tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-21 00:00:10 +02:00
Code Issues Packages Projects Releases Wiki Activity

rt: instrument task poll times with a histogram (#5685)

Browse Source 
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.
This commit is contained in:
Carl Lerche
2023-05-15 15:20:41 -07:00
committed by GitHub
parent a883fd4378
commit c84d0a14b1
14 changed files with 1105 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio/src/runtime/builder.rs
+148 -1
View File
@@ -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<Runtime> {
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<HistogramBuilder> {
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(),
},
);
tokio/src/runtime/config.rs
+3
View File
@@ -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
+51 -10
View File
@@ -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/histogram.rs
+502
View File
@@ -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<u64> {
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::<Vec<_>>()
.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]);
}
}
}
tokio/src/runtime/metrics/mock.rs
+12 -2
View File
@@ -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
+6 -1
View File
@@ -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
+191
View File
@@ -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<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
#[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`.
///
tokio/src/runtime/metrics/worker.rs
+15
View File
@@ -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
+3 -3
View File
@@ -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
+8 -4
View File
@@ -109,6 +109,8 @@ 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))),
@@ -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<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
+12 -4
View File
@@ -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
+12 -8
View File
@@ -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
+12 -7
View File
@@ -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
+130
View File
@@ -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)
}