Repository: rust-lang-nursery/futures-rs
Pull request #1778 (2019-08-05 18:58:30)

document AtomicWaker memory ordering (#942)
This is certainly no proof, but I tried to help with the part "someone gets pencil and paper out" :)

I don't think not documenting the way we think it should work helps anyone, so at least this documents the expected outcome, whether it is actually true or not.

stbuehler: So I came up with 3 interesting scenarios, and the first two fail when I remove the explicit wakes.

I think for `A` could try some spinning before yielding, and for `B` we could simply leave the registration in place, but I guess for now I just document why the wakes are needed.

### A. Spurious wake preventing registration

```
wake();                             // spurious wake
    |           pop() -> Pending;
    #           register();         // state locked by wake, registration fails
    |           pop() -> Pending;   // doesn't see push(2)
    |   push(2);
    ?   wake();                     // either registration still locked or empty
```

### B. Spurious wake triggering undo of registration

```
                pop() -> Pending;
                register();
wake();             #               // spurious wake
                    |               // makes unlock in register fail, having it unregister the waker
                pop() -> Pending;   // doesn't see push(2)
        push(2);
        wake();                     // no waker to call
```

### C. Multiple wakes during registration

```
                pop() -> Pending;
                register();
wake();             #               // spurious wake
                    |               // makes unlock in register fail, having it unregister the waker
        push(2);    |
        wake();     |               // registration locked, but syncs on unlock (swap(AcqRel))
                    |
                pop() -> 2;
```


Patch:
diff --git a/.rustfmt.toml b/.rustfmt.toml
deleted file mode 100644
index f25bd518..00000000
--- a/.rustfmt.toml
+++ /dev/null
@@ -1,2 +0,0 @@
-max_width = 80
-comment_width = 80
diff --git a/.travis.yml b/.travis.yml
index 93debacc..bb049a06 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -39,7 +39,7 @@ matrix:
     # This is the minimum Rust version supported by `async-await` feature.
     # When updating this, the reminder to update the minimum required version of `async-await` feature in README.md.
     - name: cargo +nightly build (minimum required version)
-      rust: nightly-2019-07-29
+      rust: nightly-2019-08-21
       script:
         - cargo run --manifest-path ci/remove-dev-dependencies/Cargo.toml */Cargo.toml
         - cargo build --all --all-features
@@ -75,7 +75,7 @@ matrix:
       rust: nightly
       script:
         - cargo bench --all
-        - cargo bench --manifest-path futures-util/Cargo.toml --features=bench
+        - cargo bench --manifest-path futures-util/Cargo.toml --features=bilock,unstable
 
     - name: cargo +stable build --no-default-features
       rust: stable
@@ -108,11 +108,15 @@ matrix:
         - cargo build --manifest-path futures/Cargo.toml
             --target thumbv6m-none-eabi
             --no-default-features
-            --features nightly,cfg-target-has-atomic
+            --features unstable,cfg-target-has-atomic
         - cargo build --manifest-path futures/Cargo.toml
             --target thumbv6m-none-eabi
             --no-default-features
-            --features nightly,alloc,cfg-target-has-atomic
+            --features unstable,cfg-target-has-atomic,alloc
+        - cargo build --manifest-path futures/Cargo.toml
+            --target thumbv6m-none-eabi
+            --no-default-features
+            --features unstable,cfg-target-has-atomic,async-await
 
     - name: cargo build --target=thumbv7m-none-eabi
       rust: nightly
@@ -126,30 +130,52 @@ matrix:
         - cargo build --manifest-path futures/Cargo.toml
             --target thumbv7m-none-eabi
             --no-default-features
-            --features nightly,alloc
+            --features alloc
+        - cargo build --manifest-path futures/Cargo.toml
+            --target thumbv7m-none-eabi
+            --no-default-features
+            --features async-await
 
-    - name: cargo check (futures-util)
+    - name: cargo check (sub crates)
       rust: nightly
       script:
         - cargo run --manifest-path ci/remove-dev-dependencies/Cargo.toml */Cargo.toml
 
+        # futures-io
+        # Check default-features, all-features
+        - cargo check --manifest-path futures-io/Cargo.toml
+        - cargo check --manifest-path futures-io/Cargo.toml --all-features
+        # Check each features
+        - cargo check --manifest-path futures-io/Cargo.toml --features read_initializer,unstable
+
+        # futures-util
+        # Check default-features, all-features
         - cargo check --manifest-path futures-util/Cargo.toml
         - cargo check --manifest-path futures-util/Cargo.toml --all-features
-
+        # Check each features
         - cargo check --manifest-path futures-util/Cargo.toml --features sink
         - cargo check --manifest-path futures-util/Cargo.toml --features io
         - cargo check --manifest-path futures-util/Cargo.toml --features channel
-        - cargo check --manifest-path futures-util/Cargo.toml --features nightly,async-await
-        - cargo check --manifest-path futures-util/Cargo.toml --features nightly,join-macro
-        - cargo check --manifest-path futures-util/Cargo.toml --features nightly,select-macro
+        - cargo check --manifest-path futures-util/Cargo.toml --features async-await
+        - cargo check --manifest-path futures-util/Cargo.toml --features join-macro
+        - cargo check --manifest-path futures-util/Cargo.toml --features select-macro
         - cargo check --manifest-path futures-util/Cargo.toml --features compat
         - cargo check --manifest-path futures-util/Cargo.toml --features io-compat
         - cargo check --manifest-path futures-util/Cargo.toml --features sink,compat
         - cargo check --manifest-path futures-util/Cargo.toml --features sink,channel
+        - cargo check --manifest-path futures-util/Cargo.toml --features bilock,unstable
+        - cargo check --manifest-path futures-util/Cargo.toml --features sink,bilock,unstable
+        - cargo check --manifest-path futures-util/Cargo.toml --features io,bilock,unstable
+        - cargo check --manifest-path futures-util/Cargo.toml --features sink,io
+        - cargo check --manifest-path futures-util/Cargo.toml --features read_initializer,unstable
 
         - cargo check --manifest-path futures-util/Cargo.toml --no-default-features
         - cargo check --manifest-path futures-util/Cargo.toml --no-default-features --features sink
         - cargo check --manifest-path futures-util/Cargo.toml --no-default-features --features alloc,sink
+        - cargo check --manifest-path futures-util/Cargo.toml --no-default-features --features async-await
+        - cargo check --manifest-path futures-util/Cargo.toml --no-default-features --features bilock,unstable
+        - cargo check --manifest-path futures-util/Cargo.toml --no-default-features --features sink,bilock,unstable
+        - cargo check --manifest-path futures-util/Cargo.toml --no-default-features --features io,bilock,unstable
 
     - name: cargo doc
       rust: nightly
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ebeaeb0a..6a04db2f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,4 @@
-# 0.3.0-alpha.18 - 2019-9-9
+# 0.3.0-alpha.18 - 2019-8-9
 * Rewrote `join!` and `try_join!` as procedural macros to allow passing expressions (#1783)
 * Banned manual implementation of `TryFuture` and `TryStream` for forward compatibility. See #1776 for more details. (#1777)
 * Changed `AsyncReadExt::read_to_end` to return the total number of bytes read (#1721)
diff --git a/README.md b/README.md
index 3562866a..50496782 100644
--- a/README.md
+++ b/README.md
@@ -58,16 +58,14 @@ futures-preview = { version = "=0.3.0-alpha.18", default-features = false }
 
 ### Feature `async-await`
 
-The `async-await` feature provides several convenient features using unstable
-async/await. Note that this is an unstable feature, and upstream changes might
-prevent it from compiling. To use futures-rs with async/await, use:
+The `async-await` feature provides several convenient features using async/await. To use futures-rs with async/await, use:
 
 ```toml
 [dependencies]
-futures-preview = { version = "=0.3.0-alpha.18", features = ["async-await", "nightly"] }
+futures-preview = { version = "=0.3.0-alpha.18", features = ["async-await"] }
 ```
 
-The current `async-await` feature requires Rust nightly 2019-07-29 or later.
+The current `async-await` feature requires Rust nightly 2019-08-21 or later.
 
 # License
 
diff --git a/futures-channel/Cargo.toml b/futures-channel/Cargo.toml
index f8111a4c..1f3b3cb0 100644
--- a/futures-channel/Cargo.toml
+++ b/futures-channel/Cargo.toml
@@ -19,7 +19,11 @@ default = ["std"]
 std = ["alloc", "futures-core-preview/std"]
 alloc = ["futures-core-preview/alloc"]
 sink = ["futures-sink-preview"]
-nightly = ["futures-core-preview/nightly"]
+
+# Unstable features
+# These features are outside of the normal semver guarantees and require the
+# `unstable` feature as an explicit opt-in to unstable API.
+unstable = ["futures-core-preview/unstable"]
 cfg-target-has-atomic = ["futures-core-preview/cfg-target-has-atomic"]
 
 [dependencies]
diff --git a/futures-channel/src/lib.rs b/futures-channel/src/lib.rs
index c20637e8..d421f15b 100644
--- a/futures-channel/src/lib.rs
+++ b/futures-channel/src/lib.rs
@@ -19,8 +19,8 @@
 
 #![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_channel")]
 
-#[cfg(all(feature = "cfg-target-has-atomic", not(feature = "nightly")))]
-compile_error!("The `cfg-target-has-atomic` feature requires the `nightly` feature as an explicit opt-in to unstable features");
+#[cfg(all(feature = "cfg-target-has-atomic", not(feature = "unstable")))]
+compile_error!("The `cfg-target-has-atomic` feature requires the `unstable` feature as an explicit opt-in to unstable features");
 
 macro_rules! cfg_target_has_atomic {
     ($($item:item)*) => {$(
diff --git a/futures-channel/src/mpsc/mod.rs b/futures-channel/src/mpsc/mod.rs
index 6409ebc7..fde96357 100644
--- a/futures-channel/src/mpsc/mod.rs
+++ b/futures-channel/src/mpsc/mod.rs
@@ -81,8 +81,6 @@
 use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::{Context, Poll, Waker};
 use futures_core::task::__internal::AtomicWaker;
-use std::any::Any;
-use std::error::Error;
 use std::fmt;
 use std::pin::Pin;
 use std::sync::{Arc, Mutex};
@@ -166,7 +164,7 @@ enum SendErrorKind {
 
 /// The error type returned from [`try_next`](Receiver::try_next).
 pub struct TryRecvError {
-    _inner: (),
+    _priv: (),
 }
 
 impl fmt::Display for SendError {
@@ -179,7 +177,7 @@ impl fmt::Display for SendError {
     }
 }
 
-impl Error for SendError {}
+impl std::error::Error for SendError {}
 
 impl SendError {
     /// Returns true if this error is a result of the channel being full.
@@ -217,7 +215,7 @@ impl<T> fmt::Display for TrySendError<T> {
     }
 }
 
-impl<T: Any> Error for TrySendError<T> {}
+impl<T: core::any::Any> std::error::Error for TrySendError<T> {}
 
 impl<T> TrySendError<T> {
     /// Returns true if this error is a result of the channel being full.
@@ -254,7 +252,7 @@ impl fmt::Display for TryRecvError {
     }
 }
 
-impl Error for TryRecvError {}
+impl std::error::Error for TryRecvError {}
 
 #[derive(Debug)]
 struct Inner<T> {
@@ -834,7 +832,7 @@ impl<T> Receiver<T> {
             Poll::Ready(msg) => {
                 Ok(msg)
             },
-            Poll::Pending => Err(TryRecvError { _inner: () }),
+            Poll::Pending => Err(TryRecvError { _priv: () }),
         }
     }
 
diff --git a/futures-core/Cargo.toml b/futures-core/Cargo.toml
index a1a54981..a810628c 100644
--- a/futures-core/Cargo.toml
+++ b/futures-core/Cargo.toml
@@ -17,10 +17,14 @@ name = "futures_core"
 [features]
 default = ["std"]
 std = ["alloc"]
-nightly = []
-cfg-target-has-atomic = []
 alloc = []
 
+# Unstable features
+# These features are outside of the normal semver guarantees and require the
+# `unstable` feature as an explicit opt-in to unstable API.
+unstable = []
+cfg-target-has-atomic = []
+
 [dependencies]
 
 [dev-dependencies]
diff --git a/futures-core/src/future/mod.rs b/futures-core/src/future/mod.rs
index c6273119..ba9398ee 100644
--- a/futures-core/src/future/mod.rs
+++ b/futures-core/src/future/mod.rs
@@ -57,7 +57,7 @@ mod private_try_future {
 
 /// A convenience for futures that return `Result` values that includes
 /// a variety of adapters tailored to such futures.
-pub trait TryFuture: private_try_future::Sealed {
+pub trait TryFuture: Future + private_try_future::Sealed {
     /// The type of successful values yielded by this future
     type Ok;
 
diff --git a/futures-core/src/lib.rs b/futures-core/src/lib.rs
index 2121db2e..a7d5299f 100644
--- a/futures-core/src/lib.rs
+++ b/futures-core/src/lib.rs
@@ -13,8 +13,8 @@
 
 #![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_core")]
 
-#[cfg(all(feature = "cfg-target-has-atomic", not(feature = "nightly")))]
-compile_error!("The `cfg-target-has-atomic` feature requires the `nightly` feature as an explicit opt-in to unstable features");
+#[cfg(all(feature = "cfg-target-has-atomic", not(feature = "unstable")))]
+compile_error!("The `cfg-target-has-atomic` feature requires the `unstable` feature as an explicit opt-in to unstable features");
 
 #[cfg(feature = "alloc")]
 extern crate alloc;
@@ -29,9 +29,6 @@ pub mod stream;
 pub mod task;
 #[doc(hidden)] pub use self::task::Poll;
 
-pub mod never;
-#[doc(hidden)] pub use self::never::Never;
-
 #[doc(hidden)]
 pub mod core_reexport {
     #[doc(hidden)]
diff --git a/futures-core/src/stream.rs b/futures-core/src/stream.rs
index 12bb01f3..37c60c9c 100644
--- a/futures-core/src/stream.rs
+++ b/futures-core/src/stream.rs
@@ -59,6 +59,37 @@ pub trait Stream {
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Self::Item>>;
+
+    /// Returns the bounds on the remaining length of the stream.
+    ///
+    /// Specifically, `size_hint()` returns a tuple where the first element
+    /// is the lower bound, and the second element is the upper bound.
+    ///
+    /// The second half of the tuple that is returned is an [`Option`]`<`[`usize`]`>`.
+    /// A [`None`] here means that either there is no known upper bound, or the
+    /// upper bound is larger than [`usize`].
+    ///
+    /// # Implementation notes
+    ///
+    /// It is not enforced that a stream implementation yields the declared
+    /// number of elements. A buggy stream may yield less than the lower bound
+    /// or more than the upper bound of elements.
+    ///
+    /// `size_hint()` is primarily intended to be used for optimizations such as
+    /// reserving space for the elements of the stream, but must not be
+    /// trusted to e.g., omit bounds checks in unsafe code. An incorrect
+    /// implementation of `size_hint()` should not lead to memory safety
+    /// violations.
+    ///
+    /// That said, the implementation should provide a correct estimation,
+    /// because otherwise it would be a violation of the trait's protocol.
+    ///
+    /// The default implementation returns `(0, `[`None`]`)` which is correct for any
+    /// stream.
+    #[inline]
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (0, None)
+    }
 }
 
 impl<S: ?Sized + Stream + Unpin> Stream for &mut S {
@@ -70,6 +101,10 @@ impl<S: ?Sized + Stream + Unpin> Stream for &mut S {
     ) -> Poll<Option<Self::Item>> {
         S::poll_next(Pin::new(&mut **self), cx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (**self).size_hint()
+    }
 }
 
 impl<P> Stream for Pin<P>
@@ -85,6 +120,10 @@ where
     ) -> Poll<Option<Self::Item>> {
         self.get_mut().as_mut().poll_next(cx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (**self).size_hint()
+    }
 }
 
 /// A stream which tracks whether or not the underlying stream
@@ -126,7 +165,7 @@ mod private_try_stream {
 
 /// A convenience for streams that return `Result` values that includes
 /// a variety of adapters tailored to such futures.
-pub trait TryStream: private_try_stream::Sealed {
+pub trait TryStream: Stream + private_try_stream::Sealed {
     /// The type of successful values yielded by this future
     type Ok;
 
@@ -169,10 +208,30 @@ mod if_alloc {
         ) -> Poll<Option<Self::Item>> {
             Pin::new(&mut **self).poll_next(cx)
         }
+
+        fn size_hint(&self) -> (usize, Option<usize>) {
+            (**self).size_hint()
+        }
+    }
+
+    impl<T: Unpin> Stream for alloc::collections::VecDeque<T> {
+        type Item = T;
+
+        fn poll_next(
+            mut self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+        ) -> Poll<Option<Self::Item>> {
+            Poll::Ready(self.pop_front())
+        }
+
+        fn size_hint(&self) -> (usize, Option<usize>) {
+            let len = self.len();
+            (len, Some(len))
+        }
     }
 
     #[cfg(feature = "std")]
-    impl<S: Stream> Stream for ::std::panic::AssertUnwindSafe<S> {
+    impl<S: Stream> Stream for std::panic::AssertUnwindSafe<S> {
         type Item = S::Item;
 
         fn poll_next(
@@ -181,16 +240,9 @@ mod if_alloc {
         ) -> Poll<Option<S::Item>> {
             unsafe { self.map_unchecked_mut(|x| &mut x.0) }.poll_next(cx)
         }
-    }
-
-    impl<T: Unpin> Stream for ::alloc::collections::VecDeque<T> {
-        type Item = T;
 
-        fn poll_next(
-            mut self: Pin<&mut Self>,
-            _cx: &mut Context<'_>,
-        ) -> Poll<Option<Self::Item>> {
-            Poll::Ready(self.pop_front())
+        fn size_hint(&self) -> (usize, Option<usize>) {
+            self.0.size_hint()
         }
     }
 
diff --git a/futures-core/src/task/__internal/atomic_waker.rs b/futures-core/src/task/__internal/atomic_waker.rs
index 582bdf5e..89195833 100644
--- a/futures-core/src/task/__internal/atomic_waker.rs
+++ b/futures-core/src/task/__internal/atomic_waker.rs
@@ -21,9 +21,71 @@ use crate::task::Waker;
 /// A single `AtomicWaker` may be reused for any number of calls to `register` or
 /// `wake`.
 ///
-/// `AtomicWaker` does not provide any memory ordering guarantees, as such the
-/// user should use caution and use other synchronization primitives to guard
-/// the result of the underlying computation.
+/// # Memory ordering
+///
+/// Calling `register` "acquires" all memory "released" by calls to `wake`
+/// before the call to `register`.  Later calls to `wake` will wake the
+/// registered waker (on contention this wake might be triggered in `register`).
+///
+/// For concurrent calls to `register` (should be avoided) the ordering is only
+/// guaranteed for the winning call.
+///
+/// # Examples
+///
+/// Here is a simple example providing a `Flag` that can be signalled manually
+/// when it is ready.
+///
+/// ```
+/// use futures::future::Future;
+/// use futures::task::{Context, Poll, AtomicWaker};
+/// use std::sync::Arc;
+/// use std::sync::atomic::AtomicBool;
+/// use std::sync::atomic::Ordering::Relaxed;
+/// use std::pin::Pin;
+///
+/// struct Inner {
+///     waker: AtomicWaker,
+///     set: AtomicBool,
+/// }
+///
+/// #[derive(Clone)]
+/// struct Flag(Arc<Inner>);
+///
+/// impl Flag {
+///     pub fn new() -> Self {
+///         Flag(Arc::new(Inner {
+///             waker: AtomicWaker::new(),
+///             set: AtomicBool::new(false),
+///         }))
+///     }
+///
+///     pub fn signal(&self) {
+///         self.0.set.store(true, Relaxed);
+///         self.0.waker.wake();
+///     }
+/// }
+///
+/// impl Future for Flag {
+///     type Output = ();
+///
+///     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+///         // quick check to avoid registration if already done.
+///         if self.0.set.load(Relaxed) {
+///             return Poll::Ready(());
+///         }
+///
+///         self.0.waker.register(cx.waker());
+///
+///         // Need to check condition **after** `register` to avoid a race
+///         // condition that would result in lost notifications.
+///         if self.0.set.load(Relaxed) {
+///             Poll::Ready(())
+///         } else {
+///             Poll::Pending
+///         }
+///     }
+/// }
+/// ```
 pub struct AtomicWaker {
     state: AtomicUsize,
     waker: UnsafeCell<Option<Waker>>,
@@ -204,9 +266,8 @@ impl AtomicWaker {
                     *self.waker.get() = Some(waker.clone());
 
                     // Release the lock. If the state transitioned to include
-                    // the `WAKING` bit, this means that a wake has been
-                    // called concurrently, so we have to remove the waker and
-                    // wake it.`
+                    // the `WAKING` bit, this means that at least one wake has
+                    // been called concurrently.
                     //
                     // Start by assuming that the state is `REGISTERING` as this
                     // is what we jut set it to.
@@ -214,9 +275,15 @@ impl AtomicWaker {
                         REGISTERING, WAITING, AcqRel, Acquire);
 
                     match res {
-                        Ok(_) => {}
+                        Ok(_) => {
+                            // memory ordering: acquired self.state
+                            // - if previous wakes went through it syncs with
+                            //   their final release (`fetch_and`)
+                            // - if there was no previous wake the next wake
+                            //   will wake us, no sync needed.
+                        }
                         Err(actual) => {
-                            // This branch can only be reached if a
+                            // This branch can only be reached if at least one
                             // concurrent thread called `wake`. In this
                             // case, `actual` **must** be `REGISTERING |
                             // `WAKING`.
@@ -229,8 +296,13 @@ impl AtomicWaker {
                             // Just swap, because no one could change state while state == `REGISTERING` | `WAKING`.
                             self.state.swap(WAITING, AcqRel);
 
-                            // The atomic swap was complete, now
-                            // wake the task and return.
+                            // memory ordering: we acquired the state for all
+                            // concurrent wakes, but future wakes might still
+                            // need to wake us in case we can't make progress
+                            // from the pending wakes.
+                            //
+                            // So we simply schedule to come back later (we could
+                            // also simply leave the registration in place above).
                             waker.wake();
                         }
                     }
@@ -239,7 +311,15 @@ impl AtomicWaker {
             WAKING => {
                 // Currently in the process of waking the task, i.e.,
                 // `wake` is currently being called on the old task handle.
-                // So, we call wake on the new waker
+                //
+                // memory ordering: we acquired the state for all
+                // concurrent wakes, but future wakes might still
+                // need to wake us in case we can't make progress
+                // from the pending wakes.
+                //
+                // So we simply schedule to come back later (we
+                // could also spin here trying to acquire the lock
+                // to register).
                 waker.wake_by_ref();
             }
             state => {
@@ -248,6 +328,9 @@ impl AtomicWaker {
                 // caller's code as racing to call `register` doesn't make much
                 // sense.
                 //
+                // memory ordering: don't care. a concurrent register() is going
+                // to succeed and provide proper memory ordering.
+                //
                 // We just want to maintain memory safety. It is ok to drop the
                 // call to `register`.
                 debug_assert!(
diff --git a/futures-core/src/task/spawn.rs b/futures-core/src/task/spawn.rs
index 6e66e067..895e81b5 100644
--- a/futures-core/src/task/spawn.rs
+++ b/futures-core/src/task/spawn.rs
@@ -55,7 +55,7 @@ pub trait LocalSpawn {
 
 /// An error that occurred during spawning.
 pub struct SpawnError {
-    _hidden: (),
+    _priv: (),
 }
 
 impl fmt::Debug for SpawnError {
@@ -78,7 +78,7 @@ impl std::error::Error for SpawnError {}
 impl SpawnError {
     /// Spawning failed because the executor has been shut down.
     pub fn shutdown() -> Self {
-        Self { _hidden: () }
+        Self { _priv: () }
     }
 
     /// Check whether spawning failed to the executor being shut down.
diff --git a/futures-executor/src/enter.rs b/futures-executor/src/enter.rs
index c9c0a451..5895a9ef 100644
--- a/futures-executor/src/enter.rs
+++ b/futures-executor/src/enter.rs
@@ -7,16 +7,29 @@ thread_local!(static ENTERED: Cell<bool> = Cell::new(false));
 ///
 /// For more details, see [`enter` documentation](enter()).
 pub struct Enter {
-    _a: ()
+    _priv: (),
 }
 
 /// An error returned by `enter` if an execution scope has already been
 /// entered.
-#[derive(Debug)]
 pub struct EnterError {
-    _a: (),
+    _priv: (),
 }
 
+impl fmt::Debug for EnterError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("EnterError").finish()
+    }
+}
+
+impl fmt::Display for EnterError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "an execution scope has already been entered")
+    }
+}
+
+impl std::error::Error for EnterError {}
+
 /// Marks the current thread as being within the dynamic extent of an
 /// executor.
 ///
@@ -42,11 +55,11 @@ pub struct EnterError {
 pub fn enter() -> Result<Enter, EnterError> {
     ENTERED.with(|c| {
         if c.get() {
-            Err(EnterError { _a: () })
+            Err(EnterError { _priv: () })
         } else {
             c.set(true);
 
-            Ok(Enter { _a: () })
+            Ok(Enter { _priv: () })
         }
     })
 }
diff --git a/futures-executor/src/local_pool.rs b/futures-executor/src/local_pool.rs
index 8bf72462..9985dca5 100644
--- a/futures-executor/src/local_pool.rs
+++ b/futures-executor/src/local_pool.rs
@@ -132,10 +132,9 @@ impl LocalPool {
     ///
     /// ```
     /// use futures::executor::LocalPool;
-    /// use futures::future::ready;
     ///
     /// let mut pool = LocalPool::new();
-    /// # let my_app  = ready(());
+    /// # let my_app  = async {};
     ///
     /// // run tasks in the pool until `my_app` completes, by default spawning
     /// // further tasks back onto the pool
@@ -330,9 +329,14 @@ impl<S: Stream + Unpin> BlockingStream<S> {
 
 impl<S: Stream + Unpin> Iterator for BlockingStream<S> {
     type Item = S::Item;
+
     fn next(&mut self) -> Option<Self::Item> {
         LocalPool::new().run_until(self.stream.next())
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
 }
 
 impl Spawn for LocalSpawner {
diff --git a/futures-executor/src/thread_pool.rs b/futures-executor/src/thread_pool.rs
index 070f6dc0..9a7ab8d1 100644
--- a/futures-executor/src/thread_pool.rs
+++ b/futures-executor/src/thread_pool.rs
@@ -4,12 +4,13 @@ use futures_core::future::{Future, FutureObj};
 use futures_core::task::{Context, Poll, Spawn, SpawnError};
 use futures_util::future::FutureExt;
 use futures_util::task::{ArcWake, waker_ref};
+use std::cmp;
+use std::fmt;
 use std::io;
-use std::sync::{Arc, Mutex};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::mpsc;
+use std::sync::{Arc, Mutex};
 use std::thread;
-use std::fmt;
 
 /// A general-purpose thread pool for scheduling tasks that poll futures to
 /// completion.
@@ -116,7 +117,6 @@ impl ThreadPool {
     /// completion.
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::executor::ThreadPool;
     ///
     /// let pool = ThreadPool::new().unwrap();
@@ -204,7 +204,7 @@ impl ThreadPoolBuilder {
     /// See the other methods on this type for details on the defaults.
     pub fn new() -> ThreadPoolBuilder {
         ThreadPoolBuilder {
-            pool_size: num_cpus::get(),
+            pool_size: cmp::max(1, num_cpus::get()),
             stack_size: 0,
             name_prefix: None,
             after_start: None,
@@ -214,7 +214,7 @@ impl ThreadPoolBuilder {
 
     /// Set size of a future ThreadPool
     ///
-    /// The size of a thread pool is the number of worker threads spawned.  By
+    /// The size of a thread pool is the number of worker threads spawned. By
     /// default, this is equal to the number of CPU cores.
     pub fn pool_size(&mut self, size: usize) -> &mut Self {
         self.pool_size = size;
diff --git a/futures-io/Cargo.toml b/futures-io/Cargo.toml
index 8b5c4a1f..6af7c24c 100644
--- a/futures-io/Cargo.toml
+++ b/futures-io/Cargo.toml
@@ -18,7 +18,10 @@ name = "futures_io"
 default = ["std"]
 std = []
 
-[dependencies]
+# Unstable features
+# These features are outside of the normal semver guarantees and require the
+# `unstable` feature as an explicit opt-in to unstable API.
+unstable = []
+read_initializer = []
 
-[dev-dependencies]
-futures-preview = { path = "../futures", version = "=0.3.0-alpha.18" }
+[dependencies]
diff --git a/futures-io/src/lib.rs b/futures-io/src/lib.rs
index 4cb3b959..551ee070 100644
--- a/futures-io/src/lib.rs
+++ b/futures-io/src/lib.rs
@@ -8,6 +8,8 @@
 //! All items of this library are only available when the `std` feature of this
 //! library is activated, and it is activated by default.
 
+#![cfg_attr(feature = "read_initializer", feature(read_initializer))]
+
 #![cfg_attr(not(feature = "std"), no_std)]
 
 #![warn(missing_docs, missing_debug_implementations, rust_2018_idioms, unreachable_pub)]
@@ -19,13 +21,15 @@
 
 #![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_io")]
 
+#[cfg(all(feature = "read_initializer", not(feature = "unstable")))]
+compile_error!("The `read_initializer` feature requires the `unstable` feature as an explicit opt-in to unstable features");
+
 #[cfg(feature = "std")]
 mod if_std {
     use std::cmp;
     use std::io;
     use std::ops::DerefMut;
     use std::pin::Pin;
-    use std::ptr;
     use std::task::{Context, Poll};
 
     // Re-export some types from `std::io` so that users don't have to deal
@@ -40,45 +44,9 @@ mod if_std {
         SeekFrom as SeekFrom,
     };
 
-    /// A type used to conditionally initialize buffers passed to `AsyncRead`
-    /// methods, modeled after `std`.
-    #[derive(Debug)]
-    pub struct Initializer(bool);
-
-    impl Initializer {
-        /// Returns a new `Initializer` which will zero out buffers.
-        #[inline]
-        pub fn zeroing() -> Initializer {
-            Initializer(true)
-        }
-
-        /// Returns a new `Initializer` which will not zero out buffers.
-        ///
-        /// # Safety
-        ///
-        /// This method may only be called by `AsyncRead`ers which guarantee
-        /// that they will not read from the buffers passed to `AsyncRead`
-        /// methods, and that the return value of the method accurately reflects
-        /// the number of bytes that have been written to the head of the buffer.
-        #[inline]
-        pub unsafe fn nop() -> Initializer {
-            Initializer(false)
-        }
-
-        /// Indicates if a buffer should be initialized.
-        #[inline]
-        pub fn should_initialize(&self) -> bool {
-            self.0
-        }
-
-        /// Initializes a buffer if necessary.
-        #[inline]
-        pub fn initialize(&self, buf: &mut [u8]) {
-            if self.should_initialize() {
-                unsafe { ptr::write_bytes(buf.as_mut_ptr(), 0, buf.len()) }
-            }
-        }
-    }
+    #[cfg(feature = "read_initializer")]
+    #[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411
+    pub use io::Initializer as Initializer;
 
     /// Read bytes asynchronously.
     ///
@@ -99,6 +67,7 @@ mod if_std {
         /// This method is `unsafe` because and `AsyncRead`er could otherwise
         /// return a non-zeroing `Initializer` from another `AsyncRead` type
         /// without an `unsafe` block.
+        #[cfg(feature = "read_initializer")]
         #[inline]
         unsafe fn initializer(&self) -> Initializer {
             Initializer::zeroing()
@@ -342,6 +311,7 @@ mod if_std {
 
     macro_rules! deref_async_read {
         () => {
+            #[cfg(feature = "read_initializer")]
             unsafe fn initializer(&self) -> Initializer {
                 (**self).initializer()
             }
@@ -373,6 +343,7 @@ mod if_std {
         P: DerefMut + Unpin,
         P::Target: AsyncRead,
     {
+        #[cfg(feature = "read_initializer")]
         unsafe fn initializer(&self) -> Initializer {
             (**self).initializer()
         }
@@ -390,12 +361,11 @@ mod if_std {
         }
     }
 
-    /// `unsafe` because the `io::Read` type must not access the buffer
-    /// before reading data into it.
-    macro_rules! unsafe_delegate_async_read_to_stdio {
+    macro_rules! delegate_async_read_to_stdio {
         () => {
+            #[cfg(feature = "read_initializer")]
             unsafe fn initializer(&self) -> Initializer {
-                Initializer::nop()
+                io::Read::initializer(self)
             }
 
             fn poll_read(mut self: Pin<&mut Self>, _: &mut Context<'_>, buf: &mut [u8])
@@ -413,19 +383,11 @@ mod if_std {
     }
 
     impl AsyncRead for &[u8] {
-        unsafe_delegate_async_read_to_stdio!();
-    }
-
-    impl AsyncRead for io::Repeat {
-        unsafe_delegate_async_read_to_stdio!();
-    }
-
-    impl AsyncRead for io::Empty {
-        unsafe_delegate_async_read_to_stdio!();
+        delegate_async_read_to_stdio!();
     }
 
     impl<T: AsRef<[u8]> + Unpin> AsyncRead for io::Cursor<T> {
-        unsafe_delegate_async_read_to_stdio!();
+        delegate_async_read_to_stdio!();
     }
 
     macro_rules! deref_async_write {
@@ -547,10 +509,6 @@ mod if_std {
         delegate_async_write_to_stdio!();
     }
 
-    impl AsyncWrite for io::Sink {
-        delegate_async_write_to_stdio!();
-    }
-
     macro_rules! deref_async_seek {
         () => {
             fn poll_seek(mut self: Pin<&mut Self>, cx: &mut Context<'_>, pos: SeekFrom)
@@ -651,10 +609,6 @@ mod if_std {
         delegate_async_buf_read_to_stdio!();
     }
 
-    impl AsyncBufRead for io::Empty {
-        delegate_async_buf_read_to_stdio!();
-    }
-
     impl<T: AsRef<[u8]> + Unpin> AsyncBufRead for io::Cursor<T> {
         delegate_async_buf_read_to_stdio!();
     }
diff --git a/futures-join-macro/Cargo.toml b/futures-join-macro/Cargo.toml
index 0c53d657..caba4692 100644
--- a/futures-join-macro/Cargo.toml
+++ b/futures-join-macro/Cargo.toml
@@ -18,7 +18,7 @@ proc-macro = true
 [features]
 
 [dependencies]
-proc-macro2 = "0.4"
-proc-macro-hack = "0.5.3"
-quote = "0.6"
-syn = { version = "0.15.25", features = ["full"] }
+proc-macro2 = "1.0"
+proc-macro-hack = "0.5.9"
+quote = "1.0"
+syn = { version = "1.0", features = ["full"] }
diff --git a/futures-join-macro/src/lib.rs b/futures-join-macro/src/lib.rs
index a17b1bd8..08783a4a 100644
--- a/futures-join-macro/src/lib.rs
+++ b/futures-join-macro/src/lib.rs
@@ -11,7 +11,7 @@ extern crate proc_macro;
 use proc_macro::TokenStream;
 use proc_macro2::{Span, TokenStream as TokenStream2};
 use proc_macro_hack::proc_macro_hack;
-use quote::quote;
+use quote::{format_ident, quote};
 use syn::parse::{Parse, ParseStream};
 use syn::{parenthesized, parse_quote, Expr, Ident, Token};
 
@@ -61,7 +61,7 @@ fn bind_futures(
         .into_iter()
         .enumerate()
         .map(|(i, expr)| {
-            let name = Ident::new(&format!("_fut{}", i), span);
+            let name = format_ident!("_fut{}", i, span = span);
             future_let_bindings.push(quote! {
                 // Move future into a local so that it is pinned in one place and
                 // is no longer accessible by the end user.
diff --git a/futures-select-macro/Cargo.toml b/futures-select-macro/Cargo.toml
index 2147b02b..78ef7c6a 100644
--- a/futures-select-macro/Cargo.toml
+++ b/futures-select-macro/Cargo.toml
@@ -19,7 +19,7 @@ proc-macro = true
 std = []
 
 [dependencies]
-proc-macro2 = "0.4"
-proc-macro-hack = "0.5.3"
-quote = "0.6"
-syn = { version = "0.15.25", features = ["full"] }
+proc-macro2 = "1.0"
+proc-macro-hack = "0.5.9"
+quote = "1.0"
+syn = { version = "1.0", features = ["full"] }
diff --git a/futures-select-macro/src/lib.rs b/futures-select-macro/src/lib.rs
index 2cc2a4c9..ffb4776f 100644
--- a/futures-select-macro/src/lib.rs
+++ b/futures-select-macro/src/lib.rs
@@ -11,7 +11,7 @@ extern crate proc_macro;
 use proc_macro::TokenStream;
 use proc_macro2::Span;
 use proc_macro_hack::proc_macro_hack;
-use quote::quote;
+use quote::{format_ident, quote};
 use syn::{parenthesized, parse_quote, Expr, Ident, Pat, Token};
 use syn::parse::{Parse, ParseStream};
 
@@ -119,7 +119,7 @@ fn declare_result_enum(
     // "_0", "_1", "_2"
     let variant_names: Vec<Ident> =
         (0..variants)
-            .map(|num| Ident::new(&format!("_{}", num), span))
+            .map(|num| format_ident!("_{}", num, span = span))
             .collect();
 
     let type_parameters = &variant_names;
@@ -149,7 +149,6 @@ pub fn select(input: TokenStream) -> TokenStream {
     let parsed = syn::parse_macro_input!(input as Select);
 
     let futures_crate: syn::Path = parsed.futures_crate_path.unwrap_or_else(|| parse_quote!(::futures_util));
-    let rand_crate: syn::Path = parse_quote!(#futures_crate::rand_reexport);
 
     // should be def_site, but that's unstable
     let span = Span::call_site();
@@ -265,10 +264,7 @@ pub fn select(input: TokenStream) -> TokenStream {
             #( #poll_functions )*
 
             let mut __select_arr = [#( #variant_names ),*];
-            <[_] as #rand_crate::SliceRandom>::shuffle(
-                &mut __select_arr,
-                &mut #rand_crate::thread_rng(),
-            );
+            #futures_crate::async_await::shuffle(&mut __select_arr);
             for poller in &mut __select_arr {
                 let poller: &mut &mut dyn FnMut(
                     &mut #futures_crate::task::Context<'_>
diff --git a/futures-sink/Cargo.toml b/futures-sink/Cargo.toml
index c3efb27a..ee751631 100644
--- a/futures-sink/Cargo.toml
+++ b/futures-sink/Cargo.toml
@@ -16,8 +16,7 @@ name = "futures_sink"
 
 [features]
 default = ["std"]
-std = ["alloc", "futures-core-preview/std"]
-alloc = ["futures-core-preview/alloc"]
+std = ["alloc"]
+alloc = []
 
 [dependencies]
-futures-core-preview = { path = "../futures-core", version = "=0.3.0-alpha.18", default-features = false }
diff --git a/futures-sink/src/lib.rs b/futures-sink/src/lib.rs
index 2939bf53..8400325d 100644
--- a/futures-sink/src/lib.rs
+++ b/futures-sink/src/lib.rs
@@ -16,9 +16,9 @@
 #[cfg(feature = "alloc")]
 extern crate alloc;
 
-use futures_core::task::{Context, Poll};
 use core::ops::DerefMut;
 use core::pin::Pin;
+use core::task::{Context, Poll};
 
 /// A `Sink` is a value into which other values can be sent, asynchronously.
 ///
@@ -84,8 +84,7 @@ pub trait Sink<Item> {
     ///
     /// In most cases, if the sink encounters an error, the sink will
     /// permanently be unable to receive items.
-    fn start_send(self: Pin<&mut Self>, item: Item)
-                  -> Result<(), Self::Error>;
+    fn start_send(self: Pin<&mut Self>, item: Item) -> Result<(), Self::Error>;
 
     /// Flush any remaining output from this sink.
     ///
@@ -162,7 +161,7 @@ where
 #[cfg(feature = "alloc")]
 mod if_alloc {
     use super::*;
-    use futures_core::never::Never;
+    use core::convert::Infallible as Never;
 
     impl<T> Sink<T> for alloc::vec::Vec<T> {
         type Error = Never;
diff --git a/futures-util/Cargo.toml b/futures-util/Cargo.toml
index ed6c6e57..1825ea83 100644
--- a/futures-util/Cargo.toml
+++ b/futures-util/Cargo.toml
@@ -18,17 +18,22 @@ name = "futures_util"
 default = ["std"]
 std = ["alloc", "futures-core-preview/std", "slab"]
 alloc = ["futures-core-preview/alloc"]
-async-await = ["std"]
+async-await = []
 compat = ["std", "futures_01"]
 io-compat = ["io", "compat", "tokio-io"]
-bench = []
-nightly = ["futures-core-preview/nightly"]
-cfg-target-has-atomic = ["futures-core-preview/cfg-target-has-atomic"]
 sink = ["futures-sink-preview"]
 io = ["std", "futures-io-preview", "memchr"]
 channel = ["std", "futures-channel-preview"]
 join-macro = ["async-await", "futures-join-macro-preview", "proc-macro-hack", "proc-macro-nested"]
-select-macro = ["async-await", "futures-select-macro-preview", "proc-macro-hack", "proc-macro-nested", "rand"]
+select-macro = ["async-await", "futures-select-macro-preview", "proc-macro-hack", "proc-macro-nested"]
+
+# Unstable features
+# These features are outside of the normal semver guarantees and require the
+# `unstable` feature as an explicit opt-in to unstable API.
+unstable = ["futures-core-preview/unstable"]
+cfg-target-has-atomic = ["futures-core-preview/cfg-target-has-atomic"]
+bilock = []
+read_initializer = ["io", "futures-io-preview/read_initializer", "futures-io-preview/unstable"]
 
 [dependencies]
 futures-core-preview = { path = "../futures-core", version = "=0.3.0-alpha.18", default-features = false }
@@ -37,9 +42,8 @@ futures-io-preview = { path = "../futures-io", version = "=0.3.0-alpha.18", defa
 futures-sink-preview = { path = "../futures-sink", version = "=0.3.0-alpha.18", default-features = false, optional = true }
 futures-join-macro-preview = { path = "../futures-join-macro", version = "=0.3.0-alpha.18", default-features = false, optional = true }
 futures-select-macro-preview = { path = "../futures-select-macro", version = "=0.3.0-alpha.18", default-features = false, optional = true }
-proc-macro-hack = { version = "0.5", optional = true }
+proc-macro-hack = { version = "0.5.9", optional = true }
 proc-macro-nested = { version = "0.1.2", optional = true }
-rand = { version = "0.7.0", optional = true }
 slab = { version = "0.4", optional = true }
 memchr = { version = "2.2", optional = true }
 futures_01 = { version = "0.1.25", optional = true, package = "futures" }
@@ -47,7 +51,7 @@ tokio-io = { version = "0.1.9", optional = true }
 pin-utils = "0.1.0-alpha.4"
 
 [dev-dependencies]
-futures-preview = { path = "../futures", version = "=0.3.0-alpha.18", features = ["async-await", "nightly"] }
+futures-preview = { path = "../futures", version = "=0.3.0-alpha.18", features = ["async-await"] }
 futures-test-preview = { path = "../futures-test", version = "=0.3.0-alpha.18" }
 tokio = "0.1.11"
 
diff --git a/futures-util/benches_disabled/bilock.rs b/futures-util/benches_disabled/bilock.rs
index 4b1a172e..78b5edb6 100644
--- a/futures-util/benches_disabled/bilock.rs
+++ b/futures-util/benches_disabled/bilock.rs
@@ -1,6 +1,6 @@
 #![feature(test)]
 
-#[cfg(feature = "bench")]
+#[cfg(feature = "bilock")]
 mod bench {
 use futures::task::{Context, Waker};
 use futures::executor::LocalPool;
diff --git a/futures-util/src/async_await/join_mod.rs b/futures-util/src/async_await/join_mod.rs
index 61cb0585..a2dbad9a 100644
--- a/futures-util/src/async_await/join_mod.rs
+++ b/futures-util/src/async_await/join_mod.rs
@@ -19,12 +19,11 @@ macro_rules! document_join_macro {
         /// # Examples
         ///
         /// ```
-        /// #![feature(async_await)]
         /// # futures::executor::block_on(async {
-        /// use futures::{join, future};
+        /// use futures::join;
         ///
-        /// let a = future::ready(1);
-        /// let b = future::ready(2);
+        /// let a = async { 1 };
+        /// let b = async { 2 };
         ///
         /// assert_eq!(join!(a, b), (1, 2));
         /// # });
@@ -47,12 +46,11 @@ macro_rules! document_join_macro {
         /// `Ok` of a tuple of the values:
         ///
         /// ```
-        /// #![feature(async_await)]
         /// # futures::executor::block_on(async {
-        /// use futures::{try_join, future};
+        /// use futures::try_join;
         ///
-        /// let a = future::ready(Ok::<i32, i32>(1));
-        /// let b = future::ready(Ok::<u64, i32>(2));
+        /// let a = async { Ok::<i32, i32>(1) };
+        /// let b = async { Ok::<u64, i32>(2) };
         ///
         /// assert_eq!(try_join!(a, b), Ok((1, 2)));
         /// # });
@@ -62,12 +60,11 @@ macro_rules! document_join_macro {
         /// that error:
         ///
         /// ```
-        /// #![feature(async_await)]
         /// # futures::executor::block_on(async {
-        /// use futures::{try_join, future};
+        /// use futures::try_join;
         ///
-        /// let a = future::ready(Ok::<i32, i32>(1));
-        /// let b = future::ready(Err::<u64, i32>(2));
+        /// let a = async { Ok::<i32, i32>(1) };
+        /// let b = async { Err::<u64, i32>(2) };
         ///
         /// assert_eq!(try_join!(a, b), Err(2));
         /// # });
diff --git a/futures-util/src/async_await/mod.rs b/futures-util/src/async_await/mod.rs
index ef5052f7..0a68dba6 100644
--- a/futures-util/src/async_await/mod.rs
+++ b/futures-util/src/async_await/mod.rs
@@ -31,6 +31,11 @@ mod select_mod;
 #[cfg(feature = "select-macro")]
 pub use self::select_mod::*;
 
+#[cfg(feature = "select-macro")]
+mod random;
+#[cfg(feature = "select-macro")]
+pub use self::random::*;
+
 #[doc(hidden)]
 #[inline(always)]
 pub fn assert_unpin<T: Unpin>(_: &T) {}
diff --git a/futures-util/src/async_await/random.rs b/futures-util/src/async_await/random.rs
new file mode 100644
index 00000000..4f8c7254
--- /dev/null
+++ b/futures-util/src/async_await/random.rs
@@ -0,0 +1,54 @@
+use std::{
+    cell::Cell,
+    collections::hash_map::DefaultHasher,
+    hash::Hasher,
+    num::Wrapping,
+    sync::atomic::{AtomicUsize, Ordering},
+};
+
+// Based on [Fisher–Yates shuffle].
+//
+// [Fisher–Yates shuffle]: https://en.wikipedia.org/wiki/Fisher–Yates_shuffle
+#[doc(hidden)]
+pub fn shuffle<T>(slice: &mut [T]) {
+    for i in (1..slice.len()).rev() {
+        slice.swap(i, gen_index(i + 1));
+    }
+}
+
+/// Return a value from `0..n`.
+fn gen_index(n: usize) -> usize {
+    (random() % n as u64) as usize
+}
+
+/// Pseudorandom number generator based on [xorshift*].
+///
+/// [xorshift*]: https://en.wikipedia.org/wiki/Xorshift#xorshift*
+fn random() -> u64 {
+    thread_local! {
+        static RNG: Cell<Wrapping<u64>> = Cell::new(Wrapping(prng_seed()));
+    }
+
+    fn prng_seed() -> u64 {
+        static COUNTER: AtomicUsize = AtomicUsize::new(0);
+
+        // Any non-zero seed will do
+        let mut seed = 0;
+        while seed == 0 {
+            let mut hasher = DefaultHasher::new();
+            hasher.write_usize(COUNTER.fetch_add(1, Ordering::Relaxed));
+            seed = hasher.finish();
+        }
+        seed
+    }
+
+    RNG.with(|rng| {
+        let mut x = rng.get();
+        debug_assert_ne!(x.0, 0);
+        x ^= x >> 12;
+        x ^= x << 25;
+        x ^= x >> 27;
+        rng.set(x);
+        x.0.wrapping_mul(0x2545_f491_4f6c_dd1d)
+    })
+}
diff --git a/futures-util/src/async_await/select_mod.rs b/futures-util/src/async_await/select_mod.rs
index 67efdca6..e6cdf1a9 100644
--- a/futures-util/src/async_await/select_mod.rs
+++ b/futures-util/src/async_await/select_mod.rs
@@ -27,7 +27,6 @@ macro_rules! document_select_macro {
         /// # Examples
         ///
         /// ```
-        /// #![feature(async_await)]
         /// # futures::executor::block_on(async {
         /// use futures::future;
         /// use futures::select;
@@ -43,7 +42,6 @@ macro_rules! document_select_macro {
         /// ```
         ///
         /// ```
-        /// #![feature(async_await)]
         /// # futures::executor::block_on(async {
         /// use futures::future;
         /// use futures::stream::{self, StreamExt};
@@ -65,7 +63,6 @@ macro_rules! document_select_macro {
         /// the case where all futures have completed.
         ///
         /// ```
-        /// #![feature(async_await)]
         /// # futures::executor::block_on(async {
         /// use futures::future;
         /// use futures::select;
diff --git a/futures-util/src/compat/compat01as03.rs b/futures-util/src/compat/compat01as03.rs
index 9790ef58..dde42a4b 100644
--- a/futures-util/src/compat/compat01as03.rs
+++ b/futures-util/src/compat/compat01as03.rs
@@ -68,7 +68,6 @@ pub trait Future01CompatExt: Future01 {
     /// [`Future<Output = Result<T, E>>`](futures_core::future::Future).
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// # // TODO: These should be all using `futures::compat`, but that runs up against Cargo
     /// # // feature issues
@@ -95,7 +94,6 @@ pub trait Stream01CompatExt: Stream01 {
     /// [`Stream<Item = Result<T, E>>`](futures_core::stream::Stream).
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::StreamExt;
     /// use futures_util::compat::Stream01CompatExt;
@@ -124,7 +122,6 @@ pub trait Sink01CompatExt: Sink01 {
     /// [`Sink<T, Error = E>`](futures_sink::Sink).
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::{sink::SinkExt, stream::StreamExt};
     /// use futures_util::compat::{Stream01CompatExt, Sink01CompatExt};
@@ -371,9 +368,9 @@ unsafe impl UnsafeNotify01 for NotifyWaker {
 #[cfg(feature = "io-compat")]
 mod io {
     use super::*;
-    use futures_io::{
-        AsyncRead as AsyncRead03, AsyncWrite as AsyncWrite03, Initializer,
-    };
+    #[cfg(feature = "read_initializer")]
+    use futures_io::Initializer;
+    use futures_io::{AsyncRead as AsyncRead03, AsyncWrite as AsyncWrite03};
     use std::io::Error;
     use tokio_io::{AsyncRead as AsyncRead01, AsyncWrite as AsyncWrite01};
 
@@ -383,7 +380,7 @@ mod io {
         /// [`AsyncRead`](futures_io::AsyncRead).
         ///
         /// ```
-        /// #![feature(async_await, impl_trait_in_bindings)]
+        /// #![feature(impl_trait_in_bindings)]
         /// # #![allow(incomplete_features)]
         /// # futures::executor::block_on(async {
         /// use futures::io::AsyncReadExt;
@@ -413,7 +410,6 @@ mod io {
         /// [`AsyncWrite`](futures_io::AsyncWrite).
         ///
         /// ```
-        /// #![feature(async_await)]
         /// # futures::executor::block_on(async {
         /// use futures::io::AsyncWriteExt;
         /// use futures_util::compat::AsyncWrite01CompatExt;
@@ -437,6 +433,7 @@ mod io {
     impl<W: AsyncWrite01> AsyncWrite01CompatExt for W {}
 
     impl<R: AsyncRead01> AsyncRead03 for Compat01As03<R> {
+        #[cfg(feature = "read_initializer")]
         unsafe fn initializer(&self) -> Initializer {
             // check if `prepare_uninitialized_buffer` needs zeroing
             if self.inner.get_ref().prepare_uninitialized_buffer(&mut [1]) {
diff --git a/futures-util/src/compat/compat03as01.rs b/futures-util/src/compat/compat03as01.rs
index a571826e..b9632ced 100644
--- a/futures-util/src/compat/compat03as01.rs
+++ b/futures-util/src/compat/compat03as01.rs
@@ -201,9 +201,9 @@ impl Current {
 
         let ptr = current_to_ptr(self);
         let vtable = &RawWakerVTable::new(clone, wake, wake, drop);
-        unsafe {
-            WakerRef::new(task03::Waker::from_raw(RawWaker::new(ptr, vtable)))
-        }
+        WakerRef::new_unowned(std::mem::ManuallyDrop::new(unsafe {
+            task03::Waker::from_raw(RawWaker::new(ptr, vtable))
+        }))
     }
 }
 
@@ -262,6 +262,7 @@ mod io {
     }
 
     impl<R: AsyncRead03 + Unpin> AsyncRead01 for Compat<R> {
+        #[cfg(feature = "read_initializer")]
         unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
             let initializer = self.inner.initializer();
             let does_init = initializer.should_initialize();
diff --git a/futures-util/src/compat/executor.rs b/futures-util/src/compat/executor.rs
index 6a0fa8e1..8ba7fb93 100644
--- a/futures-util/src/compat/executor.rs
+++ b/futures-util/src/compat/executor.rs
@@ -22,7 +22,6 @@ pub trait Executor01CompatExt: Executor01<Executor01Future> +
     /// futures 0.3 [`Spawn`](futures_core::task::Spawn).
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::task::SpawnExt;
     /// use futures::future::{FutureExt, TryFutureExt};
     /// use futures_util::compat::Executor01CompatExt;
diff --git a/futures-util/src/future/abortable.rs b/futures-util/src/future/abortable.rs
index 804e9dc1..281cf6b4 100644
--- a/futures-util/src/future/abortable.rs
+++ b/futures-util/src/future/abortable.rs
@@ -2,6 +2,7 @@ use crate::task::AtomicWaker;
 use futures_core::future::Future;
 use futures_core::task::{Context, Poll};
 use pin_utils::unsafe_pinned;
+use core::fmt;
 use core::pin::Pin;
 use core::sync::atomic::{AtomicBool, Ordering};
 use alloc::sync::Arc;
@@ -29,12 +30,11 @@ impl<Fut> Abortable<Fut> where Fut: Future {
     /// Example:
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{ready, Abortable, AbortHandle, Aborted};
+    /// use futures::future::{Abortable, AbortHandle, Aborted};
     ///
     /// let (abort_handle, abort_registration) = AbortHandle::new_pair();
-    /// let future = Abortable::new(ready(2), abort_registration);
+    /// let future = Abortable::new(async { 2 }, abort_registration);
     /// abort_handle.abort();
     /// assert_eq!(future.await, Err(Aborted));
     /// # });
@@ -70,12 +70,11 @@ impl AbortHandle {
     /// Example:
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{ready, Abortable, AbortHandle, Aborted};
+    /// use futures::future::{Abortable, AbortHandle, Aborted};
     ///
     /// let (abort_handle, abort_registration) = AbortHandle::new_pair();
-    /// let future = Abortable::new(ready(2), abort_registration);
+    /// let future = Abortable::new(async { 2 }, abort_registration);
     /// abort_handle.abort();
     /// assert_eq!(future.await, Err(Aborted));
     /// # });
@@ -126,6 +125,15 @@ pub fn abortable<Fut>(future: Fut) -> (Abortable<Fut>, AbortHandle)
 #[derive(Copy, Clone, Debug, Eq, PartialEq)]
 pub struct Aborted;
 
+impl fmt::Display for Aborted {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "`Abortable` future has been aborted")
+    }
+}
+
+#[cfg(feature = "std")]
+impl std::error::Error for Aborted {}
+
 impl<Fut> Future for Abortable<Fut> where Fut: Future {
     type Output = Result<Fut::Output, Aborted>;
 
diff --git a/futures-util/src/future/either.rs b/futures-util/src/future/either.rs
index 24ff16da..12bdce62 100644
--- a/futures-util/src/future/either.rs
+++ b/futures-util/src/future/either.rs
@@ -160,9 +160,10 @@ mod if_std {
     use super::Either;
     use core::pin::Pin;
     use core::task::{Context, Poll};
+    #[cfg(feature = "read_initializer")]
+    use futures_io::Initializer;
     use futures_io::{
-        AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, Initializer, IoSlice, IoSliceMut, Result,
-        SeekFrom,
+        AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, IoSlice, IoSliceMut, Result, SeekFrom,
     };
 
     impl<A, B> AsyncRead for Either<A, B>
@@ -170,6 +171,7 @@ mod if_std {
         A: AsyncRead,
         B: AsyncRead,
     {
+        #[cfg(feature = "read_initializer")]
         unsafe fn initializer(&self) -> Initializer {
             match self {
                 Either::Left(x) => x.initializer(),
diff --git a/futures-util/src/future/fuse.rs b/futures-util/src/future/fuse.rs
index 0c7fdc84..32454688 100644
--- a/futures-util/src/future/fuse.rs
+++ b/futures-util/src/future/fuse.rs
@@ -27,7 +27,6 @@ impl<Fut: Future> Fuse<Fut> {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::mpsc;
     /// use futures::future::{Fuse, FusedFuture, FutureExt};
diff --git a/futures-util/src/future/into_stream.rs b/futures-util/src/future/into_stream.rs
index adc987e6..9f8a3912 100644
--- a/futures-util/src/future/into_stream.rs
+++ b/futures-util/src/future/into_stream.rs
@@ -1,3 +1,4 @@
+use crate::stream::{self, Once};
 use core::pin::Pin;
 use futures_core::future::Future;
 use futures_core::stream::Stream;
@@ -8,15 +9,15 @@ use pin_utils::unsafe_pinned;
 #[must_use = "streams do nothing unless polled"]
 #[derive(Debug)]
 pub struct IntoStream<Fut: Future> {
-    future: Option<Fut>
+    inner: Once<Fut>
 }
 
 impl<Fut: Future> IntoStream<Fut> {
-    unsafe_pinned!(future: Option<Fut>);
+    unsafe_pinned!(inner: Once<Fut>);
 
     pub(super) fn new(future: Fut) -> IntoStream<Fut> {
         IntoStream {
-            future: Some(future)
+            inner: stream::once(future)
         }
     }
 }
@@ -24,13 +25,13 @@ impl<Fut: Future> IntoStream<Fut> {
 impl<Fut: Future> Stream for IntoStream<Fut> {
     type Item = Fut::Output;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let v = match self.as_mut().future().as_pin_mut() {
-            Some(fut) => ready!(fut.poll(cx)),
-            None => return Poll::Ready(None),
-        };
+    #[inline]
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.inner().poll_next(cx)
+    }
 
-        self.as_mut().future().set(None);
-        Poll::Ready(Some(v))
+    #[inline]
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.inner.size_hint()
     }
 }
diff --git a/futures-util/src/future/join.rs b/futures-util/src/future/join.rs
index 82832867..5e984b84 100644
--- a/futures-util/src/future/join.rs
+++ b/futures-util/src/future/join.rs
@@ -90,12 +90,11 @@ generate! {
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
-/// let a = future::ready(1);
-/// let b = future::ready(2);
+/// let a = async { 1 };
+/// let b = async { 2 };
 /// let pair = future::join(a, b);
 ///
 /// assert_eq!(pair.await, (1, 2));
@@ -115,13 +114,12 @@ where
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
-/// let a = future::ready(1);
-/// let b = future::ready(2);
-/// let c = future::ready(3);
+/// let a = async { 1 };
+/// let b = async { 2 };
+/// let c = async { 3 };
 /// let tuple = future::join3(a, b, c);
 ///
 /// assert_eq!(tuple.await, (1, 2, 3));
@@ -145,14 +143,13 @@ where
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
-/// let a = future::ready(1);
-/// let b = future::ready(2);
-/// let c = future::ready(3);
-/// let d = future::ready(4);
+/// let a = async { 1 };
+/// let b = async { 2 };
+/// let c = async { 3 };
+/// let d = async { 4 };
 /// let tuple = future::join4(a, b, c, d);
 ///
 /// assert_eq!(tuple.await, (1, 2, 3, 4));
@@ -178,15 +175,14 @@ where
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
-/// let a = future::ready(1);
-/// let b = future::ready(2);
-/// let c = future::ready(3);
-/// let d = future::ready(4);
-/// let e = future::ready(5);
+/// let a = async { 1 };
+/// let b = async { 2 };
+/// let c = async { 3 };
+/// let d = async { 4 };
+/// let e = async { 5 };
 /// let tuple = future::join5(a, b, c, d, e);
 ///
 /// assert_eq!(tuple.await, (1, 2, 3, 4, 5));
diff --git a/futures-util/src/future/join_all.rs b/futures-util/src/future/join_all.rs
index 9a6d82dc..07408856 100644
--- a/futures-util/src/future/join_all.rs
+++ b/futures-util/src/future/join_all.rs
@@ -102,9 +102,8 @@ where
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
-/// use futures::future::{join_all};
+/// use futures::future::join_all;
 ///
 /// async fn foo(i: u32) -> u32 { i }
 ///
diff --git a/futures-util/src/future/lazy.rs b/futures-util/src/future/lazy.rs
index 4fc0504a..5e72218d 100644
--- a/futures-util/src/future/lazy.rs
+++ b/futures-util/src/future/lazy.rs
@@ -19,7 +19,6 @@ impl<F> Unpin for Lazy<F> {}
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
diff --git a/futures-util/src/future/maybe_done.rs b/futures-util/src/future/maybe_done.rs
index d359191b..f16f8897 100644
--- a/futures-util/src/future/maybe_done.rs
+++ b/futures-util/src/future/maybe_done.rs
@@ -27,12 +27,11 @@ impl<Fut: Future + Unpin> Unpin for MaybeDone<Fut> {}
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 /// use futures::pin_mut;
 ///
-/// let future = future::maybe_done(future::ready(5));
+/// let future = future::maybe_done(async { 5 });
 /// pin_mut!(future);
 /// assert_eq!(future.as_mut().take_output(), None);
 /// let () = future.as_mut().await;
diff --git a/futures-util/src/future/mod.rs b/futures-util/src/future/mod.rs
index f591a647..4cd04a57 100644
--- a/futures-util/src/future/mod.rs
+++ b/futures-util/src/future/mod.rs
@@ -4,17 +4,14 @@
 //! including the `FutureExt` trait which adds methods to `Future` types.
 
 use core::pin::Pin;
-use futures_core::future::Future;
 use futures_core::stream::Stream;
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "alloc")]
 use alloc::boxed::Box;
-#[cfg(feature = "alloc")]
-use futures_core::future::{BoxFuture, LocalBoxFuture};
 
-// re-export for `select!`
-#[doc(hidden)]
-pub use futures_core::future::FusedFuture;
+pub use futures_core::future::{FusedFuture, Future};
+#[cfg(feature = "alloc")]
+pub use futures_core::future::{BoxFuture, LocalBoxFuture};
 
 // Primitive futures
 mod lazy;
@@ -129,11 +126,10 @@ pub trait FutureExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, FutureExt};
+    /// use futures::future::FutureExt;
     ///
-    /// let future = future::ready(1);
+    /// let future = async { 1 };
     /// let new_future = future.map(|x| x + 3);
     /// assert_eq!(new_future.await, 4);
     /// # });
@@ -161,12 +157,11 @@ pub trait FutureExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, FutureExt};
+    /// use futures::future::FutureExt;
     ///
-    /// let future_of_1 = future::ready(1);
-    /// let future_of_4 = future_of_1.then(|x| future::ready(x + 3));
+    /// let future_of_1 = async { 1 };
+    /// let future_of_4 = future_of_1.then(|x| async move { x + 3 });
     /// assert_eq!(future_of_4.await, 4);
     /// # });
     /// ```
@@ -187,15 +182,14 @@ pub trait FutureExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, FutureExt};
+    /// use futures::future::FutureExt;
     ///
     /// let x = 6;
     /// let future = if x < 10 {
-    ///     future::ready(true).left_future()
+    ///     async { true }.left_future()
     /// } else {
-    ///     future::ready(false).right_future()
+    ///     async { false }.right_future()
     /// };
     ///
     /// assert_eq!(future.await, true);
@@ -217,15 +211,14 @@ pub trait FutureExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, FutureExt};
+    /// use futures::future::FutureExt;
     ///
     /// let x = 6;
     /// let future = if x > 10 {
-    ///     future::ready(true).left_future()
+    ///     async { true }.left_future()
     /// } else {
-    ///     future::ready(false).right_future()
+    ///     async { false }.right_future()
     /// };
     ///
     /// assert_eq!(future.await, false);
@@ -246,12 +239,11 @@ pub trait FutureExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, FutureExt};
+    /// use futures::future::FutureExt;
     /// use futures::stream::StreamExt;
     ///
-    /// let future = future::ready(17);
+    /// let future = async { 17 };
     /// let stream = future.into_stream();
     /// let collected: Vec<_> = stream.collect().await;
     /// assert_eq!(collected, vec![17]);
@@ -280,11 +272,10 @@ pub trait FutureExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, FutureExt};
+    /// use futures::future::FutureExt;
     ///
-    /// let nested_future = future::ready(future::ready(1));
+    /// let nested_future = async { async { 1 } };
     /// let future = nested_future.flatten();
     /// assert_eq!(future.await, 1);
     /// # });
@@ -310,13 +301,12 @@ pub trait FutureExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, FutureExt};
+    /// use futures::future::FutureExt;
     /// use futures::stream::{self, StreamExt};
     ///
     /// let stream_items = vec![17, 18, 19];
-    /// let future_of_a_stream = future::ready(stream::iter(stream_items));
+    /// let future_of_a_stream = async { stream::iter(stream_items) };
     ///
     /// let stream = future_of_a_stream.flatten_stream();
     /// let list: Vec<_> = stream.collect().await;
@@ -363,11 +353,10 @@ pub trait FutureExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, FutureExt};
+    /// use futures::future::FutureExt;
     ///
-    /// let future = future::ready(1);
+    /// let future = async { 1 };
     /// let new_future = future.inspect(|&x| println!("about to resolve: {}", x));
     /// assert_eq!(new_future.await, 1);
     /// # });
@@ -398,7 +387,6 @@ pub trait FutureExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::future::{self, FutureExt, Ready};
     ///
@@ -431,11 +419,10 @@ pub trait FutureExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, FutureExt};
+    /// use futures::future::FutureExt;
     ///
-    /// let future = future::ready(6);
+    /// let future = async { 6 };
     /// let shared1 = future.shared();
     /// let shared2 = shared1.clone();
     ///
@@ -449,13 +436,12 @@ pub trait FutureExt: Future {
     /// // synchronous function to better illustrate the cross-thread aspect of
     /// // the `shared` combinator.
     ///
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, FutureExt};
+    /// use futures::future::FutureExt;
     /// use futures::executor::block_on;
     /// use std::thread;
     ///
-    /// let future = future::ready(6);
+    /// let future = async { 6 };
     /// let shared1 = future.shared();
     /// let shared2 = shared1.clone();
     /// let join_handle = thread::spawn(move || {
diff --git a/futures-util/src/future/never_error.rs b/futures-util/src/future/never_error.rs
index 6e8094c1..5a68e6f9 100644
--- a/futures-util/src/future/never_error.rs
+++ b/futures-util/src/future/never_error.rs
@@ -1,7 +1,7 @@
+use crate::never::Never;
 use core::pin::Pin;
 use futures_core::future::{FusedFuture, Future};
 use futures_core::task::{self, Poll};
-use futures_core::never::Never;
 use pin_utils::unsafe_pinned;
 
 /// Future for the [`never_error`](super::FutureExt::never_error) combinator.
diff --git a/futures-util/src/future/option.rs b/futures-util/src/future/option.rs
index 53e9a1e9..e14745e9 100644
--- a/futures-util/src/future/option.rs
+++ b/futures-util/src/future/option.rs
@@ -12,11 +12,10 @@ use pin_utils::unsafe_pinned;
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
-/// use futures::future::{self, OptionFuture};
+/// use futures::future::OptionFuture;
 ///
-/// let mut a: OptionFuture<_> = Some(future::ready(123)).into();
+/// let mut a: OptionFuture<_> = Some(async { 123 }).into();
 /// assert_eq!(a.await, Some(123));
 ///
 /// a = None.into();
diff --git a/futures-util/src/future/pending.rs b/futures-util/src/future/pending.rs
index 9c623f7b..299add72 100644
--- a/futures-util/src/future/pending.rs
+++ b/futures-util/src/future/pending.rs
@@ -24,7 +24,6 @@ impl<T> FusedFuture for Pending<T> {
 /// # Examples
 ///
 /// ```ignore
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
diff --git a/futures-util/src/future/poll_fn.rs b/futures-util/src/future/poll_fn.rs
index d926cf9f..b7b10be8 100644
--- a/futures-util/src/future/poll_fn.rs
+++ b/futures-util/src/future/poll_fn.rs
@@ -20,7 +20,6 @@ impl<F> Unpin for PollFn<F> {}
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future::poll_fn;
 /// use futures::task::{Context, Poll};
diff --git a/futures-util/src/future/ready.rs b/futures-util/src/future/ready.rs
index 33828ca0..94e6d355 100644
--- a/futures-util/src/future/ready.rs
+++ b/futures-util/src/future/ready.rs
@@ -29,7 +29,6 @@ impl<T> Future for Ready<T> {
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
@@ -46,7 +45,6 @@ pub fn ready<T>(t: T) -> Ready<T> {
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
@@ -63,7 +61,6 @@ pub fn ok<T, E>(t: T) -> Ready<Result<T, E>> {
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
diff --git a/futures-util/src/io/allow_std.rs b/futures-util/src/io/allow_std.rs
index 63cda335..f44d5b25 100644
--- a/futures-util/src/io/allow_std.rs
+++ b/futures-util/src/io/allow_std.rs
@@ -1,4 +1,6 @@
 use futures_core::task::{Context, Poll};
+#[cfg(feature = "read_initializer")]
+use futures_io::Initializer;
 use futures_io::{AsyncRead, AsyncWrite, AsyncSeek, AsyncBufRead, IoSlice, IoSliceMut, SeekFrom};
 use std::{fmt, io};
 use std::pin::Pin;
@@ -106,8 +108,10 @@ impl<T> io::Read for AllowStdIo<T> where T: io::Read {
     fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> io::Result<usize> {
         self.0.read_vectored(bufs)
     }
-    // TODO: implement the `initializer` fn when it stabilizes.
-    // See rust-lang/rust #42788
+    #[cfg(feature = "read_initializer")]
+    unsafe fn initializer(&self) -> Initializer {
+        self.0.initializer()
+    }
     fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
         self.0.read_to_end(buf)
     }
@@ -131,6 +135,11 @@ impl<T> AsyncRead for AllowStdIo<T> where T: io::Read {
     {
         Poll::Ready(Ok(try_with_interrupt!(self.0.read_vectored(bufs))))
     }
+
+    #[cfg(feature = "read_initializer")]
+    unsafe fn initializer(&self) -> Initializer {
+        self.0.initializer()
+    }
 }
 
 impl<T> io::Seek for AllowStdIo<T> where T: io::Seek {
diff --git a/futures-util/src/io/buf_reader.rs b/futures-util/src/io/buf_reader.rs
index 6f7a026c..e6df6abe 100644
--- a/futures-util/src/io/buf_reader.rs
+++ b/futures-util/src/io/buf_reader.rs
@@ -1,5 +1,7 @@
 use futures_core::task::{Context, Poll};
-use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, Initializer, IoSliceMut, SeekFrom};
+#[cfg(feature = "read_initializer")]
+use futures_io::Initializer;
+use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, IoSliceMut, SeekFrom};
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 use std::io::{self, Read};
 use std::pin::Pin;
@@ -48,7 +50,7 @@ impl<R: AsyncRead> BufReader<R> {
         unsafe {
             let mut buffer = Vec::with_capacity(capacity);
             buffer.set_len(capacity);
-            inner.initializer().initialize(&mut buffer);
+            super::initialize(&inner, &mut buffer);
             Self {
                 inner,
                 buf: buffer.into_boxed_slice(),
@@ -166,6 +168,7 @@ impl<R: AsyncRead> AsyncRead for BufReader<R> {
     }
 
     // we can't skip unconditionally because of the large buffer case in read.
+    #[cfg(feature = "read_initializer")]
     unsafe fn initializer(&self) -> Initializer {
         self.inner.initializer()
     }
diff --git a/futures-util/src/io/chain.rs b/futures-util/src/io/chain.rs
new file mode 100644
index 00000000..ec6bbff7
--- /dev/null
+++ b/futures-util/src/io/chain.rs
@@ -0,0 +1,165 @@
+use futures_core::task::{Context, Poll};
+#[cfg(feature = "read_initializer")]
+use futures_io::Initializer;
+use futures_io::{AsyncBufRead, AsyncRead, IoSliceMut};
+use pin_utils::{unsafe_pinned, unsafe_unpinned};
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+
+/// Reader for the [`chain`](super::AsyncReadExt::chain) method.
+#[must_use = "readers do nothing unless polled"]
+pub struct Chain<T, U> {
+    first: T,
+    second: U,
+    done_first: bool,
+}
+
+impl<T, U> Unpin for Chain<T, U>
+where
+    T: Unpin,
+    U: Unpin,
+{
+}
+
+impl<T, U> Chain<T, U>
+where
+    T: AsyncRead,
+    U: AsyncRead,
+{
+    unsafe_pinned!(first: T);
+    unsafe_pinned!(second: U);
+    unsafe_unpinned!(done_first: bool);
+
+    pub(super) fn new(first: T, second: U) -> Self {
+        Self {
+            first,
+            second,
+            done_first: false,
+        }
+    }
+
+    /// Gets references to the underlying readers in this `Chain`.
+    pub fn get_ref(&self) -> (&T, &U) {
+        (&self.first, &self.second)
+    }
+
+    /// Gets mutable references to the underlying readers in this `Chain`.
+    ///
+    /// Care should be taken to avoid modifying the internal I/O state of the
+    /// underlying readers as doing so may corrupt the internal state of this
+    /// `Chain`.
+    pub fn get_mut(&mut self) -> (&mut T, &mut U) {
+        (&mut self.first, &mut self.second)
+    }
+
+    /// Gets pinned mutable references to the underlying readers in this `Chain`.
+    ///
+    /// Care should be taken to avoid modifying the internal I/O state of the
+    /// underlying readers as doing so may corrupt the internal state of this
+    /// `Chain`.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> (Pin<&mut T>, Pin<&mut U>) {
+        unsafe {
+            let Self { first, second, .. } = self.get_unchecked_mut();
+            (Pin::new_unchecked(first), Pin::new_unchecked(second))
+        }
+    }
+
+    /// Consumes the `Chain`, returning the wrapped readers.
+    pub fn into_inner(self) -> (T, U) {
+        (self.first, self.second)
+    }
+}
+
+impl<T, U> fmt::Debug for Chain<T, U>
+where
+    T: fmt::Debug,
+    U: fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Chain")
+            .field("t", &self.first)
+            .field("u", &self.second)
+            .finish()
+    }
+}
+
+impl<T, U> AsyncRead for Chain<T, U>
+where
+    T: AsyncRead,
+    U: AsyncRead,
+{
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        if !self.done_first {
+            match ready!(self.as_mut().first().poll_read(cx, buf)?) {
+                0 if !buf.is_empty() => *self.as_mut().done_first() = true,
+                n => return Poll::Ready(Ok(n)),
+            }
+        }
+        self.second().poll_read(cx, buf)
+    }
+
+    fn poll_read_vectored(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &mut [IoSliceMut<'_>],
+    ) -> Poll<io::Result<usize>> {
+        if !self.done_first {
+            let n = ready!(self.as_mut().first().poll_read_vectored(cx, bufs)?);
+            if n == 0 && bufs.iter().any(|b| !b.is_empty()) {
+                *self.as_mut().done_first() = true
+            } else {
+                return Poll::Ready(Ok(n));
+            }
+        }
+        self.second().poll_read_vectored(cx, bufs)
+    }
+
+    #[cfg(feature = "read_initializer")]
+    unsafe fn initializer(&self) -> Initializer {
+        let initializer = self.first.initializer();
+        if initializer.should_initialize() {
+            initializer
+        } else {
+            self.second.initializer()
+        }
+    }
+}
+
+impl<T, U> AsyncBufRead for Chain<T, U>
+where
+    T: AsyncBufRead,
+    U: AsyncBufRead,
+{
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        let Self {
+            first,
+            second,
+            done_first,
+        } = unsafe { self.get_unchecked_mut() };
+        let first = unsafe { Pin::new_unchecked(first) };
+        let second = unsafe { Pin::new_unchecked(second) };
+
+        if !*done_first {
+            match ready!(first.poll_fill_buf(cx)?) {
+                buf if buf.is_empty() => {
+                    *done_first = true;
+                }
+                buf => return Poll::Ready(Ok(buf)),
+            }
+        }
+        second.poll_fill_buf(cx)
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        if !self.done_first {
+            self.first().consume(amt)
+        } else {
+            self.second().consume(amt)
+        }
+    }
+}
diff --git a/futures-util/src/io/empty.rs b/futures-util/src/io/empty.rs
new file mode 100644
index 00000000..60fa08a9
--- /dev/null
+++ b/futures-util/src/io/empty.rs
@@ -0,0 +1,67 @@
+use futures_core::task::{Context, Poll};
+#[cfg(feature = "read_initializer")]
+use futures_io::Initializer;
+use futures_io::{AsyncBufRead, AsyncRead};
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+
+/// Reader for the [`empty()`] function.
+#[must_use = "readers do nothing unless polled"]
+pub struct Empty {
+    _priv: (),
+}
+
+/// Constructs a new handle to an empty reader.
+///
+/// All reads from the returned reader will return `Poll::Ready(Ok(0))`.
+///
+/// # Examples
+///
+/// A slightly sad example of not reading anything into a buffer:
+///
+/// ```
+/// # futures::executor::block_on(async {
+/// use futures::io::{self, AsyncReadExt};
+///
+/// let mut buffer = String::new();
+/// let mut reader = io::empty();
+/// reader.read_to_string(&mut buffer).await?;
+/// assert!(buffer.is_empty());
+/// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+/// ```
+pub fn empty() -> Empty {
+    Empty { _priv: () }
+}
+
+impl AsyncRead for Empty {
+    #[inline]
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        _: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(Ok(0))
+    }
+
+    #[cfg(feature = "read_initializer")]
+    #[inline]
+    unsafe fn initializer(&self) -> Initializer {
+        Initializer::nop()
+    }
+}
+
+impl AsyncBufRead for Empty {
+    #[inline]
+    fn poll_fill_buf(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        Poll::Ready(Ok(&[]))
+    }
+    #[inline]
+    fn consume(self: Pin<&mut Self>, _: usize) {}
+}
+
+impl fmt::Debug for Empty {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.pad("Empty { .. }")
+    }
+}
diff --git a/futures-util/src/io/mod.rs b/futures-util/src/io/mod.rs
index 7f7e0295..db8bd134 100644
--- a/futures-util/src/io/mod.rs
+++ b/futures-util/src/io/mod.rs
@@ -9,16 +9,35 @@
 //! This module is only available when the `io` and `std` features of this
 //! library is activated, and it is activated by default.
 
+#[cfg(feature = "io-compat")]
+use crate::compat::Compat;
+use std::ptr;
+
 pub use futures_io::{
-    AsyncRead, AsyncWrite, AsyncSeek, AsyncBufRead, IoSlice, IoSliceMut, SeekFrom,
+    AsyncRead, AsyncWrite, AsyncSeek, AsyncBufRead, Error, ErrorKind,
+    IoSlice, IoSliceMut, Result, SeekFrom,
 };
-
-#[cfg(feature = "io-compat")] use crate::compat::Compat;
+#[cfg(feature = "read_initializer")]
+pub use futures_io::Initializer;
 
 // used by `BufReader` and `BufWriter`
 // https://github.com/rust-lang/rust/blob/master/src/libstd/sys_common/io.rs#L1
 const DEFAULT_BUF_SIZE: usize = 8 * 1024;
 
+/// Initializes a buffer if necessary.
+///
+/// A buffer is always initialized if `read_initializer` feature is disabled.
+#[inline]
+unsafe fn initialize<R: AsyncRead>(_reader: &R, buf: &mut [u8]) {
+    #[cfg(feature = "read_initializer")]
+    {
+        if !_reader.initializer().should_initialize() {
+            return;
+        }
+    }
+    ptr::write_bytes(buf.as_mut_ptr(), 0, buf.len())
+}
+
 mod allow_std;
 pub use self::allow_std::AllowStdIo;
 
@@ -28,12 +47,21 @@ pub use self::buf_reader::BufReader;
 mod buf_writer;
 pub use self::buf_writer::BufWriter;
 
+mod chain;
+pub use self::chain::Chain;
+
+mod close;
+pub use self::close::Close;
+
 mod copy_into;
 pub use self::copy_into::CopyInto;
 
 mod copy_buf_into;
 pub use self::copy_buf_into::CopyBufInto;
 
+mod empty;
+pub use self::empty::{empty, Empty};
+
 mod flush;
 pub use self::flush::Flush;
 
@@ -66,15 +94,21 @@ pub use self::read_to_string::ReadToString;
 mod read_until;
 pub use self::read_until::ReadUntil;
 
-mod close;
-pub use self::close::Close;
+mod repeat;
+pub use self::repeat::{repeat, Repeat};
 
 mod seek;
 pub use self::seek::Seek;
 
+mod sink;
+pub use self::sink::{sink, Sink};
+
 mod split;
 pub use self::split::{ReadHalf, WriteHalf};
 
+mod take;
+pub use self::take::Take;
+
 mod window;
 pub use self::window::Window;
 
@@ -89,6 +123,38 @@ pub use self::write_all::WriteAll;
 
 /// An extension trait which adds utility methods to `AsyncRead` types.
 pub trait AsyncReadExt: AsyncRead {
+    /// Creates an adaptor which will chain this stream with another.
+    ///
+    /// The returned `AsyncRead` instance will first read all bytes from this object
+    /// until EOF is encountered. Afterwards the output is equivalent to the
+    /// output of `next`.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::AsyncReadExt;
+    /// use std::io::Cursor;
+    ///
+    /// let reader1 = Cursor::new([1, 2, 3, 4]);
+    /// let reader2 = Cursor::new([5, 6, 7, 8]);
+    ///
+    /// let mut reader = reader1.chain(reader2);
+    /// let mut buffer = Vec::new();
+    ///
+    /// // read the value into a Vec.
+    /// reader.read_to_end(&mut buffer).await?;
+    /// assert_eq!(buffer, [1, 2, 3, 4, 5, 6, 7, 8]);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn chain<R>(self, next: R) -> Chain<Self, R>
+    where
+        Self: Sized,
+        R: AsyncRead,
+    {
+        Chain::new(self, next)
+    }
+
     /// Creates a future which copies all the bytes from one object to another.
     ///
     /// The returned future will copy all the bytes read from this `AsyncRead` into the
@@ -101,7 +167,6 @@ pub trait AsyncReadExt: AsyncRead {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::{AsyncReadExt, AsyncWriteExt};
     /// use std::io::Cursor;
@@ -133,7 +198,6 @@ pub trait AsyncReadExt: AsyncRead {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::AsyncReadExt;
     /// use std::io::Cursor;
@@ -178,7 +242,6 @@ pub trait AsyncReadExt: AsyncRead {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::AsyncReadExt;
     /// use std::io::Cursor;
@@ -195,7 +258,6 @@ pub trait AsyncReadExt: AsyncRead {
     /// ## EOF is hit before `buf` is filled
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::AsyncReadExt;
     /// use std::io::{self, Cursor};
@@ -224,7 +286,6 @@ pub trait AsyncReadExt: AsyncRead {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::AsyncReadExt;
     /// use std::io::Cursor;
@@ -254,7 +315,6 @@ pub trait AsyncReadExt: AsyncRead {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::AsyncReadExt;
     /// use std::io::Cursor;
@@ -285,7 +345,6 @@ pub trait AsyncReadExt: AsyncRead {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::AsyncReadExt;
     /// use std::io::Cursor;
@@ -314,6 +373,32 @@ pub trait AsyncReadExt: AsyncRead {
         split::split(self)
     }
 
+    /// Creates an AsyncRead adapter which will read at most `limit` bytes
+    /// from the underlying reader.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::AsyncReadExt;
+    /// use std::io::Cursor;
+    ///
+    /// let reader = Cursor::new(&b"12345678"[..]);
+    /// let mut buffer = [0; 5];
+    ///
+    /// let mut take = reader.take(4);
+    /// let n = take.read(&mut buffer).await?;
+    ///
+    /// assert_eq!(n, 4);
+    /// assert_eq!(&buffer, b"1234\0");
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    fn take(self, limit: u64) -> Take<Self>
+        where Self: Sized
+    {
+        Take::new(self, limit)
+    }
+
     /// Wraps an [`AsyncRead`] in a compatibility wrapper that allows it to be
     /// used as a futures 0.1 / tokio-io 0.1 `AsyncRead`. If the wrapped type
     /// implements [`AsyncWrite`] as well, the result will also implement the
@@ -337,7 +422,6 @@ pub trait AsyncWriteExt: AsyncWrite {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::{AllowStdIo, AsyncWriteExt};
     /// use std::io::{BufWriter, Cursor};
@@ -399,7 +483,6 @@ pub trait AsyncWriteExt: AsyncWrite {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::AsyncWriteExt;
     /// use std::io::Cursor;
@@ -439,7 +522,6 @@ pub trait AsyncWriteExt: AsyncWrite {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::AsyncWriteExt;
     /// use futures::stream::{self, StreamExt};
@@ -495,7 +577,6 @@ pub trait AsyncBufReadExt: AsyncBufRead {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::{AsyncBufReadExt, AsyncWriteExt};
     /// use std::io::Cursor;
@@ -535,7 +616,6 @@ pub trait AsyncBufReadExt: AsyncBufRead {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::AsyncBufReadExt;
     /// use std::io::Cursor;
@@ -598,7 +678,6 @@ pub trait AsyncBufReadExt: AsyncBufRead {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::AsyncBufReadExt;
     /// use std::io::Cursor;
@@ -649,7 +728,6 @@ pub trait AsyncBufReadExt: AsyncBufRead {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::io::AsyncBufReadExt;
     /// use futures::stream::StreamExt;
diff --git a/futures-util/src/io/read_to_end.rs b/futures-util/src/io/read_to_end.rs
index 4e342e05..c4dab99e 100644
--- a/futures-util/src/io/read_to_end.rs
+++ b/futures-util/src/io/read_to_end.rs
@@ -58,7 +58,7 @@ pub(super) fn read_to_end_internal<R: AsyncRead + ?Sized>(
                 g.buf.reserve(32);
                 let capacity = g.buf.capacity();
                 g.buf.set_len(capacity);
-                rd.initializer().initialize(&mut g.buf[g.len..]);
+                super::initialize(&rd, &mut g.buf[g.len..]);
             }
         }
 
diff --git a/futures-util/src/io/repeat.rs b/futures-util/src/io/repeat.rs
new file mode 100644
index 00000000..9f493d1a
--- /dev/null
+++ b/futures-util/src/io/repeat.rs
@@ -0,0 +1,73 @@
+use futures_core::task::{Context, Poll};
+#[cfg(feature = "read_initializer")]
+use futures_io::Initializer;
+use futures_io::{AsyncRead, IoSliceMut};
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+
+/// Reader for the [`repeat()`] function.
+#[must_use = "readers do nothing unless polled"]
+pub struct Repeat {
+    byte: u8,
+}
+
+/// Creates an instance of a reader that infinitely repeats one byte.
+///
+/// All reads from this reader will succeed by filling the specified buffer with
+/// the given byte.
+///
+/// # Examples
+///
+/// ```
+/// # futures::executor::block_on(async {
+/// use futures::io::{self, AsyncReadExt};
+///
+/// let mut buffer = [0; 3];
+/// let mut reader = io::repeat(0b101);
+/// reader.read_exact(&mut buffer).await.unwrap();
+/// assert_eq!(buffer, [0b101, 0b101, 0b101]);
+/// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+/// ```
+pub fn repeat(byte: u8) -> Repeat {
+    Repeat { byte }
+}
+
+impl AsyncRead for Repeat {
+    #[inline]
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        for slot in &mut *buf {
+            *slot = self.byte;
+        }
+        Poll::Ready(Ok(buf.len()))
+    }
+
+    #[inline]
+    fn poll_read_vectored(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &mut [IoSliceMut<'_>],
+    ) -> Poll<io::Result<usize>> {
+        let mut nwritten = 0;
+        for buf in bufs {
+            nwritten += ready!(self.as_mut().poll_read(cx, buf))?;
+        }
+        Poll::Ready(Ok(nwritten))
+    }
+
+    #[cfg(feature = "read_initializer")]
+    #[inline]
+    unsafe fn initializer(&self) -> Initializer {
+        Initializer::nop()
+    }
+}
+
+impl fmt::Debug for Repeat {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.pad("Repeat { .. }")
+    }
+}
diff --git a/futures-util/src/io/sink.rs b/futures-util/src/io/sink.rs
new file mode 100644
index 00000000..4a32ca70
--- /dev/null
+++ b/futures-util/src/io/sink.rs
@@ -0,0 +1,67 @@
+use futures_core::task::{Context, Poll};
+use futures_io::{AsyncWrite, IoSlice};
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+
+/// Writer for the [`sink()`] function.
+#[must_use = "writers do nothing unless polled"]
+pub struct Sink {
+    _priv: (),
+}
+
+/// Creates an instance of a writer which will successfully consume all data.
+///
+/// All calls to `poll_write` on the returned instance will return `Poll::Ready(Ok(buf.len()))`
+/// and the contents of the buffer will not be inspected.
+///
+/// # Examples
+///
+/// ```rust
+/// # futures::executor::block_on(async {
+/// use futures::io::{self, AsyncWriteExt};
+///
+/// let buffer = vec![1, 2, 3, 5, 8];
+/// let mut writer = io::sink();
+/// let num_bytes = writer.write(&buffer).await?;
+/// assert_eq!(num_bytes, 5);
+/// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+/// ```
+pub fn sink() -> Sink {
+    Sink { _priv: () }
+}
+
+impl AsyncWrite for Sink {
+    #[inline]
+    fn poll_write(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(Ok(buf.len()))
+    }
+
+    #[inline]
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Ready(Ok(bufs.iter().map(|b| b.len()).sum()))
+    }
+
+    #[inline]
+    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+    #[inline]
+    fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl fmt::Debug for Sink {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.pad("Sink { .. }")
+    }
+}
diff --git a/futures-util/src/io/take.rs b/futures-util/src/io/take.rs
new file mode 100644
index 00000000..3eaa5f5c
--- /dev/null
+++ b/futures-util/src/io/take.rs
@@ -0,0 +1,215 @@
+use futures_core::task::{Context, Poll};
+#[cfg(feature = "read_initializer")]
+use futures_io::Initializer;
+use futures_io::{AsyncRead, AsyncBufRead};
+use pin_utils::{unsafe_pinned, unsafe_unpinned};
+use std::{cmp, io};
+use std::pin::Pin;
+
+/// Reader for the [`take`](super::AsyncReadExt::take) method.
+#[derive(Debug)]
+#[must_use = "readers do nothing unless you `.await` or poll them"]
+pub struct Take<R> {
+    inner: R,
+    // Add '_' to avoid conflicts with `limit` method.
+    limit_: u64,
+}
+
+impl<R: Unpin> Unpin for Take<R> { }
+
+impl<R: AsyncRead> Take<R> {
+    unsafe_pinned!(inner: R);
+    unsafe_unpinned!(limit_: u64);
+
+    pub(super) fn new(inner: R, limit: u64) -> Self {
+        Self { inner, limit_: limit }
+    }
+
+    /// Returns the remaining number of bytes that can be
+    /// read before this instance will return EOF.
+    ///
+    /// # Note
+    ///
+    /// This instance may reach `EOF` after reading fewer bytes than indicated by
+    /// this method if the underlying [`AsyncRead`] instance reaches EOF.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::AsyncReadExt;
+    /// use std::io::Cursor;
+    ///
+    /// let reader = Cursor::new(&b"12345678"[..]);
+    /// let mut buffer = [0; 2];
+    ///
+    /// let mut take = reader.take(4);
+    /// let n = take.read(&mut buffer).await?;
+    ///
+    /// assert_eq!(take.limit(), 2);
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    pub fn limit(&self) -> u64 {
+        self.limit_
+    }
+
+    /// Sets the number of bytes that can be read before this instance will
+    /// return EOF. This is the same as constructing a new `Take` instance, so
+    /// the amount of bytes read and the previous limit value don't matter when
+    /// calling this method.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::AsyncReadExt;
+    /// use std::io::Cursor;
+    ///
+    /// let reader = Cursor::new(&b"12345678"[..]);
+    /// let mut buffer = [0; 4];
+    ///
+    /// let mut take = reader.take(4);
+    /// let n = take.read(&mut buffer).await?;
+    ///
+    /// assert_eq!(n, 4);
+    /// assert_eq!(take.limit(), 0);
+    ///
+    /// take.set_limit(10);
+    /// let n = take.read(&mut buffer).await?;
+    /// assert_eq!(n, 4);
+    ///
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    pub fn set_limit(&mut self, limit: u64) {
+        self.limit_ = limit
+    }
+
+    /// Gets a reference to the underlying reader.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::AsyncReadExt;
+    /// use std::io::Cursor;
+    ///
+    /// let reader = Cursor::new(&b"12345678"[..]);
+    /// let mut buffer = [0; 4];
+    ///
+    /// let mut take = reader.take(4);
+    /// let n = take.read(&mut buffer).await?;
+    ///
+    /// let cursor_ref = take.get_ref();
+    /// assert_eq!(cursor_ref.position(), 4);
+    ///
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    pub fn get_ref(&self) -> &R {
+        &self.inner
+    }
+
+    /// Gets a mutable reference to the underlying reader.
+    ///
+    /// Care should be taken to avoid modifying the internal I/O state of the
+    /// underlying reader as doing so may corrupt the internal limit of this
+    /// `Take`.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::AsyncReadExt;
+    /// use std::io::Cursor;
+    ///
+    /// let reader = Cursor::new(&b"12345678"[..]);
+    /// let mut buffer = [0; 4];
+    ///
+    /// let mut take = reader.take(4);
+    /// let n = take.read(&mut buffer).await?;
+    ///
+    /// let cursor_mut = take.get_mut();
+    ///
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    pub fn get_mut(&mut self) -> &mut R {
+        &mut self.inner
+    }
+
+    /// Gets a pinned mutable reference to the underlying reader.
+    ///
+    /// Care should be taken to avoid modifying the internal I/O state of the
+    /// underlying reader as doing so may corrupt the internal limit of this
+    /// `Take`.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut R> {
+        self.inner()
+    }
+
+    /// Consumes the `Take`, returning the wrapped reader.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::io::AsyncReadExt;
+    /// use std::io::Cursor;
+    ///
+    /// let reader = Cursor::new(&b"12345678"[..]);
+    /// let mut buffer = [0; 4];
+    ///
+    /// let mut take = reader.take(4);
+    /// let n = take.read(&mut buffer).await?;
+    ///
+    /// let cursor = take.into_inner();
+    /// assert_eq!(cursor.position(), 4);
+    ///
+    /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
+    /// ```
+    pub fn into_inner(self) -> R {
+        self.inner
+    }
+}
+
+impl<R: AsyncRead> AsyncRead for Take<R> {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<Result<usize, io::Error>> {
+        if self.limit_ == 0 {
+            return Poll::Ready(Ok(0));
+        }
+
+        let max = std::cmp::min(buf.len() as u64, self.limit_) as usize;
+        let n = ready!(self.as_mut().inner().poll_read(cx, &mut buf[..max]))?;
+        *self.as_mut().limit_() -= n as u64;
+        Poll::Ready(Ok(n))
+    }
+
+    #[cfg(feature = "read_initializer")]
+    unsafe fn initializer(&self) -> Initializer {
+        self.inner.initializer()
+    }
+}
+
+impl<R: AsyncBufRead> AsyncBufRead for Take<R> {
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        let Self { inner, limit_ } = unsafe { self.get_unchecked_mut() };
+        let inner = unsafe { Pin::new_unchecked(inner) };
+
+        // Don't call into inner reader at all at EOF because it may still block
+        if *limit_ == 0 {
+            return Poll::Ready(Ok(&[]));
+        }
+
+        let buf = ready!(inner.poll_fill_buf(cx)?);
+        let cap = cmp::min(buf.len() as u64, *limit_) as usize;
+        Poll::Ready(Ok(&buf[..cap]))
+    }
+
+    fn consume(mut self: Pin<&mut Self>, amt: usize) {
+        // Don't let callers reset the limit by passing an overlarge value
+        let amt = cmp::min(amt as u64, self.limit_) as usize;
+        *self.as_mut().limit_() -= amt as u64;
+        self.inner().consume(amt);
+    }
+}
diff --git a/futures-util/src/lib.rs b/futures-util/src/lib.rs
index 4c52fe7c..3f91a9eb 100644
--- a/futures-util/src/lib.rs
+++ b/futures-util/src/lib.rs
@@ -1,8 +1,8 @@
 //! Combinators and utilities for working with `Future`s, `Stream`s, `Sink`s,
 //! and the `AsyncRead` and `AsyncWrite` traits.
 
-#![cfg_attr(feature = "async-await", feature(async_await))]
 #![cfg_attr(feature = "cfg-target-has-atomic", feature(cfg_target_has_atomic))]
+#![cfg_attr(feature = "read_initializer", feature(read_initializer))]
 
 #![cfg_attr(not(feature = "std"), no_std)]
 #![warn(missing_docs, missing_debug_implementations, rust_2018_idioms, unreachable_pub)]
@@ -14,11 +14,14 @@
 
 #![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_util")]
 
-#[cfg(all(feature = "cfg-target-has-atomic", not(feature = "nightly")))]
-compile_error!("The `cfg-target-has-atomic` feature requires the `nightly` feature as an explicit opt-in to unstable features");
+#[cfg(all(feature = "cfg-target-has-atomic", not(feature = "unstable")))]
+compile_error!("The `cfg-target-has-atomic` feature requires the `unstable` feature as an explicit opt-in to unstable features");
 
-#[cfg(all(feature = "async-await", not(feature = "nightly")))]
-compile_error!("The `async-await` feature requires the `nightly` feature as an explicit opt-in to unstable features");
+#[cfg(all(feature = "bilock", not(feature = "unstable")))]
+compile_error!("The `bilock` feature requires the `unstable` feature as an explicit opt-in to unstable features");
+
+#[cfg(all(feature = "read_initializer", not(feature = "unstable")))]
+compile_error!("The `read_initializer` feature requires the `unstable` feature as an explicit opt-in to unstable features");
 
 #[cfg(feature = "alloc")]
 extern crate alloc;
@@ -30,21 +33,16 @@ extern crate futures_core;
 pub use futures_core::ready;
 pub use pin_utils::pin_mut;
 
+#[cfg(feature = "std")]
 #[cfg(feature = "async-await")]
 #[macro_use]
 #[doc(hidden)]
 pub mod async_await;
+#[cfg(feature = "std")]
 #[cfg(feature = "async-await")]
 #[doc(hidden)]
 pub use self::async_await::*;
 
-#[cfg(feature = "select-macro")]
-#[doc(hidden)]
-pub mod rand_reexport { // used by select!
-    #[doc(hidden)]
-    pub use rand::{prelude::SliceRandom, thread_rng};
-}
-
 #[doc(hidden)]
 pub use futures_core::core_reexport;
 
@@ -110,6 +108,9 @@ pub mod sink;
 
 pub mod task;
 
+pub mod never;
+#[doc(hidden)] pub use crate::never::Never;
+
 #[cfg(feature = "compat")]
 pub mod compat;
 
diff --git a/futures-util/src/lock/bilock.rs b/futures-util/src/lock/bilock.rs
index f749dc2a..5e1fe35b 100644
--- a/futures-util/src/lock/bilock.rs
+++ b/futures-util/src/lock/bilock.rs
@@ -1,21 +1,16 @@
 //! Futures-powered synchronization primitives.
-#![allow(unused)]
 
+#[cfg(feature = "bilock")]
 use futures_core::future::Future;
 use futures_core::task::{Context, Poll, Waker};
 use core::cell::UnsafeCell;
 use core::fmt;
-use core::mem;
 use core::ops::{Deref, DerefMut};
 use core::pin::Pin;
 use core::sync::atomic::AtomicUsize;
 use core::sync::atomic::Ordering::SeqCst;
 use alloc::boxed::Box;
 use alloc::sync::Arc;
-#[cfg(feature = "std")]
-use std::any::Any;
-#[cfg(feature = "std")]
-use std::error::Error;
 
 /// A type of futures-powered synchronization primitive which is a mutex between
 /// two possible owners.
@@ -35,6 +30,9 @@ use std::error::Error;
 /// example a TCP stream could be both a reader and a writer or a framing layer
 /// could be both a stream and a sink for messages. A `BiLock` enables splitting
 /// these two and then using each independently in a futures-powered fashion.
+///
+/// This type is only available when the `bilock` feature of this
+/// library is activated.
 #[derive(Debug)]
 pub struct BiLock<T> {
     arc: Arc<Inner<T>>,
@@ -140,6 +138,7 @@ impl<T> BiLock<T> {
     /// `BiLockGuard<T>`.
     ///
     /// Note that the returned future will never resolve to an error.
+    #[cfg(feature = "bilock")]
     pub fn lock(&self) -> BiLockAcquire<'_, T> {
         BiLockAcquire {
             bilock: self,
@@ -149,11 +148,12 @@ impl<T> BiLock<T> {
     /// Attempts to put the two "halves" of a `BiLock<T>` back together and
     /// recover the original value. Succeeds only if the two `BiLock<T>`s
     /// originated from the same call to `BiLock::new`.
+    #[cfg(any(feature = "bilock", feature = "sink"))]
     pub fn reunite(self, other: Self) -> Result<T, ReuniteError<T>>
     where
         T: Unpin,
     {
-        if &*self.arc as *const _ == &*other.arc as *const _ {
+        if Arc::ptr_eq(&self.arc, &other.arc) {
             drop(other);
             let inner = Arc::try_unwrap(self.arc)
                 .ok()
@@ -182,6 +182,7 @@ impl<T> BiLock<T> {
     }
 }
 
+#[cfg(any(feature = "bilock", feature = "sink"))]
 impl<T: Unpin> Inner<T> {
     unsafe fn into_value(mut self) -> T {
         self.value.take().unwrap().into_inner()
@@ -213,7 +214,7 @@ impl<T> fmt::Display for ReuniteError<T> {
 }
 
 #[cfg(feature = "std")]
-impl<T: Any> Error for ReuniteError<T> {}
+impl<T: core::any::Any> std::error::Error for ReuniteError<T> {}
 
 /// Returned RAII guard from the `poll_lock` method.
 ///
@@ -255,6 +256,7 @@ impl<T> Drop for BiLockGuard<'_, T> {
 
 /// Future returned by `BiLock::lock` which will resolve when the lock is
 /// acquired.
+#[cfg(feature = "bilock")]
 #[must_use = "futures do nothing unless you `.await` or poll them"]
 #[derive(Debug)]
 pub struct BiLockAcquire<'a, T> {
@@ -262,8 +264,10 @@ pub struct BiLockAcquire<'a, T> {
 }
 
 // Pinning is never projected to fields
+#[cfg(feature = "bilock")]
 impl<T> Unpin for BiLockAcquire<'_, T> {}
 
+#[cfg(feature = "bilock")]
 impl<'a, T> Future for BiLockAcquire<'a, T> {
     type Output = BiLockGuard<'a, T>;
 
diff --git a/futures-util/src/lock/mod.rs b/futures-util/src/lock/mod.rs
index 7a5dc114..6d831f3d 100644
--- a/futures-util/src/lock/mod.rs
+++ b/futures-util/src/lock/mod.rs
@@ -1,16 +1,18 @@
 //! Futures-powered synchronization primitives.
+//!
+//! This module is only available when the `std` or `alloc` feature of this
+//! library is activated, and it is activated by default.
 
 #[cfg(feature = "std")]
 mod mutex;
 #[cfg(feature = "std")]
 pub use self::mutex::{Mutex, MutexLockFuture, MutexGuard};
 
-#[cfg(any(feature = "sink", feature = "io"))]
-#[allow(unreachable_pub)]
+#[cfg(any(feature = "bilock", feature = "sink", feature = "io"))]
+#[cfg_attr(not(feature = "bilock"), allow(unreachable_pub))]
 mod bilock;
-#[cfg(any(feature = "sink", feature = "io"))]
-#[cfg(any(test, feature = "bench"))]
+#[cfg(feature = "bilock")]
 pub use self::bilock::{BiLock, BiLockAcquire, BiLockGuard, ReuniteError};
 #[cfg(any(feature = "sink", feature = "io"))]
-#[cfg(not(any(test, feature = "bench")))]
+#[cfg(not(feature = "bilock"))]
 pub(crate) use self::bilock::BiLock;
diff --git a/futures-util/src/lock/mutex.rs b/futures-util/src/lock/mutex.rs
index 9b1c1286..7e679782 100644
--- a/futures-util/src/lock/mutex.rs
+++ b/futures-util/src/lock/mutex.rs
@@ -25,6 +25,12 @@ impl<T: ?Sized> fmt::Debug for Mutex<T> {
     }
 }
 
+impl<T> From<T> for Mutex<T> {
+    fn from(t: T) -> Self {
+        Self::new(t)
+    }
+}
+
 impl<T: Default> Default for Mutex<T> {
     fn default() -> Mutex<T> {
         Mutex::new(Default::default())
@@ -61,10 +67,24 @@ impl<T> Mutex<T> {
     pub fn new(t: T) -> Mutex<T> {
         Mutex {
             state: AtomicUsize::new(0),
-            value: UnsafeCell::new(t),
             waiters: StdMutex::new(Slab::new()),
+            value: UnsafeCell::new(t),
         }
     }
+
+    /// Consumes this mutex, returning the underlying data.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use futures::lock::Mutex;
+    ///
+    /// let mutex = Mutex::new(0);
+    /// assert_eq!(mutex.into_inner(), 0);
+    /// ```
+    pub fn into_inner(self) -> T {
+        self.value.into_inner()
+    }
 }
 
 impl<T: ?Sized> Mutex<T> {
@@ -91,6 +111,28 @@ impl<T: ?Sized> Mutex<T> {
         }
     }
 
+    /// Returns a mutable reference to the underlying data.
+    ///
+    /// Since this call borrows the `Mutex` mutably, no actual locking needs to
+    /// take place -- the mutable borrow statically guarantees no locks exist.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # futures::executor::block_on(async {
+    /// use futures::lock::Mutex;
+    ///
+    /// let mut mutex = Mutex::new(0);
+    /// *mutex.get_mut() = 10;
+    /// assert_eq!(*mutex.lock().await, 10);
+    /// # });
+    /// ```
+    pub fn get_mut(&mut self) -> &mut T {
+        // We know statically that there are no other references to `self`, so
+        // there's no need to lock the inner mutex.
+        unsafe { &mut *self.value.get() }
+    }
+
     fn remove_waker(&self, wait_key: usize, wake_another: bool) {
         if wait_key != WAIT_KEY_NONE {
             let mut waiters = self.waiters.lock().unwrap();
diff --git a/futures-core/src/never.rs b/futures-util/src/never.rs
similarity index 100%
rename from futures-core/src/never.rs
rename to futures-util/src/never.rs
diff --git a/futures-util/src/sink/buffer.rs b/futures-util/src/sink/buffer.rs
index a39ff9d7..90480bd5 100644
--- a/futures-util/src/sink/buffer.rs
+++ b/futures-util/src/sink/buffer.rs
@@ -1,4 +1,4 @@
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
@@ -76,6 +76,16 @@ impl<S, Item> Stream for Buffer<S, Item> where S: Sink<Item> + Stream {
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<S::Item>> {
         self.sink().poll_next(cx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.sink.size_hint()
+    }
+}
+
+impl<S, Item> FusedStream for Buffer<S, Item> where S: Sink<Item> + FusedStream {
+    fn is_terminated(&self) -> bool {
+        self.sink.is_terminated()
+    }
 }
 
 impl<Si: Sink<Item>, Item> Sink<Item> for Buffer<Si, Item> {
diff --git a/futures-util/src/sink/drain.rs b/futures-util/src/sink/drain.rs
index 8cd0fa56..46de83b7 100644
--- a/futures-util/src/sink/drain.rs
+++ b/futures-util/src/sink/drain.rs
@@ -1,7 +1,7 @@
+use crate::never::Never;
 use core::marker::PhantomData;
 use core::pin::Pin;
 use futures_core::task::{Context, Poll};
-use futures_core::never::Never;
 use futures_sink::Sink;
 
 /// Sink for the [`drain`] function.
@@ -18,7 +18,6 @@ pub struct Drain<T> {
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::sink::{self, SinkExt};
 ///
diff --git a/futures-util/src/sink/err_into.rs b/futures-util/src/sink/err_into.rs
index b1b99a2b..af198fea 100644
--- a/futures-util/src/sink/err_into.rs
+++ b/futures-util/src/sink/err_into.rs
@@ -1,6 +1,6 @@
 use crate::sink::{SinkExt, SinkMapErr};
 use core::pin::Pin;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use futures_sink::{Sink};
 use pin_utils::unsafe_pinned;
@@ -57,6 +57,7 @@ impl<Si, Item, E> Sink<Item> for SinkErrInto<Si, Item, E>
     delegate_sink!(sink, Item);
 }
 
+// Forwarding impl of Stream from the underlying sink
 impl<S, Item, E> Stream for SinkErrInto<S, Item, E>
     where S: Sink<Item> + Stream,
           S::Error: Into<E>
@@ -69,4 +70,17 @@ impl<S, Item, E> Stream for SinkErrInto<S, Item, E>
     ) -> Poll<Option<S::Item>> {
         self.sink().poll_next(cx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.sink.size_hint()
+    }
+}
+
+impl<S, Item, E> FusedStream for SinkErrInto<S, Item, E>
+    where S: Sink<Item> + FusedStream,
+          S::Error: Into<E>
+{
+    fn is_terminated(&self) -> bool {
+        self.sink.is_terminated()
+    }
 }
diff --git a/futures-util/src/sink/fanout.rs b/futures-util/src/sink/fanout.rs
index 00d30624..24e4de95 100644
--- a/futures-util/src/sink/fanout.rs
+++ b/futures-util/src/sink/fanout.rs
@@ -33,11 +33,11 @@ impl<Si1, Si2> Fanout<Si1, Si2> {
     }
 
     /// Get a pinned mutable reference to the inner sinks.
-    pub fn get_pin_mut(self: Pin<&mut Self>) -> (Pin<&mut Si1>, Pin<&mut Si2>)
-        where Si1: Unpin, Si2: Unpin,
-    {
-        let Self { sink1, sink2 } = self.get_mut();
-        (Pin::new(sink1), Pin::new(sink2))
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> (Pin<&mut Si1>, Pin<&mut Si2>) {
+        unsafe {
+            let Self { sink1, sink2 } = self.get_unchecked_mut();
+            (Pin::new_unchecked(sink1), Pin::new_unchecked(sink2))
+        }
     }
 
     /// Consumes this combinator, returning the underlying sinks.
diff --git a/futures-util/src/sink/map_err.rs b/futures-util/src/sink/map_err.rs
index a7b102ca..7333d204 100644
--- a/futures-util/src/sink/map_err.rs
+++ b/futures-util/src/sink/map_err.rs
@@ -1,5 +1,5 @@
 use core::pin::Pin;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use futures_sink::{Sink};
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
@@ -85,6 +85,7 @@ impl<Si, F, E, Item> Sink<Item> for SinkMapErr<Si, F>
     }
 }
 
+// Forwarding impl of Stream from the underlying sink
 impl<S: Stream, F> Stream for SinkMapErr<S, F> {
     type Item = S::Item;
 
@@ -94,4 +95,14 @@ impl<S: Stream, F> Stream for SinkMapErr<S, F> {
     ) -> Poll<Option<S::Item>> {
         self.sink().poll_next(cx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.sink.size_hint()
+    }
+}
+
+impl<S: FusedStream, F> FusedStream for SinkMapErr<S, F> {
+    fn is_terminated(&self) -> bool {
+        self.sink.is_terminated()
+    }
 }
diff --git a/futures-util/src/sink/mod.rs b/futures-util/src/sink/mod.rs
index ea4ab6f4..c35a48f1 100644
--- a/futures-util/src/sink/mod.rs
+++ b/futures-util/src/sink/mod.rs
@@ -8,12 +8,13 @@
 
 use futures_core::future::Future;
 use futures_core::stream::Stream;
-use futures_sink::Sink;
 use crate::future::Either;
 
 #[cfg(feature = "compat")]
 use crate::compat::CompatSink;
 
+pub use futures_sink::Sink;
+
 mod close;
 pub use self::close::Close;
 
@@ -89,7 +90,6 @@ pub trait SinkExt<Item>: Sink<Item> {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::mpsc;
     /// use futures::sink::SinkExt;
diff --git a/futures-util/src/sink/send_all.rs b/futures-util/src/sink/send_all.rs
index bcc6a8c9..65e5bf52 100644
--- a/futures-util/src/sink/send_all.rs
+++ b/futures-util/src/sink/send_all.rs
@@ -87,7 +87,7 @@ where
                     return Poll::Ready(Ok(()))
                 }
                 Poll::Pending => {
-                    ready!(Pin::new(&mut this.sink).poll_ready(cx))?;
+                    ready!(Pin::new(&mut this.sink).poll_flush(cx))?;
                     return Poll::Pending
                 }
             }
diff --git a/futures-util/src/sink/with.rs b/futures-util/src/sink/with.rs
index 9a4afe4c..9ea47a73 100644
--- a/futures-util/src/sink/with.rs
+++ b/futures-util/src/sink/with.rs
@@ -97,6 +97,10 @@ impl<S, Item, U, Fut, F> Stream for With<S, Item, U, Fut, F>
     ) -> Poll<Option<S::Item>> {
         self.sink().poll_next(cx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.sink.size_hint()
+    }
 }
 
 impl<Si, Item, U, Fut, F, E> With<Si, Item, U, Fut, F>
diff --git a/futures-util/src/sink/with_flat_map.rs b/futures-util/src/sink/with_flat_map.rs
index 1aa18678..baa7536f 100644
--- a/futures-util/src/sink/with_flat_map.rs
+++ b/futures-util/src/sink/with_flat_map.rs
@@ -1,7 +1,7 @@
 use core::fmt;
 use core::marker::PhantomData;
 use core::pin::Pin;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
@@ -113,6 +113,7 @@ where
     }
 }
 
+// Forwarding impl of Stream from the underlying sink
 impl<S, Item, U, St, F> Stream for WithFlatMap<S, Item, U, St, F>
 where
     S: Stream + Sink<Item>,
@@ -120,12 +121,28 @@ where
     St: Stream<Item = Result<Item, S::Error>>,
 {
     type Item = S::Item;
+
     fn poll_next(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<S::Item>> {
         self.sink().poll_next(cx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.sink.size_hint()
+    }
+}
+
+impl<S, Item, U, St, F> FusedStream for WithFlatMap<S, Item, U, St, F>
+where
+    S: FusedStream + Sink<Item>,
+    F: FnMut(U) -> St,
+    St: Stream<Item = Result<Item, S::Error>>,
+{
+    fn is_terminated(&self) -> bool {
+        self.sink.is_terminated()
+    }
 }
 
 impl<Si, Item, U, St, F> Sink<U> for WithFlatMap<Si, Item, U, St, F>
diff --git a/futures-util/src/stream/buffer_unordered.rs b/futures-util/src/stream/buffer_unordered.rs
index 4ade2823..7a750b31 100644
--- a/futures-util/src/stream/buffer_unordered.rs
+++ b/futures-util/src/stream/buffer_unordered.rs
@@ -127,6 +127,17 @@ where
             Poll::Pending
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let queue_len = self.in_progress_queue.len();
+        let (lower, upper) = self.stream.size_hint();
+        let lower = lower.saturating_add(queue_len);
+        let upper = match upper {
+            Some(x) => x.checked_add(queue_len),
+            None => None,
+        };
+        (lower, upper)
+    }
 }
 
 impl<St> FusedStream for BufferUnordered<St>
diff --git a/futures-util/src/stream/buffered.rs b/futures-util/src/stream/buffered.rs
index 7b451c5e..1a31b639 100644
--- a/futures-util/src/stream/buffered.rs
+++ b/futures-util/src/stream/buffered.rs
@@ -122,6 +122,17 @@ where
             Poll::Pending
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let queue_len = self.in_progress_queue.len();
+        let (lower, upper) = self.stream.size_hint();
+        let lower = lower.saturating_add(queue_len);
+        let upper = match upper {
+            Some(x) => x.checked_add(queue_len),
+            None => None,
+        };
+        (lower, upper)
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/catch_unwind.rs b/futures-util/src/stream/catch_unwind.rs
index 69669884..309625b7 100644
--- a/futures-util/src/stream/catch_unwind.rs
+++ b/futures-util/src/stream/catch_unwind.rs
@@ -1,4 +1,4 @@
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 use std::any::Any;
@@ -22,8 +22,7 @@ impl<St: Stream + UnwindSafe> CatchUnwind<St> {
     }
 }
 
-impl<St: Stream + UnwindSafe> Stream for CatchUnwind<St>
-{
+impl<St: Stream + UnwindSafe> Stream for CatchUnwind<St> {
     type Item = Result<St::Item, Box<dyn Any + Send>>;
 
     fn poll_next(
@@ -46,4 +45,18 @@ impl<St: Stream + UnwindSafe> Stream for CatchUnwind<St>
             }
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        if self.caught_unwind {
+            (0, Some(0))
+        } else {
+            self.stream.size_hint()
+        }
+    }
+}
+
+impl<St: FusedStream + UnwindSafe> FusedStream for CatchUnwind<St> {
+    fn is_terminated(&self) -> bool {
+        self.caught_unwind || self.stream.is_terminated()
+    }
 }
diff --git a/futures-util/src/stream/chain.rs b/futures-util/src/stream/chain.rs
index 22f88cb0..b2ada69c 100644
--- a/futures-util/src/stream/chain.rs
+++ b/futures-util/src/stream/chain.rs
@@ -54,4 +54,22 @@ where St1: Stream,
         self.as_mut().first().set(None);
         self.as_mut().second().poll_next(cx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        if let Some(first) = &self.first {
+            let (first_lower, first_upper) = first.size_hint();
+            let (second_lower, second_upper) = self.second.size_hint();
+
+            let lower = first_lower.saturating_add(second_lower);
+
+            let upper = match (first_upper, second_upper) {
+                (Some(x), Some(y)) => x.checked_add(y),
+                _ => None
+            };
+
+            (lower, upper)
+        } else {
+            self.second.size_hint()
+        }
+    }
 }
diff --git a/futures-util/src/stream/chunks.rs b/futures-util/src/stream/chunks.rs
index bf75bb24..87e9fbb8 100644
--- a/futures-util/src/stream/chunks.rs
+++ b/futures-util/src/stream/chunks.rs
@@ -1,5 +1,5 @@
 use crate::stream::Fuse;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
@@ -105,6 +105,23 @@ impl<St: Stream> Stream for Chunks<St> {
             }
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let chunk_len = if self.items.is_empty() { 0 } else { 1 };
+        let (lower, upper) = self.stream.size_hint();
+        let lower = lower.saturating_add(chunk_len);
+        let upper = match upper {
+            Some(x) => x.checked_add(chunk_len),
+            None => None,
+        };
+        (lower, upper)
+    }
+}
+
+impl<St: FusedStream> FusedStream for Chunks<St> {
+    fn is_terminated(&self) -> bool {
+        self.stream.is_terminated() && self.items.is_empty()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/empty.rs b/futures-util/src/stream/empty.rs
index e176bc17..903af685 100644
--- a/futures-util/src/stream/empty.rs
+++ b/futures-util/src/stream/empty.rs
@@ -33,4 +33,8 @@ impl<T> Stream for Empty<T> {
     fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         Poll::Ready(None)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (0, Some(0))
+    }
 }
diff --git a/futures-util/src/stream/enumerate.rs b/futures-util/src/stream/enumerate.rs
index c5fbdf34..a688459a 100644
--- a/futures-util/src/stream/enumerate.rs
+++ b/futures-util/src/stream/enumerate.rs
@@ -81,6 +81,10 @@ impl<St: Stream> Stream for Enumerate<St> {
             None => Poll::Ready(None),
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/filter.rs b/futures-util/src/stream/filter.rs
index ad807da7..06335f1e 100644
--- a/futures-util/src/stream/filter.rs
+++ b/futures-util/src/stream/filter.rs
@@ -132,6 +132,16 @@ impl<St, Fut, F> Stream for Filter<St, Fut, F>
             }
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let pending_len = if self.pending_item.is_some() { 1 } else { 0 };
+        let (_, upper) = self.stream.size_hint();
+        let upper = match upper {
+            Some(x) => x.checked_add(pending_len),
+            None => None,
+        };
+        (0, upper) // can't know a lower bound, due to the predicate
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/filter_map.rs b/futures-util/src/stream/filter_map.rs
index d22cd8bd..532e6cad 100644
--- a/futures-util/src/stream/filter_map.rs
+++ b/futures-util/src/stream/filter_map.rs
@@ -118,6 +118,16 @@ impl<St, Fut, F, T> Stream for FilterMap<St, Fut, F>
             }
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let pending_len = if self.pending.is_some() { 1 } else { 0 };
+        let (_, upper) = self.stream.size_hint();
+        let upper = match upper {
+            Some(x) => x.checked_add(pending_len),
+            None => None,
+        };
+        (0, upper) // can't know a lower bound, due to the predicate
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/fuse.rs b/futures-util/src/stream/fuse.rs
index d4ac3e4e..9085dc55 100644
--- a/futures-util/src/stream/fuse.rs
+++ b/futures-util/src/stream/fuse.rs
@@ -88,6 +88,14 @@ impl<S: Stream> Stream for Fuse<S> {
         }
         Poll::Ready(item)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        if self.done {
+            (0, Some(0))
+        } else {
+            self.stream.size_hint()
+        }
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/futures_ordered.rs b/futures-util/src/stream/futures_ordered.rs
index 7f5d6cc0..a30cbaa4 100644
--- a/futures-util/src/stream/futures_ordered.rs
+++ b/futures-util/src/stream/futures_ordered.rs
@@ -181,6 +181,11 @@ impl<Fut: Future> Stream for FuturesOrdered<Fut> {
             }
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let len = self.len();
+        (len, Some(len))
+    }
 }
 
 impl<Fut: Future> Debug for FuturesOrdered<Fut> {
diff --git a/futures-util/src/stream/futures_unordered/mod.rs b/futures-util/src/stream/futures_unordered/mod.rs
index dd930ad2..45666183 100644
--- a/futures-util/src/stream/futures_unordered/mod.rs
+++ b/futures-util/src/stream/futures_unordered/mod.rs
@@ -427,6 +427,11 @@ impl<Fut: Future> Stream for FuturesUnordered<Fut> {
             }
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let len = self.len();
+        (len, Some(len))
+    }
 }
 
 impl<Fut> Debug for FuturesUnordered<Fut> {
diff --git a/futures-util/src/stream/inspect.rs b/futures-util/src/stream/inspect.rs
index 75ca2361..e34970ae 100644
--- a/futures-util/src/stream/inspect.rs
+++ b/futures-util/src/stream/inspect.rs
@@ -101,6 +101,10 @@ impl<St, F> Stream for Inspect<St, F>
             .poll_next(cx)
             .map(|opt| opt.map(|e| inspect(e, self.as_mut().f())))
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/into_future.rs b/futures-util/src/stream/into_future.rs
index 556e388f..dce4b402 100644
--- a/futures-util/src/stream/into_future.rs
+++ b/futures-util/src/stream/into_future.rs
@@ -3,6 +3,7 @@ use core::pin::Pin;
 use futures_core::future::{FusedFuture, Future};
 use futures_core::stream::Stream;
 use futures_core::task::{Context, Poll};
+use pin_utils::unsafe_pinned;
 
 /// Future for the [`into_future`](super::StreamExt::into_future) method.
 #[derive(Debug)]
@@ -14,6 +15,8 @@ pub struct StreamFuture<St> {
 impl<St: Stream + Unpin> Unpin for StreamFuture<St> {}
 
 impl<St: Stream + Unpin> StreamFuture<St> {
+    unsafe_pinned!(stream: Option<St>);
+
     pub(super) fn new(stream: St) -> StreamFuture<St> {
         StreamFuture { stream: Some(stream) }
     }
@@ -54,7 +57,7 @@ impl<St: Stream + Unpin> StreamFuture<St> {
     /// in order to return it to the caller of `Future::poll` if the stream yielded
     /// an element.
     pub fn get_pin_mut(self: Pin<&mut Self>) -> Option<Pin<&mut St>> {
-        Pin::new(&mut self.get_mut().stream).as_pin_mut()
+        self.stream().as_pin_mut()
     }
 
     /// Consumes this combinator, returning the underlying stream.
diff --git a/futures-util/src/stream/iter.rs b/futures-util/src/stream/iter.rs
index d72afd0f..e9df81ce 100644
--- a/futures-util/src/stream/iter.rs
+++ b/futures-util/src/stream/iter.rs
@@ -18,7 +18,6 @@ impl<I> Unpin for Iter<I> {}
 /// simply always calls `iter.next()` and returns that.
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::stream::{self, StreamExt};
 ///
@@ -42,4 +41,8 @@ impl<I> Stream for Iter<I>
     fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<I::Item>> {
         Poll::Ready(self.iter.next())
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.iter.size_hint()
+    }
 }
diff --git a/futures-util/src/stream/map.rs b/futures-util/src/stream/map.rs
index e681cc92..81194342 100644
--- a/futures-util/src/stream/map.rs
+++ b/futures-util/src/stream/map.rs
@@ -94,6 +94,10 @@ impl<St, F, T> Stream for Map<St, F>
             .poll_next(cx)
             .map(|opt| opt.map(|x| self.as_mut().f()(x)))
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/mod.rs b/futures-util/src/stream/mod.rs
index 44bbe25e..39ea5e76 100644
--- a/futures-util/src/stream/mod.rs
+++ b/futures-util/src/stream/mod.rs
@@ -6,7 +6,6 @@
 use crate::future::Either;
 use core::pin::Pin;
 use futures_core::future::Future;
-use futures_core::stream::{FusedStream, Stream};
 #[cfg(feature = "sink")]
 use futures_core::stream::TryStream;
 use futures_core::task::{Context, Poll};
@@ -14,8 +13,10 @@ use futures_core::task::{Context, Poll};
 use futures_sink::Sink;
 #[cfg(feature = "alloc")]
 use alloc::boxed::Box;
+
+pub use futures_core::stream::{FusedStream, Stream};
 #[cfg(feature = "alloc")]
-use futures_core::stream::{BoxStream, LocalBoxStream};
+pub use futures_core::stream::{BoxStream, LocalBoxStream};
 
 mod iter;
 pub use self::iter::{iter, Iter};
@@ -180,7 +181,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, StreamExt};
     ///
@@ -214,7 +214,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, StreamExt};
     ///
@@ -247,7 +246,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, StreamExt};
     ///
@@ -289,7 +287,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, StreamExt};
     ///
@@ -325,7 +322,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::future;
     /// use futures::stream::{self, StreamExt};
@@ -358,15 +354,12 @@ pub trait StreamExt: Stream {
     ///
     /// # Examples
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future;
     /// use futures::stream::{self, StreamExt};
     ///
     /// let stream = stream::iter(1..=10);
-    /// let evens = stream.filter_map(|x| {
-    ///     let ret = if x % 2 == 0 { Some(x + 1) } else { None };
-    ///     future::ready(ret)
+    /// let evens = stream.filter_map(|x| async move {
+    ///     if x % 2 == 0 { Some(x + 1) } else { None }
     /// });
     ///
     /// assert_eq!(vec![3, 5, 7, 9, 11], evens.collect::<Vec<_>>().await);
@@ -393,13 +386,11 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future;
     /// use futures::stream::{self, StreamExt};
     ///
     /// let stream = stream::iter(1..=3);
-    /// let stream = stream.then(|x| future::ready(x + 3));
+    /// let stream = stream.then(|x| async move { x + 3 });
     ///
     /// assert_eq!(vec![4, 5, 6], stream.collect::<Vec<_>>().await);
     /// # });
@@ -420,7 +411,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::mpsc;
     /// use futures::stream::StreamExt;
@@ -457,7 +447,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::mpsc;
     /// use futures::stream::StreamExt;
@@ -497,13 +486,11 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future;
     /// use futures::stream::{self, StreamExt};
     ///
     /// let number_stream = stream::iter(0..6);
-    /// let sum = number_stream.fold(0, |acc, x| future::ready(acc + x));
+    /// let sum = number_stream.fold(0, |acc, x| async move { acc + x });
     /// assert_eq!(sum.await, 15);
     /// # });
     /// ```
@@ -520,7 +507,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::mpsc;
     /// use futures::stream::StreamExt;
@@ -565,7 +551,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::future;
     /// use futures::stream::{self, StreamExt};
@@ -595,7 +580,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::future;
     /// use futures::stream::{self, StreamExt};
@@ -631,7 +615,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::future;
     /// use futures::stream::{self, StreamExt};
@@ -682,7 +665,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::oneshot;
     /// use futures::stream::{self, StreamExt};
@@ -728,7 +710,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, StreamExt};
     ///
@@ -751,7 +732,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, StreamExt};
     ///
@@ -767,7 +747,7 @@ pub trait StreamExt: Stream {
     }
 
     /// Fuse a stream such that [`poll_next`](Stream::poll_next) will never
-    /// again be called once it has finished. This method can be used t turn
+    /// again be called once it has finished. This method can be used to turn
     /// any `Stream` into a `FusedStream`.
     ///
     /// Normally, once a stream has returned [`None`] from
@@ -819,22 +799,20 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future;
     /// use futures::stream::{self, StreamExt};
     ///
     /// let mut stream = stream::iter(1..5);
     ///
     /// let sum = stream.by_ref()
     ///                 .take(2)
-    ///                 .fold(0, |a, b| future::ready(a + b))
+    ///                 .fold(0, |a, b| async move { a + b })
     ///                 .await;
     /// assert_eq!(sum, 3);
     ///
     /// // You can use the stream again
     /// let sum = stream.take(2)
-    ///                 .fold(0, |a, b| future::ready(a + b))
+    ///                 .fold(0, |a, b| async move { a + b })
     ///                 .await;
     /// assert_eq!(sum, 7);
     /// # });
@@ -865,7 +843,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, StreamExt};
     ///
@@ -955,7 +932,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::oneshot;
     /// use futures::stream::{self, StreamExt};
@@ -996,7 +972,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, StreamExt};
     ///
@@ -1017,13 +992,12 @@ pub trait StreamExt: Stream {
         Zip::new(self, other)
     }
 
-    /// Adapter for chaining two stream.
+    /// Adapter for chaining two streams.
     ///
     /// The resulting stream emits elements from the first stream, and when
     /// first stream reaches the end, emits the elements from the second stream.
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, StreamExt};
     ///
@@ -1089,9 +1063,9 @@ pub trait StreamExt: Stream {
     /// exhausted, sending each item to the sink. It will complete once the
     /// stream is exhausted, the sink has received and flushed all items, and
     /// the sink is closed. Note that neither the original stream nor provided
-    /// sink will be output by this future.  Pass the sink by `Pin<&mut S>`
+    /// sink will be output by this future. Pass the sink by `Pin<&mut S>`
     /// (for example, via `forward(&mut sink)` inside an `async` fn/block) in
-    /// order to preserve access to the Sink.
+    /// order to preserve access to the `Sink`.
     #[cfg(feature = "sink")]
     fn forward<S>(self, sink: S) -> Forward<Self, S>
     where
@@ -1189,7 +1163,6 @@ pub trait StreamExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::{future, select};
     /// use futures::stream::{StreamExt, FuturesUnordered};
diff --git a/futures-util/src/stream/once.rs b/futures-util/src/stream/once.rs
index a530bb16..4f68b0ce 100644
--- a/futures-util/src/stream/once.rs
+++ b/futures-util/src/stream/once.rs
@@ -1,18 +1,16 @@
 use core::pin::Pin;
 use futures_core::future::Future;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 use pin_utils::unsafe_pinned;
 
 /// Creates a stream of a single element.
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
-/// use futures::future;
 /// use futures::stream::{self, StreamExt};
 ///
-/// let stream = stream::once(future::ready(17));
+/// let stream = stream::once(async { 17 });
 /// let collected = stream.collect::<Vec<i32>>().await;
 /// assert_eq!(collected, vec![17]);
 /// # });
@@ -39,16 +37,27 @@ impl<Fut> Once<Fut> {
 impl<Fut: Future> Stream for Once<Fut> {
     type Item = Fut::Output;
 
-    fn poll_next(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Option<Fut::Output>> {
-        let val = if let Some(f) = self.as_mut().future().as_pin_mut() {
-            ready!(f.poll(cx))
-        } else {
-            return Poll::Ready(None)
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let v = match self.as_mut().future().as_pin_mut() {
+            Some(fut) => ready!(fut.poll(cx)),
+            None => return Poll::Ready(None),
         };
-        self.future().set(None);
-        Poll::Ready(Some(val))
+
+        self.as_mut().future().set(None);
+        Poll::Ready(Some(v))
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        if self.future.is_some() {
+            (1, Some(1))
+        } else {
+            (0, Some(0))
+        }
+    }
+}
+
+impl<Fut: Future> FusedStream for Once<Fut> {
+    fn is_terminated(&self) -> bool {
+        self.future.is_none()
     }
 }
diff --git a/futures-util/src/stream/peek.rs b/futures-util/src/stream/peek.rs
index 636f4575..ef9a219c 100644
--- a/futures-util/src/stream/peek.rs
+++ b/futures-util/src/stream/peek.rs
@@ -104,6 +104,17 @@ impl<S: Stream> Stream for Peekable<S> {
         }
         self.as_mut().stream().poll_next(cx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let peek_len = if self.peeked.is_some() { 1 } else { 0 };
+        let (lower, upper) = self.stream.size_hint();
+        let lower = lower.saturating_add(peek_len);
+        let upper = match upper {
+            Some(x) => x.checked_add(peek_len),
+            None => None,
+        };
+        (lower, upper)
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/pending.rs b/futures-util/src/stream/pending.rs
index 14730b38..bbe07504 100644
--- a/futures-util/src/stream/pending.rs
+++ b/futures-util/src/stream/pending.rs
@@ -31,4 +31,8 @@ impl<T> Stream for Pending<T> {
     fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         Poll::Pending
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (0, Some(0))
+    }
 }
diff --git a/futures-util/src/stream/repeat.rs b/futures-util/src/stream/repeat.rs
index a63c0bfa..21749eb7 100644
--- a/futures-util/src/stream/repeat.rs
+++ b/futures-util/src/stream/repeat.rs
@@ -1,5 +1,5 @@
 use core::pin::Pin;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 
 /// Stream for the [`repeat`] function.
@@ -16,7 +16,6 @@ pub struct Repeat<T> {
 /// available memory as it tries to just fill up all RAM.
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::stream::{self, StreamExt};
 ///
@@ -40,4 +39,16 @@ impl<T> Stream for Repeat<T>
     fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         Poll::Ready(Some(self.item.clone()))
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (usize::max_value(), None)
+    }
+}
+
+impl<T> FusedStream for Repeat<T>
+    where T: Clone,
+{
+    fn is_terminated(&self) -> bool {
+        false
+    }
 }
diff --git a/futures-util/src/stream/select.rs b/futures-util/src/stream/select.rs
index e40d3f82..b5fb8133 100644
--- a/futures-util/src/stream/select.rs
+++ b/futures-util/src/stream/select.rs
@@ -56,11 +56,11 @@ impl<St1, St2> Select<St1, St2> {
     ///
     /// Note that care must be taken to avoid tampering with the state of the
     /// stream which may otherwise confuse this combinator.
-    pub fn get_pin_mut(self: Pin<&mut Self>) -> (Pin<&mut St1>, Pin<&mut St2>)
-        where St1: Unpin, St2: Unpin,
-    {
-        let Self { stream1, stream2, .. } = self.get_mut();
-        (Pin::new(stream1.get_mut()), Pin::new(stream2.get_mut()))
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> (Pin<&mut St1>, Pin<&mut St2>) {
+        unsafe {
+            let Self { stream1, stream2, .. } = self.get_unchecked_mut();
+            (Pin::new_unchecked(stream1).get_pin_mut(), Pin::new_unchecked(stream2).get_pin_mut())
+        }
     }
 
     /// Consumes this combinator, returning the underlying streams.
diff --git a/futures-util/src/stream/skip.rs b/futures-util/src/stream/skip.rs
index 4dd9ade0..f9edf2ae 100644
--- a/futures-util/src/stream/skip.rs
+++ b/futures-util/src/stream/skip.rs
@@ -81,6 +81,18 @@ impl<St: Stream> Stream for Skip<St> {
 
         self.as_mut().stream().poll_next(cx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let (lower, upper) = self.stream.size_hint();
+
+        let lower = lower.saturating_sub(self.remaining as usize);
+        let upper = match upper {
+            Some(x) => Some(x.saturating_sub(self.remaining as usize)),
+            None => None,
+        };
+
+        (lower, upper)
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/skip_while.rs b/futures-util/src/stream/skip_while.rs
index a3bb320a..666d9dea 100644
--- a/futures-util/src/stream/skip_while.rs
+++ b/futures-util/src/stream/skip_while.rs
@@ -95,7 +95,7 @@ impl<St, Fut, F> FusedStream for SkipWhile<St, Fut, F>
           Fut: Future<Output = bool>,
 {
     fn is_terminated(&self) -> bool {
-        self.stream.is_terminated()
+        self.pending_item.is_none() && self.stream.is_terminated()
     }
 }
 
@@ -135,6 +135,16 @@ impl<St, Fut, F> Stream for SkipWhile<St, Fut, F>
             }
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let pending_len = if self.pending_item.is_some() { 1 } else { 0 };
+        let (_, upper) = self.stream.size_hint();
+        let upper = match upper {
+            Some(x) => x.checked_add(pending_len),
+            None => None,
+        };
+        (0, upper) // can't know a lower bound, due to the predicate
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/split.rs b/futures-util/src/stream/split.rs
index c6151fbf..28d7aa71 100644
--- a/futures-util/src/stream/split.rs
+++ b/futures-util/src/stream/split.rs
@@ -3,10 +3,6 @@ use futures_core::task::{Context, Poll};
 use futures_sink::Sink;
 use core::fmt;
 use core::pin::Pin;
-#[cfg(feature = "std")]
-use std::any::Any;
-#[cfg(feature = "std")]
-use std::error::Error;
 
 use crate::lock::BiLock;
 
@@ -47,12 +43,12 @@ fn SplitSink<S: Sink<Item>, Item>(lock: BiLock<S>) -> SplitSink<S, Item> {
 /// A `Sink` part of the split pair
 #[derive(Debug)]
 #[must_use = "sinks do nothing unless polled"]
-pub struct SplitSink<S: Sink<Item>, Item> {
+pub struct SplitSink<S, Item> {
     lock: BiLock<S>,
     slot: Option<Item>,
 }
 
-impl<S: Sink<Item>, Item> Unpin for SplitSink<S, Item> {}
+impl<S, Item> Unpin for SplitSink<S, Item> {}
 
 impl<S: Sink<Item> + Unpin, Item> SplitSink<S, Item> {
     /// Attempts to put the two "halves" of a split `Stream + Sink` back
@@ -112,9 +108,9 @@ pub(super) fn split<S: Stream + Sink<Item>, Item>(s: S) -> (SplitSink<S, Item>,
 
 /// Error indicating a `SplitSink<S>` and `SplitStream<S>` were not two halves
 /// of a `Stream + Split`, and thus could not be `reunite`d.
-pub struct ReuniteError<T: Sink<Item>, Item>(pub SplitSink<T, Item>, pub SplitStream<T>);
+pub struct ReuniteError<T, Item>(pub SplitSink<T, Item>, pub SplitStream<T>);
 
-impl<T: Sink<Item>, Item> fmt::Debug for ReuniteError<T, Item> {
+impl<T, Item> fmt::Debug for ReuniteError<T, Item> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_tuple("ReuniteError")
             .field(&"...")
@@ -122,11 +118,11 @@ impl<T: Sink<Item>, Item> fmt::Debug for ReuniteError<T, Item> {
     }
 }
 
-impl<T: Sink<Item>, Item> fmt::Display for ReuniteError<T, Item> {
+impl<T, Item> fmt::Display for ReuniteError<T, Item> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "tried to reunite a SplitStream and SplitSink that don't form a pair")
     }
 }
 
 #[cfg(feature = "std")]
-impl<T: Any + Sink<Item>, Item> Error for ReuniteError<T, Item> {}
+impl<T: core::any::Any, Item> std::error::Error for ReuniteError<T, Item> {}
diff --git a/futures-util/src/stream/take.rs b/futures-util/src/stream/take.rs
index df275083..4d0dbd31 100644
--- a/futures-util/src/stream/take.rs
+++ b/futures-util/src/stream/take.rs
@@ -1,5 +1,6 @@
+use core::cmp;
 use core::pin::Pin;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
@@ -79,6 +80,31 @@ impl<St> Stream for Take<St>
             Poll::Ready(next)
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        if self.remaining == 0 {
+            return (0, Some(0));
+        }
+
+        let (lower, upper) = self.stream.size_hint();
+
+        let lower = cmp::min(lower, self.remaining as usize);
+
+        let upper = match upper {
+            Some(x) if x < self.remaining as usize => Some(x),
+            _ => Some(self.remaining as usize)
+        };
+
+        (lower, upper)
+    }
+}
+
+impl<St> FusedStream for Take<St>
+    where St: FusedStream,
+{
+    fn is_terminated(&self) -> bool {
+        self.remaining == 0 || self.stream.is_terminated()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/take_while.rs b/futures-util/src/stream/take_while.rs
index 1f0ba32c..68606ec2 100644
--- a/futures-util/src/stream/take_while.rs
+++ b/futures-util/src/stream/take_while.rs
@@ -1,7 +1,7 @@
 use core::fmt;
 use core::pin::Pin;
 use futures_core::future::Future;
-use futures_core::stream::Stream;
+use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
@@ -127,6 +127,30 @@ impl<St, Fut, F> Stream for TakeWhile<St, Fut, F>
             Poll::Ready(None)
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        if self.done_taking {
+            return (0, Some(0));
+        }
+
+        let pending_len = if self.pending_item.is_some() { 1 } else { 0 };
+        let (_, upper) = self.stream.size_hint();
+        let upper = match upper {
+            Some(x) => x.checked_add(pending_len),
+            None => None,
+        };
+        (0, upper) // can't know a lower bound, due to the predicate
+    }
+}
+
+impl<St, Fut, F> FusedStream for TakeWhile<St, Fut, F>
+    where St: FusedStream,
+          F: FnMut(&St::Item) -> Fut,
+          Fut: Future<Output = bool>,
+{
+    fn is_terminated(&self) -> bool {
+        self.done_taking || self.pending_item.is_none() && self.stream.is_terminated()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/then.rs b/futures-util/src/stream/then.rs
index 7e16e6d6..39843b25 100644
--- a/futures-util/src/stream/then.rs
+++ b/futures-util/src/stream/then.rs
@@ -115,6 +115,17 @@ impl<St, Fut, F> Stream for Then<St, Fut, F>
         self.as_mut().future().set(None);
         Poll::Ready(Some(e))
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let future_len = if self.future.is_some() { 1 } else { 0 };
+        let (lower, upper) = self.stream.size_hint();
+        let lower = lower.saturating_add(future_len);
+        let upper = match upper {
+            Some(x) => x.checked_add(future_len),
+            None => None,
+        };
+        (lower, upper)
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/stream/unfold.rs b/futures-util/src/stream/unfold.rs
index 91835128..223f6457 100644
--- a/futures-util/src/stream/unfold.rs
+++ b/futures-util/src/stream/unfold.rs
@@ -31,18 +31,16 @@ use pin_utils::{unsafe_pinned, unsafe_unpinned};
 /// # Example
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
-/// use futures::future;
 /// use futures::stream::{self, StreamExt};
 ///
-/// let stream = stream::unfold(0, |state| {
+/// let stream = stream::unfold(0, |state| async move {
 ///     if state <= 2 {
 ///         let next_state = state + 1;
 ///         let yielded = state  * 2;
-///         future::ready(Some((yielded, next_state)))
+///         Some((yielded, next_state))
 ///     } else {
-///         future::ready(None)
+///         None
 ///     }
 /// });
 ///
diff --git a/futures-util/src/stream/zip.rs b/futures-util/src/stream/zip.rs
index 83b4a0a7..f97ac17d 100644
--- a/futures-util/src/stream/zip.rs
+++ b/futures-util/src/stream/zip.rs
@@ -1,4 +1,5 @@
 use crate::stream::{StreamExt, Fuse};
+use core::cmp;
 use core::pin::Pin;
 use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::{Context, Poll};
@@ -58,11 +59,11 @@ impl<St1: Stream, St2: Stream> Zip<St1, St2> {
     ///
     /// Note that care must be taken to avoid tampering with the state of the
     /// stream which may otherwise confuse this combinator.
-    pub fn get_pin_mut(self: Pin<&mut Self>) -> (Pin<&mut St1>, Pin<&mut St2>)
-        where St1: Unpin, St2: Unpin,
-    {
-        let Self { stream1, stream2, .. } = self.get_mut();
-        (Pin::new(stream1.get_mut()), Pin::new(stream2.get_mut()))
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> (Pin<&mut St1>, Pin<&mut St2>) {
+        unsafe {
+            let Self { stream1, stream2, .. } = self.get_unchecked_mut();
+            (Pin::new_unchecked(stream1).get_pin_mut(), Pin::new_unchecked(stream2).get_pin_mut())
+        }
     }
 
     /// Consumes this combinator, returning the underlying streams.
@@ -114,4 +115,29 @@ impl<St1, St2> Stream for Zip<St1, St2>
             Poll::Pending
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let queued1_len = if self.queued1.is_some() { 1 } else { 0 };
+        let queued2_len = if self.queued2.is_some() { 1 } else { 0 };
+        let (stream1_lower, stream1_upper) = self.stream1.size_hint();
+        let (stream2_lower, stream2_upper) = self.stream2.size_hint();
+
+        let stream1_lower = stream1_lower.saturating_add(queued1_len);
+        let stream2_lower = stream2_lower.saturating_add(queued2_len);
+
+        let lower = cmp::min(stream1_lower, stream2_lower);
+
+        let upper = match (stream1_upper, stream2_upper) {
+            (Some(x), Some(y)) => {
+                let x = x.saturating_add(queued1_len);
+                let y = y.saturating_add(queued2_len);
+                Some(cmp::min(x, y))
+            }
+            (Some(x), None) => x.checked_add(queued1_len),
+            (None, Some(y)) => y.checked_add(queued2_len),
+            (None, None) => None
+        };
+
+        (lower, upper)
+    }
 }
diff --git a/futures-util/src/task/mod.rs b/futures-util/src/task/mod.rs
index 3a24265a..fdc70136 100644
--- a/futures-util/src/task/mod.rs
+++ b/futures-util/src/task/mod.rs
@@ -1,20 +1,6 @@
 //! Task notification
 
 cfg_target_has_atomic! {
-    /// A macro for creating a `RawWaker` vtable for a type that implements
-    /// the `ArcWake` trait.
-    #[cfg(feature = "alloc")]
-    macro_rules! waker_vtable {
-        ($ty:ident) => {
-            &RawWakerVTable::new(
-                clone_arc_raw::<$ty>,
-                wake_arc_raw::<$ty>,
-                wake_by_ref_arc_raw::<$ty>,
-                drop_arc_raw::<$ty>,
-            )
-        };
-    }
-
     #[cfg(feature = "alloc")]
     mod arc_wake;
     #[cfg(feature = "alloc")]
@@ -41,6 +27,4 @@ pub use self::noop_waker::noop_waker_ref;
 mod spawn;
 pub use self::spawn::{SpawnExt, LocalSpawnExt};
 
-// re-export for `select!`
-#[doc(hidden)]
-pub use futures_core::task::{Context, Poll, Waker};
+pub use futures_core::task::{Context, Poll, Waker, RawWaker, RawWakerVTable};
diff --git a/futures-util/src/task/spawn.rs b/futures-util/src/task/spawn.rs
index 23617969..e869f1ef 100644
--- a/futures-util/src/task/spawn.rs
+++ b/futures-util/src/task/spawn.rs
@@ -33,7 +33,6 @@ pub trait SpawnExt: Spawn {
     /// today. Feel free to use this method in the meantime.
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::executor::ThreadPool;
     /// use futures::task::SpawnExt;
     ///
@@ -58,7 +57,6 @@ pub trait SpawnExt: Spawn {
     /// resolves to the output of the spawned future.
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::executor::ThreadPool;
     /// use futures::future;
     /// use futures::task::SpawnExt;
@@ -112,7 +110,6 @@ pub trait LocalSpawnExt: LocalSpawn {
     /// today. Feel free to use this method in the meantime.
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::executor::LocalPool;
     /// use futures::task::LocalSpawnExt;
     ///
@@ -138,15 +135,13 @@ pub trait LocalSpawnExt: LocalSpawn {
     /// resolves to the output of the spawned future.
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::executor::LocalPool;
-    /// use futures::future;
     /// use futures::task::LocalSpawnExt;
     ///
     /// let mut executor = LocalPool::new();
     /// let mut spawner = executor.spawner();
     ///
-    /// let future = future::ready(1);
+    /// let future = async { 1 };
     /// let join_handle_fut = spawner.spawn_local_with_handle(future).unwrap();
     /// assert_eq!(executor.run_until(join_handle_fut), 1);
     /// ```
diff --git a/futures-util/src/task/waker.rs b/futures-util/src/task/waker.rs
index 5fb2f921..1ea2f2ef 100644
--- a/futures-util/src/task/waker.rs
+++ b/futures-util/src/task/waker.rs
@@ -3,6 +3,15 @@ use core::mem;
 use core::task::{Waker, RawWaker, RawWakerVTable};
 use alloc::sync::Arc;
 
+pub(super) fn waker_vtable<W: ArcWake>() -> &'static RawWakerVTable {
+    &RawWakerVTable::new(
+        clone_arc_raw::<W>,
+        wake_arc_raw::<W>,
+        wake_by_ref_arc_raw::<W>,
+        drop_arc_raw::<W>,
+    )
+}
+
 /// Creates a [`Waker`] from an `Arc<impl ArcWake>`.
 ///
 /// The returned [`Waker`] will call
@@ -14,7 +23,7 @@ where
     let ptr = Arc::into_raw(wake) as *const ();
 
     unsafe {
-        Waker::from_raw(RawWaker::new(ptr, waker_vtable!(W)))
+        Waker::from_raw(RawWaker::new(ptr, waker_vtable::<W>()))
     }
 }
 
@@ -22,18 +31,16 @@ where
 // code here. We should guard against this by aborting.
 
 unsafe fn increase_refcount<T: ArcWake>(data: *const ()) {
-    // Retain Arc by creating a copy
-    let arc: Arc<T> = Arc::from_raw(data as *const T);
-    let arc_clone = arc.clone();
-    // Forget the Arcs again, so that the refcount isn't decrased
-    mem::forget(arc);
-    mem::forget(arc_clone);
+    // Retain Arc, but don't touch refcount by wrapping in ManuallyDrop
+    let arc = mem::ManuallyDrop::new(Arc::<T>::from_raw(data as *const T));
+    // Now increase refcount, but don't drop new refcount either
+    let _arc_clone: mem::ManuallyDrop<_> = arc.clone();
 }
 
 // used by `waker_ref`
-pub(super) unsafe fn clone_arc_raw<T: ArcWake>(data: *const ()) -> RawWaker {
+unsafe fn clone_arc_raw<T: ArcWake>(data: *const ()) -> RawWaker {
     increase_refcount::<T>(data);
-    RawWaker::new(data, waker_vtable!(T))
+    RawWaker::new(data, waker_vtable::<T>())
 }
 
 unsafe fn wake_arc_raw<T: ArcWake>(data: *const ()) {
@@ -42,10 +49,10 @@ unsafe fn wake_arc_raw<T: ArcWake>(data: *const ()) {
 }
 
 // used by `waker_ref`
-pub(super) unsafe fn wake_by_ref_arc_raw<T: ArcWake>(data: *const ()) {
-    let arc: Arc<T> = Arc::from_raw(data as *const T);
+unsafe fn wake_by_ref_arc_raw<T: ArcWake>(data: *const ()) {
+    // Retain Arc, but don't touch refcount by wrapping in ManuallyDrop
+    let arc = mem::ManuallyDrop::new(Arc::<T>::from_raw(data as *const T));
     ArcWake::wake_by_ref(&arc);
-    mem::forget(arc);
 }
 
 unsafe fn drop_arc_raw<T: ArcWake>(data: *const ()) {
diff --git a/futures-util/src/task/waker_ref.rs b/futures-util/src/task/waker_ref.rs
index 9e83742f..8dda499a 100644
--- a/futures-util/src/task/waker_ref.rs
+++ b/futures-util/src/task/waker_ref.rs
@@ -1,9 +1,10 @@
-use super::arc_wake::ArcWake;
-use super::waker::{clone_arc_raw, wake_by_ref_arc_raw};
+use super::arc_wake::{ArcWake};
+use super::waker::waker_vtable;
 use alloc::sync::Arc;
+use core::mem::ManuallyDrop;
 use core::marker::PhantomData;
 use core::ops::Deref;
-use core::task::{Waker, RawWaker, RawWakerVTable};
+use core::task::{Waker, RawWaker};
 
 /// A [`Waker`] that is only valid for a given lifetime.
 ///
@@ -11,17 +12,29 @@ use core::task::{Waker, RawWaker, RawWakerVTable};
 /// so it can be used to get a `&Waker`.
 #[derive(Debug)]
 pub struct WakerRef<'a> {
-    waker: Waker,
+    waker: ManuallyDrop<Waker>,
     _marker: PhantomData<&'a ()>,
 }
 
-impl WakerRef<'_> {
-    /// Create a new [`WakerRef`] from a [`Waker`].
+impl<'a> WakerRef<'a> {
+    /// Create a new [`WakerRef`] from a [`Waker`] reference.
+    pub fn new(waker: &'a Waker) -> Self {
+        // copy the underlying (raw) waker without calling a clone,
+        // as we won't call Waker::drop either.
+        let waker = ManuallyDrop::new(unsafe { core::ptr::read(waker) });
+        WakerRef {
+            waker,
+            _marker: PhantomData,
+        }
+    }
+
+    /// Create a new [`WakerRef`] from a [`Waker`] that must not be dropped.
     ///
-    /// Note: this function is safe, but it is generally only used
-    /// from `unsafe` contexts that need to create a `Waker`
-    /// that is guaranteed not to outlive a particular lifetime.
-    pub fn new(waker: Waker) -> Self {
+    /// Note: this if for rare cases where the caller created a [`Waker`] in
+    /// an unsafe way (that will be valid only for a lifetime to be determined
+    /// by the caller), and the [`Waker`] doesn't need to or must not be
+    /// destroyed.
+    pub fn new_unowned(waker: ManuallyDrop<Waker>) -> Self {
         WakerRef {
             waker,
             _marker: PhantomData,
@@ -37,21 +50,6 @@ impl Deref for WakerRef<'_> {
     }
 }
 
-#[inline]
-unsafe fn noop(_data: *const ()) {}
-
-unsafe fn wake_unreachable(_data: *const ()) {
-    // With only a reference, calling `wake_arc_raw()` would be unsound,
-    // since the `WakerRef` didn't increment the refcount of the `ArcWake`,
-    // and `wake_arc_raw` would *decrement* it.
-    //
-    // This should never be reachable, since `WakerRef` only provides a `Deref`
-    // to the inner `Waker`.
-    //
-    // Still, safer to panic here than to call `wake_arc_raw`.
-    unreachable!("WakerRef::wake");
-}
-
 /// Creates a reference to a [`Waker`] from a reference to `Arc<impl ArcWake>`.
 ///
 /// The resulting [`Waker`] will call
@@ -61,21 +59,12 @@ pub fn waker_ref<W>(wake: &Arc<W>) -> WakerRef<'_>
 where
     W: ArcWake
 {
-    // This uses the same mechanism as Arc::into_raw, without needing a reference.
-    // This is potentially not stable
-    let ptr = &*wake as &W as *const W as *const ();
-
-    // Similar to `waker_vtable`, but with a no-op `drop` function.
-    // Clones of the resulting `RawWaker` will still be dropped normally.
-    let vtable = &RawWakerVTable::new(
-        clone_arc_raw::<W>,
-        wake_unreachable,
-        wake_by_ref_arc_raw::<W>,
-        noop,
-    );
+    // simply copy the pointer instead of using Arc::into_raw,
+    // as we don't actually keep a refcount by using ManuallyDrop.<
+    let ptr = (&**wake as *const W) as *const ();
 
-    let waker = unsafe {
-        Waker::from_raw(RawWaker::new(ptr, vtable))
-    };
-    WakerRef::new(waker)
+    let waker = ManuallyDrop::new(unsafe {
+        Waker::from_raw(RawWaker::new(ptr, waker_vtable::<W>()))
+    });
+    WakerRef::new_unowned(waker)
 }
diff --git a/futures-util/src/try_future/mod.rs b/futures-util/src/try_future/mod.rs
index 2136ad6a..82bba476 100644
--- a/futures-util/src/try_future/mod.rs
+++ b/futures-util/src/try_future/mod.rs
@@ -4,7 +4,6 @@
 //! including the `FutureExt` trait which adds methods to `Future` types.
 
 use core::pin::Pin;
-use futures_core::future::TryFuture;
 use futures_core::stream::TryStream;
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
@@ -12,6 +11,8 @@ use futures_sink::Sink;
 
 #[cfg(feature = "compat")] use crate::compat::Compat;
 
+pub use futures_core::future::TryFuture;
+
 mod try_join;
 pub use self::try_join::{
     try_join, try_join3, try_join4, try_join5,
@@ -135,11 +136,10 @@ pub trait TryFutureExt: TryFuture {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     ///
     /// # futures::executor::block_on(async {
-    /// let future = future::ready(Ok::<i32, i32>(1));
+    /// let future = async { Ok::<i32, i32>(1) };
     /// let future = future.map_ok(|x| x + 3);
     /// assert_eq!(future.await, Ok(4));
     /// # });
@@ -149,11 +149,10 @@ pub trait TryFutureExt: TryFuture {
     /// effect:
     ///
     /// ```
-    /// #![feature(async_await)]
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     ///
     /// # futures::executor::block_on(async {
-    /// let future = future::ready(Err::<i32, i32>(1));
+    /// let future = async { Err::<i32, i32>(1) };
     /// let future = future.map_ok(|x| x + 3);
     /// assert_eq!(future.await, Err(1));
     /// # });
@@ -183,11 +182,10 @@ pub trait TryFutureExt: TryFuture {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     ///
     /// # futures::executor::block_on(async {
-    /// let future = future::ready(Err::<i32, i32>(1));
+    /// let future = async { Err::<i32, i32>(1) };
     /// let future = future.map_err(|x| x + 3);
     /// assert_eq!(future.await, Err(4));
     /// # });
@@ -197,11 +195,10 @@ pub trait TryFutureExt: TryFuture {
     /// no effect:
     ///
     /// ```
-    /// #![feature(async_await)]
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     ///
     /// # futures::executor::block_on(async {
-    /// let future = future::ready(Ok::<i32, i32>(1));
+    /// let future = async { Ok::<i32, i32>(1) };
     /// let future = future.map_err(|x| x + 3);
     /// assert_eq!(future.await, Ok(1));
     /// # });
@@ -228,11 +225,10 @@ pub trait TryFutureExt: TryFuture {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     ///
     /// # futures::executor::block_on(async {
-    /// let future_err_u8 = future::ready(Err::<(), u8>(1));
+    /// let future_err_u8 = async { Err::<(), u8>(1) };
     /// let future_err_i32 = future_err_u8.err_into::<i32>();
     /// # });
     /// ```
@@ -258,12 +254,11 @@ pub trait TryFutureExt: TryFuture {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     ///
     /// # futures::executor::block_on(async {
-    /// let future = future::ready(Ok::<i32, i32>(1));
-    /// let future = future.and_then(|x| future::ready(Ok::<i32, i32>(x + 3)));
+    /// let future = async { Ok::<i32, i32>(1) };
+    /// let future = future.and_then(|x| async move { Ok::<i32, i32>(x + 3) });
     /// assert_eq!(future.await, Ok(4));
     /// # });
     /// ```
@@ -272,12 +267,11 @@ pub trait TryFutureExt: TryFuture {
     /// effect:
     ///
     /// ```
-    /// #![feature(async_await)]
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     ///
     /// # futures::executor::block_on(async {
-    /// let future = future::ready(Err::<i32, i32>(1));
-    /// let future = future.and_then(|x| future::ready(Err::<i32, i32>(x + 3)));
+    /// let future = async { Err::<i32, i32>(1) };
+    /// let future = future.and_then(|x| async move { Err::<i32, i32>(x + 3) });
     /// assert_eq!(future.await, Err(1));
     /// # });
     /// ```
@@ -304,12 +298,11 @@ pub trait TryFutureExt: TryFuture {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     ///
     /// # futures::executor::block_on(async {
-    /// let future = future::ready(Err::<i32, i32>(1));
-    /// let future = future.or_else(|x| future::ready(Err::<i32, i32>(x + 3)));
+    /// let future = async { Err::<i32, i32>(1) };
+    /// let future = future.or_else(|x| async move { Err::<i32, i32>(x + 3) });
     /// assert_eq!(future.await, Err(4));
     /// # });
     /// ```
@@ -318,12 +311,11 @@ pub trait TryFutureExt: TryFuture {
     /// no effect:
     ///
     /// ```
-    /// #![feature(async_await)]
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     ///
     /// # futures::executor::block_on(async {
-    /// let future = future::ready(Ok::<i32, i32>(1));
-    /// let future = future.or_else(|x| future::ready(Ok::<i32, i32>(x + 3)));
+    /// let future = async { Ok::<i32, i32>(1) };
+    /// let future = future.or_else(|x| async move { Ok::<i32, i32>(x + 3) });
     /// assert_eq!(future.await, Ok(1));
     /// # });
     /// ```
@@ -345,11 +337,10 @@ pub trait TryFutureExt: TryFuture {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     ///
-    /// let future = future::ok::<_, ()>(1);
+    /// let future = async { Ok::<_, ()>(1) };
     /// let new_future = future.inspect_ok(|&x| println!("about to resolve: {}", x));
     /// assert_eq!(new_future.await, Ok(1));
     /// # });
@@ -371,11 +362,10 @@ pub trait TryFutureExt: TryFuture {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     ///
-    /// let future = future::err::<(), _>(1);
+    /// let future = async { Err::<(), _>(1) };
     /// let new_future = future.inspect_err(|&x| println!("about to error: {}", x));
     /// assert_eq!(new_future.await, Err(1));
     /// # });
@@ -400,13 +390,12 @@ pub trait TryFutureExt: TryFuture {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     /// use futures::stream::{self, TryStreamExt};
     ///
     /// let stream_items = vec![17, 18, 19].into_iter().map(Ok);
-    /// let future_of_a_stream = future::ok::<_, ()>(stream::iter(stream_items));
+    /// let future_of_a_stream = async { Ok::<_, ()>(stream::iter(stream_items)) };
     ///
     /// let stream = future_of_a_stream.try_flatten_stream();
     /// let list = stream.try_collect::<Vec<_>>().await;
@@ -434,11 +423,10 @@ pub trait TryFutureExt: TryFuture {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
-    /// use futures::future::{self, TryFutureExt};
+    /// use futures::future::TryFutureExt;
     ///
     /// # futures::executor::block_on(async {
-    /// let future = future::ready(Err::<(), &str>("Boom!"));
+    /// let future = async { Err::<(), &str>("Boom!") };
     /// let future = future.unwrap_or_else(|_| ());
     /// assert_eq!(future.await, ());
     /// # });
@@ -474,7 +462,7 @@ pub trait TryFutureExt: TryFuture {
     /// # type T = i32;
     /// # type E = ();
     /// fn make_try_future() -> impl TryFuture<Ok = T, Error = E> { // ... }
-    /// # futures::future::ready(Ok::<i32, ()>(1))
+    /// # async { Ok::<i32, ()>(1) }
     /// # }
     /// fn take_future(future: impl Future<Output = Result<T, E>>) { /* ... */ }
     ///
diff --git a/futures-util/src/try_future/try_join.rs b/futures-util/src/try_future/try_join.rs
index 24b9065f..d3c17261 100644
--- a/futures-util/src/try_future/try_join.rs
+++ b/futures-util/src/try_future/try_join.rs
@@ -133,7 +133,6 @@ generate! {
 /// [`Ok`] of a tuple of the values:
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
@@ -149,7 +148,6 @@ generate! {
 /// that error:
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
@@ -173,7 +171,6 @@ where
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
@@ -203,7 +200,6 @@ where
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
@@ -236,7 +232,6 @@ where
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future;
 ///
diff --git a/futures-util/src/try_future/try_join_all.rs b/futures-util/src/try_future/try_join_all.rs
index ccff4b78..30300e4e 100644
--- a/futures-util/src/try_future/try_join_all.rs
+++ b/futures-util/src/try_future/try_join_all.rs
@@ -100,7 +100,6 @@ where
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// # futures::executor::block_on(async {
 /// use futures::future::{self, try_join_all};
 ///
diff --git a/futures-util/src/try_stream/and_then.rs b/futures-util/src/try_stream/and_then.rs
index c500028b..809c32a9 100644
--- a/futures-util/src/try_stream/and_then.rs
+++ b/futures-util/src/try_stream/and_then.rs
@@ -1,7 +1,7 @@
 use core::fmt;
 use core::pin::Pin;
 use futures_core::future::TryFuture;
-use futures_core::stream::{Stream, TryStream};
+use futures_core::stream::{Stream, TryStream, FusedStream};
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
@@ -102,6 +102,27 @@ impl<St, Fut, F> Stream for AndThen<St, Fut, F>
         self.as_mut().future().set(None);
         Poll::Ready(Some(e))
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let future_len = if self.future.is_some() { 1 } else { 0 };
+        let (lower, upper) = self.stream.size_hint();
+        let lower = lower.saturating_add(future_len);
+        let upper = match upper {
+            Some(x) => x.checked_add(future_len),
+            None => None,
+        };
+        (lower, upper)
+    }
+}
+
+impl<St, Fut, F> FusedStream for AndThen<St, Fut, F>
+    where St: TryStream + FusedStream,
+          F: FnMut(St::Ok) -> Fut,
+          Fut: TryFuture<Error = St::Error>,
+{
+    fn is_terminated(&self) -> bool {
+        self.future.is_none() && self.stream.is_terminated()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/try_stream/err_into.rs b/futures-util/src/try_stream/err_into.rs
index 462c31a2..f5d92945 100644
--- a/futures-util/src/try_stream/err_into.rs
+++ b/futures-util/src/try_stream/err_into.rs
@@ -80,6 +80,10 @@ where
         self.stream().try_poll_next(cx)
             .map(|res| res.map(|some| some.map_err(Into::into)))
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/try_stream/inspect_err.rs b/futures-util/src/try_stream/inspect_err.rs
index 588e660d..36c21f05 100644
--- a/futures-util/src/try_stream/inspect_err.rs
+++ b/futures-util/src/try_stream/inspect_err.rs
@@ -100,6 +100,10 @@ where
             .try_poll_next(cx)
             .map(|opt| opt.map(|res| res.map_err(|e| inspect(e, self.as_mut().f()))))
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/try_stream/inspect_ok.rs b/futures-util/src/try_stream/inspect_ok.rs
index 1674d91b..29bb9544 100644
--- a/futures-util/src/try_stream/inspect_ok.rs
+++ b/futures-util/src/try_stream/inspect_ok.rs
@@ -100,6 +100,10 @@ where
             .try_poll_next(cx)
             .map(|opt| opt.map(|res| res.map(|e| inspect(e, self.as_mut().f()))))
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/try_stream/into_async_read.rs b/futures-util/src/try_stream/into_async_read.rs
index 374299a5..e053b1b7 100644
--- a/futures-util/src/try_stream/into_async_read.rs
+++ b/futures-util/src/try_stream/into_async_read.rs
@@ -6,9 +6,9 @@ use futures_io::{AsyncRead, AsyncWrite, AsyncBufRead};
 use std::cmp;
 use std::io::{Error, Result};
 
-/// An `AsyncRead` for the [`into_async_read`](super::TryStreamExt::into_async_read) combinator.
+/// Reader for the [`into_async_read`](super::TryStreamExt::into_async_read) method.
 #[derive(Debug)]
-#[must_use = "streams do nothing unless polled"]
+#[must_use = "readers do nothing unless polled"]
 pub struct IntoAsyncRead<St>
 where
     St: TryStream<Error = Error> + Unpin,
diff --git a/futures-util/src/try_stream/into_stream.rs b/futures-util/src/try_stream/into_stream.rs
index f2656174..b0fa07aa 100644
--- a/futures-util/src/try_stream/into_stream.rs
+++ b/futures-util/src/try_stream/into_stream.rs
@@ -69,6 +69,10 @@ impl<St: TryStream> Stream for IntoStream<St> {
     ) -> Poll<Option<Self::Item>> {
         self.stream().try_poll_next(cx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/try_stream/map_err.rs b/futures-util/src/try_stream/map_err.rs
index 68d5d71a..1b98d6b4 100644
--- a/futures-util/src/try_stream/map_err.rs
+++ b/futures-util/src/try_stream/map_err.rs
@@ -94,6 +94,10 @@ where
             .try_poll_next(cx)
             .map(|opt| opt.map(|res| res.map_err(|e| self.as_mut().f()(e))))
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/try_stream/map_ok.rs b/futures-util/src/try_stream/map_ok.rs
index 1738c6e7..19d01be4 100644
--- a/futures-util/src/try_stream/map_ok.rs
+++ b/futures-util/src/try_stream/map_ok.rs
@@ -94,6 +94,10 @@ where
             .try_poll_next(cx)
             .map(|opt| opt.map(|res| res.map(|x| self.as_mut().f()(x))))
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/try_stream/mod.rs b/futures-util/src/try_stream/mod.rs
index e791613e..d1f46117 100644
--- a/futures-util/src/try_stream/mod.rs
+++ b/futures-util/src/try_stream/mod.rs
@@ -5,12 +5,13 @@
 
 use core::pin::Pin;
 use futures_core::future::{Future, TryFuture};
-use futures_core::stream::TryStream;
 use futures_core::task::{Context, Poll};
 
 #[cfg(feature = "compat")]
 use crate::compat::Compat;
 
+pub use futures_core::stream::TryStream;
+
 mod and_then;
 pub use self::and_then::AndThen;
 
@@ -91,7 +92,6 @@ pub trait TryStreamExt: TryStream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, TryStreamExt};
     ///
@@ -117,7 +117,6 @@ pub trait TryStreamExt: TryStream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, TryStreamExt};
     ///
@@ -143,7 +142,6 @@ pub trait TryStreamExt: TryStream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, TryStreamExt};
     ///
@@ -297,7 +295,6 @@ pub trait TryStreamExt: TryStream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, TryStreamExt};
     ///
@@ -329,7 +326,6 @@ pub trait TryStreamExt: TryStream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::future;
     /// use futures::stream::{self, TryStreamExt};
@@ -364,7 +360,6 @@ pub trait TryStreamExt: TryStream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::future;
     /// use futures::stream::{self, TryStreamExt};
@@ -399,7 +394,6 @@ pub trait TryStreamExt: TryStream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::oneshot;
     /// use futures::stream::{self, StreamExt, TryStreamExt};
@@ -455,7 +449,6 @@ pub trait TryStreamExt: TryStream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::mpsc;
     /// use futures::stream::TryStreamExt;
@@ -497,7 +490,6 @@ pub trait TryStreamExt: TryStream {
     ///
     /// # Examples
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::future;
     /// use futures::stream::{self, StreamExt, TryStreamExt};
@@ -536,17 +528,17 @@ pub trait TryStreamExt: TryStream {
     ///
     /// # Examples
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future;
     /// use futures::stream::{self, StreamExt, TryStreamExt};
+    /// use futures::pin_mut;
     ///
     /// let stream = stream::iter(vec![Ok(1i32), Ok(6i32), Err("error")]);
-    /// let mut halves = stream.try_filter_map(|x| {
+    /// let halves = stream.try_filter_map(|x| async move {
     ///     let ret = if x % 2 == 0 { Some(x / 2) } else { None };
-    ///     future::ready(Ok(ret))
+    ///     Ok(ret)
     /// });
     ///
+    /// pin_mut!(halves);
     /// assert_eq!(halves.next().await, Some(Ok(3)));
     /// assert_eq!(halves.next().await, Some(Err("error")));
     /// # })
@@ -569,7 +561,6 @@ pub trait TryStreamExt: TryStream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::mpsc;
     /// use futures::stream::{StreamExt, TryStreamExt};
@@ -622,17 +613,15 @@ pub trait TryStreamExt: TryStream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
-    /// use futures::future;
     /// use futures::stream::{self, TryStreamExt};
     ///
     /// let number_stream = stream::iter(vec![Ok::<i32, i32>(1), Ok(2)]);
-    /// let sum = number_stream.try_fold(0, |acc, x| future::ready(Ok(acc + x)));
+    /// let sum = number_stream.try_fold(0, |acc, x| async move { Ok(acc + x) });
     /// assert_eq!(sum.await, Ok(3));
     ///
     /// let number_stream_with_err = stream::iter(vec![Ok::<i32, i32>(1), Err(2), Ok(1)]);
-    /// let sum = number_stream_with_err.try_fold(0, |acc, x| future::ready(Ok(acc + x)));
+    /// let sum = number_stream_with_err.try_fold(0, |acc, x| async move { Ok(acc + x) });
     /// assert_eq!(sum.await, Err(2));
     /// # })
     /// ```
@@ -659,7 +648,6 @@ pub trait TryStreamExt: TryStream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::mpsc;
     /// use futures::stream::TryStreamExt;
@@ -707,7 +695,6 @@ pub trait TryStreamExt: TryStream {
     ///
     /// Results are returned in the order of completion:
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::oneshot;
     /// use futures::stream::{self, StreamExt, TryStreamExt};
@@ -731,16 +718,14 @@ pub trait TryStreamExt: TryStream {
     ///
     /// Errors from the underlying stream itself are propagated:
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::mpsc;
-    /// use futures::future;
     /// use futures::stream::{StreamExt, TryStreamExt};
     ///
     /// let (sink, stream_of_futures) = mpsc::unbounded();
     /// let mut buffered = stream_of_futures.try_buffer_unordered(10);
     ///
-    /// sink.unbounded_send(Ok(future::ready(Ok(7i32))))?;
+    /// sink.unbounded_send(Ok(async { Ok(7i32) }))?;
     /// assert_eq!(buffered.next().await, Some(Ok(7i32)));
     ///
     /// sink.unbounded_send(Err("error in the stream"))?;
@@ -775,7 +760,6 @@ pub trait TryStreamExt: TryStream {
     /// Wraps a [`TryStream`] into a stream compatible with libraries using
     /// futures 0.1 `Stream`. Requires the `compat` feature to be enabled.
     /// ```
-    /// #![feature(async_await)]
     /// use futures::future::{FutureExt, TryFutureExt};
     /// # let (tx, rx) = futures::channel::oneshot::channel();
     ///
@@ -814,7 +798,6 @@ pub trait TryStreamExt: TryStream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::stream::{self, TryStreamExt};
     /// use futures::io::AsyncReadExt;
diff --git a/futures-util/src/try_stream/or_else.rs b/futures-util/src/try_stream/or_else.rs
index 0fe25657..33310d1c 100644
--- a/futures-util/src/try_stream/or_else.rs
+++ b/futures-util/src/try_stream/or_else.rs
@@ -1,7 +1,7 @@
 use core::fmt;
 use core::pin::Pin;
 use futures_core::future::TryFuture;
-use futures_core::stream::{Stream, TryStream};
+use futures_core::stream::{Stream, TryStream, FusedStream};
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
@@ -103,6 +103,27 @@ impl<St, Fut, F> Stream for OrElse<St, Fut, F>
         self.as_mut().future().set(None);
         Poll::Ready(Some(e))
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let future_len = if self.future.is_some() { 1 } else { 0 };
+        let (lower, upper) = self.stream.size_hint();
+        let lower = lower.saturating_add(future_len);
+        let upper = match upper {
+            Some(x) => x.checked_add(future_len),
+            None => None,
+        };
+        (lower, upper)
+    }
+}
+
+impl<St, Fut, F> FusedStream for OrElse<St, Fut, F>
+    where St: TryStream + FusedStream,
+          F: FnMut(St::Error) -> Fut,
+          Fut: TryFuture<Ok = St::Ok>,
+{
+    fn is_terminated(&self) -> bool {
+        self.future.is_none() && self.stream.is_terminated()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/try_stream/try_filter.rs b/futures-util/src/try_stream/try_filter.rs
index f5035c13..24a9c327 100644
--- a/futures-util/src/try_stream/try_filter.rs
+++ b/futures-util/src/try_stream/try_filter.rs
@@ -129,6 +129,16 @@ impl<St, Fut, F> Stream for TryFilter<St, Fut, F>
             }
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let pending_len = if self.pending_fut.is_some() { 1 } else { 0 };
+        let (_, upper) = self.stream.size_hint();
+        let upper = match upper {
+            Some(x) => x.checked_add(pending_len),
+            None => None,
+        };
+        (0, upper) // can't know a lower bound, due to the predicate
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/try_stream/try_filter_map.rs b/futures-util/src/try_stream/try_filter_map.rs
index 08b7aa7c..ed7eeb22 100644
--- a/futures-util/src/try_stream/try_filter_map.rs
+++ b/futures-util/src/try_stream/try_filter_map.rs
@@ -113,6 +113,16 @@ impl<St, Fut, F, T> Stream for TryFilterMap<St, Fut, F>
             }
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let pending_len = if self.pending.is_some() { 1 } else { 0 };
+        let (_, upper) = self.stream.size_hint();
+        let upper = match upper {
+            Some(x) => x.checked_add(pending_len),
+            None => None,
+        };
+        (0, upper) // can't know a lower bound, due to the predicate
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures-util/src/try_stream/try_skip_while.rs b/futures-util/src/try_stream/try_skip_while.rs
index 6d446fcc..a3d6803a 100644
--- a/futures-util/src/try_stream/try_skip_while.rs
+++ b/futures-util/src/try_stream/try_skip_while.rs
@@ -1,7 +1,7 @@
 use core::fmt;
 use core::pin::Pin;
 use futures_core::future::TryFuture;
-use futures_core::stream::{Stream, TryStream};
+use futures_core::stream::{Stream, TryStream, FusedStream};
 use futures_core::task::{Context, Poll};
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
@@ -131,6 +131,26 @@ impl<St, Fut, F> Stream for TrySkipWhile<St, Fut, F>
             }
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let pending_len = if self.pending_item.is_some() { 1 } else { 0 };
+        let (_, upper) = self.stream.size_hint();
+        let upper = match upper {
+            Some(x) => x.checked_add(pending_len),
+            None => None,
+        };
+        (0, upper) // can't know a lower bound, due to the predicate
+    }
+}
+
+impl<St, Fut, F> FusedStream for TrySkipWhile<St, Fut, F>
+    where St: TryStream + FusedStream,
+          F: FnMut(&St::Ok) -> Fut,
+          Fut: TryFuture<Ok = bool, Error = St::Error>,
+{
+    fn is_terminated(&self) -> bool {
+        self.pending_item.is_none() && self.stream.is_terminated()
+    }
 }
 
 // Forwarding impl of Sink from the underlying stream
diff --git a/futures/Cargo.toml b/futures/Cargo.toml
index 14fe4fdf..71e99ea6 100644
--- a/futures/Cargo.toml
+++ b/futures/Cargo.toml
@@ -39,11 +39,17 @@ assert_matches = "1.3.0"
 default = ["std"]
 std = ["alloc", "futures-core-preview/std", "futures-executor-preview/std", "futures-io-preview/std", "futures-sink-preview/std", "futures-util-preview/std", "futures-util-preview/io", "futures-util-preview/channel"]
 alloc = ["futures-core-preview/alloc", "futures-sink-preview/alloc", "futures-channel-preview/alloc", "futures-util-preview/alloc"]
-nightly = ["futures-core-preview/nightly", "futures-channel-preview/nightly", "futures-util-preview/nightly"]
 async-await = ["futures-util-preview/async-await", "futures-util-preview/join-macro", "futures-util-preview/select-macro"]
 compat = ["std", "futures-util-preview/compat"]
 io-compat = ["compat", "futures-util-preview/io-compat"]
+
+# Unstable features
+# These features are outside of the normal semver guarantees and require the
+# `unstable` feature as an explicit opt-in to unstable API.
+unstable = ["futures-core-preview/unstable", "futures-channel-preview/unstable", "futures-io-preview/unstable", "futures-util-preview/unstable"]
 cfg-target-has-atomic = ["futures-core-preview/cfg-target-has-atomic", "futures-channel-preview/cfg-target-has-atomic", "futures-util-preview/cfg-target-has-atomic"]
+bilock = ["futures-util-preview/bilock"]
+read_initializer = ["futures-io-preview/read_initializer", "futures-util-preview/read_initializer"]
 
 [package.metadata.docs.rs]
 all-features = true
diff --git a/futures/src/lib.rs b/futures/src/lib.rs
index 9a981591..46158ca8 100644
--- a/futures/src/lib.rs
+++ b/futures/src/lib.rs
@@ -22,6 +22,7 @@
 //! completion, but *do not block* the thread running them.
 
 #![cfg_attr(feature = "cfg-target-has-atomic", feature(cfg_target_has_atomic))]
+#![cfg_attr(feature = "read_initializer", feature(read_initializer))]
 
 #![cfg_attr(not(feature = "std"), no_std)]
 
@@ -34,11 +35,14 @@
 
 #![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures")]
 
-#[cfg(all(feature = "async-await", not(feature = "nightly")))]
-compile_error!("The `async-await` feature requires the `nightly` feature as an explicit opt-in to unstable features");
+#[cfg(all(feature = "cfg-target-has-atomic", not(feature = "unstable")))]
+compile_error!("The `cfg-target-has-atomic` feature requires the `unstable` feature as an explicit opt-in to unstable features");
 
-#[cfg(all(feature = "cfg-target-has-atomic", not(feature = "nightly")))]
-compile_error!("The `cfg-target-has-atomic` feature requires the `nightly` feature as an explicit opt-in to unstable features");
+#[cfg(all(feature = "bilock", not(feature = "unstable")))]
+compile_error!("The `bilock` feature requires the `unstable` feature as an explicit opt-in to unstable features");
+
+#[cfg(all(feature = "read_initializer", not(feature = "unstable")))]
+compile_error!("The `read_initializer` feature requires the `unstable` feature as an explicit opt-in to unstable features");
 
 #[doc(hidden)] pub use futures_core::core_reexport;
 
@@ -62,11 +66,12 @@ compile_error!("The `cfg-target-has-atomic` feature requires the `nightly` featu
 
 #[doc(hidden)] pub use futures_core::task::Poll;
 
-#[doc(hidden)] pub use futures_core::never::Never;
+#[doc(hidden)] pub use futures_util::never::Never;
 
 // Macro reexports
 pub use futures_core::ready; // Readiness propagation
 pub use futures_util::pin_mut;
+#[cfg(feature = "std")]
 #[cfg(feature = "async-await")]
 pub use futures_util::{pending, poll}; // Async-await
 
@@ -273,7 +278,7 @@ pub mod future {
 pub mod io {
     //! Asynchronous I/O.
     //!
-    //! This module is the asynchronous version of `std::io`. It defines two
+    //! This module is the asynchronous version of `std::io`. It defines four
     //! traits, [`AsyncRead`](crate::io::AsyncRead),
     //! [`AsyncWrite`](crate::io::AsyncWrite),
     //! [`AsyncSeek`](crate::io::AsyncSeek), and
@@ -297,25 +302,36 @@ pub mod io {
 
     pub use futures_io::{
         AsyncRead, AsyncWrite, AsyncSeek, AsyncBufRead, Error, ErrorKind,
-        Initializer, IoSlice, IoSliceMut, Result, SeekFrom,
+        IoSlice, IoSliceMut, Result, SeekFrom,
     };
 
+    #[cfg(feature = "read_initializer")]
+    pub use futures_io::Initializer;
+
     pub use futures_util::io::{
         AsyncReadExt, AsyncWriteExt, AsyncSeekExt, AsyncBufReadExt, AllowStdIo,
-        BufReader, BufWriter, Close, CopyInto, CopyBufInto, Flush, IntoSink,
-        Lines, Read, ReadExact, ReadHalf, ReadLine, ReadToEnd, ReadToString,
-        ReadUntil, ReadVectored, Seek, Window, Write, WriteAll, WriteHalf,
-        WriteVectored,
+        BufReader, BufWriter, Chain, Close, CopyInto, CopyBufInto, empty, Empty,
+        Flush, IntoSink, Lines, Read, ReadExact, ReadHalf, ReadLine, ReadToEnd,
+        ReadToString, ReadUntil, ReadVectored, repeat, Repeat, Seek, sink, Sink,
+        Take, Window, Write, WriteAll, WriteHalf, WriteVectored,
     };
 }
 
-#[cfg(feature = "std")]
+#[cfg_attr(
+    feature = "cfg-target-has-atomic",
+    cfg(all(target_has_atomic = "cas", target_has_atomic = "ptr"))
+)]
+#[cfg(feature = "alloc")]
 pub mod lock {
     //! Futures-powered synchronization primitives.
     //!
-    //! This module is only available when the `std` feature of this
+    //! This module is only available when the `std` or `alloc` feature of this
     //! library is activated, and it is activated by default.
 
+    #[cfg(feature = "bilock")]
+    pub use futures_util::lock::{BiLock, BiLockAcquire, BiLockGuard, ReuniteError};
+
+    #[cfg(feature = "std")]
     pub use futures_util::lock::{Mutex, MutexLockFuture, MutexGuard};
 }
 
@@ -446,8 +462,8 @@ pub mod stream {
         TryStreamExt,
         AndThen, ErrInto, MapOk, MapErr, OrElse,
         InspectOk, InspectErr,
-        TryNext, TryForEach, TryFilterMap, TryFlatten,
-        TryCollect, TryFold, TrySkipWhile,
+        TryNext, TryForEach, TryFilter, TryFilterMap, TryFlatten,
+        TryCollect, TryConcat, TryFold, TrySkipWhile,
         IntoStream,
     };
 
@@ -510,15 +526,17 @@ pub mod never {
     //!
     //! Values of this type can never be created and will never exist.
 
-    pub use futures_core::never::Never;
+    pub use futures_util::never::Never;
 }
 
 // proc-macro re-export --------------------------------------
 
+#[cfg(feature = "std")]
 #[cfg(feature = "async-await")]
 #[doc(hidden)]
-pub use futures_util::rand_reexport;
+pub use futures_util::async_await;
 
+#[cfg(feature = "std")]
 #[cfg(feature = "async-await")]
 #[doc(hidden)]
 pub mod inner_macro {
@@ -527,6 +545,7 @@ pub mod inner_macro {
     pub use futures_util::select;
 }
 
+#[cfg(feature = "std")]
 #[cfg(feature = "async-await")]
 futures_util::document_join_macro! {
     #[macro_export]
@@ -550,6 +569,7 @@ futures_util::document_join_macro! {
     }
 }
 
+#[cfg(feature = "std")]
 #[cfg(feature = "async-await")]
 futures_util::document_select_macro! {
     #[macro_export]


Test patch:
diff --git a/futures-channel/tests/channel.rs b/futures-channel/tests/channel.rs
index e6dad1dc..cea7c127 100644
--- a/futures-channel/tests/channel.rs
+++ b/futures-channel/tests/channel.rs
@@ -1,5 +1,3 @@
-#![feature(async_await)]
-
 use futures::channel::mpsc;
 use futures::executor::block_on;
 use futures::future::poll_fn;
diff --git a/futures-channel/tests/mpsc.rs b/futures-channel/tests/mpsc.rs
index df8d3b9f..ce4c3110 100644
--- a/futures-channel/tests/mpsc.rs
+++ b/futures-channel/tests/mpsc.rs
@@ -1,5 +1,3 @@
-#![feature(async_await)]
-
 use futures::channel::{mpsc, oneshot};
 use futures::executor::{block_on, block_on_stream};
 use futures::future::{FutureExt, poll_fn};
diff --git a/futures-test/src/assert.rs b/futures-test/src/assert.rs
index 3e1288cd..24b4ec8e 100644
--- a/futures-test/src/assert.rs
+++ b/futures-test/src/assert.rs
@@ -9,7 +9,6 @@ pub fn assert_is_unpin_stream<S: Stream + Unpin>(_: &mut S) {}
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// use futures::stream;
 /// use futures_test::future::FutureTestExt;
 /// use futures_test::{
@@ -46,7 +45,6 @@ macro_rules! assert_stream_pending {
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// use futures::stream;
 /// use futures_test::future::FutureTestExt;
 /// use futures_test::{
@@ -89,7 +87,6 @@ macro_rules! assert_stream_next {
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// use futures::stream;
 /// use futures_test::future::FutureTestExt;
 /// use futures_test::{
diff --git a/futures-test/src/future/mod.rs b/futures-test/src/future/mod.rs
index 6eb999ca..0317d349 100644
--- a/futures-test/src/future/mod.rs
+++ b/futures-test/src/future/mod.rs
@@ -35,7 +35,6 @@ pub trait FutureTestExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::task::Poll;
     /// use futures::future::FutureExt;
     /// use futures_test::task::noop_context;
@@ -62,7 +61,6 @@ pub trait FutureTestExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// # futures::executor::block_on(async {
     /// use futures::channel::oneshot;
     /// use futures_test::future::FutureTestExt;
@@ -88,7 +86,6 @@ pub trait FutureTestExt: Future {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::task::Poll;
     /// use futures::future::{self, Future};
     /// use futures_test::task::noop_context;
diff --git a/futures-test/src/interleave_pending.rs b/futures-test/src/interleave_pending.rs
index 302075a6..993bee56 100644
--- a/futures-test/src/interleave_pending.rs
+++ b/futures-test/src/interleave_pending.rs
@@ -104,6 +104,10 @@ impl<St: Stream> Stream for InterleavePending<St> {
             Poll::Pending
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.inner.size_hint()
+    }
 }
 
 impl<W: AsyncWrite> AsyncWrite for InterleavePending<W> {
diff --git a/futures-test/src/io/read/mod.rs b/futures-test/src/io/read/mod.rs
index 178c3574..b24a20e5 100644
--- a/futures-test/src/io/read/mod.rs
+++ b/futures-test/src/io/read/mod.rs
@@ -13,7 +13,6 @@ pub trait AsyncReadTestExt: AsyncRead {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::task::Poll;
     /// use futures::io::AsyncRead;
     /// use futures_test::task::noop_context;
@@ -44,7 +43,6 @@ pub trait AsyncReadTestExt: AsyncRead {
     /// The returned reader will also implement `AsyncBufRead` if the underlying reader does.
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::task::Poll;
     /// use futures::io::AsyncBufRead;
     /// use futures_test::task::noop_context;
@@ -79,7 +77,6 @@ pub trait AsyncReadTestExt: AsyncRead {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::task::Poll;
     /// use futures::io::AsyncRead;
     /// use futures_test::task::noop_context;
diff --git a/futures-test/src/io/write/mod.rs b/futures-test/src/io/write/mod.rs
index 55d512d6..cd167c29 100644
--- a/futures-test/src/io/write/mod.rs
+++ b/futures-test/src/io/write/mod.rs
@@ -13,7 +13,6 @@ pub trait AsyncWriteTestExt: AsyncWrite {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::task::Poll;
     /// use futures::io::AsyncWrite;
     /// use futures_test::task::noop_context;
@@ -54,7 +53,6 @@ pub trait AsyncWriteTestExt: AsyncWrite {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::task::Poll;
     /// use futures::io::AsyncWrite;
     /// use futures_test::task::noop_context;
diff --git a/futures-test/src/stream/mod.rs b/futures-test/src/stream/mod.rs
index a737a82a..5c194e71 100644
--- a/futures-test/src/stream/mod.rs
+++ b/futures-test/src/stream/mod.rs
@@ -12,7 +12,6 @@ pub trait StreamTestExt: Stream {
     /// # Examples
     ///
     /// ```
-    /// #![feature(async_await)]
     /// use futures::task::Poll;
     /// use futures::stream::{self, Stream};
     /// use futures_test::task::noop_context;
diff --git a/futures-test/src/task/context.rs b/futures-test/src/task/context.rs
index c4e356f5..602127c6 100644
--- a/futures-test/src/task/context.rs
+++ b/futures-test/src/task/context.rs
@@ -22,7 +22,6 @@ pub fn panic_context() -> Context<'static> {
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// use futures::future::Future;
 /// use futures::task::Poll;
 /// use futures_test::task::noop_context;
diff --git a/futures-test/src/task/noop_spawner.rs b/futures-test/src/task/noop_spawner.rs
index 05d48018..947fd119 100644
--- a/futures-test/src/task/noop_spawner.rs
+++ b/futures-test/src/task/noop_spawner.rs
@@ -7,7 +7,6 @@ use futures_core::task::{Spawn, SpawnError};
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// use futures::task::SpawnExt;
 /// use futures_test::task::NoopSpawner;
 ///
@@ -46,7 +45,6 @@ impl Default for NoopSpawner {
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// use futures::task::SpawnExt;
 /// use futures_test::task::noop_spawner_mut;
 ///
diff --git a/futures-test/src/task/panic_spawner.rs b/futures-test/src/task/panic_spawner.rs
index 5d525248..4206d44d 100644
--- a/futures-test/src/task/panic_spawner.rs
+++ b/futures-test/src/task/panic_spawner.rs
@@ -7,7 +7,6 @@ use futures_core::task::{Spawn, SpawnError};
 /// # Examples
 ///
 /// ```should_panic
-/// #![feature(async_await)]
 /// use futures::task::SpawnExt;
 /// use futures_test::task::PanicSpawner;
 ///
@@ -47,7 +46,6 @@ impl Default for PanicSpawner {
 /// # Examples
 ///
 /// ```should_panic
-/// #![feature(async_await)]
 /// use futures::task::SpawnExt;
 /// use futures_test::task::panic_spawner_mut;
 ///
diff --git a/futures-test/src/task/panic_waker.rs b/futures-test/src/task/panic_waker.rs
index b7826ec6..74476d82 100644
--- a/futures-test/src/task/panic_waker.rs
+++ b/futures-test/src/task/panic_waker.rs
@@ -46,7 +46,6 @@ pub fn panic_waker() -> Waker {
 /// # Examples
 ///
 /// ```should_panic
-/// #![feature(async_await)]
 /// use futures_test::task::panic_waker_ref;
 ///
 /// let waker = panic_waker_ref();
diff --git a/futures-test/src/task/record_spawner.rs b/futures-test/src/task/record_spawner.rs
index ec54518e..3b13d002 100644
--- a/futures-test/src/task/record_spawner.rs
+++ b/futures-test/src/task/record_spawner.rs
@@ -7,7 +7,6 @@ use futures_core::task::{Spawn, SpawnError};
 /// # Examples
 ///
 /// ```
-/// #![feature(async_await)]
 /// use futures::task::SpawnExt;
 /// use futures_test::task::RecordSpawner;
 ///
diff --git a/futures/tests/arc_wake.rs b/futures/tests/arc_wake.rs
index ff07ab94..1940e4f9 100644
--- a/futures/tests/arc_wake.rs
+++ b/futures/tests/arc_wake.rs
@@ -44,3 +44,34 @@ fn create_waker_from_arc() {
     drop(w1);
     assert_eq!(1, Arc::strong_count(&some_w));
 }
+
+struct PanicWaker;
+
+impl ArcWake for PanicWaker {
+    fn wake_by_ref(_arc_self: &Arc<Self>) {
+        panic!("WAKE UP");
+    }
+}
+
+#[test]
+fn proper_refcount_on_wake_panic() {
+    let some_w = Arc::new(PanicWaker);
+
+    let w1: Waker = task::waker(some_w.clone());
+    assert_eq!("WAKE UP", *std::panic::catch_unwind(|| w1.wake_by_ref()).unwrap_err().downcast::<&str>().unwrap());
+    assert_eq!(2, Arc::strong_count(&some_w)); // some_w + w1
+    drop(w1);
+    assert_eq!(1, Arc::strong_count(&some_w)); // some_w
+}
+
+#[test]
+fn waker_ref_wake_same() {
+    let some_w = Arc::new(CountingWaker::new());
+
+    let w1: Waker = task::waker(some_w.clone());
+    let w2 = task::waker_ref(&some_w);
+    let w3 = w2.clone();
+
+    assert!(w1.will_wake(&w2));
+    assert!(w2.will_wake(&w3));
+}
diff --git a/futures/tests/async_await_macros.rs b/futures/tests/async_await_macros.rs
index e21a382c..de1937ce 100644
--- a/futures/tests/async_await_macros.rs
+++ b/futures/tests/async_await_macros.rs
@@ -1,5 +1,4 @@
 #![recursion_limit="128"]
-#![feature(async_await)]
 
 use futures::{Poll, pending, pin_mut, poll, join, try_join, select};
 use futures::channel::{mpsc, oneshot};
@@ -175,7 +174,7 @@ fn join_size() {
         let ready2 = future::ready(0i32);
         join!(ready1, ready2)
     };
-    assert_eq!(::std::mem::size_of_val(&fut), 32);
+    assert_eq!(::std::mem::size_of_val(&fut), 28);
 }
 
 #[test]
@@ -191,7 +190,7 @@ fn try_join_size() {
         let ready2 = future::ready(Ok::<i32, i32>(0));
         try_join!(ready1, ready2)
     };
-    assert_eq!(::std::mem::size_of_val(&fut), 32);
+    assert_eq!(::std::mem::size_of_val(&fut), 28);
 }
 
 #[test]
diff --git a/futures/tests/compat.rs b/futures/tests/compat.rs
index 89b548b5..39adc7cb 100644
--- a/futures/tests/compat.rs
+++ b/futures/tests/compat.rs
@@ -1,4 +1,3 @@
-#![feature(async_await)]
 #![cfg(feature = "compat")]
 
 use tokio::timer::Delay;
diff --git a/futures/tests/future_obj.rs b/futures/tests/future_obj.rs
index bc1eab88..c6b18fc8 100644
--- a/futures/tests/future_obj.rs
+++ b/futures/tests/future_obj.rs
@@ -1,5 +1,3 @@
-#![feature(async_await)]
-
 use futures::future::{Future, FutureObj, FutureExt};
 use std::pin::Pin;
 use futures::task::{Context, Poll};
diff --git a/futures/tests/join_all.rs b/futures/tests/join_all.rs
index 39f2608f..63967bf9 100644
--- a/futures/tests/join_all.rs
+++ b/futures/tests/join_all.rs
@@ -1,5 +1,3 @@
-#![feature(async_await)]
-
 use futures_util::future::*;
 use std::future::Future;
 use futures::executor::block_on;
diff --git a/futures/tests/macro_comma_support.rs b/futures/tests/macro_comma_support.rs
index 3c5d9b2e..111f65af 100644
--- a/futures/tests/macro_comma_support.rs
+++ b/futures/tests/macro_comma_support.rs
@@ -1,5 +1,3 @@
-#![feature(async_await)]
-
 #[macro_use]
 extern crate futures;
 
@@ -20,7 +18,7 @@ fn ready() {
 #[test]
 fn poll() {
     block_on(async {
-        let _ = poll!(async { () }.boxed(),);
+        let _ = poll!(async {}.boxed(),);
     })
 }
 
diff --git a/futures/tests/mutex.rs b/futures/tests/mutex.rs
index 7b1a7fbf..3c6f02a9 100644
--- a/futures/tests/mutex.rs
+++ b/futures/tests/mutex.rs
@@ -1,5 +1,3 @@
-#![feature(async_await)]
-
 use futures::channel::mpsc;
 use futures::future::{ready, FutureExt};
 use futures::lock::Mutex;
diff --git a/futures/tests/object_safety.rs b/futures/tests/object_safety.rs
index b8d2bb00..30c892f5 100644
--- a/futures/tests/object_safety.rs
+++ b/futures/tests/object_safety.rs
@@ -7,7 +7,7 @@ fn future() {
 
     assert_is_object_safe::<&dyn Future<Output = ()>>();
     assert_is_object_safe::<&dyn FusedFuture<Output = ()>>();
-    assert_is_object_safe::<&dyn TryFuture<Ok = (), Error = ()>>();
+    assert_is_object_safe::<&dyn TryFuture<Ok = (), Error = (), Output = Result<(), ()>>>();
 }
 
 #[test]
@@ -17,7 +17,7 @@ fn stream() {
 
     assert_is_object_safe::<&dyn Stream<Item = ()>>();
     assert_is_object_safe::<&dyn FusedStream<Item = ()>>();
-    assert_is_object_safe::<&dyn TryStream<Ok = (), Error = ()>>();
+    assert_is_object_safe::<&dyn TryStream<Ok = (), Error = (), Item = Result<(), ()>>>();
 }
 
 #[test]
diff --git a/futures/tests/stream.rs b/futures/tests/stream.rs
index ae9cdfa5..3dd375c8 100644
--- a/futures/tests/stream.rs
+++ b/futures/tests/stream.rs
@@ -1,5 +1,3 @@
-#![feature(async_await)]
-
 use futures::executor::block_on;
 use futures::stream;
 use futures_util::StreamExt;
diff --git a/futures/tests/stream_select_next_some.rs b/futures/tests/stream_select_next_some.rs
index b47ab921..09d7e895 100644
--- a/futures/tests/stream_select_next_some.rs
+++ b/futures/tests/stream_select_next_some.rs
@@ -1,5 +1,3 @@
-#![feature(async_await)]
-
 use futures::{future, select};
 use futures::future::{FusedFuture, FutureExt};
 use futures::stream::{FuturesUnordered, StreamExt};
diff --git a/futures/tests/try_join_all.rs b/futures/tests/try_join_all.rs
index bdca3e95..ed45f209 100644
--- a/futures/tests/try_join_all.rs
+++ b/futures/tests/try_join_all.rs
@@ -1,5 +1,3 @@
-#![feature(async_await)]
-
 use futures_util::future::*;
 use futures_util::try_future::{try_join_all, TryJoinAll};
 use std::future::Future;