From 461eebe612b32d2f75fde35d737b10728ff428cd Mon Sep 17 00:00:00 2001 From: Ivan Petkov Date: Tue, 9 Jul 2019 08:48:46 -0700 Subject: [PATCH] signal: Replace `ctrl_c` with a `CtrlC` struct (#1273) * Add a new `CtrlC` struct which will represent a stream of SIGINT signals on Unix or the CTRL_C event on Windows * `CtrlC` implements `Stream` rather than `IoSteam` as previously --- tokio-signal/examples/ctrl-c.rs | 41 +++++------- tokio-signal/examples/sighup-example.rs | 19 ++---- tokio-signal/src/ctrl_c.rs | 55 ++++++++++++++++ tokio-signal/src/lib.rs | 64 ++----------------- tokio-signal/src/registry.rs | 1 - tokio-signal/src/unix.rs | 35 ++++++++++ tokio-signal/src/windows.rs | 18 ++---- tokio-signal/tests/drop_multi_loop.rs | 2 +- tokio-signal/tests/drop_then_get_a_signal.rs | 2 +- ...ing_does_not_deregister_other_instances.rs | 2 +- tokio-signal/tests/multi_loop.rs | 2 +- tokio-signal/tests/notify_both.rs | 2 +- tokio-signal/tests/simple.rs | 2 +- 13 files changed, 129 insertions(+), 116 deletions(-) create mode 100644 tokio-signal/src/ctrl_c.rs diff --git a/tokio-signal/examples/ctrl-c.rs b/tokio-signal/examples/ctrl-c.rs index 53f927443..0d3b6d21a 100644 --- a/tokio-signal/examples/ctrl-c.rs +++ b/tokio-signal/examples/ctrl-c.rs @@ -1,7 +1,6 @@ #![deny(warnings, rust_2018_idioms)] #![feature(async_await)] -use futures_util::future; use futures_util::stream::StreamExt; /// how many signals to handle before exiting @@ -12,11 +11,11 @@ async fn main() -> Result<(), Box> { // tokio_signal provides a convenience builder for Ctrl+C // this even works cross-platform: linux and windows! // - // `fn ctrl_c()` produces a `Future` of the actual stream-initialisation + // `CtrlC::new()` produces a `Future` of the actual stream-initialisation // so first we await until the signal is ready. - let endless_stream = tokio_signal::ctrl_c().await?; + let endless_stream = tokio_signal::CtrlC::new().await?; // don't keep going forever: convert the endless stream to a bounded one. - let limited_stream = endless_stream.take(STOP_AFTER); + let mut limited_stream = endless_stream.take(STOP_AFTER); // how many Ctrl+C have we received so far? let mut counter = 0; @@ -28,29 +27,19 @@ async fn main() -> Result<(), Box> { STOP_AFTER ); - // Stream::for_each is a powerful primitive provided by the Futures crate. - // It turns a Stream into a Future that completes after all stream-items - // have been completed, or the first time the closure returns an error - let future = limited_stream - .map(|result| result.expect("failed to get event")) - .for_each(|()| { - // Note how we manipulate the counter without any fancy synchronisation. - // The borrowchecker realises there can't be any conflicts, so the closure - // can just capture it. - counter += 1; - println!( - "Ctrl+C received {} times! {} more before exit", - counter, - STOP_AFTER - counter - ); - - // return a result to continue handling the stream - future::ready(()) - }); - // Up until now, we haven't really DONE anything, just prepared - // now it's time to actually the results! - future.await; + // our futures, now it's time to actually await the results! + while let Some(_) = limited_stream.next().await { + // Note how we manipulate the counter without any fancy synchronisation. + // The borrowchecker realises there can't be any conflicts, so the closure + // can just capture it. + counter += 1; + println!( + "Ctrl+C received {} times! {} more before exit", + counter, + STOP_AFTER - counter + ); + } println!("Stream ended, quiting the program."); Ok(()) diff --git a/tokio-signal/examples/sighup-example.rs b/tokio-signal/examples/sighup-example.rs index a5b00eac7..cd5685be5 100644 --- a/tokio-signal/examples/sighup-example.rs +++ b/tokio-signal/examples/sighup-example.rs @@ -6,15 +6,13 @@ use std::error::Error; // A trick to not fail build on non-unix platforms when using unix-specific features. #[cfg(unix)] mod platform { - - use futures_util::future; use futures_util::stream::StreamExt; use std::error::Error; use tokio_signal::unix::{Signal, SIGHUP}; pub async fn main() -> Result<(), Box> { // on Unix, we can listen to whatever signal we want, in this case: SIGHUP - let stream = Signal::new(SIGHUP).await?; + let mut stream = Signal::new(SIGHUP).await?; println!("Waiting for SIGHUPS (Ctrl+C to quit)"); println!( @@ -23,22 +21,15 @@ mod platform { (i.e. this binary)" ); - // for_each is a powerful primitive provided by the Futures crate - // it turns a Stream into a Future that completes after all stream-items - // have been completed. - let future = stream.for_each(|the_signal| { + // Up until now, we haven't really DONE anything, just prepared + // our futures, now it's time to actually await the results! + while let Some(the_signal) = stream.next().await { println!( "*Got signal {:#x}* I should probably reload my config \ or something", the_signal ); - - future::ready(()) - }); - - // Up until now, we haven't really DONE anything, just prepared - // now it's time to actually the results! - future.await; + } Ok(()) } diff --git a/tokio-signal/src/ctrl_c.rs b/tokio-signal/src/ctrl_c.rs new file mode 100644 index 000000000..6bde65c37 --- /dev/null +++ b/tokio-signal/src/ctrl_c.rs @@ -0,0 +1,55 @@ +#[cfg(unix)] +use crate::unix::Signal as Inner; +#[cfg(windows)] +use crate::windows::Event as Inner; +use crate::IoFuture; +use futures_core::stream::Stream; +use futures_util::future::FutureExt; +use futures_util::try_future::TryFutureExt; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio_reactor::Handle; + +/// Represents a stream which receives "ctrl-c" notifications sent to the process. +/// +/// In general signals are handled very differently across Unix and Windows, but +/// this is somewhat cross platform in terms of how it can be handled. A ctrl-c +/// event to a console process can be represented as a stream for both Windows +/// and Unix. +/// +/// Note that there are a number of caveats listening for signals, and you may +/// wish to read up on the documentation in the `unix` or `windows` module to +/// take a peek. +#[must_use = "streams do nothing unless polled"] +#[derive(Debug)] +pub struct CtrlC { + inner: Inner, +} + +impl CtrlC { + /// Creates a new stream which receives "ctrl-c" notifications sent to the + /// process. + /// + /// This function binds to the default reactor. + pub fn new() -> IoFuture { + Self::with_handle(&Handle::default()) + } + + /// Creates a new stream which receives "ctrl-c" notifications sent to the + /// process. + /// + /// This function binds to reactor specified by `handle`. + pub fn with_handle(handle: &Handle) -> IoFuture { + Inner::ctrl_c(handle).map_ok(|inner| Self { inner }).boxed() + } +} + +impl Stream for CtrlC { + type Item = (); + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner) + .poll_next(cx) + .map(|item| item.map(|_| ())) + } +} diff --git a/tokio-signal/src/lib.rs b/tokio-signal/src/lib.rs index 03082dab8..ef8e48236 100644 --- a/tokio-signal/src/lib.rs +++ b/tokio-signal/src/lib.rs @@ -1,5 +1,5 @@ #![doc(html_root_url = "https://docs.rs/tokio-signal/0.2.8")] -#![deny(missing_docs, rust_2018_idioms)] +#![deny(missing_debug_implementations, missing_docs, rust_2018_idioms)] #![cfg_attr(test, deny(warnings))] #![cfg_attr(test, feature(async_await))] #![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))] @@ -33,12 +33,10 @@ //! async fn main() -> Result<(), Box> { //! // Create an infinite stream of "Ctrl+C" notifications. Each item received //! // on this stream may represent multiple ctrl-c signals. -//! let ctrl_c = tokio_signal::ctrl_c().await?; +//! let ctrl_c = tokio_signal::CtrlC::new().await?; //! //! // Process each ctrl-c as it comes in -//! let prog = ctrl_c.for_each(|event| { -//! event.expect("failed to get event"); -//! +//! let prog = ctrl_c.for_each(|_| { //! println!("ctrl-c received!"); //! future::ready(()) //! }); @@ -62,12 +60,10 @@ //! async fn main() -> Result<(), Box> { //! // Create an infinite stream of "Ctrl+C" notifications. Each item received //! // on this stream may represent multiple ctrl-c signals. -//! let ctrl_c = tokio_signal::ctrl_c().await?; +//! let ctrl_c = tokio_signal::CtrlC::new().await?; //! //! // Process each ctrl-c as it comes in -//! let prog = ctrl_c.for_each(|event| { -//! event.expect("failed to get event"); -//! +//! let prog = ctrl_c.for_each(|_| { //! println!("ctrl-c received!"); //! future::ready(()) //! }); @@ -90,13 +86,10 @@ extern crate lazy_static; use futures_core::future::Future; use futures_core::stream::Stream; -use futures_util::future::FutureExt; -use futures_util::stream::StreamExt; -use futures_util::try_future::TryFutureExt; use std::io; use std::pin::Pin; -use tokio_reactor::Handle; +mod ctrl_c; mod registry; mod os { @@ -114,47 +107,4 @@ pub type IoFuture = Pin> + Send>>; /// A stream whose item is `io::Result` pub type IoStream = Pin> + Send>>; -/// Creates a stream which receives "ctrl-c" notifications sent to a process. -/// -/// In general signals are handled very differently across Unix and Windows, but -/// this is somewhat cross platform in terms of how it can be handled. A ctrl-c -/// event to a console process can be represented as a stream for both Windows -/// and Unix. -/// -/// This function binds to the default event loop. Note that -/// there are a number of caveats listening for signals, and you may wish to -/// read up on the documentation in the `unix` or `windows` module to take a -/// peek. -pub fn ctrl_c() -> IoFuture> { - ctrl_c_handle(&Handle::default()) -} - -/// Creates a stream which receives "ctrl-c" notifications sent to a process. -/// -/// In general signals are handled very differently across Unix and Windows, but -/// this is somewhat cross platform in terms of how it can be handled. A ctrl-c -/// event to a console process can be represented as a stream for both Windows -/// and Unix. -/// -/// This function receives a `Handle` to an event loop and returns a future -/// which when resolves yields a stream receiving all signal events. Note that -/// there are a number of caveats listening for signals, and you may wish to -/// read up on the documentation in the `unix` or `windows` module to take a -/// peek. -pub fn ctrl_c_handle(handle: &Handle) -> IoFuture> { - return ctrl_c_imp(handle); - - #[cfg(unix)] - fn ctrl_c_imp(handle: &Handle) -> IoFuture> { - unix::Signal::with_handle(unix::libc::SIGINT, &handle) - .map_ok(|signal| -> IoStream<()> { signal.map(|_| Ok(())).boxed() }) - .boxed() - } - - #[cfg(windows)] - fn ctrl_c_imp(handle: &Handle) -> IoFuture> { - windows::Event::ctrl_c_handle(&handle) - .map_ok(|event| -> IoStream<()> { event.map(|_| Ok(())).boxed() }) - .boxed() - } -} +pub use ctrl_c::CtrlC; diff --git a/tokio-signal/src/registry.rs b/tokio-signal/src/registry.rs index cfeda99ea..5705fc1a6 100644 --- a/tokio-signal/src/registry.rs +++ b/tokio-signal/src/registry.rs @@ -252,7 +252,6 @@ mod tests { drop(second_rx); let (fire, wait) = oneshot::channel(); - //let mut rt = Runtime::new().unwrap(); tokio::spawn(async { wait.await.expect("wait failed"); diff --git a/tokio-signal/src/unix.rs b/tokio-signal/src/unix.rs index 9c07c443f..fe1044cb5 100644 --- a/tokio-signal/src/unix.rs +++ b/tokio-signal/src/unix.rs @@ -159,6 +159,7 @@ fn signal_enable(signal: c_int) -> io::Result<()> { } } +#[derive(Debug)] struct Driver { wakeup: PollEvented, } @@ -249,6 +250,8 @@ impl Driver { /// If you've got any questions about this feel free to open an issue on the /// repo, though, as I'd love to chat about this! In other words, I'd love to /// alleviate some of these limitations if possible! +#[must_use = "streams do nothing unless polled"] +#[derive(Debug)] pub struct Signal { driver: Driver, signal: c_int, @@ -326,6 +329,10 @@ impl Signal { }) .boxed() } + + pub(crate) fn ctrl_c(handle: &Handle) -> IoFuture { + Self::with_handle(libc::SIGINT, handle) + } } impl Stream for Signal { @@ -341,6 +348,11 @@ impl Stream for Signal { #[cfg(test)] mod tests { use super::*; + use futures_util::future::FutureExt; + use futures_util::StreamExt; + use std::time::Duration; + use tokio_sync::oneshot; + use tokio_timer::Timeout; #[test] fn signal_enable_error_on_invalid_input() { @@ -351,4 +363,27 @@ mod tests { fn signal_enable_error_on_forbidden_input() { signal_enable(signal_hook_registry::FORBIDDEN[0]).unwrap_err(); } + + fn with_timeout(future: F) -> impl Future { + Timeout::new(future, Duration::from_secs(1)).map(|result| result.expect("timed out")) + } + + #[tokio::test] + async fn ctrl_c() { + let ctrl_c = with_timeout(crate::CtrlC::new()) + .await + .expect("failed to init ctrl_c"); + + let (fire, wait) = oneshot::channel(); + + // NB: simulate a signal coming in by exercising our signal handler + // to avoid complications with sending SIGINT to the test process + tokio::spawn(async { + wait.await.expect("wait failed"); + action(globals(), libc::SIGINT); + }); + + let _ = fire.send(()); + let _ = with_timeout(ctrl_c.into_future()).await; + } } diff --git a/tokio-signal/src/windows.rs b/tokio-signal/src/windows.rs index 784a8243d..a025192be 100644 --- a/tokio-signal/src/windows.rs +++ b/tokio-signal/src/windows.rs @@ -89,6 +89,8 @@ static INIT: Once = ONCE_INIT; /// received back-to-back, then the stream may only receive one item about the /// two notifications. // FIXME: refactor and combine with unix::Signal +#[must_use = "streams do nothing unless polled"] +#[derive(Debug)] pub struct Event { rx: Receiver<()>, } @@ -103,15 +105,7 @@ impl Event { /// /// This function will register a handler via `SetConsoleCtrlHandler` and /// deliver notifications to the returned stream. - pub fn ctrl_c() -> IoFuture { - Event::ctrl_c_handle(&Handle::default()) - } - - /// Creates a new stream listening for the `CTRL_C_EVENT` events. - /// - /// This function will register a handler via `SetConsoleCtrlHandler` and - /// deliver notifications to the returned stream. - pub fn ctrl_c_handle(handle: &Handle) -> IoFuture { + pub(crate) fn ctrl_c(handle: &Handle) -> IoFuture { Event::new(CTRL_C_EVENT, handle) } @@ -224,7 +218,7 @@ mod tests { // first event loop cannot go away let mut rt = current_thread::Runtime::new().unwrap(); let event_ctrl_c = rt - .block_on(with_timeout(Event::ctrl_c())) + .block_on(with_timeout(crate::CtrlC::new())) .expect("failed to run future"); // Windows doesn't have a good programmatic way of sending events @@ -234,7 +228,7 @@ mod tests { super::handler(CTRL_C_EVENT); } - rt.block_on(with_timeout(event_ctrl_c.into_future())); + let _ = rt.block_on(with_timeout(event_ctrl_c.into_future())); let event_ctrl_break = rt .block_on(with_timeout(Event::ctrl_break())) @@ -244,6 +238,6 @@ mod tests { super::handler(CTRL_BREAK_EVENT); } - rt.block_on(with_timeout(event_ctrl_break.into_future())); + let _ = rt.block_on(with_timeout(event_ctrl_break.into_future())); } } diff --git a/tokio-signal/tests/drop_multi_loop.rs b/tokio-signal/tests/drop_multi_loop.rs index f6fb606d6..ba0aa2ac8 100644 --- a/tokio-signal/tests/drop_multi_loop.rs +++ b/tokio-signal/tests/drop_multi_loop.rs @@ -29,5 +29,5 @@ fn dropping_loops_does_not_cause_starvation() { send_signal(TEST_SIGNAL); - run_with_timeout(&mut rt, signal.into_future()); + let _ = run_with_timeout(&mut rt, signal.into_future()); } diff --git a/tokio-signal/tests/drop_then_get_a_signal.rs b/tokio-signal/tests/drop_then_get_a_signal.rs index ae5e626b7..137a62e74 100644 --- a/tokio-signal/tests/drop_then_get_a_signal.rs +++ b/tokio-signal/tests/drop_then_get_a_signal.rs @@ -19,5 +19,5 @@ async fn drop_then_get_a_signal() { .await .expect("failed to create second signal"); - with_timeout(signal.into_future()).await; + let _ = with_timeout(signal.into_future()).await; } diff --git a/tokio-signal/tests/dropping_does_not_deregister_other_instances.rs b/tokio-signal/tests/dropping_does_not_deregister_other_instances.rs index d44a5d946..7768eab7d 100644 --- a/tokio-signal/tests/dropping_does_not_deregister_other_instances.rs +++ b/tokio-signal/tests/dropping_does_not_deregister_other_instances.rs @@ -26,5 +26,5 @@ async fn dropping_signal_does_not_deregister_any_other_instances() { drop(second_duplicate_signal); send_signal(TEST_SIGNAL); - with_timeout(signal.into_future()).await; + let _ = with_timeout(signal.into_future()).await; } diff --git a/tokio-signal/tests/multi_loop.rs b/tokio-signal/tests/multi_loop.rs index afcb2f77a..d241ce0db 100644 --- a/tokio-signal/tests/multi_loop.rs +++ b/tokio-signal/tests/multi_loop.rs @@ -22,7 +22,7 @@ fn multi_loop() { let mut rt = CurrentThreadRuntime::new().unwrap(); let signal = run_with_timeout(&mut rt, Signal::new(libc::SIGHUP)).unwrap(); sender.send(()).unwrap(); - run_with_timeout(&mut rt, signal.into_future()); + let _ = run_with_timeout(&mut rt, signal.into_future()); }) }) .collect(); diff --git a/tokio-signal/tests/notify_both.rs b/tokio-signal/tests/notify_both.rs index 0b1526a38..7662dbdb0 100644 --- a/tokio-signal/tests/notify_both.rs +++ b/tokio-signal/tests/notify_both.rs @@ -18,5 +18,5 @@ async fn notify_both() { .expect("failed to create signal2"); send_signal(libc::SIGUSR2); - with_timeout(future::join(signal1.into_future(), signal2.into_future())).await; + let _ = with_timeout(future::join(signal1.into_future(), signal2.into_future())).await; } diff --git a/tokio-signal/tests/simple.rs b/tokio-signal/tests/simple.rs index 29e528981..0ce3ad49d 100644 --- a/tokio-signal/tests/simple.rs +++ b/tokio-signal/tests/simple.rs @@ -15,5 +15,5 @@ async fn simple() { send_signal(libc::SIGUSR1); - with_timeout(signal.into_future()).await; + let _ = with_timeout(signal.into_future()).await; }