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<AtomicUsize>) 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 <noreply@anthropic.com>
This commit is contained in:
+24
-1
@@ -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<Vec<i16>>, target_node: Option<String>) -> 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<Vec<i16>>,
|
||||
target_node: Option<String>,
|
||||
ring_fill: Arc<AtomicUsize>,
|
||||
) -> Result<(), AudioError>;
|
||||
|
||||
/// Stops both capture and playback streams.
|
||||
fn stop(&self) -> Result<(), AudioError>;
|
||||
|
||||
+192
-14
@@ -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<Vec<i16>>, target_node: Option<String>) -> Result<(), AudioError> {
|
||||
fn start_playback(
|
||||
&self,
|
||||
rx: Receiver<Vec<i16>>,
|
||||
target_node: Option<String>,
|
||||
ring_fill: Arc<AtomicUsize>,
|
||||
) -> 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<Vec<i16>>, target_n
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn run_playback(cmd_rx: pw::channel::Receiver<()>, rx: Receiver<Vec<i16>>, target_node: Option<String>) -> Result<(), AudioError> {
|
||||
fn run_playback(
|
||||
cmd_rx: pw::channel::Receiver<()>,
|
||||
rx: Receiver<Vec<i16>>,
|
||||
target_node: Option<String>,
|
||||
fill_gauge: Arc<AtomicUsize>,
|
||||
) -> 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<Vec<i16>>, 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::<i16>::new(9600);
|
||||
// Ring buffer setup: 9600 samples (200ms capacity for mono 48kHz).
|
||||
const RING_CAPACITY: usize = 9600;
|
||||
let rb = HeapRb::<i16>::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<Vec<i16>>, 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<Vec<i16>>, targe
|
||||
|
||||
struct PlaybackUserData<C: ringbuf::traits::Consumer<Item = i16>> {
|
||||
consumer: C,
|
||||
underrun_samples: Arc<AtomicU64>,
|
||||
fill_gauge: Arc<AtomicUsize>,
|
||||
last_quantum: Arc<AtomicUsize>,
|
||||
callback_count: Arc<AtomicU64>,
|
||||
}
|
||||
|
||||
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<Vec<i16>>, 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<Vec<i16>>, 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<u8> = 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<Vec<i16>>, 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(())
|
||||
}
|
||||
|
||||
@@ -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<String> = 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::<Vec<i16>>();
|
||||
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)),
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
+35
-18
@@ -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<EndpointId, f32> = 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;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user