From f8a8b198a26129e97b31023700c7b42fbe7f2ca3 Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 28 Oct 2026 02:49:55 +0100 Subject: [PATCH 01/12] kernel: a pipe keeps its bytes in 4 KiB frames it takes as it fills A pipe's ring was one zeroed 2 MiB page, taken on its first write, left from when SYS_PIPE_MAP mapped it into a process. Every pipe that carried a byte held 2 MiB: a bell, each direction of a connection, and each of a TCP connect's two connection pipes, zeroed before the connect returned. A pipe's bytes now live in a queue of 4 KiB frames counted to the job that made the pipe (kernel::pipe::Bytes, host-tested). A write takes the frames it needs under one hold of the allocator's lock, all or none; a read rotates the frames it empties to the back as spares and gives back the spares past as many as still hold bytes, and one. So a pipe holds at most twice the frames its bytes need plus one, never more than MOST_FRAMES, and a stream that keeps a backlog takes nothing from the allocator. The frames are kernel-only and written before they are read, so they are not zeroed. The read end's last holder gives back what the pipe held, which nobody can read any more. Capacity stays 2 MiB, now toyos_abi::syscall::PIPE_CAPACITY, so pipe.throughput keeps its backlog. When a reader keeps up and drains the pipe each round, a 64 KiB write takes 16 frames again: on the host the allocator's take and give of 16 frames cost 89 ns a round (release, no lock), against about 9 us to move 64 KiB at the T14's 7.2 GB/s. The page-table frame path is generalised to it: Frames::alloc_held and free_held take and give a batch of frames a job holds one by one, and the ledger's `tables` count is `held`, covering both. #1057's floors and refusals apply unchanged through `admit`. A pipe write can now be refused ResourceExhausted whenever it needs a frame and its job's memory is spent, where before only the first write could be. A floor holds what must keep working. Deleted: toyos_abi::ring (Ring, RingHeader, Src, Dst, the reader and writer flags nothing read), UserBytes::read_run and UserBytesMut::write_run, Pipe::publish_ends. netstack's PIPE_BYTES and fileserver's STREAM_PRESENT read PIPE_CAPACITY. Closes issues/a-pipe-is-a-two-mebibyte-page-for-a-mapping-nothing-makes.md. Co-Authored-By: Claude Opus 5.5 --- build/sourcegate.rs | 8 +- ...bibyte-page-for-a-mapping-nothing-makes.md | 32 -- ...ns-a-user-copy-and-its-first-allocation.md | 42 +- ...as-not-bound-is-not-shared-with-a-child.md | 3 +- issues/no-physical-memory-fairness.md | 3 +- ...nused-oracle-for-kernel-region-lifetime.md | 4 +- kernel/pure/frames.rs | 128 +++-- kernel/pure/pipe.rs | 294 ++++++++++- kernel/src/arch/aarch64/paging.rs | 2 +- kernel/src/arch/x86_64/paging.rs | 2 +- kernel/src/mm/pmm.rs | 32 +- kernel/src/object/job.rs | 2 +- kernel/src/pipe.rs | 185 +++---- kernel/src/user_ptr.rs | 25 - sdk/abi/src/handle.rs | 2 +- sdk/abi/src/lib.rs | 1 - sdk/abi/src/ring.rs | 479 ------------------ sdk/abi/src/syscall.rs | 5 + system/fileserver/src/main.rs | 6 +- system/netstack/src/main.rs | 19 +- .../src/bin/kernel_objects_refused.rs | 4 +- .../guest/rust-tests/src/bin/memory_floor.rs | 14 +- .../rust-tests/src/bin/poller_capacity.rs | 3 +- 23 files changed, 542 insertions(+), 753 deletions(-) delete mode 100644 issues/a-pipe-is-a-two-mebibyte-page-for-a-mapping-nothing-makes.md delete mode 100644 sdk/abi/src/ring.rs diff --git a/build/sourcegate.rs b/build/sourcegate.rs index 3e4b8419252..4bd5dddbe9c 100644 --- a/build/sourcegate.rs +++ b/build/sourcegate.rs @@ -141,7 +141,6 @@ const AUTO_TRAIT_IMPLS: &[(&str, usize)] = &[ ("kernel/src/arch/x86_64/hw.rs", 1), ("kernel/src/mm/mmio.rs", 2), ("kernel/src/mm/region.rs", 2), - ("kernel/src/pipe.rs", 1), ("kernel/src/process.rs", 1), ("kernel/src/sched/driver.rs", 2), ("kernel/src/symbols.rs", 2), @@ -680,14 +679,11 @@ mod tests { } /// A page a process can write while the kernel is inside it never becomes a - /// slice, and never an exclusive reference. Two files, named rather than + /// slice, and never an exclusive reference. The file is named rather than /// walked: both spellings are ordinary elsewhere. - /// - /// A *shared* reference is not banned: `Ring::header` soundly takes one over - /// the same page, its whole subject being an `AtomicU32`. #[test] fn no_slice_or_exclusive_reference_is_built_over_a_mapped_page() { - const OVER_A_MAPPING: &[&str] = &["sdk/abi/src/ring.rs", "kernel/src/user_ptr.rs"]; + const OVER_A_MAPPING: &[&str] = &["kernel/src/user_ptr.rs"]; const SLICES: &[&str] = &["from_raw_parts", "from_raw_parts_mut", "from_ptr_range", "from_ptr_range_mut"]; const EXCLUSIVE: &str = "&mut *"; diff --git a/issues/a-pipe-is-a-two-mebibyte-page-for-a-mapping-nothing-makes.md b/issues/a-pipe-is-a-two-mebibyte-page-for-a-mapping-nothing-makes.md deleted file mode 100644 index aad88663e48..00000000000 --- a/issues/a-pipe-is-a-two-mebibyte-page-for-a-mapping-nothing-makes.md +++ /dev/null @@ -1,32 +0,0 @@ ---- -status: open -kind: defect -opened: 2026-10-11 ---- - -# A pipe is a 2 MiB page for a mapping nothing makes - -A pipe's ring is one `pmm::alloc_page` page, `PIPE_SIZE = PAGE_2M` -(`kernel/src/pipe.rs`), allocated and zeroed on its first write, because -`SYS_PIPE_MAP` once mapped that page into a process and a 2 MiB page was then -the only one the kernel mapped. `SYS_PIPE_MAP` is gone; the page is not. It -costs every pipe and every IPC connection, which is two pipes, 2 MiB a -direction once a byte moves, whatever the traffic: a wake bell holds 2 MiB to -carry one byte, and so does each direction of a connection that carries one -request. `Pipe::publish_ends` still writes reader and writer flags into a -ring header nothing reads (`sdk/abi/src/ring.rs`). - -A stream's holder is a connection and two bells, four pipes, so each holder -costs 8 MiB of pages once it has rung, rung back and been answered -(`HOLDER_BYTES`, `system/netstack/src/main.rs`): a stream exported to N -children holds about N × 8 MiB beside its 8 MiB region, and netstack prices -each holder past the first as a place. Shared listeners and datagram sockets -multiply it again, so this is fixed before they share. - -Owner: the orchestrator, who stages -`issues/netstack-moves-every-byte-through-two-kernel-pipes.md`, whose streams -ring their doorbells through pipes and pay a page for each until this closes. - -Exit: `rg -n -i 'publish_ends|RING_(WRITER|READER)_CLOSED' kernel sdk tests` -is empty, and a pipe's ring is no `pmm::alloc_page` page: -`rg -n 'alloc_page|PAGE_2M' kernel/src/pipe.rs` is empty. diff --git a/issues/a-pipes-lock-spans-a-user-copy-and-its-first-allocation.md b/issues/a-pipes-lock-spans-a-user-copy-and-its-first-allocation.md index 80ddd79ff86..4964972d4a2 100644 --- a/issues/a-pipes-lock-spans-a-user-copy-and-its-first-allocation.md +++ b/issues/a-pipes-lock-spans-a-user-copy-and-its-first-allocation.md @@ -4,40 +4,42 @@ kind: defect opened: 2026-08-28 --- -# A pipe's lock spans a 2 MiB user copy, and a pipe's first write allocates and zeroes 2 MiB under it +# A pipe's lock spans a 2 MiB user copy and the frames it takes for it Each pipe owns its state behind its own `sync::Lock` (`kernel/src/pipe.rs`), which disables preemption before it spins (`kernel/src/sync.rs`). The bulk copy runs inside that lock: -* `PipeRef::try_read_within` runs `ring.read(len, |off, src| buf.write_run(off, &src))` - under it; -* `PipeRef::write` runs `ring.write(n, |off, dst| buf.read_run(off, dst))` in - the same acquisition. +* `PipeRef::try_read_within` runs `bytes.read(len, ..)`, each frame's run + copied with `buf.write_at`, under it; +* `PipeRef::write` runs `bytes.write(n, ..)`, each run copied with + `buf.read_at`, in the same acquisition. -The copy is bounded only by the ring: `PIPE_SIZE` is `PAGE_2M` less the 64-byte -`RingHeader`, so **2,097,088 bytes** is the largest single copy under the lock, -and nothing above caps it (`SYS_READ`/`SYS_WRITE` pass the caller's length -through). Each 16 pages of it take the caller's space lock and populate any -untouched page, a frame from the allocator and a 4 KiB zeroing. +The copy is bounded only by the pipe: `PIPE_CAPACITY` is 2 MiB, the largest +single copy under the lock, and nothing above caps it (`SYS_READ`/`SYS_WRITE` +pass the caller's length through). Each 16 pages of it take the caller's space +lock and populate any untouched page, a frame from the allocator and a 4 KiB +zeroing. -A pipe's **first** write is worse: `State::back` allocates the ring page lazily -under the same lock, nesting the frame allocator's lock and a 2 MiB zeroing. +A write that grows the pipe also takes its frames under the same lock, +nesting the frame allocator's lock: up to 513 frames, unzeroed, under one +hold, and the heap the list of them grows into. A read gives back the frames +it empties past those kept, under the same nesting. What queues behind it is that one pipe's traffic: its other end's reads and writes, `has_data`/`has_space` from its blocking wrappers and polls, and its handles' clones and closes. The scheduler's own bar for an uninterruptible -stretch is `MAX_PASS_NS = 200_000` ns (`kernel/pure/sched/cpu.rs`); a 2 MiB copy, -and on first write a 2 MiB zeroing, is past it at any plausible bandwidth. The -hold time is not measured, and that is owed first. The length is the caller's. +stretch is `MAX_PASS_NS = 200_000` ns (`kernel/pure/sched/cpu.rs`); a 2 MiB copy +is past it at any plausible bandwidth. The hold time is not measured, and that +is owed first. The length is the caller's. ## Fix direction -The lock is also the serialization `Ring` relies on (`sdk/abi/src/ring.rs`'s -doc), so the copy cannot simply be lifted out: a claimed-range token per end, -or a copy bounded per acquisition, has to carry the ring's data-race argument. -The lazy allocation moves out from under the lock: allocate before taking it, -or drop and retry on the miss. +The lock is also the serialization `kernel::pipe::Bytes` relies on, so the +copy cannot simply be lifted out: a claimed-range token per end, or a copy +bounded per acquisition, has to carry the data-race argument. The frames can +be taken before the lock and given back after it, as the read end's last +holder already gives back what the pipe held. **Exit**: no pipe lock is held across more than a bounded copy or any allocation, the hold time is read before and after, and `audio_tone`'s worst diff --git a/issues/a-socket-libc-or-socket2-has-not-bound-is-not-shared-with-a-child.md b/issues/a-socket-libc-or-socket2-has-not-bound-is-not-shared-with-a-child.md index 01020cd52ee..d60d7eb4e36 100644 --- a/issues/a-socket-libc-or-socket2-has-not-bound-is-not-shared-with-a-child.md +++ b/issues/a-socket-libc-or-socket2-has-not-bound-is-not-shared-with-a-child.md @@ -30,8 +30,7 @@ What is not shared: Each child descriptor naming a stream is a holder of its own: inetd's `dup2(s, 0); dup2(s, 1); dup2(s, 2)` makes three holders for one child, each -a place, a watch and four pipe pages -(`issues/a-pipe-is-a-two-mebibyte-page-for-a-mapping-nothing-makes.md`). One +a place, a watch and four pipes. One holder shared by a child's duplicates would need the child to alias an adoption, since a second copy of one exported end carries no `ADOPT`. diff --git a/issues/no-physical-memory-fairness.md b/issues/no-physical-memory-fairness.md index 2ce97d215ce..9cacfae75ca 100644 --- a/issues/no-physical-memory-fairness.md +++ b/issues/no-physical-memory-fairness.md @@ -94,7 +94,8 @@ netstack's piped connections, and `SYS_CONNECT` pinning 4 MiB into an unbounded pending queue — and all three have a bound *and* a caller that hears the refusal: `toyos_desktop::max_windows` and netstack's `places_for` each divide an eighth of physical memory by what one unit costs and refuse past it, -and a pipe allocates its ring page on first use rather than at `create`. Their +and a pipe takes memory for its bytes as it holds them rather than at +`create`. Their memory is counted to whichever job made it: a client's own connection is its job's, but a pipe a server makes and hands out (fileserver's stream pipes, netstack's bells) is the server's, which the server bounds. With no diff --git a/issues/the-compiler-is-the-unused-oracle-for-kernel-region-lifetime.md b/issues/the-compiler-is-the-unused-oracle-for-kernel-region-lifetime.md index 586f66301aa..6170e006107 100644 --- a/issues/the-compiler-is-the-unused-oracle-for-kernel-region-lifetime.md +++ b/issues/the-compiler-is-the-unused-oracle-for-kernel-region-lifetime.md @@ -36,6 +36,6 @@ only where it agrees with a separate stress of the real copy path. `build/sourcegate.rs`'s `no_slice_or_exclusive_reference_is_built_over_a_mapped_page` is the precedent for preferring a compile-or-gate answer here: it closed the -boundary's version of this question over `sdk/abi/src/ring.rs` and -`kernel/src/user_ptr.rs`, and the defect record it closed is already deleted +boundary's version of this question over `kernel/src/user_ptr.rs` and the +pipe ring the kernel then kept, and the defect record it closed is already deleted from this tracker. diff --git a/kernel/pure/frames.rs b/kernel/pure/frames.rs index 2c76ec0656e..831d6a0a6c5 100644 --- a/kernel/pure/frames.rs +++ b/kernel/pure/frames.rs @@ -36,8 +36,8 @@ //! **Everything a process causes is counted to its account's [`Ledger`], //! and an account's floor is held back against every other account.** What //! a floor account does not use of its floor is held as frames, and as the -//! whole superframes it makes, so that it can still be promised, tabled, -//! lent a run and grow its own heap pool however short every other account +//! whole superframes it makes, so that it can still be promised, given +//! frames, lent a run and grow its own heap pool however short every other account //! runs. What it gives back is held for its floor again as it comes back: //! a frame a copy still pins comes back with its last unpin, and a whole //! superframe only once one is whole. An ordinary account has no floor and @@ -136,10 +136,11 @@ pub struct Ledger { /// Frames held back for it, however short every other account runs: /// zero for an ordinary account. floor: AtomicU64, - /// Pages committed to it, promised or drawn; frames of its page tables; - /// superframes of its runs. + /// Pages committed to it, promised or drawn; frames the kernel holds for + /// it one by one, its page tables' and its pipes' bytes; superframes of + /// its runs. committed: AtomicU64, - tables: AtomicU64, + held: AtomicU64, runs: AtomicU64, /// Heap bytes charged to it, those of them charged to its own pool, and /// the superframes that pool is entitled to. @@ -165,7 +166,7 @@ pub struct Ledger { pub struct Usage { pub floor: u64, pub committed: u64, - pub tables: u64, + pub held: u64, pub runs: u64, pub heap: u64, pub entitled: u64, @@ -206,7 +207,7 @@ impl Ledger { Self { floor: AtomicU64::new(0), committed: AtomicU64::new(0), - tables: AtomicU64::new(0), + held: AtomicU64::new(0), runs: AtomicU64::new(0), heap: AtomicU64::new(0), own: AtomicU64::new(0), @@ -226,7 +227,7 @@ impl Ledger { Usage { floor: read(&self.floor), committed: read(&self.committed), - tables: read(&self.tables), + held: read(&self.held), runs: read(&self.runs), heap: read(&self.heap), entitled: read(&self.entitled), @@ -239,7 +240,7 @@ impl Ledger { /// Whether it holds nothing at all. pub fn is_empty(&self) -> bool { let u = self.usage(); - (u.committed, u.tables, u.runs, u.heap) == (0, 0, 0, 0) + (u.committed, u.held, u.runs, u.heap) == (0, 0, 0, 0) } /// The heap pool a charge it takes now is taken from: its own once it @@ -256,7 +257,7 @@ impl Ledger { /// since that is what is held for its heap. fn floor_used(&self) -> u64 { let u = self.usage(); - u.committed + u.tables + PER_SUPER as u64 * (u.runs + u.entitled) + u.committed + u.held + PER_SUPER as u64 * (u.runs + u.entitled) } /// Frames of its floor it does not use. @@ -267,7 +268,7 @@ impl Ledger { /// The frames it holds, its heap rounded up. fn footprint(&self) -> u64 { let u = self.usage(); - u.committed + u.tables + PER_SUPER as u64 * u.runs + u.heap.div_ceil(FRAME) + u.committed + u.held + PER_SUPER as u64 * u.runs + u.heap.div_ceil(FRAME) } fn add(word: &AtomicU64, n: u64) { @@ -797,25 +798,32 @@ impl<'m> Frames<'m> { self.take_frame() } - /// One frame for a page table of `l`'s, which is not promised: refused + /// `count` frames the kernel holds for `l`, such as page tables, which + /// are not promised, each handed to `each`: refused whole, nothing taken, /// rather than take one held back. - pub fn alloc_table(&mut self, l: &Ledger) -> Option { - let Some(floors) = self.admit(l, 1, 1, 0) else { - self.refused(l, 1, 0, 0); - return None; + pub fn alloc_held(&mut self, l: &Ledger, count: u64, mut each: impl FnMut(u64)) -> bool { + let Some(floors) = self.admit(l, count, count, 0) else { + self.refused(l, count, 0, 0); + return false; }; self.floors = floors; - let phys = self.take_frame().expect("frames: an admitted frame is free"); - Ledger::add(&l.tables, 1); + for _ in 0..count { + each(self.take_frame().expect("frames: an admitted frame is free")); + } + Ledger::add(&l.held, count); l.grew(); - Some(phys) + true } - /// Give back a table frame [`alloc_table`](Self::alloc_table) took for - /// `l`. - pub fn free_table(&mut self, l: &Ledger, phys: u64, pins: &Pins) { - self.give_back(phys, pins); - self.give_back_for(l, |l| Ledger::sub(&l.tables, 1, "table frames")); + /// Give back `frames`, which [`alloc_held`](Self::alloc_held) took for + /// `l`; a pinned one is taken back by its last unpin instead. + pub fn free_held(&mut self, l: &Ledger, frames: impl IntoIterator, pins: &Pins) { + let mut count = 0; + for phys in frames { + self.give_back(phys, pins); + count += 1; + } + self.give_back_for(l, |l| Ledger::sub(&l.held, count, "held frames")); } // A whole superframe is split only when no split one has a frame free, so @@ -1300,6 +1308,30 @@ mod tests { machine(supers, &[(BASE, BASE + supers as u64 * SUPER)], &[]) } + /// One held frame for `l`, as a page table takes it. + fn held(frames: &mut Frames, l: &Ledger) -> Option { + let mut got = None; + frames.alloc_held(l, 1, |phys| got = Some(phys)).then(|| got.expect("an admitted frame was handed over")) + } + + /// A batch is admitted whole or not at all: past what is free it takes + /// nothing and records what it asked for, and what it took comes back. + #[test] + fn held_frames_are_taken_whole_or_not_at_all() { + let mut m = whole(2); + m.frames.keep(1); + let all = PER_SUPER as u64; + let mut taken = Vec::new(); + assert!(!m.frames.alloc_held(&m.l, all + 1, |phys| taken.push(phys)), "a batch into the kept superframe"); + assert!(taken.is_empty(), "a refused batch handed frames over"); + assert_eq!((m.frames.free_frames(), m.l.usage().held, m.l.usage().refusals), (2 * all, 0, 1)); + assert!(m.frames.alloc_held(&m.l, all, |phys| taken.push(phys))); + assert_eq!((taken.iter().collect::>().len() as u64, m.l.usage().held), (all, all)); + m.frames.free_held(&m.l, taken, &m.pins); + assert_eq!((m.frames.free_frames(), m.l.usage().held), (2 * all, 0)); + assert!(m.l.is_empty()); + } + #[test] fn a_frame_carves_one_superframe_and_its_return_makes_it_whole() { let mut m = whole(2); @@ -1337,7 +1369,7 @@ mod tests { assert!(m.frames.promise(&m.l, all - 10)); assert_eq!(m.frames.promised(), all - 10); assert_eq!(m.frames.alloc_run_for(&m.l, 1), None, "a superframe out of promised frames"); - let free: Vec = core::iter::from_fn(|| m.frames.alloc_table(&m.l)).collect(); + let free: Vec = core::iter::from_fn(|| held(&mut m.frames, &m.l)).collect(); assert_eq!(free.len(), 10, "only the unpromised frames go to anyone else"); assert!(!m.frames.promise(&m.l, 1), "nothing is left to promise"); @@ -1348,12 +1380,12 @@ mod tests { // again; one given back by none is anyone's. assert_eq!(m.frames.discard(drawn[0], &m.pins), Discarded::Freed); assert_eq!(m.frames.promised(), 1); - assert_eq!(m.frames.alloc_table(&m.l), None, "a frame promised again went to another"); + assert_eq!(held(&mut m.frames, &m.l), None, "a frame promised again went to another"); m.frames.free_drawn(&m.l, [drawn[1]], &m.pins); - assert!(m.frames.alloc_table(&m.l).is_some(), "a frame no promise holds was refused"); - assert_eq!(m.frames.alloc_table(&m.l), None, "the promised one went too"); + assert!(held(&mut m.frames, &m.l).is_some(), "a frame no promise holds was refused"); + assert_eq!(held(&mut m.frames, &m.l), None, "the promised one went too"); m.frames.unpromise(&m.l, 1); - assert!(m.frames.alloc_table(&m.l).is_some(), "an unpromised frame stayed held"); + assert!(held(&mut m.frames, &m.l).is_some(), "an unpromised frame stayed held"); } /// A kept superframe is the kernel's alone: refused to whatever a process @@ -1364,14 +1396,14 @@ mod tests { let mut m = whole(2); m.frames.keep(1); let all = PER_SUPER as u64; - let tables: Vec = core::iter::from_fn(|| m.frames.alloc_table(&m.l)).collect(); + let tables: Vec = core::iter::from_fn(|| held(&mut m.frames, &m.l)).collect(); assert_eq!(tables.len() as u64, all, "a table took a frame of the kept superframe"); assert_eq!(m.frames.alloc_run_for(&m.l, 1), None, "a process's run took the kept superframe"); assert_eq!(m.frames.held_back(), (0, 1)); let kernel = m.frames.alloc_frame().expect("the kernel was refused a kept frame"); assert_eq!(m.frames.held_back(), (0, 0), "a kept superframe split for the kernel stayed kept"); for &f in &tables { - m.frames.free_table(&m.l, f, &m.pins); + m.frames.free_held(&m.l, [f], &m.pins); } assert_eq!(m.frames.held_back(), (0, 1), "a superframe back whole was not kept again"); m.frames.free_frame(kernel, &m.pins); @@ -1638,12 +1670,12 @@ mod tests { let floor = 2 * PER + 100; let f = floored(&mut m, floor); exhaust(&mut m.frames, &m.l); - assert_eq!(m.frames.alloc_table(&m.l), None, "an ordinary table took a floor's frame"); + assert_eq!(held(&mut m.frames, &m.l), None, "an ordinary table took a floor's frame"); assert_eq!(m.frames.alloc_run_for(&m.l, 1), None, "an ordinary run took a floor's superframe"); assert!(!m.frames.charge(&m.l, Pool::Shared, SUPER), "an ordinary charge took a floor's superframe"); let run = m.frames.alloc_run_for(&f, 1).expect("the floor account was refused a run inside its floor"); - assert!(m.frames.alloc_table(&f).is_some(), "the floor account was refused a table inside its floor"); + assert!(held(&mut m.frames, &f).is_some(), "the floor account was refused a table inside its floor"); assert!(m.frames.promise(&f, floor - PER - 1), "the floor account was refused the rest of its floor"); assert!(!m.frames.promise(&f, 1), "the floor account took a frame past its floor"); assert_eq!(m.frames.floors_held(), (0, 0)); @@ -1660,9 +1692,9 @@ mod tests { let mut m = whole(4); let f = floored(&mut m, PER); // One split superframe: half its frames taken, half free and scattered. - let tables: Vec = (0..2 * PER).map(|_| m.frames.alloc_table(&m.l).unwrap()).collect(); + let tables: Vec = (0..2 * PER).map(|_| held(&mut m.frames, &m.l).unwrap()).collect(); for &t in tables.iter().step_by(2) { - m.frames.free_table(&m.l, t, &m.pins); + m.frames.free_held(&m.l, [t], &m.pins); } assert!(m.frames.room() >= PER, "the scattered frames cover a run"); assert_eq!(m.frames.alloc_run_for(&m.l, 1), Some(BASE + 2 * SUPER), "the one spare whole superframe"); @@ -1812,10 +1844,10 @@ mod tests { #[test] fn a_wait_for_a_superframe_is_answered_by_one_and_not_by_frames() { let mut m = whole(4); - let tables: Vec = core::iter::from_fn(|| m.frames.alloc_table(&m.l)).collect(); + let tables: Vec = core::iter::from_fn(|| held(&mut m.frames, &m.l)).collect(); assert_eq!(tables.len(), 4 * PER_SUPER); for &t in tables.iter().step_by(2) { - m.frames.free_table(&m.l, t, &m.pins); + m.frames.free_held(&m.l, [t], &m.pins); } assert!(m.frames.room() >= PER && m.frames.spare_supers() == 0, "the frames free are scattered"); let starved = Ledger::new(); @@ -1824,11 +1856,11 @@ mod tests { // All of the first superframe's frames but one come back: still no whole one. let first: Vec = tables.iter().copied().skip(1).step_by(2).filter(|&t| t < BASE + SUPER).collect(); for &t in &first[1..] { - m.frames.free_table(&m.l, t, &m.pins); + m.frames.free_held(&m.l, [t], &m.pins); } assert!(!m.frames.take_returned(), "answered with frames back and no whole superframe"); assert!(!m.frames.ready(&starved)); - m.frames.free_table(&m.l, first[0], &m.pins); + m.frames.free_held(&m.l, [first[0]], &m.pins); assert!(m.frames.take_returned(), "a whole superframe back went unanswered"); assert!(!m.frames.take_returned(), "answered twice"); assert!(m.frames.ready(&starved)); @@ -1842,11 +1874,11 @@ mod tests { fn a_floor_accounts_wait_is_answered_by_its_own_floor() { let mut m = whole(4); let f = floored(&mut m, PER); - let table = m.frames.alloc_table(&m.l).unwrap(); + let table = held(&mut m.frames, &m.l).unwrap(); assert!(m.pins.pin(table)); // Every other superframe and frame is taken, not promised: none is spare. while m.frames.alloc_run_for(&m.l, 1).is_some() {} - while m.frames.alloc_table(&m.l).is_some() {} + while held(&mut m.frames, &m.l).is_some() {} let run = m.frames.alloc_run_for(&f, 1).expect("a run inside its floor"); let pinned = run + 9 * FRAME; assert!(m.pins.pin(pinned)); @@ -1855,7 +1887,7 @@ mod tests { assert!(!m.frames.ready(&f)); // Something else is given back, and comes back pinned too: the floor // is still short of the frame its superframe needs. - m.frames.free_table(&m.l, table, &m.pins); + m.frames.free_held(&m.l, [table], &m.pins); assert!(!m.frames.take_returned(), "answered while its floor is short"); assert_eq!(m.pins.unpin(pinned), Unpinned::Release); m.frames.release(pinned); @@ -1955,7 +1987,7 @@ mod tests { m.frames.free_run_for(l, run, count, &m.pins); false } - 5 if a < 3 => match m.frames.alloc_table(l) { + 5 if a < 3 => match held(&mut m.frames, l) { Some(t) => { tables[a].push(t); false @@ -1964,7 +1996,7 @@ mod tests { }, 6 if !tables[a].is_empty() => { let t = tables[a].swap_remove(next() as usize % tables[a].len()); - m.frames.free_table(l, t, &m.pins); + m.frames.free_held(l, [t], &m.pins); false } 7 if a == 3 => !m.frames.promise(l, PER + next() % (4 * PER)) || { @@ -2025,8 +2057,8 @@ mod tests { let i = next() as usize % drawn.len(); m.frames.free_drawn(&m.l, [drawn.swap_remove(i)], &m.pins); } - 3 => tables.extend(m.frames.alloc_table(&m.l)), - 4 if !tables.is_empty() => m.frames.free_table(&m.l, tables.swap_remove(next() as usize % tables.len()), &m.pins), + 3 => tables.extend(held(&mut m.frames, &m.l)), + 4 if !tables.is_empty() => m.frames.free_held(&m.l, [tables.swap_remove(next() as usize % tables.len())], &m.pins), 5 => runs.extend(m.frames.alloc_run_for(&m.l, 1)), 6 if !runs.is_empty() => m.frames.free_run_for(&m.l, runs.swap_remove(next() as usize % runs.len()), 1, &m.pins), 7 => { @@ -2109,8 +2141,8 @@ mod tests { m.frames.free_drawn(&m.l, [drawn.swap_remove(i)], &m.pins); } } - 5 => tables.extend(m.frames.alloc_table(&m.l)), - 6 if !tables.is_empty() => m.frames.free_table(&m.l, tables.swap_remove(next() as usize % tables.len()), &m.pins), + 5 => tables.extend(held(&mut m.frames, &m.l)), + 6 if !tables.is_empty() => m.frames.free_held(&m.l, [tables.swap_remove(next() as usize % tables.len())], &m.pins), 7 => runs.extend(m.frames.alloc_run_for(&m.l, 1)), 8 if !runs.is_empty() => m.frames.free_run_for(&m.l, runs.swap_remove(next() as usize % runs.len()), 1, &m.pins), 9 => { diff --git a/kernel/pure/pipe.rs b/kernel/pure/pipe.rs index 44622051cc0..71515b83392 100644 --- a/kernel/pure/pipe.rs +++ b/kernel/pure/pipe.rs @@ -1,15 +1,26 @@ -//! Who holds a pipe's two ends, what a watch on one of them is told, and how -//! a write lands in its ring. +//! Who holds a pipe's two ends, what a watch on one of them is told, how a +//! write lands, and the frames its bytes are kept in. //! //! A pipe end is held by counted references, one per object that names it: a //! handle duplicated or moved to another process is the same object or another //! reference, so an end is gone only when the last of them is. Every answer -//! here is a function of the two counts and of what the ring holds, read under +//! here is a function of the two counts and of what the pipe holds, read under //! the kernel's pipe lock. #![forbid(unsafe_code)] -use toyos_abi::syscall::MAX_WHOLE_PIPE_WRITE; +use alloc::collections::vec_deque::{Drain, VecDeque}; +use core::ops::Range; + +use toyos_abi::syscall::{MAX_WHOLE_PIPE_WRITE, PIPE_CAPACITY}; + +/// The bytes of one frame a pipe's bytes are kept in. +pub const FRAME: usize = crate::frames::FRAME as usize; + +/// The most frames a pipe holds: its capacity, the first of them part-read. +pub const MOST_FRAMES: usize = PIPE_CAPACITY / FRAME + 1; + +const _: () = assert!(PIPE_CAPACITY.is_multiple_of(FRAME) && MAX_WHOLE_PIPE_WRITE <= PIPE_CAPACITY); /// One end of a pipe. #[derive(Clone, Copy, PartialEq, Eq, Debug)] @@ -82,23 +93,23 @@ impl Holders { /// A read answers now: with bytes, or with the end of a stream no writer /// can add to. - pub fn readable(&self, available: u32) -> bool { + pub fn readable(&self, available: usize) -> bool { available > 0 || !self.held(End::Write) } /// A write waiting for `want` bytes of room answers now: it lands, or is /// refused for want of a reader. - pub fn writable(&self, space: u32, want: usize) -> bool { - space as usize >= want || !self.held(End::Read) + pub fn writable(&self, space: usize, want: usize) -> bool { + space >= want || !self.held(End::Read) } - /// The end opposite `end` has no holder, whatever the ring holds. + /// The end opposite `end` has no holder, whatever the pipe holds. pub fn other_end_gone(&self, end: End) -> bool { !self.held(end.other()) } } -/// How a write lands in a pipe's ring. +/// How a write lands in a pipe. #[derive(Clone, Copy, PartialEq, Eq, Debug)] pub enum Landing { /// All of it or none: a message. @@ -123,12 +134,132 @@ impl Landing { /// How many of `len` bytes land with `space` free; `None` until there is /// [`Self::room`]. - pub fn takes(self, space: u32, len: usize) -> Option { - let space = space as usize; + pub fn takes(self, space: usize, len: usize) -> Option { (space >= self.room(len)).then(|| len.min(space)) } } + +/// What a pipe holds, in frames taken as it fills and given back as it +/// drains. Its first unread byte is `head` bytes into the first frame, and +/// the frames past its last byte are spare: kept for the next write while +/// there are no more of them than frames holding bytes, and one besides. So a +/// pipe holds at most twice the frames its bytes need and one more, never more +/// than [`MOST_FRAMES`], and a stream that keeps a backlog in it takes no +/// frame from the allocator. +pub struct Bytes { + frames: VecDeque, + head: usize, + len: usize, +} + +impl Bytes { + pub const fn new() -> Self { + Self { frames: VecDeque::new(), head: 0, len: 0 } + } + + pub fn len(&self) -> usize { + self.len + } + + pub fn is_empty(&self) -> bool { + self.len == 0 + } + + pub fn space(&self) -> usize { + PIPE_CAPACITY - self.len + } + + /// The frames it holds, spares among them. + pub fn frames(&self) -> usize { + self.frames.len() + } + + /// The frames it can list without growing its list. + pub fn listed(&self) -> usize { + self.frames.capacity() + } + + /// Grow its list to `frames`, at most [`MOST_FRAMES`]. + pub fn list(&mut self, frames: usize) { + assert!(frames <= MOST_FRAMES, "pipe: a list of {frames} frames"); + self.frames.reserve_exact(frames.saturating_sub(self.frames.len())); + } + + /// The frames holding bytes. + fn used(&self) -> usize { + (self.head + self.len).div_ceil(FRAME) + } + + /// How many frames more a write of `n` bytes, at most [`Self::space`], + /// needs than it holds. + pub fn wants(&self, n: usize) -> usize { + assert!(n <= self.space(), "pipe: a {n}-byte write with room for {}", self.space()); + (self.head + self.len + n).div_ceil(FRAME).saturating_sub(self.frames.len()) + } + + /// One frame more for the next write. + pub fn add(&mut self, frame: F) { + assert!(self.frames.len() < MOST_FRAMES, "pipe: a frame past the most it holds"); + self.frames.push_back(frame); + } + + /// Append `n` bytes, which [`Self::wants`] no more frames for: `copy` is + /// handed each run of them as its frame, where they go in it, and where + /// they start among the `n`. + pub fn write(&mut self, n: usize, mut copy: impl FnMut(&mut F, Range, usize)) { + assert_eq!(self.wants(n), 0, "pipe: a write past the frames it was given"); + let (mut done, mut at) = (0, self.head + self.len); + while done < n { + let (frame, off) = (at / FRAME, at % FRAME); + let run = (FRAME - off).min(n - done); + copy(&mut self.frames[frame], off..off + run, done); + (done, at) = (done + run, at + run); + } + self.len += n; + } + + /// Take up to `want` bytes from the front, handing `copy` each run as + /// [`Self::write`] does; a frame it empties becomes a spare. Answers how + /// many it took. + pub fn read(&mut self, want: usize, mut copy: impl FnMut(&F, Range, usize)) -> usize { + let n = want.min(self.len); + let mut done = 0; + while done < n { + let run = (FRAME - self.head).min(n - done); + copy(&self.frames[0], self.head..self.head + run, done); + (done, self.head, self.len) = (done + run, self.head + run, self.len - run); + if self.head == FRAME { + self.frames.rotate_left(1); + self.head = 0; + } + } + if self.len == 0 { + self.head = 0; + } + n + } + + /// The spares past those it keeps, for the caller to give back. + pub fn surplus(&mut self) -> Drain<'_, F> { + let used = self.used(); + let keep = used + (self.frames.len() - used).min(used.max(1)); + self.frames.drain(keep..) + } + + /// Every frame, for the caller to give back, and with them every byte. + pub fn clear(&mut self) -> Drain<'_, F> { + (self.head, self.len) = (0, 0); + self.frames.drain(..) + } +} + +impl Default for Bytes { + fn default() -> Self { + Self::new() + } +} + #[cfg(test)] mod tests { use super::*; @@ -229,8 +360,8 @@ mod tests { Self { holders: made(), ring: VecDeque::new(), cap } } - fn space(&self) -> u32 { - (self.cap - self.ring.len()) as u32 + fn space(&self) -> usize { + self.cap - self.ring.len() } fn write(&mut self, bytes: &[u8], landing: Landing) -> Wrote { @@ -459,4 +590,141 @@ mod tests { fn competing_partial_writes_tear_in_the_same_run() { assert!(compete(Landing::Partial).is_err(), "the run never split a record, so it proves nothing"); } + + /// A frame of the model: its bytes and a name, so a frame handed out twice + /// or lost is seen. + struct Frame { + id: usize, + bytes: Vec, + } + + /// A pipe's bytes with an allocator that counts what it hands out. + struct Held { + bytes: Bytes, + taken: usize, + returned: Vec, + } + + impl Held { + fn new() -> Self { + Self { bytes: Bytes::new(), taken: 0, returned: Vec::new() } + } + + /// Write `src` as `kernel::pipe` does: the frames it wants, then the + /// copy. + fn write(&mut self, src: &[u8]) { + for _ in 0..self.bytes.wants(src.len()) { + self.bytes.add(Frame { id: self.taken, bytes: vec![0xEE; FRAME] }); + self.taken += 1; + } + self.bytes.write(src.len(), |frame, within, at| { + frame.bytes[within.clone()].copy_from_slice(&src[at..at + within.len()]); + }); + } + + fn read(&mut self, want: usize) -> Vec { + let mut out = vec![0; want]; + let n = self.bytes.read(want, |frame, within, at| { + out[at..at + within.len()].copy_from_slice(&frame.bytes[within]); + }); + out.truncate(n); + self.returned.extend(self.bytes.surplus().map(|frame| frame.id)); + out + } + + /// Each frame handed out once is held or was given back, never both. + fn accounted(&self) { + let mut seen: Vec = self.bytes.frames.iter().map(|f| f.id).chain(self.returned.iter().copied()).collect(); + seen.sort_unstable(); + assert_eq!(seen, (0..self.taken).collect::>(), "a frame was lost or given back twice"); + } + } + + fn pattern(from: usize, len: usize) -> Vec { + (from..from + len).map(|i| (i % 251) as u8).collect() + } + + /// Writes and reads of every size, across frame edges and the whole + /// capacity, come back as written, and what the pipe keeps stays within + /// its bound: twice the frames its bytes need and one, at most + /// `MOST_FRAMES`. + #[test] + fn a_stream_comes_back_exact_and_the_frames_stay_bounded() { + let mut seed = 0x9e37_79b9_u64; + let mut random = move |below: usize| { + seed ^= seed << 13; + seed ^= seed >> 7; + seed ^= seed << 17; + (seed % below as u64) as usize + }; + let mut pipe = Held::new(); + let (mut written, mut read) = (0, 0); + for turn in 0..4000 { + let most = [1, 64, FRAME - 1, FRAME + 1, 64 << 10, PIPE_CAPACITY][turn % 6]; + let n = random(most + 1).min(pipe.bytes.space()); + pipe.write(&pattern(written, n)); + written += n; + assert!(pipe.bytes.frames() <= MOST_FRAMES); + let got = pipe.read(random(most + 1)); + assert_eq!(got, pattern(read, got.len()), "turn {turn}: bytes {read}.. came back wrong"); + read += got.len(); + assert!( + pipe.bytes.frames() <= 2 * pipe.bytes.used() + 1, + "turn {turn}: {} frames kept for {} bytes", + pipe.bytes.frames(), + pipe.bytes.len() + ); + pipe.accounted(); + } + let rest = pipe.read(PIPE_CAPACITY); + assert_eq!(rest, pattern(read, written - read)); + assert_eq!(pipe.bytes.frames(), 1, "a drained pipe keeps one frame"); + } + + /// A full pipe whose first frame is part-read holds `MOST_FRAMES`. + #[test] + fn a_full_pipe_part_read_holds_the_most_frames() { + let mut pipe = Held::new(); + pipe.write(&pattern(0, 10)); + assert_eq!(pipe.read(5), pattern(0, 5)); + let rest = pipe.bytes.space(); + pipe.write(&pattern(10, rest)); + assert_eq!((pipe.bytes.len(), pipe.bytes.frames()), (PIPE_CAPACITY, MOST_FRAMES)); + assert_eq!(pipe.read(PIPE_CAPACITY), [pattern(5, 5), pattern(10, rest)].concat()); + } + + /// A stream that keeps a backlog in the pipe, a little or nearly all of + /// it, takes no frame after its first round: the frames a read empties + /// carry the next write. + #[test] + fn a_steady_stream_takes_no_frames() { + const CHUNK: usize = 64 << 10; + for backlog in [CHUNK, PIPE_CAPACITY - CHUNK] { + let mut pipe = Held::new(); + pipe.write(&vec![1; backlog]); + pipe.write(&vec![1; CHUNK]); + pipe.read(CHUNK); + let taken = pipe.taken; + for _ in 0..100 { + pipe.write(&vec![1; CHUNK]); + pipe.read(CHUNK); + } + assert_eq!(pipe.taken, taken, "a backlog of {backlog} bytes took frames each round"); + } + } + + /// A pipe that held much and is drained gives back all but one frame, + /// and `clear` gives back every one. + #[test] + fn a_drained_pipe_gives_its_frames_back() { + let mut pipe = Held::new(); + pipe.write(&vec![1; PIPE_CAPACITY]); + assert_eq!(pipe.bytes.frames(), PIPE_CAPACITY / FRAME); + pipe.read(PIPE_CAPACITY); + assert_eq!((pipe.bytes.frames(), pipe.returned.len()), (1, PIPE_CAPACITY / FRAME - 1)); + pipe.write(&[1; 3]); + assert_eq!(pipe.taken, PIPE_CAPACITY / FRAME, "the kept frame did not carry the next write"); + assert_eq!(pipe.bytes.clear().count(), 1); + assert_eq!((pipe.bytes.len(), pipe.bytes.frames()), (0, 0)); + } } diff --git a/kernel/src/arch/aarch64/paging.rs b/kernel/src/arch/aarch64/paging.rs index 8201e749860..f3f41b0d091 100644 --- a/kernel/src/arch/aarch64/paging.rs +++ b/kernel/src/arch/aarch64/paging.rs @@ -197,7 +197,7 @@ fn published() { fn free_table(job: &Job, phys: u64) { // SAFETY: the caller unlinked the table, after the shootdown that let // every CPU stop walking it. - unsafe { crate::mm::pmm::free_table_frame(job, phys) }; + unsafe { crate::mm::pmm::free_held(job, [phys]) }; } /// A root and every table under it, each a frame the descriptor linking it diff --git a/kernel/src/arch/x86_64/paging.rs b/kernel/src/arch/x86_64/paging.rs index 65707e17d9a..aee401096e1 100644 --- a/kernel/src/arch/x86_64/paging.rs +++ b/kernel/src/arch/x86_64/paging.rs @@ -464,7 +464,7 @@ fn alloc_pcid() -> Option { fn free_table(job: &Job, phys: u64) { // SAFETY: the caller unlinked the table, after the shootdown that let // every CPU stop walking it. - unsafe { crate::mm::pmm::free_table_frame(job, phys) }; + unsafe { crate::mm::pmm::free_held(job, [phys]) }; } /// The kernel half every space shares: PML4[256..511] and what they name. diff --git a/kernel/src/mm/pmm.rs b/kernel/src/mm/pmm.rs index dd474d594f5..b05debc23d8 100644 --- a/kernel/src/mm/pmm.rs +++ b/kernel/src/mm/pmm.rs @@ -6,7 +6,7 @@ //! drops. Pins ([`pin_range`]) are taken without the allocator's lock. //! //! **What a process causes is counted to the job that caused it**, its -//! account — a commitment, a page table, a pipe's ring, a run a device is +//! account — a commitment, a page table, a pipe's bytes, a run a device is //! lent, a [`charge`] of the kernel heap — and is refused rather than take //! what the allocator holds back for the kernel or for another account's //! floor ([`set_floor`]); only the kernel's own allocations may take what it @@ -309,21 +309,35 @@ pub fn alloc_promised() -> Frame { /// A zeroed frame for a page table of `job`'s, refused rather than take what /// is held back: a process's tables are not promised. It goes back through -/// [`free_table_frame`]. +/// [`free_held`]. pub fn alloc_table_frame(job: &Job) -> Option { - let phys = with(|f| f.alloc_table(job.memory()))?; + let mut phys = 0; + if !with(|f| f.alloc_held(job.memory(), 1, |frame| phys = frame)) { + return None; + } // SAFETY: just taken off the allocator, so unaliased, and inside the direct map. unsafe { core::ptr::write_bytes(DirectMap::from_phys(phys).as_mut_ptr::(), 0, FRAME as usize) }; Some(phys) } -/// Give back a frame [`alloc_table_frame`] took for `job`. +/// `count` frames for kernel memory of `job`'s that no process maps and that +/// is written before it is read, such as a pipe's bytes, each handed to +/// `each` under one hold of the allocator's lock: refused whole rather than +/// take what is held back. They hold what their last owner left, and go back +/// through [`free_held`]. +pub fn alloc_held(job: &Job, count: usize, each: impl FnMut(u64)) -> bool { + count == 0 || with(|f| f.alloc_held(job.memory(), count as u64, each)) +} + +/// Give back `frames` [`alloc_table_frame`] or [`alloc_held`] took for +/// `job`, under one hold of the allocator's lock. /// /// # Safety -/// No table links it any more, and no CPU can still walk into it. -pub unsafe fn free_table_frame(job: &Job, phys: u64) { +/// Nothing reaches any of them any more: no table links it, no CPU can still +/// walk into it, and nothing else frees it. +pub unsafe fn free_held(job: &Job, frames: impl IntoIterator) { let pins = pins(); - with(|f| f.free_table(job.memory(), phys, &pins)); + with(|f| f.free_held(job.memory(), frames, &pins)); } /// One zeroed 2 MiB page for the kernel's own use. @@ -338,8 +352,8 @@ fn zeroed(page: PhysPage) -> PhysPage { page } -/// Physically contiguous 2 MiB pages a job holds, such as a pipe's ring or -/// memory a device is lent; they go back, counted to it, when this drops. +/// Physically contiguous 2 MiB pages a job holds, such as memory a device is +/// lent; they go back, counted to it, when this drops. pub struct Run { base: u64, count: usize, diff --git a/kernel/src/object/job.rs b/kernel/src/object/job.rs index 0f7128236ae..34414d1ffac 100644 --- a/kernel/src/object/job.rs +++ b/kernel/src/object/job.rs @@ -67,7 +67,7 @@ impl JobObject { AccountInfo { floor: pages(u.floor), committed: pages(u.committed), - kernel: u.heap.saturating_add(pages(u.tables)).saturating_add(u.runs.saturating_mul(PAGE_2M)), + kernel: u.heap.saturating_add(pages(u.held)).saturating_add(u.runs.saturating_mul(PAGE_2M)), peak: pages(u.peak), wanted: pages(u.wanted), refusals: u.refusals, diff --git a/kernel/src/pipe.rs b/kernel/src/pipe.rs index 548a396ec9b..47ca4bdc758 100644 --- a/kernel/src/pipe.rs +++ b/kernel/src/pipe.rs @@ -1,15 +1,14 @@ +use core::ops::Range; + use crate::mm::charge::{self, Charge, Exhausted}; -use crate::mm::pmm; +use crate::mm::{pmm, DirectMap}; use kernel::message::Window; -use kernel::pipe::{End, Holders, Landing, Released}; +use kernel::pipe::{Bytes, End, Holders, Landing, Released, MOST_FRAMES}; use kernel::sched::job::Job; -use toyos_abi::ring::Ring; use alloc::sync::Arc; - -use crate::mm::PAGE_2M; use crate::watch::Watch; use crate::sync::Lock; use crate::user_ptr::{UserBytes, UserBytesMut}; @@ -85,18 +84,55 @@ mod handle { } -pub const PIPE_SIZE: usize = PAGE_2M as usize; +/// What a pipe holds: its bytes, in frames counted to the job that made the +/// pipe, and the heap their list takes, charged to it. The frames go back +/// when it drops. +struct Held { + bytes: Bytes, + list: Charge, +} -const _: () = assert!( - toyos_abi::syscall::MAX_WHOLE_PIPE_WRITE <= PIPE_SIZE - core::mem::size_of::(), - "a whole pipe write must fit the ring", -); +impl Held { + fn new(job: &Arc) -> Self { + Self { bytes: Bytes::new(), list: Charge::none(job) } + } + + /// Take the frames a write of `n` bytes wants, all of them or none. + fn take_frames(&mut self, n: usize) -> Result<(), Exhausted> { + let wants = self.bytes.wants(n); + if wants == 0 { + return Ok(()); + } + let frames = self.bytes.frames() + wants; + if frames > self.bytes.listed() { + let list = frames.next_power_of_two().min(MOST_FRAMES); + self.list.resize(charge::array::(list))?; + self.bytes.list(list); + } + let bytes = &mut self.bytes; + pmm::alloc_held(self.list.job(), wants, |frame| bytes.add(frame)).then_some(()).ok_or(Exhausted) + } + + /// Give back the spares a read left past those it keeps. + fn give_back_surplus(&mut self) { + // SAFETY: a spare holds no byte, and nothing but this list names it. + unsafe { pmm::free_held(self.list.job(), self.bytes.surplus()) }; + } +} + +impl Drop for Held { + fn drop(&mut self) { + // SAFETY: nothing but this list names its frames. + unsafe { pmm::free_held(self.list.job(), self.bytes.clear()) }; + } +} -/// A pipe's ring page and the cursors over it, in kernel memory. -struct Backing { - /// Owns the page `ring` addresses, counted to the job that made the pipe. - _page: pmm::Run, - ring: Ring, +/// `within` of a frame a pipe holds, which no process maps and only that +/// pipe's lock reaches. +fn frame(frame: &u64, within: Range) -> *mut u8 { + assert!(within.end <= kernel::pipe::FRAME, "pipe: bytes {within:?} past a frame"); + // SAFETY: inside the frame, which the direct map covers. + unsafe { DirectMap::from_phys(*frame).as_mut_ptr::().add(within.start) } } struct Pipe { @@ -110,16 +146,14 @@ struct Pipe { /// moved through the pipe wakes no poll that asked about a leaving. left: [Arc; 2], /// Its own heap and its watches, charged to the job that made it, which - /// its ring is counted to as well. - charge: Charge, + /// what it holds is counted to as well. + _charge: Charge, } /// What a pipe's lock holds. struct State { - /// `None` until first use — allocating eagerly would charge every pending - /// `SYS_CONNECT` 4 MiB before either end sent a byte — and again once - /// neither end is held. - backing: Option, + /// Given back once no reader is left to read it. + held: Held, holders: Holders, /// Set when a write should hand the next reader transient RT priority — covers readers already runnable, which the wake-time boost misses. rt_boost_pending: bool, @@ -129,9 +163,6 @@ struct State { read: u64, } -// SAFETY: `Backing::ring`'s raw pointer and its owning `pmm::Run` move together in the same struct, and the pipe's lock is the only way to reach a `State`, serializing all access to it. -unsafe impl Send for State {} - impl Pipe { /// The heap one takes: itself and its four watches. fn heap() -> usize { @@ -139,51 +170,19 @@ impl Pipe { } } -impl State { - /// Allocate the ring page if this is the first use, counted to `job`, the - /// one that made the pipe, whoever writes first: a connection's are both - /// its connecting job's, so a server's reply spends nothing of its own. - /// `None` where what is left is held back, which the write is refused. - fn back(&mut self, job: &Arc) -> Option<&mut Backing> { - if self.backing.is_none() { - let page = pmm::Run::new(job, 1)?; - // SAFETY: a fresh 2 MiB page this `State` owns for as long as the `Ring` addresses it. - let ring = unsafe { Ring::new(page.direct_map().as_mut_ptr(), PIPE_SIZE) }; - self.backing = Some(Backing { _page: page, ring }); - self.publish_ends(); - } - self.backing.as_mut() - } - - /// Republish "is the other end gone?" into the mapped header, for netstack; the kernel itself decides from its own counts. - fn publish_ends(&mut self) { - let Some(backing) = self.backing.as_mut() else { return }; - if self.holders.held(End::Read) { backing.ring.open_reader() } else { backing.ring.close_reader() } - if self.holders.held(End::Write) { backing.ring.open_writer() } else { backing.ring.close_writer() } - } - - fn available(&self) -> u32 { - self.backing.as_ref().map_or(0, |b| b.ring.available()) - } - - /// A pipe with no page yet reports its whole capacity free — the allocation is deferred, not refused. - fn space(&self) -> u32 { - self.backing.as_ref().map_or(u32::MAX, |b| b.ring.space()) - } -} - /// Create a new pipe. Returns owned reader + writer references. /// -/// A pipe with no traffic owns no physical memory: its ring page is -/// allocated at its first `write`, which is refused if none is left. +/// A pipe holds memory for its bytes only as it holds them: a write takes +/// what it needs, or is refused if none is left. pub fn create() -> Result<(PipeReader, PipeWriter), Exhausted> { let charge = Charge::current(Pipe::heap())?; + let held = Held::new(charge.job()); let pipe = PipeRef(Arc::new(Pipe { - state: Lock::new(State { backing: None, holders: Holders::new(), rt_boost_pending: false, written: 0, read: 0 }), + state: Lock::new(State { held, holders: Holders::new(), rt_boost_pending: false, written: 0, read: 0 }), read_watch: Arc::new(Watch::new()), write_watch: Arc::new(Watch::new()), left: [Arc::new(Watch::new()), Arc::new(Watch::new())], - charge, + _charge: charge, })); Ok((PipeReader::acquire(&pipe), PipeWriter::acquire(&pipe))) } @@ -198,15 +197,14 @@ pub enum PipeWrite { /// Not the room its landing needs now: wait for `Landing::room`. NoRoom, BrokenPipe, - /// The ring page could not be allocated; no wait makes space appear, so a caller must not park on it. + /// The memory its bytes need could not be taken; no wait for room makes + /// it appear, so a caller must not park on it. NoMemory, } impl PipeRef { fn hold(&self, end: End) { - let mut state = self.0.state.lock(); - state.holders.hold(end); - state.publish_ends(); + self.0.state.lock().holders.hold(end); } pub fn try_read(&self, buf: &mut UserBytesMut) -> Option { @@ -215,7 +213,7 @@ impl PipeRef { } /// [`Self::try_read`], taking only what `window` allows of the bytes the - /// ring holds, given the reader's stream position: [`AtBatch`] where it + /// pipe holds, given the reader's stream position: [`AtBatch`] where it /// says the reader is at a pinned batch. pub fn try_read_within( &self, @@ -224,22 +222,24 @@ impl PipeRef { ) -> Option> { let (result, boost) = { let mut pipe = self.0.state.lock(); - let limit = match window(pipe.read, pipe.available() as usize) { + let available = pipe.held.bytes.len(); + let limit = match window(pipe.read, available) { Window::AtBatch => return Some(Err(AtBatch)), Window::Bytes(limit) => limit, }; - if pipe.available() > 0 && limit > 0 { + if available > 0 && limit > 0 { let len = buf.len().min(limit); - let n = pipe - .backing - .as_mut() - .expect("available() > 0 implies a ring") - .ring - .read(len, |off, src| buf.write_run(off, &src)); + let n = pipe.held.bytes.read(len, |held, within, at| { + let n = within.len(); + // SAFETY: bytes of a frame this pipe holds, which only its + // lock reaches and nothing writes while it is held. + buf.write_at(at, unsafe { core::slice::from_raw_parts(frame(held, within), n) }); + }); + pipe.held.give_back_surplus(); pipe.read += n as u64; let boost = core::mem::take(&mut pipe.rt_boost_pending); (Some(Ok(n)), boost) - } else if !pipe.holders.held(End::Write) && pipe.available() == 0 { + } else if !pipe.holders.held(End::Write) && available == 0 { (Some(Ok(0)), false) } else { (None, false) @@ -263,23 +263,29 @@ impl PipeRef { if !pipe.holders.held(End::Read) { return PipeWrite::BrokenPipe; } - let Some(backing) = pipe.back(self.0.charge.job()) else { return PipeWrite::NoMemory }; - let Some(n) = landing.takes(backing.ring.space(), buf.len()) else { return PipeWrite::NoRoom }; - let wrote = backing.ring.write(n, |off, mut dst| buf.read_run(off, &mut dst)); - assert_eq!(wrote, n, "pipe: a ring with room for a write took less of it"); + let Some(n) = landing.takes(pipe.held.bytes.space(), buf.len()) else { return PipeWrite::NoRoom }; + if pipe.held.take_frames(n).is_err() { + return PipeWrite::NoMemory; + } + pipe.held.bytes.write(n, |held, within, at| { + let n = within.len(); + // SAFETY: bytes of a frame this pipe holds, which only its lock + // reaches and nothing reads while it is held. + buf.read_at(at, unsafe { core::slice::from_raw_parts_mut(frame(held, within), n) }); + }); pipe.written += n as u64; PipeWrite::Wrote { n, end: pipe.written } } pub fn has_data(&self) -> bool { let pipe = self.0.state.lock(); - pipe.holders.readable(pipe.available()) + pipe.holders.readable(pipe.held.bytes.len()) } /// Whether a write waiting for `want` bytes of room answers now. pub fn has_space(&self, want: usize) -> bool { let pipe = self.0.state.lock(); - pipe.holders.writable(pipe.space(), want) + pipe.holders.writable(pipe.held.bytes.space(), want) } /// Whether the end opposite `end` of this pipe has no holder left. @@ -293,17 +299,20 @@ impl PipeRef { } /// One reference to `end` lets go; its end's last posts the other end's - /// watch and its leaving watch, and the last of either end frees the ring. + /// watch and its leaving watch, and the read end's last gives back what + /// the pipe holds, which nobody can read any more. fn release(&self, end: End) { - let (released, page) = { + let (released, unread) = { let mut pipe = self.0.state.lock(); let released = pipe.holders.release(end); - pipe.publish_ends(); - // Dropped outside the lock: the page goes back to the allocator. - let page = (released == Released::PipeGone).then(|| pipe.backing.take()).flatten(); - (released, page) + let unread = (!pipe.holders.held(End::Read) && pipe.held.bytes.frames() > 0).then(|| { + let job = Arc::clone(pipe.held.list.job()); + core::mem::replace(&mut pipe.held, Held::new(&job)) + }); + (released, unread) }; - drop(page); + // Outside the lock: its frames go back to the allocator. + drop(unread); match (released, end) { (Released::Held | Released::PipeGone, _) => return, (Released::EndGone, End::Read) => crate::scheduler::wake_pipe_writers(self), diff --git a/kernel/src/user_ptr.rs b/kernel/src/user_ptr.rs index 9d7978c334c..b0a97fcd342 100644 --- a/kernel/src/user_ptr.rs +++ b/kernel/src/user_ptr.rs @@ -219,18 +219,6 @@ impl UserBytes<'_> { self.0.read_into(off, dst); } - /// Copy the window at `off` into one ring run; panics if out of bounds, as - /// [`read_at`](Self::read_at) does. - pub fn read_run(&self, off: usize, dst: &mut toyos_abi::ring::Dst<'_>) { - let into = dst.as_mut_ptr(); - self.0.pieces("UserBytes::read_run", off, dst.len(), |run, at, n| { - // SAFETY: `pieces` bounds `[at, at + n)` inside the ring run, which - // `Ring::write` proved is inside its own data region, and `run..+n` - // inside one pinned page; the ring is kernel memory no window names. - unsafe { core::ptr::copy_nonoverlapping(run, into.add(at), n) } - }); - } - /// The `len`-byte window at `off` inside this one. pub fn sub(&self, off: usize, len: usize) -> UserBytes<'_> { UserBytes(self.0.sub("UserBytes::sub", off, len), PhantomData) @@ -254,19 +242,6 @@ impl UserBytesMut<'_> { self.0.write_from(off, src); } - /// Copy one ring run into the window at `off`; panics if out of bounds, as - /// [`write_at`](Self::write_at) does. - pub fn write_run(&mut self, off: usize, src: &toyos_abi::ring::Src<'_>) { - let from = src.as_ptr(); - self.0.pieces("UserBytesMut::write_run", off, src.len(), |run, at, n| { - // SAFETY: `pieces` bounds `[at, at + n)` inside the ring run, which - // `Ring::read` proved is inside its own data region, and `run..+n` - // inside one pinned page its range let the process write; the ring - // is kernel memory no window names. - unsafe { core::ptr::copy_nonoverlapping(from.add(at), run, n) } - }); - } - /// Zero `len` bytes of the window at `off`. pub fn fill_zero(&mut self, off: usize, len: usize) { self.0.pieces("UserBytesMut::fill_zero", off, len, |run, _, n| { diff --git a/sdk/abi/src/handle.rs b/sdk/abi/src/handle.rs index 9206185f8aa..86cfb1978a8 100644 --- a/sdk/abi/src/handle.rs +++ b/sdk/abi/src/handle.rs @@ -149,7 +149,7 @@ impl Rights { /// On memory: map it writable, or lend it to a device; a holder without /// it maps the region only readable. pub const WRITE: Rights = Rights(1 << 3); - /// Shared memory, a pipe's ring page, an [inbox](crate::inbox)'s rings. + /// Shared memory, an [inbox](crate::inbox)'s rings. pub const MAP: Rights = Rights(1 << 4); /// Block on it, or name it in an [`OP_WATCH`](crate::inbox::OP_WATCH). pub const WAIT: Rights = Rights(1 << 5); diff --git a/sdk/abi/src/lib.rs b/sdk/abi/src/lib.rs index 4596b33cc0d..0d011085998 100644 --- a/sdk/abi/src/lib.rs +++ b/sdk/abi/src/lib.rs @@ -28,7 +28,6 @@ pub mod inventory; pub mod log; pub mod pci; pub mod quit; -pub mod ring; pub mod syscall; pub mod trace; pub mod usersafe; diff --git a/sdk/abi/src/ring.rs b/sdk/abi/src/ring.rs deleted file mode 100644 index 3e4439d73c1..00000000000 --- a/sdk/abi/src/ring.rs +++ /dev/null @@ -1,479 +0,0 @@ -//! Lock-free SPSC ring buffer for shared-memory pipes. -//! -//! Layout: a `RingHeader` followed by `capacity` bytes of data. -//! -//! The region is memory the kernel allocates, which no process maps. `Ring` is -//! the kernel-side owner and holds every value the copies are bounded by; -//! `RingHeader` holds only flags nothing reads back, left from when a process -//! could map the ring (`issues/a-pipe-is-a-two-mebibyte-page-for-a-mapping-nothing-makes.md`). -//! -//! **The data region is never a `&[u8]` or `&mut [u8]`** — plain bytes in that -//! mapping — so [`Src`] and [`Dst`] describe the copies instead. A reference -//! over the *header* is sound and `header` takes one: one `AtomicU32`, atomics. - -use core::marker::PhantomData; -use core::sync::atomic::{AtomicU32, Ordering}; - -pub const RING_WRITER_CLOSED: u32 = 1; -pub const RING_READER_CLOSED: u32 = 2; - -/// What a process sees at offset 0 of a mapped ring. -/// -/// Every field is writable by any process holding the pipe, so nothing the -/// kernel indexes, bounds or divides by may live here. `flags` is a -/// publication netstack polls to notice a peer that went away; the kernel only -/// ever stores to it, and answers "is the other end gone?" from its own -/// refcounts. -#[repr(C, align(64))] -pub struct RingHeader { - pub flags: AtomicU32, -} - -impl RingHeader { - pub fn is_writer_closed(&self) -> bool { - self.flags.load(Ordering::Acquire) & RING_WRITER_CLOSED != 0 - } - - pub fn is_reader_closed(&self) -> bool { - self.flags.load(Ordering::Acquire) & RING_READER_CLOSED != 0 - } -} - -/// One contiguous run of a ring's data region, handed to [`Ring::read`]'s sink: -/// a pointer and a length, and no reference at all — the rule -/// `kernel::user_ptr` already states for a user buffer, for the same reason. -pub struct Src<'a> { - ptr: *const u8, - len: usize, - _scope: PhantomData<&'a ()>, -} - -impl Src<'_> { - pub fn len(&self) -> usize { - self.len - } - - pub fn is_empty(&self) -> bool { - self.len == 0 - } - - /// The run's first byte. Raw, and it stays raw: see this type's doc. - pub fn as_ptr(&self) -> *const u8 { - self.ptr - } - - /// Copy the whole run into `dst`, which must be exactly as long. - pub fn copy_to(&self, dst: &mut [u8]) { - assert_eq!(dst.len(), self.len, "a ring run copied into a {}-byte destination", dst.len()); - // SAFETY: `Ring::read` built this run inside its own data region, and - // `dst` is an owned `&mut [u8]` of the same length — a reference, so it - // cannot alias the run. - unsafe { core::ptr::copy_nonoverlapping(self.ptr, dst.as_mut_ptr(), self.len) }; - } -} - -/// [`Src`] for [`Ring::write`]'s fill: the same pointer and length, in the -/// other direction, and never a `&mut [u8]` for the same reason. -pub struct Dst<'a> { - ptr: *mut u8, - len: usize, - _scope: PhantomData<&'a mut ()>, -} - -impl Dst<'_> { - pub fn len(&self) -> usize { - self.len - } - - pub fn is_empty(&self) -> bool { - self.len == 0 - } - - pub fn as_mut_ptr(&mut self) -> *mut u8 { - self.ptr - } - - /// Copy `src`, which must be exactly as long, over the whole run. - pub fn copy_from(&mut self, src: &[u8]) { - assert_eq!(src.len(), self.len, "a {}-byte source copied into a ring run", src.len()); - // SAFETY: as [`Src::copy_to`], mirrored. - unsafe { core::ptr::copy_nonoverlapping(src.as_ptr(), self.ptr, self.len) }; - } -} - -/// Kernel-side owner of one ring region: the cursors, the capacity, and the -/// base of the region they address. -/// -/// Every access is serialized by the caller's lock (each `kernel::pipe` pipe's own), -/// which is why the cursors are plain integers rather than atomics — the -/// header's were only ever atomic because they lived in shared memory. -pub struct Ring { - base: *mut u8, - capacity: u32, - write_cursor: u32, - read_cursor: u32, -} - -impl Ring { - /// Claim `total_size` bytes at `base` as a ring, initializing the header - /// the mapping process will see. - /// - /// # Safety - /// `base` must be writable and aligned for `RingHeader`, and the region - /// must stay valid for the life of the `Ring`. - pub unsafe fn new(base: *mut u8, total_size: usize) -> Self { - let capacity = total_size - core::mem::size_of::(); - assert!( - capacity > 0 && capacity < (1usize << 31), - "ring capacity {capacity} does not leave room for the cursor modulus" - ); - // SAFETY: the caller's contract above (`# Safety`) is exactly this - // write's precondition — `base` writable and aligned for - // `RingHeader` — and the `capacity` assert just above proves - // `total_size >= size_of::()`, so the write lands inside - // the caller-promised region. - unsafe { - (base as *mut RingHeader).write(RingHeader { - flags: AtomicU32::new(0), - }) - }; - Self { - base, - capacity: capacity as u32, - write_cursor: 0, - read_cursor: 0, - } - } - - fn header(&self) -> &RingHeader { - // SAFETY: `new`'s caller promised `base` writable and aligned for - // `RingHeader` for the life of this `Ring`, and `new` wrote a valid - // one there. The header is racily writable by any process holding - // the pipe (its own doc comment says so), which is why its one field - // is an `AtomicU32` accessed only through atomic ops below — never - // read or written as a plain `u32` — so a concurrent scribble is a - // torn or stale load, never a reference to invalid data. - unsafe { &*(self.base as *const RingHeader) } - } - - fn data_ptr(&self) -> *mut u8 { - // SAFETY: `capacity = total_size - size_of::()` was - // asserted `> 0` in `new`, so this offset is strictly inside the - // `total_size`-byte region `new`'s caller promised — never dereferenced - // here, only computed. - unsafe { self.base.add(core::mem::size_of::()) } - } - - /// The cursors count modulo this, and a stream byte's ring offset is its - /// cursor value modulo `capacity`. - /// - /// Those two are consistent only if `capacity` divides the modulus, which - /// is why it is `2 * capacity` and not the `u32`'s own `2^32`: `capacity` - /// is whatever a 2 MiB page leaves after a 64-byte header, and it does not - /// divide `2^32`. Where the modulus is not a multiple of `capacity`, the - /// two cursor values naming the same stream byte on either side of the - /// wrap map to *different* offsets, and an access straddling the wrap - /// lands on the wrong bytes. - /// - /// Twice, rather than once, so that full stays distinguishable from empty: - /// `available` ranges over `0..=capacity` and both ends are representable. - fn modulus(&self) -> u64 { - self.capacity as u64 * 2 - } - - fn advance(&self, cursor: u32, by: usize) -> u32 { - ((cursor as u64 + by as u64) % self.modulus()) as u32 - } - - pub fn available(&self) -> u32 { - let w = self.write_cursor as u64; - let r = self.read_cursor as u64; - let m = self.modulus(); - ((w + m - r) % m) as u32 - } - - pub fn space(&self) -> u32 { - self.capacity - self.available() - } - - /// Read up to `want` bytes, handing `sink` each contiguous run of them with - /// its offset in the destination. Returns the number of bytes read. - /// - /// The run is a [`Src`] and not a `&[u8]`: the page is mapped into a - /// process writable. One or two runs, never more: the second is the ring's - /// own wrap. - pub fn read(&mut self, want: usize, mut sink: impl FnMut(usize, Src<'_>)) -> usize { - let avail = self.available() as usize; - if avail == 0 { - return 0; - } - let count = want.min(avail); - let cap = self.capacity as usize; - let offset = self.read_cursor as usize % cap; - let data = self.data_ptr(); - - let first = count.min(cap - offset); - // SAFETY: `offset < cap` and `first = count.min(cap - offset)`, so - // `data.add(offset)..+first` stays inside the `cap`-byte data region - // `data_ptr` computed, and the wrap run `data..+(count - first)` does - // too since `count - first <= cap`. - let (head, tail) = unsafe { (data.add(offset), data) }; - sink(0, Src { ptr: head, len: first, _scope: PhantomData }); - if first < count { - sink(first, Src { ptr: tail, len: count - first, _scope: PhantomData }); - } - - self.read_cursor = self.advance(self.read_cursor, count); - count - } - - /// Write up to `want` bytes, asking `fill` for each contiguous run of them - /// by its offset in the source. Returns the number of bytes written. - /// - /// The mirror of [`Self::read`], and the same reason for [`Dst`]. - pub fn write(&mut self, want: usize, mut fill: impl FnMut(usize, Dst<'_>)) -> usize { - let free = self.space() as usize; - if free == 0 { - return 0; - } - let count = want.min(free); - let cap = self.capacity as usize; - let offset = self.write_cursor as usize % cap; - let data = self.data_ptr(); - - let first = count.min(cap - offset); - // SAFETY: the same bounds argument as `read`'s matching block. - let (head, tail) = unsafe { (data.add(offset), data) }; - fill(0, Dst { ptr: head, len: first, _scope: PhantomData }); - if first < count { - fill(first, Dst { ptr: tail, len: count - first, _scope: PhantomData }); - } - - self.write_cursor = self.advance(self.write_cursor, count); - count - } - - pub fn open_writer(&self) { - self.header().flags.fetch_and(!RING_WRITER_CLOSED, Ordering::Release); - } - - pub fn open_reader(&self) { - self.header().flags.fetch_and(!RING_READER_CLOSED, Ordering::Release); - } - - pub fn close_writer(&self) { - self.header().flags.fetch_or(RING_WRITER_CLOSED, Ordering::Release); - } - - pub fn close_reader(&self) { - self.header().flags.fetch_or(RING_READER_CLOSED, Ordering::Release); - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::alloc::{alloc_zeroed, dealloc, Layout}; - - /// The ring `kernel::pipe` builds: one 2 MiB page, header at offset 0. - const PIPE_TOTAL: usize = 2 * 1024 * 1024; - - struct Backing { - ptr: *mut u8, - layout: Layout, - } - - impl Backing { - fn new(total: usize) -> Self { - let layout = Layout::from_size_align(total, core::mem::align_of::()).unwrap(); - // SAFETY: every caller passes a `total` (`PIPE_TOTAL` or a - // `const TOTAL`) that is a non-zero, hard-coded byte count, so - // `layout` has non-zero size. - let ptr = unsafe { alloc_zeroed(layout) }; - assert!(!ptr.is_null(), "test backing allocation failed"); - Self { ptr, layout } - } - - fn ring(&self) -> Ring { - // SAFETY: `ptr` was just allocated above with `layout`, whose - // alignment is `align_of::()` and whose size is - // `layout.size()` — exactly `Ring::new`'s two preconditions — - // and `self` (and the memory it owns) outlives every `Ring` this - // method hands out, since nothing here ever drops `self` early. - unsafe { Ring::new(self.ptr, self.layout.size()) } - } - } - - impl Drop for Backing { - fn drop(&mut self) { - // SAFETY: `ptr`/`layout` are exactly what `new` allocated them - // with, and this is the only place that frees them. - unsafe { dealloc(self.ptr, self.layout) } - } - } - - /// The slice forms the kernel cannot have: here the destination is an - /// ordinary allocation, so describing it buys nothing. - fn read_slice(ring: &mut Ring, buf: &mut [u8]) -> usize { - let want = buf.len(); - ring.read(want, |off, src| src.copy_to(&mut buf[off..off + src.len()])) - } - - fn write_slice(ring: &mut Ring, buf: &[u8]) -> usize { - ring.write(buf.len(), |off, mut dst| dst.copy_from(&buf[off..off + dst.len()])) - } - - /// Byte at absolute stream position `pos`. Every aligned 16-byte group - /// carries its own index, so a misread shows up whether it is off by a - /// multiple of 16 (wrong stamp) or not (wrong filler). - fn stream_byte(pos: u64) -> u8 { - let group = (pos >> 4) as u32; - match pos & 15 { - k @ 0..=3 => (group >> (8 * k)) as u8, - _ => 0xC3, - } - } - - /// `stream_byte` over a range, touching only the four stamped bytes of - /// each group: the test moves gibibytes and runs unoptimised, so a - /// per-byte closure is most of its runtime. - fn fill(buf: &mut [u8], start: u64) { - buf.fill(0xC3); - let end = start + buf.len() as u64; - let mut pos = start; - while pos < end { - let k = pos & 15; - if k < 4 { - buf[(pos - start) as usize] = stream_byte(pos); - pos += 1; - } else { - pos += 16 - k; - } - } - } - - /// The premise, computed rather than asserted from memory: `capacity` is - /// whatever the header's alignment leaves of a 2 MiB page, and it does not - /// divide the `u32` cursor space. - #[test] - fn the_pipe_capacity_does_not_divide_the_u32_cursor_space() { - let header = core::mem::size_of::(); - let capacity = (PIPE_TOTAL - header) as u64; - let remainder = (1u64 << 32) % capacity; - assert_ne!( - remainder, 0, - "header {header} B, capacity {capacity} B: a cursor wrapping at 2^32 \ - would be sound only if capacity divided it" - ); - assert_eq!( - (header, capacity, remainder), - (64, 2_097_088, 131_072), - "the pipe ring's shape moved — re-derive the wrap argument before \ - changing this" - ); - } - - /// The fast gate: a read split across the ring's own wrap point must - /// return what was written. Seeded, because reaching the wrap honestly - /// costs the gibibytes the next test spends. - #[test] - fn a_read_split_across_the_cursor_wrap_returns_what_was_written() { - let backing = Backing::new(PIPE_TOTAL); - let mut ring = backing.ring(); - - let seed = (ring.modulus() - 16) as u32; - ring.write_cursor = seed; - ring.read_cursor = seed; - - let sent: [u8; 32] = core::array::from_fn(|i| stream_byte(i as u64)); - assert_eq!(write_slice(&mut ring, &sent), 32); - - let mut got = [0u8; 32]; - assert_eq!(read_slice(&mut ring, &mut got[..16]), 16); - assert_eq!( - ring.read_cursor, 0, - "the seed did not put the wrap between the two halves" - ); - assert_eq!(read_slice(&mut ring, &mut got[16..]), 16); - assert_eq!( - got, sent, - "a read straddling the cursor wrap did not return what was written" - ); - } - - /// That the seeded state above is reachable at all: a long-lived pipe - /// carries a cursor there by ordinary traffic. Slow on purpose — this is - /// the proof the fast gate is guarding a state a real pipe reaches, and - /// with the modulus at 2*capacity it also crosses ~33k wraps on the way. - #[test] - fn a_stream_survives_the_cursor_reaching_two_to_the_thirty_two() { - const TOTAL: usize = 64 * 1024; - let backing = Backing::new(TOTAL); - let mut ring = backing.ring(); - let cap = ring.capacity as u64; - let target = (1u64 << 32) + 4 * cap; - - // Coprime-ish with the capacity, so accesses straddle the buffer end - // and the cursor wrap rather than lining up on either. - let mut wbuf = std::vec![0u8; 4093]; - let mut rbuf = std::vec![0u8; 3571]; - let mut expect = std::vec![0u8; 3571]; - - let mut written = 0u64; - let mut read = 0u64; - while read < target { - while written < target { - let n = wbuf.len().min((target - written) as usize); - fill(&mut wbuf[..n], written); - let put = write_slice(&mut ring, &wbuf[..n]); - written += put as u64; - if put < n { - break; - } - } - let got = read_slice(&mut ring, &mut rbuf); - assert!(got > 0, "ring stalled at {read} of {target}"); - fill(&mut expect[..got], read); - if rbuf[..got] != expect[..got] { - let off = (0..got).position(|i| rbuf[i] != expect[i]).unwrap(); - panic!( - "stream byte {} came back {:#04x}, want {:#04x} — this read \ - spans {}..{}, and the cursor modulus is {}", - read + off as u64, - rbuf[off], - expect[off], - read, - read + got as u64, - ring.modulus(), - ); - } - read += got as u64; - } - assert_eq!(read, target); - } - - /// Whatever the header holds, the stream must be unaffected — nothing the - /// copies are bounded by is in it. - #[test] - fn a_scribbled_header_leaves_the_stream_exact() { - const TOTAL: usize = 64 * 1024; - let backing = Backing::new(TOTAL); - let mut ring = backing.ring(); - let mut pos = 0u64; - // The third value is where `capacity` used to sit; zero was the - // divisor, and 0xFF.. the 4 GiB pointer offset. - for pattern in [0x00u8, 0xFF, 0xA5] { - // SAFETY: `backing.ptr` is `TOTAL` bytes from `Backing::new`, and - // `TOTAL` (64 KiB) is far larger than `size_of::()` - // (64 bytes), so the write stays inside the allocation. - unsafe { core::ptr::write_bytes(backing.ptr, pattern, core::mem::size_of::()) }; - let mut sent = std::vec![0u8; 3571]; - fill(&mut sent, pos); - assert_eq!(write_slice(&mut ring, &sent), sent.len()); - let mut got = std::vec![0u8; sent.len()]; - assert_eq!(read_slice(&mut ring, &mut got), sent.len()); - assert_eq!(got, sent, "header pattern {pattern:#04x} changed the stream"); - pos += sent.len() as u64; - } - } -} diff --git a/sdk/abi/src/syscall.rs b/sdk/abi/src/syscall.rs index 3f469e4cfe6..08b9ccc6c75 100644 --- a/sdk/abi/src/syscall.rs +++ b/sdk/abi/src/syscall.rs @@ -1195,6 +1195,11 @@ pub const MAX_PATH: usize = 4096; /// libc's `PIPE_BUF` (`sdk/libc/include/limits.h`). pub const MAX_WHOLE_PIPE_WRITE: usize = 4096; +/// The most a pipe holds unread: a write past it waits for its reader. The +/// kernel takes memory for what a pipe holds as it fills and gives it back as +/// it drains, so this is a bound, not what a pipe costs. +pub const PIPE_CAPACITY: usize = 2 << 20; + /// Open a file. pub fn open(path: &[u8], flags: OpenFlags) -> Result { check(syscall(SYS_OPEN, path.as_ptr() as u64, path.len() as u64, flags.0, 0)).map(|v| RawHandle(v as u32)) diff --git a/system/fileserver/src/main.rs b/system/fileserver/src/main.rs index 160696bd340..ac81b968af9 100644 --- a/system/fileserver/src/main.rs +++ b/system/fileserver/src/main.rs @@ -79,9 +79,9 @@ use toyos_abi::syscall::{SyscallError, MAX_BADGE, SERVE_PREFIX}; const STREAM_READ: usize = 64 * 1024; /// The turns that drain what one stream holds before a request on its -/// description: the kernel's whole pipe ring (`PIPE_SIZE`, 2 MiB), so a -/// writer that keeps writing cannot hold the loop. -const STREAM_PRESENT: usize = (2 << 20) / STREAM_READ; +/// description: the most a pipe holds, so a writer that keeps writing cannot +/// hold the loop. +const STREAM_PRESENT: usize = syscall::PIPE_CAPACITY / STREAM_READ; /// The name the private export port has in this server's own namespace. const EXPORT_PORT: &str = "export"; diff --git a/system/netstack/src/main.rs b/system/netstack/src/main.rs index 1fb7b4ddd77..813a4dfc8fb 100644 --- a/system/netstack/src/main.rs +++ b/system/netstack/src/main.rs @@ -33,7 +33,7 @@ use toyos::ipc::{self, RxStep}; use toyos::poller::{Poller, READABLE}; use toyos::say; use toyos::AsHandle; -use toyos_abi::syscall::PciId; +use toyos_abi::syscall::{PciId, PIPE_CAPACITY}; use toyos_manifest::device::PciMatch; use toyos_dhcp::{HostName, Lease}; use toyos_i219::Part; @@ -95,17 +95,18 @@ const LEASE_BOUND: Duration = Duration::from_millis(toyos_tco::LEASE_BOUND_MS); /// keeps up (`toyos-net-tcp`'s `rx`); a send buffer is the region's send ring. const TCP_BUFFER: u32 = toyos::net::stream::RECEIVE_BYTES; -/// A kernel pipe is one 2 MiB page (`kernel/src/pipe.rs`). The client -/// allocates it, and netstack holding its far end is what keeps it alive. -const PIPE_BYTES: u64 = 2 * 1024 * 1024; +/// The most a kernel pipe holds unread; the frames it takes for that add a +/// frame and their list's few KiB (`kernel::pipe::MOST_FRAMES`). A client's +/// pipes are counted to it, and netstack holding their far ends keeps what is +/// in them alive. +const PIPE_BYTES: u64 = PIPE_CAPACITY as u64; -/// What one holder of a stream holds of this machine: the two pipes of its -/// connection and its two bells, each a page once written -/// (`issues/a-pipe-is-a-two-mebibyte-page-for-a-mapping-nothing-makes.md`). +/// What one holder of a stream can make this machine hold: the two pipes of +/// its connection and its two bells, each full. const HOLDER_BYTES: u64 = 4 * PIPE_BYTES; -/// What one holder of a listener or a datagram socket holds: the two pipes of -/// its connection and its bell. +/// What one holder of a listener or a datagram socket can make it hold: the +/// two pipes of its connection and its bell. const SOCKET_HOLDER_BYTES: u64 = 3 * PIPE_BYTES; /// What one place can make this machine hold, at the largest of the things a diff --git a/tests/guest/rust-tests/src/bin/kernel_objects_refused.rs b/tests/guest/rust-tests/src/bin/kernel_objects_refused.rs index 8843f98fc53..186f54c68f7 100644 --- a/tests/guest/rust-tests/src/bin/kernel_objects_refused.rs +++ b/tests/guest/rust-tests/src/bin/kernel_objects_refused.rs @@ -37,7 +37,7 @@ const LONG_NAME: usize = 3900; fn main() { // Made before anything is taken, so what follows needs no heap of its - // own: the pipe whose ring is not yet allocated, the file not yet + // own: the pipe that holds no frame yet, the file not yet // written, every list at its bound, and stdout's buffer. let unwritten = syscall::pipe().expect("a pipe before anything is taken"); let file = syscall::open(b"/tmp/kernel_objects_refused.bin", OpenFlags::READ | OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE) @@ -64,7 +64,7 @@ fn main() { ("a pipe", syscall::pipe().map(|_| ())), ("a port", syscall::port_create().map(|_| ())), ("a directory", syscall::mkdir(b"/tmp/kernel_objects_refused_more")), - ("a pipe's first write, its ring", syscall::write(unwritten.write, b"x").map(|_| ())), + ("a pipe's first write, its frame", syscall::write(unwritten.write, b"x").map(|_| ())), ("a /tmp file's page", syscall::write(file, &[0x5a; PAGE]).map(|_| ())), ("a /tmp file's page at an offset", syscall::pwrite(file, &[0x5a; PAGE], HOLE).map(|_| ())), ("a memory object", syscall::shm_create(PAGE).map(|_| ())), diff --git a/tests/guest/rust-tests/src/bin/memory_floor.rs b/tests/guest/rust-tests/src/bin/memory_floor.rs index 61459003959..1b509a8f40c 100644 --- a/tests/guest/rust-tests/src/bin/memory_floor.rs +++ b/tests/guest/rust-tests/src/bin/memory_floor.rs @@ -8,7 +8,7 @@ //! while that holds them: //! //! - a server in an ordinary job answers this job's connection, before any of -//! that, and its account does not move: a connection's rings are its +//! that, and its account does not move: a connection's pipes are its //! client's; //! - an ordinary job's spawn is refused, so the exhaustion is real; //! - the supervisor answers a stop's request on a new connection, which takes @@ -17,7 +17,7 @@ //! refused it; the supervisor is refused its restart, which it says in the //! log, and keeps its port, where a new client's request waits; //! - with `--input`, a key the host types reaches `inputsink` through -//! ps2server, whose floor holds the ring its connection's first write takes; +//! ps2server, whose floor holds the frames its connection's writes take; //! //! and then it kills the greedy job, and the waiting request is answered by a //! `starveling` the supervisor started again because memory came back. @@ -86,7 +86,7 @@ fn control(input: bool) { println!(" memory_floor: this job's floor is {} KiB, and it holds {} KiB", account.floor >> 10, (account.committed + account.kernel) >> 10); // Every connection that crosses the exhaustion is made and used once - // before it, so neither end allocates a ring for it then. + // before it, so each of its pipes keeps a frame for what crosses it then. let keys = input.then(|| { let keys = endow::service("keys").expect("the image gives test-runner `keys`"); keys.signal(HELLO).expect("hello to inputsink"); @@ -101,7 +101,7 @@ fn control(input: bool) { let mut log = Log { pipe: logkeeper_api::read().expect("the log, before anything is taken").pipe, bytes: vec![0; 64 << 10], held: 0 }; log.drain(); - rings_are_the_clients(&cap); + pipes_are_the_clients(&cap); let job = cap.new_job().expect("an ordinary job for the greedy one"); let mut greedy = Command::new(ME) @@ -188,10 +188,10 @@ fn control(input: bool) { } /// A server in an ordinary job of its own accepts this job's connection and -/// answers it: both of the connection's rings are its client's, so the +/// answers it: both of the connection's pipes are its client's, so the /// server's reply costs its job nothing, and no client can spend a server's /// floor by connecting to it. -fn rings_are_the_clients(cap: &SysCap) { +fn pipes_are_the_clients(cap: &SysCap) { let (acceptor, connector) = toyos::port::create().expect("a port"); let job = cap.new_job().expect("a job for the server"); let mut server = Command::new(ME) @@ -217,7 +217,7 @@ fn rings_are_the_clients(cap: &SysCap) { assert_eq!(after.kernel, before.kernel, "the server's reply was charged to the server"); server.kill().expect("end the server"); server.wait().expect("the server's end"); - println!(" PASS: a connection's rings are its client's: the server's reply cost its job nothing"); + println!(" PASS: a connection's pipes are its client's: the server's reply cost its job nothing"); } /// Accept one connection on the endowed acceptor and answer its [`PING`]. diff --git a/tests/guest/rust-tests/src/bin/poller_capacity.rs b/tests/guest/rust-tests/src/bin/poller_capacity.rs index 67de4586f47..352861141f8 100644 --- a/tests/guest/rust-tests/src/bin/poller_capacity.rs +++ b/tests/guest/rust-tests/src/bin/poller_capacity.rs @@ -21,8 +21,7 @@ use toyos::poller::{Poller, READABLE}; const SELF_PATH: &str = "/system/bin/test_rs_poller_capacity"; -/// Each pipe is one 2 MiB kernel page, so this is 128 MiB of ring — enough to -/// be a real batch, small enough to leave the machine alone. +/// Enough to be a real batch, small enough to leave the machine alone. const CAP: u32 = 64; /// Printed by a child immediately before the call that must not return. From e644ae78a13803bf7d4d30d2e64bc7801a23217b Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 28 Oct 2026 05:25:00 +0100 Subject: [PATCH 02/12] review r1: a read that frees nothing takes no allocator lock, and a wake is never refused B1: pmm::free_held returns before the allocator's lock when it is handed no frame, so a pipe read that empties no frame past those kept, and a drop of a pipe that holds none, take no global lock. It takes an exact size iterator for that. S1: a pipe that has held a byte keeps what an empty one needs, so a write of at most MAX_WHOLE_PIPE_WRITE into it while empty is never refused for memory; PIPE_CAPACITY says so. toyos::wake::pair writes and takes one wake, so every bell has held a byte, and Waker::wake reads a refusal for memory as a wake already pending, which it then is. B2, S2, S4: the comments that still named a pipe's ring or its pages are corrected, and a frame's bytes are `frame_bytes`. Co-Authored-By: Claude Opus 5.5 --- kernel/pure/frames.rs | 4 ++-- kernel/src/mm/pmm.rs | 8 ++++++-- kernel/src/pipe.rs | 10 +++++----- sdk/abi/src/syscall.rs | 13 ++++++++----- sdk/toyos/src/wake.rs | 16 +++++++++++----- 5 files changed, 32 insertions(+), 19 deletions(-) diff --git a/kernel/pure/frames.rs b/kernel/pure/frames.rs index 831d6a0a6c5..937161c4278 100644 --- a/kernel/pure/frames.rs +++ b/kernel/pure/frames.rs @@ -917,8 +917,8 @@ impl<'m> Frames<'m> { Some(run) } - /// `count` superframes in a row for `l`, such as a pipe's ring or memory - /// a device is lent: refused rather than take a superframe held back. + /// `count` superframes in a row for `l`, such as memory a device is lent: + /// refused rather than take a superframe held back. pub fn alloc_run_for(&mut self, l: &Ledger, count: usize) -> Option { let supers = count as u64; let frames = supers * PER_SUPER as u64; diff --git a/kernel/src/mm/pmm.rs b/kernel/src/mm/pmm.rs index b05debc23d8..c7cde998b39 100644 --- a/kernel/src/mm/pmm.rs +++ b/kernel/src/mm/pmm.rs @@ -330,12 +330,16 @@ pub fn alloc_held(job: &Job, count: usize, each: impl FnMut(u64)) -> bool { } /// Give back `frames` [`alloc_table_frame`] or [`alloc_held`] took for -/// `job`, under one hold of the allocator's lock. +/// `job`, under one hold of the allocator's lock, which none takes. /// /// # Safety /// Nothing reaches any of them any more: no table links it, no CPU can still /// walk into it, and nothing else frees it. -pub unsafe fn free_held(job: &Job, frames: impl IntoIterator) { +pub unsafe fn free_held>(job: &Job, frames: impl IntoIterator) { + let frames = frames.into_iter(); + if frames.len() == 0 { + return; + } let pins = pins(); with(|f| f.free_held(job.memory(), frames, &pins)); } diff --git a/kernel/src/pipe.rs b/kernel/src/pipe.rs index 47ca4bdc758..15ca41d92b9 100644 --- a/kernel/src/pipe.rs +++ b/kernel/src/pipe.rs @@ -129,7 +129,7 @@ impl Drop for Held { /// `within` of a frame a pipe holds, which no process maps and only that /// pipe's lock reaches. -fn frame(frame: &u64, within: Range) -> *mut u8 { +fn frame_bytes(frame: &u64, within: Range) -> *mut u8 { assert!(within.end <= kernel::pipe::FRAME, "pipe: bytes {within:?} past a frame"); // SAFETY: inside the frame, which the direct map covers. unsafe { DirectMap::from_phys(*frame).as_mut_ptr::().add(within.start) } @@ -229,11 +229,11 @@ impl PipeRef { }; if available > 0 && limit > 0 { let len = buf.len().min(limit); - let n = pipe.held.bytes.read(len, |held, within, at| { + let n = pipe.held.bytes.read(len, |frame, within, at| { let n = within.len(); // SAFETY: bytes of a frame this pipe holds, which only its // lock reaches and nothing writes while it is held. - buf.write_at(at, unsafe { core::slice::from_raw_parts(frame(held, within), n) }); + buf.write_at(at, unsafe { core::slice::from_raw_parts(frame_bytes(frame, within), n) }); }); pipe.held.give_back_surplus(); pipe.read += n as u64; @@ -267,11 +267,11 @@ impl PipeRef { if pipe.held.take_frames(n).is_err() { return PipeWrite::NoMemory; } - pipe.held.bytes.write(n, |held, within, at| { + pipe.held.bytes.write(n, |frame, within, at| { let n = within.len(); // SAFETY: bytes of a frame this pipe holds, which only its lock // reaches and nothing reads while it is held. - buf.read_at(at, unsafe { core::slice::from_raw_parts_mut(frame(held, within), n) }); + buf.read_at(at, unsafe { core::slice::from_raw_parts_mut(frame_bytes(frame, within), n) }); }); pipe.written += n as u64; PipeWrite::Wrote { n, end: pipe.written } diff --git a/sdk/abi/src/syscall.rs b/sdk/abi/src/syscall.rs index 08b9ccc6c75..1d784f84b1e 100644 --- a/sdk/abi/src/syscall.rs +++ b/sdk/abi/src/syscall.rs @@ -1032,9 +1032,9 @@ pub mod debug_action { /// Create a pipe. Returns the read and write ends. /// -/// **Fallible, because `sys_pipe` is.** It answers `ResourceExhausted` on three -/// paths — no pipe pages, and either handle install hitting the table cap — and -/// a wrapper that cannot say so hands the caller a pair of handles that are an +/// **Fallible, because `sys_pipe` is.** It answers `ResourceExhausted` where +/// the pipe's heap cannot be charged or either handle install hits the table +/// cap, and a wrapper that cannot say so hands the caller a pair of handles that are an /// error word cut in half. /// /// A packed pair can never be mistaken for an error word: no handle is ever @@ -1197,7 +1197,10 @@ pub const MAX_WHOLE_PIPE_WRITE: usize = 4096; /// The most a pipe holds unread: a write past it waits for its reader. The /// kernel takes memory for what a pipe holds as it fills and gives it back as -/// it drains, so this is a bound, not what a pipe costs. +/// it drains, so this is a bound, not what a pipe costs. A pipe that has held +/// a byte keeps what an empty one needs: a write of at most +/// [`MAX_WHOLE_PIPE_WRITE`] bytes into it while it is empty is never refused +/// for memory. pub const PIPE_CAPACITY: usize = 2 << 20; /// Open a file. @@ -2809,7 +2812,7 @@ crate::user_safe! { /// Private and object memory its processes committed. pub committed: u64, /// Kernel memory it caused: heap its objects keep, its page tables, - /// and runs such as pipe rings and memory a device is lent. + /// pipes' bytes, and runs such as memory a device is lent. pub kernel: u64, /// The most of the two it held at once. pub peak: u64, diff --git a/sdk/toyos/src/wake.rs b/sdk/toyos/src/wake.rs index 9caaf46637d..36cfcbe9ad0 100644 --- a/sdk/toyos/src/wake.rs +++ b/sdk/toyos/src/wake.rs @@ -13,10 +13,15 @@ use toyos_abi::syscall::SyscallError; use crate::{pipe_pair, AsHandle, Pipe}; -/// A fresh bell and the waker that rings it. +/// A fresh bell and the waker that rings it. One wake is written and taken +/// here, so the pipe has held a byte and no later wake into an empty bell is +/// refused for memory (`toyos_abi::syscall::PIPE_CAPACITY`). pub fn pair() -> Result<(Waker, Bell), SyscallError> { let (read, write) = pipe_pair()?; - Ok((Waker(write), Bell(read))) + write.write_nonblock(&[1])?; + let bell = Bell(read); + bell.take(); + Ok((Waker(write), bell)) } /// Rings a [`Bell`] from any thread, and never waits. @@ -25,9 +30,10 @@ pub struct Waker(Pipe); impl Waker { pub fn wake(&self) { match self.0.write_nonblock(&[1]) { - // A full pipe is a wake already pending, and a bell nobody holds - // any more has nobody left to wake. - Ok(_) | Err(SyscallError::WouldBlock) | Err(SyscallError::Gone) => {} + // A full pipe is a wake already pending, and so is one refused + // memory, since a bell `pair` primed takes none while it is empty; + // a bell nobody holds any more has nobody left to wake. + Ok(_) | Err(SyscallError::WouldBlock) | Err(SyscallError::ResourceExhausted) | Err(SyscallError::Gone) => {} Err(e) => panic!("Waker::wake: the wake pipe refused a write: {e:?}"), } } From c20f31b1a4ad7f38a468f8d4e8a5dac6c4819de4 Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 28 Oct 2026 06:37:32 +0100 Subject: [PATCH 03/12] pipe: one walk of the user window a call, each piece copied into its frames The T14 read pipe.throughput.65536 at 4451 MB/s on 46848c491 against 6930 on 544f9cc3c, and 16384 at 3060 against 4393: a pipe copied each 4 KiB frame run with its own UserBytes::read_at or write_at, and each of those took the address space's lock, reached and pinned the user pages the run touched and unpinned them. A run at an arbitrary user offset touches two pages, so a 64 KiB write paid that walk 16 times over about 32 page reaches, where the 2 MiB ring paid it once per 16 pages. Bytes now places a write's bytes at an offset past what it holds and counts them with commit, and peeks a read's at an offset before consume lets them go. The pipe walks the user window once, through UserBytes::read_pieces and UserBytesMut::write_pieces, which hand out the pinned pieces read_at and write_at already copied through, and copies each piece into or out of the frame runs it covers. Review r2: MAX_WHOLE_PIPE_WRITE <= FRAME is asserted where the promise to an empty pipe rests on it, the post-drain test writes a whole write, and free_held's doc says plainly that giving back none takes no lock. Co-Authored-By: Claude Opus 5.5 --- kernel/pure/pipe.rs | 135 ++++++++++++++++++++++++++++++++--------- kernel/src/mm/pmm.rs | 3 +- kernel/src/pipe.rs | 34 ++++++++--- kernel/src/user_ptr.rs | 18 ++++++ 4 files changed, 152 insertions(+), 38 deletions(-) diff --git a/kernel/pure/pipe.rs b/kernel/pure/pipe.rs index 71515b83392..411c22e5c10 100644 --- a/kernel/pure/pipe.rs +++ b/kernel/pure/pipe.rs @@ -20,7 +20,9 @@ pub const FRAME: usize = crate::frames::FRAME as usize; /// The most frames a pipe holds: its capacity, the first of them part-read. pub const MOST_FRAMES: usize = PIPE_CAPACITY / FRAME + 1; -const _: () = assert!(PIPE_CAPACITY.is_multiple_of(FRAME) && MAX_WHOLE_PIPE_WRITE <= PIPE_CAPACITY); +// A whole write fits the one frame an emptied pipe keeps, which is what +// `PIPE_CAPACITY`'s promise to an empty pipe rests on. +const _: () = assert!(PIPE_CAPACITY.is_multiple_of(FRAME) && MAX_WHOLE_PIPE_WRITE <= FRAME); /// One end of a pipe. #[derive(Clone, Copy, PartialEq, Eq, Debug)] @@ -204,39 +206,67 @@ impl Bytes { self.frames.push_back(frame); } - /// Append `n` bytes, which [`Self::wants`] no more frames for: `copy` is + /// Place the `len` bytes `at` past the last byte held, inside the room a + /// write that [`Self::wants`] no more frames for may fill: `copy` is /// handed each run of them as its frame, where they go in it, and where - /// they start among the `n`. - pub fn write(&mut self, n: usize, mut copy: impl FnMut(&mut F, Range, usize)) { - assert_eq!(self.wants(n), 0, "pipe: a write past the frames it was given"); - let (mut done, mut at) = (0, self.head + self.len); - while done < n { - let (frame, off) = (at / FRAME, at % FRAME); - let run = (FRAME - off).min(n - done); + /// they start among the `len`. They are held once [`Self::commit`] counts + /// them. + pub fn place(&mut self, at: usize, len: usize, mut copy: impl FnMut(&mut F, Range, usize)) { + let end = self.head + self.len + at + len; + assert!(end <= self.frames.len() * FRAME, "pipe: bytes placed past the frames it was given"); + let (mut done, mut pos) = (0, self.head + self.len + at); + while done < len { + let (frame, off) = (pos / FRAME, pos % FRAME); + let run = (FRAME - off).min(len - done); copy(&mut self.frames[frame], off..off + run, done); - (done, at) = (done + run, at + run); + (done, pos) = (done + run, pos + run); } + } + + /// Count `n` more bytes held: what [`Self::place`] put at the end. + pub fn commit(&mut self, n: usize) { + assert_eq!(self.wants(n), 0, "pipe: bytes held past the frames it was given"); self.len += n; } - /// Take up to `want` bytes from the front, handing `copy` each run as - /// [`Self::write`] does; a frame it empties becomes a spare. Answers how - /// many it took. - pub fn read(&mut self, want: usize, mut copy: impl FnMut(&F, Range, usize)) -> usize { - let n = want.min(self.len); - let mut done = 0; - while done < n { - let run = (FRAME - self.head).min(n - done); - copy(&self.frames[0], self.head..self.head + run, done); - (done, self.head, self.len) = (done + run, self.head + run, self.len - run); - if self.head == FRAME { - self.frames.rotate_left(1); - self.head = 0; - } + /// Append `n` bytes, which [`Self::wants`] no more frames for, as + /// [`Self::place`] hands them out, and count them held. + pub fn write(&mut self, n: usize, copy: impl FnMut(&mut F, Range, usize)) { + assert_eq!(self.wants(n), 0, "pipe: a write past the frames it was given"); + self.place(0, n, copy); + self.len += n; + } + + /// The `len` held bytes `at` past the oldest, handed to `copy` as + /// [`Self::place`] hands them out; [`Self::consume`] lets them go. + pub fn peek(&self, at: usize, len: usize, mut copy: impl FnMut(&F, Range, usize)) { + assert!(at + len <= self.len, "pipe: {len} bytes at {at} of {} held", self.len); + let (mut done, mut pos) = (0, self.head + at); + while done < len { + let (frame, off) = (pos / FRAME, pos % FRAME); + let run = (FRAME - off).min(len - done); + copy(&self.frames[frame], off..off + run, done); + (done, pos) = (done + run, pos + run); } + } + + /// Let the `n` oldest bytes go; a frame it empties becomes a spare. + pub fn consume(&mut self, n: usize) { + assert!(n <= self.len, "pipe: {n} bytes let go of {} held", self.len); + let emptied = (self.head + n) / FRAME; + self.frames.rotate_left(emptied); + (self.head, self.len) = ((self.head + n) % FRAME, self.len - n); if self.len == 0 { self.head = 0; } + } + + /// Take up to `want` bytes from the front, handed to `copy` as + /// [`Self::peek`] hands them out, then let them go. Answers how many. + pub fn read(&mut self, want: usize, copy: impl FnMut(&F, Range, usize)) -> usize { + let n = want.min(self.len); + self.peek(0, n, copy); + self.consume(n); n } @@ -681,6 +711,57 @@ mod tests { assert_eq!(pipe.bytes.frames(), 1, "a drained pipe keeps one frame"); } + /// A write placed in pieces, in any order and cut anywhere, as a user + /// window's pages cut it, and a read peeked the same way, carry the same + /// bytes as one copy each would. + #[test] + fn pieces_placed_and_peeked_in_any_order_are_one_copy() { + let mut seed = 0x51ed_270b_u64; + let mut random = move |below: usize| { + seed ^= seed << 13; + seed ^= seed >> 7; + seed ^= seed << 17; + (seed % below as u64) as usize + }; + let mut pipe = Held::new(); + let (mut written, mut read) = (0, 0); + for _ in 0..300 { + let n = random(3 * FRAME + 1).min(pipe.bytes.space()); + for _ in 0..pipe.bytes.wants(n) { + pipe.bytes.add(Frame { id: pipe.taken, bytes: vec![0xEE; FRAME] }); + pipe.taken += 1; + } + let src = pattern(written, n); + let mut cuts: Vec = (0..random(6)).map(|_| random(n + 1)).chain([0, n]).collect(); + cuts.sort_unstable(); + let mut pieces: Vec<(usize, usize)> = cuts.windows(2).map(|w| (w[0], w[1] - w[0])).collect(); + pieces.reverse(); + for (at, len) in pieces { + pipe.bytes.place(at, len, |frame, within, from| { + frame.bytes[within.clone()].copy_from_slice(&src[at + from..at + from + within.len()]); + }); + } + pipe.bytes.commit(n); + written += n; + + let m = random(pipe.bytes.len() + 1); + let mut out = vec![0; m]; + let mut at = 0; + while at < m { + let len = (1 + random(2 * FRAME)).min(m - at); + pipe.bytes.peek(at, len, |frame, within, from| { + out[at + from..at + from + within.len()].copy_from_slice(&frame.bytes[within]); + }); + at += len; + } + pipe.bytes.consume(m); + pipe.returned.extend(pipe.bytes.surplus().map(|frame| frame.id)); + assert_eq!(out, pattern(read, m), "bytes {read}.. came back wrong"); + read += m; + pipe.accounted(); + } + } + /// A full pipe whose first frame is part-read holds `MOST_FRAMES`. #[test] fn a_full_pipe_part_read_holds_the_most_frames() { @@ -714,7 +795,7 @@ mod tests { } /// A pipe that held much and is drained gives back all but one frame, - /// and `clear` gives back every one. + /// which carries the next whole write, and `clear` gives back every one. #[test] fn a_drained_pipe_gives_its_frames_back() { let mut pipe = Held::new(); @@ -722,8 +803,8 @@ mod tests { assert_eq!(pipe.bytes.frames(), PIPE_CAPACITY / FRAME); pipe.read(PIPE_CAPACITY); assert_eq!((pipe.bytes.frames(), pipe.returned.len()), (1, PIPE_CAPACITY / FRAME - 1)); - pipe.write(&[1; 3]); - assert_eq!(pipe.taken, PIPE_CAPACITY / FRAME, "the kept frame did not carry the next write"); + pipe.write(&[1; MAX_WHOLE_PIPE_WRITE]); + assert_eq!(pipe.taken, PIPE_CAPACITY / FRAME, "the kept frame did not carry the next whole write"); assert_eq!(pipe.bytes.clear().count(), 1); assert_eq!((pipe.bytes.len(), pipe.bytes.frames()), (0, 0)); } diff --git a/kernel/src/mm/pmm.rs b/kernel/src/mm/pmm.rs index c7cde998b39..8783b10b153 100644 --- a/kernel/src/mm/pmm.rs +++ b/kernel/src/mm/pmm.rs @@ -330,7 +330,8 @@ pub fn alloc_held(job: &Job, count: usize, each: impl FnMut(u64)) -> bool { } /// Give back `frames` [`alloc_table_frame`] or [`alloc_held`] took for -/// `job`, under one hold of the allocator's lock, which none takes. +/// `job`, under one hold of the allocator's lock; giving back none takes no +/// lock. /// /// # Safety /// Nothing reaches any of them any more: no table links it, no CPU can still diff --git a/kernel/src/pipe.rs b/kernel/src/pipe.rs index 15ca41d92b9..3ed5b066112 100644 --- a/kernel/src/pipe.rs +++ b/kernel/src/pipe.rs @@ -229,12 +229,20 @@ impl PipeRef { }; if available > 0 && limit > 0 { let len = buf.len().min(limit); - let n = pipe.held.bytes.read(len, |frame, within, at| { - let n = within.len(); - // SAFETY: bytes of a frame this pipe holds, which only its - // lock reaches and nothing writes while it is held. - buf.write_at(at, unsafe { core::slice::from_raw_parts(frame_bytes(frame, within), n) }); + let n = len.min(available); + let bytes = &mut pipe.held.bytes; + // One walk of the user window, each pinned piece copied out of + // the frames it covers. + buf.write_pieces(0, n, |to, at, piece| { + bytes.peek(at, piece, |frame, within, from| { + // SAFETY: `to` is valid for `piece` bytes until this + // walk's callback returns, and `from + within.len()` is + // at most `piece`; the frame is this pipe's alone, + // reached only under its lock, and no process maps it. + unsafe { core::ptr::copy_nonoverlapping(frame_bytes(frame, within.clone()), to.add(from), within.len()) } + }); }); + bytes.consume(n); pipe.held.give_back_surplus(); pipe.read += n as u64; let boost = core::mem::take(&mut pipe.rt_boost_pending); @@ -267,12 +275,18 @@ impl PipeRef { if pipe.held.take_frames(n).is_err() { return PipeWrite::NoMemory; } - pipe.held.bytes.write(n, |frame, within, at| { - let n = within.len(); - // SAFETY: bytes of a frame this pipe holds, which only its lock - // reaches and nothing reads while it is held. - buf.read_at(at, unsafe { core::slice::from_raw_parts_mut(frame_bytes(frame, within), n) }); + let bytes = &mut pipe.held.bytes; + // One walk of the user window, each pinned piece copied into the + // frames it covers. + buf.read_pieces(0, n, |from, at, piece| { + bytes.place(at, piece, |frame, within, to| { + // SAFETY: `from` is valid for `piece` bytes until this walk's + // callback returns, and `to + within.len()` is at most `piece`; + // the frame is this pipe's alone, as above. + unsafe { core::ptr::copy_nonoverlapping(from.add(to), frame_bytes(frame, within.clone()), within.len()) } + }); }); + bytes.commit(n); pipe.written += n as u64; PipeWrite::Wrote { n, end: pipe.written } } diff --git a/kernel/src/user_ptr.rs b/kernel/src/user_ptr.rs index b0a97fcd342..ffdcb802c9b 100644 --- a/kernel/src/user_ptr.rs +++ b/kernel/src/user_ptr.rs @@ -219,6 +219,17 @@ impl UserBytes<'_> { self.0.read_into(off, dst); } + /// Hand `copy` each piece of the window's `[off, off + len)`, as + /// [`read_at`](Self::read_at) reaches them, one chunk of pages pinned at + /// a time: a pointer to its bytes, its offset in that range and its + /// length. The pointer is valid for that length until `copy` returns, + /// with preemption off, and is never made a reference; a page the + /// process took away reads as zeros. Panics if out of bounds, as + /// [`read_at`](Self::read_at) does. + pub fn read_pieces(&self, off: usize, len: usize, mut copy: impl FnMut(*const u8, usize, usize)) { + self.0.pieces("UserBytes::read_pieces", off, len, |run, at, n| copy(run.cast_const(), at, n)); + } + /// The `len`-byte window at `off` inside this one. pub fn sub(&self, off: usize, len: usize) -> UserBytes<'_> { UserBytes(self.0.sub("UserBytes::sub", off, len), PhantomData) @@ -242,6 +253,13 @@ impl UserBytesMut<'_> { self.0.write_from(off, src); } + /// [`UserBytes::read_pieces`] for a copy into the window: each pointer + /// may be written for its length until `copy` returns, and a page the + /// process took away drops what is written to it. + pub fn write_pieces(&mut self, off: usize, len: usize, copy: impl FnMut(*mut u8, usize, usize)) { + self.0.pieces("UserBytesMut::write_pieces", off, len, copy); + } + /// Zero `len` bytes of the window at `off`. pub fn fill_zero(&mut self, off: usize, len: usize) { self.0.pieces("UserBytesMut::fill_zero", off, len, |run, _, n| { From 4d80802e011546ea3c1aeb52e7b94a0eca2fa790 Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 28 Oct 2026 07:00:13 +0100 Subject: [PATCH 04/12] abuse_connect_flood: a connection's first byte buys a frame, not a 2 MiB ring The test's negative control still asserted the 2 MiB ring a pipe took on its first write. A pipe now takes a frame for its bytes; the control keeps its meaning, that a byte on a live connection buys memory. Co-Authored-By: Claude Opus 5.5 --- tests/guest/rust-tests/src/bin/abuse_connect_flood.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/tests/guest/rust-tests/src/bin/abuse_connect_flood.rs b/tests/guest/rust-tests/src/bin/abuse_connect_flood.rs index c14b23dd3e8..166415533fe 100644 --- a/tests/guest/rust-tests/src/bin/abuse_connect_flood.rs +++ b/tests/guest/rust-tests/src/bin/abuse_connect_flood.rs @@ -12,8 +12,8 @@ //! The other half, which the depth alone does not give: each of those calls //! built two 2 MiB rings *eagerly*, so the allowance the kernel grants a //! server that has not answered yet cost 128 MiB of physical memory before a -//! single byte had been sent. A pipe now allocates its page on first use, and -//! the memory assertion below is what says so. +//! single byte had been sent. A pipe now takes memory only as it holds bytes, +//! and the memory assertion below is what says so. //! //! **The attack's "be your own service" clause is dead and this is now a bound //! check.** It used to `SYS_LISTEN` on a name of its own, because listening was @@ -98,7 +98,7 @@ fn main() { let accepted = syscall::accept(acceptor.as_handle().raw()).expect("accept a queued connection"); // The other direction, without which "costs nothing" could be a pipe that - // never works: one byte on a live connection has to buy a ring. This is + // never works: one byte on a live connection has to buy a frame. This is // the negative control for the ceiling above — a kernel that allocated // nothing ever would pass that assertion and fail this one. let before_write = used_bytes(); @@ -106,8 +106,8 @@ fn main() { assert_eq!(wrote, 1); let ring = used_bytes().saturating_sub(before_write); assert!( - ring >= 2 * 1024 * 1024, - "the first write to a connection grew physical memory by {ring} bytes; a ring is 2 MiB" + ring >= syscall::PAGE_SIZE as u64, + "the first write to a connection grew physical memory by {ring} bytes; a pipe's bytes take a frame" ); println!(" the first write on a connection allocated {} KiB", ring / 1024); From 77bd7bef6e8f03c6e55b7019e037ea186376dbdc Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 28 Oct 2026 07:59:10 +0100 Subject: [PATCH 05/12] pipe: a drained pipe keeps the frames its last write spanned Session 3 read pipe.throughput at 5e0cb52b0 against 544f9cc3c, interleaved x2: 65536 5954/5975 against 6940/6844 MB/s (-13%), 16384 3683/3680 against 4456/4360 (-16%), 4096 2651/2702 against 2833/2742 (-3%). The size pattern names the spare rule: a drained pipe kept one frame, which a write of a page or less fits, while a 16 or 64 KiB write into a pipe its reader keeps up with took 4 or 16 frames from the allocator and the drain gave all but one back, each under the allocator's lock. A drained pipe now keeps as many spares as its last write spanned, beside those its bytes need, so a writer of one size reuses its frames whether its reader lags or keeps up, and one that moves to small writes keeps one again. The steady-stream test now covers the reader that drains the pipe each round, which the old rule failed (1516 frames taken in 100 rounds). Co-Authored-By: Claude Opus 5.5 --- kernel/pure/pipe.rs | 41 ++++++++++++++++++++++++----------------- 1 file changed, 24 insertions(+), 17 deletions(-) diff --git a/kernel/pure/pipe.rs b/kernel/pure/pipe.rs index 411c22e5c10..14859242bcd 100644 --- a/kernel/pure/pipe.rs +++ b/kernel/pure/pipe.rs @@ -145,19 +145,22 @@ impl Landing { /// What a pipe holds, in frames taken as it fills and given back as it /// drains. Its first unread byte is `head` bytes into the first frame, and /// the frames past its last byte are spare: kept for the next write while -/// there are no more of them than frames holding bytes, and one besides. So a -/// pipe holds at most twice the frames its bytes need and one more, never more -/// than [`MOST_FRAMES`], and a stream that keeps a backlog in it takes no -/// frame from the allocator. +/// there are no more of them than frames holding bytes, or than its last +/// write spanned, or one. So a pipe holds at most the frames its bytes need +/// and as many again or its last write's, never more than [`MOST_FRAMES`], +/// and a stream of writes of one size takes no frame from the allocator +/// whether its reader keeps up or lags. pub struct Bytes { frames: VecDeque, head: usize, len: usize, + /// The frames the last write spanned: the working set its next one wants. + last: usize, } impl Bytes { pub const fn new() -> Self { - Self { frames: VecDeque::new(), head: 0, len: 0 } + Self { frames: VecDeque::new(), head: 0, len: 0, last: 0 } } pub fn len(&self) -> usize { @@ -226,6 +229,7 @@ impl Bytes { /// Count `n` more bytes held: what [`Self::place`] put at the end. pub fn commit(&mut self, n: usize) { assert_eq!(self.wants(n), 0, "pipe: bytes held past the frames it was given"); + self.last = (self.head + self.len + n).div_ceil(FRAME) - (self.head + self.len) / FRAME; self.len += n; } @@ -234,7 +238,7 @@ impl Bytes { pub fn write(&mut self, n: usize, copy: impl FnMut(&mut F, Range, usize)) { assert_eq!(self.wants(n), 0, "pipe: a write past the frames it was given"); self.place(0, n, copy); - self.len += n; + self.commit(n); } /// The `len` held bytes `at` past the oldest, handed to `copy` as @@ -273,13 +277,13 @@ impl Bytes { /// The spares past those it keeps, for the caller to give back. pub fn surplus(&mut self) -> Drain<'_, F> { let used = self.used(); - let keep = used + (self.frames.len() - used).min(used.max(1)); + let keep = used + (self.frames.len() - used).min(used.max(self.last).max(1)); self.frames.drain(keep..) } /// Every frame, for the caller to give back, and with them every byte. pub fn clear(&mut self) -> Drain<'_, F> { - (self.head, self.len) = (0, 0); + (self.head, self.len, self.last) = (0, 0, 0); self.frames.drain(..) } } @@ -699,7 +703,7 @@ mod tests { assert_eq!(got, pattern(read, got.len()), "turn {turn}: bytes {read}.. came back wrong"); read += got.len(); assert!( - pipe.bytes.frames() <= 2 * pipe.bytes.used() + 1, + pipe.bytes.frames() <= pipe.bytes.used() + pipe.bytes.used().max(pipe.bytes.last).max(1), "turn {turn}: {} frames kept for {} bytes", pipe.bytes.frames(), pipe.bytes.len() @@ -708,7 +712,7 @@ mod tests { } let rest = pipe.read(PIPE_CAPACITY); assert_eq!(rest, pattern(read, written - read)); - assert_eq!(pipe.bytes.frames(), 1, "a drained pipe keeps one frame"); + assert_eq!(pipe.bytes.frames(), pipe.bytes.last.max(1), "a drained pipe keeps its last write's frames"); } /// A write placed in pieces, in any order and cut anywhere, as a user @@ -774,13 +778,13 @@ mod tests { assert_eq!(pipe.read(PIPE_CAPACITY), [pattern(5, 5), pattern(10, rest)].concat()); } - /// A stream that keeps a backlog in the pipe, a little or nearly all of - /// it, takes no frame after its first round: the frames a read empties - /// carry the next write. + /// A stream of one write size takes no frame after its first round, + /// whether its reader drains the pipe each time, keeps a little backlog + /// or nearly all of it: the frames a read empties carry the next write. #[test] fn a_steady_stream_takes_no_frames() { const CHUNK: usize = 64 << 10; - for backlog in [CHUNK, PIPE_CAPACITY - CHUNK] { + for backlog in [0, CHUNK, PIPE_CAPACITY - CHUNK] { let mut pipe = Held::new(); pipe.write(&vec![1; backlog]); pipe.write(&vec![1; CHUNK]); @@ -794,12 +798,15 @@ mod tests { } } - /// A pipe that held much and is drained gives back all but one frame, - /// which carries the next whole write, and `clear` gives back every one. + /// A pipe that held much and is drained keeps what its last write + /// spanned and gives back the rest, so after a small write and its read + /// it keeps one frame, which carries the next whole write; `clear` gives + /// back every one. #[test] fn a_drained_pipe_gives_its_frames_back() { let mut pipe = Held::new(); - pipe.write(&vec![1; PIPE_CAPACITY]); + pipe.write(&vec![1; PIPE_CAPACITY - 1]); + pipe.write(&[1]); assert_eq!(pipe.bytes.frames(), PIPE_CAPACITY / FRAME); pipe.read(PIPE_CAPACITY); assert_eq!((pipe.bytes.frames(), pipe.returned.len()), (1, PIPE_CAPACITY / FRAME - 1)); From 024df10e6815598277dadd4249da327f16961a05 Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 28 Oct 2026 08:37:58 +0100 Subject: [PATCH 06/12] issues: two holders of one socket queue can wait for good, at every revision the soak ran Co-Authored-By: Claude Opus 5.5 --- ...s-of-one-socket-queue-can-wait-for-good.md | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) create mode 100644 issues/two-holders-of-one-socket-queue-can-wait-for-good.md diff --git a/issues/two-holders-of-one-socket-queue-can-wait-for-good.md b/issues/two-holders-of-one-socket-queue-can-wait-for-good.md new file mode 100644 index 00000000000..6f88b4cc07a --- /dev/null +++ b/issues/two-holders-of-one-socket-queue-can-wait-for-good.md @@ -0,0 +1,41 @@ +--- +status: open +kind: defect +opened: 2026-10-28 +--- + +# Two holders of one socket queue can wait for good + +`netstack_holders_exported_to_children` (`tests/toyos.rs`), on QEMU x86-64's +`tests/images/netcase.toml`, sometimes stops in `test_rs_socket_export`'s +race. The job has said its line for the export ("a listener and a datagram +socket exported to a child: it accepted and spoke on them, and so did the +parent after it") and spawned the racer, so the parent and the racer are +blocked on one listener's queue and one datagram socket's. The guest then +says nothing until the harness's ceiling ends it, 690 s to 915 s later. No +kernel `exit:` line follows for either process, so this is lost progress, +not a slow guest. + +The soak ran the seven-test batch +`kernel_objects_refused_alone memory_floor_holds_the_controls stream_holders +netstack_holders_exported_to_children session_lifecycle libc_children +udp_receive_times_out` repeatedly. Three detached worktrees ran it at once, +under the same host load, on 2026-10-28: + +| revision | runs | this stall | +|---|---|---| +| `544f9cc3c` | 16 | 1 | +| `576bebf3a` (main) | 22 | 2 | +| `5e0cb52b0` (#1074) | 21 | 4 | + +It stalls at every revision, so it predates #1074. These counts do not +separate the rates. + +The two holders are woken by netstack ringing each holder's bell, a pipe +write it does not wait on. Where a wake is lost is not known. + +Owner: the orchestrator, who assigns it. + +Exit: the race's lost wake is found and fixed, shown by a host or loom +model of two holders on one queue. The same soak then runs at least 40 +batches without this stall. From e6e90873413c27cc80365a690f9f1033122ce647 Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 28 Oct 2026 09:24:08 +0100 Subject: [PATCH 07/12] issues: the exported-socket race's stall, soaked at three revisions, in its one record main already recorded this stall as two-holders-racing-on-an-exported- socket-can-both-stall.md. The soak's counts, signature and logs go there, and the second record this branch added goes. Co-Authored-By: Claude Opus 5.5 --- ...s-of-one-socket-queue-can-wait-for-good.md | 41 ------------------- 1 file changed, 41 deletions(-) delete mode 100644 issues/two-holders-of-one-socket-queue-can-wait-for-good.md diff --git a/issues/two-holders-of-one-socket-queue-can-wait-for-good.md b/issues/two-holders-of-one-socket-queue-can-wait-for-good.md deleted file mode 100644 index 6f88b4cc07a..00000000000 --- a/issues/two-holders-of-one-socket-queue-can-wait-for-good.md +++ /dev/null @@ -1,41 +0,0 @@ ---- -status: open -kind: defect -opened: 2026-10-28 ---- - -# Two holders of one socket queue can wait for good - -`netstack_holders_exported_to_children` (`tests/toyos.rs`), on QEMU x86-64's -`tests/images/netcase.toml`, sometimes stops in `test_rs_socket_export`'s -race. The job has said its line for the export ("a listener and a datagram -socket exported to a child: it accepted and spoke on them, and so did the -parent after it") and spawned the racer, so the parent and the racer are -blocked on one listener's queue and one datagram socket's. The guest then -says nothing until the harness's ceiling ends it, 690 s to 915 s later. No -kernel `exit:` line follows for either process, so this is lost progress, -not a slow guest. - -The soak ran the seven-test batch -`kernel_objects_refused_alone memory_floor_holds_the_controls stream_holders -netstack_holders_exported_to_children session_lifecycle libc_children -udp_receive_times_out` repeatedly. Three detached worktrees ran it at once, -under the same host load, on 2026-10-28: - -| revision | runs | this stall | -|---|---|---| -| `544f9cc3c` | 16 | 1 | -| `576bebf3a` (main) | 22 | 2 | -| `5e0cb52b0` (#1074) | 21 | 4 | - -It stalls at every revision, so it predates #1074. These counts do not -separate the rates. - -The two holders are woken by netstack ringing each holder's bell, a pipe -write it does not wait on. Where a wake is lost is not known. - -Owner: the orchestrator, who assigns it. - -Exit: the race's lost wake is found and fixed, shown by a host or loom -model of two holders on one queue. The same soak then runs at least 40 -batches without this stall. From ff368dfe71a6f8fdfdaaf7ef15bce171144389e0 Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 28 Oct 2026 09:47:16 +0100 Subject: [PATCH 08/12] pipe: a copy over physically adjacent frames is one copy The T14 diagnostic session (measurement-only counters, base 544f9cc3c against cd42d13a4, eight rounds each) put the whole of a 64 KiB pipe write's extra time in the user walk's copy phase: 153.8 against 123.5 ticks per KiB moved. Its page reach and pin, the lock's spin and hold outside the copy, and frame take and give were equal or under a tick per KiB. The head made 32 memcpys per 64 KiB write against base's 17, since every user page's piece was cut at a 4 KiB frame edge. The pipe now gathers the runs of each user piece and copies runs that continue each other in both source and destination as one. Frames a write takes come off the allocator in ascending order, so a piece over two of them is one copy when they are adjacent, as base's ring always was. Co-Authored-By: Claude Opus 5.5 --- kernel/src/pipe.rs | 52 ++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 50 insertions(+), 2 deletions(-) diff --git a/kernel/src/pipe.rs b/kernel/src/pipe.rs index 3ed5b066112..1c3ab5b1c12 100644 --- a/kernel/src/pipe.rs +++ b/kernel/src/pipe.rs @@ -127,6 +127,48 @@ impl Drop for Held { } } +/// One copy gathered from runs that continue each other in both source and +/// destination, so a user page's piece over two physically adjacent frames +/// is one copy, not two. +struct Copy { + from: *const u8, + to: *mut u8, + len: usize, +} + +impl Copy { + const NONE: Self = Self { from: core::ptr::null(), to: core::ptr::null_mut(), len: 0 }; + + /// Add `len` bytes from `from` to `to`, copying what was gathered first + /// unless these continue it. + /// + /// # Safety + /// `from` is readable and `to` writable for `len` bytes, and stay so + /// until the next [`Self::flush`]; the two never overlap. + unsafe fn add(&mut self, from: *const u8, to: *mut u8, len: usize) { + if self.len > 0 && self.from.wrapping_add(self.len) == from && self.to.wrapping_add(self.len) == to { + self.len += len; + return; + } + // SAFETY: what was gathered meets this function's contract. + unsafe { self.flush() }; + *self = Self { from, to, len }; + } + + /// Copy what was gathered. + /// + /// # Safety + /// As [`Self::add`] for every run gathered. + unsafe fn flush(&mut self) { + if self.len > 0 { + // SAFETY: the runs gathered are contiguous on both sides and + // each met `add`'s contract. + unsafe { core::ptr::copy_nonoverlapping(self.from, self.to, self.len) }; + } + *self = Self::NONE; + } +} + /// `within` of a frame a pipe holds, which no process maps and only that /// pipe's lock reaches. fn frame_bytes(frame: &u64, within: Range) -> *mut u8 { @@ -234,13 +276,16 @@ impl PipeRef { // One walk of the user window, each pinned piece copied out of // the frames it covers. buf.write_pieces(0, n, |to, at, piece| { + let mut copy = Copy::NONE; bytes.peek(at, piece, |frame, within, from| { // SAFETY: `to` is valid for `piece` bytes until this // walk's callback returns, and `from + within.len()` is // at most `piece`; the frame is this pipe's alone, // reached only under its lock, and no process maps it. - unsafe { core::ptr::copy_nonoverlapping(frame_bytes(frame, within.clone()), to.add(from), within.len()) } + unsafe { copy.add(frame_bytes(frame, within.clone()), to.add(from), within.len()) } }); + // SAFETY: as above, before the callback returns. + unsafe { copy.flush() }; }); bytes.consume(n); pipe.held.give_back_surplus(); @@ -279,12 +324,15 @@ impl PipeRef { // One walk of the user window, each pinned piece copied into the // frames it covers. buf.read_pieces(0, n, |from, at, piece| { + let mut copy = Copy::NONE; bytes.place(at, piece, |frame, within, to| { // SAFETY: `from` is valid for `piece` bytes until this walk's // callback returns, and `to + within.len()` is at most `piece`; // the frame is this pipe's alone, as above. - unsafe { core::ptr::copy_nonoverlapping(from.add(to), frame_bytes(frame, within.clone()), within.len()) } + unsafe { copy.add(from.add(to), frame_bytes(frame, within.clone()), within.len()) } }); + // SAFETY: as above, before the callback returns. + unsafe { copy.flush() }; }); bytes.commit(n); pipe.written += n as u64; From 878a0629e718002b3386ec4b350bbf2b873f721c Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 28 Oct 2026 10:06:34 +0100 Subject: [PATCH 09/12] review r3: the queue's test helper takes the kernel's path, and names say what they are B4: Bytes::write and Bytes::read had no caller but the test helper since the kernel moved to place/commit and peek/consume; they go, and the helper writes and reads as kernel::pipe does. B5: the lock-scope record names the walk the pipe makes, write_pieces and read_pieces with peek and place inside, and when a read gives frames back. S6: the copy gatherer is Gather, not a name that shadows the Copy trait. S7: the frame-run closures' offset into the piece is `off`. S8: the spare rule is said as the largest of three. S10: the connect flood's measure is `took`. Co-Authored-By: Claude Opus 5.5 --- ...ns-a-user-copy-and-its-first-allocation.md | 13 +++--- kernel/pure/pipe.rs | 45 +++++++------------ kernel/src/pipe.rs | 24 +++++----- .../rust-tests/src/bin/abuse_connect_flood.rs | 8 ++-- 4 files changed, 39 insertions(+), 51 deletions(-) diff --git a/issues/a-pipes-lock-spans-a-user-copy-and-its-first-allocation.md b/issues/a-pipes-lock-spans-a-user-copy-and-its-first-allocation.md index 4964972d4a2..cf022d69616 100644 --- a/issues/a-pipes-lock-spans-a-user-copy-and-its-first-allocation.md +++ b/issues/a-pipes-lock-spans-a-user-copy-and-its-first-allocation.md @@ -10,10 +10,11 @@ Each pipe owns its state behind its own `sync::Lock` (`kernel/src/pipe.rs`), which disables preemption before it spins (`kernel/src/sync.rs`). The bulk copy runs inside that lock: -* `PipeRef::try_read_within` runs `bytes.read(len, ..)`, each frame's run - copied with `buf.write_at`, under it; -* `PipeRef::write` runs `bytes.write(n, ..)`, each run copied with - `buf.read_at`, in the same acquisition. +* `PipeRef::try_read_within` makes one walk of the caller's window, + `buf.write_pieces`, and copies each pinned piece out of the frames it + covers with `bytes.peek`, under it; +* `PipeRef::write` makes one walk, `buf.read_pieces`, and copies each piece + into its frames with `bytes.place`, in the same acquisition. The copy is bounded only by the pipe: `PIPE_CAPACITY` is 2 MiB, the largest single copy under the lock, and nothing above caps it (`SYS_READ`/`SYS_WRITE` @@ -23,8 +24,8 @@ zeroing. A write that grows the pipe also takes its frames under the same lock, nesting the frame allocator's lock: up to 513 frames, unzeroed, under one -hold, and the heap the list of them grows into. A read gives back the frames -it empties past those kept, under the same nesting. +hold, and the heap the list of them grows into. A read that empties frames +past the spares kept gives them back under the same nesting. What queues behind it is that one pipe's traffic: its other end's reads and writes, `has_data`/`has_space` from its blocking wrappers and polls, and its diff --git a/kernel/pure/pipe.rs b/kernel/pure/pipe.rs index 14859242bcd..4d4a72a2ef8 100644 --- a/kernel/pure/pipe.rs +++ b/kernel/pure/pipe.rs @@ -144,12 +144,12 @@ impl Landing { /// What a pipe holds, in frames taken as it fills and given back as it /// drains. Its first unread byte is `head` bytes into the first frame, and -/// the frames past its last byte are spare: kept for the next write while -/// there are no more of them than frames holding bytes, or than its last -/// write spanned, or one. So a pipe holds at most the frames its bytes need -/// and as many again or its last write's, never more than [`MOST_FRAMES`], -/// and a stream of writes of one size takes no frame from the allocator -/// whether its reader keeps up or lags. +/// the frames past its last byte are spare, kept for the next write up to +/// the largest of: the frames holding bytes, the frames the last write +/// spanned, and one. So a pipe holds at most its bytes' frames plus that +/// largest, never more than [`MOST_FRAMES`], and a stream of writes of one +/// size takes no frame from the allocator whether its reader keeps up or +/// lags. pub struct Bytes { frames: VecDeque, head: usize, @@ -233,14 +233,6 @@ impl Bytes { self.len += n; } - /// Append `n` bytes, which [`Self::wants`] no more frames for, as - /// [`Self::place`] hands them out, and count them held. - pub fn write(&mut self, n: usize, copy: impl FnMut(&mut F, Range, usize)) { - assert_eq!(self.wants(n), 0, "pipe: a write past the frames it was given"); - self.place(0, n, copy); - self.commit(n); - } - /// The `len` held bytes `at` past the oldest, handed to `copy` as /// [`Self::place`] hands them out; [`Self::consume`] lets them go. pub fn peek(&self, at: usize, len: usize, mut copy: impl FnMut(&F, Range, usize)) { @@ -265,15 +257,6 @@ impl Bytes { } } - /// Take up to `want` bytes from the front, handed to `copy` as - /// [`Self::peek`] hands them out, then let them go. Answers how many. - pub fn read(&mut self, want: usize, copy: impl FnMut(&F, Range, usize)) -> usize { - let n = want.min(self.len); - self.peek(0, n, copy); - self.consume(n); - n - } - /// The spares past those it keeps, for the caller to give back. pub fn surplus(&mut self) -> Drain<'_, F> { let used = self.used(); @@ -644,24 +627,28 @@ mod tests { Self { bytes: Bytes::new(), taken: 0, returned: Vec::new() } } - /// Write `src` as `kernel::pipe` does: the frames it wants, then the - /// copy. + /// Write `src` as `kernel::pipe` does: the frames it wants, then + /// the copy placed and committed. fn write(&mut self, src: &[u8]) { for _ in 0..self.bytes.wants(src.len()) { self.bytes.add(Frame { id: self.taken, bytes: vec![0xEE; FRAME] }); self.taken += 1; } - self.bytes.write(src.len(), |frame, within, at| { + self.bytes.place(0, src.len(), |frame, within, at| { frame.bytes[within.clone()].copy_from_slice(&src[at..at + within.len()]); }); + self.bytes.commit(src.len()); } + /// Read up to `want` bytes as `kernel::pipe` does: peeked, consumed, + /// and the surplus given back. fn read(&mut self, want: usize) -> Vec { - let mut out = vec![0; want]; - let n = self.bytes.read(want, |frame, within, at| { + let n = want.min(self.bytes.len()); + let mut out = vec![0; n]; + self.bytes.peek(0, n, |frame, within, at| { out[at..at + within.len()].copy_from_slice(&frame.bytes[within]); }); - out.truncate(n); + self.bytes.consume(n); self.returned.extend(self.bytes.surplus().map(|frame| frame.id)); out } diff --git a/kernel/src/pipe.rs b/kernel/src/pipe.rs index 1c3ab5b1c12..b49e33f6326 100644 --- a/kernel/src/pipe.rs +++ b/kernel/src/pipe.rs @@ -130,13 +130,13 @@ impl Drop for Held { /// One copy gathered from runs that continue each other in both source and /// destination, so a user page's piece over two physically adjacent frames /// is one copy, not two. -struct Copy { +struct Gather { from: *const u8, to: *mut u8, len: usize, } -impl Copy { +impl Gather { const NONE: Self = Self { from: core::ptr::null(), to: core::ptr::null_mut(), len: 0 }; /// Add `len` bytes from `from` to `to`, copying what was gathered first @@ -276,16 +276,16 @@ impl PipeRef { // One walk of the user window, each pinned piece copied out of // the frames it covers. buf.write_pieces(0, n, |to, at, piece| { - let mut copy = Copy::NONE; - bytes.peek(at, piece, |frame, within, from| { + let mut gather = Gather::NONE; + bytes.peek(at, piece, |frame, within, off| { // SAFETY: `to` is valid for `piece` bytes until this - // walk's callback returns, and `from + within.len()` is + // walk's callback returns, and `off + within.len()` is // at most `piece`; the frame is this pipe's alone, // reached only under its lock, and no process maps it. - unsafe { copy.add(frame_bytes(frame, within.clone()), to.add(from), within.len()) } + unsafe { gather.add(frame_bytes(frame, within.clone()), to.add(off), within.len()) } }); // SAFETY: as above, before the callback returns. - unsafe { copy.flush() }; + unsafe { gather.flush() }; }); bytes.consume(n); pipe.held.give_back_surplus(); @@ -324,15 +324,15 @@ impl PipeRef { // One walk of the user window, each pinned piece copied into the // frames it covers. buf.read_pieces(0, n, |from, at, piece| { - let mut copy = Copy::NONE; - bytes.place(at, piece, |frame, within, to| { + let mut gather = Gather::NONE; + bytes.place(at, piece, |frame, within, off| { // SAFETY: `from` is valid for `piece` bytes until this walk's - // callback returns, and `to + within.len()` is at most `piece`; + // callback returns, and `off + within.len()` is at most `piece`; // the frame is this pipe's alone, as above. - unsafe { copy.add(from.add(to), frame_bytes(frame, within.clone()), within.len()) } + unsafe { gather.add(from.add(off), frame_bytes(frame, within.clone()), within.len()) } }); // SAFETY: as above, before the callback returns. - unsafe { copy.flush() }; + unsafe { gather.flush() }; }); bytes.commit(n); pipe.written += n as u64; diff --git a/tests/guest/rust-tests/src/bin/abuse_connect_flood.rs b/tests/guest/rust-tests/src/bin/abuse_connect_flood.rs index 166415533fe..8aad797f2c6 100644 --- a/tests/guest/rust-tests/src/bin/abuse_connect_flood.rs +++ b/tests/guest/rust-tests/src/bin/abuse_connect_flood.rs @@ -104,12 +104,12 @@ fn main() { let before_write = used_bytes(); let wrote = syscall::write(accepted, b"x").expect("write one byte to an accepted socket"); assert_eq!(wrote, 1); - let ring = used_bytes().saturating_sub(before_write); + let took = used_bytes().saturating_sub(before_write); assert!( - ring >= syscall::PAGE_SIZE as u64, - "the first write to a connection grew physical memory by {ring} bytes; a pipe's bytes take a frame" + took >= syscall::PAGE_SIZE as u64, + "the first write to a connection grew physical memory by {took} bytes; a pipe's bytes take a frame" ); - println!(" the first write on a connection allocated {} KiB", ring / 1024); + println!(" the first write on a connection allocated {} KiB", took / 1024); // SAFETY: no type holds this number. unsafe { syscall::close(accepted) }; From 0ae412ae18ecd504c99dd8ceba170314b8ee94ba Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 28 Oct 2026 16:43:15 +0100 Subject: [PATCH 10/12] pipe: a piece's copy is inlined, its refusal cold The T14's PMU session (measurement-only counters, Intel SDM Vol. 3B section 20.2 and Intel's Tiger Lake event table) put most of a pipe write's remaining cost in instructions: 67.9 against base's 24.1 retired per KiB copied at 64 KiB and 77.9 against 31.5 at 16 KiB, about 44 more per KiB or 175 per 4 KiB piece. Base, not the head, stalled on a full store buffer, and L1D replacements and L2 misses were no higher. 4K-alias blocks were lower at 64 KiB but about 2 per KiB higher at 16 and 4 KiB (+17 % and +23 %), and may account for part of what remains at 16 KiB. The kernel before this change had Bytes::place and Bytes::peek out of line, one call per user-page piece; Gather and frame_bytes were inlined into them. place and peek are inlined now, and Gather and frame_bytes are marked to stay inlined in the larger walk. frame_bytes' bounds refusal goes to a cold, never-inlined helper, as a hot accessor's refusal belongs (kernel CLAUDE.md). The same change, measured as a patch on this branch's pre-rebase head, read pipe.throughput.16384 at 4262/4260 MB/s against base's 4377/4434 and 4096 at 2662/2673 against 2661/2772, from 7 % and 4 % below base before. The readings, the counters and both kernels' symbol listings: https://github.com/ToyOSOrg/evidence/tree/f9107a6656605a9cbedeae3ad98d38f1b4d9a9d0/r6 Co-Authored-By: Claude Opus 5.5 --- kernel/pure/pipe.rs | 2 ++ kernel/src/pipe.rs | 13 ++++++++++++- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/kernel/pure/pipe.rs b/kernel/pure/pipe.rs index 4d4a72a2ef8..d171d5b2491 100644 --- a/kernel/pure/pipe.rs +++ b/kernel/pure/pipe.rs @@ -214,6 +214,7 @@ impl Bytes { /// handed each run of them as its frame, where they go in it, and where /// they start among the `len`. They are held once [`Self::commit`] counts /// them. + #[inline(always)] pub fn place(&mut self, at: usize, len: usize, mut copy: impl FnMut(&mut F, Range, usize)) { let end = self.head + self.len + at + len; assert!(end <= self.frames.len() * FRAME, "pipe: bytes placed past the frames it was given"); @@ -235,6 +236,7 @@ impl Bytes { /// The `len` held bytes `at` past the oldest, handed to `copy` as /// [`Self::place`] hands them out; [`Self::consume`] lets them go. + #[inline(always)] pub fn peek(&self, at: usize, len: usize, mut copy: impl FnMut(&F, Range, usize)) { assert!(at + len <= self.len, "pipe: {len} bytes at {at} of {} held", self.len); let (mut done, mut pos) = (0, self.head + at); diff --git a/kernel/src/pipe.rs b/kernel/src/pipe.rs index b49e33f6326..b26a8ffaace 100644 --- a/kernel/src/pipe.rs +++ b/kernel/src/pipe.rs @@ -145,6 +145,7 @@ impl Gather { /// # Safety /// `from` is readable and `to` writable for `len` bytes, and stay so /// until the next [`Self::flush`]; the two never overlap. + #[inline(always)] unsafe fn add(&mut self, from: *const u8, to: *mut u8, len: usize) { if self.len > 0 && self.from.wrapping_add(self.len) == from && self.to.wrapping_add(self.len) == to { self.len += len; @@ -159,6 +160,7 @@ impl Gather { /// /// # Safety /// As [`Self::add`] for every run gathered. + #[inline(always)] unsafe fn flush(&mut self) { if self.len > 0 { // SAFETY: the runs gathered are contiguous on both sides and @@ -171,12 +173,21 @@ impl Gather { /// `within` of a frame a pipe holds, which no process maps and only that /// pipe's lock reaches. +#[inline(always)] fn frame_bytes(frame: &u64, within: Range) -> *mut u8 { - assert!(within.end <= kernel::pipe::FRAME, "pipe: bytes {within:?} past a frame"); + if within.end > kernel::pipe::FRAME { + past_a_frame(within); + } // SAFETY: inside the frame, which the direct map covers. unsafe { DirectMap::from_phys(*frame).as_mut_ptr::().add(within.start) } } +#[cold] +#[inline(never)] +fn past_a_frame(within: Range) -> ! { + panic!("pipe: bytes {within:?} past a frame") +} + struct Pipe { state: Lock, /// The read end's and the write end's watches. Held by `Arc` so a blocking site or a From a60d8e63bd5ef05877de2f8fe07354b78922e98d Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 28 Oct 2026 15:56:19 +0100 Subject: [PATCH 11/12] fileserver: PIPE_CAPACITY named with its other ABI imports Main's fileserver imports toyos_abi::syscall names individually, so the rebased stream bound names PIPE_CAPACITY there rather than through a syscall module path it no longer has. Co-Authored-By: Claude Opus 5.5 --- system/fileserver/src/main.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/system/fileserver/src/main.rs b/system/fileserver/src/main.rs index ac81b968af9..34ab3b009e0 100644 --- a/system/fileserver/src/main.rs +++ b/system/fileserver/src/main.rs @@ -73,7 +73,7 @@ use toyos::port::{self, Acceptor}; use toyos::shm::{Bytes, SharedMemory}; use toyos::Pipe; use toyos_abi::handle::Rights; -use toyos_abi::syscall::{SyscallError, MAX_BADGE, SERVE_PREFIX}; +use toyos_abi::syscall::{SyscallError, MAX_BADGE, PIPE_CAPACITY, SERVE_PREFIX}; /// What one turn of a stream appends at most. const STREAM_READ: usize = 64 * 1024; @@ -81,7 +81,7 @@ const STREAM_READ: usize = 64 * 1024; /// The turns that drain what one stream holds before a request on its /// description: the most a pipe holds, so a writer that keeps writing cannot /// hold the loop. -const STREAM_PRESENT: usize = syscall::PIPE_CAPACITY / STREAM_READ; +const STREAM_PRESENT: usize = PIPE_CAPACITY / STREAM_READ; /// The name the private export port has in this server's own namespace. const EXPORT_PORT: &str = "export"; From 094323f298123095202bca3c401db0488fb2d550 Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 28 Oct 2026 17:10:28 +0100 Subject: [PATCH 12/12] issues: a 64 KiB pipe write is 2 % below the 2 MiB ring The final T14 A/B of #1074 against main 9d818b3ed, three interleaved boot pairs, read pipe.throughput.65536 lower on the head in every pair (mean -2.4 %, p50 7555 against 7744 MB/s), with 4096 and 16384 at base and pipe ping-pong faster. The gap is recorded with its one unmeasured candidate, frame placement under the adjacent-page prefetch, and its exit: a PMU session on the landed head at 64 KiB. Co-Authored-By: Claude Opus 5.5 --- ...two-percent-below-the-two-mebibyte-ring.md | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) create mode 100644 issues/a-64-kib-pipe-write-is-two-percent-below-the-two-mebibyte-ring.md diff --git a/issues/a-64-kib-pipe-write-is-two-percent-below-the-two-mebibyte-ring.md b/issues/a-64-kib-pipe-write-is-two-percent-below-the-two-mebibyte-ring.md new file mode 100644 index 00000000000..ada1434c25e --- /dev/null +++ b/issues/a-64-kib-pipe-write-is-two-percent-below-the-two-mebibyte-ring.md @@ -0,0 +1,32 @@ +--- +status: open +kind: defect +opened: 2026-10-28 +--- + +# A 64 KiB pipe write is 2 % below the 2 MiB ring + +Pipe bytes live in 4 KiB frames (`kernel/pure/pipe.rs`, `kernel/src/pipe.rs`) +instead of one physically contiguous 2 MiB page. On the T14 (NVMe boot, 4188 +MHz), three interleaved boot pairs of the frame head against the ring base +read `pipe.throughput.65536` p50 7596 / 7547 / 7521 MB/s against 7616 / 7806 +/ 7811. Every pair is lower (rank test p = 0.05, the floor for 3 against 3), a +mean of −2.4 %, with base's own boot-to-boot spread 2.5 %. 4096 and 16384 are +at base, and `pingpong.pipe.*` p50 is lower (better) on the head in every pair. + +It is not per-piece overhead: a 16 KiB write crosses as many user-page +pieces per KiB and is at base, so a frame-aware walk that saves instructions +per piece is not expected to recover it. + +The unmeasured candidate: a 64 KiB write spans 16 or 17 scattered frames +where the ring was physically contiguous, which a prefetcher that follows into +the adjacent page could use. The earlier image with physically adjacent frames +(PR #1074's evidence r6, diag4) was read while instructions dominated the gap, +on the old boot path at 3789 MHz, so it does not settle this. + +Exit: a PMU session (the measurement-only counters of evidence r6) on the +landed head at 65536 beside base and beside a physically-adjacent-frames +image, then a fix or a folded note at `kernel/src/pipe.rs` stating the +measured cost. + +Evidence: PR #1074, evidence r7 (the final A/B) and r6 (the PMU session).