From f759e4d70f12b6074912311e87058433b9bbf6b7 Mon Sep 17 00:00:00 2001 From: Markus Westerlind Date: Sat, 5 May 2018 00:33:02 +0200 Subject: [PATCH] signal: panic --- Cargo.toml | 1 + examples/multiple.rs | 1 + examples/sighup-example.rs | 2 +- src/lib.rs | 11 ++++--- src/unix.rs | 66 +++++++++++++++----------------------- tests/signal.rs | 37 +++++++++++++++++---- 6 files changed, 65 insertions(+), 53 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index b7829fe4b..92ad04524 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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" diff --git a/examples/multiple.rs b/examples/multiple.rs index 093b83c6a..39c0e5031 100644 --- a/examples/multiple.rs +++ b/examples/multiple.rs @@ -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(); diff --git a/examples/sighup-example.rs b/examples/sighup-example.rs index 88da9b8c3..a9a76db7a 100644 --- a/examples/sighup-example.rs +++ b/examples/sighup-example.rs @@ -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!( diff --git a/src/lib.rs b/src/lib.rs index a3423b0d1..1fa8cff15 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -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> { #[cfg(unix)] fn ctrl_c_imp(handle: &Handle) -> IoFuture> { - Box::new( - unix::Signal::new(unix::libc::SIGINT, handle) - .map(|x| Box::new(x.map(|_| ())) as Box + 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 + Send>) + })) } #[cfg(windows)] diff --git a/src/unix.rs b/src/unix.rs index 9971804f2..df5c17e37 100644 --- a/src/unix.rs +++ b/src/unix.rs @@ -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>, } 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, } @@ -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 { 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 { - 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) + })) } } diff --git a/tests/signal.rs b/tests/signal.rs index c16c5bc19..a53a49b27 100644 --- a/tests/signal.rs +++ b/tests/signal.rs @@ -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(); })