signal: Change constructors to return a result instead of lazy future (#1340)

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<process::Child> for GlobalOrphanQueue {
     }
 }
 
-type ChildReaperFuture = Pin<Box<dyn Stream<Item = io::Result<()>> + Send>>;
-
 #[must_use = "futures do nothing unless polled"]
 pub struct Child {
-    inner: Reaper<process::Child, GlobalOrphanQueue, ChildReaperFuture>,
+    inner: Reaper<process::Child, GlobalOrphanQueue, Signal>,
 }
 
 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),

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<W, Q, S> Future for Reaper<W, Q, S>
 where
     W: Wait + Unpin,
     Q: OrphanQueue<W> + Unpin,
-    S: TryStream<Error = io::Error> + Unpin,
+    S: Stream + Unpin,
 {
     type Output = io::Result<ExitStatus>;
 
-    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
-        let inner = Pin::get_mut(self);
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
         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));
             }
 

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};

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
 

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);
 

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);

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!(

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<Self> {
+    pub fn new() -> io::Result<Self> {
         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<Self> {
-        Inner::ctrl_c(handle).map_ok(|inner| Self { inner }).boxed()
+    pub fn with_handle(handle: &Handle) -> io::Result<Self> {
+        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<Option<Self::Item>> {
-        Pin::new(&mut self.inner)
-            .poll_next(cx)
-            .map(|item| item.map(|_| ()))
+        Pin::new(&mut self.inner).poll_next(cx)
     }
 }

tokio-signal/src/lib.rs

@@ -33,7 +33,7 @@
 //! async fn main() -> Result<(), Box<dyn std::error::Error>> {
 //!     // 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<dyn std::error::Error>> {
 //!     // 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<T>`
-pub type IoFuture<T> = Pin<Box<dyn Future<Output = io::Result<T>> + Send>>;
-
 pub use ctrl_c::CtrlC;

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<Signal> {
+    pub fn new(signal: c_int) -> io::Result<Self> {
         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<Signal> {
-        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<Self> {
+        // 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<Signal> {
+    pub(crate) fn ctrl_c(handle: &Handle) -> io::Result<Self> {
         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();
 

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<Event> {
+    pub(crate) fn ctrl_c(handle: &Handle) -> io::Result<Self> {
         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<Event> {
+    fn ctrl_break_handle(handle: &Handle) -> io::Result<Self> {
         Event::new(CTRL_BREAK_EVENT, handle)
     }
 
-    fn new(signum: DWORD, _handle: &Handle) -> IoFuture<Event> {
-        future::lazy(move |_| {
-            let mut init = None;
-            INIT.call_once(|| {
-                init = Some(global_init());
-            });
+    fn new(signum: DWORD, _handle: &Handle) -> io::Result<Self> {
+        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<Self> {
+    pub fn new() -> io::Result<Self> {
         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<Self> {
-        Event::ctrl_break_handle(handle)
-            .map_ok(|inner| Self { inner })
-            .boxed()
+    pub fn with_handle(handle: &Handle) -> io::Result<Self> {
+        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 {

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);

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;
 }

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);

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());
                 })

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;

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);
 

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);