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

refactor: decompose oversized modules into directory form (avhw + state + cap_portal + state_portal + webrtc + bench bins) #26

Merged
dailz merged 16 commits from refactor/split-avhw into master 2026-07-14 11:44:56 +08:00
Conversation 0 Commits 16 Files Changed 53
+441 -414
3 changed files with 441 additions and 414 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit bcfbd93f5a - Show all commits
src/state_portal.rs
+10 -414
View File
@@ -13,13 +13,21 @@ use anyhow::{bail, Result}; // 错误处理工具
use crate::args::Args; // 命令行参数 use crate::args::Args; // 命令行参数
use crate::avhw::{ use crate::avhw::{
self, BitrateCommand, CpuNv12Frame, EncodeOutcome, EncodedH264Frame, ResolutionChange, self, BitrateCommand, CpuNv12Frame, ResolutionChange, SwEncEncode, SwEncImport, SwEncState,
SwEncEncode, SwEncImport, SwEncState,
}; // 软件编码器状态（VAAPI 导入 + H.264 编码） }; // 软件编码器状态（VAAPI 导入 + H.264 编码）
use crate::cap_portal::{CapPortal, PwCtrlEvent, PwDmaBufFrame}; // PipeWire 屏幕采集端点 use crate::cap_portal::{CapPortal, PwCtrlEvent, PwDmaBufFrame}; // PipeWire 屏幕采集端点
use crate::stats::{FrameTimings, PipelineStats}; // 管道统计（帧计时、每秒快照） use crate::stats::{FrameTimings, PipelineStats}; // 管道统计（帧计时、每秒快照）
use crate::webrtc::WebRtcState; // WebRTC 信令与媒体传输 use crate::webrtc::WebRtcState; // WebRTC 信令与媒体传输
mod bitrate;
use bitrate::webrtc_startup_bitrate_bps;
mod threads;
use threads::{
encode_thread_loop, webrtc_thread_loop, EncodeThread, EncodeThreadTiming, WebRtcThreadChannels,
WebRtcThreadConfig, WebrtcThread,
};
/// 门户采集的阶段状态 /// 门户采集的阶段状态
/// - WaitingForFormat: 等待接收到第一帧 DMA-BUF 以确定视频格式参数 /// - WaitingForFormat: 等待接收到第一帧 DMA-BUF 以确定视频格式参数
/// - Streaming: 已完成初始化，正在持续编码流 /// - Streaming: 已完成初始化，正在持续编码流
@@ -28,44 +36,6 @@ enum PortalStage {
Streaming, Streaming,
} }
struct EncodeThreadTiming {
sws_us: u64,
encode_us: u64,
output_bytes: usize,
}
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>,
}
struct WebrtcThread {
handle: Option<std::thread::JoinHandle<()>>,
sent_gap_rx: crossbeam_channel::Receiver<(f64, Option<f64>)>,
}
/// Static configuration handed to the WebRTC sender thread. Immutable for the
/// thread's lifetime; a resolution tier change rebuilds the whole pipeline
/// (and spawns a new thread) rather than mutating this.
struct WebRtcThreadConfig {
fps: u32,
enc_width: u32,
enc_height: u32,
max_bitrate: u64,
}
/// Channel endpoints owned exclusively by the WebRTC sender thread after spawn.
/// The reverse endpoints stay with StatePortal (or the encode thread) for
/// inbound/outbound traffic.
struct WebRtcThreadChannels {
webrtc_rx: crossbeam_channel::Receiver<EncodedH264Frame>,
sent_gap_tx: crossbeam_channel::Sender<(f64, Option<f64>)>,
bitrate_tx: crossbeam_channel::Sender<BitrateCommand>,
resolution_tx: crossbeam_channel::Sender<BitrateCommand>,
}
/// 门户模式的主状态机 /// 门户模式的主状态机
/// ///
/// 负责管理从 PipeWire 采集屏幕帧、通过 VAAPI 硬件编码的完整生命周期。 /// 负责管理从 PipeWire 采集屏幕帧、通过 VAAPI 硬件编码的完整生命周期。
@@ -677,256 +647,6 @@ impl StatePortal {
} }
} }
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>,
) {
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");
}
fn webrtc_thread_loop(
mut wrtc: WebRtcState,
config: WebRtcThreadConfig,
channels: WebRtcThreadChannels,
paused: Arc<AtomicBool>,
) {
let WebRtcThreadConfig {
fps,
enc_width,
enc_height,
max_bitrate,
} = config;
let WebRtcThreadChannels {
webrtc_rx,
sent_gap_tx,
bitrate_tx,
resolution_tx,
} = channels;
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 {
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();
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 = bwe.abs_diff(last);
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(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 {
while webrtc_rx.try_recv().is_ok() {}
}
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");
}
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
}
/// 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
}
}
/// 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 { impl Drop for StatePortal {
// 析构时自动调用 shutdown，确保编码器被刷新、资源被释放 // 析构时自动调用 shutdown，确保编码器被刷新、资源被释放
fn drop(&mut self) { fn drop(&mut self) {
@@ -1082,48 +802,6 @@ mod tests {
assert_eq!(result, None); 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 描述符 /// 测试：使用自定义偏移量和 stride 构建 DRM 描述符
#[test] #[test]
fn build_drm_descriptor_custom_offset_and_stride() { fn build_drm_descriptor_custom_offset_and_stride() {
@@ -1148,86 +826,4 @@ mod tests {
} }
// ── issue #8 regression ── // ── 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);
}
} }
src/state_portal/bitrate.rs
+144
View File
@@ -0,0 +1,144 @@
pub(super) const RESOLUTION_TIERS: &[(u32, u32)] = &[(2560, 1440), (1920, 1080), (1280, 720)];
pub(super) fn resolution_bitrate_bps(width: u32, height: u32, fps: u32) -> u64 {
5 * u64::from(width) * u64::from(height) * u64::from(fps) / 100
}
/// 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.
pub(super) 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
}
}
/// Select resolution tier based on BWE estimate.
/// Returns (width, height) for the selected tier.
pub(super) 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]
}
pub(super) 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)
}
#[cfg(test)]
mod tests {
use super::*;
#[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);
}
#[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);
}
}
src/state_portal/threads.rs
+287
View File
@@ -0,0 +1,287 @@
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use crate::avhw::{BitrateCommand, CpuNv12Frame, EncodeOutcome, EncodedH264Frame, SwEncEncode};
use crate::webrtc::WebRtcState;
use super::bitrate::{next_upscale_tier, resolution_bitrate_bps, select_resolution};
pub(super) struct EncodeThreadTiming {
pub(super) sws_us: u64,
pub(super) encode_us: u64,
pub(super) output_bytes: usize,
}
pub(super) struct EncodeThread {
pub(super) handle: Option<std::thread::JoinHandle<()>>,
pub(super) input_tx: crossbeam_channel::Sender<CpuNv12Frame>,
pub(super) timing_rx: crossbeam_channel::Receiver<EncodeThreadTiming>,
pub(super) duplicate_count: std::sync::Arc<std::sync::atomic::AtomicU64>,
}
pub(super) struct WebrtcThread {
pub(super) handle: Option<std::thread::JoinHandle<()>>,
pub(super) sent_gap_rx: crossbeam_channel::Receiver<(f64, Option<f64>)>,
}
/// Static configuration handed to the WebRTC sender thread. Immutable for the
/// thread's lifetime; a resolution tier change rebuilds the whole pipeline
/// (and spawns a new thread) rather than mutating this.
pub(super) struct WebRtcThreadConfig {
pub(super) fps: u32,
pub(super) enc_width: u32,
pub(super) enc_height: u32,
pub(super) max_bitrate: u64,
}
/// Channel endpoints owned exclusively by the WebRTC sender thread after spawn.
/// The reverse endpoints stay with StatePortal (or the encode thread) for
/// inbound/outbound traffic.
pub(super) struct WebRtcThreadChannels {
pub(super) webrtc_rx: crossbeam_channel::Receiver<EncodedH264Frame>,
pub(super) sent_gap_tx: crossbeam_channel::Sender<(f64, Option<f64>)>,
pub(super) bitrate_tx: crossbeam_channel::Sender<BitrateCommand>,
pub(super) resolution_tx: crossbeam_channel::Sender<BitrateCommand>,
}
pub(super) 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>,
) {
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");
}
pub(super) fn webrtc_thread_loop(
mut wrtc: WebRtcState,
config: WebRtcThreadConfig,
channels: WebRtcThreadChannels,
paused: Arc<AtomicBool>,
) {
let WebRtcThreadConfig {
fps,
enc_width,
enc_height,
max_bitrate,
} = config;
let WebRtcThreadChannels {
webrtc_rx,
sent_gap_tx,
bitrate_tx,
resolution_tx,
} = channels;
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 {
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();
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 = bwe.abs_diff(last);
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(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 {
while webrtc_rx.try_recv().is_ok() {}
}
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");
}
#[cfg(test)]
mod tests {
#[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(_))
));
}
}