From 32eb457dadbbaef2230b53276550a1843ef26f2a Mon Sep 17 00:00:00 2001 From: Alex Crichton Date: Wed, 17 Aug 2016 23:34:44 -0700 Subject: [PATCH] Relax the 'static bound on Loop::run We know that the future will never persist beyond this stack frame, so we can just leave its ownership on the stack frame itself and receive notifications off the event loop that we need to poll it. --- src/event_loop.rs | 93 ++++++++++++++++++++++++++--------------------- tests/tcp.rs | 4 ++ 2 files changed, 55 insertions(+), 42 deletions(-) diff --git a/src/event_loop.rs b/src/event_loop.rs index aef92b62b..70f2cfd92 100644 --- a/src/event_loop.rs +++ b/src/event_loop.rs @@ -1,15 +1,14 @@ -use std::cell::{Cell, RefCell}; +use std::cell::RefCell; use std::io::{self, ErrorKind}; use std::marker; use std::mem; use std::rc::Rc; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT, Ordering}; -use std::sync::mpsc; use std::time::{Instant, Duration}; use futures::{Future, Poll}; -use futures::task::{self, TaskHandle}; +use futures::task::{self, Task, Notify, TaskHandle}; use futures::executor::{ExecuteCallback, Executor}; use mio; use slab::Slab; @@ -33,11 +32,13 @@ const SLAB_CAPACITY: usize = 1024 * 64; // TODO: expand this pub struct Loop { id: usize, - active: Cell, io: mio::Poll, + events: mio::Events, tx: Arc, rx: Receiver, dispatch: RefCell>, + _future_registration: mio::Registration, + future_readiness: Arc, // Timer wheel keeping track of all timeouts. The `usize` stored in the // timer wheel is an index into the slab below. @@ -104,7 +105,6 @@ enum Message { CancelTimeout(TimeoutToken), Run(Box), Drop(DropBox), - Shutdown, } /// Type of I/O objects inserted into the event loop, created by `Source::new`. @@ -140,13 +140,20 @@ impl Loop { mio::Token(0), mio::EventSet::readable(), mio::PollOpt::edge())); + let pair = mio::Registration::new(&io, + mio::Token(1), + mio::EventSet::readable(), + mio::PollOpt::level()); + let (registration, readiness) = pair; Ok(Loop { id: NEXT_LOOP_ID.fetch_add(1, Ordering::Relaxed), - active: Cell::new(true), io: io, + events: mio::Events::new(), tx: Arc::new(MioSender { inner: tx }), rx: rx, - dispatch: RefCell::new(Slab::new_starting_at(1, SLAB_CAPACITY)), + _future_registration: registration, + future_readiness: Arc::new(readiness), + dispatch: RefCell::new(Slab::new_starting_at(2, SLAB_CAPACITY)), timeouts: RefCell::new(Slab::new_starting_at(0, SLAB_CAPACITY)), timer_wheel: RefCell::new(TimerWheel::new()), _marker: marker::PhantomData, @@ -195,25 +202,38 @@ impl Loop { /// otherwise waiting for the future to complete. /// /// Returns the value that the future resolves to. - pub fn run(&mut self, f: F) -> Result - where F: Future + 'static, + pub fn run(&mut self, mut f: F) -> Result + where F: Future, { - let (tx_res, rx_res) = mpsc::channel(); - let handle = self.handle(); - self.add_loop_data(f.then(move |res| { - handle.shutdown(); - tx_res.send(res) - })).forget(); + struct MyNotify(Arc); - self._run(); + impl Notify for MyNotify { + fn notify(&self) { + self.0.set_readiness(mio::EventSet::readable()) + .expect("failed to set readiness"); + } + } - rx_res.recv().unwrap() + let mut task = Task::new_notify(MyNotify(self.future_readiness.clone())); + let ready = self.future_readiness.clone(); + + let mut res = None; + self._run(&mut || { + ready.set_readiness(mio::EventSet::none()) + .expect("failed to set readiness"); + assert!(res.is_none()); + match task.enter(|| f.poll()) { + Poll::NotReady => {} + Poll::Ok(e) => res = Some(Ok(e)), + Poll::Err(e) => res = Some(Err(e)), + } + res.is_some() + }); + res.expect("run should not return until future is done") } - fn _run(&mut self) { - let mut events = mio::Events::new(); - self.active.set(true); - while self.active.get() { + fn _run(&mut self, done: &mut FnMut() -> bool) { + while CURRENT_LOOP.set(self, || !done()) { let amt; // On Linux, Poll::poll is epoll_wait, which may return EINTR if a // ptracer attaches. This retry loop prevents crashing when @@ -227,7 +247,7 @@ impl Loop { t - start } }); - match self.io.poll(&mut events, timeout) { + match self.io.poll(&mut self.events, timeout) { Ok(a) => { amt = a; break; @@ -246,18 +266,24 @@ impl Loop { self.consume_timeouts(start); // Next, process all the events that came in. - for i in 0..events.len() { - let event = events.get(i).unwrap(); + for i in 0..self.events.len() { + let event = self.events.get(i).unwrap(); let token = usize::from(event.token()); // Token 0 == our incoming message queue, so this means we // process the whole queue of messages. + // + // Token 1 == we should poll the future, we'll do that right + // after we get through the rest of this tick of the event loop. if token == 0 { debug!("consuming notification queue"); CURRENT_LOOP.set(&self, || { self.consume_queue(); }); continue + } else if token == 1 { + debug!("ZOMG IT'S HERE"); + continue } trace!("event {:?} {:?}", event.kind(), event.token()); @@ -418,7 +444,6 @@ impl Loop { } Message::DropSource(tok) => self.drop_source(tok), Message::Schedule(tok, wake, dir) => self.schedule(tok, wake, dir), - Message::Shutdown => self.active.set(false), Message::AddTimeout(at, slot) => { slot.try_produce(self.add_timeout(at)) @@ -604,6 +629,7 @@ impl LoopHandle { /// This method will panic if the timeout specified was not created by this /// loop handle's `add_timeout` method. pub fn cancel_timeout(&self, timeout: &TimeoutToken) { + debug!("cancel timeout {}", timeout.token); let timeout = TimeoutToken { token: timeout.token, when: timeout.when }; self.send(Message::CancelTimeout(timeout)) } @@ -638,23 +664,6 @@ impl LoopHandle { }, } } - - /// Send a message to the associated event loop that it should shut down, or - /// otherwise break out of its current loop of iteration. - /// - /// This method does not forcibly cause the event loop to shut down or - /// perform an interrupt on whatever task is currently running, instead a - /// message is simply enqueued to at a later date process the request to - /// stop looping ASAP. - /// - /// # Panics - /// - /// This function will panic if the event loop this handle is associated - /// with has gone away, or if there is an error communicating with the event - /// loop. - pub fn shutdown(&self) { - self.send(Message::Shutdown); - } } impl LoopPin { diff --git a/tests/tcp.rs b/tests/tcp.rs index a984977ab..5fb38b1b0 100644 --- a/tests/tcp.rs +++ b/tests/tcp.rs @@ -1,3 +1,4 @@ +extern crate env_logger; extern crate futures; extern crate futures_mio; @@ -17,6 +18,7 @@ macro_rules! t { #[test] fn connect() { + drop(env_logger::init()); let mut l = t!(futures_mio::Loop::new()); let srv = t!(TcpListener::bind("127.0.0.1:0")); let addr = t!(srv.local_addr()); @@ -34,6 +36,7 @@ fn connect() { #[test] fn accept() { + drop(env_logger::init()); let mut l = t!(futures_mio::Loop::new()); let srv = l.handle().tcp_listen(&"127.0.0.1:0".parse().unwrap()); let srv = t!(l.run(srv)); @@ -59,6 +62,7 @@ fn accept() { #[test] fn accept2() { + drop(env_logger::init()); let mut l = t!(futures_mio::Loop::new()); let srv = l.handle().tcp_listen(&"127.0.0.1:0".parse().unwrap()); let srv = t!(l.run(srv));