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

[dleshchev]

Related issue

Closes #287

Summary

• Coalesce the datagrams returned by one UDP recvmmsg() call into one DAQIRI
  receive burst. rx.queues[].batch_size selects a maximum of 1-32 datagrams,
  and MSG_WAITFORONE returns partial batches without waiting for all slots.
• Reuse rx.queues[].cpu_core for the UDP receive I/O thread. Initialization
  fails if the requested affinity cannot be applied instead of continuing
  unpinned.
• Allow a UDP server to identify its expected client with the existing
  socket_config.remote_addr. DAQIRI connects that socket to the configured
  peer, so other sources cannot claim the endpoint. For compatibility, servers
  without remote_addr retain single-active-peer behavior and are limited to
  one datagram per receive burst.
• Apply socket benchmark batch and CPU overrides structurally. The generated
  server RX queue now matches the batch recorded in the artifact name and CSV,
  and SOCKET_RX_IO_CORES can place UDP receive threads independently.
• Report the configured and observed RX burst sizes plus the actual sampled CPU
  core IDs in benchmark CSVs. Multi-pair socket CPU percentages are explicitly
  pair-0 samples.

Validation

• Full project-container build with DAQIRI_ENGINE="dpdk ibverbs", examples,
  and Python bindings enabled.
• Configured-peer UDP loopback with a 32-packet RX limit: exactly 1,000 packets
  and 1,024,000 bytes sent and received in each direction. The final focused
  run observed max_rx_burst=12, proving that a returned application burst held
  more than one datagram.
• Injected 100 datagrams from an unexpected source before any expected-peer
  traffic: the configured server received zero packets from that source.
• Legacy UDP server without remote_addr: emitted the batch-size fallback
  warning and completed exactly 1,000 packets in each direction.
• Invalid RX queue CPU 9999 caused daqiri_init to fail, confirming affinity
  failures are not silently ignored.
• Structured generator assertions verify independent RX batch, RX core, TX
  core, and benchmark-worker overrides.
• clang-format, git diff --check, bash -n, Python syntax checks, and
  scripts/check_doc_refs.py pass.
• Physical Spark-to-Spark UDP was previously validated with 8 KB payloads,
  30-second paced TX, 70-second RX lifetime, and three repetitions per point.
  With batching and explicit UDP I/O placement, all three repetitions were
  counter-clean at 27.5 Gb/s with one RX worker, 40 Gb/s with two, and 75 Gb/s
  with four. PHY TX/RX deltas matched; NoPorts, RcvbufErrors, and NIC
  receive-buffer counters were flat.
• Against DAQIRI main commit 263f4b5c3aedf25698c6404309fac6158359b6ef,
  with the same benchmark-worker topology but no batching or UDP I/O affinity,
  those points were clean in 0/3, 2/3, and 0/3 runs respectively. Loss was
  primarily accounted for by receiver RcvbufErrors; four-worker runs also
  recorded NIC receive-buffer drops.

The physical throughput matrix was not rerun after the configured-peer or
benchmark-reporting follow-ups. Those earlier artifacts therefore do not contain
the new configured/observed batch and CPU-core columns.

Scope

This PR covers UDP receive batching, socket I/O-thread placement through the
existing RX queue configuration, deterministic point-to-point server peer
selection, and the benchmark/reporting changes needed to reproduce and audit
those settings. It does not add general multi-client UDP server support or
change receive-buffer sizing.

[greptile-apps]

Greptile Summary

The PR batches UDP datagrams into receive bursts, pins UDP receive threads using existing RX queue affinity, and supports configured server peers to preserve deterministic peer identity.

• Adds configurable UDP receive batches of 1–32 datagrams.
• Applies rx.queues[].cpu_core to UDP receive I/O threads and fails initialization when affinity cannot be applied.
• Connects configured UDP servers to their expected peer while retaining single-datagram legacy peer learning.
• Updates socket benchmarks, configuration generation, examples, and related documentation.

Confidence Score: 5/5

The PR appears safe to merge.

No blocking failure remains.

Important Files Changed

Filename	Overview
src/engines/socket/daqiri_socket_engine.cpp	Implements UDP batching, receive-thread affinity startup synchronization, configured-peer filtering, and safe single-packet fallback for dynamic peers.
src/engines/socket/daqiri_socket_engine.h	Adds endpoint state for UDP batching, receive affinity, configured-peer tracking, and thread-start synchronization.
examples/run_spark_bench.sh	Separates benchmark-worker and UDP I/O affinity controls and records observed burst and CPU-placement data.
scripts/gen_spark_netns_config.py	Adds structured queue batch and affinity overrides when generating per-role benchmark configurations.
examples/socket_bench.cpp	Records the maximum observed receive-burst size while preserving burst cleanup.
docs/api-reference/configuration.md	Documents UDP batching limits, RX affinity semantics, and configured versus dynamic server peers.

_{Reviews (4): Last reviewed commit: "#287 - Correct UDP benchmark metadata" | Re-trigger Greptile}