1 use crate::runtime::time::{TimerHandle, TimerShared};
2 use crate::time::error::InsertError;
3
4 mod level;
5 pub(crate) use self::level::Expiration;
6 use self::level::Level;
7
8 use std::{array, ptr::NonNull};
9
10 use super::EntryList;
11
12 /// Timing wheel implementation.
13 ///
14 /// This type provides the hashed timing wheel implementation that backs `Timer`
15 /// and `DelayQueue`.
16 ///
17 /// The structure is generic over `T: Stack`. This allows handling timeout data
18 /// being stored on the heap or in a slab. In order to support the latter case,
19 /// the slab must be passed into each function allowing the implementation to
20 /// lookup timer entries.
21 ///
22 /// See `Timer` documentation for some implementation notes.
23 #[derive(Debug)]
24 pub(crate) struct Wheel {
25 /// The number of milliseconds elapsed since the wheel started.
26 elapsed: u64,
27
28 /// Timer wheel.
29 ///
30 /// Levels:
31 ///
32 /// * 1 ms slots / 64 ms range
33 /// * 64 ms slots / ~ 4 sec range
34 /// * ~ 4 sec slots / ~ 4 min range
35 /// * ~ 4 min slots / ~ 4 hr range
36 /// * ~ 4 hr slots / ~ 12 day range
37 /// * ~ 12 day slots / ~ 2 yr range
38 levels: Box<[Level; NUM_LEVELS]>,
39
40 /// Entries queued for firing
41 pending: EntryList,
42 }
43
44 /// Number of levels. Each level has 64 slots. By using 6 levels with 64 slots
45 /// each, the timer is able to track time up to 2 years into the future with a
46 /// precision of 1 millisecond.
47 const NUM_LEVELS: usize = 6;
48
49 /// The maximum duration of a `Sleep`.
50 pub(super) const MAX_DURATION: u64 = (1 << (6 * NUM_LEVELS)) - 1;
51
52 impl Wheel {
53 /// Creates a new timing wheel.
new() -> Wheel54 pub(crate) fn new() -> Wheel {
55 Wheel {
56 elapsed: 0,
57 levels: Box::new(array::from_fn(Level::new)),
58 pending: EntryList::new(),
59 }
60 }
61
62 /// Returns the number of milliseconds that have elapsed since the timing
63 /// wheel's creation.
elapsed(&self) -> u6464 pub(crate) fn elapsed(&self) -> u64 {
65 self.elapsed
66 }
67
68 /// Inserts an entry into the timing wheel.
69 ///
70 /// # Arguments
71 ///
72 /// * `item`: The item to insert into the wheel.
73 ///
74 /// # Return
75 ///
76 /// Returns `Ok` when the item is successfully inserted, `Err` otherwise.
77 ///
78 /// `Err(Elapsed)` indicates that `when` represents an instant that has
79 /// already passed. In this case, the caller should fire the timeout
80 /// immediately.
81 ///
82 /// `Err(Invalid)` indicates an invalid `when` argument as been supplied.
83 ///
84 /// # Safety
85 ///
86 /// This function registers item into an intrusive linked list. The caller
87 /// must ensure that `item` is pinned and will not be dropped without first
88 /// being deregistered.
insert( &mut self, item: TimerHandle, ) -> Result<u64, (TimerHandle, InsertError)>89 pub(crate) unsafe fn insert(
90 &mut self,
91 item: TimerHandle,
92 ) -> Result<u64, (TimerHandle, InsertError)> {
93 let when = item.sync_when();
94
95 if when <= self.elapsed {
96 return Err((item, InsertError::Elapsed));
97 }
98
99 // Get the level at which the entry should be stored
100 let level = self.level_for(when);
101
102 unsafe {
103 self.levels[level].add_entry(item);
104 }
105
106 debug_assert!({
107 self.levels[level]
108 .next_expiration(self.elapsed)
109 .map(|e| e.deadline >= self.elapsed)
110 .unwrap_or(true)
111 });
112
113 Ok(when)
114 }
115
116 /// Removes `item` from the timing wheel.
remove(&mut self, item: NonNull<TimerShared>)117 pub(crate) unsafe fn remove(&mut self, item: NonNull<TimerShared>) {
118 unsafe {
119 let when = item.as_ref().cached_when();
120 if when == u64::MAX {
121 self.pending.remove(item);
122 } else {
123 debug_assert!(
124 self.elapsed <= when,
125 "elapsed={}; when={}",
126 self.elapsed,
127 when
128 );
129
130 let level = self.level_for(when);
131 self.levels[level].remove_entry(item);
132 }
133 }
134 }
135
136 /// Instant at which to poll.
poll_at(&self) -> Option<u64>137 pub(crate) fn poll_at(&self) -> Option<u64> {
138 self.next_expiration().map(|expiration| expiration.deadline)
139 }
140
141 /// Advances the timer up to the instant represented by `now`.
poll(&mut self, now: u64) -> Option<TimerHandle>142 pub(crate) fn poll(&mut self, now: u64) -> Option<TimerHandle> {
143 loop {
144 if let Some(handle) = self.pending.pop_back() {
145 return Some(handle);
146 }
147
148 match self.next_expiration() {
149 Some(ref expiration) if expiration.deadline <= now => {
150 self.process_expiration(expiration);
151
152 self.set_elapsed(expiration.deadline);
153 }
154 _ => {
155 // in this case the poll did not indicate an expiration
156 // _and_ we were not able to find a next expiration in
157 // the current list of timers. advance to the poll's
158 // current time and do nothing else.
159 self.set_elapsed(now);
160 break;
161 }
162 }
163 }
164
165 self.pending.pop_back()
166 }
167
168 /// Returns the instant at which the next timeout expires.
next_expiration(&self) -> Option<Expiration>169 fn next_expiration(&self) -> Option<Expiration> {
170 if !self.pending.is_empty() {
171 // Expire immediately as we have things pending firing
172 return Some(Expiration {
173 level: 0,
174 slot: 0,
175 deadline: self.elapsed,
176 });
177 }
178
179 // Check all levels
180 for (level_num, level) in self.levels.iter().enumerate() {
181 if let Some(expiration) = level.next_expiration(self.elapsed) {
182 // There cannot be any expirations at a higher level that happen
183 // before this one.
184 debug_assert!(self.no_expirations_before(level_num + 1, expiration.deadline));
185
186 return Some(expiration);
187 }
188 }
189
190 None
191 }
192
193 /// Returns the tick at which this timer wheel next needs to perform some
194 /// processing, or None if there are no timers registered.
next_expiration_time(&self) -> Option<u64>195 pub(super) fn next_expiration_time(&self) -> Option<u64> {
196 self.next_expiration().map(|ex| ex.deadline)
197 }
198
199 /// Used for debug assertions
no_expirations_before(&self, start_level: usize, before: u64) -> bool200 fn no_expirations_before(&self, start_level: usize, before: u64) -> bool {
201 let mut res = true;
202
203 for level in &self.levels[start_level..] {
204 if let Some(e2) = level.next_expiration(self.elapsed) {
205 if e2.deadline < before {
206 res = false;
207 }
208 }
209 }
210
211 res
212 }
213
214 /// iteratively find entries that are between the wheel's current
215 /// time and the expiration time. for each in that population either
216 /// queue it for notification (in the case of the last level) or tier
217 /// it down to the next level (in all other cases).
process_expiration(&mut self, expiration: &Expiration)218 pub(crate) fn process_expiration(&mut self, expiration: &Expiration) {
219 // Note that we need to take _all_ of the entries off the list before
220 // processing any of them. This is important because it's possible that
221 // those entries might need to be reinserted into the same slot.
222 //
223 // This happens only on the highest level, when an entry is inserted
224 // more than MAX_DURATION into the future. When this happens, we wrap
225 // around, and process some entries a multiple of MAX_DURATION before
226 // they actually need to be dropped down a level. We then reinsert them
227 // back into the same position; we must make sure we don't then process
228 // those entries again or we'll end up in an infinite loop.
229 let mut entries = self.take_entries(expiration);
230
231 while let Some(item) = entries.pop_back() {
232 if expiration.level == 0 {
233 debug_assert_eq!(unsafe { item.cached_when() }, expiration.deadline);
234 }
235
236 // Try to expire the entry; this is cheap (doesn't synchronize) if
237 // the timer is not expired, and updates cached_when.
238 match unsafe { item.mark_pending(expiration.deadline) } {
239 Ok(()) => {
240 // Item was expired
241 self.pending.push_front(item);
242 }
243 Err(expiration_tick) => {
244 let level = level_for(expiration.deadline, expiration_tick);
245 unsafe {
246 self.levels[level].add_entry(item);
247 }
248 }
249 }
250 }
251 }
252
set_elapsed(&mut self, when: u64)253 fn set_elapsed(&mut self, when: u64) {
254 assert!(
255 self.elapsed <= when,
256 "elapsed={:?}; when={:?}",
257 self.elapsed,
258 when
259 );
260
261 if when > self.elapsed {
262 self.elapsed = when;
263 }
264 }
265
266 /// Obtains the list of entries that need processing for the given expiration.
take_entries(&mut self, expiration: &Expiration) -> EntryList267 fn take_entries(&mut self, expiration: &Expiration) -> EntryList {
268 self.levels[expiration.level].take_slot(expiration.slot)
269 }
270
level_for(&self, when: u64) -> usize271 fn level_for(&self, when: u64) -> usize {
272 level_for(self.elapsed, when)
273 }
274 }
275
level_for(elapsed: u64, when: u64) -> usize276 fn level_for(elapsed: u64, when: u64) -> usize {
277 const SLOT_MASK: u64 = (1 << 6) - 1;
278
279 // Mask in the trailing bits ignored by the level calculation in order to cap
280 // the possible leading zeros
281 let mut masked = elapsed ^ when | SLOT_MASK;
282
283 if masked >= MAX_DURATION {
284 // Fudge the timer into the top level
285 masked = MAX_DURATION - 1;
286 }
287
288 let leading_zeros = masked.leading_zeros() as usize;
289 let significant = 63 - leading_zeros;
290
291 significant / NUM_LEVELS
292 }
293
294 #[cfg(all(test, not(loom)))]
295 mod test {
296 use super::*;
297
298 #[test]
test_level_for()299 fn test_level_for() {
300 for pos in 0..64 {
301 assert_eq!(0, level_for(0, pos), "level_for({pos}) -- binary = {pos:b}");
302 }
303
304 for level in 1..5 {
305 for pos in level..64 {
306 let a = pos * 64_usize.pow(level as u32);
307 assert_eq!(
308 level,
309 level_for(0, a as u64),
310 "level_for({a}) -- binary = {a:b}"
311 );
312
313 if pos > level {
314 let a = a - 1;
315 assert_eq!(
316 level,
317 level_for(0, a as u64),
318 "level_for({a}) -- binary = {a:b}"
319 );
320 }
321
322 if pos < 64 {
323 let a = a + 1;
324 assert_eq!(
325 level,
326 level_for(0, a as u64),
327 "level_for({a}) -- binary = {a:b}"
328 );
329 }
330 }
331 }
332 }
333 }
334