fix(webrtc): propagate real capture PTS through WebRTC channel (closes #24)
Root cause (3-stage bug found via Oracle round 1+2 review):
Browser WebRTC clients accumulated 2-3 seconds jitter buffer under
damage-driven variable frame rate (KWin Portal/PipeWire). User moved
mouse, saw action 2-3 seconds later on client.
Three coordinated bugs formed a chain that defeated any single-point fix:
1. state_portal.rs:455 used sequential frame counter as PTS instead of
real capture time. (Portal path only — wlr-screencopy already correct.)
2. avhw.rs Channel output sent only Vec<u8>, DISCARDING AVPacket PTS.
Even with correct encoder PTS, timing metadata was thrown away.
3. webrtc.rs:695 computed RTP timestamp as 'frame_number * 90000 / fps'
from a counter, ignoring any real PTS. Browser saw uniform 33ms RTP
spacing regardless of actual 1.6-57fps variable delivery, growing
jitter buffer to compensate for perceived 'network jitter'.
Fix (7-step ordered implementation per Oracle round 2):
1. EncodedH264Frame struct in avhw.rs carries data + pts_ticks
2. FrameOutput::Channel type changed from Sender<Vec<u8>> to
Sender<EncodedH264Frame>; both new_webrtc signatures updated
3. Channel drain in avhw.rs extracts pkt.pts(), normalizes via
'p - start_ts' (mirrors existing Muxer branch logic). Drops
packets with missing PTS instead of silently emitting zero.
4. write_h264_frame signature: frame_number:u64 -> pts_ticks:i64
(both WebRtcState and WebRtcInner layers). Extracted pure function
rtp_timestamp_from_pts_ticks(pts_ticks, fps) with 5 unit tests
covering zero, one-frame, one-second, negative clamp, fps=0.
5. state_portal.rs receiver loop consumes EncodedH264Frame, passes
.data and .pts_ticks to write_h264_frame.
6. state.rs (wlr-screencopy) receiver loop updated for compile
compatibility — its existing real-time PTS computation at state.rs:606
was already correct, now properly propagates through new channel type.
7. state_portal.rs:467 PTS computation GATED on output mode:
- WebRTC branch: compute_capture_pts() uses PipeWire's ns timestamp
(or Instant fallback), normalizes to first-frame-origin, converts
to encoder time_base units with i128 intermediate math, enforces
monotonicity via safe checked_add pattern.
- MP4 branch: KEEPS self.frames_encoded as i64 (sequential counter).
File output does not need real-time PTS; changing it would alter
file playback speed during static periods.
Oracle round 2 critical revisions incorporated:
- Single-point PTS normalization (only in avhw Channel drain), NOT at
source. Avoids double-subtraction with existing Muxer logic.
- MP4 path explicitly preserved — real PTS only applied to WebRTC branch.
- Safe Rust monotonicity guard (no unsafe pointer tricks).
- i64::try_from(ticks_i128) instead of broken i128::try_from(...).unwrap_or(i64::MAX).
- pkt.pts() missing -> log + drop, not silent unwrap_or(0).
- Updates span 4 files (avhw.rs, webrtc.rs, state_portal.rs, state.rs)
because channel type change ripples through both Portal and wlr paths.
Verification expectations:
- Browser jitter buffer should stabilize at 100-500ms (typical) instead
of growing to 2-3 seconds under damage-driven delivery.
- chrome://webrtc-internals: jitterBufferDelay / jitterBufferEmittedCount
ratio should drop significantly.
- Server-side metrics (output_bps, frame rate, IDR size) unchanged.
- MP4 file output (--output mode) behavior unchanged.
Out of scope (deferred):
- frame_age metric fix (Oracle: separate commit to isolate behavioral
change from observability change)
- VFR encoder redesign (1/fps time_base sufficient for this fix)
- MP4 VFR recording (semantic change, separate decision)
- Existing client jitter buffers may not auto-shrink; reconnect may be
required for users with already-accumulated latency
Tests:
- cargo build --release: clean, 0 new warnings (19 pre-existing)
- cargo test: 96 lib + 96 bin + 3 integration, 0 failed
- 5 new RTP unit tests covering edge cases
- SAFETY comments preserved verbatim
- 4 files changed, +157/-27 lines
This commit is contained in:
dailz
+40
-6
@@ -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<u8>,
|
||||
/// 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<Vec<u8>>),
|
||||
Channel(crossbeam_channel::Sender<EncodedH264Frame>),
|
||||
}
|
||||
|
||||
/// 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<Vec<u8>>,
|
||||
tx: crossbeam_channel::Sender<EncodedH264Frame>,
|
||||
webrtc_paused: Arc<AtomicBool>,
|
||||
bitrate_rx: crossbeam_channel::Receiver<BitrateCommand>,
|
||||
resolution_rx: crossbeam_channel::Receiver<ResolutionChange>,
|
||||
@@ -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<Vec<u8>>,
|
||||
tx: crossbeam_channel::Sender<EncodedH264Frame>,
|
||||
webrtc_paused: Arc<AtomicBool>,
|
||||
) -> Result<Self> {
|
||||
tracing::info!(
|
||||
|
||||
+6
-5
@@ -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<S: CaptureSource> {
|
||||
pub drm_device: Option<PathBuf>,
|
||||
pub drm_device_from_compositor: Option<PathBuf>,
|
||||
pub webrtc: Option<WebRtcState>,
|
||||
pub webrtc_tx: Option<crossbeam_channel::Sender<Vec<u8>>>,
|
||||
webrtc_rx: Option<crossbeam_channel::Receiver<Vec<u8>>>,
|
||||
pub webrtc_tx: Option<crossbeam_channel::Sender<EncodedH264Frame>>,
|
||||
webrtc_rx: Option<crossbeam_channel::Receiver<EncodedH264Frame>>,
|
||||
webrtc_frames_sent: u64,
|
||||
webrtc_paused: Option<Arc<AtomicBool>>,
|
||||
stats: PipelineStats,
|
||||
@@ -701,12 +701,13 @@ impl<S: CaptureSource> State<S> {
|
||||
|
||||
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}");
|
||||
}
|
||||
|
||||
+62
-8
@@ -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<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 {
|
||||
@@ -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<Vec<u8>>,
|
||||
webrtc_rx: crossbeam_channel::Receiver<EncodedH264Frame>,
|
||||
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);
|
||||
|
||||
+49
-8
@@ -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<bool> {
|
||||
fn write_h264_frame(&mut self, data: &[u8], pts_ticks: i64, fps: u32) -> Result<bool> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user