feat(broadcast): return shared messages from recv

CHANGELOG.md

@@ -23,6 +23,10 @@ All notable changes to this project will be documented in this file.
 
 ## Unreleased
 
+### Breaking changes
+
+* Return shared messages from `broadcast::mpmc` receivers: `recv` and `try_recv` now yield `Arc<T>`, and `bounded` and `unbounded` no longer require `T: Clone`. Callers that need an owned value use the new `recv_cloned` and `try_recv_cloned`, which still move the payload out instead of cloning it when nothing else still refers to the message; callers that sent `Arc<T>` to avoid cloning can send `T` directly. A retained `Arc<T>` keeps its payload alive after the message leaves the backlog, so a bounded channel's capacity limits retained messages rather than memory.
+
 ### New features
 
 * Add bounded MPMC `reserve` and `try_reserve` methods returning a borrowed `Permit`, so callers can wait for capacity before constructing a value; sends and reservations receive capacity in wait-queue order, and unused permits release it.

asyncband/src/broadcast/mpmc/bounded/mod.rs

@@ -66,10 +66,10 @@
 //! tx.send(10).await;
 //! tx.send(20).await;
 //!
-//! assert_eq!(rx1.recv().await, Ok(10));
-//! assert_eq!(rx1.recv().await, Ok(20));
-//! assert_eq!(rx2.recv().await, Ok(10));
-//! assert_eq!(rx2.recv().await, Ok(20));
+//! assert_eq!(*rx1.recv().await.unwrap(), 10);
+//! assert_eq!(*rx1.recv().await.unwrap(), 20);
+//! assert_eq!(*rx2.recv().await.unwrap(), 10);
+//! assert_eq!(*rx2.recv().await.unwrap(), 20);
 //! # }
 //! ```
 //!
@@ -87,7 +87,7 @@
 //! assert_eq!(tx.try_send(3), Err(TrySendError::Full(3)));
 //!
 //! // `rx1` draining is not enough: `rx2` has read neither message, so both stay retained.
-//! assert_eq!(rx1.try_recv(), Ok(1));
+//! assert_eq!(*rx1.try_recv().unwrap(), 1);
 //! assert_eq!(tx.retained_message_count(), 2);
 //! assert_eq!(tx.try_send(3), Err(TrySendError::Full(3)));
 //!
@@ -112,6 +112,7 @@ use std::task::Poll;
 use super::common;
 use super::common::Backlog;
 use super::common::Inner;
+use super::common::Received;
 use super::error::RecvError;
 use super::error::TryRecvError;
 use super::error::TrySendError;
@@ -142,10 +143,10 @@ mod tests;
 ///
 /// let (tx, mut rx) = mpmc::bounded(1);
 /// tx.try_send(10).unwrap();
-/// assert_eq!(rx.try_recv(), Ok(10));
+/// assert_eq!(*rx.try_recv().unwrap(), 10);
 /// ```
 #[track_caller]
-pub fn bounded<T: Clone>(capacity: usize) -> (BoundedSender<T>, BoundedReceiver<T>) {
+pub fn bounded<T>(capacity: usize) -> (BoundedSender<T>, BoundedReceiver<T>) {
     assert!(
         capacity > 0,
         "broadcast bounded channel requires capacity > 0"
@@ -283,7 +284,7 @@ impl<T> BoundedSender<T> {
     /// # async fn main() {
     /// let (tx, mut rx) = mpmc::bounded(1);
     /// tx.send(10).await;
-    /// assert_eq!(rx.recv().await, Ok(10));
+    /// assert_eq!(*rx.recv().await.unwrap(), 10);
     /// # }
     /// ```
     pub async fn send(&self, value: T) {
@@ -380,7 +381,7 @@ impl<T> BoundedSender<T> {
     /// tx.try_send(10).unwrap();
     /// assert_eq!(tx.try_send(20), Err(TrySendError::Full(20)));
     ///
-    /// assert_eq!(rx.try_recv(), Ok(10));
+    /// assert_eq!(*rx.try_recv().unwrap(), 10);
     /// tx.try_send(20).unwrap();
     /// ```
     pub fn try_send(&self, value: T) -> Result<(), TrySendError<T>> {
@@ -445,7 +446,7 @@ impl<T> BoundedSender<T> {
     /// tx.try_send(10).unwrap();
     /// assert_eq!(tx.retained_message_count(), 1);
     ///
-    /// assert_eq!(rx.try_recv(), Ok(10));
+    /// assert_eq!(*rx.try_recv().unwrap(), 10);
     /// assert_eq!(tx.retained_message_count(), 0);
     /// ```
     pub fn retained_message_count(&self) -> usize {
@@ -488,7 +489,7 @@ impl<T> BoundedSender<T> {
     /// let mut rx = tx.subscribe();
     /// assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
     /// tx.send(20).await;
-    /// assert_eq!(rx.recv().await, Ok(20));
+    /// assert_eq!(*rx.recv().await.unwrap(), 20);
     /// # }
     /// ```
     #[must_use = "the receiver is dropped immediately if it is not retained"]
@@ -537,12 +538,16 @@ impl<T> Drop for BoundedReceiver<T> {
     }
 }
 
-impl<T: Clone> BoundedReceiver<T> {
-    /// Receives the next value for this receiver.
+impl<T> BoundedReceiver<T> {
+    /// Receives the next value for this receiver as the channel's shared `Arc<T>`.
+    ///
+    /// Receiving never clones `T`. The `Arc` outlives the message's stay in the backlog: a
+    /// caller that keeps it keeps the payload alive but holds no capacity. Use
+    /// [`recv_cloned`](BoundedReceiver::recv_cloned) to take an owned value instead.
     ///
     /// # Returns
     ///
-    /// * `Ok(T)`: The next message.
+    /// * `Ok(Arc<T>)`: The next message.
     /// * `Err(RecvError::Disconnected)`: All senders have been dropped and this receiver has no
     ///   remaining messages.
     ///
@@ -561,22 +566,23 @@ impl<T: Clone> BoundedReceiver<T> {
     /// # async fn main() {
     /// let (tx, mut rx) = mpmc::bounded(4);
     /// tx.send(10).await;
-    /// assert_eq!(rx.recv().await, Ok(10));
+    /// assert_eq!(*rx.recv().await.unwrap(), 10);
     /// # }
     /// ```
-    pub async fn recv(&mut self) -> Result<T, RecvError> {
-        Recv {
-            receiver: self,
-            token: None,
-        }
-        .await
+    pub async fn recv(&mut self) -> Result<Arc<T>, RecvError> {
+        let (msg, reclaimed) = self.wait_next().await?;
+        Ok(common::share_msg(msg, reclaimed))
     }
 
-    /// Attempts to receive the next value for this receiver without blocking.
+    /// Attempts to receive the next value for this receiver without blocking, as the channel's
+    /// shared `Arc<T>`.
+    ///
+    /// See [`recv`](BoundedReceiver::recv) for what keeping the `Arc` means, and
+    /// [`try_recv_cloned`](BoundedReceiver::try_recv_cloned) to take an owned value instead.
     ///
     /// # Returns
     ///
-    /// * `Ok(T)`: The next message.
+    /// * `Ok(Arc<T>)`: The next message.
     /// * `Err(TryRecvError::Empty)`: No message is currently available.
     /// * `Err(TryRecvError::Disconnected)`: All senders have been dropped and this receiver has no
     ///   remaining messages.
@@ -588,21 +594,26 @@ impl<T: Clone> BoundedReceiver<T> {
     ///
     /// let (tx, mut rx) = mpmc::bounded(4);
     /// tx.try_send(10).unwrap();
-    /// assert_eq!(rx.try_recv(), Ok(10));
+    /// assert_eq!(*rx.try_recv().unwrap(), 10);
     /// ```
-    pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
+    pub fn try_recv(&mut self) -> Result<Arc<T>, TryRecvError> {
+        let (msg, reclaimed) = self.receive_released()?;
+        Ok(common::share_msg(msg, reclaimed))
+    }
+
+    /// Receives without waiting and hands the capacity this receive released to blocked producers
+    /// before the caller can touch the payload.
+    ///
+    /// `T::clone` and `T::drop` may panic, and if either ran first the reclaimed slots would never
+    /// be handed to a waiting producer, stalling it permanently. The `Recv` future orders its poll
+    /// the same way.
+    fn receive_released(&mut self) -> Result<Received<T>, TryRecvError> {
         let (msg, reclaimed) =
             common::try_receive(&self.shared.inner, &self.shared.senders, self.key)?;
-
-        // Release before taking the payload: `take_msg` runs `T::clone` and `T::drop`, and if
-        // either panics the reclaimed slots would otherwise never be handed to a waiting producer,
-        // stalling it permanently.
         self.shared.release_reclaimed(reclaimed.len());
-        Ok(common::take_msg(msg, reclaimed))
+        Ok((msg, reclaimed))
     }
-}
 
-impl<T> BoundedReceiver<T> {
     /// Re-subscribes to the channel, returning a new receiver that starts receiving messages from
     /// the *current* tail of the channel.
     ///
@@ -622,7 +633,7 @@ impl<T> BoundedReceiver<T> {
     /// let mut rx2 = rx.resubscribe();
     /// tx.try_send(3).unwrap();
     ///
-    /// assert_eq!(rx2.try_recv(), Ok(3));
+    /// assert_eq!(*rx2.try_recv().unwrap(), 3);
     /// ```
     #[must_use = "the receiver is dropped immediately if it is not retained"]
     pub fn resubscribe(&self) -> Self {
@@ -654,12 +665,83 @@ impl<T> BoundedReceiver<T> {
     /// tx.try_send(20).unwrap();
     /// assert_eq!(rx.unread_message_count(), 2);
     ///
-    /// assert_eq!(rx.try_recv(), Ok(10));
+    /// assert_eq!(*rx.try_recv().unwrap(), 10);
     /// assert_eq!(rx.unread_message_count(), 1);
     /// ```
     pub fn unread_message_count(&self) -> usize {
         self.shared.inner.lock().log.unread(self.key)
     }
+
+    fn wait_next(&mut self) -> Recv<'_, T> {
+        Recv {
+            receiver: self,
+            token: None,
+        }
+    }
+}
+
+impl<T: Clone> BoundedReceiver<T> {
+    /// Receives the next value for this receiver as an owned `T`.
+    ///
+    /// The payload is moved out of the channel without cloning when this read reclaims the message
+    /// and nothing else still refers to it, such as an `Arc` kept from a shared receive or another
+    /// receiver's clone still in progress. Otherwise it is cloned with the channel unlocked. If
+    /// `T::clone` panics, this receiver has still advanced past the value.
+    ///
+    /// # Returns
+    ///
+    /// * `Ok(T)`: The next message.
+    /// * `Err(RecvError::Disconnected)`: All senders have been dropped and this receiver has no
+    ///   remaining messages.
+    ///
+    /// # Cancel safety
+    ///
+    /// Dropping a pending `recv_cloned` future leaves this receiver's cursor unchanged. A
+    /// subsequent receive can still return the same next value, so these futures may safely be
+    /// raced with other futures in a selection construct.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use asyncband::broadcast::mpmc;
+    ///
+    /// # #[tokio::main]
+    /// # async fn main() {
+    /// let (tx, mut rx) = mpmc::bounded(4);
+    /// tx.send(String::from("owned")).await;
+    /// assert_eq!(rx.recv_cloned().await, Ok(String::from("owned")));
+    /// # }
+    /// ```
+    pub async fn recv_cloned(&mut self) -> Result<T, RecvError> {
+        let (msg, reclaimed) = self.wait_next().await?;
+        Ok(common::take_msg(msg, reclaimed))
+    }
+
+    /// Attempts to receive the next value for this receiver without blocking, as an owned `T`.
+    ///
+    /// See [`recv_cloned`](BoundedReceiver::recv_cloned) for when the payload is moved out rather
+    /// than cloned.
+    ///
+    /// # Returns
+    ///
+    /// * `Ok(T)`: The next message.
+    /// * `Err(TryRecvError::Empty)`: No message is currently available.
+    /// * `Err(TryRecvError::Disconnected)`: All senders have been dropped and this receiver has no
+    ///   remaining messages.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use asyncband::broadcast::mpmc;
+    ///
+    /// let (tx, mut rx) = mpmc::bounded(4);
+    /// tx.try_send(String::from("owned")).unwrap();
+    /// assert_eq!(rx.try_recv_cloned(), Ok(String::from("owned")));
+    /// ```
+    pub fn try_recv_cloned(&mut self) -> Result<T, TryRecvError> {
+        let (msg, reclaimed) = self.receive_released()?;
+        Ok(common::take_msg(msg, reclaimed))
+    }
 }
 
 struct Recv<'a, T> {
@@ -683,8 +765,8 @@ impl<T> Drop for Recv<'_, T> {
     }
 }
 
-impl<T: Clone> Future for Recv<'_, T> {
-    type Output = Result<T, RecvError>;
+impl<T> Future for Recv<'_, T> {
+    type Output = Result<Received<T>, RecvError>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let Self { receiver, token } = self.get_mut();
@@ -701,9 +783,10 @@ impl<T: Clone> Future for Recv<'_, T> {
             Poll::Ready(Ok(received)) => received,
         };
 
-        // Release before taking the payload, for the same reason as `try_recv`: a panicking
-        // `T::clone` must not strand producers waiting for the reclaimed slots.
+        // Release before handing out the payload, for the same reason as `receive_released`: a
+        // panicking `T::clone` or `T::drop` must not strand producers waiting for the reclaimed
+        // slots.
         receiver.shared.release_reclaimed(reclaimed.len());
-        Poll::Ready(Ok(common::take_msg(msg, reclaimed)))
+        Poll::Ready(Ok((msg, reclaimed)))
     }
 }

asyncband/src/broadcast/mpmc/bounded/tests.rs

@@ -50,10 +50,10 @@ fn buffer_is_preallocated_and_never_shrinks() {
         tx.try_send(i).unwrap();
     }
     for i in 0..capacity {
-        assert_eq!(rx.try_recv(), Ok(i));
+        assert_eq!(rx.try_recv().as_deref(), Ok(&i));
     }
     tx.try_send(0).unwrap();
-    assert_eq!(rx.try_recv(), Ok(0));
+    assert_eq!(rx.try_recv().as_deref(), Ok(&0));
 
     assert_eq!(tx.retained_message_count(), 0);
     assert_eq!(tx.shared.inner.lock().log.buffer_capacity(), allocated);

asyncband/src/broadcast/mpmc/common.rs

@@ -503,12 +503,22 @@ pub fn poll_receive<T>(
     }
 }
 
-/// Drops the reclaimed backlog, then yields the received message, both with the channel unlocked.
+/// Releases the backlog's reference to the message, then hands out the stored `Arc`, both with the
+/// channel unlocked.
+///
+/// A receive reclaims at most the message itself, and `msg` keeps the payload alive, so a shared
+/// receive runs neither `T::clone` nor `T::drop`.
+pub fn share_msg<T>(msg: Arc<T>, reclaimed: Reclaimed<T>) -> Arc<T> {
+    drop(reclaimed);
+    msg
+}
+
+/// Drops the reclaimed backlog, then yields the received message owned, both with the channel
+/// unlocked.
 ///
 /// A non-empty backlog means receiving this message removed `msg` from the buffer. Once the
 /// reclaimed backlog is dropped, the caller may hold the only remaining reference and can then
-/// move the payload out instead of cloning it. A channel with a single receiver never clones a
-/// payload.
+/// move the payload out instead of cloning it.
 ///
 /// Ownership is decided from that bookkeeping rather than by probing the reference count. An
 /// [`Arc::try_unwrap`] call for every received message would fail under fan-out, and its failed
@@ -524,9 +534,9 @@ pub fn take_msg<T: Clone>(msg: Arc<T>, reclaimed: Reclaimed<T>) -> T {
         return (*msg).clone();
     }
 
-    // Another receiver can still hold an in-flight reference to the same message, so the clone
-    // remains the fallback.
-    Arc::try_unwrap(msg).unwrap_or_else(|msg| (*msg).clone())
+    // Another receiver can still hold an in-flight reference to the same message, or an `Arc` it
+    // kept from a shared receive, so the clone remains the fallback.
+    Arc::unwrap_or_clone(msg)
 }
 
 #[cfg(test)]

asyncband/src/broadcast/mpmc/mod.rs

@@ -27,9 +27,13 @@
 //!
 //! Receiving a value advances the subscription before returning that value. The channel tracks
 //! unread messages, not application work: retaining a received value or processing it
-//! asynchronously does not hold backlog capacity. There is no acknowledgement or
-//! processing-completion barrier. If cloning a received value panics, that subscription has still
-//! advanced past the value.
+//! asynchronously does not hold backlog capacity. `recv` and `try_recv` hand out the channel's own
+//! `Arc<T>`, so an `Arc` kept past the receive keeps its payload alive after the message has left
+//! the backlog, and a bounded channel's capacity limits retained messages rather than memory.
+//! Callers that wrapped values in an `Arc` only to avoid broadcast cloning can send `T` directly,
+//! since the channel already shares one `Arc<T>` with every receiver. There is no acknowledgement
+//! or processing-completion barrier. If `T::clone` panics inside `recv_cloned` or
+//! `try_recv_cloned`, that subscription has still advanced past the value.
 //!
 //! Sending with no subscriptions discards the value and succeeds. A later subscription starts
 //! with future publications; it does not replay discarded or previously retained values.