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<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 {
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<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 {
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<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)]
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<Option<Ready>, 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));