diff --git a/crates/pf-client-core/src/audio.rs b/crates/pf-client-core/src/audio.rs index b70ff4f9..59636a35 100644 --- a/crates/pf-client-core/src/audio.rs +++ b/crates/pf-client-core/src/audio.rs @@ -107,6 +107,9 @@ pub struct AudioPlayer { recycle_rx: Receiver>, quit_tx: pipewire::channel::Sender, thread: Option>, + /// A/V sync hand-off with the PipeWire callback: it publishes the ring depth, the decode + /// thread posts the depth the sync loop wants. See [`punktfunk_core::audio::AudioSyncCell`]. + sync: Arc, } impl AudioPlayer { @@ -121,10 +124,12 @@ impl AudioPlayer { // as the data channel; a full pool just drops the Vec (plain deallocation). let (recycle_tx, recycle_rx) = std::sync::mpsc::sync_channel::>(64); let (quit_tx, quit_rx) = pipewire::channel::channel::(); + let sync: Arc = Arc::default(); + let sync_cb = sync.clone(); let thread = std::thread::Builder::new() .name("punktfunk-audio".into()) .spawn(move || { - if let Err(e) = pw_thread(pcm_rx, recycle_tx, quit_rx, channels as usize) { + if let Err(e) = pw_thread(pcm_rx, recycle_tx, quit_rx, channels as usize, sync_cb) { tracing::warn!(error = %e, "audio playback thread ended"); } }) @@ -134,9 +139,16 @@ impl AudioPlayer { recycle_rx, quit_tx, thread: Some(thread), + sync, }) } + /// The A/V sync hand-off cell — the decode thread reads the ring depth from it and posts the + /// depth the sync loop wants back through it. + pub fn sync_cell(&self) -> Arc { + self.sync.clone() + } + /// A recycled chunk Vec from the pool, empty but with its capacity intact — fill it /// and hand it back through [`push`](Self::push). Allocates only when the pool is dry /// (startup, or after the PipeWire side dropped chunks). @@ -180,6 +192,8 @@ struct PlayerData { underruns: u64, sheds: u64, callbacks: u64, + /// A/V sync hand-off with the decode thread (depth out, target in). + sync: Arc, } fn pw_thread( @@ -187,6 +201,7 @@ fn pw_thread( recycle_tx: SyncSender>, quit_rx: pipewire::channel::Receiver, channels: usize, + sync: Arc, ) -> Result<()> { use pipewire as pw; use pw::{properties::properties, spa}; @@ -240,6 +255,7 @@ fn pw_thread( underruns: 0, sheds: 0, callbacks: 0, + sync, }; let _listener = stream @@ -267,6 +283,13 @@ fn pw_thread( let want_frames = data.data().map(|s| s.len() / stride).unwrap_or(0); let want = want_frames * ud.channels; + // A/V sync: take whatever depth the decode thread's sync loop last asked for, and + // publish where the ring actually is so it can measure the result. The policy + // clamps the request between its own underrun floor and the hard cap — continuity + // outranks sync, always (see `JitterPolicy::set_sync_target`). + ud.policy.set_sync_target(ud.sync.target()); + ud.sync.publish_depth(ud.ring.len()); + // Shared de-jitter policy: prime depth in MILLISECONDS, smooth drift correction // (a crossfaded 5 ms shed) so latency returns to target instead of ratcheting, // and a hard cap as the backstop. diff --git a/crates/pf-client-core/src/audio_wasapi.rs b/crates/pf-client-core/src/audio_wasapi.rs index 687d3a03..765281c1 100644 --- a/crates/pf-client-core/src/audio_wasapi.rs +++ b/crates/pf-client-core/src/audio_wasapi.rs @@ -163,6 +163,9 @@ pub struct AudioPlayer { recycle_rx: Receiver>, stop: Arc, thread: Option>, + /// A/V sync hand-off with the render thread: it publishes the ring depth, the decode thread + /// posts the depth the sync loop wants. See [`punktfunk_core::audio::AudioSyncCell`]. + sync: Arc, } impl AudioPlayer { @@ -179,10 +182,13 @@ impl AudioPlayer { let stop = Arc::new(AtomicBool::new(false)); let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel::>(1); let stop_t = stop.clone(); + let sync: Arc = Arc::default(); + let sync_t = sync.clone(); let thread = std::thread::Builder::new() .name("punktfunk-audio".into()) .spawn(move || { - if let Err(e) = render_thread(pcm_rx, recycle_tx, stop_t, ready_tx, channels as u8) + if let Err(e) = + render_thread(pcm_rx, recycle_tx, stop_t, ready_tx, channels as u8, sync_t) { tracing::warn!(error = %format!("{e:#}"), "audio playback thread ended"); } @@ -197,6 +203,7 @@ impl AudioPlayer { recycle_rx, stop, thread: Some(thread), + sync, }) } Ok(Err(e)) => Err(e), @@ -213,6 +220,12 @@ impl AudioPlayer { self.recycle_rx.try_recv().unwrap_or_default() } + /// The A/V sync hand-off cell — the decode thread reads the ring depth from it and posts the + /// depth the sync loop wants back through it. + pub fn sync_cell(&self) -> Arc { + self.sync.clone() + } + /// Queue one interleaved f32 chunk (in the session's channel layout). Drops the chunk if the /// WASAPI side is wedged (the renderer conceals the gap; never block the session pump). pub fn push(&self, pcm: Vec) { @@ -237,6 +250,7 @@ fn render_thread( stop: Arc, ready: SyncSender>, channels: u8, + sync: Arc, ) -> Result<()> { if let Err(e) = wasapi::initialize_mta() .ok() @@ -315,6 +329,12 @@ fn render_thread( } let want = avail_frames * channels as usize; + // A/V sync: same contract as the PipeWire ring — take the decode thread's request, + // publish where the ring actually is. The policy clamps the request against its own + // underrun floor, so continuity always outranks alignment. + policy.set_sync_target(sync.target()); + sync.publish_depth(ring.len()); + let step = policy.step(ring.len(), want); if step.drop_front > 0 { sheds += 1; diff --git a/crates/pf-client-core/src/session.rs b/crates/pf-client-core/src/session.rs index 77ca0012..74240748 100644 --- a/crates/pf-client-core/src/session.rs +++ b/crates/pf-client-core/src/session.rs @@ -178,6 +178,22 @@ pub struct Stats { /// is actually going out — the muted case has its own badge, which does not need stats on. pub mic_sent: u32, pub mic_dropped: u32, + /// How much decoded audio is queued ahead of the speaker right now (ms) — the playback + /// ring's depth. + /// + /// The audio plane used to publish nothing any surface could render: depth and target existed + /// only as a `tracing::debug!` line, and on a Steam Deck the client runs under Steam's + /// `reaper` with its stdout on a pipe, so the one number that identifies a deep ring was + /// unobtainable on the device reporting the latency. A field investigation ran to its + /// conclusion without it. That is the gap this closes. + pub audio_buffer_ms: u32, + /// The A/V sync loop's smoothed offset (ms): **positive = audio playing BEHIND the picture**, + /// negative = ahead of it. `0` before the loop has evidence, or with sync disabled. + /// + /// This is the figure that says whether audio is placed correctly, and it is the one the + /// overhaul is judged by — an absolute buffer depth cannot distinguish "deep because the link + /// needs it" from "deep and therefore late". + pub audio_av_offset_ms: i32, /// The decode path frames actually took this window (`"vaapi"`/`"software"`, empty /// until the first frame) — the OSD's trailing tag; tracks a mid-session fallback. pub decoder: &'static str, @@ -1407,6 +1423,8 @@ fn pump( }, mic_sent, mic_dropped, + audio_buffer_ms: connector.audio_buffer_ms(), + audio_av_offset_ms: connector.audio_av_offset_ms() as i32, decoder: dec_path, target_kbps: connector.current_bitrate_kbps(), auto_rate, @@ -1523,15 +1541,59 @@ fn spawn_audio( let mut dec = AudioDec::new(channels) .map_err(|e| tracing::warn!(error = %e, "opus decoder failed — audio disabled")) .ok()?; + // A/V sync (audio latency overhaul). This thread is the only place that holds all three + // ingredients at once: the packet's host capture `pts_ns`, the ring depth (via the sync cell) + // and the video plane's end-to-end figure. `pts_ns` was decoded into `AudioPacket` and then + // dropped on the floor here for the plane's entire existence, which is why audio ran at + // whatever depth its jitter ring happened to settle at and nothing ever placed it against the + // picture. + // + // The escape hatch is deliberate: a field regression in a loop that steers PLAYBACK should be + // bisectable without a rebuild, the same way `PUNKTFUNK_MIC_LEGACY_BUFFER` covers the uplink. + let av_sync_enabled = !matches!( + std::env::var("PUNKTFUNK_NO_AV_SYNC").as_deref(), + Ok("1") | Ok("true") + ); + let sync_cell = player.sync_cell(); + let video_e2e = connector.video_e2e_shared(); + let av_offset_out = connector.audio_av_offset_shared(); + let buffer_ms_out = connector.audio_buffer_ms_shared(); + // Interleaved samples per ms, to report the ring depth in the unit a human reads. + let per_ms = 48 * channels.max(1) as usize; std::thread::Builder::new() .name("punktfunk-audio-rx".into()) .spawn(move || { let mut pcm = vec![0f32; 5760 * channels as usize]; // scratch: max Opus frame (120 ms) × channels let mut gaps = punktfunk_core::audio::AudioGapTracker::new(); let mut frame_samples = 0usize; // per-channel samples of the last decoded frame — the PLC unit + let mut av = punktfunk_core::audio::AvSync::new(channels); + if !av_sync_enabled { + tracing::info!("A/V sync disabled by PUNKTFUNK_NO_AV_SYNC"); + } while !stop.load(Ordering::SeqCst) { match connector.next_audio(Duration::from_millis(100)) { Ok(pkt) => { + // Place this frame against the picture it belongs with, BEFORE it is + // queued: `buffered_ahead` is everything that must still play first, so + // the depth read here is exactly what delays it. + let depth = sync_cell.depth(); + // Published unconditionally — the ring's depth is worth seeing even with + // sync off, and it is what makes a "too much latency" report triageable. + buffer_ms_out.store((depth / per_ms) as u32, Ordering::Relaxed); + if av_sync_enabled { + let ve2e = video_e2e.load(Ordering::Relaxed); + let o = punktfunk_core::audio::AvSyncObservation { + pts_ns: pkt.pts_ns, + now_local_ns: punktfunk_core::client::now_realtime_ns(), + clock_offset_ns: connector.clock_offset_now_ns(), + buffered_ahead: depth, + // 0 = nothing on the glass yet; no reference, no correction. + video_e2e_ns: (ve2e > 0).then_some(ve2e), + }; + av.observe(o); + sync_cell.set_target(av.desired_depth(depth)); + av_offset_out.store(av.offset_ms() as i64, Ordering::Relaxed); + } // Conceal lost packets (a seq gap) with libopus PLC before decoding the one // that arrived: empty input synthesizes `frame_samples` of interpolation per // missing packet — an inaudible fade instead of the click a hard gap makes. diff --git a/crates/pf-presenter/src/run.rs b/crates/pf-presenter/src/run.rs index 93a76348..fafedc1c 100644 --- a/crates/pf-presenter/src/run.rs +++ b/crates/pf-presenter/src/run.rs @@ -222,6 +222,10 @@ struct StreamState { /// Live host↔client clock offset handle (None until Connected): loaded per present so /// mid-stream re-syncs keep the end-to-end number honest after an NTP step / drift. clock_offset: Option>, + /// Where the audio plane reads the video leg it must land with (ns). Published on every + /// presented frame; see the two `e2e` sites. The presenter deliberately knows nothing about + /// audio beyond writing this number. + video_e2e: Option>, hdr: bool, /// The presented lane shows a PQ stream RAW — no tone-map pass ran — so the OSD badge /// reads `HDR→SDR (raw)` instead of claiming one that never did. @@ -391,6 +395,7 @@ impl StreamState { profile, latch_grid, clock_offset: None, + video_e2e: None, hdr: false, hdr_untonemapped: false, win_e2e_us: Vec::with_capacity(256), @@ -1295,6 +1300,7 @@ fn run_inner(mut opts: SessionOpts, mut mode: ModeCtl) -> Result .ok(); gamepad.attach(c.clone()); st.clock_offset = Some(c.clock_offset_shared()); + st.video_e2e = Some(c.video_e2e_shared()); // gamescope's EIS grants only a relative pointer — absolute sends // would be dropped, so the desktop model is pinned off there. Auto // (an older host that didn't say) stays allowed: Windows hosts and @@ -1619,6 +1625,11 @@ fn run_inner(mut opts: SessionOpts, mut mode: ModeCtl) -> Result .max(0) as u64; if e2e > 0 && e2e < 10_000_000_000 { st.win_e2e_us.push(e2e / 1000); + // Hand the audio plane the figure it has to hit. This is the TRUE + // on-glass branch, so it is the best reference we can offer. + if let Some(c) = st.video_e2e.as_ref() { + c.store(e2e, Ordering::Relaxed); + } } st.win_disp_us .push(s.displayed_ns.saturating_sub(s.decoded_ns) / 1000); @@ -1995,6 +2006,13 @@ fn run_inner(mut opts: SessionOpts, mut mode: ModeCtl) -> Result .max(0) as u64; if e2e > 0 && e2e < 10_000_000_000 { st.win_e2e_us.push(e2e / 1000); + // Same hand-off as the glass-stamped branch above. This one is anchored + // on the submit instant rather than a true latch, so it UNDERSTATES the + // video leg by up to a refresh period — the audio loop's deadband is + // wider than that, which is what keeps the approximation harmless. + if let Some(c) = st.video_e2e.as_ref() { + c.store(e2e, Ordering::Relaxed); + } } st.win_disp_us .push(displayed_ns.saturating_sub(decoded_ns) / 1000); @@ -2805,6 +2823,20 @@ fn stats_text( text.push_str(&format!(" · dropped {}", s.mic_dropped)); } } + // The audio plane's own latency, Detailed-only. `buffer` is how much decoded audio is queued + // ahead of the speaker; `a/v` is where that PUTS it relative to the picture (+ = audio behind). + // + // Both, not just the depth: a deep ring on a jittery link is correct behaviour, and only the + // offset distinguishes that from a ring that is simply holding audio late. Before this the + // plane published neither — they lived in a `tracing::debug!` line that, on the Steam Deck, + // goes to a pipe under Steam's reaper that nobody can read, so the device that reported the + // latency was the one device where the numbers could not be seen. + if detailed && s.audio_buffer_ms > 0 { + text.push_str(&format!("\naudio buffer {} ms", s.audio_buffer_ms)); + if s.audio_av_offset_ms != 0 { + text.push_str(&format!(" · a/v {:+} ms", s.audio_av_offset_ms)); + } + } // Decode integrity (M4) — the native lane's answer to "was that stream actually // clean?". Appended LAST and only when it has something to say, which keeps it // additive for the stdout `stats:` line's parsers (a machine interface: every @@ -3116,6 +3148,8 @@ mod tests { lost_pct: 0.4, mic_sent: 0, mic_dropped: 0, + audio_buffer_ms: 0, + audio_av_offset_ms: 0, // The decode-path tag as the session actually spells it since M10 — the // ladder's rung names (`NativeRung::name`), not the deleted libavcodec // ones. A fixture carrying a tag no client emits would let this test go on diff --git a/crates/punktfunk-core/src/audio.rs b/crates/punktfunk-core/src/audio.rs index b38bdbb1..984e198f 100644 --- a/crates/punktfunk-core/src/audio.rs +++ b/crates/punktfunk-core/src/audio.rs @@ -494,6 +494,9 @@ const GROW_WINDOW_MS: u32 = 5_000; const GROW_STEP_MS: u32 = 10; /// Quiet time (no underrun) before a grown target relaxes one step back toward the base. const SHRINK_QUIET_MS: u32 = 30_000; +/// The same, while the A/V sync loop is actively asking for a shallower ring — see the branch in +/// [`JitterPolicy::note_read`] that selects between them. +const SHRINK_QUIET_SYNC_MS: u32 = 5_000; /// The playback de-jitter state machine shared by every client's audio ring. /// @@ -531,6 +534,11 @@ pub struct JitterPolicy { /// `want` from the most recent [`step`](Self::step), so [`note_read`](Self::note_read) can /// advance the sample-denominated timers without the caller repeating it. last_want: usize, + /// Depth the A/V sync loop would like, in interleaved samples ([`AvSync::desired_depth`]). + /// `None` — the default, and what every un-wired ring keeps — reproduces the pre-sync + /// behaviour exactly, which is what lets the four client rings adopt this one at a time + /// without diverging in the meantime. + sync_target: Option, } impl JitterPolicy { @@ -549,9 +557,27 @@ impl JitterPolicy { window_run: 0, quiet_run: 0, last_want: 0, + sync_target: None, } } + /// Hand the ring the depth the A/V sync loop wants ([`AvSync::desired_depth`]), or `None` to + /// run unsynchronised. + /// + /// This is a REQUEST, not a command. [`effective_target`](Self::effective_target) clamps it + /// between the underrun-driven adaptive floor and the hard cap, so sync can never starve the + /// ring: if the link's jitter needs more buffer than the picture is away, the floor wins and + /// the residual shows up on the HUD instead of as a dropout. That ordering is the whole safety + /// argument for steering playback depth from a network measurement at all. + pub fn set_sync_target(&mut self, target: Option) { + self.sync_target = target; + } + + /// The sync loop is asking to run shallower than the adaptive target has grown to. + fn sync_wants_less(&self) -> bool { + self.sync_target.is_some_and(|s| s < self.target) + } + /// The live target depth in ms (grows under underrun pressure; never below the base). pub fn target_ms(&self) -> u32 { (self.target / self.per_ms) as u32 @@ -577,7 +603,25 @@ impl JitterPolicy { /// quantum, a legacy AAudio path) lifts it to `want` plus one protocol frame rather than /// oscillating prime → dropout → re-prime forever. fn effective_target(&self, want: usize) -> usize { - self.target.max(want + FRAME_MS as usize * self.per_ms) + let floor = self.target.max(want + FRAME_MS as usize * self.per_ms); + match self.sync_target { + // Continuity outranks sync — see `set_sync_target`. The loop may pull the ring + // shallower to catch the picture up, or push it deeper when audio runs early, but + // never below what underrun pressure has proven this link needs, and never past the + // hard cap that bounds added latency. + // + // The ceiling is raised to the floor rather than passed to `clamp` as-is: a device + // whose callback quantum alone exceeds the preset's `hard_cap_ms` makes `floor > cap`, + // and `Ord::clamp` PANICS when min > max. That would be a panic in a realtime audio + // callback on exactly the awkward hardware this code exists to survive — and the same + // reasoning `step` already applies when it computes its own cap with `.max(target + + // want)`. + Some(s) => { + let cap = (self.tuning.hard_cap_ms as usize * self.per_ms).max(floor); + s.clamp(floor, cap) + } + None => floor, + } } /// Decide this callback: what to trim, and whether to play. Call BEFORE reading, with the @@ -664,7 +708,19 @@ impl JitterPolicy { } else { self.empties = 0; self.quiet_run += want; - if self.quiet_run >= SHRINK_QUIET_MS as usize * self.per_ms { + // A grown target normally relaxes only after a long quiet spell, because without other + // evidence the only thing that can justify giving up hard-won slack is time. When the + // sync loop is asking to run shallower it IS that evidence — a measurement saying the + // extra depth is costing alignment right now — so test a smaller target sooner. Wrong + // guesses are cheap and self-correcting: one underrun and the growth path takes it + // straight back. Without this a ring that ratcheted to the ceiling during a transient + // would hold the audio a ceiling's worth late for minutes after the cause had gone. + let quiet_needed = if self.sync_wants_less() { + SHRINK_QUIET_SYNC_MS + } else { + SHRINK_QUIET_MS + }; + if self.quiet_run >= quiet_needed as usize * self.per_ms { // Long quiet spell: give a grown target one step back, so a single bad minute // doesn't cost latency for the rest of the session. self.quiet_run = 0; @@ -750,6 +806,205 @@ pub fn spa_positions(channels: u8) -> &'static [u32] { } } +/// The lock-free hand-off between the thread that knows the TIMESTAMPS (the decode/pull thread, +/// which sees each packet's `pts_ns`) and the one that knows the RING (the realtime audio +/// callback, which owns the depth and the [`JitterPolicy`]). Neither can do the job alone and the +/// callback must not block, so they trade two words. +/// +/// `usize::MAX` encodes "no target" rather than `0`, because `0` is a perfectly ordinary depth to +/// ask for and conflating the two would silently mean "run the ring dry". +#[derive(Debug)] +pub struct AudioSyncCell { + depth: std::sync::atomic::AtomicUsize, + target: std::sync::atomic::AtomicUsize, +} + +impl Default for AudioSyncCell { + fn default() -> Self { + AudioSyncCell { + depth: std::sync::atomic::AtomicUsize::new(0), + target: std::sync::atomic::AtomicUsize::new(usize::MAX), + } + } +} + +impl AudioSyncCell { + /// Callback side: publish the ring's current depth in interleaved samples. + pub fn publish_depth(&self, depth: usize) { + self.depth + .store(depth, std::sync::atomic::Ordering::Relaxed); + } + + /// Decode side: the ring depth as last seen by the audio callback. + pub fn depth(&self) -> usize { + self.depth.load(std::sync::atomic::Ordering::Relaxed) + } + + /// Decode side: ask the ring to aim for this depth (`None` = run unsynchronised). + pub fn set_target(&self, target: Option) { + self.target.store( + target.unwrap_or(usize::MAX), + std::sync::atomic::Ordering::Relaxed, + ); + } + + /// Callback side: the depth the sync loop wants, if any. + pub fn target(&self) -> Option { + match self.target.load(std::sync::atomic::Ordering::Relaxed) { + usize::MAX => None, + t => Some(t), + } + } +} + +/// Smoothing time constant for the measured A/V offset, in ms of consumed audio. Long enough that +/// network jitter and a single late datagram do not move it; short enough to track real drift. +const AV_EWMA_TAU_MS: u32 = 2_000; +/// Offsets inside this band are left alone. Correcting a few ms costs a (crossfaded, but real) +/// discontinuity and buys nothing a listener can perceive — detectability for A/V misalignment sits +/// an order of magnitude above it. The deadband is what keeps the loop from hunting forever around +/// zero, which would be audible in a way the misalignment it was chasing was not. +const AV_DEADBAND_MS: u32 = 10; +/// Observations folded before the first correction is offered. The offset is derived from a clock +/// skew estimate and a video figure that both need a moment to settle after connect; acting on the +/// first sample would chase the handshake, not the stream. +const AV_MIN_OBSERVATIONS: u32 = 100; +/// An offset larger than this is not believed. A wall-clock step, a paused host, or a stale video +/// figure can all produce an enormous apparent misalignment, and steering the ring by it would +/// empty or overfill it outright. Beyond this the loop reports and waits rather than acting. +const AV_SANE_LIMIT_MS: u32 = 1_000; + +/// The A/V synchronisation controller: turns "when will this audio actually play" and "when did the +/// picture it belongs with reach the glass" into a ring depth the [`JitterPolicy`] should aim for. +/// +/// **The defect it exists to fix.** The host stamps `pts_ns` on every audio datagram and the client +/// decoded it into `AudioPacket` — and then never read it. Video's `pts_ns`, by contrast, is used +/// end to end (the presenter computes a true glass-to-glass `displayed + clock_offset − pts`). So +/// audio free-ran at whatever depth its jitter ring happened to settle at, video was presented on a +/// wholly independent path, and nothing ever compared them: the A/V offset was an accident of +/// buffer depths. It moved whenever the ring ratcheted under underrun pressure, and — the way this +/// surfaced in the field — it got WORSE every time video got faster, because a quicker decoder +/// lowers the video leg while leaving the audio leg exactly where it was. +/// +/// **Video is the master.** In a game streamer the video leg is the input-feel budget and must +/// never be inflated to satisfy the audio clock; audio tolerates small, crossfaded, rate-limited +/// corrections that are inaudible, and [`crossfade_drop`] already applies them. So audio moves. +/// +/// **Continuity outranks sync.** This type only ever proposes a depth. [`JitterPolicy`] clamps the +/// proposal to its own underrun-driven floor, so a link whose jitter genuinely needs more buffer +/// than the picture is away keeps its buffer and the residual is reported instead of being taken +/// out of the listener's stream. See [`JitterPolicy::set_sync_target`]. +#[derive(Clone, Debug)] +pub struct AvSync { + /// Interleaved samples per millisecond at the negotiated layout (48 × channels). + per_ms: usize, + /// EWMA of the measured offset in ns. Positive = audio is scheduled to play LATE relative to + /// the picture it belongs with. + offset_avg_ns: f32, + observations: u32, + /// Set once an observation lands outside [`AV_SANE_LIMIT_MS`], for reporting. + implausible: bool, +} + +/// One measurement handed to [`AvSync::observe`]. Every field is in the units its source already +/// produces, so no caller has to do clock arithmetic to use it correctly. +#[derive(Clone, Copy, Debug)] +pub struct AvSyncObservation { + /// The host capture timestamp carried by the audio frame being queued (host clock). + pub pts_ns: u64, + /// Local wall-clock now, same basis the client's video latency math uses (CLOCK_REALTIME). + pub now_local_ns: i128, + /// Host clock minus client clock, from the skew handshake (`clock_offset_now_ns`). + pub clock_offset_ns: i64, + /// How much audio is already queued AHEAD of this frame, in interleaved samples — everything + /// that must play before it does. + pub buffered_ahead: usize, + /// The video plane's current end-to-end figure in ns: `displayed + clock_offset − pts`, as the + /// presenter already computes it. `None` while no frame has been presented yet. + pub video_e2e_ns: Option, +} + +impl AvSync { + /// `channels` is the negotiated interleaved channel count (2/6/8). + pub fn new(channels: u8) -> AvSync { + AvSync { + per_ms: (SAMPLE_RATE_HZ / 1000) as usize * channels.max(1) as usize, + offset_avg_ns: 0.0, + observations: 0, + implausible: false, + } + } + + /// Fold one measurement. Returns the smoothed offset in ns once there is enough evidence to + /// believe it (positive = audio late), or `None` while still settling. + /// + /// Rejecting the implausible rather than clamping it is deliberate: a wall-clock step or a + /// stale video figure produces a huge apparent offset, and a clamped-but-wrong value would be + /// acted on as though it were a small real one. + pub fn observe(&mut self, o: AvSyncObservation) -> Option { + // No frame on the glass yet ⇒ no reference to align against, so nothing to say. + let video_e2e_ns = o.video_e2e_ns?; + // When this frame's samples will actually reach the speaker, expressed in the host's + // capture clock — the same clock, and the same shape, as the video figure it is compared + // against. + let buffered_ns = (o.buffered_ahead / self.per_ms.max(1)) as i128 * 1_000_000; + let play_at_host = o.now_local_ns + buffered_ns + o.clock_offset_ns as i128; + let audio_e2e_ns = play_at_host - o.pts_ns as i128; + let offset_ns = audio_e2e_ns - video_e2e_ns as i128; + + if offset_ns.unsigned_abs() > (AV_SANE_LIMIT_MS as u128) * 1_000_000 { + self.implausible = true; + return None; + } + self.implausible = false; + + // Weight by one protocol frame so the time constant means the same thing regardless of how + // often the caller observes. + let alpha = (FRAME_MS as f32 / AV_EWMA_TAU_MS as f32).clamp(0.0, 1.0); + if self.observations == 0 { + self.offset_avg_ns = offset_ns as f32; + } else { + self.offset_avg_ns += (offset_ns as f32 - self.offset_avg_ns) * alpha; + } + self.observations = self.observations.saturating_add(1); + self.settled().then_some(self.offset_avg_ns as i64) + } + + /// Enough evidence folded to act on. + pub fn settled(&self) -> bool { + self.observations >= AV_MIN_OBSERVATIONS + } + + /// The smoothed offset in ms (positive = audio late), for the HUD. Reported as soon as it is + /// measured, including while still settling — a number the operator can watch converge is more + /// useful than a blank that hides whether the loop is working at all. + pub fn offset_ms(&self) -> i32 { + (self.offset_avg_ns / 1_000_000.0) as i32 + } + + /// The last observation was outside the believable range and was discarded. + pub fn implausible(&self) -> bool { + self.implausible + } + + /// The ring depth that would place audio with the picture, given where the ring is now. + /// `None` while unsettled or inside the deadband — the caller then leaves the policy alone. + /// + /// Audio late (offset > 0) means there is too much queued: aim shallower. Audio early means + /// aim deeper. + pub fn desired_depth(&self, current_depth: usize) -> Option { + if !self.settled() { + return None; + } + let offset_ms = self.offset_avg_ns / 1_000_000.0; + if offset_ms.abs() < AV_DEADBAND_MS as f32 { + return None; + } + let delta = (offset_ms * self.per_ms as f32) as i64; + Some((current_depth as i64 - delta).max(0) as usize) + } +} + #[cfg(test)] mod tests { use super::*; @@ -1447,4 +1702,239 @@ mod tests { } } } + + // ---- A/V sync (audio latency overhaul) ---------------------------------------------- + + /// Build an observation whose measured offset is exactly `offset_ms` (positive = audio late). + fn obs(offset_ms: i64, depth: usize, per_ms: usize) -> AvSyncObservation { + // audio_e2e = buffered + (now + skew - pts); pin now/skew/pts so the only free term is the + // buffered depth, then choose video_e2e so the difference lands on `offset_ms`. + let buffered_ms = (depth / per_ms) as i64; + let audio_e2e_ms = buffered_ms + 40; // 40 ms of transport, arbitrary but fixed + let video_e2e_ms = audio_e2e_ms - offset_ms; + AvSyncObservation { + pts_ns: 1_000_000_000, + now_local_ns: 1_000_000_000i128 + 40 * 1_000_000, + clock_offset_ns: 0, + buffered_ahead: depth, + video_e2e_ns: Some((video_e2e_ms.max(0) as u64) * 1_000_000), + } + } + + fn settle(sync: &mut AvSync, offset_ms: i64, depth: usize, per_ms: usize, n: u32) { + for _ in 0..n { + sync.observe(obs(offset_ms, depth, per_ms)); + } + } + + #[test] + fn av_sync_needs_evidence_before_acting() { + let pm = per_ms(2); + let mut s = AvSync::new(2); + // One sample is never enough — the skew estimate and the video figure both settle after + // connect, and acting on the first would chase the handshake. + assert!(s.observe(obs(50, 30 * pm, pm)).is_none()); + assert!(!s.settled()); + assert!(s.desired_depth(30 * pm).is_none()); + settle(&mut s, 50, 30 * pm, pm, AV_MIN_OBSERVATIONS); + assert!(s.settled(), "should act once the evidence is in"); + } + + #[test] + fn av_sync_aims_shallower_when_audio_is_late() { + let pm = per_ms(2); + let depth = 60 * pm; + let mut s = AvSync::new(2); + settle(&mut s, 40, depth, pm, AV_MIN_OBSERVATIONS * 4); + let want = s + .desired_depth(depth) + .expect("a 40 ms offset is actionable"); + assert!( + want < depth, + "audio late must aim shallower: {want} vs {depth}" + ); + // The correction is the offset, not a guess at it. + let shed_ms = (depth - want) / pm; + assert!( + (35..=45).contains(&shed_ms), + "should aim to shed ~40 ms, got {shed_ms}" + ); + } + + #[test] + fn av_sync_aims_deeper_when_audio_is_early() { + let pm = per_ms(2); + let depth = 20 * pm; + let mut s = AvSync::new(2); + settle(&mut s, -30, depth, pm, AV_MIN_OBSERVATIONS * 4); + let want = s + .desired_depth(depth) + .expect("a 30 ms offset is actionable"); + assert!( + want > depth, + "audio early must aim deeper: {want} vs {depth}" + ); + } + + #[test] + fn av_sync_deadbands_what_no_one_can_hear() { + let pm = per_ms(2); + let depth = 30 * pm; + let mut s = AvSync::new(2); + settle( + &mut s, + (AV_DEADBAND_MS - 2) as i64, + depth, + pm, + AV_MIN_OBSERVATIONS * 4, + ); + assert!( + s.desired_depth(depth).is_none(), + "an offset inside the deadband must not provoke a (real, if crossfaded) discontinuity" + ); + } + + #[test] + fn av_sync_rejects_the_implausible_instead_of_clamping_it() { + let pm = per_ms(2); + let depth = 30 * pm; + let mut s = AvSync::new(2); + settle(&mut s, 30, depth, pm, AV_MIN_OBSERVATIONS * 4); + let before = s.offset_ms(); + // A wall-clock step / stale video figure. Built directly rather than through `obs`: that + // helper floors the video figure at zero, which would cap the offset at a merely LARGE + // value and let this test pass without ever exercising the rejection. + let wild = AvSyncObservation { + pts_ns: 0, + now_local_ns: 5_000_000_000, + clock_offset_ns: 0, + buffered_ahead: depth, + video_e2e_ns: Some(40_000_000), + }; + assert!(s.observe(wild).is_none()); + assert!(s.implausible(), "a ~5 s offset must be refused, not folded"); + assert_eq!( + before, + s.offset_ms(), + "an implausible sample must be discarded, not folded in" + ); + } + + #[test] + fn sync_can_never_starve_the_ring() { + // THE safety invariant: sync only ever proposes. Continuity — the underrun-driven floor — + // outranks it on every preset, or a lossy link would be "synced" into dropouts. + for (name, t) in [ + ("PIPEWIRE", JitterTuning::PIPEWIRE), + ("WASAPI", JitterTuning::WASAPI), + ("COREAUDIO", JitterTuning::COREAUDIO), + ("AAUDIO", JitterTuning::AAUDIO), + ] { + let pm = per_ms(2); + let want = 5 * pm; + let mut p = JitterPolicy::new(t, 2); + let floor = p.effective_target(want); + // Ask for an absurdly shallow ring — zero. + p.set_sync_target(Some(0)); + assert_eq!( + p.effective_target(want), + floor, + "{name}: sync pulled the target below the continuity floor" + ); + // And it may not blow past the hard cap either. + p.set_sync_target(Some(usize::MAX / 2)); + assert!( + p.effective_target(want) <= t.hard_cap_ms as usize * pm, + "{name}: sync pushed the target past the hard cap" + ); + } + } + + #[test] + fn a_huge_device_quantum_does_not_panic_the_clamp() { + // `Ord::clamp` panics when min > max. A device whose callback quantum alone exceeds the + // preset's hard cap pushes the continuity floor above the ceiling, and this runs inside a + // realtime audio callback — so the ceiling yields to the floor instead. + let t = JitterTuning::PIPEWIRE; // hard_cap 80 ms + let pm = per_ms(2); + let want = 500 * pm; // a 500 ms quantum: absurd, but not a reason to abort the process + let mut p = JitterPolicy::new(t, 2); + p.set_sync_target(Some(0)); + let target = p.effective_target(want); // must not panic + assert!( + target >= want, + "the target must still be able to serve one callback" + ); + } + + #[test] + fn no_sync_target_leaves_the_policy_exactly_as_it_was() { + // The four rings adopt sync one at a time; an un-wired ring must behave bit-identically to + // before. `None` is the default, so this also pins the constructor. + let t = JitterTuning::PIPEWIRE; + let pm = per_ms(2); + let want = 5 * pm; + let mut a = JitterPolicy::new(t, 2); + let mut b = JitterPolicy::new(t, 2); + b.set_sync_target(None); + assert_eq!(a.effective_target(want), b.effective_target(want)); + for depth_ms in [0usize, 5, 15, 30, 60, 90, 200] { + let sa = a.step(depth_ms * pm, want); + let sb = b.step(depth_ms * pm, want); + assert_eq!(sa, sb, "depth {depth_ms} ms diverged with an explicit None"); + a.note_read(sa.silence); + b.note_read(sb.silence); + } + } + + #[test] + fn sync_pressure_relaxes_a_grown_target_sooner_than_time_alone() { + // A ring that ratcheted during a transient must not hold audio late for minutes after the + // cause is gone. With sync asking for less, the relax window is the short one. + let t = JitterTuning::PIPEWIRE; + let pm = per_ms(2); + let want = 5 * pm; + + let grow = |p: &mut JitterPolicy| { + // Drive underruns until the target has grown above the base. Each round hands `step` a + // DEEP ring first: `note_read` ignores everything while un-primed (a priming silence is + // not an underrun), and `deprime_after` short reads in a row un-prime the ring — so + // hammering a zero-depth ring would report nothing and grow nothing, forever. + for _ in 0..10_000 { + if p.target_ms() > t.base_target_ms { + return; + } + p.step(200 * pm, want); // (re-)prime + p.note_read(true); // then one genuine short read + } + panic!("the adaptive floor never grew — the test cannot measure a relax"); + }; + // Quiet reads needed to relax one step, with and without sync pressure. + let quiet_to_relax = |p: &mut JitterPolicy| -> usize { + let start = p.target_ms(); + let mut reads = 0usize; + while p.target_ms() == start && reads < 200_000 { + p.step(60 * pm, want); + p.note_read(false); + reads += 1; + } + reads + }; + + let mut slow = JitterPolicy::new(t, 2); + grow(&mut slow); + slow.set_sync_target(None); + let slow_reads = quiet_to_relax(&mut slow); + + let mut fast = JitterPolicy::new(t, 2); + grow(&mut fast); + // Ask for something strictly shallower than the grown target. + fast.set_sync_target(Some(pm)); + let fast_reads = quiet_to_relax(&mut fast); + + assert!( + fast_reads < slow_reads, + "sync pressure should relax sooner: {fast_reads} vs {slow_reads} quiet reads" + ); + } } diff --git a/crates/punktfunk-core/src/client/mod.rs b/crates/punktfunk-core/src/client/mod.rs index 865f6b3f..2c7d7127 100644 --- a/crates/punktfunk-core/src/client/mod.rs +++ b/crates/punktfunk-core/src/client/mod.rs @@ -297,6 +297,26 @@ pub struct NativeClient { /// the pump's first no-op clock flush). Shared with the pump and, via /// [`clock_offset_shared`](Self::clock_offset_shared), with embedder latency-math threads. clock_offset: Arc, + /// The video plane's live end-to-end latency in ns — `displayed + clock_offset − pts`, the + /// figure the presenter already computes per frame (with a TRUE on-glass stamp where + /// `VK_KHR_present_wait` is available, and the submit instant otherwise). `0` = nothing + /// presented yet. + /// + /// Written by whoever puts frames on the glass; read by the audio plane, which steers its ring + /// depth to land audio WITH the picture ([`crate::audio::AvSync`]). It lives here, next to + /// `clock_offset`, because those two are exactly the pair a synchroniser needs and neither + /// plane owns the other: the presenter must not know about audio, and the audio thread cannot + /// see the glass. + video_e2e_ns: Arc, + /// The A/V sync loop's smoothed offset in ms — positive = audio playing LATE relative to the + /// picture. Written by the audio thread, read by the stats HUD. The audio plane used to + /// publish NOTHING a surface could render (its depth and target existed only as a + /// `tracing::debug!` line, which on a Deck goes into a pipe under Steam's reaper that nobody + /// can read), so a latency report had no instrument behind it at all. + audio_av_offset_ms: Arc, + /// Decoded audio queued ahead of the speaker (ms) — the playback ring's depth, as last seen by + /// the audio callback. Written by the audio thread, read by the stats HUD. + audio_buffer_ms: Arc, /// Decode-stage latency samples from the embedder ([`report_decode_us`](Self::report_decode_us)), /// drained per window by the data-plane pump to feed the adaptive-bitrate controller's decode /// signal. Shared with the pump; see [`DecodeLatAcc`]. @@ -400,7 +420,13 @@ fn pin_thread_user_interactive() {} /// Wall-clock now in nanoseconds (CLOCK_REALTIME basis), to compare against the host-stamped /// capture `pts_ns` after the skew offset is applied — the same latency math the stats HUDs use. -fn now_realtime_ns() -> i128 { +/// +/// Public because the A/V sync loop ([`crate::audio::AvSync`]) lives in an embedder crate but must +/// read the clock in EXACTLY this basis: its whole output is a difference between a local instant +/// and a host `pts_ns`, so a caller reaching for `Instant` or a monotonic clock instead would get a +/// plausible-looking number that is wrong by the machine's boot time. Exporting the one correct +/// clock is cheaper than documenting which clocks are incorrect. +pub fn now_realtime_ns() -> i128 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_nanos() as i128) @@ -545,6 +571,9 @@ impl NativeClient { let mic_stats = Arc::new(MicUplinkCounters::default()); let hot_tids = Arc::new(Mutex::new(Vec::new())); let clock_offset = Arc::new(AtomicI64::new(0)); + let video_e2e_ns = Arc::new(AtomicU64::new(0)); + let audio_av_offset_ms = Arc::new(AtomicI64::new(0)); + let audio_buffer_ms = Arc::new(AtomicU32::new(0)); let decode_lat = Arc::new(Mutex::new(DecodeLatAcc::default())); // Seeded by the pump from the Welcome (before ready_tx), then follows every ack. let live_bitrate = Arc::new(AtomicU32::new(0)); @@ -690,6 +719,9 @@ impl NativeClient { rfi: Mutex::new(RfiRecovery::default()), hot_tids, clock_offset, + video_e2e_ns, + audio_av_offset_ms, + audio_buffer_ms, decode_lat, live_bitrate_kbps: live_bitrate, // The controller arms exactly when the pump does — all three terms, not two: Automatic @@ -960,6 +992,35 @@ impl NativeClient { self.clock_offset.clone() } + /// The shared cell carrying the video plane's end-to-end latency (ns, `0` = nothing presented + /// yet). The presenter WRITES it once per presented frame; the audio plane READS it to place + /// its samples with the picture. See the field docs on `video_e2e_ns`. + pub fn video_e2e_shared(&self) -> Arc { + self.video_e2e_ns.clone() + } + + /// The cell carrying the A/V sync loop's smoothed offset in ms (positive = audio late). + /// Written by the audio thread; read by the HUD. + pub fn audio_av_offset_shared(&self) -> Arc { + self.audio_av_offset_ms.clone() + } + + /// The A/V sync offset the audio plane last measured, in ms. Positive = audio is playing + /// behind the picture. `0` before the loop has evidence, or when sync is off. + pub fn audio_av_offset_ms(&self) -> i64 { + self.audio_av_offset_ms.load(Ordering::Relaxed) + } + + /// The cell carrying the playback ring's depth in ms. Written by the audio thread. + pub fn audio_buffer_ms_shared(&self) -> Arc { + self.audio_buffer_ms.clone() + } + + /// Decoded audio queued ahead of the speaker, in ms. + pub fn audio_buffer_ms(&self) -> u32 { + self.audio_buffer_ms.load(Ordering::Relaxed) + } + /// Report one decoded frame's decode-stage latency, in microseconds: the wall-clock elapsed from /// the access unit leaving [`next_frame`](Self::next_frame) to its decoded output becoming /// available (dequeued from the decoder). This feeds the "Automatic" bitrate controller's decode diff --git a/crates/punktfunk-host/src/audio/linux/mod.rs b/crates/punktfunk-host/src/audio/linux/mod.rs index 1275cd0f..62bd118d 100644 --- a/crates/punktfunk-host/src/audio/linux/mod.rs +++ b/crates/punktfunk-host/src/audio/linux/mod.rs @@ -352,6 +352,12 @@ struct MicUserData { /// bursting out as stale audio when recording (re)starts. const MIC_STALE: Duration = Duration::from_secs(1); +/// The graph quantum every punktfunk PipeWire stream asks for, in frames: 240 @ 48 kHz = 5 ms, +/// one protocol audio frame. Named so the `NODE_LATENCY` request and the code that CHECKS whether +/// the request was honoured cannot drift apart — the check is only meaningful while it compares +/// against the same number the ask used. +const CAPTURE_QUANTUM_FRAMES: u32 = 240; + fn mic_pw_thread( pcm_rx: Receiver<(std::time::Instant, Vec)>, quit_rx: pipewire::channel::Receiver, @@ -722,12 +728,19 @@ fn pw_thread( channels: u32, stats: crate::audio::capture_policy::CaptureStats, last_stats: std::time::Instant, + /// Whether this OPEN has reported its negotiated buffer size yet. Per-open, not the + /// process-wide `static AtomicBool` this replaces: a host runs for days across many + /// sessions, so the old form reported the very first capture of the process and then + /// never again — the one number that identifies a clamped quantum, invisible on every + /// subsequent open (including every reopen after a device change). + reported_quantum: bool, } let ud = CapUd { tx, channels, stats: Default::default(), last_stats: std::time::Instant::now(), + reported_quantum: false, }; let _listener = stream .add_local_listener_with_user_data(ud) @@ -788,10 +801,36 @@ fn pw_thread( let region = &buf[offset..(offset + size).min(buf.len())]; // Negotiated as F32LE; reinterpret the byte region as interleaved f32. let n = region.len() / 4; - static FIRST: std::sync::atomic::AtomicBool = - std::sync::atomic::AtomicBool::new(true); - if FIRST.swap(false, std::sync::atomic::Ordering::Relaxed) { - tracing::info!(samples = n, "audio first capture buffer"); + if !ud.reported_quantum { + ud.reported_quantum = true; + // What we ASKED for vs what PipeWire actually handed us. Stating only the + // result ("samples=2048") reads as a fact about the device; stating it + // next to the request is what makes a clamp legible. A VM is the common + // cause — stock `pipewire.conf` raises `default.clock.min-quantum` to + // 1024 whenever `cpu.vm.name` is set, so a 5 ms ask silently becomes + // 21.3 ms and the audio plane starts arriving in bursts. That cost a + // whole field investigation to find; it should cost one log line. + let frames = n / (ud.channels.max(1) as usize); + let want = CAPTURE_QUANTUM_FRAMES as usize; + if frames > want { + tracing::warn!( + requested_frames = want, + negotiated_frames = frames, + negotiated_ms = + format!("{:.1}", frames as f32 * 1000.0 / SAMPLE_RATE as f32), + "the audio graph refused our low-latency quantum — capture arrives \ + in bursts this size, and the client must buffer at least that \ + much to play them smoothly. On a VM this is PipeWire's \ + `default.clock.min-quantum = 1024` rule; check \ + `pw-metadata -n settings`" + ); + } else { + tracing::info!( + requested_frames = want, + negotiated_frames = frames, + "audio capture quantum negotiated" + ); + } } let mut samples = Vec::with_capacity(n); for i in 0..n { diff --git a/crates/punktfunk-host/src/native/audio.rs b/crates/punktfunk-host/src/native/audio.rs index 50cd5033..27ecb33c 100644 --- a/crates/punktfunk-host/src/native/audio.rs +++ b/crates/punktfunk-host/src/native/audio.rs @@ -89,6 +89,16 @@ pub(super) fn audio_thread( use crate::audio::SAMPLE_RATE; const FRAME_MS: usize = 5; const SAMPLES_PER_FRAME: usize = SAMPLE_RATE as usize * FRAME_MS / 1000; // 240 + /// One protocol frame of wall time — the cadence paced sends aim for. + const FRAME_INTERVAL: std::time::Duration = std::time::Duration::from_millis(FRAME_MS as u64); + /// Ceiling on a single pacing sleep. The capture channel is finite and `next_chunk` has to be + /// serviced; sleeping past a couple of frames would trade a burst on the wire for a drop at + /// the capturer, which is strictly worse (a drop is a click AND a permanent shift). + const PACE_MAX_SLEEP: std::time::Duration = std::time::Duration::from_millis(10); + /// How far behind schedule the pacer may fall before it stops trying to catch up and simply + /// re-anchors. Chasing an old schedule after a stall would send a burst — the exact thing + /// pacing exists to prevent — so past this point the debt is forgiven, not repaid. + const PACE_REANCHOR: std::time::Duration = std::time::Duration::from_millis(100); let want = punktfunk_core::audio::normalize_channels(channels); // Tier and redundancy are ONE decision, budgeted against the session's video bitrate — see // `handshake::audio_budget`. An unparseable `audio.quality` was already warned about there @@ -151,6 +161,27 @@ pub(super) fn audio_thread( // continuity breaks (a capture reopen), so we never advertise a predecessor the client's // sequence numbering does not agree with. let mut prev_frame: Vec = Vec::new(); + // W1.1/W1.2 — the audio SAMPLE clock, and the schedule frames leave on. + // + // `pts_ns` used to be `now_ns()` evaluated inside the drain loop below, which made it the + // instant we got round to ENCODING rather than the instant the samples were CAPTURED. Every + // frame carved out of one capture chunk therefore carried a near-identical timestamp, and the + // value drifted with encoder scheduling. That was harmless only for as long as nothing + // consumed it; a client-side A/V sync loop regulating against it would be regulating against + // a fiction, so this is a prerequisite for the whole overhaul, not a tidy-up. + // + // `pace_due` exists because a chunk is not a frame. A capture callback hands us a whole + // quantum (5 ms when the graph honours our ask, 21.3 ms on a VM that clamps it to 1024 — + // see `audio::linux`'s quantum warning), and the old loop drained all of it into + // back-to-back `send_datagram` calls. The wire then carried a 4-5 frame burst followed by + // ~21 ms of nothing, and a client ring can only absorb that by standing at least a burst + // period deep. Releasing frames on the audio clock instead costs no AVERAGE latency — the + // client was buffering those frames anyway — and removes the burst the ring was sized for. + // Uninitialised on purpose: every read is preceded by the re-anchor at the top of the chunk + // loop, and seeding it with a placeholder would just be a value the compiler correctly points + // out is never read. + let mut next_pts_ns: u64; + let mut pace_due: Option = None; if capturer.is_some() { tracing::info!( channels = want, @@ -194,10 +225,37 @@ pub(super) fn audio_thread( continue; } }; + // Anchor the sample clock on THIS chunk's arrival. PipeWire hands us a buffer of already + // captured audio, so the newest sample in `acc` is ~now and the oldest is one whole + // buffer-occupancy earlier. Re-deriving the anchor every chunk (rather than free-running + // a counter) keeps the stamp tied to the capture device's own cadence, so a drifting or + // resampling graph corrects itself instead of accumulating error over a long session. + let arrival_ns = now_ns(); acc.extend_from_slice(&chunk); + let queued_frames = (acc.len() / want as usize) as u64; + next_pts_ns = arrival_ns.saturating_sub(queued_frames * 1_000_000_000 / SAMPLE_RATE as u64); while acc.len() >= frame_len { + // Hold each frame until its slot on the audio clock. The FIRST frame of a chunk is + // already due (its samples are the oldest we hold), so this only ever delays the + // tail of a multi-frame chunk — exactly the burst we are trying not to send. A + // schedule that has fallen more than one frame behind is re-anchored rather than + // chased, so a scheduling hiccup cannot turn into a permanent send-time debt. + let now = std::time::Instant::now(); + match pace_due { + Some(due) if due > now => { + let wait = due - now; + // Never sleep longer than the audio we are holding: `next_chunk` has to be + // serviced or the capture channel backs up and starts dropping. + std::thread::sleep(wait.min(PACE_MAX_SLEEP)); + } + Some(due) if now.duration_since(due) > PACE_REANCHOR => pace_due = None, + _ => {} + } + pace_due = Some(pace_due.unwrap_or_else(std::time::Instant::now) + FRAME_INTERVAL); + let frame: Vec = acc.drain(..frame_len).collect(); - let pts_ns = now_ns(); + let pts_ns = next_pts_ns; + next_pts_ns += FRAME_MS as u64 * 1_000_000; match enc.encode_float(&frame, &mut opus_buf) { Ok(n) => { let opus = &opus_buf[..n];