MultiInferenceMessage & MultiResponseMessage share a new base class

morpheus/_lib/cmake/libraries/morpheus.cmake

@@ -16,6 +16,7 @@ message(STATUS "Adding library: morpheus")
 
 add_library(morpheus
     # Keep these sorted!
+    ${MORPHEUS_LIB_ROOT}/src/io/deserializers.cpp
     ${MORPHEUS_LIB_ROOT}/src/io/serializers.cpp
     ${MORPHEUS_LIB_ROOT}/src/messages/memory/inference_memory.cpp
     ${MORPHEUS_LIB_ROOT}/src/messages/memory/inference_memory_fil.cpp
@@ -30,6 +31,7 @@ add_library(morpheus
     ${MORPHEUS_LIB_ROOT}/src/messages/multi_inference_nlp.cpp
     ${MORPHEUS_LIB_ROOT}/src/messages/multi_response.cpp
     ${MORPHEUS_LIB_ROOT}/src/messages/multi_response_probs.cpp
+    ${MORPHEUS_LIB_ROOT}/src/messages/multi_tensor.cpp
     ${MORPHEUS_LIB_ROOT}/src/objects/fiber_queue.cpp
     ${MORPHEUS_LIB_ROOT}/src/objects/file_types.cpp
     ${MORPHEUS_LIB_ROOT}/src/objects/wrapped_tensor.cpp

morpheus/_lib/include/morpheus/io/deserializers.hpp

@@ -0,0 +1,55 @@
+/**
+ * SPDX-FileCopyrightText: Copyright (c) 2022, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <cudf/io/json.hpp>
+#include <cudf/io/types.hpp>
+
+#include <string>
+
+namespace morpheus {
+#pragma GCC visibility push(default)
+
+/**
+ * @brief Loads a cudf table from either CSV or JSON file
+ *
+ * @param filename
+ * @return cudf::io::table_with_metadata
+ */
+cudf::io::table_with_metadata load_table_from_file(const std::string& filename);
+
+/**
+ * @brief Loads a cudf table from a json soruce, replacing any escape characters in the source data that cudf can't
+ * handle
+ *
+ * @param json_options
+ * @return cudf::io::table_with_metadata
+ */
+cudf::io::table_with_metadata load_json_table(cudf::io::json_reader_options&& json_options);
+
+/**
+ * @brief Return the number of index columns in `data_table`, in practice this will be a `0` or `1`.
+ * If `data_table` contains a column named "Unnamed: 0" it will be renamed to "".
+ *
+ * @param data_table
+ * @return int
+ */
+int get_index_col_count(cudf::io::table_with_metadata& data_table);
+
+#pragma GCC visibility pop
+}  // namespace morpheus

morpheus/_lib/include/morpheus/messages/memory/inference_memory.hpp

@@ -26,6 +26,8 @@ namespace morpheus {
 /**
  * TODO(Documentation)
  */
+
+#pragma GCC visibility push(default)
 class InferenceMemory : public TensorMemory
 {
   public:
@@ -43,7 +45,6 @@ class InferenceMemory : public TensorMemory
 };
 
 /****** InferenceMemoryInterfaceProxy *************************/
-#pragma GCC visibility push(default)
 /**
  * @brief Interface proxy, used to insulate python bindings.
  */

morpheus/_lib/include/morpheus/messages/memory/response_memory_probs.hpp

@@ -18,6 +18,7 @@
 #pragma once
 
 #include "morpheus/messages/memory/response_memory.hpp"
+#include "morpheus/messages/memory/tensor_memory.hpp"
 #include "morpheus/objects/tensor_object.hpp"
 
 #include <cudf/types.hpp>
@@ -36,14 +37,19 @@ class ResponseMemoryProbs : public ResponseMemory
 {
   public:
     ResponseMemoryProbs(size_t count, TensorObject probs);
+    ResponseMemoryProbs(size_t count, tensor_map_t &&tensors);
 
     /**
-     * TODO(Documentation)
+     * @brief Return the tensor named 'probs', throws a `std::runtime_error` if it does not exist.
+     *
+     * @return const TensorObject&
      */
     const TensorObject &get_probs() const;
 
     /**
-     * TODO(Documentation)
+     * @brief Update the tensor named 'probs'
+     *
+     * @param probs
      */
     void set_probs(TensorObject probs);
 };

morpheus/_lib/include/morpheus/messages/memory/tensor_memory.hpp

@@ -41,6 +41,7 @@ class TensorMemory
 
     TensorMemory(size_t count);
     TensorMemory(size_t count, tensor_map_t &&tensors);
+    virtual ~TensorMemory() = default;
 
     size_t count{0};
     tensor_map_t tensors;

morpheus/_lib/include/morpheus/messages/multi_inference.hpp

@@ -20,6 +20,7 @@
 #include "morpheus/messages/memory/inference_memory.hpp"
 #include "morpheus/messages/meta.hpp"
 #include "morpheus/messages/multi.hpp"
+#include "morpheus/messages/multi_tensor.hpp"
 #include "morpheus/objects/tensor_object.hpp"
 
 #include <cudf/types.hpp>
@@ -38,7 +39,7 @@ namespace morpheus {
  * TODO(Documentation)
  */
 #pragma GCC visibility push(default)
-class MultiInferenceMessage : public DerivedMultiMessage<MultiInferenceMessage, MultiMessage>
+class MultiInferenceMessage : public DerivedMultiMessage<MultiInferenceMessage, MultiTensorMessage>
 {
   public:
     MultiInferenceMessage(const MultiInferenceMessage &other) = default;
@@ -49,32 +50,26 @@ class MultiInferenceMessage : public DerivedMultiMessage<MultiInferenceMessage,
                           std::size_t offset,
                           std::size_t count);
 
-    std::shared_ptr<morpheus::InferenceMemory> memory;
-    std::size_t offset{0};
-    std::size_t count{0};
-
     /**
-     * TODO(Documentation)
+     * @brief Return the input tensor for the given `name`. Will halt on a fatal error if the tensor does not exist.
+     *
+     * @param name
+     * @return const TensorObject
      */
     const TensorObject get_input(const std::string &name) const;
 
     /**
-     * TODO(Documentation)
+     * @brief Return the input tensor for the given `name`. Will halt on a fatal error if the tensor does not exist.
+     *
+     * @param name
+     * @return TensorObject
      */
-    const void set_input(const std::string &name, const TensorObject &value);
+    TensorObject get_input(const std::string &name);
 
-  protected:
     /**
-     * TODO(Documentation)
+     * Update the value of ain input tensor. The tensor must already exist, otherwise this will halt on a fatal error.
      */
-    void get_slice_impl(std::shared_ptr<MultiMessage> new_message, std::size_t start, std::size_t stop) const override;
-
-    void copy_ranges_impl(std::shared_ptr<MultiMessage> new_message,
-                          const std::vector<std::pair<size_t, size_t>> &ranges,
-                          size_t num_selected_rows) const override;
-
-    std::shared_ptr<InferenceMemory> copy_input_ranges(const std::vector<std::pair<size_t, size_t>> &ranges,
-                                                       size_t num_selected_rows) const;
+    void set_input(const std::string &name, const TensorObject &value);
 };
 
 /****** MultiInferenceMessageInterfaceProxy****************/

morpheus/_lib/include/morpheus/messages/multi_inference_fil.hpp

@@ -45,22 +45,32 @@ class MultiInferenceFILMessage : public MultiInferenceMessage
                              size_t count);
 
     /**
-     * TODO(Documentation)
+     * @brief Return the 'input__0' tensor, throws a `std::runtime_error` if it does not exist.
+     *
+     * @param name
+     * @return const TensorObject
      */
     const TensorObject get_input__0() const;
 
     /**
-     * TODO(Documentation)
+     * @brief Sets a tensor named 'input__0'.
+     *
+     * @param input__0
      */
     void set_input__0(const TensorObject& input__0);
 
     /**
-     * TODO(Documentation)
+     * @brief Return the 'seq_ids' tensor, throws a `std::runtime_error` if it does not exist.
+     *
+     * @param name
+     * @return const TensorObject
      */
     const TensorObject get_seq_ids() const;
 
     /**
-     * TODO(Documentation)
+     * @brief Sets a tensor named 'seq_ids'.
+     *
+     * @param seq_ids
      */
     void set_seq_ids(const TensorObject& seq_ids);
 };

morpheus/_lib/include/morpheus/messages/multi_inference_nlp.hpp

@@ -46,32 +46,47 @@ class MultiInferenceNLPMessage : public MultiInferenceMessage
                              std::size_t count);
 
     /**
-     * TODO(Documentation)
+     * @brief Return the 'input_ids' tensor, throws a `std::runtime_error` if it does not exist.
+     *
+     * @param name
+     * @return const TensorObject
      */
     const TensorObject get_input_ids() const;
 
     /**
-     * TODO(Documentation)
+     * @brief Sets a tensor named 'input_ids'.
+     *
+     * @param input_ids
      */
     void set_input_ids(const TensorObject& input_ids);
 
     /**
-     * TODO(Documentation)
+     * @brief Return the 'input_mask' tensor, throws a `std::runtime_error` if it does not exist.
+     *
+     * @param name
+     * @return const TensorObject
      */
     const TensorObject get_input_mask() const;
 
     /**
-     * TODO(Documentation)
+     * @brief Sets a tensor named 'input_mask'.
+     *
+     * @param input_mask
      */
     void set_input_mask(const TensorObject& input_mask);
 
     /**
-     * TODO(Documentation)
+     * @brief Return the 'seq_ids' tensor, throws a `std::runtime_error` if it does not exist.
+     *
+     * @param name
+     * @return const TensorObject
      */
     const TensorObject get_seq_ids() const;
 
     /**
-     * TODO(Documentation)
+     * @brief Sets a tensor named 'seq_ids'.
+     *
+     * @param seq_ids
      */
     void set_seq_ids(const TensorObject& seq_ids);
 };

morpheus/_lib/include/morpheus/messages/multi_response.hpp

@@ -20,6 +20,7 @@
 #include "morpheus/messages/memory/response_memory.hpp"
 #include "morpheus/messages/meta.hpp"
 #include "morpheus/messages/multi.hpp"
+#include "morpheus/messages/multi_tensor.hpp"
 #include "morpheus/objects/tensor_object.hpp"
 
 #include <cudf/types.hpp>
@@ -38,7 +39,7 @@ namespace morpheus {
  * TODO(Documentation)
  */
 #pragma GCC visibility push(default)
-class MultiResponseMessage : public DerivedMultiMessage<MultiResponseMessage, MultiMessage>
+class MultiResponseMessage : public DerivedMultiMessage<MultiResponseMessage, MultiTensorMessage>
 {
   public:
     MultiResponseMessage(const MultiResponseMessage &other) = default;
@@ -49,37 +50,30 @@ class MultiResponseMessage : public DerivedMultiMessage<MultiResponseMessage, Mu
                          std::size_t offset,
                          std::size_t count);
 
-    std::shared_ptr<ResponseMemory> memory;
-    std::size_t offset{0};
-    std::size_t count{0};
-
-    /**
-     * TODO(Documentation)
-     */
-    TensorObject get_output(const std::string &name);
-
     /**
-     * TODO(Documentation)
+     * @brief Returns the output tensor with the given name. Will halt on a fatal error if the tensor does not exist.
+     *
+     * @param name
+     * @return const TensorObject
      */
     const TensorObject get_output(const std::string &name) const;
 
     /**
-     * TODO(Documentation)
+     * @brief Returns the output tensor with the given name. Will halt on a fatal error if the tensor does not exist.
+     *
+     * @param name
+     * @return TensorObject
      */
-    const void set_output(const std::string &name, const TensorObject &value);
+    TensorObject get_output(const std::string &name);
 
-  protected:
     /**
-     * TODO(Documentation)
+     * @brief Update the value of a given output tensor. The tensor must already exist, otherwise this will halt on a
+     * fatal error.
+     *
+     * @param name
+     * @param value
      */
-    void get_slice_impl(std::shared_ptr<MultiMessage> new_message, std::size_t start, std::size_t stop) const override;
-
-    void copy_ranges_impl(std::shared_ptr<MultiMessage> new_message,
-                          const std::vector<std::pair<size_t, size_t>> &ranges,
-                          size_t num_selected_rows) const override;
-
-    std::shared_ptr<ResponseMemory> copy_output_ranges(const std::vector<std::pair<size_t, size_t>> &ranges,
-                                                       size_t num_selected_rows) const;
+    void set_output(const std::string &name, const TensorObject &value);
 };
 
 /****** MultiResponseMessageInterfaceProxy *************************/

morpheus/_lib/include/morpheus/messages/multi_response_probs.hpp

@@ -17,7 +17,7 @@
 
 #pragma once
 
-#include "morpheus/messages/memory/response_memory.hpp"
+#include "morpheus/messages/memory/response_memory_probs.hpp"
 #include "morpheus/messages/meta.hpp"
 #include "morpheus/messages/multi.hpp"
 #include "morpheus/messages/multi_response.hpp"
@@ -43,17 +43,21 @@ class MultiResponseProbsMessage : public DerivedMultiMessage<MultiResponseProbsM
     MultiResponseProbsMessage(std::shared_ptr<morpheus::MessageMeta> meta,
                               size_t mess_offset,
                               size_t mess_count,
-                              std::shared_ptr<morpheus::ResponseMemory> memory,
+                              std::shared_ptr<morpheus::ResponseMemoryProbs> memory,
                               size_t offset,
                               size_t count);
 
     /**
-     * TODO(Documentation)
+     * @brief Return the `probs` (probabilities) output tensor
+     *
+     * @return const TensorObject
      */
     const TensorObject get_probs() const;
 
     /**
-     * TODO(Documentation)
+     * @brief Update the `probs` output tensor. Will halt on a fatal error if the `probs` output tensor does not exist.
+     *
+     * @param probs
      */
     void set_probs(const TensorObject &probs);
 };
@@ -67,14 +71,14 @@ struct MultiResponseProbsMessageInterfaceProxy
     static std::shared_ptr<MultiResponseProbsMessage> init(std::shared_ptr<MessageMeta> meta,
                                                            cudf::size_type mess_offset,
                                                            cudf::size_type mess_count,
-                                                           std::shared_ptr<ResponseMemory> memory,
+                                                           std::shared_ptr<ResponseMemoryProbs> memory,
                                                            cudf::size_type offset,
                                                            cudf::size_type count);
 
     /**
      * TODO(Documentation)
      */
-    static std::shared_ptr<morpheus::ResponseMemory> memory(MultiResponseProbsMessage &self);
+    static std::shared_ptr<morpheus::ResponseMemoryProbs> memory(MultiResponseProbsMessage &self);
 
     /**
      * TODO(Documentation)