diff --git a/crates/pf-client-core/src/audio.rs b/crates/pf-client-core/src/audio.rs index 6e62912b..d40a185d 100644 --- a/crates/pf-client-core/src/audio.rs +++ b/crates/pf-client-core/src/audio.rs @@ -14,9 +14,12 @@ use std::sync::mpsc::{Receiver, SyncSender, TrySendError}; use std::sync::Arc; const SAMPLE_RATE: u32 = 48_000; -const CHANNELS: usize = 2; -/// Mic frames are 20 ms (960 samples/channel) — any size ≤ 120 ms is fine host-side. -const MIC_FRAME: usize = 960; +/// Mic capture is MONO: voice is mono at the source, the host accepts any Opus channel +/// layout (its stereo decoder upmixes), and half the samples halve the encode + wire cost. +const MIC_CHANNELS: usize = 1; +/// Mic frames are 10 ms (480 mono samples) — any size ≤ 120 ms is fine host-side; 10 ms +/// halves the frame-fill share of mouth-to-ear latency vs the old 20 ms. +const MIC_FRAME: usize = 480; struct Terminate; @@ -327,8 +330,9 @@ fn pw_thread( Ok(()) } -/// The microphone uplink: capture the default input device, Opus-encode 20 ms chunks, -/// ship them as 0xCB datagrams into the host's virtual PipeWire source. +/// The microphone uplink: capture the default input device (or the picked / echo-cancelled +/// source), Opus-encode 10 ms mono chunks, ship them as 0xCB datagrams into the host's +/// virtual PipeWire source. pub struct MicStreamer { quit_tx: pipewire::channel::Sender, thread: Option>, @@ -361,8 +365,8 @@ impl Drop for MicStreamer { } } -/// Capture-side state: accumulated PCM and the Opus encoder (encoding a 20 ms frame is -/// ~100 µs — fine inside the process callback). +/// Capture-side state: accumulated PCM and the Opus encoder (encoding a 10 ms frame is +/// well under 100 µs — fine inside the process callback). struct MicData { connector: Arc, ring: VecDeque, @@ -371,6 +375,55 @@ struct MicData { out: Vec, } +/// `PUNKTFUNK_NO_AEC=1` — the opt-out for every mic echo-cancellation hook (here: the +/// echo-cancelled-source preference; the WASAPI twin gates its Communications category on it). +fn aec_disabled() -> bool { + std::env::var("PUNKTFUNK_NO_AEC").is_ok_and(|v| !v.is_empty() && v != "0") +} + +/// The capture stream's `target.object`, in preference order: the Settings microphone pick +/// (`Settings::mic_device` via session main's `PUNKTFUNK_AUDIO_SOURCE`) verbatim, else — so a +/// desktop that already runs `module-echo-cancel` stops feeding its own downlink audio back +/// into the host's virtual mic — the first echo-cancelled source in the graph. `None` = the +/// user picked nothing and no such source exists: PipeWire's default routing, as before. +/// +/// Preference-only by design: loading `libpipewire-module-echo-cancel` ourselves needs +/// `pw_context_load_module`, which the pipewire crate (0.9) doesn't expose safely — until it +/// does, we only ever target processing the user (or their session) already set up. +fn mic_capture_target() -> Option { + if let Ok(target) = std::env::var("PUNKTFUNK_AUDIO_SOURCE") { + if !target.is_empty() { + return Some(target); + } + } + if aec_disabled() { + return None; + } + let name = echo_cancel_source()?; + tracing::info!( + source = %name, + "mic capture targets the echo-cancelled source (PUNKTFUNK_NO_AEC=1 to disable)" + ); + Some(name) +} + +/// Find an existing echo-cancelled capture node: the first `Audio/Source` whose `node.name` +/// or description says echo-cancel (`module-echo-cancel`'s convention — `echo-cancel-*` +/// nodes, "Echo-Cancel …" descriptions; PulseAudio-compat setups match too). One registry +/// roundtrip via [`devices`]; any failure reads as "none". +fn echo_cancel_source() -> Option { + let (_, sources) = devices().ok()?; + sources.into_iter().find_map(|d| { + let name = d.name.to_ascii_lowercase(); + let desc = d.description.to_ascii_lowercase(); + (name.contains("echo-cancel") + || name.contains("echo_cancel") + || desc.contains("echo-cancel") + || desc.contains("echo cancel")) + .then_some(d.name) + }) +} + fn mic_thread( connector: &Arc, quit_rx: pipewire::channel::Receiver, @@ -384,9 +437,14 @@ fn mic_thread( PW_INIT.call_once(pw::init); let mut encoder = - opus::Encoder::new(SAMPLE_RATE, opus::Channels::Stereo, opus::Application::Voip) + opus::Encoder::new(SAMPLE_RATE, opus::Channels::Mono, opus::Application::Voip) .map_err(|e| anyhow::anyhow!("opus encoder: {e}"))?; - let _ = encoder.set_bitrate(opus::Bitrate::Bits(64_000)); + // Voice tuning: 48 kbps mono is transparent for speech; in-band FEC + an assumed 10 % + // loss let the host's decoder rebuild a lost 0xCB datagram from its successor instead + // of concealing (datagrams are fire-and-forget — this FEC is the only redundancy). + let _ = encoder.set_bitrate(opus::Bitrate::Bits(48_000)); + let _ = encoder.set_inband_fec(true); + let _ = encoder.set_packet_loss_perc(10); let mainloop = pw::main_loop::MainLoopRc::new(None).context("pw mic MainLoop")?; let context = pw::context::ContextRc::new(&mainloop, None).context("pw mic Context")?; @@ -405,14 +463,15 @@ fn mic_thread( *pw::keys::MEDIA_ROLE => "Communication", *pw::keys::NODE_NAME => "punktfunk-mic-capture", *pw::keys::NODE_DESCRIPTION => "Punktfunk Microphone", + // ~10 ms quantum (one mic frame). Without it the capture stream inherits the graph + // quantum — commonly 1024–2048 samples, so the mic arrived in 21–43 ms bursts that + // sat ahead of the encoder as latency (the playback stream always asked for 5 ms). + *pw::keys::NODE_LATENCY => "480/48000", }; - // The Settings microphone pick (`Settings::mic_device` via session main). - if let Ok(target) = std::env::var("PUNKTFUNK_AUDIO_SOURCE") { - if !target.is_empty() { - // Raw key: the `keys::TARGET_OBJECT` constant is feature-gated on a newer - // libpipewire than we require; the wire name is stable. - props.insert("target.object", target); - } + if let Some(target) = mic_capture_target() { + // Raw key: the `keys::TARGET_OBJECT` constant is feature-gated on a newer + // libpipewire than we require; the wire name is stable. + props.insert("target.object", target); } let stream = pw::stream::StreamBox::new(&core, "punktfunk-mic-capture", props) .context("pw mic Stream")?; @@ -447,9 +506,9 @@ fn mic_thread( .push_back(f32::from_le_bytes([s[0], s[1], s[2], s[3]])); } } - // Ship every complete 20 ms stereo frame. - while ud.ring.len() >= MIC_FRAME * CHANNELS { - let pcm: Vec = ud.ring.drain(..MIC_FRAME * CHANNELS).collect(); + // Ship every complete 10 ms mono frame. + while ud.ring.len() >= MIC_FRAME * MIC_CHANNELS { + let pcm: Vec = ud.ring.drain(..MIC_FRAME * MIC_CHANNELS).collect(); match ud.encoder.encode_float(&pcm, &mut ud.out) { Ok(len) => { let pts = std::time::SystemTime::now() @@ -473,7 +532,8 @@ fn mic_thread( let mut info = AudioInfoRaw::new(); info.set_format(AudioFormat::F32LE); info.set_rate(SAMPLE_RATE); - info.set_channels(CHANNELS as u32); + // Mono: the stream's adapter downmixes whatever layout the source really has. + info.set_channels(MIC_CHANNELS as u32); let obj = pw::spa::pod::Object { type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(), id: pw::spa::param::ParamType::EnumFormat.as_raw(), diff --git a/crates/pf-client-core/src/audio_wasapi.rs b/crates/pf-client-core/src/audio_wasapi.rs index 5fb1de9f..4b191df9 100644 --- a/crates/pf-client-core/src/audio_wasapi.rs +++ b/crates/pf-client-core/src/audio_wasapi.rs @@ -23,14 +23,22 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc::{Receiver, SyncSender, TrySendError}; use std::sync::Arc; use std::time::Duration; -use wasapi::{DeviceEnumerator, Direction, SampleType, StreamMode, WaveFormat}; +use wasapi::{ + AudioClientProperties, DeviceEnumerator, Direction, SampleType, StreamCategory, StreamMode, + WaveFormat, +}; const SAMPLE_RATE: usize = 48_000; -/// The microphone uplink stays stereo (the host's virtual mic is stereo). The render path is -/// multichannel — its channel count + block align are runtime, driven by the host-resolved layout. -const CHANNELS: usize = 2; -/// Mic frames are 20 ms (960 samples/channel) — any size ≤ 120 ms is fine host-side. -const MIC_FRAME: usize = 960; +/// Mic capture requests STEREO from WASAPI (autoconvert matrixes any endpoint layout down to +/// it — the proven path; `read_from_device_to_deque` then delivers our requested format) and +/// downmixes to MONO in code before the encoder: voice is mono at the source, the host accepts +/// any Opus channel layout (its stereo decoder upmixes), and half the samples halve the encode +/// + wire cost. The render path is multichannel — its channel count + block align are runtime, +/// driven by the host-resolved layout. +const CAPT_CHANNELS: usize = 2; +/// Mic frames are 10 ms (480 mono samples) — any size ≤ 120 ms is fine host-side; 10 ms +/// halves the frame-fill share of mouth-to-ear latency vs the old 20 ms. +const MIC_FRAME: usize = 480; pub struct AudioPlayer { pcm_tx: SyncSender>, @@ -217,8 +225,8 @@ fn render_thread( res } -/// The microphone uplink: capture the default input device, Opus-encode 20 ms chunks, ship -/// them as 0xCB datagrams into the host's virtual mic source. +/// The microphone uplink: capture the default input device, Opus-encode 10 ms mono chunks, +/// ship them as 0xCB datagrams into the host's virtual mic source. pub struct MicStreamer { stop: Arc, thread: Option>, @@ -259,18 +267,37 @@ fn mic_thread(connector: &Arc, stop: Arc) -> Result<() let mut encoder = opus::Encoder::new( SAMPLE_RATE as u32, - opus::Channels::Stereo, + opus::Channels::Mono, opus::Application::Voip, ) .map_err(|e| anyhow!("opus encoder: {e}"))?; - let _ = encoder.set_bitrate(opus::Bitrate::Bits(64_000)); + // Voice tuning: 48 kbps mono is transparent for speech; in-band FEC + an assumed 10 % + // loss let the host's decoder rebuild a lost 0xCB datagram from its successor instead + // of concealing (datagrams are fire-and-forget — this FEC is the only redundancy). + let _ = encoder.set_bitrate(opus::Bitrate::Bits(48_000)); + let _ = encoder.set_inband_fec(true); + let _ = encoder.set_packet_loss_perc(10); let device = DeviceEnumerator::new() .context("DeviceEnumerator")? .get_default_device(&Direction::Capture) .context("default capture endpoint (no microphone?)")?; let mut audio_client = device.get_iaudioclient().context("IAudioClient")?; - let desired = WaveFormat::new(32, 32, &SampleType::Float, SAMPLE_RATE, CHANNELS, None); + // Communications category → the endpoint's communications signal-processing chain. A + // driver/APO stack with an echo canceller only engages it for communications-category + // streams; the default (Other) category never did, so the downlink audio playing on + // this box fed straight back into the host's virtual mic. Must precede Initialize + // (SetClientProperties is a pre-init call; the wasapi crate QIs IAudioClient2 inside). + // Best-effort: an endpoint without IAudioClient2 just keeps the default category. + // PUNKTFUNK_NO_AEC=1 opts out (same lever as the Linux echo-cancel-source preference). + if !std::env::var("PUNKTFUNK_NO_AEC").is_ok_and(|v| !v.is_empty() && v != "0") { + if let Err(e) = audio_client.set_properties( + AudioClientProperties::new().set_category(StreamCategory::Communications), + ) { + tracing::debug!(error = %e, "mic capture: Communications category not set"); + } + } + let desired = WaveFormat::new(32, 32, &SampleType::Float, SAMPLE_RATE, CAPT_CHANNELS, None); let (default_period, _min_period) = audio_client.get_device_period().context("device period")?; let mode = StreamMode::EventsShared { @@ -308,13 +335,23 @@ fn mic_thread(connector: &Arc, stop: Arc) -> Result<() Err(e) => return Err(anyhow!("get_next_packet_size: {e}")), } } - let whole = (bytes.len() / 4) * 4; - for c in bytes.drain(..whole).collect::>().chunks_exact(4) { - ring.push_back(f32::from_le_bytes([c[0], c[1], c[2], c[3]])); + // One stereo capture frame (8 bytes) → one mono sample: average L/R. Autoconvert + // already matrixed the endpoint's real layout (mono/stereo/array mic) into the + // stereo stream we initialized, so this is the only downmix left to do. + let stereo_frame = 4 * CAPT_CHANNELS; + let whole = (bytes.len() / stereo_frame) * stereo_frame; + for c in bytes + .drain(..whole) + .collect::>() + .chunks_exact(stereo_frame) + { + let l = f32::from_le_bytes([c[0], c[1], c[2], c[3]]); + let r = f32::from_le_bytes([c[4], c[5], c[6], c[7]]); + ring.push_back((l + r) * 0.5); } - // Ship every complete 20 ms stereo frame. - while ring.len() >= MIC_FRAME * CHANNELS { - let pcm: Vec = ring.drain(..MIC_FRAME * CHANNELS).collect(); + // Ship every complete 10 ms mono frame. + while ring.len() >= MIC_FRAME { + let pcm: Vec = ring.drain(..MIC_FRAME).collect(); match encoder.encode_float(&pcm, &mut out) { Ok(len) => { let pts = std::time::SystemTime::now() diff --git a/crates/pf-client-core/src/session.rs b/crates/pf-client-core/src/session.rs index 93c6dc38..0773e116 100644 --- a/crates/pf-client-core/src/session.rs +++ b/crates/pf-client-core/src/session.rs @@ -121,6 +121,12 @@ pub struct Stats { /// received+lost (%). The OSD renders the counter line only when nonzero. pub lost: u32, pub lost_pct: f32, + /// Mic uplink frames this window: handed to the QUIC datagram send, and shed anywhere + /// client-side (queue-full at the producer + the pump's stale-oldest backlog governor — + /// see [`NativeClient::mic_stats`]). Both stay 0 while the mic is off, so the OSD + /// renders the mic line only when the uplink is live. + pub mic_sent: u32, + pub mic_dropped: u32, /// The decode path frames actually took this window (`"vaapi"`/`"software"`, empty /// until the first frame) — the OSD's trailing tag; tracks a mid-session fallback. pub decoder: &'static str, @@ -434,6 +440,9 @@ fn pump( let mut dec_path: &'static str = ""; // The stats window keeps its own drop cursor — the OSD shows the per-window delta. let mut window_dropped = connector.frames_dropped(); + // Mic uplink cursor (same per-window diffing): a healthy 10 ms-frame mic reads ~100 + // sent/s; a nonzero drop delta is the queue shedding backlog (see NativeClient::mic_stats). + let mut window_mic = connector.mic_stats(); let mut last_kf_req: Option = None; // Freeze-until-reanchor: the shared post-loss gate ([`punktfunk_core::reanchor::ReanchorGate`]). // Armed on any loss signal (frame-index gap, dropped-count climb, decoder wedge/demotion), it @@ -789,6 +798,12 @@ fn pump( let (pace_p50, _) = window_percentiles(&mut pace_us_win); let lost = dropped.saturating_sub(window_dropped) as u32; window_dropped = dropped; + let mic_now = connector.mic_stats(); + let mic_sent = mic_now.sent.saturating_sub(window_mic.sent) as u32; + let mic_dropped = (mic_now.dropped_full + mic_now.dropped_stale) + .saturating_sub(window_mic.dropped_full + window_mic.dropped_stale) + as u32; + window_mic = mic_now; tracing::debug!( fps = frames_n, hostnet_p50_us = hn_p50, @@ -800,6 +815,8 @@ fn pump( pace_p50_us = pace_p50, decode_p50_us = dec_p50, lost, + mic_sent, + mic_dropped, total_frames, "stream window" ); @@ -822,6 +839,8 @@ fn pump( } else { 0.0 }, + mic_sent, + mic_dropped, decoder: dec_path, })); window_start = Instant::now(); diff --git a/crates/pf-presenter/src/run.rs b/crates/pf-presenter/src/run.rs index 50c62fad..123b5a53 100644 --- a/crates/pf-presenter/src/run.rs +++ b/crates/pf-presenter/src/run.rs @@ -2017,6 +2017,14 @@ fn stats_text( if s.lost > 0 { text.push_str(&format!("\nlost {} ({:.1}%)", s.lost, s.lost_pct)); } + // The mic uplink line renders only while the mic is live (a healthy 10 ms-frame uplink + // reads ~100 f/s) and only in Detailed — drops here are the client shedding backlog. + if detailed && (s.mic_sent > 0 || s.mic_dropped > 0) { + text.push_str(&format!("\nmic {} f/s", s.mic_sent)); + if s.mic_dropped > 0 { + text.push_str(&format!(" · dropped {}", s.mic_dropped)); + } + } text } @@ -2262,6 +2270,8 @@ mod tests { decode_ms: 1.8, lost: 3, lost_pct: 0.4, + mic_sent: 0, + mic_dropped: 0, decoder: "vulkan", }, PresentedWindow { @@ -2303,6 +2313,30 @@ mod tests { ); } + /// The mic uplink line: Detailed-only, and only while the uplink is live. + #[test] + fn stats_text_mic_line() { + let (mut s, p) = sample(); + let text = |s: &Stats, v| stats_text(v, "m", s, &p, false, false, None); + assert!( + !text(&s, StatsVerbosity::Detailed).contains("mic"), + "no mic line while the mic is off" + ); + s.mic_sent = 100; + let detailed = text(&s, StatsVerbosity::Detailed); + assert!(detailed.contains("\nmic 100 f/s")); + assert!( + !detailed.contains("dropped"), + "a healthy uplink shows no drop term" + ); + assert!( + !text(&s, StatsVerbosity::Normal).contains("mic"), + "mic line is Detailed-only" + ); + s.mic_dropped = 7; + assert!(text(&s, StatsVerbosity::Detailed).contains("mic 100 f/s · dropped 7")); + } + /// Compact omits the latency term until the presenter's first e2e window lands. #[test] fn compact_waits_for_e2e() { diff --git a/crates/punktfunk-core/src/client/mod.rs b/crates/punktfunk-core/src/client/mod.rs index 59472a0e..eda3b9e5 100644 --- a/crates/punktfunk-core/src/client/mod.rs +++ b/crates/punktfunk-core/src/client/mod.rs @@ -63,11 +63,45 @@ fn join_host_port(host: &str, port: u16) -> String { } } -/// Outbound mic uplink queue depth: 5 ms Opus frames, so 64 is ~320 ms of audio — far beyond -/// any worker stall a live mic session survives anyway. On overflow the FRESH frame is dropped -/// (a tokio mpsc can't shed from the head; by the time 320 ms are queued the stream is broken -/// either way, and the bound is about memory, not audio quality) and logged at debug. -const MIC_QUEUE: usize = 64; +/// Outbound mic uplink queue depth: desktop clients send 10 ms Opus frames (mobile still 20 ms), +/// so 12 is ~120–240 ms of audio — the hard memory cap, not the working depth. (The old 64 was +/// mislabeled "~320 ms of 5 ms frames"; the frames were 20 ms, so it really allowed 1.28 s, and +/// since a filled tokio mpsc can only drop the FRESH frame, every queued frame became permanent +/// standing latency once a stall filled it.) The pump's mic task now sheds the OLDEST frames +/// whenever more than [`MIC_BACKLOG_MAX`] are still waiting, so a stall costs a short dropout +/// and heals the moment the worker catches up. The producer stays non-blocking: on overflow it +/// still drops the fresh frame (logged at debug) — the shed loop keeps that a rare event. +const MIC_QUEUE: usize = 12; + +/// The mic backlog the pump tolerates before shedding oldest-first: ~60 ms of 10 ms frames — +/// enough slack to ride out an encode/send hiccup, small enough that voice stays conversational +/// and never accrues a session-long standing delay. +pub(crate) const MIC_BACKLOG_MAX: usize = 6; + +/// Mic uplink counters, shared between the producer side ([`NativeClient::send_mic`]) and the +/// pump's mic task. All monotonic for the session; a stats HUD windows them by diffing +/// successive [`NativeClient::mic_stats`] snapshots (the `frames_dropped` pattern). +#[derive(Debug, Default)] +pub(crate) struct MicUplinkCounters { + /// Frames handed to the QUIC datagram send (past every client-side queue). + pub(crate) sent: AtomicU64, + /// Frames shed at enqueue — the worker queue was full ([`MIC_QUEUE`]). + pub(crate) dropped_full: AtomicU64, + /// Frames shed by the pump's backlog governor (stale-oldest past [`MIC_BACKLOG_MAX`]). + pub(crate) dropped_stale: AtomicU64, +} + +/// A [`NativeClient::mic_stats`] snapshot: cumulative mic uplink frame counts per stage. +#[derive(Clone, Copy, Debug, Default)] +pub struct MicUplinkStats { + /// Frames handed to the QUIC datagram send. + pub sent: u64, + /// Frames shed at enqueue (worker queue full). + pub dropped_full: u64, + /// Frames shed by the pump's backlog governor (stale-oldest — see [`MIC_QUEUE`]'s + /// self-healing note). + pub dropped_stale: u64, +} /// Outbound control-request queue depth. The requests are sparse (mode switches, keyframe /// requests, ~1.3 loss reports/s, clock re-syncs) — 32 is hours of headroom; a full queue means @@ -101,9 +135,12 @@ pub struct NativeClient { cursor_state: Mutex>, input_tx: tokio::sync::mpsc::UnboundedSender, /// Outbound mic frames `(seq, pts_ns, opus)` → encoded as 0xCB datagrams by the worker. - /// Bounded ([`MIC_QUEUE`]): a wedged worker drops fresh frames (logged) instead of queueing - /// audio-latency (and memory) without limit — mic is best-effort end to end. + /// Bounded ([`MIC_QUEUE`]): the pump sheds stale frames oldest-first and a full queue drops + /// the fresh one (logged) instead of queueing audio-latency (and memory) without limit — + /// mic is best-effort end to end, and a standing backlog is worse than a dropout. mic_tx: tokio::sync::mpsc::Sender<(u32, u64, Vec)>, + /// Mic uplink counters (sent / dropped per stage) — see [`NativeClient::mic_stats`]. + mic_stats: Arc, /// Outbound 0xCC rich-input plane, PRE-ENCODED datagrams: [`RichInput`] touchpad/motion /// (encoded in [`NativeClient::send_rich_input`]) and stylus [`crate::quic::PenBatch`]es /// (encoded in [`NativeClient::send_pen`]) share the channel — the worker's task just @@ -396,6 +433,7 @@ impl NativeClient { let probe = Arc::new(Mutex::new(ProbeState::default())); let frames_dropped = Arc::new(AtomicU64::new(0)); let fec_recovered = Arc::new(AtomicU64::new(0)); + let mic_stats = Arc::new(MicUplinkCounters::default()); let hot_tids = Arc::new(Mutex::new(Vec::new())); let clock_offset = Arc::new(AtomicI64::new(0)); let decode_lat = Arc::new(Mutex::new(DecodeLatAcc::default())); @@ -408,6 +446,7 @@ impl NativeClient { let probe_w = probe.clone(); let frames_dropped_w = frames_dropped.clone(); let fec_recovered_w = fec_recovered.clone(); + let mic_stats_w = mic_stats.clone(); let hot_tids_w = hot_tids.clone(); let clock_offset_w = clock_offset.clone(); let decode_lat_w = decode_lat.clone(); @@ -472,6 +511,7 @@ impl NativeClient { probe: probe_w, frames_dropped: frames_dropped_w, fec_recovered: fec_recovered_w, + mic_stats: mic_stats_w, hot_tids: hot_tids_w, clock_offset: clock_offset_w, decode_lat: decode_lat_w, @@ -506,6 +546,7 @@ impl NativeClient { cursor_state: Mutex::new(cursor_state_rx), input_tx, mic_tx, + mic_stats, rich_input_tx, ctrl_tx, clip: Mutex::new(clip_event_rx), @@ -709,6 +750,18 @@ impl NativeClient { self.fec_recovered.load(Ordering::Relaxed) } + /// Cumulative mic uplink frame counts per stage: handed to the QUIC datagram send, shed at + /// enqueue (worker queue full), and shed by the pump's backlog governor (stale-oldest — see + /// [`MIC_QUEUE`]'s self-healing note). All monotonic for the session; a stats HUD windows + /// them by diffing successive reads, like [`frames_dropped`](Self::frames_dropped). + pub fn mic_stats(&self) -> MicUplinkStats { + MicUplinkStats { + sent: self.mic_stats.sent.load(Ordering::Relaxed), + dropped_full: self.mic_stats.dropped_full.load(Ordering::Relaxed), + dropped_stale: self.mic_stats.dropped_stale.load(Ordering::Relaxed), + } + } + /// Whether the underlying QUIC session has ended — the worker's connection-close watcher set the /// shutdown flag (`conn.closed()` fired: a host suspend / crash / network drop idle-timed the /// connection out, or the host closed it), or a deliberate [`disconnect_quit`](Self::disconnect_quit) @@ -1124,8 +1177,10 @@ impl NativeClient { match self.mic_tx.try_send((seq, pts_ns, opus)) { Ok(()) => Ok(()), Err(TrySendError::Full(_)) => { - // Bounded queue full = the worker stalled for ~MIC_QUEUE x 5 ms. Shed this - // frame (mic is best-effort end to end) instead of queueing latency/memory. + // Bounded queue full = the worker stalled long enough to outrun even the + // pump's oldest-first shed. Drop this frame (mic is best-effort end to end) + // instead of queueing latency/memory; the counter keeps the loss visible. + self.mic_stats.dropped_full.fetch_add(1, Ordering::Relaxed); tracing::debug!("mic uplink queue full — dropping frame"); Ok(()) } diff --git a/crates/punktfunk-core/src/client/pump.rs b/crates/punktfunk-core/src/client/pump.rs index f3774164..73b922a2 100644 --- a/crates/punktfunk-core/src/client/pump.rs +++ b/crates/punktfunk-core/src/client/pump.rs @@ -68,6 +68,7 @@ pub(super) async fn run_pump(args: WorkerArgs) { probe, frames_dropped, fec_recovered, + mic_stats, hot_tids, clock_offset, decode_lat, @@ -92,11 +93,20 @@ pub(super) async fn run_pump(args: WorkerArgs) { tokio::spawn(input_task::run(conn.clone(), input_rx, gamepad_snapshots)); // Mic task: embedder Opus mic frames → 0xCB uplink datagrams (best-effort, dropped on loss). + // Self-healing latency bound: every frame still queued once this task catches up is standing + // mic delay from then on, so a frame with more than [`MIC_BACKLOG_MAX`] successors already + // waiting is shed as stale instead of sent — a stall costs a short dropout, not a + // session-long lag (see [`MIC_QUEUE`]). The counters feed [`NativeClient::mic_stats`]. let mic_conn = conn.clone(); tokio::spawn(async move { while let Some((seq, pts_ns, opus)) = mic_rx.recv().await { + if mic_rx.len() > MIC_BACKLOG_MAX { + mic_stats.dropped_stale.fetch_add(1, Ordering::Relaxed); + continue; + } let d = crate::quic::encode_mic_datagram(seq, pts_ns, &opus); let _ = mic_conn.send_datagram(d.into()); + mic_stats.sent.fetch_add(1, Ordering::Relaxed); } }); diff --git a/crates/punktfunk-core/src/client/worker.rs b/crates/punktfunk-core/src/client/worker.rs index 07b1f961..fd02c44b 100644 --- a/crates/punktfunk-core/src/client/worker.rs +++ b/crates/punktfunk-core/src/client/worker.rs @@ -66,6 +66,9 @@ pub(crate) struct WorkerArgs { pub(crate) probe: Arc>, pub(crate) frames_dropped: Arc, pub(crate) fec_recovered: Arc, + /// Mic uplink counters (see [`NativeClient::mic_stats`]): the pump's mic task counts wire + /// sends and its own stale-shed drops here; the producer counts queue-full drops. + pub(crate) mic_stats: Arc, pub(crate) hot_tids: Arc>>, /// The live clock offset (see [`NativeClient::clock_offset`]): the worker seeds it with the /// connect-time estimate; the control task's mid-stream re-syncs update it.