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

signal: panic

Browse Source
This commit is contained in:
Markus Westerlind
2018-09-10 11:30:03 -07:00
committed by Carl Lerche
parent 1fdff707b8
commit f759e4d70f
6 changed files with 65 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+1
View File
@@ -28,6 +28,7 @@ mio-uds = "0.6"
[dev-dependencies]
tokio-core = "0.1.17"
tokio = "0.1.0"
[target.'cfg(windows)'.dependencies.winapi]
version = "0.3"
examples/multiple.rs
+1
View File
@@ -11,6 +11,7 @@ use tokio_signal::unix::{Signal, SIGINT, SIGTERM};
fn main() {
let mut core = Core::new().unwrap();
let handle = core.handle();
let handle = handle.new_tokio_handle();
// Create a stream for each of the signals we'd like to handle.
let sigint = Signal::new(SIGINT, &handle).flatten_stream();
examples/sighup-example.rs
+1 -1
View File
@@ -11,7 +11,7 @@ fn main() {
let mut core = Core::new().unwrap();
// on Unix, we can listen to whatever signal we want, in this case: SIGHUP
let stream = Signal::new(SIGHUP, &core.handle()).flatten_stream();
let stream = Signal::new(SIGHUP, &core.handle().new_tokio_handle()).flatten_stream();
println!("Waiting for SIGHUPS (Ctrl+C to quit)");
println!(
src/lib.rs
+7 -4
View File
@@ -28,6 +28,7 @@
//! fn main() {
//! let mut core = Core::new().unwrap();
//! let handle = core.handle();
//! let handle = handle.new_tokio_handle();
//!
//! // Create an infinite stream of "Ctrl+C" notifications. Each item received
//! // on this stream may represent multiple ctrl-c signals.
@@ -63,6 +64,7 @@
//! fn main() {
//! let mut core = Core::new().unwrap();
//! let handle = core.handle();
//! let handle = handle.new_tokio_handle();
//!
//! // Like the previous example, this is an infinite stream of signals
//! // being received, and signals may be coalesced while pending.
@@ -115,10 +117,11 @@ pub fn ctrl_c(handle: &Handle) -> IoFuture<IoStream<()>> {
#[cfg(unix)]
fn ctrl_c_imp(handle: &Handle) -> IoFuture<IoStream<()>> {
Box::new(
unix::Signal::new(unix::libc::SIGINT, handle)
.map(|x| Box::new(x.map(|_| ())) as Box<Stream<Item = _, Error = _> + Send>),
)
let handle = handle.clone();
Box::new(future::lazy(move || {
unix::Signal::new(unix::libc::SIGINT, &handle)
.map(|x| Box::new(x.map(|_| ())) as Box<Stream<Item = _, Error = _> + Send>)
}))
}
#[cfg(windows)]
src/unix.rs
+25 -41
View File
@@ -10,7 +10,6 @@ extern crate mio;
extern crate mio_uds;
use std::cell::UnsafeCell;
use std::collections::HashSet;
use std::io;
use std::io::prelude::*;
use std::mem;
@@ -27,7 +26,7 @@ use futures::future;
use futures::sync::mpsc::{channel, Receiver, Sender};
use futures::{Async, AsyncSink, Future};
use futures::{Poll, Sink, Stream};
use tokio_core::reactor::{CoreId, Handle, PollEvented};
use tokio_reactor::{Handle, PollEvented};
use tokio_io::IoFuture;
pub use self::libc::{SIGUSR1, SIGUSR2, SIGINT, SIGTERM};
@@ -50,7 +49,6 @@ struct Globals {
sender: UnixStream,
receiver: UnixStream,
signals: [SignalInfo; SIGNUM],
drivers: Mutex<HashSet<CoreId>>,
}
impl Default for SignalInfo {
@@ -77,7 +75,6 @@ fn globals() -> &'static Globals {
sender: sender,
receiver: receiver,
signals: Default::default(),
drivers: Mutex::new(HashSet::new()),
};
GLOBALS = Box::into_raw(Box::new(globals));
});
@@ -215,7 +212,6 @@ impl Read for EventedReceiver {
}
struct Driver {
id: CoreId,
wakeup: PollEvented<EventedReceiver>,
}
@@ -234,18 +230,10 @@ impl Future for Driver {
}
}
impl Drop for Driver {
fn drop(&mut self) {
let mut drivers = globals().drivers.lock().unwrap();
drivers.remove(&self.id);
}
}
impl Driver {
fn new(handle: &Handle) -> io::Result<Driver> {
Ok(Driver {
id: handle.id(),
wakeup: try!(PollEvented::new(EventedReceiver, handle)),
wakeup: try!(PollEvented::new_with_handle(EventedReceiver, handle)),
})
}
@@ -374,35 +362,31 @@ impl Signal {
/// multiple times. When a signal is received then all the associated
/// channels will receive the signal notification.
pub fn new(signal: c_int, handle: &Handle) -> IoFuture<Signal> {
let result = (|| {
// Turn the signal delivery on once we are ready for it
try!(signal_enable(signal));
let handle = handle.clone();
Box::new(future::lazy(move || {
let result = (|| {
// Turn the signal delivery on once we are ready for it
try!(signal_enable(signal));
// Ensure there's a driver for our associated event loop processing
// signals.
let id = handle.id();
let mut drivers = globals().drivers.lock().unwrap();
if !drivers.contains(&id) {
handle.spawn(try!(Driver::new(handle)));
drivers.insert(id);
}
drop(drivers);
// Ensure there's a driver for our associated event loop processing
// signals.
::tokio_executor::spawn(try!(Driver::new(&handle)));
// One wakeup in a queue is enough, no need for us to buffer up any
// more.
let (tx, rx) = channel(1);
let tx = Box::new(tx);
let id: *const _ = &*tx;
let idx = signal as usize;
globals().signals[idx].recipients.lock().unwrap().push(tx);
Ok(Signal {
rx: rx,
id: id,
signal: signal,
})
})();
Box::new(future::result(result))
// One wakeup in a queue is enough, no need for us to buffer up any
// more.
let (tx, rx) = channel(1);
let tx = Box::new(tx);
let id: *const _ = &*tx;
let idx = signal as usize;
globals().signals[idx].recipients.lock().unwrap().push(tx);
Ok(Signal {
rx: rx,
id: id,
signal: signal,
})
})();
future::result(result)
}))
}
}
tests/signal.rs
+30 -7
View File
@@ -2,6 +2,7 @@
extern crate futures;
extern crate libc;
extern crate tokio;
extern crate tokio_core;
extern crate tokio_signal;
@@ -10,7 +11,7 @@ use std::thread;
use std::time::Duration;
use futures::stream::Stream;
use futures::Future;
use futures::{future, Future, IntoFuture};
use tokio_core::reactor::{Core, Timeout};
use tokio_signal::unix::Signal;
@@ -18,19 +19,38 @@ use tokio_signal::unix::Signal;
fn simple() {
let mut lp = Core::new().unwrap();
let handle = lp.handle();
let signal = lp.run(Signal::new(libc::SIGUSR1, &handle)).unwrap();
let signal = lp.run(Signal::new(libc::SIGUSR1, &handle.new_tokio_handle()))
.unwrap();
unsafe {
assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0);
}
lp.run(signal.into_future()).ok().unwrap();
}
#[test]
fn tokio_simple() {
tokio::run(
future::lazy(|| {
Signal::new(libc::SIGUSR1, &tokio::reactor::Handle::default())
.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)),
)
}
#[test]
fn notify_both() {
let mut lp = Core::new().unwrap();
let handle = lp.handle();
let signal1 = lp.run(Signal::new(libc::SIGUSR2, &handle)).unwrap();
let signal2 = lp.run(Signal::new(libc::SIGUSR2, &handle)).unwrap();
let signal1 = lp.run(Signal::new(libc::SIGUSR2, &handle.new_tokio_handle()))
.unwrap();
let signal2 = lp.run(Signal::new(libc::SIGUSR2, &handle.new_tokio_handle()))
.unwrap();
unsafe {
assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR2), 0);
}
@@ -43,7 +63,8 @@ fn notify_both() {
fn drop_then_get_a_signal() {
let mut lp = Core::new().unwrap();
let handle = lp.handle();
let signal = lp.run(Signal::new(libc::SIGUSR1, &handle)).unwrap();
let signal = lp.run(Signal::new(libc::SIGUSR1, &handle.new_tokio_handle()))
.unwrap();
drop(signal);
unsafe {
assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0);
@@ -56,7 +77,8 @@ fn drop_then_get_a_signal() {
fn twice() {
let mut lp = Core::new().unwrap();
let handle = lp.handle();
let signal = lp.run(Signal::new(libc::SIGUSR1, &handle)).unwrap();
let signal = lp.run(Signal::new(libc::SIGUSR1, &handle.new_tokio_handle()))
.unwrap();
unsafe {
assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0);
}
@@ -81,7 +103,8 @@ fn multi_loop() {
thread::spawn(move || {
let mut lp = Core::new().unwrap();
let handle = lp.handle();
let signal = lp.run(Signal::new(libc::SIGHUP, &handle)).unwrap();
let signal = lp.run(Signal::new(libc::SIGHUP, &handle.new_tokio_handle()))
.unwrap();
sender.send(()).unwrap();
lp.run(signal.into_future()).ok().unwrap();
})