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}