diff --git a/tokio-signal/Cargo.toml b/tokio-signal/Cargo.toml index ceffa126a..62c250bc1 100644 --- a/tokio-signal/Cargo.toml +++ b/tokio-signal/Cargo.toml @@ -27,18 +27,20 @@ appveyor = { repository = "carllerche/tokio", id = "s83yxhy9qeb58va7" } [dependencies] futures = "0.1.11" -mio = "0.6.14" +lazy_static = "1" tokio-reactor = { version = "0.2.0", path = "../tokio-reactor" } tokio-executor = { version = "0.2.0", path = "../tokio-executor" } tokio-io = { version = "0.2.0", path = "../tokio-io" } [target.'cfg(unix)'.dependencies] libc = "0.2" +mio = "0.6.14" mio-uds = "0.6" signal-hook-registry = "~1" [dev-dependencies] tokio = { version = "0.2.0", path = "../tokio" } +tokio-timer = { version = "0.3.0", path = "../tokio-timer" } [target.'cfg(windows)'.dependencies.winapi] version = "0.3" diff --git a/tokio-signal/src/lib.rs b/tokio-signal/src/lib.rs index 2e73dd2a8..dea33afc1 100644 --- a/tokio-signal/src/lib.rs +++ b/tokio-signal/src/lib.rs @@ -54,11 +54,23 @@ //! # } //! ``` +#[macro_use] +extern crate lazy_static; + use futures::stream::Stream; use futures::{future, Future}; use std::io; use tokio_reactor::Handle; +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; diff --git a/tokio-signal/src/registry.rs b/tokio-signal/src/registry.rs new file mode 100644 index 000000000..cf6705a92 --- /dev/null +++ b/tokio-signal/src/registry.rs @@ -0,0 +1,292 @@ +use std::ops; +use std::pin::Pin; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Mutex; + +use crate::os::{OsExtraData, OsStorage}; +use futures::sync::mpsc::Sender; + +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>>, +} + +/// 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 { + 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 { + storage: S, +} + +impl Registry { + fn new(storage: S) -> Self { + Self { storage } + } +} + +impl Registry { + /// 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) { + self.storage + .event_info(event_id) + .map(|event_info| event_info.pending.store(true, Ordering::SeqCst)); + } + + /// Broadcast all previously recorded events to their respective listeners. + fn broadcast(&self) { + 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(()) => {} + Err(ref e) if e.is_disconnected() => { + 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()); + } + } + } + }); + } +} + +pub(crate) struct Globals { + extra: OsExtraData, + registry: Registry, +} + +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. + pub(crate) fn broadcast(&self) { + 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> = Pin::new(Box::new(Globals { + extra: OsExtraData::init(), + registry: Registry::new(OsStorage::init()), + })); + } + + GLOBALS.as_ref() +} + +#[cfg(test)] +mod tests { + use super::*; + use futures::sync::mpsc::channel; + use futures::sync::oneshot; + use futures::{Future, Stream}; + + use std::time::Duration; + use tokio::runtime::Runtime; + use tokio_timer::sleep; + + #[test] + fn smoke() { + let registry = Registry::new(vec![ + EventInfo::default(), + EventInfo::default(), + EventInfo::default(), + ]); + + let (first_tx, first_rx) = channel(0); + let (second_tx, second_rx) = channel(0); + let (third_tx, third_rx) = channel(0); + + registry.register_listener(0, first_tx); + registry.register_listener(1, second_tx); + registry.register_listener(2, third_tx); + + let (fire, wait) = oneshot::channel(); + let rt = Runtime::new().unwrap(); + + rt.spawn( + wait.and_then(move |_| { + // 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(); + + sleep(Duration::from_millis(100)) + .map_err(|e| panic!("{:#?}", e)) + .and_then(move |_| { + registry.record_event(0); + registry.broadcast(); + + drop(registry); + Ok(()) + }) + }) + .map_err(|e| panic!("{}", e)), + ); + + let (first_results, second_results, third_results) = rt + .block_on(futures::lazy(move || { + let _ = fire.send(()); + + first_rx + .collect() + .join3(second_rx.collect(), third_rx.collect()) + })) + .expect("failed to extract events"); + + 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(0); + registry.register_listener(1, tx); + } + + #[test] + fn record_invalid_event_does_nothing() { + let registry = Registry::new(vec![EventInfo::default()]); + registry.record_event(42); + } + + #[test] + fn broadcast_cleans_up_disconnected_listeners() { + let registry = Registry::new(vec![EventInfo::default()]); + + let (first_tx, first_rx) = channel(0); + let (second_tx, second_rx) = channel(0); + let (third_tx, third_rx) = channel(0); + + 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(); + let rt = Runtime::new().unwrap(); + + rt.spawn( + wait.and_then(move |_| { + // Record some events which should get coalesced + registry.record_event(0); + registry.broadcast(); + + assert_eq!(1, registry.storage[0].recipients.lock().unwrap().len()); + drop(registry); + + Ok(()) + }) + .map_err(|e| panic!("{}", e)), + ); + + let results = rt + .block_on(futures::lazy(move || { + let _ = fire.send(()); + third_rx.collect() + })) + .expect("failed to extract events"); + + assert_eq!(1, results.len()); + } +} diff --git a/tokio-signal/src/unix.rs b/tokio-signal/src/unix.rs index 3982ebec2..4dd06aeb2 100644 --- a/tokio-signal/src/unix.rs +++ b/tokio-signal/src/unix.rs @@ -9,11 +9,12 @@ pub use libc; use std::io::prelude::*; use std::io::{self, Error, ErrorKind}; +use std::pin::Pin; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Mutex, Once, ONCE_INIT}; +use std::sync::{Once, ONCE_INIT}; use futures::future; -use futures::sync::mpsc::{channel, Receiver, Sender}; +use futures::sync::mpsc::{channel, Receiver}; use futures::{Async, Future}; use futures::{Poll, Stream}; use libc::c_int; @@ -21,6 +22,8 @@ use mio_uds::UnixStream; use tokio_io::IoFuture; use tokio_reactor::{Handle, PollEvented}; +use crate::registry::{globals, EventId, EventInfo, Globals, Init, Storage}; + pub use libc::{SIGALRM, SIGHUP, SIGPIPE, SIGQUIT, SIGTRAP}; pub use libc::{SIGINT, SIGTERM, SIGUSR1, SIGUSR2}; @@ -43,91 +46,61 @@ pub mod bsd { pub use super::libc::SIGINFO; } +pub(crate) type OsStorage = Vec; + // Number of different unix signals // (FreeBSD has 33) const SIGNUM: usize = 33; -type SignalSender = Sender; - -struct SignalInfo { - pending: AtomicBool, - // The ones interested in this signal - recipients: Mutex>>, - - init: Once, - initialized: AtomicBool, +impl Init for OsStorage { + fn init() -> Self { + (0..SIGNUM).map(|_| SignalInfo::default()).collect() + } } -struct Globals { +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, - signals: Vec, } -impl Globals { - /// Register a new `Signal` instance's channel sender. - /// Returns a `SignalId` which should be later used for deregistering - /// this sender. - fn register_signal_sender(signal: c_int, tx: SignalSender) -> SignalId { - let tx = Box::new(tx); - let id = SignalId::from(&tx); +impl Init for OsExtraData { + fn init() -> Self { + let (receiver, sender) = UnixStream::pair().expect("failed to create UnixStream"); - let idx = signal as usize; - globals().signals[idx].recipients.lock().unwrap().push(tx); - id - } - - /// Deregister a `Signal` instance's channel sender because the `Signal` - /// is no longer interested in receiving events (e.g. dropped). - fn deregister_signal_receiver(signal: c_int, id: SignalId) { - let idx = signal as usize; - let mut list = globals().signals[idx].recipients.lock().unwrap(); - list.retain(|sender| SignalId::from(sender) != id); + Self { sender, receiver } } } -/// A newtype which represents a unique identifier for each `Signal` instance. -/// The id is derived by boxing the channel `Sender` associated with this instance -/// and using its address in memory. -#[derive(Debug, Copy, Clone, PartialEq, Eq)] -struct SignalId(usize); - -impl<'a> From<&'a Box> for SignalId { - fn from(tx: &'a Box) -> Self { - SignalId(&**tx as *const _ as usize) - } +pub(crate) struct SignalInfo { + event_info: EventInfo, + init: Once, + initialized: AtomicBool, } impl Default for SignalInfo { fn default() -> SignalInfo { SignalInfo { - pending: AtomicBool::new(false), + event_info: Default::default(), init: ONCE_INIT, initialized: AtomicBool::new(false), - recipients: Mutex::new(Vec::new()), } } } -static mut GLOBALS: *mut Globals = 0 as *mut Globals; - -fn globals() -> &'static Globals { - static INIT: Once = ONCE_INIT; - - unsafe { - INIT.call_once(|| { - let (receiver, sender) = UnixStream::pair().unwrap(); - let globals = Globals { - sender: sender, - receiver: receiver, - signals: (0..SIGNUM).map(|_| Default::default()).collect(), - }; - GLOBALS = Box::into_raw(Box::new(globals)); - }); - &*GLOBALS - } -} - /// Our global signal handler for all signals registered by this module. /// /// The purpose of this signal handler is to primarily: @@ -136,11 +109,12 @@ fn globals() -> &'static Globals { /// 2. Wake up driver tasks by writing a byte to a pipe /// /// Those two operations shoudl both be async-signal safe. -fn action(slot: &SignalInfo, mut sender: &UnixStream) { - slot.pending.store(true, Ordering::SeqCst); +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])); } @@ -150,7 +124,7 @@ fn action(slot: &SignalInfo, mut sender: &UnixStream) { /// 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_hook_registry::FORBIDDEN.contains(&signal) { + if signal < 0 || signal_hook_registry::FORBIDDEN.contains(&signal) { return Err(Error::new( ErrorKind::Other, format!("Refusing to register signal {}", signal), @@ -158,15 +132,14 @@ fn signal_enable(signal: c_int) -> io::Result<()> { } let globals = globals(); - let siginfo = match globals.signals.get(signal as usize) { + 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(siginfo, &globals.sender)) - .map(|_| ()) + signal_hook_registry::register(signal, move || action(globals, signal)).map(|_| ()) }; if registered.is_ok() { siginfo.initialized.store(true, Ordering::Relaxed); @@ -197,7 +170,7 @@ impl Future for Driver { // Drain the data from the pipe and maintain interest in getting more self.drain(); // Broadcast any signals which were received - self.broadcast(); + globals().broadcast(); // This task just lives until the end of the event loop Ok(Async::NotReady) @@ -241,44 +214,6 @@ impl Driver { } } } - - /// Go through all the signals and broadcast everything. - /// - /// Driver tasks wake up for *any* signal and simply process all globally - /// registered signal streams, so each task is sort of cooperatively working - /// for all the rest as well. - fn broadcast(&self) { - for (sig, slot) in globals().signals.iter().enumerate() { - // Any signal of this kind arrived since we checked last? - if !slot.pending.swap(false, Ordering::SeqCst) { - continue; - } - - let signum = sig as c_int; - let mut recipients = slot.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(signum) { - Ok(()) => {} - Err(ref e) if e.is_disconnected() => { - 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()); - } - } - } - } - } } /// An implementation of `Stream` for receiving a particular type of signal. @@ -318,8 +253,7 @@ impl Driver { pub struct Signal { driver: Driver, signal: c_int, - id: SignalId, - rx: Receiver, + rx: Receiver<()>, } impl Signal { @@ -385,11 +319,11 @@ impl Signal { // more. NB: channels always guarantee at least one slot per sender, // so we don't need additional slots let (tx, rx) = channel(0); - let id = Globals::register_signal_sender(signal, tx); + globals().register_listener(signal as EventId, tx); + Ok(Signal { driver: driver, rx: rx, - id: id, signal: signal, }) })(); @@ -404,53 +338,26 @@ impl Stream for Signal { fn poll(&mut self) -> Poll, io::Error> { self.driver.poll().unwrap(); - // receivers don't generate errors - self.rx.poll().map_err(|_| panic!()) - } -} -impl Drop for Signal { - fn drop(&mut self) { - Globals::deregister_signal_receiver(self.signal, self.id); + self.rx + .poll() + .map(|ready| ready.map(|item| item.map(|()| self.signal))) + // receivers don't generate errors + .map_err(|_| unreachable!()) } } #[cfg(test)] mod tests { - use tokio; - use super::*; #[test] - fn dropped_signal_senders_are_cleaned_up() { - let mut rt = - self::tokio::runtime::current_thread::Runtime::new().expect("failed to init runtime"); + fn signal_enable_error_on_invalid_input() { + signal_enable(-1).unwrap_err(); + } - let signum = libc::SIGUSR1; - let signal = rt - .block_on(Signal::new(signum)) - .expect("failed to create signal"); - - { - let recipients = globals().signals[signum as usize] - .recipients - .lock() - .unwrap(); - assert!(!recipients.is_empty()); - } - - drop(signal); - - unsafe { - assert_eq!(libc::kill(libc::getpid(), signum), 0); - } - - { - let recipients = globals().signals[signum as usize] - .recipients - .lock() - .unwrap(); - assert!(recipients.is_empty()); - } + #[test] + fn signal_enable_error_on_forbidden_input() { + signal_enable(signal_hook_registry::FORBIDDEN[0]).unwrap_err(); } } diff --git a/tokio-signal/src/windows.rs b/tokio-signal/src/windows.rs index f03bbb481..f42854a02 100644 --- a/tokio-signal/src/windows.rs +++ b/tokio-signal/src/windows.rs @@ -7,26 +7,70 @@ #![cfg(windows)] -use std::cell::RefCell; +use std::convert::TryFrom; use std::io; -use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Once, ONCE_INIT}; use futures::future; -use futures::stream::Fuse; -use futures::sync::mpsc; -use futures::sync::oneshot; +use futures::sync::mpsc::{channel, Receiver, Sender}; use futures::{Async, Future, Poll, Stream}; -use mio::Ready; -use tokio_reactor::{Handle, PollEvented}; +use tokio_reactor::Handle; use winapi::shared::minwindef::*; use winapi::um::consoleapi::SetConsoleCtrlHandler; use winapi::um::wincon::*; +use crate::registry::{globals, EventId, EventInfo, Init, Storage}; use crate::IoFuture; +#[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 { + driver_waker: Sender<()>, +} + +impl Init for OsExtraData { + fn init() -> Self { + let (driver_waker, driver_rx) = channel(0); + + ::tokio_executor::spawn(DriverTask { rx: driver_rx }); + + Self { driver_waker } + } +} + static INIT: Once = ONCE_INIT; -static mut GLOBAL_STATE: *mut GlobalState = 0 as *mut _; /// Stream of events discovered via `SetConsoleCtrlHandler`. /// @@ -41,36 +85,14 @@ static mut GLOBAL_STATE: *mut GlobalState = 0 as *mut _; /// 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 pub struct Event { - reg: PollEvented, - _finished: oneshot::Sender<()>, -} - -struct GlobalState { - ready: mio::SetReadiness, - tx: mpsc::UnboundedSender, - ctrl_c: GlobalEventState, - ctrl_break: GlobalEventState, -} - -struct GlobalEventState { - ready: AtomicBool, -} - -enum Message { - NewEvent(DWORD, oneshot::Sender>), + rx: Receiver<()>, } +#[derive(Debug)] struct DriverTask { - handle: Handle, - reg: PollEvented, - rx: Fuse>, - ctrl_c: EventState, - ctrl_break: EventState, -} - -struct EventState { - tasks: Vec<(RefCell>, mio::SetReadiness)>, + rx: Receiver<()>, } impl Event { @@ -106,29 +128,24 @@ impl Event { Event::new(CTRL_BREAK_EVENT, handle) } - fn new(signum: DWORD, handle: &Handle) -> IoFuture { - let handle = handle.clone(); + fn new(signum: DWORD, _handle: &Handle) -> IoFuture { let new_signal = future::poll_fn(move || { let mut init = None; INIT.call_once(|| { - init = Some(global_init(&handle)); + init = Some(global_init()); }); if let Some(Err(e)) = init { return Err(e); } - let (tx, rx) = oneshot::channel(); - let msg = Message::NewEvent(signum, tx); - let res = unsafe { (*GLOBAL_STATE).tx.clone().unbounded_send(msg) }; - res.expect( - "failed to request a new signal stream, did the \ - first event loop go away?", - ); - Ok(Async::Ready(rx.then(|r| r.unwrap()))) + let (tx, rx) = channel(0); + globals().register_listener(signum as EventId, tx); + + Ok(Async::Ready(Event { rx })) }); - Box::new(new_signal.flatten()) + Box::new(new_signal) } } @@ -137,54 +154,20 @@ impl Stream for Event { type Error = io::Error; fn poll(&mut self) -> Poll, io::Error> { - if !self.reg.poll_read_ready(Ready::readable())?.is_ready() { - return Ok(Async::NotReady); - } - self.reg.clear_read_ready(Ready::readable())?; - self.reg - .get_ref() - .readiness - .set_readiness(mio::Ready::empty()) - .expect("failed to set readiness"); - Ok(Async::Ready(Some(()))) + self.rx + .poll() + // receivers don't generate errors + .map_err(|_| unreachable!()) } } -fn global_init(handle: &Handle) -> io::Result<()> { - let reg = MyRegistration::new(); - let ready = reg.readiness.clone(); - - let (tx, rx) = mpsc::unbounded(); - let reg = PollEvented::new_with_handle(reg, handle)?; - +fn global_init() -> io::Result<()> { unsafe { - let state = Box::new(GlobalState { - ready: ready, - ctrl_c: GlobalEventState { - ready: AtomicBool::new(false), - }, - ctrl_break: GlobalEventState { - ready: AtomicBool::new(false), - }, - tx: tx, - }); - GLOBAL_STATE = Box::into_raw(state); - let rc = SetConsoleCtrlHandler(Some(handler), TRUE); if rc == 0 { - Box::from_raw(GLOBAL_STATE); - GLOBAL_STATE = 0 as *mut _; return Err(io::Error::last_os_error()); } - ::tokio_executor::spawn(Box::new(DriverTask { - handle: handle.clone(), - rx: rx.fuse(), - reg: reg, - ctrl_c: EventState { tasks: Vec::new() }, - ctrl_break: EventState { tasks: Vec::new() }, - })); - Ok(()) } } @@ -194,153 +177,37 @@ impl Future for DriverTask { type Error = (); fn poll(&mut self) -> Poll<(), ()> { - self.check_event_drops(); - self.check_messages(); - self.check_events().unwrap(); + loop { + // Ensure we keep polling our waker until we know there are no more + // events (and therefore we've registered interest to be woken again). + match self.rx.poll() { + Ok(Async::Ready(Some(()))) => continue, + Ok(Async::Ready(None)) => panic!("driver got disconnected?"), + Ok(Async::NotReady) => break, + // receivers don't generate errors + Err(()) => unreachable!(), + } + } - // TODO: when to finish this task? + globals().broadcast(); + + // TODO(1000): when to finish this task? Ok(Async::NotReady) } } -impl DriverTask { - fn check_event_drops(&mut self) { - self.ctrl_c - .tasks - .retain(|task| !task.0.borrow_mut().poll().is_err()); - self.ctrl_break - .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().unwrap() { - Async::Ready(Some(e)) => e, - Async::Ready(None) | Async::NotReady => break, - }; - let (sig, complete) = match message { - Message::NewEvent(sig, complete) => (sig, complete), - }; - - let event = if sig == CTRL_C_EVENT { - &mut self.ctrl_c - } else { - &mut self.ctrl_break - }; - - // Acquire the (registration, set_readiness) pair by... assuming - // we're on the event loop (true because of the spawn above). - let reg = MyRegistration::new(); - let ready = reg.readiness.clone(); - - let reg = match PollEvented::new_with_handle(reg, &self.handle) { - Ok(reg) => reg, - Err(e) => { - drop(complete.send(Err(e))); - continue; - } - }; - - // Create the `Event` to pass back and then also keep a handle to - // the `SetReadiness` for ourselves internally. - let (tx, rx) = oneshot::channel(); - drop(complete.send(Ok(Event { - reg: reg, - _finished: tx, - }))); - event.tasks.push((RefCell::new(rx), ready)); - } - } - - fn check_events(&mut self) -> io::Result<()> { - if self.reg.poll_read_ready(Ready::readable())?.is_not_ready() { - return Ok(()); - } - self.reg.clear_read_ready(Ready::readable())?; - self.reg - .get_ref() - .readiness - .set_readiness(mio::Ready::empty()) - .expect("failed to set readiness"); - - if unsafe { (*GLOBAL_STATE).ctrl_c.ready.swap(false, Ordering::SeqCst) } { - for task in self.ctrl_c.tasks.iter() { - task.1.set_readiness(mio::Ready::readable()).unwrap(); - } - } - if unsafe { - (*GLOBAL_STATE) - .ctrl_break - .ready - .swap(false, Ordering::SeqCst) - } { - for task in self.ctrl_break.tasks.iter() { - task.1.set_readiness(mio::Ready::readable()).unwrap(); - } - } - Ok(()) - } -} - unsafe extern "system" fn handler(ty: DWORD) -> BOOL { - let event = match ty { - CTRL_C_EVENT => &(*GLOBAL_STATE).ctrl_c, - CTRL_BREAK_EVENT => &(*GLOBAL_STATE).ctrl_break, - _ => return FALSE, - }; - if event.ready.swap(true, Ordering::SeqCst) { - FALSE - } else { - drop((*GLOBAL_STATE).ready.set_readiness(mio::Ready::readable())); - // TODO(1000): this will report that we handled a CTRL_BREAK_EVENT when - // in fact we may not have any streams actually created for that - // event. - TRUE - } -} + let globals = globals(); + globals.record_event(ty as EventId); -struct MyRegistration { - registration: mio::Registration, - readiness: mio::SetReadiness, -} + // FIXME: revisit this, we'd probably want to panic if the driver task goes away, + // but that would unwind across the FFI boundary... + let _ = globals.driver_waker.clone().try_send(()); -impl MyRegistration { - fn new() -> Self { - let (registration, readiness) = mio::Registration::new2(); - - Self { - registration, - readiness, - } - } -} - -impl mio::Evented for MyRegistration { - fn register( - &self, - poll: &mio::Poll, - token: mio::Token, - events: mio::Ready, - opts: mio::PollOpt, - ) -> io::Result<()> { - self.registration.register(poll, token, events, opts) - } - - fn reregister( - &self, - poll: &mio::Poll, - token: mio::Token, - events: mio::Ready, - opts: mio::PollOpt, - ) -> io::Result<()> { - self.registration.reregister(poll, token, events, opts) - } - - fn deregister(&self, poll: &mio::Poll) -> io::Result<()> { - mio::Evented::deregister(&self.registration, poll) - } + // TODO(1000): this will report that we handled a CTRL_BREAK_EVENT when + // in fact we may not have any streams actually created for that + // event. + TRUE } #[cfg(test)]