diff --git a/.gitignore b/.gitignore index c23e9a4..92a1fc0 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,11 @@ build*/ site/ bench-results/ +# ResNet inference example: locally generated dataset + model artifacts +# (built offline by applications/resnet50_inference/tools/, see the tutorial) +/data/ +/models/ + # tune_system.py default output pcie_schematic.png diff --git a/AGENTS.md b/AGENTS.md index 322c09b..9389e5e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,6 +18,7 @@ CMake options (full table in `docs/getting-started.md`): - `DAQIRI_ENGINE` — space-separated list of optional engines to compile. Valid values: `dpdk` (raw Ethernet) and `ibverbs` (RDMA/RoCE). Linux sockets (UDP/TCP) are always built in, so there is no `socket` value. Default is `"dpdk ibverbs"`. - `DAQIRI_BUILD_PYTHON` — builds `pybind11` bindings from `python/`. - `DAQIRI_BUILD_EXAMPLES` — builds the benchmark executables (default `ON`). +- `DAQIRI_BUILD_APPLICATIONS` — builds the end-to-end example applications under `applications/` (default `OFF`; requires TensorRT, e.g. the `BASE_IMAGE=torch` container). Currently builds `applications/resnet50_inference/` (DAQIRI → TensorRT ResNet inference). - `DAQIRI_ENABLE_OTEL_METRICS` — enables OpenTelemetry metrics instrumentation (default `OFF`). - `DAQIRI_REORDER_GPU_PROFILE` — enable CUDA event timing in the DPDK reorder kernels (off by default). - `DAQIRI_ENABLE_S3` — enable AWS SDK-backed asynchronous raw packet writes to S3 (off by default). @@ -55,6 +56,14 @@ Configs named `raw_rx_*` are RX-only — they initialize the RX path and wait fo When determining throughput for a benchmark use the `mlnx_perf` utility in the background to view transmit and receive rates. Using application run time with packet counts is usually not accurate enough due to startup inconsistencies. +### Example applications (`applications/`, opt-in) + +Beyond the benchmarks, end-to-end **example applications** live under `applications/` and build only with `-DDAQIRI_BUILD_APPLICATIONS=ON` (off by default; they pull in heavier optional dependencies). They are downstream `daqiri::daqiri` consumers — they reuse the bench scaffolding (`examples/raw_bench_common.*`, `bench_pipeline.cu`) but are not part of the benchmark table above. + +| Executable | Source | Typical config | +|---|---|---| +| `daqiri_resnet50_inference` | `applications/resnet50_inference/` | `configs/resnet50_wire_loopback.yaml` (example: real CIFAR-10 images → per-class feature stats), `configs/resnet50_bench_spark.yaml` (synthetic throughput sweep). DAQIRI raw/GPUDirect RX → GPU sequence-number reorder → ResNet feature extraction via **TensorRT** (FP16) → per-class mean-feature stats. Requires TensorRT (build the container with `BASE_IMAGE=torch`). Accepts `--seconds N --dataset --images-per-batch N --model resnet18|34|50|101|152`. Data prep + sweep tools in `applications/resnet50_inference/tools/`. See `docs/tutorials/daqiri-resnet-inference.md`. | + ## Formatting `clang-format` is required for contributions (CONTRIBUTING.md): @@ -112,7 +121,7 @@ The web docs live in `docs/` and are built with [MkDocs Material](https://squidf - `docs/concepts.md` — terminology glossary (stream types and endpoint URI schemes, GPUDirect, packet/burst/segment, flow/queue, memory region, zero-copy ownership, RX reorder). Meant to be opened in parallel with the rest of the docs. - `docs/api-reference/index.md` — API guide (6-step application lifecycle, configuration-first model) - `docs/api-reference/configuration.md`, `docs/api-reference/cpp.md`, `docs/api-reference/python.md` — YAML schema, C++ API, and Python bindings docs -- `docs/tutorials/` — tutorial walkthroughs (system config, config-file walkthrough, Holoscan integration) +- `docs/tutorials/` — tutorial walkthroughs (system config, config-file walkthrough, Holoscan integration, ResNet inference) - `docs/benchmarks/` — benchmark guide pages, surfaced as a top-level "Benchmarking" nav section in `mkdocs.yml` and the landing page (`docs/index.md`): - `docs/benchmarks/benchmarks.md` — overview and engine-selection decision tree - `docs/benchmarks/socket_benchmarking.md` — "Socket and RDMA Benchmarking" (TCP/UDP and RoCE/RDMA) diff --git a/CMakeLists.txt b/CMakeLists.txt index 6752096..6bc0312 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -32,6 +32,7 @@ endif() option(DAQIRI_BUILD_PYTHON "Build Python bindings" OFF) option(DAQIRI_BUILD_EXAMPLES "Build standalone examples" ON) +option(DAQIRI_BUILD_APPLICATIONS "Build example applications (requires TensorRT)" OFF) option(DAQIRI_ENABLE_OTEL_METRICS "Enable OpenTelemetry metrics instrumentation" OFF) set(DAQIRI_ENGINE "dpdk ibverbs" CACHE STRING "Optional engine implementations to build: dpdk, ibverbs (Linux sockets are always built in)") @@ -53,6 +54,10 @@ if(DAQIRI_BUILD_EXAMPLES) add_subdirectory(examples) endif() +if(DAQIRI_BUILD_APPLICATIONS) + add_subdirectory(applications) +endif() + install( EXPORT daqiriTargets NAMESPACE daqiri:: diff --git a/applications/CMakeLists.txt b/applications/CMakeLists.txt new file mode 100644 index 0000000..c54636c --- /dev/null +++ b/applications/CMakeLists.txt @@ -0,0 +1,19 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 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. + +# End-to-end example applications that build ON TOP of the DAQIRI library +# (downstream consumers). Gated by DAQIRI_BUILD_APPLICATIONS in the top-level +# CMakeLists because they pull in heavier optional dependencies (TensorRT). +add_subdirectory(resnet50_inference) diff --git a/applications/resnet50_inference/CMakeLists.txt b/applications/resnet50_inference/CMakeLists.txt new file mode 100644 index 0000000..7c24ba9 --- /dev/null +++ b/applications/resnet50_inference/CMakeLists.txt @@ -0,0 +1,118 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 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. + +cmake_minimum_required(VERSION 3.20) + +# Mirror the examples/ CUDA-arch auto-detect so Spark (sm_120/121), IGX (sm_89), +# and A100/H100 (sm_80/90) all build without manual flags. +if(NOT DEFINED CMAKE_CUDA_ARCHITECTURES) + set(CMAKE_CUDA_ARCHITECTURES 80 90) + set(DAQIRI_APP_USING_DEFAULT_CUDA_ARCHITECTURES ON) +endif() + +project(daqiri_resnet50_inference LANGUAGES CXX CUDA) + +find_package(CUDAToolkit REQUIRED) +find_package(Threads REQUIRED) + +if(DAQIRI_APP_USING_DEFAULT_CUDA_ARCHITECTURES AND + CMAKE_CUDA_COMPILER_VERSION VERSION_GREATER_EQUAL 13.0) + list(APPEND CMAKE_CUDA_ARCHITECTURES 121) +endif() +set(DAQIRI_APP_CUDA_ARCHITECTURES "${CMAKE_CUDA_ARCHITECTURES}") + +# --- TensorRT discovery (cross-platform). NGC PyTorch (BASE_IMAGE=torch) ships +# TensorRT with headers under /usr/include and libs under +# /usr/lib/{x86_64,aarch64}-linux-gnu. The CMake-find machinery for TRT is +# inconsistent across versions, so locate the header + libs manually. If TRT is +# absent we warn and skip the target rather than failing the whole build. +find_path(TRT_INCLUDE_DIR NvInfer.h + HINTS /usr/include /usr/include/x86_64-linux-gnu /usr/include/aarch64-linux-gnu) +find_library(TRT_NVINFER_LIB nvinfer + HINTS /usr/lib/aarch64-linux-gnu /usr/lib/x86_64-linux-gnu /usr/lib) +find_library(TRT_NVONNXPARSER_LIB nvonnxparser + HINTS /usr/lib/aarch64-linux-gnu /usr/lib/x86_64-linux-gnu /usr/lib) +if(NOT TRT_INCLUDE_DIR OR NOT TRT_NVINFER_LIB OR NOT TRT_NVONNXPARSER_LIB) + message(WARNING + "TensorRT not found (NvInfer.h / libnvinfer / libnvonnxparser); skipping " + "resnet50_inference application. Build the container with BASE_IMAGE=torch " + "to get TensorRT. INCLUDE=${TRT_INCLUDE_DIR} NVINFER=${TRT_NVINFER_LIB} " + "NVONNX=${TRT_NVONNXPARSER_LIB}") + return() +endif() +message(STATUS "TensorRT include: ${TRT_INCLUDE_DIR}") +message(STATUS "TensorRT libs: ${TRT_NVINFER_LIB} ${TRT_NVONNXPARSER_LIB}") + +# yaml-cpp target: set by the top-level/src CMake when built in-tree; mirror the +# examples/ resolution block for a standalone application build. +if(NOT DAQIRI_YAML_TARGET) + find_package(yaml-cpp QUIET) + if(TARGET yaml-cpp::yaml-cpp) + set(DAQIRI_YAML_TARGET yaml-cpp::yaml-cpp) + elseif(TARGET yaml-cpp) + set(DAQIRI_YAML_TARGET yaml-cpp) + else() + message(FATAL_ERROR "yaml-cpp not found; install yaml-cpp-dev") + endif() +endif() + +set(DAQIRI_EXAMPLES_DIR "${CMAKE_SOURCE_DIR}/examples") + +# Reuse the bench scaffolding: config parsers, thread affinity, UDP header +# builders, wait_for_stop, and the ReorderPipeline. raw_bench_common.cpp's +# rx_count_worker references GpuWorkload + ReorderPipeline, so we also compile +# bench_workload.cu + bench_pipeline.cu and link the math libs (same as the +# socket bench). +add_executable(daqiri_resnet50_inference + main.cpp + app_config.cpp + pcap_replayer.cpp + feature_sink.cpp + inference_pipeline.cu + trt_runner.cu + ${DAQIRI_EXAMPLES_DIR}/raw_bench_common.cpp + ${DAQIRI_EXAMPLES_DIR}/raw_bench_common_cuda.cu + ${DAQIRI_EXAMPLES_DIR}/bench_workload.cu + ${DAQIRI_EXAMPLES_DIR}/bench_pipeline.cu) + +target_compile_features(daqiri_resnet50_inference PRIVATE cxx_std_17) +target_include_directories(daqiri_resnet50_inference PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR} + ${DAQIRI_EXAMPLES_DIR} + ${CMAKE_SOURCE_DIR} + ${TRT_INCLUDE_DIR}) +target_link_libraries(daqiri_resnet50_inference PRIVATE + daqiri::daqiri + ${DAQIRI_YAML_TARGET} + CUDA::cudart + CUDA::cufft + CUDA::cublas + Threads::Threads + ${TRT_NVINFER_LIB} + ${TRT_NVONNXPARSER_LIB}) +set_target_properties(daqiri_resnet50_inference PROPERTIES + CUDA_ARCHITECTURES "${DAQIRI_APP_CUDA_ARCHITECTURES}" + BUILD_RPATH "$ORIGIN/../../src;$ORIGIN/../../src/third_party/yaml-cpp") + +set(DAQIRI_RESNET_CONFIGS + configs/resnet50_wire_loopback.yaml + configs/resnet50_bench_spark.yaml) +foreach(cfg IN LISTS DAQIRI_RESNET_CONFIGS) + configure_file(${CMAKE_CURRENT_SOURCE_DIR}/${cfg} + ${CMAKE_CURRENT_BINARY_DIR}/${cfg} COPYONLY) +endforeach() + +install(TARGETS daqiri_resnet50_inference RUNTIME DESTINATION bin) +install(FILES ${DAQIRI_RESNET_CONFIGS} DESTINATION bin/resnet50_inference) diff --git a/applications/resnet50_inference/README.md b/applications/resnet50_inference/README.md new file mode 100644 index 0000000..dd3cf12 --- /dev/null +++ b/applications/resnet50_inference/README.md @@ -0,0 +1,139 @@ + + +# DAQIRI → TensorRT ResNet inference example + +End-to-end demo wiring DAQIRI packet ingestion into a GPU inference pipeline +(GitHub issue #73): + +``` +received packets (raw / DPDK GPUDirect) + → GPU sequence-number reorder (image reassembly) + → ResNet feature extraction (TensorRT, FP16) + → per-class mean-feature stats (example mode) +``` + +Packets DMA straight into GPU memory, a CUDA kernel reorders each image's +packets by sequence number into a contiguous FP32 NCHW tensor, TensorRT runs a +ResNet feature extractor on a batch of those tensors, and the feature vectors +are summarized per class — all without a host bounce on the data path. + +The example runs over the **DGX Spark physical p0→p1 cabled loopback** (the same +two-physical-port DPDK loopback the performance report uses), so it exercises the +real NIC RX path rather than a software shortcut. The build, CMake target, and +CUDA-arch handling are platform-agnostic (x86_64 + aarch64 TensorRT discovery), +so the same code builds on IGX and RTX Pro servers. + +## Prerequisites + +Build the project container with TensorRT (and torch / torchvision for the data +prep), which the `torch` base image provides: + +```bash +BASE_IMAGE=torch BASE_TARGET=dpdk DAQIRI_ENGINE="dpdk ibverbs" scripts/build-container.sh +``` + +## Build + +Enable the applications tree (off by default; it requires TensorRT): + +```bash +cmake -S . -B build -DCMAKE_BUILD_TYPE=Release -DBUILD_SHARED_LIBS=ON \ + -DDAQIRI_ENGINE="dpdk ibverbs" -DDAQIRI_BUILD_APPLICATIONS=ON +cmake --build build -j +cmake --install build --prefix /opt/daqiri +``` + +If TensorRT is not found the application is skipped with a CMake warning (the +rest of the build is unaffected). + +## Prepare the model and dataset (offline, one-time) + +```bash +# 1. Export a ResNet feature extractor (final FC stripped) to ONNX. +python3 applications/resnet50_inference/tools/export_resnet_onnx.py \ + --model resnet50 --output models/resnet50_features.onnx --check + +# 2. Packetize CIFAR-10 into a replayable pcap (+ a labels sidecar). +python3 applications/resnet50_inference/tools/prepare_cifar10_pcap.py \ + --num-images 256 --out data/cifar10_resnet.pcap +``` + +The TensorRT engine itself is built and cached on the first run +(`models/resnet50_features.fp16.engine`) and loaded from cache afterwards. + +## Run (example mode — real images + per-class stats) + +Bring the wire loopback's network namespaces **down** (they hide the ports from +DPDK) and export the RX port MAC, exactly as for the raw/DPDK bench: + +```bash +./scripts/setup_spark_wire_loopback_netns.sh down +export ETH_DST_ADDR=$(cat /sys/class/net//address) # RX port (p1) +``` + +Fill the `` / `` placeholders in +`configs/resnet50_wire_loopback.yaml` (TX = p0, RX = p1), then: + +```bash +./build/applications/resnet50_inference/daqiri_resnet50_inference \ + ./build/applications/resnet50_inference/configs/resnet50_wire_loopback.yaml \ + --dataset data/cifar10_resnet.pcap --seconds 10 +``` + +Expected output: the engine builds + caches on the first run (loads from cache +on the next), a couple of sample feature vectors print, and at shutdown a +per-class mean-feature summary prints. Confirm nonzero RX with `mlnx_perf`. + +Example mode replays the dataset **once** by default (use `--loop` to repeat): +the whole dataset is buffered in the RX ring, so even though ResNet inference is +slower than line rate no packets drop and the in-order, drop-free image +reassembly (and label mapping) stays correct. Keep +`num_images * packets_per_image < num_bufs` (the default 256 × 84 = 21 504 fits +the 51 200-buffer ring). + +## Run (benchmark mode — synthetic, throughput sweep) + +```bash +export ETH_DST_ADDR=$(cat /sys/class/net//address) +BUILD_DIR=./build ./applications/resnet50_inference/tools/run_resnet_bench.sh +``` + +Sweeps ResNet-18/34/50/101/152 (FP16) and writes `resnet-bench.csv` (img/s per +model). No dataset is needed; the TX synthesizes frames so this measures the +full reorder + inference receive-path throughput. + +## How it works + +See the tutorial `docs/tutorials/daqiri-resnet-inference.md` for the CUDA-event +buffer logic (how bursts are ingested and freed without stalling the GPU) and a +flow diagram. The short version: + +``` +p0 --cable--> p1 -> get_rx_burst -> [reorder-scatter into image buffer (stream s)] + -> [D2D image -> NCHW batch + input_ready] -> [TrtRunner enqueueV3(s) + release_evt] + -> [D2H + d2h_event] -> FeatureSink -> per-class stats / counts + | + cudaStreamSynchronize(s) BEFORE free_all_packets_and_burst_rx +``` + +One CUDA stream serializes reorder → batch-copy → inference, so no explicit sync +is needed between them; the only barrier is a `cudaStreamSynchronize` before each +burst is freed, because the reorder kernel reads the burst's device pointers. + +## Files + +| File | Role | +|---|---| +| `main.cpp` | arg parse, `daqiri_init`, spawn TX + RX threads | +| `app_config.{h,cpp}` | YAML parse + derived reorder geometry | +| `pcap_replayer.{h,cpp}` | pcap reader, dst-MAC patch, TX worker, synthetic frames | +| `trt_runner.{hpp,cu}` | TRT engine build/cache + FP16 dynamic-batch inference | +| `inference_pipeline.{h,cu}` | RX loop: reorder → NCHW batch → TRT | +| `feature_sink.{h,cpp}` | per-class mean-feature stats / throughput | +| `tools/export_resnet_onnx.py` | torchvision ResNet → ONNX (feature extractor) | +| `tools/prepare_cifar10_pcap.py` | CIFAR-10 → preprocessed framed pcap + labels | +| `tools/run_resnet_bench.sh` | model-size throughput sweep → CSV | +| `configs/*.yaml` | wire-loopback example + benchmark configs | diff --git a/applications/resnet50_inference/app_config.cpp b/applications/resnet50_inference/app_config.cpp new file mode 100644 index 0000000..00a497a --- /dev/null +++ b/applications/resnet50_inference/app_config.cpp @@ -0,0 +1,85 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 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. + */ + +#include "app_config.h" + +#include +#include +#include + +namespace daqiri::apps::resnet { + +AppConfig AppConfig::from_yaml(const YAML::Node& root) { + AppConfig cfg; + + const auto rx_cfgs = daqiri::bench::parse_rx_configs(root); + const auto tx_cfgs = daqiri::bench::parse_tx_configs(root); + if (rx_cfgs.empty() || tx_cfgs.empty()) { + throw std::runtime_error("config must define one bench_rx and one bench_tx interface"); + } + cfg.rx = rx_cfgs.front(); + cfg.tx = tx_cfgs.front(); + cfg.trt = TrtRunnerConfig::from_yaml(root); + + const auto& reorder = root["reorder"]; + if (reorder && reorder["out_payload_len"]) { + cfg.out_payload_len = reorder["out_payload_len"].as(); + } + if (reorder && reorder["images_per_batch"]) { + cfg.images_per_batch = reorder["images_per_batch"].as(); + } else if (root["inference"] && root["inference"]["images_per_batch"]) { + cfg.images_per_batch = root["inference"]["images_per_batch"].as(); + } + + // Image size and reorder geometry derived from the model input + chunk size. + cfg.image_bytes = + static_cast(cfg.trt.channels) * cfg.trt.height * cfg.trt.width * sizeof(float); + if (cfg.out_payload_len == 0 || cfg.image_bytes % cfg.out_payload_len != 0) { + throw std::runtime_error("reorder.out_payload_len (" + std::to_string(cfg.out_payload_len) + + ") must evenly divide the image size (" + + std::to_string(cfg.image_bytes) + + " bytes = channels*height*width*4); pick a clean divisor so images " + "reassemble with no padding"); + } + cfg.packets_per_image = cfg.image_bytes / cfg.out_payload_len; + // The seq number occupies the first 4 payload bytes; pixels start after it. + cfg.seq_bit_offset = static_cast(cfg.tx.header_size * 8u); + cfg.payload_byte_offset = cfg.tx.header_size + 4u; + cfg.seq_bit_width = 32; + + if (cfg.images_per_batch == 0) cfg.images_per_batch = 1; + cfg.images_per_batch = + std::min(cfg.images_per_batch, static_cast(cfg.trt.opt_max)); + + const auto& dataset = root["dataset"]; + if (dataset && dataset["pcap_path"]) { + cfg.dataset_pcap = dataset["pcap_path"].as(); + } + if (dataset && dataset["labels_path"]) { + cfg.labels_path = dataset["labels_path"].as(); + } + + const auto& stats = root["stats"]; + if (stats) { + if (stats["enabled"]) cfg.stats_enabled = stats["enabled"].as(); + if (stats["top_k"]) cfg.stats_top_k = stats["top_k"].as(); + } + + return cfg; +} + +} // namespace daqiri::apps::resnet diff --git a/applications/resnet50_inference/app_config.h b/applications/resnet50_inference/app_config.h new file mode 100644 index 0000000..a5737af --- /dev/null +++ b/applications/resnet50_inference/app_config.h @@ -0,0 +1,72 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 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 + +#include +#include + +#include "raw_bench_common.h" // daqiri::bench::RawBench{Rx,Tx}Config + parsers +#include "trt_runner.hpp" + +namespace daqiri::apps::resnet { + +// Resolved configuration for the ResNet inference application. The rx/tx blocks +// reuse the bench parsers; the inference/reorder/dataset/stats blocks are +// app-specific. Geometry (packets_per_image, offsets) is derived from the model +// input dims and the per-packet chunk size so the runtime stays pure +// reorder -> infer. +struct AppConfig { + daqiri::bench::RawBenchRxConfig rx; + daqiri::bench::RawBenchTxConfig tx; + TrtRunnerConfig trt; + + // Derived image / reorder geometry. + uint32_t image_bytes = 0; // channels * height * width * sizeof(float) + uint32_t out_payload_len = 7168; // image-pixel bytes carried per packet + uint32_t packets_per_image = 0; // image_bytes / out_payload_len (must divide) + uint32_t payload_byte_offset = 0; // header_size + seq prefix (4 bytes) + uint16_t seq_bit_offset = 0; // header_size * 8 + uint8_t seq_bit_width = 32; + uint32_t images_per_batch = 32; // TRT dynamic batch (clamped to opt_max) + + // Example mode (dataset replay). Empty pcap path => synthetic benchmark mode. + std::string dataset_pcap; + std::string labels_path; // dataset_pcap + ".labels" (resolved if empty) + + // Per-class mean-feature stats (example mode only). + bool stats_enabled = true; + int stats_top_k = 8; + + // Bytes of a single frame on the wire: header + 4-byte seq prefix + chunk. + uint32_t frame_bytes() const { + return tx.header_size + 4u + out_payload_len; + } + + // True when a dataset pcap was supplied (drives real images + per-class stats). + bool example_mode() const { + return !dataset_pcap.empty(); + } + + // Parse the YAML config. Throws YAML::Exception / std::runtime_error on a + // malformed or geometrically-inconsistent config. + static AppConfig from_yaml(const YAML::Node& root); +}; + +} // namespace daqiri::apps::resnet diff --git a/applications/resnet50_inference/configs/resnet50_bench_spark.yaml b/applications/resnet50_inference/configs/resnet50_bench_spark.yaml new file mode 100644 index 0000000..a1964b5 --- /dev/null +++ b/applications/resnet50_inference/configs/resnet50_bench_spark.yaml @@ -0,0 +1,103 @@ +%YAML 1.2 +--- +# DAQIRI -> TensorRT ResNet inference (benchmark mode), DGX Spark physical +# p0->p1 cabled loopback. No dataset: the TX synthesizes frames so this measures +# the full reorder + inference receive-path throughput. Swept across model sizes +# by tools/run_resnet_bench.sh (which sets --model and --images-per-batch). +# Replace placeholders and export ETH_DST_ADDR as in the example +# config; the dq_wire_* netns must be DOWN. +daqiri: + cfg: + version: 1 + stream_type: "raw" + master_core: 3 + debug: false + log_level: "info" + loopback: "" + + memory_regions: + - name: "Data_TX_GPU" + kind: "host_pinned" + affinity: 0 + num_bufs: 51200 + buf_size: 8064 + - name: "Data_RX_GPU" + kind: "host_pinned" + affinity: 0 + num_bufs: 51200 + buf_size: 8064 + + interfaces: + - name: "tx_port" + address: + tx: + queues: + - name: "tx_q_0" + id: 0 + batch_size: 10240 + cpu_core: 11 + memory_regions: + - "Data_TX_GPU" + offloads: + - "tx_eth_src" + - name: "rx_port" + address: + rx: + flow_isolation: true + queues: + - name: "rq_q_0" + id: 0 + cpu_core: 9 + batch_size: 10240 + # Flush a partial (< batch_size) burst after this idle gap so a finite + # run's tail is delivered (the poller otherwise waits to fill batch_size). + timeout_us: 2000 + memory_regions: + - "Data_RX_GPU" + flows: + - name: "flow_0" + id: 0 + action: + type: queue + id: 0 + match: + udp_src: 4096 + udp_dst: 4096 + +bench_rx: +- interface_name: "rx_port" + cpu_core: 8 + +bench_tx: +- interface_name: "tx_port" + cpu_core: 10 + batch_size: 10240 + payload_size: 7172 + header_size: 64 + eth_dst_addr: + ip_src_addr: <1.2.3.4> + ip_dst_addr: <5.6.7.8> + udp_src_port: 4096 + udp_dst_port: 4096 + +reorder: + out_payload_len: 7168 + images_per_batch: 32 + +inference: + onnx_path: "models/resnet50_features.onnx" + engine_path: "models/resnet50_features.fp16.engine" + enable_fp16: true + channels: 3 + height: 224 + width: 224 + feature_dim: 2048 + opt_min: 1 + opt_avg: 32 + opt_max: 256 + input_name: "input" + output_name: "features" + +# No dataset block => synthetic benchmark mode. +stats: + enabled: false diff --git a/applications/resnet50_inference/configs/resnet50_wire_loopback.yaml b/applications/resnet50_inference/configs/resnet50_wire_loopback.yaml new file mode 100644 index 0000000..ca56491 --- /dev/null +++ b/applications/resnet50_inference/configs/resnet50_wire_loopback.yaml @@ -0,0 +1,113 @@ +%YAML 1.2 +--- +# DAQIRI -> TensorRT ResNet inference (example mode), DGX Spark physical p0->p1 +# cabled loopback. Replace every placeholder for your system and +# export ETH_DST_ADDR=$(cat /sys/class/net//address) so the replayed +# frames carry the RX port (p1) MAC. The dq_wire_* netns must be DOWN (it +# captures the ports and hides them from DPDK). See the README / the +# "Performance: DGX Spark" report for the full loopback setup. +daqiri: + cfg: + version: 1 + stream_type: "raw" + master_core: 3 + debug: false + log_level: "info" + loopback: "" + + memory_regions: + - name: "Data_TX_GPU" + kind: "host_pinned" + affinity: 0 + num_bufs: 51200 + buf_size: 8064 + - name: "Data_RX_GPU" + kind: "host_pinned" + affinity: 0 + num_bufs: 51200 + buf_size: 8064 + + interfaces: + - name: "tx_port" # physical port p0 + address: + tx: + queues: + - name: "tx_q_0" + id: 0 + batch_size: 10240 + cpu_core: 11 + memory_regions: + - "Data_TX_GPU" + offloads: + - "tx_eth_src" + - name: "rx_port" # physical port p1 (the loopback sink) + address: + rx: + flow_isolation: true + queues: + - name: "rq_q_0" + id: 0 + cpu_core: 9 + batch_size: 10240 + # Flush a partial (< batch_size) burst after this idle gap so a finite + # run's tail is delivered (the poller otherwise waits to fill batch_size; + # here that would strand the dataset's last ~1024 packets / 13 images). + timeout_us: 2000 + memory_regions: + - "Data_RX_GPU" + flows: + - name: "flow_0" + id: 0 + action: + type: queue + id: 0 + match: + udp_src: 4096 + udp_dst: 4096 + +bench_rx: +- interface_name: "rx_port" + cpu_core: 8 + +bench_tx: +- interface_name: "tx_port" + cpu_core: 10 + batch_size: 10240 + payload_size: 7172 # 4-byte seq prefix + 7168-byte image chunk + header_size: 64 + eth_dst_addr: # overridden by ETH_DST_ADDR at run time + ip_src_addr: <1.2.3.4> + ip_dst_addr: <5.6.7.8> + udp_src_port: 4096 + udp_dst_port: 4096 + +# Image reassembly geometry. 84 packets x 7168 pixel-bytes = 602112 bytes = +# 3 x 224 x 224 x 4 (FP32 NCHW), so images reassemble with no padding. +reorder: + out_payload_len: 7168 + images_per_batch: 32 + +# TensorRT engine. Run tools/export_resnet_onnx.py first; the engine is built +# and cached on the first launch, then loaded from cache afterwards. +inference: + onnx_path: "models/resnet50_features.onnx" + engine_path: "models/resnet50_features.fp16.engine" + enable_fp16: true + channels: 3 + height: 224 + width: 224 + feature_dim: 2048 + opt_min: 1 + opt_avg: 32 + opt_max: 256 + input_name: "input" + output_name: "features" + +# Real CIFAR-10 images for the example. Produced by tools/prepare_cifar10_pcap.py, +# which also writes ".labels". +dataset: + pcap_path: "data/cifar10_resnet.pcap" + +stats: + enabled: true + top_k: 8 diff --git a/applications/resnet50_inference/feature_sink.cpp b/applications/resnet50_inference/feature_sink.cpp new file mode 100644 index 0000000..287e524 --- /dev/null +++ b/applications/resnet50_inference/feature_sink.cpp @@ -0,0 +1,117 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 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. + */ + +#include "feature_sink.h" + +#include +#include +#include +#include +#include +#include + +namespace daqiri::apps::resnet { + +std::vector load_labels(const std::string& path) { + std::vector labels; + std::ifstream f(path); + if (!f) { + std::cerr << "load_labels: cannot open " << path << "\n"; + return labels; + } + int v; + while (f >> v) labels.push_back(v); + std::cerr << "load_labels: " << labels.size() << " labels from " << path << "\n"; + return labels; +} + +FeatureSink::FeatureSink(bool example_mode, int feature_dim, int top_k, std::vector labels) + : example_mode_(example_mode), + feature_dim_(feature_dim), + top_k_(std::min(top_k, feature_dim)), + labels_(std::move(labels)) { + if (example_mode_ && !labels_.empty()) { + num_classes_ = *std::max_element(labels_.begin(), labels_.end()) + 1; + class_sum_.assign(static_cast(num_classes_) * feature_dim_, 0.0); + class_count_.assign(num_classes_, 0); + } +} + +void FeatureSink::consume(const float* host_features, uint32_t n) { + ++batches_; + if (!example_mode_ || labels_.empty()) { + images_ += n; + return; + } + + for (uint32_t i = 0; i < n; ++i) { + const float* row = host_features + static_cast(i) * feature_dim_; + const int label = labels_[static_cast(images_ % labels_.size())]; + if (label >= 0 && label < num_classes_) { + double* acc = class_sum_.data() + static_cast(label) * feature_dim_; + for (int d = 0; d < feature_dim_; ++d) acc[d] += row[d]; + ++class_count_[label]; + } + ++images_; + + // Print a couple of raw feature vectors so the run shows concrete output. + if (samples_printed_ < 2) { + std::ostringstream os; + os << " feature[image " << (images_ - 1) << ", class " << label << "] = ["; + const int show = std::min(top_k_, feature_dim_); + os << std::fixed << std::setprecision(4); + for (int d = 0; d < show; ++d) os << (d ? ", " : "") << row[d]; + os << ", ...] (dim=" << feature_dim_ << ")"; + std::cerr << os.str() << "\n"; + ++samples_printed_; + } + } +} + +void FeatureSink::log_final_summary(double seconds) const { + const double imgs_per_s = seconds > 0 ? static_cast(images_) / seconds : 0.0; + std::cerr << "\n=== ResNet inference summary ===\n"; + std::cerr << "images=" << images_ << " batches=" << batches_ << " seconds=" << std::fixed + << std::setprecision(2) << seconds << " => " << imgs_per_s << " img/s\n"; + + if (!example_mode_ || labels_.empty() || num_classes_ == 0) return; + + std::cerr << "\nPer-class mean-feature stats (first " << top_k_ + << " dims + L2 norm of the mean vector):\n"; + for (int c = 0; c < num_classes_; ++c) { + const uint64_t cnt = class_count_[c]; + if (cnt == 0) continue; + const double* acc = class_sum_.data() + static_cast(c) * feature_dim_; + double norm = 0.0; + for (int d = 0; d < feature_dim_; ++d) { + const double mean = acc[d] / static_cast(cnt); + norm += mean * mean; + } + std::ostringstream os; + os << " class " << std::setw(2) << c << " (n=" << std::setw(6) << cnt << "): mean=[" + << std::fixed << std::setprecision(4); + for (int d = 0; d < top_k_; ++d) { + os << (d ? ", " : "") << acc[d] / static_cast(cnt); + } + os << ", ...] |mean|=" << std::sqrt(norm); + std::cerr << os.str() << "\n"; + } + std::cerr << "(Distinct per-class mean vectors indicate ResNet separates the " + "classes in latent space.)\n"; +} + +} // namespace daqiri::apps::resnet diff --git a/applications/resnet50_inference/feature_sink.h b/applications/resnet50_inference/feature_sink.h new file mode 100644 index 0000000..17d5533 --- /dev/null +++ b/applications/resnet50_inference/feature_sink.h @@ -0,0 +1,66 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 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 +#include +#include + +namespace daqiri::apps::resnet { + +// Load a one-class-id-per-line labels sidecar (written by prepare_cifar10_pcap.py). +// Returns an empty vector on failure. +std::vector load_labels(const std::string& path); + +// Consumes host-resident ResNet feature vectors. In benchmark mode it just +// counts (throughput). In example mode it accumulates a per-class mean feature +// vector (label per image comes from the dataset labels sidecar, indexed by a +// monotonic image counter; the wire loopback is drop-free and in-order, so +// delivery order matches send order) and prints a few sample vectors. PCA is +// intentionally not implemented; per-class mean-feature stats are the cheap, +// dependency-free readout that shows the latent space separating by class. +class FeatureSink { + public: + FeatureSink(bool example_mode, int feature_dim, int top_k, std::vector labels); + + // host_features is a row-major [n, feature_dim] FP32 matrix of one inference + // batch's outputs (n images). + void consume(const float* host_features, uint32_t n); + + void log_final_summary(double seconds) const; + + uint64_t images() const { + return images_; + } + + private: + bool example_mode_; + int feature_dim_; + int top_k_; + int num_classes_ = 0; + std::vector labels_; + + uint64_t images_ = 0; + uint64_t batches_ = 0; + uint64_t samples_printed_ = 0; + + std::vector class_sum_; // num_classes_ * feature_dim_ + std::vector class_count_; // num_classes_ +}; + +} // namespace daqiri::apps::resnet diff --git a/applications/resnet50_inference/inference_pipeline.cu b/applications/resnet50_inference/inference_pipeline.cu new file mode 100644 index 0000000..b1eca66 --- /dev/null +++ b/applications/resnet50_inference/inference_pipeline.cu @@ -0,0 +1,223 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 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. + */ + +#include "inference_pipeline.h" + +#include + +#include +#include +#include +#include +#include +#include + +#include "bench_pipeline.h" +#include "raw_bench_common.h" +#include "trt_runner.hpp" + +#include + +namespace daqiri::apps::resnet { + +void inference_rx_worker(const AppConfig& cfg, FeatureSink& sink, uint64_t expected_images, + std::atomic& ready, std::atomic& tx_done, + std::atomic& stop) { + if (!daqiri::bench::set_current_thread_affinity(cfg.rx.cpu_core, "resnet_rx")) { + stop.store(true); + return; + } + + cudaStream_t stream = nullptr; + if (cudaStreamCreateWithFlags(&stream, cudaStreamNonBlocking) != cudaSuccess) { + std::cerr << "inference_rx_worker: cudaStreamCreate failed\n"; + stop.store(true); + return; + } + + TrtRunner trt(cfg.trt, stream); + trt.initialize(); + + // One image's worth of reorder slots (packets_per_image), shared CUDA stream. + daqiri::bench::ReorderPipeline pipe; + if (!pipe.init(daqiri::bench::ReorderMode::SeqReorder, cfg.packets_per_image, cfg.out_payload_len, + cfg.payload_byte_offset, cfg.seq_bit_offset, cfg.seq_bit_width, + /*staging_needed=*/false, stream) || + !pipe.enabled()) { + std::cerr << "inference_rx_worker: reorder pipeline init failed\n"; + cudaStreamDestroy(stream); + stop.store(true); + return; + } + + const size_t img_bytes = cfg.image_bytes; + const uint32_t batch = cfg.images_per_batch; + char* d_nchw_batch = nullptr; + if (cudaMalloc(&d_nchw_batch, static_cast(batch) * img_bytes) != cudaSuccess) { + std::cerr << "inference_rx_worker: d_nchw_batch alloc failed\n"; + cudaStreamDestroy(stream); + stop.store(true); + return; + } + + cudaEvent_t input_ready = nullptr; + cudaEvent_t release_evt = nullptr; + cudaEventCreateWithFlags(&input_ready, cudaEventDisableTiming); + cudaEventCreateWithFlags(&release_evt, cudaEventDisableTiming); + + const int port_id = daqiri::get_port_id(cfg.rx.interface_name); + if (port_id < 0) { + std::cerr << "inference_rx_worker: invalid RX interface " << cfg.rx.interface_name << "\n"; + cudaFree(d_nchw_batch); + cudaEventDestroy(input_ready); + cudaEventDestroy(release_evt); + cudaStreamDestroy(stream); + stop.store(true); + return; + } + const int queue_id = cfg.rx.queue_id >= 0 ? cfg.rx.queue_id : 0; + + // img_filled / img_in_batch persist ACROSS bursts so an image (or a batch) + // that spans a burst boundary still reassembles correctly. The per-burst + // finish_batch() + stream sync run the reorder kernel before the burst is + // freed, so the persistent reorder buffer accumulates safely. + uint32_t img_filled = 0; + uint32_t img_in_batch = 0; + uint64_t images_completed = 0; // total images reassembled (across all batches) + + // Receive path is fully initialized (engine built/loaded, buffers allocated): + // release the caller so it can start the TX and the run-duration timer. + ready.store(true); + + // The NORMAL example-mode exit is the expected-image count below. This idle + // backstop only catches a genuine stall (e.g. a dropped packet means the count + // never completes): a long run of empty polls (~100us each) after the TX is + // done and we have already received at least one image. It must be far longer + // than the sub-second gaps between TX bursts (the per-packet H2D copies make + // the TX bursty), or it would quiesce mid-stream and drop the tail -- which is + // exactly what a 200ms window did (stopped at 243/256). Gating on + // images_completed>0 also prevents quiescing before the first frame arrives + // (the TX finishes *queuing* well before the first frame reaches the RX ring). + constexpr uint32_t kQuiesceIters = 50000; // ~5 s of continuous empty polls + uint32_t idle_polls = 0; + + const auto t0 = std::chrono::steady_clock::now(); + while (!stop.load()) { + daqiri::BurstParams* burst = nullptr; + if (daqiri::get_rx_burst(&burst, port_id, queue_id) != daqiri::Status::SUCCESS || + burst == nullptr) { + if (tx_done.load() && images_completed > 0 && ++idle_polls >= kQuiesceIters) break; + std::this_thread::sleep_for(std::chrono::microseconds(100)); + continue; + } + idle_polls = 0; + + const int num_pkts = static_cast(daqiri::get_num_packets(burst)); + pipe.reset_batch(); + for (int i = 0; i < num_pkts; ++i) { + pipe.add_device_packet(daqiri::get_segment_packet_ptr(burst, 0, i)); + ++img_filled; + if (img_filled == cfg.packets_per_image) { + // Scatter this burst's contribution to the current image into the + // persistent reorder buffer, then copy the completed image into its + // slot in the NCHW inference batch (all on the shared stream). + const void* ordered = pipe.finish_batch(); + if (ordered != nullptr) { + cudaMemcpyAsync(d_nchw_batch + static_cast(img_in_batch) * img_bytes, ordered, + img_bytes, cudaMemcpyDeviceToDevice, stream); + } + ++img_in_batch; + ++images_completed; + img_filled = 0; + pipe.reset_batch(); + + if (img_in_batch == batch) { + cudaEventRecord(input_ready, stream); + float* host_prev = nullptr; + uint32_t n_prev = 0; + trt.infer(reinterpret_cast(d_nchw_batch), batch, input_ready, release_evt, + host_prev, n_prev); + if (host_prev != nullptr) sink.consume(host_prev, n_prev); + img_in_batch = 0; + } + } + } + // Flush the in-progress image's this-burst packets into the reorder buffer + // BEFORE freeing the burst (its device pointers die at free). + pipe.finish_batch(); + pipe.reset_batch(); + + // Safe-free barrier: the reorder kernel reads the burst's device pointers, + // so the burst must not be freed until the stream drains. + cudaStreamSynchronize(stream); + daqiri::free_all_packets_and_burst_rx(burst); + + // Normal example-mode exit: the whole dataset has been reassembled. (The + // last full inference batch already fired inside the loop; any trailing + // partial batch is flushed below.) + if (expected_images > 0 && images_completed >= expected_images) break; + } + + // Flush a final partial inference batch (img_in_batch < batch): the dataset + // size need not be a multiple of images_per_batch, so the last few completed + // images may not have triggered an infer in the loop. This inference also + // inline-delivers the previous full batch's now-ready features. + if (img_in_batch > 0) { + cudaEventRecord(input_ready, stream); + float* host_prev = nullptr; + uint32_t n_prev = 0; + trt.infer(reinterpret_cast(d_nchw_batch), img_in_batch, input_ready, release_evt, + host_prev, n_prev); + if (host_prev != nullptr) sink.consume(host_prev, n_prev); + img_in_batch = 0; + } + + // Drain the final in-flight inference batch so its features are not lost. + cudaStreamSynchronize(stream); + float* host_final = nullptr; + uint32_t n_final = 0; + trt.drain_final(host_final, n_final); + if (host_final != nullptr) sink.consume(host_final, n_final); + + const double secs = std::chrono::duration(std::chrono::steady_clock::now() - t0).count(); + std::cerr << "inference_rx_worker: " << trt.total_batches_inferred() << " inference batches in " + << secs << " s\n"; + + // Per-batch inference latency (batch-ready -> features on host). Report + // mean/p50/p99 over all batches in a greppable line for the bench sweep. + std::vector lat = trt.batch_latencies_ms(); + if (!lat.empty()) { + std::sort(lat.begin(), lat.end()); + const double mean = + std::accumulate(lat.begin(), lat.end(), 0.0) / static_cast(lat.size()); + const auto pct = [&lat](double p) { + const size_t idx = + std::min(lat.size() - 1, static_cast(p * (static_cast(lat.size()) - 1))); + return lat[idx]; + }; + std::cerr << "inference latency (ms): mean=" << mean << " p50=" << pct(0.50) + << " p99=" << pct(0.99) << " (per batch of " << batch << " images, n=" << lat.size() + << ")\n"; + } + + cudaFree(d_nchw_batch); + cudaEventDestroy(input_ready); + cudaEventDestroy(release_evt); + cudaStreamDestroy(stream); +} + +} // namespace daqiri::apps::resnet diff --git a/applications/resnet50_inference/inference_pipeline.h b/applications/resnet50_inference/inference_pipeline.h new file mode 100644 index 0000000..7205208 --- /dev/null +++ b/applications/resnet50_inference/inference_pipeline.h @@ -0,0 +1,60 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 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 + +#include "app_config.h" +#include "feature_sink.h" + +namespace daqiri::apps::resnet { + +// Single RX thread: dequeue DAQIRI bursts, reassemble each image on the GPU via +// the sequence-number reorder kernel, batch images into a contiguous FP32 NCHW +// buffer, run TensorRT inference, and hand the resulting feature vectors to the +// sink. All GPU work shares one CUDA stream so the reorder kernel orders before +// the inference; the burst is freed only after the stream drains (so the reorder +// has finished reading the burst's device pointers). +// +// `ready` is set true once the (slow, first-run) TensorRT engine build/load and +// all GPU buffers are initialized, so the caller can defer starting the TX and +// the run-duration timer until the receive path is actually ready to consume -- +// a cold engine build can otherwise exceed the whole --seconds window. On any +// init failure the worker sets `stop` (which also releases a caller waiting on +// `ready`). +// +// `expected_images` (example mode) is the dataset size: the worker drains the RX +// ring until it has reassembled that many images, then flushes the final partial +// inference batch -- so the whole dataset is inferred even though inference runs +// slower than line rate, without abandoning un-drained bursts or an un-inferred +// partial batch when a wall-clock `stop` fires mid-stream. Pass 0 in benchmark +// mode to run until `stop`. +// +// `tx_done` lets the TX signal it has finished sending; it is only used as a +// drop-tolerant safety exit (quiescence) once at least one image has been +// received, so a stray dropped packet can't make the worker hang waiting for an +// image count it will never reach. `ready` is set true once the (slow, first-run) +// TensorRT engine build/load and all GPU buffers are initialized, so the caller +// can defer starting the TX until the receive path is ready to consume. On any +// init failure the worker sets `stop` (which also releases a caller waiting on +// `ready`). +void inference_rx_worker(const AppConfig& cfg, FeatureSink& sink, uint64_t expected_images, + std::atomic& ready, std::atomic& tx_done, + std::atomic& stop); + +} // namespace daqiri::apps::resnet diff --git a/applications/resnet50_inference/main.cpp b/applications/resnet50_inference/main.cpp new file mode 100644 index 0000000..c930602 --- /dev/null +++ b/applications/resnet50_inference/main.cpp @@ -0,0 +1,204 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 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. + */ + +// DAQIRI -> TensorRT ResNet inference example (GitHub issue #73). +// +// received packets (raw / DPDK GPUDirect) +// -> GPU sequence-number reorder (image reassembly) +// -> ResNet feature extraction (TensorRT, FP16) +// -> per-class mean-feature stats (example mode) +// +// Example mode (--dataset ) replays preprocessed CIFAR-10 images and +// prints per-class mean-feature stats. Without --dataset the TX synthesizes +// frames for the throughput benchmark (no stats). Runs over the DGX Spark +// physical p0->p1 cabled loopback (see configs/ + README). + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "app_config.h" +#include "feature_sink.h" +#include "inference_pipeline.h" +#include "pcap_replayer.h" +#include "raw_bench_common.h" + +#include + +namespace { + +const char* find_flag_value(int argc, char** argv, const std::string& flag) { + for (int i = 1; i < argc - 1; ++i) { + if (flag == argv[i]) return argv[i + 1]; + } + return nullptr; +} + +bool has_flag(int argc, char** argv, const std::string& flag) { + for (int i = 1; i < argc; ++i) { + if (flag == argv[i]) return true; + } + return false; +} + +} // namespace + +int main(int argc, char** argv) { + if (argc < 2) { + std::cerr << "Usage: " << argv[0] + << " [--seconds N] [--dataset ]" + " [--replay-once|--loop] [--images-per-batch N]" + " [--model resnet18|34|50|101|152]\n"; + return 1; + } + + namespace app = daqiri::apps::resnet; + const int run_seconds = daqiri::bench::parse_run_seconds(argc, argv); + + app::AppConfig cfg; + try { + const auto root = YAML::LoadFile(argv[1]); + cfg = app::AppConfig::from_yaml(root); + } catch (const std::exception& e) { + std::cerr << "Invalid config: " << e.what() << "\n"; + return 1; + } + + // CLI overrides. + if (const char* ds = find_flag_value(argc, argv, "--dataset")) { + cfg.dataset_pcap = ds; + } + if (const char* ipb = find_flag_value(argc, argv, "--images-per-batch")) { + cfg.images_per_batch = std::min(static_cast(std::stoul(ipb)), + static_cast(cfg.trt.opt_max)); + } + if (const char* model = find_flag_value(argc, argv, "--model")) { + const std::string m = model; + cfg.trt.onnx_path = "models/" + m + "_features.onnx"; + cfg.trt.engine_path = "models/" + m + "_features.fp16.engine"; + cfg.trt.feature_dim = (m == "resnet18" || m == "resnet34") ? 512 : 2048; + // image_bytes / geometry are independent of feature_dim (input dims fixed). + } + // Example mode replays the dataset ONCE by default: a small dataset fits + // entirely in the RX ring (num_images * packets_per_image < num_bufs), so no + // packets drop even though ResNet inference is slower than line rate -- which + // keeps the drop-free, in-order reassembly (and label mapping) intact. + // Benchmark mode loops to sustain throughput (drops are tolerable there since + // it only counts). Override with --loop / --replay-once. + bool loop = !cfg.example_mode(); + if (has_flag(argc, argv, "--loop")) loop = true; + if (has_flag(argc, argv, "--replay-once")) loop = false; + + // Destination MAC for the replayed frames: ETH_DST_ADDR env (the RX port MAC, + // matching the Spark report) overrides the config placeholder. + std::string eth_dst = cfg.tx.eth_dst_addr; + if (const char* env = std::getenv("ETH_DST_ADDR")) { + if (env[0] != '\0') eth_dst = env; + } + + // Labels sidecar (example mode only). + std::vector labels; + if (cfg.example_mode()) { + if (cfg.labels_path.empty()) cfg.labels_path = cfg.dataset_pcap + ".labels"; + labels = app::load_labels(cfg.labels_path); + if (labels.empty()) { + std::cerr << "Warning: no labels loaded; per-class stats disabled\n"; + } + } + + std::cerr << "ResNet inference: mode=" + << (cfg.example_mode() ? "example (dataset)" : "benchmark (synthetic)") + << " feature_dim=" << cfg.trt.feature_dim + << " packets_per_image=" << cfg.packets_per_image + << " out_payload_len=" << cfg.out_payload_len + << " images_per_batch=" << cfg.images_per_batch << " eth_dst=" << eth_dst << "\n"; + + // Build / load the TX frame source before init so a bad dataset fails fast. + std::vector frames; + if (cfg.example_mode()) { + app::PcapReplayer replayer; + if (!replayer.load(cfg.dataset_pcap)) { + std::cerr << "Failed to load dataset pcap: " << cfg.dataset_pcap << "\n"; + return 1; + } + frames = replayer.frames(); + } else { + frames = app::build_synthetic_frames(cfg); + } + + if (daqiri::daqiri_init(argv[1]) != daqiri::Status::SUCCESS) { + std::cerr << "daqiri_init failed\n"; + return 1; + } + + app::FeatureSink sink(cfg.example_mode() && !labels.empty(), cfg.trt.feature_dim, cfg.stats_top_k, + labels); + + std::atomic stop{false}; + std::atomic rx_ready{false}; + std::atomic tx_done{false}; + // Example mode drains exactly the dataset; benchmark mode (0) runs until stop. + const uint64_t expected_images = + cfg.example_mode() && cfg.packets_per_image > 0 ? frames.size() / cfg.packets_per_image : 0; + std::thread rx_thread(app::inference_rx_worker, std::cref(cfg), std::ref(sink), expected_images, + std::ref(rx_ready), std::ref(tx_done), std::ref(stop)); + + // The first-run TensorRT engine build can take ~a minute -- longer than the + // whole --seconds window. Wait until the RX path is ready (engine built/loaded + // and buffers allocated) before starting the TX and the run-duration timer, so + // the timer measures actual receive/inference time, not the engine build. An + // init failure sets `stop`, which also breaks this wait. + while (!rx_ready.load() && !stop.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + const auto run_t0 = std::chrono::steady_clock::now(); + std::thread tx_thread(app::pcap_tx_worker, std::cref(cfg), std::cref(frames), std::cref(eth_dst), + loop, std::ref(tx_done), std::ref(stop)); + + double summary_seconds = run_seconds; + if (loop) { + // Benchmark / continuous replay: measure a fixed wall-clock window. + daqiri::bench::wait_for_stop(run_seconds, stop); + } else { + // Replay-once (example mode): the TX sends the dataset once, then the RX + // worker drains the ring to quiescence (it sees tx_done) and flushes the + // final partial batch -- so the whole dataset is inferred even though + // inference runs slower than line rate. No fixed timer: just let TX finish + // and RX drain, then report the actual elapsed time. + if (tx_thread.joinable()) tx_thread.join(); + if (rx_thread.joinable()) rx_thread.join(); + summary_seconds = + std::chrono::duration(std::chrono::steady_clock::now() - run_t0).count(); + } + + if (tx_thread.joinable()) tx_thread.join(); + if (rx_thread.joinable()) rx_thread.join(); + + sink.log_final_summary(summary_seconds); + daqiri::print_stats(); + daqiri::shutdown(); + return 0; +} diff --git a/applications/resnet50_inference/pcap_replayer.cpp b/applications/resnet50_inference/pcap_replayer.cpp new file mode 100644 index 0000000..7b82514 --- /dev/null +++ b/applications/resnet50_inference/pcap_replayer.cpp @@ -0,0 +1,234 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 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. + */ + +#include "pcap_replayer.h" + +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace daqiri::apps::resnet { + +namespace { + +constexpr uint32_t kPcapMagicLE = 0xa1b2c3d4; // microsecond, little-endian +constexpr uint32_t kPcapMagicLE_ns = 0xa1b23c4d; +constexpr int kLinkTypeEthernet = 1; + +} // namespace + +bool PcapReplayer::load(const std::string& path) { + std::ifstream f(path, std::ios::binary); + if (!f) { + std::cerr << "PcapReplayer: cannot open " << path << "\n"; + return false; + } + + uint8_t global[24]; + if (!f.read(reinterpret_cast(global), sizeof(global))) { + std::cerr << "PcapReplayer: " << path << " truncated global header\n"; + return false; + } + uint32_t magic; + std::memcpy(&magic, global, sizeof(magic)); + if (magic != kPcapMagicLE && magic != kPcapMagicLE_ns) { + std::cerr << "PcapReplayer: " << path << " is not a little-endian pcap (magic mismatch)\n"; + return false; + } + uint32_t network; + std::memcpy(&network, global + 20, sizeof(network)); + if (network != kLinkTypeEthernet) { + std::cerr << "PcapReplayer: " << path << " linktype " << network << " is not Ethernet (1)\n"; + return false; + } + + frames_.clear(); + uint8_t rec[16]; + while (f.read(reinterpret_cast(rec), sizeof(rec))) { + uint32_t incl_len; + std::memcpy(&incl_len, rec + 8, sizeof(incl_len)); + PcapFrame frame(incl_len); + if (!f.read(reinterpret_cast(frame.data()), incl_len)) { + std::cerr << "PcapReplayer: " << path << " truncated frame body\n"; + break; + } + frames_.push_back(std::move(frame)); + } + + std::cerr << "PcapReplayer: loaded " << frames_.size() << " frames from " << path << "\n"; + return !frames_.empty(); +} + +std::vector build_synthetic_frames(const AppConfig& cfg) { + char eth_src[6] = {0}; + char eth_dst[6] = {0}; + daqiri::format_eth_addr(eth_src, cfg.tx.eth_src_addr); + daqiri::format_eth_addr(eth_dst, cfg.tx.eth_dst_addr); + + uint32_t ip_src = 0; + uint32_t ip_dst = 0; + inet_pton(AF_INET, cfg.tx.ip_src_addr.c_str(), &ip_src); + inet_pton(AF_INET, cfg.tx.ip_dst_addr.c_str(), &ip_dst); + ip_src = ntohl(ip_src); + ip_dst = ntohl(ip_dst); + + const auto src_port = static_cast(std::stoi(cfg.tx.udp_src_port)); + const auto dst_port = static_cast(std::stoi(cfg.tx.udp_dst_port)); + + const uint32_t udp_payload = 4u + cfg.out_payload_len; // seq prefix + chunk + const uint32_t frame_bytes = cfg.tx.header_size + udp_payload; + + // images_per_batch images so the TX has a full inference batch of variety. + const uint32_t num_images = cfg.images_per_batch > 0 ? cfg.images_per_batch : 1; + std::vector frames; + frames.reserve(static_cast(num_images) * cfg.packets_per_image); + + for (uint32_t img = 0; img < num_images; ++img) { + for (uint32_t seq = 0; seq < cfg.packets_per_image; ++seq) { + PcapFrame frame(frame_bytes); + daqiri::bench::populate_udp_ipv4_headers(frame.data(), cfg.tx.header_size, udp_payload, + eth_src, eth_dst, ip_src, ip_dst, src_port, + dst_port); + // Per-image sequence number (network byte order) at the payload start. + const uint32_t seq_be = htonl(seq); + std::memcpy(frame.data() + cfg.tx.header_size, &seq_be, sizeof(seq_be)); + // Arbitrary pixel bytes (content irrelevant to throughput). + for (uint32_t b = 0; b < cfg.out_payload_len; ++b) { + frame[cfg.tx.header_size + 4u + b] = static_cast((seq + b) & 0xff); + } + daqiri::bench::finalize_udp_ipv4_checksums(frame.data()); + frames.push_back(std::move(frame)); + } + } + std::cerr << "build_synthetic_frames: " << frames.size() << " frames (" << num_images + << " images x " << cfg.packets_per_image << " pkts)\n"; + return frames; +} + +void pcap_tx_worker(const AppConfig& cfg, const std::vector& frames, + const std::string& eth_dst_addr, bool loop, std::atomic& tx_done, + std::atomic& stop) { + if (!daqiri::bench::set_current_thread_affinity(cfg.tx.cpu_core, "resnet_tx")) { + stop.store(true); + return; + } + if (frames.empty()) { + std::cerr << "pcap_tx_worker: no frames to replay\n"; + stop.store(true); + return; + } + + const int port_id = daqiri::get_port_id(cfg.tx.interface_name); + if (port_id < 0) { + std::cerr << "pcap_tx_worker: invalid TX interface " << cfg.tx.interface_name << "\n"; + stop.store(true); + return; + } + + // Resolved destination MAC patched into every frame at send time so the pcap + // is host-independent and the NIC accepts the frame on the RX port. + char dst_mac[6] = {0}; + daqiri::format_eth_addr(dst_mac, eth_dst_addr); + + std::vector scratch; + size_t cursor = 0; + uint64_t total_sent = 0; + + while (!stop.load()) { + // In replay-once mode, size the final burst to the exact remainder so the + // dataset is sent exactly once. `batch_size` (>> packets_per_image) would + // otherwise over-send by up to a full burst, wrapping the cursor and + // re-sending the first images -- skewing the per-class sample counts. + uint32_t want = cfg.tx.batch_size; + if (!loop) { + if (total_sent >= frames.size()) break; + want = + static_cast(std::min(cfg.tx.batch_size, frames.size() - total_sent)); + } + + auto* msg = daqiri::create_tx_burst_params(); + daqiri::set_header(msg, static_cast(port_id), static_cast(cfg.tx.queue_id), + want, 1); + + if (!daqiri::is_tx_burst_available(msg)) { + daqiri::free_tx_metadata(msg); + std::this_thread::sleep_for(std::chrono::microseconds(100)); + continue; + } + if (daqiri::get_tx_packet_burst(msg) != daqiri::Status::SUCCESS) { + daqiri::free_tx_metadata(msg); + continue; + } + + bool failed = false; + const int num_pkts = static_cast(daqiri::get_num_packets(msg)); + for (int i = 0; i < num_pkts; ++i) { + const PcapFrame& frame = frames[cursor]; + cursor = (cursor + 1) % frames.size(); + + scratch.assign(frame.begin(), frame.end()); + if (scratch.size() >= 6) { + std::memcpy(scratch.data(), dst_mac, 6); // patch dst MAC + } + + auto* gpu_pkt = daqiri::get_segment_packet_ptr(msg, 0, i); + if (cudaMemcpy(gpu_pkt, scratch.data(), scratch.size(), cudaMemcpyHostToDevice) != + cudaSuccess) { + failed = true; + break; + } + if (daqiri::set_packet_lengths(msg, i, {static_cast(scratch.size())}) != + daqiri::Status::SUCCESS) { + failed = true; + break; + } + } + + if (failed) { + daqiri::free_all_packets_and_burst_tx(msg); + continue; + } + if (daqiri::send_tx_burst(msg) == daqiri::Status::SUCCESS) { + total_sent += static_cast(num_pkts); + } else { + // Caller owns the burst after get_tx_packet_burst; free it on a failed send + // so a persistent error doesn't drain the TX mempool one slot at a time. + daqiri::free_all_packets_and_burst_tx(msg); + continue; + } + // --replay-once: the whole dataset has now been sent exactly once (the burst + // above was sized to the remainder), so stop. + if (!loop && total_sent >= frames.size()) { + break; + } + } + + // Signal the RX worker that no more packets are coming so it can drain the + // ring to quiescence and stop (example / --replay-once mode). + tx_done.store(true); +} + +} // namespace daqiri::apps::resnet diff --git a/applications/resnet50_inference/pcap_replayer.h b/applications/resnet50_inference/pcap_replayer.h new file mode 100644 index 0000000..23b5b2a --- /dev/null +++ b/applications/resnet50_inference/pcap_replayer.h @@ -0,0 +1,65 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 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 +#include +#include +#include + +#include "app_config.h" + +namespace daqiri::apps::resnet { + +using PcapFrame = std::vector; // one full ETH/IP/UDP frame + +// Minimal standard-pcap reader (little-endian global + record headers, +// linktype 1 / Ethernet). Loads every frame into memory for replay; the +// per-image data prep keeps pcaps small (a few hundred images). +class PcapReplayer { + public: + bool load(const std::string& path); + const std::vector& frames() const { + return frames_; + } + bool empty() const { + return frames_.empty(); + } + + private: + std::vector frames_; +}; + +// Build synthetic frames for the benchmark (no dataset): images_per_batch +// images worth of frames, each carrying a per-image sequence number (0 .. +// packets_per_image-1) at the payload start and arbitrary pixel bytes. The RX +// pipeline reassembles + infers identically to the dataset path; only the pixel +// content (and thus the feature values) is meaningless, which does not affect +// throughput. +std::vector build_synthetic_frames(const AppConfig& cfg); + +// Replay `frames` into the DAQIRI TX path forever (or once if loop==false), +// patching each frame's destination MAC to `eth_dst_addr` at send time so one +// pcap works on any host. Mirrors the bench tx_worker burst/send sequence. +// Sets `tx_done` true on return (the whole dataset has been sent once, in +// replay-once mode) so the RX worker knows to drain the ring and stop. +void pcap_tx_worker(const AppConfig& cfg, const std::vector& frames, + const std::string& eth_dst_addr, bool loop, std::atomic& tx_done, + std::atomic& stop); + +} // namespace daqiri::apps::resnet diff --git a/applications/resnet50_inference/tools/export_resnet_onnx.py b/applications/resnet50_inference/tools/export_resnet_onnx.py new file mode 100755 index 0000000..5cf0782 --- /dev/null +++ b/applications/resnet50_inference/tools/export_resnet_onnx.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 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. +"""Export a torchvision ResNet *feature extractor* (final FC stripped) to ONNX. + +The fully-connected classifier head is replaced with Identity, so the network +outputs the global-average-pooled feature vector (512 dims for resnet18/34, +2048 for resnet50/101/152). A dynamic batch axis lets one engine serve any +inference batch size. Run inside the BASE_IMAGE=torch container (torch + +torchvision pre-installed). + + python3 export_resnet_onnx.py --model resnet50 --output models/resnet50_features.onnx +""" + +import argparse +import os + +import torch +import torchvision + +FEATURE_DIM = { + "resnet18": 512, + "resnet34": 512, + "resnet50": 2048, + "resnet101": 2048, + "resnet152": 2048, +} + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--model", default="resnet50", choices=sorted(FEATURE_DIM)) + ap.add_argument("--output", default=None, + help="output .onnx path (default: models/_features.onnx)") + ap.add_argument("--weights", default="DEFAULT", choices=["DEFAULT", "none"], + help="DEFAULT = ImageNet-pretrained (meaningful features); " + "none = random init (offline)") + ap.add_argument("--opset", type=int, default=18, + help="ONNX opset (>=18 avoids a noisy down-conversion in newer " + "torch.onnx exporters; TensorRT 10 handles opset 18)") + ap.add_argument("--check", action="store_true", help="run onnx.checker on the result") + args = ap.parse_args() + + out = args.output or os.path.join("models", f"{args.model}_features.onnx") + os.makedirs(os.path.dirname(out) or ".", exist_ok=True) + + weights = None if args.weights == "none" else "DEFAULT" + model = getattr(torchvision.models, args.model)(weights=weights) + model.fc = torch.nn.Identity() # output = pooled feature vector [N, feature_dim] + model.eval() + + dummy = torch.randn(1, 3, 224, 224) + torch.onnx.export( + model, + dummy, + out, + export_params=True, + opset_version=args.opset, + do_constant_folding=True, + input_names=["input"], + output_names=["features"], + dynamic_axes={"input": {0: "batch"}, "features": {0: "batch"}}, + ) + print(f"Exported {args.model} feature extractor -> {out} " + f"(feature_dim={FEATURE_DIM[args.model]}, weights={args.weights})") + + if args.check: + import onnx + m = onnx.load(out) + onnx.checker.check_model(m) + out_shape = [d.dim_value or d.dim_param + for d in m.graph.output[0].type.tensor_type.shape.dim] + print(f"onnx.checker OK; output '{m.graph.output[0].name}' shape={out_shape}") + + +if __name__ == "__main__": + main() diff --git a/applications/resnet50_inference/tools/prepare_cifar10_pcap.py b/applications/resnet50_inference/tools/prepare_cifar10_pcap.py new file mode 100755 index 0000000..b2b2e20 --- /dev/null +++ b/applications/resnet50_inference/tools/prepare_cifar10_pcap.py @@ -0,0 +1,152 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 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. +"""Packetize preprocessed CIFAR-10 images into a standard .pcap for replay. + +Each image is resized to 224x224, ImageNet-normalized, and serialized as FP32 +NCHW (3*224*224*4 = 602112 bytes). The bytes are split into fixed-size chunks +(out_payload_len) and framed into ETH/IP/UDP packets. Each packet carries a +per-image sequence number (0 .. packets_per_image-1) at the payload start so the +DAQIRI RX-side reorder kernel reassembles the image. A ".labels" sidecar +(one CIFAR class id per image, in send order) drives the per-class feature stats. + +The frame layout mirrors the bench's populate_udp_ipv4_headers: a real +14B Ethernet + 20B IPv4 + 8B UDP header, zero filler up to header_size, then the +4-byte sequence number, then the image chunk. Preprocessing is baked in here so +the runtime pipeline stays pure reorder -> infer. No scapy dependency. + + python3 prepare_cifar10_pcap.py --num-images 256 --out data/cifar10_resnet.pcap +""" + +import argparse +import os +import socket +import struct + +IMAGENET_MEAN = (0.485, 0.456, 0.406) +IMAGENET_STD = (0.229, 0.224, 0.225) + + +def ip_checksum(header: bytes) -> int: + s = 0 + for i in range(0, len(header), 2): + s += (header[i] << 8) | header[i + 1] + s = (s & 0xFFFF) + (s >> 16) + return (~s) & 0xFFFF + + +def build_frame(chunk: bytes, seq: int, header_size: int, eth_src: bytes, + eth_dst: bytes, ip_src: int, ip_dst: int, sport: int, + dport: int) -> bytes: + # UDP payload = filler (header_size - 42) + 4-byte seq + chunk. + filler = header_size - 42 + udp_payload = b"\x00" * filler + struct.pack(">I", seq) + chunk + udp_len = 8 + len(udp_payload) + ip_total = 20 + udp_len + + eth = eth_dst + eth_src + struct.pack(">H", 0x0800) + + ip = bytearray(struct.pack(">BBHHHBBH4s4s", + 0x45, 0x00, ip_total, 0x0000, 0x0000, 64, + socket.IPPROTO_UDP, 0, + struct.pack(">I", ip_src), struct.pack(">I", ip_dst))) + chk = ip_checksum(bytes(ip)) + ip[10] = (chk >> 8) & 0xFF + ip[11] = chk & 0xFF + + # IPv4 UDP checksum is optional; 0 means "not computed" (the NIC flow matches + # ports, not the checksum), which keeps this tool simple. + udp = struct.pack(">HHHH", sport, dport, udp_len, 0) + + return eth + bytes(ip) + udp + udp_payload + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--num-images", type=int, default=256) + ap.add_argument("--out", default="data/cifar10_resnet.pcap") + ap.add_argument("--data-root", default="data/cifar10") + ap.add_argument("--out-payload-len", type=int, default=7168, + help="image-pixel bytes per packet (must divide 602112)") + ap.add_argument("--header-size", type=int, default=64, + help="logical payload offset (>=42); matches the YAML header_size") + ap.add_argument("--udp-port", type=int, default=4096) + ap.add_argument("--eth-src", default="02:00:00:00:00:01") + ap.add_argument("--eth-dst", default="02:00:00:00:00:02", + help="placeholder; the replayer patches the real RX MAC at TX time") + ap.add_argument("--ip-src", default="1.2.3.4") + ap.add_argument("--ip-dst", default="5.6.7.8") + args = ap.parse_args() + + import torch + import torchvision + import torchvision.transforms as T + + image_bytes = 3 * 224 * 224 * 4 + if image_bytes % args.out_payload_len != 0: + raise SystemExit(f"--out-payload-len {args.out_payload_len} must divide " + f"{image_bytes}") + packets_per_image = image_bytes // args.out_payload_len + if args.header_size < 42: + raise SystemExit("--header-size must be >= 42 (ETH+IP+UDP)") + + transform = T.Compose([ + T.Resize(224), + T.ToTensor(), + T.Normalize(mean=IMAGENET_MEAN, std=IMAGENET_STD), + ]) + ds = torchvision.datasets.CIFAR10(root=args.data_root, train=False, + download=True, transform=transform) + n = min(args.num_images, len(ds)) + + eth_src = bytes(int(b, 16) for b in args.eth_src.split(":")) + eth_dst = bytes(int(b, 16) for b in args.eth_dst.split(":")) + ip_src = struct.unpack(">I", socket.inet_aton(args.ip_src))[0] + ip_dst = struct.unpack(">I", socket.inet_aton(args.ip_dst))[0] + + os.makedirs(os.path.dirname(args.out) or ".", exist_ok=True) + labels = [] + frames_written = 0 + with open(args.out, "wb") as pcap: + # pcap global header: magic, ver 2.4, thiszone, sigfigs, snaplen, linktype=1. + pcap.write(struct.pack("= this)") + + +if __name__ == "__main__": + main() diff --git a/applications/resnet50_inference/tools/run_resnet_bench.sh b/applications/resnet50_inference/tools/run_resnet_bench.sh new file mode 100755 index 0000000..978bba9 --- /dev/null +++ b/applications/resnet50_inference/tools/run_resnet_bench.sh @@ -0,0 +1,83 @@ +#!/usr/bin/env bash +# SPDX-FileCopyrightText: Copyright (c) 2026 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. +# +# Sweep ResNet-18/34/50/101/152 inference throughput in the DAQIRI receive path +# and emit a CSV. Platform-agnostic: drive it via env vars so the same script +# runs on DGX Spark, IGX, or an RTX Pro server. Physical wire loopback only +# (synthetic benchmark mode): the dq_wire_* netns must be DOWN and ETH_DST_ADDR +# must be exported (the RX port MAC), exactly like the raw/DPDK bench. +# +# export ETH_DST_ADDR=$(cat /sys/class/net//address) +# BUILD_DIR=./build ./tools/run_resnet_bench.sh +# +# Env knobs: BUILD_DIR, BIN, CONFIG, MODELS, SECONDS_PER, IMAGES_PER_BATCH, +# REPEATS, OUT, MODEL_DIR. +set -euo pipefail + +BUILD_DIR=${BUILD_DIR:-./build} +APP_DIR="${BUILD_DIR}/applications/resnet50_inference" +BIN=${BIN:-${APP_DIR}/daqiri_resnet50_inference} +CONFIG=${CONFIG:-${APP_DIR}/configs/resnet50_bench_spark.yaml} +MODELS=${MODELS:-"resnet18 resnet34 resnet50 resnet101 resnet152"} +SECONDS_PER=${SECONDS_PER:-30} +IMAGES_PER_BATCH=${IMAGES_PER_BATCH:-32} +REPEATS=${REPEATS:-3} +MODEL_DIR=${MODEL_DIR:-models} +OUT=${OUT:-resnet-bench.csv} + +if [[ -z "${ETH_DST_ADDR:-}" ]]; then + echo "ETH_DST_ADDR not set. Export the RX port MAC, e.g.:" >&2 + echo " export ETH_DST_ADDR=\$(cat /sys/class/net//address)" >&2 + exit 1 +fi +if [[ ! -x "${BIN}" ]]; then + echo "Binary not found: ${BIN} (build with -DDAQIRI_BUILD_APPLICATIONS=ON)" >&2 + exit 1 +fi + +declare -A FEATURE_DIM=( [resnet18]=512 [resnet34]=512 [resnet50]=2048 \ + [resnet101]=2048 [resnet152]=2048 ) + +echo "model,feature_dim,images_per_batch,seconds,rep,img_s,inference_batches,lat_p50_ms,lat_p99_ms,rx_drops" > "${OUT}" + +for model in ${MODELS}; do + onnx="${MODEL_DIR}/${model}_features.onnx" + if [[ ! -f "${onnx}" ]]; then + echo "Exporting ONNX for ${model} ..." + python3 "$(dirname "$0")/export_resnet_onnx.py" --model "${model}" --output "${onnx}" + fi + for rep in $(seq 1 "${REPEATS}"); do + echo "=== ${model} rep ${rep}/${REPEATS} ===" + log=$(mktemp) + # First rep builds + caches the engine; later reps load from cache. + "${BIN}" "${CONFIG}" --model "${model}" --images-per-batch "${IMAGES_PER_BATCH}" \ + --seconds "${SECONDS_PER}" 2>&1 | tee "${log}" || true + # `|| true` on each: grep exits non-zero when a field is absent (e.g. a + # drop-free run has no "total:" line), which under `set -euo pipefail` would + # otherwise abort the whole sweep at the assignment. + img_s=$(grep -oE '=> [0-9.]+ img/s' "${log}" | grep -oE '[0-9.]+' | head -1 || true) + batches=$(grep -oE '[0-9]+ inference batches' "${log}" | grep -oE '[0-9]+' | head -1 || true) + lat_p50=$(grep -oE 'p50=[0-9.]+' "${log}" | head -1 | cut -d= -f2 || true) + lat_p99=$(grep -oE 'p99=[0-9.]+' "${log}" | head -1 | cut -d= -f2 || true) + # Cumulative RX drop total from the periodic drop logger; absent => 0. + drops=$(grep -oE 'total: [0-9]+' "${log}" | grep -oE '[0-9]+' | tail -1 || true) + echo "${model},${FEATURE_DIM[$model]:-2048},${IMAGES_PER_BATCH},${SECONDS_PER},${rep},${img_s:-NA},${batches:-NA},${lat_p50:-NA},${lat_p99:-NA},${drops:-0}" >> "${OUT}" + rm -f "${log}" + done +done + +echo "Wrote ${OUT}" +column -s, -t "${OUT}" || cat "${OUT}" diff --git a/applications/resnet50_inference/trt_runner.cu b/applications/resnet50_inference/trt_runner.cu new file mode 100644 index 0000000..28564a2 --- /dev/null +++ b/applications/resnet50_inference/trt_runner.cu @@ -0,0 +1,313 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 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. + */ + +// TensorRT 10.x engine build + FP16 inference + double-buffered async D2H. We +// use TRT directly (rather than a higher-level inference runtime) because the +// pipeline needs exactly one engine, one input, one output, and dynamic batch. + +#include "trt_runner.hpp" + +#include +#include +#include + +#include +#include +#include +#include +#include + +namespace daqiri::apps::resnet { + +namespace { + +class TrtLogger : public nvinfer1::ILogger { + public: + void log(Severity severity, const char* msg) noexcept override { + if (severity == Severity::kERROR || severity == Severity::kINTERNAL_ERROR) { + std::cerr << "[TRT] " << msg << "\n"; + } else if (severity == Severity::kWARNING) { + std::cerr << "[TRT][warn] " << msg << "\n"; + } + } +}; + +TrtLogger& trt_logger() { + static TrtLogger logger; + return logger; +} + +} // namespace + +TrtRunnerConfig TrtRunnerConfig::from_yaml(const YAML::Node& root) { + TrtRunnerConfig cfg; + const auto& inf = root["inference"]; + if (!inf) return cfg; + if (inf["onnx_path"]) cfg.onnx_path = inf["onnx_path"].as(); + if (inf["engine_path"]) cfg.engine_path = inf["engine_path"].as(); + if (inf["opt_min"]) cfg.opt_min = inf["opt_min"].as(); + if (inf["opt_avg"]) cfg.opt_avg = inf["opt_avg"].as(); + if (inf["opt_max"]) cfg.opt_max = inf["opt_max"].as(); + if (inf["enable_fp16"]) cfg.enable_fp16 = inf["enable_fp16"].as(); + if (inf["channels"]) cfg.channels = inf["channels"].as(); + if (inf["height"]) cfg.height = inf["height"].as(); + if (inf["width"]) cfg.width = inf["width"].as(); + if (inf["feature_dim"]) cfg.feature_dim = inf["feature_dim"].as(); + if (inf["gpu_id"]) cfg.gpu_id = inf["gpu_id"].as(); + if (inf["input_name"]) cfg.input_name = inf["input_name"].as(); + if (inf["output_name"]) cfg.output_name = inf["output_name"].as(); + return cfg; +} + +TrtRunner::TrtRunner(TrtRunnerConfig cfg, cudaStream_t inf_stream) + : cfg_(std::move(cfg)), inf_stream_(inf_stream) {} + +TrtRunner::~TrtRunner() { + for (int i = 0; i < kBuffers; ++i) { + if (host_buf_[i]) cudaFreeHost(host_buf_[i]); + if (trt_out_dev_[i]) cudaFree(trt_out_dev_[i]); + if (d2h_event_[i]) cudaEventDestroy(d2h_event_[i]); + if (start_evt_[i]) cudaEventDestroy(start_evt_[i]); + } + delete context_; + delete engine_; + delete runtime_; +} + +void TrtRunner::initialize() { + std::cerr << "TrtRunner: onnx=" << cfg_.onnx_path << " engine_cache=" << cfg_.engine_path + << " fp16=" << cfg_.enable_fp16 << " input=" << cfg_.channels << "x" << cfg_.height + << "x" << cfg_.width << " feature_dim=" << cfg_.feature_dim << " opt=(" << cfg_.opt_min + << "," << cfg_.opt_avg << "," << cfg_.opt_max << ")\n"; + + runtime_ = nvinfer1::createInferRuntime(trt_logger()); + if (!runtime_) { + std::cerr << "createInferRuntime failed\n"; + std::exit(1); + } + + build_or_load_engine_(); + + context_ = engine_->createExecutionContext(); + if (!context_) { + std::cerr << "createExecutionContext failed\n"; + std::exit(1); + } + + // The pcap carries FP32 NCHW pixels, so the engine input binding must be + // FP32 (the kFP16 builder flag runs the network internally in FP16 but keeps + // the declared input binding as the ONNX type). Catch an accidentally + // FP16-input ONNX early rather than reading garbage at the first infer. + if (engine_->getTensorDataType(cfg_.input_name.c_str()) != nvinfer1::DataType::kFLOAT) { + std::cerr << "TrtRunner: input tensor '" << cfg_.input_name + << "' is not FP32; export the ONNX with an FP32 input\n"; + std::exit(1); + } + + allocate_buffers_(); + std::cerr << "TrtRunner ready\n"; +} + +void TrtRunner::build_or_load_engine_() { + std::vector plan_data; + + if (!cfg_.engine_path.empty()) { + std::ifstream f(cfg_.engine_path, std::ios::binary | std::ios::ate); + if (f) { + const auto sz = static_cast(f.tellg()); + f.seekg(0, std::ios::beg); + plan_data.resize(static_cast(sz)); + if (f.read(plan_data.data(), sz)) { + std::cerr << "Loaded cached engine from " << cfg_.engine_path << " (" << sz << " bytes)\n"; + } else { + plan_data.clear(); + } + } + } + + if (plan_data.empty()) { + std::cerr << "Building TRT engine from " << cfg_.onnx_path + << " (first run; this can take a minute)\n"; + + auto* builder = nvinfer1::createInferBuilder(trt_logger()); + if (!builder) { + std::cerr << "createInferBuilder failed\n"; + std::exit(1); + } + + // TRT 10 makes explicit batch the only (default) mode; pass no creation + // flags (the old kEXPLICIT_BATCH flag is deprecated). + auto* network = builder->createNetworkV2(0U); + if (!network) { + std::cerr << "createNetworkV2 failed\n"; + std::exit(1); + } + + auto* parser = nvonnxparser::createParser(*network, trt_logger()); + if (!parser->parseFromFile(cfg_.onnx_path.c_str(), + static_cast(nvinfer1::ILogger::Severity::kWARNING))) { + std::cerr << "ONNX parse failed: " << cfg_.onnx_path << "\n"; + std::exit(1); + } + + auto* config = builder->createBuilderConfig(); + // All supported targets (A100/H100/Ada/GB10) have fast FP16, so gate only on + // the config flag. kFP16 is marked deprecated in TRT 10 in favor of + // strongly-typed networks, but remains the simplest way to request a mixed + // FP16 build from an FP32 ONNX. + if (cfg_.enable_fp16) { + config->setFlag(nvinfer1::BuilderFlag::kFP16); + } + + auto* profile = builder->createOptimizationProfile(); + const nvinfer1::Dims4 dmin{cfg_.opt_min, cfg_.channels, cfg_.height, cfg_.width}; + const nvinfer1::Dims4 dopt{cfg_.opt_avg, cfg_.channels, cfg_.height, cfg_.width}; + const nvinfer1::Dims4 dmax{cfg_.opt_max, cfg_.channels, cfg_.height, cfg_.width}; + profile->setDimensions(cfg_.input_name.c_str(), nvinfer1::OptProfileSelector::kMIN, dmin); + profile->setDimensions(cfg_.input_name.c_str(), nvinfer1::OptProfileSelector::kOPT, dopt); + profile->setDimensions(cfg_.input_name.c_str(), nvinfer1::OptProfileSelector::kMAX, dmax); + config->addOptimizationProfile(profile); + + auto* plan = builder->buildSerializedNetwork(*network, *config); + if (!plan) { + std::cerr << "buildSerializedNetwork failed\n"; + std::exit(1); + } + + plan_data.assign(static_cast(plan->data()), + static_cast(plan->data()) + plan->size()); + + if (!cfg_.engine_path.empty()) { + std::ofstream out(cfg_.engine_path, std::ios::binary); + out.write(plan_data.data(), static_cast(plan_data.size())); + std::cerr << "Cached engine to " << cfg_.engine_path << " (" << plan_data.size() + << " bytes)\n"; + } + + delete plan; + delete config; + delete parser; + delete network; + delete builder; + } + + engine_ = runtime_->deserializeCudaEngine(plan_data.data(), plan_data.size()); + if (!engine_) { + std::cerr << "deserializeCudaEngine failed\n"; + std::exit(1); + } +} + +void TrtRunner::allocate_buffers_() { + // Feature-extractor output: feature_dim floats per image (batch * feature_dim). + const size_t max_out_bytes = static_cast(cfg_.opt_max) * cfg_.feature_dim * sizeof(float); + for (int i = 0; i < kBuffers; ++i) { + cudaMalloc(&trt_out_dev_[i], max_out_bytes); + cudaMallocHost(&host_buf_[i], max_out_bytes); + // Timing-enabled (default flags) so cudaEventElapsedTime(start_evt, d2h) + // yields the per-batch latency; both events in a pair must be timing-enabled. + cudaEventCreate(&d2h_event_[i]); + cudaEventCreate(&start_evt_[i]); + } + std::cerr << "TrtRunner buffers: 2x" << max_out_bytes << " bytes pinned host + GPU output\n"; +} + +void TrtRunner::infer(float* dev_input, uint32_t batch, cudaEvent_t input_ready, + cudaEvent_t release_evt, float*& host_out_prev, uint32_t& host_out_prev_n) { + host_out_prev = nullptr; + host_out_prev_n = 0; + + if (batch == 0 || dev_input == nullptr) { + std::cerr << "TrtRunner::infer called with batch=" << batch + << " dev_input=" << static_cast(dev_input) << "\n"; + return; + } + // setInputShape logs + returns on overflow which would silently skip + // inference; clamp so we never quietly produce zero features. + if (batch > static_cast(cfg_.opt_max)) { + std::cerr << "TrtRunner::infer batch=" << batch << " > opt_max=" << cfg_.opt_max + << "; clamping\n"; + batch = static_cast(cfg_.opt_max); + } + + // Gate this inference on the upstream reorder/copy completion. + cudaStreamWaitEvent(inf_stream_, input_ready, 0); + + // Latency clock starts once the batch is ready on the stream (post-wait), + // ending at this batch's d2h_event; read one batch late (see below). + cudaEventRecord(start_evt_[parity_], inf_stream_); + + const nvinfer1::Dims4 dims{static_cast(batch), cfg_.channels, cfg_.height, cfg_.width}; + if (!context_->setInputShape(cfg_.input_name.c_str(), dims)) { + std::cerr << "setInputShape failed for batch=" << batch << "\n"; + return; + } + context_->setTensorAddress(cfg_.input_name.c_str(), dev_input); + context_->setTensorAddress(cfg_.output_name.c_str(), trt_out_dev_[parity_]); + + if (!context_->enqueueV3(inf_stream_)) { + std::cerr << "enqueueV3 failed\n"; + return; + } + + const size_t out_bytes = static_cast(batch) * cfg_.feature_dim * sizeof(float); + cudaMemcpyAsync(host_buf_[parity_], trt_out_dev_[parity_], out_bytes, cudaMemcpyDeviceToHost, + inf_stream_); + cudaEventRecord(d2h_event_[parity_], inf_stream_); + + // Back-edge signal: when this fires the input buffer is safe to reuse. + cudaEventRecord(release_evt, inf_stream_); + + // Return the *previous* batch's now-ready buffer for the sink. + const int prev = 1 - parity_; + if (has_pending_[prev]) { + cudaEventSynchronize(d2h_event_[prev]); + float ms = 0.0f; + if (cudaEventElapsedTime(&ms, start_evt_[prev], d2h_event_[prev]) == cudaSuccess) { + batch_latency_ms_.push_back(ms); + } + host_out_prev = host_buf_[prev]; + host_out_prev_n = pending_n_[prev]; + has_pending_[prev] = false; + } + + has_pending_[parity_] = true; + pending_n_[parity_] = batch; + parity_ = prev; + + ++total_batches_inferred_; +} + +void TrtRunner::drain_final(float*& host_out, uint32_t& host_out_n) { + host_out = nullptr; + host_out_n = 0; + for (int i = 0; i < kBuffers; ++i) { + if (has_pending_[i]) { + cudaEventSynchronize(d2h_event_[i]); + float ms = 0.0f; + if (cudaEventElapsedTime(&ms, start_evt_[i], d2h_event_[i]) == cudaSuccess) { + batch_latency_ms_.push_back(ms); + } + host_out = host_buf_[i]; + host_out_n = pending_n_[i]; + has_pending_[i] = false; + return; + } + } +} + +} // namespace daqiri::apps::resnet diff --git a/applications/resnet50_inference/trt_runner.hpp b/applications/resnet50_inference/trt_runner.hpp new file mode 100644 index 0000000..93fc2e6 --- /dev/null +++ b/applications/resnet50_inference/trt_runner.hpp @@ -0,0 +1,133 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 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 + +// Builds (or loads from a disk cache) a TensorRT engine for a ResNet feature +// extractor, then runs FP16 inference per batch with double-buffered async D2H +// to pinned host memory. The caller (inference_pipeline) feeds a contiguous +// FP32 NCHW device buffer + batch and gets back the *previous* batch's host +// feature buffer (now ready, since the stream serializes) for the FeatureSink. +// +// Adapted from the daqiri reorder->inference reference pipeline. The +// adaptation is mechanical but touches several sites versus a 1-channel +// classifier: 3-channel 224x224 input, a wide feature-vector output +// (feature_dim floats per image instead of one logit), and the matching D2H +// byte math. + +#include + +#include +#include +#include + +#include + +// Forward-declare TRT types to keep this header light. +namespace nvinfer1 { +class IRuntime; +class ICudaEngine; +class IExecutionContext; +} // namespace nvinfer1 + +namespace daqiri::apps::resnet { + +struct TrtRunnerConfig { + std::string onnx_path; + std::string engine_path; // cache path; built on first launch if absent + int opt_min = 1; + int opt_avg = 32; + int opt_max = 256; + bool enable_fp16 = true; + int channels = 3; + int height = 224; + int width = 224; + int feature_dim = 2048; // 512 for resnet18/34; 2048 for resnet50/101/152 + int gpu_id = 0; + std::string input_name = "input"; + std::string output_name = "features"; + + static TrtRunnerConfig from_yaml(const YAML::Node& root); +}; + +class TrtRunner { + public: + TrtRunner(TrtRunnerConfig cfg, cudaStream_t inf_stream); + ~TrtRunner(); + + TrtRunner(const TrtRunner&) = delete; + TrtRunner& operator=(const TrtRunner&) = delete; + + void initialize(); + + // Per-batch inference. Stream-waits on input_ready before TRT runs, records + // release_evt after the input has been consumed (the back-edge signal a + // producer reusing the input buffer can wait on). Writes the *previous* + // batch's host-pinned output through host_out_prev/_n (a [n, feature_dim] + // row-major float matrix) so the caller can hand it to the sink. On the first + // call both outputs are null (no previous batch). dev_input is a contiguous + // FP32 NCHW buffer of [batch, channels, height, width]. + void infer(float* dev_input, uint32_t batch, cudaEvent_t input_ready, cudaEvent_t release_evt, + float*& host_out_prev, uint32_t& host_out_prev_n); + + // Synchronously flush the final pending batch at shutdown. + void drain_final(float*& host_out, uint32_t& host_out_n); + + int feature_dim() const { + return cfg_.feature_dim; + } + int opt_max() const { + return cfg_.opt_max; + } + uint64_t total_batches_inferred() const { + return total_batches_inferred_; + } + + // Per-batch inference latency in milliseconds (batch-ready -> features on + // host), one entry per completed batch. Measured with CUDA timing events. + const std::vector& batch_latencies_ms() const { + return batch_latency_ms_; + } + + private: + TrtRunnerConfig cfg_; + cudaStream_t inf_stream_; + + nvinfer1::IRuntime* runtime_ = nullptr; + nvinfer1::ICudaEngine* engine_ = nullptr; + nvinfer1::IExecutionContext* context_ = nullptr; + + // Double-buffered output (parity alternates between batches). + static constexpr int kBuffers = 2; + float* trt_out_dev_[kBuffers] = {nullptr, nullptr}; + float* host_buf_[kBuffers] = {nullptr, nullptr}; + cudaEvent_t d2h_event_[kBuffers] = {nullptr, nullptr}; + // Timing-enabled start marker per buffer: recorded when the batch begins on + // the inference stream; elapsed-to-d2h_event gives the batch latency. + cudaEvent_t start_evt_[kBuffers] = {nullptr, nullptr}; + bool has_pending_[kBuffers] = {false, false}; + uint32_t pending_n_[kBuffers] = {0, 0}; + int parity_ = 0; + + uint64_t total_batches_inferred_ = 0; + std::vector batch_latency_ms_; + + void build_or_load_engine_(); + void allocate_buffers_(); +}; + +} // namespace daqiri::apps::resnet diff --git a/docs/tutorials/daqiri-resnet-inference.md b/docs/tutorials/daqiri-resnet-inference.md new file mode 100644 index 0000000..0f1a441 --- /dev/null +++ b/docs/tutorials/daqiri-resnet-inference.md @@ -0,0 +1,369 @@ +--- +hide: + - navigation +--- + +# DAQIRI → TensorRT ResNet Inference + +This tutorial connects DAQIRI packet ingestion to a GPU inference pipeline: +received packets are reassembled into image tensors on the GPU, run through a +ResNet feature extractor with TensorRT, and summarized in latent space — with no +host bounce on the data path. The full source lives in +`applications/resnet50_inference/`. + +``` +received packets (raw / DPDK GPUDirect) + → GPU sequence-number reorder (image reassembly) + → ResNet feature extraction (TensorRT, FP16) + → per-class mean-feature stats +``` + +It runs over a physical NIC loopback (the DGX Spark p0→p1 cabled loopback used in +[Performance: DGX Spark](../benchmarks/performance-dgx-spark.md)), so it exercises +the real RX path; the build is platform-agnostic and also targets IGX and RTX Pro +servers. + +## Summary + +One host sends preprocessed images out one NIC port and receives them on another +over a cabled loopback. Each image is streamed as ~84 UDP frames, DMA'd straight +into GPU memory (GPUDirect), reassembled and reordered on the GPU, and run through +a ResNet feature extractor — the CPU never touches the pixel data. + +| | | +|---|---| +| **Dataset** | CIFAR-10 — 10 classes, 32×32 color images, upsampled to 224×224 for ResNet input | +| **Model** | ResNet-50 **feature extractor** via TensorRT (FP16) — the 1000-way ImageNet classifier head is replaced with `Identity`, so it emits a 2048-dim pooled feature per image (≈23.5 M params without the head; ≈25.6 M for full ResNet-50). `resnet18/34/101/152` also selectable | +| **Platform** | DGX Spark, ConnectX-7 with a p0→p1 loopback cable (build is platform-agnostic; also targets IGX and RTX Pro) | +| **Data path** | Raw Ethernet / DPDK GPUDirect RX → GPU reorder → TensorRT → per-class mean-feature stats, no host bounce | + +## The pipeline + +A single CUDA stream serializes every GPU stage — reorder → device-to-device batch +copy → inference — so no explicit syncs are needed *between* stages. The only +barrier is a `cudaStreamSynchronize` before each burst is freed, because the +reorder kernel reads the burst's device pointers. + +```mermaid +flowchart TB + subgraph NIC["ConnectX-7 (physical p0 → p1 cable)"] + TX["TX port p0
pcap replay / synthetic frames"] + RX["RX port p1
flow_isolation + UDP flow"] + TX -->|GPUDirect frames| RX + end + RX -->|"get_rx_burst()
device packet pointers"| POLL["RX poller thread"] + POLL -->|"per-burst packet ptrs"| REO["packet reorder into images
(GPU kernel, stream)"] + REO -->|"image complete → D2D copy"| BATCH["NCHW batch buffer (stream)
N images, Channels, Height, Width
record input_ready (batch ready for TRT)"] + BATCH -->|"cudaStreamWaitEvent(input_ready)"| TRT["TrtRunner::enqueueV3 (FP16)
record release_evt (input buffer free)"] + TRT -->|"async D2H + d2h_event
(double-buffered)"| SINK["FeatureSink
features [N × dim]"] + SINK -->|"example mode:
one pass over the dataset"| STATS["per-class mean-feature stats"] + SINK -->|"benchmark mode:
synthetic stream, replayed"| CNT["sample throughput
(img/s)"] + REO -.->|"cudaStreamSynchronize(stream)
BEFORE free"| FREE["free_all_packets_and_burst_rx()"] +``` + +## How it works + +The receive path is one loop on a single CUDA stream. The walkthrough below follows +that loop in order (`applications/resnet50_inference/inference_pipeline.cu`, +`trt_runner.cu`). + +Under the hood the app is multi-threaded: a DAQIRI RX poller thread pulls packets off +the NIC into the burst ring, and a separate **RX worker thread** dequeues those bursts +and runs the reorder kernel *and* TensorRT inference — both on the one CUDA stream. +A TX worker feeds the loopback. The cores are set in the config (values below are the +DGX Spark example): + +| Thread | Core | Role | +| ------ | ---: | ---- | +| DPDK EAL master + stats | 3 | `cfg.master_core` | +| DPDK RX queue poller | 9 | `rx.queues[].cpu_core` — NIC → burst ring | +| App RX worker | 8 | `bench_rx.cpu_core` — `get_rx_burst` → reorder → inference | +| DPDK TX queue poller | 11 | `tx.queues[].cpu_core` — ring → NIC | +| App TX worker | 10 | `bench_tx.cpu_core` — replay/synthesize frames → enqueue | + +The key point the "single stream" framing hides: reorder and inference share **one +thread and one stream** (the RX worker), while a *different* core (the DPDK poller) +fills the burst ring — a producer/consumer split. If the receive path can't keep up, +this core map is what you tune. + +### 1 — One stream, three events + +Everything runs on one non-blocking CUDA stream, so the GPU stages stay ordered +without per-stage synchronization. Three events coordinate the cross-stage and +back-edge signals: + +```cpp +cudaStream_t stream = nullptr; +cudaStreamCreateWithFlags(&stream, cudaStreamNonBlocking); + +cudaEvent_t input_ready = nullptr; // a full NCHW batch is assembled and ready to infer +cudaEvent_t release_evt = nullptr; // the input batch is consumed → buffer safe to reuse +cudaEventCreateWithFlags(&input_ready, cudaEventDisableTiming); +cudaEventCreateWithFlags(&release_evt, cudaEventDisableTiming); +``` + +A third event, `d2h_event`, lives inside the TRT runner: feature vectors are copied +device→host asynchronously and delivered **one batch late** (double-buffered), so the +host never stalls the current batch. + +### 2 — Receive a burst, zero-copy + +DAQIRI delivers packets in **bursts** of DAQIRI-owned buffers. The device pointers in +a burst are valid only until the burst is freed (the +[zero-copy ownership](../concepts.md) rule): + +```cpp +daqiri::BurstParams* burst = nullptr; +if (daqiri::get_rx_burst(&burst, port_id, queue_id) != daqiri::Status::SUCCESS || + burst == nullptr) { + // no packets this poll — back off and retry +} + +const int num_pkts = static_cast(daqiri::get_num_packets(burst)); +for (int i = 0; i < num_pkts; ++i) { + void* dev_ptr = daqiri::get_segment_packet_ptr(burst, /*seg=*/0, /*pkt=*/i); // GPU pointer + // ... +} +``` + +### 3 — Reassemble images across bursts + +A single CIFAR image spans ~84 packets, and an image can straddle a burst boundary. +The pipeline keeps a **persistent, application-owned reorder buffer** +(`daqiri::bench::ReorderPipeline`) and runs the sequence-number reorder kernel **per +burst**, scattering each packet's pixel chunk into `slot = seq % packets_per_image` +*before* the burst is freed. Across bursts the slots accumulate, so a split image +still reassembles correctly. The per-image sequence number occupies the first 4 +payload bytes; the reorder kernel copies pixels starting *after* it, so the sequence +number never overwrites image data. + +This is DAQIRI's own sequence-number GPU reorder kernel +(`packet_reorder_copy_payload_by_sequence`, `src/kernels.cu`) — the same primitive the +`raw_reorder_*` benchmarks use — reused here through `ReorderPipeline`. It is +**application-driven**: DAQIRI does not reorder inside `get_rx_burst()`; the pipeline +invokes the kernel per burst. The config +([`resnet50_wire_loopback.yaml`](https://github.com/NVIDIA/daqiri/blob/main/applications/resnet50_inference/configs/resnet50_wire_loopback.yaml)) +supplies only the packet geometry — `packets_per_image` is *derived* as +`image_bytes / out_payload_len`: + +```yaml +reorder: + out_payload_len: 7168 # pixel-bytes per packet, after the 4-byte sequence prefix + images_per_batch: 32 # images assembled per inference batch +``` + +84 packets × 7168 B = 602,112 B = 3 × 224 × 224 × 4 (FP32 NCHW), so images reassemble +with no padding. + +When an image completes, it is copied into its slot in the contiguous NCHW batch +buffer; once `images_per_batch` images are ready, one inference fires: + +```cpp +pipe.reset_batch(); +for (int i = 0; i < num_pkts; ++i) { + pipe.add_device_packet(daqiri::get_segment_packet_ptr(burst, 0, i)); + if (++img_filled == cfg.packets_per_image) { + const void* ordered = pipe.finish_batch(); // run reorder kernel on stream + cudaMemcpyAsync(d_nchw_batch + img_in_batch * img_bytes, ordered, + img_bytes, cudaMemcpyDeviceToDevice, stream); + ++img_in_batch; + img_filled = 0; + pipe.reset_batch(); + + if (img_in_batch == batch) { // batch full → infer + cudaEventRecord(input_ready, stream); + trt.infer(reinterpret_cast(d_nchw_batch), batch, input_ready, release_evt, + host_prev, n_prev); + if (host_prev != nullptr) sink.consume(host_prev, n_prev); + img_in_batch = 0; + } + } +} + +// End of burst: flush the CURRENT image's this-burst packets into the persistent +// reorder buffer before the burst is freed — its device pointers die at free. +pipe.finish_batch(); +pipe.reset_batch(); +``` + +This trailing `finish_batch()` runs on **every** burst, not just when an image +completes. `add_device_packet()` only records each packet's GPU pointer; the reorder +kernel that copies pixels out of those buffers runs at `finish_batch()`. So when an +image straddles a burst boundary, its partial set of packets must be scattered into +the persistent buffer *before* the burst is freed and those pointers go dead — which +is exactly what lets a split image reassemble across bursts. + +### 4 — Hand off to TensorRT + +The runner gates inference on `input_ready`, binds the NCHW batch and output buffers, +enqueues on the same stream, then records `d2h_event` (output copied to host) and +`release_evt` (input buffer reusable): + +```cpp +cudaStreamWaitEvent(inf_stream_, input_ready, 0); // wait for the assembled batch + +context_->setInputShape(cfg_.input_name.c_str(), dims); // dynamic batch dim +context_->setTensorAddress(cfg_.input_name.c_str(), dev_input); +context_->setTensorAddress(cfg_.output_name.c_str(), trt_out_dev_[parity_]); +context_->enqueueV3(inf_stream_); + +cudaMemcpyAsync(host_buf_[parity_], trt_out_dev_[parity_], out_bytes, + cudaMemcpyDeviceToHost, inf_stream_); +cudaEventRecord(d2h_event_[parity_], inf_stream_); // results ready (one batch late) +cudaEventRecord(release_evt, inf_stream_); // back-edge: input buffer free +``` + +The engine is **built and cached on the first run**, then loaded from cache. The +build requests a mixed FP16 engine from the FP32 ONNX and a dynamic batch dimension: + +```cpp +if (cfg_.enable_fp16) { + config->setFlag(nvinfer1::BuilderFlag::kFP16); +} +// ... build, then cache to cfg_.engine_path ... +std::ofstream out(cfg_.engine_path, std::ios::binary); +out.write(plan_data.data(), plan_data.size()); +``` + +### 5 — Free the burst (the one load-bearing barrier) + +The reorder kernel reads the burst's device pointers, so the burst must not be freed +until the stream drains. This is the only blocking sync in the loop: + +```cpp +cudaStreamSynchronize(stream); // reorder kernel done reading burst ptrs +daqiri::free_all_packets_and_burst_rx(burst); // return buffers to the pool +``` + +Missing the free drains the RX buffer pool, producing `NO_FREE_BURST_BUFFERS` / +`NO_FREE_PACKET_BUFFERS` errors and NIC-level drops. Free every burst, even on error +paths. + +### 6 — Drain the last batch at shutdown + +Because features are delivered **one batch late** (§1), the most recent batch is still +in flight when the loop exits. At shutdown the pipeline first flushes any trailing +partial batch (the dataset size need not be a multiple of `images_per_batch`), then +calls `trt.drain_final()` to synchronize and recover the final batch's features. Skip +this and the last batch is silently lost: + +```cpp +cudaStreamSynchronize(stream); +float* host_final = nullptr; +uint32_t n_final = 0; +trt.drain_final(host_final, n_final); // recover the one-batch-late tail +if (host_final != nullptr) sink.consume(host_final, n_final); +``` + +## Build & run + +Build the container with the `torch` base image (it ships TensorRT plus +torch / torchvision for the offline data prep), then enable the applications tree +(off by default — it requires TensorRT): + +```bash +BASE_IMAGE=torch BASE_TARGET=dpdk DAQIRI_ENGINE="dpdk ibverbs" scripts/build-container.sh + +cmake -S . -B build -DCMAKE_BUILD_TYPE=Release -DBUILD_SHARED_LIBS=ON \ + -DDAQIRI_ENGINE="dpdk ibverbs" -DDAQIRI_BUILD_APPLICATIONS=ON +cmake --build build -j +cmake --install build --prefix /opt/daqiri +``` + +Prepare the model and dataset once, offline, so the runtime pipeline stays pure +*reorder → infer*: + +```bash +# ResNet feature extractor (final FC stripped) → ONNX, dynamic batch axis. +python3 applications/resnet50_inference/tools/export_resnet_onnx.py \ + --model resnet50 --output models/resnet50_features.onnx --check + +# CIFAR-10 → preprocessed framed pcap (+ a ".labels" sidecar). +python3 applications/resnet50_inference/tools/prepare_cifar10_pcap.py \ + --num-images 256 --out data/cifar10_resnet.pcap +``` + +Bring the `dq_wire_*` namespaces down (they hide the ports from DPDK), export the RX +port MAC, fill the PCIe placeholders in `configs/resnet50_wire_loopback.yaml` +(TX = p0, RX = p1), then run: + +```bash +./scripts/setup_spark_wire_loopback_netns.sh down +export ETH_DST_ADDR=$(cat /sys/class/net//address) # RX port (p1) + +./build/applications/resnet50_inference/daqiri_resnet50_inference \ + ./build/applications/resnet50_inference/configs/resnet50_wire_loopback.yaml \ + --dataset data/cifar10_resnet.pcap --seconds 10 +``` + +Example mode replays the dataset **once** by default (use `--loop` to repeat): ResNet +inference is slower than line rate, so replaying once lets the whole dataset buffer in +the RX ring and drain through inference with zero drops, keeping the in-order image +reassembly correct. Confirm nonzero RX with `mlnx_perf`. + +## Example output + +On a successful run the engine loads from cache, a couple of raw feature vectors +print, and at shutdown a per-class mean-feature summary prints: + +```text +Loaded cached engine from models/resnet50_features.fp16.engine (94164328 bytes) + feature[image 0, class 6] = [0.1842, 0.0000, 0.4521, 0.0931, ...] (dim=2048) + feature[image 1, class 9] = [0.0000, 0.2310, 0.1102, 0.3377, ...] (dim=2048) + +=== ResNet inference summary === +images=256 batches=8 seconds=10.00 => 25.60 img/s + +Per-class mean-feature stats (first 8 dims + L2 norm of the mean vector): + class 0 (n= 25): mean=[0.2014, 0.0331, 0.3895, ...] |mean|=3.8421 + class 1 (n= 26): mean=[0.1577, 0.0902, 0.2233, ...] |mean|=3.5108 + ... + class 9 (n= 27): mean=[0.2588, 0.0117, 0.4410, ...] |mean|=4.0072 +(Distinct per-class mean vectors indicate ResNet separates the classes in latent space.) +``` + +The distinct per-class mean vectors (and their differing L2 norms) are a cheap, +dependency-free readout — a quick post-run check that the latent space separates by +class without pulling in a clustering or plotting dependency. + +## Benchmark mode + +Without `--dataset`, the TX synthesizes frames and the sink only counts, so the run +measures the full reorder + inference receive-path throughput. The sweep script builds +the ONNX for each model size as needed and writes `resnet-bench.csv` with img/s for +ResNet-18/34/50/101/152: + +```bash +export ETH_DST_ADDR=$(cat /sys/class/net//address) +BUILD_DIR=./build ./applications/resnet50_inference/tools/run_resnet_bench.sh +``` + +### Results (DGX Spark) + +Measured on a DGX Spark over the ConnectX-7 p0→p1 cable, FP16 engine, batch 32, +30 s × 3 reps (median reported). Params are the feature-extractor counts (the +1000-way ImageNet FC head is stripped). Throughput is reported both as images/s and +as the application data rate in Gb/s (`img/s × 3×224×224×4 B × 8`); the wire rate is +a few percent higher (headers). + +Latency is per **inference batch** (32 images), measured batch-ready → features on +the host with CUDA events. + +| Model | Params (M) | Feature dim | Throughput (img/s) | Throughput (Gb/s) | Latency p50 / p99 (ms) | RX drops | +| --------- | ---------: | ----------: | -----------------: | ----------------: | :--------------------: | :------: | +| ResNet-18 | 11.2 | 512 | 2109 | 10.2 | 3.0 / 5.1 | 0 | +| ResNet-34 | 21.3 | 512 | 1944 | 9.4 | 5.3 / 7.8 | 0 | +| ResNet-50 | 23.5 | 2048 | 1692 | 8.1 | 8.9 / 11.6 | 0 | +| ResNet-101 | 42.5 | 2048 | 1390 | 6.7 | 15.1 / 19.9 | 0 | +| ResNet-152 | 58.1 | 2048 | 1143 | 5.5 | 21.8 / 26.9 | ~19% | + +Throughput falls monotonically with model depth (2109 → 1143 img/s, ~1.85× from +ResNet-18 to ResNet-152) and per-batch latency rises in step (p50 3.0 → 21.8 ms) — +both tracking model FLOPs. The **RX-drop column marks the receive-bound → +compute-bound transition**: everything up through ResNet-101 keeps up **drop-free** +(the receive path sustains ~1390 img/s), and only ResNet-152's inference is slow +enough to back-pressure the RX ring — it drops ~19% of arriving packets +(~698 K of ~3.6 M). Even the top throughput (~10 Gb/s at ResNet-18) is well under +the ~95 Gb/s this link sustains on the raw DPDK bench, so the pipeline is +**GPU-compute bound**, not wire-bound: the single-stream reorder + FP16 inference on +the integrated GB10 is the limiter throughout. diff --git a/mkdocs.yml b/mkdocs.yml index 5867d1d..12ed65c 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -71,6 +71,7 @@ nav: - Bare-Metal CMake Build: tutorials/bare-metal-cmake-build.md - Configuration YAML Walkthrough: tutorials/configuration-walkthrough.md - DAQIRI + Holoscan Integration: tutorials/daqiri-holoscan-integration.md + - DAQIRI + TensorRT Inference: tutorials/daqiri-resnet-inference.md markdown_extensions: - admonition @@ -82,7 +83,11 @@ markdown_extensions: - toc: permalink: true - pymdownx.details - - pymdownx.superfences + - pymdownx.superfences: + custom_fences: + - name: mermaid + class: mermaid + format: !!python/name:pymdownx.superfences.fence_code_format - pymdownx.tabbed: alternate_style: true - pymdownx.highlight: