feat(host/audio): the virtual mic buffers for the client it has, not the worst one ever
The mic pump grows a real de-jitter (audio/mic_jitter.rs): a two-frame reorder window in front of the decoder, libopus concealment on sequence gaps (up to 5 frames — a lost datagram no longer drains the ring into a silence + re-prime crackle), and an adaptive target depth measured from inter-arrival jitter, clamped to 10–60 ms. Both backend rings now prime at one consumer quantum + that target: the old bursty Mac client still measures ~42 ms and lands where the fixed 48 ms prime protected it, a modern 10 ms-cadence client settles at ~25–35 ms, and a 2048-frame recorder on Linux stops buying 128 ms of latency from the 3-quanta clamp. Depth stuck above target sheds near-silent frames a few ms per 100 ms — never speech, never a hard clear. PUNKTFUNK_MIC_LEGACY_BUFFER=1 (documented) is the one-release escape hatch back to the fixed constants, and a "mic uplink health" line every 30 s (depth/target, cadence, gaps, conceals, reorders, drops, re-primes) finally says which side of the link a bad mic lives on. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Claude Fable 5
@@ -115,6 +115,47 @@ pub trait VirtualMic: Send {
|
||||
fn channels(&self) -> u32 {
|
||||
CHANNELS as u32
|
||||
}
|
||||
|
||||
/// The adaptive de-jitter target (per-channel samples) the pump measured from uplink
|
||||
/// arrival jitter (see `mic_jitter`). A backend with a jitter ring primes around this PLUS
|
||||
/// one of its own consumer quanta: the ring must absorb arrival burstiness (the pump's
|
||||
/// number) and pull granularity (the backend's own), and neither may buy the other's depth
|
||||
/// — a 2048-frame recorder gets its one quantum, not three. Never called ⇒ the backend
|
||||
/// keeps its legacy fixed constants, which is also how `PUNKTFUNK_MIC_LEGACY_BUFFER=1`
|
||||
/// works (the pump simply never drives the target). Default: no ring, ignored.
|
||||
fn set_target_depth(&self, _samples_per_ch: usize) {}
|
||||
|
||||
/// `(buffered, prime_target)` of the backend's jitter ring in per-channel samples — read
|
||||
/// by the pump's creep trim + telemetry. `None` while unknown (consumer not yet running,
|
||||
/// or a backend without a ring).
|
||||
fn depth(&self) -> Option<(usize, usize)> {
|
||||
None
|
||||
}
|
||||
|
||||
/// Reset-on-read telemetry counters (see [`MicBackendStats`]). Default: all zero.
|
||||
fn take_stats(&self) -> MicBackendStats {
|
||||
MicBackendStats::default()
|
||||
}
|
||||
}
|
||||
|
||||
/// Reset-on-read counters a [`VirtualMic`] backend reports into the pump's periodic mic
|
||||
/// telemetry line ("mic uplink health").
|
||||
#[derive(Debug, Default, Clone, Copy)]
|
||||
pub struct MicBackendStats {
|
||||
/// Full-drain re-prime arms: the ring emptied and gates on silence until the target depth
|
||||
/// rebuilds. One per talk spurt is normal; several per second mid-speech is the crackle.
|
||||
pub reprimes: u64,
|
||||
/// Per-channel samples dropped by the ring's overflow cap (drop-oldest).
|
||||
pub overflow_dropped: u64,
|
||||
}
|
||||
|
||||
/// One-release escape hatch (docs: configuration → Audio / microphone):
|
||||
/// `PUNKTFUNK_MIC_LEGACY_BUFFER=1` keeps the pre-adaptive fixed mic buffering — the pump never
|
||||
/// drives the backend target (so the rings stay on their legacy constants: 48 ms prime /
|
||||
/// 120 ms cap on Windows, the 3-quanta clamp on Linux) and never creep-trims depth.
|
||||
pub(crate) fn mic_legacy_buffer() -> bool {
|
||||
static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
|
||||
*ON.get_or_init(|| std::env::var_os("PUNKTFUNK_MIC_LEGACY_BUFFER").is_some_and(|v| v != "0"))
|
||||
}
|
||||
|
||||
/// Open a virtual microphone with `channels` interleaved channels (1 or 2). Linux: a PipeWire
|
||||
@@ -152,5 +193,6 @@ mod wasapi_mic;
|
||||
#[path = "audio/wiring_plan.rs"]
|
||||
pub(crate) mod wiring_plan;
|
||||
|
||||
mod mic_jitter;
|
||||
mod mic_pump;
|
||||
pub use mic_pump::{MicFrame, MicPump};
|
||||
|
||||
@@ -29,10 +29,10 @@
|
||||
|
||||
mod stream_sink;
|
||||
|
||||
use super::{AudioCapturer, VirtualMic, SAMPLE_RATE};
|
||||
use super::{AudioCapturer, MicBackendStats, VirtualMic, SAMPLE_RATE};
|
||||
use anyhow::{anyhow, Context, Result};
|
||||
use std::collections::VecDeque;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
|
||||
use std::sync::mpsc::{sync_channel, Receiver, RecvTimeoutError};
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
@@ -218,6 +218,26 @@ pub struct PwMicSource {
|
||||
alive: Arc<AtomicBool>,
|
||||
/// One-shot flush request, consumed by the process callback (clears the jitter ring).
|
||||
flush: Arc<AtomicBool>,
|
||||
/// Ring policy/telemetry shared with the RT process callback (see [`MicRingShared`]).
|
||||
ring: Arc<MicRingShared>,
|
||||
}
|
||||
|
||||
/// Atomics shared between [`PwMicSource`] (the pump's side) and the RT process callback: the
|
||||
/// pump's adaptive de-jitter target in, ring depth/prime gauges + reset-on-read counters out.
|
||||
/// All `Relaxed` — a slowly-moving target and telemetry, not synchronization.
|
||||
#[derive(Default)]
|
||||
struct MicRingShared {
|
||||
/// Pump-set jitter target (per-channel samples). `0` = the pump never spoke (legacy mode,
|
||||
/// or its first estimate hasn't landed) → the callback keeps the historical 3-quanta clamp.
|
||||
target: AtomicUsize,
|
||||
/// Ring depth after the last callback (per-channel samples).
|
||||
depth: AtomicUsize,
|
||||
/// Effective prime target of the last callback (per-channel samples).
|
||||
prime: AtomicUsize,
|
||||
/// Full-drain re-prime arms (see [`MicBackendStats`]).
|
||||
reprimes: AtomicU64,
|
||||
/// Per-channel samples dropped by the overflow cap.
|
||||
overflow: AtomicU64,
|
||||
}
|
||||
|
||||
impl PwMicSource {
|
||||
@@ -234,11 +254,13 @@ impl PwMicSource {
|
||||
// service started before the user session) must surface as an open ERROR — engaging the
|
||||
// pump's backoff — not as an instantly-dead instance the pump would churn on.
|
||||
let (ready_tx, ready_rx) = sync_channel::<Result<()>>(1);
|
||||
let (alive_t, flush_t) = (alive.clone(), flush.clone());
|
||||
let ring = Arc::new(MicRingShared::default());
|
||||
let (alive_t, flush_t, ring_t) = (alive.clone(), flush.clone(), ring.clone());
|
||||
thread::Builder::new()
|
||||
.name("punktfunk-pw-mic".into())
|
||||
.spawn(move || {
|
||||
if let Err(e) = mic_pw_thread(pcm_rx, quit_rx, channels, flush_t, ready_tx) {
|
||||
if let Err(e) = mic_pw_thread(pcm_rx, quit_rx, channels, flush_t, ring_t, ready_tx)
|
||||
{
|
||||
// Reaching here is always a setup/open failure (once the mainloop runs it exits
|
||||
// Ok) — and it was already reported to the pump via the ready handshake, which
|
||||
// owns the throttled operator-facing warn. Keep only a debug breadcrumb.
|
||||
@@ -255,6 +277,7 @@ impl PwMicSource {
|
||||
quit: quit_tx,
|
||||
alive,
|
||||
flush,
|
||||
ring,
|
||||
}),
|
||||
Ok(Err(e)) => Err(e),
|
||||
Err(_) => Err(anyhow!("pipewire virtual-mic init timed out")),
|
||||
@@ -291,6 +314,20 @@ impl VirtualMic for PwMicSource {
|
||||
fn channels(&self) -> u32 {
|
||||
self.channels
|
||||
}
|
||||
fn set_target_depth(&self, samples_per_ch: usize) {
|
||||
self.ring.target.store(samples_per_ch, Ordering::Relaxed);
|
||||
}
|
||||
fn depth(&self) -> Option<(usize, usize)> {
|
||||
let prime = self.ring.prime.load(Ordering::Relaxed);
|
||||
// 0 = the process callback hasn't run yet (no consumer) — nothing meaningful to report.
|
||||
(prime > 0).then(|| (self.ring.depth.load(Ordering::Relaxed), prime))
|
||||
}
|
||||
fn take_stats(&self) -> MicBackendStats {
|
||||
MicBackendStats {
|
||||
reprimes: self.ring.reprimes.swap(0, Ordering::Relaxed),
|
||||
overflow_dropped: self.ring.overflow.swap(0, Ordering::Relaxed),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Producer-side state for the virtual-mic loop: incoming decoded PCM and a small ring buffer
|
||||
@@ -303,6 +340,8 @@ struct MicUserData {
|
||||
primed: bool,
|
||||
/// One-shot flush request from [`PwMicSource::discard`] (stale-audio drop after a gap).
|
||||
flush: Arc<AtomicBool>,
|
||||
/// Pump-driven ring policy + telemetry (see [`MicRingShared`]).
|
||||
shared: Arc<MicRingShared>,
|
||||
/// When the process callback last ran — a long gap means the ring content predates the
|
||||
/// current consumer (the stream idles with no recorder attached) and must be dropped.
|
||||
last_run: Option<std::time::Instant>,
|
||||
@@ -318,6 +357,7 @@ fn mic_pw_thread(
|
||||
quit_rx: pipewire::channel::Receiver<Terminate>,
|
||||
channels: u32,
|
||||
flush: Arc<AtomicBool>,
|
||||
shared: Arc<MicRingShared>,
|
||||
ready: std::sync::mpsc::SyncSender<Result<()>>,
|
||||
) -> Result<()> {
|
||||
use pipewire as pw;
|
||||
@@ -400,6 +440,7 @@ fn mic_pw_thread(
|
||||
channels: channels as usize,
|
||||
primed: false,
|
||||
flush,
|
||||
shared,
|
||||
last_run: None,
|
||||
};
|
||||
|
||||
@@ -473,15 +514,31 @@ fn mic_pw_thread(
|
||||
);
|
||||
}
|
||||
|
||||
// Adaptive jitter buffer. The client pushes 5 ms frames; the recorder pulls a
|
||||
// whole *quantum* (often 20–43 ms) from an independent clock. A drain of one
|
||||
// quantum must not outrun what's buffered, or every call underruns to silence
|
||||
// (the original ~58% gaps). So prime to ~3 quanta before producing, hold there,
|
||||
// and re-prime only after a genuine full drain (the client went quiet). The ring
|
||||
// is capped at a few quanta so latency stays bounded.
|
||||
let target = (3 * want).clamp(720 * ud.channels, 9600 * ud.channels);
|
||||
// Jitter buffer. The client pushes frames on its own clock; the recorder
|
||||
// pulls a whole *quantum* (often 20–43 ms) from an independent one. A drain
|
||||
// of one quantum must not outrun what's buffered, or every call underruns
|
||||
// to silence (the original ~58% gaps). Prime target = one quantum (the pull
|
||||
// granularity) + the pump's measured-jitter target (arrival burstiness —
|
||||
// see `mic_jitter`), re-priming only after a genuine full drain (the client
|
||||
// went quiet). Until the pump's first estimate lands — and forever under
|
||||
// PUNKTFUNK_MIC_LEGACY_BUFFER — the historical 3-quanta clamp applies; that
|
||||
// formula scaled with the RECORDING APP's quantum, so a 2048-frame recorder
|
||||
// bought 128+ ms of standing mic latency for jitter it never had.
|
||||
let pump_target = ud.shared.target.load(Ordering::Relaxed) * ud.channels;
|
||||
let target = if pump_target == 0 {
|
||||
(3 * want).clamp(720 * ud.channels, 9600 * ud.channels)
|
||||
} else {
|
||||
want + pump_target
|
||||
};
|
||||
let mut dropped = 0usize;
|
||||
while ud.ring.len() > target.max(want) + want {
|
||||
ud.ring.pop_front(); // bound latency: drop the oldest beyond ~1 quantum slack
|
||||
dropped += 1;
|
||||
}
|
||||
if dropped > 0 {
|
||||
ud.shared
|
||||
.overflow
|
||||
.fetch_add((dropped / ud.channels) as u64, Ordering::Relaxed);
|
||||
}
|
||||
if !ud.primed && ud.ring.len() >= target {
|
||||
ud.primed = true;
|
||||
@@ -501,9 +558,17 @@ fn mic_pw_thread(
|
||||
} else {
|
||||
0
|
||||
};
|
||||
if ud.ring.is_empty() {
|
||||
if ud.primed && ud.ring.is_empty() {
|
||||
ud.primed = false; // fully drained — re-prime before producing again
|
||||
ud.shared.reprimes.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
// Publish depth/prime for the pump's creep trim + telemetry.
|
||||
ud.shared
|
||||
.depth
|
||||
.store(ud.ring.len() / ud.channels, Ordering::Relaxed);
|
||||
ud.shared
|
||||
.prime
|
||||
.store(target / ud.channels, Ordering::Relaxed);
|
||||
let chunk = data.chunk_mut();
|
||||
*chunk.offset_mut() = 0;
|
||||
*chunk.stride_mut() = stride as _;
|
||||
|
||||
@@ -0,0 +1,471 @@
|
||||
//! Backend-agnostic de-jitter for the client-mic uplink. The pump feeds every received `0xCB`
|
||||
//! frame (with its datagram sequence — see [`MicFrame`]) through [`MicDejitter`], which puts a
|
||||
//! two-frame reorder window in front of the Opus decoder, asks for libopus concealment on true
|
||||
//! sequence gaps (so a lost datagram doesn't drain the backend ring into a silence + re-prime
|
||||
//! cycle), and measures inter-arrival jitter into the adaptive target depth the backend rings
|
||||
//! prime against. Mirrors the client downlink's `AudioGapTracker` (punktfunk-core `audio.rs`):
|
||||
//! the same wrapping-sequence gap rules, grown the reorder hold and the jitter estimator the
|
||||
//! uplink needs — the downlink's jitter rings own their depth; the uplink's live in the
|
||||
//! backends and take their target from here.
|
||||
|
||||
use super::MicFrame;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
/// What the pump should do next, in playback order. `Frame` is a real frame to decode;
|
||||
/// `Conceal` is one frame of libopus packet-loss concealment (`decode_float(&[], …)`, sized to
|
||||
/// the last real frame) standing in for a lost one.
|
||||
pub(super) enum Deliver {
|
||||
Frame(MicFrame),
|
||||
Conceal,
|
||||
}
|
||||
|
||||
/// Most concealment frames one gap may synthesize (~100 ms at the common 20 ms frames) — the
|
||||
/// client downlink's `AudioGapTracker` cap, scaled to the uplink's bigger frames. libopus PLC
|
||||
/// fades to silence after a few frames anyway; past the cap the ring's underrun/re-prime path
|
||||
/// takes over, as before.
|
||||
const MAX_CONCEAL_FRAMES: u32 = 5;
|
||||
|
||||
/// How long one out-of-order frame may wait for its missing predecessor before the gap is
|
||||
/// concealed and the held frame plays: ~one 20 ms frame interval plus scheduling slack. With
|
||||
/// the incoming frame that is a two-frame window — anything needing more is treated as loss,
|
||||
/// exactly like the client downlink (which holds nothing at all).
|
||||
const HOLD_MAX: Duration = Duration::from_millis(30);
|
||||
|
||||
/// Arrival gaps above this are talk-spurt pauses (DTX, mute), not network jitter — folding
|
||||
/// them into the estimate would balloon the target after every pause.
|
||||
const PAUSE_GAP: Duration = Duration::from_millis(250);
|
||||
|
||||
/// The sliding jitter window: [`BUCKETS`] × [`BUCKET_LEN`] of per-bucket max inter-arrival gap.
|
||||
/// Long enough that the old bursty Mac cadence (~two 20 ms frames every ~42 ms — the 2026-07-03
|
||||
/// "crackling mic", see `wasapi_mic.rs`) is always represented; short enough that a one-off
|
||||
/// spike ages out in a few seconds.
|
||||
const BUCKETS: usize = 4;
|
||||
const BUCKET_LEN: Duration = Duration::from_millis(1000);
|
||||
|
||||
/// Target before the estimator has a full [`WARMUP`] of evidence: roughly the old fixed 48 ms
|
||||
/// prime minus the one consumer quantum the backends add — the crackle-safe assumption that
|
||||
/// the client might be an old bursty one. Modern 10 ms-cadence clients drop to ~15–25 ms once
|
||||
/// measured (after their first talk pause re-primes the ring at the lower target).
|
||||
const DEFAULT_TARGET_MS: u32 = 40;
|
||||
/// Headroom over the measured worst burst gap.
|
||||
const TARGET_PAD_MS: u32 = 5;
|
||||
/// How long the estimator distrusts its own (young) window and keeps the default as a floor.
|
||||
const WARMUP: Duration = Duration::from_secs(1);
|
||||
|
||||
/// Reset-on-read snapshot for the pump's telemetry line. Counters cover the window since the
|
||||
/// last read; `cadence_ms`/`frame_ms` are gauges (EWMAs).
|
||||
#[derive(Debug, Default, Clone, Copy)]
|
||||
pub(super) struct DejitterStats {
|
||||
pub seq_gaps: u64,
|
||||
pub concealed: u64,
|
||||
pub reorders: u64,
|
||||
pub late_drops: u64,
|
||||
/// EWMA inter-arrival gap — the uplink's burst cadence as seen by the pump.
|
||||
pub cadence_ms: f32,
|
||||
/// Nominal frame duration from consecutive frames' pts deltas (what the client encodes).
|
||||
pub frame_ms: f32,
|
||||
}
|
||||
|
||||
pub(super) struct MicDejitter {
|
||||
/// Sequence the uplink owes next; `None` before the first frame / after
|
||||
/// [`reset_stream`](Self::reset_stream).
|
||||
expected: Option<u32>,
|
||||
/// The reorder window: at most ONE newer frame parked ≤ [`HOLD_MAX`] for its predecessor.
|
||||
held: Option<(Instant, MicFrame)>,
|
||||
// --- adaptive target (inter-arrival jitter) ---
|
||||
first_arrival: Option<Instant>,
|
||||
last_arrival: Option<Instant>,
|
||||
bucket_start: Option<Instant>,
|
||||
bucket_idx: usize,
|
||||
gap_max: [f32; BUCKETS],
|
||||
ewma_gap_ms: f32,
|
||||
// --- pts bookkeeping ---
|
||||
/// `(seq, pts_ns)` of the newest ingested frame — a pts delta over exactly one seq step is
|
||||
/// the client's true frame duration.
|
||||
last_pts: Option<(u32, u64)>,
|
||||
frame_ms: f32,
|
||||
// --- counters (reset on read) ---
|
||||
seq_gaps: u64,
|
||||
concealed: u64,
|
||||
reorders: u64,
|
||||
late_drops: u64,
|
||||
}
|
||||
|
||||
impl MicDejitter {
|
||||
pub(super) fn new() -> Self {
|
||||
MicDejitter {
|
||||
expected: None,
|
||||
held: None,
|
||||
first_arrival: None,
|
||||
last_arrival: None,
|
||||
bucket_start: None,
|
||||
bucket_idx: 0,
|
||||
gap_max: [0.0; BUCKETS],
|
||||
ewma_gap_ms: 0.0,
|
||||
last_pts: None,
|
||||
frame_ms: 0.0,
|
||||
seq_gaps: 0,
|
||||
concealed: 0,
|
||||
reorders: 0,
|
||||
late_drops: 0,
|
||||
}
|
||||
}
|
||||
|
||||
/// Feed one received frame; playback-ordered work is appended to `out`.
|
||||
pub(super) fn ingest(&mut self, now: Instant, frame: MicFrame, out: &mut Vec<Deliver>) {
|
||||
self.record_arrival(now, frame.seq, frame.pts_ns);
|
||||
self.accept(now, frame, out);
|
||||
}
|
||||
|
||||
fn accept(&mut self, now: Instant, frame: MicFrame, out: &mut Vec<Deliver>) {
|
||||
let Some(expected) = self.expected else {
|
||||
// First frame (or first after a reset): the chain starts here.
|
||||
self.expected = Some(frame.seq.wrapping_add(1));
|
||||
out.push(Deliver::Frame(frame));
|
||||
return;
|
||||
};
|
||||
let delta = frame.seq.wrapping_sub(expected);
|
||||
if delta == 0 {
|
||||
self.expected = Some(expected.wrapping_add(1));
|
||||
out.push(Deliver::Frame(frame));
|
||||
self.release_held_in_order(out);
|
||||
} else if delta > u32::MAX / 2 {
|
||||
// Duplicate of, or older than, something already played — the window moved on.
|
||||
self.late_drops += 1;
|
||||
} else if let Some(held_seq) = self.held.as_ref().map(|(_, h)| h.seq) {
|
||||
if held_seq == frame.seq {
|
||||
self.late_drops += 1; // duplicate of the held frame
|
||||
return;
|
||||
}
|
||||
// Two newer-than-expected frames in hand — waiting longer can't fill the gap
|
||||
// within the window. Play them in order, concealing what's genuinely missing.
|
||||
let (_, held) = self.held.take().expect("checked above");
|
||||
let (first, second) = if frame.seq.wrapping_sub(held.seq) > u32::MAX / 2 {
|
||||
(frame, held)
|
||||
} else {
|
||||
(held, frame)
|
||||
};
|
||||
self.give_up_and_play(first, out);
|
||||
self.accept(now, second, out); // depth ≤ 1: `held` is None here
|
||||
} else {
|
||||
self.held = Some((now, frame)); // wait ≤ HOLD_MAX for the predecessor
|
||||
}
|
||||
}
|
||||
|
||||
/// If the held frame's predecessor just played, the hold is over — the reorder repaired.
|
||||
fn release_held_in_order(&mut self, out: &mut Vec<Deliver>) {
|
||||
if let Some((_, h)) = &self.held {
|
||||
if Some(h.seq) == self.expected {
|
||||
let (_, h) = self.held.take().expect("checked above");
|
||||
self.expected = Some(h.seq.wrapping_add(1));
|
||||
self.reorders += 1;
|
||||
out.push(Deliver::Frame(h));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Stop waiting for `frame`'s predecessors: conceal the gap (capped) and play it.
|
||||
fn give_up_and_play(&mut self, frame: MicFrame, out: &mut Vec<Deliver>) {
|
||||
let gap = frame
|
||||
.seq
|
||||
.wrapping_sub(self.expected.expect("callers checked"));
|
||||
if gap > 0 {
|
||||
self.seq_gaps += 1;
|
||||
let conceal = gap.min(MAX_CONCEAL_FRAMES);
|
||||
self.concealed += u64::from(conceal);
|
||||
for _ in 0..conceal {
|
||||
out.push(Deliver::Conceal);
|
||||
}
|
||||
}
|
||||
self.expected = Some(frame.seq.wrapping_add(1));
|
||||
out.push(Deliver::Frame(frame));
|
||||
}
|
||||
|
||||
/// When the current hold (if any) must be given up — the pump's next wake deadline.
|
||||
pub(super) fn hold_deadline(&self) -> Option<Instant> {
|
||||
self.held.as_ref().map(|(t, _)| *t + HOLD_MAX)
|
||||
}
|
||||
|
||||
/// Give up an expired hold (the predecessor never came): conceal + play. Called on the
|
||||
/// pump's timeout wakes.
|
||||
pub(super) fn flush_expired_hold(&mut self, now: Instant, out: &mut Vec<Deliver>) {
|
||||
if self.hold_deadline().is_some_and(|d| now >= d) {
|
||||
let (_, h) = self.held.take().expect("deadline implies held");
|
||||
self.give_up_and_play(h, out);
|
||||
}
|
||||
}
|
||||
|
||||
/// Uplink pause / backend reopen: whatever is parked predates the gap, and the next frame
|
||||
/// restarts the chain (a pause is not loss — it must not conceal or count a gap).
|
||||
pub(super) fn reset_stream(&mut self) {
|
||||
self.expected = None;
|
||||
self.held = None;
|
||||
self.last_pts = None;
|
||||
}
|
||||
|
||||
// ---- adaptive target -------------------------------------------------------------------
|
||||
|
||||
fn record_arrival(&mut self, now: Instant, seq: u32, pts_ns: u64) {
|
||||
if let Some(last) = self.last_arrival {
|
||||
let gap = now.duration_since(last);
|
||||
if gap <= PAUSE_GAP {
|
||||
let ms = gap.as_secs_f32() * 1000.0;
|
||||
self.ewma_gap_ms = if self.ewma_gap_ms == 0.0 {
|
||||
ms
|
||||
} else {
|
||||
self.ewma_gap_ms * 0.9 + ms * 0.1
|
||||
};
|
||||
self.rotate_buckets(now);
|
||||
self.gap_max[self.bucket_idx] = self.gap_max[self.bucket_idx].max(ms);
|
||||
}
|
||||
}
|
||||
self.first_arrival.get_or_insert(now);
|
||||
self.last_arrival = Some(now);
|
||||
if let Some((last_seq, last_pts)) = self.last_pts {
|
||||
if seq == last_seq.wrapping_add(1) && pts_ns > last_pts {
|
||||
let d = (pts_ns - last_pts) as f32 / 1_000_000.0;
|
||||
if (1.0..=120.0).contains(&d) {
|
||||
self.frame_ms = if self.frame_ms == 0.0 {
|
||||
d
|
||||
} else {
|
||||
self.frame_ms * 0.9 + d * 0.1
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
self.last_pts = Some((seq, pts_ns));
|
||||
}
|
||||
|
||||
fn rotate_buckets(&mut self, now: Instant) {
|
||||
let Some(mut start) = self.bucket_start else {
|
||||
self.bucket_start = Some(now);
|
||||
return;
|
||||
};
|
||||
let mut advanced = 0;
|
||||
while now.duration_since(start) >= BUCKET_LEN {
|
||||
self.bucket_idx = (self.bucket_idx + 1) % BUCKETS;
|
||||
self.gap_max[self.bucket_idx] = 0.0;
|
||||
start += BUCKET_LEN;
|
||||
advanced += 1;
|
||||
if advanced >= BUCKETS {
|
||||
// Everything aged out (long pause) — snap the window to now.
|
||||
self.gap_max = [0.0; BUCKETS];
|
||||
start = now;
|
||||
break;
|
||||
}
|
||||
}
|
||||
self.bucket_start = Some(start);
|
||||
}
|
||||
|
||||
/// The jitter component of the backend ring's prime depth, in ms — the worst burst gap in
|
||||
/// the window plus a pad, clamped to [10, 60]. Backends add ONE consumer quantum on top
|
||||
/// (see [`VirtualMic::set_target_depth`](super::VirtualMic::set_target_depth)). Old bursty
|
||||
/// clients measure ~42 ms and land near the historical 48 ms prime; modern 10 ms-cadence
|
||||
/// clients settle at ~15–25 ms.
|
||||
pub(super) fn target_ms(&self, now: Instant) -> u32 {
|
||||
let measured =
|
||||
self.gap_max.iter().fold(0f32, |a, &b| a.max(b)).ceil() as u32 + TARGET_PAD_MS;
|
||||
let warm = self
|
||||
.first_arrival
|
||||
.is_some_and(|t| now.duration_since(t) >= WARMUP);
|
||||
let t = if warm {
|
||||
measured
|
||||
} else {
|
||||
measured.max(DEFAULT_TARGET_MS)
|
||||
};
|
||||
t.clamp(10, 60)
|
||||
}
|
||||
|
||||
/// Reset-on-read counters + gauges for the pump's telemetry line.
|
||||
pub(super) fn take_stats(&mut self) -> DejitterStats {
|
||||
let s = DejitterStats {
|
||||
seq_gaps: self.seq_gaps,
|
||||
concealed: self.concealed,
|
||||
reorders: self.reorders,
|
||||
late_drops: self.late_drops,
|
||||
cadence_ms: self.ewma_gap_ms,
|
||||
frame_ms: self.frame_ms,
|
||||
};
|
||||
self.seq_gaps = 0;
|
||||
self.concealed = 0;
|
||||
self.reorders = 0;
|
||||
self.late_drops = 0;
|
||||
s
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn f(seq: u32) -> MicFrame {
|
||||
MicFrame {
|
||||
seq,
|
||||
pts_ns: u64::from(seq) * 20_000_000,
|
||||
opus: vec![1],
|
||||
}
|
||||
}
|
||||
|
||||
/// Compact delivery shape: frame seqs as-is, concealment as -1.
|
||||
fn seqs(out: &[Deliver]) -> Vec<i64> {
|
||||
out.iter()
|
||||
.map(|d| match d {
|
||||
Deliver::Frame(fr) => i64::from(fr.seq),
|
||||
Deliver::Conceal => -1,
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn in_order_passthrough() {
|
||||
let mut dj = MicDejitter::new();
|
||||
let t = Instant::now();
|
||||
let mut out = Vec::new();
|
||||
for s in 5..8 {
|
||||
dj.ingest(t, f(s), &mut out);
|
||||
}
|
||||
assert_eq!(seqs(&out), [5, 6, 7]);
|
||||
let st = dj.take_stats();
|
||||
assert_eq!(
|
||||
(st.seq_gaps, st.concealed, st.reorders, st.late_drops),
|
||||
(0, 0, 0, 0)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn swapped_pair_is_repaired_not_concealed() {
|
||||
let mut dj = MicDejitter::new();
|
||||
let t = Instant::now();
|
||||
let mut out = Vec::new();
|
||||
dj.ingest(t, f(5), &mut out);
|
||||
dj.ingest(t, f(7), &mut out); // held, waiting for 6
|
||||
assert_eq!(seqs(&out), [5]);
|
||||
dj.ingest(t, f(6), &mut out); // the missing one arrives late → both play, in order
|
||||
assert_eq!(seqs(&out), [5, 6, 7]);
|
||||
let st = dj.take_stats();
|
||||
assert_eq!(st.reorders, 1);
|
||||
assert_eq!(st.concealed, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn loss_is_concealed_when_the_window_moves_on() {
|
||||
let mut dj = MicDejitter::new();
|
||||
let t = Instant::now();
|
||||
let mut out = Vec::new();
|
||||
dj.ingest(t, f(5), &mut out);
|
||||
dj.ingest(t, f(7), &mut out); // held, waiting for 6
|
||||
dj.ingest(t, f(8), &mut out); // 6 is lost: one PLC frame, then 7 and 8
|
||||
assert_eq!(seqs(&out), [5, -1, 7, 8]);
|
||||
let st = dj.take_stats();
|
||||
assert_eq!(st.seq_gaps, 1);
|
||||
assert_eq!(st.concealed, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn expired_hold_flushes_with_concealment() {
|
||||
let mut dj = MicDejitter::new();
|
||||
let t = Instant::now();
|
||||
let mut out = Vec::new();
|
||||
dj.ingest(t, f(5), &mut out);
|
||||
dj.ingest(t, f(7), &mut out);
|
||||
assert!(dj.hold_deadline().is_some());
|
||||
dj.flush_expired_hold(t + HOLD_MAX, &mut out);
|
||||
assert_eq!(seqs(&out), [5, -1, 7]);
|
||||
assert!(dj.hold_deadline().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn big_gap_conceal_is_capped() {
|
||||
let mut dj = MicDejitter::new();
|
||||
let t = Instant::now();
|
||||
let mut out = Vec::new();
|
||||
dj.ingest(t, f(5), &mut out);
|
||||
dj.ingest(t, f(100), &mut out); // held
|
||||
dj.flush_expired_hold(t + HOLD_MAX, &mut out);
|
||||
let conceals = out.iter().filter(|d| matches!(d, Deliver::Conceal)).count();
|
||||
assert_eq!(conceals, MAX_CONCEAL_FRAMES as usize);
|
||||
assert_eq!(dj.take_stats().seq_gaps, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn duplicates_and_stale_late_frames_drop() {
|
||||
let mut dj = MicDejitter::new();
|
||||
let t = Instant::now();
|
||||
let mut out = Vec::new();
|
||||
dj.ingest(t, f(5), &mut out);
|
||||
dj.ingest(t, f(6), &mut out);
|
||||
dj.ingest(t, f(6), &mut out); // duplicate
|
||||
dj.ingest(t, f(3), &mut out); // ancient
|
||||
dj.ingest(t, f(8), &mut out); // held
|
||||
dj.ingest(t, f(8), &mut out); // duplicate of the held frame
|
||||
assert_eq!(seqs(&out), [5, 6]);
|
||||
assert_eq!(dj.take_stats().late_drops, 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn wraparound_is_a_reorder_not_an_eternity() {
|
||||
let mut dj = MicDejitter::new();
|
||||
let t = Instant::now();
|
||||
let mut out = Vec::new();
|
||||
dj.ingest(t, f(u32::MAX), &mut out);
|
||||
dj.ingest(t, f(0), &mut out); // in order across the wrap
|
||||
assert_eq!(seqs(&out), [i64::from(u32::MAX), 0]);
|
||||
dj.ingest(t, f(u32::MAX), &mut out); // pre-wrap replay: late, not a 2^31 gap
|
||||
assert_eq!(dj.take_stats().late_drops, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reset_restarts_the_chain_without_concealing() {
|
||||
let mut dj = MicDejitter::new();
|
||||
let t = Instant::now();
|
||||
let mut out = Vec::new();
|
||||
dj.ingest(t, f(5), &mut out);
|
||||
dj.reset_stream();
|
||||
dj.ingest(t, f(500), &mut out); // a pause is not loss
|
||||
assert_eq!(seqs(&out), [5, 500]);
|
||||
assert_eq!(dj.take_stats().concealed, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bursty_uplink_grows_the_target_modern_uplink_shrinks_it() {
|
||||
// Old Mac cadence: two frames back-to-back every ~42 ms.
|
||||
let mut dj = MicDejitter::new();
|
||||
let t0 = Instant::now();
|
||||
let mut out = Vec::new();
|
||||
let mut now = t0;
|
||||
let mut seq = 0u32;
|
||||
for burst in 0u64..50 {
|
||||
now = t0 + Duration::from_millis(burst * 42);
|
||||
for _ in 0..2 {
|
||||
dj.ingest(now, f(seq), &mut out);
|
||||
seq += 1;
|
||||
}
|
||||
}
|
||||
let target = dj.target_ms(now);
|
||||
assert!((42..=60).contains(&target), "bursty target {target}");
|
||||
|
||||
// Modern 10 ms cadence: past warmup the target follows the small gaps down.
|
||||
let mut dj = MicDejitter::new();
|
||||
let mut now = t0;
|
||||
for i in 0u64..200 {
|
||||
now = t0 + Duration::from_millis(i * 10);
|
||||
dj.ingest(now, f(i as u32), &mut out);
|
||||
}
|
||||
let target = dj.target_ms(now);
|
||||
assert!((10..=20).contains(&target), "steady target {target}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn warmup_holds_the_crackle_safe_default() {
|
||||
let mut dj = MicDejitter::new();
|
||||
let t0 = Instant::now();
|
||||
let mut out = Vec::new();
|
||||
dj.ingest(t0, f(0), &mut out);
|
||||
dj.ingest(t0 + Duration::from_millis(10), f(1), &mut out);
|
||||
// 10 ms of evidence must not shrink the target yet — the client might be bursty.
|
||||
assert_eq!(
|
||||
dj.target_ms(t0 + Duration::from_millis(20)),
|
||||
DEFAULT_TARGET_MS
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -5,6 +5,7 @@
|
||||
//! trait, its factory ([`open_virtual_mic`](super::open_virtual_mic)) and the audio-plane sample
|
||||
//! rate stay in `super`.
|
||||
|
||||
use super::mic_jitter::{Deliver, MicDejitter};
|
||||
use super::{VirtualMic, SAMPLE_RATE};
|
||||
use anyhow::Result;
|
||||
|
||||
@@ -34,6 +35,20 @@ const MIC_QUEUE_CAP: usize = 12;
|
||||
const DRAIN_ABOVE: usize = 6;
|
||||
const DRAIN_KEEP: usize = 4;
|
||||
|
||||
/// Cadence of the "mic uplink health" telemetry line (emitted only while frames flow).
|
||||
const TELEMETRY_EVERY: std::time::Duration = std::time::Duration::from_secs(30);
|
||||
|
||||
/// Creep trim: the backend ring must sit more than this over its prime target, for
|
||||
/// [`TRIM_AFTER`], before the pump starts shedding — and then only near-silent frames, at most
|
||||
/// one per [`TRIM_SPACING`] (a 20 ms frame per 300 ms ≈ 7 ms per 100 ms). Depth built by a
|
||||
/// burst otherwise only ever comes back down at a full drain (the device consumes exactly what
|
||||
/// it plays).
|
||||
const TRIM_MARGIN_MS: usize = 15;
|
||||
const TRIM_AFTER: std::time::Duration = std::time::Duration::from_secs(2);
|
||||
const TRIM_SPACING: std::time::Duration = std::time::Duration::from_millis(300);
|
||||
/// Peak below which a decoded frame counts as a silence stretch (≈ −48 dBFS).
|
||||
const TRIM_SILENCE_PEAK: f32 = 0.004;
|
||||
|
||||
/// Tuning for [`MicPump`]'s open/reopen/flush behaviour — parameterized so the tests can run the
|
||||
/// real pump loop in milliseconds instead of seconds.
|
||||
#[derive(Clone, Copy)]
|
||||
@@ -75,6 +90,10 @@ const PUMP_TUNING: PumpTuning = PumpTuning {
|
||||
/// every push and on an idle heartbeat, and reopened with backoff. Sessions keep their
|
||||
/// senders; nothing upstream notices.
|
||||
/// - **Stale-flush**: buffered audio is discarded after an uplink gap (see [`PumpTuning`]).
|
||||
/// - **De-jittered**: frames pass a sequence-aware reorder window + loss concealment, and an
|
||||
/// adaptive target-depth estimator drives the backend rings' priming — see
|
||||
/// [`MicDejitter`](super::mic_jitter) (`PUNKTFUNK_MIC_LEGACY_BUFFER=1` restores the fixed
|
||||
/// pre-adaptive buffering).
|
||||
///
|
||||
/// Per-frame Opus DECODE errors stay non-fatal (dropped frame): the mic is shared across every
|
||||
/// concurrent session, so one paired client's junk frames must not deny everyone's mic
|
||||
@@ -178,13 +197,33 @@ where
|
||||
let opened_at = Instant::now();
|
||||
|
||||
// Pump phase — runs until the backend dies (break) or the host shuts down (return).
|
||||
let legacy = super::mic_legacy_buffer();
|
||||
let mut decode_fails: u64 = 0;
|
||||
let mut drain_drops: u64 = 0;
|
||||
let mut trimmed: u64 = 0;
|
||||
let mut frames_seen: u64 = 0;
|
||||
let mut pcm = vec![0f32; 5760 * MIC_CHANNELS as usize]; // up to 120 ms scratch
|
||||
let mut last_push = Instant::now();
|
||||
let mut batch: Vec<MicFrame> = Vec::new();
|
||||
let mut deliveries: Vec<Deliver> = Vec::new();
|
||||
let mut jitter = MicDejitter::new();
|
||||
// Concealment must match the last real frame's duration — libopus sizes PLC from the
|
||||
// output slice. One 20 ms frame until something decodes.
|
||||
let mut plc_samples: usize = 960;
|
||||
let mut applied_target_ms: u32 = 0;
|
||||
let mut over_since: Option<Instant> = None;
|
||||
let mut last_trim = Instant::now();
|
||||
let mut last_log = Instant::now();
|
||||
'pump: loop {
|
||||
match rx.recv_timeout(tuning.heartbeat) {
|
||||
// Wake for the next frame, the heartbeat, or a parked reorder hold aging out —
|
||||
// whichever is soonest.
|
||||
let timeout = jitter
|
||||
.hold_deadline()
|
||||
.map(|d| d.saturating_duration_since(Instant::now()))
|
||||
.unwrap_or(tuning.heartbeat)
|
||||
.min(tuning.heartbeat);
|
||||
deliveries.clear();
|
||||
match rx.recv_timeout(timeout) {
|
||||
Ok(first) => {
|
||||
// Take everything already queued in one gulp: normally that's just `first`,
|
||||
// but after a stall the backlog IS standing mic latency — heal by jumping
|
||||
@@ -200,45 +239,23 @@ where
|
||||
if batch.len() > DRAIN_ABOVE {
|
||||
let drop_n = batch.len() - DRAIN_KEEP;
|
||||
drain_drops += drop_n as u64;
|
||||
if drain_drops.is_power_of_two() {
|
||||
tracing::debug!(
|
||||
dropped = drop_n,
|
||||
total = drain_drops,
|
||||
"mic pump backlog — dropped oldest frames to recover latency"
|
||||
);
|
||||
}
|
||||
batch.drain(..drop_n);
|
||||
}
|
||||
frames_seen += batch.len() as u64;
|
||||
if last_push.elapsed() > tuning.stale_gap {
|
||||
mic.discard();
|
||||
jitter.reset_stream();
|
||||
}
|
||||
let now = Instant::now();
|
||||
for frame in batch.drain(..) {
|
||||
if frame.opus.is_empty() {
|
||||
continue; // DTX silence — the source underruns to silence on its own
|
||||
}
|
||||
match decoder.decode_float(&frame.opus, &mut pcm, false) {
|
||||
Ok(samples_per_ch) => {
|
||||
let total = (samples_per_ch * MIC_CHANNELS as usize).min(pcm.len());
|
||||
if !mic.push(&pcm[..total]) {
|
||||
tracing::warn!("virtual mic backend died — reopening");
|
||||
break 'pump;
|
||||
}
|
||||
last_push = Instant::now();
|
||||
decode_fails = 0;
|
||||
}
|
||||
Err(e) => {
|
||||
// Malformed/garbage frame: drop it, keep the shared mic + decoder
|
||||
// (see the struct docs). Throttled log (1, 2, 4, … fails).
|
||||
decode_fails += 1;
|
||||
if decode_fails.is_power_of_two() {
|
||||
tracing::warn!(error = %e, fails = decode_fails,
|
||||
"mic opus decode failed — dropping frame");
|
||||
}
|
||||
}
|
||||
}
|
||||
jitter.ingest(now, frame, &mut deliveries);
|
||||
}
|
||||
// A hold whose window expired while traffic kept flowing (only late
|
||||
// duplicates arriving) must still flush — the timeout arm never fires then.
|
||||
jitter.flush_expired_hold(now, &mut deliveries);
|
||||
}
|
||||
Err(RecvTimeoutError::Timeout) => {
|
||||
jitter.flush_expired_hold(Instant::now(), &mut deliveries);
|
||||
if !mic.alive() {
|
||||
tracing::warn!("virtual mic backend died while idle — reopening");
|
||||
break;
|
||||
@@ -249,6 +266,123 @@ where
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
for d in deliveries.drain(..) {
|
||||
let samples_per_ch = match d {
|
||||
Deliver::Frame(frame) => {
|
||||
if frame.opus.is_empty() {
|
||||
continue; // DTX silence — the source underruns to silence on its own
|
||||
}
|
||||
match decoder.decode_float(&frame.opus, &mut pcm, false) {
|
||||
Ok(n) => {
|
||||
plc_samples = n.max(120); // ≥ 2.5 ms keeps PLC well-formed
|
||||
decode_fails = 0;
|
||||
n
|
||||
}
|
||||
Err(e) => {
|
||||
// Malformed/garbage frame: drop it, keep the shared mic +
|
||||
// decoder (see the struct docs). Throttled log (1, 2, 4, …).
|
||||
decode_fails += 1;
|
||||
if decode_fails.is_power_of_two() {
|
||||
tracing::warn!(error = %e, fails = decode_fails,
|
||||
"mic opus decode failed — dropping frame");
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
Deliver::Conceal => {
|
||||
// A lost datagram: an empty-input decode synthesizes libopus PLC for
|
||||
// exactly the slice's duration, so the backend ring doesn't starve
|
||||
// into a silence + re-prime cycle over one missing frame.
|
||||
let want = (plc_samples * MIC_CHANNELS as usize).min(pcm.len());
|
||||
match decoder.decode_float(&[], &mut pcm[..want], false) {
|
||||
Ok(n) => n,
|
||||
Err(_) => continue, // nothing decoded yet — nothing to extend
|
||||
}
|
||||
}
|
||||
};
|
||||
let total = (samples_per_ch * MIC_CHANNELS as usize).min(pcm.len());
|
||||
// Creep trim: ring depth persistently over target sheds a near-silent frame at
|
||||
// a time (a few ms per 100 ms) — never speech, never a hard clear. Without it,
|
||||
// depth built by a burst only ever comes back down at a full drain.
|
||||
let mut shed = false;
|
||||
if !legacy {
|
||||
match mic.depth() {
|
||||
Some((buffered, target))
|
||||
if buffered > target + TRIM_MARGIN_MS * SAMPLE_RATE as usize / 1000 =>
|
||||
{
|
||||
let since = *over_since.get_or_insert_with(Instant::now);
|
||||
if since.elapsed() >= TRIM_AFTER
|
||||
&& last_trim.elapsed() >= TRIM_SPACING
|
||||
&& pcm[..total].iter().all(|s| s.abs() < TRIM_SILENCE_PEAK)
|
||||
{
|
||||
shed = true;
|
||||
last_trim = Instant::now();
|
||||
trimmed += 1;
|
||||
}
|
||||
}
|
||||
_ => over_since = None,
|
||||
}
|
||||
}
|
||||
if shed {
|
||||
last_push = Instant::now(); // the uplink is live — a trim is not a stale gap
|
||||
continue;
|
||||
}
|
||||
if !mic.push(&pcm[..total]) {
|
||||
tracing::warn!("virtual mic backend died — reopening");
|
||||
break 'pump;
|
||||
}
|
||||
last_push = Instant::now();
|
||||
}
|
||||
|
||||
// Drive the backend ring's prime target from the measured jitter. Legacy mode
|
||||
// never calls this, which keeps the backend on its fixed constants (see
|
||||
// `VirtualMic::set_target_depth`).
|
||||
if !legacy {
|
||||
let t = jitter.target_ms(Instant::now());
|
||||
if t != applied_target_ms {
|
||||
applied_target_ms = t;
|
||||
mic.set_target_depth(t as usize * SAMPLE_RATE as usize / 1000);
|
||||
}
|
||||
}
|
||||
|
||||
// One structured health line per interval while the mic is live (reset-on-read).
|
||||
if last_log.elapsed() >= TELEMETRY_EVERY {
|
||||
if frames_seen > 0 {
|
||||
let js = jitter.take_stats();
|
||||
let bs = mic.take_stats();
|
||||
let (depth_ms, target_ms) = mic
|
||||
.depth()
|
||||
.map(|(d, t)| {
|
||||
(
|
||||
d * 1000 / SAMPLE_RATE as usize,
|
||||
t * 1000 / SAMPLE_RATE as usize,
|
||||
)
|
||||
})
|
||||
.unwrap_or((0, 0));
|
||||
tracing::info!(
|
||||
depth_ms,
|
||||
target_ms,
|
||||
cadence_ms = (js.cadence_ms * 10.0).round() / 10.0,
|
||||
frame_ms = (js.frame_ms * 10.0).round() / 10.0,
|
||||
frames = frames_seen,
|
||||
gaps = js.seq_gaps,
|
||||
concealed = js.concealed,
|
||||
reorders = js.reorders,
|
||||
late = js.late_drops,
|
||||
drained = drain_drops,
|
||||
trimmed,
|
||||
reprimes = bs.reprimes,
|
||||
overflow_ms = bs.overflow_dropped * 1000 / SAMPLE_RATE as u64,
|
||||
"mic uplink health"
|
||||
);
|
||||
}
|
||||
last_log = Instant::now();
|
||||
frames_seen = 0;
|
||||
drain_drops = 0;
|
||||
trimmed = 0;
|
||||
}
|
||||
}
|
||||
|
||||
// Death triage: an instance that lived is a one-off (PipeWire/audio-engine restart) —
|
||||
@@ -471,6 +605,23 @@ mod pump_tests {
|
||||
h.join.join().unwrap();
|
||||
}
|
||||
|
||||
/// A lost datagram is concealed: seq 0 then 2 (1 missing) must still push ~3 frames of PCM
|
||||
/// (decode, PLC, decode) once the reorder window expires — the backend ring never starves
|
||||
/// over a single missing frame.
|
||||
#[test]
|
||||
fn seq_gap_is_concealed() {
|
||||
let h = start(0);
|
||||
wait_until("instance", || h.alive.lock().unwrap().is_some());
|
||||
h.tx.send(mic_frame(0)).unwrap();
|
||||
h.tx.send(mic_frame(2)).unwrap();
|
||||
// 0 plays immediately; 2 is held ≤ 30 ms for 1, then the gap is concealed and 2 plays.
|
||||
wait_until("conceal + late frame pushed", || {
|
||||
h.pushed.load(Ordering::SeqCst) >= 3 * 960 * 2
|
||||
});
|
||||
drop(h.tx);
|
||||
h.join.join().unwrap();
|
||||
}
|
||||
|
||||
/// An uplink gap discards buffered-stale audio before the next frame plays.
|
||||
#[test]
|
||||
fn discards_after_gap() {
|
||||
|
||||
@@ -19,19 +19,21 @@
|
||||
//! returns `false` and the pump reopens (re-planning, so endpoint churn re-resolves). Before this
|
||||
//! existed, the first device change silently killed mic passthrough for the rest of the host's life.
|
||||
//!
|
||||
//! `push` enqueues decoded interleaved-f32 PCM into a bounded ring (drop-oldest beyond ~120 ms so
|
||||
//! mic latency stays bounded); a dedicated COM-apartment thread renders it event-driven through an
|
||||
//! adaptive jitter buffer (prime → hold → re-prime, see the render loop — clients arrive in bursts,
|
||||
//! the device pulls per-period), filling silence when the client isn't talking. WASAPI objects are
|
||||
//! `!Send`, so they live entirely on that thread (mirrors `WasapiLoopbackCapturer`).
|
||||
//! `push` enqueues decoded interleaved-f32 PCM into a bounded ring (drop-oldest so mic latency
|
||||
//! stays bounded — the bound follows the adaptive prime threshold, or the legacy ~120 ms until
|
||||
//! the pump drives it); a dedicated COM-apartment thread renders it event-driven through a
|
||||
//! jitter buffer (prime → hold → re-prime, see the render loop — clients arrive in bursts, the
|
||||
//! device pulls per-period) whose prime depth the mic pump sets from measured uplink jitter
|
||||
//! ([`VirtualMic::set_target_depth`]), filling silence when the client isn't talking. WASAPI
|
||||
//! objects are `!Send`, so they live entirely on that thread (mirrors `WasapiLoopbackCapturer`).
|
||||
|
||||
// Every `unsafe` block in this file carries a `// SAFETY:` proof; enforce it.
|
||||
#![deny(clippy::undocumented_unsafe_blocks)]
|
||||
|
||||
use super::{audio_control, VirtualMic, SAMPLE_RATE};
|
||||
use super::{audio_control, MicBackendStats, VirtualMic, SAMPLE_RATE};
|
||||
use anyhow::{anyhow, Context, Result};
|
||||
use std::collections::VecDeque;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
|
||||
use std::sync::mpsc::{sync_channel, SyncSender};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::thread::{self, JoinHandle};
|
||||
@@ -41,26 +43,53 @@ use wasapi::{Direction, SampleType, StreamMode, WaveFormat};
|
||||
const CHANNELS: u32 = 2;
|
||||
/// 48 kHz stereo f32: 2 channels * 4 bytes.
|
||||
const BLOCK_ALIGN: usize = 2 * 4;
|
||||
/// Jitter-buffer priming depth (~48 ms): the render loop emits pure silence until this much PCM
|
||||
/// is queued, then plays from the cushion. Clients deliver mic audio in BURSTS (the Mac client's
|
||||
/// input tap yields ~two 20 ms Opus packets every ~42 ms) while WASAPI pulls a small block every
|
||||
/// device period (~10 ms) — with no cushion the queue sits near-empty and most periods insert
|
||||
/// mid-stream silence: the "crackling mic" (heard live, Mac → Windows host 2026-07-03; the Linux
|
||||
/// backend's process callback primes the same way and the identical stream was clean there). The
|
||||
/// depth must cover the worst inter-burst gap (~42 ms), so ~48 ms with re-prime on a full drain.
|
||||
/// LEGACY jitter-buffer priming depth (~48 ms): the render loop emits pure silence until this
|
||||
/// much PCM is queued, then plays from the cushion. Old clients deliver mic audio in BURSTS
|
||||
/// (the Mac client's input tap yields ~two 20 ms Opus packets every ~42 ms) while WASAPI pulls
|
||||
/// a small block every device period (~10 ms) — with no cushion the queue sits near-empty and
|
||||
/// most periods insert mid-stream silence: the "crackling mic" (heard live, Mac → Windows host
|
||||
/// 2026-07-03; the Linux backend's process callback primes the same way and the identical
|
||||
/// stream was clean there). The depth had to cover the worst inter-burst gap (~42 ms), so
|
||||
/// ~48 ms with re-prime on a full drain. Today the pump MEASURES that gap and drives the prime
|
||||
/// threshold per client ([`VirtualMic::set_target_depth`] — bursty clients still get ~48 ms,
|
||||
/// modern 10 ms-cadence ones ~25–35 ms); this constant remains the fallback until the pump's
|
||||
/// first estimate, and forever under `PUNKTFUNK_MIC_LEGACY_BUFFER=1`.
|
||||
const PRIME_BYTES: usize = (SAMPLE_RATE as usize * 48 / 1000) * BLOCK_ALIGN;
|
||||
/// Bound the inject queue at ~120 ms so the passed-through mic stays low-latency (drop oldest
|
||||
/// beyond): the priming cushion (~48 ms) plus arrival-burst headroom.
|
||||
/// LEGACY bound for the inject queue at ~120 ms (drop oldest beyond): the fixed priming
|
||||
/// cushion plus arrival-burst headroom. Applies only while the pump isn't driving the target;
|
||||
/// adaptive mode bounds the queue at prime + [`CAP_HEADROOM_BYTES`] instead (≈ 50–105 ms).
|
||||
const MAX_QUEUE_BYTES: usize = (SAMPLE_RATE as usize * 120 / 1000) * BLOCK_ALIGN;
|
||||
/// Producer-side overflow headroom (~32 ms) over the render loop's prime threshold when the
|
||||
/// adaptive target drives the ring.
|
||||
const CAP_HEADROOM_BYTES: usize = (SAMPLE_RATE as usize * 32 / 1000) * BLOCK_ALIGN;
|
||||
|
||||
pub struct WasapiVirtualMic {
|
||||
queue: Arc<Mutex<VecDeque<u8>>>,
|
||||
stop: Arc<AtomicBool>,
|
||||
/// False once the render thread has exited (device error or stop) — the pump's reopen signal.
|
||||
alive: Arc<AtomicBool>,
|
||||
/// Ring policy/telemetry shared with the render thread (see [`RingShared`]).
|
||||
ring: Arc<RingShared>,
|
||||
join: Option<JoinHandle<()>>,
|
||||
}
|
||||
|
||||
/// Atomics shared between the pump-facing handle and the render thread: the pump's adaptive
|
||||
/// de-jitter target in, the effective prime threshold + reset-on-read counters out. All
|
||||
/// `Relaxed` — a slowly-moving target and telemetry, not synchronization.
|
||||
#[derive(Default)]
|
||||
struct RingShared {
|
||||
/// Pump-set jitter target in bytes. `0` = the pump never spoke (legacy mode, or its first
|
||||
/// estimate hasn't landed) → the render loop keeps the fixed [`PRIME_BYTES`] and `push`
|
||||
/// keeps the fixed [`MAX_QUEUE_BYTES`].
|
||||
target_bytes: AtomicUsize,
|
||||
/// Effective prime threshold (bytes) of the last render iteration.
|
||||
prime_bytes: AtomicUsize,
|
||||
/// Full-drain re-prime arms (see [`MicBackendStats`]).
|
||||
reprimes: AtomicU64,
|
||||
/// Per-channel samples dropped by the overflow cap.
|
||||
overflow: AtomicU64,
|
||||
}
|
||||
|
||||
impl WasapiVirtualMic {
|
||||
pub fn open(channels: u32) -> Result<Self> {
|
||||
anyhow::ensure!(
|
||||
@@ -70,14 +99,15 @@ impl WasapiVirtualMic {
|
||||
let queue = Arc::new(Mutex::new(VecDeque::<u8>::new()));
|
||||
let stop = Arc::new(AtomicBool::new(false));
|
||||
let alive = Arc::new(AtomicBool::new(true));
|
||||
let ring = Arc::new(RingShared::default());
|
||||
// Bring-up handshake: report the resolved device (or the error) before returning, so a missing
|
||||
// virtual-mic device surfaces as Err (the caller retries with backoff) not a silent dead thread.
|
||||
let (ready_tx, ready_rx) = sync_channel::<Result<String>>(1);
|
||||
let (q, st, al) = (queue.clone(), stop.clone(), alive.clone());
|
||||
let (q, st, rg, al) = (queue.clone(), stop.clone(), ring.clone(), alive.clone());
|
||||
let join = thread::Builder::new()
|
||||
.name("punktfunk-wasapi-mic".into())
|
||||
.spawn(move || {
|
||||
if let Err(e) = render_thread(q, st, ready_tx) {
|
||||
if let Err(e) = render_thread(q, st, rg, ready_tx) {
|
||||
tracing::error!(error = %format!("{e:#}"), "wasapi virtual-mic thread failed");
|
||||
}
|
||||
// Normal stop or device error alike: this instance is done — the pump reopens.
|
||||
@@ -92,6 +122,7 @@ impl WasapiVirtualMic {
|
||||
queue,
|
||||
stop,
|
||||
alive,
|
||||
ring,
|
||||
join: Some(join),
|
||||
})
|
||||
}
|
||||
@@ -122,10 +153,25 @@ impl VirtualMic for WasapiVirtualMic {
|
||||
for &s in pcm {
|
||||
q.extend(s.to_le_bytes());
|
||||
}
|
||||
// Drop-oldest to keep latency bounded (mic is real-time; stale audio is worse than dropped).
|
||||
if q.len() > MAX_QUEUE_BYTES {
|
||||
let excess = q.len() - MAX_QUEUE_BYTES;
|
||||
// Drop-oldest to keep latency bounded (mic is real-time; stale audio is worse than
|
||||
// dropped). With the pump driving the target, the bound follows the render loop's prime
|
||||
// threshold + headroom; otherwise (legacy / no estimate yet) the fixed 120 ms applies.
|
||||
let cap = if self.ring.target_bytes.load(Ordering::Relaxed) == 0 {
|
||||
MAX_QUEUE_BYTES
|
||||
} else {
|
||||
// `max(PRIME_BYTES)` covers the one render period before the loop first publishes.
|
||||
self.ring
|
||||
.prime_bytes
|
||||
.load(Ordering::Relaxed)
|
||||
.max(PRIME_BYTES)
|
||||
+ CAP_HEADROOM_BYTES
|
||||
};
|
||||
if q.len() > cap {
|
||||
let excess = q.len() - cap;
|
||||
q.drain(..excess);
|
||||
self.ring
|
||||
.overflow
|
||||
.fetch_add((excess / BLOCK_ALIGN) as u64, Ordering::Relaxed);
|
||||
}
|
||||
true
|
||||
}
|
||||
@@ -143,6 +189,28 @@ impl VirtualMic for WasapiVirtualMic {
|
||||
fn channels(&self) -> u32 {
|
||||
CHANNELS
|
||||
}
|
||||
|
||||
fn set_target_depth(&self, samples_per_ch: usize) {
|
||||
self.ring
|
||||
.target_bytes
|
||||
.store(samples_per_ch * BLOCK_ALIGN, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
fn depth(&self) -> Option<(usize, usize)> {
|
||||
let prime = self.ring.prime_bytes.load(Ordering::Relaxed);
|
||||
if prime == 0 {
|
||||
return None; // render loop hasn't run yet
|
||||
}
|
||||
let q = self.queue.lock().ok()?;
|
||||
Some((q.len() / BLOCK_ALIGN, prime / BLOCK_ALIGN))
|
||||
}
|
||||
|
||||
fn take_stats(&self) -> MicBackendStats {
|
||||
MicBackendStats {
|
||||
reprimes: self.ring.reprimes.swap(0, Ordering::Relaxed),
|
||||
overflow_dropped: self.ring.overflow.swap(0, Ordering::Relaxed),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve the mic inject target from the wiring plan, auto-installing the Steam Streaming pair
|
||||
@@ -273,6 +341,7 @@ fn try_install_steam_audio(inf_name: &str) -> bool {
|
||||
fn render_thread(
|
||||
queue: Arc<Mutex<VecDeque<u8>>>,
|
||||
stop: Arc<AtomicBool>,
|
||||
shared: Arc<RingShared>,
|
||||
ready: SyncSender<Result<String>>,
|
||||
) -> Result<()> {
|
||||
if let Err(e) = wasapi::initialize_mta()
|
||||
@@ -284,7 +353,7 @@ fn render_thread(
|
||||
}
|
||||
// Open + start the render stream. The WASAPI objects must outlive the loop, so build them here and
|
||||
// keep them (a closure that *returned* them would drop them); on any failure report Err and exit.
|
||||
let setup = (|| -> Result<(wasapi::AudioClient, wasapi::AudioRenderClient, wasapi::Handle, String)> {
|
||||
let setup = (|| -> Result<(wasapi::AudioClient, wasapi::AudioRenderClient, wasapi::Handle, i64, String)> {
|
||||
let (device, name) = resolve_target()?;
|
||||
let mut audio_client = device.get_iaudioclient().context("IAudioClient")?;
|
||||
// 48 kHz stereo f32; autoconvert lets WASAPI shared-mode SRC match the device mix format.
|
||||
@@ -312,9 +381,9 @@ fn render_thread(
|
||||
let buf_frames = audio_client.get_buffer_size().context("buffer size")? as usize;
|
||||
let _ = render_client.write_to_device(buf_frames, &vec![0u8; buf_frames * BLOCK_ALIGN], None);
|
||||
audio_client.start_stream().context("start render stream")?;
|
||||
Ok((audio_client, render_client, h_event, name))
|
||||
Ok((audio_client, render_client, h_event, default_period, name))
|
||||
})();
|
||||
let (audio_client, render_client, h_event, name) = match setup {
|
||||
let (audio_client, render_client, h_event, default_period, name) = match setup {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let _ = ready.send(Err(anyhow!("{e:#}")));
|
||||
@@ -322,18 +391,26 @@ fn render_thread(
|
||||
}
|
||||
};
|
||||
let _ = ready.send(Ok(name));
|
||||
// One device period in bytes (period is in 100 ns units; floor 10 ms if it reads absurd) —
|
||||
// the device-side pull granularity the adaptive prime threshold builds on.
|
||||
let period_bytes = ((default_period.max(0) as usize * SAMPLE_RATE as usize / 10_000_000)
|
||||
.max(SAMPLE_RATE as usize / 100))
|
||||
* BLOCK_ALIGN;
|
||||
|
||||
// Any error below (endpoint invalidated/removed, engine restart) propagates out of the loop,
|
||||
// ending the thread — the `alive` flag flips in the spawn wrapper and the pump reopens.
|
||||
//
|
||||
// Adaptive jitter buffer (mirrors the Linux backend's process callback): clients push mic
|
||||
// audio in bursts on their own clock while the device pulls a block every period from an
|
||||
// independent clock, so a greedy per-period drain leaves the queue near-empty and pads most
|
||||
// periods with mid-stream silence — audible as constant crackling. Instead: emit silence
|
||||
// until [`PRIME_BYTES`] is buffered, then play from the cushion (zero-filling only a
|
||||
// momentary shortfall), and re-prime only after a genuine FULL drain (the client went quiet —
|
||||
// between talk spurts the cushion rebuilds, and [`VirtualMic::discard`] resets it across
|
||||
// session gaps).
|
||||
// Jitter buffer (mirrors the Linux backend's process callback): clients push mic audio in
|
||||
// bursts on their own clock while the device pulls a block every period from an independent
|
||||
// clock, so a greedy per-period drain leaves the queue near-empty and pads most periods
|
||||
// with mid-stream silence — audible as constant crackling. Instead: emit silence until the
|
||||
// prime threshold is buffered, then play from the cushion (zero-filling only a momentary
|
||||
// shortfall), and re-prime only after a genuine FULL drain (the client went quiet — between
|
||||
// talk spurts the cushion rebuilds, and [`VirtualMic::discard`] resets it across session
|
||||
// gaps). The threshold = one device period + the pump's measured-jitter target
|
||||
// ([`VirtualMic::set_target_depth`]); the fixed [`PRIME_BYTES`] until the pump's first
|
||||
// estimate, and forever under `PUNKTFUNK_MIC_LEGACY_BUFFER=1` (the pump then never drives
|
||||
// the target).
|
||||
let mut buf: Vec<u8> = Vec::new();
|
||||
let mut primed = false;
|
||||
while !stop.load(Ordering::Relaxed) {
|
||||
@@ -351,11 +428,18 @@ fn render_thread(
|
||||
if buf.len() < need {
|
||||
buf.resize(need, 0);
|
||||
}
|
||||
let target = shared.target_bytes.load(Ordering::Relaxed);
|
||||
let prime = if target == 0 {
|
||||
PRIME_BYTES
|
||||
} else {
|
||||
period_bytes + target
|
||||
};
|
||||
shared.prime_bytes.store(prime, Ordering::Relaxed);
|
||||
// Silence base; overwrite with queued mic PCM once the cushion is primed.
|
||||
buf[..need].fill(0);
|
||||
{
|
||||
let mut q = queue.lock().unwrap();
|
||||
if !primed && q.len() >= PRIME_BYTES {
|
||||
if !primed && q.len() >= prime {
|
||||
primed = true;
|
||||
}
|
||||
if primed {
|
||||
@@ -365,6 +449,7 @@ fn render_thread(
|
||||
}
|
||||
if q.is_empty() {
|
||||
primed = false; // fully drained — re-prime before producing again
|
||||
shared.reprimes.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -150,6 +150,7 @@ See your desktop page ([KDE](/docs/kde), [GNOME](/docs/gnome)) for when to set t
|
||||
|---|---|---|
|
||||
| `PUNKTFUNK_AUDIO_GAIN` | float (default `1.0`) | **(Moonlight/GameStream sessions only)** Linear gain applied to captured desktop audio — bump it for a quiet source. The native `punktfunk/1` path ignores it; adjust the source's own volume there instead. |
|
||||
| `PUNKTFUNK_MIC_DEVICE` | name substring | **(Windows)** Target mic-uplink device by friendly-name substring (first match wins). |
|
||||
| `PUNKTFUNK_MIC_LEGACY_BUFFER` | `1` | Restore the fixed pre-adaptive mic buffering (a ~48 ms prime and ~120 ms cap on Windows; a buffer scaled to the recording app's audio quantum on Linux) instead of the adaptive per-client jitter target. One-release escape hatch: if the microphone coming out of the host only sounds right *with* this set, that's a bug — please report it. |
|
||||
| `PUNKTFUNK_NO_MIC_INSTALL` | set | **(Windows)** Skip installing the virtual-mic driver (e.g. when the host runs as SYSTEM). |
|
||||
| `PUNKTFUNK_HOST_AUDIO` | set | **(Windows)** Also play the stream's audio on the host's own speakers. While a session is capturing desktop audio the host parks the default playback device on a silent sink, so sound comes out of the *client* only — that's why the PC goes quiet when a stream starts. Set this to prefer a real output device instead (audible on both ends). The default is put back when the capture closes. |
|
||||
| `PUNKTFUNK_KEEP_DEFAULT` | set | **(Windows)** Never touch the default playback/recording devices at all — the host leaves whatever you chose in Sound settings in place. The mic uplink still picks a target device; you may then have to select it yourself. |
|
||||
|
||||
Reference in New Issue
Block a user