feat(mpmc): add competing queues

AGENTS.md

@@ -8,6 +8,14 @@ Use `cargo x` as the source of truth for repository workflows. Read `cargo x --h
 
 Declare restricted visibility at the module boundary and use `pub` for items in that module's API.
 
+## Waker Contract
+
+- Allow normal executor `Waker::clone` inside short state critical sections. Custom clone panic recovery and reentrancy are not general guarantees; do not require them in reviews unless an explicit local contract does.
+- Reuse borrowed-waker registration and avoid redundant clones.
+- Keep wake callbacks and replaced or cancelled waker destruction outside primitive locks. If a batch wake panics, attempt the remaining wakes and propagate the first panic.
+
+Decision: [#257](https://github.com/apache/asyncband/pull/257).
+
 ## Documentation
 
 Keep each Markdown prose paragraph and list item on one source line.

CHANGELOG.md

@@ -6,6 +6,7 @@ All notable changes to this project will be documented in this file.
 
 ### New features
 
+* 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

@@ -81,6 +81,7 @@ Runnable examples live in the [`examples`](examples) workspace crate. They demon
 |                            | [`Group`](https://docs.rs/asyncband/*/asyncband/singleflight/struct.Group.html)                | `singleflight` | Coalesce overlapping work per key without retaining completed values.                                         |
 | Communication              | [`Completion`](https://docs.rs/asyncband/*/asyncband/completion/struct.Completion.html)       | `completion`   | Publish one shared result to any number of current and future observers.                                       |
 |                            | [`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.      |
 |                            | [`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. |

asyncband/Cargo.toml

@@ -52,6 +52,7 @@ condvar = ["mutex"]
 event = []
 latch = []
 lazy-cell = ["mutex"]
+mpmc = []
 mpsc = []
 mutex = []
 once = ["semaphore"]

asyncband/src/internal/mod.rs

@@ -55,6 +55,7 @@ pub(crate) fn wake_all(mut wakers: impl Iterator<Item = Waker>) {
     feature = "event",
     feature = "completion",
     feature = "latch",
+    feature = "mpmc",
     feature = "mpsc",
     feature = "mutex",
     feature = "phaser",
@@ -83,6 +84,7 @@ pub(crate) mod value_cell;
     feature = "event",
     feature = "completion",
     feature = "latch",
+    feature = "mpmc",
     feature = "mpsc",
     feature = "mutex",
     feature = "phaser",
@@ -95,14 +97,20 @@ pub(crate) mod value_cell;
 #[allow(dead_code)]
 pub(crate) mod mutex;
 
-#[cfg(any(feature = "mutex", feature = "rwlock", feature = "semaphore"))]
-// Mutexes and rwlocks use the acquire/release operations; the public semaphore also exposes
-// permit accounting. Each single-primitive build leaves part of this shared API unused.
+#[cfg(any(
+    feature = "mpmc",
+    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.
 #[allow(dead_code)]
 pub(crate) mod semaphore;
 
 #[cfg(any(
     feature = "event",
+    feature = "mpmc",
     feature = "mpsc",
     feature = "mutex",
     feature = "rwlock",
@@ -119,6 +127,7 @@ pub(crate) mod waitlist;
     feature = "event",
     feature = "completion",
     feature = "latch",
+    feature = "mpmc",
     feature = "mpsc",
     feature = "mutex",
     feature = "once",

asyncband/src/internal/semaphore.rs

@@ -205,6 +205,41 @@ impl Semaphore {
         }
     }
 
+    /// Adds `n` permits to the semaphore if there is any waiter.
+    #[cfg(feature = "mpmc")]
+    pub fn release_if_nonempty(&self, n: usize) {
+        let waiters = self.waiters.lock();
+        if !waiters.is_empty() {
+            self.insert_permits_with_lock(n, waiters);
+        }
+    }
+
+    /// Adds as many permits until there is no waiter.
+    #[cfg(feature = "mpmc")]
+    pub fn notify_all(&self) {
+        let mut waiters = self.waiters.lock();
+        let mut wakers = vec![];
+        loop {
+            match waiters.unlink_first_waiter(|node| {
+                node.permits = 0;
+                true
+            }) {
+                None => break,
+                Some((id, waiter)) => {
+                    let remove_now = waiter.waker.is_none();
+                    if let Some(waker) = waiter.waker.take() {
+                        wakers.push(waker);
+                    }
+                    if remove_now {
+                        waiters.remove_unlinked_waiter(id);
+                    }
+                }
+            }
+        }
+        drop(waiters);
+        crate::internal::wake_all(wakers.into_iter());
+    }
+
     fn insert_permits_with_lock(
         &self,
         mut rem: usize,

asyncband/src/lib.rs

@@ -73,6 +73,7 @@
 //! |                            | [`Group`](singleflight::Group)                | `singleflight` | Coalesce overlapping work per key without retaining completed values.                                         |
 //! | Communication              | [`Completion`](completion::Completion)       | `completion`   | Publish one shared result to any number of current and future observers.                                       |
 //! |                            | [`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.      |
 //! |                            | [`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. |
@@ -133,6 +134,8 @@ pub mod condvar;
 pub mod event;
 #[cfg(feature = "latch")]
 pub mod latch;
+#[cfg(feature = "mpmc")]
+pub mod mpmc;
 #[cfg(feature = "mpsc")]
 pub mod mpsc;
 #[cfg(feature = "mutex")]

asyncband/src/mpmc/bounded.rs

@@ -0,0 +1,153 @@
+// 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 super::queue::Shared;
+
+/// Creates a bounded multi-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, "mpmc 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 can be cloned to add producers.
+pub struct BoundedSender<T> {
+    shared: Arc<Shared<T>>,
+}
+
+impl<T> Clone for BoundedSender<T> {
+    fn clone(&self) -> Self {
+        self.shared.clone_sender();
+        Self {
+            shared: self.shared.clone(),
+        }
+    }
+}
+
+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. Any selected capacity notification is passed to
+    /// the next waiting sender before `value` is dropped. Use [`try_send`](Self::try_send) when
+    /// the caller must retain ownership if capacity is unavailable.
+    pub async fn send(&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(&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 final 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 all senders have been dropped.
+    pub fn try_recv(&self) -> Result<T, TryRecvError> {
+        self.shared.try_recv()
+    }
+}