signal: Refactor Signal tests

* Added timeouts to all tests that were missing them
 - any issue we have will likely result in deadlocks/starvation so its
   best if all tests quickly timeout rather than require getting killed
   or have the CI timeout itself
* Added a `support` module and put a bunch of helpers there to DRY the
tests

tests/drop_multi_loop.rs

@@ -0,0 +1,39 @@
+#![cfg(unix)]
+
+extern crate libc;
+
+pub mod support;
+use support::*;
+
+const TEST_SIGNAL: libc::c_int = libc::SIGUSR1;
+
+#[test]
+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 second_rt = CurrentThreadRuntime::new()
+            .expect("failed to init second runtime");
+
+        let second_signal = run_with_timeout(&mut second_rt, Signal::new(TEST_SIGNAL))
+            .expect("failed to register second signal");
+
+        drop(first_rt);
+        drop(first_signal);
+
+        (second_rt, second_signal)
+    };
+
+    send_signal(TEST_SIGNAL);
+
+    let signal_future = signal.into_future()
+        .map_err(|(e, _)| e);
+
+    run_with_timeout(&mut rt, signal_future)
+        .expect("failed to get signal");
+}
+

tests/drop_then_get_a_signal.rs

@@ -1,28 +1,28 @@
 #![cfg(unix)]
 
-extern crate futures;
 extern crate libc;
-extern crate tokio;
-extern crate tokio_core;
-extern crate tokio_signal;
 
-use std::time::Duration;
-
-use tokio_core::reactor::{Core, Timeout};
-use tokio_signal::unix::Signal;
+pub mod support;
+use support::*;
 
 #[test]
 fn drop_then_get_a_signal() {
     let mut lp = Core::new().unwrap();
     let handle = lp.handle();
-    let signal = lp.run(Signal::with_handle(
+
+    let signal = run_core_with_timeout(&mut lp, Signal::with_handle(
         libc::SIGUSR1,
         &handle.new_tokio_handle(),
-    )).unwrap();
+    )).expect("failed to create first signal");
     drop(signal);
-    unsafe {
-        assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0);
-    }
-    let timeout = Timeout::new(Duration::from_millis(1), &lp.handle()).unwrap();
-    lp.run(timeout).unwrap();
+
+    send_signal(libc::SIGUSR1);
+    let signal = lp.run(Signal::with_handle(libc::SIGUSR1, &handle.new_tokio_handle()))
+        .expect("failed to create signal")
+        .into_future()
+        .map(|_| ())
+        .map_err(|(e, _)| panic!("{}", e));
+
+    run_core_with_timeout(&mut lp, signal)
+        .expect("failed to get signal");
 }

tests/dropping_does_not_deregister_other_instances.rs

@@ -1,14 +1,9 @@
 #![cfg(unix)]
 
-extern crate futures;
 extern crate libc;
-extern crate tokio;
-extern crate tokio_signal;
 
-use futures::stream::Stream;
-use futures::Future;
-use tokio_signal::unix::Signal;
-use std::time::{Duration, Instant};
+pub mod support;
+use support::*;
 
 const TEST_SIGNAL: libc::c_int = libc::SIGUSR1;
 
@@ -17,26 +12,21 @@ fn dropping_signal_does_not_deregister_any_other_instances() {
     // NB: Deadline requires a timer registration which is provided by
     // tokio's `current_thread::Runtime`, but isn't available by just using
     // tokio's default CurrentThread executor which powers `current_thread::block_on_all`.
-    let mut rt = tokio::runtime::current_thread::Runtime::new()
+    let mut rt = CurrentThreadRuntime::new()
         .expect("failed to init runtime");
 
     // NB: Testing for issue #38: signals should not starve based on ordering
-    let first_duplicate_signal = rt.block_on(Signal::new(TEST_SIGNAL))
+    let first_duplicate_signal = run_with_timeout(&mut rt, Signal::new(TEST_SIGNAL))
         .expect("failed to register first duplicate signal");
-    let signal = rt.block_on(Signal::new(TEST_SIGNAL))
+    let signal = run_with_timeout(&mut rt, Signal::new(TEST_SIGNAL))
         .expect("failed to register signal");
-    let second_duplicate_signal = rt.block_on(Signal::new(TEST_SIGNAL))
+    let second_duplicate_signal = run_with_timeout(&mut rt, Signal::new(TEST_SIGNAL))
         .expect("failed to register second duplicate signal");
 
     drop(first_duplicate_signal);
     drop(second_duplicate_signal);
 
-    unsafe { assert_eq!(libc::kill(libc::getpid(), TEST_SIGNAL), 0); }
-
-    let signal_future = signal.into_future()
-        .map_err(|(e, _)| e);
-
-    let deadline_time = Instant::now() + Duration::from_secs(1);
-    rt.block_on(tokio::timer::Deadline::new(signal_future, deadline_time))
+    send_signal(TEST_SIGNAL);
+    run_with_timeout(&mut rt, signal.into_future().map_err(|(e, _)| e))
         .expect("failed to get signal");
 }

tests/multi_loop.rs

@@ -1,17 +1,12 @@
 #![cfg(unix)]
 
-extern crate futures;
 extern crate libc;
-extern crate tokio;
-extern crate tokio_core;
-extern crate tokio_signal;
 
 use std::sync::mpsc::channel;
 use std::thread;
 
-use futures::stream::Stream;
-use tokio_core::reactor::Core;
-use tokio_signal::unix::Signal;
+pub mod support;
+use support::*;
 
 #[test]
 fn multi_loop() {
@@ -27,7 +22,7 @@ fn multi_loop() {
                     let mut lp = Core::new().unwrap();
                     let signal = lp.run(Signal::new(libc::SIGHUP)).unwrap();
                     sender.send(()).unwrap();
-                    lp.run(signal.into_future()).ok().unwrap();
+                    run_core_with_timeout(&mut lp, signal.into_future()).ok().unwrap();
                 })
             })
             .collect();
@@ -36,9 +31,7 @@ fn multi_loop() {
             receiver.recv().unwrap();
         }
         // Send a signal
-        unsafe {
-            assert_eq!(libc::kill(libc::getpid(), libc::SIGHUP), 0);
-        }
+        send_signal(libc::SIGHUP);
         // Make sure the threads terminated correctly
         for t in threads {
             t.join().unwrap();

tests/notify_both.rs

@@ -1,32 +1,27 @@
 #![cfg(unix)]
 
-extern crate futures;
 extern crate libc;
-extern crate tokio;
-extern crate tokio_core;
-extern crate tokio_signal;
 
-use futures::stream::Stream;
-use futures::Future;
-use tokio_core::reactor::Core;
-use tokio_signal::unix::Signal;
+pub mod support;
+use support::*;
 
 #[test]
 fn notify_both() {
     let mut lp = Core::new().unwrap();
     let handle = lp.handle();
-    let signal1 = lp.run(Signal::with_handle(
+
+    let signal1 = run_core_with_timeout(&mut lp, Signal::with_handle(
         libc::SIGUSR2,
         &handle.new_tokio_handle(),
-    )).unwrap();
-    let signal2 = lp.run(Signal::with_handle(
+    )).expect("failed to create signal1");
+
+    let signal2 = run_core_with_timeout(&mut lp, Signal::with_handle(
         libc::SIGUSR2,
         &handle.new_tokio_handle(),
-    )).unwrap();
-    unsafe {
-        assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR2), 0);
-    }
-    lp.run(signal1.into_future().join(signal2.into_future()))
+    )).expect("failed to create signal2");
+
+    send_signal(libc::SIGUSR2);
+    run_core_with_timeout(&mut lp, signal1.into_future().join(signal2.into_future()))
         .ok()
-        .unwrap();
+        .expect("failed to create signal2");
 }

tests/signal.rs

@@ -1,26 +1,20 @@
 #![cfg(unix)]
 
-extern crate futures;
 extern crate libc;
-extern crate tokio;
-extern crate tokio_core;
-extern crate tokio_signal;
 
-use futures::stream::Stream;
-use futures::{Future, IntoFuture};
-use tokio_signal::unix::Signal;
+pub mod support;
+use support::*;
 
 #[test]
 fn tokio_simple() {
-    tokio::run(
-        Signal::new(libc::SIGUSR1)
-            .into_future()
-            .and_then(|signal| {
-                unsafe {
-                    assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0);
-                }
-                signal.into_future().map(|_| ()).map_err(|(err, _)| err)
-            })
-            .map_err(|err| panic!("{}", err)),
-    )
+    let signal_future = Signal::new(libc::SIGUSR1)
+        .and_then(|signal| {
+            send_signal(libc::SIGUSR1);
+            signal.into_future().map(|_| ()).map_err(|(err, _)| err)
+        });
+
+    let mut rt = CurrentThreadRuntime::new()
+        .expect("failed to init runtime");
+    run_with_timeout(&mut rt, signal_future)
+        .expect("failed");
 }

tests/simple.rs

@@ -1,20 +1,20 @@
 #![cfg(unix)]
 
-extern crate futures;
 extern crate libc;
-extern crate tokio_core;
-extern crate tokio_signal;
 
-use futures::stream::Stream;
-use tokio_core::reactor::Core;
-use tokio_signal::unix::Signal;
+pub mod support;
+use support::*;
 
 #[test]
 fn simple() {
     let mut lp = Core::new().unwrap();
-    let signal = lp.run(Signal::new(libc::SIGUSR1)).unwrap();
-    unsafe {
-        assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0);
-    }
-    lp.run(signal.into_future()).ok().unwrap();
+
+    let signal = run_core_with_timeout(&mut lp, Signal::new(libc::SIGUSR1))
+        .expect("failed to create signal");
+
+    send_signal(libc::SIGUSR1);
+
+    run_core_with_timeout(&mut lp, signal.into_future())
+        .ok()
+        .expect("failed to get signal");
 }

tests/support.rs

@@ -0,0 +1,50 @@
+extern crate libc;
+extern crate futures;
+extern crate tokio;
+extern crate tokio_core;
+extern crate tokio_signal;
+
+use self::libc::{c_int, getpid, kill};
+use std::time::{Duration, Instant};
+use self::tokio::timer::Deadline;
+use self::tokio_core::reactor::Timeout;
+
+pub use self::futures::{Future, Stream};
+pub use self::tokio_core::reactor::Core;
+pub use self::tokio::runtime::current_thread::Runtime as CurrentThreadRuntime;
+pub use self::tokio_signal::unix::Signal;
+
+pub fn run_core_with_timeout<F>(lp: &mut Core, future: F) -> Result<F::Item, F::Error>
+    where F: Future
+{
+    let timeout = Timeout::new(Duration::from_secs(1), &lp.handle())
+        .expect("failed to register timeout")
+        .map(|()| panic!("timeout exceeded"))
+        .map_err(|e| panic!("timeout error: {}", e));
+
+    lp.run(future.select(timeout))
+        .map(|(r, _)| r)
+        .map_err(|(e, _)| e)
+}
+
+pub fn run_with_timeout<F>(rt: &mut CurrentThreadRuntime, future: F) -> Result<F::Item, F::Error>
+    where F: Future
+{
+    let deadline = Deadline::new(future, Instant::now() + Duration::from_secs(1))
+        .map_err(|e| if e.is_timer() {
+            panic!("failed to register timer");
+        } else if e.is_elapsed() {
+            panic!("timed out")
+        } else {
+            e.into_inner().expect("missing inner error")
+        });
+
+    rt.block_on(deadline)
+}
+
+#[cfg(unix)]
+pub fn send_signal(signal: c_int) {
+    unsafe {
+        assert_eq!(kill(getpid(), signal), 0);
+    }
+}

tests/twice.rs

@@ -1,26 +1,19 @@
 #![cfg(unix)]
 
-extern crate futures;
 extern crate libc;
-extern crate tokio;
-extern crate tokio_core;
-extern crate tokio_signal;
 
-use futures::stream::Stream;
-use tokio_core::reactor::Core;
-use tokio_signal::unix::Signal;
+pub mod support;
+use support::*;
 
 #[test]
 fn twice() {
     let mut lp = Core::new().unwrap();
-    let signal = lp.run(Signal::new(libc::SIGUSR1)).unwrap();
-    unsafe {
-        assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0);
-    }
-    let (num, signal) = lp.run(signal.into_future()).ok().unwrap();
+    let signal = run_core_with_timeout(&mut lp, Signal::new(libc::SIGUSR1)).unwrap();
+
+    send_signal(libc::SIGUSR1);
+    let (num, signal) = run_core_with_timeout(&mut lp, signal.into_future()).ok().unwrap();
     assert_eq!(num, Some(libc::SIGUSR1));
-    unsafe {
-        assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0);
-    }
-    lp.run(signal.into_future()).ok().unwrap();
+
+    send_signal(libc::SIGUSR1);
+    run_core_with_timeout(&mut lp, signal.into_future()).ok().unwrap();
 }