From 48f9c632e58f411e912ed85c2a50ab0890196b66 Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 21 Oct 2026 16:00:32 +0200 Subject: [PATCH 1/2] pipe: a write of at most MAX_WHOLE_PIPE_WRITE bytes lands whole A pipe write took min(len, free space), so whenever the reader lagged a 40-byte line could split, and another writer's bytes could land between its parts. Its writers are separate processes sharing no lock, so only the pipe can keep a write whole. Owner decision (Jan, 2026-10-21): up to a documented bound a pipe write lands whole or waits for room for all of it; the non-blocking form lands whole or answers WouldBlock; longer writes may stay partial; libc exposes the bound as PIPE_BUF. Pipes only, not files, sockets or devices. - toyos-abi: MAX_WHOLE_PIPE_WRITE = 4096, libc's PIPE_BUF. The kernel asserts at compile time that it fits the ring. - kernel::pipe::Landing says how a write lands: Whole (a message), Pipe (whole up to the bound, as far as it fits past it) or Partial (a connection's plain write, unchanged). room(len) is what the write waits for and takes(space, len) what it writes, from one comparison. - crate::pipe::write(id, buf, landing) replaces both try_write and write_whole: the message send is the same write with Landing::Whole. - sys_write waits on has_space(id, landing.room(len)) instead of one byte, so a waiter wakes only when its write lands, and every read and the reader's leaving still post the write watch, which wakes every waiter to re-read its condition. - A pipe end polls writable only with MAX_WHOLE_PIPE_WRITE bytes free, so a non-blocking write refused and retried on readiness lands rather than spinning. - An empty write waits for nothing: Holders::writable no longer raises the wanted room to one byte, so a zero-byte write to a full pipe or connection answers 0 (or Gone without a reader) instead of parking. Host tests in kernel/pure/pipe.rs model the pipe's write, wait condition and poll readiness over the real Landing and Holders: whole-or-wait, non-blocking refusal answered by readiness, empty writes, the reader leaving, wake-iff-lands at every amount of room, and three writers with a lagging reader whose records never tear (with the same run under Partial as the test's own control). Closes issues/a-pipe-write-of-a-few-bytes-can-tear-between-two-writers.md. Co-Authored-By: Claude Opus 5.5 --- ...-few-bytes-can-tear-between-two-writers.md | 46 --- kernel/pure/pipe.rs | 302 +++++++++++++++++- kernel/src/object/ops.rs | 35 +- kernel/src/object/service.rs | 13 +- kernel/src/pipe.rs | 68 ++-- kernel/src/syscall/io.rs | 9 +- sdk/abi/src/syscall.rs | 13 +- sdk/libc/include/limits.h | 2 + 8 files changed, 364 insertions(+), 124 deletions(-) delete mode 100644 issues/a-pipe-write-of-a-few-bytes-can-tear-between-two-writers.md diff --git a/issues/a-pipe-write-of-a-few-bytes-can-tear-between-two-writers.md b/issues/a-pipe-write-of-a-few-bytes-can-tear-between-two-writers.md deleted file mode 100644 index 04bf46ac1cc..00000000000 --- a/issues/a-pipe-write-of-a-few-bytes-can-tear-between-two-writers.md +++ /dev/null @@ -1,46 +0,0 @@ ---- -status: owner -kind: question -opened: 2026-10-21 ---- - -# A pipe write of a few bytes can tear between two writers - -`pipe::try_write` (`kernel/src/pipe.rs`) writes `min(len, free space)` and -`sys_write` (`kernel/src/syscall/io.rs`) returns that short count. It waits -only while the ring is full, and the caller's write loop sends the rest later. -So whenever the reader lags and the ring has less free space than a write, -even a 40-byte line is split. Another writer's bytes can land between the two -parts. POSIX promises a pipe write of at most `PIPE_BUF` bytes is never -interleaved with another's, and `sdk/libc/include/limits.h` declares no -`PIPE_BUF` at all. - -Every multi-writer pipe has this gap: `{ a & b; } | c`, `make -j | tee log`, -and a stdout stream several children share, whose lines tear in one log -while the file server behind it is busy. A one-byte write, such as a -jobserver token, cannot tear. Record boundaries matter to any -multi-writer pipe, so the fix is native, not POSIX's alone. - -Only the pipe can keep such a write whole: its writers are separate -processes holding one pipe and sharing no lock, so no userland library can -order them. - -The change is one branch in the pipe's write path, plus libc's constant, and -it costs two semantics. A write of at most `PIPE_BUF` waits for room for the -whole write (`pipe::has_space(id, len)`) where today it takes part of the -ring, and a non-blocking one (`sys_write_nonblock`, -`kernel/src/syscall/io.rs`) answers `WouldBlock` where today it answers a -short count, as POSIX has `O_NONBLOCK` do. Writes longer than `PIPE_BUF` -behave as today. - -Owner: the kernel's pipe. It is the owner's call because it changes the -kernel's write semantics (ruling M: no kernel change for POSIX, so the case -for it is the native one above). - -## Exit - -A pipe write of at most `PIPE_BUF` bytes lands whole or waits for room, a -non-blocking one lands whole or answers `WouldBlock`, libc declares -`PIPE_BUF`, and a host test of the pipe shows two writers' writes of at most -`PIPE_BUF` never interleave while a reader lags. Longer writes may still be -split. diff --git a/kernel/pure/pipe.rs b/kernel/pure/pipe.rs index 960d1d38669..b1c846c13bd 100644 --- a/kernel/pure/pipe.rs +++ b/kernel/pure/pipe.rs @@ -8,6 +8,8 @@ #![forbid(unsafe_code)] +use toyos_abi::syscall::MAX_WHOLE_PIPE_WRITE; + /// One end of a pipe. #[derive(Clone, Copy, PartialEq, Eq, Debug)] pub enum End { @@ -83,10 +85,10 @@ impl Holders { available > 0 || !self.held(End::Write) } - /// A write that needs `want` bytes at once, at least one, answers now: it - /// takes them, or is refused for want of a reader. - pub fn writable(&self, space: u32, want: u32) -> bool { - space >= want.max(1) || !self.held(End::Read) + /// 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) } /// The end opposite `end` has no holder, whatever the ring holds. @@ -95,9 +97,45 @@ impl Holders { } } +/// How a write lands in a pipe's ring. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum Landing { + /// All of it or none: a message. + Whole, + /// Whole up to [`MAX_WHOLE_PIPE_WRITE`] bytes, and as far as it fits past + /// that: a pipe end's write. + Pipe, + /// As far as it fits: a connection's plain write. + Partial, +} + +impl Landing { + /// The room a write of `len` bytes waits for: it lands as soon as there is + /// this much, and not before. + pub fn room(self, len: usize) -> usize { + match self { + Self::Whole => len, + Self::Pipe if len <= MAX_WHOLE_PIPE_WRITE => len, + Self::Pipe | Self::Partial => len.min(1), + } + } + + /// 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; + (space >= self.room(len)).then(|| len.min(space)) + } +} + #[cfg(test)] mod tests { use super::*; + use alloc::collections::VecDeque; + use alloc::format; + use alloc::string::String; + use alloc::vec; + use alloc::vec::Vec; /// A pipe as `pipe::create` leaves it: one holder of each end. fn made() -> Holders { @@ -164,4 +202,260 @@ mod tests { assert_eq!(pipe.release(End::Read), Released::EndGone); assert_eq!(pipe.release(End::Write), Released::PipeGone); } + + const MAX: usize = MAX_WHOLE_PIPE_WRITE; + + /// What a write answers, as `kernel::pipe::write` does. + #[derive(PartialEq, Eq, Debug)] + enum Wrote { + Bytes(usize), + /// A blocking write waits for [`Landing::room`]; a non-blocking one + /// answers `WouldBlock`. + NoRoom, + Broken, + } + + /// One pipe as the kernel holds it under its lock: the holders and the + /// bytes in a ring of `cap`. + struct Model { + holders: Holders, + ring: VecDeque, + cap: usize, + } + + impl Model { + fn new(cap: usize) -> Self { + Self { holders: made(), ring: VecDeque::new(), cap } + } + + fn space(&self) -> u32 { + (self.cap - self.ring.len()) as u32 + } + + fn write(&mut self, bytes: &[u8], landing: Landing) -> Wrote { + if !self.holders.held(End::Read) { + return Wrote::Broken; + } + match landing.takes(self.space(), bytes.len()) { + Some(n) => { + self.ring.extend(&bytes[..n]); + Wrote::Bytes(n) + } + None => Wrote::NoRoom, + } + } + + /// A blocking writer's wait condition, as `sys_write` asks it. + fn wakes(&self, len: usize, landing: Landing) -> bool { + self.holders.writable(self.space(), landing.room(len)) + } + + /// A poll's writable, as `ops::has_space` asks it of a pipe end. + fn polls_writable(&self) -> bool { + self.holders.writable(self.space(), MAX) + } + + fn read(&mut self, want: usize) -> Vec { + let n = want.min(self.ring.len()); + self.ring.drain(..n).collect() + } + } + + /// A blocking pipe write of at most `MAX` bytes lands whole or not at all + /// at every amount of room, and a longer one as far as it fits. + #[test] + fn a_small_pipe_write_lands_whole_or_waits() { + for len in [1, 2, 40, MAX - 1, MAX] { + let bytes = vec![7; len]; + for space in [0, 1, len / 2, len - 1, len, len + 1] { + let mut pipe = Model::new(4 * MAX); + pipe.ring.extend(core::iter::repeat_n(0, 4 * MAX - space)); + let answer = pipe.write(&bytes, Landing::Pipe); + if space >= len { + assert_eq!(answer, Wrote::Bytes(len), "a {len}-byte write with room for it"); + } else { + assert_eq!(answer, Wrote::NoRoom, "a {len}-byte write split at {space} bytes of room"); + assert_eq!(pipe.ring.len(), 4 * MAX - space, "a refused write left bytes behind"); + } + } + } + let mut pipe = Model::new(4 * MAX); + pipe.ring.extend(core::iter::repeat_n(0, 4 * MAX - 1)); + assert_eq!(pipe.write(&[7; MAX + 1], Landing::Pipe), Wrote::Bytes(1), "a long write does not wait for all of it"); + } + + /// The non-blocking form is refused whole, and a poll answers writable + /// only when a refused write of at most `MAX` bytes would then land: a + /// retry on readiness is never refused again, so it cannot spin. + #[test] + fn a_refused_nonblocking_write_is_answered_by_readiness() { + let cap = 3 * MAX; + for len in [1, 40, MAX] { + let mut pipe = Model::new(cap); + pipe.ring.extend(core::iter::repeat_n(0, cap - len + 1)); + assert!(!pipe.polls_writable(), "writable with less room than a whole write"); + assert_eq!(pipe.write(&vec![7; len], Landing::Pipe), Wrote::NoRoom); + while !pipe.polls_writable() { + pipe.read(97); + } + assert_eq!(pipe.write(&vec![7; len], Landing::Pipe), Wrote::Bytes(len), "a {len}-byte retry on readiness refused"); + } + } + + /// An empty write moves nothing, so it waits for nothing, even on a full + /// ring, and still answers a missing reader. + #[test] + fn an_empty_write_waits_for_nothing() { + for landing in [Landing::Whole, Landing::Pipe, Landing::Partial] { + let mut pipe = Model::new(64); + pipe.ring.extend([0; 64]); + assert!(pipe.wakes(0, landing), "{landing:?}: an empty write waits on a full ring"); + assert_eq!(pipe.write(&[], landing), Wrote::Bytes(0)); + assert_eq!(pipe.holders.release(End::Read), Released::EndGone); + assert_eq!(pipe.write(&[], landing), Wrote::Broken); + } + } + + /// A writer waiting for room is answered when the reader leaves, and its + /// write is refused rather than landing in a ring nobody reads. + #[test] + fn a_reader_leaving_answers_a_waiting_writer() { + let mut pipe = Model::new(2 * MAX); + pipe.ring.extend(core::iter::repeat_n(0, 2 * MAX - 10)); + assert!(!pipe.wakes(40, Landing::Pipe)); + assert!(!pipe.polls_writable()); + assert_eq!(pipe.holders.release(End::Read), Released::EndGone); + assert!(pipe.wakes(40, Landing::Pipe), "a waiting writer was not answered by the reader leaving"); + assert!(pipe.polls_writable()); + assert_eq!(pipe.write(&[7; 40], Landing::Pipe), Wrote::Broken); + } + + /// The wait condition is exactly the write's: at every amount of room, a + /// waiting writer wakes if and only if its write would land, so it is + /// neither woken to be refused nor left asleep with room for it. + #[test] + fn a_waiting_writer_wakes_exactly_when_its_write_lands() { + let cap = 2 * MAX + 3; + for landing in [Landing::Whole, Landing::Pipe, Landing::Partial] { + for len in [0, 1, 2, 40, MAX - 1, MAX, MAX + 1, 2 * MAX] { + for space in 0..=cap { + let mut pipe = Model::new(cap); + pipe.ring.extend(core::iter::repeat_n(0, cap - space)); + let wakes = pipe.wakes(len, landing); + let lands = pipe.write(&vec![7; len], landing) != Wrote::NoRoom; + assert_eq!(wakes, lands, "{landing:?}, {len} bytes, {space} free: woken {wakes}, lands {lands}"); + } + } + } + } + + /// One writer of a run of records, each written whole or in parts as its + /// write lands, with a `write_all` loop for what is left. + struct Writer { + tag: u8, + lens: Vec, + next: usize, + /// Of the record `next`, already written. + done: usize, + parked: bool, + } + + impl Writer { + fn pending(&self) -> Option { + self.lens.get(self.next).map(|len| len - self.done) + } + } + + /// Three writers of records of up to `MAX` bytes share one pipe whose + /// reader lags, so most writes meet a ring without room for them. A + /// writer whose write is refused parks until a read's post finds its wait + /// condition true. Every record reaches the reader in one piece, and every + /// writer finishes: no wake is missed. + /// Answers how many writes were refused for room. + fn compete(landing: Landing) -> Result { + let mut seed = 0x2545_f491_u64; + let mut random = move |below: usize| { + seed ^= seed << 13; + seed ^= seed >> 7; + seed ^= seed << 17; + (seed % below as u64) as usize + }; + let mut writers: Vec = (0..3u8) + .map(|tag| Writer { + tag: b'a' + tag, + lens: (0..300).map(|_| 1 + random(MAX)).collect(), + next: 0, + done: 0, + parked: false, + }) + .collect(); + let mut pipe = Model::new(2 * MAX + 13); + let mut stream = Vec::new(); + let mut refused = 0; + for _ in 0..1_000_000 { + if writers.iter().all(|w| w.pending().is_none()) && pipe.ring.is_empty() { + break; + } + let runnable: Vec = (0..writers.len()) + .filter(|&i| !writers[i].parked && writers[i].pending().is_some()) + .collect(); + if runnable.is_empty() && pipe.ring.is_empty() { + return Err("every writer parked on an empty ring: a missed wake".into()); + } + // The reader runs one turn in four, or whenever no writer can. + if !pipe.ring.is_empty() && (runnable.is_empty() || random(4) == 0) { + stream.extend(pipe.read(1 + random(MAX / 2))); + // The read posts the write watch: every parked writer re-reads its condition. + for w in &mut writers { + if let Some(len) = w.pending() { + w.parked = w.parked && !pipe.wakes(len, landing); + } + } + continue; + } + let w = &mut writers[runnable[random(runnable.len())]]; + let len = w.pending().expect("a runnable writer has a record"); + match pipe.write(&vec![w.tag; len], landing) { + Wrote::Bytes(n) => { + w.done += n; + if w.done == w.lens[w.next] { + (w.next, w.done) = (w.next + 1, 0); + } + } + Wrote::NoRoom => { + w.parked = true; + refused += 1; + } + Wrote::Broken => unreachable!("the reader stays"), + } + } + if writers.iter().any(|w| w.pending().is_some()) { + return Err("the run did not finish".into()); + } + let mut at = 0; + let mut next = [0; 3]; + while at < stream.len() { + let tag = stream[at]; + let i = (tag - b'a') as usize; + let len = writers[i].lens[next[i]]; + if let Some(off) = stream[at..at + len].iter().position(|&b| b != tag) { + return Err(format!("{}'s record {} of {len} bytes torn at {off}", tag as char, next[i])); + } + (at, next[i]) = (at + len, next[i] + 1); + } + Ok(refused) + } + + #[test] + fn competing_small_writes_never_interleave_while_the_reader_lags() { + let refused = compete(Landing::Pipe).unwrap(); + assert!(refused > 100, "only {refused} writes met a ring without room for them"); + } + + /// The control on the test itself: the same run, writing each record as + /// far as it fits, does tear. + #[test] + 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"); + } } diff --git a/kernel/src/object/ops.rs b/kernel/src/object/ops.rs index d2dc509d7d9..eddf3a05ea2 100644 --- a/kernel/src/object/ops.rs +++ b/kernel/src/object/ops.rs @@ -8,14 +8,14 @@ use alloc::sync::Arc; use alloc::vec::Vec; use toyos_abi::handle::{RawHandle, Rights}; -use toyos_abi::syscall::{FileType, OpenFlags, SeekFrom, SyscallError}; +use toyos_abi::syscall::{FileType, OpenFlags, SeekFrom, SyscallError, MAX_WHOLE_PIPE_WRITE}; use kernel::sched::task::WaitClass; use crate::drivers::serial; use crate::file_cache; use crate::time::Deadline; use crate::pipe::{self, PipeId}; -use kernel::pipe::End; +use kernel::pipe::{End, Landing}; use crate::process::PipeMap; use crate::user_ptr::{UserBytes, UserBytesMut}; use crate::inbox::PollEntry; @@ -211,7 +211,7 @@ pub fn pipe_id_read(object: &KObjectRef) -> Option { } pub fn pipe_id_write(object: &KObjectRef) -> Option { - pipe_write(object).map(|(id, _)| id) + pipe_write(object).map(|(id, _, _)| id) } /// The pipe a blocking read of `object` parks on, and the class its wait is @@ -228,11 +228,12 @@ pub fn pipe_read(object: &KObjectRef) -> Option<(PipeId, WaitClass)> { } } -/// [`pipe_read`]'s answer for a blocking write. -pub fn pipe_write(object: &KObjectRef) -> Option<(PipeId, WaitClass)> { +/// [`pipe_read`]'s answer for a blocking write, and how a write lands in +/// that pipe. +pub fn pipe_write(object: &KObjectRef) -> Option<(PipeId, WaitClass, Landing)> { match object { - KObjectRef::PipeWrite(w) => Some((w.id(), WaitClass::Pipe)), - KObjectRef::Connection(c) => Some((c.tx(), WaitClass::Ipc)), + KObjectRef::PipeWrite(w) => Some((w.id(), WaitClass::Pipe, Landing::Pipe)), + KObjectRef::Connection(c) => Some((c.tx(), WaitClass::Ipc, Landing::Partial)), KObjectRef::PipeRead(_) | KObjectRef::File(_) | KObjectRef::Device(_) | KObjectRef::Console(_) | KObjectRef::Acceptor(_) | KObjectRef::Inbox(_) | KObjectRef::SysCap(_) @@ -546,13 +547,14 @@ pub fn try_read(object: &KObjectRef, buf: &mut UserBytesMut) -> Option { } } -fn write_pipe(id: PipeId, buf: &UserBytes) -> Option { - match pipe::try_write(id, buf) { +fn write_pipe(object: &KObjectRef, buf: &UserBytes) -> Option { + let (id, _, landing) = pipe_write(object).expect("a pipe's write end writes its pipe"); + match pipe::write(id, buf, landing) { // The reader is gone, not absent: `NotFound` means the name is not there and nothing else may say so. - Some(pipe::PipeWrite::BrokenPipe) => Some(SyscallError::Gone.to_u64()), - Some(pipe::PipeWrite::NoMemory) => Some(SyscallError::ResourceExhausted.to_u64()), - Some(pipe::PipeWrite::Wrote(n)) => Some(n as u64), - None => None, + pipe::PipeWrite::BrokenPipe => Some(SyscallError::Gone.to_u64()), + pipe::PipeWrite::NoMemory => Some(SyscallError::ResourceExhausted.to_u64()), + pipe::PipeWrite::Wrote { n, .. } => Some(n as u64), + pipe::PipeWrite::NoRoom => None, } } @@ -593,8 +595,7 @@ pub fn try_write(object: &KObjectRef, buf: &UserBytes) -> Option { file_cache::touch(state.file_id, state.mtime); Some(written as u64) }), - KObjectRef::PipeWrite(w) => write_pipe(w.id(), buf), - KObjectRef::Connection(c) => write_pipe(c.tx(), buf), + KObjectRef::PipeWrite(_) | KObjectRef::Connection(_) => write_pipe(object, buf), // The whole lines `klogd`'s queue has room for, and a trailing partial one: // a short count is a writer ahead of the console, told so. KObjectRef::Console(c) => Some(c.write(buf) as u64), @@ -854,7 +855,9 @@ pub fn read_posts_are_readiness(object: &KObjectRef) -> bool { pub fn has_space(object: &KObjectRef) -> bool { match object { - KObjectRef::PipeWrite(w) => pipe::has_space(w.id(), 1), + // Writable is room for any write that lands whole, so a non-blocking + // one refused, retried on readiness, lands. + KObjectRef::PipeWrite(w) => pipe::has_space(w.id(), MAX_WHOLE_PIPE_WRITE), KObjectRef::Connection(c) => pipe::has_space(c.tx(), 1), KObjectRef::File(_) | KObjectRef::Null(_) => true, KObjectRef::Console(_) => crate::log::console::has_room(), diff --git a/kernel/src/object/service.rs b/kernel/src/object/service.rs index a44a3779436..6b494932a14 100644 --- a/kernel/src/object/service.rs +++ b/kernel/src/object/service.rs @@ -4,9 +4,10 @@ use alloc::sync::Arc; use alloc::vec::Vec; use kernel::message::{Batches, Refused, Window}; +use kernel::pipe::Landing; use toyos_abi::syscall::{SyscallError, MAX_QUEUED_BATCHES}; -use crate::pipe::{self, AtBatch, PipeId, PipeReader, PipeWriter, WholeWrite}; +use crate::pipe::{self, AtBatch, PipeId, PipeReader, PipeWrite, PipeWriter}; use crate::user_ptr::{UserBytes, UserBytesMut}; use crate::sync::Lock; @@ -47,16 +48,16 @@ impl HandleQueue { Err(Refused::Empty) => return Err((batch, SendRefused::Error(SyscallError::InvalidArgument))), Err(Refused::QueueFull) => return Err((batch, SendRefused::Error(SyscallError::ResourceExhausted))), } - match pipe::write_whole(tx, bytes) { - WholeWrite::Wrote(end) => { + match pipe::write(tx, bytes, Landing::Whole) { + PipeWrite::Wrote { end, .. } => { if carries { batches.pin(end, batch); } Ok(()) } - WholeWrite::NoRoom => Err((batch, SendRefused::NoRoom)), - WholeWrite::BrokenPipe => Err((batch, SendRefused::Error(SyscallError::Gone))), - WholeWrite::NoMemory => Err((batch, SendRefused::Error(SyscallError::ResourceExhausted))), + PipeWrite::NoRoom => Err((batch, SendRefused::NoRoom)), + PipeWrite::BrokenPipe => Err((batch, SendRefused::Error(SyscallError::Gone))), + PipeWrite::NoMemory => Err((batch, SendRefused::Error(SyscallError::ResourceExhausted))), } } diff --git a/kernel/src/pipe.rs b/kernel/src/pipe.rs index 822b9ccf1df..7e6df7fb157 100644 --- a/kernel/src/pipe.rs +++ b/kernel/src/pipe.rs @@ -1,7 +1,7 @@ use crate::mm::pmm; use kernel::message::Window; -use kernel::pipe::{End, Holders, Released}; +use kernel::pipe::{End, Holders, Landing, Released}; use toyos_abi::ring::Ring; use alloc::sync::Arc; @@ -101,6 +101,11 @@ mod handle { pub const PIPE_SIZE: usize = PAGE_2M as usize; +const _: () = assert!( + toyos_abi::syscall::MAX_WHOLE_PIPE_WRITE <= PIPE_SIZE - core::mem::size_of::(), + "a whole pipe write must fit the ring", +); + /// A pipe's ring page and the cursors over it — cursors stay in kernel memory /// because `SYS_PIPE_MAP` maps `page` writable, so anything read back from the page is user-chosen. struct Backing { @@ -200,7 +205,7 @@ pub fn init() { /// Create a new pipe. Returns owned reader + writer references. /// -/// Infallible: a pipe with no traffic owns no physical memory, so nothing here can be exhausted; the ring page is allocated on first `try_write` or `map_page`. +/// Infallible: a pipe with no traffic owns no physical memory, so nothing here can be exhausted; the ring page is allocated on first `write` or `map_page`. pub fn create() -> (PipeReader, PipeWriter) { with_pipes_mut(|pipes| { let id = pipes.insert_with(Pipe::new); @@ -271,57 +276,29 @@ pub fn read_position(pipe_id: PipeId) -> Option { with_pipes(|pipes| pipes.get(pipe_id).map(|p| p.read)) } -/// What [`write_whole`] did: all of it, or nothing. -pub enum WholeWrite { - /// Every byte, ending at this stream position. - Wrote(u64), - /// Not enough room for all of it now. +pub enum PipeWrite { + /// This many bytes, the stream ending at `end`. + Wrote { n: usize, end: u64 }, + /// 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. NoMemory, } -/// Write all of `buf` or none of it, under the one lock hold. -pub fn write_whole(pipe_id: PipeId, buf: &UserBytes) -> WholeWrite { +/// Write `buf` as `landing` has it land, under the one lock hold. +pub fn write(pipe_id: PipeId, buf: &UserBytes, landing: Landing) -> PipeWrite { with_pipes_mut(|pipes| { - let Some(pipe) = pipes.get_mut(pipe_id) else { return WholeWrite::BrokenPipe }; + let Some(pipe) = pipes.get_mut(pipe_id) else { return PipeWrite::BrokenPipe }; if !pipe.holders.held(End::Read) { - return WholeWrite::BrokenPipe; - } - let Some(backing) = pipe.back() else { return WholeWrite::NoMemory }; - if (backing.ring.space() as usize) < buf.len() { - return WholeWrite::NoRoom; + return PipeWrite::BrokenPipe; } - let n = backing.ring.write(buf.len(), |off, mut dst| buf.read_run(off, &mut dst)); - assert_eq!(n, buf.len(), "pipe: a ring with room for a write took less of it"); + let Some(backing) = pipe.back() 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"); pipe.written += n as u64; - WholeWrite::Wrote(pipe.written) - }) -} - -pub enum PipeWrite { - Wrote(usize), - BrokenPipe, - /// The ring page could not be allocated; distinct from `None`, no wait makes space appear so a caller must not park on it. - NoMemory, -} - -pub fn try_write(pipe_id: PipeId, buf: &UserBytes) -> Option { - with_pipes_mut(|pipes| { - let pipe = pipes.get_mut(pipe_id)?; - if !pipe.holders.held(End::Read) { - return Some(PipeWrite::BrokenPipe); - } - let Some(backing) = pipe.back() else { - return Some(PipeWrite::NoMemory); - }; - if backing.ring.space() > 0 { - let n = backing.ring.write(buf.len(), |off, mut dst| buf.read_run(off, &mut dst)); - pipe.written += n as u64; - Some(PipeWrite::Wrote(n)) - } else { - None - } + PipeWrite::Wrote { n, end: pipe.written } }) } @@ -331,9 +308,8 @@ pub fn has_data(pipe_id: PipeId) -> bool { }) } -/// Whether a write needing `want` bytes at once answers now. +/// Whether a write waiting for `want` bytes of room answers now. pub fn has_space(pipe_id: PipeId, want: usize) -> bool { - let want = u32::try_from(want).unwrap_or(u32::MAX); with_pipes(|pipes| pipes.get(pipe_id).is_some_and(|p| p.holders.writable(p.space(), want))) } diff --git a/kernel/src/syscall/io.rs b/kernel/src/syscall/io.rs index d231a185eba..786f808f732 100644 --- a/kernel/src/syscall/io.rs +++ b/kernel/src/syscall/io.rs @@ -22,7 +22,8 @@ use super::handles::with_object_ref; /// What `sys_write` does when the object took nothing. enum WriteBlock { - Pipe(pipe::PipeId, WaitClass), + /// The pipe, the class of the wait, and the room the write waits for. + Pipe(pipe::PipeId, WaitClass, usize), Refused(u64), /// Carried out of the process's lock: `HandleError::refuse` may take the /// process down and cannot run under a guard. @@ -55,7 +56,7 @@ pub(super) fn sys_write(h: RawHandle, buf: &UserBytes) -> u64 { match ops::try_write(object, buf) { Some(n) => Ok((n, ops::pipe_id_write(object))), None => Err(match ops::pipe_write(object) { - Some((id, class)) => WriteBlock::Pipe(id, class), + Some((id, class, landing)) => WriteBlock::Pipe(id, class, landing.room(buf.len())), None => WriteBlock::Refused(SyscallError::NotFound.to_u64()), }), } @@ -65,7 +66,7 @@ pub(super) fn sys_write(h: RawHandle, buf: &UserBytes) -> u64 { if let Some(id) = pipe_id { process::wake_pipe_readers(id); } return n; } - Err(WriteBlock::Pipe(id, class)) => match pipe::write_watch(id) { + Err(WriteBlock::Pipe(id, class, room)) => match pipe::write_watch(id) { Some(end) => { let parkable = crate::scheduler::Parkable::at_entry(); if watch::wait_until( @@ -74,7 +75,7 @@ pub(super) fn sys_write(h: RawHandle, buf: &UserBytes) -> u64 { 0, class, Deadline::never(), - || pipe::has_space(id, 1), + || pipe::has_space(id, room), ) .is_err() { diff --git a/sdk/abi/src/syscall.rs b/sdk/abi/src/syscall.rs index 3dc565fbfaa..a9a16a438cf 100644 --- a/sdk/abi/src/syscall.rs +++ b/sdk/abi/src/syscall.rs @@ -752,7 +752,8 @@ fn encode_timeout(timeout: Option) -> u64 { } /// Write bytes to a handle. Returns number of bytes written; a pipe or -/// connection whose other end has closed is [`SyscallError::Gone`]. +/// connection whose other end has closed is [`SyscallError::Gone`]. A pipe +/// write of at most [`MAX_WHOLE_PIPE_WRITE`] bytes writes all of them. pub fn write(handle: RawHandle, buf: &[u8]) -> Result { check(syscall(SYS_WRITE, handle.0 as u64, buf.as_ptr() as u64, buf.len() as u64, 0)).map(|n| n as usize) } @@ -1067,6 +1068,12 @@ pub const MAX_LINKS: u32 = 10; /// libc's `PATH_MAX` (`sdk/libc/include/limits.h`). pub const MAX_PATH: usize = 4096; +/// The longest pipe write that lands whole: never interleaved with another +/// writer's bytes, it waits for room for all of it, and a pipe end polls +/// writable only with this much room. A longer write may land in parts. +/// libc's `PIPE_BUF` (`sdk/libc/include/limits.h`). +pub const MAX_WHOLE_PIPE_WRITE: usize = 4096; + /// 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)) @@ -2328,7 +2335,9 @@ pub fn read_nonblock(handle: RawHandle, buf: &mut [u8]) -> Result Result { check(syscall(SYS_WRITE_NONBLOCK, handle.0 as u64, buf.as_ptr() as u64, buf.len() as u64, 0)).map(|n| n as usize) } diff --git a/sdk/libc/include/limits.h b/sdk/libc/include/limits.h index 54df895a2dd..206b071649c 100644 --- a/sdk/libc/include/limits.h +++ b/sdk/libc/include/limits.h @@ -26,5 +26,7 @@ #define PATH_MAX 4096 #define _POSIX_ARG_MAX 4096 #define NAME_MAX 255 +/* A pipe write of at most this many bytes lands whole. */ +#define PIPE_BUF 4096 #endif From 1fcbfd9d23b8cfd7f316fcb082a9c89c8d84a6aa Mon Sep 17 00:00:00 2001 From: japabu Date: Wed, 21 Oct 2026 16:20:00 +0200 Subject: [PATCH 2/2] review r1: hold PIPE_BUF to the kernel's bound; drop stale try_write names - sdk/libc/arch: a host test asserts limits.h's PIPE_BUF equals MAX_WHOLE_PIPE_WRITE, as path_max_is_limits_h_s does for PATH_MAX. - kernel/loom/tests/inbox_answer.rs and issues/the-global-pipe-lock-spans-a-user-copy.md name pipe::write, which replaced pipe::try_write. - kernel/pure/pipe.rs's header says it also owns how a write lands. Co-Authored-By: Claude Opus 5.5 --- issues/the-global-pipe-lock-spans-a-user-copy.md | 8 ++++---- kernel/loom/tests/inbox_answer.rs | 2 +- kernel/pure/pipe.rs | 3 ++- sdk/libc/arch/src/header.rs | 5 +++++ 4 files changed, 12 insertions(+), 6 deletions(-) diff --git a/issues/the-global-pipe-lock-spans-a-user-copy.md b/issues/the-global-pipe-lock-spans-a-user-copy.md index d0972d18e96..df127a234e1 100644 --- a/issues/the-global-pipe-lock-spans-a-user-copy.md +++ b/issues/the-global-pipe-lock-spans-a-user-copy.md @@ -4,7 +4,7 @@ kind: defect opened: 2026-08-28 --- -# `pipe::try_read`/`try_write` hold the one global `PIPES` spinlock across a 2 MiB user copy, and a pipe's first write adds a bitmap scan and a 2 MiB zeroing under it +# `pipe::try_read`/`write` hold the one global `PIPES` spinlock across a 2 MiB user copy, and a pipe's first write adds a bitmap scan and a 2 MiB zeroing under it Every pipe and every IPC `Connection` in the machine lives behind one static: `static PIPES: Lock>>` (`kernel/src/pipe.rs:174`). `with_pipes_mut` takes it and runs the whole closure under it (`pipe.rs:181-184`), and `sync::Lock::lock` calls `crate::preempt::disable()` before it spins (`kernel/src/sync.rs:56`), so the holder is non-preemptible and every other CPU that wants a pipe spins with preemption off too. @@ -13,12 +13,12 @@ Every pipe and every IPC `Connection` in the machine lives behind one static: `s The bulk copy is inside that closure, not outside it. * `pipe::try_read` (`pipe.rs:210`) calls `ring.read(len, |off, src| buf.write_at(off, src))` at `pipe.rs:220`, inside `with_pipes_mut`. -* `pipe::try_write` (`pipe.rs:244`) calls `backing.ring.write(buf.len(), |off, dst| buf.read_at(off, dst))` at `pipe.rs:254`, same acquisition. +* `pipe::write` (`pipe.rs:244`) calls `backing.ring.write(buf.len(), |off, dst| buf.read_at(off, dst))` at `pipe.rs:254`, same acquisition. * `Ring::read`/`Ring::write` (`sdk/abi/src/ring.rs:141`, `:178`) hand the closure one or two contiguous runs; the closure is `UserBytesMut::write_at` / `UserBytes::read_at` (`kernel/src/user_ptr.rs:206`, `:164`), a `copy_nonoverlapping` of the whole run. The size is bounded only by the ring: `PIPE_SIZE = PAGE_2M` (`pipe.rs:104`), `PAGE_2M = 2 * 1024 * 1024` (`lib/userbound/src/span.rs:27`), and `capacity = total_size - size_of::()` (`ring.rs:60`) with `RingHeader` `#[repr(C, align(64))]` holding one `AtomicU32` (`ring.rs:24-27`) — so **2,097,088 bytes** is the largest single copy under the lock. Nothing above caps it: `SYS_READ`/`SYS_WRITE` pass the userland length straight through (`kernel/src/syscall/dispatch.rs:110-116`), `object::ops::try_read` hands the full window to `pipe::try_read` (`kernel/src/object/ops.rs:339-340`), and the only other bound, `user_ptr::window` (`user_ptr.rs:269`), requires physical contiguity — which a demand-paged 2 MiB frame satisfies exactly. -A pipe's **first** write is worse, because the page is allocated lazily under the same lock. `try_write` calls `pipe.back()` (`pipe.rs:250`), which calls `pmm::alloc_page()` (`pipe.rs:148`). That takes `BITMAP` nested inside `PIPES` (`kernel/src/mm/pmm.rs:221`), linearly scans up to the whole physical bitmap for a free frame (`pmm.rs:224-241`), and then `write_bytes(..., 0, PAGE_2M)` — a 2 MiB zeroing (`pmm.rs:233-237`) — before `Ring::new` and the user copy that follows it. +A pipe's **first** write is worse, because the page is allocated lazily under the same lock. `write` calls `pipe.back()` (`pipe.rs:250`), which calls `pmm::alloc_page()` (`pipe.rs:148`). That takes `BITMAP` nested inside `PIPES` (`kernel/src/mm/pmm.rs:221`), linearly scans up to the whole physical bitmap for a free frame (`pmm.rs:224-241`), and then `write_bytes(..., 0, PAGE_2M)` — a 2 MiB zeroing (`pmm.rs:233-237`) — before `Ring::new` and the user copy that follows it. ## What queues behind it @@ -44,7 +44,7 @@ For the allocation half, the repro is simpler: `SYS_CONNECT` then a single first `PIPES` is doing two jobs and only one of them needs the global. It is the table lock, and it is also the serialization `Ring` relies on — `ring.rs:41-43` says so plainly: "Every access is serialized by the caller's lock (`kernel::pipe`'s `PIPES`), which is why the cursors are plain integers rather than atomics." So the copy cannot simply be lifted out; the ring's data-race argument has to move with it. -The shape that keeps that argument true is a per-pipe lock under the table lock: `PIPES` guards the `IdMap` and the refcounts, a `Lock` (or a claimed-range token) inside `Pipe` guards `Backing`, and the table lock is dropped once the per-pipe guard is in hand. `try_read`/`try_write` then hold only the pipe's own lock across `Ring::read`/`Ring::write`, and `has_data`/`has_space`/`create`/`clone`/`close` stop queueing behind a stranger's copy. `Ring`'s header sentence changes with it — the serializing lock becomes the pipe's, not the table's. +The shape that keeps that argument true is a per-pipe lock under the table lock: `PIPES` guards the `IdMap` and the refcounts, a `Lock` (or a claimed-range token) inside `Pipe` guards `Backing`, and the table lock is dropped once the per-pipe guard is in hand. `try_read`/`write` then hold only the pipe's own lock across `Ring::read`/`Ring::write`, and `has_data`/`has_space`/`create`/`clone`/`close` stop queueing behind a stranger's copy. `Ring`'s header sentence changes with it — the serializing lock becomes the pipe's, not the table's. The lazy allocation should move out from under both: allocate the ring page before taking the table lock (or, on the miss, drop and retry the way `file_cache::read_page` already does at `kernel/src/file_cache.rs:217-231`), so neither the bitmap scan nor the 2 MiB zeroing runs with a global held. diff --git a/kernel/loom/tests/inbox_answer.rs b/kernel/loom/tests/inbox_answer.rs index 7266833180d..af9d55b9108 100644 --- a/kernel/loom/tests/inbox_answer.rs +++ b/kernel/loom/tests/inbox_answer.rs @@ -231,7 +231,7 @@ impl Submitter for Kernel { } /// A peer's write of no bytes posts the pipe all the same -/// (`pipe::try_write` answers `Wrote(0)` and `sys_write` wakes the readers). +/// (`pipe::write` answers zero bytes and `sys_write` wakes the readers). /// Nothing is there to read, so nothing is answered, and the poll armed again /// answers the bytes that do come. #[test] diff --git a/kernel/pure/pipe.rs b/kernel/pure/pipe.rs index b1c846c13bd..44622051cc0 100644 --- a/kernel/pure/pipe.rs +++ b/kernel/pure/pipe.rs @@ -1,4 +1,5 @@ -//! Who holds a pipe's two ends, and what a watch on one of them is told. +//! Who holds a pipe's two ends, what a watch on one of them is told, and how +//! a write lands in its ring. //! //! 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 diff --git a/sdk/libc/arch/src/header.rs b/sdk/libc/arch/src/header.rs index 50f4e81318a..975e095eebe 100644 --- a/sdk/libc/arch/src/header.rs +++ b/sdk/libc/arch/src/header.rs @@ -29,6 +29,11 @@ pub(crate) fn int(header: &str, name: &str) -> i32 { as_int(found.unwrap_or_else(|| panic!("no integer #define {name}"))).1 } +#[test] +fn pipe_buf_is_the_kernels_whole_pipe_write() { + assert_eq!(toyos_abi::syscall::MAX_WHOLE_PIPE_WRITE, int(LIMITS_H, "PIPE_BUF") as usize); +} + /// Each signal `signal.h` numbers: `SIGHUP`, and not `SIG_BLOCK`. pub(crate) fn signals() -> Vec<(&'static str, i32)> { defines(SIGNAL_H).filter(|(name, _)| name.starts_with("SIG") && !name.contains('_')).map(as_int).collect()