diff --git a/src/avhw.rs b/src/avhw.rs index 8e3d888..1c5b3d8 100644 --- a/src/avhw.rs +++ b/src/avhw.rs @@ -711,9 +711,25 @@ impl EncState { // SwEncState - VAAPI GPU downscale + software H.264 encode // --------------------------------------------------------------------------- +/// Encoded H.264 frame with timing metadata for WebRTC output. +/// +/// MP4 file output (FrameOutput::Muxer) does NOT use this - it writes via +/// avformat which preserves PTS internally. WebRTC output (FrameOutput::Channel) +/// requires explicit PTS propagation so RTP timestamps reflect real capture time. +/// Without this, WebRTC clients' jitter buffers grow to seconds under +/// damage-driven variable frame rate. See issue #24. +#[derive(Debug)] +pub struct EncodedH264Frame { + /// H.264 NAL byte stream (Annex B or AVCC depending on encoder configuration) + pub data: Vec, + /// PTS in encoder time_base units (1/fps seconds), normalized so first frame = 0. + /// Derived from real capture time, NOT frame counter. + pub pts_ticks: i64, +} + pub enum FrameOutput { Muxer(ff::format::context::Output), - Channel(crossbeam_channel::Sender>), + Channel(crossbeam_channel::Sender), } /// Owned CPU NV12 frame data for cross-thread transfer. @@ -1078,7 +1094,7 @@ impl SwEncEncode { fps: u32, bitrate: u64, gop_size: u32, - tx: crossbeam_channel::Sender>, + tx: crossbeam_channel::Sender, webrtc_paused: Arc, bitrate_rx: crossbeam_channel::Receiver, resolution_rx: crossbeam_channel::Receiver, @@ -1377,18 +1393,36 @@ impl SwEncEncode { // slice is copied into a Vec before the packet is unreffed. let data: &[u8] = unsafe { std::slice::from_raw_parts(raw.data, raw.size as usize) }; - match tx.try_send(data.to_vec()) { + // Normalize PTS: subtract starting_timestamp so first frame = 0. + // Mirrors the Muxer branch normalization above. `start_ts` is + // self.starting_timestamp.unwrap_or(0) (passed by caller), so + // when no origin is recorded yet the subtraction is a no-op. + let pts_ticks = match pkt.pts() { + Some(p) => p - start_ts, + None => { + // libx264 should always set PTS; emitting RTP ts=0 + // here would recreate issue #24. Drop the packet. + tracing::warn!( + "encoder produced packet without PTS, dropping" + ); + continue; + } + }; + match tx.try_send(EncodedH264Frame { + data: data.to_vec(), + pts_ticks, + }) { Ok(()) => {} Err(crossbeam_channel::TrySendError::Full(frame)) => { tracing::warn!( "WebRTC channel full, dropping frame: {} bytes lost", - frame.len() + frame.data.len() ); } Err(crossbeam_channel::TrySendError::Disconnected(frame)) => { tracing::warn!( "WebRTC channel disconnected: {} bytes lost", - frame.len() + frame.data.len() ); self.webrtc_disconnected = true; break; @@ -1458,7 +1492,7 @@ impl SwEncState { fps: u32, bitrate: u64, gop_size: u32, - tx: crossbeam_channel::Sender>, + tx: crossbeam_channel::Sender, webrtc_paused: Arc, ) -> Result { tracing::info!( diff --git a/src/state.rs b/src/state.rs index 4513c71..352f4f7 100644 --- a/src/state.rs +++ b/src/state.rs @@ -43,7 +43,7 @@ use ffmpeg_next as ff; use ffmpeg_next::ffi; use crate::args::Args; -use crate::avhw::{AvHwDevCtx, EncState, SwEncState}; +use crate::avhw::{AvHwDevCtx, EncState, EncodedH264Frame, SwEncState}; use crate::cap_wlr_screencopy::CapWlrScreencopy; use crate::fps_limit::FpsLimit; use crate::stats::{FrameTimings, PipelineStats}; @@ -226,8 +226,8 @@ pub struct State { pub drm_device: Option, pub drm_device_from_compositor: Option, pub webrtc: Option, - pub webrtc_tx: Option>>, - webrtc_rx: Option>>, + pub webrtc_tx: Option>, + webrtc_rx: Option>, webrtc_frames_sent: u64, webrtc_paused: Option>, stats: PipelineStats, @@ -701,12 +701,13 @@ impl State { if let Some(ref rx) = self.webrtc_rx { let mut count = 0u32; - while let Ok(data) = rx.try_recv() { + while let Ok(enc_frame) = rx.try_recv() { if !connected { continue; } count += 1; - if let Err(e) = wrtc.write_h264_frame(&data, self.webrtc_frames_sent, self.args.fps) + if let Err(e) = wrtc + .write_h264_frame(&enc_frame.data, enc_frame.pts_ticks, self.args.fps) { tracing::debug!("WebRTC write frame error: {e}"); } diff --git a/src/state_portal.rs b/src/state_portal.rs index c2f5e3d..47f80c8 100644 --- a/src/state_portal.rs +++ b/src/state_portal.rs @@ -9,8 +9,8 @@ use anyhow::{bail, Result}; // 错误处理工具 use crate::args::Args; // 命令行参数 use crate::avhw::{ - self, BitrateCommand, CpuNv12Frame, EncodeOutcome, ResolutionChange, SwEncEncode, SwEncImport, - SwEncState, + 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}; // 管道统计（帧计时、每秒快照） @@ -64,6 +64,10 @@ pub struct StatePortal { last_capture_arrival: Option, // timestamp of last real frame arrival idle_log_start: Option, // 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, + capture_start: Option, + last_pts_emitted: Option, } impl StatePortal { @@ -107,6 +111,9 @@ impl StatePortal { last_capture_arrival: None, idle_log_start: None, shutdown_started: false, + first_pts_ns: None, + capture_start: None, + last_pts_emitted: None, }) } @@ -452,7 +459,16 @@ impl StatePortal { } } let t_import_start = Instant::now(); - let pts = self.frames_encoded as i64; + // 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.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 硬件帧池 @@ -537,6 +553,43 @@ impl StatePortal { Ok(()) } + /// Compute PTS in encoder time_base units (1/fps) 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(i128::from(self.args.fps))) / NS_PER_SEC; + let mut pts = i64::try_from(ticks_i128).unwrap_or(i64::MAX); + + 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。 @@ -633,7 +686,7 @@ fn encode_thread_loop( fn webrtc_thread_loop( mut wrtc: WebRtcState, - webrtc_rx: crossbeam_channel::Receiver>, + webrtc_rx: crossbeam_channel::Receiver, fps: u32, enc_width: u32, enc_height: u32, @@ -736,8 +789,8 @@ fn webrtc_thread_loop( } if connected { - while let Ok(data) = webrtc_rx.try_recv() { - if let Err(e) = wrtc.write_h264_frame(&data, frames_sent, fps) { + while let Ok(enc_frame) = webrtc_rx.try_recv() { + if let Err(e) = wrtc.write_h264_frame(&enc_frame.data, enc_frame.pts_ticks, fps) { tracing::debug!("WebRTC write frame error: {e}"); } frames_sent = frames_sent.saturating_add(1); @@ -752,9 +805,10 @@ fn webrtc_thread_loop( } match webrtc_rx.recv_timeout(timeout) { - Ok(data) => { + Ok(enc_frame) => { if wrtc.is_connected() { - if let Err(e) = wrtc.write_h264_frame(&data, frames_sent, fps) { + if let Err(e) = wrtc.write_h264_frame(&enc_frame.data, enc_frame.pts_ticks, fps) + { tracing::debug!("WebRTC write frame error: {e}"); } frames_sent = frames_sent.saturating_add(1); diff --git a/src/webrtc.rs b/src/webrtc.rs index 3aff975..36ee2e9 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -334,9 +334,9 @@ impl WebRtcState { self.poll_rtc() } - pub fn write_h264_frame(&mut self, data: &[u8], frame_number: u64, fps: u32) -> Result<()> { + pub fn write_h264_frame(&mut self, data: &[u8], pts_ticks: i64, fps: u32) -> Result<()> { let should_destroy = if let Some(inner) = self.inner.as_mut() { - inner.write_h264_frame(data, frame_number, fps)? + inner.write_h264_frame(data, pts_ticks, fps)? } else { false }; @@ -655,7 +655,7 @@ impl WebRtcInner { Ok(()) } - fn write_h264_frame(&mut self, data: &[u8], frame_number: u64, fps: u32) -> Result { + fn write_h264_frame(&mut self, data: &[u8], pts_ticks: i64, fps: u32) -> Result { if !self.connected { return Ok(false); } @@ -690,11 +690,9 @@ impl WebRtcInner { self.need_keyframe = false; } - let ticks_per_second = 90_000u64; - let fps = fps.max(1) as u64; - let rtp_timestamp = frame_number.saturating_mul(ticks_per_second) / fps; - self.rtp_clock = rtp_timestamp as u32; - let rtp_time = MediaTime::new(rtp_timestamp, Frequency::NINETY_KHZ); + let rtp_timestamp = rtp_timestamp_from_pts_ticks(pts_ticks, fps); + self.rtp_clock = rtp_timestamp; + let rtp_time = MediaTime::new(rtp_timestamp as u64, Frequency::NINETY_KHZ); let writer = match self.rtc.writer(mid) { Some(w) => w, @@ -724,6 +722,19 @@ impl WebRtcInner { } } +/// Convert PTS in encoder time_base units (1/fps) to RTP timestamp (90kHz clock). +/// +/// Extracted as a pure function for unit testing. Clamps negative pts_ticks to 0 +/// (encoder should never emit negative PTS, but defensive). Saturating multiply +/// to avoid overflow on long sessions. +pub fn rtp_timestamp_from_pts_ticks(pts_ticks: i64, fps: u32) -> u32 { + const TICKS_PER_SECOND: u64 = 90_000; + let fps_safe = (fps.max(1) as u64).max(1); + let pts_u64 = (pts_ticks.max(0) as u64).min(u64::MAX / TICKS_PER_SECOND); + let rtp_ts = pts_u64.saturating_mul(TICKS_PER_SECOND) / fps_safe; + rtp_ts as u32 +} + // ── 工具函数 ────────────────────────────────────────────────────────────── /// 从 HTTP 请求中提取 body（在 \r\n\r\n 之后） @@ -866,4 +877,34 @@ mod tests { let bps = default.as_u64(); assert_eq!(bps, 5_000_000); } + + // ── RTP timestamp conversion (issue #24) ── + + #[test] + fn rtp_timestamp_zero_pts() { + assert_eq!(rtp_timestamp_from_pts_ticks(0, 30), 0); + } + + #[test] + fn rtp_timestamp_one_frame() { + // 1 frame at 30fps = 33ms = 3000 RTP ticks (90kHz / 30) + assert_eq!(rtp_timestamp_from_pts_ticks(1, 30), 3000); + } + + #[test] + fn rtp_timestamp_one_second() { + // 30 frames at 30fps = 1 second = 90000 RTP ticks + assert_eq!(rtp_timestamp_from_pts_ticks(30, 30), 90000); + } + + #[test] + fn rtp_timestamp_negative_clamps_to_zero() { + assert_eq!(rtp_timestamp_from_pts_ticks(-5, 30), 0); + } + + #[test] + fn rtp_timestamp_zero_fps_does_not_panic() { + // fps=0 should clamp to 1 internally, not divide by zero + let _ = rtp_timestamp_from_pts_ticks(100, 0); + } }