From 73c8b3e2928ea64b70af8b7039db062d22d1d742 Mon Sep 17 00:00:00 2001 From: Denis Leshchev Date: Mon, 24 Aug 2026 22:23:31 +0000 Subject: [PATCH 1/5] #287 - Batch UDP socket receives Coalesce the datagrams returned by one recvmmsg call into one DAQIRI receive burst before handing it to the application queue. The affinity experiment is intentionally excluded; this change only reduces per-datagram metadata allocation and queue handoff overhead. Signed-off-by: Denis Leshchev --- src/engines/socket/daqiri_socket_engine.cpp | 36 ++++++++++++--------- 1 file changed, 20 insertions(+), 16 deletions(-) diff --git a/src/engines/socket/daqiri_socket_engine.cpp b/src/engines/socket/daqiri_socket_engine.cpp index e8056f0e..ca00202f 100644 --- a/src/engines/socket/daqiri_socket_engine.cpp +++ b/src/engines/socket/daqiri_socket_engine.cpp @@ -1356,27 +1356,31 @@ void SocketEngine::udp_rx_loop(int if_index) { ep->udp_peer_valid = true; } + // One recvmmsg batch becomes one DAQIRI burst. One-packet bursts create a + // queue lock/unlock and metadata allocation for every UDP datagram. + auto* burst = create_tx_burst_params(); + burst->hdr.hdr.port_id = ep->port; + burst->hdr.hdr.q_id = ep->rx_queue; + burst->hdr.hdr.num_pkts = received; + burst->hdr.hdr.num_segs = 1; + burst->pkts[0] = new void*[static_cast(received)]; + burst->pkt_lens[0] = new uint32_t[static_cast(received)]; + + uint64_t received_bytes = 0; for (int i = 0; i < received; ++i) { const auto rx = static_cast(msgs[static_cast(i)].msg_len); - auto* burst = create_tx_burst_params(); - burst->hdr.hdr.port_id = ep->port; - burst->hdr.hdr.q_id = ep->rx_queue; - burst->hdr.hdr.num_pkts = 1; - burst->hdr.hdr.num_segs = 1; - burst->pkts[0] = new void*[1]; - burst->pkt_lens[0] = new uint32_t[1]; - auto* payload = new uint8_t[rx]; std::memcpy(payload, iovs[static_cast(i)].iov_base, rx); - burst->pkts[0][0] = payload; - burst->pkt_lens[0][0] = static_cast(rx); - set_connection_id(burst, ep->primary_conn_id); - - push_rx_burst(ep->rx_queue_state, burst); - rx_pkts_.fetch_add(1); - rx_bytes_.fetch_add(static_cast(rx)); - metrics::add_rx(ep->rx_metrics, 1, static_cast(rx)); + burst->pkts[0][i] = payload; + burst->pkt_lens[0][i] = static_cast(rx); + received_bytes += rx; } + set_connection_id(burst, ep->primary_conn_id); + + push_rx_burst(ep->rx_queue_state, burst); + rx_pkts_.fetch_add(static_cast(received)); + rx_bytes_.fetch_add(received_bytes); + metrics::add_rx(ep->rx_metrics, static_cast(received), received_bytes); } } From ada5f01356a1c62090b973a3659c0d60e843adf1 Mon Sep 17 00:00:00 2001 From: Denis Leshchev Date: Thu, 3 Sep 2026 17:02:25 +0000 Subject: [PATCH 2/5] #287 - Add UDP receive I/O thread affinity Signed-off-by: Denis Leshchev --- docs/api-reference/configuration.md | 3 ++ ...ri_bench_socket_udp_tx_rx_spark_netns.yaml | 2 ++ include/daqiri/types.h | 1 + src/common.cpp | 9 ++++++ src/engines/socket/daqiri_socket_engine.cpp | 28 +++++++++++++++++++ 5 files changed, 43 insertions(+) diff --git a/docs/api-reference/configuration.md b/docs/api-reference/configuration.md index 1a592640..39696d9b 100644 --- a/docs/api-reference/configuration.md +++ b/docs/api-reference/configuration.md @@ -135,6 +135,9 @@ Endpoint addresses are URI strings. Supported schemes are `tcp://`, `udp://`, an `udp://10.250.0.2:5021`. Required for TCP/UDP client mode. RoCE clients choose the peer in application code (for example by calling `rdma_connect_to_server`), not in DAQIRI config. +- **`socket_config.udp_rx_cpu_core`**: Optional CPU index for the UDP receive I/O + thread. The default is `-1` (unpinned); non-negative values are valid only for + `udp://` endpoints. - **`socket_config.local_ip`** / **`socket_config.local_port`** and **`socket_config.remote_ip`** / **`socket_config.remote_port`**: Legacy endpoint fields accepted for older configs when a top-level engine override provides the diff --git a/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml b/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml index a0d19c27..d58b8000 100644 --- a/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml +++ b/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml @@ -47,6 +47,8 @@ daqiri: mode: server local_addr: "udp://10.250.0.2:5001" max_payload_size: 65535 + # -1 leaves the DAQIRI UDP receive I/O thread unpinned. + udp_rx_cpu_core: -1 rx: queues: - name: "Server_RX_Queue" diff --git a/include/daqiri/types.h b/include/daqiri/types.h index 88c0b16d..d906e718 100644 --- a/include/daqiri/types.h +++ b/include/daqiri/types.h @@ -981,6 +981,7 @@ struct SocketConfig { uint64_t max_burst_interval_ms_ = 0; uint32_t min_ipg_ns_ = 0; int32_t retry_connect_s_ = 1; + int32_t udp_rx_cpu_core_ = -1; }; struct RoCEConfig { diff --git a/src/common.cpp b/src/common.cpp index 420fd22c..5735f6f2 100644 --- a/src/common.cpp +++ b/src/common.cpp @@ -1573,6 +1573,15 @@ bool YAML::convert::parse_socket_config( socket_cfg.max_burst_interval_ms_ = socket_item["max_burst_interval_ms"].as(0); socket_cfg.min_ipg_ns_ = socket_item["min_ipg_ns"].as(0); socket_cfg.retry_connect_s_ = socket_item["retry_connect_s"].as(1); + socket_cfg.udp_rx_cpu_core_ = socket_item["udp_rx_cpu_core"].as(-1); + if (socket_cfg.udp_rx_cpu_core_ < -1) { + DAQIRI_LOG_ERROR("socket_config.udp_rx_cpu_core must be -1 or a non-negative CPU index"); + return false; + } + if (socket_cfg.udp_rx_cpu_core_ >= 0 && protocol != daqiri::SocketProtocol::UDP) { + DAQIRI_LOG_ERROR("socket_config.udp_rx_cpu_core is valid only for udp:// endpoints"); + return false; + } const bool roce_client = socket_cfg.mode_ == daqiri::SocketMode::CLIENT && protocol == daqiri::SocketProtocol::ROCE; diff --git a/src/engines/socket/daqiri_socket_engine.cpp b/src/engines/socket/daqiri_socket_engine.cpp index ca00202f..89005515 100644 --- a/src/engines/socket/daqiri_socket_engine.cpp +++ b/src/engines/socket/daqiri_socket_engine.cpp @@ -27,6 +27,8 @@ #include #include #include +#include +#include #include #include #include @@ -72,6 +74,31 @@ std::string sockaddr_to_ip(const sockaddr_in& addr) { return std::string(ip_buf); } +void pin_udp_rx_thread(int cpu_core, uint16_t port) { + if (cpu_core < 0) { return; } + if (cpu_core >= CPU_SETSIZE) { + DAQIRI_LOG_ERROR("UDP RX I/O thread for port {} requested invalid CPU {}; continuing unpinned", + port, + cpu_core); + return; + } + + cpu_set_t cpuset; + CPU_ZERO(&cpuset); + CPU_SET(cpu_core, &cpuset); + const int status = pthread_setaffinity_np(pthread_self(), sizeof(cpuset), &cpuset); + if (status != 0) { + DAQIRI_LOG_ERROR( + "Failed to pin UDP RX I/O thread for port {} to CPU {}: {}; continuing unpinned", + port, + cpu_core, + strerror(status)); + return; + } + + DAQIRI_LOG_INFO("UDP RX I/O thread for port {} pinned to CPU {}", port, cpu_core); +} + } // namespace SocketEngine::~SocketEngine() { @@ -1317,6 +1344,7 @@ void SocketEngine::udp_rx_loop(int if_index) { if (if_index < 0 || if_index >= static_cast(endpoints_.size())) { return; } auto* ep = endpoints_[if_index].get(); if (ep == nullptr) { return; } + pin_udp_rx_thread(ep->socket_cfg.udp_rx_cpu_core_, ep->port); if (ep->udp_fd < 0) { return; } From 589f97c7482bebb5fe0c1558902a15815f6da674 Mon Sep 17 00:00:00 2001 From: Denis Leshchev Date: Fri, 4 Sep 2026 15:36:33 +0000 Subject: [PATCH 3/5] #287 - Address UDP receive batching review findings Signed-off-by: Denis Leshchev --- docs/api-reference/configuration.md | 20 ++- docs/api-reference/cpp.md | 2 +- docs/benchmarks/socket_benchmarking.md | 14 +- examples/daqiri_bench_socket_udp_tx_rx.yaml | 4 +- ...ri_bench_socket_udp_tx_rx_spark_netns.yaml | 8 +- examples/run_spark_bench.sh | 37 +++- examples/socket_bench.cpp | 4 +- include/daqiri/types.h | 1 - scripts/gen_spark_netns_config.py | 34 +++- src/common.cpp | 9 - src/engines/socket/daqiri_socket_engine.cpp | 159 ++++++++++++++---- src/engines/socket/daqiri_socket_engine.h | 12 +- 12 files changed, 229 insertions(+), 75 deletions(-) diff --git a/docs/api-reference/configuration.md b/docs/api-reference/configuration.md index 39696d9b..bfcca4ec 100644 --- a/docs/api-reference/configuration.md +++ b/docs/api-reference/configuration.md @@ -135,9 +135,6 @@ Endpoint addresses are URI strings. Supported schemes are `tcp://`, `udp://`, an `udp://10.250.0.2:5021`. Required for TCP/UDP client mode. RoCE clients choose the peer in application code (for example by calling `rdma_connect_to_server`), not in DAQIRI config. -- **`socket_config.udp_rx_cpu_core`**: Optional CPU index for the UDP receive I/O - thread. The default is `-1` (unpinned); non-negative values are valid only for - `udp://` endpoints. - **`socket_config.local_ip`** / **`socket_config.local_port`** and **`socket_config.remote_ip`** / **`socket_config.remote_port`**: Legacy endpoint fields accepted for older configs when a top-level engine override provides the @@ -148,6 +145,10 @@ after connection setup with `socket_setsockopt(conn_id, level, optname, optval, optlen)`, using the numeric constants from the target system headers. The API is not supported for `roce://` endpoints. +A UDP server endpoint accepts one peer for its lifetime. It learns the source +address of the first datagram and drops datagrams from other source addresses or +ports, keeping subsequent server transmissions bound to that first peer. + When using RoCE, set `stream_type: "socket"` and use `roce://` endpoint addresses plus a `roce_config` block for transport settings. A RoCE URI may include `?engine=ibverbs`; when omitted, `ibverbs` is the default and only supported RoCE @@ -174,12 +175,15 @@ engine. - values: `indirect`, `direct` - default: `indirect` - **`cpu_core`**: CPU core ID for the RX worker thread. Required in indirect mode and forbidden - in direct mode. Should be an isolated core for best performance. + in direct mode. For `udp://` socket endpoints, this pins the thread that calls `recvmmsg()`; + application threads use their own affinity settings. Should be an isolated core for best + performance. Use `-1` to leave a socket UDP receive thread unpinned. - type: `string` -- **`batch_size`**: Number of packets per batch passed from the NIC to the application. Larger - values increase throughput, and smaller values reduce latency. Required in indirect mode and - forbidden in direct mode. A direct poll returns the packets currently ready, up to 256, - without waiting. +- **`batch_size`**: Maximum number of packets per batch passed to the application. Larger values + increase throughput, and smaller values reduce latency. For `udp://` socket endpoints, one + `recvmmsg()` call returns up to this many datagrams; valid values are 1-32. Required in indirect + mode and forbidden in direct mode. A direct poll returns the packets currently ready, up to + 256, without waiting. - type: `integer` - **`memory_regions`**: List of memory region names (defined in [Memory Regions](#memory-regions)). The order determines segment mapping: first region = segment 0, second = segment 1, etc. diff --git a/docs/api-reference/cpp.md b/docs/api-reference/cpp.md index b0d8ec47..a977eb56 100644 --- a/docs/api-reference/cpp.md +++ b/docs/api-reference/cpp.md @@ -80,7 +80,7 @@ int queue_id = 0; auto status = daqiri::get_rx_burst(&burst, port_id, queue_id); ``` -`get_rx_burst()` is non-blocking. It returns `Status::SUCCESS` when a complete batch is +`get_rx_burst()` is non-blocking. It returns `Status::SUCCESS` when a burst is available. When no burst is ready, engines return `Status::NULL_PTR` or `Status::NOT_READY`; applications should handle both as an empty poll. There are also overloads that dequeue from any queue on a port, or from any queue on any port: diff --git a/docs/benchmarks/socket_benchmarking.md b/docs/benchmarks/socket_benchmarking.md index 636f8ab2..be20ab77 100644 --- a/docs/benchmarks/socket_benchmarking.md +++ b/docs/benchmarks/socket_benchmarking.md @@ -158,6 +158,16 @@ The shipped configs run both endpoints on `127.0.0.1` and are useful for a smoke For an on-wire namespace test, use separate server and client YAML files. The important fields are the endpoint URI scheme, namespace IPs, server port, `max_payload_size`, memory-region `buf_size`, and benchmark `message_size`. +For UDP, `rx.queues[].cpu_core` pins the DAQIRI socket I/O thread that drains +`recvmmsg()`. The separate `socket_bench_*.cpu_core` pins the application worker +that consumes the resulting bursts. Assign different CPUs when measuring the +receive path without intentional time-sharing. `rx.queues[].batch_size` controls +the maximum number of datagrams coalesced into one DAQIRI burst (up to 32). +`run_spark_bench.sh` normally preserves its historical self-pacing by assigning +the server I/O and benchmark worker to the same core. Set, for example, +`SOCKET_RX_IO_CORES="15 17 5 7"` to give concurrent UDP pairs dedicated receive +I/O cores while leaving their benchmark-worker placement unchanged. + Applications can tune the underlying TCP/UDP socket after resolving a connection ID with `socket_connect_to_server()` or `socket_get_server_conn_id()`. Use `socket_setsockopt(conn_id, level, optname, optval, optlen)` with the integer @@ -193,7 +203,7 @@ daqiri: - name: "RX_Queue" id: 0 cpu_core: 8 - batch_size: 1 + batch_size: 32 memory_regions: ["DATA_SOCKET_SERVER"] tx: queues: @@ -246,7 +256,7 @@ daqiri: - name: "RX_Queue" id: 0 cpu_core: 8 - batch_size: 1 + batch_size: 32 memory_regions: ["DATA_SOCKET_CLIENT"] tx: queues: diff --git a/examples/daqiri_bench_socket_udp_tx_rx.yaml b/examples/daqiri_bench_socket_udp_tx_rx.yaml index 249adef4..7934de36 100644 --- a/examples/daqiri_bench_socket_udp_tx_rx.yaml +++ b/examples/daqiri_bench_socket_udp_tx_rx.yaml @@ -32,7 +32,7 @@ daqiri: - name: "Server_RX_Queue" id: 0 cpu_core: 8 - batch_size: 1 + batch_size: 32 memory_regions: - "DATA_SOCKET_SERVER" tx: @@ -56,7 +56,7 @@ daqiri: - name: "Client_RX_Queue" id: 0 cpu_core: 7 - batch_size: 1 + batch_size: 32 memory_regions: - "DATA_SOCKET_CLIENT" tx: diff --git a/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml b/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml index d58b8000..73d4d5c2 100644 --- a/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml +++ b/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml @@ -7,7 +7,8 @@ # scripts/gen_spark_netns_config.py splits it to one role at sweep time and # run_spark_bench.sh runs each role in its own namespace, assigning ports/cores per # pair (cores 16-19 across four pairs; the send and receive sides of a pair share -# one core and self-pace, App TX ~= App RX). To split by hand: +# one core and self-pace, App TX ~= App RX). SOCKET_RX_IO_CORES can instead assign +# dedicated server I/O cores without changing the benchmark-worker cores. To split by hand: # scripts/gen_spark_netns_config.py examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml \ # --role server > server.yaml # ip netns exec dq_wire_server daqiri_bench_socket server.yaml --seconds N --mode server @@ -47,14 +48,13 @@ daqiri: mode: server local_addr: "udp://10.250.0.2:5001" max_payload_size: 65535 - # -1 leaves the DAQIRI UDP receive I/O thread unpinned. - udp_rx_cpu_core: -1 rx: queues: - name: "Server_RX_Queue" id: 0 cpu_core: 16 - batch_size: 1 + # The socket engine passes up to one recvmmsg batch per DAQIRI burst. + batch_size: 32 memory_regions: - "DATA_SOCKET_SERVER" tx: diff --git a/examples/run_spark_bench.sh b/examples/run_spark_bench.sh index 25358e3a..badd9c30 100755 --- a/examples/run_spark_bench.sh +++ b/examples/run_spark_bench.sh @@ -40,6 +40,9 @@ # (--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. +# SOCKET_RX_IO_CORES — optional space-separated UDP receive I/O cores, one +# per concurrent pair. These override the server RX queue +# core independently of the socket_bench worker core. # # 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 / @@ -137,6 +140,16 @@ 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 +SOCKET_RX_IO_PIN_CORES=() +if [[ -n "${SOCKET_RX_IO_CORES:-}" ]]; then + read -r -a SOCKET_RX_IO_PIN_CORES <<< "$SOCKET_RX_IO_CORES" + for core in "${SOCKET_RX_IO_PIN_CORES[@]}"; do + if [[ ! "$core" =~ ^-1$|^[0-9]+$ ]]; then + echo "Invalid SOCKET_RX_IO_CORES entry '$core' (expected -1 or a CPU index)" >&2 + exit 1 + fi + done +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 @@ -450,6 +463,14 @@ 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 ))]}"; } +pair_server_io_core() { + local idx="$1" fallback="$2" + if (( ${#SOCKET_RX_IO_PIN_CORES[@]} == 0 )); then + echo "$fallback" + else + echo "${SOCKET_RX_IO_PIN_CORES[$(( idx % ${#SOCKET_RX_IO_PIN_CORES[@]} ))]}" + fi +} # 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), @@ -462,27 +483,31 @@ generate_socket_yaml() { # 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 + local server_core client_core server_io_core if [[ -n "${SOCKET_NOPIN:-}" ]]; then - server_core=-1; client_core=-1 + server_core=-1; client_core=-1; server_io_core=-1 else server_core="$(pair_server_core "$idx")" client_core="$(pair_client_core "$idx")" + server_io_core="$server_core" + if [[ "$BACKEND" == "socket-udp" ]]; then + server_io_core="$(pair_server_io_core "$idx" "$server_core")" + fi fi - python3 "$NETNS_GEN" "$BASE_YAML" --role server | \ + python3 "$NETNS_GEN" "$BASE_YAML" --role server \ + --rx-queue-cpu-core "$server_io_core" --bench-cpu-core "$server_core" | \ 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|^( *server_port: ).*|\1$srv_port|" \ - -e "s|^( *cpu_core: ).*|\1$server_core|" \ > "$server_out" - python3 "$NETNS_GEN" "$BASE_YAML" --role client | \ + python3 "$NETNS_GEN" "$BASE_YAML" --role client \ + --rx-queue-cpu-core "$client_core" --bench-cpu-core "$client_core" | \ 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|^( *server_port: ).*|\1$srv_port|" \ - -e "s|^( *cpu_core: ).*|\1$client_core|" \ > "$client_out" } diff --git a/examples/socket_bench.cpp b/examples/socket_bench.cpp index e5108a21..163afbfa 100644 --- a/examples/socket_bench.cpp +++ b/examples/socket_bench.cpp @@ -210,8 +210,8 @@ void socket_worker(const SocketBenchConfig& cfg, daqiri::bench::TokenBucketPacer if (run_workload) { // Stage each received (host) payload to the GPU; the copy persists in - // the pipeline's device buffer, so a batch can accumulate across bursts - // (UDP bursts are one datagram). When a batch fills, reorder/gather it + // the pipeline's device buffer, so a batch can accumulate across bursts. + // A UDP burst can contain multiple datagrams. When a batch fills, reorder/gather it // and run the compute. TCP (packets_per_batch == 1) flushes per chunk. for (int i = 0; i < num_pkts; ++i) { const auto len = daqiri::get_packet_length(burst, i); diff --git a/include/daqiri/types.h b/include/daqiri/types.h index d906e718..88c0b16d 100644 --- a/include/daqiri/types.h +++ b/include/daqiri/types.h @@ -981,7 +981,6 @@ struct SocketConfig { uint64_t max_burst_interval_ms_ = 0; uint32_t min_ipg_ns_ = 0; int32_t retry_connect_s_ = 1; - int32_t udp_rx_cpu_core_ = -1; }; struct RoCEConfig { diff --git a/scripts/gen_spark_netns_config.py b/scripts/gen_spark_netns_config.py index c00102af..fc8404ca 100755 --- a/scripts/gen_spark_netns_config.py +++ b/scripts/gen_spark_netns_config.py @@ -15,11 +15,13 @@ * interfaces: keep the one whose socket_config.mode == role. * memory regions: keep those whose name contains the role (SERVER / CLIENT). * bench sections: drop the other role's _bench_ mapping. + * optional core overrides: update RX-queue and benchmark-worker affinity + independently without text substitutions that conflate the two. This is the structural inverse of unioning the old _netns_server / _netns_client -files. run_spark_bench.sh pipes the output through its existing per-message-size -awk/sed rewrites (num_bufs/buf_size/depths for RDMA; ports/cores for sockets), -so this script intentionally does NOT touch those fields. Output goes to stdout. +files. run_spark_bench.sh supplies optional structured core overrides, then pipes +the output through per-message-size awk/sed rewrites (num_bufs/buf_size/depths +for RDMA; payload and ports for sockets). Output goes to stdout. """ from __future__ import annotations @@ -30,7 +32,12 @@ import yaml -def split_role(base: dict, role: str) -> dict: +def split_role( + base: dict, + role: str, + rx_queue_cpu_core: int | None = None, + bench_cpu_core: int | None = None, +) -> dict: cfg = base["daqiri"]["cfg"] cfg["interfaces"] = [ i for i in cfg["interfaces"] @@ -38,6 +45,10 @@ def split_role(base: dict, role: str) -> dict: ] if not cfg["interfaces"]: raise SystemExit(f"no interface with socket_config.mode == {role!r}") + if rx_queue_cpu_core is not None: + for interface in cfg["interfaces"]: + for queue in interface.get("rx", {}).get("queues", []): + queue["cpu_core"] = rx_queue_cpu_core cfg["memory_regions"] = [ m for m in cfg["memory_regions"] if role.upper() in m["name"] ] @@ -45,8 +56,12 @@ def split_role(base: dict, role: str) -> dict: other = "client" if role == "server" else "server" for key in [k for k in base if k.endswith(f"_bench_{other}")]: del base[key] - if not any(k.endswith(f"_bench_{role}") for k in base): + role_bench_keys = [k for k in base if k.endswith(f"_bench_{role}")] + if not role_bench_keys: raise SystemExit(f"base has no *_bench_{role} section") + if bench_cpu_core is not None: + for key in role_bench_keys: + base[key]["cpu_core"] = bench_cpu_core return base @@ -54,12 +69,19 @@ def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("base", help="path to the combined (both-role) netns base YAML") ap.add_argument("--role", choices=("server", "client"), required=True) + ap.add_argument("--rx-queue-cpu-core", type=int) + ap.add_argument("--bench-cpu-core", type=int) args = ap.parse_args() with open(args.base, encoding="utf-8") as fh: base = yaml.safe_load(fh) - out = split_role(base, args.role) + out = split_role( + base, + args.role, + rx_queue_cpu_core=args.rx_queue_cpu_core, + bench_cpu_core=args.bench_cpu_core, + ) yaml.safe_dump(out, sys.stdout, sort_keys=False, default_flow_style=False) return 0 diff --git a/src/common.cpp b/src/common.cpp index 5735f6f2..420fd22c 100644 --- a/src/common.cpp +++ b/src/common.cpp @@ -1573,15 +1573,6 @@ bool YAML::convert::parse_socket_config( socket_cfg.max_burst_interval_ms_ = socket_item["max_burst_interval_ms"].as(0); socket_cfg.min_ipg_ns_ = socket_item["min_ipg_ns"].as(0); socket_cfg.retry_connect_s_ = socket_item["retry_connect_s"].as(1); - socket_cfg.udp_rx_cpu_core_ = socket_item["udp_rx_cpu_core"].as(-1); - if (socket_cfg.udp_rx_cpu_core_ < -1) { - DAQIRI_LOG_ERROR("socket_config.udp_rx_cpu_core must be -1 or a non-negative CPU index"); - return false; - } - if (socket_cfg.udp_rx_cpu_core_ >= 0 && protocol != daqiri::SocketProtocol::UDP) { - DAQIRI_LOG_ERROR("socket_config.udp_rx_cpu_core is valid only for udp:// endpoints"); - return false; - } const bool roce_client = socket_cfg.mode_ == daqiri::SocketMode::CLIENT && protocol == daqiri::SocketProtocol::ROCE; diff --git a/src/engines/socket/daqiri_socket_engine.cpp b/src/engines/socket/daqiri_socket_engine.cpp index 89005515..5bc812b1 100644 --- a/src/engines/socket/daqiri_socket_engine.cpp +++ b/src/engines/socket/daqiri_socket_engine.cpp @@ -50,6 +50,7 @@ namespace daqiri { namespace { constexpr size_t kMaxUdpPayloadBytes = 65507; +constexpr uint32_t kMaxUdpRxBatch = 32; bool parse_ipv4_addr(const std::string& ip, uint16_t port, sockaddr_in* addr) { if (addr == nullptr) { return false; } @@ -74,13 +75,18 @@ std::string sockaddr_to_ip(const sockaddr_in& addr) { return std::string(ip_buf); } -void pin_udp_rx_thread(int cpu_core, uint16_t port) { - if (cpu_core < 0) { return; } +bool same_udp_peer(const sockaddr_in& lhs, const sockaddr_in& rhs) { + return lhs.sin_family == rhs.sin_family && lhs.sin_port == rhs.sin_port && + lhs.sin_addr.s_addr == rhs.sin_addr.s_addr; +} + +bool pin_udp_rx_thread(int cpu_core, uint16_t port) { + if (cpu_core < 0) { + return true; + } if (cpu_core >= CPU_SETSIZE) { - DAQIRI_LOG_ERROR("UDP RX I/O thread for port {} requested invalid CPU {}; continuing unpinned", - port, - cpu_core); - return; + DAQIRI_LOG_ERROR("UDP RX I/O thread for port {} requested invalid CPU {}", port, cpu_core); + return false; } cpu_set_t cpuset; @@ -88,15 +94,13 @@ void pin_udp_rx_thread(int cpu_core, uint16_t port) { CPU_SET(cpu_core, &cpuset); const int status = pthread_setaffinity_np(pthread_self(), sizeof(cpuset), &cpuset); if (status != 0) { - DAQIRI_LOG_ERROR( - "Failed to pin UDP RX I/O thread for port {} to CPU {}: {}; continuing unpinned", - port, - cpu_core, - strerror(status)); - return; + DAQIRI_LOG_ERROR("Failed to pin UDP RX I/O thread for port {} to CPU {}: {}", port, cpu_core, + strerror(status)); + return false; } DAQIRI_LOG_INFO("UDP RX I/O thread for port {} pinned to CPU {}", port, cpu_core); + return true; } } // namespace @@ -197,6 +201,10 @@ void SocketEngine::initialize() { ep->socket_cfg = if_cfg.socket_; ep->rx_queue = select_queue_id(if_cfg.rx_.queues_); ep->tx_queue = select_queue_id(if_cfg.tx_.queues_); + if (cfg_.common_.protocol == SocketProtocol::UDP) { + ep->rx_cpu_core = select_cpu_core(if_cfg.rx_.queues_); + ep->rx_batch_size = select_batch_size(if_cfg.rx_.queues_); + } ep->tx_batch_size = select_batch_size(if_cfg.tx_.queues_); ep->max_packet_size = static_cast(std::max(1, select_max_packet_size(if_cfg))); ep->rx_queue_state = get_or_create_rx_queue(ep->port, ep->rx_queue); @@ -625,7 +633,15 @@ Status SocketEngine::pop_rx_burst(const std::shared_ptr& qstate, B } void SocketEngine::push_rx_burst(const std::shared_ptr& qstate, BurstParams* burst) { - if (qstate == nullptr || burst == nullptr) { return; } + if (burst == nullptr) { + return; + } + if (qstate == nullptr) { + DAQIRI_LOG_ERROR("Socket RX burst has no destination queue; releasing it"); + free_all_packets(burst); + free_rx_burst(burst); + return; + } std::lock_guard lock(qstate->mutex); qstate->bursts.push(burst); } @@ -1033,6 +1049,33 @@ uint16_t SocketEngine::select_queue_id(const std::vector& queues) return static_cast(queues.front().common_.id_); } +int SocketEngine::select_cpu_core(const std::vector& queues) const { + if (queues.empty() || queues.front().common_.cpu_core_.empty()) { + return -1; + } + + const auto& value = queues.front().common_.cpu_core_; + size_t parsed_chars = 0; + const int cpu_core = std::stoi(value, &parsed_chars); + if (parsed_chars != value.size() || cpu_core < -1) { + throw std::invalid_argument("invalid socket RX queue cpu_core '" + value + "'"); + } + return cpu_core; +} + +uint32_t SocketEngine::select_batch_size(const std::vector& queues) const { + if (queues.empty()) { + return 1; + } + + const int batch_size = queues.front().common_.batch_size_; + if (batch_size < 1 || batch_size > static_cast(kMaxUdpRxBatch)) { + throw std::invalid_argument("socket UDP RX queue batch_size must be between 1 and " + + std::to_string(kMaxUdpRxBatch)); + } + return static_cast(batch_size); +} + uint32_t SocketEngine::select_batch_size(const std::vector& queues) const { if (queues.empty()) { return 1; } return static_cast(std::max(1, queues.front().common_.batch_size_)); @@ -1245,6 +1288,13 @@ void SocketEngine::setup_udp_endpoint(EndpointState& ep) { } ep.io_thread = std::thread(&SocketEngine::udp_rx_loop, this, ep.if_index); + { + std::unique_lock lock(ep.io_start_mutex); + ep.io_start_cv.wait(lock, [&ep] { return ep.io_start_complete; }); + if (!ep.io_start_success) { + throw std::runtime_error("failed to start UDP RX I/O thread"); + } + } DAQIRI_LOG_INFO("UDP {} on {}:{} (port={})", ep.socket_cfg.mode_ == SocketMode::SERVER ? "server" : "client", @@ -1344,17 +1394,27 @@ void SocketEngine::udp_rx_loop(int if_index) { if (if_index < 0 || if_index >= static_cast(endpoints_.size())) { return; } auto* ep = endpoints_[if_index].get(); if (ep == nullptr) { return; } - pin_udp_rx_thread(ep->socket_cfg.udp_rx_cpu_core_, ep->port); + + const bool affinity_ok = pin_udp_rx_thread(ep->rx_cpu_core, ep->port); + { + std::lock_guard lock(ep->io_start_mutex); + ep->io_start_success = affinity_ok; + ep->io_start_complete = true; + } + ep->io_start_cv.notify_one(); + if (!affinity_ok) { + return; + } if (ep->udp_fd < 0) { return; } - constexpr size_t kUdpRxBatch = 32; - std::vector rx_storage(ep->max_packet_size * kUdpRxBatch); - std::vector msgs(kUdpRxBatch); - std::vector iovs(kUdpRxBatch); - std::vector peers(kUdpRxBatch); + const size_t rx_batch_size = ep->rx_batch_size; + std::vector rx_storage(ep->max_packet_size * rx_batch_size); + std::vector msgs(rx_batch_size); + std::vector iovs(rx_batch_size); + std::vector peers(rx_batch_size); - for (size_t i = 0; i < kUdpRxBatch; ++i) { + for (size_t i = 0; i < rx_batch_size; ++i) { iovs[i].iov_base = rx_storage.data() + (i * ep->max_packet_size); iovs[i].iov_len = ep->max_packet_size; std::memset(&msgs[i], 0, sizeof(mmsghdr)); @@ -1364,12 +1424,15 @@ void SocketEngine::udp_rx_loop(int if_index) { msgs[i].msg_hdr.msg_namelen = sizeof(sockaddr_in); } + std::vector accepted_indices; + accepted_indices.reserve(rx_batch_size); while (running_.load()) { - for (size_t i = 0; i < kUdpRxBatch; ++i) { + for (size_t i = 0; i < rx_batch_size; ++i) { msgs[i].msg_hdr.msg_namelen = sizeof(sockaddr_in); } - const int received = ::recvmmsg(ep->udp_fd, msgs.data(), static_cast(kUdpRxBatch), 0, nullptr); + const int received = ::recvmmsg( + ep->udp_fd, msgs.data(), static_cast(rx_batch_size), MSG_WAITFORONE, nullptr); if (received < 0) { if (errno == EINTR) { continue; } if (!running_.load()) { break; } @@ -1378,10 +1441,39 @@ void SocketEngine::udp_rx_loop(int if_index) { } if (received == 0) { continue; } + accepted_indices.clear(); + uint64_t peer_mismatch_drops = 0; if (ep->socket_cfg.mode_ == SocketMode::SERVER) { - std::lock_guard lock(state_mutex_); - ep->udp_peer_addr = peers[static_cast(received - 1)]; - ep->udp_peer_valid = true; + { + std::lock_guard lock(state_mutex_); + if (!ep->udp_peer_valid) { + ep->udp_peer_addr = peers[0]; + ep->udp_peer_valid = true; + } + for (int i = 0; i < received; ++i) { + if (same_udp_peer(ep->udp_peer_addr, peers[static_cast(i)])) { + accepted_indices.push_back(static_cast(i)); + } else { + ++peer_mismatch_drops; + } + } + } + if (peer_mismatch_drops > 0) { + metrics::add_dropped(ep->rx_metrics, "peer_mismatch", peer_mismatch_drops); + if (!ep->udp_peer_mismatch_warned) { + DAQIRI_LOG_WARN( + "UDP server on port {} accepts one peer; dropping datagrams from other senders", + ep->port); + ep->udp_peer_mismatch_warned = true; + } + } + } else { + for (int i = 0; i < received; ++i) { + accepted_indices.push_back(static_cast(i)); + } + } + if (accepted_indices.empty()) { + continue; } // One recvmmsg batch becomes one DAQIRI burst. One-packet bursts create a @@ -1389,16 +1481,17 @@ void SocketEngine::udp_rx_loop(int if_index) { auto* burst = create_tx_burst_params(); burst->hdr.hdr.port_id = ep->port; burst->hdr.hdr.q_id = ep->rx_queue; - burst->hdr.hdr.num_pkts = received; + burst->hdr.hdr.num_pkts = accepted_indices.size(); burst->hdr.hdr.num_segs = 1; - burst->pkts[0] = new void*[static_cast(received)]; - burst->pkt_lens[0] = new uint32_t[static_cast(received)]; + burst->pkts[0] = new void*[accepted_indices.size()]; + burst->pkt_lens[0] = new uint32_t[accepted_indices.size()]; uint64_t received_bytes = 0; - for (int i = 0; i < received; ++i) { - const auto rx = static_cast(msgs[static_cast(i)].msg_len); + for (size_t i = 0; i < accepted_indices.size(); ++i) { + const size_t source_index = accepted_indices[i]; + const auto rx = static_cast(msgs[source_index].msg_len); auto* payload = new uint8_t[rx]; - std::memcpy(payload, iovs[static_cast(i)].iov_base, rx); + std::memcpy(payload, iovs[source_index].iov_base, rx); burst->pkts[0][i] = payload; burst->pkt_lens[0][i] = static_cast(rx); received_bytes += rx; @@ -1406,9 +1499,9 @@ void SocketEngine::udp_rx_loop(int if_index) { set_connection_id(burst, ep->primary_conn_id); push_rx_burst(ep->rx_queue_state, burst); - rx_pkts_.fetch_add(static_cast(received)); + rx_pkts_.fetch_add(static_cast(accepted_indices.size())); rx_bytes_.fetch_add(received_bytes); - metrics::add_rx(ep->rx_metrics, static_cast(received), received_bytes); + metrics::add_rx(ep->rx_metrics, static_cast(accepted_indices.size()), received_bytes); } } diff --git a/src/engines/socket/daqiri_socket_engine.h b/src/engines/socket/daqiri_socket_engine.h index 2dd5d3dc..6ea94744 100644 --- a/src/engines/socket/daqiri_socket_engine.h +++ b/src/engines/socket/daqiri_socket_engine.h @@ -18,14 +18,15 @@ #pragma once #include +#include #include #include +#include #include #include #include #include #include -#include #include #include "src/engine.h" @@ -139,6 +140,8 @@ class SocketEngine : public Engine { std::string address; uint16_t tx_queue = 0; uint16_t rx_queue = 0; + int rx_cpu_core = -1; + uint32_t rx_batch_size = 1; uint32_t tx_batch_size = 1; size_t max_packet_size = 65536; SocketConfig socket_cfg; @@ -147,11 +150,16 @@ class SocketEngine : public Engine { std::atomic accept_running{false}; std::thread accept_thread; std::thread io_thread; + std::mutex io_start_mutex; + std::condition_variable io_start_cv; + bool io_start_complete = false; + bool io_start_success = false; std::shared_ptr rx_queue_state; std::shared_ptr rx_metrics; std::shared_ptr tx_metrics; sockaddr_in udp_peer_addr{}; bool udp_peer_valid = false; + bool udp_peer_mismatch_warned = false; uintptr_t primary_conn_id = 0; }; @@ -188,6 +196,8 @@ class SocketEngine : public Engine { int select_max_packet_size(const InterfaceConfig& if_cfg) const; uint16_t select_queue_id(const std::vector& queues) const; uint16_t select_queue_id(const std::vector& queues) const; + int select_cpu_core(const std::vector& queues) const; + uint32_t select_batch_size(const std::vector& queues) const; uint32_t select_batch_size(const std::vector& queues) const; Status pop_rx_burst(const std::shared_ptr& qstate, BurstParams** burst); From 2063b17d6b3bd65587b232fc53a79014b5ab1858 Mon Sep 17 00:00:00 2001 From: Denis Leshchev Date: Fri, 4 Sep 2026 16:51:30 +0000 Subject: [PATCH 4/5] #287 - Prevent UDP peer lockout during batching Signed-off-by: Denis Leshchev --- docs/api-reference/configuration.md | 18 +-- docs/benchmarks/socket_benchmarking.md | 4 + examples/daqiri_bench_socket_udp_tx_rx.yaml | 1 + ...ri_bench_socket_udp_tx_rx_spark_netns.yaml | 1 + examples/run_spark_bench.sh | 7 +- scripts/gen_spark_netns_config.py | 11 +- src/engines/socket/daqiri_socket_engine.cpp | 103 ++++++++---------- src/engines/socket/daqiri_socket_engine.h | 2 +- 8 files changed, 80 insertions(+), 67 deletions(-) diff --git a/docs/api-reference/configuration.md b/docs/api-reference/configuration.md index bfcca4ec..063578a5 100644 --- a/docs/api-reference/configuration.md +++ b/docs/api-reference/configuration.md @@ -134,7 +134,9 @@ Endpoint addresses are URI strings. Supported schemes are `tcp://`, `udp://`, an - **`socket_config.remote_addr`**: Remote peer endpoint, for example `udp://10.250.0.2:5021`. Required for TCP/UDP client mode. RoCE clients choose the peer in application code (for example by calling `rdma_connect_to_server`), - not in DAQIRI config. + not in DAQIRI config. It is optional for UDP server mode; when present, DAQIRI + connects the socket to that expected peer so other sources are rejected by the + kernel and multi-datagram receive batching is safe. - **`socket_config.local_ip`** / **`socket_config.local_port`** and **`socket_config.remote_ip`** / **`socket_config.remote_port`**: Legacy endpoint fields accepted for older configs when a top-level engine override provides the @@ -145,9 +147,10 @@ after connection setup with `socket_setsockopt(conn_id, level, optname, optval, optlen)`, using the numeric constants from the target system headers. The API is not supported for `roce://` endpoints. -A UDP server endpoint accepts one peer for its lifetime. It learns the source -address of the first datagram and drops datagrams from other source addresses or -ports, keeping subsequent server transmissions bound to that first peer. +For compatibility, a UDP server without `remote_addr` learns the sender of each +received datagram and uses it for subsequent server transmissions. DAQIRI limits +such endpoints to one datagram per receive burst. Configure `remote_addr` for +point-to-point operation and receive batching. When using RoCE, set `stream_type: "socket"` and use `roce://` endpoint addresses plus a `roce_config` block for transport settings. A RoCE URI may include @@ -181,9 +184,10 @@ engine. - type: `string` - **`batch_size`**: Maximum number of packets per batch passed to the application. Larger values increase throughput, and smaller values reduce latency. For `udp://` socket endpoints, one - `recvmmsg()` call returns up to this many datagrams; valid values are 1-32. Required in indirect - mode and forbidden in direct mode. A direct poll returns the packets currently ready, up to - 256, without waiting. + `recvmmsg()` call returns up to this many datagrams; valid values are 1-32. UDP servers without + a configured `remote_addr` are limited to one datagram per burst. Required in indirect mode and + forbidden in direct mode. A direct poll returns the packets currently ready, up to 256, without + waiting. - type: `integer` - **`memory_regions`**: List of memory region names (defined in [Memory Regions](#memory-regions)). The order determines segment mapping: first region = segment 0, second = segment 1, etc. diff --git a/docs/benchmarks/socket_benchmarking.md b/docs/benchmarks/socket_benchmarking.md index be20ab77..3c73e07d 100644 --- a/docs/benchmarks/socket_benchmarking.md +++ b/docs/benchmarks/socket_benchmarking.md @@ -163,6 +163,9 @@ For UDP, `rx.queues[].cpu_core` pins the DAQIRI socket I/O thread that drains that consumes the resulting bursts. Assign different CPUs when measuring the receive path without intentional time-sharing. `rx.queues[].batch_size` controls the maximum number of datagrams coalesced into one DAQIRI burst (up to 32). +Set `socket_config.remote_addr` on a UDP server to identify its expected client; +this lets the kernel reject other senders and permits receive batches larger than +one datagram. `run_spark_bench.sh` normally preserves its historical self-pacing by assigning the server I/O and benchmark worker to the same core. Set, for example, `SOCKET_RX_IO_CORES="15 17 5 7"` to give concurrent UDP pairs dedicated receive @@ -197,6 +200,7 @@ daqiri: socket_config: mode: server local_addr: "udp://10.250.0.2:5021" + remote_addr: "udp://10.250.0.1:5121" max_payload_size: 65535 rx: queues: diff --git a/examples/daqiri_bench_socket_udp_tx_rx.yaml b/examples/daqiri_bench_socket_udp_tx_rx.yaml index 7934de36..7a2cbb75 100644 --- a/examples/daqiri_bench_socket_udp_tx_rx.yaml +++ b/examples/daqiri_bench_socket_udp_tx_rx.yaml @@ -26,6 +26,7 @@ daqiri: socket_config: mode: server local_addr: "udp://127.0.0.1:5001" + remote_addr: "udp://127.0.0.1:5002" max_payload_size: 2048 rx: queues: diff --git a/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml b/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml index 73d4d5c2..a6c99292 100644 --- a/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml +++ b/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml @@ -47,6 +47,7 @@ daqiri: socket_config: mode: server local_addr: "udp://10.250.0.2:5001" + remote_addr: "udp://10.250.0.1:5002" max_payload_size: 65535 rx: queues: diff --git a/examples/run_spark_bench.sh b/examples/run_spark_bench.sh index badd9c30..04a8329e 100755 --- a/examples/run_spark_bench.sh +++ b/examples/run_spark_bench.sh @@ -495,14 +495,17 @@ generate_socket_yaml() { fi fi python3 "$NETNS_GEN" "$BASE_YAML" --role server \ - --rx-queue-cpu-core "$server_io_core" --bench-cpu-core "$server_core" | \ + --rx-queue-cpu-core "$server_io_core" --tx-queue-cpu-core "$server_core" \ + --bench-cpu-core "$server_core" | \ 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|^( *remote_addr: \"?[a-z]+://[0-9.]+:)[0-9]+(\"?)|\1$cli_port\2|" \ -e "s|^( *server_port: ).*|\1$srv_port|" \ > "$server_out" python3 "$NETNS_GEN" "$BASE_YAML" --role client \ - --rx-queue-cpu-core "$client_core" --bench-cpu-core "$client_core" | \ + --rx-queue-cpu-core "$client_core" --tx-queue-cpu-core "$client_core" \ + --bench-cpu-core "$client_core" | \ sed -E \ -e "s|^( *message_size: ).*|\1$payload|g" \ -e "s|^( *local_addr: \"?[a-z]+://[0-9.]+:)[0-9]+(\"?)|\1$cli_port\2|" \ diff --git a/scripts/gen_spark_netns_config.py b/scripts/gen_spark_netns_config.py index fc8404ca..dcb80119 100755 --- a/scripts/gen_spark_netns_config.py +++ b/scripts/gen_spark_netns_config.py @@ -15,8 +15,8 @@ * interfaces: keep the one whose socket_config.mode == role. * memory regions: keep those whose name contains the role (SERVER / CLIENT). * bench sections: drop the other role's _bench_ mapping. - * optional core overrides: update RX-queue and benchmark-worker affinity - independently without text substitutions that conflate the two. + * optional core overrides: update RX queue, TX queue, and benchmark-worker + affinity independently without text substitutions that conflate them. This is the structural inverse of unioning the old _netns_server / _netns_client files. run_spark_bench.sh supplies optional structured core overrides, then pipes @@ -36,6 +36,7 @@ def split_role( base: dict, role: str, rx_queue_cpu_core: int | None = None, + tx_queue_cpu_core: int | None = None, bench_cpu_core: int | None = None, ) -> dict: cfg = base["daqiri"]["cfg"] @@ -49,6 +50,10 @@ def split_role( for interface in cfg["interfaces"]: for queue in interface.get("rx", {}).get("queues", []): queue["cpu_core"] = rx_queue_cpu_core + if tx_queue_cpu_core is not None: + for interface in cfg["interfaces"]: + for queue in interface.get("tx", {}).get("queues", []): + queue["cpu_core"] = tx_queue_cpu_core cfg["memory_regions"] = [ m for m in cfg["memory_regions"] if role.upper() in m["name"] ] @@ -70,6 +75,7 @@ def main() -> int: ap.add_argument("base", help="path to the combined (both-role) netns base YAML") ap.add_argument("--role", choices=("server", "client"), required=True) ap.add_argument("--rx-queue-cpu-core", type=int) + ap.add_argument("--tx-queue-cpu-core", type=int) ap.add_argument("--bench-cpu-core", type=int) args = ap.parse_args() @@ -80,6 +86,7 @@ def main() -> int: base, args.role, rx_queue_cpu_core=args.rx_queue_cpu_core, + tx_queue_cpu_core=args.tx_queue_cpu_core, bench_cpu_core=args.bench_cpu_core, ) yaml.safe_dump(out, sys.stdout, sort_keys=False, default_flow_style=False) diff --git a/src/engines/socket/daqiri_socket_engine.cpp b/src/engines/socket/daqiri_socket_engine.cpp index 5bc812b1..2c74a40d 100644 --- a/src/engines/socket/daqiri_socket_engine.cpp +++ b/src/engines/socket/daqiri_socket_engine.cpp @@ -75,11 +75,6 @@ std::string sockaddr_to_ip(const sockaddr_in& addr) { return std::string(ip_buf); } -bool same_udp_peer(const sockaddr_in& lhs, const sockaddr_in& rhs) { - return lhs.sin_family == rhs.sin_family && lhs.sin_port == rhs.sin_port && - lhs.sin_addr.s_addr == rhs.sin_addr.s_addr; -} - bool pin_udp_rx_thread(int cpu_core, uint16_t port) { if (cpu_core < 0) { return true; @@ -204,6 +199,13 @@ void SocketEngine::initialize() { if (cfg_.common_.protocol == SocketProtocol::UDP) { ep->rx_cpu_core = select_cpu_core(if_cfg.rx_.queues_); ep->rx_batch_size = select_batch_size(if_cfg.rx_.queues_); + if (ep->socket_cfg.mode_ == SocketMode::SERVER && ep->socket_cfg.remote_ip_.empty() && + ep->rx_batch_size > 1) { + DAQIRI_LOG_WARN( + "UDP server '{}' has no configured remote_addr; limiting RX batch_size to 1", + if_cfg.name_); + ep->rx_batch_size = 1; + } } ep->tx_batch_size = select_batch_size(if_cfg.tx_.queues_); ep->max_packet_size = static_cast(std::max(1, select_max_packet_size(if_cfg))); @@ -840,7 +842,7 @@ bool SocketEngine::send_udp_burst(EndpointState& ep, BurstParams* burst, size_t* return false; } peer = ep.udp_peer_addr; - use_sendto = true; + use_sendto = !ep.udp_peer_configured; } for (size_t i = 0; i < num_pkts; ++i) { @@ -1243,6 +1245,23 @@ void SocketEngine::setup_udp_endpoint(EndpointState& ep) { close_fd(fd); throw std::runtime_error("failed to bind UDP server socket: " + err); } + + if (!ep.socket_cfg.remote_ip_.empty() || ep.socket_cfg.remote_port_ != 0) { + sockaddr_in peer_addr{}; + if (ep.socket_cfg.remote_ip_.empty() || ep.socket_cfg.remote_port_ == 0 || + !parse_ipv4_addr(ep.socket_cfg.remote_ip_, ep.socket_cfg.remote_port_, &peer_addr)) { + close_fd(fd); + throw std::runtime_error("invalid UDP server remote address"); + } + if (::connect(fd, reinterpret_cast(&peer_addr), sizeof(peer_addr)) != 0) { + const auto err = std::string(strerror(errno)); + close_fd(fd); + throw std::runtime_error("failed to connect UDP server socket to configured peer: " + err); + } + ep.udp_peer_addr = peer_addr; + ep.udp_peer_valid = true; + ep.udp_peer_configured = true; + } } else { if (!ep.socket_cfg.local_ip_.empty() || ep.socket_cfg.local_port_ != 0) { const std::string bind_ip = ep.socket_cfg.local_ip_.empty() ? std::string("0.0.0.0") : ep.socket_cfg.local_ip_; @@ -1395,19 +1414,22 @@ void SocketEngine::udp_rx_loop(int if_index) { auto* ep = endpoints_[if_index].get(); if (ep == nullptr) { return; } - const bool affinity_ok = pin_udp_rx_thread(ep->rx_cpu_core, ep->port); + bool startup_ok = ep->udp_fd >= 0; + if (!startup_ok) { + DAQIRI_LOG_ERROR("UDP RX I/O thread for port {} has no socket", ep->port); + } else { + startup_ok = pin_udp_rx_thread(ep->rx_cpu_core, ep->port); + } { std::lock_guard lock(ep->io_start_mutex); - ep->io_start_success = affinity_ok; + ep->io_start_success = startup_ok; ep->io_start_complete = true; } ep->io_start_cv.notify_one(); - if (!affinity_ok) { + if (!startup_ok) { return; } - if (ep->udp_fd < 0) { return; } - const size_t rx_batch_size = ep->rx_batch_size; std::vector rx_storage(ep->max_packet_size * rx_batch_size); std::vector msgs(rx_batch_size); @@ -1424,8 +1446,6 @@ void SocketEngine::udp_rx_loop(int if_index) { msgs[i].msg_hdr.msg_namelen = sizeof(sockaddr_in); } - std::vector accepted_indices; - accepted_indices.reserve(rx_batch_size); while (running_.load()) { for (size_t i = 0; i < rx_batch_size; ++i) { msgs[i].msg_hdr.msg_namelen = sizeof(sockaddr_in); @@ -1439,41 +1459,15 @@ void SocketEngine::udp_rx_loop(int if_index) { DAQIRI_LOG_WARN("UDP recvmmsg failed on port {}: {}", ep->port, strerror(errno)); continue; } + if (!running_.load()) { + break; + } if (received == 0) { continue; } - accepted_indices.clear(); - uint64_t peer_mismatch_drops = 0; - if (ep->socket_cfg.mode_ == SocketMode::SERVER) { - { - std::lock_guard lock(state_mutex_); - if (!ep->udp_peer_valid) { - ep->udp_peer_addr = peers[0]; - ep->udp_peer_valid = true; - } - for (int i = 0; i < received; ++i) { - if (same_udp_peer(ep->udp_peer_addr, peers[static_cast(i)])) { - accepted_indices.push_back(static_cast(i)); - } else { - ++peer_mismatch_drops; - } - } - } - if (peer_mismatch_drops > 0) { - metrics::add_dropped(ep->rx_metrics, "peer_mismatch", peer_mismatch_drops); - if (!ep->udp_peer_mismatch_warned) { - DAQIRI_LOG_WARN( - "UDP server on port {} accepts one peer; dropping datagrams from other senders", - ep->port); - ep->udp_peer_mismatch_warned = true; - } - } - } else { - for (int i = 0; i < received; ++i) { - accepted_indices.push_back(static_cast(i)); - } - } - if (accepted_indices.empty()) { - continue; + if (ep->socket_cfg.mode_ == SocketMode::SERVER && !ep->udp_peer_configured) { + std::lock_guard lock(state_mutex_); + ep->udp_peer_addr = peers[static_cast(received - 1)]; + ep->udp_peer_valid = true; } // One recvmmsg batch becomes one DAQIRI burst. One-packet bursts create a @@ -1481,17 +1475,16 @@ void SocketEngine::udp_rx_loop(int if_index) { auto* burst = create_tx_burst_params(); burst->hdr.hdr.port_id = ep->port; burst->hdr.hdr.q_id = ep->rx_queue; - burst->hdr.hdr.num_pkts = accepted_indices.size(); + burst->hdr.hdr.num_pkts = static_cast(received); burst->hdr.hdr.num_segs = 1; - burst->pkts[0] = new void*[accepted_indices.size()]; - burst->pkt_lens[0] = new uint32_t[accepted_indices.size()]; + burst->pkts[0] = new void*[static_cast(received)]; + burst->pkt_lens[0] = new uint32_t[static_cast(received)]; uint64_t received_bytes = 0; - for (size_t i = 0; i < accepted_indices.size(); ++i) { - const size_t source_index = accepted_indices[i]; - const auto rx = static_cast(msgs[source_index].msg_len); + for (int i = 0; i < received; ++i) { + const auto rx = static_cast(msgs[static_cast(i)].msg_len); auto* payload = new uint8_t[rx]; - std::memcpy(payload, iovs[source_index].iov_base, rx); + std::memcpy(payload, iovs[static_cast(i)].iov_base, rx); burst->pkts[0][i] = payload; burst->pkt_lens[0][i] = static_cast(rx); received_bytes += rx; @@ -1499,9 +1492,9 @@ void SocketEngine::udp_rx_loop(int if_index) { set_connection_id(burst, ep->primary_conn_id); push_rx_burst(ep->rx_queue_state, burst); - rx_pkts_.fetch_add(static_cast(accepted_indices.size())); + rx_pkts_.fetch_add(static_cast(received)); rx_bytes_.fetch_add(received_bytes); - metrics::add_rx(ep->rx_metrics, static_cast(accepted_indices.size()), received_bytes); + metrics::add_rx(ep->rx_metrics, static_cast(received), received_bytes); } } diff --git a/src/engines/socket/daqiri_socket_engine.h b/src/engines/socket/daqiri_socket_engine.h index 6ea94744..129ca548 100644 --- a/src/engines/socket/daqiri_socket_engine.h +++ b/src/engines/socket/daqiri_socket_engine.h @@ -159,7 +159,7 @@ class SocketEngine : public Engine { std::shared_ptr tx_metrics; sockaddr_in udp_peer_addr{}; bool udp_peer_valid = false; - bool udp_peer_mismatch_warned = false; + bool udp_peer_configured = false; uintptr_t primary_conn_id = 0; }; From 6a2770523bf53b975a30f3277d2292f07cfcc462 Mon Sep 17 00:00:00 2001 From: Denis Leshchev Date: Fri, 4 Sep 2026 20:42:29 +0000 Subject: [PATCH 5/5] #287 - Correct UDP benchmark metadata Signed-off-by: Denis Leshchev --- docs/api-reference/configuration.md | 12 +- docs/benchmarks/performance-dgx-spark.md | 6 +- docs/benchmarks/socket_benchmarking.md | 17 ++- ...ri_bench_socket_udp_tx_rx_spark_netns.yaml | 12 +- examples/run_spark_bench.sh | 114 +++++++++++++++--- examples/socket_bench.cpp | 50 +++++--- scripts/gen_spark_netns_config.py | 18 ++- 7 files changed, 169 insertions(+), 60 deletions(-) diff --git a/docs/api-reference/configuration.md b/docs/api-reference/configuration.md index 063578a5..282b0784 100644 --- a/docs/api-reference/configuration.md +++ b/docs/api-reference/configuration.md @@ -147,10 +147,12 @@ after connection setup with `socket_setsockopt(conn_id, level, optname, optval, optlen)`, using the numeric constants from the target system headers. The API is not supported for `roce://` endpoints. -For compatibility, a UDP server without `remote_addr` learns the sender of each -received datagram and uses it for subsequent server transmissions. DAQIRI limits -such endpoints to one datagram per receive burst. Configure `remote_addr` for -point-to-point operation and receive batching. +For compatibility, a UDP server without `remote_addr` operates in a +single-active-peer mode: each received datagram replaces the endpoint's current +reply target. This does not preserve request/reply association when multiple +clients interleave traffic. DAQIRI limits such endpoints to one datagram per +receive burst. Configure `remote_addr` for point-to-point operation, peer +filtering, and receive batching. When using RoCE, set `stream_type: "socket"` and use `roce://` endpoint addresses plus a `roce_config` block for transport settings. A RoCE URI may include @@ -189,6 +191,8 @@ engine. forbidden in direct mode. A direct poll returns the packets currently ready, up to 256, without waiting. - type: `integer` + - C++ or Python callers constructing an indirect UDP RX queue programmatically must set this + field explicitly. The default `CommonQueueConfig` value of `0` is not a valid UDP batch size. - **`memory_regions`**: List of memory region names (defined in [Memory Regions](#memory-regions)). The order determines segment mapping: first region = segment 0, second = segment 1, etc. A single region means all packet data lands in one place, while two regions enables header-data diff --git a/docs/benchmarks/performance-dgx-spark.md b/docs/benchmarks/performance-dgx-spark.md index bb1aeb30..0662e6c4 100644 --- a/docs/benchmarks/performance-dgx-spark.md +++ b/docs/benchmarks/performance-dgx-spark.md @@ -468,7 +468,11 @@ client/server pairs (sockets). The workload lands in the CSV `post_process` colu (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 +Each run writes `bench-results/--/runs.csv`. The CSV +records the configured `batch`; for socket runs, `observed_max_rx_burst` reports +the largest burst returned to the application. Its CPU core columns identify the +actual sampled cores; socket runs with multiple pairs report pair 0 rather than +aggregate CPU utilization. See [Socket and RDMA Benchmarking](socket_benchmarking.md) and [Raw Ethernet Benchmarking](raw_benchmarking.md) for the namespace setup and per-transport details. diff --git a/docs/benchmarks/socket_benchmarking.md b/docs/benchmarks/socket_benchmarking.md index 3c73e07d..18f101bc 100644 --- a/docs/benchmarks/socket_benchmarking.md +++ b/docs/benchmarks/socket_benchmarking.md @@ -166,10 +166,19 @@ the maximum number of datagrams coalesced into one DAQIRI burst (up to 32). Set `socket_config.remote_addr` on a UDP server to identify its expected client; this lets the kernel reject other senders and permits receive batches larger than one datagram. -`run_spark_bench.sh` normally preserves its historical self-pacing by assigning -the server I/O and benchmark worker to the same core. Set, for example, -`SOCKET_RX_IO_CORES="15 17 5 7"` to give concurrent UDP pairs dedicated receive -I/O cores while leaving their benchmark-worker placement unchanged. +`run_spark_bench.sh` normally preserves its historical server-side placement by +assigning the server I/O and benchmark worker to the same core. To measure a +fully separated pair on DGX Spark, select pair 0 and the spare core 15: + +```bash +PAIRS_OVERRIDE=1 SOCKET_RX_IO_CORES=15 \ + ./examples/run_spark_bench.sh socket-udp smoke +``` + +That places the master on 8, server worker on 16, client worker on 17, and UDP +I/O on 15. The fixed four-pair map consumes the other big cores, so a four-pair +run cannot give every I/O thread a dedicated core without changing the worker +map or allowing deliberate overlap. Applications can tune the underlying TCP/UDP socket after resolving a connection ID with `socket_connect_to_server()` or `socket_get_server_conn_id()`. Use diff --git a/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml b/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml index a6c99292..7cfbeecd 100644 --- a/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml +++ b/examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml @@ -6,9 +6,9 @@ # it as-is across namespaces (it would init the peer namespace's IP), so # scripts/gen_spark_netns_config.py splits it to one role at sweep time and # run_spark_bench.sh runs each role in its own namespace, assigning ports/cores per -# pair (cores 16-19 across four pairs; the send and receive sides of a pair share -# one core and self-pace, App TX ~= App RX). SOCKET_RX_IO_CORES can instead assign -# dedicated server I/O cores without changing the benchmark-worker cores. To split by hand: +# pair. Server workers use [16, 18, 5, 7], client workers use [17, 19, 6, 9], and +# the server UDP I/O thread shares its server-worker core by default. +# SOCKET_RX_IO_CORES overrides the server I/O placement independently. To split by hand: # scripts/gen_spark_netns_config.py examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml \ # --role server > server.yaml # ip netns exec dq_wire_server daqiri_bench_socket server.yaml --seconds N --mode server @@ -91,9 +91,9 @@ daqiri: - "DATA_SOCKET_CLIENT" socket_bench_server: - # Bench worker thread affinity (PR #149). run_spark_bench.sh rewrites this to the - # pair core, so the server worker, client worker, and both queues share one core - # per pair -- the deliberate self-pacing setup (App TX ~= App RX). + # Bench worker thread affinity (PR #149). run_spark_bench.sh rewrites this to + # the pair's server-worker core, separately from the client worker and optional + # UDP receive I/O core. cpu_core: 16 server: true send: false diff --git a/examples/run_spark_bench.sh b/examples/run_spark_bench.sh index 04a8329e..546b29f7 100755 --- a/examples/run_spark_bench.sh +++ b/examples/run_spark_bench.sh @@ -43,6 +43,9 @@ # SOCKET_RX_IO_CORES — optional space-separated UDP receive I/O cores, one # per concurrent pair. These override the server RX queue # core independently of the socket_bench worker core. +# BATCHES_OVERRIDE — optional space-separated batch sizes for one-off runs. +# PAIRS_OVERRIDE — optional space-separated socket pair counts. For example, +# PAIRS_OVERRIDE=1 selects the pair-0 CPU placement only. # # 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 / @@ -83,8 +86,15 @@ CSV="$OUT_DIR/runs.csv" # 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_gemm_dim = GEMM_DIM pinned dimension (default 1024). +# CPU core/percentage columns identify the actual sampled cores. For a multi-pair +# socket run, TX and RX are pair-0 samples rather than aggregate utilization. # 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" +CSV_HEADER="lang,backend,post_process,payload,batch,observed_max_rx_burst,pairs" +CSV_HEADER+=",target_gbps,rep,seconds,packets,bytes,pps,gbps,rx_gbps,drops,drops_kind" +CSV_HEADER+=",cpu_master_core,cpu_tx_core,cpu_rx_core" +CSV_HEADER+=",cpu_master_pct,cpu_tx_pct,cpu_rx_pct,gpu_sm_pct,gpu_mem_pct" +CSV_HEADER+=",post_process_gemm_dim,post_process_sync" +echo "$CSV_HEADER" > "$CSV" # Capture slow-moving environment state once per result set. "$SCRIPT_DIR/bench_capture_environment.sh" "$OUT_DIR" @@ -239,11 +249,11 @@ case "$BACKEND" in # carry a large buf_size so message_size never overflows the TX buffer. socket-udp) PAYLOADS_SWEEP=(8000 1000) - BATCHES_SWEEP=(1) + BATCHES_SWEEP=(32) PAYLOADS_HEADLINE=(8000) - BATCHES_HEADLINE=(1) - # Concurrent client/server pairs. A single pair is core-bound well below line rate; - # the published matrix reaches ~12 Gb/s aggregate by running four pairs. + BATCHES_HEADLINE=(32) + # Concurrent client/server pairs. A single pair is core-bound well below line + # rate; the published matrix scales aggregate throughput with four pairs. PAIRS_SWEEP=(1 2 4) PAIRS_HEADLINE=(4) SRV_PORT_BASE=5001; CLI_PORT_BASE=5101 @@ -253,9 +263,8 @@ case "$BACKEND" in # cutting same-host IPs through the loopback (lo) device. BASE_YAML="$SCRIPT_DIR/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml" BENCH_BIN="$BUILD_DIR/examples/daqiri_bench_socket" - # Pair 0 pins to core 16 (see pair_core); report that core's busy% as the per-pair - # bottleneck. cpu_tx_pct and cpu_rx_pct therefore both refer to the pair-0 core. - CPU_MASTER=8; CPU_TX=16; CPU_RX=16 + # Final pair-0 TX/RX attribution is derived from the structured pinning below. + CPU_MASTER=8; CPU_TX=17; CPU_RX=16 ;; socket-tcp) # 1 MiB / 8000 / 1000 to mirror the published TCP matrix. The bench memsets a full @@ -272,9 +281,8 @@ case "$BACKEND" in # One combined base (both roles, netns IPs 10.250.0.1/2); see socket-udp note. BASE_YAML="$SCRIPT_DIR/daqiri_bench_socket_tcp_tx_rx_spark_netns.yaml" BENCH_BIN="$BUILD_DIR/examples/daqiri_bench_socket" - # Pair 0 pins to core 16 (see pair_core); report that core's busy% as the per-pair - # bottleneck. cpu_tx_pct and cpu_rx_pct therefore both refer to the pair-0 core. - CPU_MASTER=8; CPU_TX=16; CPU_RX=16 + # Final pair-0 TX/RX attribution is derived from the structured pinning below. + CPU_MASTER=8; CPU_TX=17; CPU_RX=16 ;; *) echo "Unknown backend: $BACKEND" >&2; exit 1 ;; esac @@ -288,8 +296,44 @@ esac 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" +# Optional space-separated overrides for one-off experiments. Apply them to both +# sweep and headline modes so smoke and drop-curve runs use the requested values. +if [[ -n "${BATCHES_OVERRIDE:-}" ]]; then + read -r -a BATCHES_SWEEP <<< "$BATCHES_OVERRIDE" + if (( ${#BATCHES_SWEEP[@]} == 0 )); then + echo "BATCHES_OVERRIDE must contain at least one batch size" >&2 + exit 1 + fi + BATCHES_HEADLINE=("${BATCHES_SWEEP[@]}") + for batch in "${BATCHES_SWEEP[@]}"; do + if [[ ! "$batch" =~ ^[1-9][0-9]*$ ]]; then + echo "Invalid BATCHES_OVERRIDE entry '$batch' (expected a positive integer)" >&2 + exit 1 + fi + if [[ "$BACKEND" == "socket-udp" && "$batch" -gt 32 ]]; then + echo "Invalid UDP batch size '$batch' (expected 1-32)" >&2 + exit 1 + fi + done +fi +if [[ -n "${PAIRS_OVERRIDE:-}" ]]; then + if [[ ! "$BACKEND" =~ ^socket- ]]; then + echo "PAIRS_OVERRIDE is supported only for socket backends" >&2 + exit 1 + fi + read -r -a PAIRS_SWEEP <<< "$PAIRS_OVERRIDE" + if (( ${#PAIRS_SWEEP[@]} == 0 )); then + echo "PAIRS_OVERRIDE must contain at least one pair count" >&2 + exit 1 + fi + PAIRS_HEADLINE=("${PAIRS_SWEEP[@]}") + for pairs in "${PAIRS_SWEEP[@]}"; do + if [[ ! "$pairs" =~ ^[1-9][0-9]*$ ]]; then + echo "Invalid PAIRS_OVERRIDE entry '$pairs' (expected a positive integer)" >&2 + exit 1 + fi + done +fi # 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 @@ -365,6 +409,10 @@ snapshot_cpu_stat() { # Compute busy% for a single cpu index between two /proc/stat snapshots. cpu_busy_pct() { local before="$1" after="$2" cpu_idx="$3" + if [[ ! "$cpu_idx" =~ ^[0-9]+$ ]]; then + echo "nan" + return + fi awk -v cpu="cpu$cpu_idx" ' NR == FNR { b_total[$1] = $2; b_busy[$1] = $3; next } { a_total[$1] = $2; a_busy[$1] = $3 } @@ -472,12 +520,28 @@ pair_server_io_core() { fi } +# Socket CSV utilization samples pair 0. Attribute TX to the client application +# worker and RX to the actual server receive role: UDP's recvmmsg() I/O thread, +# or TCP's server application worker. An unpinned run has no meaningful core sample. +if [[ "$BACKEND" =~ ^socket- ]]; then + if [[ -n "${SOCKET_NOPIN:-}" ]]; then + CPU_TX=-1 + CPU_RX=-1 + else + CPU_TX="$(pair_client_core 0)" + CPU_RX="$(pair_server_core 0)" + if [[ "$BACKEND" == "socket-udp" ]]; then + CPU_RX="$(pair_server_io_core 0 "$CPU_RX")" + fi + fi +fi + # 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 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 idx="$1" payload="$2" batch="$3" server_out="$4" client_out="$5" local srv_port=$(( SRV_PORT_BASE + idx )) local cli_port=$(( CLI_PORT_BASE + idx )) # SOCKET_NOPIN=1 runs the bench workers unpinned (cpu_core -1 -> no affinity, the @@ -495,7 +559,8 @@ generate_socket_yaml() { fi fi python3 "$NETNS_GEN" "$BASE_YAML" --role server \ - --rx-queue-cpu-core "$server_io_core" --tx-queue-cpu-core "$server_core" \ + --rx-queue-cpu-core "$server_io_core" --rx-queue-batch-size "$batch" \ + --tx-queue-cpu-core "$server_core" \ --bench-cpu-core "$server_core" | \ sed -E \ -e "s|^( *message_size: ).*|\1$payload|g" \ @@ -542,7 +607,7 @@ run_cell() { local stdout="$cell_dir/stdout.txt" local stderr="$cell_dir/stderr.txt" local bench_rc=0 - local pkts="" bytes="" secs="" rx_bytes="" + local pkts="" bytes="" secs="" rx_bytes="" observed_max_rx_burst=0 local bench_extra=() [[ "$target_gbps" != "0" ]] && bench_extra+=(--target-gbps "$target_gbps") @@ -566,11 +631,11 @@ run_cell() { if [[ "$BACKEND" =~ ^socket- ]]; then # `pairs` independent client/server processes, each in the wire-loopback # namespaces with unique ports and cores. A single pair is core-bound below line - # rate; the published Spark matrix reaches ~12 Gb/s by aggregating four pairs. + # rate; the published Spark matrix scales aggregate throughput with four pairs. # App TX (client sent) and App RX (server recv) are summed across pairs. local i server_pids=() client_pids=() for ((i = 0; i < pairs; i++)); do - generate_socket_yaml "$i" "$payload" \ + generate_socket_yaml "$i" "$payload" "$batch" \ "$cell_dir/server_p$i.yaml" "$cell_dir/client_p$i.yaml" done for ((i = 0; i < pairs; i++)); do @@ -591,14 +656,19 @@ run_cell() { local tx_pkts=0 tx_bytes=0 agg_rx_bytes=0 max_secs=0 for ((i = 0; i < pairs; i++)); do - local sp sb se rb + local sp sb se rb max_burst sp="$(extract_field 'Client complete' sent_packets "$cell_dir/client_p$i.stdout")" sb="$(extract_field 'Client complete' sent_bytes "$cell_dir/client_p$i.stdout")" se="$(extract_field 'Client complete' seconds "$cell_dir/client_p$i.stdout")" rb="$(extract_field 'Server complete' recv_bytes "$cell_dir/server_p$i.stdout")" + max_burst="$(extract_field 'Server complete' max_rx_burst \ + "$cell_dir/server_p$i.stdout")" tx_pkts=$(( tx_pkts + ${sp:-0} )) tx_bytes=$(( tx_bytes + ${sb:-0} )) agg_rx_bytes=$(( agg_rx_bytes + ${rb:-0} )) + if (( ${max_burst:-0} > observed_max_rx_burst )); then + observed_max_rx_burst="${max_burst:-0}" + fi max_secs="$(awk -v a="$max_secs" -v b="${se:-0}" 'BEGIN { print (b+0>a+0)?b:a }')" done pkts="$tx_pkts"; bytes="$tx_bytes"; rx_bytes="$agg_rx_bytes"; secs="$max_secs" @@ -775,7 +845,11 @@ run_cell() { 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" \ + local row="$lang,$BACKEND,$WORKLOAD_EFF,$payload,$batch,$observed_max_rx_burst,$pairs" + row+=",$target_gbps,$rep,$secs,$pkts,$bytes,$pps,$gbps,$rx_gbps,$drops,$drops_kind" + row+=",$CPU_MASTER,$CPU_TX,$CPU_RX,$cpu_master_pct,$cpu_tx_pct,$cpu_rx_pct" + row+=",$gpu_sm,$gpu_mem,$pp_gemm_dim,$pp_sync" + echo "$row" \ | tee -a "$CSV" } diff --git a/examples/socket_bench.cpp b/examples/socket_bench.cpp index 163afbfa..8261d4dd 100644 --- a/examples/socket_bench.cpp +++ b/examples/socket_bench.cpp @@ -38,7 +38,9 @@ namespace { volatile std::sig_atomic_t g_stop_requested = 0; void signal_handler(int signum) { - if (signum == SIGINT) { g_stop_requested = 1; } + if (signum == SIGINT) { + g_stop_requested = 1; + } } struct SocketBenchConfig { @@ -58,6 +60,7 @@ struct SocketWorkerStats { uint64_t received_packets = 0; uint64_t sent_bytes = 0; uint64_t received_bytes = 0; + uint32_t max_rx_burst = 0; }; SocketBenchConfig parse_socket_cfg(const YAML::Node& node) { @@ -104,8 +107,7 @@ void socket_worker(const SocketBenchConfig& cfg, daqiri::bench::TokenBucketPacer std::atomic& stop, SocketWorkerStats& stats, 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"; + const char* thread_name = cfg.server ? "socket_bench_server" : "socket_bench_client"; if (!daqiri::bench::set_current_thread_affinity(cfg.cpu_core, thread_name)) { stop.store(true); return; @@ -147,8 +149,8 @@ void socket_worker(const SocketBenchConfig& cfg, daqiri::bench::TokenBucketPacer if (cfg.server) { s = daqiri::socket_get_server_conn_id(cfg.server_address, cfg.server_port, &conn_id); } else { - s = daqiri::socket_connect_to_server( - cfg.server_address, cfg.server_port, cfg.client_address, &conn_id); + s = daqiri::socket_connect_to_server(cfg.server_address, cfg.server_port, + cfg.client_address, &conn_id); } if (s != daqiri::Status::SUCCESS) { @@ -165,13 +167,15 @@ void socket_worker(const SocketBenchConfig& cfg, daqiri::bench::TokenBucketPacer // When cfg.iterations <= 0, the loop is time-bounded (driven by stop.load() // set by --seconds). Otherwise the iteration cap applies as before. - const bool send_done = !cfg.send || - (cfg.iterations > 0 && - stats.sent_packets >= static_cast(cfg.iterations)); - const bool recv_done = !cfg.receive || - (cfg.iterations > 0 && - stats.received_packets >= static_cast(cfg.iterations)); - if (send_done && recv_done) { break; } + const bool send_done = + !cfg.send || + (cfg.iterations > 0 && stats.sent_packets >= static_cast(cfg.iterations)); + const bool recv_done = + !cfg.receive || + (cfg.iterations > 0 && stats.received_packets >= static_cast(cfg.iterations)); + if (send_done && recv_done) { + break; + } if (cfg.send && !send_done) { auto* msg = daqiri::create_tx_burst_params(); @@ -207,6 +211,7 @@ void socket_worker(const SocketBenchConfig& cfg, daqiri::bench::TokenBucketPacer const int num_pkts = static_cast(daqiri::get_num_packets(burst)); stats.received_packets += static_cast(num_pkts); stats.received_bytes += daqiri::get_burst_tot_byte(burst); + stats.max_rx_burst = std::max(stats.max_rx_burst, static_cast(num_pkts)); if (run_workload) { // Stage each received (host) payload to the GPU; the copy persists in @@ -293,7 +298,7 @@ int main(int argc, char** argv) { run_client = true; client_cfg = parse_socket_cfg(root["socket_bench_client"]); } - } catch (const std::exception &e) { + } catch (const std::exception& e) { std::cerr << "Invalid benchmark config: " << e.what() << "\n"; daqiri::shutdown(); return 1; @@ -322,33 +327,38 @@ int main(int argc, char** argv) { if (run_seconds > 0) { const auto elapsed = std::chrono::duration_cast( std::chrono::steady_clock::now() - start); - if (elapsed.count() >= run_seconds) { break; } + if (elapsed.count() >= run_seconds) { + break; + } } std::this_thread::sleep_for(std::chrono::milliseconds(100)); } stop.store(true); - if (server_thread.joinable()) { server_thread.join(); } - if (client_thread.joinable()) { client_thread.join(); } + if (server_thread.joinable()) { + server_thread.join(); + } + if (client_thread.joinable()) { + client_thread.join(); + } const double secs = - std::chrono::duration(std::chrono::steady_clock::now() - start) - .count(); + std::chrono::duration(std::chrono::steady_clock::now() - start).count(); if (run_server) { std::cout << "Server complete: sent_packets=" << server_stats.sent_packets << " recv_packets=" << server_stats.received_packets << " sent_bytes=" << server_stats.sent_bytes << " recv_bytes=" << server_stats.received_bytes - << " seconds=" << secs << '\n'; + << " max_rx_burst=" << server_stats.max_rx_burst << " seconds=" << secs << '\n'; } if (run_client) { std::cout << "Client complete: sent_packets=" << client_stats.sent_packets << " recv_packets=" << client_stats.received_packets << " sent_bytes=" << client_stats.sent_bytes << " recv_bytes=" << client_stats.received_bytes - << " seconds=" << secs << '\n'; + << " max_rx_burst=" << client_stats.max_rx_burst << " seconds=" << secs << '\n'; } daqiri::print_stats(); diff --git a/scripts/gen_spark_netns_config.py b/scripts/gen_spark_netns_config.py index dcb80119..2284da15 100755 --- a/scripts/gen_spark_netns_config.py +++ b/scripts/gen_spark_netns_config.py @@ -15,13 +15,14 @@ * interfaces: keep the one whose socket_config.mode == role. * memory regions: keep those whose name contains the role (SERVER / CLIENT). * bench sections: drop the other role's _bench_ mapping. - * optional core overrides: update RX queue, TX queue, and benchmark-worker - affinity independently without text substitutions that conflate them. + * optional queue overrides: update RX batch size plus RX queue, TX queue, and + benchmark-worker affinity without text substitutions that conflate them. This is the structural inverse of unioning the old _netns_server / _netns_client -files. run_spark_bench.sh supplies optional structured core overrides, then pipes -the output through per-message-size awk/sed rewrites (num_bufs/buf_size/depths -for RDMA; payload and ports for sockets). Output goes to stdout. +files. run_spark_bench.sh supplies optional structured queue overrides, then +pipes the output through per-message-size awk/sed rewrites +(num_bufs/buf_size/depths for RDMA; payload and ports for sockets). Output goes +to stdout. """ from __future__ import annotations @@ -36,6 +37,7 @@ def split_role( base: dict, role: str, rx_queue_cpu_core: int | None = None, + rx_queue_batch_size: int | None = None, tx_queue_cpu_core: int | None = None, bench_cpu_core: int | None = None, ) -> dict: @@ -50,6 +52,10 @@ def split_role( for interface in cfg["interfaces"]: for queue in interface.get("rx", {}).get("queues", []): queue["cpu_core"] = rx_queue_cpu_core + if rx_queue_batch_size is not None: + for interface in cfg["interfaces"]: + for queue in interface.get("rx", {}).get("queues", []): + queue["batch_size"] = rx_queue_batch_size if tx_queue_cpu_core is not None: for interface in cfg["interfaces"]: for queue in interface.get("tx", {}).get("queues", []): @@ -75,6 +81,7 @@ def main() -> int: ap.add_argument("base", help="path to the combined (both-role) netns base YAML") ap.add_argument("--role", choices=("server", "client"), required=True) ap.add_argument("--rx-queue-cpu-core", type=int) + ap.add_argument("--rx-queue-batch-size", type=int) ap.add_argument("--tx-queue-cpu-core", type=int) ap.add_argument("--bench-cpu-core", type=int) args = ap.parse_args() @@ -86,6 +93,7 @@ def main() -> int: base, args.role, rx_queue_cpu_core=args.rx_queue_cpu_core, + rx_queue_batch_size=args.rx_queue_batch_size, tx_queue_cpu_core=args.tx_queue_cpu_core, bench_cpu_core=args.bench_cpu_core, )