From ba8e9762e3bc918ec9aa679dd7e1dbb411bd11c4 Mon Sep 17 00:00:00 2001 From: mxsm Date: Fri, 11 Sep 2026 05:46:44 -0700 Subject: [PATCH] feat(spmc): add competing queues --- CHANGELOG.md | 1 + README.md | 1 + asyncband/Cargo.toml | 1 + .../competing_queue}/error.rs | 8 +- .../competing_queue/mod.rs} | 22 +- asyncband/src/internal/mod.rs | 13 +- asyncband/src/internal/semaphore.rs | 4 +- asyncband/src/lib.rs | 3 + asyncband/src/mpmc/bounded.rs | 2 +- asyncband/src/mpmc/mod.rs | 10 +- asyncband/src/mpmc/unbounded.rs | 5 +- asyncband/src/spmc/bounded.rs | 145 +++++++ asyncband/src/spmc/mod.rs | 99 +++++ asyncband/src/spmc/unbounded.rs | 128 +++++++ benchmarks/Cargo.toml | 1 + benchmarks/ecosystem/main.rs | 1 + benchmarks/ecosystem/spmc/README.md | 61 +++ benchmarks/ecosystem/spmc/adapters.rs | 108 ++++++ benchmarks/ecosystem/spmc/bounded.rs | 48 +++ benchmarks/ecosystem/spmc/mod.rs | 21 ++ benchmarks/ecosystem/spmc/support.rs | 87 +++++ benchmarks/ecosystem/spmc/unbounded.rs | 48 +++ tests-integration/Cargo.toml | 1 + tests-integration/tests/spmc_test.rs | 356 ++++++++++++++++++ xtask/src/main.rs | 1 + 25 files changed, 1150 insertions(+), 25 deletions(-) rename asyncband/src/{mpmc => internal/competing_queue}/error.rs (96%) rename asyncband/src/{mpmc/queue.rs => internal/competing_queue/mod.rs} (93%) create mode 100644 asyncband/src/spmc/bounded.rs create mode 100644 asyncband/src/spmc/mod.rs create mode 100644 asyncband/src/spmc/unbounded.rs create mode 100644 benchmarks/ecosystem/spmc/README.md create mode 100644 benchmarks/ecosystem/spmc/adapters.rs create mode 100644 benchmarks/ecosystem/spmc/bounded.rs create mode 100644 benchmarks/ecosystem/spmc/mod.rs create mode 100644 benchmarks/ecosystem/spmc/support.rs create mode 100644 benchmarks/ecosystem/spmc/unbounded.rs create mode 100644 tests-integration/tests/spmc_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 72bf624e..fbe74c99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,7 @@ All notable changes to this project will be documented in this file. ### New features +* Add opt-in bounded and unbounded `asyncband::spmc` queues with one non-cloneable sender requiring exclusive access, cloneable competing receivers, and cancellation-safe receive notification handoff. * Add opt-in bounded and unbounded `asyncband::mpmc` queues with cloneable producers and competing consumers, delivering each accepted value to exactly one receiver while a receiver remains. * Add an opt-in runtime-agnostic `Phaser` with shared observer handles, dynamic RAII participants registered individually or in batches through an owning iterator, `u64` phase numbers, split arrival/wait with cancellation-resilient retries, and a `close` operation that releases unfinished waits with `Closed`. * Add bounded MPSC `reserve` and `try_reserve` methods returning a `Permit`, allowing callers to wait for capacity before constructing a message; pending sends and reservations receive capacity in wait-queue order, and unused permits release capacity without claiming message order. diff --git a/README.md b/README.md index 287ac380..5e56b824 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,7 @@ Runnable examples live in the [`examples`](examples) workspace crate. They demon | | [`oneshot`](https://docs.rs/asyncband/*/asyncband/oneshot/) | `oneshot` | Send one value from one sender to one receiver. | | | [`mpmc`](https://docs.rs/asyncband/*/asyncband/mpmc/) | `mpmc` | Distribute each value to exactly one of multiple competing receivers. | | | [`mpsc`](https://docs.rs/asyncband/*/asyncband/mpsc/) | `mpsc` | Send each value from multiple producers to one receiver with bounded backpressure or an unbounded queue. | +| | [`spmc`](https://docs.rs/asyncband/*/asyncband/spmc/) | `spmc` | Distribute work from one exclusive sender to multiple competing receivers, with bounded or unbounded storage. | | | [`broadcast`](https://docs.rs/asyncband/*/asyncband/broadcast/) | `broadcast` | Deliver every value to receivers active at send time; retain an unbounded backlog until each consumes or drops. | | | [`watch`](https://docs.rs/asyncband/*/asyncband/watch/) | `watch` | Publish cloneable latest state from one or more senders; receivers independently coalesce intermediate updates. | | Object reuse | [`pool`](https://docs.rs/asyncband/*/asyncband/pool/) | `pool` | Reuse objects through bounded or unbounded pool variants. | diff --git a/asyncband/Cargo.toml b/asyncband/Cargo.toml index d34a926b..819ef36b 100644 --- a/asyncband/Cargo.toml +++ b/asyncband/Cargo.toml @@ -65,6 +65,7 @@ rwlock = [] semaphore = [] shutdown = ["latch", "waitgroup"] singleflight = ["dep:hashbrown", "once-cell"] +spmc = [] waitgroup = [] watch = [] diff --git a/asyncband/src/mpmc/error.rs b/asyncband/src/internal/competing_queue/error.rs similarity index 96% rename from asyncband/src/mpmc/error.rs rename to asyncband/src/internal/competing_queue/error.rs index 7bb7525e..1ad71e45 100644 --- a/asyncband/src/mpmc/error.rs +++ b/asyncband/src/internal/competing_queue/error.rs @@ -34,10 +34,10 @@ impl SendError { pub fn into_inner(self) -> T { self.0 } +} - pub(super) fn new(value: T) -> Self { - Self(value) - } +pub fn send_error(value: T) -> SendError { + SendError(value) } impl fmt::Display for SendError { @@ -54,7 +54,7 @@ impl fmt::Debug for SendError { impl std::error::Error for SendError {} -/// Error returned by [`BoundedSender::try_send`](crate::mpmc::BoundedSender::try_send). +/// Error returned when attempting to send without waiting for capacity. #[derive(Clone, PartialEq, Eq)] pub enum TrySendError { /// The queue is full, so the value cannot be sent without waiting for capacity. diff --git a/asyncband/src/mpmc/queue.rs b/asyncband/src/internal/competing_queue/mod.rs similarity index 93% rename from asyncband/src/mpmc/queue.rs rename to asyncband/src/internal/competing_queue/mod.rs index fe51bce6..537b4a50 100644 --- a/asyncband/src/mpmc/queue.rs +++ b/asyncband/src/internal/competing_queue/mod.rs @@ -22,15 +22,20 @@ use std::pin::Pin; use std::task::Context; use std::task::Poll; -use super::RecvError; -use super::SendError; -use super::TryRecvError; -use super::TrySendError; use crate::internal::mutex::Mutex; use crate::internal::semaphore::Acquire; use crate::internal::semaphore::Semaphore; -pub(super) struct Shared { +mod error; + +pub use self::error::RecvError; +pub use self::error::SendError; +pub use self::error::TryRecvError; +pub use self::error::TrySendError; +pub use self::error::send_error; + +// Shared by MPMC and SPMC; endpoint wrappers decide which producer capabilities are exposed. +pub struct Shared { state: Mutex>, recv_waiters: Semaphore, send_waiters: Semaphore, @@ -65,6 +70,7 @@ impl Shared { } } + #[cfg(feature = "mpmc")] pub fn clone_sender(&self) { let mut state = self.state.lock(); state.senders = state @@ -89,7 +95,7 @@ impl Shared { state.receivers = state .receivers .checked_add(1) - .expect("mpmc receiver count overflow"); + .expect("competing queue receiver count overflow"); } pub fn drop_receiver(&self) { @@ -125,7 +131,7 @@ impl Shared { pub async fn send(&self, value: T) -> Result<(), SendError> { let value = match self.try_send(value) { Ok(()) => return Ok(()), - Err(TrySendError::Disconnected(value)) => return Err(SendError::new(value)), + Err(TrySendError::Disconnected(value)) => return Err(send_error(value)), Err(TrySendError::Full(value)) => value, }; let mut send = Send { @@ -181,7 +187,7 @@ impl Send<'_, T> { value = match self.shared.try_send(value) { Ok(()) => return Poll::Ready(Ok(())), Err(TrySendError::Disconnected(value)) => { - return Poll::Ready(Err(SendError::new(value))); + return Poll::Ready(Err(send_error(value))); } Err(TrySendError::Full(value)) => value, }; diff --git a/asyncband/src/internal/mod.rs b/asyncband/src/internal/mod.rs index 7b8cb352..8e6daf97 100644 --- a/asyncband/src/internal/mod.rs +++ b/asyncband/src/internal/mod.rs @@ -56,6 +56,7 @@ pub(crate) fn wake_all(mut wakers: impl Iterator) { feature = "completion", feature = "latch", feature = "mpmc", + feature = "spmc", feature = "mpsc", feature = "mutex", feature = "phaser", @@ -69,6 +70,9 @@ pub(crate) fn wake_all(mut wakers: impl Iterator) { #[allow(dead_code)] pub(crate) mod arena; +#[cfg(any(feature = "mpmc", feature = "spmc"))] +pub(crate) mod competing_queue; + #[cfg(any(feature = "latch", feature = "once"))] pub(crate) mod countdown; @@ -85,6 +89,7 @@ pub(crate) mod value_cell; feature = "completion", feature = "latch", feature = "mpmc", + feature = "spmc", feature = "mpsc", feature = "mutex", feature = "phaser", @@ -99,18 +104,21 @@ pub(crate) mod mutex; #[cfg(any( feature = "mpmc", + feature = "spmc", feature = "mutex", feature = "rwlock", feature = "semaphore", ))] -// MPMC uses waiter notifications; mutexes and rwlocks use acquire/release operations; the public -// semaphore also exposes permit accounting. Single-primitive builds leave part of this API unused. +// Competing queues use waiter notifications; mutexes and rwlocks use acquire/release operations; +// the public semaphore also exposes permit accounting. Single-primitive builds leave part of this +// API unused. #[allow(dead_code)] pub(crate) mod semaphore; #[cfg(any( feature = "event", feature = "mpmc", + feature = "spmc", feature = "mpsc", feature = "mutex", feature = "rwlock", @@ -128,6 +136,7 @@ pub(crate) mod waitlist; feature = "completion", feature = "latch", feature = "mpmc", + feature = "spmc", feature = "mpsc", feature = "mutex", feature = "once", diff --git a/asyncband/src/internal/semaphore.rs b/asyncband/src/internal/semaphore.rs index 8c2de0c1..7f8cd287 100644 --- a/asyncband/src/internal/semaphore.rs +++ b/asyncband/src/internal/semaphore.rs @@ -206,7 +206,7 @@ impl Semaphore { } /// Adds `n` permits to the semaphore if there is any waiter. - #[cfg(feature = "mpmc")] + #[cfg(any(feature = "mpmc", feature = "spmc"))] pub fn release_if_nonempty(&self, n: usize) { let waiters = self.waiters.lock(); if !waiters.is_empty() { @@ -215,7 +215,7 @@ impl Semaphore { } /// Adds as many permits until there is no waiter. - #[cfg(feature = "mpmc")] + #[cfg(any(feature = "mpmc", feature = "spmc"))] pub fn notify_all(&self) { let mut waiters = self.waiters.lock(); let mut wakers = vec![]; diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index ae99158c..773b89fb 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -75,6 +75,7 @@ //! | | [`oneshot`] | `oneshot` | Send one value from one sender to one receiver. | //! | | [`mpmc`] | `mpmc` | Distribute each value to exactly one of multiple competing receivers. | //! | | [`mpsc`] | `mpsc` | Send each value from multiple producers to one receiver with bounded backpressure or an unbounded queue. | +//! | | [`spmc`] | `spmc` | Distribute work from one exclusive sender to multiple competing receivers, with bounded or unbounded storage. | //! | | [`broadcast`] | `broadcast` | Deliver every value to receivers active at send time; retain an unbounded backlog until each consumes or drops. | //! | | [`watch`] | `watch` | Publish cloneable latest state from one or more senders; receivers independently coalesce intermediate updates. | //! | Object reuse | [`pool`] | `pool` | Reuse objects through bounded or unbounded pool variants. | @@ -161,6 +162,8 @@ pub mod semaphore; pub mod shutdown; #[cfg(feature = "singleflight")] pub mod singleflight; +#[cfg(feature = "spmc")] +pub mod spmc; #[cfg(feature = "waitgroup")] pub mod waitgroup; #[cfg(feature = "watch")] diff --git a/asyncband/src/mpmc/bounded.rs b/asyncband/src/mpmc/bounded.rs index 3bcd73f9..1a08117b 100644 --- a/asyncband/src/mpmc/bounded.rs +++ b/asyncband/src/mpmc/bounded.rs @@ -22,7 +22,7 @@ use super::RecvError; use super::SendError; use super::TryRecvError; use super::TrySendError; -use super::queue::Shared; +use crate::internal::competing_queue::Shared; /// Creates a bounded multi-producer, multi-consumer queue. /// diff --git a/asyncband/src/mpmc/mod.rs b/asyncband/src/mpmc/mod.rs index 8a3f0392..204d6eed 100644 --- a/asyncband/src/mpmc/mod.rs +++ b/asyncband/src/mpmc/mod.rs @@ -23,17 +23,15 @@ //! sends return their value in an error. mod bounded; -mod error; -mod queue; mod unbounded; pub use self::bounded::BoundedReceiver; pub use self::bounded::BoundedSender; pub use self::bounded::bounded; -pub use self::error::RecvError; -pub use self::error::SendError; -pub use self::error::TryRecvError; -pub use self::error::TrySendError; pub use self::unbounded::UnboundedReceiver; pub use self::unbounded::UnboundedSender; pub use self::unbounded::unbounded; +pub use crate::internal::competing_queue::RecvError; +pub use crate::internal::competing_queue::SendError; +pub use crate::internal::competing_queue::TryRecvError; +pub use crate::internal::competing_queue::TrySendError; diff --git a/asyncband/src/mpmc/unbounded.rs b/asyncband/src/mpmc/unbounded.rs index 7e1db12a..5c7654d1 100644 --- a/asyncband/src/mpmc/unbounded.rs +++ b/asyncband/src/mpmc/unbounded.rs @@ -22,7 +22,8 @@ use super::RecvError; use super::SendError; use super::TryRecvError; use super::TrySendError; -use super::queue::Shared; +use crate::internal::competing_queue::Shared; +use crate::internal::competing_queue::send_error; /// Creates an unbounded multi-producer, multi-consumer queue. /// @@ -76,7 +77,7 @@ impl UnboundedSender { pub fn send(&self, value: T) -> Result<(), SendError> { match self.shared.try_send(value) { Ok(()) => Ok(()), - Err(TrySendError::Disconnected(value)) => Err(SendError::new(value)), + Err(TrySendError::Disconnected(value)) => Err(send_error(value)), Err(TrySendError::Full(_)) => unreachable!("unbounded queue cannot be full"), } } diff --git a/asyncband/src/spmc/bounded.rs b/asyncband/src/spmc/bounded.rs new file mode 100644 index 00000000..7524bddc --- /dev/null +++ b/asyncband/src/spmc/bounded.rs @@ -0,0 +1,145 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::fmt; +use std::sync::Arc; + +use super::RecvError; +use super::SendError; +use super::TryRecvError; +use super::TrySendError; +use crate::internal::competing_queue::Shared; + +/// Creates a bounded single-producer, multi-consumer queue. +/// +/// The queue stores at most `capacity` values. Sending waits for a receiver to free capacity when +/// the queue is full. +/// +/// Operations briefly acquire internal mutexes. No lock is held across an await point, while +/// waking tasks, or while dropping messages. The `try_*` methods do not wait for capacity or +/// messages, but may wait to acquire a mutex. +/// +/// # Panics +/// +/// Panics if `capacity` is zero. +#[track_caller] +pub fn bounded(capacity: usize) -> (BoundedSender, BoundedReceiver) { + assert!(capacity > 0, "spmc bounded queue requires capacity > 0"); + let shared = Arc::new(Shared::bounded(capacity)); + ( + BoundedSender { + shared: shared.clone(), + }, + BoundedReceiver { shared }, + ) +} + +/// Sends values to the associated [`BoundedReceiver`] handles. +/// +/// Instances are created by [`bounded`] and cannot be cloned. Sending requires exclusive access to +/// this endpoint. +pub struct BoundedSender { + shared: Arc>, +} + +impl fmt::Debug for BoundedSender { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("BoundedSender").finish_non_exhaustive() + } +} + +impl Drop for BoundedSender { + fn drop(&mut self) { + self.shared.drop_sender(); + } +} + +impl BoundedSender { + /// Sends a value, waiting until capacity is available if the queue is full. + /// + /// If all receivers have been dropped, the value is returned in [`SendError`]. + /// + /// # Cancel safety + /// + /// Dropping a pending `send` removes it from the wait queue and drops `value`; a call that has + /// returned `Pending` has not sent the value. Cancelling releases the exclusive sender borrow + /// and leaves available capacity usable by the next send. Use [`try_send`](Self::try_send) when + /// the caller must retain ownership if capacity is unavailable. + pub async fn send(&mut self, value: T) -> Result<(), SendError> { + self.shared.send(value).await + } + + /// Attempts to send a value without waiting for capacity. + /// + /// Returns [`TrySendError::Full`] when the queue has reached its exact capacity and + /// [`TrySendError::Disconnected`] when all receivers have been dropped. + pub fn try_send(&mut self, value: T) -> Result<(), TrySendError> { + self.shared.try_send(value) + } +} + +/// Receives values from the associated [`BoundedSender`] handles. +/// +/// Cloned receivers compete for values, and every accepted value is returned by exactly one +/// receiver while a receiver remains. Dropping the final receiver releases buffered values. +pub struct BoundedReceiver { + shared: Arc>, +} + +impl Clone for BoundedReceiver { + fn clone(&self) -> Self { + self.shared.clone_receiver(); + Self { + shared: self.shared.clone(), + } + } +} + +impl fmt::Debug for BoundedReceiver { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("BoundedReceiver").finish_non_exhaustive() + } +} + +impl Drop for BoundedReceiver { + fn drop(&mut self) { + self.shared.drop_receiver(); + } +} + +impl BoundedReceiver { + /// Receives the next available value. + /// + /// Buffered values remain available after the sender is dropped. Once they are drained, + /// this method returns [`RecvError::Disconnected`]. + /// + /// # Cancel safety + /// + /// Dropping a pending `recv` does not consume a value. Any selected value notification is + /// passed to another waiting receiver, so cancellation does not prevent it from receiving. + pub async fn recv(&self) -> Result { + self.shared.recv().await + } + + /// Attempts to receive the next available value without waiting for a message. + /// + /// Returns [`TryRecvError::Empty`] while the queue is empty and a sender remains, or + /// [`TryRecvError::Disconnected`] once the queue is empty and the sender has been dropped. + pub fn try_recv(&self) -> Result { + self.shared.try_recv() + } +} diff --git a/asyncband/src/spmc/mod.rs b/asyncband/src/spmc/mod.rs new file mode 100644 index 00000000..d04ce170 --- /dev/null +++ b/asyncband/src/spmc/mod.rs @@ -0,0 +1,99 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Single-producer, multi-consumer queues for distributing work between asynchronous tasks. +//! +//! Enable the `spmc` Cargo feature to use this module. [`bounded`] applies backpressure at its +//! exact capacity; [`unbounded`] sends synchronously and can grow until memory is exhausted. +//! Receivers are cloneable and compete for messages: each accepted message is delivered to one +//! receiver while receivers remain. Values leave the queue in FIFO order, but consumer completion +//! order and an equal distribution of work are not guaranteed. +//! +//! The sender cannot be cloned, and every send operation requires `&mut self`, including for the +//! lifetime of a bounded send future. It can move between tasks, but shared references cannot send. +//! Dropping the sender lets receivers drain buffered messages before observing disconnection. +//! Dropping the last receiver releases buffered messages and makes sending return the unsent value. +//! +//! # Example +//! +//! ``` +//! # #[tokio::main] +//! # async fn main() { +//! use asyncband::spmc; +//! +//! let (mut sender, receiver) = spmc::bounded(2); +//! let competing = receiver.clone(); +//! sender.send("first").await.unwrap(); +//! sender.send("second").await.unwrap(); +//! drop(sender); +//! +//! assert_eq!(receiver.recv().await, Ok("first")); +//! assert_eq!(competing.recv().await, Ok("second")); +//! assert_eq!(receiver.recv().await, Err(spmc::RecvError::Disconnected)); +//! # } +//! ``` +//! +//! # Single-producer capability +//! +//! Neither sender supports cloning: +//! +//! ```compile_fail,E0599 +//! let (sender, _receiver) = asyncband::spmc::bounded::(1); +//! let second_producer = sender.clone(); +//! ``` +//! +//! ```compile_fail,E0599 +//! let (sender, _receiver) = asyncband::spmc::unbounded::(); +//! let second_producer = sender.clone(); +//! ``` +//! +//! Sending through a shared reference is rejected: +//! +//! ```compile_fail,E0596 +//! fn send(sender: &asyncband::spmc::BoundedSender) { +//! let _ = sender.try_send(1); +//! } +//! ``` +//! +//! ```compile_fail,E0596 +//! fn send(sender: &asyncband::spmc::UnboundedSender) { +//! let _ = sender.send(1); +//! } +//! ``` +//! +//! A bounded send future retains the exclusive borrow until completion or cancellation: +//! +//! ```compile_fail,E0499 +//! let (mut sender, _receiver) = asyncband::spmc::bounded(1); +//! let pending = sender.send(1); +//! let _ = sender.try_send(2); +//! drop(pending); +//! ``` + +mod bounded; +mod unbounded; + +pub use self::bounded::BoundedReceiver; +pub use self::bounded::BoundedSender; +pub use self::bounded::bounded; +pub use self::unbounded::UnboundedReceiver; +pub use self::unbounded::UnboundedSender; +pub use self::unbounded::unbounded; +pub use crate::internal::competing_queue::RecvError; +pub use crate::internal::competing_queue::SendError; +pub use crate::internal::competing_queue::TryRecvError; +pub use crate::internal::competing_queue::TrySendError; diff --git a/asyncband/src/spmc/unbounded.rs b/asyncband/src/spmc/unbounded.rs new file mode 100644 index 00000000..c8cd00df --- /dev/null +++ b/asyncband/src/spmc/unbounded.rs @@ -0,0 +1,128 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::fmt; +use std::sync::Arc; + +use super::RecvError; +use super::SendError; +use super::TryRecvError; +use super::TrySendError; +use crate::internal::competing_queue::Shared; +use crate::internal::competing_queue::send_error; + +/// Creates an unbounded single-producer, multi-consumer queue. +/// +/// Sends are synchronous and values may be buffered until available memory is exhausted. +/// +/// Operations briefly acquire internal mutexes. No lock is held across an await point, while +/// waking tasks, or while dropping messages. Sending and trying to receive may wait to acquire +/// a mutex, but never wait for capacity or new messages. +pub fn unbounded() -> (UnboundedSender, UnboundedReceiver) { + let shared = Arc::new(Shared::unbounded()); + ( + UnboundedSender { + shared: shared.clone(), + }, + UnboundedReceiver { shared }, + ) +} + +/// Sends values to the associated [`UnboundedReceiver`] handles. +/// +/// Instances are created by [`unbounded`] and cannot be cloned. Sending requires exclusive access +/// to this endpoint. +pub struct UnboundedSender { + shared: Arc>, +} + +impl fmt::Debug for UnboundedSender { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("UnboundedSender").finish_non_exhaustive() + } +} + +impl Drop for UnboundedSender { + fn drop(&mut self) { + self.shared.drop_sender(); + } +} + +impl UnboundedSender { + /// Sends a value without waiting for capacity. + /// + /// If all receivers have been dropped, the value is returned in [`SendError`]. + pub fn send(&mut self, value: T) -> Result<(), SendError> { + match self.shared.try_send(value) { + Ok(()) => Ok(()), + Err(TrySendError::Disconnected(value)) => Err(send_error(value)), + Err(TrySendError::Full(_)) => unreachable!("unbounded queue cannot be full"), + } + } +} + +/// Receives values from the associated [`UnboundedSender`] handles. +/// +/// Cloned receivers compete for values, and every accepted value is returned by exactly one +/// receiver while a receiver remains. Dropping the final receiver releases buffered values. +pub struct UnboundedReceiver { + shared: Arc>, +} + +impl Clone for UnboundedReceiver { + fn clone(&self) -> Self { + self.shared.clone_receiver(); + Self { + shared: self.shared.clone(), + } + } +} + +impl fmt::Debug for UnboundedReceiver { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("UnboundedReceiver").finish_non_exhaustive() + } +} + +impl Drop for UnboundedReceiver { + fn drop(&mut self) { + self.shared.drop_receiver(); + } +} + +impl UnboundedReceiver { + /// Receives the next available value. + /// + /// Buffered values remain available after the sender is dropped. Once they are drained, + /// this method returns [`RecvError::Disconnected`]. + /// + /// # Cancel safety + /// + /// Dropping a pending `recv` does not consume a value. Any selected value notification is + /// passed to another waiting receiver, so cancellation does not prevent it from receiving. + pub async fn recv(&self) -> Result { + self.shared.recv().await + } + + /// Attempts to receive the next available value without waiting for a message. + /// + /// Returns [`TryRecvError::Empty`] while the queue is empty and a sender remains, or + /// [`TryRecvError::Disconnected`] once the queue is empty and the sender has been dropped. + pub fn try_recv(&self) -> Result { + self.shared.try_recv() + } +} diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml index bcab8f24..b49b8eb1 100644 --- a/benchmarks/Cargo.toml +++ b/benchmarks/Cargo.toml @@ -46,6 +46,7 @@ asyncband = { workspace = true, features = [ "semaphore", "shutdown", "singleflight", + "spmc", "waitgroup", "watch", ] } diff --git a/benchmarks/ecosystem/main.rs b/benchmarks/ecosystem/main.rs index 6cc187c2..fa687e45 100644 --- a/benchmarks/ecosystem/main.rs +++ b/benchmarks/ecosystem/main.rs @@ -22,6 +22,7 @@ mod mpmc; #[path = "../mpmc/mod.rs"] mod mpmc_support; mod mpsc; +mod spmc; mod waitgroup; mod watch; diff --git a/benchmarks/ecosystem/spmc/README.md b/benchmarks/ecosystem/spmc/README.md new file mode 100644 index 00000000..09ec699d --- /dev/null +++ b/benchmarks/ecosystem/spmc/README.md @@ -0,0 +1,61 @@ + + +# SPMC competing queue benchmarks + +These benchmarks exercise Issue [#212](https://github.com/apache/asyncband/issues/212): one producer owns a non-cloneable sender, and 1, 2, 4, or 8 receivers compete for values. The same harness compares Asyncband SPMC, Asyncband MPMC used with one producer, async-channel, and flume. Peer dependencies stay in the benchmark crate; their exact versions are recorded in `Cargo.lock`. + +Every sample transfers 16,384 `usize` values; bounded queues have capacity 64. One sender moves into a producer task without cloning or a synchronization wrapper. Consumers drain freely until disconnection, with no equal-work quotas. All data operations run in spawned Tokio tasks, including on the current-thread runtime. The coordinator only releases the start barrier and joins the tasks. Each sample checks the total count and checksum. + +The two runtime configurations are `WORKERS = 0` for Tokio current-thread and `WORKERS = 4` for four worker threads. Queue and runtime construction are outside the measured region; sending, receiving, terminal disconnection, and task completion are inside. Unbounded sending is synchronous for every implementation. Its producer can fill the queue before consumers run on a current-thread executor, so that configuration measures draining rather than parallel consumer contention. Use the four-worker results and the targeted-wakeup tests to evaluate the 1P/8C case. + +Run the repository benchmark workflow: + +```powershell +cargo x --help +cargo x bench --help +cargo x bench +``` + +For repeated focused measurements, use `cargo x bench --no-run`, then run the ecosystem executable printed by Cargo with `--bench --color never --sample-count 100 'spmc::'`. The default is 20 samples with one batch per sample. Run repeated comparisons serially, without concurrent builds or tests, and record the machine, OS, Rust version, commit, sample settings, and results for all consumer counts. Lower elapsed time is better. Investigate sustained gaps above 3x against a comparable peer; sustained order-of-magnitude gaps block acceptance under [#208](https://github.com/apache/asyncband/issues/208). + +## Development measurements (2026-09-11) + +Measured on an Intel Core i7-11700K (8 cores / 16 logical processors), 64-bit Windows 11 Pro 10.0.26200, Rust 1.96.1 (`31fca3adb`), and the default optimized bench profile, with this implementation based on `main` at `8204e14`. Peers: async-channel 2.5.0, flume 0.12.0; runtime: Tokio 1.53.1; harness: Divan 0.1.21. The full `cargo x bench` suite passed before three serial focused runs of 100 samples per case. No builds or tests ran alongside the measurements. Each value below is the median of three run medians in milliseconds per batch; `current` denotes Tokio current-thread. The ratio compares SPMC with the faster of async-channel and flume in that row. + +| Queue | Runtime | Consumers | SPMC | MPMC | async-channel | flume | Peer ratio | +| --------- | ------- | --------: | ------: | ------: | ------------: | ------: | ---------: | +| bounded | current | 1 | 0.819 | 0.875 | 1.207 | 0.652 | 1.25x | +| bounded | current | 2 | 0.811 | 0.866 | 1.200 | 0.645 | 1.26x | +| bounded | current | 4 | 0.873 | 0.931 | 1.469 | 0.689 | 1.27x | +| bounded | current | 8 | 0.999 | 1.041 | 1.492 | 0.770 | 1.30x | +| bounded | 4 | 1 | 1.054 | 1.123 | 1.480 | 0.898 | 1.17x | +| bounded | 4 | 2 | 3.830 | 4.002 | 2.021 | 3.193 | 1.90x | +| bounded | 4 | 4 | 8.008 | 7.979 | 6.074 | 4.889 | 1.64x | +| bounded | 4 | 8 | 11.570 | 11.400 | 10.610 | 6.926 | 1.67x | +| unbounded | current | 1 | 0.565 | 0.613 | 1.314 | 0.521 | 1.08x | +| unbounded | current | 2 | 0.543 | 0.588 | 1.262 | 0.495 | 1.10x | +| unbounded | current | 4 | 0.542 | 0.588 | 1.254 | 0.498 | 1.09x | +| unbounded | current | 8 | 0.546 | 0.596 | 1.263 | 0.506 | 1.08x | +| unbounded | 4 | 1 | 0.567 | 0.597 | 1.298 | 0.524 | 1.08x | +| unbounded | 4 | 2 | 3.003 | 3.250 | 1.529 | 2.542 | 1.96x | +| unbounded | 4 | 4 | 6.751 | 6.537 | 2.543 | 3.418 | 2.65x | +| unbounded | 4 | 8 | 10.360 | 9.824 | 2.667 | 3.816 | 3.88x | + +The largest sustained gap is unbounded 1P/8C with four workers: 10.360 ms versus async-channel at 2.667 ms (3.88x). The same harness measures the shared-core MPMC variant at 9.824 ms, only about 5% below SPMC. Both variants serialize storage and endpoint state under a mutex and use semaphore-backed receiver notifications; SPMC adds no extra queue lock or per-message allocation over MPMC. Together with the deterministic eight-waiter test proving one ordinary notification and cancellation handoff, this points to shared storage/waiter contention rather than a wake-all implementation. This is an inference from the implementation and comparison, not a lock profile. No topology reaches the 10x rejection threshold on this host, but the 3.88x gap remains a performance limitation. Reusing the core avoids a second backend while the exclusive, non-cloneable sender supplies the required static capability; these results are not a claim of general superiority over MPMC or peers. diff --git a/benchmarks/ecosystem/spmc/adapters.rs b/benchmarks/ecosystem/spmc/adapters.rs new file mode 100644 index 00000000..6930ad86 --- /dev/null +++ b/benchmarks/ecosystem/spmc/adapters.rs @@ -0,0 +1,108 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::future::Future; +use std::marker::PhantomData; + +pub use crate::mpmc_support::adapters::AsyncChannel; +pub use crate::mpmc_support::adapters::Asyncband as Mpmc; +use crate::mpmc_support::adapters::BoundedMpmc; +pub use crate::mpmc_support::adapters::Flume; +use crate::mpmc_support::adapters::UnboundedMpmc; +use crate::mpmc_support::support::BOUNDED_CAPACITY; + +pub struct Spmc; +pub struct Bounded(PhantomData); +pub struct Unbounded(PhantomData); + +// The harness moves the only sender into one task, without adding Clone or Sync requirements. +pub trait Channel: Send + Sync + 'static { + type Sender: Send + 'static; + type Receiver: Clone + Send + Sync + 'static; + + fn channel() -> (Self::Sender, Self::Receiver); + fn send(sender: &mut Self::Sender, value: usize) -> impl Future + Send; + fn recv(receiver: &Self::Receiver) -> impl Future> + Send; +} + +impl Channel for Bounded { + type Sender = asyncband::spmc::BoundedSender; + type Receiver = asyncband::spmc::BoundedReceiver; + + fn channel() -> (Self::Sender, Self::Receiver) { + asyncband::spmc::bounded(BOUNDED_CAPACITY) + } + + async fn send(sender: &mut Self::Sender, value: usize) { + sender.send(value).await.unwrap(); + } + + async fn recv(receiver: &Self::Receiver) -> Option { + receiver.recv().await.ok() + } +} + +impl Channel for Unbounded { + type Sender = asyncband::spmc::UnboundedSender; + type Receiver = asyncband::spmc::UnboundedReceiver; + + fn channel() -> (Self::Sender, Self::Receiver) { + asyncband::spmc::unbounded() + } + + async fn send(sender: &mut Self::Sender, value: usize) { + sender.send(value).unwrap(); + } + + async fn recv(receiver: &Self::Receiver) -> Option { + receiver.recv().await.ok() + } +} + +impl Channel for Bounded { + type Sender = C::Sender; + type Receiver = C::Receiver; + + fn channel() -> (Self::Sender, Self::Receiver) { + C::channel(BOUNDED_CAPACITY) + } + + async fn send(sender: &mut Self::Sender, value: usize) { + C::send_async(sender, value).await; + } + + async fn recv(receiver: &Self::Receiver) -> Option { + C::recv_async(receiver).await + } +} + +impl Channel for Unbounded { + type Sender = C::Sender; + type Receiver = C::Receiver; + + fn channel() -> (Self::Sender, Self::Receiver) { + C::channel() + } + + async fn send(sender: &mut Self::Sender, value: usize) { + C::send(sender, value); + } + + async fn recv(receiver: &Self::Receiver) -> Option { + C::recv_async(receiver).await + } +} diff --git a/benchmarks/ecosystem/spmc/bounded.rs b/benchmarks/ecosystem/spmc/bounded.rs new file mode 100644 index 00000000..8ec46a0e --- /dev/null +++ b/benchmarks/ecosystem/spmc/bounded.rs @@ -0,0 +1,48 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use divan::Bencher; +use divan::counter::ItemsCount; + +use super::adapters::AsyncChannel; +use super::adapters::Bounded; +use super::adapters::Channel; +use super::adapters::Flume; +use super::adapters::Mpmc; +use super::adapters::Spmc; +use super::support::BATCH_MESSAGES; +use super::support::CONSUMERS; +use super::support::TaskBatch; +use super::support::runtime; + +#[divan::bench( + types = [Spmc, Mpmc, AsyncChannel, Flume], + consts = [0, 4], + args = CONSUMERS, + sample_count = 20, + sample_size = 1, + counter = ItemsCount::new(BATCH_MESSAGES), +)] +fn tokio_tasks(bencher: Bencher, consumers: usize) +where + Bounded: Channel, +{ + let runtime = runtime(WORKERS); + bencher + .with_inputs(|| TaskBatch::new::>(&runtime, consumers)) + .bench_local_refs(|batch| runtime.block_on(batch.run())); +} diff --git a/benchmarks/ecosystem/spmc/mod.rs b/benchmarks/ecosystem/spmc/mod.rs new file mode 100644 index 00000000..dd09282b --- /dev/null +++ b/benchmarks/ecosystem/spmc/mod.rs @@ -0,0 +1,21 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +mod adapters; +mod bounded; +mod support; +mod unbounded; diff --git a/benchmarks/ecosystem/spmc/support.rs b/benchmarks/ecosystem/spmc/support.rs new file mode 100644 index 00000000..f88258a6 --- /dev/null +++ b/benchmarks/ecosystem/spmc/support.rs @@ -0,0 +1,87 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::sync::Arc; + +use divan::black_box; +use tokio::runtime::Runtime; +use tokio::task::JoinSet; + +use super::adapters::Channel; +pub use crate::mpmc_support::support::BATCH_MESSAGES; +pub use crate::mpmc_support::support::runtime; + +pub const CONSUMERS: &[usize] = &[1, 2, 4, 8]; + +pub struct TaskBatch { + start: Arc, + tasks: JoinSet<(usize, usize)>, +} + +impl TaskBatch { + pub fn new(runtime: &Runtime, consumers: usize) -> Self { + let (mut sender, receiver) = C::channel(); + let start = Arc::new(tokio::sync::Barrier::new(consumers + 2)); + let mut tasks = JoinSet::new(); + let producer_start = start.clone(); + tasks.spawn_on( + async move { + producer_start.wait().await; + for value in 0..BATCH_MESSAGES { + C::send(&mut sender, black_box(value)).await; + } + // The sender is moved once and dropped on completion so consumers can drain. + (0, 0) + }, + runtime.handle(), + ); + for _ in 0..consumers { + let receiver = receiver.clone(); + let start = start.clone(); + tasks.spawn_on( + async move { + start.wait().await; + let mut count = 0; + let mut checksum = 0usize; + // Consumers compete freely, with no fixed per-consumer quota. + while let Some(value) = C::recv(&receiver).await { + count += 1; + checksum = checksum.wrapping_add(value); + } + (count, checksum) + }, + runtime.handle(), + ); + } + drop(receiver); + Self { start, tasks } + } + + pub async fn run(&mut self) -> (usize, usize) { + self.start.wait().await; + let mut count = 0; + let mut checksum = 0usize; + while let Some(result) = self.tasks.join_next().await { + let (received, sum) = result.expect("benchmark task panicked"); + count += received; + checksum = checksum.wrapping_add(sum); + } + assert_eq!(count, BATCH_MESSAGES); + assert_eq!(checksum, BATCH_MESSAGES * (BATCH_MESSAGES - 1) / 2); + black_box((count, checksum)) + } +} diff --git a/benchmarks/ecosystem/spmc/unbounded.rs b/benchmarks/ecosystem/spmc/unbounded.rs new file mode 100644 index 00000000..4c5d446c --- /dev/null +++ b/benchmarks/ecosystem/spmc/unbounded.rs @@ -0,0 +1,48 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use divan::Bencher; +use divan::counter::ItemsCount; + +use super::adapters::AsyncChannel; +use super::adapters::Channel; +use super::adapters::Flume; +use super::adapters::Mpmc; +use super::adapters::Spmc; +use super::adapters::Unbounded; +use super::support::BATCH_MESSAGES; +use super::support::CONSUMERS; +use super::support::TaskBatch; +use super::support::runtime; + +#[divan::bench( + types = [Spmc, Mpmc, AsyncChannel, Flume], + consts = [0, 4], + args = CONSUMERS, + sample_count = 20, + sample_size = 1, + counter = ItemsCount::new(BATCH_MESSAGES), +)] +fn tokio_tasks(bencher: Bencher, consumers: usize) +where + Unbounded: Channel, +{ + let runtime = runtime(WORKERS); + bencher + .with_inputs(|| TaskBatch::new::>(&runtime, consumers)) + .bench_local_refs(|batch| runtime.block_on(batch.run())); +} diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index 44942255..e93720d7 100644 --- a/tests-integration/Cargo.toml +++ b/tests-integration/Cargo.toml @@ -48,6 +48,7 @@ asyncband = { workspace = true, features = [ "semaphore", "shutdown", "singleflight", + "spmc", "waitgroup", "watch", ] } diff --git a/tests-integration/tests/spmc_test.rs b/tests-integration/tests/spmc_test.rs new file mode 100644 index 00000000..6c9e3535 --- /dev/null +++ b/tests-integration/tests/spmc_test.rs @@ -0,0 +1,356 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::cell::Cell; +use std::future::Future; +use std::pin::pin; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::task::Context; +use std::task::Poll; +use std::task::Waker; + +use asyncband::spmc; +use asyncband::spmc::RecvError; +use asyncband::spmc::TryRecvError; +use asyncband::spmc::TrySendError; +use tests_integration::WakeCounter; +use tests_integration::poll_once; + +#[derive(Debug)] +struct DropSpy(Arc); + +impl Drop for DropSpy { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::SeqCst); + } +} + +macro_rules! receiver_contract { + ($name:ident, $channel:expr, $send:ident) => { + mod $name { + use super::*; + + #[test] + fn receivers_compete_in_fifo_order_and_drain_after_sender_drop() { + let (mut sender, receiver) = $channel; + let competing = receiver.clone(); + sender.$send(10).unwrap(); + sender.$send(20).unwrap(); + assert_eq!(receiver.try_recv(), Ok(10)); + assert_eq!(competing.try_recv(), Ok(20)); + assert_eq!(receiver.try_recv(), Err(TryRecvError::Empty)); + sender.$send(30).unwrap(); + sender.$send(40).unwrap(); + drop(sender); + assert_eq!(poll_once(pin!(receiver.recv())), Poll::Ready(Ok(30))); + assert_eq!(poll_once(pin!(competing.recv())), Poll::Ready(Ok(40))); + assert_eq!(receiver.try_recv(), Err(TryRecvError::Disconnected)); + assert_eq!( + poll_once(pin!(competing.recv())), + Poll::Ready(Err(RecvError::Disconnected)) + ); + } + + #[test] + fn only_last_receiver_disconnects_and_returns_unsent_value() { + let (mut sender, receiver) = $channel; + let competing = receiver.clone(); + drop(receiver); + sender.$send(10).unwrap(); + assert_eq!(competing.try_recv(), Ok(10)); + drop(competing); + let error = sender.$send(20).unwrap_err(); + assert_eq!(error.as_inner(), &20); + assert_eq!(error.into_inner(), 20); + } + + #[test] + fn one_send_wakes_one_of_eight_receivers_and_cancellation_hands_off() { + let (mut sender, receiver) = $channel; + let receivers: Vec<_> = (0..8).map(|_| receiver.clone()).collect(); + let counts: Vec<_> = (0..8).map(|_| Arc::new(WakeCounter::default())).collect(); + let wakers: Vec<_> = counts.iter().cloned().map(Waker::from).collect(); + let mut pending: Vec<_> = receivers.iter().map(|rx| Box::pin(rx.recv())).collect(); + for (future, waker) in pending.iter_mut().zip(&wakers) { + assert!( + future + .as_mut() + .poll(&mut Context::from_waker(waker)) + .is_pending() + ); + } + sender.$send(7).unwrap(); + assert_eq!(counts[0].count(), 1); + assert!(counts[1..].iter().all(|count| count.count() == 0)); + + // Every selected receiver cancels before consuming. The message and notification + // must survive the entire chain without waking every remaining receiver at once. + for next in 1..8 { + drop(pending.remove(0)); + assert_eq!(counts[next].count(), 1); + assert!(counts[next + 1..].iter().all(|count| count.count() == 0)); + } + assert_eq!(poll_once(pending[0].as_mut()), Poll::Ready(Ok(7))); + assert_eq!(receiver.try_recv(), Err(TryRecvError::Empty)); + } + + #[test] + fn cancelling_before_notification_removes_the_waiter() { + let (mut sender, receiver) = $channel; + let competing = receiver.clone(); + let cancelled_count = Arc::new(WakeCounter::default()); + let waiting_count = Arc::new(WakeCounter::default()); + let cancelled_waker = Waker::from(cancelled_count.clone()); + let waiting_waker = Waker::from(waiting_count.clone()); + let mut cancelled = Box::pin(receiver.recv()); + let mut waiting = Box::pin(competing.recv()); + assert!( + cancelled + .as_mut() + .poll(&mut Context::from_waker(&cancelled_waker)) + .is_pending() + ); + assert!( + waiting + .as_mut() + .poll(&mut Context::from_waker(&waiting_waker)) + .is_pending() + ); + drop(cancelled); + sender.$send(5).unwrap(); + assert_eq!(cancelled_count.count(), 0); + assert_eq!(waiting_count.count(), 1); + assert_eq!(poll_once(waiting.as_mut()), Poll::Ready(Ok(5))); + } + + #[test] + fn sender_disconnection_wakes_all_receivers() { + let (sender, receiver) = $channel; + let receivers: Vec<_> = (0..8).map(|_| receiver.clone()).collect(); + let counts: Vec<_> = (0..8).map(|_| Arc::new(WakeCounter::default())).collect(); + let wakers: Vec<_> = counts.iter().cloned().map(Waker::from).collect(); + let mut pending: Vec<_> = receivers.iter().map(|rx| Box::pin(rx.recv())).collect(); + for (future, waker) in pending.iter_mut().zip(&wakers) { + assert!( + future + .as_mut() + .poll(&mut Context::from_waker(waker)) + .is_pending() + ); + } + drop(sender); + assert!(counts.iter().all(|count| count.count() == 1)); + for future in &mut pending { + assert_eq!( + poll_once(future.as_mut()), + Poll::Ready(Err(RecvError::Disconnected)) + ); + } + // Fix the payload type without sending into a disconnected queue. + let _: Result = receiver.try_recv(); + } + + #[test] + fn buffered_received_and_rejected_values_are_each_dropped_once() { + let (mut sender, receiver) = $channel; + let competing = receiver.clone(); + let drops: Vec<_> = (0..4).map(|_| Arc::new(AtomicUsize::new(0))).collect(); + sender.$send(DropSpy(drops[0].clone())).unwrap(); + sender.$send(DropSpy(drops[1].clone())).unwrap(); + let received = receiver.try_recv().unwrap(); + sender.$send(DropSpy(drops[2].clone())).unwrap(); + drop(receiver); + assert!(drops.iter().all(|count| count.load(Ordering::SeqCst) == 0)); + drop(competing); + assert_eq!(drops[1].load(Ordering::SeqCst), 1); + assert_eq!(drops[2].load(Ordering::SeqCst), 1); + let rejected = sender.$send(DropSpy(drops[3].clone())).unwrap_err(); + assert_eq!(drops[3].load(Ordering::SeqCst), 0); + drop(rejected.into_inner()); + drop(received); + drop(sender); + assert!(drops.iter().all(|count| count.load(Ordering::SeqCst) == 1)); + } + } + }; +} + +receiver_contract!(bounded, spmc::bounded(2), try_send); +receiver_contract!(unbounded, spmc::unbounded(), send); + +#[test] +#[should_panic(expected = "spmc bounded queue requires capacity > 0")] +fn bounded_rejects_zero_capacity() { + let _ = spmc::bounded::<()>(0); +} + +#[test] +fn bounded_capacity_and_pending_send_progress() { + for capacity in [1, 2, 3, 8] { + let (mut sender, receiver) = spmc::bounded(capacity); + for value in 0..capacity { + sender.try_send(value).unwrap(); + } + assert_eq!(sender.try_send(capacity), Err(TrySendError::Full(capacity))); + let mut waiting = Box::pin(sender.send(capacity)); + let count = Arc::new(WakeCounter::default()); + let waker = Waker::from(count.clone()); + assert!( + waiting + .as_mut() + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); + assert_eq!(receiver.try_recv(), Ok(0)); + assert_eq!(count.count(), 1); + assert_eq!(poll_once(waiting.as_mut()), Poll::Ready(Ok(()))); + drop(waiting); + drop(sender); + for value in 1..=capacity { + assert_eq!(receiver.try_recv(), Ok(value)); + } + assert_eq!(receiver.try_recv(), Err(TryRecvError::Disconnected)); + } +} + +#[test] +fn cancelling_a_send_before_or_after_notification_preserves_capacity() { + for notified in [false, true] { + let (mut sender, receiver) = spmc::bounded(1); + let drops = Arc::new(AtomicUsize::new(0)); + sender.try_send(DropSpy(drops.clone())).unwrap(); + let mut cancelled = Box::pin(sender.send(DropSpy(drops.clone()))); + assert!(poll_once(cancelled.as_mut()).is_pending()); + if notified { + drop(receiver.try_recv().unwrap()); + } + drop(cancelled); + assert_eq!(drops.load(Ordering::SeqCst), 1 + usize::from(notified)); + if !notified { + drop(receiver.try_recv().unwrap()); + } + assert_eq!(drops.load(Ordering::SeqCst), 2); + assert!(matches!( + poll_once(pin!(sender.send(DropSpy(drops.clone())))), + Poll::Ready(Ok(())) + )); + drop(receiver); + drop(sender); + assert_eq!(drops.load(Ordering::SeqCst), 3); + } +} + +#[test] +fn last_receiver_wakes_pending_sender_and_returns_its_value() { + let (mut sender, receiver) = spmc::bounded(1); + let competing = receiver.clone(); + sender.try_send(0).unwrap(); + let count = Arc::new(WakeCounter::default()); + let waker = Waker::from(count.clone()); + let mut pending = Box::pin(sender.send(1)); + assert!( + pending + .as_mut() + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); + drop(receiver); + assert_eq!(count.count(), 0); + drop(competing); + assert_eq!(count.count(), 1); + let Poll::Ready(Err(error)) = poll_once(pending.as_mut()) else { + panic!("disconnected sender must return the unsent value"); + }; + assert_eq!(error.into_inner(), 1); +} + +#[test] +fn endpoint_and_future_traits_allow_send_but_not_sync_payloads() { + fn assert_traits() {} + fn assert_send(_: T) {} + assert_traits::>>(); + assert_traits::>>(); + assert_traits::>>(); + assert_traits::>>(); + let (mut sender, receiver) = spmc::bounded::>(1); + assert_send(sender.send(Cell::new(1))); + assert_send(receiver.recv()); + let (_sender, receiver) = spmc::unbounded::>(); + assert_send(receiver.recv()); +} + +// Exercise the single producer on a different worker from eight competing consumers. Keep Miri +// focused on the deterministic notification, cancellation, and destruction contracts above. +#[cfg(not(miri))] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn one_producer_delivers_every_value_once_to_eight_competing_consumers() { + use std::time::Duration; + + macro_rules! run { + ($channel:expr, $send:expr) => {{ + const TOTAL: usize = 2_048; + let (mut sender, receiver) = $channel; + let start = Arc::new(tokio::sync::Barrier::new(9)); + let consumers: Vec<_> = (0..8) + .map(|_| { + let receiver = receiver.clone(); + let start = start.clone(); + tokio::spawn(async move { + start.wait().await; + let mut values = Vec::new(); + while let Ok(value) = receiver.recv().await { + values.push(value); + } + values + }) + }) + .collect(); + drop(receiver); + let producer = tokio::spawn(async move { + start.wait().await; + for value in 0..TOTAL { + $send(&mut sender, value).await; + } + }); + tokio::time::timeout(Duration::from_secs(10), async { + producer.await.unwrap(); + let mut received = Vec::new(); + for consumer in consumers { + let values = consumer.await.unwrap(); + assert!(values.windows(2).all(|pair| pair[0] < pair[1])); + received.extend(values); + } + received.sort_unstable(); + assert_eq!(received, (0..TOTAL).collect::>()); + }) + .await + .expect("SPMC sender and all consumers must make progress"); + }}; + } + + async fn bounded_send(sender: &mut spmc::BoundedSender, value: usize) { + sender.send(value).await.unwrap(); + } + async fn unbounded_send(sender: &mut spmc::UnboundedSender, value: usize) { + sender.send(value).unwrap(); + } + run!(spmc::bounded(1), bounded_send); + run!(spmc::bounded(64), bounded_send); + run!(spmc::unbounded(), unbounded_send); +} diff --git a/xtask/src/main.rs b/xtask/src/main.rs index 51fb9866..a6449358 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -121,6 +121,7 @@ impl CommandMiri { &["--test", "unsafe_paths_test"], )); run_command(make_miri_cmd("tests-integration", &["--test", "mpsc_test"])); + run_command(make_miri_cmd("tests-integration", &["--test", "spmc_test"])); run_command(make_miri_cmd( "tests-integration", &["--test", "phaser_test"],