From fac44506949a8d0c8e29e8ad4c2a6df6c1b235e8 Mon Sep 17 00:00:00 2001
From: Martti Malmi <sirius@iki.fi>
Date: Sat, 9 May 2026 18:17:42 +0300
Subject: [PATCH] node: drain packet_rx / tun_outbound_rx in batches in
 run_rx_loop
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The run_rx_loop's `tokio::select!` was costing one full scheduler hop
+ futex per inbound packet and per outbound TUN packet. Under
sustained load that capped throughput at one event per scheduler
quantum — independent of CPU (which sat near-idle) because every
iteration parked the worker, woke it via futex, processed one event,
then parked again.

After the await on `packet_rx.recv()` / `tun_outbound_rx.recv()`
fires, drain up to 256 additional ready items via `try_recv()` in a
tight inner loop before yielding back to `select!`. `biased` ordering
gives the data-plane branches priority over tick / control / DNS
under sustained load.

The 256 cap is empirically tuned to keep the worker on a busy stream
between yield points (a contiguous burst of ~256 MTU-sized packets
≈ 400 KB of contiguous traffic) while still bounding the inner loop
so a flood on one branch can't starve the periodic tick or control
socket. Lower caps (64) left perf on the table; higher caps (1024+)
delayed tick handling visibly under stress.

Pairs with the recvmmsg(2) change in the previous commit: the kernel
UDP queue now hands packets to `packet_rx` in 32-batches, and the
rx_loop drains them without a per-packet scheduler hop.
---
 src/node/handlers/rx_loop.rs | 28 ++++++++++++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs
index 3399b77..e45684e 100644
--- a/src/node/handlers/rx_loop.rs
+++ b/src/node/handlers/rx_loop.rs
@@ -82,14 +82,42 @@ impl Node {
 
         loop {
             tokio::select! {
+                biased;
                 packet = packet_rx.recv() => {
                     match packet {
                         Some(p) => self.process_packet(p).await,
                         None => break, // channel closed
                     }
+                    // Drain remaining ready inbound packets in a tight loop
+                    // before yielding back to select! — every yield is a
+                    // futex hop on tokio's multi-thread scheduler, and at
+                    // line rate the kernel UDP queue typically has several
+                    // datagrams available per wake. Caps at a batch
+                    // boundary so other branches (tick, control) eventually
+                    // get a turn even under sustained load.
+                    let mut drained = 0;
+                    while drained < 256 {
+                        match packet_rx.try_recv() {
+                            Ok(p) => {
+                                self.process_packet(p).await;
+                                drained += 1;
+                            }
+                            Err(_) => break,
+                        }
+                    }
                 }
                 Some(ipv6_packet) = tun_outbound_rx.recv() => {
                     self.handle_tun_outbound(ipv6_packet).await;
+                    let mut drained = 0;
+                    while drained < 256 {
+                        match tun_outbound_rx.try_recv() {
+                            Ok(p) => {
+                                self.handle_tun_outbound(p).await;
+                                drained += 1;
+                            }
+                            Err(_) => break,
+                        }
+                    }
                 }
                 Some(identity) = dns_identity_rx.recv() => {
                     debug!(