diff --git a/tokio/src/runtime/time/entry.rs b/tokio/src/runtime/time/entry.rs index 0998b5301..f6f56e277 100644 --- a/tokio/src/runtime/time/entry.rs +++ b/tokio/src/runtime/time/entry.rs @@ -75,7 +75,7 @@ const STATE_MIN_VALUE: u64 = STATE_PENDING_FIRE; /// The largest safe integer to use for ticks. /// /// This value should be updated if any other signal values are added above. -pub(super) const MAX_SAFE_MILLIS_DURATION: u64 = u64::MAX - 2; +pub(super) const MAX_SAFE_MILLIS_DURATION: u64 = STATE_MIN_VALUE - 1; /// This structure holds the current shared state of the timer - its scheduled /// time (if registered), or otherwise the result of the timer completing, as @@ -187,18 +187,14 @@ impl StateCell { break Err(cur_state); } - match self.state.compare_exchange( + match self.state.compare_exchange_weak( cur_state, STATE_PENDING_FIRE, Ordering::AcqRel, Ordering::Acquire, ) { - Ok(_) => { - break Ok(()); - } - Err(actual_state) => { - cur_state = actual_state; - } + Ok(_) => break Ok(()), + Err(actual_state) => cur_state = actual_state, } } } @@ -266,12 +262,8 @@ impl StateCell { Ordering::AcqRel, Ordering::Acquire, ) { - Ok(_) => { - return Ok(()); - } - Err(true_prior) => { - prior = true_prior; - } + Ok(_) => return Ok(()), + Err(true_prior) => prior = true_prior, } } } @@ -564,9 +556,7 @@ impl TimerEntry { self.as_mut().reset(deadline, true); } - let this = unsafe { self.get_unchecked_mut() }; - - this.inner().state.poll(cx.waker()) + self.inner().state.poll(cx.waker()) } pub(crate) fn driver(&self) -> &super::Handle { diff --git a/tokio/src/runtime/time/mod.rs b/tokio/src/runtime/time/mod.rs index a30393a02..8cd51c5cb 100644 --- a/tokio/src/runtime/time/mod.rs +++ b/tokio/src/runtime/time/mod.rs @@ -23,9 +23,10 @@ use crate::loom::sync::Mutex; use crate::runtime::driver::{self, IoHandle, IoStack}; use crate::time::error::Error; use crate::time::{Clock, Duration}; +use crate::util::WakeList; use std::fmt; -use std::{num::NonZeroU64, ptr::NonNull, task::Waker}; +use std::{num::NonZeroU64, ptr::NonNull}; /// Time implementation that drives [`Sleep`][sleep], [`Interval`][interval], and [`Timeout`][timeout]. /// @@ -253,8 +254,7 @@ impl Handle { } pub(self) fn process_at_time(&self, mut now: u64) { - let mut waker_list: [Option; 32] = Default::default(); - let mut waker_idx = 0; + let mut waker_list = WakeList::new(); let mut lock = self.inner.lock(); @@ -273,19 +273,13 @@ impl Handle { // SAFETY: We hold the driver lock, and just removed the entry from any linked lists. if let Some(waker) = unsafe { entry.fire(Ok(())) } { - waker_list[waker_idx] = Some(waker); + waker_list.push(waker); - waker_idx += 1; - - if waker_idx == waker_list.len() { + if !waker_list.can_push() { // Wake a batch of wakers. To avoid deadlock, we must do this with the lock temporarily dropped. drop(lock); - for waker in waker_list.iter_mut() { - waker.take().unwrap().wake(); - } - - waker_idx = 0; + waker_list.wake_all(); lock = self.inner.lock(); } @@ -299,9 +293,7 @@ impl Handle { drop(lock); - for waker in &mut waker_list[0..waker_idx] { - waker.take().unwrap().wake(); - } + waker_list.wake_all(); } /// Removes a registered timer from the driver. diff --git a/tokio/src/runtime/time/source.rs b/tokio/src/runtime/time/source.rs index c709dc538..e371c207c 100644 --- a/tokio/src/runtime/time/source.rs +++ b/tokio/src/runtime/time/source.rs @@ -21,9 +21,7 @@ impl TimeSource { 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 dur: Duration = t.saturating_duration_since(self.start_time); let ms = dur.as_millis(); ms.try_into().unwrap_or(MAX_SAFE_MILLIS_DURATION) diff --git a/tokio/src/runtime/time/wheel/level.rs b/tokio/src/runtime/time/wheel/level.rs index a828c0067..d31eaf468 100644 --- a/tokio/src/runtime/time/wheel/level.rs +++ b/tokio/src/runtime/time/wheel/level.rs @@ -1,6 +1,6 @@ use crate::runtime::time::{EntryList, TimerHandle, TimerShared}; -use std::{fmt, ptr::NonNull}; +use std::{array, fmt, ptr::NonNull}; /// Wheel for a single level in the timer. This wheel contains 64 slots. pub(crate) struct Level { @@ -39,89 +39,10 @@ const LEVEL_MULT: usize = 64; impl Level { pub(crate) fn new(level: usize) -> Level { - // A value has to be Copy in order to use syntax like: - // let stack = Stack::default(); - // ... - // slots: [stack; 64], - // - // Alternatively, since Stack is Default one can - // use syntax like: - // let slots: [Stack; 64] = Default::default(); - // - // However, that is only supported for arrays of size - // 32 or fewer. So in our case we have to explicitly - // invoke the constructor for each array element. - let ctor = EntryList::default; - Level { level, occupied: 0, - slot: [ - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ctor(), - ], + slot: array::from_fn(|_| EntryList::default()), } } @@ -130,10 +51,7 @@ impl Level { pub(crate) fn next_expiration(&self, now: u64) -> Option { // Use the `occupied` bit field to get the index of the next slot that // needs to be processed. - let slot = match self.next_occupied_slot(now) { - Some(slot) => slot, - None => return None, - }; + let slot = self.next_occupied_slot(now)?; // From the slot index, calculate the `Instant` at which it needs to be // processed. This value *must* be in the future with respect to `now`. @@ -196,7 +114,7 @@ impl Level { let now_slot = (now / slot_range(self.level)) as usize; let occupied = self.occupied.rotate_right(now_slot as u32); let zeros = occupied.trailing_zeros() as usize; - let slot = (zeros + now_slot) % 64; + let slot = (zeros + now_slot) % LEVEL_MULT; Some(slot) } diff --git a/tokio/src/runtime/time/wheel/mod.rs b/tokio/src/runtime/time/wheel/mod.rs index bf13b7b24..f2b422851 100644 --- a/tokio/src/runtime/time/wheel/mod.rs +++ b/tokio/src/runtime/time/wheel/mod.rs @@ -5,7 +5,7 @@ mod level; pub(crate) use self::level::Expiration; use self::level::Level; -use std::ptr::NonNull; +use std::{array, ptr::NonNull}; use super::EntryList; @@ -35,7 +35,7 @@ pub(crate) struct Wheel { /// * ~ 4 min slots / ~ 4 hr range /// * ~ 4 hr slots / ~ 12 day range /// * ~ 12 day slots / ~ 2 yr range - levels: Vec, + levels: Box<[Level; NUM_LEVELS]>, /// Entries queued for firing pending: EntryList, @@ -52,11 +52,9 @@ pub(super) const MAX_DURATION: u64 = (1 << (6 * NUM_LEVELS)) - 1; impl Wheel { /// Creates a new timing wheel. pub(crate) fn new() -> Wheel { - let levels = (0..NUM_LEVELS).map(Level::new).collect(); - Wheel { elapsed: 0, - levels, + levels: Box::new(array::from_fn(Level::new)), pending: EntryList::new(), } } @@ -130,7 +128,6 @@ impl Wheel { ); let level = self.level_for(when); - self.levels[level].remove_entry(item); } } @@ -180,11 +177,11 @@ impl Wheel { } // Check all levels - for level in 0..NUM_LEVELS { - if let Some(expiration) = self.levels[level].next_expiration(self.elapsed) { + for (level_num, level) in self.levels.iter().enumerate() { + if let Some(expiration) = level.next_expiration(self.elapsed) { // There cannot be any expirations at a higher level that happen // before this one. - debug_assert!(self.no_expirations_before(level + 1, expiration.deadline)); + debug_assert!(self.no_expirations_before(level_num + 1, expiration.deadline)); return Some(expiration); } @@ -203,8 +200,8 @@ impl Wheel { fn no_expirations_before(&self, start_level: usize, before: u64) -> bool { let mut res = true; - for l2 in start_level..NUM_LEVELS { - if let Some(e2) = self.levels[l2].next_expiration(self.elapsed) { + for level in &self.levels[start_level..] { + if let Some(e2) = level.next_expiration(self.elapsed) { if e2.deadline < before { res = false; } @@ -267,7 +264,6 @@ impl Wheel { } /// Obtains the list of entries that need processing for the given expiration. - /// fn take_entries(&mut self, expiration: &Expiration) -> EntryList { self.levels[expiration.level].take_slot(expiration.slot) } @@ -292,7 +288,7 @@ fn level_for(elapsed: u64, when: u64) -> usize { let leading_zeros = masked.leading_zeros() as usize; let significant = 63 - leading_zeros; - significant / 6 + significant / NUM_LEVELS } #[cfg(all(test, not(loom)))] diff --git a/tokio/src/util/mod.rs b/tokio/src/util/mod.rs index abdb70406..d821ec897 100644 --- a/tokio/src/util/mod.rs +++ b/tokio/src/util/mod.rs @@ -19,6 +19,8 @@ pub(crate) mod once_cell; // rt and signal use `Notify`, which requires `WakeList`. feature = "rt", feature = "signal", + // time driver uses `WakeList` in `Handle::process_at_time`. + feature = "time", ))] mod wake_list; #[cfg(any( @@ -28,6 +30,7 @@ mod wake_list; feature = "fs", feature = "rt", feature = "signal", + feature = "time", ))] pub(crate) use wake_list::WakeList;