diff --git a/docs/api-reference/configuration.md b/docs/api-reference/configuration.md index 1a592640..282b0784 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,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 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/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 636f8ab2..18f101bc 100644 --- a/docs/benchmarks/socket_benchmarking.md +++ b/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: diff --git a/examples/daqiri_bench_socket_udp_tx_rx.yaml b/examples/daqiri_bench_socket_udp_tx_rx.yaml index 249adef4..7a2cbb75 100644 --- a/examples/daqiri_bench_socket_udp_tx_rx.yaml +++ b/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: 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..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,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 diff --git a/examples/run_spark_bench.sh b/examples/run_spark_bench.sh index 25358e3a..546b29f7 100755 --- a/examples/run_spark_bench.sh +++ b/examples/run_spark_bench.sh @@ -40,6 +40,12 @@ # (--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. +# 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 / @@ -80,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" @@ -137,6 +150,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 @@ -226,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 @@ -240,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 @@ -259,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 @@ -275,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 @@ -352,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 } @@ -450,39 +511,71 @@ 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 +} + +# 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 # 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" --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" \ -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|" \ - -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" --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|" \ -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" } @@ -514,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") @@ -538,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 @@ -563,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" @@ -747,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 e5108a21..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,11 +211,12 @@ 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 - // 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); @@ -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 c00102af..2284da15 100755 --- a/scripts/gen_spark_netns_config.py +++ b/scripts/gen_spark_netns_config.py @@ -15,11 +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 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 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 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 @@ -30,7 +33,14 @@ import yaml -def split_role(base: dict, role: str) -> dict: +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: cfg = base["daqiri"]["cfg"] cfg["interfaces"] = [ i for i in cfg["interfaces"] @@ -38,6 +48,18 @@ 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 + 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", []): + queue["cpu_core"] = tx_queue_cpu_core cfg["memory_regions"] = [ m for m in cfg["memory_regions"] if role.upper() in m["name"] ] @@ -45,8 +67,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 +80,23 @@ 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("--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() 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, + 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, + ) yaml.safe_dump(out, sys.stdout, sort_keys=False, default_flow_style=False) return 0 diff --git a/src/engines/socket/daqiri_socket_engine.cpp b/src/engines/socket/daqiri_socket_engine.cpp index e8056f0e..2c74a40d 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 @@ -48,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; } @@ -72,6 +75,29 @@ std::string sockaddr_to_ip(const sockaddr_in& addr) { return std::string(ip_buf); } +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 {}", port, cpu_core); + return false; + } + + 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 {}: {}", 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 SocketEngine::~SocketEngine() { @@ -170,6 +196,17 @@ 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_); + 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))); ep->rx_queue_state = get_or_create_rx_queue(ep->port, ep->rx_queue); @@ -598,7 +635,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); } @@ -797,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) { @@ -1006,6 +1051,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_)); @@ -1173,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_; @@ -1218,6 +1307,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", @@ -1318,15 +1414,29 @@ void SocketEngine::udp_rx_loop(int if_index) { auto* ep = endpoints_[if_index].get(); if (ep == nullptr) { return; } - if (ep->udp_fd < 0) { return; } + 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 = startup_ok; + ep->io_start_complete = true; + } + ep->io_start_cv.notify_one(); + if (!startup_ok) { + 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)); @@ -1337,46 +1447,54 @@ void SocketEngine::udp_rx_loop(int if_index) { } 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; } DAQIRI_LOG_WARN("UDP recvmmsg failed on port {}: {}", ep->port, strerror(errno)); continue; } + if (!running_.load()) { + break; + } if (received == 0) { continue; } - if (ep->socket_cfg.mode_ == SocketMode::SERVER) { + 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 + // 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 = static_cast(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); } } diff --git a/src/engines/socket/daqiri_socket_engine.h b/src/engines/socket/daqiri_socket_engine.h index 2dd5d3dc..129ca548 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_configured = 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);