Rust in Everything
基于 tokio-metrics 0.5.2

intervals：迭代器差分

主线位置：两条主线的汇合点。任务那边是 RawMetrics 里只增不减的原子变量，运行时那边是 tokio 只增不减的计数器；intervals() 返回的迭代器把它们切成一段一段的区间。本页讲任务这边的 TaskIntervals，再和运行时的 RuntimeIntervals 对照。

sequenceDiagram
  participant U as 调用方
  participant I as TaskIntervals
  participant R as RawMetrics
  U->>I: next()（第一次）
  I->>R: metrics()：逐个 load
  I->>R: swap(0) 取出区间最大空闲
  I-->>U: 返回累计值，previous = 本次读数
  U->>U: sleep 一个采样周期
  U->>I: next()
  I->>R: metrics()，swap(0)
  I->>I: 与 previous 逐字段 wrapping_sub
  I-->>U: 返回差值，previous = 本次读数

TaskIntervals：记住上一次

TaskIntervals 只有两个字段：

#[derive(Debug)]
pub struct TaskIntervals<M: AsRef<TaskMonitorCore> + Send + Sync + 'static = TaskMonitor> {
    monitor: M,
    previous: Option<TaskMetrics>,
}

monitor 可以是 TaskMonitor，也可以是 &'static TaskMonitorCore（见 TaskMonitorCore 与 FutureMonitor）。previous 一开始是 None。它实现的是 Iterator 而不是 Stream，next() 永远返回 Some(self.probe())；多久调一次由调用方自己决定，库里没有定时器。

probe 的骨架：

    fn probe(&mut self) -> TaskMetrics {
        let latest = self.monitor.as_ref().metrics.metrics();
        let local_max_idle_duration = self
            .monitor
            .as_ref()
            .metrics
            .get_and_reset_local_max_idle_duration();

        let next = if let Some(previous) = self.previous {
            TaskMetrics {
                instrumented_count: latest
                    .instrumented_count
                    .wrapping_sub(previous.instrumented_count),
                // ... 其余计数同样 wrapping_sub
                total_poll_duration: sub(latest.total_poll_duration, previous.total_poll_duration),
                // ... 其余时长同样 sub
                max_idle_duration: local_max_idle_duration,
                // ...
            }
        } else {
            latest
        };

        self.previous = Some(latest);

        next
    }

三点：

  1. 第一次返回累计值。 previous 是 None 时直接返回 latest，也就是从 monitor 创建到现在的全部。想让第一个区间从「现在」开始，就先空调一次 next()，怎么读这一卷的程序就是这么做的。
  2. 差分用回绕减法。 计数是 u64::wrapping_sub；时长先用 to_nanos 转成纳秒的 u64（乘法和加法也是 wrapping_*），再 wrapping_sub，见 sub。u64 纳秒能表示 584 年，累计值回绕在实践中不会发生；用回绕运算的效果是任何情况下都不 panic，哪怕读到的两个值顺序颠倒，也只是得到一个很大的数。运行时那边不一样：worker 计数用普通减法，运行时级的计数用 saturating_sub（上一部分的 Worker 差分）。
  3. 最大空闲不做差。 最大值没法从两个累计最大值相减得到，所以它走另一条路：直接取一个「区间内」的原子变量，读完清零。

区间最大空闲：swap(0)

任务主线里，每记一次空闲，instrument_poll 就对两个原子变量各做一次 fetch_max（第 2946 行）：

        metrics
            .local_max_idle_duration_ns
            .fetch_max(idle_ns, SeqCst);
        metrics
            .global_max_idle_duration_ns
            .fetch_max(idle_ns, SeqCst);

cumulative() 读 global_*；TaskIntervals::probe 读 local_*，用的是 get_and_reset_local_max_idle_duration：

    fn get_and_reset_local_max_idle_duration(&self) -> Duration {
        Duration::from_nanos(self.local_max_idle_duration_ns.swap(0, SeqCst))
    }

swap(0) 把值取出来的同时清零，下一个区间从 0 开始取最大值。问题在于 local_max_idle_duration_ns 在 RawMetrics 里，一个 monitor 只有一份，而 TaskIntervals 可以开任意多个。其他字段每个迭代器各自记 previous，互不干扰；这一个字段，谁先读谁拿走。

另外，第一次 probe 返回的是 latest，里面的 max_idle_duration 来自 global_*；但 swap(0) 照样执行了，local_* 同样被清零。

实验：同一个 monitor 开 a、b 两个迭代器（比如两个看板），每轮让一个被包装的 sleep 空闲一段，然后两个迭代器各取一次；最后再晚开一个 c：

use std::time::Duration;
use tokio::time::sleep;
use tokio_metrics::TaskMonitor;

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let m = TaskMonitor::new();
  let mut a = m.intervals(); // 看板 A
  let mut b = m.intervals(); // 看板 B，同一个 monitor

  // 第 1 轮：一段约 100ms 的空闲
  m.instrument(sleep(Duration::from_millis(100))).await;
  let (x, y) = (a.next().unwrap(), b.next().unwrap());
  println!("round1 a: polls={} max_idle={:?} | b: polls={} max_idle={:?}", x.total_poll_count, x.max_idle_duration, y.total_poll_count, y.max_idle_duration);

  // 第 2 轮：一段约 60ms 的空闲
  m.instrument(sleep(Duration::from_millis(60))).await;
  let (x, y) = (a.next().unwrap(), b.next().unwrap());
  println!("round2 a: polls={} max_idle={:?} | b: polls={} max_idle={:?}", x.total_poll_count, x.max_idle_duration, y.total_poll_count, y.max_idle_duration);

  // 第 3 轮：什么都不做
  let (x, y) = (a.next().unwrap(), b.next().unwrap());
  println!("round3 a: polls={} max_idle={:?} slow_poll_ratio={} | b: polls={} max_idle={:?}", x.total_poll_count, x.max_idle_duration, x.slow_poll_ratio(), y.total_poll_count, y.max_idle_duration);

  // 晚开的迭代器：第一次 next() 是从 monitor 创建起的累计值
  let mut c = m.intervals();
  let z = c.next().unwrap();
  let z2 = c.next().unwrap();
  println!("late c first: polls={} max_idle={:?}; second: polls={} | cumulative max_idle={:?}", z.total_poll_count, z.max_idle_duration, z2.total_poll_count, m.cumulative().max_idle_duration);
}

依赖同前（tokio-metrics = "=0.5.2"、tokio = { version = "=1.54.1", features = ["full"] }）。M2 Max 笔记本、--release，连跑 4 次，计数和「谁是 0」每次都一样，时长在零点几毫秒内变化。其中一次：

round1 a: polls=2 max_idle=103.019333ms | b: polls=2 max_idle=103.019333ms
round2 a: polls=2 max_idle=62.638916ms | b: polls=2 max_idle=0ns
round3 a: polls=0 max_idle=0ns slow_poll_ratio=NaN | b: polls=0 max_idle=0ns
late c first: polls=4 max_idle=103.019333ms; second: polls=0 | cumulative max_idle=103.019333ms
  • 第 1 轮两边相同：都是第一次 next()，返回累计值，max_idle_duration 来自全局变量。
  • 第 2 轮 b 是 0：a 先 swap(0) 拿走了 62.6ms，b 读到的是清零后的值。polls=2 两边都对，因为普通字段各自做差。哪个看板先取样，哪个就有最大空闲；两个看板的取样周期不同步时，这个值会在两边随机地出现和消失。
  • 第 3 轮：没有任何 poll，slow_poll_ratio 是 0/0 = NaN（TaskMetrics 与派生指标）。
  • 晚开的 c：第一次 next() 把 monitor 创建以来的 4 次 poll 全部返回，max_idle_duration 是全局最大的 103ms，而不是「c 创建以来」的。

每个空闲比 sleep 的时长多出两三毫秒，原因和怎么读这一卷里的实验一样：时间轮的精度是 1ms，woke_at 记的是 wake() 被调用的时刻。

一个 monitor 需要喂给多个消费者时，只开一个 intervals()，把每次得到的 TaskMetrics（它是 Copy）分发出去；或者每个消费者一个 monitor。

RuntimeIntervals：从调用时起算

运行时那边没有 previous: Option。RuntimeMonitor::intervals 在调用时就给每个 worker 记下当前累计值（上一部分的 Worker 差分），所以第一次 next() 已经是差分。两者对照：

TaskIntervalsRuntimeIntervals
基准在什么时候定第一次 next()调用 intervals() 时
第一次 next() 返回从 monitor 创建起的累计值从 intervals() 起的差分
计数做差wrapping_subworker 计数普通 -，运行时级 saturating_sub
区间的长度没有字段，调用方自己记elapsed
多个迭代器互相抢 max_idle_duration互不干扰，每个迭代器各有一份 worker 快照
有没有 cumulative()有没有，只能读 tokio 的 Handle::metrics()

RuntimeIntervals 没有任何「读完清零」的状态，开几个都可以。TaskMetrics 没有 elapsed，算速率（每秒 poll 次数）要自己记取样时刻。

cumulative() 还是 intervals()

TaskMonitor::cumulative 直接返回 RawMetrics::metrics() 的结果，不改任何状态，随便调多少次。两者的区别只在「差分由谁来做」：

  • 导出到 Prometheus 这类自己会算速率的系统：计数和时长本来就适合当单调递增的 counter，用 cumulative() 原样导出，让 rate() 去做差，不需要本库的差分，也就没有「两个迭代器抢最大值」的问题。代价是 max_idle_duration 只有全局最大值，看不到「最近一分钟的最大空闲」。
  • 打日志、做单机判断（比如「这 10 秒的慢 poll 比例超过 20% 就告警」）：用 intervals()，每个区间的数字直接可读。只开一个迭代器。

本库自带的 metrics-rs 集成走的是第二条路：每个区间调一次 next()，把差值交给 metrics-rs。它怎么把差值映射成 counter 和 gauge，放在下一页 metrics-rs 与 metrique。

小结

  • TaskIntervals 记一个 previous: Option<TaskMetrics>：第一次 next() 返回累计值，之后逐字段与上次相减；计数和时长都用回绕运算，不会 panic。
  • 区间最大空闲不能做差，靠 local_max_idle_duration_ns.swap(0) 读完清零；它一个 monitor 只有一份，同一个 monitor 开两个 intervals()，后取样的一方读到 0。
  • RuntimeIntervals 在调用 intervals() 时记下基准，第一次 next() 已经是差分，带 elapsed，开几个都互不影响。
  • 导出给会算速率的系统，用 cumulative() 让对方做差；要直接可读的区间数字，用 intervals()，一个 monitor 只开一个。

💬 关联讨论

发起讨论
加载中...