node: drain packet_rx / tun_outbound_rx in batches in run_rx_loop

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.
This commit is contained in:
Martti Malmi
2026-05-10 00:28:45 +03:00
committed by Dev
parent 253dddabe3
commit fac4450694
+28
View File
@@ -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!(