tokio_timer/timer/mod.rs
1//! Timer implementation.
2//!
3//! This module contains the types needed to run a timer.
4//!
5//! The [`Timer`] type runs the timer logic. It holds all the necessary state
6//! to track all associated [`Delay`] instances and delivering notifications
7//! once the deadlines are reached.
8//!
9//! The [`Handle`] type is a reference to a [`Timer`] instance. This type is
10//! `Clone`, `Send`, and `Sync`. This type is used to create instances of
11//! [`Delay`].
12//!
13//! The [`Now`] trait describes how to get an [`Instant`] representing the
14//! current moment in time. [`SystemNow`] is the default implementation, where
15//! [`Now::now`] is implemented by calling [`Instant::now`].
16//!
17//! [`Timer`] is generic over [`Now`]. This allows the source of time to be
18//! customized. This ability is especially useful in tests and any environment
19//! where determinism is necessary.
20//!
21//! Note, when using the Tokio runtime, the [`Timer`] does not need to be manually
22//! setup as the runtime comes pre-configured with a [`Timer`] instance.
23//!
24//! [`Timer`]: struct.Timer.html
25//! [`Handle`]: struct.Handle.html
26//! [`Delay`]: ../struct.Delay.html
27//! [`Now`]: ../clock/trait.Now.html
28//! [`Now::now`]: ../clock/trait.Now.html#method.now
29//! [`SystemNow`]: struct.SystemNow.html
30//! [`Instant`]: https://doc.rust-lang.org/std/time/struct.Instant.html
31//! [`Instant::now`]: https://doc.rust-lang.org/std/time/struct.Instant.html#method.now
32
33// This allows the usage of the old `Now` trait.
34#![allow(deprecated)]
35
36mod atomic_stack;
37mod entry;
38mod handle;
39mod now;
40mod registration;
41mod stack;
42
43use self::atomic_stack::AtomicStack;
44use self::entry::Entry;
45use self::stack::Stack;
46
47pub(crate) use self::handle::HandlePriv;
48pub use self::handle::{set_default, with_default, DefaultGuard, Handle};
49pub use self::now::{Now, SystemNow};
50pub(crate) use self::registration::Registration;
51
52use atomic::AtomicU64;
53use wheel;
54use Error;
55
56use tokio_executor::park::{Park, ParkThread, Unpark};
57
58use std::sync::atomic::AtomicUsize;
59use std::sync::atomic::Ordering::SeqCst;
60use std::sync::Arc;
61use std::time::{Duration, Instant};
62use std::usize;
63use std::{cmp, fmt};
64
65/// Timer implementation that drives [`Delay`], [`Interval`], and [`Timeout`].
66///
67/// A `Timer` instance tracks the state necessary for managing time and
68/// notifying the [`Delay`] instances once their deadlines are reached.
69///
70/// It is expected that a single `Timer` instance manages many individual
71/// [`Delay`] instances. The `Timer` implementation is thread-safe and, as such,
72/// is able to handle callers from across threads.
73///
74/// Callers do not use `Timer` directly to create [`Delay`] instances. Instead,
75/// [`Handle`][Handle.struct] is used. A handle for the timer instance is obtained by calling
76/// [`handle`]. [`Handle`][Handle.struct] is the type that implements `Clone` and is `Send +
77/// Sync`.
78///
79/// After creating the `Timer` instance, the caller must repeatedly call
80/// [`turn`]. The timer will perform no work unless [`turn`] is called
81/// repeatedly.
82///
83/// The `Timer` has a resolution of one millisecond. Any unit of time that falls
84/// between milliseconds are rounded up to the next millisecond.
85///
86/// When the `Timer` instance is dropped, any outstanding [`Delay`] instance that
87/// has not elapsed will be notified with an error. At this point, calling
88/// `poll` on the [`Delay`] instance will result in `Err` being returned.
89///
90/// # Implementation
91///
92/// `Timer` is based on the [paper by Varghese and Lauck][paper].
93///
94/// A hashed timing wheel is a vector of slots, where each slot handles a time
95/// slice. As time progresses, the timer walks over the slot for the current
96/// instant, and processes each entry for that slot. When the timer reaches the
97/// end of the wheel, it starts again at the beginning.
98///
99/// The `Timer` implementation maintains six wheels arranged in a set of levels.
100/// As the levels go up, the slots of the associated wheel represent larger
101/// intervals of time. At each level, the wheel has 64 slots. Each slot covers a
102/// range of time equal to the wheel at the lower level. At level zero, each
103/// slot represents one millisecond of time.
104///
105/// The wheels are:
106///
107/// * Level 0: 64 x 1 millisecond slots.
108/// * Level 1: 64 x 64 millisecond slots.
109/// * Level 2: 64 x ~4 second slots.
110/// * Level 3: 64 x ~4 minute slots.
111/// * Level 4: 64 x ~4 hour slots.
112/// * Level 5: 64 x ~12 day slots.
113///
114/// When the timer processes entries at level zero, it will notify all the
115/// [`Delay`] instances as their deadlines have been reached. For all higher
116/// levels, all entries will be redistributed across the wheel at the next level
117/// down. Eventually, as time progresses, entries will [`Delay`] instances will
118/// either be canceled (dropped) or their associated entries will reach level
119/// zero and be notified.
120///
121/// [`Delay`]: ../struct.Delay.html
122/// [`Interval`]: ../struct.Interval.html
123/// [`Timeout`]: ../struct.Timeout.html
124/// [paper]: http://www.cs.columbia.edu/~nahum/w6998/papers/ton97-timing-wheels.pdf
125/// [`handle`]: #method.handle
126/// [`turn`]: #method.turn
127/// [Handle.struct]: struct.Handle.html
128#[derive(Debug)]
129pub struct Timer<T, N = SystemNow> {
130 /// Shared state
131 inner: Arc<Inner>,
132
133 /// Timer wheel
134 wheel: wheel::Wheel<Stack>,
135
136 /// Thread parker. The `Timer` park implementation delegates to this.
137 park: T,
138
139 /// Source of "now" instances
140 now: N,
141}
142
143/// Return value from the `turn` method on `Timer`.
144///
145/// Currently this value doesn't actually provide any functionality, but it may
146/// in the future give insight into what happened during `turn`.
147#[derive(Debug)]
148pub struct Turn(());
149
150/// Timer state shared between `Timer`, `Handle`, and `Registration`.
151pub(crate) struct Inner {
152 /// The instant at which the timer started running.
153 start: Instant,
154
155 /// The last published timer `elapsed` value.
156 elapsed: AtomicU64,
157
158 /// Number of active timeouts
159 num: AtomicUsize,
160
161 /// Head of the "process" linked list.
162 process: AtomicStack,
163
164 /// Unparks the timer thread.
165 unpark: Box<dyn Unpark>,
166}
167
168/// Maximum number of timeouts the system can handle concurrently.
169const MAX_TIMEOUTS: usize = usize::MAX >> 1;
170
171// ===== impl Timer =====
172
173impl<T> Timer<T>
174where
175 T: Park,
176{
177 /// Create a new `Timer` instance that uses `park` to block the current
178 /// thread.
179 ///
180 /// Once the timer has been created, a handle can be obtained using
181 /// [`handle`]. The handle is used to create `Delay` instances.
182 ///
183 /// Use `default` when constructing a `Timer` using the default `park`
184 /// instance.
185 ///
186 /// [`handle`]: #method.handle
187 pub fn new(park: T) -> Self {
188 Timer::new_with_now(park, SystemNow::new())
189 }
190}
191
192impl<T, N> Timer<T, N> {
193 /// Returns a reference to the underlying `Park` instance.
194 pub fn get_park(&self) -> &T {
195 &self.park
196 }
197
198 /// Returns a mutable reference to the underlying `Park` instance.
199 pub fn get_park_mut(&mut self) -> &mut T {
200 &mut self.park
201 }
202}
203
204impl<T, N> Timer<T, N>
205where
206 T: Park,
207 N: Now,
208{
209 /// Create a new `Timer` instance that uses `park` to block the current
210 /// thread and `now` to get the current `Instant`.
211 ///
212 /// Specifying the source of time is useful when testing.
213 pub fn new_with_now(park: T, mut now: N) -> Self {
214 let unpark = Box::new(park.unpark());
215
216 Timer {
217 inner: Arc::new(Inner::new(now.now(), unpark)),
218 wheel: wheel::Wheel::new(),
219 park,
220 now,
221 }
222 }
223
224 /// Returns a handle to the timer.
225 ///
226 /// The `Handle` is how `Delay` instances are created. The `Delay` instances
227 /// can either be created directly or the `Handle` instance can be passed to
228 /// `with_default`, setting the timer as the default timer for the execution
229 /// context.
230 pub fn handle(&self) -> Handle {
231 Handle::new(Arc::downgrade(&self.inner))
232 }
233
234 /// Performs one iteration of the timer loop.
235 ///
236 /// This function must be called repeatedly in order for the `Timer`
237 /// instance to make progress. This is where the work happens.
238 ///
239 /// The `Timer` will use the `Park` instance that was specified in [`new`]
240 /// to block the current thread until the next `Delay` instance elapses. One
241 /// call to `turn` results in at most one call to `park.park()`.
242 ///
243 /// # Return
244 ///
245 /// On success, `Ok(Turn)` is returned, where `Turn` is a placeholder type
246 /// that currently does nothing but may, in the future, have functions add
247 /// to provide information about the call to `turn`.
248 ///
249 /// If the call to `park.park()` fails, then `Err` is returned with the
250 /// error.
251 ///
252 /// [`new`]: #method.new
253 pub fn turn(&mut self, max_wait: Option<Duration>) -> Result<Turn, T::Error> {
254 match max_wait {
255 Some(timeout) => self.park_timeout(timeout)?,
256 None => self.park()?,
257 }
258
259 Ok(Turn(()))
260 }
261
262 /// Converts an `Expiration` to an `Instant`.
263 fn expiration_instant(&self, when: u64) -> Instant {
264 self.inner.start + Duration::from_millis(when)
265 }
266
267 /// Run timer related logic
268 fn process(&mut self) {
269 let now = ::ms(self.now.now() - self.inner.start, ::Round::Down);
270 let mut poll = wheel::Poll::new(now);
271
272 while let Some(entry) = self.wheel.poll(&mut poll, &mut ()) {
273 let when = entry.when_internal().expect("invalid internal entry state");
274
275 // Fire the entry
276 entry.fire(when);
277
278 // Track that the entry has been fired
279 entry.set_when_internal(None);
280 }
281
282 // Update the elapsed cache
283 self.inner.elapsed.store(self.wheel.elapsed(), SeqCst);
284 }
285
286 /// Process the entry queue
287 ///
288 /// This handles adding and canceling timeouts.
289 fn process_queue(&mut self) {
290 for entry in self.inner.process.take() {
291 match (entry.when_internal(), entry.load_state()) {
292 (None, None) => {
293 // Nothing to do
294 }
295 (Some(_), None) => {
296 // Remove the entry
297 self.clear_entry(&entry);
298 }
299 (None, Some(when)) => {
300 // Queue the entry
301 self.add_entry(entry, when);
302 }
303 (Some(_), Some(next)) => {
304 self.clear_entry(&entry);
305 self.add_entry(entry, next);
306 }
307 }
308 }
309 }
310
311 fn clear_entry(&mut self, entry: &Arc<Entry>) {
312 self.wheel.remove(entry, &mut ());
313 entry.set_when_internal(None);
314 }
315
316 /// Fire the entry if it needs to, otherwise queue it to be processed later.
317 ///
318 /// Returns `None` if the entry was fired.
319 fn add_entry(&mut self, entry: Arc<Entry>, when: u64) {
320 use wheel::InsertError;
321
322 entry.set_when_internal(Some(when));
323
324 match self.wheel.insert(when, entry, &mut ()) {
325 Ok(_) => {}
326 Err((entry, InsertError::Elapsed)) => {
327 // The entry's deadline has elapsed, so fire it and update the
328 // internal state accordingly.
329 entry.set_when_internal(None);
330 entry.fire(when);
331 }
332 Err((entry, InsertError::Invalid)) => {
333 // The entry's deadline is invalid, so error it and update the
334 // internal state accordingly.
335 entry.set_when_internal(None);
336 entry.error();
337 }
338 }
339 }
340}
341
342impl Default for Timer<ParkThread, SystemNow> {
343 fn default() -> Self {
344 Timer::new(ParkThread::new())
345 }
346}
347
348impl<T, N> Park for Timer<T, N>
349where
350 T: Park,
351 N: Now,
352{
353 type Unpark = T::Unpark;
354 type Error = T::Error;
355
356 fn unpark(&self) -> Self::Unpark {
357 self.park.unpark()
358 }
359
360 fn park(&mut self) -> Result<(), Self::Error> {
361 self.process_queue();
362
363 match self.wheel.poll_at() {
364 Some(when) => {
365 let now = self.now.now();
366 let deadline = self.expiration_instant(when);
367
368 if deadline > now {
369 self.park.park_timeout(deadline - now)?;
370 } else {
371 self.park.park_timeout(Duration::from_secs(0))?;
372 }
373 }
374 None => {
375 self.park.park()?;
376 }
377 }
378
379 self.process();
380
381 Ok(())
382 }
383
384 fn park_timeout(&mut self, duration: Duration) -> Result<(), Self::Error> {
385 self.process_queue();
386
387 match self.wheel.poll_at() {
388 Some(when) => {
389 let now = self.now.now();
390 let deadline = self.expiration_instant(when);
391
392 if deadline > now {
393 self.park.park_timeout(cmp::min(deadline - now, duration))?;
394 } else {
395 self.park.park_timeout(Duration::from_secs(0))?;
396 }
397 }
398 None => {
399 self.park.park_timeout(duration)?;
400 }
401 }
402
403 self.process();
404
405 Ok(())
406 }
407}
408
409impl<T, N> Drop for Timer<T, N> {
410 fn drop(&mut self) {
411 use std::u64;
412
413 // Shutdown the stack of entries to process, preventing any new entries
414 // from being pushed.
415 self.inner.process.shutdown();
416
417 // Clear the wheel, using u64::MAX allows us to drain everything
418 let mut poll = wheel::Poll::new(u64::MAX);
419
420 while let Some(entry) = self.wheel.poll(&mut poll, &mut ()) {
421 entry.error();
422 }
423 }
424}
425
426// ===== impl Inner =====
427
428impl Inner {
429 fn new(start: Instant, unpark: Box<dyn Unpark>) -> Inner {
430 Inner {
431 num: AtomicUsize::new(0),
432 elapsed: AtomicU64::new(0),
433 process: AtomicStack::new(),
434 start,
435 unpark,
436 }
437 }
438
439 fn elapsed(&self) -> u64 {
440 self.elapsed.load(SeqCst)
441 }
442
443 /// Increment the number of active timeouts
444 fn increment(&self) -> Result<(), Error> {
445 let mut curr = self.num.load(SeqCst);
446
447 loop {
448 if curr == MAX_TIMEOUTS {
449 return Err(Error::at_capacity());
450 }
451
452 let actual = self.num.compare_and_swap(curr, curr + 1, SeqCst);
453
454 if curr == actual {
455 return Ok(());
456 }
457
458 curr = actual;
459 }
460 }
461
462 /// Decrement the number of active timeouts
463 fn decrement(&self) {
464 let prev = self.num.fetch_sub(1, SeqCst);
465 debug_assert!(prev <= MAX_TIMEOUTS);
466 }
467
468 fn queue(&self, entry: &Arc<Entry>) -> Result<(), Error> {
469 if self.process.push(entry)? {
470 // The timer is notified so that it can process the timeout
471 self.unpark.unpark();
472 }
473
474 Ok(())
475 }
476
477 fn normalize_deadline(&self, deadline: Instant) -> u64 {
478 if deadline < self.start {
479 return 0;
480 }
481
482 ::ms(deadline - self.start, ::Round::Up)
483 }
484}
485
486impl fmt::Debug for Inner {
487 fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
488 fmt.debug_struct("Inner").finish()
489 }
490}