feat(mpmc): add competing queues

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 
 ## Unreleased
 
+### 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.
+
 ### Bug fixes
 
 * Release MPSC receiver wakers when the receiver is dropped, avoiding retained tasks and ownership cycles when a waker holds a sender.

README.md

@@ -80,6 +80,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

@@ -60,6 +60,7 @@ pub(crate) mod atomic_waker;
     feature = "event",
     feature = "completion",
     feature = "latch",
+    feature = "mpmc",
     feature = "mpsc",
     feature = "mutex",
     feature = "rwlock",
@@ -90,6 +91,7 @@ pub(crate) mod value_cell;
     feature = "event",
     feature = "completion",
     feature = "latch",
+    feature = "mpmc",
     feature = "mpsc",
     feature = "mutex",
     feature = "rwlock",
@@ -101,14 +103,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",
@@ -125,6 +133,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

@@ -72,6 +72,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. |
@@ -132,6 +133,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,140 @@
+// 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.
+///
+/// # 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`]. This method is
+    /// cancel safe: cancelling a pending send leaves its value with the future and passes any
+    /// selected capacity notification to the next sender.
+    pub async fn send(&self, value: T) -> Result<(), SendError<T>> {
+        self.shared.send(value).await
+    }
+
+    /// Attempts to send a value without waiting.
+    ///
+    /// 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`]. This method is cancel safe and passes a
+    /// selected value notification to another receiver if the pending future is cancelled.
+    pub async fn recv(&self) -> Result<T, RecvError> {
+        self.shared.recv().await
+    }
+
+    /// Attempts to receive the next available value without waiting.
+    ///
+    /// 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()
+    }
+}