Merge pull request #2 from mdemoret-nv/mdd_fix-doca-split

Mdd fix doca split

.devcontainer/Dockerfile

@@ -13,8 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-#FROM rapidsai/devcontainers:23.12-cpp-cuda12.1-mambaforge-ubuntu22.04 AS base
-
 FROM rapidsai/devcontainers:23.12-cpp-mambaforge-ubuntu22.04 AS base
 
 ENV MORPHEUS_SUPPORT_DOCA=ON

.devcontainer/devcontainer.json

@@ -96,11 +96,6 @@
             "source": "${localWorkspaceFolder}/.devcontainer/opt/morpheus",
             "target": "/opt/morpheus"
         }
-        // {
-        //     "type": "volume",
-        //     "source": "/dev/hugepages",
-        //     "target": "/dev/hugepages"
-        // }
     ],
 
     "features": {

examples/doca/run_tcp.py

@@ -16,7 +16,6 @@
 
 import click
 
-from morpheus._lib.messages import RawPacketMessage
 from morpheus.config import Config
 from morpheus.config import CppConfig
 from morpheus.config import PipelineModes
@@ -99,9 +98,6 @@ def run_pipeline(pipeline_batch_size, model_max_batch_size, model_fea_length, ou
 
     config.edge_buffer_size = 128
 
-    def count_raw_packets(message: RawPacketMessage):
-        return message.num
-
     pipeline = LinearPipeline(config)
 
     # add doca source stage
@@ -112,7 +108,7 @@ def count_raw_packets(message: RawPacketMessage):
     pipeline.add_stage(DeserializeStage(config))
     pipeline.add_stage(MonitorStage(config, description="Deserialize rate", unit='pkts'))
 
-    hashfile = '/workspace/models/training-tuning-scripts/sid-models/resources/bert-base-uncased-hash.txt'
+    hashfile = 'data/bert-base-uncased-hash.txt'
 
     # add preprocessing stage
     pipeline.add_stage(
@@ -127,13 +123,11 @@ def count_raw_packets(message: RawPacketMessage):
 
     # add inference stage
     pipeline.add_stage(
-        TritonInferenceStage(
-            config,
-            # model_name="sid-minibert-trt",
-            model_name="sid-minibert-onnx",
-            server_url="localhost:8000",
-            force_convert_inputs=True,
-            use_shared_memory=True))
+        TritonInferenceStage(config,
+                             model_name="sid-minibert-onnx",
+                             server_url="localhost:8000",
+                             force_convert_inputs=True,
+                             use_shared_memory=True))
 
     pipeline.add_stage(MonitorStage(config, description="Inference rate", unit='pkts'))
 

examples/doca/run_udp_convert.py

@@ -16,24 +16,17 @@
 
 import click
 
-from morpheus._lib.messages import RawPacketMessage
 from morpheus.config import Config
 from morpheus.config import CppConfig
 from morpheus.config import PipelineModes
 from morpheus.pipeline.linear_pipeline import LinearPipeline
 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.general.trigger_stage import TriggerStage
 from morpheus.utils.logger import configure_logging
 
 
 @click.command()
-@click.option(
-    "--out_file",
-    default="doca_output.csv",
-    help="File in which to store output",
-)
 @click.option(
     "--nic_addr",
     help="NIC PCI Address",
@@ -44,7 +37,7 @@
     help="GPU PCI Address",
     required=True,
 )
-def run_pipeline(out_file, nic_addr, gpu_addr):
+def run_pipeline(nic_addr, gpu_addr):
     # Enable the default logger
     configure_logging(log_level=logging.DEBUG)
 
@@ -61,7 +54,6 @@ def run_pipeline(out_file, nic_addr, gpu_addr):
 
     # 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(MonitorStage(config, description="DOCA GPUNetIO rate", unit='pkts'))
 

examples/doca/run_udp_raw.py

@@ -16,22 +16,17 @@
 
 import click
 
-from morpheus._lib.messages import RawPacketMessage
 from morpheus.config import Config
 from morpheus.config import CppConfig
 from morpheus.config import PipelineModes
+from morpheus.messages import RawPacketMessage
 from morpheus.pipeline.linear_pipeline import LinearPipeline
 from morpheus.stages.doca.doca_source_stage import DocaSourceStage
 from morpheus.stages.general.monitor_stage import MonitorStage
 from morpheus.utils.logger import configure_logging
 
 
 @click.command()
-@click.option(
-    "--out_file",
-    default="doca_output.csv",
-    help="File in which to store output",
-)
 @click.option(
     "--nic_addr",
     help="NIC PCI Address",
@@ -42,7 +37,7 @@
     help="GPU PCI Address",
     required=True,
 )
-def run_pipeline(out_file, nic_addr, gpu_addr):
+def run_pipeline(nic_addr, gpu_addr):
     # Enable the default logger
     configure_logging(log_level=logging.DEBUG)
 

examples/doca/vdb_realtime/README.md

@@ -32,7 +32,7 @@ wget https://github.com/milvus-io/milvus/releases/download/v2.3.3/milvus-standal
 Start Milvus
 
 ```bash
-sudo docker-compose up -d
+docker compose up -d
 ```
 
 ## Launch Triton Inference Server
@@ -66,9 +66,6 @@ pip install scapy
 sudo python3 examples/doca/vdb_realtime/sender/send.py
 ```
 
-Please modify inside the script all the 
-
-
 On the original machine, wait for the "Upload rate" to match the "DOCA GPUNetIO Source rate" and then press `Ctrl+C` to stop the script. The output should look like the following
 
 ```
@@ -97,9 +94,7 @@ export NGC_API_KEY="<YOUR_NGC_API>"
 
 Then install basic requirements:
 ```bash
-pip install langchain
-pip install sentence-transformers
-conda env update --solver=libmamba -n morpheus --file conda/environments/dev_cuda-121_arch-x86_64.yaml --prune
+conda env update --solver=libmamba -n morpheus --file conda/environments/examples_cuda-121_arch-x86_64.yaml --prune
 ```
 
 Run the RAG example to query the Milvus database:

examples/doca/vdb_realtime/sender/send.py

@@ -1,27 +1,42 @@
 #!/usr/bin/python
+# SPDX-FileCopyrightText: Copyright (c) 2024, 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.
 
 import glob
 import os
-import socket
-import sys
 
-from scapy.all import *
+from scapy.all import IP  # pylint: disable=no-name-in-module
+from scapy.all import UDP  # pylint: disable=no-name-in-module
+from scapy.all import RandShort
+from scapy.all import Raw
+from scapy.all import send
 
 
-def main(args):
+def main():
     os.chdir("dataset")
     for file in glob.glob("*.txt"):
-        fp = open(file, 'r')
-        while True:
-            content = fp.read(1024)
-            if not content:
-                break
-            pkt = IP(src="192.168.2.28", dst="192.168.2.27") / UDP(sport=RandShort(),
-                                                                   dport=5001) / Raw(load=content.encode('utf-8'))
-            print(pkt)
-            send(pkt, iface="enp202s0f0np0")
-            #sock.sendto(line.encode('utf-8'), (ip, port))
-        fp.close()
+        with open(file, 'r', encoding='utf-8') as fp:
+            while True:
+                content = fp.read(1024)
+                if not content:
+                    break
+                pkt = IP(src="192.168.2.28", dst="192.168.2.27") / UDP(sport=RandShort(),
+                                                                       dport=5001) / Raw(load=content.encode('utf-8'))
+                print(pkt)
+                send(pkt, iface="enp202s0f0np0")
 
 
-main(sys.argv)
+if __name__ == "__main__":
+    main()

examples/doca/vdb_realtime/vdb.py

@@ -17,13 +17,10 @@
 import click
 import pymilvus
 
-from morpheus._lib.messages import RawPacketMessage
 from morpheus.config import Config
 from morpheus.config import CppConfig
 from morpheus.config import PipelineModes
 from morpheus.pipeline.linear_pipeline import LinearPipeline
-from morpheus.service.vdb.milvus_vector_db_service import MilvusVectorDBService
-from morpheus.service.vdb.utils import VectorDBServiceFactory
 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
@@ -75,11 +72,6 @@ def build_milvus_service(embedding_size):
 
 
 @click.command()
-@click.option(
-    "--out_file",
-    default="doca_output.csv",
-    help="File in which to store output",
-)
 @click.option(
     "--nic_addr",
     help="NIC PCI Address",
@@ -90,7 +82,7 @@ def build_milvus_service(embedding_size):
     help="GPU PCI Address",
     required=True,
 )
-def run_pipeline(out_file, nic_addr, gpu_addr):
+def run_pipeline(nic_addr, gpu_addr):
     # Enable the default logger
     configure_logging(log_level=logging.DEBUG)
 
@@ -141,4 +133,4 @@ def run_pipeline(out_file, nic_addr, gpu_addr):
 
 
 if __name__ == "__main__":
-    run_pipeline()
+    run_pipeline()

examples/llm/rag/standalone_pipeline.py

@@ -14,8 +14,6 @@
 import logging
 import time
 
-import pandas as pd
-
 import cudf
 
 from morpheus.config import Config
@@ -48,8 +46,21 @@ def _build_engine(model_name: str, vdb_resource_name: str, llm_service: str, emb
 
     prompt = """You are a helpful assistant. Given the following background information:\n
 {% for c in contexts -%}
+{% if c.title %}
+Title: {{ c.title }}
+{% endif %}
+{% if c.summary %}
+Summary: {{ c.summary }}
+{% endif %}
+{% if c.page_content %}
+Text: {{ c.page_content }}
+{% endif %}
+{% if c.header %}
 Header: {{ c.header }}
+{% endif %}
+{% if c.data %}
 Data: {{ c.data }}
+{% endif %}
 {% endfor %}
 
 Please answer the following question: \n{{ query }}"""
@@ -136,4 +147,4 @@ def standalone(num_threads,
             logger.debug("Question:\n%s", row.questions)
             logger.debug("Response:\n%s", row.response)
 
-    return start_time
+    return start_time

examples/llm/vdb_upload/helper.py

@@ -107,8 +107,26 @@ def setup_filesystem_source(pipe: Pipeline, config: Config, source_name: str, fs
     return file_pipe
 
 
+# pylint: disable=unused-argument
 def setup_doca_source(pipe: Pipeline, config: Config, source_name: str, stage_config: typing.Dict[str, typing.Any]):
     """
+    Set up the DOCA source stage in the pipeline.
+
+    Parameters
+    ----------
+    pipe : Pipeline
+        The pipeline to which the DOCA source stage will be added.
+    config : Config
+        Configuration object for the pipeline.
+    source_name : str
+        The name of the DOCA source stage.
+    stage_config : typing.Dict[str, Any]
+        Configuration parameters for the DOCA source stage.
+
+    Returns
+    -------
+    SubPipeline
+        The sub-pipeline stage created for the DOCA source.
     """
     from morpheus.stages.doca.doca_convert_stage import DocaConvertStage
     from morpheus.stages.doca.doca_source_stage import DocaSourceStage
@@ -119,7 +137,7 @@ def setup_doca_source(pipe: Pipeline, config: Config, source_name: str, stage_co
                         stage_config.pop('nic_addr'),
                         stage_config.pop('gpu_addr'),
                         stage_config.pop('traffic_type')))
-    convert_stage = pipe.add_stage(DocaConvertStage(config, False))
+    convert_stage = pipe.add_stage(DocaConvertStage(config))
     deserialize_stage = pipe.add_stage(DeserializeStage(config))
 
     pipe.add_edge(source_stage, convert_stage)

examples/llm/vdb_upload/pipeline.py

@@ -120,4 +120,4 @@ def embedding_tensor_to_df(message: ControlMessage, *, embedding_tensor_name='pr
 
     pipe.run()
 
-    return start_time
+    return start_time

morpheus/_lib/cmake/libmorpheus.cmake

@@ -82,6 +82,7 @@ add_library(morpheus
   src/utilities/cudf_util.cpp
   src/utilities/cupy_util.cpp
   src/utilities/http_server.cpp
+  src/utilities/json_types.cpp
   src/utilities/matx_util.cu
   src/utilities/python_util.cpp
   src/utilities/string_util.cpp
@@ -123,7 +124,7 @@ target_include_directories(morpheus
 add_dependencies(morpheus ${cudf_helpers_target})
 
 # In debug mode, dont allow missing symbols
-# target_link_options(morpheus PUBLIC "$<$<CONFIG:Debug>:-Wl,--no-allow-shlib-undefined>")
+target_link_options(morpheus PUBLIC "$<$<CONFIG:Debug>:-Wl,--no-allow-shlib-undefined>")
 
 # Generates an include file for specifying external linkage since everything is hidden by default
 generate_export_header(morpheus

morpheus/_lib/doca/CMakeLists.txt

@@ -19,22 +19,16 @@ morpheus_configure_libbsd()
 set(doca_ROOT "/opt/mellanox/doca")
 find_package(doca REQUIRED)
 
-# set(CMAKE_BUILD_TYPE Debug)
-# set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -O0 -ggdb")
-# set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -O0 -ggdb")
-# set(CMAKE_CUDA_FLAGS "${CMAKE_CUDA_FLAGS} --generate-line-info ")
-# set(CMAKE_CUDA_FLAGS "${CMAKE_CUDA_FLAGS} -G")
-
 add_library(morpheus_doca
   # Keep these sorted!
   src/doca_context.cpp
+  src/doca_convert_kernel.cu
+  src/doca_convert.cpp
   src/doca_rx_pipe.cpp
   src/doca_rx_queue.cpp
   src/doca_semaphore.cpp
   src/doca_source_kernel.cu
   src/doca_source.cpp
-  src/doca_convert_kernel.cu
-  src/doca_convert.cpp
   src/rte_context.cpp
 )