diff --git a/issues/toyos-has-its-own-network-stack.md b/issues/toyos-has-its-own-network-stack.md index 0e239de0d8a..b7f1571f2de 100644 --- a/issues/toyos-has-its-own-network-stack.md +++ b/issues/toyos-has-its-own-network-stack.md @@ -17,13 +17,19 @@ netd runs smoltcp. Its replacement is ToyOS's own stack, built clean-room: reade - Then multi-core, netring (blocked on the owner's ABI ruling), TCP and IP hardening, IPv6, offloads, soak. - This track owns the T14's outbound rows, the machine reaching its router and the internet on the I219, judged from the stick: they arrive on this stack and not on smoltcp (owner: "no smoltcp."), and until they do no T14 row reads the wired card (`issues/the-host-cannot-reach-the-t14-while-it-runs-toyos.md`). -Stage 5 lands in slices, the orchestrator's cut under the owner's words "i want smoltcp out as fast as possible": netd's decisions go into `toyos-net-node` (`userland/netstack/node`), pure and host-tested and shipped in nothing, and then one change moves netd onto it and deletes smoltcp. In the tree: the node, its DHCP lease, its clients' datagram sockets, its mDNS name and its resolver. Still to build on it: streams, listeners; then the move. +Stage 5 lands in slices, the orchestrator's cut under the owner's words "i want smoltcp out as fast as possible": netd's decisions go into `toyos-net-node` (`userland/netstack/node`), pure and host-tested and shipped in nothing, and then one change moves netd onto it and deletes smoltcp. In the tree: the node, its DHCP lease, its clients' datagram sockets, its mDNS name, its resolver and streams. Still to build on it: listeners; then the move. What the node does not yet meet: - Of what a third party wrote, only slirp's recorded OFFER and ACK have reached it (`userland/netstack/node/tests/slirp.rs`); every other frame it has answered is the tests' own, from the RFCs' layouts. Exit: netd runs on it against slirp in a guest and a router on the T14. - It counts a DHCP message [udp] refused as `node.dhcp-unsent`, whatever the rule; the `dhcp.renew-unroutable` scenario owed above is not written. Exit: that scenario names the counter, or the node counts the renewal apart. - With the link down the DHCP client keeps its timers: every 4 to 64 s the node has a deadline, builds a DISCOVER that [udp] refuses, and counts it in `dhcp.tx.discover` and `node.dhcp-unsent`, from `Node::new` on. Exit: `toyos-dhcp`'s client is told the link went down and waits for it, and the node's first DISCOVER is the one that leaves. +- No second TCP has been on the far end of a stream: the peer of `userland/netstack/node/tests/streams.rs` is the tests' own script, which acknowledges what arrives in order and loses, reorders and repeats nothing; what is not ours there is `etherparse`, which reads every segment the node emits and builds every one it receives. Exit: netd runs on the node against the host kernel's TCP through slirp in a guest. +- The node has no bound on its streams, where netd refuses a connect past `max_piped_connections`, and a connection whose client let go is [tcp]'s alone and counted by nobody; [tcp] has no table limit of its own, and each connection is two 64 KiB buffers of netd's. Exit, before netd runs on the node: a connect past a bound is refused, and the bound counts connecting streams and the connections [tcp] finishes alone. +- A stream its client can see no more, its reader gone or the peer's FIN read through and its client writing no more, is reset once its pipe has given up no byte for 100 s, and a peer address restarts that clock for at most 16 such streams at once (`OWNERLESS_PER_PEER`, `userland/netstack/node/src/streams.rs`); one past them has 100 s from the pass that found it so, until one of the 16 is done and it takes its place. What holds in this stage, from the code: per address, 16, an estimate no measurement set. There is no floor on the rate: each of the 16 is kept for as long as its pipe gives up one byte per 100 s, which for a full 2 MiB pipe is 2,097,152 x 100 s, over six years, and then for its tail in [tcp] (the line below on a client gone with nothing left in its pipe). There is no bound across addresses: addresses are not counted, so the node holds 16 times as many as there are addresses holding one, and where a connect's address is its client's choice, an accepted stream's is the peer's, which on the link answers for as many addresses as it likes. That bound is the stream bound of the line above, built one stage on as the node's places, of which every stream holds one until the node lets it go; this stage builds no second one. `issues/netstack-cuts-a-departed-clients-unsent-tail-at-the-ceiling.md` stays open until netd runs on the node: on a host its exit's second clause is met (`a_departed_clients_tail_arrives_whole_while_its_peer_takes_it`) and its first for 16 streams an address. Exit: a test of the node's places in which peers at more addresses than the node has places for each hold 16 such streams, and the node holds no stream past its places; the T14 or a guest shows a peer holding one at a byte per 100 s, and the cut takes a floor on the rate, or the owner rules it needs none; and the move closes that issue with its bench row. +- A stream its peer reset is gone from the node at once, so for a client that still holds it `Node::set_nodelay` answers `false` and `Node::nodelay` `None`: `issues/a-stream-its-peer-reset-refuses-the-option-requests-a-host-answers.md`, carried by the node as built. std's `nodelay()` asks netd nothing: it answers from the value its `TcpStream` holds (`sdk/std/sys/net/connection.rs`). What the node carries is `set_nodelay` on a reset stream, refused where macOS refuses it under another kind and Linux answers, and libc's `getsockopt` of `TCP_NODELAY`, which asks through `toyos::net::tcp_get_option` (`userland/libc/src/socket.rs`) and is refused where both hosts answer. The node cannot hold the stream for the client: its two pipe ends are all it has of the client, letting both go is how the client learns of the reset, and with both gone nothing tells it when the client leaves, so a kept stream would be kept for ever. Exit, outside the node: that issue's guest test is green, and a `getsockopt` of `TCP_NODELAY` on a stream its peer reset is answered, by libc from a value its socket holds as std's does or by a pipe ABI that gives netd a handle whose other end's leaving the kernel reports. +- Every frame and every deadline ends in a pass over every stream, every transmit opportunity walks every stream twice (once for the count each peer address keeps alive, once to pass those whose connect is not answered), and `Node::next_deadline` reads every stream: linear in streams per frame, as netd's `bridge_piped` is today. For an idle stream a pass is one `recv_with`, two `status` and one read of its client's pipe, which in netd is a system call: 1,000 system calls a frame with 1,000 idle streams. A pass's `recv_with` also re-files its connection's deadline and offers it to the transmit round, so one frame makes every idle stream eligible and the next opportunity serves each to learn it has nothing. [tcp] cannot say which connections a segment, a timer, an ICMP error or a failed next hop moved: it reports `drain_eligible` and `drain_gone`, which are the round's. The change: [tcp] marks a held connection in `for_conn`, `tick`, `icmp` and `end` and hands the marks out once (`drain_moved`); the node maps a connection to its stream, keeps its deadlines ordered, and passes only the streams [tcp] moved, the one a pipe's wake names, and those due. Exit: at the move, netd's CPU time per frame on the T14 with 1, 100 and 1,000 idle streams beside one bulk stream is measured, and the change lands if 1,000 idle streams cost more than 12.3 µs a frame over what one costs. The 12.3 µs is the time a full frame takes on a wire of 1 Gbit/s, 1,538 bytes of 8 ns with its preamble and gap: past it the passes alone take longer than the bulk stream's frames take to arrive. It is a threshold by arithmetic and no measurement; by estimate, not measured, 1,000 system calls exceed it. +- A client that is gone with nothing left in its pipe is finished by [tcp]'s rules for a user who let go: a reset once it has been idle 60 s, where netd on smoltcp reset it 100 s after the client left whatever the peer said. A peer that acknowledges a byte a minute holds such a connection for as long as it has bytes to acknowledge. Exit: the T14 or a guest shows such a peer holding one, and [tcp]'s rule takes a bound from the client's leaving; or the owner rules the idle bound is the one. - The word each of [udp]'s refusals is answered in on the pipe ABI is `Refused` in `userland/netstack/node/src/datagram.rs`, tested by `each_refusal_is_answered_in_the_pipes_word_for_it` against no reader's scenario; `udp.no-ephemeral-port` has a word and no test that reaches it. Exit: the scenario owed below exists, and the test names its id. - A client's datagram to a broadcast address is refused `udp.broadcast-not-permitted` and answered invalid input, where smoltcp sends it: the pipe ABI has no request that permits it, and std's `set_broadcast` (`sdk/std/sys/net/connection.rs`) answers `Ok` to a permission nothing grants. No program in the tree sends to a broadcast address or sets the option; the sources of the third-party C programs an image builds from a recipe have not been searched. The owner's ruling: "Of course carry the permission why would you even ask. The premise is and always has been to do things properly". The move of netd onto the node does not land with this open. Exit, both before the move, as a stage of their own: that search, for `SO_BROADCAST` and a `sendto` to a broadcast address, is posted with its command and output; and the pipe ABI carries the permission from std's and libc's setters to `Udp::set_broadcast`, a send refused for want of it answered in a word that says permission and not input. - A refusal [udp] logs against a client's call waits in the stack until the node's next `receive`, `fire` or `link`, because only those pass over the stack's reports. Exit, at the move and not after it: the datagram calls end in that pass, or netd drains it once per turn of its loop. @@ -59,6 +65,7 @@ What stage 4 departs from its specifications: - [ip]'s `Event::Cleared` and `Event::Room` are not in the specifications. Exit: they name them. - The readers' stage 4 design has a flow waiting for its next hop keep its place in the round. [shard]'s round lets it go, deficit and all, and [tcp] offers it again when woken, at the round's tail, so a waiting flow costs the round nothing. Exit: a `toyos-net-shard/tests/drr.rs` test finds a woken flow served in the place it held when it began to wait. - The readers' stage 4 design has a flow leave the round once it has nothing eligible, its deficit reset. [udp] says so as it hands out a sender's last datagram; [tcp] learns it only when it serves the connection, since only its `next_segment` decides what is due, so a connection whose turn ended on its last segment stays in the round with its deficit until its next turn, and what it queues before then is sent against that deficit, debt included, as RFC 8290 §4.2 keeps an emptied queue until it is next selected. Exit: [tcp] reports at hand-off that nothing is left, and a `toyos-net-shard/tests/drr.rs` test finds such a connection out of the round, its deficit reset. +- `Tcp::recv_with`, the read of a reader that may take less than it is shown, is not in the specifications: the node's bridge to a client's pipe needs it, since a byte `recv` copied out has opened the window and cannot be put back. Its tests (`toyos-net-shard/tcp/tests/take.rs`) name no scenario. Exit: the specification names the call and its scenarios, and the tests take their ids. - The readers' TCP specification disagrees with itself, and [tcp] follows its user timeout: one replaces R2 and runs only while sent data is unacknowledged or a zero window holds data back, so with one set a connection whose data never left, for want of credit or of a next hop, has no give-up, where its rule that give-up clocks run on wall time has a device that never offers credit end its connections. Exit: RFC 9293 §3.10.8 ends a connection in any state once its user timeout expires: a test that sets one and withholds credit, as `s_pl_009_give_up_without_credit` does without one, finds the connection ended. Open for the owner. On 2026-10-03 he chose "Fair scheduler now": "Build the byte-fair scheduler as the network spec describes, about 280 lines, and decide later on the laptop whether new connections should jump the queue." It is built as one round, in the order flows became eligible, and nobody jumps it. The track reads a queue that new connections jump as RFC 8290's new-flows list, and offers him this evidence: a sparse flow such as a DNS query waits one turn of every flow ahead of it in the round, up to a quantum of 1,514 bytes each, where [udp]'s datagrams and [tcp]'s segments alternated frame by frame before, so a query waited at most one TCP frame: in DRR-02 a 100-byte segment that becomes eligible leaves behind a full frame of each of the three bulk flows ahead of it (`s_shard_drr_002_a_bulk_flow_starves_no_other_beyond_its_quantum`); a T14 measurement at stage 5 of a sparse flow's wait behind bulk flows would add to it. Exit: the owner rules on whether new connections jump the queue. diff --git a/toyos-net-shard/src/lib.rs b/toyos-net-shard/src/lib.rs index 05c4b94b85c..1c64171f249 100644 --- a/toyos-net-shard/src/lib.rs +++ b/toyos-net-shard/src/lib.rs @@ -489,6 +489,19 @@ impl Shard { done } + /// [`Tcp::recv_with`]: `take` is shown the oldest bytes held and answers how many it took. + pub fn recv_with(&mut self, now: Instant, id: ConnId, take: impl FnOnce(&[u8]) -> usize) -> Result { + let done = self.tcp.recv_with(now, id, take); + self.settle(now); + done + } + + pub fn set_options(&mut self, now: Instant, id: ConnId, options: toyos_net_tcp::Options) -> Result<(), toyos_net_tcp::Error> { + let done = self.tcp.set_options(now, id, options); + self.settle(now); + done + } + pub fn shutdown_write(&mut self, now: Instant, id: ConnId) -> Result<(), toyos_net_tcp::Error> { let done = self.tcp.shutdown_write(now, id); self.settle(now); diff --git a/toyos-net-shard/tcp/src/conn.rs b/toyos-net-shard/tcp/src/conn.rs index 3bf2354bee8..b0086dfff11 100644 --- a/toyos-net-shard/tcp/src/conn.rs +++ b/toyos-net-shard/tcp/src/conn.rs @@ -865,6 +865,18 @@ impl Sync { } } + /// [`Self::recv`] for a reader that may take less than it is shown. + pub fn recv_with(&mut self, take: impl FnOnce(&[u8]) -> usize) -> Result { + if self.rx.unread() == 0 { + return if self.rx.closed { Ok(Received::End) } else { Err(Error::WouldBlock) }; + } + let n = self.rx.read_with(take); + if n > 0 { + self.rx.after_read(self.smss()); + } + Ok(Received::Data(n)) + } + pub fn shutdown_write(&mut self, now: Instant) { if self.tx.fin.is_some() { return; diff --git a/toyos-net-shard/tcp/src/rx.rs b/toyos-net-shard/tcp/src/rx.rs index 45ff8cf13bd..8a5900d80f0 100644 --- a/toyos-net-shard/tcp/src/rx.rs +++ b/toyos-net-shard/tcp/src/rx.rs @@ -338,6 +338,15 @@ impl Rx { self.buf.read(out) } + /// Shows `take` the oldest bytes held, as far as they lie in one piece, and lets go of as + /// many of them as it answers it took. + pub fn read_with(&mut self, take: impl FnOnce(&[u8]) -> usize) -> usize { + let (held, _) = self.buf.slices(0, self.buf.len()); + let n = take(held).min(held.len()); + self.buf.consume(n); + n + } + /// `shutdown_read`: what is held is dropped, and later text is dropped as if read. pub fn stop_reading(&mut self) { self.discard = true; diff --git a/toyos-net-shard/tcp/src/stack.rs b/toyos-net-shard/tcp/src/stack.rs index 3e86ba5954f..3bdcbb9afde 100644 --- a/toyos-net-shard/tcp/src/stack.rs +++ b/toyos-net-shard/tcp/src/stack.rs @@ -998,6 +998,25 @@ impl Tcp { result } + /// [`Self::recv`] for a reader that may take less than it is shown, such as a pipe that is + /// nearly full: `take` sees the oldest bytes held, as far as they lie in one piece, and answers + /// how many of them it took; no other byte leaves the buffer. With nothing held it is not + /// called, and the answer is [`Self::recv`]'s; `Data(0)` is a reader that took nothing. + pub fn recv_with(&mut self, now: Instant, id: ConnId, take: impl FnOnce(&[u8]) -> usize) -> Result { + let conn = self.conn(id)?; + let result = match &mut conn.state { + Tcb::SynSent(_) | Tcb::SynRcvd(_) => Err(Error::WouldBlock), + Tcb::Sync(sync) => sync.recv_with(take), + Tcb::Ended(Ended { failure: Some(failure), .. }) => Err(Error::Failed(*failure)), + Tcb::Ended(Ended { failure: None, rx }) => match rx.as_mut() { + Some(rx) if rx.unread() > 0 => Ok(Received::Data(rx.read_with(take))), + _ => Ok(Received::End), + }, + }; + self.settle(id.index, now); + result + } + pub fn shutdown_write(&mut self, now: Instant, id: ConnId) -> Result<(), Error> { let conn = self.conn(id)?; let result = match &mut conn.state { diff --git a/toyos-net-shard/tcp/tests/take.rs b/toyos-net-shard/tcp/tests/take.rs new file mode 100644 index 00000000000..3e3567abdca --- /dev/null +++ b/toyos-net-shard/tcp/tests/take.rs @@ -0,0 +1,130 @@ +//! `recv_with`: the read of a reader that may take less than it is shown. No scenario ids: the +//! specifications hold no such call, which the track lists among stage 4's departures; their +//! receive scenarios are `receive.rs`'s, on `recv`, and these hold `recv_with` to the same buffer +//! and the same window update. From E. + +mod common; + +use common::*; +use toyos_net_tcp::{Error, Received, State}; + +/// A read whose reader answers `count`, without letting the clock move: the answer, what the +/// reader was shown, and what the read made A send. +fn take(h: &mut H, count: usize) -> (Result, Vec, Vec) { + let mut shown = Vec::new(); + let (r, outs) = h.call(h.t, |tcp, now, id| { + tcp.recv_with(now, id, |held| { + shown = held.to_vec(); + count + }) + }); + (r, shown, outs) +} + +/// The text a well-behaved peer sends in `[from, to)`. +fn text(from: u32, to: u32) -> Vec { + (from..to).map(pattern).collect() +} + +#[test] +fn what_the_reader_did_not_take_stays_held_in_order() { + let mut h = fixture_e(); + h.input(0, seg(5001).ack(1001).len(100)); + let (r, shown, _) = take(&mut h, 40); + assert_eq!((r, shown), (Ok(Received::Data(40)), text(5001, 5101))); + assert_eq!(h.status().readable, 60); + let (r, shown, _) = take(&mut h, 0); + assert_eq!((r, shown), (Ok(Received::Data(0)), text(5041, 5101)), "a reader that took nothing"); + assert_eq!(h.read(100), text(5041, 5101)); +} + +#[test] +fn with_nothing_held_the_reader_is_not_asked() { + let mut h = fixture_e(); + let mut asked = false; + let (r, _) = h.call(h.t, |tcp, now, id| { + tcp.recv_with(now, id, |_| { + asked = true; + 0 + }) + }); + assert_eq!((r, asked), (Err(Error::WouldBlock), false)); + h.input(0, seg(5001).ack(1001).fin()); + let (r, _) = h.call(h.t, |tcp, now, id| { + tcp.recv_with(now, id, |_| { + asked = true; + 0 + }) + }); + assert_eq!((r, asked), (Ok(Received::End), false)); +} + +// The window update RX-13 has a read owe: 65,535 unread bytes shut the window, and a take that +// moves the edge by 1,500 from a window of 0 is acknowledged at once. +#[test] +fn a_take_owes_the_window_update_a_read_does() { + let mut h = fixture_e(); + let mut seq = 5001u32; + while seq < 70_536 { + let n = (70_536 - seq).min(1460); + h.arrive(seg(seq).ack(1001).len(n as usize)); + seq += n; + } + h.at(1); + let (r, _, outs) = take(&mut h, 1000); + assert_eq!(r, Ok(Received::Data(1000))); + nothing(&outs); + let (r, _, outs) = take(&mut h, 500); + assert_eq!(r, Ok(Received::Data(500))); + expect(&outs, &["ACK=70536 WND=1500"]); +} + +// The buffer is a ring of 65,535 bytes: text that wraps it is shown in two pieces, and a reader +// that claims more than it was shown takes the piece and no byte of the next. +#[test] +fn text_across_the_rings_wrap_is_shown_in_two_pieces_and_none_is_lost() { + let mut h = fixture_e(); + let mut seq = 5001u32; + while seq < 70_536 { + let n = (70_536 - seq).min(1460); + h.arrive(seg(seq).ack(1001).len(n as usize)); + seq += n; + } + h.at(1); + let (r, _, _) = take(&mut h, 65_000); + assert_eq!(r, Ok(Received::Data(65_000))); + h.input(2, seg(70_536).ack(1001).len(1000)); + assert_eq!(h.status().readable, 1535); + + let mut pieces = Vec::new(); + loop { + match take(&mut h, usize::MAX) { + (Ok(Received::Data(n)), shown, _) => { + assert_eq!(n, shown.len(), "the count is what was shown"); + pieces.push(shown); + } + (Err(Error::WouldBlock), _, _) => break, + (other, _, _) => panic!("{other:?}"), + } + } + assert_eq!(pieces.iter().map(Vec::len).collect::>(), [535, 1000]); + assert_eq!(pieces.concat(), text(70_001, 71_536)); +} + +// Both FINs end the connection (RFC 9293 §3.6, case 1), and what the peer sent before its own is +// still the reader's: shown, taken, and only then the end. +#[test] +fn text_unread_when_both_fins_ended_the_connection_is_still_taken() { + let mut h = fixture_e(); + let (r, _) = h.call(0, |tcp, now, id| tcp.shutdown_write(now, id)); + r.unwrap(); + h.input(10, seg(5001).ack(1002)); + h.input(20, seg(5001).ack(1002).len(100).fin()); + assert_eq!(h.status().state, State::TimeWait); + let (r, shown, _) = take(&mut h, 40); + assert_eq!((r, shown), (Ok(Received::Data(40)), text(5001, 5101))); + let (r, shown, _) = take(&mut h, usize::MAX); + assert_eq!((r, shown), (Ok(Received::Data(60)), text(5041, 5101))); + let (r, shown, _) = take(&mut h, usize::MAX); + assert_eq!((r, shown), (Ok(Received::End), Vec::new())); +} diff --git a/userland/netstack/node/Cargo.toml b/userland/netstack/node/Cargo.toml index b2e5663d211..9da7f469b45 100644 --- a/userland/netstack/node/Cargo.toml +++ b/userland/netstack/node/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "toyos-net-node" -description = "Netstack's decisions, pure: one interface of ToyOS's own stack, its DHCP lease, its clients' datagram sockets and its mDNS name, frames and time in, frames out against the device's credit." +description = "Netstack's decisions, pure: one interface of ToyOS's own stack, its DHCP lease, its clients' datagram sockets and streams, its mDNS name and its resolver, frames and time in, frames out against the device's credit." version = "0.1.0" edition = "2021" license = "MIT OR Apache-2.0" diff --git a/userland/netstack/node/src/lease.rs b/userland/netstack/node/src/lease.rs index 625ad8624bc..70af5c6b884 100644 --- a/userland/netstack/node/src/lease.rs +++ b/userland/netstack/node/src/lease.rs @@ -13,6 +13,8 @@ //! [ip] reports on no address but the one this module added, and one report at a time: each is //! read against the state as it stands. +mod tcp; + use alloc::collections::VecDeque; use core::net::Ipv4Addr; diff --git a/userland/netstack/node/src/lease/tcp.rs b/userland/netstack/node/src/lease/tcp.rs new file mode 100644 index 00000000000..51cc26cae19 --- /dev/null +++ b/userland/netstack/node/src/lease/tcp.rs @@ -0,0 +1,64 @@ +//! [tcp]'s calls on the interface, for `streams`. They reach no address, gateway or resolver: the +//! shard stays `lease`'s to configure. + +use toyos_net_shard::ConnectError; +use toyos_net_tcp::{ConnId, Endpoint, Error, Options, Received, Status}; +use toyos_net_wire::{Instant, Port}; + +use super::Stack; + +/// [tcp] refuses a caller nothing about a connection that caller holds. +fn held(answer: Result) -> T { + match answer { + Ok(value) => value, + Err(refusal) => unreachable!("[tcp] refused the holder of a connection: {refusal:?}"), + } +} + +impl Stack { + /// An active open from a port [tcp] draws. + pub(crate) fn tcp_connect(&mut self, now: Instant, remote: Endpoint) -> Result { + self.shard.connect(now, None, remote) + } + + pub(crate) fn tcp_status(&mut self, id: ConnId) -> Status { + held(self.shard.status(id)) + } + + pub(crate) fn tcp_local_port(&mut self, id: ConnId) -> Port { + held(self.shard.tuple(id)).local.port + } + + /// Queues `data`, all of it: the caller offers no more than [`Status::writable`]. + pub(crate) fn tcp_send(&mut self, now: Instant, id: ConnId, data: &[u8]) { + let taken = held(self.shard.send(now, id, data)); + if taken != data.len() { + unreachable!("[tcp] took {taken} of {} bytes it had room for", data.len()); + } + } + + /// [`toyos_net_tcp::Tcp::recv_with`]. + pub(crate) fn tcp_recv(&mut self, now: Instant, id: ConnId, take: impl FnOnce(&[u8]) -> usize) -> Result { + self.shard.recv_with(now, id, take) + } + + /// The connection's FIN, after what is queued. Of a synchronized connection that has not + /// failed. + pub(crate) fn tcp_shutdown_write(&mut self, now: Instant, id: ConnId) { + held(self.shard.shutdown_write(now, id)); + } + + /// The holder lets go, and [tcp] finishes the connection alone; `id` names nothing after. + pub(crate) fn tcp_close(&mut self, now: Instant, id: ConnId) { + held(self.shard.close(now, id)); + } + + /// RFC 9293 §3.10.5; `id` names nothing after. + pub(crate) fn tcp_abort(&mut self, now: Instant, id: ConnId) { + held(self.shard.abort(now, id)); + } + + pub(crate) fn tcp_set_options(&mut self, now: Instant, id: ConnId, options: Options) { + held(self.shard.set_options(now, id, options)); + } +} diff --git a/userland/netstack/node/src/lib.rs b/userland/netstack/node/src/lib.rs index ae7ef358aad..cd211878c25 100644 --- a/userland/netstack/node/src/lib.rs +++ b/userland/netstack/node/src/lib.rs @@ -6,7 +6,7 @@ //! //! The node carries out what the client asks and tells it what became of its address. What the //! interface holds is `lease`'s to write and nobody else's: see that module for the three rules -//! its types keep. +//! its types keep. A client's TCP connection and its two pipes are `streams`'. //! //! A client's datagram sockets are `datagram`'s, the machine's `.local` name is `name`'s //! and the lookups of other machines' names are `resolve`'s: each reads the lease and writes none @@ -38,6 +38,9 @@ mod datagram; mod lease; mod name; mod resolve; +mod streams; + +pub use streams::{FromClient, PipeEnd, Pipes, ReadRefusal, StreamEvent, StreamId, ToClient, Watch, WriteRefusal}; use alloc::vec; use alloc::vec::Vec; @@ -77,6 +80,7 @@ pub struct Node { events: Vec, /// Room for the largest message the client accepts. datagram: Vec, + streams: streams::Streams, } impl Node { @@ -94,6 +98,7 @@ impl Node { counters: Counters::default(), events: Vec::new(), datagram: vec![0; usize::from(toyos_dhcp::limits::MAX_MESSAGE)], + streams: streams::Streams::default(), }) } @@ -126,12 +131,15 @@ impl Node { pub fn receive(&mut self, now: Instant, frame: &[u8], mut draw: impl FnMut() -> u32) { self.stack.receive(now, frame); self.settle(now, &mut draw); + self.bridge(now); } /// A transmit opportunity with room for `credit` frames, each handed to `sink` as it is built. /// Returns how many left. pub fn transmit(&mut self, now: Instant, credit: usize, sink: impl FnMut(&[u8])) -> usize { - self.stack.transmit(now, credit, sink) + let sent = self.stack.transmit(now, credit, sink); + self.pass(now, true); + sent } /// The link came up or went down; the caller reports a change, not a state. Down, a held @@ -149,7 +157,7 @@ impl Node { pub fn next_deadline(&self) -> Option { let name = self.name.as_ref().and_then(name::Name::next_deadline); - self.stack.next_deadline().into_iter().chain(self.client.next_deadline()).chain(name).chain(self.resolver.next_deadline()).min() + self.stack.next_deadline().into_iter().chain(self.client.next_deadline()).chain(name).chain(self.resolver.next_deadline()).chain(self.streams.next_deadline()).min() } /// Every deadline at or before `now`; the frames they make due wait for [`Self::transmit`]. @@ -159,6 +167,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); } /// Hands the client what the shard reported and what reached its socket, and carries out diff --git a/userland/netstack/node/src/streams.rs b/userland/netstack/node/src/streams.rs new file mode 100644 index 00000000000..c940e0daeef --- /dev/null +++ b/userland/netstack/node/src/streams.rs @@ -0,0 +1,411 @@ +//! Streams: a connection of [tcp]'s and the two pipes its client reads and writes it by. The pipe +//! ends are traits, so the kernel's calls stay in netstack and the tests fake them. +//! +//! **The node holds a stream for as long as it holds one of its pipes.** A pass +//! ([`Node::bridge`]) moves what each side takes and lets go of an end that is finished: the +//! to-client end once the peer's FIN or a failure has been read through it or its reader is gone, +//! the from-client end once its last byte is queued and the FIN after it, or the connection is +//! over. With neither end left the node closes the connection, which [tcp] then finishes alone +//! by its own rules for a user who let go (a reset if text is unread, RFC 9293 §3.6.1), and the +//! stream's id names nothing. +//! +//! **Nothing leaves the stack that the client's pipe did not take**, and **nothing leaves the +//! pipe that the stack will not take**: a byte moved is a byte acknowledged to whoever sent it, +//! so one the other side refused would be cut out of the middle of the stream with nothing +//! saying so. What a full pipe refuses stays in [tcp], whose window closes on the peer; what +//! [tcp] has no room for stays in the pipe, which blocks the client. +//! +//! **A stream its client can see no more leaves its unsent bytes [`OWNERLESS_LIFE`] without +//! progress.** That is a stream whose to-client end the node let go, whichever way, and whose +//! client writes no more: no byte and no end reaches the client through it again, and nothing +//! tells the node whether that client still lives. The peer's FIN read through counts as much as +//! a reader that left, since behind it the node cannot learn of the leaving. The bytes the pipe +//! still holds are sent as [tcp] makes room, and the connection is reset once the pipe has given +//! up no byte for [`OWNERLESS_LIFE`]. A peer that takes a byte now and then would so hold such a +//! connection for as long as it liked (RFC 9293 §3.8.6.1 lets one be reclaimed), so a peer +//! address restarts that clock for at most [`OWNERLESS_PER_PEER`] streams at once; any other has +//! [`OWNERLESS_LIFE`] from the pass that first found it so, whatever it gives up, until it +//! becomes one of them: a pass over the streams that begins with fewer makes it one, the oldest +//! stream first. The count is an address's and addresses are not counted, so how many such +//! streams the node holds in all is not bounded here, nor is how slowly a peer may take. A +//! client that still holds its reading end, or may still write, is never timed here: what it +//! cannot send is [tcp]'s to give up on. +//! +//! Every call that can move a stream ends in a pass: a frame, 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. + +use alloc::boxed::Box; +use alloc::collections::{BTreeMap, VecDeque}; +use core::net::Ipv4Addr; +use core::time::Duration; + +use toyos_net_shard::ConnectError; +use toyos_net_tcp::{ConnId, Endpoint, Error, Failure, Options, Received, State}; +use toyos_net_wire::{Instant, Port}; + +use crate::lease::Stack; +use crate::Node; + +/// How long the pipe of a stream its client can see no more may give up no byte: R2, the time +/// RFC 9293 §3.8.3 gives a segment's retransmission before the connection is closed, at the 100 +/// seconds it asks for at least. +const OWNERLESS_LIFE: Duration = Duration::from_secs(100); + +/// How many such streams one peer address keeps alive at once by taking their bytes. An +/// estimate: no measurement sets it. It bounds an address, not the node, and not how slowly a +/// peer may take. +const OWNERLESS_PER_PEER: usize = 16; + +/// The most one read of a client's pipe takes. +const CHUNK: usize = 4096; + +/// Why the write end of a pipe took nothing. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum WriteRefusal { + Full, + /// Nobody holds the read end. + Gone, + /// Anything else: the handle is not the pipe end its client said it was. + Broken, +} + +/// Why the read end of a pipe gave nothing. A pipe with no writer left is no refusal: it gives +/// what it holds and then the end. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ReadRefusal { + /// Nothing to read, and a writer is left. + Empty, + /// Anything else: the handle is not the pipe end its client said it was. + Broken, +} + +/// The write end of the pipe a client reads its stream from. Dropped, the client reads the end. +pub trait ToClient { + /// Takes a prefix of `bytes` and answers its length. + fn write(&mut self, bytes: &[u8]) -> Result; +} + +/// The read end of the pipe a client writes its stream to. Dropped, the client's writes fail. +pub trait FromClient { + /// Fills a prefix of `out` and answers its length. 0 is the end: the pipe is empty and no + /// writer is left. + fn read(&mut self, out: &mut [u8]) -> Result; +} + +/// The two ends a client's connect hands over. +pub struct Pipes { + pub to_client: Box, + pub from_client: Box, +} + +/// One of a stream's two pipe ends. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum PipeEnd { + ToClient, + FromClient, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct StreamId(u64); + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum StreamEvent { + /// The answer to a connect: established, from this port. + Connected { id: StreamId, local: Port }, + /// The answer to a connect: the peer refused it or [tcp] gave it up. + Failed { id: StreamId, failure: Failure }, + /// The answer to a connect: its deadline passed first. + TimedOut { id: StreamId }, + /// The answer to a connect: its stream was closed first. + Closed { id: StreamId }, + /// A line for the log: reset with its bytes unsent and nobody left to see it, at the end of + /// its [`OWNERLESS_LIFE`]. + Cut { id: StreamId }, +} + +/// What netstack asks the kernel about an established stream's pipes until the next pass. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Watch { + /// Bytes in the from-client pipe: [tcp] has room for them. + pub readable: bool, + /// The from-client pipe's writer leaving, reported by [`Node::pipe_gone`]. + pub writer: bool, + /// Room in the to-client pipe: it refused bytes [tcp] still holds. + pub writable: bool, + /// The to-client pipe's reader leaving, reported by [`Node::pipe_gone`]. + pub reader: bool, +} + +struct Stream { + conn: ConnId, + /// The peer's address: what [`OWNERLESS_PER_PEER`] counts by. + remote: Ipv4Addr, + to_client: Option>, + from_client: Option>, + /// The connect is not answered yet, and no byte moves. + connecting: bool, + /// The connect's, while connecting; then the cut's, once the client can see the stream no + /// more. + deadline: Option, + /// The client writes no more: its writer is gone, it shut its sending half down or it + /// closed. An empty pipe is then the end. + done_writing: bool, + /// One of its peer address's [`OWNERLESS_PER_PEER`]: a byte given up restarts the cut's clock. + extended: bool, + /// The to-client pipe refused bytes [tcp] holds. + held: bool, + /// [tcp] had room for the client's bytes when the last pass ended. + room: bool, + /// The options last written to [tcp]. + options: Options, +} + +impl Stream { + /// One pass. `false` lets the stream go: its pipe ends drop with it, and its connection is + /// [tcp]'s alone or gone. `extended` is how many streams each peer address kept alive when + /// the pass over the streams began, and those it has made so since. + fn pass(&mut self, id: StreamId, now: Instant, stack: &mut Stack, events: &mut VecDeque, extended: &mut BTreeMap) -> bool { + let conn = self.conn; + if self.connecting { + let status = stack.tcp_status(conn); + let event = match (status.state, status.failure) { + (_, Some(failure)) => StreamEvent::Failed { id, failure }, + (State::SynSent | State::SynReceived, None) if self.deadline.is_some_and(|at| at <= now) => StreamEvent::TimedOut { id }, + (State::SynSent | State::SynReceived, None) => return true, + (_, None) => StreamEvent::Connected { id, local: stack.tcp_local_port(conn) }, + }; + events.push_back(event); + if !matches!(event, StreamEvent::Connected { .. }) { + stack.tcp_abort(now, conn); + return false; + } + self.connecting = false; + self.deadline = None; + } + + self.held = false; + while let Some(pipe) = self.to_client.as_mut() { + let mut refused = None; + let received = stack.tcp_recv(now, conn, |bytes| match pipe.write(bytes) { + Ok(taken) => taken, + Err(refusal) => { + refused = Some(refusal); + 0 + } + }); + match (received, refused) { + (Ok(Received::Data(taken)), None) if taken > 0 => {} + (Ok(Received::Data(_)), None | Some(WriteRefusal::Full)) => { + self.held = true; + break; + } + // Nobody reads what the peer sends. + (Ok(Received::Data(_)), Some(WriteRefusal::Gone)) => self.to_client = None, + (Ok(Received::Data(_)), Some(WriteRefusal::Broken)) => { + stack.tcp_abort(now, conn); + return false; + } + // The peer's FIN after its last byte, or the connection's failure: the client + // reads the end. + (Ok(Received::End) | Err(Error::Failed(_)), _) => self.to_client = None, + (Err(Error::WouldBlock), _) => break, + (Err(refusal), _) => unreachable!("[tcp] refused the holder of a connection a read: {refusal:?}"), + } + } + + let mut gave_up = false; + while let Some(pipe) = self.from_client.as_mut() { + // Never more than [tcp] has room for, and so never a read of no bytes, whose answer + // would be the end's. + let room = stack.tcp_status(conn).writable.min(CHUNK); + if room == 0 { + break; + } + let mut chunk = [0u8; CHUNK]; + let Some(space) = chunk.get_mut(..room) else { unreachable!("room is at most a chunk") }; + let read = match pipe.read(space) { + Ok(read) => read, + Err(ReadRefusal::Empty) if self.done_writing => 0, + Err(ReadRefusal::Empty) => break, + Err(ReadRefusal::Broken) => { + stack.tcp_abort(now, conn); + return false; + } + }; + if read == 0 { + stack.tcp_shutdown_write(now, conn); + self.from_client = None; + break; + } + let Some(bytes) = space.get(..read) else { unreachable!("a pipe read {read} bytes into room for {room}") }; + stack.tcp_send(now, conn, bytes); + gave_up = true; + } + + let status = stack.tcp_status(conn); + // A connection that is over takes no more of the client's bytes: its writes fail rather + // than fill a pipe nobody drains. + if matches!(status.state, State::Closed | State::TimeWait) { + self.from_client = None; + } + self.room = status.writable > 0; + if self.to_client.is_none() && self.from_client.is_none() { + stack.tcp_close(now, conn); + return false; + } + // Nothing of the stream reaches its client again, alive or not. + if self.to_client.is_none() && self.done_writing { + if !self.extended { + let kept = extended.entry(self.remote).or_insert(0); + if *kept < OWNERLESS_PER_PEER { + *kept = kept.saturating_add(1); + self.extended = true; + } + } + if self.deadline.is_none() || (self.extended && gave_up) { + self.deadline = Some(now.after(OWNERLESS_LIFE)); + } else if self.deadline.is_some_and(|at| at <= now) { + stack.tcp_abort(now, conn); + events.push_back(StreamEvent::Cut { id }); + return false; + } + } + true + } +} + +#[derive(Default)] +pub(crate) struct Streams { + live: BTreeMap, + /// The next stream's id: none is used twice. + next: u64, + events: VecDeque, +} + +impl Streams { + pub(crate) fn next_deadline(&self) -> Option { + self.live.values().filter_map(|stream| stream.deadline).min() + } +} + +impl Node { + /// An active open to `remote`, answered by a [`StreamEvent`] once the handshake ends or + /// `timeout` passes. Refused, nothing was sent and the pipe ends are dropped. + pub fn connect(&mut self, now: Instant, remote: Endpoint, timeout: Option, pipes: Pipes) -> Result { + let conn = self.stack.tcp_connect(now, remote)?; + let id = StreamId(self.streams.next); + self.streams.next = self.streams.next.saturating_add(1); + let stream = Stream { + conn, + remote: remote.addr, + to_client: Some(pipes.to_client), + from_client: Some(pipes.from_client), + connecting: true, + deadline: timeout.map(|within| now.after(within)), + done_writing: false, + extended: false, + held: false, + room: false, + options: Options::default(), + }; + self.streams.live.insert(id, stream); + Ok(id) + } + + /// One pass over every stream: netstack calls it when a pipe it watches is ready. + pub fn bridge(&mut self, now: Instant) { + self.pass(now, false); + } + + /// A pass over every stream, or over those whose connect is not answered. + pub(crate) fn pass(&mut self, now: Instant, connects: bool) { + let Streams { live, events, .. } = &mut self.streams; + let stack = &mut self.stack; + let mut extended = BTreeMap::new(); + for stream in live.values().filter(|stream| stream.extended) { + let peers = extended.entry(stream.remote).or_insert(0usize); + *peers = peers.saturating_add(1); + } + live.retain(|id, stream| (connects && !stream.connecting) || stream.pass(*id, now, stack, events, &mut extended)); + } + + /// The client lets go of the stream: nobody reads it, and what its pipe still holds is sent + /// before the FIN. A connect not yet answered is answered [`StreamEvent::Closed`]. + pub fn close(&mut self, now: Instant, id: StreamId) { + let Some(stream) = self.streams.live.get_mut(&id) else { return }; + if stream.connecting { + self.stack.tcp_abort(now, stream.conn); + self.streams.live.remove(&id); + self.streams.events.push_back(StreamEvent::Closed { id }); + return; + } + stream.to_client = None; + stream.done_writing = true; + self.bridge(now); + } + + /// The client sends no more: what its pipe still holds is sent, then the FIN. `false` is an + /// id that names no established stream. + 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); + true + } + + /// Nagle's algorithm off or on (RFC 9293 §3.7.4). `false` is an id that names no stream. + pub fn set_nodelay(&mut self, now: Instant, id: StreamId, nodelay: bool) -> bool { + let Some(stream) = self.streams.live.get_mut(&id) else { return false }; + stream.options.nodelay = nodelay; + self.stack.tcp_set_options(now, stream.conn, stream.options); + true + } + + pub fn nodelay(&self, id: StreamId) -> Option { + self.streams.live.get(&id).map(|stream| stream.options.nodelay) + } + + /// The kernel said nobody holds the other end of one of an established stream's pipes: the + /// to-client pipe has no reader, or the from-client pipe no writer, whatever it still holds. + pub fn pipe_gone(&mut self, now: Instant, id: StreamId, end: PipeEnd) { + let Some(stream) = self.streams.live.get_mut(&id).filter(|stream| !stream.connecting) else { return }; + match end { + PipeEnd::ToClient => stream.to_client = None, + PipeEnd::FromClient => stream.done_writing = true, + } + self.bridge(now); + } + + /// The kernel refused netstack's watch of a pipe end the node still holds: the handle is no + /// pipe end, and its client's connection is reset. + pub fn pipe_broken(&mut self, now: Instant, id: StreamId, end: PipeEnd) { + let Some(stream) = self.streams.live.get(&id).filter(|stream| !stream.connecting) else { return }; + let held = match end { + PipeEnd::ToClient => stream.to_client.is_some(), + PipeEnd::FromClient => stream.from_client.is_some(), + }; + if held { + self.stack.tcp_abort(now, stream.conn); + self.streams.live.remove(&id); + } + } + + /// What to ask the kernel about each established stream's pipes, as the last pass left them. + pub fn watches(&self) -> impl Iterator + '_ { + self.streams.live.iter().filter(|(_, stream)| !stream.connecting).map(|(id, stream)| { + let (from, to) = (stream.from_client.is_some(), stream.to_client.is_some()); + (*id, Watch { readable: from && stream.room, writer: from && !stream.done_writing, writable: to && stream.held, reader: to }) + }) + } + + /// The streams the node holds, connecting ones included. + pub fn streams(&self) -> usize { + self.streams.live.len() + } + + /// Connect answers and log lines since the last call. + pub fn drain_stream_events(&mut self) -> impl Iterator + '_ { + self.streams.events.drain(..) + } +} diff --git a/userland/netstack/node/tests/streams.rs b/userland/netstack/node/tests/streams.rs new file mode 100644 index 00000000000..e704b73026e --- /dev/null +++ b/userland/netstack/node/tests/streams.rs @@ -0,0 +1,1243 @@ +//! Streams on the node, against peers the test scripts, at 192.0.2.2:80 unless a test puts one +//! elsewhere, and pipe ends it fakes. No scenario ids: the specifications' TCP scenarios are +//! `toyos-net-tcp`'s; these are what the node does between [tcp] and a client's two pipes. +//! +//! What is not ours: every segment the node emits is read by `etherparse` as it leaves +//! ([`Far::hears`]), its IPv4 and TCP checksums and lengths that crate's sums, and every segment +//! the peer sends is built by it. The order and numbers each test asserts are RFC 9293's, cited +//! where asserted. The scripted peer itself is ours: it acknowledges what arrives in order and +//! says only what a test tells it to. + +mod common; + +use std::cell::RefCell; +use std::collections::VecDeque; +use std::net::Ipv4Addr; +use std::rc::Rc; +use std::time::Duration; + +use common::{arp, terms, Wire, A, MAC, MAC_B, MAC_R, R}; +use etherparse::{ArpOperation, LinkSlice, NetSlice, PacketBuilder, SlicedPacket, TcpOptionElement, TransportSlice}; +use toyos_net_node::{FromClient, Node, PipeEnd, Pipes, ReadRefusal, StreamEvent, StreamId, ToClient, Watch, WriteRefusal}; +use toyos_net_shard::ConnectError; +use toyos_net_tcp::{Counter, Endpoint, Failure}; +use toyos_net_wire::{Instant, Port}; + +const B: Ipv4Addr = Ipv4Addr::new(192, 0, 2, 2); +/// A second peer's address, and its MAC. +const C: Ipv4Addr = Ipv4Addr::new(192, 0, 2, 3); +const MAC_C: [u8; 6] = [2, 0, 0, 0, 0, 0x0c]; +const PORT: u16 = 80; +/// The peer's initial sequence number. +const ISS: u32 = 5000; +/// [tcp]'s send buffer, as `common::Wire` configures it. +const SEND_BUFFER: usize = 65_535; + +fn peer() -> Endpoint { + Endpoint { addr: B, port: Port::new(PORT).unwrap() } +} + +/// Text that repeats at no power of two. +fn text(len: usize) -> Vec { + (0..len).map(|i| (i % 251) as u8 ^ (i / 251) as u8).collect() +} + +// ---- the client's pipes ---- + +#[derive(Default)] +struct Ends { + /// What the client wrote and the node has not read. + outbox: VecDeque, + /// The client holds no write end: the outbox, once empty, reads as the end. + writer_gone: bool, + /// What the node wrote and the client has not read: at most `room` bytes. + inbox: Vec, + room: usize, + /// The client holds no read end. + reader_gone: bool, + /// The handle the client gave for an end is no pipe end. + write_broken: bool, + read_broken: bool, + to_dropped: bool, + from_dropped: bool, +} + +type Client = Rc>; + +struct WriteEnd(Client); +struct ReadEnd(Client); + +impl ToClient for WriteEnd { + fn write(&mut self, bytes: &[u8]) -> Result { + let mut ends = self.0.borrow_mut(); + if ends.write_broken { + return Err(WriteRefusal::Broken); + } + if ends.reader_gone { + return Err(WriteRefusal::Gone); + } + let taken = bytes.len().min(ends.room - ends.inbox.len()); + if taken == 0 { + return Err(WriteRefusal::Full); + } + ends.inbox.extend_from_slice(&bytes[..taken]); + Ok(taken) + } +} + +impl Drop for WriteEnd { + fn drop(&mut self) { + self.0.borrow_mut().to_dropped = true; + } +} + +impl FromClient for ReadEnd { + fn read(&mut self, out: &mut [u8]) -> Result { + let mut ends = self.0.borrow_mut(); + if ends.read_broken { + return Err(ReadRefusal::Broken); + } + let read = out.len().min(ends.outbox.len()); + if read == 0 { + return if ends.writer_gone { Ok(0) } else { Err(ReadRefusal::Empty) }; + } + for slot in &mut out[..read] { + *slot = ends.outbox.pop_front().unwrap(); + } + Ok(read) + } +} + +impl Drop for ReadEnd { + fn drop(&mut self) { + self.0.borrow_mut().from_dropped = true; + } +} + +/// A client whose receive pipe holds 65,536 bytes, and the two ends it hands over. +fn client() -> (Client, Pipes) { + let client = Rc::new(RefCell::new(Ends { room: 65_536, ..Ends::default() })); + let pipes = Pipes { to_client: Box::new(WriteEnd(client.clone())), from_client: Box::new(ReadEnd(client.clone())) }; + (client, pipes) +} + +fn dropped(client: &Client) -> (bool, bool) { + let ends = client.borrow(); + (ends.to_dropped, ends.from_dropped) +} + +// ---- the far end ---- + +/// A segment the node emitted, as `etherparse` read it. +#[derive(Clone, Debug)] +struct Segment { + seq: u32, + ack: Option, + syn: bool, + fin: bool, + rst: bool, + window: u16, + text: Vec, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Manner { + /// Completes the handshake and acknowledges what arrives in order. + Answers, + /// Says nothing to any segment. + Deaf, + /// Answers, and takes text only as far as the window it offers, which it then shuts: a + /// reader that reads when the test says so ([`Net::run_slowly_until`]). + Slow, + /// Resets a SYN (RFC 9293 §3.10.7.1). + Refuses, +} + +struct Far { + addr: Ipv4Addr, + mac: [u8; 6], + manner: Manner, + /// Nobody answers an ARP request for the peer's address. + absent: bool, + /// The test turned to another peer ([`Net::turn_to`]): of the node's segments this one hears + /// its own connection's and no other, a SYN least of all. + parked: bool, + /// The window the peer's segments offer. + offers: u16, + /// The node's port, from its last SYN to this address: segments from another are heard and + /// not answered. + port: u16, + /// The peer's next sequence number. + snd_nxt: u32, + /// The node's next sequence number the peer has everything before. + rcv_nxt: u32, + /// The node's last acknowledgment, and the window it came with. + acked: u32, + window: u16, + received: Vec, + /// The node's FIN arrived, after everything before it. + fin: bool, + resets: Vec, + /// Every segment the node emitted, in order. + segments: Vec, +} + +fn v4(bytes: &[u8]) -> Ipv4Addr { + Ipv4Addr::from(<[u8; 4]>::try_from(bytes).expect("four bytes")) +} + +impl Far { + fn new() -> Self { + Self::at(B, MAC_B) + } + + /// A peer at `addr`, whose frames come from `mac`. + fn at(addr: Ipv4Addr, mac: [u8; 6]) -> Self { + Self { + addr, + mac, + manner: Manner::Answers, + absent: false, + parked: false, + offers: 65_535, + port: 0, + snd_nxt: ISS, + rcv_nxt: 0, + acked: 0, + window: 0, + received: Vec::new(), + fin: false, + resets: Vec::new(), + segments: Vec::new(), + } + } + + /// A segment from the peer in its frame, built by `etherparse`: PSH with text. + fn frame(&self, seq: u32, ack: Option, syn: bool, fin: bool, rst: bool, text: &[u8]) -> Vec { + let mut step = PacketBuilder::ethernet2(self.mac, MAC).ipv4(self.addr.octets(), A.octets(), 64).tcp(PORT, self.port, seq, self.offers); + if syn { + step = step.syn().options(&[TcpOptionElement::MaximumSegmentSize(1460)]).unwrap(); + } + if let Some(ack) = ack { + step = step.ack(ack); + } + if fin { + step = step.fin(); + } + if rst { + step = step.rst(); + } + if !text.is_empty() { + step = step.psh(); + } + let mut frame = Vec::new(); + step.write(&mut frame, text).unwrap(); + frame + } + + /// The peer's next text. + fn text(&mut self, text: &[u8]) -> Vec { + let frame = self.frame(self.snd_nxt, Some(self.rcv_nxt), false, false, false, text); + self.snd_nxt = self.snd_nxt.wrapping_add(u32::try_from(text.len()).unwrap()); + frame + } + + /// The peer's FIN. + fn fin(&mut self) -> Vec { + let frame = self.frame(self.snd_nxt, Some(self.rcv_nxt), false, true, false, &[]); + self.snd_nxt = self.snd_nxt.wrapping_add(1); + frame + } + + /// A reset at exactly the sequence number the node expects (RFC 9293 §3.10.7.4, first check). + fn rst(&self) -> Vec { + self.frame(self.snd_nxt, Some(self.rcv_nxt), false, false, true, &[]) + } + + /// An acknowledgment that says nothing new but the window the peer offers. + fn update(&self) -> Vec { + self.frame(self.snd_nxt, Some(self.rcv_nxt), false, false, false, &[]) + } + + /// The peer offers a window of `bytes` from its next segment on. + fn offer(&mut self, bytes: u16) { + self.offers = bytes; + } + + /// How many bytes the node's window still lets the peer send. + fn open(&self) -> usize { + self.acked.wrapping_add(u32::from(self.window)).wrapping_sub(self.snd_nxt) as usize + } + + /// The outside reading of a frame the node emitted, and the frames it is answered with. + /// Ethernet II from the node's MAC; then ARP, whose requests for this peer and the router + /// are answered; a datagram of the DHCP client's, which `lease.rs` reads; or a TCP segment + /// to port 80, with the lengths and checksums `etherparse` computes, heard if it is to this + /// peer's address. + fn hears(&mut self, frame: &[u8]) -> Vec> { + let packet = SlicedPacket::from_ethernet(frame).expect("Ethernet II"); + let Some(LinkSlice::Ethernet2(ethernet)) = &packet.link else { panic!("{:?}", packet.link) }; + assert_eq!(ethernet.source(), MAC, "from the node's MAC"); + match (&packet.net, &packet.transport) { + (Some(NetSlice::Arp(asked)), None) => { + let asks = asked.operation() == ArpOperation::REQUEST && v4(asked.sender_protocol_addr()) == A; + match v4(asked.target_protocol_addr()) { + target if asks && target == self.addr && !self.absent => vec![arp(MAC, false, self.mac, self.addr, A)], + target if asks && target == R => vec![arp(MAC, false, MAC_R, R, A)], + _ => Vec::new(), + } + } + (Some(NetSlice::Ipv4(_)), Some(TransportSlice::Udp(_))) => Vec::new(), + (Some(NetSlice::Ipv4(ip)), Some(TransportSlice::Tcp(tcp))) => { + let header = ip.header(); + assert_eq!(header.header_checksum(), header.to_header().calc_header_checksum(), "the IPv4 header checksum"); + assert_eq!(usize::from(header.total_len()), 20 + tcp.slice().len(), "IPv4's length is the segment's"); + assert_eq!(header.source_addr(), A); + assert_eq!(tcp.checksum(), tcp.calc_checksum_ipv4(header.source(), header.destination()).unwrap(), "the TCP checksum"); + assert_eq!(tcp.destination_port(), PORT); + if header.destination_addr() != self.addr || (self.parked && tcp.source_port() != self.port) { + return Vec::new(); + } + assert_eq!(ethernet.destination(), self.mac, "to the peer's MAC"); + let segment = Segment { + seq: tcp.sequence_number(), + ack: tcp.ack().then(|| tcp.acknowledgment_number()), + syn: tcp.syn(), + fin: tcp.fin(), + rst: tcp.rst(), + window: tcp.window_size(), + text: tcp.payload().to_vec(), + }; + self.segments.push(segment.clone()); + if segment.syn { + self.port = tcp.source_port(); + self.received.clear(); + self.fin = false; + } else if tcp.source_port() != self.port { + return Vec::new(); + } + self.answer(&segment) + } + other => panic!("neither ARP, UDP nor TCP: {other:?}"), + } + } + + fn answer(&mut self, segment: &Segment) -> Vec> { + if segment.rst { + self.resets.push(segment.clone()); + return Vec::new(); + } + if let Some(ack) = segment.ack { + (self.acked, self.window) = (ack, segment.window); + } + match self.manner { + Manner::Deaf => return Vec::new(), + Manner::Refuses => return vec![self.frame(0, Some(segment.seq.wrapping_add(1)), false, false, true, &[])], + Manner::Answers | Manner::Slow => {} + } + if segment.syn { + self.rcv_nxt = segment.seq.wrapping_add(1); + self.snd_nxt = ISS.wrapping_add(1); + return vec![self.frame(ISS, Some(self.rcv_nxt), true, false, false, &[])]; + } + let fits = self.manner != Manner::Slow || segment.text.len() <= usize::from(self.offers); + if segment.seq == self.rcv_nxt && fits { + if self.manner == Manner::Slow && !segment.text.is_empty() { + self.offers = 0; + } + self.received.extend_from_slice(&segment.text); + self.rcv_nxt = self.rcv_nxt.wrapping_add(u32::try_from(segment.text.len()).unwrap()); + if segment.fin { + self.fin = true; + self.rcv_nxt = self.rcv_nxt.wrapping_add(1); + } + } + if segment.text.is_empty() && !segment.fin { + return Vec::new(); + } + vec![self.frame(self.snd_nxt, Some(self.rcv_nxt), false, false, false, &[])] + } + + /// The lengths of the node's segments that carried text, in order. + fn texts(&self) -> Vec { + self.segments.iter().filter(|segment| !segment.text.is_empty()).map(|segment| segment.text.len()).collect() + } +} + +// ---- the node between them ---- + +struct Net { + node: Node, + now: Instant, + draws: u32, + far: Far, + /// The peers the test turned from, each with its connection. + parked: Vec, +} + +fn draw(draws: &mut u32) -> impl FnMut() -> u32 + '_ { + move || { + *draws = draws.wrapping_add(1); + *draws + } +} + +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() } + } + + /// Leaves the present peer with the connection it has and makes `next` the peer the test + /// talks to and connects to. + fn turn_to(&mut self, next: Far) { + let mut far = std::mem::replace(&mut self.far, next); + far.parked = true; + self.parked.push(far); + } + + /// What the peers answer a frame the node emitted with. Two at one address give an ARP + /// request the same answer, which is one frame. A segment to an address no peer is at is + /// the node's mistake, and no peer would hear it. + fn hears(&mut self, frame: &[u8]) -> Vec> { + let packet = SlicedPacket::from_ethernet(frame).expect("Ethernet II"); + if let (Some(NetSlice::Ipv4(ip)), Some(TransportSlice::Tcp(_))) = (&packet.net, &packet.transport) { + let to = ip.header().destination_addr(); + assert!(self.far.addr == to || self.parked.iter().any(|far| far.addr == to), "a segment to {to}, where no peer is"); + } + let mut answers = self.far.hears(frame); + for far in &mut self.parked { + for answer in far.hears(frame) { + if !answers.contains(&answer) { + answers.push(answer); + } + } + } + answers + } + + /// Offers the node all the credit it wants and delivers what each frame is answered with, + /// until it sends nothing more. + fn pump(&mut self) { + for _ in 0..100_000 { + if self.opportunity() == 0 { + return; + } + } + panic!("100,000 opportunities and the node still sends"); + } + + /// One transmit opportunity with all the credit the node wants, each frame answered once + /// the opportunity is over. Returns how many frames left. + fn opportunity(&mut self) -> usize { + let mut frames = Vec::new(); + self.node.transmit(self.now, usize::MAX, |frame| frames.push(frame.to_vec())); + for frame in &frames { + for answer in self.hears(frame) { + self.node.receive(self.now, &answer, draw(&mut self.draws)); + } + } + frames.len() + } + + fn deliver(&mut self, frame: &[u8]) { + self.node.receive(self.now, frame, draw(&mut self.draws)); + self.pump(); + } + + /// A pass, as netstack runs one when a pipe it watches is ready, and what it sends. + fn bridge(&mut self) { + self.node.bridge(self.now); + self.pump(); + } + + /// Moves the clock to `at` and fires what is due. + fn fire(&mut self, at: Instant) { + self.now = self.now.max(at); + self.node.fire(self.now, draw(&mut self.draws)); + self.pump(); + } + + /// Fires deadline after deadline until `done`, giving up at the first deadline more than + /// `limit` away. + fn run_until(&mut self, limit: Duration, done: impl Fn(&Net) -> bool) -> bool { + let end = self.now.after(limit); + for _ in 0..100_000 { + if done(self) { + return true; + } + match self.node.next_deadline() { + Some(at) if at <= end => self.fire(at), + _ => return false, + } + } + panic!("100,000 deadlines without the clock passing {limit:?}"); + } + + /// [`Self::run_until`] with every [`Manner::Slow`] peer reading 1,460 bytes a second: each + /// second one whose connection has had neither the node's FIN nor its reset offers that + /// window again. + fn run_slowly_until(&mut self, limit: Duration, done: impl Fn(&Net) -> bool) -> bool { + let end = self.now.after(limit); + let mut reads = self.now.after(Duration::from_secs(1)); + for _ in 0..100_000 { + if done(self) { + return true; + } + let at = self.node.next_deadline().map_or(reads, |at| at.min(reads)); + if at > end { + return false; + } + self.fire(at); + if self.now >= reads { + reads = self.now.after(Duration::from_secs(1)); + let mut updates = Vec::new(); + for far in self.parked.iter_mut().chain([&mut self.far]) { + if far.manner == Manner::Slow && !far.fin && far.resets.is_empty() { + far.offer(1_460); + updates.push(far.update()); + } + } + for update in &updates { + self.deliver(update); + } + } + } + panic!("100,000 deadlines without the clock passing {limit:?}"); + } + + fn connect(&mut self, timeout: Option) -> (StreamId, Client) { + 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.pump(); + (id, client) + } + + fn events(&mut self) -> Vec { + self.node.drain_stream_events().collect() + } + + fn watch(&self, id: StreamId) -> Option { + self.node.watches().find(|(watched, _)| *watched == id).map(|(_, watch)| watch) + } + + /// The sequence number of the node's first byte of text: its SYN's, plus one. + fn first(&self) -> u32 { + let syn = self.far.segments.iter().find(|segment| segment.syn).expect("a SYN"); + syn.seq.wrapping_add(1) + } +} + +/// A stream whose connect was answered `Connected`. +fn established() -> (Net, StreamId, Client) { + established_within(None) +} + +/// A stream whose connect, with `timeout`, was answered `Connected`. +fn established_within(timeout: Option) -> (Net, StreamId, Client) { + let mut net = Net::new(); + let (id, client) = net.connect(timeout); + let local = Port::new(net.far.port).expect("the SYN's source port"); + assert_eq!(net.events(), [StreamEvent::Connected { id, local }]); + (net, id, client) +} + +const IDLE: Watch = Watch { readable: true, writer: true, writable: false, reader: true }; + +// ---- connect ---- + +// RFC 9293 §3.5, figure 6: a SYN that acknowledges nothing, and after the peer's SYN-ACK an ACK +// of its sequence number plus one from ours plus one. +#[test] +fn a_connect_is_a_handshake_and_its_answer_names_the_port() { + let (net, id, client) = established(); + let [syn, ack] = &net.far.segments[..] else { panic!("a SYN and an ACK, not {:?}", net.far.segments) }; + assert!(syn.syn && syn.ack.is_none() && syn.text.is_empty() && !syn.fin && !syn.rst, "{syn:?}"); + assert_eq!((ack.syn, ack.seq, ack.ack, ack.text.len()), (false, syn.seq.wrapping_add(1), Some(ISS + 1), 0)); + assert_eq!((net.node.streams(), net.watch(id), net.node.nodelay(id)), (1, Some(IDLE), Some(false))); + assert_eq!(dropped(&client), (false, false)); +} + +#[test] +fn a_connect_before_the_lease_is_refused_and_sends_nothing() { + let mut wire = Wire::new(); + wire.link(true); + let (client, pipes) = client(); + let refused = wire.node.connect(wire.now, peer(), None, pipes); + assert!(matches!(refused, Err(ConnectError::Route(_))), "{refused:?}"); + assert_eq!((wire.node.streams(), dropped(&client)), (0, (true, true))); + // `common::outside` takes no TCP segment. + wire.pump(); +} + +// RFC 9293 §3.10.7.3: a reset that acknowledges the SYN, in SYN-SENT, is "connection refused". +#[test] +fn a_refused_connect_is_answered_with_the_refusal() { + let mut net = Net::new(); + net.far.manner = Manner::Refuses; + let (id, client) = net.connect(None); + assert_eq!(net.events(), [StreamEvent::Failed { id, failure: Failure::Refused }]); + assert_eq!((net.node.streams(), dropped(&client)), (0, (true, true))); +} + +// The deadline is the node's own and falls between two of the SYN's retransmissions; RFC 9293 +// §3.10.5: an abort in SYN-SENT sends nothing. +#[test] +fn a_connect_past_its_deadline_is_timed_out_at_the_deadline() { + let mut net = Net::new(); + net.far.manner = Manner::Deaf; + let start = net.now; + let (id, client) = net.connect(Some(Duration::from_millis(2_500))); + assert_eq!(net.node.streams(), 1); + assert!(net.run_until(Duration::from_secs(10), |net| net.node.streams() == 0), "the connect ends"); + assert_eq!(net.now, start.after(Duration::from_millis(2_500))); + assert_eq!(net.events(), [StreamEvent::TimedOut { id }]); + assert_eq!(dropped(&client), (true, true)); + assert!(!net.far.segments.is_empty() && net.far.segments.iter().all(|segment| segment.syn), "{:?}", net.far.segments); + // The connection is nobody's from here: its SYN is not sent again. + let sent = net.far.segments.len(); + assert!(!net.run_until(Duration::from_secs(60), |_| false)); + assert_eq!((net.far.segments.len(), net.far.resets.len()), (sent, 0)); +} + +// The peer's address answers no ARP request, so the connect's next hop fails inside a transmit +// opportunity: the answer is there when that opportunity is over, with no frame and no deadline +// of the stream's own to bring it. +#[test] +fn a_connect_nobody_answers_arp_for_fails_when_tcp_knows() { + let mut net = Net::new(); + net.far.absent = true; + let renewal = net.now.after(Duration::from_secs(1_700)); + let (id, client) = net.connect(None); + for _ in 0..10_000 { + let failed = net.node.shard().tcp_counters().get(Counter::NextHopFailed) > 0; + assert_eq!(net.node.streams() == 0, failed, "the stream goes in the opportunity that failed its connect"); + if failed { + break; + } + let at = net.node.next_deadline().expect("address resolution is running"); + assert!(at < renewal, "before the lease's renewal"); + net.fire(at); + } + let events = net.events(); + assert!(matches!(events[..], [StreamEvent::Failed { id: failed, failure: Failure::Unreachable(_) }] if failed == id), "{events:?}"); + assert_eq!((dropped(&client), net.far.segments.len()), ((true, true), 0)); +} + +// A connect's deadline ends with its answer: the stream has none while its client holds it. +#[test] +fn an_established_streams_connect_deadline_is_gone() { + let (mut net, _, _client) = established_within(Some(Duration::from_secs(30))); + net.fire(net.now.after(Duration::from_secs(31))); + assert!(net.node.next_deadline().is_none_or(|at| at > net.now), "{:?} at {:?}", net.node.next_deadline(), net.now); + assert_eq!(net.node.streams(), 1); +} + +#[test] +fn a_connect_closed_before_its_answer_is_answered_closed() { + let mut net = Net::new(); + net.far.manner = Manner::Deaf; + let (id, client) = net.connect(None); + assert!(!net.node.shutdown_write(net.now, id), "not established"); + assert_eq!(net.watch(id), None); + net.node.close(net.now, id); + assert_eq!(net.events(), [StreamEvent::Closed { id }]); + assert_eq!((net.node.streams(), dropped(&client)), (0, (true, true))); + // The connection is nobody's from here: its SYN is not sent again. + let sent = net.far.segments.len(); + assert!(!net.run_until(Duration::from_secs(60), |_| false)); + assert_eq!((net.far.segments.len(), net.far.resets.len()), (sent, 0)); +} + +// A frame 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. +#[test] +fn a_frame_moves_an_established_stream_with_no_opportunity_after_it() { + let (mut net, _, client) = established(); + let frame = net.far.text(b"at once"); + net.node.receive(net.now, &frame, draw(&mut net.draws)); + assert_eq!(client.borrow().inbox, b"at once"); +} + +#[test] +fn a_deadline_moves_an_established_stream_with_no_opportunity_after_it() { + let mut net = Net::new(); + let id = departed(&mut net, 100_000, false); + net.events(); + let cut = net.now.after(Duration::from_secs(100)); + net.node.fire(cut, draw(&mut net.draws)); + assert_eq!((net.node.streams(), net.events()), (0, vec![StreamEvent::Cut { id }])); +} + +// ---- the bridge ---- + +#[test] +fn nothing_leaves_the_pipe_that_the_stack_will_not_take() { + let (mut net, id, client) = established(); + 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); + 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 })); + + net.pump(); + assert!(net.run_until(Duration::from_secs(5), |net| net.far.received.len() >= sent.len()), "{} bytes arrived", net.far.received.len()); + assert!(net.far.received == sent, "the peer's stream is the client's, byte for byte"); + assert!(client.borrow().outbox.is_empty()); + assert_eq!(net.watch(id), Some(IDLE)); +} + +#[test] +fn a_full_pipe_keeps_the_rest_in_the_stack_and_is_watched_for_room() { + let (mut net, id, client) = established(); + client.borrow_mut().room = 100; + let sent = text(300); + let frame = net.far.text(&sent); + net.deliver(&frame); + for taken in [0, 100, 200] { + assert!(client.borrow().inbox == sent[taken..taken + 100], "after {taken} bytes"); + // Watched for room only while the stack holds bytes it refused. + assert_eq!(net.watch(id), Some(Watch { writable: taken < 200, ..IDLE }), "after {taken} bytes"); + client.borrow_mut().inbox.clear(); + net.bridge(); + } + assert!(client.borrow().inbox.is_empty()); + assert_eq!(net.watch(id), Some(IDLE)); +} + +#[test] +fn nothing_leaves_the_stack_that_the_pipe_did_not_take() { + let (mut net, _, client) = established(); + client.borrow_mut().room = 1_000; + let sent = text(200_000); + let (mut read, mut offset) = (Vec::new(), 0); + for _ in 0..10_000 { + if read.len() == sent.len() { + break; + } + // The peer sends what the node's window allows (RFC 9293 §3.8.6), and no more. + let len = net.far.open().min(1_460).min(sent.len() - offset); + if len > 0 { + let frame = net.far.text(&sent[offset..offset + len]); + offset += len; + net.deliver(&frame); + continue; + } + // The window is shut or the text all sent: the client reads what its pipe holds. + let taken = std::mem::take(&mut client.borrow_mut().inbox); + if taken.is_empty() { + let at = net.node.next_deadline().expect("the lease's renewal, if nothing else"); + net.fire(at); + } else { + read.extend(taken); + net.bridge(); + } + } + assert_eq!(offset, sent.len(), "the window let all of it through"); + assert!(read == sent, "the client's stream is the peer's, byte for byte: {} of {} bytes", read.len(), sent.len()); +} + +// RFC 9293 §3.8.6.2.2: the window a full pipe shut is opened again by an acknowledgment once the +// pipe has taken what the stack held, at the first opportunity after the pass that moved it. +#[test] +fn room_in_the_pipe_opens_the_window_at_the_next_opportunity() { + let (mut net, _, client) = established(); + client.borrow_mut().room = 0; + let sent = text(65_535); + let mut offset = 0; + for _ in 0..1_000 { + let len = net.far.open().min(1_460).min(sent.len() - offset); + if offset == sent.len() { + break; + } else if len > 0 { + let frame = net.far.text(&sent[offset..offset + len]); + offset += len; + net.deliver(&frame); + } else { + let at = net.node.next_deadline().expect("a delayed acknowledgment"); + net.fire(at); + } + } + assert_eq!(offset, sent.len(), "a window of 65,535 bytes let all of it through"); + let shut = |net: &Net| (net.far.acked, net.far.window) == (net.far.snd_nxt, 0); + 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); + assert!(client.borrow().inbox == sent); + assert_eq!(net.opportunity(), 1, "the window update"); + let update = net.far.segments.last().expect("a segment"); + assert_eq!((update.ack, update.window, update.text.len()), (Some(net.far.snd_nxt), 65_535, 0)); +} + +// ---- the stream's ends ---- + +// RFC 9293 §3.6, case 1: our FIN follows our last byte, and in FIN-WAIT-2 the peer's text is +// still received; its FIN is acknowledged and the connection is the stack's to wait out. +#[test] +fn the_end_of_the_clients_writing_is_a_fin_and_the_peer_still_sends() { + let (mut net, id, client) = established(); + client.borrow_mut().outbox.extend(b"request"); + client.borrow_mut().writer_gone = true; + net.bridge(); + assert_eq!((net.far.received.as_slice(), net.far.fin), (&b"request"[..], true)); + let fin = net.far.segments.iter().find(|segment| segment.fin).expect("a FIN"); + assert_eq!(fin.seq.wrapping_add(u32::try_from(fin.text.len()).unwrap()), net.first().wrapping_add(7)); + assert_eq!((dropped(&client), net.node.streams()), ((false, true), 1)); + assert_eq!(net.watch(id), Some(Watch { readable: false, writer: false, writable: false, reader: true })); + + let frame = net.far.text(b"response"); + net.deliver(&frame); + assert_eq!(client.borrow().inbox, b"response"); + let frame = net.far.fin(); + net.deliver(&frame); + assert_eq!((dropped(&client), net.node.streams()), ((true, true), 0)); + assert_eq!(net.far.acked, net.far.snd_nxt, "the peer's FIN is acknowledged"); + assert!(net.far.resets.is_empty() && net.events().is_empty()); +} + +// RFC 9293 §3.6, case 2: the peer's FIN ends the client's reading after its last byte, and the +// client still writes until its own end. +#[test] +fn the_peers_fin_ends_the_clients_reading_and_not_its_writing() { + let (mut net, _, client) = established(); + let frame = net.far.text(b"all there is"); + net.deliver(&frame); + let frame = net.far.fin(); + net.deliver(&frame); + assert_eq!(client.borrow().inbox, b"all there is"); + assert_eq!((dropped(&client), net.node.streams()), ((true, false), 1)); + + // CLOSE-WAIT lasts as long as the client likes. + net.fire(net.now.after(Duration::from_secs(150))); + assert_eq!((net.node.streams(), net.events()), (1, vec![])); + client.borrow_mut().outbox.extend(b"noted"); + client.borrow_mut().writer_gone = true; + net.bridge(); + assert_eq!((net.far.received.as_slice(), net.far.fin), (&b"noted"[..], true)); + assert_eq!((dropped(&client), net.node.streams()), ((true, true), 0)); + assert!(net.far.resets.is_empty()); +} + +// RFC 9293 §3.6: both FINs end the connection, not the client's reading: text its pipe had no +// room for when the peer's FIN arrived is still its to read, and the end comes after it. +#[test] +fn text_the_pipe_had_no_room_for_outlives_both_fins() { + let (mut net, _, client) = established(); + client.borrow_mut().room = 0; + client.borrow_mut().outbox.extend(b"request"); + client.borrow_mut().writer_gone = true; + net.bridge(); + assert!(net.far.fin); + let frame = net.far.text(b"held back"); + net.deliver(&frame); + let frame = net.far.fin(); + net.deliver(&frame); + assert_eq!(net.far.acked, net.far.snd_nxt, "the peer's FIN is acknowledged"); + assert_eq!((client.borrow().inbox.len(), dropped(&client), net.node.streams()), (0, (false, true), 1)); + + client.borrow_mut().room = 65_536; + net.bridge(); + assert_eq!(client.borrow().inbox, b"held back"); + assert_eq!((dropped(&client), net.node.streams()), ((true, true), 0)); +} + +// A client that shut its writing down and still reads is alive: what it cannot send is [tcp]'s +// to give up on (R2, RFC 9293 §3.8.3), not the node's to cut. +#[test] +fn a_live_client_that_is_done_writing_is_not_timed() { + let (mut net, id, client) = established(); + net.far.manner = Manner::Deaf; + client.borrow_mut().outbox.extend(text(100_000)); + assert!(net.node.shutdown_write(net.now, id)); + net.pump(); + assert!(!net.run_until(Duration::from_secs(150), |_| false)); + assert_eq!((dropped(&client), net.node.streams(), net.far.resets.len()), ((false, false), 1, 0)); + assert!(net.events().is_empty()); +} + +#[test] +fn a_shutdown_sends_what_the_pipe_held_and_then_the_fin() { + let (mut net, id, client) = established(); + client.borrow_mut().outbox.extend(b"written before the shutdown"); + assert!(net.node.shutdown_write(net.now, id)); + net.pump(); + assert_eq!((net.far.received.as_slice(), net.far.fin), (&b"written before the shutdown"[..], true)); + assert_eq!((dropped(&client), net.node.streams()), ((false, true), 1)); +} + +#[test] +fn a_close_sends_what_the_pipe_held_and_then_the_fin() { + let (mut net, id, client) = established(); + client.borrow_mut().outbox.extend(b"written and dropped"); + net.node.close(net.now, id); + net.pump(); + assert_eq!((net.far.received.as_slice(), net.far.fin), (&b"written and dropped"[..], true)); + assert_eq!((dropped(&client), net.node.streams()), ((true, true), 0)); + assert!(net.far.resets.is_empty()); + // The id names nothing. + net.node.close(net.now, id); + assert_eq!((net.node.nodelay(id), net.node.set_nodelay(net.now, id, true), net.node.shutdown_write(net.now, id)), (None, false, false)); +} + +// RFC 9293 §3.6.1 (SHLD-3): a close with text unread is a reset, which shows the peer it was +// lost. The rule is [tcp]'s; the node's is to hand it the close. +#[test] +fn a_close_with_text_unread_resets() { + let (mut net, id, client) = established(); + client.borrow_mut().room = 0; + let frame = net.far.text(b"never read"); + net.deliver(&frame); + net.node.close(net.now, id); + net.pump(); + assert_eq!((net.far.resets.len(), net.node.streams()), (1, 0)); +} + +// RFC 9293 §3.10.7.4: a reset in sequence ends the connection, and the client learns it as the +// end of both pipes. +#[test] +fn a_reset_ends_both_pipes_and_the_stream() { + let (mut net, _, client) = established(); + client.borrow_mut().outbox.extend(b"never sent"); + net.far.manner = Manner::Deaf; + let frame = net.far.rst(); + net.deliver(&frame); + assert_eq!((dropped(&client), net.node.streams()), ((true, true), 0)); + assert!(net.events().is_empty() && net.far.resets.is_empty()); +} + +// ---- a client that leaves ---- + +#[test] +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; + 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); + net.pump(); + assert_eq!((net.far.received.as_slice(), net.far.fin), (&b"last words"[..], true)); + let frame = net.far.fin(); + net.deliver(&frame); + assert_eq!(net.far.acked, net.far.snd_nxt, "the peer's FIN is acknowledged by a connection nobody holds"); + assert!(net.far.resets.is_empty() && net.events().is_empty()); +} + +#[test] +fn a_reader_that_left_is_not_written_to_again() { + let (mut net, id, client) = established(); + client.borrow_mut().reader_gone = true; + let frame = net.far.text(b"to nobody"); + net.deliver(&frame); + assert_eq!((dropped(&client), net.node.streams()), ((true, false), 1), "its writer may still write"); + assert_eq!(net.watch(id), Some(Watch { reader: false, ..IDLE })); + net.fire(net.now.after(Duration::from_secs(150))); + assert_eq!((net.node.streams(), net.events(), net.far.resets.len()), (1, vec![], 0), "and for as long as it likes"); +} + +/// How a client's reader is found gone. +#[derive(Clone, Copy, Debug)] +enum Leaving { + /// The kernel says so of the to-client pipe. + Kernel, + /// A write to it is refused for it. + Write, + /// The client closes the stream. + Close, +} + +// R2 of RFC 9293 §3.8.3, at the 100 seconds it asks for at least, from the moment nobody is left +// when the pipe gives up nothing after; the reset is §3.10.5's. +#[test] +fn a_departed_clients_unsent_bytes_have_100_seconds() { + for leaving in [Leaving::Kernel, Leaving::Write, Leaving::Close] { + let (mut net, id, client) = established_within(Some(Duration::from_secs(30))); + match leaving { + Leaving::Kernel => net.node.pipe_gone(net.now, id, PipeEnd::ToClient), + Leaving::Write => { + client.borrow_mut().reader_gone = true; + let frame = net.far.text(b"to nobody"); + net.deliver(&frame); + } + Leaving::Close => {} + } + net.far.manner = Manner::Deaf; + client.borrow_mut().outbox.extend(text(100_000)); + net.bridge(); + assert_eq!(client.borrow().outbox.len(), 100_000 - SEND_BUFFER, "{leaving:?}"); + + net.fire(net.now.after(Duration::from_secs(5))); + assert_eq!(net.node.streams(), 1, "{leaving:?}: its writer is still there"); + let left = net.now; + match leaving { + Leaving::Close => net.node.close(left, id), + Leaving::Kernel | Leaving::Write => { + client.borrow_mut().writer_gone = true; + net.node.pipe_gone(left, id, PipeEnd::FromClient); + } + } + net.pump(); + assert_eq!((dropped(&client), net.node.streams()), ((true, false), 1), "{leaving:?}"); + assert_eq!(net.watch(id), Some(Watch { readable: false, writer: false, writable: false, reader: false }), "{leaving:?}"); + + assert!(net.run_until(Duration::from_secs(200), |net| net.node.streams() == 0), "{leaving:?}: the stream is let go"); + assert_eq!(net.now, left.after(Duration::from_secs(100)), "{leaving:?}: from the moment nobody was left"); + assert_eq!(net.events(), [StreamEvent::Cut { id }], "{leaving:?}"); + assert_eq!((dropped(&client), net.far.resets.len()), ((true, true), 1), "{leaving:?}"); + assert_eq!(client.borrow().outbox.len(), 100_000 - SEND_BUFFER, "{leaving:?}: and no byte more was taken"); + } +} + +/// A stream whose client wrote `len` bytes and is gone, its writer and its reader, at the +/// clock's present. Its peer hears nothing after the handshake, or is a [`Manner::Slow`] reader +/// offering 1,460 bytes. +fn departed(net: &mut Net, len: usize, slow: bool) -> StreamId { + net.far.manner = Manner::Answers; + net.far.offer(65_535); + let (id, client) = net.connect(None); + if slow { + net.far.manner = Manner::Slow; + net.far.offer(1_460); + let update = net.far.update(); + net.deliver(&update); + } else { + net.far.manner = Manner::Deaf; + } + client.borrow_mut().outbox.extend(text(len)); + client.borrow_mut().writer_gone = true; + net.bridge(); + net.node.pipe_gone(net.now, id, PipeEnd::FromClient); + net.node.pipe_gone(net.now, id, PipeEnd::ToClient); + id +} + +// A peer that takes one segment a second, past the 100 seconds, sees a departed client's whole +// tail and its FIN: the cut's clock runs from the last byte the pipe gave up. +#[test] +fn a_departed_clients_tail_arrives_whole_while_its_peer_takes_it() { + let mut net = Net::new(); + let sent = text(250_000); + let id = departed(&mut net, sent.len(), true); + let left = net.now; + assert_eq!((net.node.streams(), net.events()), (1, vec![StreamEvent::Connected { id, local: Port::new(net.far.port).unwrap() }])); + + assert!(net.run_slowly_until(Duration::from_secs(1_500), |net| net.far.fin), "the FIN arrives: {} bytes did", net.far.received.len()); + assert!(net.now > left.after(Duration::from_secs(100)), "{:?} after {left:?}", net.now); + assert!(net.far.received == sent, "the peer's stream is the client's, byte for byte"); + assert_eq!((net.node.streams(), net.events(), net.far.resets.len()), (0, vec![], 0)); +} + +// Once the peer's FIN is read through and the client writes no more, nothing of the stream +// reaches the client again and nothing tells the node whether it lives: its unsent bytes have +// the 100 seconds of a departed client's. The peer offers no window and answers every probe +// (RFC 9293 §3.8.6.1), which keeps [tcp] from giving the connection up, so the cut is the node's. +#[test] +fn a_client_done_writing_after_the_peers_fin_has_100_seconds_without_progress() { + let (mut net, id, client) = established(); + net.far.manner = Manner::Slow; + net.far.offer(0); + let frame = net.far.fin(); + net.deliver(&frame); + assert_eq!((dropped(&client), net.node.streams()), ((true, false), 1), "the client reads the end"); + + client.borrow_mut().outbox.extend(text(100_000)); + net.bridge(); + assert_eq!(client.borrow().outbox.len(), 100_000 - SEND_BUFFER); + let left = net.now; + client.borrow_mut().writer_gone = true; + net.node.pipe_gone(left, id, PipeEnd::FromClient); + net.pump(); + assert_eq!((net.node.streams(), net.events()), (1, vec![])); + + assert!(net.run_until(Duration::from_secs(200), |net| net.node.streams() == 0), "the stream is let go"); + assert_eq!(net.now, left.after(Duration::from_secs(100)), "from the moment its client wrote no more"); + assert_eq!(net.events(), [StreamEvent::Cut { id }]); + assert_eq!((dropped(&client), net.far.resets.len()), ((true, true), 1)); + assert_eq!((client.borrow().outbox.len(), net.far.received.len()), (100_000 - SEND_BUFFER, 0), "no byte was given up, and none taken"); + // The peer's script acknowledges every segment that carries text, taken or not. + let probes = net.far.texts(); + assert!(probes.len() > 2, "the peer answered probes: {probes:?}"); +} + +// Sixteen connections of departed clients are a peer's to keep alive, all at once; the +// seventeenth has its 100 seconds from the leaving, whatever the peer takes of it. +#[test] +fn a_peer_keeps_sixteen_departed_clients_connections_alive_and_no_more() { + let mut net = Net::new(); + let sent = text(250_000); + for _ in 0..16 { + departed(&mut net, sent.len(), true); + net.turn_to(Far::new()); + } + let id = departed(&mut net, sent.len(), true); + let left = net.now; + assert_eq!(net.node.streams(), 17); + net.events(); + + assert!(net.run_slowly_until(Duration::from_secs(1_500), |net| net.node.streams() < 17), "a stream is let go"); + assert_eq!(net.now, left.after(Duration::from_secs(100)), "the peer took {} bytes of the seventeenth", net.far.received.len()); + assert_eq!((net.node.streams(), net.events(), net.far.resets.len()), (16, vec![StreamEvent::Cut { id }], 1)); + assert!(net.far.received.len() > 1_460, "and it did take some"); + + assert!(net.run_slowly_until(Duration::from_secs(1_500), |net| net.parked.iter().all(|far| far.fin)), "the sixteen FINs arrive"); + for (nth, far) in net.parked.iter().enumerate() { + assert!(far.received == sent, "connection {nth}: the peer's stream is the client's, byte for byte"); + assert!(far.resets.is_empty(), "connection {nth}"); + } + assert_eq!((net.node.streams(), net.events()), (0, vec![])); +} + +// The seventeenth becomes one of the sixteen when one of them is done: from then its peer keeps +// it alive too, past the 100 seconds it had from its client's leaving. +#[test] +fn a_departed_clients_connection_past_the_sixteen_is_one_of_them_once_one_is_done() { + let mut net = Net::new(); + for _ in 0..16 { + departed(&mut net, 100_000, true); + net.turn_to(Far::new()); + } + let sent = text(250_000); + departed(&mut net, sent.len(), true); + let left = net.now; + assert_eq!(net.node.streams(), 17); + net.events(); + + // The sixteen peers read all there is, at once. + let mut updates = Vec::new(); + for far in &mut net.parked { + far.manner = Manner::Answers; + far.offer(65_535); + updates.push(far.update()); + } + for update in &updates { + net.deliver(update); + } + let done = |net: &Net| net.node.streams() == 1 && net.parked.iter().all(|far| far.fin && far.received.len() == 100_000); + assert!(net.run_until(Duration::from_secs(10), done), "the sixteen are done: {} streams are left", net.node.streams()); + assert!(net.now < left.after(Duration::from_secs(100)), "before the seventeenth's 100 seconds: {:?} after {left:?}", net.now); + + assert!(net.run_slowly_until(Duration::from_secs(1_500), |net| net.far.fin), "the FIN arrives: {} bytes did", net.far.received.len()); + assert!(net.now > left.after(Duration::from_secs(100)), "{:?} after {left:?}", net.now); + assert!(net.far.received == sent, "the peer's stream is the client's, byte for byte"); + assert_eq!((net.node.streams(), net.events(), net.far.resets.len()), (0, vec![], 0)); +} + +// The sixteen are an address's: with sixteen held at one address, a peer at another keeps its +// own departed client's connection alive. +#[test] +fn a_peer_at_another_address_keeps_its_own_sixteen() { + let mut net = Net::new(); + for _ in 0..16 { + departed(&mut net, 100_000, false); + } + net.turn_to(Far::at(C, MAC_C)); + let sent = text(250_000); + departed(&mut net, sent.len(), true); + let left = net.now; + assert_eq!(net.node.streams(), 17); + net.events(); + + assert!(net.run_slowly_until(Duration::from_secs(1_500), |net| net.far.fin), "the FIN arrives: {} bytes did", net.far.received.len()); + assert!(net.now > left.after(Duration::from_secs(100)), "{:?} after {left:?}", net.now); + assert!(net.far.received == sent, "the peer's stream is the client's, byte for byte"); + // The sixteen at the first address gave up nothing and were cut. + let cuts = net.events(); + assert!(cuts.len() == 16 && cuts.iter().all(|event| matches!(event, StreamEvent::Cut { .. })), "{cuts:?}"); + assert_eq!((net.node.streams(), net.far.resets.len()), (0, 0)); +} + +// ---- pipes that refuse the node ---- + +// RFC 9293 §3.10.5: the reset an abort sends is at SND.NXT. +#[test] +fn a_pipe_that_is_no_pipe_resets_its_connection() { + for write in [true, false] { + let (mut net, _, client) = established(); + if write { + client.borrow_mut().write_broken = true; + let frame = net.far.text(b"to a handle that is no pipe"); + net.deliver(&frame); + } else { + client.borrow_mut().read_broken = true; + net.bridge(); + } + assert_eq!((dropped(&client), net.node.streams()), ((true, true), 0), "write: {write}"); + let [reset] = &net.far.resets[..] else { panic!("one reset, not {:?}", net.far.resets) }; + assert_eq!(reset.seq, net.first(), "write: {write}"); + } +} + +#[test] +fn a_refused_watch_resets_only_a_stream_that_holds_the_end() { + let (mut net, id, client) = established(); + client.borrow_mut().writer_gone = true; + net.bridge(); + assert_eq!(dropped(&client), (false, true)); + net.node.pipe_broken(net.now, id, PipeEnd::FromClient); + net.pump(); + assert_eq!((net.node.streams(), net.far.resets.len()), (1, 0), "the end was let go before its watch answered"); + net.node.pipe_broken(net.now, id, PipeEnd::ToClient); + net.pump(); + assert_eq!((dropped(&client), net.node.streams(), net.far.resets.len()), ((true, true), 0, 1)); +} + +// ---- options ---- + +// RFC 9293 §3.7.4: with text unacknowledged a short segment waits (TX-02), and with Nagle's +// algorithm off it leaves (TX-03). +#[test] +fn nodelay_reaches_the_stack() { + let (mut net, id, client) = established(); + net.far.manner = Manner::Deaf; + client.borrow_mut().outbox.extend(text(100)); + net.bridge(); + assert_eq!(net.far.texts(), [100]); + client.borrow_mut().outbox.extend(text(50)); + net.bridge(); + assert_eq!(net.far.texts(), [100], "the second write waits for the first's acknowledgment"); + assert!(net.node.set_nodelay(net.now, id, true)); + net.pump(); + assert_eq!((net.far.texts(), net.node.nodelay(id)), (vec![100, 50], Some(true))); +} + +// ---- the wire is not trusted ---- + +/// An established stream, and a frame of the peer's carrying `text`. +fn expecting(text: &[u8]) -> (Net, Client, Vec) { + let (mut net, _, client) = established(); + let frame = net.far.text(text); + (net, client, frame) +} + +#[test] +fn no_cut_of_a_segment_is_a_segment() { + let (_, _, frame) = expecting(b"whole or nothing"); + for len in 0..frame.len() { + let (mut net, client, frame) = expecting(b"whole or nothing"); + net.deliver(&frame[..len]); + assert!(client.borrow().inbox.is_empty(), "cut to {len} bytes"); + net.deliver(&frame); + assert_eq!(client.borrow().inbox, b"whole or nothing", "whole, after the cut to {len} bytes"); + } +} + +#[test] +fn no_flipped_bit_of_a_segment_delivers_other_bytes() { + let (_, _, frame) = expecting(b"whole or nothing"); + for index in 0..frame.len() { + for bit in 0..8 { + let (mut net, client, mut frame) = expecting(b"whole or nothing"); + frame[index] ^= 1 << bit; + net.deliver(&frame); + // A flip no checksum covers is in the link addresses: taken or refused, the client + // reads the peer's text or nothing. + let inbox = client.borrow().inbox.clone(); + assert!(inbox.is_empty() || inbox == b"whole or nothing", "byte {index} bit {bit}: {inbox:?}"); + assert_eq!(net.node.streams(), 1, "byte {index} bit {bit}"); + } + } +}