diff --git a/tokio-signal/Cargo.toml b/tokio-signal/Cargo.toml index 0c0633f41..48030d551 100644 --- a/tokio-signal/Cargo.toml +++ b/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" } diff --git a/tokio-signal/src/registry.rs b/tokio-signal/src/registry.rs index 215deceaa..604a19c95 100644 --- a/tokio-signal/src/registry.rs +++ b/tokio-signal/src/registry.rs @@ -82,7 +82,10 @@ impl Registry { } /// 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 Registry { // 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 Registry { } } }); + + 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); + } } diff --git a/tokio-signal/src/windows.rs b/tokio-signal/src/windows.rs index c6ebd7e39..bda0a0c45 100644 --- a/tokio-signal/src/windows.rs +++ b/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 { - 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(future: F) -> impl Future { 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; } }