Merge remote-tracking branch 'upstream/branch-24.10' into complete-re…

…move-multi-message

.pre-commit-config.yaml

@@ -19,6 +19,26 @@ repos:
         hooks:
             - id: rapids-dependency-file-generator
               args: ["--clean"]
+      - repo: https://github.com/PyCQA/isort
+        rev: 5.12.0
+        hooks:
+              - id: isort
+                args: ["--settings-file=./pyproject.toml"]
+                files: ^python/
+      - repo: https://github.com/PyCQA/flake8
+        rev: 6.1.0
+        hooks:
+              - id: flake8
+                entry: pflake8
+                additional_dependencies: [pyproject-flake8]
+                args: ["--config=./pyproject.toml"]
+                files: ^python/
+      - repo: https://github.com/google/yapf
+        rev: v0.40.2
+        hooks:
+              - id: yapf
+                args: ["--style", "./pyproject.toml"]
+                files: ^python/
 
 default_language_version:
       python: python3

ci/scripts/fix_all.sh

@@ -105,5 +105,5 @@ fi
 # Run yapf
 if [[ "${SKIP_YAPF}" == "" ]]; then
    echo "Running yapf..."
-   python3 -m yapf -i --style ${PY_CFG} -r ${PY_MODIFIED_FILES[@]}
+   python3 -m yapf -i --style ${PROJ_TOML} -r ${PY_MODIFIED_FILES[@]}
 fi

ci/scripts/python_checks.sh

@@ -26,10 +26,7 @@ LC_ALL=C.UTF-8
 LANG=C.UTF-8
 
 # Pre-populate the return values in case they are skipped
-ISORT_RETVAL=0
-FLAKE_RETVAL=0
 PYLINT_RETVAL=0
-YAPF_RETVAL=0
 
 get_modified_files ${PYTHON_FILE_REGEX} MORPHEUS_MODIFIED_FILES
 
@@ -41,45 +38,16 @@ if [[ -n "${MORPHEUS_MODIFIED_FILES}" ]]; then
       echo "  $f"
    done
 
-   if [[ "${SKIP_ISORT}" == "" ]]; then
-      ISORT_OUTPUT=`python3 -m isort --settings-file ${PROJ_TOML} --filter-files --check-only  ${MORPHEUS_MODIFIED_FILES[@]} 2>&1`
-      ISORT_RETVAL=$?
-   fi
-
    if [[ "${SKIP_PYLINT}" == "" ]]; then
       NUM_PROC=$(get_num_proc)
       PYLINT_OUTPUT=`pylint -j ${NUM_PROC} ${MORPHEUS_MODIFIED_FILES[@]} 2>&1`
       PYLINT_RETVAL=$?
    fi
 
-   if [[ "${SKIP_FLAKE}" == "" ]]; then
-      FLAKE_OUTPUT=`python3 -m flake8 --config ${PY_CFG} ${MORPHEUS_MODIFIED_FILES[@]} 2>&1`
-      FLAKE_RETVAL=$?
-   fi
-
-   if [[ "${SKIP_YAPF}" == "" ]]; then
-      # Run yapf. Will return 1 if there are any diffs
-      YAPF_OUTPUT=`python3 -m yapf --style ${PY_CFG} --diff ${MORPHEUS_MODIFIED_FILES[@]} 2>&1`
-      YAPF_RETVAL=$?
-   fi
-
 else
    echo "No modified Python files to check"
 fi
 
-# Output results if failure otherwise show pass
-if [[ "${SKIP_ISORT}" != "" ]]; then
-   echo -e "\n\n>>>> SKIPPED: isort check\n\n"
-elif [ "${ISORT_RETVAL}" != "0" ]; then
-   echo -e "\n\n>>>> FAILED: isort style check; begin output\n\n"
-   echo -e "${ISORT_OUTPUT}"
-   echo -e "\n\n>>>> FAILED: isort style check; end output\n\n" \
-           "To auto-fix many issues (not all) run:\n" \
-           "   ./ci/scripts/fix_all.sh\n\n"
-else
-  echo -e "\n\n>>>> PASSED: isort style check\n\n"
-fi
-
 if [[ "${SKIP_PYLINT}" != "" ]]; then
    echo -e "\n\n>>>> SKIPPED: pylint check\n\n"
 elif [ "${PYLINT_RETVAL}" != "0" ]; then
@@ -90,33 +58,7 @@ else
   echo -e "\n\n>>>> PASSED: pylint style check\n\n"
 fi
 
-if [[ "${SKIP_FLAKE}" != "" ]]; then
-   echo -e "\n\n>>>> SKIPPED: flake8 check\n\n"
-elif [ "${FLAKE_RETVAL}" != "0" ]; then
-   echo -e "\n\n>>>> FAILED: flake8 style check; begin output\n\n"
-   echo -e "${FLAKE_OUTPUT}"
-   echo -e "\n\n>>>> FAILED: flake8 style check; end output\n\n" \
-           "To auto-fix many issues (not all) run:\n" \
-           "   ./ci/scripts/fix_all.sh\n\n"
-else
-  echo -e "\n\n>>>> PASSED: flake8 style check\n\n"
-fi
-
-if [[ "${SKIP_YAPF}" != "" ]]; then
-   echo -e "\n\n>>>> SKIPPED: yapf check\n\n"
-elif [ "${YAPF_RETVAL}" != "0" ]; then
-   echo -e "\n\n>>>> FAILED: yapf style check; begin output\n\n"
-   echo -e "Incorrectly formatted files:"
-   YAPF_OUTPUT=`echo "${YAPF_OUTPUT}" | sed -nr 's/^\+\+\+ ([^ ]*) *\(reformatted\)$/\1/p'`
-   echo -e "${YAPF_OUTPUT}"
-   echo -e "\n\n>>>> FAILED: yapf style check; end output\n\n" \
-           "To auto-fix many issues (not all) run:\n" \
-           "   ./ci/scripts/fix_all.sh\n\n"
-else
-  echo -e "\n\n>>>> PASSED: yapf style check\n\n"
-fi
-
-RETVALS=(${ISORT_RETVAL} ${FLAKE_RETVAL} ${PYLINT_RETVAL} ${YAPF_RETVAL})
+RETVALS=(${PYLINT_RETVAL})
 IFS=$'\n'
 RETVAL=`echo "${RETVALS[*]}" | sort -nr | head -n1`
 

docs/source/developer_guide/guides/2_real_world_phishing.md

@@ -388,7 +388,7 @@ To start, we will need to instantiate and set a few attributes of the `Config` c
 config = Config()
 config.mode = PipelineModes.NLP
 
-config.num_threads = os.cpu_count()
+config.num_threads = len(os.sched_getaffinity(0))
 config.feature_length = model_fea_length
 
 with open(labels_file, encoding='UTF-8') as fh:
@@ -563,7 +563,7 @@ def run_pipeline(use_stage_function: bool,
     config.mode = PipelineModes.NLP
 
     # Set the thread count to match our cpu count
-    config.num_threads = os.cpu_count()
+    config.num_threads = len(os.sched_getaffinity(0))
     config.feature_length = model_fea_length
 
     with open(labels_file, encoding='UTF-8') as fh:

docs/source/developer_guide/guides/6_digital_fingerprinting_reference.md

@@ -41,7 +41,7 @@ from morpheus.cli.utils import load_labels_file
 CppConfig.set_should_use_cpp(False)
 
 config = Config()
-config.num_threads = os.cpu_count()
+config.num_threads = len(os.sched_getaffinity(0))
 config.ae = ConfigAutoEncoder()
 config.ae.feature_columns = load_labels_file(get_package_relative_file("data/columns_ae_azure.txt"))
 ```

examples/abp_pcap_detection/run.py

@@ -39,7 +39,7 @@
 @click.command()
 @click.option(
     "--num_threads",
-    default=os.cpu_count(),
+    default=len(os.sched_getaffinity(0)),
     type=click.IntRange(min=1),
     help="Number of internal pipeline threads to use.",
 )

examples/developer_guide/2_1_real_world_phishing/run.py

@@ -98,7 +98,7 @@ def run_pipeline(use_stage_function: bool,
     config.mode = PipelineModes.NLP
 
     # Set the thread count to match our cpu count
-    config.num_threads = os.cpu_count()
+    config.num_threads = len(os.sched_getaffinity(0))
     config.feature_length = model_fea_length
 
     with open(labels_file, encoding='UTF-8') as fh:

examples/developer_guide/2_2_rabbitmq/read_simple.py

@@ -38,7 +38,7 @@ def run_pipeline(use_source_function: bool):
     configure_logging(log_level=logging.DEBUG)
 
     config = Config()
-    config.num_threads = os.cpu_count()
+    config.num_threads = len(os.sched_getaffinity(0))
 
     # Create a linear pipeline object
     pipeline = LinearPipeline(config)

examples/developer_guide/2_2_rabbitmq/write_simple.py

@@ -32,7 +32,7 @@ def run_pipeline():
     input_file = os.path.join(root_dir, 'examples/data/email.jsonlines')
 
     config = Config()
-    config.num_threads = os.cpu_count()
+    config.num_threads = len(os.sched_getaffinity(0))
 
     # Create a linear pipeline object
     pipeline = LinearPipeline(config)

examples/developer_guide/4_rabbitmq_cpp_stage/README.md

@@ -18,7 +18,7 @@ limitations under the License.
 # Example RabbitMQ stages
 This example builds upon the `examples/developer_guide/2_2_rabbitmq` example adding a C++ implementation for the `RabbitMQSourceStage` along with adding package install scripts.
 
-This example adds two flags to the `read_simple.py` script. A `--use_cpp` flag which defaults to `True` and a `--num_threads` flag which defaults to the number of cores on the system as returned by `os.cpu_count()`.
+This example adds two flags to the `read_simple.py` script. A `--use_cpp` flag which defaults to `True` and a `--num_threads` flag which defaults to the number of cores on the system as returned by `len(os.sched_getaffinity(0))`.
 
 ## Supported Environments
 | Environment | Supported | Notes |

examples/developer_guide/4_rabbitmq_cpp_stage/src/read_simple.py

@@ -32,7 +32,7 @@
 @click.option('--use_cpp', default=True)
 @click.option(
     "--num_threads",
-    default=os.cpu_count(),
+    default=len(os.sched_getaffinity(0)),
     type=click.IntRange(min=1),
     help="Number of internal pipeline threads to use",
 )

examples/developer_guide/4_rabbitmq_cpp_stage/src/write_simple.py

@@ -32,7 +32,7 @@ def run_pipeline():
     input_file = os.path.join(root_dir, 'examples/data/email.jsonlines')
 
     config = Config()
-    config.num_threads = os.cpu_count()
+    config.num_threads = len(os.sched_getaffinity(0))
 
     # Create a linear pipeline object
     pipeline = LinearPipeline(config)

examples/digital_fingerprinting/production/grafana/run.py

@@ -245,7 +245,7 @@ def run_pipeline(train_users,
 
     CppConfig.set_should_use_cpp(False)
 
-    config.num_threads = os.cpu_count()
+    config.num_threads = len(os.sched_getaffinity(0))
 
     config.ae = ConfigAutoEncoder()
 

examples/digital_fingerprinting/production/morpheus/dfp/utils/config_generator.py

@@ -174,7 +174,7 @@ def generate_ae_config(source: str,
                        pipeline_batch_size: int = 0,
                        edge_buffer_size: int = 0,
                        use_cpp: bool = False,
-                       num_threads: int = os.cpu_count()):
+                       num_threads: int = len(os.sched_getaffinity(0))):
     config = Config()
 
     CppConfig.set_should_use_cpp(use_cpp)

examples/digital_fingerprinting/production/morpheus/dfp_azure_pipeline.py

@@ -233,7 +233,7 @@ def run_pipeline(train_users,
 
     CppConfig.set_should_use_cpp(False)
 
-    config.num_threads = os.cpu_count()
+    config.num_threads = len(os.sched_getaffinity(0))
 
     config.ae = ConfigAutoEncoder()
 

examples/digital_fingerprinting/production/morpheus/dfp_duo_pipeline.py

@@ -230,7 +230,7 @@ def run_pipeline(train_users,
 
     CppConfig.set_should_use_cpp(False)
 
-    config.num_threads = os.cpu_count()
+    config.num_threads = len(os.sched_getaffinity(0))
 
     config.ae = ConfigAutoEncoder()
 

examples/digital_fingerprinting/production/morpheus/notebooks/dfp_azure_inference.ipynb

@@ -219,7 +219,7 @@
     "\n",
     "CppConfig.set_should_use_cpp(False)\n",
     "\n",
-    "config.num_threads = os.cpu_count()\n",
+    "config.num_threads = len(os.sched_getaffinity(0))\n",
     "\n",
     "config.ae = ConfigAutoEncoder()\n",
     "\n",

examples/digital_fingerprinting/production/morpheus/notebooks/dfp_azure_training.ipynb

@@ -216,7 +216,7 @@
     "\n",
     "CppConfig.set_should_use_cpp(False)\n",
     "\n",
-    "config.num_threads = os.cpu_count()\n",
+    "config.num_threads = len(os.sched_getaffinity(0))\n",
     "\n",
     "config.ae = ConfigAutoEncoder()\n",
     "\n",

examples/digital_fingerprinting/production/morpheus/notebooks/dfp_duo_inference.ipynb

@@ -218,7 +218,7 @@
     "\n",
     "CppConfig.set_should_use_cpp(False)\n",
     "\n",
-    "config.num_threads = os.cpu_count()\n",
+    "config.num_threads = len(os.sched_getaffinity(0))\n",
     "\n",
     "config.ae = ConfigAutoEncoder()\n",
     "\n",

examples/digital_fingerprinting/production/morpheus/notebooks/dfp_duo_training.ipynb

@@ -217,7 +217,7 @@
     "\n",
     "CppConfig.set_should_use_cpp(False)\n",
     "\n",
-    "config.num_threads = os.cpu_count()\n",
+    "config.num_threads = len(os.sched_getaffinity(0))\n",
     "\n",
     "config.ae = ConfigAutoEncoder()\n",
     "\n",

examples/digital_fingerprinting/starter/run_cloudtrail_dfp.py

@@ -39,7 +39,7 @@
 @click.command()
 @click.option(
     "--num_threads",
-    default=os.cpu_count(),
+    default=len(os.sched_getaffinity(0)),
     type=click.IntRange(min=1),
     help="Number of internal pipeline threads to use",
 )

examples/digital_fingerprinting/visualization/dfp_viz_azure_pipeline.py

@@ -183,7 +183,7 @@ def run_pipeline(train_users,
 
     CppConfig.set_should_use_cpp(False)
 
-    config.num_threads = os.cpu_count()
+    config.num_threads = len(os.sched_getaffinity(0))
 
     config.ae = ConfigAutoEncoder()
 

examples/digital_fingerprinting/visualization/dfp_viz_duo_pipeline.py

@@ -186,7 +186,7 @@ def run_pipeline(train_users,
 
     CppConfig.set_should_use_cpp(False)
 
-    config.num_threads = os.cpu_count()
+    config.num_threads = len(os.sched_getaffinity(0))
 
     config.ae = ConfigAutoEncoder()
 

examples/doca/run_udp_convert.py

@@ -43,7 +43,7 @@
 )
 @click.option(
     "--num_threads",
-    default=os.cpu_count(),
+    default=len(os.sched_getaffinity(0)),
     type=click.IntRange(min=1),
     show_default=True,
     help="Number of internal pipeline threads to use.",

examples/gnn_fraud_detection_pipeline/run.py

@@ -38,7 +38,7 @@
 @click.command()
 @click.option(
     "--num_threads",
-    default=os.cpu_count(),
+    default=len(os.sched_getaffinity(0)),
     type=click.IntRange(min=1),
     help="Number of internal pipeline threads to use.",
 )

examples/llm/agents/run.py

@@ -27,7 +27,7 @@ def run():
 @run.command(help="Runs a simple finite pipeline with a single execution of a LangChain agent from a fixed input")
 @click.option(
     "--num_threads",
-    default=os.cpu_count(),
+    default=len(os.sched_getaffinity(0)),
     type=click.IntRange(min=1),
     help="Number of internal pipeline threads to use",
 )
@@ -67,7 +67,7 @@ def simple(**kwargs):
 @run.command(help="Runs a pipeline LangChain agents which pulls inputs from a Kafka message bus")
 @click.option(
     "--num_threads",
-    default=os.cpu_count(),
+    default=len(os.sched_getaffinity(0)),
     type=click.IntRange(min=1),
     help="Number of internal pipeline threads to use",
 )

examples/llm/completion/run.py

@@ -28,7 +28,7 @@ def run():
 @run.command()
 @click.option(
     "--num_threads",
-    default=os.cpu_count(),
+    default=len(os.sched_getaffinity(0)),
     type=click.IntRange(min=1),
     help="Number of internal pipeline threads to use",
 )

examples/llm/rag/run.py

@@ -27,7 +27,7 @@ def run():
 @run.command()
 @click.option(
     "--num_threads",
-    default=os.cpu_count(),
+    default=len(os.sched_getaffinity(0)),
     type=click.IntRange(min=1),
     help="Number of internal pipeline threads to use",
 )

examples/llm/vdb_upload/run.py

@@ -78,7 +78,7 @@ def run():
 )
 @click.option(
     "--num_threads",
-    default=os.cpu_count(),
+    default=len(os.sched_getaffinity(0)),
     type=click.IntRange(min=1),
     help="Number of internal pipeline threads to use",
 )

examples/log_parsing/run.py

@@ -35,7 +35,7 @@
 @click.command()
 @click.option(
     "--num_threads",
-    default=os.cpu_count(),
+    default=len(os.sched_getaffinity(0)),
     type=click.IntRange(min=1),
     help="Number of internal pipeline threads to use.",
 )

examples/ransomware_detection/run.py

@@ -41,7 +41,7 @@
 @click.option('--use_cpp', default=False, help="Enable C++ execution for this pipeline, currently this is unsupported.")
 @click.option(
     "--num_threads",
-    default=os.cpu_count(),
+    default=len(os.sched_getaffinity(0)),
     type=click.IntRange(min=1),
     help="Number of internal pipeline threads to use.",
 )