DocaConvert stage allows dynamic number of packets + fix to python pi…

…peline

examples/doca/run_udp_convert.py

@@ -24,11 +24,9 @@
 from morpheus.stages.doca.doca_convert_stage import DocaConvertStage
 from morpheus.stages.doca.doca_source_stage import DocaSourceStage
 from morpheus.stages.general.monitor_stage import MonitorStage
-from morpheus.stages.preprocess.deserialize_stage import DeserializeStage
-from morpheus.stages.preprocess.preprocess_nlp_stage import PreprocessNLPStage
+# from morpheus.stages.general.trigger_stage import TriggerStage
 from morpheus.utils.logger import configure_logging
 
-
 @click.command()
 @click.option(
     "--out_file",
@@ -55,15 +53,15 @@ def run_pipeline(out_file, nic_addr, gpu_addr):
     config.mode = PipelineModes.NLP
 
     # Below properties are specified by the command line
-    config.num_threads = 5
+    config.num_threads = 7
+    config.edge_buffer_size = 512
 
     pipeline = LinearPipeline(config)
 
     # add doca source stage
     pipeline.set_source(DocaSourceStage(config, nic_addr, gpu_addr, 'udp'))
+    # pipeline.add_stage(TriggerStage(config))
     pipeline.add_stage(DocaConvertStage(config))
-    pipeline.add_stage(DeserializeStage(config))
-    pipeline.add_stage(PreprocessNLPStage(config))
     pipeline.add_stage(MonitorStage(config, description="DOCA GPUNetIO rate", unit='pkts'))
 
     # Build the pipeline here to see types in the vizualization

examples/doca/run_udp_raw.py

@@ -52,7 +52,8 @@ def run_pipeline(out_file, nic_addr, gpu_addr):
     config.mode = PipelineModes.NLP
 
     # Below properties are specified by the command line
-    config.num_threads = 1
+    config.num_threads = 4
+    config.edge_buffer_size = 512
 
     def count_raw_packets(message: RawPacketMessage):
         return message.num
@@ -61,8 +62,7 @@ def count_raw_packets(message: RawPacketMessage):
 
     # add doca source stage
     pipeline.set_source(DocaSourceStage(config, nic_addr, gpu_addr, 'udp'))
-    pipeline.add_stage(
-        MonitorStage(config, description="DOCA GPUNetIO rate", unit='pkts', determine_count_fn=count_raw_packets))
+    pipeline.add_stage(MonitorStage(config, description="DOCA GPUNetIO rate", unit='pkts', determine_count_fn=count_raw_packets))
 
     # Build the pipeline here to see types in the vizualization
     pipeline.build()

examples/doca/vdb_realtime/vdb.py

@@ -98,17 +98,16 @@ def run_pipeline(out_file, nic_addr, gpu_addr):
 
     config = Config()
     config.mode = PipelineModes.NLP
-    config.pipeline_batch_size = 2048
+    config.pipeline_batch_size = 1024
     config.feature_length = 512
-
-    # Below properties are specified by the command line
-    # config.num_threads = 5
+    config.edge_buffer_size = 512
+    config.num_threads = 15
 
     pipeline = LinearPipeline(config)
 
     # add doca source stage
     pipeline.set_source(DocaSourceStage(config, nic_addr, gpu_addr, 'udp'))
-    pipeline.add_stage(DocaConvertStage(config, False))
+    pipeline.add_stage(DocaConvertStage(config))
 
     pipeline.add_stage(MonitorStage(config, description="DOCA GPUNetIO Source rate", unit='pkts'))
 

models/triton-model-repo/all-MiniLM-L6-v2/config.pbtxt

@@ -40,14 +40,6 @@ optimization {
         key: "precision_mode"
         value: "FP16"
       }
-      parameters {
-        key: "trt_engine_cache_enable"
-        value: "true"
-      }
-      parameters {
-        key: "trt_engine_cache_path"
-        value: "/models/triton-model-repo/.cache/triton"
-      }
     }
   }
 }

morpheus/_lib/doca/include/morpheus/doca/common.hpp

@@ -29,7 +29,8 @@ uint32_t const PACKETS_PER_BLOCK    = PACKETS_PER_THREAD * THREADS_PER_BLOCK;
 uint32_t const PACKET_RX_TIMEOUT_NS = 500000;  // 1ms //500us
 
 uint32_t const MAX_PKT_RECEIVE = PACKETS_PER_BLOCK;
-uint32_t const MAX_PKT_SIZE    = 4096;
+uint32_t const MAX_PKT_CONVERT = MAX_PKT_RECEIVE * 5;
+uint32_t const MAX_PKT_SIZE    = 2048;
 uint32_t const MAX_PKT_HDR     = 64;
 uint32_t const MAX_PKT_NUM     = 65536;
 uint32_t const MAX_QUEUE       = 2;

morpheus/_lib/doca/include/morpheus/doca/doca_kernels.hpp

@@ -35,6 +35,7 @@ std::unique_ptr<cudf::column> gather_payload(
     uintptr_t* packets_buffer,
     uint32_t* header_sizes,
     uint32_t* payload_sizes,
+    uint32_t*    fixed_size_list,
     rmm::cuda_stream_view stream,
     rmm::mr::device_memory_resource* mr = rmm::mr::get_current_device_resource());
 

morpheus/_lib/doca/include/morpheus/doca/doca_stages.hpp

@@ -109,6 +109,8 @@ class DocaConvertStage : public mrc::pymrc::PythonNode<std::shared_ptr<RawPacket
 
     cudaStream_t m_stream;
     rmm::cuda_stream_view m_stream_cpp;
+    uint32_t* fixed_size_list;
+    uint32_t* fixed_size_list_cpu;
 };
 
 /****** DocaConvertStageInterfaceProxy***********************/

morpheus/_lib/doca/src/doca_context.cpp

@@ -94,11 +94,11 @@ doca_flow_port* init_doca_flow(uint16_t port_id, uint8_t rxq_num)
     doca_flow_cfg rxq_flow_cfg = {0};
     rte_eth_dev_info dev_info  = {nullptr};
     rte_eth_conf eth_conf      = {
-             .rxmode =
+            .rxmode =
             {
-                     .mtu = 2048, /* Not really used, just to initialize DPDK */
+                     .mtu = 1024, /* Not really used, just to initialize DPDK */
             },
-             .txmode =
+            .txmode =
             {
                      .offloads = RTE_ETH_TX_OFFLOAD_IPV4_CKSUM | RTE_ETH_TX_OFFLOAD_UDP_CKSUM | RTE_ETH_TX_OFFLOAD_TCP_CKSUM,
             },

morpheus/_lib/doca/src/doca_convert.cpp

@@ -90,10 +90,17 @@ DocaConvertStage::DocaConvertStage() :
 {
     cudaStreamCreateWithFlags(&m_stream, cudaStreamNonBlocking);
     m_stream_cpp = rmm::cuda_stream_view(reinterpret_cast<cudaStream_t>(m_stream));
+    fixed_size_list_cpu = (uint32_t *) calloc(MAX_PKT_RECEIVE * 4, sizeof(uint32_t));
+    cudaMalloc((void **)&fixed_size_list, MAX_PKT_RECEIVE * 4 * sizeof(uint32_t));
+    for (int idx = 0; idx < MAX_PKT_RECEIVE * 4; idx++)
+        fixed_size_list_cpu[idx] = MAX_PKT_SIZE;
+    cudaMemcpy(fixed_size_list, fixed_size_list_cpu, MAX_PKT_RECEIVE * 4 * sizeof(uint32_t), cudaMemcpyDefault);
 }
 
 DocaConvertStage::~DocaConvertStage()
 {
+    free(fixed_size_list_cpu);
+    cudaFree(fixed_size_list);
     cudaStreamDestroy(m_stream);
 }
 
@@ -123,16 +130,22 @@ DocaConvertStage::source_type_t DocaConvertStage::on_raw_packet_message(sink_typ
 
     // LOG(WARNING) << "New RawPacketMessage with " << packet_count << " packets from queue id " << queue_idx;
 
+    // const auto t0 = now_ns();
+
     // gather header data
     auto header_src_ip_col = cudf::make_column_from_scalar(cudf::string_scalar("111.111.111.111"), packet_count);
     auto header_src_ip_addr = header_src_ip_col->mutable_view().data<uint8_t>();
     doca::gather_header_scalar(packet_count, pkt_addr_list, pkt_hdr_size_list, pkt_pld_size_list, header_src_ip_addr, m_stream_cpp);
 
+    // const auto t1 = now_ns();
+
     // gather payload data
     auto payload_col =
-        doca::gather_payload(packet_count, pkt_addr_list, pkt_hdr_size_list, pkt_pld_size_list, m_stream_cpp);
+        doca::gather_payload(packet_count, pkt_addr_list,
+                            pkt_hdr_size_list, pkt_pld_size_list,
+                            fixed_size_list, m_stream_cpp);
 
-    // const auto gather_payload_stop = now_ns();
+    // const auto t2 = now_ns();
 
     std::vector<std::unique_ptr<cudf::column>> gathered_columns;
     gathered_columns.emplace_back(std::move(header_src_ip_col));
@@ -141,7 +154,7 @@ DocaConvertStage::source_type_t DocaConvertStage::on_raw_packet_message(sink_typ
     // After this point buffers can be reused -> copies actual packets' data
     auto gathered_table = std::make_unique<cudf::table>(std::move(gathered_columns));
 
-    // const auto gather_table_meta = now_ns();
+    // const auto t3 = now_ns();
 
     auto gathered_metadata = cudf::io::table_metadata();
     gathered_metadata.schema_info.emplace_back("src_ip");
@@ -150,11 +163,11 @@ DocaConvertStage::source_type_t DocaConvertStage::on_raw_packet_message(sink_typ
     auto gathered_table_w_metadata =
         cudf::io::table_with_metadata{std::move(gathered_table), std::move(gathered_metadata)};
 
-    // const auto create_message_cpp = now_ns();
+    // const auto t4 = now_ns();
 
     auto meta = MessageMeta::create_from_cpp(std::move(gathered_table_w_metadata), 0);
 
-    // const auto gather_meta_stop = now_ns();
+    // const auto t5 = now_ns();
 
     cudaStreamSynchronize(m_stream_cpp);
 

morpheus/_lib/doca/src/doca_convert_kernel.cu

@@ -48,9 +48,23 @@ __global__ void _packet_gather_payload_kernel(
   uintptr_t*  packets_buffer,
   uint32_t* header_sizes,
   uint32_t* payload_sizes,
-  uint8_t*    payload_chars_out
+  uint8_t*  payload_chars_out
 )
 {
+  int pkt_idx = threadIdx.x;
+  int j = 0;
+
+  while (pkt_idx < packet_count) {
+    uint8_t* pkt_hdr_addr = (uint8_t*)(packets_buffer[pkt_idx] + header_sizes[pkt_idx]);
+    for (j = 0; j < payload_sizes[pkt_idx]; j++)
+      payload_chars_out[(MAX_PKT_SIZE * pkt_idx) + j] = pkt_hdr_addr[j];
+    for (; j < MAX_PKT_SIZE; j++)
+      payload_chars_out[(MAX_PKT_SIZE * pkt_idx) + j] = '\0';
+    pkt_idx += blockDim.x;
+  }
+
+#if 0
+
   // Specialize BlockScan for a 1D block of 128 threads of type int
   using BlockScan = cub::BlockScan<int32_t, THREADS_PER_BLOCK>;
   // Allocate shared memory for BlockScan
@@ -86,6 +100,7 @@ __global__ void _packet_gather_payload_kernel(
       //     packets_buffer[packet_idx]);
     }
   }
+#endif
 }
 
 __global__ void _packet_gather_header_kernel(
@@ -115,12 +130,13 @@ std::unique_ptr<cudf::column> gather_payload(
   uintptr_t*    packets_buffer,
   uint32_t*    header_sizes,
   uint32_t*    payload_sizes,
+  uint32_t*    fixed_size_list,
   rmm::cuda_stream_view stream,
   rmm::mr::device_memory_resource* mr)
 {
   auto [offsets_column, bytes] = cudf::detail::make_offsets_child_column(
-    payload_sizes,
-    payload_sizes + packet_count,
+    fixed_size_list,
+    fixed_size_list + packet_count,
     stream,
     mr
   );

morpheus/_lib/doca/src/doca_source.cpp

@@ -147,7 +147,7 @@ DocaSourceStage::subscriber_fn_t DocaSourceStage::build()
                 continue;
             }
 
-            const auto start_kernel = now_ns();
+            // const auto start_kernel = now_ns();
 
             // Assume MAX_QUEUE == 2
             morpheus::doca::packet_receive_kernel(m_rxq[0]->rxq_info_gpu(), m_rxq[1]->rxq_info_gpu(),
@@ -158,11 +158,11 @@ DocaSourceStage::subscriber_fn_t DocaSourceStage::build()
                                                   rstream);
             cudaStreamSynchronize(rstream);
 
-            const auto end_kernel = now_ns();
+            // const auto end_kernel = now_ns();
 
             for (int queue_idx = 0; queue_idx < MAX_QUEUE; queue_idx++) {
                 if (m_semaphore[queue_idx]->is_ready(sem_idx[queue_idx])) {
-                    const auto start_sem = now_ns();
+                    // const auto start_sem = now_ns();
                     // LOG(WARNING) << "CPU READY sem " << sem_idx[queue_idx] << " queue " << thread_idx << std::endl;
 
                     pkt_ptr = static_cast<struct packets_info*>(m_semaphore[queue_idx]->get_info_cpu(sem_idx[queue_idx]));
@@ -182,21 +182,21 @@ DocaSourceStage::subscriber_fn_t DocaSourceStage::build()
                                                                 true,
                                                                 queue_idx);
 
-                    const auto create_msg = now_ns();
+                    // const auto create_msg = now_ns();
 
                     output.on_next(std::move(meta));
 
                     m_semaphore[queue_idx]->set_free(sem_idx[queue_idx]);
                     sem_idx[queue_idx] = (sem_idx[queue_idx] + 1) % MAX_SEM_X_QUEUE;
 
-                    const auto end = now_ns();
+                    // const auto end = now_ns();
 
-                    LOG(WARNING) << "Queue " << queue_idx
-                                << " packets " << pkt_ptr->packet_count_out
-                                << " kernel time ns " << end_kernel - start_kernel
-                                << " Sem + msg ns " << create_msg - start_sem
-                                << " End ns " << end - create_msg
-                                << std::endl;
+                    // LOG(WARNING) << "Queue " << queue_idx
+                    //             << " packets " << pkt_ptr->packet_count_out
+                    //             << " kernel time ns " << end_kernel - start_kernel
+                    //             << " Sem + msg ns " << create_msg - start_sem
+                    //             << " End ns " << end - create_msg
+                    //             << std::endl;
                 }
             }
         }

morpheus/stages/doca/doca_convert_stage.py

@@ -53,7 +53,7 @@ def __init__(self, c: Config):
                                        "Ensure the DOCA components have been built and installed. Error message: ") +
                                       ex.msg) from ex
 
-        self._max_concurrent = c.num_threads
+        self._max_concurrent = 3
 
     @property
     def name(self) -> str:

morpheus/stages/doca/doca_source_stage.py

@@ -65,7 +65,6 @@ def __init__(
 
         self._batch_size = c.pipeline_batch_size
         self._input_count = None
-        self._max_concurrent = 1 # Only 1 thread is enough for 2 queues
         self._nic_pci_address = nic_pci_address
         self._gpu_pci_address = gpu_pci_address
         self._traffic_type = traffic_type.lower()
@@ -96,7 +95,8 @@ def _build_source(self, builder: mrc.Builder) -> mrc.SegmentObject:
                                            self._nic_pci_address,
                                            self._gpu_pci_address,
                                            self._traffic_type)
-            # node.launch_options.pe_count = self._max_concurrent
+            # Only 1 thread is enough for 2 queues
+            node.launch_options.pe_count = 1
             return node
 
         raise NotImplementedError("Does not support Python nodes")