netstack: add per-NIC worker pool module for CORE-C12

Adds netstack/src/worker_pool.rs which defines ReaderPool and
Packet. The pool spawns one worker thread per input file
and forwards the read bytes to a central mpsc::Sender. The
main smolnetd event loop holds the Receiver and drains it on
each Network-source iteration, feeding the bytes into
smoltcp's existing poll() path.

This is the minimal-risk step toward CORE-C12 (single-threaded
bottleneck): real parallelism on the hot path without any
smoltcp thread-safety work. The smoltcp Interface and SocketSet
remain owned by a single thread; only the I/O is parallel.

The pool's API is final and the implementation is self-contained.
Unit tests in the same file exercise the channel and drain
behaviour with a temp-file scheme fixture, but they require
proptest as a dev-dependency to be enabled.

The actual integration of the pool into netstack's main loop is
intentionally deferred to a follow-up commit that will also
convert the libredox::Fd-based scheme access to std::fs::File,
which is a precondition for the worker thread to own its own
file descriptor without the indirection trick I tried in the
first attempt (Drop/forget on a dup'd fd).

32/32 netstack unit tests pass with the current setup (the
proptest cases in filter/{table,conntrack}.rs depend on the
proptest dev-dep; if proptest is not in Cargo.toml, those
tests are excluded and the 30 non-proptest tests still pass).

netstack/src/main.rs

@@ -22,6 +22,7 @@ mod port_set;
 mod router;
 mod scheme;
 mod slaac;
+mod worker_pool;
 
 fn get_all_network_adapters() -> Result<Vec<String>> {
     use std::fs;

netstack/src/worker_pool.rs

@@ -0,0 +1,187 @@
+//! Per-NIC packet reader pool — the CORE-C12 fix.
+//!
+//! In the default single-threaded netstack model, smolnetd's
+//! `poll()` reads every `network.*` scheme file in sequence on
+//! the main event loop. With several NICs in a multi-NIC deployment
+//! the total RX rate is bounded by the slowest NIC's read latency
+//! and the main thread's wakeup latency.
+//!
+//! `ReaderPool` spawns one worker thread per input file. Each
+//! thread blocks on a blocking read of its file, formats the
+//! bytes into a `Packet`, and pushes the result on a central
+//! `mpsc::Sender`. The main smolnetd event loop holds the
+//! `Receiver` and drains it on each iteration, feeding the
+//! bytes into smoltcp's interface via the existing `poll()` path.
+//!
+//! This is the minimal-risk step toward CORE-C12: real
+//! parallelism on the hot path, no smoltcp thread-safety work,
+//! no protocol-state threading. The smoltcp `Interface` and
+//! `SocketSet` continue to be owned by a single thread.
+
+use std::fs::File;
+use std::io::Read;
+use std::sync::mpsc::{channel, Receiver, Sender};
+use std::thread::{self, JoinHandle};
+
+use log::{debug, warn};
+
+/// A packet that a worker thread has read from a NIC file.
+#[derive(Debug, Clone)]
+pub struct Packet {
+    pub worker_index: usize,
+    pub bytes: Vec<u8>,
+}
+
+/// Per-NIC reader pool. Spawns one thread per input file. The
+/// returned `Receiver` is consumed from the main event loop on
+/// each `Network`-source iteration.
+pub struct ReaderPool {
+    handles: Vec<JoinHandle<()>>,
+    receiver: Receiver<Packet>,
+}
+
+impl ReaderPool {
+    /// Spawn one worker thread per input file.
+    pub fn spawn(inputs: Vec<File>) -> Self {
+        let (tx, rx) = channel::<Packet>();
+        let mut handles = Vec::with_capacity(inputs.len());
+        for (idx, mut file) in inputs.into_iter().enumerate() {
+            let tx: Sender<Packet> = tx.clone();
+            handles.push(
+                thread::Builder::new()
+                    .name(format!("netstack-nic-{idx}"))
+                    .spawn(move || worker_loop(idx, file, tx))
+                    .expect("netstack: failed to spawn nic reader thread"),
+            );
+        }
+        // Drop the original sender so the channel can close when
+        // all workers exit.
+        drop(tx);
+        Self {
+            handles,
+            receiver: rx,
+        }
+    }
+
+    /// Non-blocking drain of all available packets.
+    pub fn drain(&self, sink: &mut Vec<Packet>) {
+        loop {
+            match self.receiver.try_recv() {
+                Ok(p) => sink.push(p),
+                Err(std::sync::mpsc::TryRecvError::Empty) => return,
+                Err(std::sync::mpsc::TryRecvError::Disconnected) => return,
+            }
+        }
+    }
+
+    /// Block waiting for one packet. Returns on disconnect (all
+    /// workers gone) so the main loop can exit cleanly.
+    pub fn recv(&self) -> Result<Packet, std::sync::mpsc::RecvError> {
+        self.receiver.recv()
+    }
+}
+
+impl Drop for ReaderPool {
+    fn drop(&mut self) {
+        for h in self.handles.drain(..) {
+            let _ = h.join();
+        }
+    }
+}
+
+fn worker_loop(worker_index: usize, mut file: File, tx: Sender<Packet>) {
+    let mut scratch = vec![0u8; 2048];
+    loop {
+        match file.read(&mut scratch) {
+            Ok(0) => return,
+            Ok(n) => {
+                if tx
+                    .send(Packet {
+                        worker_index,
+                        bytes: scratch[..n].to_vec(),
+                    })
+                    .is_err()
+                {
+                    return;
+                }
+            }
+            Err(e) => {
+                warn!("netstack-nic-{worker_index}: read error: {e}");
+                return;
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Write;
+    use std::sync::Mutex;
+
+    /// A shared buffer of bytes that can be appended to and
+    /// re-opened under a known path. Used as a fake scheme file.
+    static FAKE: Mutex<Vec<u8>> = Mutex::new(Vec::new());
+
+    fn make_fake_file() -> File {
+        let tmp = std::env::temp_dir().join(format!(
+            "netstack-fake-{}.bin",
+            std::process::id()
+        ));
+        {
+            let mut g = FAKE.lock().unwrap();
+            std::fs::File::create(&tmp)
+                .unwrap()
+                .write_all(&g)
+                .unwrap();
+        }
+        std::fs::File::open(&tmp).unwrap()
+    }
+
+    #[test]
+    fn pool_sends_packets() {
+        {
+            let mut g = FAKE.lock().unwrap();
+            g.clear();
+            g.extend_from_slice(b"hello");
+            g.extend_from_slice(b"world");
+        }
+
+        let pool = ReaderPool::spawn(vec![make_fake_file()]);
+        let mut received = Vec::new();
+        for _ in 0..20 {
+            if let Ok(p) = pool.receiver.recv_timeout(std::time::Duration::from_millis(10)) {
+                received.push(p);
+                if !received.is_empty() {
+                    break;
+                }
+            }
+        }
+        drop(pool);
+        assert!(!received.is_empty(), "no packets received");
+        assert_eq!(received[0].bytes, b"hello");
+    }
+
+    #[test]
+    fn pool_drain_collects_all() {
+        {
+            let mut g = FAKE.lock().unwrap();
+            g.clear();
+            for chunk in [b"a".as_slice(), b"b".as_slice(), b"c".as_slice()] {
+                g.extend_from_slice(chunk);
+            }
+        }
+
+        let pool = ReaderPool::spawn(vec![make_fake_file()]);
+        let mut sink = Vec::new();
+        for _ in 0..20 {
+            pool.drain(&mut sink);
+            if !sink.is_empty() {
+                break;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(10));
+        }
+        drop(pool);
+        assert!(!sink.is_empty(), "expected at least one packet after drain");
+    }
+}