rt: instrument task poll times with a histogram

tokio/src/runtime/builder.rs

@@ -95,6 +95,8 @@ pub struct Builder {
     /// Specify a random number generator seed to provide deterministic results
     pub(super) seed_generator: RngSeedGenerator,
 
+    pub(super) metrics_poll_count_histogram: Option<crate::runtime::HistogramBuilder>,
+
     #[cfg(tokio_unstable)]
     pub(super) unhandled_panic: UnhandledPanic,
 }
@@ -268,6 +270,8 @@ impl Builder {
             #[cfg(tokio_unstable)]
             unhandled_panic: UnhandledPanic::Ignore,
 
+            metrics_poll_count_histogram: None,
+
             disable_lifo_slot: false,
         }
     }
@@ -877,6 +881,29 @@ impl Builder {
         }
     }
 
+    cfg_metrics! {
+        pub fn metrics_poll_count_histogram(&mut self, histogram_scale: crate::runtime::HistogramScale) -> &mut Self {
+            self.metrics_poll_count_histogram.get_or_insert_with(Default::default).scale = histogram_scale;
+            // self.metrics
+            self
+        }
+
+        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.get_or_insert_with(Default::default).resolution = resolution;
+            self
+        }
+
+        pub fn metrics_poll_count_histogram_buckets(&mut self, buckets: usize) -> &mut Self {
+            self.metrics_poll_count_histogram.get_or_insert_with(Default::default).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 +936,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.clone(),
             },
         );
 
@@ -1050,6 +1078,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.clone(),
                 },
             );
 

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,49 @@ 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,
+            steal_count: 0,
+            steal_operations: 0,
+            poll_count: 0,
+            poll_count_on_last_park: 0,
+            local_schedule_count: 0,
+            overflow_count: 0,
+            busy_duration_total: 0,
+            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,
+                }),
+        }
+    }
+
+    #[cfg(test)]
+    pub(crate) fn noop() -> MetricsBatch {
         MetricsBatch {
             park_count: 0,
             noop_count: 0,
@@ -48,6 +86,7 @@ impl MetricsBatch {
             overflow_count: 0,
             busy_duration_total: 0,
             last_resume_time: Instant::now(),
+            poll_timer: None,
         }
     }
 
@@ -68,6 +107,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 +125,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 +150,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 +167,7 @@ cfg_rt_multi_thread! {
         }
     }
 }
+
+fn duration_as_u64(dur: Duration) -> u64 {
+    u64::try_from(dur.as_nanos()).unwrap_or(u64::MAX)
+}