diff --git a/issues/closing-a-pipe-end-ends-every-holders-polls-on-it.md b/issues/closing-a-pipe-end-ends-every-holders-polls-on-it.md new file mode 100644 index 00000000000..2e467ff1557 --- /dev/null +++ b/issues/closing-a-pipe-end-ends-every-holders-polls-on-it.md @@ -0,0 +1,24 @@ +--- +status: open +kind: defect +opened: 2026-10-11 +--- + +# Closing a pipe end ends every holder's polls on it + +`ops::close` (`kernel/src/object/ops.rs`) cancels every poll on the watches of +the object the closed handle names: a pipe end's data watches and its leaving +watch, and those of a connection, an acceptor, a file and the other objects +`close_ends_polls` lists. `Watch::cancel_rings` (`kernel/pure/sched/watch.rs`) +takes every ring registered on the watch, whoever registered it. So when two +handles name one pipe end — a duplicate, or a handle moved to another process +that kept one — closing either answers the other's poll on its live handle +with `NotFound`, as though that handle had been closed. Its holder reads a +watch it still needs as over. + +Found by reading, in review of pull request #861, which put the pipe end's +leaving watch in that same loop. No test or machine has shown it. + +**Exit**: a close ends only the polls made through the handle it closes, and a +guest test that holds a duplicate of a pipe end, watches it, and closes the +other handle reads no answer on it. diff --git a/issues/netstacks-latency-under-a-receive-batch-over-64-frames-is-unmeasured.md b/issues/netstacks-latency-under-a-receive-batch-over-64-frames-is-unmeasured.md new file mode 100644 index 00000000000..1997ed22945 --- /dev/null +++ b/issues/netstacks-latency-under-a-receive-batch-over-64-frames-is-unmeasured.md @@ -0,0 +1,35 @@ +--- +status: open +kind: tooling +opened: 2026-10-12 +--- + +# netstack's latency under a receive batch of more than 64 frames is unmeasured + +`Node::receive` takes at most the card's ring in one batch (`Card::receive`, +`system/netstack/src/card.rs`): `toyos_i219::RX_RING - 1`, 999 frames, on the +I219, and the queue's 256 on virtio. Before pull request #861 it took 64. Every +frame of a batch reaches the stack under one `now`, read before the batch +(`system/netstack/src/main.rs`'s loop), and the deadlines that are due, the +transmit opportunity and the clients' answers run only after the batch's last +frame. So a backlog that was served as several 64-frame passes, with those +three between them, is now one pass: a frame late in it is stamped with the +batch's start, and a deadline due during it fires after it. + +Nothing moves while a pass brings 64 frames or fewer, and nothing has read a +pass that brings more. The T14's LAN round trip was read only under one +gigabit download of full-sized frames, at about 12 frames a pass (round 4 of +pull request #861: 59.7 passes per MB, and some 700 full-sized frames carry a +MB; derived). A pass brings more than 64 when netstack is about 0.79 ms behind +full-sized frames at a gigabit (64 frames of 12.3 µs on the wire, derived), or +under minimum-sized frames, of which a gigabit carries about 195 in one +131 µs throttle interval (672 ns a frame on the wire, derived). + +**Exit**: a T14 reading, in one session, of the LAN round trip and of how late +netstack fires a due deadline, under a load that brings more than 64 frames a +pass, with the ring's budget and with a budget of 64, recorded in the commit +that closes this. A gap between the two is investigated where it shows, and is +no verdict on the budget by itself. + +**Owner**: whoever holds `issues/the-lan-is-not-yet-production-grade.md`, whose +stage 2 sets the ring sizes and the interrupt moderation this batch follows. diff --git a/issues/netstacks-receiver-round-trip-reads-no-finer-than-a-millisecond.md b/issues/netstacks-receiver-round-trip-reads-no-finer-than-a-millisecond.md new file mode 100644 index 00000000000..ef6eb67f355 --- /dev/null +++ b/issues/netstacks-receiver-round-trip-reads-no-finer-than-a-millisecond.md @@ -0,0 +1,22 @@ +--- +status: open +kind: defect +opened: 2026-10-11 +--- + +# netstack's receiver round trip reads no finer than a millisecond + +A downloading stream's round trip is the receiver's own estimate +(`system/netstack/shard/tcp/src/rx.rs`): the age of an echoed TSval. TSval +ticks once a millisecond, so an echo's age is a whole number of milliseconds +and never less than one tick. A path whose round trip is under a millisecond, +a LAN's, reads as at least 1 ms. + +What that moves is the receive buffer's growth, twice what the user read per +round trip (`rx.rs`): a 200 µs path that reads 1 ms grows its buffer to five +times what it needs. That figure is derived from the tick and the growth rule. +No machine has measured a sub-millisecond path. + +**Exit**: a stream on a sub-millisecond path grows its receive buffer to twice +what the user reads in the path's own round trip, read by a host test that +delivers echoes on such a path. diff --git a/issues/the-lan-is-not-yet-production-grade.md b/issues/the-lan-is-not-yet-production-grade.md index 3785e7e0efe..40b956a0d6f 100644 --- a/issues/the-lan-is-not-yet-production-grade.md +++ b/issues/the-lan-is-not-yet-production-grade.md @@ -40,7 +40,7 @@ why, and the boot still leases. `issues/no-machine-has-read-an-intel-nic-across-a-link-change.md` waits on it. - **Throughput** is measured against Ubuntu on the same machine and cable, with line rate (about 940 Mb/s of TCP) as the target: interrupt moderation (a - fixed 128 µs throttle today, not adaptive), ring sizes, checksum and + fixed throttle today, not adaptive), ring sizes, checksum and segmentation offload. - **The part is handed back** in the state the next operating system and the management firmware that shares it expect. Ubuntu has had link after every diff --git a/kernel/src/inbox/mod.rs b/kernel/src/inbox/mod.rs index 32310a66ae0..d542b06e522 100644 --- a/kernel/src/inbox/mod.rs +++ b/kernel/src/inbox/mod.rs @@ -536,9 +536,10 @@ fn arm( let ready = readiness_of(object, flags).any(); let read = if flags.readable() { ops::read_watch(object) } else { None }; let write = if flags.writable() { ops::write_watch(object) } else { None }; - // A pipe end's own watch is the one its other end's last holder posts; - // an object with no other end is refused the question, whatever else the - // watch asks. A direction asked above has taken that watch already. + // A pipe end's leaving watch is posted by its other end's last holder + // alone; an object with no other end is refused the question, whatever + // else the watch asks. A direction asked above needs no second entry: that + // last holder posts the end's own data watch too. let own = match flags.other_end_gone().then(|| ops::pipe_end_watch(object)) { Some(None) => return Err(SyscallError::NotSupported), Some(Some(_)) if read.is_some() || write.is_some() => None, diff --git a/kernel/src/object/ops.rs b/kernel/src/object/ops.rs index 6981fe0e48a..a17b15c32a7 100644 --- a/kernel/src/object/ops.rs +++ b/kernel/src/object/ops.rs @@ -188,7 +188,7 @@ pub fn close( // Only a watch this handle's object ends is answered: one it shares (every `Console` and a // `Device(Keyboard)` name the keyboard's) cannot cancel another's poll. if close_ends_polls(&object) { - for watch in [read_watch(&object), write_watch(&object)].into_iter().flatten() { + for watch in [read_watch(&object), write_watch(&object), pipe_end_watch(&object)].into_iter().flatten() { watch.cancel_polls(); } } @@ -315,12 +315,12 @@ pub fn write_watch(object: &KObjectRef) -> Option { } } -/// The watch of a pipe end, which its other end's last holder posts as it -/// lets go; `None` for an object that is no pipe end. +/// A pipe end's leaving watch, which only its other end's last holder posts as +/// it lets go; `None` for an object that is no pipe end. pub fn pipe_end_watch(object: &KObjectRef) -> Option { match object { - KObjectRef::PipeRead(_) => read_watch(object), - KObjectRef::PipeWrite(_) => write_watch(object), + KObjectRef::PipeRead(r) => pipe::left_watch(r.id(), End::Read).map(WatchRef::Shared), + KObjectRef::PipeWrite(w) => pipe::left_watch(w.id(), End::Write).map(WatchRef::Shared), KObjectRef::Connection(_) | KObjectRef::Acceptor(_) | KObjectRef::Process(_) | KObjectRef::Console(_) | KObjectRef::Device(_) | KObjectRef::SysCap(_) | KObjectRef::File(_) | KObjectRef::Inbox(_) | KObjectRef::Connector(_) diff --git a/kernel/src/pipe.rs b/kernel/src/pipe.rs index de41909970d..396c50dcf5b 100644 --- a/kernel/src/pipe.rs +++ b/kernel/src/pipe.rs @@ -117,6 +117,10 @@ struct Pipe { /// poll registration can clone one out from under the table lock and hold it across its own park. read_watch: Arc, write_watch: Arc, + /// Each end's leaving watch, by [`End`]: an `OTHER_END_GONE` poll on that + /// end sits on it, and only the other end's last holder posts it — a byte + /// moved through the pipe wakes no poll that asked about a leaving. + left: [Arc; 2], /// Set when a write should hand the next reader transient RT priority — covers readers already runnable, which the wake-time boost misses. rt_boost_pending: bool, } @@ -132,6 +136,7 @@ impl Pipe { holders: Holders::new(), read_watch: Arc::new(Watch::new()), write_watch: Arc::new(Watch::new()), + left: [Arc::new(Watch::new()), Arc::new(Watch::new())], rt_boost_pending: false, } } @@ -283,23 +288,33 @@ pub fn set_rt_boost_pending(pipe_id: PipeId) { }); } -/// One reference to `end` lets go; its end's last posts the other end's watch. +/// One reference to `end` lets go; its end's last posts the other end's watch +/// and its leaving watch. fn release(pipe_id: PipeId, end: End) { - let released = with_pipes_mut(|pipes| { + let (released, left) = with_pipes_mut(|pipes| { let pipe = pipes.get_mut(pipe_id).expect("release: pipe not found"); let released = pipe.holders.release(end); pipe.publish_ends(); + let left = pipe.left[slot(end.other())].clone(); if released == Released::PipeGone { let pipe = pipes.remove(pipe_id).unwrap(); free_pipe(pipe); } - released + (released, left) }); match (released, end) { - (Released::Held | Released::PipeGone, _) => {} + (Released::Held | Released::PipeGone, _) => return, (Released::EndGone, End::Read) => crate::scheduler::wake_pipe_writers(pipe_id), (Released::EndGone, End::Write) => crate::scheduler::wake_pipe_readers(pipe_id), } + left.post(); +} + +fn slot(end: End) -> usize { + match end { + End::Read => 0, + End::Write => 1, + } } fn free_pipe(pipe: Pipe) { @@ -315,3 +330,9 @@ pub fn read_watch(pipe_id: PipeId) -> Option> { pub fn write_watch(pipe_id: PipeId) -> Option> { with_pipes(|pipes| pipes.get(pipe_id).map(|p| p.write_watch.clone())) } + +/// `end`'s leaving watch, the same way: posted once, by the other end's last +/// holder. +pub fn left_watch(pipe_id: PipeId, end: End) -> Option> { + with_pipes(|pipes| pipes.get(pipe_id).map(|p| p.left[slot(end)].clone())) +} diff --git a/system/netstack/i219/src/lib.rs b/system/netstack/i219/src/lib.rs index a7983803ce5..8569405645c 100644 --- a/system/netstack/i219/src/lib.rs +++ b/system/netstack/i219/src/lib.rs @@ -148,25 +148,39 @@ pub trait Interrupts { fn taken(&self) -> Result; } -/// How many receive descriptors the ring holds. §7.1.8: `RDLEN` "must be a -/// multiple of 128", so the count is a multiple of eight. -pub const RX_RING: usize = 256; +/// How much memory the driver asks to be granted, in one region: one 2 MiB +/// span, which is what the kernel maps a grant of any smaller size in, so +/// every byte of it is the receive ring's that the transmit ring does not +/// need. +pub const GRANT_BYTES: u64 = 2 << 20; +/// How many receive descriptors the ring holds: the most whose descriptors +/// and buffers fit the grant beside the transmit ring's. §7.1.8: `RDLEN` +/// "must be a multiple of 128", so the count is a multiple of eight. +pub const RX_RING: usize = 1000; /// How many transmit descriptors the ring holds. A frame is handed to the /// device the moment it is filled and reclaimed when room is next asked for, /// so what this bounds is how many may be in flight at once: one fewer than /// this, [`I219::tx_room`]'s most. pub const TX_RING: usize = 16; -/// `ITR` (§10.2.4.2): the least interval between two of the function's -/// interrupts, in 256 ns units — 128 µs. It runs from the last interrupt, so a -/// frame after a quiet spell that long interrupts at once, and a gigabit -/// stream's full-sized frames, one each 12.3 µs, wake their reader about ten at -/// a time. -const THROTTLE: u32 = 500; +/// Free receive descriptors at which the part reports itself short: +/// `RCTL.RDMTS = 00b`, "1/2" of `RDLEN` (§10.2.5.1), raising `RXDMT0`. +const RX_SHORT: usize = RX_RING / 2; +/// Receive buffers given back and not yet handed to the part at most: the +/// tail moves once a batch ([`I219::rx_flush`]), and sooner once this many +/// wait. A quarter of [`RX_SHORT`], so what a batch holds back never brings +/// the part to its report alone. +pub const HELD_MAX: usize = RX_SHORT / 4; /// Bytes per receive buffer — `RCTL.BSIZE = 00b` with `BSEX` clear (§10.2.5.1). pub const RX_BUF_BYTES: usize = 2048; /// Bytes per transmit buffer, and therefore the longest frame [`I219::tx_reserve`] /// takes. pub const TX_BUF_BYTES: usize = 2048; +/// `ITR` (§10.2.4.2): the least interval between two of the function's +/// interrupts, in 256 ns units — 131 µs. It runs from the last interrupt, so a +/// frame after a quiet spell that long interrupts at once, and a gigabit +/// stream's full-sized frames, one each 12.3 µs, wake their reader about ten at +/// a time. +const THROTTLE: u32 = 512; const _: () = { // §7.1.8 and §7.2.4: both ring lengths are programmed in bytes and "must be @@ -181,20 +195,30 @@ const _: () = { /// The grant's layout. One grant, because every byte of it is this process's /// own: what a wrong offset here corrupts is this driver's memory and nobody -/// else's. +/// else's. Each buffer starts on a multiple of its own size. const OFF_RX_RING: usize = 0x0000; const OFF_TX_RING: usize = OFF_RX_RING + RX_RING * rx_desc::BYTES; -const OFF_RX_BUFS: usize = 0x2000; +const OFF_RX_BUFS: usize = layout(RX_RING).0; const OFF_TX_BUFS: usize = OFF_RX_BUFS + RX_RING * RX_BUF_BYTES; -/// How much memory the driver needs granted, in one region. -pub const GRANT_BYTES: u64 = (OFF_TX_BUFS + TX_RING * TX_BUF_BYTES) as u64; + +/// Where the receive buffers start, and where the grant's last byte ends, for +/// a receive ring of `rx` descriptors. +const fn layout(rx: usize) -> (usize, usize) { + let rings = rx * rx_desc::BYTES + TX_RING * tx_desc::BYTES; + let rx_bufs = rings.div_ceil(RX_BUF_BYTES) * RX_BUF_BYTES; + (rx_bufs, rx_bufs + rx * RX_BUF_BYTES + TX_RING * TX_BUF_BYTES) +} const _: () = { assert!(OFF_TX_RING + TX_RING * tx_desc::BYTES <= OFF_RX_BUFS); + // The grant holds this ring and not the next multiple of eight. + assert!(layout(RX_RING).1 <= GRANT_BYTES as usize); + assert!(layout(RX_RING + 8).1 > GRANT_BYTES as usize); // §7.1.8 and §7.2.4: the ring base "is aligned on a 16-byte boundary" and // "hardware ignores the lower 4 bits", so a ring at an unaligned offset // would be programmed somewhere else entirely. assert!(OFF_RX_RING.is_multiple_of(16) && OFF_TX_RING.is_multiple_of(16)); + assert!(HELD_MAX > 0); }; /// Why the function was not brought up. Each keeps its own word: a caller asks @@ -765,6 +789,8 @@ pub struct I219 { /// receive ring keeps in software's hands (§7.1.8: the queue is empty when /// head equals tail). rx_tail: usize, + /// Buffers given back since the tail last moved. + rx_held: usize, /// Next transmit descriptor to hand out, which is also `TDT`. tx_next: usize, /// Next transmit descriptor to reclaim. @@ -913,6 +939,7 @@ impl I219 { link_up_at: None, rx_next: 0, rx_tail: RX_RING - 1, + rx_held: 0, tx_next: 0, tx_clean: 0, tx_wake: false, @@ -1253,19 +1280,20 @@ impl I219 { } /// Give a frame's buffer back to the device, its bytes having been read. + /// It reaches the device at the next [`Self::rx_flush`], or here once + /// [`HELD_MAX`] buffers wait. pub fn rx_done(&mut self, frame: Frame) { self.give_back(frame.index); } - /// Republish descriptor `index` and move the tail over every ready - /// descriptor from where it stands. - /// - /// **The tail moves over a run, never to an index.** §7.1.8's tail - /// "identifies the location beyond the last descriptor hardware can - /// process", so writing it with the descriptor that came back last would - /// hand the device every descriptor between — including one whose bytes - /// their reader still holds — and, for a buffer returned before an older - /// one, walk `RDT` backwards over the rest of the ring. + /// Hand the device every buffer given back since the tail last moved: the + /// end of a batch, and one `RDT` write for all of it. + pub fn rx_flush(&mut self) { + self.move_tail(); + } + + /// Republish descriptor `index`, and move the tail once [`HELD_MAX`] + /// buffers wait for it. /// /// **Ready is read out of the descriptor, so there is no second record of /// it to disagree.** [`publish_rx`] zeroes the status word, and a @@ -1274,6 +1302,22 @@ impl I219 { /// touches neither until the tail passes it. fn give_back(&mut self, index: usize) { self.publish_rx(index); + self.rx_held += 1; + if self.rx_held >= HELD_MAX { + self.move_tail(); + } + } + + /// Move the tail over every ready descriptor from where it stands. + /// + /// **The tail moves over a run, never to an index.** §7.1.8's tail + /// "identifies the location beyond the last descriptor hardware can + /// process", so writing it with the descriptor that came back last would + /// hand the device every descriptor between — including one whose bytes + /// their reader still holds — and, for a buffer returned before an older + /// one, walk `RDT` backwards over the rest of the ring. + fn move_tail(&mut self) { + self.rx_held = 0; let mut tail = self.rx_tail; loop { let next = (tail + 1) % RX_RING; diff --git a/system/netstack/i219/src/tests.rs b/system/netstack/i219/src/tests.rs index 7a197b77a24..57e67d4fcd2 100644 --- a/system/netstack/i219/src/tests.rs +++ b/system/netstack/i219/src/tests.rs @@ -47,6 +47,7 @@ fn drain(nic: &Nic, driver: &mut Driver) -> Vec> { for frame in taken { driver.rx_done(frame); } + driver.rx_flush(); got } @@ -123,10 +124,11 @@ fn bring_up_programs_what_the_initialization_sections_name() { assert_eq!(nic.peek(regs::TDT), 0); // Both timers off and the throttle on: the first frame after a quiet - // spell is not held back, and a stream's are taken a batch a wake. + // spell is not held back, and no peer's frames wake the reader more often + // than the throttle. assert_eq!(nic.peek(regs::RDTR), 0); assert_eq!(nic.peek(regs::RADV), 0); - assert_eq!(u64::from(nic.peek(regs::ITR)) * 256, 128_000, "§10.2.4.2 counts in 256 ns"); + assert_eq!(u64::from(nic.peek(regs::ITR)) * 256, 131_072, "§10.2.4.2 counts in 256 ns"); assert_eq!(driver.mac(), NVM_MAC); } @@ -528,6 +530,7 @@ fn a_buffer_given_back_out_of_turn_does_not_carry_the_tail_over_an_older_one() { assert_eq!(armed as usize, RX_RING - 1, "{}", nic.because("the ring did not arm its tail")); driver.rx_done(second); + driver.rx_flush(); assert_eq!( nic.peek(regs::RDT), armed, @@ -536,6 +539,7 @@ fn a_buffer_given_back_out_of_turn_does_not_carry_the_tail_over_an_older_one() { ); driver.rx_done(first); + driver.rx_flush(); assert_eq!( nic.peek(regs::RDT), 1, @@ -574,6 +578,7 @@ fn a_refused_descriptor_does_not_carry_the_tail_over_a_frame_still_held() { nic.because("the refused descriptor was handed up as a frame") ); assert_eq!(driver.counters().over_length, 1, "{}", nic.because("the length was believed")); + driver.rx_flush(); assert_eq!( nic.peek(regs::RDT) as usize, RX_RING - 1, @@ -582,6 +587,7 @@ fn a_refused_descriptor_does_not_carry_the_tail_over_a_frame_still_held() { ); driver.rx_done(held); + driver.rx_flush(); assert_eq!( nic.peek(regs::RDT), 1, @@ -590,6 +596,96 @@ fn a_refused_descriptor_does_not_carry_the_tail_over_a_frame_still_held() { ); } +/// The ring that fills the grant takes frames whole and in order through its wrap, twice over, +/// in batches of up to a third of it with a flush at each end. +#[test] +fn frames_come_up_whole_and_in_order_through_the_rings_wrap() { + let nic = Nic::new(26); + let mut driver = open(&nic); + nic.set_link(true); + let mut sent = Vec::new(); + let mut got = Vec::new(); + for batch in 0..8 { + for i in 0..RX_RING / 3 { + let tag = (batch * RX_RING / 3 + i) as u32; + let mut bytes = frame(tag as u8, 64 + (tag % 1400) as usize); + bytes[7..11].copy_from_slice(&tag.to_le_bytes()); + nic.deliver(&bytes); + sent.push(bytes); + } + for _ in 0..RX_RING { + nic.run(); + one_pass(&mut driver); + got.extend(drain(&nic, &mut driver)); + if got.len() == sent.len() { + break; + } + } + } + assert!(sent.len() > 2 * RX_RING, "the ring did not wrap twice"); + assert_eq!(got.len(), sent.len(), "{}", nic.because("frames were lost or invented")); + assert!(got == sent, "{}", nic.because("the frames did not come up whole and in order")); +} + +/// How many times the driver has written `reg`. +fn writes_of(nic: &Nic, reg: usize) -> usize { + nic.trace().iter().filter(|access| **access == Access::Write { reg }).count() +} + +/// A batch's buffers reach the part in one `RDT` write at its end, and sooner +/// only once [`HELD_MAX`] wait: the part is never kept from more than that. +#[test] +fn the_tail_moves_once_a_batch_and_sooner_only_once_held_max_wait() { + let nic = Nic::with( + 24, + Part::E82574, + Permits { batched_writeback: false, spurious_interrupts: false, ..Permits::default() }, + ); + let mut driver = open(&nic); + nic.set_link(true); + let late = 3; + for i in 0..HELD_MAX + late { + nic.deliver(&frame(i as u8, 64)); + } + nic.run(); + one_pass(&mut driver); + let frames: Vec = core::iter::from_fn(|| driver.poll_rx()).collect(); + assert_eq!(frames.len(), HELD_MAX + late, "{}", nic.because("the frames did not all arrive")); + let (armed, writes) = (nic.peek(regs::RDT), writes_of(&nic, regs::RDT)); + + let mut frames = frames.into_iter(); + for frame in frames.by_ref().take(HELD_MAX - 1) { + driver.rx_done(frame); + } + assert_eq!( + (nic.peek(regs::RDT), writes_of(&nic, regs::RDT)), + (armed, writes), + "{}", + nic.because("the tail moved before a batch ended or HELD_MAX buffers waited") + ); + + driver.rx_done(frames.next().expect("the HELD_MAX-th frame")); + assert_eq!( + (nic.peek(regs::RDT) as usize, writes_of(&nic, regs::RDT)), + (HELD_MAX - 1, writes + 1), + "{}", + nic.because("HELD_MAX buffers waited and the tail did not move over them once") + ); + + for frame in frames { + driver.rx_done(frame); + } + driver.rx_flush(); + assert_eq!( + (nic.peek(regs::RDT) as usize, writes_of(&nic, regs::RDT)), + (HELD_MAX + late - 1, writes + 2), + "{}", + nic.because("the batch's end did not hand the rest over in one write") + ); + driver.rx_flush(); + assert_eq!(writes_of(&nic, regs::RDT), writes + 2, "{}", nic.because("a flush with nothing held wrote RDT")); +} + // --- transmit --- #[test] diff --git a/system/netstack/node/src/lib.rs b/system/netstack/node/src/lib.rs index 3a25ececfe0..026abb3054d 100644 --- a/system/netstack/node/src/lib.rs +++ b/system/netstack/node/src/lib.rs @@ -69,11 +69,6 @@ toyos_net_wire::counters! { QueryFailed = "node.query-failed"; } -/// The most frames one [`Node::receive`] takes. A device may refill its ring as fast as it is -/// read, and a batch that waited for it to empty would keep the deadlines, the transmit -/// opportunity and every stream's pass waiting with it. -const RECEIVE_BUDGET: usize = 64; - /// A line for the log. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Event { @@ -155,15 +150,20 @@ impl Node { // ---- the device ---- - /// The frames the device received since the last call, one by one and at most - /// `RECEIVE_BUDGET`: `next` hands its next frame to the sink it is given and answers - /// `false` once it had none. One pass over the streams follows the last frame, so what the - /// batch brought reaches each client in one write and not one a frame. `true` is a batch - /// that stopped at the bound, and another is owed without waiting for the device to say so. - pub fn receive(&mut self, now: Instant, mut next: impl FnMut(&mut dyn FnMut(&[u8])) -> bool, mut draw: impl FnMut() -> u32) -> bool { + /// The frames the device received since the last call, one by one and at most `budget`: + /// `next` hands its next frame to the sink it is given and answers `false` once it had none. + /// One pass over the streams follows the last frame, so what the batch brought reaches each + /// client in one write and not one a frame. `true` is a batch that stopped at the budget, and + /// another is owed without waiting for the device to say so. + /// + /// **The budget is the device's ring**: a device may refill it as fast as it is read, and a + /// batch that waited for it to empty would keep the deadlines, the transmit opportunity and + /// every stream's pass waiting with it, while a bound below what one interrupt's wait can + /// bring would owe a second pass for every long wait. + pub fn receive(&mut self, now: Instant, budget: usize, mut next: impl FnMut(&mut dyn FnMut(&[u8])) -> bool, mut draw: impl FnMut() -> u32) -> bool { let mut taken = 0usize; let owed = loop { - if taken == RECEIVE_BUDGET { + if taken == budget { break true; } if !next(&mut |frame| { diff --git a/system/netstack/node/tests/common/mod.rs b/system/netstack/node/tests/common/mod.rs index 33c130e1345..f5114b2517c 100644 --- a/system/netstack/node/tests/common/mod.rs +++ b/system/netstack/node/tests/common/mod.rs @@ -294,6 +294,9 @@ pub fn echo_request(destination: Ipv4Addr, id: u16, seq: u16, data: &[u8]) -> Ve /// What [`Node::receive`] hands each frame to. type Sink<'s> = dyn FnMut(&[u8]) + 's; +/// The most frames a test's [`Node::receive`] takes: a ring of a device's size. +pub const BUDGET: usize = 256; + /// `frames`, in order, as the batch [`Node::receive`] takes. pub fn batch<'a>(frames: &'a [&'a [u8]]) -> impl FnMut(&mut Sink<'_>) -> bool + 'a { let mut left = frames.iter(); @@ -350,7 +353,7 @@ impl Wire { } let asked = self.sent[from..].iter().filter(|seen| matches!(seen, Seen::Arp { request: true, sender, target, .. } if *target == R && *sender == A)).count(); for _ in 0..asked { - self.node.receive(self.now, batch(&[&arp(MAC, false, MAC_R, R, A)]), draw(&mut self.draws)); + self.node.receive(self.now, BUDGET, batch(&[&arp(MAC, false, MAC_R, R, A)]), draw(&mut self.draws)); } } } @@ -361,7 +364,7 @@ impl Wire { } pub fn deliver(&mut self, frame: &[u8]) { - self.node.receive(self.now, batch(&[frame]), draw(&mut self.draws)); + self.node.receive(self.now, BUDGET, batch(&[frame]), draw(&mut self.draws)); self.pump(); } diff --git a/system/netstack/node/tests/lan/mod.rs b/system/netstack/node/tests/lan/mod.rs index 5d00ae9dd57..a1e5b3889a3 100644 --- a/system/netstack/node/tests/lan/mod.rs +++ b/system/netstack/node/tests/lan/mod.rs @@ -17,7 +17,7 @@ use etherparse::{ArpOperation, Icmpv4Type, LinkSlice, NetSlice, SlicedPacket, Tr use toyos_net_node::Node; use toyos_net_wire::Instant; -use crate::common::{arp, batch, from_server, message_of, option, sum, terms, xid, Wire, A, ACK, BROADCAST, DISCOVER, ELSEWHERE, MAC, MAC_B, MAC_R, OFFER, R}; +use crate::common::{arp, batch, BUDGET, from_server, message_of, option, sum, terms, xid, Wire, A, ACK, BROADCAST, DISCOVER, ELSEWHERE, MAC, MAC_B, MAC_R, OFFER, R}; pub const B: Ipv4Addr = ELSEWHERE; pub const OFF_LINK: Ipv4Addr = Ipv4Addr::new(198, 51, 100, 7); @@ -161,13 +161,13 @@ impl Lan { .collect(); for target in asked { let mac = if target == R { MAC_R } else { MAC_B }; - self.node.receive(self.now, batch(&[&arp(MAC, false, mac, target, A)]), draw(&mut self.draws)); + self.node.receive(self.now, BUDGET, batch(&[&arp(MAC, false, mac, target, A)]), draw(&mut self.draws)); } } } pub fn deliver(&mut self, frame: &[u8]) { - self.node.receive(self.now, batch(&[frame]), draw(&mut self.draws)); + self.node.receive(self.now, BUDGET, batch(&[frame]), draw(&mut self.draws)); self.pump(); } diff --git a/system/netstack/node/tests/listeners.rs b/system/netstack/node/tests/listeners.rs index ae863e13f70..f6f59cf7bd4 100644 --- a/system/netstack/node/tests/listeners.rs +++ b/system/netstack/node/tests/listeners.rs @@ -18,7 +18,7 @@ use std::net::Ipv4Addr; use std::rc::Rc; use std::time::Duration; -use common::{arp, batch, terms, Segment, Wire, A, ELSEWHERE, MAC, MAC_R, R}; +use common::{arp, batch, terms, Segment, Wire, A, BUDGET, ELSEWHERE, MAC, MAC_R, R}; use etherparse::{ArpOperation, LinkSlice, NetSlice, PacketBuilder, SlicedPacket, TcpOptionElement, TransportSlice}; use toyos_net_node::{AcceptRefused, Accepted, ConnectRefused, FromClient, ListenRefused, ListenerId, Node, PipeEnd, Pipes, ReadRefusal, Refused, StreamEvent, StreamId, ToClient, Wake, WriteRefusal}; use toyos_net_tcp::{limits, Counter, Endpoint}; @@ -227,7 +227,7 @@ impl Net { } for frame in &frames { if let Some(answer) = self.hears(frame) { - self.node.receive(self.now, batch(&[&answer]), draw(&mut self.draws)); + self.node.receive(self.now, BUDGET, batch(&[&answer]), draw(&mut self.draws)); } } } @@ -263,7 +263,7 @@ impl Net { } fn deliver(&mut self, frame: &[u8]) { - self.node.receive(self.now, batch(&[frame]), draw(&mut self.draws)); + self.node.receive(self.now, BUDGET, batch(&[frame]), draw(&mut self.draws)); self.pump(); } @@ -809,7 +809,7 @@ fn a_listener_ended_for_its_wake_gives_its_place_to_another_in_the_same_pass() { refusing.borrow_mut().refusal = Some(WriteRefusal::Gone); net.syn(P3); let ack = net.iss(P3).wrapping_add(1); - net.node.receive(net.now, batch(&[&frame(P3, SSH, ISS + 1, Some(ack), 0, &[])]), draw(&mut net.draws)); + net.node.receive(net.now, BUDGET, batch(&[&frame(P3, SSH, ISS + 1, Some(ack), 0, &[])]), draw(&mut net.draws)); assert_eq!(wakes(&owner), 2); let ended: Vec<(ListenerId, WriteRefusal)> = net.node.drain_ended_listeners().collect(); assert_eq!((ended, net.node.listeners(), net.node.held()), (vec![(first, WriteRefusal::Gone)], 1, 1)); diff --git a/system/netstack/node/tests/streams.rs b/system/netstack/node/tests/streams.rs index 8af4d0e46b9..89bbc81b462 100644 --- a/system/netstack/node/tests/streams.rs +++ b/system/netstack/node/tests/streams.rs @@ -16,7 +16,7 @@ use std::net::Ipv4Addr; use std::rc::Rc; use std::time::Duration; -use common::{arp, batch, terms, Segment, Wire, A, MAC, MAC_B, MAC_R, R}; +use common::{arp, batch, terms, Segment, Wire, A, BUDGET, MAC, MAC_B, MAC_R, R}; use etherparse::{ArpOperation, LinkSlice, NetSlice, PacketBuilder, SlicedPacket, TcpOptionElement, TransportSlice}; use toyos_net_node::{ConnectRefused, FromClient, Node, PipeEnd, Pipes, ReadRefusal, StreamEvent, StreamId, ToClient, Watch, WriteRefusal}; use toyos_net_shard::ConnectError; @@ -424,14 +424,14 @@ impl Net { self.node.transmit(self.now, usize::MAX, |frame| frames.push(frame.to_vec()), draw(&mut self.draws)); for frame in &frames { for answer in self.hears(frame) { - self.node.receive(self.now, batch(&[&answer]), draw(&mut self.draws)); + self.node.receive(self.now, BUDGET, batch(&[&answer]), draw(&mut self.draws)); } } frames.len() } fn deliver(&mut self, frame: &[u8]) { - self.node.receive(self.now, batch(&[frame]), draw(&mut self.draws)); + self.node.receive(self.now, BUDGET, batch(&[frame]), draw(&mut self.draws)); self.pump(); } @@ -660,33 +660,32 @@ fn a_batch_of_frames_moves_an_established_stream_once_with_no_opportunity_after_ let (mut net, _, client) = established(); let frames = [net.far.text(b"all "), net.far.text(b"at "), net.far.text(b"once")]; let (writes, reads) = (client.borrow().writes, client.borrow().reads); - net.node.receive(net.now, batch(&[&frames[0], &frames[1], &frames[2]]), draw(&mut net.draws)); + net.node.receive(net.now, BUDGET, batch(&[&frames[0], &frames[1], &frames[2]]), draw(&mut net.draws)); let ends = client.borrow(); assert_eq!(ends.inbox, b"all at once"); assert_eq!((ends.writes - writes, ends.reads - reads), (1, 0), "the pipes' calls in the batch's pass"); } // A device that refills its ring as fast as it is read never says it has no frame: a batch -// stops at the bound and answers that another pass is owed, and one that found the ring empty -// owes none, a ring one short of the bound included. +// stops at the budget it was given and answers that another pass is owed, and one that found the +// ring empty owes none, a ring one short of the budget included. #[test] -fn a_batch_stops_at_its_bound_and_says_a_pass_is_owed() { +fn a_batch_stops_at_its_budget_and_says_a_pass_is_owed() { let (mut net, _, client) = established(); let frame = net.far.text(b"again"); let mut handed = 0usize; let endless = |sink: &mut dyn FnMut(&[u8])| { handed += 1; - assert!(handed <= 1 << 16, "a batch with no bound"); sink(&frame); true }; - assert!(net.node.receive(net.now, endless, draw(&mut net.draws)), "a pass is owed"); - assert!(handed > 1, "a bound of one frame is a pass a frame"); + let owed = net.node.receive(net.now, BUDGET, endless, draw(&mut net.draws)); + assert_eq!((owed, handed), (true, BUDGET), "the batch is the budget, and a pass is owed"); assert_eq!(client.borrow().inbox, b"again", "the rest of the batch was the same segment again"); let next = net.far.text(b"once"); - let short: Vec<&[u8]> = vec![&next; handed - 1]; - assert!(!net.node.receive(net.now, batch(&short), draw(&mut net.draws)), "the ring was emptied"); + let short: Vec<&[u8]> = vec![&next; BUDGET - 1]; + assert!(!net.node.receive(net.now, BUDGET, batch(&short), draw(&mut net.draws)), "the ring was emptied"); assert_eq!(client.borrow().inbox, b"againonce"); } @@ -1084,7 +1083,7 @@ fn a_reset_in_the_batch_of_the_text_before_it_ends_the_pipes_after_the_text() { let (mut net, _, client) = established(); let text = net.far.text(b"last"); let rst = net.far.rst(); - net.node.receive(net.now, batch(&[&text, &rst]), draw(&mut net.draws)); + net.node.receive(net.now, BUDGET, batch(&[&text, &rst]), draw(&mut net.draws)); assert_eq!((client.borrow().inbox.clone(), let_go(&client)), (b"last".to_vec(), vec![PipeEnd::FromClient, PipeEnd::ToClient])); } @@ -1096,7 +1095,7 @@ fn text_the_pipe_had_no_room_for_outlives_the_reset() { client.borrow_mut().room = 0; let text = net.far.text(b"held back"); let rst = net.far.rst(); - net.node.receive(net.now, batch(&[&text, &rst]), draw(&mut net.draws)); + net.node.receive(net.now, BUDGET, batch(&[&text, &rst]), draw(&mut net.draws)); assert_eq!((client.borrow().inbox.len(), let_go(&client), net.node.streams()), (0, vec![PipeEnd::FromClient], 1)); client.borrow_mut().room = 65_536; diff --git a/system/netstack/src/card.rs b/system/netstack/src/card.rs index e95a53dfaca..29cc9a8cbf9 100644 --- a/system/netstack/src/card.rs +++ b/system/netstack/src/card.rs @@ -130,9 +130,24 @@ impl Card { snap.put("errors.crc", wire.crc_errors); } + /// One batch of received frames: `batch` is given the most it may take — the card's ring, + /// all of what one wait can bring — and the card's next frame on each call, and every buffer + /// it gave back reaches the card once it returns. + pub fn receive(&self, batch: impl FnOnce(usize, &mut dyn FnMut(&mut dyn FnMut(&[u8])) -> bool) -> R) -> R { + let budget = match self { + Self::Virtio(_) => crate::virtio_net::RX_BUF_COUNT, + Self::Intel(_) => toyos_i219::RX_RING - 1, + }; + let took = batch(budget, &mut |sink| self.rx(sink)); + if let Self::Intel(nic) = self { + nic.rx_flush(); + } + took + } + /// Hands the next received frame to `take` and gives its buffer back to /// the card once `take` returns; `false` when none waits. - pub fn rx(&self, take: impl FnOnce(&[u8])) -> bool { + fn rx(&self, take: &mut dyn FnMut(&[u8])) -> bool { match self { Self::Virtio(nic) => { let Some((index, len)) = nic.poll_rx() else { return false }; diff --git a/system/netstack/src/i219.rs b/system/netstack/src/i219.rs index a71fbc5a289..2cf5f50d47a 100644 --- a/system/netstack/src/i219.rs +++ b/system/netstack/src/i219.rs @@ -262,6 +262,11 @@ impl Nic { self.driver.borrow_mut().rx_done(frame); } + /// Hand the part every buffer the batch gave back. + pub fn rx_flush(&self) { + self.driver.borrow_mut().rx_flush(); + } + /// How many frames the transmit ring takes now. pub fn tx_room(&self) -> usize { self.driver.borrow_mut().tx_room() diff --git a/system/netstack/src/main.rs b/system/netstack/src/main.rs index 73960cddf28..b61b1922a27 100644 --- a/system/netstack/src/main.rs +++ b/system/netstack/src/main.rs @@ -48,6 +48,7 @@ mod i219; mod pipes; mod serve; mod virtio_net; +mod watches; use card::Card; use client::{Client, ClientRx, HANDSHAKE_TIMEOUT, MAX_KEPT_REQUEST, MAX_PENDING_CONNS, PendingConn, Request}; @@ -356,7 +357,7 @@ fn main() { node.link(clock(), true, draw); } - let poller = Poller::new( + let mut poller = watches::Watches::new( FIXED_WATCHES + serve::WATCHES_PER_PLACE * MAX_PLACES + MAX_PENDING_CONNS + serve::LOOKUP_WATCHES, ); const TOKEN_ACCEPTOR: u64 = 0; @@ -379,7 +380,7 @@ fn main() { node.link(clock(), link_up, draw); } } - let owed = node.receive(clock(), |sink| card.rx(sink), draw); + let owed = card.receive(|budget, next| node.receive(clock(), budget, next, draw)); let now = clock(); if node.next_deadline().is_some_and(|at| at <= now) { node.fire(now, draw); @@ -410,7 +411,7 @@ fn main() { poller.watch(&acceptor, READABLE, TOKEN_ACCEPTOR); poller.watch(card.claim(), READABLE, TOKEN_NIC); - sockets.watch(&node, &poller); + sockets.watch(&node, &mut poller); for p in pending.iter() { poller.watch(&p.conn, READABLE, TOKEN_PENDING_BASE + p.conn.as_handle().0 as u64); } diff --git a/system/netstack/src/serve.rs b/system/netstack/src/serve.rs index 7a95dbf023e..b6fd96f5e4c 100644 --- a/system/netstack/src/serve.rs +++ b/system/netstack/src/serve.rs @@ -28,7 +28,7 @@ use std::net::Ipv4Addr; use std::time::Duration; use toyos::net::*; -use toyos::poller::{OTHER_END_GONE, READABLE, WRITABLE, Poller}; +use toyos::poller::{OTHER_END_GONE, READABLE, WRITABLE}; use toyos::{ipc, say, AsHandle, Pipe}; use toyos_abi::syscall::SyscallError; use toyos_inspect::Snapshot; @@ -42,6 +42,7 @@ use toyos_net_wire::{Instant, Port}; use crate::client::{Client, Request}; use crate::pipes::{self, Watched}; +use crate::watches::Watches; use crate::HOSTNAME; /// A watch's token names the socket it was asked of by its id, in the low @@ -737,7 +738,7 @@ impl Sockets { /// one pipe of each datagram socket for its owner's leaving, and the /// connection of each client that waits for a lookup, which hangs up by /// closing it. - pub fn watch(&self, node: &Node, poller: &Poller) { + pub fn watch(&self, node: &Node, poller: &mut Watches) { for (stream, watch) in node.watches() { let Some((socket_id, ends)) = self.by_stream.get(&stream).and_then(|id| Some((*id, self.ends.get(id)?))) else { unreachable!("the node holds a stream whose ends netstack does not") diff --git a/system/netstack/src/watches.rs b/system/netstack/src/watches.rs new file mode 100644 index 00000000000..a3c668c67a8 --- /dev/null +++ b/system/netstack/src/watches.rs @@ -0,0 +1,164 @@ +//! What netstack has asked the kernel to watch, so that a pass submits a watch +//! only where its interest changed or the kernel has answered it. +//! +//! **A kernel watch stays armed until it is answered or replaced**, and its +//! answer is what the object holds at the look that answers it +//! (`kernel/src/inbox/polls.rs`). So a watch submitted again with the same +//! handle, interest and token replaces a poll with its twin and costs the +//! submit the poll's registration on every watch it names, for nothing. +//! +//! **The kernel keeps one poll a handle** (`admit` in `kernel/src/inbox/polls.rs`): +//! a watch of a handle under another token withdraws the handle's earlier poll +//! unanswered. So a record is kept by handle, and a watch under another token +//! replaces it as the kernel replaces the poll. +//! +//! **Every poll the kernel keeps is answered, once**: by readiness, by a +//! refusal, or by `NotFound` when a handle it names is closed, so a record here +//! is let go by the answer under its token and never outlives the poll it +//! records. A record whose poll was answered and whose answer is not yet read +//! costs one needless submit after that answer, and never a missed one. + +use std::collections::HashMap; + +use toyos::poller::Poller; +use toyos::AsHandle; +use toyos_abi::syscall::SyscallError; + +/// The watches the kernel holds armed for this process: each handle's token +/// and interest, and each token's handle. +#[derive(Default)] +pub struct Armed { + by_handle: HashMap, + by_token: HashMap, +} + +impl Armed { + /// Whether a watch of `flags` on `handle` under `token` has to be + /// submitted, recording it as armed where it does. + pub fn submit(&mut self, token: u64, handle: u32, flags: u32) -> bool { + let was = self.by_handle.insert(handle, (token, flags)); + if was == Some((token, flags)) { + return false; + } + if let Some((other, _)) = was.filter(|&(other, _)| other != token) { + self.by_token.remove(&other); + } + if let Some(moved) = self.by_token.insert(token, handle).filter(|&moved| moved != handle) { + self.by_handle.remove(&moved); + } + true + } + + /// The kernel answered the watch under `token`: it is armed no more. + pub fn answered(&mut self, token: u64) { + if let Some(handle) = self.by_token.remove(&token) { + self.by_handle.remove(&handle); + } + } +} + +/// A [`Poller`] that submits only what [`Armed`] says has to be. +pub struct Watches { + poller: Poller, + armed: Armed, +} + +impl Watches { + pub fn new(capacity: u32) -> Self { + Self { poller: Poller::new(capacity), armed: Armed::default() } + } + + pub fn watch(&mut self, handle: &impl AsHandle, flags: u32, token: u64) { + let raw = handle.as_handle(); + if self.armed.submit(token, raw.0, flags) { + self.poller.watch_raw(raw, flags, token); + } + } + + /// [`Poller::wait_answers`], each answer letting its watch's record go + /// before `f` sees it. + pub fn wait_answers(&mut self, min_complete: u32, timeout_nanos: u64, mut f: impl FnMut(u64, Result)) { + let armed = &mut self.armed; + self.poller.wait_answers(min_complete, timeout_nanos, |token, answer| { + armed.answered(token); + f(token, answer); + }); + } +} + +#[cfg(test)] +mod tests { + use super::Armed; + + const TOKEN: u64 = 0x1_0000_0007; + const READABLE: u32 = 1; + const WRITABLE: u32 = 2; + + #[test] + fn an_armed_watch_is_not_submitted_again() { + let mut armed = Armed::default(); + assert!(armed.submit(TOKEN, 7, READABLE)); + assert!(!armed.submit(TOKEN, 7, READABLE)); + assert!(!armed.submit(TOKEN, 7, READABLE)); + } + + #[test] + fn an_answered_watch_is_submitted_again() { + let mut armed = Armed::default(); + assert!(armed.submit(TOKEN, 7, READABLE)); + armed.answered(TOKEN); + assert!(armed.submit(TOKEN, 7, READABLE)); + } + + #[test] + fn a_changed_interest_is_submitted() { + let mut armed = Armed::default(); + assert!(armed.submit(TOKEN, 7, READABLE)); + assert!(armed.submit(TOKEN, 7, READABLE | WRITABLE)); + assert!(armed.submit(TOKEN, 7, READABLE)); + assert!(!armed.submit(TOKEN, 7, READABLE)); + } + + /// A handle number its closed holder freed, named again under the same + /// token: the closed one's poll ended with it and the new one has none. + #[test] + fn a_token_on_another_handle_is_submitted() { + let mut armed = Armed::default(); + assert!(armed.submit(TOKEN, 7, READABLE)); + assert!(armed.submit(TOKEN, 8, READABLE)); + } + + /// A handle watched under another token is submitted, and so is the first + /// token's watch of it after that: the kernel withdrew that poll when the + /// second replaced it. + #[test] + fn a_handle_under_another_token_is_submitted_and_so_is_the_first_again() { + let mut armed = Armed::default(); + assert!(armed.submit(TOKEN, 7, READABLE)); + assert!(armed.submit(TOKEN + 1, 7, READABLE)); + assert!(armed.submit(TOKEN, 7, READABLE)); + assert!(!armed.submit(TOKEN, 7, READABLE)); + } + + /// The first token's answer, read after the handle moved to a second, + /// lets the second's record stand. + #[test] + fn an_answer_under_a_replaced_token_lets_the_handles_record_stand() { + let mut armed = Armed::default(); + assert!(armed.submit(TOKEN, 7, READABLE)); + assert!(armed.submit(TOKEN + 1, 7, READABLE)); + armed.answered(TOKEN); + assert!(!armed.submit(TOKEN + 1, 7, READABLE)); + } + + /// The answer of one token lets no other's record go. + #[test] + fn an_answer_is_its_own_tokens_alone() { + let mut armed = Armed::default(); + assert!(armed.submit(TOKEN, 7, READABLE)); + assert!(armed.submit(TOKEN + 1, 8, READABLE)); + armed.answered(TOKEN); + assert!(!armed.submit(TOKEN + 1, 8, READABLE)); + assert!(armed.submit(TOKEN, 7, READABLE)); + } +} diff --git a/tests/guest/rust-tests/src/bin/pipe_other_end_gone.rs b/tests/guest/rust-tests/src/bin/pipe_other_end_gone.rs index f67c2e683ec..4787208946f 100644 --- a/tests/guest/rust-tests/src/bin/pipe_other_end_gone.rs +++ b/tests/guest/rust-tests/src/bin/pipe_other_end_gone.rs @@ -12,6 +12,10 @@ //! **The refusal.** A connection and a file have no other end, and a watch that //! asks for one is refused whatever else it asks. //! +//! **Its own close.** An end whose holder closes it while the other end lives +//! answers its watch with `NotFound`: the poll is over, and nothing else would +//! end it. +//! //! **Not gone.** The other end has a holder while a duplicate of its handle is //! held, while the handle sits unreceived in a connection's queue, and after //! it moved to a child that still lives: the watch stays armed through each, @@ -75,6 +79,17 @@ fn test() { syscall::close(file); println!(" a connection and a file were refused the question"); + for own in [End::Read, End::Write] { + let (reader, writer) = pipe(""); + let (closed, kept) = match own { + End::Read => (reader, writer), + End::Write => (writer, reader), + }; + assert_eq!(closed_own(closed), [(TOKEN, Err::(SyscallError::NotFound))], "{own:?} end closed by its holder, its other end held"); + drop(kept); + } + println!(" an end closed by its own holder answered its watch with NotFound"); + let (reader, writer) = pipe(""); let second = syscall::dup(writer.as_handle()).expect("a second handle to the write end"); drop(writer); @@ -105,6 +120,31 @@ fn test() { println!("pipe_other_end_gone: an end is told when its other end's last holder goes, and not before"); } +#[derive(Clone, Copy, Debug)] +enum End { + Read, + Write, +} + +/// How long [`closed_own`] waits for the answer its close owes: a close +/// answers it before it returns, so this is a hang's ceiling. +const CLOSE_ANSWERED_NANOS: u64 = 10_000_000_000; + +/// What one `OTHER_END_GONE` watch of `end`, held by the kernel, is answered +/// once `end` is closed: every answer a wait bounded by +/// [`CLOSE_ANSWERED_NANOS`] collects. +fn closed_own(end: Pipe) -> Vec<(u64, Result)> { + let poller = Poller::new(1); + poller.watch(&end, OTHER_END_GONE, TOKEN); + let mut answers = Vec::new(); + // A non-blocking enter, so the watch is the kernel's before the close. + poller.wait_answers(0, 0, |token, said| answers.push((token, said))); + assert!(answers.is_empty(), "an end was told its other end is gone while it is held: {answers:?}"); + syscall::close(end.into_raw()); + poller.wait_answers(1, CLOSE_ANSWERED_NANOS, |token, said| answers.push((token, said))); + answers +} + /// A fresh pipe with `unread` written into it. fn pipe(unread: &str) -> (Pipe, Pipe) { let (reader, writer) = pipe_pair().expect("a pipe");