Repository: alexcrichton/futures-rs
Pull request #436 (2017-03-31 21:32:45)

Task system overhaul
## Updated description

The primary change made in this PR is to restructure memory management and
notifications throughout the "task system" in the `futures` crate. It is
intended that this will have very little impact, if any, on consumers of
futures. Implementations of runtimes of futures (e.g. crates like tokio-core)
are the target for this series of changes, enabling a suite of possible
optimizations that were not previously feasible. One major use case that is 
now enabled is usage of the `task` and `executor` modules in the `no_std` ecosystem.
This means that bare-metal applications of futures should be able to use the same
task system that the std-based futures ecosystem uses.

One of the largest changes being made to support this is an update to the memory management of
objects behind `Task` handles. Previously it was required that `Arc<Unpark>`
instances were passed into the various `Spawn::poll_*` functions, but new
functions `Spawn::poll_*_notify` were added which operate with a `NotifyHandle`
instead. A `NotifyHandle` is conceptually very similar to an `Arc<Unpark>`
instance, but it works through an `unsafe trait UnsafeNotify` to manage memory
instead of requiring `Arc` is used. You can still use `Arc` safely, however, if
you'd like.

In addition to supporting more forms of memory management, the `poll_*_notify`
functions also take a new `id` parameter. This parameter is intended to be an
opaque bag-of-bits to the `futures` crate itself but runtimes can use this to
identify the future being notified. This is intended to enable situations where
the same instance of a `NotifyHandle` can be used for all futures executed by
using the `id` field to distinguish *which* future is ready when it gets
notified.

## API Additions

* A `FuturesUnordered::push` method was added and the `FuturesUnordered` type
  itself was completely rewritten to efficiently track a large number of
  futures.
* A `Task::will_notify_current` method was added with a slightly different
  implementation than `Task::is_current` but with stronger guarantees and
  documentation wording about its purpose.

## Compatibility Notes

As with all 0.1.x releases this PR is intended to be 100% backwards compatible.
All code that previously compiled should continue to do so with these changes.
As with other changes, though, there are also some updates to be aware of:

* The `task::park` function has been renamed to `task::current`.
* The `Task::unpark` function has been renamed to `Task::notify`, and in general
  terminology around "unpark" has shifted to terminology around "notify"
* The `Unpark` trait has been deprecated in favor of the `Notify` trait
  mentioned above.
* The `UnparkEvent` structure has been deprecated. It currently should perform
  the same as it used to, but it's planned that in a future 0.1.x release the
  performance will regress for crates that have not transitioned away. The
  primary primitive to replace this is the addition of a `push` function on the
  `FuturesUnordered` type. If this does not help implement your use case though,
  please let us know!
* The `Task::is_current` method is now deprecated, and you likely want to use
  `Task::will_notify_current` instead, but let us know if this doesn't suffice!


## Original description


**This PR is a work in progress**

I'm submitting this PR early to hopefully try to illustrate my thoughts and get some early feedback.

# Checklist

* [x] Land #442 
* [x] Update tokio-core
* [x] Run sscache tests using new task system
* [x] Switch `Arc<Unpark>` to `UnparkHandle` #432
* [x] Decide on #312 (leaning towards yes)
* [x] Allow executors to customize wait behavior #360 (deferring this until later)
* [x] Fix `Task::is_current`
* [x] Remove `Notify::is_current`, I don't think this is needed anymore.
* [x] Consider `GetNotifyHandle` https://github.com/alexcrichton/futures-rs/issues/129
* [x] Should `ref_inc` -> `ref_dec` be moved to `UnsafeNotify` (@alexcrichton says no).
* [x] Consider getting rid of `poll_*_notify` on `Stream` and `Sink`. Also, maybe name it `poll_notify` if it is only for `Future`.
* [x] Merge https://github.com/carllerche/futures-rs/pull/4
* [x] `u64` vs. `usize`

# Overview

The previous implementation of the task system required a number of allocations per task instance. Each task required a dedicated `Arc<Unpark>` handle which means that executors require at least two allocations per task.

Things get worse when using `with_unpark_event` as nested calls to `with_unpark_event` result in Vec allocation and cloning during each call to `task::park`.

This commit provides an overhaul to the task system to work around these problems. The `Unpark` trait is changed so that only one instance is required per executor. In order to identify which task is being unparked, `Unpark::unpark` takes an `unpark_id: u64` argument.

`with_unpark_event` is removed in favor of `UnparkContext` which satisfies a similar end goal, but requires only a single allocation per lifetime of the `UnparkContext`.

# The new `Unpark` trait

In general, tasks are driven to completion by executors and executors are able to handle large number of tasks. As such, the `Unpark` trait has been tweaked to require a single allocation per executor instead of one per task. The idea is that the executor creates one `Arc<Unpark>` handle and uses the `unpark_id: u64` to identify which task is unparked.

In the case of tokio-core, each task is stored in a slab and the the `unpark_id` is the slab index. Now, given that an Arc is no longer used, it can be that a slab slot is released and repurposed for a different task while there are still outstanding Task handles referencing the now released task.

There are two potential ways to deal with this.

a) Not care. Futures need to be able to handle spurious wake ups already. Spurious wake ups can be reduced by splitting the `u64` into 28 bits for the slab offset and use the rest of the u64 as a slot usage counter.

b) Use `Unpark::ref_inc` and `Unpark::ref_dec` to allow the executor implementation to handle its own reference counting.

Option b) would allow an executor implementation to store a pointer as the `unpark_id` and the `ref_inc` and `ref_dec` allow for atomic reference counting. This could be used in cases where using a slab is not an option.

# `UnparkContext`

This is quite similar in spirit to `with_unpark_event` except it requires significantly less allocations when used in a nested situation.

It does have some different behavior. Given the following:

```rust
// Currently in task A
let my_context = UnparkContext::new(my_unpark);

my_context.with(0, || {
    let task = task::park();

    thread::spawn(move || {
        thread::sleep(a_few_secs);
        task.unpark();
    });
});

my_executor.spawn(move || {
    // Currently in task B

    my_context.with(0, || {
        // some work here
    });
});
```

Task B will be the root task that is notified.

carllerche: I should elaborate a bit.

This change should also enable to implement a tokio-core server that can run without allocations. Given that `UnparkContext` is now pretty lightweight, it seems reasonable for a server task to store a slab itself for the futures handling each accepted connection. In this case, the only allocations would be for the task handling the `TcpListener` and a pre-allocated slab for futures handling client sockets. The task then uses `UnparkContext` to either accept from the listener or poll a socket connection task.
carllerche: If this ends up being a direction that we want to pursue, I think that it would be safe to keep this in a `task2` module. I'm fairly confident that the existing task API (`task` module) could use `task2` under the hood.
carllerche: This PR is related to #432. If this strategy is taken, I don't think #432 is relevant anymore.
alexcrichton: This looks pretty slick, I really like it how it's making `with_unpark_event` essentially free (which is not the case today). 

Your code snippet of how behavior changes was intially troubling to me, but in practice I don't think it will matter. Each `UnparkContext` will typically be associated with a particular future, which means that if you call `with` you're typically in a `Future::poll`. In that case you always want to override any previously stored task.

So to make sure I get this straight in my head:

* All top-level executors assign ids to futures.
* Top-level executors create **one** `Arc<Unpark>` for the entire duration of that executor.
* Code currently using `with_unpark_event` will now instead use `UnparkContext`
* `UnparkContext` requires on allocation at creation, and no more afterwards
* Code uses `UnparkContext::with` to receive a custom notification with a custom id whenever something is unparked, allowing it to arrange for it to learn later about precisely which ids were unparked.

Does that sound like a correct interpretation? I can't really see many downsides to this!

Some other random thoughts:

* This is essentially creating a linked list of `Arc`, right? I'd want to think careful about cycles here to see if they're possible, and if necessary we may wish to use weak pointers in some locations.
* Can you explain in more detail how this would work in tokio-core? Currently we create a `SetReadiness` for each future, but would we perhaps no longer do that and just have a linked list of "ready futures"?

carllerche: Re: cycles, I don't think they are possible. `UnparkContext::with` takes `&mut self` and a closure, which prevents nesting `UnparkContext::with` calls w/ the same `self`. Only currently running tasks can be linked together... so it should be fine.

Re: how to hook this into tokio-core, you could most likely get rid of the `SetReadiness` in favor of a linked list of readiness. The slab in -core that tracks tasks could have have a `next_readiness: Option<usize>` for each entry..
alexcrichton: Ok yeah, that both makes sense to me.

Before merging I'd want to see tokio-core working (ideally something like sccache compiling llvm) just to be sure though.
carllerche: Seems fine to me, there is still work to be done though!
carllerche: Ok, I have a tweaked proposal.

* Not introduce `task_impl2` (as requested by @alexcrichton).
* Deprecate the `Unpark` trait.
* Introduce the `Notify` trait:

```rust
pub trait Notify: Send + Sync {
    /// Indicates that an associated future and/or task are ready to make
    /// progress.
    ///
    /// Typically this means that the receiver of the notification should
    /// arrange for the future to get poll'd in a prompt fashion.
    fn notify(&self, id: u64);

    /// Called when an instance of `NotifyHandle` has been cloned.
    fn ref_inc(&self);

    /// Called when an instance of `NotifyHandle` has been dropped.
    fn ref_dec(&self);

    /// A new `Task` handle referencing `id` has been created.
    fn id_ref_inc(&self, id: u64);

    /// A `Task` handle referencing `id` has been dropped.
    fn id_ref_dec(&self, id: u64);
}
```

* Instead of `Arc<Notify>` (or `Arc<Unpark>`), introduce `NotifyHandle` as a wrapper struct:

```rust
struct NotifyHandle {
    ptr: *const NotifyHandle,
}

impl NotifyHandle {
    unsafe pub fn from_ptr(ptr: *const NotifyHandle) -> NotifyHandle {
        unimplemented!();
    }

    pub fn noop() -> NotifyHandle {
        unimplemented!();
    }

    pub fn notify_fn<F: Fn() + Sync + Send>(f: F) -> NotifyHandle {
        unimplemented!();
    }
}
```

* `Rename `task::park()` -> `task::current()` and `Task::unpark()` -> `Task::notify()` (#312 after discussing w/ a number of users, I think this is much clearer naming).

On one hand, `Notify` is a much more complex trait to implement... and very unsafe. However, the `Notify` trait would be able to support executors that run without `std`. `NotifyHandle` would provide a couple of additional constructors (`noop`, `notify_fn`) to assist in constructing "simple" `Notify` implementations.

Another option would be to provide `trait NotifySimple` which is safe and has fewer methods to implement. I prefer sticking with one trait `Notify` as increasing the number of traits increases the complexity of both decisions that users have to make + the implementation.

The one unknown here is how to add `Notify` support to `Spawn`. The easiest option would be to deprecate all existing `poll_*` fns and introduce new ones with a `_notify` syntax.

IMO, `poll_future()` (and co.) should be repurposed in 0.2 to be:

```rust
fn poll_future(&mut self) -> Poll<F::Item, F::Error> {
    self.poll_future_notify(&NotifyHandle::noop())
}
```

Patch:
diff --git a/benches/poll.rs b/benches/poll.rs
new file mode 100644
index 00000000..1fec653f
--- /dev/null
+++ b/benches/poll.rs
@@ -0,0 +1,72 @@
+#![feature(test)]
+
+extern crate futures;
+extern crate test;
+
+use futures::*;
+use futures::executor::{Notify, NotifyHandle};
+use futures::task::Task;
+
+use test::Bencher;
+
+fn notify_noop() -> NotifyHandle {
+    struct Noop;
+
+    impl Notify for Noop {
+        fn notify(&self, _id: usize) {}
+    }
+
+    const NOOP : &'static Noop = &Noop;
+
+    NotifyHandle::from(NOOP)
+}
+
+#[bench]
+fn task_init(b: &mut Bencher) {
+    const NUM: u32 = 100_000;
+
+    struct MyFuture {
+        num: u32,
+        task: Option<Task>,
+    };
+
+    impl Future for MyFuture {
+        type Item = ();
+        type Error = ();
+
+        fn poll(&mut self) -> Poll<(), ()> {
+            if self.num == NUM {
+                Ok(Async::Ready(()))
+            } else {
+                self.num += 1;
+
+                if let Some(ref t) = self.task {
+                    if t.will_notify_current() {
+                        t.notify();
+                        return Ok(Async::NotReady);
+                    }
+                }
+
+                let t = task::current();
+                t.notify();
+                self.task = Some(t);
+
+                Ok(Async::NotReady)
+            }
+        }
+    }
+
+    let notify = notify_noop();
+
+    let mut fut = executor::spawn(MyFuture {
+        num: 0,
+        task: None,
+    });
+
+    b.iter(|| {
+        fut.get_mut().num = 0;
+
+        while let Ok(Async::NotReady) = fut.poll_future_notify(&notify, 0) {
+        }
+    });
+}
diff --git a/src/executor.rs b/src/executor.rs
index 65ebaf0b..518a6e9c 100644
--- a/src/executor.rs
+++ b/src/executor.rs
@@ -7,4 +7,10 @@
 //!
 //! [online]: https://tokio.rs/docs/going-deeper-futures/tasks/
 
-pub use task_impl::{Spawn, spawn, Unpark, Executor, Run};
+#[allow(deprecated)]
+#[cfg(feature = "use_std")]
+pub use task_impl::{Unpark, Executor, Run};
+
+pub use task_impl::{Spawn, spawn, Notify};
+
+pub use task_impl::{UnsafeNotify, NotifyHandle};
diff --git a/src/future/shared.rs b/src/future/shared.rs
index cae415d3..d933093d 100644
--- a/src/future/shared.rs
+++ b/src/future/shared.rs
@@ -14,8 +14,8 @@
 //! ```
 
 use {Future, Poll, Async};
-use executor::{self, Spawn, Unpark};
 use task::{self, Task};
+use executor::{self, Notify, Spawn};
 
 use std::{fmt, mem, ops};
 use std::cell::UnsafeCell;
@@ -49,10 +49,10 @@ struct Inner<F: Future> {
     next_clone_id: AtomicUsize,
     future: UnsafeCell<Option<Spawn<F>>>,
     result: UnsafeCell<Option<Result<SharedItem<F::Item>, SharedError<F::Error>>>>,
-    unparker: Arc<Unparker>,
+    notifier: Arc<Notifier>,
 }
 
-struct Unparker {
+struct Notifier {
     state: AtomicUsize,
     waiters: Mutex<HashMap<usize, Task>>,
 }
@@ -67,7 +67,7 @@ pub fn new<F: Future>(future: F) -> Shared<F> {
     Shared {
         inner: Arc::new(Inner {
             next_clone_id: AtomicUsize::new(1),
-            unparker: Arc::new(Unparker {
+            notifier: Arc::new(Notifier {
                 state: AtomicUsize::new(IDLE),
                 waiters: Mutex::new(HashMap::new()),
             }),
@@ -91,7 +91,7 @@ impl<F> Shared<F> where F: Future {
     /// without blocking. Otherwise, returns None without triggering the work represented by
     /// this `Shared`.
     pub fn peek(&self) -> Option<Result<SharedItem<F::Item>, SharedError<F::Error>>> {
-        match self.inner.unparker.state.load(SeqCst) {
+        match self.inner.notifier.state.load(SeqCst) {
             COMPLETE => {
                 Some(unsafe { self.clone_result() })
             }
@@ -101,8 +101,8 @@ impl<F> Shared<F> where F: Future {
     }
 
     fn set_waiter(&mut self) {
-        let mut waiters = self.inner.unparker.waiters.lock().unwrap();
-        waiters.insert(self.waiter, task::park());
+        let mut waiters = self.inner.notifier.waiters.lock().unwrap();
+        waiters.insert(self.waiter, task::current());
     }
 
     unsafe fn clone_result(&self) -> Result<SharedItem<F::Item>, SharedError<F::Error>> {
@@ -115,8 +115,8 @@ impl<F> Shared<F> where F: Future {
 
     fn complete(&self) {
         unsafe { *self.inner.future.get() = None };
-        self.inner.unparker.state.store(COMPLETE, SeqCst);
-        self.inner.unparker.unpark();
+        self.inner.notifier.state.store(COMPLETE, SeqCst);
+        self.inner.notifier.notify(0);
     }
 }
 
@@ -129,7 +129,7 @@ impl<F> Future for Shared<F>
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         self.set_waiter();
 
-        match self.inner.unparker.state.compare_and_swap(IDLE, POLLING, SeqCst) {
+        match self.inner.notifier.state.compare_and_swap(IDLE, POLLING, SeqCst) {
             IDLE => {
                 // Lock acquired, fall through
             }
@@ -159,23 +159,24 @@ impl<F> Future for Shared<F>
                 }
             }
 
-            let _reset = Reset(&self.inner.unparker.state);
-
-            // Get a handle to the unparker
-            let unpark: Arc<Unpark> = self.inner.unparker.clone();
+            let _reset = Reset(&self.inner.notifier.state);
 
             // Poll the future
-            match unsafe { (*self.inner.future.get()).as_mut().unwrap().poll_future(unpark) } {
+            let res = unsafe {
+                (*self.inner.future.get()).as_mut().unwrap()
+                    .poll_future_notify(&self.inner.notifier, 0)
+            };
+            match res {
                 Ok(Async::NotReady) => {
                     // Not ready, try to release the handle
-                    match self.inner.unparker.state.compare_and_swap(POLLING, IDLE, SeqCst) {
+                    match self.inner.notifier.state.compare_and_swap(POLLING, IDLE, SeqCst) {
                         POLLING => {
                             // Success
                             return Ok(Async::NotReady);
                         }
                         REPOLL => {
                             // Gotta poll again!
-                            let prev = self.inner.unparker.state.swap(POLLING, SeqCst);
+                            let prev = self.inner.notifier.state.swap(POLLING, SeqCst);
                             assert_eq!(prev, REPOLL);
                         }
                         _ => unreachable!(),
@@ -217,19 +218,19 @@ impl<F> Clone for Shared<F> where F: Future {
 
 impl<F> Drop for Shared<F> where F: Future {
     fn drop(&mut self) {
-        let mut waiters = self.inner.unparker.waiters.lock().unwrap();
+        let mut waiters = self.inner.notifier.waiters.lock().unwrap();
         waiters.remove(&self.waiter);
     }
 }
 
-impl Unpark for Unparker {
-    fn unpark(&self) {
+impl Notify for Notifier {
+    fn notify(&self, _id: usize) {
         self.state.compare_and_swap(POLLING, REPOLL, SeqCst);
 
         let waiters = mem::replace(&mut *self.waiters.lock().unwrap(), HashMap::new());
 
         for (_, waiter) in waiters {
-            waiter.unpark();
+            waiter.notify();
         }
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index 48af1121..ed8c93f9 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -196,17 +196,11 @@ pub use future::{
     SelectNext, Then
 };
 
-
 #[cfg(feature = "use_std")]
 mod lock;
-#[cfg(feature = "use_std")]
 mod task_impl;
-#[cfg(feature = "use_std")]
-mod stack;
 
-#[cfg(feature = "use_std")]
 pub mod task;
-#[cfg(feature = "use_std")]
 pub mod executor;
 #[cfg(feature = "use_std")]
 pub mod sync;
diff --git a/src/stack.rs b/src/stack.rs
deleted file mode 100644
index a9639c44..00000000
--- a/src/stack.rs
+++ /dev/null
@@ -1,140 +0,0 @@
-//! A lock-free stack which supports concurrent pushes and a concurrent call to
-//! drain the entire stack all at once.
-
-use std::prelude::v1::*;
-
-use std::mem;
-use std::ptr;
-use std::sync::atomic::AtomicPtr;
-use std::sync::atomic::Ordering::SeqCst;
-
-use task::EventSet;
-
-#[derive(Debug)]
-pub struct Stack<T> {
-    head: AtomicPtr<Node<T>>,
-}
-
-struct Node<T> {
-    data: T,
-    next: *mut Node<T>,
-}
-
-#[derive(Debug)]
-pub struct Drain<T> {
-    head: *mut Node<T>,
-}
-
-unsafe impl<T: Send> Send for Drain<T> {}
-unsafe impl<T: Sync> Sync for Drain<T> {}
-
-impl<T> Stack<T> {
-    pub fn new() -> Stack<T> {
-        Stack {
-            head: AtomicPtr::default(),
-        }
-    }
-
-    pub fn push(&self, data: T) {
-        let mut node = Box::new(Node { data: data, next: ptr::null_mut() });
-        let mut head = self.head.load(SeqCst);
-        loop {
-            node.next = head;
-            match self.head.compare_exchange(head, &mut *node, SeqCst, SeqCst) {
-                Ok(_) => {
-                    mem::forget(node);
-                    return
-                }
-                Err(cur) => head = cur,
-            }
-        }
-    }
-
-    pub fn drain(&self) -> Drain<T> {
-        Drain {
-            head: self.head.swap(ptr::null_mut(), SeqCst),
-        }
-    }
-}
-
-impl<T> Drop for Stack<T> {
-    fn drop(&mut self) {
-        self.drain();
-    }
-}
-
-impl<T> Iterator for Drain<T> {
-    type Item = T;
-
-    fn next(&mut self) -> Option<T> {
-        if self.head.is_null() {
-            return None
-        }
-        unsafe {
-            let node = Box::from_raw(self.head);
-            self.head = node.next;
-            return Some(node.data)
-        }
-    }
-}
-
-impl<T> Drop for Drain<T> {
-    fn drop(&mut self) {
-        for item in self.by_ref() {
-            drop(item);
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::prelude::v1::*;
-    use std::rc::Rc;
-    use std::cell::Cell;
-
-    use super::Stack;
-
-    struct Set(Rc<Cell<usize>>, usize);
-
-    impl Drop for Set {
-        fn drop(&mut self) {
-            self.0.set(self.1);
-        }
-    }
-
-    #[test]
-    fn simple() {
-        let s = Stack::new();
-        s.push(1);
-        s.push(2);
-        s.push(4);
-        assert_eq!(s.drain().collect::<Vec<_>>(), vec![4, 2, 1]);
-        s.push(5);
-        assert_eq!(s.drain().collect::<Vec<_>>(), vec![5]);
-        assert_eq!(s.drain().collect::<Vec<_>>(), vec![]);
-    }
-
-    #[test]
-    fn drain_drops() {
-        let data = Rc::new(Cell::new(0));
-        let s = Stack::new();
-        s.push(Set(data.clone(), 1));
-        drop(s.drain());
-        assert_eq!(data.get(), 1);
-    }
-
-    #[test]
-    fn drop_drops() {
-        let data = Rc::new(Cell::new(0));
-        let s = Stack::new();
-        s.push(Set(data.clone(), 1));
-        drop(s);
-        assert_eq!(data.get(), 1);
-    }
-}
-
-impl EventSet for Stack<usize> {
-    fn insert(&self, id: usize) {
-        self.push(id);
-    }
-}
diff --git a/src/stream/buffer_unordered.rs b/src/stream/buffer_unordered.rs
index 04935d1f..cdece4e7 100644
--- a/src/stream/buffer_unordered.rs
+++ b/src/stream/buffer_unordered.rs
@@ -1,13 +1,8 @@
 use std::prelude::v1::*;
 use std::fmt;
-use std::mem;
-use std::sync::Arc;
 
-use task::{self, UnparkEvent};
-
-use {Async, IntoFuture, Poll, Future};
-use stream::{Stream, Fuse};
-use stack::{Stack, Drain};
+use {Async, IntoFuture, Poll};
+use stream::{Stream, Fuse, FuturesUnordered};
 
 /// An adaptor for a stream of futures to execute the futures concurrently, if
 /// possible, delivering results as they become available.
@@ -22,25 +17,11 @@ pub struct BufferUnordered<S>
 {
     stream: Fuse<S>,
 
-    // A slab of futures that are being executed. Each slot in this vector is
-    // either an active future or a pointer to the next empty slot. This is used
-    // to get O(1) deallocation in the slab and O(1) allocation.
-    //
-    // The `next_future` field is the next slot in the `futures` array that's a
-    // `Slot::Next` variant. If it points to the end of the array then the array
-    // is full.
-    futures: Vec<Slot<<S::Item as IntoFuture>::Future>>,
-    next_future: usize,
-
-    // A list of events that will get pushed onto concurrently by our many
-    // futures. This is filled in and used with the `with_unpark_event`
-    // function. The `pending` list here is the last time we drained events from
-    // our stack.
-    stack: Arc<Stack<usize>>,
-    pending: Drain<usize>,
-
-    // Number of active futures running in the `futures` slab
-    active: usize,
+    // Handles "in-flight" futures
+    queue: FuturesUnordered<<S::Item as IntoFuture>::Future>,
+
+    // Max number of futures
+    max: usize,
 }
 
 impl<S> fmt::Debug for BufferUnordered<S>
@@ -51,32 +32,20 @@ impl<S> fmt::Debug for BufferUnordered<S>
     fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
         fmt.debug_struct("BufferUnordered")
             .field("stream", &self.stream)
-            .field("futures", &self.futures)
-            .field("next_future", &self.next_future)
-            .field("stack", &self.stack)
-            .field("pending", &self.pending)
-            .field("active", &self.active)
+            .field("queue", &self.queue)
+            .field("max", &self.max)
             .finish()
     }
 }
 
-#[derive(Debug)]
-enum Slot<T> {
-    Next(usize),
-    Data(T),
-}
-
 pub fn new<S>(s: S, amt: usize) -> BufferUnordered<S>
     where S: Stream,
           S::Item: IntoFuture<Error=<S as Stream>::Error>,
 {
     BufferUnordered {
         stream: super::fuse::new(s),
-        futures: (0..amt).map(|i| Slot::Next(i + 1)).collect(),
-        next_future: 0,
-        pending: Stack::new().drain(),
-        stack: Arc::new(Stack::new()),
-        active: 0,
+        queue: FuturesUnordered::new(),
+        max: amt,
     }
 }
 
@@ -84,29 +53,6 @@ impl<S> BufferUnordered<S>
     where S: Stream,
           S::Item: IntoFuture<Error=<S as Stream>::Error>,
 {
-    fn poll_pending(&mut self)
-                    -> Option<Poll<Option<<S::Item as IntoFuture>::Item>,
-                                   S::Error>> {
-        while let Some(idx) = self.pending.next() {
-            let result = match self.futures[idx] {
-                Slot::Data(ref mut f) => {
-                    let event = UnparkEvent::new(self.stack.clone(), idx);
-                    match task::with_unpark_event(event, || f.poll()) {
-                        Ok(Async::NotReady) => continue,
-                        Ok(Async::Ready(e)) => Ok(Async::Ready(Some(e))),
-                        Err(e) => Err(e),
-                    }
-                },
-                Slot::Next(_) => continue,
-            };
-            self.active -= 1;
-            self.futures[idx] = Slot::Next(self.next_future);
-            self.next_future = idx;
-            return Some(result)
-        }
-        None
-    }
-
     /// Acquires a reference to the underlying stream that this combinator is
     /// pulling from.
     pub fn get_ref(&self) -> &S {
@@ -141,43 +87,29 @@ impl<S> Stream for BufferUnordered<S>
     fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
         // First up, try to spawn off as many futures as possible by filling up
         // our slab of futures.
-        while self.next_future < self.futures.len() {
+        while self.queue.len() < self.max {
             let future = match try!(self.stream.poll()) {
                 Async::Ready(Some(s)) => s.into_future(),
                 Async::Ready(None) |
                 Async::NotReady => break,
             };
-            self.active += 1;
-            self.stack.push(self.next_future);
-            match mem::replace(&mut self.futures[self.next_future],
-                               Slot::Data(future)) {
-                Slot::Next(next) => self.next_future = next,
-                Slot::Data(_) => panic!(),
-            }
-        }
 
-        // Next, see if our list of `pending` events from last time has any
-        // items, and if so process them here.
-        if let Some(ret) = self.poll_pending() {
-            return ret
+            self.queue.push(future);
         }
 
-        // And finally, take a look at our stack of events, attempting to
-        // process all of those.
-        assert!(self.pending.next().is_none());
-        self.pending = self.stack.drain();
-        if let Some(ret) = self.poll_pending() {
-            return ret
+        // Try polling a new future
+        if let Some(val) = try_ready!(self.queue.poll()) {
+            return Ok(Async::Ready(Some(val)));
         }
 
-        // If we've gotten this far then there's no events for us to process and
-        // nothing was ready, so figure out if we're not done yet or if we've
-        // reached the end.
-        Ok(if self.active > 0 || !self.stream.is_done() {
-            Async::NotReady
+        // If we've gotten this far, then there are no events for us to process
+        // and nothing was ready, so figure out if we're not done yet  or if
+        // we've reached the end.
+        if self.stream.is_done() {
+            Ok(Async::Ready(None))
         } else {
-            Async::Ready(None)
-        })
+            Ok(Async::NotReady)
+        }
     }
 }
 
diff --git a/src/stream/futures_unordered.rs b/src/stream/futures_unordered.rs
index d33f8e5f..17743eae 100644
--- a/src/stream/futures_unordered.rs
+++ b/src/stream/futures_unordered.rs
@@ -1,30 +1,55 @@
-use future::{Future, IntoFuture};
-use stream::Stream;
-use poll::Poll;
-use Async;
-use stack::{Stack, Drain};
-use std::sync::Arc;
-use task::{self, UnparkEvent};
-
-use std::prelude::v1::*;
-
-/// An adaptor for a stream of futures to execute the futures concurrently, if
-/// possible, delivering results as they become available.
+use std::cell::UnsafeCell;
+use std::fmt::{self, Debug};
+use std::marker::PhantomData;
+use std::mem;
+use std::ptr;
+use std::sync::atomic::Ordering::{Relaxed, SeqCst, Acquire, Release, AcqRel};
+use std::sync::atomic::{AtomicPtr, AtomicBool};
+use std::sync::{Arc, Weak};
+use std::usize;
+
+use {task, Stream, Future, Poll, Async, IntoFuture};
+use executor::{Notify, UnsafeNotify, NotifyHandle};
+use task_impl::{self, AtomicTask};
+
+/// An unbounded queue of futures.
+///
+/// This "combinator" also serves a special function in this library, providing
+/// the ability to maintain a queue of futures that and manage driving them all
+/// to completion.
+///
+/// Futures are pushed into this queue and their realized values are yielded as
+/// they are ready. This structure is optimized to manage a large number of
+/// futures. Futures managed by `FuturesUnordered` will only be polled when they
+/// generate notifications. This reduces the required amount of work needed to
+/// coordinate large numbers of futures.
+///
+/// When a `FuturesUnordered` is first created, it does not contain any futures.
+/// Calling `poll` in this state will result in `Ok(Async::Ready(None))` to be
+/// returned. Futures are submitted to the queue using `push`; however, the
+/// future will **not** be polled at this point. `FuturesUnordered` will only
+/// poll managged futures when `FuturesUnordered::poll` is called. As such, it
+/// is important to call `poll` after pushing new futures.
 ///
-/// This adaptor will return their results in the order that they complete.
-/// This is created by the `futures` method.
+/// If `FuturesUnordered::poll` returns `Ok(Async::Ready(None))` this means that
+/// 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
+/// `FuturesUnordered::poll` will either return the future's resolved value
+/// **or** `Ok(Async::NotReady)` if the future has not yet completed.
 ///
-#[derive(Debug)]
+/// Note that you can create a ready-made `FuturesUnordered` via the
+/// `futures_unordered` function in the `stream` module, or you can start with a
+/// blank queue with the `FuturesUnordered::new` constructor.
 #[must_use = "streams do nothing unless polled"]
-pub struct FuturesUnordered<F>
-    where F: Future
-{
-    futures: Vec<Option<F>>,
-    stack: Arc<Stack<usize>>,
-    pending: Option<Drain<usize>>,
-    active: usize,
+pub struct FuturesUnordered<F> {
+    inner: Arc<Inner<F>>,
+    len: usize,
+    head_all: *const Node<F>,
 }
 
+unsafe impl<T: Send> Send for FuturesUnordered<T> {}
+unsafe impl<T: Sync> Sync for FuturesUnordered<T> {}
+
 /// 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,
@@ -32,76 +57,587 @@ pub struct FuturesUnordered<F>
 /// available on the futures internally, in the order that they become
 /// available. This function is similar to `buffer_unordered` in that it may
 /// return items in a different order than in the list specified.
+///
+/// Note that the returned queue can also be used to dynamically push more
+/// futures onto the queue as they become available.
 pub fn futures_unordered<I>(futures: I) -> FuturesUnordered<<I::Item as IntoFuture>::Future>
     where I: IntoIterator,
           I::Item: IntoFuture
 {
-    let futures = futures.into_iter()
-                         .map(IntoFuture::into_future)
-                         .map(Some)
-                         .collect::<Vec<_>>();
-    let stack = Arc::new(Stack::new());
-    for i in 0..futures.len() {
-        stack.push(i);
-    }
-    FuturesUnordered {
-        active: futures.len(),
-        futures: futures,
-        pending: None,
-        stack: stack,
+    let mut queue = FuturesUnordered::new();
+
+    for future in futures {
+        queue.push(future.into_future());
     }
+
+    return queue
 }
 
-impl<F> FuturesUnordered<F>
-    where F: Future
+// FuturesUnordered is implemented using two linked lists. One which links all
+// futures managed by a `FuturesUnordered` and one that tracks futures that have
+// been scheduled for polling. The first linked list is not thread safe and is
+// only accessed by the thread that owns the `FuturesUnordered` value. The
+// second linked list is an implementation of the intrusive MPSC queue algorithm
+// described by 1024cores.net.
+//
+// When a future is submitted to the queue a node is allocated and inserted in
+// both linked lists. The next call to `poll` will (eventually) see this node
+// and call `poll` on the future.
+//
+// Before a managed future is polled, the current task's `Notify` is replaced
+// with one that is aware of the specific future being run. This ensures that
+// task notifications generated by that specific future are visible to
+// `FuturesUnordered`. When a notification is received, the node is scheduled
+// for polling by being inserted into the concurrent linked list.
+//
+// Each node uses an `AtomicUisze` to track it's state. The node state is the
+// reference count (the number of outstanding handles to the node) as well as a
+// flag tracking if the node is currently inserted in the atomic queue. When the
+// future is notified, it will only insert itself into the linked list if it
+// isn't currently inserted.
+
+#[allow(missing_debug_implementations)]
+struct Inner<T> {
+    // The task using `FuturesUnordered`.
+    parent: AtomicTask,
+
+    // Head/tail of the readiness queue
+    head_readiness: AtomicPtr<Node<T>>,
+    tail_readiness: UnsafeCell<*const Node<T>>,
+    stub: Arc<Node<T>>,
+}
+
+struct Node<T> {
+    // The future
+    future: UnsafeCell<Option<T>>,
+
+    // Next pointer for linked list tracking all active nodes
+    next_all: UnsafeCell<*const Node<T>>,
+
+    // Previous node in linked list tracking all active nodes
+    prev_all: UnsafeCell<*const Node<T>>,
+
+    // Next pointer in readiness queue
+    next_readiness: AtomicPtr<Node<T>>,
+
+    // Queue that we'll be enqueued to when notified
+    queue: Weak<Inner<T>>,
+
+    // Whether or not this node is currently in the mpsc queue.
+    queued: AtomicBool,
+}
+
+enum Dequeue<T> {
+    Data(*const Node<T>),
+    Empty,
+    Inconsistent,
+}
+
+impl<T> FuturesUnordered<T>
+    where T: Future,
 {
-    fn poll_pending(&mut self, mut drain: Drain<usize>)
-                    -> Option<Poll<Option<F::Item>, F::Error>> {
-        while let Some(id) = drain.next() {
-            // If this future was already done just skip the notification
-            if self.futures[id].is_none() {
-                continue
+    /// Constructs a new, empty `FuturesUnordered`
+    ///
+    /// The returned `FuturesUnordered` does not contain any futures and, in this
+    /// state, `FuturesUnordered::poll` will return `Ok(Async::Ready(None))`.
+    pub fn new() -> FuturesUnordered<T> {
+        let stub = Arc::new(Node {
+            future: UnsafeCell::new(None),
+            next_all: UnsafeCell::new(ptr::null()),
+            prev_all: UnsafeCell::new(ptr::null()),
+            next_readiness: AtomicPtr::new(ptr::null_mut()),
+            queued: AtomicBool::new(true),
+            queue: Weak::new(),
+        });
+        let stub_ptr = &*stub as *const Node<T>;
+        let inner = Arc::new(Inner {
+            parent: AtomicTask::new(),
+            head_readiness: AtomicPtr::new(stub_ptr as *mut _),
+            tail_readiness: UnsafeCell::new(stub_ptr),
+            stub: stub,
+        });
+
+        FuturesUnordered {
+            len: 0,
+            head_all: ptr::null_mut(),
+            inner: inner,
+        }
+    }
+}
+
+impl<T> FuturesUnordered<T> {
+    /// Returns the number of futures contained by the queue.
+    ///
+    /// This represents the total number of in-flight futures.
+    pub fn len(&self) -> usize {
+        self.len
+    }
+
+    /// Returns `true` if the queue contains no futures
+    pub fn is_empty(&self) -> bool {
+        self.len == 0
+    }
+
+    /// Push a future into the queue.
+    ///
+    /// This function submits the given future to the queue for managing. This
+    /// function will not call `poll` on the submitted future. The caller must
+    /// ensure that `FuturesUnordered::poll` is called in order to receive task
+    /// notifications.
+    pub fn push(&mut self, future: T) {
+        let node = Arc::new(Node {
+            future: UnsafeCell::new(Some(future)),
+            next_all: UnsafeCell::new(ptr::null_mut()),
+            prev_all: UnsafeCell::new(ptr::null_mut()),
+            next_readiness: AtomicPtr::new(ptr::null_mut()),
+            queued: AtomicBool::new(true),
+            queue: Arc::downgrade(&self.inner),
+        });
+
+        // Right now our node has a strong reference count of 1. We transfer
+        // ownership of this reference count to our internal linked list
+        // and we'll reclaim ownership through the `unlink` function below.
+        let ptr = self.link(node);
+
+        // We'll need to get the future "into the system" to start tracking it,
+        // e.g. getting its unpark notifications going to us tracking which
+        // futures are ready. To do that we unconditionally enqueue it for
+        // polling here.
+        self.inner.enqueue(ptr);
+    }
+
+    fn release_node(&mut self, node: Arc<Node<T>>) {
+        // The future is done, try to reset the queued flag. This will prevent
+        // `notify` from doing any work in the future
+        let prev = node.queued.swap(true, SeqCst);
+
+        // Drop the future, even if it hasn't finished yet. This is safe
+        // because we're dropping the future on the thread that owns
+        // `FuturesUnordered`, which correctly tracks T's lifetimes and such.
+        unsafe {
+            drop((*node.future.get()).take());
+        }
+
+        // If the queued flag was previously set then it means that this node
+        // is still in our internal mpsc queue. We then transfer ownership
+        // of our reference count to the mpsc queue, and it'll come along and
+        // free it later, noticing that the future is `None`.
+        //
+        // If, however, the queued flag was *not* set then we're safe to
+        // release our reference count on the internal node. The queued flag
+        // was set above so all future `enqueue` operations will not actually
+        // enqueue the node, so our node will never see the mpsc queue again.
+        // The node itself will be deallocated once all reference counts have
+        // been dropped by the various owning tasks elsewhere.
+        if prev {
+            mem::forget(node);
+        }
+    }
+
+    /// Insert a new node into the internal linked list.
+    fn link(&mut self, node: Arc<Node<T>>) -> *const Node<T> {
+        let ptr = arc2ptr(node);
+        unsafe {
+            *(*ptr).next_all.get() = self.head_all;
+            if !self.head_all.is_null() {
+                *(*self.head_all).prev_all.get() = ptr;
             }
-            let event = UnparkEvent::new(self.stack.clone(), id);
-            let ret = match task::with_unpark_event(event, || {
-                self.futures[id]
-                    .as_mut()
-                    .unwrap()
-                    .poll()
-            }) {
-                Ok(Async::NotReady) => continue,
-                Ok(Async::Ready(val)) => Ok(Async::Ready(Some(val))),
-                Err(e) => Err(e),
-            };
-            self.pending = Some(drain);
-            self.active -= 1;
-            self.futures[id] = None;
-            return Some(ret)
         }
-        None
+
+        self.head_all = ptr;
+        self.len += 1;
+        return ptr
+    }
+
+    /// Remove the node from the linked list tracking all nodes currently
+    /// managed by `FuturesUnordered`.
+    unsafe fn unlink(&mut self, node: *const Node<T>) -> Arc<Node<T>> {
+        let node = ptr2arc(node);
+        let next = *node.next_all.get();
+        let prev = *node.prev_all.get();
+        *node.next_all.get() = ptr::null_mut();
+        *node.prev_all.get() = ptr::null_mut();
+
+        if !next.is_null() {
+            *(*next).prev_all.get() = prev;
+        }
+
+        if !prev.is_null() {
+            *(*prev).next_all.get() = next;
+        } else {
+            self.head_all = next;
+        }
+        self.len -= 1;
+        return node
     }
 }
 
-impl<F> Stream for FuturesUnordered<F>
-    where F: Future
+impl<T> Stream for FuturesUnordered<T>
+    where T: Future
 {
-    type Item = F::Item;
-    type Error = F::Error;
+    type Item = T::Item;
+    type Error = T::Error;
 
-    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-        if self.active == 0 {
-            return Ok(Async::Ready(None))
-        }
-        if let Some(drain) = self.pending.take() {
-            if let Some(ret) = self.poll_pending(drain) {
+    fn poll(&mut self) -> Poll<Option<T::Item>, T::Error> {
+        // Ensure `parent` is correctly set. Note that the `unsafe` here is
+        // because the `park` method underneath needs mutual exclusion from
+        // other calls to `park`, which we guarantee with `&mut self` above and
+        // this is the only method which calls park.
+        unsafe { self.inner.parent.park() };
+
+        loop {
+            let node = match unsafe { self.inner.dequeue() } {
+                Dequeue::Empty => {
+                    if self.is_empty() {
+                        return Ok(Async::Ready(None));
+                    } else {
+                        return Ok(Async::NotReady)
+                    }
+                }
+                Dequeue::Inconsistent => {
+                    // At this point, it may be worth yielding the thread &
+                    // spinning a few times... but for now, just yield using the
+                    // task system.
+                    task::current().notify();
+                    return Ok(Async::NotReady);
+                }
+                Dequeue::Data(node) => node,
+            };
+
+            debug_assert!(node != self.inner.stub());
+
+            unsafe {
+                let mut future = match (*(*node).future.get()).take() {
+                    Some(future) => future,
+
+                    // If the future has already gone away then we're just
+                    // cleaning out this node. See the comment in
+                    // `release_node` for more information, but we're basically
+                    // just taking ownership of our reference count here.
+                    None => {
+                        let node = ptr2arc(node);
+                        assert!((*node.next_all.get()).is_null());
+                        assert!((*node.prev_all.get()).is_null());
+                        continue
+                    }
+                };
+
+                // Unset queued flag... this must be done before
+                // polling. This ensures that the future gets
+                // rescheduled if it is notified **during** a call
+                // to `poll`.
+                let prev = (*node).queued.swap(false, SeqCst);
+                assert!(prev);
+
+                // We're going to need to be very careful if the `poll`
+                // function below panics. We need to (a) not leak memory and
+                // (b) ensure that we still don't have any use-after-frees. To
+                // manage this we do a few things:
+                //
+                // * This "bomb" here will call `release_node` if dropped
+                //   abnormally. That way we'll be sure the memory management
+                //   of the `node` is managed correctly.
+                // * The future was extracted above (taken ownership). That way
+                //   if it panics we're guaranteed that the future is
+                //   dropped on this thread and doesn't accidentally get
+                //   dropped on a different thread (bad).
+                // * We unlink the node from our internal queue to preemptively
+                //   assume it'll panic, in which case we'll want to discard it
+                //   regardless.
+                struct Bomb<'a, T: 'a> {
+                    queue: &'a mut FuturesUnordered<T>,
+                    node: Option<Arc<Node<T>>>,
+                }
+                impl<'a, T> Drop for Bomb<'a, T> {
+                    fn drop(&mut self) {
+                        if let Some(node) = self.node.take() {
+                            self.queue.release_node(node);
+                        }
+                    }
+                }
+                let mut bomb = Bomb {
+                    node: Some(self.unlink(node)),
+                    queue: self,
+                };
+
+                // Poll the underlying future with the appropriate `notify`
+                // implementation. This is where a large bit of the unsafety
+                // starts to stem from internally. The `notify` instance itself
+                // is basically just our `Arc<Node<T>>` and tracks the mpsc
+                // queue of ready futures.
+                //
+                // Critically though `Node<T>` won't actually access `T`, the
+                // future, while it's floating around inside of `Task`
+                // instances. These structs will basically just use `T` to size
+                // the internal allocation, appropriately accessing fields and
+                // deallocating the node if need be.
+                let res = {
+                    let notify = NodeToHandle(bomb.node.as_ref().unwrap());
+                    task_impl::with_notify(&notify, 0, || {
+                        future.poll()
+                    })
+                };
+
+                let ret = match res {
+                    Ok(Async::NotReady) => {
+                        let node = bomb.node.take().unwrap();
+                        *node.future.get() = Some(future);
+                        bomb.queue.link(node);
+                        continue
+                    }
+                    Ok(Async::Ready(e)) => Ok(Async::Ready(Some(e))),
+                    Err(e) => Err(e),
+                };
                 return ret
             }
         }
-        let drain = self.stack.drain();
-        if let Some(ret) = self.poll_pending(drain) {
-            return ret
+    }
+}
+
+impl<T: Debug> Debug for FuturesUnordered<T> {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        write!(fmt, "FuturesUnordered {{ ... }}")
+    }
+}
+
+impl<T> Drop for FuturesUnordered<T> {
+    fn drop(&mut self) {
+        // When a `FuturesUnordered` is dropped we want to drop all futures associated
+        // with it. At the same time though there may be tons of `Task` handles
+        // flying around which contain `Node<T>` references inside them. We'll
+        // let those naturally get deallocated when the `Task` itself goes out
+        // of scope or gets notified.
+        unsafe {
+            while !self.head_all.is_null() {
+                let head = self.head_all;
+                let node = self.unlink(head);
+                self.release_node(node);
+            }
+        }
+
+        // Note that at this point we could still have a bunch of nodes in the
+        // mpsc queue. None of those nodes, however, have futures associated
+        // with them so they're safe to destroy on any thread. At this point
+        // the `FuturesUnordered` struct, the onwer of the one strong reference
+        // to `Inner<T>` will drop the strong reference. At that point
+        // whichever thread releases the strong refcuont last (be it this
+        // thread or some other thread as part of an `upgrade`) will clear out
+        // the mpsc queue and free all remaining nodes.
+        //
+        // While that freeing operation isn't guaranteed to happen here, it's
+        // guaranteed to happen "promptly" as no more "blocking work" will
+        // happen while there's a strong refcount held.
+    }
+}
+
+impl<T> Inner<T> {
+    /// The enqueue function from the 1024cores intrusive MPSC queue algorithm.
+    fn enqueue(&self, node: *const Node<T>) {
+        unsafe {
+            debug_assert!((*node).queued.load(Relaxed));
+
+            // This action does not require any coordination
+            (*node).next_readiness.store(ptr::null_mut(), Relaxed);
+
+            // Note that these atomic orderings come from 1024cores
+            let node = node as *mut _;
+            let prev = self.head_readiness.swap(node, AcqRel);
+            (*prev).next_readiness.store(node, Release);
         }
-        assert!(self.active > 0);
-        Ok(Async::NotReady)
     }
+
+    /// The dequeue function from the 1024cores intrusive MPSC queue algorithm
+    ///
+    /// Note that this unsafe as it required mutual exclusion (only one thread
+    /// can call this) to be guaranteed elsewhere.
+    unsafe fn dequeue(&self) -> Dequeue<T> {
+        let mut tail = *self.tail_readiness.get();
+        let mut next = (*tail).next_readiness.load(Acquire);
+
+        if tail == self.stub() {
+            if next.is_null() {
+                return Dequeue::Empty;
+            }
+
+            *self.tail_readiness.get() = next;
+            tail = next;
+            next = (*next).next_readiness.load(Acquire);
+        }
+
+        if !next.is_null() {
+            *self.tail_readiness.get() = next;
+            debug_assert!(tail != self.stub());
+            return Dequeue::Data(tail);
+        }
+
+        if self.head_readiness.load(Acquire) as *const _ != tail {
+            return Dequeue::Inconsistent;
+        }
+
+        self.enqueue(self.stub());
+
+        next = (*tail).next_readiness.load(Acquire);
+
+        if !next.is_null() {
+            *self.tail_readiness.get() = next;
+            return Dequeue::Data(tail);
+        }
+
+        Dequeue::Inconsistent
+    }
+
+    fn stub(&self) -> *const Node<T> {
+        &*self.stub
+    }
+}
+
+impl<T> Drop for Inner<T> {
+    fn drop(&mut self) {
+        // Once we're in the destructor for `Inner<T>` we need to clear out the
+        // mpsc queue of nodes if there's anything left in there.
+        //
+        // Note that each node has a strong reference count associated with it
+        // which is owned by the mpsc queue. All nodes should have had their
+        // futures dropped already by the `FuturesUnordered` destructor above,
+        // so we're just pulling out nodes and dropping their refcounts.
+        unsafe {
+            loop {
+                match self.dequeue() {
+                    Dequeue::Empty => break,
+                    Dequeue::Inconsistent => abort("inconsistent in drop"),
+                    Dequeue::Data(ptr) => drop(ptr2arc(ptr)),
+                }
+            }
+        }
+    }
+}
+
+#[allow(missing_debug_implementations)]
+struct NodeToHandle<'a, T: 'a>(&'a Arc<Node<T>>);
+
+impl<'a, T> Clone for NodeToHandle<'a, T> {
+    fn clone(&self) -> Self {
+        NodeToHandle(self.0)
+    }
+}
+
+impl<'a, T> From<NodeToHandle<'a, T>> for NotifyHandle {
+    fn from(handle: NodeToHandle<'a, T>) -> NotifyHandle {
+        unsafe {
+            let ptr = handle.0.clone();
+            let ptr = mem::transmute::<Arc<Node<T>>, *mut ArcNode<T>>(ptr);
+            NotifyHandle::new(hide_lt(ptr))
+        }
+    }
+}
+
+struct ArcNode<T>(PhantomData<T>);
+
+// We should never touch `T` on any thread other than the one owning
+// `FuturesUnordered`, so this should be a safe operation.
+unsafe impl<T> Send for ArcNode<T> {}
+unsafe impl<T> Sync for ArcNode<T> {}
+
+impl<T> Notify for ArcNode<T> {
+    fn notify(&self, _id: usize) {
+        unsafe {
+            let me: *const ArcNode<T> = self;
+            let me: *const *const ArcNode<T> = &me;
+            let me = me as *const Arc<Node<T>>;
+            Node::notify(&*me)
+        }
+    }
+}
+
+unsafe impl<T> UnsafeNotify for ArcNode<T> {
+    unsafe fn clone_raw(&self) -> NotifyHandle {
+        let me: *const ArcNode<T> = self;
+        let me: *const *const ArcNode<T> = &me;
+        let me = &*(me as *const Arc<Node<T>>);
+        NodeToHandle(me).into()
+    }
+
+    unsafe fn drop_raw(&self) {
+        let mut me: *const ArcNode<T> = self;
+        let me = &mut me as *mut *const ArcNode<T> as *mut Arc<Node<T>>;
+        ptr::drop_in_place(me);
+    }
+}
+
+unsafe fn hide_lt<T>(p: *mut ArcNode<T>) -> *mut UnsafeNotify {
+    mem::transmute(p as *mut UnsafeNotify)
+}
+
+impl<T> Node<T> {
+    fn notify(me: &Arc<Node<T>>) {
+        let inner = match me.queue.upgrade() {
+            Some(inner) => inner,
+            None => return,
+        };
+
+        // It's our job to notify the node that it's ready to get polled,
+        // meaning that we need to enqueue it into the readiness queue. To
+        // do this we flag that we're ready to be queued, and if successful
+        // we then do the literal queueing operation, ensuring that we're
+        // only queued once.
+        //
+        // Once the node is inserted we be sure to notify the parent task,
+        // as it'll want to come along and pick up our node now.
+        //
+        // Note that we don't change the reference count of the node here,
+        // we're just enqueueing the raw pointer. The `FuturesUnordered`
+        // implementation guarantees that if we set the `queued` flag true that
+        // there's a reference count held by the main `FuturesUnordered` queue
+        // still.
+        let prev = me.queued.swap(true, SeqCst);
+        if !prev {
+            inner.enqueue(&**me);
+            inner.parent.notify();
+        }
+    }
+}
+
+impl<T> Drop for Node<T> {
+    fn drop(&mut self) {
+        // Currently a `Node<T>` is sent across all threads for any lifetime,
+        // regardless of `T`. This means that for memory safety we can't
+        // actually touch `T` at any time except when we have a reference to the
+        // `FuturesUnordered` itself.
+        //
+        // Consequently it *should* be the case that we always drop futures from
+        // the `FuturesUnordered` instance, but this is a bomb in place to catch
+        // any bugs in that logic.
+        unsafe {
+            if (*self.future.get()).is_some() {
+                abort("future still here when dropping");
+            }
+        }
+    }
+}
+
+fn arc2ptr<T>(ptr: Arc<T>) -> *const T {
+    let addr = &*ptr as *const T;
+    mem::forget(ptr);
+    return addr
+}
+
+unsafe fn ptr2arc<T>(ptr: *const T) -> Arc<T> {
+    let anchor = mem::transmute::<usize, Arc<T>>(0x10);
+    let addr = &*anchor as *const T;
+    mem::forget(anchor);
+    let offset = addr as isize - 0x10;
+    mem::transmute::<isize, Arc<T>>(ptr as isize - offset)
+}
+
+fn abort(s: &str) -> ! {
+    struct DoublePanic;
+
+    impl Drop for DoublePanic {
+        fn drop(&mut self) {
+            panic!("panicking twice to abort the program");
+        }
+    }
+
+    let _bomb = DoublePanic;
+    panic!("{}", s);
 }
diff --git a/src/sync/bilock.rs b/src/sync/bilock.rs
index b76d5d00..f36426b1 100644
--- a/src/sync/bilock.rs
+++ b/src/sync/bilock.rs
@@ -90,7 +90,7 @@ impl<T> BiLock<T> {
                 }
             }
 
-            let me = Box::new(task::park());
+            let me = Box::new(task::current());
             let me = Box::into_raw(me) as usize;
 
             match self.inner.state.compare_exchange(1, me, SeqCst, SeqCst) {
@@ -143,7 +143,7 @@ impl<T> BiLock<T> {
             // Another task has parked themselves on this lock, let's wake them
             // up as its now their turn.
             n => unsafe {
-                Box::from_raw(n as *mut Task).unpark();
+                Box::from_raw(n as *mut Task).notify();
             }
         }
     }
diff --git a/src/sync/mpsc/mod.rs b/src/sync/mpsc/mod.rs
index 34212b1f..e5b8c01c 100644
--- a/src/sync/mpsc/mod.rs
+++ b/src/sync/mpsc/mod.rs
@@ -331,7 +331,7 @@ impl<T> Sender<T> {
         // be parked. This will send the task handle on the parked task queue.
         //
         // However, when `do_send` is called while dropping the `Sender`,
-        // `task::park()` can't be called safely. In this case, in order to
+        // `task::current()` can't be called safely. In this case, in order to
         // maintain internal consistency, a blank message is pushed onto the
         // parked task queue.
         if park_self {
@@ -435,15 +435,15 @@ impl<T> Sender<T> {
         };
 
         if let Some(task) = task {
-            task.unpark();
+            task.notify();
         }
     }
 
     fn park(&mut self, can_park: bool) {
-        // TODO: clean up internal state if the task::park will fail
+        // TODO: clean up internal state if the task::current will fail
 
         let task = if can_park {
-            Some(task::park())
+            Some(task::current())
         } else {
             None
         };
@@ -478,7 +478,7 @@ impl<T> Sender<T> {
             //
             // Update the task in case the `Sender` has been moved to another
             // task
-            *task = Some(task::park());
+            *task = Some(task::current());
 
             Async::NotReady
         } else {
@@ -647,7 +647,7 @@ impl<T> Receiver<T> {
                 PopResult::Data(task) => {
                     let task = task.lock().unwrap().take();
                     if let Some(task) = task {
-                        task.unpark();
+                        task.notify();
                     }
                 }
                 PopResult::Empty => break,
@@ -675,7 +675,7 @@ impl<T> Receiver<T> {
                     //
                     // 1) Spin
                     // 2) thread::yield_now()
-                    // 3) task::park().unwrap() & return NotReady
+                    // 3) task::current().unwrap() & return NotReady
                     //
                     // For now, thread::yield_now() is used, but it would
                     // probably be better to spin a few times then yield.
@@ -695,7 +695,7 @@ impl<T> Receiver<T> {
                     let task = task.lock().unwrap().take();
 
                     if let Some(task) = task {
-                        task.unpark();
+                        task.notify();
                     }
 
                     return;
@@ -731,7 +731,7 @@ impl<T> Receiver<T> {
             return TryPark::NotEmpty;
         }
 
-        recv_task.task = Some(task::park());
+        recv_task.task = Some(task::current());
         TryPark::Parked
     }
 
diff --git a/src/sync/oneshot.rs b/src/sync/oneshot.rs
index 68d369a4..2aa55985 100644
--- a/src/sync/oneshot.rs
+++ b/src/sync/oneshot.rs
@@ -183,7 +183,7 @@ impl<T> Sender<T> {
         // may have been dropped. The first thing it does is set the flag, and
         // if it fails to acquire the lock it assumes that we'll see the flag
         // later on. So... we then try to see the flag later on!
-        let handle = task::park();
+        let handle = task::current();
         match self.inner.tx_task.try_lock() {
             Some(mut p) => *p = Some(handle),
             None => return Ok(Async::Ready(())),
@@ -222,7 +222,7 @@ impl<T> Drop for Sender<T> {
         if let Some(mut slot) = self.inner.rx_task.try_lock() {
             if let Some(task) = slot.take() {
                 drop(slot);
-                task.unpark();
+                task.notify();
             }
         }
     }
@@ -259,7 +259,7 @@ impl<T> Receiver<T> {
         if let Some(mut handle) = self.inner.tx_task.try_lock() {
             if let Some(task) = handle.take() {
                 drop(handle);
-                task.unpark()
+                task.notify()
             }
         }
     }
@@ -282,7 +282,7 @@ impl<T> Future for Receiver<T> {
         if self.inner.complete.load(SeqCst) {
             done = true;
         } else {
-            let task = task::park();
+            let task = task::current();
             match self.inner.rx_task.try_lock() {
                 Some(mut slot) => *slot = Some(task),
                 None => done = true,
@@ -334,7 +334,7 @@ impl<T> Drop for Receiver<T> {
         if let Some(mut handle) = self.inner.tx_task.try_lock() {
             if let Some(task) = handle.take() {
                 drop(handle);
-                task.unpark()
+                task.notify()
             }
         }
     }
diff --git a/src/task.rs b/src/task.rs
index cd12f382..fa018c4e 100644
--- a/src/task.rs
+++ b/src/task.rs
@@ -22,19 +22,25 @@
 //!
 //! ## Functions
 //!
-//! There is an important bare function in this module: `park`. The `park`
-//! function is similar to the standard library's `thread::park` method where it
-//! returns a handle to wake up a task at a later date (via an `unpark` method).
+//! There is an important bare function in this module: `current`. The
+//! `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.
 
 #[doc(hidden)]
 #[deprecated(since = "0.1.4", note = "import through the executor module instead")]
-#[cfg(feature = "with-deprecated")]
-pub use task_impl::{Spawn, spawn, Unpark, Executor, Run};
+#[cfg(all(feature = "with-deprecated", feature = "use_std"))]
+#[allow(deprecated)]
+pub use task_impl::{Spawn, spawn, Unpark, Executor, Run, park};
+
+pub use task_impl::{Task, current, init};
 
-pub use task_impl::{Task, LocalKey, park, with_unpark_event, UnparkEvent, EventSet};
+#[allow(deprecated)]
+#[cfg(feature = "use_std")]
+pub use task_impl::{LocalKey, with_unpark_event, UnparkEvent, EventSet};
 
 #[doc(hidden)]
 #[deprecated(since = "0.1.4", note = "import through the executor module instead")]
-#[cfg(feature = "with-deprecated")]
+#[cfg(all(feature = "with-deprecated", feature = "use_std"))]
 #[allow(deprecated)]
 pub use task_impl::TaskRc;
diff --git a/src/task_impl/atomic_task.rs b/src/task_impl/atomic_task.rs
new file mode 100644
index 00000000..002959d4
--- /dev/null
+++ b/src/task_impl/atomic_task.rs
@@ -0,0 +1,147 @@
+#![allow(dead_code)]
+
+use super::Task;
+
+use core::fmt;
+use core::cell::UnsafeCell;
+use core::sync::atomic::AtomicUsize;
+use core::sync::atomic::Ordering::{Acquire, Release};
+
+/// A coordinated `Task` handle enabling concurrent operations on a task.
+pub struct AtomicTask {
+    state: AtomicUsize,
+    task: UnsafeCell<Option<Task>>,
+}
+
+/// Initial state, the `AtomicTask` is currently not being used.
+///
+/// The value `2` is picked specifically because it between the write lock &
+/// read lock values. Since the read lock is represented by an incrementing
+/// counter, this enables an atomic fetch_sub operation to be used for releasing
+/// a lock.
+const WAITING: usize = 2;
+
+/// The `park` function has determined that the task is no longer current. This
+/// implies that `AtomicTask::park` is being called from a different task than
+/// is represented by the currently stored task. The write lock is obtained to
+/// update the task cell.
+const LOCKED_WRITE: usize = 0;
+
+/// At least one call to `notify` happened concurrently to `park` updating the
+/// task cell. This state is detected when `park` exits the mutation code and
+/// signals to `park` that it is responsible for notifying its own task.
+const LOCKED_WRITE_NOTIFIED: usize = 1;
+
+
+/// The `notify` function has locked access to the task cell for notification.
+///
+/// The constant is left here mostly for documentation reasons.
+#[allow(dead_code)]
+const LOCKED_READ: usize = 3;
+
+impl AtomicTask {
+    /// Create an `AtomicTask` initialized with the given `Task`
+    pub fn new() -> AtomicTask {
+        // Make sure that task is Sync
+        fn is_sync<T: Sync>() {}
+        is_sync::<Task>();
+
+        AtomicTask {
+            state: AtomicUsize::new(WAITING),
+            task: UnsafeCell::new(None),
+        }
+    }
+
+    /// The caller must ensure mutual exclusion
+    pub unsafe fn park(&self) {
+        if let Some(ref task) = *self.task.get() {
+            if task.will_notify_current() {
+                // Nothing more to do
+                return
+            }
+        }
+
+        // Get a new task handle
+        let task = super::current();
+
+        match self.state.compare_and_swap(WAITING, LOCKED_WRITE, Acquire) {
+            WAITING => {
+                // Locked acquired, update the task cell
+                *self.task.get() = Some(task);
+
+                // Release the lock. If the state transitioned to
+                // `LOCKED_NOTIFIED`, this means that an notify has been
+                // signaled, so notify the task.
+                if LOCKED_WRITE_NOTIFIED == self.state.swap(WAITING, Release) {
+                    (*self.task.get()).as_ref().unwrap().notify();
+                }
+            }
+            state => {
+                debug_assert!(state != LOCKED_WRITE, "unexpected state LOCKED_WRITE");
+                debug_assert!(state != LOCKED_WRITE_NOTIFIED, "unexpected state LOCKED_WRITE_NOTIFIED");
+
+                // Currently in a read locked state, this implies that `notify`
+                // is currently being called on the old task handle. So, we call
+                // notify on the new task handle
+                task.notify();
+            }
+        }
+    }
+
+    pub fn notify(&self) {
+        let mut curr = WAITING;
+
+        loop {
+            if curr == LOCKED_WRITE {
+                // Transition the state to LOCKED_NOTIFIED
+                let actual = self.state.compare_and_swap(LOCKED_WRITE, LOCKED_WRITE_NOTIFIED, Release);
+
+                if curr == actual {
+                    // Success, return
+                    return;
+                }
+
+                // update current state variable and try again
+                curr = actual;
+
+            } else if curr == LOCKED_WRITE_NOTIFIED {
+                // Currently in `LOCKED_WRITE_NOTIFIED` state, nothing else to do.
+                return;
+
+            } else {
+                // Currently in a LOCKED_READ state, so attempt to increment the
+                // lock count.
+                let actual = self.state.compare_and_swap(curr, curr + 1, Acquire);
+
+                // Locked acquired
+                if actual == curr {
+                    // Notify the task
+                    unsafe {
+                        if let Some(ref task) = *self.task.get() {
+                            task.notify();
+                        }
+                    }
+
+                    // Release the lock
+                    self.state.fetch_sub(1, Release);
+
+                    // Done
+                    return;
+                }
+
+                // update current state variable and try again
+                curr = actual;
+
+            }
+        }
+    }
+}
+
+impl fmt::Debug for AtomicTask {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        write!(fmt, "AtomicTask")
+    }
+}
+
+unsafe impl Send for AtomicTask {}
+unsafe impl Sync for AtomicTask {}
diff --git a/src/task_impl/core.rs b/src/task_impl/core.rs
new file mode 100644
index 00000000..05ce8cfd
--- /dev/null
+++ b/src/task_impl/core.rs
@@ -0,0 +1,173 @@
+#![cfg_attr(feature = "use_std", allow(dead_code))]
+
+use core::marker;
+use core::mem;
+use core::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT};
+use core::sync::atomic::Ordering::{SeqCst, Relaxed};
+
+use super::{BorrowedTask, NotifyHandle};
+
+pub struct LocalKey;
+pub struct LocalMap;
+pub fn local_map() -> LocalMap { LocalMap }
+
+#[derive(Copy, Clone)]
+pub struct BorrowedEvents<'a>(marker::PhantomData<&'a ()>);
+
+#[derive(Copy, Clone)]
+pub struct BorrowedUnpark<'a> {
+    f: &'a Fn() -> NotifyHandle,
+    id: usize,
+}
+
+pub struct TaskUnpark {
+    handle: NotifyHandle,
+    id: usize,
+}
+
+#[derive(Clone)]
+pub struct UnparkEvents;
+
+impl<'a> BorrowedEvents<'a> {
+    pub fn new() -> BorrowedEvents<'a> {
+        BorrowedEvents(marker::PhantomData)
+    }
+
+    pub fn to_owned(&self) -> UnparkEvents {
+        UnparkEvents
+    }
+}
+
+impl<'a> BorrowedUnpark<'a> {
+    #[inline]
+    pub fn new(f: &'a Fn() -> NotifyHandle, id: usize) -> BorrowedUnpark<'a> {
+        BorrowedUnpark { f: f, id: id }
+    }
+
+    #[inline]
+    pub fn to_owned(&self) -> TaskUnpark {
+        let handle = (self.f)();
+        let id = handle.clone_id(self.id);
+        TaskUnpark { handle: handle, id: id }
+    }
+}
+
+impl UnparkEvents {
+    pub fn notify(&self) {}
+
+    pub fn will_notify(&self, _other: &BorrowedEvents) -> bool {
+        true
+    }
+}
+
+impl TaskUnpark {
+    pub fn notify(&self) {
+        self.handle.notify(self.id);
+    }
+
+    pub fn will_notify(&self, other: &BorrowedUnpark) -> bool {
+        self.id == other.id && self.handle.inner == (other.f)().inner
+    }
+}
+
+impl Clone for TaskUnpark {
+    fn clone(&self) -> TaskUnpark {
+        let handle = self.handle.clone();
+        let id = handle.clone_id(self.id);
+        TaskUnpark { handle: handle, id: id }
+    }
+}
+
+impl Drop for TaskUnpark {
+    fn drop(&mut self) {
+        self.handle.drop_id(self.id);
+    }
+}
+
+static GET: AtomicUsize = ATOMIC_USIZE_INIT;
+static SET: AtomicUsize = ATOMIC_USIZE_INIT;
+
+/// Initialize the `futures` task system.
+///
+/// This function is an unsafe low-level implementation detail typically only
+/// used by crates using `futures` in `no_std` context. Users of this crate
+/// who also use the standard library never need to invoke this function.
+///
+/// The task system in the `futures` crate relies on some notion of "local
+/// storage" for the running thread and/or context. The `task::current` function
+/// can get invoked in any context, for example, and needs to be able to return
+/// a `Task`. Typically with the standard library this is supported with
+/// thread-local-storage, but this is not available in `no_std` contexts!
+///
+/// This function is provided to allow `no_std` contexts to continue to be able
+/// to use the standard task system in this crate. The functions provided here
+/// will be used as-if they were thread-local-storage getters/setters. The `get`
+/// function provided is used to retrieve the current thread-local value of the
+/// task system's pointer, returning null if not initialized. The `set` function
+/// updates the value of the pointer.
+///
+/// # Return value
+///
+/// This function will return whether initialization succeeded or not. This
+/// function can be called concurrently and only the first invocation will
+/// succeed. If `false` is returned then the `get` and `set` pointers provided
+/// were *not* registered for use with the task system, but if `true` was
+/// provided then they will be called when the task system is used.
+///
+/// Note that while safe to call concurrently it's recommended to still perform
+/// external synchronization when calling this function. This task system is
+/// not guaranteed to be ready to go until a call to this function returns
+/// `true`. In other words, if you call this function and see `false`, the
+/// task system may not be ready to go as another thread may still be calling
+/// `init`.
+///
+/// # Unsafety
+///
+/// This function is unsafe due to the requirements on the behavior of the
+/// `get` and `set` functions. The pointers returned from these functions must
+/// reflect the semantics specified above and must also be thread-local,
+/// depending on the definition of a "thread" in the calling context.
+pub unsafe fn init(get: fn() -> *mut u8, set: fn(*mut u8)) -> bool {
+    if GET.compare_exchange(0, get as usize, SeqCst, SeqCst).is_ok() {
+        SET.store(set as usize, SeqCst);
+        true
+    } else {
+        false
+    }
+}
+
+#[inline]
+pub fn get_ptr() -> *mut u8 {
+    match GET.load(Relaxed) {
+        0 => panic!("not initialized"),
+        n => unsafe { mem::transmute::<usize, fn() -> *mut u8>(n)() },
+    }
+}
+
+#[cfg(feature = "use_std")]
+#[inline]
+pub fn is_get_ptr(f: usize) -> bool {
+    GET.load(Relaxed) == f
+}
+
+pub fn set<'a, F, R>(task: &BorrowedTask<'a>, f: F) -> R
+    where F: FnOnce() -> R
+{
+    let set = match SET.load(Relaxed) {
+        0 => panic!("not initialized"),
+        n => unsafe { mem::transmute::<usize, fn(*mut u8)>(n) },
+    };
+
+    struct Reset(fn(*mut u8), *mut u8);
+
+    impl Drop for Reset {
+        #[inline]
+        fn drop(&mut self) {
+            (self.0)(self.1);
+        }
+    }
+
+    let _reset = Reset(set, get_ptr());
+    set(task as *const _ as *mut u8);
+    f()
+}
diff --git a/src/task_impl/mod.rs b/src/task_impl/mod.rs
index a261a76a..de41781c 100644
--- a/src/task_impl/mod.rs
+++ b/src/task_impl/mod.rs
@@ -1,39 +1,36 @@
-use std::prelude::v1::*;
+use core::fmt;
+use core::marker::PhantomData;
 
-use std::cell::Cell;
-use std::fmt;
-use std::mem;
-use std::sync::Arc;
-use std::sync::atomic::{Ordering, AtomicBool, AtomicUsize, ATOMIC_USIZE_INIT};
-use std::thread;
+use {Poll, Future, Stream, Sink, StartSend};
 
-use {Poll, Future, Async, Stream, Sink, StartSend, AsyncSink};
-use future::BoxFuture;
+mod atomic_task;
+pub use self::atomic_task::AtomicTask;
 
-mod unpark_mutex;
-use self::unpark_mutex::UnparkMutex;
+mod core;
 
-mod task_rc;
-mod data;
-#[allow(deprecated)]
-#[cfg(feature = "with-deprecated")]
-pub use self::task_rc::TaskRc;
-pub use self::data::LocalKey;
+#[cfg(feature = "use_std")]
+mod std;
+#[cfg(feature = "use_std")]
+pub use self::std::*;
+#[cfg(not(feature = "use_std"))]
+pub use self::core::*;
 
-struct BorrowedTask<'a> {
+pub struct BorrowedTask<'a> {
     id: usize,
-    unpark: &'a Arc<Unpark>,
-    map: &'a data::LocalMap,
-    events: Events,
+    unpark: BorrowedUnpark<'a>,
+    events: BorrowedEvents<'a>,
+    // Task-local storage
+    map: &'a LocalMap,
 }
 
-thread_local!(static CURRENT_TASK: Cell<*const BorrowedTask<'static>> = {
-    Cell::new(0 as *const _)
-});
-
 fn fresh_task_id() -> usize {
+    use core::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
+
     // TODO: this assert is a real bummer, need to figure out how to reuse
     //       old IDs that are no longer in use.
+    //
+    // Note, though, that it is intended that these ids go away entirely
+    // eventually, see the comment on `is_current` below.
     static NEXT_ID: AtomicUsize = ATOMIC_USIZE_INIT;
     let id = NEXT_ID.fetch_add(1, Ordering::Relaxed);
     assert!(id < usize::max_value() / 2,
@@ -41,32 +38,11 @@ fn fresh_task_id() -> usize {
     id
 }
 
-fn set<'a, F, R>(task: &BorrowedTask<'a>, f: F) -> R
-    where F: FnOnce() -> R
-{
-    struct Reset(*const BorrowedTask<'static>);
-    impl Drop for Reset {
-        fn drop(&mut self) {
-            CURRENT_TASK.with(|c| c.set(self.0));
-        }
-    }
-
-    CURRENT_TASK.with(move |c| {
-        let _reset = Reset(c.get());
-        let task = unsafe {
-            mem::transmute::<&BorrowedTask<'a>,
-                             *const BorrowedTask<'static>>(task)
-        };
-        c.set(task);
-        f()
-    })
-}
-
 fn with<F: FnOnce(&BorrowedTask) -> R, R>(f: F) -> R {
-    let task = CURRENT_TASK.with(|c| c.get());
-    assert!(!task.is_null(), "no Task is currently running");
     unsafe {
-        f(&*task)
+        let task = get_ptr();
+        assert!(!task.is_null(), "no Task is currently running");
+        f(&*(task as *const BorrowedTask))
     }
 }
 
@@ -77,12 +53,12 @@ fn with<F: FnOnce(&BorrowedTask) -> R, R>(f: F) -> R {
 /// spawned as tasks onto an *executor*. The executor is responsible for polling
 /// the future as notifications arrive, until the future terminates.
 ///
-/// This is obtained by the `task::park` function.
+/// This is obtained by the `task::current` function.
 #[derive(Clone)]
 pub struct Task {
     id: usize,
-    unpark: Arc<Unpark>,
-    events: Events,
+    unpark: TaskUnpark,
+    events: UnparkEvents,
 }
 
 fn _assert_kinds() {
@@ -90,11 +66,7 @@ fn _assert_kinds() {
     _assert_send::<Task>();
 }
 
-/// Returns a handle to the current task to call `unpark` at a later date.
-///
-/// This function is similar to the standard library's `thread::park` function
-/// except that it won't block the current thread but rather the current future
-/// that is being executed.
+/// Returns a handle to the current task to call `notify` at a later date.
 ///
 /// The returned handle implements the `Send` and `'static` bounds and may also
 /// be cheaply cloned. This is useful for squirreling away the handle into a
@@ -102,173 +74,134 @@ fn _assert_kinds() {
 ///
 /// Implementations of the `Future` trait typically use this function if they
 /// would otherwise perform a blocking operation. When something isn't ready
-/// yet, this `park` function is called to acquire a handle to the current
-/// task, and then the future arranges it such that when the block operation
-/// otherwise finishes (perhaps in the background) it will `unpark` the returned
-/// handle.
+/// yet, this `current` function is called to acquire a handle to the current
+/// task, and then the future arranges it such that when the blocking operation
+/// otherwise finishes (perhaps in the background) it will `notify` the
+/// returned handle.
 ///
 /// It's sometimes necessary to pass extra information to the task when
-/// unparking it, so that the task knows something about *why* it was woken. See
-/// the `with_unpark_event` for details on how to do this.
+/// unparking it, so that the task knows something about *why* it was woken.
+/// See the `FutureQueue` documentation for details on how to do this.
 ///
 /// # Panics
 ///
 /// This function will panic if a task is not currently being executed. That
 /// is, this method can be dangerous to call outside of an implementation of
 /// `poll`.
-pub fn park() -> Task {
-    with(|task| {
+pub fn current() -> Task {
+    with(|borrowed| {
+        let unpark = borrowed.unpark.to_owned();
+        let events = borrowed.events.to_owned();
+
         Task {
-            id: task.id,
-            events: task.events.clone(),
-            unpark: task.unpark.clone(),
+            id: borrowed.id,
+            unpark: unpark,
+            events: events,
         }
     })
 }
 
+#[doc(hidden)]
+#[deprecated(note = "renamed to `current`")]
+pub fn park() -> Task {
+    current()
+}
+
 impl Task {
     /// Indicate that the task should attempt to poll its future in a timely
     /// fashion.
     ///
-    /// It's typically guaranteed that, for each call to `unpark`, `poll` will
-    /// be called at least once subsequently (unless the task has terminated).
-    /// If the task is currently polling its future when `unpark` is called, it
+    /// It's typically guaranteed that, for each call to `notify`, `poll` will
+    /// be called at least once subsequently (unless the future has terminated).
+    /// If the task is currently polling its future when `notify` is called, it
     /// must poll the future *again* afterwards, ensuring that all relevant
     /// events are eventually observed by the future.
+    pub fn notify(&self) {
+        self.events.notify();
+        self.unpark.notify();
+    }
+
+    #[doc(hidden)]
+    #[deprecated(note = "renamed to `notify`")]
     pub fn unpark(&self) {
-        self.events.trigger();
-        self.unpark.unpark();
+        self.notify()
     }
 
-    /// Returns `true` when called from within the context of the task. In
-    /// other words, the task is currently running on the thread calling the
-    /// function.
+    /// Returns `true` when called from within the context of the task.
+    ///
+    /// In other words, the task is currently running on the thread calling the
+    /// function. Note that this is currently, and has historically, been
+    /// implemented by tracking an `id` on every instance of `Spawn` created.
+    /// When a `Spawn` is being polled it stores in thread-local-storage the id
+    /// of the instance, and then `task::current` will return a `Task` that also
+    /// stores this id.
+    ///
+    /// The intention of this function was to answer questions like "if I
+    /// `notify` this task, is it equivalent to `task::current().notify()`?"
+    /// The answer "yes" may be able to avoid some extra work to block the
+    /// current task, such as sending a task along a channel or updating a
+    /// stored `Task` somewhere. An answer of "no" typically results in doing
+    /// the work anyway.
+    ///
+    /// Unfortunately this function has been somewhat buggy in the past and is
+    /// not intended to be supported in the future. By simply matching `id` the
+    /// intended question above isn't accurately taking into account, for
+    /// example, unpark events (now deprecated, but still a feature). Thus many
+    /// old users of this API weren't fully accounting for the question it was
+    /// intended they were asking.
+    ///
+    /// This API continues to be implemented but will in the future, e.g. in the
+    /// 0.1.x series of this crate, eventually return `false` unconditionally.
+    /// It is intended that this function will be removed in the next breaking
+    /// change of this crate. If you'd like to continue to be able to answer the
+    /// example question above, it's recommended you use the
+    /// `will_notify_current` method.
+    ///
+    /// If you've got questions about this though please let us know! We'd like
+    /// to learn about other use cases here that we did not consider.
+    ///
+    /// # Panics
+    ///
+    /// This function will panic if no current future is being polled.
+    #[deprecated(note = "intended to be removed, see docs for details")]
     pub fn is_current(&self) -> bool {
         with(|current| current.id == self.id)
     }
-}
-
-impl fmt::Debug for Task {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("Task")
-         .field("id", &self.id)
-         .finish()
-    }
-}
 
-/// For the duration of the given callback, add an "unpark event" to be
-/// triggered when the task handle is used to unpark the task.
-///
-/// Unpark events are used to pass information about what event caused a task to
-/// be unparked. In some cases, tasks are waiting on a large number of possible
-/// events, and need precise information about the wakeup to avoid extraneous
-/// polling.
-///
-/// Every `Task` handle comes with a set of unpark events which will fire when
-/// `unpark` is called. When fired, these events insert an identifer into a
-/// concurrent set, which the task can read from to determine what events
-/// occurred.
-///
-/// This function immediately invokes the closure, `f`, but arranges things so
-/// that `task::park` will produce a `Task` handle that includes the given
-/// unpark event.
-///
-/// # Panics
-///
-/// This function will panic if a task is not currently being executed. That
-/// is, this method can be dangerous to call outside of an implementation of
-/// `poll`.
-pub fn with_unpark_event<F, R>(event: UnparkEvent, f: F) -> R
-    where F: FnOnce() -> R
-{
-    with(|task| {
-        let new_task = BorrowedTask {
-            id: task.id,
-            unpark: task.unpark,
-            events: task.events.with_event(event),
-            map: task.map,
-        };
-        set(&new_task, f)
-    })
-}
-
-#[derive(Clone)]
-/// A set insertion to trigger upon `unpark`.
-///
-/// Unpark events are used to communicate information about *why* an unpark
-/// occured, in particular populating sets with event identifiers so that the
-/// unparked task can avoid extraneous polling. See `with_unpark_event` for
-/// more.
-pub struct UnparkEvent {
-    set: Arc<EventSet>,
-    item: usize,
-}
-
-impl UnparkEvent {
-    /// Construct an unpark event that will insert `id` into `set` when
-    /// triggered.
-    pub fn new(set: Arc<EventSet>, id: usize) -> UnparkEvent {
-        UnparkEvent {
-            set: set,
-            item: id,
-        }
+    /// This function is intended as a performance optimization for structures
+    /// which store a `Task` internally.
+    ///
+    /// The purpose of this function is to answer the question "if I `notify`
+    /// this task is it equivalent to `task::current().notify()`". An answer
+    /// "yes" may mean that you don't actually need to call `task::current()`
+    /// and store it, but rather you can simply leave a stored task in place. An
+    /// answer of "no" typically means that you need to call `task::current()`
+    /// and store it somewhere.
+    ///
+    /// As this is purely a peformance optimization a valid implementation for
+    /// this function is to always return `false`. A best effort is done to
+    /// return `true` where possible, but false negatives may happen. Note that
+    /// this function will not return a false positive, however.
+    ///
+    /// # Panics
+    ///
+    /// This function will panic if no current future is being polled.
+    #[allow(deprecated)]
+    pub fn will_notify_current(&self) -> bool {
+        with(|current| {
+            self.unpark.will_notify(&current.unpark) &&
+                self.events.will_notify(&current.events)
+        })
     }
 }
 
-impl fmt::Debug for UnparkEvent {
+impl fmt::Debug for Task {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("UnparkEvent")
-         .field("set", &"...")
-         .field("item", &self.item)
+        f.debug_struct("Task")
          .finish()
     }
 }
 
-/// A concurrent set which allows for the insertion of `usize` values.
-///
-/// `EventSet`s are used to communicate precise information about the event(s)
-/// that trigged a task notification. See `task::with_unpark_event` for details.
-pub trait EventSet: Send + Sync + 'static {
-    /// Insert the given ID into the set
-    fn insert(&self, id: usize);
-}
-
-// A collection of UnparkEvents to trigger on `unpark`
-#[derive(Clone)]
-enum Events {
-    Zero,
-    One(UnparkEvent),
-    Lots(Vec<UnparkEvent>),
-}
-
-impl Events {
-    fn new() -> Events {
-        Events::Zero
-    }
-
-    fn trigger(&self) {
-        match *self {
-            Events::Zero => {}
-            Events::One(ref event) => event.set.insert(event.item),
-            Events::Lots(ref list) => {
-                for event in list {
-                    event.set.insert(event.item);
-                }
-            }
-        }
-    }
-
-    fn with_event(&self, event: UnparkEvent) -> Events {
-        let mut list = match *self {
-            Events::Zero => return Events::One(event),
-            Events::One(ref event) => vec![event.clone()],
-            Events::Lots(ref list) => list.clone(),
-        };
-        list.push(event);
-        Events::Lots(list)
-    }
-}
-
 /// Representation of a spawned future/stream.
 ///
 /// This object is returned by the `spawn` function in this module. This
@@ -280,9 +213,9 @@ impl Events {
 /// with either futures or streams, with different methods being available on
 /// `Spawn` depending which is used.
 pub struct Spawn<T> {
-    obj: T,
     id: usize,
-    data: data::LocalMap,
+    obj: T,
+    data: LocalMap,
 }
 
 /// Spawns a new future, returning the fused future and task.
@@ -296,9 +229,9 @@ pub struct Spawn<T> {
 /// until the methods on `Spawn` are called in turn.
 pub fn spawn<T>(obj: T) -> Spawn<T> {
     Spawn {
-        obj: obj,
         id: fresh_task_id(),
-        data: data::local_map(),
+        obj: obj,
+        data: local_map(),
     }
 }
 
@@ -321,167 +254,102 @@ impl<T> Spawn<T> {
 
 impl<F: Future> Spawn<F> {
     /// Polls the internal future, scheduling notifications to be sent to the
-    /// `unpark` argument.
+    /// `notify` argument.
     ///
     /// This method will poll the internal future, testing if it's completed
-    /// yet. The `unpark` argument is used as a sink for notifications sent to
+    /// yet. The `notify` argument is used as a sink for notifications sent to
     /// this future. That is, while the future is being polled, any call to
-    /// `task::park()` will return a handle that contains the `unpark`
+    /// `task::current()` will return a handle that contains the `notify`
     /// specified.
     ///
-    /// If this function returns `NotReady`, then the `unpark` should have been
+    /// If this function returns `NotReady`, 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.
-    pub fn poll_future(&mut self, unpark: Arc<Unpark>) -> Poll<F::Item, F::Error> {
-        self.enter(&unpark, |f| f.poll())
-    }
-
-    /// Waits for the internal future to complete, blocking this thread's
-    /// execution until it does.
-    ///
-    /// This function will call `poll_future` in a loop, waiting for the future
-    /// to complete. When a future cannot make progress it will use
-    /// `thread::park` to block the current thread.
-    pub fn wait_future(&mut self) -> Result<F::Item, F::Error> {
-        let unpark = Arc::new(ThreadUnpark::new(thread::current()));
-        loop {
-            match try!(self.poll_future(unpark.clone())) {
-                Async::NotReady => unpark.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.
+    /// Note that `notify` itself is passed as a shared reference, and is itself
+    /// not required to be a `NotifyHandle`. The `Clone` and `Into` trait bounds
+    /// will be used to convert this `notify` to a `NotifyHandle` if necessary.
+    /// This construction can avoid an unnecessary atomic reference count bump
+    /// in some situations.
     ///
-    /// 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.
+    /// ## Unsafety and `id`
     ///
-    /// This method is not appropriate for all futures, and other kinds of
-    /// executors typically provide a similar function with perhaps relaxed
-    /// bounds as well.
-    pub fn execute(self, exec: Arc<Executor>)
-        where F: Future<Item=(), Error=()> + Send + 'static,
+    /// This function and all other `*_notify` functions on this type will treat
+    /// the `id` specified very carefully, explicitly calling functions like the
+    /// `notify` argument's `clone_id` and `drop_id` functions. It should be
+    /// safe to encode a pointer itself into the `id` specified, such as an
+    /// `Arc<T>` or a `Box<T>`. The `clone_id` and `drop_id` functions are then
+    /// intended to be sufficient for the memory management related to that
+    /// pointer.
+    pub fn poll_future_notify<T>(&mut self,
+                                 notify: &T,
+                                 id: usize) -> Poll<F::Item, F::Error>
+        where T: Clone + Into<NotifyHandle>,
     {
-        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 {
-                id: self.id,
-                data: self.data,
-                obj: self.obj.boxed(),
-            },
-            inner: Arc::new(Inner {
-                exec: exec,
-                mutex: UnparkMutex::new()
-            }),
-        })
+        let mk = || notify.clone().into();
+        self.enter(BorrowedUnpark::new(&mk, id), |f| f.poll())
     }
 }
 
 impl<S: Stream> Spawn<S> {
-    /// Like `poll_future`, except polls the underlying stream.
-    pub fn poll_stream(&mut self, unpark: Arc<Unpark>)
-                       -> Poll<Option<S::Item>, S::Error> {
-        self.enter(&unpark, |stream| stream.poll())
-    }
-
-    /// Like `wait_future`, except only waits for the next element to arrive on
-    /// the underlying stream.
-    pub fn wait_stream(&mut self) -> Option<Result<S::Item, S::Error>> {
-        let unpark = Arc::new(ThreadUnpark::new(thread::current()));
-        loop {
-            match self.poll_stream(unpark.clone()) {
-                Ok(Async::NotReady) => unpark.park(),
-                Ok(Async::Ready(Some(e))) => return Some(Ok(e)),
-                Ok(Async::Ready(None)) => return None,
-                Err(e) => return Some(Err(e)),
-            }
-        }
+    /// Like `poll_future_notify`, except polls the underlying stream.
+    pub fn poll_stream_notify<T>(&mut self,
+                                 notify: &T,
+                                 id: usize)
+                                 -> Poll<Option<S::Item>, S::Error>
+        where T: Clone + Into<NotifyHandle>,
+    {
+        let mk = || notify.clone().into();
+        self.enter(BorrowedUnpark::new(&mk, id), |s| s.poll())
     }
 }
 
 impl<S: Sink> Spawn<S> {
     /// Invokes the underlying `start_send` method with this task in place.
     ///
-    /// If the underlying operation returns `NotReady` then the `unpark` value
+    /// 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(&mut self, value: S::SinkItem, unpark: &Arc<Unpark>)
-                       -> StartSend<S::SinkItem, S::SinkError> {
-        self.enter(unpark, |sink| sink.start_send(value))
+    pub fn start_send_notify<T>(&mut self,
+                                value: S::SinkItem,
+                                notify: &T,
+                                id: usize)
+                               -> StartSend<S::SinkItem, S::SinkError>
+        where T: Clone + Into<NotifyHandle>,
+    {
+        let mk = || notify.clone().into();
+        self.enter(BorrowedUnpark::new(&mk, id), |s| s.start_send(value))
     }
 
     /// Invokes the underlying `poll_complete` method with this task in place.
     ///
-    /// If the underlying operation returns `NotReady` then the `unpark` value
+    /// 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(&mut self, unpark: &Arc<Unpark>)
-                       -> Poll<(), S::SinkError> {
-        self.enter(unpark, |sink| sink.poll_complete())
-    }
-
-    /// Blocks the current thread until it's able to send `value` on this sink.
-    ///
-    /// This function will send the `value` on the sink that this task wraps. If
-    /// the sink is not ready to send the value yet then the current thread will
-    /// be blocked until it's able to send the value.
-    pub fn wait_send(&mut self, mut value: S::SinkItem)
-                     -> Result<(), S::SinkError> {
-        let unpark = Arc::new(ThreadUnpark::new(thread::current()));
-        let unpark2 = unpark.clone() as Arc<Unpark>;
-        loop {
-            value = match try!(self.start_send(value, &unpark2)) {
-                AsyncSink::NotReady(v) => v,
-                AsyncSink::Ready => return Ok(()),
-            };
-            unpark.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> {
-        let unpark = Arc::new(ThreadUnpark::new(thread::current()));
-        let unpark2 = unpark.clone() as Arc<Unpark>;
-        loop {
-            if try!(self.poll_flush(&unpark2)).is_ready() {
-                return Ok(())
-            }
-            unpark.park();
-        }
+    pub fn poll_flush_notify<T>(&mut self,
+                                notify: &T,
+                                id: usize)
+                                -> Poll<(), S::SinkError>
+        where T: Clone + Into<NotifyHandle>,
+    {
+        let mk = || notify.clone().into();
+        self.enter(BorrowedUnpark::new(&mk, id), |s| s.poll_complete())
     }
 }
 
 impl<T> Spawn<T> {
-    fn enter<F, R>(&mut self, unpark: &Arc<Unpark>, f: F) -> R
+    fn enter<F, R>(&mut self, unpark: BorrowedUnpark, f: F) -> R
         where F: FnOnce(&mut T) -> R
     {
-        let task = BorrowedTask {
+        let borrowed = BorrowedTask {
             id: self.id,
             unpark: unpark,
-            events: Events::new(),
+            events: BorrowedEvents::new(),
             map: &self.data,
         };
         let obj = &mut self.obj;
-        set(&task, || f(obj))
+        set(&borrowed, || f(obj))
     }
 }
 
@@ -489,7 +357,6 @@ impl<T: fmt::Debug> fmt::Debug for Spawn<T> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         f.debug_struct("Spawn")
          .field("obj", &self.obj)
-         .field("id", &self.id)
          .finish()
     }
 }
@@ -497,108 +364,286 @@ impl<T: fmt::Debug> fmt::Debug for Spawn<T> {
 /// A trait which represents a sink of notifications that a future is ready to
 /// make progress.
 ///
-/// This trait is provided as an argument to the `Spawn::poll_future` and
-/// `Spawn::poll_stream` functions. It's transitively used as part of the
-/// `Task::unpark` method to internally deliver notifications of readiness of a
-/// future to move forward.
-pub trait Unpark: Send + Sync {
+/// This trait is provided as an argument to the `Spawn::*_notify` family of
+/// functions. It's transitively used as part of the `Task::notify` method to
+/// internally deliver notifications of readiness of a future to move forward.
+///
+/// An instance of `Notify` has one primary method, `notify`, which is given a
+/// contextual argument as to what's being notified. This contextual argument is
+/// *also* provided to the `Spawn::*_notify` family of functions and can be used
+/// to reuse an instance of `Notify` across many futures.
+///
+/// Instances of `Notify` must be safe to share across threads, and the methods
+/// be invoked concurrently. They must also live for the `'static` lifetime,
+/// not containing any stack references.
+pub trait Notify: Send + Sync {
     /// Indicates that an associated future and/or task are ready to make
     /// progress.
     ///
     /// Typically this means that the receiver of the notification should
     /// arrange for the future to get poll'd in a prompt fashion.
-    fn unpark(&self);
+    ///
+    /// This method takes an `id` as an argument which was transitively passed
+    /// in from the original call to `Spawn::*_notify`. This id can be used to
+    /// disambiguate which precise future became ready for polling.
+    fn notify(&self, id: usize);
+
+    /// This function is called whenever a new copy of `id` is needed.
+    ///
+    /// This is called in one of two situations:
+    ///
+    /// * A `Task` is being created through `task::current` while a future is
+    ///   being polled. In that case the instance of `Notify` passed in to one
+    ///   of the `poll_*` functions is called with the `id` passed into the same
+    ///   `poll_*` function.
+    /// * A `Task` is itself being cloned. Each `Task` contains its own id and a
+    ///   handle to the `Notify` behind it, and the task's `Notify` is used to
+    ///   clone the internal `id` to assign to the new task.
+    ///
+    /// The `id` returned here will be stored in the `Task`-to-be and used later
+    /// to pass to `notify` when the `Task::notify` function is called on that
+    /// `Task`.
+    ///
+    /// Note that typically this is just the identity function, passing through
+    /// the identifier. For more unsafe situations, however, if `id` is itself a
+    /// pointer of some kind this can be used as a hook to "clone" the pointer,
+    /// depending on what that means for the specified pointer.
+    fn clone_id(&self, id: usize) -> usize {
+        id
+    }
+
+    /// All instances of `Task` store an `id` that they're going to internally
+    /// notify with, and this function is called when the `Task` is dropped.
+    ///
+    /// This function provides a hook for schemes which encode pointers in this
+    /// `id` argument to deallocate resources associated with the pointer. It's
+    /// guaranteed that after this function is called the `Task` containing this
+    /// `id` will no longer use the `id`.
+    fn drop_id(&self, id: usize) {
+        drop(id);
+    }
+}
+
+pub fn with_notify<F, T, R>(notify: &T, id: usize, f: F) -> R
+    where F: FnOnce() -> R,
+          T: Clone + Into<NotifyHandle>,
+{
+    with(|task| {
+        let mk = || notify.clone().into();
+        let new_task = BorrowedTask {
+            id: task.id,
+            unpark: BorrowedUnpark::new(&mk, id),
+            events: task.events,
+            map: task.map,
+        };
+
+        set(&new_task, f)
+    })
 }
 
-/// A trait representing requests to poll futures.
+/// An unsafe trait for implementing custom forms of memory management behind a
+/// `Task`.
+///
+/// The `futures` critically relies on "notification handles" to extract for
+/// futures to contain and then later inform that they're ready to make
+/// progress. These handles, however, must be cheap to create and cheap
+/// to clone to ensure that this operation is efficient throughout the
+/// execution of a program.
+///
+/// Typically this sort of memory management is done in the standard library
+/// with the `Arc` type. An `Arc` is relatively cheap to allocate an is
+/// quite cheap to clone and pass around. Plus, it's 100% safe!
+///
+/// When working outside the standard library, however, you don't always have
+/// and `Arc` type available to you. This trait, `UnsafeNotify`, is intended
+/// to be the "unsafe version" of the `Notify` trait. This trait encodes the
+/// memory management operations of a `Task`'s notification handle, allowing
+/// custom implementations for the memory management of a notification handle.
+///
+/// Put another way, the core notification type in this library,
+/// `NotifyHandle`, simply internally contains an instance of
+/// `*mut UnsafeNotify`. This "unsafe trait object" is then used exclusively
+/// to operate with, dynamically dispatching calls to clone, drop, and notify.
+/// Critically though as a raw pointer it doesn't require a particular form
+/// of memory management, allowing external implementations.
 ///
-/// 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.
-pub trait Executor: Send + Sync + 'static {
-    /// Requests that `Run` is executed soon on the given executor.
-    fn execute(&self, r: Run);
+/// A default implementation of the `UnsafeNotify` trait is provided for the
+/// `Arc` type in the standard library. If the `use_std` feature of this crate
+/// is not available however, you'll be required to implement your own
+/// instance of this trait to pass it into `NotifyHandle::new`.
+///
+/// # Unsafety
+///
+/// This trait is manually encoding the memory management of the underlying
+/// handle, and as a result is quite unsafe to implement! Implementors of
+/// this trait must guarantee:
+///
+/// * Calls to `clone_raw` produce uniquely owned handles. It should be safe
+///   to drop the current handle and have the returned handle still be valid.
+/// * Calls to `drop_raw` work with `self` as a raw pointer, deallocating
+///   resources associated with it. This is a pretty unsafe operation as it's
+///   invalidating the `self` pointer, so extreme care needs to be taken.
+///
+/// In general it's recommended to review the trait documentation as well as
+/// the implementation for `Arc` in this crate. When in doubt ping the
+/// `futures` authors to clarify an unsafety question here.
+pub unsafe trait UnsafeNotify: Notify {
+    /// Creates a new `NotifyHandle` from this instance of `UnsafeNotify`.
+    ///
+    /// This function will create a new uniquely owned handle that under the
+    /// hood references the same notification instance. In other words calls
+    /// to `notify` on the returned handle should be equivalent to calls to
+    /// `notify` on this handle.
+    ///
+    /// # Unsafety
+    ///
+    /// This trait is unsafe to implement, as are all these methods. This
+    /// method is also unsafe to call as it's asserting the `UnsafeNotify`
+    /// value is in a consistent state. In general it's recommended to
+    /// review the trait documentation as well as the implementation for `Arc`
+    /// in this crate. When in doubt ping the `futures` authors to clarify
+    /// an unsafety question here.
+    unsafe fn clone_raw(&self) -> NotifyHandle;
+
+    /// Drops this instance of `UnsafeNotify`, deallocating resources
+    /// associated with it.
+    ///
+    /// This method is intended to have a signature such as:
+    ///
+    /// ```ignore
+    /// fn drop_raw(self: *mut Self);
+    /// ```
+    ///
+    /// Unfortunately in Rust today that signature is not object safe.
+    /// Nevertheless it's recommended to implement this function *as if* that
+    /// were its signature. As such it is not safe to call on an invalid
+    /// pointer, nor is the validity of the pointer guaranteed after this
+    /// function returns.
+    ///
+    /// # Unsafety
+    ///
+    /// This trait is unsafe to implement, as are all these methods. This
+    /// method is also unsafe to call as it's asserting the `UnsafeNotify`
+    /// value is in a consistent state. In general it's recommended to
+    /// review the trait documentation as well as the implementation for `Arc`
+    /// in this crate. When in doubt ping the `futures` authors to clarify
+    /// an unsafety question here.
+    unsafe fn drop_raw(&self);
 }
 
-struct ThreadUnpark {
-    thread: thread::Thread,
-    ready: AtomicBool,
+/// A `NotifyHandle` is the core value through which notifications are routed
+/// in the `futures` crate.
+///
+/// All instances of `Task` will contain a `NotifyHandle` handle internally.
+/// This handle itself contains a trait object pointing to an instance of the
+/// `Notify` trait, allowing notifications to get routed through it.
+///
+/// The `NotifyHandle` type internally does not codify any particular memory
+/// management strategy. Internally it contains an instance of `*mut
+/// UnsafeNotify`, and more details about that trait can be found on its own
+/// documentation. Consequently, though, the one constructor of this type,
+/// `NotifyHandle::new`, is `unsafe` to call. It is not recommended to call
+/// this constructor directly.
+///
+/// If you're working with the standard library then it's recommended to
+/// work with the `Arc` type. If you have a struct, `T`, which implements the
+/// `Notify` trait, then you can construct this with
+/// `NotifyHandle::from(t: Arc<T>)`. The coercion to `UnsafeNotify` will
+/// happen automatically and safely for you.
+///
+/// When working externally from the standard library it's recommended to
+/// provide a similar safe constructor for your custom type as opposed to
+/// recommending an invocation of `NotifyHandle::new` directly.
+pub struct NotifyHandle {
+    inner: *mut UnsafeNotify,
 }
 
-impl ThreadUnpark {
-    fn new(thread: thread::Thread) -> ThreadUnpark {
-        ThreadUnpark {
-            thread: thread,
-            ready: AtomicBool::new(false),
-        }
-    }
+unsafe impl Send for NotifyHandle {}
+unsafe impl Sync for NotifyHandle {}
 
-    fn park(&self) {
-        if !self.ready.swap(false, Ordering::SeqCst) {
-            thread::park();
-        }
+impl NotifyHandle {
+    /// Constructs a new `NotifyHandle` directly.
+    ///
+    /// Note that most code will not need to call this. Implementors of the
+    /// `UnsafeNotify` trait will typically provide a wrapper that calls this
+    /// but you otherwise shouldn't call it directly.
+    ///
+    /// If you're working with the standard library then it's recommended to
+    /// use the `NotifyHandle::from` function instead which works with the safe
+    /// `Arc` type and the safe `Notify` trait.
+    #[inline]
+    pub unsafe fn new(inner: *mut UnsafeNotify) -> NotifyHandle {
+        NotifyHandle { inner: inner }
     }
-}
 
-impl Unpark for ThreadUnpark {
-    fn unpark(&self) {
-        self.ready.store(true, Ordering::SeqCst);
-        self.thread.unpark()
+    /// Invokes the underlying instance of `Notify` with the provided `id`.
+    pub fn notify(&self, id: usize) {
+        unsafe { (*self.inner).notify(id) }
     }
-}
 
-/// Units of work submitted to an `Executor`, currently only created
-/// internally.
-pub struct Run {
-    spawn: Spawn<BoxFuture<(), ()>>,
-    inner: Arc<Inner>,
-}
+    fn clone_id(&self, id: usize) -> usize {
+        unsafe { (*self.inner).clone_id(id) }
+    }
 
-struct Inner {
-    mutex: UnparkMutex<Run>,
-    exec: Arc<Executor>,
+    fn drop_id(&self, id: usize) {
+        unsafe { (*self.inner).drop_id(id) }
+    }
 }
 
-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.
+impl Clone for NotifyHandle {
+    #[inline]
+    fn clone(&self) -> Self {
         unsafe {
-            inner.mutex.start_poll();
-
-            loop {
-                match spawn.poll_future(inner.clone()) {
-                    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
-                }
-            }
+            (*self.inner).clone_raw()
         }
     }
 }
 
-impl fmt::Debug for Run {
+impl fmt::Debug for NotifyHandle {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct("Run")
-         .field("contents", &"...")
+        f.debug_struct("NotifyHandle")
          .finish()
     }
 }
 
-impl Unpark for Inner {
-    fn unpark(&self) {
-        match self.mutex.notify() {
-            Ok(run) => self.exec.execute(run),
-            Err(()) => {}
+impl Drop for NotifyHandle {
+    fn drop(&mut self) {
+        unsafe {
+            (*self.inner).drop_raw()
         }
     }
 }
+
+/// Marker for a `T` that is behind &'static.
+struct StaticRef<T>(PhantomData<T>);
+
+impl<T: Notify> Notify for StaticRef<T> {
+    fn notify(&self, id: usize) {
+        let me = unsafe { &*(self as *const _ as *const T) };
+        me.notify(id);
+    }
+
+    fn clone_id(&self, id: usize) -> usize {
+        let me = unsafe { &*(self as *const _ as *const T) };
+        me.clone_id(id)
+    }
+
+    fn drop_id(&self, id: usize) {
+        let me = unsafe { &*(self as *const _ as *const T) };
+        me.drop_id(id);
+    }
+}
+
+unsafe impl<T: Notify + 'static> UnsafeNotify for StaticRef<T> {
+    unsafe fn clone_raw(&self) -> NotifyHandle {
+        NotifyHandle::new(self as *const _ as *mut StaticRef<T>)
+    }
+
+    unsafe fn drop_raw(&self) {}
+}
+
+impl<T: Notify> From<&'static T> for NotifyHandle {
+    fn from(src : &'static T) -> NotifyHandle {
+        unsafe { NotifyHandle::new(src as *const _ as *mut StaticRef<T>) }
+    }
+}
diff --git a/src/task_impl/data.rs b/src/task_impl/std/data.rs
similarity index 98%
rename from src/task_impl/data.rs
rename to src/task_impl/std/data.rs
index f7076e4b..770912b2 100644
--- a/src/task_impl/data.rs
+++ b/src/task_impl/std/data.rs
@@ -5,6 +5,8 @@ use std::cell::RefCell;
 use std::hash::{BuildHasherDefault, Hasher};
 use std::collections::HashMap;
 
+use task_impl::with;
+
 /// A macro to create a `static` of type `LocalKey`
 ///
 /// This macro is intentionally similar to the `thread_local!`, and creates a
@@ -113,7 +115,7 @@ impl<T: Send + 'static> LocalKey<T> {
         where F: FnOnce(&T) -> R
     {
         let key = (self.__key)();
-        super::with(|task| {
+        with(|task| {
             let raw_pointer = {
                 let mut data = task.map.borrow_mut();
                 let entry = data.entry(key).or_insert_with(|| {
diff --git a/src/task_impl/std/mod.rs b/src/task_impl/std/mod.rs
new file mode 100644
index 00000000..2e65c008
--- /dev/null
+++ b/src/task_impl/std/mod.rs
@@ -0,0 +1,653 @@
+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, Once, ONCE_INIT};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::thread;
+
+use {Future, Stream, Sink, Poll, Async, StartSend, AsyncSink};
+use future::BoxFuture;
+use super::core;
+use super::{BorrowedTask, NotifyHandle, Spawn, spawn, Notify, UnsafeNotify};
+
+mod unpark_mutex;
+pub use self::unpark_mutex::UnparkMutex;
+
+mod data;
+pub use self::data::*;
+
+mod task_rc;
+#[allow(deprecated)]
+#[cfg(feature = "with-deprecated")]
+pub use self::task_rc::TaskRc;
+
+pub use task_impl::core::init;
+
+thread_local!(static CURRENT_TASK: Cell<*mut u8> = Cell::new(ptr::null_mut()));
+
+// This function and the one below are never used, but they are used to pass in
+// as arguments to the task system initialization. However, thread local logic
+// is special cased before, and `tls_get_ptr` is used as a token to check that
+// TLS is being used to track the current task.
+//
+// This function should never be inlined as it is used as a token in the
+// conditional below.
+#[inline(never)]
+fn tls_get_ptr() -> *mut u8 {
+    unreachable!();
+}
+
+#[inline(never)]
+fn tls_set_ptr(_: *mut u8) {
+    unreachable!();
+}
+
+static INIT: Once = ONCE_INIT;
+
+pub fn get_ptr() -> *mut u8 {
+    // Since this condition will always return true when TLS task storage is
+    // used (the default), the branch predictor will be able to optimize the
+    // branching and a dynamic dispatch will be avoided, which makes the
+    // compiler happier.
+    if core::is_get_ptr(tls_get_ptr as usize) {
+        CURRENT_TASK.with(|c| c.get())
+    } else {
+        core::get_ptr()
+    }
+}
+
+fn tls_slot() -> *const Cell<*mut u8> {
+    CURRENT_TASK.with(|c| c as *const _)
+}
+
+pub fn set<'a, F, R>(task: &BorrowedTask<'a>, f: F) -> R
+    where F: FnOnce() -> R
+{
+    // Lazily initialze the get / set ptrs
+    INIT.call_once(|| unsafe { init(tls_get_ptr, tls_set_ptr); });
+
+    // Same as above.
+    if core::is_get_ptr(tls_get_ptr as usize) {
+        struct Reset(*const Cell<*mut u8>, *mut u8);
+
+        impl Drop for Reset {
+            #[inline]
+            fn drop(&mut self) {
+                unsafe {
+                    (*self.0).set(self.1);
+                }
+            }
+        }
+
+        unsafe {
+            let slot = tls_slot();
+            let _reset = Reset(slot, (*slot).get());
+            (*slot).set(task as *const _ as *mut u8);
+            f()
+        }
+    } else {
+        core::set(task, f)
+    }
+}
+
+#[derive(Copy, Clone)]
+#[allow(deprecated)]
+pub enum BorrowedUnpark<'a> {
+    Old(&'a Arc<Unpark>),
+    New(core::BorrowedUnpark<'a>),
+}
+
+#[derive(Copy, Clone)]
+#[allow(deprecated)]
+pub enum BorrowedEvents<'a> {
+    None,
+    One(&'a UnparkEvent, &'a BorrowedEvents<'a>),
+}
+
+#[derive(Clone)]
+pub enum TaskUnpark {
+    #[allow(deprecated)]
+    Old(Arc<Unpark>),
+    New(core::TaskUnpark),
+}
+
+#[derive(Clone)]
+#[allow(deprecated)]
+pub enum UnparkEvents {
+    None,
+    One(UnparkEvent),
+    Many(Box<[UnparkEvent]>),
+}
+
+impl<'a> BorrowedUnpark<'a> {
+    #[inline]
+    pub fn new(f: &'a Fn() -> NotifyHandle, id: usize) -> BorrowedUnpark<'a> {
+        BorrowedUnpark::New(core::BorrowedUnpark::new(f, id))
+    }
+
+    #[inline]
+    pub fn to_owned(&self) -> TaskUnpark {
+        match *self {
+            BorrowedUnpark::Old(old) => TaskUnpark::Old(old.clone()),
+            BorrowedUnpark::New(new) => TaskUnpark::New(new.to_owned()),
+        }
+    }
+}
+
+impl<'a> BorrowedEvents<'a> {
+    #[inline]
+    pub fn new() -> BorrowedEvents<'a> {
+        BorrowedEvents::None
+    }
+
+    #[inline]
+    pub fn to_owned(&self) -> UnparkEvents {
+        let mut one_event = None;
+        let mut list = Vec::new();
+        let mut cur = self;
+        while let BorrowedEvents::One(event, next) = *cur {
+            let event = event.clone();
+            match one_event.take() {
+                None if list.len() == 0 => one_event = Some(event),
+                None => list.push(event),
+                Some(event2) =>  {
+                    list.push(event2);
+                    list.push(event);
+                }
+            }
+            cur = next;
+        }
+
+        match one_event {
+            None if list.len() == 0 => UnparkEvents::None,
+            None => UnparkEvents::Many(list.into_boxed_slice()),
+            Some(e) => UnparkEvents::One(e),
+        }
+    }
+}
+
+impl UnparkEvents {
+    pub fn notify(&self) {
+        match *self {
+            UnparkEvents::None => {}
+            UnparkEvents::One(ref e) => e.unpark(),
+            UnparkEvents::Many(ref list) => {
+                for event in list.iter() {
+                    event.unpark();
+                }
+            }
+        }
+    }
+
+    pub fn will_notify(&self, events: &BorrowedEvents) -> bool {
+        // Pessimistically assume that any unpark events mean that we're not
+        // equivalent to the current task.
+        match *self {
+            UnparkEvents::None => {}
+            _ => return false,
+        }
+
+        match *events {
+            BorrowedEvents::None => return true,
+            _ => {},
+        }
+
+        return false
+    }
+}
+
+#[allow(deprecated)]
+impl TaskUnpark {
+    pub fn notify(&self) {
+        match *self {
+            TaskUnpark::Old(ref old) => old.unpark(),
+            TaskUnpark::New(ref new) => new.notify(),
+        }
+    }
+
+    pub fn will_notify(&self, unpark: &BorrowedUnpark) -> bool {
+        match (unpark, self) {
+            (&BorrowedUnpark::Old(old1), &TaskUnpark::Old(ref old2)) => {
+                &**old1 as *const Unpark == &**old2 as *const Unpark
+            }
+            (&BorrowedUnpark::New(ref new1), &TaskUnpark::New(ref new2)) => {
+                new2.will_notify(new1)
+            }
+            _ => false,
+        }
+    }
+}
+
+impl<F: Future> Spawn<F> {
+    /// Polls the internal future, scheduling notifications to be sent to the
+    /// `unpark` argument.
+    ///
+    /// This method will poll the internal future, testing if it's completed
+    /// yet. The `unpark` argument is used as a sink for notifications sent to
+    /// this future. That is, while the future is being polled, any call to
+    /// `task::park()` will return a handle that contains the `unpark`
+    /// specified.
+    ///
+    /// If this function returns `NotReady`, then the `unpark` 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.
+    #[deprecated(note = "recommended to use `poll_future_notify` instead")]
+    #[allow(deprecated)]
+    pub fn poll_future(&mut self, unpark: Arc<Unpark>) -> Poll<F::Item, F::Error> {
+        self.enter(BorrowedUnpark::Old(&unpark), |f| f.poll())
+    }
+
+    /// Waits for the internal future to complete, blocking this thread's
+    /// execution until it does.
+    ///
+    /// This function will call `poll_future` in a loop, waiting for the future
+    /// to complete. When a future cannot make progress it will use
+    /// `thread::park` to block the current thread.
+    pub fn wait_future(&mut self) -> Result<F::Item, F::Error> {
+        let unpark = Arc::new(ThreadUnpark::new(thread::current()));
+
+        loop {
+            match try!(self.poll_future_notify(&unpark, 0)) {
+                Async::NotReady => unpark.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.
+    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(self.into_inner().boxed()),
+            inner: Arc::new(RunInner {
+                exec: exec,
+                mutex: UnparkMutex::new()
+            }),
+        })
+    }
+}
+
+impl<S: Stream> Spawn<S> {
+    /// Like `poll_future`, except polls the underlying stream.
+    #[deprecated(note = "recommended to use `poll_stream_notify` instead")]
+    #[allow(deprecated)]
+    pub fn poll_stream(&mut self, unpark: Arc<Unpark>)
+                       -> Poll<Option<S::Item>, S::Error> {
+        self.enter(BorrowedUnpark::Old(&unpark), |s| s.poll())
+    }
+
+    /// Like `wait_future`, except only waits for the next element to arrive on
+    /// the underlying stream.
+    pub fn wait_stream(&mut self) -> Option<Result<S::Item, S::Error>> {
+        let unpark = Arc::new(ThreadUnpark::new(thread::current()));
+        loop {
+            match self.poll_stream_notify(&unpark, 0) {
+                Ok(Async::NotReady) => unpark.park(),
+                Ok(Async::Ready(Some(e))) => return Some(Ok(e)),
+                Ok(Async::Ready(None)) => return None,
+                Err(e) => return Some(Err(e)),
+            }
+        }
+    }
+}
+
+impl<S: Sink> Spawn<S> {
+    /// Invokes the underlying `start_send` method with this task in place.
+    ///
+    /// If the underlying operation returns `NotReady` then the `unpark` value
+    /// passed in will receive a notification when the operation is ready to be
+    /// attempted again.
+    #[deprecated(note = "recommended to use `start_send_notify` instead")]
+    #[allow(deprecated)]
+    pub fn start_send(&mut self, value: S::SinkItem, unpark: &Arc<Unpark>)
+                       -> StartSend<S::SinkItem, S::SinkError> {
+        self.enter(BorrowedUnpark::Old(unpark), |s| s.start_send(value))
+    }
+
+    /// Invokes the underlying `poll_complete` method with this task in place.
+    ///
+    /// If the underlying operation returns `NotReady` then the `unpark` value
+    /// passed in will receive a notification when the operation is ready to be
+    /// attempted again.
+    #[deprecated(note = "recommended to use `poll_flush_notify` instead")]
+    #[allow(deprecated)]
+    pub fn poll_flush(&mut self, unpark: &Arc<Unpark>)
+                       -> Poll<(), S::SinkError> {
+        self.enter(BorrowedUnpark::Old(unpark), |s| s.poll_complete())
+    }
+
+    /// Blocks the current thread until it's able to send `value` on this sink.
+    ///
+    /// This function will send the `value` on the sink that this task wraps. If
+    /// the sink is not ready to send the value yet then the current thread will
+    /// be blocked until it's able to send the value.
+    pub fn wait_send(&mut self, mut value: S::SinkItem)
+                     -> Result<(), S::SinkError> {
+        let notify = Arc::new(ThreadUnpark::new(thread::current()));
+        loop {
+            value = match try!(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> {
+        let notify = Arc::new(ThreadUnpark::new(thread::current()));
+        loop {
+            if try!(self.poll_flush_notify(&notify, 0)).is_ready() {
+                return Ok(())
+            }
+            notify.park();
+        }
+    }
+}
+
+/// A trait which represents a sink of notifications that a future is ready to
+/// make progress.
+///
+/// This trait is provided as an argument to the `Spawn::poll_future` and
+/// `Spawn::poll_stream` functions. It's transitively used as part of the
+/// `Task::unpark` method to internally deliver notifications of readiness of a
+/// future to move forward.
+#[deprecated(note = "recommended to use `Notify` instead")]
+pub trait Unpark: Send + Sync {
+    /// Indicates that an associated future and/or task are ready to make
+    /// progress.
+    ///
+    /// Typically this means that the receiver of the notification should
+    /// arrange for the future to get poll'd in a prompt fashion.
+    fn unpark(&self);
+}
+
+/// 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.
+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<BoxFuture<(), ()>>,
+    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.
+    #[allow(deprecated)]
+    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 {
+                let unpark: Arc<Unpark> = inner.clone();
+                match spawn.poll_future(unpark) {
+                    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()
+    }
+}
+
+#[allow(deprecated)]
+impl Unpark for RunInner {
+    fn unpark(&self) {
+        match self.mutex.notify() {
+            Ok(run) => self.exec.execute(run),
+            Err(()) => {}
+        }
+    }
+}
+
+// ===== ThreadUnpark =====
+
+struct ThreadUnpark {
+    thread: thread::Thread,
+    ready: AtomicBool,
+}
+
+impl ThreadUnpark {
+    fn new(thread: thread::Thread) -> ThreadUnpark {
+        ThreadUnpark {
+            thread: thread,
+            ready: AtomicBool::new(false),
+        }
+    }
+
+    fn park(&self) {
+        if !self.ready.swap(false, Ordering::SeqCst) {
+            thread::park();
+        }
+    }
+}
+
+impl Notify for ThreadUnpark {
+    fn notify(&self, _unpark_id: usize) {
+        self.ready.store(true, Ordering::SeqCst);
+        self.thread.unpark()
+    }
+}
+
+// ===== UnparkEvent =====
+
+/// For the duration of the given callback, add an "unpark event" to be
+/// triggered when the task handle is used to unpark the task.
+///
+/// Unpark events are used to pass information about what event caused a task to
+/// be unparked. In some cases, tasks are waiting on a large number of possible
+/// events, and need precise information about the wakeup to avoid extraneous
+/// polling.
+///
+/// Every `Task` handle comes with a set of unpark events which will fire when
+/// `unpark` is called. When fired, these events insert an identifer into a
+/// concurrent set, which the task can read from to determine what events
+/// occurred.
+///
+/// This function immediately invokes the closure, `f`, but arranges things so
+/// that `task::park` will produce a `Task` handle that includes the given
+/// unpark event.
+///
+/// # Panics
+///
+/// This function will panic if a task is not currently being executed. That
+/// is, this method can be dangerous to call outside of an implementation of
+/// `poll`.
+#[deprecated(note = "recommended to use `FuturesUnordered` instead")]
+#[allow(deprecated)]
+pub fn with_unpark_event<F, R>(event: UnparkEvent, f: F) -> R
+    where F: FnOnce() -> R
+{
+    super::with(|task| {
+        let new_task = BorrowedTask {
+            id: task.id,
+            unpark: task.unpark,
+            events: BorrowedEvents::One(&event, &task.events),
+            map: task.map,
+        };
+
+        super::set(&new_task, f)
+    })
+}
+
+/// A set insertion to trigger upon `unpark`.
+///
+/// Unpark events are used to communicate information about *why* an unpark
+/// occured, in particular populating sets with event identifiers so that the
+/// unparked task can avoid extraneous polling. See `with_unpark_event` for
+/// more.
+#[derive(Clone)]
+#[deprecated(note = "recommended to use `FuturesUnordered` instead")]
+#[allow(deprecated)]
+pub struct UnparkEvent {
+    set: Arc<EventSet>,
+    item: usize,
+}
+
+#[allow(deprecated)]
+impl UnparkEvent {
+    /// Construct an unpark event that will insert `id` into `set` when
+    /// triggered.
+    #[deprecated(note = "recommended to use `FuturesUnordered` instead")]
+    pub fn new(set: Arc<EventSet>, id: usize) -> UnparkEvent {
+        UnparkEvent {
+            set: set,
+            item: id,
+        }
+    }
+
+    fn unpark(&self) {
+        self.set.insert(self.item);
+    }
+}
+
+#[allow(deprecated)]
+impl fmt::Debug for UnparkEvent {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_struct("UnparkEvent")
+         .field("set", &"...")
+         .field("item", &self.item)
+         .finish()
+    }
+}
+
+/// A concurrent set which allows for the insertion of `usize` values.
+///
+/// `EventSet`s are used to communicate precise information about the event(s)
+/// that trigged a task notification. See `task::with_unpark_event` for details.
+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
+// these two values are considered equivalent:
+//
+// * Arc<T>
+// * *const ArcWrapped<T>
+//
+// We don't actually know the layout of `ArcWrapped<T>` as it's an
+// implementation detail in the standard library. We can work, though, by
+// casting it through and back an `Arc<T>`.
+//
+// This also means that you wn't actually fine `UnsafeNotify for Arc<T>`
+// because it's the wrong level of indirection. These methods are sort of
+// receiving Arc<T>, but not an owned version. It's... complicated. We may be
+// one of the first users of unsafe trait objects!
+
+struct ArcWrapped<T>(PhantomData<T>);
+
+impl<T: Notify + 'static> Notify for ArcWrapped<T> {
+    fn notify(&self, id: usize) {
+        unsafe {
+            let me: *const ArcWrapped<T> = self;
+            T::notify(&*(&me as *const *const ArcWrapped<T> as *const Arc<T>),
+                      id)
+        }
+    }
+
+    fn clone_id(&self, id: usize) -> usize {
+        unsafe {
+            let me: *const ArcWrapped<T> = self;
+            T::clone_id(&*(&me as *const *const ArcWrapped<T> as *const Arc<T>),
+                        id)
+        }
+    }
+
+    fn drop_id(&self, id: usize) {
+        unsafe {
+            let me: *const ArcWrapped<T> = self;
+            T::drop_id(&*(&me as *const *const ArcWrapped<T> as *const Arc<T>),
+                       id)
+        }
+    }
+}
+
+unsafe impl<T: Notify + 'static> UnsafeNotify for ArcWrapped<T> {
+    unsafe fn clone_raw(&self) -> NotifyHandle {
+        let me: *const ArcWrapped<T> = self;
+        let arc = (*(&me as *const *const ArcWrapped<T> as *const Arc<T>)).clone();
+        NotifyHandle::from(arc)
+    }
+
+    unsafe fn drop_raw(&self) {
+        let mut me: *const ArcWrapped<T> = self;
+        let me = &mut me as *mut *const ArcWrapped<T> as *mut Arc<T>;
+        ptr::drop_in_place(me);
+    }
+}
+
+impl<T> From<Arc<T>> for NotifyHandle
+    where T: Notify + 'static,
+{
+    fn from(rc: Arc<T>) -> NotifyHandle {
+        unsafe {
+            let ptr = mem::transmute::<Arc<T>, *mut ArcWrapped<T>>(rc);
+            NotifyHandle::new(ptr)
+        }
+    }
+}
diff --git a/src/task_impl/task_rc.rs b/src/task_impl/std/task_rc.rs
similarity index 91%
rename from src/task_impl/task_rc.rs
rename to src/task_impl/std/task_rc.rs
index b942f26a..64d5eb52 100644
--- a/src/task_impl/task_rc.rs
+++ b/src/task_impl/std/task_rc.rs
@@ -7,6 +7,7 @@
 use std::prelude::v1::*;
 use std::sync::Arc;
 use std::cell::UnsafeCell;
+use task_impl;
 
 // One critical piece of this module's contents are the `TaskRc<A>` handles.
 // The purpose of this is to conceptually be able to store data in a task,
@@ -63,7 +64,7 @@ use std::cell::UnsafeCell;
 /// change over time, if the task migrates, so `A` must be `Send`.
 #[derive(Debug)]
 pub struct TaskRc<A> {
-    task_id: usize,
+    task: task_impl::Task,
     ptr: Arc<UnsafeCell<A>>,
 }
 
@@ -89,12 +90,10 @@ impl<A> TaskRc<A> {
     ///
     /// This function will panic if a task is not currently running.
     pub fn new(a: A) -> TaskRc<A> {
-        super::with(|task| {
-            TaskRc {
-                task_id: task.id,
-                ptr: Arc::new(UnsafeCell::new(a)),
-            }
-        })
+        TaskRc {
+            task: task_impl::park(),
+            ptr: Arc::new(UnsafeCell::new(a)),
+        }
     }
 
     /// Operate with a reference to the underlying data.
@@ -112,19 +111,18 @@ impl<A> TaskRc<A> {
     pub fn with<F, R>(&self, f: F) -> R
         where F: FnOnce(&A) -> R
     {
-        // for safety here, see docs at the top of this module
-        super::with(|task| {
-            assert!(self.task_id == task.id,
-                    "TaskRc being accessed on task it does not belong to");
-            f(unsafe { &*self.ptr.get() })
-        })
+        if !self.task.is_current() {
+            panic!("TaskRc being accessed on task it does not belong to");
+        }
+
+        f(unsafe { &*self.ptr.get() })
     }
 }
 
 impl<A> Clone for TaskRc<A> {
     fn clone(&self) -> TaskRc<A> {
         TaskRc {
-            task_id: self.task_id,
+            task: self.task.clone(),
             ptr: self.ptr.clone(),
         }
     }
diff --git a/src/task_impl/unpark_mutex.rs b/src/task_impl/std/unpark_mutex.rs
similarity index 100%
rename from src/task_impl/unpark_mutex.rs
rename to src/task_impl/std/unpark_mutex.rs
diff --git a/src/unsync/mpsc.rs b/src/unsync/mpsc.rs
index 2a474438..d4c38211 100644
--- a/src/unsync/mpsc.rs
+++ b/src/unsync/mpsc.rs
@@ -66,14 +66,14 @@ impl<T> Sender<T> {
 
         match shared.capacity {
             Some(capacity) if shared.buffer.len() == capacity => {
-                shared.blocked_senders.push_back(task::park());
+                shared.blocked_senders.push_back(task::current());
                 Ok(AsyncSink::NotReady(msg))
             }
             _ => {
                 shared.buffer.push_back(msg);
                 if let Some(task) = shared.blocked_recv.take() {
                     drop(shared);
-                    task.unpark();
+                    task.notify();
                 }
                 Ok(AsyncSink::Ready)
             }
@@ -120,7 +120,7 @@ impl<T> Drop for Sender<T> {
             if let Some(task) = shared.blocked_recv.take() {
                 // Wake up receiver as its stream has ended
                 drop(shared);
-                task.unpark();
+                task.notify();
             }
         }
     }
@@ -159,7 +159,7 @@ impl<T> Receiver<T> {
         };
         self.state = State::Closed(items);
         for task in blockers {
-            task.unpark();
+            task.notify();
         }
     }
 }
@@ -186,11 +186,11 @@ impl<T> Stream for Receiver<T> {
         if let Some(msg) = shared.buffer.pop_front() {
             if let Some(task) = shared.blocked_senders.pop_front() {
                 drop(shared);
-                task.unpark();
+                task.notify();
             }
             Ok(Async::Ready(Some(msg)))
         } else {
-            shared.blocked_recv = Some(task::park());
+            shared.blocked_recv = Some(task::current());
             Ok(Async::NotReady)
         }
     }
@@ -261,7 +261,7 @@ impl<T> UnboundedSender<T> {
         shared.buffer.push_back(msg);
         if let Some(task) = shared.blocked_recv.take() {
             drop(shared);
-            task.unpark();
+            task.notify();
         }
         Ok(())
     }
diff --git a/src/unsync/oneshot.rs b/src/unsync/oneshot.rs
index 6008efa6..0cd44bed 100644
--- a/src/unsync/oneshot.rs
+++ b/src/unsync/oneshot.rs
@@ -110,7 +110,7 @@ impl<T> Sender<T> {
     pub fn poll_cancel(&mut self) -> Poll<(), ()> {
         match self.inner.upgrade() {
             Some(inner) => {
-                inner.borrow_mut().tx_task = Some(task::park());
+                inner.borrow_mut().tx_task = Some(task::current());
                 Ok(Async::NotReady)
             }
             None => Ok(().into()),
@@ -130,7 +130,7 @@ impl<T> Drop for Sender<T> {
             borrow.rx_task.take()
         };
         if let Some(task) = rx_task {
-            task.unpark();
+            task.notify();
         }
     }
 }
@@ -153,7 +153,7 @@ impl<T> Receiver<T> {
         };
         self.state = State::Closed(item);
         if let Some(task) = task {
-            task.unpark();
+            task.notify();
         }
     }
 }
@@ -184,7 +184,7 @@ impl<T> Future for Receiver<T> {
         if Rc::get_mut(inner).is_some() {
             Err(Canceled)
         } else {
-            inner.borrow_mut().rx_task = Some(task::park());
+            inner.borrow_mut().rx_task = Some(task::current());
             Ok(Async::NotReady)
         }
     }


Test patch:
diff --git a/tests/all.rs b/tests/all.rs
index bc872bb3..716d1df5 100644
--- a/tests/all.rs
+++ b/tests/all.rs
@@ -127,7 +127,7 @@ fn smoke_oneshot() {
 
     let (c, p) = oneshot::channel::<i32>();
     drop(c);
-    let res = executor::spawn(p).poll_future(unpark_panic());
+    let res = executor::spawn(p).poll_future_notify(&notify_panic(), 0);
     assert!(res.is_err());
     let (c, p) = oneshot::channel::<i32>();
     drop(c);
@@ -150,7 +150,7 @@ fn select_cancels() {
     assert!(brx.try_recv().is_err());
     assert!(drx.try_recv().is_err());
     a.send(1).unwrap();
-    let res = executor::spawn(f).poll_future(unpark_panic());
+    let res = executor::spawn(f).poll_future_notify(&notify_panic(), 0);
     assert!(res.ok().unwrap().is_ready());
     assert_eq!(brx.recv().unwrap(), 1);
     drop(c);
@@ -162,10 +162,10 @@ fn select_cancels() {
     let d = d.map(move |d| { dtx.send(d).unwrap(); d });
 
     let mut f = executor::spawn(b.select(d).then(unselect));
-    assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
-    assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
+    assert!(f.poll_future_notify(&notify_noop(), 0).ok().unwrap().is_not_ready());
+    assert!(f.poll_future_notify(&notify_noop(), 0).ok().unwrap().is_not_ready());
     a.send(1).unwrap();
-    assert!(f.poll_future(unpark_panic()).ok().unwrap().is_ready());
+    assert!(f.poll_future_notify(&notify_panic(), 0).ok().unwrap().is_ready());
     drop((c, f));
     assert!(drx.recv().is_err());
 }
@@ -179,7 +179,7 @@ fn join_cancels() {
 
     let f = b.join(d);
     drop(a);
-    let res = executor::spawn(f).poll_future(unpark_panic());
+    let res = executor::spawn(f).poll_future_notify(&notify_panic(), 0);
     assert!(res.is_err());
     drop(c);
     assert!(drx.recv().is_err());
@@ -208,37 +208,37 @@ fn join_incomplete() {
     let (a, b) = oneshot::channel::<i32>();
     let (tx, rx) = channel();
     let mut f = executor::spawn(ok(1).join(b).map(move |r| tx.send(r).unwrap()));
-    assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
+    assert!(f.poll_future_notify(&notify_noop(), 0).ok().unwrap().is_not_ready());
     assert!(rx.try_recv().is_err());
     a.send(2).unwrap();
-    assert!(f.poll_future(unpark_noop()).ok().unwrap().is_ready());
+    assert!(f.poll_future_notify(&notify_noop(), 0).ok().unwrap().is_ready());
     assert_eq!(rx.recv().unwrap(), (1, 2));
 
     let (a, b) = oneshot::channel::<i32>();
     let (tx, rx) = channel();
     let mut f = executor::spawn(b.join(Ok(2)).map(move |r| tx.send(r).unwrap()));
-    assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
+    assert!(f.poll_future_notify(&notify_noop(), 0).ok().unwrap().is_not_ready());
     assert!(rx.try_recv().is_err());
     a.send(1).unwrap();
-    assert!(f.poll_future(unpark_noop()).ok().unwrap().is_ready());
+    assert!(f.poll_future_notify(&notify_noop(), 0).ok().unwrap().is_ready());
     assert_eq!(rx.recv().unwrap(), (1, 2));
 
     let (a, b) = oneshot::channel::<i32>();
     let (tx, rx) = channel();
     let mut f = executor::spawn(ok(1).join(b).map_err(move |_r| tx.send(2).unwrap()));
-    assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
+    assert!(f.poll_future_notify(&notify_noop(), 0).ok().unwrap().is_not_ready());
     assert!(rx.try_recv().is_err());
     drop(a);
-    assert!(f.poll_future(unpark_noop()).is_err());
+    assert!(f.poll_future_notify(&notify_noop(), 0).is_err());
     assert_eq!(rx.recv().unwrap(), 2);
 
     let (a, b) = oneshot::channel::<i32>();
     let (tx, rx) = channel();
     let mut f = executor::spawn(b.join(Ok(2)).map_err(move |_r| tx.send(1).unwrap()));
-    assert!(f.poll_future(unpark_noop()).ok().unwrap().is_not_ready());
+    assert!(f.poll_future_notify(&notify_noop(), 0).ok().unwrap().is_not_ready());
     assert!(rx.try_recv().is_err());
     drop(a);
-    assert!(f.poll_future(unpark_noop()).is_err());
+    assert!(f.poll_future_notify(&notify_noop(), 0).is_err());
     assert_eq!(rx.recv().unwrap(), 1);
 }
 
@@ -323,7 +323,7 @@ fn select2() {
         let b = b.map(move |v| { btx.send(v).unwrap(); v });
         let d = d.map(move |v| { dtx.send(v).unwrap(); v });
         let f = b.select(d);
-        drop(executor::spawn(f).poll_future(support::unpark_noop()));
+        drop(executor::spawn(f).poll_future_notify(&support::notify_noop(), 0));
         assert!(drx.recv().is_err());
         assert!(brx.recv().is_err());
     }
diff --git a/tests/bilock.rs b/tests/bilock.rs
index 1f94397e..0458779a 100644
--- a/tests/bilock.rs
+++ b/tests/bilock.rs
@@ -39,7 +39,7 @@ fn smoke() {
     });
 
     assert!(executor::spawn(future)
-                .poll_future(unpark_noop())
+                .poll_future_notify(&notify_noop(), 0)
                 .expect("failure in poll")
                 .is_ready());
 }
diff --git a/tests/fuse.rs b/tests/fuse.rs
index cffb8be9..670e3f0c 100644
--- a/tests/fuse.rs
+++ b/tests/fuse.rs
@@ -9,6 +9,6 @@ use support::*;
 #[test]
 fn fuse() {
     let mut future = executor::spawn(ok::<i32, u32>(2).fuse());
-    assert!(future.poll_future(unpark_panic()).unwrap().is_ready());
-    assert!(future.poll_future(unpark_panic()).unwrap().is_not_ready());
+    assert!(future.poll_future_notify(&notify_panic(), 0).unwrap().is_ready());
+    assert!(future.poll_future_notify(&notify_panic(), 0).unwrap().is_not_ready());
 }
diff --git a/tests/futures_unordered.rs b/tests/futures_unordered.rs
index d877f03b..e7b7bd6f 100644
--- a/tests/futures_unordered.rs
+++ b/tests/futures_unordered.rs
@@ -38,9 +38,9 @@ fn works_2() {
     let mut spawn = futures::executor::spawn(stream);
     a_tx.send(33).unwrap();
     b_tx.send(33).unwrap();
-    assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_ready());
+    assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_ready());
     c_tx.send(33).unwrap();
-    assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_ready());
+    assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_ready());
 }
 
 #[test]
@@ -56,13 +56,13 @@ fn finished_future_ok() {
 
     let mut spawn = futures::executor::spawn(stream);
     for _ in 0..10 {
-        assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_not_ready());
+        assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_not_ready());
     }
 
     b_tx.send(Box::new(())).unwrap();
-    let next = spawn.poll_stream(support::unpark_noop()).unwrap();
+    let next = spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap();
     assert!(next.is_ready());
     c_tx.send(Box::new(())).unwrap();
-    assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_not_ready());
-    assert!(spawn.poll_stream(support::unpark_noop()).unwrap().is_not_ready());
+    assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_not_ready());
+    assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_not_ready());
 }
diff --git a/tests/oneshot.rs b/tests/oneshot.rs
index 405906c1..7097a591 100644
--- a/tests/oneshot.rs
+++ b/tests/oneshot.rs
@@ -21,7 +21,7 @@ fn smoke_poll() {
         assert!(tx.poll_cancel().unwrap().is_ready());
         ok::<(), ()>(())
     }));
-    assert!(task.poll_future(unpark_noop()).unwrap().is_ready());
+    assert!(task.poll_future_notify(&notify_noop(), 0).unwrap().is_ready());
 }
 
 #[test]
diff --git a/tests/ready_queue.rs b/tests/ready_queue.rs
new file mode 100644
index 00000000..256b966e
--- /dev/null
+++ b/tests/ready_queue.rs
@@ -0,0 +1,170 @@
+extern crate futures;
+
+use std::panic::{self, AssertUnwindSafe};
+
+use futures::{Future, Stream, Poll};
+use futures::Async::*;
+use futures::future;
+use futures::stream::FuturesUnordered;
+use futures::sync::oneshot;
+
+#[test]
+fn bounds() {
+    fn is_send<T: Send>() {}
+    fn is_sync<T: Sync>() {}
+
+    is_send::<FuturesUnordered<()>>();
+    is_sync::<FuturesUnordered<()>>();
+}
+
+#[test]
+fn basic_usage() {
+    future::lazy(move || {
+        let mut queue = FuturesUnordered::new();
+        let (tx1, rx1) = oneshot::channel();
+        let (tx2, rx2) = oneshot::channel();
+        let (tx3, rx3) = oneshot::channel();
+
+        queue.push(rx1);
+        queue.push(rx2);
+        queue.push(rx3);
+
+        assert!(!queue.poll().unwrap().is_ready());
+
+        tx2.send("hello").unwrap();
+
+        assert_eq!(Ready(Some("hello")), queue.poll().unwrap());
+        assert!(!queue.poll().unwrap().is_ready());
+
+        tx1.send("world").unwrap();
+        tx3.send("world2").unwrap();
+
+        assert_eq!(Ready(Some("world")), queue.poll().unwrap());
+        assert_eq!(Ready(Some("world2")), queue.poll().unwrap());
+        assert_eq!(Ready(None), queue.poll().unwrap());
+
+        Ok::<_, ()>(())
+    }).wait().unwrap();
+}
+
+#[test]
+fn resolving_errors() {
+    future::lazy(move || {
+        let mut queue = FuturesUnordered::new();
+        let (tx1, rx1) = oneshot::channel();
+        let (tx2, rx2) = oneshot::channel();
+        let (tx3, rx3) = oneshot::channel();
+
+        queue.push(rx1);
+        queue.push(rx2);
+        queue.push(rx3);
+
+        assert!(!queue.poll().unwrap().is_ready());
+
+        drop(tx2);
+
+        assert!(queue.poll().is_err());
+        assert!(!queue.poll().unwrap().is_ready());
+
+        drop(tx1);
+        tx3.send("world2").unwrap();
+
+        assert!(queue.poll().is_err());
+        assert_eq!(Ready(Some("world2")), queue.poll().unwrap());
+        assert_eq!(Ready(None), queue.poll().unwrap());
+
+        Ok::<_, ()>(())
+    }).wait().unwrap();
+}
+
+#[test]
+fn dropping_ready_queue() {
+    future::lazy(move || {
+        let mut queue = FuturesUnordered::new();
+        let (mut tx1, rx1) = oneshot::channel::<()>();
+        let (mut tx2, rx2) = oneshot::channel::<()>();
+        let (mut tx3, rx3) = oneshot::channel::<()>();
+
+        queue.push(rx1);
+        queue.push(rx2);
+        queue.push(rx3);
+
+        assert!(!tx1.poll_cancel().unwrap().is_ready());
+        assert!(!tx2.poll_cancel().unwrap().is_ready());
+        assert!(!tx3.poll_cancel().unwrap().is_ready());
+
+        drop(queue);
+
+        assert!(tx1.poll_cancel().unwrap().is_ready());
+        assert!(tx2.poll_cancel().unwrap().is_ready());
+        assert!(tx3.poll_cancel().unwrap().is_ready());
+
+        Ok::<_, ()>(())
+    }).wait().unwrap();
+}
+
+#[test]
+fn stress() {
+    const ITER: usize = 300;
+
+    use std::sync::{Arc, Barrier};
+    use std::thread;
+
+    for i in 0..ITER {
+        let n = (i % 10) + 1;
+
+        let mut queue = FuturesUnordered::new();
+
+        for _ in 0..5 {
+            let barrier = Arc::new(Barrier::new(n + 1));
+
+            for num in 0..n {
+                let barrier = barrier.clone();
+                let (tx, rx) = oneshot::channel();
+
+                queue.push(rx);
+
+                thread::spawn(move || {
+                    barrier.wait();
+                    tx.send(num).unwrap();
+                });
+            }
+
+            barrier.wait();
+
+            let mut sync = queue.wait();
+
+            let mut rx: Vec<_> = (&mut sync)
+                .take(n)
+                .map(|res| res.unwrap())
+                .collect();
+
+            assert_eq!(rx.len(), n);
+
+            rx.sort();
+
+            for num in 0..n {
+                assert_eq!(rx[num], num);
+            }
+
+            queue = sync.into_inner();
+        }
+    }
+}
+
+#[test]
+fn panicking_future_dropped() {
+    future::lazy(move || {
+        let mut queue = FuturesUnordered::new();
+        queue.push(future::poll_fn(|| -> Poll<i32, i32> {
+            panic!()
+        }));
+
+        let r = panic::catch_unwind(AssertUnwindSafe(|| queue.poll()));
+        assert!(r.is_err());
+        assert!(queue.is_empty());
+        assert_eq!(Ready(None), queue.poll().unwrap());
+
+        Ok::<_, ()>(())
+    }).wait().unwrap();
+}
diff --git a/tests/shared.rs b/tests/shared.rs
index f74f5a94..3a733a97 100644
--- a/tests/shared.rs
+++ b/tests/shared.rs
@@ -186,7 +186,7 @@ fn recursive_poll_with_unpark() {
     let f2 = f1.clone();
     let f3 = f1.clone();
     tx0.send(Box::new(future::lazy(move || {
-        task::park().unpark();
+        task::current().notify();
         f1.map(|_|()).map_err(|_|())
             .select(rx1.map_err(|_|()))
             .map(|_| ()).map_err(|_|())
diff --git a/tests/sink.rs b/tests/sink.rs
index a65cdeec..54a4a0e1 100644
--- a/tests/sink.rs
+++ b/tests/sink.rs
@@ -11,7 +11,7 @@ use futures::future::ok;
 use futures::stream;
 use futures::sync::{oneshot, mpsc};
 use futures::task::{self, Task};
-use futures::executor::{self, Unpark};
+use futures::executor::{self, Notify};
 use futures::sink::*;
 
 mod support;
@@ -72,8 +72,8 @@ impl Flag {
     }
 }
 
-impl Unpark for Flag {
-    fn unpark(&self) {
+impl Notify for Flag {
+    fn notify(&self, _id: usize) {
         self.set(true)
     }
 }
@@ -115,12 +115,12 @@ fn mpsc_blocking_start_send() {
         let flag = Flag::new();
         let mut task = executor::spawn(StartSendFut::new(tx, 1));
 
-        assert!(task.poll_future(flag.clone()).unwrap().is_not_ready());
+        assert!(task.poll_future_notify(&flag, 0).unwrap().is_not_ready());
         assert!(!flag.get());
         sassert_next(&mut rx, 0);
         assert!(flag.get());
         flag.set(false);
-        assert!(task.poll_future(flag.clone()).unwrap().is_ready());
+        assert!(task.poll_future_notify(&flag, 0).unwrap().is_ready());
         assert!(!flag.get());
         sassert_next(&mut rx, 1);
 
@@ -143,11 +143,11 @@ fn with_flush() {
 
     let flag = Flag::new();
     let mut task = executor::spawn(sink.flush());
-    assert!(task.poll_future(flag.clone()).unwrap().is_not_ready());
+    assert!(task.poll_future_notify(&flag, 0).unwrap().is_not_ready());
     tx.send(()).unwrap();
     assert!(flag.get());
 
-    let sink = match task.poll_future(flag.clone()).unwrap() {
+    let sink = match task.poll_future_notify(&flag, 0).unwrap() {
         Async::Ready(sink) => sink,
         _ => panic!()
     };
@@ -191,7 +191,7 @@ impl<T> Sink for ManualFlush<T> {
         if self.data.is_empty() {
             Ok(Async::Ready(()))
         } else {
-            self.waiting_tasks.push(task::park());
+            self.waiting_tasks.push(task::current());
             Ok(Async::NotReady)
         }
     }
@@ -211,7 +211,7 @@ impl<T> ManualFlush<T> {
 
     fn force_flush(&mut self) -> Vec<T> {
         for task in self.waiting_tasks.drain(..) {
-            task.unpark()
+            task.notify()
         }
         mem::replace(&mut self.data, Vec::new())
     }
@@ -227,11 +227,11 @@ fn with_flush_propagate() {
 
     let flag = Flag::new();
     let mut task = executor::spawn(sink.flush());
-    assert!(task.poll_future(flag.clone()).unwrap().is_not_ready());
+    assert!(task.poll_future_notify(&flag, 0).unwrap().is_not_ready());
     assert!(!flag.get());
     assert_eq!(task.get_mut().get_mut().get_mut().force_flush(), vec![0, 1]);
     assert!(flag.get());
-    assert!(task.poll_future(flag.clone()).unwrap().is_ready());
+    assert!(task.poll_future_notify(&flag, 0).unwrap().is_ready());
 }
 
 #[test]
@@ -270,7 +270,7 @@ impl Allow {
         if self.flag.get() {
             true
         } else {
-            self.tasks.borrow_mut().push(task::park());
+            self.tasks.borrow_mut().push(task::current());
             false
         }
     }
@@ -279,7 +279,7 @@ impl Allow {
         self.flag.set(true);
         let mut tasks = self.tasks.borrow_mut();
         for task in tasks.drain(..) {
-            task.unpark();
+            task.notify();
         }
     }
 }
@@ -327,11 +327,11 @@ fn buffer() {
 
     let flag = Flag::new();
     let mut task = executor::spawn(sink.send(2));
-    assert!(task.poll_future(flag.clone()).unwrap().is_not_ready());
+    assert!(task.poll_future_notify(&flag, 0).unwrap().is_not_ready());
     assert!(!flag.get());
     allow.start();
     assert!(flag.get());
-    match task.poll_future(flag.clone()).unwrap() {
+    match task.poll_future_notify(&flag, 0).unwrap() {
         Async::Ready(sink) => {
             assert_eq!(sink.get_ref().data, vec![0, 1, 2]);
         }
diff --git a/tests/stream.rs b/tests/stream.rs
index 2a776dd2..0d5961a5 100644
--- a/tests/stream.rs
+++ b/tests/stream.rs
@@ -339,10 +339,10 @@ fn forward() {
 #[test]
 fn concat() {
     let a = iter(vec![Ok::<_, ()>(vec![1, 2, 3]), Ok(vec![4, 5, 6]), Ok(vec![7, 8, 9])]);
-    assert_done(move || a.concat(), Ok(vec![1, 2, 3, 4, 5, 6, 7, 8, 9]));
+    assert_done(move || a.concat2(), Ok(vec![1, 2, 3, 4, 5, 6, 7, 8, 9]));
 
     let b = iter(vec![Ok::<_, ()>(vec![1, 2, 3]), Err(()), Ok(vec![7, 8, 9])]);
-    assert_done(move || b.concat(), Err(()));
+    assert_done(move || b.concat2(), Err(()));
 }
 
 #[test]
diff --git a/tests/support/local_executor.rs b/tests/support/local_executor.rs
index 8fc0d977..704d968c 100644
--- a/tests/support/local_executor.rs
+++ b/tests/support/local_executor.rs
@@ -17,10 +17,10 @@ use futures::executor::{self, Spawn};
 
 /// Main loop object
 pub struct Core {
-    unpark_send: mpsc::Sender<u64>,
-    unpark: mpsc::Receiver<u64>,
-    live: HashMap<u64, Spawn<Box<Future<Item=(), Error=()>>>>,
-    next_id: u64,
+    unpark_send: mpsc::Sender<usize>,
+    unpark: mpsc::Receiver<usize>,
+    live: HashMap<usize, Spawn<Box<Future<Item=(), Error=()>>>>,
+    next_id: usize,
 }
 
 impl Core {
@@ -70,9 +70,9 @@ impl Core {
 
     fn turn(&mut self) {
         let task = self.unpark.recv().unwrap(); // Safe to unwrap because self.unpark_send keeps the channel alive
-        let unpark = Arc::new(Unpark { task: task, send: Mutex::new(self.unpark_send.clone()), });
+        let notify = Arc::new(Notify { task: task, send: Mutex::new(self.unpark_send.clone()), });
         let mut task = if let hash_map::Entry::Occupied(x) = self.live.entry(task) { x } else { return };
-        let result = task.get_mut().poll_future(unpark);
+        let result = task.get_mut().poll_future_notify(&notify, 0);
         match result {
             Ok(Async::Ready(())) => { task.remove(); }
             Err(()) => { task.remove(); }
@@ -81,13 +81,13 @@ impl Core {
     }
 }
 
-struct Unpark {
-    task: u64,
-    send: Mutex<mpsc::Sender<u64>>,
+struct Notify {
+    task: usize,
+    send: Mutex<mpsc::Sender<usize>>,
 }
 
-impl executor::Unpark for Unpark {
-    fn unpark(&self) {
+impl executor::Notify for Notify {
+    fn notify(&self, _id: usize) {
         let _ = self.send.lock().unwrap().send(self.task);
     }
 }
diff --git a/tests/support/mod.rs b/tests/support/mod.rs
index df4c8c60..65df6d23 100755
--- a/tests/support/mod.rs
+++ b/tests/support/mod.rs
@@ -7,7 +7,7 @@ use std::thread;
 use futures::{Future, IntoFuture, Async, Poll};
 use futures::future::FutureResult;
 use futures::stream::Stream;
-use futures::executor::{self, Unpark};
+use futures::executor::{self, NotifyHandle, Notify};
 use futures::task;
 
 pub mod local_executor;
@@ -27,11 +27,11 @@ pub fn assert_done<T, F>(f: F, result: Result<T::Item, T::Error>)
 }
 
 pub fn assert_empty<T: Future, F: FnMut() -> T>(mut f: F) {
-    assert!(executor::spawn(f()).poll_future(unpark_panic()).ok().unwrap().is_not_ready());
+    assert!(executor::spawn(f()).poll_future_notify(&notify_panic(), 0).ok().unwrap().is_not_ready());
 }
 
 pub fn sassert_done<S: Stream>(s: &mut S) {
-    match executor::spawn(s).poll_stream(unpark_panic()) {
+    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"),
@@ -40,7 +40,7 @@ pub fn sassert_done<S: Stream>(s: &mut S) {
 }
 
 pub fn sassert_empty<S: Stream>(s: &mut S) {
-    match executor::spawn(s).poll_stream(unpark_noop()) {
+    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) => {}
@@ -51,7 +51,7 @@ pub fn sassert_empty<S: Stream>(s: &mut S) {
 pub fn sassert_next<S: Stream>(s: &mut S, item: S::Item)
     where S::Item: Eq + fmt::Debug
 {
-    match executor::spawn(s).poll_stream(unpark_panic()) {
+    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"),
@@ -62,7 +62,7 @@ pub fn sassert_next<S: Stream>(s: &mut S, item: S::Item)
 pub fn sassert_err<S: Stream>(s: &mut S, err: S::Error)
     where S::Error: Eq + fmt::Debug
 {
-    match executor::spawn(s).poll_stream(unpark_panic()) {
+    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"),
@@ -70,26 +70,28 @@ pub fn sassert_err<S: Stream>(s: &mut S, err: S::Error)
     }
 }
 
-pub fn unpark_panic() -> Arc<Unpark> {
+pub fn notify_panic() -> NotifyHandle {
     struct Foo;
 
-    impl Unpark for Foo {
-        fn unpark(&self) {
-            panic!("should not be unparked");
+    impl Notify for Foo {
+        fn notify(&self, _id: usize) {
+            panic!("should not be notified");
         }
     }
 
-    Arc::new(Foo)
+    NotifyHandle::from(Arc::new(Foo))
 }
 
-pub fn unpark_noop() -> Arc<Unpark> {
-    struct Foo;
+pub fn notify_noop() -> NotifyHandle {
+    struct Noop;
 
-    impl Unpark for Foo {
-        fn unpark(&self) {}
+    impl Notify for Noop {
+        fn notify(&self, _id: usize) {}
     }
 
-    Arc::new(Foo)
+    const NOOP : &'static Noop = &Noop;
+
+    NotifyHandle::from(NOOP)
 }
 
 pub trait ForgetExt {
@@ -117,7 +119,7 @@ impl<F: Future> Future for DelayFuture<F> {
             self.0.poll()
         } else {
             self.1 = true;
-            task::park().unpark();
+            task::current().notify();
             Ok(Async::NotReady)
         }
     }