#287 - Batch UDP socket receives and honor RX queue affinity

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,6 +147,13 @@ 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` 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
 `?engine=ibverbs`; when omitted, `ibverbs` is the default and only supported RoCE
@@ -171,13 +180,19 @@ 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. 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`
+  - 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

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:

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/<timestamp>-<backend>-<mode>/runs.csv`. See
+Each run writes `bench-results/<timestamp>-<backend>-<mode>/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.

docs/benchmarks/socket_benchmarking.md

@@ -158,6 +158,28 @@ 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).
+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 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
 `socket_setsockopt(conn_id, level, optname, optval, optlen)` with the integer
@@ -187,13 +209,14 @@ 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:
         - name: "RX_Queue"
           id: 0
           cpu_core: 8
-          batch_size: 1
+          batch_size: 32
           memory_regions: ["DATA_SOCKET_SERVER"]
       tx:
         queues:
@@ -246,7 +269,7 @@ daqiri:
         - name: "RX_Queue"
           id: 0
           cpu_core: 8
-          batch_size: 1
+          batch_size: 32
           memory_regions: ["DATA_SOCKET_CLIENT"]
       tx:
         queues:

examples/daqiri_bench_socket_udp_tx_rx.yaml

@@ -26,13 +26,14 @@ 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:
         - name: "Server_RX_Queue"
           id: 0
           cpu_core: 8
-          batch_size: 1
+          batch_size: 32
           memory_regions:
             - "DATA_SOCKET_SERVER"
       tx:
@@ -56,7 +57,7 @@ daqiri:
         - name: "Client_RX_Queue"
           id: 0
           cpu_core: 7
-          batch_size: 1
+          batch_size: 32
           memory_regions:
             - "DATA_SOCKET_CLIENT"
       tx:

examples/daqiri_bench_socket_udp_tx_rx_spark_netns.yaml

@@ -6,8 +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). 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
@@ -46,13 +47,15 @@ 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:
         - 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:
@@ -88,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