aboutsummaryrefslogtreecommitdiff
path: root/src/future
diff options
context:
space:
mode:
authorJoel Galenson <jgalenson@google.com>2020-10-05 08:16:15 -0700
committerJoel Galenson <jgalenson@google.com>2020-10-05 08:16:15 -0700
commitf03b3ba785a6d336884bfc525046906f8c2a9904 (patch)
tree14e2bd707d8d152ea0476ec9e686deb2a2f55b34 /src/future
parent40b8b369b069afb314a9d4bb92be1bdd038979f8 (diff)
downloadtokio-f03b3ba785a6d336884bfc525046906f8c2a9904.tar.gz
Import tokio-0.2.22
Test: None Change-Id: Iea7ee5e62819c9b16dbfad05a6146775df72506a
Diffstat (limited to 'src/future')
-rw-r--r--src/future/maybe_done.rs76
-rw-r--r--src/future/mod.rs15
-rw-r--r--src/future/pending.rs44
-rw-r--r--src/future/poll_fn.rs38
-rw-r--r--src/future/ready.rs27
-rw-r--r--src/future/try_join.rs82
6 files changed, 282 insertions, 0 deletions
diff --git a/src/future/maybe_done.rs b/src/future/maybe_done.rs
new file mode 100644
index 0000000..1e083ad
--- /dev/null
+++ b/src/future/maybe_done.rs
@@ -0,0 +1,76 @@
+//! Definition of the MaybeDone combinator
+
+use std::future::Future;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A future that may have completed.
+#[derive(Debug)]
+pub enum MaybeDone<Fut: Future> {
+ /// A not-yet-completed future
+ Future(Fut),
+ /// The output of the completed future
+ Done(Fut::Output),
+ /// The empty variant after the result of a [`MaybeDone`] has been
+ /// taken using the [`take_output`](MaybeDone::take_output) method.
+ Gone,
+}
+
+// Safe because we never generate `Pin<&mut Fut::Output>`
+impl<Fut: Future + Unpin> Unpin for MaybeDone<Fut> {}
+
+/// Wraps a future into a `MaybeDone`
+pub fn maybe_done<Fut: Future>(future: Fut) -> MaybeDone<Fut> {
+ MaybeDone::Future(future)
+}
+
+impl<Fut: Future> MaybeDone<Fut> {
+ /// Returns an [`Option`] containing a mutable reference to the output of the future.
+ /// The output of this method will be [`Some`] if and only if the inner
+ /// future has been completed and [`take_output`](MaybeDone::take_output)
+ /// has not yet been called.
+ pub fn output_mut(self: Pin<&mut Self>) -> Option<&mut Fut::Output> {
+ unsafe {
+ let this = self.get_unchecked_mut();
+ match this {
+ MaybeDone::Done(res) => Some(res),
+ _ => None,
+ }
+ }
+ }
+
+ /// Attempts to take the output of a `MaybeDone` without driving it
+ /// towards completion.
+ #[inline]
+ pub fn take_output(self: Pin<&mut Self>) -> Option<Fut::Output> {
+ unsafe {
+ let this = self.get_unchecked_mut();
+ match this {
+ MaybeDone::Done(_) => {}
+ MaybeDone::Future(_) | MaybeDone::Gone => return None,
+ };
+ if let MaybeDone::Done(output) = mem::replace(this, MaybeDone::Gone) {
+ Some(output)
+ } else {
+ unreachable!()
+ }
+ }
+ }
+}
+
+impl<Fut: Future> Future for MaybeDone<Fut> {
+ type Output = ();
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let res = unsafe {
+ match self.as_mut().get_unchecked_mut() {
+ MaybeDone::Future(a) => ready!(Pin::new_unchecked(a).poll(cx)),
+ MaybeDone::Done(_) => return Poll::Ready(()),
+ MaybeDone::Gone => panic!("MaybeDone polled after value taken"),
+ }
+ };
+ self.set(MaybeDone::Done(res));
+ Poll::Ready(())
+ }
+}
diff --git a/src/future/mod.rs b/src/future/mod.rs
new file mode 100644
index 0000000..770753f
--- /dev/null
+++ b/src/future/mod.rs
@@ -0,0 +1,15 @@
+#![allow(unused_imports, dead_code)]
+
+//! Asynchronous values.
+
+mod maybe_done;
+pub use maybe_done::{maybe_done, MaybeDone};
+
+mod poll_fn;
+pub use poll_fn::poll_fn;
+
+mod ready;
+pub(crate) use ready::{ok, Ready};
+
+mod try_join;
+pub(crate) use try_join::try_join3;
diff --git a/src/future/pending.rs b/src/future/pending.rs
new file mode 100644
index 0000000..287e836
--- /dev/null
+++ b/src/future/pending.rs
@@ -0,0 +1,44 @@
+use sdt::pin::Pin;
+use std::future::Future;
+use std::marker;
+use std::task::{Context, Poll};
+
+/// Future for the [`pending()`] function.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+struct Pending<T> {
+ _data: marker::PhantomData<T>,
+}
+
+/// Creates a future which never resolves, representing a computation that never
+/// finishes.
+///
+/// The returned future will forever return [`Poll::Pending`].
+///
+/// # Examples
+///
+/// ```no_run
+/// use tokio::future;
+///
+/// #[tokio::main]
+/// async fn main {
+/// future::pending().await;
+/// unreachable!();
+/// }
+/// ```
+pub async fn pending() -> ! {
+ Pending {
+ _data: marker::PhantomData,
+ }
+ .await
+}
+
+impl<T> Future for Pending<T> {
+ type Output = !;
+
+ fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<T> {
+ Poll::Pending
+ }
+}
+
+impl<T> Unpin for Pending<T> {}
diff --git a/src/future/poll_fn.rs b/src/future/poll_fn.rs
new file mode 100644
index 0000000..9b3d137
--- /dev/null
+++ b/src/future/poll_fn.rs
@@ -0,0 +1,38 @@
+//! Definition of the `PollFn` adapter combinator
+
+use std::fmt;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Future for the [`poll_fn`] function.
+pub struct PollFn<F> {
+ f: F,
+}
+
+impl<F> Unpin for PollFn<F> {}
+
+/// Creates a new future wrapping around a function returning [`Poll`].
+pub fn poll_fn<T, F>(f: F) -> PollFn<F>
+where
+ F: FnMut(&mut Context<'_>) -> Poll<T>,
+{
+ PollFn { f }
+}
+
+impl<F> fmt::Debug for PollFn<F> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("PollFn").finish()
+ }
+}
+
+impl<T, F> Future for PollFn<F>
+where
+ F: FnMut(&mut Context<'_>) -> Poll<T>,
+{
+ type Output = T;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
+ (&mut self.f)(cx)
+ }
+}
diff --git a/src/future/ready.rs b/src/future/ready.rs
new file mode 100644
index 0000000..de2d60c
--- /dev/null
+++ b/src/future/ready.rs
@@ -0,0 +1,27 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Future for the [`ok`](ok()) function.
+///
+/// `pub` in order to use the future as an associated type in a sealed trait.
+#[derive(Debug)]
+// Used as an associated type in a "sealed" trait.
+#[allow(unreachable_pub)]
+pub struct Ready<T>(Option<T>);
+
+impl<T> Unpin for Ready<T> {}
+
+impl<T> Future for Ready<T> {
+ type Output = T;
+
+ #[inline]
+ fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<T> {
+ Poll::Ready(self.0.take().unwrap())
+ }
+}
+
+/// Creates a future that is immediately ready with a success value.
+pub(crate) fn ok<T, E>(t: T) -> Ready<Result<T, E>> {
+ Ready(Some(Ok(t)))
+}
diff --git a/src/future/try_join.rs b/src/future/try_join.rs
new file mode 100644
index 0000000..5bd80dc
--- /dev/null
+++ b/src/future/try_join.rs
@@ -0,0 +1,82 @@
+use crate::future::{maybe_done, MaybeDone};
+
+use pin_project_lite::pin_project;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pub(crate) fn try_join3<T1, F1, T2, F2, T3, F3, E>(
+ future1: F1,
+ future2: F2,
+ future3: F3,
+) -> TryJoin3<F1, F2, F3>
+where
+ F1: Future<Output = Result<T1, E>>,
+ F2: Future<Output = Result<T2, E>>,
+ F3: Future<Output = Result<T3, E>>,
+{
+ TryJoin3 {
+ future1: maybe_done(future1),
+ future2: maybe_done(future2),
+ future3: maybe_done(future3),
+ }
+}
+
+pin_project! {
+ pub(crate) struct TryJoin3<F1, F2, F3>
+ where
+ F1: Future,
+ F2: Future,
+ F3: Future,
+ {
+ #[pin]
+ future1: MaybeDone<F1>,
+ #[pin]
+ future2: MaybeDone<F2>,
+ #[pin]
+ future3: MaybeDone<F3>,
+ }
+}
+
+impl<T1, F1, T2, F2, T3, F3, E> Future for TryJoin3<F1, F2, F3>
+where
+ F1: Future<Output = Result<T1, E>>,
+ F2: Future<Output = Result<T2, E>>,
+ F3: Future<Output = Result<T3, E>>,
+{
+ type Output = Result<(T1, T2, T3), E>;
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let mut all_done = true;
+
+ let mut me = self.project();
+
+ if me.future1.as_mut().poll(cx).is_pending() {
+ all_done = false;
+ } else if me.future1.as_mut().output_mut().unwrap().is_err() {
+ return Poll::Ready(Err(me.future1.take_output().unwrap().err().unwrap()));
+ }
+
+ if me.future2.as_mut().poll(cx).is_pending() {
+ all_done = false;
+ } else if me.future2.as_mut().output_mut().unwrap().is_err() {
+ return Poll::Ready(Err(me.future2.take_output().unwrap().err().unwrap()));
+ }
+
+ if me.future3.as_mut().poll(cx).is_pending() {
+ all_done = false;
+ } else if me.future3.as_mut().output_mut().unwrap().is_err() {
+ return Poll::Ready(Err(me.future3.take_output().unwrap().err().unwrap()));
+ }
+
+ if all_done {
+ Poll::Ready(Ok((
+ me.future1.take_output().unwrap().ok().unwrap(),
+ me.future2.take_output().unwrap().ok().unwrap(),
+ me.future3.take_output().unwrap().ok().unwrap(),
+ )))
+ } else {
+ Poll::Pending
+ }
+ }
+}