Merge branch 'audio/host-mic-jitter' into audio/mic-latency-echo
This commit is contained in:
@@ -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::MicPump;
|
||||
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,15 +5,49 @@
|
||||
//! 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;
|
||||
|
||||
/// Mic is 48 kHz stereo — matches the Opus stereo decoder and the host→client audio layout.
|
||||
pub const MIC_CHANNELS: u32 = 2;
|
||||
/// Bound for the shared mic frame queue (drop-newest when full): the host-lifetime queue is
|
||||
/// shared across all concurrent sessions and must not grow without limit under a near-line-rate
|
||||
/// flood (security-review 2026-06-28 S6). 64 × 5–20 ms frames ≈ 0.3–1.3 s of slack.
|
||||
const MIC_QUEUE_CAP: usize = 64;
|
||||
|
||||
/// One client mic frame off the wire (`0xCB` — `punktfunk_core::quic::decode_mic_datagram`).
|
||||
/// The datagram's seq + pts used to be decoded at ingest and thrown away; the pump's de-jitter
|
||||
/// needs them (reorder window, loss concealment, cadence/drift), so they ride the queue with
|
||||
/// the Opus payload.
|
||||
pub struct MicFrame {
|
||||
pub seq: u32,
|
||||
pub pts_ns: u64,
|
||||
pub opus: Vec<u8>,
|
||||
}
|
||||
|
||||
/// Bound for the shared mic frame queue (drop-newest at the producer when full): the
|
||||
/// host-lifetime queue is shared across all concurrent sessions and must not grow without limit
|
||||
/// under a near-line-rate flood (security-review 2026-06-28 S6). 12 × 5–20 ms frames ≈
|
||||
/// 60–240 ms of slack — enough for a scheduling hiccup, and the consumer heals the rest (see
|
||||
/// [`DRAIN_ABOVE`]). The old cap of 64 could park 1.28 s of standing latency here that only a
|
||||
/// >600 ms silence gap ever flushed.
|
||||
const MIC_QUEUE_CAP: usize = 12;
|
||||
/// Consumer self-healing: a pump that wakes to a backlog deeper than this drops down to the
|
||||
/// newest [`DRAIN_KEEP`] frames before decoding — the oldest frames are already late, and
|
||||
/// playing them would turn a one-off stall into permanent mic delay.
|
||||
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.
|
||||
@@ -56,20 +90,24 @@ 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
|
||||
/// (security-review 2026-06-28 S2). The thread exits when every sender is dropped (host
|
||||
/// shutdown), tearing the backend down.
|
||||
pub struct MicPump {
|
||||
tx: std::sync::mpsc::SyncSender<Vec<u8>>,
|
||||
tx: std::sync::mpsc::SyncSender<MicFrame>,
|
||||
}
|
||||
|
||||
impl MicPump {
|
||||
/// Start the host-lifetime pump (Linux/Windows). On platforms without a virtual-mic backend
|
||||
/// the thread just drains and drops frames (sessions still count the datagrams).
|
||||
pub fn start() -> MicPump {
|
||||
let (tx, rx) = std::sync::mpsc::sync_channel::<Vec<u8>>(MIC_QUEUE_CAP);
|
||||
let (tx, rx) = std::sync::mpsc::sync_channel::<MicFrame>(MIC_QUEUE_CAP);
|
||||
let spawned = std::thread::Builder::new()
|
||||
.name("punktfunk-mic-pump".into())
|
||||
.spawn(move || {
|
||||
@@ -90,7 +128,7 @@ impl MicPump {
|
||||
/// A sender a session forwards the client's Opus mic frames to (`try_send` — never block a
|
||||
/// datagram loop). Cloned per session; dropping a clone does NOT stop the pump (it holds
|
||||
/// the original sender for the host life).
|
||||
pub fn sender(&self) -> std::sync::mpsc::SyncSender<Vec<u8>> {
|
||||
pub fn sender(&self) -> std::sync::mpsc::SyncSender<MicFrame> {
|
||||
self.tx.clone()
|
||||
}
|
||||
}
|
||||
@@ -99,7 +137,7 @@ impl MicPump {
|
||||
/// never accumulates a stale backlog and senders never see a wedged queue. Returns `false` when
|
||||
/// every sender is gone (host shutdown).
|
||||
#[cfg_attr(not(any(target_os = "linux", target_os = "windows")), allow(dead_code))]
|
||||
fn drain_sleep(rx: &std::sync::mpsc::Receiver<Vec<u8>>, dur: std::time::Duration) -> bool {
|
||||
fn drain_sleep(rx: &std::sync::mpsc::Receiver<MicFrame>, dur: std::time::Duration) -> bool {
|
||||
use std::sync::mpsc::RecvTimeoutError;
|
||||
let deadline = std::time::Instant::now() + dur;
|
||||
loop {
|
||||
@@ -118,7 +156,7 @@ fn drain_sleep(rx: &std::sync::mpsc::Receiver<Vec<u8>>, dur: std::time::Duration
|
||||
/// The pump loop. `opener` is injected so the tests can run the REAL loop against a mock
|
||||
/// backend; production passes [`open_virtual_mic`](super::open_virtual_mic).
|
||||
#[cfg_attr(not(any(target_os = "linux", target_os = "windows")), allow(dead_code))]
|
||||
fn pump_thread<O>(rx: std::sync::mpsc::Receiver<Vec<u8>>, opener: O, tuning: PumpTuning)
|
||||
fn pump_thread<O>(rx: std::sync::mpsc::Receiver<MicFrame>, opener: O, tuning: PumpTuning)
|
||||
where
|
||||
O: Fn() -> Result<Box<dyn VirtualMic>>,
|
||||
{
|
||||
@@ -159,40 +197,65 @@ 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();
|
||||
loop {
|
||||
match rx.recv_timeout(tuning.heartbeat) {
|
||||
Ok(frame) => {
|
||||
if frame.is_empty() {
|
||||
continue; // DTX silence — the source underruns to silence on its own
|
||||
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 {
|
||||
// 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
|
||||
// to the newest few frames instead of replaying the pile.
|
||||
batch.clear();
|
||||
batch.push(first);
|
||||
while batch.len() <= MIC_QUEUE_CAP {
|
||||
match rx.try_recv() {
|
||||
Ok(f) => batch.push(f),
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
if batch.len() > DRAIN_ABOVE {
|
||||
let drop_n = batch.len() - DRAIN_KEEP;
|
||||
drain_drops += drop_n as u64;
|
||||
batch.drain(..drop_n);
|
||||
}
|
||||
frames_seen += batch.len() as u64;
|
||||
if last_push.elapsed() > tuning.stale_gap {
|
||||
mic.discard();
|
||||
jitter.reset_stream();
|
||||
}
|
||||
match decoder.decode_float(&frame, &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;
|
||||
}
|
||||
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");
|
||||
}
|
||||
}
|
||||
let now = Instant::now();
|
||||
for frame in batch.drain(..) {
|
||||
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;
|
||||
@@ -203,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) —
|
||||
@@ -252,7 +432,7 @@ mod pump_tests {
|
||||
}
|
||||
|
||||
struct Harness {
|
||||
tx: std::sync::mpsc::SyncSender<Vec<u8>>,
|
||||
tx: std::sync::mpsc::SyncSender<MicFrame>,
|
||||
opens: Arc<AtomicUsize>,
|
||||
alive: Arc<Mutex<Option<Arc<AtomicBool>>>>, // latest instance's kill switch
|
||||
pushed: Arc<AtomicUsize>,
|
||||
@@ -265,7 +445,7 @@ mod pump_tests {
|
||||
/// pre-killed (exercises the rapid-death churn guard). `stable_after` mirrors the tuning
|
||||
/// field (ZERO = every death counts as stable → immediate reopen, keeping tests fast).
|
||||
fn start_tuned(fail_first: usize, dead_on_arrival: bool, stable_after: Duration) -> Harness {
|
||||
let (tx, rx) = std::sync::mpsc::sync_channel::<Vec<u8>>(MIC_QUEUE_CAP);
|
||||
let (tx, rx) = std::sync::mpsc::sync_channel::<MicFrame>(MIC_QUEUE_CAP);
|
||||
let opens = Arc::new(AtomicUsize::new(0));
|
||||
let alive = Arc::new(Mutex::new(None::<Arc<AtomicBool>>));
|
||||
let pushed = Arc::new(AtomicUsize::new(0));
|
||||
@@ -336,6 +516,15 @@ mod pump_tests {
|
||||
out
|
||||
}
|
||||
|
||||
/// A wire-shaped frame: `seq` with the matching 20 ms pts, payload from [`opus_frame`].
|
||||
fn mic_frame(seq: u32) -> MicFrame {
|
||||
MicFrame {
|
||||
seq,
|
||||
pts_ns: seq as u64 * 20_000_000,
|
||||
opus: opus_frame(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Eager: the backend opens (after transient failures) with NO frame ever sent.
|
||||
#[test]
|
||||
fn opens_eagerly_with_backoff() {
|
||||
@@ -352,7 +541,7 @@ mod pump_tests {
|
||||
fn decodes_and_pushes() {
|
||||
let h = start(0);
|
||||
wait_until("open", || h.alive.lock().unwrap().is_some());
|
||||
h.tx.send(opus_frame()).unwrap();
|
||||
h.tx.send(mic_frame(0)).unwrap();
|
||||
wait_until("pcm pushed", || h.pushed.load(Ordering::SeqCst) > 0);
|
||||
drop(h.tx);
|
||||
h.join.join().unwrap();
|
||||
@@ -388,9 +577,9 @@ mod pump_tests {
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.store(false, Ordering::Release);
|
||||
h.tx.send(opus_frame()).unwrap(); // push sees death → reopen
|
||||
h.tx.send(mic_frame(0)).unwrap(); // push sees death → reopen
|
||||
wait_until("reopen", || h.opens.load(Ordering::SeqCst) >= 2);
|
||||
h.tx.send(opus_frame()).unwrap();
|
||||
h.tx.send(mic_frame(1)).unwrap();
|
||||
wait_until("pcm after reopen", || h.pushed.load(Ordering::SeqCst) > 0);
|
||||
drop(h.tx);
|
||||
h.join.join().unwrap();
|
||||
@@ -416,15 +605,32 @@ 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() {
|
||||
let h = start(0);
|
||||
wait_until("instance", || h.alive.lock().unwrap().is_some());
|
||||
h.tx.send(opus_frame()).unwrap();
|
||||
h.tx.send(mic_frame(0)).unwrap();
|
||||
wait_until("first push", || h.pushed.load(Ordering::SeqCst) > 0);
|
||||
std::thread::sleep(Duration::from_millis(150)); // > stale_gap
|
||||
h.tx.send(opus_frame()).unwrap();
|
||||
h.tx.send(mic_frame(1)).unwrap();
|
||||
wait_until("discard on gap", || h.discards.load(Ordering::SeqCst) >= 1);
|
||||
drop(h.tx);
|
||||
h.join.join().unwrap();
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -69,11 +69,18 @@ fn capture_for(mic_render_lname: &str) -> &'static [&'static str] {
|
||||
}
|
||||
|
||||
/// A render endpoint no loopback should capture: the VB-CABLE (reserved for the mic even when it
|
||||
/// isn't the chosen target — capturing a cable someone else feeds echoes too) and the Steam
|
||||
/// Streaming Speakers, whose loopback is silent (validated live). Also the capture-side
|
||||
/// watchdog's test for "the operator's new default can never work — snap back to the plan".
|
||||
/// isn't the chosen target — capturing a cable someone else feeds echoes too), the Steam
|
||||
/// Streaming Speakers, whose loopback is silent (validated live), and any VoiceMeeter or
|
||||
/// generically-"virtual" endpoint. VoiceMeeter's strips share one internal mixer: capturing ANY
|
||||
/// of its render endpoints re-captures what the mic wrote into another one — a digital feedback
|
||||
/// loop with no acoustic path to break it (mic=`Voicemeeter Input`, loopback=`Voicemeeter Aux
|
||||
/// Input` used to pass the old name checks). Also the capture-side watchdog's test for "the
|
||||
/// operator's new default can never work — snap back to the plan".
|
||||
pub(crate) fn excluded_from_loopback(lname: &str) -> bool {
|
||||
lname.contains("cable") || lname.contains("steam streaming speakers")
|
||||
lname.contains("cable")
|
||||
|| lname.contains("steam streaming speakers")
|
||||
|| lname.contains("voicemeeter")
|
||||
|| lname.contains("virtual")
|
||||
}
|
||||
|
||||
/// A render endpoint that is SILENT on the host but loopback-capturable — the client-only audio
|
||||
@@ -140,10 +147,14 @@ pub(crate) fn plan(
|
||||
.iter()
|
||||
.find(|(n, id)| not_mic(id) && silent_sink(&n.to_lowercase()))
|
||||
};
|
||||
// `virtualish` here too: a virtual endpoint that slipped past `excluded_from_loopback`'s
|
||||
// name list (a future cable/mixer sibling) is an internal-feedback loop waiting to happen —
|
||||
// no loopback is the honest answer, exactly like the cable-only case.
|
||||
let leftover = || {
|
||||
renders
|
||||
.iter()
|
||||
.find(|(n, id)| not_mic(id) && !excluded_from_loopback(&n.to_lowercase()))
|
||||
renders.iter().find(|(n, id)| {
|
||||
let ln = n.to_lowercase();
|
||||
not_mic(id) && !excluded_from_loopback(&ln) && !virtualish(&ln)
|
||||
})
|
||||
};
|
||||
let loopback_render = if host_audio {
|
||||
real_hw().or_else(silent).or_else(leftover)
|
||||
@@ -359,4 +370,59 @@ mod tests {
|
||||
assert!(w.mic_render.is_none());
|
||||
assert_eq!(w.loopback_render.unwrap().0, "Speakers (Realtek HD Audio)");
|
||||
}
|
||||
|
||||
/// VoiceMeeter box with real hardware: no cable, so the mic takes `Voicemeeter Input` — and
|
||||
/// the loopback must NEVER be `Voicemeeter Aux Input` (same internal mixer: capturing any
|
||||
/// VoiceMeeter render re-captures what the mic wrote — a digital feedback loop). Real
|
||||
/// hardware wins the loopback in both modes.
|
||||
#[test]
|
||||
fn voicemeeter_aux_never_pairs_with_voicemeeter_mic() {
|
||||
let renders = [
|
||||
ep("Voicemeeter Input (VB-Audio Voicemeeter VAIO)"),
|
||||
ep("Voicemeeter Aux Input (VB-Audio Voicemeeter AUX VAIO)"),
|
||||
ep("Speakers (Realtek HD Audio)"),
|
||||
];
|
||||
let captures = [ep("Voicemeeter Out B1 (VB-Audio Voicemeeter VAIO)")];
|
||||
for host_audio in [false, true] {
|
||||
let w = plan(&renders, &captures, None, host_audio);
|
||||
assert_eq!(
|
||||
w.mic_render.as_ref().unwrap().0,
|
||||
"Voicemeeter Input (VB-Audio Voicemeeter VAIO)",
|
||||
"host_audio={host_audio}"
|
||||
);
|
||||
assert_eq!(
|
||||
w.loopback_render.as_ref().unwrap().0,
|
||||
"Speakers (Realtek HD Audio)",
|
||||
"host_audio={host_audio}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Only VoiceMeeter endpoints (headless mixer box): mic wins one, the loopback is honestly
|
||||
/// absent — like the cable-only case, never another strip of the same mixer.
|
||||
#[test]
|
||||
fn voicemeeter_only_no_loopback() {
|
||||
let renders = [
|
||||
ep("Voicemeeter Input (VB-Audio Voicemeeter VAIO)"),
|
||||
ep("Voicemeeter Aux Input (VB-Audio Voicemeeter AUX VAIO)"),
|
||||
];
|
||||
for host_audio in [false, true] {
|
||||
let w = plan(&renders, &[], None, host_audio);
|
||||
assert!(w.mic_render.is_some(), "host_audio={host_audio}");
|
||||
assert!(w.loopback_render.is_none(), "host_audio={host_audio}");
|
||||
}
|
||||
}
|
||||
|
||||
/// A generically-"virtual" leftover (unknown vendor cable) is refused too: `leftover()`
|
||||
/// applies `virtualish`, so a virtual endpoint that slips past the name list can't become
|
||||
/// the loopback.
|
||||
#[test]
|
||||
fn unknown_virtual_never_loopback() {
|
||||
let renders = [
|
||||
ep("CABLE Input (VB-Audio Virtual Cable)"),
|
||||
ep("Speakers (Some Virtual Audio Device)"),
|
||||
];
|
||||
let w = plan(&renders, &[], None, false);
|
||||
assert!(w.loopback_render.is_none());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -760,7 +760,7 @@ async fn serve_session(
|
||||
opts: &Punktfunk1Options,
|
||||
audio_cap: &AudioCapSlot,
|
||||
inj_tx: std::sync::mpsc::Sender<InputEvent>,
|
||||
mic_tx: std::sync::mpsc::SyncSender<Vec<u8>>,
|
||||
mic_tx: std::sync::mpsc::SyncSender<crate::audio::MicFrame>,
|
||||
host_fp: &[u8; 32],
|
||||
np: &NativePairing,
|
||||
last_pairing: &std::sync::Mutex<Option<std::time::Instant>>,
|
||||
@@ -1178,11 +1178,17 @@ async fn serve_session(
|
||||
tokio::spawn(async move {
|
||||
let (mut input_count, mut mic_count, mut rich_count) = (0u64, 0u64, 0u64);
|
||||
while let Ok(d) = input_conn.read_datagram().await {
|
||||
if let Some((_seq, _pts, opus)) = punktfunk_core::quic::decode_mic_datagram(&d) {
|
||||
if let Some((seq, pts, opus)) = punktfunk_core::quic::decode_mic_datagram(&d) {
|
||||
mic_count += 1;
|
||||
// Host-lifetime mic service (bounded queue): `try_send` drops the frame when the
|
||||
// service is full or gone, never blocking this datagram loop (security-review S6).
|
||||
let _ = mic_tx.try_send(opus.to_vec());
|
||||
// seq + pts ride along — the pump's de-jitter reorders, conceals losses and
|
||||
// tracks cadence with them (they used to be decoded here and thrown away).
|
||||
let _ = mic_tx.try_send(crate::audio::MicFrame {
|
||||
seq,
|
||||
pts_ns: pts,
|
||||
opus: opus.to_vec(),
|
||||
});
|
||||
} else if let Some(rich) = punktfunk_core::quic::RichInput::decode(&d) {
|
||||
rich_count += 1;
|
||||
if rich_tx.send(ClientInput::Rich(rich)).is_err() {
|
||||
|
||||
@@ -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. |
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
---
|
||||
title: Why do I hear myself
|
||||
description: Echo while streaming — the four places it can come from (client speakers, Windows monitoring loops, host speakers, virtual mixers) and how to stop each one.
|
||||
---
|
||||
|
||||
You talk into your device's microphone and hear your own voice come back a beat later. The echo
|
||||
is almost never "in the stream" itself — it's a loop in one of four well-known places. Work
|
||||
through them in order; the first two cover nearly every report.
|
||||
|
||||
## Your device's speakers
|
||||
|
||||
If the stream's audio plays out of the **speakers of the device you're streaming on** — a
|
||||
phone, tablet or laptop without headphones — its microphone picks that sound back up and sends
|
||||
it to the host along with your voice. Everyone in your voice chat hears the game twice, and you
|
||||
hear yourself whenever anything routes the mic back.
|
||||
|
||||
**Fix: use headphones on the device you're streaming on.** Newer clients cancel this
|
||||
automatically (acoustic echo cancellation is rolling out per platform); headphones are the
|
||||
reliable fix everywhere today.
|
||||
|
||||
## "Listen to this device" and app monitoring (Windows hosts)
|
||||
|
||||
Windows can play a microphone straight out of the speakers. If **Listen to this device** is
|
||||
ticked for the Punktfunk mic (usually *CABLE Output*), your voice plays on the host's output —
|
||||
which the stream then captures and sends right back to you.
|
||||
|
||||
Open **Sound settings → More sound settings → Recording**, double-click *CABLE Output*, and on
|
||||
the **Listen** tab untick *Listen to this device*.
|
||||
|
||||
The same loop hides in apps: **Discord's** *Mic Test* / input monitoring, **OBS's** *Monitor
|
||||
audio* on a mic source, and similar monitoring features in other tools all play your mic into
|
||||
the host's output. Turn the monitoring off rather than the mic.
|
||||
|
||||
## The host's own speakers
|
||||
|
||||
If you set `PUNKTFUNK_HOST_AUDIO` (Windows) so the stream's sound also plays in the room, and
|
||||
you're streaming **from that same room**, your device's mic hears the host's speakers. Remove
|
||||
the setting while you stream from nearby, or turn the host's volume down.
|
||||
|
||||
## Virtual mixers (VoiceMeeter and friends)
|
||||
|
||||
VoiceMeeter's virtual devices all share one internal mixer. Older Punktfunk hosts could pick one
|
||||
VoiceMeeter strip as the microphone target and *another* as the audio capture — a feedback loop
|
||||
with no acoustics involved at all. Current hosts refuse to capture VoiceMeeter or other virtual
|
||||
endpoints for desktop audio, so this fixes itself with an update. If you route audio through
|
||||
VoiceMeeter on purpose, make sure no strip that hears the Punktfunk mic feeds the output being
|
||||
streamed.
|
||||
|
||||
## What the host log can tell you
|
||||
|
||||
While your microphone is in use, the host writes a **`mic uplink health`** line to its log every
|
||||
30 seconds (web console → **Logs**). It shows how much of your voice is buffered on the host
|
||||
(`depth_ms` vs `target_ms`), how much the network lost (`gaps`, `concealed`), and how steadily
|
||||
your client is delivering audio (`cadence_ms`). It won't point at an echo loop directly — echo
|
||||
is a routing problem, not a network one — but if your voice also sounds choppy or delayed,
|
||||
include that line in a bug report. The startup log also names exactly which devices the host
|
||||
picked for the microphone and for audio capture, which is the quickest way to spot a monitoring
|
||||
loop like the ones above.
|
||||
@@ -47,6 +47,7 @@
|
||||
"wake-on-lan",
|
||||
"---Troubleshooting---",
|
||||
"troubleshooting",
|
||||
"echo",
|
||||
"stats",
|
||||
"forgot-password",
|
||||
"---Project---",
|
||||
|
||||
Reference in New Issue
Block a user