diff --git a/AGENTS.md b/AGENTS.md index ae942e67..322c09bb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,7 +42,7 @@ There is no unit test suite. Verification is done via the benchmark executables | `daqiri_bench_socket` | `socket_bench.cpp` | `daqiri_bench_socket_{udp,tcp}_tx_rx.yaml`, `daqiri_bench_socket_{udp,tcp}_tx_rx_spark_netns.yaml` (combined-role netns bases) | | `daqiri_pool_ring_bench` | `pool_ring_bench.cpp` | none — microbenchmark comparing `daqiri::Ring`/`daqiri::ObjectPool` vs DPDK `rte_ring`/`rte_mempool` (SPSC/MPMC, single/bulk, thread sweep). The `rte_*` comparison arm compiles only in a DPDK-enabled build; takes no YAML/CLI args | -The four `raw_*` benches share `raw_bench_common.{cpp,h}` and accept `--seconds N`. `daqiri_bench_rdma` and `daqiri_bench_socket` also take `--mode {tx,rx,both}`. +The four `raw_*` benches share `raw_bench_common.{cpp,h}` and accept `--seconds N`. `daqiri_bench_rdma` and `daqiri_bench_socket` also take `--mode {tx,rx,both}`. `daqiri_bench_raw_gpudirect`, `daqiri_bench_raw_hds`, `daqiri_bench_rdma`, and `daqiri_bench_socket` additionally accept `--workload none|fft|gemm|gemm_fp16` — a reusable representative GPU workload (`examples/bench_workload.{h,cu}`, cuFFT/cuBLAS) run once per received reorder window on the **actual received payload**. Each backend first assembles the burst's payloads into one contiguous GPU buffer via `examples/bench_pipeline.{h,cu}` (`ReorderPipeline`): a sequence-number reorder kernel for the out-of-order transports (DPDK raw, UDP) and an arrival-order gather for the in-order ones (RoCE RC, TCP); sockets stage host→device first since their payloads land in pageable host memory. The reorder/gather kernels (`packet_reorder_copy_payload_by_sequence`, `packet_gather_copy_payload`) live in `src/kernels.cu`. `gemm` is FP32 `cublasSgemm`; `gemm_fp16` is the same-size mixed-precision FP16/tensor-core `cublasGemmEx` (inference-style); the contiguous buffer supplies the FFT input / GEMM A operand. `--workload-batch-bytes N` decouples the compute working set from the I/O unit (RoCE sub-chunks each message; raw sizes its reorder window), enabling a batch-size sweep. Used by `run_spark_bench.sh`'s `WORKLOAD` / `WORKLOAD_BATCH` env (all backends) to fill the CSV `post_process` / `post_process_batch` columns (issue #15). ```bash ./build/examples/daqiri_bench_raw_gpudirect ./build/examples/daqiri_bench_raw_tx_rx.yaml --seconds 10 diff --git a/docs/benchmarks/performance-dgx-spark.md b/docs/benchmarks/performance-dgx-spark.md index ecabf556..b0f5014b 100644 --- a/docs/benchmarks/performance-dgx-spark.md +++ b/docs/benchmarks/performance-dgx-spark.md @@ -25,11 +25,11 @@ loopback). The exact commands are collected under [Reproduce](#reproduce) below. | Loopback | Raw/DPDK uses the two physical ports directly; socket/RoCE use the `dq_wire_*` network-namespace wire loopback | | Core pinning | Each direction has a busy-spin queue poller and an app worker on separate isolated X925 cores (PR #149). Single-queue: DPDK pollers 17/18, workers 16/19. Multi-queue: TX pollers 16/19, RX pollers 18/9, each with its own worker core, master 8 (with `isolcpus=5-9,15-19`). | -## Results Summary — native-shape peak (C++ loopback) +## Results Summary (C++ loopback) -Each transport at its best-case operation size. Raw/RoCE are single-stream; -socket TCP/UDP scale with the number of client/server pairs, so the four-pair -aggregate is shown. +Each transport at its best-case **native operation size**. Raw/RoCE are +single-stream; socket TCP/UDP scale with the number of client/server pairs, so the +four-pair aggregate is shown. | Stream / Protocol | Best case | Throughput | Drops | Notes | | ----------------- | --------- | ---------: | ----- | ----- | @@ -235,6 +235,132 @@ shared per-namespace pool, so under multi-pair unpaced load delivery collapses (≈100% loss at 65507 B / 4 pairs). The wire itself is loss-free here; the loss is host-side socket-buffer and reassembly pressure. +## GPU workloads in the receive path + +A common question for a GPU-attached receiver is how much line rate it holds while +the GPU also crunches the incoming data. The benchmarks accept +`--workload none|fft|gemm|gemm_fp16`, exposed by `run_spark_bench.sh` as the +`WORKLOAD` env var (recorded in the CSV `post_process` column); more workload kinds +can be added to the same reusable component over time. The workload runs on the +**actual received packet data** — every backend first assembles the burst's +payloads into one contiguous GPU buffer (a sequence-number **reorder** on the +out-of-order transports, an arrival-order **gather** on the in-order ones) and the +compute consumes that buffer. + +**What the two workloads compute** (the reusable component +`examples/bench_workload.{h,cu}`): + +- **FFT** — a batched 1-D **complex-to-complex forward FFT** via cuFFT + (`cufftExecC2C`). The reordered buffer is treated as an array of single-precision + complex samples and transformed as many independent length-1024 FFTs, batched so + the transforms cover the whole reorder window. This models a streaming + signal-processing receiver — channelization or spectral analysis that FFTs every + frame as it arrives. +- **GEMM** — a dense **matrix multiply** `C = A·B` via cuBLAS on square *n×n* + matrices, with the reordered buffer supplying the *A* operand and *n* chosen so + the matrices match the working-set size. Two precisions: **`gemm`** is FP32 + (`cublasSgemm`); **`gemm_fp16`** is the *same-size* mixed-precision matmul (FP16 + inputs, FP32 accumulate) on the **tensor cores** (`cublasGemmEx`, + `CUBLAS_GEMM_DEFAULT_TENSOR_OP`) — the core op of GPU inference. Running both at + one matrix size isolates the precision / tensor-core effect. This models a + receiver feeding incoming data into a dense linear-algebra or neural-network + stage (beamforming, correlation, an inference layer). + +**The reorder/gather step is per-backend** (`examples/bench_pipeline.{h,cu}`), +chosen to be representative for each transport: + +| Backend | Payload source | Pre-workload step | +| ------- | -------------- | ----------------- | +| Raw / GPUDirect (DPDK) | GPU-accessible RX buffers | **seq reorder** kernel → contiguous device buffer (out-of-order capable) | +| RoCE (RC) | GPU-accessible recv MR | **gather** (in-order); one large message is a zero-copy pass-through | +| UDP sockets | host RX buffers | **host→device stage**, then **seq reorder** | +| TCP sockets | host RX buffers | **host→device stage**, then **gather** (in-order stream) | + +Each compute runs **once per reorder window** on a dedicated CUDA stream (shared +with the reorder/gather kernel so the two serialize without an extra sync), with up +to two kernels in flight (sync depth 2) to overlap GPU work with ingest. The +reorder window is sized so the contiguous buffer is ~8 MB on every backend, giving +a comparable GPU working set across transports. GPU SM% is from `nvidia-smi dmon` +across the run. Throughput is mean ± std over 3 reps, 30 s each. + +!!! note "Where the data lives, per backend" + On the integrated GB10 the GPU shares memory with the CPU, so the raw and RoCE + receive buffers (`host_pinned`) are GPU-accessible with **no copy** — the + reorder/gather kernel reads them in place. Sockets are different: the kernel + hands received bytes to the application in pageable host memory, so the socket + path must **stage each payload host→device** before the GPU can touch it — a + copy on the measured path that the raw/RoCE paths avoid. Lost packets (raw/UDP) + leave their reorder slots zero-filled; the FLOP/copy volume is unchanged. + +Raw / GPUDirect, 8 KB native shape (batch 10240), GPU-resident payloads, seq reorder: + +| Workload | Throughput | Drops | GPU SM% | Notes | +| -------- | ---------: | ----- | ------: | ----- | +| none (baseline) | 98.4 ±0.2 Gb/s | 0 | ~0 | Bare loopback (no GPU compute) | +| FFT | 95.5 ±1.7 Gb/s | 0 | 16.4% | Light compute; line rate held within ~3% | +| GEMM (FP32) | 94.9 ±0.0 Gb/s | 0 | 55.9% | FP32 cores; GPU-bound end, still **drop-free** | +| GEMM (FP16 tensor) | 97.7 ±0.2 Gb/s | 0 | 16.8% | Same matrix, tensor cores; near line rate at a third of the SM | + +RoCE, 8 MB native message (single QP, batch 1), gather pass-through: + +| Workload | Throughput | Drops | GPU SM% | Notes | +| -------- | ---------: | ----- | ------: | ----- | +| none (baseline) | 101.7 ±0.2 Gb/s | 0 | ~0 | Pass-through, no compute | +| FFT | 93.6 ±1.6 Gb/s | 0 | 36.0% | Light compute; ~8% off line rate | +| GEMM (FP32) | 35.0 ±0.3 Gb/s | 0 | 79.0% | GPU-bound at this batch (one GEMM/message); see sweep | +| GEMM (FP16 tensor) | 85.8 ±0.2 Gb/s | 0 | 53.6% | Same matrix, tensor cores | + +Both paths drive the GPU the same way (each holds a batch of received data and drains +the GPU stream **once per batch** — the raw path a burst of ~10 reorder windows, the +RoCE path a small batch of messages whose recv buffers stay live during the pass-through +compute), so compute overlaps with receive on both and every cell is drop-free. The +fixed-batch rows above are the 8 MB native operating point, batch-matched to the raw +path's ~8.19 MB reorder window (1024 packets × 8000 B). + +The remaining RoCE-vs-raw gap on the heavy GEMM is **pipelining depth, not transport**: +at the 8 MB default, one RoCE message is a single GEMM with no neighbor to overlap, while +a raw burst packs ~10 GEMMs back-to-back. Sub-dividing the message into a smaller compute +batch packs several GEMMs per message and recovers most of the throughput — at a **4 MB +batch (2 GEMMs/message), 3-rep**: + +| Workload | Throughput | Drops | GPU SM% | vs 8 MB default | +| -------- | ---------: | ----- | ------: | --------------- | +| GEMM (FP32) | 76.3 ±0.2 Gb/s | 0 | 77.7% | 2.2× (was 35.0) | +| GEMM (FP16 tensor) | 92.3 ±2.0 Gb/s | 0 | 38.7% | within ~5% of raw (was 85.8) | + +The full curve is in the [batch-size sweep](#workload-batch-size-sweep) below. + +### Workload batch-size sweep + +The workload's compute working set is **decoupled from the I/O unit** via +`--workload-batch-bytes` (env `WORKLOAD_BATCH` in `run_spark_bench.sh`): it sets the +bytes fed to one compute call — the GEMM dimension is `n = √(batch/4)` — independent of +the 8 MB RoCE message or 8 KB raw frame. RoCE sub-divides each message into batch-sized +slices (one compute per slice); raw sizes its reorder window to the batch. Sweep run on +`gemm_fp16` (cuBLAS takes the dimension per call); the tables above are the fixed-batch +(8 MB) operating points. + +The two paths respond very differently, which is the whole point. **Raw is flat at line +rate (~98 Gb/s) across every batch** — a raw burst always packs ~10 reorder windows, so +the GPU stays fed regardless of the per-window size; the workload is never the bottleneck. +**RoCE is strongly batch-sensitive** and **non-monotonic with a sweet spot at 2–4 MB**: +one 8 MB message yields a single GEMM with nothing to overlap (underpipelined, 85 Gb/s), +sub-dividing it to 2–4 MB packs 2–4 GEMMs per message and recovers **~92 Gb/s — on par +with raw**, and tiny 512 KB slices collapse to 37 Gb/s on per-call launch/sync overhead +(16 GEMMs/message). So the RoCE↔raw gap at the default batch is purely pipelining depth: +give RoCE a batch that packs a few GEMMs per message and the two paths converge. + +| Batch | RoCE Gb/s | RoCE GPU SM% | Raw Gb/s | Raw GPU SM% | +| ----: | --------: | -----------: | -------: | ----------: | +| 512 KB | 37.0 | 76.9% | 98.0 | 24.8% | +| 1 MB | 76.9 | 75.9% | 98.3 | 16.8% | +| 2 MB | 90.8 | 54.0% | 98.2 | 13.6% | +| 4 MB | 92.5 | 40.4% | 97.8 | 15.1% | +| 8 MB | 85.3 | 52.2% | 96.7 | 16.0% | + +(`gemm_fp16`, single rep per point. Raw cells use the nearest whole-packet window to the +batch size; RoCE caps the batch at the 8 MB message.) + ## Reproduce Run inside the project container (privileged, GPUs passed through, hugepages @@ -294,6 +420,34 @@ before running; tear it down when finished: ./scripts/setup_spark_wire_loopback_netns.sh down # tear down when done ``` +**GPU workload (FFT / GEMM)** re-runs a backend with a representative GPU +workload in the receive path by exporting `WORKLOAD`. It composes with the same +modes and netns setup as above (dpdk in the default namespace, rdma in the +`dq_wire_*` namespaces): + +```bash +# Raw / GPUDirect (netns down, ETH_DST_ADDR exported as above) +WORKLOAD=fft ./examples/run_spark_bench.sh dpdk smoke +WORKLOAD=gemm ./examples/run_spark_bench.sh dpdk smoke +# Socket / RoCE (netns up) +WORKLOAD=fft ./examples/run_spark_bench.sh rdma smoke +WORKLOAD=gemm ./examples/run_spark_bench.sh rdma smoke +``` + +The chosen workload lands in the CSV `post_process` column; compare `gbps` / +`gpu_sm_pct` against the matching `WORKLOAD=none` baseline. + +**Workload batch-size sweep** decouples the compute working set from the I/O unit +by also exporting `WORKLOAD_BATCH` (bytes); it lands in the CSV `post_process_batch` +column: + +```bash +for B in 262144 524288 1048576 2097152 4194304 8388608; do + WORKLOAD=gemm_fp16 WORKLOAD_BATCH=$B ./examples/run_spark_bench.sh rdma smoke # netns up +done +# raw: netns down, ETH_DST_ADDR exported; same loop with `dpdk` +``` + Each run writes `bench-results/--/runs.csv`. See [Socket and RDMA Benchmarking](socket_benchmarking.md) and [Raw Ethernet Benchmarking](raw_benchmarking.md) for the namespace setup and diff --git a/docs/benchmarks/raw_benchmarking.md b/docs/benchmarks/raw_benchmarking.md index b47d3252..fa18426b 100644 --- a/docs/benchmarks/raw_benchmarking.md +++ b/docs/benchmarks/raw_benchmarking.md @@ -238,6 +238,8 @@ After having modified the configuration file, ensure you have connected an SFP c By default the application runs for 10 seconds and then exits. You can change the duration by passing `--seconds ` after the YAML path, or stop it gracefully at any time with `Ctrl-C`. +`daqiri_bench_raw_gpudirect` and `daqiri_bench_raw_hds` also accept `--workload none|fft|gemm|gemm_fp16`, which runs a representative GPU workload once per received reorder window on the **actual received packet data**: `fft` (batched cuFFT C2C transform), `gemm` (FP32 `cublasSgemm`), or `gemm_fp16` (the same-size mixed-precision FP16/tensor-core matmul that models inference). Each received burst's payloads are first reordered by sequence number into a contiguous GPU buffer (`examples/bench_pipeline.{h,cu}`) that the compute then consumes. The same `--workload` flag is honoured by the RoCE bench (`daqiri_bench_rdma`, in-order gather) and the socket bench (`daqiri_bench_socket`, host→device stage then UDP reorder / TCP gather). See the [DGX Spark GPU-workload results](performance-dgx-spark.md#gpu-workloads-in-the-receive-path). + ## Flow programming smoke test Raw Ethernet flow rules are programmed into the NIC during `daqiri_init()`. Software loopback diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index fa97b1c6..be938e4a 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -85,9 +85,10 @@ function(link_daqiri_bench target) endfunction() function(add_daqiri_raw_bench target source) - add_executable(${target} ${source} raw_bench_common.cpp raw_bench_common_cuda.cu) + add_executable(${target} ${source} raw_bench_common.cpp raw_bench_common_cuda.cu + bench_workload.cu bench_pipeline.cu) link_daqiri_bench(${target}) - target_link_libraries(${target} PRIVATE CUDA::cudart) + target_link_libraries(${target} PRIVATE CUDA::cudart CUDA::cufft CUDA::cublas) set_target_properties(${target} PROPERTIES CUDA_ARCHITECTURES "${DAQIRI_EXAMPLES_CUDA_ARCHITECTURES}" BUILD_RPATH "$ORIGIN/../src;$ORIGIN/../src/third_party/yaml-cpp" @@ -115,17 +116,24 @@ add_daqiri_raw_bench(daqiri_example_dynamic_rx_flow dynamic_rx_flow_example.cpp) add_daqiri_raw_bench(daqiri_example_gds_write gds_write_example.cpp) add_daqiri_raw_bench(daqiri_example_pcap_writer pcap_writer_example.cpp) -add_executable(daqiri_bench_rdma rdma_bench.cpp raw_bench_common.cpp) +add_executable(daqiri_bench_rdma rdma_bench.cpp raw_bench_common.cpp + bench_workload.cu bench_pipeline.cu) link_daqiri_bench(daqiri_bench_rdma) -target_link_libraries(daqiri_bench_rdma PRIVATE CUDA::cudart) +target_link_libraries(daqiri_bench_rdma PRIVATE CUDA::cudart CUDA::cufft CUDA::cublas) set_target_properties(daqiri_bench_rdma PROPERTIES + CUDA_ARCHITECTURES "${DAQIRI_EXAMPLES_CUDA_ARCHITECTURES}" BUILD_RPATH "$ORIGIN/../src;$ORIGIN/../src/third_party/yaml-cpp" ) -add_executable(daqiri_bench_socket socket_bench.cpp raw_bench_common.cpp) +# raw_bench_common.cpp's rx_count_worker references GpuWorkload + ReorderPipeline, +# so the socket bench must also compile/link bench_workload.cu + bench_pipeline.cu +# (and the math libs). +add_executable(daqiri_bench_socket socket_bench.cpp raw_bench_common.cpp + bench_workload.cu bench_pipeline.cu) link_daqiri_bench(daqiri_bench_socket) -target_link_libraries(daqiri_bench_socket PRIVATE CUDA::cudart) +target_link_libraries(daqiri_bench_socket PRIVATE CUDA::cudart CUDA::cufft CUDA::cublas) set_target_properties(daqiri_bench_socket PROPERTIES + CUDA_ARCHITECTURES "${DAQIRI_EXAMPLES_CUDA_ARCHITECTURES}" BUILD_RPATH "$ORIGIN/../src;$ORIGIN/../src/third_party/yaml-cpp" ) diff --git a/examples/bench_pipeline.cu b/examples/bench_pipeline.cu new file mode 100644 index 00000000..7e0e4c45 --- /dev/null +++ b/examples/bench_pipeline.cu @@ -0,0 +1,174 @@ +/* + * 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 "bench_pipeline.h" + +#include + +#include +#include +#include + +#include "../src/kernels.h" + +namespace daqiri::bench { + +namespace { +// kReorderDataTypeSame / kReorderEndiannessNetwork from src/kernels.cu — a pure +// vectorized copy (no quantization), reading the seq number network-byte-order. +constexpr uint8_t kDataTypeSame = 0; +constexpr uint8_t kEndianNetwork = 1; + +cudaStream_t as_stream(void* s) { + return static_cast(s); +} +} // namespace + +ReorderPipeline::~ReorderPipeline() { + destroy(); +} + +void ReorderPipeline::destroy() { + if (ordered_ != nullptr) { + cudaFree(ordered_); + ordered_ = nullptr; + } + if (staging_ != nullptr) { + cudaFree(staging_); + staging_ = nullptr; + } + if (dev_ptrs_ != nullptr) { + cudaFree(dev_ptrs_); + dev_ptrs_ = nullptr; + } + if (host_ptrs_ != nullptr) { + delete[] static_cast(host_ptrs_); + host_ptrs_ = nullptr; + } + stream_ = nullptr; // not owned + ok_ = false; +} + +bool ReorderPipeline::init(ReorderMode mode, uint32_t packets_per_batch, uint32_t out_payload_len, + uint32_t payload_byte_offset, uint16_t seq_bit_offset, + uint8_t seq_bit_width, bool staging_needed, void* stream) { + mode_ = mode; + staging_needed_ = staging_needed; + packets_per_batch_ = packets_per_batch; + out_payload_len_ = out_payload_len; + payload_byte_offset_ = payload_byte_offset; + seq_bit_offset_ = seq_bit_offset; + seq_bit_width_ = seq_bit_width; + collected_ = 0; + staged_ = 0; + + if (stream == nullptr || packets_per_batch == 0 || out_payload_len == 0) { + ok_ = false; // disabled (e.g. workload == none): inert no-op, not an error + return true; + } + stream_ = stream; + + batch_bytes_ = static_cast(packets_per_batch) * out_payload_len; + stage_slot_bytes_ = static_cast(payload_byte_offset) + out_payload_len; + + if (cudaMalloc(&ordered_, batch_bytes_) != cudaSuccess || + cudaMemset(ordered_, 0, batch_bytes_) != cudaSuccess) { + std::cerr << "ReorderPipeline: ordered buffer alloc failed\n"; + destroy(); + return false; + } + if (cudaMalloc(&dev_ptrs_, static_cast(packets_per_batch) * sizeof(const void*)) != + cudaSuccess) { + std::cerr << "ReorderPipeline: pointer array alloc failed\n"; + destroy(); + return false; + } + host_ptrs_ = new (std::nothrow) const void*[packets_per_batch]; + if (host_ptrs_ == nullptr) { + std::cerr << "ReorderPipeline: host pointer array alloc failed\n"; + destroy(); + return false; + } + if (staging_needed_) { + if (cudaMalloc(&staging_, static_cast(packets_per_batch) * stage_slot_bytes_) != + cudaSuccess) { + std::cerr << "ReorderPipeline: staging buffer alloc failed\n"; + destroy(); + return false; + } + } + + ok_ = true; + return true; +} + +void ReorderPipeline::reset_batch() { + collected_ = 0; + staged_ = 0; +} + +void ReorderPipeline::add_device_packet(const void* dptr) { + if (!ok_ || dptr == nullptr || collected_ >= packets_per_batch_) { + return; + } + static_cast(host_ptrs_)[collected_++] = dptr; +} + +void* ReorderPipeline::stage_host_packet(const void* hptr, uint32_t len) { + if (!ok_ || !staging_needed_ || hptr == nullptr || staged_ >= packets_per_batch_) { + return nullptr; + } + const uint32_t copy_len = + len < stage_slot_bytes_ ? len : static_cast(stage_slot_bytes_); + auto* dst = static_cast(staging_) + static_cast(staged_) * stage_slot_bytes_; + cudaMemcpyAsync(dst, hptr, copy_len, cudaMemcpyHostToDevice, as_stream(stream_)); + ++staged_; + return dst; +} + +const void* ReorderPipeline::finish_batch() { + if (!ok_ || collected_ == 0) { + return nullptr; + } + + // Single-packet in-order case (one large RoCE/TCP message): the source buffer + // is already contiguous and device-resident, so feed it straight to the + // workload with no kernel and no copy. + if (mode_ == ReorderMode::GatherOnly && collected_ == 1) { + const auto* src = static_cast(static_cast(host_ptrs_)[0]); + return src + payload_byte_offset_; + } + + cudaMemcpyAsync(dev_ptrs_, host_ptrs_, static_cast(collected_) * sizeof(const void*), + cudaMemcpyHostToDevice, as_stream(stream_)); + + const auto* const* in = static_cast(dev_ptrs_); + if (mode_ == ReorderMode::SeqReorder) { + packet_reorder_copy_payload_by_sequence( + ordered_, in, out_payload_len_, out_payload_len_, payload_byte_offset_, collected_, + seq_bit_offset_, seq_bit_width_, /*batch_bit_offset=*/0, + /*batch_bit_width=*/0, /*has_batch_number=*/0, packets_per_batch_, + /*max_slot_idx=*/packets_per_batch_ - 1, kDataTypeSame, kDataTypeSame, kEndianNetwork, + /*batch_id_out=*/nullptr, as_stream(stream_)); + } else { + packet_gather_copy_payload(ordered_, in, out_payload_len_, payload_byte_offset_, collected_, + as_stream(stream_)); + } + return ordered_; +} + +} // namespace daqiri::bench diff --git a/examples/bench_pipeline.h b/examples/bench_pipeline.h new file mode 100644 index 00000000..9393d481 --- /dev/null +++ b/examples/bench_pipeline.h @@ -0,0 +1,123 @@ +/* + * 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 + +namespace daqiri::bench { + +// How a burst's packets are assembled into the contiguous device buffer that the +// GpuWorkload then consumes. +enum class ReorderMode { + // Out-of-order capable transports (DPDK raw, UDP): read a per-packet sequence + // number and place each payload at slot = seq % packets_per_batch. + SeqReorder, + // Reliable / in-order transports (RoCE RC, TCP): place each payload at its + // arrival index (no reshuffle). When a single already-device-resident buffer + // makes up the batch (one large message), this is a zero-copy pass-through. + GatherOnly, +}; + +// Per-RX-thread helper that turns a burst of received packets into one +// contiguous device buffer for the GpuWorkload, doing the reorder/gather (and, +// for sockets, the host->device staging) on the workload's CUDA stream so the +// kernel orders before the workload with no extra synchronization. +// +// CUDA types are hidden behind opaque void* so this header is includable from +// plain .cpp bench mains (same pattern as bench_workload.h). +// +// Usage per burst: +// pipe.reset_batch(); +// for each packet p: +// const void* d = staging_needed ? pipe.stage_host_packet(host_ptr, len) +// : device_ptr; // GPU-accessible src +// pipe.add_device_packet(d); +// const void* ordered = pipe.finish_batch(); // launches kernel +// workload.run(ordered); +class ReorderPipeline { + public: + ReorderPipeline() = default; + ~ReorderPipeline(); + ReorderPipeline(const ReorderPipeline&) = delete; + ReorderPipeline& operator=(const ReorderPipeline&) = delete; + + // packets_per_batch slots of out_payload_len bytes form the contiguous output + // buffer. payload_byte_offset / seq_bit_offset / seq_bit_width describe the + // packet layout (seq fields ignored in GatherOnly). staging_needed allocates a + // device staging buffer for host sources (sockets). stream is the + // GpuWorkload's cudaStream_t (share it); pass nullptr to disable the pipeline. + // Returns false on CUDA error (caller may warn and continue disabled). + bool init(ReorderMode mode, uint32_t packets_per_batch, uint32_t out_payload_len, + uint32_t payload_byte_offset, uint16_t seq_bit_offset, uint8_t seq_bit_width, + bool staging_needed, void* stream); + + // Begin accumulating a new batch. + void reset_batch(); + + // Record one GPU-accessible source pointer (packet base, before + // payload_byte_offset). Ignored once packets_per_batch is reached. + void add_device_packet(const void* dptr); + + // Copy `len` host bytes into the device staging buffer and return the device + // pointer to feed add_device_packet(). Returns nullptr if disabled/full. + void* stage_host_packet(const void* hptr, uint32_t len); + + uint32_t collected() const { + return collected_; + } + + // Launch the reorder/gather kernel for the collected packets and return the + // contiguous ordered device buffer (packets_per_batch * out_payload_len) to + // hand to GpuWorkload::run(). For the single-packet GatherOnly pass-through it + // returns the source pointer directly (no kernel, no copy). Returns nullptr if + // disabled or nothing was collected. + const void* finish_batch(); + + size_t batch_bytes() const { + return batch_bytes_; + } + bool enabled() const { + return ok_; + } + + private: + void destroy(); + + ReorderMode mode_ = ReorderMode::GatherOnly; + bool ok_ = false; + bool staging_needed_ = false; + uint32_t packets_per_batch_ = 0; + uint32_t out_payload_len_ = 0; + uint32_t payload_byte_offset_ = 0; + uint16_t seq_bit_offset_ = 0; + uint8_t seq_bit_width_ = 0; + size_t batch_bytes_ = 0; + size_t stage_slot_bytes_ = 0; + + uint32_t collected_ = 0; + uint32_t staged_ = 0; + + void* stream_ = nullptr; // cudaStream_t (shared, not owned) + void* ordered_ = nullptr; // device output buffer (owned) + void* staging_ = nullptr; // device staging buffer (owned, sockets only) + void* dev_ptrs_ = nullptr; // device void*[packets_per_batch] (owned) + void* host_ptrs_ = nullptr; // host const void*[packets_per_batch] (owned) +}; + +} // namespace daqiri::bench diff --git a/examples/bench_workload.cu b/examples/bench_workload.cu new file mode 100644 index 00000000..4d602da1 --- /dev/null +++ b/examples/bench_workload.cu @@ -0,0 +1,278 @@ +/* + * 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 "bench_workload.h" + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace daqiri::bench { +namespace { + +// 1D FFT length; the burst working set is fanned out across as many batched +// transforms of this length as fit. +constexpr int kFftLen = 1024; +// Default working-set size when the caller passes bytes_per_burst == 0. +constexpr size_t kDefaultBytes = 1u << 16; // 64 KiB + +cufftHandle as_fft_plan(int p) { + return static_cast(p); +} +cudaStream_t as_stream(void* s) { + return static_cast(s); +} +cublasHandle_t as_cublas(void* h) { + return static_cast(h); +} + +} // namespace + +BenchWorkload parse_workload(int argc, char** argv) { + BenchWorkload workload = BenchWorkload::None; + for (int i = 2; i + 1 < argc; i += 2) { + if (std::string(argv[i]) == "--workload") { + const std::string val = argv[i + 1]; + if (val == "fft") { + workload = BenchWorkload::Fft; + } else if (val == "gemm") { + workload = BenchWorkload::Gemm; + } else if (val == "gemm_fp16" || val == "gemm-fp16") { + workload = BenchWorkload::GemmFp16; + } else if (val == "none") { + workload = BenchWorkload::None; + } else { + std::cerr << "Unknown --workload value '" << val + << "' (expected none|fft|gemm|gemm_fp16); using none\n"; + workload = BenchWorkload::None; + } + } + } + return workload; +} + +size_t parse_workload_batch_bytes(int argc, char** argv) { + for (int i = 2; i + 1 < argc; i += 2) { + if (std::string(argv[i]) == "--workload-batch-bytes") { + const long long v = std::atoll(argv[i + 1]); + if (v > 0) { + return static_cast(v); + } + } + } + return 0; +} + +const char* workload_name(BenchWorkload workload) { + switch (workload) { + case BenchWorkload::Fft: + return "fft"; + case BenchWorkload::Gemm: + return "gemm"; + case BenchWorkload::GemmFp16: + return "gemm_fp16"; + case BenchWorkload::None: + default: + return "none"; + } +} + +GpuWorkload::~GpuWorkload() { + destroy(); +} + +void GpuWorkload::destroy() { + if (fft_plan_ >= 0) { + cufftDestroy(as_fft_plan(fft_plan_)); + fft_plan_ = -1; + } + if (cublas_ != nullptr) { + cublasDestroy(as_cublas(cublas_)); + cublas_ = nullptr; + } + if (fft_out_ != nullptr) { + cudaFree(fft_out_); + fft_out_ = nullptr; + } + if (gemm_b_ != nullptr) { + cudaFree(gemm_b_); + gemm_b_ = nullptr; + } + if (gemm_c_ != nullptr) { + cudaFree(gemm_c_); + gemm_c_ = nullptr; + } + if (stream_ != nullptr) { + cudaStreamDestroy(as_stream(stream_)); + stream_ = nullptr; + } + ok_ = false; +} + +bool GpuWorkload::init(BenchWorkload kind, size_t batch_bytes, int sync_interval) { + kind_ = kind; + sync_interval_ = sync_interval > 0 ? sync_interval : 1; + run_count_ = 0; + if (kind_ == BenchWorkload::None) { + ok_ = false; + return true; // inert no-op object; not an error + } + + const size_t bytes = batch_bytes > 0 ? batch_bytes : kDefaultBytes; + + cudaStream_t stream = nullptr; + if (cudaStreamCreate(&stream) != cudaSuccess) { + std::cerr << "GpuWorkload: cudaStreamCreate failed\n"; + destroy(); + return false; + } + stream_ = stream; + + if (kind_ == BenchWorkload::Fft) { + // Fan the working set out across batched length-kFftLen C2C transforms. The + // transform reads the caller's input buffer and writes the owned fft_out_; + // total*sizeof(cufftComplex) <= bytes so the input always holds enough data. + const size_t n_complex = std::max(kFftLen, bytes / sizeof(cufftComplex)); + const int batch = std::max(1, static_cast(n_complex / kFftLen)); + const size_t total = static_cast(kFftLen) * batch; + + if (cudaMalloc(&fft_out_, total * sizeof(cufftComplex)) != cudaSuccess || + cudaMemset(fft_out_, 0, total * sizeof(cufftComplex)) != cudaSuccess) { + std::cerr << "GpuWorkload: FFT buffer alloc failed\n"; + destroy(); + return false; + } + cufftHandle plan = 0; + if (cufftPlan1d(&plan, kFftLen, CUFFT_C2C, batch) != CUFFT_SUCCESS) { + std::cerr << "GpuWorkload: cufftPlan1d failed\n"; + destroy(); + return false; + } + fft_plan_ = static_cast(plan); + if (cufftSetStream(as_fft_plan(fft_plan_), stream) != CUFFT_SUCCESS) { + std::cerr << "GpuWorkload: cufftSetStream failed\n"; + destroy(); + return false; + } + } else { // Gemm or GemmFp16 + // Square matmul whose matrices match the working set. Size n from FP32 in + // both cases so gemm and gemm_fp16 use the SAME dimension -> identical FLOP + // count, isolating the precision / tensor-core effect. The A operand is the + // caller's input buffer (n*n*elem_size <= bytes); B and C are owned scratch. + int n = static_cast(std::sqrt(static_cast(bytes) / sizeof(float))); + n = std::max(64, (n / 8) * 8); // multiple of 8, sane floor + gemm_n_ = n; + const size_t elems = static_cast(n) * n; + const size_t elem_size = kind_ == BenchWorkload::GemmFp16 ? sizeof(__half) : sizeof(float); + if (cudaMalloc(&gemm_b_, elems * elem_size) != cudaSuccess || + cudaMalloc(&gemm_c_, elems * elem_size) != cudaSuccess || + cudaMemset(gemm_b_, 0, elems * elem_size) != cudaSuccess) { + std::cerr << "GpuWorkload: GEMM buffer alloc failed\n"; + destroy(); + return false; + } + cublasHandle_t handle = nullptr; + if (cublasCreate(&handle) != CUBLAS_STATUS_SUCCESS) { + std::cerr << "GpuWorkload: cublasCreate failed\n"; + destroy(); + return false; + } + cublas_ = handle; + if (cublasSetStream(handle, stream) != CUBLAS_STATUS_SUCCESS) { + std::cerr << "GpuWorkload: cublasSetStream failed\n"; + destroy(); + return false; + } + } + + // Warm up and validate the chosen op once against a transient zeroed input + // buffer: this surfaces a misconfigured cuFFT/cuBLAS call (which would + // otherwise no-op silently on the hot path and look like a free workload) and + // excludes one-time library setup from the measured run. The warmup buffer is + // freed immediately; the measured runs read the caller's received-data buffer. + void* warmup_in = nullptr; + const bool warmup_ok = cudaMalloc(&warmup_in, bytes) == cudaSuccess && + cudaMemset(warmup_in, 0, bytes) == cudaSuccess && issue_op(warmup_in) && + cudaStreamSynchronize(stream) == cudaSuccess; + if (warmup_in != nullptr) { + cudaFree(warmup_in); + } + if (!warmup_ok) { + std::cerr << "GpuWorkload: warmup of '" << workload_name(kind_) + << "' failed (cuda=" << cudaGetErrorString(cudaGetLastError()) << ")\n"; + destroy(); + return false; + } + + ok_ = true; + return true; +} + +bool GpuWorkload::issue_op(const void* input) { + if (kind_ == BenchWorkload::Fft) { + return cufftExecC2C(as_fft_plan(fft_plan_), + const_cast(static_cast(input)), + static_cast(fft_out_), CUFFT_FORWARD) == CUFFT_SUCCESS; + } + const float alpha = 1.0f; + const float beta = 0.0f; + if (kind_ == BenchWorkload::Gemm) { + return cublasSgemm(as_cublas(cublas_), CUBLAS_OP_N, CUBLAS_OP_N, gemm_n_, gemm_n_, gemm_n_, + &alpha, static_cast(input), gemm_n_, + static_cast(gemm_b_), gemm_n_, &beta, + static_cast(gemm_c_), gemm_n_) == CUBLAS_STATUS_SUCCESS; + } + // GemmFp16: FP16 inputs, FP32 accumulate, on the tensor cores. + return cublasGemmEx(as_cublas(cublas_), CUBLAS_OP_N, CUBLAS_OP_N, gemm_n_, gemm_n_, gemm_n_, + &alpha, input, CUDA_R_16F, gemm_n_, gemm_b_, CUDA_R_16F, gemm_n_, &beta, + gemm_c_, CUDA_R_16F, gemm_n_, CUBLAS_COMPUTE_32F, + CUBLAS_GEMM_DEFAULT_TENSOR_OP) == CUBLAS_STATUS_SUCCESS; +} + +void GpuWorkload::run(const void* input) { + if (!enabled() || input == nullptr) { + return; + } + issue_op(input); + ++run_count_; + maybe_sync(); +} + +void GpuWorkload::maybe_sync() { + if (!enabled()) { + return; + } + if (run_count_ % static_cast(sync_interval_) == 0) { + cudaStreamSynchronize(as_stream(stream_)); + } +} + +void GpuWorkload::sync() { + if (stream_ != nullptr) { + cudaStreamSynchronize(as_stream(stream_)); + } +} + +} // namespace daqiri::bench diff --git a/examples/bench_workload.h b/examples/bench_workload.h new file mode 100644 index 00000000..63ccbead --- /dev/null +++ b/examples/bench_workload.h @@ -0,0 +1,126 @@ +/* + * 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 + +namespace daqiri::bench { + +// Representative GPU workloads that can be dropped into any benchmark's receive +// path to model downstream GPU compute. Bare loopback (None) vs loopback + FFT +// vs loopback + GEMM. GemmFp16 is the same square matmul as Gemm but in +// mixed-precision (FP16 inputs, FP32 accumulate) on the tensor cores — the core +// op of GPU inference, and far faster than the FP32 path on tensor-core GPUs. +enum class BenchWorkload { None, Fft, Gemm, GemmFp16 }; + +// Parse "--workload none|fft|gemm|gemm_fp16" from argv (default None). Mirrors +// the flag/value stride used by parse_run_seconds / parse_target_gbps. +BenchWorkload parse_workload(int argc, char** argv); + +// Parse "--workload-batch-bytes N" from argv: the working-set size (bytes) fed to +// one compute call, decoupled from the I/O unit (RoCE message / raw frame). The +// GEMM matrix dimension and FFT batch scale from it. Returns 0 if unset (the bench +// then falls back to its backend-default batch). Mirrors parse_workload's stride. +size_t parse_workload_batch_bytes(int argc, char** argv); + +// Lower-case name ("none"/"fft"/"gemm"); used for the run_spark_bench.sh +// post_process CSV column and log lines. +const char* workload_name(BenchWorkload workload); + +// Engine-agnostic representative GPU compute, run once per received burst on the +// ACTUAL received packet data. +// +// The caller hands run() a contiguous device buffer holding the reordered / +// gathered payload of one burst (see ReorderPipeline in bench_pipeline.h). The +// op reads that buffer as its input operand: FFT transforms it in place of the +// scratch input; GEMM uses it as the A matrix. The bytes are arbitrary packet +// data, which is fine for a throughput benchmark — the FLOP profile and memory +// footprint are unchanged; only the input source differs. This makes the +// measurement an honest end-to-end "receive then process the data" cost. +// +// Owns its own CUDA stream, cuFFT plan, and cuBLAS handle. cuFFT plans and +// cuBLAS handles are not safe to share across threads, so construct one +// GpuWorkload per RX worker thread (each multi-queue RX thread gets its own). +// The stream is shared with the ReorderPipeline so the reorder/gather kernel and +// the workload are serialized on the same stream without an explicit sync. +class GpuWorkload { + public: + GpuWorkload() = default; + ~GpuWorkload(); + GpuWorkload(const GpuWorkload&) = delete; + GpuWorkload& operator=(const GpuWorkload&) = delete; + + // Build the plan/handle and size the problem to the contiguous input buffer of + // batch_bytes the caller will pass to run() (0 => an internal default). The op + // reads at most batch_bytes from that buffer. kind == None leaves the object an + // inert no-op. sync_interval bounds outstanding GPU work (sync every N runs). + // Returns false on CUDA / library error; the caller may warn and continue with + // the workload disabled (enabled() will report false). + bool init(BenchWorkload kind, size_t batch_bytes, int sync_interval = 2); + + // Enqueue one representative FFT/SGEMM on the internal stream, reading `input` + // (a device pointer to >= batch_bytes valid bytes). No-op unless enabled(). + void run(const void* input); + + // Every sync_interval runs, block until the stream drains so the GPU stays on + // the critical path without unbounded queueing. + void maybe_sync(); + + // Drain any remaining queued work (call once on shutdown). + void sync(); + + bool enabled() const { + return kind_ != BenchWorkload::None && ok_; + } + BenchWorkload kind() const { + return kind_; + } + + // CUDA stream (cudaStream_t) this workload runs on; share it with the + // ReorderPipeline so the reorder/gather kernel orders before run(). null when + // the workload is disabled. + void* stream() const { + return stream_; + } + + private: + void destroy(); + // Enqueue one op on the stream reading `input`; returns false if the + // cuFFT/cuBLAS call reports an error at enqueue. Shared by run() (hot path) + // and init() (warmup/validate, against an internal zeroed buffer). + bool issue_op(const void* input); + + BenchWorkload kind_ = BenchWorkload::None; + bool ok_ = false; + int sync_interval_ = 16; + unsigned long long run_count_ = 0; + + // Opaque handles, cast in the .cu so this header stays free of CUDA library + // includes (it is included from plain .cpp bench mains). + void* stream_ = nullptr; // cudaStream_t + void* cublas_ = nullptr; // cublasHandle_t + int fft_plan_ = -1; // cufftHandle (-1 == unset) + + // Device scratch (operands NOT sourced from received data). + void* fft_out_ = nullptr; // cufftComplex[fft_total_] (FFT output) + void* gemm_b_ = nullptr; // B operand + void* gemm_c_ = nullptr; // C output + int gemm_n_ = 0; +}; + +} // namespace daqiri::bench diff --git a/examples/daqiri_bench_rdma_tx_rx_spark_netns.yaml b/examples/daqiri_bench_rdma_tx_rx_spark_netns.yaml index 6556aa18..510309b9 100644 --- a/examples/daqiri_bench_rdma_tx_rx_spark_netns.yaml +++ b/examples/daqiri_bench_rdma_tx_rx_spark_netns.yaml @@ -80,7 +80,10 @@ daqiri: address: 10.250.0.1 socket_config: mode: client - remote_addr: "roce://10.250.0.2:4096" + # Client binds its own RoCE address here; the server peer endpoint lives + # in the rdma_bench_client app config (server_address/server_port). Since + # #156 a RoCE client's socket_config.remote_addr is rejected by daqiri_init. + local_addr: "roce://10.250.0.1" roce_config: transport_mode: RC tx: diff --git a/examples/raw_bench_common.cpp b/examples/raw_bench_common.cpp index e46847b2..1d514a17 100644 --- a/examples/raw_bench_common.cpp +++ b/examples/raw_bench_common.cpp @@ -548,12 +548,31 @@ void print_queue_stats(const char *direction, const std::string &interface_name, << std::endl; } -void rx_count_worker(const RawBenchRxConfig &cfg, std::atomic &stop) { +void rx_count_worker(const RawBenchRxConfig& cfg, std::atomic& stop, BenchWorkload workload, + const ReorderGeometry& geom) { if (!set_current_thread_affinity(cfg.cpu_core, "bench_rx")) { stop.store(true); return; } + // Per-thread representative GPU workload + reorder pipeline (both inert unless + // --workload is set). If init fails we warn and keep counting so the run still + // produces throughput. The pipeline shares the workload's CUDA stream so the + // reorder kernel orders before the compute with no extra sync. + const uint32_t out_payload_len = geom.out_payload_len > 0 ? geom.out_payload_len : 1; + const size_t batch_bytes = static_cast(geom.packets_per_batch) * out_payload_len; + GpuWorkload gpu_workload; + ReorderPipeline pipeline; + if (workload != BenchWorkload::None) { + if (!gpu_workload.init(workload, batch_bytes) || + !pipeline.init(ReorderMode::SeqReorder, geom.packets_per_batch, out_payload_len, + geom.payload_byte_offset, geom.seq_bit_offset, geom.seq_bit_width, + /*staging_needed=*/false, gpu_workload.stream())) { + std::cerr << "RX workload init failed; continuing without GPU workload\n"; + } + } + const bool run_workload = gpu_workload.enabled() && pipeline.enabled(); + const int port_id = daqiri::get_port_id(cfg.interface_name); if (port_id < 0) { std::cerr << "Invalid RX interface_name: " << cfg.interface_name << "\n"; @@ -591,15 +610,35 @@ void rx_count_worker(const RawBenchRxConfig &cfg, std::atomic &stop) { } got_any = true; auto &stats = queue_stats[static_cast(q)]; - stats.packets += static_cast(daqiri::get_num_packets(burst)); + const int num_pkts = static_cast(daqiri::get_num_packets(burst)); + stats.packets += static_cast(num_pkts); stats.bytes += daqiri::get_burst_tot_byte(burst); ++stats.bursts; + + if (run_workload) { + // Reorder the real received payloads into a contiguous device buffer and + // run the compute on it. Packet pointers are only valid until the burst + // is freed, so process complete batches within this burst (the remainder + // < packets_per_batch is dropped), then drain the stream so the reorder + // kernel has finished reading the buffers before they are recycled. + pipeline.reset_batch(); + for (int i = 0; i < num_pkts; ++i) { + pipeline.add_device_packet( + daqiri::get_segment_packet_ptr(burst, geom.payload_segment, i)); + if (pipeline.collected() == geom.packets_per_batch) { + gpu_workload.run(pipeline.finish_batch()); + pipeline.reset_batch(); + } + } + gpu_workload.sync(); + } daqiri::free_all_packets_and_burst_rx(burst); } if (!got_any) { std::this_thread::sleep_for(std::chrono::microseconds(100)); } } + gpu_workload.sync(); const double secs = std::chrono::duration(std::chrono::steady_clock::now() - t0) .count(); diff --git a/examples/raw_bench_common.h b/examples/raw_bench_common.h index 17e9705d..c79a605b 100644 --- a/examples/raw_bench_common.h +++ b/examples/raw_bench_common.h @@ -20,6 +20,9 @@ #include #include +#include "bench_pipeline.h" +#include "bench_workload.h" + #include #include #include @@ -137,6 +140,27 @@ void wait_for_stop(int run_seconds, std::atomic &stop); void print_queue_stats(const char *direction, const std::string &interface_name, int queue_id, const RawBenchQueueStats &stats, double seconds); -void rx_count_worker(const RawBenchRxConfig &cfg, std::atomic &stop); +// Describes how received packets are reordered/gathered into the contiguous +// device buffer the GpuWorkload consumes. Defaults match the raw native shape; +// the bench main fills it from the TX config. +struct ReorderGeometry { + uint32_t packets_per_batch = 1024; // slots in the ordered buffer + uint32_t out_payload_len = 0; // bytes/slot (0 => use payload_size) + uint32_t payload_byte_offset = 0; // offset of payload within the packet + uint16_t seq_bit_offset = 0; // bit offset of the seq number + uint8_t seq_bit_width = 32; // seq number width (bits) + int payload_segment = 0; // packet segment carrying payload (1 = HDS) +}; + +// When `workload != None`, each RX thread builds its own GpuWorkload (cuFFT/ +// cuBLAS handles are not thread-safe to share) and its own ReorderPipeline, then +// for each received burst reorders the real packet payloads into a contiguous +// device buffer (geom) and runs one representative compute on THAT buffer. The +// stream is shared so the reorder kernel orders before the workload; the burst +// is freed only after the stream drains (so the reorder has read the buffers). +// workload == None leaves the bare-loopback count-only path untouched. +void rx_count_worker(const RawBenchRxConfig& cfg, std::atomic& stop, + BenchWorkload workload = BenchWorkload::None, + const ReorderGeometry& geom = {}); } // namespace daqiri::bench diff --git a/examples/raw_gpudirect_bench.cpp b/examples/raw_gpudirect_bench.cpp index 4240c05a..9aaf5747 100644 --- a/examples/raw_gpudirect_bench.cpp +++ b/examples/raw_gpudirect_bench.cpp @@ -19,6 +19,7 @@ #include #include +#include #include #include #include @@ -112,6 +113,14 @@ void tx_worker(const daqiri::bench::RawBenchTxConfig &cfg, eth_dst, ip_src, ip_dst, src_port, dst_port); std::memcpy(packet_template.data() + cfg.header_size, payload_template.data(), cfg.payload_size); + // Inject a per-packet sequence number (network byte order) at the start + // of the payload so the RX-side reorder workload has a real seq to sort + // by. The buffer is initialized once and reused, so seq == the packet's + // fixed index within the batch. + if (cfg.payload_size >= sizeof(uint32_t)) { + const uint32_t seq = htonl(static_cast(i)); + std::memcpy(packet_template.data() + cfg.header_size, &seq, sizeof(seq)); + } daqiri::bench::finalize_udp_ipv4_checksums(packet_template.data()); if (cudaMemcpy(gpu_pkt, packet_template.data(), packet_template.size(), cudaMemcpyHostToDevice) != cudaSuccess) { @@ -153,7 +162,8 @@ void tx_worker(const daqiri::bench::RawBenchTxConfig &cfg, int main(int argc, char **argv) { if (argc < 2) { std::cerr << "Usage: " << argv[0] - << " [--seconds N] [--target-gbps G]\n"; + << " [--seconds N] [--target-gbps G] " + "[--workload none|fft|gemm|gemm_fp16] [--workload-batch-bytes N]\n"; return 1; } @@ -161,6 +171,8 @@ int main(int argc, char **argv) { daqiri::bench::grafana::init_prometheus_metrics_from_env(); const int run_seconds = daqiri::bench::parse_run_seconds(argc, argv); const double target_gbps = daqiri::bench::parse_target_gbps(argc, argv); + const auto workload = daqiri::bench::parse_workload(argc, argv); + const size_t workload_batch_bytes = daqiri::bench::parse_workload_batch_bytes(argc, argv); const auto root = YAML::LoadFile(argv[1]); std::vector rx_configs; @@ -188,10 +200,30 @@ int main(int argc, char **argv) { std::vector rx_threads; daqiri::bench::TokenBucketPacer tx_pacer(target_gbps); + // Reorder geometry for the real-data workload: the seq number sits at the + // payload start (just after the UDP/IP header), and a window of up to 1024 + // packets is reordered into a contiguous ~8 MB device buffer (matching the + // RoCE 8 MB working set) that the compute then consumes. + daqiri::bench::ReorderGeometry geom; + if (!tx_configs.empty()) { + const auto& tx = tx_configs.front(); + geom.payload_segment = 0; + geom.payload_byte_offset = tx.header_size; + geom.seq_bit_offset = static_cast(tx.header_size * 8); + geom.seq_bit_width = 32; + geom.out_payload_len = tx.payload_size; + // Reorder window = the workload batch. --workload-batch-bytes sets it + // (rounded to whole packets); default ~8 MB (1024 packets at the native + // shape). Capped to one burst's packet count. + const uint32_t ppb = + workload_batch_bytes > 0 + ? std::max(1, static_cast(workload_batch_bytes / tx.payload_size)) + : 1024; + geom.packets_per_batch = std::min(ppb, tx.batch_size); + } rx_threads.reserve(rx_configs.size()); for (const auto &cfg : rx_configs) { - rx_threads.emplace_back(daqiri::bench::rx_count_worker, cfg, - std::ref(stop)); + rx_threads.emplace_back(daqiri::bench::rx_count_worker, cfg, std::ref(stop), workload, geom); } tx_threads.reserve(tx_configs.size()); for (const auto &cfg : tx_configs) { diff --git a/examples/raw_hds_bench.cpp b/examples/raw_hds_bench.cpp index 250b57b7..4a1abca5 100644 --- a/examples/raw_hds_bench.cpp +++ b/examples/raw_hds_bench.cpp @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -138,11 +139,15 @@ void tx_worker(const daqiri::bench::RawBenchTxConfig &cfg, int main(int argc, char **argv) { if (argc < 2) { - std::cerr << "Usage: " << argv[0] << " [--seconds N]\n"; + std::cerr << "Usage: " << argv[0] + << " [--seconds N] " + "[--workload none|fft|gemm|gemm_fp16] [--workload-batch-bytes N]\n"; return 1; } const int run_seconds = daqiri::bench::parse_run_seconds(argc, argv); + const auto workload = daqiri::bench::parse_workload(argc, argv); + const size_t workload_batch_bytes = daqiri::bench::parse_workload_batch_bytes(argc, argv); const auto root = YAML::LoadFile(argv[1]); if (daqiri::daqiri_init(argv[1]) != daqiri::Status::SUCCESS) { std::cerr << "daqiri_init failed\n"; @@ -162,8 +167,27 @@ int main(int argc, char **argv) { std::thread rx_thread; if (has_rx) { - rx_thread = std::thread(daqiri::bench::rx_count_worker, - daqiri::bench::parse_rx(root), std::ref(stop)); + // HDS reorder geometry: the payload lives in segment 1 (GPU memory), so the + // reorder reads it from offset 0 of that segment. The HDS TX does not inject + // a per-packet sequence number, so the seq-based reorder still exercises the + // kernel but mostly collides on slot 0 — fine for a throughput benchmark + // (the FLOP/copy volume is unchanged). + daqiri::bench::ReorderGeometry geom; + if (has_tx) { + const auto tx = daqiri::bench::parse_tx(root); + geom.payload_segment = 1; + geom.payload_byte_offset = 0; + geom.seq_bit_offset = 0; + geom.seq_bit_width = 32; + geom.out_payload_len = tx.payload_size; + const uint32_t ppb = + workload_batch_bytes > 0 + ? std::max(1, static_cast(workload_batch_bytes / tx.payload_size)) + : 1024; + geom.packets_per_batch = std::min(ppb, tx.batch_size); + } + rx_thread = std::thread(daqiri::bench::rx_count_worker, daqiri::bench::parse_rx(root), + std::ref(stop), workload, geom); } if (has_tx) { tx_thread = diff --git a/examples/rdma_bench.cpp b/examples/rdma_bench.cpp index 77a4fbc9..9cec4889 100644 --- a/examples/rdma_bench.cpp +++ b/examples/rdma_bench.cpp @@ -17,6 +17,7 @@ #include +#include #include #include #include @@ -25,6 +26,7 @@ #include #include #include +#include #include "raw_bench_common.h" #include @@ -78,7 +80,8 @@ RdmaBenchConfig parse_rdma_cfg(const YAML::Node& node) { } void rdma_worker(const RdmaBenchConfig& cfg, daqiri::bench::TokenBucketPacer& pacer, - std::atomic& stop, RdmaWorkerStats& stats) { + std::atomic& stop, RdmaWorkerStats& stats, + daqiri::bench::BenchWorkload workload, size_t workload_batch_bytes) { const char *thread_name = cfg.server ? "rdma_bench_server" : "rdma_bench_client"; if (!daqiri::bench::set_current_thread_affinity(cfg.cpu_core, thread_name)) { @@ -86,6 +89,55 @@ void rdma_worker(const RdmaBenchConfig& cfg, daqiri::bench::TokenBucketPacer& pa return; } + // Representative GPU workload run on each received message's REAL data (no-op + // unless --workload set). RoCE/RC delivers one large in-order message, so the + // recv buffer is already contiguous and (host_pinned on the integrated GB10) + // GPU-accessible: the gather-only pipeline is a zero-copy pass-through that + // hands the recv pointer straight to the compute. + // + // The compute working set is decoupled from the 8 MB message: --workload-batch- + // bytes sub-divides each message into chunk-sized slices, one compute per slice + // (the per-message sync then amortizes over num_chunks computes). 0 or >= the + // message size means one compute per message. + const uint32_t msg = static_cast(cfg.message_size); + const uint32_t chunk = (workload_batch_bytes > 0 && workload_batch_bytes < msg) + ? static_cast(workload_batch_bytes) + : msg; + const uint32_t num_chunks = chunk > 0 ? msg / chunk : 1; + daqiri::bench::GpuWorkload gpu_workload; + daqiri::bench::ReorderPipeline pipeline; + if (workload != daqiri::bench::BenchWorkload::None) { + if (!gpu_workload.init(workload, chunk) || + !pipeline.init(daqiri::bench::ReorderMode::GatherOnly, + /*packets_per_batch=*/1, msg, + /*payload_byte_offset=*/0, /*seq_bit_offset=*/0, + /*seq_bit_width=*/0, /*staging_needed=*/false, gpu_workload.stream())) { + std::cerr << "RDMA workload init failed; continuing without GPU workload\n"; + } + } + const bool run_workload = gpu_workload.enabled() && pipeline.enabled(); + + // To overlap receive with compute (apples-to-apples with the raw path, which + // holds a whole burst and drains the GPU once per burst), hold a small batch of + // RECEIVE completions instead of draining + freeing after each one. Holding a + // completion keeps its recv buffer alive, so the pass-through compute can read it + // while later messages keep arriving into other pool buffers; we drain the stream + // once per batch, then free them all. Bounded well under rx_depth so receives are + // not starved. + std::vector held_recv; + const size_t recv_hold_batch = + run_workload ? std::max(1, std::min(cfg.rx_depth / 2, 8)) : 0; + auto flush_held_recv = [&]() { + if (held_recv.empty()) { + return; + } + gpu_workload.sync(); + for (auto* held : held_recv) { + daqiri::free_tx_burst(held); + } + held_recv.clear(); + }; + int outstanding_send = 0; int outstanding_recv = 0; uint64_t send_wr_id = 0x1234; @@ -203,6 +255,24 @@ void rdma_worker(const RdmaBenchConfig& cfg, daqiri::bench::TokenBucketPacer& pa outstanding_recv--; stats.recv_completions++; stats.recv_bytes += static_cast(cfg.message_size); + if (run_workload) { + pipeline.reset_batch(); + pipeline.add_device_packet(daqiri::get_segment_packet_ptr(completion, 0, 0)); + const auto* base = static_cast(pipeline.finish_batch()); + // One compute per chunk-sized slice of the message (num_chunks == 1 when + // the batch is the whole message). Enqueue only; the GPU work overlaps + // with subsequent receives and is drained per batch (below). + for (uint32_t k = 0; k < num_chunks; ++k) { + gpu_workload.run(base + static_cast(k) * chunk); + } + // Hold the completion (and thus its recv buffer) until the batch drains, + // so the pass-through compute reads valid data without a per-message sync. + held_recv.push_back(completion); + if (held_recv.size() >= recv_hold_batch) { + flush_held_recv(); + } + continue; // do not free yet; freed by flush_held_recv() + } } daqiri::free_tx_burst(completion); } @@ -218,9 +288,14 @@ void rdma_worker(const RdmaBenchConfig& cfg, daqiri::bench::TokenBucketPacer& pa } if (!got_completion && !posted_work) { + // Idle: return any held recv buffers so they are not pinned while traffic + // pauses, then back off. + flush_held_recv(); std::this_thread::sleep_for(std::chrono::microseconds(100)); } } + flush_held_recv(); + gpu_workload.sync(); } } // namespace @@ -228,7 +303,9 @@ void rdma_worker(const RdmaBenchConfig& cfg, daqiri::bench::TokenBucketPacer& pa int main(int argc, char** argv) { if (argc < 2) { std::cerr << "Usage: " << argv[0] - << " [--seconds N] [--mode server|client|both] [--target-gbps G]\n"; + << " [--seconds N] [--mode server|client|both] " + "[--target-gbps G] [--workload none|fft|gemm|gemm_fp16] " + "[--workload-batch-bytes N]\n"; return 1; } @@ -244,6 +321,8 @@ int main(int argc, char** argv) { target_gbps = std::stod(argv[i + 1]); } } + const auto workload = daqiri::bench::parse_workload(argc, argv); + const size_t workload_batch_bytes = daqiri::bench::parse_workload_batch_bytes(argc, argv); const auto root = YAML::LoadFile(argv[1]); if (daqiri::daqiri_init(argv[1]) != daqiri::Status::SUCCESS) { @@ -279,12 +358,12 @@ int main(int argc, char** argv) { } if (run_server) { - server_thread = std::thread(rdma_worker, server_cfg, std::ref(server_pacer), - std::ref(stop), std::ref(server_stats)); + server_thread = std::thread(rdma_worker, server_cfg, std::ref(server_pacer), std::ref(stop), + std::ref(server_stats), workload, workload_batch_bytes); } if (run_client) { - client_thread = std::thread(rdma_worker, client_cfg, std::ref(client_pacer), - std::ref(stop), std::ref(client_stats)); + client_thread = std::thread(rdma_worker, client_cfg, std::ref(client_pacer), std::ref(stop), + std::ref(client_stats), workload, workload_batch_bytes); } if (!server_thread.joinable() && !client_thread.joinable()) { diff --git a/examples/run_spark_bench.sh b/examples/run_spark_bench.sh index 6b345060..4e97e7cd 100755 --- a/examples/run_spark_bench.sh +++ b/examples/run_spark_bench.sh @@ -23,6 +23,12 @@ # ETH_DST_ADDR — required for dpdk backend (the RX iface MAC). # REPEATS — repeats per cell for error bars (default 1; use 3 for the # published re-run). Each rep is an independent run + CSV row. +# WORKLOAD — representative GPU workload run on the REAL received data +# in the receive path (preceded by a reorder/gather step): +# none (default) | fft | gemm (FP32) | gemm_fp16 (FP16 +# tensor-core matmul). Honoured by all backends (dpdk, rdma, +# socket-udp, socket-tcp); recorded in the CSV post_process +# column. # # Optional (dpdk only): DPDK_{TX,RX}_PCI / DPDK_{TX,RX}_NETDEV override the p0/p1 # ports used for the per-cell *_phy wire-transit check (defaults p0 0000:01:00.0 / @@ -62,7 +68,9 @@ CSV="$OUT_DIR/runs.csv" # `pairs` = number of concurrent client/server process pairs (socket backends sweep # this; dpdk/rdma are always 1). `gbps` is aggregate App TX, `rx_gbps` aggregate App RX # (summed across pairs); App-level loss is (gbps - rx_gbps) / gbps. -echo "lang,backend,post_process,payload,batch,pairs,target_gbps,rep,seconds,packets,bytes,pps,gbps,rx_gbps,drops,drops_kind,cpu_master_pct,cpu_tx_pct,cpu_rx_pct,gpu_sm_pct,gpu_mem_pct" > "$CSV" +# post_process_batch (last column) = WORKLOAD_BATCH bytes, or "default" when unset. +# Appended at the end so existing column positions are unchanged. +echo "lang,backend,post_process,payload,batch,pairs,target_gbps,rep,seconds,packets,bytes,pps,gbps,rx_gbps,drops,drops_kind,cpu_master_pct,cpu_tx_pct,cpu_rx_pct,gpu_sm_pct,gpu_mem_pct,post_process_batch" > "$CSV" # Capture slow-moving environment state once per result set. "$SCRIPT_DIR/bench_capture_environment.sh" "$OUT_DIR" @@ -72,6 +80,24 @@ RUN_SECONDS=30 # capture dir (-r) and CSV row; the perf-doc tables report mean +/- std # across reps. Default 1; set REPEATS=3 for the published re-run. REPEATS="${REPEATS:-1}" +# Representative GPU workload run on the REAL received data (after a reorder/ +# gather step) in the receive path: none | fft | gemm (FP32 SGEMM) | gemm_fp16 +# (mixed-precision FP16/tensor-core matmul, the inference-style GEMM). Recorded in +# the CSV post_process column. Honoured by ALL backends (dpdk, rdma, socket-udp, +# socket-tcp). Default none = bare loopback (no GPU compute). +WORKLOAD="${WORKLOAD:-none}" +case "$WORKLOAD" in + none|fft|gemm|gemm_fp16) ;; + *) echo "Invalid WORKLOAD '$WORKLOAD' (expected none|fft|gemm|gemm_fp16)" >&2; exit 1 ;; +esac +# WORKLOAD_BATCH (bytes): the GPU compute working set per call, decoupled from the +# I/O unit (RoCE sub-chunks each message; raw sets the reorder window). Empty = +# backend default. Sweep it (e.g. 262144 524288 1048576 2097152 4194304 8388608) +# to trace the compute-intensity vs throughput curve. Recorded in post_process_batch. +WORKLOAD_BATCH="${WORKLOAD_BATCH:-}" +if [[ -n "$WORKLOAD_BATCH" && ! "$WORKLOAD_BATCH" =~ ^[0-9]+$ ]]; then + echo "Invalid WORKLOAD_BATCH '$WORKLOAD_BATCH' (expected a positive byte count)" >&2; exit 1 +fi DRIVER_LOG="$OUT_DIR/last_run.stderr" FAILURES=0 @@ -161,6 +187,11 @@ case "$BACKEND" in *) echo "Unknown backend: $BACKEND" >&2; exit 1 ;; esac +# All backends (dpdk, rdma, socket-udp, socket-tcp) now run the workload on real +# received data, so the CSV post_process column records the requested workload +# for every backend. +WORKLOAD_EFF="$WORKLOAD" + DROP_CURVE_TARGETS=(1 5 10 25 50 75 100 0) # 0 means unpaced (line rate) # -------------------------------------------------------------------------- @@ -365,6 +396,11 @@ run_cell() { local bench_extra=() [[ "$target_gbps" != "0" ]] && bench_extra+=(--target-gbps "$target_gbps") + # Every backend honours --workload (runs it on real received data); none = skip. + if [[ "$WORKLOAD_EFF" != "none" ]]; then + bench_extra+=(--workload "$WORKLOAD_EFF") + [[ -n "$WORKLOAD_BATCH" ]] && bench_extra+=(--workload-batch-bytes "$WORKLOAD_BATCH") + fi # Shutdown ordering: the server must keep receiving until the client has fully # stopped sending, otherwise the client's last in-flight messages have no peer # to land on -- for RDMA that flushes the QP (status 5, "Work Request Flushed @@ -536,7 +572,9 @@ run_cell() { gpu_mem="$(awk '/^ *[0-9]/ { count++; sum += $6 } END { if (count) printf "%.1f", sum/count; else print 0 }' \ "$cell_dir/nvidia_smi_dmon.txt" 2>/dev/null || echo 0)" - echo "$lang,$BACKEND,none,$payload,$batch,$pairs,$target_gbps,$rep,$secs,$pkts,$bytes,$pps,$gbps,$rx_gbps,$drops,$drops_kind,$cpu_master_pct,$cpu_tx_pct,$cpu_rx_pct,$gpu_sm,$gpu_mem" \ + local pp_batch="${WORKLOAD_BATCH:-default}" + [[ -z "$pp_batch" ]] && pp_batch="default" + echo "$lang,$BACKEND,$WORKLOAD_EFF,$payload,$batch,$pairs,$target_gbps,$rep,$secs,$pkts,$bytes,$pps,$gbps,$rx_gbps,$drops,$drops_kind,$cpu_master_pct,$cpu_tx_pct,$cpu_rx_pct,$gpu_sm,$gpu_mem,$pp_batch" \ | tee -a "$CSV" } diff --git a/examples/socket_bench.cpp b/examples/socket_bench.cpp index c3aaaa18..05d13392 100644 --- a/examples/socket_bench.cpp +++ b/examples/socket_bench.cpp @@ -17,6 +17,9 @@ #include +#include + +#include #include #include #include @@ -74,8 +77,33 @@ SocketBenchConfig parse_socket_cfg(const YAML::Node& node) { return cfg; } +// Detect the socket transport from the YAML: TCP is an in-order byte stream +// (gather-only), UDP datagrams can reorder (seq-based reorder). Scans the +// interfaces' socket_config.local_addr scheme. +bool socket_transport_is_tcp(const YAML::Node& root) { + const auto ifaces = root["interfaces"]; + if (ifaces && ifaces.IsSequence()) { + for (const auto& iface : ifaces) { + const auto sc = iface["socket_config"]; + if (!sc) { + continue; + } + const auto addr = sc["local_addr"].as(""); + if (addr.rfind("tcp://", 0) == 0) { + return true; + } + if (addr.rfind("udp://", 0) == 0) { + return false; + } + } + } + return false; // default to UDP semantics +} + void socket_worker(const SocketBenchConfig& cfg, daqiri::bench::TokenBucketPacer& pacer, - std::atomic& stop, SocketWorkerStats& stats) { + std::atomic& stop, SocketWorkerStats& stats, + daqiri::bench::BenchWorkload workload, bool is_tcp, + size_t workload_batch_bytes) { const char *thread_name = cfg.server ? "socket_bench_server" : "socket_bench_client"; if (!daqiri::bench::set_current_thread_affinity(cfg.cpu_core, thread_name)) { @@ -83,6 +111,32 @@ void socket_worker(const SocketBenchConfig& cfg, daqiri::bench::TokenBucketPacer return; } + // Representative GPU workload on the REAL received data. Socket data lands in + // pageable host memory, so the pipeline stages it to the GPU (a host->device + // copy on the workload stream -- the honest cost of the socket path). UDP can + // reorder by sequence number; TCP is in-order, so it gathers per chunk. + const uint32_t msg = static_cast(cfg.message_size); + // --workload-batch-bytes sets the UDP ordered-buffer / working-set size + // (default ~8 MB, matching the RoCE working set); the UDP reorder window is that + // many datagrams. TCP gathers one chunk at a time (one compute per recv). + const uint32_t target_bytes = + workload_batch_bytes > 0 ? static_cast(workload_batch_bytes) : 8u * 1024u * 1024u; + const uint32_t packets_per_batch = + is_tcp ? 1u : std::max(1u, std::min(8192u, target_bytes / std::max(1u, msg))); + daqiri::bench::GpuWorkload gpu_workload; + daqiri::bench::ReorderPipeline pipeline; + if (workload != daqiri::bench::BenchWorkload::None) { + if (!gpu_workload.init(workload, static_cast(packets_per_batch) * msg) || + !pipeline.init(is_tcp ? daqiri::bench::ReorderMode::GatherOnly + : daqiri::bench::ReorderMode::SeqReorder, + packets_per_batch, msg, /*payload_byte_offset=*/0, + /*seq_bit_offset=*/0, /*seq_bit_width=*/is_tcp ? 0 : 32, + /*staging_needed=*/true, gpu_workload.stream())) { + std::cerr << "Socket workload init failed; continuing without GPU workload\n"; + } + } + const bool run_workload = gpu_workload.enabled() && pipeline.enabled(); + uintptr_t conn_id = 0; uint16_t port = 0; uint16_t queue = 0; @@ -126,6 +180,12 @@ void socket_worker(const SocketBenchConfig& cfg, daqiri::bench::TokenBucketPacer if (daqiri::get_tx_packet_burst(msg) == daqiri::Status::SUCCESS) { auto* payload = reinterpret_cast(daqiri::get_packet_ptr(msg, 0)); std::memset(payload, static_cast(stats.sent_packets & 0xff), cfg.message_size); + // Inject a sequence number (network byte order) at the payload start so + // the UDP RX-side reorder has a real seq to sort by. Harmless for TCP. + if (cfg.message_size >= static_cast(sizeof(uint32_t))) { + const uint32_t seq = htonl(static_cast(stats.sent_packets)); + std::memcpy(payload, &seq, sizeof(seq)); + } daqiri::set_packet_lengths(msg, 0, {cfg.message_size}); daqiri::set_connection_id(msg, conn_id); @@ -144,9 +204,28 @@ void socket_worker(const SocketBenchConfig& cfg, daqiri::bench::TokenBucketPacer daqiri::BurstParams* burst = nullptr; if (daqiri::get_rx_burst(&burst, conn_id, cfg.server) == daqiri::Status::SUCCESS && burst != nullptr) { - const uint64_t rx_pkts = static_cast(daqiri::get_num_packets(burst)); - stats.received_packets += rx_pkts; + const int num_pkts = static_cast(daqiri::get_num_packets(burst)); + stats.received_packets += static_cast(num_pkts); stats.received_bytes += daqiri::get_burst_tot_byte(burst); + + if (run_workload) { + // Stage each received (host) payload to the GPU; the copy persists in + // the pipeline's device buffer, so a batch can accumulate across bursts + // (UDP bursts are one datagram). When a batch fills, reorder/gather it + // and run the compute. TCP (packets_per_batch == 1) flushes per chunk. + for (int i = 0; i < num_pkts; ++i) { + const auto len = daqiri::get_packet_length(burst, i); + void* d = pipeline.stage_host_packet(daqiri::get_packet_ptr(burst, i), len); + pipeline.add_device_packet(d); + if (pipeline.collected() == packets_per_batch) { + gpu_workload.run(pipeline.finish_batch()); + pipeline.reset_batch(); + } + } + // Drain so the host->device staging copies (which read this burst's + // memory) complete before the burst is freed/recycled. + gpu_workload.sync(); + } daqiri::free_all_packets_and_burst_rx(burst); } else { std::this_thread::sleep_for(std::chrono::microseconds(100)); @@ -160,7 +239,9 @@ void socket_worker(const SocketBenchConfig& cfg, daqiri::bench::TokenBucketPacer int main(int argc, char** argv) { if (argc < 2) { std::cerr << "Usage: " << argv[0] - << " [--seconds N] [--mode server|client|both] [--target-gbps G]\n"; + << " [--seconds N] [--mode server|client|both] " + "[--target-gbps G] [--workload none|fft|gemm|gemm_fp16] " + "[--workload-batch-bytes N]\n"; return 1; } @@ -176,8 +257,11 @@ int main(int argc, char** argv) { target_gbps = std::stod(argv[i + 1]); } } + const auto workload = daqiri::bench::parse_workload(argc, argv); + const size_t workload_batch_bytes = daqiri::bench::parse_workload_batch_bytes(argc, argv); const auto root = YAML::LoadFile(argv[1]); + const bool is_tcp = socket_transport_is_tcp(root); if (daqiri::daqiri_init(argv[1]) != daqiri::Status::SUCCESS) { std::cerr << "daqiri_init failed\n"; return 1; @@ -211,12 +295,12 @@ int main(int argc, char** argv) { } if (run_server) { - server_thread = std::thread(socket_worker, server_cfg, std::ref(server_pacer), - std::ref(stop), std::ref(server_stats)); + server_thread = std::thread(socket_worker, server_cfg, std::ref(server_pacer), std::ref(stop), + std::ref(server_stats), workload, is_tcp, workload_batch_bytes); } if (run_client) { - client_thread = std::thread(socket_worker, client_cfg, std::ref(client_pacer), - std::ref(stop), std::ref(client_stats)); + client_thread = std::thread(socket_worker, client_cfg, std::ref(client_pacer), std::ref(stop), + std::ref(client_stats), workload, is_tcp, workload_batch_bytes); } if (!server_thread.joinable() && !client_thread.joinable()) { diff --git a/src/kernels.cu b/src/kernels.cu index d55046c5..4dc0c3a2 100644 --- a/src/kernels.cu +++ b/src/kernels.cu @@ -580,3 +580,33 @@ extern "C" void packet_reorder_copy_payload_by_sequence(void* out, input_endianness, batch_id_out); } + +__global__ void packet_gather_copy_payload_kernel(void* __restrict__ out, + const void* const* const __restrict__ in, + uint32_t payload_len, + uint32_t payload_byte_offset, uint32_t num_pkts) { + const uint32_t pkt_idx = static_cast(blockIdx.x); + if (pkt_idx >= num_pkts) { + return; + } + + const auto* src_pkt = static_cast(in[pkt_idx]); + if (src_pkt == nullptr) { + return; + } + + const auto* src = src_pkt + payload_byte_offset; + auto* dst = static_cast(out) + (static_cast(pkt_idx) * payload_len); + copy_payload_vectorized(dst, src, payload_len); +} + +extern "C" void packet_gather_copy_payload(void* out, const void* const* const in, + uint32_t payload_len, uint32_t payload_byte_offset, + uint32_t num_pkts, cudaStream_t stream) { + if (out == nullptr || in == nullptr || payload_len == 0 || num_pkts == 0) { + return; + } + + packet_gather_copy_payload_kernel<<>>(out, in, payload_len, + payload_byte_offset, num_pkts); +} diff --git a/src/kernels.h b/src/kernels.h index 7989bff1..68cc1c7c 100644 --- a/src/kernels.h +++ b/src/kernels.h @@ -43,6 +43,14 @@ __attribute__((__visibility__("default"))) void packet_reorder_copy_payload_by_s uint8_t input_endianness, uint64_t* batch_id_out, cudaStream_t stream); + +// Gather-only companion to packet_reorder_copy_payload_by_sequence: places each +// input packet's payload into the output buffer at its ARRIVAL index (slot = +// packet index), with no sequence-number read and no reshuffle. Used for the +// in-order transports (RoCE/TCP) where the seq-based reorder is meaningless. +__attribute__((__visibility__("default"))) void packet_gather_copy_payload( + void* out, const void* const* const in, uint32_t payload_len, uint32_t payload_byte_offset, + uint32_t num_pkts, cudaStream_t stream); #if __cplusplus } #endif