signal: factor out event delivery into its own module to share between Unix and Windows (#1174)

Today the Unix and Windows implementations have similar yet differing
implementations of hooking into OS events and propagating them to any
listening futures. Rather than re-implement the same behavior two
different ways, we should factor out any commonality into a shared
module and keep the Unix/Windows modules focused solely on OS
integrations.

Reusing the same implementation across OS versions also allows for more
consistent behavior between platforms, which also makes squashing bugs
much easier.

This change introduces the `registry` module which handles creating and
initializing a global map of signals/events and their registered
listeners. Each OS specific module is expected to implement the OS hooks
which delegate to invoking the registry module's methods for
distributing the event notifications.

# Use registry module for Windows implementation

Note this still uses the same architecture as previously: a driver task
is spawned by the first registered event, and that task is responsible
for delivering any events to registered futures. (If that first event
loop goes away, all events will deadlock). A solution to this issue will
be explored at a later time.

tokio-signal/Cargo.toml

@@ -27,18 +27,20 @@ appveyor = { repository = "carllerche/tokio", id = "s83yxhy9qeb58va7" }
 
 [dependencies]
 futures = "0.1.11"
-mio = "0.6.14"
+lazy_static = "1"
 tokio-reactor = { version = "0.2.0", path = "../tokio-reactor" }
 tokio-executor = { version = "0.2.0", path = "../tokio-executor" }
 tokio-io = { version = "0.2.0", path = "../tokio-io" }
 
 [target.'cfg(unix)'.dependencies]
 libc = "0.2"
+mio = "0.6.14"
 mio-uds = "0.6"
 signal-hook-registry = "~1"
 
 [dev-dependencies]
 tokio = { version = "0.2.0", path = "../tokio" }
+tokio-timer = { version = "0.3.0", path = "../tokio-timer" }
 
 [target.'cfg(windows)'.dependencies.winapi]
 version = "0.3"

tokio-signal/src/lib.rs

@@ -54,11 +54,23 @@
 //! # }
 //! ```
 
+#[macro_use]
+extern crate lazy_static;
+
 use futures::stream::Stream;
 use futures::{future, Future};
 use std::io;
 use tokio_reactor::Handle;
 
+mod registry;
+
+mod os {
+    #[cfg(unix)]
+    pub(crate) use super::unix::{OsExtraData, OsStorage};
+    #[cfg(windows)]
+    pub(crate) use super::windows::{OsExtraData, OsStorage};
+}
+
 pub mod unix;
 pub mod windows;
 

tokio-signal/src/registry.rs

@@ -0,0 +1,292 @@
+use std::ops;
+use std::pin::Pin;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Mutex;
+
+use crate::os::{OsExtraData, OsStorage};
+use futures::sync::mpsc::Sender;
+
+pub(crate) type EventId = usize;
+
+/// State for a specific event, whether a notification is pending delivery,
+/// and what listeners are registered.
+#[derive(Default, Debug)]
+pub(crate) struct EventInfo {
+    pending: AtomicBool,
+    recipients: Mutex<Vec<Sender<()>>>,
+}
+
+/// An interface for retrieving the `EventInfo` for a particular eventId.
+pub(crate) trait Storage {
+    /// Get the `EventInfo` for `id` if it exists.
+    fn event_info(&self, id: EventId) -> Option<&EventInfo>;
+
+    /// Invoke `f` once for each defined `EventInfo` in this storage.
+    fn for_each<'a, F>(&'a self, f: F)
+    where
+        F: FnMut(&'a EventInfo);
+}
+
+impl Storage for Vec<EventInfo> {
+    fn event_info(&self, id: EventId) -> Option<&EventInfo> {
+        self.get(id)
+    }
+
+    fn for_each<'a, F>(&'a self, f: F)
+    where
+        F: FnMut(&'a EventInfo),
+    {
+        self.iter().for_each(f)
+    }
+}
+
+/// An interface for initializing a type. Useful for situations where we cannot
+/// inject a configured instance in the constructor of another type.
+pub(crate) trait Init {
+    fn init() -> Self;
+}
+
+/// Manages and distributes event notifications to any registered listeners.
+///
+/// Generic over the underlying storage to allow for domain specific
+/// optimizations (e.g. eventIds may or may not be contiguous).
+#[derive(Debug)]
+pub(crate) struct Registry<S> {
+    storage: S,
+}
+
+impl<S> Registry<S> {
+    fn new(storage: S) -> Self {
+        Self { storage }
+    }
+}
+
+impl<S: Storage> Registry<S> {
+    /// Register a new listener for `event_id`.
+    fn register_listener(&self, event_id: EventId, listener: Sender<()>) {
+        self.storage
+            .event_info(event_id)
+            .unwrap_or_else(|| panic!("invalid event_id: {}", event_id))
+            .recipients
+            .lock()
+            .unwrap()
+            .push(listener);
+    }
+
+    /// Mark `event_id` as having been delivered, without broadcasting it to
+    /// any listeners.
+    fn record_event(&self, event_id: EventId) {
+        self.storage
+            .event_info(event_id)
+            .map(|event_info| event_info.pending.store(true, Ordering::SeqCst));
+    }
+
+    /// Broadcast all previously recorded events to their respective listeners.
+    fn broadcast(&self) {
+        self.storage.for_each(|event_info| {
+            // Any signal of this kind arrived since we checked last?
+            if !event_info.pending.swap(false, Ordering::SeqCst) {
+                return;
+            }
+
+            let mut recipients = event_info.recipients.lock().unwrap();
+
+            // Notify all waiters on this signal that the signal has been
+            // received. If we can't push a message into the queue then we don't
+            // worry about it as everything is coalesced anyway. If the channel
+            // has gone away then we can remove that slot.
+            for i in (0..recipients.len()).rev() {
+                match recipients[i].try_send(()) {
+                    Ok(()) => {}
+                    Err(ref e) if e.is_disconnected() => {
+                        recipients.swap_remove(i);
+                    }
+
+                    // Channel is full, ignore the error since the
+                    // receiver has already been woken up
+                    Err(e) => {
+                        // Sanity check in case this error type ever gets
+                        // additional variants we have not considered.
+                        debug_assert!(e.is_full());
+                    }
+                }
+            }
+        });
+    }
+}
+
+pub(crate) struct Globals {
+    extra: OsExtraData,
+    registry: Registry<OsStorage>,
+}
+
+impl ops::Deref for Globals {
+    type Target = OsExtraData;
+
+    fn deref(&self) -> &Self::Target {
+        &self.extra
+    }
+}
+
+impl Globals {
+    /// Register a new listener for `event_id`.
+    pub(crate) fn register_listener(&self, event_id: EventId, listener: Sender<()>) {
+        self.registry.register_listener(event_id, listener);
+    }
+
+    /// Mark `event_id` as having been delivered, without broadcasting it to
+    /// any listeners.
+    pub(crate) fn record_event(&self, event_id: EventId) {
+        self.registry.record_event(event_id);
+    }
+
+    /// Broadcast all previously recorded events to their respective listeners.
+    pub(crate) fn broadcast(&self) {
+        self.registry.broadcast()
+    }
+
+    #[cfg(unix)]
+    pub(crate) fn storage(&self) -> &OsStorage {
+        &self.registry.storage
+    }
+}
+
+pub(crate) fn globals() -> Pin<&'static Globals>
+where
+    OsExtraData: 'static + Send + Sync + Init,
+    OsStorage: 'static + Send + Sync + Init,
+{
+    lazy_static! {
+        static ref GLOBALS: Pin<Box<Globals>> = Pin::new(Box::new(Globals {
+            extra: OsExtraData::init(),
+            registry: Registry::new(OsStorage::init()),
+        }));
+    }
+
+    GLOBALS.as_ref()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::sync::mpsc::channel;
+    use futures::sync::oneshot;
+    use futures::{Future, Stream};
+
+    use std::time::Duration;
+    use tokio::runtime::Runtime;
+    use tokio_timer::sleep;
+
+    #[test]
+    fn smoke() {
+        let registry = Registry::new(vec![
+            EventInfo::default(),
+            EventInfo::default(),
+            EventInfo::default(),
+        ]);
+
+        let (first_tx, first_rx) = channel(0);
+        let (second_tx, second_rx) = channel(0);
+        let (third_tx, third_rx) = channel(0);
+
+        registry.register_listener(0, first_tx);
+        registry.register_listener(1, second_tx);
+        registry.register_listener(2, third_tx);
+
+        let (fire, wait) = oneshot::channel();
+        let rt = Runtime::new().unwrap();
+
+        rt.spawn(
+            wait.and_then(move |_| {
+                // Record some events which should get coalesced
+                registry.record_event(0);
+                registry.record_event(0);
+                registry.record_event(1);
+                registry.record_event(1);
+                registry.broadcast();
+
+                sleep(Duration::from_millis(100))
+                    .map_err(|e| panic!("{:#?}", e))
+                    .and_then(move |_| {
+                        registry.record_event(0);
+                        registry.broadcast();
+
+                        drop(registry);
+                        Ok(())
+                    })
+            })
+            .map_err(|e| panic!("{}", e)),
+        );
+
+        let (first_results, second_results, third_results) = rt
+            .block_on(futures::lazy(move || {
+                let _ = fire.send(());
+
+                first_rx
+                    .collect()
+                    .join3(second_rx.collect(), third_rx.collect())
+            }))
+            .expect("failed to extract events");
+
+        assert_eq!(2, first_results.len());
+        assert_eq!(1, second_results.len());
+        assert_eq!(0, third_results.len());
+    }
+
+    #[test]
+    #[should_panic = "invalid event_id: 1"]
+    fn register_panics_on_invalid_input() {
+        let registry = Registry::new(vec![EventInfo::default()]);
+
+        let (tx, _) = channel(0);
+        registry.register_listener(1, tx);
+    }
+
+    #[test]
+    fn record_invalid_event_does_nothing() {
+        let registry = Registry::new(vec![EventInfo::default()]);
+        registry.record_event(42);
+    }
+
+    #[test]
+    fn broadcast_cleans_up_disconnected_listeners() {
+        let registry = Registry::new(vec![EventInfo::default()]);
+
+        let (first_tx, first_rx) = channel(0);
+        let (second_tx, second_rx) = channel(0);
+        let (third_tx, third_rx) = channel(0);
+
+        registry.register_listener(0, first_tx);
+        registry.register_listener(0, second_tx);
+        registry.register_listener(0, third_tx);
+
+        drop(first_rx);
+        drop(second_rx);
+
+        let (fire, wait) = oneshot::channel();
+        let rt = Runtime::new().unwrap();
+
+        rt.spawn(
+            wait.and_then(move |_| {
+                // Record some events which should get coalesced
+                registry.record_event(0);
+                registry.broadcast();
+
+                assert_eq!(1, registry.storage[0].recipients.lock().unwrap().len());
+                drop(registry);
+
+                Ok(())
+            })
+            .map_err(|e| panic!("{}", e)),
+        );
+
+        let results = rt
+            .block_on(futures::lazy(move || {
+                let _ = fire.send(());
+                third_rx.collect()
+            }))
+            .expect("failed to extract events");
+
+        assert_eq!(1, results.len());
+    }
+}

tokio-signal/src/unix.rs

@@ -9,11 +9,12 @@ pub use libc;
 
 use std::io::prelude::*;
 use std::io::{self, Error, ErrorKind};
+use std::pin::Pin;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Mutex, Once, ONCE_INIT};
+use std::sync::{Once, ONCE_INIT};
 
 use futures::future;
-use futures::sync::mpsc::{channel, Receiver, Sender};
+use futures::sync::mpsc::{channel, Receiver};
 use futures::{Async, Future};
 use futures::{Poll, Stream};
 use libc::c_int;
@@ -21,6 +22,8 @@ use mio_uds::UnixStream;
 use tokio_io::IoFuture;
 use tokio_reactor::{Handle, PollEvented};
 
+use crate::registry::{globals, EventId, EventInfo, Globals, Init, Storage};
+
 pub use libc::{SIGALRM, SIGHUP, SIGPIPE, SIGQUIT, SIGTRAP};
 pub use libc::{SIGINT, SIGTERM, SIGUSR1, SIGUSR2};
 
@@ -43,91 +46,61 @@ pub mod bsd {
     pub use super::libc::SIGINFO;
 }
 
+pub(crate) type OsStorage = Vec<SignalInfo>;
+
 // Number of different unix signals
 // (FreeBSD has 33)
 const SIGNUM: usize = 33;
 
-type SignalSender = Sender<c_int>;
-
-struct SignalInfo {
-    pending: AtomicBool,
-    // The ones interested in this signal
-    recipients: Mutex<Vec<Box<SignalSender>>>,
-
-    init: Once,
-    initialized: AtomicBool,
+impl Init for OsStorage {
+    fn init() -> Self {
+        (0..SIGNUM).map(|_| SignalInfo::default()).collect()
+    }
 }
 
-struct Globals {
+impl Storage for OsStorage {
+    fn event_info(&self, id: EventId) -> Option<&EventInfo> {
+        self.get(id).map(|si| &si.event_info)
+    }
+
+    fn for_each<'a, F>(&'a self, f: F)
+    where
+        F: FnMut(&'a EventInfo),
+    {
+        self.iter().map(|si| &si.event_info).for_each(f)
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct OsExtraData {
     sender: UnixStream,
     receiver: UnixStream,
-    signals: Vec<SignalInfo>,
 }
 
-impl Globals {
-    /// Register a new `Signal` instance's channel sender.
-    /// Returns a `SignalId` which should be later used for deregistering
-    /// this sender.
-    fn register_signal_sender(signal: c_int, tx: SignalSender) -> SignalId {
-        let tx = Box::new(tx);
-        let id = SignalId::from(&tx);
+impl Init for OsExtraData {
+    fn init() -> Self {
+        let (receiver, sender) = UnixStream::pair().expect("failed to create UnixStream");
 
-        let idx = signal as usize;
-        globals().signals[idx].recipients.lock().unwrap().push(tx);
-        id
-    }
-
-    /// Deregister a `Signal` instance's channel sender because the `Signal`
-    /// is no longer interested in receiving events (e.g. dropped).
-    fn deregister_signal_receiver(signal: c_int, id: SignalId) {
-        let idx = signal as usize;
-        let mut list = globals().signals[idx].recipients.lock().unwrap();
-        list.retain(|sender| SignalId::from(sender) != id);
+        Self { sender, receiver }
     }
 }
 
-/// A newtype which represents a unique identifier for each `Signal` instance.
-/// The id is derived by boxing the channel `Sender` associated with this instance
-/// and using its address in memory.
-#[derive(Debug, Copy, Clone, PartialEq, Eq)]
-struct SignalId(usize);
-
-impl<'a> From<&'a Box<SignalSender>> for SignalId {
-    fn from(tx: &'a Box<SignalSender>) -> Self {
-        SignalId(&**tx as *const _ as usize)
-    }
+pub(crate) struct SignalInfo {
+    event_info: EventInfo,
+    init: Once,
+    initialized: AtomicBool,
 }
 
 impl Default for SignalInfo {
     fn default() -> SignalInfo {
         SignalInfo {
-            pending: AtomicBool::new(false),
+            event_info: Default::default(),
             init: ONCE_INIT,
             initialized: AtomicBool::new(false),
-            recipients: Mutex::new(Vec::new()),
         }
     }
 }
 
-static mut GLOBALS: *mut Globals = 0 as *mut Globals;
-
-fn globals() -> &'static Globals {
-    static INIT: Once = ONCE_INIT;
-
-    unsafe {
-        INIT.call_once(|| {
-            let (receiver, sender) = UnixStream::pair().unwrap();
-            let globals = Globals {
-                sender: sender,
-                receiver: receiver,
-                signals: (0..SIGNUM).map(|_| Default::default()).collect(),
-            };
-            GLOBALS = Box::into_raw(Box::new(globals));
-        });
-        &*GLOBALS
-    }
-}
-
 /// Our global signal handler for all signals registered by this module.
 ///
 /// The purpose of this signal handler is to primarily:
@@ -136,11 +109,12 @@ fn globals() -> &'static Globals {
 /// 2. Wake up driver tasks by writing a byte to a pipe
 ///
 /// Those two operations shoudl both be async-signal safe.
-fn action(slot: &SignalInfo, mut sender: &UnixStream) {
-    slot.pending.store(true, Ordering::SeqCst);
+fn action(globals: Pin<&'static Globals>, signal: c_int) {
+    globals.record_event(signal as EventId);
 
     // Send a wakeup, ignore any errors (anything reasonably possible is
     // full pipe and then it will wake up anyway).
+    let mut sender = &globals.sender;
     drop(sender.write(&[1]));
 }
 
@@ -150,7 +124,7 @@ fn action(slot: &SignalInfo, mut sender: &UnixStream) {
 /// This will register the signal handler if it hasn't already been registered,
 /// returning any error along the way if that fails.
 fn signal_enable(signal: c_int) -> io::Result<()> {
-    if signal_hook_registry::FORBIDDEN.contains(&signal) {
+    if signal < 0 || signal_hook_registry::FORBIDDEN.contains(&signal) {
         return Err(Error::new(
             ErrorKind::Other,
             format!("Refusing to register signal {}", signal),
@@ -158,15 +132,14 @@ fn signal_enable(signal: c_int) -> io::Result<()> {
     }
 
     let globals = globals();
-    let siginfo = match globals.signals.get(signal as usize) {
+    let siginfo = match globals.storage().get(signal as EventId) {
         Some(slot) => slot,
         None => return Err(io::Error::new(io::ErrorKind::Other, "signal too large")),
     };
     let mut registered = Ok(());
     siginfo.init.call_once(|| {
         registered = unsafe {
-            signal_hook_registry::register(signal, move || action(siginfo, &globals.sender))
-                .map(|_| ())
+            signal_hook_registry::register(signal, move || action(globals, signal)).map(|_| ())
         };
         if registered.is_ok() {
             siginfo.initialized.store(true, Ordering::Relaxed);
@@ -197,7 +170,7 @@ impl Future for Driver {
         // Drain the data from the pipe and maintain interest in getting more
         self.drain();
         // Broadcast any signals which were received
-        self.broadcast();
+        globals().broadcast();
 
         // This task just lives until the end of the event loop
         Ok(Async::NotReady)
@@ -241,44 +214,6 @@ impl Driver {
             }
         }
     }
-
-    /// Go through all the signals and broadcast everything.
-    ///
-    /// Driver tasks wake up for *any* signal and simply process all globally
-    /// registered signal streams, so each task is sort of cooperatively working
-    /// for all the rest as well.
-    fn broadcast(&self) {
-        for (sig, slot) in globals().signals.iter().enumerate() {
-            // Any signal of this kind arrived since we checked last?
-            if !slot.pending.swap(false, Ordering::SeqCst) {
-                continue;
-            }
-
-            let signum = sig as c_int;
-            let mut recipients = slot.recipients.lock().unwrap();
-
-            // Notify all waiters on this signal that the signal has been
-            // received. If we can't push a message into the queue then we don't
-            // worry about it as everything is coalesced anyway. If the channel
-            // has gone away then we can remove that slot.
-            for i in (0..recipients.len()).rev() {
-                match recipients[i].try_send(signum) {
-                    Ok(()) => {}
-                    Err(ref e) if e.is_disconnected() => {
-                        recipients.swap_remove(i);
-                    }
-
-                    // Channel is full, ignore the error since the
-                    // receiver has already been woken up
-                    Err(e) => {
-                        // Sanity check in case this error type ever gets
-                        // additional variants we have not considered.
-                        debug_assert!(e.is_full());
-                    }
-                }
-            }
-        }
-    }
 }
 
 /// An implementation of `Stream` for receiving a particular type of signal.
@@ -318,8 +253,7 @@ impl Driver {
 pub struct Signal {
     driver: Driver,
     signal: c_int,
-    id: SignalId,
-    rx: Receiver<c_int>,
+    rx: Receiver<()>,
 }
 
 impl Signal {
@@ -385,11 +319,11 @@ impl Signal {
                 // more. NB: channels always guarantee at least one slot per sender,
                 // so we don't need additional slots
                 let (tx, rx) = channel(0);
-                let id = Globals::register_signal_sender(signal, tx);
+                globals().register_listener(signal as EventId, tx);
+
                 Ok(Signal {
                     driver: driver,
                     rx: rx,
-                    id: id,
                     signal: signal,
                 })
             })();
@@ -404,53 +338,26 @@ impl Stream for Signal {
 
     fn poll(&mut self) -> Poll<Option<c_int>, io::Error> {
         self.driver.poll().unwrap();
-        // receivers don't generate errors
-        self.rx.poll().map_err(|_| panic!())
-    }
-}
 
-impl Drop for Signal {
-    fn drop(&mut self) {
-        Globals::deregister_signal_receiver(self.signal, self.id);
+        self.rx
+            .poll()
+            .map(|ready| ready.map(|item| item.map(|()| self.signal)))
+            // receivers don't generate errors
+            .map_err(|_| unreachable!())
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use tokio;
-
     use super::*;
 
     #[test]
-    fn dropped_signal_senders_are_cleaned_up() {
-        let mut rt =
-            self::tokio::runtime::current_thread::Runtime::new().expect("failed to init runtime");
+    fn signal_enable_error_on_invalid_input() {
+        signal_enable(-1).unwrap_err();
+    }
 
-        let signum = libc::SIGUSR1;
-        let signal = rt
-            .block_on(Signal::new(signum))
-            .expect("failed to create signal");
-
-        {
-            let recipients = globals().signals[signum as usize]
-                .recipients
-                .lock()
-                .unwrap();
-            assert!(!recipients.is_empty());
-        }
-
-        drop(signal);
-
-        unsafe {
-            assert_eq!(libc::kill(libc::getpid(), signum), 0);
-        }
-
-        {
-            let recipients = globals().signals[signum as usize]
-                .recipients
-                .lock()
-                .unwrap();
-            assert!(recipients.is_empty());
-        }
+    #[test]
+    fn signal_enable_error_on_forbidden_input() {
+        signal_enable(signal_hook_registry::FORBIDDEN[0]).unwrap_err();
     }
 }

tokio-signal/src/windows.rs

@@ -7,26 +7,70 @@
 
 #![cfg(windows)]
 
-use std::cell::RefCell;
+use std::convert::TryFrom;
 use std::io;
-use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Once, ONCE_INIT};
 
 use futures::future;
-use futures::stream::Fuse;
-use futures::sync::mpsc;
-use futures::sync::oneshot;
+use futures::sync::mpsc::{channel, Receiver, Sender};
 use futures::{Async, Future, Poll, Stream};
-use mio::Ready;
-use tokio_reactor::{Handle, PollEvented};
+use tokio_reactor::Handle;
 use winapi::shared::minwindef::*;
 use winapi::um::consoleapi::SetConsoleCtrlHandler;
 use winapi::um::wincon::*;
 
+use crate::registry::{globals, EventId, EventInfo, Init, Storage};
 use crate::IoFuture;
 
+#[derive(Debug)]
+pub(crate) struct OsStorage {
+    ctrl_c: EventInfo,
+    ctrl_break: EventInfo,
+}
+
+impl Init for OsStorage {
+    fn init() -> Self {
+        Self {
+            ctrl_c: EventInfo::default(),
+            ctrl_break: EventInfo::default(),
+        }
+    }
+}
+
+impl Storage for OsStorage {
+    fn event_info(&self, id: EventId) -> Option<&EventInfo> {
+        match DWORD::try_from(id) {
+            Ok(CTRL_C_EVENT) => Some(&self.ctrl_c),
+            Ok(CTRL_BREAK_EVENT) => Some(&self.ctrl_break),
+            _ => None,
+        }
+    }
+
+    fn for_each<'a, F>(&'a self, mut f: F)
+    where
+        F: FnMut(&'a EventInfo),
+    {
+        f(&self.ctrl_c);
+        f(&self.ctrl_break);
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct OsExtraData {
+    driver_waker: Sender<()>,
+}
+
+impl Init for OsExtraData {
+    fn init() -> Self {
+        let (driver_waker, driver_rx) = channel(0);
+
+        ::tokio_executor::spawn(DriverTask { rx: driver_rx });
+
+        Self { driver_waker }
+    }
+}
+
 static INIT: Once = ONCE_INIT;
-static mut GLOBAL_STATE: *mut GlobalState = 0 as *mut _;
 
 /// Stream of events discovered via `SetConsoleCtrlHandler`.
 ///
@@ -41,36 +85,14 @@ static mut GLOBAL_STATE: *mut GlobalState = 0 as *mut _;
 ///   processed quickly enough. This means that if two notifications are
 ///   received back-to-back, then the stream may only receive one item about the
 ///   two notifications.
+// FIXME: refactor and combine with unix::Signal
 pub struct Event {
-    reg: PollEvented<MyRegistration>,
-    _finished: oneshot::Sender<()>,
-}
-
-struct GlobalState {
-    ready: mio::SetReadiness,
-    tx: mpsc::UnboundedSender<Message>,
-    ctrl_c: GlobalEventState,
-    ctrl_break: GlobalEventState,
-}
-
-struct GlobalEventState {
-    ready: AtomicBool,
-}
-
-enum Message {
-    NewEvent(DWORD, oneshot::Sender<io::Result<Event>>),
+    rx: Receiver<()>,
 }
 
+#[derive(Debug)]
 struct DriverTask {
-    handle: Handle,
-    reg: PollEvented<MyRegistration>,
-    rx: Fuse<mpsc::UnboundedReceiver<Message>>,
-    ctrl_c: EventState,
-    ctrl_break: EventState,
-}
-
-struct EventState {
-    tasks: Vec<(RefCell<oneshot::Receiver<()>>, mio::SetReadiness)>,
+    rx: Receiver<()>,
 }
 
 impl Event {
@@ -106,29 +128,24 @@ impl Event {
         Event::new(CTRL_BREAK_EVENT, handle)
     }
 
-    fn new(signum: DWORD, handle: &Handle) -> IoFuture<Event> {
-        let handle = handle.clone();
+    fn new(signum: DWORD, _handle: &Handle) -> IoFuture<Event> {
         let new_signal = future::poll_fn(move || {
             let mut init = None;
             INIT.call_once(|| {
-                init = Some(global_init(&handle));
+                init = Some(global_init());
             });
 
             if let Some(Err(e)) = init {
                 return Err(e);
             }
 
-            let (tx, rx) = oneshot::channel();
-            let msg = Message::NewEvent(signum, tx);
-            let res = unsafe { (*GLOBAL_STATE).tx.clone().unbounded_send(msg) };
-            res.expect(
-                "failed to request a new signal stream, did the \
-                 first event loop go away?",
-            );
-            Ok(Async::Ready(rx.then(|r| r.unwrap())))
+            let (tx, rx) = channel(0);
+            globals().register_listener(signum as EventId, tx);
+
+            Ok(Async::Ready(Event { rx }))
         });
 
-        Box::new(new_signal.flatten())
+        Box::new(new_signal)
     }
 }
 
@@ -137,54 +154,20 @@ impl Stream for Event {
     type Error = io::Error;
 
     fn poll(&mut self) -> Poll<Option<()>, io::Error> {
-        if !self.reg.poll_read_ready(Ready::readable())?.is_ready() {
-            return Ok(Async::NotReady);
-        }
-        self.reg.clear_read_ready(Ready::readable())?;
-        self.reg
-            .get_ref()
-            .readiness
-            .set_readiness(mio::Ready::empty())
-            .expect("failed to set readiness");
-        Ok(Async::Ready(Some(())))
+        self.rx
+            .poll()
+            // receivers don't generate errors
+            .map_err(|_| unreachable!())
     }
 }
 
-fn global_init(handle: &Handle) -> io::Result<()> {
-    let reg = MyRegistration::new();
-    let ready = reg.readiness.clone();
-
-    let (tx, rx) = mpsc::unbounded();
-    let reg = PollEvented::new_with_handle(reg, handle)?;
-
+fn global_init() -> io::Result<()> {
     unsafe {
-        let state = Box::new(GlobalState {
-            ready: ready,
-            ctrl_c: GlobalEventState {
-                ready: AtomicBool::new(false),
-            },
-            ctrl_break: GlobalEventState {
-                ready: AtomicBool::new(false),
-            },
-            tx: tx,
-        });
-        GLOBAL_STATE = Box::into_raw(state);
-
         let rc = SetConsoleCtrlHandler(Some(handler), TRUE);
         if rc == 0 {
-            Box::from_raw(GLOBAL_STATE);
-            GLOBAL_STATE = 0 as *mut _;
             return Err(io::Error::last_os_error());
         }
 
-        ::tokio_executor::spawn(Box::new(DriverTask {
-            handle: handle.clone(),
-            rx: rx.fuse(),
-            reg: reg,
-            ctrl_c: EventState { tasks: Vec::new() },
-            ctrl_break: EventState { tasks: Vec::new() },
-        }));
-
         Ok(())
     }
 }
@@ -194,153 +177,37 @@ impl Future for DriverTask {
     type Error = ();
 
     fn poll(&mut self) -> Poll<(), ()> {
-        self.check_event_drops();
-        self.check_messages();
-        self.check_events().unwrap();
+        loop {
+            // Ensure we keep polling our waker until we know there are no more
+            // events (and therefore we've registered interest to be woken again).
+            match self.rx.poll() {
+                Ok(Async::Ready(Some(()))) => continue,
+                Ok(Async::Ready(None)) => panic!("driver got disconnected?"),
+                Ok(Async::NotReady) => break,
+                // receivers don't generate errors
+                Err(()) => unreachable!(),
+            }
+        }
 
-        // TODO: when to finish this task?
+        globals().broadcast();
+
+        // TODO(1000): when to finish this task?
         Ok(Async::NotReady)
     }
 }
 
-impl DriverTask {
-    fn check_event_drops(&mut self) {
-        self.ctrl_c
-            .tasks
-            .retain(|task| !task.0.borrow_mut().poll().is_err());
-        self.ctrl_break
-            .tasks
-            .retain(|task| !task.0.borrow_mut().poll().is_err());
-    }
-
-    fn check_messages(&mut self) {
-        loop {
-            // Acquire the next message
-            let message = match self.rx.poll().unwrap() {
-                Async::Ready(Some(e)) => e,
-                Async::Ready(None) | Async::NotReady => break,
-            };
-            let (sig, complete) = match message {
-                Message::NewEvent(sig, complete) => (sig, complete),
-            };
-
-            let event = if sig == CTRL_C_EVENT {
-                &mut self.ctrl_c
-            } else {
-                &mut self.ctrl_break
-            };
-
-            // Acquire the (registration, set_readiness) pair by... assuming
-            // we're on the event loop (true because of the spawn above).
-            let reg = MyRegistration::new();
-            let ready = reg.readiness.clone();
-
-            let reg = match PollEvented::new_with_handle(reg, &self.handle) {
-                Ok(reg) => reg,
-                Err(e) => {
-                    drop(complete.send(Err(e)));
-                    continue;
-                }
-            };
-
-            // Create the `Event` to pass back and then also keep a handle to
-            // the `SetReadiness` for ourselves internally.
-            let (tx, rx) = oneshot::channel();
-            drop(complete.send(Ok(Event {
-                reg: reg,
-                _finished: tx,
-            })));
-            event.tasks.push((RefCell::new(rx), ready));
-        }
-    }
-
-    fn check_events(&mut self) -> io::Result<()> {
-        if self.reg.poll_read_ready(Ready::readable())?.is_not_ready() {
-            return Ok(());
-        }
-        self.reg.clear_read_ready(Ready::readable())?;
-        self.reg
-            .get_ref()
-            .readiness
-            .set_readiness(mio::Ready::empty())
-            .expect("failed to set readiness");
-
-        if unsafe { (*GLOBAL_STATE).ctrl_c.ready.swap(false, Ordering::SeqCst) } {
-            for task in self.ctrl_c.tasks.iter() {
-                task.1.set_readiness(mio::Ready::readable()).unwrap();
-            }
-        }
-        if unsafe {
-            (*GLOBAL_STATE)
-                .ctrl_break
-                .ready
-                .swap(false, Ordering::SeqCst)
-        } {
-            for task in self.ctrl_break.tasks.iter() {
-                task.1.set_readiness(mio::Ready::readable()).unwrap();
-            }
-        }
-        Ok(())
-    }
-}
-
 unsafe extern "system" fn handler(ty: DWORD) -> BOOL {
-    let event = match ty {
-        CTRL_C_EVENT => &(*GLOBAL_STATE).ctrl_c,
-        CTRL_BREAK_EVENT => &(*GLOBAL_STATE).ctrl_break,
-        _ => return FALSE,
-    };
-    if event.ready.swap(true, Ordering::SeqCst) {
-        FALSE
-    } else {
-        drop((*GLOBAL_STATE).ready.set_readiness(mio::Ready::readable()));
-        // TODO(1000): this will report that we handled a CTRL_BREAK_EVENT when
-        //       in fact we may not have any streams actually created for that
-        //       event.
-        TRUE
-    }
-}
+    let globals = globals();
+    globals.record_event(ty as EventId);
 
-struct MyRegistration {
-    registration: mio::Registration,
-    readiness: mio::SetReadiness,
-}
+    // FIXME: revisit this, we'd probably want to panic if the driver task goes away,
+    // but that would unwind across the FFI boundary...
+    let _ = globals.driver_waker.clone().try_send(());
 
-impl MyRegistration {
-    fn new() -> Self {
-        let (registration, readiness) = mio::Registration::new2();
-
-        Self {
-            registration,
-            readiness,
-        }
-    }
-}
-
-impl mio::Evented for MyRegistration {
-    fn register(
-        &self,
-        poll: &mio::Poll,
-        token: mio::Token,
-        events: mio::Ready,
-        opts: mio::PollOpt,
-    ) -> io::Result<()> {
-        self.registration.register(poll, token, events, opts)
-    }
-
-    fn reregister(
-        &self,
-        poll: &mio::Poll,
-        token: mio::Token,
-        events: mio::Ready,
-        opts: mio::PollOpt,
-    ) -> io::Result<()> {
-        self.registration.reregister(poll, token, events, opts)
-    }
-
-    fn deregister(&self, poll: &mio::Poll) -> io::Result<()> {
-        mio::Evented::deregister(&self.registration, poll)
-    }
+    // TODO(1000): this will report that we handled a CTRL_BREAK_EVENT when
+    //       in fact we may not have any streams actually created for that
+    //       event.
+    TRUE
 }
 
 #[cfg(test)]