socket export: the race's peer connects once a racer answered its last connection; a failed job reports the peer's verdict

sdk/toyos/src/net/socket.rs

@@ -201,12 +201,6 @@ fn call_on<R>(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<T>(slot: usize, end: End, wait: Wait, mut ask: impl FnMut(&dyn Fn()) -> Result<T, NetError>) -> Result<T, NetError> {
-    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<R>(&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<R>(&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<T>(bell: &impl Bell, wait: Wait, mut ask: impl FnMut(Drain<'_>) -> Result<T, NetError>) -> Result<T, NetError> {
+    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<T>(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<stream::Accepted, NetError> {
-    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<Received, NetError> {
     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<u32>,
+        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<alloc::vec::Vec<Event>>,
+        /// What arrives once an ask that found nothing is answered, before its holder reads
+        /// the answer.
+        after_nothing: Option<u32>,
+        taken: [Option<u32>; 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<u32, NetError> {
+            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<Server>,
+    }
+
+    impl Bell for Held<'_> {
+        fn look_now<R>(&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<Server>) -> Result<u32, NetError> {
+        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] {

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<Accepted, NetError> {
+pub(super) fn accept(listener: RawHandle, drain: super::socket::Drain<'_>) -> Result<Accepted, NetError> {
     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];

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");

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<std::net::SocketAddr, String> {
             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<std::net::TcpStream> = (0..2)
-            .map(|_| std::net::TcpStream::connect(("127.0.0.1", forwarded)).map_err(|e| format!("a racer's connect: {e}")))
-            .collect::<Result<_, _>>()?;
-        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<T>(heard: std::thread::JoinHandle<Result<T, String>>) -> 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";