Support ControlMessage for Preprocess and PostProcess stages

morpheus/_lib/cmake/libmorpheus.cmake

@@ -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

morpheus/_lib/include/morpheus/stages/filter_detection.hpp → morpheus/_lib/include/morpheus/stages/filter_detections.hpp

@@ -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 *******************/
@@ -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;
@@ -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;
@@ -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
@@ -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

morpheus/_lib/src/stages/add_scores_stage_base.cpp

@@ -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
@@ -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;
 }