diff --git a/clients/android/native/src/audio.rs b/clients/android/native/src/audio.rs index 2a76dcba..6dbec5cc 100644 --- a/clients/android/native/src/audio.rs +++ b/clients/android/native/src/audio.rs @@ -12,10 +12,14 @@ //! realtime callback and makes us own the buffer. So this client diverges deliberately to stop the //! Android-only crackle: (1) the callback is allocation/free-free — decoded buffers are recycled to //! the producer via a free-list instead of being freed on the audio thread (Android's Scudo `free` -//! has unbounded tail latency); (2) the jitter ring is deeper (~40 ms prime / ~150 ms hard cap) and -//! decoupled from the tiny LowLatency burst size, with de-prime hysteresis so a transient drain -//! doesn't manufacture a silence; (3) the AAudio HW buffer is primed above its 2-burst default and -//! grown on XRuns (Google's anti-glitch technique). +//! has unbounded tail latency); (2) the jitter ring is deeper than the other clients' and decoupled +//! from the tiny LowLatency burst size, with de-prime hysteresis so a transient drain doesn't +//! manufacture a silence; (3) the AAudio HW buffer is primed above its 2-burst default and grown on +//! XRuns (Google's anti-glitch technique). +//! +//! (2) is now the SHARED `punktfunk_core::audio::JitterPolicy` at `JitterTuning::AAUDIO`, which also +//! fixed what this ring was missing: it had a hard cap but nothing that walked the depth back down, +//! so drift and arrival bursts raised latency permanently and Android settled on its ceiling. use ndk::audio::{ AudioCallbackResult, AudioContentType, AudioDirection, AudioFormat, AudioPerformanceMode, @@ -34,26 +38,18 @@ const SAMPLE_RATE: i32 = 48_000; /// Decoded-chunk hand-off depth: 64 × 5 ms = 320 ms slack (matches the core's AUDIO_QUEUE). const RING_CHUNKS: usize = 64; -// --- Jitter-ring depths, in MILLISECONDS (scaled to interleaved-f32 samples at runtime). -------- -// The channel count is negotiated, not a compile-time const, so these are kept in ms and multiplied -// by `ms` (interleaved-f32 samples per millisecond at the resolved layout) inside `start`. -// Unlike the Linux client (PipeWire adaptively rate-matches the stream to the graph clock, masking -// host↔DAC drift + a shallow ring), AAudio hands us a raw callback and we own the buffer: drift and -// WiFi power-save bunching land as underruns/overflows = crackle. So Android runs a deliberately -// deeper, smoothly-managed ring than Linux — keep the two clients' depths intentionally divergent. -/// Prime/target floor: fill to ~40 ms before playing (and after a sustained drain). Deep enough to -/// ride out WiFi arrival jitter + clock drift; the dominant Android-only anti-crackle lever. -const PRIME_FLOOR_MS: usize = 40; -/// Ceiling for the burst-scaled target (so a large quantum can't push the prime depth too high). -const PRIME_CEIL_MS: usize = 80; -/// Drop-oldest headroom above the target before trimming — a ~80 ms band swallows an arrival burst -/// without overflowing. -const JITTER_HEADROOM_MS: usize = 80; -/// Hard latency bound: never let the ring exceed ~150 ms (the only thing that caps added latency). -const HARD_CAP_MS: usize = 150; -/// Re-prime (go silent to refill) only after this many CONSECUTIVE empty callbacks, so one transient -/// drain doesn't manufacture a fresh 40 ms silence (the old `if ring.is_empty()` re-primed instantly). -const DEPRIME_AFTER_CALLBACKS: u32 = 5; +// --- Jitter-ring depths now come from the SHARED policy (`punktfunk_core::audio::JitterTuning`). -- +// They used to be four Android-only constants here. The rationale for Android being DEEPER than the +// other clients still holds and is preserved in `JitterTuning::AAUDIO`: unlike PipeWire, which +// adaptively rate-matches the stream to the graph clock and masks host↔DAC drift, AAudio hands us a +// raw callback and we own the buffer, so drift and Wi-Fi power-save bunching land as +// underruns/overflows = crackle. +// +// Two things changed with the move. The prime floor drops 40 ms → 25 ms, because the policy GROWS +// the target on the devices that actually underrun instead of every device pre-paying for the worst +// one. And the ring finally sheds: it had a hard cap but nothing that walked the depth back down, so +// any drift or burst raised latency permanently and Android converged on its 120 ms ceiling and +// stayed there — the "audio latency is too high" report. /// Throttle the AAudio XRun-driven HW-buffer grow check (cheap, but no need to poll every quantum). const XRUN_CHECK_EVERY: u32 = 128; @@ -104,6 +100,7 @@ struct Counters { pcm_written: AtomicU64, // PCM frames copied out to AAudio (device clock is pulling) underruns: AtomicU64, // callbacks that emitted silence (ring not primed / drained) ring_depth: AtomicU64, // ring sample count at the last callback + target_ms: AtomicU64, // the policy's LIVE target depth (it grows on this device's underruns) } /// Owned by [`crate::session::SessionHandle`]: the live AAudio stream + the decode thread. @@ -126,10 +123,9 @@ impl AudioPlayback { // Interleaved f32 samples per millisecond at this layout (48 kHz × channels); the ms- // denominated jitter-ring depths scale by it. let ms = (SAMPLE_RATE as usize / 1000) * channels; - let prime_floor = PRIME_FLOOR_MS * ms; - let prime_ceil = PRIME_CEIL_MS * ms; - let jitter_headroom = JITTER_HEADROOM_MS * ms; - let hard_cap_max = HARD_CAP_MS * ms; + let tuning = punktfunk_core::audio::JitterTuning::AAUDIO; + // Worst transient the ring can hold before the policy trims it. + let hard_cap_max = tuning.hard_cap_ms as usize * ms; let counters = Arc::new(Counters::default()); // One open attempt at a given sharing mode. Everything the realtime callback captures @@ -157,8 +153,10 @@ impl AudioPlayback { // `decode_loop`. let mut ring: VecDeque = VecDeque::with_capacity(hard_cap_max + RING_CHUNKS * 5 * ms); - let mut primed = false; - let mut empties: u32 = 0; // consecutive empty callbacks (de-prime hysteresis) + // Shared de-jitter policy — prime depth, drift correction, de-prime hysteresis. The + // hysteresis this replaces was Android-only; Linux and Windows carried the instant + // `if ring.is_empty()` re-prime until now. + let mut policy = punktfunk_core::audio::JitterPolicy::new(tuning, channels as u8); let mut cb_count: u32 = 0; // callbacks since open (throttles the XRun grow check) let mut last_xrun: i32 = 0; // last AAudio XRun count we grew the buffer for let callback = move |s: &AudioStream, data: *mut c_void, num_frames: i32| { @@ -173,21 +171,25 @@ impl AudioPlayback { ring.extend(chunk.drain(..)); let _ = free_tx.try_send(chunk); } - // Jitter buffer: prime to ~40 ms (prime_floor) before playing and after a sustained - // drain; drop-oldest only above a wide ~120 ms band. Decoupled from the AAudio burst - // `want` (tiny on the LowLatency MMAP path) so the depth doesn't collapse to a single - // quantum. - let target = (3 * want).clamp(prime_floor, prime_ceil); - let hard_cap = (target + jitter_headroom).min(hard_cap_max); - while ring.len() > hard_cap { - ring.pop_front(); + // Jitter buffer: the shared policy decides prime/silence, trims a burst, and — + // new here — sheds ONE crossfaded 5 ms frame when the depth average has sat above + // target long enough to be drift rather than jitter. Without that shed this ring + // had no way back down: it clamped at 120 ms and stayed pinned there. + let step = policy.step(ring.len(), want); + if step.drop_front > 0 { + punktfunk_core::audio::crossfade_drop( + &mut ring, + step.drop_front, + step.crossfade, + ); } - if !primed && ring.len() >= target { - primed = true; - } - if primed { + let mut ran_short = false; + if !step.silence { for slot in out.iter_mut() { - *slot = ring.pop_front().unwrap_or(0.0); + *slot = ring.pop_front().unwrap_or_else(|| { + ran_short = true; + 0.0 + }); } cb_counters .pcm_written @@ -196,20 +198,15 @@ impl AudioPlayback { out.fill(0.0); cb_counters.underruns.fetch_add(1, Ordering::Relaxed); } - // Re-prime only after a RUN of empty callbacks, not a single transient one — - // otherwise every momentary drain costs a fresh 40 ms silence (the old behaviour, - // self-inflicted crackle on any jitter spike). - if ring.is_empty() { - empties += 1; - if empties >= DEPRIME_AFTER_CALLBACKS { - primed = false; - } - } else { - empties = 0; - } + // No-op while un-primed, so a deliberate priming silence is never counted as an + // underrun (which would otherwise drive the adaptive floor up for no reason). + policy.note_read(ran_short); cb_counters .ring_depth .store(ring.len() as u64, Ordering::Relaxed); + cb_counters + .target_ms + .store(policy.target_ms() as u64, Ordering::Relaxed); // Google's AAudio anti-glitch technique: when the device reports new XRuns, grow the // HW buffer by one burst (up to capacity). getXRunCount + setBufferSizeInFrames are // both callback-safe / non-blocking, and set clamps to capacity so it self-limits. @@ -408,10 +405,11 @@ fn decode_loop( } if count % 600 == 0 { log::info!( - "audio: opus={count} pcm_frames={} underruns={} ring={} peak={window_peak:.3}", + "audio: opus={count} pcm_frames={} underruns={} buffer_ms={} target_ms={} peak={window_peak:.3}", counters.pcm_written.load(Ordering::Relaxed), counters.underruns.load(Ordering::Relaxed), - counters.ring_depth.load(Ordering::Relaxed), + counters.ring_depth.load(Ordering::Relaxed) / ms.max(1) as u64, + counters.target_ms.load(Ordering::Relaxed), ); window_peak = 0.0; } diff --git a/clients/apple/Sources/PunktfunkKit/Audio/AudioRing.swift b/clients/apple/Sources/PunktfunkKit/Audio/AudioRing.swift index bca90d98..7654111d 100644 --- a/clients/apple/Sources/PunktfunkKit/Audio/AudioRing.swift +++ b/clients/apple/Sources/PunktfunkKit/Audio/AudioRing.swift @@ -3,28 +3,61 @@ import os /// SPSC-ish jitter ring (interleaved float, `channels` per frame), drain thread → render /// callback. The unfair lock is held for microseconds; fine at render-callback rates. Priming: -/// reads return silence until enough is buffered (at least `prefill`, and at least one +/// reads return silence until enough is buffered (at least the target, and at least one /// packet more than the device's render quantum — large-buffer devices would otherwise -/// chronically out-demand the prefill and oscillate prime → dropout → re-prime), and an -/// underrun re-primes, concealing jitter as one short dip instead of sustained crackle. +/// chronically out-demand the prefill and oscillate prime → dropout → re-prime). /// All counts stay whole frames (multiples of `channels`), so the interleave can never slip. +/// +/// **Drift correction.** Both ends run at 48 kHz but on different crystals, so backlog from a +/// network stall or plain host-vs-DAC skew never drains on its own: without correction one 300 ms +/// hiccup leaves audio 300 ms behind video for the rest of the session. This used to be handled by +/// a `highWater` shed that dropped a whole `2 × prefill` at once — its own comment called that "one +/// audible blip". It is now the same two-stage scheme the Rust clients share +/// (`punktfunk_core::audio::JitterPolicy`): a slow depth average that sits above target for a +/// sustained window sheds ONE 5 ms frame with a crossfade, and the hard cap is only a backstop. +/// Keep the constants here in step with `JitterTuning.COREAUDIO`. final class AudioRing: @unchecked Sendable { + /// Mirrors `JitterTuning::COREAUDIO` — see that type for the rationale. + private static let targetMS = 20 + private static let headroomMS = 30 + private static let hardCapMS = 90 + private static let deprimeAfter = 4 + /// The protocol's frame: the shed unit, and the slack added over a large device quantum. + private static let frameMS = 5 + /// Depth average must exceed target by this before drift correction fires — the middle of the + /// headroom band, so the smooth shed always gets its chance BEFORE the hard cap trims. + private static let shedExcessMS = 15 + /// …and must stay there for this much consumed audio. Long, because a shed is the only thing + /// here a listener could notice; it must never fire on a transient. + private static let shedSustainMS = 2_000 + private static let crossfadeMS = 2 + /// Time constant of the depth average. + private static let ewmaTauMS = 1_000 + private var buf: [Float] private var readIdx = 0 private var writeIdx = 0 private var primed = false private var renderQuantum = 0 - private let prefill: Int - private let highWater: Int + private var emptyReads = 0 + private var depthAvg: Double = 0 + private var overRun = 0 private let channels: Int + private let perMS: Int private let lock = OSAllocatedUnfairLock() /// `capacity`/`prefill` in samples (interleaved — `channels` per frame, both whole frames). - init(capacity: Int, prefill: Int, channels: Int) { + /// `prefill` is accepted for source compatibility but the target now comes from `targetMS`. + init(capacity: Int, prefill: Int = 0, channels: Int) { buf = [Float](repeating: 0, count: capacity) - self.prefill = prefill self.channels = channels - highWater = prefill * 4 + perMS = 48 * channels + } + + /// Live target depth in interleaved samples, lifted so it can always serve one device quantum + /// plus a packet (a large-buffer device cannot sustain a target below its own quantum). + private var target: Int { + max(Self.targetMS * perMS, renderQuantum + Self.frameMS * perMS) } func write(_ samples: UnsafePointer, count: Int) { @@ -42,12 +75,12 @@ final class AudioRing: @unchecked Sendable { buf[(writeIdx + i) % capacity] = samples[i] } writeIdx += count - // Latency clamp: both ends run at 48 kHz, so backlog from a network stall (or - // creeping host-vs-DAC clock skew) never drains on its own — without this, one - // 300 ms hiccup leaves audio 300 ms behind video for the rest of the session. - // Shedding down to 2× prefill costs one audible blip instead. - if writeIdx - readIdx > highWater { - readIdx = writeIdx - prefill * 2 + // Backstop only: the smooth shed in `read` is what normally holds the depth down. + let cap = min(target + Self.headroomMS * perMS, Self.hardCapMS * perMS) + if writeIdx - readIdx > cap { + readIdx = writeIdx - cap + depthAvg = Double(cap) + overRun = 0 } } @@ -57,16 +90,36 @@ final class AudioRing: @unchecked Sendable { defer { lock.unlock() } renderQuantum = max(renderQuantum, count) let available = writeIdx - readIdx + + // Depth average, weighted by the callback size so its time constant is independent of the + // device quantum. + let alpha = min(1.0, Double(count) / Double(Self.ewmaTauMS * perMS)) + depthAvg += (Double(available) - depthAvg) * alpha + if !primed { - // One 5 ms host packet (240 frames × channels) of slack beyond the device's demand. - if available >= max(prefill, renderQuantum + 240 * channels) { + if available >= target { primed = true + emptyReads = 0 } else { for i in 0.. Double(target + Self.shedExcessMS * perMS) { + overRun += count + if overRun >= Self.shedSustainMS * perMS { + overRun = 0 + shedOneFrame() + depthAvg = Double(writeIdx - readIdx) + } + } else { + overRun = 0 + } + + let n = min(writeIdx - readIdx, count) let capacity = buf.count for i in 0..= Self.deprimeAfter { primed = false } + } else { + emptyReads = 0 } } + + /// Drop one protocol frame from the front, linearly crossfading the seam so the correction is + /// inaudible rather than a click. Mirrors `punktfunk_core::audio::crossfade_drop`; caller holds + /// the lock. + private func shedOneFrame() { + let drop = Self.frameMS * perMS + let available = writeIdx - readIdx + guard available > drop else { return } + let fade = min(Self.crossfadeMS * perMS, min(drop, available - drop)) + let capacity = buf.count + if fade > 0 { + // The tail of what we discard fades out into the head of what survives. + for i in 0.. (Int, Int, Int) { + let ring = AudioRing(capacity: 48_000 * channels, channels: channels) + let want = quantumMS * perMS + var scratch = [Float](repeating: 0, count: want) + // Non-zero so a silent callback is distinguishable from real audio. + let producer = [Float](repeating: 0.25, count: want + 8) + var carry = 0, peak = 0, final = 0, silent = 0 + + for i in 0..<(ms / quantumMS) { + carry += want * driftPPM + let extra = carry / 1_000_000 + carry -= extra * 1_000_000 + producer.withUnsafeBufferPointer { ring.write($0.baseAddress!, count: want + extra) } + + scratch.withUnsafeMutableBufferPointer { ring.read(into: $0.baseAddress!, count: want) } + // Skip the priming window at the very start. + if i > 20, scratch.allSatisfy({ $0 == 0 }) { silent += 1 } + peak = max(peak, ring.bufferedMS) + final = ring.bufferedMS + } + return (final, peak, silent) + } + + /// THE regression: with the host clock running fast, buffered latency must return to target + /// instead of climbing to the hard cap and staying pinned there. +200 ppm is deliberately + /// harsher than real hardware (tens of ppm). + func testDriftDoesNotRatchetLatencyToTheCeiling() { + let (final, peak, silent) = simulate(ms: 5 * 60 * 1_000, quantumMS: 5, driftPPM: 200) + // Must settle inside the headroom band (target 20 + headroom 30), never near the 90 ms cap. + XCTAssertLessThanOrEqual(final, 50, "settled at \(final) ms — that is the ratchet") + XCTAssertLessThanOrEqual(peak, 50, "peaked at \(peak) ms") + XCTAssertEqual(silent, 0, "drift correction must never starve the callback") + } + + /// The mirror case: a host clock running SLOW must keep audio flowing rather than being + /// "corrected" into a stutter. + func testNegativeDriftKeepsPlaying() { + let (_, _, silent) = simulate(ms: 2 * 60 * 1_000, quantumMS: 5, driftPPM: -200) + XCTAssertEqual(silent, 0, "a draining ring must re-prime, not chatter") + } + + /// A device that pulls a large quantum cannot sustain a target below it — the ring must lift + /// its target rather than oscillating prime → dropout → re-prime forever. + func testLargeDeviceQuantumStillPlays() { + let (_, _, silent) = simulate(ms: 60 * 1_000, quantumMS: 40, driftPPM: 0) + XCTAssertEqual(silent, 0, "a 40 ms quantum must not starve a 20 ms target") + } + + /// One transient drain must not manufacture a whole target's worth of fresh silence: the ring + /// de-primes only after a RUN of short reads. + func testSingleShortReadDoesNotDeprime() { + let ring = AudioRing(capacity: 48_000 * channels, channels: channels) + let want = 5 * perMS + var scratch = [Float](repeating: 0, count: want) + // Prime well past target. + let big = [Float](repeating: 0.5, count: 60 * perMS) + big.withUnsafeBufferPointer { ring.write($0.baseAddress!, count: big.count) } + scratch.withUnsafeMutableBufferPointer { ring.read(into: $0.baseAddress!, count: want) } + XCTAssertTrue(scratch.contains { $0 != 0 }, "should be playing after priming") + + // Drain it dry with one oversized read, then feed a normal quantum again. + var huge = [Float](repeating: 0, count: 200 * perMS) + huge.withUnsafeMutableBufferPointer { ring.read(into: $0.baseAddress!, count: huge.count) } + let feed = [Float](repeating: 0.5, count: want) + feed.withUnsafeBufferPointer { ring.write($0.baseAddress!, count: want) } + scratch.withUnsafeMutableBufferPointer { ring.read(into: $0.baseAddress!, count: want) } + XCTAssertTrue( + scratch.contains { $0 != 0 }, + "a single short read must not force a full re-prime") + } +} +#endif diff --git a/crates/pf-client-core/src/audio.rs b/crates/pf-client-core/src/audio.rs index 30ceaaf7..b70ff4f9 100644 --- a/crates/pf-client-core/src/audio.rs +++ b/crates/pf-client-core/src/audio.rs @@ -168,9 +168,18 @@ struct PlayerData { /// Drained chunk Vecs go back here for the decode side to refill (allocation pool). recycle: SyncSender>, ring: VecDeque, - primed: bool, + /// Shared ms-denominated de-jitter policy: prime depth, drift correction, de-prime + /// hysteresis. Replaces the old `3 × quantum` target, which meant 15 ms at a 5 ms graph + /// quantum and a silent 64 ms at a 20 ms one, and the `if ring.is_empty()` re-prime, where + /// one transient drain manufactured a whole target's worth of fresh silence. + policy: punktfunk_core::audio::JitterPolicy, /// Interleaved channel count this stream was opened with (2/6/8). channels: usize, + /// Diagnostics (WP0.3), logged ~every 10 s: the audio plane used to be entirely silent in a + /// client log, so a latency or dropout report had nothing to go on. + underruns: u64, + sheds: u64, + callbacks: u64, } fn pw_thread( @@ -223,8 +232,14 @@ fn pw_thread( rx: pcm_rx, recycle: recycle_tx, ring: VecDeque::new(), - primed: false, + policy: punktfunk_core::audio::JitterPolicy::new( + punktfunk_core::audio::JitterTuning::PIPEWIRE, + channels as u8, + ), channels, + underruns: 0, + sheds: 0, + callbacks: 0, }; let _listener = stream @@ -252,23 +267,29 @@ fn pw_thread( let want_frames = data.data().map(|s| s.len() / stride).unwrap_or(0); let want = want_frames * ud.channels; - // Adaptive jitter buffer (same shape as the host's virtual mic): prime to - // ~3 quanta, cap at ~1 quantum of slack beyond that, re-prime after a - // genuine drain. - let target = (3 * want).clamp(720 * ud.channels, 9600 * ud.channels); - while ud.ring.len() > target.max(want) + want { - ud.ring.pop_front(); - } - if !ud.primed && ud.ring.len() >= target { - ud.primed = true; + // 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. + let step = ud.policy.step(ud.ring.len(), want); + if step.drop_front > 0 { + ud.sheds += 1; + punktfunk_core::audio::crossfade_drop( + &mut ud.ring, + step.drop_front, + step.crossfade, + ); } + let mut ran_short = false; let n_frames = if let Some(slice) = data.data() { for k in 0..want { - let s = if ud.primed { - ud.ring.pop_front().unwrap_or(0.0) - } else { + let s = if step.silence { 0.0 + } else { + ud.ring.pop_front().unwrap_or_else(|| { + ran_short = true; + 0.0 + }) }; let off = k * 4; slice[off..off + 4].copy_from_slice(&s.to_le_bytes()); @@ -277,8 +298,21 @@ fn pw_thread( } else { 0 }; - if ud.ring.is_empty() { - ud.primed = false; + // No-op while un-primed (the policy ignores it), so a deliberate priming silence + // is never miscounted as an underrun. + ud.policy.note_read(ran_short); + ud.underruns += u64::from(ran_short); + ud.callbacks += 1; + // ~10 s at a 5 ms quantum; the exact cadence does not matter, only that the + // plane stops being invisible. + if ud.callbacks % 2_000 == 0 { + tracing::debug!( + buffer_ms = ud.policy.avg_depth_ms(), + target_ms = ud.policy.target_ms(), + underruns = ud.underruns, + drift_sheds = ud.sheds, + "audio playback" + ); } let chunk = data.chunk_mut(); *chunk.offset_mut() = 0; diff --git a/crates/pf-client-core/src/audio_wasapi.rs b/crates/pf-client-core/src/audio_wasapi.rs index 2df9f3b8..0c9b6d69 100644 --- a/crates/pf-client-core/src/audio_wasapi.rs +++ b/crates/pf-client-core/src/audio_wasapi.rs @@ -250,10 +250,20 @@ fn render_thread( audio_client.start_stream().context("start render stream")?; let _ = ready.send(Ok(())); - // Adaptive jitter buffer, in f32-byte units (same shape as the host's virtual mic). - let mut ring: VecDeque = VecDeque::new(); - let mut primed = false; + // De-jitter ring, in interleaved f32 SAMPLES (it used to be raw bytes, which made the + // depth arithmetic byte-vs-sample and kept it from sharing the policy and the crossfade + // helper with the other three clients). + let mut ring: VecDeque = VecDeque::new(); + // Shared ms-denominated policy: prime depth, crossfaded drift correction so latency + // returns to target instead of ratcheting, and de-prime hysteresis — the last replacing + // the old `if ring.is_empty()`, where a single transient drain manufactured a whole + // target's worth of fresh silence. + let mut policy = punktfunk_core::audio::JitterPolicy::new( + punktfunk_core::audio::JitterTuning::WASAPI, + channels, + ); let mut out = Vec::new(); // per-quantum scratch, reused across iterations + let (mut underruns, mut sheds, mut callbacks) = (0u64, 0u64, 0u64); while !stop.load(Ordering::Relaxed) { if h_event.wait_for_event(100).is_err() { @@ -262,9 +272,7 @@ fn render_thread( // Drain everything the pump has queued into the ring, returning each drained // Vec to the pool (a full/closed pool drops it). while let Ok(mut chunk) = pcm_rx.try_recv() { - for s in chunk.iter() { - ring.extend(s.to_le_bytes()); - } + ring.extend(chunk.iter().copied()); chunk.clear(); let _ = recycle_tx.try_send(chunk); } @@ -274,28 +282,40 @@ fn render_thread( if avail_frames == 0 { continue; } - let want_bytes = avail_frames * block_align; + let want = avail_frames * channels as usize; - // Prime to ~3 quanta; cap at ~1 quantum of slack beyond that; re-prime on drain. - let target = (3 * want_bytes).clamp(720 * block_align, 9600 * block_align); - let cap = target.max(want_bytes) + want_bytes; - if ring.len() > cap { - ring.drain(..ring.len() - cap); - } - if !primed && ring.len() >= target { - primed = true; + let step = policy.step(ring.len(), want); + if step.drop_front > 0 { + sheds += 1; + punktfunk_core::audio::crossfade_drop(&mut ring, step.drop_front, step.crossfade); } out.clear(); - out.resize(want_bytes, 0); - if primed { - let n = ring.len().min(want_bytes); - for (dst, b) in out.iter_mut().zip(ring.drain(..n)) { - *dst = b; + out.resize(avail_frames * block_align, 0); + let mut ran_short = false; + if !step.silence { + // `out` is exactly `want` f32s wide (avail_frames × channels × 4 bytes). + for dst in out.chunks_exact_mut(4) { + let s = ring.pop_front().unwrap_or_else(|| { + ran_short = true; + 0.0 + }); + dst.copy_from_slice(&s.to_le_bytes()); } } - if ring.is_empty() { - primed = false; + // No-op while un-primed (the policy ignores it), so a deliberate priming silence is + // never miscounted as an underrun. + policy.note_read(ran_short); + underruns += u64::from(ran_short); + callbacks += 1; + if callbacks % 1_000 == 0 { + tracing::debug!( + buffer_ms = policy.avg_depth_ms(), + target_ms = policy.target_ms(), + underruns, + drift_sheds = sheds, + "audio playback" + ); } render_client .write_to_device(avail_frames, &out, None) diff --git a/crates/punktfunk-core/src/audio.rs b/crates/punktfunk-core/src/audio.rs index a006fc51..f20d7780 100644 --- a/crates/punktfunk-core/src/audio.rs +++ b/crates/punktfunk-core/src/audio.rs @@ -243,6 +243,54 @@ impl AudioGapTracker { } } +/// Rebuilds the audio stream from the redundant `0xD2` plane, so a single lost datagram is +/// RECOVERED rather than concealed. +/// +/// Deliberately lives in core, on the demux side, rather than in the four client decoders. The +/// recovered frame is re-inserted into the same queue in order, so every embedder — Linux, +/// Windows, Android, Apple, and any C-ABI consumer — gets a complete stream with no change at all, +/// and their [`AudioGapTracker`] simply stops seeing the gap. +/// +/// **Only the immediately-preceding frame can be recovered**, because that is all the wire carries +/// (see [`crate::quic::encode_audio_red_datagram`]). A longer burst still falls through to +/// packet-loss concealment — but it falls through one frame shorter, which is strictly better. +#[derive(Debug, Default)] +pub struct AudioRedRecovery { + /// Sequence of the newest packet handed downstream. + last_seq: Option, +} + +impl AudioRedRecovery { + pub fn new() -> Self { + Self::default() + } + + /// Feed the arriving datagram's sequence and whether it carried a redundant copy. Returns + /// `true` when that copy should be emitted (as `seq - 1`) BEFORE the packet itself. + /// + /// Wrapping-safe, and conservative in both directions: a reorder or duplicate recovers + /// nothing, and neither does the first packet of a session (nothing is known to be missing). + pub fn recover_before(&mut self, seq: u32, has_prev: bool) -> bool { + let recover = match self.last_seq { + // Nothing emitted yet: no evidence anything was lost, so inserting the predecessor + // would prepend audio the client never missed. + None => false, + Some(last) => { + let delta = seq.wrapping_sub(last); + // `delta == 1` is in-order; `delta >= 2` (forward half of the space only) means + // at least the predecessor is missing. + has_prev && (2..u32::MAX / 2).contains(&delta) + } + }; + self.last_seq = Some(match self.last_seq { + // A reorder must not drag the anchor backwards. + Some(last) if seq.wrapping_sub(last) > u32::MAX / 2 => last, + _ => seq, + }); + recover + } +} + // ---- the shared playback de-jitter policy ------------------------------------------------- /// The protocol's audio frame, in milliseconds — every host datagram carries exactly one @@ -692,6 +740,92 @@ mod tests { assert_eq!(t.missing_before(0), 0, "pre-wrap reorder, not a 2^31 gap"); } + // ---- redundant-plane recovery --------------------------------------------------------- + + #[test] + fn red_recovery_rebuilds_exactly_the_single_missing_frame() { + let mut r = AudioRedRecovery::new(); + // First packet: nothing is known to be missing, so nothing is prepended. + assert!(!r.recover_before(10, true)); + // In order. + assert!(!r.recover_before(11, true)); + // 12 lost: 13 carries it. + assert!(r.recover_before(13, true)); + // Back in order from the new anchor. + assert!(!r.recover_before(14, true)); + } + + #[test] + fn red_recovery_is_conservative() { + let mut r = AudioRedRecovery::new(); + r.recover_before(10, true); + // A datagram with no redundant copy recovers nothing, however big the gap. + assert!(!r.recover_before(20, false)); + // Duplicates and reorders recover nothing, and must not move the anchor backwards. + let mut r = AudioRedRecovery::new(); + r.recover_before(10, true); + r.recover_before(11, true); + assert!(!r.recover_before(11, true), "duplicate"); + assert!(!r.recover_before(9, true), "late reorder"); + assert!( + !r.recover_before(12, true), + "the reorder must not have moved the anchor" + ); + } + + /// A longer burst still recovers its last frame — the gap the client has to conceal gets one + /// frame shorter, which is strictly better than concealing all of it. + #[test] + fn red_recovery_shortens_a_longer_burst() { + let mut r = AudioRedRecovery::new(); + r.recover_before(100, true); + assert!( + r.recover_before(105, true), + "104 is recoverable even though 101-103 are not" + ); + } + + #[test] + fn red_recovery_survives_seq_wraparound() { + let mut r = AudioRedRecovery::new(); + assert!(!r.recover_before(u32::MAX - 1, true)); + assert!( + !r.recover_before(u32::MAX, true), + "in order across the edge" + ); + assert!(r.recover_before(1, true), "seq 0 lost across the wrap"); + assert!(!r.recover_before(2, true)); + } + + /// The two halves must agree: whatever `AudioRedRecovery` rebuilds, `AudioGapTracker` must + /// then see as no gap at all — that is the whole point of doing recovery on the demux side. + #[test] + fn recovery_and_the_gap_tracker_agree() { + let mut rec = AudioRedRecovery::new(); + let mut gaps = AudioGapTracker::new(); + let mut concealed = 0; + // Deliver 0..20 with 7 and 13 lost; each survivor carries its predecessor. + let mut emitted: Vec = Vec::new(); + for seq in (0..20u32).filter(|s| *s != 7 && *s != 13) { + if rec.recover_before(seq, true) { + emitted.push(seq - 1); + } + emitted.push(seq); + } + for seq in &emitted { + concealed += gaps.missing_before(*seq); + } + assert_eq!( + concealed, 0, + "recovered stream must need no concealment: {emitted:?}" + ); + assert_eq!(emitted.len(), 20, "every frame accounted for"); + assert!( + emitted.windows(2).all(|w| w[1] == w[0] + 1), + "and in order: {emitted:?}" + ); + } + // ---- bitrate tiers ------------------------------------------------------------------- /// `Standard` must reproduce the historical table EXACTLY — that is what makes the tier diff --git a/crates/punktfunk-core/src/client/mod.rs b/crates/punktfunk-core/src/client/mod.rs index eaece606..0d4c48d6 100644 --- a/crates/punktfunk-core/src/client/mod.rs +++ b/crates/punktfunk-core/src/client/mod.rs @@ -490,7 +490,13 @@ impl NativeClient { video_codecs, preferred_codec, display_hdr, - client_caps, + // Redundant audio (`0xD2`) is advertised by CORE, not by the embedder: the + // recovery happens on the demux side (`AudioRedRecovery` in the datagram + // task) and re-inserts the rebuilt frame into the same queue, so every + // embedder benefits without knowing the plane exists — and none of them can + // forget to opt in. The bit is a pure "I can decode it"; the host still + // decides whether to spend the extra ~1 %. + client_caps: client_caps | crate::quic::CLIENT_CAP_AUDIO_RED, frame_parts, launch, name, diff --git a/crates/punktfunk-core/src/client/pump/datagram_task.rs b/crates/punktfunk-core/src/client/pump/datagram_task.rs index b62c8dd7..59bae19f 100644 --- a/crates/punktfunk-core/src/client/pump/datagram_task.rs +++ b/crates/punktfunk-core/src/client/pump/datagram_task.rs @@ -23,6 +23,10 @@ pub(super) async fn run( // gate): a datagram the network reordered must not roll a stopped motor back on. Legacy v1 // datagrams carry no seq and bypass it (an old host's own periodic re-send is the only heal). let mut rumble_last_seq: [Option; crate::input::MAX_PADS] = [None; crate::input::MAX_PADS]; + // Redundant-audio-plane rebuild (`0xD2`). Recovery happens HERE rather than in the four + // client decoders: the recovered frame is re-inserted into this queue in order, so every + // embedder gets a complete stream without knowing the plane exists. + let mut audio_red = crate::audio::AudioRedRecovery::new(); while let Ok(d) = conn.read_datagram().await { match d.first() { Some(&crate::quic::AUDIO_MAGIC) => { @@ -34,6 +38,26 @@ pub(super) async fn run( }); } } + Some(&crate::quic::AUDIO_RED_MAGIC) => { + if let Some((seq, pts_ns, opus, prev)) = crate::quic::decode_audio_red_datagram(&d) + { + if audio_red.recover_before(seq, prev.is_some()) { + // The copy is the frame BEFORE this one, so it carries the previous + // sequence and presentation time — one protocol frame earlier. + let _ = audio_tx.try_send(AudioPacket { + seq: seq.wrapping_sub(1), + pts_ns: pts_ns + .saturating_sub(crate::audio::FRAME_MS as u64 * 1_000_000), + data: prev.unwrap_or_default().to_vec(), + }); + } + let _ = audio_tx.try_send(AudioPacket { + seq, + pts_ns, + data: opus.to_vec(), + }); + } + } Some(&crate::quic::RUMBLE_MAGIC) => { if let Some(u) = crate::quic::decode_rumble_envelope(&d) { // Gate v2 envelopes on their per-pad seq; forward v1 (envelope: None) as-is.