tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-25 00:00:18 +02:00
Code Issues Packages Projects Releases Wiki Activity

Remove Future::schedule

Browse Source 
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.
This commit is contained in:
Alex Crichton
2016-08-17 18:41:34 -07:00
parent 62f306629d
commit 293d104177
6 changed files with 52 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/event_loop.rs
+36 -63
View File
@@ -652,12 +652,8 @@ impl Future for AddSource {
type Item = usize;
type Error = io::Error;
fn poll(&mut self, _task: &mut Task) -> Poll<usize, io::Error> {
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<usize, io::Error> {
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<TimeoutToken, io::Error> {
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<TimeoutToken, io::Error> {
self.inner.poll(task, Loop::add_timeout, Message::AddTimeout)
}
}
@@ -723,9 +715,14 @@ impl<F, A> Future for AddLoopData<F, A>
type Item = LoopData<A>;
type Error = io::Error;
fn poll(&mut self, _task: &mut Task) -> Poll<LoopData<A>, io::Error> {
let ret = self.inner.poll(|_lp, f| {
fn poll(&mut self, task: &mut Task) -> Poll<LoopData<A>, 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<F, A> Future for AddLoopData<F, A>
}
})
}
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<A: 'static> LoopData<A> {
@@ -798,15 +786,6 @@ impl<A: Future> Future for LoopData<A> {
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<A: 'static> Drop for LoopData<A> {
@@ -975,49 +954,43 @@ struct LoopFuture<T, U> {
impl<T, U> LoopFuture<T, U>
where T: 'static,
{
fn poll<F>(&mut self, f: F) -> Poll<T, io::Error>
fn poll<F, G>(&mut self, task: &mut Task, f: F, g: G) -> Poll<T, io::Error>
where F: FnOnce(&Loop, U) -> io::Result<T>,
G: FnOnce(U, Arc<Slot<io::Result<T>>>) -> 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<F>(&mut self, task: &mut Task, f: F)
where F: FnOnce(U, Arc<Slot<io::Result<T>>>) -> 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 {
src/readiness_stream.rs
+11 -11
View File
@@ -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<Option<Ready>, io::Error> {
fn poll(&mut self, task: &mut Task) -> Poll<Option<Ready>, 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 {
src/tcp.rs
-17
View File
@@ -141,10 +141,6 @@ impl Stream for TcpListener {
fn poll(&mut self, task: &mut Task) -> Poll<Option<Ready>, 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<Option<Ready>, io::Error> {
self.ready.poll(task)
}
fn schedule(&mut self, task: &mut Task) {
self.ready.schedule(task)
}
}
#[cfg(unix)]
src/timeout.rs
+2 -5
View File
@@ -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 {
src/udp.rs
-4
View File
@@ -243,10 +243,6 @@ impl Stream for UdpSocket {
fn poll(&mut self, task: &mut Task) -> Poll<Option<Ready>, io::Error> {
self.ready.poll(task)
}
fn schedule(&mut self, task: &mut Task) {
self.ready.schedule(task)
}
}
#[cfg(unix)]
tests/echo.rs
+3
View File
@@ -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));