| //! An iterator over incoming signals. |
| //! |
| //! This provides a higher abstraction over the signals, providing a structure |
| //! ([`Signals`](struct.Signals.html)) able to iterate over the incoming signals. |
| //! |
| //! In case the `tokio-support` feature is turned on, the [`Async`](struct.Async.html) is also |
| //! available, making it possible to integrate with the tokio runtime. |
| //! |
| //! # Examples |
| //! |
| //! ```rust |
| //! extern crate libc; |
| //! extern crate signal_hook; |
| //! |
| //! use std::io::Error; |
| //! |
| //! use signal_hook::iterator::Signals; |
| //! |
| //! fn main() -> Result<(), Error> { |
| //! let signals = Signals::new(&[ |
| //! signal_hook::SIGHUP, |
| //! signal_hook::SIGTERM, |
| //! signal_hook::SIGINT, |
| //! signal_hook::SIGQUIT, |
| //! # signal_hook::SIGUSR1, |
| //! ])?; |
| //! # // A trick to terminate the example when run as doc-test. Not part of the real code. |
| //! # unsafe { libc::raise(signal_hook::SIGUSR1) }; |
| //! 'outer: loop { |
| //! // Pick up signals that arrived since last time |
| //! for signal in signals.pending() { |
| //! match signal as libc::c_int { |
| //! signal_hook::SIGHUP => { |
| //! // Reload configuration |
| //! // Reopen the log file |
| //! } |
| //! signal_hook::SIGTERM | signal_hook::SIGINT | signal_hook::SIGQUIT => { |
| //! break 'outer; |
| //! }, |
| //! # signal_hook::SIGUSR1 => return Ok(()), |
| //! _ => unreachable!(), |
| //! } |
| //! } |
| //! // Do some bit of work ‒ something with upper limit on waiting, so we don't block |
| //! // forever with a SIGTERM already waiting. |
| //! } |
| //! println!("Terminating. Bye bye"); |
| //! Ok(()) |
| //! } |
| //! ``` |
| |
| use std::borrow::Borrow; |
| use std::io::Error; |
| use std::iter::Enumerate; |
| use std::os::unix::io::AsRawFd; |
| use std::os::unix::net::UnixStream; |
| use std::slice::Iter; |
| use std::sync::atomic::{AtomicBool, Ordering}; |
| use std::sync::{Arc, Mutex}; |
| |
| use libc::{self, c_int}; |
| |
| use crate::SigId; |
| |
| /// Maximal signal number we support. |
| const MAX_SIGNUM: usize = 128; |
| |
| #[derive(Debug)] |
| struct Waker { |
| pending: Vec<AtomicBool>, |
| closed: AtomicBool, |
| read: UnixStream, |
| write: UnixStream, |
| } |
| |
| impl Waker { |
| /// Sends a wakeup signal to the internal wakeup pipe. |
| fn wake(&self) { |
| unsafe { |
| // See the comment at pipe::write. |
| // |
| // We don't use pipe::write, because it expects the FD to be already in non-blocking |
| // mode. That's because it needs to support actual pipes. We can afford send here, |
| // which has flags. |
| libc::send( |
| self.write.as_raw_fd(), |
| b"X" as *const _ as *const _, |
| 1, |
| libc::MSG_DONTWAIT, |
| ); |
| } |
| } |
| } |
| |
| #[derive(Debug)] |
| struct RegisteredSignals(Mutex<Vec<Option<SigId>>>); |
| |
| impl Drop for RegisteredSignals { |
| fn drop(&mut self) { |
| let lock = self.0.lock().unwrap(); |
| for id in lock.iter().filter_map(|s| *s) { |
| crate::unregister(id); |
| } |
| } |
| } |
| |
| /// The main structure of the module, representing interest in some signals. |
| /// |
| /// Unlike the helpers in other modules, this registers the signals when created and unregisters |
| /// them on drop. It provides the pending signals during its lifetime, either in batches or as an |
| /// infinite iterator. |
| /// |
| /// # Multiple consumers |
| /// |
| /// You may have noticed this structure can be used simultaneously by multiple threads. If it is |
| /// done, a signal arrives to one of the threads (on the first come, first serve basis). The signal |
| /// is *not* broadcasted to all currently active threads. |
| /// |
| /// A similar thing applies to cloning the structure ‒ at least one of the copies gets the signal, |
| /// but it is not broadcasted to all of them. |
| /// |
| /// If you need multiple recipients, you can create multiple independent instances (not by cloning, |
| /// but by the constructor). |
| /// |
| /// # Examples |
| /// |
| /// ```rust |
| /// # extern crate signal_hook; |
| /// # |
| /// # use std::io::Error; |
| /// # use std::thread; |
| /// use signal_hook::iterator::Signals; |
| /// |
| /// # |
| /// # fn main() -> Result<(), Error> { |
| /// let signals = Signals::new(&[signal_hook::SIGUSR1, signal_hook::SIGUSR2])?; |
| /// thread::spawn(move || { |
| /// for signal in &signals { |
| /// match signal { |
| /// signal_hook::SIGUSR1 => {}, |
| /// signal_hook::SIGUSR2 => {}, |
| /// _ => unreachable!(), |
| /// } |
| /// } |
| /// }); |
| /// # Ok(()) |
| /// # } |
| /// ``` |
| /// |
| /// # `mio` support |
| /// |
| /// If the crate is compiled with the `mio-support` or `mio-0_7-support` flags, the `Signals` |
| /// becomes pluggable into `mio` version `0.6` or `0.7` respectively (it implements the `Source` |
| /// trait). If it becomes readable, there may be new signals to pick up. |
| /// |
| /// # `tokio` support |
| /// |
| /// If the crate is compiled with the `tokio-support` flag, the [`into_async`](#method.into_async) |
| /// method becomes available. This method turns the iterator into an asynchronous stream of |
| /// received signals. |
| #[derive(Clone, Debug)] |
| pub struct Signals { |
| ids: Arc<RegisteredSignals>, |
| waker: Arc<Waker>, |
| } |
| |
| impl Signals { |
| /// Creates the `Signals` structure. |
| /// |
| /// This registers all the signals listed. The same restrictions (panics, errors) apply as with |
| /// [`register`](../fn.register.html). |
| pub fn new<I, S>(signals: I) -> Result<Self, Error> |
| where |
| I: IntoIterator<Item = S>, |
| S: Borrow<c_int>, |
| { |
| let (read, write) = UnixStream::pair()?; |
| let pending = (0..MAX_SIGNUM).map(|_| AtomicBool::new(false)).collect(); |
| let waker = Arc::new(Waker { |
| pending, |
| closed: AtomicBool::new(false), |
| read, |
| write, |
| }); |
| let ids = (0..MAX_SIGNUM).map(|_| None).collect(); |
| let me = Self { |
| ids: Arc::new(RegisteredSignals(Mutex::new(ids))), |
| waker, |
| }; |
| for sig in signals { |
| me.add_signal(*sig.borrow())?; |
| } |
| Ok(me) |
| } |
| |
| /// Registers another signal to the set watched by this [`Signals`] instance. |
| /// |
| /// # Notes |
| /// |
| /// * This is safe to call concurrently from whatever thread. |
| /// * This is *not* safe to call from within a signal handler. |
| /// * If the signal number was already registered previously, this is a no-op. |
| /// * If this errors, the original set of signals is left intact. |
| /// * This actually registers the signal into the whole group of [`Signals`] cloned from each |
| /// other, so any of them might start receiving the signals. |
| /// |
| /// # Panics |
| /// |
| /// * If the given signal is [forbidden][::FORBIDDEN]. |
| /// * If the signal number is negative or larger than internal limit. The limit should be |
| /// larger than any supported signal the OS supports. |
| pub fn add_signal(&self, signal: c_int) -> Result<(), Error> { |
| assert!(signal >= 0); |
| assert!( |
| (signal as usize) < MAX_SIGNUM, |
| "Signal number {} too large. If your OS really supports such signal, file a bug", |
| signal, |
| ); |
| let mut lock = self.ids.0.lock().unwrap(); |
| // Already registered, ignoring |
| if lock[signal as usize].is_some() { |
| return Ok(()); |
| } |
| |
| let waker = Arc::clone(&self.waker); |
| let action = move || { |
| waker.pending[signal as usize].store(true, Ordering::SeqCst); |
| waker.wake(); |
| }; |
| let id = unsafe { crate::register(signal, action) }?; |
| lock[signal as usize] = Some(id); |
| Ok(()) |
| } |
| |
| /// Reads data from the internal self-pipe. |
| /// |
| /// If `wait` is `true` and there are no data in the self pipe, it blocks until some come. |
| /// |
| /// Returns weather it successfully read something. |
| fn flush(&self, wait: bool) -> bool { |
| // Just an optimisation.. would work without it too. |
| if self.waker.closed.load(Ordering::SeqCst) { |
| return false; |
| } |
| const SIZE: usize = 1024; |
| let mut buff = [0u8; SIZE]; |
| let res = unsafe { |
| // We ignore all errors on purpose. This should not be something like closed file |
| // descriptor. It could EAGAIN, but that's OK in case we say MSG_DONTWAIT. If it's |
| // EINTR, then it's OK too, it'll only create a spurious wakeup. |
| libc::recv( |
| self.waker.read.as_raw_fd(), |
| buff.as_mut_ptr() as *mut libc::c_void, |
| SIZE, |
| if wait { 0 } else { libc::MSG_DONTWAIT }, |
| ) |
| }; |
| |
| if res > 0 { |
| unsafe { |
| // Finish draining the data in case there's more |
| while libc::recv( |
| self.waker.read.as_raw_fd(), |
| buff.as_mut_ptr() as *mut libc::c_void, |
| SIZE, |
| libc::MSG_DONTWAIT, |
| ) > 0 |
| {} |
| } |
| } |
| |
| if self.waker.closed.load(Ordering::SeqCst) { |
| // Wake any other sleeping ends |
| // (if none wait, it'll only leave garbage inside the pipe, but we'll close it soon |
| // anyway). |
| self.waker.wake(); |
| } |
| res > 0 |
| } |
| |
| /// Returns an iterator of already received signals. |
| /// |
| /// This returns an iterator over all the signal numbers of the signals received since last |
| /// time they were read (out of the set registered by this `Signals` instance). Note that they |
| /// are returned in arbitrary order and a signal number is returned only once even if it was |
| /// received multiple times. |
| /// |
| /// This method returns immediately (does not block) and may produce an empty iterator if there |
| /// are no signals ready. |
| pub fn pending(&self) -> Pending { |
| self.flush(false); |
| |
| Pending(self.waker.pending.iter().enumerate()) |
| } |
| |
| /// Waits for some signals to be available and returns an iterator. |
| /// |
| /// This is similar to [`pending`](#method.pending). If there are no signals available, it |
| /// tries to wait for some to arrive. However, due to implementation details, this still can |
| /// produce an empty iterator. |
| /// |
| /// This can block for arbitrary long time. |
| /// |
| /// Note that the blocking is done in this method, not in the iterator. |
| pub fn wait(&self) -> Pending { |
| self.flush(true); |
| |
| Pending(self.waker.pending.iter().enumerate()) |
| } |
| |
| /// Returns an infinite iterator over arriving signals. |
| /// |
| /// The iterator's `next()` blocks as necessary to wait for signals to arrive. This is adequate |
| /// if you want to designate a thread solely to handling signals. If multiple signals come at |
| /// the same time (between two values produced by the iterator), they will be returned in |
| /// arbitrary order. Multiple instances of the same signal may be collated. |
| /// |
| /// This is also the iterator returned by `IntoIterator` implementation on `&Signals`. |
| /// |
| /// This iterator terminates only if the [`Signals`] is explicitly [closed][Signals::close]. |
| /// |
| /// # Examples |
| /// |
| /// ```rust |
| /// # extern crate libc; |
| /// # extern crate signal_hook; |
| /// # |
| /// # use std::io::Error; |
| /// # use std::thread; |
| /// # |
| /// use signal_hook::iterator::Signals; |
| /// |
| /// # fn main() -> Result<(), Error> { |
| /// let signals = Signals::new(&[signal_hook::SIGUSR1, signal_hook::SIGUSR2])?; |
| /// thread::spawn(move || { |
| /// for signal in signals.forever() { |
| /// match signal { |
| /// signal_hook::SIGUSR1 => {}, |
| /// signal_hook::SIGUSR2 => {}, |
| /// _ => unreachable!(), |
| /// } |
| /// } |
| /// }); |
| /// # Ok(()) |
| /// # } |
| /// ``` |
| pub fn forever(&self) -> Forever { |
| Forever { |
| signals: self, |
| iter: self.pending(), |
| } |
| } |
| |
| /// Is it closed? |
| /// |
| /// See [`close`][Signals::close]. |
| pub fn is_closed(&self) -> bool { |
| self.waker.closed.load(Ordering::SeqCst) |
| } |
| |
| /// Closes the instance. |
| /// |
| /// This is meant to signalize termination through all the interrelated instances ‒ the ones |
| /// created by cloning the same original [`Signals`] instance (and all the [`Async`] ones |
| /// created from them). After calling close: |
| /// |
| /// * [`is_closed`][Signals::is_closed] will return true. |
| /// * All currently blocking operations on all threads and all the instances are interrupted |
| /// and terminate. |
| /// * Any further operations will never block. |
| /// * Further signals may or may not be returned from the iterators. However, if any are |
| /// returned, these are real signals that happened. |
| /// * The [`forever`][Signals::forever] terminates (follows from the above). |
| /// |
| /// The goal is to be able to shut down any background thread that handles only the signals. |
| /// |
| /// ```rust |
| /// # use signal_hook::iterator::Signals; |
| /// # use signal_hook::SIGUSR1; |
| /// # fn main() -> Result<(), std::io::Error> { |
| /// let signals = Signals::new(&[SIGUSR1])?; |
| /// let signals_bg = signals.clone(); |
| /// let thread = std::thread::spawn(move || { |
| /// for signal in &signals_bg { |
| /// // Whatever with the signal |
| /// # let _ = signal; |
| /// } |
| /// }); |
| /// |
| /// signals.close(); |
| /// |
| /// // The thread will terminate on its own now (the for cycle runs out of signals). |
| /// thread.join().expect("background thread panicked"); |
| /// # Ok(()) } |
| /// ``` |
| pub fn close(&self) { |
| self.waker.closed.store(true, Ordering::SeqCst); |
| self.waker.wake(); |
| } |
| } |
| |
| impl<'a> IntoIterator for &'a Signals { |
| type Item = c_int; |
| type IntoIter = Forever<'a>; |
| fn into_iter(self) -> Forever<'a> { |
| self.forever() |
| } |
| } |
| |
| /// The iterator of one batch of signals. |
| /// |
| /// This is returned by the [`pending`](struct.Signals.html#method.pending) and |
| /// [`wait`](struct.Signals.html#method.wait) methods. |
| pub struct Pending<'a>(Enumerate<Iter<'a, AtomicBool>>); |
| |
| impl<'a> Iterator for Pending<'a> { |
| type Item = c_int; |
| |
| fn next(&mut self) -> Option<c_int> { |
| while let Some((sig, flag)) = self.0.next() { |
| if flag |
| .compare_exchange(true, false, Ordering::SeqCst, Ordering::Relaxed) |
| .is_ok() |
| { |
| return Some(sig as c_int); |
| } |
| } |
| |
| None |
| } |
| } |
| |
| /// The infinite iterator of signals. |
| /// |
| /// It is returned by the [`forever`](struct.Signals.html#method.forever) and by the `IntoIterator` |
| /// implementation of [`&Signals`](struct.Signals.html). |
| pub struct Forever<'a> { |
| signals: &'a Signals, |
| iter: Pending<'a>, |
| } |
| |
| impl<'a> Iterator for Forever<'a> { |
| type Item = c_int; |
| |
| fn next(&mut self) -> Option<c_int> { |
| while !self.signals.is_closed() { |
| if let Some(result) = self.iter.next() { |
| return Some(result); |
| } |
| |
| self.iter = self.signals.wait(); |
| } |
| |
| None |
| } |
| } |
| #[cfg(feature = "mio-support")] |
| mod mio_support { |
| use std::io::Error; |
| use std::os::unix::io::AsRawFd; |
| |
| use mio::event::Evented; |
| use mio::unix::EventedFd; |
| use mio::{Poll, PollOpt, Ready, Token}; |
| |
| use super::Signals; |
| |
| impl Evented for Signals { |
| fn register( |
| &self, |
| poll: &Poll, |
| token: Token, |
| interest: Ready, |
| opts: PollOpt, |
| ) -> Result<(), Error> { |
| EventedFd(&self.waker.read.as_raw_fd()).register(poll, token, interest, opts) |
| } |
| |
| fn reregister( |
| &self, |
| poll: &Poll, |
| token: Token, |
| interest: Ready, |
| opts: PollOpt, |
| ) -> Result<(), Error> { |
| EventedFd(&self.waker.read.as_raw_fd()).reregister(poll, token, interest, opts) |
| } |
| |
| fn deregister(&self, poll: &Poll) -> Result<(), Error> { |
| EventedFd(&self.waker.read.as_raw_fd()).deregister(poll) |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use std::time::Duration; |
| |
| use libc; |
| use mio::Events; |
| |
| use super::*; |
| |
| #[test] |
| fn mio_wakeup() { |
| let signals = Signals::new(&[crate::SIGUSR1]).unwrap(); |
| let token = Token(0); |
| let poll = Poll::new().unwrap(); |
| poll.register(&signals, token, Ready::readable(), PollOpt::level()) |
| .unwrap(); |
| let mut events = Events::with_capacity(10); |
| unsafe { libc::raise(crate::SIGUSR1) }; |
| poll.poll(&mut events, Some(Duration::from_secs(10))) |
| .unwrap(); |
| let event = events.iter().next().unwrap(); |
| assert!(event.readiness().is_readable()); |
| assert_eq!(token, event.token()); |
| let sig = signals.pending().next().unwrap(); |
| assert_eq!(crate::SIGUSR1, sig); |
| } |
| } |
| } |
| |
| #[cfg(any(test, feature = "mio-0_7-support"))] |
| mod mio_0_7_support { |
| use std::io::Error; |
| use std::os::unix::io::AsRawFd; |
| |
| use mio_0_7::event::Source; |
| use mio_0_7::unix::SourceFd; |
| use mio_0_7::{Interest, Registry, Token}; |
| |
| use super::Signals; |
| |
| impl Source for Signals { |
| fn register( |
| &mut self, |
| registry: &Registry, |
| token: Token, |
| interest: Interest, |
| ) -> Result<(), Error> { |
| SourceFd(&self.waker.read.as_raw_fd()).register(registry, token, interest) |
| } |
| |
| fn reregister( |
| &mut self, |
| registry: &Registry, |
| token: Token, |
| interest: Interest, |
| ) -> Result<(), Error> { |
| SourceFd(&self.waker.read.as_raw_fd()).reregister(registry, token, interest) |
| } |
| |
| fn deregister(&mut self, registry: &Registry) -> Result<(), Error> { |
| SourceFd(&self.waker.read.as_raw_fd()).deregister(registry) |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use std::time::Duration; |
| |
| use mio_0_7::{Events, Poll}; |
| |
| use super::*; |
| |
| #[test] |
| fn mio_wakeup() { |
| let mut signals = Signals::new(&[crate::SIGUSR1]).unwrap(); |
| let mut poll = Poll::new().unwrap(); |
| let token = Token(0); |
| poll.registry() |
| .register(&mut signals, token, Interest::READABLE) |
| .unwrap(); |
| |
| let mut events = Events::with_capacity(10); |
| unsafe { libc::raise(crate::SIGUSR1) }; |
| poll.poll(&mut events, Some(Duration::from_secs(10))) |
| .unwrap(); |
| let event = events.iter().next().unwrap(); |
| |
| assert!(event.is_readable()); |
| assert_eq!(token, event.token()); |
| let sig = signals.pending().next().unwrap(); |
| assert_eq!(crate::SIGUSR1, sig); |
| } |
| } |
| } |
| |
| #[cfg(feature = "tokio-support")] |
| mod tokio_support { |
| use std::io::Error; |
| use std::sync::atomic::Ordering; |
| |
| use futures::stream::Stream; |
| use futures::{Async as AsyncResult, Poll}; |
| use libc::{self, c_int}; |
| use tokio_reactor::{Handle, Registration}; |
| |
| use super::Signals; |
| |
| /// An asynchronous stream of registered signals. |
| /// |
| /// It is created by converting [`Signals`](struct.Signals.html). See |
| /// [`Signals::into_async`](struct.Signals.html#method.into_async). |
| /// |
| /// # Cloning |
| /// |
| /// If you register multiple signals, then create multiple `Signals` instances by cloning and |
| /// convert them to `Async`, one of them can „steal“ wakeups for several signals at once. This |
| /// one will produce the signals while the others will be silent. |
| /// |
| /// This has an effect if the one consumes them slowly or is dropped after the first one. |
| /// |
| /// It is recommended not to clone the `Signals` instances and keep just one `Async` stream |
| /// around. |
| #[derive(Debug)] |
| pub struct Async { |
| registration: Registration, |
| inner: Signals, |
| // It seems we can't easily use the iterator into the array here because of lifetimes ‒ |
| // using non-'static things in around futures is real pain. |
| position: usize, |
| } |
| |
| impl Async { |
| /// Creates a new `Async`. |
| pub fn new(signals: Signals, handle: &Handle) -> Result<Self, Error> { |
| let registration = Registration::new(); |
| registration.register_with(&signals, handle)?; |
| Ok(Async { |
| registration, |
| inner: signals, |
| position: 0, |
| }) |
| } |
| } |
| |
| impl Stream for Async { |
| type Item = libc::c_int; |
| type Error = Error; |
| fn poll(&mut self) -> Poll<Option<libc::c_int>, Self::Error> { |
| while !self.inner.is_closed() { |
| if self.position >= self.inner.waker.pending.len() { |
| if self.registration.poll_read_ready()?.is_not_ready() { |
| return Ok(AsyncResult::NotReady); |
| } |
| // Non-blocking clean of the pipe |
| while self.inner.flush(false) {} |
| // By now we have an indication there might be some stuff inside the signals, |
| // reset the scanning position |
| self.position = 0; |
| } |
| assert!(self.position < self.inner.waker.pending.len()); |
| let sig = &self.inner.waker.pending[self.position]; |
| let sig_num = self.position; |
| self.position += 1; |
| if sig |
| .compare_exchange(true, false, Ordering::SeqCst, Ordering::Relaxed) |
| .is_ok() |
| { |
| // Successfully claimed a signal, return it |
| return Ok(AsyncResult::Ready(Some(sig_num as c_int))); |
| } |
| } |
| Ok(AsyncResult::Ready(None)) |
| } |
| } |
| |
| impl Signals { |
| /// Turns the iterator into an asynchronous stream. |
| /// |
| /// This allows getting the signals in asynchronous way in a tokio event loop. Available |
| /// only if compiled with the `tokio-support` feature enabled. |
| /// |
| /// # Examples |
| /// |
| /// ```rust |
| /// extern crate libc; |
| /// extern crate signal_hook; |
| /// extern crate tokio; |
| /// |
| /// use std::io::Error; |
| /// |
| /// use signal_hook::iterator::Signals; |
| /// use tokio::prelude::*; |
| /// |
| /// fn main() -> Result<(), Error> { |
| /// let wait_signal = Signals::new(&[signal_hook::SIGUSR1])? |
| /// .into_async()? |
| /// .into_future() |
| /// .map(|sig| assert_eq!(sig.0.unwrap(), signal_hook::SIGUSR1)) |
| /// .map_err(|e| panic!("{}", e.0)); |
| /// unsafe { libc::raise(signal_hook::SIGUSR1) }; |
| /// tokio::run(wait_signal); |
| /// Ok(()) |
| /// } |
| /// ``` |
| pub fn into_async(self) -> Result<Async, Error> { |
| Async::new(self, &Handle::default()) |
| } |
| |
| /// Turns the iterator into a stream, tied into a specific tokio reactor. |
| pub fn into_async_with_handle(self, handle: &Handle) -> Result<Async, Error> { |
| Async::new(self, handle) |
| } |
| } |
| } |
| |
| #[cfg(feature = "tokio-support")] |
| pub use self::tokio_support::Async; |
