Skip to content

Commit

Permalink
Merge branch 'branch-24.06' of github.com:nv-morpheus/Morpheus into e…
Browse files Browse the repository at this point in the history 
…nv-config-value
  • Loading branch information
@cwharris
cwharris committed May 20, 2024
2 parents f47fe92 + 08e40dc commit b917f29
Show file tree
Hide file tree
Showing 29 changed files with 1,279 additions and 568 deletions.
2 changes: 1 addition & 1 deletion morpheus/_lib/cmake/libmorpheus.cmake
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ add_library(morpheus
src/stages/add_scores.cpp
src/stages/deserialize.cpp
src/stages/file_source.cpp
src/stages/filter_detection.cpp
src/stages/filter_detections.cpp
src/stages/http_server_source_stage.cpp
src/stages/inference_client_stage.cpp
src/stages/kafka_source.cpp
Expand Down
77 changes: 52 additions & 25 deletions ...lude/morpheus/stages/filter_detection.hpp → ...ude/morpheus/stages/filter_detections.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,22 +17,22 @@

#pragma once

#include "morpheus/export.h"
#include "morpheus/messages/multi.hpp"
#include "morpheus/objects/dev_mem_info.hpp" // for DevMemInfo
#include "morpheus/objects/filter_source.hpp"
#include "morpheus/export.h" // for MORPHEUS_EXPORT
#include "morpheus/messages/control.hpp" // for ControlMessage
#include "morpheus/messages/multi.hpp" // for MultiMessage
#include "morpheus/objects/dev_mem_info.hpp" // for DevMemInfo
#include "morpheus/objects/filter_source.hpp" // for FilterSource

#include <boost/fiber/context.hpp>
#include <mrc/segment/builder.hpp>
#include <mrc/segment/object.hpp>
#include <pymrc/node.hpp>
#include <rxcpp/rx.hpp>
#include <cuda_runtime.h> // for cudaMemcpy
#include <mrc/segment/builder.hpp> // for Builder
#include <mrc/segment/object.hpp> // for Object
#include <pymrc/node.hpp> // for PythonNode
#include <rxcpp/rx.hpp> // for observable_member, trace_activity, map, decay_t, from

#include <cstddef> // for size_t
#include <map>
#include <memory>
#include <string>
#include <thread>
#include <map> // for map
#include <memory> // for allocator, shared_ptr
#include <string> // for string

namespace morpheus {
/****** Component public implementations *******************/
Expand Down Expand Up @@ -68,11 +68,12 @@ namespace morpheus {
* Depending on the downstream stages, this can cause performance issues, especially if those stages need to acquire
* the Python GIL.
*/
template <typename MessageT>
class MORPHEUS_EXPORT FilterDetectionsStage
: public mrc::pymrc::PythonNode<std::shared_ptr<MultiMessage>, std::shared_ptr<MultiMessage>>
: public mrc::pymrc::PythonNode<std::shared_ptr<MessageT>, std::shared_ptr<MessageT>>
{
public:
using base_t = mrc::pymrc::PythonNode<std::shared_ptr<MultiMessage>, std::shared_ptr<MultiMessage>>;
using base_t = mrc::pymrc::PythonNode<std::shared_ptr<MessageT>, std::shared_ptr<MessageT>>;
using typename base_t::sink_type_t;
using typename base_t::source_type_t;
using typename base_t::subscribe_fn_t;
Expand All @@ -90,8 +91,8 @@ class MORPHEUS_EXPORT FilterDetectionsStage

private:
subscribe_fn_t build_operator();
DevMemInfo get_tensor_filter_source(const std::shared_ptr<morpheus::MultiMessage>& x);
DevMemInfo get_column_filter_source(const std::shared_ptr<morpheus::MultiMessage>& x);
DevMemInfo get_tensor_filter_source(const sink_type_t& x);
DevMemInfo get_column_filter_source(const sink_type_t& x);

float m_threshold;
bool m_copy;
Expand All @@ -101,14 +102,39 @@ class MORPHEUS_EXPORT FilterDetectionsStage
std::map<std::size_t, std::string> m_idx2label;
};

using FilterDetectionsStageMM = // NOLINT(readability-identifier-naming)
FilterDetectionsStage<MultiMessage>;
using FilterDetectionsStageCM = // NOLINT(readability-identifier-naming)
FilterDetectionsStage<ControlMessage>;

/****** FilterDetectionStageInterfaceProxy******************/
/**
* @brief Interface proxy, used to insulate python bindings.
*/
struct MORPHEUS_EXPORT FilterDetectionStageInterfaceProxy
{
/**
* @brief Create and initialize a FilterDetectionStage, and return the result
* @brief Create and initialize a FilterDetectionStage that receives MultiMessage and emits MultiMessage, and return
* the result
*
* @param builder : Pipeline context object reference
* @param name : Name of a stage reference
* @param threshold : Threshold to classify
* @param copy : Whether or not to perform a copy default=true
* @param filter_source : Indicate if the values used for filtering exist in either an output tensor
* (`FilterSource::TENSOR`) or a column in a Dataframe (`FilterSource::DATAFRAME`).
* @param field_name : Name of the tensor or Dataframe column to filter on default="probs"
* @return std::shared_ptr<mrc::segment::Object<FilterDetectionsStage<MultiMessage, MultiMessage>>>
*/
static std::shared_ptr<mrc::segment::Object<FilterDetectionsStageMM>> init_mm(mrc::segment::Builder& builder,
const std::string& name,
float threshold,
bool copy,
FilterSource filter_source,
std::string field_name);
/**
* @brief Create and initialize a FilterDetectionStage that receives ControlMessage and emits ControlMessage, and
* return the result
*
* @param builder : Pipeline context object reference
* @param name : Name of a stage reference
Expand All @@ -117,14 +143,15 @@ struct MORPHEUS_EXPORT FilterDetectionStageInterfaceProxy
* @param filter_source : Indicate if the values used for filtering exist in either an output tensor
* (`FilterSource::TENSOR`) or a column in a Dataframe (`FilterSource::DATAFRAME`).
* @param field_name : Name of the tensor or Dataframe column to filter on default="probs"
* @return std::shared_ptr<mrc::segment::Object<FilterDetectionsStage>>
* @return std::shared_ptr<mrc::segment::Object<FilterDetectionsStage<ControlMessage, ControlMessage>>>
*/
static std::shared_ptr<mrc::segment::Object<FilterDetectionsStage>> init(mrc::segment::Builder& builder,
const std::string& name,
float threshold,
bool copy,
FilterSource filter_source,
std::string field_name);
static std::shared_ptr<mrc::segment::Object<FilterDetectionsStageCM>> init_cm(mrc::segment::Builder& builder,
const std::string& name,
float threshold,
bool copy,
FilterSource filter_source,
std::string field_name);
};

/** @} */ // end of group
} // namespace morpheus
30 changes: 13 additions & 17 deletions morpheus/_lib/src/stages/add_scores_stage_base.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,26 +18,24 @@
#include "morpheus/stages/add_scores_stage_base.hpp"

#include "morpheus/messages/memory/tensor_memory.hpp" // for TensorMemory
#include "morpheus/messages/meta.hpp"
#include "morpheus/messages/multi_response.hpp" // for MultiResponseMessage
#include "morpheus/objects/dtype.hpp" // for DType
#include "morpheus/objects/tensor.hpp" // for Tensor
#include "morpheus/objects/tensor_object.hpp" // for TensorObject
#include "morpheus/types.hpp" // for TensorIndex
#include "morpheus/utilities/matx_util.hpp" // for MatxUtil
#include "morpheus/utilities/string_util.hpp" // for StringUtil
#include "morpheus/utilities/tensor_util.hpp" // for TensorUtils

#include <glog/logging.h> // for CHECK, COMPACT_GOOGLE_LOG_FATAL, LogMessageFatal, COMP...
#include "morpheus/messages/meta.hpp" // for MessageMeta
#include "morpheus/messages/multi_response.hpp" // for MultiResponseMessage
#include "morpheus/objects/dtype.hpp" // for DType
#include "morpheus/objects/tensor.hpp" // for Tensor
#include "morpheus/objects/tensor_object.hpp" // for TensorObject
#include "morpheus/types.hpp" // for TensorIndex
#include "morpheus/utilities/matx_util.hpp" // for MatxUtil
#include "morpheus/utilities/string_util.hpp" // for StringUtil
#include "morpheus/utilities/tensor_util.hpp" // for TensorUtils

#include <glog/logging.h> // for CHECK, COMPACT_GOOGLE_LOG_FATAL, LogMessageFatal
#include <rxcpp/rx.hpp> // for observable_member, trace_activity, decay_t, operator|

#include <cstddef> // for size_t
#include <iterator> // for reverse_iterator
#include <memory> // for shared_ptr, allocator, __shared_ptr_access
#include <ostream> // for basic_ostream, operator<<, basic_ostream::operator<<
#include <stdexcept> // for runtime_error
#include <type_traits> // for is_same_v
#include <typeinfo> // for type_info
#include <utility> // for move, pair
#include <vector> // for vector
// IWYU thinks we need __alloc_traits<>::value_type for vector assignments
Expand Down Expand Up @@ -72,12 +70,10 @@ AddScoresStageBase<InputT, OutputT>::source_type_t AddScoresStageBase<InputT, Ou
{
this->on_control_message(x);
}
// sink_type_t not supported
else
{
std::string error_msg{"AddScoresStageBase receives unsupported input type: " + std::string(typeid(x).name())};
LOG(ERROR) << error_msg;
throw std::runtime_error(error_msg);
// sink_type_t not supported
static_assert(!sizeof(sink_type_t), "AddScoresStageBase receives unsupported input type");
}
return x;
}
Expand Down
Loading

0 comments on commit b917f29

Please sign in to comment.