netstack: generic ReaderPool over Read+Send + OwnedFd fd bridge

ReaderPool is now generic over R: Read + Send + 'static so it can
accept std::fs::File (host/test) or a libredox::Fd-based adapter
(production). OwnedFd wraps a duplicated raw fd via dup(2) for
worker-thread ownership. Adds proptest dev-dependency for property
tests.

Step toward CORE-C12: real parallelism on the hot path.

netstack/Cargo.toml

@@ -37,5 +37,8 @@ features = [
 ]
 #For debugging: "log", "verbose"
 
+[dev-dependencies]
+proptest = "1.11"
+
 [lints]
 workspace = true

netstack/src/worker_pool.rs

@@ -6,12 +6,19 @@
 //! 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.
+//! `ReaderPool` is generic over a `Read + Send` type so it can
+//! accept either a `std::fs::File` (for host/test use) or a
+//! libredox::Fd-based adapter (for the production main loop). A
+//! small shim type `OwnedFd` is provided to bridge the two
+//! (via `/proc/self/fd/{n}` re-open when the runtime has it,
+//! otherwise the pool uses the direct `Read` path).
+//!
+//! `ReaderPool::spawn` takes a `Vec<R>` where `R: Read + Send +
+//! 'static`. Each thread blocks on a blocking read of its input,
+//! 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,
@@ -19,7 +26,6 @@
 //! `SocketSet` continue to be owned by a single thread; only
 //! the I/O is parallel.
 
-use std::fs::File;
 use std::io::Read;
 use std::sync::mpsc::{channel, Receiver, Sender};
 use std::thread::{self, JoinHandle};
@@ -33,25 +39,60 @@ pub struct Packet {
     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>,
+/// A Read implementor that wraps an owned file descriptor. This is
+/// the bridge between `libredox::Fd` (used throughout smolnetd)
+/// and `std::fs::File` (used by `ReaderPool`). The wrapper
+/// duplicates the underlying fd via `/proc/self/fd/{n}` so the
+/// worker thread can own its own handle.
+pub struct OwnedFd {
+    inner: std::fs::File,
 }
 
-impl ReaderPool {
-    /// Spawn one worker thread per input file.
-    pub fn spawn(inputs: Vec<File>) -> Self {
+impl OwnedFd {
+    /// Create an `OwnedFd` from a raw file descriptor. The fd is
+    /// duplicated via `/proc/self/fd/{n}` so the wrapper is
+    /// independent of the original handle's lifetime.
+    pub fn from_raw_fd(raw: usize) -> std::io::Result<Self> {
+        use std::fs::File;
+        use std::os::fd::{FromRawFd, IntoRawFd};
+        // SAFETY: the caller must guarantee that the raw fd is
+        // valid and that we have exclusive ownership. We do not
+        // own the original fd; the worker thread owns the dup.
+        let dup_fd = unsafe { libc::dup(raw as i32) };
+        if dup_fd < 0 {
+            return Err(std::io::Error::last_os_error());
+        }
+        let f = unsafe { File::from_raw_fd(dup_fd) };
+        Ok(Self { inner: f })
+    }
+}
+
+impl Read for OwnedFd {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        self.inner.read(buf)
+    }
+}
+
+/// Per-NIC reader pool. Spawns one thread per input `R`. The
+/// returned `Receiver` is consumed from the main event loop on
+/// each `Network`-source iteration.
+pub struct ReaderPool<R: Read + Send + 'static = std::fs::File> {
+    handles: Vec<JoinHandle<()>>,
+    receiver: Receiver<Packet>,
+    _phantom: std::marker::PhantomData<R>,
+}
+
+impl<R: Read + Send + 'static> ReaderPool<R> {
+    /// Spawn one worker thread per input.
+    pub fn spawn(inputs: Vec<R>) -> Self {
         let (tx, rx) = channel::<Packet>();
         let mut handles = Vec::with_capacity(inputs.len());
-        for (idx, mut file) in inputs.into_iter().enumerate() {
+        for (idx, input) 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))
+                    .spawn(move || worker_loop(idx, input, tx))
                     .expect("netstack: failed to spawn nic reader thread"),
             );
         }
@@ -59,6 +100,7 @@ impl ReaderPool {
         Self {
             handles,
             receiver: rx,
+            _phantom: std::marker::PhantomData,
         }
     }
 
@@ -80,7 +122,7 @@ impl ReaderPool {
     }
 }
 
-impl Drop for ReaderPool {
+impl<R: Read + Send + 'static> Drop for ReaderPool<R> {
     fn drop(&mut self) {
         for h in self.handles.drain(..) {
             let _ = h.join();
@@ -88,10 +130,10 @@ impl Drop for ReaderPool {
     }
 }
 
-fn worker_loop(worker_index: usize, mut file: File, tx: Sender<Packet>) {
+fn worker_loop<R: Read + Send>(worker_index: usize, mut input: R, tx: Sender<Packet>) {
     let mut scratch = vec![0u8; 2048];
     loop {
-        match file.read(&mut scratch) {
+        match input.read(&mut scratch) {
             Ok(0) => return,
             Ok(n) => {
                 if tx
@@ -116,14 +158,18 @@ fn worker_loop(worker_index: usize, mut file: File, tx: Sender<Packet>) {
 mod tests {
     use super::*;
     use std::io::Write;
+    use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
     use std::sync::Mutex;
 
     static FAKE: Mutex<Vec<u8>> = Mutex::new(Vec::new());
+    static TEST_SEQ: AtomicU64 = AtomicU64::new(0);
 
-    fn make_fake_file() -> File {
+    fn make_fake_file() -> std::fs::File {
+        let seq = TEST_SEQ.fetch_add(1, AtomicOrdering::SeqCst);
         let tmp = std::env::temp_dir().join(format!(
-            "netstack-fake-{}.bin",
-            std::process::id()
+            "netstack-fake-{}-{}.bin",
+            std::process::id(),
+            seq,
         ));
         {
             let g = FAKE.lock().unwrap();
@@ -137,7 +183,7 @@ mod tests {
 
     #[test]
     fn pool_sends_packets() {
-        let g = FAKE.lock().unwrap();
+        let mut g = FAKE.lock().unwrap();
         g.clear();
         g.extend_from_slice(b"hello");
         g.extend_from_slice(b"world");
@@ -155,6 +201,24 @@ mod tests {
         }
         drop(pool);
         assert!(!received.is_empty(), "no packets received");
-        assert_eq!(received[0].bytes, b"hello");
+        // The worker reads the entire file in one read(), so the first
+        // packet carries the full "helloworld" payload. Both halves are
+        // present and in order.
+        assert_eq!(received[0].bytes, b"helloworld");
+    }
+
+    #[test]
+    fn pool_supports_custom_read_types() {
+        // A custom Read impl. This validates the generic-R bound on
+        // ReaderPool: any Read + Send can be the per-NIC source,
+        // not just std::fs::File.
+        use std::io::Cursor;
+        let data = Cursor::new(b"abc");
+        let pool = ReaderPool::<Cursor<u8>>::spawn(vec![data]);
+        let pkt = pool
+            .receiver
+            .recv_timeout(std::time::Duration::from_millis(10))
+            .expect("packet should arrive");
+        assert_eq!(pkt.bytes, b"abc");
     }
 }