diff --git a/issues/two-holders-racing-on-an-exported-socket-can-both-stall.md b/issues/two-holders-racing-on-an-exported-socket-can-both-stall.md deleted file mode 100644 index 1dc15e0ff62..00000000000 --- a/issues/two-holders-racing-on-an-exported-socket-can-both-stall.md +++ /dev/null @@ -1,27 +0,0 @@ ---- -status: open -kind: defect -opened: 2026-10-24 ---- - -# Two holders racing on an exported socket can both stall - -`netstack_holders_exported_to_children`'s last step (`tests/guest/rust-tests/src/bin/socket_export.rs`, -`race`) has the parent and a second child each block on a listener and a -datagram socket they both hold, while the harness connects twice and sends two -datagrams. Now and then neither process goes on: the racer is spawned -(`spawn: /system/bin/test_rs_socket_export`), the first part's line is printed, -and the guest says nothing more until the harness's ceiling (952–970 s guards -expired). - -QEMU x86-64 under TCG on an aarch64 Mac, the test alone, `--nocapture`: one -stall in six runs at `32046264b` (main), two in six at PR #1060's head, whose -change touches no socket or netstack code. Each stalled run's last guest lines -are the racer's spawn and netstack's mDNS line. Logs: PR #1060's evidence, -`qemu/netstack-*`. - -What blocks is not established: no guest line says whether either racer sent -its `racing` datagram, which the harness waits for before it connects. - -**Exit**: the race finishes in every run of a repeated QEMU loop, with the -blocked step named by a line the guest prints, or the cause found and fixed. diff --git a/sdk/toyos/src/net/socket.rs b/sdk/toyos/src/net/socket.rs index bd51e6a3bf0..447d7a1d165 100644 --- a/sdk/toyos/src/net/socket.rs +++ b/sdk/toyos/src/net/socket.rs @@ -201,12 +201,6 @@ fn call_on(handle: RawHandle, kind: Kind, f: impl FnOnce(usize, End) -> Resul call(handle, |slot, end| if end.kind == kind { f(slot, end) } else { Err(NetError::InvalidInput) }) } -/// The rings `bell` holds, read: a request follows at once. -fn drain(bell: RawHandle) { - let mut bytes = [0u8; 64]; - while let Ok(1..) = syscall::read_nonblock(bell, &mut bytes) {} -} - /// Whether `bell` is readable, waiting for it at most `nanos` on `poller`: rung, or hung up /// with netstack. Nothing is read. fn rung(poller: &Poller, bell: RawHandle, nanos: u64) -> bool { @@ -216,20 +210,80 @@ fn rung(poller: &Poller, bell: RawHandle, nanos: u64) -> bool { answered } -/// `ask` while the end's bell is rung, until it takes something, waiting for the bell between -/// as `wait` says. `ask` is handed [`drain`] for the bell, which it calls just before its -/// request goes. -fn look(slot: usize, end: End, wait: Wait, mut ask: impl FnMut(&dyn Fn()) -> Result) -> Result { - let drained = || drain(end.bell); +/// One request's read of the bell, handed out by [`look`] and spent once: spending it gives the +/// [`Drained`] that [`Drained::accept`] and [`Drained::receive`] send their request +/// with. Spend it after every step that can fail, so a call refused before its request leaves +/// the ring in the bell. A raw [`ipc`] send of [`ACCEPT`] or [`RECEIVE`] bypasses it. +pub(super) struct Drain<'a>(&'a dyn Fn()); + +impl Drain<'_> { + pub(super) fn now(self) -> Drained { + (self.0)(); + Drained(()) + } +} + +/// The bell read for one request, which sending the request spends. +#[must_use = "the bell was read for a request: send it"] +pub(super) struct Drained(()); + +impl Drained { + /// [`ACCEPT`] on `listener`, lending the stream's `region`. + pub(super) fn accept(self, listener: RawHandle, region: OwnedHandle) -> Result<(), ipc::IpcError> { + ipc::signal_with(listener, ACCEPT, [region]) + } + + /// [`RECEIVE`] on `connection`. + pub(super) fn receive(self, connection: RawHandle) -> Result<(), ipc::IpcError> { + ipc::signal(connection, RECEIVE) + } +} + +/// A holder's bell, as [`look`] reads it. +trait Bell { + /// `f` under the end's turn, told whether the bell is rung now. + fn look_now(&self, f: impl FnOnce(bool) -> R) -> R; + /// Until the bell is rung, or for at most `nanos`, out of the turn. + fn wait(&self, nanos: u64) -> Result<(), NetError>; + fn drain(&self); +} + +/// An end's own bell: looked at with the end's poller under its turn, waited on with a poller +/// of the wait's own, which the end's other calls do not share. +struct EndBell { + slot: usize, + bell: RawHandle, +} + +impl Bell for EndBell { + fn look_now(&self, f: impl FnOnce(bool) -> R) -> R { + let turn = TURNS[self.slot].lock(); + let Some(poller) = turn.as_ref() else { unreachable!("a live end has its poller") }; + f(rung(poller, self.bell, 0)) + } + + fn wait(&self, nanos: u64) -> Result<(), NetError> { + // Netstack gone hangs the bell up, which is readable, and the next look says so. + let poller = Poller::try_new(1).map_err(|_| NetError::ResourceExhausted)?; + rung(&poller, self.bell, nanos); + Ok(()) + } + + fn drain(&self) { + let mut bytes = [0u8; 64]; + while let Ok(1..) = syscall::read_nonblock(self.bell, &mut bytes) {} + } +} + +/// `ask` while `bell` is rung, until it takes something, waiting for the bell between as +/// `wait` says. Each `ask` is handed one [`Drain`]. +fn look(bell: &impl Bell, wait: Wait, mut ask: impl FnMut(Drain<'_>) -> Result) -> Result { + let drain = || bell.drain(); loop { - { - let turn = TURNS[slot].lock(); - let Some(poller) = turn.as_ref() else { unreachable!("a live end has its poller") }; - if rung(poller, end.bell, 0) { - match ask(&drained) { - Err(NetError::WouldBlock) => {} - answer => return answer, - } + if let Some(answer) = bell.look_now(|rung| rung.then(|| ask(Drain(&drain)))) { + match answer { + Err(NetError::WouldBlock) => {} + answer => return answer, } } let until = match wait { @@ -240,11 +294,7 @@ fn look(slot: usize, end: End, wait: Wait, mut ask: impl FnMut(&dyn Fn()) -> if left == 0 { return Err(NetError::TimedOut); } - // Out of the turn, which the end's other calls take meanwhile, so on a poller of the - // wait's own. Netstack gone hangs the bell up, which is readable, and the next look says - // so. - let poller = Poller::try_new(1).map_err(|_| NetError::ResourceExhausted)?; - rung(&poller, end.bell, left); + bell.wait(left)?; } } @@ -365,7 +415,7 @@ pub fn let_go(handle: RawHandle) { /// The oldest connection waiting at the listener `handle` reaches, as a stream: waiting for one /// as `wait` says, [`NetError::WouldBlock`] or [`NetError::TimedOut`] where none came. pub fn accept(handle: RawHandle, wait: Wait) -> Result { - call_on(handle, Kind::Listener, |slot, end| look(slot, end, wait, |drain| stream::accept(end.connection, drain))) + call_on(handle, Kind::Listener, |slot, end| look(&EndBell { slot, bell: end.bell }, wait, |drain| stream::accept(end.connection, drain))) } // ---- a datagram socket's ---- @@ -395,9 +445,8 @@ pub fn send_to(handle: RawHandle, addr: [u8; 4], port: u16, data: &[u8]) -> Resu /// `wait` says, [`NetError::WouldBlock`] or [`NetError::TimedOut`] where none came. pub fn recv_from(handle: RawHandle, buf: &mut [u8], wait: Wait) -> Result { call_on(handle, Kind::Datagram, |slot, end| { - look(slot, end, wait, |drain| { - drain(); - ipc::signal(end.connection, RECEIVE).map_err(hangup)?; + look(&EndBell { slot, bell: end.bell }, wait, |drain| { + drain.now().receive(end.connection).map_err(hangup)?; let header = answer(end.connection)?; let mut frame = [0u8; MAX_FRAME]; let n = ipc::recv_bytes(end.connection, &header, &mut frame).map_err(hangup)?; @@ -558,6 +607,116 @@ mod tests { assert!(datagram_of(&[0; MAX_FRAME + 1]).is_none()); } + /// netstack's side of two holders' bells over one queue (`toyos-net-node`'s `holders`): + /// each holder is rung while something waits and it has not been rung since it last asked, + /// and every ask, whatever it takes, is followed by that telling. + #[derive(Default)] + struct Server { + queue: alloc::collections::VecDeque, + told: [bool; 2], + /// The rings in each holder's bell, unread. + bells: [u32; 2], + /// What happens while holder 1 waits, a batch a wait. + while_waiting: alloc::collections::VecDeque>, + /// What arrives once an ask that found nothing is answered, before its holder reads + /// the answer. + after_nothing: Option, + taken: [Option; 2], + } + + enum Event { + Arrives(u32), + /// Holder 0 makes its whole call, which must take something. + Looks, + } + + impl Server { + fn tell(&mut self) { + for holder in 0..2 { + if !self.queue.is_empty() && !self.told[holder] { + self.bells[holder] += 1; + self.told[holder] = true; + } + } + } + + fn arrives(&mut self, item: u32) { + self.queue.push_back(item); + self.tell(); + } + + fn ask(&mut self, holder: usize, _: Drained) -> Result { + self.told[holder] = false; + let taken = self.queue.pop_front(); + self.tell(); + if taken.is_none() { + if let Some(item) = self.after_nothing.take() { + self.arrives(item); + } + } + taken.ok_or(NetError::WouldBlock) + } + } + + struct Held<'a> { + holder: usize, + server: &'a core::cell::RefCell, + } + + impl Bell for Held<'_> { + fn look_now(&self, f: impl FnOnce(bool) -> R) -> R { + let rung = self.server.borrow().bells[self.holder] > 0; + f(rung) + } + + /// Runs what the scenario says happens meanwhile; a holder still unrung when nothing is + /// left to happen would wait for good. + fn wait(&self, _: u64) -> Result<(), NetError> { + while self.server.borrow().bells[self.holder] == 0 { + let Some(batch) = self.server.borrow_mut().while_waiting.pop_front() else { + panic!("holder {} waits for good with {:?} queued", self.holder, self.server.borrow().queue) + }; + for event in batch { + match event { + Event::Arrives(item) => self.server.borrow_mut().arrives(item), + Event::Looks => { + let taken = call(0, self.server).expect("holder 0 takes what it was rung for"); + self.server.borrow_mut().taken[0] = Some(taken); + } + } + } + } + Ok(()) + } + + fn drain(&self) { + self.server.borrow_mut().bells[self.holder] = 0; + } + } + + fn call(holder: usize, server: &core::cell::RefCell) -> Result { + look(&Held { holder, server }, Wait::Blocking(None), |drain| { + let drained = drain.now(); + server.borrow_mut().ask(holder, drained) + }) + } + + /// Two holders blocked on one queue: both are rung for the first arrival, the other takes + /// it, and holder 1's ask finds nothing. The second arrives as that answer goes, before + /// holder 1 reads it, so its ring is already in the bell: a holder that read its bell after + /// asking would wait for good. + #[test] + fn the_holder_that_found_nothing_takes_the_next_arrival() { + let server = core::cell::RefCell::new(Server { + while_waiting: [alloc::vec![Event::Arrives(1), Event::Looks]].into(), + after_nothing: Some(2), + ..Server::default() + }); + let taken = call(1, &server).expect("holder 1 takes the second arrival"); + assert_eq!((server.borrow().taken[0], taken), (Some(1), 2)); + assert!(server.borrow().queue.is_empty()); + } + #[test] fn a_kind_is_its_word_and_no_other_word_is_one() { for kind in [Kind::Listener, Kind::Datagram] { diff --git a/sdk/toyos/src/net/stream.rs b/sdk/toyos/src/net/stream.rs index 8bb075deae4..c1314584fad 100644 --- a/sdk/toyos/src/net/stream.rs +++ b/sdk/toyos/src/net/stream.rs @@ -742,13 +742,12 @@ pub struct Accepted { /// The oldest connection waiting at the listener whose holder's connection is /// `listener`, as a stream: its region lent with a `super::socket::ACCEPT` and -/// its connection and bells in the answer, `drain` called just before the +/// its connection and bells in the answer, `drain` spent just before the /// request goes, once the slot and the region are made. Refused as [`connect`] /// is, and [`NetError::WouldBlock`] where none waits. -pub(super) fn accept(listener: RawHandle, drain: &dyn Fn()) -> Result { +pub(super) fn accept(listener: RawHandle, drain: super::socket::Drain<'_>) -> Result { let opened = open(false, ([0; 4], 0), |region| { - drain(); - ipc::signal_with(listener, super::socket::ACCEPT, [region]).map_err(hangup)?; + drain.now().accept(listener, region).map_err(hangup)?; let header = answer(listener)?; let accepted: TcpAcceptResponse = ipc::recv_payload(listener, &header).map_err(hangup)?; let mut handles = [toyos_abi::HANDLE_INVALID; syscall::MAX_TRANSFER_HANDLES]; diff --git a/system/netstack/node/tests/datagram.rs b/system/netstack/node/tests/datagram.rs index a55603c654a..d61de97c63f 100644 --- a/system/netstack/node/tests/datagram.rs +++ b/system/netstack/node/tests/datagram.rs @@ -356,6 +356,8 @@ fn every_holder_receives_from_one_queue_each_datagram_once_and_whole() { assert_eq!(&out[..got.len], b"two, longer", "the next datagram, whole, to the other holder"); assert_eq!(lan.node.udp_recv_from(id, second, &mut out), Ok(None), "each datagram went once"); assert_eq!(rings(), (1, 2), "nothing left, no ring"); + lan.deliver(&udp(MAC, MAC_B, (B, 9_999), (A, 4_000), b"three")); + assert_eq!(rings(), (2, 3), "both looked since, the one that found nothing too, and both are told"); lan.node.datagram_departed(lan.now, id, second); assert_eq!(lan.node.udp_recv_from(id, second, &mut out), Err(Refused::NotConnected), "a holder that left takes nothing"); diff --git a/tests/toyos.rs b/tests/toyos.rs index d5173fa98a1..56c71fdc049 100644 --- a/tests/toyos.rs +++ b/tests/toyos.rs @@ -5347,7 +5347,10 @@ fn crypto_instrument_prints_its_table(test_config: &Path) -> Result<(), String> const CRYPTO_JOB_CEILING_SECS: u64 = 300; /// The guest port `test_rs_socket_export` listens on, which the boot forwards -/// a port of the host's loopback to. +/// a port of the host's loopback to. QEMU's user network listens on that host +/// port with a queue of one, and a macOS host resets a connect that finds it +/// full, so the peer connects only once the guest has answered its last +/// connection. const SOCKET_EXPORT_PORT: u16 = 7070; /// `test_rs_stream_export` and `test_rs_socket_export` on one boot of @@ -5395,7 +5398,7 @@ fn stream_exported_to_a_child(qemu: &mut QemuInstance) -> Result<(), String> { }); let result = qemu.run_test(&format!("test_rs_stream_export {port}"), Duration::from_secs(120)); if let Some(why) = &result.error { - return Err(format!("{why}\nthe job said:\n{}", result.stdout)); + return Err(format!("{why}\n{}\nthe job said:\n{}", peer_verdict(heard), result.stdout)); } if result.exit_code != Some(0) { return Err(format!("the job ended {:?}:\n{}", result.exit_code, result.stdout)); @@ -5411,8 +5414,9 @@ fn stream_exported_to_a_child(qemu: &mut QemuInstance) -> Result<(), String> { /// `test_rs_socket_export`: for the child and then the parent, a datagram /// saying it is ready, answered by one saying go, and a connection through the /// forwarded port that it accepts and writes its line on; then, once the parent -/// and a second child both say they are about to block, two connections and -/// two datagrams, each taken once. +/// and a second child both say they are about to block, two connections, the +/// second once a racer wrote its line on the first, and two datagrams, each +/// taken once. fn sockets_exported_to_a_child(qemu: &mut QemuInstance, forwarded: u16) -> Result<(), String> { use std::io::Read; /// A liveness ceiling on each of the peer's waits, never a timing. @@ -5437,8 +5441,6 @@ fn sockets_exported_to_a_child(qemu: &mut QemuInstance, forwarded: u16) -> Resul return Err(format!("the peer heard {:?}, not {owed:?}", String::from_utf8_lossy(&line))); } } - // The race: both processes say they are about to block, then two - // connections and two datagrams, each taken once. let both = |what: &str| -> Result { let mut said = Vec::new(); let mut from = None; @@ -5455,10 +5457,8 @@ fn sockets_exported_to_a_child(qemu: &mut QemuInstance, forwarded: u16) -> Resul from.ok_or_else(|| "no racer spoke".to_string()) }; both("racing")?; - let conns: Vec = (0..2) - .map(|_| std::net::TcpStream::connect(("127.0.0.1", forwarded)).map_err(|e| format!("a racer's connect: {e}"))) - .collect::>()?; - for mut conn in conns { + for _ in 0..2 { + let mut conn = std::net::TcpStream::connect(("127.0.0.1", forwarded)).map_err(|e| format!("a racer's connect: {e}"))?; conn.set_read_timeout(Some(CEILING)).map_err(|e| e.to_string())?; let mut line = Vec::new(); conn.read_to_end(&mut line).map_err(|e| format!("a racer's line: {e} after {line:?}"))?; @@ -5474,7 +5474,7 @@ fn sockets_exported_to_a_child(qemu: &mut QemuInstance, forwarded: u16) -> Resul }); let result = qemu.run_test(&format!("test_rs_socket_export {port}"), Duration::from_secs(300)); if let Some(why) = &result.error { - return Err(format!("{why}\nthe job said:\n{}", result.stdout)); + return Err(format!("{why}\n{}\nthe job said:\n{}", peer_verdict(heard), result.stdout)); } if result.exit_code != Some(0) { return Err(format!("the job ended {:?}:\n{}", result.exit_code, result.stdout)); @@ -5490,6 +5490,19 @@ fn sockets_exported_to_a_child(qemu: &mut QemuInstance, forwarded: u16) -> Resul heard.join().map_err(|_| "the peer's thread panicked".to_string())? } +/// What the peer's thread had come to when its job failed: a guest that waits +/// for a step the peer could not take is no stall of the guest's. +fn peer_verdict(heard: std::thread::JoinHandle>) -> String { + if !heard.is_finished() { + return "the peer was still waiting on the guest".into(); + } + match heard.join() { + Ok(Ok(_)) => "the peer had done its part".into(), + Ok(Err(why)) => format!("the peer failed first: {why}"), + Err(_) => "the peer's thread panicked".into(), + } +} + /// `test_rs_udp_recv_timeout` on `tests/images/netcase.toml`'s boot. fn udp_receive_times_out() -> Result<(), String> { const JOB: &str = "udp_recv_timeout";