Introduce the Tokio runtime: Reactor + Threadpool (#141)

This patch is an intial implementation of the Tokio runtime. The Tokio
runtime provides an out of the box configuration for running I/O heavy
asynchronous applications.

As of now, the Tokio runtime is a combination of a work-stealing thread
pool as well as a background reactor to drive I/O resources.

This patch also includes tokio-executor, a hopefully short lived crate
that is based on the futures 0.2 executor RFC.

* Implement `Park` for `Reactor`

This enables the reactor to be used as the thread parker for executors.
This also adds an `Error` component to `Park`. With this change, a
`Reactor` and a `CurrentThread` can be combined to achieve the
capabilities of tokio-core.

tokio-threadpool/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "tokio-threadpool"
+version = "0.1.0"
+documentation = "https://docs.rs/tokio-threadpool"
+repository = "https://github.com/tokio-rs/tokio"
+homepage = "https://github.com/tokio-rs/tokio"
+license = "MIT/Apache-2.0"
+authors = ["Carl Lerche <[email protected]>"]
+description = """
+A Future aware thread pool based on work stealing.
+"""
+keywords = ["futures", "tokio"]
+categories = ["concurrency", "asynchronous"]
+
+[dependencies]
+tokio-executor = { version = "0.1", path = "../tokio-executor" }
+futures = "0.1"
+coco = "0.3"
+num_cpus = "1.2"
+rand = "0.3"
+log = "0.3"
+
+[dev-dependencies]
+tokio-timer = "0.1"
+env_logger = "0.4"
+futures-cpupool = "0.1.7"

tokio-threadpool/README.md

@@ -0,0 +1,52 @@
+# Tokio Thread Pool
+
+A library for scheduling execution of futures concurrently across a pool of
+threads.
+
+**Note**: This library isn't quite ready for use.
+
+### Why not Rayon?
+
+Rayon is designed to handle parallelizing single computations by breaking them
+into smaller chunks. The scheduling for each individual chunk doesn't matter as
+long as the root computation completes in a timely fashion. In other words,
+Rayon does not provide any guarantees of fairness with regards to how each task
+gets scheduled.
+
+On the other hand, `tokio-threadpool` is a general purpose scheduler and
+attempts to schedule each task fairly. This is the ideal behavior when
+scheduling a set of unrelated tasks.
+
+### Why not futures-cpupool?
+
+It's 10x slower.
+
+## Examples
+
+```rust
+extern crate tokio_threadpool;
+extern crate futures;
+
+use tokio_threadpool::*;
+use futures::*;
+use futures::sync::oneshot;
+
+pub fn main() {
+    let (tx, _pool) = ThreadPool::new();
+
+    let res = oneshot::spawn(future::lazy(|| {
+        println!("Running on the pool");
+        Ok::<_, ()>("complete")
+    }), &tx);
+
+    println!("Result: {:?}", res.wait());
+}
+```
+
+## License
+
+`tokio-threadpool` is primarily distributed under the terms of both the MIT
+license and the Apache License (Version 2.0), with portions covered by various
+BSD-like licenses.
+
+See LICENSE-APACHE, and LICENSE-MIT for details.

tokio-threadpool/benches/basic.rs

@@ -0,0 +1,162 @@
+#![feature(test)]
+
+extern crate futures;
+extern crate futures_pool;
+extern crate futures_cpupool;
+extern crate num_cpus;
+extern crate test;
+
+const NUM_SPAWN: usize = 10_000;
+const NUM_YIELD: usize = 1_000;
+const TASKS_PER_CPU: usize = 50;
+
+mod us {
+    use futures::{task, Async};
+    use futures::future::{self, Executor};
+    use futures_pool::*;
+    use num_cpus;
+    use test;
+    use std::sync::{mpsc, Arc};
+    use std::sync::atomic::AtomicUsize;
+    use std::sync::atomic::Ordering::SeqCst;
+
+    #[bench]
+    fn spawn_many(b: &mut test::Bencher) {
+        let (sched_tx, _scheduler) = Pool::new();
+
+        let (tx, rx) = mpsc::sync_channel(10);
+        let rem = Arc::new(AtomicUsize::new(0));
+
+        b.iter(move || {
+            rem.store(super::NUM_SPAWN, SeqCst);
+
+            for _ in 0..super::NUM_SPAWN {
+                let tx = tx.clone();
+                let rem = rem.clone();
+
+                sched_tx.execute(future::lazy(move || {
+                    if 1 == rem.fetch_sub(1, SeqCst) {
+                        tx.send(()).unwrap();
+                    }
+
+                    Ok(())
+                })).ok().unwrap();
+            }
+
+            let _ = rx.recv().unwrap();
+        });
+    }
+
+    #[bench]
+    fn yield_many(b: &mut test::Bencher) {
+        let (sched_tx, _scheduler) = Pool::new();
+        let tasks = super::TASKS_PER_CPU * num_cpus::get();
+
+        let (tx, rx) = mpsc::sync_channel(tasks);
+
+        b.iter(move || {
+            for _ in 0..tasks {
+                let mut rem = super::NUM_YIELD;
+                let tx = tx.clone();
+
+                sched_tx.execute(future::poll_fn(move || {
+                    rem -= 1;
+
+                    if rem == 0 {
+                        tx.send(()).unwrap();
+                        Ok(Async::Ready(()))
+                    } else {
+                        // Notify the current task
+                        task::current().notify();
+
+                        // Not ready
+                        Ok(Async::NotReady)
+                    }
+                })).ok().unwrap();
+            }
+
+            for _ in 0..tasks {
+                let _ = rx.recv().unwrap();
+            }
+        });
+    }
+}
+
+// In this case, CPU pool completes the benchmark faster, but this is due to how
+// CpuPool currently behaves, starving other futures. This completes the
+// benchmark quickly but results in poor runtime characteristics for a thread
+// pool.
+//
+// See alexcrichton/futures-rs#617
+//
+mod cpupool {
+    use futures::{task, Async};
+    use futures::future::{self, Executor};
+    use futures_cpupool::*;
+    use num_cpus;
+    use test;
+    use std::sync::{mpsc, Arc};
+    use std::sync::atomic::AtomicUsize;
+    use std::sync::atomic::Ordering::SeqCst;
+
+    #[bench]
+    fn spawn_many(b: &mut test::Bencher) {
+        let pool = CpuPool::new(num_cpus::get());
+
+        let (tx, rx) = mpsc::sync_channel(10);
+        let rem = Arc::new(AtomicUsize::new(0));
+
+        b.iter(move || {
+            rem.store(super::NUM_SPAWN, SeqCst);
+
+            for _ in 0..super::NUM_SPAWN {
+                let tx = tx.clone();
+                let rem = rem.clone();
+
+                pool.execute(future::lazy(move || {
+                    if 1 == rem.fetch_sub(1, SeqCst) {
+                        tx.send(()).unwrap();
+                    }
+
+                    Ok(())
+                })).ok().unwrap();
+            }
+
+            let _ = rx.recv().unwrap();
+        });
+    }
+
+    #[bench]
+    fn yield_many(b: &mut test::Bencher) {
+        let pool = CpuPool::new(num_cpus::get());
+        let tasks = super::TASKS_PER_CPU * num_cpus::get();
+
+        let (tx, rx) = mpsc::sync_channel(tasks);
+
+        b.iter(move || {
+            for _ in 0..tasks {
+                let mut rem = super::NUM_YIELD;
+                let tx = tx.clone();
+
+                pool.execute(future::poll_fn(move || {
+                    rem -= 1;
+
+                    if rem == 0 {
+                        tx.send(()).unwrap();
+                        Ok(Async::Ready(()))
+                    } else {
+                        // Notify the current task
+                        task::current().notify();
+
+                        // Not ready
+                        Ok(Async::NotReady)
+                    }
+                })).ok().unwrap();
+            }
+
+            for _ in 0..tasks {
+                let _ = rx.recv().unwrap();
+            }
+        });
+    }
+}

tokio-threadpool/benches/depth.rs

@@ -0,0 +1,72 @@
+#![feature(test)]
+
+extern crate futures;
+extern crate futures_pool;
+extern crate futures_cpupool;
+extern crate num_cpus;
+extern crate test;
+
+const ITER: usize = 20_000;
+
+mod us {
+    use futures::future::{self, Executor};
+    use futures_pool::*;
+    use test;
+    use std::sync::mpsc;
+
+    #[bench]
+    fn chained_spawn(b: &mut test::Bencher) {
+        let (sched_tx, _scheduler) = Pool::new();
+
+        fn spawn(sched_tx: Sender, res_tx: mpsc::Sender<()>, n: usize) {
+            if n == 0 {
+                res_tx.send(()).unwrap();
+            } else {
+                let sched_tx2 = sched_tx.clone();
+                sched_tx.execute(future::lazy(move || {
+                    spawn(sched_tx2, res_tx, n - 1);
+                    Ok(())
+                })).ok().unwrap();
+            }
+        }
+
+        b.iter(move || {
+            let (res_tx, res_rx) = mpsc::channel();
+
+            spawn(sched_tx.clone(), res_tx, super::ITER);
+            res_rx.recv().unwrap();
+        });
+    }
+}
+
+mod cpupool {
+    use futures::future::{self, Executor};
+    use futures_cpupool::*;
+    use num_cpus;
+    use test;
+    use std::sync::mpsc;
+
+    #[bench]
+    fn chained_spawn(b: &mut test::Bencher) {
+        let pool = CpuPool::new(num_cpus::get());
+
+        fn spawn(pool: CpuPool, res_tx: mpsc::Sender<()>, n: usize) {
+            if n == 0 {
+                res_tx.send(()).unwrap();
+            } else {
+                let pool2 = pool.clone();
+                pool.execute(future::lazy(move || {
+                    spawn(pool2, res_tx, n - 1);
+                    Ok(())
+                })).ok().unwrap();
+            }
+        }
+
+        b.iter(move || {
+            let (res_tx, res_rx) = mpsc::channel();
+
+            spawn(pool.clone(), res_tx, super::ITER);
+            res_rx.recv().unwrap();
+        });
+    }
+}

tokio-threadpool/examples/depth.rs

@@ -0,0 +1,46 @@
+extern crate futures;
+extern crate tokio_threadpool;
+extern crate env_logger;
+
+use tokio_threadpool::*;
+use futures::future::{self, Executor};
+
+use std::sync::mpsc;
+
+const ITER: usize = 2_000_000;
+// const ITER: usize = 30;
+
+fn chained_spawn() {
+    let pool = ThreadPool::new();
+    let tx = pool.sender().clone();
+
+    fn spawn(tx: Sender, res_tx: mpsc::Sender<()>, n: usize) {
+        if n == 0 {
+            res_tx.send(()).unwrap();
+        } else {
+            let tx2 = tx.clone();
+            tx.execute(future::lazy(move || {
+                spawn(tx2, res_tx, n - 1);
+                Ok(())
+            })).ok().unwrap();
+        }
+    }
+
+    loop {
+        println!("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
+        let (res_tx, res_rx) = mpsc::channel();
+
+        for _ in 0..10 {
+            spawn(tx.clone(), res_tx.clone(), ITER);
+        }
+
+        for _ in 0..10 {
+            res_rx.recv().unwrap();
+        }
+    }
+}
+
+pub fn main() {
+    let _ = ::env_logger::init();
+    chained_spawn();
+}

tokio-threadpool/examples/hello.rs

@@ -0,0 +1,21 @@
+extern crate futures;
+extern crate tokio_threadpool;
+extern crate env_logger;
+
+use tokio_threadpool::*;
+use futures::*;
+use futures::sync::oneshot;
+
+pub fn main() {
+    let _ = ::env_logger::init();
+
+    let pool = ThreadPool::new();
+    let tx = pool.sender().clone();
+
+    let res = oneshot::spawn(future::lazy(|| {
+        println!("Running on the pool");
+        Ok::<_, ()>("complete")
+    }), &tx);
+
+    println!("Result: {:?}", res.wait());
+}

tokio-threadpool/examples/smoke.rs

@@ -0,0 +1,34 @@
+extern crate futures;
+extern crate tokio_threadpool;
+extern crate tokio_timer;
+extern crate env_logger;
+
+use tokio_threadpool::*;
+use tokio_timer::Timer;
+
+use futures::*;
+use futures::sync::oneshot::spawn;
+
+use std::thread;
+use std::time::Duration;
+
+pub fn main() {
+    let _ = ::env_logger::init();
+
+    let timer = Timer::default();
+    {
+        let pool = ThreadPool::new();
+        let tx = pool.sender().clone();
+
+        let fut = timer.interval(Duration::from_millis(300))
+            .for_each(|_| {
+                println!("~~~~~ Hello ~~~");
+                Ok(())
+            })
+            .map_err(|_| unimplemented!());
+
+        spawn(fut, &tx).wait().unwrap();
+    }
+
+    thread::sleep(Duration::from_millis(100));
+}

tokio-threadpool/src/task.rs

@@ -0,0 +1,429 @@
+use Notifier;
+
+use futures::{future, Future, Async};
+use futures::executor::{self, Spawn};
+
+use std::{fmt, mem, ptr};
+use std::cell::Cell;
+use std::sync::Arc;
+use std::sync::atomic::{self, AtomicUsize, AtomicPtr};
+use std::sync::atomic::Ordering::{AcqRel, Acquire, Release, Relaxed};
+
+pub(crate) struct Task {
+    ptr: *mut Inner,
+}
+
+#[derive(Debug)]
+pub(crate) struct Queue {
+    head: AtomicPtr<Inner>,
+    tail: Cell<*mut Inner>,
+    stub: Box<Inner>,
+}
+
+#[derive(Debug)]
+pub(crate) enum Poll {
+    Empty,
+    Inconsistent,
+    Data(Task),
+}
+
+#[derive(Debug)]
+pub(crate) enum Run {
+    Idle,
+    Schedule,
+    Complete,
+}
+
+struct Inner {
+    // Next pointer in the queue that submits tasks to a worker.
+    next: AtomicPtr<Inner>,
+
+    // Task state
+    state: AtomicUsize,
+
+    // Number of outstanding references to the task
+    ref_count: AtomicUsize,
+
+    // Store the future at the head of the struct
+    //
+    // The future is dropped immediately when it transitions to Complete
+    future: Option<Spawn<BoxFuture>>,
+}
+
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+enum State {
+    /// Task is currently idle
+    Idle,
+    /// Task is currently running
+    Running,
+    /// Task is currently running, but has been notified that it must run again.
+    Notified,
+    /// Task has been scheduled
+    Scheduled,
+    /// Task is complete
+    Complete,
+}
+
+type BoxFuture = Box<Future<Item = (), Error = ()> + Send + 'static>;
+
+// ===== impl Task =====
+
+impl Task {
+    /// Create a new task handle
+    pub fn new(future: BoxFuture) -> Task {
+        let inner = Box::new(Inner {
+            next: AtomicPtr::new(ptr::null_mut()),
+            state: AtomicUsize::new(State::new().into()),
+            ref_count: AtomicUsize::new(1),
+            future: Some(executor::spawn(future)),
+        });
+
+        Task { ptr: Box::into_raw(inner) }
+    }
+
+    /// Transmute a u64 to a Task
+    pub unsafe fn from_notify_id(unpark_id: usize) -> Task {
+        mem::transmute(unpark_id)
+    }
+
+    /// Transmute a u64 to a task ref
+    pub unsafe fn from_notify_id_ref<'a>(unpark_id: &'a usize) -> &'a Task {
+        mem::transmute(unpark_id)
+    }
+
+    /// Execute the task returning `Run::Schedule` if the task needs to be
+    /// scheduled again.
+    pub fn run(&self, unpark: &Arc<Notifier>) -> Run {
+        use self::State::*;
+
+        // Transition task to running state. At this point, the task must be
+        // scheduled.
+        let actual: State = self.inner().state.compare_and_swap(
+            Scheduled.into(), Running.into(), AcqRel).into();
+
+        trace!("running; state={:?}", actual);
+
+        match actual {
+            Scheduled => {},
+            _ => panic!("unexpected task state; {:?}", actual),
+        }
+
+        trace!("Task::run; state={:?}", State::from(self.inner().state.load(Relaxed)));
+
+        let res = self.inner_mut().future.as_mut().unwrap()
+            .poll_future_notify(unpark, self.ptr as usize);
+
+        match res {
+            Ok(Async::Ready(_)) | Err(_) => {
+                trace!("    -> task complete");
+
+                // Drop the future
+                self.inner_mut().drop_future();
+
+                // Transition to the completed state
+                self.inner().state.store(State::Complete.into(), Release);
+
+                Run::Complete
+            }
+            _ => {
+                trace!("    -> not ready");
+
+                // Attempt to transition from Running -> Idle, if successful,
+                // then the task does not need to be scheduled again. If the CAS
+                // fails, then the task has been unparked concurrent to running,
+                // in which case it transitions immediately back to scheduled
+                // and we return `true`.
+                let prev: State = self.inner().state.compare_and_swap(
+                    Running.into(), Idle.into(), AcqRel).into();
+
+                match prev {
+                    Running => Run::Idle,
+                    Notified => {
+                        self.inner().state.store(Scheduled.into(), Release);
+                        Run::Schedule
+                    }
+                    _ => unreachable!(),
+                }
+            }
+        }
+    }
+
+    /// Transition the task state to scheduled.
+    ///
+    /// Returns `true` if the caller is permitted to schedule the task.
+    pub fn schedule(&self) -> bool {
+        use self::State::*;
+
+        loop {
+            let actual = self.inner().state.compare_and_swap(
+                Idle.into(),
+                Scheduled.into(),
+                Relaxed).into();
+
+            match actual {
+                Idle => return true,
+                Running => {
+                    let actual = self.inner().state.compare_and_swap(
+                        Running.into(), Notified.into(), Relaxed).into();
+
+                    match actual {
+                        Idle => continue,
+                        _ => return false,
+                    }
+                }
+                Complete | Notified | Scheduled => return false,
+            }
+        }
+    }
+
+    #[inline]
+    fn inner(&self) -> &Inner {
+        unsafe { &*self.ptr }
+    }
+
+    #[inline]
+    fn inner_mut(&self) -> &mut Inner {
+        unsafe { &mut *self.ptr }
+    }
+}
+
+impl fmt::Debug for Task {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        fmt.debug_struct("Task")
+            .field("inner", self.inner())
+            .finish()
+    }
+}
+
+impl Clone for Task {
+    fn clone(&self) -> Task {
+        use std::isize;
+
+        const MAX_REFCOUNT: usize = (isize::MAX) as usize;
+        // Using a relaxed ordering is alright here, as knowledge of the
+        // original reference prevents other threads from erroneously deleting
+        // the object.
+        //
+        // As explained in the [Boost documentation][1], Increasing the
+        // reference counter can always be done with memory_order_relaxed: New
+        // references to an object can only be formed from an existing
+        // reference, and passing an existing reference from one thread to
+        // another must already provide any required synchronization.
+        //
+        // [1]: (www.boost.org/doc/libs/1_55_0/doc/html/atomic/usage_examples.html)
+        let old_size = self.inner().ref_count.fetch_add(1, Relaxed);
+
+        // However we need to guard against massive refcounts in case someone
+        // is `mem::forget`ing Arcs. If we don't do this the count can overflow
+        // and users will use-after free. We racily saturate to `isize::MAX` on
+        // the assumption that there aren't ~2 billion threads incrementing
+        // the reference count at once. This branch will never be taken in
+        // any realistic program.
+        //
+        // We abort because such a program is incredibly degenerate, and we
+        // don't care to support it.
+        if old_size > MAX_REFCOUNT {
+            // TODO: abort
+            panic!();
+        }
+
+        Task { ptr: self.ptr }
+    }
+}
+
+impl Drop for Task {
+    fn drop(&mut self) {
+        // Because `fetch_sub` is already atomic, we do not need to synchronize
+        // with other threads unless we are going to delete the object. This
+        // same logic applies to the below `fetch_sub` to the `weak` count.
+        if self.inner().ref_count.fetch_sub(1, Release) != 1 {
+            return;
+        }
+
+        // This fence is needed to prevent reordering of use of the data and
+        // deletion of the data.  Because it is marked `Release`, the decreasing
+        // of the reference count synchronizes with this `Acquire` fence. This
+        // means that use of the data happens before decreasing the reference
+        // count, which happens before this fence, which happens before the
+        // deletion of the data.
+        //
+        // As explained in the [Boost documentation][1],
+        //
+        // > It is important to enforce any possible access to the object in one
+        // > thread (through an existing reference) to *happen before* deleting
+        // > the object in a different thread. This is achieved by a "release"
+        // > operation after dropping a reference (any access to the object
+        // > through this reference must obviously happened before), and an
+        // > "acquire" operation before deleting the object.
+        //
+        // [1]: (www.boost.org/doc/libs/1_55_0/doc/html/atomic/usage_examples.html)
+        atomic::fence(Acquire);
+
+        unsafe {
+            let _ = Box::from_raw(self.ptr);
+        }
+    }
+}
+
+unsafe impl Send for Task {}
+
+// ===== impl Inner =====
+
+impl Inner {
+    fn stub() -> Inner {
+        Inner {
+            next: AtomicPtr::new(ptr::null_mut()),
+            state: AtomicUsize::new(State::stub().into()),
+            ref_count: AtomicUsize::new(0),
+            future: Some(executor::spawn(Box::new(future::empty()))),
+        }
+    }
+
+    fn drop_future(&mut self) {
+        let _ = self.future.take();
+    }
+}
+
+impl Drop for Inner {
+    fn drop(&mut self) {
+        self.drop_future();
+    }
+}
+
+impl fmt::Debug for Inner {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        fmt.debug_struct("Inner")
+            .field("next", &self.next)
+            .field("state", &self.state)
+            .field("ref_count", &self.ref_count)
+            .field("future", &"Spawn<BoxFuture>")
+            .finish()
+    }
+}
+
+// ===== impl Queue =====
+
+impl Queue {
+    pub fn new() -> Queue {
+        let stub = Box::new(Inner::stub());
+        let ptr = &*stub as *const _ as *mut _;
+
+        Queue {
+            head: AtomicPtr::new(ptr),
+            tail: Cell::new(ptr),
+            stub: stub,
+        }
+    }
+
+    pub fn push(&self, handle: Task) {
+        unsafe {
+            self.push2(handle.ptr);
+
+            // Forgetting the handle is necessary to avoid the ref dec
+            mem::forget(handle);
+        }
+    }
+
+    unsafe fn push2(&self, handle: *mut Inner) {
+        // Set the next pointer. This does not require an atomic operation as
+        // this node is not accessible. The write will be flushed with the next
+        // operation
+        (*handle).next = AtomicPtr::new(ptr::null_mut());
+
+        // Update the head to point to the new node. We need to see the previous
+        // node in order to update the next pointer as well as release `handle`
+        // to any other threads calling `push`.
+        let prev = self.head.swap(handle, AcqRel);
+
+        // Release `handle` to the consume end.
+        (*prev).next.store(handle, Release);
+    }
+
+    pub unsafe fn poll(&self) -> Poll {
+        let mut tail = self.tail.get();
+        let mut next = (*tail).next.load(Acquire);
+        let stub = &*self.stub as *const _ as *mut _;
+
+        if tail == stub {
+            if next.is_null() {
+                return Poll::Empty;
+            }
+
+            self.tail.set(next);
+            tail = next;
+            next = (*next).next.load(Acquire);
+        }
+
+        if !next.is_null() {
+            self.tail.set(next);
+
+            // No ref_count inc is necessary here as this poll is paired
+            // with a `push` which "forgets" the handle.
+            return Poll::Data(Task {
+                ptr: tail,
+            });
+        }
+
+        if self.head.load(Acquire) != tail {
+            return Poll::Inconsistent;
+        }
+
+        self.push2(stub);
+
+        next = (*tail).next.load(Acquire);
+
+        if !next.is_null() {
+            self.tail.set(next);
+            return Poll::Data(Task {
+                ptr: tail,
+            });
+        }
+
+        Poll::Inconsistent
+    }
+}
+
+// ===== impl State =====
+
+impl State {
+    /// Returns the initial task state.
+    ///
+    /// Tasks start in the scheduled state as they are immediately scheduled on
+    /// creation.
+    fn new() -> State {
+        State::Scheduled
+    }
+
+    fn stub() -> State {
+        State::Idle
+    }
+}
+
+impl From<usize> for State {
+    fn from(src: usize) -> Self {
+        use self::State::*;
+
+        match src {
+            0 => Idle,
+            1 => Running,
+            2 => Notified,
+            3 => Scheduled,
+            4 => Complete,
+            _ => unreachable!(),
+        }
+    }
+}
+
+impl From<State> for usize {
+    fn from(src: State) -> Self {
+        use self::State::*;
+
+        match src {
+            Idle => 0,
+            Running => 1,
+            Notified => 2,
+            Scheduled => 3,
+            Complete => 4,
+        }
+    }
+}

tokio-threadpool/tests/threadpool.rs

@@ -0,0 +1,331 @@
+extern crate tokio_threadpool;
+extern crate tokio_executor;
+extern crate futures;
+extern crate env_logger;
+
+use tokio_threadpool::*;
+use futures::{Poll, Sink, Stream, Async};
+use futures::future::{Future, lazy};
+
+use std::cell::Cell;
+use std::sync::{mpsc, Arc};
+use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT};
+use std::sync::atomic::Ordering::Relaxed;
+use std::time::Duration;
+
+thread_local!(static FOO: Cell<u32> = Cell::new(0));
+
+#[test]
+fn natural_shutdown_simple_futures() {
+    let _ = ::env_logger::init();
+
+    for _ in 0..1_000 {
+        static NUM_INC: AtomicUsize = ATOMIC_USIZE_INIT;
+        static NUM_DEC: AtomicUsize = ATOMIC_USIZE_INIT;
+
+        FOO.with(|f| {
+            f.set(1);
+
+            let pool = Builder::new()
+                .around_worker(|w, _| {
+                    NUM_INC.fetch_add(1, Relaxed);
+                    w.run();
+                    NUM_DEC.fetch_add(1, Relaxed);
+                })
+                .build();
+            let tx = pool.sender().clone();
+
+            let a = {
+                let (t, rx) = mpsc::channel();
+                tx.spawn(lazy(move || {
+                    // Makes sure this runs on a worker thread
+                    FOO.with(|f| assert_eq!(f.get(), 0));
+
+                    t.send("one").unwrap();
+                    Ok(())
+                })).unwrap();
+                rx
+            };
+
+            let b = {
+                let (t, rx) = mpsc::channel();
+                tx.spawn(lazy(move || {
+                    // Makes sure this runs on a worker thread
+                    FOO.with(|f| assert_eq!(f.get(), 0));
+
+                    t.send("two").unwrap();
+                    Ok(())
+                })).unwrap();
+                rx
+            };
+
+            drop(tx);
+
+            assert_eq!("one", a.recv().unwrap());
+            assert_eq!("two", b.recv().unwrap());
+
+            // Wait for the pool to shutdown
+            pool.shutdown().wait().unwrap();
+
+            // Assert that at least one thread started
+            let num_inc = NUM_INC.load(Relaxed);
+            assert!(num_inc > 0);
+
+            // Assert that all threads shutdown
+            let num_dec = NUM_DEC.load(Relaxed);
+            assert_eq!(num_inc, num_dec);
+        });
+    }
+}
+
+#[test]
+fn force_shutdown_drops_futures() {
+    let _ = ::env_logger::init();
+
+    for _ in 0..1_000 {
+        let num_inc = Arc::new(AtomicUsize::new(0));
+        let num_dec = Arc::new(AtomicUsize::new(0));
+        let num_drop = Arc::new(AtomicUsize::new(0));
+
+        struct Never(Arc<AtomicUsize>);
+
+        impl Future for Never {
+            type Item = ();
+            type Error = ();
+
+            fn poll(&mut self) -> Poll<(), ()> {
+                Ok(Async::NotReady)
+            }
+        }
+
+        impl Drop for Never {
+            fn drop(&mut self) {
+                self.0.fetch_add(1, Relaxed);
+            }
+        }
+
+        let a = num_inc.clone();
+        let b = num_dec.clone();
+
+        let mut pool = Builder::new()
+            .around_worker(move |w, _| {
+                a.fetch_add(1, Relaxed);
+                w.run();
+                b.fetch_add(1, Relaxed);
+            })
+            .build();
+        let mut tx = pool.sender().clone();
+
+        tx.spawn(Never(num_drop.clone())).unwrap();
+
+        // Wait for the pool to shutdown
+        pool.shutdown_now().wait().unwrap();
+
+        // Assert that only a single thread was spawned.
+        let a = num_inc.load(Relaxed);
+        assert!(a >= 1);
+
+        // Assert that all threads shutdown
+        let b = num_dec.load(Relaxed);
+        assert_eq!(a, b);
+
+        // Assert that the future was dropped
+        let c = num_drop.load(Relaxed);
+        assert_eq!(c, 1);
+    }
+}
+
+#[test]
+fn thread_shutdown_timeout() {
+    use std::sync::Mutex;
+
+    let _ = ::env_logger::init();
+
+    let (shutdown_tx, shutdown_rx) = mpsc::channel();
+    let (complete_tx, complete_rx) = mpsc::channel();
+
+    let t = Mutex::new(shutdown_tx);
+
+    let pool = Builder::new()
+        .keep_alive(Some(Duration::from_millis(200)))
+        .around_worker(move |w, _| {
+            w.run();
+            // There could be multiple threads here
+            let _ = t.lock().unwrap().send(());
+        })
+        .build();
+    let tx = pool.sender().clone();
+
+    let t = complete_tx.clone();
+    tx.spawn(lazy(move || {
+        t.send(()).unwrap();
+        Ok(())
+    })).unwrap();
+
+    // The future completes
+    complete_rx.recv().unwrap();
+
+    // The thread shuts down eventually
+    shutdown_rx.recv().unwrap();
+
+    // Futures can still be run
+    tx.spawn(lazy(move || {
+        complete_tx.send(()).unwrap();
+        Ok(())
+    })).unwrap();
+
+    complete_rx.recv().unwrap();
+
+    pool.shutdown().wait().unwrap();
+}
+
+#[test]
+fn many_oneshot_futures() {
+    const NUM: usize = 10_000;
+
+    let _ = ::env_logger::init();
+
+    for _ in 0..50 {
+        let pool = ThreadPool::new();
+        let mut tx = pool.sender().clone();
+        let cnt = Arc::new(AtomicUsize::new(0));
+
+        for _ in 0..NUM {
+            let cnt = cnt.clone();
+            tx.spawn(lazy(move || {
+                cnt.fetch_add(1, Relaxed);
+                Ok(())
+            })).unwrap();
+        }
+
+        // Wait for the pool to shutdown
+        pool.shutdown().wait().unwrap();
+
+        let num = cnt.load(Relaxed);
+        assert_eq!(num, NUM);
+    }
+}
+
+#[test]
+fn many_multishot_futures() {
+    use futures::sync::mpsc;
+
+    const CHAIN: usize = 200;
+    const CYCLES: usize = 5;
+    const TRACKS: usize = 50;
+
+    let _ = ::env_logger::init();
+
+    for _ in 0..50 {
+        let pool = ThreadPool::new();
+        let mut pool_tx = pool.sender().clone();
+
+        let mut start_txs = Vec::with_capacity(TRACKS);
+        let mut final_rxs = Vec::with_capacity(TRACKS);
+
+        for _ in 0..TRACKS {
+            let (start_tx, mut chain_rx) = mpsc::channel(10);
+
+            for _ in 0..CHAIN {
+                let (next_tx, next_rx) = mpsc::channel(10);
+
+                let rx = chain_rx
+                    .map_err(|e| panic!("{:?}", e));
+
+                // Forward all the messages
+                pool_tx.spawn(next_tx
+                    .send_all(rx)
+                    .map(|_| ())
+                    .map_err(|e| panic!("{:?}", e))
+                ).unwrap();
+
+                chain_rx = next_rx;
+            }
+
+            // This final task cycles if needed
+            let (final_tx, final_rx) = mpsc::channel(10);
+            let cycle_tx = start_tx.clone();
+            let mut rem = CYCLES;
+
+            pool_tx.spawn(chain_rx.take(CYCLES as u64).for_each(move |msg| {
+                rem -= 1;
+                let send = if rem == 0 {
+                    final_tx.clone().send(msg)
+                } else {
+                    cycle_tx.clone().send(msg)
+                };
+
+                send.then(|res| {
+                    res.unwrap();
+                    Ok(())
+                })
+            })).unwrap();
+
+            start_txs.push(start_tx);
+            final_rxs.push(final_rx);
+        }
+
+        for start_tx in start_txs {
+            start_tx.send("ping").wait().unwrap();
+        }
+
+        for final_rx in final_rxs {
+            final_rx.wait().next().unwrap().unwrap();
+        }
+
+        // Shutdown the pool
+        pool.shutdown().wait().unwrap();
+    }
+}
+
+#[test]
+fn global_executor_is_configured() {
+    let pool = ThreadPool::new();
+    let tx = pool.sender().clone();
+
+    let (signal_tx, signal_rx) = mpsc::channel();
+
+    tx.spawn(lazy(move || {
+        tokio_executor::spawn(lazy(move || {
+            signal_tx.send(()).unwrap();
+            Ok(())
+        }));
+
+        Ok(())
+    })).unwrap();
+
+    signal_rx.recv().unwrap();
+
+    pool.shutdown().wait().unwrap();
+}
+
+#[test]
+fn new_threadpool_is_idle() {
+    let pool = ThreadPool::new();
+    pool.shutdown_on_idle().wait().unwrap();
+}
+
+#[test]
+fn busy_threadpool_is_not_idle() {
+    use futures::sync::oneshot;
+
+    let pool = ThreadPool::new();
+    let tx = pool.sender().clone();
+
+    let (term_tx, term_rx) = oneshot::channel();
+
+    tx.spawn(term_rx.then(|_| {
+        Ok(())
+    })).unwrap();
+
+    let mut idle = pool.shutdown_on_idle();
+
+    futures::lazy(|| {
+        assert!(idle.poll().unwrap().is_not_ready());
+        Ok::<_, ()>(())
+    }).wait().unwrap();
+
+    term_tx.send(()).unwrap();
+
+    idle.wait().unwrap();
+}