time: lazy init TimerShared in TimerEntry

benches/Cargo.toml

@@ -90,3 +90,8 @@ harness = false
 name = "time_now"
 path = "time_now.rs"
 harness = false
+
+[[bench]]
+name = "time_timeout"
+path = "time_timeout.rs"
+harness = false

benches/time_timeout.rs

@@ -0,0 +1,114 @@
+//! Benchmark spawning a task onto the basic and threaded Tokio executors.
+//! This essentially measure the time to enqueue a task in the local and remote
+//! case.
+
+use std::time::{Duration, Instant};
+
+use criterion::{black_box, criterion_group, criterion_main, Criterion};
+use tokio::{
+    runtime::Runtime,
+    time::{sleep, timeout},
+};
+
+// a vevry quick async task, but might timeout
+async fn quick_job() -> usize {
+    1
+}
+
+fn build_run_time(workers: usize) -> Runtime {
+    if workers == 1 {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .worker_threads(workers)
+            .build()
+            .unwrap()
+    } else {
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .worker_threads(workers)
+            .build()
+            .unwrap()
+    }
+}
+
+fn single_thread_scheduler_timeout(c: &mut Criterion) {
+    do_timeout_test(c, 1, "single_thread_timeout");
+}
+
+fn multi_thread_scheduler_timeout(c: &mut Criterion) {
+    do_timeout_test(c, 8, "multi_thread_timeout-8");
+}
+
+fn do_timeout_test(c: &mut Criterion, workers: usize, name: &str) {
+    let runtime = build_run_time(workers);
+    c.bench_function(name, |b| {
+        b.iter_custom(|iters| {
+            let start = Instant::now();
+            runtime.block_on(async {
+                black_box(spawn_timeout_job(iters as usize, workers).await);
+            });
+            start.elapsed()
+        })
+    });
+}
+
+async fn spawn_timeout_job(iters: usize, procs: usize) {
+    let mut handles = Vec::with_capacity(procs);
+    for _ in 0..procs {
+        handles.push(tokio::spawn(async move {
+            for _ in 0..iters / procs {
+                let h = timeout(Duration::from_secs(1), quick_job());
+                assert_eq!(black_box(h.await.unwrap()), 1);
+            }
+        }));
+    }
+    for handle in handles {
+        handle.await.unwrap();
+    }
+}
+
+fn single_thread_scheduler_sleep(c: &mut Criterion) {
+    do_sleep_test(c, 1, "single_thread_sleep");
+}
+
+fn multi_thread_scheduler_sleep(c: &mut Criterion) {
+    do_sleep_test(c, 8, "multi_thread_sleep-8");
+}
+
+fn do_sleep_test(c: &mut Criterion, workers: usize, name: &str) {
+    let runtime = build_run_time(workers);
+
+    c.bench_function(name, |b| {
+        b.iter_custom(|iters| {
+            let start = Instant::now();
+            runtime.block_on(async {
+                black_box(spawn_sleep_job(iters as usize, workers).await);
+            });
+            start.elapsed()
+        })
+    });
+}
+
+async fn spawn_sleep_job(iters: usize, procs: usize) {
+    let mut handles = Vec::with_capacity(procs);
+    for _ in 0..procs {
+        handles.push(tokio::spawn(async move {
+            for _ in 0..iters / procs {
+                let _h = black_box(sleep(Duration::from_secs(1)));
+            }
+        }));
+    }
+    for handle in handles {
+        handle.await.unwrap();
+    }
+}
+
+criterion_group!(
+    timeout_benchmark,
+    single_thread_scheduler_timeout,
+    multi_thread_scheduler_timeout,
+    single_thread_scheduler_sleep,
+    multi_thread_scheduler_sleep
+);
+
+criterion_main!(timeout_benchmark);

tokio/src/runtime/time/entry.rs

@@ -496,10 +496,6 @@ impl TimerEntry {
         unsafe { &*self.inner.get() }
     }
 
-    pub(crate) fn deadline(&self) -> Instant {
-        self.deadline
-    }
-
     pub(crate) fn is_elapsed(&self) -> bool {
         !self.inner().state.might_be_registered() && self.registered
     }

tokio/src/time/sleep.rs

@@ -1,3 +1,4 @@
+use crate::runtime::scheduler;
 use crate::runtime::time::TimerEntry;
 use crate::time::{error::Error, Duration, Instant};
 use crate::util::trace;
@@ -226,10 +227,11 @@ pin_project! {
     #[must_use = "futures do nothing unless you `.await` or poll them"]
     pub struct Sleep {
         inner: Inner,
-
+        deadline : Instant,
+        handle: scheduler::Handle,
         // The link between the `Sleep` instance and the timer that drives it.
         #[pin]
-        entry: TimerEntry,
+        entry: Option<TimerEntry>,
     }
 }
 
@@ -256,8 +258,16 @@ impl Sleep {
         use crate::runtime::scheduler;
 
         let handle = scheduler::Handle::current();
-        let entry = TimerEntry::new(&handle, deadline);
+        Self::new_timeout_with_handle(deadline, location, handle)
+    }
 
+    #[cfg_attr(not(all(tokio_unstable, feature = "tracing")), allow(unused_variables))]
+    #[track_caller]
+    pub(crate) fn new_timeout_with_handle(
+        deadline: Instant,
+        location: Option<&'static Location<'static>>,
+        handle: crate::runtime::scheduler::Handle,
+    ) -> Sleep {
         #[cfg(all(tokio_unstable, feature = "tracing"))]
         let inner = {
             let clock = handle.driver().clock();
@@ -302,23 +312,27 @@ impl Sleep {
         #[cfg(not(all(tokio_unstable, feature = "tracing")))]
         let inner = Inner {};
 
-        Sleep { inner, entry }
-    }
-
-    pub(crate) fn far_future(location: Option<&'static Location<'static>>) -> Sleep {
-        Self::new_timeout(Instant::far_future(), location)
+        Sleep {
+            inner,
+            deadline,
+            handle,
+            entry: None,
+        }
     }
 
     /// Returns the instant at which the future will complete.
     pub fn deadline(&self) -> Instant {
-        self.entry.deadline()
+        self.deadline
     }
 
     /// Returns `true` if `Sleep` has elapsed.
     ///
     /// A `Sleep` instance is elapsed when the requested duration has elapsed.
     pub fn is_elapsed(&self) -> bool {
-        self.entry.is_elapsed()
+        if self.entry.is_none() {
+            return false;
+        }
+        self.entry.as_ref().unwrap().is_elapsed()
     }
 
     /// Resets the `Sleep` instance to a new deadline.
@@ -362,17 +376,27 @@ impl Sleep {
     /// without having it registered. This is required in e.g. the
     /// [`crate::time::Interval`] where we want to reset the internal [Sleep]
     /// without having it wake up the last task that polled it.
-    pub(crate) fn reset_without_reregister(self: Pin<&mut Self>, deadline: Instant) {
-        let mut me = self.project();
-        me.entry.as_mut().reset(deadline, false);
+    pub(crate) fn reset_without_reregister(mut self: Pin<&mut Self>, deadline: Instant) {
+        self.as_mut().lazy_init_timer_entry(deadline);
+        self.project()
+            .entry
+            .as_pin_mut()
+            .unwrap()
+            .reset(deadline, false);
     }
 
-    fn reset_inner(self: Pin<&mut Self>, deadline: Instant) {
-        let mut me = self.project();
-        me.entry.as_mut().reset(deadline, true);
+    fn reset_inner(mut self: Pin<&mut Self>, deadline: Instant) {
+        self.as_mut().lazy_init_timer_entry(deadline);
+        self.as_mut()
+            .project()
+            .entry
+            .as_pin_mut()
+            .unwrap()
+            .reset(deadline, true);
 
         #[cfg(all(tokio_unstable, feature = "tracing"))]
         {
+            let me = self.as_mut().project();
             let _resource_enter = me.inner.ctx.resource_span.enter();
             me.inner.ctx.async_op_span =
                 tracing::trace_span!("runtime.resource.async_op", source = "Sleep::reset");
@@ -382,8 +406,15 @@ impl Sleep {
                 tracing::trace_span!("runtime.resource.async_op.poll");
 
             let duration = {
-                let clock = me.entry.clock();
-                let time_source = me.entry.driver().time_source();
+                let clock = self.as_mut().project().entry.as_pin_mut().unwrap().clock();
+                let time_source = self
+                    .as_mut()
+                    .project()
+                    .entry
+                    .as_pin_mut()
+                    .unwrap()
+                    .driver()
+                    .time_source();
                 let now = time_source.now(clock);
                 let deadline_tick = time_source.deadline_to_tick(deadline);
                 deadline_tick.saturating_sub(now)
@@ -398,10 +429,15 @@ impl Sleep {
         }
     }
 
-    fn poll_elapsed(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Result<(), Error>> {
-        let me = self.project();
-
+    fn poll_elapsed(
+        mut self: Pin<&mut Self>,
+        cx: &mut task::Context<'_>,
+    ) -> Poll<Result<(), Error>> {
         ready!(crate::trace::trace_leaf(cx));
+        let deadline = self.deadline;
+        self.as_mut().lazy_init_timer_entry(deadline);
+
+        let me = self.project();
 
         // Keep track of task budget
         #[cfg(all(tokio_unstable, feature = "tracing"))]
@@ -412,18 +448,34 @@ impl Sleep {
 
         #[cfg(any(not(tokio_unstable), not(feature = "tracing")))]
         let coop = ready!(crate::runtime::coop::poll_proceed(cx));
-
-        let result = me.entry.poll_elapsed(cx).map(move |r| {
-            coop.made_progress();
-            r
-        });
+        // Safety: we have just assigned it a value of `Some`.
+        let result = me
+            .entry
+            .as_pin_mut()
+            .unwrap()
+            .poll_elapsed(cx)
+            .map(move |r| {
+                coop.made_progress();
+                r
+            });
 
         #[cfg(all(tokio_unstable, feature = "tracing"))]
         return trace_poll_op!("poll_elapsed", result);
 
         #[cfg(any(not(tokio_unstable), not(feature = "tracing")))]
         return result;
     }
+
+    // This lazy initiation is for performance purposes,
+    // it can avoid the unnecessary creation and drop of `TimerEntry`.
+    fn lazy_init_timer_entry(self: Pin<&mut Self>, deadline: Instant) {
+        let mut me = self.project();
+        *me.deadline = deadline;
+        if me.entry.is_none() {
+            let entry = TimerEntry::new(me.handle, deadline);
+            me.entry.as_mut().set(Some(entry));
+        }
+    }
 }
 
 impl Future for Sleep {

tokio/src/time/timeout.rs

@@ -6,7 +6,8 @@
 
 use crate::{
     runtime::coop,
-    time::{error::Elapsed, sleep_until, Duration, Instant, Sleep},
+    runtime::scheduler,
+    time::{error::Elapsed, Duration, Instant, Sleep},
     util::trace,
 };
 
@@ -87,14 +88,7 @@ pub fn timeout<F>(duration: Duration, future: F) -> Timeout<F>
 where
     F: Future,
 {
-    let location = trace::caller_location();
-
-    let deadline = Instant::now().checked_add(duration);
-    let delay = match deadline {
-        Some(deadline) => Sleep::new_timeout(deadline, location),
-        None => Sleep::far_future(location),
-    };
-    Timeout::new_with_delay(future, delay)
+    Timeout::new_with_delay(future, Instant::now().checked_add(duration))
 }
 
 /// Requires a `Future` to complete before the specified instant in time.
@@ -146,7 +140,11 @@ pub fn timeout_at<F>(deadline: Instant, future: F) -> Timeout<F>
 where
     F: Future,
 {
-    let delay = sleep_until(deadline);
+    let handle = scheduler::Handle::current();
+    // Panic if the time driver is not enabled.
+    let _ = handle.driver().time();
+
+    let delay = Sleep::new_timeout_with_handle(deadline, trace::caller_location(), handle);
 
     Timeout {
         value: future,
@@ -167,7 +165,19 @@ pin_project! {
 }
 
 impl<T> Timeout<T> {
-    pub(crate) fn new_with_delay(value: T, delay: Sleep) -> Timeout<T> {
+    #[track_caller]
+    pub(crate) fn new_with_delay(value: T, deadline: Option<Instant>) -> Timeout<T> {
+        let handle = scheduler::Handle::current();
+        // Panic if the time driver is not enabled.
+        let _ = handle.driver().time();
+
+        let deadline = match deadline {
+            Some(deadline) => deadline,
+            None => Instant::far_future(),
+        };
+
+        let delay = Sleep::new_timeout_with_handle(deadline, trace::caller_location(), handle);
+
         Timeout { value, delay }
     }