From 3fd214732df2278afa167555c95227b903d9c732 Mon Sep 17 00:00:00 2001 From: Thomas Lekanger Date: Thu, 10 Sep 2026 15:17:13 +0200 Subject: [PATCH] fix(irl-source): delay late-arriving video into pacing instead of dropping or judder --- CLAUDE.md | 8 +- README.md | 1 + crates/irl-core/src/consts.rs | 20 ++ crates/irl-core/src/lib.rs | 2 + crates/irl-core/src/pacing.rs | 39 +++ crates/irl-core/src/stats.rs | 19 +- crates/irl-core/src/video_delay.rs | 382 ++++++++++++++++++++++ crates/irl-source/src/receiver/decode.rs | 1 + crates/irl-source/src/receiver/stream.rs | 3 +- crates/irl-source/src/shared.rs | 6 + crates/irl-source/src/source.rs | 1 + crates/irl-source/src/video/output.rs | 15 +- crates/irl-source/src/video/thread.rs | 182 +++++++++-- crates/irl-source/tests/network_sim.rs | 2 + crates/irl-source/tests/shell_stats.rs | 1 + crates/irl-source/tests/video_pipeline.rs | 304 ++++++++++++++--- docs/viewer-quality-plan.md | 5 + 17 files changed, 911 insertions(+), 80 deletions(-) create mode 100644 crates/irl-core/src/video_delay.rs diff --git a/CLAUDE.md b/CLAUDE.md index 3a86624..fdd2bd2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -129,6 +129,7 @@ Buffer regulation happens through playback speed only, asymmetric like IRLToolki | `timing.rs` | Output-clock arithmetic: next timestamp, lead, expected samples, soft compensation, prime threshold. | | `dsp.rs` | Fades, shaped concealment silence, last-sample memory. | | `video_time.rs` | Mapping video PTS through the audio playout offset, the fallback anchor and its clamps, the frame-interval EMA. | +| `video_delay.rs` | The standing video delay: sized from the worst arrival margin (due time minus packet arrival) so every frame is in hand a delivery lead early, raised at once before the play head is anchored and only for a shortfall that recurs across a window after it, capped, never lowered within a connection. | | `url_opts.rs` | The demuxer option table (probe sizes, SRT latency, RIST/UDP buffers, `tls_verify=0`), parsing of the user's FFmpeg Options, and `url_awaits_caller`, which decides whether the I/O stall deadline applies before a connection exists. | | `stats.rs` | `FIELDS`, `StatsSnapshot`, `proc_declaration()`. | | `config.rs` | `HwDecode`, `Watermarks::derive`. | @@ -170,6 +171,8 @@ Four threads. The C plugin enforced its lock contract by convention and a debug- Frames are handed to libobs a couple of canvas ticks *before* their due time. libobs is a scheduler too — `ready_async_frame` advances its play head by wall-clock deltas and takes the frame it has just passed — so a frame handed over exactly at its due time is not queued yet when its render tick runs and slips to the next one, which at 30fps on a 60fps canvas is visible judder. The frame keeps its due time as its timestamp, so the lead changes when libobs *receives* it, not when it is shown. The exception is the frame that re-anchors libobs's play head after a start or a clear: `get_closest_frame` displays that one on arrival whatever its timestamp and anchors from it, so a lead there would run the whole connection early. That frame goes at its due time, and the anchor only clears once a frame libobs actually received went out at its real due time. And while an audio stream is present, that frame's due time has to come from the *audio mapping*, not the video-only fallback: the two disagree by ~100 ms (the fallback schedules the first frame one Target Buffer out; the mapping puts it at the first audio chunk, a prime threshold plus a chunk after the first *kept* audio), which way depends on whether the audio warm-up or the first keyframe won, and libobs freezes whichever error the anchor frame carried into the connection's lip sync. So video holds until audio primes (the pump wakes the video thread the moment it publishes the mapping), drops anything the mapping lands more than a canvas tick in the past, and anchors on the first frame that is on time. A stream whose audio never primes is let through on the fallback after `VIDEO_ANCHOR_WAIT_MARGIN_MS` past the expected prime. +The lead only exists for a frame that is in hand a lead before it is due. How early a frame is in hand — its *arrival margin*, due time minus the OBS time its packet reached the video thread (`TimedPacket::received_ns`) — is the sender's to set: video that leaves the encoder later than the audio of the same instant has that much less margin, and once the skew exceeds what Target Buffer plus the audio output lead covers, every frame is late. A late frame goes out on arrival, unpaced, and libobs then drops one whenever two arrive inside a canvas tick, which is what a stream with trailing video looks like: low fps. `irl_core::video_delay` closes that gap with a standing delay on every video due time, sized from the worst margin seen so that frames are a full lead early again. Before the anchor a shortfall raises it on the spot (nothing has been shown; `VideoThread::settle_anchor_candidate` does this while it also drops the stale backlog). After the anchor a raise moves the picture, so it takes `VIDEO_DELAY_MIN_FRAMES` late frames spread across `VIDEO_DELAY_WINDOW_MS`, and the delay is capped at `VIDEO_DELAY_MAX_MS` and never lowered within a connection. It is a fixed, known lip-sync error in return for a smooth picture: `video_delay_ms` reports it, and the warning that sets it says how much more Target Buffer would remove it. A clear and a decoder handover reset it. The video-only fallback schedules its first frame at arrival, so a stream without audio always carries a delay of exactly one lead. + Video decode is on the video thread and not the receiver for two reasons, and the second is the load-bearing one. Decoding eagerly would mean holding the stream's whole latency as decoded frames — 8s of 4K60 is ~6GB — where the same 8s of packets is ~20MB. And the receiver spends a network stall blocked in `av_read_frame`, which is exactly when video must keep draining the buffer it already has, so the thread that decodes cannot be the thread that reads. Lock order, and the whole of it: **`audio_state` → `audio_buf` → `hot.watermarks`.** `video.q` is never held together with any of them. The audio pump takes `audio_state` exactly once per iteration and passes `&mut AudioState` down, so nothing below it can take it again — parking_lot mutexes are not recursive, and a nested acquire would hang the audio thread and then the video thread behind it. @@ -226,7 +229,7 @@ Releases are tag driven (`.github/workflows/release.yml`, see `RELEASING.md`). P The port is behaviour-identical except for these, which are intentional: 1. The dead `network_buffer_mb` setting is gone. Nothing read it; the transport buffer is `irl_core::consts::NETWORK_BUFFER_MB`. -2. The `video_decoder_flushes` stat is gone (it was always 0 after the video decoder stopped being flushed). 27 stat fields remain. +2. The `video_decoder_flushes` stat is gone (it was always 0 after the video decoder stopped being flushed). 27 stat fields remained; deviation 16 adds `video_delay_ms`, making 28. 3. `irl-stats.lua` finds the source by its plugin id instead of by display name, and takes source names as script properties. 4. The vestigial `hw_map_ok` flag is not ported. 5. `w32-pthreads.dll` is no longer shipped on Windows: Rust never calls `pthread_*`, so the librist shim hazard that `include/irl-threading.h` existed for is gone. The installer deletes a stale copy. @@ -239,7 +242,8 @@ The port is behaviour-identical except for these, which are intentional: 12. The I/O stall deadline is not armed while a listener URL waits to be called, and once connected it is measured from the last byte that arrived rather than from the start of the call (master `6d09dea`). `InterruptWatch` therefore tracks the `AVFormatContext` so the callback can read `pb->bytes_read`; `FormatContext` clears that pointer on a failed open and in `Drop`, which cannot wait for the watch's own `Drop` because the receiver holds the same `Arc` across connections. 13. Video is decoded on the video thread, just before each frame is due, and the receiver → video queue carries compressed packets instead of decoded frames. The C decoded eagerly on the receiver thread, which made the Target Buffer cost decoded-frame memory: 1 GiB of pacing budget is 5.7s of 1080p60 but only 1.4s of 4K60 and 0.7s of 4K60 10-bit, and past that frames were emitted early and dropped. Decoded memory is now bounded by `VIDEO_DECODE_LEAD_MS` regardless of the target. `PacingQueue` gained the matching soft/hard bound split: holding the decode lead is normal and must not emit early, while the byte and frame ceilings are memory limits that still do. The stats line reports `pktq=` instead of `pinned_peak=`, since no decoded frame pins a decoder surface any more. 14. PTS repair treats a gap of at most one time-base tick, in either direction, as the sender being on time rather than as a discontinuity. The C tested only `< 1 ms` and only forwards, which is a threshold a 90 kHz clock cannot express a 44.1 kHz frame against: 1024 samples is 2089.795 ticks and `duration` can carry only 2090, so the stream landed a tick *early* every few frames, fell into the leading-edge rule for backward jumps (which deliberately freezes the baseline), and the next frame was then interpolated onto the frozen baseline. The repaired timeline stayed one frame short from there on. Since the audio→video mapping is derived from those PTS, every 44.1 kHz stream — which is what phone encoders send — carried a standing ~23 ms lip-sync error, and its `norm=` counter climbed at the frame rate. 48 kHz divides 90 kHz exactly and was never affected. -15. Video does not anchor libobs's play head on a fallback-scheduled frame when an audio stream is present. The C handed over whichever frame came due first, fallback or mapped, and libobs — which anchors to that frame's *arrival* and never moves the anchor — then played the whole connection with the ~100 ms disagreement between the two schedules baked in, in whichever direction the warm-up/keyframe race went. In the port the race was almost always lost the same way (the video thread decodes the keyframe while the receiver is still working through the probe backlog, so the warm-up has drained and the fallback runs ~100 ms early), which surfaced as audio consistently lagging on phone encoders. Video now waits for the mapping, drops frames it lands in the past, and anchors on the first on-time frame (`VideoThread::awaiting_audio_mapping`, `drop_stale_before_anchor`). +15. Video does not anchor libobs's play head on a fallback-scheduled frame when an audio stream is present. The C handed over whichever frame came due first, fallback or mapped, and libobs — which anchors to that frame's *arrival* and never moves the anchor — then played the whole connection with the ~100 ms disagreement between the two schedules baked in, in whichever direction the warm-up/keyframe race went. In the port the race was almost always lost the same way (the video thread decodes the keyframe while the receiver is still working through the probe backlog, so the warm-up has drained and the fallback runs ~100 ms early), which surfaced as audio consistently lagging on phone encoders. Video now waits for the mapping, drops frames it lands in the past, and anchors on the first on-time frame (`VideoThread::awaiting_audio_mapping`, `settle_anchor_candidate`). Deviation 16 is what keeps this from dropping a stream whose frames are *all* in the past. +16. Video that arrives too late to be paced is delayed by a measured, standing amount instead of played unpaced. The C mapped each frame through the audio playout and handed over whatever was past due on arrival, so a sender whose video trailed its audio by more than Target Buffer covered — 120 ms by default, against phone hardware encoders that routinely run 100 to 300 ms behind their audio — played the whole connection with zero lead: every frame handed to libobs at or after its due time, and one dropped by `ready_async_frame` whenever two arrived inside a canvas tick. That is the "low fps" such a stream showed, and the delivery lead (master `964f74b`) could not help it, because a lead needs a frame in hand early. After deviation 15 the same stream was dropped forever instead, since no frame was ever on time. The port measures each frame's arrival margin and adds the shortfall to the schedule as `irl_core::video_delay` (see the threading section); `video_delay_ms` reports it and the warning that sets it tells the user how much Target Buffer restores lip sync. A stream without audio is affected too: the video-only fallback scheduled its first frame at arrival, with no lead for any frame that followed it on time, and now carries a delay of one lead. ## Contributing diff --git a/README.md b/README.md index 61b1e03..e6ee8f1 100644 --- a/README.md +++ b/README.md @@ -330,6 +330,7 @@ Stats are exposed through OBS's `proc_handler` API under the `get_stats` call, a | `video_corrupt_held` | int | HEVC frames held back instead of shown because they were predicted from a missing reference and would have rendered gray; the last good frame stays on screen until the next keyframe | | `video_lead_ms` | int | How far ahead of real time the last video frame was timestamped. Tracks the audio buffer; a value climbing well past Target Buffer and staying there means concealment has inflated the A/V mapping | | `video_lead_excess` | int | Frames whose lead exceeded what OBS's async queue can absorb. Harmless while the lead is steady; sustained growth is what makes OBS drop queued video | +| `video_delay_ms` | int | Standing delay added to the video schedule because video reached the plugin too late to be paced against the audio playout (the encoder sends video later than audio by more than Target Buffer covers). Lip sync is off by this much; raising Target Buffer by at least this takes it back to zero | | `stream_delay_ms` | int | End-to-end stream delay (SRT latency + decode + buffering) | | `low_latency_audio` | bool | Whether OBS async unbuffered low-latency mode is enabled | | `reconnect_count` | int | Number of reconnect attempts since the source was created | diff --git a/crates/irl-core/src/consts.rs b/crates/irl-core/src/consts.rs index 70e1712..146e574 100644 --- a/crates/irl-core/src/consts.rs +++ b/crates/irl-core/src/consts.rs @@ -307,6 +307,23 @@ pub const VIDEO_PACING_MAX_WAIT_MS: u64 = 50; /// this is the slack past that before a stream whose audio never arrives is /// let through on the fallback anyway. pub const VIDEO_ANCHOR_WAIT_MARGIN_MS: i64 = 1000; + +/// Ceiling on the standing video delay (`irl_core::video_delay`). +/// +/// The delay covers a sender whose video reaches the plugin later than the +/// audio of the same instant by more than Target Buffer absorbs. A whole +/// second of that is no longer a skew any encoder produces; it is a decoder +/// or a host that cannot keep up, and past this the frames stay late (shown +/// on arrival, as before the delay existed) rather than the picture drifting +/// ever further behind the sound. +pub const VIDEO_DELAY_MAX_MS: u64 = 1000; +/// After the play head is anchored, a raise of the video delay moves the +/// picture, so late frames must recur across this window before one is made. +pub const VIDEO_DELAY_WINDOW_MS: u64 = 1000; +/// Late frames within that window, spread over at least half of it, that count +/// as recurring. A single scheduling hiccup on the host makes a couple of +/// consecutive frames late; a sender skew makes them late all window long. +pub const VIDEO_DELAY_MIN_FRAMES: u32 = 3; /// How long the last audio playout offset is reused after it goes away. pub const VIDEO_OFFSET_HOLD_NS: u64 = 500_000_000; /// Video-only fallback: clamp on drift between stream and system clock. @@ -443,6 +460,9 @@ mod tests { assert_eq!(VIDEO_PACING_SLACK_NS, 1_000_000); // IRL_VIDEO_PACING_SLACK_NS assert_eq!(VIDEO_PACING_LEAD_TICKS, 2); // IRL_VIDEO_PACING_LEAD_TICKS assert_eq!(VIDEO_ANCHOR_WAIT_MARGIN_MS, 1000); + assert_eq!(VIDEO_DELAY_MAX_MS, 1000); + assert_eq!(VIDEO_DELAY_WINDOW_MS, 1000); + assert_eq!(VIDEO_DELAY_MIN_FRAMES, 3); assert_eq!(VIDEO_PACING_MAX_LEAD_NS, 50_000_000); // IRL_VIDEO_PACING_MAX_LEAD_NS assert_eq!(VIDEO_CANVAS_TICK_DEFAULT_NS, 16_666_667); // IRL_VIDEO_CANVAS_TICK_DEFAULT_NS assert_eq!(VIDEO_PACING_MAX_WAIT_MS, 50); // IRL_VIDEO_PACING_MAX_WAIT_MS diff --git a/crates/irl-core/src/lib.rs b/crates/irl-core/src/lib.rs index 1e14288..09b4036 100644 --- a/crates/irl-core/src/lib.rs +++ b/crates/irl-core/src/lib.rs @@ -23,6 +23,7 @@ pub mod speed; pub mod stats; pub mod timing; pub mod url_opts; +pub mod video_delay; pub mod video_time; pub use audio_buffer::{AudioBuffer, BufferState}; @@ -35,3 +36,4 @@ pub use speed::{ }; pub use stats::{StatKind, StatValue, StatsSnapshot}; pub use url_opts::url_awaits_caller; +pub use video_delay::{DelayRaise, VideoDelay}; diff --git a/crates/irl-core/src/pacing.rs b/crates/irl-core/src/pacing.rs index e618b9a..0d6620b 100644 --- a/crates/irl-core/src/pacing.rs +++ b/crates/irl-core/src/pacing.rs @@ -153,6 +153,21 @@ impl PacingQueue { self.entries.front().map(|e| e.due_ns) } + /// The head frame itself, for the caller's per-frame bookkeeping before + /// it decides whether to pop it. + pub fn head(&self) -> Option<&F> { + self.entries.front().map(|e| &e.frame) + } + + /// Move every due time later by `delta_ns`: the video delay was raised. + /// Unlike [`Self::reschedule`] this needs no mapping, so it also serves a + /// queue scheduled on the video-only fallback. + pub fn shift(&mut self, delta_ns: u64) { + for entry in &mut self.entries { + entry.due_ns = entry.due_ns.saturating_add(delta_ns); + } + } + /// Pop the head. pub fn pop(&mut self) -> Option { let entry = self.entries.pop_front()?; @@ -289,6 +304,30 @@ mod tests { assert_eq!(q.next_due(), Some(100_000_000)); } + #[test] + fn shift_moves_every_due_time_and_keeps_the_head() { + let mut q = queue(); + q.push( + TestFrame { + pts_ns: 0, + bytes: 1, + }, + 1_000, + ); + q.push( + TestFrame { + pts_ns: 40, + bytes: 1, + }, + 1_040, + ); + q.shift(500); + assert_eq!(q.next_due(), Some(1_500)); + assert_eq!(q.head().map(|f| f.pts_ns), Some(0)); + q.pop(); + assert_eq!(q.next_due(), Some(1_540)); + } + #[test] fn empty_queue_has_no_verdict() { let mut q = queue(); diff --git a/crates/irl-core/src/stats.rs b/crates/irl-core/src/stats.rs index 57c447e..1c7cb6b 100644 --- a/crates/irl-core/src/stats.rs +++ b/crates/irl-core/src/stats.rs @@ -24,7 +24,7 @@ pub enum StatValue { Bool(bool), } -/// The 27 stat fields in proc-declaration order. +/// The 28 stat fields in proc-declaration order. pub const FIELDS: &[(&str, StatKind)] = &[ ("buffer_fill_ms", StatKind::Int), ("current_speed", StatKind::Float), @@ -50,6 +50,7 @@ pub const FIELDS: &[(&str, StatKind)] = &[ ("video_corrupt_held", StatKind::Int), ("video_lead_ms", StatKind::Int), ("video_lead_excess", StatKind::Int), + ("video_delay_ms", StatKind::Int), ("stream_delay_ms", StatKind::Int), ("low_latency_audio", StatKind::Bool), ("reconnect_count", StatKind::Int), @@ -106,6 +107,9 @@ pub struct StatsSnapshot { pub video_lead_ms: i64, /// Video lead excess events. pub video_lead_excess: i64, + /// Standing delay added to the video schedule so late-arriving video can + /// still be paced; lip sync is off by this much. + pub video_delay_ms: i64, /// Estimated end-to-end delay. pub stream_delay_ms: i64, /// Low latency audio enabled. @@ -153,6 +157,7 @@ impl StatsSnapshot { StatValue::Int(self.video_corrupt_held), StatValue::Int(self.video_lead_ms), StatValue::Int(self.video_lead_excess), + StatValue::Int(self.video_delay_ms), StatValue::Int(self.stream_delay_ms), StatValue::Bool(self.low_latency_audio), StatValue::Int(self.reconnect_count), @@ -190,7 +195,8 @@ mod tests { /// The declaration `irl_source_create` passed to `proc_handler_add` /// (`src/irl-source.c`), with `out int video_decoder_flushes` removed — - /// that stat was always zero and is not ported. + /// that stat was always zero and is not ported — and `out int + /// video_delay_ms` added for the port's standing video delay. const C_DECLARATION: &str = "void get_stats(out int buffer_fill_ms, \ out float current_speed, out bool adaptive_latency_control, \ out bool reconnecting, \ @@ -206,12 +212,12 @@ out int audio_output_restarts, out int obs_lead_ms, \ out int audio_decoder_flushes, \ out int video_corrupt_frames, out int video_corrupt_held, \ out int video_lead_ms, out int video_lead_excess, \ -out int stream_delay_ms, out bool low_latency_audio, \ +out int video_delay_ms, out int stream_delay_ms, out bool low_latency_audio, \ out int reconnect_count)"; #[test] - fn there_are_twenty_seven_fields() { - assert_eq!(FIELDS.len(), 27); + fn there_are_twenty_eight_fields() { + assert_eq!(FIELDS.len(), 28); // video_decoder_flushes was removed (it was always 0 in C). assert!( !FIELDS @@ -263,6 +269,7 @@ out int reconnect_count)"; video_corrupt_held: 19, video_lead_ms: 20, video_lead_excess: 21, + video_delay_ms: 24, stream_delay_ms: 22, low_latency_audio: true, reconnect_count: 23, @@ -300,7 +307,7 @@ out int reconnect_count)"; assert_eq!(values[1], StatValue::Float(1.05)); assert_eq!(values[2], StatValue::Bool(true)); assert_eq!(values[3], StatValue::Bool(true)); - assert_eq!(values[25], StatValue::Bool(true)); + assert_eq!(values[26], StatValue::Bool(true)); // Spot-check the by-name accessor against the same snapshot. assert_eq!(snap.get("buffer_fill_ms"), Some(StatValue::Int(1))); diff --git a/crates/irl-core/src/video_delay.rs b/crates/irl-core/src/video_delay.rs new file mode 100644 index 0000000..5a5feb6 --- /dev/null +++ b/crates/irl-core/src/video_delay.rs @@ -0,0 +1,382 @@ +//! The standing delay on the video schedule. +//! +//! Pacing hands each frame to libobs a delivery lead before it is due, which +//! is what keeps libobs's async queue about one frame deep and the cadence +//! smooth. That only works for a frame that is in hand at least a lead before +//! its due time. How early it is in hand — its *arrival margin*, the due time +//! minus the moment its packet reached the video thread — is set by the +//! sender: video that leaves the encoder later than the audio of the same +//! instant has that much less margin, and once that skew exceeds what Target +//! Buffer covers the margin is negative and every frame is late. A late frame +//! goes out on arrival, unpaced, and bursty arrival then shows as dropped +//! frames, because libobs discards a frame it finds behind its play head +//! whenever a newer one is already queued. That is the "low fps" a stream with +//! trailing video shows, and it is what the delay here prevents. +//! +//! The delay is added to every due time and sized from the worst margin seen, +//! so that frames are in hand a full lead early again. It trades a fixed, +//! known lip-sync error for a smooth picture; the caller's log line says how +//! much more Target Buffer would take the error back to zero. Before libobs's +//! play head is anchored nothing has been shown yet, so a shortfall is acted +//! on at once. Afterwards a raise moves the picture (one frame holds for the +//! size of the raise), so it takes a shortfall that recurs across a window, +//! not a single late frame from a scheduling hiccup. The delay never shrinks +//! within a connection: lowering it would be a second visible step, for a +//! margin that might well tighten again. + +/// One change of the delay, for the caller to log and mirror into the stats. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DelayRaise { + /// The delay before. + pub from_ns: u64, + /// The delay now. + pub to_ns: u64, + /// Frames that fell short in the window that caused the raise; zero for a + /// raise before the anchor, which one frame is enough for. + pub frames: u32, + /// The margin called for more than the ceiling allows. + pub capped: bool, +} + +/// Late frames seen since the play head was anchored, awaiting a verdict. +#[derive(Debug, Clone, Copy)] +struct Window { + since_ns: u64, + last_ns: u64, + frames: u32, + /// The largest delay any frame in the window needed. + worst_ns: u64, +} + +/// The standing delay and the observation window behind it. +#[derive(Debug)] +pub struct VideoDelay { + delay_ns: u64, + max_ns: u64, + window_ns: u64, + min_frames: u32, + window: Option, +} + +impl VideoDelay { + /// A zero delay that may grow to `max_ns`, and after the anchor only when + /// at least `min_frames` frames fall short within `window_ns`, spread over + /// at least half of it. + pub fn new(max_ns: u64, window_ns: u64, min_frames: u32) -> Self { + Self { + delay_ns: 0, + max_ns, + window_ns, + min_frames, + window: None, + } + } + + /// The delay every due time carries. + pub fn delay_ns(&self) -> u64 { + self.delay_ns + } + + /// Back to zero, for a new connection or a cleared source. + pub fn reset(&mut self) { + self.delay_ns = 0; + self.window = None; + } + + /// The whole delay a frame needs so that it is in hand `lead_ns` before it + /// is due, in whole canvas ticks. `due_ns` is the frame's scheduled time + /// *including* the current delay, as the pacing queue holds it, and + /// `received_ns` when its packet arrived. + fn required_ns(&self, due_ns: u64, received_ns: u64, lead_ns: u64, tick_ns: u64) -> u64 { + let margin_ns = due_ns as i64 - self.delay_ns as i64 - received_ns as i64; + let shortfall_ns = lead_ns as i64 - margin_ns; + if shortfall_ns <= 0 { + return 0; + } + round_up_to_tick(shortfall_ns as u64, tick_ns) + } + + /// Raise to `need_ns` if that is more than the delay already is. + fn raise_to(&mut self, need_ns: u64, frames: u32) -> Option { + let to_ns = need_ns.min(self.max_ns); + if to_ns <= self.delay_ns { + return None; + } + let raise = DelayRaise { + from_ns: self.delay_ns, + to_ns, + frames, + capped: need_ns > self.max_ns, + }; + self.delay_ns = to_ns; + Some(raise) + } + + /// A frame considered for anchoring the play head. Nothing has been shown + /// yet, so a shortfall raises the delay at once. + pub fn before_anchor( + &mut self, + due_ns: u64, + received_ns: u64, + lead_ns: u64, + tick_ns: u64, + ) -> Option { + self.window = None; + let need_ns = self.required_ns(due_ns, received_ns, lead_ns, tick_ns); + self.raise_to(need_ns, 0) + } + + /// A frame handed over after the anchor. A shortfall is recorded; the + /// delay is raised only once the window says it recurs. + pub fn note( + &mut self, + now_ns: u64, + due_ns: u64, + received_ns: u64, + lead_ns: u64, + tick_ns: u64, + ) -> Option { + // Against the delay its due time carries: a raise below must not make + // this frame look short by the amount just added. + let need_ns = self.required_ns(due_ns, received_ns, lead_ns, tick_ns); + // An elapsed window is judged on what it holds, before this frame + // starts the next one. + let raise = self.expire(now_ns); + if need_ns > self.delay_ns { + let window = self.window.get_or_insert(Window { + since_ns: now_ns, + last_ns: now_ns, + frames: 0, + worst_ns: 0, + }); + window.last_ns = now_ns; + window.frames += 1; + window.worst_ns = window.worst_ns.max(need_ns); + } + raise + } + + /// Judge a window that has run its course. Called once per pacing cycle as + /// well as from [`Self::note`], so a window whose late frames simply + /// stopped is closed rather than kept open for the next one. + pub fn expire(&mut self, now_ns: u64) -> Option { + let window = self.window?; + if now_ns.saturating_sub(window.since_ns) < self.window_ns { + return None; + } + self.window = None; + let recurring = window.frames >= self.min_frames + && window.last_ns.saturating_sub(window.since_ns) >= self.window_ns / 2; + if !recurring { + return None; + } + self.raise_to(window.worst_ns, window.frames) + } +} + +/// `ns` rounded up to a whole number of canvas ticks. libobs displays on its +/// ticks, so a fraction of one buys nothing. +fn round_up_to_tick(ns: u64, tick_ns: u64) -> u64 { + if tick_ns == 0 { + return ns; + } + ns.div_ceil(tick_ns) * tick_ns +} + +#[cfg(test)] +mod tests { + use super::*; + + const TICK: u64 = 16_666_667; + const LEAD: u64 = 2 * TICK; + const WINDOW: u64 = 1_000_000_000; + const MAX: u64 = 1_000_000_000; + + fn delay() -> VideoDelay { + VideoDelay::new(MAX, WINDOW, 3) + } + + #[test] + fn a_frame_with_a_full_lead_in_hand_needs_no_delay() { + let mut d = delay(); + assert_eq!(d.before_anchor(1_000 + LEAD, 1_000, LEAD, TICK), None); + assert_eq!( + d.before_anchor(1_000 + 500_000_000, 1_000, LEAD, TICK), + None + ); + assert_eq!(d.delay_ns(), 0); + } + + #[test] + fn a_frame_with_no_margin_gets_the_lead_before_the_anchor() { + let mut d = delay(); + let raise = d.before_anchor(5_000, 5_000, LEAD, TICK).expect("raised"); + assert_eq!(raise.from_ns, 0); + assert_eq!(raise.to_ns, LEAD); + assert!(!raise.capped); + assert_eq!(d.delay_ns(), LEAD); + } + + #[test] + fn a_late_frame_gets_its_lateness_plus_the_lead_in_whole_ticks() { + let mut d = delay(); + // 150 ms late: 150 + 33.3 = 183.3 ms, which is 11 ticks (183.33 ms). + let received = 1_000_000_000; + let due = received - 150_000_000; + let raise = d.before_anchor(due, received, LEAD, TICK).expect("raised"); + assert_eq!(raise.to_ns, 11 * TICK); + } + + #[test] + fn the_delay_only_ever_grows_before_the_anchor() { + let mut d = delay(); + d.before_anchor(0, 100_000_000, LEAD, TICK).expect("raised"); + let first = d.delay_ns(); + // A frame with more margin (its due already carries the delay). + assert_eq!( + d.before_anchor(first + 100_000_000, 0, LEAD, TICK), + None, + "a smaller shortfall must not lower the delay" + ); + assert_eq!(d.delay_ns(), first); + // A frame with less raises it further. + let more = d + .before_anchor(first, 200_000_000, LEAD, TICK) + .expect("raised"); + assert_eq!(more.from_ns, first); + assert!(more.to_ns > first); + } + + #[test] + fn the_existing_delay_counts_toward_the_margin() { + let mut d = delay(); + d.before_anchor(0, 0, LEAD, TICK) + .expect("raised to the lead"); + // Due times now carry the delay: a frame due `LEAD` after it arrived + // has, net of the delay, no margin — exactly what the delay covers. + assert_eq!(d.before_anchor(10_000 + LEAD, 10_000, LEAD, TICK), None); + } + + #[test] + fn the_ceiling_caps_the_delay_and_says_so() { + let mut d = delay(); + let received = 5_000_000_000; + let raise = d + .before_anchor(received - 2_000_000_000, received, LEAD, TICK) + .expect("raised"); + assert_eq!(raise.to_ns, MAX); + assert!(raise.capped); + // Still short at the ceiling: nothing more to do, nothing to report. + assert_eq!( + d.before_anchor(received - 3_000_000_000, received, LEAD, TICK), + None + ); + } + + #[test] + fn after_the_anchor_one_late_frame_does_not_raise_the_delay() { + let mut d = delay(); + let t0 = 10_000_000_000; + assert_eq!(d.note(t0, t0, t0, LEAD, TICK), None); + assert_eq!(d.expire(t0 + WINDOW), None); + assert_eq!(d.delay_ns(), 0); + } + + #[test] + fn after_the_anchor_a_burst_of_late_frames_does_not_raise_the_delay() { + let mut d = delay(); + let t0 = 10_000_000_000; + // Six frames late within 100 ms: one hiccup, not a standing shortfall. + for i in 0..6 { + let t = t0 + i * TICK; + assert_eq!(d.note(t, t, t, LEAD, TICK), None); + } + assert_eq!(d.expire(t0 + WINDOW), None); + assert_eq!(d.delay_ns(), 0); + } + + #[test] + fn after_the_anchor_a_recurring_shortfall_raises_the_delay_to_its_worst() { + let mut d = delay(); + let t0 = 10_000_000_000; + // Three frames spread across the window, 10 ms, 0 ms and 5 ms of + // margin against a 33.3 ms lead: the worst needs 33.3 ms, two ticks. + assert_eq!(d.note(t0, t0 + 10_000_000, t0, LEAD, TICK), None); + assert_eq!( + d.note( + t0 + 600_000_000, + t0 + 600_000_000, + t0 + 600_000_000, + LEAD, + TICK + ), + None + ); + assert_eq!( + d.note( + t0 + 900_000_000, + t0 + 905_000_000, + t0 + 900_000_000, + LEAD, + TICK + ), + None, + "the window has not run its course" + ); + let raise = d.expire(t0 + WINDOW).expect("raised at the window's end"); + assert_eq!(raise.to_ns, 2 * TICK); + assert_eq!(raise.frames, 3); + assert_eq!(d.delay_ns(), 2 * TICK); + } + + #[test] + fn a_late_frame_after_the_window_judges_it_and_starts_the_next() { + let mut d = delay(); + let t0 = 10_000_000_000; + for i in 0..3 { + let t = t0 + i * 400_000_000; + assert_eq!(d.note(t, t, t, LEAD, TICK), None); + } + // The fourth frame arrives after the window elapsed: the raise comes + // out of this call, judged on the three before it. + let t = t0 + WINDOW + 1; + let raise = d.note(t, t, t, LEAD, TICK).expect("raised"); + assert_eq!(raise.frames, 3); + assert_eq!(raise.to_ns, LEAD); + // The fourth frame needed the lead too, and the raise just provided + // it: it does not start a new window. + assert_eq!(d.expire(t + WINDOW), None); + } + + #[test] + fn frames_covered_by_the_delay_are_not_counted() { + let mut d = delay(); + d.before_anchor(0, 0, LEAD, TICK) + .expect("raised to the lead"); + let t0 = 10_000_000_000; + for i in 0..5 { + let t = t0 + i * 300_000_000; + // Due carries the delay; net margin is zero, which the delay covers. + assert_eq!(d.note(t, t + LEAD, t, LEAD, TICK), None); + } + assert_eq!(d.expire(t0 + 2 * WINDOW), None); + assert_eq!(d.delay_ns(), LEAD); + } + + #[test] + fn reset_returns_to_zero() { + let mut d = delay(); + d.before_anchor(0, 0, LEAD, TICK).expect("raised"); + d.reset(); + assert_eq!(d.delay_ns(), 0); + assert_eq!(d.expire(u64::MAX), None); + } + + #[test] + fn rounding_goes_up_to_whole_ticks() { + assert_eq!(round_up_to_tick(1, TICK), TICK); + assert_eq!(round_up_to_tick(TICK, TICK), TICK); + assert_eq!(round_up_to_tick(TICK + 1, TICK), 2 * TICK); + assert_eq!(round_up_to_tick(12_345, 0), 12_345); + } +} diff --git a/crates/irl-source/src/receiver/decode.rs b/crates/irl-source/src/receiver/decode.rs index 806173e..85d928e 100644 --- a/crates/irl-source/src/receiver/decode.rs +++ b/crates/irl-source/src/receiver/decode.rs @@ -200,6 +200,7 @@ impl Receiver { packet, pts_ns, bytes, + received_ns: obs::time::gettime_ns(), }, &self.shared.lifetime, ), diff --git a/crates/irl-source/src/receiver/stream.rs b/crates/irl-source/src/receiver/stream.rs index 9ccdc6f..b738028 100644 --- a/crates/irl-source/src/receiver/stream.rs +++ b/crates/irl-source/src/receiver/stream.rs @@ -555,7 +555,7 @@ impl Receiver { obs_lead={}ms chunk={}@{} \ stream_chunk={}ms obs_chunk={}ms \ restarts={} av_drift={}ms reanchors={} \ - vlead={}ms peak={}ms excess={} vfps={:.1} \ + vlead={}ms peak={}ms excess={} vdelay={}ms vfps={:.1} \ pktq={}/{}({}KB,{}ms) paced={}/{}({}MB) early={} eagain={}/{} pktdrop={}/{} res={}x{}", conn.total_video_frames.load(Relaxed), conn.total_audio_frames.load(Relaxed), @@ -594,6 +594,7 @@ impl Receiver { conn.video_lead_ns.load(Relaxed) / 1_000_000, lifetime.video_lead_peak_ns.load(Relaxed) / 1_000_000, lifetime.video_lead_excess.load(Relaxed), + conn.video_delay_ns.load(Relaxed) / 1_000_000, if video_frame_interval_ns > 0 { 1_000_000_000.0 / video_frame_interval_ns as f64 } else { diff --git a/crates/irl-source/src/shared.rs b/crates/irl-source/src/shared.rs index b89b363..167c6ae 100644 --- a/crates/irl-source/src/shared.rs +++ b/crates/irl-source/src/shared.rs @@ -205,6 +205,8 @@ pub struct ConnStats { pub last_frames_out: AtomicU32, pub last_samples_per_sec: AtomicU32, pub video_lead_ns: AtomicI64, + /// Mirror of the video thread's standing delay, for the stats. + pub video_delay_ns: AtomicU64, /// EMA of decoded PTS deltas; written by the receiver, read by video. pub video_frame_interval_ns: AtomicI64, // Mirrors of the video thread's anchors for stats / media_get_state. @@ -292,6 +294,10 @@ pub struct TimedPacket { pub packet: ffmpeg::Packet, pub pts_ns: i64, pub bytes: usize, + /// OBS clock when the receiver queued it. The frames it decodes into can + /// be in hand no earlier, so this is where their arrival margin is + /// measured from (`irl_core::video_delay`). + pub received_ns: u64, } /// What the receiver sends the video thread, in order. diff --git a/crates/irl-source/src/source.rs b/crates/irl-source/src/source.rs index 7f7da04..11b0d19 100644 --- a/crates/irl-source/src/source.rs +++ b/crates/irl-source/src/source.rs @@ -495,6 +495,7 @@ fn snapshot(state: &ObsState, lifetime: &LifetimeStats) -> StatsSnapshot { snap.video_corrupt_frames = conn.video_corrupt_frames.load(Relaxed) as i64; snap.video_corrupt_held = conn.video_corrupt_held.load(Relaxed) as i64; snap.video_lead_ms = conn.video_lead_ns.load(Relaxed) / 1_000_000; + snap.video_delay_ms = (conn.video_delay_ns.load(Relaxed) / 1_000_000) as i64; // Stream delay: how far behind real time the video output is, computed as // wall clock minus the anchored video PTS. Includes SRT latency, decode diff --git a/crates/irl-source/src/video/output.rs b/crates/irl-source/src/video/output.rs index c4ffd17..798337d 100644 --- a/crates/irl-source/src/video/output.rs +++ b/crates/irl-source/src/video/output.rs @@ -212,8 +212,13 @@ impl VideoThread { // No audio-stream test: a published mapping already implies the pump // handed OBS a real chunk, so it implies the audio stream. + // Both schedules carry the standing video delay; see + // `VideoThread::settle_anchor_candidate`. + let delay_ns = self.delay.delay_ns(); + if obs_end != 0 && buffered_end > 0 { - let mapped = video_time::map_through_playout(pts_ns, obs_end, buffered_end); + let mapped = video_time::map_through_playout(pts_ns, obs_end, buffered_end) + .saturating_add(delay_ns); self.record_lead(mapped as i64, now, frame_interval_ns); return mapped; } @@ -234,7 +239,8 @@ impl VideoThread { self.shared.conn.video_ts_init.store(true, Relaxed); } - let mut computed = video_time::fallback_anchor(pts_ns, self.pts_base, self.sys_base, now); + let mut computed = video_time::fallback_anchor(pts_ns, self.pts_base, self.sys_base, now) + .saturating_add(delay_ns); // Startup fallback before the audio playout mapping exists. Here the // mapping cannot stand in for "there is audio" — the whole point is @@ -277,7 +283,8 @@ impl VideoThread { /// every pacing cycle would report the queue rather than the stream. /// `None` when there is no audio to slave to, or when the mapping has been /// gone long enough that holding it would be a guess; the caller then - /// keeps the due times the frames arrived with. + /// keeps the due times the frames arrived with. The standing video delay + /// rides on the offset, so a reschedule keeps it. pub fn playout_offset(&mut self) -> Option { let (obs_end, buffered_end) = { let state = self.shared.audio_state(); @@ -294,7 +301,7 @@ impl VideoThread { return None; } - Some(self.playout_offset_ns) + Some(self.playout_offset_ns + self.delay.delay_ns() as i64) } /// `video_record_lead` (`video-handler.c:285-327`): how far ahead of wall diff --git a/crates/irl-source/src/video/thread.rs b/crates/irl-source/src/video/thread.rs index 6cbdb17..5dcee4c 100644 --- a/crates/irl-source/src/video/thread.rs +++ b/crates/irl-source/src/video/thread.rs @@ -16,6 +16,7 @@ use std::time::Duration; use ffmpeg::{AVPixelFormat, Frame, FramePool, Scaler}; use irl_core::consts; use irl_core::pacing::{DueVerdict, PacedFrame, PacingQueue}; +use irl_core::video_delay::{DelayRaise, VideoDelay}; use crate::shared::{Shared, VideoDecoder, VideoMsg}; use crate::video::VideoSink; @@ -24,15 +25,18 @@ use crate::video::intake::DecodeState; /// A frame waiting for its due time. `pts_ns` and `bytes` are cached at /// intake: the pacing queue re-derives due times from the PTS every cycle, and -/// the byte total bounds the queue. +/// the byte total bounds the queue. `received_ns` is when the packet it was +/// decoded from reached this thread, which is what its arrival margin is +/// measured from. pub struct Paced { frame: Frame, pts_ns: i64, bytes: usize, + received_ns: u64, } impl Paced { - fn new(frame: Frame) -> Self { + fn new(frame: Frame, received_ns: u64) -> Self { let pts_ns = frame.pts(); // `av_image_get_buffer_size(fmt, w, h, 1)`, as `pacing_frame_bytes`. let bytes = frame.image_buffer_size().unwrap_or(0); @@ -40,6 +44,7 @@ impl Paced { frame, pts_ns, bytes, + received_ns, } } @@ -47,6 +52,11 @@ impl Paced { pub fn frame(&self) -> &Frame { &self.frame } + + /// OBS clock at which the packet behind this frame arrived. + pub fn received_ns(&self) -> u64 { + self.received_ns + } } impl PacedFrame for Paced { @@ -109,11 +119,15 @@ pub struct VideoThread { pub(crate) lead_warn_time_ns: u64, /// Set while libobs's async play head is unanchored, so the next frame out /// goes at its due time rather than a lead early. See - /// [`Self::emit_slack_ns`]. + /// [`Self::emit_slack_ns`] and [`Self::settle_anchor_candidate`]. anchor_pending: bool, /// Where the wait for the audio playout mapping stands. See /// [`Self::awaiting_audio_mapping`]. anchor_wait: AnchorWait, + /// The standing delay on the video schedule, so that video which reaches + /// this thread too late to be paced against the audio playout still can + /// be. See [`Self::settle_anchor_candidate`]. + pub(crate) delay: VideoDelay, /// The OBS canvas tick. Injectable so tests can drive pacing without a /// running libobs — `obs_get_frame_interval_ns` reads libobs's global /// video state and faults when `obs_startup` never ran. @@ -155,6 +169,11 @@ impl VideoThread { // A fresh source has libobs's `last_frame_ts` at 0 too. anchor_pending: true, anchor_wait: AnchorWait::Idle, + delay: VideoDelay::new( + consts::VIDEO_DELAY_MAX_MS * 1_000_000, + consts::VIDEO_DELAY_WINDOW_MS * 1_000_000, + consts::VIDEO_DELAY_MIN_FRAMES, + ), canvas_tick_ns: Box::new(obs::time::canvas_frame_interval_ns), sink, } @@ -202,6 +221,8 @@ impl VideoThread { // to 0, so the next frame re-anchors the play head. self.anchor_pending = true; self.anchor_wait = AnchorWait::Idle; + // The delay was sized for the connection that just ended. + self.reset_delay(); self.sink.output_video_none(); return Duration::ZERO; } @@ -260,6 +281,7 @@ impl VideoThread { if let Some(VideoMsg::Decoder(decoder)) = shared.video.pop() { self.decoder = Some(*decoder); self.state.reset(); + self.reset_delay(); } } while self.pacing.has_room() { @@ -272,8 +294,10 @@ impl VideoThread { // already paced or was cleared with the disconnect. self.decoder = Some(*decoder); self.state.reset(); + self.reset_delay(); } VideoMsg::Packet(packet) => { + let received_ns = packet.received_ns; let mut produced = std::mem::take(&mut self.decoded); if let Some(decoder) = self.decoder.as_mut() { decode::decode_packet( @@ -286,7 +310,7 @@ impl VideoThread { ); } for frame in produced.drain(..) { - self.pace_decoded(frame); + self.pace_decoded(frame, received_ns); } self.decoded = produced; } @@ -295,11 +319,12 @@ impl VideoThread { } /// Copy one decoded frame out of the hardware pool and schedule it. + /// `received_ns` is when the packet it came from reached this thread. /// /// Public as the seam the pacing tests use to put a frame in front of the /// loop without a decoder; production reaches it only through /// [`Self::decode_intake`]. - pub fn pace_decoded(&mut self, frame: Frame) { + pub fn pace_decoded(&mut self, frame: Frame, received_ns: u64) { let sysmem = self.to_sysmem(&frame); let due_ns = match &sysmem { Some(f) => self.due_time(f), @@ -308,7 +333,7 @@ impl VideoThread { // Releases the decoder's surface before the next packet is sent. drop(frame); if let Some(f) = sysmem { - self.pacing.push(Paced::new(f), due_ns); + self.pacing.push(Paced::new(f, received_ns), due_ns); } } @@ -340,8 +365,11 @@ impl VideoThread { /// video is what the un-paced path did all the time, and it beats a hole /// in the picture. fn pacing_emit_due(&mut self, now_ns: u64, slack_ns: i64) { + let tick_ns = self.canvas_tick_ns(); if self.anchor_pending { - self.drop_stale_before_anchor(now_ns); + self.settle_anchor_candidate(now_ns, tick_ns); + } else if let Some(raise) = self.delay.expire(now_ns) { + self.apply_delay_raise(raise, false); } loop { // While the play head needs anchoring, a hard ceiling must not @@ -359,6 +387,15 @@ impl VideoThread { let Some(paced) = self.pacing.pop() else { return; }; + if !self.anchor_pending { + let lead_ns = self.delivery_lead_ns(tick_ns); + if let Some(raise) = + self.delay + .note(now_ns, due_ns, paced.received_ns(), lead_ns, tick_ns) + { + self.apply_delay_raise(raise, false); + } + } let submitted = self.output_frame(paced.frame(), due_ns); // The play head is only anchored by a frame libobs actually // received, at its real due time. A conversion that failed @@ -418,34 +455,73 @@ impl VideoThread { false } - /// The frame that anchors libobs's play head must go out at its due time, - /// so a head that is already past due cannot be it: drop such frames until - /// one that is on time is at the head. + /// Settle the frame that will anchor libobs's play head. + /// + /// Two things are decided for each head frame, in this order. First its + /// arrival margin: nothing has been shown yet, so a frame that reached this + /// thread less than a delivery lead before it is due raises the standing + /// video delay on the spot ([`VideoDelay::before_anchor`]) and the queue is + /// moved onto it. That is what keeps a sender whose video trails its audio + /// by more than Target Buffer covers from playing unpaced — every frame + /// late, handed over on arrival, and dropped by libobs whenever two arrive + /// inside one canvas tick — or, before this existed, from being dropped + /// here indefinitely. The delay is a fixed lip-sync error in return for a + /// smooth picture, and the log line says how much more Target Buffer would + /// remove it. /// - /// The mapping arriving is what makes frames overdue here — everything - /// decoded during the wait maps to the moments its audio was discarded by - /// the warm-up or already played — and handing them over would anchor the - /// connection late by however stale the first one was. They amount to a - /// fraction of a second at connection start and nothing has been shown yet. + /// Second, with audio present, whether the head is stale. The anchoring + /// frame must go out at its due time, so a head already past due cannot be + /// it, and such frames are dropped until one that is on time is at the + /// head. The mapping arriving is what makes frames overdue here — + /// everything decoded during the wait maps to the moments its audio was + /// discarded by the warm-up or already played — and handing them over would + /// anchor the connection late by however stale the first one was. They + /// amount to a fraction of a second at connection start and nothing has + /// been shown yet. Their margins are still valid samples: a margin is set + /// by when a packet arrived against when its audio plays, not by how long + /// the frame has been queued, so the backlog measures the same sender skew + /// the live frames will. /// /// "Past due" is measured against a canvas tick, not the emit slack: libobs /// quantises display to its ticks anyway, and a box with a coarse timer can /// oversleep by most of one, which must not make it drop every candidate in /// turn. Without audio there is nothing to be in sync with, and the - /// fallback frames go out as they always did. - fn drop_stale_before_anchor(&mut self, now_ns: u64) { - if !self.shared.flags.audio_present.load(Relaxed) { - return; - } - let tick = (self.canvas_tick_ns)().unwrap_or(consts::VIDEO_CANVAS_TICK_DEFAULT_NS) as i64; + /// fallback frames go out as they always did, now a lead after they arrive + /// instead of on arrival. + fn settle_anchor_candidate(&mut self, now_ns: u64, tick_ns: u64) { + let audio_present = self.shared.flags.audio_present.load(Relaxed); + let lead_ns = self.delivery_lead_ns(tick_ns); let mut dropped = 0u32; - while let Some(due_ns) = self.pacing.next_due() { - if now_ns as i64 - due_ns as i64 <= tick { + let mut raised: Option = None; + while let (Some(due_ns), Some(received_ns)) = ( + self.pacing.next_due(), + self.pacing.head().map(Paced::received_ns), + ) { + if let Some(raise) = self + .delay + .before_anchor(due_ns, received_ns, lead_ns, tick_ns) + { + self.pacing.shift(raise.to_ns - raise.from_ns); + // One line for the whole settlement, from the first delay to + // the last. + raised = Some(match raised { + Some(first) => DelayRaise { + from_ns: first.from_ns, + ..raise + }, + None => raise, + }); + continue; + } + if !audio_present || now_ns as i64 - due_ns as i64 <= tick_ns as i64 { break; } self.pacing.pop(); dropped += 1; } + if let Some(raise) = raised { + self.apply_delay_raise(raise, true); + } if dropped > 0 { irl_info!( "Dropped {dropped} stale video frame(s) before anchoring to the audio playout" @@ -453,6 +529,62 @@ impl VideoThread { } } + /// Publish a raised delay: mirror it for the stats and say why, with what + /// the user can do about it. The queue itself is moved by the caller + /// (before the anchor, inside the settling loop) or is already scheduled + /// on the new delay (after it, the next reschedule carries it; the frames + /// queued now are shifted here). + fn apply_delay_raise(&mut self, raise: DelayRaise, at_anchor: bool) { + if !at_anchor { + self.pacing.shift(raise.to_ns - raise.from_ns); + } + self.shared.conn.video_delay_ns.store(raise.to_ns, Relaxed); + let to_ms = raise.to_ns / 1_000_000; + let by_ms = (raise.to_ns - raise.from_ns) / 1_000_000; + let audio_present = self.shared.flags.audio_present.load(Relaxed); + match (at_anchor, audio_present) { + (true, true) => irl_warn!( + "Video arrives too late for its audio to be paced; delaying video by {to_ms}ms so it can be. Raise Target Buffer by at least {to_ms}ms to keep lip sync instead" + ), + (true, false) => { + irl_info!("Video arrives without its pacing lead; delaying video by {to_ms}ms") + } + (false, true) => irl_warn!( + "Video ran late on {} frames in the last {}ms; delaying video by {by_ms}ms more, {to_ms}ms in all. Raise Target Buffer by at least {to_ms}ms to keep lip sync instead", + raise.frames, + consts::VIDEO_DELAY_WINDOW_MS + ), + (false, false) => irl_info!( + "Video ran late on {} frames in the last {}ms; delaying video by {by_ms}ms more, {to_ms}ms in all", + raise.frames, + consts::VIDEO_DELAY_WINDOW_MS + ), + } + if raise.capped { + irl_warn!( + "Video is late by more than the {}ms delay ceiling; the decoder or the host is not keeping up, and frames past it go out on arrival", + consts::VIDEO_DELAY_MAX_MS + ); + } + } + + /// Forget the delay: a new connection or a cleared source sizes its own. + fn reset_delay(&mut self) { + self.delay.reset(); + self.shared.conn.video_delay_ns.store(0, Relaxed); + } + + /// The canvas tick, or the default when libobs has not reported one. + fn canvas_tick_ns(&self) -> u64 { + (self.canvas_tick_ns)().unwrap_or(consts::VIDEO_CANVAS_TICK_DEFAULT_NS) + } + + /// How early a frame is handed to libobs once the play head is anchored; + /// see [`consts::VIDEO_PACING_LEAD_TICKS`]. + fn delivery_lead_ns(&self, tick_ns: u64) -> u64 { + (tick_ns * consts::VIDEO_PACING_LEAD_TICKS).min(consts::VIDEO_PACING_MAX_LEAD_NS) + } + /// How early a frame is handed to libobs: the emit slack plus the delivery /// lead. /// @@ -474,9 +606,7 @@ impl VideoThread { if self.anchor_pending { return consts::VIDEO_PACING_SLACK_NS; } - let tick = (self.canvas_tick_ns)().unwrap_or(consts::VIDEO_CANVAS_TICK_DEFAULT_NS); - let lead = (tick * consts::VIDEO_PACING_LEAD_TICKS).min(consts::VIDEO_PACING_MAX_LEAD_NS); - lead as i64 + consts::VIDEO_PACING_SLACK_NS + self.delivery_lead_ns(self.canvas_tick_ns()) as i64 + consts::VIDEO_PACING_SLACK_NS } /// Mirror the pacing counters for the stats line. diff --git a/crates/irl-source/tests/network_sim.rs b/crates/irl-source/tests/network_sim.rs index 84f0bec..7bcb3a5 100644 --- a/crates/irl-source/tests/network_sim.rs +++ b/crates/irl-source/tests/network_sim.rs @@ -557,6 +557,7 @@ fn video_keeps_flowing_while_the_receiver_is_blocked() { packet: ffmpeg::Packet::new().unwrap(), pts_ns: i * 33_333_333, bytes: 4096, + received_ns: 0, }, &sim.shared.lifetime, ); @@ -591,6 +592,7 @@ fn decoded_memory_does_not_grow_with_the_target() { packet: ffmpeg::Packet::new().unwrap(), pts_ns: i * 16_666_667, bytes: 16 * 1024, + received_ns: 0, }, &deep.shared.lifetime, ); diff --git a/crates/irl-source/tests/shell_stats.rs b/crates/irl-source/tests/shell_stats.rs index 54b2529..321ff10 100644 --- a/crates/irl-source/tests/shell_stats.rs +++ b/crates/irl-source/tests/shell_stats.rs @@ -38,6 +38,7 @@ fn distinct_snapshot() -> StatsSnapshot { video_corrupt_held: 119, video_lead_ms: 120, video_lead_excess: 121, + video_delay_ms: 124, stream_delay_ms: 122, low_latency_audio: true, reconnect_count: 123, diff --git a/crates/irl-source/tests/video_pipeline.rs b/crates/irl-source/tests/video_pipeline.rs index 506abe3..8c7b6c1 100644 --- a/crates/irl-source/tests/video_pipeline.rs +++ b/crates/irl-source/tests/video_pipeline.rs @@ -28,6 +28,11 @@ use video::VideoSink; use video::output; use video::thread::VideoThread; +/// A 60fps canvas tick, what `thread_with` reports. +const TICK: u64 = 16_666_667; +/// One 30fps frame interval. +const FRAME: u64 = 33_333_333; + /* ── Harness ──────────────────────────────────────────────── */ /// One `obs_source_output_video` call, flattened. @@ -393,6 +398,7 @@ fn the_packet_queue_is_bounded_by_duration_and_bytes() { packet: ffmpeg::Packet::new().unwrap(), pts_ns: ms(i * 500), bytes: 1024, + received_ns: 0, }, &shared.lifetime, ); @@ -416,6 +422,7 @@ fn the_packet_queue_is_bounded_by_bytes_whatever_the_timestamps_say() { packet: ffmpeg::Packet::new().unwrap(), pts_ns: 0, bytes: 4 * 1024 * 1024, + received_ns: 0, }, &big.lifetime, ); @@ -437,6 +444,7 @@ fn the_channel_delivers_packets_after_the_decoder_that_owns_them() { packet: ffmpeg::Packet::new().unwrap(), pts_ns: 0, bytes: 1, + received_ns: 0, }, &shared.lifetime, ); @@ -516,19 +524,24 @@ fn a_frame_is_handed_over_a_lead_before_it_is_due() { let (mut thread, recorder) = thread_with(shared.clone()); // Anchor the play head first: the very first frame out deliberately gets - // no lead, so a second frame is needed to see one. + // no lead, so a second frame is needed to see one. The video-only fallback + // schedules it at its arrival, so it is delayed by a lead first (see + // `video_without_audio_is_delayed_by_a_lead_and_anchors_on_the_fallback`). + let t0 = obs::time::gettime_ns(); let mut first = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); - first.set_pts(obs::time::gettime_ns() as i64); - thread.pace_decoded(first); - thread.run_once(obs::time::gettime_ns()); + first.set_pts(t0 as i64); + thread.pace_decoded(first, t0); + let now = t0 + 2 * TICK; + thread.run_once(now); assert_eq!(recorder.emitted().len(), 1, "anchor frame emitted"); // A frame due 25ms out: inside two 60fps ticks (33.3ms) of the lead, so - // it should go now even though its due time has not arrived. - let now = obs::time::gettime_ns(); + // it should go now even though its due time has not arrived. (Its due + // time is its PTS on the fallback epoch plus the delay: `t0 + 25 ms + + // 2 ticks`, 25 ms from `now`.) let mut soon = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); - soon.set_pts(now as i64 + 25_000_000); - thread.pace_decoded(soon); + soon.set_pts(t0 as i64 + 25_000_000); + thread.pace_decoded(soon, t0); thread.run_once(now); assert_eq!( recorder.emitted().len(), @@ -537,10 +550,9 @@ fn a_frame_is_handed_over_a_lead_before_it_is_due() { ); // One due far out still waits. - let now = obs::time::gettime_ns(); let mut later = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); - later.set_pts(now as i64 + 500_000_000); - thread.pace_decoded(later); + later.set_pts(t0 as i64 + 500_000_000); + thread.pace_decoded(later, now); let wait = thread.run_once(now); assert_eq!(recorder.emitted().len(), 2, "not due, not delivered"); assert!(!wait.is_zero(), "should sleep toward the lead"); @@ -566,6 +578,7 @@ fn a_full_pacing_queue_does_not_keep_the_video_thread_awake() { packet: ffmpeg::Packet::new().unwrap(), pts_ns: i * 33_333_333, bytes: 2048, + received_ns: 0, }, &shared.lifetime, ); @@ -578,7 +591,7 @@ fn a_full_pacing_queue_does_not_keep_the_video_thread_awake() { while thread.pacing_has_room() { let mut frame = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); frame.set_pts(now as i64 + 60_000_000_000 + pts); - thread.pace_decoded(frame); + thread.pace_decoded(frame, now); pts += 33_333_333; } assert!(!thread.pacing_has_room(), "queue should be at its lead"); @@ -609,6 +622,7 @@ fn a_decoder_handover_is_taken_even_with_no_pacing_room() { packet: ffmpeg::Packet::new().unwrap(), pts_ns: 0, bytes: 1, + received_ns: 0, }, &shared.lifetime, ); @@ -620,6 +634,7 @@ fn a_decoder_handover_is_taken_even_with_no_pacing_room() { packet: ffmpeg::Packet::new().unwrap(), pts_ns: 0, bytes: 1, + received_ns: 0, }, &shared.lifetime, ); @@ -635,11 +650,13 @@ fn a_queued_frame_is_transferred_paced_and_emitted() { // The receiver hands the queue nanosecond PTS. queued.set_pts(5_000_000_000); let source_plane = queued.plane(0).unwrap().as_ptr() as usize; - thread.pace_decoded(queued); + let now = obs::time::gettime_ns(); + thread.pace_decoded(queued, now); - // With no audio mapping the video-only anchor puts the first frame at - // `now`, so one cycle takes it all the way out. - let wait = thread.run_once(obs::time::gettime_ns()); + // With no audio mapping the video-only anchor puts the first frame at its + // arrival, and the delivery lead it then lacks is added as the standing + // delay: one cycle at that due time takes it all the way out. + let wait = thread.run_once(now + 2 * TICK); let emitted = recorder.only(); assert_eq!(emitted.planes[0].0, source_plane, "still zero-copy"); @@ -663,14 +680,15 @@ fn a_future_frame_waits_instead_of_being_emitted() { // 200 ms into the future on the video-only anchor: the fallback anchors on // the first frame, so pace the *second* one forward. queued.set_pts(0); - thread.pace_decoded(queued); + thread.pace_decoded(queued, now); + let now = now + 2 * TICK; thread.run_once(now); assert_eq!(recorder.emitted().len(), 1, "the anchor frame goes out"); let mut later = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); later.set_pts(200_000_000); - thread.pace_decoded(later); - let wait = thread.run_once(obs::time::gettime_ns()); + thread.pace_decoded(later, now); + let wait = thread.run_once(now); assert_eq!(recorder.emitted().len(), 1, "not due yet"); assert_eq!(thread.paced_len(), 1); @@ -687,20 +705,22 @@ fn a_clear_request_drops_the_queue_and_blanks_the_source() { let shared = shared(); let (mut thread, recorder) = thread_with(shared.clone()); + let now = obs::time::gettime_ns(); let mut queued = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); queued.set_pts(0); - thread.pace_decoded(queued); - thread.run_once(obs::time::gettime_ns()); + thread.pace_decoded(queued, now); + let now = now + 2 * TICK; + thread.run_once(now); assert_eq!(recorder.emitted().len(), 1); let mut pending = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); pending.set_pts(10_000_000_000); - thread.pace_decoded(pending); - thread.run_once(obs::time::gettime_ns()); + thread.pace_decoded(pending, now); + thread.run_once(now); assert_eq!(thread.paced_len(), 1, "parked until its due time"); shared.video.request_clear(); - let wait = thread.run_once(obs::time::gettime_ns()); + let wait = thread.run_once(now); assert_eq!(recorder.cleared.load(Relaxed), 1); assert!( @@ -732,7 +752,7 @@ fn queued_frames_reschedule_onto_the_audio_playout_offset() { let mut queued = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); queued.set_pts(10_000_000_000); - thread.pace_decoded(queued); + thread.pace_decoded(queued, now); thread.run_once(now); assert!(recorder.emitted().is_empty(), "due a second from now"); @@ -748,14 +768,18 @@ fn queued_frames_reschedule_onto_the_audio_playout_offset() { thread.run_once(now); assert_eq!(thread.next_due_ns(), Some(now + 500_000_000)); - // Once the offset says "now", it goes out. + // Once the offset says "now", the frame has no margin left to be paced + // with: the delivery lead is added as the standing delay, and it goes out + // at the end of it. { let mut state = shared.audio_state(); state.latest_obs_end_ts_ns = now; } thread.run_once(now); - assert_eq!(recorder.emitted().len(), 1); - assert_eq!(recorder.only().timestamp, now); + assert!(recorder.emitted().is_empty()); + assert_eq!(thread.next_due_ns(), Some(now + 2 * TICK)); + thread.run_once(now + 2 * TICK); + assert_eq!(recorder.only().timestamp, now + 2 * TICK); } #[test] @@ -867,7 +891,7 @@ fn video_waits_for_the_audio_mapping_before_anchoring_the_play_head() { let now = obs::time::gettime_ns(); let mut first = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); first.set_pts(10_000_000_000); - thread.pace_decoded(first); + thread.pace_decoded(first, now); // Well past the fallback due time (+120 ms): still held. let wait = thread.run_once(now + 300_000_000); @@ -888,18 +912,21 @@ fn video_waits_for_the_audio_mapping_before_anchoring_the_play_head() { /// The mapping can also land frames in the past — the ones whose audio the /// warm-up discarded. Handing those over would anchor the play head late by /// however stale the first one was; they are dropped instead, and the first -/// frame that is on time anchors. +/// frame that is on time anchors. Their arrival margins are the same as the +/// live frames' (60 ms here, more than a lead), so the backlog raises no +/// video delay either. #[test] fn frames_already_past_due_when_the_mapping_arrives_do_not_anchor_the_play_head() { let shared = shared_with_audio(); let (mut thread, recorder) = thread_with(shared.clone()); - // 25 fps: 40 ms apart, so no two frames fall inside one 60fps tick. + // 25 fps: 40 ms apart, so no two frames fall inside one 60fps tick, + // arriving in real time over the 160 ms before the mapping. let now = obs::time::gettime_ns(); for i in 0..4 { let mut frame = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); frame.set_pts(10_000_000_000 + i * 40_000_000); - thread.pace_decoded(frame); + thread.pace_decoded(frame, now - 160_000_000 + i as u64 * 40_000_000); } thread.run_once(now); assert!(recorder.emitted().is_empty()); @@ -914,6 +941,7 @@ fn frames_already_past_due_when_the_mapping_arrives_do_not_anchor_the_play_head( "the stale frames must not go out in place of the on-time one" ); assert_eq!(thread.paced_len(), 1, "three stale frames dropped"); + assert_eq!(shared.conn.video_delay_ns.load(Relaxed), 0); thread.run_once(now + 20_000_000); assert_eq!(recorder.only().timestamp, now + 20_000_000); @@ -929,7 +957,7 @@ fn a_frame_late_by_under_a_canvas_tick_still_anchors() { let now = obs::time::gettime_ns(); let mut frame = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); frame.set_pts(10_000_000_000); - thread.pace_decoded(frame); + thread.pace_decoded(frame, now - 50_000_000); publish_mapping(&shared, now, 10_000_000_000); thread.run_once(now + 10_000_000); @@ -950,7 +978,7 @@ fn video_stops_waiting_for_audio_that_never_primes() { let now = obs::time::gettime_ns(); let mut first = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); first.set_pts(10_000_000_000); - thread.pace_decoded(first); + thread.pace_decoded(first, now); thread.run_once(now); assert!(recorder.emitted().is_empty()); @@ -969,24 +997,218 @@ fn video_stops_waiting_for_audio_that_never_primes() { "the stale frame was dropped, not anchored" ); + // The fallback schedules against the real clock (`due_time` reads it), + // so the frame's arrival has to be on that clock too: stamped with the + // test's 1.35 s of pretend time it would look a second late, which in + // production cannot happen (a packet never arrives after a due time the + // fallback capped at now + 200 ms). let mut fresh = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); fresh.set_pts(10_000_000_000 + 1_400_000_000); - thread.pace_decoded(fresh); + thread.pace_decoded(fresh, obs::time::gettime_ns()); let due = thread.next_due_ns().expect("paced on the fallback"); thread.run_once(due); assert_eq!(recorder.only().timestamp, due); } -/// Without an audio stream the video-only fallback anchors immediately, as it -/// always did: there is nothing to wait for and nothing to be in sync with. +/// Without an audio stream there is nothing to wait for and nothing to be in +/// sync with: the video-only fallback anchors as soon as the first frame has +/// its delivery lead. The fallback schedules that frame at its arrival, which +/// leaves it no lead at all, so the lead is added as the standing video delay +/// and the frame goes out two canvas ticks after it arrived. #[test] -fn video_without_audio_anchors_on_the_fallback_at_once() { +fn video_without_audio_is_delayed_by_a_lead_and_anchors_on_the_fallback() { let shared = shared(); let (mut thread, recorder) = thread_with(shared.clone()); + let t0 = obs::time::gettime_ns(); let mut first = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); - first.set_pts(obs::time::gettime_ns() as i64); - thread.pace_decoded(first); - thread.run_once(obs::time::gettime_ns()); + first.set_pts(t0 as i64); + thread.pace_decoded(first, t0); + thread.run_once(t0); + assert!( + recorder.emitted().is_empty(), + "no audio to wait for, but no lead yet either" + ); + assert_eq!(shared.conn.video_delay_ns.load(Relaxed), 2 * TICK); + // The fallback anchors on the clock as `due_time` read it, a hair after + // `t0`, so the due time is two ticks past that rather than past `t0`. + let due = thread.next_due_ns().expect("paced"); + assert!( + due >= t0 + 2 * TICK && due < t0 + 2 * TICK + 1_000_000, + "due {due} vs t0 {t0}" + ); + + thread.run_once(due); + assert_eq!(recorder.only().timestamp, due); +} + +/* ── The standing video delay ─────────────────────────────── */ + +/// The mapping can also put *every* frame in the past: a sender whose video +/// leaves the encoder later than the audio of the same instant by more than +/// Target Buffer covers. Dropping stale frames until an on-time one turns up +/// would drop such a stream forever, and anchoring on a late frame would play +/// the whole connection unpaced — each frame handed over on arrival, and +/// dropped by libobs whenever two arrive inside one canvas tick, which is the +/// "low fps" such a stream shows. The shortfall is measured instead and added +/// to the schedule as a standing delay, sized so that frames are in hand a +/// full delivery lead early again, and the picture is paced from the first +/// frame. +#[test] +fn video_that_trails_its_audio_is_delayed_into_pacing_not_dropped() { + let shared = shared_with_audio(); + let (mut thread, recorder) = thread_with(shared.clone()); + + // Audio has primed such that a video frame arriving now maps 150 ms into + // the past, and every later frame likewise. + let t0 = obs::time::gettime_ns(); + let late = 150_000_000; + publish_mapping(&shared, t0 - late, 10_000_000_000); + + for i in 0..30u64 { + let arrival = t0 + i * FRAME; + let mut frame = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); + frame.set_pts(10_000_000_000 + (i * FRAME) as i64); + thread.pace_decoded(frame, arrival); + thread.run_once(arrival); + } + + // 150 ms of lateness plus the 33.3 ms lead is 183.3 ms: 11 ticks. + let delay = 11 * TICK; + assert_eq!(shared.conn.video_delay_ns.load(Relaxed), delay); + let emitted = recorder.emitted(); + assert_eq!(emitted.len(), 30, "every frame shown, none dropped"); + for (i, frame) in emitted.iter().enumerate() { + let arrival = t0 + i as u64 * FRAME; + assert_eq!(frame.timestamp, arrival - late + delay, "frame {i}"); + } + assert_eq!(thread.paced_len(), 0); +} + +/// A sender whose skew sits right at the edge of what Target Buffer covers +/// gets some frames in hand with a lead and some without. The anchor frame may +/// well be one of the former, and from then on every frame that arrives short +/// is handed over on arrival, unpaced: the same dropped-frame judder in +/// libobs. A shortfall that recurs across a window raises the delay to cover +/// it. A raise moves the picture, so it happens once, not per frame. +#[test] +fn a_recurring_shortfall_after_the_anchor_raises_the_delay_once() { + let shared = shared_with_audio(); + let (mut thread, recorder) = thread_with(shared.clone()); + + // A frame arriving now maps 100 ms out: a comfortable margin to anchor on. + let t0 = obs::time::gettime_ns(); + publish_mapping(&shared, t0 + 100_000_000, 10_000_000_000); + let mut anchor = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); + anchor.set_pts(10_000_000_000); + thread.pace_decoded(anchor, t0); + thread.run_once(t0 + 100_000_000); + assert_eq!(recorder.emitted().len(), 1, "anchored with a full margin"); + assert_eq!(shared.conn.video_delay_ns.load(Relaxed), 0); + + // From here every frame arrives with only 10 ms in hand against a 33.3 ms + // lead, 90 ms later relative to its due time than the anchor frame did. + let arrival_of = |i: u64| t0 + 90_000_000 + i * FRAME; + let mut delay_seen = Vec::new(); + for i in 1..=80u64 { + let mut frame = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); + frame.set_pts(10_000_000_000 + (i * FRAME) as i64); + thread.pace_decoded(frame, arrival_of(i)); + thread.run_once(arrival_of(i)); + delay_seen.push(shared.conn.video_delay_ns.load(Relaxed)); + } + thread.run_once(arrival_of(81)); + + // The window opened with frame 1 and ran its course a second later, at + // frame 32: raised then, by two ticks, and never again for the same + // shortfall. + assert_eq!(delay_seen[30], 0, "frame 31: still inside the window"); + assert_eq!(delay_seen[31], 2 * TICK, "frame 32: raised"); + assert!( + delay_seen[31..].iter().all(|&d| d == 2 * TICK), + "raised once" + ); + + let emitted = recorder.emitted(); + assert_eq!(emitted.len(), 81, "no frame dropped across the raise"); + for (i, frame) in emitted.iter().enumerate().skip(1) { + let i = i as u64; + let due = arrival_of(i) + 10_000_000; + let expected = if i < 32 { due } else { due + 2 * TICK }; + assert_eq!(frame.timestamp, expected, "frame {i}"); + } +} + +/// A few late frames in a row are a scheduling hiccup on the host, not a +/// sender skew: they go out late, as they always did, and the delay stays. +#[test] +fn a_single_late_burst_after_the_anchor_does_not_raise_the_delay() { + let shared = shared_with_audio(); + let (mut thread, recorder) = thread_with(shared.clone()); + + let t0 = obs::time::gettime_ns(); + publish_mapping(&shared, t0 + 100_000_000, 10_000_000_000); + let mut anchor = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); + anchor.set_pts(10_000_000_000); + thread.pace_decoded(anchor, t0); + thread.run_once(t0 + 100_000_000); assert_eq!(recorder.emitted().len(), 1); + + let due_of = |i: u64| t0 + 100_000_000 + i * FRAME; + // Six frames 20 ms late, inside 200 ms ... + for i in 1..=6u64 { + let mut frame = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); + frame.set_pts(10_000_000_000 + (i * FRAME) as i64); + thread.pace_decoded(frame, due_of(i) + 20_000_000); + thread.run_once(due_of(i) + 20_000_000); + } + // ... then a second and a half of frames with their full margin. + for i in 7..=50u64 { + let mut frame = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); + frame.set_pts(10_000_000_000 + (i * FRAME) as i64); + thread.pace_decoded(frame, due_of(i) - 100_000_000); + thread.run_once(due_of(i) - 100_000_000); + } + thread.run_once(due_of(50)); + + assert_eq!(shared.conn.video_delay_ns.load(Relaxed), 0); + assert_eq!( + recorder.emitted().len(), + 51, + "late frames go out late, not dropped" + ); +} + +/// The delay was sized for one connection's sender. A clear — disconnect, +/// hide, restart — forgets it along with the play head, and the next +/// connection is measured afresh. +#[test] +fn a_clear_forgets_the_video_delay() { + let shared = shared_with_audio(); + let (mut thread, recorder) = thread_with(shared.clone()); + + let t0 = obs::time::gettime_ns(); + publish_mapping(&shared, t0 - 150_000_000, 10_000_000_000); + let mut frame = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); + frame.set_pts(10_000_000_000); + thread.pace_decoded(frame, t0); + thread.run_once(t0); + assert_eq!(shared.conn.video_delay_ns.load(Relaxed), 11 * TICK); + + shared.video.request_clear(); + let t1 = t0 + 5_000_000_000; + thread.run_once(t1); + assert_eq!(shared.conn.video_delay_ns.load(Relaxed), 0); + + // The next connection's audio primes with room to spare: no delay. + publish_mapping(&shared, t1 + 200_000_000, 20_000_000_000); + let mut frame = sw_frame(Pix::AV_PIX_FMT_YUV420P, 64, 32); + frame.set_pts(20_000_000_000); + thread.pace_decoded(frame, t1); + thread.run_once(t1 + 200_000_000); + assert_eq!( + recorder.emitted().last().map(|f| f.timestamp), + Some(t1 + 200_000_000) + ); + assert_eq!(shared.conn.video_delay_ns.load(Relaxed), 0); } diff --git a/docs/viewer-quality-plan.md b/docs/viewer-quality-plan.md index 33d0304..9206ba4 100644 --- a/docs/viewer-quality-plan.md +++ b/docs/viewer-quality-plan.md @@ -41,6 +41,11 @@ so a single log line describes the health of the whole path. ## Video behavior - First-keyframe gating is on by default. +- Video that reaches the plugin too late to be paced (the encoder sends video + later than audio by more than Target Buffer covers) is delayed by a + standing, measured amount rather than dropped or shown unpaced. + `video_delay_ms` reports it. It is the lip-sync error, and the log line that + sets it says how much more Target Buffer would remove it. - Timestamped damaged H.264 frames are passed through during decoder corruption so video cadence stays smooth: H.264 concealment patches a damaged frame from the previous one, which is a usable picture.