mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-08-26 00:00:16 +02:00
signal: move into tokio-net (#1463)
This commit is contained in:
Carl Lerche
@@ -40,6 +40,9 @@
|
||||
pub mod driver;
|
||||
pub mod util;
|
||||
|
||||
#[cfg(feature = "signal")]
|
||||
pub mod signal;
|
||||
|
||||
#[cfg(feature = "tcp")]
|
||||
pub mod tcp;
|
||||
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
#[cfg(unix)]
|
||||
use super::unix::Signal as Inner;
|
||||
#[cfg(windows)]
|
||||
use super::windows::Event as Inner;
|
||||
use crate::driver::Handle;
|
||||
|
||||
use futures_core::stream::Stream;
|
||||
use std::io;
|
||||
use std::pin::Pin;
|
||||
use std::task::{Context, Poll};
|
||||
|
||||
/// Represents a stream which receives "ctrl-c" notifications sent to the process.
|
||||
///
|
||||
/// In general signals are handled very differently across Unix and Windows, but
|
||||
/// this is somewhat cross platform in terms of how it can be handled. A ctrl-c
|
||||
/// event to a console process can be represented as a stream for both Windows
|
||||
/// and Unix.
|
||||
///
|
||||
/// Note that there are a number of caveats listening for signals, and you may
|
||||
/// wish to read up on the documentation in the `unix` or `windows` module to
|
||||
/// take a peek.
|
||||
///
|
||||
/// Notably, a notification to this process notifies *all* streams listening to
|
||||
/// this event. Moreover, the notifications **are coalesced** if they aren't processed
|
||||
/// quickly enough. This means that if two notifications are received back-to-back,
|
||||
/// then the stream may only receive one item about the two notifications.
|
||||
#[must_use = "streams do nothing unless polled"]
|
||||
#[derive(Debug)]
|
||||
pub struct CtrlC {
|
||||
inner: Inner,
|
||||
}
|
||||
|
||||
impl CtrlC {
|
||||
/// Creates a new stream which receives "ctrl-c" notifications sent to the
|
||||
/// process.
|
||||
///
|
||||
/// This function binds to the default reactor.
|
||||
pub fn new() -> io::Result<Self> {
|
||||
Self::with_handle(&Handle::default())
|
||||
}
|
||||
|
||||
/// Creates a new stream which receives "ctrl-c" notifications sent to the
|
||||
/// process.
|
||||
///
|
||||
/// This function binds to reactor specified by `handle`.
|
||||
pub fn with_handle(handle: &Handle) -> io::Result<Self> {
|
||||
Inner::ctrl_c(handle).map(|inner| Self { inner })
|
||||
}
|
||||
}
|
||||
|
||||
impl Stream for CtrlC {
|
||||
type Item = ();
|
||||
|
||||
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||
Pin::new(&mut self.inner).poll_next(cx)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
//! Asynchronous signal handling for Tokio
|
||||
//!
|
||||
//! This crate implements asynchronous signal handling for Tokio, an
|
||||
//! 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.
|
||||
//!
|
||||
//! # Examples
|
||||
//!
|
||||
//! Print out all ctrl-C notifications received
|
||||
//!
|
||||
//! ```rust,no_run
|
||||
//! #![feature(async_await)]
|
||||
//!
|
||||
//! use tokio_net::signal;
|
||||
//!
|
||||
//! 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 = signal::CtrlC::new()?;
|
||||
//!
|
||||
//! // Process each ctrl-c as it comes in
|
||||
//! let prog = ctrl_c.for_each(|_| {
|
||||
//! println!("ctrl-c received!");
|
||||
//! future::ready(())
|
||||
//! });
|
||||
//!
|
||||
//! prog.await;
|
||||
//!
|
||||
//! Ok(())
|
||||
//! }
|
||||
//! ```
|
||||
//!
|
||||
//! Wait for SIGHUP on Unix
|
||||
//!
|
||||
//! ```rust,no_run
|
||||
//! #![feature(async_await)]
|
||||
//! # #[cfg(unix)] {
|
||||
//!
|
||||
//! use tokio_net::signal::{self, unix::{Signal, SignalKind}};
|
||||
//!
|
||||
//! 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 = signal::CtrlC::new()?;
|
||||
//!
|
||||
//! // Process each ctrl-c as it comes in
|
||||
//! let prog = ctrl_c.for_each(|_| {
|
||||
//! 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(SignalKind::hangup())?;
|
||||
//!
|
||||
//! // Convert out stream into a future and block the program
|
||||
//! let (signal, _signal) = stream.into_future().await;
|
||||
//! println!("got signal {:?}", signal);
|
||||
//! Ok(())
|
||||
//! }
|
||||
//! # }
|
||||
//! ```
|
||||
|
||||
mod ctrl_c;
|
||||
mod registry;
|
||||
|
||||
mod os {
|
||||
#[cfg(unix)]
|
||||
pub(crate) use super::unix::{OsExtraData, OsStorage};
|
||||
#[cfg(windows)]
|
||||
pub(crate) use super::windows::{OsExtraData, OsStorage};
|
||||
}
|
||||
|
||||
pub mod unix;
|
||||
pub mod windows;
|
||||
|
||||
pub use self::ctrl_c::CtrlC;
|
||||
@@ -0,0 +1,303 @@
|
||||
use super::os::{OsExtraData, OsStorage};
|
||||
|
||||
use tokio_sync::mpsc::Sender;
|
||||
|
||||
use lazy_static::lazy_static;
|
||||
use std::ops;
|
||||
use std::pin::Pin;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Mutex;
|
||||
|
||||
pub(crate) type EventId = usize;
|
||||
|
||||
/// State for a specific event, whether a notification is pending delivery,
|
||||
/// and what listeners are registered.
|
||||
#[derive(Default, Debug)]
|
||||
pub(crate) struct EventInfo {
|
||||
pending: AtomicBool,
|
||||
recipients: Mutex<Vec<Sender<()>>>,
|
||||
}
|
||||
|
||||
/// An interface for retrieving the `EventInfo` for a particular eventId.
|
||||
pub(crate) trait Storage {
|
||||
/// Get the `EventInfo` for `id` if it exists.
|
||||
fn event_info(&self, id: EventId) -> Option<&EventInfo>;
|
||||
|
||||
/// Invoke `f` once for each defined `EventInfo` in this storage.
|
||||
fn for_each<'a, F>(&'a self, f: F)
|
||||
where
|
||||
F: FnMut(&'a EventInfo);
|
||||
}
|
||||
|
||||
impl Storage for Vec<EventInfo> {
|
||||
fn event_info(&self, id: EventId) -> Option<&EventInfo> {
|
||||
self.get(id)
|
||||
}
|
||||
|
||||
fn for_each<'a, F>(&'a self, f: F)
|
||||
where
|
||||
F: FnMut(&'a EventInfo),
|
||||
{
|
||||
self.iter().for_each(f)
|
||||
}
|
||||
}
|
||||
|
||||
/// An interface for initializing a type. Useful for situations where we cannot
|
||||
/// inject a configured instance in the constructor of another type.
|
||||
pub(crate) trait Init {
|
||||
fn init() -> Self;
|
||||
}
|
||||
|
||||
/// Manages and distributes event notifications to any registered listeners.
|
||||
///
|
||||
/// Generic over the underlying storage to allow for domain specific
|
||||
/// optimizations (e.g. eventIds may or may not be contiguous).
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct Registry<S> {
|
||||
storage: S,
|
||||
}
|
||||
|
||||
impl<S> Registry<S> {
|
||||
fn new(storage: S) -> Self {
|
||||
Self { storage }
|
||||
}
|
||||
}
|
||||
|
||||
impl<S: Storage> Registry<S> {
|
||||
/// Register a new listener for `event_id`.
|
||||
fn register_listener(&self, event_id: EventId, listener: Sender<()>) {
|
||||
self.storage
|
||||
.event_info(event_id)
|
||||
.unwrap_or_else(|| panic!("invalid event_id: {}", event_id))
|
||||
.recipients
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push(listener);
|
||||
}
|
||||
|
||||
/// Mark `event_id` as having been delivered, without broadcasting it to
|
||||
/// any listeners.
|
||||
fn record_event(&self, event_id: EventId) {
|
||||
if let Some(event_info) = self.storage.event_info(event_id) {
|
||||
event_info.pending.store(true, Ordering::SeqCst)
|
||||
}
|
||||
}
|
||||
|
||||
/// Broadcast all previously recorded events to their respective listeners.
|
||||
///
|
||||
/// Returns true if an event was delivered to at least one listener.
|
||||
fn broadcast(&self) -> bool {
|
||||
let mut did_notify = false;
|
||||
self.storage.for_each(|event_info| {
|
||||
// Any signal of this kind arrived since we checked last?
|
||||
if !event_info.pending.swap(false, Ordering::SeqCst) {
|
||||
return;
|
||||
}
|
||||
|
||||
let mut recipients = event_info.recipients.lock().unwrap();
|
||||
|
||||
// Notify all waiters on this signal that the signal has been
|
||||
// received. If we can't push a message into the queue then we don't
|
||||
// worry about it as everything is coalesced anyway. If the channel
|
||||
// has gone away then we can remove that slot.
|
||||
for i in (0..recipients.len()).rev() {
|
||||
match recipients[i].try_send(()) {
|
||||
Ok(()) => did_notify = true,
|
||||
Err(ref e) if e.is_closed() => {
|
||||
recipients.swap_remove(i);
|
||||
}
|
||||
|
||||
// Channel is full, ignore the error since the
|
||||
// receiver has already been woken up
|
||||
Err(e) => {
|
||||
// Sanity check in case this error type ever gets
|
||||
// additional variants we have not considered.
|
||||
debug_assert!(e.is_full());
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
did_notify
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct Globals {
|
||||
extra: OsExtraData,
|
||||
registry: Registry<OsStorage>,
|
||||
}
|
||||
|
||||
impl ops::Deref for Globals {
|
||||
type Target = OsExtraData;
|
||||
|
||||
fn deref(&self) -> &Self::Target {
|
||||
&self.extra
|
||||
}
|
||||
}
|
||||
|
||||
impl Globals {
|
||||
/// Register a new listener for `event_id`.
|
||||
pub(crate) fn register_listener(&self, event_id: EventId, listener: Sender<()>) {
|
||||
self.registry.register_listener(event_id, listener);
|
||||
}
|
||||
|
||||
/// Mark `event_id` as having been delivered, without broadcasting it to
|
||||
/// any listeners.
|
||||
pub(crate) fn record_event(&self, event_id: EventId) {
|
||||
self.registry.record_event(event_id);
|
||||
}
|
||||
|
||||
/// Broadcast all previously recorded events to their respective listeners.
|
||||
///
|
||||
/// Returns true if an event was delivered to at least one listener.
|
||||
pub(crate) fn broadcast(&self) -> bool {
|
||||
self.registry.broadcast()
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
pub(crate) fn storage(&self) -> &OsStorage {
|
||||
&self.registry.storage
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn globals() -> Pin<&'static Globals>
|
||||
where
|
||||
OsExtraData: 'static + Send + Sync + Init,
|
||||
OsStorage: 'static + Send + Sync + Init,
|
||||
{
|
||||
lazy_static! {
|
||||
static ref GLOBALS: Pin<Box<Globals>> = Pin::new(Box::new(Globals {
|
||||
extra: OsExtraData::init(),
|
||||
registry: Registry::new(OsStorage::init()),
|
||||
}));
|
||||
}
|
||||
|
||||
GLOBALS.as_ref()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use futures_util::{future, StreamExt};
|
||||
use tokio_sync::mpsc::channel;
|
||||
use tokio_sync::oneshot;
|
||||
|
||||
#[tokio::test]
|
||||
async fn smoke() {
|
||||
let registry = Registry::new(vec![
|
||||
EventInfo::default(),
|
||||
EventInfo::default(),
|
||||
EventInfo::default(),
|
||||
]);
|
||||
|
||||
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();
|
||||
|
||||
tokio::spawn(async {
|
||||
wait.await.expect("wait failed");
|
||||
|
||||
// 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();
|
||||
|
||||
// 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) = all.await;
|
||||
assert_eq!(2, first_results.len());
|
||||
assert_eq!(1, second_results.len());
|
||||
assert_eq!(0, third_results.len());
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[should_panic = "invalid event_id: 1"]
|
||||
fn register_panics_on_invalid_input() {
|
||||
let registry = Registry::new(vec![EventInfo::default()]);
|
||||
|
||||
let (tx, _) = channel(1);
|
||||
registry.register_listener(1, tx);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn record_invalid_event_does_nothing() {
|
||||
let registry = Registry::new(vec![EventInfo::default()]);
|
||||
registry.record_event(42);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn broadcast_cleans_up_disconnected_listeners() {
|
||||
let registry = Registry::new(vec![EventInfo::default()]);
|
||||
|
||||
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);
|
||||
registry.register_listener(0, third_tx);
|
||||
|
||||
drop(first_rx);
|
||||
drop(second_rx);
|
||||
|
||||
let (fire, wait) = oneshot::channel();
|
||||
|
||||
tokio::spawn(async {
|
||||
wait.await.expect("wait failed");
|
||||
|
||||
registry.record_event(0);
|
||||
registry.broadcast();
|
||||
|
||||
assert_eq!(1, registry.storage[0].recipients.lock().unwrap().len());
|
||||
drop(registry);
|
||||
});
|
||||
|
||||
let _ = fire.send(());
|
||||
let results: Vec<()> = third_rx.collect().await;
|
||||
|
||||
assert_eq!(1, results.len());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn broadcast_returns_if_at_least_one_event_fired() {
|
||||
let registry = Registry::new(vec![EventInfo::default()]);
|
||||
|
||||
registry.record_event(0);
|
||||
assert_eq!(false, registry.broadcast());
|
||||
|
||||
let (first_tx, first_rx) = channel(1);
|
||||
let (second_tx, second_rx) = channel(1);
|
||||
|
||||
registry.register_listener(0, first_tx);
|
||||
registry.register_listener(0, second_tx);
|
||||
|
||||
registry.record_event(0);
|
||||
assert_eq!(true, registry.broadcast());
|
||||
|
||||
drop(first_rx);
|
||||
registry.record_event(0);
|
||||
assert_eq!(false, registry.broadcast());
|
||||
|
||||
drop(second_rx);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,451 @@
|
||||
//! 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)]
|
||||
|
||||
use super::registry::{globals, EventId, EventInfo, Globals, Init, Storage};
|
||||
use crate::driver::Handle;
|
||||
use crate::util::PollEvented;
|
||||
|
||||
use tokio_io::AsyncRead;
|
||||
use tokio_sync::mpsc::{channel, Receiver};
|
||||
|
||||
pub use libc;
|
||||
|
||||
use futures_core::stream::Stream;
|
||||
use libc::c_int;
|
||||
use mio_uds::UnixStream;
|
||||
use std::future::Future;
|
||||
use std::io::{self, Error, ErrorKind, Write};
|
||||
use std::pin::Pin;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Once;
|
||||
use std::task::{Context, Poll};
|
||||
|
||||
pub(crate) type OsStorage = Vec<SignalInfo>;
|
||||
|
||||
// Number of different unix signals
|
||||
// (FreeBSD has 33)
|
||||
const SIGNUM: usize = 33;
|
||||
|
||||
impl Init for OsStorage {
|
||||
fn init() -> Self {
|
||||
(0..SIGNUM).map(|_| SignalInfo::default()).collect()
|
||||
}
|
||||
}
|
||||
|
||||
impl Storage for OsStorage {
|
||||
fn event_info(&self, id: EventId) -> Option<&EventInfo> {
|
||||
self.get(id).map(|si| &si.event_info)
|
||||
}
|
||||
|
||||
fn for_each<'a, F>(&'a self, f: F)
|
||||
where
|
||||
F: FnMut(&'a EventInfo),
|
||||
{
|
||||
self.iter().map(|si| &si.event_info).for_each(f)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct OsExtraData {
|
||||
sender: UnixStream,
|
||||
receiver: UnixStream,
|
||||
}
|
||||
|
||||
impl Init for OsExtraData {
|
||||
fn init() -> Self {
|
||||
let (receiver, sender) = UnixStream::pair().expect("failed to create UnixStream");
|
||||
|
||||
Self { sender, receiver }
|
||||
}
|
||||
}
|
||||
|
||||
/// Represents the specific kind of signal to listen for.
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub struct SignalKind(c_int);
|
||||
|
||||
impl SignalKind {
|
||||
/// Allows for listening to any valid OS signal.
|
||||
///
|
||||
/// For example, this can be used for listening for platform-specific
|
||||
/// signals.
|
||||
/// ```rust,no_run
|
||||
/// # use tokio_net::signal::unix::SignalKind;
|
||||
/// # let signum = -1;
|
||||
/// // let signum = libc::OS_SPECIFIC_SIGNAL;
|
||||
/// let kind = SignalKind::from_raw(signum);
|
||||
/// ```
|
||||
pub fn from_raw(signum: c_int) -> Self {
|
||||
Self(signum)
|
||||
}
|
||||
|
||||
/// Represents the SIGALRM signal.
|
||||
///
|
||||
/// On Unix systems this signal is sent when a real-time timer has expired.
|
||||
/// By default, the process is terminated by this signal.
|
||||
pub fn alarm() -> Self {
|
||||
Self(libc::SIGALRM)
|
||||
}
|
||||
|
||||
/// Represents the SIGCHLD signal.
|
||||
///
|
||||
/// On Unix systems this signal is sent when the status of a child process
|
||||
/// has changed. By default, this signal is ignored.
|
||||
pub fn child() -> Self {
|
||||
Self(libc::SIGCHLD)
|
||||
}
|
||||
|
||||
/// Represents the SIGHUP signal.
|
||||
///
|
||||
/// On Unix systems this signal is sent when the terminal is disconnected.
|
||||
/// By default, the process is terminated by this signal.
|
||||
pub fn hangup() -> Self {
|
||||
Self(libc::SIGHUP)
|
||||
}
|
||||
|
||||
/// Represents the SIGINFO signal.
|
||||
///
|
||||
/// On Unix systems this signal is sent to request a status update from the
|
||||
/// process. By default, this signal is ignored.
|
||||
#[cfg(any(
|
||||
target_os = "dragonfly",
|
||||
target_os = "freebsd",
|
||||
target_os = "macos",
|
||||
target_os = "netbsd",
|
||||
target_os = "openbsd"
|
||||
))]
|
||||
pub fn info() -> Self {
|
||||
Self(libc::SIGINFO)
|
||||
}
|
||||
|
||||
/// Represents the SIGINT signal.
|
||||
///
|
||||
/// On Unix systems this signal is sent to interrupt a program.
|
||||
/// By default, the process is terminated by this signal.
|
||||
pub fn interrupt() -> Self {
|
||||
Self(libc::SIGINT)
|
||||
}
|
||||
|
||||
/// Represents the SIGIO signal.
|
||||
///
|
||||
/// On Unix systems this signal is sent when I/O operations are possible
|
||||
/// on some file descriptor. By default, this signal is ignored.
|
||||
pub fn io() -> Self {
|
||||
Self(libc::SIGIO)
|
||||
}
|
||||
|
||||
/// Represents the SIGPIPE signal.
|
||||
///
|
||||
/// On Unix systems this signal is sent when the process attempts to write
|
||||
/// to a pipe which has no reader. By default, the process is terminated by
|
||||
/// this signal.
|
||||
pub fn pipe() -> Self {
|
||||
Self(libc::SIGPIPE)
|
||||
}
|
||||
|
||||
/// Represents the SIGQUIT signal.
|
||||
///
|
||||
/// On Unix systems this signal is sent to issue a shutdown of the
|
||||
/// process, after which the OS will dump the process core.
|
||||
/// By default, the process is terminated by this signal.
|
||||
pub fn quit() -> Self {
|
||||
Self(libc::SIGQUIT)
|
||||
}
|
||||
|
||||
/// Represents the SIGTERM signal.
|
||||
///
|
||||
/// On Unix systems this signal is sent to issue a shutdown of the
|
||||
/// process. By default, the process is terminated by this signal.
|
||||
pub fn terminate() -> Self {
|
||||
Self(libc::SIGTERM)
|
||||
}
|
||||
|
||||
/// Represents the SIGUSR1 signal.
|
||||
///
|
||||
/// On Unix systems this is a user defined signal.
|
||||
/// By default, the process is terminated by this signal.
|
||||
pub fn user_defined1() -> Self {
|
||||
Self(libc::SIGUSR1)
|
||||
}
|
||||
|
||||
/// Represents the SIGUSR2 signal.
|
||||
///
|
||||
/// On Unix systems this is a user defined signal.
|
||||
/// By default, the process is terminated by this signal.
|
||||
pub fn user_defined2() -> Self {
|
||||
Self(libc::SIGUSR2)
|
||||
}
|
||||
|
||||
/// Represents the SIGWINCH signal.
|
||||
///
|
||||
/// On Unix systems this signal is sent when the terminal window is resized.
|
||||
/// By default, this signal is ignored.
|
||||
pub fn window_change() -> Self {
|
||||
Self(libc::SIGWINCH)
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct SignalInfo {
|
||||
event_info: EventInfo,
|
||||
init: Once,
|
||||
initialized: AtomicBool,
|
||||
}
|
||||
|
||||
impl Default for SignalInfo {
|
||||
fn default() -> SignalInfo {
|
||||
SignalInfo {
|
||||
event_info: Default::default(),
|
||||
init: Once::new(),
|
||||
initialized: AtomicBool::new(false),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Our global signal handler for all signals registered by this module.
|
||||
///
|
||||
/// The purpose of this signal handler is to primarily:
|
||||
///
|
||||
/// 1. Flag that our specific signal was received (e.g. store an atomic flag)
|
||||
/// 2. Wake up driver tasks by writing a byte to a pipe
|
||||
///
|
||||
/// Those two operations shoudl both be async-signal safe.
|
||||
fn action(globals: Pin<&'static Globals>, signal: c_int) {
|
||||
globals.record_event(signal as EventId);
|
||||
|
||||
// Send a wakeup, ignore any errors (anything reasonably possible is
|
||||
// full pipe and then it will wake up anyway).
|
||||
let mut sender = &globals.sender;
|
||||
drop(sender.write(&[1]));
|
||||
}
|
||||
|
||||
/// Enable this module to receive signal notifications for the `signal`
|
||||
/// provided.
|
||||
///
|
||||
/// This will register the signal handler if it hasn't already been registered,
|
||||
/// returning any error along the way if that fails.
|
||||
fn signal_enable(signal: c_int) -> io::Result<()> {
|
||||
if signal < 0 || signal_hook_registry::FORBIDDEN.contains(&signal) {
|
||||
return Err(Error::new(
|
||||
ErrorKind::Other,
|
||||
format!("Refusing to register signal {}", signal),
|
||||
));
|
||||
}
|
||||
|
||||
let globals = globals();
|
||||
let siginfo = match globals.storage().get(signal as EventId) {
|
||||
Some(slot) => slot,
|
||||
None => return Err(io::Error::new(io::ErrorKind::Other, "signal too large")),
|
||||
};
|
||||
let mut registered = Ok(());
|
||||
siginfo.init.call_once(|| {
|
||||
registered = unsafe {
|
||||
signal_hook_registry::register(signal, move || action(globals, signal)).map(|_| ())
|
||||
};
|
||||
if registered.is_ok() {
|
||||
siginfo.initialized.store(true, Ordering::Relaxed);
|
||||
}
|
||||
});
|
||||
registered?;
|
||||
// If the call_once failed, it won't be retried on the next attempt to register the signal. In
|
||||
// such case it is not run, registered is still `Ok(())`, initialized is still false.
|
||||
if siginfo.initialized.load(Ordering::Relaxed) {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(Error::new(
|
||||
ErrorKind::Other,
|
||||
"Failed to register signal handler",
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct Driver {
|
||||
wakeup: PollEvented<UnixStream>,
|
||||
}
|
||||
|
||||
impl Future for Driver {
|
||||
type Output = ();
|
||||
|
||||
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(cx);
|
||||
// Broadcast any signals which were received
|
||||
globals().broadcast();
|
||||
|
||||
Poll::Pending
|
||||
}
|
||||
}
|
||||
|
||||
impl Driver {
|
||||
fn new(handle: &Handle) -> io::Result<Driver> {
|
||||
// NB: We give each driver a "fresh" reciever file descriptor to avoid
|
||||
// the issues described in alexcrichton/tokio-process#42.
|
||||
//
|
||||
// In the past we would reuse the actual receiver file descriptor and
|
||||
// swallow any errors around double registration of the same descriptor.
|
||||
// I'm not sure if the second (failed) registration simply doesn't end up
|
||||
// receiving wake up notifications, or there could be some race condition
|
||||
// when consuming readiness events, but having distinct descriptors for
|
||||
// distinct PollEvented instances appears to mitigate this.
|
||||
//
|
||||
// Unfortunately we cannot just use a single global PollEvented instance
|
||||
// either, since we can't compare Handles or assume they will always
|
||||
// point to the exact same reactor.
|
||||
let stream = globals().receiver.try_clone()?;
|
||||
let wakeup = PollEvented::new_with_handle(stream, handle)?;
|
||||
|
||||
Ok(Driver { wakeup })
|
||||
}
|
||||
|
||||
/// Drain all data in the global receiver, ensuring we'll get woken up when
|
||||
/// there is a write on the other end.
|
||||
///
|
||||
/// 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: Pin<&mut Self>, cx: &mut Context<'_>) {
|
||||
loop {
|
||||
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,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 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:
|
||||
///
|
||||
/// * 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.
|
||||
///
|
||||
/// Put another way, any element pulled off the returned stream corresponds to
|
||||
/// *at least one* signal, but possibly more.
|
||||
///
|
||||
/// * 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. This
|
||||
/// driver task runs in the background of the event loop provided, and
|
||||
/// in general you shouldn't need to worry about it.
|
||||
///
|
||||
/// 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!
|
||||
#[must_use = "streams do nothing unless polled"]
|
||||
#[derive(Debug)]
|
||||
pub struct Signal {
|
||||
driver: Driver,
|
||||
rx: Receiver<()>,
|
||||
}
|
||||
|
||||
impl Signal {
|
||||
/// Creates a new stream which will receive notifications when the current
|
||||
/// process receives the signal `signal`.
|
||||
///
|
||||
/// This function will create a new stream which binds to the default reactor.
|
||||
/// 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.
|
||||
/// * Once a signal handler 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.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// * If the lower-level C functions fail for some reason.
|
||||
/// * If the previous initialization of this specific signal failed.
|
||||
/// * If the signal is one of
|
||||
/// [`signal_hook::FORBIDDEN`](https://docs.rs/signal-hook/*/signal_hook/fn.register.html#panics)
|
||||
pub fn new(kind: SignalKind) -> io::Result<Self> {
|
||||
Signal::with_handle(kind, &Handle::default())
|
||||
}
|
||||
|
||||
/// Creates a new stream which will receive notifications when the current
|
||||
/// process receives the signal `signal`.
|
||||
///
|
||||
/// This function will create a new stream which may be based on the
|
||||
/// provided reactor handle.
|
||||
/// 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.
|
||||
/// * Once a signal handler 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 with_handle(kind: SignalKind, handle: &Handle) -> io::Result<Self> {
|
||||
let signal = kind.0;
|
||||
|
||||
// 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)?;
|
||||
|
||||
// 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, rx })
|
||||
}
|
||||
|
||||
pub(crate) fn ctrl_c(handle: &Handle) -> io::Result<Self> {
|
||||
Self::with_handle(SignalKind::interrupt(), handle)
|
||||
}
|
||||
}
|
||||
|
||||
impl Stream for Signal {
|
||||
type Item = ();
|
||||
|
||||
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_recv(cx)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn signal_enable_error_on_invalid_input() {
|
||||
signal_enable(-1).unwrap_err();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn signal_enable_error_on_forbidden_input() {
|
||||
signal_enable(signal_hook_registry::FORBIDDEN[0]).unwrap_err();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,240 @@
|
||||
//! Windows-specific types for signal handling.
|
||||
//!
|
||||
//! This module is only defined on Windows and contains the primary `Event` type
|
||||
//! for receiving notifications of events. These events are listened for via the
|
||||
//! `SetConsoleCtrlHandler` function which receives events of the type
|
||||
//! `CTRL_C_EVENT` and `CTRL_BREAK_EVENT`
|
||||
|
||||
#![cfg(windows)]
|
||||
|
||||
use super::registry::{globals, EventId, EventInfo, Init, Storage};
|
||||
use crate::driver::Handle;
|
||||
|
||||
use tokio_sync::mpsc::{channel, Receiver};
|
||||
|
||||
use futures_core::stream::Stream;
|
||||
use std::convert::TryFrom;
|
||||
use std::io;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Once;
|
||||
use std::task::{Context, Poll};
|
||||
use winapi::shared::minwindef::*;
|
||||
use winapi::um::consoleapi::SetConsoleCtrlHandler;
|
||||
use winapi::um::wincon::*;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct OsStorage {
|
||||
ctrl_c: EventInfo,
|
||||
ctrl_break: EventInfo,
|
||||
}
|
||||
|
||||
impl Init for OsStorage {
|
||||
fn init() -> Self {
|
||||
Self {
|
||||
ctrl_c: EventInfo::default(),
|
||||
ctrl_break: EventInfo::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Storage for OsStorage {
|
||||
fn event_info(&self, id: EventId) -> Option<&EventInfo> {
|
||||
match DWORD::try_from(id) {
|
||||
Ok(CTRL_C_EVENT) => Some(&self.ctrl_c),
|
||||
Ok(CTRL_BREAK_EVENT) => Some(&self.ctrl_break),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn for_each<'a, F>(&'a self, mut f: F)
|
||||
where
|
||||
F: FnMut(&'a EventInfo),
|
||||
{
|
||||
f(&self.ctrl_c);
|
||||
f(&self.ctrl_break);
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct OsExtraData {}
|
||||
|
||||
impl Init for OsExtraData {
|
||||
fn init() -> Self {
|
||||
Self {}
|
||||
}
|
||||
}
|
||||
|
||||
/// Stream of events discovered via `SetConsoleCtrlHandler`.
|
||||
///
|
||||
/// This structure can be used to listen for events of the type `CTRL_C_EVENT`
|
||||
/// and `CTRL_BREAK_EVENT`. The `Stream` trait is implemented for this struct
|
||||
/// and will resolve for each notification received by the process. Note that
|
||||
/// there are few limitations with this as well:
|
||||
///
|
||||
/// * A notification to this process notifies *all* `Event` streams for that
|
||||
/// event type.
|
||||
/// * Notifications to an `Event` stream **are coalesced** if they aren't
|
||||
/// processed quickly enough. This means that if two notifications are
|
||||
/// received back-to-back, then the stream may only receive one item about the
|
||||
/// two notifications.
|
||||
// FIXME: refactor and combine with unix::Signal
|
||||
#[must_use = "streams do nothing unless polled"]
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct Event {
|
||||
rx: Receiver<()>,
|
||||
}
|
||||
|
||||
impl Event {
|
||||
/// Creates a new stream listening for the `CTRL_C_EVENT` events.
|
||||
///
|
||||
/// This function will register a handler via `SetConsoleCtrlHandler` and
|
||||
/// deliver notifications to the returned stream.
|
||||
pub(crate) fn ctrl_c(handle: &Handle) -> io::Result<Self> {
|
||||
Event::new(CTRL_C_EVENT, handle)
|
||||
}
|
||||
|
||||
/// Creates a new stream listening for the `CTRL_BREAK_EVENT` events.
|
||||
///
|
||||
/// This function will register a handler via `SetConsoleCtrlHandler` and
|
||||
/// deliver notifications to the returned stream.
|
||||
fn ctrl_break_handle(handle: &Handle) -> io::Result<Self> {
|
||||
Event::new(CTRL_BREAK_EVENT, handle)
|
||||
}
|
||||
|
||||
fn new(signum: DWORD, _handle: &Handle) -> io::Result<Self> {
|
||||
global_init()?;
|
||||
|
||||
let (tx, rx) = channel(1);
|
||||
globals().register_listener(signum as EventId, tx);
|
||||
|
||||
Ok(Event { rx })
|
||||
}
|
||||
}
|
||||
|
||||
impl Stream for Event {
|
||||
type Item = ();
|
||||
|
||||
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||
self.rx.poll_recv(cx)
|
||||
}
|
||||
}
|
||||
|
||||
fn global_init() -> io::Result<()> {
|
||||
static INIT: Once = Once::new();
|
||||
|
||||
let mut init = None;
|
||||
INIT.call_once(|| unsafe {
|
||||
let rc = SetConsoleCtrlHandler(Some(handler), TRUE);
|
||||
let ret = if rc == 0 {
|
||||
Err(io::Error::last_os_error())
|
||||
} else {
|
||||
Ok(())
|
||||
};
|
||||
|
||||
init = Some(ret);
|
||||
});
|
||||
|
||||
init.unwrap_or_else(|| Ok(()))
|
||||
}
|
||||
|
||||
unsafe extern "system" fn handler(ty: DWORD) -> BOOL {
|
||||
let globals = globals();
|
||||
globals.record_event(ty as EventId);
|
||||
|
||||
// According to https://docs.microsoft.com/en-us/windows/console/handlerroutine
|
||||
// the handler routine is always invoked in a new thread, thus we don't
|
||||
// have the same restrictions as in Unix signal handlers, meaning we can
|
||||
// go ahead and perform the broadcast here.
|
||||
if globals.broadcast() {
|
||||
TRUE
|
||||
} else {
|
||||
// No one is listening for this notification any more
|
||||
// let the OS fire the next (possibly the default) handler.
|
||||
FALSE
|
||||
}
|
||||
}
|
||||
|
||||
/// Represents a stream which receives "ctrl-break" notifications sent to the process
|
||||
/// via `SetConsoleCtrlHandler`.
|
||||
///
|
||||
/// A notification to this process notifies *all* streams listening to
|
||||
/// this event. Moreover, the notifications **are coalesced** if they aren't processed
|
||||
/// quickly enough. This means that if two notifications are received back-to-back,
|
||||
/// then the stream may only receive one item about the two notifications.
|
||||
#[must_use = "streams do nothing unless polled"]
|
||||
#[derive(Debug)]
|
||||
pub struct CtrlBreak {
|
||||
inner: Event,
|
||||
}
|
||||
|
||||
impl CtrlBreak {
|
||||
/// Creates a new stream which receives "ctrl-break" notifications sent to the
|
||||
/// process.
|
||||
///
|
||||
/// This function binds to the default reactor.
|
||||
pub fn new() -> io::Result<Self> {
|
||||
Self::with_handle(&Handle::default())
|
||||
}
|
||||
|
||||
/// Creates a new stream which receives "ctrl-break" notifications sent to the
|
||||
/// process.
|
||||
///
|
||||
/// This function binds to reactor specified by `handle`.
|
||||
pub fn with_handle(handle: &Handle) -> io::Result<Self> {
|
||||
Event::ctrl_break_handle(handle).map(|inner| Self { inner })
|
||||
}
|
||||
}
|
||||
|
||||
impl Stream for CtrlBreak {
|
||||
type Item = ();
|
||||
|
||||
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||
Pin::new(&mut self.inner)
|
||||
.poll_next(cx)
|
||||
.map(|item| item.map(|_| ()))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
use tokio::timer::Timeout;
|
||||
|
||||
use futures_util::future::FutureExt;
|
||||
use futures_util::stream::StreamExt;
|
||||
use std::future::Future;
|
||||
use std::time::Duration;
|
||||
|
||||
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"))
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ctrl_c() {
|
||||
let ctrl_c = crate::signal::CtrlC::new().expect("failed to create CtrlC");
|
||||
|
||||
// Windows doesn't have a good programmatic way of sending events
|
||||
// like sending signals on Unix, so we'll stub out the actual OS
|
||||
// integration and test that our handling works.
|
||||
unsafe {
|
||||
super::handler(CTRL_C_EVENT);
|
||||
}
|
||||
|
||||
let _ = with_timeout(ctrl_c.into_future()).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ctrl_break() {
|
||||
let ctrl_break = super::CtrlBreak::new().expect("failed to create CtrlC");
|
||||
|
||||
// Windows doesn't have a good programmatic way of sending events
|
||||
// like sending signals on Unix, so we'll stub out the actual OS
|
||||
// integration and test that our handling works.
|
||||
unsafe {
|
||||
super::handler(CTRL_BREAK_EVENT);
|
||||
}
|
||||
|
||||
let _ = with_timeout(ctrl_break.into_future()).await;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user