diff --git a/issues/netstacks-virtio-receive-has-no-per-pass-budget.md b/issues/netstacks-virtio-receive-has-no-per-pass-budget.md deleted file mode 100644 index 39c8e5d31de..00000000000 --- a/issues/netstacks-virtio-receive-has-no-per-pass-budget.md +++ /dev/null @@ -1,31 +0,0 @@ ---- -status: open -kind: defect -opened: 2026-10-10 ---- - -# netstack's virtio receive has no per-pass budget - -`VirtioNet::poll_rx` (`userland/netstack/src/virtio_net.rs`) answers a frame for -as long as the used ring holds one, and `Node::receive` -(`userland/netstack/node/src/lib.rs`) takes frames until the card says none is -left. A device that refills the ring as fast as it is read keeps one pass in -receive: the transmit opportunity, the deadlines and every stream's pass wait -for the ring to empty, and the whole batch is read at one `now`. The I219 is -bounded at `toyos_i219::RX_BUDGET` frames a pass. - -The I219's bound is not one to copy. A pass that stops at it leaves frames whose -interrupts the pass's `begin_pass` already took, and nothing in netstack's loop -(`userland/netstack/src/main.rs`) makes the next wait return at once, so they -wait for the next interrupt or deadline. Read from the code, not measured. - -The fix is one bound for every driver where the pass takes its frames, at -`Node::receive`, answering that it stopped at the bound so the loop's next wait -is zero; the I219's own budget is then deleted. That touches `toyos-i219`, which -the branch that found this (#828) was not briefed to change. - -**Exit condition**: a host test feeds `Node::receive` a source that never empties -and sees it stop at the bound and answer that a pass is owed; netstack's loop -waits zero after such a pass; `toyos_i219::RX_BUDGET` is gone. - -**Owner**: whoever holds `issues/toyos-has-its-own-network-stack.md`. diff --git a/issues/the-lan-is-not-yet-production-grade.md b/issues/the-lan-is-not-yet-production-grade.md index 744210c307f..3785e7e0efe 100644 --- a/issues/the-lan-is-not-yet-production-grade.md +++ b/issues/the-lan-is-not-yet-production-grade.md @@ -39,8 +39,9 @@ why, and the boot still leases. and raise it under traffic. Nothing has one, and `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 (off - today), ring sizes, checksum and segmentation offload. + 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 + 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 ToyOS boot so far; that becomes a tested property. diff --git a/toyos-i219/src/lib.rs b/toyos-i219/src/lib.rs index 6fa33319e30..a7983803ce5 100644 --- a/toyos-i219/src/lib.rs +++ b/toyos-i219/src/lib.rs @@ -156,6 +156,12 @@ pub const RX_RING: usize = 256; /// 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; /// 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`] @@ -191,13 +197,6 @@ const _: () = { assert!(OFF_RX_RING.is_multiple_of(16) && OFF_TX_RING.is_multiple_of(16)); }; -/// Frames this driver hands up in one pass before it says there are no more. -/// -/// **A ring that never empties is a ring the specification permits**: at line -/// rate the device refills descriptors as fast as they are returned, and a -/// caller that loops until the ring is empty never returns to its other work. -pub const RX_BUDGET: u32 = 64; - /// Why the function was not brought up. Each keeps its own word: a caller asks /// different things of a grant that is too small and of a part with no station /// address. @@ -766,7 +765,6 @@ pub struct I219 { /// receive ring keeps in software's hands (§7.1.8: the queue is empty when /// head equals tail). rx_tail: usize, - rx_budget: u32, /// Next transmit descriptor to hand out, which is also `TDT`. tx_next: usize, /// Next transmit descriptor to reclaim. @@ -915,7 +913,6 @@ impl I219 { link_up_at: None, rx_next: 0, rx_tail: RX_RING - 1, - rx_budget: RX_BUDGET, tx_next: 0, tx_clean: 0, tx_wake: false, @@ -933,10 +930,10 @@ impl I219 { nic.regs.write32(regs::IVAR, ivar::ALL_ON_VECTOR_ZERO); nic.accepted(regs::IVAR, ivar::ALL_ON_VECTOR_ZERO)?; - // No moderation on either side: §10.2.4.2's throttle and the two - // receive timers all hold an interrupt back, and what this driver - // waits on is the frame that has already arrived. - nic.regs.write32(regs::ITR, 0); + // The throttle alone: the two receive timers hold back every frame's + // interrupt, the first after a quiet spell too, and the throttle none + // of those. + nic.regs.write32(regs::ITR, THROTTLE); nic.regs.write32(regs::RDTR, 0); nic.regs.write32(regs::RADV, 0); nic.regs.write32(regs::TIDV, 0); @@ -1085,8 +1082,7 @@ impl I219 { self.link_up_at.map(|at| at.saturating_sub(self.opened_at)) } - /// Take the interrupt, read and acknowledge its causes, and give the - /// receive budget back. + /// Take the interrupt, and read and acknowledge its causes. /// /// **The record has to be taken, not merely noticed**: a claim reads ready /// while it holds an undrained interrupt, so a caller that saw the token @@ -1103,7 +1099,6 @@ impl I219 { /// down, so a link that reads up on both sides of it starts the same /// deadline. pub fn begin_pass(&mut self) -> Result> { - self.rx_budget = RX_BUDGET; let messages = match self.irq.taken() { Ok(count) => count, Err(why) if why == I::IDLE => 0, @@ -1212,17 +1207,13 @@ impl I219 { } } - /// The next received frame, or `None` — for an empty ring and for a budget - /// already spent alike. + /// The next received frame, or `None` for an empty ring. /// /// A refused descriptor is counted, its buffer given straight back, and /// the walk continues: one frame this driver will not act on may not hide /// the ones behind it. pub fn poll_rx(&mut self) -> Option { loop { - if self.rx_budget == 0 { - return None; - } let index = self.rx_next; let at = OFF_RX_RING + index * rx_desc::BYTES; let word = self.dma.read64(at + 8); @@ -1235,7 +1226,6 @@ impl I219 { self.dma.observe(); let word = self.dma.read64(at + 8); self.rx_next = (index + 1) % RX_RING; - self.rx_budget -= 1; match parse_rx(word) { Ok(len) => { self.counters.received = self.counters.received.saturating_add(1); diff --git a/toyos-i219/src/tests.rs b/toyos-i219/src/tests.rs index 7d2f159f8af..7a197b77a24 100644 --- a/toyos-i219/src/tests.rs +++ b/toyos-i219/src/tests.rs @@ -122,11 +122,11 @@ fn bring_up_programs_what_the_initialization_sections_name() { assert_eq!(nic.peek(regs::TDLEN) as usize, TX_RING * tx_desc::BYTES); assert_eq!(nic.peek(regs::TDT), 0); - // Both timers off: an interrupt this driver waits on may not be held back - // by one nothing else expires. + // 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. assert_eq!(nic.peek(regs::RDTR), 0); assert_eq!(nic.peek(regs::RADV), 0); - assert_eq!(nic.peek(regs::ITR), 0); + assert_eq!(u64::from(nic.peek(regs::ITR)) * 256, 128_000, "§10.2.4.2 counts in 256 ns"); assert_eq!(driver.mac(), NVM_MAC); } @@ -503,41 +503,6 @@ fn null_descriptor_padding_is_not_a_frame() { assert_eq!(got.len(), 1, "{}", nic.because("the real frame did not come up behind it")); } -/// **A ring that never empties is what the wire does at line rate.** A pass -/// has to end, or the caller's event loop never runs again — so the budget is -/// what makes "no more this pass" something this driver can say. -#[test] -fn one_pass_hands_up_at_most_its_budget() { - let nic = Nic::new(8); - let mut driver = open(&nic); - nic.set_link(true); - // Twice the budget, arriving faster than they are taken: every buffer - // returned is another frame placed. - for i in 0..(RX_BUDGET as usize * 2) { - nic.deliver(&frame((i % 200) as u8 + 1, 128)); - } - // Enough ticks for every one of them to be placed and written back, so - // what bounds the pass below is the budget and not the device. - for _ in 0..(RX_BUDGET * 4) { - nic.run(); - } - one_pass(&mut driver); - let first = drain(&nic, &mut driver).len(); - assert_eq!( - first, RX_BUDGET as usize, - "{}", - nic.because("a pass did not stop at its budget") - ); - // And the next pass picks the rest up rather than losing them. - let mut rest = 0; - for _ in 0..8 { - nic.run(); - one_pass(&mut driver); - rest += drain(&nic, &mut driver).len(); - } - assert_eq!(first + rest, RX_BUDGET as usize * 2, "{}", nic.because("frames were lost")); -} - /// §7.1.8's tail "identifies the location beyond the last descriptor hardware /// can process", so it moves over a run of ready descriptors and never to the /// index that came back last. A buffer given back before an older one may not diff --git a/userland/netstack/node/src/lib.rs b/userland/netstack/node/src/lib.rs index 420501a525c..3a25ececfe0 100644 --- a/userland/netstack/node/src/lib.rs +++ b/userland/netstack/node/src/lib.rs @@ -69,6 +69,11 @@ 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 { @@ -150,16 +155,27 @@ impl Node { // ---- the device ---- - /// The frames the device received since the last call, one by one: `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. - pub fn receive(&mut self, now: Instant, mut next: impl FnMut(&mut dyn FnMut(&[u8])) -> bool, mut draw: impl FnMut() -> u32) { - while next(&mut |frame| { - self.stack.receive(now, frame); - self.settle(now, &mut draw); - }) {} - self.bridge(now); + /// 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 { + let mut taken = 0usize; + let owed = loop { + if taken == RECEIVE_BUDGET { + break true; + } + if !next(&mut |frame| { + self.stack.receive(now, frame); + self.settle(now, &mut draw); + }) { + break false; + } + taken = taken.saturating_add(1); + }; + self.pass(now, false); + owed } /// A transmit opportunity with room for `credit` frames, each handed to `sink` as it is built. @@ -207,7 +223,7 @@ impl Node { let out = self.client.timer(now, &mut draw); self.carry_out(now, out, None, &mut draw); self.settle(now, &mut draw); - self.bridge(now); + self.pass(now, false); } /// Hands the client what the shard reported and what reached its socket, and carries out diff --git a/userland/netstack/node/src/listeners.rs b/userland/netstack/node/src/listeners.rs index 6ee247b8ec1..36e65b65c09 100644 --- a/userland/netstack/node/src/listeners.rs +++ b/userland/netstack/node/src/listeners.rs @@ -156,7 +156,7 @@ impl Node { listener.unspent = listener.unspent.saturating_sub(1); let bound = listener.bound; let answer = self.take(bound, pipes); - self.bridge(now); + self.pass(now, false); answer } diff --git a/userland/netstack/node/src/streams.rs b/userland/netstack/node/src/streams.rs index 025a510a42b..1201fa2e44b 100644 --- a/userland/netstack/node/src/streams.rs +++ b/userland/netstack/node/src/streams.rs @@ -40,6 +40,12 @@ //! Every call that can move a stream ends in a pass: a batch of frames, a deadline, and each call //! here. A transmit opportunity can only fail a connect, whose next hop it found to answer //! nobody, and ends in a pass over the connects. +//! +//! **A send pipe found empty is read again only once the kernel says it holds bytes** +//! ([`Node::bridge`]'s `readable`), or once its client writes no more, when an empty pipe is the +//! end. netstack watches such a pipe for bytes as long as [tcp] has room for them +//! ([`Watch::readable`]), and the kernel answers a watch of a pipe that already holds some at +//! once, so no byte waits on a pass that never reads. use alloc::boxed::Box; use alloc::collections::{BTreeMap, VecDeque}; @@ -178,6 +184,9 @@ struct Stream { held: bool, /// [tcp] had room for the client's bytes when the last pass ended. room: bool, + /// The last read of the from-client pipe found it empty, and the kernel has not said since + /// that it holds bytes. + drained: bool, /// The options last written to [tcp]. options: Options, } @@ -199,6 +208,7 @@ impl Stream { extended: false, held: false, room: false, + drained: false, options, } } @@ -266,6 +276,9 @@ impl Stream { let mut gave_up = false; while let Some(pipe) = self.from_client.as_mut() { + if self.drained && !self.done_writing { + break; + } // Never more than [tcp] has room for, and so never a read of no bytes, whose answer // would be the end's; [tcp] has none once the FIN is queued. let room = stack.tcp_status(conn).writable.min(CHUNK); @@ -278,7 +291,10 @@ impl Stream { let (read, writer_gone) = match pipe.read(space) { Ok(read) => (read, read == 0), Err(ReadRefusal::Empty) if self.done_writing => (0, false), - Err(ReadRefusal::Empty) => break, + Err(ReadRefusal::Empty) => { + self.drained = true; + break; + } Err(ReadRefusal::Broken) => { stack.tcp_abort(now, conn); return false; @@ -377,8 +393,15 @@ impl Node { Ok(self.streams.hold(Stream { connecting: true, deadline, ..Stream::established(conn, remote.addr, Options::default(), pipes) })) } - /// One pass over every stream: netstack calls it when a pipe it watches is ready. - pub fn bridge(&mut self, now: Instant) { + /// One pass over every stream: netstack calls it once the kernel has answered pipes it + /// watches ready as [`Self::watches`] asked, and `readable` are the streams whose from-client + /// pipe it said holds bytes. + pub fn bridge(&mut self, now: Instant, readable: impl IntoIterator) { + for id in readable { + if let Some(stream) = self.streams.live.get_mut(&id) { + stream.drained = false; + } + } self.pass(now, false); } @@ -409,7 +432,7 @@ impl Node { stream.to_client = None; stream.done_writing = true; stream.left = true; - self.bridge(now); + self.pass(now, false); } /// The client sends no more: what its pipe still holds is sent, then the FIN. `false` is an @@ -417,7 +440,7 @@ impl Node { pub fn shutdown_write(&mut self, now: Instant, id: StreamId) -> bool { let Some(stream) = self.streams.live.get_mut(&id).filter(|stream| !stream.connecting) else { return false }; stream.done_writing = true; - self.bridge(now); + self.pass(now, false); true } @@ -442,7 +465,7 @@ impl Node { PipeEnd::FromClient if stream.fin_queued => stream.from_client = None, PipeEnd::FromClient => stream.done_writing = true, } - self.bridge(now); + self.pass(now, false); } /// The kernel refused netstack's watch of a pipe end the node still holds: the handle is no diff --git a/userland/netstack/node/tests/listeners.rs b/userland/netstack/node/tests/listeners.rs index 4ecc470ac85..ae863e13f70 100644 --- a/userland/netstack/node/tests/listeners.rs +++ b/userland/netstack/node/tests/listeners.rs @@ -398,10 +398,11 @@ impl Net { (answer, client) } - /// The client writes `text`, and the pass netstack runs when its pipe has bytes. - fn says(&mut self, client: &Client, text: &[u8]) { + /// The client of stream `id` writes `text`, and the pass netstack runs when its pipe has + /// bytes. + fn says(&mut self, id: StreamId, client: &Client, text: &[u8]) { client.borrow_mut().outbox.extend(text); - self.node.bridge(self.now); + self.node.bridge(self.now, [id]); self.pump(); } @@ -454,7 +455,7 @@ fn a_listener_answers_a_syn_and_wakes_its_owner_when_the_handshake_ends() { assert_eq!(client.borrow().inbox, b"hello", "what arrived before the accept moves in it"); assert_eq!((net.node.streams(), net.node.listeners(), net.node.held()), (1, 1, 2)); client.borrow_mut().outbox.extend(b"welcome"); - net.node.bridge(net.now); + net.node.bridge(net.now, [accepted.id]); net.pump(); let said = net.last(P1); assert_eq!((said.text.as_slice(), said.seq), (&b"welcome"[..], synack.seq.wrapping_add(1))); @@ -855,16 +856,16 @@ fn a_stream_starts_with_the_options_its_connection_took_from_its_listener() { let (second, b) = net.accept(id).unwrap(); assert_eq!((first.remote, first.nodelay, second.remote, second.nodelay), (P1.endpoint(), false, P2.endpoint(), true)); let (first, second) = (first.id, second.id); - for client in [&a, &b] { - net.says(client, b"a"); - net.says(client, b"b"); + for (stream, client) in [(first, &a), (second, &b)] { + net.says(stream, client, b"a"); + net.says(stream, client, b"b"); } assert_eq!(net.texts(P1), [b"a".to_vec()], "the second write waits for the first's acknowledgment"); assert_eq!(net.texts(P2), [b"a".to_vec(), b"b".to_vec()], "the second write waits for no acknowledgment"); // An accepted stream's options are its own from then on, and the listener's its own. assert!(net.node.set_nodelay(net.now, second, false)); - net.says(&b, b"c"); + net.says(second, &b, b"c"); assert_eq!(net.texts(P2), [b"a".to_vec(), b"b".to_vec()], "the third write waits"); assert!(net.node.set_nodelay(net.now, first, true)); net.pump(); @@ -874,8 +875,8 @@ fn a_stream_starts_with_the_options_its_connection_took_from_its_listener() { net.handshake(P3); let (third, c) = net.accept(id).unwrap(); assert_eq!((third.remote, third.nodelay), (P3.endpoint(), false)); - net.says(&c, b"a"); - net.says(&c, b"b"); + net.says(third.id, &c, b"a"); + net.says(third.id, &c, b"b"); assert_eq!(net.texts(P3), [b"a".to_vec()]); assert!(net.node.close_listener(net.now, id)); @@ -892,8 +893,8 @@ fn a_listener_holds_the_option_its_listen_named_from_its_first_connection() { net.handshake(P1); let (accepted, client) = net.accept(id).unwrap(); assert!(accepted.nodelay); - net.says(&client, b"a"); - net.says(&client, b"b"); + net.says(accepted.id, &client, b"a"); + net.says(accepted.id, &client, b"b"); assert_eq!(net.texts(P1), [b"a".to_vec(), b"b".to_vec()], "the second write waits for no acknowledgment"); assert!(net.node.set_listener_nodelay(id, false)); diff --git a/userland/netstack/node/tests/streams.rs b/userland/netstack/node/tests/streams.rs index 7c521334881..8af4d0e46b9 100644 --- a/userland/netstack/node/tests/streams.rs +++ b/userland/netstack/node/tests/streams.rs @@ -362,6 +362,8 @@ struct Net { far: Far, /// The peers the test turned from, each with its connection. parked: Vec, + /// Each stream's client, for the kernel's answers [`Self::bridge`] gives. + clients: Vec<(StreamId, Client)>, } fn draw(draws: &mut u32) -> impl FnMut() -> u32 + '_ { @@ -375,7 +377,7 @@ impl Net { /// The node holding its lease of 192.0.2.1/24 for an hour. fn new() -> Self { let Wire { node, now, .. } = Wire::leased(&terms(3_600, Some(R))); - Self { node, now, draws: 0x7c00_0000, far: Far::new(), parked: Vec::new() } + Self { node, now, draws: 0x7c00_0000, far: Far::new(), parked: Vec::new(), clients: Vec::new() } } /// Leaves the present peer with the connection it has and makes `next` the peer the test @@ -433,9 +435,19 @@ impl Net { self.pump(); } - /// A pass, as netstack runs one when a pipe it watches is ready, and what it sends. + /// A pass, as netstack runs one when a pipe it watches is ready, and what it sends. The + /// kernel answers a send pipe readable that holds bytes, has no writer left or is no pipe. fn bridge(&mut self) { - self.node.bridge(self.now); + let readable: Vec = self + .clients + .iter() + .filter(|(_, client)| { + let ends = client.borrow(); + !ends.outbox.is_empty() || ends.writer_gone || ends.read_broken + }) + .map(|(id, _)| *id) + .collect(); + self.node.bridge(self.now, readable); self.pump(); } @@ -498,6 +510,7 @@ impl Net { let (client, pipes) = client(); let remote = Endpoint { addr: self.far.addr, ..peer() }; let id = self.node.connect(self.now, remote, timeout, pipes).expect("a route to the peer"); + self.clients.push((id, client.clone())); self.pump(); (id, client) } @@ -640,7 +653,8 @@ fn a_connect_closed_before_its_answer_is_answered_closed() { // A batch of frames and a deadline each end in a pass of their own: an opportunity's pass is over // the connects only, so neither test offers one. The batch's pass is one, after its last frame: -// what the batch brought is one write to the client's pipe, and its send pipe is asked once. +// what the batch brought is one write to the client's pipe, and its send pipe, found empty as the +// connect was answered, is not asked. #[test] fn a_batch_of_frames_moves_an_established_stream_once_with_no_opportunity_after_it() { let (mut net, _, client) = established(); @@ -649,7 +663,51 @@ fn a_batch_of_frames_moves_an_established_stream_once_with_no_opportunity_after_ net.node.receive(net.now, 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, 1), "the pipes' calls in the batch's pass"); + 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. +#[test] +fn a_batch_stops_at_its_bound_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"); + 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"); + assert_eq!(client.borrow().inbox, b"againonce"); +} + +// A send pipe found empty is asked again only once the kernel says it holds bytes: a batch of +// frames, a deadline and the to-client pipe's room each pass over the stream without reading it. +#[test] +fn a_send_pipe_found_empty_is_read_again_only_once_the_kernel_says_it_holds_bytes() { + let (mut net, id, client) = established(); + let reads = client.borrow().reads; + client.borrow_mut().outbox.extend(b"request"); + let frame = net.far.text(b"response"); + net.deliver(&frame); + net.fire(net.now.after(Duration::from_secs(1))); + net.node.bridge(net.now, []); + net.pump(); + assert_eq!((client.borrow().reads - reads, net.far.received.len()), (0, 0)); + + net.node.bridge(net.now, [id]); + net.pump(); + assert_eq!(net.far.received, b"request"); + assert_eq!(client.borrow().reads - reads, 2, "the request, then the pipe found empty again"); } #[test] @@ -670,7 +728,7 @@ fn nothing_leaves_the_pipe_that_the_stack_will_not_take() { let sent = text(100_000); client.borrow_mut().outbox.extend(&sent); // A pass with no transmit opportunity after it: the send buffer fills and nothing leaves it. - net.node.bridge(net.now); + net.node.bridge(net.now, [id]); assert_eq!(client.borrow().outbox.len(), sent.len() - SEND_BUFFER, "what [tcp] had no room for is still the pipe's"); assert_eq!(net.watch(id), Some(Watch { readable: false, ..IDLE })); @@ -757,7 +815,7 @@ fn room_in_the_pipe_opens_the_window_at_the_next_opportunity() { assert!(net.run_until(Duration::from_secs(1), shut), "all of it acknowledged, and the window shut: {} {}", net.far.acked, net.far.window); client.borrow_mut().room = 65_536; - net.node.bridge(net.now); + net.node.bridge(net.now, []); assert!(client.borrow().inbox == sent); assert_eq!(net.opportunity(), 1, "the window update"); let update = net.far.segments.last().expect("a segment"); @@ -1054,9 +1112,10 @@ fn a_client_that_left_is_finished_by_the_stack() { let (mut net, id, client) = established(); client.borrow_mut().outbox.extend(b"last words"); client.borrow_mut().writer_gone = true; + // The kernel's two answers to a client's leaving, one an end. net.node.pipe_gone(net.now, id, PipeEnd::ToClient); - assert_eq!((dropped(&client), net.node.streams()), ((true, true), 0), "nothing of the client's is left to serve"); net.node.pipe_gone(net.now, id, PipeEnd::FromClient); + assert_eq!((dropped(&client), net.node.streams()), ((true, true), 0), "nothing of the client's is left to serve"); net.pump(); assert_eq!((net.far.received.as_slice(), net.far.fin), (&b"last words"[..], true)); let frame = net.far.fin(); diff --git a/userland/netstack/src/card.rs b/userland/netstack/src/card.rs index 38b7e31dce1..e95a53dfaca 100644 --- a/userland/netstack/src/card.rs +++ b/userland/netstack/src/card.rs @@ -54,7 +54,6 @@ impl Card { /// while it holds an undrained interrupt, so a pass that saw the token and /// left it would find the same one on the next `wait` and every one after /// it. - /// This is also where a driver with a per-pass budget gets it back. /// /// A claim that refuses the read for anything but `WouldBlock` is the /// kernel saying this function is no longer this process's: a fault at the diff --git a/userland/netstack/src/main.rs b/userland/netstack/src/main.rs index bc588329fbf..73960cddf28 100644 --- a/userland/netstack/src/main.rs +++ b/userland/netstack/src/main.rs @@ -371,8 +371,7 @@ fn main() { let mut accept_refused: u64 = 0; loop { - // First, because it is what makes the interrupt taken and what gives - // a driver with a per-pass receive budget that budget back. + // First, because it is what makes the interrupt taken. if let Some(link) = card.begin_pass() { // A change of state only: a speed change is no new network. if link.is_up() != link_up { @@ -380,7 +379,7 @@ fn main() { node.link(clock(), link_up, draw); } } - node.receive(clock(), |sink| card.rx(sink), draw); + let owed = node.receive(clock(), |sink| card.rx(sink), draw); let now = clock(); if node.next_deadline().is_some_and(|at| at <= now) { node.fire(now, draw); @@ -417,9 +416,14 @@ fn main() { } // The node's next deadline: a retransmission, a lease's timer, a - // lookup's wait, a connect's. Zero when one is due. + // lookup's wait, a connect's. Zero when one is due, and when the + // receive stopped at its bound: the interrupts of the frames it left + // were taken with this pass's. let nanos = |left: Duration| u64::try_from(left.as_nanos()).unwrap_or(u64::MAX); let mut timeout = node.next_deadline().map_or(u64::MAX, |at| nanos(at.since(clock()))); + if owed { + timeout = 0; + } if let Some(left) = leases.due_in() { timeout = timeout.min(nanos(left)); } diff --git a/userland/netstack/src/serve.rs b/userland/netstack/src/serve.rs index 478cd9b2ffd..7a95dbf023e 100644 --- a/userland/netstack/src/serve.rs +++ b/userland/netstack/src/serve.rs @@ -115,6 +115,8 @@ pub struct Sockets { /// A stream's pipe was answered ready in this wake: one pass over the /// streams follows the wake, however many answers it carried. bridge: bool, + /// The streams whose send pipe was answered readable in this wake. + readable: Vec, } /// The two ends a connect or an accept moves, each split into the node's half @@ -216,6 +218,7 @@ impl Sockets { receiving: BTreeMap::new(), places, bridge: false, + readable: Vec::new(), } } @@ -785,7 +788,12 @@ impl Sockets { node.pipe_broken(now, ends.stream, end); } Ok(met) if met & OTHER_END_GONE != 0 => node.pipe_gone(now, ends.stream, end), - Ok(_) => self.bridge = true, + Ok(_) => { + if end == PipeEnd::FromClient { + self.readable.push(ends.stream); + } + self.bridge = true; + } } } TOKEN_LISTENER => { @@ -829,7 +837,7 @@ impl Sockets { /// The pass over the streams the last wake's answers asked for, once. pub fn bridge(&mut self, node: &mut Node, now: Instant) { if std::mem::take(&mut self.bridge) { - node.bridge(now); + node.bridge(now, self.readable.drain(..)); } }