1346 lines
59 KiB
Rust
1346 lines
59 KiB
Rust
//! Portal 后端的主状态机：通过 PipeWire + DMA-BUF 进行屏幕采集并软件编码。
|
||
//!
|
||
//! ## 整体角色
|
||
//!
|
||
//! `StatePortal` 与 `src/state.rs::State` 是平行的两条采集路径：
|
||
//! - `state.rs`（wlroots 路径）：由外层 `mio` 事件循环驱动（手工版 epoll），
|
||
//! 通过 `zwlr_screencopy_manager_v1` 协议一帧一帧地拉取。
|
||
//! - `state_portal.rs`（本文件，XDG Portal / PipeWire 路径）：由 `CapPortal`
|
||
//! 通过 `crossbeam_channel::Receiver<PwDmaBufFrame>` 推帧；本状态机只负责"消费"。
|
||
//!
|
||
//! ## 异步模型的真相
|
||
//!
|
||
//! 本文件**不**使用 `mio` 或 `tokio`——`CapPortal` 内部在独立线程跑 PipeWire
|
||
//! asyncio loop，把 DMA-BUF 帧通过 crossbeam channel 投递出来；外层 `main.rs`
|
||
//! 只需在 `while !is_errored()` 循环里轮询 `poll_and_encode(block)`。编码线程与
|
||
//! WebRTC 线程通过 `std::thread::spawn`（不是 `tokio::spawn`）启动，再借助
|
||
//! crossbeam channel 与主线程通信——类比 Go 的 `go func()` + channel。
|
||
//!
|
||
//! ## 阶段机
|
||
//!
|
||
//! `PortalStage::WaitingForFormat`（等首帧以确定格式）→ `Streaming`（持续编码）。
|
||
//!
|
||
//! ## 注意
|
||
//!
|
||
//! - T9a（本块）覆盖文件头 + struct 定义 + `impl StatePortal`（至 `fn encode_thread_loop` 之前）；
|
||
//! T9b 覆盖 `encode_thread_loop` / `webrtc_thread_loop` / `resolve_drm_device` 等自由函数。
|
||
//! - 多处 `unsafe` 调用 FFmpeg/VAAPI FFI；现有英文 SAFETY 标记保留不动，
|
||
//! 本任务在每个 unsafe 块上方加普通 `//` 中文概述（不新增 SAFETY 标记）。
|
||
|
||
// 采集门户状态模块 —— 通过 PipeWire/DMA-BUF 进行屏幕采集并编码
|
||
use std::os::fd::AsRawFd;
|
||
use std::path::PathBuf;
|
||
use std::sync::atomic::{AtomicBool, Ordering};
|
||
use std::sync::Arc;
|
||
use std::time::{Duration, Instant};
|
||
|
||
use anyhow::{bail, Result}; // 错误处理工具
|
||
|
||
use crate::args::Args; // 命令行参数
|
||
use crate::avhw::{
|
||
self, BitrateCommand, CpuNv12Frame, EncodeOutcome, EncodedH264Frame, ResolutionChange,
|
||
SwEncEncode, SwEncImport, SwEncState,
|
||
}; // 软件编码器状态（VAAPI 导入 + H.264 编码）
|
||
use crate::cap_portal::{CapPortal, PwCtrlEvent, PwDmaBufFrame}; // PipeWire 屏幕采集端点
|
||
use crate::stats::{FrameTimings, PipelineStats}; // 管道统计（帧计时、每秒快照）
|
||
use crate::webrtc::WebRtcState; // WebRTC 信令与媒体传输
|
||
|
||
/// 门户采集的阶段状态
|
||
/// - WaitingForFormat: 等待接收到第一帧 DMA-BUF 以确定视频格式参数
|
||
/// - Streaming: 已完成初始化，正在持续编码流
|
||
enum PortalStage {
|
||
WaitingForFormat,
|
||
Streaming,
|
||
}
|
||
|
||
/// 编码线程单帧计时回执——由 `encode_thread_loop` 通过 `timing_tx` 发回主线程，
|
||
/// 用于在 `PipelineStats` 中窗口化统计 `sws_us`（libswscale 缩放开销）和
|
||
/// `encode_us`（H.264 软件编码开销）。类比 Go 的 `type EncodeThreadTiming struct`。
|
||
struct EncodeThreadTiming {
|
||
sws_us: u64,
|
||
encode_us: u64,
|
||
output_bytes: usize,
|
||
}
|
||
|
||
/// 编码工作线程的句柄与通信端点。
|
||
///
|
||
/// 由主线程持有，负责把 NV12 帧 (`CpuNv12Frame`) 通过 `input_tx` 投递给
|
||
/// `encode_thread_loop`；编码完成后通过 `timing_rx` 收回单帧计时；`duplicate_count`
|
||
/// 是跨线程共享的 `Arc<AtomicU64>`（类比 Go 的 `*uint64` protected by atomic），
|
||
/// 用于统计被去重跳过的帧数（影响 BWE 与丢弃策略）。
|
||
///
|
||
/// 字段全部用 `Option<...>`/`Sender`/`Receiver` 包装，是为了在 `shutdown` 时
|
||
/// 能用 `Option::take()` 把所有权转移到本地变量、显式 drop `input_tx`、再 `join()`。
|
||
struct EncodeThread {
|
||
handle: Option<std::thread::JoinHandle<()>>,
|
||
input_tx: crossbeam_channel::Sender<CpuNv12Frame>,
|
||
timing_rx: crossbeam_channel::Receiver<EncodeThreadTiming>,
|
||
duplicate_count: std::sync::Arc<std::sync::atomic::AtomicU64>,
|
||
}
|
||
|
||
/// WebRTC 工作线程的句柄与单向上行通道。
|
||
///
|
||
/// 主线程只能通过 `sent_gap_rx` **被动接收** WebRTC 线程上报的"已发送帧间隔/老化"
|
||
/// 指标（用于 `PipelineStats::record_send_from_thread`）。下行的码率 / 分辨率 / 暂停
|
||
/// 控制走另一组 channel（`bitrate_tx` / `resolution_tx` / `webrtc_paused`），不在此处。
|
||
struct WebrtcThread {
|
||
handle: Option<std::thread::JoinHandle<()>>,
|
||
sent_gap_rx: crossbeam_channel::Receiver<(f64, Option<f64>)>,
|
||
}
|
||
|
||
/// 门户模式的主状态机
|
||
///
|
||
/// 负责管理从 PipeWire 采集屏幕帧、通过 VAAPI 硬件编码的完整生命周期。
|
||
/// 工作流程：等待第一帧 → 创建编码器 → 持续编码帧数据。
|
||
pub struct StatePortal {
|
||
stage: PortalStage, // 当前采集阶段（等待首帧 / 流式编码中）
|
||
enc: Option<SwEncState>, // 软件编码器，首帧到达后初始化
|
||
enc_import: Option<SwEncImport>,
|
||
enc_thread: Option<EncodeThread>,
|
||
cap: CapPortal, // PipeWire 屏幕采集端点
|
||
args: Args, // 用户命令行参数
|
||
errored: bool, // 是否遇到不可恢复的错误
|
||
drm_device: Option<PathBuf>, // DRM 渲染设备路径（可自动检测）
|
||
frames_encoded: u64, // 已编码帧数（用于 PTS 编号）
|
||
start_time: Option<Instant>, // 编码开始时间
|
||
stats: PipelineStats, // 管道统计（窗口化帧计时 + 每秒快照）
|
||
pw_dropped_prev: u64, // 上一窗口的 PipeWire 丢弃帧数（用于增量计算）
|
||
webrtc: Option<WebRtcState>,
|
||
webrtc_thread: Option<WebrtcThread>,
|
||
webrtc_paused: Option<Arc<AtomicBool>>,
|
||
last_capture_arrival: Option<Instant>, // timestamp of last real frame arrival
|
||
idle_log_start: Option<Instant>, // when current idle period began (one-shot DEBUG log guard)
|
||
shutdown_started: bool, // idempotency guard; plain bool because &mut self is exclusive (not AtomicBool)
|
||
// Issue #24: real-capture PTS origin/tracking for WebRTC RTP timestamps.
|
||
first_pts_ns: Option<i128>,
|
||
capture_start: Option<Instant>,
|
||
last_pts_emitted: Option<i64>,
|
||
}
|
||
|
||
// `impl StatePortal` 块集中了门户路径的所有主线程逻辑：
|
||
// - `new`：构造（DRM 设备探测 + CapPortal 初始化；编码器延后到首帧）。
|
||
// - `poll_and_encode`：外层 main 循环每轮调用一次，处理 1 个 PipeWire 帧 / 控制事件。
|
||
// - `shutdown`：幂等清理（编码线程 → WebRTC 线程 → MP4 flush）。
|
||
// - 私有辅助：`record_capture_timeout` / `record_frame_arrival`（采集空闲日志节流）、
|
||
// `resolve_drm_device_for_frame`（DMA-BUF 导入兼容性探测）、
|
||
// `handle_pw_frame`（VAAPI 导入 + 软件编码）、`compute_capture_pts`（90kHz RTP PTS）。
|
||
// 内部不使用任何锁——所有 `&mut self` 由外层 main 循环单线程串行化保证独占。
|
||
impl StatePortal {
|
||
/// 创建门户状态实例
|
||
///
|
||
/// 初始化 DRM 设备路径和 PipeWire 采集端点，编码器延迟到第一帧到达时创建。
|
||
pub fn new(args: Args) -> Result<Self> {
|
||
let drm_device = resolve_drm_device(&args)?;
|
||
if let Some(ref drm_device) = drm_device {
|
||
tracing::info!("Using DRM device: {}", drm_device.display());
|
||
} else {
|
||
tracing::info!("DRM device auto-detection enabled");
|
||
}
|
||
|
||
let cap = CapPortal::new(&args)?;
|
||
|
||
let (webrtc, webrtc_paused) = if args.port > 0 {
|
||
let wrtc = WebRtcState::new(args.port, args.fps)?;
|
||
let paused = Arc::new(AtomicBool::new(true));
|
||
(Some(wrtc), Some(paused))
|
||
} else {
|
||
(None, None)
|
||
};
|
||
|
||
Ok(Self {
|
||
stage: PortalStage::WaitingForFormat,
|
||
enc: None,
|
||
enc_import: None,
|
||
enc_thread: None,
|
||
cap,
|
||
args,
|
||
errored: false,
|
||
drm_device,
|
||
frames_encoded: 0,
|
||
start_time: None,
|
||
stats: PipelineStats::new(),
|
||
pw_dropped_prev: 0,
|
||
webrtc,
|
||
webrtc_thread: None,
|
||
webrtc_paused,
|
||
last_capture_arrival: None,
|
||
idle_log_start: None,
|
||
shutdown_started: false,
|
||
first_pts_ns: None,
|
||
capture_start: None,
|
||
last_pts_emitted: None,
|
||
})
|
||
}
|
||
|
||
/// 轮询 PipeWire 事件并编码帧
|
||
///
|
||
/// `block=true` 时使用 recv_timeout 阻塞等待帧（最多 2ms），
|
||
/// `block=false` 时使用 try_recv 非阻塞检查。
|
||
/// 返回 `Ok(true)` 表示已处理事件，`Ok(false)` 表示暂无数据。
|
||
pub fn poll_and_encode(&mut self, block: bool) -> Result<bool> {
|
||
// 检查 PipeWire 控制事件（流结束 / 错误）
|
||
if let Ok(ctrl) = self.cap.event_receiver().try_recv() {
|
||
match ctrl {
|
||
PwCtrlEvent::StreamEnded => {
|
||
tracing::warn!("PipeWire stream ended");
|
||
self.errored = true;
|
||
return Ok(true);
|
||
}
|
||
PwCtrlEvent::Error(e) => {
|
||
tracing::error!("PipeWire error: {e}");
|
||
self.errored = true;
|
||
return Ok(true);
|
||
}
|
||
PwCtrlEvent::FormatChanged { width, height } => {
|
||
tracing::warn!(
|
||
"PipeWire format renegotiation: new dimensions {}x{} — encoder output remains at original resolution",
|
||
width,
|
||
height
|
||
);
|
||
// No action yet — VAAPI import/scale handles the conversion.
|
||
// Full encoder reinit is a future enhancement.
|
||
}
|
||
}
|
||
}
|
||
|
||
// 根据阻塞模式选择不同的帧接收策略
|
||
let frame = if block {
|
||
// 阻塞模式：最多等待 2ms 接收帧
|
||
match self
|
||
.cap
|
||
.frame_receiver()
|
||
.recv_timeout(std::time::Duration::from_millis(2))
|
||
{
|
||
Ok(frame) => frame,
|
||
Err(_) => {
|
||
self.record_capture_timeout();
|
||
return Ok(false);
|
||
}
|
||
}
|
||
} else {
|
||
// 非阻塞模式：立即尝试接收，无数据则返回
|
||
match self.cap.frame_receiver().try_recv() {
|
||
Ok(frame) => frame,
|
||
Err(_) => {
|
||
self.record_capture_timeout();
|
||
return Ok(false);
|
||
}
|
||
}
|
||
};
|
||
self.record_frame_arrival();
|
||
|
||
match self.stage {
|
||
PortalStage::WaitingForFormat => {
|
||
tracing::info!(
|
||
"First DMA-BUF frame: {}x{} format=0x{:08X} stride={} modifier=0x{:X}",
|
||
frame.width,
|
||
frame.height,
|
||
frame.format,
|
||
frame.stride,
|
||
frame.modifier
|
||
);
|
||
|
||
// 自动检测或确认 DRM 设备是否支持导入该帧
|
||
let drm_path = self.resolve_drm_device_for_frame(&frame)?;
|
||
// 计算编码目标分辨率（不超过 2560x1440）
|
||
let (enc_width, enc_height) = portal_encode_dimensions(frame.width, frame.height);
|
||
tracing::info!(
|
||
"Portal software encode target: {}x{} -> {}x{} @ {} fps",
|
||
frame.width,
|
||
frame.height,
|
||
enc_width,
|
||
enc_height,
|
||
self.args.fps,
|
||
);
|
||
// 码率:WebRTC 模式用保守默认(BWE 连接后立即覆盖),MP4 用公式
|
||
let actual_bitrate = self.args.bitrate.unwrap_or_else(|| {
|
||
if self.webrtc.is_some() {
|
||
webrtc_startup_bitrate_bps(enc_width, enc_height)
|
||
} else {
|
||
5 * (enc_width as u64) * (enc_height as u64) * (self.args.fps as u64) / 100
|
||
}
|
||
});
|
||
// GOP 大小：WebRTC 模式使用较大的 GOP（fps*2，最低20），MP4 模式使用 fps
|
||
let actual_gop_size = self.args.gop_size.unwrap_or_else(|| {
|
||
if self.webrtc.is_some() {
|
||
(self.args.fps * 2).max(20)
|
||
} else {
|
||
self.args.fps
|
||
}
|
||
});
|
||
|
||
// 根据是否启用 WebRTC 选择不同的编码器构造方式
|
||
if self.webrtc.is_some() {
|
||
let paused = self.webrtc_paused.as_ref()
|
||
.ok_or_else(|| anyhow::anyhow!("internal invariant broken: webrtc_paused missing while WebRTC mode is active"))?;
|
||
// WebRTC 模式需要 6 路 crossbeam channel 协调主线程 ↔ 编码线程 ↔ WebRTC 线程。
|
||
// `crossbeam_channel::bounded::<T>(n)` 类比 Go 的 `make(chan T, n)`——
|
||
// 容量满时 `send` 阻塞、空时 `recv` 阻塞；返回的 `(Sender, Receiver)` 各占一份
|
||
//所有权，可 move 到不同线程（前提是元素类型 `T: Send`）。
|
||
let (resolution_tx, resolution_rx) =
|
||
crossbeam_channel::bounded::<BitrateCommand>(4);
|
||
let (encoder_resolution_tx, encoder_resolution_rx) =
|
||
crossbeam_channel::bounded::<ResolutionChange>(4);
|
||
let import = SwEncImport::new_with_resolution_control(
|
||
&drm_path,
|
||
frame.width,
|
||
frame.height,
|
||
enc_width,
|
||
enc_height,
|
||
self.args.fps,
|
||
resolution_rx,
|
||
encoder_resolution_tx,
|
||
)?;
|
||
let (webrtc_tx, webrtc_rx) = crossbeam_channel::bounded(2);
|
||
let (input_tx, input_rx) = crossbeam_channel::bounded::<CpuNv12Frame>(1);
|
||
let (timing_tx, timing_rx) =
|
||
crossbeam_channel::bounded::<EncodeThreadTiming>(32);
|
||
let (bitrate_tx, bitrate_rx) = crossbeam_channel::bounded::<BitrateCommand>(4);
|
||
let encode = SwEncEncode::new_webrtc(
|
||
enc_width,
|
||
enc_height,
|
||
self.args.fps,
|
||
actual_bitrate,
|
||
actual_gop_size,
|
||
webrtc_tx,
|
||
paused.clone(),
|
||
bitrate_rx,
|
||
encoder_resolution_rx,
|
||
)?;
|
||
let duplicate_count = std::sync::Arc::new(
|
||
std::sync::atomic::AtomicU64::new(0),
|
||
);
|
||
// Arc 引用计数克隆（不是深拷贝）——`duplicate_count` 留在主线程，
|
||
// `duplicate_count_for_thread` move 进编码线程；两者指向同一原子。
|
||
// 类比 Go 的 `*uint64` + atomic.Store，但 Rust 用类型系统保证线程安全。
|
||
let duplicate_count_for_thread = duplicate_count.clone();
|
||
// `std::thread::Builder::new().name(...).spawn(move || {...})?`：
|
||
// - 类比 Go 的 `go func() {...}()`，但返回 `JoinHandle<T>` 而非 fire-and-forget——
|
||
// 主线程可在 shutdown 时 `handle.join()` 等待子线程退出。
|
||
// - **不**用 `tokio::spawn`：编码是 CPU 密集 + 阻塞 FFmpeg 调用，
|
||
// 不需要 async/await；标准线程更直接。
|
||
// - `move ||` 闭包：把 `encode` / `input_rx` / `timing_tx` /
|
||
// `duplicate_count_for_thread` 的所有权**转移**给子线程（类比 Go 里把变量
|
||
// 显式传入 goroutine 闭包参数）。
|
||
// - `?` 传播 `io::Error`——线程创建可能失败（资源限制）。
|
||
let handle = std::thread::Builder::new()
|
||
.name("wl-webrtc-encode".into())
|
||
.spawn(move || {
|
||
encode_thread_loop(
|
||
encode,
|
||
input_rx,
|
||
timing_tx,
|
||
duplicate_count_for_thread,
|
||
)
|
||
})?;
|
||
self.enc_import = Some(import);
|
||
self.enc_thread = Some(EncodeThread {
|
||
handle: Some(handle),
|
||
input_tx,
|
||
timing_rx,
|
||
duplicate_count,
|
||
});
|
||
|
||
let wrtc = self.webrtc.take().ok_or_else(|| {
|
||
anyhow::anyhow!("internal: WebRtcState missing during init")
|
||
})?;
|
||
let paused = self
|
||
.webrtc_paused
|
||
.as_ref()
|
||
.ok_or_else(|| anyhow::anyhow!("internal: webrtc_paused missing"))?
|
||
.clone();
|
||
let fps = self.args.fps;
|
||
let max_bitrate = self.args.max_bitrate;
|
||
let (sent_gap_tx, sent_gap_rx) =
|
||
crossbeam_channel::bounded::<(f64, Option<f64>)>(64);
|
||
// WebRTC 工作线程：同上 `std::thread::spawn(move || ...)` 模式——
|
||
// 内部跑 str0m 的 asyncio loop（`WebRtcState` 自己驱动），
|
||
// 通过 `webrtc_rx` 接收 H.264 帧、通过 `bitrate_tx` / `resolution_tx`
|
||
// 接收码率/分辨率指令、通过 `sent_gap_tx` 上报发送指标。
|
||
let webrtc_handle = std::thread::Builder::new()
|
||
.name("wl-webrtc-webrtc".into())
|
||
.spawn(move || {
|
||
webrtc_thread_loop(
|
||
wrtc,
|
||
webrtc_rx,
|
||
fps,
|
||
enc_width,
|
||
enc_height,
|
||
max_bitrate,
|
||
paused,
|
||
sent_gap_tx,
|
||
bitrate_tx,
|
||
resolution_tx,
|
||
)
|
||
})?;
|
||
self.webrtc_thread = Some(WebrtcThread {
|
||
handle: Some(webrtc_handle),
|
||
sent_gap_rx,
|
||
});
|
||
} else {
|
||
// MP4 模式：编码输出写入文件
|
||
let output_path = self.args.output.as_deref()
|
||
.ok_or_else(|| anyhow::anyhow!("--output is required in MP4 file output mode; use --port > 0 for WebRTC mode"))?;
|
||
let enc = avhw::SwEncState::new(
|
||
&drm_path,
|
||
std::path::Path::new(output_path),
|
||
frame.width,
|
||
frame.height,
|
||
enc_width,
|
||
enc_height,
|
||
self.args.fps,
|
||
actual_bitrate,
|
||
actual_gop_size,
|
||
)?;
|
||
self.enc = Some(enc);
|
||
};
|
||
self.stage = PortalStage::Streaming; // 切换到流式编码阶段
|
||
self.start_time = Some(Instant::now());
|
||
tracing::info!(
|
||
"First frame processed, encoder initialized, transitioning to Streaming"
|
||
);
|
||
drop(frame); // 首帧仅用于初始化，不参与编码
|
||
}
|
||
PortalStage::Streaming => {
|
||
// 记录采集帧到达（用于 capture gap 和 capture_fps 统计）
|
||
self.stats.record_capture();
|
||
self.last_capture_arrival = Some(Instant::now());
|
||
// 流式编码阶段：直接处理帧
|
||
self.handle_pw_frame(frame)?;
|
||
}
|
||
}
|
||
|
||
// 每秒输出一次结构化管道统计（仅 --stats 启用时记录日志）
|
||
if self.args.stats && self.stats.should_snapshot() {
|
||
self.stats.set_pipewire_dropped(0, 0);
|
||
self.stats.set_queue_depths(0, 0);
|
||
if let Some(ref enc_thread) = self.enc_thread {
|
||
while let Ok(timing) = enc_thread.timing_rx.try_recv() {
|
||
self.stats.record_encode_thread(
|
||
timing.sws_us,
|
||
timing.encode_us,
|
||
timing.output_bytes,
|
||
);
|
||
}
|
||
// Read duplicate counter (delta computed in setter)
|
||
let total = enc_thread
|
||
.duplicate_count
|
||
.load(std::sync::atomic::Ordering::Relaxed);
|
||
self.stats.set_duplicate_frames_skipped(total);
|
||
}
|
||
if let Some(ref webrtc_thread) = self.webrtc_thread {
|
||
while let Ok((gap_ms, age_ms)) = webrtc_thread.sent_gap_rx.try_recv() {
|
||
self.stats.record_send_from_thread(gap_ms, age_ms);
|
||
}
|
||
}
|
||
let snap = self.stats.snapshot_and_reset();
|
||
tracing::info!("stats: {snap}");
|
||
}
|
||
|
||
Ok(true)
|
||
}
|
||
|
||
/// 记录"采集超时"——本次轮询未取到帧（PipeWire 队列空）。
|
||
///
|
||
/// 因为 Wayland 是 damage-driven（只有画面变化才推帧），静态画面下长时间无帧
|
||
/// 是**正常**行为，不是 compositor 卡死。所以本函数只做"5 秒阈值后的 DEBUG 一次性日志"，
|
||
/// 用 `idle_log_start` 字段保证每次空闲区间只发一条日志（issue #15 / #18）。
|
||
fn record_capture_timeout(&mut self) {
|
||
let Some(last_capture_arrival) = self.last_capture_arrival else {
|
||
return;
|
||
};
|
||
|
||
let now = Instant::now();
|
||
// Wayland damage-driven delivery: static content means no new frames.
|
||
// This is normal Wayland behavior, not a compositor hang. Only log DEBUG
|
||
// after a meaningful idle period, and only once per idle episode.
|
||
// See issues #15 and #18.
|
||
const CAPTURE_IDLE_LOG_THRESHOLD: Duration = Duration::from_secs(5);
|
||
if now.duration_since(last_capture_arrival) <= CAPTURE_IDLE_LOG_THRESHOLD {
|
||
return;
|
||
}
|
||
|
||
if self.idle_log_start.is_none() {
|
||
// Use last_capture_arrival as idle start for accurate elapsed duration.
|
||
self.idle_log_start = Some(last_capture_arrival);
|
||
tracing::debug!(
|
||
elapsed_ms = now.duration_since(last_capture_arrival).as_millis(),
|
||
"portal capture idle; no damage frames received (normal Wayland behavior)"
|
||
);
|
||
}
|
||
}
|
||
|
||
/// 记录"采集到达"——本次轮询成功取到一帧。
|
||
///
|
||
/// 与 `record_capture_timeout` 互补：若之前处于空闲区间，则通过 `Option::take()`
|
||
/// 取出 `idle_log_start` 并发一条 "resumed after idle" DEBUG 日志；然后刷新
|
||
/// `last_capture_arrival` 时间戳。两者共同实现"一次性空闲日志"语义。
|
||
fn record_frame_arrival(&mut self) {
|
||
if let Some(idle_start) = self.idle_log_start.take() {
|
||
tracing::debug!(
|
||
idle_ms = idle_start.elapsed().as_millis(),
|
||
"portal capture resumed after idle period"
|
||
);
|
||
}
|
||
self.last_capture_arrival = Some(Instant::now());
|
||
}
|
||
|
||
/// 为当前帧解析可用的 DRM 渲染设备
|
||
///
|
||
/// 如果用户已通过 `--drm-device` 指定设备，直接返回；
|
||
/// 否则遍历系统中所有 DRM render node，逐个尝试导入 DMA-BUF 帧来找到兼容设备。
|
||
fn resolve_drm_device_for_frame(&mut self, frame: &PwDmaBufFrame) -> Result<PathBuf> {
|
||
// 用户已显式指定 DRM 设备，直接使用
|
||
if let Some(ref drm) = self.drm_device {
|
||
return Ok(drm.clone());
|
||
}
|
||
|
||
// 查找系统中所有 DRM render node（如 /dev/dri/renderD128）
|
||
let candidates = crate::state::find_drm_render_nodes();
|
||
if candidates.is_empty() {
|
||
bail!("No DRM render device found. Specify --drm-device.");
|
||
}
|
||
|
||
// 逐个尝试导入 DMA-BUF 帧，找到第一个兼容的设备
|
||
let mut failures = Vec::new();
|
||
for candidate in &candidates {
|
||
match crate::avhw::test_dma_buf_import(candidate, frame) {
|
||
Ok(()) => {
|
||
// 成功导入，缓存检测结果并返回
|
||
tracing::info!(
|
||
"Auto-detected DRM device: {} (tested {} candidates)",
|
||
candidate.display(),
|
||
candidates.len(),
|
||
);
|
||
self.drm_device = Some(candidate.clone());
|
||
return Ok(candidate.clone());
|
||
}
|
||
Err(e) => {
|
||
// 导入失败，记录原因，继续尝试下一个设备
|
||
tracing::debug!(
|
||
"DRM device {} cannot import DMA-BUF: {e}",
|
||
candidate.display(),
|
||
);
|
||
failures.push((candidate, e));
|
||
}
|
||
}
|
||
}
|
||
|
||
// 所有候选设备均失败，返回详细错误信息
|
||
bail!(failures
|
||
.into_iter()
|
||
.map(|(p, e)| format!("{} ({e})", p.display()))
|
||
.collect::<Vec<_>>()
|
||
.join(", "));
|
||
}
|
||
|
||
/// 处理单帧 DMA-BUF 数据
|
||
///
|
||
/// 通过 `av_hwframe_map` 零拷贝导入 VAAPI，然后交给 SwEncState 完成：
|
||
/// scale_vaapi GPU 缩放、2K NV12 回读、YUV420P 格式转换、软件 H.264 编码。
|
||
fn handle_pw_frame(&mut self, frame: PwDmaBufFrame) -> Result<()> {
|
||
// #19: When WebRTC mode is paused (no client connected), skip ALL frame
|
||
// processing — DMA-BUF import, VAAPI scale, NV12 clone, channel send, and
|
||
// encode thread wakeup. This eliminates ~60fps of pointless work during
|
||
// the pre-connect idle window. MP4 mode (webrtc_paused == None) is unaffected.
|
||
// `Arc<AtomicBool>` 类比 Go 的 `*atomic.Bool`——`Arc` 提供跨线程共享所有权
|
||
// （引用计数原子递增/递减），`AtomicBool` 提供无锁读/写。
|
||
// `Ordering::Relaxed`：只保证单变量原子性，不建立与其他变量的 happens-before 关系——
|
||
// 对"暂停标志"足够（不需要它做屏障同步）。
|
||
if let Some(paused) = &self.webrtc_paused {
|
||
if paused.load(Ordering::Relaxed) {
|
||
return Ok(());
|
||
}
|
||
}
|
||
let t_import_start = Instant::now();
|
||
// WebRTC: use real PipeWire capture time so RTP timestamps reflect reality
|
||
// (sequential counter caused client jitter buffers to grow to 2-3s under
|
||
// damage-driven variable fps — issue #24). MP4: keep sequential counter;
|
||
// file output doesn't need real-time PTS and changing it would alter
|
||
// playback speed during static periods.
|
||
let pts = if self.webrtc_thread.is_some() {
|
||
self.compute_capture_pts(frame.pts)
|
||
} else {
|
||
self.frames_encoded as i64
|
||
};
|
||
|
||
if let Some(enc) = self.enc.as_mut() {
|
||
// 将 DMA-BUF 帧零拷贝导入 VAAPI 硬件帧池
|
||
// unsafe：FFI 调用 FFmpeg `av_hwframe_ctx_init` / `av_hwframe_map` 系列，
|
||
// 内部会读取 `enc.frames_rgb()` 指向的 `AVBufferRef`（硬件帧池），
|
||
// 并把 `frame.fd.as_raw_fd()`（DMA-BUF dmabuf fd）注册到 VAAPI。
|
||
// 安全性前提：`enc` 在本线程独占（main 串行化保证）、`frame.fd` 未被 close。
|
||
let mut vaapi_frame = unsafe {
|
||
avhw::import_dma_buf_to_vaapi(
|
||
enc.frames_rgb().as_ptr(),
|
||
frame.fd.as_raw_fd(),
|
||
frame.width,
|
||
frame.height,
|
||
frame.format,
|
||
frame.modifier,
|
||
frame.stride,
|
||
frame.offset,
|
||
)
|
||
}?;
|
||
|
||
let import_us = t_import_start.elapsed().as_micros() as u64;
|
||
let t_encode_start = Instant::now();
|
||
|
||
// 设置帧的显示时间戳（PTS），基于已编码帧序号
|
||
unsafe {
|
||
(*vaapi_frame.as_mut_ptr()).pts = pts;
|
||
}
|
||
|
||
// 送入编码器完成：缩放 → 回读 → 格式转换 → H.264 编码
|
||
enc.encode_frame(&vaapi_frame)?;
|
||
let total_us = t_import_start.elapsed().as_micros() as u64;
|
||
let encode_us = t_encode_start.elapsed().as_micros() as u64;
|
||
|
||
self.frames_encoded += 1;
|
||
|
||
// 记录帧计时到管道统计（import + encode 内部各阶段暂不可分离，用 total 覆盖）
|
||
let timings = FrameTimings {
|
||
import_us,
|
||
encode_us,
|
||
total_us,
|
||
..Default::default()
|
||
};
|
||
self.stats.record_encode(&timings);
|
||
} else if let Some(import) = self.enc_import.as_mut() {
|
||
// 同上 unsafe：DMA-BUF → VAAPI 导入；`import.frames_rgb()` 是与编码线程
|
||
// **不共享**的独立硬件帧池（避免与 `import_and_scale` 的回读路径竞争）。
|
||
let mut vaapi_frame = unsafe {
|
||
avhw::import_dma_buf_to_vaapi(
|
||
import.frames_rgb().as_ptr(),
|
||
frame.fd.as_raw_fd(),
|
||
frame.width,
|
||
frame.height,
|
||
frame.format,
|
||
frame.modifier,
|
||
frame.stride,
|
||
frame.offset,
|
||
)
|
||
}?;
|
||
unsafe {
|
||
(*vaapi_frame.as_mut_ptr()).pts = pts;
|
||
}
|
||
|
||
let cpu_nv12 = import.import_and_scale(&vaapi_frame)?;
|
||
let import_us = t_import_start.elapsed().as_micros() as u64;
|
||
self.stats.record_import(import_us);
|
||
|
||
let enc_thread = self.enc_thread.as_ref().ok_or_else(|| {
|
||
anyhow::anyhow!(
|
||
"internal invariant broken: encode thread missing while async import is active"
|
||
)
|
||
})?;
|
||
// `try_send` 类比 Go 的 `select { case ch <- v: default: }`——
|
||
// 非阻塞投递；三种结果分别处理：成功递增、满了丢弃（DEBUG 日志）、
|
||
// 对端关闭（致命，置 `errored=true` 让外层循环退出）。
|
||
match enc_thread.input_tx.try_send(cpu_nv12) {
|
||
Ok(()) => {
|
||
self.frames_encoded += 1;
|
||
}
|
||
Err(crossbeam_channel::TrySendError::Full(_)) => {
|
||
tracing::debug!("Encode thread input full, dropping portal frame");
|
||
}
|
||
Err(crossbeam_channel::TrySendError::Disconnected(_frame)) => {
|
||
tracing::error!("Encode thread input disconnected");
|
||
self.errored = true;
|
||
}
|
||
}
|
||
} else {
|
||
bail!("encoder not initialized");
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// Compute PTS in 90kHz media-clock ticks from PipeWire's nanosecond
|
||
/// capture timestamp. Falls back to `Instant`-based elapsed time when PipeWire
|
||
/// does not provide PTS. Maintains strict monotonicity (encoder requirement).
|
||
fn compute_capture_pts(&mut self, pw_pts_ns: i64) -> i64 {
|
||
const NS_PER_SEC: i128 = 1_000_000_000;
|
||
|
||
let raw_ns: i128 = if pw_pts_ns > 0 {
|
||
i128::from(pw_pts_ns)
|
||
} else {
|
||
let start = self.capture_start.get_or_insert_with(Instant::now);
|
||
i128::try_from(start.elapsed().as_nanos()).unwrap_or(0)
|
||
};
|
||
|
||
if self.first_pts_ns.is_none() && raw_ns > 0 {
|
||
self.first_pts_ns = Some(raw_ns);
|
||
}
|
||
|
||
let origin = self.first_pts_ns.unwrap_or(0);
|
||
let relative_ns = if raw_ns >= origin {
|
||
raw_ns - origin
|
||
} else {
|
||
// PipeWire PTS went backwards (stream restart) — reset origin.
|
||
self.first_pts_ns = Some(raw_ns);
|
||
0
|
||
};
|
||
let ticks_i128 =
|
||
(relative_ns.saturating_mul(crate::avhw::WEBRTC_RTP_CLOCK_HZ)) / NS_PER_SEC;
|
||
let computed_pts = i64::try_from(ticks_i128).unwrap_or(i64::MAX);
|
||
let mut pts = computed_pts;
|
||
|
||
if let Some(last) = self.last_pts_emitted {
|
||
if pts <= last {
|
||
pts = last.checked_add(1).unwrap_or(last);
|
||
}
|
||
}
|
||
self.last_pts_emitted = Some(pts);
|
||
pts
|
||
}
|
||
|
||
/// 关闭状态：刷新编码器并清理资源（幂等）。
|
||
///
|
||
/// `shutdown_started` 守卫在清理之前置位——防止 panic 时 `Drop` 重入 unwinding。
|
||
pub fn shutdown(&mut self) {
|
||
if self.shutdown_started {
|
||
return;
|
||
}
|
||
self.shutdown_started = true;
|
||
|
||
// 1. Stop encode thread (drops webrtc_tx → signals WebRTC thread to exit)
|
||
// `Option::take()` 把 `EncodeThread` 的所有权从 `self.enc_thread` 转移到本地 `enc_thread`，
|
||
// 同时 `self.enc_thread` 变成 `None`——这是 Rust 里"消费字段但保留父结构体"的标准习语，
|
||
// 类比 Go 里把字段设为 nil 但保留外层 struct。接下来显式 `drop(input_tx)` 关闭 channel，
|
||
// 编码线程的 `input_rx.recv()` 会返回 `Err(Disconnected)` 从而退出循环。
|
||
if let Some(mut enc_thread) = self.enc_thread.take() {
|
||
drop(enc_thread.input_tx);
|
||
if let Some(handle) = enc_thread.handle.take() {
|
||
if handle.join().is_err() {
|
||
tracing::error!("Encode thread panicked during shutdown");
|
||
}
|
||
}
|
||
}
|
||
self.enc_import = None;
|
||
// 2. Wait for WebRTC thread (exits when webrtc_tx is dropped by encode thread)
|
||
if let Some(mut webrtc_thread) = self.webrtc_thread.take() {
|
||
if let Some(handle) = webrtc_thread.handle.take() {
|
||
if handle.join().is_err() {
|
||
tracing::error!("WebRTC thread panicked during shutdown");
|
||
}
|
||
}
|
||
}
|
||
// 3. Flush MP4 encoder if present
|
||
if let Some(mut enc) = self.enc.take() {
|
||
if let Err(e) = enc.flush() {
|
||
tracing::error!("Flush error during shutdown: {e}");
|
||
}
|
||
}
|
||
if let Some(start) = self.start_time {
|
||
if self.frames_encoded > 0 {
|
||
let elapsed = start.elapsed().as_secs_f64();
|
||
let fps = self.frames_encoded as f64 / elapsed;
|
||
tracing::info!(
|
||
"Total: {} frames in {:.1}s, avg {:.1}fps",
|
||
self.frames_encoded,
|
||
elapsed,
|
||
fps,
|
||
);
|
||
}
|
||
}
|
||
tracing::info!("StatePortal shutdown complete");
|
||
}
|
||
|
||
/// 返回是否遇到不可恢复的错误
|
||
pub fn is_errored(&self) -> bool {
|
||
self.errored
|
||
}
|
||
}
|
||
|
||
// === 编码线程主循环（独立 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<CpuNv12Frame>,
|
||
timing_tx: crossbeam_channel::Sender<EncodeThreadTiming>,
|
||
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 {
|
||
match input_rx.recv() {
|
||
Ok(frame) => {
|
||
match encode.encode_cpu_frame(&frame) {
|
||
Ok(EncodeOutcome::Encoded) => {
|
||
let t = encode.take_timing();
|
||
let _ = timing_tx.try_send(EncodeThreadTiming {
|
||
sws_us: t.sws_us,
|
||
encode_us: t.encode_us,
|
||
output_bytes: t.output_bytes,
|
||
});
|
||
}
|
||
Ok(EncodeOutcome::SkippedDuplicate) => {
|
||
duplicate_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||
}
|
||
Ok(_) => {
|
||
// SkippedPaused / SkippedDisconnected — no counter needed
|
||
}
|
||
Err(e) => {
|
||
tracing::error!("Encode thread error: {e}");
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
Err(_) => {
|
||
tracing::info!("Encode thread input closed, flushing encoder");
|
||
if let Err(e) = encode.flush() {
|
||
tracing::error!("Encode thread flush error: {e}");
|
||
}
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
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(
|
||
mut wrtc: WebRtcState,
|
||
webrtc_rx: crossbeam_channel::Receiver<EncodedH264Frame>,
|
||
fps: u32,
|
||
enc_width: u32,
|
||
enc_height: u32,
|
||
max_bitrate: u64,
|
||
paused: Arc<AtomicBool>,
|
||
sent_gap_tx: crossbeam_channel::Sender<(f64, Option<f64>)>,
|
||
bitrate_tx: crossbeam_channel::Sender<BitrateCommand>,
|
||
resolution_tx: crossbeam_channel::Sender<BitrateCommand>,
|
||
) {
|
||
let mut frames_sent: u64 = 0;
|
||
let mut last_send: Option<std::time::Instant> = None;
|
||
let mut last_sent_bitrate: Option<u64> = None;
|
||
let initial_tier = (enc_width, enc_height);
|
||
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 {
|
||
if let Err(e) = wrtc.handle_signaling() {
|
||
tracing::error!("WebRTC signaling error: {e}");
|
||
break;
|
||
}
|
||
if let Err(e) = wrtc.poll_and_feed() {
|
||
tracing::error!("WebRTC poll error: {e}");
|
||
break;
|
||
}
|
||
|
||
if wrtc.take_force_keyframe() {
|
||
let _ = bitrate_tx.try_send(BitrateCommand::ForceKeyframe);
|
||
}
|
||
|
||
let connected = wrtc.is_connected();
|
||
// Arc<AtomicBool> 跨线程协调：编码线程 Relaxed 读 paused；本线程 Relaxed 写。
|
||
// Relaxed 取舍：暂停标志无内存序需求（不保护其他共享数据），只需原子可见性。
|
||
let was_paused = paused.load(Ordering::Relaxed);
|
||
let now_paused = !connected;
|
||
if was_paused && !now_paused {
|
||
tracing::info!("WebRTC client connected, resuming encoding");
|
||
} else if !was_paused && now_paused {
|
||
tracing::warn!("WebRTC client disconnected, pausing encoding");
|
||
}
|
||
paused.store(now_paused, Ordering::Relaxed);
|
||
|
||
if let Some(bwe) = wrtc.get_bwe_estimate() {
|
||
// #23: Cap BWE to prevent runaway bitrate escalation. Without this, BWE
|
||
// estimates can rise to 10+ Mbps, causing IDR bursts and PLI storms.
|
||
let effective_bwe = bwe.min(max_bitrate);
|
||
if effective_bwe != bwe {
|
||
tracing::debug!(
|
||
bwe,
|
||
effective_bwe,
|
||
max_bitrate,
|
||
"BWE exceeds --max-bitrate cap, clamping"
|
||
);
|
||
}
|
||
let bwe = effective_bwe;
|
||
|
||
let should_send = match last_sent_bitrate {
|
||
None => true,
|
||
Some(last) => {
|
||
let diff = if bwe > last { bwe - last } else { last - bwe };
|
||
diff * 10 > last
|
||
}
|
||
};
|
||
if should_send {
|
||
let _ = bitrate_tx.try_send(BitrateCommand::UpdateBitrate { target_bps: bwe });
|
||
last_sent_bitrate = Some(bwe);
|
||
}
|
||
|
||
if last_resolution_eval.elapsed() >= Duration::from_secs(1) {
|
||
last_resolution_eval = Instant::now();
|
||
let selected = select_resolution(current_tier.0, current_tier.1, bwe, fps);
|
||
if selected != current_tier {
|
||
current_tier = selected;
|
||
upscale_counter = 0;
|
||
let _ = resolution_tx.try_send(BitrateCommand::UpdateResolution {
|
||
width: current_tier.0,
|
||
height: current_tier.1,
|
||
});
|
||
wrtc.set_need_keyframe();
|
||
} else if let Some(next_tier) = next_upscale_tier(current_tier, initial_tier) {
|
||
let needed = resolution_bitrate_bps(next_tier.0, next_tier.1, fps);
|
||
if bwe > needed.saturating_mul(120) / 100 {
|
||
upscale_counter = upscale_counter.saturating_add(1);
|
||
if upscale_counter >= 10 {
|
||
current_tier = next_tier;
|
||
upscale_counter = 0;
|
||
let _ = resolution_tx.try_send(BitrateCommand::UpdateResolution {
|
||
width: current_tier.0,
|
||
height: current_tier.1,
|
||
});
|
||
wrtc.set_need_keyframe();
|
||
}
|
||
} else {
|
||
upscale_counter = 0;
|
||
}
|
||
} else {
|
||
upscale_counter = 0;
|
||
}
|
||
}
|
||
}
|
||
|
||
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}");
|
||
}
|
||
frames_sent = frames_sent.saturating_add(1);
|
||
let gap_ms = last_send
|
||
.map(|l| l.elapsed().as_secs_f64() * 1000.0)
|
||
.unwrap_or(0.0);
|
||
// Compute capture-to-send age on the sending thread so the
|
||
// frame_age stat stays accurate when batch-drained later.
|
||
let age_ms =
|
||
Some(enc_frame.capture_time.elapsed().as_secs_f64() * 1000.0);
|
||
last_send = Some(std::time::Instant::now());
|
||
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() {
|
||
if let Err(e) = wrtc.write_h264_frame(&enc_frame.data, enc_frame.pts_ticks)
|
||
{
|
||
tracing::debug!("WebRTC write frame error: {e}");
|
||
}
|
||
frames_sent = frames_sent.saturating_add(1);
|
||
let gap_ms = last_send
|
||
.map(|l| l.elapsed().as_secs_f64() * 1000.0)
|
||
.unwrap_or(0.0);
|
||
let age_ms =
|
||
Some(enc_frame.capture_time.elapsed().as_secs_f64() * 1000.0);
|
||
last_send = Some(std::time::Instant::now());
|
||
let _ = sent_gap_tx.try_send((gap_ms, age_ms));
|
||
}
|
||
}
|
||
Err(crossbeam_channel::RecvTimeoutError::Timeout) => {}
|
||
Err(crossbeam_channel::RecvTimeoutError::Disconnected) => {
|
||
tracing::info!("WebRTC channel disconnected, exiting thread");
|
||
return;
|
||
}
|
||
}
|
||
}
|
||
|
||
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.
|
||
fn webrtc_startup_bitrate_bps(width: u32, height: u32) -> u64 {
|
||
let pixels = u64::from(width) * u64::from(height);
|
||
if pixels <= 1_000_000 {
|
||
1_000_000
|
||
} else if pixels <= 2_500_000 {
|
||
2_000_000
|
||
} else if pixels <= 4_500_000 {
|
||
4_000_000
|
||
} else {
|
||
8_000_000
|
||
}
|
||
}
|
||
|
||
// 基于 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) {
|
||
let current = (current_w, current_h);
|
||
let current_bitrate = resolution_bitrate_bps(current_w, current_h, fps);
|
||
if bwe_bps >= current_bitrate.saturating_mul(60) / 100 {
|
||
return current;
|
||
}
|
||
|
||
// 在 RESOLUTION_TIERS 中找当前分辨率的位置；若不在表中（如 1366×768），
|
||
// 用 unwrap_or_else 退回到第一个宽高都不超过 current 的档位，最终兜底取最小档（720p）。
|
||
let current_index = RESOLUTION_TIERS
|
||
.iter()
|
||
.position(|&tier| tier == current)
|
||
.unwrap_or_else(|| {
|
||
RESOLUTION_TIERS
|
||
.iter()
|
||
.position(|&(w, h)| w <= current_w && h <= current_h)
|
||
.unwrap_or(RESOLUTION_TIERS.len() - 1)
|
||
});
|
||
let next_index = (current_index + 1).min(RESOLUTION_TIERS.len() - 1);
|
||
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)
|
||
}
|
||
|
||
impl Drop for StatePortal {
|
||
// 析构时自动调用 shutdown，确保编码器被刷新、资源被释放
|
||
fn drop(&mut self) {
|
||
self.shutdown();
|
||
}
|
||
}
|
||
|
||
/// 计算编码目标分辨率
|
||
///
|
||
/// 将原始分辨率等比缩放至不超过 2560×1440（2K），并确保宽高为偶数
|
||
/// （H.264 编码要求偶数尺寸）。
|
||
fn portal_encode_dimensions(width: u32, height: u32) -> (u32, u32) {
|
||
const TARGET_W: u32 = 2560; // 目标最大宽度
|
||
const TARGET_H: u32 = 1440; // 目标最大高度
|
||
|
||
// 原始分辨率已在 2K 以内，直接对齐偶数
|
||
if width <= TARGET_W && height <= TARGET_H {
|
||
return (width & !1, height & !1); // & !1 确保为偶数
|
||
}
|
||
|
||
// 按宽度限制等比缩放
|
||
let width_limited_h = ((height as u64) * (TARGET_W as u64) / (width as u64)) as u32;
|
||
if width_limited_h <= TARGET_H {
|
||
(TARGET_W & !1, width_limited_h & !1)
|
||
} else {
|
||
// 按高度限制等比缩放
|
||
let height_limited_w = ((width as u64) * (TARGET_H as u64) / (height as u64)) as u32;
|
||
(height_limited_w & !1, TARGET_H & !1)
|
||
}
|
||
}
|
||
|
||
/// 解析 DRM 渲染设备路径
|
||
///
|
||
/// 仅使用命令行指定的设备路径；未指定则在首帧到达时自动检测。
|
||
fn resolve_drm_device(args: &Args) -> Result<Option<PathBuf>> {
|
||
if let Some(ref drm) = args.drm_device {
|
||
return Ok(Some(PathBuf::from(drm)));
|
||
}
|
||
Ok(None)
|
||
}
|
||
|
||
/// 构建测试用的 AVDRMFrameDescriptor（仅测试用途）
|
||
///
|
||
/// 将 PwDmaBufFrame 转换为 FFmpeg 的 DRM 帧描述符结构体，
|
||
/// 用于验证 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 文件描述符
|
||
desc.objects[0].size = 0; // 大小设为 0（内核自动确定）
|
||
desc.objects[0].format_modifier = frame.modifier; // DRM 格式修饰符（如线性、tiled）
|
||
desc.nb_layers = 1; // 单层
|
||
desc.layers[0].format = frame.format; // 像素格式（如 XR24）
|
||
desc.layers[0].nb_planes = 1; // 单平面
|
||
desc.layers[0].planes[0].object_index = 0; // 指向第 0 个对象
|
||
desc.layers[0].planes[0].offset = frame.offset as isize; // 帧数据偏移
|
||
desc.layers[0].planes[0].pitch = frame.stride as isize; // 行跨度（stride）
|
||
desc
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use std::os::fd::{FromRawFd, OwnedFd};
|
||
|
||
/// 创建测试用的 DMA-BUF 帧数据（使用 stderr fd 的副本作为占位）
|
||
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,
|
||
offset: 0,
|
||
stride: 1920 * 4, // 每行 1920 像素 × 4 字节（XRGB）
|
||
modifier: 0, // DRM_FORMAT_MOD_LINEAR（线性布局）
|
||
width: 1920,
|
||
height: 1080,
|
||
format: 0x34325258, // XR24 little-endian（XRGB8888）
|
||
pts: 12345,
|
||
}
|
||
}
|
||
|
||
/// 测试 DRM 描述符构建（单平面情况）
|
||
#[test]
|
||
fn build_drm_descriptor_single_plane() {
|
||
let frame = make_test_frame();
|
||
let desc = build_drm_descriptor(&frame);
|
||
|
||
assert_eq!(desc.nb_objects, 1);
|
||
assert_eq!(desc.objects[0].format_modifier, 0);
|
||
assert_eq!(desc.nb_layers, 1);
|
||
assert_eq!(desc.layers[0].format, 0x34325258);
|
||
assert_eq!(desc.layers[0].nb_planes, 1);
|
||
assert_eq!(desc.layers[0].planes[0].object_index, 0);
|
||
assert_eq!(desc.layers[0].planes[0].offset, 0);
|
||
assert_eq!(desc.layers[0].planes[0].pitch, 1920 * 4);
|
||
}
|
||
|
||
/// 测试显式指定 DRM 设备时的解析
|
||
#[test]
|
||
fn resolve_drm_device_explicit() {
|
||
let args = Args {
|
||
output: Some("test.mp4".to_string()),
|
||
output_name: None,
|
||
fps: 30,
|
||
codec: "h264".to_string(),
|
||
hw_accel: "vaapi".to_string(),
|
||
drm_device: Some("/dev/dri/renderD128".to_string()),
|
||
bitrate: None,
|
||
max_bitrate: 8_000_000,
|
||
gop_size: None,
|
||
verbose: false,
|
||
backend: None,
|
||
port: 0,
|
||
no_persist: false,
|
||
stats: false,
|
||
};
|
||
let result = resolve_drm_device(&args).unwrap();
|
||
assert_eq!(
|
||
result,
|
||
Some(std::path::PathBuf::from("/dev/dri/renderD128"))
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn resolve_drm_device_none_when_not_specified() {
|
||
let args = Args {
|
||
output: Some("test.mp4".to_string()),
|
||
output_name: None,
|
||
fps: 30,
|
||
codec: "h264".to_string(),
|
||
hw_accel: "vaapi".to_string(),
|
||
drm_device: None,
|
||
bitrate: None,
|
||
max_bitrate: 8_000_000,
|
||
gop_size: None,
|
||
verbose: false,
|
||
backend: None,
|
||
port: 0,
|
||
no_persist: false,
|
||
stats: false,
|
||
};
|
||
let result = resolve_drm_device(&args).unwrap();
|
||
assert_eq!(result, None);
|
||
}
|
||
|
||
#[test]
|
||
fn webrtc_startup_bitrate_tiers_by_pixel_count() {
|
||
assert_eq!(webrtc_startup_bitrate_bps(1280, 720), 1_000_000);
|
||
assert_eq!(webrtc_startup_bitrate_bps(1920, 1080), 2_000_000);
|
||
assert_eq!(webrtc_startup_bitrate_bps(2560, 1440), 4_000_000);
|
||
assert_eq!(webrtc_startup_bitrate_bps(3840, 2160), 8_000_000);
|
||
}
|
||
|
||
#[test]
|
||
fn select_resolution_downscales_one_tier_below_sixty_percent() {
|
||
let fps = 30;
|
||
let current = resolution_bitrate_bps(1920, 1080, fps);
|
||
assert_eq!(
|
||
select_resolution(1920, 1080, current * 59 / 100, fps),
|
||
(1280, 720)
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn select_resolution_keeps_tier_at_sixty_percent() {
|
||
let fps = 30;
|
||
let current = resolution_bitrate_bps(1920, 1080, fps);
|
||
assert_eq!(
|
||
select_resolution(1920, 1080, current * 60 / 100, fps),
|
||
(1920, 1080)
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn select_resolution_never_goes_below_720p() {
|
||
assert_eq!(select_resolution(1280, 720, 1, 30), (1280, 720));
|
||
}
|
||
|
||
#[test]
|
||
fn next_upscale_tier_respects_initial_ceiling() {
|
||
assert_eq!(
|
||
next_upscale_tier((1280, 720), (1920, 1080)),
|
||
Some((1920, 1080))
|
||
);
|
||
assert_eq!(next_upscale_tier((1920, 1080), (1920, 1080)), None);
|
||
}
|
||
|
||
/// 测试：使用自定义偏移量和 stride 构建 DRM 描述符
|
||
#[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 字节
|
||
modifier: 0x0100000000000001, // AMD modifiers
|
||
width: 3840,
|
||
height: 2160,
|
||
format: 0x34325258,
|
||
pts: 0,
|
||
};
|
||
let desc = build_drm_descriptor(&frame);
|
||
|
||
assert_eq!(desc.nb_objects, 1);
|
||
assert_eq!(desc.objects[0].format_modifier, 0x0100000000000001);
|
||
assert_eq!(desc.layers[0].planes[0].offset, 4096);
|
||
assert_eq!(desc.layers[0].planes[0].pitch, 3840 * 4);
|
||
}
|
||
|
||
// ── issue #8 regression ──
|
||
|
||
#[test]
|
||
fn try_send_full_channel_returns_full_not_block() {
|
||
let (tx, rx) = crossbeam_channel::bounded::<Vec<u8>>(2);
|
||
tx.send(vec![1]).unwrap();
|
||
tx.send(vec![2]).unwrap();
|
||
|
||
assert!(matches!(
|
||
tx.try_send(vec![3]),
|
||
Err(crossbeam_channel::TrySendError::Full(_))
|
||
));
|
||
assert_eq!(rx.len(), 2);
|
||
}
|
||
|
||
#[test]
|
||
fn try_send_after_rx_dropped_returns_disconnected() {
|
||
let (tx, rx) = crossbeam_channel::bounded::<Vec<u8>>(2);
|
||
drop(rx);
|
||
|
||
assert!(matches!(
|
||
tx.try_send(vec![1]),
|
||
Err(crossbeam_channel::TrySendError::Disconnected(_))
|
||
));
|
||
}
|
||
|
||
// given: full bounded channel
|
||
// when: rx is dropped, then try_send
|
||
// expect: Disconnected, not blocking
|
||
#[test]
|
||
fn shutdown_rx_drop_prevents_deadlock_on_full_channel() {
|
||
let (tx, rx) = crossbeam_channel::bounded::<Vec<u8>>(2);
|
||
tx.send(vec![1]).unwrap();
|
||
tx.send(vec![2]).unwrap();
|
||
drop(rx);
|
||
|
||
assert!(matches!(
|
||
tx.try_send(vec![3]),
|
||
Err(crossbeam_channel::TrySendError::Disconnected(_))
|
||
));
|
||
}
|
||
|
||
// ── Task 7: Additional resolution tier edge cases ──
|
||
|
||
#[test]
|
||
fn select_resolution_keeps_720p_when_bwe_sufficient() {
|
||
let fps = 30;
|
||
let bitrate_720 = resolution_bitrate_bps(1280, 720, fps);
|
||
assert_eq!(
|
||
select_resolution(1280, 720, bitrate_720, fps),
|
||
(1280, 720)
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn select_resolution_downscales_1440p_to_1080p() {
|
||
let fps = 30;
|
||
let bitrate_1440 = resolution_bitrate_bps(2560, 1440, fps);
|
||
assert_eq!(
|
||
select_resolution(2560, 1440, bitrate_1440 * 59 / 100, fps),
|
||
(1920, 1080)
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn select_resolution_1080p_to_720p_at_very_low_bwe() {
|
||
let fps = 30;
|
||
let bitrate_1080 = resolution_bitrate_bps(1920, 1080, fps);
|
||
assert_eq!(
|
||
select_resolution(1920, 1080, bitrate_1080 / 10, fps),
|
||
(1280, 720)
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn next_upscale_tier_from_720p_to_1080p() {
|
||
assert_eq!(
|
||
next_upscale_tier((1280, 720), (2560, 1440)),
|
||
Some((1920, 1080))
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn next_upscale_tier_returns_none_at_highest() {
|
||
assert_eq!(next_upscale_tier((2560, 1440), (2560, 1440)), None);
|
||
}
|
||
}
|