diff --git a/tokio-process/src/unix/mod.rs b/tokio-process/src/unix/mod.rs index 585b43673..140c85daf 100644 --- a/tokio-process/src/unix/mod.rs +++ b/tokio-process/src/unix/mod.rs @@ -35,11 +35,7 @@ use self::orphan::{AtomicOrphanQueue, OrphanQueue, Wait}; use self::reap::Reaper; use super::SpawnedChild; use crate::kill::Kill; -use futures_core::stream::Stream; -use futures_util::future; use futures_util::future::FutureExt; -use futures_util::stream::StreamExt; -use futures_util::try_future::TryFutureExt; use std::fmt; use std::future::Future; use std::io; @@ -89,11 +85,9 @@ impl OrphanQueue for GlobalOrphanQueue { } } -type ChildReaperFuture = Pin> + Send>>; - #[must_use = "futures do nothing unless polled"] pub struct Child { - inner: Reaper, + inner: Reaper, } impl fmt::Debug for Child { @@ -110,10 +104,8 @@ pub(crate) fn spawn_child(cmd: &mut process::Command, handle: &Handle) -> io::Re let stdout = stdio(child.stdout.take(), handle)?; let stderr = stdio(child.stderr.take(), handle)?; - let signal = Signal::with_handle(libc::SIGCHLD, handle) - .and_then(|stream| future::ok(stream.map(Ok))) - .try_flatten_stream() - .boxed(); + let signal = Signal::with_handle(libc::SIGCHLD, handle)?; + Ok(SpawnedChild { child: Child { inner: Reaper::new(child, GlobalOrphanQueue, signal), diff --git a/tokio-process/src/unix/reap.rs b/tokio-process/src/unix/reap.rs index 8a70715f4..693507e43 100644 --- a/tokio-process/src/unix/reap.rs +++ b/tokio-process/src/unix/reap.rs @@ -1,7 +1,6 @@ use super::orphan::{OrphanQueue, Wait}; use crate::kill::Kill; -use futures_core::stream::TryStream; -use futures_util::try_stream::TryStreamExt; +use futures_core::stream::Stream; use std::future::Future; use std::io; use std::ops::Deref; @@ -61,12 +60,11 @@ impl Future for Reaper where W: Wait + Unpin, Q: OrphanQueue + Unpin, - S: TryStream + Unpin, + S: Stream + Unpin, { type Output = io::Result; - fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll { - let inner = Pin::get_mut(self); + fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll { loop { // If the child hasn't exited yet, then it's our responsibility to // ensure the current task gets notified when it might be able to @@ -87,14 +85,10 @@ where // this future's task will be notified/woken up again. Since the // futures model allows for spurious wake ups this extra wakeup // should not cause significant issues with parent futures. - let signal_poll = inner.signal.try_poll_next_unpin(cx); - if let Poll::Ready(Some(Err(err))) = signal_poll { - return Poll::Ready(Err(err)); - } - let registered_interest = signal_poll.is_pending(); + let registered_interest = Pin::new(&mut self.signal).poll_next(cx).is_pending(); - inner.orphan_queue.reap_orphans(); - if let Some(status) = inner.inner_mut().try_wait()? { + self.orphan_queue.reap_orphans(); + if let Some(status) = self.inner_mut().try_wait()? { return Poll::Ready(Ok(status)); } diff --git a/tokio-process/tests/stdio.rs b/tokio-process/tests/stdio.rs index 5a3b64a83..c442f0091 100644 --- a/tokio-process/tests/stdio.rs +++ b/tokio-process/tests/stdio.rs @@ -13,7 +13,6 @@ use std::process::{Command, ExitStatus, Stdio}; use futures_util::future; use futures_util::future::FutureExt; use futures_util::io::AsyncBufReadExt; -use futures_util::io::AsyncReadExt; use futures_util::io::AsyncWriteExt; use futures_util::io::BufReader; use futures_util::stream::{self, StreamExt}; diff --git a/tokio-signal/CHANGELOG.md b/tokio-signal/CHANGELOG.md index 5e8b94340..c25075992 100644 --- a/tokio-signal/CHANGELOG.md +++ b/tokio-signal/CHANGELOG.md @@ -6,6 +6,7 @@ a separate `windows::CtrlBreak` struct. - **Breaking:** `ctrl_c{,_with_handle}` has been replaced with a `CtrlC` struct (which can be constructed via `CtrlC::{new, with_handle}`. - **Breaking:** `unix::Signal` returns `()` instead of the signal number used in registration +- **Breaking:** `unis::Signal` constructors now return a simple result rather than a lazy future # 0.2.9 diff --git a/tokio-signal/examples/ctrl-c.rs b/tokio-signal/examples/ctrl-c.rs index dcce6696c..740b6c014 100644 --- a/tokio-signal/examples/ctrl-c.rs +++ b/tokio-signal/examples/ctrl-c.rs @@ -10,10 +10,7 @@ const STOP_AFTER: u64 = 10; async fn main() { // tokio_signal provides a convenience builder for Ctrl+C // this even works cross-platform: linux and windows! - // - // `CtrlC::new()` produces a `Future` of the actual stream-initialisation - // so first we await until the signal is ready. - let endless_stream = tokio_signal::CtrlC::new().await.unwrap(); + let endless_stream = tokio_signal::CtrlC::new().expect("failed to create CtrlC"); // don't keep going forever: convert the endless stream to a bounded one. let mut limited_stream = endless_stream.take(STOP_AFTER); diff --git a/tokio-signal/examples/multiple.rs b/tokio-signal/examples/multiple.rs index 0c5a17fd6..996af50d6 100644 --- a/tokio-signal/examples/multiple.rs +++ b/tokio-signal/examples/multiple.rs @@ -11,8 +11,8 @@ mod platform { pub async fn main() { // Create a stream for each of the signals we'd like to handle. - let sigint = Signal::new(SIGINT).await.unwrap().map(|_| SIGINT); - let sigterm = Signal::new(SIGTERM).await.unwrap().map(|_| SIGTERM); + let sigint = Signal::new(SIGINT).unwrap().map(|_| SIGINT); + let sigterm = Signal::new(SIGTERM).unwrap().map(|_| SIGTERM); // Use the `select` combinator to merge these two streams into one let stream = stream::select(sigint, sigterm); diff --git a/tokio-signal/examples/sighup-example.rs b/tokio-signal/examples/sighup-example.rs index ef84194de..5762d6878 100644 --- a/tokio-signal/examples/sighup-example.rs +++ b/tokio-signal/examples/sighup-example.rs @@ -9,7 +9,7 @@ mod platform { pub async fn main() { // on Unix, we can listen to whatever signal we want, in this case: SIGHUP - let mut stream = Signal::new(SIGHUP).await.unwrap(); + let mut stream = Signal::new(SIGHUP).unwrap(); println!("Waiting for SIGHUPS (Ctrl+C to quit)"); println!( diff --git a/tokio-signal/src/ctrl_c.rs b/tokio-signal/src/ctrl_c.rs index 377ee39db..03a5050c4 100644 --- a/tokio-signal/src/ctrl_c.rs +++ b/tokio-signal/src/ctrl_c.rs @@ -2,10 +2,8 @@ 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::io; use std::pin::Pin; use std::task::{Context, Poll}; use tokio_reactor::Handle; @@ -36,7 +34,7 @@ impl CtrlC { /// process. /// /// This function binds to the default reactor. - pub fn new() -> IoFuture { + pub fn new() -> io::Result { Self::with_handle(&Handle::default()) } @@ -44,8 +42,8 @@ impl CtrlC { /// 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() + pub fn with_handle(handle: &Handle) -> io::Result { + Inner::ctrl_c(handle).map(|inner| Self { inner }) } } @@ -53,8 +51,6 @@ 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(|_| ())) + Pin::new(&mut self.inner).poll_next(cx) } } diff --git a/tokio-signal/src/lib.rs b/tokio-signal/src/lib.rs index 50d907a56..086c0346d 100644 --- a/tokio-signal/src/lib.rs +++ b/tokio-signal/src/lib.rs @@ -33,7 +33,7 @@ //! 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::CtrlC::new().await?; +//! let ctrl_c = tokio_signal::CtrlC::new()?; //! //! // Process each ctrl-c as it comes in //! let prog = ctrl_c.for_each(|_| { @@ -60,7 +60,7 @@ //! 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::CtrlC::new().await?; +//! let ctrl_c = tokio_signal::CtrlC::new()?; //! //! // Process each ctrl-c as it comes in //! let prog = ctrl_c.for_each(|_| { @@ -72,7 +72,7 @@ //! //! // Like the previous example, this is an infinite stream of signals //! // being received, and signals may be coalesced while pending. -//! let stream = Signal::new(SIGHUP).await?; +//! let stream = Signal::new(SIGHUP)?; //! //! // Convert out stream into a future and block the program //! let (signal, _signal) = stream.into_future().await; @@ -84,10 +84,6 @@ #[macro_use] extern crate lazy_static; -use futures_core::future::Future; -use std::io; -use std::pin::Pin; - mod ctrl_c; mod registry; @@ -101,7 +97,4 @@ mod os { pub mod unix; pub mod windows; -/// A future whose output is `io::Result` -pub type IoFuture = Pin> + Send>>; - pub use ctrl_c::CtrlC; diff --git a/tokio-signal/src/unix.rs b/tokio-signal/src/unix.rs index d42cb18d3..65cdc99f6 100644 --- a/tokio-signal/src/unix.rs +++ b/tokio-signal/src/unix.rs @@ -12,9 +12,7 @@ use std::pin::Pin; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Once; -use crate::IoFuture; use futures_core::stream::Stream; -use futures_util::future::{self, FutureExt}; use libc::c_int; use mio_uds::UnixStream; use std::future::Future; @@ -283,7 +281,7 @@ impl Signal { /// * If the previous initialization of this specific signal failed. /// * If the signal is one of /// [`signal_hook::FORBIDDEN`](https://docs.rs/signal-hook/*/signal_hook/fn.register.html#panics) - pub fn new(signal: c_int) -> IoFuture { + pub fn new(signal: c_int) -> io::Result { Signal::with_handle(signal, &Handle::default()) } @@ -305,27 +303,23 @@ impl Signal { /// A `Signal` stream can be created for a particular signal number /// multiple times. When a signal is received then all the associated /// channels will receive the signal notification. - pub fn with_handle(signal: c_int, handle: &Handle) -> IoFuture { - let handle = handle.clone(); - future::lazy(move |_| { - // Turn the signal delivery on once we are ready for it - signal_enable(signal)?; + pub fn with_handle(signal: c_int, handle: &Handle) -> io::Result { + // Turn the signal delivery on once we are ready for it + signal_enable(signal)?; - // Ensure there's a driver for our associated event loop processing - // signals. - let driver = Driver::new(&handle)?; + // Ensure there's a driver for our associated event loop processing + // signals. + let driver = Driver::new(&handle)?; - // One wakeup in a queue is enough, no need for us to buffer up any - // more. - let (tx, rx) = channel(1); - globals().register_listener(signal as EventId, tx); + // One wakeup in a queue is enough, no need for us to buffer up any + // more. + let (tx, rx) = channel(1); + globals().register_listener(signal as EventId, tx); - Ok(Signal { driver, rx }) - }) - .boxed() + Ok(Signal { driver, rx }) } - pub(crate) fn ctrl_c(handle: &Handle) -> IoFuture { + pub(crate) fn ctrl_c(handle: &Handle) -> io::Result { Self::with_handle(libc::SIGINT, handle) } } @@ -365,9 +359,7 @@ mod tests { #[tokio::test] async fn ctrl_c() { - let ctrl_c = with_timeout(crate::CtrlC::new()) - .await - .expect("failed to init ctrl_c"); + let ctrl_c = crate::CtrlC::new().expect("failed to init ctrl_c"); let (fire, wait) = oneshot::channel(); diff --git a/tokio-signal/src/windows.rs b/tokio-signal/src/windows.rs index 9c699fb68..c6ebd7e39 100644 --- a/tokio-signal/src/windows.rs +++ b/tokio-signal/src/windows.rs @@ -15,8 +15,6 @@ use std::sync::Once; use std::task::{Context, Poll}; use futures_core::stream::Stream; -use futures_util::future::{self, FutureExt}; -use futures_util::try_future::TryFutureExt; use tokio_reactor::Handle; use tokio_sync::mpsc::{channel, Receiver, Sender}; use winapi::shared::minwindef::*; @@ -24,7 +22,6 @@ 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 { @@ -74,8 +71,6 @@ impl Init for OsExtraData { } } -static INIT: Once = Once::new(); - /// Stream of events discovered via `SetConsoleCtrlHandler`. /// /// This structure can be used to listen for events of the type `CTRL_C_EVENT` @@ -106,7 +101,7 @@ impl Event { /// /// This function will register a handler via `SetConsoleCtrlHandler` and /// deliver notifications to the returned stream. - pub(crate) fn ctrl_c(handle: &Handle) -> IoFuture { + pub(crate) fn ctrl_c(handle: &Handle) -> io::Result { Event::new(CTRL_C_EVENT, handle) } @@ -114,27 +109,17 @@ impl Event { /// /// This function will register a handler via `SetConsoleCtrlHandler` and /// deliver notifications to the returned stream. - fn ctrl_break_handle(handle: &Handle) -> IoFuture { + fn ctrl_break_handle(handle: &Handle) -> io::Result { Event::new(CTRL_BREAK_EVENT, handle) } - fn new(signum: DWORD, _handle: &Handle) -> IoFuture { - future::lazy(move |_| { - let mut init = None; - INIT.call_once(|| { - init = Some(global_init()); - }); + fn new(signum: DWORD, _handle: &Handle) -> io::Result { + global_init()?; - if let Some(Err(e)) = init { - return Err(e); - } + let (tx, rx) = channel(1); + globals().register_listener(signum as EventId, tx); - let (tx, rx) = channel(1); - globals().register_listener(signum as EventId, tx); - - Ok(Event { rx }) - }) - .boxed() + Ok(Event { rx }) } } @@ -147,14 +132,21 @@ impl Stream for Event { } fn global_init() -> io::Result<()> { - unsafe { - let rc = SetConsoleCtrlHandler(Some(handler), TRUE); - if rc == 0 { - return Err(io::Error::last_os_error()); - } + static INIT: Once = Once::new(); - Ok(()) - } + let mut init = None; + INIT.call_once(|| unsafe { + let rc = SetConsoleCtrlHandler(Some(handler), TRUE); + let ret = if rc == 0 { + Err(io::Error::last_os_error()) + } else { + Ok(()) + }; + + init = Some(ret); + }); + + init.unwrap_or_else(|| Ok(())) } impl Future for DriverTask { @@ -210,7 +202,7 @@ impl CtrlBreak { /// process. /// /// This function binds to the default reactor. - pub fn new() -> IoFuture { + pub fn new() -> io::Result { Self::with_handle(&Handle::default()) } @@ -218,10 +210,8 @@ impl CtrlBreak { /// process. /// /// This function binds to reactor specified by `handle`. - pub fn with_handle(handle: &Handle) -> IoFuture { - Event::ctrl_break_handle(handle) - .map_ok(|inner| Self { inner }) - .boxed() + pub fn with_handle(handle: &Handle) -> io::Result { + Event::ctrl_break_handle(handle).map(|inner| Self { inner }) } } @@ -238,7 +228,7 @@ impl Stream for CtrlBreak { #[cfg(test)] mod tests { use super::*; - use futures_util::future::FutureExt; + use futures_util::future::{self, FutureExt}; use futures_util::stream::StreamExt; use std::time::Duration; use tokio::runtime::current_thread; @@ -254,7 +244,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(crate::CtrlC::new())) + .block_on(with_timeout(future::lazy(|_| crate::CtrlC::new()))) .expect("failed to run future"); // Windows doesn't have a good programmatic way of sending events @@ -267,7 +257,7 @@ mod tests { let _ = rt.block_on(with_timeout(event_ctrl_c.into_future())); let event_ctrl_break = rt - .block_on(with_timeout(CtrlBreak::new())) + .block_on(with_timeout(future::lazy(|_| CtrlBreak::new()))) .expect("failed to run future"); unsafe { diff --git a/tokio-signal/tests/drop_multi_loop.rs b/tokio-signal/tests/drop_multi_loop.rs index ba0aa2ac8..1ba338cf8 100644 --- a/tokio-signal/tests/drop_multi_loop.rs +++ b/tokio-signal/tests/drop_multi_loop.rs @@ -12,14 +12,18 @@ const TEST_SIGNAL: libc::c_int = libc::SIGUSR1; fn dropping_loops_does_not_cause_starvation() { let (mut rt, signal) = { let mut first_rt = CurrentThreadRuntime::new().expect("failed to init first runtime"); - - let first_signal = run_with_timeout(&mut first_rt, Signal::new(TEST_SIGNAL)) - .expect("failed to register first signal"); + let mut first_signal = Signal::new(TEST_SIGNAL).expect("failed to register first signal"); let mut second_rt = CurrentThreadRuntime::new().expect("failed to init second runtime"); + let mut second_signal = Signal::new(TEST_SIGNAL).expect("failed to register second signal"); - let second_signal = run_with_timeout(&mut second_rt, Signal::new(TEST_SIGNAL)) - .expect("failed to register second signal"); + send_signal(TEST_SIGNAL); + + let _ = run_with_timeout(&mut first_rt, first_signal.next()) + .expect("failed to await first signal"); + + let _ = run_with_timeout(&mut second_rt, second_signal.next()) + .expect("failed to await second signal"); drop(first_rt); drop(first_signal); diff --git a/tokio-signal/tests/drop_then_get_a_signal.rs b/tokio-signal/tests/drop_then_get_a_signal.rs index 137a62e74..6f5a628c9 100644 --- a/tokio-signal/tests/drop_then_get_a_signal.rs +++ b/tokio-signal/tests/drop_then_get_a_signal.rs @@ -9,15 +9,11 @@ use crate::support::*; #[tokio::test] async fn drop_then_get_a_signal() { - let signal = with_timeout(Signal::new(libc::SIGUSR1)) - .await - .expect("failed to create first signal"); + let signal = Signal::new(libc::SIGUSR1).expect("failed to create first signal"); drop(signal); send_signal(libc::SIGUSR1); - let signal = with_timeout(Signal::new(libc::SIGUSR1)) - .await - .expect("failed to create second signal"); + let signal = Signal::new(libc::SIGUSR1).expect("failed to create second signal"); 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 7768eab7d..ca8bb1b0c 100644 --- a/tokio-signal/tests/dropping_does_not_deregister_other_instances.rs +++ b/tokio-signal/tests/dropping_does_not_deregister_other_instances.rs @@ -11,16 +11,13 @@ const TEST_SIGNAL: libc::c_int = libc::SIGUSR1; #[tokio::test] async fn dropping_signal_does_not_deregister_any_other_instances() { - // NB: Testing for issue #38: signals should not starve based on ordering - let first_duplicate_signal = with_timeout(Signal::new(TEST_SIGNAL)) - .await - .expect("failed to register first duplicate signal"); - let signal = with_timeout(Signal::new(TEST_SIGNAL)) - .await - .expect("failed to register signal"); - let second_duplicate_signal = with_timeout(Signal::new(TEST_SIGNAL)) - .await - .expect("failed to register second duplicate signal"); + // NB: Testing for issue alexcrichton/tokio-signal#38: + // signals should not starve based on ordering + let first_duplicate_signal = + Signal::new(TEST_SIGNAL).expect("failed to register first duplicate signal"); + let signal = Signal::new(TEST_SIGNAL).expect("failed to register signal"); + let second_duplicate_signal = + Signal::new(TEST_SIGNAL).expect("failed to register second duplicate signal"); drop(first_duplicate_signal); drop(second_duplicate_signal); diff --git a/tokio-signal/tests/multi_loop.rs b/tokio-signal/tests/multi_loop.rs index d241ce0db..1d75e873f 100644 --- a/tokio-signal/tests/multi_loop.rs +++ b/tokio-signal/tests/multi_loop.rs @@ -20,7 +20,7 @@ fn multi_loop() { let sender = sender.clone(); thread::spawn(move || { let mut rt = CurrentThreadRuntime::new().unwrap(); - let signal = run_with_timeout(&mut rt, Signal::new(libc::SIGHUP)).unwrap(); + let signal = Signal::new(libc::SIGHUP).unwrap(); sender.send(()).unwrap(); let _ = run_with_timeout(&mut rt, signal.into_future()); }) diff --git a/tokio-signal/tests/notify_both.rs b/tokio-signal/tests/notify_both.rs index 7662dbdb0..6b82df7aa 100644 --- a/tokio-signal/tests/notify_both.rs +++ b/tokio-signal/tests/notify_both.rs @@ -9,13 +9,9 @@ use libc; #[tokio::test] async fn notify_both() { - let signal1 = with_timeout(Signal::new(libc::SIGUSR2)) - .await - .expect("failed to create signal1"); + let signal1 = Signal::new(libc::SIGUSR2).expect("failed to create signal1"); - let signal2 = with_timeout(Signal::new(libc::SIGUSR2)) - .await - .expect("failed to create signal2"); + let signal2 = Signal::new(libc::SIGUSR2).expect("failed to create signal2"); send_signal(libc::SIGUSR2); 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 0ce3ad49d..1456cb61d 100644 --- a/tokio-signal/tests/simple.rs +++ b/tokio-signal/tests/simple.rs @@ -9,9 +9,7 @@ use libc; #[tokio::test] async fn simple() { - let signal = with_timeout(Signal::new(libc::SIGUSR1)) - .await - .expect("failed to create signal"); + let signal = Signal::new(libc::SIGUSR1).expect("failed to create signal"); send_signal(libc::SIGUSR1); diff --git a/tokio-signal/tests/twice.rs b/tokio-signal/tests/twice.rs index b56385514..a51478bc5 100644 --- a/tokio-signal/tests/twice.rs +++ b/tokio-signal/tests/twice.rs @@ -9,9 +9,7 @@ use libc; #[tokio::test] async fn twice() { - let mut signal = with_timeout(Signal::new(libc::SIGUSR1)) - .await - .expect("failed to get signal"); + let mut signal = Signal::new(libc::SIGUSR1).expect("failed to get signal"); for _ in 0..2 { send_signal(libc::SIGUSR1);