feat(spmc): add competing queues

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.

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.                                                     |

asyncband/Cargo.toml

@@ -65,6 +65,7 @@ rwlock = []
 semaphore = []
 shutdown = ["latch", "waitgroup"]
 singleflight = ["dep:hashbrown", "once-cell"]
+spmc = []
 waitgroup = []
 watch = []
 

asyncband/src/mpmc/error.rs → asyncband/src/internal/competing_queue/error.rs

@@ -34,10 +34,10 @@ impl<T> SendError<T> {
     pub fn into_inner(self) -> T {
         self.0
     }
+}
 
-    pub(super) fn new(value: T) -> Self {
-        Self(value)
-    }
+pub fn send_error<T>(value: T) -> SendError<T> {
+    SendError(value)
 }
 
 impl<T> fmt::Display for SendError<T> {
@@ -54,7 +54,7 @@ impl<T> fmt::Debug for SendError<T> {
 
 impl<T> std::error::Error for SendError<T> {}
 
-/// 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<T> {
     /// The queue is full, so the value cannot be sent without waiting for capacity.

asyncband/src/mpmc/queue.rs → 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<T> {
+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<T> {
     state: Mutex<State<T>>,
     recv_waiters: Semaphore,
     send_waiters: Semaphore,
@@ -65,6 +70,7 @@ impl<T> Shared<T> {
         }
     }
 
+    #[cfg(feature = "mpmc")]
     pub fn clone_sender(&self) {
         let mut state = self.state.lock();
         state.senders = state
@@ -89,7 +95,7 @@ impl<T> Shared<T> {
         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<T> Shared<T> {
     pub async fn send(&self, value: T) -> Result<(), SendError<T>> {
         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<T> 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,
             };

asyncband/src/internal/mod.rs

@@ -56,6 +56,7 @@ pub(crate) fn wake_all(mut wakers: impl Iterator<Item = Waker>) {
     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<Item = Waker>) {
 #[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",

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![];

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")]

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.
 ///

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;

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<T> UnboundedSender<T> {
     pub fn send(&self, value: T) -> Result<(), SendError<T>> {
         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"),
         }
     }

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<T>(capacity: usize) -> (BoundedSender<T>, BoundedReceiver<T>) {
+    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<T> {
+    shared: Arc<Shared<T>>,
+}
+
+impl<T> fmt::Debug for BoundedSender<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("BoundedSender").finish_non_exhaustive()
+    }
+}
+
+impl<T> Drop for BoundedSender<T> {
+    fn drop(&mut self) {
+        self.shared.drop_sender();
+    }
+}
+
+impl<T> BoundedSender<T> {
+    /// 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<T>> {
+        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<T>> {
+        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<T> {
+    shared: Arc<Shared<T>>,
+}
+
+impl<T> Clone for BoundedReceiver<T> {
+    fn clone(&self) -> Self {
+        self.shared.clone_receiver();
+        Self {
+            shared: self.shared.clone(),
+        }
+    }
+}
+
+impl<T> fmt::Debug for BoundedReceiver<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("BoundedReceiver").finish_non_exhaustive()
+    }
+}
+
+impl<T> Drop for BoundedReceiver<T> {
+    fn drop(&mut self) {
+        self.shared.drop_receiver();
+    }
+}
+
+impl<T> BoundedReceiver<T> {
+    /// 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<T, RecvError> {
+        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<T, TryRecvError> {
+        self.shared.try_recv()
+    }
+}