remove from preprocess_fil cpp impl

morpheus/_lib/include/morpheus/stages/preprocess_fil.hpp

@@ -50,12 +50,11 @@ namespace morpheus {
 /**
  * @brief FIL input data for inference
  */
-template <typename InputT, typename OutputT>
 class MORPHEUS_EXPORT PreprocessFILStage
-  : public mrc::pymrc::PythonNode<std::shared_ptr<InputT>, std::shared_ptr<OutputT>>
+  : public mrc::pymrc::PythonNode<std::shared_ptr<ControlMessage>, std::shared_ptr<ControlMessage>>
 {
   public:
-    using base_t = mrc::pymrc::PythonNode<std::shared_ptr<InputT>, std::shared_ptr<OutputT>>;
+    using base_t = mrc::pymrc::PythonNode<std::shared_ptr<ControlMessage>, std::shared_ptr<ControlMessage>>;
     using typename base_t::sink_type_t;
     using typename base_t::source_type_t;
     using typename base_t::subscribe_fn_t;
@@ -73,47 +72,27 @@ class MORPHEUS_EXPORT PreprocessFILStage
     source_type_t on_data(sink_type_t x);
 
   private:
-    std::shared_ptr<MultiInferenceMessage> on_multi_message(std::shared_ptr<MultiMessage> x);
-    std::shared_ptr<ControlMessage> on_control_message(std::shared_ptr<ControlMessage> x);
-    void transform_bad_columns(std::vector<std::string>& fea_cols, morpheus::MutableTableInfo& mutable_info);
     TableInfo fix_bad_columns(sink_type_t x);
 
     std::vector<std::string> m_fea_cols;
     std::string m_vocab_file;
 };
 
-using PreprocessFILStageMM =  // NOLINT(readability-identifier-naming)
-    PreprocessFILStage<MultiMessage, MultiInferenceMessage>;
-using PreprocessFILStageCM =  // NOLINT(readability-identifier-naming)
-    PreprocessFILStage<ControlMessage, ControlMessage>;
-
 /****** PreprocessFILStageInferenceProxy********************/
 /**
  * @brief Interface proxy, used to insulate python bindings.
  */
 struct MORPHEUS_EXPORT PreprocessFILStageInterfaceProxy
 {
-    /**
-     * @brief Create and initialize a PreprocessFILStage that receives MultiMessage and emits MultiInferenceMessage,
-     * and return the result
-     *
-     * @param builder : Pipeline context object reference
-     * @param name : Name of a stage reference
-     * @param features : Reference to the features that are required for model inference
-     * @return std::shared_ptr<mrc::segment::Object<PreprocessFILStage<MultiMessage, MultiInferenceMessage>>>
-     */
-    static std::shared_ptr<mrc::segment::Object<PreprocessFILStage<MultiMessage, MultiInferenceMessage>>> init_multi(
-        mrc::segment::Builder& builder, const std::string& name, const std::vector<std::string>& features);
-
     /**
      * @brief Create and initialize a PreprocessFILStage, and return the result
      *
      * @param builder : Pipeline context object reference
      * @param name : Name of a stage reference
      * @param features : Reference to the features that are required for model inference
-     * @return std::shared_ptr<mrc::segment::Object<PreprocessFILStage<ControlMessage, ControlMessage>>>
+     * @return std::shared_ptr<mrc::segment::Object<PreprocessFILStage>>
      */
-    static std::shared_ptr<mrc::segment::Object<PreprocessFILStage<ControlMessage, ControlMessage>>> init_cm(
+    static std::shared_ptr<mrc::segment::Object<PreprocessFILStage>> init(
         mrc::segment::Builder& builder, const std::string& name, const std::vector<std::string>& features);
 };
 /** @} */  // end of group

morpheus/_lib/src/stages/preprocess_fil.cpp

@@ -55,183 +55,77 @@
 namespace morpheus {
 // Component public implementations
 // ************ PreprocessFILStage ************************* //
-template <typename InputT, typename OutputT>
-PreprocessFILStage<InputT, OutputT>::PreprocessFILStage(const std::vector<std::string>& features) :
-  base_t(rxcpp::operators::map([this](sink_type_t x) {
-      return this->on_data(std::move(x));
+PreprocessFILStage::PreprocessFILStage(const std::vector<std::string>& features) :
+  base_t(rxcpp::operators::map([this](sink_type_t msg) {
+      return this->on_data(std::move(msg));
   })),
   m_fea_cols(std::move(features))
 {}
 
-template <typename InputT, typename OutputT>
-void PreprocessFILStage<InputT, OutputT>::transform_bad_columns(std::vector<std::string>& fea_cols,
-                                                                morpheus::MutableTableInfo& mutable_info)
+TableInfo PreprocessFILStage::fix_bad_columns(sink_type_t msg)
 {
-    auto df_meta_col_names = mutable_info.get_column_names();
-    std::vector<std::string> bad_cols;
-    // Only check the feature columns. Leave the rest unchanged
-    for (auto& fea_col : fea_cols)
     {
-        // Find the index of the column in the dataframe
-        auto col_idx =
-            std::find(df_meta_col_names.begin(), df_meta_col_names.end(), fea_col) - df_meta_col_names.begin();
+        // Get the mutable info for the entire meta object so we only do this once
+        // per dataframe
+        auto mutable_info      = msg->payload()->get_mutable_info();
+        auto df_meta_col_names = mutable_info.get_column_names();
 
-        if (col_idx == df_meta_col_names.size())
-        {
-            // This feature was not found. Ignore it.
-            continue;
-        }
-
-        if (mutable_info.get_column(col_idx).type().id() == cudf::type_id::STRING)
-        {
-            bad_cols.push_back(fea_col);
-        }
-    }
-
-    // Exit early if there is nothing to do
-    if (!bad_cols.empty())
-    {
-        // Need to ensure all string columns have been converted to numbers. This requires running a
-        // regex which is too difficult to do from C++ at this time. So grab the GIL, make the
-        // conversions, and release. This is horribly inefficient, but so is the JSON lines format for
-        // this workflow
-        using namespace pybind11::literals;
-        pybind11::gil_scoped_acquire gil;
-
-        // pybind11::object df = x->meta->get_py_table();
-        auto pdf = mutable_info.checkout_obj();
-        auto& df = *pdf;
-
-        std::string regex = R"((\d+))";
+        std::vector<std::string> bad_cols;
 
-        for (auto c : bad_cols)
+        // Only check the feature columns. Leave the rest unchanged
+        for (auto& fea_col : m_fea_cols)
         {
-            df[pybind11::str(c)] = df[pybind11::str(c)]
-                                       .attr("str")
-                                       .attr("extract")(pybind11::str(regex), "expand"_a = true)
-                                       .attr("astype")(pybind11::str("float32"));
+            // Find the index of the column in the dataframe
+            auto col_idx =
+                std::find(df_meta_col_names.begin(), df_meta_col_names.end(), fea_col) - df_meta_col_names.begin();
+
+            if (col_idx == df_meta_col_names.size())
+            {
+                // This feature was not found. Ignore it.
+                continue;
+            }
+
+            if (mutable_info.get_column(col_idx).type().id() == cudf::type_id::STRING)
+            {
+                bad_cols.push_back(fea_col);
+            }
         }
 
-        mutable_info.return_obj(std::move(pdf));
-    }
-}
-
-template <typename InputT, typename OutputT>
-TableInfo PreprocessFILStage<InputT, OutputT>::fix_bad_columns(sink_type_t x)
-{
-    if constexpr (std::is_same_v<sink_type_t, std::shared_ptr<MultiMessage>>)
-    {
-        {
-            // Get the mutable info for the entire meta object so we only do this once per dataframe
-            auto mutable_info = x->meta->get_mutable_info();
-            transform_bad_columns(this->m_fea_cols, mutable_info);
-        }
-
-        // Now re-get the meta
-        return x->get_meta(m_fea_cols);
-    }
-    else if constexpr (std::is_same_v<sink_type_t, std::shared_ptr<ControlMessage>>)
-    {
-        {
-            // Get the mutable info for the entire meta object so we only do this once per dataframe
-            auto mutable_info = x->payload()->get_mutable_info();
-            transform_bad_columns(this->m_fea_cols, mutable_info);
-        }
-
-        // Now re-get the meta
-        return x->payload()->get_info(m_fea_cols);
-    }
-    else
-    {
-        // sink_type_t not supported
-        static_assert(!sizeof(sink_type_t), "PreProcessFILStage receives unsupported input type");
-    }
-}
-
-template <typename InputT, typename OutputT>
-PreprocessFILStage<InputT, OutputT>::source_type_t PreprocessFILStage<InputT, OutputT>::on_data(sink_type_t x)
-{
-    if constexpr (std::is_same_v<sink_type_t, std::shared_ptr<MultiMessage>>)
-    {
-        return on_multi_message(x);
-    }
-    else if constexpr (std::is_same_v<sink_type_t, std::shared_ptr<ControlMessage>>)
-    {
-        return on_control_message(x);
-    }
-    else
-    {
-        // sink_type_t not supported
-        static_assert(!sizeof(sink_type_t), "PreProcessFILStage receives unsupported input type");
-    }
-}
-
-template <>
-std::shared_ptr<MultiInferenceMessage> PreprocessFILStage<MultiMessage, MultiInferenceMessage>::on_multi_message(
-    std::shared_ptr<MultiMessage> x)
-{
-    auto packed_data = std::make_shared<rmm::device_buffer>(m_fea_cols.size() * x->mess_count * sizeof(float),
-                                                            rmm::cuda_stream_per_thread);
-    auto df_meta     = this->fix_bad_columns(x);
-    for (size_t i = 0; i < df_meta.num_columns(); ++i)
-    {
-        auto curr_col = df_meta.get_column(i);
-
-        auto curr_ptr = static_cast<float*>(packed_data->data()) + i * df_meta.num_rows();
-
-        // Check if we are something other than float
-        if (curr_col.type().id() != cudf::type_id::FLOAT32)
+        // Exit early if there is nothing to do
+        if (!bad_cols.empty())
         {
-            auto float_data = cudf::cast(curr_col, cudf::data_type(cudf::type_id::FLOAT32))->release();
-
-            // Do the copy here before it goes out of scope
-            MRC_CHECK_CUDA(cudaMemcpy(
-                curr_ptr, float_data.data->data(), df_meta.num_rows() * sizeof(float), cudaMemcpyDeviceToDevice));
-        }
-        else
-        {
-            MRC_CHECK_CUDA(cudaMemcpy(curr_ptr,
-                                      curr_col.template data<float>(),
-                                      df_meta.num_rows() * sizeof(float),
-                                      cudaMemcpyDeviceToDevice));
+            // Need to ensure all string columns have been converted to numbers. This
+            // requires running a regex which is too difficult to do from C++ at this
+            // time. So grab the GIL, make the conversions, and release. This is
+            // horribly inefficient, but so is the JSON lines format for this workflow
+            using namespace pybind11::literals;
+            pybind11::gil_scoped_acquire gil;
+
+            // pybind11::object df = x->meta->get_py_table();
+            auto pdf = mutable_info.checkout_obj();
+            auto& df = *pdf;
+
+            std::string regex = R"((\d+))";
+
+            for (auto c : bad_cols)
+            {
+                df[pybind11::str(c)] = df[pybind11::str(c)]
+                                           .attr("str")
+                                           .attr("extract")(pybind11::str(regex), "expand"_a = true)
+                                           .attr("astype")(pybind11::str("float32"));
+            }
+
+            mutable_info.return_obj(std::move(pdf));
         }
     }
 
-    // Need to convert from row major to column major
-    // Easiest way to do this is to transpose the data from [fea_len, row_count] to [row_count, fea_len]
-    auto transposed_data = MatxUtil::transpose(DevMemInfo{packed_data,
-                                                          TypeId::FLOAT32,
-                                                          {static_cast<TensorIndex>(m_fea_cols.size()), x->mess_count},
-                                                          {x->mess_count, 1}});
-
-    // Create the tensor which will be row-major and size [row_count, fea_len]
-    auto input__0 = Tensor::create(
-        transposed_data, DType::create<float>(), {x->mess_count, static_cast<TensorIndex>(m_fea_cols.size())}, {}, 0);
-
-    auto seq_id_dtype = DType::create<TensorIndex>();
-    auto seq_ids      = Tensor::create(
-        MatxUtil::create_seq_ids(
-            x->mess_count, m_fea_cols.size(), seq_id_dtype.type_id(), input__0.get_memory(), x->mess_offset),
-        seq_id_dtype,
-        {x->mess_count, 3},
-        {},
-        0);
-
-    // Build the results
-    auto memory = std::make_shared<InferenceMemoryFIL>(x->mess_count, std::move(input__0), std::move(seq_ids));
-
-    auto next = std::make_shared<MultiInferenceMessage>(
-        x->meta, x->mess_offset, x->mess_count, std::move(memory), 0, memory->count);
-
-    return next;
+    // Now re-get the meta
+    return msg->payload()->get_info(m_fea_cols);
 }
 
-template <>
-std::shared_ptr<ControlMessage> PreprocessFILStage<ControlMessage, ControlMessage>::on_control_message(
-    std::shared_ptr<ControlMessage> x)
-
+PreprocessFILStage::source_type_t PreprocessFILStage::on_data(sink_type_t msg)
 {
-    auto df_meta        = this->fix_bad_columns(x);
+    auto df_meta        = this->fix_bad_columns(msg);
     const auto num_rows = df_meta.num_rows();
 
     auto packed_data =
@@ -259,7 +153,8 @@ std::shared_ptr<ControlMessage> PreprocessFILStage<ControlMessage, ControlMessag
     }
 
     // Need to convert from row major to column major
-    // Easiest way to do this is to transpose the data from [fea_len, row_count] to [row_count, fea_len]
+    // Easiest way to do this is to transpose the data from [fea_len, row_count]
+    // to [row_count, fea_len]
     auto transposed_data = MatxUtil::transpose(DevMemInfo{
         packed_data, TypeId::FLOAT32, {static_cast<TensorIndex>(m_fea_cols.size()), num_rows}, {num_rows, 1}});
 
@@ -279,27 +174,16 @@ std::shared_ptr<ControlMessage> PreprocessFILStage<ControlMessage, ControlMessag
     auto memory = std::make_shared<TensorMemory>(num_rows);
     memory->set_tensor("input__0", std::move(input__0));
     memory->set_tensor("seq_ids", std::move(seq_ids));
-    x->tensors(memory);
+    msg->tensors(memory);
 
-    return x;
+    return msg;
 }
 
-template class PreprocessFILStage<MultiMessage, MultiInferenceMessage>;
-template class PreprocessFILStage<ControlMessage, ControlMessage>;
-
 // ************ PreprocessFILStageInterfaceProxy *********** //
-std::shared_ptr<mrc::segment::Object<PreprocessFILStageMM>> PreprocessFILStageInterfaceProxy::init_multi(
-    mrc::segment::Builder& builder, const std::string& name, const std::vector<std::string>& features)
-{
-    auto stage = builder.construct_object<PreprocessFILStageMM>(name, features);
-
-    return stage;
-}
-
-std::shared_ptr<mrc::segment::Object<PreprocessFILStageCM>> PreprocessFILStageInterfaceProxy::init_cm(
+std::shared_ptr<mrc::segment::Object<PreprocessFILStage>> PreprocessFILStageInterfaceProxy::init(
     mrc::segment::Builder& builder, const std::string& name, const std::vector<std::string>& features)
 {
-    auto stage = builder.construct_object<PreprocessFILStageCM>(name, features);
+    auto stage = builder.construct_object<PreprocessFILStage>(name, features);
 
     return stage;
 }

morpheus/_lib/stages/__init__.pyi

@@ -29,8 +29,7 @@ __all__ = [
     "PreallocateControlMessageStage",
     "PreallocateMessageMetaStage",
     "PreallocateMultiMessageStage",
-    "PreprocessFILControlMessageStage",
-    "PreprocessFILMultiMessageStage",
+    "PreprocessFILStage",
     "PreprocessNLPStage",
     "SerializeControlMessageStage",
     "SerializeMultiMessageStage",
@@ -86,10 +85,7 @@ class PreallocateMessageMetaStage(mrc.core.segment.SegmentObject):
 class PreallocateMultiMessageStage(mrc.core.segment.SegmentObject):
     def __init__(self, builder: mrc.core.segment.Builder, name: str, needed_columns: typing.List[typing.Tuple[str, morpheus._lib.common.TypeId]]) -> None: ...
     pass
-class PreprocessFILControlMessageStage(mrc.core.segment.SegmentObject):
-    def __init__(self, builder: mrc.core.segment.Builder, name: str, features: typing.List[str]) -> None: ...
-    pass
-class PreprocessFILMultiMessageStage(mrc.core.segment.SegmentObject):
+class PreprocessFILStage(mrc.core.segment.SegmentObject):
     def __init__(self, builder: mrc.core.segment.Builder, name: str, features: typing.List[str]) -> None: ...
     pass
 class PreprocessNLPStage(mrc.core.segment.SegmentObject):

morpheus/_lib/stages/module.cpp

@@ -256,20 +256,11 @@ PYBIND11_MODULE(stages, _module)
              py::arg("name"),
              py::arg("needed_columns"));
 
-    py::class_<mrc::segment::Object<PreprocessFILStageMM>,
+    py::class_<mrc::segment::Object<PreprocessFILStage>,
                mrc::segment::ObjectProperties,
-               std::shared_ptr<mrc::segment::Object<PreprocessFILStageMM>>>(
-        _module, "PreprocessFILMultiMessageStage", py::multiple_inheritance())
-        .def(py::init<>(&PreprocessFILStageInterfaceProxy::init_multi),
-             py::arg("builder"),
-             py::arg("name"),
-             py::arg("features"));
-
-    py::class_<mrc::segment::Object<PreprocessFILStageCM>,
-               mrc::segment::ObjectProperties,
-               std::shared_ptr<mrc::segment::Object<PreprocessFILStageCM>>>(
-        _module, "PreprocessFILControlMessageStage", py::multiple_inheritance())
-        .def(py::init<>(&PreprocessFILStageInterfaceProxy::init_cm),
+               std::shared_ptr<mrc::segment::Object<PreprocessFILStage>>>(
+        _module, "PreprocessFILStage", py::multiple_inheritance())
+        .def(py::init<>(&PreprocessFILStageInterfaceProxy::init),
              py::arg("builder"),
              py::arg("name"),
              py::arg("features"));