diff --git a/build/ci.rs b/build/ci.rs index 6b8418f2fb3..eb37ecb4f7a 100644 --- a/build/ci.rs +++ b/build/ci.rs @@ -651,10 +651,23 @@ pub(crate) const CONTROLS: &[Control] = &[ red(BLOCKRING, "mutate-no-reissue-after-loss", None, &[ Fails("model::tests::what_a_flush_calls_durable_is_on_the_medium"), ]), - red(TRANSPORT, "publish-relaxed", Some("loom"), &[Fails("a_published_entry_is_read_whole")]), + red(TRANSPORT, "publish-relaxed", Some("loom"), &[ + Fails("a_published_entry_is_read_whole"), + Fails("a_published_byte_is_read_as_written"), + ]), + red(TRANSPORT, "head-relaxed", Some("loom"), &[Fails("a_published_byte_is_read_as_written")]), red(TRANSPORT, "no-clamp", Some("loom"), &[Fails( "a_hostile_producer_yields_entries_or_a_violation", )]), + red(TRANSPORT, "no-clamp", None, &[Fails( + "stream::tests::a_hostile_producer_hands_the_consumer_at_most_the_ring", + )]), + // The stream's lost wake: a sleeper whose exchange read the waker's clear + // looks at a cursor from before it, and sleeps with nobody to ring. + red(TRANSPORT, "want-relaxed", Some("loom"), &[ + Fails("a_consumer_that_looked_again_is_rung_for_bytes_it_missed"), + Fails("a_producer_that_looked_again_is_rung_for_room_it_missed"), + ]), red(TRANSPORT, "end-keeps-inflight", None, &[Fails( "inflight::tests::an_end_answers_every_tag_once_and_a_late_completion_nothing", )]), diff --git a/issues/loom-arc-panics-in-its-drop-while-a-failed-model-unwinds.md b/issues/loom-arc-panics-in-its-drop-while-a-failed-model-unwinds.md new file mode 100644 index 00000000000..42b9beec917 --- /dev/null +++ b/issues/loom-arc-panics-in-its-drop-while-a-failed-model-unwinds.md @@ -0,0 +1,34 @@ +--- +status: open +kind: tooling +opened: 2026-10-12 +--- + +# Loom's `Arc` panics in its drop while a failed model unwinds + +The loom fork, ToyOSOrg/loom `toyos` at `dfe3797`, carries v0.7.2's +`src/sync/arc.rs` unchanged. `Arc`'s `Drop` calls `rt::Arc::ref_dec`, which +branches the scheduler. A model that deadlocks has no thread active when +`Execution::schedule` panics, so the drop that panic's unwinding reaches +unwraps `None` in `rt::thread::Set::active_mut` (`src/rt/thread.rs:276`), and +a destructor panicking during cleanup aborts the process. The test binary +prints loom's `deadlock; threads = ...`, then aborts before the harness prints +the test's `FAILED` line or its summary: a gate that reads the harness's +verdicts finds none. Loom's cells and atomics skip their checks +while `std::thread::panicking()`; its `Arc` does not. + +**Evidence** ([ToyOSOrg/evidence `29c8ebcb`, `arc-probe/`](https://github.com/ToyOSOrg/evidence/tree/29c8ebcb04eae318598ff7cb0a045e6e9a1c4f53/arc-probe)): +at `6d2e78400`, with the stream model in +`lib/blockring/transport/tests/loom.rs` on `loom::sync::Arc` in place of +std's, `cargo test -p toyos-transport --features want-relaxed --test loom -- +a_consumer_that_looked_again_is_rung_for_bytes_it_missed --exact` kills the +test binary with SIGABRT, `panic in a destructor during cleanup`, through +` as Drop>::drop` and `rt::arc::Arc::ref_dec`. +On std's `Arc` the same run prints the test's `FAILED` line and loom's +`deadlock; threads = ...`. + +**Owner**: the orchestrator. Nobody holds it. + +**Exit**: a test on the fork's `toyos` branch in which a model that deadlocks +with a `loom::sync::Arc` live prints loom's `deadlock` message and its test +fails without an abort, and that stream model holds its state in loom's `Arc`. diff --git a/lib/blockring/src/model.rs b/lib/blockring/src/model.rs index 90d6a58b609..433755e272f 100644 --- a/lib/blockring/src/model.rs +++ b/lib/blockring/src/model.rs @@ -98,6 +98,9 @@ impl Word for Shared { fn store(&self, value: u32, _: Ordering) { self.0.set(value) } + fn swap(&self, value: u32, _: Ordering) -> u32 { + self.0.replace(value) + } } /// How deep the model's rings are: shallow enough that a script fills and diff --git a/lib/blockring/transport/Cargo.toml b/lib/blockring/transport/Cargo.toml index 55602697a99..60ea174eac3 100644 --- a/lib/blockring/transport/Cargo.toml +++ b/lib/blockring/transport/Cargo.toml @@ -1,16 +1,16 @@ # A member of the host workspace (root `Cargo.toml`). The one transport every -# client/server session runs over: the rings, the arena's runs and the tag -# table, as decisions over words an adapter hands in. Nothing here maps, copies -# or blocks. +# client/server session runs over: the rings, the byte streams and their wants, +# the arena's runs and the tag table, as decisions over words an adapter hands +# in. Nothing here maps, copies or blocks. # -# Its defects are orders and hostile peers, so `tests/loom.rs` holds the -# publication edge and a hostile peer. +# Its defects are orders, lost wakes and hostile peers, so `tests/loom.rs` +# holds the publication edge, the wake handshake and a hostile peer. [package] name = "toyos-transport" version = "0.1.0" edition = "2021" -description = "The transport every client/server session runs over: rings, an arena of runs and a tag table, decided over words an adapter hands in." +description = "The transport every client/server session runs over: entry rings, byte streams and their wakes, an arena of runs and a tag table, decided over words an adapter hands in." license = "MIT OR Apache-2.0" publish = false @@ -20,11 +20,19 @@ publish = false # `CONTROLS` runs each and demands the named test's own FAILED line. # # A producer publishes its tail `Relaxed`, so a consumer can see the index -# before the entry's words. +# before the entry's words, or a stream's tail before its bytes. publish-relaxed = [] +# A stream's consumer stores its head `Relaxed` and its producer loads it +# `Relaxed`, so the producer can copy into a byte the consumer is still +# copying out. +head-relaxed = [] # A peer's cursor is believed however far it claims to be, so a hostile -# producer hands the consumer more entries than the ring holds. +# producer hands the consumer more entries, or bytes, than the ring holds. no-clamp = [] +# A stream's waker clears the peer's want `Relaxed`, so a sleeper whose +# exchange reads the clear need not see the cursor stored before it, and +# sleeps over bytes or room nobody rings for. +want-relaxed = [] # `Inflight::end` answers nothing and keeps every tag in flight, so a session's # requests outlive it unanswered and a late completion answers one. end-keeps-inflight = [] diff --git a/lib/blockring/transport/src/lib.rs b/lib/blockring/transport/src/lib.rs index 1fbfcbfecb1..940dfff917a 100644 --- a/lib/blockring/transport/src/lib.rs +++ b/lib/blockring/transport/src/lib.rs @@ -3,8 +3,9 @@ //! **A session is one connection and, at most, one region.** The connection //! carries the hello, handles, doorbells and the hang-up; the region carries //! [`Producer`]/[`Consumer`] queues of fixed-size entries and an arena of -//! [`Run`]s. A protocol says what the entries mean; this crate says only what -//! is safe. +//! [`Run`]s, or byte [`stream`]s and the wants that say when to ring. A +//! protocol says what the entries and bytes mean; this crate says only what is +//! safe. //! //! **Nothing here holds the region.** An adapter hands every call the `N` //! [`Word`]s an end was made over, and copies bytes itself, through a run's @@ -29,6 +30,7 @@ mod arena; mod inflight; mod queue; +pub mod stream; use core::sync::atomic::{AtomicU32, Ordering}; @@ -42,6 +44,7 @@ pub use toyos_untrusted::Untrusted; pub trait Word { fn load(&self, order: Ordering) -> u32; fn store(&self, value: u32, order: Ordering); + fn swap(&self, value: u32, order: Ordering) -> u32; } impl Word for AtomicU32 { @@ -51,6 +54,29 @@ impl Word for AtomicU32 { fn store(&self, value: u32, order: Ordering) { AtomicU32::store(self, value, order) } + fn swap(&self, value: u32, order: Ordering) -> u32 { + AtomicU32::swap(self, value, order) + } +} + +/// How an end stores the cursor that hands what it wrote to the peer. +#[cfg(not(feature = "publish-relaxed"))] +const PUBLISH: Ordering = Ordering::Release; +#[cfg(feature = "publish-relaxed")] +const PUBLISH: Ordering = Ordering::Relaxed; + +/// Word `at` of an end's words, and the one index into them: every `at` an end +/// forms is below `N`, because its place is and a ring position is reduced +/// below the depth. +#[allow(clippy::indexing_slicing)] +fn word(page: &[W; N], at: usize) -> &W { + &page[at] +} + +/// A distance the peer's cursor claims, believed only up to `cap`. +fn clamp(claimed: Untrusted, cap: u32, broken: Violation) -> Result { + let cap = if cfg!(feature = "no-clamp") { u32::MAX } else { cap }; + claimed.at_most(u64::from(cap)).ok().and_then(|n| u32::try_from(n).ok()).ok_or(broken) } /// The peer wrote what no peer of the protocol writes. The session is over; diff --git a/lib/blockring/transport/src/queue.rs b/lib/blockring/transport/src/queue.rs index 681b101b243..865c66ac991 100644 --- a/lib/blockring/transport/src/queue.rs +++ b/lib/blockring/transport/src/queue.rs @@ -8,26 +8,7 @@ use core::sync::atomic::Ordering; -use crate::{Untrusted, Violation, Word}; - -#[cfg(not(feature = "publish-relaxed"))] -const PUBLISH: Ordering = Ordering::Release; -#[cfg(feature = "publish-relaxed")] -const PUBLISH: Ordering = Ordering::Relaxed; - -/// Word `at` of an end's words, and the one index into them: every `at` an end -/// forms is below `N`, because its [`Place`] is and a ring position is masked -/// below the depth. -#[allow(clippy::indexing_slicing)] -fn word(page: &[W; N], at: usize) -> &W { - &page[at] -} - -/// A distance the peer's cursor claims, believed only up to `cap`. -fn clamp(claimed: Untrusted, cap: u32, broken: Violation) -> Result { - let cap = if cfg!(feature = "no-clamp") { u32::MAX } else { cap }; - claimed.at_most(u64::from(cap)).ok().and_then(|n| u32::try_from(n).ok()).ok_or(broken) -} +use crate::{clamp, word, Untrusted, Violation, Word, PUBLISH}; /// Where a queue of `E`-word entries, `D` deep, is among `N` words: the word /// its consumer stores its head in, the word its producer stores its tail in, diff --git a/lib/blockring/transport/src/stream.rs b/lib/blockring/transport/src/stream.rs new file mode 100644 index 00000000000..ead60173214 --- /dev/null +++ b/lib/blockring/transport/src/stream.rs @@ -0,0 +1,563 @@ +//! A single-producer ring of `C` bytes, its cursors and wants among `N` words. +//! +//! **A producer copies bytes in through the [`Span`]s it is handed, then +//! publishes its tail with `Release`; a consumer loads the tail with `Acquire`, +//! then copies the bytes out.** The head goes back the other way. Either end +//! addresses bytes by their offset from its own cursor, so a producer may place +//! a byte anywhere in its room before it advances over it. An end reads the +//! peer's cursor only when it asks for its room or what is ready, and stores +//! its own once a batch. +//! +//! **A cursor is a position modulo twice the ring**, so a full ring and an empty +//! one differ and the ring may be any size. The peer's is believed only as a +//! position at most the ring away from what this end last stored; otherwise +//! [`Violation::TailPastDepth`] or [`Violation::HeadPastTail`] names it. Every +//! span lies in the ring whatever the peer wrote. +//! +//! **A sleeper never misses a wake.** An end about to sleep [`arm`]s its want, +//! which exchanges 1 into it and only then looks at the peer's cursor again; it +//! sleeps only if that look finds nothing. An end that publishes or releases +//! exchanges the peer's want for 0 after its cursor and rings the doorbell only +//! if the want was set. The two exchanges are on one word, so the waker's is +//! read by the sleeper's and carries its cursor with it, or follows it and +//! finds it set. A sleeper drains its doorbell before it looks, never between +//! arming and sleeping, or it can drain the ring that answers its want. +//! +//! **The peer's want words cost at most a doorbell**: any value but 0 is set. +//! +//! [`arm`]: Consumer::arm + +use core::sync::atomic::Ordering; + +use crate::{clamp, word, Span, Untrusted, Violation, Word, PUBLISH}; + +/// A sleeper's exchange: what the waker published before clearing the want is +/// seen by the look that follows. +const ARM: Ordering = Ordering::Acquire; +#[cfg(not(feature = "want-relaxed"))] +const WAKE: Ordering = Ordering::Release; +#[cfg(feature = "want-relaxed")] +const WAKE: Ordering = Ordering::Relaxed; +/// The head's edge: a producer that loads the head copies into a byte only +/// after the consumer's copy out of it, which the consumer's store carries. +const GIVE_BACK: Ordering = if cfg!(feature = "head-relaxed") { Ordering::Relaxed } else { Ordering::Release }; +const TAKE_BACK: Ordering = if cfg!(feature = "head-relaxed") { Ordering::Relaxed } else { Ordering::Acquire }; + +/// `n` as an offset into the region; `Place::new` holds `usize` to every `u32`. +#[allow(clippy::as_conversions)] +const fn bytes(n: u32) -> usize { + n as usize +} + +/// Where a ring of `C` bytes is: its four words among `N` and the region offset +/// of its first byte. +/// +/// ``` +/// use toyos_transport::stream::Place; +/// const RING: Place<4096, 64> = Place::new::<0, 16, 32, 48, 4096>(); +/// ``` +/// +/// A place with a word at `N` or past it does not compile: +/// +/// ```compile_fail,E0080 +/// use toyos_transport::stream::Place; +/// const RING: Place<4096, 64> = Place::new::<0, 16, 32, 64, 4096>(); +/// ``` +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct Place { + head: usize, + tail: usize, + /// Set by a consumer about to sleep on an empty ring. + data: usize, + /// Set by a producer about to sleep on a full ring. + room: usize, + base: usize, +} + +impl Place { + /// What a cursor counts modulo. + const LAP: u32 = C.wrapping_mul(2); + + pub const fn new( + ) -> Self { + const { + assert!(C > 0 && C <= 1 << 30, "a ring holds a byte and at most a gibibyte"); + assert!(usize::BITS >= u32::BITS, "a ring offset is a u32"); + assert!(HEAD < N && TAIL < N && DATA < N && ROOM < N, "a place names a word past its words"); + assert!( + HEAD != TAIL && HEAD != DATA && HEAD != ROOM && TAIL != DATA && TAIL != ROOM && DATA != ROOM, + "a ring's four words are four" + ); + assert!(BASE.checked_add(bytes(C)).is_some(), "a ring ends inside the address space"); + } + Self { head: HEAD, tail: TAIL, data: DATA, room: ROOM, base: BASE } + } + + /// `at` moved `n` on; both are below the lap. + fn forward(at: u32, n: u32) -> u32 { + let to = at.wrapping_add(n); + if to >= Self::LAP { to.wrapping_sub(Self::LAP) } else { to } + } + + /// How far `to` is past `from`; both are below the lap. + fn distance(from: u32, to: u32) -> u32 { + let apart = to.wrapping_sub(from); + if to >= from { apart } else { apart.wrapping_add(Self::LAP) } + } + + /// How far the peer's `claimed` position is from this end's, by `apart`: + /// at most the ring. A word past the lap names no position, so it is + /// farther than any ring. + fn reach(claimed: Untrusted, apart: impl FnOnce(u32) -> u32, broken: Violation) -> Result { + clamp(claimed.map(|at| if at < Self::LAP { apart(at) } else { u32::MAX }), C, broken) + } + + /// The region's bytes `len` on from position `at`, in ring order: the + /// second span is empty unless they wrap. + fn spans(&self, at: u32, len: u32) -> [Span; 2] { + let slot = if at >= C { at.wrapping_sub(C) } else { at }; + let first = len.min(C.wrapping_sub(slot)); + let start = self.base.wrapping_add(bytes(slot)); + [Span { offset: start, len: bytes(first) }, Span { offset: self.base, len: bytes(len.wrapping_sub(first)) }] + } +} + +/// The end of a ring that writes bytes. It holds its cursors and not the +/// region; every call is given the region's words. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct Producer { + place: Place, + local: u32, + published: u32, + /// Bytes past `local` that may be written, as last seen. + room: u32, +} + +impl Producer { + /// This end of the ring at `place`, its tail stored 0. + pub fn new(page: &[W; N], place: Place) -> Self { + word(page, place.tail).store(0, Ordering::Release); + Self { place, local: 0, published: 0, room: C } + } + + /// How many bytes past the tail may be written, the consumer's head read + /// again. + pub fn room(&mut self, page: &[W; N]) -> Result { + let head = Untrusted::new(word(page, self.place.head).load(TAKE_BACK)); + let published = self.published; + let unreleased = + Place::::reach(head, |head| Place::::distance(head, published), Violation::HeadPastTail)?; + let pending = Place::::distance(self.published, self.local); + self.room = C.saturating_sub(unreleased.saturating_add(pending)); + Ok(self.room) + } + + /// Where to copy the `len` bytes `at` past the tail; `None` past the room + /// last seen. + pub fn span(&self, at: u32, len: u32) -> Option<[Span; 2]> { + (at.checked_add(len)? <= self.room).then(|| self.place.spans(Place::::forward(self.local, at), len)) + } + + /// Count `n` more bytes written at the tail; `false`, and nothing counted, + /// past the room last seen. They are the consumer's once + /// [`Self::publish`] runs. + #[must_use] + pub fn advance(&mut self, n: u32) -> bool { + let Some(room) = self.room.checked_sub(n) else { return false }; + self.local = Place::::forward(self.local, n); + self.room = room; + true + } + + /// Publish every byte advanced over so far; `true` if the consumer armed + /// its want and the doorbell is to be rung. + #[must_use] + pub fn publish(&mut self, page: &[W; N]) -> bool { + if self.published == self.local { + return false; + } + word(page, self.place.tail).store(self.local, PUBLISH); + self.published = self.local; + word(page, self.place.data).swap(0, WAKE) != 0 + } + + /// Ask to be rung once the consumer releases, then read its head again: + /// the room, and 0 is safe to sleep on. + pub fn arm(&mut self, page: &[W; N]) -> Result { + word(page, self.place.room).swap(1, ARM); + self.room(page) + } +} + +/// The end of a ring that reads bytes; like [`Producer`], it holds cursors and +/// is given the region's words. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct Consumer { + place: Place, + local: u32, + released: u32, + /// Bytes published past `local`, as last seen. + ready: u32, +} + +impl Consumer { + /// This end of the ring at `place`, its head stored 0. + pub fn new(page: &[W; N], place: Place) -> Self { + word(page, place.head).store(0, Ordering::Release); + Self { place, local: 0, released: 0, ready: 0 } + } + + /// How many bytes past the head are published, the producer's tail read + /// again. + pub fn ready(&mut self, page: &[W; N]) -> Result { + let tail = Untrusted::new(word(page, self.place.tail).load(Ordering::Acquire)); + let released = self.released; + let published = + Place::::reach(tail, |tail| Place::::distance(released, tail), Violation::TailPastDepth)?; + self.ready = published.saturating_sub(Place::::distance(self.released, self.local)); + Ok(self.ready) + } + + /// Where to copy the `len` bytes `at` past the head from; `None` past what + /// was last seen ready. + pub fn span(&self, at: u32, len: u32) -> Option<[Span; 2]> { + (at.checked_add(len)? <= self.ready).then(|| self.place.spans(Place::::forward(self.local, at), len)) + } + + /// Count `n` more bytes read at the head; `false`, and nothing counted, + /// past what was last seen ready. They are the producer's once + /// [`Self::release`] runs. + #[must_use] + pub fn advance(&mut self, n: u32) -> bool { + let Some(ready) = self.ready.checked_sub(n) else { return false }; + self.local = Place::::forward(self.local, n); + self.ready = ready; + true + } + + /// Give every byte advanced over so far back to the producer; `true` if it + /// armed its want and the doorbell is to be rung. + #[must_use] + pub fn release(&mut self, page: &[W; N]) -> bool { + if self.released == self.local { + return false; + } + word(page, self.place.head).store(self.local, GIVE_BACK); + self.released = self.local; + word(page, self.place.room).swap(0, WAKE) != 0 + } + + /// Ask to be rung once the producer publishes, then read its tail again: + /// what is ready, and 0 is safe to sleep on. + pub fn arm(&mut self, page: &[W; N]) -> Result { + word(page, self.place.data).swap(1, ARM); + self.ready(page) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use core::sync::atomic::AtomicU32; + + const WORDS: usize = 4; + /// Off any alignment, so a span that forgets the base lands outside it. + const BASE: usize = 3; + /// Bytes on either side of the ring that no span may reach. + const GUARD: usize = 3; + + fn page() -> [AtomicU32; WORDS] { + core::array::from_fn(|_| AtomicU32::new(0)) + } + + fn place() -> Place { + Place::new::<0, 1, 2, 3, BASE>() + } + + /// A region with the ring between two guards. + fn region() -> Vec { + vec![0xA5; BASE + C as usize + GUARD] + } + + /// The byte at stream position `at`: one read from the wrong slot or the + /// wrong lap is, but for one chance in 256 a byte, another byte. + fn byte(at: u64) -> u8 { + (at.wrapping_add(1).wrapping_mul(0x9E37_79B9_7F4A_7C15) >> 56) as u8 + } + + /// xorshift64: deterministic, and seeded per run. + struct Rng(u64); + + impl Rng { + fn next(&mut self) -> u64 { + self.0 ^= self.0 << 13; + self.0 ^= self.0 >> 7; + self.0 ^= self.0 << 17; + self.0 + } + + fn below(&mut self, n: u32) -> u32 { + (self.next() % u64::from(n)) as u32 + } + } + + fn inside(spans: [Span; 2], len: u32) { + assert_eq!(spans[0].len + spans[1].len, len as usize, "{spans:?} is not {len} bytes"); + for span in spans { + assert!(span.offset >= BASE && span.offset + span.len <= BASE + C as usize, "{span:?} leaves the ring"); + } + } + + fn put(region: &mut [u8], spans: [Span; 2], from: u64) { + let mut at = from; + for span in spans { + for b in &mut region[span.offset..span.offset + span.len] { + *b = byte(at); + at += 1; + } + } + } + + fn get(region: &[u8], spans: [Span; 2]) -> Vec { + spans.iter().flat_map(|span| region[span.offset..span.offset + span.len].iter().copied()).collect() + } + + fn expected(from: u64, len: u32) -> Vec { + (from..from + u64::from(len)).map(byte).collect() + } + + /// The oracle is a stream that never wraps: four counts of bytes since the + /// start. The producer places each batch out of order, the consumer reads + /// at offsets as well as at its head, and every room, every count ready and + /// every byte read is what the counts say, round the lap many times. + fn agrees(seed: u64) { + let page = page(); + let (mut tx, mut rx) = (Producer::new(&page, place::()), Consumer::new(&page, place::())); + let mut region = region::(); + let (mut written, mut published, mut read, mut released) = (0u64, 0u64, 0u64, 0u64); + let mut rng = Rng(seed); + for _ in 0..400 * C + 400 { + match rng.below(4) { + 0 => { + let room = tx.room(&page).unwrap(); + assert_eq!(u64::from(room), u64::from(C) - (written - released), "the room"); + let n = rng.below(room + 1); + let split = rng.below(n + 1); + for (at, len) in [(split, n - split), (0, split)] { + put(&mut region, tx.span(at, len).unwrap(), written + u64::from(at)); + } + assert!(tx.advance(n)); + written += u64::from(n); + } + 1 => { + let _ = tx.publish(&page); + published = written; + } + 2 => { + let ready = rx.ready(&page).unwrap(); + assert_eq!(u64::from(ready), published - read, "the bytes ready"); + let (len, at) = { + let len = rng.below(ready + 1); + (len, rng.below(ready - len + 1)) + }; + assert_eq!(get(®ion, rx.span(at, len).unwrap()), expected(read + u64::from(at), len)); + let n = rng.below(ready + 1); + assert_eq!(get(®ion, rx.span(0, n).unwrap()), expected(read, n)); + assert!(rx.advance(n)); + read += u64::from(n); + } + _ => { + let _ = rx.release(&page); + released = read; + } + } + } + assert!(read > 4 * u64::from(C), "{read} bytes did not go round the lap twice"); + assert!(region[..BASE].iter().chain(®ion[BASE + C as usize..]).all(|&b| b == 0xA5), "a guard was written"); + } + + #[test] + fn the_ring_agrees_with_a_stream_that_never_wraps() { + for seed in 1..=16 { + agrees::<1>(seed); + agrees::<2>(seed); + agrees::<5>(seed); + agrees::<8>(seed); + agrees::<61>(seed); + } + } + + /// A word a peer might write: anything, a position near the lap, or an edge. + fn hostile(rng: &mut Rng) -> u32 { + match rng.below(3) { + 0 => rng.next() as u32, + 1 => rng.below(4 * C + 2), + _ => [0, C - 1, C, C + 1, 2 * C - 1, 2 * C, 2 * C + 1, u32::MAX, u32::MAX - C][rng.below(9) as usize], + } + } + + /// Whatever a producer stores in any of the four words, the consumer holds + /// at most the ring past what it released, every span it is handed lies in + /// the ring, and a tail it does not believe is `TailPastDepth`. + fn hostile_producer(seed: u64) { + let page = page(); + let mut rx = Consumer::new(&page, place::()); + let mut rng = Rng(seed); + let mut held = 0; + for _ in 0..2000 { + page[rng.below(4) as usize].store(hostile::(&mut rng), Ordering::Release); + let seen = match rng.below(4) { + 0 => rx.ready(&page), + 1 => rx.arm(&page), + 2 => { + let n = rng.below(C + 1); + match rx.span(0, n) { + Some(spans) => { + inside::(spans, n); + assert!(rx.advance(n)); + held += n; + } + None => assert!(!rx.advance(n), "advanced over {n} bytes it was never handed"), + } + continue; + } + _ => { + let _ = rx.release(&page); + held = 0; + continue; + } + }; + match seen { + Ok(ready) => assert!( + u64::from(held) + u64::from(ready) <= u64::from(C), + "holds {held} and was told of {ready} more in a ring of {C}" + ), + Err(violation) => assert_eq!(violation, Violation::TailPastDepth), + } + } + } + + #[test] + fn a_hostile_producer_hands_the_consumer_at_most_the_ring() { + for seed in 1..=64 { + hostile_producer::<1>(seed); + hostile_producer::<5>(seed); + hostile_producer::<8>(seed); + } + } + + /// Whatever a consumer stores in any of the four words, the producer + /// writes at most the ring past the head it was shown, every span it is + /// handed lies in the ring, and a head it does not believe is + /// `HeadPastTail`. + fn hostile_consumer(seed: u64) { + let page = page(); + let mut tx = Producer::new(&page, place::()); + let mut rng = Rng(seed); + let mut pending = 0; + for _ in 0..2000 { + page[rng.below(4) as usize].store(hostile::(&mut rng), Ordering::Release); + let seen = match rng.below(4) { + 0 => tx.room(&page), + 1 => tx.arm(&page), + 2 => { + let n = rng.below(C + 1); + match tx.span(0, n) { + Some(spans) => { + inside::(spans, n); + assert!(tx.advance(n)); + pending += n; + } + None => assert!(!tx.advance(n), "advanced over {n} bytes it was never handed"), + } + continue; + } + _ => { + let _ = tx.publish(&page); + pending = 0; + continue; + } + }; + match seen { + Ok(room) => assert!( + u64::from(pending) + u64::from(room) <= u64::from(C), + "{pending} unpublished and {room} more in a ring of {C}" + ), + Err(violation) => assert_eq!(violation, Violation::HeadPastTail), + } + } + } + + #[test] + fn a_hostile_consumer_gives_the_producer_at_most_the_ring() { + for seed in 1..=64 { + hostile_consumer::<1>(seed); + hostile_consumer::<5>(seed); + hostile_consumer::<8>(seed); + } + } + + /// From every position round the lap, twice: a tail the ring ahead is + /// believed, and one a byte further, a word past the lap or the last word + /// is not; a head the ring behind is believed, and one a byte past what was + /// published or a word past the lap is not. + #[test] + fn a_cursor_is_believed_up_to_the_ring_and_no_further() { + const C: u32 = 5; + const LAP: u32 = 2 * C; + let page = page(); + let (mut tx, mut rx) = (Producer::new(&page, place::()), Consumer::new(&page, place::())); + for at in 0..2 * LAP { + let here = at % LAP; + let tails = [ + ((here + C) % LAP, Ok(C)), + ((here + C + 1) % LAP, Err(Violation::TailPastDepth)), + (LAP, Err(Violation::TailPastDepth)), + (u32::MAX, Err(Violation::TailPastDepth)), + ]; + for (tail, believed) in tails { + page[1].store(tail, Ordering::Release); + assert_eq!(rx.ready(&page), believed, "tail {tail} with {here} released"); + } + let heads = [ + ((here + LAP - C) % LAP, Ok(0)), + ((here + 1) % LAP, Err(Violation::HeadPastTail)), + (LAP, Err(Violation::HeadPastTail)), + ]; + for (head, believed) in heads { + page[0].store(head, Ordering::Release); + assert_eq!(tx.room(&page), believed, "head {head} with {here} published"); + } + page[0].store(here, Ordering::Release); + assert_eq!(tx.room(&page), Ok(C), "an honest head at {here}"); + assert!(tx.advance(1)); + let _ = tx.publish(&page); + assert_eq!(rx.ready(&page), Ok(1)); + assert!(rx.advance(1)); + let _ = rx.release(&page); + } + } + + /// An end rings exactly when the peer armed since the last ring, and only + /// when it moved its cursor. + #[test] + fn an_end_rings_only_a_peer_that_armed() { + const C: u32 = 4; + let page = page(); + let (mut tx, mut rx) = (Producer::new(&page, place::()), Consumer::new(&page, place::())); + assert!(tx.advance(1)); + assert!(!tx.publish(&page), "nobody armed"); + assert_eq!(rx.arm(&page), Ok(1), "the look after arming finds the byte"); + assert!(!tx.publish(&page), "nothing new to publish"); + assert!(tx.advance(1)); + assert!(tx.publish(&page), "the consumer armed"); + assert!(tx.advance(2)); + assert!(!tx.publish(&page), "the ring cleared the want"); + assert_eq!(tx.arm(&page), Ok(0), "a full ring"); + assert!(rx.ready(&page) == Ok(4) && rx.advance(2)); + assert!(rx.release(&page), "the producer armed"); + assert!(rx.advance(2)); + assert!(!rx.release(&page), "the ring cleared the want"); + assert!(!rx.release(&page), "nothing new to release"); + } +} diff --git a/lib/blockring/transport/tests/loom.rs b/lib/blockring/transport/tests/loom.rs index 8051dc92a57..8403b9b9a6f 100644 --- a/lib/blockring/transport/tests/loom.rs +++ b/lib/blockring/transport/tests/loom.rs @@ -10,10 +10,12 @@ //! cargo test -p toyos-transport --features --test loom use core::sync::atomic::Ordering; +use std::rc::Rc; +use loom::cell::Cell; use loom::sync::atomic::AtomicU32; -use loom::sync::Arc; -use toyos_transport::{Consumer, Place, Producer, Untrusted, Violation, Word}; +use loom::sync::{Arc, Notify}; +use toyos_transport::{stream, Consumer, Place, Producer, Untrusted, Violation, Word}; /// A loom atomic as a region word: the trait is this crate's and the type is /// loom's, so the two meet through a wrapper. @@ -26,6 +28,9 @@ impl Word for Shared { fn store(&self, value: u32, order: Ordering) { self.0.store(value, order) } + fn swap(&self, value: u32, order: Ordering) -> u32 { + self.0.swap(value, order) + } } const E: usize = 2; @@ -139,3 +144,133 @@ fn a_hostile_consumer_yields_room_or_a_violation() { assert!(pushed <= D, "pushed {pushed} into a ring of {D} nobody released"); }); } + +/// A stream's four words, its bytes, and a doorbell each way. A byte is a loom +/// cell: a copy into or out of it that the cursors do not order after the +/// other end's last copy of it is loom's own `Causality violation`. A doorbell +/// is loom's `Notify`, which keeps one ring and may wake its waiter for +/// nothing, and loom fails a run that ends with a thread waiting on one. It is +/// held in an `Rc`, as loom's threads take no `Send` bound and run on the one +/// thread that runs the model: loom's `Arc` panics again in its drop when that +/// failure unwinds the model, which aborts the run before it says which test +/// failed (`issues/loom-arc-panics-in-its-drop-while-a-failed-model-unwinds.md`). +struct Stream { + page: [Shared; 4], + bytes: [Cell; RING as usize], + to_consumer: Notify, + to_producer: Notify, +} + +const RING: u32 = 1; +const STREAM: stream::Place = stream::Place::new::<0, 1, 2, 3, 0>(); +type StreamEnds = (stream::Producer, stream::Consumer); + +fn stream() -> (Rc, StreamEnds) { + let s = Rc::new(Stream { + page: core::array::from_fn(|_| Shared(AtomicU32::new(0))), + bytes: core::array::from_fn(|_| Cell::new(0)), + to_consumer: Notify::new(), + to_producer: Notify::new(), + }); + let ends = (stream::Producer::new(&s.page, STREAM), stream::Consumer::new(&s.page, STREAM)); + (s, ends) +} + +/// Writes `byte` into the room the producer was last shown and publishes it, +/// ringing a consumer that armed. +fn send(s: &Stream, tx: &mut stream::Producer, byte: u32) { + for span in tx.span(0, 1).expect("a byte of room") { + for at in span.offset..span.offset + span.len { + s.bytes[at].set(byte); + } + } + assert!(tx.advance(1)); + if tx.publish(&s.page) { + s.to_consumer.notify(); + } +} + +/// Reads the byte the consumer was last shown and releases it, ringing a +/// producer that armed. +fn receive(s: &Stream, rx: &mut stream::Consumer) -> u32 { + let mut read = Vec::new(); + for span in rx.span(0, 1).expect("a byte ready") { + read.extend((span.offset..span.offset + span.len).map(|at| s.bytes[at].get())); + } + assert!(rx.advance(1)); + if rx.release(&s.page) { + s.to_producer.notify(); + } + read[0] +} + +/// A consumer that polls reads each byte as the producer wrote it, through a +/// ring that wraps under it: its copy out follows the producer's copy in, and +/// the producer's next copy into that byte follows its copy out. +/// `publish-relaxed` takes the first edge away. +#[test] +fn a_published_byte_is_read_as_written() { + loom::model(|| { + let (s, (mut tx, mut rx)) = stream(); + let producer_side = Rc::clone(&s); + let producer = loom::thread::spawn(move || { + for byte in [0x5A, 0xA5] { + while tx.room(&producer_side.page).expect("the consumer keeps the protocol") == 0 { + loom::thread::yield_now(); + } + send(&producer_side, &mut tx, byte); + } + }); + let mut read = Vec::new(); + while read.len() < 2 { + match rx.ready(&s.page).expect("the producer keeps the protocol") { + 0 => loom::thread::yield_now(), + _ => read.push(receive(&s, &mut rx)), + } + } + producer.join().expect("the producer thread"); + assert_eq!(read, [0x5A, 0xA5], "a byte was read before it was written"); + }); +} + +/// A consumer sleeps whenever the look after arming finds the ring empty, and +/// the byte published around its sleep still reaches it. `want-relaxed` lets +/// it miss the tail stored before the want it read cleared. +#[test] +fn a_consumer_that_looked_again_is_rung_for_bytes_it_missed() { + loom::model(|| { + let (s, (mut tx, mut rx)) = stream(); + let producer_side = Rc::clone(&s); + let producer = loom::thread::spawn(move || send(&producer_side, &mut tx, 0x5A)); + while rx.ready(&s.page).expect("the producer keeps the protocol") == 0 { + if rx.arm(&s.page).expect("the producer keeps the protocol") == 0 { + s.to_consumer.wait(); + } + } + assert_eq!(receive(&s, &mut rx), 0x5A); + producer.join().expect("the producer thread"); + }); +} + +/// A producer sleeps whenever the look after arming finds the ring full, and +/// the room released around its sleep still reaches it. `want-relaxed` lets +/// it miss the head stored before the want it read cleared. +#[test] +fn a_producer_that_looked_again_is_rung_for_room_it_missed() { + loom::model(|| { + let (s, (mut tx, mut rx)) = stream(); + send(&s, &mut tx, 0x5A); + let consumer_side = Rc::clone(&s); + let consumer = loom::thread::spawn(move || { + assert_eq!(rx.ready(&consumer_side.page), Ok(1), "published before the consumer started"); + receive(&consumer_side, &mut rx) + }); + while tx.room(&s.page).expect("the consumer keeps the protocol") == 0 { + if tx.arm(&s.page).expect("the consumer keeps the protocol") == 0 { + s.to_producer.wait(); + } + } + send(&s, &mut tx, 0xA5); + assert_eq!(consumer.join().expect("the consumer thread"), 0x5A); + }); +}