From 293d1041770384c9eeb34ac7d97214feaf3b88c3 Mon Sep 17 00:00:00 2001 From: Alex Crichton Date: Tue, 16 Aug 2016 15:56:57 -0700 Subject: [PATCH] Remove Future::schedule A more appealing model is actually just automatically inferring what needs to be scheduled based on what actions are done during poll. For example if during a poll you check a oneshot channel, then the current task is registered for being woken up if it's not ready. Similarly this will apply to I/O where if I/O is attempted but we see EAGAIN then we'll schedule the task to get notified when it's ready. This may also have performance benefits in some niche situations because you don't need to recompute where you are in the state machine both during poll and during schedule. Instead, it now happens all at once. --- src/event_loop.rs | 99 +++++++++++++++-------------------------- src/readiness_stream.rs | 22 ++++----- src/tcp.rs | 17 ------- src/timeout.rs | 7 +-- src/udp.rs | 4 -- tests/echo.rs | 3 ++ 6 files changed, 52 insertions(+), 100 deletions(-) diff --git a/src/event_loop.rs b/src/event_loop.rs index 06ee3a01a..8a68013c3 100644 --- a/src/event_loop.rs +++ b/src/event_loop.rs @@ -652,12 +652,8 @@ impl Future for AddSource { type Item = usize; type Error = io::Error; - fn poll(&mut self, _task: &mut Task) -> Poll { - self.inner.poll(Loop::add_source) - } - - fn schedule(&mut self, task: &mut Task) { - self.inner.schedule(task, Message::AddSource) + fn poll(&mut self, task: &mut Task) -> Poll { + self.inner.poll(task, Loop::add_source, Message::AddSource) } } @@ -676,12 +672,8 @@ impl Future for AddTimeout { type Item = TimeoutToken; type Error = io::Error; - fn poll(&mut self, _task: &mut Task) -> Poll { - self.inner.poll(Loop::add_timeout) - } - - fn schedule(&mut self, task: &mut Task) { - self.inner.schedule(task, Message::AddTimeout) + fn poll(&mut self, task: &mut Task) -> Poll { + self.inner.poll(task, Loop::add_timeout, Message::AddTimeout) } } @@ -723,9 +715,14 @@ impl Future for AddLoopData type Item = LoopData; type Error = io::Error; - fn poll(&mut self, _task: &mut Task) -> Poll, io::Error> { - let ret = self.inner.poll(|_lp, f| { + fn poll(&mut self, task: &mut Task) -> Poll, io::Error> { + let ret = self.inner.poll(task, |_lp, f| { Ok(DropBox::new(f())) + }, |f, slot| { + Message::Run(Box::new(move || { + slot.try_produce(Ok(DropBox::new(f()))).ok() + .expect("add loop data try_produce intereference"); + })) }); ret.map(|data| { @@ -735,15 +732,6 @@ impl Future for AddLoopData } }) } - - fn schedule(&mut self, task: &mut Task) { - self.inner.schedule(task, |f, slot| { - Message::Run(Box::new(move || { - slot.try_produce(Ok(DropBox::new(f()))).ok() - .expect("add loop data try_produce intereference"); - })) - }) - } } impl LoopData { @@ -798,15 +786,6 @@ impl Future for LoopData { task.poll_on(self.executor()); Poll::NotReady } - - fn schedule(&mut self, task: &mut Task) { - // If we're on the right thread, then we're good to go, otherwise we - // need to get poll'd to tell the task to move somewhere else. - match self.get_mut() { - Some(inner) => inner.schedule(task), - None => task.notify(), - } - } } impl Drop for LoopData { @@ -975,49 +954,43 @@ struct LoopFuture { impl LoopFuture where T: 'static, { - fn poll(&mut self, f: F) -> Poll + fn poll(&mut self, task: &mut Task, f: F, g: G) -> Poll where F: FnOnce(&Loop, U) -> io::Result, + G: FnOnce(U, Arc>>) -> Message, { match self.result { - Some((ref result, ref token)) => { + Some((ref result, ref mut token)) => { result.cancel(*token); match result.try_consume() { - Ok(t) => t.into(), - Err(_) => Poll::NotReady, + Ok(t) => return t.into(), + Err(_) => {} } + let handle = task.handle().clone(); + *token = result.on_full(move |_| { + handle.notify(); + }); + return Poll::NotReady } None => { let data = &mut self.data; - self.loop_handle.with_loop(|lp| { - match lp { - Some(lp) => f(lp, data.take().unwrap()).into(), - None => Poll::NotReady, - } - }) + let ret = self.loop_handle.with_loop(|lp| { + lp.map(|lp| f(lp, data.take().unwrap())) + }); + if let Some(ret) = ret { + return ret.into() + } + + let handle = task.handle().clone(); + let result = Arc::new(Slot::new(None)); + let token = result.on_full(move |_| { + handle.notify(); + }); + self.result = Some((result.clone(), token)); + self.loop_handle.send(g(data.take().unwrap(), result)); + Poll::NotReady } } } - - fn schedule(&mut self, task: &mut Task, f: F) - where F: FnOnce(U, Arc>>) -> Message, - { - if let Some((ref result, ref mut token)) = self.result { - result.cancel(*token); - let handle = task.handle().clone(); - *token = result.on_full(move |_| { - handle.notify(); - }); - return - } - - let handle = task.handle().clone(); - let result = Arc::new(Slot::new(None)); - let token = result.on_full(move |_| { - handle.notify(); - }); - self.result = Some((result.clone(), token)); - self.loop_handle.send(f(self.data.take().unwrap(), result)) - } } impl TimeoutState { diff --git a/src/readiness_stream.rs b/src/readiness_stream.rs index 9e825f8c9..efa180421 100644 --- a/src/readiness_stream.rs +++ b/src/readiness_stream.rs @@ -60,26 +60,26 @@ impl Future for ReadinessStreamNew { } }) } - - fn schedule(&mut self, task: &mut Task) { - self.inner.schedule(task) - } } impl Stream for ReadinessStream { type Item = Ready; type Error = io::Error; - fn poll(&mut self, _task: &mut Task) -> Poll, io::Error> { + fn poll(&mut self, task: &mut Task) -> Poll, io::Error> { match self.source.take_readiness() { - None => Poll::NotReady, - Some(r) => Poll::Ok(Some(r)), + None => { + self.loop_handle.schedule(self.io_token, task); + Poll::NotReady + } + Some(r) => { + if !r.is_read() || !r.is_write() { + self.loop_handle.schedule(self.io_token, task); + } + Poll::Ok(Some(r)) + } } } - - fn schedule(&mut self, task: &mut Task) { - self.loop_handle.schedule(self.io_token, task) - } } impl Drop for ReadinessStream { diff --git a/src/tcp.rs b/src/tcp.rs index 0f8384981..90c768465 100644 --- a/src/tcp.rs +++ b/src/tcp.rs @@ -141,10 +141,6 @@ impl Stream for TcpListener { fn poll(&mut self, task: &mut Task) -> Poll, io::Error> { self.ready.poll(task) } - - fn schedule(&mut self, task: &mut Task) { - self.ready.schedule(task) - } } /// An I/O object representing a TCP stream connected to a remote endpoint. @@ -297,15 +293,6 @@ impl Future for TcpStreamNew { *self = TcpStreamNew::Waiting(stream); Poll::NotReady } - - fn schedule(&mut self, task: &mut Task) { - match *self { - TcpStreamNew::Waiting(ref mut s) => { - s.ready.schedule(task); - } - TcpStreamNew::Empty => task.notify(), - } - } } impl Read for TcpStream { @@ -355,10 +342,6 @@ impl Stream for TcpStream { fn poll(&mut self, task: &mut Task) -> Poll, io::Error> { self.ready.poll(task) } - - fn schedule(&mut self, task: &mut Task) { - self.ready.schedule(task) - } } #[cfg(unix)] diff --git a/src/timeout.rs b/src/timeout.rs index c0f6eb4db..dccbd91e7 100644 --- a/src/timeout.rs +++ b/src/timeout.rs @@ -50,18 +50,15 @@ impl Future for Timeout { type Item = (); type Error = io::Error; - fn poll(&mut self, _task: &mut Task) -> Poll<(), io::Error> { + fn poll(&mut self, task: &mut Task) -> Poll<(), io::Error> { // TODO: is this fast enough? if self.at <= Instant::now() { Poll::Ok(()) } else { + self.handle.update_timeout(&self.token, task); Poll::NotReady } } - - fn schedule(&mut self, task: &mut Task) { - self.handle.update_timeout(&self.token, task); - } } impl Drop for Timeout { diff --git a/src/udp.rs b/src/udp.rs index 847694c2b..754fc8b9c 100644 --- a/src/udp.rs +++ b/src/udp.rs @@ -243,10 +243,6 @@ impl Stream for UdpSocket { fn poll(&mut self, task: &mut Task) -> Poll, io::Error> { self.ready.poll(task) } - - fn schedule(&mut self, task: &mut Task) { - self.ready.schedule(task) - } } #[cfg(unix)] diff --git a/tests/echo.rs b/tests/echo.rs index a9ee4e56c..6247a10a8 100644 --- a/tests/echo.rs +++ b/tests/echo.rs @@ -1,3 +1,4 @@ +extern crate env_logger; extern crate futures; extern crate futures_io; extern crate futures_mio; @@ -19,6 +20,8 @@ macro_rules! t { #[test] fn echo_server() { + 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));