Skip to main content

sozu_lib/
timer.rs

1//! Timer based on timing wheels
2//!
3//! code imported from mio-extras
4//! License: MIT or Apache 2.0
5use std::{
6    cmp,
7    fmt::Display,
8    time::{Duration, Instant},
9};
10
11use mio::Token;
12use slab::Slab;
13
14use crate::server::TIMER;
15
16// Conversion utilities
17mod convert {
18    use std::time::Duration;
19
20    /// Convert a `Duration` to milliseconds, rounding up and saturating at
21    /// `u64::MAX`.
22    ///
23    /// The saturating is fine because `u64::MAX` milliseconds are still many
24    /// million years.
25    pub fn millis(duration: Duration) -> u64 {
26        u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
27    }
28}
29
30/// A timer.
31///
32/// Typical usage goes like this:
33///
34/// * register the timer with a `mio::Poll`.
35/// * set a timeout, by calling `Timer::set_timeout`.  Here you provide some
36///   state to be associated with this timeout.
37/// * poll the `Poll`, to learn when a timeout has occurred.
38/// * retrieve state associated with the timeout by calling `Timer::poll`.
39///
40/// You can omit use of the `Poll` altogether, if you like, and just poll the
41/// `Timer` directly.
42pub struct Timer<T> {
43    // Size of each tick in milliseconds
44    tick_ms: u64,
45    // Slab of timeout entries
46    entries: Slab<Entry<T>>,
47    // Timeout wheel. Each tick, the timer will look at the next slot for
48    // timeouts that match the current tick.
49    wheel: Vec<WheelEntry>,
50    // Tick 0's time instant
51    start: Instant,
52    // The current tick
53    tick: Tick,
54    // The next entry to possibly timeout
55    next: Token,
56    // Masks the target tick to get the slot
57    mask: u64,
58}
59
60/// Used to create a `Timer`.
61pub struct Builder {
62    // Approximate duration of each tick
63    tick: Duration,
64    // Number of slots in the timer wheel
65    num_slots: usize,
66    // Max number of timeouts that can be in flight at a given time.
67    capacity: usize,
68}
69
70/// A timeout, as returned by `Timer::set_timeout`.
71///
72/// Use this as the argument to `Timer::cancel_timeout`, to cancel this timeout.
73#[derive(Clone, Debug)]
74pub struct Timeout {
75    // Reference into the timer entry slab
76    token: Token,
77    // Tick that it should match up with
78    tick: u64,
79}
80
81#[derive(Clone, Debug)]
82pub struct TimeoutContainer {
83    // mark it as an option, so we do not try to cancel a timeout multiple times
84    timeout: Option<Timeout>,
85    duration: Duration,
86    token: Option<Token>,
87    /// Absolute instant the currently armed wheel entry is meant to fire at,
88    /// recorded at arm time; `None` whenever no entry is armed.
89    ///
90    /// The wheel is duration-based and rounds a delay to the NEAREST tick
91    /// ([`duration_to_tick`]), so an entry armed for deadline `D` is delivered
92    /// from `tick * round(D / tick) - tick/2` onwards. With the default 100 ms
93    /// tick the earliness is `(D_ms + 50) mod 100`, i.e. up to **99 ms** — a
94    /// full tick minus a millisecond, not half a tick. A holder that wants to
95    /// distinguish "the deadline really elapsed"
96    /// from "the wheel rounded down" needs the absolute instant the wheel
97    /// itself does not keep. That is the whole purpose of this field: it is
98    /// the mirror of what the wheel holds, not a second clock, and it is
99    /// stamped with [`Instant::now`] at exactly the sites that touch the
100    /// wheel, so it cannot drift from the entry it describes.
101    ///
102    /// It is deliberately cleared by [`Self::triggered`]: the expiry that made
103    /// the caller run consumed the wheel entry, so the container holds nothing
104    /// until it is armed again. Callers that memoize on "the deadline has not
105    /// moved" depend on that clearing — without it, an early delivery is a
106    /// lost wakeup.
107    deadline: Option<Instant>,
108}
109
110/// `now + duration`, saturating at the far end of the [`Instant`] range
111/// instead of panicking. A duration large enough to overflow can only come
112/// from a configuration value, and a session that never times out is a far
113/// better outcome than a worker that aborts.
114fn deadline_from(duration: Duration) -> Instant {
115    let now = Instant::now();
116    now.checked_add(duration).unwrap_or(now)
117}
118
119impl TimeoutContainer {
120    pub fn new(duration: Duration, token: Token) -> TimeoutContainer {
121        let timeout = TIMER.with(|timer| timer.borrow_mut().set_timeout(duration, token));
122        TimeoutContainer {
123            timeout: Some(timeout),
124            duration,
125            token: Some(token),
126            deadline: Some(deadline_from(duration)),
127        }
128    }
129
130    pub fn new_empty(duration: Duration) -> TimeoutContainer {
131        TimeoutContainer {
132            timeout: None,
133            duration,
134            token: None,
135            deadline: None,
136        }
137    }
138
139    pub fn take(&mut self) -> TimeoutContainer {
140        TimeoutContainer {
141            timeout: self.timeout.take(),
142            duration: self.duration,
143            token: self.token.take(),
144            deadline: self.deadline.take(),
145        }
146    }
147
148    /// must be called when a timeout was triggered, to prevent errors when canceling
149    pub fn triggered(&mut self) {
150        let _ = self.timeout.take();
151        self.deadline = None;
152    }
153
154    pub fn set(&mut self, token: Token) {
155        if let Some(timeout) = self.timeout.take() {
156            TIMER.with(|timer| timer.borrow_mut().cancel_timeout(&timeout));
157        }
158
159        let timeout = TIMER.with(|timer| timer.borrow_mut().set_timeout(self.duration, token));
160
161        self.timeout = Some(timeout);
162        self.token = Some(token);
163        self.deadline = Some(deadline_from(self.duration));
164    }
165
166    /// Arm the entry so it fires at `deadline` rather than one full
167    /// [`Self::duration`] from now, leaving the configured duration untouched.
168    ///
169    /// This is what a holder uses to put an entry BACK after the wheel handed
170    /// it over early: re-arming with [`Self::set`] would push the deadline out
171    /// by a fresh full duration, so an entry delivered a few tens of
172    /// milliseconds before a 60 s timeout would fire at ~120 s — silently
173    /// doubling the operator's configured timeout on every early delivery.
174    ///
175    /// A deadline that has already passed arms the shortest entry the wheel
176    /// can express (`set_timeout_at` forces at least one tick ahead), so the
177    /// caller is called back promptly rather than never — and, because that
178    /// clamp exists, never spins: a zero-length delay still costs a whole tick.
179    pub fn set_at(&mut self, token: Token, deadline: Instant) {
180        if let Some(timeout) = self.timeout.take() {
181            TIMER.with(|timer| timer.borrow_mut().cancel_timeout(&timeout));
182        }
183
184        let delay = deadline.saturating_duration_since(Instant::now());
185        let timeout = TIMER.with(|timer| timer.borrow_mut().set_timeout(delay, token));
186
187        self.timeout = Some(timeout);
188        self.token = Some(token);
189        self.deadline = Some(deadline);
190    }
191
192    /// warning: this does not reset the timer
193    pub fn set_duration(&mut self, duration: Duration) {
194        self.duration = duration;
195
196        if let Some(timeout) = self.timeout.take() {
197            TIMER.with(|timer| timer.borrow_mut().cancel_timeout(&timeout));
198        }
199
200        if let Some(token) = self.token {
201            self.timeout =
202                Some(TIMER.with(|timer| timer.borrow_mut().set_timeout(self.duration, token)));
203            self.deadline = Some(deadline_from(self.duration));
204        } else {
205            self.deadline = None;
206        }
207    }
208
209    pub fn duration(&self) -> Duration {
210        self.duration
211    }
212
213    /// Adopt a new configured duration **without touching the armed entry**.
214    ///
215    /// [`Self::set_duration`] cancels and re-arms; an owner that drives arming
216    /// itself through [`Self::set_at`] must not pay that, and must not have its
217    /// entry moved behind its back. The duration still has to be kept current,
218    /// because a container handed on to another protocol state — the WebSocket
219    /// upgrade path hands the mux's containers to `Pipe`, which calls
220    /// [`Self::reset`] — re-arms from it.
221    pub fn retune(&mut self, duration: Duration) {
222        self.duration = duration;
223    }
224
225    /// Whether a wheel entry is currently held.
226    ///
227    /// Distinct from `deadline().is_some()` only while the two are out of step,
228    /// which is exactly the state an owner that memoizes on the deadline must
229    /// be able to see: after [`Self::triggered`] both are cleared together, but
230    /// an owner keeping its own deadline mirror can believe an entry is armed
231    /// when the wheel has already handed it over. That belief is the lost
232    /// wakeup, so the check is available rather than inferred.
233    pub fn is_armed(&self) -> bool {
234        self.timeout.is_some()
235    }
236
237    /// The absolute instant the armed entry is meant to fire at, or `None`
238    /// when nothing is armed. See the field documentation: this is the mirror
239    /// of the wheel entry, and the only way to tell an early delivery from a
240    /// real expiry.
241    pub fn deadline(&self) -> Option<Instant> {
242        self.deadline
243    }
244
245    pub fn cancel(&mut self) -> bool {
246        self.deadline = None;
247        match self.timeout.take() {
248            None => {
249                //error!("cannot cancel non existing timeout");
250                //error!("self.duration was {:?}", self.duration);
251                false
252            }
253            Some(timeout) => {
254                TIMER.with(|timer| timer.borrow_mut().cancel_timeout(&timeout));
255                true
256            }
257        }
258    }
259
260    // Reset the timeout to its optional timeout, or to its defined duration
261    pub fn reset(&mut self) -> bool {
262        match self.timeout.take() {
263            None => {
264                if let Some(token) = self.token {
265                    self.timeout = Some(
266                        TIMER.with(|timer| timer.borrow_mut().set_timeout(self.duration, token)),
267                    );
268                } else {
269                    //error!("cannot reset non existing timeout");
270                    self.deadline = None;
271                    return false;
272                }
273            }
274            Some(timeout) => {
275                self.timeout =
276                    TIMER.with(|timer| timer.borrow_mut().reset_timeout(&timeout, self.duration));
277            }
278        };
279        self.deadline = self.timeout.is_some().then(|| deadline_from(self.duration));
280        self.timeout.is_some()
281    }
282}
283
284impl Display for TimeoutContainer {
285    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
286        write!(f, "{:?}", self.duration)
287    }
288}
289
290impl std::ops::Drop for TimeoutContainer {
291    fn drop(&mut self) {
292        if self.cancel() {
293            debug!(
294                "Cancel a dangling timeout that haven't be handled in session lifecycle, token ({:?}), duration {}",
295                self.token, self
296            );
297        }
298    }
299}
300
301#[derive(Copy, Clone, Debug)]
302struct WheelEntry {
303    next_tick: Tick,
304    head: Token,
305}
306
307// Doubly linked list of timer entries. Allows for efficient insertion /
308// removal of timeouts.
309struct Entry<T> {
310    state: T,
311    links: EntryLinks,
312}
313
314#[derive(Copy, Clone)]
315struct EntryLinks {
316    tick: Tick,
317    prev: Token,
318    next: Token,
319}
320
321type Tick = u64;
322const TICK_MAX: Tick = u64::MAX;
323const EMPTY: Token = Token(usize::MAX);
324
325impl Builder {
326    /// Set the tick duration.  Default is 100ms.
327    pub fn tick_duration(mut self, duration: Duration) -> Builder {
328        self.tick = duration;
329        self
330    }
331
332    /// Set the number of slots.  Default is 256.
333    pub fn num_slots(mut self, num_slots: usize) -> Builder {
334        self.num_slots = num_slots;
335        self
336    }
337
338    /// Set the capacity.  Default is 65536.
339    pub fn capacity(mut self, capacity: usize) -> Builder {
340        self.capacity = capacity;
341        self
342    }
343
344    /// Build a `Timer` with the parameters set on this `Builder`.
345    pub fn build<T>(self) -> Timer<T> {
346        Timer::new(
347            convert::millis(self.tick),
348            self.num_slots,
349            self.capacity,
350            Instant::now(),
351        )
352    }
353}
354
355impl Default for Builder {
356    fn default() -> Builder {
357        Builder {
358            tick: Duration::from_millis(100),
359            num_slots: 1 << 8,
360            capacity: 1 << 16,
361        }
362    }
363}
364
365impl<T> Timer<T> {
366    fn new(tick_ms: u64, num_slots: usize, capacity: usize, start: Instant) -> Timer<T> {
367        let num_slots = num_slots.next_power_of_two();
368        let capacity = capacity.next_power_of_two();
369        let mask = (num_slots as u64) - 1;
370        let wheel = std::iter::repeat_n(
371            WheelEntry {
372                next_tick: TICK_MAX,
373                head: EMPTY,
374            },
375            num_slots,
376        )
377        .collect();
378
379        Timer {
380            tick_ms,
381            entries: Slab::with_capacity(capacity),
382            wheel,
383            start,
384            tick: 0,
385            next: EMPTY,
386            mask,
387        }
388    }
389
390    /// Set a timeout.
391    ///
392    /// When the timeout occurs, the given state becomes available via `poll`.
393    pub fn set_timeout(&mut self, delay_from_now: Duration, state: T) -> Timeout {
394        self.set_timeout_from(Instant::now(), delay_from_now, state)
395    }
396
397    /// [`Self::set_timeout`], counting `delay_from_now` from the caller's
398    /// `now` instead of re-reading the clock.
399    ///
400    /// A caller that already holds the instant its deadline was computed
401    /// against uses this so the entry lands on the grid point that deadline
402    /// maps to, however long ago `now` was read — and a test can place an
403    /// entry at an exact instant.
404    pub fn set_timeout_from(
405        &mut self,
406        now: Instant,
407        delay_from_now: Duration,
408        state: T,
409    ) -> Timeout {
410        let delay_from_start = now.saturating_duration_since(self.start) + delay_from_now;
411        self.set_timeout_at(delay_from_start, state)
412    }
413
414    fn set_timeout_at(&mut self, delay_from_start: Duration, state: T) -> Timeout {
415        let mut tick = duration_to_tick(delay_from_start, self.tick_ms);
416        trace!(
417            "setting timeout; delay={:?}; tick={:?}; current-tick={:?}",
418            delay_from_start, tick, self.tick
419        );
420
421        // Always target at least 1 tick in the future
422        if tick <= self.tick {
423            tick = self.tick + 1;
424        }
425
426        self.insert(tick, state)
427    }
428
429    fn insert(&mut self, tick: Tick, state: T) -> Timeout {
430        // Get the slot for the requested tick
431        let slot = (tick & self.mask) as usize;
432        let curr = self.wheel[slot];
433
434        // Insert the new entry
435        let entry = Entry::new(state, tick, curr.head);
436        let token = Token(self.entries.insert(entry));
437
438        if curr.head != EMPTY {
439            // If there was a previous entry, set its prev pointer to the new
440            // entry
441            self.entries[curr.head.into()].links.prev = token;
442        }
443
444        // Update the head slot
445        self.wheel[slot] = WheelEntry {
446            next_tick: cmp::min(tick, curr.next_tick),
447            head: token,
448        };
449
450        trace!("inserted timeout; slot={}; token={:?}", slot, token);
451
452        // Return the new timeout
453        Timeout { token, tick }
454    }
455
456    /// Resets a timeout.
457    ///
458    pub fn reset_timeout(
459        &mut self,
460        timeout: &Timeout,
461        delay_from_now: Duration,
462    ) -> Option<Timeout> {
463        self.cancel_timeout(timeout)
464            .map(|state| self.set_timeout(delay_from_now, state))
465    }
466
467    // TODO: return Result with context
468    /// Cancel a timeout.
469    ///
470    /// If the timeout has not yet occurred, the return value holds the
471    /// associated state.
472    pub fn cancel_timeout(&mut self, timeout: &Timeout) -> Option<T> {
473        let links = match self.entries.get(timeout.token.into()) {
474            Some(e) => e.links,
475            None => {
476                debug!("timeout token {:?} not found", timeout.token);
477                return None;
478            }
479        };
480
481        // Sanity check
482        if links.tick != timeout.tick {
483            return None;
484        }
485
486        self.unlink(&links, timeout.token);
487        Some(self.entries.remove(timeout.token.into()).state)
488    }
489
490    /// Poll for an expired timer.
491    ///
492    /// The return value holds the state associated with the first expired
493    /// timer, if any.
494    pub fn poll(&mut self) -> Option<T> {
495        let target_tick = current_tick(self.start, self.tick_ms);
496        self.poll_to(target_tick)
497    }
498
499    /// [`Self::poll`] as of `now` rather than the current instant, so a test
500    /// can drive the wheel from a simulated clock.
501    pub fn poll_at(&mut self, now: Instant) -> Option<T> {
502        let target_tick = duration_to_tick(now.saturating_duration_since(self.start), self.tick_ms);
503        self.poll_to(target_tick)
504    }
505
506    fn poll_to(&mut self, mut target_tick: Tick) -> Option<T> {
507        trace!(
508            "tick_to; target_tick={}; current_tick={}",
509            target_tick, self.tick
510        );
511
512        if target_tick < self.tick {
513            target_tick = self.tick;
514        }
515
516        while self.tick <= target_tick {
517            let curr = self.next;
518
519            //info!("ticking; curr={:?}", curr);
520
521            if curr == EMPTY {
522                self.tick += 1;
523
524                let slot = self.slot_for(self.tick);
525                self.next = self.wheel[slot].head;
526
527                // Handle the case when a slot has a single timeout which gets
528                // canceled before the timeout expires. In this case, the
529                // slot's head is EMPTY but there is a value for next_tick. Not
530                // resetting next_tick here causes the timer to get stuck in a
531                // loop.
532                if self.next == EMPTY {
533                    self.wheel[slot].next_tick = TICK_MAX;
534                }
535            } else {
536                let slot = self.slot_for(self.tick);
537
538                if curr == self.wheel[slot].head {
539                    self.wheel[slot].next_tick = TICK_MAX;
540                }
541
542                let links = self.entries[curr.into()].links;
543
544                if links.tick <= self.tick {
545                    trace!("triggering; token={:?}", curr);
546
547                    // Unlink will also advance self.next
548                    self.unlink(&links, curr);
549
550                    // Remove and return the token
551                    return Some(self.entries.remove(curr.into()).state);
552                } else {
553                    let next_tick = self.wheel[slot].next_tick;
554                    self.wheel[slot].next_tick = cmp::min(next_tick, links.tick);
555                    self.next = links.next;
556                }
557            }
558        }
559
560        None
561    }
562
563    fn unlink(&mut self, links: &EntryLinks, token: Token) {
564        trace!(
565            "unlinking timeout; slot={}; token={:?}",
566            self.slot_for(links.tick),
567            token
568        );
569
570        if links.prev == EMPTY {
571            let slot = self.slot_for(links.tick);
572            self.wheel[slot].head = links.next;
573        } else {
574            self.entries[links.prev.into()].links.next = links.next;
575        }
576
577        if links.next != EMPTY {
578            self.entries[links.next.into()].links.prev = links.prev;
579
580            if token == self.next {
581                self.next = links.next;
582            }
583        } else if token == self.next {
584            self.next = EMPTY;
585        }
586    }
587
588    // Next tick containing a timeout
589    fn next_tick(&self) -> Option<Tick> {
590        if self.next != EMPTY {
591            let slot = self.slot_for(self.entries[self.next.into()].links.tick);
592
593            if self.wheel[slot].next_tick == self.tick {
594                // There is data ready right now
595                return Some(self.tick);
596            }
597        }
598
599        self.wheel.iter().map(|e| e.next_tick).min()
600    }
601
602    pub fn next_poll_date(&self) -> Option<Instant> {
603        self.next_tick()
604            .map(|tick| self.start + Duration::from_millis(self.tick_ms.saturating_mul(tick)))
605    }
606
607    fn slot_for(&self, tick: Tick) -> usize {
608        (self.mask & tick) as usize
609    }
610}
611
612impl<T> Default for Timer<T> {
613    fn default() -> Timer<T> {
614        Builder::default().build()
615    }
616}
617
618/// Map a duration onto the wheel's tick grid, rounding to the NEAREST tick.
619///
620/// Not rounding UP: `elapsed` in `[tick_ms*N - tick_ms/2, tick_ms*N + tick_ms/2)`
621/// yields tick `N`, so a timeout can be delivered up to half a tick — 50 ms at
622/// the 100 ms default — BEFORE the caller asked for it *when the poll lands on
623/// the grid*, which is what [`Timer::next_poll_date`] schedules (it returns
624/// `start + tick_ms * tick`, the grid point itself).
625///
626/// The DELIVERY bound is wider, because `Timer::poll` recomputes
627/// `current_tick` from the real clock and fires everything whose tick has come:
628/// a poll anywhere in `[tick_ms*N - tick_ms/2, tick_ms*N)` already sees tick
629/// `N`. An event loop reaches that window routinely — another session's earlier
630/// entry plus loop latency, or a non-timer wakeup. Total earliness is therefore
631/// `(delay_ms + tick_ms/2) mod tick_ms`, spanning `[0, tick_ms - 1]`: up to
632/// **99 ms** at the default tick. Reason with 99 ms, not 50. That is deliberate (it
633/// halves the worst-case error instead of always overshooting), and it is why
634/// every consumer must treat an expiry as "my entry was consumed" rather than
635/// "my deadline arrived": `Timer::poll` removes the entry either way, and
636/// nothing re-arms it on the consumer's behalf. See
637/// `test_timeout_fires_up_to_half_a_tick_early` below, and the
638/// consume-then-reschedule rule in `lib/src/protocol/udp/LIFECYCLE.md` §7.
639fn duration_to_tick(elapsed: Duration, tick_ms: u64) -> Tick {
640    let elapsed_ms = convert::millis(elapsed);
641    elapsed_ms.saturating_add(tick_ms / 2) / tick_ms
642}
643
644fn current_tick(start: Instant, tick_ms: u64) -> Tick {
645    duration_to_tick(start.elapsed(), tick_ms)
646}
647
648impl<T> Entry<T> {
649    fn new(state: T, tick: u64, next: Token) -> Entry<T> {
650        Entry {
651            state,
652            links: EntryLinks {
653                tick,
654                prev: EMPTY,
655                next,
656            },
657        }
658    }
659}
660
661#[cfg(test)]
662mod test {
663    use std::time::{Duration, Instant};
664
665    use super::*;
666
667    #[test]
668    pub fn test_timeout_next_tick() {
669        let mut t = timer();
670
671        t.set_timeout_at(Duration::from_millis(100), "a");
672
673        let mut tick = ms_to_tick(&t, 50);
674        assert_eq!(None, t.poll_to(tick));
675
676        tick = ms_to_tick(&t, 100);
677        assert_eq!(Some("a"), t.poll_to(tick));
678        assert_eq!(None, t.poll_to(tick));
679
680        tick = ms_to_tick(&t, 150);
681        assert_eq!(None, t.poll_to(tick));
682
683        tick = ms_to_tick(&t, 200);
684        assert_eq!(None, t.poll_to(tick));
685
686        assert_eq!(count(&t), 0);
687    }
688
689    /// The wheel rounds a requested delay to the NEAREST tick
690    /// ([`duration_to_tick`]) — it does not round up, despite what that
691    /// function's own comment says. With the default 100 ms tick a delay in
692    /// `[100N - 50, 100N + 50)` lands in tick `N`, so an entry can be delivered
693    /// up to 50 ms BEFORE the caller's deadline when the poll lands ON the grid
694    /// — which is what this test does and what [`Timer::next_poll_date`]
695    /// schedules. A poll inside `[100N - 50, 100N)` already sees tick `N`, so
696    /// the bound a CONSUMER must tolerate is up to 99 ms; see
697    /// [`duration_to_tick`]. Either way the delivery CONSUMES the slab entry.
698    ///
699    /// Every consumer of this wheel must therefore read a wakeup as "my entry
700    /// is gone", never as "my deadline has arrived", and re-arm on its own if
701    /// it still wants one. `UdpManager::handle_timeout`
702    /// (`lib/src/protocol/udp/manager.rs`) clears `armed_deadline` on entry for
703    /// exactly this reason; `early_expiry_that_finds_nothing_due_still_rearms`
704    /// there is this test's other half.
705    ///
706    /// This pins the rounding deliberately so nobody re-derives the intuitive
707    /// but wrong "a timer never fires early" assumption. The rounding itself is
708    /// a design choice and is NOT the bug.
709    ///
710    /// To SEE THIS RED: make the rounding round up in [`duration_to_tick`] —
711    /// replace `elapsed_ms.saturating_add(tick_ms / 2) / tick_ms` with
712    /// `elapsed_ms.div_ceil(tick_ms)`. The 149 ms entry then lands in tick 2,
713    /// `poll_to` at tick 1 returns `None`, and the first assertion fails with
714    /// `left: Some("early")  right: None`.
715    #[test]
716    pub fn test_timeout_fires_up_to_half_a_tick_early() {
717        let mut t = timer();
718
719        // 149 ms against a 100 ms tick: (149 + 50) / 100 == 1.
720        t.set_timeout_at(Duration::from_millis(149), "early");
721
722        // Tick 1 is 100 ms — 49 ms BEFORE the requested deadline — and it fires.
723        assert_eq!(
724            Some("early"),
725            t.poll_to(ms_to_tick(&t, 100)),
726            "a 149ms delay must be delivered at the 100ms tick"
727        );
728
729        // ...and the entry is gone. Polling again at the real deadline yields
730        // nothing: the wheel does not re-arm on the caller's behalf. This is the
731        // lost wakeup every consumer has to handle.
732        assert_eq!(None, t.poll_to(ms_to_tick(&t, 149)));
733        assert_eq!(0, count(&t));
734
735        // The complementary half: 150 ms rounds up to tick 2, so it is NOT
736        // early. 50 ms is the exact boundary for a poll that lands ON the grid,
737        // which is what this test does (`poll_to(ms_to_tick(..))`) and what
738        // `next_poll_date` schedules. A poll inside `[100N - 50, 100N)` already
739        // sees tick `N`, so the earliness a CONSUMER must tolerate is up to
740        // 99 ms — see `duration_to_tick`.
741        let mut late = timer();
742        late.set_timeout_at(Duration::from_millis(150), "late");
743        assert_eq!(None, late.poll_to(ms_to_tick(&late, 100)));
744        assert_eq!(Some("late"), late.poll_to(ms_to_tick(&late, 200)));
745    }
746
747    /// A DELIVERED `Timeout` handle can never cancel the entry that reuses its
748    /// slab slot. The slot really is reused — `poll_to` does
749    /// `self.entries.remove(...)` — so the only thing standing between a stale
750    /// handle and someone else's timeout is `cancel_timeout`'s tick guard, and
751    /// that guard is exact rather than probabilistic:
752    ///
753    /// - [`insert`](Timer::insert) has exactly one caller,
754    ///   [`set_timeout_at`](Timer::set_timeout_at), which clamps every new entry
755    ///   to `tick > self.tick`;
756    /// - `self.tick` never decreases (`poll_to` only increments it, and floors
757    ///   `target_tick` at it);
758    /// - delivery requires `links.tick <= self.tick`.
759    ///
760    /// So a delivered handle's tick is `<= self.tick`, and every entry created
761    /// afterwards has a tick `> self.tick` — they can never be equal, whatever
762    /// the slab does with the slot. `TIMER` is a thread-local shared by every
763    /// session in a worker, so this is what makes a forgotten handle merely
764    /// untidy rather than a cross-session cancel.
765    ///
766    /// To SEE THIS RED: delete the `if links.tick != timeout.tick { return
767    /// None; }` sanity check from [`Timer::cancel_timeout`]. The stale handle
768    /// then cancels the unrelated entry occupying its slot and this fails with
769    /// `a delivered handle must not cancel the entry that reused its slot`.
770    #[test]
771    pub fn test_a_delivered_timeout_handle_cannot_cancel_its_slot_successor() {
772        let mut t = timer();
773
774        let stale = t.set_timeout_at(Duration::from_millis(100), "first");
775        assert_eq!(Some("first"), t.poll_to(ms_to_tick(&t, 100)));
776        assert_eq!(0, count(&t), "delivery must free the slab slot");
777
778        // The successor lands in the very slot the delivered entry vacated...
779        let successor = t.set_timeout_at(Duration::from_millis(300), "second");
780        assert_eq!(
781            stale.token, successor.token,
782            "precondition: the slab must reuse the delivered entry's slot"
783        );
784        // ...but never at the same tick, because `set_timeout_at` clamps every
785        // new entry past `self.tick` and the delivered one was at or below it.
786        assert!(
787            successor.tick > stale.tick,
788            "a new entry must outrank a delivered one: {} > {}",
789            successor.tick,
790            stale.tick
791        );
792
793        assert_eq!(
794            None,
795            t.cancel_timeout(&stale),
796            "a delivered handle must not cancel the entry that reused its slot"
797        );
798        assert_eq!(
799            Some("second"),
800            t.poll_to(ms_to_tick(&t, 300)),
801            "the successor must survive the stale cancel and still fire"
802        );
803    }
804
805    /// The attained maximum earliness ON THE GRID is a full half-tick — 50 ms
806    /// at the 100 ms default — not 49 ms. The rounding interval
807    /// `[100N-50, 100N+50)` is closed on the EARLY side and open on the late
808    /// side, so a 50 ms delay lands in tick 1 and is delivered at 100 ms, while
809    /// 150 ms is pushed out to tick 2. Lateness is what is capped at 49 ms.
810    ///
811    /// "On the grid" is the scope of this test: it polls at grid points, as
812    /// [`Timer::next_poll_date`] schedules. A poll inside `[100N-50, 100N)`
813    /// already sees tick `N`, so the earliness a consumer must tolerate reaches
814    /// 99 ms — see [`duration_to_tick`].
815    ///
816    /// To SEE THIS RED: change the rounding in [`duration_to_tick`] from
817    /// `saturating_add(tick_ms / 2)` to `saturating_add(tick_ms / 2 - 1)`. The
818    /// 50 ms entry then lands in tick 0, is clamped to tick 1 by
819    /// `set_timeout_at`'s "at least one tick in the future" rule, and still
820    /// fires at 100 ms — so the first assertion holds; it is the 149 ms case in
821    /// `test_timeout_fires_up_to_half_a_tick_early` that goes red. Use that test
822    /// for the rounding and this one for the boundary.
823    #[test]
824    pub fn test_maximum_earliness_is_a_full_half_tick() {
825        let mut t = timer();
826
827        // 50 ms: (50 + 50) / 100 == 1. Delivered at 100 ms — exactly 50 ms early.
828        t.set_timeout_at(Duration::from_millis(50), "half-tick-early");
829        assert_eq!(
830            Some("half-tick-early"),
831            t.poll_to(ms_to_tick(&t, 100)),
832            "a 50ms delay must be delivered at the 100ms tick: 50ms early"
833        );
834
835        // 49 ms is the far side of the same interval for tick 0, which
836        // `set_timeout_at` clamps up to tick 1 — the wheel never fires in the
837        // past, so sub-half-tick delays are the one case that is always LATE.
838        let mut clamped = timer();
839        clamped.set_timeout_at(Duration::from_millis(49), "clamped");
840        assert_eq!(Some("clamped"), clamped.poll_to(ms_to_tick(&clamped, 100)));
841    }
842
843    #[test]
844    pub fn test_clearing_timeout() {
845        let mut t = timer();
846
847        let to = t.set_timeout_at(Duration::from_millis(100), "a");
848        assert_eq!("a", t.cancel_timeout(&to).unwrap());
849
850        let mut tick = ms_to_tick(&t, 100);
851        assert_eq!(None, t.poll_to(tick));
852
853        tick = ms_to_tick(&t, 200);
854        assert_eq!(None, t.poll_to(tick));
855
856        assert_eq!(count(&t), 0);
857    }
858
859    #[test]
860    pub fn test_multiple_timeouts_same_tick() {
861        let mut t = timer();
862
863        t.set_timeout_at(Duration::from_millis(100), "a");
864        t.set_timeout_at(Duration::from_millis(100), "b");
865
866        let mut rcv = vec![];
867
868        let mut tick = ms_to_tick(&t, 100);
869        rcv.push(t.poll_to(tick).unwrap());
870        rcv.push(t.poll_to(tick).unwrap());
871
872        assert_eq!(None, t.poll_to(tick));
873
874        rcv.sort_unstable();
875        assert!(rcv == ["a", "b"], "actual={rcv:?}");
876
877        tick = ms_to_tick(&t, 200);
878        assert_eq!(None, t.poll_to(tick));
879
880        assert_eq!(count(&t), 0);
881    }
882
883    #[test]
884    pub fn test_multiple_timeouts_diff_tick() {
885        let mut t = timer();
886
887        t.set_timeout_at(Duration::from_millis(110), "a");
888        t.set_timeout_at(Duration::from_millis(220), "b");
889        t.set_timeout_at(Duration::from_millis(230), "c");
890        t.set_timeout_at(Duration::from_millis(440), "d");
891        t.set_timeout_at(Duration::from_millis(560), "e");
892
893        let mut tick = ms_to_tick(&t, 100);
894        assert_eq!(Some("a"), t.poll_to(tick));
895        assert_eq!(None, t.poll_to(tick));
896
897        tick = ms_to_tick(&t, 200);
898        assert_eq!(Some("c"), t.poll_to(tick));
899        assert_eq!(Some("b"), t.poll_to(tick));
900        assert_eq!(None, t.poll_to(tick));
901
902        tick = ms_to_tick(&t, 300);
903        assert_eq!(None, t.poll_to(tick));
904
905        tick = ms_to_tick(&t, 400);
906        assert_eq!(Some("d"), t.poll_to(tick));
907        assert_eq!(None, t.poll_to(tick));
908
909        tick = ms_to_tick(&t, 500);
910        assert_eq!(None, t.poll_to(tick));
911
912        tick = ms_to_tick(&t, 600);
913        assert_eq!(Some("e"), t.poll_to(tick));
914        assert_eq!(None, t.poll_to(tick));
915    }
916
917    #[test]
918    pub fn test_catching_up() {
919        let mut t = timer();
920
921        t.set_timeout_at(Duration::from_millis(110), "a");
922        t.set_timeout_at(Duration::from_millis(220), "b");
923        t.set_timeout_at(Duration::from_millis(230), "c");
924        t.set_timeout_at(Duration::from_millis(440), "d");
925
926        let tick = ms_to_tick(&t, 600);
927        assert_eq!(Some("a"), t.poll_to(tick));
928        assert_eq!(Some("c"), t.poll_to(tick));
929        assert_eq!(Some("b"), t.poll_to(tick));
930        assert_eq!(Some("d"), t.poll_to(tick));
931        assert_eq!(None, t.poll_to(tick));
932    }
933
934    #[test]
935    pub fn test_timeout_hash_collision() {
936        let mut t = timer();
937
938        t.set_timeout_at(Duration::from_millis(100), "a");
939        t.set_timeout_at(Duration::from_millis(100 + TICK * SLOTS as u64), "b");
940
941        let mut tick = ms_to_tick(&t, 100);
942        assert_eq!(Some("a"), t.poll_to(tick));
943        assert_eq!(1, count(&t));
944
945        tick = ms_to_tick(&t, 200);
946        assert_eq!(None, t.poll_to(tick));
947        assert_eq!(1, count(&t));
948
949        tick = ms_to_tick(&t, 100 + TICK * SLOTS as u64);
950        assert_eq!(Some("b"), t.poll_to(tick));
951        assert_eq!(0, count(&t));
952    }
953
954    #[test]
955    pub fn test_clearing_timeout_between_triggers() {
956        let mut t = timer();
957
958        let a = t.set_timeout_at(Duration::from_millis(100), "a");
959        let _ = t.set_timeout_at(Duration::from_millis(100), "b");
960        let _ = t.set_timeout_at(Duration::from_millis(200), "c");
961
962        let mut tick = ms_to_tick(&t, 100);
963        assert_eq!(Some("b"), t.poll_to(tick));
964        assert_eq!(2, count(&t));
965
966        t.cancel_timeout(&a);
967        assert_eq!(1, count(&t));
968
969        assert_eq!(None, t.poll_to(tick));
970
971        tick = ms_to_tick(&t, 200);
972        assert_eq!(Some("c"), t.poll_to(tick));
973        assert_eq!(0, count(&t));
974    }
975
976    /// `TimeoutContainer` mirrors the absolute instant its armed wheel entry
977    /// is meant to fire at, because the wheel itself keeps only a rounded
978    /// tick. Three properties, all load-bearing for the mux timeout path:
979    ///
980    /// * arming records a deadline (`set` one duration out, `set_at` at the
981    ///   caller's instant, leaving `duration` alone);
982    /// * `triggered` CLEARS it — the expiry consumed the entry, so the
983    ///   container holds nothing. A holder that memoizes on "the deadline has
984    ///   not moved" depends on this clearing; without it, an early delivery is
985    ///   a lost wakeup;
986    /// * `cancel` clears it too.
987    ///
988    /// To SEE THIS RED: drop `self.deadline = None;` from
989    /// [`TimeoutContainer::triggered`] — the `triggered` assertion below then
990    /// still reports the stale deadline.
991    #[test]
992    fn timeout_container_mirrors_the_deadline_of_its_armed_entry() {
993        let token = Token(7);
994        let duration = Duration::from_secs(60);
995        let mut container = TimeoutContainer::new_empty(duration);
996        assert_eq!(
997            container.deadline(),
998            None,
999            "an empty container arms nothing"
1000        );
1001
1002        let before = Instant::now();
1003        container.set(token);
1004        let armed = container.deadline().expect("set must record a deadline");
1005        assert!(
1006            armed >= before + duration,
1007            "set must record now + duration, got {armed:?} against {before:?} + {duration:?}"
1008        );
1009
1010        let explicit = before + Duration::from_secs(5);
1011        container.set_at(token, explicit);
1012        assert_eq!(
1013            container.deadline(),
1014            Some(explicit),
1015            "set_at must record the caller's instant verbatim"
1016        );
1017        assert_eq!(
1018            container.duration(),
1019            duration,
1020            "set_at must not disturb the configured duration"
1021        );
1022
1023        container.triggered();
1024        assert_eq!(
1025            container.deadline(),
1026            None,
1027            "a consumed entry leaves the container holding nothing"
1028        );
1029
1030        container.set(token);
1031        assert!(container.deadline().is_some(), "re-armed after triggered");
1032        container.cancel();
1033        assert_eq!(container.deadline(), None, "cancel releases the deadline");
1034    }
1035
1036    const TICK: u64 = 100;
1037    const SLOTS: usize = 16;
1038    const CAPACITY: usize = 32;
1039
1040    fn count<T>(timer: &Timer<T>) -> usize {
1041        timer.entries.len()
1042    }
1043
1044    fn timer() -> Timer<&'static str> {
1045        Timer::new(TICK, SLOTS, CAPACITY, Instant::now())
1046    }
1047
1048    fn ms_to_tick<T>(timer: &Timer<T>, ms: u64) -> u64 {
1049        ms / timer.tick_ms
1050    }
1051}