sync: oneshot::Receiver::is_finished()

this commit introduces a method that can be used to inquire whether a
oneshot channel's receiver has or has not yielded a value.

this is useful for callers that may be polling the receiver as a future,
to avoid inducing a panic. the receiver panics if polled after yielding
a `Poll::Ready<T>`. note that this 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.

NB: this commit makes one somewhat noteworthy change to the
implementation of `<Receiver<T> as Future>::poll()`. the inner state is
now taken when an error is yielded. this also follows the rules
proscribed by `std::future::Future::poll()`, to be clear!

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: <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

@@ -1028,6 +1028,67 @@ impl<T> Receiver<T> {
         }
     }
 
+    /// Checks if this receiver is finished.
+    ///
+    /// This returns true if this receiver has already yielded a [`Poll::Ready`] result, whether
+    /// that was a value `T`, or a [`RecvError`].
+    ///
+    /// # 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 finished when it is initialized.
+    ///     assert!(!rx.is_finished());
+    ///
+    ///     // A receiver is not finished 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);
+    ///
+    ///     // A receiver is not finished if a value has been sent, but not yet read.
+    ///     tx.send(0).unwrap();
+    ///     assert!(!rx.is_finished());
+    ///
+    ///     // A receiver *is* finished after it has been polled and yielded a value.
+    ///     let poll = Pin::new(&mut rx).poll(&mut Context::from_waker(noop_waker_ref()));
+    ///     assert_eq!(poll, Poll::Ready(Ok(0)));
+    ///     assert!(rx.is_finished());
+    /// }
+    /// ```
+    ///
+    /// Dropping the sender.
+    ///
+    /// ```
+    /// use tokio::sync::oneshot;
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let (tx, mut rx) = oneshot::channel::<()>();
+    ///
+    ///     // A receiver is not immediately finished when the sender is dropped.
+    ///     drop(tx);
+    ///     assert!(!rx.is_finished());
+    ///
+    ///     // A receiver *is* finished after it has been polled and yielded an error.
+    ///     let _ = (&mut rx).await.unwrap_err();
+    ///     assert!(rx.is_finished());
+    /// }
+    /// ```
+    pub fn is_finished(&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
@@ -1203,18 +1264,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

@@ -360,3 +360,59 @@ fn receiver_is_empty_rx_close() {
     rx.close();
     assert!(rx.is_empty());
 }
+
+#[test]
+fn receiver_is_finished_send() {
+    let (tx, mut rx) = oneshot::channel::<i32>();
+
+    assert!(
+        !rx.is_finished(),
+        "channel is NOT finished before value is sent"
+    );
+    tx.send(17).unwrap();
+    assert!(
+        !rx.is_finished(),
+        "channel is NOT finished 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_finished(), "channel IS finished after value is read");
+}
+
+#[test]
+fn receiver_is_finished_drop() {
+    let (tx, mut rx) = oneshot::channel::<i32>();
+
+    assert!(
+        !rx.is_finished(),
+        "channel is NOT finished before sender is dropped"
+    );
+    drop(tx);
+    assert!(
+        !rx.is_finished(),
+        "channel is NOT finished 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_finished(), "channel IS finished after value is read");
+}
+
+#[test]
+fn receiver_is_finished_rx_close() {
+    let (_tx, mut rx) = oneshot::channel::<i32>();
+    assert!(!rx.is_finished(), "channel is NOT finished before closing");
+    rx.close();
+    assert!(!rx.is_finished(), "channel is NOT finished 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_finished(), "channel IS finished after value is read");
+}