Get rid of Chain as the building block for other combinators and remove some unsafe code

Author: Diggsey

futures-util/Cargo.toml

@@ -45,6 +45,7 @@ memchr = { version = "2.2", optional = true }
 futures_01 = { version = "0.1.25", optional = true, package = "futures" }
 tokio-io = { version = "0.1.9", optional = true }
 pin-utils = "0.1.0-alpha.4"
+pin-project = "0.4.8"
 
 [dev-dependencies]
 futures = { path = "../futures", version = "0.3.4", features = ["async-await", "thread-pool"] }

futures-util/src/future/future/chain.rs

@@ -1,56 +1,84 @@
-use super::chain::Chain;
-use core::fmt;
+use core::fmt::{self, Debug};
 use core::pin::Pin;
 use futures_core::future::{FusedFuture, Future};
 use futures_core::task::{Context, Poll};
-use pin_utils::unsafe_pinned;
+use pin_project::pin_project;
 
-/// Future for the [`flatten`](super::FutureExt::flatten) method.
-#[must_use = "futures do nothing unless you `.await` or poll them"]
-pub struct Flatten<Fut>
+#[pin_project]
+#[derive(Debug)]
+enum InternalFlatten<Fut: Future> {
+    First(#[pin] Fut),
+    Second(#[pin] Fut::Output),
+    Empty,
+}
+
+impl<Fut: Future> InternalFlatten<Fut> {
+    fn new(future: Fut) -> Self {
+        Self::First(future)
+    }
+}
+
+impl<Fut> FusedFuture for InternalFlatten<Fut>
     where Fut: Future,
+          Fut::Output: Future,
 {
-    state: Chain<Fut, Fut::Output, ()>,
+    fn is_terminated(&self) -> bool {
+        match self {
+            Self::Empty => true,
+            _ => false,
+        }
+    }
 }
 
-impl<Fut> Flatten<Fut>
+impl<Fut> Future for InternalFlatten<Fut>
     where Fut: Future,
           Fut::Output: Future,
 {
-    unsafe_pinned!(state: Chain<Fut, Fut::Output, ()>);
+    type Output = <Fut::Output as Future>::Output;
 
-    pub(super) fn new(future: Fut) -> Flatten<Fut> {
-        Flatten {
-            state: Chain::new(future, ()),
-        }
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Poll::Ready(loop {
+            match self.as_mut().project() {
+                __InternalFlattenProjection::First(f) => {
+                    let f = ready!(f.poll(cx));
+                    self.set(Self::Second(f));
+                },
+                __InternalFlattenProjection::Second(f) => {
+                    let output = ready!(f.poll(cx));
+                    self.set(Self::Empty);
+                    break output;
+                },
+                __InternalFlattenProjection::Empty => unreachable!()
+            }
+        })
     }
 }
 
-impl<Fut> fmt::Debug for Flatten<Fut>
-    where Fut: Future + fmt::Debug,
-          Fut::Output: fmt::Debug,
-{
+/// Future for the [`flatten`](super::FutureExt::flatten) method.
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+#[pin_project]
+pub struct Flatten<Fut: Future>(#[pin] InternalFlatten<Fut>);
+
+impl<Fut: Debug + Future> Debug for Flatten<Fut> where Fut::Output: Debug {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("Flatten")
-            .field("state", &self.state)
-            .finish()
+        self.0.fmt(f)
     }
 }
 
-impl<Fut> FusedFuture for Flatten<Fut>
-    where Fut: Future,
-          Fut::Output: Future,
-{
-    fn is_terminated(&self) -> bool { self.state.is_terminated() }
+impl<Fut: Future> Flatten<Fut> {
+    pub(super) fn new(future: Fut) -> Self {
+        Self(InternalFlatten::new(future))
+    }
 }
 
-impl<Fut> Future for Flatten<Fut>
-    where Fut: Future,
-          Fut::Output: Future,
-{
+impl<Fut: Future> FusedFuture for Flatten<Fut> where Fut::Output: Future {
+    fn is_terminated(&self) -> bool { self.0.is_terminated() }
+}
+
+impl<Fut: Future> Future for Flatten<Fut> where Fut::Output: Future {
     type Output = <Fut::Output as Future>::Output;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        self.state().poll(cx, |a, ()| a)
+        self.project().0.poll(cx)
     }
 }

futures-util/src/future/future/flatten.rs

@@ -73,11 +73,6 @@ mod shared;
 #[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411
 pub use self::shared::Shared;
 
-// Implementation details
-
-mod chain;
-pub(crate) use self::chain::Chain;
-
 impl<T: ?Sized> FutureExt for T where T: Future {}
 
 /// An extension trait for `Future`s that provides a variety of convenient
@@ -137,13 +132,13 @@ pub trait FutureExt: Future {
     /// assert_eq!(future_of_4.await, 4);
     /// # });
     /// ```
-    fn then<Fut, F>(self, f: F) -> Then<Self, Fut, F>
+    fn then<Fut, F>(self, f: F) -> Then<Self, F>
     where
         F: FnOnce(Self::Output) -> Fut,
         Fut: Future,
         Self: Sized,
     {
-        assert_future::<Fut::Output, _>(Then::new(self, f))
+        assert_future::<Fut::Output, _>(Flatten::new(Map::new(self, f)))
     }
 
     /// Wrap this future in an `Either` future, making it the left-hand variant

futures-util/src/future/future/mod.rs

@@ -1,46 +1,4 @@
-use super::Chain;
-use core::pin::Pin;
-use futures_core::future::{FusedFuture, Future};
-use futures_core::task::{Context, Poll};
-use pin_utils::unsafe_pinned;
+use super::{Map, Flatten};
 
 /// Future for the [`then`](super::FutureExt::then) method.
-#[derive(Debug)]
-#[must_use = "futures do nothing unless you `.await` or poll them"]
-pub struct Then<Fut1, Fut2, F> {
-    chain: Chain<Fut1, Fut2, F>,
-}
-
-impl<Fut1, Fut2, F> Then<Fut1, Fut2, F>
-    where Fut1: Future,
-          Fut2: Future,
-{
-    unsafe_pinned!(chain: Chain<Fut1, Fut2, F>);
-
-    /// Creates a new `Then`.
-    pub(super) fn new(future: Fut1, f: F) -> Then<Fut1, Fut2, F> {
-        Then {
-            chain: Chain::new(future, f),
-        }
-    }
-}
-
-impl<Fut1, Fut2, F> FusedFuture for Then<Fut1, Fut2, F>
-    where Fut1: Future,
-          Fut2: Future,
-          F: FnOnce(Fut1::Output) -> Fut2,
-{
-    fn is_terminated(&self) -> bool { self.chain.is_terminated() }
-}
-
-impl<Fut1, Fut2, F> Future for Then<Fut1, Fut2, F>
-    where Fut1: Future,
-          Fut2: Future,
-          F: FnOnce(Fut1::Output) -> Fut2,
-{
-    type Output = Fut2::Output;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Fut2::Output> {
-        self.as_mut().chain().poll(cx, |output, f| f(output))
-    }
-}
+pub type Then<Fut1, F> = Flatten<Map<Fut1, F>>;