tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-07 00:00:09 +02:00
Code Issues Packages Projects Releases Wiki Activity

signal: migrate to std::futures (#1218)

Browse Source 
Migrate to std::futures and the futures 0.3 preview and use async/await
where possible

**Breaking change:** the IoFuture and IoStream definitions used to refer
to Box<dyn Future> and Box<dyn Stream>, but now they are defined as
Pin<...> versions which are technically breaking.

No other breaking or functional changes have been made
This commit is contained in:
Ivan Petkov
2019-07-03 10:40:59 -07:00
committed by Carl Lerche
parent bd9760e124
commit cbad83f362
19 changed files with 304 additions and 338 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+1 -1
View File
@@ -11,7 +11,7 @@ members = [
"tokio-io",
"tokio-macros",
"tokio-reactor",
# "tokio-signal",
"tokio-signal",
"tokio-sync",
"tokio-test",
"tokio-threadpool",
azure-pipelines.yml
+1 -1
View File
@@ -32,7 +32,7 @@ jobs:
crates:
# - tokio-fs
tokio-reactor: []
# - tokio-signal
tokio-signal: []
tokio-tcp:
- incoming
# - tokio-tls
tokio-signal/Cargo.toml
+4 -1
View File
@@ -26,11 +26,13 @@ travis-ci = { repository = "tokio-rs/tokio" }
appveyor = { repository = "carllerche/tokio", id = "s83yxhy9qeb58va7" }
[dependencies]
futures = "0.1.11"
futures-core-preview = "0.3.0-alpha.16"
futures-util-preview = "0.3.0-alpha.16"
lazy_static = "1"
tokio-reactor = { version = "0.2.0", path = "../tokio-reactor" }
tokio-executor = { version = "0.2.0", path = "../tokio-executor" }
tokio-io = { version = "0.2.0", path = "../tokio-io" }
tokio-sync = { version = "0.2.0", path = "../tokio-sync" }
[target.'cfg(unix)'.dependencies]
libc = "0.2"
@@ -41,6 +43,7 @@ signal-hook-registry = "~1"
[dev-dependencies]
tokio = { version = "0.2.0", path = "../tokio" }
tokio-timer = { version = "0.3.0", path = "../tokio-timer" }
tokio-sync = { version = "0.2.0", path = "../tokio-sync", features = ["async-traits"]}
[target.'cfg(windows)'.dependencies.winapi]
version = "0.3"
tokio-signal/examples/ctrl-c.rs
+24 -30
View File
@@ -1,22 +1,20 @@
#![deny(warnings, rust_2018_idioms)]
#![feature(async_await)]
use tokio;
use tokio_signal;
use futures::{Future, Stream};
use futures_util::future;
use futures_util::stream::StreamExt;
/// how many signals to handle before exiting
const STOP_AFTER: u64 = 10;
fn main() -> Result<(), Box<dyn std::error::Error>> {
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// 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
// the `flatten_stream()` convenience method lazily defers that
// initialisation, allowing us to use it 'as if' it is already the
// stream we want, reducing boilerplate Future-handling.
let endless_stream = tokio_signal::ctrl_c().flatten_stream();
// so first we await until the signal is ready.
let endless_stream = tokio_signal::ctrl_c().await?;
// don't keep going forever: convert the endless stream to a bounded one.
let limited_stream = endless_stream.take(STOP_AFTER);
@@ -36,30 +34,26 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
// 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.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
);
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 Ok-result to continue handling the stream
Ok(())
});
// 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 schedule, and thus execute, the stream
// on our event loop
// FIXME(1000): windows uses a global driver task which doesn't terminate
// on its own, so if we use block_on_all our application will never exit
//tokio::runtime::current_thread::block_on_all(future)?;
tokio::runtime::current_thread::Runtime::new()
.expect("failed to start runtime on current thread")
.block_on(future)?;
// now it's time to actually the results!
future.await;
println!("Stream ended, quiting the program.");
Ok(())
tokio-signal/examples/multiple.rs
+16 -10
View File
@@ -1,21 +1,25 @@
#![deny(warnings, rust_2018_idioms)]
#![feature(async_await)]
//! A small example of how to listen for two signals at the same time
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::{Future, Stream};
use futures_util::stream::{self, StreamExt};
use std::error::Error;
use tokio_signal::unix::{Signal, SIGINT, SIGTERM};
pub fn main() -> Result<(), Box<dyn (::std::error::Error)>> {
pub async fn main() -> Result<(), Box<dyn Error>> {
// Create a stream for each of the signals we'd like to handle.
let sigint = Signal::new(SIGINT).flatten_stream();
let sigterm = Signal::new(SIGTERM).flatten_stream();
let sigint = Signal::new(SIGINT).await?;
let sigterm = Signal::new(SIGTERM).await?;
// Use the `select` combinator to merge these two streams into one
let stream = sigint.select(sigterm);
let stream = stream::select(sigint, sigterm);
// Wait for a signal to arrive
println!("Waiting for SIGINT or SIGTERM");
@@ -24,8 +28,7 @@ mod platform {
to send a SIGINT to all processes named 'multiple' \
(i.e. this binary)"
);
let (item, _rest) = ::tokio::runtime::current_thread::block_on_all(stream.into_future())
.map_err(|_| "failed to wait for signals")?;
let (item, _rest) = stream.into_future().await;
// Figure out which signal we received
let item = item.ok_or("received no signal")?;
@@ -35,6 +38,7 @@ mod platform {
assert_eq!(item, SIGTERM);
println!("received SIGTERM");
}
Ok(())
}
@@ -42,11 +46,13 @@ mod platform {
#[cfg(not(unix))]
mod platform {
pub fn main() -> Result<(), Box<dyn std::error::Error>> {
use std::error::Error;
pub async fn main() -> Result<(), Box<dyn Error>> {
Ok(())
}
}
fn main() -> Result<(), Box<dyn std::error::Error>> {
platform::main()
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
platform::main().await
}
tokio-signal/examples/sighup-example.rs
+18 -11
View File
@@ -1,15 +1,20 @@
#![deny(warnings, rust_2018_idioms)]
#![feature(async_await)]
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::{Future, Stream};
use futures_util::future;
use futures_util::stream::StreamExt;
use std::error::Error;
use tokio_signal::unix::{Signal, SIGHUP};
pub fn main() -> Result<(), Box<dyn (::std::error::Error)>> {
pub async fn main() -> Result<(), Box<dyn Error>> {
// on Unix, we can listen to whatever signal we want, in this case: SIGHUP
let stream = Signal::new(SIGHUP).flatten_stream();
let stream = Signal::new(SIGHUP).await?;
println!("Waiting for SIGHUPS (Ctrl+C to quit)");
println!(
@@ -27,25 +32,27 @@ mod platform {
or something",
the_signal
);
Ok(())
future::ready(())
});
// Up until now, we haven't really DONE anything, just prepared
// now it's time to actually schedule, and thus execute, the stream
// on our event loop, and loop forever
::tokio::runtime::current_thread::block_on_all(future)?;
// now it's time to actually the results!
future.await;
Ok(())
}
}
#[cfg(not(unix))]
mod platform {
pub fn main() -> Result<(), Box<dyn ::std::error::Error>> {
use std::error::Error;
pub async fn main() -> Result<(), Box<dyn Error>> {
Ok(())
}
}
fn main() -> Result<(), Box<dyn std::error::Error>> {
platform::main()
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
platform::main().await
}
tokio-signal/src/lib.rs
+65 -35
View File
@@ -1,6 +1,7 @@
#![doc(html_root_url = "https://docs.rs/tokio-signal/0.2.8")]
#![deny(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))))]
//! Asynchronous signal handling for Tokio
@@ -23,43 +24,77 @@
//! Print out all ctrl-C notifications received
//!
//! ```rust,no_run
//! use futures::{Future, Stream};
//! #![feature(async_await)]
//!
//! // 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().flatten_stream();
//! use futures_util::future;
//! use futures_util::stream::StreamExt;
//!
//! #[tokio::main]
//! 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::ctrl_c().await?;
//!
//! // Process each ctrl-c as it comes in
//! let prog = ctrl_c.for_each(|event| {
//! event.expect("failed to get event");
//!
//! println!("ctrl-c received!");
//! future::ready(())
//! });
//!
//! prog.await;
//!
//! // Process each ctrl-c as it comes in
//! let prog = ctrl_c.for_each(|()| {
//! println!("ctrl-c received!");
//! Ok(())
//! });
//!
//! tokio::runtime::current_thread::block_on_all(prog).unwrap();
//! }
//! ```
//!
//! Wait for SIGHUP on Unix
//!
//! ```rust,no_run
//! # #[cfg(unix)] fn dox() {
//! use futures::{Future, Stream};
//! #![feature(async_await)]
//!
//! use futures_util::future;
//! use futures_util::stream::StreamExt;
//! use tokio_signal::unix::{Signal, SIGHUP};
//!
//! // 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).flatten_stream();
//! #[tokio::main]
//! 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::ctrl_c().await?;
//!
//! // Convert out stream into a future and block the program
//! tokio::runtime::current_thread::block_on_all(stream.into_future()).ok().unwrap();
//! # }
//! // Process each ctrl-c as it comes in
//! let prog = ctrl_c.for_each(|event| {
//! event.expect("failed to get event");
//!
//! println!("ctrl-c received!");
//! future::ready(())
//! });
//!
//! prog.await;
//!
//! // 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?;
//!
//! // Convert out stream into a future and block the program
//! let (signal, _signal) = stream.into_future().await;
//! println!("got signal {:?}", signal);
//! Ok(())
//! }
//! ```
#[macro_use]
extern crate lazy_static;
use futures::stream::Stream;
use futures::{future, Future};
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 registry;
@@ -74,10 +109,10 @@ mod os {
pub mod unix;
pub mod windows;
/// A future whose error is `io::Error`
pub type IoFuture<T> = Box<dyn Future<Item = T, Error = io::Error> + Send>;
/// A stream whose error is `io::Error`
pub type IoStream<T> = Box<dyn Stream<Item = T, Error = io::Error> + Send>;
/// A future whose output is `io::Result<T>`
pub type IoFuture<T> = Pin<Box<dyn Future<Output = io::Result<T>> + Send>>;
/// A stream whose item is `io::Result<T>`
pub type IoStream<T> = Pin<Box<dyn Stream<Item = io::Result<T>> + Send>>;
/// Creates a stream which receives "ctrl-c" notifications sent to a process.
///
@@ -111,20 +146,15 @@ pub fn ctrl_c_handle(handle: &Handle) -> IoFuture<IoStream<()>> {
#[cfg(unix)]
fn ctrl_c_imp(handle: &Handle) -> IoFuture<IoStream<()>> {
let handle = handle.clone();
Box::new(future::lazy(move || {
unix::Signal::with_handle(unix::libc::SIGINT, &handle)
.map(|x| Box::new(x.map(|_| ())) as Box<dyn Stream<Item = _, Error = _> + Send>)
}))
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<IoStream<()>> {
let handle = handle.clone();
// Use lazy to ensure that `ctrl_c` gets called while on an event loop
Box::new(future::lazy(move || {
windows::Event::ctrl_c_handle(&handle)
.map(|x| Box::new(x) as Box<dyn Stream<Item = _, Error = _> + Send>)
}))
windows::Event::ctrl_c_handle(&handle)
.map_ok(|event| -> IoStream<()> { event.map(|_| Ok(())).boxed() })
.boxed()
}
}
tokio-signal/src/registry.rs
+47 -67
View File
@@ -4,7 +4,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Mutex;
use crate::os::{OsExtraData, OsStorage};
use futures::sync::mpsc::Sender;
use tokio_sync::mpsc::Sender;
pub(crate) type EventId = usize;
@@ -98,7 +98,7 @@ impl<S: Storage> Registry<S> {
for i in (0..recipients.len()).rev() {
match recipients[i].try_send(()) {
Ok(()) => {}
Err(ref e) if e.is_disconnected() => {
Err(ref e) if e.is_closed() => {
recipients.swap_remove(i);
}
@@ -169,65 +169,53 @@ where
#[cfg(test)]
mod tests {
use super::*;
use futures::sync::mpsc::channel;
use futures::sync::oneshot;
use futures::{Future, Stream};
use futures_util::{future, StreamExt};
use tokio_sync::mpsc::channel;
use tokio_sync::oneshot;
use std::time::Duration;
use tokio::runtime::Runtime;
use tokio_timer::sleep;
#[test]
fn smoke() {
#[tokio::test]
async fn smoke() {
let registry = Registry::new(vec![
EventInfo::default(),
EventInfo::default(),
EventInfo::default(),
]);
let (first_tx, first_rx) = channel(0);
let (second_tx, second_rx) = channel(0);
let (third_tx, third_rx) = channel(0);
let (first_tx, first_rx) = channel(3);
let (second_tx, second_rx) = channel(3);
let (third_tx, third_rx) = channel(3);
registry.register_listener(0, first_tx);
registry.register_listener(1, second_tx);
registry.register_listener(2, third_tx);
let (fire, wait) = oneshot::channel();
let rt = Runtime::new().unwrap();
rt.spawn(
wait.and_then(move |_| {
// Record some events which should get coalesced
registry.record_event(0);
registry.record_event(0);
registry.record_event(1);
registry.record_event(1);
registry.broadcast();
tokio::spawn(async {
wait.await.expect("wait failed");
sleep(Duration::from_millis(100))
.map_err(|e| panic!("{:#?}", e))
.and_then(move |_| {
registry.record_event(0);
registry.broadcast();
// Record some events which should get coalesced
registry.record_event(0);
registry.record_event(0);
registry.record_event(1);
registry.record_event(1);
registry.broadcast();
drop(registry);
Ok(())
})
})
.map_err(|e| panic!("{}", e)),
// Send subsequent signal
registry.record_event(0);
registry.broadcast();
drop(registry);
});
let _ = fire.send(());
let all = future::join3(
first_rx.collect::<Vec<_>>(),
second_rx.collect::<Vec<_>>(),
third_rx.collect::<Vec<_>>(),
);
let (first_results, second_results, third_results) = rt
.block_on(futures::lazy(move || {
let _ = fire.send(());
first_rx
.collect()
.join3(second_rx.collect(), third_rx.collect())
}))
.expect("failed to extract events");
let (first_results, second_results, third_results) = all.await;
assert_eq!(2, first_results.len());
assert_eq!(1, second_results.len());
assert_eq!(0, third_results.len());
@@ -238,7 +226,7 @@ mod tests {
fn register_panics_on_invalid_input() {
let registry = Registry::new(vec![EventInfo::default()]);
let (tx, _) = channel(0);
let (tx, _) = channel(1);
registry.register_listener(1, tx);
}
@@ -248,13 +236,13 @@ mod tests {
registry.record_event(42);
}
#[test]
fn broadcast_cleans_up_disconnected_listeners() {
#[tokio::test]
async fn broadcast_cleans_up_disconnected_listeners() {
let registry = Registry::new(vec![EventInfo::default()]);
let (first_tx, first_rx) = channel(0);
let (second_tx, second_rx) = channel(0);
let (third_tx, third_rx) = channel(0);
let (first_tx, first_rx) = channel(1);
let (second_tx, second_rx) = channel(1);
let (third_tx, third_rx) = channel(1);
registry.register_listener(0, first_tx);
registry.register_listener(0, second_tx);
@@ -264,28 +252,20 @@ mod tests {
drop(second_rx);
let (fire, wait) = oneshot::channel();
let rt = Runtime::new().unwrap();
//let mut rt = Runtime::new().unwrap();
rt.spawn(
wait.and_then(move |_| {
// Record some events which should get coalesced
registry.record_event(0);
registry.broadcast();
tokio::spawn(async {
wait.await.expect("wait failed");
assert_eq!(1, registry.storage[0].recipients.lock().unwrap().len());
drop(registry);
registry.record_event(0);
registry.broadcast();
Ok(())
})
.map_err(|e| panic!("{}", e)),
);
assert_eq!(1, registry.storage[0].recipients.lock().unwrap().len());
drop(registry);
});
let results = rt
.block_on(futures::lazy(move || {
let _ = fire.send(());
third_rx.collect()
}))
.expect("failed to extract events");
let _ = fire.send(());
let results: Vec<()> = third_rx.collect().await;
assert_eq!(1, results.len());
}
tokio-signal/src/unix.rs
+38 -47
View File
@@ -7,20 +7,21 @@
pub use libc;
use std::io::prelude::*;
use std::io::{self, Error, ErrorKind};
use std::io::{self, Error, ErrorKind, Write};
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Once, ONCE_INIT};
use futures::future;
use futures::sync::mpsc::{channel, Receiver};
use futures::{Async, Future};
use futures::{Poll, Stream};
use crate::IoFuture;
use futures_core::stream::Stream;
use futures_util::future::{self, FutureExt};
use libc::c_int;
use mio_uds::UnixStream;
use tokio_io::IoFuture;
use std::future::Future;
use std::task::{Context, Poll};
use tokio_io::AsyncRead;
use tokio_reactor::{Handle, PollEvented};
use tokio_sync::mpsc::{channel, Receiver};
use crate::registry::{globals, EventId, EventInfo, Globals, Init, Storage};
@@ -163,17 +164,15 @@ struct Driver {
}
impl Future for Driver {
type Item = ();
type Error = ();
type Output = ();
fn poll(&mut self) -> Poll<(), ()> {
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// Drain the data from the pipe and maintain interest in getting more
self.drain();
self.drain(cx);
// Broadcast any signals which were received
globals().broadcast();
// This task just lives until the end of the event loop
Ok(Async::NotReady)
Poll::Pending
}
}
@@ -204,13 +203,13 @@ impl Driver {
/// We do *NOT* use the existence of any read bytes as evidence a sigal was
/// received since the `pending` flags would have already been set if that
/// was the case. See #38 for more info.
fn drain(&mut self) {
fn drain(mut self: Pin<&mut Self>, cx: &mut Context<'_>) {
loop {
match self.wakeup.read(&mut [0; 128]) {
Ok(0) => panic!("EOF on self-pipe"),
Ok(_) => {}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
Err(e) => panic!("Bad read on self-pipe: {}", e),
match Pin::new(&mut self.wakeup).poll_read(cx, &mut [0; 128]) {
Poll::Ready(Ok(0)) => panic!("EOF on self-pipe"),
Poll::Ready(Ok(_)) => {}
Poll::Ready(Err(e)) => panic!("Bad read on self-pipe: {}", e),
Poll::Pending => break,
}
}
}
@@ -306,44 +305,36 @@ impl Signal {
/// channels will receive the signal notification.
pub fn with_handle(signal: c_int, handle: &Handle) -> IoFuture<Signal> {
let handle = handle.clone();
Box::new(future::lazy(move || {
let result = (|| {
// Turn the signal delivery on once we are ready for it
signal_enable(signal)?;
future::lazy(move |_| {
// 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. NB: channels always guarantee at least one slot per sender,
// so we don't need additional slots
let (tx, rx) = channel(0);
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: driver,
rx: rx,
signal: signal,
})
})();
future::result(result)
}))
Ok(Signal {
driver: driver,
rx: rx,
signal: signal,
})
})
.boxed()
}
}
impl Stream for Signal {
type Item = c_int;
type Error = io::Error;
fn poll(&mut self) -> Poll<Option<c_int>, io::Error> {
self.driver.poll().unwrap();
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let _ = Pin::new(&mut self.driver).poll(cx);
self.rx
.poll()
.map(|ready| ready.map(|item| item.map(|()| self.signal)))
// receivers don't generate errors
.map_err(|_| unreachable!())
self.rx.poll_recv(cx).map(|item| item.map(|()| self.signal))
}
}
tokio-signal/src/windows.rs
+30 -44
View File
@@ -8,13 +8,16 @@
#![cfg(windows)]
use std::convert::TryFrom;
use std::future::Future;
use std::io;
use std::pin::Pin;
use std::sync::{Once, ONCE_INIT};
use std::task::{Context, Poll};
use futures::future;
use futures::sync::mpsc::{channel, Receiver, Sender};
use futures::{Async, Future, Poll, Stream};
use futures_core::stream::Stream;
use futures_util::future::{self, FutureExt};
use tokio_reactor::Handle;
use tokio_sync::mpsc::{channel, Receiver, Sender};
use winapi::shared::minwindef::*;
use winapi::um::consoleapi::SetConsoleCtrlHandler;
use winapi::um::wincon::*;
@@ -62,9 +65,9 @@ pub(crate) struct OsExtraData {
impl Init for OsExtraData {
fn init() -> Self {
let (driver_waker, driver_rx) = channel(0);
let (driver_waker, driver_rx) = channel(1);
::tokio_executor::spawn(DriverTask { rx: driver_rx });
tokio_executor::spawn(DriverTask { rx: driver_rx });
Self { driver_waker }
}
@@ -129,7 +132,7 @@ impl Event {
}
fn new(signum: DWORD, _handle: &Handle) -> IoFuture<Event> {
let new_signal = future::poll_fn(move || {
future::lazy(move |_| {
let mut init = None;
INIT.call_once(|| {
init = Some(global_init());
@@ -139,25 +142,20 @@ impl Event {
return Err(e);
}
let (tx, rx) = channel(0);
let (tx, rx) = channel(1);
globals().register_listener(signum as EventId, tx);
Ok(Async::Ready(Event { rx }))
});
Box::new(new_signal)
Ok(Event { rx })
})
.boxed()
}
}
impl Stream for Event {
type Item = ();
type Error = io::Error;
fn poll(&mut self) -> Poll<Option<()>, io::Error> {
self.rx
.poll()
// receivers don't generate errors
.map_err(|_| unreachable!())
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.rx.poll_recv(cx)
}
}
@@ -173,26 +171,23 @@ fn global_init() -> io::Result<()> {
}
impl Future for DriverTask {
type Item = ();
type Error = ();
type Output = ();
fn poll(&mut self) -> Poll<(), ()> {
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
loop {
// Ensure we keep polling our waker until we know there are no more
// events (and therefore we've registered interest to be woken again).
match self.rx.poll() {
Ok(Async::Ready(Some(()))) => continue,
Ok(Async::Ready(None)) => panic!("driver got disconnected?"),
Ok(Async::NotReady) => break,
// receivers don't generate errors
Err(()) => unreachable!(),
match self.rx.poll_recv(cx) {
Poll::Ready(Some(())) => continue,
Poll::Ready(None) => panic!("driver got disconnected?"),
Poll::Pending => break,
}
}
globals().broadcast();
// TODO(1000): when to finish this task?
Ok(Async::NotReady)
Poll::Pending
}
}
@@ -213,20 +208,14 @@ unsafe extern "system" fn handler(ty: DWORD) -> BOOL {
#[cfg(test)]
mod tests {
use super::*;
use futures_util::future::FutureExt;
use futures_util::stream::StreamExt;
use std::time::Duration;
use tokio::runtime::current_thread;
use tokio::timer::Timeout;
use tokio_timer::Timeout;
fn with_timeout<F: Future>(future: F) -> impl Future<Item = F::Item, Error = F::Error> {
Timeout::new(future, 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")
}
})
fn with_timeout<F: Future>(future: F) -> impl Future<Output = F::Output> {
Timeout::new(future, Duration::from_secs(1)).map(|result| result.expect("timed out"))
}
#[test]
@@ -245,19 +234,16 @@ mod tests {
super::handler(CTRL_C_EVENT);
}
rt.block_on(with_timeout(event_ctrl_c.into_future()))
.ok()
.expect("failed to run event");
rt.block_on(with_timeout(event_ctrl_c.into_future()));
let event_ctrl_break = rt
.block_on(with_timeout(Event::ctrl_break()))
.expect("failed to run future");
unsafe {
super::handler(CTRL_BREAK_EVENT);
}
rt.block_on(with_timeout(event_ctrl_break.into_future()))
.ok()
.expect("failed to run event");
rt.block_on(with_timeout(event_ctrl_break.into_future()));
}
}
tokio-signal/tests/drop_multi_loop.rs
+1 -3
View File
@@ -29,7 +29,5 @@ fn dropping_loops_does_not_cause_starvation() {
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");
run_with_timeout(&mut rt, signal.into_future());
}
tokio-signal/tests/drop_then_get_a_signal.rs
+9 -10
View File
@@ -1,24 +1,23 @@
#![cfg(unix)]
#![deny(warnings, rust_2018_idioms)]
#![feature(async_await)]
use libc;
pub mod support;
use crate::support::*;
#[test]
fn drop_then_get_a_signal() {
let mut rt = CurrentThreadRuntime::new().unwrap();
let signal = run_with_timeout(&mut rt, Signal::new(libc::SIGUSR1))
#[tokio::test]
async fn drop_then_get_a_signal() {
let signal = with_timeout(Signal::new(libc::SIGUSR1))
.await
.expect("failed to create first signal");
drop(signal);
send_signal(libc::SIGUSR1);
let signal = run_with_timeout(&mut rt, Signal::new(libc::SIGUSR1))
.expect("failed to create signal")
.into_future()
.map(|_| ())
.map_err(|(e, _)| panic!("{}", e));
let signal = with_timeout(Signal::new(libc::SIGUSR1))
.await
.expect("failed to create second signal");
run_with_timeout(&mut rt, signal).expect("failed to get signal");
with_timeout(signal.into_future()).await;
}
tokio-signal/tests/dropping_does_not_deregister_other_instances.rs
+11 -13
View File
@@ -1,5 +1,6 @@
#![cfg(unix)]
#![deny(warnings, rust_2018_idioms)]
#![feature(async_await)]
use libc;
@@ -8,25 +9,22 @@ use crate::support::*;
const TEST_SIGNAL: libc::c_int = libc::SIGUSR1;
#[test]
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 = CurrentThreadRuntime::new().expect("failed to init runtime");
#[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 = run_with_timeout(&mut rt, Signal::new(TEST_SIGNAL))
let first_duplicate_signal = with_timeout(Signal::new(TEST_SIGNAL))
.await
.expect("failed to register first duplicate signal");
let signal =
run_with_timeout(&mut rt, Signal::new(TEST_SIGNAL)).expect("failed to register signal");
let second_duplicate_signal = run_with_timeout(&mut rt, Signal::new(TEST_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");
drop(first_duplicate_signal);
drop(second_duplicate_signal);
send_signal(TEST_SIGNAL);
run_with_timeout(&mut rt, signal.into_future().map_err(|(e, _)| e))
.expect("failed to get signal");
with_timeout(signal.into_future()).await;
}
tokio-signal/tests/multi_loop.rs
+1 -3
View File
@@ -22,9 +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())
.ok()
.unwrap();
run_with_timeout(&mut rt, signal.into_future());
})
})
.collect();
tokio-signal/tests/notify_both.rs
+10 -10
View File
@@ -1,22 +1,22 @@
#![cfg(unix)]
#![deny(warnings, rust_2018_idioms)]
#![feature(async_await)]
pub mod support;
use crate::support::*;
use libc;
#[test]
fn notify_both() {
let mut rt = CurrentThreadRuntime::new().unwrap();
let signal1 =
run_with_timeout(&mut rt, Signal::new(libc::SIGUSR2)).expect("failed to create signal1");
#[tokio::test]
async fn notify_both() {
let signal1 = with_timeout(Signal::new(libc::SIGUSR2))
.await
.expect("failed to create signal1");
let signal2 =
run_with_timeout(&mut rt, Signal::new(libc::SIGUSR2)).expect("failed to create signal2");
let signal2 = with_timeout(Signal::new(libc::SIGUSR2))
.await
.expect("failed to create signal2");
send_signal(libc::SIGUSR2);
run_with_timeout(&mut rt, signal1.into_future().join(signal2.into_future()))
.ok()
.expect("failed to receive");
with_timeout(future::join(signal1.into_future(), signal2.into_future())).await;
}
tokio-signal/tests/signal.rs
-18
View File
@@ -1,18 +0,0 @@
#![cfg(unix)]
#![deny(warnings, rust_2018_idioms)]
pub mod support;
use crate::support::*;
use libc;
#[test]
fn tokio_simple() {
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");
}
tokio-signal/tests/simple.rs
+7 -8
View File
@@ -1,20 +1,19 @@
#![cfg(unix)]
#![deny(warnings, rust_2018_idioms)]
#![feature(async_await)]
pub mod support;
use crate::support::*;
use libc;
#[test]
fn simple() {
let mut rt = CurrentThreadRuntime::new().unwrap();
let signal =
run_with_timeout(&mut rt, Signal::new(libc::SIGUSR1)).expect("failed to create signal");
#[tokio::test]
async fn simple() {
let signal = with_timeout(Signal::new(libc::SIGUSR1))
.await
.expect("failed to create signal");
send_signal(libc::SIGUSR1);
run_with_timeout(&mut rt, signal.into_future())
.ok()
.expect("failed to get signal");
with_timeout(signal.into_future()).await;
}
tokio-signal/tests/support.rs
+8 -13
View File
@@ -1,27 +1,22 @@
#![cfg(unix)]
#![deny(warnings, rust_2018_idioms)]
use futures_util::future::FutureExt;
use libc::{c_int, getpid, kill};
use std::future::Future;
use std::time::Duration;
use tokio::timer::Timeout;
use tokio_timer::Timeout;
pub use futures::{Future, Stream};
pub use futures_util::future;
pub use futures_util::stream::StreamExt;
pub use tokio::runtime::current_thread::{self, Runtime as CurrentThreadRuntime};
pub use tokio_signal::unix::Signal;
pub fn with_timeout<F: Future>(future: F) -> impl Future<Item = F::Item, Error = F::Error> {
Timeout::new(future, 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")
}
})
pub fn with_timeout<F: Future>(future: F) -> impl Future<Output = F::Output> {
Timeout::new(future, Duration::from_secs(1)).map(Result::unwrap)
}
pub fn run_with_timeout<F>(rt: &mut CurrentThreadRuntime, future: F) -> Result<F::Item, F::Error>
pub fn run_with_timeout<F>(rt: &mut CurrentThreadRuntime, future: F) -> F::Output
where
F: Future,
{
tokio-signal/tests/twice.rs
+13 -13
View File
@@ -1,24 +1,24 @@
#![cfg(unix)]
#![deny(warnings, rust_2018_idioms)]
#![feature(async_await)]
pub mod support;
use crate::support::*;
use libc;
#[test]
fn twice() {
let mut rt = CurrentThreadRuntime::new().unwrap();
let signal = run_with_timeout(&mut rt, Signal::new(libc::SIGUSR1)).unwrap();
#[tokio::test]
async fn twice() {
let mut signal = with_timeout(Signal::new(libc::SIGUSR1))
.await
.expect("failed to get signal");
send_signal(libc::SIGUSR1);
let (num, signal) = run_with_timeout(&mut rt, signal.into_future())
.ok()
.unwrap();
assert_eq!(num, Some(libc::SIGUSR1));
for _ in 0..2 {
send_signal(libc::SIGUSR1);
send_signal(libc::SIGUSR1);
run_with_timeout(&mut rt, signal.into_future())
.ok()
.unwrap();
let (num, sig) = with_timeout(signal.into_future()).await;
assert_eq!(num, Some(libc::SIGUSR1));
signal = sig;
}
}