signal: move driver to runtime thread (#2835)

Refactors the signal infrastructure to move the driver to the runtime
thread. This follows the model put forth by the I/O driver and time
driver.

2 files changed, 104 insertions, 0 deletions

diff --git a/benches/Cargo.toml b/benches/Cargo.toml
index 3a90943f..cca0ece5 100644
--- a/benches/Cargo.toml
+++ b/benches/Cargo.toml
@@ -8,6 +8,9 @@ edition = "2018"
 tokio = { version = "0.3.0", path = "../tokio", features = ["full"] }
 bencher = "0.1.5"
 
+[target.'cfg(unix)'.dependencies]
+libc = "0.2.42"
+
 [[bench]]
 name = "spawn"
 path = "spawn.rs"
@@ -33,3 +36,8 @@ harness = false
 name = "sync_semaphore"
 path = "sync_semaphore.rs"
 harness = false
+
+[[bench]]
+name = "signal"
+path = "signal.rs"
+harness = false
diff --git a/benches/signal.rs b/benches/signal.rs
new file mode 100644
index 00000000..d891326d
--- /dev/null
+++ b/benches/signal.rs
@@ -0,0 +1,96 @@
+//! Benchmark the delay in propagating OS signals to any listeners.
+#![cfg(unix)]
+
+use bencher::{benchmark_group, benchmark_main, Bencher};
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::runtime;
+use tokio::signal::unix::{signal, SignalKind};
+use tokio::sync::mpsc;
+
+struct Spinner {
+    count: usize,
+}
+
+impl Future for Spinner {
+    type Output = ();
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if self.count > 3 {
+            Poll::Ready(())
+        } else {
+            self.count += 1;
+            cx.waker().wake_by_ref();
+            Poll::Pending
+        }
+    }
+}
+
+impl Spinner {
+    fn new() -> Self {
+        Self { count: 0 }
+    }
+}
+
+pub fn send_signal(signal: libc::c_int) {
+    use libc::{getpid, kill};
+
+    unsafe {
+        assert_eq!(kill(getpid(), signal), 0);
+    }
+}
+
+fn many_signals(bench: &mut Bencher) {
+    let num_signals = 10;
+    let (tx, mut rx) = mpsc::channel(num_signals);
+
+    let rt = runtime::Builder::new()
+        // Intentionally single threaded to measure delays in propagating wakes
+        .basic_scheduler()
+        .enable_all()
+        .build()
+        .unwrap();
+
+    let spawn_signal = |kind| {
+        let mut tx = tx.clone();
+        rt.spawn(async move {
+            let mut signal = signal(kind).expect("failed to create signal");
+
+            while signal.recv().await.is_some() {
+                if tx.send(()).await.is_err() {
+                    break;
+                }
+            }
+        });
+    };
+
+    for _ in 0..num_signals {
+        // Pick some random signals which don't terminate the test harness
+        spawn_signal(SignalKind::child());
+        spawn_signal(SignalKind::io());
+    }
+    drop(tx);
+
+    // Turn the runtime for a while to ensure that all the spawned
+    // tasks have been polled at least once
+    rt.block_on(Spinner::new());
+
+    bench.iter(|| {
+        rt.block_on(async {
+            send_signal(libc::SIGCHLD);
+            for _ in 0..num_signals {
+                rx.recv().await.expect("channel closed");
+            }
+
+            send_signal(libc::SIGIO);
+            for _ in 0..num_signals {
+                rx.recv().await.expect("channel closed");
+            }
+        });
+    });
+}
+
+benchmark_group!(signal_group, many_signals,);
+
+benchmark_main!(signal_group);