//! A pure, minimal, and scalable structure for tracking expiration of timers //! //! ``` //! # use timer_queue::TimerQueue; //! let mut q = TimerQueue::new(); //! q.insert(42, "second"); //! q.insert(17, "first"); //! assert!(q.next_timeout().unwrap() <= 17); //! assert_eq!(q.poll(16), None); //! assert_eq!(q.poll(17), Some("first")); //! assert_eq!(q.poll(100), Some("second")); //! ``` #![no_std] use core::fmt; use slab::Slab; #[cfg(feature = "serde")] use serde::ser::SerializeSeq; /// Stores values to be yielded at specific times in the future /// /// Time is expressed as a bare u64 representing an absolute point in time. The caller may use any /// consistent unit, e.g. milliseconds, and any consistent definition of time zero. Larger units /// limit resolution but make `poll`ing over the same real-time interval proportionately faster, /// whereas smaller units improve resolution, limit total range, and reduce `poll` performance. #[derive(Debug, Clone)] pub struct TimerQueue { /// Definitions of each active timer /// /// Timers are defined here, and referenced indirectly by index from `levels` and in the public /// API. This allows for safe construction of intrusive linked lists between timers, and helps /// reduce the amount of data that needs to be routinely shuffled around in `levels` as time /// passes. timers: Slab>, /// A hierarchical timer wheel /// /// This data structure breaks down points in time into digits. The base of those digits can be /// chosen arbitrarily; this implementation uses base `2^LOG_2_SLOTS`. A power of two makes it /// easy to manipulate individual digits using bit shifts and masking because each digit /// corresponds directly to `LOG_2_SLOTS` bits in the binary representation. For familiarity, we /// will illustrate a timer wheel built instead on base 10, but the behavior is identical. /// /// Consider this timer wheel where timers are set at times 32, 42, and 46, and `next_tick` is /// between 30 and 32 inclusive. Note that the number of slots in each level is equal to the /// base of the digits used, in this case 10. /// /// ```text /// +--+--+--+--+-- /// Level 0 |30|31|32|33| ... /// +--+--+--+--+-- /// \ | / /// \ V / /// \ +--+ / /// \ |32| / /// \+--+ / /// \ / /// +--+--+--+--+--+--+--+--+--+--+ /// Level 1 |00|10|20|30|40|50|60|70|80|90| /// +--+--+--+--+--+--+--+--+--+--+ /// | /// V /// +--+ /// |46| /// +--+ /// ^| /// |V /// +--+ /// |42| /// +--+ /// ``` /// /// Timers are organized into buckets (or slots) at a resolution that decreases exponentially /// with distance from `next_tick`, the present. Higher-numbered levels cover larger intervals, /// until the highest-numbered level covers the complete representable of timers, from 0 to /// `u64::MAX`. Every lower level covers the slot in the next highest level which `next_tick` /// lies within. Level 0 represents the maximum resolution, where each slot covers exactly one /// unit of time. /// /// The slot that a timer should be stored in is easily computed based on `next_tick` and the /// desired expiry time. For a base 10 structure, find the most significant digit in the base 10 /// representations of `next_tick` and the desired expiry time that differs between the two. The /// position of that digit is the level, and the value of that digit is the position in the /// level. For example, if `next_tick` is 7342, and a timer is scheduled for time 7361, the /// timer would be stored at level 1, slot 6. Note that no subtraction is performed: the start /// of each level is always the greatest integer multiple of the level's span which is less than /// or equal to `next_tick`. /// /// Calls to `poll` move `next_tick` towards the passed-in time. When `next_tick` reaches a /// timer in level 0, it stops there and the timer is removed and returned from `poll`. Reaching /// the end of level 0 redefines level 0 to represent the next slot in level 1, at which point /// all timers stored in that slot are unpacked into appropriate slots of level 0, and traversal /// of level 0 begins again from the start. When level 1 is exhausted, the next slot in level 2 /// is unpacked into levels 1 and 0, and so on for higher levels. Slots preceding `next_tick` /// are therefore empty at any level, and for levels above 0, the slot containing `next_tick` is /// also empty, having necessarily been unpacked into lower levels. /// /// Assuming the number of timers scheduled within a period of time is on average proportional /// to the size of that period, advancing the queue by a constant amount of time has amortized /// constant time complexity, because the frequency with which slots at a particular level are /// unpacked is inversely proportional to the expected number of timers stored in that /// slot. /// /// Inserting, removing, and updating timers are constant-time operations thanks to the above /// and the use of unordered doubly linked lists to represent the contents of a slot. We can /// also compute a lower bound for the next timeout in constant time by scanning for the /// earliest nonempty slot. levels: [Level; LEVELS], /// Earliest point at which a timer may be pending /// /// Each `LOG_2_SLOTS` bits of this are a cursor into the associated level, in order of /// ascending significance. next_tick: u64, } impl TimerQueue { /// Create an empty queue starting at time `0` pub const fn new() -> Self { Self { timers: Slab::new(), levels: [Level::new(); LEVELS], next_tick: 0, } } /// Create a queue for which at least `n` calls to `insert` will not require a reallocation pub fn with_capacity(n: usize) -> Self { Self { timers: Slab::with_capacity(n), levels: [Level::new(); LEVELS], next_tick: 0, } } /// Returns a timer that has expired by `now`, if any /// /// `now` must be at least the largest previously passed value pub fn poll(&mut self, now: u64) -> Option { debug_assert!(now >= self.next_tick, "time advances monotonically"); loop { // Advance towards the next timeout self.advance_towards(now); // Check for timeouts in the immediate future if let Some(value) = self.scan_bottom(now) { return Some(value); } // If we can't advance any further, bail out if self.next_tick >= now { return None; } } } /// Find a timer expired by `now` in level 0 fn scan_bottom(&mut self, now: u64) -> Option { let index = self.levels[0].first_index()?; if slot_start(self.next_tick, 0, index) > now { return None; } let timer = self.levels[0].slots[index]; let state = self.timers.remove(timer.0); debug_assert_eq!(state.prev, None, "head of list has no predecessor"); debug_assert!(state.expiry <= now); if let Some(next) = state.next { debug_assert_eq!( self.timers[next.0].prev, Some(timer), "successor links to head" ); self.timers[next.0].prev = None; } self.levels[0].set(index, state.next); self.next_tick = state.expiry; self.maybe_shrink(); Some(state.value) } /// Advance to the start of the first nonempty slot or `now`, whichever is sooner fn advance_towards(&mut self, now: u64) { for level in 0..LEVELS { if let Some(slot) = self.levels[level].first_index() { if slot_start(self.next_tick, level, slot) > now { break; } self.advance_to(level, slot); return; } } self.next_tick = now; } /// Advance to a specific slot, which must be the first nonempty slot fn advance_to(&mut self, level: usize, slot: usize) { debug_assert!( self.levels[..level].iter().all(|level| level.is_empty()), "lower levels are empty" ); debug_assert!( self.levels[level].first_index().map_or(true, |x| x >= slot), "lower slots in this level are empty" ); // Advance into the slot self.next_tick = slot_start(self.next_tick, level, slot); if level == 0 { // No lower levels exist to unpack timers into return; } // Unpack all timers in this slot into lower levels while let Some(timer) = self.levels[level].take(slot) { let next = self.timers[timer.0].next; self.levels[level].set(slot, next); if let Some(next) = next { self.timers[next.0].prev = None; } self.list_unlink(timer); self.schedule(timer); } } /// Link `timer` from the slot associated with its expiry fn schedule(&mut self, timer: Timer) { debug_assert_eq!( self.timers[timer.0].next, None, "timer isn't already scheduled" ); debug_assert_eq!( self.timers[timer.0].prev, None, "timer isn't already scheduled" ); let (level, slot) = timer_index(self.next_tick, self.timers[timer.0].expiry); // Insert `timer` at the head of the list in the target slot let head = self.levels[level].get(slot); self.timers[timer.0].next = head; if let Some(head) = head { self.timers[head.0].prev = Some(timer); } self.levels[level].set(slot, Some(timer)); } /// Lower bound on when the next timer will expire, if any pub fn next_timeout(&self) -> Option { for level in 0..LEVELS { let start = ((self.next_tick >> (level * LOG_2_SLOTS)) & (SLOTS - 1) as u64) as usize; for slot in start..SLOTS { if self.levels[level].get(slot).is_some() { return Some(slot_start(self.next_tick, level, slot)); } } } None } /// Register a timer that will yield `value` at `timeout` pub fn insert(&mut self, timeout: u64, value: T) -> Timer { let timer = Timer(self.timers.insert(TimerState { expiry: timeout.max(self.next_tick), prev: None, next: None, value, })); self.schedule(timer); timer } /// Adjust `timer` to expire at `timeout` pub fn reset(&mut self, timer: Timer, timeout: u64) { self.unlink(timer); self.timers[timer.0].expiry = timeout.max(self.next_tick); self.schedule(timer); } /// Cancel `timer` pub fn remove(&mut self, timer: Timer) -> T { self.unlink(timer); let state = self.timers.remove(timer.0); self.maybe_shrink(); state.value } /// Release timer state memory if it's mostly unused fn maybe_shrink(&mut self) { if self.timers.capacity() / 16 > self.timers.len() { self.timers.shrink_to_fit(); } } /// Iterate over the expiration and value of all scheduled timers pub fn iter(&self) -> impl ExactSizeIterator { self.timers.iter().map(|(_, x)| (x.expiry, &x.value)) } /// Iterate over the expiration and value of all scheduled timers pub fn iter_mut(&mut self) -> impl ExactSizeIterator { self.timers .iter_mut() .map(|(_, x)| (x.expiry, &mut x.value)) } /// Borrow the value associated with `timer` pub fn get(&self, timer: Timer) -> &T { &self.timers[timer.0].value } /// Uniquely borrow the value associated with `timer` pub fn get_mut(&mut self, timer: Timer) -> &mut T { &mut self.timers[timer.0].value } /// Number of scheduled timers pub fn len(&self) -> usize { self.timers.len() } /// Whether no timers are scheduled pub fn is_empty(&self) -> bool { self.timers.is_empty() } /// Remove all references to `timer` fn unlink(&mut self, timer: Timer) { let (level, slot) = timer_index(self.next_tick, self.timers[timer.0].expiry); // If necessary, remove a reference to `timer` from its slot by replacing it with its // successor let slot_head = self.levels[level].get(slot).unwrap(); if slot_head == timer { self.levels[level].set(slot, self.timers[slot_head.0].next); debug_assert_eq!( self.timers[timer.0].prev, None, "head of list has no predecessor" ); } // Remove references to `timer` from other timers self.list_unlink(timer); } /// Remove `timer` from its list fn list_unlink(&mut self, timer: Timer) { let prev = self.timers[timer.0].prev.take(); let next = self.timers[timer.0].next.take(); if let Some(prev) = prev { // Remove reference from predecessor self.timers[prev.0].next = next; } if let Some(next) = next { // Remove reference from successor self.timers[next.0].prev = prev; } } } /// Compute the first tick that lies within a slot fn slot_start(base: u64, level: usize, slot: usize) -> u64 { let shift = (level * LOG_2_SLOTS) as u64; // Shifting twice avoids an overflow when level = 10. (base & ((!0 << shift) << LOG_2_SLOTS as u64)) | ((slot as u64) << shift) } /// Compute the level and slot for a certain expiry fn timer_index(base: u64, expiry: u64) -> (usize, usize) { // The level is the position of the first bit set in `expiry` but not in `base`, divided by the // number of bits spanned by each level. let differing_bits = base ^ expiry; let level = (63 - (differing_bits | 1).leading_zeros()) as usize / LOG_2_SLOTS; debug_assert!(level < LEVELS, "every possible expiry is in range"); // The slot in that level is the difference between the expiry time and the time at which the // level's span begins, after both times are shifted down to the level's granularity. Each // level's spans starts at `base`, rounded down to a multiple of the size of its span. let slot_base = (base >> (level * LOG_2_SLOTS)) & (!0 << LOG_2_SLOTS); let slot = (expiry >> (level * LOG_2_SLOTS)) - slot_base; debug_assert!(slot < SLOTS as u64); (level, slot as usize) } impl Default for TimerQueue { fn default() -> Self { Self::new() } } #[derive(Debug, Clone)] struct TimerState { /// Lowest argument to `poll` for which this timer may be returned expiry: u64, /// Value returned to the caller on expiry value: T, /// Predecessor within a slot's list prev: Option, /// Successor within a slot's list next: Option, } /// A set of contiguous timer lists, ordered by expiry /// /// Level `n` spans `2^(LOG_2_SLOTS * (n+1))` ticks, and each of its slots corresponds to a span of /// `2^(LOG_2_SLOTS * n)`. #[derive(Copy, Clone)] struct Level { slots: [Timer; SLOTS], /// Bit n indicates whether slot n is occupied, counting from LSB up occupied: u64, } impl Level { const fn new() -> Self { Self { slots: [Timer(usize::MAX); SLOTS], occupied: 0, } } fn first_index(&self) -> Option { let x = self.occupied.trailing_zeros() as usize; if x == self.slots.len() { return None; } Some(x) } fn get(&self, slot: usize) -> Option { if self.occupied & (1 << slot) == 0 { return None; } Some(self.slots[slot]) } fn take(&mut self, slot: usize) -> Option { let x = self.get(slot)?; self.set(slot, None); Some(x) } fn set(&mut self, slot: usize, timer: Option) { match timer { None => { self.slots[slot] = Timer(usize::MAX); self.occupied &= !(1 << slot); } Some(x) => { self.slots[slot] = x; self.occupied |= 1 << slot; } } } fn is_empty(&self) -> bool { self.occupied == 0 } } impl fmt::Debug for Level { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { let mut m = f.debug_map(); let numbered_nonempty_slots = self .slots .iter() .enumerate() .filter(|(i, _)| self.occupied & (1 << i) != 0); for (i, Timer(t)) in numbered_nonempty_slots { m.entry(&i, &t); } m.finish() } } const LOG_2_SLOTS: usize = 6; const LEVELS: usize = 1 + 64 / LOG_2_SLOTS; const SLOTS: usize = 1 << LOG_2_SLOTS; // Index in `TimerQueue::timers`. Future work: add a niche here. /// Handle to a specific timer, obtained from [`TimerQueue::insert`] #[derive(Debug, Copy, Clone, Eq, PartialEq)] pub struct Timer(usize); #[cfg(feature = "serde")] impl serde::Serialize for TimerQueue { fn serialize( &self, serializer: S, ) -> Result<::Ok, ::Error> where S: serde::Serializer, { let mut seq = serializer.serialize_seq(Some(self.len()))?; for v in self.iter() { let t: (u64, &T) = v; seq.serialize_element(&t)?; } seq.end() } } #[cfg(feature = "serde")] impl<'de, T> serde::Deserialize<'de> for TimerQueue where T: serde::Deserialize<'de>, { fn deserialize(deserializer: D) -> Result where D: serde::Deserializer<'de>, { use core::fmt::Formatter; use core::marker::PhantomData; struct TimerQueueVisitor(PhantomData); impl<'de, T> serde::de::Visitor<'de> for TimerQueueVisitor where T: serde::Deserialize<'de>, { type Value = TimerQueue; fn expecting(&self, formatter: &mut Formatter) -> fmt::Result { write!( formatter, "a sequence of (u64, {}) tuples", core::any::type_name::() ) } fn visit_seq(self, mut seq: A) -> Result where A: serde::de::SeqAccess<'de>, { let mut timer_queue = if let Some(size) = seq.size_hint() { TimerQueue::::with_capacity(size) } else { TimerQueue::::new() }; while let Some((time, value)) = seq.next_element::<(u64, T)>()? { timer_queue.insert(time, value); } Ok(timer_queue) } } deserializer.deserialize_seq(TimerQueueVisitor(PhantomData)) } } #[cfg(test)] mod tests { extern crate alloc; extern crate std; use std::{collections::HashMap, vec::Vec}; use super::*; use proptest::prelude::*; #[test] fn max_timeout() { let mut queue = TimerQueue::new(); queue.insert(u64::MAX, ()); assert!(queue.poll(u64::MAX - 1).is_none()); assert!(queue.poll(u64::MAX).is_some()); } #[test] fn slot_starts() { for i in 0..SLOTS { assert_eq!(slot_start(0, 0, i), i as u64); assert_eq!(slot_start(SLOTS as u64, 0, i), SLOTS as u64 + i as u64); assert_eq!(slot_start(SLOTS as u64 + 1, 0, i), SLOTS as u64 + i as u64); for j in 1..LEVELS { assert_eq!( slot_start(0, j, i), (SLOTS as u64).pow(j as u32).wrapping_mul(i as u64) ); } } } #[test] fn indexes() { assert_eq!(timer_index(0, 0), (0, 0)); assert_eq!(timer_index(0, SLOTS as u64 - 1), (0, SLOTS - 1)); assert_eq!( timer_index(SLOTS as u64 - 1, SLOTS as u64 - 1), (0, SLOTS - 1) ); assert_eq!(timer_index(0, SLOTS as u64), (1, 1)); for i in 0..LEVELS { assert_eq!(timer_index(0, (SLOTS as u64).pow(i as u32)), (i, 1)); if i < LEVELS - 1 { assert_eq!( timer_index(0, (SLOTS as u64).pow(i as u32 + 1) - 1), (i, SLOTS - 1) ); assert_eq!( timer_index(SLOTS as u64 - 1, (SLOTS as u64).pow(i as u32 + 1) - 1), (i, SLOTS - 1) ); } } } #[test] fn next_timeout() { let mut queue = TimerQueue::new(); assert_eq!(queue.next_timeout(), None); let k = queue.insert(0, ()); assert_eq!(queue.next_timeout(), Some(0)); queue.remove(k); assert_eq!(queue.next_timeout(), None); queue.insert(1234, ()); assert!(queue.next_timeout().unwrap() > 12); queue.insert(12, ()); assert_eq!(queue.next_timeout(), Some(12)); } #[test] fn poll_boundary() { let mut queue = TimerQueue::new(); queue.insert(SLOTS as u64 - 1, 'a'); queue.insert(SLOTS as u64, 'b'); assert_eq!(queue.poll(SLOTS as u64 - 2), None); assert_eq!(queue.poll(SLOTS as u64 - 1), Some('a')); assert_eq!(queue.poll(SLOTS as u64 - 1), None); assert_eq!(queue.poll(SLOTS as u64), Some('b')); } #[test] /// Validate that `reset` properly updates intrusive list links fn reset_list_middle() { let mut queue = TimerQueue::new(); let slot = SLOTS as u64 / 2; let a = queue.insert(slot, ()); let b = queue.insert(slot, ()); let c = queue.insert(slot, ()); queue.reset(b, slot + 1); assert_eq!(queue.levels[0].get(slot as usize + 1), Some(b)); assert_eq!(queue.timers[b.0].prev, None); assert_eq!(queue.timers[b.0].next, None); assert_eq!(queue.levels[0].get(slot as usize), Some(c)); assert_eq!(queue.timers[c.0].prev, None); assert_eq!(queue.timers[c.0].next, Some(a)); assert_eq!(queue.timers[a.0].prev, Some(c)); assert_eq!(queue.timers[a.0].next, None); } proptest! { #[test] fn poll(ts in times()) { let mut queue = TimerQueue::new(); let mut time_values = HashMap::>::new(); for (i, t) in ts.into_iter().enumerate() { queue.insert(t, i); time_values.entry(t).or_default().push(i); } let mut time_values = time_values.into_iter().collect::)>>(); time_values.sort_unstable_by_key(|&(t, _)| t); for &(t, ref is) in &time_values { assert!(queue.next_timeout().unwrap() <= t); if t > 0 { assert_eq!(queue.poll(t-1), None); } let mut values = Vec::new(); while let Some(i) = queue.poll(t) { values.push(i); } assert_eq!(values.len(), is.len()); for i in is { assert!(values.contains(i)); } } } #[test] fn reset(ts_a in times(), ts_b in times()) { let mut queue = TimerQueue::new(); let timers = ts_a.map(|t| queue.insert(t, ())); for (timer, t) in timers.into_iter().zip(ts_b) { queue.reset(timer, t); } let mut n = 0; while let Some(()) = queue.poll(u64::MAX) { n += 1; } assert_eq!(n, timers.len()); } #[test] fn index_start_consistency(a in time(), b in time()) { let base = a.min(b); let t = a.max(b); let (level, slot) = timer_index(base, t); let start = slot_start(base, level, slot); assert!(start <= t); if let Some(end) = start.checked_add((SLOTS as u64).pow(level as u32)) { assert!(end > t); } else { // Slot contains u64::MAX assert!(start >= slot_start(0, LEVELS - 1, 15)); if level == LEVELS - 1 { assert_eq!(slot, 15); } else { assert_eq!(slot, SLOTS - 1); } } } } #[test] #[cfg(feature = "serde")] fn serialization() { const VALUES: [(u64, usize); 17] = [ (23, 5132), (87, 6), (45, 7839), (122, 345), (67, 12333), (34, 8), (90, 234), (151, 82290), (56, 32), (78, 567), (19, 345), (22, 78), (33, 890), (44, 123), (51235, 6), (66, 89), (727, 890), ]; let mut queue = TimerQueue::::new(); for (t, v) in VALUES { queue.insert(t, v); } let serialized: Vec = bincode::serialize(&queue).expect("Serialization failed"); let mut deserialized: TimerQueue = bincode::deserialize(&serialized).expect("Deserialization failed"); loop { let r1 = queue.poll(u64::MAX); let r2 = deserialized.poll(u64::MAX); assert!(r1 == r2); if r1.is_none() { break; } } } /// Generates a time whose level/slot is more or less uniformly distributed fn time() -> impl Strategy { ((0..LEVELS as u32), (0..SLOTS as u64)).prop_perturb(|(level, mut slot), mut rng| { if level == LEVELS as u32 - 1 { slot %= 16; } let slot_size = (SLOTS as u64).pow(level); let slot_start = slot * slot_size; let slot_end = (slot + 1).saturating_mul(slot_size); rng.gen_range(slot_start..slot_end) }) } #[rustfmt::skip] fn times() -> impl Strategy { [time(), time(), time(), time(), time(), time(), time(), time(), time(), time(), time(), time(), time(), time(), time(), time()] } }