From 2ad347465e8b634b785e7637d899a14da8a0fba3 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Wed, 7 Sep 2022 13:24:02 -0700 Subject: [PATCH] rt: minor time driver refactors (#4989) This patch makes some minor refactors. It renames `ClockTime` to `TimeSource` since that is how all variables refer to it. It also moves the type into a new file. Finally, it moves the `unpark` handle out of the mutex as it does not need to be there. Note, the call to `unpark` is still called while the mutex is held, so there is no functional change. Moving it out of the mutex is in preparation for moving the unpark handle completely out of the time driver. --- tokio/src/runtime/time/handle.rs | 8 ++-- tokio/src/runtime/time/mod.rs | 63 ++++++----------------------- tokio/src/runtime/time/source.rs | 42 +++++++++++++++++++ tokio/src/runtime/time/tests/mod.rs | 12 +++--- tokio/src/time/sleep.rs | 4 +- 5 files changed, 67 insertions(+), 62 deletions(-) create mode 100644 tokio/src/runtime/time/source.rs diff --git a/tokio/src/runtime/time/handle.rs b/tokio/src/runtime/time/handle.rs index f68786747..ad81867bd 100644 --- a/tokio/src/runtime/time/handle.rs +++ b/tokio/src/runtime/time/handle.rs @@ -1,12 +1,12 @@ use crate::loom::sync::Arc; -use crate::runtime::time::ClockTime; +use crate::runtime::time::TimeSource; use std::fmt; /// Handle to time driver instance. #[derive(Clone)] pub(crate) struct Handle { - time_source: ClockTime, - inner: Arc, + time_source: TimeSource, + pub(super) inner: Arc, } impl Handle { @@ -17,7 +17,7 @@ impl Handle { } /// Returns the time source associated with this handle. - pub(crate) fn time_source(&self) -> &ClockTime { + pub(crate) fn time_source(&self) -> &TimeSource { &self.time_source } diff --git a/tokio/src/runtime/time/mod.rs b/tokio/src/runtime/time/mod.rs index b5cad2ee0..0d898470e 100644 --- a/tokio/src/runtime/time/mod.rs +++ b/tokio/src/runtime/time/mod.rs @@ -13,15 +13,17 @@ use entry::{EntryList, TimerHandle, TimerShared}; mod handle; pub(crate) use self::handle::Handle; +mod source; +pub(crate) use source::TimeSource; + mod wheel; use crate::loom::sync::atomic::{AtomicBool, Ordering}; use crate::loom::sync::{Arc, Mutex}; use crate::park::{Park, Unpark}; use crate::time::error::Error; -use crate::time::{Clock, Duration, Instant}; +use crate::time::{Clock, Duration}; -use std::convert::TryInto; use std::fmt; use std::{num::NonZeroU64, ptr::NonNull, task::Waker}; @@ -83,7 +85,7 @@ use std::{num::NonZeroU64, ptr::NonNull, task::Waker}; #[derive(Debug)] pub(crate) struct Driver { /// Timing backend in use. - time_source: ClockTime, + time_source: TimeSource, /// Shared state. handle: Handle, @@ -101,45 +103,6 @@ pub(crate) struct Driver { did_wake: Arc, } -/// A structure which handles conversion from Instants to u64 timestamps. -#[derive(Debug, Clone)] -pub(crate) struct ClockTime { - clock: crate::time::Clock, - start_time: Instant, -} - -impl ClockTime { - pub(self) fn new(clock: Clock) -> Self { - Self { - start_time: clock.now(), - clock, - } - } - - pub(crate) fn deadline_to_tick(&self, t: Instant) -> u64 { - // Round up to the end of a ms - self.instant_to_tick(t + Duration::from_nanos(999_999)) - } - - pub(self) fn instant_to_tick(&self, t: Instant) -> u64 { - // round up - let dur: Duration = t - .checked_duration_since(self.start_time) - .unwrap_or_else(|| Duration::from_secs(0)); - let ms = dur.as_millis(); - - ms.try_into().unwrap_or(u64::MAX) - } - - pub(self) fn tick_to_duration(&self, t: u64) -> Duration { - Duration::from_millis(t) - } - - pub(crate) fn now(&self) -> u64 { - self.instant_to_tick(self.clock.now()) - } -} - /// Timer state shared between `Driver`, `Handle`, and `Registration`. struct Inner { // The state is split like this so `Handle` can access `is_shutdown` without locking the mutex @@ -147,12 +110,15 @@ struct Inner { /// True if the driver is being shutdown. pub(super) is_shutdown: AtomicBool, + + /// Unparker that can be used to wake the time driver. + unpark: Box, } /// Time state shared which must be protected by a `Mutex` struct InnerState { /// Timing backend in use. - time_source: ClockTime, + time_source: TimeSource, /// The last published timer `elapsed` value. elapsed: u64, @@ -162,9 +128,6 @@ struct InnerState { /// Timer wheel. wheel: wheel::Wheel, - - /// Unparker that can be used to wake the time driver. - unpark: Box, } // ===== impl Driver ===== @@ -178,7 +141,7 @@ where /// /// Specifying the source of time is useful when testing. pub(crate) fn new(park: P, clock: Clock) -> Driver

{ - let time_source = ClockTime::new(clock); + let time_source = TimeSource::new(clock); let inner = Inner::new(time_source.clone(), Box::new(park.unpark())); @@ -397,7 +360,7 @@ impl Handle { .map(|next_wake| when < next_wake.get()) .unwrap_or(true) { - lock.unpark.unpark(); + self.inner.unpark.unpark(); } None @@ -493,15 +456,15 @@ impl Unpark for TimerUnpark

{ // ===== impl Inner ===== impl Inner { - pub(self) fn new(time_source: ClockTime, unpark: Box) -> Self { + pub(self) fn new(time_source: TimeSource, unpark: Box) -> Self { Inner { state: Mutex::new(InnerState { time_source, elapsed: 0, next_wake: None, - unpark, wheel: wheel::Wheel::new(), }), + unpark, is_shutdown: AtomicBool::new(false), } } diff --git a/tokio/src/runtime/time/source.rs b/tokio/src/runtime/time/source.rs new file mode 100644 index 000000000..1cdb86891 --- /dev/null +++ b/tokio/src/runtime/time/source.rs @@ -0,0 +1,42 @@ +use crate::time::{Clock, Duration, Instant}; + +use std::convert::TryInto; + +/// A structure which handles conversion from Instants to u64 timestamps. +#[derive(Debug, Clone)] +pub(crate) struct TimeSource { + pub(crate) clock: Clock, + start_time: Instant, +} + +impl TimeSource { + pub(crate) fn new(clock: Clock) -> Self { + Self { + start_time: clock.now(), + clock, + } + } + + pub(crate) fn deadline_to_tick(&self, t: Instant) -> u64 { + // Round up to the end of a ms + self.instant_to_tick(t + Duration::from_nanos(999_999)) + } + + pub(crate) fn instant_to_tick(&self, t: Instant) -> u64 { + // round up + let dur: Duration = t + .checked_duration_since(self.start_time) + .unwrap_or_else(|| Duration::from_secs(0)); + let ms = dur.as_millis(); + + ms.try_into().unwrap_or(u64::MAX) + } + + pub(crate) fn tick_to_duration(&self, t: u64) -> Duration { + Duration::from_millis(t) + } + + pub(crate) fn now(&self) -> u64 { + self.instant_to_tick(self.clock.now()) + } +} diff --git a/tokio/src/runtime/time/tests/mod.rs b/tokio/src/runtime/time/tests/mod.rs index 541e78256..ff69c8407 100644 --- a/tokio/src/runtime/time/tests/mod.rs +++ b/tokio/src/runtime/time/tests/mod.rs @@ -49,7 +49,7 @@ fn model(f: impl Fn() + Send + Sync + 'static) { fn single_timer() { model(|| { let clock = crate::time::Clock::new(true, false); - let time_source = super::ClockTime::new(clock.clone()); + let time_source = super::TimeSource::new(clock.clone()); let inner = super::Inner::new(time_source.clone(), MockUnpark::mock()); let handle = Handle::new(Arc::new(inner)); @@ -80,7 +80,7 @@ fn single_timer() { fn drop_timer() { model(|| { let clock = crate::time::Clock::new(true, false); - let time_source = super::ClockTime::new(clock.clone()); + let time_source = super::TimeSource::new(clock.clone()); let inner = super::Inner::new(time_source.clone(), MockUnpark::mock()); let handle = Handle::new(Arc::new(inner)); @@ -111,7 +111,7 @@ fn drop_timer() { fn change_waker() { model(|| { let clock = crate::time::Clock::new(true, false); - let time_source = super::ClockTime::new(clock.clone()); + let time_source = super::TimeSource::new(clock.clone()); let inner = super::Inner::new(time_source.clone(), MockUnpark::mock()); let handle = Handle::new(Arc::new(inner)); @@ -146,7 +146,7 @@ fn reset_future() { let finished_early = Arc::new(AtomicBool::new(false)); let clock = crate::time::Clock::new(true, false); - let time_source = super::ClockTime::new(clock.clone()); + let time_source = super::TimeSource::new(clock.clone()); let inner = super::Inner::new(time_source.clone(), MockUnpark::mock()); let handle = Handle::new(Arc::new(inner)); @@ -204,7 +204,7 @@ fn poll_process_levels() { let clock = crate::time::Clock::new(true, false); clock.pause(); - let time_source = super::ClockTime::new(clock.clone()); + let time_source = super::TimeSource::new(clock.clone()); let inner = super::Inner::new(time_source, MockUnpark::mock()); let handle = Handle::new(Arc::new(inner)); @@ -245,7 +245,7 @@ fn poll_process_levels_targeted() { let clock = crate::time::Clock::new(true, false); clock.pause(); - let time_source = super::ClockTime::new(clock.clone()); + let time_source = super::TimeSource::new(clock.clone()); let inner = super::Inner::new(time_source, MockUnpark::mock()); let handle = Handle::new(Arc::new(inner)); diff --git a/tokio/src/time/sleep.rs b/tokio/src/time/sleep.rs index 29db2fbee..378ca6e88 100644 --- a/tokio/src/time/sleep.rs +++ b/tokio/src/time/sleep.rs @@ -1,5 +1,5 @@ #[cfg(all(tokio_unstable, feature = "tracing"))] -use crate::runtime::time::ClockTime; +use crate::runtime::time::TimeSource; use crate::runtime::time::{Handle, TimerEntry}; use crate::time::{error::Error, Duration, Instant}; use crate::util::trace; @@ -239,7 +239,7 @@ cfg_trace! { struct Inner { deadline: Instant, ctx: trace::AsyncOpTracingCtx, - time_source: ClockTime, + time_source: TimeSource, } }