From bcfbd93f5ae51bfd5c7361db4cf5274b541b0198 Mon Sep 17 00:00:00 2001 From: dailz Date: Mon, 13 Jul 2026 16:35:44 +0800 Subject: [PATCH] refactor(state_portal): extract bitrate helpers + thread loops to submodules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 3: split state_portal.rs (1241 -> 829 LOC) into three modules. - src/state_portal.rs (829 LOC): keeps StatePortal struct + impl (with poll_and_encode / handle_pw_frame / shutdown / etc.) + Drop + PortalStage enum + DRM helpers + DRM tests. Per Oracle/Explore audit, all 21 StatePortal fields are private and poll_and_encode interleaves three channel reads with state-machine transitions; moving it would force pub(crate) on every field, so it stays in mod.rs. - src/state_portal/bitrate.rs (144 LOC): RESOLUTION_TIERS + 4 pure fns (resolution_bitrate_bps / webrtc_startup_bitrate_bps / select_resolution / next_upscale_tier) + 10 tests that exercise them. Pure fns with no StatePortal field access — the cleanest possible extract. - src/state_portal/threads.rs (287 LOC): the 5 thread-related types (EncodeThreadTiming / EncodeThread / WebrtcThread / WebRtcThreadConfig / WebRtcThreadChannels) + the two free fns encode_thread_loop / webrtc_thread_loop + the 3 channel-semantics regression tests (try_send_* / shutdown_rx_drop_*) that document crossbeam invariants the shutdown logic relies on. Struct fields widened to pub(super) so StatePortal in mod.rs can construct and join them. Test preservation: - state_portal test count: 17 (mod.rs=4 drm tests + bitrate.rs=10 + threads.rs=3 channel tests) — matches baseline. Verification (all green): - cargo build / cargo build --release - cargo test (79 lib + 3 integration = 82 pass, 1 ignored — unchanged) - cargo clippy --all-targets -- -D warnings - cargo fmt --check --- src/state_portal.rs | 424 +----------------------------------- src/state_portal/bitrate.rs | 144 ++++++++++++ src/state_portal/threads.rs | 287 ++++++++++++++++++++++++ 3 files changed, 441 insertions(+), 414 deletions(-) create mode 100644 src/state_portal/bitrate.rs create mode 100644 src/state_portal/threads.rs diff --git a/src/state_portal.rs b/src/state_portal.rs index 0200c55..e4ebf92 100644 --- a/src/state_portal.rs +++ b/src/state_portal.rs @@ -13,13 +13,21 @@ use anyhow::{bail, Result}; // 错误处理工具 use crate::args::Args; // 命令行参数 use crate::avhw::{ - self, BitrateCommand, CpuNv12Frame, EncodeOutcome, EncodedH264Frame, ResolutionChange, - SwEncEncode, SwEncImport, SwEncState, + 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 信令与媒体传输 +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 以确定视频格式参数 /// - Streaming: 已完成初始化，正在持续编码流 @@ -28,44 +36,6 @@ enum PortalStage { Streaming, } -struct EncodeThreadTiming { - sws_us: u64, - encode_us: u64, - output_bytes: usize, -} - -struct EncodeThread { - handle: Option>, - input_tx: crossbeam_channel::Sender, - timing_rx: crossbeam_channel::Receiver, - duplicate_count: std::sync::Arc, -} - -struct WebrtcThread { - handle: Option>, - sent_gap_rx: crossbeam_channel::Receiver<(f64, Option)>, -} - -/// 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, - sent_gap_tx: crossbeam_channel::Sender<(f64, Option)>, - bitrate_tx: crossbeam_channel::Sender, - resolution_tx: crossbeam_channel::Sender, -} - /// 门户模式的主状态机 /// /// 负责管理从 PipeWire 采集屏幕帧、通过 VAAPI 硬件编码的完整生命周期。 @@ -677,256 +647,6 @@ impl StatePortal { } } -fn encode_thread_loop( - mut encode: SwEncEncode, - input_rx: crossbeam_channel::Receiver, - timing_tx: crossbeam_channel::Sender, - duplicate_count: std::sync::Arc, -) { - 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, -) { - 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 = None; - let mut last_sent_bitrate: Option = 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 { // 析构时自动调用 shutdown，确保编码器被刷新、资源被释放 fn drop(&mut self) { @@ -1082,48 +802,6 @@ mod tests { 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() { @@ -1148,86 +826,4 @@ mod tests { } // ── issue #8 regression ── - - #[test] - fn try_send_full_channel_returns_full_not_block() { - let (tx, rx) = crossbeam_channel::bounded::>(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::>(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::>(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); - } } diff --git a/src/state_portal/bitrate.rs b/src/state_portal/bitrate.rs new file mode 100644 index 0000000..d7ec03e --- /dev/null +++ b/src/state_portal/bitrate.rs @@ -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); + } +} diff --git a/src/state_portal/threads.rs b/src/state_portal/threads.rs new file mode 100644 index 0000000..f0c6edf --- /dev/null +++ b/src/state_portal/threads.rs @@ -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>, + pub(super) input_tx: crossbeam_channel::Sender, + pub(super) timing_rx: crossbeam_channel::Receiver, + pub(super) duplicate_count: std::sync::Arc, +} + +pub(super) struct WebrtcThread { + pub(super) handle: Option>, + pub(super) sent_gap_rx: crossbeam_channel::Receiver<(f64, Option)>, +} + +/// 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, + pub(super) sent_gap_tx: crossbeam_channel::Sender<(f64, Option)>, + pub(super) bitrate_tx: crossbeam_channel::Sender, + pub(super) resolution_tx: crossbeam_channel::Sender, +} + +pub(super) fn encode_thread_loop( + mut encode: SwEncEncode, + input_rx: crossbeam_channel::Receiver, + timing_tx: crossbeam_channel::Sender, + duplicate_count: std::sync::Arc, +) { + 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, +) { + 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 = None; + let mut last_sent_bitrate: Option = 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::>(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::>(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::>(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(_)) + )); + } +}