diff --git a/Cargo.lock b/Cargo.lock index 98a2ba2b0f..46252a3ac1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/netstack/Cargo.toml b/netstack/Cargo.toml index e2a5c954c1..a727cf22d4 100644 --- a/netstack/Cargo.toml +++ b/netstack/Cargo.toml @@ -37,5 +37,8 @@ features = [ ] #For debugging: "log", "verbose" +[dev-dependencies] +proptest = "1.11" + [lints] workspace = true diff --git a/netstack/src/worker_pool.rs b/netstack/src/worker_pool.rs index 3f5ada156f..39f7ea3144 100644 --- a/netstack/src/worker_pool.rs +++ b/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` 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, } -/// 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>, - receiver: Receiver, +/// 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) -> 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 { + 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 { + 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 { + handles: Vec>, + receiver: Receiver, + _phantom: std::marker::PhantomData, +} + +impl ReaderPool { + /// Spawn one worker thread per input. + pub fn spawn(inputs: Vec) -> Self { let (tx, rx) = channel::(); 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 = 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 Drop for ReaderPool { 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) { +fn worker_loop(worker_index: usize, mut input: R, tx: Sender) { 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) { mod tests { use super::*; use std::io::Write; + use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering}; use std::sync::Mutex; static FAKE: Mutex> = 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::>::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"); } }