dailz/wl-webrtc
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(portal): BWE-driven resolution adaptation + duplicate frame skipping

Browse Source 
WebRTC client bandwidth estimate now drives both encoder bitrate and
resolution tier selection, replacing the previous static-target encoder.

- webrtc.rs: enable str0m BWE (seeded at 5 Mbps), surface
  EgressBitrateEstimate + KeyframeRequest events, expose
  get_bwe_estimate() / set_need_keyframe()
- state_portal.rs: wire bitrate/resolution channels between the WebRTC
  thread and the encode thread; tier ladder [1440p, 1080p, 720p] with
  downscale at 60% budget and upscale hysteresis (120% sustained 10s)
- avhw.rs: SwEncImport::poll_resolution_commands() rebuilds the import
  filter graph on UpdateResolution; SwEncEncode::recreate_encoder()
  rebuilds sws/enc_video/yuv_frame atomically; hash_sampled_y_plane()
  skips duplicate frames; VBV x264opts cap IDR bursts; H.264 level 4.0
  (muxer) / 4.2 (WebRTC)
- state.rs: sync wlr-screencopy GOP to fps*2 max 20 for parity
- fix: drain bitrate_rx + resolution_rx BEFORE the stride check in
  encode_cpu_frame() so the new (smaller-stride) frame produced after
  a resolution change does not hit the stale (larger) enc_width and
  crash the encode thread
- WebRTC GOP widened to fps*2 max 20 (was fps/2 max 10)
This commit is contained in:
dailz
2026-06-13 22:46:33 +08:00
parent 503e4dbc22
commit 3e60258627
9 changed files with 905 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/state_portal.rs
+274 -60
View File
@@ -5,10 +5,12 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{bail, Result}; // 错误处理工具
use anyhow::{bail, Result}; // 错误处理工具
use crate::args::Args; // 命令行参数
use crate::avhw::{self, CpuNv12Frame, SwEncEncode, SwEncImport, SwEncState}; // 软件编码器状态（VAAPI 导入 + H.264 编码）
use crate::args::Args; // 命令行参数
use crate::avhw::{
self, BitrateCommand, CpuNv12Frame, 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 信令与媒体传输
@@ -43,26 +45,26 @@ struct WebrtcThread {
/// 负责管理从 PipeWire 采集屏幕帧、通过 VAAPI 硬件编码的完整生命周期。
/// 工作流程：等待第一帧 → 创建编码器 → 持续编码帧数据。
pub struct StatePortal {
stage: PortalStage, // 当前采集阶段（等待首帧 / 流式编码中）
enc: Option<SwEncState>, // 软件编码器，首帧到达后初始化
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 丢弃帧数（用于增量计算）
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
stall_start: Option<Instant>, // when current stall began
last_stall_log: Option<Instant>, // rate-limiting for stall warnings
last_fillable_frame: Option<CpuNv12Frame>, // cached last frame for filler duplication
next_filler_at: Option<Instant>, // when to send next filler frame
last_capture_arrival: Option<Instant>, // timestamp of last real frame arrival
stall_start: Option<Instant>, // when current stall began
last_stall_log: Option<Instant>, // rate-limiting for stall warnings
last_fillable_frame: Option<CpuNv12Frame>, // cached last frame for filler duplication
next_filler_at: Option<Instant>, // when to send next filler frame
filler_frames_sent: u64,
}
@@ -147,7 +149,11 @@ impl StatePortal {
// 根据阻塞模式选择不同的帧接收策略
let frame = if block {
// 阻塞模式：最多等待 2ms 接收帧
match self.cap.frame_receiver().recv_timeout(std::time::Duration::from_millis(2)) {
match self
.cap
.frame_receiver()
.recv_timeout(std::time::Duration::from_millis(2))
{
Ok(frame) => frame,
Err(_) => {
self.record_capture_timeout();
@@ -194,10 +200,10 @@ impl StatePortal {
let actual_bitrate = self.args.bitrate.unwrap_or_else(|| {
5 * (enc_width as u64) * (enc_height as u64) * (self.args.fps as u64) / 100
});
// GOP 大小：WebRTC 模式使用更小的 GOP（fps/2，最低10），MP4 模式使用 fps
// 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(10)
(self.args.fps * 2).max(20)
} else {
self.args.fps
}
@@ -207,17 +213,25 @@ impl StatePortal {
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"))?;
let import = SwEncImport::new(
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 (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,
@@ -226,24 +240,48 @@ impl StatePortal {
actual_gop_size,
webrtc_tx,
paused.clone(),
bitrate_rx,
encoder_resolution_rx,
)?;
let handle = std::thread::Builder::new()
.name("wl-webrtc-encode".into())
.spawn(move || encode_thread_loop(encode, input_rx, timing_tx))?;
self.enc_import = Some(import);
self.enc_thread = Some(EncodeThread { handle: Some(handle), input_tx, timing_rx });
self.enc_thread = Some(EncodeThread {
handle: Some(handle),
input_tx,
timing_rx,
});
let wrtc = self.webrtc.take()
.ok_or_else(|| anyhow::anyhow!("internal: WebRtcState missing during init"))?;
let paused = self.webrtc_paused.as_ref()
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 (sent_gap_tx, sent_gap_rx) = crossbeam_channel::bounded(64);
let webrtc_handle = std::thread::Builder::new()
.name("wl-webrtc-webrtc".into())
.spawn(move || webrtc_thread_loop(wrtc, webrtc_rx, fps, paused, sent_gap_tx))?;
self.webrtc_thread = Some(WebrtcThread { handle: Some(webrtc_handle), sent_gap_rx });
.spawn(move || {
webrtc_thread_loop(
wrtc,
webrtc_rx,
fps,
enc_width,
enc_height,
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()
@@ -263,7 +301,9 @@ impl StatePortal {
};
self.stage = PortalStage::Streaming; // 切换到流式编码阶段
self.start_time = Some(Instant::now());
tracing::info!("First frame processed, encoder initialized, transitioning to Streaming");
tracing::info!(
"First frame processed, encoder initialized, transitioning to Streaming"
);
drop(frame); // 首帧仅用于初始化，不参与编码
}
PortalStage::Streaming => {
@@ -295,7 +335,10 @@ impl StatePortal {
}
let snap = self.stats.snapshot_and_reset();
if self.filler_frames_sent > 0 {
tracing::info!("stats: {snap} filler_frames_sent={}", self.filler_frames_sent);
tracing::info!(
"stats: {snap} filler_frames_sent={}",
self.filler_frames_sent
);
} else {
tracing::info!("stats: {snap}");
}
@@ -321,9 +364,9 @@ impl StatePortal {
self.last_stall_log = Some(now);
tracing::warn!("compositor frame delivery stalled");
} else {
let should_log = self
.last_stall_log
.map_or(true, |last_log| now.duration_since(last_log) >= Duration::from_secs(1));
let should_log = self.last_stall_log.map_or(true, |last_log| {
now.duration_since(last_log) >= Duration::from_secs(1)
});
if should_log {
self.last_stall_log = Some(now);
tracing::warn!("compositor frame delivery stalled");
@@ -437,13 +480,11 @@ impl StatePortal {
}
// 所有候选设备均失败，返回详细错误信息
bail!(
failures
.into_iter()
.map(|(p, e)| format!("{} ({e})", p.display()))
.collect::<Vec<_>>()
.join(", ")
);
bail!(failures
.into_iter()
.map(|(p, e)| format!("{} ({e})", p.display()))
.collect::<Vec<_>>()
.join(", "));
}
/// 处理单帧 DMA-BUF 数据
@@ -513,8 +554,11 @@ impl StatePortal {
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"))?;
let enc_thread = self.enc_thread.as_ref().ok_or_else(|| {
anyhow::anyhow!(
"internal invariant broken: encode thread missing while async import is active"
)
})?;
let fillable_frame = CpuNv12Frame {
y_data: cpu_nv12.y_data.clone(),
uv_data: cpu_nv12.uv_data.clone(),
@@ -632,11 +676,20 @@ fn webrtc_thread_loop(
mut wrtc: WebRtcState,
webrtc_rx: crossbeam_channel::Receiver<Vec<u8>>,
fps: u32,
enc_width: u32,
enc_height: u32,
paused: Arc<AtomicBool>,
sent_gap_tx: crossbeam_channel::Sender<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();
let timeout = Duration::from_millis(1);
loop {
@@ -659,6 +712,52 @@ fn webrtc_thread_loop(
}
paused.store(now_paused, Ordering::Relaxed);
if let Some(bwe) = wrtc.get_bwe_estimate() {
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 {
while let Ok(data) = webrtc_rx.try_recv() {
if let Err(e) = wrtc.write_h264_frame(&data, frames_sent, fps) {
@@ -700,6 +799,43 @@ fn webrtc_thread_loop(
tracing::info!("WebRTC thread exiting");
}
const RESOLUTION_TIERS: &[(u32, u32)] = &[(2560, 1440), (1920, 1080), (1280, 720)];
fn resolution_bitrate_bps(width: u32, height: u32, fps: u32) -> u64 {
5 * u64::from(width) * u64::from(height) * u64::from(fps) / 100
}
/// 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;
}
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]
}
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];
(next.0 <= ceiling.0 && next.1 <= ceiling.1).then_some(next)
}
impl Drop for StatePortal {
// 析构时自动调用 shutdown，确保编码器被刷新、资源被释放
fn drop(&mut self) {
@@ -712,8 +848,8 @@ impl Drop for StatePortal {
/// 将原始分辨率等比缩放至不超过 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; // 目标最大高度
const TARGET_W: u32 = 2560; // 目标最大宽度
const TARGET_H: u32 = 1440; // 目标最大高度
// 原始分辨率已在 2K 以内，直接对齐偶数
if width <= TARGET_W && height <= TARGET_H {
@@ -748,16 +884,16 @@ fn resolve_drm_device(args: &Args) -> Result<Option<PathBuf>> {
#[cfg(test)]
fn build_drm_descriptor(frame: &PwDmaBufFrame) -> ffmpeg_next::ffi::AVDRMFrameDescriptor {
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.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.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
}
@@ -774,11 +910,11 @@ mod tests {
PwDmaBufFrame {
fd,
offset: 0,
stride: 1920 * 4, // 每行 1920 像素 × 4 字节（XRGB）
modifier: 0, // DRM_FORMAT_MOD_LINEAR（线性布局）
stride: 1920 * 4, // 每行 1920 像素 × 4 字节（XRGB）
modifier: 0, // DRM_FORMAT_MOD_LINEAR（线性布局）
width: 1920,
height: 1080,
format: 0x34325258, // XR24 little-endian（XRGB8888）
format: 0x34325258, // XR24 little-endian（XRGB8888）
pts: 12345,
}
}
@@ -845,14 +981,48 @@ mod tests {
assert_eq!(result, None);
}
#[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 {
fd: unsafe { OwnedFd::from_raw_fd(libc::dup(2)) },
offset: 4096, // 4KB 对齐偏移
stride: 3840 * 4, // 4K 宽度 × 4 字节
modifier: 0x0100000000000001, // AMD modifiers
offset: 4096, // 4KB 对齐偏移
stride: 3840 * 4, // 4K 宽度 × 4 字节
modifier: 0x0100000000000001, // AMD modifiers
width: 3840,
height: 2160,
format: 0x34325258,
@@ -908,4 +1078,48 @@ mod tests {
));
}
// ── 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);
}
}