From c84d0a14b189c45da8f5e963fd3a83790ec92f8e Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Mon, 15 May 2023 15:20:41 -0700 Subject: [PATCH] rt: instrument task poll times with a histogram (#5685) Adds support for instrumenting the poll times of all spawned tasks. Data is tracked in a histogram. The user must specify the histogram scale and bucket ranges. Implementation-wise, the same strategy is used in the runtime where we are just using atomic counters. Because instrumenting each poll duration will result in frequent calls to `Instant::now()`, I think it should be an opt-in metric. --- tokio/src/runtime/builder.rs | 149 +++++- tokio/src/runtime/config.rs | 3 + tokio/src/runtime/metrics/batch.rs | 61 ++- tokio/src/runtime/metrics/histogram.rs | 502 ++++++++++++++++++ tokio/src/runtime/metrics/mock.rs | 14 +- tokio/src/runtime/metrics/mod.rs | 7 +- tokio/src/runtime/metrics/runtime.rs | 191 +++++++ tokio/src/runtime/metrics/worker.rs | 15 + tokio/src/runtime/mod.rs | 6 +- tokio/src/runtime/scheduler/current_thread.rs | 12 +- .../runtime/scheduler/multi_thread/worker.rs | 16 +- tokio/src/runtime/tests/loom_queue.rs | 20 +- tokio/src/runtime/tests/queue.rs | 19 +- tokio/tests/rt_metrics.rs | 130 +++++ 14 files changed, 1105 insertions(+), 40 deletions(-) create mode 100644 tokio/src/runtime/metrics/histogram.rs diff --git a/tokio/src/runtime/builder.rs b/tokio/src/runtime/builder.rs index ea0df2e3b..dda21a3ae 100644 --- a/tokio/src/runtime/builder.rs +++ b/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,133 @@ 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 + /// [`Instant::now()`]: std::time::Instant::now + 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. + /// + /// The last bucket tracks all greater values that fall out of other + /// ranges. So, configuring the histogram using a linear scale, + /// resolution of 50ms, and 10 buckets, the 10th bucket will track task + /// polls that take more than 450ms to complete. + /// + /// **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 { use crate::runtime::scheduler::{self, CurrentThread}; use crate::runtime::{runtime::Scheduler, Config}; @@ -909,6 +1046,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 +1060,14 @@ impl Builder { blocking_pool, )) } + + fn metrics_poll_count_histogram_builder(&self) -> Option { + if self.metrics_poll_count_histogram_enable { + Some(self.metrics_poll_count_histogram.clone()) + } else { + None + } + } } cfg_io_driver! { @@ -1050,6 +1196,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(), }, ); diff --git a/tokio/src/runtime/config.rs b/tokio/src/runtime/config.rs index 39eb1cf11..4af5eee47 100644 --- a/tokio/src/runtime/config.rs +++ b/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, + #[cfg(tokio_unstable)] /// How to respond to unhandled task panics. pub(crate) unhandled_panic: crate::runtime::UnhandledPanic, diff --git a/tokio/src/runtime/metrics/batch.rs b/tokio/src/runtime/metrics/batch.rs index ec2997413..296792d7f 100644 --- a/tokio/src/runtime/metrics/batch.rs +++ b/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, +} + +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) +} diff --git a/tokio/src/runtime/metrics/histogram.rs b/tokio/src/runtime/metrics/histogram.rs new file mode 100644 index 000000000..976f54fe8 --- /dev/null +++ b/tokio/src/runtime/metrics/histogram.rs @@ -0,0 +1,502 @@ +use crate::loom::sync::atomic::{AtomicU64, Ordering::Relaxed}; + +use std::cmp; +use std::ops::Range; + +#[derive(Debug)] +pub(crate) struct Histogram { + /// The histogram buckets + buckets: Box<[AtomicU64]>, + + /// Bucket scale, linear or log + scale: HistogramScale, + + /// Minimum resolution + resolution: u64, +} + +#[derive(Debug, Clone)] +pub(crate) struct HistogramBuilder { + /// Histogram scale + pub(crate) scale: HistogramScale, + + /// Must be a power of 2 + pub(crate) resolution: u64, + + /// Number of buckets + pub(crate) num_buckets: usize, +} + +#[derive(Debug)] +pub(crate) struct HistogramBatch { + buckets: Box<[u64]>, + scale: HistogramScale, + resolution: u64, +} + +cfg_unstable! { + /// Whether the histogram used to aggregate a metric uses a linear or + /// logarithmic scale. + #[derive(Debug, Copy, Clone, Eq, PartialEq)] + #[non_exhaustive] + pub enum HistogramScale { + /// Linear bucket scale + Linear, + + /// Logarithmic bucket scale + Log, + } +} + +impl Histogram { + pub(crate) fn num_buckets(&self) -> usize { + self.buckets.len() + } + + pub(crate) fn get(&self, bucket: usize) -> u64 { + self.buckets[bucket].load(Relaxed) + } + + pub(crate) fn bucket_range(&self, bucket: usize) -> Range { + match self.scale { + HistogramScale::Log => Range { + start: if bucket == 0 { + 0 + } else { + self.resolution << (bucket - 1) + }, + end: if bucket == self.buckets.len() - 1 { + u64::MAX + } else { + self.resolution << bucket + }, + }, + HistogramScale::Linear => Range { + start: self.resolution * bucket as u64, + end: if bucket == self.buckets.len() - 1 { + u64::MAX + } else { + self.resolution * (bucket as u64 + 1) + }, + }, + } + } +} + +impl HistogramBatch { + pub(crate) fn from_histogram(histogram: &Histogram) -> HistogramBatch { + let buckets = vec![0; histogram.buckets.len()].into_boxed_slice(); + + HistogramBatch { + buckets, + scale: histogram.scale, + resolution: histogram.resolution, + } + } + + pub(crate) fn measure(&mut self, value: u64, count: u64) { + self.buckets[self.value_to_bucket(value)] += count; + } + + pub(crate) fn submit(&self, histogram: &Histogram) { + debug_assert_eq!(self.scale, histogram.scale); + debug_assert_eq!(self.resolution, histogram.resolution); + debug_assert_eq!(self.buckets.len(), histogram.buckets.len()); + + for i in 0..self.buckets.len() { + histogram.buckets[i].store(self.buckets[i], Relaxed); + } + } + + fn value_to_bucket(&self, value: u64) -> usize { + match self.scale { + HistogramScale::Linear => { + let max = self.buckets.len() - 1; + cmp::min(value / self.resolution, max as u64) as usize + } + HistogramScale::Log => { + let max = self.buckets.len() - 1; + + if value < self.resolution { + 0 + } else { + let significant_digits = 64 - value.leading_zeros(); + let bucket_digits = 64 - (self.resolution - 1).leading_zeros(); + cmp::min(significant_digits as usize - bucket_digits as usize, max) + } + } + } + } +} + +impl HistogramBuilder { + pub(crate) fn new() -> HistogramBuilder { + HistogramBuilder { + scale: HistogramScale::Linear, + // Resolution is in nanoseconds. + resolution: 100_000, + num_buckets: 10, + } + } + + pub(crate) fn build(&self) -> Histogram { + let mut resolution = self.resolution; + + assert!(resolution > 0); + + if matches!(self.scale, HistogramScale::Log) { + resolution = resolution.next_power_of_two(); + } + + Histogram { + buckets: (0..self.num_buckets) + .map(|_| AtomicU64::new(0)) + .collect::>() + .into_boxed_slice(), + resolution, + scale: self.scale, + } + } +} + +impl Default for HistogramBuilder { + fn default() -> HistogramBuilder { + HistogramBuilder::new() + } +} + +#[cfg(test)] +mod test { + use super::*; + + macro_rules! assert_bucket_eq { + ($h:expr, $bucket:expr, $val:expr) => {{ + assert_eq!($h.buckets[$bucket], $val); + }}; + } + + #[test] + fn log_scale_resolution_1() { + let h = HistogramBuilder { + scale: HistogramScale::Log, + resolution: 1, + num_buckets: 10, + } + .build(); + + assert_eq!(h.bucket_range(0), 0..1); + assert_eq!(h.bucket_range(1), 1..2); + assert_eq!(h.bucket_range(2), 2..4); + assert_eq!(h.bucket_range(3), 4..8); + assert_eq!(h.bucket_range(9), 256..u64::MAX); + + let mut b = HistogramBatch::from_histogram(&h); + + b.measure(0, 1); + assert_bucket_eq!(b, 0, 1); + assert_bucket_eq!(b, 1, 0); + + b.measure(1, 1); + assert_bucket_eq!(b, 0, 1); + assert_bucket_eq!(b, 1, 1); + assert_bucket_eq!(b, 2, 0); + + b.measure(2, 1); + assert_bucket_eq!(b, 0, 1); + assert_bucket_eq!(b, 1, 1); + assert_bucket_eq!(b, 2, 1); + + b.measure(3, 1); + assert_bucket_eq!(b, 0, 1); + assert_bucket_eq!(b, 1, 1); + assert_bucket_eq!(b, 2, 2); + + b.measure(4, 1); + assert_bucket_eq!(b, 0, 1); + assert_bucket_eq!(b, 1, 1); + assert_bucket_eq!(b, 2, 2); + assert_bucket_eq!(b, 3, 1); + + b.measure(100, 1); + assert_bucket_eq!(b, 7, 1); + + b.measure(128, 1); + assert_bucket_eq!(b, 8, 1); + + b.measure(4096, 1); + assert_bucket_eq!(b, 9, 1); + } + + #[test] + fn log_scale_resolution_2() { + let h = HistogramBuilder { + scale: HistogramScale::Log, + resolution: 2, + num_buckets: 10, + } + .build(); + + assert_eq!(h.bucket_range(0), 0..2); + assert_eq!(h.bucket_range(1), 2..4); + assert_eq!(h.bucket_range(2), 4..8); + assert_eq!(h.bucket_range(3), 8..16); + assert_eq!(h.bucket_range(9), 512..u64::MAX); + + let mut b = HistogramBatch::from_histogram(&h); + + b.measure(0, 1); + assert_bucket_eq!(b, 0, 1); + assert_bucket_eq!(b, 1, 0); + + b.measure(1, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 0); + + b.measure(2, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 1); + assert_bucket_eq!(b, 2, 0); + + b.measure(3, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 2); + assert_bucket_eq!(b, 2, 0); + + b.measure(4, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 2); + assert_bucket_eq!(b, 2, 1); + + b.measure(5, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 2); + assert_bucket_eq!(b, 2, 2); + + b.measure(6, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 2); + assert_bucket_eq!(b, 2, 3); + + b.measure(7, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 2); + assert_bucket_eq!(b, 2, 4); + + b.measure(8, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 2); + assert_bucket_eq!(b, 2, 4); + assert_bucket_eq!(b, 3, 1); + + b.measure(100, 1); + assert_bucket_eq!(b, 6, 1); + + b.measure(128, 1); + assert_bucket_eq!(b, 7, 1); + + b.measure(4096, 1); + assert_bucket_eq!(b, 9, 1); + + for bucket in h.buckets.iter() { + assert_eq!(bucket.load(Relaxed), 0); + } + + b.submit(&h); + + for i in 0..h.buckets.len() { + assert_eq!(h.buckets[i].load(Relaxed), b.buckets[i]); + } + + b.submit(&h); + + for i in 0..h.buckets.len() { + assert_eq!(h.buckets[i].load(Relaxed), b.buckets[i]); + } + } + + #[test] + fn linear_scale_resolution_1() { + let h = HistogramBuilder { + scale: HistogramScale::Linear, + resolution: 1, + num_buckets: 10, + } + .build(); + + assert_eq!(h.bucket_range(0), 0..1); + assert_eq!(h.bucket_range(1), 1..2); + assert_eq!(h.bucket_range(2), 2..3); + assert_eq!(h.bucket_range(3), 3..4); + assert_eq!(h.bucket_range(9), 9..u64::MAX); + + let mut b = HistogramBatch::from_histogram(&h); + + b.measure(0, 1); + assert_bucket_eq!(b, 0, 1); + assert_bucket_eq!(b, 1, 0); + + b.measure(1, 1); + assert_bucket_eq!(b, 0, 1); + assert_bucket_eq!(b, 1, 1); + assert_bucket_eq!(b, 2, 0); + + b.measure(2, 1); + assert_bucket_eq!(b, 0, 1); + assert_bucket_eq!(b, 1, 1); + assert_bucket_eq!(b, 2, 1); + assert_bucket_eq!(b, 3, 0); + + b.measure(3, 1); + assert_bucket_eq!(b, 0, 1); + assert_bucket_eq!(b, 1, 1); + assert_bucket_eq!(b, 2, 1); + assert_bucket_eq!(b, 3, 1); + + b.measure(5, 1); + assert_bucket_eq!(b, 5, 1); + + b.measure(4096, 1); + assert_bucket_eq!(b, 9, 1); + + for bucket in h.buckets.iter() { + assert_eq!(bucket.load(Relaxed), 0); + } + + b.submit(&h); + + for i in 0..h.buckets.len() { + assert_eq!(h.buckets[i].load(Relaxed), b.buckets[i]); + } + + b.submit(&h); + + for i in 0..h.buckets.len() { + assert_eq!(h.buckets[i].load(Relaxed), b.buckets[i]); + } + } + + #[test] + fn linear_scale_resolution_100() { + let h = HistogramBuilder { + scale: HistogramScale::Linear, + resolution: 100, + num_buckets: 10, + } + .build(); + + assert_eq!(h.bucket_range(0), 0..100); + assert_eq!(h.bucket_range(1), 100..200); + assert_eq!(h.bucket_range(2), 200..300); + assert_eq!(h.bucket_range(3), 300..400); + assert_eq!(h.bucket_range(9), 900..u64::MAX); + + let mut b = HistogramBatch::from_histogram(&h); + + b.measure(0, 1); + assert_bucket_eq!(b, 0, 1); + assert_bucket_eq!(b, 1, 0); + + b.measure(50, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 0); + + b.measure(100, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 1); + assert_bucket_eq!(b, 2, 0); + + b.measure(101, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 2); + assert_bucket_eq!(b, 2, 0); + + b.measure(200, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 2); + assert_bucket_eq!(b, 2, 1); + + b.measure(299, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 2); + assert_bucket_eq!(b, 2, 2); + + b.measure(222, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 2); + assert_bucket_eq!(b, 2, 3); + + b.measure(300, 1); + assert_bucket_eq!(b, 0, 2); + assert_bucket_eq!(b, 1, 2); + assert_bucket_eq!(b, 2, 3); + assert_bucket_eq!(b, 3, 1); + + b.measure(888, 1); + assert_bucket_eq!(b, 8, 1); + + b.measure(4096, 1); + assert_bucket_eq!(b, 9, 1); + + for bucket in h.buckets.iter() { + assert_eq!(bucket.load(Relaxed), 0); + } + + b.submit(&h); + + for i in 0..h.buckets.len() { + assert_eq!(h.buckets[i].load(Relaxed), b.buckets[i]); + } + + b.submit(&h); + + for i in 0..h.buckets.len() { + assert_eq!(h.buckets[i].load(Relaxed), b.buckets[i]); + } + } + + #[test] + fn inc_by_more_than_one() { + let h = HistogramBuilder { + scale: HistogramScale::Linear, + resolution: 100, + num_buckets: 10, + } + .build(); + + let mut b = HistogramBatch::from_histogram(&h); + + b.measure(0, 3); + assert_bucket_eq!(b, 0, 3); + assert_bucket_eq!(b, 1, 0); + + b.measure(50, 5); + assert_bucket_eq!(b, 0, 8); + assert_bucket_eq!(b, 1, 0); + + b.measure(100, 2); + assert_bucket_eq!(b, 0, 8); + assert_bucket_eq!(b, 1, 2); + assert_bucket_eq!(b, 2, 0); + + b.measure(101, 19); + assert_bucket_eq!(b, 0, 8); + assert_bucket_eq!(b, 1, 21); + assert_bucket_eq!(b, 2, 0); + + for bucket in h.buckets.iter() { + assert_eq!(bucket.load(Relaxed), 0); + } + + b.submit(&h); + + for i in 0..h.buckets.len() { + assert_eq!(h.buckets[i].load(Relaxed), b.buckets[i]); + } + + b.submit(&h); + + for i in 0..h.buckets.len() { + assert_eq!(h.buckets[i].load(Relaxed), b.buckets[i]); + } + } +} diff --git a/tokio/src/runtime/metrics/mock.rs b/tokio/src/runtime/metrics/mock.rs index c388dc069..4d5e7b5b1 100644 --- a/tokio/src/runtime/metrics/mock.rs +++ b/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! { diff --git a/tokio/src/runtime/metrics/mod.rs b/tokio/src/runtime/metrics/mod.rs index 4b96f1b71..88be4a521 100644 --- a/tokio/src/runtime/metrics/mod.rs +++ b/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}; } diff --git a/tokio/src/runtime/metrics/runtime.rs b/tokio/src/runtime/metrics/runtime.rs index 014995c7e..1f990a1f8 100644 --- a/tokio/src/runtime/metrics/runtime.rs +++ b/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,196 @@ 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. + /// + /// # Panics + /// + /// The method panics if `bucket` represents an invalid bucket index, i.e. + /// is greater than or equal to `poll_count_histogram_num_buckets()`. + /// + /// # 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 + #[track_caller] + pub fn poll_count_histogram_bucket_range(&self, bucket: usize) -> Range { + 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 + #[track_caller] + 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`. /// diff --git a/tokio/src/runtime/metrics/worker.rs b/tokio/src/runtime/metrics/worker.rs index a40c76eff..e0f23e6a0 100644 --- a/tokio/src/runtime/metrics/worker.rs +++ b/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, } 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, } } diff --git a/tokio/src/runtime/mod.rs b/tokio/src/runtime/mod.rs index 4165143f5..a2efba479 100644 --- a/tokio/src/runtime/mod.rs +++ b/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 diff --git a/tokio/src/runtime/scheduler/current_thread.rs b/tokio/src/runtime/scheduler/current_thread.rs index 634abf3ee..becf9e244 100644 --- a/tokio/src/runtime/scheduler/current_thread.rs +++ b/tokio/src/runtime/scheduler/current_thread.rs @@ -109,6 +109,8 @@ impl CurrentThread { seed_generator: RngSeedGenerator, config: Config, ) -> (CurrentThread, Arc) { + let worker_metrics = WorkerMetrics::from_config(&config); + let handle = Arc::new(Handle { shared: Shared { queue: Mutex::new(Some(VecDeque::with_capacity(INITIAL_CAPACITY))), @@ -116,7 +118,7 @@ impl CurrentThread { 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(&self, mut core: Box, f: impl FnOnce() -> R) -> (Box, 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, diff --git a/tokio/src/runtime/scheduler/multi_thread/worker.rs b/tokio/src/runtime/scheduler/multi_thread/worker.rs index f83c6bc1e..c59f19e59 100644 --- a/tokio/src/runtime/scheduler/multi_thread/worker.rs +++ b/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(); diff --git a/tokio/src/runtime/tests/loom_queue.rs b/tokio/src/runtime/tests/loom_queue.rs index fc93bf3e4..4c3511cde 100644 --- a/tokio/src/runtime/tests/loom_queue.rs +++ b/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) -> 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); diff --git a/tokio/src/runtime/tests/queue.rs b/tokio/src/runtime/tests/queue.rs index 68d2e892e..ac80fa733 100644 --- a/tokio/src/runtime/tests/queue.rs +++ b/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; diff --git a/tokio/tests/rt_metrics.rs b/tokio/tests/rt_metrics.rs index 76aa0587f..22bd5b3d7 100644 --- a/tokio/tests/rt_metrics.rs +++ b/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) +}