From 039c34322ceb655702856d088c52c3a71fba8329 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Sun, 31 May 2026 22:35:52 -0400 Subject: [PATCH] fix: eliminate playback crackle by pinning the PipeWire buffer to one quantum The playback stream was negotiated with a ~256ms (12288-frame) maxsize buffer. The sink drains the graph quantum (1024 frames) per cycle, so one of our buffers lasted ~12 cycles and `process` was called only ~4x/sec, each time asking us to fill all 12288 frames -- far more than the 200ms (9600-sample) playout ring could ever hold. So ~half of every buffer was silence-fill: a steady ~46% underrun, audible as constant crackle. This is a consumer-side buffer-size bug, upstream of production pacing, which is why earlier mixer-pacing attempts never moved the numbers. Fix: pass an explicit SPA_TYPE_OBJECT_ParamBuffers param on connect, pinning buffer size to one 1024-frame quantum (2048 bytes mono S16LE). PipeWire now hands us a quantum-sized buffer ~47x/sec, the ring satisfies every callback, and slice.len()/stride equals the quantum so we never over-pull. A node.latency hint is added too (not load-bearing on its own -- the hint alone changed nothing; the Buffers param is the fix). Note: pipewire 0.9.2 only exposes feature v0_3_32, so Buffer::requested() is unreachable -- pinning the buffer size is the available lever. Verified with the probe (quantum=1024, 47 cb/s, underrun +0 steady) and by ear: clean 440Hz tone, no clicks. Local playout path only -- not yet verified on a live two-peer call. Also in this commit (the investigation scaffolding that proved it out): - Fill-paced mixer: production tracks the hardware clock via a shared exact ring-occupancy gauge (Arc) kept near PLAYBACK_TARGET_SAMPLES, replacing the fixed 20ms timer that beat against the 1024 quantum. - src/bin/audio_probe.rs: drives a sine through the real start_playback path with no network/mic, for isolating the local output stage. - playout-health logging: quiet in normal use (logs only on underrun/ dropped > 0); set PEERSPEAK_AUDIO_VERBOSE=1 for the per-second heartbeat (audio_probe sets it automatically). Co-Authored-By: Claude Opus 4.8 --- src/audio/mod.rs | 25 ++++- src/audio/pipewire_impl.rs | 206 ++++++++++++++++++++++++++++++++++--- src/bin/audio_probe.rs | 119 +++++++++++++++++++++ src/core/mod.rs | 53 ++++++---- src/lib.rs | 6 ++ 5 files changed, 376 insertions(+), 33 deletions(-) create mode 100644 src/bin/audio_probe.rs diff --git a/src/audio/mod.rs b/src/audio/mod.rs index 5822226..e5ce4ac 100644 --- a/src/audio/mod.rs +++ b/src/audio/mod.rs @@ -1,6 +1,18 @@ use std::sync::mpsc::{Sender, Receiver}; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; use thiserror::Error; +/// Target depth of the playback ring buffer, in samples (48kHz mono). +/// +/// The playout chain is paced to keep the ring near this level: production is +/// driven by how fast PipeWire actually drains the ring (the hardware clock), +/// not by a fixed software timer — which is what eliminates the producer/ +/// consumer beat that otherwise churns ~20% of audio into drops + silence. +/// 2880 = 60ms = 3×20ms frames, comfortably above the 2048-sample max quantum +/// so a single hardware pull can never empty the ring before the mixer refills. +pub const PLAYBACK_TARGET_SAMPLES: usize = 2880; + #[derive(Error, Debug)] pub enum AudioError { #[error("Failed to initialize audio backend: {0}")] @@ -22,7 +34,18 @@ pub trait AudioBackend: Send + Sync { /// Starts playing back raw PCM audio to the output device (speaker), /// reading mixed/incoming chunks of samples from the provided Receiver. - fn start_playback(&self, rx: Receiver>, target_node: Option) -> Result<(), AudioError>; + /// + /// `ring_fill` is updated with the playback ring's current occupancy (in + /// samples) as the device drains and the worker fills it. The caller (the + /// mixer) reads it to pace production to the hardware clock — produce only + /// while the ring is below [`PLAYBACK_TARGET_SAMPLES`] — instead of on a + /// fixed timer that beats against the device quantum. + fn start_playback( + &self, + rx: Receiver>, + target_node: Option, + ring_fill: Arc, + ) -> Result<(), AudioError>; /// Stops both capture and playback streams. fn stop(&self) -> Result<(), AudioError>; diff --git a/src/audio/pipewire_impl.rs b/src/audio/pipewire_impl.rs index 6483cae..f4ad278 100644 --- a/src/audio/pipewire_impl.rs +++ b/src/audio/pipewire_impl.rs @@ -1,7 +1,7 @@ use crate::audio::{AudioBackend, AudioError}; use std::sync::mpsc::{Sender, Receiver}; use std::sync::{Arc, Mutex}; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::thread::{self, JoinHandle}; use std::time::Duration; use pipewire as pw; @@ -63,7 +63,12 @@ impl AudioBackend for PipeWireBackend { Ok(()) } - fn start_playback(&self, rx: Receiver>, target_node: Option) -> Result<(), AudioError> { + fn start_playback( + &self, + rx: Receiver>, + target_node: Option, + ring_fill: Arc, + ) -> Result<(), AudioError> { let mut playback_guard = self.playback_state.lock().unwrap(); if playback_guard.is_some() { return Err(AudioError::Stream("Playback already started".to_string())); @@ -74,7 +79,7 @@ impl AudioBackend for PipeWireBackend { let thread = thread::Builder::new() .name("peerspeak-playback".to_string()) .spawn(move || { - if let Err(e) = run_playback(cmd_rx, rx, target_node) { + if let Err(e) = run_playback(cmd_rx, rx, target_node, ring_fill) { eprintln!("Playback thread error: {:?}", e); } }) @@ -219,7 +224,12 @@ fn run_capture(cmd_rx: pw::channel::Receiver<()>, tx: Sender>, target_n Ok(()) } -fn run_playback(cmd_rx: pw::channel::Receiver<()>, rx: Receiver>, target_node: Option) -> Result<(), AudioError> { +fn run_playback( + cmd_rx: pw::channel::Receiver<()>, + rx: Receiver>, + target_node: Option, + fill_gauge: Arc, +) -> Result<(), AudioError> { let mainloop = pw::main_loop::MainLoopRc::new(None) .map_err(|e| AudioError::Init(e.to_string()))?; let context = pw::context::ContextRc::new(&mainloop, None) @@ -227,10 +237,46 @@ fn run_playback(cmd_rx: pw::channel::Receiver<()>, rx: Receiver>, targe let core = context.connect_rc(None) .map_err(|e| AudioError::Init(e.to_string()))?; - // Ring buffer setup: 9600 samples (200ms capacity for mono 48kHz) - let rb = HeapRb::::new(9600); + // Ring buffer setup: 9600 samples (200ms capacity for mono 48kHz). + const RING_CAPACITY: usize = 9600; + let rb = HeapRb::::new(RING_CAPACITY); let (mut producer, consumer) = rb.split(); + // Prefill to the target depth so playout starts at its steady-state level: + // the mixer keeps the ring near `PLAYBACK_TARGET_SAMPLES` by reading + // `fill_gauge`, so production tracks the PipeWire hardware clock instead of + // a fixed timer. + // + // `fill_gauge` is an EXACT occupancy counter, maintained by deltas: the + // worker `fetch_add`s every pushed sample, the RT callback `fetch_sub`s + // every popped one. We can't use ringbuf's `occupied_len()` for this — the + // split producer/consumer keep *cached* head/tail indices, so their length + // observers are approximate and stale, which would feed the mixer a fill + // reading that lies high and starve the ring. (The try_push/try_pop data + // path itself is exact; only the length observers are cached.) + for _ in 0..crate::audio::PLAYBACK_TARGET_SAMPLES { + let _ = producer.try_push(0); + } + fill_gauge.store(crate::audio::PLAYBACK_TARGET_SAMPLES, Ordering::Relaxed); + + // Playout-health instrumentation (diagnostic). `underrun_samples` counts + // samples the RT callback had to substitute with silence because the ring + // was empty (clicks); `dropped_frames` counts whole frames the worker + // discarded because the ring was full (overrun). With clock-paced + // production both should stay at zero; a steady non-zero trend means the + // pacing isn't keeping up, bursts mean scheduling jitter. All RT-safe: the + // callback does a single wait-free fetch_add/store per quantum. + let underrun_samples = Arc::new(AtomicU64::new(0)); + let dropped_frames = Arc::new(AtomicU64::new(0)); + // Diagnostic: the per-cycle frame count the RT callback is asked to fill + // (`slice.len()/stride`) and how many callbacks fire per second. If the + // count is ~1024 (the graph quantum) at ~47/s the consumer is normal; if + // it's large (the buffer maxsize) we're over-pulling the whole slice + // instead of the requested quantum — the leading suspect for the ~46% + // underrun. See handoff.md 2026-05-31 PM entry. + let last_quantum = Arc::new(AtomicUsize::new(0)); + let callback_count = Arc::new(AtomicU64::new(0)); + // Command receiver to quit main loop let mainloop_clone = mainloop.clone(); let _cmd_recv = cmd_rx.attach(mainloop.loop_(), move |_| { @@ -241,6 +287,10 @@ fn run_playback(cmd_rx: pw::channel::Receiver<()>, rx: Receiver>, targe *pw::keys::MEDIA_TYPE => "Audio", *pw::keys::MEDIA_CATEGORY => "Playback", *pw::keys::MEDIA_ROLE => "Communication", + // Low-latency hint (~21ms @ 48kHz). On its own this does NOT shrink the + // buffer — the real fix is the explicit Buffers param below — but it + // expresses the intended quantum for any node that honours it. + *pw::keys::NODE_LATENCY => "1024/48000", }; if let Some(target) = target_node { props.insert("node.target", target); @@ -251,10 +301,20 @@ fn run_playback(cmd_rx: pw::channel::Receiver<()>, rx: Receiver>, targe struct PlaybackUserData> { consumer: C, + underrun_samples: Arc, + fill_gauge: Arc, + last_quantum: Arc, + callback_count: Arc, } let _listener = stream - .add_local_listener_with_user_data(PlaybackUserData { consumer }) + .add_local_listener_with_user_data(PlaybackUserData { + consumer, + underrun_samples: underrun_samples.clone(), + fill_gauge: fill_gauge.clone(), + last_quantum: last_quantum.clone(), + callback_count: callback_count.clone(), + }) .process(|stream, user_data| { if let Some(mut buffer) = stream.dequeue_buffer() { let datas = buffer.datas_mut(); @@ -264,13 +324,36 @@ fn run_playback(cmd_rx: pw::channel::Receiver<()>, rx: Receiver>, targe if let Some(slice) = data.data() { let stride = 2; // S16LE Mono = 2 bytes per frame let n_frames = slice.len() / stride; + // Diagnostic: record the quantum the device asked for and + // count the callback. Wait-free, RT-safe. + user_data.last_quantum.store(n_frames, Ordering::Relaxed); + user_data.callback_count.fetch_add(1, Ordering::Relaxed); + let mut starved = 0u64; for i in 0..n_frames { - let val = user_data.consumer.try_pop().unwrap_or(0); + let val = match user_data.consumer.try_pop() { + Some(v) => v, + None => { + starved += 1; + 0 + } + }; let bytes = val.to_le_bytes(); let start = i * stride; slice[start] = bytes[0]; slice[start + 1] = bytes[1]; } + if starved > 0 { + // One wait-free atomic add per quantum — RT-safe. + user_data.underrun_samples.fetch_add(starved, Ordering::Relaxed); + } + // Decrement the exact occupancy counter by the samples we + // actually pulled (excluding underruns, which removed + // nothing) so the mixer paces against true ring depth. + // Wait-free fetch_sub, RT-safe. + let popped = n_frames - starved as usize; + if popped > 0 { + user_data.fill_gauge.fetch_sub(popped, Ordering::Relaxed); + } total_size = n_frames * stride; } let chunk = data.chunk_mut(); @@ -301,7 +384,56 @@ fn run_playback(cmd_rx: pw::channel::Receiver<()>, rx: Receiver>, targe .0 .into_inner(); - let mut params = [Pod::from_bytes(&values).unwrap()]; + // Explicit Buffers param — THE fix for the playback crackle. Without it + // PipeWire handed this stream a ~256ms (12288-frame) maxsize buffer and + // only called `process` ~4x/sec; each callback then asked us to fill all + // 12288 frames, far more than the 200ms ring could ever hold, so ~half of + // every buffer was silence (the crackle). Pinning the buffer size to one + // graph quantum (1024 frames = 2048 bytes mono S16LE) makes the device hand + // us a ~1024-frame buffer ~47x/sec, which the ring satisfies comfortably and + // keeps `slice.len()/stride` equal to the quantum so we never over-pull. + const QUANTUM_FRAMES: i32 = 1024; + const STRIDE: i32 = 2; // S16LE mono = 2 bytes/frame + let buffers_obj = pw::spa::pod::Object { + type_: pw::spa::utils::SpaTypes::ObjectParamBuffers.as_raw(), + id: pw::spa::param::ParamType::Buffers.as_raw(), + properties: vec![ + // Let PipeWire pick the buffer count (>=2 for double-buffering). + pw::spa::pod::Property::new( + pw::spa::sys::SPA_PARAM_BUFFERS_buffers, + pw::spa::pod::Value::Choice(pw::spa::pod::ChoiceValue::Int( + pw::spa::utils::Choice( + pw::spa::utils::ChoiceFlags::empty(), + pw::spa::utils::ChoiceEnum::Range { default: 8, min: 2, max: 64 }, + ), + )), + ), + pw::spa::pod::Property::new( + pw::spa::sys::SPA_PARAM_BUFFERS_blocks, + pw::spa::pod::Value::Int(1), + ), + pw::spa::pod::Property::new( + pw::spa::sys::SPA_PARAM_BUFFERS_size, + pw::spa::pod::Value::Int(QUANTUM_FRAMES * STRIDE), + ), + pw::spa::pod::Property::new( + pw::spa::sys::SPA_PARAM_BUFFERS_stride, + pw::spa::pod::Value::Int(STRIDE), + ), + ], + }; + let buffers_values: Vec = pw::spa::pod::serialize::PodSerializer::serialize( + std::io::Cursor::new(Vec::new()), + &pw::spa::pod::Value::Object(buffers_obj), + ) + .unwrap() + .0 + .into_inner(); + + let mut params = [ + Pod::from_bytes(&values).unwrap(), + Pod::from_bytes(&buffers_values).unwrap(), + ]; stream.connect( spa::utils::Direction::Output, @@ -316,25 +448,71 @@ fn run_playback(cmd_rx: pw::channel::Receiver<()>, rx: Receiver>, targe // Spawn a worker thread to read from rx and push to producer let running = Arc::new(AtomicBool::new(true)); let running_clone = running.clone(); + let worker_dropped = dropped_frames.clone(); + let worker_fill = fill_gauge.clone(); let worker_handle = thread::spawn(move || { while running_clone.load(Ordering::Relaxed) { if let Ok(frame) = rx.recv() { - for &sample in &frame { - // Try to push. If buffer is full, drop to avoid growing latency. - if producer.try_push(sample).is_err() { - break; - } + // Drop the whole frame (rather than tearing it) only if it truly + // won't fit — checked against the exact occupancy counter, not + // the ring's stale length observer. With clock-paced production + // this should never fire. + if worker_fill.load(Ordering::Relaxed) + frame.len() > RING_CAPACITY { + worker_dropped.fetch_add(1, Ordering::Relaxed); + continue; } + for &sample in &frame { + let _ = producer.try_push(sample); + } + worker_fill.fetch_add(frame.len(), Ordering::Relaxed); } else { return; } } }); + // Diagnostic logger: once per second, report the ring fill and the delta in + // underrun samples / dropped frames since the last report. Quiet line (all + // zeros) means the local playout path is healthy; a steady non-zero trend is + // the clock-drift signature, bursts are scheduling jitter. + let log_running = running.clone(); + let log_underrun = underrun_samples.clone(); + let log_dropped = dropped_frames.clone(); + let log_fill = fill_gauge.clone(); + let log_quantum = last_quantum.clone(); + let log_callbacks = callback_count.clone(); + // In normal operation this stays quiet — it only logs a second where the + // playout actually glitched (underrun or dropped > 0). Set + // PEERSPEAK_AUDIO_VERBOSE=1 (the `audio_probe` tool does) to get the full + // once-per-second heartbeat for diagnostics. + let verbose = std::env::var_os("PEERSPEAK_AUDIO_VERBOSE").is_some(); + let logger_handle = thread::spawn(move || { + let (mut last_u, mut last_d, mut last_c) = (0u64, 0u64, 0u64); + while log_running.load(Ordering::Relaxed) { + thread::sleep(Duration::from_secs(1)); + let u = log_underrun.load(Ordering::Relaxed); + let d = log_dropped.load(Ordering::Relaxed); + let fill = log_fill.load(Ordering::Relaxed); + let q = log_quantum.load(Ordering::Relaxed); + let c = log_callbacks.load(Ordering::Relaxed); + let (du, dd, dc) = (u - last_u, d - last_d, c - last_c); + last_u = u; + last_d = d; + last_c = c; + if verbose || du > 0 || dd > 0 { + crate::log_msg(&format!( + "playout-health: fill={fill} samples (~{}ms) | underrun +{du} samples/s (total {u}) | dropped +{dd} frames/s (total {d}) | quantum={q} frames, {dc} callbacks/s", + fill / 48, + )); + } + } + }); + mainloop.run(); running.store(false, Ordering::Relaxed); let _ = worker_handle.join(); + let _ = logger_handle.join(); Ok(()) } diff --git a/src/bin/audio_probe.rs b/src/bin/audio_probe.rs new file mode 100644 index 0000000..240179d --- /dev/null +++ b/src/bin/audio_probe.rs @@ -0,0 +1,119 @@ +//! Audio playout diagnostic probe. +//! +//! Drives a phase-continuous sine tone through the *real* PipeWire playback path +//! (`PipeWireBackend::start_playback`), using the *same* fill-paced production +//! the production mixer uses (`core/mod.rs`): generate a frame only while the +//! playback ring is below `PLAYBACK_TARGET_SAMPLES`, so production tracks the +//! PipeWire hardware clock. No network, no microphone — this isolates the local +//! output path so we can confirm the clock-paced playout is glitch-free. +//! +//! Use your ears on the tone (any click/pop is a glitch) together with the +//! `playout-health:` lines tailed to stdout: +//! - all-zero health lines + clean tone → local playout is healthy; any +//! crackle on real calls is upstream (per-peer jitter buffer), not this ring. +//! - steady `underrun +N samples/s` (or steady `dropped +N frames/s`) with a +//! slowly drifting `fill` → clock drift (producer vs hardware clock). +//! - random bursts correlated with system load → scheduling jitter. +//! +//! Run: cargo run --bin audio_probe -- [freq_hz] [seconds] [target_node] +//! e.g. cargo run --release --bin audio_probe -- 440 30 + +use std::io::{BufRead, BufReader, Seek, SeekFrom}; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::mpsc; +use std::time::Duration; + +use peerspeak::audio::AudioBackend; +use peerspeak::audio::pipewire_impl::PipeWireBackend; +use peerspeak::core::jitter::FRAME_SAMPLES; // 960 samples = 20ms @ 48kHz mono + +const SAMPLE_RATE: f32 = 48_000.0; + +#[tokio::main] +async fn main() { + let mut args = std::env::args().skip(1); + let freq: f32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(440.0); + let secs: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(30); + let target_node: Option = args.next(); + + // The playout-health logger is quiet in normal operation (it only logs + // glitches); ask it for the full once-per-second heartbeat so the probe can + // show the steady-state numbers. + // SAFETY: set before any playback thread starts, so no concurrent env read. + unsafe { std::env::set_var("PEERSPEAK_AUDIO_VERBOSE", "1") }; + + println!("audio_probe: {freq} Hz tone for {secs}s through the real playback path."); + println!("Listen for clicks/pops; watch the playout-health lines below.\n"); + + // Tail the app log (where playout-health lines land) to stdout in the + // background so it's all in one terminal. + spawn_log_tailer(); + + let backend = PipeWireBackend::new(); + let (tx, rx) = mpsc::channel::>(); + let ring_fill = Arc::new(AtomicUsize::new(0)); + if let Err(e) = backend.start_playback(rx, target_node, ring_fill.clone()) { + eprintln!("failed to start playback: {e}"); + return; + } + + // Phase-continuous sine, generated one 20ms frame at a time, fill-paced + // exactly like the production mixer: only produce while the ring is below + // target, so production tracks the PipeWire hardware clock. + use std::sync::atomic::Ordering; + let deadline = tokio::time::Instant::now() + Duration::from_secs(secs); + let mut n: u64 = 0; // running sample index keeps phase continuous across frames + while tokio::time::Instant::now() < deadline { + if ring_fill.load(Ordering::Relaxed) >= peerspeak::audio::PLAYBACK_TARGET_SAMPLES { + tokio::time::sleep(Duration::from_millis(2)).await; + continue; + } + let mut frame = Vec::with_capacity(FRAME_SAMPLES); + for _ in 0..FRAME_SAMPLES { + let t = n as f32 / SAMPLE_RATE; + // 0.25 amplitude: clearly audible but not harsh. + let sample = (0.25 * i16::MAX as f32 * (2.0 * std::f32::consts::PI * freq * t).sin()) as i16; + frame.push(sample); + n += 1; + } + if tx.send(frame).is_err() { + eprintln!("playback channel closed early"); + break; + } + } + + // Let the ring drain, then stop. + tokio::time::sleep(Duration::from_millis(300)).await; + let _ = backend.stop(); + println!("\naudio_probe: done."); +} + +/// Open the app log, seek to the end, and echo new lines (the `playout-health:` +/// reports) to stdout once they appear. +fn spawn_log_tailer() { + let path = peerspeak::log_file_path(); + std::thread::spawn(move || { + // Wait for the file to exist (first log_msg creates it). + let file = loop { + if let Ok(f) = std::fs::File::open(&path) { + break f; + } + std::thread::sleep(Duration::from_millis(100)); + }; + let mut reader = BufReader::new(file); + let _ = reader.seek(SeekFrom::End(0)); + loop { + let mut line = String::new(); + match reader.read_line(&mut line) { + Ok(0) => std::thread::sleep(Duration::from_millis(150)), + Ok(_) => { + if line.contains("playout-health:") { + print!("{line}"); + } + } + Err(_) => std::thread::sleep(Duration::from_millis(150)), + } + } + }); +} diff --git a/src/core/mod.rs b/src/core/mod.rs index 5e3c808..91af23a 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -17,7 +17,7 @@ use iroh_gossip::net::Gossip; use tokio::sync::{mpsc, Mutex}; use std::collections::HashMap; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::time::Duration; pub struct CoreController { @@ -213,7 +213,11 @@ async fn run_core_loop( continue; } - if let Err(e) = audio_backend.start_playback(playback_rx, output_device) { + // Shared gauge: PipeWire publishes the playback ring's live depth + // here (drain side + fill side); the mixer reads it to pace + // production to the hardware clock instead of a fixed timer. + let ring_fill = Arc::new(AtomicUsize::new(0)); + if let Err(e) = audio_backend.start_playback(playback_rx, output_device, ring_fill.clone()) { let _ = ui_tx.send(UiEvent::Error(format!("Failed to start playback: {}", e))).await; let _ = audio_backend.stop(); let _ = room_state.leave().await; @@ -314,28 +318,41 @@ async fn run_core_loop( } }); - // 3. Mixing & level extraction loop task. Every 20ms, pull one - // concealed frame per peer from its jitter buffer, apply - // per-peer volume, sum, and hand the mix to playback. + // 3. Mixing & level extraction loop task. Production is paced by + // the playback ring's fill level (the PipeWire hardware clock), + // NOT a fixed software timer: we produce a 20ms frame only when + // the ring is below its target depth, so the long-run mix rate + // auto-matches the device drain rate and the producer/consumer + // beat (which otherwise churns ~20% of audio) disappears. Each + // produced frame pulls one concealed frame per peer from its + // jitter buffer, applies per-peer volume, and sums. let jitter_mixer = jitter.clone(); let is_deafened_clone = is_deafened.clone(); let peer_volumes_mixer = peer_volumes.clone(); let ui_tx_mixer = ui_tx.clone(); + let ring_fill_mixer = ring_fill.clone(); let mixer_task = tokio::spawn(async move { - let mut interval = tokio::time::interval(Duration::from_millis(20)); - interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + // When the ring is at/above target we have nothing to do; nap + // briefly and re-check. Short enough (relative to the ~60ms + // target and ~21ms device quantum) that we always refill well + // before the ring can run dry. + const IDLE_NAP: Duration = Duration::from_millis(2); - // The 20ms mix cadence is fixed by playback, but pushing a - // level event every tick floods the UI runtime at ~50/sec. We - // peak-hold per-peer levels across this many ticks and emit - // once per window (~10/sec) — peak-hold so a brief transient - // inside the window still lights the speaking indicator. - const LEVEL_EMIT_TICKS: u32 = 5; + // Pushing a level event per frame floods the UI runtime at + // ~50/sec. We peak-hold per-peer levels across this many + // produced frames and emit once per window (~10/sec) — + // peak-hold so a brief transient still lights the indicator. + const LEVEL_EMIT_FRAMES: u32 = 5; let mut level_peaks: HashMap = HashMap::new(); - let mut ticks_since_emit: u32 = 0; + let mut frames_since_emit: u32 = 0; loop { - interval.tick().await; + // Pace to the hardware clock: only produce while the ring + // is draining below target. Otherwise yield and re-check. + if ring_fill_mixer.load(Ordering::Relaxed) >= crate::audio::PLAYBACK_TARGET_SAMPLES { + tokio::time::sleep(IDLE_NAP).await; + continue; + } let current_volumes = peer_volumes_mixer.lock().await.clone(); let mut peer_frames = Vec::new(); @@ -389,11 +406,11 @@ async fn run_core_loop( } // Emit coalesced peaks once per window, then reset. - ticks_since_emit += 1; - if ticks_since_emit >= LEVEL_EMIT_TICKS { + frames_since_emit += 1; + if frames_since_emit >= LEVEL_EMIT_FRAMES { let levels: Vec<(EndpointId, f32)> = level_peaks.drain().collect(); let _ = ui_tx_mixer.send(UiEvent::AudioLevels(levels)).await; - ticks_since_emit = 0; + frames_since_emit = 0; } } }); diff --git a/src/lib.rs b/src/lib.rs index 801729c..5007d6c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24,6 +24,12 @@ fn log_path() -> &'static PathBuf { }) } +/// The resolved log file path. Exposed so diagnostic tools (e.g. the audio +/// probe) can tail the same log the app writes to. +pub fn log_file_path() -> PathBuf { + log_path().clone() +} + pub fn log_msg(msg: &str) { if let Ok(mut file) = std::fs::OpenOptions::new() .create(true)