From 09123fcd6511eeba3b6896e46b64b7a74c6c4a07 Mon Sep 17 00:00:00 2001 From: dailz Date: Mon, 22 Jun 2026 17:28:53 +0800 Subject: [PATCH] =?UTF-8?q?docs(state=5Fportal):=20[2/2]=20=E4=B8=AD?= =?UTF-8?q?=E6=96=87=E6=B3=A8=E9=87=8A=20Portal=20=E5=B8=A7=E5=BE=AA?= =?UTF-8?q?=E7=8E=AF=E4=B8=8E=E7=BC=96=E7=A0=81=E8=B0=83=E5=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/state_portal.rs | 46 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/src/state_portal.rs b/src/state_portal.rs index 40d3a54..16c5f40 100644 --- a/src/state_portal.rs +++ b/src/state_portal.rs @@ -753,12 +753,21 @@ impl StatePortal { } } +// === 编码线程主循环（独立 std::thread，非 tokio） === +// 类比 Go：`go func(input <-chan Frame) { for f := range input { encode(f) } }`。 +// 线程持有 SwEncEncode 的所有权（move 语义），消费 input_rx 直到对端 drop 所有 Sender。 +// 编码结果通过 timing_tx（单帧耗时）+ duplicate_count（重复帧统计）回传主线程。 fn encode_thread_loop( mut encode: SwEncEncode, input_rx: crossbeam_channel::Receiver, timing_tx: crossbeam_channel::Sender, duplicate_count: std::sync::Arc, ) { + // 阻塞循环：`match input_rx.recv()` 类比 Go `for frame := range input_rx {}`。 + // - Ok(frame) → 调用 encode_cpu_frame；match EncodeOutcome 各 variant 分别处理； + // timing_tx.try_send 非阻塞回执（满则丢，类比 Go `select { case ch <- v: default: }`）； + // 重复帧计数通过 Arc::fetch_add + Relaxed 累加——无锁、无需 happens-before。 + // - Err(_) → 所有 Sender 已 drop，flush 编码器后退出循环。 loop { match input_rx.recv() { Ok(frame) => { @@ -795,6 +804,13 @@ fn encode_thread_loop( tracing::info!("Encode thread exiting"); } +// === WebRTC 信令 + 帧发送主循环（独立 std::thread，非 tokio） === +// 该线程串行处理 4 件事： +// 1. str0m 信令（ICE/DTLS）+ RTP 打包发送（wrtc.handle_signaling / poll_and_feed）； +// 2. 自适应码率（BWE）→ bitrate_tx 下发 UpdateBitrate/ForceKeyframe 给编码线程； +// 3. 自适应分辨率（每 1s 评估）→ resolution_tx 下发 UpdateResolution； +// 4. 从 webrtc_rx 取已编码 H264 帧，写入 str0m RTP sink。 +// 暂停状态由 Arc 跨线程共享：编码线程读，本线程写。 fn webrtc_thread_loop( mut wrtc: WebRtcState, webrtc_rx: crossbeam_channel::Receiver, @@ -814,6 +830,7 @@ fn webrtc_thread_loop( let mut current_tier = initial_tier; let mut upscale_counter = 0u32; let mut last_resolution_eval = Instant::now(); + // recv 超时 1ms——既能让循环周期性处理 str0m 信令，又能在帧到达时立即返回。 let timeout = Duration::from_millis(1); loop { @@ -831,6 +848,8 @@ fn webrtc_thread_loop( } let connected = wrtc.is_connected(); + // Arc 跨线程协调：编码线程 Relaxed 读 paused；本线程 Relaxed 写。 + // Relaxed 取舍：暂停标志无内存序需求（不保护其他共享数据），只需原子可见性。 let was_paused = paused.load(Ordering::Relaxed); let now_paused = !connected; if was_paused && !now_paused { @@ -900,6 +919,8 @@ fn webrtc_thread_loop( } if connected { + // 已连接：批量 drain 已编码帧队列（类比 Go `for { select { case f := <-rx: send(f); default: break } }`）。 + // saturating_add 防止计数器溢出（Go 没有，Rust 默认 panic-on-overflow，debug 下尤其危险）。 while let Ok(enc_frame) = webrtc_rx.try_recv() { if let Err(e) = wrtc.write_h264_frame(&enc_frame.data, enc_frame.pts_ticks) { tracing::debug!("WebRTC write frame error: {e}"); @@ -916,9 +937,12 @@ fn webrtc_thread_loop( let _ = sent_gap_tx.try_send((gap_ms, age_ms)); } } else { + // 未连接：丢弃积压帧防止 drain 时刻反向堆积（类比 Go `for { select { case <-rx: default: return } }`）。 while webrtc_rx.try_recv().is_ok() {} } + // recv_timeout：阻塞至下一帧或最多 1ms——保证 str0m 信令循环周期性推进。 + // 三路 Result：Ok → 处理帧；Err(Timeout) → 继续下一轮循环处理信令；Err(Disconnected) → 编码线程已退出，本线程返回。 match webrtc_rx.recv_timeout(timeout) { Ok(enc_frame) => { if wrtc.is_connected() { @@ -947,12 +971,19 @@ fn webrtc_thread_loop( tracing::info!("WebRTC thread exiting"); } +// 自适应分辨率阶梯（从高到低）。下标 0 = 最高分辨率（2K），下标 2 = 最低（720p）。 +// BWE 不足时 select_resolution 从数组下标小的（高分辨率）向大的（低分辨率）切换； +// 反向 upscale 由 next_upscale_tier 处理，受 initial_tier 上限约束（不会超过初始分辨率）。 const RESOLUTION_TIERS: &[(u32, u32)] = &[(2560, 1440), (1920, 1080), (1280, 720)]; +// 启发式码率估算：`5 × W × H × fps / 100` 即 0.05 bits/pixel/frame。 +// 类似 H.264 平均量化参考值，作为 BWE 充分性判据（≥ 60% 认为可承载当前分辨率）。 fn resolution_bitrate_bps(width: u32, height: u32, fps: u32) -> u64 { 5 * u64::from(width) * u64::from(height) * u64::from(fps) / 100 } +// WebRTC 启动码率：按总像素数分 4 档（≤1M / ≤2.5M / ≤4.5M / 其他 → 1/2/4/8 Mbps）。 +// 仅影响客户端连接后第一个 IDR；BWE 估计（毫秒级到达）会覆盖此值。详见 issue #21。 /// Conservative startup bitrate for WebRTC mode, tier-based by total pixel count. /// BWE estimate arrives within milliseconds of client connect and overrides this; /// the startup value only affects the first IDR. See issue #21. @@ -969,6 +1000,8 @@ fn webrtc_startup_bitrate_bps(width: u32, height: u32) -> u64 { } } +// 基于 BWE 选择分辨率阶梯。返回 (width, height)。 +// 决策逻辑：若 BWE ≥ 当前分辨率所需码率的 60%，保持不变；否则降到下一档（最低 720p）。 /// Select resolution tier based on BWE estimate. /// Returns (width, height) for the selected tier. fn select_resolution(current_w: u32, current_h: u32, bwe_bps: u64, fps: u32) -> (u32, u32) { @@ -978,6 +1011,8 @@ fn select_resolution(current_w: u32, current_h: u32, bwe_bps: u64, fps: u32) -> return current; } + // 在 RESOLUTION_TIERS 中找当前分辨率的位置；若不在表中（如 1366×768）， + // 用 unwrap_or_else 退回到第一个宽高都不超过 current 的档位，最终兜底取最小档（720p）。 let current_index = RESOLUTION_TIERS .iter() .position(|&tier| tier == current) @@ -991,12 +1026,16 @@ fn select_resolution(current_w: u32, current_h: u32, bwe_bps: u64, fps: u32) -> RESOLUTION_TIERS[next_index] } +// 反向 upscale：在 ceiling 上限内尝试升一档；若已在最高档或下一档超出 ceiling 则返回 None。 +// 调用方需要"连续 10 次 BWE 充足"才真正切换，避免 BWE 抖动导致频繁分辨率变化。 fn next_upscale_tier(current: (u32, u32), ceiling: (u32, u32)) -> Option<(u32, u32)> { let current_index = RESOLUTION_TIERS.iter().position(|&tier| tier == current)?; if current_index == 0 { return None; } let next = RESOLUTION_TIERS[current_index - 1]; + // bool::then_some(true → Some(next)，false → None）：将谓词结果转换为 Option， + // 类比 Go `if ok { return &tier } else { return nil }`。 (next.0 <= ceiling.0 && next.1 <= ceiling.1).then_some(next) } @@ -1047,6 +1086,10 @@ fn resolve_drm_device(args: &Args) -> Result> { /// 用于验证 DMA-BUF 元数据映射的正确性。 #[cfg(test)] fn build_drm_descriptor(frame: &PwDmaBufFrame) -> ffmpeg_next::ffi::AVDRMFrameDescriptor { + // unsafe：调用 std::mem::zeroed() 对 #[repr(C)] 结构体进行零初始化—— + // AVDRMFrameDescriptor 是 FFmpeg C 结构体，零值是合法的"空"状态（nb_objects/nb_layers=0， + // 后续字段在下方显式赋值）。`std::mem::zeroed` 对带指针字段的类型可能产生空悬指针（UB）， + // 此处安全：descriptor 的所有字段都是整数/数组，没有指针/引用。 let mut desc: ffmpeg_next::ffi::AVDRMFrameDescriptor = unsafe { std::mem::zeroed() }; desc.nb_objects = 1; // 单个 DMA-BUF 对象 desc.objects[0].fd = frame.fd.as_raw_fd(); // DMA-BUF 文件描述符 @@ -1070,6 +1113,8 @@ mod tests { fn make_test_frame() -> PwDmaBufFrame { // Create a dummy fd from stderr (always valid fd 2) // 使用 stderr（fd 2）的副本作为虚拟文件描述符 + // unsafe：libc::dup(2) 复制 stderr fd → 返回新整数 fd；OwnedFd::from_raw_fd 接管 + // 该 fd 的 close 责任（RAII）。前提：libc::dup 调用成功（fd 2 始终有效，不检查返回值是测试代码约定）。 let fd = unsafe { OwnedFd::from_raw_fd(libc::dup(2)) }; PwDmaBufFrame { fd, @@ -1193,6 +1238,7 @@ mod tests { #[test] fn build_drm_descriptor_custom_offset_and_stride() { let frame = PwDmaBufFrame { + // unsafe：同 make_test_frame——dup(2) 复制 stderr fd 并交给 OwnedFd 管理。 fd: unsafe { OwnedFd::from_raw_fd(libc::dup(2)) }, offset: 4096, // 4KB 对齐偏移 stride: 3840 * 4, // 4K 宽度 × 4 字节