Worker 差分与 poll 时间直方图
主线位置：运行时主线的中段。上一页列了字段和它们背后的 tokio 计数器；本页讲
RuntimeIntervals::next()怎么把这些累计值变成「本区间」的数字，以及 tokio 那边什么时候才把计数器写出来。
sequenceDiagram participant W as tokio worker participant B as MetricsBatch（线程私有） participant A as WorkerMetrics（原子变量） participant R as RuntimeIntervals W->>B: poll 任务：poll_count 加 1，偷到、溢出各自加 W->>B: 一批任务处理完：busy 时长加上，更新 EWMA W->>A: park 前、每 61 个 tick 的维护：submit，整批写出 R->>A: next()：逐个 worker 读累计值 R->>R: 减去上次读数，求和、取最大最小 R->>R: 按本区间 poll 次数给各 worker 的 EWMA 加权
intervals()：在这一刻定下基准
RuntimeMonitor::intervals 做两件事：记下当前时刻，给每个 worker 建一个 Worker 快照，快照里是此刻的累计值：
pub fn intervals(&self) -> RuntimeIntervals {
let started_at = Instant::now();
let workers = (0..self.runtime.num_workers())
.map(|worker| Worker::new(worker, &self.runtime))
.collect();
RuntimeIntervals {
runtime: self.runtime.clone(),
started_at,
workers,
#[cfg(tokio_unstable)]
num_remote_schedules: self.runtime.remote_schedule_count(),
// ... 预算强制让出、IO 就绪数、直方图的桶范围
}
}
所以第一次 next() 返回的已经是「从调用 intervals() 到现在」的差，而不是从运行时创建起的累计值。任务那边的 TaskIntervals 第一次返回累计值，两边不一样，下一部分的 intervals：迭代器差分 会把两者放在一起比。worker 的个数也在这一刻定下：tokio 的 worker 数在运行时创建后不变，所以这不会出问题。
probe：每次 next() 做什么
RuntimeIntervals::probe 先填瞬时值和墙钟时间，再把最小值初始化成一个很大的数，交给每个 worker 去更新：
fn probe(&mut self) -> RuntimeMetrics {
let now = Instant::now();
let mut metrics = RuntimeMetrics {
workers_count: self.runtime.num_workers(),
live_tasks_count: self.runtime.num_alive_tasks(),
elapsed: now.saturating_duration_since(self.started_at),
global_queue_depth: self.runtime.global_queue_depth(),
min_park_count: u64::MAX,
min_busy_duration: Duration::from_secs(1000000000),
..Default::default()
};
#[cfg(tokio_unstable)]
{
// 运行时级的累计值：remote_schedule_count 等三个，saturating_sub 做差
// 其余最小值字段也设成 u64::MAX / usize::MAX / Duration::MAX
// ...
}
self.started_at = now;
for worker in &mut self.workers {
worker.probe(&self.runtime, &mut metrics);
}
// ...
metrics
}
elapsed 是本库自己记的墙钟时间，上次 probe 到这次 probe；它是 busy_ratio 的分母。最小忙碌时长初始化成 10 亿秒，第一个 worker 一定会把它压下去。
Worker::probe 里的 metric! 宏是差分的全部逻辑：
macro_rules! metric {
( $sum:ident, $max:ident, $min:ident, $probe:ident ) => {{
let val = rt.$probe(self.worker);
let delta = val - self.$sum;
self.$sum = val;
metrics.$sum += delta;
if delta > metrics.$max {
metrics.$max = delta;
}
if delta < metrics.$min {
metrics.$min = delta;
}
}};
}
metric!(
total_park_count,
max_park_count,
min_park_count,
worker_park_count
);
metric!(
total_busy_duration,
max_busy_duration,
min_busy_duration,
worker_total_busy_duration
);
读 tokio 的累计值，减去这个 worker 上次的读数，得到本区间增量；增量加进总和，再更新最大最小。max_* 和 min_* 是「增量最大的那个 worker 的增量」，不是某个时刻的峰值。
这里的减法是普通的 -。0.4.8 的 #114（「use saturating_sub to prevent overflow panics in runtime metrics」）把运行时级的几个计数、elapsed、直方图的减法都换成了 saturating_sub，metric! 宏里这一行没换。tokio 的每个 worker 计数器只增不减，正常情况下减法不会出负数；这个宏依赖的就是这一点。
busy_ratio：0 到 worker 数
派生指标 busy_ratio 只有一行：
pub fn busy_ratio(&self) -> f64 {
self.total_busy_duration.as_nanos() as f64 / self.elapsed.as_nanos() as f64
}
分子是所有 worker 的忙碌时长之和，分母是一段墙钟时间，所以取值范围是 0 到 worker 数，不是 0 到 1。4 个 worker 各跑一个忙 300ms 的任务：
#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main() {
let rm = RuntimeMonitor::new(&tokio::runtime::Handle::current());
let mut ri = rm.intervals();
ri.next();
let hs: Vec<_> = (0..4).map(|_| tokio::spawn(async { spin(Duration::from_millis(300)) })).collect();
for h in hs { h.await.unwrap(); }
let m = ri.next().unwrap();
println!("workers={} busy_ratio={:.2} total_busy={:?} elapsed={:?}", m.workers_count, m.busy_ratio(), m.total_busy_duration, m.elapsed);
}
M2 Max 笔记本、--release，连跑 3 次：
workers=4 busy_ratio=4.00 total_busy=1.200039252s elapsed=300.073042ms
workers=4 busy_ratio=4.00 total_busy=1.200038709s elapsed=300.187375ms
workers=4 busy_ratio=4.00 total_busy=1.200072292s elapsed=300.06725ms
要百分比，自己再除以 workers_count。
tokio 什么时候才把计数写出来
上面那个实验里 busy_ratio 正好是 4.00，是因为取样前 4 个任务都已结束、worker 都已 park。换一种取法就不是这样了。
worker 的计数先记在线程私有的 MetricsBatch 里，忙碌时长在一批任务处理完时才加上；整批写进原子变量的 submit 只在两个地方调：worker park 之前，和每 event_interval（默认 61）个 tick 一次的维护（worker 主循环见 tokio 卷 worker 主循环）。
一个 poll 跑了 2.5 秒，这段时间 worker 既不 park，也走不到下一个 tick，计数器就一直停在原地。实验：2 个 worker，一个任务在一次 poll 里忙 2.5 秒，主线程在运行时外面每 500ms 取一次样：
fn multi(n: usize) -> tokio::runtime::Runtime {
tokio::runtime::Builder::new_multi_thread().worker_threads(n).enable_all().build().unwrap()
}
// ...
let rt = multi(2);
let monitor = RuntimeMonitor::new(rt.handle());
let mut iv = monitor.intervals();
rt.spawn(async { spin(Duration::from_millis(2500)) });
for i in 0..7 {
std::thread::sleep(Duration::from_millis(500));
let m = iv.next().unwrap();
println!(
"t={}ms busy_total={:?} busy_max={:?} busy_min={:?} busy_ratio={:.2} parks={}",
(i + 1) * 500, m.total_busy_duration, m.max_busy_duration, m.min_busy_duration, m.busy_ratio(), m.total_park_count
);
}
其中一次的输出：
t=500ms busy_total=125ns busy_max=125ns busy_min=0ns busy_ratio=0.00 parks=1
t=1000ms busy_total=0ns busy_max=0ns busy_min=0ns busy_ratio=0.00 parks=0
t=1500ms busy_total=0ns busy_max=0ns busy_min=0ns busy_ratio=0.00 parks=0
t=2000ms busy_total=0ns busy_max=0ns busy_min=0ns busy_ratio=0.00 parks=0
t=2500ms busy_total=2.500003792s busy_max=2.500003792s busy_min=0ns busy_ratio=4.90 parks=1
t=3000ms busy_total=0ns busy_max=0ns busy_min=0ns busy_ratio=0.00 parks=0
t=3500ms busy_total=0ns busy_max=0ns busy_min=0ns busy_ratio=0.00 parks=0
另外跑了 3 次，形状完全相同：前 2 秒的四个区间里忙碌时长都是 0（或一百多纳秒），2.5 秒的那个区间一次性记进 2.5 秒，busy_ratio 为 4.90–4.95。2 个 worker 的运行时，busy_ratio 超过了 worker 数。
这说明两件事：
- 区间指标反映的是「这个区间里提交了什么」，不是「这个区间里发生了什么」。 一个 worker 被一次长 poll 卡住，正是最想看到的时候，它的忙碌时长却是 0，等它跑完才一次补上。这和任务指标不同：
Instrumented在每次 poll 结束时直接写原子变量，没有批次。 - 取样周期越短，这种错位越明显。tokio-metrics 自己的文档也碰到了这个问题：
total_busy_duration、num_remote_schedules等字段的文档示例在取第二个区间之前都先drop(runtime)，注释写的是// flush metrics（第 414 行）。
要在看板上发现「某个 worker 被卡住」，与其等 busy_ratio，不如看任务指标里的慢 poll，或者 park 次数长时间为 0 的 worker。后者 tokio-metrics 只给了全体的总和与最大最小，没有逐个 worker 的值，需要的话得直接读 tokio 的 worker_park_count。
mean_poll_duration：按 poll 次数加权的 EWMA
mean_poll_duration 要 tokio_unstable。名字像是「本区间 poll 时长的均值」，但 Worker::probe 读的是 tokio 的 worker_mean_poll_time：
// Update the mean task poll duration if there were polls
if worker_polls_count > 0 {
let val = rt.worker_mean_poll_time(self.worker);
if val > metrics.mean_poll_duration_worker_max {
metrics.mean_poll_duration_worker_max = val;
}
if val < metrics.mean_poll_duration_worker_min {
metrics.mean_poll_duration_worker_min = val;
}
// First, scale the current value down
let ratio = total_polls_count as f64 / metrics.total_polls_count as f64;
let mut mean = metrics.mean_poll_duration.as_nanos() as f64 * ratio;
// Add the scaled current worker's mean poll duration
let ratio = worker_polls_count as f64 / metrics.total_polls_count as f64;
mean += val.as_nanos() as f64 * ratio;
metrics.mean_poll_duration = Duration::from_nanos(mean as u64);
}
这段是在线算加权平均：每加进一个 worker，把已有的均值按「之前的 poll 数 ÷ 现在的 poll 数」缩小，再加上这个 worker 的值乘以它的权重。权重是本区间的 poll 次数，没问题；问题在被加权的那个值。tokio 的 worker_mean_poll_time 来自多线程调度器的 Stats：
pub(crate) fn end_processing_scheduled_tasks(&mut self) {
self.batch.end_processing_scheduled_tasks();
// Update the EWMA task poll time
if self.tasks_polled_in_batch > 0 {
let now = Instant::now();
// ...
let elapsed = (now - self.processing_scheduled_tasks_started_at).as_nanos() as f64;
let num_polls = self.tasks_polled_in_batch as f64;
// Calculate the mean poll duration for a single task in the batch
let mean_poll_duration = elapsed / num_polls;
// Compute the alpha weighted by the number of tasks polled this batch.
let weighted_alpha = 1.0 - (1.0 - TASK_POLL_TIME_EWMA_ALPHA).powf(num_polls);
// Now compute the new weighted average task poll time.
self.task_poll_time_ewma = weighted_alpha * mean_poll_duration
+ (1.0 - weighted_alpha) * self.task_poll_time_ewma;
}
}
它是指数加权移动平均（EWMA），TASK_POLL_TIME_EWMA_ALPHA 是 0.1，源码注释说这个值「plucked from thin air」；初值是 200µs ÷ 61 ≈ 3.3µs。tokio 算它是为了自动调节多久看一次全局队列（见 tokio 卷全局注入队列），不是为了监控。由此有三个后果：
- 它记的是最近的 poll，越早的 poll 权重越小，和区间边界无关。
- 它的样本是「一批任务的总耗时 ÷ 任务数」，批内任务之间调度器自己的开销也算进去了。
- 只有多线程调度器算它。current_thread 调度器提交计数时直接传 0：
fn submit_metrics(&mut self, handle: &Handle) {
self.metrics.submit(&handle.shared.worker_metrics, 0);
}
两个实验，都用 RUSTFLAGS="--cfg tokio_unstable"。第一个：1 个 worker，一个区间里跑 1 个忙 20ms 的任务和 1000 个忙 10µs 的任务，只换先后顺序，同时用一个 TaskMonitor 量真实的均值：
for slow_first in [true, false] {
let rt = multi(1);
let monitor = RuntimeMonitor::new(rt.handle());
let tm = TaskMonitor::new();
let mut iv = monitor.intervals();
rt.block_on(async {
let slow = || tokio::spawn(tm.instrument(async { spin(Duration::from_millis(20)) }));
if slow_first {
slow().await.unwrap();
}
for _ in 0..1000 {
tokio::spawn(tm.instrument(async { spin(Duration::from_micros(10)) })).await.unwrap();
}
if !slow_first {
slow().await.unwrap();
}
});
let m = iv.next().unwrap();
let t = tm.cumulative();
println!(
"slow_first={} runtime: polls={} mean_poll_duration={:?} | task monitor: polls={} mean_poll_duration={:?}",
slow_first, m.total_polls_count, m.mean_poll_duration, t.total_poll_count, t.mean_poll_duration()
);
}
第二个：current_thread 运行时上跑 100 个忙 200µs 的任务：
let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
let monitor = RuntimeMonitor::new(rt.handle());
let mut iv = monitor.intervals();
rt.block_on(async {
for _ in 0..100 {
tokio::spawn(async { spin(Duration::from_micros(200)) }).await.unwrap();
}
});
let m = iv.next().unwrap();
println!(
"current_thread: workers={} polls={} mean_poll_duration={:?} worker_min/max={:?}/{:?} busy={:?}",
m.workers_count, m.total_polls_count, m.mean_poll_duration, m.mean_poll_duration_worker_min,
m.mean_poll_duration_worker_max, m.total_busy_duration
);
各跑 3 次。第一个实验的一次：
slow_first=true runtime: polls=1001 mean_poll_duration=11.216µs | task monitor: polls=1001 mean_poll_duration=30.091µs
slow_first=false runtime: polls=1001 mean_poll_duration=2.010693ms | task monitor: polls=1001 mean_poll_duration=30.084µs
另两次 slow_first=true 是 11.4µs、11.5µs，slow_first=false 都是 2.0106–2.0109ms；TaskMonitor 的均值三次都在 30.08–30.13µs。第二个实验三次的输出只有 busy 在 20.05–20.06ms 之间变化：
current_thread: workers=1 polls=100 mean_poll_duration=0ns worker_min/max=0ns/0ns busy=20.059873ms
- 同样的 1001 次 poll，真实均值约 30µs（(20ms + 1000 × 10µs) ÷ 1001）。慢任务在前，EWMA 被后面 1000 次快 poll 冲淡到 11µs；慢任务在最后，这一次就占了 10% 的权重：0.1 × 20ms + 0.9 × 约 11µs ≈ 2.01ms，和输出吻合（每个任务单独成一批，
weighted_alpha正好是 0.1）。 - current_thread 上有 100 次 poll、20ms 的忙碌，
mean_poll_duration和两个 worker 极值都是 0。tokio 文档对worker_mean_poll_time的说明里有一句「is only provided by the multi-threaded runtime」，tokio-metrics 的字段文档没有提。
要按区间统计 poll 时长，用下一节的直方图，或者任务指标的 mean_poll_duration()；运行时的 mean_poll_duration 只能当「最近的 poll 大概多长」的趋势看。
poll 时间直方图
poll_time_histogram 也要 tokio_unstable，而且要在建运行时时打开：Builder::enable_metrics_poll_time_histogram 本身也在 cfg_unstable_metrics! 里。默认配置在 HistogramBuilder::new：线性 10 个桶，每个 100µs 宽，最后一个桶从 900µs 一直到 u64::MAX 纳秒。
tokio-metrics 在 intervals() 时把桶的范围读一次存起来，每次 probe 用它造一个空直方图，再由每个 worker 把自己每个桶的增量加进去（第 1570 行）：
// Update the histogram counts if there were polls since last count
if worker_polls_count > 0 {
for (bucket, entry) in metrics.poll_time_histogram.buckets_mut().iter_mut().enumerate() {
let new = rt.poll_time_histogram_bucket_count(self.worker, bucket);
let delta = new.saturating_sub(self.poll_time_histogram[bucket]);
self.poll_time_histogram[bucket] = new;
entry.add_count(delta);
}
}
没在 Builder 上打开时，tokio 的 poll_time_histogram_num_buckets 返回 0，直方图就是空的，不报错。
PollTimeHistogram 是一个 Vec<HistogramBucket>，每个桶带 range_start、range_end、count。0.5.0 之前这个字段是 Vec<u64>，只有计数，桶的范围要自己去问 tokio；0.5.0（#121）改成了带范围的类型，是一次破坏性改动。
实验：2 个 worker，分别跑 10 个忙 50µs、250µs、2ms 的任务，比较打开和不打开直方图：
for enable in [false, true] {
let mut b = tokio::runtime::Builder::new_multi_thread();
b.worker_threads(2).enable_all();
if enable {
b.enable_metrics_poll_time_histogram();
}
let rt = b.build().unwrap();
let monitor = RuntimeMonitor::new(rt.handle());
let mut iv = monitor.intervals();
rt.block_on(async {
for us in [50u64, 250, 2000] {
for _ in 0..10 {
tokio::spawn(async move { spin(Duration::from_micros(us)) }).await.unwrap();
}
}
});
let m = iv.next().unwrap();
let h = &m.poll_time_histogram;
let parts: Vec<String> = h
.buckets()
.iter()
.map(|b| format!("{}..{}:{}", b.range_start().as_micros(), if b.range_end() == Duration::from_nanos(u64::MAX) { "MAX".to_string() } else { b.range_end().as_micros().to_string() }, b.count()))
.collect();
println!("enabled={} polls={} buckets={} [{}]", enable, m.total_polls_count, h.buckets().len(), parts.join(" "));
}
连跑 2 次，输出相同：
enabled=false polls=30 buckets=0 []
enabled=true polls=30 buckets=10 [0..100:10 100..200:0 200..300:10 300..400:0 400..500:0 500..600:0 600..700:0 700..800:0 800..900:0 900..MAX:10]
不打开时 poll 次数照样有（它只要 cfg），桶数是 0；打开后 30 次 poll 按时长落进三个桶。默认的线性桶对 1ms 以上的 poll 一概归进最后一个桶，2ms 和 2s 分不出来。要看长尾，用 metrics_poll_time_histogram_configuration 换成对数桶（tokio 文档推荐的 LogHistogram），tokio-metrics 原样照搬 tokio 给的桶范围，不需要改。
tokio 1.54.1 还有一个排队延迟直方图（enable_metrics_schedule_latency_histogram），除了 tokio_unstable 还要 tokio 的 schedule-latency feature。它量的是任务从被调度到被 poll 的时间，正是任务指标里「排队」的运行时版本，而且不受嵌套的 Instrumented 的限制。tokio-metrics 0.5.2 没有接它。
小结
intervals()在调用那一刻记下每个 worker 的累计值，第一次next()已经是差分；probe逐个 worker 读累计值、减上次、求和取最大最小，metric!宏里的减法是普通的-。busy_ratio= 各 worker 忙碌时长之和 ÷ 墙钟时间，范围 0 到 worker 数；4 个 worker 各忙 300ms 是 4.00。- tokio 只在 worker park 前和每 61 个 tick 的维护里提交计数。一次 poll 忙 2.5 秒，前 2 秒的区间里忙碌时长是 0，最后一个区间一次补上，2 个 worker 的
busy_ratio到了 4.9。 mean_poll_duration是 tokio 为调节全局队列间隔算的 EWMA（α = 0.1），按各 worker 本区间 poll 次数加权；同一组 poll 换个顺序，11µs 和 2ms 都可能，真实均值是 30µs；current_thread 上恒为 0。- poll 时间直方图要
tokio_unstable加Builder::enable_metrics_poll_time_histogram()，不打开时桶数为 0；默认 10 个 100µs 的线性桶，900µs 以上全归最后一个桶。