Repository: rust-lang-nursery/futures-rs
Pull request #1848 (2019-08-31 12:44:17)

Use Arc::ptr_eq in Bilock::reunite
Refs: https://doc.rust-lang.org/nightly/alloc/sync/struct.Arc.html#method.ptr_eq


Patch:
diff --git a/.travis.yml b/.travis.yml
index 5bcf70a5..bb049a06 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -136,14 +136,23 @@ matrix:
             --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
@@ -158,6 +167,7 @@ matrix:
         - 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
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/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/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/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 78ba2c34..9985dca5 100644
--- a/futures-executor/src/local_pool.rs
+++ b/futures-executor/src/local_pool.rs
@@ -329,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-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 ff2aff9f..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,11 +383,11 @@ mod if_std {
     }
 
     impl AsyncRead for &[u8] {
-        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 {
diff --git a/futures-util/Cargo.toml b/futures-util/Cargo.toml
index 33f0428c..1825ea83 100644
--- a/futures-util/Cargo.toml
+++ b/futures-util/Cargo.toml
@@ -33,6 +33,7 @@ select-macro = ["async-await", "futures-select-macro-preview", "proc-macro-hack"
 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 }
diff --git a/futures-util/src/compat/compat01as03.rs b/futures-util/src/compat/compat01as03.rs
index f9471031..dde42a4b 100644
--- a/futures-util/src/compat/compat01as03.rs
+++ b/futures-util/src/compat/compat01as03.rs
@@ -368,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};
 
@@ -433,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 763d19b2..b9632ced 100644
--- a/futures-util/src/compat/compat03as01.rs
+++ b/futures-util/src/compat/compat03as01.rs
@@ -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/future/abortable.rs b/futures-util/src/future/abortable.rs
index 3c57cfc0..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;
@@ -124,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/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/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
index 35adfdbb..ec6bbff7 100644
--- a/futures-util/src/io/chain.rs
+++ b/futures-util/src/io/chain.rs
@@ -1,5 +1,7 @@
 use futures_core::task::{Context, Poll};
-use futures_io::{AsyncBufRead, AsyncRead, Initializer, IoSliceMut};
+#[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;
@@ -117,6 +119,7 @@ where
         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() {
diff --git a/futures-util/src/io/empty.rs b/futures-util/src/io/empty.rs
index 58e0d52f..60fa08a9 100644
--- a/futures-util/src/io/empty.rs
+++ b/futures-util/src/io/empty.rs
@@ -1,5 +1,7 @@
 use futures_core::task::{Context, Poll};
-use futures_io::{AsyncBufRead, AsyncRead, Initializer};
+#[cfg(feature = "read_initializer")]
+use futures_io::Initializer;
+use futures_io::{AsyncBufRead, AsyncRead};
 use std::fmt;
 use std::io;
 use std::pin::Pin;
@@ -42,6 +44,7 @@ impl AsyncRead for Empty {
         Poll::Ready(Ok(0))
     }
 
+    #[cfg(feature = "read_initializer")]
     #[inline]
     unsafe fn initializer(&self) -> Initializer {
         Initializer::nop()
diff --git a/futures-util/src/io/mod.rs b/futures-util/src/io/mod.rs
index 3f2e6ba2..db8bd134 100644
--- a/futures-util/src/io/mod.rs
+++ b/futures-util/src/io/mod.rs
@@ -9,17 +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, 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;
 
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
index 0821ec9b..9f493d1a 100644
--- a/futures-util/src/io/repeat.rs
+++ b/futures-util/src/io/repeat.rs
@@ -1,5 +1,7 @@
 use futures_core::task::{Context, Poll};
-use futures_io::{AsyncRead, Initializer, IoSliceMut};
+#[cfg(feature = "read_initializer")]
+use futures_io::Initializer;
+use futures_io::{AsyncRead, IoSliceMut};
 use std::fmt;
 use std::io;
 use std::pin::Pin;
@@ -57,6 +59,7 @@ impl AsyncRead for Repeat {
         Poll::Ready(Ok(nwritten))
     }
 
+    #[cfg(feature = "read_initializer")]
     #[inline]
     unsafe fn initializer(&self) -> Initializer {
         Initializer::nop()
diff --git a/futures-util/src/io/take.rs b/futures-util/src/io/take.rs
index 39513714..3eaa5f5c 100644
--- a/futures-util/src/io/take.rs
+++ b/futures-util/src/io/take.rs
@@ -1,5 +1,7 @@
 use futures_core::task::{Context, Poll};
-use futures_io::{AsyncRead, AsyncBufRead, Initializer};
+#[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;
@@ -183,6 +185,7 @@ impl<R: AsyncRead> AsyncRead for Take<R> {
         Poll::Ready(Ok(n))
     }
 
+    #[cfg(feature = "read_initializer")]
     unsafe fn initializer(&self) -> Initializer {
         self.inner.initializer()
     }
diff --git a/futures-util/src/lib.rs b/futures-util/src/lib.rs
index 4446c147..3f91a9eb 100644
--- a/futures-util/src/lib.rs
+++ b/futures-util/src/lib.rs
@@ -2,6 +2,7 @@
 //! and the `AsyncRead` and `AsyncWrite` traits.
 
 #![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)]
@@ -19,6 +20,9 @@ compile_error!("The `cfg-target-has-atomic` feature requires the `unstable` feat
 #[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;
 
diff --git a/futures-util/src/lock/bilock.rs b/futures-util/src/lock/bilock.rs
index 8ea809d0..5e1fe35b 100644
--- a/futures-util/src/lock/bilock.rs
+++ b/futures-util/src/lock/bilock.rs
@@ -11,10 +11,6 @@ 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.
@@ -157,7 +153,7 @@ impl<T> BiLock<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()
@@ -218,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.
 ///
diff --git a/futures-util/src/sink/buffer.rs b/futures-util/src/sink/buffer.rs
index 046efcaa..90480bd5 100644
--- a/futures-util/src/sink/buffer.rs
+++ b/futures-util/src/sink/buffer.rs
@@ -76,6 +76,10 @@ 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 {
diff --git a/futures-util/src/sink/err_into.rs b/futures-util/src/sink/err_into.rs
index d93c8aa3..af198fea 100644
--- a/futures-util/src/sink/err_into.rs
+++ b/futures-util/src/sink/err_into.rs
@@ -70,6 +70,10 @@ 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>
diff --git a/futures-util/src/sink/map_err.rs b/futures-util/src/sink/map_err.rs
index bd2ed7ae..7333d204 100644
--- a/futures-util/src/sink/map_err.rs
+++ b/futures-util/src/sink/map_err.rs
@@ -95,6 +95,10 @@ 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> {
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 0b2726e0..baa7536f 100644
--- a/futures-util/src/sink/with_flat_map.rs
+++ b/futures-util/src/sink/with_flat_map.rs
@@ -121,12 +121,17 @@ 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>
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 0486c51c..309625b7 100644
--- a/futures-util/src/stream/catch_unwind.rs
+++ b/futures-util/src/stream/catch_unwind.rs
@@ -45,6 +45,14 @@ 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> {
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 1c8b518d..87e9fbb8 100644
--- a/futures-util/src/stream/chunks.rs
+++ b/futures-util/src/stream/chunks.rs
@@ -105,6 +105,17 @@ 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> {
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/iter.rs b/futures-util/src/stream/iter.rs
index 9701d750..e9df81ce 100644
--- a/futures-util/src/stream/iter.rs
+++ b/futures-util/src/stream/iter.rs
@@ -41,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/once.rs b/futures-util/src/stream/once.rs
index 8dc6ed6a..4f68b0ce 100644
--- a/futures-util/src/stream/once.rs
+++ b/futures-util/src/stream/once.rs
@@ -37,17 +37,22 @@ 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))
+        }
     }
 }
 
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 d869b5df..21749eb7 100644
--- a/futures-util/src/stream/repeat.rs
+++ b/futures-util/src/stream/repeat.rs
@@ -39,6 +39,10 @@ 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>
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 f18f6c42..666d9dea 100644
--- a/futures-util/src/stream/skip_while.rs
+++ b/futures-util/src/stream/skip_while.rs
@@ -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 cd9113e5..4d0dbd31 100644
--- a/futures-util/src/stream/take.rs
+++ b/futures-util/src/stream/take.rs
@@ -1,3 +1,4 @@
+use core::cmp;
 use core::pin::Pin;
 use futures_core::stream::{Stream, FusedStream};
 use futures_core::task::{Context, Poll};
@@ -79,6 +80,23 @@ 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>
diff --git a/futures-util/src/stream/take_while.rs b/futures-util/src/stream/take_while.rs
index e6225ca3..68606ec2 100644
--- a/futures-util/src/stream/take_while.rs
+++ b/futures-util/src/stream/take_while.rs
@@ -127,6 +127,20 @@ 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>
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/zip.rs b/futures-util/src/stream/zip.rs
index c8d7f3df..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};
@@ -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/try_stream/and_then.rs b/futures-util/src/try_stream/and_then.rs
index d92b9a93..809c32a9 100644
--- a/futures-util/src/try_stream/and_then.rs
+++ b/futures-util/src/try_stream/and_then.rs
@@ -102,6 +102,17 @@ 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>
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_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/or_else.rs b/futures-util/src/try_stream/or_else.rs
index 4a18fe96..33310d1c 100644
--- a/futures-util/src/try_stream/or_else.rs
+++ b/futures-util/src/try_stream/or_else.rs
@@ -103,6 +103,17 @@ 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>
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 0c0355a9..a3d6803a 100644
--- a/futures-util/src/try_stream/try_skip_while.rs
+++ b/futures-util/src/try_stream/try_skip_while.rs
@@ -131,6 +131,16 @@ 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>
diff --git a/futures/Cargo.toml b/futures/Cargo.toml
index e17f4833..71e99ea6 100644
--- a/futures/Cargo.toml
+++ b/futures/Cargo.toml
@@ -46,9 +46,10 @@ 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-util-preview/unstable"]
+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 33f6e79a..f4973ff7 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)]
 
@@ -40,6 +41,9 @@ compile_error!("The `cfg-target-has-atomic` feature requires the `unstable` feat
 #[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;
 
 #[doc(hidden)] pub use futures_core::future::Future;
@@ -298,9 +302,12 @@ 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, Chain, Close, CopyInto, CopyBufInto, empty, Empty,


Test patch:
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/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]