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 04bf46ac1c..0000000000 --- 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/issues/the-global-pipe-lock-spans-a-user-copy.md b/issues/the-global-pipe-lock-spans-a-user-copy.md index d0972d18e9..df127a234e 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 7266833180..af9d55b910 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 960d1d3866..44622051cc 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 @@ -8,6 +9,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 +86,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 +98,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 +203,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 d2dc509d7d..eddf3a05ea 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 a44a377943..6b494932a1 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 822b9ccf1d..7e6df7fb15 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 d231a185eb..786f808f73 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 3dc565fbfa..a9a16a438c 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/arch/src/header.rs b/sdk/libc/arch/src/header.rs index 50f4e81318..975e095eeb 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() diff --git a/sdk/libc/include/limits.h b/sdk/libc/include/limits.h index 54df895a2d..206b071649 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