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.
2026-07-27 06:16:21 +09:00
parent c33f6ce762
commit 36dddf2300
3 changed files with 258 additions and 27 deletions
@@ -250,6 +250,21 @@ dependencies = [
 "serde",
 ]

+[[package]]
+name = "bit-set"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08807e080ed7f9d5433fa9b275196cfc35414f66a0c79d864dc51a0d825231a3"
+dependencies = [
+ "bit-vec",
+]
+
+[[package]]
+name = "bit-vec"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7"
+
 [[package]]
 name = "bit_field"
 version = "0.10.3"
@@ -779,6 +794,12 @@ version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e88a8acf291dafb59c2d96e8f59828f3838bb1a70398823ade51a84de6a6deed"

+[[package]]
+name = "fastrand"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223"
+
 [[package]]
 name = "fbbootlogd"
 version = "0.1.0"
@@ -835,6 +856,12 @@ version = "0.1.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"

+[[package]]
+name = "fnv"
+version = "1.0.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
+
 [[package]]
 name = "fuchsia-cprng"
 version = "0.1.1"
@@ -949,6 +976,18 @@ dependencies = [
 "wasi",
 ]

+[[package]]
+name = "getrandom"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
+dependencies = [
+ "cfg-if 1.0.4",
+ "libc",
+ "r-efi 5.3.0",
+ "wasip2",
+]
+
 [[package]]
 name = "getrandom"
 version = "0.4.3"
@@ -957,7 +996,7 @@ checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099"
 dependencies = [
 "cfg-if 1.0.4",
 "libc",
- "r-efi",
+ "r-efi 6.0.0",
 "rand_core 0.10.1",
 ]

@@ -1493,6 +1532,7 @@ dependencies = [
 "daemon",
 "libredox",
 "log",
+ "proptest",
 "redox-log",
 "redox-scheme",
 "redox_event",
@@ -1718,6 +1758,25 @@ dependencies = [
 "unicode-ident",
 ]

+[[package]]
+name = "proptest"
+version = "1.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4b45fcc2344c680f5025fe57779faef368840d0bd1f42f216291f0dc4ace4744"
+dependencies = [
+ "bit-set",
+ "bit-vec",
+ "bitflags 2.13.0",
+ "num-traits",
+ "rand 0.9.5",
+ "rand_chacha 0.9.0",
+ "rand_xorshift",
+ "regex-syntax",
+ "rusty-fork",
+ "tempfile",
+ "unarray",
+]
+
 [[package]]
 name = "ps2d"
 version = "0.1.0"
@@ -1754,6 +1813,12 @@ version = "3.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8bb0fd6580eeed0103c054e3fba2c2618ff476943762f28a645b63b8692b21c9"

+[[package]]
+name = "quick-error"
+version = "1.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0"
+
 [[package]]
 name = "quote"
 version = "1.0.46"
@@ -1763,6 +1828,12 @@ dependencies = [
 "proc-macro2",
 ]

+[[package]]
+name = "r-efi"
+version = "5.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
+
 [[package]]
 name = "r-efi"
 version = "6.0.0"
@@ -1806,6 +1877,16 @@ dependencies = [
 "rand_core 0.6.4",
 ]

+[[package]]
+name = "rand"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41"
+dependencies = [
+ "rand_chacha 0.9.0",
+ "rand_core 0.9.5",
+]
+
 [[package]]
 name = "rand"
 version = "0.10.2"
@@ -1837,6 +1918,16 @@ dependencies = [
 "rand_core 0.6.4",
 ]

+[[package]]
+name = "rand_chacha"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
+dependencies = [
+ "ppv-lite86",
+ "rand_core 0.9.5",
+]
+
 [[package]]
 name = "rand_core"
 version = "0.3.1"
@@ -1867,12 +1958,30 @@ dependencies = [
 "getrandom 0.2.17",
 ]

+[[package]]
+name = "rand_core"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c"
+dependencies = [
+ "getrandom 0.3.4",
+]
+
 [[package]]
 name = "rand_core"
 version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"

+[[package]]
+name = "rand_xorshift"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "513962919efc330f829edb2535844d1b912b0fbe2ca165d613e4e8788bb05a5a"
+dependencies = [
+ "rand_core 0.9.5",
+]
+
 [[package]]
 name = "randd"
 version = "0.1.0"
@@ -2151,6 +2260,18 @@ version = "1.0.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f"

+[[package]]
+name = "rusty-fork"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc6bf79ff24e648f6da1f8d1f011e9cac26491b619e6b9280f2b47f1774e6ee2"
+dependencies = [
+ "fnv",
+ "quick-error",
+ "tempfile",
+ "wait-timeout",
+]
+
 [[package]]
 name = "sb16d"
 version = "0.1.0"
@@ -2419,6 +2540,19 @@ dependencies = [
 "unicode-ident",
 ]

+[[package]]
+name = "tempfile"
+version = "3.27.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
+dependencies = [
+ "fastrand",
+ "getrandom 0.4.3",
+ "once_cell",
+ "rustix",
+ "windows-sys",
+]
+
 [[package]]
 name = "termcolor"
 version = "1.4.1"
@@ -2602,6 +2736,12 @@ dependencies = [
 "serde",
 ]

+[[package]]
+name = "unarray"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eaea85b334db583fe3274d12b4cd1880032beab409c0d774be044d4480ab9a94"
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.24"
@@ -2811,12 +2951,30 @@ dependencies = [
 "utf8parse 0.1.1",
 ]

+[[package]]
+name = "wait-timeout"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09ac3b126d3914f9849036f826e054cbabdc8519970b8998ddaf3b5bd3c65f11"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "wasi"
 version = "0.11.1+wasi-snapshot-preview1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"

+[[package]]
+name = "wasip2"
+version = "1.0.4+wasi-0.2.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487"
+dependencies = [
+ "wit-bindgen",
+]
+
 [[package]]
 name = "wasm-bindgen"
 version = "0.2.126"
@@ -3040,6 +3198,12 @@ version = "1.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0592e1c9d151f854e6fd382574c3a0855250e1d9b2f99d9281c6e6391af352f1"

+[[package]]
+name = "wit-bindgen"
+version = "0.57.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"
+
 [[package]]
 name = "xhcid"
 version = "0.1.0"
@@ -37,5 +37,8 @@ features = [
 ]
 #For debugging: "log", "verbose"

+[dev-dependencies]
+proptest = "1.11"
+
 [lints]
 workspace = true
@@ -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");
    }
 }