From 70eca184f0f52dd2d634b750bba698da9418ee5f Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Mon, 1 Jul 2019 18:27:13 -0700 Subject: [PATCH] tokio: re-enable timer in runtimes (#1237) This also brings back the timer tests in the tokio crate. --- tokio/Cargo.toml | 8 +- tokio/src/lib.rs | 2 +- tokio/src/prelude.rs | 2 +- tokio/src/runtime/current_thread/builder.rs | 22 ++-- tokio/src/runtime/current_thread/runtime.rs | 49 ++++---- tokio/src/timer.rs | 9 -- tokio/src/util/future.rs | 35 +----- tokio/src/util/mod.rs | 6 +- tokio/src/util/stream.rs | 2 +- tokio/tests/timer.rs | 131 +++++++++----------- 10 files changed, 104 insertions(+), 162 deletions(-) diff --git a/tokio/Cargo.toml b/tokio/Cargo.toml index 161d3b639..2b68f70b6 100644 --- a/tokio/Cargo.toml +++ b/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 diff --git a/tokio/src/lib.rs b/tokio/src/lib.rs index d554ab007..943d840d2 100644 --- a/tokio/src/lib.rs +++ b/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; diff --git a/tokio/src/prelude.rs b/tokio/src/prelude.rs index 000a680c9..921c8eef0 100644 --- a/tokio/src/prelude.rs +++ b/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}; diff --git a/tokio/src/runtime/current_thread/builder.rs b/tokio/src/runtime/current_thread/builder.rs index 7b0f0c012..ff8550f83 100644 --- a/tokio/src/runtime/current_thread/builder.rs +++ b/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::timer::Timer; +use tokio_timer::clock::Clock; +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 -//clock: Clock, + /// The clock to use + 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 { @@ -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_handle = timer.handle(); + let timer = Timer::new_with_now(reactor, self.clock.clone()); + 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, - //self.clock.clone(), + timer_handle, + self.clock.clone(), executor, ); diff --git a/tokio/src/runtime/current_thread/runtime.rs b/tokio/src/runtime/current_thread/runtime.rs index 0ef5988ec..3a8f8832a 100644 --- a/tokio/src/runtime/current_thread/runtime.rs +++ b/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::timer::{self, Timer}; +use tokio_timer::clock::{self, Clock}; +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, - //clock: Clock, + timer_handle: timer::Handle, + clock: Clock, executor: CurrentThread, } -//pub(super) type Parker = Timer; -pub(super) type Parker = Reactor; +pub(super) type Parker = Timer; /// 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, - //clock: Clock, + timer_handle: timer::Handle, + clock: Clock, executor: CurrentThread, ) -> Runtime { Runtime { reactor_handle, - //timer_handle, - //clock, + timer_handle, + clock, executor, } } @@ -197,8 +196,8 @@ impl Runtime { { let Runtime { ref reactor_handle, - //ref timer_handle, - //ref clock, + ref timer_handle, + 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| { - // timer::with_default(&timer_handle, enter, |enter| { - // The TaskExecutor is a fake executor that looks into the - // current single-threaded executor when used. This is a trick, - // because we need two mutable references to the executor (one - // to run the provided future, another to install as the default - // one). We use the fake one here as the default one. - let mut default_executor = current_thread::TaskExecutor::current(); - tokio_executor::with_default(&mut default_executor, enter, |enter| { - let mut executor = executor.enter(enter); - f(&mut executor) + clock::with_default(clock, || { + timer::with_default(&timer_handle, || { + // The TaskExecutor is a fake executor that looks into the + // current single-threaded executor when used. This is a trick, + // because we need two mutable references to the executor (one + // to run the provided future, another to install as the default + // one). We use the fake one here as the default one. + let mut default_executor = current_thread::TaskExecutor::current(); + tokio_executor::with_default(&mut default_executor, enter, |enter| { + let mut executor = executor.enter(enter); + f(&mut executor) + }) + }) }) - // }) - //}) }) } } diff --git a/tokio/src/timer.rs b/tokio/src/timer.rs index 98de0018f..716524c81 100644 --- a/tokio/src/timer.rs +++ b/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 = ::tokio_timer::Deadline; -#[deprecated(since = "0.1.8", note = "use Timeout instead")] -#[allow(deprecated)] -#[doc(hidden)] -pub type DeadlineError = ::tokio_timer::DeadlineError; diff --git a/tokio/src/util/future.rs b/tokio/src/util/future.rs index ea9d0ff39..a5fb70a0d 100644 --- a/tokio/src/util/future.rs +++ b/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 - where - Self: Sized, - { - Deadline::new(self, deadline) - } } impl 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()); - } -} diff --git a/tokio/src/util/mod.rs b/tokio/src/util/mod.rs index 58fd3d0b0..3a9ec31f9 100644 --- a/tokio/src/util/mod.rs +++ b/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; diff --git a/tokio/src/util/stream.rs b/tokio/src/util/stream.rs index 9b2d08522..001a4496b 100644 --- a/tokio/src/util/stream.rs +++ b/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}; diff --git a/tokio/tests/timer.rs b/tokio/tests/timer.rs index 8790f5bad..afb9f8367 100644 --- a/tokio/tests/timer.rs +++ b/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] -fn timer_with_runtime() { - let _ = env_logger::try_init(); +use std::sync::mpsc; +use std::time::{Duration, Instant}; - 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({ - Delay::new(when) - .map_err(|e| panic!("unexpected error; err={:?}", e)) - .and_then(move |_| { - assert!(Instant::now() >= when); - tx.send(()).unwrap(); - Ok(()) - }) + 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(); } #[test] -fn starving() { - use futures::{task, Async, Poll}; +fn timer_with_current_thread_runtime() { + 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 Error = (); + type Output = u64; - fn poll(&mut self) -> Poll { - if self.0.poll().unwrap().is_ready() { - return Ok(self.1.into()); + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + if Pin::new(&mut self.0).poll(cx).is_ready() { + 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(); - - tokio::run({ - starve.and_then(move |_ticks| { - assert!(Instant::now() >= when); - tx.send(()).unwrap(); - Ok(()) - }) - }); - - rx.recv().unwrap(); + starve.await; + assert!(Instant::now() >= when); } -#[test] -fn deadline() { - use futures::future; +#[tokio::test] +async fn timeout() { + use tokio::sync::oneshot; - let _ = env_logger::try_init(); + let (_tx, rx) = oneshot::channel::<()>(); - let when = Instant::now() + Duration::from_millis(20); - let (tx, rx) = mpsc::channel(); + let now = Instant::now(); + let dur = Duration::from_millis(20); - #[allow(deprecated)] - tokio::run({ - future::empty::<(), ()>().deadline(when).then(move |res| { - 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(); + let res = rx.timeout(dur).await; + assert!(res.is_err()); + assert!(Instant::now() >= now + dur); }