rt: instrument task poll times with a histogram

tokio/src/runtime/builder.rs

@@ -1,5 +1,5 @@
 use crate::runtime::handle::Handle;
-use crate::runtime::{blocking, driver, Callback, Runtime};
+use crate::runtime::{blocking, driver, Callback, HistogramBuilder, Runtime};
 use crate::util::rand::{RngSeed, RngSeedGenerator};
 
 use std::fmt;
@@ -95,6 +95,12 @@ pub struct Builder {
     /// Specify a random number generator seed to provide deterministic results
     pub(super) seed_generator: RngSeedGenerator,
 
+    /// When true, enables task poll count histogram instrumentation.
+    pub(super) metrics_poll_count_histogram_enable: bool,
+
+    /// Configures the task poll count histogram
+    pub(super) metrics_poll_count_histogram: HistogramBuilder,
+
     #[cfg(tokio_unstable)]
     pub(super) unhandled_panic: UnhandledPanic,
 }
@@ -268,6 +274,10 @@ impl Builder {
             #[cfg(tokio_unstable)]
             unhandled_panic: UnhandledPanic::Ignore,
 
+            metrics_poll_count_histogram_enable: false,
+
+            metrics_poll_count_histogram: Default::default(),
+
             disable_lifo_slot: false,
         }
     }
@@ -877,6 +887,127 @@ impl Builder {
         }
     }
 
+    cfg_metrics! {
+        /// Enables tracking the distribution of task poll times.
+        ///
+        /// Task poll times are not instrumented by default as doing so requires
+        /// calling [`Instant::now()]` twice per task poll, which could add
+        /// measurable overhead. Use the [`Handle::metrics()`] to access the
+        /// metrics data.
+        ///
+        /// The histogram uses fixed bucket sizes. In other words, the histogram
+        /// buckets are not dynamic based on input values. Use the
+        /// `metrics_poll_count_histogram_` builder methods to configure the
+        /// histogram details.
+        ///
+        /// # Examples
+        ///
+        /// ```
+        /// use tokio::runtime;
+        ///
+        /// let rt = runtime::Builder::new_multi_thread()
+        ///     .enable_metrics_poll_count_histogram()
+        ///     .build()
+        ///     .unwrap();
+        /// # // Test default values here
+        /// # fn us(n: u64) -> std::time::Duration { std::time::Duration::from_micros(n) }
+        /// # let m = rt.handle().metrics();
+        /// # assert_eq!(m.poll_count_histogram_num_buckets(), 10);
+        /// # assert_eq!(m.poll_count_histogram_bucket_range(0), us(0)..us(100));
+        /// # assert_eq!(m.poll_count_histogram_bucket_range(1), us(100)..us(200));
+        /// ```
+        ///
+        /// [`Handle::metrics()`]: crate::runtime::Handle::metrics
+        pub fn enable_metrics_poll_count_histogram(&mut self) -> &mut Self {
+            self.metrics_poll_count_histogram_enable = true;
+            self
+        }
+
+        /// Sets the histogram scale for tracking the distribution of task poll
+        /// times.
+        ///
+        /// Tracking the distribution of task poll times can be done using a
+        /// linear or log scale. When using linear scale, each histogram bucket
+        /// will represent the same range of poll times. When using log scale,
+        /// each histogram bucket will cover a range twice as big as the
+        /// previous bucket.
+        ///
+        /// **Default:** linear scale.
+        ///
+        /// # Examples
+        ///
+        /// ```
+        /// use tokio::runtime::{self, HistogramScale};
+        ///
+        /// let rt = runtime::Builder::new_multi_thread()
+        ///     .enable_metrics_poll_count_histogram()
+        ///     .metrics_poll_count_histogram_scale(HistogramScale::Log)
+        ///     .build()
+        ///     .unwrap();
+        /// ```
+        pub fn metrics_poll_count_histogram_scale(&mut self, histogram_scale: crate::runtime::HistogramScale) -> &mut Self {
+            self.metrics_poll_count_histogram.scale = histogram_scale;
+            self
+        }
+
+        /// Sets the histogram resolution for tracking the distribution of task
+        /// poll times.
+        ///
+        /// The resolution is the histogram's first bucket's range. When using a
+        /// linear histogram scale, each bucket will cover the same range. When
+        /// using a log scale, each bucket will cover a range twice as big as
+        /// the previous bucket. In the log case, the resolution represents the
+        /// smallest bucket range.
+        ///
+        /// Note that, when using log scale, the resolution is rounded up to the
+        /// nearest power of 2 in nanoseconds.
+        ///
+        /// **Default:** 100 microseconds.
+        ///
+        /// # Examples
+        ///
+        /// ```
+        /// use tokio::runtime;
+        /// use std::time::Duration;
+        ///
+        /// let rt = runtime::Builder::new_multi_thread()
+        ///     .enable_metrics_poll_count_histogram()
+        ///     .metrics_poll_count_histogram_resolution(Duration::from_micros(100))
+        ///     .build()
+        ///     .unwrap();
+        /// ```
+        pub fn metrics_poll_count_histogram_resolution(&mut self, resolution: Duration) -> &mut Self {
+            assert!(resolution > Duration::from_secs(0));
+            // Sanity check the argument and also make the cast below safe.
+            assert!(resolution <= Duration::from_secs(1));
+
+            let resolution = resolution.as_nanos() as u64;
+            self.metrics_poll_count_histogram.resolution = resolution;
+            self
+        }
+
+        /// Sets the number of buckets for the histogram tracking the
+        /// distribution of task poll times.
+        ///
+        /// **Default:** 10
+        ///
+        /// # Examples
+        ///
+        /// ```
+        /// use tokio::runtime;
+        ///
+        /// let rt = runtime::Builder::new_multi_thread()
+        ///     .enable_metrics_poll_count_histogram()
+        ///     .metrics_poll_count_histogram_buckets(15)
+        ///     .build()
+        ///     .unwrap();
+        /// ```
+        pub fn metrics_poll_count_histogram_buckets(&mut self, buckets: usize) -> &mut Self {
+            self.metrics_poll_count_histogram.num_buckets = buckets;
+            self
+        }
+    }
+
     fn build_current_thread_runtime(&mut self) -> io::Result<Runtime> {
         use crate::runtime::scheduler::{self, CurrentThread};
         use crate::runtime::{runtime::Scheduler, Config};
@@ -909,6 +1040,7 @@ impl Builder {
                 unhandled_panic: self.unhandled_panic.clone(),
                 disable_lifo_slot: self.disable_lifo_slot,
                 seed_generator: seed_generator_1,
+                metrics_poll_count_histogram: self.metrics_poll_count_histogram_builder(),
             },
         );
 
@@ -922,6 +1054,14 @@ impl Builder {
             blocking_pool,
         ))
     }
+
+    fn metrics_poll_count_histogram_builder(&self) -> Option<HistogramBuilder> {
+        if self.metrics_poll_count_histogram_enable {
+            Some(self.metrics_poll_count_histogram.clone())
+        } else {
+            None
+        }
+    }
 }
 
 cfg_io_driver! {
@@ -1050,6 +1190,7 @@ cfg_rt_multi_thread! {
                     unhandled_panic: self.unhandled_panic.clone(),
                     disable_lifo_slot: self.disable_lifo_slot,
                     seed_generator: seed_generator_1,
+                    metrics_poll_count_histogram: self.metrics_poll_count_histogram_builder(),
                 },
             );
 

tokio/src/runtime/config.rs

@@ -28,6 +28,9 @@ pub(crate) struct Config {
     /// deterministic way.
     pub(crate) seed_generator: RngSeedGenerator,
 
+    /// How to build poll time histograms
+    pub(crate) metrics_poll_count_histogram: Option<crate::runtime::HistogramBuilder>,
+
     #[cfg(tokio_unstable)]
     /// How to respond to unhandled task panics.
     pub(crate) unhandled_panic: crate::runtime::UnhandledPanic,

tokio/src/runtime/metrics/batch.rs

@@ -1,7 +1,7 @@
-use crate::runtime::WorkerMetrics;
+use crate::runtime::metrics::{HistogramBatch, WorkerMetrics};
 
 use std::sync::atomic::Ordering::Relaxed;
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 pub(crate) struct MetricsBatch {
     /// Number of times the worker parked.
@@ -32,11 +32,26 @@ pub(crate) struct MetricsBatch {
 
     /// The total busy duration in nanoseconds.
     busy_duration_total: u64,
+
+    /// Instant at which work last resumed (continued after park).
     last_resume_time: Instant,
+
+    /// If `Some`, tracks poll times in nanoseconds
+    poll_timer: Option<PollTimer>,
+}
+
+struct PollTimer {
+    /// Histogram of poll counts within each band.
+    poll_counts: HistogramBatch,
+
+    /// Instant when the most recent task started polling.
+    poll_started_at: Instant,
 }
 
 impl MetricsBatch {
-    pub(crate) fn new() -> MetricsBatch {
+    pub(crate) fn new(worker_metrics: &WorkerMetrics) -> MetricsBatch {
+        let now = Instant::now();
+
         MetricsBatch {
             park_count: 0,
             noop_count: 0,
@@ -47,7 +62,14 @@ impl MetricsBatch {
             local_schedule_count: 0,
             overflow_count: 0,
             busy_duration_total: 0,
-            last_resume_time: Instant::now(),
+            last_resume_time: now,
+            poll_timer: worker_metrics
+                .poll_count_histogram
+                .as_ref()
+                .map(|worker_poll_counts| PollTimer {
+                    poll_counts: HistogramBatch::from_histogram(worker_poll_counts),
+                    poll_started_at: now,
+                }),
         }
     }
 
@@ -68,6 +90,11 @@ impl MetricsBatch {
             .local_schedule_count
             .store(self.local_schedule_count, Relaxed);
         worker.overflow_count.store(self.overflow_count, Relaxed);
+
+        if let Some(poll_timer) = &self.poll_timer {
+            let dst = worker.poll_count_histogram.as_ref().unwrap();
+            poll_timer.poll_counts.submit(dst);
+        }
     }
 
     /// The worker is about to park.
@@ -81,8 +108,22 @@ impl MetricsBatch {
         }
 
         let busy_duration = self.last_resume_time.elapsed();
-        let busy_duration = u64::try_from(busy_duration.as_nanos()).unwrap_or(u64::MAX);
-        self.busy_duration_total += busy_duration;
+        self.busy_duration_total += duration_as_u64(busy_duration);
+    }
+
+    pub(crate) fn start_poll(&mut self) {
+        self.poll_count += 1;
+
+        if let Some(poll_timer) = &mut self.poll_timer {
+            poll_timer.poll_started_at = Instant::now();
+        }
+    }
+
+    pub(crate) fn end_poll(&mut self) {
+        if let Some(poll_timer) = &mut self.poll_timer {
+            let elapsed = duration_as_u64(poll_timer.poll_started_at.elapsed());
+            poll_timer.poll_counts.measure(elapsed, 1);
+        }
     }
 
     pub(crate) fn returned_from_park(&mut self) {
@@ -92,10 +133,6 @@ impl MetricsBatch {
     pub(crate) fn inc_local_schedule_count(&mut self) {
         self.local_schedule_count += 1;
     }
-
-    pub(crate) fn incr_poll_count(&mut self) {
-        self.poll_count += 1;
-    }
 }
 
 cfg_rt_multi_thread! {
@@ -113,3 +150,7 @@ cfg_rt_multi_thread! {
         }
     }
 }
+
+fn duration_as_u64(dur: Duration) -> u64 {
+    u64::try_from(dur.as_nanos()).unwrap_or(u64::MAX)
+}

tokio/src/runtime/metrics/mock.rs

@@ -6,6 +6,9 @@ pub(crate) struct WorkerMetrics {}
 
 pub(crate) struct MetricsBatch {}
 
+#[derive(Clone, Default)]
+pub(crate) struct HistogramBuilder {}
+
 impl SchedulerMetrics {
     pub(crate) fn new() -> Self {
         Self {}
@@ -20,19 +23,26 @@ impl WorkerMetrics {
         Self {}
     }
 
+    pub(crate) fn from_config(config: &crate::runtime::Config) -> Self {
+        // Prevent the dead-code warning from being triggered
+        let _ = &config.metrics_poll_count_histogram;
+        Self::new()
+    }
+
     pub(crate) fn set_queue_depth(&self, _len: usize) {}
 }
 
 impl MetricsBatch {
-    pub(crate) fn new() -> Self {
+    pub(crate) fn new(_: &WorkerMetrics) -> Self {
         Self {}
     }
 
     pub(crate) fn submit(&mut self, _to: &WorkerMetrics) {}
     pub(crate) fn about_to_park(&mut self) {}
     pub(crate) fn returned_from_park(&mut self) {}
-    pub(crate) fn incr_poll_count(&mut self) {}
     pub(crate) fn inc_local_schedule_count(&mut self) {}
+    pub(crate) fn start_poll(&mut self) {}
+    pub(crate) fn end_poll(&mut self) {}
 }
 
 cfg_rt_multi_thread! {

tokio/src/runtime/metrics/mod.rs

@@ -12,6 +12,11 @@ cfg_metrics! {
     mod batch;
     pub(crate) use batch::MetricsBatch;
 
+    mod histogram;
+    pub(crate) use histogram::{Histogram, HistogramBatch, HistogramBuilder};
+    #[allow(unreachable_pub)] // rust-lang/rust#57411
+    pub use histogram::HistogramScale;
+
     mod runtime;
     #[allow(unreachable_pub)] // rust-lang/rust#57411
     pub use runtime::RuntimeMetrics;
@@ -31,5 +36,5 @@ cfg_metrics! {
 cfg_not_metrics! {
     mod mock;
 
-    pub(crate) use mock::{SchedulerMetrics, WorkerMetrics, MetricsBatch};
+    pub(crate) use mock::{SchedulerMetrics, WorkerMetrics, MetricsBatch, HistogramBuilder};
 }

tokio/src/runtime/metrics/runtime.rs

@@ -1,5 +1,6 @@
 use crate::runtime::Handle;
 
+use std::ops::Range;
 use std::sync::atomic::Ordering::Relaxed;
 use std::time::Duration;
 
@@ -578,6 +579,189 @@ impl RuntimeMetrics {
         self.handle.inner.worker_local_queue_depth(worker)
     }
 
+    /// Returns `true` if the runtime is tracking the distribution of task poll
+    /// times.
+    ///
+    /// Task poll times are not instrumented by default as doing so requires
+    /// calling [`Instant::now()`] twice per task poll. The feature is enabled
+    /// by calling [`enable_metrics_poll_count_histogram()`] when building the
+    /// runtime.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::runtime::{self, Handle};
+    ///
+    /// fn main() {
+    ///     runtime::Builder::new_current_thread()
+    ///         .enable_metrics_poll_count_histogram()
+    ///         .build()
+    ///         .unwrap()
+    ///         .block_on(async {
+    ///             let metrics = Handle::current().metrics();
+    ///             let enabled = metrics.poll_count_histogram_enabled();
+    ///
+    ///             println!("Tracking task poll time distribution: {:?}", enabled);
+    ///         });
+    /// }
+    /// ```
+    ///
+    /// [`enable_metrics_poll_count_histogram()`]: crate::runtime::Builder::enable_metrics_poll_count_histogram
+    /// [`Instant::now()`]: std::time::Instant::now
+    pub fn poll_count_histogram_enabled(&self) -> bool {
+        self.handle
+            .inner
+            .worker_metrics(0)
+            .poll_count_histogram
+            .is_some()
+    }
+
+    /// Returns the number of histogram buckets tracking the distribution of
+    /// task poll times.
+    ///
+    /// This value is configured by calling
+    /// [`metrics_poll_count_histogram_buckets()`] when building the runtime.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::runtime::{self, Handle};
+    ///
+    /// fn main() {
+    ///     runtime::Builder::new_current_thread()
+    ///         .enable_metrics_poll_count_histogram()
+    ///         .build()
+    ///         .unwrap()
+    ///         .block_on(async {
+    ///             let metrics = Handle::current().metrics();
+    ///             let buckets = metrics.poll_count_histogram_num_buckets();
+    ///
+    ///             println!("Histogram buckets: {:?}", buckets);
+    ///         });
+    /// }
+    /// ```
+    ///
+    /// [`metrics_poll_count_histogram_buckets()`]:
+    ///     crate::runtime::Builder::metrics_poll_count_histogram_buckets
+    pub fn poll_count_histogram_num_buckets(&self) -> usize {
+        self.handle
+            .inner
+            .worker_metrics(0)
+            .poll_count_histogram
+            .as_ref()
+            .map(|histogram| histogram.num_buckets())
+            .unwrap_or_default()
+    }
+
+    /// Returns the range of task poll times tracked by the given bucket.
+    ///
+    /// This value is configured by calling
+    /// [`metrics_poll_count_histogram_resolution()`] when building the runtime.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::runtime::{self, Handle};
+    ///
+    /// fn main() {
+    ///     runtime::Builder::new_current_thread()
+    ///         .enable_metrics_poll_count_histogram()
+    ///         .build()
+    ///         .unwrap()
+    ///         .block_on(async {
+    ///             let metrics = Handle::current().metrics();
+    ///             let buckets = metrics.poll_count_histogram_num_buckets();
+    ///
+    ///             for i in 0..buckets {
+    ///                 let range = metrics.poll_count_histogram_bucket_range(i);
+    ///                 println!("Histogram bucket {} range: {:?}", i, range);
+    ///             }
+    ///         });
+    /// }
+    /// ```
+    ///
+    /// [`metrics_poll_count_histogram_resolution()`]:
+    ///     crate::runtime::Builder::metrics_poll_count_histogram_resolution
+    pub fn poll_count_histogram_bucket_range(&self, bucket: usize) -> Range<Duration> {
+        self.handle
+            .inner
+            .worker_metrics(0)
+            .poll_count_histogram
+            .as_ref()
+            .map(|histogram| {
+                let range = histogram.bucket_range(bucket);
+                std::ops::Range {
+                    start: Duration::from_nanos(range.start),
+                    end: Duration::from_nanos(range.end),
+                }
+            })
+            .unwrap_or_default()
+    }
+
+    /// Returns the number of times the given worker polled tasks with a poll
+    /// duration within the given bucket's range.
+    ///
+    /// Each worker maintains its own histogram and the counts for each bucket
+    /// starts at zero when the runtime is created. Each time the worker polls a
+    /// task, it tracks the duration the task poll time took and increments the
+    /// associated bucket by 1.
+    ///
+    /// Each bucket is a monotonically increasing counter. It is never
+    /// decremented or reset to zero.
+    ///
+    /// # Arguments
+    ///
+    /// `worker` is the index of the worker being queried. The given value must
+    /// be between 0 and `num_workers()`. The index uniquely identifies a single
+    /// worker and will continue to identify the worker throughout the lifetime
+    /// of the runtime instance.
+    ///
+    /// `bucket` is the index of the bucket being queried. The bucket is scoped
+    /// to the worker. The range represented by the bucket can be queried by
+    /// calling [`poll_count_histogram_bucket_range()`]. Each worker maintains
+    /// identical bucket ranges.
+    ///
+    /// # Panics
+    ///
+    /// The method panics when `worker` represents an invalid worker, i.e. is
+    /// greater than or equal to `num_workers()` or if `bucket` represents an
+    /// invalid bucket.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::runtime::{self, Handle};
+    ///
+    /// fn main() {
+    ///     runtime::Builder::new_current_thread()
+    ///         .enable_metrics_poll_count_histogram()
+    ///         .build()
+    ///         .unwrap()
+    ///         .block_on(async {
+    ///             let metrics = Handle::current().metrics();
+    ///             let buckets = metrics.poll_count_histogram_num_buckets();
+    ///
+    ///             for worker in 0..metrics.num_workers() {
+    ///                 for i in 0..buckets {
+    ///                     let count = metrics.poll_count_histogram_bucket_count(worker, i);
+    ///                     println!("Poll count {}", count);
+    ///                 }
+    ///             }
+    ///         });
+    /// }
+    /// ```
+    ///
+    /// [`poll_count_histogram_bucket_range()`]: crate::runtime::RuntimeMetrics::poll_count_histogram_bucket_range
+    pub fn poll_count_histogram_bucket_count(&self, worker: usize, bucket: usize) -> u64 {
+        self.handle
+            .inner
+            .worker_metrics(worker)
+            .poll_count_histogram
+            .as_ref()
+            .map(|histogram| histogram.get(bucket))
+            .unwrap_or_default()
+    }
+
     /// Returns the number of tasks currently scheduled in the blocking
     /// thread pool, spawned using `spawn_blocking`.
     ///

tokio/src/runtime/metrics/worker.rs

@@ -1,5 +1,7 @@
 use crate::loom::sync::atomic::Ordering::Relaxed;
 use crate::loom::sync::atomic::{AtomicU64, AtomicUsize};
+use crate::runtime::metrics::Histogram;
+use crate::runtime::Config;
 
 /// Retrieve runtime worker metrics.
 ///
@@ -38,9 +40,21 @@ pub(crate) struct WorkerMetrics {
     /// Number of tasks currently in the local queue. Used only by the
     /// current-thread scheduler.
     pub(crate) queue_depth: AtomicUsize,
+
+    /// If `Some`, tracks the the number of polls by duration range.
+    pub(super) poll_count_histogram: Option<Histogram>,
 }
 
 impl WorkerMetrics {
+    pub(crate) fn from_config(config: &Config) -> WorkerMetrics {
+        let mut worker_metrics = WorkerMetrics::new();
+        worker_metrics.poll_count_histogram = config
+            .metrics_poll_count_histogram
+            .as_ref()
+            .map(|histogram_builder| histogram_builder.build());
+        worker_metrics
+    }
+
     pub(crate) fn new() -> WorkerMetrics {
         WorkerMetrics {
             park_count: AtomicU64::new(0),
@@ -52,6 +66,7 @@ impl WorkerMetrics {
             busy_duration_total: AtomicU64::new(0),
             local_schedule_count: AtomicU64::new(0),
             queue_depth: AtomicUsize::new(0),
+            poll_count_histogram: None,
         }
     }
 

tokio/src/runtime/mod.rs

@@ -249,9 +249,9 @@ cfg_rt! {
 
     cfg_metrics! {
         mod metrics;
-        pub use metrics::RuntimeMetrics;
+        pub use metrics::{RuntimeMetrics, HistogramScale};
 
-        pub(crate) use metrics::{MetricsBatch, SchedulerMetrics, WorkerMetrics};
+        pub(crate) use metrics::{MetricsBatch, SchedulerMetrics, WorkerMetrics, HistogramBuilder};
 
         cfg_net! {
         pub(crate) use metrics::IoDriverMetrics;
@@ -260,7 +260,7 @@ cfg_rt! {
 
     cfg_not_metrics! {
         pub(crate) mod metrics;
-        pub(crate) use metrics::{SchedulerMetrics, WorkerMetrics, MetricsBatch};
+        pub(crate) use metrics::{SchedulerMetrics, WorkerMetrics, MetricsBatch, HistogramBuilder};
     }
 
     /// After thread starts / before thread stops

tokio/src/runtime/scheduler/current_thread.rs

@@ -109,14 +109,16 @@ impl CurrentThread {
         seed_generator: RngSeedGenerator,
         config: Config,
     ) -> (CurrentThread, Arc<Handle>) {
+        let worker_metrics = WorkerMetrics::from_config(&config);
+
         let handle = Arc::new(Handle {
             shared: Shared {
                 queue: Mutex::new(Some(VecDeque::with_capacity(INITIAL_CAPACITY))),
                 owned: OwnedTasks::new(),
                 woken: AtomicBool::new(false),
                 config,
                 scheduler_metrics: SchedulerMetrics::new(),
-                worker_metrics: WorkerMetrics::new(),
+                worker_metrics,
             },
             driver: driver_handle,
             blocking_spawner,
@@ -127,7 +129,7 @@ impl CurrentThread {
             tasks: VecDeque::with_capacity(INITIAL_CAPACITY),
             tick: 0,
             driver: Some(driver),
-            metrics: MetricsBatch::new(),
+            metrics: MetricsBatch::new(&handle.shared.worker_metrics),
             unhandled_panic: false,
         })));
 
@@ -291,8 +293,10 @@ impl Context {
     /// Execute the closure with the given scheduler core stored in the
     /// thread-local context.
     fn run_task<R>(&self, mut core: Box<Core>, f: impl FnOnce() -> R) -> (Box<Core>, R) {
-        core.metrics.incr_poll_count();
-        self.enter(core, || crate::runtime::coop::budget(f))
+        core.metrics.start_poll();
+        let mut ret = self.enter(core, || crate::runtime::coop::budget(f));
+        ret.0.metrics.end_poll();
+        ret
     }
 
     /// Blocks the current thread until an event is received by the driver,

tokio/src/runtime/scheduler/multi_thread/worker.rs

@@ -203,6 +203,7 @@ pub(super) fn create(
 
         let park = park.clone();
         let unpark = park.unpark();
+        let metrics = WorkerMetrics::from_config(&config);
 
         cores.push(Box::new(Core {
             tick: 0,
@@ -211,12 +212,12 @@ pub(super) fn create(
             is_searching: false,
             is_shutdown: false,
             park: Some(park),
-            metrics: MetricsBatch::new(),
+            metrics: MetricsBatch::new(&metrics),
             rand: FastRand::new(config.seed_generator.next_seed()),
         }));
 
         remotes.push(Remote { steal, unpark });
-        worker_metrics.push(WorkerMetrics::new());
+        worker_metrics.push(metrics);
     }
 
     let handle = Arc::new(Handle {
@@ -456,7 +457,7 @@ impl Context {
         core.transition_from_searching(&self.worker);
 
         // Make the core available to the runtime context
-        core.metrics.incr_poll_count();
+        core.metrics.start_poll();
         *self.core.borrow_mut() = Some(core);
 
         // Run the task
@@ -473,6 +474,13 @@ impl Context {
                     None => return Err(()),
                 };
 
+                // If task poll times is enabled, measure the poll time. Note
+                // that, if the `core` is stolen, this means `block_in_place`
+                // was called, turning the poll into a "blocking op". In this
+                // case, we don't want to measure the poll time as it doesn't
+                // really count as an async poll anymore.
+                core.metrics.end_poll();
+
                 // Check for a task in the LIFO slot
                 let task = match core.lifo_slot.take() {
                     Some(task) => task,
@@ -487,7 +495,7 @@ impl Context {
 
                 if coop::has_budget_remaining() {
                     // Run the LIFO task, then loop
-                    core.metrics.incr_poll_count();
+                    core.metrics.start_poll();
                     *self.core.borrow_mut() = Some(core);
                     let task = self.worker.handle.shared.owned.assert_owner(task);
                     task.run();

tokio/src/runtime/tests/loom_queue.rs

@@ -5,15 +5,19 @@ use crate::runtime::MetricsBatch;
 
 use loom::thread;
 
+fn metrics_batch() -> MetricsBatch {
+    MetricsBatch::new(&crate::runtime::WorkerMetrics::new())
+}
+
 #[test]
 fn basic() {
     loom::model(|| {
         let (steal, mut local) = queue::local();
         let inject = Inject::new();
-        let mut metrics = MetricsBatch::new();
+        let mut metrics = metrics_batch();
 
         let th = thread::spawn(move || {
-            let mut metrics = MetricsBatch::new();
+            let mut metrics = metrics_batch();
             let (_, mut local) = queue::local();
             let mut n = 0;
 
@@ -66,10 +70,10 @@ fn steal_overflow() {
     loom::model(|| {
         let (steal, mut local) = queue::local();
         let inject = Inject::new();
-        let mut metrics = MetricsBatch::new();
+        let mut metrics = metrics_batch();
 
         let th = thread::spawn(move || {
-            let mut metrics = MetricsBatch::new();
+            let mut metrics = metrics_batch();
             let (_, mut local) = queue::local();
             let mut n = 0;
 
@@ -118,7 +122,7 @@ fn multi_stealer() {
     const NUM_TASKS: usize = 5;
 
     fn steal_tasks(steal: queue::Steal<NoopSchedule>) -> usize {
-        let mut metrics = MetricsBatch::new();
+        let mut metrics = metrics_batch();
         let (_, mut local) = queue::local();
 
         if steal.steal_into(&mut local, &mut metrics).is_none() {
@@ -137,7 +141,7 @@ fn multi_stealer() {
     loom::model(|| {
         let (steal, mut local) = queue::local();
         let inject = Inject::new();
-        let mut metrics = MetricsBatch::new();
+        let mut metrics = metrics_batch();
 
         // Push work
         for _ in 0..NUM_TASKS {
@@ -172,7 +176,7 @@ fn multi_stealer() {
 #[test]
 fn chained_steal() {
     loom::model(|| {
-        let mut metrics = MetricsBatch::new();
+        let mut metrics = metrics_batch();
         let (s1, mut l1) = queue::local();
         let (s2, mut l2) = queue::local();
         let inject = Inject::new();
@@ -188,7 +192,7 @@ fn chained_steal() {
 
         // Spawn a task to steal from **our** queue
         let th = thread::spawn(move || {
-            let mut metrics = MetricsBatch::new();
+            let mut metrics = metrics_batch();
             let (_, mut local) = queue::local();
             s1.steal_into(&mut local, &mut metrics);
 

tokio/src/runtime/tests/queue.rs

@@ -21,11 +21,16 @@ macro_rules! assert_metrics {
     }};
 }
 
+fn metrics_batch() -> MetricsBatch {
+    use crate::runtime::WorkerMetrics;
+    MetricsBatch::new(&WorkerMetrics::new())
+}
+
 #[test]
 fn fits_256() {
     let (_, mut local) = queue::local();
     let inject = Inject::new();
-    let mut metrics = MetricsBatch::new();
+    let mut metrics = metrics_batch();
 
     for _ in 0..256 {
         let (task, _) = super::unowned(async {});
@@ -45,7 +50,7 @@ fn fits_256() {
 fn overflow() {
     let (_, mut local) = queue::local();
     let inject = Inject::new();
-    let mut metrics = MetricsBatch::new();
+    let mut metrics = metrics_batch();
 
     for _ in 0..257 {
         let (task, _) = super::unowned(async {});
@@ -71,7 +76,7 @@ fn overflow() {
 
 #[test]
 fn steal_batch() {
-    let mut metrics = MetricsBatch::new();
+    let mut metrics = metrics_batch();
 
     let (steal1, mut local1) = queue::local();
     let (_, mut local2) = queue::local();
@@ -117,14 +122,14 @@ fn stress1() {
     const NUM_PUSH: usize = normal_or_miri(500, 10);
     const NUM_POP: usize = normal_or_miri(250, 10);
 
-    let mut metrics = MetricsBatch::new();
+    let mut metrics = metrics_batch();
 
     for _ in 0..NUM_ITER {
         let (steal, mut local) = queue::local();
         let inject = Inject::new();
 
         let th = thread::spawn(move || {
-            let mut metrics = MetricsBatch::new();
+            let mut metrics = metrics_batch();
             let (_, mut local) = queue::local();
             let mut n = 0;
 
@@ -180,14 +185,14 @@ fn stress2() {
     const NUM_TASKS: usize = normal_or_miri(1_000_000, 50);
     const NUM_STEAL: usize = normal_or_miri(1_000, 10);
 
-    let mut metrics = MetricsBatch::new();
+    let mut metrics = metrics_batch();
 
     for _ in 0..NUM_ITER {
         let (steal, mut local) = queue::local();
         let inject = Inject::new();
 
         let th = thread::spawn(move || {
-            let mut stats = MetricsBatch::new();
+            let mut stats = metrics_batch();
             let (_, mut local) = queue::local();
             let mut n = 0;
 

tokio/tests/rt_metrics.rs

@@ -228,6 +228,12 @@ fn worker_poll_count() {
     drop(rt);
     assert_eq!(N, metrics.worker_poll_count(0));
 
+    // Does not populate the histogram
+    assert!(!metrics.poll_count_histogram_enabled());
+    for i in 0..10 {
+        assert_eq!(0, metrics.poll_count_histogram_bucket_count(0, i));
+    }
+
     let rt = threaded();
     let metrics = rt.metrics();
     rt.block_on(async {
@@ -242,6 +248,126 @@ fn worker_poll_count() {
         .sum();
 
     assert_eq!(N, n);
+
+    // Does not populate the histogram
+    assert!(!metrics.poll_count_histogram_enabled());
+    for n in 0..metrics.num_workers() {
+        for i in 0..10 {
+            assert_eq!(0, metrics.poll_count_histogram_bucket_count(n, i));
+        }
+    }
+}
+
+#[test]
+fn worker_poll_count_histogram() {
+    const N: u64 = 5;
+
+    let rts = [
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .enable_metrics_poll_count_histogram()
+            .metrics_poll_count_histogram_scale(tokio::runtime::HistogramScale::Linear)
+            .metrics_poll_count_histogram_buckets(3)
+            .metrics_poll_count_histogram_resolution(Duration::from_millis(50))
+            .build()
+            .unwrap(),
+        tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(2)
+            .enable_all()
+            .enable_metrics_poll_count_histogram()
+            .metrics_poll_count_histogram_scale(tokio::runtime::HistogramScale::Linear)
+            .metrics_poll_count_histogram_buckets(3)
+            .metrics_poll_count_histogram_resolution(Duration::from_millis(50))
+            .build()
+            .unwrap(),
+    ];
+
+    for rt in rts {
+        let metrics = rt.metrics();
+        rt.block_on(async {
+            for _ in 0..N {
+                tokio::spawn(async {}).await.unwrap();
+            }
+        });
+        drop(rt);
+
+        let num_workers = metrics.num_workers();
+        let num_buckets = metrics.poll_count_histogram_num_buckets();
+
+        assert!(metrics.poll_count_histogram_enabled());
+        assert_eq!(num_buckets, 3);
+
+        let n = (0..num_workers)
+            .flat_map(|i| (0..num_buckets).map(move |j| (i, j)))
+            .map(|(worker, bucket)| metrics.poll_count_histogram_bucket_count(worker, bucket))
+            .sum();
+        assert_eq!(N, n);
+    }
+}
+
+#[test]
+fn worker_poll_count_histogram_range() {
+    let max = Duration::from_nanos(u64::MAX);
+
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .enable_metrics_poll_count_histogram()
+        .metrics_poll_count_histogram_scale(tokio::runtime::HistogramScale::Linear)
+        .metrics_poll_count_histogram_buckets(3)
+        .metrics_poll_count_histogram_resolution(us(50))
+        .build()
+        .unwrap();
+    let metrics = rt.metrics();
+
+    assert_eq!(metrics.poll_count_histogram_bucket_range(0), us(0)..us(50));
+    assert_eq!(
+        metrics.poll_count_histogram_bucket_range(1),
+        us(50)..us(100)
+    );
+    assert_eq!(metrics.poll_count_histogram_bucket_range(2), us(100)..max);
+
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .enable_metrics_poll_count_histogram()
+        .metrics_poll_count_histogram_scale(tokio::runtime::HistogramScale::Log)
+        .metrics_poll_count_histogram_buckets(3)
+        .metrics_poll_count_histogram_resolution(us(50))
+        .build()
+        .unwrap();
+    let metrics = rt.metrics();
+
+    let a = Duration::from_nanos(50000_u64.next_power_of_two());
+    let b = a * 2;
+
+    assert_eq!(metrics.poll_count_histogram_bucket_range(0), us(0)..a);
+    assert_eq!(metrics.poll_count_histogram_bucket_range(1), a..b);
+    assert_eq!(metrics.poll_count_histogram_bucket_range(2), b..max);
+}
+
+#[test]
+fn worker_poll_count_histogram_disabled_without_explicit_enable() {
+    let rts = [
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .metrics_poll_count_histogram_scale(tokio::runtime::HistogramScale::Linear)
+            .metrics_poll_count_histogram_buckets(3)
+            .metrics_poll_count_histogram_resolution(Duration::from_millis(50))
+            .build()
+            .unwrap(),
+        tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(2)
+            .enable_all()
+            .metrics_poll_count_histogram_scale(tokio::runtime::HistogramScale::Linear)
+            .metrics_poll_count_histogram_buckets(3)
+            .metrics_poll_count_histogram_resolution(Duration::from_millis(50))
+            .build()
+            .unwrap(),
+    ];
+
+    for rt in rts {
+        let metrics = rt.metrics();
+        assert!(!metrics.poll_count_histogram_enabled());
+    }
 }
 
 #[test]
@@ -572,3 +698,7 @@ fn threaded() -> Runtime {
         .build()
         .unwrap()
 }
+
+fn us(n: u64) -> Duration {
+    Duration::from_micros(n)
+}