sync: oneshot::Receiver::is_terminated()

this commit introduces a new method to
`tokio::sync::oneshot::Receiver<T>`.

broadly speaking, users of the oneshot channel are encouraged to
`.await` the `Receiver<T>` directly, as it will only yield a single
value.

users implementing their own `std::future::Future`s directly may instead
poll the receiver via
`<Receiver<T> as Future<Output = Result<T, RecvError>::poll(..)`.
note that the contract of `Future::poll()` states that clients should
not poll a future after it has yielded `Poll::Ready(value)`.

this commit provides a way to inspect the
state of a receiver, to avoid violating the contact of
`Future::poll(..)`, or requiring that a oneshot channel users track this
state themselves externally via mechanisms like
`futures::future::FusedFuture`, or wrapping the receiver in an
`Option<T>`.

NB: this makes a small behavioral change to the implementation of
`<Receiver<T> as Future<Output = Result<T, RecvError>::poll(..)`. this
change is acceptable, per the `Future::poll()` documentation regarding
panics:

> Once a future has completed (returned Ready from poll), calling its
poll method again may panic, block forever, or cause other kinds of
problems; the Future trait places no requirements on the effects of such
a call.

the upside of this is that it means a broken or closed channel, e.g.
when the sender is dropped, will settle as "finished" after it yields an
error.

see:
* tokio-rs#7137 (comment)
* https://doc.rust-lang.org/stable/std/future/trait.Future.html#panics

Signed-off-by: katelyn martin <[email protected]>

Author: cratelyn

tokio/src/sync/oneshot.rs

@@ -931,6 +931,67 @@ impl<T> Receiver<T> {
         }
     }
 
+    /// Checks if this receiver is terminated.
+    ///
+    /// This function returns true if this receiver has already yielded a [`Poll::Ready`] result.
+    /// If so, this receiver should no longer be polled.
+    ///
+    /// # Examples
+    ///
+    /// Sending a value and polling it.
+    ///
+    /// ```
+    /// use futures::task::noop_waker_ref;
+    /// use tokio::sync::oneshot;
+    ///
+    /// use std::future::Future;
+    /// use std::pin::Pin;
+    /// use std::task::{Context, Poll};
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let (tx, mut rx) = oneshot::channel();
+    ///
+    ///     // A receiver is not terminated when it is initialized.
+    ///     assert!(!rx.is_terminated());
+    ///
+    ///     // A receiver is not terminated it is polled and is still pending.
+    ///     let poll = Pin::new(&mut rx).poll(&mut Context::from_waker(noop_waker_ref()));
+    ///     assert_eq!(poll, Poll::Pending);
+    ///     assert!(!rx.is_terminated());
+    ///
+    ///     // A receiver is not terminated if a value has been sent, but not yet read.
+    ///     tx.send(0).unwrap();
+    ///     assert!(!rx.is_terminated());
+    ///
+    ///     // A receiver *is* terminated after it has been polled and yielded a value.
+    ///     assert_eq!((&mut rx).await, Ok(0));
+    ///     assert!(rx.is_terminated());
+    /// }
+    /// ```
+    ///
+    /// Dropping the sender.
+    ///
+    /// ```
+    /// use tokio::sync::oneshot;
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let (tx, mut rx) = oneshot::channel::<()>();
+    ///
+    ///     // A receiver is not immediately terminated when the sender is dropped.
+    ///     drop(tx);
+    ///     assert!(!rx.is_terminated());
+    ///
+    ///     // A receiver *is* terminated after it has been polled and yielded an error.
+    ///     let _ = (&mut rx).await.unwrap_err();
+    ///     assert!(rx.is_terminated());
+    /// }
+    /// ```
+    pub fn is_terminated(&self) -> bool {
+        self.inner.is_none()
+    }
+
     /// Attempts to receive a value.
     ///
     /// If a pending value exists in the channel, it is returned. If no value
@@ -1106,18 +1167,18 @@ impl<T> Future for Receiver<T> {
 
         let ret = if let Some(inner) = self.as_ref().get_ref().inner.as_ref() {
             #[cfg(all(tokio_unstable, feature = "tracing"))]
-            let res = ready!(trace_poll_op!("poll_recv", inner.poll_recv(cx)))?;
+            let res = ready!(trace_poll_op!("poll_recv", inner.poll_recv(cx))).map_err(Into::into);
 
             #[cfg(any(not(tokio_unstable), not(feature = "tracing")))]
-            let res = ready!(inner.poll_recv(cx))?;
+            let res = ready!(inner.poll_recv(cx)).map_err(Into::into);
 
             res
         } else {
             panic!("called after complete");
         };
 
         self.inner = None;
-        Ready(Ok(ret))
+        Ready(ret)
     }
 }
 

tokio/tests/sync_oneshot.rs

@@ -292,3 +292,74 @@ fn sender_changes_task() {
 
     assert_ready!(task2.enter(|cx, _| tx.poll_closed(cx)));
 }
+
+#[test]
+fn receiver_is_terminated_send() {
+    let (tx, mut rx) = oneshot::channel::<i32>();
+
+    assert!(
+        !rx.is_terminated(),
+        "channel is NOT terminated before value is sent"
+    );
+    tx.send(17).unwrap();
+    assert!(
+        !rx.is_terminated(),
+        "channel is NOT terminated after value is sent"
+    );
+
+    let mut task = task::spawn(());
+    let poll = task.enter(|cx, _| Pin::new(&mut rx).poll(cx));
+    assert_ready_eq!(poll, Ok(17));
+
+    assert!(
+        rx.is_terminated(),
+        "channel IS terminated after value is read"
+    );
+}
+
+#[test]
+fn receiver_is_terminated_drop() {
+    let (tx, mut rx) = oneshot::channel::<i32>();
+
+    assert!(
+        !rx.is_terminated(),
+        "channel is NOT terminated before sender is dropped"
+    );
+    drop(tx);
+    assert!(
+        !rx.is_terminated(),
+        "channel is NOT terminated after sender is dropped"
+    );
+
+    let mut task = task::spawn(());
+    let poll = task.enter(|cx, _| Pin::new(&mut rx).poll(cx));
+    assert_ready_err!(poll);
+
+    assert!(
+        rx.is_terminated(),
+        "channel IS terminated after value is read"
+    );
+}
+
+#[test]
+fn receiver_is_terminated_rx_close() {
+    let (_tx, mut rx) = oneshot::channel::<i32>();
+    assert!(
+        !rx.is_terminated(),
+        "channel is NOT terminated before closing"
+    );
+    rx.close();
+    assert!(
+        !rx.is_terminated(),
+        "channel is NOT terminated before closing"
+    );
+
+    let mut task = task::spawn(());
+    let poll = task.enter(|cx, _| Pin::new(&mut rx).poll(cx));
+    assert_ready_err!(poll);
+
+    assert!(
+        rx.is_terminated(),
+        "channel IS terminated after value is read"
+    );
+}