From 7bf2df3ad45591faa9cfd217a7bf565fe785e003 Mon Sep 17 00:00:00 2001 From: Lars Kroll Date: Sat, 11 Apr 2026 12:15:55 +0200 Subject: [PATCH 1/3] Add manual timer and shared queue core --- README.md | 7 +- rustfmt.toml | 2 +- src/lib.rs | 19 +- src/manual_timer.rs | 450 ++++++++++++++++++++++++++++++++++++++++++++ src/queue_timer.rs | 350 ++++++++++++++++++++++++++++++++++ src/thread_timer.rs | 382 +++++++++++-------------------------- 6 files changed, 931 insertions(+), 279 deletions(-) create mode 100644 src/manual_timer.rs create mode 100644 src/queue_timer.rs diff --git a/README.md b/README.md index 8c8154b..707fbd9 100644 --- a/README.md +++ b/README.md @@ -27,9 +27,9 @@ This crate provides two variant implementations of this four level wheel structu - The `wheels::cancellable::QuadWheelWithOverflow` additionally supports the cancellation of outstanding timers before they expire. In order to do so, however, it requires the generic timer entry type to provide a unique identifier field. It also uses `std::rc::Rc` internally to avoid double storing the actual entry, which makes it (potentially) unsuitable for situations where the timer must be able to move between threads. ### 3 – High Level APIs -This crate also provides two high levels APIs that can either be used directly or can be seen as examples of how to use the lower level APIs in an application. +This crate also provides three high-level APIs that can either be used directly or can be seen as examples of how to use the lower level APIs in an application. -Both higher level APIs also offer built-in support for periodically repeating timers, in addition to the normal timers which are scheduled once and discarded once expired. +All higher-level APIs also offer built-in support for periodically repeating timers, in addition to the normal timers which are scheduled once and discarded once expired. #### Simulation Timer The `simulation` module provides an implementation for an event timer used to drive a discrete event simulation. @@ -39,6 +39,9 @@ Its particular feature is that it can skip quickly through periods where no even The `thread_timer` module provides a timer for real-time event scheduling with millisecond accuracy. It runs on its own dedicated thread and uses a shareable handle called a `TimerRef` for communication with other threads. +#### Manual Timer +The `manual_timer` module provides the same queue-based scheduling API as the thread timer, but advances only when the caller explicitly steps time forward. + ## Documentation For reference and examples check the [API Docs](https://docs.rs/hierarchical_hash_wheel_timer). diff --git a/rustfmt.toml b/rustfmt.toml index 0b35b63..2d6a4d0 100644 --- a/rustfmt.toml +++ b/rustfmt.toml @@ -43,7 +43,7 @@ use_field_init_shorthand = false force_explicit_abi = true condense_wildcard_suffixes = false color = "Auto" -required_version = "1.8.0" +required_version = "1.9.0" unstable_features = false disable_all_formatting = false skip_children = false diff --git a/src/lib.rs b/src/lib.rs index 39a614c..ff10117 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -32,11 +32,13 @@ //! be able to move threads (since Rc](std::rc::Rc) is not `Send`). //! //! # 3 – High Level APIs -//! This crate also provides two high levels APIs that can either be used directly or can be seen as examples -//! of how to use the lower level APIs in an application. +//! This crate also provides three high-level APIs that can either be used +//! directly or can be seen as examples of how to use the lower level APIs in +//! an application. //! -//! Bother higher level APIs also offer built-in support for periodically repeating timers, in addition to the normal timers -//! which are schedulled once and discarded once expired. +//! All higher-level APIs also offer built-in support for periodically +//! repeating timers, in addition to the normal timers which are scheduled once +//! and discarded once expired. //! //! ## Simulation Timer //! The [simulation](simulation) module provides an implementation for an event timer used to drive a discrete event simulation. @@ -46,6 +48,11 @@ //! ## Thread Timer //! The [thread_timer](thread_timer) module provides a timer for real-time event schedulling with millisecond accuracy. //! It runs on its own dedicated thread and uses a shareable handle called a `TimerRef` for communication with other threads. +//! +//! ## Manual Timer +//! The [manual_timer](manual_timer) module provides the same queue-based +//! scheduling API as the thread timer, but advances only when the caller +//! explicitly steps time forward. #![deny(missing_docs)] @@ -58,6 +65,10 @@ use wheels::{cancellable::CancellableTimerEntry, TimerEntryWithDelay}; mod timers; pub use self::timers::*; +#[cfg(feature = "thread-timer")] +pub mod manual_timer; +#[cfg(feature = "thread-timer")] +mod queue_timer; #[cfg(feature = "thread-timer")] pub mod thread_timer; diff --git a/src/manual_timer.rs b/src/manual_timer.rs new file mode 100644 index 0000000..51ffe1a --- /dev/null +++ b/src/manual_timer.rs @@ -0,0 +1,450 @@ +//! This module provides a manually-driven timer with the same queue-based +//! scheduling handle as the threaded timer. +//! +//! The timer does not own a worker thread. Instead, tests or simulations drive +//! time forward explicitly through [`ManualTimer::advance_by`] or +//! [`ManualTimer::advance_to_next`]. + +use super::*; + +use crate::{ + queue_timer::{ClockRef, QueueTimerControl, QueueTimerCore, TimerMsg, TimerRef}, + wheels::Skip, +}; +use crossbeam_channel as channel; +use std::{ + fmt, + hash::Hash, + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, + Mutex, + }, + time::{Duration, Instant}, +}; + +/// A timer implementation advanced explicitly by the caller. +/// +/// This timer is useful for tests that need deterministic control over timeout +/// delivery without depending on wall-clock time or an extra scheduling thread. +pub struct ManualTimer +where + I: Hash + Clone + Eq + fmt::Debug, + O: OneshotState + fmt::Debug, + P: PeriodicState + fmt::Debug, +{ + inner: Arc>>, + work_queue: channel::Sender>, + base: Instant, + elapsed_millis: Arc, +} + +impl ManualTimer +where + I: Hash + Clone + Eq + fmt::Debug, + O: OneshotState + fmt::Debug, + P: PeriodicState + fmt::Debug, +{ + /// Create a new manual timer. + pub fn new() -> Self { + let (work_queue, receiver) = channel::unbounded(); + let elapsed_millis = Arc::new(AtomicU64::new(0)); + let inner = ManualTimerInner::new(receiver, Arc::clone(&elapsed_millis)); + Self { + inner: Arc::new(Mutex::new(inner)), + work_queue, + base: Instant::now(), + elapsed_millis, + } + } + + /// Returns a shareable scheduling handle for this timer. + pub fn timer_ref(&self) -> TimerRef { + TimerRef::new( + self.work_queue.clone(), + ClockRef::manual(self.base, Arc::clone(&self.elapsed_millis)), + ) + } + + /// Advances the timer by the supplied duration. + /// + /// The timer has millisecond resolution, matching the threaded timer. + /// Sub-millisecond advances are accumulated and take effect once they sum + /// to at least one millisecond. + pub fn advance_by(&self, duration: Duration) { + let mut inner = self.inner.lock().expect("manual timer mutex poisoned"); + inner.advance_by(duration); + } + + /// Advances the timer just far enough to trigger the next due timeout. + /// + /// Returns `false` when there is currently no outstanding timeout to fire. + pub fn advance_to_next(&self) -> bool { + let mut inner = self.inner.lock().expect("manual timer mutex poisoned"); + inner.advance_to_next() + } + + /// Returns the current manually-driven logical time. + /// + /// This is rounded down to the manual timer's millisecond resolution so it + /// agrees with the deadlines the wheel can actually execute. + pub fn current_time(&self) -> Duration { + Duration::from_millis(self.elapsed_millis.load(Ordering::Acquire)) + } + + /// Returns whether the timer has neither queued commands nor scheduled + /// deadlines left to execute. + pub fn is_idle(&self) -> bool { + let mut inner = self.inner.lock().expect("manual timer mutex poisoned"); + inner.prepare(); + !inner.running || inner.core.is_idle() + } + + /// Stops the timer so future advances no longer execute queued work. + pub fn shutdown(&self) { + let mut inner = self.inner.lock().expect("manual timer mutex poisoned"); + inner.running = false; + } +} + +impl Clone for ManualTimer +where + I: Hash + Clone + Eq + fmt::Debug, + O: OneshotState + fmt::Debug, + P: PeriodicState + fmt::Debug, +{ + fn clone(&self) -> Self { + Self { + inner: Arc::clone(&self.inner), + work_queue: self.work_queue.clone(), + base: self.base, + elapsed_millis: Arc::clone(&self.elapsed_millis), + } + } +} + +impl Default for ManualTimer +where + I: Hash + Clone + Eq + fmt::Debug, + O: OneshotState + fmt::Debug, + P: PeriodicState + fmt::Debug, +{ + fn default() -> Self { + Self::new() + } +} + +impl fmt::Debug for ManualTimer +where + I: Hash + Clone + Eq + fmt::Debug, + O: OneshotState + fmt::Debug, + P: PeriodicState + fmt::Debug, +{ + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "") + } +} + +struct ManualTimerInner +where + I: Hash + Clone + Eq + fmt::Debug, + O: OneshotState + fmt::Debug, + P: PeriodicState + fmt::Debug, +{ + core: QueueTimerCore, + /// Shared logical clock visible to non-blocking readers through `TimerRef::now()`. + elapsed_millis: Arc, + /// Carries sub-millisecond advances until they add up to one wheel tick. + sub_millis_remainder: Duration, + /// Counts how many logical milliseconds have already been applied to the wheel. + processed_millis: u64, + running: bool, +} + +// Safety: the queue timer core still uses `Rc`/`Weak` internally through the +// cancellable wheel implementation, but a `ManualTimerInner` is only ever +// accessed while holding the surrounding `Mutex`. The shared scheduling handle +// talks to the timer over a channel and never touches the wheel state +// directly, so moving the manually-driven timer between threads is sound as +// long as the user-provided timer state is itself `Send`. +unsafe impl Send for ManualTimerInner +where + I: Hash + Clone + Eq + fmt::Debug + Send, + O: OneshotState + fmt::Debug + Send, + P: PeriodicState + fmt::Debug + Send, +{ +} + +impl ManualTimerInner +where + I: Hash + Clone + Eq + fmt::Debug, + O: OneshotState + fmt::Debug, + P: PeriodicState + fmt::Debug, +{ + fn new( + work_queue: channel::Receiver>, + elapsed_millis: Arc, + ) -> Self { + Self { + core: QueueTimerCore::new(work_queue), + elapsed_millis, + sub_millis_remainder: Duration::ZERO, + processed_millis: 0, + running: true, + } + } + + fn advance_by(&mut self, duration: Duration) { + if !self.running { + return; + } + + self.prepare(); + let total_advanced = self + .sub_millis_remainder + .checked_add(duration) + .expect("manual timer duration overflow"); + let advanced_millis = + u64::try_from(total_advanced.as_millis()).expect("manual timer duration overflow"); + self.sub_millis_remainder = total_advanced + .checked_sub(Duration::from_millis(advanced_millis)) + .expect("manual timer remainder underflow"); + let target_millis = if advanced_millis == 0 { + self.current_millis() + } else { + self.advance_clock(advanced_millis) + }; + let ticks_to_process = target_millis.saturating_sub(self.processed_millis); + self.advance_ticks(ticks_to_process); + self.prepare(); + } + + fn advance_to_next(&mut self) -> bool { + if !self.running { + return false; + } + + self.prepare(); + loop { + if !self.running { + return false; + } + + match self.core.can_skip() { + Skip::Empty => return false, + Skip::Millis(can_skip) if can_skip > 0 => { + self.skip_millis(u64::from(can_skip)); + self.prepare(); + } + Skip::Millis(_) | Skip::None => { + // Once the next entry sits in the primary wheel, `can_skip` + // can only promise that something may fire on the next tick, + // not which exact future millisecond will do it. + if self.advance_one_tick() > 0 { + return true; + } + self.prepare(); + } + } + } + } + + fn prepare(&mut self) { + if matches!(self.core.drain_messages(), QueueTimerControl::Stop) { + self.running = false; + } + } + + fn current_millis(&self) -> u64 { + self.elapsed_millis.load(Ordering::Acquire) + } + + fn advance_clock(&self, advanced_millis: u64) -> u64 { + let next_millis = self + .current_millis() + .checked_add(advanced_millis) + .expect("manual timer duration overflow"); + self.elapsed_millis.store(next_millis, Ordering::Release); + next_millis + } + + fn advance_ticks(&mut self, mut remaining_ticks: u64) { + while self.running && remaining_ticks > 0 { + self.prepare(); + if !self.running { + return; + } + + match self.core.can_skip() { + Skip::Empty => { + self.processed_millis += remaining_ticks; + return; + } + Skip::Millis(can_skip) if can_skip > 0 => { + let skipped_ticks = remaining_ticks.min(u64::from(can_skip)); + self.core + .skip(u32::try_from(skipped_ticks).expect("skip count fits in u32")); + self.processed_millis += skipped_ticks; + remaining_ticks -= skipped_ticks; + } + Skip::Millis(_) | Skip::None => { + self.core.tick(); + self.processed_millis += 1; + remaining_ticks -= 1; + } + } + } + } + + fn skip_millis(&mut self, skip_millis: u64) { + if skip_millis == 0 { + return; + } + + self.advance_clock(skip_millis); + self.core + .skip(u32::try_from(skip_millis).expect("skip count fits in u32")); + self.processed_millis += skip_millis; + } + + fn advance_one_tick(&mut self) -> usize { + debug_assert!( + self.sub_millis_remainder < Duration::from_millis(1), + "sub-millisecond remainder must stay below one millisecond" + ); + self.sub_millis_remainder = Duration::ZERO; + self.advance_clock(1); + self.processed_millis += 1; + self.core.tick() + } +} + +#[cfg(feature = "uuid-extras")] +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{Arc, Mutex}; + use uuid::Uuid; + + #[test] + fn oneshot_triggers_only_after_manual_advance() { + let timer = ManualTimer::for_uuid_closures(); + let mut timer_ref = timer.timer_ref(); + let fired = Arc::new(Mutex::new(false)); + let fired_clone = Arc::clone(&fired); + + timer_ref.schedule_action_once(Uuid::new_v4(), Duration::from_millis(5), move |_| { + let mut guard = fired_clone.lock().unwrap(); + *guard = true; + }); + + timer.advance_by(Duration::from_millis(4)); + assert!(!*fired.lock().unwrap()); + + timer.advance_by(Duration::from_millis(1)); + assert!(*fired.lock().unwrap()); + } + + #[test] + fn cancel_prevents_future_trigger() { + let timer = ManualTimer::for_uuid_closures(); + let mut timer_ref = timer.timer_ref(); + let fired = Arc::new(Mutex::new(false)); + let fired_clone = Arc::clone(&fired); + let id = Uuid::new_v4(); + + timer_ref.schedule_action_once(id, Duration::from_millis(5), move |_| { + let mut guard = fired_clone.lock().unwrap(); + *guard = true; + }); + timer_ref.cancel(&id); + + timer.advance_by(Duration::from_millis(5)); + assert!(!*fired.lock().unwrap()); + } + + #[test] + fn advance_to_next_triggers_periodic_once_per_call() { + let timer = ManualTimer::for_uuid_closures(); + let mut timer_ref = timer.timer_ref(); + let fired = Arc::new(Mutex::new(Vec::new())); + let fired_clone = Arc::clone(&fired); + + timer_ref.schedule_action_periodic( + Uuid::new_v4(), + Duration::from_millis(5), + Duration::from_millis(7), + move |_| { + fired_clone.lock().unwrap().push(()); + if fired_clone.lock().unwrap().len() < 3 { + TimerReturn::Reschedule(()) + } else { + TimerReturn::Cancel + } + }, + ); + + assert!(timer.advance_to_next()); + assert_eq!(fired.lock().unwrap().len(), 1); + assert!(timer.advance_to_next()); + assert_eq!(fired.lock().unwrap().len(), 2); + assert!(timer.advance_to_next()); + assert_eq!(fired.lock().unwrap().len(), 3); + assert!(!timer.advance_to_next()); + } + + #[test] + fn timer_ref_now_tracks_manual_advance_at_millisecond_resolution() { + let timer = ManualTimer::for_uuid_closures(); + let timer_ref = timer.timer_ref(); + let base = timer_ref.now(); + + timer.advance_by(Duration::from_micros(999)); + assert_eq!(timer.current_time(), Duration::ZERO); + assert_eq!(timer_ref.now(), base); + + timer.advance_by(Duration::from_micros(1)); + assert_eq!(timer.current_time(), Duration::from_millis(1)); + assert_eq!( + timer_ref.now().duration_since(base), + Duration::from_millis(1) + ); + } + + #[test] + fn timer_ref_now_inside_handler_is_not_earlier_than_due_time() { + let timer = ManualTimer::for_uuid_closures(); + let mut timer_ref = timer.timer_ref(); + let scheduled_from = timer_ref.now(); + let delay = Duration::from_millis(5); + let handler_timer_ref = timer_ref.clone(); + let observed_now = Arc::new(Mutex::new(None)); + let observed_now_clone = Arc::clone(&observed_now); + + timer_ref.schedule_action_once(Uuid::new_v4(), delay, move |_| { + let now = handler_timer_ref.now(); + *observed_now_clone.lock().unwrap() = Some(now); + }); + + timer.advance_by(Duration::from_millis(20)); + + let observed_now = observed_now + .lock() + .unwrap() + .expect("oneshot callback should record observed time"); + assert!( + observed_now.duration_since(scheduled_from) >= delay, + "handler observed {observed_now:?}, which is earlier than scheduled lower bound {:?}", + scheduled_from + delay + ); + } +} + +#[cfg(feature = "uuid-extras")] +impl ManualTimer, PeriodicClosureState> { + /// Shorthand for creating a manual timer using `Uuid` identifiers and + /// closure state. + pub fn for_uuid_closures() -> Self { + Self::new() + } +} diff --git a/src/queue_timer.rs b/src/queue_timer.rs new file mode 100644 index 0000000..2d6aacf --- /dev/null +++ b/src/queue_timer.rs @@ -0,0 +1,350 @@ +use super::*; + +use crate::wheels::{cancellable::*, *}; +use crossbeam_channel as channel; +use std::{ + fmt, + hash::Hash, + rc::Rc, + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, + }, + time::{Duration, Instant}, +}; + +#[derive(Debug)] +pub(crate) enum TimerMsg +where + I: Hash + Clone + Eq, + O: OneshotState, + P: PeriodicState, +{ + Schedule(TimerEntry), + Cancel(I), + Stop, +} + +/// A shareable scheduling handle for queue-backed timer drivers. +/// +/// This handle is used by both the threaded and manually-driven timers in this crate. +pub struct TimerRef +where + I: Hash + Clone + Eq, + O: OneshotState, + P: PeriodicState, +{ + work_queue: channel::Sender>, + clock: ClockRef, +} + +impl TimerRef +where + I: Hash + Clone + Eq, + O: OneshotState, + P: PeriodicState, +{ + pub(crate) fn new(work_queue: channel::Sender>, clock: ClockRef) -> Self { + Self { work_queue, clock } + } + + /// Returns the current time according to the underlying timer driver. + /// + /// For the threaded timer this is wall-clock time. For the manual timer + /// this advances only when the caller explicitly advances the timer. + /// + /// When called from a timeout handler, the returned instant is guaranteed + /// not to be earlier than the timeout deadline that permitted the handler + /// to run, but it may be later because timeout delivery itself can lag + /// behind the nominal deadline. + pub fn now(&self) -> Instant { + self.clock.now() + } +} + +impl Timer for TimerRef +where + I: Hash + Clone + Eq, + O: OneshotState, + P: PeriodicState, +{ + type Id = I; + type OneshotState = O; + type PeriodicState = P; + + fn schedule_once(&mut self, timeout: Duration, state: Self::OneshotState) { + let entry = TimerEntry::OneShot { timeout, state }; + self.work_queue + .send(TimerMsg::Schedule(entry)) + .unwrap_or_else(|error| eprintln!("Could not send Schedule msg: {:?}", error)); + } + + fn schedule_periodic(&mut self, delay: Duration, period: Duration, state: Self::PeriodicState) { + let entry = TimerEntry::Periodic { + delay, + period, + state, + }; + self.work_queue + .send(TimerMsg::Schedule(entry)) + .unwrap_or_else(|error| eprintln!("Could not send Schedule msg: {:?}", error)); + } + + fn cancel(&mut self, id: &Self::Id) { + self.work_queue + .send(TimerMsg::Cancel(id.clone())) + .unwrap_or_else(|error| eprintln!("Could not send Cancel msg: {:?}", error)); + } +} + +impl Clone for TimerRef +where + I: Hash + Clone + Eq, + O: OneshotState, + P: PeriodicState, +{ + fn clone(&self) -> Self { + Self { + work_queue: self.work_queue.clone(), + clock: self.clock.clone(), + } + } +} + +/// Non-blocking clock source shared with queue-backed timer handles. +/// +/// The manual variant stores logical time as an atomic elapsed offset from a +/// fixed base instant so component threads can read the current timer-aligned +/// instant without taking the manual timer's mutex. +#[derive(Clone, Debug)] +pub(crate) enum ClockRef { + Realtime, + Manual { + base: Instant, + elapsed_millis: Arc, + }, +} + +impl ClockRef { + pub(crate) fn realtime() -> Self { + Self::Realtime + } + + pub(crate) fn manual(base: Instant, elapsed_millis: Arc) -> Self { + Self::Manual { + base, + elapsed_millis, + } + } + + fn now(&self) -> Instant { + match self { + Self::Realtime => Instant::now(), + Self::Manual { + base, + elapsed_millis, + } => { + let elapsed = elapsed_millis.load(Ordering::Acquire); + base.checked_add(Duration::from_millis(elapsed)) + .expect("manual timer instant overflow") + } + } + } +} + +#[derive(Debug)] +pub(crate) enum QueueTimerControl { + Continue, + Stop, +} + +#[derive(Debug)] +enum QueueTimerEntry +where + I: Hash + Clone + Eq, + O: OneshotState, + P: PeriodicState, +{ + OneShot { state: O }, + Periodic { period: Duration, state: P }, +} + +impl QueueTimerEntry +where + I: Hash + Clone + Eq + fmt::Debug, + O: OneshotState + fmt::Debug, + P: PeriodicState + fmt::Debug, +{ + fn from_timer_entry(entry: TimerEntry) -> (Self, Duration) { + match entry { + TimerEntry::OneShot { timeout, state } => { + let queue_entry = QueueTimerEntry::OneShot { state }; + (queue_entry, timeout) + } + TimerEntry::Periodic { + delay, + period, + state, + } => { + let queue_entry = QueueTimerEntry::Periodic { period, state }; + (queue_entry, delay) + } + } + } + + fn execute(self) -> Option<(Self, Duration)> { + match self { + QueueTimerEntry::OneShot { state } => { + state.trigger(); + None + } + QueueTimerEntry::Periodic { period, state } => match state.trigger() { + TimerReturn::Reschedule(new_state) => { + let new_entry = QueueTimerEntry::Periodic { + period, + state: new_state, + }; + Some((new_entry, period)) + } + TimerReturn::Cancel => None, + }, + } + } + + fn execute_unique_ref(unique_ref: Rc) -> Option<(Rc, Duration)> { + let unique = Rc::try_unwrap(unique_ref).expect("shouldn't hold on to these refs anywhere"); + unique + .execute() + .map(|(new_unique, delay)| (Rc::new(new_unique), delay)) + } +} + +impl CancellableTimerEntry for QueueTimerEntry +where + I: Hash + Clone + Eq + fmt::Debug, + O: OneshotState + fmt::Debug, + P: PeriodicState + fmt::Debug, +{ + type Id = I; + + fn id(&self) -> &Self::Id { + match self { + QueueTimerEntry::OneShot { state } => state.id(), + QueueTimerEntry::Periodic { state, .. } => state.id(), + } + } +} + +pub(crate) struct QueueTimerCore +where + I: Hash + Clone + Eq + fmt::Debug, + O: OneshotState + fmt::Debug, + P: PeriodicState + fmt::Debug, +{ + timer: QuadWheelWithOverflow>, + work_queue: channel::Receiver>, +} + +impl QueueTimerCore +where + I: Hash + Clone + Eq + fmt::Debug, + O: OneshotState + fmt::Debug, + P: PeriodicState + fmt::Debug, +{ + pub(crate) fn new(work_queue: channel::Receiver>) -> Self { + Self { + timer: QuadWheelWithOverflow::new(), + work_queue, + } + } + + pub(crate) fn work_queue(&self) -> &channel::Receiver> { + &self.work_queue + } + + pub(crate) fn try_recv(&self) -> Result, channel::TryRecvError> { + self.work_queue.try_recv() + } + + pub(crate) fn recv(&self) -> Result, channel::RecvError> { + self.work_queue.recv() + } + + pub(crate) fn drain_messages(&mut self) -> QueueTimerControl { + loop { + match self.work_queue.try_recv() { + Ok(message) => { + if matches!(self.handle_msg(message), QueueTimerControl::Stop) { + return QueueTimerControl::Stop; + } + } + Err(channel::TryRecvError::Empty) => return QueueTimerControl::Continue, + Err(channel::TryRecvError::Disconnected) => { + panic!("Timer work_queue unexpectedly shut down!") + } + } + } + } + + pub(crate) fn handle_msg(&mut self, message: TimerMsg) -> QueueTimerControl { + match message { + TimerMsg::Stop => QueueTimerControl::Stop, + TimerMsg::Schedule(entry) => { + let (queue_entry, delay) = QueueTimerEntry::from_timer_entry(entry); + match self + .timer + .insert_ref_with_delay(Rc::new(queue_entry), delay) + { + Ok(_) => {} + Err(TimerError::Expired(entry)) => { + self.trigger_entry(entry); + } + Err(error) => panic!("Could not insert timer entry! {:?}", error), + } + QueueTimerControl::Continue + } + TimerMsg::Cancel(id) => { + match self.timer.cancel(&id) { + Ok(_) => {} + Err(TimerError::NotFound) => {} + Err(error) => panic!("Unexpected error cancelling timer! {:?}", error), + } + QueueTimerControl::Continue + } + } + } + + pub(crate) fn can_skip(&self) -> Skip { + self.timer.can_skip() + } + + pub(crate) fn is_idle(&self) -> bool { + self.work_queue.is_empty() && matches!(self.timer.can_skip(), Skip::Empty) + } + + pub(crate) fn skip(&mut self, amount: u32) { + self.timer.skip(amount); + } + + pub(crate) fn tick(&mut self) -> usize { + let expired = self.timer.tick(); + let expired_count = expired.len(); + for entry in expired { + self.trigger_entry(entry); + } + expired_count + } + + fn trigger_entry(&mut self, entry: Rc>) { + if let Some((new_entry, delay)) = QueueTimerEntry::execute_unique_ref(entry) { + match self.timer.insert_ref_with_delay(new_entry, delay) { + Ok(_) => {} + Err(TimerError::Expired(entry)) => panic!( + "Trying to insert periodic timer entry with 0ms period! {:?}", + entry + ), + Err(error) => panic!("Could not insert timer entry! {:?}", error), + } + } + } +} diff --git a/src/thread_timer.rs b/src/thread_timer.rs index e6bb8b0..aaf78bc 100644 --- a/src/thread_timer.rs +++ b/src/thread_timer.rs @@ -43,83 +43,16 @@ use super::*; -use crate::wheels::{cancellable::*, *}; +use crate::{ + queue_timer::{ClockRef, QueueTimerControl, QueueTimerCore, TimerMsg}, + wheels::Skip, +}; use channel::select; use crossbeam_channel as channel; -use std::{cmp::Ordering, fmt, io, rc::Rc, thread, time::Instant}; +use std::{cmp::Ordering, fmt, io, thread, time::Instant}; -#[derive(Debug)] -enum TimerMsg -where - I: Hash + Clone + Eq, - O: OneshotState, - P: PeriodicState, -{ - Schedule(TimerEntry), - Cancel(I), - Stop, -} - -/// A reference to a thread timer -/// -/// This is used to schedule events on the timer from other threads. -/// -/// You can get an instance via [timer_ref](TimerWithThread::timer_ref). -pub struct TimerRef -where - I: Hash + Clone + Eq, - O: OneshotState, - P: PeriodicState, -{ - work_queue: channel::Sender>, -} - -impl Timer for TimerRef -where - I: Hash + Clone + Eq, - O: OneshotState, - P: PeriodicState, -{ - type Id = I; - type OneshotState = O; - type PeriodicState = P; - - fn schedule_once(&mut self, timeout: Duration, state: Self::OneshotState) { - let e = TimerEntry::OneShot { timeout, state }; - self.work_queue - .send(TimerMsg::Schedule(e)) - .unwrap_or_else(|e| eprintln!("Could not send Schedule msg: {:?}", e)); - } +pub use crate::queue_timer::TimerRef; - fn schedule_periodic(&mut self, delay: Duration, period: Duration, state: Self::PeriodicState) { - let e = TimerEntry::Periodic { - delay, - period, - state, - }; - self.work_queue - .send(TimerMsg::Schedule(e)) - .unwrap_or_else(|e| eprintln!("Could not send Schedule msg: {:?}", e)); - } - - fn cancel(&mut self, id: &Self::Id) { - self.work_queue - .send(TimerMsg::Cancel(id.clone())) - .unwrap_or_else(|e| eprintln!("Could not send Cancel msg: {:?}", e)); - } -} -impl Clone for TimerRef -where - I: Hash + Clone + Eq, - O: OneshotState, - P: PeriodicState, -{ - fn clone(&self) -> Self { - Self { - work_queue: self.work_queue.clone(), - } - } -} /// A timer implementation that uses its own thread /// /// This struct acts as a main handle for the timer and its thread. @@ -143,18 +76,17 @@ where /// /// The thread will be called `"timer-thread"`. pub fn new() -> io::Result> { - let (s, r) = channel::unbounded(); + let (sender, receiver) = channel::unbounded(); let handle = thread::Builder::new() .name("timer-thread".to_string()) .spawn(move || { - let timer = TimerThread::new(r); + let timer = TimerThread::new(receiver); timer.run(); })?; - let twt = TimerWithThread { + Ok(TimerWithThread { timer_thread: handle, - work_queue: s, - }; - Ok(twt) + work_queue: sender, + }) } /// Returns a shareable reference to this timer @@ -162,9 +94,7 @@ where /// The reference contains the timer's work queue /// and can be used to schedule timeouts on this timer. pub fn timer_ref(&self) -> TimerRef { - TimerRef { - work_queue: self.work_queue.clone(), - } + TimerRef::new(self.work_queue.clone(), ClockRef::realtime()) } /// Shut this timer down @@ -174,7 +104,7 @@ where pub fn shutdown(self) -> Result<(), ThreadTimerError> { self.work_queue .send(TimerMsg::Stop) - .unwrap_or_else(|e| eprintln!("Could not send Stop msg: {:?}", e)); + .unwrap_or_else(|error| eprintln!("Could not send Stop msg: {:?}", error)); match self.timer_thread.join() { Ok(_) => Ok(()), Err(_) => { @@ -188,7 +118,7 @@ where pub fn shutdown_async(&self) -> Result<(), ThreadTimerError> { self.work_queue .send(TimerMsg::Stop) - .unwrap_or_else(|e| eprintln!("Could not send Stop msg: {:?}", e)); + .unwrap_or_else(|error| eprintln!("Could not send Stop msg: {:?}", error)); Ok(()) } } @@ -232,93 +162,13 @@ where CouldNotJoinThread, } -// Almost the same as `TimerEntry`, but not storing unnecessary things -#[derive(Debug)] -enum ThreadTimerEntry -where - I: Hash + Clone + Eq, - O: OneshotState, - P: PeriodicState, -{ - OneShot { state: O }, - Periodic { period: Duration, state: P }, -} - -impl ThreadTimerEntry -where - I: Hash + Clone + Eq + fmt::Debug, - O: OneshotState + fmt::Debug, - P: PeriodicState + fmt::Debug, -{ - fn from(e: TimerEntry) -> (Self, Duration) { - match e { - TimerEntry::OneShot { timeout, state } => { - let tte = ThreadTimerEntry::OneShot { state }; - (tte, timeout) - } - TimerEntry::Periodic { - delay, - period, - state, - } => { - let tte = ThreadTimerEntry::Periodic { period, state }; - (tte, delay) - } - } - } - - fn execute(self) -> Option<(Self, Duration)> { - match self { - ThreadTimerEntry::OneShot { state } => { - state.trigger(); - None - } - ThreadTimerEntry::Periodic { period, state } => match state.trigger() { - TimerReturn::Reschedule(new_state) => { - let new_entry = ThreadTimerEntry::Periodic { - period, - state: new_state, - }; - Some((new_entry, period)) - } - TimerReturn::Cancel => None, - }, - } - } - - fn execute_unique_ref(unique_ref: Rc) -> Option<(Rc, Duration)> { - let unique = Rc::try_unwrap(unique_ref).expect("shouldn't hold on to these refs anywhere"); - unique.execute().map(|t| { - let (new_unique, delay) = t; - (Rc::new(new_unique), delay) - }) - } -} - -impl CancellableTimerEntry for ThreadTimerEntry -where - I: Hash + Clone + Eq + fmt::Debug, - O: OneshotState + fmt::Debug, - P: PeriodicState + fmt::Debug, -{ - type Id = I; - - fn id(&self) -> &Self::Id { - match self { - ThreadTimerEntry::OneShot { state, .. } => state.id(), - ThreadTimerEntry::Periodic { state, .. } => state.id(), - } - } -} - struct TimerThread where I: Hash + Clone + Eq + fmt::Debug, O: OneshotState + fmt::Debug, P: PeriodicState + fmt::Debug, { - timer: QuadWheelWithOverflow>, - work_queue: channel::Receiver>, + core: QueueTimerCore, running: bool, start: Instant, last_check: u128, @@ -330,67 +180,58 @@ where O: OneshotState + fmt::Debug, P: PeriodicState + fmt::Debug, { - fn new(work_queue: channel::Receiver>) -> TimerThread { - TimerThread { - timer: QuadWheelWithOverflow::new(), - work_queue, + fn new(work_queue: channel::Receiver>) -> Self { + Self { + core: QueueTimerCore::new(work_queue), running: true, start: Instant::now(), - last_check: 0u128, + last_check: 0, } } fn run(mut self) { while self.running { - let elap = self.elapsed(); - if elap > 0 { - for _ in 0..elap { - self.tick(); + let elapsed = self.elapsed(); + if elapsed > 0 { + for _ in 0..elapsed { + self.core.tick(); } } - match self.work_queue.try_recv() { - Ok(msg) => self.handle_msg(msg), - Err(channel::TryRecvError::Empty) => { - match self.timer.can_skip() { - Skip::None => { - thread::yield_now(); // try again after yielding for a bit - } - Skip::Empty => { - // wait until something is scheduled - // don't even need to bother skipping time in the wheel, - // since all times in there are relative - match self.work_queue.recv() { - Ok(msg) => { - self.reset(); // since we waited for an arbitrary time and taking a new timestamp incurs no error - self.handle_msg(msg) - } - Err(channel::RecvError) => { - panic!("Timer work_queue unexpectedly shut down!") - } - } + match self.core.try_recv() { + Ok(message) => self.apply_message(message), + Err(channel::TryRecvError::Empty) => match self.core.can_skip() { + Skip::None => { + thread::yield_now(); + } + Skip::Empty => match self.core.recv() { + Ok(message) => { + self.reset(); + self.apply_message(message); } - Skip::Millis(can_skip) if can_skip > 5 => { - let waiting_time = can_skip - 5; // balance OS scheduler inaccuracy - // wait until something is scheduled but max skip - let timeout = Duration::from_millis(waiting_time as u64); - let res = select! { - recv(self.work_queue) -> msg => msg.ok(), - default(timeout) => None, - }; - let elap = self.elapsed(); - self.skip_and_tick(can_skip, elap); - if let Some(msg) = res { - self.handle_msg(msg) - } + Err(channel::RecvError) => { + panic!("Timer work_queue unexpectedly shut down!") } - Skip::Millis(can_skip) => { - thread::yield_now(); - let elap = self.elapsed(); - self.skip_and_tick(can_skip, elap); + }, + Skip::Millis(can_skip) if can_skip > 5 => { + let waiting_time = can_skip - 5; + let timeout = Duration::from_millis(u64::from(waiting_time)); + let result = select! { + recv(self.core.work_queue()) -> message => message.ok(), + default(timeout) => None, + }; + let elapsed = self.elapsed(); + self.skip_and_tick(can_skip, elapsed); + if let Some(message) = result { + self.apply_message(message); } } - } + Skip::Millis(can_skip) => { + thread::yield_now(); + let elapsed = self.elapsed(); + self.skip_and_tick(can_skip, elapsed); + } + }, Err(channel::TryRecvError::Disconnected) => { panic!("Timer work_queue unexpectedly shut down!") } @@ -400,34 +241,31 @@ where #[inline(always)] fn skip_and_tick(&mut self, can_skip: u32, elapsed: u128) { - let can_skip_u128 = can_skip as u128; + let can_skip_u128 = u128::from(can_skip); match elapsed.cmp(&can_skip_u128) { Ordering::Greater => { - // took longer to get rescheduled than we wanted - self.timer.skip(can_skip); + self.core.skip(can_skip); let ticks = elapsed - can_skip_u128; for _ in 0..ticks { - self.tick(); + self.core.tick(); } } Ordering::Less => { - // we got woken up early, no need to tick - self.timer.skip(elapsed as u32); + self.core.skip(elapsed as u32); } Ordering::Equal => { - // elapsed == can_skip - self.timer.skip(can_skip); + self.core.skip(can_skip); } } } #[inline(always)] fn elapsed(&mut self) -> u128 { - let elap = self.start.elapsed().as_millis(); - let rel_elap = elap - self.last_check; - self.last_check = elap; - rel_elap + let elapsed = self.start.elapsed().as_millis(); + let relative_elapsed = elapsed - self.last_check; + self.last_check = elapsed; + relative_elapsed } #[inline(always)] @@ -437,45 +275,9 @@ where } #[inline(always)] - fn handle_msg(&mut self, msg: TimerMsg) { - match msg { - TimerMsg::Stop => self.running = false, - TimerMsg::Schedule(entry) => { - let (e, delay) = ThreadTimerEntry::from(entry); - match self.timer.insert_ref_with_delay(Rc::new(e), delay) { - Ok(_) => (), // ok - Err(TimerError::Expired(e)) => { - self.trigger_entry(e); - } - Err(f) => panic!("Could not insert timer entry! {:?}", f), - } - } - TimerMsg::Cancel(ref id) => match self.timer.cancel(id) { - Ok(_) => (), // ok - Err(TimerError::NotFound) => (), // also ok, might have been triggered already - Err(f) => panic!("Unexpected error cancelling timer! {:?}", f), - }, - } - } - - fn trigger_entry(&mut self, e: Rc>) { - if let Some((new_e, delay)) = ThreadTimerEntry::execute_unique_ref(e) { - match self.timer.insert_ref_with_delay(new_e, delay) { - Ok(_) => (), // ok - Err(TimerError::Expired(e)) => panic!( - "Trying to insert periodic timer entry with 0ms period! {:?}", - e - ), - Err(f) => panic!("Could not insert timer entry! {:?}", f), - } - } // otherwise: timer is not rescheduled - } - - #[inline(always)] - fn tick(&mut self) { - let res = self.timer.tick(); - for e in res { - self.trigger_entry(e); + fn apply_message(&mut self, message: TimerMsg) { + if matches!(self.core.handle_msg(message), QueueTimerControl::Stop) { + self.running = false; } } } @@ -485,6 +287,7 @@ where mod tests { use super::*; use crate::test_helpers::*; + use crossbeam_channel as channel; use std::sync::{Arc, Mutex}; use uuid::Uuid; @@ -504,15 +307,15 @@ mod tests { total_wait += timeout; let now = Instant::now(); timer.schedule_action_once(id, timeout, move |_| { - let elap = now.elapsed().as_nanos(); + let elapsed = now.elapsed().as_nanos(); let target = timeout.as_nanos(); - match elap.cmp(&target) { + match elapsed.cmp(&target) { Ordering::Greater => { - let diff = ((elap - target) as f64) / 1000000.0; + let diff = ((elapsed - target) as f64) / 1000000.0; println!("Running action {} {}ms late", i, diff); } Ordering::Less => { - let diff = ((target - elap) as f64) / 1000000.0; + let diff = ((target - elapsed) as f64) / 1000000.0; println!("Running action {} {}ms early", i, diff); } Ordering::Equal => { @@ -529,8 +332,8 @@ mod tests { .shutdown() .expect("Timer didn't shutdown properly!"); println!("Timing run done!"); - for b in barriers { - let guard = b.lock().unwrap(); + for barrier in barriers { + let guard = barrier.lock().unwrap(); assert!(*guard); } } @@ -568,19 +371,54 @@ mod tests { .shutdown() .expect("Timer didn't shutdown properly!"); println!("Timing run done!"); - for b in barriers { - let guard = b.lock().unwrap(); + for barrier in barriers { + let guard = barrier.lock().unwrap(); assert!(*guard); } } + #[test] + fn timer_ref_now_inside_handler_is_not_earlier_than_due_time() { + let timer_core = TimerWithThread::for_uuid_closures(); + let mut timer_ref = timer_core.timer_ref(); + let scheduled_from = timer_ref.now(); + let delay = Duration::from_millis(20); + let deadline = scheduled_from + .checked_add(delay) + .expect("thread timer deadline should not overflow"); + let handler_timer_ref = timer_ref.clone(); + let (observed_now_sender, observed_now_receiver) = channel::bounded(1); + + timer_ref.schedule_action_once(Uuid::new_v4(), delay, move |_| { + let now = handler_timer_ref.now(); + observed_now_sender + .send(now) + .expect("oneshot receiver should stay alive"); + }); + + let wait_deadline = Instant::now() + .checked_add(Duration::from_secs(1)) + .expect("thread timer receive deadline should not overflow"); + let observed_now = observed_now_receiver + .recv_deadline(wait_deadline) + .expect("oneshot callback should record observed time before the test deadline"); + + timer_core + .shutdown() + .expect("Timer didn't shutdown properly!"); + + assert!( + observed_now >= deadline, + "handler observed {observed_now:?}, which is earlier than scheduled lower bound {deadline:?}" + ); + } + /// Check that the TimeRef is cloneable, even if its type parameters aren't. #[test] fn timer_ref_clone() { let timer = TimerWithThread::for_uuid_closures(); let timer_ref: TimerRef, PeriodicClosureState> = timer.timer_ref(); - // No need to use it. Just checking that it compiles. let _cloned_ref = timer_ref.clone(); } } From 993f7f07af7e674a0bd7b6fd09da11991f4688fc Mon Sep 17 00:00:00 2001 From: Lars Kroll Date: Sat, 11 Apr 2026 13:33:22 +0200 Subject: [PATCH 2/3] Refine manual timer API naming --- src/manual_timer.rs | 14 +++++++------- src/queue_timer.rs | 1 - src/thread_timer.rs | 31 ++++++++++++++++++++----------- 3 files changed, 27 insertions(+), 19 deletions(-) diff --git a/src/manual_timer.rs b/src/manual_timer.rs index 51ffe1a..8ea6966 100644 --- a/src/manual_timer.rs +++ b/src/manual_timer.rs @@ -6,7 +6,6 @@ //! [`ManualTimer::advance_to_next`]. use super::*; - use crate::{ queue_timer::{ClockRef, QueueTimerControl, QueueTimerCore, TimerMsg, TimerRef}, wheels::Skip, @@ -84,11 +83,11 @@ where inner.advance_to_next() } - /// Returns the current manually-driven logical time. + /// Returns the elapsed logical time since this manual timer was created. /// /// This is rounded down to the manual timer's millisecond resolution so it /// agrees with the deadlines the wheel can actually execute. - pub fn current_time(&self) -> Duration { + pub fn elapsed(&self) -> Duration { Duration::from_millis(self.elapsed_millis.load(Ordering::Acquire)) } @@ -100,8 +99,9 @@ where !inner.running || inner.core.is_idle() } - /// Stops the timer so future advances no longer execute queued work. - pub fn shutdown(&self) { + /// Stops this shared manual timer so future advances no longer execute + /// queued work, even through other clones of the same timer handle. + pub fn stop(&self) { let mut inner = self.inner.lock().expect("manual timer mutex poisoned"); inner.running = false; } @@ -400,11 +400,11 @@ mod tests { let base = timer_ref.now(); timer.advance_by(Duration::from_micros(999)); - assert_eq!(timer.current_time(), Duration::ZERO); + assert_eq!(timer.elapsed(), Duration::ZERO); assert_eq!(timer_ref.now(), base); timer.advance_by(Duration::from_micros(1)); - assert_eq!(timer.current_time(), Duration::from_millis(1)); + assert_eq!(timer.elapsed(), Duration::from_millis(1)); assert_eq!( timer_ref.now().duration_since(base), Duration::from_millis(1) diff --git a/src/queue_timer.rs b/src/queue_timer.rs index 2d6aacf..c2b2f81 100644 --- a/src/queue_timer.rs +++ b/src/queue_timer.rs @@ -1,5 +1,4 @@ use super::*; - use crate::wheels::{cancellable::*, *}; use crossbeam_channel as channel; use std::{ diff --git a/src/thread_timer.rs b/src/thread_timer.rs index aaf78bc..b0d402d 100644 --- a/src/thread_timer.rs +++ b/src/thread_timer.rs @@ -42,7 +42,6 @@ //! ``` use super::*; - use crate::{ queue_timer::{ClockRef, QueueTimerControl, QueueTimerCore, TimerMsg}, wheels::Skip, @@ -202,19 +201,25 @@ where Ok(message) => self.apply_message(message), Err(channel::TryRecvError::Empty) => match self.core.can_skip() { Skip::None => { - thread::yield_now(); + thread::yield_now(); // Try again after yielding for a bit. } - Skip::Empty => match self.core.recv() { - Ok(message) => { - self.reset(); - self.apply_message(message); + Skip::Empty => { + // Wait until something is scheduled. + // Don't even need to bother skipping time in the wheel, + // since all times in there are relative. + match self.core.recv() { + Ok(message) => { + self.reset(); // Since we waited for an arbitrary time and taking a new timestamp incurs no error. + self.apply_message(message); + } + Err(channel::RecvError) => { + panic!("Timer work_queue unexpectedly shut down!") + } } - Err(channel::RecvError) => { - panic!("Timer work_queue unexpectedly shut down!") - } - }, + } Skip::Millis(can_skip) if can_skip > 5 => { - let waiting_time = can_skip - 5; + // Wait until something is scheduled but max can_skip. + let waiting_time = can_skip - 5; // Balance OS scheduler inaccuracy. let timeout = Duration::from_millis(u64::from(waiting_time)); let result = select! { recv(self.core.work_queue()) -> message => message.ok(), @@ -245,6 +250,7 @@ where match elapsed.cmp(&can_skip_u128) { Ordering::Greater => { + // It took longer to get rescheduled than we wanted. self.core.skip(can_skip); let ticks = elapsed - can_skip_u128; for _ in 0..ticks { @@ -252,9 +258,11 @@ where } } Ordering::Less => { + // We got woken up early, no need to tick. self.core.skip(elapsed as u32); } Ordering::Equal => { + // elapsed == can_skip self.core.skip(can_skip); } } @@ -419,6 +427,7 @@ mod tests { let timer = TimerWithThread::for_uuid_closures(); let timer_ref: TimerRef, PeriodicClosureState> = timer.timer_ref(); + // No need to use it. Just checking that it compiles. let _cloned_ref = timer_ref.clone(); } } From d2ff7115bb77c21f3a41d6f07981629422aead77 Mon Sep 17 00:00:00 2001 From: Lars Kroll Date: Sat, 11 Apr 2026 13:43:47 +0200 Subject: [PATCH 3/3] Fix manual timer clippy layout --- src/manual_timer.rs | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/src/manual_timer.rs b/src/manual_timer.rs index 8ea6966..dc5c398 100644 --- a/src/manual_timer.rs +++ b/src/manual_timer.rs @@ -319,6 +319,15 @@ where } } +#[cfg(feature = "uuid-extras")] +impl ManualTimer, PeriodicClosureState> { + /// Shorthand for creating a manual timer using `Uuid` identifiers and + /// closure state. + pub fn for_uuid_closures() -> Self { + Self::new() + } +} + #[cfg(feature = "uuid-extras")] #[cfg(test)] mod tests { @@ -439,12 +448,3 @@ mod tests { ); } } - -#[cfg(feature = "uuid-extras")] -impl ManualTimer, PeriodicClosureState> { - /// Shorthand for creating a manual timer using `Uuid` identifiers and - /// closure state. - pub fn for_uuid_closures() -> Self { - Self::new() - } -}