signal: remove driver task for Windows event implementation (#1429)

Windows guarantees handler routines are always invoked in a new thread
(https://docs.microsoft.com/en-us/windows/console/handlerroutine), so we
don't need to use the handler-wake-another-driver technique used in the
Unix implementation

By broadcasting the event notifications from the handler, we no longer
need the Driver task to be spawned, which fixes the starvation issue if
the executor which runs the Driver task goes away

Also changed the behavior so that the default event handler runs if
all listeners for CTRL_{C, BREAK} events go away.

tokio-signal/Cargo.toml

@@ -24,7 +24,6 @@ futures-core-preview = "=0.3.0-alpha.18"
 futures-util-preview = "=0.3.0-alpha.18"
 lazy_static = "1"
 tokio-reactor = { version = "=0.2.0-alpha.1", path = "../tokio-reactor" }
-tokio-executor = { version = "=0.2.0-alpha.1", path = "../tokio-executor" }
 tokio-io = { version = "=0.2.0-alpha.1", path = "../tokio-io" }
 tokio-sync = { version = "=0.2.0-alpha.1", path = "../tokio-sync" }
 

tokio-signal/src/registry.rs

@@ -82,7 +82,10 @@ impl<S: Storage> Registry<S> {
     }
 
     /// Broadcast all previously recorded events to their respective listeners.
-    fn broadcast(&self) {
+    ///
+    /// Returns true if an event was delivered to at least one listener.
+    fn broadcast(&self) -> bool {
+        let mut did_notify = false;
         self.storage.for_each(|event_info| {
             // Any signal of this kind arrived since we checked last?
             if !event_info.pending.swap(false, Ordering::SeqCst) {
@@ -97,7 +100,7 @@ impl<S: Storage> Registry<S> {
             // has gone away then we can remove that slot.
             for i in (0..recipients.len()).rev() {
                 match recipients[i].try_send(()) {
-                    Ok(()) => {}
+                    Ok(()) => did_notify = true,
                     Err(ref e) if e.is_closed() => {
                         recipients.swap_remove(i);
                     }
@@ -112,6 +115,8 @@ impl<S: Storage> Registry<S> {
                 }
             }
         });
+
+        did_notify
     }
 }
 
@@ -141,7 +146,9 @@ impl Globals {
     }
 
     /// Broadcast all previously recorded events to their respective listeners.
-    pub(crate) fn broadcast(&self) {
+    ///
+    /// Returns true if an event was delivered to at least one listener.
+    pub(crate) fn broadcast(&self) -> bool {
         self.registry.broadcast()
     }
 
@@ -268,4 +275,27 @@ mod tests {
 
         assert_eq!(1, results.len());
     }
+
+    #[test]
+    fn broadcast_returns_if_at_least_one_event_fired() {
+        let registry = Registry::new(vec![EventInfo::default()]);
+
+        registry.record_event(0);
+        assert_eq!(false, registry.broadcast());
+
+        let (first_tx, first_rx) = channel(1);
+        let (second_tx, second_rx) = channel(1);
+
+        registry.register_listener(0, first_tx);
+        registry.register_listener(0, second_tx);
+
+        registry.record_event(0);
+        assert_eq!(true, registry.broadcast());
+
+        drop(first_rx);
+        registry.record_event(0);
+        assert_eq!(false, registry.broadcast());
+
+        drop(second_rx);
+    }
 }

tokio-signal/src/windows.rs

@@ -8,7 +8,6 @@
 #![cfg(windows)]
 
 use std::convert::TryFrom;
-use std::future::Future;
 use std::io;
 use std::pin::Pin;
 use std::sync::Once;
@@ -16,7 +15,7 @@ use std::task::{Context, Poll};
 
 use futures_core::stream::Stream;
 use tokio_reactor::Handle;
-use tokio_sync::mpsc::{channel, Receiver, Sender};
+use tokio_sync::mpsc::{channel, Receiver};
 use winapi::shared::minwindef::*;
 use winapi::um::consoleapi::SetConsoleCtrlHandler;
 use winapi::um::wincon::*;
@@ -57,17 +56,11 @@ impl Storage for OsStorage {
 }
 
 #[derive(Debug)]
-pub(crate) struct OsExtraData {
-    driver_waker: Sender<()>,
-}
+pub(crate) struct OsExtraData {}
 
 impl Init for OsExtraData {
     fn init() -> Self {
-        let (driver_waker, driver_rx) = channel(1);
-
-        tokio_executor::spawn(DriverTask { rx: driver_rx });
-
-        Self { driver_waker }
+        Self {}
     }
 }
 
@@ -91,11 +84,6 @@ pub(crate) struct Event {
     rx: Receiver<()>,
 }
 
-#[derive(Debug)]
-struct DriverTask {
-    rx: Receiver<()>,
-}
-
 impl Event {
     /// Creates a new stream listening for the `CTRL_C_EVENT` events.
     ///
@@ -149,39 +137,21 @@ fn global_init() -> io::Result<()> {
     init.unwrap_or_else(|| Ok(()))
 }
 
-impl Future for DriverTask {
-    type Output = ();
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        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_recv(cx) {
-                Poll::Ready(Some(())) => continue,
-                Poll::Ready(None) => panic!("driver got disconnected?"),
-                Poll::Pending => break,
-            }
-        }
-
-        globals().broadcast();
-
-        // TODO(1000): when to finish this task?
-        Poll::Pending
-    }
-}
-
 unsafe extern "system" fn handler(ty: DWORD) -> BOOL {
     let globals = globals();
     globals.record_event(ty as EventId);
 
-    // 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(());
-
-    // 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
+    // According to https://docs.microsoft.com/en-us/windows/console/handlerroutine
+    // the handler routine is always invoked in a new thread, thus we don't
+    // have the same restrictions as in Unix signal handlers, meaning we can
+    // go ahead and perform the broadcast here.
+    if globals.broadcast() {
+        TRUE
+    } else {
+        // No one is listening for this notification any more
+        // let the OS fire the next (possibly the default) handler.
+        FALSE
+    }
 }
 
 /// Represents a stream which receives "ctrl-break" notifications sent to the process
@@ -228,24 +198,19 @@ impl Stream for CtrlBreak {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use futures_util::future::{self, FutureExt};
+    use futures_util::future::FutureExt;
     use futures_util::stream::StreamExt;
+    use std::future::Future;
     use std::time::Duration;
-    use tokio::runtime::current_thread;
     use tokio_timer::Timeout;
 
     fn with_timeout<F: Future>(future: F) -> impl Future<Output = F::Output> {
         Timeout::new(future, Duration::from_secs(1)).map(|result| result.expect("timed out"))
     }
 
-    #[test]
-    fn ctrl_c_and_ctrl_break() {
-        // FIXME(1000): combining into one test due to a restriction where the
-        // first event loop cannot go away
-        let mut rt = current_thread::Runtime::new().unwrap();
-        let event_ctrl_c = rt
-            .block_on(with_timeout(future::lazy(|_| crate::CtrlC::new())))
-            .expect("failed to run future");
+    #[tokio::test]
+    async fn ctrl_c() {
+        let ctrl_c = crate::CtrlC::new().expect("failed to create CtrlC");
 
         // Windows doesn't have a good programmatic way of sending events
         // like sending signals on Unix, so we'll stub out the actual OS
@@ -254,16 +219,20 @@ mod tests {
             super::handler(CTRL_C_EVENT);
         }
 
-        let _ = rt.block_on(with_timeout(event_ctrl_c.into_future()));
+        let _ = with_timeout(ctrl_c.into_future()).await;
+    }
 
-        let event_ctrl_break = rt
-            .block_on(with_timeout(future::lazy(|_| CtrlBreak::new())))
-            .expect("failed to run future");
+    #[tokio::test]
+    async fn ctrl_break() {
+        let ctrl_break = super::CtrlBreak::new().expect("failed to create CtrlC");
 
+        // Windows doesn't have a good programmatic way of sending events
+        // like sending signals on Unix, so we'll stub out the actual OS
+        // integration and test that our handling works.
         unsafe {
             super::handler(CTRL_BREAK_EVENT);
         }
 
-        let _ = rt.block_on(with_timeout(event_ctrl_break.into_future()));
+        let _ = with_timeout(ctrl_break.into_future()).await;
     }
 }