Repository: rust-lang-nursery/futures-rs
Pull request #734 (2018-02-06 21:41:30)

Separate futures-rs into workspace crates
The tests, `futures-cpupool`, and `futures-unordered`/executor related things are not working / commented out for now-- we can fix them when `futures-executor` gets integrated back in and the `task::Context` changes are made (or decided against).


Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 87a4fca2..da8f2e3d 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,28 +1,10 @@
-[package]
-name = "futures"
-version = "0.1.18"
-authors = ["Alex Crichton <alex@alexcrichton.com>"]
-license = "MIT/Apache-2.0"
-readme = "README.md"
-keywords = ["futures", "async", "future"]
-repository = "https://github.com/alexcrichton/futures-rs"
-homepage = "https://github.com/alexcrichton/futures-rs"
-documentation = "https://docs.rs/futures"
-description = """
-An implementation of futures and streams featuring zero allocations,
-composability, and iterator-like interfaces.
-"""
-categories = ["asynchronous"]
-
-[badges]
-travis-ci = { repository = "alexcrichton/futures-rs" }
-appveyor = { repository = "alexcrichton/futures-rs" }
-
-[dependencies]
-
-[features]
-std = []
-default = ["std"]
-
 [workspace]
-members = ["futures-cpupool"]
+members = [
+  "futures",
+  "futures-core",
+  "futures-channel",
+  "futures-cpupool",
+  "futures-executor",
+  "futures-util",
+  "futures-sink",
+]
diff --git a/TODO.md b/TODO.md
new file mode 100644
index 00000000..8892738a
--- /dev/null
+++ b/TODO.md
@@ -0,0 +1,4 @@
+# A list of things to be resolved before 1.0 release
+
+- `SpawnHandle` or similar functionality
+- Executors
diff --git a/benches/bilock.rs b/benches/bilock.rs
index 0f840289..f971b91d 100644
--- a/benches/bilock.rs
+++ b/benches/bilock.rs
@@ -54,7 +54,7 @@ impl Stream for LockStream {
     fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
         self.lock.poll().map(|a| match a {
             Async::Ready(a) => Async::Ready(Some(a)),
-            Async::NotReady => Async::NotReady,
+            Async::Pending => Async::Pending,
         })
     }
 }
@@ -76,7 +76,7 @@ fn contended(b: &mut Bencher) {
 
             // Try poll second lock while first lock still holds the lock
             match y.poll_stream_notify(&notify_noop(), 11) {
-                Ok(Async::NotReady) => (),
+                Ok(Async::Pending) => (),
                 _ => panic!(),
             };
 
diff --git a/benches/poll.rs b/benches/poll.rs
index 1fec653f..c39dfd83 100644
--- a/benches/poll.rs
+++ b/benches/poll.rs
@@ -43,7 +43,7 @@ fn task_init(b: &mut Bencher) {
                 if let Some(ref t) = self.task {
                     if t.will_notify_current() {
                         t.notify();
-                        return Ok(Async::NotReady);
+                        return Ok(Async::Pending);
                     }
                 }
 
@@ -51,7 +51,7 @@ fn task_init(b: &mut Bencher) {
                 t.notify();
                 self.task = Some(t);
 
-                Ok(Async::NotReady)
+                Ok(Async::Pending)
             }
         }
     }
@@ -66,7 +66,7 @@ fn task_init(b: &mut Bencher) {
     b.iter(|| {
         fut.get_mut().num = 0;
 
-        while let Ok(Async::NotReady) = fut.poll_future_notify(&notify, 0) {
+        while let Ok(Async::Pending) = fut.poll_future_notify(&notify, 0) {
         }
     });
 }
diff --git a/benches/sync_mpsc.rs b/benches/sync_mpsc.rs
index 34e4dd38..f3edc2c5 100644
--- a/benches/sync_mpsc.rs
+++ b/benches/sync_mpsc.rs
@@ -43,7 +43,7 @@ fn unbounded_1_tx(b: &mut Bencher) {
         for i in 0..1000 {
 
             // Poll, not ready, park
-            assert_eq!(Ok(Async::NotReady), rx.poll_stream_notify(&notify_noop(), 1));
+            assert_eq!(Ok(Async::Pending), rx.poll_stream_notify(&notify_noop(), 1));
 
             UnboundedSender::unbounded_send(&tx, i).unwrap();
 
@@ -66,7 +66,7 @@ fn unbounded_100_tx(b: &mut Bencher) {
         // 1000 send/recv operations total, result should be divided by 1000
         for _ in 0..10 {
             for i in 0..tx.len() {
-                assert_eq!(Ok(Async::NotReady), rx.poll_stream_notify(&notify_noop(), 1));
+                assert_eq!(Ok(Async::Pending), rx.poll_stream_notify(&notify_noop(), 1));
 
                 UnboundedSender::unbounded_send(&tx[i], i).unwrap();
 
@@ -109,8 +109,8 @@ impl Stream for TestSender {
                 assert_eq!(Ok(Async::Ready(())), self.tx.poll_complete());
                 Ok(Async::Ready(Some(self.last)))
             }
-            Ok(AsyncSink::NotReady(_)) => {
-                Ok(Async::NotReady)
+            Ok(AsyncSink::Pending(_)) => {
+                Ok(Async::Pending)
             }
         }
     }
@@ -132,7 +132,7 @@ fn bounded_1_tx(b: &mut Bencher) {
 
         for i in 0..1000 {
             assert_eq!(Ok(Async::Ready(Some(i + 1))), tx.poll_stream_notify(&notify_noop(), 1));
-            assert_eq!(Ok(Async::NotReady), tx.poll_stream_notify(&notify_noop(), 1));
+            assert_eq!(Ok(Async::Pending), tx.poll_stream_notify(&notify_noop(), 1));
             assert_eq!(Ok(Async::Ready(Some(i + 1))), rx.poll_stream_notify(&notify_noop(), 1));
         }
     })
@@ -159,7 +159,7 @@ fn bounded_100_tx(b: &mut Bencher) {
                 // Send an item
                 assert_eq!(Ok(Async::Ready(Some(i + 1))), tx[j].poll_stream_notify(&notify_noop(), 1));
                 // Then block
-                assert_eq!(Ok(Async::NotReady), tx[j].poll_stream_notify(&notify_noop(), 1));
+                assert_eq!(Ok(Async::Pending), tx[j].poll_stream_notify(&notify_noop(), 1));
                 // Recv the item
                 assert_eq!(Ok(Async::Ready(Some(i + 1))), rx.poll_stream_notify(&notify_noop(), 1));
             }
diff --git a/benches/thread_notify.rs b/benches/thread_notify.rs
index 92932353..d36cd7ed 100644
--- a/benches/thread_notify.rs
+++ b/benches/thread_notify.rs
@@ -26,7 +26,7 @@ fn thread_yield_single_thread_one_wait(b: &mut Bencher) {
             } else {
                 self.rem -= 1;
                 task::current().notify();
-                Ok(Async::NotReady)
+                Ok(Async::Pending)
             }
         }
     }
@@ -55,7 +55,7 @@ fn thread_yield_single_thread_many_wait(b: &mut Bencher) {
             } else {
                 self.rem -= 1;
                 task::current().notify();
-                Ok(Async::NotReady)
+                Ok(Async::Pending)
             }
         }
     }
@@ -92,7 +92,7 @@ fn thread_yield_multi_thread(b: &mut Bencher) {
             } else {
                 self.rem -= 1;
                 self.tx.send(task::current()).unwrap();
-                Ok(Async::NotReady)
+                Ok(Async::Pending)
             }
         }
     }
diff --git a/futures-channel/Cargo.toml b/futures-channel/Cargo.toml
new file mode 100644
index 00000000..eeb5a07b
--- /dev/null
+++ b/futures-channel/Cargo.toml
@@ -0,0 +1,17 @@
+[package]
+name = "futures-channel"
+version = "0.1.0"
+authors = ["Alex Crichton <alex@alexcrichton.com>"]
+license = "MIT/Apache-2.0"
+repository = "https://github.com/alexcrichton/futures-rs"
+homepage = "https://github.com/alexcrichton/futures-rs"
+documentation = "https://docs.rs/futures-core"
+description = """
+Channels for asynchronous communication using futures-rs.
+"""
+
+[dependencies.futures-core]
+version = "0.1.0"
+path = "../futures-core"
+default-features = false
+features = ["std"]
diff --git a/futures-channel/src/lib.rs b/futures-channel/src/lib.rs
new file mode 100644
index 00000000..dae0cc75
--- /dev/null
+++ b/futures-channel/src/lib.rs
@@ -0,0 +1,15 @@
+//! Asynchronous channels
+//!
+//! This crate provides channels which can be used to communicate between futures.
+
+#![deny(missing_docs, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/futures/0.2")]
+
+extern crate futures_core;
+
+use futures_core::{Future, Stream, Poll, Async};
+use futures_core::task::{self, Task};
+
+mod lock;
+pub mod mpsc;
+pub mod oneshot;
diff --git a/src/lock.rs b/futures-channel/src/lock.rs
similarity index 100%
rename from src/lock.rs
rename to futures-channel/src/lock.rs
diff --git a/src/sync/mpsc/mod.rs b/futures-channel/src/mpsc/mod.rs
similarity index 82%
rename from src/sync/mpsc/mod.rs
rename to futures-channel/src/mpsc/mod.rs
index 63a17f8c..602650fb 100644
--- a/src/sync/mpsc/mod.rs
+++ b/futures-channel/src/mpsc/mod.rs
@@ -76,13 +76,9 @@ use std::sync::{Arc, Mutex};
 use std::thread;
 use std::usize;
 
-use sync::mpsc::queue::{Queue, PopResult};
-use sync::oneshot;
+use mpsc::queue::{Queue, PopResult};
 use task::{self, Task};
-use future::Executor;
-use sink::SendAll;
-use resultstream::{self, Results};
-use {Async, AsyncSink, Future, Poll, StartSend, Sink, Stream};
+use {Async, Poll, Stream};
 
 mod queue;
 
@@ -415,6 +411,26 @@ impl<T> Sender<T> {
             })
     }
 
+    /// Attempt to start sending a message on the channel.
+    ///
+    /// If there is not room on the channel, this function will return
+    /// `Ok(Err(msg))`, and the current `Task` will be
+    /// awoken when the channel is ready to receive more messages.
+    ///
+    /// On successful completion, this function will return `Ok(Ok(()))`.
+    pub fn start_send(&mut self, msg: T) -> Result<Result<(), T>, SendError<T>> {
+        // If the sender is currently blocked, reject the message before doing
+        // any work.
+        if !self.poll_unparked(true).is_ready() {
+            return Ok(Err(msg));
+        }
+
+        // The channel has capacity to accept the message, so send it.
+        self.do_send(Some(msg), true)?;
+
+        Ok(Ok(()))
+    }
+
     // Do the send without failing
     // None means close
     fn do_send(&mut self, msg: Option<T>, do_park: bool) -> Result<(), SendError<T>> {
@@ -585,7 +601,7 @@ impl<T> Sender<T> {
     /// item without waiting.
     ///
     /// Returns `Ok(Async::Ready(_))` if there is sufficient capacity, or returns
-    /// `Ok(Async::NotReady)` if the channel is not guaranteed to have capacity. Returns
+    /// `Ok(Async::Pending)` if the channel is not guaranteed to have capacity. Returns
     /// `Err(SendError(_))` if the receiver has been dropped.
     ///
     /// # Panics
@@ -624,40 +640,26 @@ impl<T> Sender<T> {
                 None
             };
 
-            Async::NotReady
+            Async::Pending
         } else {
             Async::Ready(())
         }
     }
 }
 
-impl<T> Sink for Sender<T> {
-    type SinkItem = T;
-    type SinkError = SendError<T>;
-
-    fn start_send(&mut self, msg: T) -> StartSend<T, SendError<T>> {
-        // If the sender is currently blocked, reject the message before doing
-        // any work.
-        if !self.poll_unparked(true).is_ready() {
-            return Ok(AsyncSink::NotReady(msg));
-        }
-
-        // The channel has capacity to accept the message, so send it.
-        self.do_send(Some(msg), true)?;
-
-        Ok(AsyncSink::Ready)
-    }
-
-    fn poll_complete(&mut self) -> Poll<(), SendError<T>> {
-        Ok(Async::Ready(()))
+impl<T> UnboundedSender<T> {
+    /// Attempt to start sending a message on the channel.
+    ///
+    /// If there is not room on the channel, this function will return
+    /// `Ok(Err(msg))`, and the current `Task` will be
+    /// awoken when the channel is ready to receive more messages.
+    ///
+    /// On a successful `start_send`, this function will return
+    /// `Ok(Ok(())`.
+    pub fn start_send(&mut self, msg: T) -> Result<Result<(), T>, SendError<T>> {
+        self.0.start_send(msg)
     }
 
-    fn close(&mut self) -> Poll<(), SendError<T>> {
-        Ok(Async::Ready(()))
-    }
-}
-
-impl<T> UnboundedSender<T> {
     /// Sends the provided message along this channel.
     ///
     /// This is an unbounded sender, so this function differs from `Sink::send`
@@ -668,41 +670,6 @@ impl<T> UnboundedSender<T> {
     }
 }
 
-impl<T> Sink for UnboundedSender<T> {
-    type SinkItem = T;
-    type SinkError = SendError<T>;
-
-    fn start_send(&mut self, msg: T) -> StartSend<T, SendError<T>> {
-        self.0.start_send(msg)
-    }
-
-    fn poll_complete(&mut self) -> Poll<(), SendError<T>> {
-        self.0.poll_complete()
-    }
-
-    fn close(&mut self) -> Poll<(), SendError<T>> {
-        Ok(Async::Ready(()))
-    }
-}
-
-impl<'a, T> Sink for &'a UnboundedSender<T> {
-    type SinkItem = T;
-    type SinkError = SendError<T>;
-
-    fn start_send(&mut self, msg: T) -> StartSend<T, SendError<T>> {
-        self.0.do_send_nb(msg)?;
-        Ok(AsyncSink::Ready)
-    }
-
-    fn poll_complete(&mut self) -> Poll<(), SendError<T>> {
-        Ok(Async::Ready(()))
-    }
-
-    fn close(&mut self) -> Poll<(), SendError<T>> {
-        Ok(Async::Ready(()))
-    }
-}
-
 impl<T> Clone for UnboundedSender<T> {
     fn clone(&self) -> UnboundedSender<T> {
         UnboundedSender(self.0.clone())
@@ -805,8 +772,8 @@ impl<T> Receiver<T> {
                     return Async::Ready(msg);
                 }
                 PopResult::Empty => {
-                    // The queue is empty, return NotReady
-                    return Async::NotReady;
+                    // The queue is empty, return Pending
+                    return Async::Pending;
                 }
                 PopResult::Inconsistent => {
                     // Inconsistent means that there will be a message to pop
@@ -816,7 +783,7 @@ impl<T> Receiver<T> {
                     //
                     // 1) Spin
                     // 2) thread::yield_now()
-                    // 3) task::current().unwrap() & return NotReady
+                    // 3) task::current().unwrap() & return Pending
                     //
                     // For now, thread::yield_now() is used, but it would
                     // probably be better to spin a few times then yield.
@@ -895,15 +862,15 @@ impl<T> Stream for Receiver<T> {
             // Try to read a message off of the message queue.
             let msg = match self.next_message() {
                 Async::Ready(msg) => msg,
-                Async::NotReady => {
+                Async::Pending => {
                     // There are no messages to read, in this case, attempt to
                     // park. The act of parking will verify that the channel is
                     // still empty after the park operation has completed.
                     match self.try_park() {
                         TryPark::Parked => {
                             // The task was parked, and the channel is still
-                            // empty, return NotReady.
-                            return Ok(Async::NotReady);
+                            // empty, return Pending.
+                            return Ok(Async::Pending);
                         }
                         TryPark::Closed => {
                             // The channel is closed, there will be no further
@@ -962,138 +929,6 @@ impl<T> Stream for UnboundedReceiver<T> {
     }
 }
 
-/// Handle returned from the `spawn` function.
-///
-/// This handle is a stream that proxies a stream on a separate `Executor`.
-/// Created through the `mpsc::spawn` function, this handle will produce
-/// the same values as the proxied stream, as they are produced in the executor,
-/// and uses a limited buffer to exert back-pressure on the remote stream.
-///
-/// If this handle is dropped, then the stream will no longer be polled and is
-/// scheduled to be dropped.
-pub struct SpawnHandle<Item, Error> {
-    rx: Receiver<Result<Item, Error>>,
-    _cancel_tx: oneshot::Sender<()>,
-}
-
-/// Type of future which `Executor` instances must be able to execute for `spawn`.
-pub struct Execute<S: Stream> {
-    inner: SendAll<Sender<Result<S::Item, S::Error>>, Results<S, SendError<Result<S::Item, S::Error>>>>,
-    cancel_rx: oneshot::Receiver<()>,
-}
-
-/// Spawns a `stream` onto the instance of `Executor` provided, `executor`,
-/// returning a handle representing the remote stream.
-///
-/// The `stream` will be canceled if the `SpawnHandle` is dropped.
-///
-/// The `SpawnHandle` returned is a stream that is a proxy for `stream` itself.
-/// When `stream` has additional items available, then the `SpawnHandle`
-/// will have those same items available.
-///
-/// At most `buffer + 1` elements will be buffered at a time. If the buffer
-/// is full, then `stream` will stop progressing until more space is available.
-/// This allows the `SpawnHandle` to exert backpressure on the `stream`.
-///
-/// # Panics
-///
-/// This function will panic if `executor` is unable spawn a `Future` containing
-/// the entirety of the `stream`.
-pub fn spawn<S, E>(stream: S, executor: &E, buffer: usize) -> SpawnHandle<S::Item, S::Error>
-    where S: Stream,
-          E: Executor<Execute<S>>
-{
-    let (cancel_tx, cancel_rx) = oneshot::channel();
-    let (tx, rx) = channel(buffer);
-    executor.execute(Execute {
-        inner: tx.send_all(resultstream::new(stream)),
-        cancel_rx: cancel_rx,
-    }).expect("failed to spawn stream");
-    SpawnHandle {
-        rx: rx,
-        _cancel_tx: cancel_tx,
-    }
-}
-
-/// Spawns a `stream` onto the instance of `Executor` provided, `executor`,
-/// returning a handle representing the remote stream, with unbounded buffering.
-///
-/// The `stream` will be canceled if the `SpawnHandle` is dropped.
-///
-/// The `SpawnHandle` returned is a stream that is a proxy for `stream` itself.
-/// When `stream` has additional items available, then the `SpawnHandle`
-/// will have those same items available.
-///
-/// An unbounded buffer is used, which means that values will be buffered as
-/// fast as `stream` can produce them, without any backpressure. Therefore, if
-/// `stream` is an infinite stream, it can use an unbounded amount of memory, and
-/// potentially hog CPU resources.
-///
-/// # Panics
-///
-/// This function will panic if `executor` is unable spawn a `Future` containing
-/// the entirety of the `stream`.
-pub fn spawn_unbounded<S, E>(stream: S, executor: &E) -> SpawnHandle<S::Item, S::Error>
-    where S: Stream,
-          E: Executor<Execute<S>>
-{
-    let (cancel_tx, cancel_rx) = oneshot::channel();
-    let (tx, rx) = channel2(None);
-    executor.execute(Execute {
-        inner: tx.send_all(resultstream::new(stream)),
-        cancel_rx: cancel_rx,
-    }).expect("failed to spawn stream");
-    SpawnHandle {
-        rx: rx,
-        _cancel_tx: cancel_tx,
-    }
-}
-
-impl<I, E> Stream for SpawnHandle<I, E> {
-    type Item = I;
-    type Error = E;
-
-    fn poll(&mut self) -> Poll<Option<I>, E> {
-        match self.rx.poll() {
-            Ok(Async::Ready(Some(Ok(t)))) => Ok(Async::Ready(Some(t.into()))),
-            Ok(Async::Ready(Some(Err(e)))) => Err(e),
-            Ok(Async::Ready(None)) => Ok(Async::Ready(None)),
-            Ok(Async::NotReady) => Ok(Async::NotReady),
-            Err(_) => unreachable!("mpsc::Receiver should never return Err"),
-        }
-    }
-}
-
-impl<I, E> fmt::Debug for SpawnHandle<I, E> {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("SpawnHandle")
-         .finish()
-    }
-}
-
-impl<S: Stream> Future for Execute<S> {
-    type Item = ();
-    type Error = ();
-
-    fn poll(&mut self) -> Poll<(), ()> {
-        match self.cancel_rx.poll() {
-            Ok(Async::NotReady) => (),
-            _ => return Ok(Async::Ready(())),
-        }
-        match self.inner.poll() {
-            Ok(Async::NotReady) => Ok(Async::NotReady),
-            _ => Ok(Async::Ready(()))
-        }
-    }
-}
-
-impl<S: Stream> fmt::Debug for Execute<S> {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("Execute")
-         .finish()
-    }
-}
-
 /*
  *
  * ===== impl Inner =====
diff --git a/src/sync/mpsc/queue.rs b/futures-channel/src/mpsc/queue.rs
similarity index 100%
rename from src/sync/mpsc/queue.rs
rename to futures-channel/src/mpsc/queue.rs
diff --git a/src/sync/oneshot.rs b/futures-channel/src/oneshot.rs
similarity index 72%
rename from src/sync/oneshot.rs
rename to futures-channel/src/oneshot.rs
index 9542bdcc..d986582a 100644
--- a/src/sync/oneshot.rs
+++ b/futures-channel/src/oneshot.rs
@@ -6,10 +6,9 @@ use std::sync::atomic::Ordering::SeqCst;
 use std::error::Error;
 use std::fmt;
 
-use {Future, Poll, Async};
-use future::{lazy, Lazy, Executor, IntoFuture};
+use {Future, Poll, Async, Task};
+use task;
 use lock::Lock;
-use task::{self, Task};
 
 /// A future representing the completion of a computation happening elsewhere in
 /// memory.
@@ -178,7 +177,7 @@ impl<T> Inner<T> {
         if self.complete.load(SeqCst) {
             Ok(Async::Ready(()))
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 
@@ -255,7 +254,7 @@ impl<T> Inner<T> {
         // otherwise.
         //
         // If we're not done, and we're not complete, though, then we've
-        // successfully blocked our task and we return `NotReady`.
+        // successfully blocked our task and we return `Pending`.
         if done || self.complete.load(SeqCst) {
             // If taking the lock fails, the sender will realise that the we're
             // `done` when it checks the `complete` flag on the way out, and will
@@ -267,7 +266,7 @@ impl<T> Inner<T> {
             }
             Err(Canceled)
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 
@@ -328,7 +327,7 @@ impl<T> Sender<T> {
     /// and the result this `Sender` would otherwise produce should no longer
     /// be produced.
     ///
-    /// If `NotReady` is returned then the `Receiver` is still alive and may be
+    /// If `Pending` is returned then the `Receiver` is still alive and may be
     /// able to receive a message if sent. The current task, however, is
     /// scheduled to receive a notification if the corresponding `Receiver` goes
     /// away.
@@ -412,158 +411,3 @@ impl<T> Drop for Receiver<T> {
         self.inner.drop_rx()
     }
 }
-
-/// Handle returned from the `spawn` function.
-///
-/// This handle is a future representing the completion of a different future on
-/// a separate executor. Created through the `oneshot::spawn` function this
-/// handle will resolve when the future provided to `spawn` resolves on the
-/// `Executor` instance provided to that function.
-///
-/// If this handle is dropped then the future will automatically no longer be
-/// polled and is scheduled to be dropped. This can be canceled with the
-/// `forget` function, however.
-pub struct SpawnHandle<T, E> {
-    rx: Arc<ExecuteInner<Result<T, E>>>,
-}
-
-struct ExecuteInner<T> {
-    inner: Inner<T>,
-    keep_running: AtomicBool,
-}
-
-/// Type of future which `Execute` instances below must be able to spawn.
-pub struct Execute<F: Future> {
-    future: F,
-    tx: Arc<ExecuteInner<Result<F::Item, F::Error>>>,
-}
-
-/// Spawns a `future` onto the instance of `Executor` provided, `executor`,
-/// returning a handle representing the completion of the future.
-///
-/// The `SpawnHandle` returned is a future that is a proxy for `future` itself.
-/// When `future` completes on `executor` then the `SpawnHandle` will itself be
-/// resolved.  Internally `SpawnHandle` contains a `oneshot` channel and is
-/// thus safe to send across threads.
-///
-/// The `future` will be canceled if the `SpawnHandle` is dropped. If this is
-/// not desired then the `SpawnHandle::forget` function can be used to continue
-/// running the future to completion.
-///
-/// # Panics
-///
-/// This function will panic if the instance of `Spawn` provided is unable to
-/// spawn the `future` provided.
-///
-/// If the provided instance of `Spawn` does not actually run `future` to
-/// completion, then the returned handle may panic when polled. Typically this
-/// is not a problem, though, as most instances of `Spawn` will run futures to
-/// completion.
-///
-/// Note that the returned future will likely panic if the `futures` provided
-/// panics. If a future running on an executor panics that typically means that
-/// the executor drops the future, which falls into the above case of not
-/// running the future to completion essentially.
-pub fn spawn<F, E>(future: F, executor: &E) -> SpawnHandle<F::Item, F::Error>
-    where F: Future,
-          E: Executor<Execute<F>>,
-{
-    let data = Arc::new(ExecuteInner {
-        inner: Inner::new(),
-        keep_running: AtomicBool::new(false),
-    });
-    executor.execute(Execute {
-        future: future,
-        tx: data.clone(),
-    }).expect("failed to spawn future");
-    SpawnHandle { rx: data }
-}
-
-/// Spawns a function `f` onto the `Spawn` instance provided `s`.
-///
-/// For more information see the `spawn` function in this module. This function
-/// is just a thin wrapper around `spawn` which will execute the closure on the
-/// executor provided and then complete the future that the closure returns.
-pub fn spawn_fn<F, R, E>(f: F, executor: &E) -> SpawnHandle<R::Item, R::Error>
-    where F: FnOnce() -> R,
-          R: IntoFuture,
-          E: Executor<Execute<Lazy<F, R>>>,
-{
-    spawn(lazy(f), executor)
-}
-
-impl<T, E> SpawnHandle<T, E> {
-    /// Drop this future without canceling the underlying future.
-    ///
-    /// When `SpawnHandle` is dropped, the spawned future will be canceled as
-    /// well if the future hasn't already resolved. This function can be used
-    /// when to drop this future but keep executing the underlying future.
-    pub fn forget(self) {
-        self.rx.keep_running.store(true, SeqCst);
-    }
-}
-
-impl<T, E> Future for SpawnHandle<T, E> {
-    type Item = T;
-    type Error = E;
-
-    fn poll(&mut self) -> Poll<T, E> {
-        match self.rx.inner.recv() {
-            Ok(Async::Ready(Ok(t))) => Ok(t.into()),
-            Ok(Async::Ready(Err(e))) => Err(e),
-            Ok(Async::NotReady) => Ok(Async::NotReady),
-            Err(_) => panic!("future was canceled before completion"),
-        }
-    }
-}
-
-impl<T: fmt::Debug, E: fmt::Debug> fmt::Debug for SpawnHandle<T, E> {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("SpawnHandle")
-         .finish()
-    }
-}
-
-impl<T, E> Drop for SpawnHandle<T, E> {
-    fn drop(&mut self) {
-        self.rx.inner.drop_rx();
-    }
-}
-
-impl<F: Future> Future for Execute<F> {
-    type Item = ();
-    type Error = ();
-
-    fn poll(&mut self) -> Poll<(), ()> {
-        // If we're canceled then we may want to bail out early.
-        //
-        // If the `forget` function was called, though, then we keep going.
-        if self.tx.inner.poll_cancel().unwrap().is_ready() {
-            if !self.tx.keep_running.load(SeqCst) {
-                return Ok(().into())
-            }
-        }
-
-        let result = match self.future.poll() {
-            Ok(Async::NotReady) => return Ok(Async::NotReady),
-            Ok(Async::Ready(t)) => Ok(t),
-            Err(e) => Err(e),
-        };
-        drop(self.tx.inner.send(result));
-        Ok(().into())
-    }
-}
-
-impl<F: Future + fmt::Debug> fmt::Debug for Execute<F> {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("Execute")
-         .field("future", &self.future)
-         .finish()
-    }
-}
-
-impl<F: Future> Drop for Execute<F> {
-    fn drop(&mut self) {
-        self.tx.inner.drop_tx();
-    }
-}
diff --git a/futures-core/Cargo.toml b/futures-core/Cargo.toml
new file mode 100644
index 00000000..4c7d164b
--- /dev/null
+++ b/futures-core/Cargo.toml
@@ -0,0 +1,15 @@
+[package]
+name = "futures-core"
+version = "0.1.0"
+authors = ["Alex Crichton <alex@alexcrichton.com>"]
+license = "MIT/Apache-2.0"
+repository = "https://github.com/alexcrichton/futures-rs"
+homepage = "https://github.com/alexcrichton/futures-rs"
+documentation = "https://docs.rs/futures-core"
+description = """
+The core traits and types in for the `futures` library.
+"""
+
+[features]
+default = ["std"]
+std = []
diff --git a/futures-core/src/future/mod.rs b/futures-core/src/future/mod.rs
new file mode 100644
index 00000000..86cf8728
--- /dev/null
+++ b/futures-core/src/future/mod.rs
@@ -0,0 +1,236 @@
+//! Futures
+//!
+//! This module contains the `Future` and `IntoFuture` traits.
+//! Implementations are also provided for core types like
+//! `Box`, `Option`, and `Result`.
+
+use Poll;
+
+mod option;
+#[path = "result.rs"]
+mod result_;
+pub use self::result_::{result, ok, err, Result};
+
+/// Trait for types which are a placeholder of a value that may become
+/// available at some later point in time.
+///
+/// In addition to the documentation here you can also find more information
+/// about futures [online] at [https://tokio.rs](https://tokio.rs)
+///
+/// [online]: https://tokio.rs/docs/getting-started/futures/
+///
+/// Futures are used to provide a sentinel through which a value can be
+/// referenced. They crucially allow chaining and composing operations through
+/// consumption which allows expressing entire trees of computation as one
+/// sentinel value.
+///
+/// The ergonomics and implementation of the `Future` trait are very similar to
+/// the `Iterator` trait in that there is just one methods you need
+/// to implement, but you get a whole lot of others for free as a result.
+///
+/// # The `poll` method
+///
+/// The core method of future, `poll`, is used to attempt to generate the value
+/// of a `Future`. This method *does not block* but is allowed to inform the
+/// caller that the value is not ready yet. Implementations of `poll` may
+/// themselves do work to generate the value, but it's guaranteed that this will
+/// never block the calling thread.
+///
+/// A key aspect of this method is that if the value is not yet available the
+/// current task is scheduled to receive a notification when it's later ready to
+/// be made available. This follows what's typically known as a "readiness" or
+/// "pull" model where values are pulled out of futures on demand, and
+/// otherwise a task is notified when a value might be ready to get pulled out.
+///
+/// The `poll` method is not intended to be called in general, but rather is
+/// typically called in the context of a "task" which drives a future to
+/// completion. For more information on this see the `task` module.
+///
+/// More information about the details of `poll` and the nitty-gritty of tasks
+/// can be [found online at tokio.rs][poll-dox].
+///
+/// [poll-dox]: https://tokio.rs/docs/going-deeper-futures/futures-model/
+///
+/// # Combinators
+///
+/// Like iterators, futures provide a large number of combinators to work with
+/// futures to express computations in a much more natural method than
+/// scheduling a number of callbacks. For example the `map` method can change
+/// a `Future<Item=T>` to a `Future<Item=U>` or an `and_then` combinator could
+/// create a future after the first one is done and only be resolved when the
+/// second is done.
+///
+/// Combinators act very similarly to the methods on the `Iterator` trait itself
+/// or those on `Option` and `Result`. Like with iterators, the combinators are
+/// zero-cost and don't impose any extra layers of indirection you wouldn't
+/// otherwise have to write down.
+///
+/// More information about combinators can be found [on tokio.rs].
+///
+/// [on tokio.rs]: https://tokio.rs/docs/going-deeper-futures/futures-mechanics/
+pub trait Future {
+    /// The type of value that this future will resolved with if it is
+    /// successful.
+    type Item;
+
+    /// The type of error that this future will resolve with if it fails in a
+    /// normal fashion.
+    type Error;
+
+    /// Query this future to see if its value has become available, registering
+    /// interest if it is not.
+    ///
+    /// This function will check the internal state of the future and assess
+    /// whether the value is ready to be produced. Implementers of this function
+    /// should ensure that a call to this **never blocks** as event loops may
+    /// not work properly otherwise.
+    ///
+    /// When a future is not ready yet, the `Async::Pending` value will be
+    /// returned. In this situation the future will *also* register interest of
+    /// the current task in the value being produced. This is done by calling
+    /// `task::park` to retrieve a handle to the current `Task`. When the future
+    /// is then ready to make progress (e.g. it should be `poll`ed again) the
+    /// `unpark` method is called on the `Task`.
+    ///
+    /// More information about the details of `poll` and the nitty-gritty of
+    /// tasks can be [found online at tokio.rs][poll-dox].
+    ///
+    /// [poll-dox]: https://tokio.rs/docs/going-deeper-futures/futures-model/
+    ///
+    /// # Runtime characteristics
+    ///
+    /// This function, `poll`, is the primary method for 'making progress'
+    /// within a tree of futures. For example this method will be called
+    /// repeatedly as the internal state machine makes its various transitions.
+    /// Executors are responsible for ensuring that this function is called in
+    /// the right location (e.g. always on an I/O thread or not). Unless it is
+    /// otherwise arranged to be so, it should be ensured that **implementations
+    /// of this function finish very quickly**.
+    ///
+    /// Returning quickly prevents unnecessarily clogging up threads and/or
+    /// event loops while a `poll` function call, for example, takes up compute
+    /// resources to perform some expensive computation. If it is known ahead
+    /// of time that a call to `poll` may end up taking awhile, the work should
+    /// be offloaded to a thread pool (or something similar) to ensure that
+    /// `poll` can return quickly.
+    ///
+    /// Note that the `poll` function is not called repeatedly in a loop for
+    /// futures typically, but only whenever the future itself is ready. If
+    /// you're familiar with the `poll(2)` or `select(2)` syscalls on Unix
+    /// it's worth noting that futures typically do *not* suffer the same
+    /// problems of "all wakeups must poll all events". Futures have enough
+    /// support for only polling futures which cause a wakeup.
+    ///
+    /// # Return value
+    ///
+    /// This function returns `Async::Pending` if the future is not ready yet,
+    /// `Err` if the future is finished but resolved to an error, or
+    /// `Async::Ready` with the result of this future if it's finished
+    /// successfully. Once a future has finished it is considered a contract
+    /// error to continue polling the future.
+    ///
+    /// If `Pending` is returned, then the future will internally register
+    /// interest in the value being produced for the current task (through
+    /// `task::park`). In other words, the current task will receive a
+    /// notification (through the `unpark` method) once the value is ready to be
+    /// produced or the future can make progress.
+    ///
+    /// Note that if `Pending` is returned it only means that *this* task will
+    /// receive a notification. Historical calls to `poll` with different tasks
+    /// will not receive notifications. In other words, implementers of the
+    /// `Future` trait need not store a queue of tasks to notify, but only the
+    /// last task that called this method. Alternatively callers of this method
+    /// can only rely on the most recent task which call `poll` being notified
+    /// when a future is ready.
+    ///
+    /// # Panics
+    ///
+    /// Once a future has completed (returned `Ready` or `Err` from `poll`),
+    /// then any future calls to `poll` may panic, block forever, or otherwise
+    /// cause wrong behavior. The `Future` trait itself provides no guarantees
+    /// about the behavior of `poll` after a future has completed.
+    ///
+    /// Callers who may call `poll` too many times may want to consider using
+    /// the `fuse` adaptor which defines the behavior of `poll`, but comes with
+    /// a little bit of extra cost.
+    ///
+    /// Additionally, calls to `poll` must always be made from within the
+    /// context of a task. If a current task is not set then this method will
+    /// likely panic.
+    ///
+    /// # Errors
+    ///
+    /// This future may have failed to finish the computation, in which case
+    /// the `Err` variant will be returned with an appropriate payload of an
+    /// error.
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error>;
+}
+
+impl<'a, F: ?Sized + Future> Future for &'a mut F {
+    type Item = F::Item;
+    type Error = F::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        (**self).poll()
+    }
+}
+
+if_std! {
+    impl<F: ?Sized + Future> Future for ::std::boxed::Box<F> {
+        type Item = F::Item;
+        type Error = F::Error;
+
+        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+            (**self).poll()
+        }
+    }
+
+
+    impl<F: Future> Future for ::std::panic::AssertUnwindSafe<F> {
+        type Item = F::Item;
+        type Error = F::Error;
+
+        fn poll(&mut self) -> Poll<F::Item, F::Error> {
+            self.0.poll()
+        }
+    }
+}
+
+/// Class of types which can be converted into a future.
+///
+/// This trait is very similar to the `IntoIterator` trait and is intended to be
+/// used in a very similar fashion.
+pub trait IntoFuture {
+    /// The future that this type can be converted into.
+    type Future: Future<Item=Self::Item, Error=Self::Error>;
+
+    /// The item that the future may resolve with.
+    type Item;
+    /// The error that the future may resolve with.
+    type Error;
+
+    /// Consumes this object and produces a future.
+    fn into_future(self) -> Self::Future;
+}
+
+impl<F> IntoFuture for F where F: Future {
+    type Future = Self;
+    type Item = <Self as Future>::Item;
+    type Error = <Self as Future>::Error;
+    fn into_future(self) -> Self { self }
+}
+
+/// Asynchronous conversion from a type `T`.
+///
+/// This trait is analogous to `std::convert::From`, adapted to asynchronous
+/// computation.
+pub trait FutureFrom<T>: Sized {
+    /// The future for the conversion.
+    type Future: Future<Item=Self, Error=Self::Error>;
+
+    /// Possible errors during conversion.
+    type Error;
+
+    /// Consume the given value, beginning the conversion.
+    fn future_from(T) -> Self::Future;
+}
diff --git a/futures-core/src/future/option.rs b/futures-core/src/future/option.rs
new file mode 100644
index 00000000..fb1c726c
--- /dev/null
+++ b/futures-core/src/future/option.rs
@@ -0,0 +1,36 @@
+//! Definition of the `Option` (optional step) combinator
+
+use {Future, IntoFuture, Poll, Async};
+
+use core::option;
+
+/// A future representing a value which may or may not be present.
+///
+/// Created by the `IntoFuture` implementation for `std::option::Option`.
+#[derive(Debug, Clone)]
+#[must_use = "futures do nothing unless polled"]
+pub struct Option<T> {
+    inner: option::Option<T>,
+}
+
+impl<T> IntoFuture for option::Option<T> where T: IntoFuture {
+    type Future = Option<T::Future>;
+    type Item = option::Option<T::Item>;
+    type Error = T::Error;
+
+    fn into_future(self) -> Option<T::Future> {
+        Option { inner: self.map(IntoFuture::into_future) }
+    }
+}
+
+impl<F, T, E> Future for Option<F> where F: Future<Item=T, Error=E> {
+    type Item = option::Option<T>;
+    type Error = E;
+
+    fn poll(&mut self) -> Poll<option::Option<T>, E> {
+        match self.inner {
+            None => Ok(Async::Ready(None)),
+            Some(ref mut x) => x.poll().map(|x| x.map(Some)),
+        }
+    }
+}
diff --git a/src/future/result.rs b/futures-core/src/future/result.rs
similarity index 73%
rename from src/future/result.rs
rename to futures-core/src/future/result.rs
index 5c44a63e..69ab2c72 100644
--- a/src/future/result.rs
+++ b/futures-core/src/future/result.rs
@@ -1,19 +1,25 @@
-//! Definition of the `Result` (immediately finished) combinator
-
+use {Future, IntoFuture, Poll, Async};
 use core::result;
 
-use {Future, Poll, Async};
-
 /// A future representing a value that is immediately ready.
 ///
 /// Created by the `result` function.
 #[derive(Debug, Clone)]
 #[must_use = "futures do nothing unless polled"]
-// TODO: rename this to `Result` on the next major version
-pub struct FutureResult<T, E> {
+pub struct Result<T, E> {
     inner: Option<result::Result<T, E>>,
 }
 
+impl<T, E> IntoFuture for result::Result<T, E> {
+    type Future = Result<T, E>;
+    type Item = T;
+    type Error = E;
+
+    fn into_future(self) -> Result<T, E> {
+        result(self)
+    }
+}
+
 /// Creates a new "leaf future" which will resolve with the given result.
 ///
 /// The returned future represents a computation which is finished immediately.
@@ -28,8 +34,8 @@ pub struct FutureResult<T, E> {
 /// let future_of_1 = result::<u32, u32>(Ok(1));
 /// let future_of_err_2 = result::<u32, u32>(Err(2));
 /// ```
-pub fn result<T, E>(r: result::Result<T, E>) -> FutureResult<T, E> {
-    FutureResult { inner: Some(r) }
+pub fn result<T, E>(r: result::Result<T, E>) -> Result<T, E> {
+    Result { inner: Some(r) }
 }
 
 /// Creates a "leaf future" from an immediate value of a finished and
@@ -45,7 +51,7 @@ pub fn result<T, E>(r: result::Result<T, E>) -> FutureResult<T, E> {
 ///
 /// let future_of_1 = ok::<u32, u32>(1);
 /// ```
-pub fn ok<T, E>(t: T) -> FutureResult<T, E> {
+pub fn ok<T, E>(t: T) -> Result<T, E> {
     result(Ok(t))
 }
 
@@ -61,11 +67,11 @@ pub fn ok<T, E>(t: T) -> FutureResult<T, E> {
 ///
 /// let future_of_err_1 = err::<u32, u32>(1);
 /// ```
-pub fn err<T, E>(e: E) -> FutureResult<T, E> {
+pub fn err<T, E>(e: E) -> Result<T, E> {
     result(Err(e))
 }
 
-impl<T, E> Future for FutureResult<T, E> {
+impl<T, E> Future for Result<T, E> {
     type Item = T;
     type Error = E;
 
@@ -74,8 +80,8 @@ impl<T, E> Future for FutureResult<T, E> {
     }
 }
 
-impl<T, E> From<Result<T, E>> for FutureResult<T, E> {
-    fn from(r: Result<T, E>) -> Self {
+impl<T, E> From<result::Result<T, E>> for Result<T, E> {
+    fn from(r: result::Result<T, E>) -> Self {
         result(r)
     }
 }
diff --git a/futures-core/src/lib.rs b/futures-core/src/lib.rs
new file mode 100644
index 00000000..748df505
--- /dev/null
+++ b/futures-core/src/lib.rs
@@ -0,0 +1,31 @@
+//! Core traits and types for asynchronous operations in Rust.
+
+#![no_std]
+#![deny(missing_docs, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/futures/0.2")]
+
+#[macro_use]
+#[cfg(feature = "std")]
+extern crate std;
+
+macro_rules! if_std {
+    ($($i:item)*) => ($(
+        #[cfg(feature = "std")]
+        $i
+    )*)
+}
+
+#[macro_use]
+mod poll;
+pub use poll::{Async, Poll};
+
+pub mod future;
+pub use future::{Future, IntoFuture};
+
+pub mod stream;
+pub use stream::Stream;
+
+mod task_impl;
+pub mod task;
+
+pub mod never;
diff --git a/futures-core/src/never.rs b/futures-core/src/never.rs
new file mode 100644
index 00000000..1796b9b4
--- /dev/null
+++ b/futures-core/src/never.rs
@@ -0,0 +1,36 @@
+//! Definition and trait implementations for the `Never` type.
+
+use {Future, Stream, Poll};
+
+/// A type that can never exist.
+/// This is used to indicate values which can never be created, such as the
+/// error type of infallible futures.
+///
+/// This type is a stable equivalent to the `!` type from `std`.
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd)]
+pub enum Never {}
+
+impl Never {
+    /// Convert the `Never` type into any other type.
+    pub fn never_into<T>(self) -> T {
+        match self {}
+    }
+}
+
+impl Future for Never {
+    type Item = Never;
+    type Error = Never;
+
+    fn poll(&mut self) -> Poll<Never, Never> {
+        match *self {}
+    }
+}
+
+impl Stream for Never {
+    type Item = Never;
+    type Error = Never;
+
+    fn poll(&mut self) -> Poll<Option<Never>, Never> {
+        match *self {}
+    }
+}
diff --git a/futures-core/src/poll.rs b/futures-core/src/poll.rs
new file mode 100644
index 00000000..0a616b6b
--- /dev/null
+++ b/futures-core/src/poll.rs
@@ -0,0 +1,71 @@
+/// A macro for extracting the successful type of a `Poll<T, E>`.
+///
+/// This macro bakes propagation of both errors and `Pending` signals by
+/// returning early.
+#[macro_export]
+macro_rules! try_ready {
+    ($e:expr) => (match $e {
+        Ok($crate::Async::Ready(t)) => t,
+        Ok($crate::Async::Pending) => return Ok($crate::Async::Pending),
+        Err(e) => return Err(From::from(e)),
+    })
+}
+
+/// A convenience wrapper for `Result<Async<T>, E>`.
+///
+/// `Poll` is the return type of the `poll` method on the `Future` trait.
+///
+/// * `Ok(Async::Ready(t))` means the future has successfully resolved.
+/// * `Ok(Async::Pending)` means the future is not able to fully resolve yet.
+///    The current `Task` will be awoken when the future can make further
+///    progress.
+/// * `Err(e)` means that an error was encountered when attempting to complet
+///    the future. `Future`s which have returned errors are complete, and
+///    should not be polled again. However,`Stream`s that have returned errors
+///    may not be complete and should still be polled.
+pub type Poll<T, E> = Result<Async<T>, E>;
+
+/// Return type of future, indicating whether a value is ready or not.
+#[derive(Copy, Clone, Debug, PartialEq)]
+pub enum Async<T> {
+    /// Represents that a value is immediately ready.
+    Ready(T),
+
+    /// Represents that a value is not ready yet.
+    /// 
+    /// When a function returns `Pending`, the function should also
+    /// ensure that the current `Task` is scheduled to be awoken when
+    /// progress can be made.
+    Pending,
+}
+
+impl<T> Async<T> {
+    /// Change the success value of this `Async` with the closure provided
+    pub fn map<F, U>(self, f: F) -> Async<U>
+        where F: FnOnce(T) -> U
+    {
+        match self {
+            Async::Ready(t) => Async::Ready(f(t)),
+            Async::Pending => Async::Pending,
+        }
+    }
+
+    /// Returns whether this is `Async::Ready`
+    pub fn is_ready(&self) -> bool {
+        match *self {
+            Async::Ready(_) => true,
+            Async::Pending => false,
+        }
+    }
+
+    /// Returns whether this is `Async::Pending`
+    pub fn is_not_ready(&self) -> bool {
+        !self.is_ready()
+    }
+}
+
+impl<T> From<T> for Async<T> {
+    fn from(t: T) -> Async<T> {
+        Async::Ready(t)
+    }
+}
diff --git a/futures-core/src/stream/mod.rs b/futures-core/src/stream/mod.rs
new file mode 100644
index 00000000..46cdfa78
--- /dev/null
+++ b/futures-core/src/stream/mod.rs
@@ -0,0 +1,120 @@
+//! Asynchronous streams
+//!
+//! This module contains the `Stream` trait and a number of adaptors for this
+//! trait. This trait is very similar to the `Iterator` trait in the standard
+//! library except that it expresses the concept of blocking as well. A stream
+//! here is a sequential sequence of values which may take some amount of time
+//! in between to produce.
+//!
+//! A stream may request that it is blocked between values while the next value
+//! is calculated, and provides a way to get notified once the next value is
+//! ready as well.
+//!
+//! You can find more information/tutorials about streams [online at
+//! https://tokio.rs][online]
+//!
+//! [online]: https://tokio.rs/docs/getting-started/streams-and-sinks/
+
+use Poll;
+
+/// A stream of values, not all of which may have been produced yet.
+///
+/// `Stream` is a trait to represent any source of sequential events or items
+/// which acts like an iterator but long periods of time may pass between
+/// items. Like `Future` the methods of `Stream` never block and it is thus
+/// suitable for programming in an asynchronous fashion. This trait is very
+/// similar to the `Iterator` trait in the standard library where `Some` is
+/// used to signal elements of the stream and `None` is used to indicate that
+/// the stream is finished.
+///
+/// Like futures a stream has basic combinators to transform the stream, perform
+/// more work on each item, etc.
+///
+/// You can find more information/tutorials about streams [online at
+/// https://tokio.rs][online]
+///
+/// [online]: https://tokio.rs/docs/getting-started/streams-and-sinks/
+///
+/// # Streams as Futures
+///
+/// Any instance of `Stream` can also be viewed as a `Future` where the resolved
+/// value is the next item in the stream along with the rest of the stream. The
+/// `into_future` adaptor can be used here to convert any stream into a future
+/// for use with other future methods like `join` and `select`.
+///
+/// # Errors
+///
+/// Streams, like futures, can also model errors in their computation. All
+/// streams have an associated `Error` type like with futures. Currently as of
+/// the 0.1 release of this library an error on a stream **does not terminate
+/// the stream**. That is, after one error is received, another error may be
+/// received from the same stream (it's valid to keep polling).
+///
+/// This property of streams, however, is [being considered] for change in 0.2
+/// where an error on a stream is similar to `None`, it terminates the stream
+/// entirely. If one of these use cases suits you perfectly and not the other,
+/// please feel welcome to comment on [the issue][being considered]!
+///
+/// [being considered]: https://github.com/alexcrichton/futures-rs/issues/206
+pub trait Stream {
+    /// The type of item this stream will yield on success.
+    type Item;
+
+    /// The type of error this stream may generate.
+    type Error;
+
+    /// Attempt to pull out the next value of this stream, returning `None` if
+    /// the stream is finished.
+    ///
+    /// This method, like `Future::poll`, is the sole method of pulling out a
+    /// value from a stream. This method must also be run within the context of
+    /// a task typically and implementors of this trait must ensure that
+    /// implementations of this method do not block, as it may cause consumers
+    /// to behave badly.
+    ///
+    /// # Return value
+    ///
+    /// If `Pending` is returned then this stream's next value is not ready
+    /// yet and implementations will ensure that the current task will be
+    /// notified when the next value may be ready. If `Some` is returned then
+    /// the returned value represents the next value on the stream. `Err`
+    /// indicates an error happened, while `Ok` indicates whether there was a
+    /// new item on the stream or whether the stream has terminated.
+    ///
+    /// # Panics
+    ///
+    /// Once a stream is finished, that is `Ready(None)` has been returned,
+    /// further calls to `poll` may result in a panic or other "bad behavior".
+    /// If this is difficult to guard against then the `fuse` adapter can be
+    /// used to ensure that `poll` always has well-defined semantics.
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error>;
+}
+
+impl<'a, S: ?Sized + Stream> Stream for &'a mut S {
+    type Item = S::Item;
+    type Error = S::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        (**self).poll()
+    }
+}
+
+if_std! {
+    impl<S: ?Sized + Stream> Stream for ::std::boxed::Box<S> {
+        type Item = S::Item;
+        type Error = S::Error;
+
+        fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+            (**self).poll()
+        }
+    }
+
+    impl<S: Stream> Stream for ::std::panic::AssertUnwindSafe<S> {
+        type Item = S::Item;
+        type Error = S::Error;
+
+        fn poll(&mut self) -> Poll<Option<S::Item>, S::Error> {
+            self.0.poll()
+        }
+    }
+}
diff --git a/src/task.rs b/futures-core/src/task.rs
similarity index 85%
rename from src/task.rs
rename to futures-core/src/task.rs
index e89a1391..76ee85e3 100644
--- a/src/task.rs
+++ b/futures-core/src/task.rs
@@ -26,7 +26,25 @@
 //! `current` function returns a handle to the currently running task, panicking
 //! if one isn't present. This handle is then used to later notify the task that
 //! it's ready to make progress through the `Task::notify` method.
-pub use task_impl::{Task, AtomicTask, current, init};
+pub use task_impl::{
+    // Types
+    NotifyHandle,
+    Spawn,
+    Task,
+
+    // Traits
+    Notify,
+    UnsafeNotify,
+
+    // Functions
+    current,
+    init,
+    spawn,
+    with_notify,
+};
 
 #[cfg(feature = "std")]
-pub use task_impl::LocalKey;
+pub use task_impl::{
+    AtomicTask,
+    LocalKey,
+};
diff --git a/src/task_impl/atomic_task.rs b/futures-core/src/task_impl/atomic_task.rs
similarity index 100%
rename from src/task_impl/atomic_task.rs
rename to futures-core/src/task_impl/atomic_task.rs
diff --git a/src/task_impl/core.rs b/futures-core/src/task_impl/core.rs
similarity index 100%
rename from src/task_impl/core.rs
rename to futures-core/src/task_impl/core.rs
diff --git a/src/task_impl/mod.rs b/futures-core/src/task_impl/mod.rs
similarity index 90%
rename from src/task_impl/mod.rs
rename to futures-core/src/task_impl/mod.rs
index 331a3322..8b7498df 100644
--- a/src/task_impl/mod.rs
+++ b/futures-core/src/task_impl/mod.rs
@@ -1,7 +1,7 @@
 use core::fmt;
 use core::marker::PhantomData;
 
-use {Poll, Future, Stream, Sink, StartSend};
+use {Poll, Future, Stream};
 
 mod atomic_task;
 pub use self::atomic_task::AtomicTask;
@@ -180,7 +180,7 @@ impl<T: ?Sized> Spawn<T> {
     /// `task::current()` will return a handle that contains the `notify`
     /// specified.
     ///
-    /// If this function returns `NotReady`, then the `notify` should have been
+    /// If this function returns `Pending`, then the `notify` should have been
     /// scheduled to receive a notification when poll can be called again.
     /// Otherwise if `Ready` or `Err` is returned, the `Spawn` task can be
     /// safely destroyed.
@@ -206,8 +206,7 @@ impl<T: ?Sized> Spawn<T> {
         where N: Clone + Into<NotifyHandle>,
               T: Future,
     {
-        let mk = || notify.clone().into();
-        self.enter(BorrowedUnpark::new(&mk, id), |f| f.poll())
+        self.poll_notify(notify, id, |s| s.poll())
     }
 
     /// Like `poll_future_notify`, except polls the underlying stream.
@@ -218,57 +217,16 @@ impl<T: ?Sized> Spawn<T> {
         where N: Clone + Into<NotifyHandle>,
               T: Stream,
     {
-        let mk = || notify.clone().into();
-        self.enter(BorrowedUnpark::new(&mk, id), |s| s.poll())
-    }
-
-    /// Invokes the underlying `start_send` method with this task in place.
-    ///
-    /// If the underlying operation returns `NotReady` then the `notify` value
-    /// passed in will receive a notification when the operation is ready to be
-    /// attempted again.
-    pub fn start_send_notify<N>(&mut self,
-                                value: T::SinkItem,
-                                notify: &N,
-                                id: usize)
-                               -> StartSend<T::SinkItem, T::SinkError>
-        where N: Clone + Into<NotifyHandle>,
-              T: Sink,
-    {
-        let mk = || notify.clone().into();
-        self.enter(BorrowedUnpark::new(&mk, id), |s| s.start_send(value))
+        self.poll_notify(notify, id, |s| s.poll())
     }
 
-    /// Invokes the underlying `poll_complete` method with this task in place.
-    ///
-    /// If the underlying operation returns `NotReady` then the `notify` value
-    /// passed in will receive a notification when the operation is ready to be
-    /// attempted again.
-    pub fn poll_flush_notify<N>(&mut self,
-                                notify: &N,
-                                id: usize)
-                                -> Poll<(), T::SinkError>
-        where N: Clone + Into<NotifyHandle>,
-              T: Sink,
-    {
-        let mk = || notify.clone().into();
-        self.enter(BorrowedUnpark::new(&mk, id), |s| s.poll_complete())
-    }
-
-    /// Invokes the underlying `close` method with this task in place.
-    ///
-    /// If the underlying operation returns `NotReady` then the `notify` value
-    /// passed in will receive a notification when the operation is ready to be
-    /// attempted again.
-    pub fn close_notify<N>(&mut self,
-                           notify: &N,
-                           id: usize)
-                           -> Poll<(), T::SinkError>
+    /// Invokes the function with the provided `notify` in the task context.
+    pub fn poll_notify<N, F, R>(&mut self, notify: &N, id: usize, f: F) -> R
         where N: Clone + Into<NotifyHandle>,
-              T: Sink,
+              F: FnOnce(&mut T) -> R
     {
         let mk = || notify.clone().into();
-        self.enter(BorrowedUnpark::new(&mk, id), |s| s.close())
+        self.enter(BorrowedUnpark::new(&mk, id), f)
     }
 
     fn enter<F, R>(&mut self, unpark: BorrowedUnpark, f: F) -> R
diff --git a/src/task_impl/std/data.rs b/futures-core/src/task_impl/std/data.rs
similarity index 100%
rename from src/task_impl/std/data.rs
rename to futures-core/src/task_impl/std/data.rs
diff --git a/src/task_impl/std/mod.rs b/futures-core/src/task_impl/std/mod.rs
similarity index 54%
rename from src/task_impl/std/mod.rs
rename to futures-core/src/task_impl/std/mod.rs
index b051b49a..aa815372 100644
--- a/src/task_impl/std/mod.rs
+++ b/futures-core/src/task_impl/std/mod.rs
@@ -1,21 +1,17 @@
 use std::prelude::v1::*;
 
 use std::cell::Cell;
-use std::fmt;
 use std::marker::PhantomData;
 use std::mem;
 use std::ptr;
 use std::sync::{Arc, Mutex, Condvar, Once, ONCE_INIT};
 use std::sync::atomic::{AtomicUsize, Ordering};
 
-use {Future, Stream, Sink, Async, AsyncSink};
+use {Future, Stream, Async};
 use super::core;
-use super::{BorrowedTask, NotifyHandle, Spawn, spawn, Notify, UnsafeNotify};
+use super::{BorrowedTask, NotifyHandle, Spawn, Notify, UnsafeNotify};
 pub use super::core::{BorrowedUnpark, TaskUnpark};
 
-mod unpark_mutex;
-pub use self::unpark_mutex::UnparkMutex;
-
 mod data;
 pub use self::data::*;
 
@@ -90,47 +86,12 @@ impl<F: Future> Spawn<F> {
 
             loop {
                 match self.poll_future_notify(notify, 0)? {
-                    Async::NotReady => notify.park(),
+                    Async::Pending => notify.park(),
                     Async::Ready(e) => return Ok(e),
                 }
             }
         })
     }
-
-    /// A specialized function to request running a future to completion on the
-    /// specified executor.
-    ///
-    /// This function only works for futures whose item and error types are `()`
-    /// and also implement the `Send` and `'static` bounds. This will submit
-    /// units of work (instances of `Run`) to the `exec` argument provided
-    /// necessary to drive the future to completion.
-    ///
-    /// When the future would block, it's arranged that when the future is again
-    /// ready it will submit another unit of work to the `exec` provided. This
-    /// will happen in a loop until the future has completed.
-    ///
-    /// This method is not appropriate for all futures, and other kinds of
-    /// executors typically provide a similar function with perhaps relaxed
-    /// bounds as well.
-    ///
-    /// Note that this method is likely to be deprecated in favor of the
-    /// `futures::Executor` trait and `execute` method, but if this'd cause
-    /// difficulty for you please let us know!
-    pub fn execute(self, exec: Arc<Executor>)
-        where F: Future<Item=(), Error=()> + Send + 'static,
-    {
-        exec.clone().execute(Run {
-            // Ideally this method would be defined directly on
-            // `Spawn<BoxFuture<(), ()>>` so we wouldn't have to box here and
-            // it'd be more explicit, but unfortunately that currently has a
-            // link error on nightly: rust-lang/rust#36155
-            spawn: spawn(Box::new(self.into_inner())),
-            inner: Arc::new(RunInner {
-                exec: exec,
-                mutex: UnparkMutex::new()
-            }),
-        })
-    }
 }
 
 impl<S: Stream> Spawn<S> {
@@ -141,7 +102,7 @@ impl<S: Stream> Spawn<S> {
 
             loop {
                 match self.poll_stream_notify(notify, 0) {
-                    Ok(Async::NotReady) => notify.park(),
+                    Ok(Async::Pending) => notify.park(),
                     Ok(Async::Ready(Some(e))) => return Some(Ok(e)),
                     Ok(Async::Ready(None)) => return None,
                     Err(e) => return Some(Err(e)),
@@ -151,134 +112,6 @@ impl<S: Stream> Spawn<S> {
     }
 }
 
-impl<S: Sink> Spawn<S> {
-    /// 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.
-    pub fn wait_send(&mut self, mut value: S::SinkItem)
-                     -> Result<(), S::SinkError> {
-        ThreadNotify::with_current(|notify| {
-
-            loop {
-                value = match self.start_send_notify(value, notify, 0)? {
-                    AsyncSink::NotReady(v) => v,
-                    AsyncSink::Ready => return Ok(()),
-                };
-                notify.park();
-            }
-        })
-    }
-
-    /// 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.
-    pub fn wait_flush(&mut self) -> Result<(), S::SinkError> {
-        ThreadNotify::with_current(|notify| {
-
-            loop {
-                if self.poll_flush_notify(notify, 0)?.is_ready() {
-                    return Ok(())
-                }
-                notify.park();
-            }
-        })
-    }
-
-    /// 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.
-    pub fn wait_close(&mut self) -> Result<(), S::SinkError> {
-        ThreadNotify::with_current(|notify| {
-
-            loop {
-                if self.close_notify(notify, 0)?.is_ready() {
-                    return Ok(())
-                }
-                notify.park();
-            }
-        })
-    }
-}
-
-/// A trait representing requests to poll futures.
-///
-/// This trait is an argument to the `Spawn::execute` which is used to run a
-/// future to completion. An executor will receive requests to run a future and
-/// an executor is responsible for ensuring that happens in a timely fashion.
-///
-/// Note that this trait is likely to be deprecated and/or renamed to avoid
-/// clashing with the `future::Executor` trait. If you've got a use case for
-/// this or would like to comment on the name please let us know!
-pub trait Executor: Send + Sync + 'static {
-    /// Requests that `Run` is executed soon on the given executor.
-    fn execute(&self, r: Run);
-}
-
-/// Units of work submitted to an `Executor`, currently only created
-/// internally.
-pub struct Run {
-    spawn: Spawn<Box<Future<Item = (), Error = ()> + Send>>,
-    inner: Arc<RunInner>,
-}
-
-struct RunInner {
-    mutex: UnparkMutex<Run>,
-    exec: Arc<Executor>,
-}
-
-impl Run {
-    /// Actually run the task (invoking `poll` on its future) on the current
-    /// thread.
-    pub fn run(self) {
-        let Run { mut spawn, inner } = self;
-
-        // SAFETY: the ownership of this `Run` object is evidence that
-        // we are in the `POLLING`/`REPOLL` state for the mutex.
-        unsafe {
-            inner.mutex.start_poll();
-
-            loop {
-                match spawn.poll_future_notify(&inner, 0) {
-                    Ok(Async::NotReady) => {}
-                    Ok(Async::Ready(())) |
-                    Err(()) => return inner.mutex.complete(),
-                }
-                let run = Run { spawn: spawn, inner: inner.clone() };
-                match inner.mutex.wait(run) {
-                    Ok(()) => return,            // we've waited
-                    Err(r) => spawn = r.spawn,   // someone's notified us
-                }
-            }
-        }
-    }
-}
-
-impl fmt::Debug for Run {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("Run")
-         .field("contents", &"...")
-         .finish()
-    }
-}
-
-impl Notify for RunInner {
-    fn notify(&self, _id: usize) {
-        match self.mutex.notify() {
-            Ok(run) => self.exec.execute(run),
-            Err(()) => {}
-        }
-    }
-}
-
 // ===== ThreadNotify =====
 
 struct ThreadNotify {
@@ -367,16 +200,6 @@ impl Notify for ThreadNotify {
     }
 }
 
-/// A concurrent set which allows for the insertion of `usize` values.
-///
-/// `EventSet`s are used to communicate precise information about the event(s)
-/// that triggered a task notification. See `task::with_unpark_event` for details.
-#[deprecated(since="0.1.18", note = "recommended to use `FuturesUnordered` instead")]
-pub trait EventSet: Send + Sync + 'static {
-    /// Insert the given ID into the set
-    fn insert(&self, id: usize);
-}
-
 // Safe implementation of `UnsafeNotify` for `Arc` in the standard library.
 //
 // Note that this is a very unsafe implementation! The crucial pieces is that
diff --git a/futures-cpupool/Cargo.toml b/futures-cpupool/Cargo.toml
index 35d9854c..242570de 100644
--- a/futures-cpupool/Cargo.toml
+++ b/futures-cpupool/Cargo.toml
@@ -15,7 +15,7 @@ computation on the threads themselves.
 num_cpus = "1.0"
 
 [dependencies.futures]
-path = ".."
-version = "0.1"
+path = "../futures"
+version = "0.2"
 default-features = false
 features = ["std"]
diff --git a/futures-cpupool/src/lib.rs b/futures-cpupool/src/lib.rs
index 0614368b..eb6f1094 100644
--- a/futures-cpupool/src/lib.rs
+++ b/futures-cpupool/src/lib.rs
@@ -40,6 +40,9 @@
 extern crate futures;
 extern crate num_cpus;
 
+mod unpark_mutex;
+use unpark_mutex::UnparkMutex;
+
 use std::panic::{self, AssertUnwindSafe};
 use std::sync::{Arc, Mutex};
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
@@ -47,10 +50,10 @@ use std::sync::mpsc;
 use std::thread;
 use std::fmt;
 
-use futures::{IntoFuture, Future, Poll, Async};
-use futures::future::{lazy, Executor, ExecuteError};
-use futures::sync::oneshot::{channel, Sender, Receiver};
-use futures::executor::{self, Run, Executor as OldExecutor};
+use futures::prelude::*;
+use futures::future::lazy;
+use futures::channel::oneshot::{channel, Sender, Receiver};
+use futures::task::{self, Notify, Spawn};
 
 /// A thread pool intended to run CPU intensive work.
 ///
@@ -176,6 +179,24 @@ impl CpuPool {
         Builder::new().create()
     }
 
+    fn execute_run(&self, run: Run) {
+        self.inner.send(Message::Run(run));
+    }
+
+    /// TODO
+    pub fn execute<F>(&self, future: F)
+        where F: Future<Item = (), Error = ()> + Send + 'static
+    {
+        let run = Run {
+            spawn: task::spawn(Box::new(future)),
+            inner: Arc::new(RunInner {
+                exec: self.clone(),
+                mutex: UnparkMutex::new(),
+            }),
+        };
+        self.execute_run(run);
+    }
+
     /// Spawns a future to run on this thread pool, returning a future
     /// representing the produced value.
     ///
@@ -211,7 +232,7 @@ impl CpuPool {
             tx: Some(tx),
             keep_running_flag: keep_running_flag.clone(),
         };
-        executor::spawn(sender).execute(self.inner.clone());
+        self.execute(sender);
         CpuFuture { inner: rx , keep_running_flag: keep_running_flag.clone() }
     }
 
@@ -235,15 +256,6 @@ impl CpuPool {
     }
 }
 
-impl<F> Executor<F> for CpuPool
-    where F: Future<Item = (), Error = ()> + Send + 'static,
-{
-    fn execute(&self, future: F) -> Result<(), ExecuteError<F>> {
-        executor::spawn(future).execute(self.inner.clone());
-        Ok(())
-    }
-}
-
 impl Inner {
     fn send(&self, msg: Message) {
         self.tx.lock().unwrap().send(msg).unwrap();
@@ -279,12 +291,6 @@ impl Drop for CpuPool {
     }
 }
 
-impl OldExecutor for Inner {
-    fn execute(&self, run: Run) {
-        self.send(Message::Run(run))
-    }
-}
-
 impl<T, E> CpuFuture<T, E> {
     /// Drop this future without canceling the underlying future.
     ///
@@ -305,7 +311,7 @@ impl<T: Send + 'static, E: Send + 'static> Future for CpuFuture<T, E> {
             Async::Ready(Ok(Ok(e))) => Ok(e.into()),
             Async::Ready(Ok(Err(e))) => Err(e),
             Async::Ready(Err(e)) => panic::resume_unwind(e),
-            Async::NotReady => Ok(Async::NotReady),
+            Async::Pending => Ok(Async::Pending),
         }
     }
 }
@@ -324,7 +330,7 @@ impl<F: Future> Future for MySender<F, Result<F::Item, F::Error>> {
 
         let res = match self.fut.poll() {
             Ok(Async::Ready(e)) => Ok(e),
-            Ok(Async::NotReady) => return Ok(Async::NotReady),
+            Ok(Async::Pending) => return Ok(Async::Pending),
             Err(e) => Err(e),
         };
 
@@ -430,6 +436,62 @@ impl Builder {
     }
 }
 
+/// Units of work submitted to an `Executor`, currently only created
+/// internally.
+struct Run {
+    spawn: Spawn<Box<Future<Item = (), Error = ()> + Send>>,
+    inner: Arc<RunInner>,
+}
+
+struct RunInner {
+    mutex: UnparkMutex<Run>,
+    exec: CpuPool,
+}
+
+impl Run {
+    /// Actually run the task (invoking `poll` on its future) on the current
+    /// thread.
+    pub fn run(self) {
+        let Run { mut spawn, inner } = self;
+
+        // SAFETY: the ownership of this `Run` object is evidence that
+        // we are in the `POLLING`/`REPOLL` state for the mutex.
+        unsafe {
+            inner.mutex.start_poll();
+
+            loop {
+                match spawn.poll_future_notify(&inner, 0) {
+                    Ok(Async::Pending) => {}
+                    Ok(Async::Ready(())) |
+                    Err(()) => return inner.mutex.complete(),
+                }
+                let run = Run { spawn: spawn, inner: inner.clone() };
+                match inner.mutex.wait(run) {
+                    Ok(()) => return,            // we've waited
+                    Err(r) => spawn = r.spawn,   // someone's notified us
+                }
+            }
+        }
+    }
+}
+
+impl fmt::Debug for Run {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_struct("Run")
+         .field("contents", &"...")
+         .finish()
+    }
+}
+
+impl Notify for RunInner {
+    fn notify(&self, _id: usize) {
+        match self.mutex.notify() {
+            Ok(run) => self.exec.execute_run(run),
+            Err(()) => {}
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/futures-executor/Cargo.toml b/futures-executor/Cargo.toml
new file mode 100644
index 00000000..f0453881
--- /dev/null
+++ b/futures-executor/Cargo.toml
@@ -0,0 +1,18 @@
+[package]
+name = "futures-executor"
+version = "0.1.0"
+authors = ["Alex Crichton <alex@alexcrichton.com>"]
+license = "MIT/Apache-2.0"
+repository = "https://github.com/alexcrichton/futures-rs"
+homepage = "https://github.com/alexcrichton/futures-rs"
+documentation = "https://docs.rs/futures-core"
+description = """
+Executors for asynchronous tasks based on the futures-rs library.
+"""
+
+[features]
+std = ["futures-core/std"]
+default = ["std"]
+
+[dependencies]
+futures-core = { path = "../futures-core", version = "0.1.0" }
diff --git a/futures-executor/src/lib.rs b/futures-executor/src/lib.rs
new file mode 100644
index 00000000..e69de29b
diff --git a/futures-sink/Cargo.toml b/futures-sink/Cargo.toml
new file mode 100644
index 00000000..68f7dd60
--- /dev/null
+++ b/futures-sink/Cargo.toml
@@ -0,0 +1,19 @@
+[package]
+name = "futures-sink"
+version = "0.1.0"
+authors = ["Alex Crichton <alex@alexcrichton.com>"]
+license = "MIT/Apache-2.0"
+repository = "https://github.com/alexcrichton/futures-rs"
+homepage = "https://github.com/alexcrichton/futures-rs"
+documentation = "https://docs.rs/futures-core"
+description = """
+The asynchronous `Sink` trait for the futures-rs library.
+"""
+
+[features]
+std = ["futures-core/std"]
+default = ["std"]
+
+[dependencies]
+futures-core = { path = "../futures-core", version = "0.1.0" }
+futures-channel = { path = "../futures-channel", version = "0.1.0" }
diff --git a/futures-sink/src/channel_impls.rs b/futures-sink/src/channel_impls.rs
new file mode 100644
index 00000000..6d7c1cfd
--- /dev/null
+++ b/futures-sink/src/channel_impls.rs
@@ -0,0 +1,61 @@
+use {Async, Sink, AsyncSink, StartSend, Poll};
+use futures_channel::mpsc::{Sender, SendError, UnboundedSender};
+
+fn res_to_async_sink<T>(res: Result<(), T>) -> AsyncSink<T> {
+    match res {
+        Ok(()) => AsyncSink::Ready,
+        Err(x) => AsyncSink::Pending(x),
+    }
+}
+
+impl<T> Sink for Sender<T> {
+    type SinkItem = T;
+    type SinkError = SendError<T>;
+
+    fn start_send(&mut self, msg: T) -> StartSend<T, SendError<T>> {
+        self.start_send(msg).map(res_to_async_sink)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), SendError<T>> {
+        Ok(Async::Ready(()))
+    }
+
+    fn close(&mut self) -> Poll<(), SendError<T>> {
+        Ok(Async::Ready(()))
+    }
+}
+
+impl<T> Sink for UnboundedSender<T> {
+    type SinkItem = T;
+    type SinkError = SendError<T>;
+
+    fn start_send(&mut self, msg: T) -> StartSend<T, SendError<T>> {
+        self.start_send(msg).map(res_to_async_sink)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), SendError<T>> {
+        Ok(Async::Ready(()))
+    }
+
+    fn close(&mut self) -> Poll<(), SendError<T>> {
+        Ok(Async::Ready(()))
+    }
+}
+
+impl<'a, T> Sink for &'a UnboundedSender<T> {
+    type SinkItem = T;
+    type SinkError = SendError<T>;
+
+    fn start_send(&mut self, msg: T) -> StartSend<T, SendError<T>> {
+        self.unbounded_send(msg)?;
+        Ok(AsyncSink::Ready)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), SendError<T>> {
+        Ok(Async::Ready(()))
+    }
+
+    fn close(&mut self) -> Poll<(), SendError<T>> {
+        Ok(Async::Ready(()))
+    }
+}
diff --git a/src/sink/mod.rs b/futures-sink/src/lib.rs
similarity index 57%
rename from src/sink/mod.rs
rename to futures-sink/src/lib.rs
index fcb18727..ed2ab2da 100644
--- a/src/sink/mod.rs
+++ b/futures-sink/src/lib.rs
@@ -1,39 +1,80 @@
 //! Asynchronous sinks
 //!
-//! This module contains the `Sink` trait, along with a number of adapter types
-//! for it. An overview is available in the documentation for the trait itself.
-//!
-//! You can find more information/tutorials about streams [online at
-//! https://tokio.rs][online]
-//!
-//! [online]: https://tokio.rs/docs/getting-started/streams-and-sinks/
+//! This crate contains the `Sink` trait which allows values to be sent
+//! asynchronously.
+
+#![no_std]
+#![deny(missing_docs, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/futures/0.2")]
+
+#[cfg(feature = "std")]
+extern crate std;
+
+extern crate futures_core;
+#[cfg(feature = "std")]
+extern crate futures_channel;
+
+macro_rules! if_std {
+    ($($i:item)*) => ($(
+        #[cfg(feature = "std")]
+        $i
+    )*)
+}
+
+use futures_core::Poll;
 
-use {IntoFuture, Poll, StartSend};
-use stream::Stream;
+/// The result of an asynchronous attempt to send a value to a sink.
+#[derive(Copy, Clone, Debug, PartialEq)]
+pub enum AsyncSink<T> {
+    /// The `start_send` attempt succeeded, so the sending process has
+    /// *started*; you must use `Sink::poll_complete` to drive the send
+    /// to completion.
+    Ready,
 
-mod with;
-mod with_flat_map;
-// mod with_map;
-// mod with_filter;
-// mod with_filter_map;
-mod flush;
-mod from_err;
-mod send;
-mod send_all;
-mod map_err;
-mod fanout;
+    /// The `start_send` attempt failed due to the sink being full. The value
+    /// being sent is returned, and the current `Task` will be automatically
+    /// notified again once the sink has room.
+    Pending(T),
+}
+
+impl<T> AsyncSink<T> {
+    /// Change the Pending value of this `AsyncSink` with the closure provided
+    pub fn map<F, U>(self, f: F) -> AsyncSink<U>
+        where F: FnOnce(T) -> U,
+    {
+        match self {
+            AsyncSink::Ready => AsyncSink::Ready,
+            AsyncSink::Pending(t) => AsyncSink::Pending(f(t)),
+        }
+    }
+
+    /// Returns whether this is `AsyncSink::Ready`
+    pub fn is_ready(&self) -> bool {
+        match *self {
+            AsyncSink::Ready => true,
+            AsyncSink::Pending(_) => false,
+        }
+    }
+
+    /// Returns whether this is `AsyncSink::Pending`
+    pub fn is_not_ready(&self) -> bool {
+        !self.is_ready()
+    }
+}
+
+/// Return type of the `Sink::start_send` method, indicating the outcome of a
+/// send attempt. See `AsyncSink` for more details.
+pub type StartSend<T, E> = Result<AsyncSink<T>, E>;
 
 if_std! {
-    mod buffer;
-    mod wait;
+    mod channel_impls;
 
-    pub use self::buffer::Buffer;
-    pub use self::wait::Wait;
+    use futures_core::Async;
+    use futures_core::never::Never;
 
-    // TODO: consider expanding this via e.g. FromIterator
     impl<T> Sink for ::std::vec::Vec<T> {
         type SinkItem = T;
-        type SinkError = (); // Change this to ! once it stabilizes
+        type SinkError = Never;
 
         fn start_send(&mut self, item: Self::SinkItem)
                       -> StartSend<Self::SinkItem, Self::SinkError>
@@ -74,15 +115,6 @@ if_std! {
     }
 }
 
-pub use self::with::With;
-pub use self::with_flat_map::WithFlatMap;
-pub use self::flush::Flush;
-pub use self::send::Send;
-pub use self::send_all::SendAll;
-pub use self::map_err::SinkMapErr;
-pub use self::from_err::SinkFromErr;
-pub use self::fanout::Fanout;
-
 /// A `Sink` is a value into which other values can be sent, asynchronously.
 ///
 /// Basic examples of sinks include the sending side of:
@@ -139,7 +171,7 @@ pub trait Sink {
     /// that several calls to `start_send` can be made prior to calling
     /// `poll_complete`, which will work on completing all pending items.
     ///
-    /// The method returns `AsyncSink::NotReady` if the sink was unable to begin
+    /// The method returns `AsyncSink::Pending` if the sink was unable to begin
     /// sending, usually due to being full. The sink must have attempted to
     /// complete processing any outstanding requests (equivalent to
     /// `poll_complete`) before yielding this result. The current task will be
@@ -181,7 +213,7 @@ pub trait Sink {
     /// via `start_send` will be guaranteed to be available to a listening
     /// receiver.
     ///
-    /// Returns `Ok(Async::NotReady)` if there is more work left to do, in which
+    /// Returns `Ok(Async::Pending)` if there is more work left to do, in which
     /// case the current task is scheduled to wake up when more progress may be
     /// possible.
     ///
@@ -222,7 +254,7 @@ pub trait Sink {
     /// down is followed by a little more blocking work.
     ///
     /// Callers of this function should work it it in a similar fashion to
-    /// `poll_complete`. Once called it may return `NotReady` which indicates
+    /// `poll_complete`. Once called it may return `Pending` which indicates
     /// that more external work needs to happen to make progress. The current
     /// task will be scheduled to receive a notification in such an event,
     /// however.
@@ -239,7 +271,7 @@ pub trait Sink {
     /// `Ready` once the close operation has completed. At that point it should
     /// be safe to drop the sink and deallocate associated resources.
     ///
-    /// If the value returned is `NotReady` then the sink is not yet closed and
+    /// If the value returned is `Pending` then the sink is not yet closed and
     /// work needs to be done to close it. The work has been scheduled and the
     /// current task will receive a notification when it's next ready to call
     /// this method again.
@@ -278,189 +310,6 @@ pub trait Sink {
     /// to implement it whenever you implement `Sink` locally. It is especially
     /// crucial to be sure to close inner sinks, if applicable.
     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 = "std")]
-    fn wait(self) -> Wait<Self>
-        where Self: Sized
-    {
-        wait::new(self)
-    }
-
-    /// Composes a function *in front of* the sink.
-    ///
-    /// This adapter produces a new sink that passes each value through the
-    /// given function `f` before sending it to `self`.
-    ///
-    /// To process each value, `f` produces a *future*, which is then polled to
-    /// completion before passing its result down to the underlying sink. If the
-    /// future produces an error, that error is returned by the new sink.
-    ///
-    /// Note that this function consumes the given sink, returning a wrapped
-    /// version, much like `Iterator::map`.
-    fn with<U, F, Fut>(self, f: F) -> With<Self, U, F, Fut>
-        where F: FnMut(U) -> Fut,
-              Fut: IntoFuture<Item = Self::SinkItem>,
-              Fut::Error: From<Self::SinkError>,
-              Self: Sized
-    {
-        with::new(self, f)
-    }
-
-    /// Composes a function *in front of* the sink.
-    ///
-    /// This adapter produces a new sink that passes each value through the
-    /// given function `f` before sending it to `self`.
-    ///
-    /// To process each value, `f` produces a *stream*, of which each value
-    /// is passed to the underlying sink. A new value will not be accepted until
-    /// the stream has been drained
-    ///
-    /// Note that this function consumes the given sink, returning a wrapped
-    /// version, much like `Iterator::flat_map`.
-    ///
-    /// # Examples
-    /// ---
-    /// Using this function with an iterator through use of the `stream::iter_ok()`
-    /// function
-    ///
-    /// ```
-    /// use futures::prelude::*;
-    /// use futures::stream;
-    /// use futures::sync::mpsc;
-    ///
-    /// let (tx, rx) = mpsc::channel::<i32>(5);
-    ///
-    /// let tx = tx.with_flat_map(|x| {
-    ///     stream::iter_ok(vec![42; x].into_iter().map(|y| y))
-    /// });
-    /// tx.send(5).wait().unwrap();
-    /// assert_eq!(rx.collect().wait(), Ok(vec![42, 42, 42, 42, 42]))
-    /// ```
-    fn with_flat_map<U, F, St>(self, f: F) -> WithFlatMap<Self, U, F, St>
-        where F: FnMut(U) -> St,
-              St: Stream<Item = Self::SinkItem, Error=Self::SinkError>,
-              Self: Sized
-        {
-            with_flat_map::new(self, f)
-        }
-
-    /*
-    fn with_map<U, F>(self, f: F) -> WithMap<Self, U, F>
-        where F: FnMut(U) -> Self::SinkItem,
-              Self: Sized;
-
-    fn with_filter<F>(self, f: F) -> WithFilter<Self, F>
-        where F: FnMut(Self::SinkItem) -> bool,
-              Self: Sized;
-
-    fn with_filter_map<U, F>(self, f: F) -> WithFilterMap<Self, U, F>
-        where F: FnMut(U) -> Option<Self::SinkItem>,
-              Self: Sized;
-     */
-
-    /// Transforms the error returned by the sink.
-    fn sink_map_err<F, E>(self, f: F) -> SinkMapErr<Self, F>
-        where F: FnOnce(Self::SinkError) -> E,
-              Self: Sized,
-    {
-        map_err::new(self, f)
-    }
-
-    /// Map this sink's error to any error implementing `From` for this sink's
-    /// `Error`, returning a new sink.
-    ///
-    /// If wanting to map errors of a `Sink + Stream`, use `.sink_from_err().from_err()`.
-    fn sink_from_err<E: From<Self::SinkError>>(self) -> from_err::SinkFromErr<Self, E>
-        where Self: Sized,
-    {
-        from_err::new(self)
-    }
-
-
-    /// Adds a fixed-size buffer to the current sink.
-    ///
-    /// The resulting sink will buffer up to `amt` items when the underlying
-    /// sink is unwilling to accept additional items. Calling `poll_complete` on
-    /// the buffered sink will attempt to both empty the buffer and complete
-    /// processing on the underlying sink.
-    ///
-    /// Note that this function consumes the given sink, returning a wrapped
-    /// version, much like `Iterator::map`.
-    ///
-    /// This method is only available when the `std` feature of this
-    /// library is activated, and it is activated by default.
-    #[cfg(feature = "std")]
-    fn buffer(self, amt: usize) -> Buffer<Self>
-        where Self: Sized
-    {
-        buffer::new(self, amt)
-    }
-
-    /// Fanout items to multiple sinks.
-    ///
-    /// This adapter clones each incoming item and forwards it to both this as well as
-    /// the other sink at the same time.
-    fn fanout<S>(self, other: S) -> Fanout<Self, S>
-        where Self: Sized,
-              Self::SinkItem: Clone,
-              S: Sink<SinkItem=Self::SinkItem, SinkError=Self::SinkError>
-    {
-        fanout::new(self, other)
-    }
-
-    /// A future that completes when the sink has finished processing all
-    /// pending requests.
-    ///
-    /// The sink itself is returned after flushing is complete; this adapter is
-    /// intended to be used when you want to stop sending to the sink until
-    /// all current requests are processed.
-    fn flush(self) -> Flush<Self>
-        where Self: Sized
-    {
-        flush::new(self)
-    }
-
-    /// A future that completes after the given item has been fully processed
-    /// into the sink, including flushing.
-    ///
-    /// Note that, **because of the flushing requirement, it is usually better
-    /// to batch together items to send via `send_all`, rather than flushing
-    /// between each item.**
-    ///
-    /// On completion, the sink is returned.
-    fn send(self, item: Self::SinkItem) -> Send<Self>
-        where Self: Sized
-    {
-        send::new(self, item)
-    }
-
-    /// A future that completes after the given stream has been fully processed
-    /// into the sink, including flushing.
-    ///
-    /// This future will drive the stream to keep producing items until it is
-    /// exhausted, sending each item to the sink. It will complete once both the
-    /// stream is exhausted, the sink has received all items, the sink has been
-    /// flushed, and the sink has been closed.
-    ///
-    /// Doing `sink.send_all(stream)` is roughly equivalent to
-    /// `stream.forward(sink)`. The returned future will exhaust all items from
-    /// `stream` and send them to `self`, closing `self` when all items have been
-    /// received.
-    ///
-    /// On completion, the pair `(sink, source)` is returned.
-    fn send_all<S>(self, stream: S) -> SendAll<Self, S>
-        where S: Stream<Item = Self::SinkItem>,
-              Self::SinkError: From<S::Error>,
-              Self: Sized
-    {
-        send_all::new(self, stream)
-    }
 }
 
 impl<'a, S: ?Sized + Sink> Sink for &'a mut S {
diff --git a/futures-util/Cargo.toml b/futures-util/Cargo.toml
new file mode 100644
index 00000000..f7c514eb
--- /dev/null
+++ b/futures-util/Cargo.toml
@@ -0,0 +1,21 @@
+[package]
+name = "futures-util"
+version = "0.1.0"
+authors = ["Alex Crichton <alex@alexcrichton.com>"]
+license = "MIT/Apache-2.0"
+repository = "https://github.com/alexcrichton/futures-rs"
+homepage = "https://github.com/alexcrichton/futures-rs"
+documentation = "https://docs.rs/futures-core"
+description = """
+Common utilities and extension traits for the futures-rs library.
+"""
+
+[features]
+std = ["futures-core/std", "futures-executor/std", "futures-sink/std"]
+default = ["std"]
+
+[dependencies]
+futures-core = { path = "../futures-core", version = "0.1.0" }
+futures-channel = { path = "../futures-channel", version = "0.1.0" }
+futures-executor = { path = "../futures-executor", version = "0.1.0" }
+futures-sink = { path = "../futures-sink", version = "0.1.0" }
diff --git a/src/future/and_then.rs b/futures-util/src/future/and_then.rs
similarity index 100%
rename from src/future/and_then.rs
rename to futures-util/src/future/and_then.rs
diff --git a/src/future/catch_unwind.rs b/futures-util/src/future/catch_unwind.rs
similarity index 75%
rename from src/future/catch_unwind.rs
rename to futures-util/src/future/catch_unwind.rs
index f87f1181..b318be40 100644
--- a/src/future/catch_unwind.rs
+++ b/futures-util/src/future/catch_unwind.rs
@@ -1,6 +1,6 @@
 use std::prelude::v1::*;
 use std::any::Any;
-use std::panic::{catch_unwind, UnwindSafe, AssertUnwindSafe};
+use std::panic::{catch_unwind, UnwindSafe};
 
 use {Future, Poll, Async};
 
@@ -31,21 +31,12 @@ impl<F> Future for CatchUnwind<F>
         let mut future = self.future.take().expect("cannot poll twice");
         let (res, future) = catch_unwind(|| (future.poll(), future))?;
         match res {
-            Ok(Async::NotReady) => {
+            Ok(Async::Pending) => {
                 self.future = Some(future);
-                Ok(Async::NotReady)
+                Ok(Async::Pending)
             }
             Ok(Async::Ready(t)) => Ok(Async::Ready(Ok(t))),
             Err(e) => Ok(Async::Ready(Err(e))),
         }
     }
 }
-
-impl<F: Future> Future for AssertUnwindSafe<F> {
-    type Item = F::Item;
-    type Error = F::Error;
-
-    fn poll(&mut self) -> Poll<F::Item, F::Error> {
-        self.0.poll()
-    }
-}
diff --git a/src/future/chain.rs b/futures-util/src/future/chain.rs
similarity index 94%
rename from src/future/chain.rs
rename to futures-util/src/future/chain.rs
index 1bf5cd63..11050513 100644
--- a/src/future/chain.rs
+++ b/futures-util/src/future/chain.rs
@@ -24,7 +24,7 @@ impl<A, B, C> Chain<A, B, C>
         let a_result = match *self {
             Chain::First(ref mut a, _) => {
                 match a.poll() {
-                    Ok(Async::NotReady) => return Ok(Async::NotReady),
+                    Ok(Async::Pending) => return Ok(Async::Pending),
                     Ok(Async::Ready(t)) => Ok(t),
                     Err(e) => Err(e),
                 }
diff --git a/src/future/either.rs b/futures-util/src/future/either.rs
similarity index 100%
rename from src/future/either.rs
rename to futures-util/src/future/either.rs
diff --git a/src/future/empty.rs b/futures-util/src/future/empty.rs
similarity index 87%
rename from src/future/empty.rs
rename to futures-util/src/future/empty.rs
index fbb56b26..372f5599 100644
--- a/src/future/empty.rs
+++ b/futures-util/src/future/empty.rs
@@ -16,7 +16,7 @@ pub struct Empty<T, E> {
 /// Creates a future which never resolves, representing a computation that never
 /// finishes.
 ///
-/// The returned future will forever return `Async::NotReady`.
+/// The returned future will forever return `Async::Pending`.
 pub fn empty<T, E>() -> Empty<T, E> {
     Empty { _data: marker::PhantomData }
 }
@@ -26,6 +26,6 @@ impl<T, E> Future for Empty<T, E> {
     type Error = E;
 
     fn poll(&mut self) -> Poll<T, E> {
-        Ok(Async::NotReady)
+        Ok(Async::Pending)
     }
 }
diff --git a/src/future/flatten.rs b/futures-util/src/future/flatten.rs
similarity index 94%
rename from src/future/flatten.rs
rename to futures-util/src/future/flatten.rs
index bfe28697..7fabd6f3 100644
--- a/src/future/flatten.rs
+++ b/futures-util/src/future/flatten.rs
@@ -14,7 +14,7 @@ pub struct Flatten<A> where A: Future, A::Item: IntoFuture {
 impl<A> fmt::Debug for Flatten<A>
     where A: Future + fmt::Debug,
           A::Item: IntoFuture,
-          <<A as IntoFuture>::Item as IntoFuture>::Future: fmt::Debug,
+          <<A as Future>::Item as IntoFuture>::Future: fmt::Debug,
 {
     fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
         fmt.debug_struct("Flatten")
diff --git a/src/future/flatten_stream.rs b/futures-util/src/future/flatten_stream.rs
similarity index 95%
rename from src/future/flatten_stream.rs
rename to futures-util/src/future/flatten_stream.rs
index 7bf3b9ca..b69c6154 100644
--- a/src/future/flatten_stream.rs
+++ b/futures-util/src/future/flatten_stream.rs
@@ -1,6 +1,6 @@
 use {Async, Future, Poll};
 use core::fmt;
-use stream::Stream;
+use Stream;
 
 /// Future for the `flatten_stream` combinator, flattening a
 /// future-of-a-stream to get just the result of the final stream as a stream.
@@ -61,9 +61,9 @@ impl<F> Stream for FlattenStream<F>
             let (next_state, ret_opt) = match self.state {
                 State::Future(ref mut f) => {
                     match f.poll() {
-                        Ok(Async::NotReady) => {
+                        Ok(Async::Pending) => {
                             // State is not changed, early return.
-                            return Ok(Async::NotReady)
+                            return Ok(Async::Pending)
                         },
                         Ok(Async::Ready(stream)) => {
                             // Future resolved to stream.
diff --git a/src/future/from_err.rs b/futures-util/src/future/from_err.rs
similarity index 92%
rename from src/future/from_err.rs
rename to futures-util/src/future/from_err.rs
index 97e35d7c..ab351e6c 100644
--- a/src/future/from_err.rs
+++ b/futures-util/src/future/from_err.rs
@@ -27,7 +27,7 @@ impl<A:Future, E:From<A::Error>> Future for FromErr<A, E> {
 
     fn poll(&mut self) -> Poll<A::Item, E> {
         let e = match self.future.poll() {
-            Ok(Async::NotReady) => return Ok(Async::NotReady),
+            Ok(Async::Pending) => return Ok(Async::Pending),
             other => other,
         };
         e.map_err(From::from)
diff --git a/src/future/fuse.rs b/futures-util/src/future/fuse.rs
similarity index 88%
rename from src/future/fuse.rs
rename to futures-util/src/future/fuse.rs
index e37eace0..30d752fd 100644
--- a/src/future/fuse.rs
+++ b/futures-util/src/future/fuse.rs
@@ -3,7 +3,7 @@ use {Future, Poll, Async};
 /// A future which "fuses" a future once it's been resolved.
 ///
 /// Normally futures can behave unpredictable once they're used after a future
-/// has been resolved, but `Fuse` is always defined to return `Async::NotReady`
+/// has been resolved, but `Fuse` is always defined to return `Async::Pending`
 /// from `poll` after it has resolved successfully or returned an error.
 ///
 /// This is created by the `Future::fuse` method.
@@ -25,13 +25,13 @@ impl<A: Future> Future for Fuse<A> {
 
     fn poll(&mut self) -> Poll<A::Item, A::Error> {
         let res = self.future.as_mut().map(|f| f.poll());
-        match res.unwrap_or(Ok(Async::NotReady)) {
+        match res.unwrap_or(Ok(Async::Pending)) {
             res @ Ok(Async::Ready(_)) |
             res @ Err(_) => {
                 self.future = None;
                 res
             }
-            Ok(Async::NotReady) => Ok(Async::NotReady)
+            Ok(Async::Pending) => Ok(Async::Pending)
         }
     }
 }
diff --git a/src/future/inspect.rs b/futures-util/src/future/inspect.rs
similarity index 94%
rename from src/future/inspect.rs
rename to futures-util/src/future/inspect.rs
index 59fcd786..0900523f 100644
--- a/src/future/inspect.rs
+++ b/futures-util/src/future/inspect.rs
@@ -29,7 +29,7 @@ impl<A, F> Future for Inspect<A, F>
 
     fn poll(&mut self) -> Poll<A::Item, A::Error> {
         match self.future.poll() {
-            Ok(Async::NotReady) => Ok(Async::NotReady),
+            Ok(Async::Pending) => Ok(Async::Pending),
             Ok(Async::Ready(e)) => {
                 (self.f.take().expect("cannot poll Inspect twice"))(&e);
                 Ok(Async::Ready(e))
diff --git a/src/future/into_stream.rs b/futures-util/src/future/into_stream.rs
similarity index 90%
rename from src/future/into_stream.rs
rename to futures-util/src/future/into_stream.rs
index 6e299e6a..8b358eaa 100644
--- a/src/future/into_stream.rs
+++ b/futures-util/src/future/into_stream.rs
@@ -1,6 +1,6 @@
 use {Async, Poll};
 use Future;
-use stream::Stream;
+use Stream;
 
 /// Future that forwards one element from the underlying future
 /// (whether it is success of error) and emits EOF after that.
@@ -24,7 +24,7 @@ impl<F: Future> Stream for IntoStream<F> {
             None => return Ok(Async::Ready(None)),
             Some(ref mut future) => {
                 match future.poll() {
-                    Ok(Async::NotReady) => return Ok(Async::NotReady),
+                    Ok(Async::Pending) => return Ok(Async::Pending),
                     Err(e) => Err(e),
                     Ok(Async::Ready(r)) => Ok(r),
                 }
diff --git a/src/future/join.rs b/futures-util/src/future/join.rs
similarity index 96%
rename from src/future/join.rs
rename to futures-util/src/future/join.rs
index 45212120..b5d8dcac 100644
--- a/src/future/join.rs
+++ b/futures-util/src/future/join.rs
@@ -3,7 +3,7 @@
 use core::fmt;
 use core::mem;
 
-use {Future, Poll, IntoFuture, Async};
+use {Future, Poll, Async};
 
 macro_rules! generate {
     ($(
@@ -84,11 +84,13 @@ macro_rules! generate {
                 if all_done {
                     Ok(Async::Ready((self.a.take(), $(self.$B.take()),*)))
                 } else {
-                    Ok(Async::NotReady)
+                    Ok(Async::Pending)
                 }
             }
         }
 
+        // Incoherent-- add to futures-core when stable.
+        /*
         impl<A, $($B),*> IntoFuture for (A, $($B),*)
             where A: IntoFuture,
         $(
@@ -110,6 +112,7 @@ macro_rules! generate {
                 }
             }
         }
+        */
 
     )*)
 }
@@ -159,7 +162,7 @@ impl<A: Future> MaybeDone<A> {
                 *self = MaybeDone::Done(res);
                 Ok(true)
             }
-            Async::NotReady => Ok(false),
+            Async::Pending => Ok(false),
         }
     }
 
diff --git a/src/future/join_all.rs b/futures-util/src/future/join_all.rs
similarity index 97%
rename from src/future/join_all.rs
rename to futures-util/src/future/join_all.rs
index e1cddd42..a4278624 100644
--- a/src/future/join_all.rs
+++ b/futures-util/src/future/join_all.rs
@@ -96,7 +96,7 @@ impl<Item> Future for JoinAll<Item>
                 ElemState::Pending(ref mut t) => {
                     match t.poll() {
                         Ok(Async::Ready(v)) => Ok(v),
-                        Ok(Async::NotReady) => {
+                        Ok(Async::Pending) => {
                             all_done = false;
                             continue
                         }
@@ -127,7 +127,7 @@ impl<Item> Future for JoinAll<Item>
             }).collect();
             Ok(Async::Ready(result))
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 }
diff --git a/src/future/lazy.rs b/futures-util/src/future/lazy.rs
similarity index 100%
rename from src/future/lazy.rs
rename to futures-util/src/future/lazy.rs
diff --git a/src/future/loop_fn.rs b/futures-util/src/future/loop_fn.rs
similarity index 100%
rename from src/future/loop_fn.rs
rename to futures-util/src/future/loop_fn.rs
diff --git a/src/future/map.rs b/futures-util/src/future/map.rs
similarity index 92%
rename from src/future/map.rs
rename to futures-util/src/future/map.rs
index 4b1f4cd7..f350f5b4 100644
--- a/src/future/map.rs
+++ b/futures-util/src/future/map.rs
@@ -28,7 +28,7 @@ impl<U, A, F> Future for Map<A, F>
 
     fn poll(&mut self) -> Poll<U, A::Error> {
         let e = match self.future.poll() {
-            Ok(Async::NotReady) => return Ok(Async::NotReady),
+            Ok(Async::Pending) => return Ok(Async::Pending),
             Ok(Async::Ready(e)) => Ok(e),
             Err(e) => Err(e),
         };
diff --git a/src/future/map_err.rs b/futures-util/src/future/map_err.rs
similarity index 92%
rename from src/future/map_err.rs
rename to futures-util/src/future/map_err.rs
index 4ea12f45..a34c45ea 100644
--- a/src/future/map_err.rs
+++ b/futures-util/src/future/map_err.rs
@@ -28,7 +28,7 @@ impl<U, A, F> Future for MapErr<A, F>
 
     fn poll(&mut self) -> Poll<A::Item, U> {
         let e = match self.future.poll() {
-            Ok(Async::NotReady) => return Ok(Async::NotReady),
+            Ok(Async::Pending) => return Ok(Async::Pending),
             other => other,
         };
         e.map_err(self.f.take().expect("cannot poll MapErr twice"))
diff --git a/src/future/mod.rs b/futures-util/src/future/mod.rs
similarity index 67%
rename from src/future/mod.rs
rename to futures-util/src/future/mod.rs
index 47c3d761..2274a216 100644
--- a/src/future/mod.rs
+++ b/futures-util/src/future/mod.rs
@@ -1,23 +1,20 @@
 //! Futures
 //!
-//! This module contains the `Future` trait and a number of adaptors for this
+//! This module contains the `FutureExt` trait and a number of adaptors for this
 //! trait. See the crate docs, and the docs for `Future`, for full detail.
 
-use core::fmt;
 use core::result;
 
+use {Future, IntoFuture, Stream};
+
 // Primitive futures
 mod empty;
 mod lazy;
 mod poll_fn;
-#[path = "result.rs"]
-mod result_;
 mod loop_fn;
-mod option;
 pub use self::empty::{empty, Empty};
 pub use self::lazy::{lazy, Lazy};
 pub use self::poll_fn::{poll_fn, PollFn};
-pub use self::result_::{result, ok, err, FutureResult};
 pub use self::loop_fn::{loop_fn, Loop, LoopFn};
 
 // combinators
@@ -67,173 +64,13 @@ if_std! {
     pub use self::select_all::{SelectAll, SelectAllNext, select_all};
     pub use self::select_ok::{SelectOk, select_ok};
     pub use self::shared::{Shared, SharedItem, SharedError};
-
-    impl<F: ?Sized + Future> Future for ::std::boxed::Box<F> {
-        type Item = F::Item;
-        type Error = F::Error;
-
-        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-            (**self).poll()
-        }
-    }
 }
 
-use {Poll, stream};
-
-/// Trait for types which are a placeholder of a value that may become
-/// available at some later point in time.
-///
-/// In addition to the documentation here you can also find more information
-/// about futures [online] at [https://tokio.rs](https://tokio.rs)
-///
-/// [online]: https://tokio.rs/docs/getting-started/futures/
-///
-/// Futures are used to provide a sentinel through which a value can be
-/// referenced. They crucially allow chaining and composing operations through
-/// consumption which allows expressing entire trees of computation as one
-/// sentinel value.
-///
-/// The ergonomics and implementation of the `Future` trait are very similar to
-/// the `Iterator` trait in that there is just one methods you need
-/// to implement, but you get a whole lot of others for free as a result.
-///
-/// # The `poll` method
-///
-/// The core method of future, `poll`, is used to attempt to generate the value
-/// of a `Future`. This method *does not block* but is allowed to inform the
-/// caller that the value is not ready yet. Implementations of `poll` may
-/// themselves do work to generate the value, but it's guaranteed that this will
-/// never block the calling thread.
-///
-/// A key aspect of this method is that if the value is not yet available the
-/// current task is scheduled to receive a notification when it's later ready to
-/// be made available. This follows what's typically known as a "readiness" or
-/// "pull" model where values are pulled out of futures on demand, and
-/// otherwise a task is notified when a value might be ready to get pulled out.
-///
-/// The `poll` method is not intended to be called in general, but rather is
-/// typically called in the context of a "task" which drives a future to
-/// completion. For more information on this see the `task` module.
-///
-/// More information about the details of `poll` and the nitty-gritty of tasks
-/// can be [found online at tokio.rs][poll-dox].
-///
-/// [poll-dox]: https://tokio.rs/docs/going-deeper-futures/futures-model/
-///
-/// # Combinators
-///
-/// Like iterators, futures provide a large number of combinators to work with
-/// futures to express computations in a much more natural method than
-/// scheduling a number of callbacks. For example the `map` method can change
-/// a `Future<Item=T>` to a `Future<Item=U>` or an `and_then` combinator could
-/// create a future after the first one is done and only be resolved when the
-/// second is done.
-///
-/// Combinators act very similarly to the methods on the `Iterator` trait itself
-/// or those on `Option` and `Result`. Like with iterators, the combinators are
-/// zero-cost and don't impose any extra layers of indirection you wouldn't
-/// otherwise have to write down.
-///
-/// More information about combinators can be found [on tokio.rs].
-///
-/// [on tokio.rs]: https://tokio.rs/docs/going-deeper-futures/futures-mechanics/
-pub trait Future {
-    /// The type of value that this future will resolved with if it is
-    /// successful.
-    type Item;
-
-    /// The type of error that this future will resolve with if it fails in a
-    /// normal fashion.
-    type Error;
-
-    /// Query this future to see if its value has become available, registering
-    /// interest if it is not.
-    ///
-    /// This function will check the internal state of the future and assess
-    /// whether the value is ready to be produced. Implementers of this function
-    /// should ensure that a call to this **never blocks** as event loops may
-    /// not work properly otherwise.
-    ///
-    /// When a future is not ready yet, the `Async::NotReady` value will be
-    /// returned. In this situation the future will *also* register interest of
-    /// the current task in the value being produced. This is done by calling
-    /// `task::park` to retrieve a handle to the current `Task`. When the future
-    /// is then ready to make progress (e.g. it should be `poll`ed again) the
-    /// `unpark` method is called on the `Task`.
-    ///
-    /// More information about the details of `poll` and the nitty-gritty of
-    /// tasks can be [found online at tokio.rs][poll-dox].
-    ///
-    /// [poll-dox]: https://tokio.rs/docs/going-deeper-futures/futures-model/
-    ///
-    /// # Runtime characteristics
-    ///
-    /// This function, `poll`, is the primary method for 'making progress'
-    /// within a tree of futures. For example this method will be called
-    /// repeatedly as the internal state machine makes its various transitions.
-    /// Executors are responsible for ensuring that this function is called in
-    /// the right location (e.g. always on an I/O thread or not). Unless it is
-    /// otherwise arranged to be so, it should be ensured that **implementations
-    /// of this function finish very quickly**.
-    ///
-    /// Returning quickly prevents unnecessarily clogging up threads and/or
-    /// event loops while a `poll` function call, for example, takes up compute
-    /// resources to perform some expensive computation. If it is known ahead
-    /// of time that a call to `poll` may end up taking awhile, the work should
-    /// be offloaded to a thread pool (or something similar) to ensure that
-    /// `poll` can return quickly.
-    ///
-    /// Note that the `poll` function is not called repeatedly in a loop for
-    /// futures typically, but only whenever the future itself is ready. If
-    /// you're familiar with the `poll(2)` or `select(2)` syscalls on Unix
-    /// it's worth noting that futures typically do *not* suffer the same
-    /// problems of "all wakeups must poll all events". Futures have enough
-    /// support for only polling futures which cause a wakeup.
-    ///
-    /// # Return value
-    ///
-    /// This function returns `Async::NotReady` if the future is not ready yet,
-    /// `Err` if the future is finished but resolved to an error, or
-    /// `Async::Ready` with the result of this future if it's finished
-    /// successfully. Once a future has finished it is considered a contract
-    /// error to continue polling the future.
-    ///
-    /// If `NotReady` is returned, then the future will internally register
-    /// interest in the value being produced for the current task (through
-    /// `task::park`). In other words, the current task will receive a
-    /// notification (through the `unpark` method) once the value is ready to be
-    /// produced or the future can make progress.
-    ///
-    /// Note that if `NotReady` is returned it only means that *this* task will
-    /// receive a notification. Historical calls to `poll` with different tasks
-    /// will not receive notifications. In other words, implementers of the
-    /// `Future` trait need not store a queue of tasks to notify, but only the
-    /// last task that called this method. Alternatively callers of this method
-    /// can only rely on the most recent task which call `poll` being notified
-    /// when a future is ready.
-    ///
-    /// # Panics
-    ///
-    /// Once a future has completed (returned `Ready` or `Err` from `poll`),
-    /// then any future calls to `poll` may panic, block forever, or otherwise
-    /// cause wrong behavior. The `Future` trait itself provides no guarantees
-    /// about the behavior of `poll` after a future has completed.
-    ///
-    /// Callers who may call `poll` too many times may want to consider using
-    /// the `fuse` adaptor which defines the behavior of `poll`, but comes with
-    /// a little bit of extra cost.
-    ///
-    /// Additionally, calls to `poll` must always be made from within the
-    /// context of a task. If a current task is not set then this method will
-    /// likely panic.
-    ///
-    /// # Errors
-    ///
-    /// This future may have failed to finish the computation, in which case
-    /// the `Err` variant will be returned with an appropriate payload of an
-    /// error.
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error>;
+impl<T: ?Sized> FutureExt for T where T: Future {}
 
+/// An extension trait for `Future`s that provides a variety of convenient
+/// combinator functions.
+pub trait FutureExt: Future {
     /// Block the current thread until this future is resolved.
     ///
     /// This method will consume ownership of this future, driving it to
@@ -259,7 +96,7 @@ pub trait Future {
     fn wait(self) -> result::Result<Self::Item, Self::Error>
         where Self: Sized
     {
-        ::executor::spawn(self).wait_future()
+        ::task::spawn(self).wait_future()
     }
 
     /// Map this future's result to a different type, returning a new future of
@@ -786,7 +623,7 @@ pub trait Future {
     /// assert_eq!(None, iter.next());
     /// ```
     fn flatten_stream(self) -> FlattenStream<Self>
-        where <Self as Future>::Item: stream::Stream<Error=Self::Error>,
+        where <Self as Future>::Item: Stream<Error=Self::Error>,
               Self: Sized
     {
         flatten_stream::new(self)
@@ -802,7 +639,7 @@ pub trait Future {
     /// has defined semantics.
     ///
     /// Once a future has been `fuse`d and it returns a completion from `poll`,
-    /// then it will forever return `NotReady` from `poll` again (never
+    /// then it will forever return `Pending` from `poll` again (never
     /// resolve).  This, unlike the trait's `poll` method, is guaranteed.
     ///
     /// This combinator will drop this future as soon as it's been completed to
@@ -823,7 +660,7 @@ pub trait Future {
     /// // This, however, is guaranteed to not panic
     /// let mut future = future::ok::<i32, u32>(2).fuse();
     /// assert_eq!(future.poll(), Ok(Async::Ready(2)));
-    /// assert_eq!(future.poll(), Ok(Async::NotReady));
+    /// assert_eq!(future.poll(), Ok(Async::Pending));
     /// ```
     fn fuse(self) -> Fuse<Self>
         where Self: Sized
@@ -943,15 +780,6 @@ pub trait Future {
     }
 }
 
-impl<'a, F: ?Sized + Future> Future for &'a mut F {
-    type Item = F::Item;
-    type Error = F::Error;
-
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        (**self).poll()
-    }
-}
-
 // Just a helper function to ensure the futures we're returning all have the
 // right implementations.
 fn assert_future<A, B, F>(t: F) -> F
@@ -960,140 +788,3 @@ fn assert_future<A, B, F>(t: F) -> F
     t
 }
 
-/// Class of types which can be converted into a future.
-///
-/// This trait is very similar to the `IntoIterator` trait and is intended to be
-/// used in a very similar fashion.
-pub trait IntoFuture {
-    /// The future that this type can be converted into.
-    type Future: Future<Item=Self::Item, Error=Self::Error>;
-
-    /// The item that the future may resolve with.
-    type Item;
-    /// The error that the future may resolve with.
-    type Error;
-
-    /// Consumes this object and produces a future.
-    fn into_future(self) -> Self::Future;
-}
-
-impl<F: Future> IntoFuture for F {
-    type Future = F;
-    type Item = F::Item;
-    type Error = F::Error;
-
-    fn into_future(self) -> F {
-        self
-    }
-}
-
-impl<T, E> IntoFuture for result::Result<T, E> {
-    type Future = FutureResult<T, E>;
-    type Item = T;
-    type Error = E;
-
-    fn into_future(self) -> FutureResult<T, E> {
-        result(self)
-    }
-}
-
-/// Asynchronous conversion from a type `T`.
-///
-/// This trait is analogous to `std::convert::From`, adapted to asynchronous
-/// computation.
-pub trait FutureFrom<T>: Sized {
-    /// The future for the conversion.
-    type Future: Future<Item=Self, Error=Self::Error>;
-
-    /// Possible errors during conversion.
-    type Error;
-
-    /// Consume the given value, beginning the conversion.
-    fn future_from(T) -> Self::Future;
-}
-
-/// A trait for types which can spawn fresh futures.
-///
-/// This trait is typically implemented for "executors", or those types which
-/// can execute futures to completion. Futures passed to `Spawn::spawn`
-/// typically get turned into a *task* and are then driven to completion.
-///
-/// On spawn, the executor takes ownership of the future and becomes responsible
-/// to call `Future::poll()` whenever a readiness notification is raised.
-pub trait Executor<F: Future<Item = (), Error = ()>> {
-    /// Spawns a future to run on this `Executor`, typically in the
-    /// "background".
-    ///
-    /// This function will return immediately, and schedule the future `future`
-    /// to run on `self`. The details of scheduling and execution are left to
-    /// the implementations of `Executor`, but this is typically a primary point
-    /// for injecting concurrency in a futures-based system. Futures spawned
-    /// through this `execute` function tend to run concurrently while they're
-    /// waiting on events.
-    ///
-    /// # Errors
-    ///
-    /// Implementers of this trait are allowed to reject accepting this future
-    /// as well. This can happen for various reason such as:
-    ///
-    /// * The executor is shut down
-    /// * The executor has run out of capacity to execute futures
-    ///
-    /// The decision is left to the caller how to work with this form of error.
-    /// The error returned transfers ownership of the future back to the caller.
-    fn execute(&self, future: F) -> Result<(), ExecuteError<F>>;
-}
-
-/// Errors returned from the `Spawn::spawn` function.
-pub struct ExecuteError<F> {
-    future: F,
-    kind: ExecuteErrorKind,
-}
-
-/// Kinds of errors that can be returned from the `Execute::spawn` function.
-///
-/// Executors which may not always be able to accept a future may return one of
-/// these errors, indicating why it was unable to spawn a future.
-#[derive(Debug, Copy, Clone, PartialEq)]
-pub enum ExecuteErrorKind {
-    /// This executor has shut down and will no longer accept new futures to
-    /// spawn.
-    Shutdown,
-
-    /// This executor has no more capacity to run more futures. Other futures
-    /// need to finish before this executor can accept another.
-    NoCapacity,
-
-    #[doc(hidden)]
-    __Nonexhaustive,
-}
-
-impl<F> ExecuteError<F> {
-    /// Create a new `ExecuteError`
-    pub fn new(kind: ExecuteErrorKind, future: F) -> ExecuteError<F> {
-        ExecuteError {
-            future: future,
-            kind: kind,
-        }
-    }
-
-    /// Returns the associated reason for the error
-    pub fn kind(&self) -> ExecuteErrorKind {
-        self.kind
-    }
-
-    /// Consumes self and returns the original future that was spawned.
-    pub fn into_future(self) -> F {
-        self.future
-    }
-}
-
-impl<F> fmt::Debug for ExecuteError<F> {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        match self.kind {
-            ExecuteErrorKind::Shutdown => "executor has shut down".fmt(f),
-            ExecuteErrorKind::NoCapacity => "executor has no more capacity".fmt(f),
-            ExecuteErrorKind::__Nonexhaustive => panic!(),
-        }
-    }
-}
diff --git a/src/future/or_else.rs b/futures-util/src/future/or_else.rs
similarity index 100%
rename from src/future/or_else.rs
rename to futures-util/src/future/or_else.rs
diff --git a/src/future/poll_fn.rs b/futures-util/src/future/poll_fn.rs
similarity index 100%
rename from src/future/poll_fn.rs
rename to futures-util/src/future/poll_fn.rs
diff --git a/src/future/select.rs b/futures-util/src/future/select.rs
similarity index 95%
rename from src/future/select.rs
rename to futures-util/src/future/select.rs
index c48e1c0a..92e93a23 100644
--- a/src/future/select.rs
+++ b/futures-util/src/future/select.rs
@@ -48,11 +48,11 @@ impl<A, B> Future for Select<A, B>
                 match a.poll() {
                     Err(a) => (Err(a), true),
                     Ok(Async::Ready(a)) => (Ok(a), true),
-                    Ok(Async::NotReady) => {
+                    Ok(Async::Pending) => {
                         match b.poll() {
                             Err(a) => (Err(a), false),
                             Ok(Async::Ready(a)) => (Ok(a), false),
-                            Ok(Async::NotReady) => return Ok(Async::NotReady),
+                            Ok(Async::Pending) => return Ok(Async::Pending),
                         }
                     }
                 }
diff --git a/src/future/select2.rs b/futures-util/src/future/select2.rs
similarity index 89%
rename from src/future/select2.rs
rename to futures-util/src/future/select2.rs
index 073f67be..5acc9eb1 100644
--- a/src/future/select2.rs
+++ b/futures-util/src/future/select2.rs
@@ -26,12 +26,12 @@ impl<A, B> Future for Select2<A, B> where A: Future, B: Future {
         match a.poll() {
             Err(e) => Err(Either::A((e, b))),
             Ok(Async::Ready(x)) => Ok(Async::Ready(Either::A((x, b)))),
-            Ok(Async::NotReady) => match b.poll() {
+            Ok(Async::Pending) => match b.poll() {
                 Err(e) => Err(Either::B((e, a))),
                 Ok(Async::Ready(x)) => Ok(Async::Ready(Either::B((x, a)))),
-                Ok(Async::NotReady) => {
+                Ok(Async::Pending) => {
                     self.inner = Some((a, b));
-                    Ok(Async::NotReady)
+                    Ok(Async::Pending)
                 }
             }
         }
diff --git a/src/future/select_all.rs b/futures-util/src/future/select_all.rs
similarity index 95%
rename from src/future/select_all.rs
rename to futures-util/src/future/select_all.rs
index 1fbc9869..73a0123d 100644
--- a/src/future/select_all.rs
+++ b/futures-util/src/future/select_all.rs
@@ -51,7 +51,7 @@ impl<A> Future for SelectAll<A>
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         let item = self.inner.iter_mut().enumerate().filter_map(|(i, f)| {
             match f.poll() {
-                Ok(Async::NotReady) => None,
+                Ok(Async::Pending) => None,
                 Ok(Async::Ready(e)) => Some((i, Ok(e))),
                 Err(e) => Some((i, Err(e))),
             }
@@ -65,7 +65,7 @@ impl<A> Future for SelectAll<A>
                     Err(e) => Err((e, idx, rest)),
                 }
             }
-            None => Ok(Async::NotReady),
+            None => Ok(Async::Pending),
         }
     }
 }
diff --git a/src/future/select_ok.rs b/futures-util/src/future/select_ok.rs
similarity index 96%
rename from src/future/select_ok.rs
rename to futures-util/src/future/select_ok.rs
index f122a0ea..7d6ea835 100644
--- a/src/future/select_ok.rs
+++ b/futures-util/src/future/select_ok.rs
@@ -49,7 +49,7 @@ impl<A> Future for SelectOk<A> where A: Future {
         loop {
             let item = self.inner.iter_mut().enumerate().filter_map(|(i, f)| {
                 match f.poll() {
-                    Ok(Async::NotReady) => None,
+                    Ok(Async::Pending) => None,
                     Ok(Async::Ready(e)) => Some((i, Ok(e))),
                     Err(e) => Some((i, Err(e))),
                 }
@@ -73,7 +73,7 @@ impl<A> Future for SelectOk<A> where A: Future {
                 }
                 None => {
                     // based on the filter above, nothing is ready, return
-                    return Ok(Async::NotReady)
+                    return Ok(Async::Pending)
                 },
             }
         }
diff --git a/src/future/shared.rs b/futures-util/src/future/shared.rs
similarity index 96%
rename from src/future/shared.rs
rename to futures-util/src/future/shared.rs
index 9b6081ad..d6ecba4c 100644
--- a/src/future/shared.rs
+++ b/futures-util/src/future/shared.rs
@@ -14,8 +14,7 @@
 //! ```
 
 use {Future, Poll, Async};
-use task::{self, Task};
-use executor::{self, Notify, Spawn};
+use task::{self, Notify, Spawn, Task};
 
 use std::{error, fmt, mem, ops};
 use std::cell::UnsafeCell;
@@ -71,7 +70,7 @@ pub fn new<F: Future>(future: F) -> Shared<F> {
                 state: AtomicUsize::new(IDLE),
                 waiters: Mutex::new(HashMap::new()),
             }),
-            future: UnsafeCell::new(Some(executor::spawn(future))),
+            future: UnsafeCell::new(Some(task::spawn(future))),
             result: UnsafeCell::new(None),
         }),
         waiter: 0,
@@ -129,7 +128,7 @@ impl<F> Future for Shared<F>
                 // Another task is currently polling, at this point we just want
                 // to ensure that our task handle is currently registered
 
-                return Ok(Async::NotReady);
+                return Ok(Async::Pending);
             }
             COMPLETE => {
                 return unsafe { self.clone_result().map(Async::Ready) };
@@ -159,12 +158,12 @@ impl<F> Future for Shared<F>
                     .poll_future_notify(&self.inner.notifier, 0)
             };
             match res {
-                Ok(Async::NotReady) => {
+                Ok(Async::Pending) => {
                     // Not ready, try to release the handle
                     match self.inner.notifier.state.compare_and_swap(POLLING, IDLE, SeqCst) {
                         POLLING => {
                             // Success
-                            return Ok(Async::NotReady);
+                            return Ok(Async::Pending);
                         }
                         REPOLL => {
                             // Gotta poll again!
diff --git a/src/future/then.rs b/futures-util/src/future/then.rs
similarity index 100%
rename from src/future/then.rs
rename to futures-util/src/future/then.rs
diff --git a/futures-util/src/lib.rs b/futures-util/src/lib.rs
new file mode 100644
index 00000000..b6c7f571
--- /dev/null
+++ b/futures-util/src/lib.rs
@@ -0,0 +1,42 @@
+//! Combinators and utilities for working with futures-rs `Futures`s, `Stream`s, and `Sink`s.
+
+#![no_std]
+#![deny(missing_docs, missing_debug_implementations)]
+#![doc(html_root_url = "https://docs.rs/futures/0.1")]
+
+#[macro_use]
+extern crate futures_core;
+extern crate futures_executor;
+extern crate futures_sink;
+
+macro_rules! if_std {
+    ($($i:item)*) => ($(
+        #[cfg(feature = "std")]
+        $i
+    )*)
+}
+
+use futures_core::{Async, Future, IntoFuture, Poll, Stream};
+use futures_core::task;
+use futures_sink::{AsyncSink, Sink, StartSend};
+
+#[macro_use]
+#[cfg(feature = "std")]
+extern crate std;
+
+#[cfg(feature = "std")]
+pub mod lock;
+
+pub mod future;
+pub use future::FutureExt;
+
+pub mod stream;
+pub use stream::StreamExt;
+
+pub mod sink;
+pub use sink::SinkExt;
+
+pub mod prelude {
+    //! Prelude with common traits from the `futures-util` crate.
+    pub use {FutureExt, StreamExt, SinkExt};
+}
diff --git a/src/sync/bilock.rs b/futures-util/src/lock.rs
similarity index 96%
rename from src/sync/bilock.rs
rename to futures-util/src/lock.rs
index af9e1eeb..c0cddcba 100644
--- a/src/sync/bilock.rs
+++ b/futures-util/src/lock.rs
@@ -1,3 +1,5 @@
+//! Futures-powered synchronization primitives.
+
 use std::any::Any;
 use std::boxed::Box;
 use std::cell::UnsafeCell;
@@ -59,7 +61,7 @@ impl<T> BiLock<T> {
         (BiLock { inner: inner.clone() }, BiLock { inner: inner })
     }
 
-    /// Attempt to acquire this lock, returning `NotReady` if it can't be
+    /// Attempt to acquire this lock, returning `Pending` if it can't be
     /// acquired.
     ///
     /// This function will acquire the lock in a nonblocking fashion, returning
@@ -69,7 +71,7 @@ impl<T> BiLock<T> {
     /// lock is unlocked when the returned `BiLockGuard` is dropped.
     ///
     /// If the lock is already held then this function will return
-    /// `Async::NotReady`. In this case the current task will also be scheduled
+    /// `Async::Pending`. In this case the current task will also be scheduled
     /// to receive a notification when the lock would otherwise become
     /// available.
     ///
@@ -99,7 +101,7 @@ impl<T> BiLock<T> {
             match self.inner.state.compare_exchange(1, me, SeqCst, SeqCst) {
                 // The lock is still locked, but we've now parked ourselves, so
                 // just report that we're scheduled to receive a notification.
-                Ok(_) => return Async::NotReady,
+                Ok(_) => return Async::Pending,
 
                 // Oops, looks like the lock was unlocked after our swap above
                 // and before the compare_exchange. Deallocate what we just
@@ -248,7 +250,7 @@ impl<T> Future for BiLockAcquire<T> {
             Async::Ready(r) => {
                 mem::forget(r);
             }
-            Async::NotReady => return Ok(Async::NotReady),
+            Async::Pending => return Ok(Async::Pending),
         }
         Ok(Async::Ready(BiLockAcquired { inner: self.inner.take() }))
     }
diff --git a/src/sink/buffer.rs b/futures-util/src/sink/buffer.rs
similarity index 93%
rename from src/sink/buffer.rs
rename to futures-util/src/sink/buffer.rs
index 034d571c..3b8628e9 100644
--- a/src/sink/buffer.rs
+++ b/futures-util/src/sink/buffer.rs
@@ -2,8 +2,8 @@ use std::collections::VecDeque;
 
 use {Poll, Async};
 use {StartSend, AsyncSink};
-use sink::Sink;
-use stream::Stream;
+use Sink;
+use Stream;
 
 /// Sink for the `Sink::buffer` combinator, which buffers up to some fixed
 /// number of values when the underlying sink is unable to accept them.
@@ -46,13 +46,13 @@ impl<S: Sink> Buffer<S> {
 
     fn try_empty_buffer(&mut self) -> Poll<(), S::SinkError> {
         while let Some(item) = self.buf.pop_front() {
-            if let AsyncSink::NotReady(item) = self.sink.start_send(item)? {
+            if let AsyncSink::Pending(item) = self.sink.start_send(item)? {
                 self.buf.push_front(item);
 
                 // ensure that we attempt to complete any pushes we've started
                 self.sink.poll_complete()?;
 
-                return Ok(Async::NotReady);
+                return Ok(Async::Pending);
             }
         }
 
@@ -81,7 +81,7 @@ impl<S: Sink> Sink for Buffer<S> {
 
         self.try_empty_buffer()?;
         if self.buf.len() == self.cap {
-            return Ok(AsyncSink::NotReady(item));
+            return Ok(AsyncSink::Pending(item));
         }
         self.buf.push_back(item);
         Ok(AsyncSink::Ready)
diff --git a/src/sink/fanout.rs b/futures-util/src/sink/fanout.rs
similarity index 93%
rename from src/sink/fanout.rs
rename to futures-util/src/sink/fanout.rs
index 8d2456e7..10557e7c 100644
--- a/src/sink/fanout.rs
+++ b/futures-util/src/sink/fanout.rs
@@ -62,7 +62,7 @@ impl<A, B> Sink for Fanout<A, B>
             self.right.state = self.right.sink.start_send(item)?;
             Ok(AsyncSink::Ready)
         } else {
-            Ok(AsyncSink::NotReady(item))
+            Ok(AsyncSink::Pending(item))
         }
     }
 
@@ -73,7 +73,7 @@ impl<A, B> Sink for Fanout<A, B>
         if left_async.is_ready() && right_async.is_ready() {
             Ok(Async::Ready(()))
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 
@@ -84,7 +84,7 @@ impl<A, B> Sink for Fanout<A, B>
         if left_async.is_ready() && right_async.is_ready() {
             Ok(Async::Ready(()))
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         } 
     }
 }
@@ -105,7 +105,7 @@ impl<S: Sink> Downstream<S> {
     }
 
     fn keep_flushing(&mut self) -> Result<(), S::SinkError> {
-        if let AsyncSink::NotReady(item) = replace(&mut self.state, AsyncSink::Ready) {
+        if let AsyncSink::Pending(item) = replace(&mut self.state, AsyncSink::Ready) {
             self.state = self.sink.start_send(item)?;
         }
         Ok(())
@@ -119,7 +119,7 @@ impl<S: Sink> Downstream<S> {
         if self.state.is_ready() && async.is_ready() {
             Ok(Async::Ready(()))
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 
@@ -129,7 +129,7 @@ impl<S: Sink> Downstream<S> {
         if self.state.is_ready() {
             self.sink.close()
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 }
diff --git a/src/sink/flush.rs b/futures-util/src/sink/flush.rs
similarity index 96%
rename from src/sink/flush.rs
rename to futures-util/src/sink/flush.rs
index ed152931..26a9a561 100644
--- a/src/sink/flush.rs
+++ b/futures-util/src/sink/flush.rs
@@ -1,5 +1,5 @@
 use {Poll, Async, Future};
-use sink::Sink;
+use Sink;
 
 /// Future for the `Sink::flush` combinator, which polls the sink until all data
 /// has been flushed.
@@ -43,7 +43,7 @@ impl<S: Sink> Future for Flush<S> {
             Ok(Async::Ready(sink))
         } else {
             self.sink = Some(sink);
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 }
diff --git a/src/sink/from_err.rs b/futures-util/src/sink/from_err.rs
similarity index 94%
rename from src/sink/from_err.rs
rename to futures-util/src/sink/from_err.rs
index 92c218fe..0433196d 100644
--- a/src/sink/from_err.rs
+++ b/futures-util/src/sink/from_err.rs
@@ -1,6 +1,6 @@
 use core::marker::PhantomData;
 
-use {Sink, Poll, StartSend};
+use {Sink, Stream, Poll, StartSend};
 
 /// A sink combinator to change the error type of a sink.
 ///
@@ -61,7 +61,7 @@ impl<S, E> Sink for SinkFromErr<S, E>
     }
 }
 
-impl<S: ::stream::Stream, E> ::stream::Stream for SinkFromErr<S, E> {
+impl<S: Stream, E> Stream for SinkFromErr<S, E> {
     type Item = S::Item;
     type Error = S::Error;
 
diff --git a/src/sink/map_err.rs b/futures-util/src/sink/map_err.rs
similarity index 99%
rename from src/sink/map_err.rs
rename to futures-util/src/sink/map_err.rs
index cccad399..0f2a2156 100644
--- a/src/sink/map_err.rs
+++ b/futures-util/src/sink/map_err.rs
@@ -1,4 +1,4 @@
-use sink::Sink;
+use Sink;
 
 use {Poll, StartSend, Stream};
 
diff --git a/futures-util/src/sink/mod.rs b/futures-util/src/sink/mod.rs
new file mode 100644
index 00000000..938c2150
--- /dev/null
+++ b/futures-util/src/sink/mod.rs
@@ -0,0 +1,211 @@
+//! Asynchronous sinks
+//!
+//! This module contains the `Sink` trait, along with a number of adapter types
+//! for it. An overview is available in the documentation for the trait itself.
+//!
+//! You can find more information/tutorials about streams [online at
+//! https://tokio.rs][online]
+//!
+//! [online]: https://tokio.rs/docs/getting-started/streams-and-sinks/
+
+use {IntoFuture, Sink};
+use Stream;
+
+mod with;
+mod with_flat_map;
+mod flush;
+mod from_err;
+mod send;
+mod send_all;
+mod map_err;
+mod fanout;
+
+if_std! {
+    mod buffer;
+    pub use self::buffer::Buffer;
+}
+
+pub use self::with::With;
+pub use self::with_flat_map::WithFlatMap;
+pub use self::flush::Flush;
+pub use self::send::Send;
+pub use self::send_all::SendAll;
+pub use self::map_err::SinkMapErr;
+pub use self::from_err::SinkFromErr;
+pub use self::fanout::Fanout;
+
+impl<T: ?Sized> SinkExt for T where T: Sink {}
+
+/// An extension trait for `Sink`s that provides a variety of convenient
+/// combinator functions.
+pub trait SinkExt: Sink {
+    /// Composes a function *in front of* the sink.
+    ///
+    /// This adapter produces a new sink that passes each value through the
+    /// given function `f` before sending it to `self`.
+    ///
+    /// To process each value, `f` produces a *future*, which is then polled to
+    /// completion before passing its result down to the underlying sink. If the
+    /// future produces an error, that error is returned by the new sink.
+    ///
+    /// Note that this function consumes the given sink, returning a wrapped
+    /// version, much like `Iterator::map`.
+    fn with<U, F, Fut>(self, f: F) -> With<Self, U, F, Fut>
+        where F: FnMut(U) -> Fut,
+              Fut: IntoFuture<Item = Self::SinkItem>,
+              Fut::Error: From<Self::SinkError>,
+              Self: Sized
+    {
+        with::new(self, f)
+    }
+
+    /// Composes a function *in front of* the sink.
+    ///
+    /// This adapter produces a new sink that passes each value through the
+    /// given function `f` before sending it to `self`.
+    ///
+    /// To process each value, `f` produces a *stream*, of which each value
+    /// is passed to the underlying sink. A new value will not be accepted until
+    /// the stream has been drained
+    ///
+    /// Note that this function consumes the given sink, returning a wrapped
+    /// version, much like `Iterator::flat_map`.
+    ///
+    /// # Examples
+    /// ---
+    /// Using this function with an iterator through use of the `stream::iter_ok()`
+    /// function
+    ///
+    /// ```
+    /// use futures::prelude::*;
+    /// use futures::stream;
+    /// use futures::sync::mpsc;
+    ///
+    /// let (tx, rx) = mpsc::channel::<i32>(5);
+    ///
+    /// let tx = tx.with_flat_map(|x| {
+    ///     stream::iter_ok(vec![42; x].into_iter().map(|y| y))
+    /// });
+    /// tx.send(5).wait().unwrap();
+    /// assert_eq!(rx.collect().wait(), Ok(vec![42, 42, 42, 42, 42]))
+    /// ```
+    fn with_flat_map<U, F, St>(self, f: F) -> WithFlatMap<Self, U, F, St>
+        where F: FnMut(U) -> St,
+              St: Stream<Item = Self::SinkItem, Error=Self::SinkError>,
+              Self: Sized
+        {
+            with_flat_map::new(self, f)
+        }
+
+    /*
+    fn with_map<U, F>(self, f: F) -> WithMap<Self, U, F>
+        where F: FnMut(U) -> Self::SinkItem,
+              Self: Sized;
+
+    fn with_filter<F>(self, f: F) -> WithFilter<Self, F>
+        where F: FnMut(Self::SinkItem) -> bool,
+              Self: Sized;
+
+    fn with_filter_map<U, F>(self, f: F) -> WithFilterMap<Self, U, F>
+        where F: FnMut(U) -> Option<Self::SinkItem>,
+              Self: Sized;
+     */
+
+    /// Transforms the error returned by the sink.
+    fn sink_map_err<F, E>(self, f: F) -> SinkMapErr<Self, F>
+        where F: FnOnce(Self::SinkError) -> E,
+              Self: Sized,
+    {
+        map_err::new(self, f)
+    }
+
+    /// Map this sink's error to any error implementing `From` for this sink's
+    /// `Error`, returning a new sink.
+    ///
+    /// If wanting to map errors of a `Sink + Stream`, use `.sink_from_err().from_err()`.
+    fn sink_from_err<E: From<Self::SinkError>>(self) -> from_err::SinkFromErr<Self, E>
+        where Self: Sized,
+    {
+        from_err::new(self)
+    }
+
+
+    /// Adds a fixed-size buffer to the current sink.
+    ///
+    /// The resulting sink will buffer up to `amt` items when the underlying
+    /// sink is unwilling to accept additional items. Calling `poll_complete` on
+    /// the buffered sink will attempt to both empty the buffer and complete
+    /// processing on the underlying sink.
+    ///
+    /// Note that this function consumes the given sink, returning a wrapped
+    /// version, much like `Iterator::map`.
+    ///
+    /// This method is only available when the `std` feature of this
+    /// library is activated, and it is activated by default.
+    #[cfg(feature = "std")]
+    fn buffer(self, amt: usize) -> Buffer<Self>
+        where Self: Sized
+    {
+        buffer::new(self, amt)
+    }
+
+    /// Fanout items to multiple sinks.
+    ///
+    /// This adapter clones each incoming item and forwards it to both this as well as
+    /// the other sink at the same time.
+    fn fanout<S>(self, other: S) -> Fanout<Self, S>
+        where Self: Sized,
+              Self::SinkItem: Clone,
+              S: Sink<SinkItem=Self::SinkItem, SinkError=Self::SinkError>
+    {
+        fanout::new(self, other)
+    }
+
+    /// A future that completes when the sink has finished processing all
+    /// pending requests.
+    ///
+    /// The sink itself is returned after flushing is complete; this adapter is
+    /// intended to be used when you want to stop sending to the sink until
+    /// all current requests are processed.
+    fn flush(self) -> Flush<Self>
+        where Self: Sized
+    {
+        flush::new(self)
+    }
+
+    /// A future that completes after the given item has been fully processed
+    /// into the sink, including flushing.
+    ///
+    /// Note that, **because of the flushing requirement, it is usually better
+    /// to batch together items to send via `send_all`, rather than flushing
+    /// between each item.**
+    ///
+    /// On completion, the sink is returned.
+    fn send(self, item: Self::SinkItem) -> Send<Self>
+        where Self: Sized
+    {
+        send::new(self, item)
+    }
+
+    /// A future that completes after the given stream has been fully processed
+    /// into the sink, including flushing.
+    ///
+    /// This future will drive the stream to keep producing items until it is
+    /// exhausted, sending each item to the sink. It will complete once both the
+    /// stream is exhausted, the sink has received all items, the sink has been
+    /// flushed, and the sink has been closed.
+    ///
+    /// Doing `sink.send_all(stream)` is roughly equivalent to
+    /// `stream.forward(sink)`. The returned future will exhaust all items from
+    /// `stream` and send them to `self`, closing `self` when all items have been
+    /// received.
+    ///
+    /// On completion, the pair `(sink, source)` is returned.
+    fn send_all<S>(self, stream: S) -> SendAll<Self, S>
+        where S: Stream<Item = Self::SinkItem>,
+              Self::SinkError: From<S::Error>,
+              Self: Sized
+    {
+        send_all::new(self, stream)
+    }
+}
diff --git a/src/sink/send.rs b/futures-util/src/sink/send.rs
similarity index 91%
rename from src/sink/send.rs
rename to futures-util/src/sink/send.rs
index 71173fa8..942b9635 100644
--- a/src/sink/send.rs
+++ b/futures-util/src/sink/send.rs
@@ -1,5 +1,5 @@
 use {Poll, Async, Future, AsyncSink};
-use sink::Sink;
+use Sink;
 
 /// Future for the `Sink::send` combinator, which sends a value to a sink and
 /// then waits until the sink has fully flushed.
@@ -43,9 +43,9 @@ impl<S: Sink> Future for Send<S> {
 
     fn poll(&mut self) -> Poll<S, S::SinkError> {
         if let Some(item) = self.item.take() {
-            if let AsyncSink::NotReady(item) = self.sink_mut().start_send(item)? {
+            if let AsyncSink::Pending(item) = self.sink_mut().start_send(item)? {
                 self.item = Some(item);
-                return Ok(Async::NotReady);
+                return Ok(Async::Pending);
             }
         }
 
diff --git a/src/sink/send_all.rs b/futures-util/src/sink/send_all.rs
similarity index 89%
rename from src/sink/send_all.rs
rename to futures-util/src/sink/send_all.rs
index a230903d..02722ffb 100644
--- a/src/sink/send_all.rs
+++ b/futures-util/src/sink/send_all.rs
@@ -1,6 +1,6 @@
-use {Poll, Async, Future, AsyncSink};
-use stream::{Stream, Fuse};
-use sink::Sink;
+use {Poll, Async, Future, AsyncSink, Stream};
+use stream::{StreamExt, Fuse};
+use Sink;
 
 /// Future for the `Sink::send_all` combinator, which sends a stream of values
 /// to a sink and then waits until the sink has fully flushed those values.
@@ -48,9 +48,9 @@ impl<T, U> SendAll<T, U>
 
     fn try_start_send(&mut self, item: U::Item) -> Poll<(), T::SinkError> {
         debug_assert!(self.buffered.is_none());
-        if let AsyncSink::NotReady(item) = self.sink_mut().start_send(item)? {
+        if let AsyncSink::Pending(item) = self.sink_mut().start_send(item)? {
             self.buffered = Some(item);
-            return Ok(Async::NotReady)
+            return Ok(Async::Pending)
         }
         Ok(Async::Ready(()))
     }
@@ -78,9 +78,9 @@ impl<T, U> Future for SendAll<T, U>
                     try_ready!(self.sink_mut().close());
                     return Ok(Async::Ready(self.take_result()))
                 }
-                Async::NotReady => {
+                Async::Pending => {
                     try_ready!(self.sink_mut().poll_complete());
-                    return Ok(Async::NotReady)
+                    return Ok(Async::Pending)
                 }
             }
         }
diff --git a/src/sink/wait.rs b/futures-util/src/sink/wait.rs
similarity index 88%
rename from src/sink/wait.rs
rename to futures-util/src/sink/wait.rs
index 940a5886..403f7b0e 100644
--- a/src/sink/wait.rs
+++ b/futures-util/src/sink/wait.rs
@@ -1,4 +1,4 @@
-use sink::Sink;
+use Sink;
 use executor;
 
 /// A sink combinator which converts an asynchronous sink to a **blocking
@@ -25,7 +25,7 @@ impl<S: Sink> Wait<S> {
     ///
     /// 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
+    /// the function returns `Pending` then the current thread is blocked
     /// until it is otherwise ready to accept the value.
     ///
     /// # Return value
@@ -42,7 +42,7 @@ impl<S: Sink> Wait<S> {
     ///
     /// 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
+    /// `Pending` the current thread will be blocked until it's otherwise
     /// ready to proceed.
     pub fn flush(&mut self) -> Result<(), S::SinkError> {
         self.sink.wait_flush()
@@ -52,7 +52,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.
+    /// `Pending` the current thread will be blocked until it's otherwise closed.
     pub fn close(&mut self) -> Result<(), S::SinkError> {
         self.sink.wait_close()
     }
diff --git a/src/sink/with.rs b/futures-util/src/sink/with.rs
similarity index 94%
rename from src/sink/with.rs
rename to futures-util/src/sink/with.rs
index b0d5c54c..12146eec 100644
--- a/src/sink/with.rs
+++ b/futures-util/src/sink/with.rs
@@ -2,8 +2,8 @@ use core::mem;
 use core::marker::PhantomData;
 
 use {IntoFuture, Future, Poll, Async, StartSend, AsyncSink};
-use sink::Sink;
-use stream::Stream;
+use Sink;
+use Stream;
 
 /// Sink for the `Sink::with` combinator, chaining a computation to run *prior*
 /// to pushing a value into the underlying sink.
@@ -98,14 +98,14 @@ impl<S, U, F, Fut> With<S, U, F, Fut>
                         Async::Ready(item) => {
                             self.state = State::Buffered(item);
                         }
-                        Async::NotReady => {
+                        Async::Pending => {
                             self.state = State::Process(fut);
                             break
                         }
                     }
                 }
                 State::Buffered(item) => {
-                    if let AsyncSink::NotReady(item) = self.sink.start_send(item)? {
+                    if let AsyncSink::Pending(item) = self.sink.start_send(item)? {
                         self.state = State::Buffered(item);
                         break
                     }
@@ -116,7 +116,7 @@ impl<S, U, F, Fut> With<S, U, F, Fut>
         if self.state.is_empty() {
             Ok(Async::Ready(()))
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 }
@@ -132,7 +132,7 @@ impl<S, U, F, Fut> Sink for With<S, U, F, Fut>
 
     fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Fut::Error> {
         if self.poll()?.is_not_ready() {
-            return Ok(AsyncSink::NotReady(item))
+            return Ok(AsyncSink::Pending(item))
         }
         self.state = State::Process((self.f)(item).into_future());
         Ok(AsyncSink::Ready)
diff --git a/src/sink/with_flat_map.rs b/futures-util/src/sink/with_flat_map.rs
similarity index 88%
rename from src/sink/with_flat_map.rs
rename to futures-util/src/sink/with_flat_map.rs
index 80c4f660..2dde711b 100644
--- a/src/sink/with_flat_map.rs
+++ b/futures-util/src/sink/with_flat_map.rs
@@ -1,8 +1,8 @@
 use core::marker::PhantomData;
 
 use {Poll, Async, StartSend, AsyncSink};
-use sink::Sink;
-use stream::Stream;
+use Sink;
+use Stream;
 
 /// Sink for the `Sink::with_flat_map` combinator, chaining a computation that returns an iterator
 /// to run prior to pushing a value into the underlying sink
@@ -62,17 +62,17 @@ where
 
     fn try_empty_stream(&mut self) -> Poll<(), S::SinkError> {
         if let Some(x) = self.buffer.take() {
-            if let AsyncSink::NotReady(x) = self.sink.start_send(x)? {
+            if let AsyncSink::Pending(x) = self.sink.start_send(x)? {
                 self.buffer = Some(x);
-                return Ok(Async::NotReady);
+                return Ok(Async::Pending);
             }
         }
         if let Some(mut stream) = self.stream.take() {
             while let Some(x) = try_ready!(stream.poll()) {
-                if let AsyncSink::NotReady(x) = self.sink.start_send(x)? {
+                if let AsyncSink::Pending(x) = self.sink.start_send(x)? {
                     self.stream = Some(stream);
                     self.buffer = Some(x);
-                    return Ok(Async::NotReady);
+                    return Ok(Async::Pending);
                 }
             }
         }
@@ -103,7 +103,7 @@ where
     type SinkError = S::SinkError;
     fn start_send(&mut self, i: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
         if self.try_empty_stream()?.is_not_ready() {
-            return Ok(AsyncSink::NotReady(i));
+            return Ok(AsyncSink::Pending(i));
         }
         assert!(self.stream.is_none());
         self.stream = Some((self.f)(i));
@@ -112,13 +112,13 @@ where
     }
     fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
         if self.try_empty_stream()?.is_not_ready() {
-            return Ok(Async::NotReady);
+            return Ok(Async::Pending);
         }
         self.sink.poll_complete()
     }
     fn close(&mut self) -> Poll<(), Self::SinkError> {
         if self.try_empty_stream()?.is_not_ready() {
-            return Ok(Async::NotReady);
+            return Ok(Async::Pending);
         }
         assert!(self.stream.is_none());
         self.sink.close()
diff --git a/src/stream/and_then.rs b/futures-util/src/stream/and_then.rs
similarity index 94%
rename from src/stream/and_then.rs
rename to futures-util/src/stream/and_then.rs
index 1fac8b95..ad57c445 100644
--- a/src/stream/and_then.rs
+++ b/futures-util/src/stream/and_then.rs
@@ -1,5 +1,5 @@
 use {IntoFuture, Future, Poll, Async};
-use stream::Stream;
+use Stream;
 
 /// A stream combinator which chains a computation onto values produced by a
 /// stream.
@@ -55,8 +55,8 @@ impl<S, F, U> AndThen<S, F, U>
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S, F, U: IntoFuture> ::sink::Sink for AndThen<S, F, U>
-    where S: ::sink::Sink
+impl<S, F, U: IntoFuture> ::Sink for AndThen<S, F, U>
+    where S: ::Sink
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
@@ -100,7 +100,7 @@ impl<S, F, U> Stream for AndThen<S, F, U>
                 self.future = None;
                 Err(e)
             }
-            Ok(Async::NotReady) => Ok(Async::NotReady)
+            Ok(Async::Pending) => Ok(Async::Pending)
         }
     }
 }
diff --git a/src/stream/buffer_unordered.rs b/futures-util/src/stream/buffer_unordered.rs
similarity index 96%
rename from src/stream/buffer_unordered.rs
rename to futures-util/src/stream/buffer_unordered.rs
index 3011108c..38bbae28 100644
--- a/src/stream/buffer_unordered.rs
+++ b/futures-util/src/stream/buffer_unordered.rs
@@ -86,7 +86,7 @@ impl<S> Stream for BufferUnordered<S>
             let future = match self.stream.poll()? {
                 Async::Ready(Some(s)) => s.into_future(),
                 Async::Ready(None) |
-                Async::NotReady => break,
+                Async::Pending => break,
             };
 
             self.queue.push(future);
@@ -103,14 +103,14 @@ impl<S> Stream for BufferUnordered<S>
         if self.stream.is_done() {
             Ok(Async::Ready(None))
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S> ::sink::Sink for BufferUnordered<S>
-    where S: ::sink::Sink + Stream,
+impl<S> ::Sink for BufferUnordered<S>
+    where S: ::Sink + Stream,
           S::Item: IntoFuture,
 {
     type SinkItem = S::SinkItem;
diff --git a/src/stream/buffered.rs b/futures-util/src/stream/buffered.rs
similarity index 96%
rename from src/stream/buffered.rs
rename to futures-util/src/stream/buffered.rs
index 5616b73d..422acb7c 100644
--- a/src/stream/buffered.rs
+++ b/futures-util/src/stream/buffered.rs
@@ -75,8 +75,8 @@ impl<S> Buffered<S>
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S> ::sink::Sink for Buffered<S>
-    where S: ::sink::Sink + Stream,
+impl<S> ::Sink for Buffered<S>
+    where S: ::Sink + Stream,
           S::Item: IntoFuture,
 {
     type SinkItem = S::SinkItem;
@@ -109,7 +109,7 @@ impl<S> Stream for Buffered<S>
             let future = match self.stream.poll()? {
                 Async::Ready(Some(s)) => s.into_future(),
                 Async::Ready(None) |
-                Async::NotReady => break,
+                Async::Pending => break,
             };
 
             self.queue.push(future);
@@ -126,7 +126,7 @@ impl<S> Stream for Buffered<S>
         if self.stream.is_done() {
             Ok(Async::Ready(None))
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 }
diff --git a/src/stream/catch_unwind.rs b/futures-util/src/stream/catch_unwind.rs
similarity index 84%
rename from src/stream/catch_unwind.rs
rename to futures-util/src/stream/catch_unwind.rs
index d3244946..02938259 100644
--- a/src/stream/catch_unwind.rs
+++ b/futures-util/src/stream/catch_unwind.rs
@@ -1,6 +1,6 @@
 use std::prelude::v1::*;
 use std::any::Any;
-use std::panic::{catch_unwind, UnwindSafe, AssertUnwindSafe};
+use std::panic::{catch_unwind, UnwindSafe};
 use std::mem;
 
 use super::super::{Poll, Async};
@@ -52,7 +52,7 @@ impl<S> Stream for CatchUnwind<S>
                 self.state = CatchUnwindState::Stream(stream);
                 match poll {
                     Err(e) => Ok(Async::Ready(Some(Err(e)))),
-                    Ok(Async::NotReady) => Ok(Async::NotReady),
+                    Ok(Async::Pending) => Ok(Async::Pending),
                     Ok(Async::Ready(Some(r))) => Ok(Async::Ready(Some(Ok(r)))),
                     Ok(Async::Ready(None)) => Ok(Async::Ready(None)),
                 }
@@ -60,12 +60,3 @@ impl<S> Stream for CatchUnwind<S>
         }
     }
 }
-
-impl<S: Stream> Stream for AssertUnwindSafe<S> {
-    type Item = S::Item;
-    type Error = S::Error;
-
-    fn poll(&mut self) -> Poll<Option<S::Item>, S::Error> {
-        self.0.poll()
-    }
-}
diff --git a/src/stream/chain.rs b/futures-util/src/stream/chain.rs
similarity index 98%
rename from src/stream/chain.rs
rename to futures-util/src/stream/chain.rs
index 0ff0e5ce..e56cb2f2 100644
--- a/src/stream/chain.rs
+++ b/futures-util/src/stream/chain.rs
@@ -1,6 +1,6 @@
 use core::mem;
 
-use stream::Stream;
+use Stream;
 use {Async, Poll};
 
 
diff --git a/src/stream/chunks.rs b/futures-util/src/stream/chunks.rs
similarity index 96%
rename from src/stream/chunks.rs
rename to futures-util/src/stream/chunks.rs
index 3a361eb6..bc618c38 100644
--- a/src/stream/chunks.rs
+++ b/futures-util/src/stream/chunks.rs
@@ -32,8 +32,8 @@ pub fn new<S>(s: S, capacity: usize) -> Chunks<S>
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S> ::sink::Sink for Chunks<S>
-    where S: ::sink::Sink + Stream
+impl<S> ::Sink for Chunks<S>
+    where S: ::Sink + Stream
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
@@ -96,7 +96,7 @@ impl<S> Stream for Chunks<S>
         let cap = self.items.capacity();
         loop {
             match self.stream.poll() {
-                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Ok(Async::Pending) => return Ok(Async::Pending),
 
                 // Push the item into the buffer and check whether it is full.
                 // If so, replace our buffer with a new and empty one and return
diff --git a/src/stream/collect.rs b/futures-util/src/stream/collect.rs
similarity index 92%
rename from src/stream/collect.rs
rename to futures-util/src/stream/collect.rs
index 8bd9d0e1..da6208b9 100644
--- a/src/stream/collect.rs
+++ b/futures-util/src/stream/collect.rs
@@ -3,7 +3,7 @@ use std::prelude::v1::*;
 use std::mem;
 
 use {Future, Poll, Async};
-use stream::Stream;
+use Stream;
 
 /// A future which collects all of the values of a stream into a vector.
 ///
@@ -41,7 +41,7 @@ impl<S> Future for Collect<S>
             match self.stream.poll() {
                 Ok(Async::Ready(Some(e))) => self.items.push(e),
                 Ok(Async::Ready(None)) => return Ok(Async::Ready(self.finish())),
-                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Ok(Async::Pending) => return Ok(Async::Pending),
                 Err(e) => {
                     self.finish();
                     return Err(e)
diff --git a/src/stream/concat.rs b/futures-util/src/stream/concat.rs
similarity index 94%
rename from src/stream/concat.rs
rename to futures-util/src/stream/concat.rs
index 164a50f7..105842db 100644
--- a/src/stream/concat.rs
+++ b/futures-util/src/stream/concat.rs
@@ -2,9 +2,7 @@ use core::mem;
 use core::fmt::{Debug, Formatter, Result as FmtResult};
 use core::default::Default;
 
-use {Poll, Async};
-use future::Future;
-use stream::Stream;
+use {Async, Future, Poll, Stream};
 
 /// A stream combinator to concatenate the results of a stream into the first
 /// yielded item.
@@ -45,7 +43,7 @@ impl<S> Future for Concat<S>
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         self.inner.poll().map(|a| {
             match a {
-                Async::NotReady => Async::NotReady,
+                Async::Pending => Async::Pending,
                 Async::Ready(None) => Async::Ready(Default::default()),
                 Async::Ready(Some(e)) => Async::Ready(e)
             }
@@ -101,7 +99,7 @@ impl<S> Future for ConcatSafe<S>
                         Inner::Done => panic!("cannot poll Concat again")
                     }
                 },
-                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Ok(Async::Pending) => return Ok(Async::Pending),
                 Err(e) => {
                     self.extend = Inner::Done;
                     return Err(e)
diff --git a/src/stream/empty.rs b/futures-util/src/stream/empty.rs
similarity index 97%
rename from src/stream/empty.rs
rename to futures-util/src/stream/empty.rs
index c53fb802..688363d0 100644
--- a/src/stream/empty.rs
+++ b/futures-util/src/stream/empty.rs
@@ -1,6 +1,6 @@
 use core::marker;
 
-use stream::Stream;
+use Stream;
 use {Poll, Async};
 
 /// A stream which contains no elements.
diff --git a/src/stream/filter.rs b/futures-util/src/stream/filter.rs
similarity index 96%
rename from src/stream/filter.rs
rename to futures-util/src/stream/filter.rs
index c07d7a7b..ac792711 100644
--- a/src/stream/filter.rs
+++ b/futures-util/src/stream/filter.rs
@@ -1,5 +1,5 @@
 use {Async, Future, IntoFuture, Poll, Sink};
-use stream::Stream;
+use Stream;
 
 /// A stream combinator used to filter the results of a stream and only yield
 /// some values.
@@ -59,7 +59,7 @@ impl<S, P, R> Filter<S, P, R>
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S, P, R> ::sink::Sink for Filter<S, P, R>
+impl<S, P, R> ::Sink for Filter<S, P, R>
     where S: Stream,
           P: FnMut(&S::Item) -> R,
           R: IntoFuture<Item=bool, Error=S::Error>,
@@ -106,7 +106,7 @@ impl<S, P, R> Stream for Filter<S, P, R>
                     return Ok(Async::Ready(Some(item)));
                 }
                 Ok(Async::Ready(false)) => self.pending = None,
-                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Ok(Async::Pending) => return Ok(Async::Pending),
                 Err(e) => {
                     self.pending = None;
                     return Err(e)
diff --git a/src/stream/filter_map.rs b/futures-util/src/stream/filter_map.rs
similarity index 94%
rename from src/stream/filter_map.rs
rename to futures-util/src/stream/filter_map.rs
index 79bc57f1..db293d74 100644
--- a/src/stream/filter_map.rs
+++ b/futures-util/src/stream/filter_map.rs
@@ -1,5 +1,5 @@
 use {Async, Future, IntoFuture, Poll};
-use stream::Stream;
+use Stream;
 
 /// A combinator used to filter the results of a stream and simultaneously map
 /// them to a different type.
@@ -59,8 +59,8 @@ impl<S, F, R> FilterMap<S, F, R>
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S, F, R> ::sink::Sink for FilterMap<S, F, R>
-    where S: Stream + ::sink::Sink,
+impl<S, F, R> ::Sink for FilterMap<S, F, R>
+    where S: Stream + ::Sink,
           F: FnMut(S::Item) -> R,
           R: IntoFuture<Error=S::Error>,
 {
@@ -105,7 +105,7 @@ impl<S, F, R, B> Stream for FilterMap<S, F, R>
                     return x
                 }
                 Ok(Async::Ready(None)) => self.pending = None,
-                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Ok(Async::Pending) => return Ok(Async::Pending),
                 Err(e) => {
                     self.pending = None;
                     return Err(e)
diff --git a/src/stream/flatten.rs b/futures-util/src/stream/flatten.rs
similarity index 96%
rename from src/stream/flatten.rs
rename to futures-util/src/stream/flatten.rs
index 74f4e0c6..3f6595a7 100644
--- a/src/stream/flatten.rs
+++ b/futures-util/src/stream/flatten.rs
@@ -1,5 +1,5 @@
 use {Poll, Async};
-use stream::Stream;
+use Stream;
 
 /// A combinator used to flatten a stream-of-streams into one long stream of
 /// elements.
@@ -50,8 +50,8 @@ impl<S: Stream> Flatten<S> {
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S> ::sink::Sink for Flatten<S>
-    where S: ::sink::Sink + Stream
+impl<S> ::Sink for Flatten<S>
+    where S: ::Sink + Stream
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
diff --git a/src/stream/fold.rs b/futures-util/src/stream/fold.rs
similarity index 91%
rename from src/stream/fold.rs
rename to futures-util/src/stream/fold.rs
index 15a0a940..6b18e8e8 100644
--- a/src/stream/fold.rs
+++ b/futures-util/src/stream/fold.rs
@@ -1,7 +1,7 @@
 use core::mem;
 
 use {Future, Poll, IntoFuture, Async};
-use stream::Stream;
+use Stream;
 
 /// A future used to collect all the results of a stream into one generic type.
 ///
@@ -58,18 +58,18 @@ impl<S, F, Fut, T> Future for Fold<S, F, Fut, T>
                             self.state = State::Processing(future);
                         }
                         Async::Ready(None) => return Ok(Async::Ready(state)),
-                        Async::NotReady => {
+                        Async::Pending => {
                             self.state = State::Ready(state);
-                            return Ok(Async::NotReady)
+                            return Ok(Async::Pending)
                         }
                     }
                 }
                 State::Processing(mut fut) => {
                     match fut.poll()? {
                         Async::Ready(state) => self.state = State::Ready(state),
-                        Async::NotReady => {
+                        Async::Pending => {
                             self.state = State::Processing(fut);
-                            return Ok(Async::NotReady)
+                            return Ok(Async::Pending)
                         }
                     }
                 }
diff --git a/src/stream/for_each.rs b/futures-util/src/stream/for_each.rs
similarity index 94%
rename from src/stream/for_each.rs
rename to futures-util/src/stream/for_each.rs
index c7e1cde5..e19dbb47 100644
--- a/src/stream/for_each.rs
+++ b/futures-util/src/stream/for_each.rs
@@ -1,5 +1,5 @@
 use {Async, Future, IntoFuture, Poll};
-use stream::Stream;
+use Stream;
 
 /// A stream combinator which executes a unit closure over each item on a
 /// stream.
@@ -38,7 +38,7 @@ impl<S, F, U> Future for ForEach<S, F, U>
             if let Some(mut fut) = self.fut.take() {
                 if fut.poll()?.is_not_ready() {
                     self.fut = Some(fut);
-                    return Ok(Async::NotReady);
+                    return Ok(Async::Pending);
                 }
             }
 
diff --git a/src/stream/forward.rs b/futures-util/src/stream/forward.rs
similarity index 92%
rename from src/stream/forward.rs
rename to futures-util/src/stream/forward.rs
index 2ecde100..b08a5137 100644
--- a/src/stream/forward.rs
+++ b/futures-util/src/stream/forward.rs
@@ -1,6 +1,6 @@
-use {Poll, Async, Future, AsyncSink};
-use stream::{Stream, Fuse};
-use sink::Sink;
+use {Async, AsyncSink, Future, Poll, Stream};
+use stream::{StreamExt, Fuse};
+use Sink;
 
 /// Future for the `Stream::forward` combinator, which sends a stream of values
 /// to a sink and then waits until the sink has fully flushed those values.
@@ -64,12 +64,12 @@ impl<T, U> Forward<T, U>
 
     fn try_start_send(&mut self, item: T::Item) -> Poll<(), U::SinkError> {
         debug_assert!(self.buffered.is_none());
-        if let AsyncSink::NotReady(item) = self.sink_mut()
+        if let AsyncSink::Pending(item) = self.sink_mut()
             .take().expect("Attempted to poll Forward after completion")
             .start_send(item)?
         {
             self.buffered = Some(item);
-            return Ok(Async::NotReady)
+            return Ok(Async::Pending)
         }
         Ok(Async::Ready(()))
     }
@@ -100,9 +100,9 @@ impl<T, U> Future for Forward<T, U>
                     try_ready!(self.sink_mut().take().expect("Attempted to poll Forward after completion").close());
                     return Ok(Async::Ready(self.take_result()))
                 }
-                Async::NotReady => {
+                Async::Pending => {
                     try_ready!(self.sink_mut().take().expect("Attempted to poll Forward after completion").poll_complete());
-                    return Ok(Async::NotReady)
+                    return Ok(Async::Pending)
                 }
             }
         }
diff --git a/src/stream/from_err.rs b/futures-util/src/stream/from_err.rs
similarity index 91%
rename from src/stream/from_err.rs
rename to futures-util/src/stream/from_err.rs
index 4028542d..dace5f65 100644
--- a/src/stream/from_err.rs
+++ b/futures-util/src/stream/from_err.rs
@@ -1,7 +1,5 @@
 use core::marker::PhantomData;
-use poll::Poll;
-use Async;
-use stream::Stream;
+use {Async, Poll, Stream};
 
 /// A stream combinator to change the error type of a stream.
 ///
@@ -54,7 +52,7 @@ impl<S: Stream, E: From<S::Error>> Stream for FromErr<S, E> {
 
     fn poll(&mut self) -> Poll<Option<S::Item>, E> {
         let e = match self.stream.poll() {
-            Ok(Async::NotReady) => return Ok(Async::NotReady),
+            Ok(Async::Pending) => return Ok(Async::Pending),
             other => other,
         };
         e.map_err(From::from)
@@ -62,7 +60,7 @@ impl<S: Stream, E: From<S::Error>> Stream for FromErr<S, E> {
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S: Stream + ::sink::Sink, E> ::sink::Sink for FromErr<S, E> {
+impl<S: Stream + ::Sink, E> ::Sink for FromErr<S, E> {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
diff --git a/src/stream/fuse.rs b/futures-util/src/stream/fuse.rs
similarity index 96%
rename from src/stream/fuse.rs
rename to futures-util/src/stream/fuse.rs
index e39c31f3..d34dc9d8 100644
--- a/src/stream/fuse.rs
+++ b/futures-util/src/stream/fuse.rs
@@ -1,5 +1,5 @@
 use {Poll, Async};
-use stream::Stream;
+use Stream;
 
 /// A stream which "fuse"s a stream once it's terminated.
 ///
@@ -14,8 +14,8 @@ pub struct Fuse<S> {
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S> ::sink::Sink for Fuse<S>
-    where S: ::sink::Sink
+impl<S> ::Sink for Fuse<S>
+    where S: ::Sink
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
diff --git a/src/stream/future.rs b/futures-util/src/stream/future.rs
similarity index 96%
rename from src/stream/future.rs
rename to futures-util/src/stream/future.rs
index 5b052ee4..33a6d958 100644
--- a/src/stream/future.rs
+++ b/futures-util/src/stream/future.rs
@@ -1,5 +1,5 @@
 use {Future, Poll, Async};
-use stream::Stream;
+use Stream;
 
 /// A combinator used to temporarily convert a stream into a future.
 ///
@@ -62,7 +62,7 @@ impl<S: Stream> Future for StreamFuture<S> {
         let item = {
             let s = self.stream.as_mut().expect("polling StreamFuture twice");
             match s.poll() {
-                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Ok(Async::Pending) => return Ok(Async::Pending),
                 Ok(Async::Ready(e)) => Ok(e),
                 Err(e) => Err(e),
             }
diff --git a/src/stream/futures_ordered.rs b/futures-util/src/stream/futures_ordered.rs
similarity index 95%
rename from src/stream/futures_ordered.rs
rename to futures-util/src/stream/futures_ordered.rs
index 64e2d6f9..2762c7e7 100644
--- a/src/stream/futures_ordered.rs
+++ b/futures-util/src/stream/futures_ordered.rs
@@ -72,9 +72,9 @@ impl<T> Future for OrderWrapper<T>
 /// the queue is currently not managing any futures. A future may be submitted
 /// to the queue at a later time. At that point, a call to
 /// `FuturesOrdered::poll` will either return the future's resolved value
-/// **or** `Ok(Async::NotReady)` if the future has not yet completed. When
+/// **or** `Ok(Async::Pending)` if the future has not yet completed. When
 /// multiple futures are submitted to the queue, `FuturesOrdered::poll` will
-/// return `Ok(Async::NotReady)` until the first future completes, even if
+/// return `Ok(Async::Pending)` until the first future completes, even if
 /// some of the later futures have already completed.
 ///
 /// Note that you can create a ready-made `FuturesOrdered` via the
@@ -171,17 +171,17 @@ impl<T> Stream for FuturesOrdered<T>
         loop {
             match self.in_progress.poll()? {
                 Async::Ready(Some(result)) => self.queued_results.push(result),
-                Async::Ready(None) | Async::NotReady => break,
+                Async::Ready(None) | Async::Pending => break,
             }
         }
 
         if let Some(next_result) = self.queued_results.peek() {
             // PeekMut::pop is not stable yet QQ
             if next_result.index != self.next_outgoing_index {
-                return Ok(Async::NotReady);
+                return Ok(Async::Pending);
             }
         } else if !self.in_progress.is_empty() {
-            return Ok(Async::NotReady);
+            return Ok(Async::Pending);
         } else {
             return Ok(Async::Ready(None));
         }
diff --git a/src/stream/futures_unordered.rs b/futures-util/src/stream/futures_unordered.rs
similarity index 98%
rename from src/stream/futures_unordered.rs
rename to futures-util/src/stream/futures_unordered.rs
index 2940fd34..0d47d500 100644
--- a/src/stream/futures_unordered.rs
+++ b/futures-util/src/stream/futures_unordered.rs
@@ -12,8 +12,7 @@ use std::sync::{Arc, Weak};
 use std::usize;
 
 use {task, Stream, Future, Poll, Async};
-use executor::{Notify, UnsafeNotify, NotifyHandle};
-use task_impl::{self, AtomicTask};
+use task::{AtomicTask, Notify, UnsafeNotify, NotifyHandle};
 
 /// An unbounded set of futures.
 ///
@@ -38,7 +37,7 @@ use task_impl::{self, AtomicTask};
 /// the set is currently not managing any futures. A future may be submitted
 /// to the set at a later time. At that point, a call to
 /// `FuturesUnordered::poll` will either return the future's resolved value
-/// **or** `Ok(Async::NotReady)` if the future has not yet completed.
+/// **or** `Ok(Async::Pending)` if the future has not yet completed.
 ///
 /// Note that you can create a ready-made `FuturesUnordered` via the
 /// `futures_unordered` function in the `stream` module, or you can start with an
@@ -277,7 +276,7 @@ impl<T> Stream for FuturesUnordered<T>
                     if self.is_empty() {
                         return Ok(Async::Ready(None));
                     } else {
-                        return Ok(Async::NotReady)
+                        return Ok(Async::Pending)
                     }
                 }
                 Dequeue::Inconsistent => {
@@ -285,7 +284,7 @@ impl<T> Stream for FuturesUnordered<T>
                     // spinning a few times... but for now, just yield using the
                     // task system.
                     task::current().notify();
-                    return Ok(Async::NotReady);
+                    return Ok(Async::Pending);
                 }
                 Dequeue::Data(node) => node,
             };
@@ -359,13 +358,13 @@ impl<T> Stream for FuturesUnordered<T>
                 // deallocating the node if need be.
                 let res = {
                     let notify = NodeToHandle(bomb.node.as_ref().unwrap());
-                    task_impl::with_notify(&notify, 0, || {
+                    task::with_notify(&notify, 0, || {
                         future.poll()
                     })
                 };
 
                 let ret = match res {
-                    Ok(Async::NotReady) => {
+                    Ok(Async::Pending) => {
                         let node = bomb.node.take().unwrap();
                         *node.future.get() = Some(future);
                         bomb.queue.link(node);
diff --git a/src/stream/inspect.rs b/futures-util/src/stream/inspect.rs
similarity index 96%
rename from src/stream/inspect.rs
rename to futures-util/src/stream/inspect.rs
index fc8f7f4e..254a148e 100644
--- a/src/stream/inspect.rs
+++ b/futures-util/src/stream/inspect.rs
@@ -46,8 +46,8 @@ impl<S: Stream, F> Inspect<S, F> {
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S, F> ::sink::Sink for Inspect<S, F>
-    where S: ::sink::Sink + Stream
+impl<S, F> ::Sink for Inspect<S, F>
+    where S: ::Sink + Stream
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
diff --git a/src/stream/inspect_err.rs b/futures-util/src/stream/inspect_err.rs
similarity index 96%
rename from src/stream/inspect_err.rs
rename to futures-util/src/stream/inspect_err.rs
index 5c56a217..d495923a 100644
--- a/src/stream/inspect_err.rs
+++ b/futures-util/src/stream/inspect_err.rs
@@ -46,8 +46,8 @@ impl<S: Stream, F> InspectErr<S, F> {
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S, F> ::sink::Sink for InspectErr<S, F>
-    where S: ::sink::Sink + Stream
+impl<S, F> ::Sink for InspectErr<S, F>
+    where S: ::Sink + Stream
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
diff --git a/src/stream/iter_ok.rs b/futures-util/src/stream/iter_ok.rs
similarity index 98%
rename from src/stream/iter_ok.rs
rename to futures-util/src/stream/iter_ok.rs
index 9c8d8713..ec33b50a 100644
--- a/src/stream/iter_ok.rs
+++ b/futures-util/src/stream/iter_ok.rs
@@ -1,7 +1,7 @@
 use core::marker;
 
 use {Async, Poll};
-use stream::Stream;
+use Stream;
 
 /// A stream which is just a shim over an underlying instance of `Iterator`.
 ///
diff --git a/src/stream/iter_result.rs b/futures-util/src/stream/iter_result.rs
similarity index 98%
rename from src/stream/iter_result.rs
rename to futures-util/src/stream/iter_result.rs
index 4eef5da0..8d595414 100644
--- a/src/stream/iter_result.rs
+++ b/futures-util/src/stream/iter_result.rs
@@ -1,5 +1,5 @@
 use {Async, Poll};
-use stream::Stream;
+use Stream;
 
 /// A stream which is just a shim over an underlying instance of `Iterator`.
 ///
diff --git a/src/stream/map.rs b/futures-util/src/stream/map.rs
similarity index 95%
rename from src/stream/map.rs
rename to futures-util/src/stream/map.rs
index 702e980b..357d5196 100644
--- a/src/stream/map.rs
+++ b/futures-util/src/stream/map.rs
@@ -1,5 +1,5 @@
 use {Async, Poll};
-use stream::Stream;
+use Stream;
 
 /// A stream combinator which will change the type of a stream from one
 /// type to another.
@@ -48,8 +48,8 @@ impl<S, F> Map<S, F> {
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S, F> ::sink::Sink for Map<S, F>
-    where S: ::sink::Sink
+impl<S, F> ::Sink for Map<S, F>
+    where S: ::Sink
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
diff --git a/src/stream/map_err.rs b/futures-util/src/stream/map_err.rs
similarity index 95%
rename from src/stream/map_err.rs
rename to futures-util/src/stream/map_err.rs
index 8d1c0fc0..59bc789d 100644
--- a/src/stream/map_err.rs
+++ b/futures-util/src/stream/map_err.rs
@@ -1,5 +1,5 @@
 use Poll;
-use stream::Stream;
+use Stream;
 
 /// A stream combinator which will change the error type of a stream from one
 /// type to another.
@@ -48,8 +48,8 @@ impl<S, F> MapErr<S, F> {
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S, F> ::sink::Sink for MapErr<S, F>
-    where S: ::sink::Sink
+impl<S, F> ::Sink for MapErr<S, F>
+    where S: ::Sink
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
diff --git a/src/stream/mod.rs b/futures-util/src/stream/mod.rs
similarity index 88%
rename from src/stream/mod.rs
rename to futures-util/src/stream/mod.rs
index e82ef0a5..b9684a18 100644
--- a/src/stream/mod.rs
+++ b/futures-util/src/stream/mod.rs
@@ -15,7 +15,7 @@
 //!
 //! [online]: https://tokio.rs/docs/getting-started/streams-and-sinks/
 
-use {IntoFuture, Poll};
+use {IntoFuture, Stream, Sink};
 
 mod iter_ok;
 pub use self::iter_ok::{iter_ok, IterOk};
@@ -83,7 +83,6 @@ pub use self::then::Then;
 pub use self::unfold::{Unfold, unfold};
 pub use self::zip::Zip;
 pub use self::forward::Forward;
-use sink::{Sink};
 
 if_std! {
     use std;
@@ -106,94 +105,13 @@ if_std! {
     pub use self::split::{SplitStream, SplitSink, ReuniteError};
     pub use self::futures_unordered::FuturesUnordered;
     pub use self::futures_ordered::{futures_ordered, FuturesOrdered};
-
-    impl<S: ?Sized + Stream> Stream for ::std::boxed::Box<S> {
-        type Item = S::Item;
-        type Error = S::Error;
-
-        fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-            (**self).poll()
-        }
-    }
 }
 
-/// A stream of values, not all of which may have been produced yet.
-///
-/// `Stream` is a trait to represent any source of sequential events or items
-/// which acts like an iterator but long periods of time may pass between
-/// items. Like `Future` the methods of `Stream` never block and it is thus
-/// suitable for programming in an asynchronous fashion. This trait is very
-/// similar to the `Iterator` trait in the standard library where `Some` is
-/// used to signal elements of the stream and `None` is used to indicate that
-/// the stream is finished.
-///
-/// Like futures a stream has basic combinators to transform the stream, perform
-/// more work on each item, etc.
-///
-/// You can find more information/tutorials about streams [online at
-/// https://tokio.rs][online]
-///
-/// [online]: https://tokio.rs/docs/getting-started/streams-and-sinks/
-///
-/// # Streams as Futures
-///
-/// Any instance of `Stream` can also be viewed as a `Future` where the resolved
-/// value is the next item in the stream along with the rest of the stream. The
-/// `into_future` adaptor can be used here to convert any stream into a future
-/// for use with other future methods like `join` and `select`.
-///
-/// # Errors
-///
-/// Streams, like futures, can also model errors in their computation. All
-/// streams have an associated `Error` type like with futures. Currently as of
-/// the 0.1 release of this library an error on a stream **does not terminate
-/// the stream**. That is, after one error is received, another error may be
-/// received from the same stream (it's valid to keep polling).
-///
-/// This property of streams, however, is [being considered] for change in 0.2
-/// where an error on a stream is similar to `None`, it terminates the stream
-/// entirely. If one of these use cases suits you perfectly and not the other,
-/// please feel welcome to comment on [the issue][being considered]!
-///
-/// [being considered]: https://github.com/alexcrichton/futures-rs/issues/206
-pub trait Stream {
-    /// The type of item this stream will yield on success.
-    type Item;
-
-    /// The type of error this stream may generate.
-    type Error;
-
-    /// Attempt to pull out the next value of this stream, returning `None` if
-    /// the stream is finished.
-    ///
-    /// This method, like `Future::poll`, is the sole method of pulling out a
-    /// value from a stream. This method must also be run within the context of
-    /// a task typically and implementors of this trait must ensure that
-    /// implementations of this method do not block, as it may cause consumers
-    /// to behave badly.
-    ///
-    /// # Return value
-    ///
-    /// If `NotReady` is returned then this stream's next value is not ready
-    /// yet and implementations will ensure that the current task will be
-    /// notified when the next value may be ready. If `Some` is returned then
-    /// the returned value represents the next value on the stream. `Err`
-    /// indicates an error happened, while `Ok` indicates whether there was a
-    /// new item on the stream or whether the stream has terminated.
-    ///
-    /// # Panics
-    ///
-    /// Once a stream is finished, that is `Ready(None)` has been returned,
-    /// further calls to `poll` may result in a panic or other "bad behavior".
-    /// If this is difficult to guard against then the `fuse` adapter can be
-    /// used to ensure that `poll` always has well-defined semantics.
-    // TODO: more here
-    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error>;
-
-    // TODO: should there also be a method like `poll` but doesn't return an
-    //       item? basically just says "please make more progress internally"
-    //       seems crucial for buffering to actually make any sense.
+impl<T: ?Sized> StreamExt for T where T: Stream {}
 
+/// An extension trait for `Stream`s that provides a variety of convenient
+/// combinator functions.
+pub trait StreamExt: Stream {
     /// Creates an iterator which blocks the current thread until each item of
     /// this stream is resolved.
     ///
@@ -973,7 +891,7 @@ pub trait Stream {
     /// library is activated, and it is activated by default.
     #[cfg(feature = "std")]
     fn split(self) -> (SplitSink<Self>, SplitStream<Self>)
-        where Self: super::sink::Sink + Sized
+        where Self: super::Sink + Sized
     {
         split::split(self)
     }
@@ -1003,15 +921,6 @@ pub trait Stream {
     }
 }
 
-impl<'a, S: ?Sized + Stream> Stream for &'a mut S {
-    type Item = S::Item;
-    type Error = S::Error;
-
-    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-        (**self).poll()
-    }
-}
-
 /// Converts a list of futures into a `Stream` of results from the futures.
 ///
 /// This function will take an list of futures (e.g. a vector, an iterator,
diff --git a/src/stream/once.rs b/futures-util/src/stream/once.rs
similarity index 97%
rename from src/stream/once.rs
rename to futures-util/src/stream/once.rs
index 24fb327b..a13793a7 100644
--- a/src/stream/once.rs
+++ b/futures-util/src/stream/once.rs
@@ -1,5 +1,5 @@
 use {Poll, Async};
-use stream::Stream;
+use Stream;
 
 /// A stream which emits single element and then EOF.
 ///
diff --git a/src/stream/or_else.rs b/futures-util/src/stream/or_else.rs
similarity index 89%
rename from src/stream/or_else.rs
rename to futures-util/src/stream/or_else.rs
index 2d15fa2b..190f745f 100644
--- a/src/stream/or_else.rs
+++ b/futures-util/src/stream/or_else.rs
@@ -1,5 +1,5 @@
 use {IntoFuture, Future, Poll, Async};
-use stream::Stream;
+use Stream;
 
 /// A stream combinator which chains a computation onto errors produced by a
 /// stream.
@@ -28,8 +28,8 @@ pub fn new<S, F, U>(s: S, f: F) -> OrElse<S, F, U>
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S, F, U> ::sink::Sink for OrElse<S, F, U>
-    where S: ::sink::Sink, U: IntoFuture
+impl<S, F, U> ::Sink for OrElse<S, F, U>
+    where S: ::Sink, U: IntoFuture
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
@@ -59,7 +59,7 @@ impl<S, F, U> Stream for OrElse<S, F, U>
         if self.future.is_none() {
             let item = match self.stream.poll() {
                 Ok(Async::Ready(e)) => return Ok(Async::Ready(e)),
-                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Ok(Async::Pending) => return Ok(Async::Pending),
                 Err(e) => e,
             };
             self.future = Some((self.f)(item).into_future());
@@ -74,7 +74,7 @@ impl<S, F, U> Stream for OrElse<S, F, U>
                 self.future = None;
                 Err(e)
             }
-            Ok(Async::NotReady) => Ok(Async::NotReady)
+            Ok(Async::Pending) => Ok(Async::Pending)
         }
     }
 }
diff --git a/src/stream/peek.rs b/futures-util/src/stream/peek.rs
similarity index 94%
rename from src/stream/peek.rs
rename to futures-util/src/stream/peek.rs
index 96e65766..3a98d568 100644
--- a/src/stream/peek.rs
+++ b/futures-util/src/stream/peek.rs
@@ -1,5 +1,5 @@
-use {Async, Poll};
-use stream::{Stream, Fuse};
+use {Async, Poll, Stream};
+use stream::{StreamExt, Fuse};
 
 /// A `Stream` that implements a `peek` method.
 ///
@@ -22,8 +22,8 @@ pub fn new<S: Stream>(stream: S) -> Peekable<S> {
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S> ::sink::Sink for Peekable<S>
-    where S: ::sink::Sink + Stream
+impl<S> ::Sink for Peekable<S>
+    where S: ::Sink + Stream
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
diff --git a/src/stream/poll_fn.rs b/futures-util/src/stream/poll_fn.rs
similarity index 100%
rename from src/stream/poll_fn.rs
rename to futures-util/src/stream/poll_fn.rs
diff --git a/src/stream/repeat.rs b/futures-util/src/stream/repeat.rs
similarity index 98%
rename from src/stream/repeat.rs
rename to futures-util/src/stream/repeat.rs
index e3cb5ff4..956b978a 100644
--- a/src/stream/repeat.rs
+++ b/futures-util/src/stream/repeat.rs
@@ -1,7 +1,7 @@
 use core::marker;
 
 
-use stream::Stream;
+use Stream;
 
 use {Async, Poll};
 
diff --git a/src/stream/select.rs b/futures-util/src/stream/select.rs
similarity index 91%
rename from src/stream/select.rs
rename to futures-util/src/stream/select.rs
index ae6b66cf..5b312b05 100644
--- a/src/stream/select.rs
+++ b/futures-util/src/stream/select.rs
@@ -1,5 +1,5 @@
-use {Poll, Async};
-use stream::{Stream, Fuse};
+use {Async, Poll, Stream};
+use stream::{StreamExt, Fuse};
 
 /// An adapter for merging the output of two streams.
 ///
@@ -45,7 +45,7 @@ impl<S1, S2> Stream for Select<S1, S2>
         let a_done = match a.poll()? {
             Async::Ready(Some(item)) => return Ok(Some(item).into()),
             Async::Ready(None) => true,
-            Async::NotReady => false,
+            Async::Pending => false,
         };
 
         match b.poll()? {
@@ -58,7 +58,7 @@ impl<S1, S2> Stream for Select<S1, S2>
                 Ok(Some(item).into())
             }
             Async::Ready(None) if a_done => Ok(None.into()),
-            Async::Ready(None) | Async::NotReady => Ok(Async::NotReady),
+            Async::Ready(None) | Async::Pending => Ok(Async::Pending),
         }
     }
 }
diff --git a/src/stream/skip.rs b/futures-util/src/stream/skip.rs
similarity index 96%
rename from src/stream/skip.rs
rename to futures-util/src/stream/skip.rs
index a1d7b497..b3c493b8 100644
--- a/src/stream/skip.rs
+++ b/futures-util/src/stream/skip.rs
@@ -1,5 +1,5 @@
 use {Poll, Async};
-use stream::Stream;
+use Stream;
 
 /// A stream combinator which skips a number of elements before continuing.
 ///
@@ -46,8 +46,8 @@ impl<S> Skip<S> {
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S> ::sink::Sink for Skip<S>
-    where S: ::sink::Sink
+impl<S> ::Sink for Skip<S>
+    where S: ::Sink
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
diff --git a/src/stream/skip_while.rs b/futures-util/src/stream/skip_while.rs
similarity index 94%
rename from src/stream/skip_while.rs
rename to futures-util/src/stream/skip_while.rs
index 6a9ba202..95ef36b3 100644
--- a/src/stream/skip_while.rs
+++ b/futures-util/src/stream/skip_while.rs
@@ -1,5 +1,5 @@
 use {Async, Poll, IntoFuture, Future};
-use stream::Stream;
+use Stream;
 
 /// A stream combinator which skips elements of a stream while a predicate
 /// holds.
@@ -53,8 +53,8 @@ impl<S, P, R> SkipWhile<S, P, R> where S: Stream, R: IntoFuture {
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S, P, R> ::sink::Sink for SkipWhile<S, P, R>
-    where S: ::sink::Sink + Stream, R: IntoFuture
+impl<S, P, R> ::Sink for SkipWhile<S, P, R>
+    where S: ::Sink + Stream, R: IntoFuture
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
@@ -101,7 +101,7 @@ impl<S, P, R> Stream for SkipWhile<S, P, R>
                     self.done_skipping = true;
                     return Ok(Async::Ready(Some(item)))
                 }
-                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Ok(Async::Pending) => return Ok(Async::Pending),
                 Err(e) => {
                     self.pending = None;
                     return Err(e)
diff --git a/src/stream/split.rs b/futures-util/src/stream/split.rs
similarity index 92%
rename from src/stream/split.rs
rename to futures-util/src/stream/split.rs
index ddaa5299..dc808456 100644
--- a/src/stream/split.rs
+++ b/futures-util/src/stream/split.rs
@@ -3,7 +3,7 @@ use std::error::Error;
 use std::fmt;
 
 use {StartSend, Sink, Stream, Poll, Async, AsyncSink};
-use sync::BiLock;
+use lock::BiLock;
 
 /// A `Stream` part of the split pair
 #[derive(Debug)]
@@ -25,7 +25,7 @@ impl<S: Stream> Stream for SplitStream<S> {
     fn poll(&mut self) -> Poll<Option<S::Item>, S::Error> {
         match self.0.poll_lock() {
             Async::Ready(mut inner) => inner.poll(),
-            Async::NotReady => Ok(Async::NotReady),
+            Async::Pending => Ok(Async::Pending),
         }
     }
 }
@@ -54,21 +54,21 @@ impl<S: Sink> Sink for SplitSink<S> {
     {
         match self.0.poll_lock() {
             Async::Ready(mut inner) => inner.start_send(item),
-            Async::NotReady => Ok(AsyncSink::NotReady(item)),
+            Async::Pending => Ok(AsyncSink::Pending(item)),
         }
     }
 
     fn poll_complete(&mut self) -> Poll<(), S::SinkError> {
         match self.0.poll_lock() {
             Async::Ready(mut inner) => inner.poll_complete(),
-            Async::NotReady => Ok(Async::NotReady),
+            Async::Pending => Ok(Async::Pending),
         }
     }
 
     fn close(&mut self) -> Poll<(), S::SinkError> {
         match self.0.poll_lock() {
             Async::Ready(mut inner) => inner.close(),
-            Async::NotReady => Ok(Async::NotReady),
+            Async::Pending => Ok(Async::Pending),
         }
     }
 }
diff --git a/src/stream/take.rs b/futures-util/src/stream/take.rs
similarity index 96%
rename from src/stream/take.rs
rename to futures-util/src/stream/take.rs
index 0ca68496..bb46a495 100644
--- a/src/stream/take.rs
+++ b/futures-util/src/stream/take.rs
@@ -1,5 +1,5 @@
 use {Async, Poll};
-use stream::Stream;
+use Stream;
 
 /// A stream combinator which returns a maximum number of elements.
 ///
@@ -46,8 +46,8 @@ impl<S> Take<S> {
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S> ::sink::Sink for Take<S>
-    where S: ::sink::Sink + Stream
+impl<S> ::Sink for Take<S>
+    where S: ::Sink + Stream
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
diff --git a/src/stream/take_while.rs b/futures-util/src/stream/take_while.rs
similarity index 94%
rename from src/stream/take_while.rs
rename to futures-util/src/stream/take_while.rs
index 732ae855..cffd2e8c 100644
--- a/src/stream/take_while.rs
+++ b/futures-util/src/stream/take_while.rs
@@ -1,5 +1,5 @@
 use {Async, Poll, IntoFuture, Future};
-use stream::Stream;
+use Stream;
 
 /// A stream combinator which takes elements from a stream while a predicate
 /// holds.
@@ -53,8 +53,8 @@ impl<S, P, R> TakeWhile<S, P, R> where S: Stream, R: IntoFuture {
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S, P, R> ::sink::Sink for TakeWhile<S, P, R>
-    where S: ::sink::Sink + Stream, R: IntoFuture
+impl<S, P, R> ::Sink for TakeWhile<S, P, R>
+    where S: ::Sink + Stream, R: IntoFuture
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
@@ -103,7 +103,7 @@ impl<S, P, R> Stream for TakeWhile<S, P, R>
                 self.done_taking = true;
                 Ok(Async::Ready(None))
             }
-            Ok(Async::NotReady) => Ok(Async::NotReady),
+            Ok(Async::Pending) => Ok(Async::Pending),
             Err(e) => {
                 self.pending = None;
                 Err(e)
diff --git a/src/stream/then.rs b/futures-util/src/stream/then.rs
similarity index 89%
rename from src/stream/then.rs
rename to futures-util/src/stream/then.rs
index cab338e9..dd25ab72 100644
--- a/src/stream/then.rs
+++ b/futures-util/src/stream/then.rs
@@ -1,5 +1,5 @@
 use {Async, IntoFuture, Future, Poll};
-use stream::Stream;
+use Stream;
 
 /// A stream combinator which chains a computation onto each item produced by a
 /// stream.
@@ -28,8 +28,8 @@ pub fn new<S, F, U>(s: S, f: F) -> Then<S, F, U>
 }
 
 // Forwarding impl of Sink from the underlying stream
-impl<S, F, U> ::sink::Sink for Then<S, F, U>
-    where S: ::sink::Sink, U: IntoFuture,
+impl<S, F, U> ::Sink for Then<S, F, U>
+    where S: ::Sink, U: IntoFuture,
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
@@ -58,7 +58,7 @@ impl<S, F, U> Stream for Then<S, F, U>
     fn poll(&mut self) -> Poll<Option<U::Item>, U::Error> {
         if self.future.is_none() {
             let item = match self.stream.poll() {
-                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Ok(Async::Pending) => return Ok(Async::Pending),
                 Ok(Async::Ready(None)) => return Ok(Async::Ready(None)),
                 Ok(Async::Ready(Some(e))) => Ok(e),
                 Err(e) => Err(e),
@@ -75,7 +75,7 @@ impl<S, F, U> Stream for Then<S, F, U>
                 self.future = None;
                 Err(e)
             }
-            Ok(Async::NotReady) => Ok(Async::NotReady)
+            Ok(Async::Pending) => Ok(Async::Pending)
         }
     }
 }
diff --git a/src/stream/unfold.rs b/futures-util/src/stream/unfold.rs
similarity index 96%
rename from src/stream/unfold.rs
rename to futures-util/src/stream/unfold.rs
index ac427b8c..c10a6532 100644
--- a/src/stream/unfold.rs
+++ b/futures-util/src/stream/unfold.rs
@@ -1,7 +1,7 @@
 use core::mem;
 
 use {Future, IntoFuture, Async, Poll};
-use stream::Stream;
+use Stream;
 
 /// Creates a `Stream` from a seed and a closure returning a `Future`.
 ///
@@ -90,9 +90,9 @@ impl <T, F, Fut, It> Stream for Unfold<T, F, Fut>
                             self.state = State::Ready(next_state);
                             return Ok(Async::Ready(Some(item)));
                         }
-                        Async::NotReady => {
+                        Async::Pending => {
                             self.state = State::Processing(fut);
-                            return Ok(Async::NotReady);
+                            return Ok(Async::Pending);
                         }
                     }
                 }
diff --git a/src/stream/wait.rs b/futures-util/src/stream/wait.rs
similarity index 93%
rename from src/stream/wait.rs
rename to futures-util/src/stream/wait.rs
index 80acb6c2..754bff2c 100644
--- a/src/stream/wait.rs
+++ b/futures-util/src/stream/wait.rs
@@ -1,5 +1,5 @@
-use stream::Stream;
-use executor;
+use Stream;
+use task;
 
 /// A stream combinator which converts an asynchronous stream to a **blocking
 /// iterator**.
@@ -10,7 +10,7 @@ use executor;
 #[must_use = "iterators do nothing unless advanced"]
 #[derive(Debug)]
 pub struct Wait<S> {
-    stream: executor::Spawn<S>,
+    stream: task::Spawn<S>,
 }
 
 impl<S> Wait<S> {
@@ -40,7 +40,7 @@ impl<S> Wait<S> {
 
 pub fn new<S: Stream>(s: S) -> Wait<S> {
     Wait {
-        stream: executor::spawn(s),
+        stream: task::spawn(s),
     }
 }
 
diff --git a/src/stream/zip.rs b/futures-util/src/stream/zip.rs
similarity index 88%
rename from src/stream/zip.rs
rename to futures-util/src/stream/zip.rs
index 17e3c69f..da526858 100644
--- a/src/stream/zip.rs
+++ b/futures-util/src/stream/zip.rs
@@ -1,5 +1,5 @@
-use {Async, Poll};
-use stream::{Stream, Fuse};
+use {Async, Poll, Stream};
+use stream::{StreamExt, Fuse};
 
 /// An adapter for merging the output of two streams.
 ///
@@ -36,13 +36,13 @@ impl<S1, S2> Stream for Zip<S1, S2>
         if self.queued1.is_none() {
             match self.stream1.poll()? {
                 Async::Ready(Some(item1)) => self.queued1 = Some(item1),
-                Async::Ready(None) | Async::NotReady => {}
+                Async::Ready(None) | Async::Pending => {}
             }
         }
         if self.queued2.is_none() {
             match self.stream2.poll()? {
                 Async::Ready(Some(item2)) => self.queued2 = Some(item2),
-                Async::Ready(None) | Async::NotReady => {}
+                Async::Ready(None) | Async::Pending => {}
             }
         }
 
@@ -53,7 +53,7 @@ impl<S1, S2> Stream for Zip<S1, S2>
         } else if self.stream1.is_done() || self.stream2.is_done() {
             Ok(Async::Ready(None))
         } else {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 }
diff --git a/futures/Cargo.toml b/futures/Cargo.toml
new file mode 100644
index 00000000..b3f77338
--- /dev/null
+++ b/futures/Cargo.toml
@@ -0,0 +1,30 @@
+[package]
+name = "futures"
+version = "0.2.0"
+authors = ["Alex Crichton <alex@alexcrichton.com>"]
+license = "MIT/Apache-2.0"
+readme = "README.md"
+keywords = ["futures", "async", "future"]
+repository = "https://github.com/alexcrichton/futures-rs"
+homepage = "https://github.com/alexcrichton/futures-rs"
+documentation = "https://docs.rs/futures"
+description = """
+An implementation of futures and streams featuring zero allocations,
+composability, and iterator-like interfaces.
+"""
+categories = ["asynchronous"]
+
+[badges]
+travis-ci = { repository = "alexcrichton/futures-rs" }
+appveyor = { repository = "alexcrichton/futures-rs" }
+
+[dependencies]
+futures-core = { path = "../futures-core", version = "0.1.0", default-features = false }
+futures-channel = { path = "../futures-channel", version = "0.1.0", default-features = false }
+futures-executor = { path = "../futures-executor", version = "0.1.0", default-features = false }
+futures-sink = { path = "../futures-sink", version = "0.1.0", default-features = false }
+futures-util = { path = "../futures-util", version = "0.1.0", default-features = false }
+
+[features]
+std = ["futures-core/std", "futures-executor/std", "futures-sink/std", "futures-util/std"]
+default = ["std"]
diff --git a/src/lib.rs b/futures/src/lib.rs
similarity index 63%
rename from src/lib.rs
rename to futures/src/lib.rs
index 01021940..6823df82 100644
--- a/src/lib.rs
+++ b/futures/src/lib.rs
@@ -155,61 +155,137 @@
 //!
 //! [README]: https://github.com/alexcrichton/futures-rs#futures-rs
 
-#![no_std]
-#![deny(missing_docs, missing_debug_implementations)]
-#![doc(html_root_url = "https://docs.rs/futures/0.1")]
+extern crate futures_core;
+extern crate futures_channel;
+extern crate futures_executor;
+extern crate futures_sink;
+extern crate futures_util;
+
+pub use futures_core::future::{Future, IntoFuture};
+pub use futures_util::future::FutureExt;
+pub use futures_core::stream::Stream;
+pub use futures_util::stream::StreamExt;
+pub use futures_sink::Sink;
+pub use futures_util::sink::SinkExt;
+
+// Macros redefined here because macro re-exports are unstable.
+
+/// A macro for extracting the successful type of a `Poll<T, E>`.
+///
+/// This macro bakes propagation of both errors and `Pending` signals by
+/// returning early.
+#[macro_export]
+macro_rules! try_ready {
+    ($e:expr) => (match $e {
+        Ok($crate::Async::Ready(t)) => t,
+        Ok($crate::Async::Pending) => return Ok($crate::Async::Pending),
+        Err(e) => return Err(From::from(e)),
+    })
+}
+
+// TODO: task_local macro
 
-#[macro_use]
 #[cfg(feature = "std")]
-extern crate std;
+pub mod channel {
+    //! Channels
+    //!
+    //! This module contains channels for asynchronous
+    //! communication.
+    pub use futures_channel::*;
+}
+
+pub mod executor {
+    //! Executors
+    //!
+    //! This module contains the `Executor` trait, which allows futures to be
+    //! spawned and executed asynchronously.
 
-macro_rules! if_std {
-    ($($i:item)*) => ($(
-        #[cfg(feature = "std")]
-        $i
-    )*)
+    pub use futures_executor::*;
 }
 
-#[macro_use]
-mod poll;
-pub use poll::{Poll, Async, AsyncSink, StartSend};
+pub mod future {
+    //! Futures
+    //!
+    //! This module contains the `Future` trait and adapters for this trait.
 
-pub mod future;
-pub use future::{Future, IntoFuture};
+    pub use futures_core::future::*;
+    pub use futures_util::future::*;
+}
 
-pub mod stream;
-pub use stream::Stream;
+pub mod prelude {
+	//! A "prelude" for crates using the `futures` crate.
+	//!
+	//! This prelude is similar to the standard library's prelude in that you'll
+	//! almost always want to import its entire contents, but unlike the standard
+	//! library's prelude you'll have to do so manually. An example of using this is:
+	//!
+	//! ```
+	//! use futures::prelude::*;
+	//! ```
+	//!
+	//! We may add items to this over time as they become ubiquitous as well, but
+	//! otherwise this should help cut down on futures-related imports when you're
+	//! working with the `futures` crate!
 
-pub mod sink;
-pub use sink::Sink;
+    pub use futures_core::{
+        Future,
+        IntoFuture,
+        Stream,
+        Async,
+        Poll,
+    };
 
-#[cfg(feature = "std")]
-mod lock;
-mod task_impl;
+    pub use futures_sink::{
+        Sink,
+        AsyncSink,
+        StartSend,
+    };
 
-mod resultstream;
+    pub use futures_util::{
+		FutureExt,
+		StreamExt,
+        SinkExt,
+    };
+}
 
-pub mod task;
-pub mod executor;
-#[cfg(feature = "std")]
-pub mod sync;
+pub mod sink {
+	//! Asynchronous sinks
+	//!
+	//! This module contains the `Sink` trait, along with a number of adapter types
+	//! for it. An overview is available in the documentation for the trait itself.
+	//!
+	//! You can find more information/tutorials about streams [online at
+	//! https://tokio.rs][online]
+	//!
+	//! [online]: https://tokio.rs/docs/getting-started/streams-and-sinks/
 
-/// A "prelude" for crates using the `futures` crate.
-///
-/// This prelude is similar to the standard library's prelude in that you'll
-/// almost always want to import its entire contents, but unlike the standard
-/// library's prelude you'll have to do so manually. An example of using this is:
-///
-/// ```
-/// use futures::prelude::*;
-/// ```
-///
-/// We may add items to this over time as they become ubiquitous as well, but
-/// otherwise this should help cut down on futures-related imports when you're
-/// working with the `futures` crate!
-pub mod prelude {
-    #[doc(no_inline)]
-    pub use {Future, Stream, Sink, Async, AsyncSink, Poll, StartSend};
-    #[doc(no_inline)]
-    pub use IntoFuture;
+    pub use futures_sink::*;
+    pub use futures_util::sink::*;
+}
+
+pub mod stream {
+	//! Asynchronous streams
+	//!
+	//! This module contains the `Stream` trait and a number of adaptors for this
+	//! trait. This trait is very similar to the `Iterator` trait in the standard
+	//! library except that it expresses the concept of blocking as well. A stream
+	//! here is a sequential sequence of values which may take some amount of time
+	//! in between to produce.
+	//!
+	//! A stream may request that it is blocked between values while the next value
+	//! is calculated, and provides a way to get notified once the next value is
+	//! ready as well.
+	//!
+	//! You can find more information/tutorials about streams [online at
+	//! https://tokio.rs][online]
+	//!
+	//! [online]: https://tokio.rs/docs/getting-started/streams-and-sinks/
+    
+    pub use futures_core::stream::*;
+    pub use futures_util::stream::*;
+}
+
+pub mod task {
+    //! Task executon
+    pub use futures_core::task::*;
 }
diff --git a/src/executor.rs b/src/executor.rs
deleted file mode 100644
index 02a3bd29..00000000
--- a/src/executor.rs
+++ /dev/null
@@ -1,16 +0,0 @@
-//! Executors
-//!
-//! This module contains tools for managing the raw execution of futures,
-//! which is needed when building *executors* (places where futures can run).
-//!
-//! More information about executors can be [found online at tokio.rs][online].
-//!
-//! [online]: https://tokio.rs/docs/going-deeper-futures/tasks/
-
-#[allow(deprecated)]
-#[cfg(feature = "std")]
-pub use task_impl::{Executor, Run};
-
-pub use task_impl::{Spawn, spawn, Notify, with_notify};
-
-pub use task_impl::{UnsafeNotify, NotifyHandle};
diff --git a/src/future/option.rs b/src/future/option.rs
deleted file mode 100644
index 1b204d37..00000000
--- a/src/future/option.rs
+++ /dev/null
@@ -1,15 +0,0 @@
-//! Definition of the `Option` (optional step) combinator
-
-use {Future, Poll, Async};
-
-impl<F, T, E> Future for Option<F> where F: Future<Item=T, Error=E> {
-    type Item = Option<T>;
-    type Error = E;
-
-    fn poll(&mut self) -> Poll<Option<T>, E> {
-        match *self {
-            None => Ok(Async::Ready(None)),
-            Some(ref mut x) => x.poll().map(|x| x.map(Some)),
-        }
-    }
-}
diff --git a/src/poll.rs b/src/poll.rs
deleted file mode 100644
index c568e726..00000000
--- a/src/poll.rs
+++ /dev/null
@@ -1,105 +0,0 @@
-/// A macro for extracting the successful type of a `Poll<T, E>`.
-///
-/// This macro bakes propagation of both errors and `NotReady` signals by
-/// returning early.
-#[macro_export]
-macro_rules! try_ready {
-    ($e:expr) => (match $e {
-        Ok($crate::Async::Ready(t)) => t,
-        Ok($crate::Async::NotReady) => return Ok($crate::Async::NotReady),
-        Err(e) => return Err(From::from(e)),
-    })
-}
-
-/// Return type of the `Future::poll` method, indicates whether a future's value
-/// is ready or not.
-///
-/// * `Ok(Async::Ready(t))` means that a future has successfully resolved
-/// * `Ok(Async::NotReady)` means that a future is not ready to complete yet
-/// * `Err(e)` means that a future has completed with the given failure
-pub type Poll<T, E> = Result<Async<T>, E>;
-
-/// Return type of future, indicating whether a value is ready or not.
-#[derive(Copy, Clone, Debug, PartialEq)]
-pub enum Async<T> {
-    /// Represents that a value is immediately ready.
-    Ready(T),
-
-    /// Represents that a value is not ready yet, but may be so later.
-    NotReady,
-}
-
-impl<T> Async<T> {
-    /// Change the success value of this `Async` with the closure provided
-    pub fn map<F, U>(self, f: F) -> Async<U>
-        where F: FnOnce(T) -> U
-    {
-        match self {
-            Async::Ready(t) => Async::Ready(f(t)),
-            Async::NotReady => Async::NotReady,
-        }
-    }
-
-    /// Returns whether this is `Async::Ready`
-    pub fn is_ready(&self) -> bool {
-        match *self {
-            Async::Ready(_) => true,
-            Async::NotReady => false,
-        }
-    }
-
-    /// Returns whether this is `Async::NotReady`
-    pub fn is_not_ready(&self) -> bool {
-        !self.is_ready()
-    }
-}
-
-impl<T> From<T> for Async<T> {
-    fn from(t: T) -> Async<T> {
-        Async::Ready(t)
-    }
-}
-
-/// The result of an asynchronous attempt to send a value to a sink.
-#[derive(Copy, Clone, Debug, PartialEq)]
-pub enum AsyncSink<T> {
-    /// The `start_send` attempt succeeded, so the sending process has
-    /// *started*; you must use `Sink::poll_complete` to drive the send
-    /// to completion.
-    Ready,
-
-    /// The `start_send` attempt failed due to the sink being full. The value
-    /// being sent is returned, and the current `Task` will be automatically
-    /// notified again once the sink has room.
-    NotReady(T),
-}
-
-impl<T> AsyncSink<T> {
-    /// Change the NotReady value of this `AsyncSink` with the closure provided
-    pub fn map<F, U>(self, f: F) -> AsyncSink<U>
-        where F: FnOnce(T) -> U,
-    {
-        match self {
-            AsyncSink::Ready => AsyncSink::Ready,
-            AsyncSink::NotReady(t) => AsyncSink::NotReady(f(t)),
-        }
-    }
-
-    /// Returns whether this is `AsyncSink::Ready`
-    pub fn is_ready(&self) -> bool {
-        match *self {
-            AsyncSink::Ready => true,
-            AsyncSink::NotReady(_) => false,
-        }
-    }
-
-    /// Returns whether this is `AsyncSink::NotReady`
-    pub fn is_not_ready(&self) -> bool {
-        !self.is_ready()
-    }
-}
-
-
-/// Return type of the `Sink::start_send` method, indicating the outcome of a
-/// send attempt. See `AsyncSink` for more details.
-pub type StartSend<T, E> = Result<AsyncSink<T>, E>;
diff --git a/src/resultstream.rs b/src/resultstream.rs
deleted file mode 100644
index 23a99819..00000000
--- a/src/resultstream.rs
+++ /dev/null
@@ -1,46 +0,0 @@
-// This should really be in the stream module,
-// but `pub(crate)` isn't available until Rust 1.18,
-// and pre-1.18 there isn't a really good way to have a sub-module
-// available to the crate, but not without it.
-use core::marker::PhantomData;
-
-use {Poll, Async};
-use stream::Stream;
-
-
-/// A stream combinator used to convert a `Stream<Item=T,Error=E>`
-/// to a `Stream<Item=Result<T,E>>`.
-///
-/// A poll on this stream will never return an `Err`. As such the
-/// actual error type is parameterized, so it can match whatever error
-/// type is needed.
-///
-/// This structure is produced by the `Stream::results` method.
-#[derive(Debug)]
-#[must_use = "streams do nothing unless polled"]
-pub struct Results<S: Stream, E> {
-    inner: S,
-    phantom: PhantomData<E>
-}
-
-pub fn new<S, E>(s: S) -> Results<S, E> where S: Stream {
-    Results {
-        inner: s,
-        phantom: PhantomData
-    }
-}
-
-impl<S: Stream, E> Stream for Results<S, E> {
-    type Item = Result<S::Item, S::Error>;
-    type Error = E;
-
-    fn poll(&mut self) -> Poll<Option<Result<S::Item, S::Error>>, E> {
-        match self.inner.poll() {
-            Ok(Async::Ready(Some(item))) => Ok(Async::Ready(Some(Ok(item)))),
-            Err(e) => Ok(Async::Ready(Some(Err(e)))),
-            Ok(Async::Ready(None)) => Ok(Async::Ready(None)),
-            Ok(Async::NotReady) => Ok(Async::NotReady)
-        }
-    }
-}
-
diff --git a/src/sync/mod.rs b/src/sync/mod.rs
deleted file mode 100644
index 0a46e9af..00000000
--- a/src/sync/mod.rs
+++ /dev/null
@@ -1,17 +0,0 @@
-//! Future-aware synchronization
-//!
-//! This module, which is modeled after `std::sync`, contains user-space
-//! synchronization tools that work with futures, streams and sinks. In
-//! particular, these synchronizers do *not* block physical OS threads, but
-//! instead work at the task level.
-//!
-//! More information and examples of how to use these synchronization primitives
-//! can be found [online at tokio.rs].
-//!
-//! [online at tokio.rs]: https://tokio.rs/docs/going-deeper-futures/synchronization/
-
-pub mod oneshot;
-pub mod mpsc;
-mod bilock;
-
-pub use self::bilock::{BiLock, BiLockGuard, BiLockAcquire, BiLockAcquired};


Test patch:
diff --git a/src/task_impl/std/unpark_mutex.rs b/futures-cpupool/src/unpark_mutex.rs
similarity index 100%
rename from src/task_impl/std/unpark_mutex.rs
rename to futures-cpupool/src/unpark_mutex.rs
diff --git a/futures-cpupool/tests/smoke.rs b/futures-cpupool/tests/smoke.rs
index 1b267f2f..ed8be10c 100644
--- a/futures-cpupool/tests/smoke.rs
+++ b/futures-cpupool/tests/smoke.rs
@@ -5,7 +5,7 @@ use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
 use std::thread;
 use std::time::Duration;
 
-use futures::future::Future;
+use futures::future::{Future, FutureExt};
 use futures_cpupool::{CpuPool, Builder};
 
 fn done<T: Send + 'static>(t: T) -> Box<Future<Item = T, Error = ()> + Send> {
diff --git a/tests/bilock.rs b/tests/bilock.rs
index 6e0c3442..d4c4f9f2 100644
--- a/tests/bilock.rs
+++ b/tests/bilock.rs
@@ -19,7 +19,7 @@ fn smoke() {
         {
             let mut lock = match a.poll_lock() {
                 Async::Ready(l) => l,
-                Async::NotReady => panic!("poll not ready"),
+                Async::Pending => panic!("poll not ready"),
             };
             assert_eq!(*lock, 1);
             *lock = 2;
@@ -34,7 +34,7 @@ fn smoke() {
         {
             let lock = match b.poll_lock() {
                 Async::Ready(l) => l,
-                Async::NotReady => panic!("poll not ready"),
+                Async::Pending => panic!("poll not ready"),
             };
             assert_eq!(*lock, 2);
         }
@@ -72,11 +72,11 @@ fn concurrent() {
 
     match a.poll_lock() {
         Async::Ready(l) => assert_eq!(*l, 2 * N),
-        Async::NotReady => panic!("poll not ready"),
+        Async::Pending => panic!("poll not ready"),
     }
     match b.poll_lock() {
         Async::Ready(l) => assert_eq!(*l, 2 * N),
-        Async::NotReady => panic!("poll not ready"),
+        Async::Pending => panic!("poll not ready"),
     }
 
     assert_eq!(a.reunite(b).expect("bilock/concurrent: reunite error"), 2 * N);
@@ -99,7 +99,7 @@ fn concurrent() {
                 let a = self.a.as_ref().unwrap();
                 let mut a = match a.poll_lock() {
                     Async::Ready(l) => l,
-                    Async::NotReady => return Ok(Async::NotReady),
+                    Async::Pending => return Ok(Async::Pending),
                 };
                 self.remaining -= 1;
                 *a += 1;
diff --git a/tests/mpsc.rs b/tests/mpsc.rs
index 86e44fee..2c4b996d 100644
--- a/tests/mpsc.rs
+++ b/tests/mpsc.rs
@@ -172,7 +172,7 @@ fn spawn_kill_dead_stream() {
         type Error = ();
 
         fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 
@@ -319,7 +319,7 @@ fn stress_receiver_multi_task_bounded_hard() {
                                     Async::Ready(None) => {
                                         true
                                     }
-                                    Async::NotReady => {
+                                    Async::Pending => {
                                         *lock = Some(rx);
                                         false
                                     }
diff --git a/tests/oneshot.rs b/tests/oneshot.rs
index 45c19968..7f2bbd38 100644
--- a/tests/oneshot.rs
+++ b/tests/oneshot.rs
@@ -151,7 +151,7 @@ fn spawn_kill_dead_stream() {
         type Error = ();
 
         fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 
@@ -213,7 +213,7 @@ fn spawn_dont_kill_forgot_dead_stream() {
         type Error = ();
 
         fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 
diff --git a/tests/sink.rs b/tests/sink.rs
index 4d5355b1..3993eabb 100644
--- a/tests/sink.rs
+++ b/tests/sink.rs
@@ -94,9 +94,9 @@ impl<S: Sink> Future for StartSendFut<S> {
     fn poll(&mut self) -> Poll<S, S::SinkError> {
         match self.0.as_mut().unwrap().start_send(self.1.take().unwrap())? {
             AsyncSink::Ready => Ok(Async::Ready(self.0.take().unwrap())),
-            AsyncSink::NotReady(item) => {
+            AsyncSink::Pending(item) => {
                 self.1 = Some(item);
-                Ok(Async::NotReady)
+                Ok(Async::Pending)
             }
         }
 
@@ -205,7 +205,7 @@ impl<T> Sink for ManualFlush<T> {
             Ok(Async::Ready(()))
         } else {
             self.waiting_tasks.push(task::current());
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 
@@ -306,7 +306,7 @@ impl<T> Sink for ManualAllow<T> {
             self.data.push(item);
             Ok(AsyncSink::Ready)
         } else {
-            Ok(AsyncSink::NotReady(item))
+            Ok(AsyncSink::Pending(item))
         }
     }
 
diff --git a/tests/support/local_executor.rs b/tests/support/local_executor.rs
index 615efc1f..40fc0802 100644
--- a/tests/support/local_executor.rs
+++ b/tests/support/local_executor.rs
@@ -86,7 +86,7 @@ impl Core {
             if self.turn() {
                 match f.poll_future_notify(&self.notify, id)? {
                     Async::Ready(e) => return Ok(e),
-                    Async::NotReady => {}
+                    Async::Pending => {}
                 }
             }
         }
@@ -124,7 +124,7 @@ impl Core {
         // done then we put it back in the tasks array.
         let done = match future.poll_future_notify(&self.notify, task_id) {
             Ok(Async::Ready(())) | Err(()) => true,
-            Ok(Async::NotReady) => false,
+            Ok(Async::Pending) => false,
         };
         let mut tasks = self.tasks.borrow_mut();
         if done {
diff --git a/tests/support/mod.rs b/tests/support/mod.rs
index 29774977..8eca69d0 100644
--- a/tests/support/mod.rs
+++ b/tests/support/mod.rs
@@ -34,7 +34,7 @@ pub fn sassert_done<S: Stream>(s: &mut S) {
     match executor::spawn(s).poll_stream_notify(&notify_panic(), 0) {
         Ok(Async::Ready(None)) => {}
         Ok(Async::Ready(Some(_))) => panic!("stream had more elements"),
-        Ok(Async::NotReady) => panic!("stream wasn't ready"),
+        Ok(Async::Pending) => panic!("stream wasn't ready"),
         Err(_) => panic!("stream had an error"),
     }
 }
@@ -43,7 +43,7 @@ pub fn sassert_empty<S: Stream>(s: &mut S) {
     match executor::spawn(s).poll_stream_notify(&notify_noop(), 0) {
         Ok(Async::Ready(None)) => panic!("stream is at its end"),
         Ok(Async::Ready(Some(_))) => panic!("stream had more elements"),
-        Ok(Async::NotReady) => {}
+        Ok(Async::Pending) => {}
         Err(_) => panic!("stream had an error"),
     }
 }
@@ -54,7 +54,7 @@ pub fn sassert_next<S: Stream>(s: &mut S, item: S::Item)
     match executor::spawn(s).poll_stream_notify(&notify_panic(), 0) {
         Ok(Async::Ready(None)) => panic!("stream is at its end"),
         Ok(Async::Ready(Some(e))) => assert_eq!(e, item),
-        Ok(Async::NotReady) => panic!("stream wasn't ready"),
+        Ok(Async::Pending) => panic!("stream wasn't ready"),
         Err(_) => panic!("stream had an error"),
     }
 }
@@ -65,7 +65,7 @@ pub fn sassert_err<S: Stream>(s: &mut S, err: S::Error)
     match executor::spawn(s).poll_stream_notify(&notify_panic(), 0) {
         Ok(Async::Ready(None)) => panic!("stream is at its end"),
         Ok(Async::Ready(Some(_))) => panic!("stream had more elements"),
-        Ok(Async::NotReady) => panic!("stream wasn't ready"),
+        Ok(Async::Pending) => panic!("stream wasn't ready"),
         Err(e) => assert_eq!(e, err),
     }
 }
@@ -120,12 +120,12 @@ impl<F: Future> Future for DelayFuture<F> {
         } else {
             self.1 = true;
             task::current().notify();
-            Ok(Async::NotReady)
+            Ok(Async::Pending)
         }
     }
 }
 
-/// Introduces one `Ok(Async::NotReady)` before polling the given future
+/// Introduces one `Ok(Async::Pending)` before polling the given future
 pub fn delay_future<F>(f: F) -> DelayFuture<F::Future>
     where F: IntoFuture,
 {