perf(mpmc): coordinate queue values and waiters under one lock

asyncband/src/internal/mod.rs

@@ -99,14 +99,13 @@ pub(crate) mod mutex;
 
 #[cfg(any(
     feature = "broadcast",
-    feature = "mpmc",
     feature = "mutex",
     feature = "rwlock",
     feature = "semaphore",
 ))]
-// Broadcast and MPMC 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.
+// Broadcast 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;
 

asyncband/src/internal/semaphore.rs

@@ -204,7 +204,7 @@ impl Semaphore {
     }
 
     /// Adds `n` permits to the semaphore if there is any waiter.
-    #[cfg(any(feature = "broadcast", feature = "mpmc"))]
+    #[cfg(feature = "broadcast")]
     pub fn release_if_nonempty(&self, n: usize) {
         let waiters = self.waiters.lock();
         if !waiters.is_empty() {
@@ -213,7 +213,7 @@ impl Semaphore {
     }
 
     /// Adds as many permits until there is no waiter.
-    #[cfg(any(feature = "broadcast", feature = "mpmc"))]
+    #[cfg(feature = "broadcast")]
     pub fn notify_all(&self) {
         let mut waiters = self.waiters.lock();
         let mut wakers = vec![];

asyncband/src/mpmc/bounded.rs

@@ -29,9 +29,9 @@ use super::queue::Shared;
 /// 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.
+/// Operations briefly acquire an internal mutex; no lock is held across an await point or while
+/// invoking waker callbacks or message destructors. The `try_*` methods do not wait for capacity
+/// or messages, but may wait to acquire this mutex.
 ///
 /// # Panics
 ///
@@ -84,9 +84,10 @@ impl<T> BoundedSender<T> {
     /// # 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.
+    /// returned `Pending` has not sent the value. If this call was woken for capacity that is still
+    /// free, cancelling it wakes 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
     }
@@ -137,8 +138,8 @@ impl<T> BoundedReceiver<T> {
     ///
     /// # 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.
+    /// Dropping a pending `recv` does not consume a value. If this call was woken for a value that
+    /// is still queued, cancelling it wakes the next waiting receiver instead.
     pub async fn recv(&self) -> Result<T, RecvError> {
         self.shared.recv().await
     }