Repository: alexcrichton/futures-rs
Pull request #634 (2017-11-08 06:47:52)

Deprecate Future::wait.
This commit deprecates `Future::wait`. Instead of having a trait function, a wrapper is provided at `future::Blocking`. `Blocking` wraps a `Future` value and provides thread-blocking operation on it.

The deprecation strategy differs from that proposed in tokio-rs/tokio-rfcs#3. The intent is that there still will be current-thread executor functions described in the PR but those will come after.

This PR proposes adding `future::Blocking` which wraps a `T: Future` and provides thread-blocking operations. Currently, only `wait` and `poll` are provided, but `wait_timeout` could be added easily later.

I opted for this strategy because in [@alex's branch](https://github.com/alexcrichton/futures-rs/tree/spawn), he provided both `BlockingStream` and `BlockingSink`, yet `BlockingFuture` was omitted. If `BlockingSink` and `BlockingStream` are worth having (and I think they are), then `BlockingFuture` should also be provided. This proves to be a natural place to add any functions for using a future outside the scope of an executor.

I believe that this solves the problems being encountered with `wait` by not being a trait function (you won't find this function by looking at what the `Future` trait provides). Also, in the future, thread-blocking operations will panic when invoked from the context of an executor.

This PR opts to name the type `future::Blocking`.

If this PR is accepted, I will apply a similar treatment to `Sink` and `Stream`.

This PR is based against the `tokio-reform` branch, and not master.

## Remaining

* [x] Deprecate `Wait` types
* [x] Deprecate functions on `Spawn`.
* [x] Check docs
* [x] Add test for new functionality.

carllerche: Also worth noting, calling `future:::Blocking::wait` will *only* drive the current future and *not* the current thread executor. This avoids the problem of accidentally dropping tasks that never completed.
carllerche: I would also want to add some tests for `Blocking::poll` (naming TBD?) before merging. I am waiting to hear if this is PR represents a good strategy.
aturon: @carllerche It'd be helpful to see the rationale for this API surface as an addition beyond what's in the RFC (maybe I missed the motivation somewhere?) How should one think about having both this and the current-thread executor?

> Also worth noting, calling `future:::Blocking::wait` will *only* drive the current future and *not* the current thread executor. This avoids the problem of accidentally dropping tasks that never completed.

I'm also curious to understand how this fits into the RFC design, which was also attempting to solve this problem.
carllerche: @aturon A few things:

* The RFC doesn't touch on what to do with `Stream::wait` and `Sink::wait`.
* The latest version of the RFC doesn't actually provide a way to block the thread waiting until a single future completes.

@alexcrichton's PR deals with `Stream::wait` and `Sink:wait` by providing `BlockingStream` and `BlockingSink`, but there is no way in the PR to block waiting for a single future to complete. This fact is apparent when trying to update the tests.

It doesn't make sense to include `BlockingStream` and `BlockingSink` without including `BlockingFuture`. So, either those two types should go or a `BlockingFuture` equivalent should be added.

> I'm also curious to understand how this fits into the RFC design, which was also attempting to solve this problem.

The goal of the RFC was to prevent being able to spawn current-thread tasks and having them be implicitly dropped. This PR maintains this goal.

* `Blocking::wait` does not "enter" the current-thread executor, as such a current-thread spawn will panic.
aturon: @carllerche Got it, thanks! And yeah, an earlier version of the RFC had something like this (though it was less elaborate; it wasn't a wrapper), which was lost in the shuffle.

I'm good with this approach!
twmb: Is there a reason for `try_take` to exist at the moment? I realize that a no-op notifier is the only way to internally implement it, but w.r.t. the inability to avoid actually registering interest, it seems like fitting a square peg into a round hole. Also, I'm not sure why I would wrap a future in `Blocking` and then have the desire to check it in a non-blocking manner. This also extends down into the `get_ref`, `get_mut`, and `into_inner` functions - if the only purpose of `Blocking` is truly to block until the future returns, there is no need to access a `Blocking`'s internals.

Additionally, I feel as if `blocking`, and its corresponding type type `Blocking`, feel ergonomically awkward to read and I think that feeling would translate to use as well. I recognize that a pattern in the futures crate is to name wrapper types after the functions that produce them (`and_then` -> `AndThen`, etc.), but those function names are all verbs that extend off a `Future`, and the types are really only used in in return types (something that should go away once `impl Trait` is stable).

I'd prefer something closer to `blocking::wrap(f)` or `block::wrap(f)` - this _reads_ to me as something that provides blocking is wrapping a future, which is exactly what `blocking(f)` does. Assuming that `.wait()` will be chained on that 99% of the time, `blocking::wrap(f).wait()` still gives me a bit more context as to what is happening as opposed to `blocking(f).wait()`. Alternatively, `block_on(f)`, which would directly replace `wait` as a purely blocking function call).

In terms of the type `Blocking` itself - this feels the weirdest, because the -ing suffix makes me read it as a verb, but the word itself is a [verbal noun](https://forum.wordreference.com/threads/deverbal-and-verbal-nouns.2689882/#post-13577342). The lack of a sentence as context makes the isolation weird. This is nitpicky, and feel free to ignore this and the prior two paragraphs. I bring this up (and will add more in other tickets / prs / rfc#3) because I had a difficult time reading and understanding the futures api and source due to small semantic things such as this.
carllerche: @twmb 

`try_take` is very useful in tests.

> if the only purpose of Blocking is truly to block until the future returns, there is no need to access a Blocking's internals.

I don't understand why you believe this to be true. A future is very stateful, there is plenty of reason to want to access the inner T before or after blocking. Also, at some point there will be a `wait_timeout`.

Re naming, the only thing that I care about is that it is succinct, and clearly communicates that the type can block the thread.
twmb: > I don't understand why you believe this to be true.

I meant that only if `try_take` were to be removed and the only option were to be `wait` - `wait_timeout` is more compelling. I'm still a bit skeptical on providing a public function that is mostly useful for tests, but that's not really a showstopper for me.

re: Re naming, if anybody else has input, that would be great as well. `Blocking` and `blocking` are likely perfectly fine, but I figured I would throw that out there.
carllerche: @alexcrichton could you enable "squash & merge" and "rebase and merge" for this repo?

Patch:
diff --git a/src/future/blocking.rs b/src/future/blocking.rs
new file mode 100644
index 00000000..268af166
--- /dev/null
+++ b/src/future/blocking.rs
@@ -0,0 +1,94 @@
+use Async;
+use future::Future;
+use executor::{self, NotifyHandle};
+use task_impl::ThreadNotify;
+
+/// Provides thread-blocking operations on a future.
+///
+/// See [`blocking`](fn.blocking.html) documentation for more details.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless used"]
+pub struct Blocking<T> {
+    inner: executor::Spawn<T>,
+}
+
+/// Provides thread-blocking operations on a future.
+///
+/// `blocking` consumes ownership of `future`, returning a `Blocking` backed by
+/// the future. The `Blocking` value exposes thread-blocking operations that
+/// allow getting the realized value of the future. For example,
+/// `Blocking::wait` will block the current thread until the inner future has
+/// completed and return the completed value.
+///
+/// **These operations will block the current thread**. This means that they
+/// should not be called while in the context of a task executor as this will
+/// block the task executor's progress.
+pub fn blocking<T: Future>(future: T) -> Blocking<T> {
+    let inner = executor::spawn(future);
+    Blocking { inner: inner }
+}
+
+impl<T: Future> Blocking<T> {
+    /// Query the inner future to see if its value has become available.
+    ///
+    /// Unlike `Future::poll`, this function does **not** register interest if
+    /// the inner future is not in a ready state.
+    ///
+    /// This function will return immediately if the inner future is not ready.
+    pub fn try_take(&mut self) -> Option<Result<T::Item, T::Error>> {
+        match self.inner.poll_future_notify(&NotifyHandle::noop(), 0) {
+            Ok(Async::NotReady) => None,
+            Ok(Async::Ready(v)) => Some(Ok(v)),
+            Err(e) => Some(Err(e)),
+        }
+    }
+
+    /// Block the current thread until this future is resolved.
+    ///
+    /// This method will drive the inner future to completion via
+    /// `Future::poll`. **The current thread will be blocked** until the future
+    /// transitions to a ready state. Once the future is complete, the result of
+    /// this future is returned.
+    ///
+    /// > **Note:** This method is not appropriate to call on event loops or
+    /// >           similar I/O situations because it will prevent the event
+    /// >           loop from making progress (this blocks the thread). This
+    /// >           method should only be called when it's guaranteed that the
+    /// >           blocking work associated with this future will be completed
+    /// >           by another thread.
+    ///
+    /// This method is only available when the `use_std` feature of this
+    /// library is activated, and it is activated by default.
+    ///
+    /// # Panics
+    ///
+    /// This method does not attempt to catch panics. If the `poll` function of
+    /// the inner future panics, the panic will be propagated to the caller.
+    pub fn wait(&mut self) -> Result<T::Item, T::Error> {
+        ThreadNotify::with_current(|notify| {
+            loop {
+                match self.inner.poll_future_notify(notify, 0)? {
+                    Async::NotReady => notify.park(),
+                    Async::Ready(e) => return Ok(e),
+                }
+            }
+        })
+    }
+}
+
+impl<T> Blocking<T> {
+    /// Get a shared reference to the inner future.
+    pub fn get_ref(&self) -> &T {
+        self.inner.get_ref()
+    }
+
+    /// Get a mutable reference to the inner future.
+    pub fn get_mut(&mut self) -> &mut T {
+        self.inner.get_mut()
+    }
+
+    /// Consume the `Blocking`, returning its inner future.
+    pub fn into_inner(self) -> T {
+        self.inner.into_inner()
+    }
+}
diff --git a/src/future/mod.rs b/src/future/mod.rs
index c858a583..b20c22e7 100644
--- a/src/future/mod.rs
+++ b/src/future/mod.rs
@@ -78,11 +78,13 @@ pub use self::either::Either;
 pub use self::inspect::Inspect;
 
 if_std! {
+    mod blocking;
     mod catch_unwind;
     mod join_all;
     mod select_all;
     mod select_ok;
     mod shared;
+    pub use self::blocking::{blocking, Blocking};
     pub use self::catch_unwind::CatchUnwind;
     pub use self::join_all::{join_all, JoinAll};
     pub use self::select_all::{SelectAll, SelectAllNext, select_all};
@@ -271,28 +273,10 @@ pub trait Future {
     /// error.
     fn poll(&mut self) -> Poll<Self::Item, Self::Error>;
 
-    /// Block the current thread until this future is resolved.
-    ///
-    /// This method will consume ownership of this future, driving it to
-    /// completion via `poll` and blocking the current thread while it's waiting
-    /// for the value to become available. Once the future is resolved the
-    /// result of this future is returned.
-    ///
-    /// > **Note:** This method is not appropriate to call on event loops or
-    /// >           similar I/O situations because it will prevent the event
-    /// >           loop from making progress (this blocks the thread). This
-    /// >           method should only be called when it's guaranteed that the
-    /// >           blocking work associated with this future will be completed
-    /// >           by another thread.
-    ///
-    /// This method is only available when the `use_std` feature of this
-    /// library is activated, and it is activated by default.
-    ///
-    /// # Panics
-    ///
-    /// This function does not attempt to catch panics. If the `poll` function
-    /// of this future panics, panics will be propagated to the caller.
     #[cfg(feature = "use_std")]
+    #[doc(hidden)]
+    #[allow(deprecated)]
+    #[deprecated(note = "use `future::blocking` instead")]
     fn wait(self) -> result::Result<Self::Item, Self::Error>
         where Self: Sized
     {
diff --git a/src/sink/blocking.rs b/src/sink/blocking.rs
new file mode 100644
index 00000000..0987ce7d
--- /dev/null
+++ b/src/sink/blocking.rs
@@ -0,0 +1,121 @@
+use {Async, AsyncSink};
+use sink::Sink;
+use executor;
+use task_impl::ThreadNotify;
+
+/// Provides thread-blocking operations on a sink.
+///
+/// See [`blocking`](fn.blocking.html) documentation for more details.
+#[derive(Debug)]
+#[must_use = "sinks do nothing unless used"]
+pub struct Blocking<T: Sink> {
+    inner: Option<executor::Spawn<T>>,
+}
+
+/// Provides thread-blocking operations on a sink.
+///
+/// `blocking` consumes ownership of `sink`, returning a `Blocking` backed by
+/// the sink. The `Blocking` value exposes thread-blocking operations that allow
+/// sending values into the sink. For example, `Blocking::send` will block the
+/// current thread until the inner sink has capacity to accept the item being
+/// sent.
+///
+/// **These operations will block the current thread**. This means that they
+/// should not be called while in the context of a task executor as this will
+/// block the task executor's progress. Also, note that **this value will block
+/// the current thread on drop**. Droping `Blocking` will ensure that the inner
+/// sink is fully flushed before completing the drop. This could block the
+/// current thread.
+///
+/// Any errors that happen in the process of flushing when `Blocking` is dropped
+/// will be ignored. Code that wishes to handle such errors must manually call
+/// flush before the value is dropped. This behavior is similar to [`BufWriter`]
+/// in std.
+///
+/// [`BufWriter`]: https://doc.rust-lang.org/std/io/struct.BufWriter.html
+pub fn blocking<T: Sink>(sink: T) -> Blocking<T> {
+    let inner = executor::spawn(sink);
+    Blocking { inner: Some(inner) }
+}
+
+impl<T: Sink> Blocking<T> {
+    /// Sends a value to this sink, blocking the current thread until it's able
+    /// to do so.
+    ///
+    /// This function will take the `value` provided and call the underlying
+    /// sink's `start_send` function until it's ready to accept the value. If
+    /// the function returns `NotReady` then the current thread is blocked
+    /// until it is otherwise ready to accept the value.
+    ///
+    /// # Return value
+    ///
+    /// If `Ok(())` is returned then the `value` provided was successfully sent
+    /// along the sink, and if `Err(e)` is returned then an error occurred
+    /// which prevented the value from being sent.
+    pub fn send(&mut self, mut value: T::SinkItem) -> Result<(), T::SinkError> {
+        ThreadNotify::with_current(|notify| {
+            loop {
+                let inner = self.inner.as_mut().unwrap();
+                match inner.start_send_notify(value, notify, 0) {
+                    Ok(AsyncSink::Ready) => return Ok(()),
+                    Ok(AsyncSink::NotReady(v)) => {
+                        value = v;
+                        notify.park();
+                    }
+                    Err(e) => return Err(e),
+                }
+            }
+        })
+    }
+
+    /// Flushes any buffered data in this sink, blocking the current thread
+    /// until it's entirely flushed.
+    ///
+    /// This function will call the underlying sink's `poll_complete` method
+    /// until it returns that it's ready to proceed. If the method returns
+    /// `NotReady` the current thread will be blocked until it's otherwise
+    /// ready to proceed.
+    pub fn flush(&mut self) -> Result<(), T::SinkError> {
+        ThreadNotify::with_current(|notify| {
+            loop {
+                let inner = self.inner.as_mut().unwrap();
+                match inner.poll_flush_notify(notify, 0) {
+                    Ok(Async::Ready(_)) => return Ok(()),
+                    Ok(Async::NotReady) => notify.park(),
+                    Err(e) => return Err(e),
+                }
+            }
+        })
+    }
+
+    /// Get a shared reference to the inner sink.
+    pub fn get_ref(&self) -> &T {
+        self.inner.as_ref().unwrap().get_ref()
+    }
+
+    /// Get a mutable reference to the inner sink.
+    pub fn get_mut(&mut self) -> &mut T {
+        self.inner.as_mut().unwrap().get_mut()
+    }
+
+    /// Consume the `Blocking`, returning its inner sink.
+    pub fn into_inner(mut self) -> T {
+        self.inner.take().unwrap().into_inner()
+    }
+}
+
+impl<T: Sink> Drop for Blocking<T> {
+    fn drop(&mut self) {
+        ThreadNotify::with_current(|notify| {
+            if let Some(ref mut inner) = self.inner {
+                loop {
+                    match inner.close_notify(notify, 0) {
+                        Ok(Async::Ready(_)) => break,
+                        Ok(Async::NotReady) => notify.park(),
+                        Err(_) => break,
+                    }
+                }
+            }
+        })
+    }
+}
diff --git a/src/sink/mod.rs b/src/sink/mod.rs
index 9dece12b..b219d202 100644
--- a/src/sink/mod.rs
+++ b/src/sink/mod.rs
@@ -23,10 +23,15 @@ mod send_all;
 mod map_err;
 
 if_std! {
+    mod blocking;
     mod buffer;
     mod wait;
 
+    pub use self::blocking::{blocking, Blocking};
     pub use self::buffer::Buffer;
+
+    #[doc(hidden)]
+    #[allow(deprecated)]
     pub use self::wait::Wait;
 
     // TODO: consider expanding this via e.g. FromIterator
@@ -284,13 +289,10 @@ pub trait Sink {
     #[cfg(not(feature = "with-deprecated"))]
     fn close(&mut self) -> Poll<(), Self::SinkError>;
 
-    /// Creates a new object which will produce a synchronous sink.
-    ///
-    /// The sink returned does **not** implement the `Sink` trait, and instead
-    /// only has two methods: `send` and `flush`. These two methods correspond
-    /// to `start_send` and `poll_complete` above except are executed in a
-    /// blocking fashion.
     #[cfg(feature = "use_std")]
+    #[doc(hidden)]
+    #[allow(deprecated)]
+    #[deprecated(note = "use `sink::blocking` instead")]
     fn wait(self) -> Wait<Self>
         where Self: Sized
     {
diff --git a/src/sink/wait.rs b/src/sink/wait.rs
index 940a5886..1f25d4a4 100644
--- a/src/sink/wait.rs
+++ b/src/sink/wait.rs
@@ -1,13 +1,11 @@
+#![allow(deprecated)]
+
 use sink::Sink;
 use executor;
 
-/// A sink combinator which converts an asynchronous sink to a **blocking
-/// sink**.
-///
-/// Created by the `Sink::wait` method, this function transforms any sink into a
-/// blocking version. This is implemented by blocking the current thread when a
-/// sink is otherwise unable to make progress.
 #[must_use = "sinks do nothing unless used"]
+#[doc(hidden)]
+#[deprecated(note = "use `sink::blocking` instead")]
 #[derive(Debug)]
 pub struct Wait<S> {
     sink: executor::Spawn<S>,
@@ -33,6 +31,7 @@ impl<S: Sink> Wait<S> {
     /// If `Ok(())` is returned then the `value` provided was successfully sent
     /// along the sink, and if `Err(e)` is returned then an error occurred
     /// which prevented the value from being sent.
+    #[allow(deprecated)]
     pub fn send(&mut self, value: S::SinkItem) -> Result<(), S::SinkError> {
         self.sink.wait_send(value)
     }
@@ -44,6 +43,7 @@ impl<S: Sink> Wait<S> {
     /// until it returns that it's ready to proceed. If the method returns
     /// `NotReady` the current thread will be blocked until it's otherwise
     /// ready to proceed.
+    #[allow(deprecated)]
     pub fn flush(&mut self) -> Result<(), S::SinkError> {
         self.sink.wait_flush()
     }
@@ -53,6 +53,7 @@ impl<S: Sink> Wait<S> {
     /// This function will call the underlying sink's `close` method
     /// until it returns that it's closed. If the method returns
     /// `NotReady` the current thread will be blocked until it's otherwise closed.
+    #[allow(deprecated)]
     pub fn close(&mut self) -> Result<(), S::SinkError> {
         self.sink.wait_close()
     }
diff --git a/src/stream/blocking.rs b/src/stream/blocking.rs
new file mode 100644
index 00000000..3f79a367
--- /dev/null
+++ b/src/stream/blocking.rs
@@ -0,0 +1,61 @@
+use Async;
+use stream::Stream;
+use executor;
+use task_impl::ThreadNotify;
+
+/// Provides thread-blocking stream iteration.
+///
+/// See [`blocking`](fn.blocking.html) documentation for more details.
+#[derive(Debug)]
+#[must_use = "iterators do nothing unless advanced"]
+pub struct Blocking<T> {
+    inner: executor::Spawn<T>,
+}
+
+/// Provides thread-blocking stream iteration.
+///
+/// `blocking` consumes ownership of `stream`, returning a `Blocking` backed by
+/// the stream. The `Blocking` value provides a thread-blocking iterator that
+/// yields each consecutive value from stream.
+///
+/// **Iteration will block the current thread**. This means that it should not
+/// be performed while in the context of a task executor as this will block the
+/// task executor's progress.
+pub fn blocking<T: Stream>(stream: T) -> Blocking<T> {
+    let inner = executor::spawn(stream);
+    Blocking { inner: inner }
+}
+
+impl<T> Blocking<T> {
+    /// Get a shared reference to the inner stream.
+    pub fn get_ref(&self) -> &T {
+        self.inner.get_ref()
+    }
+
+    /// Get a mutable reference to the inner stream.
+    pub fn get_mut(&mut self) -> &mut T {
+        self.inner.get_mut()
+    }
+
+    /// Consume the `Blocking`, returning its inner stream.
+    pub fn into_inner(self) -> T {
+        self.inner.into_inner()
+    }
+}
+
+impl<T: Stream> Iterator for Blocking<T> {
+    type Item = Result<T::Item, T::Error>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        ThreadNotify::with_current(|notify| {
+            loop {
+                match self.inner.poll_stream_notify(notify, 0) {
+                    Ok(Async::Ready(Some(v))) => return Some(Ok(v)),
+                    Ok(Async::Ready(None)) => return None,
+                    Ok(Async::NotReady) => notify.park(),
+                    Err(e) => return Some(Err(e)),
+                }
+            }
+        })
+    }
+}
diff --git a/src/stream/mod.rs b/src/stream/mod.rs
index e3ea2e3f..ffe61ba0 100644
--- a/src/stream/mod.rs
+++ b/src/stream/mod.rs
@@ -95,6 +95,7 @@ use sink::{Sink};
 if_std! {
     use std;
 
+    mod blocking;
     mod buffered;
     mod buffer_unordered;
     mod catch_unwind;
@@ -105,16 +106,20 @@ if_std! {
     mod split;
     pub mod futures_unordered;
     mod futures_ordered;
+    pub use self::blocking::{blocking, Blocking};
     pub use self::buffered::Buffered;
     pub use self::buffer_unordered::BufferUnordered;
     pub use self::catch_unwind::CatchUnwind;
     pub use self::chunks::Chunks;
     pub use self::collect::Collect;
-    pub use self::wait::Wait;
     pub use self::split::{SplitStream, SplitSink};
     pub use self::futures_unordered::FuturesUnordered;
     pub use self::futures_ordered::{futures_ordered, FuturesOrdered};
 
+    #[doc(hidden)]
+    #[allow(deprecated)]
+    pub use self::wait::Wait;
+
     #[doc(hidden)]
     #[cfg(feature = "with-deprecated")]
     #[allow(deprecated)]
@@ -214,29 +219,10 @@ pub trait Stream {
     //       item? basically just says "please make more progress internally"
     //       seems crucial for buffering to actually make any sense.
 
-    /// Creates an iterator which blocks the current thread until each item of
-    /// this stream is resolved.
-    ///
-    /// This method will consume ownership of this stream, returning an
-    /// implementation of a standard iterator. This iterator will *block the
-    /// current thread* on each call to `next` if the item in the stream isn't
-    /// ready yet.
-    ///
-    /// > **Note:** This method is not appropriate to call on event loops or
-    /// >           similar I/O situations because it will prevent the event
-    /// >           loop from making progress (this blocks the thread). This
-    /// >           method should only be called when it's guaranteed that the
-    /// >           blocking work associated with this stream will be completed
-    /// >           by another thread.
-    ///
-    /// This method is only available when the `use_std` feature of this
-    /// library is activated, and it is activated by default.
-    ///
-    /// # Panics
-    ///
-    /// The returned iterator does not attempt to catch panics. If the `poll`
-    /// function panics, panics will be propagated to the caller of `next`.
     #[cfg(feature = "use_std")]
+    #[doc(hidden)]
+    #[allow(deprecated)]
+    #[deprecated(note = "use `stream::blocking` instead")]
     fn wait(self) -> Wait<Self>
         where Self: Sized
     {
diff --git a/src/stream/wait.rs b/src/stream/wait.rs
index 80acb6c2..7dbabfef 100644
--- a/src/stream/wait.rs
+++ b/src/stream/wait.rs
@@ -1,13 +1,11 @@
+#![allow(deprecated)]
+
 use stream::Stream;
 use executor;
 
-/// A stream combinator which converts an asynchronous stream to a **blocking
-/// iterator**.
-///
-/// Created by the `Stream::wait` method, this function transforms any stream
-/// into a standard iterator. This is implemented by blocking the current thread
-/// while items on the underlying stream aren't ready yet.
 #[must_use = "iterators do nothing unless advanced"]
+#[doc(hidden)]
+#[deprecated(note = "use `stream::blocking` instead")]
 #[derive(Debug)]
 pub struct Wait<S> {
     stream: executor::Spawn<S>,
@@ -47,6 +45,7 @@ pub fn new<S: Stream>(s: S) -> Wait<S> {
 impl<S: Stream> Iterator for Wait<S> {
     type Item = Result<S::Item, S::Error>;
 
+    #[allow(deprecated)]
     fn next(&mut self) -> Option<Self::Item> {
         self.stream.wait_stream()
     }
diff --git a/src/task_impl/mod.rs b/src/task_impl/mod.rs
index da42a78d..c2fa722d 100644
--- a/src/task_impl/mod.rs
+++ b/src/task_impl/mod.rs
@@ -615,6 +615,19 @@ impl NotifyHandle {
         NotifyHandle { inner: inner }
     }
 
+    /// Return a no-op notify handle
+    pub fn noop() -> NotifyHandle {
+        struct Noop;
+
+        impl Notify for Noop {
+            fn notify(&self, _id: usize) {}
+        }
+
+        const NOOP: &'static Noop = &Noop;
+
+        NotifyHandle::from(NOOP)
+    }
+
     /// Invokes the underlying instance of `Notify` with the provided `id`.
     pub fn notify(&self, id: usize) {
         unsafe { (*self.inner).notify(id) }
diff --git a/src/task_impl/std/mod.rs b/src/task_impl/std/mod.rs
index 108c4e57..ce58aafc 100644
--- a/src/task_impl/std/mod.rs
+++ b/src/task_impl/std/mod.rs
@@ -230,12 +230,10 @@ impl<F: Future> Spawn<F> {
         self.enter(BorrowedUnpark::Old(&unpark), |f| f.poll())
     }
 
-    /// Waits for the internal future to complete, blocking this thread's
-    /// execution until it does.
-    ///
-    /// This function will call `poll_future` in a loop, waiting for the future
-    /// to complete. When a future cannot make progress it will use
-    /// `thread::park` to block the current thread.
+    #[cfg(feature = "use_std")]
+    #[doc(hidden)]
+    #[allow(deprecated)]
+    #[deprecated(note = "use `future::blocking` instead")]
     pub fn wait_future(&mut self) -> Result<F::Item, F::Error> {
         ThreadNotify::with_current(|notify| {
 
@@ -293,8 +291,9 @@ impl<S: Stream> Spawn<S> {
         self.enter(BorrowedUnpark::Old(&unpark), |s| s.poll())
     }
 
-    /// Like `wait_future`, except only waits for the next element to arrive on
-    /// the underlying stream.
+    #[doc(hidden)]
+    #[allow(deprecated)]
+    #[deprecated(note = "use `stream::blocking` instead")]
     pub fn wait_stream(&mut self) -> Option<Result<S::Item, S::Error>> {
         ThreadNotify::with_current(|notify| {
 
@@ -335,11 +334,9 @@ impl<S: Sink> Spawn<S> {
         self.enter(BorrowedUnpark::Old(unpark), |s| s.poll_complete())
     }
 
-    /// Blocks the current thread until it's able to send `value` on this sink.
-    ///
-    /// This function will send the `value` on the sink that this task wraps. If
-    /// the sink is not ready to send the value yet then the current thread will
-    /// be blocked until it's able to send the value.
+    #[doc(hidden)]
+    #[allow(deprecated)]
+    #[deprecated(note = "use `sink::blocking` instead")]
     pub fn wait_send(&mut self, mut value: S::SinkItem)
                      -> Result<(), S::SinkError> {
         ThreadNotify::with_current(|notify| {
@@ -354,14 +351,9 @@ impl<S: Sink> Spawn<S> {
         })
     }
 
-    /// Blocks the current thread until it's able to flush this sink.
-    ///
-    /// This function will call the underlying sink's `poll_complete` method
-    /// until it returns that it's ready, proxying out errors upwards to the
-    /// caller if one occurs.
-    ///
-    /// The thread will be blocked until `poll_complete` returns that it's
-    /// ready.
+    #[doc(hidden)]
+    #[allow(deprecated)]
+    #[deprecated(note = "use `sink::blocking` instead")]
     pub fn wait_flush(&mut self) -> Result<(), S::SinkError> {
         ThreadNotify::with_current(|notify| {
 
@@ -374,11 +366,9 @@ impl<S: Sink> Spawn<S> {
         })
     }
 
-    /// Blocks the current thread until it's able to close this sink.
-    ///
-    /// This function will close the sink that this task wraps. If the sink
-    /// is not ready to be close yet, then the current thread will be blocked
-    /// until it's closed.
+    #[doc(hidden)]
+    #[allow(deprecated)]
+    #[deprecated(note = "use `future::blocking` instead")]
     pub fn wait_close(&mut self) -> Result<(), S::SinkError> {
         ThreadNotify::with_current(|notify| {
 
@@ -484,7 +474,7 @@ impl Unpark for RunInner {
 
 // ===== ThreadNotify =====
 
-struct ThreadNotify {
+pub struct ThreadNotify {
     state: AtomicUsize,
     mutex: Mutex<()>,
     condvar: Condvar,
@@ -503,13 +493,13 @@ thread_local! {
 }
 
 impl ThreadNotify {
-    fn with_current<F, R>(f: F) -> R
+    pub fn with_current<F, R>(f: F) -> R
         where F: FnOnce(&Arc<ThreadNotify>) -> R,
     {
         CURRENT_THREAD_NOTIFY.with(|notify| f(notify))
     }
 
-    fn park(&self) {
+    pub fn park(&self) {
         // If currently notified, then we skip sleeping. This is checked outside
         // of the lock to avoid acquiring a mutex if not necessary.
         match self.state.compare_and_swap(NOTIFY, IDLE, Ordering::SeqCst) {


Test patch:
diff --git a/tests/all.rs b/tests/all.rs
index bdd67315..1711e4ee 100644
--- a/tests/all.rs
+++ b/tests/all.rs
@@ -356,8 +356,11 @@ fn select2() {
 
 #[test]
 fn option() {
-    assert_eq!(Ok(Some(())), Some(ok::<(), ()>(())).wait());
-    assert_eq!(Ok(None), <Option<FutureResult<(), ()>> as Future>::wait(None));
+    let some = Some(ok::<(), ()>(()));
+    let none: Option<FutureResult<(), ()>> = None;
+
+    assert_eq!(Ok(Some(())), blocking(some).wait());
+    assert_eq!(Ok(None), blocking(none).wait());
 }
 
 #[test]
diff --git a/tests/bilock.rs b/tests/bilock.rs
index 78d87363..61c8bb5b 100644
--- a/tests/bilock.rs
+++ b/tests/bilock.rs
@@ -5,7 +5,7 @@ use std::thread;
 use futures::prelude::*;
 use futures::executor;
 use futures::stream;
-use futures::future;
+use futures::future::{self, blocking};
 use futures::sync::BiLock;
 
 mod support;
@@ -66,8 +66,8 @@ fn concurrent() {
         })
     });
 
-    let t1 = thread::spawn(move || a.wait());
-    let b = b.wait().expect("b error");
+    let t1 = thread::spawn(move || blocking(a).wait());
+    let b = blocking(b).wait().expect("b error");
     let a = t1.join().unwrap().expect("a error");
 
     match a.poll_lock() {
diff --git a/tests/blocking.rs b/tests/blocking.rs
new file mode 100644
index 00000000..371fc720
--- /dev/null
+++ b/tests/blocking.rs
@@ -0,0 +1,16 @@
+extern crate futures;
+
+use futures::future::blocking;
+use futures::unsync::oneshot;
+
+#[test]
+fn future_try_take() {
+    let (tx, rx) = oneshot::channel::<u32>();
+    let mut rx = blocking(rx);
+
+    assert!(rx.try_take().is_none());
+
+    tx.send(1).unwrap();
+
+    assert_eq!(Some(Ok(1)), rx.try_take());
+}
diff --git a/tests/buffer_unordered.rs b/tests/buffer_unordered.rs
index 005bbd98..e3335cd3 100644
--- a/tests/buffer_unordered.rs
+++ b/tests/buffer_unordered.rs
@@ -3,7 +3,9 @@ extern crate futures;
 use std::sync::mpsc as std_mpsc;
 use std::thread;
 
+use futures::future::blocking;
 use futures::prelude::*;
+use futures::stream;
 use futures::sync::oneshot;
 use futures::sync::mpsc;
 
@@ -18,20 +20,20 @@ fn works() {
     let t1 = thread::spawn(move || {
         for _ in 0..N+1 {
             let (mytx, myrx) = oneshot::channel();
-            tx = tx.send(myrx).wait().unwrap();
+            tx = blocking(tx.send(myrx)).wait().unwrap();
             tx3.send(mytx).unwrap();
         }
         rx2.recv().unwrap();
         for _ in 0..N {
             let (mytx, myrx) = oneshot::channel();
-            tx = tx.send(myrx).wait().unwrap();
+            tx = blocking(tx.send(myrx)).wait().unwrap();
             tx3.send(mytx).unwrap();
         }
     });
 
     let (tx4, rx4) = std_mpsc::channel();
     let t2 = thread::spawn(move || {
-        for item in rx.map_err(|_| panic!()).buffer_unordered(N).wait() {
+        for item in stream::blocking(rx.map_err(|_| panic!()).buffer_unordered(N)) {
             tx4.send(item.unwrap()).unwrap();
         }
     });
diff --git a/tests/channel.rs b/tests/channel.rs
index 58c611b5..eba81b11 100644
--- a/tests/channel.rs
+++ b/tests/channel.rs
@@ -3,7 +3,8 @@ extern crate futures;
 use std::sync::atomic::*;
 
 use futures::prelude::*;
-use futures::future::result;
+use futures::future::{blocking, result};
+use futures::stream;
 use futures::sync::mpsc;
 
 mod support;
@@ -18,7 +19,7 @@ fn sequence() {
 
     let amt = 20;
     send(amt, tx).forget();
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
     for i in (1..amt + 1).rev() {
         assert_eq!(rx.next(), Some(Ok(i)));
     }
@@ -45,9 +46,9 @@ fn drop_sender() {
 #[test]
 fn drop_rx() {
     let (tx, rx) = mpsc::channel::<u32>(1);
-    let tx = tx.send(1).wait().ok().unwrap();
+    let tx = blocking(tx.send(1)).wait().ok().unwrap();
     drop(rx);
-    assert!(tx.send(1).wait().is_err());
+    assert!(blocking(tx.send(1)).wait().is_err());
 }
 
 #[test]
@@ -63,10 +64,10 @@ fn drop_order() {
         }
     }
 
-    let tx = tx.send(A).wait().unwrap();
+    let tx = blocking(tx.send(A)).wait().unwrap();
     assert_eq!(DROPS.load(Ordering::SeqCst), 0);
     drop(rx);
     assert_eq!(DROPS.load(Ordering::SeqCst), 1);
-    assert!(tx.send(A).wait().is_err());
+    assert!(blocking(tx.send(A)).wait().is_err());
     assert_eq!(DROPS.load(Ordering::SeqCst), 2);
 }
diff --git a/tests/future_flatten_stream.rs b/tests/future_flatten_stream.rs
index 442d381f..1d071fe0 100644
--- a/tests/future_flatten_stream.rs
+++ b/tests/future_flatten_stream.rs
@@ -14,7 +14,7 @@ fn successful_future() {
 
     let stream = future_of_a_stream.flatten_stream();
 
-    let mut iter = stream.wait();
+    let mut iter = stream::blocking(stream);
     assert_eq!(Ok(17), iter.next().unwrap());
     assert_eq!(Ok(19), iter.next().unwrap());
     assert_eq!(None, iter.next());
@@ -37,7 +37,7 @@ impl<T, E> Stream for PanickingStream<T, E> {
 fn failed_future() {
     let future_of_a_stream = err::<PanickingStream<bool, u32>, _>(10);
     let stream = future_of_a_stream.flatten_stream();
-    let mut iter = stream.wait();
+    let mut iter = stream::blocking(stream);
     assert_eq!(Err(10), iter.next().unwrap());
     assert_eq!(None, iter.next());
 }
diff --git a/tests/futures_ordered.rs b/tests/futures_ordered.rs
index 229a8e58..6559aef7 100644
--- a/tests/futures_ordered.rs
+++ b/tests/futures_ordered.rs
@@ -2,6 +2,7 @@ extern crate futures;
 
 use std::any::Any;
 
+use futures::future::blocking;
 use futures::sync::oneshot;
 use futures::stream::futures_ordered;
 use futures::prelude::*;
@@ -59,7 +60,7 @@ fn from_iterator() {
         ok::<u32, ()>(3)
     ].into_iter().collect::<FuturesOrdered<_>>();
     assert_eq!(stream.len(), 3);
-    assert_eq!(stream.collect().wait(), Ok(vec![1,2,3]));
+    assert_eq!(blocking(stream.collect()).wait(), Ok(vec![1,2,3]));
 }
 
 #[test]
diff --git a/tests/futures_unordered.rs b/tests/futures_unordered.rs
index 9b8c08d0..7e23291c 100644
--- a/tests/futures_unordered.rs
+++ b/tests/futures_unordered.rs
@@ -2,8 +2,9 @@ extern crate futures;
 
 use std::any::Any;
 
+use futures::future::blocking;
 use futures::sync::oneshot;
-use futures::stream::futures_unordered;
+use futures::stream::{self, futures_unordered};
 use futures::prelude::*;
 
 mod support;
@@ -16,15 +17,15 @@ fn works_1() {
 
     let stream = futures_unordered(vec![a_rx, b_rx, c_rx]);
 
-    let mut spawn = futures::executor::spawn(stream);
+    let mut stream = stream::blocking(stream);
     b_tx.send(99).unwrap();
-    assert_eq!(Some(Ok(99)), spawn.wait_stream());
+    assert_eq!(Some(Ok(99)), stream.next());
 
     a_tx.send(33).unwrap();
     c_tx.send(33).unwrap();
-    assert_eq!(Some(Ok(33)), spawn.wait_stream());
-    assert_eq!(Some(Ok(33)), spawn.wait_stream());
-    assert_eq!(None, spawn.wait_stream());
+    assert_eq!(Some(Ok(33)), stream.next());
+    assert_eq!(Some(Ok(33)), stream.next());
+    assert_eq!(None, stream.next());
 }
 
 #[test]
@@ -57,7 +58,7 @@ fn from_iterator() {
         ok::<u32, ()>(3)
     ].into_iter().collect::<FuturesUnordered<_>>();
     assert_eq!(stream.len(), 3);
-    assert_eq!(stream.collect().wait(), Ok(vec![1,2,3]));
+    assert_eq!(blocking(stream.collect()).wait(), Ok(vec![1,2,3]));
 }
 
 #[test]
@@ -100,11 +101,11 @@ fn iter_mut_cancel() {
     assert!(b_tx.is_canceled());
     assert!(c_tx.is_canceled());
 
-    let mut spawn = futures::executor::spawn(stream);
-    assert_eq!(Some(Err(futures::sync::oneshot::Canceled)), spawn.wait_stream());
-    assert_eq!(Some(Err(futures::sync::oneshot::Canceled)), spawn.wait_stream());
-    assert_eq!(Some(Err(futures::sync::oneshot::Canceled)), spawn.wait_stream());
-    assert_eq!(None, spawn.wait_stream());
+    let mut stream = stream::blocking(stream);
+    assert_eq!(Some(Err(futures::sync::oneshot::Canceled)), stream.next());
+    assert_eq!(Some(Err(futures::sync::oneshot::Canceled)), stream.next());
+    assert_eq!(Some(Err(futures::sync::oneshot::Canceled)), stream.next());
+    assert_eq!(None, stream.next());
 }
 
 #[test]
diff --git a/tests/inspect.rs b/tests/inspect.rs
index c16372ed..adc81988 100644
--- a/tests/inspect.rs
+++ b/tests/inspect.rs
@@ -1,7 +1,7 @@
 extern crate futures;
 
 use futures::prelude::*;
-use futures::future::{ok, err};
+use futures::future::{blocking, ok, err};
 
 #[test]
 fn smoke() {
@@ -9,14 +9,14 @@ fn smoke() {
 
     {
         let work = ok::<u32, u32>(40).inspect(|val| { counter += *val; });
-        assert_eq!(work.wait(), Ok(40));
+        assert_eq!(blocking(work).wait(), Ok(40));
     }
 
     assert_eq!(counter, 40);
 
     {
         let work = err::<u32, u32>(4).inspect(|val| { counter += *val; });
-        assert_eq!(work.wait(), Err(4));
+        assert_eq!(blocking(work).wait(), Err(4));
     }
 
     assert_eq!(counter, 40);
diff --git a/tests/mpsc-close.rs b/tests/mpsc-close.rs
index 253e0157..0e8f9ad7 100644
--- a/tests/mpsc-close.rs
+++ b/tests/mpsc-close.rs
@@ -2,6 +2,7 @@ extern crate futures;
 
 use std::thread;
 
+use futures::future::blocking;
 use futures::prelude::*;
 use futures::sync::mpsc::*;
 
@@ -10,12 +11,12 @@ fn smoke() {
     let (mut sender, receiver) = channel(1);
 
     let t = thread::spawn(move ||{
-        while let Ok(s) = sender.send(42).wait() {
+        while let Ok(s) = blocking(sender.send(42)).wait() {
             sender = s;
         }
     });
 
-    receiver.take(3).for_each(|_| Ok(())).wait().unwrap();
+    blocking(receiver.take(3).for_each(|_| Ok(()))).wait().unwrap();
 
     t.join().unwrap()
 }
diff --git a/tests/mpsc.rs b/tests/mpsc.rs
index a5f7e55e..d2cb9104 100644
--- a/tests/mpsc.rs
+++ b/tests/mpsc.rs
@@ -4,8 +4,8 @@
 extern crate futures;
 
 use futures::prelude::*;
-use futures::future::{lazy, ok};
-use futures::stream::unfold;
+use futures::future::{blocking, lazy, ok};
+use futures::stream::{self, unfold};
 use futures::sync::mpsc;
 
 use std::time::Duration;
@@ -24,9 +24,9 @@ impl AssertSend for mpsc::Receiver<i32> {}
 #[test]
 fn send_recv() {
     let (tx, rx) = mpsc::channel::<i32>(16);
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
 
-    tx.send(1).wait().unwrap();
+    blocking(tx.send(1)).wait().unwrap();
 
     assert_eq!(rx.next().unwrap(), Ok(1));
 }
@@ -36,7 +36,7 @@ fn send_recv_no_buffer() {
     let (mut tx, mut rx) = mpsc::channel::<i32>(0);
 
     // Run on a task context
-    lazy(move || {
+    blocking(lazy(move || {
         assert!(tx.poll_complete().unwrap().is_ready());
         assert!(tx.poll_ready().unwrap().is_ready());
 
@@ -62,29 +62,29 @@ fn send_recv_no_buffer() {
         assert!(tx.poll_ready().unwrap().is_ready());
 
         Ok::<(), ()>(())
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 #[test]
 fn send_shared_recv() {
     let (tx1, rx) = mpsc::channel::<i32>(16);
     let tx2 = tx1.clone();
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
 
-    tx1.send(1).wait().unwrap();
+    blocking(tx1.send(1)).wait().unwrap();
     assert_eq!(rx.next().unwrap(), Ok(1));
 
-    tx2.send(2).wait().unwrap();
+    blocking(tx2.send(2)).wait().unwrap();
     assert_eq!(rx.next().unwrap(), Ok(2));
 }
 
 #[test]
 fn send_recv_threads() {
     let (tx, rx) = mpsc::channel::<i32>(16);
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
 
     thread::spawn(move|| {
-        tx.send(1).wait().unwrap();
+        blocking(tx.send(1)).wait().unwrap();
     });
 
     assert_eq!(rx.next().unwrap(), Ok(1));
@@ -93,11 +93,11 @@ fn send_recv_threads() {
 #[test]
 fn send_recv_threads_no_capacity() {
     let (mut tx, rx) = mpsc::channel::<i32>(0);
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
 
     let t = thread::spawn(move|| {
-        tx = tx.send(1).wait().unwrap();
-        tx = tx.send(2).wait().unwrap();
+        tx = blocking(tx.send(1)).wait().unwrap();
+        tx = blocking(tx.send(2)).wait().unwrap();
     });
 
     thread::sleep(Duration::from_millis(100));
@@ -114,7 +114,7 @@ fn recv_close_gets_none() {
     let (mut tx, mut rx) = mpsc::channel::<i32>(10);
 
     // Run on a task context
-    lazy(move || {
+    blocking(lazy(move || {
         rx.close();
 
         assert_eq!(rx.poll(), Ok(Async::Ready(None)));
@@ -123,7 +123,7 @@ fn recv_close_gets_none() {
         drop(tx);
 
         Ok::<(), ()>(())
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 
@@ -132,12 +132,12 @@ fn tx_close_gets_none() {
     let (_, mut rx) = mpsc::channel::<i32>(10);
 
     // Run on a task context
-    lazy(move || {
+    blocking(lazy(move || {
         assert_eq!(rx.poll(), Ok(Async::Ready(None)));
         assert_eq!(rx.poll(), Ok(Async::Ready(None)));
 
         Ok::<(), ()>(())
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 #[test]
@@ -154,7 +154,7 @@ fn stress_shared_unbounded() {
     const AMT: u32 = 10000;
     const NTHREADS: u32 = 8;
     let (tx, rx) = mpsc::unbounded::<i32>();
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
 
     let t = thread::spawn(move|| {
         for _ in 0..AMT * NTHREADS {
@@ -186,7 +186,7 @@ fn stress_shared_bounded_hard() {
     const AMT: u32 = 10000;
     const NTHREADS: u32 = 8;
     let (tx, rx) = mpsc::channel::<i32>(0);
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
 
     let t = thread::spawn(move|| {
         for _ in 0..AMT * NTHREADS {
@@ -203,7 +203,7 @@ fn stress_shared_bounded_hard() {
 
         thread::spawn(move|| {
             for _ in 0..AMT {
-                tx = tx.send(1).wait().unwrap();
+                tx = blocking(tx.send(1)).wait().unwrap();
             }
         });
     }
@@ -238,7 +238,7 @@ fn stress_receiver_multi_task_bounded_hard() {
                 match lock.take() {
                     Some(mut rx) => {
                         if i % 5 == 0 {
-                            let (item, rest) = rx.into_future().wait().ok().unwrap();
+                            let (item, rest) = blocking(rx.into_future()).wait().ok().unwrap();
 
                             if item.is_none() {
                                 break;
@@ -249,7 +249,7 @@ fn stress_receiver_multi_task_bounded_hard() {
                         } else {
                             // Just poll
                             let n = n.clone();
-                            let r = lazy(move || {
+                            let r = blocking(lazy(move || {
                                 let r = match rx.poll().unwrap() {
                                     Async::Ready(Some(_)) => {
                                         n.fetch_add(1, Ordering::Relaxed);
@@ -266,7 +266,7 @@ fn stress_receiver_multi_task_bounded_hard() {
                                 };
 
                                 Ok::<bool, ()>(r)
-                            }).wait().unwrap();
+                            })).wait().unwrap();
 
                             if r {
                                 break;
@@ -282,7 +282,7 @@ fn stress_receiver_multi_task_bounded_hard() {
     }
 
     for i in 0..AMT {
-        tx = tx.send(i).wait().unwrap();
+        tx = blocking(tx.send(i)).wait().unwrap();
     }
 
     drop(tx);
@@ -308,7 +308,7 @@ fn stress_drop_sender() {
     }
 
     for _ in 0..10000 {
-        assert_eq!(list().wait().collect::<Result<Vec<_>, _>>(),
+        assert_eq!(stream::blocking(list()).collect::<Result<Vec<_>, _>>(),
         Ok(vec![1, 2, 3]));
     }
 }
@@ -327,7 +327,7 @@ fn stress_close_receiver_iter() {
         }
     });
 
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
 
     // Read one message to make sure thread effectively started
     assert_eq!(Some(Ok(1)), rx.next());
@@ -389,15 +389,15 @@ fn stress_poll_ready() {
         for _ in 0..NTHREADS {
             let sender = tx.clone();
             threads.push(thread::spawn(move || {
-                SenderTask {
+                blocking(SenderTask {
                     sender: sender,
                     count: AMT,
-                }.wait()
+                }).wait()
             }));
         }
         drop(tx);
 
-        let mut rx = rx.wait();
+        let mut rx = stream::blocking(rx);
         for _ in 0..AMT * NTHREADS {
             assert!(rx.next().is_some());
         }
@@ -436,7 +436,7 @@ fn try_send_1() {
             }
         }
     });
-    for (i, j) in rx.wait().enumerate() {
+    for (i, j) in stream::blocking(rx).enumerate() {
         assert_eq!(i, j.unwrap());
     }
     t.join().unwrap();
@@ -452,18 +452,18 @@ fn try_send_2() {
     tx.try_send("hello").unwrap();
 
     let th = thread::spawn(|| {
-        lazy(|| {
+        blocking(lazy(|| {
             assert!(tx.start_send("fail").unwrap().is_not_ready());
             Ok::<_, ()>(())
-        }).wait().unwrap();
+        })).wait().unwrap();
 
-        tx.send("goodbye").wait().unwrap();
+        blocking(tx.send("goodbye")).wait().unwrap();
     });
 
     // Little sleep to hopefully get the action on the thread to happen first
     thread::sleep(Duration::from_millis(300));
 
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
 
     assert_eq!(rx.next(), Some(Ok("hello")));
     assert_eq!(rx.next(), Some(Ok("goodbye")));
@@ -475,7 +475,7 @@ fn try_send_2() {
 #[test]
 fn try_send_fail() {
     let (mut tx, rx) = mpsc::channel(0);
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
 
     tx.try_send("hello").unwrap();
 
diff --git a/tests/oneshot.rs b/tests/oneshot.rs
index fabc4f06..3b19a990 100644
--- a/tests/oneshot.rs
+++ b/tests/oneshot.rs
@@ -4,7 +4,7 @@ use std::sync::mpsc;
 use std::thread;
 
 use futures::prelude::*;
-use futures::future::{lazy, ok};
+use futures::future::{blocking, lazy, ok};
 use futures::sync::oneshot::*;
 
 mod support;
@@ -85,7 +85,7 @@ fn close_wakes() {
         rx.close();
         rx2.recv().unwrap();
     });
-    WaitForCancel { tx: tx }.wait().unwrap();
+    blocking(WaitForCancel { tx: tx }).wait().unwrap();
     tx2.send(()).unwrap();
     t.join().unwrap();
 }
diff --git a/tests/ready_queue.rs b/tests/ready_queue.rs
index b0dc2375..2e9f3df9 100644
--- a/tests/ready_queue.rs
+++ b/tests/ready_queue.rs
@@ -4,8 +4,8 @@ use std::panic::{self, AssertUnwindSafe};
 
 use futures::prelude::*;
 use futures::Async::*;
-use futures::future;
-use futures::stream::FuturesUnordered;
+use futures::future::{self, blocking};
+use futures::stream::{self, FuturesUnordered};
 use futures::sync::oneshot;
 
 trait AssertSendSync: Send + Sync {}
@@ -13,7 +13,7 @@ impl AssertSendSync for FuturesUnordered<()> {}
 
 #[test]
 fn basic_usage() {
-    future::lazy(move || {
+    blocking(future::lazy(move || {
         let mut queue = FuturesUnordered::new();
         let (tx1, rx1) = oneshot::channel();
         let (tx2, rx2) = oneshot::channel();
@@ -38,12 +38,12 @@ fn basic_usage() {
         assert_eq!(Ready(None), queue.poll().unwrap());
 
         Ok::<_, ()>(())
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 #[test]
 fn resolving_errors() {
-    future::lazy(move || {
+    blocking(future::lazy(move || {
         let mut queue = FuturesUnordered::new();
         let (tx1, rx1) = oneshot::channel();
         let (tx2, rx2) = oneshot::channel();
@@ -68,12 +68,12 @@ fn resolving_errors() {
         assert_eq!(Ready(None), queue.poll().unwrap());
 
         Ok::<_, ()>(())
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 #[test]
 fn dropping_ready_queue() {
-    future::lazy(move || {
+    blocking(future::lazy(move || {
         let mut queue = FuturesUnordered::new();
         let (mut tx1, rx1) = oneshot::channel::<()>();
         let (mut tx2, rx2) = oneshot::channel::<()>();
@@ -94,7 +94,7 @@ fn dropping_ready_queue() {
         assert!(tx3.poll_cancel().unwrap().is_ready());
 
         Ok::<_, ()>(())
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 #[test]
@@ -126,7 +126,7 @@ fn stress() {
 
             barrier.wait();
 
-            let mut sync = queue.wait();
+            let mut sync = stream::blocking(queue);
 
             let mut rx: Vec<_> = (&mut sync)
                 .take(n)
@@ -148,7 +148,7 @@ fn stress() {
 
 #[test]
 fn panicking_future_dropped() {
-    future::lazy(move || {
+    blocking(future::lazy(move || {
         let mut queue = FuturesUnordered::new();
         queue.push(future::poll_fn(|| -> Poll<i32, i32> {
             panic!()
@@ -160,5 +160,5 @@ fn panicking_future_dropped() {
         assert_eq!(Ready(None), queue.poll().unwrap());
 
         Ok::<_, ()>(())
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
diff --git a/tests/recurse.rs b/tests/recurse.rs
index 4eb024ac..c7010ba6 100644
--- a/tests/recurse.rs
+++ b/tests/recurse.rs
@@ -2,7 +2,7 @@ extern crate futures;
 
 use std::sync::mpsc::channel;
 
-use futures::future::ok;
+use futures::future::{blocking, ok};
 use futures::prelude::*;
 
 #[test]
@@ -17,7 +17,7 @@ fn lots() {
 
     let (tx, rx) = channel();
     ::std::thread::spawn(|| {
-        doit(1_000).map(move |_| tx.send(()).unwrap()).wait()
+        blocking(doit(1_000).map(move |_| tx.send(()).unwrap())).wait()
     });
     rx.recv().unwrap();
 }
diff --git a/tests/select_all.rs b/tests/select_all.rs
index 7780aa30..552f2489 100644
--- a/tests/select_all.rs
+++ b/tests/select_all.rs
@@ -1,7 +1,6 @@
 extern crate futures;
 
-use futures::prelude::*;
-use futures::future::{ok, select_all, err};
+use futures::future::{blocking, ok, select_all, err};
 
 #[test]
 fn smoke() {
@@ -11,15 +10,15 @@ fn smoke() {
         ok(3),
     ];
 
-    let (i, idx, v) = select_all(v).wait().ok().unwrap();
+    let (i, idx, v) = blocking(select_all(v)).wait().ok().unwrap();
     assert_eq!(i, 1);
     assert_eq!(idx, 0);
 
-    let (i, idx, v) = select_all(v).wait().err().unwrap();
+    let (i, idx, v) = blocking(select_all(v)).wait().err().unwrap();
     assert_eq!(i, 2);
     assert_eq!(idx, 0);
 
-    let (i, idx, v) = select_all(v).wait().ok().unwrap();
+    let (i, idx, v) = blocking(select_all(v)).wait().ok().unwrap();
     assert_eq!(i, 3);
     assert_eq!(idx, 0);
 
diff --git a/tests/select_ok.rs b/tests/select_ok.rs
index 85f39e2d..c823e878 100644
--- a/tests/select_ok.rs
+++ b/tests/select_ok.rs
@@ -11,12 +11,12 @@ fn ignore_err() {
         ok(4),
     ];
 
-    let (i, v) = select_ok(v).wait().ok().unwrap();
+    let (i, v) = blocking(select_ok(v)).wait().ok().unwrap();
     assert_eq!(i, 3);
 
     assert_eq!(v.len(), 1);
 
-    let (i, v) = select_ok(v).wait().ok().unwrap();
+    let (i, v) = blocking(select_ok(v)).wait().ok().unwrap();
     assert_eq!(i, 4);
 
     assert!(v.is_empty());
@@ -30,11 +30,11 @@ fn last_err() {
         err(3),
     ];
 
-    let (i, v) = select_ok(v).wait().ok().unwrap();
+    let (i, v) = blocking(select_ok(v)).wait().ok().unwrap();
     assert_eq!(i, 1);
 
     assert_eq!(v.len(), 2);
 
-    let i = select_ok(v).wait().err().unwrap();
+    let i = blocking(select_ok(v)).wait().err().unwrap();
     assert_eq!(i, 3);
 }
diff --git a/tests/shared.rs b/tests/shared.rs
index 99d2b381..77b4e0f4 100644
--- a/tests/shared.rs
+++ b/tests/shared.rs
@@ -8,7 +8,7 @@ use std::thread;
 
 use futures::sync::oneshot;
 use futures::prelude::*;
-use futures::future;
+use futures::future::{self, blocking};
 
 fn send_shared_oneshot_and_wait_on_multiple_threads(threads_number: u32) {
     let (tx, rx) = oneshot::channel::<u32>();
@@ -16,11 +16,11 @@ fn send_shared_oneshot_and_wait_on_multiple_threads(threads_number: u32) {
     let threads = (0..threads_number).map(|_| {
         let cloned_future = f.clone();
         thread::spawn(move || {
-            assert_eq!(*cloned_future.wait().unwrap(), 6);
+            assert_eq!(*blocking(cloned_future).wait().unwrap(), 6);
         })
     }).collect::<Vec<_>>();
     tx.send(6).unwrap();
-    assert_eq!(*f.wait().unwrap(), 6);
+    assert_eq!(*blocking(f).wait().unwrap(), 6);
     for f in threads {
         f.join().unwrap();
     }
@@ -51,20 +51,20 @@ fn drop_on_one_task_ok() {
 
     let t1 = thread::spawn(|| {
         let f = f1.map_err(|_| ()).map(|x| *x).select(rx2.map_err(|_| ()));
-        drop(f.wait());
+        drop(blocking(f).wait());
     });
 
     let (tx3, rx3) = oneshot::channel::<u32>();
 
     let t2 = thread::spawn(|| {
-        let _ = f2.map(|x| tx3.send(*x).unwrap()).map_err(|_| ()).wait();
+        let _ = blocking(f2.map(|x| tx3.send(*x).unwrap()).map_err(|_| ())).wait();
     });
 
     tx2.send(11).unwrap(); // cancel `f1`
     t1.join().unwrap();
 
     tx.send(42).unwrap(); // Should cause `f2` and then `rx3` to get resolved.
-    let result = rx3.wait().unwrap();
+    let result = blocking(rx3).wait().unwrap();
     assert_eq!(result, 42);
     t2.join().unwrap();
 }
@@ -79,7 +79,7 @@ fn drop_in_poll() {
     }).shared();
     let future2 = Box::new(future.clone()) as Box<Future<Item=_, Error=_>>;
     *slot.borrow_mut() = Some(future2);
-    assert_eq!(*future.wait().unwrap(), 1);
+    assert_eq!(*blocking(future).wait().unwrap(), 1);
 }
 
 #[test]
diff --git a/tests/sink.rs b/tests/sink.rs
index 6d940934..5aa5848a 100644
--- a/tests/sink.rs
+++ b/tests/sink.rs
@@ -7,7 +7,7 @@ use std::cell::{Cell, RefCell};
 use std::sync::atomic::{Ordering, AtomicBool};
 
 use futures::prelude::*;
-use futures::future::ok;
+use futures::future::{blocking, ok};
 use futures::stream;
 use futures::sync::{oneshot, mpsc};
 use futures::task::{self, Task};
@@ -30,10 +30,10 @@ fn vec_sink() {
 fn send() {
     let v = Vec::new();
 
-    let v = v.send(0).wait().unwrap();
+    let v = blocking(v.send(0)).wait().unwrap();
     assert_eq!(v, vec![0]);
 
-    let v = v.send(1).wait().unwrap();
+    let v = blocking(v.send(1)).wait().unwrap();
     assert_eq!(v, vec![0, 1]);
 
     assert_done(move || v.send(2),
@@ -44,10 +44,10 @@ fn send() {
 fn send_all() {
     let v = Vec::new();
 
-    let (v, _) = v.send_all(stream::iter_ok(vec![0, 1])).wait().unwrap();
+    let (v, _) = blocking(v.send_all(stream::iter_ok(vec![0, 1]))).wait().unwrap();
     assert_eq!(v, vec![0, 1]);
 
-    let (v, _) = v.send_all(stream::iter_ok(vec![2, 3])).wait().unwrap();
+    let (v, _) = blocking(v.send_all(stream::iter_ok(vec![2, 3]))).wait().unwrap();
     assert_eq!(v, vec![0, 1, 2, 3]);
 
     assert_done(
@@ -109,7 +109,7 @@ impl<S: Sink> Future for StartSendFut<S> {
 fn mpsc_blocking_start_send() {
     let (mut tx, mut rx) = mpsc::channel::<i32>(0);
 
-    futures::future::lazy(|| {
+    blocking(futures::future::lazy(|| {
         assert_eq!(tx.start_send(0).unwrap(), AsyncSink::Ready);
 
         let flag = Flag::new();
@@ -125,7 +125,7 @@ fn mpsc_blocking_start_send() {
         sassert_next(&mut rx, 1);
 
         Ok::<(), ()>(())
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 #[test]
@@ -152,7 +152,7 @@ fn with_flush() {
         _ => panic!()
     };
 
-    assert_eq!(sink.send(1).wait().unwrap().get_ref(), &[1, 2]);
+    assert_eq!(blocking(sink.send(1)).wait().unwrap().get_ref(), &[1, 2]);
 }
 
 #[test]
@@ -161,9 +161,9 @@ fn with_as_map() {
     let sink = Vec::new().with(|item| -> Result<i32, ()> {
         Ok(item * 2)
     });
-    let sink = sink.send(0).wait().unwrap();
-    let sink = sink.send(1).wait().unwrap();
-    let sink = sink.send(2).wait().unwrap();
+    let sink = blocking(sink.send(0)).wait().unwrap();
+    let sink = blocking(sink.send(1)).wait().unwrap();
+    let sink = blocking(sink.send(2)).wait().unwrap();
     assert_eq!(sink.get_ref(), &[0, 2, 4]);
 }
 
@@ -173,10 +173,10 @@ fn with_flat_map() {
     let sink = Vec::new().with_flat_map(|item| {
         stream::iter_ok(vec![item; item])
     });
-    let sink = sink.send(0).wait().unwrap();
-    let sink = sink.send(1).wait().unwrap();
-    let sink = sink.send(2).wait().unwrap();
-    let sink = sink.send(3).wait().unwrap();
+    let sink = blocking(sink.send(0)).wait().unwrap();
+    let sink = blocking(sink.send(1)).wait().unwrap();
+    let sink = blocking(sink.send(2)).wait().unwrap();
+    let sink = blocking(sink.send(3)).wait().unwrap();
     assert_eq!(sink.get_ref(), &[1,2,2,3,3,3]);
 }
 
@@ -251,13 +251,13 @@ fn with_flush_propagate() {
 // test that a buffer is a no-nop around a sink that always accepts sends
 fn buffer_noop() {
     let sink = Vec::new().buffer(0);
-    let sink = sink.send(0).wait().unwrap();
-    let sink = sink.send(1).wait().unwrap();
+    let sink = blocking(sink.send(0)).wait().unwrap();
+    let sink = blocking(sink.send(1)).wait().unwrap();
     assert_eq!(sink.get_ref(), &[0, 1]);
 
     let sink = Vec::new().buffer(1);
-    let sink = sink.send(0).wait().unwrap();
-    let sink = sink.send(1).wait().unwrap();
+    let sink = blocking(sink.send(0)).wait().unwrap();
+    let sink = blocking(sink.send(1)).wait().unwrap();
     assert_eq!(sink.get_ref(), &[0, 1]);
 }
 
@@ -335,8 +335,8 @@ fn buffer() {
     let (sink, allow) = manual_allow::<i32>();
     let sink = sink.buffer(2);
 
-    let sink = StartSendFut::new(sink, 0).wait().unwrap();
-    let sink = StartSendFut::new(sink, 1).wait().unwrap();
+    let sink = blocking(StartSendFut::new(sink, 0)).wait().unwrap();
+    let sink = blocking(StartSendFut::new(sink, 1)).wait().unwrap();
 
     let flag = Flag::new();
     let mut task = executor::spawn(sink.send(2));
diff --git a/tests/split.rs b/tests/split.rs
index 7a0667f1..0eb02122 100644
--- a/tests/split.rs
+++ b/tests/split.rs
@@ -1,5 +1,6 @@
 extern crate futures;
 
+use futures::future::blocking;
 use futures::prelude::*;
 use futures::stream::iter_ok;
 
@@ -41,7 +42,7 @@ fn test_split() {
         let (sink, stream) = j.split();
         let j = sink.reunite(stream).expect("test_split: reunite error");
         let (sink, stream) = j.split();
-        sink.send_all(stream).wait().unwrap();
+        blocking(sink.send_all(stream)).wait().unwrap();
     }
     assert_eq!(dest, vec![10, 20, 30]);
 }
diff --git a/tests/stream.rs b/tests/stream.rs
index b0e8a361..274dc89e 100644
--- a/tests/stream.rs
+++ b/tests/stream.rs
@@ -2,9 +2,8 @@
 extern crate futures;
 
 use futures::prelude::*;
-use futures::executor;
-use futures::future::{err, ok};
-use futures::stream::{empty, iter_ok, poll_fn, Peekable};
+use futures::future::{blocking, err, ok};
+use futures::stream::{self, empty, iter_ok, poll_fn, Peekable};
 use futures::sync::oneshot;
 use futures::sync::mpsc;
 
@@ -136,9 +135,8 @@ fn skip() {
 
 #[test]
 fn skip_passes_errors_through() {
-    let mut s = iter(vec![Err(1), Err(2), Ok(3), Ok(4), Ok(5)])
-        .skip(1)
-        .wait();
+    let mut s = stream::blocking(iter(vec![Err(1), Err(2), Ok(3), Ok(4), Ok(5)])
+        .skip(1));
     assert_eq!(s.next(), Some(Err(1)));
     assert_eq!(s.next(), Some(Err(2)));
     assert_eq!(s.next(), Some(Ok(4)));
@@ -164,15 +162,14 @@ fn take_while() {
 
 #[test]
 fn take_passes_errors_through() {
-    let mut s = iter(vec![Err(1), Err(2), Ok(3), Ok(4), Err(4)])
-        .take(1)
-        .wait();
+    let mut s = stream::blocking(iter(vec![Err(1), Err(2), Ok(3), Ok(4), Err(4)])
+        .take(1));
     assert_eq!(s.next(), Some(Err(1)));
     assert_eq!(s.next(), Some(Err(2)));
     assert_eq!(s.next(), Some(Ok(3)));
     assert_eq!(s.next(), None);
 
-    let mut s = iter(vec![Ok(1), Err(2)]).take(1).wait();
+    let mut s = stream::blocking(iter(vec![Ok(1), Err(2)]).take(1));
     assert_eq!(s.next(), Some(Ok(1)));
     assert_eq!(s.next(), None);
 }
@@ -184,7 +181,7 @@ fn peekable() {
 
 #[test]
 fn fuse() {
-    let mut stream = list().fuse().wait();
+    let mut stream = stream::blocking(list().fuse());
     assert_eq!(stream.next(), Some(Ok(1)));
     assert_eq!(stream.next(), Some(Ok(2)));
     assert_eq!(stream.next(), Some(Ok(3)));
@@ -208,7 +205,7 @@ fn buffered() {
     c.send(3).unwrap();
     sassert_empty(&mut rx);
     a.send(5).unwrap();
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
     assert_eq!(rx.next(), Some(Ok(5)));
     assert_eq!(rx.next(), Some(Ok(3)));
     assert_eq!(rx.next(), None);
@@ -226,7 +223,7 @@ fn buffered() {
     c.send(3).unwrap();
     sassert_empty(&mut rx);
     a.send(5).unwrap();
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
     assert_eq!(rx.next(), Some(Ok(5)));
     assert_eq!(rx.next(), Some(Ok(3)));
     assert_eq!(rx.next(), None);
@@ -244,7 +241,7 @@ fn unordered() {
 
     let mut rx = rx.buffer_unordered(2);
     sassert_empty(&mut rx);
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
     c.send(3).unwrap();
     assert_eq!(rx.next(), Some(Ok(3)));
     a.send(5).unwrap();
@@ -265,7 +262,7 @@ fn unordered() {
     c.send(3).unwrap();
     sassert_empty(&mut rx);
     a.send(5).unwrap();
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
     assert_eq!(rx.next(), Some(Ok(5)));
     assert_eq!(rx.next(), Some(Ok(3)));
     assert_eq!(rx.next(), None);
@@ -305,14 +302,14 @@ fn peek() {
         }
     }
 
-    Peek {
+    blocking(Peek {
         inner: list().peekable(),
-    }.wait().unwrap()
+    }).wait().unwrap()
 }
 
 #[test]
 fn wait() {
-    assert_eq!(list().wait().collect::<Result<Vec<_>, _>>(),
+    assert_eq!(stream::blocking(list()).collect::<Result<Vec<_>, _>>(),
                Ok(vec![1, 2, 3]));
 }
 
@@ -321,10 +318,10 @@ fn chunks() {
     assert_done(|| list().chunks(3).collect(), Ok(vec![vec![1, 2, 3]]));
     assert_done(|| list().chunks(1).collect(), Ok(vec![vec![1], vec![2], vec![3]]));
     assert_done(|| list().chunks(2).collect(), Ok(vec![vec![1, 2], vec![3]]));
-    let mut list = executor::spawn(err_list().chunks(3));
-    let i = list.wait_stream().unwrap().unwrap();
+    let mut list = stream::blocking(err_list().chunks(3));
+    let i = list.next().unwrap().unwrap();
     assert_eq!(i, vec![1, 2]);
-    let i = list.wait_stream().unwrap().unwrap_err();
+    let i = list.next().unwrap().unwrap_err();
     assert_eq!(i, 3);
 }
 
@@ -352,10 +349,10 @@ fn select() {
 #[test]
 fn forward() {
     let v = Vec::new();
-    let v = iter_ok::<_, ()>(vec![0, 1]).forward(v).wait().unwrap().1;
+    let v = blocking(iter_ok::<_, ()>(vec![0, 1]).forward(v)).wait().unwrap().1;
     assert_eq!(v, vec![0, 1]);
 
-    let v = iter_ok::<_, ()>(vec![2, 3]).forward(v).wait().unwrap().1;
+    let v = blocking(iter_ok::<_, ()>(vec![2, 3]).forward(v)).wait().unwrap().1;
     assert_eq!(v, vec![0, 1, 2, 3]);
 
     assert_done(move || iter_ok(vec![4, 5]).forward(v).map(|(_, s)| s),
@@ -396,5 +393,5 @@ fn stream_poll_fn() {
         Ok(Async::Ready(Some(counter)))
     });
 
-    assert_eq!(read_stream.wait().count(), 5);
+    assert_eq!(stream::blocking(read_stream).count(), 5);
 }
diff --git a/tests/stream_catch_unwind.rs b/tests/stream_catch_unwind.rs
index a06748d0..89382b9e 100644
--- a/tests/stream_catch_unwind.rs
+++ b/tests/stream_catch_unwind.rs
@@ -1,6 +1,6 @@
 extern crate futures;
 
-use futures::stream;
+use futures::stream::{self, blocking};
 use futures::prelude::*;
 
 #[test]
@@ -9,7 +9,7 @@ fn panic_in_the_middle_of_the_stream() {
 
     // panic on second element
     let stream_panicking = stream.map(|o| o.unwrap());
-    let mut iter = stream_panicking.catch_unwind().wait();
+    let mut iter = blocking(stream_panicking.catch_unwind());
 
     assert_eq!(Ok(10), iter.next().unwrap().ok().unwrap());
     assert!(iter.next().unwrap().is_err());
@@ -20,7 +20,7 @@ fn panic_in_the_middle_of_the_stream() {
 fn no_panic() {
     let stream = stream::iter_ok::<_, bool>(vec![10, 11, 12]);
 
-    let mut iter = stream.catch_unwind().wait();
+    let mut iter = blocking(stream.catch_unwind());
 
     assert_eq!(Ok(10), iter.next().unwrap().ok().unwrap());
     assert_eq!(Ok(11), iter.next().unwrap().ok().unwrap());
diff --git a/tests/support/mod.rs b/tests/support/mod.rs
index 29774977..a996f249 100644
--- a/tests/support/mod.rs
+++ b/tests/support/mod.rs
@@ -5,7 +5,7 @@ use std::sync::Arc;
 use std::thread;
 
 use futures::{Future, IntoFuture, Async, Poll};
-use futures::future::FutureResult;
+use futures::future::{blocking, FutureResult};
 use futures::stream::Stream;
 use futures::executor::{self, NotifyHandle, Notify};
 use futures::task;
@@ -23,7 +23,7 @@ pub fn assert_done<T, F>(f: F, result: Result<T::Item, T::Error>)
           T::Error: Eq + fmt::Debug,
           F: FnOnce() -> T,
 {
-    assert_eq!(f().wait(), result);
+    assert_eq!(blocking(f()).wait(), result);
 }
 
 pub fn assert_empty<T: Future, F: FnMut() -> T>(mut f: F) {
@@ -104,7 +104,7 @@ impl<F> ForgetExt for F
           F::Error: Send
 {
     fn forget(self) {
-        thread::spawn(|| self.wait());
+        thread::spawn(|| blocking(self).wait());
     }
 }
 
diff --git a/tests/unsync-oneshot.rs b/tests/unsync-oneshot.rs
index 7f19d830..3df056d9 100644
--- a/tests/unsync-oneshot.rs
+++ b/tests/unsync-oneshot.rs
@@ -1,20 +1,20 @@
 extern crate futures;
 
 use futures::prelude::*;
-use futures::future;
+use futures::future::{self, blocking};
 use futures::unsync::oneshot::{channel, Canceled};
 
 #[test]
 fn smoke() {
     let (tx, rx) = channel();
     tx.send(33).unwrap();
-    assert_eq!(rx.wait().unwrap(), 33);
+    assert_eq!(blocking(rx).wait().unwrap(), 33);
 }
 
 #[test]
 fn canceled() {
     let (_, rx) = channel::<()>();
-    assert_eq!(rx.wait().unwrap_err(), Canceled);
+    assert_eq!(blocking(rx).wait().unwrap_err(), Canceled);
 }
 
 #[test]
@@ -31,7 +31,7 @@ fn tx_complete_rx_unparked() {
         tx.send(55).unwrap();
         Ok(11)
     }));
-    assert_eq!(res.wait().unwrap(), (55, 11));
+    assert_eq!(blocking(res).wait().unwrap(), (55, 11));
 }
 
 #[test]
@@ -42,7 +42,7 @@ fn tx_dropped_rx_unparked() {
         let _tx = tx;
         Ok(11)
     }));
-    assert_eq!(res.wait().unwrap_err(), Canceled);
+    assert_eq!(blocking(res).wait().unwrap_err(), Canceled);
 }
 
 
diff --git a/tests/unsync.rs b/tests/unsync.rs
index 1ad0ed31..72c44a9a 100644
--- a/tests/unsync.rs
+++ b/tests/unsync.rs
@@ -7,17 +7,17 @@ mod support;
 use futures::prelude::*;
 use futures::unsync::oneshot;
 use futures::unsync::mpsc::{self, SendError};
-use futures::future::lazy;
-use futures::stream::{iter_ok, unfold};
+use futures::future::{blocking, lazy};
+use futures::stream::{self, iter_ok, unfold};
 
 use support::local_executor::Core;
 
 #[test]
 fn mpsc_send_recv() {
     let (tx, rx) = mpsc::channel::<i32>(1);
-    let mut rx = rx.wait();
+    let mut rx = stream::blocking(rx);
 
-    tx.send(42).wait().unwrap();
+    blocking(tx.send(42)).wait().unwrap();
 
     assert_eq!(rx.next(), Some(Ok(42)));
     assert_eq!(rx.next(), None);
@@ -27,65 +27,65 @@ fn mpsc_send_recv() {
 fn mpsc_rx_notready() {
     let (_tx, mut rx) = mpsc::channel::<i32>(1);
 
-    lazy(|| {
+    blocking(lazy(|| {
         assert_eq!(rx.poll().unwrap(), Async::NotReady);
         Ok(()) as Result<(), ()>
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 #[test]
 fn mpsc_rx_end() {
     let (_, mut rx) = mpsc::channel::<i32>(1);
 
-    lazy(|| {
+    blocking(lazy(|| {
         assert_eq!(rx.poll().unwrap(), Async::Ready(None));
         Ok(()) as Result<(), ()>
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 #[test]
 fn mpsc_tx_notready() {
     let (tx, _rx) = mpsc::channel::<i32>(1);
-    let tx = tx.send(1).wait().unwrap();
-    lazy(move || {
+    let tx = blocking(tx.send(1)).wait().unwrap();
+    blocking(lazy(move || {
         assert!(tx.send(2).poll().unwrap().is_not_ready());
         Ok(()) as Result<(), ()>
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 #[test]
 fn mpsc_tx_err() {
     let (tx, _) = mpsc::channel::<i32>(1);
-    lazy(move || {
+    blocking(lazy(move || {
         assert!(tx.send(2).poll().is_err());
         Ok(()) as Result<(), ()>
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 #[test]
 fn mpsc_backpressure() {
     let (tx, rx) = mpsc::channel::<i32>(1);
-    lazy(move || {
+    blocking(lazy(move || {
         iter_ok(vec![1, 2, 3])
             .forward(tx)
             .map_err(|e: SendError<i32>| panic!("{}", e))
             .join(rx.take(3).collect().map(|xs| {
                 assert_eq!(xs, [1, 2, 3]);
             }))
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 #[test]
 fn mpsc_unbounded() {
     let (tx, rx) = mpsc::unbounded::<i32>();
-    lazy(move || {
+    blocking(lazy(move || {
         iter_ok(vec![1, 2, 3])
             .forward(tx)
             .map_err(|e: SendError<i32>| panic!("{}", e))
             .join(rx.take(3).collect().map(|xs| {
                 assert_eq!(xs, [1, 2, 3]);
             }))
-    }).wait().unwrap();
+    })).wait().unwrap();
 }
 
 #[test]