diff --git a/src/unix.rs b/src/unix.rs index f41fe38ff..bb577762a 100644 --- a/src/unix.rs +++ b/src/unix.rs @@ -16,11 +16,12 @@ 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 futures::{self, Future, IntoFuture, 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}; +use tokio_core::reactor::{PollEvented, Handle}; +use tokio_core::channel::{channel, Sender, Receiver}; static INIT: Once = ONCE_INIT; static mut GLOBAL_STATE: *mut GlobalState = 0 as *mut _; @@ -64,7 +65,7 @@ static mut GLOBAL_STATE: *mut GlobalState = 0 as *mut _; /// alleviate some of these limitations if possible! pub struct Signal { signum: c_int, - reg: ReadinessStream, + reg: PollEvented, _finished: Complete<()>, } @@ -84,7 +85,7 @@ enum Message { } struct DriverTask { - handle: LoopHandle, + handle: Handle, read: UnixStream, rx: Fuse>, signals: [SignalState; 32], @@ -117,7 +118,7 @@ impl Signal { /// 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 { + pub fn new(signum: c_int, handle: &Handle) -> IoFuture { let mut init = None; INIT.call_once(|| { init = Some(global_init(handle)); @@ -133,7 +134,7 @@ impl Signal { rx.then(|r| r.unwrap()) }); match init { - Some(init) => init.and_then(|()| new_signal).boxed(), + Some(init) => init.into_future().and_then(|()| new_signal).boxed(), None => new_signal.boxed(), } } @@ -144,7 +145,9 @@ impl Stream for Signal { type Error = io::Error; fn poll(&mut self) -> Poll, io::Error> { - try_ready!(self.reg.poll_read()); + if !self.reg.poll_read().is_ready() { + return Ok(Async::NotReady) + } self.reg.get_ref() .inner.borrow() .as_ref().unwrap().1 @@ -154,52 +157,49 @@ impl Stream for Signal { } } -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() }, - } +fn global_init(handle: &Handle) -> io::Result<()> { + let (tx, rx) = try!(channel(handle)); + let (read, write) = try!(UnixStream::pair(handle)); + 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() + [ + 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.spawn(DriverTask { + handle: handle.clone(), + 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(), + ] + }, + }); + + Ok(()) + } } impl Future for DriverTask { @@ -267,10 +267,8 @@ impl DriverTask { // 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"), + let reg = match PollEvented::new(reg, &self.handle) { + Ok(reg) => reg, Err(e) => { complete.complete(Err(e)); continue diff --git a/tests/signal.rs b/tests/signal.rs index ef7ccb771..529e740c2 100644 --- a/tests/signal.rs +++ b/tests/signal.rs @@ -12,7 +12,7 @@ use std::time::Duration; use futures::Future; use futures::stream::Stream; -use tokio_core::Loop; +use tokio_core::reactor::{Core, Timeout}; use tokio_signal::unix::Signal; static INIT: Once = ONCE_INIT; @@ -24,7 +24,7 @@ fn lock() -> MutexGuard<'static, ()> { LOCK = Box::into_raw(Box::new(Mutex::new(()))); let (tx, rx) = channel(); thread::spawn(move || { - let mut lp = Loop::new().unwrap(); + let mut lp = Core::new().unwrap(); let handle = lp.handle(); let _signal = lp.run(Signal::new(libc::SIGALRM, &handle)).unwrap(); tx.send(()).unwrap(); @@ -40,7 +40,7 @@ fn lock() -> MutexGuard<'static, ()> { fn simple() { let _lock = lock(); - let mut lp = Loop::new().unwrap(); + let mut lp = Core::new().unwrap(); let handle = lp.handle(); let signal = lp.run(Signal::new(libc::SIGUSR1, &handle)).unwrap(); unsafe { @@ -53,7 +53,7 @@ fn simple() { fn notify_both() { let _lock = lock(); - let mut lp = Loop::new().unwrap(); + let mut lp = Core::new().unwrap(); let handle = lp.handle(); let signal1 = lp.run(Signal::new(libc::SIGUSR2, &handle)).unwrap(); let signal2 = lp.run(Signal::new(libc::SIGUSR2, &handle)).unwrap(); @@ -67,22 +67,22 @@ fn notify_both() { fn drop_then_get_a_signal() { let _lock = lock(); - let mut lp = Loop::new().unwrap(); + let mut lp = Core::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(); + let timeout = Timeout::new(Duration::from_millis(1), &lp.handle()).unwrap(); + lp.run(timeout).unwrap(); } #[test] fn twice() { let _lock = lock(); - let mut lp = Loop::new().unwrap(); + let mut lp = Core::new().unwrap(); let handle = lp.handle(); let signal = lp.run(Signal::new(libc::SIGUSR1, &handle)).unwrap(); unsafe {