From b2ba34ef04dd2727218bd4e35e9f2be4b4e5a38e Mon Sep 17 00:00:00 2001 From: dailz Date: Mon, 22 Jun 2026 18:03:56 +0800 Subject: [PATCH] =?UTF-8?q?docs(stats):=20=E4=B8=AD=E6=96=87=E6=B3=A8?= =?UTF-8?q?=E9=87=8A=E7=AE=A1=E9=81=93=E6=80=A7=E8=83=BD=E7=BB=9F=E8=AE=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/stats.rs | 265 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 265 insertions(+) diff --git a/src/stats.rs b/src/stats.rs index 2c6bfb6..15ba72e 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -1,3 +1,25 @@ +//! 管道性能统计模块 —— 用于卡顿诊断的轻量级滑动窗口统计。 +//! +//! 本模块跟踪 capture / encode / send 三段流水线的每秒指标快照： +//! - **FPS**：捕获帧率、编码帧率、发送帧率 +//! - **延迟分布**：各阶段（DMA-BUF import / VAAPI scale / GPU→CPU transfer / +//! sws_scale / H.264 encode）的 avg / p95 / max（微秒→毫秒） +//! - **队列深度**：capture 队列与 encoded 队列的瞬时观测值 +//! - **丢帧计数**：PipeWire 丢弃、重复帧去重跳过、超预算帧 +//! +//! 设计目标为低开销：仅收集计数器和时间样本，每秒输出一行结构化日志 +//! （仅在 `--stats` 启用时）。所有统计在主线程独占持有 `&mut PipelineStats`， +//! 跨线程数据（如 PipeWire dropped 计数、encode 线程的 duplicate 计数） +//! 通过外部 `AtomicU64` 在调用方读取后再传入本结构（见 `set_*` 系列）。 +//! +//! ## 与 Go 类比 +//! +//! - [`Instant::now()`] ≈ Go `time.Now()`，但精度更高（通常单调时钟） +//! - [`Duration::as_secs_f64`] ≈ Go `time.Duration.Seconds()`，但保留 f64 +//! - `&mut self` ≈ Go 中显式持有 `sync.Mutex` 的写锁；本模块的字段独占模型 +//! 天然无需 `Mutex`（外部跨线程读取后再以 `&mut self` 传入） +//! - `Vec` 样本缓冲 ≈ Go 中 `[]float64`，每窗口 `clear()` 复用容量 + // stats.rs — Lightweight windowed pipeline statistics for stutter diagnosis // // Tracks per-second snapshots of capture/encode/send pipeline metrics. @@ -6,6 +28,15 @@ use std::time::Instant; +/// 单帧流水线各阶段的耗时样本（中文概要，详细说明见下方英文文档）。 +/// +/// # 派生宏说明 +/// +/// - `#[derive(Debug)]`：便于 `dbg!()` 调试输出，类比 Go 的 `%+v` 格式化 +/// - `#[derive(Default)]`：所有字段为 `u64`/`usize` 零值时构造默认实例， +/// 测试中可用 `FrameTimings { total_us: 5000, ..Default::default() }` +/// 仅指定关注字段（见 `record_and_snapshot_counts` 测试） +/// /// Per-stage timing for a single encode pipeline frame. /// /// All values are in microseconds. The caller records timestamps around @@ -28,6 +59,27 @@ pub struct FrameTimings { pub output_bytes: usize, } +/// 一秒窗口内的管道统计聚合器（中文概要，详细说明见下方英文文档）。 +/// +/// 本结构通过 `&mut self` 接口收集三类原始数据：计数器（`capture_frames` 等）、 +/// 样本缓冲（`Vec`/`Vec`，窗口结束时 `clear()` 复用容量）、 +/// 时间锚点（`Option`，`None` 表示尚未观测过）。 +/// +/// # 所有权与并发模型 +/// +/// - 本结构**非 `Sync`**：`Vec` 字段无锁，跨线程读写需外部同步 +/// - 主线程独占持有 `&mut self`；跨线程数据通过外部 `AtomicU64` 在调用方 +/// 读取后以 `set_*` 接口注入 +/// - 这与 Go 中 `sync.Mutex` 不同：Rust 借用检查器在编译期 +/// 保证单一可变借用，无需运行时锁 +/// +/// # 与 Go 类比 +/// +/// - `Option` ≈ Go `*time.Time`（`nil` 表示未设置），但 Rust 用枚举 +/// 强制调用方处理"未设置"分支，避免 nil-pointer panic +/// - `Instant` 内部使用单调时钟，不受系统时间跳变影响；Go 1.9+ 的 +/// `time.Since()` 也使用单调时钟，行为一致 +/// /// Windowed statistics aggregator for the encode/send pipeline. /// /// Collects counters and timing samples within a one-second window, @@ -74,6 +126,11 @@ pub struct PipelineStats { } impl PipelineStats { + /// 构造一个空的统计聚合器（类比 Go 的 `NewXxx()` 工厂函数）。 + /// + /// `window_start` 初始化为当前时刻，确保 `should_snapshot()` 至少 + /// 在 1 秒后才返回 true（首窗口可能短于 1 秒有效数据，但 elapsed_secs + /// 是真实窗口长度，FPS 计算依然准确）。 pub fn new() -> Self { Self { capture_frames: 0, @@ -104,6 +161,21 @@ impl PipelineStats { } } + /// 记录一次来自 PipeWire 的捕获帧到达事件（中文 L3 解析见此）。 + /// + /// # 时间间隔（gap）计算 + /// + /// - [`Instant::now()`]：获取当前单调时刻（≈ Go `time.Now()`，但精度更高） + /// - `last.elapsed()`：返回 `Duration`，类比 Go `time.Since(last)` + /// - [`Duration::as_secs_f64`]：将 `Duration` 转为秒（f64），类比 Go + /// `dur.Seconds()`；此处乘以 1000.0 转毫秒，便于日志可读 + /// + /// # 首帧处理 + /// + /// `Option::None` 表示窗口内首帧，没有"上一帧"参照点， + /// 因此首帧不产生 gap 样本（这与 Go 中 `*time.Time == nil` 检查等价， + /// 但 Rust 强制处理 None 分支，编译期避免 nil 解引用）。 + /// /// Record that a capture frame was received from PipeWire. pub fn record_capture(&mut self) { let now = Instant::now(); @@ -115,6 +187,14 @@ impl PipelineStats { self.capture_frames += 1; } + /// 记录一帧完成编码（`FrameTimings` 路径，含各阶段微秒样本）。 + /// + /// # 参数借用 + /// + /// `timings: &FrameTimings`：以共享借用（`&`）读取，不获取所有权。 + /// 类比 Go 中显式传递 `*FrameTimings` 指针；Rust 借用检查保证本调用 + /// 期间原 `timings` 不会被释放。其余 gap 计算同 `record_capture`。 + /// /// Record that a frame completed encoding with the given timings. pub fn record_encode(&mut self, timings: &FrameTimings) { let now = Instant::now(); @@ -134,10 +214,19 @@ impl PipelineStats { self.output_bytes.push(timings.output_bytes); } + /// 仅记录 import 阶段微秒数（用于 `record_encode_thread` 路径补齐 import 样本）。 pub fn record_import(&mut self, import_us: u64) { self.import_us.push(import_us); } + /// 编码线程路径：散参传入 sws / encode / output_bytes，跳过 `FrameTimings`。 + /// + /// # 溢出保护 + /// + /// `saturating_add` 在 `u64::MAX` 处饱和而非回绕，避免极端情况下 + /// `total_us` 出现荒谬的小值。类比 Go 中需手动 `if total > MaxUint64 - x` + /// 检查；Rust 的 `saturating_*` / `checked_*` / `wrapping_*` 三件套 + /// 让溢出策略在调用点显式表达。 pub fn record_encode_thread(&mut self, sws_us: u64, encode_us: u64, output_bytes: usize) { let now = Instant::now(); if let Some(last) = self.last_encode_time { @@ -153,6 +242,19 @@ impl PipelineStats { self.output_bytes.push(output_bytes); } + /// 记录一帧通过 WebRTC 发送（中文 L3 解析见此）。 + /// + /// - `wait_ms`：阻塞等待发送通道可写入的时间（毫秒）；为 0 时不入样本 + /// （避免拉低 p95，因为大多数帧应无等待） + /// - `capture_time`：该帧的原始捕获时刻；用于计算 **frame age** + /// （捕获→发送端到端延迟）。`Option` 表示调用方未提供 + /// （例如 XDG/screen-copy 路径无原始时间戳），此时不入样本 + /// + /// # frame_age 计算 + /// + /// `ct.elapsed()` 返回 `Duration`，类同 `record_capture` 中 gap 计算， + /// 但锚点是"捕获时刻"而非"上一帧发送时刻"，因此测量的是端到端延迟。 + /// /// Record that a frame was sent via WebRTC. /// `wait_ms` is time spent blocked waiting to send into the channel. /// `capture_time` is when the frame was originally captured (for frame age). @@ -174,6 +276,18 @@ impl PipelineStats { } } + /// 从后台 WebRTC 发送线程记录一帧（gap_ms / age_ms 均已在调用方预算好）。 + /// + /// # 为何预算参数 + /// + /// 后台线程无法安全访问 `&mut self`（本结构非 `Sync`），因此调用方在 + /// 发送时刻直接计算 `gap_ms` / `age_ms`（`Instant::now()` 在该线程 + /// 局部调用），稍后批量 drain 到主线程的 `&mut self`。这样： + /// - 单调时钟读取在事件发生线程完成，时间戳精确 + /// - 主线程仅做 `Vec::push`，无需锁 + /// + /// `gap_ms == 0.0` 表示首帧（无前一帧参照），不入样本。 + /// /// Record a frame sent from a background WebRTC thread. /// `gap_ms` is the pre-computed time since the previous send (0.0 = first frame). /// `age_ms` is the pre-computed capture-to-send latency (None if unavailable). @@ -189,11 +303,32 @@ impl PipelineStats { } } + /// 设置 PipeWire dropped 计数（绝对值，由调用方从外部 `AtomicU64` 读取）。 + /// + /// # 增量计算 + /// + /// 外部 `AtomicU64` 累计**总会话**的 dropped 帧数（从不重置）， + /// 因此本函数计算 `total - prev` 得到本窗口内的增量。 + /// `saturating_sub` 防止极端竞态（如原子读顺序不一致）导致负数回绕。 + /// + /// # 与 Go 类比 + /// + /// - 调用方代码 ≈ Go `atomic.LoadUint64(&pw.dropped)`（`Ordering::SeqCst` + /// 或 `Relaxed` 取决于是否需要与其他原子操作建立 happens-before） + /// - `Mutex` 在本模块**未使用**：统计字段集固定，无需 Go + /// `sync.Map` 那样的动态键值存储；跨线程仅通过原子计数器通信 + /// /// Update PipeWire dropped counter (absolute value from AtomicU64). pub fn set_pipewire_dropped(&mut self, total_dropped: u64, prev_dropped: u64) { self.pipewire_dropped = total_dropped.saturating_sub(prev_dropped); } + /// 设置 duplicate frames skipped 计数（绝对值，由调用方从 encode 线程原子读取）。 + /// + /// 与 `set_pipewire_dropped` 增量算法一致，但**保留 `prev`** 在 + /// `self.prev_duplicate_frames_skipped` 字段中（因为本结构才是状态持有者， + /// 调用方仅传入当前 total）。 + /// /// Update duplicate frames skipped counter (absolute value from atomic). /// Computes delta from previous value, like set_pipewire_dropped. pub fn set_duplicate_frames_skipped(&mut self, total_skipped: u64) { @@ -201,23 +336,49 @@ impl PipelineStats { self.prev_duplicate_frames_skipped = total_skipped; } + /// 更新队列深度瞬时观测值（capture 队列与 encoded 队列各一个值）。 + /// + /// 队列深度为快照值而非累计值，每窗口只保留最后一次观测。 + /// /// Update queue depth observations. pub fn set_queue_depths(&mut self, capture: usize, encoded: usize) { self.capture_queue_depth = capture; self.encoded_queue_depth = encoded; } + /// 记录一帧超出预算（用于跟踪编码耗时超过 1/fps 的频次）。 + /// /// Record that a frame exceeded its time budget. pub fn record_over_budget(&mut self) { self.over_budget_count += 1; } + /// 判断是否到达快照点（距离上次 `snapshot_and_reset` 或构造时刻 ≥ 1 秒）。 + /// + /// 仅需 `&self`（共享借用）：本方法不修改任何字段，借用检查器允许 + /// 多个 `&self` 共存或与 `&mut self` 之外的调用并存。 + /// /// Returns true if at least 1 second has elapsed since the last snapshot /// (or since creation). If true, call `snapshot_and_reset` to get the stats. pub fn should_snapshot(&self) -> bool { self.window_start.elapsed().as_secs() >= 1 } + /// 计算当前窗口的统计快照并重置所有计数器与样本缓冲。 + /// + /// # 重置策略 + /// + /// - 计数器：置零 + /// - `Vec`：调用 `clear()`（保留已分配容量 `Vec::capacity()`，避免下个窗口 + /// 反复分配）。类比 Go 中 `s = s[:0]` 复用底层数组 + /// - `window_start = Instant::now()`：重置窗口起点 + /// + /// # 返回值 + /// + /// 返回 `StatsSnapshot` 值（拷贝语义，调用方可自由使用与丢弃）。 + /// 类比 Go 中返回值结构体的拷贝；Rust 中 `StatsSnapshot` 全部字段为 + /// `Copy` 或 `Vec`（移动语义），返回时所有权转移至调用方。 + /// /// Compute a snapshot of the current window and reset all counters. pub fn snapshot_and_reset(&mut self) -> StatsSnapshot { let elapsed = self.window_start.elapsed().as_secs_f64(); @@ -291,6 +452,17 @@ impl PipelineStats { } } +/// 一秒窗口的管道统计快照（不可变值对象，由 `snapshot_and_reset` 返回）。 +/// +/// 本结构持有所有派生指标（FPS、avg/p95/max、计数器快照），是日志输出的 +/// 数据源。一旦创建即不可变（所有字段为 `f64`/`u64`/`usize`，天然 `Copy`）， +/// 调用方可以安全地打印、记录或丢弃。 +/// +/// # `#[derive(Debug)]` 用途 +/// +/// 调试场景下可直接 `dbg!(&snap)` 或 `tracing::debug!(?snap)`， +/// 类比 Go 的 `spew.Dump(snap)` / `fmt.Printf("%+v", snap)`。 +/// /// A one-second snapshot of pipeline statistics. #[derive(Debug)] pub struct StatsSnapshot { @@ -344,6 +516,40 @@ pub struct StatsSnapshot { pub output_frame_bytes_max: usize, } +/// 单行结构化日志格式化器（中文 L3 解析见此，英文说明保留在下方）。 +/// +/// # trait 与签名说明 +/// +/// - `impl std::fmt::Display for StatsSnapshot`：为本类型实现标准库 trait， +/// 使得 `format!("{snap}")` / `println!("{}", snap)` / `tracing::info!("{}", snap)` +/// 均可工作（隐式调用 `fmt` 方法） +/// - `fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result`： +/// - `&self` 共享借用，格式化不改自身 +/// - `Formatter<'_>`：匿名生命周期（`'_`），表示该借用与调用方持有的 +/// `String`/输出流绑定；类比 Go 的 `io.Writer` 参数 +/// - `std::fmt::Result`：`Result<(), std::fmt::Error>`，专用于格式化 trait +/// （比通用 `Result` 更窄，避免 `?` 跨类型传播） +/// +/// # `write!` 宏 vs `format!` 宏 +/// +/// - [`write!`]：直接写入 `Formatter`（零分配），类比 Go `fmt.Fprintf(w, ...)` +/// - [`format!`]：分配新 `String` 后返回，类比 Go `fmt.Sprintf(...)` +/// - 本实现选 `write!`：写入日志流时避免多余分配 +/// +/// # `?` 运算符 +/// +/// `write!(...)?` 中的 `?` 是早期返回：若 `write!` 返回 `Err(fmt::Error)`， +/// 则立即从 `fmt` 返回该错误。类比 Go 中 +/// `if _, err := w.Write(...); err != nil { return err }`， +/// 但 Rust 的 `?` 让快乐路径线性化。 +/// +/// # 格式说明 +/// +/// - `{:.1}`：保留 1 位小数 +/// - `{:.0}`：整数显示（无小数点） +/// - `{}`：默认 `Display` 格式（整数原样） +/// - 行尾反斜杠 `\` 跨行延续字符串字面量，类比 Python 隐式行连接； +/// 输出时不会引入额外换行或空格 impl std::fmt::Display for StatsSnapshot { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!( @@ -392,6 +598,13 @@ impl std::fmt::Display for StatsSnapshot { // Statistics helpers // --------------------------------------------------------------------------- +/// 计算 `f64` 切片平均值（空切片返回 0.0）。 +/// +/// # 切片借用 +/// +/// `data: &[f64]`：共享借用切片（fat pointer = 指针 + 长度），类比 Go 中 +/// `func avg(data []float64)`。`&` 表示本函数不获取所有权，调用后原 `Vec` +/// 仍可用。 fn avg_f64(data: &[f64]) -> f64 { if data.is_empty() { return 0.0; @@ -399,6 +612,21 @@ fn avg_f64(data: &[f64]) -> f64 { data.iter().sum::() / data.len() as f64 } +/// 计算 p95（第 95 百分位），类比 Go 中需要手动 sort + index。 +/// +/// # 算法 +/// +/// 1. 复制输入到新 `Vec`（不修改调用方原数据）：`data.to_vec()` 类比 Go +/// `append([]T{}, data...)` +/// 2. 排序：`sort_by` + `partial_cmp` —— `f64` 没有全序（NaN 特殊）， +/// 不能直接用 `sort()`；`partial_cmp(b).unwrap_or(Equal)` 在 NaN 时 +/// 降级为相等，避免 panic +/// 3. 计算 idx = `floor(len * 0.95)`，`idx.min(len-1)` 防越界 +/// +/// # 为何不用 `sort_unstable` +/// +/// `f64` 的 `Ord` 未实现（NaN 不等于自身），故只能用 `sort_by` + 比较 +/// 函数；`u64`/`usize` 实现 `Ord`，可用 `sort_unstable`（更快、内存友好）。 fn p95_f64(data: &[f64]) -> f64 { if data.is_empty() { return 0.0; @@ -409,10 +637,22 @@ fn p95_f64(data: &[f64]) -> f64 { sorted[idx.min(sorted.len() - 1)] } +/// 返回切片最大值，空切片返回 0.0（业务上"无样本"等价于"无延迟"）。 +/// +/// # `fold` + `f64::max` +/// +/// `fold(0.0, f64::max)`：从初始值 0.0 开始，逐元素取较大值。 +/// 类比 Go： +/// ```go +/// m := 0.0 +/// for _, v := range data { m = math.Max(m, v) } +/// ``` +/// 注意：若样本全为负，0.0 仍是结果（业务上 latency 非负，不会出现）。 fn max_f64(data: &[f64]) -> f64 { data.iter().copied().fold(0.0_f64, f64::max) } +/// 计算 `u64` 微秒样本的平均值并转毫秒（÷1000）。 fn avg_ms(data: &[u64]) -> f64 { if data.is_empty() { return 0.0; @@ -420,6 +660,10 @@ fn avg_ms(data: &[u64]) -> f64 { data.iter().sum::() as f64 / data.len() as f64 / 1000.0 } +/// 计算 `u64` 微秒样本的 p95 并转毫秒。 +/// +/// 与 `p95_f64` 算法相同，但 `u64` 实现 `Ord`，可用 `sort_unstable` +/// （无内存开销、更快；稳定性对本场景无关，因为只取索引位置）。 fn p95_ms(data: &[u64]) -> f64 { if data.is_empty() { return 0.0; @@ -430,10 +674,15 @@ fn p95_ms(data: &[u64]) -> f64 { sorted[idx.min(sorted.len() - 1)] as f64 / 1000.0 } +/// 对 `usize` 切片求和（用于累计 output_bytes 总量）。 +/// +/// `data.iter().sum()` 由标准库自动推导类型（`usize`），等价于 +/// Go 中 `var total uint; for _, v := range data { total += v }`。 fn sum_usize(data: &[usize]) -> usize { data.iter().sum() } +/// 计算 `usize` 样本的 p95（字节大小分布），不转换单位。 fn p95_usize(data: &[usize]) -> usize { if data.is_empty() { return 0; @@ -444,10 +693,26 @@ fn p95_usize(data: &[usize]) -> usize { sorted[idx.min(sorted.len() - 1)] } +/// 返回 `usize` 切片最大值，空切片返回 0。 +/// +/// `iter().copied().max()` 返回 `Option`（空时为 `None`）， +/// `unwrap_or(0)` 提供默认值，类比 Go 中显式 `if len(data) == 0 { return 0 }`。 fn max_usize(data: &[usize]) -> usize { data.iter().copied().max().unwrap_or(0) } +/// 单元测试模块（仅在 `#[cfg(test)]` 时编译）。 +/// +/// # Rust 测试模式 +/// +/// - `#[cfg(test)]`：条件编译属性，`cargo test` 时才编译本模块， +/// 正常 `cargo build` 不包含本模块代码（类比 Go 中 `_test.go` 后缀 +/// 仅在 `go test` 时段编译，但 Rust 用显式属性而非文件名约定） +/// - `use super::*`：导入父模块（本文件）所有 `pub` 与私密 item， +/// 类比 Go test 文件无需 import 即可访问同包符号 +/// - `#[test]`：标记测试函数；`cargo test` 自动发现并执行 +/// - `assert_eq!` / `assert!`：宏（不是函数），失败时打印表达式原文 +/// 便于调试，类比 Go 中 `t.Errorf` 但更早终止当前测试 #[cfg(test)] mod tests { use super::*;