docs(state_portal): [2/2] 中文注释 Portal 帧循环与编码调度
This commit is contained in:
@@ -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(
|
fn encode_thread_loop(
|
||||||
mut encode: SwEncEncode,
|
mut encode: SwEncEncode,
|
||||||
input_rx: crossbeam_channel::Receiver<CpuNv12Frame>,
|
input_rx: crossbeam_channel::Receiver<CpuNv12Frame>,
|
||||||
timing_tx: crossbeam_channel::Sender<EncodeThreadTiming>,
|
timing_tx: crossbeam_channel::Sender<EncodeThreadTiming>,
|
||||||
duplicate_count: std::sync::Arc<std::sync::atomic::AtomicU64>,
|
duplicate_count: std::sync::Arc<std::sync::atomic::AtomicU64>,
|
||||||
) {
|
) {
|
||||||
|
// 阻塞循环:`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<AtomicU64>::fetch_add + Relaxed 累加——无锁、无需 happens-before。
|
||||||
|
// - Err(_) → 所有 Sender 已 drop,flush 编码器后退出循环。
|
||||||
loop {
|
loop {
|
||||||
match input_rx.recv() {
|
match input_rx.recv() {
|
||||||
Ok(frame) => {
|
Ok(frame) => {
|
||||||
@@ -795,6 +804,13 @@ fn encode_thread_loop(
|
|||||||
tracing::info!("Encode thread exiting");
|
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<AtomicBool> 跨线程共享:编码线程读,本线程写。
|
||||||
fn webrtc_thread_loop(
|
fn webrtc_thread_loop(
|
||||||
mut wrtc: WebRtcState,
|
mut wrtc: WebRtcState,
|
||||||
webrtc_rx: crossbeam_channel::Receiver<EncodedH264Frame>,
|
webrtc_rx: crossbeam_channel::Receiver<EncodedH264Frame>,
|
||||||
@@ -814,6 +830,7 @@ fn webrtc_thread_loop(
|
|||||||
let mut current_tier = initial_tier;
|
let mut current_tier = initial_tier;
|
||||||
let mut upscale_counter = 0u32;
|
let mut upscale_counter = 0u32;
|
||||||
let mut last_resolution_eval = Instant::now();
|
let mut last_resolution_eval = Instant::now();
|
||||||
|
// recv 超时 1ms——既能让循环周期性处理 str0m 信令,又能在帧到达时立即返回。
|
||||||
let timeout = Duration::from_millis(1);
|
let timeout = Duration::from_millis(1);
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
@@ -831,6 +848,8 @@ fn webrtc_thread_loop(
|
|||||||
}
|
}
|
||||||
|
|
||||||
let connected = wrtc.is_connected();
|
let connected = wrtc.is_connected();
|
||||||
|
// Arc<AtomicBool> 跨线程协调:编码线程 Relaxed 读 paused;本线程 Relaxed 写。
|
||||||
|
// Relaxed 取舍:暂停标志无内存序需求(不保护其他共享数据),只需原子可见性。
|
||||||
let was_paused = paused.load(Ordering::Relaxed);
|
let was_paused = paused.load(Ordering::Relaxed);
|
||||||
let now_paused = !connected;
|
let now_paused = !connected;
|
||||||
if was_paused && !now_paused {
|
if was_paused && !now_paused {
|
||||||
@@ -900,6 +919,8 @@ fn webrtc_thread_loop(
|
|||||||
}
|
}
|
||||||
|
|
||||||
if connected {
|
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() {
|
while let Ok(enc_frame) = webrtc_rx.try_recv() {
|
||||||
if let Err(e) = wrtc.write_h264_frame(&enc_frame.data, enc_frame.pts_ticks) {
|
if let Err(e) = wrtc.write_h264_frame(&enc_frame.data, enc_frame.pts_ticks) {
|
||||||
tracing::debug!("WebRTC write frame error: {e}");
|
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));
|
let _ = sent_gap_tx.try_send((gap_ms, age_ms));
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
|
// 未连接:丢弃积压帧防止 drain 时刻反向堆积(类比 Go `for { select { case <-rx: default: return } }`)。
|
||||||
while webrtc_rx.try_recv().is_ok() {}
|
while webrtc_rx.try_recv().is_ok() {}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// recv_timeout:阻塞至下一帧或最多 1ms——保证 str0m 信令循环周期性推进。
|
||||||
|
// 三路 Result:Ok → 处理帧;Err(Timeout) → 继续下一轮循环处理信令;Err(Disconnected) → 编码线程已退出,本线程返回。
|
||||||
match webrtc_rx.recv_timeout(timeout) {
|
match webrtc_rx.recv_timeout(timeout) {
|
||||||
Ok(enc_frame) => {
|
Ok(enc_frame) => {
|
||||||
if wrtc.is_connected() {
|
if wrtc.is_connected() {
|
||||||
@@ -947,12 +971,19 @@ fn webrtc_thread_loop(
|
|||||||
tracing::info!("WebRTC thread exiting");
|
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)];
|
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 {
|
fn resolution_bitrate_bps(width: u32, height: u32, fps: u32) -> u64 {
|
||||||
5 * u64::from(width) * u64::from(height) * u64::from(fps) / 100
|
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.
|
/// Conservative startup bitrate for WebRTC mode, tier-based by total pixel count.
|
||||||
/// BWE estimate arrives within milliseconds of client connect and overrides this;
|
/// BWE estimate arrives within milliseconds of client connect and overrides this;
|
||||||
/// the startup value only affects the first IDR. See issue #21.
|
/// 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.
|
/// Select resolution tier based on BWE estimate.
|
||||||
/// Returns (width, height) for the selected tier.
|
/// Returns (width, height) for the selected tier.
|
||||||
fn select_resolution(current_w: u32, current_h: u32, bwe_bps: u64, fps: u32) -> (u32, u32) {
|
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;
|
return current;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 在 RESOLUTION_TIERS 中找当前分辨率的位置;若不在表中(如 1366×768),
|
||||||
|
// 用 unwrap_or_else 退回到第一个宽高都不超过 current 的档位,最终兜底取最小档(720p)。
|
||||||
let current_index = RESOLUTION_TIERS
|
let current_index = RESOLUTION_TIERS
|
||||||
.iter()
|
.iter()
|
||||||
.position(|&tier| tier == current)
|
.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]
|
RESOLUTION_TIERS[next_index]
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 反向 upscale:在 ceiling 上限内尝试升一档;若已在最高档或下一档超出 ceiling 则返回 None。
|
||||||
|
// 调用方需要"连续 10 次 BWE 充足"才真正切换,避免 BWE 抖动导致频繁分辨率变化。
|
||||||
fn next_upscale_tier(current: (u32, u32), ceiling: (u32, u32)) -> Option<(u32, u32)> {
|
fn next_upscale_tier(current: (u32, u32), ceiling: (u32, u32)) -> Option<(u32, u32)> {
|
||||||
let current_index = RESOLUTION_TIERS.iter().position(|&tier| tier == current)?;
|
let current_index = RESOLUTION_TIERS.iter().position(|&tier| tier == current)?;
|
||||||
if current_index == 0 {
|
if current_index == 0 {
|
||||||
return None;
|
return None;
|
||||||
}
|
}
|
||||||
let next = RESOLUTION_TIERS[current_index - 1];
|
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)
|
(next.0 <= ceiling.0 && next.1 <= ceiling.1).then_some(next)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1047,6 +1086,10 @@ fn resolve_drm_device(args: &Args) -> Result<Option<PathBuf>> {
|
|||||||
/// 用于验证 DMA-BUF 元数据映射的正确性。
|
/// 用于验证 DMA-BUF 元数据映射的正确性。
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
fn build_drm_descriptor(frame: &PwDmaBufFrame) -> ffmpeg_next::ffi::AVDRMFrameDescriptor {
|
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() };
|
let mut desc: ffmpeg_next::ffi::AVDRMFrameDescriptor = unsafe { std::mem::zeroed() };
|
||||||
desc.nb_objects = 1; // 单个 DMA-BUF 对象
|
desc.nb_objects = 1; // 单个 DMA-BUF 对象
|
||||||
desc.objects[0].fd = frame.fd.as_raw_fd(); // DMA-BUF 文件描述符
|
desc.objects[0].fd = frame.fd.as_raw_fd(); // DMA-BUF 文件描述符
|
||||||
@@ -1070,6 +1113,8 @@ mod tests {
|
|||||||
fn make_test_frame() -> PwDmaBufFrame {
|
fn make_test_frame() -> PwDmaBufFrame {
|
||||||
// Create a dummy fd from stderr (always valid fd 2)
|
// Create a dummy fd from stderr (always valid fd 2)
|
||||||
// 使用 stderr(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)) };
|
let fd = unsafe { OwnedFd::from_raw_fd(libc::dup(2)) };
|
||||||
PwDmaBufFrame {
|
PwDmaBufFrame {
|
||||||
fd,
|
fd,
|
||||||
@@ -1193,6 +1238,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn build_drm_descriptor_custom_offset_and_stride() {
|
fn build_drm_descriptor_custom_offset_and_stride() {
|
||||||
let frame = PwDmaBufFrame {
|
let frame = PwDmaBufFrame {
|
||||||
|
// unsafe:同 make_test_frame——dup(2) 复制 stderr fd 并交给 OwnedFd 管理。
|
||||||
fd: unsafe { OwnedFd::from_raw_fd(libc::dup(2)) },
|
fd: unsafe { OwnedFd::from_raw_fd(libc::dup(2)) },
|
||||||
offset: 4096, // 4KB 对齐偏移
|
offset: 4096, // 4KB 对齐偏移
|
||||||
stride: 3840 * 4, // 4K 宽度 × 4 字节
|
stride: 3840 * 4, // 4K 宽度 × 4 字节
|
||||||
|
|||||||
Reference in New Issue
Block a user