I/O resources lazily bind to reactor. (#160)

This patch makes a significant change to how I/O resources bind to a
reactor. Currently, an I/O resource (TCP, UDP, PollEvented) will bind
itself with a reactor upon creation.

First, some history.

Originally, tokio-core required that I/O resources be explicitly
associated with a reactor upon creation by passing in a `&Handle`. Tokio
reform introduced a default reactor. If I/O resources do not specify a
reactor upon creation, then the default reactor is used.

However, futures tend to favor being lazy. Creating a future should do
no work, instead it is defining a computation to be performed once the
future is executed. Binding an I/O resource with a reactor on creation
goes against this pattern.

This patch fixes this by allowing I/O resources to lazily bind to a
reactor. An explicit `&Handle` can still be used on creation, but if no
reactor is specified, then the default reactor is used. However, this
binding happens during execution time (read / write) and not creation.

src/reactor/background.rs

@@ -4,9 +4,10 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering::SeqCst;
 
+use atomic_task::AtomicTask;
+
 use reactor::{Reactor, Handle};
 use futures::{Future, Async, Poll};
-use futures::task::AtomicTask;
 
 /// Handle to the reactor running on a background thread.
 #[derive(Debug)]
@@ -117,6 +118,8 @@ impl Drop for Background {
             None => return,
         };
 
+        inner.shutdown_now();
+
         let shutdown = Shutdown { inner };
         let _ = shutdown.wait();
     }

src/reactor/mod.rs

@@ -19,6 +19,8 @@
 use tokio_executor::Enter;
 use tokio_executor::park::{Park, Unpark};
 
+use atomic_task::AtomicTask;
+
 use std::{fmt, usize};
 use std::io::{self, ErrorKind};
 use std::mem;
@@ -29,17 +31,24 @@ use std::sync::{Arc, Weak, RwLock};
 use std::time::{Duration, Instant};
 
 use log::Level;
-use futures::task::AtomicTask;
 use mio;
 use mio::event::Evented;
 use slab::Slab;
+use futures::task::Task;
 
 pub(crate) mod background;
 use self::background::Background;
 
 mod poll_evented;
+#[allow(deprecated)]
 pub use self::poll_evented::PollEvented;
 
+mod registration;
+pub use self::registration::Registration;
+
+mod poll_evented2;
+pub use self::poll_evented2::PollEvented as PollEvented2;
+
 /// The core reactor, or event loop.
 ///
 /// The event loop is the main source of blocking in an application which drives
@@ -100,7 +109,8 @@ struct ScheduledIo {
     writer: AtomicTask,
 }
 
-enum Direction {
+#[derive(Debug, Eq, PartialEq, Clone, Copy)]
+pub(crate) enum Direction {
     Read,
     Write,
 }
@@ -358,11 +368,24 @@ impl fmt::Debug for Reactor {
 impl Handle {
     /// Returns a handle to the current reactor.
     pub fn current() -> Handle {
-        Handle::default()
+        Handle::try_current()
+            .unwrap_or(Handle { inner: Weak::new() })
+    }
+
+    /// Try to get a handle to the current reactor.
+    ///
+    /// Returns `Err` if no handle is found.
+    pub(crate) fn try_current() -> io::Result<Handle> {
+        CURRENT_REACTOR.with(|current| {
+            match *current.borrow() {
+                Some(ref handle) => Ok(handle.clone()),
+                None => Handle::fallback(),
+            }
+        })
     }
 
     /// Returns a handle to the fallback reactor.
-    fn fallback() -> Handle {
+    fn fallback() -> io::Result<Handle> {
         let mut fallback = HANDLE_FALLBACK.load(SeqCst);
 
         // If the fallback hasn't been previously initialized then let's spin
@@ -373,7 +396,8 @@ impl Handle {
         if fallback == 0 {
             let reactor = match Reactor::new() {
                 Ok(reactor) => reactor,
-                Err(_) => return Handle { inner: Weak::new() },
+                Err(_) => return Err(io::Error::new(io::ErrorKind::Other,
+                                                    "failed to create reactor")),
             };
 
             // If we successfully set ourselves as the actual fallback then we
@@ -392,7 +416,7 @@ impl Handle {
                     Err(_) => {}
                 }
 
-                return ret
+                return Ok(ret);
             }
 
             fallback = HANDLE_FALLBACK.load(SeqCst);
@@ -403,12 +427,14 @@ impl Handle {
         // handle as we don't actually have an owning reference to it.
         assert!(fallback != 0);
 
-        unsafe {
+        let ret = unsafe {
             let handle = Handle::from_usize(fallback);
             let ret = handle.clone();
             drop(handle.into_usize());
-            return ret
-        }
+            ret
+        };
+
+        Ok(ret)
     }
 
     /// Forces a reactor blocked in a call to `turn` to wakeup, or otherwise
@@ -450,12 +476,7 @@ impl Unpark for Handle {
 
 impl Default for Handle {
     fn default() -> Handle {
-        CURRENT_REACTOR.with(|current| {
-            match *current.borrow() {
-                Some(ref handle) => handle.clone(),
-                None => Handle::fallback(),
-            }
-        })
+        Handle::current()
     }
 }
 
@@ -490,7 +511,8 @@ impl Inner {
         let mut io_dispatch = self.io_dispatch.write().unwrap();
 
         if io_dispatch.len() == MAX_SOURCES {
-            return Err(io::Error::new(io::ErrorKind::Other, "reactor at max registered I/O resources"));
+            return Err(io::Error::new(io::ErrorKind::Other, "reactor at max \
+                                      registered I/O resources"));
         }
 
         // Acquire a write lock
@@ -520,7 +542,7 @@ impl Inner {
     }
 
     /// Registers interest in the I/O resource associated with `token`.
-    fn schedule(&self, token: usize, dir: Direction) {
+    fn register(&self, token: usize, dir: Direction, t: Task) {
         debug!("scheduling direction for: {}", token);
         let io_dispatch = self.io_dispatch.read().unwrap();
         let sched = io_dispatch.get(token).unwrap();
@@ -530,7 +552,7 @@ impl Inner {
             Direction::Write => (&sched.writer, mio::Ready::writable()),
         };
 
-        task.register();
+        task.register_task(t);
 
         if sched.readiness.load(SeqCst) & ready2usize(ready) != 0 {
             task.notify();
@@ -551,14 +573,33 @@ impl Drop for Inner {
     }
 }
 
+impl Direction {
+    fn ready(&self) -> mio::Ready {
+        match *self {
+            Direction::Read => read_ready(),
+            Direction::Write => write_ready(),
+        }
+    }
+
+    fn mask(&self) -> usize {
+        ready2usize(self.ready())
+    }
+}
+
 // ===== misc =====
 
+const READ: usize = 1 << 0;
+const WRITE: usize = 1 << 1;
+
 fn read_ready() -> mio::Ready {
     mio::Ready::readable() | platform::hup()
 }
 
-const READ: usize = 1 << 0;
-const WRITE: usize = 1 << 1;
+fn write_ready() -> mio::Ready {
+    mio::Ready::writable()
+}
+
+// === legacy
 
 fn ready2usize(ready: mio::Ready) -> usize {
     let mut bits = 0;

src/reactor/poll_evented.rs

@@ -6,11 +6,13 @@
 //! acquisition of a token, and tracking of the readiness state on the
 //! underlying I/O primitive.
 
+#![allow(deprecated)]
+
 use std::fmt;
 use std::io::{self, Read, Write};
 use std::sync::atomic::Ordering;
 
-use futures::{Async, Poll};
+use futures::{task, Async, Poll};
 use mio::event::Evented;
 use mio::Ready;
 use tokio_io::{AsyncRead, AsyncWrite};
@@ -68,6 +70,8 @@ struct Registration {
 /// Essentially a good rule of thumb is that if you're using the `poll_ready`
 /// method you want to also use `need_read` to signal blocking and you should
 /// otherwise probably avoid using two tasks on the same `PollEvented`.
+#[deprecated(since = "0.1.2", note = "PollEvented2 instead")]
+#[doc(hidden)]
 pub struct PollEvented<E> {
     registration: Registration,
     io: E,
@@ -233,12 +237,7 @@ impl<E> PollEvented<E> {
         let bits = super::ready2usize(super::read_ready());
         self.registration.readiness &= !bits;
 
-        let inner = match self.registration.handle.inner() {
-            Some(inner) => inner,
-            None => return Err(io::Error::new(io::ErrorKind::Other, "reactor gone")),
-        };
-        inner.schedule(self.registration.token, Direction::Read);
-        Ok(())
+        self.register(Direction::Read)
     }
 
     /// Indicates to this source of events that the corresponding I/O object is
@@ -273,12 +272,7 @@ impl<E> PollEvented<E> {
         let bits = super::ready2usize(Ready::writable());
         self.registration.readiness &= !bits;
 
-        let inner = match self.registration.handle.inner() {
-            Some(inner) => inner,
-            None => return Err(io::Error::new(io::ErrorKind::Other, "reactor gone")),
-        };
-        inner.schedule(self.registration.token, Direction::Write);
-        Ok(())
+        self.register(Direction::Write)
     }
 
     /// Returns a reference to the event loop handle that this readiness stream
@@ -326,6 +320,16 @@ impl<E> PollEvented<E> {
 
         inner.deregister_source(&self.io)
     }
+
+    fn register(&self, dir: Direction) -> io::Result<()> {
+        let inner = match self.registration.handle.inner() {
+            Some(inner) => inner,
+            None => return Err(io::Error::new(io::ErrorKind::Other, "reactor gone")),
+        };
+
+        inner.register(self.registration.token, dir, task::current());
+        Ok(())
+    }
 }
 
 impl<E: Read> Read for PollEvented<E> {

src/reactor/poll_evented2.rs

@@ -0,0 +1,419 @@
+//!
+//! Readiness tracking streams, backing I/O objects.
+//!
+//! This module contains the core type which is used to back all I/O on object
+//! in `tokio-core`. The `PollEvented` type is the implementation detail of
+//! all I/O. Each `PollEvented` manages registration with a reactor,
+//! acquisition of a token, and tracking of the readiness state on the
+//! underlying I/O primitive.
+
+#![allow(warnings)]
+
+use reactor::Handle;
+use reactor::registration::Registration;
+
+use futures::{task, Async, Poll};
+use mio;
+use mio::event::Evented;
+use tokio_io::{AsyncRead, AsyncWrite};
+
+use std::fmt;
+use std::io::{self, Read, Write};
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering::Relaxed;
+
+/// A concrete implementation of a stream of readiness notifications for I/O
+/// objects that originates from an event loop.
+///
+/// Created by the `PollEvented::new` method, each `PollEvented` is
+/// associated with a specific event loop and source of events that will be
+/// registered with an event loop.
+///
+/// An instance of `PollEvented` is essentially the bridge between the `mio`
+/// world and the `tokio-core` world, providing abstractions to receive
+/// notifications about changes to an object's `mio::Ready` state.
+///
+/// Each readiness stream has a number of methods to test whether the underlying
+/// object is readable or writable. Once the methods return that an object is
+/// readable/writable, then it will continue to do so until the `need_read` or
+/// `need_write` methods are called.
+///
+/// That is, this object is typically wrapped in another form of I/O object.
+/// It's the responsibility of the wrapper to inform the readiness stream when a
+/// "would block" I/O event is seen. The readiness stream will then take care of
+/// any scheduling necessary to get notified when the event is ready again.
+///
+/// You can find more information about creating a custom I/O object [online].
+///
+/// [online]: https://tokio.rs/docs/going-deeper-tokio/core-low-level/#custom-io
+///
+/// ## Readiness to read/write
+///
+/// A `PollEvented` allows listening and waiting for an arbitrary `mio::Ready`
+/// instance, including the platform-specific contents of `mio::Ready`. At most
+/// two future tasks, however, can be waiting on a `PollEvented`. The
+/// `need_read` and `need_write` methods can block two separate tasks, one on
+/// reading and one on writing. Not all I/O events correspond to read/write,
+/// however!
+///
+/// To account for this a `PollEvented` gets a little interesting when working
+/// with an arbitrary instance of `mio::Ready` that may not map precisely to
+/// "write" and "read" tasks. Currently it is defined that instances of
+/// `mio::Ready` that do *not* return true from `is_writable` are all notified
+/// through `need_read`, or the read task.
+///
+/// In other words, `poll_ready` with the `mio::UnixReady::hup` event will block
+/// the read task of this `PollEvented` if the `hup` event isn't available.
+/// Essentially a good rule of thumb is that if you're using the `poll_ready`
+/// method you want to also use `need_read` to signal blocking and you should
+/// otherwise probably avoid using two tasks on the same `PollEvented`.
+pub struct PollEvented<E> {
+    io: E,
+    inner: Inner,
+}
+
+struct Inner {
+    registration: Registration,
+
+    /// Currently visible read readiness
+    read_readiness: AtomicUsize,
+
+    /// Currently visible write readiness
+    write_readiness: AtomicUsize,
+}
+
+// ===== impl PollEvented =====
+
+impl<E> PollEvented<E>
+where E: Evented
+{
+    /// Creates a new `PollEvented` associated with the default reactor.
+    pub fn new(io: E) -> PollEvented<E> {
+        PollEvented {
+            io: io,
+            inner: Inner {
+                registration: Registration::new(),
+                read_readiness: AtomicUsize::new(0),
+                write_readiness: AtomicUsize::new(0),
+            }
+        }
+    }
+
+    /// Creates a new `PollEvented` associated with the specified reactor.
+    pub fn new_with_handle(io: E, handle: &Handle) -> io::Result<Self> {
+        let ret = PollEvented::new(io);
+        ret.inner.registration.register_with(&ret.io, handle)?;
+        Ok(ret)
+    }
+
+    /// Tests to see if this source is ready to be read from or not.
+    ///
+    /// If this stream is not ready for a read then `Async::NotReady` will be
+    /// returned and the current task will be scheduled to receive a
+    /// notification when the stream is readable again. In other words, this
+    /// method is only safe to call from within the context of a future's task,
+    /// typically done in a `Future::poll` method.
+    ///
+    /// # Panics
+    ///
+    /// This function will panic if called outside the context of a future's
+    /// task.
+    pub fn poll_read_ready(&self) -> Poll<mio::Ready, io::Error> {
+        self.register()?;
+
+        // Load the cached readiness
+        match self.inner.read_readiness.load(Relaxed) {
+            0 => {}
+            mut n => {
+                // Check what's new with the reactor.
+                if let Some(ready) = self.inner.registration.take_read_ready()? {
+                    n |= super::ready2usize(ready);
+                    self.inner.read_readiness.store(n, Relaxed);
+                }
+
+                return Ok(super::usize2ready(n).into());
+            }
+        }
+
+        let ready = try_ready!(self.inner.registration.poll_read_ready());
+
+        // Cache the value
+        self.inner.read_readiness.store(super::ready2usize(ready), Relaxed);
+
+        Ok(ready.into())
+    }
+
+    /// Indicates to this source of events that the corresponding I/O object is
+    /// no longer readable, but it needs to be.
+    ///
+    /// This function, like `poll_read`, is only safe to call from the context
+    /// of a future's task (typically in a `Future::poll` implementation). It
+    /// informs this readiness stream that the underlying object is no longer
+    /// readable, typically because a "would block" error was seen.
+    ///
+    /// *All* readiness bits associated with this stream except the writable bit
+    /// will be reset when this method is called. The current task is then
+    /// scheduled to receive a notification whenever anything changes other than
+    /// the writable bit. Note that this typically just means the readable bit
+    /// is used here, but if you're using a custom I/O object for events like
+    /// hup/error this may also be relevant.
+    ///
+    /// Note that it is also only valid to call this method if `poll_read`
+    /// previously indicated that the object is readable. That is, this function
+    /// must always be paired with calls to `poll_read` previously.
+    ///
+    /// # Panics
+    ///
+    /// This function will panic if called outside the context of a future's
+    /// task.
+    pub fn need_read(&self) -> io::Result<()> {
+        self.inner.read_readiness.store(0, Relaxed);
+
+        if self.poll_read_ready()?.is_ready() {
+            // Notify the current task
+            task::current().notify();
+        }
+
+        Ok(())
+    }
+
+    /// Tests to see if this source is ready to be written to or not.
+    ///
+    /// If this stream is not ready for a write then `Async::NotReady` will be
+    /// returned and the current task will be scheduled to receive a
+    /// notification when the stream is writable again. In other words, this
+    /// method is only safe to call from within the context of a future's task,
+    /// typically done in a `Future::poll` method.
+    ///
+    /// # Panics
+    ///
+    /// This function will panic if called outside the context of a future's
+    /// task.
+    pub fn poll_write_ready(&self) -> Poll<mio::Ready, io::Error> {
+        self.register()?;
+
+        match self.inner.write_readiness.load(Relaxed) {
+            0 => {}
+            mut n => {
+                // Check what's new with the reactor.
+                if let Some(ready) = self.inner.registration.take_write_ready()? {
+                    n |= super::ready2usize(ready);
+                    self.inner.write_readiness.store(n, Relaxed);
+                }
+
+                return Ok(super::usize2ready(n).into());
+            }
+        }
+
+        let ready = try_ready!(self.inner.registration.poll_write_ready());
+
+        // Cache the value
+        self.inner.write_readiness.store(super::ready2usize(ready), Relaxed);
+
+        Ok(ready.into())
+    }
+
+    /// Indicates to this source of events that the corresponding I/O object is
+    /// no longer writable, but it needs to be.
+    ///
+    /// This function, like `poll_write_ready`, is only safe to call from the
+    /// context of a future's task (typically in a `Future::poll`
+    /// implementation). It informs this readiness stream that the underlying
+    /// object is no longer writable, typically because a "would block" error
+    /// was seen.
+    ///
+    /// The flag indicating that this stream is writable is unset and the
+    /// current task is scheduled to receive a notification when the stream is
+    /// then again writable.
+    ///
+    /// Note that it is also only valid to call this method if
+    /// `poll_write_ready` previously indicated that the object is writable.
+    /// That is, this function must always be paired with calls to `poll_write`
+    /// previously.
+    ///
+    /// # Panics
+    ///
+    /// This function will panic if called outside the context of a future's
+    /// task.
+    pub fn need_write(&self) -> io::Result<()> {
+        self.inner.write_readiness.store(0, Relaxed);
+
+        if self.poll_write_ready()?.is_ready() {
+            // Notify the current task
+            task::current().notify();
+        }
+
+        Ok(())
+    }
+
+    /// Ensure that the I/O resource is registered with the reactor.
+    fn register(&self) -> io::Result<()> {
+        self.inner.registration.register(&self.io)?;
+        Ok(())
+    }
+}
+
+impl<E> PollEvented<E> {
+    /// Returns a shared reference to the underlying I/O object this readiness
+    /// stream is wrapping.
+    pub fn get_ref(&self) -> &E {
+        &self.io
+    }
+
+    /// Returns a mutable reference to the underlying I/O object this readiness
+    /// stream is wrapping.
+    pub fn get_mut(&mut self) -> &mut E {
+        &mut self.io
+    }
+
+    /// Consumes self, returning the inner I/O object
+    pub fn into_inner(self) -> E {
+        self.io
+    }
+}
+
+// ===== Read / Write impls =====
+
+impl<E> Read for PollEvented<E>
+where E: Evented + Read,
+{
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        if let Async::NotReady = self.poll_read_ready()? {
+            return Err(io::ErrorKind::WouldBlock.into())
+        }
+
+        let r = self.get_mut().read(buf);
+
+        if is_wouldblock(&r) {
+            self.need_read()?;
+        }
+
+        return r
+    }
+}
+
+impl<E> Write for PollEvented<E>
+where E: Evented + Write,
+{
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        if let Async::NotReady = self.poll_write_ready()? {
+            return Err(io::ErrorKind::WouldBlock.into())
+        }
+
+        let r = self.get_mut().write(buf);
+
+        if is_wouldblock(&r) {
+            self.need_write()?;
+        }
+
+        return r
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        if let Async::NotReady = self.poll_write_ready()? {
+            return Err(io::ErrorKind::WouldBlock.into())
+        }
+
+        let r = self.get_mut().flush();
+
+        if is_wouldblock(&r) {
+            self.need_write()?;
+        }
+
+        return r
+    }
+}
+
+impl<E> AsyncRead for PollEvented<E>
+where E: Evented + Read,
+{
+}
+
+impl<E> AsyncWrite for PollEvented<E>
+where E: Evented + Write,
+{
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        Ok(().into())
+    }
+}
+
+// ===== &'a Read / &'a Write impls =====
+
+impl<'a, E> Read for &'a PollEvented<E>
+where E: Evented, &'a E: Read,
+{
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        if let Async::NotReady = self.poll_read_ready()? {
+            return Err(io::ErrorKind::WouldBlock.into())
+        }
+
+        let r = self.get_ref().read(buf);
+
+        if is_wouldblock(&r) {
+            self.need_read()?;
+        }
+
+        return r
+    }
+}
+
+impl<'a, E> Write for &'a PollEvented<E>
+where E: Evented, &'a E: Write,
+{
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        if let Async::NotReady = self.poll_write_ready()? {
+            return Err(io::ErrorKind::WouldBlock.into())
+        }
+
+        let r = self.get_ref().write(buf);
+
+        if is_wouldblock(&r) {
+            self.need_write()?;
+        }
+
+        return r
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        if let Async::NotReady = self.poll_write_ready()? {
+            return Err(io::ErrorKind::WouldBlock.into())
+        }
+
+        let r = self.get_ref().flush();
+
+        if is_wouldblock(&r) {
+            self.need_write()?;
+        }
+
+        return r
+    }
+}
+
+impl<'a, E> AsyncRead for &'a PollEvented<E>
+where E: Evented, &'a E: Read,
+{
+}
+
+impl<'a, E> AsyncWrite for &'a PollEvented<E>
+where E: Evented, &'a E: Write,
+{
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        Ok(().into())
+    }
+}
+
+fn is_wouldblock<T>(r: &io::Result<T>) -> bool {
+    match *r {
+        Ok(_) => false,
+        Err(ref e) => e.kind() == io::ErrorKind::WouldBlock,
+    }
+}
+
+
+impl<E: fmt::Debug> fmt::Debug for PollEvented<E> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_struct("PollEvented")
+         .field("io", &self.io)
+         .finish()
+    }
+}

src/reactor/registration.rs

@@ -0,0 +1,394 @@
+use reactor::{Handle, Direction};
+
+use futures::{Async, Poll};
+use futures::task::{self, Task};
+use mio::{self, Evented};
+
+use std::{io, mem, usize};
+use std::cell::UnsafeCell;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering::SeqCst;
+
+/// Handle to a reactor registration.
+///
+/// A registration represents an I/O resource registered with a Reactor such
+/// that it will receive task notifications on readiness.
+///
+/// The registration is lazily made and supports concurrent operations. This
+/// allows a `Registration` instance to be created without the reactor handle
+/// that will eventually be used to drive the resource.
+///
+/// The difficulty is due to the fact that a single registration drives two
+/// separate tasks -- A read half and a write half.
+#[derive(Debug)]
+pub struct Registration {
+    /// Stores the handle. Once set, the value is not changed.
+    ///
+    /// Setting this requires acquiring the lock from state.
+    inner: UnsafeCell<Option<Inner>>,
+
+    /// Tracks the state of the registration.
+    ///
+    /// The least significant 2 bits are used to track the lifecycle of the
+    /// registration. The rest of the `state` variable is a pointer to tasks
+    /// that must be notified once the lock is released.
+    state: AtomicUsize,
+}
+
+#[derive(Debug)]
+struct Inner {
+    handle: Handle,
+    token: usize,
+}
+
+/// Tasks waiting on readiness notifications.
+#[derive(Debug)]
+struct Node {
+    direction: Direction,
+    task: Task,
+    next: Option<Box<Node>>,
+}
+
+/// Initial state. The handle is not set and the registration is idle.
+const INIT: usize = 0;
+
+/// A thread locked the state and will associate a handle.
+const LOCKED: usize = 1;
+
+/// A handle has been associated with the registration.
+const READY: usize = 2;
+
+/// Masks the lifecycle state
+const LIFECYCLE_MASK: usize = 0b11;
+
+/// A fake token used to identify error situations
+const ERROR: usize = usize::MAX;
+
+// ===== impl Registration =====
+
+impl Registration {
+    /// Create a new `Registration`.
+    ///
+    /// This registration is not associated with a Reactor instance. Call
+    /// `register` to establish the association.
+    pub fn new() -> Registration {
+        Registration {
+            inner: UnsafeCell::new(None),
+            state: AtomicUsize::new(INIT),
+        }
+    }
+
+    /// Register the I/O resource with the default reactor.
+    ///
+    /// This function is safe to call concurrently and repeatedly. However, only
+    /// the first call will establish the registration. Subsequent calls will be
+    /// no-ops.
+    ///
+    /// If the registration happened successfully, `Ok(true)` is returned.
+    ///
+    /// If an I/O resource has previously been successfully registered,
+    /// `Ok(false)` is returned.
+    ///
+    /// If an error is encountered during registration, `Err` is returned.
+    pub fn register<T>(&self, io: &T) -> io::Result<bool>
+    where T: Evented,
+    {
+        self.register2(io, || Handle::try_current())
+    }
+
+    /// Register the I/O resource with the specified reactor.
+    ///
+    /// This function is safe to call concurrently and repeatedly. However, only
+    /// the first call will establish the registration. Subsequent calls will be
+    /// no-ops.
+    ///
+    /// If the registration happened successfully, `Ok(true)` is returned.
+    ///
+    /// If an I/O resource has previously been successfully registered,
+    /// `Ok(false)` is returned.
+    ///
+    /// If an error is encountered during registration, `Err` is returned.
+    pub fn register_with<T>(&self, io: &T, handle: &Handle) -> io::Result<bool>
+    where T: Evented,
+    {
+        self.register2(io, || Ok(handle.clone()))
+    }
+
+    fn register2<T, F>(&self, io: &T, f: F) -> io::Result<bool>
+    where T: Evented,
+          F: Fn() -> io::Result<Handle>,
+    {
+        let mut state = self.state.load(SeqCst);
+
+        loop {
+            match state {
+                INIT => {
+                    // Registration is currently not associated with a handle.
+                    // Get a handle then attempt to lock the state.
+                    let handle = f()?;
+
+                    let actual = self.state.compare_and_swap(INIT, LOCKED, SeqCst);
+
+                    if actual != state {
+                        state = actual;
+                        continue;
+                    }
+
+                    // Create the actual registration
+                    let (inner, res) = Inner::new(io, handle);
+
+                    unsafe { *self.inner.get() = Some(inner); }
+
+                    // Transition out of the locked state. This acquires the
+                    // current value, potentially having a list of tasks that
+                    // are pending readiness notifications.
+                    let actual = self.state.swap(READY, SeqCst);
+
+                    // Consume the stack of nodes.
+                    let ptr = actual & !LIFECYCLE_MASK;
+
+                    if ptr != 0 {
+                        let mut read = false;
+                        let mut write = false;
+                        let mut curr = unsafe { Box::from_raw(ptr as *mut Node) };
+
+                        let inner = unsafe { (*self.inner.get()).as_ref().unwrap() };
+
+                        loop {
+                            let node = *curr;
+                            let Node {
+                                direction,
+                                task,
+                                next,
+                            } = node;
+
+                            let flag = match direction {
+                                Direction::Read => &mut read,
+                                Direction::Write => &mut write,
+                            };
+
+                            if !*flag {
+                                *flag = true;
+
+                                inner.register(direction, task);
+                            }
+
+                            match next {
+                                Some(next) => curr = next,
+                                None => break,
+                            }
+                        }
+                    }
+
+                    return res.map(|_| true);
+                }
+                _ => return Ok(false),
+            }
+        }
+    }
+
+    /// Poll for changes in the I/O resource's read readiness.
+    pub fn poll_read_ready(&self) -> Poll<mio::Ready, io::Error> {
+        self.poll_ready(Direction::Read, true)
+            .map(|v| match v {
+                Some(v) => Async::Ready(v),
+                _ => Async::NotReady,
+            })
+    }
+
+    /// Try taking the I/O resource's read readiness.
+    ///
+    /// Unlike `poll_read_ready`, this does not register the current task for
+    /// notification.
+    pub fn take_read_ready(&self) -> io::Result<Option<mio::Ready>> {
+        self.poll_ready(Direction::Read, false)
+
+    }
+
+    /// Poll for changes in the I/O resource's write readiness.
+    pub fn poll_write_ready(&self) -> Poll<mio::Ready, io::Error> {
+        self.poll_ready(Direction::Write, true)
+            .map(|v| match v {
+                Some(v) => Async::Ready(v),
+                _ => Async::NotReady,
+            })
+    }
+
+    /// Try taking the I/O resource's write readiness.
+    ///
+    /// Unlike `poll_write_ready`, this does not register the current task for
+    /// notification.
+    pub fn take_write_ready(&self) -> io::Result<Option<mio::Ready>> {
+        self.poll_ready(Direction::Write, false)
+    }
+
+    fn poll_ready(&self, direction: Direction, notify: bool)
+        -> io::Result<Option<mio::Ready>>
+    {
+        let mut state = self.state.load(SeqCst);
+
+        // Cache the node pointer
+        let mut node = None;
+
+        loop {
+            match state {
+                INIT => {
+                    return Err(io::Error::new(io::ErrorKind::Other, "must call `register`
+                                              before poll_read_ready"));
+                }
+                READY => {
+                    let inner = unsafe { (*self.inner.get()).as_ref().unwrap() };
+                    return inner.poll_ready(direction, notify);
+                }
+                _ => {
+                    if !notify {
+                        // Skip the notification tracking junk.
+                        return Ok(None);
+                    }
+
+                    let ptr = state & !LIFECYCLE_MASK;
+
+                    // Get the node
+                    let mut n = node.take().unwrap_or_else(|| {
+                        Box::new(Node {
+                            direction,
+                            task: task::current(),
+                            next: None,
+                        })
+                    });
+
+                    n.next = if ptr == 0 {
+                        None
+                    } else {
+                        // Great care must be taken of the CAS fails
+                        Some(unsafe { Box::from_raw(ptr as *mut Node) })
+                    };
+
+                    let ptr = Box::into_raw(n);
+                    let next = ptr as usize | (state & LIFECYCLE_MASK);
+
+                    let actual = self.state.compare_and_swap(state, next, SeqCst);
+
+                    if actual != state {
+                        // Back out of the node boxing
+                        let mut n = unsafe { Box::from_raw(ptr) };
+
+                        // We don't really own this
+                        mem::forget(n.next.take());
+
+                        // Save this for next loop
+                        node = Some(n);
+
+                        state = actual;
+                        continue;
+                    }
+
+                    return Ok(None);
+                }
+            }
+        }
+    }
+}
+
+unsafe impl Send for Registration {}
+unsafe impl Sync for Registration {}
+
+// ===== impl Inner =====
+
+impl Inner {
+    fn new<T>(io: &T, handle: Handle) -> (Self, io::Result<()>)
+    where T: Evented,
+    {
+        let mut res = Ok(());
+
+        let token = match handle.inner() {
+            Some(inner) => match inner.add_source(io) {
+                Ok(token) => token,
+                Err(e) => {
+                    res = Err(e);
+                    ERROR
+                }
+            },
+            None => {
+                res = Err(io::Error::new(io::ErrorKind::Other, "event loop gone"));
+                ERROR
+            }
+        };
+
+        let inner = Inner {
+            handle,
+            token,
+        };
+
+        (inner, res)
+    }
+
+    fn register(&self, direction: Direction, task: Task) {
+        if self.token == ERROR {
+            task.notify();
+            return;
+        }
+
+        let inner = match self.handle.inner() {
+            Some(inner) => inner,
+            None => {
+                task.notify();
+                return;
+            }
+        };
+
+        inner.register(self.token, direction, task);
+    }
+
+    fn poll_ready(&self, direction: Direction, notify: bool)
+        -> io::Result<Option<mio::Ready>>
+    {
+        if self.token == ERROR {
+            return Err(io::Error::new(io::ErrorKind::Other, "failed to associate with reactor"));
+        }
+
+        let inner = match self.handle.inner() {
+            Some(inner) => inner,
+            None => return Err(io::Error::new(io::ErrorKind::Other, "reactor gone")),
+        };
+
+        let mask = direction.mask();
+
+        let io_dispatch = inner.io_dispatch.read().unwrap();
+        let sched = &io_dispatch[self.token];
+
+        let mut ready = mask & sched.readiness.fetch_and(!mask, SeqCst);
+
+        if ready == 0 && notify {
+            // Update the task info
+            match direction {
+                Direction::Read => sched.reader.register(),
+                Direction::Write => sched.writer.register(),
+            }
+
+            // Try again
+            ready = mask & sched.readiness.fetch_and(!mask, SeqCst);
+        }
+
+        if ready == 0 {
+            Ok(None)
+        } else {
+            Ok(Some(super::usize2ready(ready)))
+        }
+    }
+}
+
+impl Drop for Inner {
+    fn drop(&mut self) {
+        if self.token == ERROR {
+            return;
+        }
+
+        let inner = match self.handle.inner() {
+            Some(inner) => inner,
+            None => return,
+        };
+
+        inner.drop_source(self.token);
+    }
+}