subms_timer_wheel/features/
concurrent.rs

1//! Thread-safe timer wheel: short-mutex wrapper around the base
2//! `TimerWheel`. Schedule + cancel + tick all serialize on a single
3//! `Mutex` because the critical sections are O(1) (or O(slot) on
4//! tick - bounded by entries-in-bucket, typically tiny).
5//!
6//! Why a mutex and not a lock-free or sharded design: timer-wheel
7//! operations are short enough that a contended mutex still wins on
8//! tail latency vs the cache-line ping-pong of an atomic-list shape,
9//! provided callers don't hold long external locks while inside a
10//! callback. The `tick()` method returns the fired values out of the
11//! critical section, so a caller can release the lock between
12//! retrieval and dispatch.
13//!
14//! Tradeoff vs the base wheel: every operation pays a lock + unlock.
15//! For single-threaded workloads, prefer the base `TimerWheel`.
16
17use crate::{TimerError, TimerWheel};
18use std::sync::{Arc, Mutex};
19
20pub struct ConcurrentTimerWheel<V> {
21    inner: Arc<Mutex<TimerWheel<V>>>,
22}
23
24impl<V> ConcurrentTimerWheel<V> {
25    pub fn new(num_slots: usize) -> Self {
26        Self {
27            inner: Arc::new(Mutex::new(TimerWheel::new(num_slots))),
28        }
29    }
30
31    fn locked(&self) -> std::sync::MutexGuard<'_, TimerWheel<V>> {
32        self.inner.lock().expect("timer-wheel mutex poisoned")
33    }
34
35    pub fn num_slots(&self) -> usize {
36        self.locked().num_slots()
37    }
38
39    pub fn max_delay(&self) -> u64 {
40        self.locked().max_delay()
41    }
42
43    pub fn pending(&self) -> usize {
44        self.locked().pending()
45    }
46
47    pub fn is_empty(&self) -> bool {
48        self.locked().is_empty()
49    }
50
51    pub fn slot_len(&self, slot: usize) -> usize {
52        self.locked().slot_len(slot)
53    }
54
55    pub fn schedule(&self, delay_ticks: usize, value: V) -> u64 {
56        self.locked().schedule(delay_ticks, value)
57    }
58
59    pub fn try_schedule(&self, delay_ticks: usize, value: V) -> Result<u64, TimerError> {
60        self.locked().try_schedule(delay_ticks, value)
61    }
62
63    pub fn cancel(&self, id: u64) -> bool {
64        self.locked().cancel(id)
65    }
66
67    pub fn reschedule(&self, id: u64, delay_ticks: usize) -> bool {
68        self.locked().reschedule(id, delay_ticks)
69    }
70
71    /// Advance one tick. Returns the fired values; the mutex is
72    /// released before the caller dispatches them.
73    pub fn tick(&self) -> Vec<V> {
74        self.locked().tick()
75    }
76
77    pub fn advance(&self, ticks: usize) -> Vec<V> {
78        self.locked().advance(ticks)
79    }
80
81    pub fn drain(&self) -> Vec<V> {
82        self.locked().drain()
83    }
84
85    pub fn clear(&self) {
86        self.locked().clear()
87    }
88}
89
90impl<V> Clone for ConcurrentTimerWheel<V> {
91    fn clone(&self) -> Self {
92        Self {
93            inner: Arc::clone(&self.inner),
94        }
95    }
96}
97
98#[cfg(test)]
99#[path = "concurrent_tests.rs"]
100mod tests;
subms_timer_wheel/features/concurrent.rs

subms_timer_wheel/features/
concurrent.rs
