rt: instrument task poll times with a histogram

tokio/src/runtime/builder.rs

@@ -97,6 +97,9 @@ pub struct Builder {
 
     #[cfg(tokio_unstable)]
     pub(super) unhandled_panic: UnhandledPanic,
+
+    #[cfg(tokio_unstable)]
+    pub(super) metrics_poll_time_histogram: Option<crate::runtime::HistogramBuilder>,
 }
 
 cfg_unstable! {
@@ -268,6 +271,9 @@ impl Builder {
             #[cfg(tokio_unstable)]
             unhandled_panic: UnhandledPanic::Ignore,
 
+            #[cfg(tokio_unstable)]
+            metrics_poll_time_histogram: None,
+
             disable_lifo_slot: false,
         }
     }
@@ -875,6 +881,39 @@ impl Builder {
             self.seed_generator = RngSeedGenerator::new(seed);
             self
         }
+
+        pub fn metrics_poll_time_histogram(&mut self, histogram_scale: crate::runtime::HistogramScale) -> &mut Self {
+            self.metrics_poll_time_histogram.get_or_insert_with(Default::default).scale = histogram_scale;
+            // self.metrics
+            self
+        }
+
+        pub fn metrics_poll_time_histogram_resolution(&mut self, resolution: Duration) -> &mut Self {
+            // 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_time_histogram.get_or_insert_with(Default::default).resolution = resolution;
+            self
+        }
+
+        pub fn metrics_poll_time_histogram_buckets(&mut self, buckets: usize) -> &mut Self {
+            self.metrics_poll_time_histogram.get_or_insert_with(Default::default).num_buckets = buckets;
+            self
+        }
+
+        fn poll_time_histogram_builder(&self) -> Option<crate::runtime::HistogramBuilder> {
+            let mut builder = self.metrics_poll_time_histogram.clone();
+
+            // Do some sanity checks
+            if let Some(builder) = &mut builder {
+                if matches!(builder.scale, crate::runtime::HistogramScale::Log) {
+                    builder.resolution = builder.resolution.next_power_of_two();
+                }
+            }
+
+            builder
+        }
     }
 
     fn build_current_thread_runtime(&mut self) -> io::Result<Runtime> {
@@ -909,6 +948,8 @@ impl Builder {
                 unhandled_panic: self.unhandled_panic.clone(),
                 disable_lifo_slot: self.disable_lifo_slot,
                 seed_generator: seed_generator_1,
+                #[cfg(tokio_unstable)]
+                metrics_poll_time_histogram: self.poll_time_histogram_builder(),
             },
         );
 
@@ -1050,6 +1091,8 @@ cfg_rt_multi_thread! {
                     unhandled_panic: self.unhandled_panic.clone(),
                     disable_lifo_slot: self.disable_lifo_slot,
                     seed_generator: seed_generator_1,
+                    #[cfg(tokio_unstable)]
+                    metrics_poll_time_histogram: self.poll_time_histogram_builder(),
                 },
             );
 

tokio/src/runtime/config.rs

@@ -31,4 +31,8 @@ pub(crate) struct Config {
     #[cfg(tokio_unstable)]
     /// How to respond to unhandled task panics.
     pub(crate) unhandled_panic: crate::runtime::UnhandledPanic,
+
+    #[cfg(tokio_unstable)]
+    /// How to build poll time histograms
+    pub(crate) metrics_poll_time_histogram: Option<crate::runtime::HistogramBuilder>,
 }

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,29 @@ 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 `ns`
+    poll_timer: Option<PollTimer>,
+}
+
+struct PollTimer {
+    /// Histogram of mean time spent polling within each band.
+    poll_times: HistogramBatch,
+
+    /// 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 +65,15 @@ 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_times
+                .as_ref()
+                .map(|worker_poll_times| PollTimer {
+                    poll_times: HistogramBatch::from_histogram(&worker_poll_times.poll_times),
+                    poll_counts: HistogramBatch::from_histogram(&worker_poll_times.poll_counts),
+                    poll_started_at: now,
+                }),
         }
     }
 
@@ -68,6 +94,12 @@ 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_times.as_ref().unwrap();
+            poll_timer.poll_times.submit(&dst.poll_times);
+            poll_timer.poll_counts.submit(&dst.poll_counts);
+        }
     }
 
     /// The worker is about to park.
@@ -81,8 +113,23 @@ 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_times.measure(elapsed, elapsed);
+            poll_timer.poll_counts.measure(elapsed, 1);
+        }
     }
 
     pub(crate) fn returned_from_park(&mut self) {
@@ -92,10 +139,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 +156,7 @@ cfg_rt_multi_thread! {
         }
     }
 }
+
+fn duration_as_u64(dur: Duration) -> u64 {
+    u64::try_from(dur.as_nanos()).unwrap_or(u64::MAX)
+}