//! A bounded FIFO queue in a ring buffer, with one owner and shared thieves. //! //! Each pool worker's run queue is a [`Queue`] of its runnables. Only the //! holder of its [`Owner`] pushes and pops. Other workers hold the shared //! handle and steal half of an unclaimed backlog with [`Queue::steal_into`]. //! //! # Overview //! //! Work usually stays in the run queue of the worker that queued it. An idle //! worker steals the oldest half of another worker's backlog, and a push into a //! full queue moves the newest half to the pool's global queue. //! //! The queue keeps its values in a ring of [`CAPACITY`] slots, and three //! positions divide the ring. The owner pushes at `tail` and pops at `real`, //! so the values from `real` up to `tail` are queued. A thief claims the oldest //! half of them, rounded down, by moving `real` past that half. It then moves //! the claimed values into its own queue while `steal` still marks where they //! start, and `steal` catches up with `real` once it is done. Every other slot //! is free, and the free slots wrap around the end of the ring. With eight //! slots, while a thief moves `b` and `c`: //! //! ```text //! steal real tail //! v v v //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! | | | b | c | d | e | | | //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! |<--- free ---->|<-- claimed -->|<-- queued --->|<--- free ---->| //! ``` //! //! # Positions //! //! Positions keep counting past the end of the ring and wrap as `u32` values, //! so a position selects its slot modulo [`CAPACITY`]. The head holds `real` //! and the gap `real - steal` in one atomic word, so a claim or a pop moves //! `real` in the same exchange that checks whether a claim is held. The gap is //! at most `tail - steal`, which never exceeds [`CAPACITY`], so it fits in 9 //! bits. The head counts `real` in the 55 bits below them, and the low 32 of //! those are the position the tail and the slots use. The higher bits only //! make the head repeat later (see [Stealing](#stealing)). Below, a head is //! written `(real, gap)`. //! //! ```text //! 63 55 54 32 31 0 //! +---------+--------------------+-------------------+ //! head | gap | real, high bits | real, low bits | AtomicU64 //! +---------+--------------------+-------------------+ //! //! 31 0 //! +-------------------+ //! tail | tail | AtomicU32 //! +-------------------+ //! ``` //! //! The gap is zero, and `steal` equals `real`, unless a claim is held. A second //! thief backs off while one is. Owner pops may move `real` past a held claim, //! widening the gap, so releasing the claim clears the gap and lifts `steal` to //! the current `real`. Only the owner writes `tail`. //! //! # Invariants //! //! The following invariants use wrapping distances, rather than numeric //! comparisons of the indices: //! - `steal <= real <= tail` and `tail - steal <= CAPACITY`. //! - `[real, tail)` contains the queue's initialized values. An overflow //! iterator separately owns the removed newest half until it is exhausted. //! - `steal != real` means one claim holds `[steal, claim_end)`, where //! `claim_end` is the `real` its exchange stored. The owner's pops can move //! `real` beyond `claim_end` while the claim is held. //! - `[tail, steal + CAPACITY)` is free except for a live overflow's slots. //! Only the owner writes free slots, and a release store of `tail` //! publishes their values. //! - Every value is read at most once, and its slot is written again only //! after that read. //! //! # Ordering //! //! A release store of `tail` publishes initialized slots to a thief's //! acquiring tail load. Releasing a claim publishes its completed reads //! through `head`, which the owner acquires before reusing those slots. //! Shared head updates also release the tail publications they observed, so //! acquiring the head before loading the tail keeps that tail from lagging //! behind the head. An overflow can still advance the tail after the head //! snapshot, so a claim refreshes an apparent backlog larger than capacity. //! The owner reads its own tail with relaxed loads, since only it writes it. //! //! # Overflow //! //! A push into a full queue exchanges the head to `(real + CAPACITY, 0)`, whose //! position is the tail, taking every value. It then republishes the oldest //! half by storing `tail + CAPACITY / 2`. //! //! In a full ring buffer, the `CAPACITY / 2` positions after the old tail map //! onto the oldest half's slots. An iterator that borrows the owner takes the //! newest half, and the borrow excludes writes to those slots until the //! iterator is consumed. A push decides from one snapshot of the head whether //! it publishes, overflows, or hands back only the new value because a claim //! is held. It retries when its overflow exchange finds the head moved. //! //! An overflow in a queue of eight slots: //! //! ```text //! before: full, with steal = real = 0 and tail = 8, as the owner pushes i //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! | a | b | c | d | e | f | g | h | //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! |<-------- oldest half -------->|<-------- newest half -------->| //! //! after: head (0, 0) -> (8, 0), then tail 8 -> 12 //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! | a | b | c | d | e | f | g | h | //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! |<- queued, positions 8 to 11 ->|<--- read by the Overflow ---->| //! //! the caller then sends e f g h, and after them i, to the global queue //! ``` //! //! # Stealing //! //! A steal takes half the unclaimed backlog, rounded down, so a lone value //! stays with its owner. It claims the values with one exchange on the head, //! moves them into the thief's own queue, releases the claim, then publishes //! them with one tail store. It goes only into a queue with at least half its //! slots free, so the batch always fits. [`Queue::is_stealable`] is a hint //! about the source's backlog. The claim checks the source again, and the //! steal also checks the destination's room. //! //! A steal from a queue of eight slots into an empty one: //! //! ```text //! before: steal = real = 0, tail = 7 //! 0 1 2 3 4 5 6 7 //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! source | a | b | c | d | e | f | g | | //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! |<---------------------- queued ----------------------->| //! //! 1. the thief claims 3 of 7 (real 0 -> 3), then the owner pops d and e (real 3 -> 5) //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! source | a | b | c | | | f | g | | //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! |<------ claimed ------>|<-- popped --->|<-- queued --->| //! //! 2. the thief copies a b c, releases (steal 0 -> 5), then publishes (its tail 0 -> 3) //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! source | | | | | | f | g | | //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! |<-- queued --->| //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! thief | a | b | c | | | | | | //! +-------+-------+-------+-------+-------+-------+-------+-------+ //! |<------ queued ------->| //! ``` //! //! A steal assumes that a thief does not stall between its head load and its //! claim exchange while the indices advance through a full `2^55` positions, //! after which the head could read as it did and the exchange succeed with a //! stale count. The tail and the slots only need `u32` positions, and the head //! counts `real` over 55 bits so that it repeats only that rarely. Once the //! exchange succeeds the claim pins `steal`, so `tail - steal <= CAPACITY` //! bounds further progress to `CAPACITY` positions. At one position per //! nanosecond, the exposed window would need to last `2^55` nanoseconds, about //! 417 days. An overflow advances `real` by `CAPACITY` and `tail` by half as //! much, so the assumption counts positions rather than operations. A stale //! count can exceed the values then queued, so even a claim of one value could //! read a free slot. //! //! The owner's own exchanges need no such assumption. While the owner stalls it //! pushes nothing, so thieves can move `real` no further than `tail`, and the //! head cannot return to the value the owner loaded. //! //! # Pops without an exchange //! //! A queue built exclusive has no reader but its owner, so the owner pops with //! plain loads and a store rather than an exchange. [`Queue::steal_into`] //! asserts that its source is shared. //! //! The owner of a shared queue also pops a lone value with a store when no //! claim is held. A claim takes half the backlog, rounded down, so no claim //! takes a lone value. Only the owner can push more, and the pop's mutable //! borrow excludes a push. A thief still expecting the pre-pop head needs a //! later tail publication to see two values. That publication follows the head //! store, so acquiring the tail orders the store before the thief's stale //! exchange. The exchange can succeed only if the thief stalled through a full //! `2^55` positions, and [Stealing](#stealing) assumes no thief does. //! //! # Dropping handles //! //! The owner must remove its unclaimed values before dropping. A thief may //! still own claimed values after that. An overflow must be exhausted before //! dropping. A nonempty owner or an unexhausted overflow panics on drop unless //! the thread is already unwinding. These drops do not destroy unread values, //! which may leak. use super::task::UnsafeCell; use commonware_utils::Widen; use crossbeam_utils::CachePadded; use std::{cell::Cell, marker::PhantomData, mem::MaybeUninit, thread}; cfg_if::cfg_if! { if #[cfg(feature = "loom")] { use loom::sync::{Arc, atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}}; } else { use std::sync::{Arc, atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}}; } } /// Number of slots in a queue, small under loom so models reach a full queue. pub const CAPACITY: usize = if cfg!(feature = "loom") { 4 } else { 256 }; /// Half the capacity: what an overflow keeps, as a position count. const HALF: u32 = CAPACITY as u32 / 2; /// Fewest values an unclaimed queue holds for a steal to take any, since a /// steal takes half rounded down. pub const STEALABLE: usize = 2; /// Mask that selects a position's slot. const MASK: usize = CAPACITY - 1; /// Bits of a head that count `real`. The gap `real - steal` fills the rest. const REAL_BITS: u32 = 55; /// Mask that selects a head's `real` count. const REAL_MASK: u64 = (1 << REAL_BITS) - 1; /// Positions select slots by masking, a full queue's distance must fit in a /// wrapping `u32` comparison, and the gap field must hold every gap from zero /// to the capacity. const _: () = assert!(CAPACITY.is_power_of_two() && CAPACITY < 1 << (u64::BITS - REAL_BITS)); /// The low 32 bits of a head's count are its `u32` position, which needs the /// count to be at least that wide. const _: () = assert!(REAL_BITS >= u32::BITS); /// The slots of a queue's ring and the positions that divide them, shared by /// its [`Queue`] and [`Owner`] handles. See the module docs for the invariants /// every access keeps. struct Inner { /// `gap << REAL_BITS | real`: the next position the owner pops, counted /// over [`REAL_BITS`] bits, and how far it lies past the oldest unreleased /// position. head: AtomicU64, /// One past the newest published value, written only by the owner. tail: AtomicU32, /// Whether the unique owner has been taken. owned: AtomicBool, /// Whether no reader but the owner ever touches the queue, fixed at /// construction. exclusive: bool, /// Storage for unclaimed, claimed, and overflow values. A pop owns its /// slot from the head update until its read completes. slots: Box<[UnsafeCell>]>, } // SAFETY: the module's invariants give each slot one writer and one reader, // and the head and tail orderings transfer values between threads without // overlap. unsafe impl Sync for Inner {} // SAFETY: the queue owns its values and each value can move between threads. unsafe impl Send for Inner {} /// Shared handle of a queue, through which thieves steal and from which its /// owner is taken. pub struct Queue( /// Shared indices and storage, kept alive by every queue and owner handle. Arc>>, ); /// Unique owner of a queue, movable between threads but never shared. pub struct Owner( /// The same indices and storage reached by the shared queue handle. Arc>>, /// Prevent sharing the owner while permitting it to move between threads. PhantomData>, ); /// Result of publishing a value into a queue. pub enum Push<'a, T> { /// The value was published. Pushed( /// Available length computed from the push's head snapshot. A /// concurrent claim can make it an overestimate. usize, ), /// The queue was full, so the newest half and the new value must be /// published elsewhere, in that order. Overflow( /// Removed newest half, which must be exhausted. Overflow<'a, T>, /// New value that was not inserted. T, ), /// No slot was free while a thief held a claim, so only the new value must /// move elsewhere. Claimed( /// New value that was not inserted. T, ), } /// Newest half of a full queue, yielded oldest first. /// /// The iterator must be exhausted before it is dropped. Its mutable borrow /// excludes owner writes until every removed slot has been read. pub struct Overflow<'a, T> { /// The queue's owner, borrowed until the batch is consumed. owner: &'a mut Owner, /// Next position to read. next: u32, /// One past the last position to read. end: u32, } /// Split a head into `real`, counted over 55 bits, and the gap `real - steal`. /// The low 32 bits of `real` are its position. const fn unpack(head: u64) -> (u64, u32) { (head & REAL_MASK, (head >> REAL_BITS) as u32) } /// Pack `real`, reduced to 55 bits, and the gap `real - steal` into a head. const fn pack(real: u64, gap: u32) -> u64 { (gap as u64) << REAL_BITS | (real & REAL_MASK) } /// The position of `steal`, the oldest unreleased value, from an unpacked /// head. const fn steal(real: u64, gap: u32) -> u32 { (real as u32).wrapping_sub(gap) } impl Queue { /// An empty queue that thieves may steal from. pub fn shared() -> Self { Self::at(0, false) } /// An empty queue that no thief may steal from. pub fn exclusive() -> Self { Self::at(0, true) } /// An empty queue whose head starts at a chosen 55-bit count, so its /// position is that count's low 32 bits. fn at(real: u64, exclusive: bool) -> Self { assert!(real <= REAL_MASK, "start must fit the head's count"); Self(Arc::new(CachePadded::new(Inner { head: AtomicU64::new(pack(real, 0)), tail: AtomicU32::new(real as u32), owned: AtomicBool::new(false), exclusive, slots: (0..CAPACITY) .map(|_| UnsafeCell::new(MaybeUninit::uninit())) .collect(), }))) } /// Take the owner, panicking if it was already taken, even if it has since /// been dropped. pub fn owner(&self) -> Owner { assert!( !self.0.owned.swap(true, Ordering::AcqRel), "queue owner must be taken at most once" ); Owner(self.0.clone(), PhantomData) } /// Whether a snapshot shows no claim and at least [`STEALABLE`] values. /// Concurrent changes can make this hint stale before a steal claims them. pub fn is_stealable(&self) -> bool { // Shared head updates release the tail they observed. Acquire the // head first so the tail cannot lag behind it and turn the wrapping // distance into a false backlog. let (real, gap) = unpack(self.0.head.load(Ordering::Acquire)); let tail = self.0.tail.load(Ordering::Acquire); gap == 0 && tail.wrapping_sub(real as u32) >= STEALABLE as u32 } /// Move half the unclaimed backlog, rounded down, into `into`, returning /// how many values moved. /// /// Moves none when the backlog holds fewer than [`STEALABLE`] values, while /// another thief holds a claim, or when `into` has fewer than half its /// slots free. The claim is released before the values are published in /// `into` with one tail store. Panics if `into` belongs to this queue or /// this queue is exclusive. #[must_use] pub fn steal_into(&self, into: &mut Owner) -> usize { assert!(!self.0.exclusive, "steal source must be a shared queue"); assert!( !Arc::ptr_eq(&into.0, &self.0), "steal destination must differ from the source queue" ); // Only this owner writes the destination tail. Acquiring its head // observes completed claim reads before their slots are reused. let tail = into.0.tail.load(Ordering::Relaxed); let (real, gap) = unpack(into.0.head.load(Ordering::Acquire)); if tail.wrapping_sub(steal(real, gap)) > HALF { return 0; } let Some((first, count)) = self.claim() else { return 0; }; for offset in 0..count { // SAFETY: the source claim owns this initialized slot. let value = unsafe { self.0.read((first as u32).wrapping_add(offset)) }; // SAFETY: the destination had half its slots free before the // claim, the owner is borrowed, and thieves only add room. unsafe { into.0.write(tail.wrapping_add(offset), value) }; } self.release_claim(first, count); into.0 .tail .store(tail.wrapping_add(count), Ordering::Release); Widen::widen(count) } /// Claim half the published backlog, rounded down, returning the first /// claimed position as a 55-bit count and the count claimed, or `None` /// while another claim is held or fewer than [`STEALABLE`] values are /// published. Every bound is checked before the exchange, so the caller /// can move and release the claimed values without further bound checks. fn claim(&self) -> Option<(u64, u32)> { let mut head = self.0.head.load(Ordering::Acquire); loop { let (real, gap) = unpack(head); if gap != 0 { return None; } // Acquire the owner's release publication before reading any // slot that a successful claim grants us. let tail = self.0.tail.load(Ordering::Acquire); let available = tail.wrapping_sub(real as u32); // An overflow can pair an old head with the republished tail. That // head cannot be claimed, so refresh it before checking bounds. if available > CAPACITY as u32 { head = self.0.head.load(Ordering::Acquire); continue; } let count = available / 2; if count == 0 { return None; } assert!(count <= HALF, "claim must not exceed half the queue"); let claimed = pack(real.wrapping_add(u64::from(count)), count); // Release carries the observed tail to later acquiring head // readers, preserving their head-before-tail snapshot order. match self .0 .head .compare_exchange(head, claimed, Ordering::AcqRel, Ordering::Acquire) { Ok(_) => return Some((real, count)), Err(actual) => head = actual, } } } /// Release a claim once its slots have been read, clearing the gap to lift /// `steal` to the current `real`. Owner pops may have moved `real` past the /// claim. The caller must hold exactly this claim, starting at the 55-bit /// count `first`, and have read every claimed slot. fn release_claim(&self, first: u64, count: u32) { let mut head = pack(first.wrapping_add(u64::from(count)), count); loop { let (real, _) = unpack(head); // Release lets an owner that acquires this head reuse the read // slots. match self.0.head.compare_exchange( head, pack(real, 0), Ordering::AcqRel, Ordering::Acquire, ) { Ok(_) => return, Err(actual) => head = actual, } } } } impl Owner { /// Values available to the owner, excluding a pending claim. pub fn len(&self) -> usize { let (real, _) = unpack(self.0.head.load(Ordering::Acquire)); let tail = self.0.tail.load(Ordering::Relaxed); Widen::widen(tail.wrapping_sub(real as u32)) } /// Slots free for publication. A pending claim holds its slots, and those /// the owner popped past it, until it is released. pub fn room(&self) -> usize { let (real, gap) = unpack(self.0.head.load(Ordering::Acquire)); let tail = self.0.tail.load(Ordering::Relaxed); CAPACITY - Widen::widen(tail.wrapping_sub(steal(real, gap))) } /// Whether the owner has no available values. pub fn is_empty(&self) -> bool { self.len() == 0 } /// Publish a value, or hand back what a full queue cannot hold, as /// [`Push`] describes. /// /// Every attempt decides from one head snapshot. A failed overflow /// exchange retries publication, since a released claim can make room. pub fn push(&mut self, value: T) -> Push<'_, T> { self.push_with(value, || {}) } /// Publish as [`Self::push`] does, running `after_snapshot` after each /// attempt's head snapshot, where a test pauses the owner. fn push_with(&mut self, value: T, mut after_snapshot: impl FnMut()) -> Push<'_, T> { loop { // Acquire completed claim reads before writing a freed slot. let head = self.0.head.load(Ordering::Acquire); let (real, gap) = unpack(head); let tail = self.0.tail.load(Ordering::Relaxed); after_snapshot(); if tail.wrapping_sub(steal(real, gap)) < CAPACITY as u32 { // SAFETY: the acquired head excludes pending readers, and the // unpublished tail slot is free. unsafe { self.0.write(tail, value) }; let tail = tail.wrapping_add(1); self.0.tail.store(tail, Ordering::Release); return Push::Pushed(Widen::widen(tail.wrapping_sub(real as u32))); } if gap != 0 { return Push::Claimed(value); } assert_eq!( tail.wrapping_sub(real as u32), CAPACITY as u32, "queue must be full to overflow" ); // The exchange arbitrates against a claim and releases the // owner's preceding tail publications to later head readers. The // queue is full, so `real` moves to the tail. if self .0 .head .compare_exchange( head, pack(real.wrapping_add(Widen::widen(CAPACITY)), 0), Ordering::AcqRel, Ordering::Acquire, ) .is_err() { continue; } // The `HALF` positions after the old tail map onto the oldest // half's slots, so publishing them republishes that half. self.0 .tail .store(tail.wrapping_add(HALF), Ordering::Release); return Push::Overflow( Overflow { owner: self, next: (real as u32).wrapping_add(HALF), end: tail, }, value, ); } } /// Take the oldest value available to the owner. #[must_use] #[inline] pub fn pop(&mut self) -> Option { if self.0.exclusive { // No other reader moves the head, so plain accesses suffice. let (real, _) = unpack(self.0.head.load(Ordering::Relaxed)); let tail = self.0.tail.load(Ordering::Relaxed); if real as u32 == tail { return None; } self.0 .head .store(pack(real.wrapping_add(1), 0), Ordering::Relaxed); // SAFETY: an exclusive queue has no reader but its owner, so this // initialized slot is the owner's alone, and the mutable borrow // excludes writes. return Some(unsafe { self.0.read(real as u32) }); } let mut head = self.0.head.load(Ordering::Acquire); loop { let (real, gap) = unpack(head); let position = real as u32; let tail = self.0.tail.load(Ordering::Relaxed); if position == tail { return None; } // No claim is held, and none can take a lone value, so a store // pops it. Release orders this head after the tail stores before // it, as the exchange does, so a thief that acquires it loads a // tail past the popped position. if gap == 0 && tail.wrapping_sub(position) == 1 { self.0 .head .store(pack(real.wrapping_add(1), 0), Ordering::Release); // SAFETY: no claim holds or can take this initialized slot, // and the mutable borrow excludes writes. return Some(unsafe { self.0.read(position) }); } // A held claim keeps `steal` in place, so the gap grows with // `real`. let next = pack(real.wrapping_add(1), if gap == 0 { 0 } else { gap + 1 }); // Release carries the owner's tail stores to thieves that acquire // this head before loading the tail. match self .0 .head .compare_exchange(head, next, Ordering::AcqRel, Ordering::Acquire) { Ok(_) => { // SAFETY: the exchange gave the owner this initialized // slot, and the mutable borrow excludes writes. return Some(unsafe { self.0.read(position) }); } Err(actual) => head = actual, } } } /// Publish a batch with one tail store. Panics if it exceeds /// [`Self::room`] or the iterator misreports its length. pub fn push_batch(&mut self, mut values: impl ExactSizeIterator) { let count = values.len(); assert!(count <= self.room(), "batch must fit the queue's room"); let tail = self.0.tail.load(Ordering::Relaxed); for offset in 0..count { let value = values .next() .expect("batch must yield at least its reported length"); // SAFETY: room bounds this unpublished span, the owner is // borrowed, and pending claims cannot overlap it. unsafe { self.0.write(tail.wrapping_add(offset as u32), value) }; } assert!( values.next().is_none(), "batch must yield at most its reported length" ); self.0 .tail .store(tail.wrapping_add(count as u32), Ordering::Release); } } impl Inner { /// Read a value from an exclusively owned initialized slot. /// /// # Safety /// /// The caller must own `position` through a successful pop, a claim, or an /// overflow exchange, and exclude writes until the read completes. unsafe fn read(&self, position: u32) -> T { self.slots[Widen::widen(position) & MASK].with_mut(|ptr| { // SAFETY: the caller owns this initialized slot by contract. unsafe { (*ptr).assume_init_read() } }) } /// Initialize a free slot before it is published. /// /// # Safety /// /// The caller must be the owner and must have established room at /// `position`, excluding pending readers until the write completes. unsafe fn write(&self, position: u32, value: T) { self.slots[Widen::widen(position) & MASK].with_mut(|ptr| { // SAFETY: the caller owns this free slot by contract. unsafe { ptr.write(MaybeUninit::new(value)) }; }); } } impl Iterator for Overflow<'_, T> { type Item = T; fn next(&mut self) -> Option { if self.next == self.end { return None; } let position = self.next; self.next = self.next.wrapping_add(1); // SAFETY: the overflow exchange gave this iterator the newest half. // Its owner borrow excludes writes, and the republished oldest half // occupies disjoint slots. Some(unsafe { self.owner.0.read(position) }) } fn size_hint(&self) -> (usize, Option) { let count = Widen::widen(self.end.wrapping_sub(self.next)); (count, Some(count)) } } impl ExactSizeIterator for Overflow<'_, T> {} impl Drop for Overflow<'_, T> { fn drop(&mut self) { // An unread value would leak, so an unexhausted overflow panics on // drop. During an unwind the values leak silently, since a second // panic would abort. if !thread::panicking() { assert_eq!( self.next, self.end, "overflow must be exhausted before it drops" ); } } } impl Drop for Owner { fn drop(&mut self) { // Unclaimed values must be removed before the owner drops. Catch an // undrained queue on its owner's thread rather than wherever the last // shared handle drops. Pending claims belong to their thieves. if !thread::panicking() { assert!(self.is_empty(), "owner must be empty when it drops"); } } } impl Drop for Inner { fn drop(&mut self) { // A claim or value still here when the last handle drops was leaked. if !thread::panicking() { let (real, gap) = unpack(self.head.load(Ordering::Relaxed)); assert_eq!(gap, 0, "queue must have no pending claim when it drops"); assert_eq!( real as u32, self.tail.load(Ordering::Relaxed), "queue must be empty when it drops" ); } } } #[cfg(test)] #[commonware_macros::test_group("miri")] pub mod tests { use super::*; use std::{ panic::{AssertUnwindSafe, catch_unwind}, sync::{Barrier, atomic::AtomicUsize, mpsc}, }; /// Claims held by a test in place of a paused thief. impl Queue { /// Claim the oldest `count` values as a thief paused after its claim /// exchange would hold them. Panics unless no claim is held and `count` /// is positive, at most [`HALF`], and within the available values, or if /// another head update wins the exchange. The caller must read and release /// the returned claim exactly once through [`Self::finish_claim`]. pub(crate) fn hold_claim(&self, count: u32) -> (u64, u32) { let head = self.0.head.load(Ordering::Acquire); let (real, gap) = unpack(head); let tail = self.0.tail.load(Ordering::Acquire); assert_eq!(gap, 0); assert!(count > 0 && count <= HALF && count <= tail.wrapping_sub(real as u32)); assert!( self.0 .head .compare_exchange( head, pack(real + u64::from(count), count), Ordering::AcqRel, Ordering::Acquire ) .is_ok() ); (real, count) } /// Read a held claim's values and release it, as the paused thief would. /// The caller must own the unread claim returned by [`Self::hold_claim`]. pub(crate) fn finish_claim(&self, claim: (u64, u32)) -> Vec { let (first, count) = claim; let values = (0..count) .map(|offset| { // SAFETY: hold_claim gave the caller exclusive ownership of this // initialized slot, and the release follows every read. unsafe { self.0.read((first as u32).wrapping_add(offset)) } }) .collect(); self.release_claim(first, count); values } } /// Push a value that must fit. fn push(owner: &mut Owner, value: T) { assert!(matches!(owner.push(value), Push::Pushed(_))); } /// Pop every value the owner holds. fn drain(owner: &mut Owner) -> Vec { std::iter::from_fn(|| owner.pop()).collect() } /// Check the physical head, including the count's bits above its position. fn assert_head(queue: &Queue, real: u64, gap: u32) { assert_eq!( queue.0.head.load(Ordering::Acquire), (u64::from(gap) << REAL_BITS) | (real & REAL_MASK) ); } /// Take every value from a queue whose owner is gone. fn take_unowned(queue: &Queue) -> Vec { // A held claim would belong to its thief, so none may be pending. let (real, gap) = unpack(queue.0.head.load(Ordering::Acquire)); assert_eq!(gap, 0); let tail = queue.0.tail.load(Ordering::Acquire); let count = tail.wrapping_sub(real as u32); let values = (0..count) .map(|offset| { // SAFETY: the owner is gone and no claim is held, so this test // is the only reader of the queue's initialized values. unsafe { queue.0.read((real as u32).wrapping_add(offset)) } }) .collect(); // Moving the head to the tail marks the values taken, so the queue drops empty. queue .0 .head .store(pack(real + u64::from(count), 0), Ordering::Release); values } /// Counts its drops, so a value read twice or never shows. struct Counted( /// Identity used to check that every published value is returned. usize, /// Shared count of destroyed values. Arc, ); impl Drop for Counted { fn drop(&mut self) { self.1.fetch_add(1, Ordering::Relaxed); } } /// Head counts whose positions wrap one past them: at `u32::MAX` the /// position wraps and carries into the head's higher bits, and at the /// 55-bit maximum the whole count wraps to zero as well. const WRAPS: [u64; 2] = [u32::MAX as u64, REAL_MASK]; /// Both queue kinds hold `CAPACITY` values in FIFO order, pushed one at a /// time or as a batch, and interleaved pops and pushes keep the order. #[test] fn test_fifo_up_to_capacity() { for queue in [Queue::shared(), Queue::exclusive()] { // A new queue is empty with every slot free. let mut owner = queue.owner(); assert!(owner.is_empty()); assert_eq!(owner.room(), CAPACITY); // Each push reports the length including the new value, up to capacity. for value in 0..CAPACITY { match owner.push(value) { Push::Pushed(len) => assert_eq!(len, value + 1), _ => panic!("push below capacity must publish"), } } // The full queue has no room, and pops return the values in push order. assert_eq!(owner.len(), CAPACITY); assert_eq!(owner.room(), 0); assert_eq!(drain(&mut owner), (0..CAPACITY).collect::>()); // A batch fills the same capacity and drains in the same order. owner.push_batch(0..CAPACITY); assert_eq!(drain(&mut owner), (0..CAPACITY).collect::>()); // Pops and pushes interleave in order. push(&mut owner, 0); push(&mut owner, 1); assert_eq!(owner.pop(), Some(0)); push(&mut owner, 2); assert_eq!(drain(&mut owner), [1, 2]); } } /// Pushes, pops and an overflow keep their order while positions wrap /// past `u32::MAX`, and while the head's count wraps past its 55 bits. #[test] fn test_positions_wrap_across_the_maximum() { for (end, exclusive) in WRAPS .into_iter() .flat_map(|end| [(end, false), (end, true)]) { // Start eight positions before the wrap, so a full ring wraps past it. let queue = Queue::at(end - 7, exclusive); let mut owner = queue.owner(); // A full batch wraps past the maximum and drains in order. owner.push_batch(0..CAPACITY); assert_eq!(drain(&mut owner), (0..CAPACITY).collect::>()); // Refill so the next push overflows with positions already wrapped. owner.push_batch(0..CAPACITY); match owner.push(CAPACITY) { Push::Overflow(batch, value) => { assert_eq!( batch.collect::>(), (CAPACITY / 2..CAPACITY).collect::>() ); assert_eq!(value, CAPACITY); } _ => panic!("full wrapping queue must overflow"), } // The oldest half stays queued in order. assert_eq!(drain(&mut owner), (0..CAPACITY / 2).collect::>()); } } /// A shared owner pops a lone value with a store, here at zero and across /// both wraps, leaving the queue empty and unclaimed. #[test] fn test_lone_value_pops_across_the_maximum() { for start in [0, u64::from(u32::MAX), REAL_MASK] { let queue = Queue::at(start, false); let mut owner = queue.owner(); // A lone value is not stealable, since a steal takes half rounded down. push(&mut owner, 7); assert!(!queue.is_stealable()); // The pop advances the head's count with a zero gap, so no claim // appears held. assert_eq!(owner.pop(), Some(7)); let next = (start + 1) & REAL_MASK; assert_eq!(unpack(queue.0.head.load(Ordering::Acquire)), (next, 0)); assert!(!queue.is_stealable()); assert!(owner.pop().is_none()); } } /// An overflow hands out the newest half, then the new value, and keeps /// the oldest half with room for as many more. #[test] fn test_overflow_keeps_the_oldest_half() { for queue in [Queue::shared(), Queue::exclusive()] { let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); // A push into the full queue overflows. match owner.push(CAPACITY) { Push::Overflow(mut batch, value) => { // The batch reports its exact length as it is consumed. assert_eq!(batch.len(), CAPACITY / 2); assert_eq!(batch.next(), Some(CAPACITY / 2)); assert_eq!(batch.len(), CAPACITY / 2 - 1); // The newest half comes out oldest first, followed by the new value. let mut exported = vec![CAPACITY / 2]; exported.extend(batch); exported.push(value); assert_eq!(exported, (CAPACITY / 2..=CAPACITY).collect::>()); } _ => panic!("full queue must overflow"), } // The oldest half stays queued, with room for as many more. assert_eq!(owner.len(), CAPACITY / 2); assert_eq!(owner.room(), CAPACITY / 2); // New values fill the freed half and queue behind the oldest half. owner.push_batch(CAPACITY + 1..CAPACITY + 1 + CAPACITY / 2); let mut expected = (0..CAPACITY / 2).collect::>(); expected.extend(CAPACITY + 1..CAPACITY + 1 + CAPACITY / 2); assert_eq!(drain(&mut owner), expected); } } /// A steal moves half a backlog between queues whose positions wrap past /// `u32::MAX`, or whose head counts wrap past their 55 bits. #[test] fn test_steal_wraps_across_the_maximum() { for end in WRAPS { // Both queues start near the wrap, so the source's backlog and the // destination's slots wrap past it. let source = Queue::at(end - 7, false); let destination = Queue::at(end - 3, false); let mut owner = source.owner(); let mut into = destination.owner(); owner.push_batch(0..CAPACITY); // The steal takes the oldest half across the wrap. assert_eq!(source.steal_into(&mut into), CAPACITY / 2); // The stolen half, then the owner's remaining half, restore the // push order. let mut values = drain(&mut into); values.extend(drain(&mut owner)); assert_eq!(values, (0..CAPACITY).collect::>()); } } /// A claim held while the owner pops past it and across a wrap keeps its /// slots, and its release frees every one. #[test] fn test_claim_and_pops_across_the_maximum() { for end in WRAPS { // The claim ends before the wrap, and the owner's half crosses it. let queue = Queue::at(end - u64::from(HALF) - 7, false); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); let claim = queue.hold_claim(HALF); // The claim leaves the owner the newest half and no room, so a push // hands back only the new value. assert_eq!(owner.len(), CAPACITY / 2); assert_eq!(owner.room(), 0); assert!(matches!(owner.push(CAPACITY), Push::Claimed(value) if value == CAPACITY)); // The exchange pops past the claim carry the count across the wrap // and widen the gap to the whole capacity, so the room stays zero // and no steal is offered. let popped = drain(&mut owner); assert_eq!(popped, (CAPACITY / 2..CAPACITY).collect::>()); assert_head(&queue, end + u64::from(HALF) - 7, CAPACITY as u32); assert_eq!(owner.room(), 0); assert!(!queue.is_stealable()); // Releasing the claim frees every slot, and the claim and the pops // together hand out every value in push order. let mut seen = queue.finish_claim(claim); assert_eq!(owner.room(), CAPACITY); seen.extend(popped); assert_eq!(seen, (0..CAPACITY).collect::>()); } } /// Every head update retains the count above its position and carries /// across both wraps, including a claim release after an owner pop. #[test] fn test_every_head_update_preserves_the_widened_count() { for start in [ u64::from(u32::MAX) - 1, u64::from(u32::MAX), (1u64 << 32) + 7, REAL_MASK - 1, REAL_MASK, ] { for exclusive in [false, true] { // Two values exercise an exchange or exclusive pop, then a // lone-value or exclusive store. let queue = Queue::at(start, exclusive); let mut owner = queue.owner(); owner.push_batch(0..2); // Each pop advances the full count, even when the position wraps. assert_eq!(owner.pop(), Some(0)); assert_head(&queue, start + 1, 0); assert_eq!(owner.pop(), Some(1)); assert_head(&queue, start + 2, 0); } for exclusive in [false, true] { // Fill a fresh queue so the overflow itself crosses the wrap. let queue = Queue::at(start, exclusive); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); // Overflow advances the full count by the capacity. match owner.push(CAPACITY) { Push::Overflow(batch, value) => { assert_eq!( batch.collect::>(), (CAPACITY / 2..CAPACITY).collect::>() ); assert_eq!(value, CAPACITY); } _ => panic!("full queue must overflow"), } assert_head(&queue, start + Widen::widen(CAPACITY), 0); // The retained oldest half still drains in order. assert_eq!(drain(&mut owner), (0..CAPACITY / 2).collect::>()); } for pop_past_claim in [false, true] { // A real claim advances across the wrap and pins its source slots. let queue = Queue::at(start, false); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); let claim = queue.claim().expect("full queue must be claimable"); assert_eq!(claim, (start, HALF)); assert_head(&queue, start + u64::from(HALF), HALF); // An owner pop changes the head that release first expects. let mut seen = Vec::new(); let advance = if pop_past_claim { seen.push(owner.pop().expect("owner must retain half the values")); assert_head(&queue, start + u64::from(HALF) + 1, HALF + 1); u64::from(HALF) + 1 } else { u64::from(HALF) }; // Both release paths clear the gap while retaining the full count. seen.extend(queue.finish_claim(claim)); assert_head(&queue, start + advance, 0); // The claim and owner return every value exactly once. seen.extend(drain(&mut owner)); seen.sort_unstable(); assert_eq!(seen, (0..CAPACITY).collect::>()); } } } /// The head encoding repeats only after a full `2^55` positions: the head /// a stalled thief loaded differs from the head after any shorter advance, /// `2^32` included, so its claim exchange would fail. #[test] fn test_stale_head_matches_only_after_a_full_period() { // The head a thief loads from a queue holding two values, starting at // a count. let head = |start: u64| { let queue = Queue::at(start & REAL_MASK, false); let mut owner = queue.owner(); owner.push_batch(0..2); let head = queue.0.head.load(Ordering::Acquire); drain(&mut owner); head }; let stale = head(3); // Shorter advances leave a different head, so the stale exchange fails. for count in [1, Widen::widen(CAPACITY), 1 << 32, 1 << 54, REAL_MASK] { assert_ne!(head(3 + count), stale); } // A full period brings the same head back. assert_eq!(head(3 + REAL_MASK + 1), stale); } /// A steal takes half the backlog rounded down, so a lone value is never /// stolen, and `is_stealable` agrees. #[test] fn test_steal_takes_half_rounded_down() { for available in [1usize, 2, 3, 7, 8, 255, 256] { let source = Queue::shared(); let destination = Queue::shared(); let mut owner = source.owner(); let mut into = destination.owner(); owner.push_batch(0..available); // The hint agrees with the steal, which takes half the backlog rounded down. let count = available / 2; assert_eq!(source.is_stealable(), count != 0); assert_eq!(source.steal_into(&mut into), count); // The thief holds the oldest values, and the owner keeps the rest in order. assert_eq!(drain(&mut into), (0..count).collect::>()); assert_eq!(drain(&mut owner), (count..available).collect::>()); } } /// A steal refuses a destination whose room a held claim pins at zero, /// even once its owner is empty, and proceeds after the release. #[test] fn test_steal_room_counts_a_pending_claim() { let source = Queue::shared(); let destination = Queue::shared(); let mut owner = source.owner(); let mut into = destination.owner(); owner.push_batch(CAPACITY..2 * CAPACITY); into.push_batch(0..CAPACITY); // A thief claims half the full destination, and its owner pops the rest. let claim = destination.hold_claim(HALF); let mut seen = drain(&mut into); // The held claim still pins every slot, so the empty destination has no // room and a steal into it moves nothing. assert!(into.is_empty()); assert_eq!(into.room(), 0); assert!(!destination.is_stealable()); assert_eq!(source.steal_into(&mut into), 0); assert_eq!(owner.len(), CAPACITY); // Releasing the claim frees every slot, so the steal proceeds. seen.extend(destination.finish_claim(claim)); assert_eq!(into.room(), CAPACITY); assert_eq!(source.steal_into(&mut into), CAPACITY / 2); // Every value from both queues is handed out exactly once. seen.extend(drain(&mut into)); seen.extend(drain(&mut owner)); seen.sort_unstable(); assert_eq!(seen, (0..2 * CAPACITY).collect::>()); } /// A steal fills a destination holding exactly half its capacity and /// refuses one holding one value more. #[test] fn test_steal_needs_half_the_destination_free() { for (held, moved) in [ (Widen::widen(HALF), CAPACITY / 2), (Widen::widen(HALF) + 1, 0), ] { // Fill the source, and give the destination `held` values. let source = Queue::shared(); let destination = Queue::shared(); let mut owner = source.owner(); let mut into = destination.owner(); owner.push_batch(0..CAPACITY); into.push_batch(CAPACITY..CAPACITY + held); // The steal moves half the source only if half the destination is free. assert_eq!(source.steal_into(&mut into), moved); // Every value from both queues is handed out exactly once. let mut seen = drain(&mut into); seen.extend(drain(&mut owner)); seen.sort_unstable(); assert_eq!(seen, (0..CAPACITY + held).collect::>()); } } /// A push into a full queue under a claim hands back only the new value. #[test] fn test_full_queue_with_a_claim_moves_only_the_new_value() { let queue = Queue::shared(); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); // A claim held in the full queue rules out an overflow, so the push // returns only the new value, and a second thief sees the claim. let claim = queue.hold_claim(HALF); assert!(matches!(owner.push(CAPACITY), Push::Claimed(value) if value == CAPACITY)); assert!(!queue.is_stealable()); // The claim and the owner together hold every queued value, in push order. let mut seen = queue.finish_claim(claim); seen.extend(drain(&mut owner)); assert_eq!(seen, (0..CAPACITY).collect::>()); } /// A steal takes nothing while another thief holds a claim. #[test] fn test_second_thief_backs_off_a_pending_claim() { let source = Queue::shared(); let destination = Queue::shared(); let mut owner = source.owner(); let mut into = destination.owner(); owner.push_batch(0..CAPACITY); // While one thief holds a claim, a second steal moves nothing. let claim = source.hold_claim(HALF); assert_eq!(source.steal_into(&mut into), 0); // The claim and the owner together hold every value, in push order. let mut seen = source.finish_claim(claim); seen.extend(drain(&mut owner)); assert_eq!(seen, (0..CAPACITY).collect::>()); } /// A steal's claim records its count in the head, so until the release /// the owner has no room, a push into the full queue hands back only the /// new value, and a second claim backs off. #[test] fn test_claim_records_its_count_until_released() { let queue = Queue::shared(); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); // The claim takes the oldest half and leaves the owner the newest half // with no room, since the claim still holds its slots. let claim = queue.claim().expect("full queue must be claimable"); assert_eq!(claim, (0, HALF)); assert_eq!(owner.len(), CAPACITY / 2); assert_eq!(owner.room(), 0); // While the claim is held, a second claim backs off, the hint agrees, // and a push hands back only the new value. assert!(queue.claim().is_none()); assert!(!queue.is_stealable()); assert!(matches!(owner.push(CAPACITY), Push::Claimed(value) if value == CAPACITY)); // Reading and releasing the claim frees its slots. let mut seen = queue.finish_claim(claim); assert_eq!(owner.room(), CAPACITY / 2); // The claim and the owner together hold every value, in push order. seen.extend(drain(&mut owner)); assert_eq!(seen, (0..CAPACITY).collect::>()); } /// A held claim does not block a push while the queue has a free slot. #[test] fn test_pending_claim_with_room_allows_a_push() { let queue = Queue::shared(); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY - 1); // A claim on half the capacity leaves the last slot free, so the push // publishes and reports only the unclaimed length. let claim = queue.hold_claim(HALF); assert!(matches!(owner.push(CAPACITY - 1), Push::Pushed(len) if len == CAPACITY / 2)); // The claim and the owner together hold every value, in push order. let mut seen = queue.finish_claim(claim); seen.extend(drain(&mut owner)); assert_eq!(seen, (0..CAPACITY).collect::>()); } /// A claim taken after a full push's snapshot fails the overflow exchange, /// and the retry hands back only the new value. #[test] fn test_claim_between_push_snapshot_and_exchange_retries() { let queue = Queue::shared(); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); // Take a claim after the first attempt's snapshot, so its overflow // exchange fails and the retry sees the claim. let mut claim = None; let mut attempts = 0; match owner.push_with(CAPACITY, || { attempts += 1; if claim.is_none() { claim = Some(queue.hold_claim(HALF)); } }) { Push::Claimed(value) => assert_eq!(value, CAPACITY), _ => panic!("retry must hand back only the new value"), } // Exactly one retry follows the failed exchange. assert_eq!(attempts, 2); // The claim and the owner together hold every value, in push order. let mut seen = queue.finish_claim(claim.unwrap()); seen.extend(drain(&mut owner)); assert_eq!(seen, (0..CAPACITY).collect::>()); } /// A steal that completes between a push's snapshot of a full queue and its /// decision makes room, so the push retries and publishes rather than /// overflowing from a stale snapshot. #[test] fn test_completed_steal_between_push_snapshot_and_exchange() { let queue = Queue::shared(); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); // The thief steals only once signaled, then reports that its steal is done. let (start_tx, start_rx) = mpsc::channel(); let (done_tx, done_rx) = mpsc::channel(); let thief = thread::spawn(move || { let destination = Queue::shared(); let mut into = destination.owner(); // Wait for the push's snapshot, then take half the full queue. start_rx.recv().unwrap(); assert_eq!(queue.steal_into(&mut into), CAPACITY / 2); let values = drain(&mut into); done_tx.send(()).unwrap(); values }); // Pause the first attempt after its snapshot until the steal completes. let mut first = true; let result = owner.push_with(CAPACITY, || { if first { first = false; start_tx.send(()).unwrap(); done_rx.recv().unwrap(); } }); // The stale overflow exchange fails, and the retry finds the freed half // and publishes. assert!(matches!(result, Push::Pushed(len) if len == CAPACITY / 2 + 1)); drop(result); // The thief and the owner together hold every value once. let mut seen = thief.join().unwrap(); seen.extend(drain(&mut owner)); seen.sort_unstable(); assert_eq!(seen, (0..=CAPACITY).collect::>()); } /// A batch beyond the room panics and publishes nothing, whether queued /// values or a pending claim take the room. #[test] fn test_batch_asserts_room() { let queue = Queue::::shared(); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY - 1); // Two values exceed the one free slot, so the batch panics and the // queue keeps only its earlier values. assert!(catch_unwind(AssertUnwindSafe(|| owner.push_batch(0..2))).is_err()); assert_eq!(drain(&mut owner), (0..CAPACITY - 1).collect::>()); // Popping past a held claim leaves the owner empty with no room. owner.push_batch(0..CAPACITY); let claim = queue.hold_claim(HALF); let mut seen = drain(&mut owner); assert!(owner.is_empty()); // The claim pins the room at zero, so even one value exceeds it. assert!( catch_unwind(AssertUnwindSafe( || owner.push_batch(std::iter::once(CAPACITY)) )) .is_err() ); // Only the claimed and popped values come out, so the batch published nothing. seen.extend(queue.finish_claim(claim)); seen.sort_unstable(); assert_eq!(seen, (0..CAPACITY).collect::>()); } /// Dropping an owner with a queued value panics on the owner's thread. #[test] fn test_nonempty_owner_drop_panics() { let queue = Queue::shared(); let mut owner = queue.owner(); push(&mut owner, 1); // The drop panics while a value is still queued. let result = catch_unwind(AssertUnwindSafe(|| drop(owner))); assert!(result.is_err()); // The drop did not destroy the value, so take it back before the queue drops. assert_eq!(take_unowned(&queue), [1]); } /// Dropping an overflow before it is exhausted panics. #[test] fn test_unexhausted_overflow_drop_panics() { let queue = Queue::shared(); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); // Dropping the overflow without reading its newest half panics. let result = catch_unwind(AssertUnwindSafe(|| match owner.push(CAPACITY) { Push::Overflow(batch, _) => drop(batch), _ => panic!("full queue must overflow"), })); assert!(result.is_err()); // The oldest half stays queued for the owner. assert_eq!(drain(&mut owner), (0..CAPACITY / 2).collect::>()); } /// An unwind through an unexhausted overflow and a nonempty owner does /// not panic again. #[test] fn test_unwinding_skips_drop_assertions() { assert!( catch_unwind(|| { let queue = Queue::shared(); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); // Hold an unexhausted overflow while the owner keeps the oldest half. let _batch = match owner.push(CAPACITY) { Push::Overflow(batch, _) => batch, _ => panic!("full queue must overflow"), }; // Unwinding drops both without a second panic, which would abort. panic!("unwind with an unexhausted overflow"); }) .is_err() ); } /// A queue hands out its owner once. #[test] fn test_second_owner_panics() { let queue = Queue::::shared(); let _owner = queue.owner(); assert!(catch_unwind(AssertUnwindSafe(|| queue.owner())).is_err()); } /// A steal into the source queue panics. #[test] fn test_self_steal_panics() { let queue = Queue::::shared(); let mut owner = queue.owner(); assert!(catch_unwind(AssertUnwindSafe(|| queue.steal_into(&mut owner))).is_err()); } /// A steal from an exclusive queue panics and leaves its values in place. #[test] fn test_steal_from_an_exclusive_queue_panics() { let source = Queue::exclusive(); let destination = Queue::shared(); let mut owner = source.owner(); let mut into = destination.owner(); owner.push_batch(0..4); // The steal panics before claiming, so the owner still pops every value. assert!(catch_unwind(AssertUnwindSafe(|| source.steal_into(&mut into))).is_err()); assert_eq!(drain(&mut owner), [0, 1, 2, 3]); } /// An owner pushing, popping and overflowing against two stealing threads /// hands every value out once and drops it once. #[test] fn test_owner_and_two_thieves_account_for_every_value_once() { let total = if cfg!(miri) { 64 } else { 4096 }; let queue = Arc::new(Queue::::shared()); let mut owner = queue.owner(); let done = Arc::new(AtomicBool::new(false)); let barrier = Arc::new(Barrier::new(3)); let drops = Arc::new(AtomicUsize::new(0)); // Each thief steals into its own queue and collects what it moved. let thieves: Vec<_> = (0..2) .map(|_| { let queue = queue.clone(); let done = done.clone(); let barrier = barrier.clone(); thread::spawn(move || { let destination = Queue::shared(); let mut into = destination.owner(); let mut seen = Vec::new(); barrier.wait(); // Keep stealing until a steal finds nothing after the owner is done. loop { if queue.steal_into(&mut into) > 0 { for value in drain(&mut into) { seen.push(value.0); } } else if done.load(Ordering::Acquire) { break; } else { thread::yield_now(); } } seen }) }) .collect(); // Start the owner together with both thieves. let mut seen = Vec::new(); barrier.wait(); // Push every value, collecting what a claim or an overflow hands back, // and pop after the first push and every third push after it to race // the thieves. for value in 0..total { match owner.push(Counted(value, drops.clone())) { Push::Pushed(_) => {} Push::Claimed(value) => seen.push(value.0), Push::Overflow(batch, value) => { seen.extend(batch.map(|value| value.0)); seen.push(value.0); } } if value % 3 == 0 && let Some(value) = owner.pop() { seen.push(value.0); } } // Tell the thieves to stop once a steal finds nothing. done.store(true, Ordering::Release); // Gather the thieves' values and whatever the owner still holds. for thief in thieves { seen.extend(thief.join().unwrap()); } seen.extend(drain(&mut owner).into_iter().map(|value| value.0)); // Every value is handed out once and dropped once. seen.sort_unstable(); assert_eq!(seen, (0..total).collect::>()); assert_eq!(drops.load(Ordering::Relaxed), total); } /// Queues and owners move between threads, queues are shared, and owners /// are not. #[test] fn test_handles_are_send_and_the_owner_is_not_sync() { fn send() {} fn sync() {} /// Resolving the check is ambiguous exactly when the type is `Sync`, /// since both implementations then match. trait AmbiguousIfSync { fn check() {} } impl AmbiguousIfSync<()> for T {} struct SyncMarker; impl AmbiguousIfSync for T {} // Both handles move between threads and the queue is shared, even for // values that are not `Sync`. send::>>(); sync::>>(); send::>>(); // This resolves only because the owner is not `Sync`. let _ = as AmbiguousIfSync<_>>::check; } } #[cfg(all(test, feature = "loom"))] mod loom_tests { use super::*; use loom::thread; /// Pop every value the owner holds. fn drain(owner: &mut Owner) -> Vec { std::iter::from_fn(|| owner.pop()).collect() } /// Push `value`, collecting whatever the queue hands back into `seen`. fn collect_push(owner: &mut Owner, value: usize, seen: &mut Vec) { match owner.push(value) { Push::Pushed(_) => {} Push::Claimed(value) => seen.push(value), Push::Overflow(batch, value) => { seen.extend(batch); seen.push(value); } } } /// Steal once from `queue` into a fresh queue, returning every value taken. fn steal_once(queue: &Queue) -> Vec { let destination = Queue::shared(); let mut into = destination.owner(); // The thief's queue holds exactly as many values as the steal reports. let moved = queue.steal_into(&mut into); let seen = drain(&mut into); assert_eq!(seen.len(), moved); seen } /// Lone values pushed and popped, then a refill that overflows and a /// drain, racing two steals, hand every value out once. In one run the /// lone values cross the position's wrap, in the other the refill and its /// overflow cross the count's wrap. #[test] fn test_lone_pops_refill_and_overflow_across_the_maximum_against_a_thief() { // Two positions before the position's wrap, the lone values cross it. // Four positions before the count's wrap, the lone values come before // it and the refill and its overflow straddle it. for start in [u64::from(u32::MAX) - 1, REAL_MASK - 3] { let mut builder = loom::model::Builder::new(); builder.preemption_bound = Some(3); builder.check(move || { let queue = Queue::at(start, false); let mut owner = queue.owner(); // The thief steals twice while the owner works. let thief = thread::spawn(move || { let mut seen = steal_once(&queue); seen.extend(steal_once(&queue)); seen }); // Push and pop two lone values, which no steal can take. let mut seen = Vec::new(); for value in 0..2 { collect_push(&mut owner, value, &mut seen); seen.extend(owner.pop()); } // Push one value more than the capacity, overflowing unless a // steal made room, then drain the rest. for value in 2..=CAPACITY + 2 { collect_push(&mut owner, value, &mut seen); } seen.extend(drain(&mut owner)); // The owner and the thief together hand out every value once. seen.extend(thief.join().unwrap()); seen.sort_unstable(); assert_eq!(seen, (0..=CAPACITY + 2).collect::>()); }); } } /// An owner that drains and drops its queue while a thief holds a claim /// across the wrapping position leaves the claimed values to the thief, /// and every value is handed out once. #[test] fn test_owner_drops_while_a_thief_holds_a_claim() { loom::model(|| { // Fill a queue whose positions wrap past the maximum. let queue = Queue::at(u64::from(u32::MAX) - 1, false); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); // A thief holds the oldest half and reads it on its own thread. let claim = queue.hold_claim(HALF); let thief = thread::spawn(move || queue.finish_claim(claim)); // The owner drains its half and drops, possibly while the claim is held. let mut seen = drain(&mut owner); drop(owner); // The thief and the owner together hand out every value once. seen.extend(thief.join().unwrap()); seen.sort_unstable(); assert_eq!(seen, (0..CAPACITY).collect::>()); }); } /// An owner's pop racing two thieves hands every value out once. #[test] fn test_owner_pop_against_two_thieves() { let mut builder = loom::model::Builder::new(); builder.preemption_bound = Some(2); builder.check(|| { let queue = Arc::new(Queue::shared()); let mut owner = queue.owner(); owner.push_batch(0..3); // Two thieves each attempt one steal. let thieves: Vec<_> = (0..2) .map(|_| { let queue = queue.clone(); thread::spawn(move || steal_once(&queue)) }) .collect(); // The owner pops once, racing both steals. let mut seen = Vec::new(); seen.extend(owner.pop()); // Every value comes out once, from a thief or from the owner. for thief in thieves { seen.extend(thief.join().unwrap()); } seen.extend(drain(&mut owner)); seen.sort_unstable(); assert_eq!(seen, vec![0, 1, 2]); }); } /// Pushes and pops across the wrapping position racing a steal hand every /// value out once, with loom tracking each slot's accesses. #[test] fn test_push_against_steal_tracks_cells() { loom::model(|| { // Two values end at the maximum, so the next pushes wrap past it. let queue = Queue::at(u64::from(u32::MAX) - 1, false); let mut owner = queue.owner(); owner.push_batch(0..2); // One steal races the owner's pops and pushes. let thief = thread::spawn(move || steal_once(&queue)); // The owner pops both values, then pushes two more past the wrap. let mut seen = Vec::new(); seen.extend(owner.pop()); seen.extend(owner.pop()); collect_push(&mut owner, 2, &mut seen); collect_push(&mut owner, 3, &mut seen); // Every value comes out once, from the thief or from the owner. seen.extend(thief.join().unwrap()); seen.extend(drain(&mut owner)); seen.sort_unstable(); assert_eq!(seen, vec![0, 1, 2, 3]); }); } /// A thief racing the owner's push and pop of a lone value never sees a /// stealable backlog and takes nothing, while the pop stores the head. #[test] fn test_lone_value_pop_against_a_thief() { let mut builder = loom::model::Builder::new(); builder.preemption_bound = Some(2); builder.check(|| { let queue = Queue::shared(); let mut owner = queue.owner(); // The thief checks and steals twice, and never finds a backlog. let thief = thread::spawn(move || { for _ in 0..2 { assert!(!queue.is_stealable()); assert!(steal_once(&queue).is_empty()); } }); // The owner pushes and pops one value, so the queue never holds two. assert!(matches!(owner.push(7), Push::Pushed(1))); assert_eq!(owner.pop(), Some(7)); // The pop took the value, so nothing remains once the thief finishes. thief.join().unwrap(); assert!(owner.is_empty()); }); } /// A push into a full queue racing a steal overflows, retries, or moves /// only the new value, and every value is handed out once. #[test] fn test_full_overflow_against_a_thief() { loom::model(|| { let queue = Queue::shared(); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); // One steal races a push into the full queue. let thief = thread::spawn(move || steal_once(&queue)); // The push yields after each snapshot, so the steal can complete // before the push decides. let mut seen = Vec::new(); match owner.push_with(CAPACITY, thread::yield_now) { Push::Pushed(_) => {} Push::Claimed(value) => seen.push(value), Push::Overflow(batch, value) => { seen.extend(batch); seen.push(value); } } // Every value comes out once, from the push, the thief, or the owner. seen.extend(thief.join().unwrap()); seen.extend(drain(&mut owner)); seen.sort_unstable(); assert_eq!(seen, (0..=CAPACITY).collect::>()); }); } /// Values a steal publishes in its own queue are visible to a third worker /// that steals from that queue, and the slots they vacate are reused. #[test] fn test_stolen_values_are_published_and_slots_reused() { let mut builder = loom::model::Builder::new(); builder.preemption_bound = Some(2); builder.check(|| { // Fill three source queues with distinct values. let sources: Vec<_> = (0..3).map(|_| Arc::new(Queue::shared())).collect(); let mut owners: Vec<_> = sources.iter().map(|queue| queue.owner()).collect(); for (index, owner) in owners.iter_mut().enumerate() { owner.push_batch(index * CAPACITY..(index + 1) * CAPACITY); } // The destination starts half full, so the first steal into it fits, // and later ones fit only after the third worker steals from it. let destination = Arc::new(Queue::shared()); let mut into = destination.owner(); into.push_batch(100..102); // One worker steals from each source into the destination. let transferring = thread::spawn(move || { for queue in sources { let _ = queue.steal_into(&mut into); } into }); // A third worker steals what the transfers publish in the destination. let stealing = thread::spawn(move || { let last = Queue::shared(); let mut owner = last.owner(); let mut seen = Vec::new(); for _ in 0..3 { let _ = destination.steal_into(&mut owner); seen.extend(drain(&mut owner)); } seen }); // Gather what the third worker, the destination, and each source hold. let mut seen = stealing.join().unwrap(); let mut into = transferring.join().unwrap(); seen.extend(drain(&mut into)); for owner in &mut owners { seen.extend(drain(owner)); } // Every source value and both initial destination values come out once. seen.sort_unstable(); let mut expected: Vec<_> = (0..3 * CAPACITY).collect(); expected.extend(100..102); assert_eq!(seen, expected); }); } /// A batch published into the room a released claim frees, as a take from /// the global queue publishes its share, writes the slots the thief read /// only after those reads, and every value comes out once. #[test] fn test_batch_after_a_release_reuses_the_claimed_slots() { loom::model(|| { let queue = Queue::shared(); let mut owner = queue.owner(); owner.push_batch(0..CAPACITY); // One steal claims the oldest half while the owner drains the rest. let thief = thread::spawn(move || steal_once(&queue)); let mut seen = drain(&mut owner); // Fill the room the owner sees. Once the claim is released, that // room includes the slots the thief read. let room = owner.room(); owner.push_batch(CAPACITY..CAPACITY + room); seen.extend(drain(&mut owner)); // Every value comes out once, from the thief or from the owner. seen.extend(thief.join().unwrap()); seen.sort_unstable(); assert_eq!(seen, (0..CAPACITY + room).collect::>()); }); } }