1 | use crate::future::poll_fn; |
2 | use crate::time::{sleep_until, Duration, Instant, Sleep}; |
3 | use crate::util::trace; |
4 | |
5 | use std::future::Future; |
6 | use std::panic::Location; |
7 | use std::pin::Pin; |
8 | use std::task::{Context, Poll}; |
9 | |
10 | /// Creates new [`Interval`] that yields with interval of `period`. The first |
11 | /// tick completes immediately. The default [`MissedTickBehavior`] is |
12 | /// [`Burst`](MissedTickBehavior::Burst), but this can be configured |
13 | /// by calling [`set_missed_tick_behavior`](Interval::set_missed_tick_behavior). |
14 | /// |
15 | /// An interval will tick indefinitely. At any time, the [`Interval`] value can |
16 | /// be dropped. This cancels the interval. |
17 | /// |
18 | /// This function is equivalent to |
19 | /// [`interval_at(Instant::now(), period)`](interval_at). |
20 | /// |
21 | /// # Panics |
22 | /// |
23 | /// This function panics if `period` is zero. |
24 | /// |
25 | /// # Examples |
26 | /// |
27 | /// ``` |
28 | /// use tokio::time::{self, Duration}; |
29 | /// |
30 | /// #[tokio::main] |
31 | /// async fn main() { |
32 | /// let mut interval = time::interval(Duration::from_millis(10)); |
33 | /// |
34 | /// interval.tick().await; // ticks immediately |
35 | /// interval.tick().await; // ticks after 10ms |
36 | /// interval.tick().await; // ticks after 10ms |
37 | /// |
38 | /// // approximately 20ms have elapsed. |
39 | /// } |
40 | /// ``` |
41 | /// |
42 | /// A simple example using `interval` to execute a task every two seconds. |
43 | /// |
44 | /// The difference between `interval` and [`sleep`] is that an [`Interval`] |
45 | /// measures the time since the last tick, which means that [`.tick().await`] |
46 | /// may wait for a shorter time than the duration specified for the interval |
47 | /// if some time has passed between calls to [`.tick().await`]. |
48 | /// |
49 | /// If the tick in the example below was replaced with [`sleep`], the task |
50 | /// would only be executed once every three seconds, and not every two |
51 | /// seconds. |
52 | /// |
53 | /// ``` |
54 | /// use tokio::time; |
55 | /// |
56 | /// async fn task_that_takes_a_second() { |
57 | /// println!("hello" ); |
58 | /// time::sleep(time::Duration::from_secs(1)).await |
59 | /// } |
60 | /// |
61 | /// #[tokio::main] |
62 | /// async fn main() { |
63 | /// let mut interval = time::interval(time::Duration::from_secs(2)); |
64 | /// for _i in 0..5 { |
65 | /// interval.tick().await; |
66 | /// task_that_takes_a_second().await; |
67 | /// } |
68 | /// } |
69 | /// ``` |
70 | /// |
71 | /// [`sleep`]: crate::time::sleep() |
72 | /// [`.tick().await`]: Interval::tick |
73 | #[track_caller ] |
74 | pub fn interval(period: Duration) -> Interval { |
75 | assert!(period > Duration::new(0, 0), "`period` must be non-zero." ); |
76 | internal_interval_at(start:Instant::now(), period, location:trace::caller_location()) |
77 | } |
78 | |
79 | /// Creates new [`Interval`] that yields with interval of `period` with the |
80 | /// first tick completing at `start`. The default [`MissedTickBehavior`] is |
81 | /// [`Burst`](MissedTickBehavior::Burst), but this can be configured |
82 | /// by calling [`set_missed_tick_behavior`](Interval::set_missed_tick_behavior). |
83 | /// |
84 | /// An interval will tick indefinitely. At any time, the [`Interval`] value can |
85 | /// be dropped. This cancels the interval. |
86 | /// |
87 | /// # Panics |
88 | /// |
89 | /// This function panics if `period` is zero. |
90 | /// |
91 | /// # Examples |
92 | /// |
93 | /// ``` |
94 | /// use tokio::time::{interval_at, Duration, Instant}; |
95 | /// |
96 | /// #[tokio::main] |
97 | /// async fn main() { |
98 | /// let start = Instant::now() + Duration::from_millis(50); |
99 | /// let mut interval = interval_at(start, Duration::from_millis(10)); |
100 | /// |
101 | /// interval.tick().await; // ticks after 50ms |
102 | /// interval.tick().await; // ticks after 10ms |
103 | /// interval.tick().await; // ticks after 10ms |
104 | /// |
105 | /// // approximately 70ms have elapsed. |
106 | /// } |
107 | /// ``` |
108 | #[track_caller ] |
109 | pub fn interval_at(start: Instant, period: Duration) -> Interval { |
110 | assert!(period > Duration::new(0, 0), "`period` must be non-zero." ); |
111 | internal_interval_at(start, period, location:trace::caller_location()) |
112 | } |
113 | |
114 | #[cfg_attr (not(all(tokio_unstable, feature = "tracing" )), allow(unused_variables))] |
115 | fn internal_interval_at( |
116 | start: Instant, |
117 | period: Duration, |
118 | location: Option<&'static Location<'static>>, |
119 | ) -> Interval { |
120 | #[cfg (all(tokio_unstable, feature = "tracing" ))] |
121 | let resource_span = { |
122 | let location = location.expect("should have location if tracing" ); |
123 | |
124 | tracing::trace_span!( |
125 | "runtime.resource" , |
126 | concrete_type = "Interval" , |
127 | kind = "timer" , |
128 | loc.file = location.file(), |
129 | loc.line = location.line(), |
130 | loc.col = location.column(), |
131 | ) |
132 | }; |
133 | |
134 | #[cfg (all(tokio_unstable, feature = "tracing" ))] |
135 | let delay = resource_span.in_scope(|| Box::pin(sleep_until(start))); |
136 | |
137 | #[cfg (not(all(tokio_unstable, feature = "tracing" )))] |
138 | let delay = Box::pin(sleep_until(start)); |
139 | |
140 | Interval { |
141 | delay, |
142 | period, |
143 | missed_tick_behavior: Default::default(), |
144 | #[cfg (all(tokio_unstable, feature = "tracing" ))] |
145 | resource_span, |
146 | } |
147 | } |
148 | |
149 | /// Defines the behavior of an [`Interval`] when it misses a tick. |
150 | /// |
151 | /// Sometimes, an [`Interval`]'s tick is missed. For example, consider the |
152 | /// following: |
153 | /// |
154 | /// ``` |
155 | /// use tokio::time::{self, Duration}; |
156 | /// # async fn task_that_takes_one_to_three_millis() {} |
157 | /// |
158 | /// #[tokio::main] |
159 | /// async fn main() { |
160 | /// // ticks every 2 milliseconds |
161 | /// let mut interval = time::interval(Duration::from_millis(2)); |
162 | /// for _ in 0..5 { |
163 | /// interval.tick().await; |
164 | /// // if this takes more than 2 milliseconds, a tick will be delayed |
165 | /// task_that_takes_one_to_three_millis().await; |
166 | /// } |
167 | /// } |
168 | /// ``` |
169 | /// |
170 | /// Generally, a tick is missed if too much time is spent without calling |
171 | /// [`Interval::tick()`]. |
172 | /// |
173 | /// By default, when a tick is missed, [`Interval`] fires ticks as quickly as it |
174 | /// can until it is "caught up" in time to where it should be. |
175 | /// `MissedTickBehavior` can be used to specify a different behavior for |
176 | /// [`Interval`] to exhibit. Each variant represents a different strategy. |
177 | /// |
178 | /// Note that because the executor cannot guarantee exact precision with timers, |
179 | /// these strategies will only apply when the delay is greater than 5 |
180 | /// milliseconds. |
181 | #[derive (Debug, Clone, Copy, PartialEq, Eq)] |
182 | pub enum MissedTickBehavior { |
183 | /// Ticks as fast as possible until caught up. |
184 | /// |
185 | /// When this strategy is used, [`Interval`] schedules ticks "normally" (the |
186 | /// same as it would have if the ticks hadn't been delayed), which results |
187 | /// in it firing ticks as fast as possible until it is caught up in time to |
188 | /// where it should be. Unlike [`Delay`] and [`Skip`], the ticks yielded |
189 | /// when `Burst` is used (the [`Instant`]s that [`tick`](Interval::tick) |
190 | /// yields) aren't different than they would have been if a tick had not |
191 | /// been missed. Like [`Skip`], and unlike [`Delay`], the ticks may be |
192 | /// shortened. |
193 | /// |
194 | /// This looks something like this: |
195 | /// ```text |
196 | /// Expected ticks: | 1 | 2 | 3 | 4 | 5 | 6 | |
197 | /// Actual ticks: | work -----| delay | work | work | work -| work -----| |
198 | /// ``` |
199 | /// |
200 | /// In code: |
201 | /// |
202 | /// ``` |
203 | /// use tokio::time::{interval, Duration}; |
204 | /// # async fn task_that_takes_200_millis() {} |
205 | /// |
206 | /// # #[tokio::main(flavor = "current_thread" )] |
207 | /// # async fn main() { |
208 | /// let mut interval = interval(Duration::from_millis(50)); |
209 | /// |
210 | /// // First tick resolves immediately after creation |
211 | /// interval.tick().await; |
212 | /// |
213 | /// task_that_takes_200_millis().await; |
214 | /// // The `Interval` has missed a tick |
215 | /// |
216 | /// // Since we have exceeded our timeout, this will resolve immediately |
217 | /// interval.tick().await; |
218 | /// |
219 | /// // Since we are more than 100ms after the start of `interval`, this will |
220 | /// // also resolve immediately. |
221 | /// interval.tick().await; |
222 | /// |
223 | /// // Also resolves immediately, because it was supposed to resolve at |
224 | /// // 150ms after the start of `interval` |
225 | /// interval.tick().await; |
226 | /// |
227 | /// // Resolves immediately |
228 | /// interval.tick().await; |
229 | /// |
230 | /// // Since we have gotten to 200ms after the start of `interval`, this |
231 | /// // will resolve after 50ms |
232 | /// interval.tick().await; |
233 | /// # } |
234 | /// ``` |
235 | /// |
236 | /// This is the default behavior when [`Interval`] is created with |
237 | /// [`interval`] and [`interval_at`]. |
238 | /// |
239 | /// [`Delay`]: MissedTickBehavior::Delay |
240 | /// [`Skip`]: MissedTickBehavior::Skip |
241 | Burst, |
242 | |
243 | /// Tick at multiples of `period` from when [`tick`] was called, rather than |
244 | /// from `start`. |
245 | /// |
246 | /// When this strategy is used and [`Interval`] has missed a tick, instead |
247 | /// of scheduling ticks to fire at multiples of `period` from `start` (the |
248 | /// time when the first tick was fired), it schedules all future ticks to |
249 | /// happen at a regular `period` from the point when [`tick`] was called. |
250 | /// Unlike [`Burst`] and [`Skip`], ticks are not shortened, and they aren't |
251 | /// guaranteed to happen at a multiple of `period` from `start` any longer. |
252 | /// |
253 | /// This looks something like this: |
254 | /// ```text |
255 | /// Expected ticks: | 1 | 2 | 3 | 4 | 5 | 6 | |
256 | /// Actual ticks: | work -----| delay | work -----| work -----| work -----| |
257 | /// ``` |
258 | /// |
259 | /// In code: |
260 | /// |
261 | /// ``` |
262 | /// use tokio::time::{interval, Duration, MissedTickBehavior}; |
263 | /// # async fn task_that_takes_more_than_50_millis() {} |
264 | /// |
265 | /// # #[tokio::main(flavor = "current_thread" )] |
266 | /// # async fn main() { |
267 | /// let mut interval = interval(Duration::from_millis(50)); |
268 | /// interval.set_missed_tick_behavior(MissedTickBehavior::Delay); |
269 | /// |
270 | /// task_that_takes_more_than_50_millis().await; |
271 | /// // The `Interval` has missed a tick |
272 | /// |
273 | /// // Since we have exceeded our timeout, this will resolve immediately |
274 | /// interval.tick().await; |
275 | /// |
276 | /// // But this one, rather than also resolving immediately, as might happen |
277 | /// // with the `Burst` or `Skip` behaviors, will not resolve until |
278 | /// // 50ms after the call to `tick` up above. That is, in `tick`, when we |
279 | /// // recognize that we missed a tick, we schedule the next tick to happen |
280 | /// // 50ms (or whatever the `period` is) from right then, not from when |
281 | /// // were *supposed* to tick |
282 | /// interval.tick().await; |
283 | /// # } |
284 | /// ``` |
285 | /// |
286 | /// [`Burst`]: MissedTickBehavior::Burst |
287 | /// [`Skip`]: MissedTickBehavior::Skip |
288 | /// [`tick`]: Interval::tick |
289 | Delay, |
290 | |
291 | /// Skips missed ticks and tick on the next multiple of `period` from |
292 | /// `start`. |
293 | /// |
294 | /// When this strategy is used, [`Interval`] schedules the next tick to fire |
295 | /// at the next-closest tick that is a multiple of `period` away from |
296 | /// `start` (the point where [`Interval`] first ticked). Like [`Burst`], all |
297 | /// ticks remain multiples of `period` away from `start`, but unlike |
298 | /// [`Burst`], the ticks may not be *one* multiple of `period` away from the |
299 | /// last tick. Like [`Delay`], the ticks are no longer the same as they |
300 | /// would have been if ticks had not been missed, but unlike [`Delay`], and |
301 | /// like [`Burst`], the ticks may be shortened to be less than one `period` |
302 | /// away from each other. |
303 | /// |
304 | /// This looks something like this: |
305 | /// ```text |
306 | /// Expected ticks: | 1 | 2 | 3 | 4 | 5 | 6 | |
307 | /// Actual ticks: | work -----| delay | work ---| work -----| work -----| |
308 | /// ``` |
309 | /// |
310 | /// In code: |
311 | /// |
312 | /// ``` |
313 | /// use tokio::time::{interval, Duration, MissedTickBehavior}; |
314 | /// # async fn task_that_takes_75_millis() {} |
315 | /// |
316 | /// # #[tokio::main(flavor = "current_thread" )] |
317 | /// # async fn main() { |
318 | /// let mut interval = interval(Duration::from_millis(50)); |
319 | /// interval.set_missed_tick_behavior(MissedTickBehavior::Skip); |
320 | /// |
321 | /// task_that_takes_75_millis().await; |
322 | /// // The `Interval` has missed a tick |
323 | /// |
324 | /// // Since we have exceeded our timeout, this will resolve immediately |
325 | /// interval.tick().await; |
326 | /// |
327 | /// // This one will resolve after 25ms, 100ms after the start of |
328 | /// // `interval`, which is the closest multiple of `period` from the start |
329 | /// // of `interval` after the call to `tick` up above. |
330 | /// interval.tick().await; |
331 | /// # } |
332 | /// ``` |
333 | /// |
334 | /// [`Burst`]: MissedTickBehavior::Burst |
335 | /// [`Delay`]: MissedTickBehavior::Delay |
336 | Skip, |
337 | } |
338 | |
339 | impl MissedTickBehavior { |
340 | /// If a tick is missed, this method is called to determine when the next tick should happen. |
341 | fn next_timeout(&self, timeout: Instant, now: Instant, period: Duration) -> Instant { |
342 | match self { |
343 | Self::Burst => timeout + period, |
344 | Self::Delay => now + period, |
345 | Self::Skip => { |
346 | now + period |
347 | - Duration::from_nanos( |
348 | ((now - timeout).as_nanos() % period.as_nanos()) |
349 | .try_into() |
350 | // This operation is practically guaranteed not to |
351 | // fail, as in order for it to fail, `period` would |
352 | // have to be longer than `now - timeout`, and both |
353 | // would have to be longer than 584 years. |
354 | // |
355 | // If it did fail, there's not a good way to pass |
356 | // the error along to the user, so we just panic. |
357 | .expect( |
358 | "too much time has elapsed since the interval was supposed to tick" , |
359 | ), |
360 | ) |
361 | } |
362 | } |
363 | } |
364 | } |
365 | |
366 | impl Default for MissedTickBehavior { |
367 | /// Returns [`MissedTickBehavior::Burst`]. |
368 | /// |
369 | /// For most usecases, the [`Burst`] strategy is what is desired. |
370 | /// Additionally, to preserve backwards compatibility, the [`Burst`] |
371 | /// strategy must be the default. For these reasons, |
372 | /// [`MissedTickBehavior::Burst`] is the default for [`MissedTickBehavior`]. |
373 | /// See [`Burst`] for more details. |
374 | /// |
375 | /// [`Burst`]: MissedTickBehavior::Burst |
376 | fn default() -> Self { |
377 | Self::Burst |
378 | } |
379 | } |
380 | |
381 | /// Interval returned by [`interval`] and [`interval_at`]. |
382 | /// |
383 | /// This type allows you to wait on a sequence of instants with a certain |
384 | /// duration between each instant. Unlike calling [`sleep`] in a loop, this lets |
385 | /// you count the time spent between the calls to [`sleep`] as well. |
386 | /// |
387 | /// An `Interval` can be turned into a `Stream` with [`IntervalStream`]. |
388 | /// |
389 | /// [`IntervalStream`]: https://docs.rs/tokio-stream/latest/tokio_stream/wrappers/struct.IntervalStream.html |
390 | /// [`sleep`]: crate::time::sleep() |
391 | #[derive (Debug)] |
392 | pub struct Interval { |
393 | /// Future that completes the next time the `Interval` yields a value. |
394 | delay: Pin<Box<Sleep>>, |
395 | |
396 | /// The duration between values yielded by `Interval`. |
397 | period: Duration, |
398 | |
399 | /// The strategy `Interval` should use when a tick is missed. |
400 | missed_tick_behavior: MissedTickBehavior, |
401 | |
402 | #[cfg (all(tokio_unstable, feature = "tracing" ))] |
403 | resource_span: tracing::Span, |
404 | } |
405 | |
406 | impl Interval { |
407 | /// Completes when the next instant in the interval has been reached. |
408 | /// |
409 | /// # Cancel safety |
410 | /// |
411 | /// This method is cancellation safe. If `tick` is used as the branch in a `tokio::select!` and |
412 | /// another branch completes first, then no tick has been consumed. |
413 | /// |
414 | /// # Examples |
415 | /// |
416 | /// ``` |
417 | /// use tokio::time; |
418 | /// |
419 | /// use std::time::Duration; |
420 | /// |
421 | /// #[tokio::main] |
422 | /// async fn main() { |
423 | /// let mut interval = time::interval(Duration::from_millis(10)); |
424 | /// |
425 | /// interval.tick().await; |
426 | /// // approximately 0ms have elapsed. The first tick completes immediately. |
427 | /// interval.tick().await; |
428 | /// interval.tick().await; |
429 | /// |
430 | /// // approximately 20ms have elapsed. |
431 | /// } |
432 | /// ``` |
433 | pub async fn tick(&mut self) -> Instant { |
434 | #[cfg (all(tokio_unstable, feature = "tracing" ))] |
435 | let resource_span = self.resource_span.clone(); |
436 | #[cfg (all(tokio_unstable, feature = "tracing" ))] |
437 | let instant = trace::async_op( |
438 | || poll_fn(|cx| self.poll_tick(cx)), |
439 | resource_span, |
440 | "Interval::tick" , |
441 | "poll_tick" , |
442 | false, |
443 | ); |
444 | #[cfg (not(all(tokio_unstable, feature = "tracing" )))] |
445 | let instant = poll_fn(|cx| self.poll_tick(cx)); |
446 | |
447 | instant.await |
448 | } |
449 | |
450 | /// Polls for the next instant in the interval to be reached. |
451 | /// |
452 | /// This method can return the following values: |
453 | /// |
454 | /// * `Poll::Pending` if the next instant has not yet been reached. |
455 | /// * `Poll::Ready(instant)` if the next instant has been reached. |
456 | /// |
457 | /// When this method returns `Poll::Pending`, the current task is scheduled |
458 | /// to receive a wakeup when the instant has elapsed. Note that on multiple |
459 | /// calls to `poll_tick`, only the [`Waker`](std::task::Waker) from the |
460 | /// [`Context`] passed to the most recent call is scheduled to receive a |
461 | /// wakeup. |
462 | pub fn poll_tick(&mut self, cx: &mut Context<'_>) -> Poll<Instant> { |
463 | // Wait for the delay to be done |
464 | ready!(Pin::new(&mut self.delay).poll(cx)); |
465 | |
466 | // Get the time when we were scheduled to tick |
467 | let timeout = self.delay.deadline(); |
468 | |
469 | let now = Instant::now(); |
470 | |
471 | // If a tick was not missed, and thus we are being called before the |
472 | // next tick is due, just schedule the next tick normally, one `period` |
473 | // after `timeout` |
474 | // |
475 | // However, if a tick took excessively long and we are now behind, |
476 | // schedule the next tick according to how the user specified with |
477 | // `MissedTickBehavior` |
478 | let next = if now > timeout + Duration::from_millis(5) { |
479 | self.missed_tick_behavior |
480 | .next_timeout(timeout, now, self.period) |
481 | } else { |
482 | timeout + self.period |
483 | }; |
484 | |
485 | // When we arrive here, the internal delay returned `Poll::Ready`. |
486 | // Reset the delay but do not register it. It should be registered with |
487 | // the next call to [`poll_tick`]. |
488 | self.delay.as_mut().reset_without_reregister(next); |
489 | |
490 | // Return the time when we were scheduled to tick |
491 | Poll::Ready(timeout) |
492 | } |
493 | |
494 | /// Resets the interval to complete one period after the current time. |
495 | /// |
496 | /// This method ignores [`MissedTickBehavior`] strategy. |
497 | /// |
498 | /// # Examples |
499 | /// |
500 | /// ``` |
501 | /// use tokio::time; |
502 | /// |
503 | /// use std::time::Duration; |
504 | /// |
505 | /// #[tokio::main] |
506 | /// async fn main() { |
507 | /// let mut interval = time::interval(Duration::from_millis(100)); |
508 | /// |
509 | /// interval.tick().await; |
510 | /// |
511 | /// time::sleep(Duration::from_millis(50)).await; |
512 | /// interval.reset(); |
513 | /// |
514 | /// interval.tick().await; |
515 | /// interval.tick().await; |
516 | /// |
517 | /// // approximately 250ms have elapsed. |
518 | /// } |
519 | /// ``` |
520 | pub fn reset(&mut self) { |
521 | self.delay.as_mut().reset(Instant::now() + self.period); |
522 | } |
523 | |
524 | /// Returns the [`MissedTickBehavior`] strategy currently being used. |
525 | pub fn missed_tick_behavior(&self) -> MissedTickBehavior { |
526 | self.missed_tick_behavior |
527 | } |
528 | |
529 | /// Sets the [`MissedTickBehavior`] strategy that should be used. |
530 | pub fn set_missed_tick_behavior(&mut self, behavior: MissedTickBehavior) { |
531 | self.missed_tick_behavior = behavior; |
532 | } |
533 | |
534 | /// Returns the period of the interval. |
535 | pub fn period(&self) -> Duration { |
536 | self.period |
537 | } |
538 | } |
539 | |