diff --git a/AGENTS.md b/AGENTS.md index d9d639a..7448afc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -44,7 +44,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}`. `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). +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-gemm-dim N` pins the square GEMM side length (default 1024), so the FLOP count per call (2·n³) is FIXED and the compute working set is exactly n·n·elem_size, read from the front of each received I/O unit (the unit must be at least that large). `--workload-fft-len N` pins the 1-D C2C transform length for `fft` (default 1024; independent of the GEMM dimension); the working set is fanned out across as many batched length-N transforms as fit. Used by `run_spark_bench.sh`'s `WORKLOAD` / `GEMM_DIM` / `FFT_LEN` env (all backends) to fill the CSV `post_process` / `post_process_gemm_dim` 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 b0f5014..fdc0777 100644 --- a/docs/benchmarks/performance-dgx-spark.md +++ b/docs/benchmarks/performance-dgx-spark.md @@ -9,6 +9,10 @@ Measured C++-loopback throughput for each stream/protocol on a single DGX Spark (GB10), driven over a physical cabled loopback on one ConnectX-7. Numbers are from a Release build via `examples/run_spark_bench.sh` (30 s per cell). +All backends are measured **one-way** (unidirectional) by default: one side sends, +the other receives and computes. For a bidirectional test, set `send: true` on the +receiving role (and `receive: true` on the sending role) in the bench config. + For the loopback setup these numbers depend on and the per-transport benchmarking procedure, see [Socket and RDMA Benchmarking](socket_benchmarking.md) (the `dq_wire_*` network-namespace wire loopback used by RoCE and sockets) and @@ -20,39 +24,44 @@ loopback). The exact commands are collected under [Reproduce](#reproduce) below. | Component | Detail | | --------- | ------ | | Platform | DGX Spark (GB10), 20 cores, isolcpus `16-19` (the multi-queue sweep expands this; see [Multi-queue core scaling](#multi-queue-core-scaling)) | -| NIC | ConnectX-7, ports p0 ↔ p1 cross-cabled (single-host loopback), MTU 9000 | +| NIC | ConnectX-7, ports p0 ↔ p1 cross-cabled with a **100 GbE QSFP28** loopback cable (single-host loopback), MTU 9000 | | Build | Release (`-DCMAKE_BUILD_TYPE=Release`), `DAQIRI_ENGINE="dpdk ibverbs"` | | 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`). | +| 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. Sockets pin each pair's send and receive to separate cores in the same CPU cluster (all with `isolcpus=5-9,15-19`). | ## Results Summary (C++ loopback) -Each transport at its best-case **native operation size**. Raw/RoCE are +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. +The **100 GbE QSFP28 loopback cable sets the maximum data rate** here — not the +ConnectX-7 (which is rated for higher line rates) or the software path. A 100 GbE +link tops out near ~99.6 Gb/s of payload, so every large-transfer result saturates +just under that ceiling. + | Stream / Protocol | Best case | Throughput | Drops | Notes | | ----------------- | --------- | ---------: | ----- | ----- | -| Raw Ethernet / GPUDirect | 4 KB packet | **105.5 ±0.9 Gb/s** | 0 | 98.5 Gb/s single-queue at the 8 KB native shape | -| Socket / RoCE (SEND) | 8 MB message | **102.2 ±0.3 Gb/s** | 0 | Single QP, batch 1 | -| Socket / TCP | 8 KB × 4 pairs | **97.2 ±2.8 Gb/s** | ~0 | Flow-controlled (App TX = App RX) | -| Socket / UDP | 8 KB × 4 pairs | **29.8 ±0.2 Gb/s** | ~51% loss | Receiver goodput; unpaced sender | +| Raw Ethernet / GPUDirect | 8 KB packet | **98.8 ±0.1 Gb/s** | 0 | Flat ~98.7 across 4–8 KB, all batch sizes | +| Socket / RoCE (SEND) | 64 KB message | **97.6 ±0.1 Gb/s** | 0 | Single QP, batch 1 | +| Socket / TCP | 8 KB × 4 pairs | **87.3 ±2.2 Gb/s** | ~0 | Flow-controlled (App TX = App RX) | +| Socket / UDP | 8 KB × 4 pairs | **34.5 ±0.6 Gb/s** | ~48% loss | Receiver goodput; unpaced sender | -Each transport is best read at its own native operation size (see the per-transport +Each transport is best read at its own best-case operation size (see the per-transport tables below); a single cross-transport unit of work isn't meaningful here, since RoCE at 8 KB is op-rate-bound well below its large-message peak and TCP has no operation boundary. ## Raw Ethernet / GPUDirect (DPDK) -Physical port-to-port loopback, GPU-resident payloads. Native 8 KB packets run -at **98.5 Gb/s** drop-free across all batch sizes; the throughput peak is -**105.5 Gb/s** at a 4 KB payload. Packet handling is CPU-bound (see the CPU -utilization table below). Throughput is flat across batch size and stable -run-to-run (3 reps per cell, ≤1% spread). +Physical port-to-port loopback, GPU-resident payloads. Throughput saturates the +100 GbE line at **~98.8 Gb/s** for 4–8 KB payloads, drop-free across all batch +sizes. Packet handling is CPU-bound (see the CPU utilization table below). +Throughput is flat across batch size and stable run-to-run (3 reps per cell, +≤1% spread). Achieved Gb/s measured at App RX (equal to App TX, since every cell is -drop-free), unpaced, mean of 3 reps; run-to-run spread ≤0.9 Gb/s (<1%): +drop-free), unpaced, mean of 3 reps; run-to-run spread ≤0.5 Gb/s (<1%): @@ -65,29 +74,29 @@ drop-free), unpaced, mean of 3 reps; run-to-run spread ≤0.9 Gb/s (<1%): - - - - - + + + + +
8000 B98.598.098.098.5
4096 B105.2105.3105.5105.1
1024 B92.191.891.791.7
256 B50.049.849.849.8
64 B20.420.520.520.4
8000 B98.898.898.898.7
4096 B98.698.898.798.6
1024 B97.197.297.297.1
256 B49.749.649.649.5
64 B20.220.220.420.2
-At ≥4 KB the link saturates (~98–105 Gb/s) regardless of batch. Below that the -path is packet-rate-bound: 1 KB ~92 Gb/s (10.5 M pps), 256 B ~50 Gb/s (19.5 M pps), -64 B ~20 Gb/s (20 M pps) — a ~20 M pps single-queue ceiling (the multi-queue +At ≥1 KB the link saturates near line rate (~97–99 Gb/s) regardless of batch. +Below that the path is packet-rate-bound: 256 B ~50 Gb/s (19.5 M pps), 64 B +~20 Gb/s (20 M pps) — a ~20 M pps single-queue ceiling (the multi-queue section lifts it). Gb/s here is the L2 frame rate including the 64 B header, so pps ≈ Gb/s ÷ ((payload + 64) × 8). These small-payload cells are flat across batch size and stable run-to-run. Because every cell is drop-free, the achieved rate is also the no-drop rate: pacing the sender below it hits the target with zero drops. -**CPU utilization** (headline cell, 8000 B / batch 10240, unpaced): +**CPU utilization** (summary-table cell, 8000 B / batch 10240, unpaced): | Core | Busy% | Note | | ------------------------ | ----: | ------------------------------------- | | Master (CPU 8) | 3.7% | Orchestration only; mostly idle | -| TX queue poller (CPU 17) | ~92% | Poll-mode spin; rate-independent | -| RX queue poller (CPU 18) | ~92% | Poll-mode spin; rate-independent | +| TX queue poller (CPU 17) | ~92% | Poll-mode busy-spin | +| RX queue poller (CPU 18) | ~92% | Poll-mode busy-spin | The benchmark app workers run on their own cores (TX 16, RX 19) alongside these pollers; this run sampled only the poller cores. @@ -98,10 +107,10 @@ not a compute engine. ### Multi-queue core scaling -Each packet-handling core spins in poll-mode. At the native 8 KB shape a single -TX core caps throughput near ~98 Gb/s while a single RX core already drains the -line, so the second TX core is the lever: it scales `(1,1)` to `(2,1)` from -97.5 to 108.8 Gb/s, while a second RX core adds little. The matrix sweeps +Each packet-handling core spins in poll-mode. At large payloads (≥1 KB) a single +queue already saturates the 100 GbE line (~97–99 Gb/s), so extra cores add +nothing there — the multi-queue win is confined to the small, +packet-rate-bound payloads, where **RX cores** are the lever. The matrix sweeps (TX cores, RX cores) over `(1,1)`, `(1,2)`, `(2,1)`, `(2,2)`. Each queue is served by a poll-mode driver core plus a separate bench-worker @@ -113,26 +122,29 @@ base `daqiri_bench_raw_tx_rx_spark_mq.yaml` (the balanced 2,2 superset) by `scripts/gen_spark_mq_config.py`; generated by `examples/run_spark_mq_bench.sh`, 30 s per cell, 0 drops. +Achieved Gb/s at a **256 B payload** (the packet-rate-bound regime where core +count matters); at ≥1 KB every cell converges at the wire ceiling regardless: + | Cell | TX pollers | RX pollers | Achieved Gb/s | | ---- | ---------- | ---------- | ------------: | -| (1,1) | 16 | 18 | 97.5 | -| (1,2) | 16 | 18,9 | 99.0 | -| (2,1) | 16,19 | 18 | **108.8** | -| (2,2) | 16,19 | 18,9 | **108.8** | +| (1,1) | 16 | 18 | 50.0 | +| (1,2) | 16 | 18,9 | **66.4** | +| (2,1) | 16,19 | 18 | 49.0 | +| (2,2) | 16,19 | 18,9 | 64.7 | -Which core is the bottleneck flips with payload size. Sweeping each cell from -64 B to 8 KB: +A second **RX** core lifts 256 B from 50.0 to 66.4 Gb/s; a second **TX** core does +nothing (49.0 ≈ 50.0). The full payload sweep shows why — the bottleneck depends +on payload size: ![DPDK multi-queue throughput vs UDP payload size on DGX Spark, one line per (TX,RX) core count](../images/spark-mq-payload-sweep.svg) At small payloads the path is packet-rate-bound, so **RX cores** are the lever: -at 64 B a second RX core lifts throughput from 20.3 to 26.8 Gb/s (~20 M → ~26 M -pps) while a second TX core does nothing. At large payloads it inverts to the -byte/line-rate-bound regime where **TX cores** are the lever: at 8 KB the second -TX core takes 97.5 to 108.8 Gb/s (the native-shape result above) while a second -RX core adds nothing. The curves cross around 1–4 KB, where the link saturates -and all four cells converge near ~104–109 Gb/s. Every cell is drop-free. -Generated by `examples/run_spark_mq_bench.sh` (30 s per point) and +a second RX core lifts 64 B from 20.3 to 26.9 Gb/s (~20 M → ~27 M pps) and 256 B +from 50.0 to 66.4 Gb/s, while a second TX core does nothing. At large payloads a +single queue already saturates the wire, so all four cells converge near +~97–99 Gb/s at ≥1 KB and neither extra core helps. Every cell is drop-free. +Generated by +`examples/run_spark_mq_bench.sh` (30 s per point) and `scripts/plot_mq_payload_sweep.py`. ## Socket / RoCE @@ -142,44 +154,52 @@ is App RX goodput, equal to App TX with 0 drops. Large messages up to 64 KB saturate the wire; the smallest messages are bound by per-operation software overhead. -**Message-size sweep (single QP, batch 1, 0 drops).** Mean of 3 reps; run-to-run -spread ≤0.8 Gb/s (<2%) in every cell. +**Message-size sweep (single QP, batch 1, 0 drops).** Mean ± sample std over 3 +reps; run-to-run spread <1% in every cell. | Message size | Gb/s | | ------------ | ---: | -| 8 MB | **102.2** | -| 1 MB | 101.3 | -| 64 KB | 101.6 | -| 8 KB | 60.7 | -| 4 KB | 38.0 | - -Messages ≥64 KB hold ~101–102 Gb/s at the wire ceiling. Below that the path is -operation-rate-bound (per-operation software overhead, not a stall) rather than -wire-bound, and every cell is drop-free. At 8 KB (60.7 Gb/s) and 4 KB (38.0 Gb/s) +| 8 MB | 96.8 ±0.3 | +| 1 MB | 96.9 ±0.2 | +| 64 KB | **97.6 ±0.1** | +| 8 KB | 51.8 ±0.2 | +| 4 KB | 28.5 ±0.0 | + +Messages ≥64 KB hold ~97 Gb/s at the 100 GbE wire ceiling (line rate). Below that +the path is operation-rate-bound (per-operation software overhead, not a stall) +rather than wire-bound, and every cell is drop-free. At 8 KB (51.8 Gb/s) and 4 KB (28.5 Gb/s) a dedicated bench-worker core, separate from the RoCE engine thread, sustains the operation rate, as it does for small DPDK packets. A per-message flow-control window keeps enough operations in flight to amortize that overhead: it pre-posts `rx_depth` receives before sending and caps the transmit side at `tx_depth`, each sized to the message so the in-flight window stays full. -**CPU utilization** (headline cell, 8 MB message, batch 1, unpaced): +**CPU utilization** (summary-table cell, 8 MB message, batch 1, unpaced): -| Core | Busy% | Note | -| ------------------- | ----: | ----------------------------------------------- | -| Master (CPU 8) | 5.9% | Orchestration only | -| Client TX (CPU 17) | 74.9% | Post-and-poll spin; rate-independent | -| Server RX (CPU 18) | 0.0% | HCA writes straight to memory; CPU uninvolved | +| Core | Busy% | Note | +| -------------------- | ----: | ----------------------------------------------- | +| Master (CPU 8) | 0.7% | Orchestration only | +| Client TX (CPU 17) | 74.8% | Busy-spins posting sends and polling completions | +| Server RX (CPU 19) | 1.1% | HCA DMAs straight to memory; worker only reaps completions | -The idle RX core is the expected RoCE RC signature — the HCA places incoming -data directly into registered memory with no CPU involvement. The GPU stays -idle here too (SM and memory-controller ~0%; DMA target, not a compute engine). +The TX core busy-spins in a post-and-poll loop, so its ~75% busy time is set by +that spin, not by the throughput: it stays near this level whether the link runs +at 10 or 100 Gb/s (the same reason the DPDK pollers sit near 92% regardless of +offered load). The near-idle RX core is the expected RoCE RC signature — the HCA +places incoming data directly into registered memory, so the receive worker only +reaps completions and reposts (~1% at this message rate). The GPU stays idle here +too (SM and memory-controller ~0%; DMA target, not a compute engine). ## Socket / TCP -Four one-way TCP client/server pairs over the netns wire loopback, each pair -pinned to one isolated core (16–19). TCP self-paces via flow control, so App TX -equals App RX with effectively no app-level loss. `message_size` is the per-send -byte count of a stream (no datagram boundary, no fragmentation). +Four one-way TCP client/server pairs over the netns wire loopback. Each pair's +send (client) and receive (server) sides pin to **separate** isolated cores in one +CPU cluster (pairs 1–2 in 15–19, pairs 3–4 in 5–9). A shared send/receive core +ping-pongs a single stream and can wedge it at half rate for a whole run, so +splitting the two sides keeps single-stream throughput stable. TCP self-paces via +flow control, so App TX equals App RX with effectively no app-level loss. +`message_size` is the per-send byte count of a stream (no datagram boundary, no +fragmentation). Throughput in Gb/s (App TX = App RX), mean ± std over 3 reps: @@ -194,21 +214,23 @@ Throughput in Gb/s (App TX = App RX), mean ± std over 3 reps: - 1000 B13.5±0.227.3±0.254.8±0.2 - 8000 B30.8±5.868.0±9.797.2±2.8 - 1 MiB31.6±0.158.7±0.393.7±1.3 + 1000 B14.2±0.427.6±0.445.2±0.1 + 8000 B28.9±2.942.4±2.787.3±2.2 + 1 MiB32.1±2.251.5±2.483.7±0.4 Throughput scales with the pair count; retransmits stay negligible over the run. +At four pairs the eight cores span both clusters, and the pairs in 5–9 sit farther +from the NIC, so the 4-pair cells scale slightly sub-linearly. ## Socket / UDP -Four one-way UDP client/server pairs, same one-core-per-pair pinning. UDP has no -flow control, so each sender runs flat-out and the receiver drops whatever it -cannot drain — the loss column is an inherent property of unpaced UDP, not a -fault. App RX is the delivered goodput; App-level loss is `(App TX − App RX) / -App TX`. +Four one-way UDP client/server pairs, same per-side pinning (send and receive on +separate cores). UDP has no flow control, so each sender runs flat-out and the +receiver drops whatever it cannot drain — the loss column is an inherent property +of unpaced UDP. App RX is the delivered goodput; App-level loss is +`(App TX − App RX) / App TX`. Each cell shows **receiver goodput in Gb/s** (mean ± std over 3 reps) with the **app-level loss %** dimmed beneath it: @@ -224,8 +246,8 @@ Each cell shows **receiver goodput in Gb/s** (mean ± std over 3 reps) with the - 1000 B3.6 ±0.011% loss7.7 ±0.016% loss13.5 ±0.16% loss - 8000 B9.6 ±0.556% loss20.9 ±1.357% loss29.8 ±0.251% loss + 1000 B4.0 ±0.144% loss6.8 ±0.852% loss15.2 ±0.340% loss + 8000 B12.2 ±1.269% loss18.9 ±0.259% loss34.5 ±0.648% loss @@ -242,13 +264,13 @@ 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 +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}`): +**What the two workloads compute** — both in **FP32** (single precision) — from 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 @@ -257,12 +279,11 @@ compute consumes that buffer. 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 + matrices, with the reordered buffer supplying the *A* operand. The side length is + **pinned at n=1024** (`--workload-gemm-dim`, env `GEMM_DIM`), so every call is an + identical **2.15 GFLOP** matmul reading the first **4 MB** (n²·4 B, FP32) of each + received unit — the compute is fixed regardless of message size, which is what + makes it comparable across transports. The matmul is FP32 (`cublasSgemm`). This models a receiver feeding incoming data into a dense linear-algebra or neural-network stage (beamforming, correlation, an inference layer). @@ -276,12 +297,10 @@ chosen to be representative for each transport: | 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. +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 and +compute overlaps ingest. The reorder window is sized so the contiguous buffer is +~8 MB on every backend, giving a comparable GPU working set across transports. !!! note "Where the data lives, per backend" On the integrated GB10 the GPU shares memory with the CPU, so the raw and RoCE @@ -292,74 +311,24 @@ across the run. Throughput is mean ± std over 3 reps, 30 s each. 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.) +Fixed **n=1024**, one GEMM (or a length-1024 batched FFT) per received unit. DPDK runs +at an **8 KB payload** (~8 MB reorder window, 1024 packets × 8000 B), matched to RoCE's +**8 MB message** so the GPU working set and per-unit compute are the same on both. +3 reps, 30 s each, GPU SM% from `nvidia-smi dmon`; 0 drops on every cell. + +| Workload | DPDK (Raw / GPUDirect) | RoCE (RC) | +| -------- | ---------------------: | --------: | +| none (baseline) | 98.7 ±0.0 | 96.6 ±0.3 | +| FFT | 95.7 ±0.8 | 95.6 ±0.1 | +| GEMM (FP32) | 96.6 ±0.2 | 90.2 ±1.1 | + +Throughput in Gb/s. Both `none` baselines sit at the ~97–99 Gb/s wire ceiling +(DPDK 98.7, RoCE 96.6), as expected for two line-rate transports. + +**GPU compute dents line rate only modestly, and stays wire-limited, not +compute-limited** (SM well under 100% throughout). FFT is nearly free on both +(SM ~6–17%, ≤1 Gb/s off baseline). The reorder/gather step assembles +each unit's payload into one contiguous GPU buffer. ## Reproduce @@ -383,6 +352,12 @@ apt-get install -y iproute2 iputils-ping ethtool iperf3 rdma-core ibverbs-utils These provide `ip`/`nstat` (`iproute2`), `ethtool`, and `ib_send_bw` (`perftest`). +Each `run_spark_bench.sh ` invocation takes a **mode** that sets +which cells run: `sweep` runs the full payload × batch × pairs matrix (the +per-transport message-size tables above), while `smoke` runs just the single +summary-table cell — one payload/batch/pairs operating point. `REPEATS=N` repeats +every cell N times for error bars. + **Raw Ethernet / GPUDirect (DPDK)** drives the two physical ports directly, so the `dq_wire_*` namespaces must **not** be up — they capture the ports and hide them from DPDK. Tear them down first (no-op if they were never created). @@ -420,34 +395,30 @@ 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): +**GPU workload (FFT / GEMM)** re-runs a backend with a representative GPU workload +in the receive path by exporting `WORKLOAD` (`none` | `fft` | `gemm` | +`gemm_fp16`), run once per received I/O unit on the real payload. Each call is a +fixed **1024³ GEMM** (override with `GEMM_DIM` / `--workload-gemm-dim`) or a batched +**length-1024 FFT** (override with `FFT_LEN` / `--workload-fft-len`) — both compute +sizes are held constant while the message size varies, so the FLOP count per call +is fixed. It composes with the same netns setup as above (dpdk in the default +namespace, rdma in the `dq_wire_*` namespaces). Use `smoke` — the single +summary-table cell that the fixed-n table reports — and run all three workloads +with error bars: ```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 +# RoCE (netns up); Raw is identical with `dpdk`, netns down, ETH_DST_ADDR exported. +for WL in none fft gemm; do + WORKLOAD=$WL REPEATS=3 ./examples/run_spark_bench.sh rdma smoke done -# raw: netns down, ETH_DST_ADDR exported; same loop with `dpdk` ``` +In the workload case the payload size is fixed per backend (8 KB for DPDK, 8 MB +message for RoCE), so a `sweep` only steps through batch size (DPDK) or +client/server pairs (sockets). The workload lands in the CSV `post_process` column +(with the GEMM dimension in `post_process_gemm_dim`); compare each `gbps` / +`gpu_sm_pct` against the `WORKLOAD=none` baseline from the same loop. + 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 fa10a2a..59c4c57 100644 --- a/docs/benchmarks/raw_benchmarking.md +++ b/docs/benchmarks/raw_benchmarking.md @@ -238,7 +238,7 @@ 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). +`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. `--workload-gemm-dim N` (default 1024) pins the square GEMM side length and `--workload-fft-len N` (default 1024) the 1-D FFT transform length, so the FLOP count per call stays constant as the I/O unit is swept. The same flags are 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 diff --git a/docs/images/spark-mq-payload-sweep.svg b/docs/images/spark-mq-payload-sweep.svg index 8d6b309..998761d 100644 --- a/docs/images/spark-mq-payload-sweep.svg +++ b/docs/images/spark-mq-payload-sweep.svg @@ -6,11 +6,11 @@ - 2026-06-11T23:20:48.401040 + 2026-07-10T18:39:19.745269 image/svg+xml - Matplotlib v3.10.9, https://matplotlib.org/ + Matplotlib v3.11.0, https://matplotlib.org/ @@ -30,35 +30,35 @@ z - - + - - + - + - - - - + + - + - + - + - - - - - + + + - + - + - + - - - - - - + + + + - + - + - + - - - - - + + + + - + - + - + - - - - - + + + + - + - - - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + - + - - + - - + + - + - + - - - + + + - + - + - - - + + + - + - + - - - + + + - + - + - - - + + + - + - + - - - - + + + + - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + - + - - - - - - - + + + + + + - + - - - - - - - + + + + + + - + - - - - - - - + + + + + + - + - - - - - - - + + + + + + - - - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + - + - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + - - + - + - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + - - + - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + - - + - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + - - + - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + - - + + diff --git a/examples/bench_workload.cu b/examples/bench_workload.cu index 4d602da..2c71611 100644 --- a/examples/bench_workload.cu +++ b/examples/bench_workload.cu @@ -37,6 +37,13 @@ namespace { constexpr int kFftLen = 1024; // Default working-set size when the caller passes bytes_per_burst == 0. constexpr size_t kDefaultBytes = 1u << 16; // 64 KiB +// Fixed CUDA-event pool for record_event(); comfortably above the RoCE recv-path +// in-flight cap (default min(rx_depth/2, 32), overridable up to kMaxWorkloadInflight +// via --workload-max-inflight). Created lazily on first record_event(). +constexpr int kEventPoolSize = 64; +static_assert(kEventPoolSize > daqiri::bench::kMaxWorkloadInflight, + "event pool must stay above the max in-flight cap so record_event() " + "never starves exactly at the cap"); cufftHandle as_fft_plan(int p) { return static_cast(p); @@ -47,6 +54,9 @@ cudaStream_t as_stream(void* s) { cublasHandle_t as_cublas(void* h) { return static_cast(h); } +cudaEvent_t as_event(void* e) { + return static_cast(e); +} } // namespace @@ -73,16 +83,52 @@ BenchWorkload parse_workload(int argc, char** argv) { return workload; } -size_t parse_workload_batch_bytes(int argc, char** argv) { +int parse_workload_gemm_dim(int argc, char** argv) { + for (int i = 2; i + 1 < argc; i += 2) { + if (std::string(argv[i]) == "--workload-gemm-dim") { + const long long v = std::atoll(argv[i + 1]); + if (v > 0) { + return static_cast(v); + } + } + } + return 1024; // default square GEMM side length +} + +int parse_workload_fft_len(int argc, char** argv) { for (int i = 2; i + 1 < argc; i += 2) { - if (std::string(argv[i]) == "--workload-batch-bytes") { + if (std::string(argv[i]) == "--workload-fft-len") { const long long v = std::atoll(argv[i + 1]); if (v > 0) { - return static_cast(v); + return static_cast(v); } } } - return 0; + return kFftLen; // default 1-D C2C transform length +} + +int parse_workload_sync_interval(int argc, char** argv) { + for (int i = 2; i + 1 < argc; i += 2) { + if (std::string(argv[i]) == "--workload-sync-interval") { + const long long v = std::atoll(argv[i + 1]); + if (v > 0) { + return static_cast(v); + } + } + } + return 2; // default sync depth +} + +int parse_workload_max_inflight(int argc, char** argv) { + for (int i = 2; i + 1 < argc; i += 2) { + if (std::string(argv[i]) == "--workload-max-inflight") { + const long long v = std::atoll(argv[i + 1]); + if (v > 0) { + return static_cast(v); + } + } + } + return 0; // unset -> caller applies its computed default } const char* workload_name(BenchWorkload workload) { @@ -124,6 +170,11 @@ void GpuWorkload::destroy() { cudaFree(gemm_c_); gemm_c_ = nullptr; } + for (void* ev : event_pool_) { + cudaEventDestroy(as_event(ev)); + } + event_pool_.clear(); + free_events_.clear(); if (stream_ != nullptr) { cudaStreamDestroy(as_stream(stream_)); stream_ = nullptr; @@ -131,7 +182,8 @@ void GpuWorkload::destroy() { ok_ = false; } -bool GpuWorkload::init(BenchWorkload kind, size_t batch_bytes, int sync_interval) { +bool GpuWorkload::init(BenchWorkload kind, size_t input_bytes, int sync_interval, + int gemm_dim, int fft_len) { kind_ = kind; sync_interval_ = sync_interval > 0 ? sync_interval : 1; run_count_ = 0; @@ -140,7 +192,7 @@ bool GpuWorkload::init(BenchWorkload kind, size_t batch_bytes, int sync_interval return true; // inert no-op object; not an error } - const size_t bytes = batch_bytes > 0 ? batch_bytes : kDefaultBytes; + const size_t bytes = input_bytes > 0 ? input_bytes : kDefaultBytes; cudaStream_t stream = nullptr; if (cudaStreamCreate(&stream) != cudaSuccess) { @@ -151,12 +203,15 @@ bool GpuWorkload::init(BenchWorkload kind, size_t batch_bytes, int sync_interval stream_ = stream; if (kind_ == BenchWorkload::Fft) { - // Fan the working set out across batched length-kFftLen C2C transforms. The + // 1-D C2C transform length, pinned via --workload-fft-len (default kFftLen); + // held fixed while the I/O unit is swept, like the GEMM side length. + fft_len_ = std::max(2, fft_len > 0 ? fft_len : kFftLen); + // Fan the working set out across batched length-fft_len_ 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; + const size_t n_complex = std::max(fft_len_, bytes / sizeof(cufftComplex)); + const int batch = std::max(1, static_cast(n_complex / fft_len_)); + const size_t total = static_cast(fft_len_) * batch; if (cudaMalloc(&fft_out_, total * sizeof(cufftComplex)) != cudaSuccess || cudaMemset(fft_out_, 0, total * sizeof(cufftComplex)) != cudaSuccess) { @@ -165,7 +220,7 @@ bool GpuWorkload::init(BenchWorkload kind, size_t batch_bytes, int sync_interval return false; } cufftHandle plan = 0; - if (cufftPlan1d(&plan, kFftLen, CUFFT_C2C, batch) != CUFFT_SUCCESS) { + if (cufftPlan1d(&plan, fft_len_, CUFFT_C2C, batch) != CUFFT_SUCCESS) { std::cerr << "GpuWorkload: cufftPlan1d failed\n"; destroy(); return false; @@ -176,16 +231,35 @@ bool GpuWorkload::init(BenchWorkload kind, size_t batch_bytes, int sync_interval destroy(); return false; } + // Explicit shape so every published FFT number carries its compute size. + // ~5·N·log2(N) flops per length-N C2C transform, times `batch` transforms. + const double flops = 5.0 * fft_len_ * std::log2(static_cast(fft_len_)) * batch; + std::cerr << "GpuWorkload: fft shape = " << batch << " x C2C length-" << fft_len_ + << " (working set " << (bytes >> 10) << " KiB, " << (flops / 1e6) << " MFLOP/call)\n"; } 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))); + // Square matmul with the side length n pinned directly (FP32 and FP16 use the + // SAME dimension -> identical FLOP count, isolating the precision / tensor-core + // effect). n is held FIXED while the I/O unit is swept, so 2*n^3 FLOPs/call + // stays constant -- this isolates pipelining depth from problem size in the + // RoCE-vs-raw study. The A operand is the caller's input buffer (n*n*elem_size + // must be <= input_bytes, enforced below); B and C are owned scratch. + int n = gemm_dim > 0 ? gemm_dim : 1024; 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); + // The A operand is read from the caller's received-data buffer, which holds + // `bytes`. The pinned n must fit -- otherwise cuBLAS reads past the buffer. + // Reject rather than silently OOB-read; the caller should use a larger I/O unit + // (message / reorder window) for this GEMM dimension. + if (elems * elem_size > bytes) { + std::cerr << "GpuWorkload: gemm n=" << n << " needs " << (elems * elem_size >> 10) + << " KiB for the A operand but the input buffer is only " << (bytes >> 10) + << " KiB; use a larger I/O unit (>= n*n*elem_size) or a smaller " + "--workload-gemm-dim\n"; + destroy(); + return false; + } if (cudaMalloc(&gemm_b_, elems * elem_size) != cudaSuccess || cudaMalloc(&gemm_c_, elems * elem_size) != cudaSuccess || cudaMemset(gemm_b_, 0, elems * elem_size) != cudaSuccess) { @@ -205,6 +279,12 @@ bool GpuWorkload::init(BenchWorkload kind, size_t batch_bytes, int sync_interval destroy(); return false; } + // Explicit shape so every published GEMM number carries its compute size. A + // square n×n×n matmul is 2·n³ flops; n is pinned via --workload-gemm-dim. + const double flops = 2.0 * static_cast(n) * n * n; + std::cerr << "GpuWorkload: " << workload_name(kind_) << " shape = " << n << "x" << n << "x" << n + << " (pinned), " << (elem_size == sizeof(__half) ? "fp16" : "fp32") << " A/B, " + << (flops / 1e9) << " GFLOP/call, matrix " << (elems * elem_size >> 10) << " KiB\n"; } // Warm up and validate the chosen op once against a transient zeroed input @@ -260,6 +340,64 @@ void GpuWorkload::run(const void* input) { maybe_sync(); } +void GpuWorkload::run_async(const void* input) { + if (!enabled() || input == nullptr) { + return; + } + issue_op(input); + ++run_count_; + // No maybe_sync: the caller bounds outstanding work with events. +} + +void* GpuWorkload::record_event() { + if (!enabled()) { + return nullptr; + } + // Lazily create the pool on first use so the run()+sync() callers (DPDK/socket) + // that never record events pay nothing. + if (event_pool_.empty()) { + event_pool_.reserve(kEventPoolSize); + free_events_.reserve(kEventPoolSize); + for (int i = 0; i < kEventPoolSize; ++i) { + cudaEvent_t ev = nullptr; + if (cudaEventCreateWithFlags(&ev, cudaEventDisableTiming) != cudaSuccess) { + break; // partial pool is fine; record_event just returns null when empty + } + event_pool_.push_back(ev); + free_events_.push_back(ev); + } + } + if (free_events_.empty()) { + return nullptr; // caller must drain (event_done / wait_event) and retry + } + void* ev = free_events_.back(); + free_events_.pop_back(); + if (cudaEventRecord(as_event(ev), as_stream(stream_)) != cudaSuccess) { + free_events_.push_back(ev); + return nullptr; + } + return ev; +} + +bool GpuWorkload::event_done(void* ev) const { + if (ev == nullptr) { + return true; + } + return cudaEventQuery(as_event(ev)) == cudaSuccess; +} + +void GpuWorkload::wait_event(void* ev) { + if (ev != nullptr) { + cudaEventSynchronize(as_event(ev)); + } +} + +void GpuWorkload::release_event(void* ev) { + if (ev != nullptr) { + free_events_.push_back(ev); + } +} + void GpuWorkload::maybe_sync() { if (!enabled()) { return; diff --git a/examples/bench_workload.h b/examples/bench_workload.h index 63ccbea..0eb18b6 100644 --- a/examples/bench_workload.h +++ b/examples/bench_workload.h @@ -18,6 +18,7 @@ #pragma once #include +#include namespace daqiri::bench { @@ -32,11 +33,42 @@ enum class BenchWorkload { None, Fft, Gemm, GemmFp16 }; // 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); +// Parse "--workload-gemm-dim N" from argv: the square GEMM side length n, held +// FIXED so the FLOP count per call (2*n^3) is constant as the I/O unit (message / +// burst window) is swept -- this isolates pipelining depth from problem size in +// the RoCE-vs-raw comparison. The compute working set is n*n*elem_size, derived +// entirely from n; the assembled input buffer must be at least that large. +// Returns 1024 (the default) if unset. Mirrors parse_workload's stride. +int parse_workload_gemm_dim(int argc, char** argv); + +// Parse "--workload-fft-len N" from argv: the 1-D C2C transform length. The burst +// working set is fanned out across as many batched length-N transforms as fit, so +// N sets the per-transform cost while the batch count tracks the I/O unit. +// Independent of --workload-gemm-dim. Returns 1024 (the default) if +// unset. Mirrors parse_workload's stride. +int parse_workload_fft_len(int argc, char** argv); + +// Parse "--workload-sync-interval N" from argv: drain the GPU stream every N compute +// calls (bounds outstanding GPU work). Larger N = deeper async queue, fewer CPU +// stalls waiting on the GPU; N=1 is fully synchronous. Used to characterize how much +// the single-threaded receive+compute loop is limited by sync stalls. Returns 2 (the +// default) if unset. Mirrors parse_workload's stride. +int parse_workload_sync_interval(int argc, char** argv); + +// Upper bound for the RoCE recv-path in-flight cap (--workload-max-inflight, +// below). Kept one below the internal CUDA-event pool (kEventPoolSize in +// bench_workload.cu) so record_event() never starves exactly at the cap and +// falls back to a per-message sync. +constexpr int kMaxWorkloadInflight = 63; + +// Parse "--workload-max-inflight N" from argv: for the event-recycling RoCE recv +// path, the max number of recv buffers that may have in-flight GPU work before +// the receive thread blocks on the oldest event (backpressure). Larger N absorbs +// more receive/compute jitter before stalling reposts, bounded by the event pool +// (kMaxWorkloadInflight) and the posted-receive window (rx_depth). Returns 0 when +// unset, so the caller applies its computed default (min(rx_depth/2, 32)). Mirrors +// parse_workload's stride. Ignored by the run()+sync() DPDK/socket path. +int parse_workload_max_inflight(int argc, char** argv); // Lower-case name ("none"/"fft"/"gemm"); used for the run_spark_bench.sh // post_process CSV column and log lines. @@ -65,18 +97,33 @@ class 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); + // Build the plan/handle for `kind`. `input_bytes` is the size of the contiguous + // buffer the caller will pass to run() (its reorder window / message size, 0 => + // an internal default); it bounds the op's read and is validated against the + // pinned GEMM working set. `gemm_dim` is the square GEMM side length n, held + // fixed so 2*n^3 FLOPs/call stays constant across the I/O sweep (>0 required for + // GEMM; <=0 falls back to the 1024 default). kind == None leaves the object an + // inert no-op. sync_interval bounds outstanding GPU work (sync every N runs) for + // the run()+sync() callers (DPDK/socket); the RoCE path bounds work with events + // instead and ignores it. `fft_len` is the 1-D FFT transform length for the FFT + // kind (>0; <=0 falls back to the 1024 default), ignored by the GEMM kinds. + // Logs the chosen problem shape (GEMM n / FFT length+batch) and FLOP count so + // every published benchmark number carries its explicit compute size. 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 input_bytes, int sync_interval = 2, int gemm_dim = 1024, + int fft_len = 1024); // Enqueue one representative FFT/SGEMM on the internal stream, reading `input` - // (a device pointer to >= batch_bytes valid bytes). No-op unless enabled(). + // (a device pointer to >= the GEMM working set of valid bytes). No-op unless + // enabled(). void run(const void* input); + // Like run(), but enqueue only -- never blocks (no maybe_sync). For callers that + // bound outstanding GPU work with events (record_event) instead of the periodic + // maybe_sync drain, so the receive thread is never parked on the GPU. + void run_async(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(); @@ -84,6 +131,22 @@ class GpuWorkload { // Drain any remaining queued work (call once on shutdown). void sync(); + // Event-based buffer recycling for callers that read a still-owned input buffer + // zero-copy (the RoCE recv path): record an event on the stream after the ops + // for one input have been enqueued; the event completes only once those ops + // (and thus the reads of that input) finish. Poll it non-blocking with + // event_done() to free/repost the input, or block on it (wait_event) for + // backpressure / shutdown. Handles come from a fixed internal pool; release_event + // returns one for reuse. All are no-ops / return true / null when disabled. + // + // Enabled callers must keep at most pool-capacity events outstanding (wait+ + // release the oldest before exceeding it); record_event returns null if the pool + // is momentarily exhausted so the caller can drain and retry. + void* record_event(); + bool event_done(void* ev) const; + void wait_event(void* ev); + void release_event(void* ev); + bool enabled() const { return kind_ != BenchWorkload::None && ok_; } @@ -116,11 +179,19 @@ class GpuWorkload { void* cublas_ = nullptr; // cublasHandle_t int fft_plan_ = -1; // cufftHandle (-1 == unset) + int fft_len_ = 0; // 1-D C2C transform length (set in init) + // 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; + + // Fixed pool of CUDA events (cudaEvent_t, cast in the .cu) for record_event(). + // Created lazily on first use; freed in destroy(). free_events_ holds the + // currently-available handles. + std::vector event_pool_; // all owned events (for teardown) + std::vector free_events_; // subset available to hand out }; } // 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 510309b..0a11087 100644 --- a/examples/daqiri_bench_rdma_tx_rx_spark_netns.yaml +++ b/examples/daqiri_bench_rdma_tx_rx_spark_netns.yaml @@ -114,7 +114,10 @@ rdma_bench_server: # peer's posted receives and trigger RNR NACKs. Keep <= the MR num_bufs above. rx_depth: 128 tx_depth: 128 - send: true + # One-way (unidirectional) by default, matching the DPDK and socket benches: + # the server only receives, the client only sends. Set send: true here (and + # receive: true on the client) for a bidirectional test. + send: false receive: true server: true @@ -131,6 +134,8 @@ rdma_bench_client: client_address: 10.250.0.1 server_address: 10.250.0.2 server_port: 4096 - receive: true + # One-way (see server note above): the client only sends. Set receive: true for + # a bidirectional test. + receive: false send: true server: false diff --git a/examples/raw_bench_common.cpp b/examples/raw_bench_common.cpp index 1d514a1..7ca75b3 100644 --- a/examples/raw_bench_common.cpp +++ b/examples/raw_bench_common.cpp @@ -549,7 +549,8 @@ void print_queue_stats(const char *direction, const std::string &interface_name, } void rx_count_worker(const RawBenchRxConfig& cfg, std::atomic& stop, BenchWorkload workload, - const ReorderGeometry& geom) { + const ReorderGeometry& geom, int workload_gemm_dim, + int workload_sync_interval, int workload_fft_len) { if (!set_current_thread_affinity(cfg.cpu_core, "bench_rx")) { stop.store(true); return; @@ -564,7 +565,8 @@ void rx_count_worker(const RawBenchRxConfig& cfg, std::atomic& stop, Bench GpuWorkload gpu_workload; ReorderPipeline pipeline; if (workload != BenchWorkload::None) { - if (!gpu_workload.init(workload, batch_bytes) || + if (!gpu_workload.init(workload, batch_bytes, workload_sync_interval, workload_gemm_dim, + workload_fft_len) || !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())) { @@ -621,6 +623,11 @@ void rx_count_worker(const RawBenchRxConfig& cfg, std::atomic& stop, Bench // 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. + // Unlike the RoCE path (which reads the recv buffer zero-copy and recycles + // it with CUDA events), the reorder here copies into the pipeline's single + // shared output buffer, so overlapping GEMMs across bursts would need that + // buffer double/ring-buffered first -- hence the per-burst blocking drain. + // The raw path already sustains line rate, so events are not wired in here. pipeline.reset_batch(); for (int i = 0; i < num_pkts; ++i) { pipeline.add_device_packet( diff --git a/examples/raw_bench_common.h b/examples/raw_bench_common.h index c79a605..e57cf41 100644 --- a/examples/raw_bench_common.h +++ b/examples/raw_bench_common.h @@ -160,7 +160,8 @@ struct ReorderGeometry { // 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 = {}); + BenchWorkload workload = BenchWorkload::None, const ReorderGeometry& geom = {}, + int workload_gemm_dim = 1024, int workload_sync_interval = 2, + int workload_fft_len = 1024); } // namespace daqiri::bench diff --git a/examples/raw_gpudirect_bench.cpp b/examples/raw_gpudirect_bench.cpp index 9aaf574..c6eda0b 100644 --- a/examples/raw_gpudirect_bench.cpp +++ b/examples/raw_gpudirect_bench.cpp @@ -163,7 +163,8 @@ int main(int argc, char **argv) { if (argc < 2) { std::cerr << "Usage: " << argv[0] << " [--seconds N] [--target-gbps G] " - "[--workload none|fft|gemm|gemm_fp16] [--workload-batch-bytes N]\n"; + "[--workload none|fft|gemm|gemm_fp16] [--workload-gemm-dim N] " + "[--workload-fft-len N] [--workload-sync-interval N]\n"; return 1; } @@ -172,7 +173,9 @@ int main(int argc, char **argv) { 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 int workload_gemm_dim = daqiri::bench::parse_workload_gemm_dim(argc, argv); + const int workload_fft_len = daqiri::bench::parse_workload_fft_len(argc, argv); + const int workload_sync_interval = daqiri::bench::parse_workload_sync_interval(argc, argv); const auto root = YAML::LoadFile(argv[1]); std::vector rx_configs; @@ -212,18 +215,16 @@ int main(int argc, char **argv) { 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); + // Reorder window: ~8 MB (1024 packets at the native shape), matching the RoCE + // working set. Capped to one burst's packet count. The compute reads the first + // n*n*elem_size bytes of this window (GEMM dimension pinned via + // --workload-gemm-dim), so the window must be at least that large. + geom.packets_per_batch = std::min(1024, 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), workload, geom); + rx_threads.emplace_back(daqiri::bench::rx_count_worker, cfg, std::ref(stop), workload, geom, + workload_gemm_dim, workload_sync_interval, workload_fft_len); } 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 4a1abca..b9e4c6a 100644 --- a/examples/raw_hds_bench.cpp +++ b/examples/raw_hds_bench.cpp @@ -141,13 +141,16 @@ int main(int argc, char **argv) { if (argc < 2) { std::cerr << "Usage: " << argv[0] << " [--seconds N] " - "[--workload none|fft|gemm|gemm_fp16] [--workload-batch-bytes N]\n"; + "[--workload none|fft|gemm|gemm_fp16] [--workload-gemm-dim N] " + "[--workload-fft-len N] [--workload-sync-interval 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 int workload_gemm_dim = daqiri::bench::parse_workload_gemm_dim(argc, argv); + const int workload_fft_len = daqiri::bench::parse_workload_fft_len(argc, argv); + const int workload_sync_interval = daqiri::bench::parse_workload_sync_interval(argc, argv); const auto root = YAML::LoadFile(argv[1]); if (daqiri::daqiri_init(argv[1]) != daqiri::Status::SUCCESS) { std::cerr << "daqiri_init failed\n"; @@ -180,14 +183,14 @@ int main(int argc, char **argv) { 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); + // Reorder window: ~8 MB (1024 packets at the native shape), capped to one + // burst. The compute reads the first n*n*elem_size bytes (GEMM dimension + // pinned via --workload-gemm-dim). + geom.packets_per_batch = std::min(1024, tx.batch_size); } - rx_thread = std::thread(daqiri::bench::rx_count_worker, daqiri::bench::parse_rx(root), - std::ref(stop), workload, geom); + rx_thread = + std::thread(daqiri::bench::rx_count_worker, daqiri::bench::parse_rx(root), std::ref(stop), + workload, geom, workload_gemm_dim, workload_sync_interval, workload_fft_len); } if (has_tx) { tx_thread = diff --git a/examples/rdma_bench.cpp b/examples/rdma_bench.cpp index 9cec488..0b6dcce 100644 --- a/examples/rdma_bench.cpp +++ b/examples/rdma_bench.cpp @@ -22,6 +22,7 @@ #include #include #include +#include #include #include #include @@ -81,7 +82,8 @@ RdmaBenchConfig parse_rdma_cfg(const YAML::Node& node) { void rdma_worker(const RdmaBenchConfig& cfg, daqiri::bench::TokenBucketPacer& pacer, std::atomic& stop, RdmaWorkerStats& stats, - daqiri::bench::BenchWorkload workload, size_t workload_batch_bytes) { + daqiri::bench::BenchWorkload workload, int workload_gemm_dim, + int workload_sync_interval, int workload_max_inflight, int workload_fft_len) { const char *thread_name = cfg.server ? "rdma_bench_server" : "rdma_bench_client"; if (!daqiri::bench::set_current_thread_affinity(cfg.cpu_core, thread_name)) { @@ -95,19 +97,15 @@ void rdma_worker(const RdmaBenchConfig& cfg, daqiri::bench::TokenBucketPacer& pa // 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. + // One fixed-dimension compute per received message, reading the first + // n*n*elem_size bytes of the message (GpuWorkload::init rejects a message too + // small to hold the pinned GEMM operand). 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) || + if (!gpu_workload.init(workload, msg, workload_sync_interval, workload_gemm_dim, + workload_fft_len) || !pipeline.init(daqiri::bench::ReorderMode::GatherOnly, /*packets_per_batch=*/1, msg, /*payload_byte_offset=*/0, /*seq_bit_offset=*/0, @@ -117,23 +115,49 @@ void rdma_worker(const RdmaBenchConfig& cfg, daqiri::bench::TokenBucketPacer& pa } 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; + // To overlap receive with compute (apples-to-apples with the raw path), hold each + // RECEIVE completion until the GPU has finished reading its recv buffer, then + // free/repost it -- WITHOUT blocking the single receive thread. The compute reads + // the recv buffer zero-copy, so the buffer must stay alive until its GEMM + // completes. We stamp a CUDA event on the stream after each message's GEMM and + // free the buffer only once event_done() reports the read is finished, so the + // thread keeps reaping completions and reposting meanwhile. Outstanding GPU work + // is bounded by max_inflight (backpressure, not a periodic hard drain), which is + // what keeps the throughput measurement honest. Bounded well under rx_depth (and + // the event pool) so receives are not starved. + struct HeldRecv { + daqiri::BurstParams* buf; + void* event; + }; + std::deque held_recv; + // Default cap min(rx_depth/2, 32); --workload-max-inflight overrides it. Clamp to + // the event pool (kMaxWorkloadInflight -- else record_event() starves at the cap + // and falls back to a per-message sync) and to rx_depth (else we would block + // reposting more receives than are posted). + size_t inflight_cap = workload_max_inflight > 0 + ? static_cast(workload_max_inflight) + : std::min(cfg.rx_depth / 2, 32); + inflight_cap = + std::min(inflight_cap, static_cast(daqiri::bench::kMaxWorkloadInflight)); + inflight_cap = std::min(inflight_cap, static_cast(cfg.rx_depth)); + const size_t max_inflight = run_workload ? std::max(1, inflight_cap) : 0; + // Free every held buffer whose GEMM has completed (non-blocking); keeps the recv + // pool drained without parking the thread on the GPU. + auto reclaim_completed = [&]() { + while (!held_recv.empty() && gpu_workload.event_done(held_recv.front().event)) { + daqiri::free_tx_burst(held_recv.front().buf); + gpu_workload.release_event(held_recv.front().event); + held_recv.pop_front(); } - gpu_workload.sync(); - for (auto* held : held_recv) { - daqiri::free_tx_burst(held); + }; + // Block until every held buffer's GEMM is done, then free them all (shutdown / + // connection loss). Events complete in stream order, so waiting front-to-back + // drains the whole queue. + auto drain_held_recv = [&]() { + for (auto& held : held_recv) { + gpu_workload.wait_event(held.event); + daqiri::free_tx_burst(held.buf); + gpu_workload.release_event(held.event); } held_recv.clear(); }; @@ -258,20 +282,38 @@ void rdma_worker(const RdmaBenchConfig& cfg, daqiri::bench::TokenBucketPacer& pa 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); + // One fixed-dimension compute per message on its real (zero-copy) data. + // Enqueue only (never blocks); the GPU work overlaps with subsequent + // receives and is reclaimed once its event completes. + gpu_workload.run_async(pipeline.finish_batch()); + void* ev = gpu_workload.record_event(); + if (ev == nullptr) { + // Event pool momentarily exhausted (defensive: max_inflight is clamped + // below the pool, so steady state never starves). Drain and free now so + // we never read/repost a buffer the GPU is still using. + gpu_workload.sync(); + daqiri::free_tx_burst(completion); + } else { + // Hold the completion (its recv buffer) until the event says the GEMM + // has finished reading it. + held_recv.push_back({completion, ev}); } - // 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(); + // Reclaim completed GEMMs FIRST (non-blocking) so finished work frees its + // held slot before we ever consider blocking. In steady state the GPU keeps + // up (GEMM ~337us << message interval), so this drops held below the cap + // every iteration and the block below never fires -- the thread keeps + // launching GEMMs and reposting receives, so the GPU stays fed and the wire + // stays full. (Parking here instead is what starved the GPU and throttled + // the sender: it stalled launches/reposts every message.) + reclaim_completed(); + // Last-resort backpressure: only park if, even after reclaiming, held is + // still at the cap -- i.e. the GPU is genuinely behind (rare; bounds CPU + // run-ahead for measurement honesty), not the per-message stall it was. + if (held_recv.size() >= max_inflight) { + gpu_workload.wait_event(held_recv.front().event); + reclaim_completed(); // free the one we just drained } - continue; // do not free yet; freed by flush_held_recv() + continue; // do not free yet; freed by reclaim_completed()/drain_held_recv() } } daqiri::free_tx_burst(completion); @@ -288,13 +330,19 @@ 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)); + // Out of completions for the moment. Do NOT touch the GPU here: neither block + // (drain_held_recv parks the thread on each in-flight GEMM) NOR hot-spin + // reclaim_completed() -- polling cudaEventQuery millions of times stole ~40% of + // the thread from wire servicing, throttling the sender just as badly. The main + // receive path already reclaims finished GEMMs on every message, so here we just + // keep polling get_rx_burst (cheap, no CUDA) for the next completion, and back + // off only when there is genuinely nothing in flight. + if (held_recv.empty()) { + std::this_thread::sleep_for(std::chrono::microseconds(100)); + } } } - flush_held_recv(); + drain_held_recv(); gpu_workload.sync(); } @@ -305,7 +353,8 @@ int main(int argc, char** argv) { std::cerr << "Usage: " << argv[0] << " [--seconds N] [--mode server|client|both] " "[--target-gbps G] [--workload none|fft|gemm|gemm_fp16] " - "[--workload-batch-bytes N]\n"; + "[--workload-gemm-dim N] [--workload-fft-len N] " + "[--workload-sync-interval N] [--workload-max-inflight N]\n"; return 1; } @@ -322,7 +371,10 @@ int main(int argc, char** 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 int workload_gemm_dim = daqiri::bench::parse_workload_gemm_dim(argc, argv); + const int workload_fft_len = daqiri::bench::parse_workload_fft_len(argc, argv); + const int workload_sync_interval = daqiri::bench::parse_workload_sync_interval(argc, argv); + const int workload_max_inflight = daqiri::bench::parse_workload_max_inflight(argc, argv); const auto root = YAML::LoadFile(argv[1]); if (daqiri::daqiri_init(argv[1]) != daqiri::Status::SUCCESS) { @@ -359,11 +411,13 @@ 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), workload, workload_batch_bytes); + std::ref(server_stats), workload, workload_gemm_dim, + workload_sync_interval, workload_max_inflight, workload_fft_len); } if (run_client) { client_thread = std::thread(rdma_worker, client_cfg, std::ref(client_pacer), std::ref(stop), - std::ref(client_stats), workload, workload_batch_bytes); + std::ref(client_stats), workload, workload_gemm_dim, + workload_sync_interval, workload_max_inflight, workload_fft_len); } if (!server_thread.joinable() && !client_thread.joinable()) { diff --git a/examples/run_spark_bench.sh b/examples/run_spark_bench.sh index 4e97e7c..25358e3 100755 --- a/examples/run_spark_bench.sh +++ b/examples/run_spark_bench.sh @@ -29,6 +29,17 @@ # tensor-core matmul). Honoured by all backends (dpdk, rdma, # socket-udp, socket-tcp); recorded in the CSV post_process # column. +# GEMM_DIM — the square GEMM side length n (--workload-gemm-dim; default +# 1024), held fixed so FLOPs/call (2·n³) is constant. The +# compute working set is n·n·elem_size, read from the front of +# each received I/O unit. Recorded in post_process_gemm_dim. +# FFT_LEN — the 1-D C2C transform length (--workload-fft-len; default +# 1024) for WORKLOAD=fft, held fixed while the I/O unit is +# swept. Independent of GEMM_DIM. +# SYNC_INTERVAL — drain the GPU stream every N compute calls +# (--workload-sync-interval; default 2). Sweep it (1 2 4 8 16 32) +# to see how much of the receive+compute ceiling is single-thread +# GPU sync-stall. Recorded in post_process_sync. # # 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 / @@ -68,14 +79,17 @@ 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. -# 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" +# post_process_gemm_dim = GEMM_DIM pinned dimension (default 1024). +# post_process_sync (last column) = SYNC_INTERVAL, or "default" (2) when unset. +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_gemm_dim,post_process_sync" > "$CSV" # Capture slow-moving environment state once per result set. "$SCRIPT_DIR/bench_capture_environment.sh" "$OUT_DIR" -RUN_SECONDS=30 +RUN_SECONDS="${RUN_SECONDS:-30}" +if [[ ! "$RUN_SECONDS" =~ ^[0-9]+$ || "$RUN_SECONDS" -lt 1 ]]; then + echo "Invalid RUN_SECONDS '$RUN_SECONDS' (expected a positive integer)" >&2; exit 1 +fi # Repeats per cell for error bars. Each rep is a full independent run with its own # 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. @@ -90,13 +104,62 @@ 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 +# GEMM_DIM: the square GEMM side length n (--workload-gemm-dim), held FIXED so the +# FLOP count per call (2·n³) is constant. The compute working set is n·n·elem_size, +# read from the front of each received I/O unit (RoCE message / raw reorder window), +# so that unit must be at least that large. Default 1024. Recorded in the CSV +# post_process_gemm_dim column. +GEMM_DIM="${GEMM_DIM:-1024}" +if [[ ! "$GEMM_DIM" =~ ^[1-9][0-9]*$ ]]; then + echo "Invalid GEMM_DIM '$GEMM_DIM' (expected a positive integer)" >&2; exit 1 +fi +# FFT_LEN: the 1-D C2C transform length (--workload-fft-len) for WORKLOAD=fft, held +# FIXED while the I/O unit is swept. Independent of GEMM_DIM. Default 1024. +FFT_LEN="${FFT_LEN:-1024}" +if [[ ! "$FFT_LEN" =~ ^[1-9][0-9]*$ ]]; then + echo "Invalid FFT_LEN '$FFT_LEN' (expected a positive integer)" >&2; exit 1 +fi +# SYNC_INTERVAL: drain the GPU stream every N compute calls (--workload-sync-interval). +# Larger N lets more GEMMs queue asynchronously before the single receive+compute +# thread blocks on the GPU, so sweeping it (e.g. 1 2 4 8 16 32) shows how much of the +# ceiling is CPU sync-stall vs the receive path. Empty = default (2). Recorded in the +# CSV post_process_sync column. +SYNC_INTERVAL="${SYNC_INTERVAL:-}" +if [[ -n "$SYNC_INTERVAL" && ! "$SYNC_INTERVAL" =~ ^[0-9]+$ ]]; then + echo "Invalid SYNC_INTERVAL '$SYNC_INTERVAL' (expected a positive integer)" >&2; exit 1 +fi +# MAX_INFLIGHT: for the RoCE event-recycling recv path, the max recv buffers with +# in-flight GPU work before the receive thread blocks on the oldest event +# (--workload-max-inflight). Larger absorbs more jitter before stalling reposts; +# clamped in the bench to the event pool (63) and rx_depth. Empty = bench default +# (min(rx_depth/2, 32)). Ignored by the run()+sync() DPDK/socket path. +MAX_INFLIGHT="${MAX_INFLIGHT:-}" +if [[ -n "$MAX_INFLIGHT" && ! "$MAX_INFLIGHT" =~ ^[0-9]+$ ]]; then + echo "Invalid MAX_INFLIGHT '$MAX_INFLIGHT' (expected a positive integer)" >&2; exit 1 +fi +# NSYS: when set (NSYS=1), wrap the RoCE *server* (the receive + GPU-workload +# process) in `nsys profile` to capture the CUDA/GPU timeline and thread states, +# so we can see whether the GPU stream idles between GEMMs (receive-thread cadence +# gaps) or the SM is genuinely busy. Only the server is traced -- the client is a +# pure traffic generator. Per-cell report lands as /roce_server.nsys-rep. +# Use `rdma smoke` + a short RUN_SECONDS so the report stays small. +NSYS="${NSYS:-}" +NSYS_BIN="${NSYS_BIN:-}" +if [[ -n "$NSYS" ]]; then + if [[ -z "$NSYS_BIN" ]]; then + # Resolve nsys: PATH first, then the usual CUDA toolkit location. + if command -v nsys >/dev/null 2>&1; then + NSYS_BIN="$(command -v nsys)" + elif [[ -x /usr/local/cuda/bin/nsys ]]; then + NSYS_BIN="/usr/local/cuda/bin/nsys" + fi + fi + if [[ -z "$NSYS_BIN" || ! -x "$NSYS_BIN" ]]; then + echo "NSYS set but no nsys binary found (looked on PATH and /usr/local/cuda/bin)." >&2 + echo "Set NSYS_BIN=/path/to/nsys explicitly." >&2 + exit 1 + fi + echo "nsys profiling enabled: $NSYS_BIN" >&2 fi DRIVER_LOG="$OUT_DIR/last_run.stderr" FAILURES=0 @@ -138,7 +201,23 @@ case "$BACKEND" in # the kernel's local routing table. BASE_YAML="$SCRIPT_DIR/daqiri_bench_rdma_tx_rx_spark_netns.yaml" BENCH_BIN="$BUILD_DIR/examples/daqiri_bench_rdma" - CPU_MASTER=8; CPU_TX=17; CPU_RX=18 + # One-way roles: the client (send-only) drives the TX-queue core 17; the server + # (receive-only) runs its RX-queue poller AND bench worker on core 19. Measure + # the sender core as cpu_tx and the SERVER RECEIVE core (19, not the idle + # client-RX core 18) as cpu_rx, so cpu_rx_pct reflects the true RoCE RC + # receiver cost. + CPU_MASTER=8; CPU_TX=17; CPU_RX=19 + # Resolve the server (RX) / client (TX) netdevs inside the wire-loopback + # namespaces so each cell can assert wire transit via *_phy SerDes counters, + # exactly like the dpdk path. RoCE loops over the SAME cable, so a non-advancing + # server rx_packets_phy flags the on-chip eswitch short-cut instead of a true + # over-the-cable loopback -- the tell for a RoCE number above the ~99 Gb/s + # 100GbE line-rate ceiling. Empty if the netns is not up yet (the per-cell check + # then just skips with a WARN). Override RDMA_{SERVER,CLIENT}_NETDEV if needed. + RDMA_SERVER_NS="${RDMA_SERVER_NS:-dq_wire_server}" + RDMA_CLIENT_NS="${RDMA_CLIENT_NS:-dq_wire_client}" + RDMA_SERVER_NETDEV="${RDMA_SERVER_NETDEV:-$(ip netns exec "$RDMA_SERVER_NS" ls /sys/class/net 2>/dev/null | grep -vx lo | head -n1 || true)}" + RDMA_CLIENT_NETDEV="${RDMA_CLIENT_NETDEV:-$(ip netns exec "$RDMA_CLIENT_NS" ls /sys/class/net 2>/dev/null | grep -vx lo | head -n1 || true)}" ;; # Single-frame UDP sizes (<= the ~8972 B MTU payload, so no IP fragmentation). # 65507 is intentionally excluded: it fragments into ~8 packets and, under @@ -187,6 +266,18 @@ case "$BACKEND" in *) echo "Unknown backend: $BACKEND" >&2; exit 1 ;; esac +# When a GPU workload is active, its pinned GEMM operand (n·n·elem_size, from +# GEMM_DIM) must fit inside each received I/O unit -- the small entries in the +# default payload sweep can't hold it. Restrict the sweep to the headline +# (native-shape) size so the fixed-n comparison is a single clean point per +# backend instead of silently disabling the workload on the small cells. +# e.g. WORKLOAD=gemm_fp16 GEMM_DIM=1024 REPEATS=3 ./run_spark_bench.sh rdma sweep +if [[ "$WORKLOAD" != "none" ]]; then + PAYLOADS_SWEEP=("${PAYLOADS_HEADLINE[@]}") +fi +# Optional space-separated override for the batch ladder (one-off experiments). +[[ -n "${BATCHES_OVERRIDE:-}" ]] && read -r -a BATCHES_SWEEP <<< "$BATCHES_OVERRIDE" + # 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. @@ -275,10 +366,13 @@ cpu_busy_pct() { # Sum a NIC *_phy SerDes counter (proves traffic crossed the cable, not an on-chip # eswitch short-cut). Empty netdev -> 0. Mirrors run_spark_mq_bench.sh's phy check. +# Optional third arg is a network namespace (the RoCE/socket wire loopback moves the +# cabled netdevs into dq_wire_{server,client}); empty reads the current namespace. phy_counter() { - local netdev="$1" key="$2" + local netdev="$1" key="$2" ns="${3:-}" [[ -z "$netdev" ]] && { echo 0; return; } - ethtool -S "$netdev" 2>/dev/null \ + { if [[ -n "$ns" ]]; then ip netns exec "$ns" ethtool -S "$netdev" 2>/dev/null + else ethtool -S "$netdev" 2>/dev/null; fi; } \ | awk -F'[: ]+' -v k="$key" '$2 == k { s += $3 } END { printf "%d", s+0 }' } @@ -300,10 +394,14 @@ generate_yaml() { # (rx 512 / tx 128) but are capped so num_bufs * buf_size stays within a memory # budget per region -- small messages get the full deep window (where RNR NACKs # bite), large messages stay memory-bounded (where window depth is irrelevant). - local budget=1073741824 # 1 GiB pinned per memory region + # Overridable for the buffer-depth experiment: RDMA_BUDGET_GIB raises the + # per-region pinned-memory cap, RDMA_RX_NB / RDMA_TX_NB raise the window + # ceilings (num_bufs + {rx,tx}_depth). Defaults reproduce the shipped sizing + # (1 GiB, rx 512 / tx 128). + local budget=$(( ${RDMA_BUDGET_GIB:-1} * 1073741824 )) # pinned per memory region local cap=$(( budget / payload )); (( cap < 1 )) && cap=1 - local rx_nb=512; (( rx_nb > cap )) && rx_nb=$cap - local tx_nb=128; (( tx_nb > cap )) && tx_nb=$cap + local rx_nb=${RDMA_RX_NB:-512}; (( rx_nb > cap )) && rx_nb=$cap + local tx_nb=${RDMA_TX_NB:-128}; (( tx_nb > cap )) && tx_nb=$cap # Split the combined base per role, then apply the per-message-size window # rewrite to each. Server -> $out, client -> ${out%.yaml}_client.yaml. local role dst @@ -311,7 +409,9 @@ generate_yaml() { if [[ "$role" == server ]]; then dst="$out"; else dst="${out%.yaml}_client.yaml"; fi # name-anchored num_bufs rewrite: RX regions -> rx_nb, TX regions -> tx_nb. # Depths are clamped to their region's num_bufs so the window never exceeds - # the buffers backing it. + # the buffers backing it. One-way (server receive-only + GEMM, client + # send-only) is baked into the base config, matching the DPDK and socket + # benches -- see the send:/receive: notes in the base YAML. python3 "$NETNS_GEN" "$BASE_YAML" --role "$role" | \ awk -v p="$payload" -v bs="$payload" -v rxnb="$rx_nb" -v txnb="$tx_nb" ' /^[[:space:]]*- name:/ { region = $0 } @@ -331,36 +431,58 @@ generate_yaml() { esac } -# One isolated core per socket pair: 16 + (idx % 4) -> cores 16-19 for pairs 0-3 (Spark -# isolcpus), wrapping (oversubscribing) beyond four. Both the server and the client of a -# pair pin to this same core. Sharing one CPU across the send and receive sides couples -# the send rate to the receive rate (the sender cannot outrun the receiver it time-slices -# with), so each pair sits near the per-core ceiling and N pairs scale to ~N x that. This -# matches the reference four-pair methodology and keeps App TX ~= App RX with low loss. -pair_core() { echo $(( 16 + ($1 % 4) )); } +# Separate isolated cores for the send (client) and receive (server) sides of each +# socket pair, so a pair's two processes never time-slice one CPU. Co-locating them +# (the old "one core per pair" setup) forces a send/recv ping-pong on a single core +# that, at mid-size messages (e.g. 8000 B), makes a single TCP stream metastable -- +# it locks into an efficient (~30 Gb/s) or a serialized (~15 Gb/s) mode for a whole +# run, producing bimodal, high-variance results. +# +# CRITICAL: keep each pair's two cores in the SAME CPU cluster. GB10's ten big X925 +# cores are two clusters of five (5-9 and 15-19); straddling a pair across clusters +# makes every payload pay cross-cluster cache-coherence latency and *regresses* large +# messages (1 MiB 4-pair dropped ~90 -> ~77 Gb/s in testing). So pairs 0-1 sit in the +# 15-19 cluster and pairs 2-3 in the 5-9 cluster, each send/recv intra-cluster +# (master 8, spare 15). NOTE: separating send/recv also lets the UDP sender outrun the +# receiver (no shared-core self-pacing), so UDP loss rises vs the co-pinned setup -- +# the more representative behaviour for an unpaced sender. +SRV_PIN_CORES=(16 18 5 7) +CLI_PIN_CORES=(17 19 6 9) +pair_server_core() { echo "${SRV_PIN_CORES[$(( $1 % 4 ))]}"; } +pair_client_core() { echo "${CLI_PIN_CORES[$(( $1 % 4 ))]}"; } # Write the server/client YAML pair for socket pair `idx`: split the combined base # per role, then substitute message_size, unique ports (SRV/CLI_PORT_BASE + idx), -# and pin every queue to the pair's core. +# and pin the server (receive) side and client (send) side to DIFFERENT isolated +# cores so the pair does not time-slice one CPU (see pair_server_core comment). generate_socket_yaml() { local idx="$1" payload="$2" server_out="$3" client_out="$4" local srv_port=$(( SRV_PORT_BASE + idx )) local cli_port=$(( CLI_PORT_BASE + idx )) - local core; core="$(pair_core "$idx")" + # SOCKET_NOPIN=1 runs the bench workers unpinned (cpu_core -1 -> no affinity, the + # scheduler places them). For gathering the pinned-vs-non-pinned comparison only; + # the published report stays pinned like the other backends. + local server_core client_core + if [[ -n "${SOCKET_NOPIN:-}" ]]; then + server_core=-1; client_core=-1 + else + server_core="$(pair_server_core "$idx")" + client_core="$(pair_client_core "$idx")" + fi python3 "$NETNS_GEN" "$BASE_YAML" --role server | \ sed -E \ -e "s|^( *message_size: ).*|\1$payload|g" \ - -e "s|^( *local_addr: \"[a-z]+://[0-9.]+:)[0-9]+(\")|\1$srv_port\2|" \ + -e "s|^( *local_addr: \"?[a-z]+://[0-9.]+:)[0-9]+(\"?)|\1$srv_port\2|" \ -e "s|^( *server_port: ).*|\1$srv_port|" \ - -e "s|^( *cpu_core: ).*|\1$core|" \ + -e "s|^( *cpu_core: ).*|\1$server_core|" \ > "$server_out" python3 "$NETNS_GEN" "$BASE_YAML" --role client | \ sed -E \ -e "s|^( *message_size: ).*|\1$payload|g" \ - -e "s|^( *local_addr: \"[a-z]+://[0-9.]+:)[0-9]+(\")|\1$cli_port\2|" \ - -e "s|^( *remote_addr: \"[a-z]+://[0-9.]+:)[0-9]+(\")|\1$srv_port\2|" \ + -e "s|^( *local_addr: \"?[a-z]+://[0-9.]+:)[0-9]+(\"?)|\1$cli_port\2|" \ + -e "s|^( *remote_addr: \"?[a-z]+://[0-9.]+:)[0-9]+(\"?)|\1$srv_port\2|" \ -e "s|^( *server_port: ).*|\1$srv_port|" \ - -e "s|^( *cpu_core: ).*|\1$core|" \ + -e "s|^( *cpu_core: ).*|\1$client_core|" \ > "$client_out" } @@ -399,7 +521,10 @@ run_cell() { # 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") + bench_extra+=(--workload-gemm-dim "$GEMM_DIM") + bench_extra+=(--workload-fft-len "$FFT_LEN") + [[ -n "$SYNC_INTERVAL" ]] && bench_extra+=(--workload-sync-interval "$SYNC_INTERVAL") + [[ -n "$MAX_INFLIGHT" ]] && bench_extra+=(--workload-max-inflight "$MAX_INFLIGHT") 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 @@ -456,7 +581,20 @@ run_cell() { # addresses over the wire rather than short-cutting the kernel's local table. local yaml="$cell_dir/config.yaml" generate_yaml "$yaml" "$payload" "$batch" - ip netns exec dq_wire_server "$BENCH_BIN" "$yaml" \ + # Optionally wrap the server (receive + GPU-workload) in nsys. Placed AFTER the + # netns exec so nsys traces the bench, not `ip netns exec`. + local -a nsys_pre=() + if [[ -n "$NSYS" ]]; then + nsys_pre=("$NSYS_BIN" profile --trace=cuda,osrt --sample=cpu --cpuctxsw=process-tree + --force-overwrite=true --output "$cell_dir/roce_server") + fi + # Snapshot the netns *_phy counters around the run to assert the RoCE traffic + # actually crossed the cable (client tx -> server rx over the wire), not the + # on-chip eswitch short-cut that lets a loopback exceed the 100GbE line rate. + local phy_tx_before phy_rx_before + phy_tx_before="$(phy_counter "$RDMA_CLIENT_NETDEV" tx_packets_phy "$RDMA_CLIENT_NS")" + phy_rx_before="$(phy_counter "$RDMA_SERVER_NETDEV" rx_packets_phy "$RDMA_SERVER_NS")" + ip netns exec dq_wire_server "${nsys_pre[@]}" "$BENCH_BIN" "$yaml" \ --seconds "$server_seconds" "${bench_extra[@]}" --mode server \ > "$cell_dir/server_stdout.txt" 2> "$cell_dir/server_stderr.txt" & local server_pid=$! @@ -467,11 +605,39 @@ run_cell() { wait "$server_pid" 2>/dev/null || true cat "$cell_dir/server_stdout.txt" >> "$stdout" cat "$cell_dir/server_stderr.txt" >> "$stderr" + if [[ -n "$NSYS" && -f "$cell_dir/roce_server.nsys-rep" ]]; then + echo "=== nsys GPU kernel summary ($cell) ===" >&2 + "$NSYS_BIN" stats --report cuda_gpu_kern_sum --format table \ + "$cell_dir/roce_server.nsys-rep" >&2 || true + echo "=== nsys CUDA API summary ($cell) ===" >&2 + "$NSYS_BIN" stats --report cuda_api_sum --format table \ + "$cell_dir/roce_server.nsys-rep" >&2 || true + echo "nsys report: $cell_dir/roce_server.nsys-rep" >&2 + fi + local phy_tx_delta phy_rx_delta + phy_tx_delta=$(( $(phy_counter "$RDMA_CLIENT_NETDEV" tx_packets_phy "$RDMA_CLIENT_NS") - phy_tx_before )) + phy_rx_delta=$(( $(phy_counter "$RDMA_SERVER_NETDEV" rx_packets_phy "$RDMA_SERVER_NS") - phy_rx_before )) # RDMA prints "Client complete: ... send_completions=N send_bytes=N seconds=S". pkts="$(extract_field 'Client complete' send_completions "$stdout")" bytes="$(extract_field 'Client complete' send_bytes "$stdout")" secs="$(extract_field 'Client complete' seconds "$stdout")" rx_bytes="$bytes" + # Wire-transit check. On the cable the server's rx_*_phy advances by at least one + # SerDes packet per RDMA message -- many more once a message exceeds the MTU and + # segments -- so a genuine over-the-wire run shows phy_rx_delta >= the message + # count. An on-chip eswitch short-cut leaves rx_*_phy essentially FLAT (only a + # handful of stray background/control packets), so a bare ">0" is not enough: a + # +22 on a 23M-message run passed the old check while never touching the cable. + # Require the delta to reach half the message count (generous margin for counter + # slack) before certifying wire transit. + local phy_min=$(( ${pkts:-0} / 2 )) + if [[ -z "$RDMA_SERVER_NETDEV" ]]; then + echo "WARN: $cell could not resolve server netdev in $RDMA_SERVER_NS (netns up?); skipped wire (*_phy) check" >&2 + elif [[ "$phy_rx_delta" -lt "$phy_min" || "$phy_rx_delta" -le 0 ]]; then + echo "WARN: $cell rx_packets_phy advanced only +$phy_rx_delta on $RDMA_SERVER_NETDEV (expected >= ~$phy_min, one per message) -- traffic likely did NOT cross the wire (on-chip eswitch short-cut?)" >&2 + else + echo "INFO: $cell wire OK -- client tx_packets_phy +$phy_tx_delta, server rx_packets_phy +$phy_rx_delta (>= $phy_min msgs)" >&2 + fi else local yaml="$cell_dir/config.yaml" generate_yaml "$yaml" "$payload" "$batch" @@ -485,13 +651,6 @@ run_cell() { local phy_tx_delta phy_rx_delta phy_tx_delta=$(( $(phy_counter "$DPDK_TX_NETDEV" tx_packets_phy) - phy_tx_before )) phy_rx_delta=$(( $(phy_counter "$DPDK_RX_NETDEV" rx_packets_phy) - phy_rx_before )) - if [[ -z "$DPDK_RX_NETDEV" ]]; then - echo "WARN: $cell could not resolve rx_port netdev ($DPDK_RX_PCI); skipped wire (*_phy) check" >&2 - elif [[ "$phy_rx_delta" -le 0 ]]; then - echo "WARN: $cell rx_packets_phy did not advance ($phy_rx_delta) on $DPDK_RX_NETDEV -- traffic may NOT have crossed the wire (on-chip eswitch short-cut?)" >&2 - else - echo "INFO: $cell wire OK -- p0 tx_packets_phy +$phy_tx_delta, p1 rx_packets_phy +$phy_rx_delta" >&2 - fi # For RX-bearing benches "RX complete" is authoritative; fall back to "TX complete". pkts="$(extract_field 'RX complete' packets "$stdout")" bytes="$(extract_field 'RX complete' bytes "$stdout")" @@ -502,6 +661,19 @@ run_cell() { secs="$(extract_field 'TX complete' seconds "$stdout")" fi rx_bytes="$bytes" + # Wire-transit check. Raw Ethernet is one SerDes packet per app packet, so a + # genuine over-the-wire run advances rx_*_phy by ~the received packet count; an + # on-chip eswitch short-cut leaves it near-flat. Require the delta to reach half + # the packet count -- a bare ">0" would let a stray background packet false-pass + # (the same hole the RoCE check had). + local phy_min=$(( ${pkts:-0} / 2 )) + if [[ -z "$DPDK_RX_NETDEV" ]]; then + echo "WARN: $cell could not resolve rx_port netdev ($DPDK_RX_PCI); skipped wire (*_phy) check" >&2 + elif [[ "$phy_rx_delta" -lt "$phy_min" || "$phy_rx_delta" -le 0 ]]; then + echo "WARN: $cell rx_packets_phy advanced only +$phy_rx_delta on $DPDK_RX_NETDEV (expected >= ~$phy_min) -- traffic likely did NOT cross the wire (on-chip eswitch short-cut?)" >&2 + else + echo "INFO: $cell wire OK -- p0 tx_packets_phy +$phy_tx_delta, p1 rx_packets_phy +$phy_rx_delta (>= $phy_min)" >&2 + fi fi cp "$stderr" "$DRIVER_LOG" @@ -572,9 +744,10 @@ 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)" - 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" \ + local pp_gemm_dim="$GEMM_DIM" + local pp_sync="${SYNC_INTERVAL:-default}" + [[ -z "$pp_sync" ]] && pp_sync="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_gemm_dim,$pp_sync" \ | tee -a "$CSV" } diff --git a/examples/run_spark_mq_bench.sh b/examples/run_spark_mq_bench.sh index 4787678..88a61f9 100755 --- a/examples/run_spark_mq_bench.sh +++ b/examples/run_spark_mq_bench.sh @@ -83,9 +83,11 @@ REPEATS="${REPEATS:-1}" MQ_BASE="$SCRIPT_DIR/daqiri_bench_raw_tx_rx_spark_mq.yaml" MQ_GEN="$SCRIPT_DIR/../scripts/gen_spark_mq_config.py" -# Match run_spark_bench.sh: prefer the installed shared libs, falling back to -# the build tree. Keep both so a fresh build dir still resolves. -export LD_LIBRARY_PATH="/opt/daqiri/lib:$BUILD_DIR:${LD_LIBRARY_PATH:-}" +# Resolve the DAQIRI shared libs from the build tree first. The per-engine +# sub-libraries (libdaqiri_dpdk.so.0 etc.) live in $BUILD_DIR/src, not $BUILD_DIR, +# so that directory must be on the path or the bench fails to load them; keep +# $BUILD_DIR and the installed /opt/daqiri/lib as fallbacks. +export LD_LIBRARY_PATH="$BUILD_DIR/src:$BUILD_DIR:/opt/daqiri/lib:${LD_LIBRARY_PATH:-}" TS="$(date -u +%Y%m%dT%H%M%SZ)" OUT_DIR="$SCRIPT_DIR/../bench-results/$TS-dpdk-mq" @@ -265,14 +267,18 @@ run_cell() { local drops; drops="$(parse_dpdk_drops "$stderr")" - # Wire-traffic confirmation: rx_packets_phy must advance over the run. + # Wire-traffic confirmation: rx_packets_phy must advance by ~the received packet + # count (raw Ethernet is one SerDes packet per app packet). A bare ">0" would let a + # stray background packet false-pass an on-chip eswitch short-cut, so require the + # delta to reach half the packet count. local phy_delta=$(( phy_after - phy_before )) + local phy_min=$(( ${pkts:-0} / 2 )) if [[ -z "$RX_NETDEV" ]]; then echo "WARN: $cell p$payload could not resolve RX netdev for $RX_PCI; skipped wire (*_phy) check" >&2 - elif [[ "$phy_delta" -le 0 ]]; then - echo "WARN: $cell p$payload rx_packets_phy did not advance ($phy_delta) -- traffic may not have crossed the wire" >&2 + elif [[ "$phy_delta" -lt "$phy_min" || "$phy_delta" -le 0 ]]; then + echo "WARN: $cell p$payload rx_packets_phy advanced only +$phy_delta (expected >= ~$phy_min) -- traffic likely did NOT cross the wire (on-chip eswitch short-cut?)" >&2 else - echo "INFO: $cell p$payload wire OK -- rx_packets_phy +$phy_delta" >&2 + echo "INFO: $cell p$payload wire OK -- rx_packets_phy +$phy_delta (>= $phy_min)" >&2 fi # Per-core busy% over the bench window, in CSV column order (see CPU_CORES). diff --git a/examples/socket_bench.cpp b/examples/socket_bench.cpp index 05d1339..1812b62 100644 --- a/examples/socket_bench.cpp +++ b/examples/socket_bench.cpp @@ -102,8 +102,8 @@ bool socket_transport_is_tcp(const YAML::Node& root) { void socket_worker(const SocketBenchConfig& cfg, daqiri::bench::TokenBucketPacer& pacer, std::atomic& stop, SocketWorkerStats& stats, - daqiri::bench::BenchWorkload workload, bool is_tcp, - size_t workload_batch_bytes) { + daqiri::bench::BenchWorkload workload, bool is_tcp, int workload_gemm_dim, + int workload_sync_interval, int workload_fft_len) { const char *thread_name = cfg.server ? "socket_bench_server" : "socket_bench_client"; if (!daqiri::bench::set_current_thread_affinity(cfg.cpu_core, thread_name)) { @@ -116,17 +116,17 @@ void socket_worker(const SocketBenchConfig& cfg, daqiri::bench::TokenBucketPacer // 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; + // The UDP reorder window is a fixed ~8 MB of datagrams (matching the RoCE + // working set); TCP gathers one chunk at a time (one compute per recv). The GEMM + // reads the first n*n*elem_size bytes of that window (dimension pinned below). + const uint32_t target_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) || + if (!gpu_workload.init(workload, static_cast(packets_per_batch) * msg, + workload_sync_interval, workload_gemm_dim, workload_fft_len) || !pipeline.init(is_tcp ? daqiri::bench::ReorderMode::GatherOnly : daqiri::bench::ReorderMode::SeqReorder, packets_per_batch, msg, /*payload_byte_offset=*/0, @@ -241,7 +241,8 @@ int main(int argc, char** argv) { std::cerr << "Usage: " << argv[0] << " [--seconds N] [--mode server|client|both] " "[--target-gbps G] [--workload none|fft|gemm|gemm_fp16] " - "[--workload-batch-bytes N]\n"; + "[--workload-gemm-dim N] [--workload-fft-len N] " + "[--workload-sync-interval N]\n"; return 1; } @@ -258,7 +259,9 @@ int main(int argc, char** 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 int workload_gemm_dim = daqiri::bench::parse_workload_gemm_dim(argc, argv); + const int workload_fft_len = daqiri::bench::parse_workload_fft_len(argc, argv); + const int workload_sync_interval = daqiri::bench::parse_workload_sync_interval(argc, argv); const auto root = YAML::LoadFile(argv[1]); const bool is_tcp = socket_transport_is_tcp(root); @@ -296,11 +299,13 @@ 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), workload, is_tcp, workload_batch_bytes); + std::ref(server_stats), workload, is_tcp, workload_gemm_dim, + workload_sync_interval, workload_fft_len); } if (run_client) { 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); + std::ref(client_stats), workload, is_tcp, workload_gemm_dim, + workload_sync_interval, workload_fft_len); } if (!server_thread.joinable() && !client_thread.joinable()) { diff --git a/scripts/plot_mq_payload_sweep.py b/scripts/plot_mq_payload_sweep.py index 173c3fb..1ecbf4f 100644 --- a/scripts/plot_mq_payload_sweep.py +++ b/scripts/plot_mq_payload_sweep.py @@ -4,8 +4,9 @@ """Render the DGX Spark DPDK multi-queue payload sweep as a line plot. Reads a runs.csv produced by examples/run_spark_mq_bench.sh (columns: -cell,tx_cores,rx_cores,payload,gbps,pps,drops,cpu8,cpu16,cpu17,cpu18,cpu19) -and plots achieved Gb/s vs payload size, one line per (TX,RX) core cell. +cell,tx_cores,rx_cores,payload,rep,gbps,pps,drops,cpu...) and plots achieved +Gb/s vs payload size, one line per (TX,RX) core cell. Multiple reps per +(cell, payload) are averaged. Usage: scripts/plot_mq_payload_sweep.py [output.svg] @@ -36,16 +37,21 @@ def load(csv_path): - """cell -> list of (payload_bytes, gbps), sorted by payload.""" - series = OrderedDict((c, []) for c in CELL_ORDER) + """cell -> list of (payload_bytes, mean gbps over reps), sorted by payload.""" + from collections import defaultdict + + acc = OrderedDict((c, defaultdict(list)) for c in CELL_ORDER) with open(csv_path, newline="") as fh: for row in csv.DictReader(fh): cell = row["cell"].strip() - if cell not in series: - series[cell] = [] # tolerate unexpected cells - series[cell].append((int(row["payload"]), float(row["gbps"]))) - for cell in series: - series[cell].sort(key=lambda pt: pt[0]) + if cell not in acc: + acc[cell] = defaultdict(list) # tolerate unexpected cells + acc[cell][int(row["payload"])].append(float(row["gbps"])) + series = OrderedDict() + for cell, by_payload in acc.items(): + series[cell] = sorted( + (payload, sum(vals) / len(vals)) for payload, vals in by_payload.items() + ) return series diff --git a/scripts/setup_spark_wire_loopback_netns.sh b/scripts/setup_spark_wire_loopback_netns.sh index 67f9ac0..5473da0 100755 --- a/scripts/setup_spark_wire_loopback_netns.sh +++ b/scripts/setup_spark_wire_loopback_netns.sh @@ -41,10 +41,11 @@ # ibdev2netdev # (mlnx) same mapping, friendlier # ethtool | grep -i 'link detected' # the two CABLED ports show "yes" # -# CLIENT_IF / SERVER_IF must be the TWO PORTS YOUR QSFP CABLE PHYSICALLY -# CONNECTS. Your colleague's box cabled f0np0<->f1np1; the repo YAMLs assume the -# two f0np0 ports across PCI segments. Confirm with `ethtool ... link detected` -# -- exactly the two cabled ports report carrier up. +# CLIENT_IF / SERVER_IF must be on the TWO DIFFERENT PHYSICAL PORTS (p0, p1) of +# the CX-7 -- the on-chip loopback the QSFP cable closes. Confirm each netdev's +# physical port with `cat /sys/class/net//phys_port_name` (p0 / p1); a cabled +# loopback lights carrier on all four PFs, so carrier alone does not distinguish +# them. See docs/tutorials/system_configuration.md "Port topology: 4 PFs, 2 ports". # # CLIENT_RDMA / SERVER_RDMA are the RDMA devices bound to those netdevs # (from `rdma link show` / `ibdev2netdev`). @@ -55,13 +56,23 @@ CLIENT_NS=dq_wire_client SERVER_NS=dq_wire_server -CLIENT_IF=enp1s0f0np0 # p0 via PCI segment 0000:01:00.0 (carrier up) -SERVER_IF=enP2p1s0f1np1 # p1 via PCI segment 0002:01:00.1 (carrier up) -- the - # cabled peer of CLIENT_IF (p0<->p1 QSFP loopback). NOT - # enP2p1s0f0np0, which is p0 again (same physical port). - -CLIENT_RDMA=rocep1s0f0 # rdma dev for enp1s0f0np0 -SERVER_RDMA=roceP2p1s0f1 # rdma dev for enP2p1s0f1np1 +# Leave CLIENT_IF / SERVER_IF blank to AUTO-DETECT the cabled data-plane ports +# (recommended): autodetect_ports() collects the carrier-up RoCE netdevs, groups +# them by phys_port_name, and keeps one per physical port -- so the pair always +# straddles both ports (over-the-wire). PCIe enumeration / interface names drift +# across reboots and kernel upgrades on this box, so hardcoding them is fragile +# (a stale name aborts setup, the netns never gets created, and the netns-based +# benches silently have nowhere to run); phys_port_name does not drift. Override +# from the environment only if auto-detect picks the wrong pair, e.g.: +# CLIENT_IF=enp1s0f1np1 SERVER_IF=enp1s0f0np0 sudo -E ./setup_spark_wire_loopback_netns.sh up +CLIENT_IF="${CLIENT_IF:-}" +SERVER_IF="${SERVER_IF:-}" + +# RDMA devices backing the netdevs above. Blank = derive from the netdev via +# sysfs (/sys/class/net//device/infiniband). Only needed for RDMA exclusive +# netns mode; the socket/UDP/TCP benches do not use these. +CLIENT_RDMA="${CLIENT_RDMA:-}" +SERVER_RDMA="${SERVER_RDMA:-}" # Leave these blank to auto-detect from the interfaces (recommended). Only set # them if you need to override the peer MAC for some reason. @@ -83,6 +94,79 @@ require_root() { fi } +rdma_for_netdev() { # netdev -> backing RDMA device name (sysfs); empty if none + local ifc="$1" d + for d in /sys/class/net/"$ifc"/device/infiniband/*; do + [[ -e "$d" ]] || continue + basename "$d" + return 0 + done + return 1 +} + +autodetect_ports() { + # Pick the cabled data-plane ports automatically when not overridden. Spark's + # single ConnectX-7 exposes each of its TWO physical ports (p0, p1) over BOTH + # PCIe segments (socket-direct), so it presents up to 4 PFs and a cabled + # loopback lights carrier on ALL of them at once -- not just two. Counting + # carrier-up netdevs therefore over-counts the physical ports. Instead group by + # phys_port_name (the hardware port label, stable across PCIe re-enumeration and + # interface renames) and keep ONE netdev per physical port, so CLIENT_IF and + # SERVER_IF always land on DIFFERENT physical ports (over-the-wire, never the + # on-chip same-port eswitch shortcut). See docs/tutorials/system_configuration.md + # "Port topology: 4 PFs, 2 ports". Ports must be in init_net here (up() calls + # down() first, returning any moved netdevs), same precondition as detect_macs. + if [[ -z "$CLIENT_IF" || -z "$SERVER_IF" ]]; then + local ifc port pci + declare -A rep_for_port=() pci_for_port=() + for ifc in $(ls /sys/class/net 2>/dev/null | sort); do + [[ "$ifc" == lo ]] && continue + rdma_for_netdev "$ifc" >/dev/null 2>&1 || continue + [[ "$(cat "/sys/class/net/$ifc/carrier" 2>/dev/null || echo 0)" == 1 ]] || continue + port="$(cat "/sys/class/net/$ifc/phys_port_name" 2>/dev/null || true)" + # Without phys_port_name we cannot fold the two PCIe reps of a port + # together; fall back to the netdev name as its own group so the 2-port + # check below still guards (it just won't dedupe). + [[ -z "$port" ]] && port="$ifc" + pci="$(basename "$(readlink -f "/sys/class/net/$ifc/device" 2>/dev/null || echo "$ifc")")" + # Prefer the lowest-BDF rep per port (the 0000: segment, matching the doc's + # canonical enp1s0f{0,1}np{0,1} names) for a stable, least-surprise pick. + if [[ -z "${rep_for_port[$port]:-}" || "$pci" < "${pci_for_port[$port]}" ]]; then + rep_for_port["$port"]="$ifc" + pci_for_port["$port"]="$pci" + fi + done + local ports + mapfile -t ports < <(printf '%s\n' "${!rep_for_port[@]}" | LC_ALL=C sort) + if [[ "${#ports[@]}" -ne 2 ]]; then + echo "ERROR: auto-detect expected exactly 2 carrier-up physical ports, found ${#ports[@]}: ${ports[*]:-none}." >&2 + echo " (netdevs are grouped by phys_port_name; Spark exposes each port over 2 PCIe segments.)" >&2 + echo " Cable the loopback, or set CLIENT_IF / SERVER_IF explicitly, e.g.:" >&2 + echo " CLIENT_IF=enp1s0f1np1 SERVER_IF=enp1s0f0np0 sudo -E $0 up" >&2 + echo " List candidates and their ports with:" >&2 + echo " for d in /sys/class/net/*/device; do n=\${d%/device}; n=\${n##*/}; printf '%s port=%s\\n' \"\$n\" \"\$(cat /sys/class/net/\$n/phys_port_name 2>/dev/null)\"; done" >&2 + exit 1 + fi + [[ -z "$CLIENT_IF" ]] && CLIENT_IF="${rep_for_port[${ports[0]}]}" + [[ -z "$SERVER_IF" ]] && SERVER_IF="${rep_for_port[${ports[1]}]}" + fi + [[ -z "$CLIENT_RDMA" ]] && CLIENT_RDMA="$(rdma_for_netdev "$CLIENT_IF" || true)" + [[ -z "$SERVER_RDMA" ]] && SERVER_RDMA="$(rdma_for_netdev "$SERVER_IF" || true)" + echo "Using ports: CLIENT_IF=$CLIENT_IF (rdma=${CLIENT_RDMA:-none}) SERVER_IF=$SERVER_IF (rdma=${SERVER_RDMA:-none})" +} + +resolve_ports() { + # For verify()/monitor(), which run AFTER up() with the ports already moved + # into the namespaces: read each ns's non-lo netdev name. If the namespaces + # are absent, fall back to init_net auto-detect. Honours explicit overrides. + if ip netns list 2>/dev/null | grep -qw "$CLIENT_NS"; then + [[ -z "$CLIENT_IF" ]] && CLIENT_IF="$(ip netns exec "$CLIENT_NS" ls /sys/class/net 2>/dev/null | grep -vx lo | head -1)" + [[ -z "$SERVER_IF" ]] && SERVER_IF="$(ip netns exec "$SERVER_NS" ls /sys/class/net 2>/dev/null | grep -vx lo | head -1)" + else + autodetect_ports + fi +} + detect_macs() { # Interfaces must be in the current (init) namespace at this point -- up() # calls down() first, which returns any moved netdevs to init_net. @@ -108,8 +192,10 @@ down() { while read -r dev _; do [[ -n "$dev" ]] && ip netns exec "$NS" rdma dev set "$dev" netns 1 2>/dev/null || true done < <(ip netns exec "$NS" rdma dev show 2>/dev/null | awk -F': ' '/link/ {print $2}' | awk '{print $1}') - # Move netdevs back to init_net. - for IF in "$CLIENT_IF" "$SERVER_IF"; do + # Move every (non-lo) netdev back to init_net. Name-independent so + # teardown can't break on a stale/auto-detected interface name. + for IF in $(ip netns exec "$NS" ls /sys/class/net 2>/dev/null); do + [[ "$IF" == lo ]] && continue ip netns exec "$NS" ip link set "$IF" netns 1 2>/dev/null || true done ip netns delete "$NS" 2>/dev/null || true @@ -123,6 +209,7 @@ up() { echo "== Tearing down any previous state ==" down + autodetect_ports detect_macs echo "== Clearing shared-namespace IPs on the cabled ports ==" @@ -184,6 +271,7 @@ up() { } verify() { + resolve_ports echo "== RDMA device per namespace (each should show exactly its own dev) ==" echo "-- $CLIENT_NS --"; ip netns exec "$CLIENT_NS" rdma dev show || true echo "-- $SERVER_NS --"; ip netns exec "$SERVER_NS" rdma dev show || true @@ -230,6 +318,7 @@ monitor() { # enP2p1s0f0np0), not the rdma device (rocep1s0f0). # Auto-detects whether the ports live in the netns (this script's setup) or # the default namespace (the policy-routed setup_spark_rdma_loopback.sh setup). + resolve_ports local cns sns if ip netns list 2>/dev/null | grep -qw "$CLIENT_NS"; then cns="$CLIENT_NS"; sns="$SERVER_NS"