feat: RX backpressure accounting and socket drop counters

.github/workflows/perf-tests.yml

@@ -18,7 +18,7 @@ on:
       configs:
         description: 'DUT configs to benchmark'
         required: false
-        default: 'plain-rust,rust-dpdk,native-dpdk'
+        default: 'plain-rust,rust-dpdk,tokio-dpdk,native-dpdk'
       teardown:
         description: 'Destroy stack after tests'
         required: false
@@ -151,7 +151,7 @@ jobs:
             --packet-sizes "${{ inputs.packet_sizes || '64,512,1400,8500' }}" \
             --duration "${{ inputs.duration || '30' }}" \
             --rate-steps "${{ inputs.rate_steps || '70000,140000,350000,700000' }}" \
-            --configs "${{ inputs.configs || 'plain-rust,rust-dpdk,native-dpdk' }}" \
+            --configs "${{ inputs.configs || 'plain-rust,rust-dpdk,tokio-dpdk,native-dpdk' }}" \
             --json-summary
         env:
           DPDK_AMI_ID: ${{ steps.amis.outputs.dpdk_ami }}

README.md

@@ -135,6 +135,31 @@ RX: Backend `recv_frames()` → parse headers → ARP/ICMP inline → UDP payloa
 
 Two packet construction paths exist by design: `build_udp_packet(&mut Mbuf)` writes directly into DPDK mbufs (zero-copy), while `build_udp_frame() -> Vec<u8>` produces owned bytes for the generic backend path. Both emit identical wire-format frames.
 
+### RX Drop Hierarchy
+
+An incoming packet can be dropped at five distinct layers between the wire and
+the application. When diagnosing loss, narrow down *which* layer is dropping
+before touching code — the fix is different at each one. The perf instrumentation
+(`[PERF]` log lines + perf-test harness) exposes counters at every layer we own,
+and the comparison table surfaces them as **NIC Drops** and **App Drops** columns:
+
+| # | Layer | Dropped because... | Counter | Column |
+|---|---|---|---|---|
+| 1 | Wire / NIC ingress | AWS ENA rate limiter, VPC shaping, bad cabling, upstream congestion | — (not owned by this stack) | inferred: `(TX − RX) − NIC Drops − App Drops` |
+| 2 | NIC RX descriptor ring (HW) | Software polled too slowly → ring fills → NIC has nowhere to DMA new packets | `rte_eth_stats.imissed` | **NIC Drops** |
+| 2b | NIC RX refill (HW) | Mempool exhausted → can't hand a free mbuf to the NIC for the next DMA | `rte_eth_stats.rx_nombuf` | **NIC Drops** |
+| 3 | dpdk-udp worker ring (SW, multi-core) | Internal SpscRing between RX worker thread and app thread is full | `PerfCounters.rx_drops_ring_full` | **App Drops** |
+| 4 | dpdk-udp `recv_queue` (SW, per-socket) | Per-socket `SO_RCVBUF`-equivalent (4096 pkts / 256 KiB) is full — app isn't calling `recv_from` fast enough | `PerfCounters.rx_drops_buffer_full` | **App Drops** |
+
+**How to read the columns in perf reports:**
+
+- **NIC Drops > 0, App Drops ≈ 0** → the poller isn't calling `rte_eth_rx_burst` fast enough to drain the HW ring (layer 2), or the mempool is too small (layer 2b). Fix: faster polling loop, larger mempool, more RX queues.
+- **NIC Drops ≈ 0, App Drops > 0** → the packet made it into the Rust stack but got stuck in the worker ring (layer 3) or the socket buffer (layer 4). Fix: faster consumer / larger `recv_queue` cap / move work off the app thread.
+- **Both ≈ 0 but RX < TX** → loss is at layer 1 (wire), which we can't directly count. Cross-reference with `native-dpdk` at the same rate to confirm it's environmental rather than something the stack is doing.
+- **Both > 0** → backpressure is propagating from app layer down through the stack. Start with layer 4, work down.
+
+For async backends, note that the `tokio-dpdk` compat layer adds a `spawn_blocking` hop per `recv_from`/`send_to` call, which caps throughput around 40K pps and makes layer 2 saturate easily under load. For raw throughput use the sync `dpdk_udp::UdpSocket` directly.
+
 ## Development
 
 ### Build and Test
@@ -237,6 +262,7 @@ This is **not a general-purpose network stack**. It does not replace the Linux k
 | Hardware checksum offload | Complete | TX offload on capable NICs, RX validation on all packets |
 | Multiple backends | 3 backends | DPDK, AF_PACKET, AF_PACKET+MMAP |
 | Ephemeral port allocation | Complete | Linux-compatible range (32768-60999) |
+| RX backpressure + drop counters | Complete | `SO_RCVBUF`-style byte limit, atomic `recv_drops()`, 256 KiB default |
 | Multicast join/leave | Basic | IPv4 only, simplified group tracking |
 | Connected socket filtering | Complete | Buffers non-matching packets |
 | Socket timeouts | Complete | Read and write deadlines |
@@ -248,7 +274,7 @@ The Linux kernel's UDP path (`net/ipv4/udp.c` and surrounding infrastructure) ha
 | Feature | Kernel | Us | Impact |
 |---------|--------|-----|--------|
 | **Subnet-aware routing** | Full FIB with longest-prefix match | Subnet-aware with longest-prefix-match, auto-detection from OS | Done |
-| **RX backpressure and drop counters** | `sk_rmem_alloc` / `sk_rcvbuf` with per-socket drop counters | None — unbounded buffering until hardware ring fills | Planned |
+| **RX backpressure and drop counters** | `sk_rmem_alloc` / `sk_rcvbuf` with per-socket drop counters | Configurable byte-based buffer with atomic drop counters (`recv_drops`, `set_recv_buffer_size`) | Done |
 | **RX checksum validation** | Hardware or software verification on every packet | Software verification on every RX packet | Done |
 | **TX hardware checksum offload** | `CHECKSUM_PARTIAL` — NIC computes checksum | NIC computes when capable, software fallback | Done |
 | **ICMP error handling** | Destination/port unreachable queued to originating socket | Echo reply only — all ICMP errors ignored | Planned |
@@ -289,9 +315,9 @@ Integration testing runs on **AWS EC2 with VPC networking**, which has specific
 
 **TX hardware checksum offload** — When the NIC supports `CHECKSUM_PARTIAL` mode (e.g., ENA on AWS), the DPDK backend sets mbuf offload flags (`RTE_MBUF_F_TX_IP_CKSUM`, `RTE_MBUF_F_TX_UDP_CKSUM`) and writes the pseudo-header checksum so the NIC computes final checksums. Falls back to software checksums on NICs without offload or on non-DPDK backends.
 
-### Planned
+**RX backpressure and drop counters** — Socket-level receive buffer accounting with a configurable byte limit (SO_RCVBUF equivalent) and lock-free atomic drop counters. Applications call `set_recv_buffer_size(bytes)` to tune the limit and `recv_drops()` to read a `RecvDropStats { packets, bytes }` snapshot for production monitoring. Drops are also surfaced via the existing `rx_drops_buffer_full` perf counter and rolled into the `rx_drops` rate on the perf reporter. Default is 256 KiB, mirroring Linux `net.core.rmem_default`.
 
-**RX backpressure and drop counters** — Implement socket-level receive buffer accounting with configurable limits and exposed drop counters. Applications need visibility into packet loss. This is the most important gap for production use.
+### Planned
 
 **ICMP error handling** — Process destination unreachable and fragmentation needed messages. Surface errors to the application via the socket error API. Enables path MTU discovery.
 

apps/echo/src/main.rs

@@ -1,5 +1,6 @@
 use clap::Parser;
 use std::io;
+use std::mem::MaybeUninit;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
@@ -12,6 +13,27 @@ extern "C" fn signal_handler(_sig: libc::c_int) {
     SHUTDOWN.store(true, Ordering::Relaxed);
 }
 
+/// Install `signal_handler` for SIGTERM and SIGINT via `sigaction`.
+///
+/// The old `libc::signal(sig, fn as *const () as sighandler_t)` cast is
+/// brittle on some toolchains: with certain codegen paths the resulting
+/// `usize` doesn't match the actual function address, leaving the default
+/// handler in place, which means SIGTERM terminates the process without
+/// running any destructors — and in particular without running
+/// `PerfReporter::drop`, so the one-shot `[NIC-FINAL]` log line the perf
+/// harness relies on is never emitted. `sigaction` takes a typed
+/// `sa_sigaction`/`sa_handler` field and avoids the cast entirely.
+fn install_signal_handlers() {
+    unsafe {
+        let mut sa: libc::sigaction = MaybeUninit::zeroed().assume_init();
+        sa.sa_sigaction = signal_handler as *const () as usize;
+        libc::sigemptyset(&mut sa.sa_mask);
+        sa.sa_flags = 0;
+        libc::sigaction(libc::SIGTERM, &sa, std::ptr::null_mut());
+        libc::sigaction(libc::SIGINT, &sa, std::ptr::null_mut());
+    }
+}
+
 #[derive(Parser)]
 #[command(name = "echo")]
 #[command(about = "UDP echo server using dpdk-udp (DPDK-accelerated drop-in for std::net)")]
@@ -30,10 +52,7 @@ struct Args {
 }
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    unsafe {
-        libc::signal(libc::SIGTERM, signal_handler as *const () as libc::sighandler_t);
-        libc::signal(libc::SIGINT, signal_handler as *const () as libc::sighandler_t);
-    }
+    install_signal_handlers();
 
     let args = Args::parse();
     let bind_addr = format!("{}:{}", args.ip, args.port);

apps/tokio-echo/Cargo.toml

@@ -6,7 +6,13 @@ description = "Async UDP Echo Server using Tokio with DPDK support"
 publish = false
 
 [features]
-default = []
+# `dpdk` is on by default so a workspace-level build like
+# `cargo build --features dpdk-stdlib-sys/bindgen` produces a binary that
+# actually uses the DPDK backend on machines where DPDK is installed.
+# When DPDK is not present, dpdk-stdlib-sys silently falls back to stubs,
+# so local `cargo build` (no DPDK) still works — it just runs against the
+# stub backend, identical to today's behavior.
+default = ["dpdk"]
 dpdk = ["dpdk-tokio/dpdk"]
 
 [dependencies]

apps/tokio-echo/src/main.rs

@@ -61,6 +61,12 @@ struct Args {
     /// Statistics report interval in seconds
     #[arg(long, default_value_t = 10)]
     stats_interval: u64,
+
+    /// Performance reporting interval in seconds (0 = disabled).
+    /// On the DPDK backend this starts the PerfReporter, emitting `[PERF]`
+    /// log lines with rx/tx pps, drop counters, and latency percentiles.
+    #[arg(long, default_value_t = 0)]
+    perf_interval: u64,
 }
 
 /// Statistics tracker
@@ -136,6 +142,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("Local address: {}", socket.local_addr()?);
     println!("Max concurrent handlers: {}", args.workers);
     println!("Request timeout: {}ms", args.timeout);
+
+    if args.perf_interval > 0 {
+        socket.enable_perf_reporting(Duration::from_secs(args.perf_interval)).await?;
+        println!("Perf reporting interval: {}s", args.perf_interval);
+    }
+
     println!();
     println!("Echo server running... (Ctrl+C to stop)");
     println!();
@@ -159,20 +171,57 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Semaphore for limiting concurrent handlers
     let semaphore = Arc::new(Semaphore::new(args.workers));
 
+    // Install a shutdown signal that fires on SIGTERM / SIGINT. Racing
+    // this against recv_from in a tokio::select! lets the main loop exit
+    // cleanly on signal, which drops the Arc<dyn AsyncUdpSocket>, which
+    // drops the underlying UdpSocket, which drops the PerfReporter — and
+    // that Drop impl is what emits the one-shot `[NIC-FINAL]` line the
+    // perf harness cross-checks against per-tick [PERF] deltas. Without
+    // this, the process is killed by pkill -TERM before any destructor
+    // can run and [NIC-FINAL] is never emitted.
+    let shutdown = async {
+        #[cfg(unix)]
+        {
+            use tokio::signal::unix::{signal, SignalKind};
+            let mut sigterm = signal(SignalKind::terminate())
+                .expect("failed to install SIGTERM handler");
+            let mut sigint = signal(SignalKind::interrupt())
+                .expect("failed to install SIGINT handler");
+            tokio::select! {
+                _ = sigterm.recv() => {},
+                _ = sigint.recv() => {},
+            }
+        }
+        #[cfg(not(unix))]
+        {
+            let _ = tokio::signal::ctrl_c().await;
+        }
+    };
+    tokio::pin!(shutdown);
+
     // Main receive loop
     let mut buf = [0u8; 65535];
 
     loop {
-        // Receive a packet
-        let (len, from_addr) = match socket.recv_from(&mut buf).await {
-            Ok(result) => result,
-            Err(e) => {
-                stats.record_error();
-                if args.verbose {
-                    eprintln!("Receive error: {}", e);
-                }
-                continue;
+        // Receive a packet, racing against shutdown. On signal we break
+        // out of the loop, fall through to the `Shutting down` log, and
+        // let main() return so PerfReporter::drop() emits [NIC-FINAL].
+        let (len, from_addr) = tokio::select! {
+            biased;
+            _ = &mut shutdown => {
+                println!("Received shutdown signal, stopping echo server...");
+                break;
             }
+            recv = socket.recv_from(&mut buf) => match recv {
+                Ok(result) => result,
+                Err(e) => {
+                    stats.record_error();
+                    if args.verbose {
+                        eprintln!("Receive error: {}", e);
+                    }
+                    continue;
+                }
+            },
         };
 
         stats.record_recv(len);
@@ -230,4 +279,16 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             }
         });
     }
+
+    // Explicitly stop the PerfReporter before returning. This joins the
+    // reporter thread synchronously inside a spawn_blocking and triggers
+    // the one-shot `[NIC-FINAL]` stderr line that the perf harness pairs
+    // with `[NIC-BASELINE]` for its instrumentation self-check. Doing this
+    // explicitly (rather than relying on Drop) removes all timing dependence
+    // on Arc refcounts and tokio runtime shutdown ordering.
+    if args.perf_interval > 0 {
+        socket.disable_perf_reporting().await;
+    }
+    println!("Shutting down gracefully...");
+    Ok(())
 }

deploy/cdk/lib/perf-test-stack.ts

@@ -324,10 +324,11 @@ export class PerfTestStack extends cdk.Stack {
       'source /root/.cargo/env',
       'cd /opt/dpdk-stdlib',
       'PKG_CONFIG_PATH=/usr/local/lib/pkgconfig pkg-config --modversion libdpdk',
-      // Build all binaries including plain-echo (no dpdk feature needed)
+      // Single workspace build with bindgen → produces real-DPDK binaries for echo,
+      // tokio-echo (which has `dpdk` as a default feature), and plain-echo (no dpdk dep).
       'PKG_CONFIG_PATH=/usr/local/lib/pkgconfig cargo build --release --features dpdk-sys/bindgen',
       'echo "=== Build complete ==="',
-      'ls -la target/release/echo target/release/plain-echo',
+      'ls -la target/release/echo target/release/plain-echo target/release/tokio-echo',
       // Collect environment info
       'echo "=== DUT Environment ==="',
       'echo "Instance type: $(curl -s -H \"X-aws-ec2-metadata-token: $(curl -s -X PUT http://169.254.169.254/latest/api/token -H X-aws-ec2-metadata-token-ttl-seconds:21600)\" http://169.254.169.254/latest/meta-data/instance-type)"',