sync: Update broadcast::recv to return a named future

tokio/src/sync/broadcast.rs

@@ -119,10 +119,11 @@
 use crate::loom::cell::UnsafeCell;
 use crate::loom::sync::atomic::{AtomicBool, AtomicUsize};
 use crate::loom::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard};
-use crate::runtime::coop::cooperative;
+use crate::runtime::coop::{cooperative, Coop};
 use crate::util::linked_list::{self, GuardedLinkedList, LinkedList};
 use crate::util::WakeList;
 
+use pin_project_lite::pin_project;
 use std::fmt;
 use std::future::Future;
 use std::marker::PhantomPinned;
@@ -389,8 +390,30 @@ struct RecvGuard<'a, T> {
     slot: RwLockReadGuard<'a, Slot<T>>,
 }
 
+pin_project! {
+    /// Future for the [`Receiver::recv`] method.
+    pub struct Recv<'a, T>
+    where
+        T: Clone,
+    {
+        #[pin]
+        inner: Coop<RecvInner<'a, T>>,
+    }
+}
+
+impl<'a, T> Future for Recv<'a, T>
+where
+    T: Clone,
+{
+    type Output = Result<T, RecvError>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.project().inner.poll(cx)
+    }
+}
+
 /// Receive a value future.
-struct Recv<'a, T> {
+struct RecvInner<'a, T> {
     /// Receiver being waited on.
     receiver: &'a mut Receiver<T>,
 
@@ -398,8 +421,8 @@ struct Recv<'a, T> {
     waiter: UnsafeCell<Waiter>,
 }
 
-unsafe impl<'a, T: Send> Send for Recv<'a, T> {}
-unsafe impl<'a, T: Send> Sync for Recv<'a, T> {}
+unsafe impl<'a, T: Send> Send for RecvInner<'a, T> {}
+unsafe impl<'a, T: Send> Sync for RecvInner<'a, T> {}
 
 /// Max number of receivers. Reserve space to lock.
 const MAX_RECEIVERS: usize = usize::MAX >> 2;
@@ -1262,8 +1285,10 @@ impl<T: Clone> Receiver<T> {
     ///     assert_eq!(30, rx.recv().await.unwrap());
     /// }
     /// ```
-    pub async fn recv(&mut self) -> Result<T, RecvError> {
-        cooperative(Recv::new(self)).await
+    pub fn recv(&mut self) -> Recv<'_, T> {
+        Recv {
+            inner: cooperative(RecvInner::new(self)),
+        }
     }
 
     /// Attempts to return a pending value on this receiver without awaiting.
@@ -1363,9 +1388,9 @@ impl<T> Drop for Receiver<T> {
     }
 }
 
-impl<'a, T> Recv<'a, T> {
-    fn new(receiver: &'a mut Receiver<T>) -> Recv<'a, T> {
-        Recv {
+impl<'a, T> RecvInner<'a, T> {
+    fn new(receiver: &'a mut Receiver<T>) -> RecvInner<'a, T> {
+        RecvInner {
             receiver,
             waiter: UnsafeCell::new(Waiter {
                 queued: AtomicBool::new(false),
@@ -1389,7 +1414,7 @@ impl<'a, T> Recv<'a, T> {
     }
 }
 
-impl<'a, T> Future for Recv<'a, T>
+impl<'a, T> Future for RecvInner<'a, T>
 where
     T: Clone,
 {
@@ -1411,7 +1436,7 @@ where
     }
 }
 
-impl<'a, T> Drop for Recv<'a, T> {
+impl<'a, T> Drop for RecvInner<'a, T> {
     fn drop(&mut self) {
         // Safety: `waiter.queued` is atomic.
         // Acquire ordering is required to synchronize with