tokio: re-enable timer in runtimes (#1237)

This also brings back the timer tests in the tokio crate.

tokio/Cargo.toml

@@ -33,7 +33,7 @@ default = [
   "rt-full",
   "sync",
   "tcp",
-#  "timer",
+  "timer",
   "udp",
 #  "uds",
 ]
@@ -45,7 +45,7 @@ reactor = ["io", "tokio-reactor"]
 rt-full = [
   "num_cpus",
   "reactor",
-#  "timer",
+  "timer",
   "tokio-current-thread",
   "tokio-executor",
   "tokio-macros",
@@ -54,7 +54,7 @@ rt-full = [
 ]
 sync = ["tokio-sync"]
 tcp = ["tokio-tcp"]
-#timer = ["tokio-timer"]
+timer = ["tokio-timer"]
 udp = ["tokio-udp"]
 #uds = ["tokio-uds"]
 
@@ -76,7 +76,7 @@ tokio-sync = { version = "0.2.0", optional = true, path = "../tokio-sync" }
 #tokio-threadpool = { version = "0.2.0", optional = true, path = "../tokio-threadpool" }
 tokio-tcp = { version = "0.2.0", optional = true, path = "../tokio-tcp" }
 tokio-udp = { version = "0.2.0", optional = true, path = "../tokio-udp" }
-#tokio-timer = { version = "0.3.0", optional = true, path = "../tokio-timer" }
+tokio-timer = { version = "0.3.0", optional = true, path = "../tokio-timer" }
 tracing-core = { version = "0.1", optional = true }
 
 # Needed for async/await preview support

tokio/src/lib.rs

@@ -100,7 +100,7 @@ pub mod reactor;
 pub mod sync;
 #[cfg(feature = "timer")]
 pub mod timer;
-//pub mod util;
+pub mod util;
 
 if_runtime! {
     pub mod executor;

tokio/src/prelude.rs

@@ -10,8 +10,8 @@
 //!
 //! The prelude may grow over time as additional items see ubiquitous use.
 
+pub use crate::util::FutureExt;
 pub use std::future::Future;
 pub use std::task::{self, Poll};
-//pub use crate::util::{FutureExt, StreamExt};
 #[cfg(feature = "io")]
 pub use tokio_io::{AsyncRead, AsyncWrite};

tokio/src/runtime/current_thread/builder.rs

@@ -1,8 +1,8 @@
 use crate::runtime::current_thread::Runtime;
 use tokio_current_thread::CurrentThread;
 use tokio_reactor::Reactor;
-//use tokio_timer::clock::Clock;
+use tokio_timer::clock::Clock;
-//use tokio_timer::timer::Timer;
+use tokio_timer::timer::Timer;
 use std::io;
 
 /// Builds a Single-threaded runtime with custom configuration values.
@@ -35,8 +35,8 @@ use std::io;
 /// ```
 #[derive(Debug)]
 pub struct Builder {
-    // /// The clock to use
+    /// The clock to use
-//clock: Clock,
+    clock: Clock,
 }
 
 impl Builder {
@@ -46,17 +46,15 @@ impl Builder {
     /// Configuration methods can be chained on the return value.
     pub fn new() -> Builder {
         Builder {
-            //clock: Clock::new(),
+            clock: Clock::new(),
         }
     }
 
-    /*
     /// Set the `Clock` instance that will be used by the runtime.
     pub fn clock(&mut self, clock: Clock) -> &mut Self {
         self.clock = clock;
         self
     }
-    */
 
     /// Create the configured `Runtime`.
     pub fn build(&mut self) -> io::Result<Runtime> {
@@ -66,18 +64,18 @@ impl Builder {
 
         // Place a timer wheel on top of the reactor. If there are no timeouts to fire, it'll let the
         // reactor pick up some new external events.
-        //let timer = Timer::new_with_now(reactor, self.clock.clone());
+        let timer = Timer::new_with_now(reactor, self.clock.clone());
-        //let timer_handle = timer.handle();
+        let timer_handle = timer.handle();
 
         // And now put a single-threaded executor on top of the timer. When there are no futures ready
         // to do something, it'll let the timer or the reactor to generate some new stimuli for the
         // futures to continue in their life.
-        let executor = CurrentThread::new_with_park(reactor /*timer*/);
+        let executor = CurrentThread::new_with_park(timer);
 
         let runtime = Runtime::new2(
             reactor_handle,
-            //timer_handle,
+            timer_handle,
-            //self.clock.clone(),
+            self.clock.clone(),
             executor,
         );
 

tokio/src/runtime/current_thread/runtime.rs

@@ -3,8 +3,8 @@ use tokio_current_thread::Handle as ExecutorHandle;
 use tokio_current_thread::{self as current_thread, CurrentThread};
 use tokio_executor;
 use tokio_reactor::{self, Reactor};
-//use tokio_timer::clock::{self, Clock};
+use tokio_timer::clock::{self, Clock};
-//use tokio_timer::timer::{self, Timer};
+use tokio_timer::timer::{self, Timer};
 use std::error::Error;
 use std::fmt;
 use std::future::Future;
@@ -19,13 +19,12 @@ use std::io;
 #[derive(Debug)]
 pub struct Runtime {
     reactor_handle: tokio_reactor::Handle,
-    //timer_handle: timer::Handle,
+    timer_handle: timer::Handle,
-    //clock: Clock,
+    clock: Clock,
     executor: CurrentThread<Parker>,
 }
 
-//pub(super) type Parker = Timer<Reactor>;
+pub(super) type Parker = Timer<Reactor>;
-pub(super) type Parker = Reactor;
 
 /// Handle to spawn a future on the corresponding `CurrentThread` runtime instance
 #[derive(Debug, Clone)]
@@ -101,14 +100,14 @@ impl Runtime {
 
     pub(super) fn new2(
         reactor_handle: tokio_reactor::Handle,
-        //timer_handle: timer::Handle,
+        timer_handle: timer::Handle,
-        //clock: Clock,
+        clock: Clock,
         executor: CurrentThread<Parker>,
     ) -> Runtime {
         Runtime {
             reactor_handle,
-            //timer_handle,
+            timer_handle,
-            //clock,
+            clock,
             executor,
         }
     }
@@ -197,8 +196,8 @@ impl Runtime {
     {
         let Runtime {
             ref reactor_handle,
-            //ref timer_handle,
+            ref timer_handle,
-            //ref clock,
+            ref clock,
             ref mut executor,
             ..
         } = *self;
@@ -209,20 +208,20 @@ impl Runtime {
         // This will set the default handle and timer to use inside the closure
         // and run the future.
         tokio_reactor::with_default(&reactor_handle, &mut enter, |enter| {
-            //clock::with_default(clock, enter, |enter| {
+            clock::with_default(clock, || {
-            //    timer::with_default(&timer_handle, enter, |enter| {
+                timer::with_default(&timer_handle, || {
-            // The TaskExecutor is a fake executor that looks into the
+                    // The TaskExecutor is a fake executor that looks into the
-            // current single-threaded executor when used. This is a trick,
+                    // current single-threaded executor when used. This is a trick,
-            // because we need two mutable references to the executor (one
+                    // because we need two mutable references to the executor (one
-            // to run the provided future, another to install as the default
+                    // to run the provided future, another to install as the default
-            // one). We use the fake one here as the default one.
+                    // one). We use the fake one here as the default one.
-            let mut default_executor = current_thread::TaskExecutor::current();
+                    let mut default_executor = current_thread::TaskExecutor::current();
-            tokio_executor::with_default(&mut default_executor, enter, |enter| {
+                    tokio_executor::with_default(&mut default_executor, enter, |enter| {
-                let mut executor = executor.enter(enter);
+                        let mut executor = executor.enter(enter);
-                f(&mut executor)
+                        f(&mut executor)
+                    })
+                })
             })
-            //    })
-            //})
         })
     }
 }

tokio/src/timer.rs

@@ -81,12 +81,3 @@
 //! [`DelayQueue`]: struct.DelayQueue.html
 
 pub use tokio_timer::{delay_queue, timeout, Delay, DelayQueue, Error, Interval, Timeout};
-
-#[deprecated(since = "0.1.8", note = "use Timeout instead")]
-#[allow(deprecated)]
-#[doc(hidden)]
-pub type Deadline<T> = ::tokio_timer::Deadline<T>;
-#[deprecated(since = "0.1.8", note = "use Timeout instead")]
-#[allow(deprecated)]
-#[doc(hidden)]
-pub type DeadlineError<T> = ::tokio_timer::DeadlineError<T>;

tokio/src/util/future.rs

@@ -1,14 +1,10 @@
-use futures::Future;
-
-#[cfg(feature = "timer")]
-#[allow(deprecated)]
-use tokio_timer::Deadline;
-
 #[cfg(feature = "timer")]
 use tokio_timer::Timeout;
 
 #[cfg(feature = "timer")]
-use std::time::{Duration, Instant};
+use std::time::Duration;
+
+use std::future::Future;
 
 /// An extension trait for `Future` that provides a variety of convenient
 /// combinator functions.
@@ -62,31 +58,6 @@ pub trait FutureExt: Future {
     {
         Timeout::new(self, timeout)
     }
-
-    #[cfg(feature = "timer")]
-    #[deprecated(since = "0.1.8", note = "use `timeout` instead")]
-    #[allow(deprecated)]
-    #[doc(hidden)]
-    fn deadline(self, deadline: Instant) -> Deadline<Self>
-    where
-        Self: Sized,
-    {
-        Deadline::new(self, deadline)
-    }
 }
 
 impl<T: ?Sized> FutureExt for T where T: Future {}
-
-#[cfg(test)]
-mod test {
-    use super::*;
-    use crate::prelude::future;
-
-    #[cfg(feature = "timer")]
-    #[test]
-    fn timeout_polls_at_least_once() {
-        let base_future = future::result::<(), ()>(Ok(()));
-        let timeouted_future = base_future.timeout(Duration::new(0, 0));
-        assert!(timeouted_future.wait().is_ok());
-    }
-}

tokio/src/util/mod.rs

@@ -7,9 +7,9 @@
 //! [`FutureExt`]: trait.FutureExt.html
 //! [`StreamExt`]: trait.StreamExt.html
 
-mod enumerate;
+// mod enumerate;
 mod future;
-mod stream;
+// mod stream;
 
 pub use self::future::FutureExt;
-pub use self::stream::StreamExt;
+// pub use self::stream::StreamExt;

tokio/src/util/stream.rs

@@ -1,8 +1,8 @@
 pub use crate::util::enumerate::Enumerate;
 
-use futures::Stream;
 #[cfg(feature = "timer")]
 use std::time::Duration;
+
 #[cfg(feature = "timer")]
 use tokio_timer::{throttle::Throttle, Timeout};
 

tokio/tests/timer.rs

@@ -1,112 +1,95 @@
-#![cfg(feature = "broken")]
 #![deny(warnings, rust_2018_idioms)]
+#![feature(async_await)]
 
-use env_logger;
-use std::sync::mpsc;
-use std::time::{Duration, Instant};
 use tokio;
 use tokio::prelude::*;
+// use tokio::sync::mpsc;
 use tokio::timer::*;
 
-#[test]
+use std::sync::mpsc;
-fn timer_with_runtime() {
+use std::time::{Duration, Instant};
-    let _ = env_logger::try_init();
 
-    let when = Instant::now() + Duration::from_millis(100);
+#[test]
+fn timer_with_threaded_runtime() {
+    use tokio::runtime::Runtime;
+
+    let mut rt = Runtime::new().unwrap();
     let (tx, rx) = mpsc::channel();
 
-    tokio::run({
+    rt.spawn(async move {
-        Delay::new(when)
+        let when = Instant::now() + Duration::from_millis(100);
-            .map_err(|e| panic!("unexpected error; err={:?}", e))
+
-            .and_then(move |_| {
+        Delay::new(when).await;
-                assert!(Instant::now() >= when);
+        assert!(Instant::now() >= when);
-                tx.send(()).unwrap();
+
-                Ok(())
+        tx.send(()).unwrap();
-            })
     });
 
+    rt.run().unwrap();
     rx.recv().unwrap();
 }
 
 #[test]
-fn starving() {
+fn timer_with_current_thread_runtime() {
-    use futures::{task, Async, Poll};
+    use tokio::runtime::current_thread::Runtime;
 
-    let _ = env_logger::try_init();
+    let mut rt = Runtime::new().unwrap();
+    let (tx, rx) = mpsc::channel();
+
+    rt.spawn(async move {
+        let when = Instant::now() + Duration::from_millis(100);
+
+        Delay::new(when).await;
+        assert!(Instant::now() >= when);
+
+        tx.send(()).unwrap();
+    });
+
+    rt.run().unwrap();
+    rx.recv().unwrap();
+}
+
+#[tokio::test]
+async fn starving() {
+    use std::future::Future;
+    use std::pin::Pin;
+    use std::task::{Context, Poll};
 
     struct Starve(Delay, u64);
 
     impl Future for Starve {
-        type Item = u64;
+        type Output = u64;
-        type Error = ();
 
-        fn poll(&mut self) -> Poll<Self::Item, ()> {
+        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u64> {
-            if self.0.poll().unwrap().is_ready() {
+            if Pin::new(&mut self.0).poll(cx).is_ready() {
-                return Ok(self.1.into());
+                return Poll::Ready(self.1);
             }
 
             self.1 += 1;
 
-            task::current().notify();
+            cx.waker().wake_by_ref();
 
-            Ok(Async::NotReady)
+            Poll::Pending
         }
     }
 
     let when = Instant::now() + Duration::from_millis(20);
     let starve = Starve(Delay::new(when), 0);
 
-    let (tx, rx) = mpsc::channel();
+    starve.await;
-
+    assert!(Instant::now() >= when);
-    tokio::run({
-        starve.and_then(move |_ticks| {
-            assert!(Instant::now() >= when);
-            tx.send(()).unwrap();
-            Ok(())
-        })
-    });
-
-    rx.recv().unwrap();
 }
 
-#[test]
+#[tokio::test]
-fn deadline() {
+async fn timeout() {
-    use futures::future;
+    use tokio::sync::oneshot;
 
-    let _ = env_logger::try_init();
+    let (_tx, rx) = oneshot::channel::<()>();
 
-    let when = Instant::now() + Duration::from_millis(20);
+    let now = Instant::now();
-    let (tx, rx) = mpsc::channel();
+    let dur = Duration::from_millis(20);
 
-    #[allow(deprecated)]
+    let res = rx.timeout(dur).await;
-    tokio::run({
+    assert!(res.is_err());
-        future::empty::<(), ()>().deadline(when).then(move |res| {
+    assert!(Instant::now() >= now + dur);
-            assert!(res.is_err());
-            tx.send(()).unwrap();
-            Ok(())
-        })
-    });
-
-    rx.recv().unwrap();
-}
-
-#[test]
-fn timeout() {
-    use futures::future;
-
-    let _ = env_logger::try_init();
-
-    let (tx, rx) = mpsc::channel();
-
-    tokio::run({
-        future::empty::<(), ()>()
-            .timeout(Duration::from_millis(20))
-            .then(move |res| {
-                assert!(res.is_err());
-                tx.send(()).unwrap();
-                Ok(())
-            })
-    });
-
-    rx.recv().unwrap();
 }