Skip to main content

tokio_metrics/
task.rs

1use futures_util::task::{ArcWake, AtomicWaker};
2use pin_project_lite::pin_project;
3use std::cell::RefCell;
4use std::future::Future;
5use std::ops::Deref;
6use std::pin::Pin;
7use std::sync::atomic::{AtomicBool, AtomicU64, Ordering::SeqCst};
8use std::sync::{Arc, OnceLock};
9use std::task::{Context, Poll};
10use tokio_stream::Stream;
11
12#[cfg(feature = "rt")]
13use tokio::time::{Duration, Instant};
14
15use crate::derived_metrics::derived_metrics;
16#[cfg(not(feature = "rt"))]
17use std::time::{Duration, Instant};
18
19#[cfg(feature = "metrics-rs-integration")]
20pub(crate) mod metrics_rs_integration;
21
22/// Monitors key metrics of instrumented tasks.
23///
24/// This struct is preferred for generating a variable number of monitors at runtime.
25/// If you can construct a fixed count of `static` monitors instead, see [`TaskMonitorCore`].
26///
27/// ### Basic Usage
28/// A [`TaskMonitor`] tracks key [metrics][TaskMetrics] of async tasks that have been
29/// [instrumented][`TaskMonitor::instrument`] with the monitor.
30///
31/// In the below example, a [`TaskMonitor`] is [constructed][TaskMonitor::new] and used to
32/// [instrument][TaskMonitor::instrument] three worker tasks; meanwhile, a fourth task
33/// prints [metrics][TaskMetrics] in 500ms [intervals][TaskMonitor::intervals].
34/// ```
35/// use std::time::Duration;
36///
37/// #[tokio::main]
38/// async fn main() {
39///     // construct a metrics monitor
40///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
41///
42///     // print task metrics every 500ms
43///     {
44///         let metrics_monitor = metrics_monitor.clone();
45///         tokio::spawn(async move {
46///             for interval in metrics_monitor.intervals() {
47///                 // pretty-print the metric interval
48///                 println!("{:?}", interval);
49///                 // wait 500ms
50///                 tokio::time::sleep(Duration::from_millis(500)).await;
51///             }
52///         });
53///     }
54///
55///     // instrument some tasks and await them
56///     // note that the same TaskMonitor can be used for multiple tasks
57///     tokio::join![
58///         metrics_monitor.instrument(do_work()),
59///         metrics_monitor.instrument(do_work()),
60///         metrics_monitor.instrument(do_work())
61///     ];
62/// }
63///
64/// async fn do_work() {
65///     for _ in 0..25 {
66///         tokio::task::yield_now().await;
67///         tokio::time::sleep(Duration::from_millis(100)).await;
68///     }
69/// }
70/// ```
71///
72/// ### What should I instrument?
73/// In most cases, you should construct a *distinct* [`TaskMonitor`] for each kind of key task.
74///
75/// #### Instrumenting a web application
76/// For instance, a web service should have a distinct [`TaskMonitor`] for each endpoint. Within
77/// each endpoint, it's prudent to additionally instrument major sub-tasks, each with their own
78/// distinct [`TaskMonitor`]s. [*Why are my tasks slow?*](#why-are-my-tasks-slow) explores a
79/// debugging scenario for a web service that takes this approach to instrumentation. This
80/// approach is exemplified in the below example:
81/// ```no_run
82/// // The unabridged version of this snippet is in the examples directory of this crate.
83///
84/// #[tokio::main]
85/// async fn main() {
86///     // construct a TaskMonitor for root endpoint
87///     let monitor_root = tokio_metrics::TaskMonitor::new();
88///
89///     // construct TaskMonitors for create_users endpoint
90///     let monitor_create_user = CreateUserMonitors {
91///         // monitor for the entire endpoint
92///         route: tokio_metrics::TaskMonitor::new(),
93///         // monitor for database insertion subtask
94///         insert: tokio_metrics::TaskMonitor::new(),
95///     };
96///
97///     // build our application with two instrumented endpoints
98///     let app = axum::Router::new()
99///         // `GET /` goes to `root`
100///         .route("/", axum::routing::get({
101///             let monitor = monitor_root.clone();
102///             move || monitor.instrument(async { "Hello, World!" })
103///         }))
104///         // `POST /users` goes to `create_user`
105///         .route("/users", axum::routing::post({
106///             let monitors = monitor_create_user.clone();
107///             let route = monitors.route.clone();
108///             move |payload| {
109///                 route.instrument(create_user(payload, monitors))
110///             }
111///         }));
112///
113///     // print task metrics for each endpoint every 1s
114///     let metrics_frequency = std::time::Duration::from_secs(1);
115///     tokio::spawn(async move {
116///         let root_intervals = monitor_root.intervals();
117///         let create_user_route_intervals =
118///             monitor_create_user.route.intervals();
119///         let create_user_insert_intervals =
120///             monitor_create_user.insert.intervals();
121///         let create_user_intervals =
122///             create_user_route_intervals.zip(create_user_insert_intervals);
123///
124///         let intervals = root_intervals.zip(create_user_intervals);
125///         for (root_route, (create_user_route, create_user_insert)) in intervals {
126///             println!("root_route = {:#?}", root_route);
127///             println!("create_user_route = {:#?}", create_user_route);
128///             println!("create_user_insert = {:#?}", create_user_insert);
129///             tokio::time::sleep(metrics_frequency).await;
130///         }
131///     });
132///
133///     // run the server
134///     let addr = std::net::SocketAddr::from(([127, 0, 0, 1], 3000));
135///     let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
136///     axum::serve(listener, app)
137///         .await
138///         .unwrap();
139/// }
140///
141/// async fn create_user(
142///     axum::Json(payload): axum::Json<CreateUser>,
143///     monitors: CreateUserMonitors,
144/// ) -> impl axum::response::IntoResponse {
145///     let user = User { id: 1337, username: payload.username, };
146///     // instrument inserting the user into the db:
147///     let _ = monitors.insert.instrument(insert_user(user.clone())).await;
148///     (axum::http::StatusCode::CREATED, axum::Json(user))
149/// }
150///
151/// /* definitions of CreateUserMonitors, CreateUser and User omitted for brevity */
152///
153/// #
154/// # #[derive(Clone)]
155/// # struct CreateUserMonitors {
156/// #     // monitor for the entire endpoint
157/// #     route: tokio_metrics::TaskMonitor,
158/// #     // monitor for database insertion subtask
159/// #     insert: tokio_metrics::TaskMonitor,
160/// # }
161/// #
162/// # #[derive(serde::Deserialize)] struct CreateUser { username: String, }
163/// # #[derive(Clone, serde::Serialize)] struct User { id: u64, username: String, }
164/// #
165/// // insert the user into the database
166/// async fn insert_user(_: User) {
167///     /* implementation details elided */
168///     tokio::time::sleep(std::time::Duration::from_secs(1)).await;
169/// }
170/// ```
171///
172/// ### Why are my tasks slow?
173/// **Scenario:** You track key, high-level metrics about the customer response time. An alarm warns
174/// you that P90 latency for an endpoint exceeds your targets. What is causing the increase?
175///
176/// #### Identifying the high-level culprits
177/// A set of tasks will appear to execute more slowly if:
178/// - they are taking longer to poll (i.e., they consume too much CPU time)
179/// - they are waiting longer to be polled (e.g., they're waiting longer in tokio's scheduling
180///   queues)
181/// - they are waiting longer on external events to complete (e.g., asynchronous network requests)
182///
183/// The culprits, at a high level, may be some combination of these sources of latency. Fortunately,
184/// you have instrumented the key tasks of each of your endpoints with distinct [`TaskMonitor`]s.
185/// Using the monitors on the endpoint experiencing elevated latency, you begin by answering:
186/// - [*Are my tasks taking longer to poll?*](#are-my-tasks-taking-longer-to-poll)
187/// - [*Are my tasks spending more time waiting to be polled?*](#are-my-tasks-spending-more-time-waiting-to-be-polled)
188/// - [*Are my tasks spending more time waiting on external events to complete?*](#are-my-tasks-spending-more-time-waiting-on-external-events-to-complete)
189///
190/// ##### Are my tasks taking longer to poll?
191/// - **Did [`mean_poll_duration`][TaskMetrics::mean_poll_duration] increase?**
192///   This metric reflects the mean poll duration. If it increased, it means that, on average,
193///   individual polls tended to take longer. However, this does not necessarily imply increased
194///   task latency: An increase in poll durations could be offset by fewer polls.
195/// - **Did [`slow_poll_ratio`][TaskMetrics::slow_poll_ratio] increase?**
196///   This metric reflects the proportion of polls that were 'slow'. If it increased, it means that
197///   a greater proportion of polls performed excessive computation before yielding. This does not
198///   necessarily imply increased task latency: An increase in the proportion of slow polls could be
199///   offset by fewer or faster polls.
200/// - **Did [`mean_slow_poll_duration`][TaskMetrics::mean_slow_poll_duration] increase?**
201///   This metric reflects the mean duration of slow polls. If it increased, it means that, on
202///   average, slow polls got slower. This does not necessarily imply increased task latency: An
203///   increase in average slow poll duration could be offset by fewer or faster polls.
204///
205/// If so, [*why are my tasks taking longer to poll?*](#why-are-my-tasks-taking-longer-to-poll)
206///
207/// ##### Are my tasks spending more time waiting to be polled?
208/// - **Did [`mean_first_poll_delay`][TaskMetrics::mean_first_poll_delay] increase?**
209///   This metric reflects the mean delay between the instant a task is first instrumented and the
210///   instant it is first polled. If it increases, it means that, on average, tasks spent longer
211///   waiting to be initially run.
212/// - **Did [`mean_scheduled_duration`][TaskMetrics::mean_scheduled_duration] increase?**
213///   This metric reflects the mean duration that tasks spent in the scheduled state. The
214///   'scheduled' state of a task is the duration between the instant a task is awoken and the
215///   instant it is subsequently polled. If this metric increases, it means that, on average, tasks
216///   spent longer in tokio's queues before being polled.
217/// - **Did [`long_delay_ratio`][TaskMetrics::long_delay_ratio] increase?**
218///   This metric reflects the proportion of scheduling delays which were 'long'. If it increased,
219///   it means that a greater proportion of tasks experienced excessive delays before they could
220///   execute after being woken. This does not necessarily indicate an increase in latency, as this
221///   could be offset by fewer or faster task polls.
222/// - **Did [`mean_long_delay_duration`][TaskMetrics::mean_long_delay_duration] increase?**
223///   This metric reflects the mean duration of long delays. If it increased, it means that, on
224///   average, long delays got even longer. This does not necessarily imply increased task latency:
225///   an increase in average long delay duration could be offset by fewer or faster polls or more
226///   short schedules.
227///
228/// If so, [*why are my tasks spending more time waiting to be polled?*](#why-are-my-tasks-spending-more-time-waiting-to-be-polled)
229///
230/// ##### Are my tasks spending more time waiting on external events to complete?
231/// - **Did [`mean_idle_duration`][TaskMetrics::mean_idle_duration] increase?**
232///   This metric reflects the mean duration that tasks spent in the idle state. The idle state is
233///   the duration spanning the instant a task completes a poll, and the instant that it is next
234///   awoken. Tasks inhabit this state when they are waiting for task-external events to complete
235///   (e.g., an asynchronous sleep, a network request, file I/O, etc.). If this metric increases,
236///   tasks, in aggregate, spent more time waiting for task-external events to complete.
237///
238/// If so, [*why are my tasks spending more time waiting on external events to complete?*](#why-are-my-tasks-spending-more-time-waiting-on-external-events-to-complete)
239///
240/// #### Digging deeper
241/// Having [established the high-level culprits](#identifying-the-high-level-culprits), you now
242/// search for further explanation...
243///
244/// ##### Why are my tasks taking longer to poll?
245/// You observed that [your tasks are taking longer to poll](#are-my-tasks-taking-longer-to-poll).
246/// The culprit is likely some combination of:
247/// - **Your tasks are accidentally blocking.** Common culprits include:
248///     1. Using the Rust standard library's [filesystem](https://doc.rust-lang.org/std/fs/) or
249///        [networking](https://doc.rust-lang.org/std/net/) APIs.
250///        These APIs are synchronous; use tokio's [filesystem](https://docs.rs/tokio/latest/tokio/fs/)
251///        and [networking](https://docs.rs/tokio/latest/tokio/net/) APIs, instead.
252///     3. Calling [`block_on`](https://docs.rs/tokio/latest/tokio/runtime/struct.Handle.html#method.block_on).
253///     4. Invoking `println!` or other synchronous logging routines.
254///        Invocations of `println!` involve acquiring an exclusive lock on stdout, followed by a
255///        synchronous write to stdout.
256/// 2. **Your tasks are computationally expensive.** Common culprits include:
257///     1. TLS/cryptographic routines
258///     2. doing a lot of processing on bytes
259///     3. calling non-Tokio resources
260///
261/// ##### Why are my tasks spending more time waiting to be polled?
262/// You observed that [your tasks are spending more time waiting to be polled](#are-my-tasks-spending-more-time-waiting-to-be-polled)
263/// suggesting some combination of:
264/// - Your application is inflating the time elapsed between instrumentation and first poll.
265/// - Your tasks are being scheduled into tokio's global queue.
266/// - Other tasks are spending too long without yielding, thus backing up tokio's queues.
267///
268/// Start by asking: [*Is time-to-first-poll unusually high?*](#is-time-to-first-poll-unusually-high)
269///
270/// ##### Why are my tasks spending more time waiting on external events to complete?
271/// You observed that [your tasks are spending more time waiting waiting on external events to
272/// complete](#are-my-tasks-spending-more-time-waiting-on-external-events-to-complete). But what
273/// event? Fortunately, within the task experiencing increased idle times, you monitored several
274/// sub-tasks with distinct [`TaskMonitor`]s. For each of these sub-tasks, you [*you try to identify
275/// the performance culprits...*](#identifying-the-high-level-culprits)
276///
277/// #### Digging even deeper
278///
279/// ##### Is time-to-first-poll unusually high?
280/// Contrast these two metrics:
281/// - **[`mean_first_poll_delay`][TaskMetrics::mean_first_poll_delay]**
282///   This metric reflects the mean delay between the instant a task is first instrumented and the
283///   instant it is *first* polled.
284/// - **[`mean_scheduled_duration`][TaskMetrics::mean_scheduled_duration]**
285///   This metric reflects the mean delay between the instant when tasks were awoken and the
286///   instant they were subsequently polled.
287///
288/// If the former metric exceeds the latter (or increased unexpectedly more than the latter), then
289/// start by investigating [*if your application is artificially delaying the time-to-first-poll*](#is-my-application-delaying-the-time-to-first-poll).
290///
291/// Otherwise, investigate [*if other tasks are polling too long without yielding*](#are-other-tasks-polling-too-long-without-yielding).
292///
293/// ##### Is my application delaying the time-to-first-poll?
294/// You observed that [`mean_first_poll_delay`][TaskMetrics::mean_first_poll_delay] increased, more
295/// than [`mean_scheduled_duration`][TaskMetrics::mean_scheduled_duration]. Your application may be
296/// needlessly inflating the time elapsed between instrumentation and first poll. Are you
297/// constructing (and instrumenting) tasks separately from awaiting or spawning them?
298///
299/// For instance, in the below example, the application induces 1 second delay between when `task`
300/// is instrumented and when it is awaited:
301/// ```rust
302/// #[tokio::main]
303/// async fn main() {
304///     use tokio::time::Duration;
305///     let monitor = tokio_metrics::TaskMonitor::new();
306///
307///     let task = monitor.instrument(async move {});
308///
309///     let one_sec = Duration::from_secs(1);
310///     tokio::time::sleep(one_sec).await;
311///
312///     let _ = tokio::spawn(task).await;
313///
314///     assert!(monitor.cumulative().total_first_poll_delay >= one_sec);
315/// }
316/// ```
317///
318/// Otherwise, [`mean_first_poll_delay`][TaskMetrics::mean_first_poll_delay] might be unusually high
319/// because [*your application is spawning key tasks into tokio's global queue...*](#is-my-application-spawning-more-tasks-into-tokio’s-global-queue)
320///
321/// ##### Is my application spawning more tasks into tokio's global queue?
322/// Tasks awoken from threads *not* managed by the tokio runtime are scheduled with a slower,
323/// global "injection" queue.
324///
325/// You may be notifying runtime tasks from off-runtime. For instance, Given the following:
326/// ```ignore
327/// #[tokio::main]
328/// async fn main() {
329///     for _ in 0..100 {
330///         let (tx, rx) = oneshot::channel();
331///         tokio::spawn(async move {
332///             tx.send(());
333///         })
334///
335///         rx.await;
336///     }
337/// }
338/// ```
339/// One would expect this to run efficiently, however, the main task is run *off* the main runtime
340/// and the spawned tasks are *on* runtime, which means the snippet will run much slower than:
341/// ```ignore
342/// #[tokio::main]
343/// async fn main() {
344///     tokio::spawn(async {
345///         for _ in 0..100 {
346///             let (tx, rx) = oneshot::channel();
347///             tokio::spawn(async move {
348///                 tx.send(());
349///             })
350///
351///             rx.await;
352///         }
353///     }).await;
354/// }
355/// ```
356/// The slowdown is caused by a higher time between the `rx` task being notified (in `tx.send()`)
357/// and the task being polled.
358///
359/// ##### Are other tasks polling too long without yielding?
360/// You suspect that your tasks are slow because they're backed up in tokio's scheduling queues. For
361/// *each* of your application's [`TaskMonitor`]s you check to see [*if their associated tasks are
362/// taking longer to poll...*](#are-my-tasks-taking-longer-to-poll)
363///
364/// ### Limitations
365/// The [`TaskMetrics`] type uses [`u64`] to represent both event counters and durations (measured
366/// in nanoseconds). Consequently, event counters are accurate for ≤ [`u64::MAX`] events, and
367/// durations are accurate for ≤ [`u64::MAX`] nanoseconds.
368///
369/// The counters and durations of [`TaskMetrics`] produced by [`TaskMonitor::cumulative`] increase
370/// monotonically with each successive invocation of [`TaskMonitor::cumulative`]. Upon overflow,
371/// counters and durations wrap.
372///
373/// The counters and durations of [`TaskMetrics`] produced by [`TaskMonitor::intervals`] are
374/// calculated by computing the difference of metrics in successive invocations of
375/// [`TaskMonitor::cumulative`]. If, within a monitoring interval, an event occurs more than
376/// [`u64::MAX`] times, or a monitored duration exceeds [`u64::MAX`] nanoseconds, the metrics for
377/// that interval will overflow and not be accurate.
378///
379/// ##### Examples at the limits
380/// Consider the [`TaskMetrics::total_first_poll_delay`] metric. This metric accurately reflects
381/// delays between instrumentation and first-poll ≤ [`u64::MAX`] nanoseconds:
382/// ```
383/// use tokio::time::Duration;
384///
385/// #[tokio::main(flavor = "current_thread", start_paused = true)]
386/// async fn main() {
387///     let monitor = tokio_metrics::TaskMonitor::new();
388///     let mut interval = monitor.intervals();
389///     let mut next_interval = || interval.next().unwrap();
390///
391///     // construct and instrument a task, but do not `await` it
392///     let task = monitor.instrument(async {});
393///
394///     // this is the maximum duration representable by tokio_metrics
395///     let max_duration = Duration::from_nanos(u64::MAX);
396///
397///     // let's advance the clock by this amount and poll `task`
398///     let _ = tokio::time::advance(max_duration).await;
399///     task.await;
400///
401///     // durations ≤ `max_duration` are accurately reflected in this metric
402///     assert_eq!(next_interval().total_first_poll_delay, max_duration);
403///     assert_eq!(monitor.cumulative().total_first_poll_delay, max_duration);
404/// }
405/// ```
406/// If the total delay between instrumentation and first poll exceeds [`u64::MAX`] nanoseconds,
407/// [`total_first_poll_delay`][TaskMetrics::total_first_poll_delay] will overflow:
408/// ```
409/// # use tokio::time::Duration;
410/// #
411/// # #[tokio::main(flavor = "current_thread", start_paused = true)]
412/// # async fn main() {
413/// #    let monitor = tokio_metrics::TaskMonitor::new();
414/// #
415///  // construct and instrument a task, but do not `await` it
416///  let task_a = monitor.instrument(async {});
417///  let task_b = monitor.instrument(async {});
418///
419///  // this is the maximum duration representable by tokio_metrics
420///  let max_duration = Duration::from_nanos(u64::MAX);
421///
422///  // let's advance the clock by 1.5x this amount and await `task`
423///  let _ = tokio::time::advance(3 * (max_duration / 2)).await;
424///  task_a.await;
425///  task_b.await;
426///
427///  // the `total_first_poll_delay` has overflowed
428///  assert!(monitor.cumulative().total_first_poll_delay < max_duration);
429/// # }
430/// ```
431/// If *many* tasks are spawned, it will take far less than a [`u64::MAX`]-nanosecond delay to bring
432/// this metric to the precipice of overflow:
433/// ```
434/// # use tokio::time::Duration;
435/// #
436/// # #[tokio::main(flavor = "current_thread", start_paused = true)]
437/// # async fn main() {
438/// #     let monitor = tokio_metrics::TaskMonitor::new();
439/// #     let mut interval = monitor.intervals();
440/// #     let mut next_interval = || interval.next().unwrap();
441/// #
442/// // construct and instrument u16::MAX tasks, but do not `await` them
443/// let first_poll_count = u16::MAX as u64;
444/// let mut tasks = Vec::with_capacity(first_poll_count as usize);
445/// for _ in 0..first_poll_count { tasks.push(monitor.instrument(async {})); }
446///
447/// // this is the maximum duration representable by tokio_metrics
448/// let max_duration = u64::MAX;
449///
450/// // let's advance the clock justenough such that all of the time-to-first-poll
451/// // delays summed nearly equals `max_duration_nanos`, less some remainder...
452/// let iffy_delay = max_duration / (first_poll_count as u64);
453/// let small_remainder = max_duration % first_poll_count;
454/// let _ = tokio::time::advance(Duration::from_nanos(iffy_delay)).await;
455///
456/// // ...then poll all of the instrumented tasks:
457/// for task in tasks { task.await; }
458///
459/// // `total_first_poll_delay` is at the precipice of overflowing!
460/// assert_eq!(
461///     next_interval().total_first_poll_delay.as_nanos(),
462///     (max_duration - small_remainder) as u128
463/// );
464/// assert_eq!(
465///     monitor.cumulative().total_first_poll_delay.as_nanos(),
466///     (max_duration - small_remainder) as u128
467/// );
468/// # }
469/// ```
470/// Frequent, interval-sampled metrics will retain their accuracy, even if the cumulative
471/// metrics counter overflows at most once in the midst of an interval:
472/// ```
473/// # use tokio::time::Duration;
474/// # use tokio_metrics::TaskMonitor;
475/// #
476/// # #[tokio::main(flavor = "current_thread", start_paused = true)]
477/// # async fn main() {
478/// #     let monitor = TaskMonitor::new();
479/// #     let mut interval = monitor.intervals();
480/// #     let mut next_interval = || interval.next().unwrap();
481/// #
482///  let first_poll_count = u16::MAX as u64;
483///  let batch_size = first_poll_count / 3;
484///
485///  let max_duration_ns = u64::MAX;
486///  let iffy_delay_ns = max_duration_ns / first_poll_count;
487///
488///  // Instrument `batch_size` number of tasks, wait for `delay` nanoseconds,
489///  // then await the instrumented tasks.
490///  async fn run_batch(monitor: &TaskMonitor, batch_size: usize, delay: u64) {
491///      let mut tasks = Vec::with_capacity(batch_size);
492///      for _ in 0..batch_size { tasks.push(monitor.instrument(async {})); }
493///      let _ = tokio::time::advance(Duration::from_nanos(delay)).await;
494///      for task in tasks { task.await; }
495///  }
496///
497///  // this is how much `total_time_to_first_poll_ns` will
498///  // increase with each batch we run
499///  let batch_delay = iffy_delay_ns * batch_size;
500///
501///  // run batches 1, 2, and 3
502///  for i in 1..=3 {
503///      run_batch(&monitor, batch_size as usize, iffy_delay_ns).await;
504///      assert_eq!(1 * batch_delay as u128, next_interval().total_first_poll_delay.as_nanos());
505///      assert_eq!(i * batch_delay as u128, monitor.cumulative().total_first_poll_delay.as_nanos());
506///  }
507///
508///  /* now, the `total_time_to_first_poll_ns` counter is at the precipice of overflow */
509///  assert_eq!(monitor.cumulative().total_first_poll_delay.as_nanos(), max_duration_ns as u128);
510///
511///  // run batch 4
512///  run_batch(&monitor, batch_size as usize, iffy_delay_ns).await;
513///  // the interval counter remains accurate
514///  assert_eq!(1 * batch_delay as u128, next_interval().total_first_poll_delay.as_nanos());
515///  // but the cumulative counter has overflowed
516///  assert_eq!(batch_delay as u128 - 1, monitor.cumulative().total_first_poll_delay.as_nanos());
517/// # }
518/// ```
519/// If a cumulative metric overflows *more than once* in the midst of an interval,
520/// its interval-sampled counterpart will also overflow.
521#[derive(Clone, Debug)]
522pub struct TaskMonitor {
523    base: Arc<TaskMonitorCore>,
524}
525
526impl Deref for TaskMonitor {
527    type Target = TaskMonitorCore;
528
529    fn deref(&self) -> &Self::Target {
530        &self.base
531    }
532}
533
534impl AsRef<TaskMonitorCore> for TaskMonitor {
535    fn as_ref(&self) -> &TaskMonitorCore {
536        &self.base
537    }
538}
539
540/// A non-`Clone`, non-allocated, static-friendly version of [`TaskMonitor`].
541/// See full docs on the [`TaskMonitor`] struct.
542///
543/// You should use [`TaskMonitorCore`] if you have a known count of monitors
544/// that you want to initialize as compile-time `static` structs.
545///
546/// You can also use [`TaskMonitorCore`] if you are already passing around an `Arc`-wrapped
547/// struct that you want to store your monitor in. This way, you can avoid double-`Arc`'ing it.
548///
549/// For other most other non-static usage, [`TaskMonitor`] will be more ergonomic.
550///
551/// ##### Examples
552///
553/// Static usage:
554/// ```
555/// use tokio_metrics::TaskMonitorCore;
556///
557/// static MONITOR: TaskMonitorCore = TaskMonitorCore::new();
558///
559/// #[tokio::main]
560/// async fn main() {
561///     assert_eq!(MONITOR.cumulative().first_poll_count, 0);
562///
563///     MONITOR.instrument(async {}).await;
564///     assert_eq!(MONITOR.cumulative().first_poll_count, 1);
565/// }
566/// ```
567///
568/// Usage with wrapper struct and [`TaskMonitorCore::instrument_with`]:
569/// ```
570/// use std::sync::Arc;
571/// use tokio_metrics::TaskMonitorCore;
572///
573/// #[derive(Clone)]
574/// struct SharedState(Arc<SharedStateInner>);
575/// struct SharedStateInner {
576///     monitor: TaskMonitorCore,
577///     other_state: SomeOtherSharedState,
578/// }
579/// /// Imagine: a type that wasn't `Clone` that you want to pass around
580/// /// in a similar way as the monitor
581/// struct SomeOtherSharedState;
582///
583/// impl AsRef<TaskMonitorCore> for SharedState {
584///     fn as_ref(&self) -> &TaskMonitorCore {
585///         &self.0.monitor
586///     }
587/// }
588///
589/// #[tokio::main]
590/// async fn main() {
591///     let state = SharedState(Arc::new(SharedStateInner {
592///         monitor: TaskMonitorCore::new(),
593///         other_state: SomeOtherSharedState,
594///     }));
595///
596///     assert_eq!(state.0.monitor.cumulative().first_poll_count, 0);
597///
598///     TaskMonitorCore::instrument_with(async {}, state.clone()).await;
599///     assert_eq!(state.0.monitor.cumulative().first_poll_count, 1);
600/// }
601/// ```
602#[derive(Debug)]
603pub struct TaskMonitorCore {
604    metrics: RawMetrics,
605    /// Whether instrumented tasks should publish their scheduling delay as a
606    /// task-local for [`FutureMonitor`] to sample. Off by default so the common
607    /// case pays nothing; enabled via
608    /// [`TaskMonitorBuilder::publish_scheduling_delay`]. Kept out of
609    /// [`RawMetrics`] so that struct's (heavily contended) layout is unchanged.
610    record_scheduling_log: bool,
611}
612
613/// Provides an interface for constructing a [`TaskMonitor`] with specialized configuration
614/// parameters.
615#[derive(Clone, Debug, Default)]
616pub struct TaskMonitorBuilder(TaskMonitorCoreBuilder);
617
618impl TaskMonitorBuilder {
619    /// Creates a new [`TaskMonitorBuilder`].
620    pub fn new() -> Self {
621        Self(TaskMonitorCoreBuilder::new())
622    }
623
624    /// Specifies the threshold at which polls are considered 'slow'.
625    pub fn with_slow_poll_threshold(&mut self, threshold: Duration) -> &mut Self {
626        self.0.slow_poll_threshold = Some(threshold);
627        self
628    }
629
630    /// Specifies the threshold at which schedules are considered 'long'.
631    pub fn with_long_delay_threshold(&mut self, threshold: Duration) -> &mut Self {
632        self.0.long_delay_threshold = Some(threshold);
633        self
634    }
635
636    /// Records each instrumented task's scheduling delay so that a
637    /// [`FutureMonitor`] running within the task can attribute scheduling delay
638    /// to a single future via [`TaskScheduling`].
639    ///
640    /// Off by default: tasks instrumented by a monitor without this enabled pay
641    /// no extra per-poll cost. Enable it on the monitor wrapping the larger task
642    /// when you want [`FutureMonitor`]'s scheduling metrics to be populated.
643    pub fn publish_scheduling_delay(&mut self) -> &mut Self {
644        self.0.record_scheduling_log = true;
645        self
646    }
647
648    /// Consume the builder, producing a [`TaskMonitor`].
649    pub fn build(self) -> TaskMonitor {
650        TaskMonitor {
651            base: Arc::new(self.0.build()),
652        }
653    }
654}
655
656/// Provides an interface for constructing a [`TaskMonitorCore`] with specialized configuration
657/// parameters.
658///
659/// ```
660/// use std::time::Duration;
661/// use tokio_metrics::TaskMonitorCoreBuilder;
662///
663/// static MONITOR: tokio_metrics::TaskMonitorCore = TaskMonitorCoreBuilder::new()
664///     .with_slow_poll_threshold(Duration::from_micros(100))
665///     .build();
666/// ```
667#[derive(Clone, Debug, Default)]
668pub struct TaskMonitorCoreBuilder {
669    slow_poll_threshold: Option<Duration>,
670    long_delay_threshold: Option<Duration>,
671    record_scheduling_log: bool,
672}
673
674impl TaskMonitorCoreBuilder {
675    /// Creates a new [`TaskMonitorCoreBuilder`].
676    pub const fn new() -> Self {
677        Self {
678            slow_poll_threshold: None,
679            long_delay_threshold: None,
680            record_scheduling_log: false,
681        }
682    }
683
684    /// Specifies the threshold at which polls are considered 'slow'.
685    pub const fn with_slow_poll_threshold(self, threshold: Duration) -> Self {
686        Self {
687            slow_poll_threshold: Some(threshold),
688            ..self
689        }
690    }
691
692    /// Specifies the threshold at which schedules are considered 'long'.
693    pub const fn with_long_delay_threshold(self, threshold: Duration) -> Self {
694        Self {
695            long_delay_threshold: Some(threshold),
696            ..self
697        }
698    }
699
700    /// Records each instrumented task's scheduling delay so that a
701    /// [`FutureMonitor`] running within the task can attribute scheduling delay
702    /// to a single future via [`TaskScheduling`].
703    ///
704    /// Off by default; see
705    /// [`TaskMonitorBuilder::publish_scheduling_delay`] for details.
706    pub const fn publish_scheduling_delay(self) -> Self {
707        Self {
708            record_scheduling_log: true,
709            ..self
710        }
711    }
712
713    /// Consume the builder, producing a [`TaskMonitorCore`].
714    pub const fn build(self) -> TaskMonitorCore {
715        let slow = match self.slow_poll_threshold {
716            Some(v) => v,
717            None => TaskMonitor::DEFAULT_SLOW_POLL_THRESHOLD,
718        };
719        let long = match self.long_delay_threshold {
720            Some(v) => v,
721            None => TaskMonitor::DEFAULT_LONG_DELAY_THRESHOLD,
722        };
723        TaskMonitorCore::create(slow, long, self.record_scheduling_log)
724    }
725}
726
727pin_project! {
728    /// An async task that has been instrumented with [`TaskMonitor::instrument`].
729    #[derive(Debug)]
730    pub struct Instrumented<T, M: AsRef<TaskMonitorCore> = TaskMonitor> {
731        // The task being instrumented
732        #[pin]
733        task: T,
734
735        // True when the task is polled for the first time
736        did_poll_once: bool,
737
738        // The instant, tracked as nanoseconds since `instrumented_at`, at which the future finished
739        // its last poll.
740        idled_at: u64,
741
742        // State shared between the task and its instrumented waker.
743        state: Arc<State<M>>,
744    }
745
746    impl<T, M: AsRef<TaskMonitorCore>> PinnedDrop for Instrumented<T, M> {
747        fn drop(this: Pin<&mut Self>) {
748            this.state.monitor.as_ref().metrics.dropped_count.fetch_add(1, SeqCst);
749        }
750    }
751}
752
753/// Key metrics of [instrumented][`TaskMonitor::instrument`] tasks.
754#[non_exhaustive]
755#[derive(Debug, Clone, Copy, Default)]
756pub struct TaskMetrics {
757    /// The number of tasks instrumented.
758    ///
759    /// ##### Examples
760    /// ```
761    /// #[tokio::main]
762    /// async fn main() {
763    ///     let monitor = tokio_metrics::TaskMonitor::new();
764    ///     let mut interval = monitor.intervals();
765    ///     let mut next_interval = || interval.next().unwrap();
766    ///
767    ///     // 0 tasks have been instrumented
768    ///     assert_eq!(next_interval().instrumented_count, 0);
769    ///
770    ///     monitor.instrument(async {});
771    ///
772    ///     // 1 task has been instrumented
773    ///     assert_eq!(next_interval().instrumented_count, 1);
774    ///
775    ///     monitor.instrument(async {});
776    ///     monitor.instrument(async {});
777    ///
778    ///     // 2 tasks have been instrumented
779    ///     assert_eq!(next_interval().instrumented_count, 2);
780    ///
781    ///     // since the last interval was produced, 0 tasks have been instrumented
782    ///     assert_eq!(next_interval().instrumented_count, 0);
783    /// }
784    /// ```
785    pub instrumented_count: u64,
786
787    /// The number of tasks dropped.
788    ///
789    /// ##### Examples
790    /// ```
791    /// #[tokio::main]
792    /// async fn main() {
793    ///     let monitor = tokio_metrics::TaskMonitor::new();
794    ///     let mut interval = monitor.intervals();
795    ///     let mut next_interval = || interval.next().unwrap();
796    ///
797    ///     // 0 tasks have been dropped
798    ///     assert_eq!(next_interval().dropped_count, 0);
799    ///
800    ///     let _task = monitor.instrument(async {});
801    ///
802    ///     // 0 tasks have been dropped
803    ///     assert_eq!(next_interval().dropped_count, 0);
804    ///
805    ///     monitor.instrument(async {}).await;
806    ///     drop(monitor.instrument(async {}));
807    ///
808    ///     // 2 tasks have been dropped
809    ///     assert_eq!(next_interval().dropped_count, 2);
810    ///
811    ///     // since the last interval was produced, 0 tasks have been dropped
812    ///     assert_eq!(next_interval().dropped_count, 0);
813    /// }
814    /// ```
815    pub dropped_count: u64,
816
817    /// The number of tasks polled for the first time.
818    ///
819    /// ##### Derived metrics
820    /// - **[`mean_first_poll_delay`][TaskMetrics::mean_first_poll_delay]**
821    ///   The mean duration elapsed between the instant tasks are instrumented, and the instant they
822    ///   are first polled.
823    ///
824    /// ##### Examples
825    /// In the below example, no tasks are instrumented or polled in the first sampling interval;
826    /// one task is instrumented (but not polled) in the second sampling interval; that task is
827    /// awaited to completion (and, thus, polled at least once) in the third sampling interval; no
828    /// additional tasks are polled for the first time within the fourth sampling interval:
829    /// ```
830    /// #[tokio::main]
831    /// async fn main() {
832    ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
833    ///     let mut interval = metrics_monitor.intervals();
834    ///     let mut next_interval = || interval.next().unwrap();
835    ///
836    ///     // no tasks have been constructed, instrumented, and polled at least once
837    ///     assert_eq!(next_interval().first_poll_count, 0);
838    ///
839    ///     let task = metrics_monitor.instrument(async {});
840    ///
841    ///     // `task` has been constructed and instrumented, but has not yet been polled
842    ///     assert_eq!(next_interval().first_poll_count, 0);
843    ///
844    ///     // poll `task` to completion
845    ///     task.await;
846    ///
847    ///     // `task` has been constructed, instrumented, and polled at least once
848    ///     assert_eq!(next_interval().first_poll_count, 1);
849    ///
850    ///     // since the last interval was produced, 0 tasks have been constructed, instrumented and polled
851    ///     assert_eq!(next_interval().first_poll_count, 0);
852    ///
853    /// }
854    /// ```
855    pub first_poll_count: u64,
856
857    /// The total duration elapsed between the instant tasks are instrumented, and the instant they
858    /// are first polled.
859    ///
860    /// ##### Derived metrics
861    /// - **[`mean_first_poll_delay`][TaskMetrics::mean_first_poll_delay]**
862    ///   The mean duration elapsed between the instant tasks are instrumented, and the instant they
863    ///   are first polled.
864    ///
865    /// ##### Examples
866    /// In the below example, 0 tasks have been instrumented or polled within the first sampling
867    /// interval, a total of 500ms elapse between the instrumentation and polling of tasks within
868    /// the second sampling interval, and a total of 350ms elapse between the instrumentation and
869    /// polling of tasks within the third sampling interval:
870    /// ```
871    /// use tokio::time::Duration;
872    ///
873    /// #[tokio::main(flavor = "current_thread", start_paused = true)]
874    /// async fn main() {
875    ///     let monitor = tokio_metrics::TaskMonitor::new();
876    ///     let mut interval = monitor.intervals();
877    ///     let mut next_interval = || interval.next().unwrap();
878    ///
879    ///     // no tasks have yet been created, instrumented, or polled
880    ///     assert_eq!(monitor.cumulative().total_first_poll_delay, Duration::ZERO);
881    ///     assert_eq!(next_interval().total_first_poll_delay, Duration::ZERO);
882    ///
883    ///     // constructs and instruments a task, pauses a given duration, then awaits the task
884    ///     async fn instrument_pause_await(monitor: &tokio_metrics::TaskMonitor, pause: Duration) {
885    ///         let task = monitor.instrument(async move {});
886    ///         tokio::time::sleep(pause).await;
887    ///         task.await;
888    ///     }
889    ///
890    ///     // construct and await a task that pauses for 500ms between instrumentation and first poll
891    ///     let task_a_pause_time = Duration::from_millis(500);
892    ///     instrument_pause_await(&monitor, task_a_pause_time).await;
893    ///
894    ///     assert_eq!(next_interval().total_first_poll_delay, task_a_pause_time);
895    ///     assert_eq!(monitor.cumulative().total_first_poll_delay, task_a_pause_time);
896    ///
897    ///     // construct and await a task that pauses for 250ms between instrumentation and first poll
898    ///     let task_b_pause_time = Duration::from_millis(250);
899    ///     instrument_pause_await(&monitor, task_b_pause_time).await;
900    ///
901    ///     // construct and await a task that pauses for 100ms between instrumentation and first poll
902    ///     let task_c_pause_time = Duration::from_millis(100);
903    ///     instrument_pause_await(&monitor, task_c_pause_time).await;
904    ///
905    ///     assert_eq!(
906    ///         next_interval().total_first_poll_delay,
907    ///         task_b_pause_time + task_c_pause_time
908    ///     );
909    ///     assert_eq!(
910    ///         monitor.cumulative().total_first_poll_delay,
911    ///         task_a_pause_time + task_b_pause_time + task_c_pause_time
912    ///     );
913    /// }
914    /// ```
915    ///
916    /// ##### When is this metric recorded?
917    /// The delay between instrumentation and first poll is not recorded until the first poll
918    /// actually occurs:
919    /// ```
920    /// # use tokio::time::Duration;
921    /// #
922    /// # #[tokio::main(flavor = "current_thread", start_paused = true)]
923    /// # async fn main() {
924    /// #     let monitor = tokio_metrics::TaskMonitor::new();
925    /// #     let mut interval = monitor.intervals();
926    /// #     let mut next_interval = || interval.next().unwrap();
927    /// #
928    /// // we construct and instrument a task, but do not `await` it
929    /// let task = monitor.instrument(async {});
930    ///
931    /// // let's sleep for 1s before we poll `task`
932    /// let one_sec = Duration::from_secs(1);
933    /// let _ = tokio::time::sleep(one_sec).await;
934    ///
935    /// // although 1s has now elapsed since the instrumentation of `task`,
936    /// // this is not reflected in `total_first_poll_delay`...
937    /// assert_eq!(next_interval().total_first_poll_delay, Duration::ZERO);
938    /// assert_eq!(monitor.cumulative().total_first_poll_delay, Duration::ZERO);
939    ///
940    /// // ...and won't be until `task` is actually polled
941    /// task.await;
942    ///
943    /// // now, the 1s delay is reflected in `total_first_poll_delay`:
944    /// assert_eq!(next_interval().total_first_poll_delay, one_sec);
945    /// assert_eq!(monitor.cumulative().total_first_poll_delay, one_sec);
946    /// # }
947    /// ```
948    ///
949    /// ##### What if first-poll-delay is very large?
950    /// The first-poll-delay of *individual* tasks saturates at `u64::MAX` nanoseconds. However, if
951    /// the *total* first-poll-delay *across* monitored tasks exceeds `u64::MAX` nanoseconds, this
952    /// metric will wrap around:
953    /// ```
954    /// use tokio::time::Duration;
955    ///
956    /// #[tokio::main(flavor = "current_thread", start_paused = true)]
957    /// async fn main() {
958    ///     let monitor = tokio_metrics::TaskMonitor::new();
959    ///
960    ///     // construct and instrument a task, but do not `await` it
961    ///     let task = monitor.instrument(async {});
962    ///
963    ///     // this is the maximum duration representable by tokio_metrics
964    ///     let max_duration = Duration::from_nanos(u64::MAX);
965    ///
966    ///     // let's advance the clock by double this amount and await `task`
967    ///     let _ = tokio::time::advance(max_duration * 2).await;
968    ///     task.await;
969    ///
970    ///     // the time-to-first-poll of `task` saturates at `max_duration`
971    ///     assert_eq!(monitor.cumulative().total_first_poll_delay, max_duration);
972    ///
973    ///     // ...but note that the metric *will* wrap around if more tasks are involved
974    ///     let task = monitor.instrument(async {});
975    ///     let _ = tokio::time::advance(Duration::from_nanos(1)).await;
976    ///     task.await;
977    ///     assert_eq!(monitor.cumulative().total_first_poll_delay, Duration::ZERO);
978    /// }
979    /// ```
980    pub total_first_poll_delay: Duration,
981
982    /// The total number of times that tasks idled, waiting to be awoken.
983    ///
984    /// An idle is recorded as occurring if a non-zero duration elapses between the instant a
985    /// task completes a poll, and the instant that it is next awoken.
986    ///
987    /// ##### Derived metrics
988    /// - **[`mean_idle_duration`][TaskMetrics::mean_idle_duration]**
989    ///   The mean duration of idles.
990    ///
991    /// ##### Examples
992    /// ```
993    /// #[tokio::main(flavor = "current_thread", start_paused = true)]
994    /// async fn main() {
995    ///     let monitor = tokio_metrics::TaskMonitor::new();
996    ///     let mut interval = monitor.intervals();
997    ///     let mut next_interval = move || interval.next().unwrap();
998    ///     let one_sec = std::time::Duration::from_secs(1);
999    ///
1000    ///     monitor.instrument(async {}).await;
1001    ///
1002    ///     assert_eq!(next_interval().total_idled_count, 0);
1003    ///     assert_eq!(monitor.cumulative().total_idled_count, 0);
1004    ///
1005    ///     monitor.instrument(async move {
1006    ///         tokio::time::sleep(one_sec).await;
1007    ///     }).await;
1008    ///
1009    ///     assert_eq!(next_interval().total_idled_count, 1);
1010    ///     assert_eq!(monitor.cumulative().total_idled_count, 1);
1011    ///
1012    ///     monitor.instrument(async {
1013    ///         tokio::time::sleep(one_sec).await;
1014    ///         tokio::time::sleep(one_sec).await;
1015    ///     }).await;
1016    ///
1017    ///     assert_eq!(next_interval().total_idled_count, 2);
1018    ///     assert_eq!(monitor.cumulative().total_idled_count, 3);
1019    /// }
1020    /// ```
1021    pub total_idled_count: u64,
1022
1023    /// The total duration that tasks idled.
1024    ///
1025    /// An idle is recorded as occurring if a non-zero duration elapses between the instant a
1026    /// task completes a poll, and the instant that it is next awoken.
1027    ///
1028    /// ##### Derived metrics
1029    /// - **[`mean_idle_duration`][TaskMetrics::mean_idle_duration]**
1030    ///   The mean duration of idles.
1031    ///
1032    /// ##### Examples
1033    /// ```
1034    /// #[tokio::main(flavor = "current_thread", start_paused = true)]
1035    /// async fn main() {
1036    ///     let monitor = tokio_metrics::TaskMonitor::new();
1037    ///     let mut interval = monitor.intervals();
1038    ///     let mut next_interval = move || interval.next().unwrap();
1039    ///     let one_sec = std::time::Duration::from_secs(1);
1040    ///     let two_sec = std::time::Duration::from_secs(2);
1041    ///
1042    ///     assert_eq!(next_interval().total_idle_duration.as_nanos(), 0);
1043    ///     assert_eq!(monitor.cumulative().total_idle_duration.as_nanos(), 0);
1044    ///
1045    ///     monitor.instrument(async move {
1046    ///         tokio::time::sleep(one_sec).await;
1047    ///     }).await;
1048    ///
1049    ///     assert_eq!(next_interval().total_idle_duration, one_sec);
1050    ///     assert_eq!(monitor.cumulative().total_idle_duration, one_sec);
1051    ///
1052    ///     monitor.instrument(async move {
1053    ///         tokio::time::sleep(two_sec).await;
1054    ///     }).await;
1055    ///
1056    ///     assert_eq!(next_interval().total_idle_duration, two_sec);
1057    ///     assert_eq!(monitor.cumulative().total_idle_duration, one_sec + two_sec);
1058    /// }
1059    /// ```
1060    pub total_idle_duration: Duration,
1061
1062    /// The maximum idle duration that a task took.
1063    ///
1064    /// An idle is recorded as occurring if a non-zero duration elapses between the instant a
1065    /// task completes a poll, and the instant that it is next awoken.
1066    ///
1067    /// ##### Examples
1068    /// ```
1069    /// #[tokio::main(flavor = "current_thread", start_paused = true)]
1070    /// async fn main() {
1071    ///     let monitor = tokio_metrics::TaskMonitor::new();
1072    ///     let mut interval = monitor.intervals();
1073    ///     let mut next_interval = move || interval.next().unwrap();
1074    ///     let one_sec = std::time::Duration::from_secs(1);
1075    ///     let two_sec = std::time::Duration::from_secs(2);
1076    ///
1077    ///     assert_eq!(next_interval().max_idle_duration.as_nanos(), 0);
1078    ///     assert_eq!(monitor.cumulative().max_idle_duration.as_nanos(), 0);
1079    ///
1080    ///     monitor.instrument(async move {
1081    ///         tokio::time::sleep(one_sec).await;
1082    ///     }).await;
1083    ///
1084    ///     assert_eq!(next_interval().max_idle_duration, one_sec);
1085    ///     assert_eq!(monitor.cumulative().max_idle_duration, one_sec);
1086    ///
1087    ///     monitor.instrument(async move {
1088    ///         tokio::time::sleep(two_sec).await;
1089    ///     }).await;
1090    ///
1091    ///     assert_eq!(next_interval().max_idle_duration, two_sec);
1092    ///     assert_eq!(monitor.cumulative().max_idle_duration, two_sec);
1093    ///
1094    ///     monitor.instrument(async move {
1095    ///         tokio::time::sleep(one_sec).await;
1096    ///     }).await;
1097    ///
1098    ///     assert_eq!(next_interval().max_idle_duration, one_sec);
1099    ///     assert_eq!(monitor.cumulative().max_idle_duration, two_sec);
1100    /// }
1101    /// ```
1102    pub max_idle_duration: Duration,
1103
1104    /// The total number of times that tasks were awoken (and then, presumably, scheduled for
1105    /// execution).
1106    ///
1107    /// ##### Definition
1108    /// This metric is equal to [`total_short_delay_count`][TaskMetrics::total_short_delay_count]
1109    /// \+ [`total_long_delay_count`][TaskMetrics::total_long_delay_count].
1110    ///
1111    /// ##### Derived metrics
1112    /// - **[`mean_scheduled_duration`][TaskMetrics::mean_scheduled_duration]**
1113    ///   The mean duration that tasks spent waiting to be executed after awakening.
1114    ///
1115    /// ##### Examples
1116    /// In the below example, a task yields to the scheduler a varying number of times between
1117    /// sampling intervals; this metric is equal to the number of times the task yielded:
1118    /// ```
1119    /// #[tokio::main]
1120    /// async fn main(){
1121    ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
1122    ///
1123    ///     // [A] no tasks have been created, instrumented, and polled more than once
1124    ///     assert_eq!(metrics_monitor.cumulative().total_scheduled_count, 0);
1125    ///
1126    ///     // [B] a `task` is created and instrumented
1127    ///     let task = {
1128    ///         let monitor = metrics_monitor.clone();
1129    ///         metrics_monitor.instrument(async move {
1130    ///             let mut interval = monitor.intervals();
1131    ///             let mut next_interval = move || interval.next().unwrap();
1132    ///
1133    ///             // [E] `task` has not yet yielded to the scheduler, and
1134    ///             // thus has not yet been scheduled since its first `poll`
1135    ///             assert_eq!(next_interval().total_scheduled_count, 0);
1136    ///
1137    ///             tokio::task::yield_now().await; // yield to the scheduler
1138    ///
1139    ///             // [F] `task` has yielded to the scheduler once (and thus been
1140    ///             // scheduled once) since the last sampling interval
1141    ///             assert_eq!(next_interval().total_scheduled_count, 1);
1142    ///
1143    ///             tokio::task::yield_now().await; // yield to the scheduler
1144    ///             tokio::task::yield_now().await; // yield to the scheduler
1145    ///             tokio::task::yield_now().await; // yield to the scheduler
1146    ///
1147    ///             // [G] `task` has yielded to the scheduler thrice (and thus been
1148    ///             // scheduled thrice) since the last sampling interval
1149    ///             assert_eq!(next_interval().total_scheduled_count, 3);
1150    ///
1151    ///             tokio::task::yield_now().await; // yield to the scheduler
1152    ///
1153    ///             next_interval
1154    ///         })
1155    ///     };
1156    ///
1157    ///     // [C] `task` has not yet been polled at all
1158    ///     assert_eq!(metrics_monitor.cumulative().first_poll_count, 0);
1159    ///     assert_eq!(metrics_monitor.cumulative().total_scheduled_count, 0);
1160    ///
1161    ///     // [D] poll `task` to completion
1162    ///     let mut next_interval = task.await;
1163    ///
1164    ///     // [H] `task` has been polled 1 times since the last sample
1165    ///     assert_eq!(next_interval().total_scheduled_count, 1);
1166    ///
1167    ///     // [I] `task` has been polled 0 times since the last sample
1168    ///     assert_eq!(next_interval().total_scheduled_count, 0);
1169    ///
1170    ///     // [J] `task` has yielded to the scheduler a total of five times
1171    ///     assert_eq!(metrics_monitor.cumulative().total_scheduled_count, 5);
1172    /// }
1173    /// ```
1174    #[doc(alias = "total_delay_count")]
1175    pub total_scheduled_count: u64,
1176
1177    /// The total duration that tasks spent waiting to be polled after awakening.
1178    ///
1179    /// ##### Definition
1180    /// This metric is equal to [`total_short_delay_duration`][TaskMetrics::total_short_delay_duration]
1181    /// \+ [`total_long_delay_duration`][TaskMetrics::total_long_delay_duration].
1182    ///
1183    /// ##### Derived metrics
1184    /// - **[`mean_scheduled_duration`][TaskMetrics::mean_scheduled_duration]**
1185    ///   The mean duration that tasks spent waiting to be executed after awakening.
1186    ///
1187    /// ##### Examples
1188    /// ```
1189    /// use tokio::time::Duration;
1190    ///
1191    /// #[tokio::main(flavor = "current_thread")]
1192    /// async fn main() {
1193    ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
1194    ///     let mut interval = metrics_monitor.intervals();
1195    ///     let mut next_interval = || interval.next().unwrap();
1196    ///
1197    ///     // construct and instrument and spawn a task that yields endlessly
1198    ///     tokio::spawn(metrics_monitor.instrument(async {
1199    ///         loop { tokio::task::yield_now().await }
1200    ///     }));
1201    ///
1202    ///     tokio::task::yield_now().await;
1203    ///
1204    ///     // block the executor for 1 second
1205    ///     std::thread::sleep(Duration::from_millis(1000));
1206    ///
1207    ///     tokio::task::yield_now().await;
1208    ///
1209    ///     // `endless_task` will have spent approximately one second waiting
1210    ///     let total_scheduled_duration = next_interval().total_scheduled_duration;
1211    ///     assert!(total_scheduled_duration >= Duration::from_millis(1000));
1212    ///     assert!(total_scheduled_duration <= Duration::from_millis(1100));
1213    /// }
1214    /// ```
1215    #[doc(alias = "total_delay_duration")]
1216    pub total_scheduled_duration: Duration,
1217
1218    /// The total number of times that tasks were polled.
1219    ///
1220    /// ##### Definition
1221    /// This metric is equal to [`total_fast_poll_count`][TaskMetrics::total_fast_poll_count]
1222    /// \+ [`total_slow_poll_count`][TaskMetrics::total_slow_poll_count].
1223    ///
1224    /// ##### Derived metrics
1225    /// - **[`mean_poll_duration`][TaskMetrics::mean_poll_duration]**
1226    ///   The mean duration of polls.
1227    ///
1228    /// ##### Examples
1229    /// In the below example, a task with multiple yield points is await'ed to completion; this
1230    /// metric reflects the number of `await`s within each sampling interval:
1231    /// ```
1232    /// #[tokio::main]
1233    /// async fn main() {
1234    ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
1235    ///
1236    ///     // [A] no tasks have been created, instrumented, and polled more than once
1237    ///     assert_eq!(metrics_monitor.cumulative().first_poll_count, 0);
1238    ///
1239    ///     // [B] a `task` is created and instrumented
1240    ///     let task = {
1241    ///         let monitor = metrics_monitor.clone();
1242    ///         metrics_monitor.instrument(async move {
1243    ///             let mut interval = monitor.intervals();
1244    ///             let mut next_interval = move || interval.next().unwrap();
1245    ///
1246    ///             // [E] task is in the midst of its first poll
1247    ///             assert_eq!(next_interval().total_poll_count, 0);
1248    ///
1249    ///             tokio::task::yield_now().await; // poll 1
1250    ///
1251    ///             // [F] task has been polled 1 time
1252    ///             assert_eq!(next_interval().total_poll_count, 1);
1253    ///
1254    ///             tokio::task::yield_now().await; // poll 2
1255    ///             tokio::task::yield_now().await; // poll 3
1256    ///             tokio::task::yield_now().await; // poll 4
1257    ///
1258    ///             // [G] task has been polled 3 times
1259    ///             assert_eq!(next_interval().total_poll_count, 3);
1260    ///
1261    ///             tokio::task::yield_now().await; // poll 5
1262    ///
1263    ///             next_interval                   // poll 6
1264    ///         })
1265    ///     };
1266    ///
1267    ///     // [C] `task` has not yet been polled at all
1268    ///     assert_eq!(metrics_monitor.cumulative().total_poll_count, 0);
1269    ///
1270    ///     // [D] poll `task` to completion
1271    ///     let mut next_interval = task.await;
1272    ///
1273    ///     // [H] `task` has been polled 2 times since the last sample
1274    ///     assert_eq!(next_interval().total_poll_count, 2);
1275    ///
1276    ///     // [I] `task` has been polled 0 times since the last sample
1277    ///     assert_eq!(next_interval().total_poll_count, 0);
1278    ///
1279    ///     // [J] `task` has been polled 6 times
1280    ///     assert_eq!(metrics_monitor.cumulative().total_poll_count, 6);
1281    /// }
1282    /// ```
1283    pub total_poll_count: u64,
1284
1285    /// The total duration elapsed during polls.
1286    ///
1287    /// ##### Definition
1288    /// This metric is equal to [`total_fast_poll_duration`][TaskMetrics::total_fast_poll_duration]
1289    /// \+ [`total_slow_poll_duration`][TaskMetrics::total_slow_poll_duration].
1290    ///
1291    /// ##### Derived metrics
1292    /// - **[`mean_poll_duration`][TaskMetrics::mean_poll_duration]**
1293    ///   The mean duration of polls.
1294    ///
1295    /// #### Examples
1296    /// ```
1297    /// use tokio::time::Duration;
1298    ///
1299    /// #[tokio::main(flavor = "current_thread", start_paused = true)]
1300    /// async fn main() {
1301    ///     let monitor = tokio_metrics::TaskMonitor::new();
1302    ///     let mut interval = monitor.intervals();
1303    ///     let mut next_interval = move || interval.next().unwrap();
1304    ///
1305    ///     assert_eq!(next_interval().total_poll_duration, Duration::ZERO);
1306    ///
1307    ///     monitor.instrument(async {
1308    ///         tokio::time::advance(Duration::from_secs(1)).await; // poll 1 (1s)
1309    ///         tokio::time::advance(Duration::from_secs(1)).await; // poll 2 (1s)
1310    ///         ()                                                  // poll 3 (0s)
1311    ///     }).await;
1312    ///
1313    ///     assert_eq!(next_interval().total_poll_duration, Duration::from_secs(2));
1314    /// }
1315    /// ```
1316    pub total_poll_duration: Duration,
1317
1318    /// The total number of times that polling tasks completed swiftly.
1319    ///
1320    /// Here, 'swiftly' is defined as completing in strictly less time than
1321    /// [`slow_poll_threshold`][TaskMonitor::slow_poll_threshold].
1322    ///
1323    /// ##### Derived metrics
1324    /// - **[`mean_fast_poll_duration`][TaskMetrics::mean_fast_poll_duration]**
1325    ///   The mean duration of fast polls.
1326    ///
1327    /// ##### Examples
1328    /// In the below example, 0 polls occur within the first sampling interval, 3 fast polls occur
1329    /// within the second sampling interval, and 2 fast polls occur within the third sampling
1330    /// interval:
1331    /// ```
1332    /// use std::future::Future;
1333    /// use std::time::Duration;
1334    ///
1335    /// #[tokio::main]
1336    /// async fn main() {
1337    ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
1338    ///     let mut interval = metrics_monitor.intervals();
1339    ///     let mut next_interval = || interval.next().unwrap();
1340    ///
1341    ///     // no tasks have been constructed, instrumented, or polled
1342    ///     assert_eq!(next_interval().total_fast_poll_count, 0);
1343    ///
1344    ///     let fast = Duration::ZERO;
1345    ///
1346    ///     // this task completes in three fast polls
1347    ///     let _ = metrics_monitor.instrument(async {
1348    ///         spin_for(fast).await; // fast poll 1
1349    ///         spin_for(fast).await; // fast poll 2
1350    ///         spin_for(fast)        // fast poll 3
1351    ///     }).await;
1352    ///
1353    ///     assert_eq!(next_interval().total_fast_poll_count, 3);
1354    ///
1355    ///     // this task completes in two fast polls
1356    ///     let _ = metrics_monitor.instrument(async {
1357    ///         spin_for(fast).await; // fast poll 1
1358    ///         spin_for(fast)        // fast poll 2
1359    ///     }).await;
1360    ///
1361    ///     assert_eq!(next_interval().total_fast_poll_count, 2);
1362    /// }
1363    ///
1364    /// /// Block the current thread for a given `duration`, then (optionally) yield to the scheduler.
1365    /// fn spin_for(duration: Duration) -> impl Future<Output=()> {
1366    ///     let start = tokio::time::Instant::now();
1367    ///     while start.elapsed() <= duration {}
1368    ///     tokio::task::yield_now()
1369    /// }
1370    /// ```
1371    pub total_fast_poll_count: u64,
1372
1373    /// The total duration of fast polls.
1374    ///
1375    /// Here, 'fast' is defined as completing in strictly less time than
1376    /// [`slow_poll_threshold`][TaskMonitor::slow_poll_threshold].
1377    ///
1378    /// ##### Derived metrics
1379    /// - **[`mean_fast_poll_duration`][TaskMetrics::mean_fast_poll_duration]**
1380    ///   The mean duration of fast polls.
1381    ///
1382    /// ##### Examples
1383    /// In the below example, no tasks are polled in the first sampling interval; three fast polls
1384    /// consume a total of 3μs time in the second sampling interval; and two fast polls consume a
1385    /// total of 2μs time in the third sampling interval:
1386    /// ```
1387    /// use std::future::Future;
1388    /// use std::time::Duration;
1389    ///
1390    /// #[tokio::main]
1391    /// async fn main() {
1392    ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
1393    ///     let mut interval = metrics_monitor.intervals();
1394    ///     let mut next_interval = || interval.next().unwrap();
1395    ///
1396    ///     // no tasks have been constructed, instrumented, or polled
1397    ///     let interval = next_interval();
1398    ///     assert_eq!(interval.total_fast_poll_duration, Duration::ZERO);
1399    ///
1400    ///     let fast = Duration::from_micros(1);
1401    ///
1402    ///     // this task completes in three fast polls
1403    ///     let task_a_time = time(metrics_monitor.instrument(async {
1404    ///         spin_for(fast).await; // fast poll 1
1405    ///         spin_for(fast).await; // fast poll 2
1406    ///         spin_for(fast)        // fast poll 3
1407    ///     })).await;
1408    ///
1409    ///     let interval = next_interval();
1410    ///     assert!(interval.total_fast_poll_duration >= fast * 3);
1411    ///     assert!(interval.total_fast_poll_duration <= task_a_time);
1412    ///
1413    ///     // this task completes in two fast polls
1414    ///     let task_b_time = time(metrics_monitor.instrument(async {
1415    ///         spin_for(fast).await; // fast poll 1
1416    ///         spin_for(fast)        // fast poll 2
1417    ///     })).await;
1418    ///
1419    ///     let interval = next_interval();
1420    ///     assert!(interval.total_fast_poll_duration >= fast * 2);
1421    ///     assert!(interval.total_fast_poll_duration <= task_b_time);
1422    /// }
1423    ///
1424    /// /// Produces the amount of time it took to await a given async task.
1425    /// async fn time(task: impl Future) -> Duration {
1426    ///     let start = tokio::time::Instant::now();
1427    ///     task.await;
1428    ///     start.elapsed()
1429    /// }
1430    ///
1431    /// /// Block the current thread for a given `duration`, then (optionally) yield to the scheduler.
1432    /// fn spin_for(duration: Duration) -> impl Future<Output=()> {
1433    ///     let start = tokio::time::Instant::now();
1434    ///     while start.elapsed() <= duration {}
1435    ///     tokio::task::yield_now()
1436    /// }
1437    /// ```
1438    pub total_fast_poll_duration: Duration,
1439
1440    /// The total number of times that polling tasks completed slowly.
1441    ///
1442    /// Here, 'slowly' is defined as completing in at least as much time as
1443    /// [`slow_poll_threshold`][TaskMonitor::slow_poll_threshold].
1444    ///
1445    /// ##### Derived metrics
1446    /// - **[`mean_slow_poll_duration`][`TaskMetrics::mean_slow_poll_duration`]**
1447    ///   The mean duration of slow polls.
1448    ///
1449    /// ##### Examples
1450    /// In the below example, 0 polls occur within the first sampling interval, 3 slow polls occur
1451    /// within the second sampling interval, and 2 slow polls occur within the third sampling
1452    /// interval:
1453    /// ```
1454    /// use std::future::Future;
1455    /// use std::time::Duration;
1456    ///
1457    /// #[tokio::main]
1458    /// async fn main() {
1459    ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
1460    ///     let mut interval = metrics_monitor.intervals();
1461    ///     let mut next_interval = || interval.next().unwrap();
1462    ///
1463    ///     // no tasks have been constructed, instrumented, or polled
1464    ///     assert_eq!(next_interval().total_slow_poll_count, 0);
1465    ///
1466    ///     let slow = 10 * metrics_monitor.slow_poll_threshold();
1467    ///
1468    ///     // this task completes in three slow polls
1469    ///     let _ = metrics_monitor.instrument(async {
1470    ///         spin_for(slow).await; // slow poll 1
1471    ///         spin_for(slow).await; // slow poll 2
1472    ///         spin_for(slow)        // slow poll 3
1473    ///     }).await;
1474    ///
1475    ///     assert_eq!(next_interval().total_slow_poll_count, 3);
1476    ///
1477    ///     // this task completes in two slow polls
1478    ///     let _ = metrics_monitor.instrument(async {
1479    ///         spin_for(slow).await; // slow poll 1
1480    ///         spin_for(slow)        // slow poll 2
1481    ///     }).await;
1482    ///
1483    ///     assert_eq!(next_interval().total_slow_poll_count, 2);
1484    /// }
1485    ///
1486    /// /// Block the current thread for a given `duration`, then (optionally) yield to the scheduler.
1487    /// fn spin_for(duration: Duration) -> impl Future<Output=()> {
1488    ///     let start = tokio::time::Instant::now();
1489    ///     while start.elapsed() <= duration {}
1490    ///     tokio::task::yield_now()
1491    /// }
1492    /// ```
1493    pub total_slow_poll_count: u64,
1494
1495    /// The total duration of slow polls.
1496    ///
1497    /// Here, 'slowly' is defined as completing in at least as much time as
1498    /// [`slow_poll_threshold`][TaskMonitor::slow_poll_threshold].
1499    ///
1500    /// ##### Derived metrics
1501    /// - **[`mean_slow_poll_duration`][`TaskMetrics::mean_slow_poll_duration`]**
1502    ///   The mean duration of slow polls.
1503    ///
1504    /// ##### Examples
1505    /// In the below example, no tasks are polled in the first sampling interval; three slow polls
1506    /// consume a total of
1507    /// 30 × [`DEFAULT_SLOW_POLL_THRESHOLD`][TaskMonitor::DEFAULT_SLOW_POLL_THRESHOLD]
1508    /// time in the second sampling interval; and two slow polls consume a total of
1509    /// 20 × [`DEFAULT_SLOW_POLL_THRESHOLD`][TaskMonitor::DEFAULT_SLOW_POLL_THRESHOLD] time in the
1510    /// third sampling interval:
1511    /// ```
1512    /// use std::future::Future;
1513    /// use std::time::Duration;
1514    ///
1515    /// #[tokio::main]
1516    /// async fn main() {
1517    ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
1518    ///     let mut interval = metrics_monitor.intervals();
1519    ///     let mut next_interval = || interval.next().unwrap();
1520    ///
1521    ///     // no tasks have been constructed, instrumented, or polled
1522    ///     let interval = next_interval();
1523    ///     assert_eq!(interval.total_slow_poll_duration, Duration::ZERO);
1524    ///
1525    ///     let slow = 10 * metrics_monitor.slow_poll_threshold();
1526    ///
1527    ///     // this task completes in three slow polls
1528    ///     let task_a_time = time(metrics_monitor.instrument(async {
1529    ///         spin_for(slow).await; // slow poll 1
1530    ///         spin_for(slow).await; // slow poll 2
1531    ///         spin_for(slow)        // slow poll 3
1532    ///     })).await;
1533    ///
1534    ///     let interval = next_interval();
1535    ///     assert!(interval.total_slow_poll_duration >= slow * 3);
1536    ///     assert!(interval.total_slow_poll_duration <= task_a_time);
1537    ///
1538    ///     // this task completes in two slow polls
1539    ///     let task_b_time = time(metrics_monitor.instrument(async {
1540    ///         spin_for(slow).await; // slow poll 1
1541    ///         spin_for(slow)        // slow poll 2
1542    ///     })).await;
1543    ///
1544    ///     let interval = next_interval();
1545    ///     assert!(interval.total_slow_poll_duration >= slow * 2);
1546    ///     assert!(interval.total_slow_poll_duration <= task_b_time);
1547    /// }
1548    ///
1549    /// /// Produces the amount of time it took to await a given async task.
1550    /// async fn time(task: impl Future) -> Duration {
1551    ///     let start = tokio::time::Instant::now();
1552    ///     task.await;
1553    ///     start.elapsed()
1554    /// }
1555    ///
1556    /// /// Block the current thread for a given `duration`, then (optionally) yield to the scheduler.
1557    /// fn spin_for(duration: Duration) -> impl Future<Output=()> {
1558    ///     let start = tokio::time::Instant::now();
1559    ///     while start.elapsed() <= duration {}
1560    ///     tokio::task::yield_now()
1561    /// }
1562    /// ```
1563    pub total_slow_poll_duration: Duration,
1564
1565    /// The total count of tasks with short scheduling delays.
1566    ///
1567    /// This is defined as tasks taking strictly less than
1568    /// [`long_delay_threshold`][TaskMonitor::long_delay_threshold] to be executed after being
1569    /// scheduled.
1570    ///
1571    /// ##### Derived metrics
1572    /// - **[`mean_short_delay_duration`][TaskMetrics::mean_short_delay_duration]**
1573    ///   The mean duration of short scheduling delays.
1574    pub total_short_delay_count: u64,
1575
1576    /// The total duration of tasks with short scheduling delays.
1577    ///
1578    /// This is defined as tasks taking strictly less than
1579    /// [`long_delay_threshold`][TaskMonitor::long_delay_threshold] to be executed after being
1580    /// scheduled.
1581    ///
1582    /// ##### Derived metrics
1583    /// - **[`mean_short_delay_duration`][TaskMetrics::mean_short_delay_duration]**
1584    ///   The mean duration of short scheduling delays.
1585    pub total_short_delay_duration: Duration,
1586
1587    /// The total count of tasks with long scheduling delays.
1588    ///
1589    /// This is defined as tasks taking
1590    /// [`long_delay_threshold`][TaskMonitor::long_delay_threshold] or longer to be executed
1591    /// after being scheduled.
1592    ///
1593    /// ##### Derived metrics
1594    /// - **[`mean_long_delay_duration`][TaskMetrics::mean_long_delay_duration]**
1595    ///   The mean duration of short scheduling delays.
1596    pub total_long_delay_count: u64,
1597
1598    /// The total duration of tasks with long scheduling delays.
1599    ///
1600    /// This is defined as tasks taking
1601    /// [`long_delay_threshold`][TaskMonitor::long_delay_threshold] or longer to be executed
1602    /// after being scheduled.
1603    ///
1604    /// ##### Derived metrics
1605    /// - **[`mean_long_delay_duration`][TaskMetrics::mean_long_delay_duration]**
1606    ///   The mean duration of short scheduling delays.
1607    pub total_long_delay_duration: Duration,
1608}
1609
1610/// Tracks the metrics, shared across the various types.
1611#[derive(Debug)]
1612struct RawMetrics {
1613    /// A task poll takes longer than this, it is considered a slow poll.
1614    slow_poll_threshold: Duration,
1615
1616    /// A scheduling delay of at least this long will be considered a long delay
1617    long_delay_threshold: Duration,
1618
1619    /// Total number of instrumented tasks.
1620    instrumented_count: AtomicU64,
1621
1622    /// Total number of instrumented tasks polled at least once.
1623    first_poll_count: AtomicU64,
1624
1625    /// Total number of times tasks entered the `idle` state.
1626    total_idled_count: AtomicU64,
1627
1628    /// Total number of times tasks were scheduled.
1629    total_scheduled_count: AtomicU64,
1630
1631    /// Total number of times tasks were polled fast
1632    total_fast_poll_count: AtomicU64,
1633
1634    /// Total number of times tasks were polled slow
1635    total_slow_poll_count: AtomicU64,
1636
1637    /// Total number of times tasks had long delay,
1638    total_long_delay_count: AtomicU64,
1639
1640    /// Total number of times tasks had little delay
1641    total_short_delay_count: AtomicU64,
1642
1643    /// Total number of times tasks were dropped
1644    dropped_count: AtomicU64,
1645
1646    /// Total amount of time until the first poll
1647    total_first_poll_delay_ns: AtomicU64,
1648
1649    /// Total amount of time tasks spent in the `idle` state.
1650    total_idle_duration_ns: AtomicU64,
1651
1652    /// The longest time tasks spent in the `idle` state locally.
1653    /// This will be used to track the local max between interval
1654    /// metric snapshots.
1655    local_max_idle_duration_ns: AtomicU64,
1656
1657    /// The longest time tasks spent in the `idle` state.
1658    global_max_idle_duration_ns: AtomicU64,
1659
1660    /// Total amount of time tasks spent in the waking state.
1661    total_scheduled_duration_ns: AtomicU64,
1662
1663    /// Total amount of time tasks spent being polled below the slow cut off.
1664    total_fast_poll_duration_ns: AtomicU64,
1665
1666    /// Total amount of time tasks spent being polled above the slow cut off.
1667    total_slow_poll_duration: AtomicU64,
1668
1669    /// Total amount of time tasks spent being polled below the long delay cut off.
1670    total_short_delay_duration_ns: AtomicU64,
1671
1672    /// Total amount of time tasks spent being polled at or above the long delay cut off.
1673    total_long_delay_duration_ns: AtomicU64,
1674}
1675
1676#[derive(Debug)]
1677struct State<M> {
1678    /// Where metrics should be recorded
1679    monitor: M,
1680
1681    /// Instant at which the task was instrumented. This is used to track the time to first poll.
1682    instrumented_at: Instant,
1683
1684    /// The instant, tracked as nanoseconds since `instrumented_at`, at which the future
1685    /// was last woken.
1686    woke_at: AtomicU64,
1687
1688    /// Waker to forward notifications to.
1689    waker: AtomicWaker,
1690
1691    /// Cumulative scheduling info for this task, published as a task-local while
1692    /// the task is being polled so that nested futures (e.g. a per-future
1693    /// [`FutureMonitor`]) can attribute scheduling delay to themselves. `None`
1694    /// unless the monitor opted in via
1695    /// [`TaskMonitorBuilder::publish_scheduling_delay`], so the common case
1696    /// allocates nothing and pays no per-poll cost.
1697    log: Option<Arc<SchedulingLog>>,
1698}
1699
1700/// Cumulative scheduling-delay counters for a single instrumented task.
1701///
1702/// Scheduling delay (the time a task spends in the runtime's queues between
1703/// being woken and being polled) is only observable by the root future that the
1704/// runtime actually schedules. [`Instrumented`] records it here and publishes
1705/// the log as a task-local for the duration of each poll, so that a future
1706/// running *within* the task can read the delay accrued during its own lifetime
1707/// via [`TaskScheduling::try_current`].
1708#[derive(Debug, Default)]
1709struct SchedulingLog {
1710    scheduled_count: AtomicU64,
1711    scheduled_duration_ns: AtomicU64,
1712    long_delay_count: AtomicU64,
1713}
1714
1715thread_local! {
1716    /// The scheduling log of the innermost [`Instrumented`] task currently being
1717    /// polled on this thread, if any.
1718    ///
1719    /// This is a hand-rolled task-local built on a `thread_local!`: it is only
1720    /// set for the duration of a single synchronous poll (see
1721    /// [`SchedulingLogGuard`]) and never held across an `.await`, so the usual
1722    /// hazard of thread-locals in async code does not apply.
1723    static CURRENT_SCHEDULING_LOG: RefCell<Option<Arc<SchedulingLog>>> =
1724        const { RefCell::new(None) };
1725}
1726
1727/// Sets the current task's [`SchedulingLog`] for the duration of a poll,
1728/// restoring the previous value on drop. This mirrors what
1729/// `tokio::task::LocalKey::sync_scope` does, but without requiring the optional
1730/// `tokio` dependency, so it works with and without the `rt` feature.
1731struct SchedulingLogGuard(Option<Arc<SchedulingLog>>);
1732
1733impl SchedulingLogGuard {
1734    fn enter(log: Arc<SchedulingLog>) -> Self {
1735        let prev = CURRENT_SCHEDULING_LOG.with(|cell| cell.borrow_mut().replace(log));
1736        SchedulingLogGuard(prev)
1737    }
1738}
1739
1740impl Drop for SchedulingLogGuard {
1741    fn drop(&mut self) {
1742        let prev = self.0.take();
1743        CURRENT_SCHEDULING_LOG.with(|cell| *cell.borrow_mut() = prev);
1744    }
1745}
1746
1747/// A snapshot of the cumulative scheduling delay observed by the root
1748/// instrumented task that the current future is running on.
1749///
1750/// Obtain one with [`TaskScheduling::try_current`] from inside a task that has
1751/// been instrumented with [`TaskMonitor::instrument`]. Scheduling delay can only
1752/// be measured by the root future the runtime schedules, so a future running
1753/// within a larger task samples this at two points and reports the difference —
1754/// this is exactly what [`FutureMonitor`] does to attribute scheduling delay to
1755/// a single future.
1756#[non_exhaustive]
1757#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
1758pub struct TaskScheduling {
1759    /// The number of times the task was scheduled (woken, then waiting to be polled).
1760    pub scheduled_count: u64,
1761    /// The total time the task spent waiting to be polled after being woken.
1762    pub total_scheduled_duration: Duration,
1763    /// The number of scheduling delays that crossed the monitor's long-delay threshold.
1764    pub long_delay_count: u64,
1765}
1766
1767impl TaskScheduling {
1768    /// Reads the cumulative scheduling delay of the task currently being polled.
1769    ///
1770    /// Returns `None` when called outside the poll of a future instrumented with
1771    /// [`TaskMonitor::instrument`].
1772    pub fn try_current() -> Option<TaskScheduling> {
1773        CURRENT_SCHEDULING_LOG.with(|cell| {
1774            cell.borrow().as_ref().map(|log| TaskScheduling {
1775                scheduled_count: log.scheduled_count.load(SeqCst),
1776                total_scheduled_duration: Duration::from_nanos(
1777                    log.scheduled_duration_ns.load(SeqCst),
1778                ),
1779                long_delay_count: log.long_delay_count.load(SeqCst),
1780            })
1781        })
1782    }
1783
1784    /// The change in each counter relative to an `earlier` snapshot, saturating at zero.
1785    fn since(self, earlier: TaskScheduling) -> TaskScheduling {
1786        TaskScheduling {
1787            scheduled_count: self.scheduled_count.saturating_sub(earlier.scheduled_count),
1788            total_scheduled_duration: self
1789                .total_scheduled_duration
1790                .saturating_sub(earlier.total_scheduled_duration),
1791            long_delay_count: self
1792                .long_delay_count
1793                .saturating_sub(earlier.long_delay_count),
1794        }
1795    }
1796}
1797
1798impl TaskMonitor {
1799    /// The default duration at which polls cross the threshold into being categorized as 'slow' is
1800    /// 50μs.
1801    #[cfg(not(test))]
1802    pub const DEFAULT_SLOW_POLL_THRESHOLD: Duration = Duration::from_micros(50);
1803    #[cfg(test)]
1804    #[allow(missing_docs)]
1805    pub const DEFAULT_SLOW_POLL_THRESHOLD: Duration = Duration::from_millis(500);
1806
1807    /// The default duration at which schedules cross the threshold into being categorized as 'long'
1808    /// is 50μs.
1809    #[cfg(not(test))]
1810    pub const DEFAULT_LONG_DELAY_THRESHOLD: Duration = Duration::from_micros(50);
1811    #[cfg(test)]
1812    #[allow(missing_docs)]
1813    pub const DEFAULT_LONG_DELAY_THRESHOLD: Duration = Duration::from_millis(500);
1814
1815    /// Constructs a new task monitor.
1816    ///
1817    /// Uses [`Self::DEFAULT_SLOW_POLL_THRESHOLD`] as the threshold at which polls will be
1818    /// considered 'slow'.
1819    ///
1820    /// Uses [`Self::DEFAULT_LONG_DELAY_THRESHOLD`] as the threshold at which scheduling will be
1821    /// considered 'long'.
1822    pub fn new() -> TaskMonitor {
1823        TaskMonitor::with_slow_poll_threshold(Self::DEFAULT_SLOW_POLL_THRESHOLD)
1824    }
1825
1826    /// Constructs a builder for a task monitor.
1827    pub fn builder() -> TaskMonitorBuilder {
1828        TaskMonitorBuilder::new()
1829    }
1830
1831    /// Constructs a new task monitor with a given threshold at which polls are considered 'slow'.
1832    ///
1833    /// ##### Selecting an appropriate threshold
1834    /// TODO. What advice can we give here?
1835    ///
1836    /// ##### Examples
1837    /// In the below example, low-threshold and high-threshold monitors are constructed and
1838    /// instrument identical tasks; the low-threshold monitor reports4 slow polls, and the
1839    /// high-threshold monitor reports only 2 slow polls:
1840    /// ```
1841    /// use std::future::Future;
1842    /// use std::time::Duration;
1843    /// use tokio_metrics::TaskMonitor;
1844    ///
1845    /// #[tokio::main]
1846    /// async fn main() {
1847    ///     let lo_threshold = Duration::from_micros(10);
1848    ///     let hi_threshold = Duration::from_millis(10);
1849    ///
1850    ///     let lo_monitor = TaskMonitor::with_slow_poll_threshold(lo_threshold);
1851    ///     let hi_monitor = TaskMonitor::with_slow_poll_threshold(hi_threshold);
1852    ///
1853    ///     let make_task = || async {
1854    ///         spin_for(lo_threshold).await; // faster poll 1
1855    ///         spin_for(lo_threshold).await; // faster poll 2
1856    ///         spin_for(hi_threshold).await; // slower poll 3
1857    ///         spin_for(hi_threshold).await  // slower poll 4
1858    ///     };
1859    ///
1860    ///     lo_monitor.instrument(make_task()).await;
1861    ///     hi_monitor.instrument(make_task()).await;
1862    ///
1863    ///     // the low-threshold monitor reported 4 slow polls:
1864    ///     assert_eq!(lo_monitor.cumulative().total_slow_poll_count, 4);
1865    ///     // the high-threshold monitor reported only 2 slow polls:
1866    ///     assert_eq!(hi_monitor.cumulative().total_slow_poll_count, 2);
1867    /// }
1868    ///
1869    /// /// Block the current thread for a given `duration`, then (optionally) yield to the scheduler.
1870    /// fn spin_for(duration: Duration) -> impl Future<Output=()> {
1871    ///     let start = tokio::time::Instant::now();
1872    ///     while start.elapsed() <= duration {}
1873    ///     tokio::task::yield_now()
1874    /// }
1875    /// ```
1876    pub fn with_slow_poll_threshold(slow_poll_cut_off: Duration) -> TaskMonitor {
1877        let base =
1878            TaskMonitorCore::create(slow_poll_cut_off, Self::DEFAULT_LONG_DELAY_THRESHOLD, false);
1879        TaskMonitor {
1880            base: Arc::new(base),
1881        }
1882    }
1883
1884    /// Produces the duration greater-than-or-equal-to at which polls are categorized as slow.
1885    ///
1886    /// ##### Examples
1887    /// In the below example, [`TaskMonitor`] is initialized with [`TaskMonitor::new`];
1888    /// consequently, its slow-poll threshold equals [`TaskMonitor::DEFAULT_SLOW_POLL_THRESHOLD`]:
1889    /// ```
1890    /// use tokio_metrics::TaskMonitor;
1891    ///
1892    /// #[tokio::main]
1893    /// async fn main() {
1894    ///     let metrics_monitor = TaskMonitor::new();
1895    ///
1896    ///     assert_eq!(
1897    ///         metrics_monitor.slow_poll_threshold(),
1898    ///         TaskMonitor::DEFAULT_SLOW_POLL_THRESHOLD
1899    ///     );
1900    /// }
1901    /// ```
1902    pub fn slow_poll_threshold(&self) -> Duration {
1903        self.base.metrics.slow_poll_threshold
1904    }
1905
1906    /// Produces the duration greater-than-or-equal-to at which scheduling delays are categorized
1907    /// as long.
1908    pub fn long_delay_threshold(&self) -> Duration {
1909        self.base.metrics.long_delay_threshold
1910    }
1911
1912    /// Produces an instrumented façade around a given async task.
1913    ///
1914    /// ##### Examples
1915    /// Instrument an async task by passing it to [`TaskMonitor::instrument`]:
1916    /// ```
1917    /// #[tokio::main]
1918    /// async fn main() {
1919    ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
1920    ///
1921    ///     // 0 tasks have been instrumented, much less polled
1922    ///     assert_eq!(metrics_monitor.cumulative().first_poll_count, 0);
1923    ///
1924    ///     // instrument a task and poll it to completion
1925    ///     metrics_monitor.instrument(async {}).await;
1926    ///
1927    ///     // 1 task has been instrumented and polled
1928    ///     assert_eq!(metrics_monitor.cumulative().first_poll_count, 1);
1929    ///
1930    ///     // instrument a task and poll it to completion
1931    ///     metrics_monitor.instrument(async {}).await;
1932    ///
1933    ///     // 2 tasks have been instrumented and polled
1934    ///     assert_eq!(metrics_monitor.cumulative().first_poll_count, 2);
1935    /// }
1936    /// ```
1937    /// An aync task may be tracked by multiple [`TaskMonitor`]s; e.g.:
1938    /// ```
1939    /// #[tokio::main]
1940    /// async fn main() {
1941    ///     let monitor_a = tokio_metrics::TaskMonitor::new();
1942    ///     let monitor_b = tokio_metrics::TaskMonitor::new();
1943    ///
1944    ///     // 0 tasks have been instrumented, much less polled
1945    ///     assert_eq!(monitor_a.cumulative().first_poll_count, 0);
1946    ///     assert_eq!(monitor_b.cumulative().first_poll_count, 0);
1947    ///
1948    ///     // instrument a task and poll it to completion
1949    ///     monitor_a.instrument(monitor_b.instrument(async {})).await;
1950    ///
1951    ///     // 1 task has been instrumented and polled
1952    ///     assert_eq!(monitor_a.cumulative().first_poll_count, 1);
1953    ///     assert_eq!(monitor_b.cumulative().first_poll_count, 1);
1954    /// }
1955    /// ```
1956    /// It is also possible (but probably undesirable) to instrument an async task multiple times
1957    /// with the same [`TaskMonitor`]; e.g.:
1958    /// ```
1959    /// #[tokio::main]
1960    /// async fn main() {
1961    ///     let monitor = tokio_metrics::TaskMonitor::new();
1962    ///
1963    ///     // 0 tasks have been instrumented, much less polled
1964    ///     assert_eq!(monitor.cumulative().first_poll_count, 0);
1965    ///
1966    ///     // instrument a task and poll it to completion
1967    ///     monitor.instrument(monitor.instrument(async {})).await;
1968    ///
1969    ///     // 2 tasks have been instrumented and polled, supposedly
1970    ///     assert_eq!(monitor.cumulative().first_poll_count, 2);
1971    /// }
1972    /// ```
1973    pub fn instrument<F>(&self, task: F) -> Instrumented<F> {
1974        TaskMonitorCore::instrument_with(task, self.clone())
1975    }
1976
1977    /// Produces [`TaskMetrics`] for the tasks instrumented by this [`TaskMonitor`], collected since
1978    /// the construction of [`TaskMonitor`].
1979    ///
1980    /// ##### See also
1981    /// - [`TaskMonitor::intervals`]:
1982    ///   produces [`TaskMetrics`] for user-defined sampling intervals, instead of cumulatively
1983    ///
1984    /// ##### Examples
1985    /// In the below example, 0 polls occur within the first sampling interval, 3 slow polls occur
1986    /// within the second sampling interval, and 2 slow polls occur within the third sampling
1987    /// interval; five slow polls occur across all sampling intervals:
1988    /// ```
1989    /// use std::future::Future;
1990    /// use std::time::Duration;
1991    ///
1992    /// #[tokio::main]
1993    /// async fn main() {
1994    ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
1995    ///
1996    ///     // initialize a stream of sampling intervals
1997    ///     let mut intervals = metrics_monitor.intervals();
1998    ///     // each call of `next_interval` will produce metrics for the last sampling interval
1999    ///     let mut next_interval = || intervals.next().unwrap();
2000    ///
2001    ///     let slow = 10 * metrics_monitor.slow_poll_threshold();
2002    ///
2003    ///     // this task completes in three slow polls
2004    ///     let _ = metrics_monitor.instrument(async {
2005    ///         spin_for(slow).await; // slow poll 1
2006    ///         spin_for(slow).await; // slow poll 2
2007    ///         spin_for(slow)        // slow poll 3
2008    ///     }).await;
2009    ///
2010    ///     // in the previous sampling interval, there were 3 slow polls
2011    ///     assert_eq!(next_interval().total_slow_poll_count, 3);
2012    ///     assert_eq!(metrics_monitor.cumulative().total_slow_poll_count, 3);
2013    ///
2014    ///     // this task completes in two slow polls
2015    ///     let _ = metrics_monitor.instrument(async {
2016    ///         spin_for(slow).await; // slow poll 1
2017    ///         spin_for(slow)        // slow poll 2
2018    ///     }).await;
2019    ///
2020    ///     // in the previous sampling interval, there were 2 slow polls
2021    ///     assert_eq!(next_interval().total_slow_poll_count, 2);
2022    ///
2023    ///     // across all sampling interval, there were a total of 5 slow polls
2024    ///     assert_eq!(metrics_monitor.cumulative().total_slow_poll_count, 5);
2025    /// }
2026    ///
2027    /// /// Block the current thread for a given `duration`, then (optionally) yield to the scheduler.
2028    /// fn spin_for(duration: Duration) -> impl Future<Output=()> {
2029    ///     let start = tokio::time::Instant::now();
2030    ///     while start.elapsed() <= duration {}
2031    ///     tokio::task::yield_now()
2032    /// }
2033    /// ```
2034    pub fn cumulative(&self) -> TaskMetrics {
2035        self.base.metrics.metrics()
2036    }
2037
2038    /// Produces an unending iterator of metric sampling intervals.
2039    ///
2040    /// Each sampling interval is defined by the time elapsed between advancements of the iterator
2041    /// produced by [`TaskMonitor::intervals`]. The item type of this iterator is [`TaskMetrics`],
2042    /// which is a bundle of task metrics that describe *only* events occurring within that sampling
2043    /// interval.
2044    ///
2045    /// ##### Examples
2046    /// In the below example, 0 polls occur within the first sampling interval, 3 slow polls occur
2047    /// within the second sampling interval, and 2 slow polls occur within the third sampling
2048    /// interval; five slow polls occur across all sampling intervals:
2049    /// ```
2050    /// use std::future::Future;
2051    /// use std::time::Duration;
2052    ///
2053    /// #[tokio::main]
2054    /// async fn main() {
2055    ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
2056    ///
2057    ///     // initialize a stream of sampling intervals
2058    ///     let mut intervals = metrics_monitor.intervals();
2059    ///     // each call of `next_interval` will produce metrics for the last sampling interval
2060    ///     let mut next_interval = || intervals.next().unwrap();
2061    ///
2062    ///     let slow = 10 * metrics_monitor.slow_poll_threshold();
2063    ///
2064    ///     // this task completes in three slow polls
2065    ///     let _ = metrics_monitor.instrument(async {
2066    ///         spin_for(slow).await; // slow poll 1
2067    ///         spin_for(slow).await; // slow poll 2
2068    ///         spin_for(slow)        // slow poll 3
2069    ///     }).await;
2070    ///
2071    ///     // in the previous sampling interval, there were 3 slow polls
2072    ///     assert_eq!(next_interval().total_slow_poll_count, 3);
2073    ///
2074    ///     // this task completes in two slow polls
2075    ///     let _ = metrics_monitor.instrument(async {
2076    ///         spin_for(slow).await; // slow poll 1
2077    ///         spin_for(slow)        // slow poll 2
2078    ///     }).await;
2079    ///
2080    ///     // in the previous sampling interval, there were 2 slow polls
2081    ///     assert_eq!(next_interval().total_slow_poll_count, 2);
2082    ///
2083    ///     // across all sampling intervals, there were a total of 5 slow polls
2084    ///     assert_eq!(metrics_monitor.cumulative().total_slow_poll_count, 5);
2085    /// }
2086    ///
2087    /// /// Block the current thread for a given `duration`, then (optionally) yield to the scheduler.
2088    /// fn spin_for(duration: Duration) -> impl Future<Output=()> {
2089    ///     let start = tokio::time::Instant::now();
2090    ///     while start.elapsed() <= duration {}
2091    ///     tokio::task::yield_now()
2092    /// }
2093    /// ```
2094    pub fn intervals(&self) -> TaskIntervals {
2095        TaskIntervals {
2096            monitor: self.clone(),
2097            previous: None,
2098        }
2099    }
2100}
2101
2102impl TaskMonitorCore {
2103    /// Returns a const-friendly [`TaskMonitorCoreBuilder`].
2104    pub const fn builder() -> TaskMonitorCoreBuilder {
2105        TaskMonitorCoreBuilder::new()
2106    }
2107
2108    /// Constructs a new [`TaskMonitorCore`]. Refer to the struct documentation for more discussion
2109    /// of benefits compared to [`TaskMonitor`].
2110    ///
2111    /// Uses [`TaskMonitor::DEFAULT_SLOW_POLL_THRESHOLD`] as the threshold at which polls will be
2112    /// considered 'slow'.
2113    ///
2114    /// Uses [`TaskMonitor::DEFAULT_LONG_DELAY_THRESHOLD`] as the threshold at which scheduling will be
2115    /// considered 'long'.
2116    pub const fn new() -> TaskMonitorCore {
2117        TaskMonitorCore::with_slow_poll_threshold(TaskMonitor::DEFAULT_SLOW_POLL_THRESHOLD)
2118    }
2119
2120    /// Constructs a new task monitor with a given threshold at which polls are considered 'slow'.
2121    ///
2122    /// Refer to [`TaskMonitor::with_slow_poll_threshold`] for examples.
2123    pub const fn with_slow_poll_threshold(slow_poll_cut_off: Duration) -> TaskMonitorCore {
2124        Self::create(
2125            slow_poll_cut_off,
2126            TaskMonitor::DEFAULT_LONG_DELAY_THRESHOLD,
2127            false,
2128        )
2129    }
2130
2131    /// Produces the duration greater-than-or-equal-to at which polls are categorized as slow.
2132    ///
2133    /// Refer to [`TaskMonitor::slow_poll_threshold`] for examples.
2134    pub fn slow_poll_threshold(&self) -> Duration {
2135        self.metrics.slow_poll_threshold
2136    }
2137
2138    /// Produces the duration greater-than-or-equal-to at which scheduling delays are categorized
2139    /// as long.
2140    pub fn long_delay_threshold(&self) -> Duration {
2141        self.metrics.long_delay_threshold
2142    }
2143
2144    /// Produces an instrumented façade around a given async task.
2145    ///
2146    /// ##### Examples
2147    /// ```
2148    /// use tokio_metrics::TaskMonitorCore;
2149    ///
2150    /// static MONITOR: TaskMonitorCore = TaskMonitorCore::new();
2151    ///
2152    /// #[tokio::main]
2153    /// async fn main() {
2154    ///     assert_eq!(MONITOR.cumulative().first_poll_count, 0);
2155    ///
2156    ///     MONITOR.instrument(async {}).await;
2157    ///     assert_eq!(MONITOR.cumulative().first_poll_count, 1);
2158    /// }
2159    /// ```
2160    pub fn instrument<F>(&'static self, task: F) -> Instrumented<F, &'static Self> {
2161        Self::instrument_with(task, self)
2162    }
2163
2164    /// Produces an instrumented façade around a given async task, with an explicit monitor.
2165    ///
2166    /// Use this when you have a non-static monitor reference, such as an `Arc<TaskMonitorCore>`.
2167    ///
2168    /// ##### Examples
2169    /// ```
2170    /// use std::sync::Arc;
2171    /// use tokio_metrics::TaskMonitorCore;
2172    ///
2173    /// #[derive(Clone)]
2174    /// struct SharedState(Arc<SharedStateInner>);
2175    /// struct SharedStateInner {
2176    ///     monitor: TaskMonitorCore,
2177    ///     other_state: SomeOtherSharedState,
2178    /// }
2179    /// /// Imagine: a type that wasn't `Clone` that you want to pass around
2180    /// /// in a similar way as the monitor
2181    /// struct SomeOtherSharedState;
2182    ///
2183    /// impl AsRef<TaskMonitorCore> for SharedState {
2184    ///     fn as_ref(&self) -> &TaskMonitorCore {
2185    ///         &self.0.monitor
2186    ///     }
2187    /// }
2188    ///
2189    /// #[tokio::main]
2190    /// async fn main() {
2191    ///     let state = SharedState(Arc::new(SharedStateInner {
2192    ///         monitor: TaskMonitorCore::new(),
2193    ///         other_state: SomeOtherSharedState,
2194    ///     }));
2195    ///
2196    ///     assert_eq!(state.0.monitor.cumulative().first_poll_count, 0);
2197    ///
2198    ///     TaskMonitorCore::instrument_with(async {}, state.clone()).await;
2199    ///     assert_eq!(state.0.monitor.cumulative().first_poll_count, 1);
2200    /// }
2201    /// ```
2202    pub fn instrument_with<F, M: AsRef<TaskMonitorCore> + Send + Sync + 'static>(
2203        task: F,
2204        monitor: M,
2205    ) -> Instrumented<F, M> {
2206        monitor
2207            .as_ref()
2208            .metrics
2209            .instrumented_count
2210            .fetch_add(1, SeqCst);
2211
2212        let log = monitor
2213            .as_ref()
2214            .record_scheduling_log
2215            .then(|| Arc::new(SchedulingLog::default()));
2216
2217        let state: State<M> = State {
2218            monitor,
2219            instrumented_at: Instant::now(),
2220            woke_at: AtomicU64::new(0),
2221            waker: AtomicWaker::new(),
2222            log,
2223        };
2224
2225        let instrumented: Instrumented<F, M> = Instrumented {
2226            task,
2227            did_poll_once: false,
2228            idled_at: 0,
2229            state: Arc::new(state),
2230        };
2231
2232        instrumented
2233    }
2234
2235    /// Produces [`TaskMetrics`] for the tasks instrumented by this [`TaskMonitorCore`], collected since
2236    /// the construction of [`TaskMonitorCore`].
2237    ///
2238    /// ##### See also
2239    /// - [`TaskMonitorCore::intervals`]:
2240    ///   produces [`TaskMetrics`] for user-defined sampling intervals, instead of cumulatively
2241    ///
2242    /// See [`TaskMonitor::cumulative`] for examples.
2243    pub fn cumulative(&self) -> TaskMetrics {
2244        self.metrics.metrics()
2245    }
2246
2247    /// Produces an unending iterator of metric sampling intervals.
2248    ///
2249    /// Each sampling interval is defined by the time elapsed between advancements of the iterator
2250    /// produced by [`TaskMonitorCore::intervals`]. The item type of this iterator is [`TaskMetrics`],
2251    /// which is a bundle of task metrics that describe *only* events occurring within that sampling
2252    /// interval.
2253    ///
2254    /// ##### Examples
2255    /// The below example demonstrates construction of [`TaskIntervals`] with [`TaskMonitorCore`].
2256    ///
2257    /// See [`TaskMonitor::intervals`] for more usage examples.
2258    ///
2259    /// ```
2260    /// use std::sync::Arc;
2261    ///
2262    /// fn main() {
2263    ///     let metrics_monitor = Arc::new(tokio_metrics::TaskMonitorCore::new());
2264    ///
2265    ///     let mut _intervals = tokio_metrics::TaskMonitorCore::intervals(metrics_monitor);
2266    /// }
2267    /// ```
2268    pub fn intervals<Monitor: AsRef<TaskMonitorCore> + Send + Sync + 'static>(
2269        monitor: Monitor,
2270    ) -> TaskIntervals<Monitor> {
2271        let intervals: TaskIntervals<Monitor> = TaskIntervals {
2272            monitor,
2273            previous: None,
2274        };
2275
2276        intervals
2277    }
2278}
2279
2280impl AsRef<TaskMonitorCore> for TaskMonitorCore {
2281    fn as_ref(&self) -> &TaskMonitorCore {
2282        self
2283    }
2284}
2285
2286impl TaskMonitorCore {
2287    const fn create(
2288        slow_poll_cut_off: Duration,
2289        long_delay_cut_off: Duration,
2290        record_scheduling_log: bool,
2291    ) -> TaskMonitorCore {
2292        TaskMonitorCore {
2293            record_scheduling_log,
2294            metrics: RawMetrics {
2295                slow_poll_threshold: slow_poll_cut_off,
2296                first_poll_count: AtomicU64::new(0),
2297                total_idled_count: AtomicU64::new(0),
2298                total_scheduled_count: AtomicU64::new(0),
2299                total_fast_poll_count: AtomicU64::new(0),
2300                total_slow_poll_count: AtomicU64::new(0),
2301                total_long_delay_count: AtomicU64::new(0),
2302                instrumented_count: AtomicU64::new(0),
2303                dropped_count: AtomicU64::new(0),
2304                total_first_poll_delay_ns: AtomicU64::new(0),
2305                total_scheduled_duration_ns: AtomicU64::new(0),
2306                local_max_idle_duration_ns: AtomicU64::new(0),
2307                global_max_idle_duration_ns: AtomicU64::new(0),
2308                total_idle_duration_ns: AtomicU64::new(0),
2309                total_fast_poll_duration_ns: AtomicU64::new(0),
2310                total_slow_poll_duration: AtomicU64::new(0),
2311                total_short_delay_duration_ns: AtomicU64::new(0),
2312                long_delay_threshold: long_delay_cut_off,
2313                total_short_delay_count: AtomicU64::new(0),
2314                total_long_delay_duration_ns: AtomicU64::new(0),
2315            },
2316        }
2317    }
2318}
2319
2320impl RawMetrics {
2321    fn get_and_reset_local_max_idle_duration(&self) -> Duration {
2322        Duration::from_nanos(self.local_max_idle_duration_ns.swap(0, SeqCst))
2323    }
2324
2325    fn metrics(&self) -> TaskMetrics {
2326        let total_fast_poll_count = self.total_fast_poll_count.load(SeqCst);
2327        let total_slow_poll_count = self.total_slow_poll_count.load(SeqCst);
2328
2329        let total_fast_poll_duration =
2330            Duration::from_nanos(self.total_fast_poll_duration_ns.load(SeqCst));
2331        let total_slow_poll_duration =
2332            Duration::from_nanos(self.total_slow_poll_duration.load(SeqCst));
2333
2334        let total_poll_count = total_fast_poll_count.saturating_add(total_slow_poll_count);
2335        let total_poll_duration = total_fast_poll_duration.saturating_add(total_slow_poll_duration);
2336
2337        TaskMetrics {
2338            instrumented_count: self.instrumented_count.load(SeqCst),
2339            dropped_count: self.dropped_count.load(SeqCst),
2340
2341            total_poll_count,
2342            total_poll_duration,
2343            first_poll_count: self.first_poll_count.load(SeqCst),
2344            total_idled_count: self.total_idled_count.load(SeqCst),
2345            total_scheduled_count: self.total_scheduled_count.load(SeqCst),
2346            total_fast_poll_count: self.total_fast_poll_count.load(SeqCst),
2347            total_slow_poll_count: self.total_slow_poll_count.load(SeqCst),
2348            total_short_delay_count: self.total_short_delay_count.load(SeqCst),
2349            total_long_delay_count: self.total_long_delay_count.load(SeqCst),
2350            total_first_poll_delay: Duration::from_nanos(
2351                self.total_first_poll_delay_ns.load(SeqCst),
2352            ),
2353            max_idle_duration: Duration::from_nanos(self.global_max_idle_duration_ns.load(SeqCst)),
2354            total_idle_duration: Duration::from_nanos(self.total_idle_duration_ns.load(SeqCst)),
2355            total_scheduled_duration: Duration::from_nanos(
2356                self.total_scheduled_duration_ns.load(SeqCst),
2357            ),
2358            total_fast_poll_duration: Duration::from_nanos(
2359                self.total_fast_poll_duration_ns.load(SeqCst),
2360            ),
2361            total_slow_poll_duration: Duration::from_nanos(
2362                self.total_slow_poll_duration.load(SeqCst),
2363            ),
2364            total_short_delay_duration: Duration::from_nanos(
2365                self.total_short_delay_duration_ns.load(SeqCst),
2366            ),
2367            total_long_delay_duration: Duration::from_nanos(
2368                self.total_long_delay_duration_ns.load(SeqCst),
2369            ),
2370        }
2371    }
2372}
2373
2374impl Default for TaskMonitor {
2375    fn default() -> TaskMonitor {
2376        TaskMonitor::new()
2377    }
2378}
2379
2380impl Default for TaskMonitorCore {
2381    fn default() -> TaskMonitorCore {
2382        TaskMonitorCore::new()
2383    }
2384}
2385
2386derived_metrics!(
2387    [TaskMetrics] {
2388        stable {
2389            /// The mean duration elapsed between the instant tasks are instrumented, and the instant they
2390            /// are first polled.
2391            ///
2392            /// ##### Definition
2393            /// This metric is derived from [`total_first_poll_delay`][TaskMetrics::total_first_poll_delay]
2394            /// ÷ [`first_poll_count`][TaskMetrics::first_poll_count].
2395            ///
2396            /// ##### Interpretation
2397            /// If this metric increases, it means that, on average, tasks spent longer waiting to be
2398            /// initially polled.
2399            ///
2400            /// ##### See also
2401            /// - **[`mean_scheduled_duration`][TaskMetrics::mean_scheduled_duration]**
2402            ///   The mean duration that tasks spent waiting to be executed after awakening.
2403            ///
2404            /// ##### Examples
2405            /// In the below example, no tasks are instrumented or polled within the first sampling
2406            /// interval; in the second sampling interval, 500ms elapse between the instrumentation of a
2407            /// task and its first poll; in the third sampling interval, a mean of 750ms elapse between the
2408            /// instrumentation and first poll of two tasks:
2409            /// ```
2410            /// use std::time::Duration;
2411            ///
2412            /// #[tokio::main]
2413            /// async fn main() {
2414            ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
2415            ///     let mut interval = metrics_monitor.intervals();
2416            ///     let mut next_interval = || interval.next().unwrap();
2417            ///
2418            ///     // no tasks have yet been created, instrumented, or polled
2419            ///     assert_eq!(next_interval().mean_first_poll_delay(), Duration::ZERO);
2420            ///
2421            ///     // constructs and instruments a task, pauses for `pause_time`, awaits the task, then
2422            ///     // produces the total time it took to do all of the aforementioned
2423            ///     async fn instrument_pause_await(
2424            ///         metrics_monitor: &tokio_metrics::TaskMonitor,
2425            ///         pause_time: Duration
2426            ///     ) -> Duration
2427            ///     {
2428            ///         let before_instrumentation = tokio::time::Instant::now();
2429            ///         let task = metrics_monitor.instrument(async move {});
2430            ///         tokio::time::sleep(pause_time).await;
2431            ///         task.await;
2432            ///         before_instrumentation.elapsed()
2433            ///     }
2434            ///
2435            ///     // construct and await a task that pauses for 500ms between instrumentation and first poll
2436            ///     let task_a_pause_time = Duration::from_millis(500);
2437            ///     let task_a_total_time = instrument_pause_await(&metrics_monitor, task_a_pause_time).await;
2438            ///
2439            ///     // the `mean_first_poll_delay` will be some duration greater-than-or-equal-to the
2440            ///     // pause time of 500ms, and less-than-or-equal-to the total runtime of `task_a`
2441            ///     let mean_first_poll_delay = next_interval().mean_first_poll_delay();
2442            ///     assert!(mean_first_poll_delay >= task_a_pause_time);
2443            ///     assert!(mean_first_poll_delay <= task_a_total_time);
2444            ///
2445            ///     // construct and await a task that pauses for 500ms between instrumentation and first poll
2446            ///     let task_b_pause_time = Duration::from_millis(500);
2447            ///     let task_b_total_time = instrument_pause_await(&metrics_monitor, task_b_pause_time).await;
2448            ///
2449            ///     // construct and await a task that pauses for 1000ms between instrumentation and first poll
2450            ///     let task_c_pause_time = Duration::from_millis(1000);
2451            ///     let task_c_total_time = instrument_pause_await(&metrics_monitor, task_c_pause_time).await;
2452            ///
2453            ///     // the `mean_first_poll_delay` will be some duration greater-than-or-equal-to the
2454            ///     // average pause time of 500ms, and less-than-or-equal-to the combined total runtime of
2455            ///     // `task_b` and `task_c`
2456            ///     let mean_first_poll_delay = next_interval().mean_first_poll_delay();
2457            ///     assert!(mean_first_poll_delay >= (task_b_pause_time + task_c_pause_time) / 2);
2458            ///     assert!(mean_first_poll_delay <= (task_b_total_time + task_c_total_time) / 2);
2459            /// }
2460            /// ```
2461            pub fn mean_first_poll_delay(&self) -> Duration {
2462                mean(self.total_first_poll_delay, self.first_poll_count)
2463            }
2464
2465            /// The mean duration of idles.
2466            ///
2467            /// ##### Definition
2468            /// This metric is derived from [`total_idle_duration`][TaskMetrics::total_idle_duration] ÷
2469            /// [`total_idled_count`][TaskMetrics::total_idled_count].
2470            ///
2471            /// ##### Interpretation
2472            /// The idle state is the duration spanning the instant a task completes a poll, and the instant
2473            /// that it is next awoken. Tasks inhabit this state when they are waiting for task-external
2474            /// events to complete (e.g., an asynchronous sleep, a network request, file I/O, etc.). If this
2475            /// metric increases, it means that tasks, in aggregate, spent more time waiting for
2476            /// task-external events to complete.
2477            ///
2478            /// ##### Examples
2479            /// ```
2480            /// #[tokio::main]
2481            /// async fn main() {
2482            ///     let monitor = tokio_metrics::TaskMonitor::new();
2483            ///     let one_sec = std::time::Duration::from_secs(1);
2484            ///
2485            ///     monitor.instrument(async move {
2486            ///         tokio::time::sleep(one_sec).await;
2487            ///     }).await;
2488            ///
2489            ///     assert!(monitor.cumulative().mean_idle_duration() >= one_sec);
2490            /// }
2491            /// ```
2492            pub fn mean_idle_duration(&self) -> Duration {
2493                mean(self.total_idle_duration, self.total_idled_count)
2494            }
2495
2496            /// The mean duration that tasks spent waiting to be executed after awakening.
2497            ///
2498            /// ##### Definition
2499            /// This metric is derived from
2500            /// [`total_scheduled_duration`][TaskMetrics::total_scheduled_duration] ÷
2501            /// [`total_scheduled_count`][`TaskMetrics::total_scheduled_count`].
2502            ///
2503            /// ##### Interpretation
2504            /// If this metric increases, it means that, on average, tasks spent longer in the runtime's
2505            /// queues before being polled.
2506            ///
2507            /// ##### See also
2508            /// - **[`mean_first_poll_delay`][TaskMetrics::mean_first_poll_delay]**
2509            ///   The mean duration elapsed between the instant tasks are instrumented, and the instant they
2510            ///   are first polled.
2511            ///
2512            /// ##### Examples
2513            /// ```
2514            /// use tokio::time::Duration;
2515            ///
2516            /// #[tokio::main(flavor = "current_thread")]
2517            /// async fn main() {
2518            ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
2519            ///     let mut interval = metrics_monitor.intervals();
2520            ///     let mut next_interval = || interval.next().unwrap();
2521            ///
2522            ///     // construct and instrument and spawn a task that yields endlessly
2523            ///     tokio::spawn(metrics_monitor.instrument(async {
2524            ///         loop { tokio::task::yield_now().await }
2525            ///     }));
2526            ///
2527            ///     tokio::task::yield_now().await;
2528            ///
2529            ///     // block the executor for 1 second
2530            ///     std::thread::sleep(Duration::from_millis(1000));
2531            ///
2532            ///     // get the task to run twice
2533            ///     // the first will have a 1 sec scheduling delay, the second will have almost none
2534            ///     tokio::task::yield_now().await;
2535            ///     tokio::task::yield_now().await;
2536            ///
2537            ///     // `endless_task` will have spent approximately one second waiting
2538            ///     let mean_scheduled_duration = next_interval().mean_scheduled_duration();
2539            ///     assert!(mean_scheduled_duration >= Duration::from_millis(500), "{}", mean_scheduled_duration.as_secs_f64());
2540            ///     assert!(mean_scheduled_duration <= Duration::from_millis(600), "{}", mean_scheduled_duration.as_secs_f64());
2541            /// }
2542            /// ```
2543            pub fn mean_scheduled_duration(&self) -> Duration {
2544                mean(self.total_scheduled_duration, self.total_scheduled_count)
2545            }
2546
2547            /// The mean duration of polls.
2548            ///
2549            /// ##### Definition
2550            /// This metric is derived from [`total_poll_duration`][TaskMetrics::total_poll_duration] ÷
2551            /// [`total_poll_count`][TaskMetrics::total_poll_count].
2552            ///
2553            /// ##### Interpretation
2554            /// If this metric increases, it means that, on average, individual polls are tending to take
2555            /// longer. However, this does not necessarily imply increased task latency: An increase in poll
2556            /// durations could be offset by fewer polls.
2557            ///
2558            /// ##### See also
2559            /// - **[`slow_poll_ratio`][TaskMetrics::slow_poll_ratio]**
2560            ///   The ratio between the number polls categorized as slow and fast.
2561            /// - **[`mean_slow_poll_duration`][TaskMetrics::mean_slow_poll_duration]**
2562            ///   The mean duration of slow polls.
2563            ///
2564            /// ##### Examples
2565            /// ```
2566            /// use std::time::Duration;
2567            ///
2568            /// #[tokio::main(flavor = "current_thread", start_paused = true)]
2569            /// async fn main() {
2570            ///     let monitor = tokio_metrics::TaskMonitor::new();
2571            ///     let mut interval = monitor.intervals();
2572            ///     let mut next_interval = move || interval.next().unwrap();
2573            ///
2574            ///     assert_eq!(next_interval().mean_poll_duration(), Duration::ZERO);
2575            ///
2576            ///     monitor.instrument(async {
2577            ///         tokio::time::advance(Duration::from_secs(1)).await; // poll 1 (1s)
2578            ///         tokio::time::advance(Duration::from_secs(1)).await; // poll 2 (1s)
2579            ///         ()                                                  // poll 3 (0s)
2580            ///     }).await;
2581            ///
2582            ///     assert_eq!(next_interval().mean_poll_duration(), Duration::from_secs(2) / 3);
2583            /// }
2584            /// ```
2585            pub fn mean_poll_duration(&self) -> Duration {
2586                mean(self.total_poll_duration, self.total_poll_count)
2587            }
2588
2589            /// The ratio between the number polls categorized as slow and fast.
2590            ///
2591            /// ##### Definition
2592            /// This metric is derived from [`total_slow_poll_count`][TaskMetrics::total_slow_poll_count] ÷
2593            /// [`total_poll_count`][TaskMetrics::total_poll_count].
2594            ///
2595            /// ##### Interpretation
2596            /// If this metric increases, it means that a greater proportion of polls took excessively long
2597            /// before yielding to the scheduler. This does not necessarily imply increased task latency:
2598            /// An increase in the proportion of slow polls could be offset by fewer or faster polls.
2599            /// However, as a rule, *should* yield to the scheduler frequently.
2600            ///
2601            /// ##### See also
2602            /// - **[`mean_poll_duration`][TaskMetrics::mean_poll_duration]**
2603            ///   The mean duration of polls.
2604            /// - **[`mean_slow_poll_duration`][TaskMetrics::mean_slow_poll_duration]**
2605            ///   The mean duration of slow polls.
2606            ///
2607            /// ##### Examples
2608            /// Changes in this metric may be observed by varying the ratio of slow and slow fast within
2609            /// sampling intervals; for instance:
2610            /// ```
2611            /// use std::future::Future;
2612            /// use std::time::Duration;
2613            ///
2614            /// #[tokio::main]
2615            /// async fn main() {
2616            ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
2617            ///     let mut interval = metrics_monitor.intervals();
2618            ///     let mut next_interval = || interval.next().unwrap();
2619            ///
2620            ///     // no tasks have been constructed, instrumented, or polled
2621            ///     let interval = next_interval();
2622            ///     assert_eq!(interval.total_fast_poll_count, 0);
2623            ///     assert_eq!(interval.total_slow_poll_count, 0);
2624            ///     assert!(interval.slow_poll_ratio().is_nan());
2625            ///
2626            ///     let fast = Duration::ZERO;
2627            ///     let slow = 10 * metrics_monitor.slow_poll_threshold();
2628            ///
2629            ///     // this task completes in three fast polls
2630            ///     metrics_monitor.instrument(async {
2631            ///         spin_for(fast).await;   // fast poll 1
2632            ///         spin_for(fast).await;   // fast poll 2
2633            ///         spin_for(fast);         // fast poll 3
2634            ///     }).await;
2635            ///
2636            ///     // this task completes in two slow polls
2637            ///     metrics_monitor.instrument(async {
2638            ///         spin_for(slow).await;   // slow poll 1
2639            ///         spin_for(slow);         // slow poll 2
2640            ///     }).await;
2641            ///
2642            ///     let interval = next_interval();
2643            ///     assert_eq!(interval.total_fast_poll_count, 3);
2644            ///     assert_eq!(interval.total_slow_poll_count, 2);
2645            ///     assert_eq!(interval.slow_poll_ratio(), ratio(2., 3.));
2646            ///
2647            ///     // this task completes in three slow polls
2648            ///     metrics_monitor.instrument(async {
2649            ///         spin_for(slow).await;   // slow poll 1
2650            ///         spin_for(slow).await;   // slow poll 2
2651            ///         spin_for(slow);         // slow poll 3
2652            ///     }).await;
2653            ///
2654            ///     // this task completes in two fast polls
2655            ///     metrics_monitor.instrument(async {
2656            ///         spin_for(fast).await; // fast poll 1
2657            ///         spin_for(fast);       // fast poll 2
2658            ///     }).await;
2659            ///
2660            ///     let interval = next_interval();
2661            ///     assert_eq!(interval.total_fast_poll_count, 2);
2662            ///     assert_eq!(interval.total_slow_poll_count, 3);
2663            ///     assert_eq!(interval.slow_poll_ratio(), ratio(3., 2.));
2664            /// }
2665            ///
2666            /// fn ratio(a: f64, b: f64) -> f64 {
2667            ///     a / (a + b)
2668            /// }
2669            ///
2670            /// /// Block the current thread for a given `duration`, then (optionally) yield to the scheduler.
2671            /// fn spin_for(duration: Duration) -> impl Future<Output=()> {
2672            ///     let start = tokio::time::Instant::now();
2673            ///     while start.elapsed() <= duration {}
2674            ///     tokio::task::yield_now()
2675            /// }
2676            /// ```
2677            pub fn slow_poll_ratio(&self) -> f64 {
2678                self.total_slow_poll_count as f64 / self.total_poll_count as f64
2679            }
2680
2681            /// The ratio of tasks exceeding [`long_delay_threshold`][TaskMonitor::long_delay_threshold].
2682            ///
2683            /// ##### Definition
2684            /// This metric is derived from [`total_long_delay_count`][TaskMetrics::total_long_delay_count] ÷
2685            /// [`total_scheduled_count`][TaskMetrics::total_scheduled_count].
2686            pub fn long_delay_ratio(&self) -> f64 {
2687                self.total_long_delay_count as f64 / self.total_scheduled_count as f64
2688            }
2689
2690            /// The mean duration of fast polls.
2691            ///
2692            /// ##### Definition
2693            /// This metric is derived from
2694            /// [`total_fast_poll_duration`][TaskMetrics::total_fast_poll_duration] ÷
2695            /// [`total_fast_poll_count`][TaskMetrics::total_fast_poll_count].
2696            ///
2697            /// ##### Examples
2698            /// In the below example, no tasks are polled in the first sampling interval; three fast polls
2699            /// consume a mean of
2700            /// ⅜ × [`DEFAULT_SLOW_POLL_THRESHOLD`][TaskMonitor::DEFAULT_SLOW_POLL_THRESHOLD] time in the
2701            /// second sampling interval; and two fast polls consume a total of
2702            /// ½ × [`DEFAULT_SLOW_POLL_THRESHOLD`][TaskMonitor::DEFAULT_SLOW_POLL_THRESHOLD] time in the
2703            /// third sampling interval:
2704            /// ```
2705            /// use std::future::Future;
2706            /// use std::time::Duration;
2707            ///
2708            /// #[tokio::main]
2709            /// async fn main() {
2710            ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
2711            ///     let mut interval = metrics_monitor.intervals();
2712            ///     let mut next_interval = || interval.next().unwrap();
2713            ///
2714            ///     // no tasks have been constructed, instrumented, or polled
2715            ///     assert_eq!(next_interval().mean_fast_poll_duration(), Duration::ZERO);
2716            ///
2717            ///     let threshold = metrics_monitor.slow_poll_threshold();
2718            ///     let fast_1 = 1 * Duration::from_micros(1);
2719            ///     let fast_2 = 2 * Duration::from_micros(1);
2720            ///     let fast_3 = 3 * Duration::from_micros(1);
2721            ///
2722            ///     // this task completes in two fast polls
2723            ///     let total_time = time(metrics_monitor.instrument(async {
2724            ///         spin_for(fast_1).await; // fast poll 1
2725            ///         spin_for(fast_2)        // fast poll 2
2726            ///     })).await;
2727            ///
2728            ///     // `mean_fast_poll_duration` ≈ the mean of `fast_1` and `fast_2`
2729            ///     let mean_fast_poll_duration = next_interval().mean_fast_poll_duration();
2730            ///     assert!(mean_fast_poll_duration >= (fast_1 + fast_2) / 2);
2731            ///     assert!(mean_fast_poll_duration <= total_time / 2);
2732            ///
2733            ///     // this task completes in three fast polls
2734            ///     let total_time = time(metrics_monitor.instrument(async {
2735            ///         spin_for(fast_1).await; // fast poll 1
2736            ///         spin_for(fast_2).await; // fast poll 2
2737            ///         spin_for(fast_3)        // fast poll 3
2738            ///     })).await;
2739            ///
2740            ///     // `mean_fast_poll_duration` ≈ the mean of `fast_1`, `fast_2`, `fast_3`
2741            ///     let mean_fast_poll_duration = next_interval().mean_fast_poll_duration();
2742            ///     assert!(mean_fast_poll_duration >= (fast_1 + fast_2 + fast_3) / 3);
2743            ///     assert!(mean_fast_poll_duration <= total_time / 3);
2744            /// }
2745            ///
2746            /// /// Produces the amount of time it took to await a given task.
2747            /// async fn time(task: impl Future) -> Duration {
2748            ///     let start = tokio::time::Instant::now();
2749            ///     task.await;
2750            ///     start.elapsed()
2751            /// }
2752            ///
2753            /// /// Block the current thread for a given `duration`, then (optionally) yield to the scheduler.
2754            /// fn spin_for(duration: Duration) -> impl Future<Output=()> {
2755            ///     let start = tokio::time::Instant::now();
2756            ///     while start.elapsed() <= duration {}
2757            ///     tokio::task::yield_now()
2758            /// }
2759            /// ```
2760            pub fn mean_fast_poll_duration(&self) -> Duration {
2761                mean(self.total_fast_poll_duration, self.total_fast_poll_count)
2762            }
2763
2764            /// The mean duration of slow polls.
2765            ///
2766            /// ##### Definition
2767            /// This metric is derived from
2768            /// [`total_slow_poll_duration`][TaskMetrics::total_slow_poll_duration] ÷
2769            /// [`total_slow_poll_count`][TaskMetrics::total_slow_poll_count].
2770            ///
2771            /// ##### Interpretation
2772            /// If this metric increases, it means that a greater proportion of polls took excessively long
2773            /// before yielding to the scheduler. This does not necessarily imply increased task latency:
2774            /// An increase in the proportion of slow polls could be offset by fewer or faster polls.
2775            ///
2776            /// ##### See also
2777            /// - **[`mean_poll_duration`][TaskMetrics::mean_poll_duration]**
2778            ///   The mean duration of polls.
2779            /// - **[`slow_poll_ratio`][TaskMetrics::slow_poll_ratio]**
2780            ///   The ratio between the number polls categorized as slow and fast.
2781            ///
2782            /// ##### Interpretation
2783            /// If this metric increases, it means that, on average, slow polls got even slower. This does
2784            /// necessarily imply increased task latency: An increase in average slow poll duration could be
2785            /// offset by fewer or faster polls. However, as a rule, *should* yield to the scheduler
2786            /// frequently.
2787            ///
2788            /// ##### Examples
2789            /// In the below example, no tasks are polled in the first sampling interval; three slow polls
2790            /// consume a mean of
2791            /// 1.5 × [`DEFAULT_SLOW_POLL_THRESHOLD`][TaskMonitor::DEFAULT_SLOW_POLL_THRESHOLD] time in the
2792            /// second sampling interval; and two slow polls consume a total of
2793            /// 2 × [`DEFAULT_SLOW_POLL_THRESHOLD`][TaskMonitor::DEFAULT_SLOW_POLL_THRESHOLD] time in the
2794            /// third sampling interval:
2795            /// ```
2796            /// use std::future::Future;
2797            /// use std::time::Duration;
2798            ///
2799            /// #[tokio::main]
2800            /// async fn main() {
2801            ///     let metrics_monitor = tokio_metrics::TaskMonitor::new();
2802            ///     let mut interval = metrics_monitor.intervals();
2803            ///     let mut next_interval = || interval.next().unwrap();
2804            ///
2805            ///     // no tasks have been constructed, instrumented, or polled
2806            ///     assert_eq!(next_interval().mean_slow_poll_duration(), Duration::ZERO);
2807            ///
2808            ///     let threshold = metrics_monitor.slow_poll_threshold();
2809            ///     let slow_1 = 1 * threshold;
2810            ///     let slow_2 = 2 * threshold;
2811            ///     let slow_3 = 3 * threshold;
2812            ///
2813            ///     // this task completes in two slow polls
2814            ///     let total_time = time(metrics_monitor.instrument(async {
2815            ///         spin_for(slow_1).await; // slow poll 1
2816            ///         spin_for(slow_2)        // slow poll 2
2817            ///     })).await;
2818            ///
2819            ///     // `mean_slow_poll_duration` ≈ the mean of `slow_1` and `slow_2`
2820            ///     let mean_slow_poll_duration = next_interval().mean_slow_poll_duration();
2821            ///     assert!(mean_slow_poll_duration >= (slow_1 + slow_2) / 2);
2822            ///     assert!(mean_slow_poll_duration <= total_time / 2);
2823            ///
2824            ///     // this task completes in three slow polls
2825            ///     let total_time = time(metrics_monitor.instrument(async {
2826            ///         spin_for(slow_1).await; // slow poll 1
2827            ///         spin_for(slow_2).await; // slow poll 2
2828            ///         spin_for(slow_3)        // slow poll 3
2829            ///     })).await;
2830            ///
2831            ///     // `mean_slow_poll_duration` ≈ the mean of `slow_1`, `slow_2`, `slow_3`
2832            ///     let mean_slow_poll_duration = next_interval().mean_slow_poll_duration();
2833            ///     assert!(mean_slow_poll_duration >= (slow_1 + slow_2 + slow_3) / 3);
2834            ///     assert!(mean_slow_poll_duration <= total_time / 3);
2835            /// }
2836            ///
2837            /// /// Produces the amount of time it took to await a given task.
2838            /// async fn time(task: impl Future) -> Duration {
2839            ///     let start = tokio::time::Instant::now();
2840            ///     task.await;
2841            ///     start.elapsed()
2842            /// }
2843            ///
2844            /// /// Block the current thread for a given `duration`, then (optionally) yield to the scheduler.
2845            /// fn spin_for(duration: Duration) -> impl Future<Output=()> {
2846            ///     let start = tokio::time::Instant::now();
2847            ///     while start.elapsed() <= duration {}
2848            ///     tokio::task::yield_now()
2849            /// }
2850            /// ```
2851            pub fn mean_slow_poll_duration(&self) -> Duration {
2852                mean(self.total_slow_poll_duration, self.total_slow_poll_count)
2853            }
2854
2855            /// The average time taken for a task with a short scheduling delay to be executed after being
2856            /// scheduled.
2857            ///
2858            /// ##### Definition
2859            /// This metric is derived from
2860            /// [`total_short_delay_duration`][TaskMetrics::total_short_delay_duration] ÷
2861            /// [`total_short_delay_count`][TaskMetrics::total_short_delay_count].
2862            pub fn mean_short_delay_duration(&self) -> Duration {
2863                mean(
2864                    self.total_short_delay_duration,
2865                    self.total_short_delay_count,
2866                )
2867            }
2868
2869            /// The average scheduling delay for a task which takes a long time to start executing after
2870            /// being scheduled.
2871            ///
2872            /// ##### Definition
2873            /// This metric is derived from
2874            /// [`total_long_delay_duration`][TaskMetrics::total_long_delay_duration] ÷
2875            /// [`total_long_delay_count`][TaskMetrics::total_long_delay_count].
2876            pub fn mean_long_delay_duration(&self) -> Duration {
2877                mean(self.total_long_delay_duration, self.total_long_delay_count)
2878            }
2879        }
2880        unstable {}
2881    }
2882);
2883
2884impl<T: Future, M: AsRef<TaskMonitorCore> + Send + Sync + 'static> Future for Instrumented<T, M> {
2885    type Output = T::Output;
2886
2887    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
2888        instrument_poll(cx, self, Future::poll)
2889    }
2890}
2891
2892impl<T: Stream> Stream for Instrumented<T> {
2893    type Item = T::Item;
2894
2895    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
2896        instrument_poll(cx, self, Stream::poll_next)
2897    }
2898}
2899
2900fn instrument_poll<T, M: AsRef<TaskMonitorCore> + Send + Sync + 'static, Out>(
2901    cx: &mut Context<'_>,
2902    instrumented: Pin<&mut Instrumented<T, M>>,
2903    poll_fn: impl FnOnce(Pin<&mut T>, &mut Context<'_>) -> Poll<Out>,
2904) -> Poll<Out> {
2905    let poll_start = Instant::now();
2906    let this = instrumented.project();
2907    let idled_at = this.idled_at;
2908    let state = this.state;
2909    let instrumented_at = state.instrumented_at;
2910    let metrics = &state.monitor.as_ref().metrics;
2911    /* accounting for time-to-first-poll and tasks-count */
2912    // is this the first time this task has been polled?
2913    if !*this.did_poll_once {
2914        // if so, we need to do three things:
2915        /* 1. note that this task *has* been polled */
2916        *this.did_poll_once = true;
2917
2918        /* 2. account for the time-to-first-poll of this task */
2919        // if the time-to-first-poll of this task exceeds `u64::MAX` ns,
2920        // round down to `u64::MAX` nanoseconds
2921        let elapsed = poll_start
2922            .saturating_duration_since(instrumented_at)
2923            .as_nanos()
2924            .try_into()
2925            .unwrap_or(u64::MAX);
2926        // add this duration to `time_to_first_poll_ns_total`
2927        metrics.total_first_poll_delay_ns.fetch_add(elapsed, SeqCst);
2928
2929        /* 3. increment the count of tasks that have been polled at least once */
2930        metrics.first_poll_count.fetch_add(1, SeqCst);
2931    }
2932    /* accounting for time-idled and time-scheduled */
2933    // 1. note (and reset) the instant this task was last awoke
2934    let woke_at = state.woke_at.swap(0, SeqCst);
2935    // The state of a future is *idling* in the interim between the instant
2936    // it completes a `poll`, and the instant it is next awoken.
2937    if *idled_at < woke_at {
2938        // increment the counter of how many idles occurred
2939        metrics.total_idled_count.fetch_add(1, SeqCst);
2940
2941        // compute the duration of the idle
2942        let idle_ns = woke_at.saturating_sub(*idled_at);
2943
2944        // update the max time tasks spent idling, both locally and
2945        // globally.
2946        metrics
2947            .local_max_idle_duration_ns
2948            .fetch_max(idle_ns, SeqCst);
2949        metrics
2950            .global_max_idle_duration_ns
2951            .fetch_max(idle_ns, SeqCst);
2952        // adjust the total elapsed time monitored tasks spent idling
2953        metrics.total_idle_duration_ns.fetch_add(idle_ns, SeqCst);
2954    }
2955    // if this task spent any time in the scheduled state after instrumentation,
2956    // and after first poll, `woke_at` will be greater than 0.
2957    if woke_at > 0 {
2958        // increment the counter of how many schedules occurred
2959        metrics.total_scheduled_count.fetch_add(1, SeqCst);
2960
2961        // recall that the `woke_at` field is internally represented as
2962        // nanoseconds-since-instrumentation. here, for accounting purposes,
2963        // we need to instead represent it as a proper `Instant`.
2964        let woke_instant = instrumented_at
2965            .checked_add(Duration::from_nanos(woke_at))
2966            .unwrap_or(poll_start);
2967
2968        // the duration this task spent scheduled is time time elapsed between
2969        // when this task was awoke, and when it was polled.
2970        let scheduled_ns = poll_start
2971            .saturating_duration_since(woke_instant)
2972            .as_nanos()
2973            .try_into()
2974            .unwrap_or(u64::MAX);
2975
2976        let scheduled = Duration::from_nanos(scheduled_ns);
2977
2978        let (count_bucket, duration_bucket) = // was the scheduling delay long or short?
2979            if scheduled >= metrics.long_delay_threshold {
2980                (&metrics.total_long_delay_count, &metrics.total_long_delay_duration_ns)
2981            } else {
2982                (&metrics.total_short_delay_count, &metrics.total_short_delay_duration_ns)
2983            };
2984        // update the appropriate bucket
2985        count_bucket.fetch_add(1, SeqCst);
2986        duration_bucket.fetch_add(scheduled_ns, SeqCst);
2987
2988        // add `scheduled_ns` to the Monitor's total
2989        metrics
2990            .total_scheduled_duration_ns
2991            .fetch_add(scheduled_ns, SeqCst);
2992
2993        // mirror the same scheduling accounting into this task's own log so that
2994        // futures running within the task can attribute scheduling delay to
2995        // their own lifetime (see `FutureMonitor`). Only present when the
2996        // monitor opted in, so the common case skips this entirely.
2997        if let Some(log) = &state.log {
2998            log.scheduled_count.fetch_add(1, SeqCst);
2999            log.scheduled_duration_ns.fetch_add(scheduled_ns, SeqCst);
3000            if scheduled >= metrics.long_delay_threshold {
3001                log.long_delay_count.fetch_add(1, SeqCst);
3002            }
3003        }
3004    }
3005    // Register the waker
3006    state.waker.register(cx.waker());
3007    // Get the instrumented waker
3008    let waker_ref = futures_util::task::waker_ref(state);
3009    let mut cx = Context::from_waker(&waker_ref);
3010    // Poll the task, publishing this task's scheduling log as a task-local for
3011    // the duration of the poll so nested futures can read it.
3012    let inner_poll_start = Instant::now();
3013    // Only publish the task-local when the monitor opted in. When it didn't
3014    // (the common case) this is a single predictable branch and the poll path is
3015    // byte-for-byte the original — no guard, thread-local, or `Arc` clone.
3016    let ret = if let Some(log) = &state.log {
3017        let _sched_guard = SchedulingLogGuard::enter(log.clone());
3018        poll_fn(this.task, &mut cx)
3019    } else {
3020        poll_fn(this.task, &mut cx)
3021    };
3022    let inner_poll_end = Instant::now();
3023    /* idle time starts now */
3024    *idled_at = inner_poll_end
3025        .saturating_duration_since(instrumented_at)
3026        .as_nanos()
3027        .try_into()
3028        .unwrap_or(u64::MAX);
3029    /* accounting for poll time */
3030    let inner_poll_duration = inner_poll_end.saturating_duration_since(inner_poll_start);
3031    let inner_poll_ns: u64 = inner_poll_duration
3032        .as_nanos()
3033        .try_into()
3034        .unwrap_or(u64::MAX);
3035    let (count_bucket, duration_bucket) = // was this a slow or fast poll?
3036            if inner_poll_duration >= metrics.slow_poll_threshold {
3037                (&metrics.total_slow_poll_count, &metrics.total_slow_poll_duration)
3038            } else {
3039                (&metrics.total_fast_poll_count, &metrics.total_fast_poll_duration_ns)
3040            };
3041    // update the appropriate bucket
3042    count_bucket.fetch_add(1, SeqCst);
3043    duration_bucket.fetch_add(inner_poll_ns, SeqCst);
3044    ret
3045}
3046
3047impl<M> State<M> {
3048    fn on_wake(&self) {
3049        let woke_at: u64 = match self.instrumented_at.elapsed().as_nanos().try_into() {
3050            Ok(woke_at) => woke_at,
3051            // This is highly unlikely as it would mean the task ran for over
3052            // 500 years. If you ran your service for 500 years. If you are
3053            // reading this 500 years in the future, I'm sorry.
3054            Err(_) => return,
3055        };
3056
3057        // We don't actually care about the result
3058        let _ = self.woke_at.compare_exchange(0, woke_at, SeqCst, SeqCst);
3059    }
3060}
3061
3062impl<M: Send + Sync> ArcWake for State<M> {
3063    fn wake_by_ref(arc_self: &Arc<State<M>>) {
3064        arc_self.on_wake();
3065        arc_self.waker.wake();
3066    }
3067
3068    fn wake(self: Arc<State<M>>) {
3069        self.on_wake();
3070        self.waker.wake();
3071    }
3072}
3073
3074/// Key metrics of a single instrumented future running within a larger task, as
3075/// captured by a [`FutureMonitor`].
3076///
3077/// Unlike [`TaskMetrics`], which aggregates across every future instrumented by
3078/// a [`TaskMonitor`], these metrics describe just one future. Idle, poll, and
3079/// first-poll metrics are measured locally from that future's own polls, so they
3080/// remain accurate even when the surrounding task interleaves other work.
3081/// Scheduling metrics are sampled from the root task's [`TaskScheduling`] log
3082/// over the future's lifetime (see [`FutureMonitor`]).
3083#[non_exhaustive]
3084#[cfg_attr(
3085    feature = "metrique-integration",
3086    metrique::unit_of_work::metrics(subfield)
3087)]
3088#[derive(Debug, Clone, Default)]
3089pub struct FutureMetrics {
3090    /// The number of times the future was polled.
3091    pub poll_count: u64,
3092    /// The total time spent polling the future.
3093    pub total_poll_duration: Duration,
3094    /// The number of polls that exceeded the monitor's slow-poll threshold.
3095    pub slow_poll_count: u64,
3096    /// The number of times the future idled, waiting to be awoken.
3097    pub idle_count: u64,
3098    /// The total time the future spent idle, waiting on external events.
3099    pub total_idle_duration: Duration,
3100    /// The longest single idle the future experienced.
3101    pub max_idle_duration: Duration,
3102    /// The delay between the future being instrumented and its first poll.
3103    pub first_poll_delay: Duration,
3104    /// The wall-clock time from the future's first poll to its completion.
3105    pub total_duration: Duration,
3106    /// The number of times the underlying task was scheduled while this future
3107    /// was active.
3108    ///
3109    /// This tracks the *root* task's scheduling, not this future's polls, so it
3110    /// can exceed [`poll_count`](Self::poll_count) — for example when the future
3111    /// is one branch of a `select!` and the task is scheduled to advance the
3112    /// other branches.
3113    pub scheduled_count: u64,
3114    /// The total scheduling delay the underlying task incurred while this future
3115    /// was active.
3116    pub total_scheduled_duration: Duration,
3117    /// The number of those scheduling delays that crossed the long-delay threshold.
3118    pub long_delay_count: u64,
3119}
3120
3121/// Per-future capture state shared between a [`FutureMonitor`] and the
3122/// [`MonitoredFuture`] future it produces.
3123#[derive(Debug, Default)]
3124struct SpanState {
3125    started: AtomicBool,
3126    start_scheduled_count: AtomicU64,
3127    start_scheduled_duration_ns: AtomicU64,
3128    start_long_delay_count: AtomicU64,
3129    end_scheduled_count: AtomicU64,
3130    end_scheduled_duration_ns: AtomicU64,
3131    end_long_delay_count: AtomicU64,
3132    first_poll: OnceLock<Instant>,
3133    total_duration_ns: AtomicU64,
3134}
3135
3136fn duration_to_nanos(d: Duration) -> u64 {
3137    d.as_nanos().try_into().unwrap_or(u64::MAX)
3138}
3139
3140impl SpanState {
3141    /// Records the root scheduling snapshot for a poll. Called at the top of
3142    /// every poll of the [`MonitoredFuture`], while the root task's
3143    /// scheduling log is in scope.
3144    fn record_poll_start(&self, now: TaskScheduling) {
3145        if !self.started.swap(true, SeqCst) {
3146            self.start_scheduled_count
3147                .store(now.scheduled_count, SeqCst);
3148            self.start_scheduled_duration_ns
3149                .store(duration_to_nanos(now.total_scheduled_duration), SeqCst);
3150            self.start_long_delay_count
3151                .store(now.long_delay_count, SeqCst);
3152            let _ = self.first_poll.set(Instant::now());
3153        }
3154        self.end_scheduled_count.store(now.scheduled_count, SeqCst);
3155        self.end_scheduled_duration_ns
3156            .store(duration_to_nanos(now.total_scheduled_duration), SeqCst);
3157        self.end_long_delay_count
3158            .store(now.long_delay_count, SeqCst);
3159    }
3160
3161    /// Records total elapsed time *after* a poll completes, so the execution
3162    /// time of the poll just finished — including the final one — is counted.
3163    fn record_poll_end(&self) {
3164        if let Some(first_poll) = self.first_poll.get() {
3165            self.total_duration_ns
3166                .store(duration_to_nanos(first_poll.elapsed()), SeqCst);
3167        }
3168    }
3169
3170    fn scheduling_delta(&self) -> TaskScheduling {
3171        let end = TaskScheduling {
3172            scheduled_count: self.end_scheduled_count.load(SeqCst),
3173            total_scheduled_duration: Duration::from_nanos(
3174                self.end_scheduled_duration_ns.load(SeqCst),
3175            ),
3176            long_delay_count: self.end_long_delay_count.load(SeqCst),
3177        };
3178        let start = TaskScheduling {
3179            scheduled_count: self.start_scheduled_count.load(SeqCst),
3180            total_scheduled_duration: Duration::from_nanos(
3181                self.start_scheduled_duration_ns.load(SeqCst),
3182            ),
3183            long_delay_count: self.start_long_delay_count.load(SeqCst),
3184        };
3185        end.since(start)
3186    }
3187
3188    fn total_duration(&self) -> Duration {
3189        Duration::from_nanos(self.total_duration_ns.load(SeqCst))
3190    }
3191}
3192
3193fn build_future_metrics(monitor: &TaskMonitor, span: &SpanState) -> FutureMetrics {
3194    let local = monitor.cumulative();
3195    let scheduling = span.scheduling_delta();
3196    FutureMetrics {
3197        poll_count: local.total_poll_count,
3198        total_poll_duration: local.total_poll_duration,
3199        slow_poll_count: local.total_slow_poll_count,
3200        idle_count: local.total_idled_count,
3201        total_idle_duration: local.total_idle_duration,
3202        max_idle_duration: local.max_idle_duration,
3203        first_poll_delay: local.total_first_poll_delay,
3204        total_duration: span.total_duration(),
3205        scheduled_count: scheduling.scheduled_count,
3206        total_scheduled_duration: scheduling.total_scheduled_duration,
3207        long_delay_count: scheduling.long_delay_count,
3208    }
3209}
3210
3211pin_project! {
3212    /// The future returned by [`FutureMonitor::instrument`].
3213    ///
3214    /// Wraps the future, capturing its per-poll metrics locally and
3215    /// sampling the surrounding task's scheduling delay at each poll. Resolves to
3216    /// the wrapped future's output paired with the captured [`FutureMetrics`].
3217    pub struct MonitoredFuture<F> {
3218        #[pin]
3219        inner: Instrumented<F, TaskMonitor>,
3220        monitor: TaskMonitor,
3221        span: Arc<SpanState>,
3222    }
3223}
3224
3225impl<F> std::fmt::Debug for MonitoredFuture<F> {
3226    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
3227        f.debug_struct("MonitoredFuture").finish_non_exhaustive()
3228    }
3229}
3230
3231impl<F: Future> Future for MonitoredFuture<F> {
3232    type Output = (F::Output, FutureMetrics);
3233
3234    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
3235        let this = self.project();
3236        // Sample the root task's scheduling log *before* delegating to the inner
3237        // `Instrumented`, while the root task's log is the one in scope (the
3238        // inner `Instrumented` shadows it with the future's own log during its
3239        // poll, which we deliberately ignore).
3240        this.span
3241            .record_poll_start(TaskScheduling::try_current().unwrap_or_default());
3242        let ret = this.inner.poll(cx);
3243        // Stamp total elapsed *after* the poll so the execution time of the poll
3244        // just finished — including the final one that returns `Ready` — is
3245        // captured in `total_duration`.
3246        this.span.record_poll_end();
3247        match ret {
3248            Poll::Ready(output) => {
3249                let metrics = build_future_metrics(this.monitor, this.span);
3250                Poll::Ready((output, metrics))
3251            }
3252            Poll::Pending => Poll::Pending,
3253        }
3254    }
3255}
3256
3257/// Monitors the metrics of a single future running within a larger,
3258/// already-[instrumented][TaskMonitor::instrument] task.
3259///
3260/// [`TaskMonitor`] aggregates metrics across all the futures it instruments;
3261/// `FutureMonitor` instead captures the metrics of *one* future so they can be
3262/// attached to that future's own record. Idle, poll, and first-poll metrics are
3263/// measured locally from the future. Scheduling delay — which only the
3264/// root future the runtime schedules can observe — is read from the surrounding
3265/// task's [`TaskScheduling`] log over the future's lifetime, so for scheduling
3266/// metrics to be populated the surrounding task must itself be instrumented with
3267/// [`TaskMonitor::instrument`].
3268///
3269/// `FutureMonitor` instruments exactly one future: [`instrument`] consumes the
3270/// monitor, so it cannot be reused, and the returned future resolves to the
3271/// wrapped output together with the captured [`FutureMetrics`].
3272///
3273/// ##### Examples
3274/// ```
3275/// use tokio_metrics::{FutureMonitor, TaskMonitor};
3276///
3277/// #[tokio::main]
3278/// async fn main() {
3279///     // the larger task is instrumented once; `publish_scheduling_delay`
3280///     // enables per-future scheduling-delay capture
3281///     let mut builder = TaskMonitor::builder();
3282///     builder.publish_scheduling_delay();
3283///     let task_monitor = builder.build();
3284///     task_monitor.instrument(async {
3285///         // each unit of work within the task is measured on its own
3286///         let (_output, metrics) = FutureMonitor::new()
3287///             .instrument(async {
3288///                 tokio::task::yield_now().await;
3289///             })
3290///             .await;
3291///
3292///         assert!(metrics.poll_count >= 1);
3293///     }).await;
3294/// }
3295/// ```
3296///
3297/// [`instrument`]: FutureMonitor::instrument
3298// Deliberately not `Clone`: cloning would share the underlying span and let two
3299// instrumented futures be measured as one, which is exactly the reuse `instrument`
3300// (by consuming `self`) is meant to prevent.
3301#[derive(Debug, Default)]
3302pub struct FutureMonitor {
3303    monitor: TaskMonitor,
3304    span: Arc<SpanState>,
3305}
3306
3307impl FutureMonitor {
3308    /// Constructs a new `FutureMonitor` for a single future.
3309    pub fn new() -> FutureMonitor {
3310        FutureMonitor::default()
3311    }
3312
3313    /// Constructs a new `FutureMonitor` whose local poll metrics use a custom
3314    /// slow-poll threshold (see [`TaskMonitor::with_slow_poll_threshold`]).
3315    pub fn with_slow_poll_threshold(slow_poll_threshold: Duration) -> FutureMonitor {
3316        FutureMonitor {
3317            monitor: TaskMonitor::with_slow_poll_threshold(slow_poll_threshold),
3318            span: Arc::new(SpanState::default()),
3319        }
3320    }
3321
3322    /// Instruments the future, consuming the monitor.
3323    ///
3324    /// Taking `self` by value means a `FutureMonitor` can only instrument one
3325    /// future. Await the returned [`MonitoredFuture`] to get the future's
3326    /// output alongside its [`FutureMetrics`].
3327    pub fn instrument<F>(self, task: F) -> MonitoredFuture<F> {
3328        MonitoredFuture {
3329            inner: self.monitor.instrument(task),
3330            monitor: self.monitor,
3331            span: self.span,
3332        }
3333    }
3334}
3335
3336/// Iterator returned by [`TaskMonitor::intervals`].
3337///
3338/// See that method's documentation for more details.
3339#[derive(Debug)]
3340pub struct TaskIntervals<M: AsRef<TaskMonitorCore> + Send + Sync + 'static = TaskMonitor> {
3341    monitor: M,
3342    previous: Option<TaskMetrics>,
3343}
3344
3345impl<M: AsRef<TaskMonitorCore> + Send + Sync + 'static> TaskIntervals<M> {
3346    fn probe(&mut self) -> TaskMetrics {
3347        let latest = self.monitor.as_ref().metrics.metrics();
3348        let local_max_idle_duration = self
3349            .monitor
3350            .as_ref()
3351            .metrics
3352            .get_and_reset_local_max_idle_duration();
3353
3354        let next = if let Some(previous) = self.previous {
3355            TaskMetrics {
3356                instrumented_count: latest
3357                    .instrumented_count
3358                    .wrapping_sub(previous.instrumented_count),
3359                dropped_count: latest.dropped_count.wrapping_sub(previous.dropped_count),
3360                total_poll_count: latest
3361                    .total_poll_count
3362                    .wrapping_sub(previous.total_poll_count),
3363                total_poll_duration: sub(latest.total_poll_duration, previous.total_poll_duration),
3364                first_poll_count: latest
3365                    .first_poll_count
3366                    .wrapping_sub(previous.first_poll_count),
3367                total_idled_count: latest
3368                    .total_idled_count
3369                    .wrapping_sub(previous.total_idled_count),
3370                total_scheduled_count: latest
3371                    .total_scheduled_count
3372                    .wrapping_sub(previous.total_scheduled_count),
3373                total_fast_poll_count: latest
3374                    .total_fast_poll_count
3375                    .wrapping_sub(previous.total_fast_poll_count),
3376                total_short_delay_count: latest
3377                    .total_short_delay_count
3378                    .wrapping_sub(previous.total_short_delay_count),
3379                total_slow_poll_count: latest
3380                    .total_slow_poll_count
3381                    .wrapping_sub(previous.total_slow_poll_count),
3382                total_long_delay_count: latest
3383                    .total_long_delay_count
3384                    .wrapping_sub(previous.total_long_delay_count),
3385                total_first_poll_delay: sub(
3386                    latest.total_first_poll_delay,
3387                    previous.total_first_poll_delay,
3388                ),
3389                max_idle_duration: local_max_idle_duration,
3390                total_idle_duration: sub(latest.total_idle_duration, previous.total_idle_duration),
3391                total_scheduled_duration: sub(
3392                    latest.total_scheduled_duration,
3393                    previous.total_scheduled_duration,
3394                ),
3395                total_fast_poll_duration: sub(
3396                    latest.total_fast_poll_duration,
3397                    previous.total_fast_poll_duration,
3398                ),
3399                total_short_delay_duration: sub(
3400                    latest.total_short_delay_duration,
3401                    previous.total_short_delay_duration,
3402                ),
3403                total_slow_poll_duration: sub(
3404                    latest.total_slow_poll_duration,
3405                    previous.total_slow_poll_duration,
3406                ),
3407                total_long_delay_duration: sub(
3408                    latest.total_long_delay_duration,
3409                    previous.total_long_delay_duration,
3410                ),
3411            }
3412        } else {
3413            latest
3414        };
3415
3416        self.previous = Some(latest);
3417
3418        next
3419    }
3420}
3421
3422impl<M: AsRef<TaskMonitorCore> + Send + Sync + 'static> Iterator for TaskIntervals<M> {
3423    type Item = TaskMetrics;
3424
3425    fn next(&mut self) -> Option<Self::Item> {
3426        Some(self.probe())
3427    }
3428}
3429
3430#[inline(always)]
3431fn to_nanos(d: Duration) -> u64 {
3432    debug_assert!(d <= Duration::from_nanos(u64::MAX));
3433    d.as_secs()
3434        .wrapping_mul(1_000_000_000)
3435        .wrapping_add(d.subsec_nanos() as u64)
3436}
3437
3438#[inline(always)]
3439fn sub(a: Duration, b: Duration) -> Duration {
3440    let nanos = to_nanos(a).wrapping_sub(to_nanos(b));
3441    Duration::from_nanos(nanos)
3442}
3443
3444#[inline(always)]
3445fn mean(d: Duration, count: u64) -> Duration {
3446    if let Some(quotient) = to_nanos(d).checked_div(count) {
3447        Duration::from_nanos(quotient)
3448    } else {
3449        Duration::ZERO
3450    }
3451}
3452
3453#[cfg(test)]
3454mod inference_tests {
3455    use super::*;
3456    use std::future::Future;
3457    use std::pin::Pin;
3458
3459    // Type alias — M defaults to TaskMonitor
3460    type _BoxedInstrumented = Instrumented<Pin<Box<dyn Future<Output = ()>>>>;
3461
3462    // Struct field — M defaults to TaskMonitor
3463    struct _Wrapper {
3464        _fut: Instrumented<Pin<Box<dyn Future<Output = ()>>>>,
3465    }
3466
3467    // Partial type annotation — M defaults to TaskMonitor
3468    async fn _partial_annotation(monitor: &TaskMonitor) {
3469        let fut: Instrumented<_> = monitor.instrument(async { 42 });
3470        fut.await;
3471    }
3472
3473    // Common path — fully inferred from instrument()'s return type
3474    async fn _common_usage(monitor: &TaskMonitor) {
3475        monitor.instrument(async { 42 }).await;
3476    }
3477
3478    // Storing without annotation — both T and M inferred
3479    async fn _store_without_annotation(monitor: &TaskMonitor) {
3480        let fut = monitor.instrument(async { 42 });
3481        fut.await;
3482    }
3483
3484    // Function boundary — M defaults to TaskMonitor in the signature
3485    async fn _function_boundary(fut: Instrumented<impl Future<Output = i32>>) -> i32 {
3486        fut.await
3487    }
3488
3489    // Return position — M defaults to TaskMonitor
3490    fn _return_position(monitor: &TaskMonitor) -> Instrumented<impl Future<Output = i32> + '_> {
3491        monitor.instrument(async { 42 })
3492    }
3493
3494    // intervals() inference
3495    fn _intervals_inference(monitor: &TaskMonitor) {
3496        let mut intervals = monitor.intervals();
3497        let _: Option<TaskMetrics> = intervals.next();
3498    }
3499
3500    #[tokio::test]
3501    async fn inference_compiles() {
3502        let monitor = TaskMonitor::new();
3503        _partial_annotation(&monitor).await;
3504        _common_usage(&monitor).await;
3505        _store_without_annotation(&monitor).await;
3506        _function_boundary(monitor.instrument(async { 42 })).await;
3507        _return_position(&monitor).await;
3508        _intervals_inference(&monitor);
3509    }
3510}
3511
3512#[cfg(test)]
3513mod future_monitor_tests {
3514    use super::*;
3515
3516    // Asserts on durations measured by the crate's clock, which is only the
3517    // runtime's virtual clock under `start_paused` when the `rt` feature selects
3518    // `tokio::time::Instant`; without `rt` the crate uses the real `std` clock.
3519    #[cfg(feature = "rt")]
3520    #[tokio::test(flavor = "current_thread", start_paused = true)]
3521    async fn captures_per_future_idle_active_polls() {
3522        let task_monitor = TaskMonitor::new();
3523        task_monitor
3524            .instrument(async {
3525                let (_, m) = FutureMonitor::new()
3526                    .instrument(async {
3527                        tokio::task::yield_now().await; // an extra poll
3528                        tokio::time::sleep(Duration::from_secs(1)).await; // idle
3529                    })
3530                    .await;
3531
3532                assert!(m.poll_count >= 2, "poll_count = {}", m.poll_count);
3533                assert!(m.idle_count >= 1, "idle_count = {}", m.idle_count);
3534                assert!(
3535                    m.total_idle_duration >= Duration::from_secs(1),
3536                    "total_idle_duration = {:?}",
3537                    m.total_idle_duration
3538                );
3539            })
3540            .await;
3541    }
3542
3543    #[tokio::test(flavor = "current_thread", start_paused = true)]
3544    async fn scheduling_is_zero_without_instrumented_root() {
3545        // No surrounding `TaskMonitor::instrument`, so `try_current()` is `None`
3546        // and no scheduling delay is attributed — but local metrics still work.
3547        let (_, m) = FutureMonitor::new()
3548            .instrument(async {
3549                tokio::task::yield_now().await;
3550            })
3551            .await;
3552
3553        assert_eq!(m.scheduled_count, 0);
3554        assert_eq!(m.total_scheduled_duration, Duration::ZERO);
3555        assert!(m.poll_count >= 1, "poll_count = {}", m.poll_count);
3556    }
3557
3558    #[tokio::test]
3559    async fn try_current_tracks_root_scheduling_when_opted_in() {
3560        assert!(TaskScheduling::try_current().is_none());
3561
3562        let mut builder = TaskMonitor::builder();
3563        builder.publish_scheduling_delay();
3564        let monitor = builder.build();
3565        monitor
3566            .instrument(async {
3567                // The log exists from the first poll, even before any scheduling.
3568                let before = TaskScheduling::try_current().expect("inside instrumented task");
3569                tokio::task::yield_now().await;
3570                let after = TaskScheduling::try_current().expect("inside instrumented task");
3571                assert!(
3572                    after.scheduled_count > before.scheduled_count,
3573                    "scheduled_count should increase after yielding: {} -> {}",
3574                    before.scheduled_count,
3575                    after.scheduled_count
3576                );
3577            })
3578            .await;
3579
3580        assert!(TaskScheduling::try_current().is_none());
3581    }
3582
3583    #[tokio::test]
3584    async fn try_current_is_none_without_opt_in() {
3585        // A default monitor does not publish the scheduling log, so the common
3586        // case pays nothing and `FutureMonitor` scheduling stays zero.
3587        TaskMonitor::new()
3588            .instrument(async {
3589                assert!(TaskScheduling::try_current().is_none());
3590            })
3591            .await;
3592    }
3593
3594    #[tokio::test]
3595    async fn instrument_yields_output_and_metrics() {
3596        // `instrument` consumes the monitor and resolves to the future's output
3597        // paired with its captured metrics — so a monitor can only ever measure
3598        // a single future.
3599        let (output, m) = FutureMonitor::new()
3600            .instrument(async {
3601                tokio::task::yield_now().await;
3602                "done"
3603            })
3604            .await;
3605
3606        assert_eq!(output, "done");
3607        assert!(m.poll_count >= 1, "poll_count = {}", m.poll_count);
3608    }
3609
3610    #[tokio::test(flavor = "current_thread")]
3611    async fn total_duration_includes_final_poll_execution() {
3612        // The future completes in a single poll that spends real time
3613        // executing. `total_duration` is stamped after the poll, so it must
3614        // include that execution time (a regression would record it before the
3615        // poll and report ~0).
3616        let (_, m) = FutureMonitor::new()
3617            .instrument(async {
3618                let start = std::time::Instant::now();
3619                while start.elapsed() < Duration::from_millis(20) {}
3620            })
3621            .await;
3622
3623        assert!(m.poll_count == 1, "poll_count = {}", m.poll_count);
3624        assert!(
3625            m.total_duration >= Duration::from_millis(15),
3626            "total_duration = {:?}",
3627            m.total_duration
3628        );
3629    }
3630
3631    #[tokio::test(flavor = "current_thread", start_paused = true)]
3632    async fn nested_future_monitors_capture_independently() {
3633        let mut builder = TaskMonitor::builder();
3634        builder.publish_scheduling_delay();
3635        let root = builder.build();
3636
3637        root.instrument(async {
3638            // An outer monitored future that does one extra poll of its own, then
3639            // runs an inner monitored future that sleeps.
3640            let (inner, outer) = FutureMonitor::new()
3641                .instrument(async {
3642                    tokio::task::yield_now().await; // outer-only extra poll
3643                    let (_, inner) = FutureMonitor::new()
3644                        .instrument(async {
3645                            tokio::time::sleep(Duration::from_secs(1)).await;
3646                        })
3647                        .await;
3648                    inner
3649                })
3650                .await;
3651
3652            // Each monitor captures only its own future: the inner one sees just
3653            // the sleep, the outer one additionally sees its yield poll.
3654            assert_eq!(
3655                inner.poll_count, 2,
3656                "inner poll_count = {}",
3657                inner.poll_count
3658            );
3659            assert_eq!(inner.idle_count, 1);
3660            assert_eq!(inner.total_idle_duration, Duration::from_secs(1));
3661
3662            assert!(
3663                outer.poll_count > inner.poll_count,
3664                "outer {} should exceed inner {}",
3665                outer.poll_count,
3666                inner.poll_count
3667            );
3668            // The outer future is idle for the whole time the inner one sleeps.
3669            assert_eq!(outer.total_idle_duration, Duration::from_secs(1));
3670        })
3671        .await;
3672    }
3673}