diff --git a/clients/apple/Sources/PunktfunkKit/Audio/AudioRing.swift b/clients/apple/Sources/PunktfunkKit/Audio/AudioRing.swift index f139d82e..222d4be0 100644 --- a/clients/apple/Sources/PunktfunkKit/Audio/AudioRing.swift +++ b/clients/apple/Sources/PunktfunkKit/Audio/AudioRing.swift @@ -15,10 +15,16 @@ import os /// 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`. +/// +/// **Adaptive depth.** The target is a floor, not a constant: repeated genuine underruns grow it +/// a step at a time (`noteRead`, mirroring `JitterPolicy::note_read`) up to `maxTargetMS`, and a +/// long quiet spell relaxes it back toward the base — so a session on Wi-Fi that bunches arrivals +/// deepens until it stops crackling, while a clean LAN keeps the tight base latency. 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 maxTargetMS = 70 private static let headroomMS = 30 private static let hardCapMS = 90 private static let deprimeAfter = 4 @@ -33,6 +39,15 @@ final class AudioRing: @unchecked Sendable { private static let crossfadeMS = 2 /// Time constant of the depth average. private static let ewmaTauMS = 1_000 + /// Adaptive target floor, mirroring `JitterPolicy::note_read`: this many genuine underruns + /// inside one window grow the live target a step (up to `maxTargetMS`), and a long quiet + /// spell relaxes it a step back toward the base — so only the sessions that actually starve + /// (Wi-Fi power-save bunching is the classic) pay for extra depth, and only while they need + /// it. All spans are measured in consumed samples, like the Rust policy. + private static let growUnderruns = 3 + private static let growWindowMS = 5_000 + private static let growStepMS = 10 + private static let shrinkQuietMS = 30_000 private var buf: [Float] private var readIdx = 0 @@ -42,6 +57,14 @@ final class AudioRing: @unchecked Sendable { private var emptyReads = 0 private var depthAvg: Double = 0 private var overRun = 0 + /// The live target in interleaved samples — `targetMS` grown by underrun pressure + /// (`noteRead`), never below the base. Set in `init` (needs `perMS`). + private var targetLive = 0 + /// Underruns seen in the current growth window, and the window's consumed-sample count. + private var underrunsInWindow = 0 + private var windowRun = 0 + /// Consumed samples since the last underrun (drives the relax-back-down step). + private var quietRun = 0 /// Reported, not acted on: short reads that actually starved the callback, and smooth drift /// corrections. A rising underrun count means the ring is being starved (network or CPU), /// which is a different problem from the depth being wrong. @@ -57,12 +80,14 @@ final class AudioRing: @unchecked Sendable { buf = [Float](repeating: 0, count: capacity) self.channels = channels perMS = 48 * channels + targetLive = Self.targetMS * perMS } - /// 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). + /// Effective target depth in interleaved samples: the (adaptively grown) live target, 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) + max(targetLive, renderQuantum + Self.frameMS * perMS) } func write(_ samples: UnsafePointer, count: Int) { @@ -80,8 +105,13 @@ final class AudioRing: @unchecked Sendable { buf[(writeIdx + i) % capacity] = samples[i] } writeIdx += count - // Backstop only: the smooth shed in `read` is what normally holds the depth down. - let cap = min(target + Self.headroomMS * perMS, Self.hardCapMS * perMS) + // Backstop only: the smooth shed in `read` is what normally holds the depth down. The + // hard cap must always leave room for one device quantum past the target (mirrors the + // Rust policy's `.max(target + want)`) or a large-quantum device would trim itself into + // a permanent underrun. + let cap = max( + min(target + Self.headroomMS * perMS, Self.hardCapMS * perMS), + target + renderQuantum) if writeIdx - readIdx > cap { readIdx = writeIdx - cap depthAvg = Double(cap) @@ -133,13 +163,43 @@ final class AudioRing: @unchecked Sendable { readIdx += n if n < count { for i in n..= Self.growWindowMS * perMS { + windowRun = 0 + underrunsInWindow = 0 + } + if ranShort { + quietRun = 0 emptyReads += 1 underrunCount += 1 - if emptyReads >= Self.deprimeAfter { primed = false } + if emptyReads >= Self.deprimeAfter { + primed = false + emptyReads = 0 + } + underrunsInWindow += 1 + if underrunsInWindow >= Self.growUnderruns { + underrunsInWindow = 0 + windowRun = 0 + targetLive = min(targetLive + Self.growStepMS * perMS, Self.maxTargetMS * perMS) + } } else { emptyReads = 0 + quietRun += count + if quietRun >= Self.shrinkQuietMS * perMS { + quietRun = 0 + targetLive = max(targetLive - Self.growStepMS * perMS, Self.targetMS * perMS) + } } } diff --git a/clients/apple/Tests/PunktfunkKitTests/AudioRingDriftTests.swift b/clients/apple/Tests/PunktfunkKitTests/AudioRingDriftTests.swift index a86949dd..7fb85be1 100644 --- a/clients/apple/Tests/PunktfunkKitTests/AudioRingDriftTests.swift +++ b/clients/apple/Tests/PunktfunkKitTests/AudioRingDriftTests.swift @@ -91,5 +91,112 @@ final class AudioRingDriftTests: XCTestCase { scratch.contains { $0 != 0 }, "a single short read must not force a full re-prime") } + + /// Mirror of the Rust `target_grows_on_underruns_and_relaxes_when_quiet`: clustered genuine + /// underruns raise the target floor (that session needs the slack), a long quiet spell gives + /// it back — and the floor never dips below the base. + func testTargetGrowsOnUnderrunsAndRelaxesWhenQuiet() { + let ring = AudioRing(capacity: 48_000 * channels, channels: channels) + let want = 5 * perMS + var scratch = [Float](repeating: 0, count: want) + let feed = [Float](repeating: 0.5, count: 25 * perMS) + func write(ms: Int) { + feed.withUnsafeBufferPointer { ring.write($0.baseAddress!, count: ms * perMS) } + } + func read() { + scratch.withUnsafeMutableBufferPointer { ring.read(into: $0.baseAddress!, count: want) } + } + XCTAssertEqual(ring.stats.targetMS, 20, "base target must match JitterTuning.COREAUDIO") + + // Prime, drain dry, then alternate starve/refill: each dry read is a genuine underrun, + // each full read in between keeps the de-prime hysteresis from tripping. + write(ms: 25) + for _ in 0..<5 { read() } // drains to zero + read() // short — underrun 1 + write(ms: 5); read() // full — hysteresis reset + read() // short — underrun 2 + write(ms: 5); read() // full + read() // short — underrun 3 → the floor grows one step + XCTAssertEqual(ring.stats.targetMS, 30, "3 clustered underruns must grow the target 10 ms") + XCTAssertEqual(ring.stats.underruns, 3) + + // A long clean run (30 s of consumed audio) relaxes the growth back to the base… + for _ in 0..<(30_000 / 5 + 10) { + write(ms: 5) + read() + } + XCTAssertEqual(ring.stats.targetMS, 20, "a quiet spell must give the growth back") + // …and stays there: quiet forever never dips below the base. + for _ in 0..<(30_000 / 5 + 10) { + write(ms: 5) + read() + } + XCTAssertEqual(ring.stats.targetMS, 20, "the floor must never go below the base target") + } + + /// Growth is capped at `maxTargetMS`, exactly like `JitterPolicy` respects + /// `JitterTuning.max_target_ms`. + func testTargetGrowthRespectsTheCap() { + let ring = AudioRing(capacity: 48_000 * channels, channels: channels) + let want = 5 * perMS + var scratch = [Float](repeating: 0, count: want) + let feed = [Float](repeating: 0.5, count: 25 * perMS) + feed.withUnsafeBufferPointer { ring.write($0.baseAddress!, count: 25 * perMS) } + // Starve it far past what six growth steps (20 → 70) would need. + for _ in 0..<40 { + for _ in 0..<5 { + scratch.withUnsafeMutableBufferPointer { + ring.read(into: $0.baseAddress!, count: want) + } + } + feed.withUnsafeBufferPointer { ring.write($0.baseAddress!, count: 25 * perMS) } + } + XCTAssertLessThanOrEqual(ring.stats.targetMS, 70, "growth must respect maxTargetMS") + } + + /// THE field scenario: Wi-Fi power-save bunches arrivals — audio is produced steadily but + /// delivered in bursts, some of them late. A fixed 20 ms target crackles on every late burst + /// forever; the adaptive floor must deepen until the bunching rides through, and the tail of + /// the session must be silence-free. + func testWifiBunchingConvergesToSilenceFree() { + let ring = AudioRing(capacity: 48_000 * channels, channels: channels) + let want = 5 * perMS + var scratch = [Float](repeating: 0, count: want) + var pending = 0 // ms produced by the host but still "in flight" + var burst = 0 + var silentTail = 0 + let steps = 4000 // 20 s in 5 ms callbacks + let feed = [Float](repeating: 0.5, count: 200 * perMS) + for step in 0..= 60 { + // The held burst lands, together with everything produced since. + feed.withUnsafeBufferPointer { + ring.write($0.baseAddress!, count: pending * perMS) + } + pending = 0 + } + scratch.withUnsafeMutableBufferPointer { ring.read(into: $0.baseAddress!, count: want) } + if step >= steps - 600, scratch.allSatisfy({ $0 == 0 }) { silentTail += 1 } + } + XCTAssertGreaterThanOrEqual( + ring.stats.targetMS, 30, + "bunched delivery must have grown the target floor") + XCTAssertEqual( + silentTail, 0, + "after adapting, the last 3 s must play through the bunching without a dropout") + } } #endif diff --git a/crates/punktfunk-core/src/abi.rs b/crates/punktfunk-core/src/abi.rs index 686b6c38..82a6b008 100644 --- a/crates/punktfunk-core/src/abi.rs +++ b/crates/punktfunk-core/src/abi.rs @@ -654,10 +654,88 @@ pub struct PunktfunkConnection { #[derive(Default)] struct AudioPcmState { decoder: Option, - /// Interleaved f32 PCM, wire channel order. Pre-sized to the largest legal Opus frame - /// (120 ms @ 48 kHz = 5760 samples/ch) × 8 channels so decode never reallocates (which would + /// Interleaved f32 PCM, wire channel order. Pre-sized in `decode_packet` for the largest + /// legal Opus frame plus a full concealment run, so decode never reallocates (which would /// dangle the pointer handed to the embedder). pcm: Vec, + /// Loss detector — the same seq-gap accounting the other clients run in their own decode + /// loops (`pf-client-core`'s session pump, Android's native pump), here for the one decoder + /// that lives in core. Without it a lost 5 ms packet reaches the embedder's playout ring as + /// a hard time-domain gap: a click per loss, sustained crackle on lossy Wi-Fi. + gaps: crate::audio::AudioGapTracker, + /// Per-channel sample count of the last real decode — sizes each synthesized concealment + /// frame. 0 until the first decode, which skips concealment (nothing to size it from), + /// exactly like the other clients. + frame_samples: usize, +} + +#[cfg(feature = "quic")] +impl AudioPcmState { + /// Decode one arriving audio packet into `self.pcm`, synthesizing libopus packet-loss + /// concealment for any packets the sequence says went missing immediately before it — the + /// concealed frames land first, the real frame after, one contiguous interleaved buffer. + /// + /// Returns the interleaved sample count now valid at the front of `pcm`; `Ok(0)` means + /// nothing to hand out this call (a DTX silence marker with no loss before it). An empty + /// `data` is the DTX marker: it still advances the loss accounting (so the silent slot is + /// never itself "concealed" later) and flushes any concealment owed, but is never decoded — + /// `decode_float` would treat it as a loss and synthesize the buffer's full capacity. + fn decode_packet( + &mut self, + data: &[u8], + seq: u32, + channels: u8, + ) -> Result { + let ch = channels as usize; + if self.decoder.is_none() { + let layout = crate::audio::layout_for(channels, false); + match opus::MSDecoder::new(48_000, layout.streams, layout.coupled, layout.mapping) { + Ok(d) => { + // Largest legal Opus frame is 120 ms = 5760 samples/ch, and a gap can owe up + // to MAX_CONCEAL_PACKETS concealed frames of the same size in front of it. + self.pcm = + vec![0f32; (1 + crate::audio::MAX_CONCEAL_PACKETS as usize) * 5760 * ch]; + self.decoder = Some(d); + } + Err(_) => return Err(PunktfunkStatus::Unsupported), + } + } + let dec = self.decoder.as_mut().unwrap(); + + // Conceal lost packets (a seq gap) 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 in the ring. Mirrors the Linux/Windows session + // pump and the Android native pump; capped by the tracker at 50 ms. + let missing = self.gaps.missing_before(seq); + let mut filled = 0usize; + if self.frame_samples > 0 { + for _ in 0..missing { + let plc = self.frame_samples * ch; + match dec.decode_float(&[], &mut self.pcm[filled..filled + plc], false) { + Ok(samples) => filled += samples * ch, + Err(_) => break, + } + } + } + + if data.is_empty() { + // DTX silence marker (a legal wire form) — never decoded (see above); the sink + // underruns to silence on its own. Concealment owed for losses before it still + // goes out. + return Ok(filled); + } + match dec.decode_float(data, &mut self.pcm[filled..], false) { + Ok(samples) => { + self.frame_samples = samples; + Ok(filled + samples * ch) + } + // An undecodable packet: hand out whatever concealment the gap before it earned + // rather than dropping it with the packet. Its own 5 ms slot plays as a ring gap, + // as on every other client (the tracker has already anchored at this seq). + Err(_) if filled > 0 => Ok(filled), + Err(_) => Err(PunktfunkStatus::BadPacket), + } + } } /// `PunktfunkHidOutput::kind` — lightbar RGB (`r`/`g`/`b` valid). @@ -2340,6 +2418,13 @@ pub struct PunktfunkAudioPcm { /// [`punktfunk_connection_next_audio`] on a given connection, from one dedicated audio thread — /// not both (they share the underlying queue). /// +/// **Loss concealment**: packets the wire lost (a gap in the sequence, after the redundant-plane +/// recovery has had its chance) are synthesized via libopus packet-loss concealment and returned +/// IN FRONT of the arriving frame in the same buffer — `out->frame_count` then covers the +/// concealed frames plus the real one (`out->seq`/`out->pts_ns` are the real packet's). The +/// embedder just writes the whole buffer to its ring, same as any other frame; gaps arrive +/// pre-healed, exactly as they do on the clients that decode outside core. +/// /// # Safety /// `c` is a valid connection handle; `out` is writable. At most one thread pulls audio. #[cfg(feature = "quic")] @@ -2369,36 +2454,16 @@ pub unsafe extern "C" fn punktfunk_connection_next_audio_pcm( Err(e) => return e.status(), }; let mut state = c.audio_pcm.lock().unwrap(); - if state.decoder.is_none() { - let layout = crate::audio::layout_for(channels, false); - match opus::MSDecoder::new(48_000, layout.streams, layout.coupled, layout.mapping) { - Ok(d) => { - // Largest legal Opus frame is 120 ms = 5760 samples/ch. - state.pcm = vec![0f32; 5760 * channels as usize]; - state.decoder = Some(d); - } - Err(_) => return PunktfunkStatus::Unsupported, - } - } - let AudioPcmState { decoder, pcm } = &mut *state; - let dec = decoder.as_mut().unwrap(); - // A header-only datagram (DTX silence — a legal wire form) must be SKIPPED, not - // decoded: `decode_float` treats an empty payload as a loss and synthesizes a full - // 120 ms of concealment for a ~5 ms slot, growing the playout ring without bound. - // Mirrors the host mic pump's guard; the sink underruns to silence on its own. - if pkt.data.is_empty() { - return PunktfunkStatus::NoFrame; - } - // `decode_float` divides the output buffer length by the channel count to get the - // per-channel capacity; an empty payload requests packet-loss concealment. - match dec.decode_float(&pkt.data, pcm, false) { - Ok(frame_count) => { + match state.decode_packet(&pkt.data, pkt.seq, channels) { + // Nothing to hand out this call: a DTX silence marker with no loss owed before it. + Ok(0) => PunktfunkStatus::NoFrame, + Ok(samples) => { // SAFETY: per the ABI contract - `out` is a caller-owned writable slot of the // matching `#[repr(C)]` type, written once by value. unsafe { *out = PunktfunkAudioPcm { - samples: pcm.as_ptr(), - frame_count: frame_count as u32, + samples: state.pcm.as_ptr(), + frame_count: (samples / channels.max(1) as usize) as u32, channels, seq: pkt.seq, pts_ns: pkt.pts_ns, @@ -2406,7 +2471,7 @@ pub unsafe extern "C" fn punktfunk_connection_next_audio_pcm( } PunktfunkStatus::Ok } - Err(_) => PunktfunkStatus::BadPacket, + Err(status) => status, } }) } @@ -4656,4 +4721,53 @@ mod tests { .is_none() ); } + + /// The in-core PCM decoder heals seq gaps with concealment, exactly like the decode loops + /// the other clients run themselves: a lost packet's worth of PLC lands in front of the + /// arriving frame, DTX markers advance the accounting without being decoded, and a gap is + /// capped at the tracker's 50 ms. + #[test] + fn audio_pcm_decode_conceals_seq_gaps() { + const FRAME: usize = 240; // 5 ms @ 48 kHz, per channel + let l = crate::audio::LAYOUT_STEREO; + let mut enc = opus::MSEncoder::new( + 48_000, + l.streams, + l.coupled, + l.mapping, + opus::Application::LowDelay, + ) + .expect("MSEncoder"); + enc.set_vbr(false).unwrap(); + let mut packet = |tone: f32| { + let mut frame = vec![0f32; FRAME * 2]; + for (i, s) in frame.iter_mut().enumerate() { + *s = 0.25 * (i as f32 * tone).sin(); + } + let mut out = vec![0u8; 1500]; + let n = enc.encode_float(&frame, &mut out).unwrap(); + out.truncate(n); + out + }; + + let mut state = AudioPcmState::default(); + // In-order packets decode to exactly one frame each. + assert_eq!(state.decode_packet(&packet(0.05), 0, 2), Ok(FRAME * 2)); + assert_eq!(state.decode_packet(&packet(0.05), 1, 2), Ok(FRAME * 2)); + // Seq 2 lost: one concealed frame precedes the real one, contiguously. + assert_eq!(state.decode_packet(&packet(0.06), 3, 2), Ok(2 * FRAME * 2)); + // A duplicate conceals nothing. + assert_eq!(state.decode_packet(&packet(0.06), 3, 2), Ok(FRAME * 2)); + // DTX marker, nothing lost before it: nothing to emit (the ABI maps 0 to NoFrame)... + assert_eq!(state.decode_packet(&[], 4, 2), Ok(0)); + // ...but a DTX marker AFTER a loss still flushes the concealment owed (seq 5 lost). + assert_eq!(state.decode_packet(&[], 6, 2), Ok(FRAME * 2)); + // And the DTX slot itself was accounted, not treated as a loss. + assert_eq!(state.decode_packet(&packet(0.07), 7, 2), Ok(FRAME * 2)); + // A huge gap is capped at MAX_CONCEAL_PACKETS of concealment. + assert_eq!( + state.decode_packet(&packet(0.07), 1000, 2), + Ok((crate::audio::MAX_CONCEAL_PACKETS as usize + 1) * FRAME * 2) + ); + } } diff --git a/crates/punktfunk-core/src/audio.rs b/crates/punktfunk-core/src/audio.rs index 4ebaba8a..b38bdbb1 100644 --- a/crates/punktfunk-core/src/audio.rs +++ b/crates/punktfunk-core/src/audio.rs @@ -301,8 +301,9 @@ pub struct AudioGapTracker { /// Most packets a single gap will ask concealment for (50 ms at the protocol's 5 ms frames). /// Crate-internal: callers only ever see `missing_before`'s already-capped count (and cbindgen -/// must not export it — it's not part of the C ABI). -const MAX_CONCEAL_PACKETS: u32 = 10; +/// must not export it — it's not part of the C ABI). `pub(crate)` for the in-core PCM decoder +/// (`abi.rs`), which sizes its no-realloc output buffer from it. +pub(crate) const MAX_CONCEAL_PACKETS: u32 = 10; impl AudioGapTracker { pub fn new() -> Self { diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index 9bf9d924..aa19227d 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -2594,6 +2594,13 @@ PunktfunkStatus punktfunk_connection_end_reason(PunktfunkConnection *c, uint8_t // [`punktfunk_connection_next_audio`] on a given connection, from one dedicated audio thread — // not both (they share the underlying queue). // +// **Loss concealment**: packets the wire lost (a gap in the sequence, after the redundant-plane +// recovery has had its chance) are synthesized via libopus packet-loss concealment and returned +// IN FRONT of the arriving frame in the same buffer — `out->frame_count` then covers the +// concealed frames plus the real one (`out->seq`/`out->pts_ns` are the real packet's). The +// embedder just writes the whole buffer to its ring, same as any other frame; gaps arrive +// pre-healed, exactly as they do on the clients that decode outside core. +// // # Safety // `c` is a valid connection handle; `out` is writable. At most one thread pulls audio. PunktfunkStatus punktfunk_connection_next_audio_pcm(PunktfunkConnection *c, diff --git a/scripts/build-xcframework.sh b/scripts/build-xcframework.sh index eb7bbedb..cb5f5ba8 100755 --- a/scripts/build-xcframework.sh +++ b/scripts/build-xcframework.sh @@ -53,6 +53,15 @@ if [[ -z "${DEVELOPER_DIR:-}" ]]; then esac # a non-beta xcode-select default is fine as-is fi +# Hermetic Opus: never let audiopus_sys link a Homebrew libopus via pkg-config. A brew lib +# is built for the RUNNING macOS (its objects carry that minos, tripping the version guard +# below) and only exists for the host arch — the other slice silently falls back to the +# vendored build, so the two slices ship different libopus builds. Force the vendored CMake +# build everywhere; the policy floor keeps modern CMake (≥4) accepting libopus's old +# `cmake_minimum_required`. +export OPUS_NO_PKG_CONFIG=1 +export CMAKE_POLICY_VERSION_MINIMUM=3.5 + # Deployment targets must match Package.swift's platforms, or every consumer link emits # "object file was built for newer macOS version" warnings. for t in "${TARGETS_MAC[@]}"; do