fix(client/apple): audio stops crackling on lossy, bunching Wi-Fi #82

Merged
enricobuehler merged 3 commits from worktree-audio-wifi-distortion into main 2026-08-07 07:37:26 +00:00
6 changed files with 338 additions and 40 deletions
@@ -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<Float>, 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..<count { out[i] = 0 }
// De-prime only after a RUN of short reads: a single transient drain must not
// manufacture a whole target's worth of fresh silence.
}
noteRead(ranShort: n < count, count: count)
}
/// The outcome accounting of one primed read the Swift mirror of
/// `JitterPolicy::note_read`. A short read drives both the de-prime hysteresis (a single
/// transient drain must not manufacture a whole target's worth of fresh silence) and the
/// adaptive target floor: a device that genuinely keeps starving gets more slack, one step
/// per window, capped and gives it back after a long quiet spell, so one bad minute
/// doesn't cost latency for the rest of the session. Caller holds the lock.
private func noteRead(ranShort: Bool, count: Int) {
windowRun += count
if windowRun >= 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)
}
}
}
@@ -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..<steps {
pending += 5 // the host encodes 5 ms per 5 ms of wall clock, stall or not
// Delivery bunches into ~60 ms bursts; every 4th burst arrives a further 30 ms late.
if step % 12 == 11 {
if burst % 4 == 3 {
// Hold this burst 30 ms: it is flushed 6 callbacks later instead.
burst += 1
} else {
feed.withUnsafeBufferPointer {
ring.write($0.baseAddress!, count: pending * perMS)
}
pending = 0
burst += 1
}
} else if step % 12 == 5, pending >= 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
+143 -29
View File
@@ -654,10 +654,88 @@ pub struct PunktfunkConnection {
#[derive(Default)]
struct AudioPcmState {
decoder: Option<opus::MSDecoder>,
/// 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<f32>,
/// 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<usize, PunktfunkStatus> {
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)
);
}
}
+3 -2
View File
@@ -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 {
+7
View File
@@ -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,
+9
View File
@@ -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