From eca7f0760f3f8ea9ecdf9e16e5226959da12d564 Mon Sep 17 00:00:00 2001 From: Alex Crichton Date: Tue, 6 Sep 2016 23:00:17 -0700 Subject: [PATCH] signal: Initial commit --- .gitignore | 2 + .travis.yml | 24 +++ Cargo.toml | 20 +++ LICENSE-APACHE | 201 ++++++++++++++++++++++++ LICENSE-MIT | 25 +++ README.md | 32 ++++ src/lib.rs | 28 ++++ src/unix.rs | 399 ++++++++++++++++++++++++++++++++++++++++++++++++ tests/signal.rs | 97 ++++++++++++ 9 files changed, 828 insertions(+) create mode 100644 .gitignore create mode 100644 .travis.yml create mode 100644 Cargo.toml create mode 100644 LICENSE-APACHE create mode 100644 LICENSE-MIT create mode 100644 README.md create mode 100644 src/lib.rs create mode 100644 src/unix.rs create mode 100644 tests/signal.rs diff --git a/.gitignore b/.gitignore new file mode 100644 index 000000000..a9d37c560 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +target +Cargo.lock diff --git a/.travis.yml b/.travis.yml new file mode 100644 index 000000000..82a10188d --- /dev/null +++ b/.travis.yml @@ -0,0 +1,24 @@ +language: rust + +rust: + - stable + - beta + - nightly +sudo: false +before_script: + - pip install 'travis-cargo<0.2' --user && export PATH=$HOME/.local/bin:$PATH +script: + - cargo build + - cargo test + - cargo doc --no-deps +after_success: + - travis-cargo --only nightly doc-upload +env: + global: + - secure: 6jMQVEVzg0JrRAHmGCSo93u5sFTkWQZffcsCKYvQhB2H4HCWQJ8ECykNeh4gbjIGt7hdpqCCL1C5HC+vUIp6uLHl41P2dvg7jT3P1dUcl3bSOEhjQZmfO6/JuSGrdHm34lQkIXzE5IpmY1PZKxmIOC5Sv3NU8Feu6bFuT+cHhy6TMW743S5b0XoViLocCcqGrsXhsc2rwets1zBP9WnkuNJgcPbU5ZBEcSGFufFYJFNMqajDLr1T1vYlGFH5Y3xFsbEVSSV6sOmZaE2zGlwDCW1aBflWOuRkTc1wrQUoVT/1DvlLI1Q8O/CVreGthiiVaFCm2hLbHXmUODQKYUG37yffglRSEPnsp+ia1xEVDqK77sykiU1BoORvK7EJB9rFxsKeRsg/c5V3Wi1FA1V278ZxddQqdH4gsMPMOwSAOFj7ni4CMoH/ihm5XgrAc8jzNWl1ZvdHpSfQbeeuog3b9KSev9Di0cv1JPaMYyt2Lfvex6hYqb47qjJTFbLf3j50mtAM1jrZaQsyxvuxB9eQaT9rRgS8xJWkp3fkb+26YK1n6JgCKooy62usK06pY6i4XZyFOlt3cxxh0fgd7cg7jbanZjrJkecDxBoKEneL2r1FqAN/BzYNiM1y4xH80kBvW+eZAOyfMmvXPvD6Ro2Cys5y0xr7B7wNPPibcbokhM4= +notifications: + email: + on_success: never +os: + - linux + - osx diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 000000000..b55545168 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "tokio-signal" +version = "0.1.0" +authors = ["Alex Crichton "] +license = "MIT/Apache-2.0" +repository = "https://github.com/alexcrichton/tokio-signal" +homepage = "https://github.com/alexcrichton/tokio-signal" +documentation = "https://alexcrichton.github.io/tokio-signal" +description = """ +An implementation of an asynchronous Unix signal handling backed futures. +""" + +[dependencies] +tokio-core = { git = "https://github.com/tokio-rs/tokio-core" } +futures = { git = "https://github.com/alexcrichton/futures-rs" } + +[target.'cfg(unix)'.dependencies] +tokio-uds = { git = "https://github.com/tokio-rs/tokio-uds" } +libc = "0.2" +mio = "0.6" diff --git a/LICENSE-APACHE b/LICENSE-APACHE new file mode 100644 index 000000000..16fe87b06 --- /dev/null +++ b/LICENSE-APACHE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + +TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + +1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + +2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + +3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + +4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + +5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + +6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + +7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + +8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + +9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + +END OF TERMS AND CONDITIONS + +APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + +Copyright [yyyy] [name of copyright owner] + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. diff --git a/LICENSE-MIT b/LICENSE-MIT new file mode 100644 index 000000000..28e630cf4 --- /dev/null +++ b/LICENSE-MIT @@ -0,0 +1,25 @@ +Copyright (c) 2016 Alex Crichton + +Permission is hereby granted, free of charge, to any +person obtaining a copy of this software and associated +documentation files (the "Software"), to deal in the +Software without restriction, including without +limitation the rights to use, copy, modify, merge, +publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software +is furnished to do so, subject to the following +conditions: + +The above copyright notice and this permission notice +shall be included in all copies or substantial portions +of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 000000000..6cedacb64 --- /dev/null +++ b/README.md @@ -0,0 +1,32 @@ +# tokio-signal + +An implementation of Unix signal handling for Tokio + +[![Build Status](https://travis-ci.org/alexcrichton/tokio-signal.svg?branch=master)](https://travis-ci.org/alexcrichton/tokio-signal) + +[Documentation](https://alexcrichton.github.io/tokio-signal) + +## Usage + +First, add this to your `Cargo.toml`: + +```toml +[dependencies] +tokio-signal = { git = "https://github.com/alexcrichton/tokio-signal" } +``` + +Next, add this to your crate: + +```rust +extern crate tokio_signal; +``` + +# License + +`tokio-signal` is primarily distributed under the terms of both the MIT +license and the Apache License (Version 2.0), with portions covered by various +BSD-like licenses. + +See LICENSE-APACHE, and LICENSE-MIT for details. + + diff --git a/src/lib.rs b/src/lib.rs new file mode 100644 index 000000000..2165715ab --- /dev/null +++ b/src/lib.rs @@ -0,0 +1,28 @@ +//! Asynchronous signal handling for Tokio +//! +//! This crate implements asynchronous signal handling for Tokio, and +//! asynchronous I/O framework in Rust. The primary type exported from this +//! crate, `unix::Signal`, allows listening for arbitrary signals on Unix +//! platforms, receiving them in an asynchronous fashion. +//! +//! Note that signal handling is in general a very tricky topic and should be +//! used with great care. This crate attempts to implement 'best practice' for +//! signal handling, but it should be evaluated for your own applications' needs +//! to see if it's suitable. +//! +//! The are some fundamental limitations of this crate documented on the +//! `Signal` structure as well. +//! +//! > **Note**: This crate compiles on Windows, but currently contains no +//! > bindings. Windows does not have signals like Unix does, but it +//! > does have a way to receive ctrl-c notifications at the console. +//! > It's planned that this will be bound and exported outside the +//! > `unix` module in the future! + +#![deny(missing_docs)] + +#[macro_use] +extern crate futures; +extern crate tokio_core; + +pub mod unix; diff --git a/src/unix.rs b/src/unix.rs new file mode 100644 index 000000000..f41fe38ff --- /dev/null +++ b/src/unix.rs @@ -0,0 +1,399 @@ +//! Unix-specific types for signal handling. +//! +//! This module is only defined on Unix platforms and contains the primary +//! `Signal` type for receiving notifications of signals. + +#![cfg(unix)] + +extern crate libc; +extern crate mio; +extern crate tokio_uds; + +use std::cell::RefCell; +use std::io::{self, Write, Read}; +use std::mem; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Once, ONCE_INIT, Mutex}; + +use futures::stream::{Stream, Fuse}; +use futures::{self, Future, Complete, Oneshot, Poll, Async}; +use self::libc::c_int; +use self::tokio_uds::UnixStream; +use tokio_core::io::IoFuture; +use tokio_core::{LoopHandle, Sender, Receiver, ReadinessStream}; + +static INIT: Once = ONCE_INIT; +static mut GLOBAL_STATE: *mut GlobalState = 0 as *mut _; + +/// An implementation of `Stream` for receiving a particular type of signal. +/// +/// This structure implements the `Stream` trait and represents notifications +/// of the current process receiving a particular signal. The signal being +/// listened for is passed to `Signal::new`, and the same signal number is then +/// yielded as each element for the stream. +/// +/// In general signal handling on Unix is a pretty tricky topic, and this +/// structure is no exception! There are some important limitations to keep in +/// mind when using `Signal` streams: +/// +/// * While multiple event loops are supported, the *first* event loop to +/// register a signal handler is required to be active to ensure that signals +/// for other event loops are delivered. In other words, once an event loop +/// registers a signal, it's best to keep it around and running. This is +/// normally just a problem for tests, and the "workaround" is to spawn a +/// thread in the background at the beginning of the test suite which is +/// running an event loop (and listening for a signal). +/// +/// * Signals handling in Unix already necessitates coalescing signals +/// together sometimes. This `Signal` stream is also no exception here in +/// that it will also coalesce signals. That is, even if the signal handler +/// for this process runs multiple times, the `Signal` stream may only return +/// one signal notification. Specifically, before `poll` is called, all +/// signal notifications are coalesced into one item returned from `poll`. +/// Once `poll` has been called, however, a further signal is guaranteed to +/// be yielded as an item. +/// +/// * Signal handling in general is relatively inefficient. Although some +/// improvements are possible in this crate, it's recommended to not plan on +/// having millions of signal channels open. +/// +/// * Currently the "driver task" to process incoming signals never exits. +/// +/// If you've got any questions about this feel free to open an issue on the +/// repo, though, as I'd love to chat about this! In other words, I'd love to +/// alleviate some of these limitations if possible! +pub struct Signal { + signum: c_int, + reg: ReadinessStream, + _finished: Complete<()>, +} + +struct GlobalState { + write: UnixStream, + tx: Mutex>, + signals: [GlobalSignalState; 32], +} + +struct GlobalSignalState { + ready: AtomicBool, + prev: libc::sigaction, +} + +enum Message { + NewSignal(c_int, Complete>), +} + +struct DriverTask { + handle: LoopHandle, + read: UnixStream, + rx: Fuse>, + signals: [SignalState; 32], +} + +struct SignalState { + registered: bool, + tasks: Vec<(RefCell>, mio::SetReadiness)>, +} + +impl Signal { + /// Creates a new stream which will receive notifications when the current + /// process receives the signal `signum`. + /// + /// This function will create a new stream which may be based on the + /// event loop handle provided. This function returns a future which will + /// then resolve to the signal stream, if successful. + /// + /// The `Signal` stream is an infinite stream which will receive + /// notifications whenever a signal is received. More documentation can be + /// found on `Signal` itself, but to reiterate: + /// + /// * Signals may be coalesced beyond what the kernel already does. + /// * While multiple event loops are supported, the first event loop to + /// register a signal handler must be active to deliver signal + /// notifications + /// * Once a signal handle is registered with the process the underlying + /// libc signal handler is never unregistered. + /// + /// A `Signal` stream can be created for a particular signal number + /// multiple times. When a signal is received then all the associated + /// channels will receive the signal notification. + pub fn new(signum: c_int, handle: &LoopHandle) -> IoFuture { + let mut init = None; + INIT.call_once(|| { + init = Some(global_init(handle)); + }); + let new_signal = futures::lazy(move || { + let (tx, rx) = futures::oneshot(); + let msg = Message::NewSignal(signum, tx); + let res = unsafe { + (*GLOBAL_STATE).tx.lock().unwrap().send(msg) + }; + res.expect("failed to request a new signal stream, did the \ + first event loop go away?"); + rx.then(|r| r.unwrap()) + }); + match init { + Some(init) => init.and_then(|()| new_signal).boxed(), + None => new_signal.boxed(), + } + } +} + +impl Stream for Signal { + type Item = c_int; + type Error = io::Error; + + fn poll(&mut self) -> Poll, io::Error> { + try_ready!(self.reg.poll_read()); + self.reg.get_ref() + .inner.borrow() + .as_ref().unwrap().1 + .set_readiness(mio::Ready::none()) + .expect("failed to set readiness"); + Ok(Async::Ready(Some(self.signum))) + } +} + +fn global_init(handle: &LoopHandle) -> IoFuture<()> { + let handle = handle.clone(); + let (tx, rx) = handle.clone().channel(); + let io = rx.join(UnixStream::pair(handle.clone())); + io.map(move |(rx, (read, write))| { + unsafe { + let state = Box::new(GlobalState { + write: write, + signals: { + fn new() -> GlobalSignalState { + GlobalSignalState { + ready: AtomicBool::new(false), + prev: unsafe { mem::zeroed() }, + } + } + [ + new(), new(), new(), new(), new(), new(), new(), new(), + new(), new(), new(), new(), new(), new(), new(), new(), + new(), new(), new(), new(), new(), new(), new(), new(), + new(), new(), new(), new(), new(), new(), new(), new(), + ] + }, + tx: Mutex::new(tx.clone()), + }); + GLOBAL_STATE = Box::into_raw(state); + + handle.clone().spawn(|_| { + DriverTask { + handle: handle, + rx: rx.fuse(), + read: read, + signals: { + fn new() -> SignalState { + SignalState { registered: false, tasks: Vec::new() } + } + [ + new(), new(), new(), new(), new(), new(), new(), new(), + new(), new(), new(), new(), new(), new(), new(), new(), + new(), new(), new(), new(), new(), new(), new(), new(), + new(), new(), new(), new(), new(), new(), new(), new(), + ] + }, + } + }); + } + }).boxed() +} + +impl Future for DriverTask { + type Item = (); + type Error = (); + + fn poll(&mut self) -> Poll<(), ()> { + self.check_signal_drops(); + self.check_messages(); + self.check_signals(); + + // TODO: when to finish this task? + Ok(Async::NotReady) + } +} + +impl DriverTask { + fn check_signal_drops(&mut self) { + for signal in self.signals.iter_mut() { + signal.tasks.retain(|task| { + !task.0.borrow_mut().poll().is_err() + }); + } + } + + fn check_messages(&mut self) { + loop { + // Acquire the next message + let message = match self.rx.poll() { + Ok(Async::Ready(Some(e))) => e, + Ok(Async::Ready(None)) | + Ok(Async::NotReady) => break, + Err(e) => panic!("error on rx: {}", e), + }; + let (sig, complete) = match message { + Message::NewSignal(sig, complete) => (sig, complete), + }; + + // If the signal's too large, then we return an error, otherwise we + // use this index to look at the signal slot. + // + // If the signal wasn't previously registered then we do so now. + let signal = match self.signals.get_mut(sig as usize) { + Some(signal) => signal, + None => { + complete.complete(Err(io::Error::new(io::ErrorKind::Other, + "signum too large"))); + continue + } + }; + if !signal.registered { + unsafe { + let mut new: libc::sigaction = mem::zeroed(); + new.sa_sigaction = handler as usize; + new.sa_flags = libc::SA_RESTART | libc::SA_SIGINFO; + let mut prev = mem::zeroed(); + if libc::sigaction(sig, &new, &mut prev) != 0 { + complete.complete(Err(io::Error::last_os_error())); + continue + } + signal.registered = true; + } + } + + // Acquire the (registration, set_readiness) pair by... assuming + // we're on the event loop (true because of the spawn above). + let reg = MyRegistration { inner: RefCell::new(None) }; + let mut new = ReadinessStream::new(self.handle.clone(), reg); + let reg = match new.poll() { + Ok(Async::Ready(reg)) => reg, + Ok(Async::NotReady) => panic!("should be on event loop"), + Err(e) => { + complete.complete(Err(e)); + continue + } + }; + + // Create the `Signal` to pass back and then also keep a handle to + // the `SetReadiness` for ourselves internally. + let (tx, rx) = futures::oneshot(); + let ready = reg.get_ref().inner.borrow_mut().as_mut().unwrap().1.clone(); + complete.complete(Ok(Signal { + signum: sig, + reg: reg, + _finished: tx, + })); + signal.tasks.push((RefCell::new(rx), ready)); + } + } + + fn check_signals(&mut self) { + // Drain all data from the pipe + let mut buf = [0; 32]; + let mut any = false; + loop { + match self.read.read(&mut buf) { + Ok(0) => { // EOF == something happened + any = true; + break + } + Ok(..) => any = true, // data read, but keep draining + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break, + Err(e) => panic!("bad read: {}", e), + } + } + + // If nothing happened, no need to check the signals + if !any { + return + } + + for (i, slot) in self.signals.iter().enumerate() { + // No need to go farther if we haven't even registered a signal + if !slot.registered { + continue + } + + // See if this signal actually happened since we last checked + unsafe { + if !(*GLOBAL_STATE).signals[i].ready.swap(false, Ordering::SeqCst) { + continue + } + } + + // Wake up all the tasks waiting on this signal + for task in slot.tasks.iter() { + task.1.set_readiness(mio::Ready::readable()) + .expect("failed to set readiness"); + } + } + } +} + +extern fn handler(signum: c_int, + info: *mut libc::siginfo_t, + ptr: *mut libc::c_void) { + type FnSigaction = extern fn(c_int, *mut libc::siginfo_t, *mut libc::c_void); + type FnHandler = extern fn(c_int); + + unsafe { + let state = match (*GLOBAL_STATE).signals.get(signum as usize) { + Some(state) => state, + None => return, + }; + + if !state.ready.swap(true, Ordering::SeqCst) { + match (&(*GLOBAL_STATE).write).write(&[1]) { + Ok(..) => {} + Err(e) => { + if e.kind() != io::ErrorKind::WouldBlock { + panic!("bad error on write fd: {}", e) + } + } + } + } + + let fnptr = state.prev.sa_sigaction; + if fnptr == 0 || fnptr == libc::SIG_DFL || fnptr == libc::SIG_IGN { + return + } + if state.prev.sa_flags & libc::SA_SIGINFO == 0 { + let action = mem::transmute::(fnptr); + action(signum) + } else { + let action = mem::transmute::(fnptr); + action(signum, info, ptr) + } + } +} + +struct MyRegistration { + inner: RefCell>, +} + +impl mio::Evented for MyRegistration { + fn register(&self, + poll: &mio::Poll, + token: mio::Token, + events: mio::Ready, + opts: mio::PollOpt) -> io::Result<()> { + let reg = mio::Registration::new(poll, token, events, opts); + *self.inner.borrow_mut() = Some(reg); + Ok(()) + } + + fn reregister(&self, + _poll: &mio::Poll, + _token: mio::Token, + _events: mio::Ready, + _opts: mio::PollOpt) -> io::Result<()> { + Ok(()) + } + + fn deregister(&self, _poll: &mio::Poll) -> io::Result<()> { + Ok(()) + } +} diff --git a/tests/signal.rs b/tests/signal.rs new file mode 100644 index 000000000..ef7ccb771 --- /dev/null +++ b/tests/signal.rs @@ -0,0 +1,97 @@ +#![cfg(unix)] + +extern crate futures; +extern crate libc; +extern crate tokio_core; +extern crate tokio_signal; + +use std::sync::mpsc::channel; +use std::sync::{Once, ONCE_INIT, Mutex, MutexGuard}; +use std::thread; +use std::time::Duration; + +use futures::Future; +use futures::stream::Stream; +use tokio_core::Loop; +use tokio_signal::unix::Signal; + +static INIT: Once = ONCE_INIT; +static mut LOCK: *mut Mutex<()> = 0 as *mut _; + +fn lock() -> MutexGuard<'static, ()> { + unsafe { + INIT.call_once(|| { + LOCK = Box::into_raw(Box::new(Mutex::new(()))); + let (tx, rx) = channel(); + thread::spawn(move || { + let mut lp = Loop::new().unwrap(); + let handle = lp.handle(); + let _signal = lp.run(Signal::new(libc::SIGALRM, &handle)).unwrap(); + tx.send(()).unwrap(); + drop(lp.run(futures::empty::<(), ()>())); + }); + rx.recv().unwrap(); + }); + (*LOCK).lock().unwrap() + } +} + +#[test] +fn simple() { + let _lock = lock(); + + let mut lp = Loop::new().unwrap(); + let handle = lp.handle(); + let signal = lp.run(Signal::new(libc::SIGUSR1, &handle)).unwrap(); + unsafe { + assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0); + } + lp.run(signal.into_future()).ok().unwrap(); +} + +#[test] +fn notify_both() { + let _lock = lock(); + + let mut lp = Loop::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(); + unsafe { + assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR2), 0); + } + lp.run(signal1.into_future().join(signal2.into_future())).ok().unwrap(); +} + +#[test] +fn drop_then_get_a_signal() { + let _lock = lock(); + + let mut lp = Loop::new().unwrap(); + let handle = lp.handle(); + let signal = lp.run(Signal::new(libc::SIGUSR1, &handle)).unwrap(); + drop(signal); + unsafe { + assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0); + } + let timeout = lp.handle().timeout(Duration::from_millis(1)); + lp.run(timeout.and_then(|t| t)).unwrap(); +} + +#[test] +fn twice() { + let _lock = lock(); + + let mut lp = Loop::new().unwrap(); + let handle = lp.handle(); + let signal = lp.run(Signal::new(libc::SIGUSR1, &handle)).unwrap(); + unsafe { + assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0); + } + let (num, signal) = lp.run(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(); +}