molluskandClaude Opus 5
bcf1343a55
Zero behaviour change. This is what makes the screenshare exclusion engine able to see us at all (plan §5.1, impl plan §3): pixelpass must refuse to fan out our own playback, and until now it had no way to recognise it. Two carriers, matched by pixelpass as a union — `peerspeak.owned=1` and a `node.name` prefix `peerspeak_owned_<role>_<pid>`. Round 8 added the second after the phase-5 audit found a node property is invisible to the PipeWire registry `global` event and recoverable only by binding the node; the prefix is announced directly. A union is also the fail-closed direction: a missed tag leaks call audio into a share, a spurious one only over-excludes. Three tagging sites, all three verified live on this host: - native call playback → props on the stream dict - screenshare mpv/VLC → PULSE_PROP + PIPEWIRE_PROPS on the child - notification chimes → same, on pw-play/paplay The literals are a cross-repo wire contract, so they appear once here as named constants and are pinned in a fixture committed byte-identical in both repos (tests/fixtures/ownership-tag-contract.txt). The contract test is black-box: it builds a real child `Command` and reads back the environment it would carry, rather than testing our own formatter. Three live `#[ignore]`d exit-gate tests drive the real call sites and poll `pw-dump` for the resulting node — the plan requires the tag be shown landing on a live node, not just in the env. All three mutation-verified (drop either carrier, or the role, and the matching gate fails). Measured while verifying: mpv, VLC, pw-play and paplay all honour `node.name` from those env vars. The native stream set neither `application.name` nor a description, so a mixer fell back to `node.name` — which the tag turns into an internal identifier. Added an explicit `node.description = "PeerSpeak"` there, which keeps the plan's rule (the prefix must not reach `node.description`) while preserving its intent: mixers stay readable. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
829 lines
34 KiB
Rust
829 lines
34 KiB
Rust
use crate::audio::ownership;
|
|
use crate::audio::{AudioBackend, AudioError};
|
|
use pipewire as pw;
|
|
use pw::{properties::properties, spa};
|
|
use ringbuf::{
|
|
HeapRb,
|
|
traits::{Consumer, Producer, Split},
|
|
};
|
|
use spa::pod::Pod;
|
|
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
|
|
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender};
|
|
use std::sync::{Arc, Mutex};
|
|
use std::thread::{self, JoinHandle};
|
|
use std::time::Duration;
|
|
|
|
pub struct PipeWireBackend {
|
|
capture_state: Mutex<Option<CaptureState>>,
|
|
playback_state: Mutex<Option<PlaybackState>>,
|
|
}
|
|
|
|
struct CaptureState {
|
|
cmd_tx: pw::channel::Sender<()>,
|
|
thread: JoinHandle<()>,
|
|
}
|
|
|
|
struct PlaybackState {
|
|
cmd_tx: pw::channel::Sender<()>,
|
|
thread: JoinHandle<()>,
|
|
}
|
|
|
|
impl Default for PipeWireBackend {
|
|
fn default() -> Self {
|
|
Self::new()
|
|
}
|
|
}
|
|
|
|
impl PipeWireBackend {
|
|
pub fn new() -> Self {
|
|
pw::init();
|
|
Self {
|
|
capture_state: Mutex::new(None),
|
|
playback_state: Mutex::new(None),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl AudioBackend for PipeWireBackend {
|
|
fn start_capture(
|
|
&self,
|
|
tx: Sender<Vec<i16>>,
|
|
target_node: Option<String>,
|
|
) -> Result<(), AudioError> {
|
|
let mut capture_guard = self.capture_state.lock().unwrap();
|
|
if capture_guard.is_some() {
|
|
return Err(AudioError::Stream("Capture already started".to_string()));
|
|
}
|
|
|
|
let (cmd_tx, cmd_rx) = pw::channel::channel::<()>();
|
|
let tx_clone = tx.clone();
|
|
|
|
let thread = thread::Builder::new()
|
|
.name("peerspeak-capture".to_string())
|
|
.spawn(move || {
|
|
if let Err(e) = run_capture(cmd_rx, tx_clone, target_node) {
|
|
eprintln!("Capture thread error: {:?}", e);
|
|
}
|
|
})
|
|
.map_err(|e| AudioError::Init(e.to_string()))?;
|
|
|
|
*capture_guard = Some(CaptureState { cmd_tx, thread });
|
|
Ok(())
|
|
}
|
|
|
|
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()));
|
|
}
|
|
|
|
let (cmd_tx, cmd_rx) = pw::channel::channel::<()>();
|
|
|
|
let thread = thread::Builder::new()
|
|
.name("peerspeak-playback".to_string())
|
|
.spawn(move || {
|
|
if let Err(e) = run_playback(cmd_rx, rx, target_node, ring_fill) {
|
|
eprintln!("Playback thread error: {:?}", e);
|
|
}
|
|
})
|
|
.map_err(|e| AudioError::Init(e.to_string()))?;
|
|
|
|
*playback_guard = Some(PlaybackState { cmd_tx, thread });
|
|
Ok(())
|
|
}
|
|
|
|
fn stop(&self) -> Result<(), AudioError> {
|
|
// Stop capture
|
|
let mut capture_guard = self.capture_state.lock().unwrap();
|
|
if let Some(state) = capture_guard.take() {
|
|
let _ = state.cmd_tx.send(());
|
|
let _ = state.thread.join();
|
|
}
|
|
|
|
// Stop playback
|
|
let mut playback_guard = self.playback_state.lock().unwrap();
|
|
if let Some(state) = playback_guard.take() {
|
|
let _ = state.cmd_tx.send(());
|
|
let _ = state.thread.join();
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
fn run_capture(
|
|
cmd_rx: pw::channel::Receiver<()>,
|
|
tx: Sender<Vec<i16>>,
|
|
target_node: Option<String>,
|
|
) -> 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)
|
|
.map_err(|e| AudioError::Init(e.to_string()))?;
|
|
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);
|
|
let (producer, mut consumer) = rb.split();
|
|
|
|
// Command receiver to quit main loop
|
|
let mainloop_clone = mainloop.clone();
|
|
let _cmd_recv = cmd_rx.attach(mainloop.loop_(), move |_| {
|
|
mainloop_clone.quit();
|
|
});
|
|
|
|
let mut props = properties! {
|
|
*pw::keys::MEDIA_TYPE => "Audio",
|
|
*pw::keys::MEDIA_CATEGORY => "Capture",
|
|
*pw::keys::MEDIA_ROLE => "Communication",
|
|
};
|
|
if let Some(target) = target_node {
|
|
props.insert("node.target", target);
|
|
}
|
|
|
|
let stream = pw::stream::StreamBox::new(&core, "peerspeak-capture-stream", props)
|
|
.map_err(|e| AudioError::Init(e.to_string()))?;
|
|
|
|
struct CaptureUserData<P: ringbuf::traits::Producer<Item = i16>> {
|
|
producer: P,
|
|
}
|
|
|
|
let _listener = stream
|
|
.add_local_listener_with_user_data(CaptureUserData { producer })
|
|
.process(|stream, user_data| {
|
|
if let Some(mut buffer) = stream.dequeue_buffer() {
|
|
let datas = buffer.datas_mut();
|
|
if !datas.is_empty() {
|
|
let data = &mut datas[0];
|
|
let size = data.chunk().size() as usize;
|
|
if let Some(slice) = data.data() {
|
|
for_each_capture_sample(slice, size, |sample| {
|
|
let _ = user_data.producer.try_push(sample);
|
|
});
|
|
}
|
|
}
|
|
}
|
|
})
|
|
.register()
|
|
.map_err(|e| AudioError::Init(e.to_string()))?;
|
|
|
|
let mut audio_info = spa::param::audio::AudioInfoRaw::new();
|
|
audio_info.set_format(spa::param::audio::AudioFormat::S16LE);
|
|
audio_info.set_rate(48000);
|
|
audio_info.set_channels(1); // Mono
|
|
|
|
let obj = pw::spa::pod::Object {
|
|
type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(),
|
|
id: pw::spa::param::ParamType::EnumFormat.as_raw(),
|
|
properties: audio_info.into(),
|
|
};
|
|
let values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize(
|
|
std::io::Cursor::new(Vec::new()),
|
|
&pw::spa::pod::Value::Object(obj),
|
|
)
|
|
.unwrap()
|
|
.0
|
|
.into_inner();
|
|
|
|
let mut params = [Pod::from_bytes(&values).unwrap()];
|
|
|
|
stream
|
|
.connect(
|
|
spa::utils::Direction::Input,
|
|
None,
|
|
pw::stream::StreamFlags::AUTOCONNECT
|
|
| pw::stream::StreamFlags::MAP_BUFFERS
|
|
| pw::stream::StreamFlags::RT_PROCESS,
|
|
&mut params,
|
|
)
|
|
.map_err(|e| AudioError::Stream(e.to_string()))?;
|
|
|
|
// Spawn the worker thread to pop from consumer and send Vec<i16> frames
|
|
let running = Arc::new(AtomicBool::new(true));
|
|
let running_clone = running.clone();
|
|
let worker_handle = thread::spawn(move || {
|
|
let mut frame = Vec::with_capacity(960);
|
|
while running_clone.load(Ordering::Relaxed) {
|
|
let mut popped = false;
|
|
while let Some(sample) = consumer.try_pop() {
|
|
popped = true;
|
|
frame.push(sample);
|
|
if frame.len() == 960 {
|
|
if tx.send(frame).is_err() {
|
|
return;
|
|
}
|
|
frame = Vec::with_capacity(960);
|
|
}
|
|
}
|
|
if !popped {
|
|
thread::sleep(Duration::from_millis(5));
|
|
}
|
|
}
|
|
});
|
|
|
|
mainloop.run();
|
|
|
|
running.store(false, Ordering::Relaxed);
|
|
let _ = worker_handle.join();
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Visit the complete S16LE samples in the portion PipeWire reports as filled.
|
|
/// Clamp the reported byte count to the mapped slice before indexing: a bad
|
|
/// chunk size must not panic from the realtime capture callback.
|
|
fn for_each_capture_sample(slice: &[u8], size: usize, mut visit: impl FnMut(i16)) {
|
|
let size = size.min(slice.len());
|
|
for chunk in slice[..size].chunks_exact(2) {
|
|
visit(i16::from_le_bytes([chunk[0], chunk[1]]));
|
|
}
|
|
}
|
|
|
|
/// Frames the playback RT callback should produce this cycle.
|
|
///
|
|
/// `requested` is the graph's per-cycle quantum from `Buffer::requested()` (0 if
|
|
/// the graph didn't set it — first buffers / non-driver cycles). `mapped_frames`
|
|
/// is how many frames the mapped buffer slice can physically hold.
|
|
///
|
|
/// Honoring the requested quantum is what makes playback correct on any graph
|
|
/// quantum, instead of the old hard-pinned 1024 (which crackled when a machine's
|
|
/// `clock.quantum` wasn't 1024). Rules:
|
|
/// - never write past the mapped slice (clamp to `mapped_frames`);
|
|
/// - when `requested == 0`, fall back to a safe cap (≤ one 1024-frame quantum) —
|
|
/// never the whole slice, because over-pulling past the ring depth is exactly
|
|
/// the sustained crackle this replaces. Under-filling a cycle is a harmless
|
|
/// brief glitch (PipeWire pads).
|
|
// How often the playback worker wakes to re-check its `running` flag while idle,
|
|
// bounding how long `stop` can block joining the worker (bug A7). A plain
|
|
// blocking receive would never re-check the flag, waking only on a new frame or
|
|
// the sender dropping, so shutdown would hang if the sender outlives `stop`.
|
|
const WORKER_POLL: Duration = Duration::from_millis(100);
|
|
|
|
/// Pump frames from `rx` to `on_frame` until `running` goes false or the sender
|
|
/// disconnects. Uses a timed receive so the loop re-checks `running` at least
|
|
/// every `WORKER_POLL` even when no frames arrive — this is what lets `stop()`
|
|
/// join the worker promptly instead of hanging on a parked blocking `recv()`
|
|
/// (bug A7). Pure w.r.t. its inputs (no PipeWire), so it's unit-testable.
|
|
fn drain_loop(rx: &Receiver<Vec<i16>>, running: &AtomicBool, mut on_frame: impl FnMut(Vec<i16>)) {
|
|
while running.load(Ordering::Relaxed) {
|
|
match rx.recv_timeout(WORKER_POLL) {
|
|
Ok(frame) => on_frame(frame),
|
|
Err(RecvTimeoutError::Timeout) => continue, // re-check `running`
|
|
Err(RecvTimeoutError::Disconnected) => return, // sender gone for good
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Reserve exact occupancy before making a frame visible to the consumer.
|
|
/// `after_reserve` is empty in production and lets the regression test force a
|
|
/// consumer interleaving at the critical ordering boundary.
|
|
fn publish_frame<P: Producer<Item = i16>>(
|
|
fill: &AtomicUsize,
|
|
dropped: &AtomicU64,
|
|
producer: &mut P,
|
|
frame: &[i16],
|
|
after_reserve: impl FnOnce(),
|
|
) {
|
|
fill.fetch_add(frame.len(), Ordering::Relaxed);
|
|
after_reserve();
|
|
let pushed = producer.push_slice(frame);
|
|
if pushed != frame.len() {
|
|
fill.fetch_sub(frame.len() - pushed, Ordering::Relaxed);
|
|
dropped.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
}
|
|
|
|
fn frames_to_produce(requested: usize, mapped_frames: usize) -> usize {
|
|
/// Safe per-cycle fallback when the graph doesn't report a quantum.
|
|
const FALLBACK_FRAMES: usize = 1024;
|
|
let want = if requested > 0 {
|
|
requested
|
|
} else {
|
|
FALLBACK_FRAMES
|
|
};
|
|
want.min(mapped_frames)
|
|
}
|
|
|
|
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)
|
|
.map_err(|e| AudioError::Init(e.to_string()))?;
|
|
let core = context
|
|
.connect_rc(None)
|
|
.map_err(|e| AudioError::Init(e.to_string()))?;
|
|
|
|
// Ring buffer setup: 19200 interleaved samples (200ms capacity for stereo
|
|
// 48kHz).
|
|
const RING_CAPACITY: usize = 9600 * crate::audio::PLAYBACK_CHANNELS;
|
|
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 |_| {
|
|
mainloop_clone.quit();
|
|
});
|
|
|
|
// Ownership tag, both carriers (`crate::audio::ownership`, plan §5.1).
|
|
// This is the node that carries the far end's voice, so it is the single
|
|
// most important thing for pixelpass to refuse to fan out: sharing it
|
|
// would send the call back to the person already speaking on it.
|
|
let owned_node_name = ownership::owned_node_name(ownership::NATIVE_PLAYBACK_ROLE);
|
|
let mut props = properties! {
|
|
*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",
|
|
ownership::OWNED_PROP_KEY => ownership::OWNED_PROP_VALUE,
|
|
// Set explicitly rather than relying on the stream name passed to
|
|
// `StreamBox::new` below: props win over that name, and this one has
|
|
// to be exact.
|
|
*pw::keys::NODE_NAME => owned_node_name.as_str(),
|
|
// Measured: this stream sets neither `application.name` nor a
|
|
// description, so a mixer falls back to `node.name` — which the line
|
|
// above just turned into an internal identifier. The plan's rule is
|
|
// that the ownership prefix must not reach `node.description`; a
|
|
// human label there is what keeps that rule's *intent* (mixers stay
|
|
// readable) true for our own stream, exactly as mpv's own
|
|
// description does for the spawned players.
|
|
*pw::keys::NODE_DESCRIPTION => "PeerSpeak",
|
|
};
|
|
if let Some(target) = target_node {
|
|
props.insert("node.target", target);
|
|
}
|
|
|
|
let stream = pw::stream::StreamBox::new(&core, "peerspeak-playback-stream", props)
|
|
.map_err(|e| AudioError::Init(e.to_string()))?;
|
|
|
|
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,
|
|
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() {
|
|
// Read the graph's requested quantum BEFORE the mutable
|
|
// `datas_mut()` borrow below (`requested()` borrows `&buffer`).
|
|
let requested = buffer.requested() as usize;
|
|
let datas = buffer.datas_mut();
|
|
if !datas.is_empty() {
|
|
let data = &mut datas[0];
|
|
let mut total_size = 0;
|
|
if let Some(slice) = data.data() {
|
|
let stride = 2 * crate::audio::PLAYBACK_CHANNELS; // S16LE stereo
|
|
// Fill exactly what the graph asked for this cycle (with
|
|
// a safe fallback), never the whole mapped slice — that
|
|
// over-pull past the ring depth was the original crackle.
|
|
let mapped_frames = slice.len() / stride;
|
|
let n_frames = frames_to_produce(requested, mapped_frames);
|
|
// 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 start = i * stride;
|
|
for ch in 0..crate::audio::PLAYBACK_CHANNELS {
|
|
let val = match user_data.consumer.try_pop() {
|
|
Some(v) => v,
|
|
None => {
|
|
starved += 1;
|
|
0
|
|
}
|
|
};
|
|
let bytes = val.to_le_bytes();
|
|
let offset = start + ch * 2;
|
|
slice[offset] = bytes[0];
|
|
slice[offset + 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 requested_samples = n_frames * crate::audio::PLAYBACK_CHANNELS;
|
|
let popped = requested_samples - 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();
|
|
*chunk.offset_mut() = 0;
|
|
*chunk.stride_mut() = (2 * crate::audio::PLAYBACK_CHANNELS) as _;
|
|
*chunk.size_mut() = total_size as _;
|
|
}
|
|
}
|
|
})
|
|
.register()
|
|
.map_err(|e| AudioError::Init(e.to_string()))?;
|
|
|
|
let mut audio_info = spa::param::audio::AudioInfoRaw::new();
|
|
audio_info.set_format(spa::param::audio::AudioFormat::S16LE);
|
|
audio_info.set_rate(48000);
|
|
audio_info.set_channels(crate::audio::PLAYBACK_CHANNELS as u32); // Stereo playback
|
|
|
|
let obj = pw::spa::pod::Object {
|
|
type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(),
|
|
id: pw::spa::param::ParamType::EnumFormat.as_raw(),
|
|
properties: audio_info.into(),
|
|
};
|
|
let values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize(
|
|
std::io::Cursor::new(Vec::new()),
|
|
&pw::spa::pod::Value::Object(obj),
|
|
)
|
|
.unwrap()
|
|
.0
|
|
.into_inner();
|
|
|
|
// Explicit Buffers param. History: without ANY Buffers param PipeWire handed
|
|
// this stream a ~256ms (12288-frame) maxsize buffer and called `process`
|
|
// only ~4x/sec; the callback then over-pulled the whole 12288-frame slice
|
|
// (far more than the 200ms ring holds → ~half silence → the crackle). The
|
|
// first fix pinned the size to exactly one 1024-frame quantum, which only
|
|
// works when the graph quantum IS 1024 — on other hardware (quantum 512 or
|
|
// 2048) the pinned slice mismatches the device's per-cycle demand and the
|
|
// crackle returns. The correct fix: give the buffer enough room for any
|
|
// plausible quantum (the desktop's quantum-limit is 8192) and let the RT
|
|
// callback fill exactly `Buffer::requested()` frames per cycle (see
|
|
// `frames_to_produce`). `requested()`, not the buffer size, now governs
|
|
// per-cycle output, so this is a generous max rather than a hard pin.
|
|
const MAX_QUANTUM_FRAMES: i32 = 8192;
|
|
const STRIDE: i32 = 2 * crate::audio::PLAYBACK_CHANNELS as i32; // S16LE stereo
|
|
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(MAX_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,
|
|
None,
|
|
pw::stream::StreamFlags::AUTOCONNECT
|
|
| pw::stream::StreamFlags::MAP_BUFFERS
|
|
| pw::stream::StreamFlags::RT_PROCESS,
|
|
&mut params,
|
|
)
|
|
.map_err(|e| AudioError::Stream(e.to_string()))?;
|
|
|
|
// 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 || {
|
|
drain_loop(&rx, &running_clone, |frame| {
|
|
// 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);
|
|
return;
|
|
}
|
|
// Reserve occupancy BEFORE publishing samples. Otherwise the RT
|
|
// consumer can pop a newly-visible sample before it is counted and
|
|
// wrap the exact fill gauge to usize::MAX, wedging mixer pacing.
|
|
// `push_slice` also publishes the frame as one operation rather than
|
|
// exposing a half-written stereo pair.
|
|
publish_frame(&worker_fill, &worker_dropped, &mut producer, &frame, || {});
|
|
});
|
|
});
|
|
|
|
// 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 * crate::audio::PLAYBACK_CHANNELS),
|
|
));
|
|
}
|
|
}
|
|
});
|
|
|
|
mainloop.run();
|
|
|
|
running.store(false, Ordering::Relaxed);
|
|
let _ = worker_handle.join();
|
|
let _ = logger_handle.join();
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{drain_loop, for_each_capture_sample, frames_to_produce, publish_frame};
|
|
use ringbuf::{
|
|
HeapRb,
|
|
traits::{Consumer, Producer, Split},
|
|
};
|
|
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
|
|
use std::sync::{Arc, Mutex};
|
|
use std::time::Duration;
|
|
use std::{sync::mpsc, thread};
|
|
|
|
/// Phase-1 exit gate, native-playback half (impl plan §3): the stream
|
|
/// that carries the far end's voice appears on the graph with **both**
|
|
/// ownership carriers, and still with the `Communication` media role.
|
|
///
|
|
/// The third and most important of the three tagged paths — this is the
|
|
/// node whose audio, if fanned out, would send the call back to whoever
|
|
/// is speaking on it.
|
|
///
|
|
/// Feeds silence, so the gate is inaudible. Live: needs PipeWire and
|
|
/// `pw-dump`. `cargo test --lib -- --ignored native_playback`
|
|
#[test]
|
|
#[ignore = "live: requires a running PipeWire daemon and pw-dump"]
|
|
fn native_playback_node_carries_both_ownership_carriers() {
|
|
use crate::audio::ownership::{self, live_test};
|
|
use crate::audio::{AudioBackend, PLAYBACK_TARGET_SAMPLES};
|
|
|
|
let backend = super::PipeWireBackend::new();
|
|
let (tx, rx) = mpsc::channel::<Vec<i16>>();
|
|
let ring_fill = Arc::new(AtomicUsize::new(0));
|
|
backend
|
|
.start_playback(rx, None, ring_fill.clone())
|
|
.expect("playback starts");
|
|
|
|
// Keep the ring fed so the node stays live for the whole poll; the
|
|
// stream is created on connect, but a starved one is not a fair test
|
|
// of what a real call looks like on the graph.
|
|
let feeder = thread::spawn(move || {
|
|
let silence = vec![0i16; 960 * 2];
|
|
for _ in 0..300 {
|
|
if ring_fill.load(Ordering::Relaxed) < PLAYBACK_TARGET_SAMPLES
|
|
&& tx.send(silence.clone()).is_err()
|
|
{
|
|
return;
|
|
}
|
|
thread::sleep(Duration::from_millis(20));
|
|
}
|
|
});
|
|
|
|
let prefix = live_test::expected_prefix(ownership::NATIVE_PLAYBACK_ROLE);
|
|
let found = live_test::poll_for_owned_node(&prefix, Duration::from_secs(5));
|
|
let _ = backend.stop();
|
|
let _ = feeder.join();
|
|
|
|
let (name, owned) =
|
|
found.unwrap_or_else(|| panic!("no live node named {prefix:?} appeared within 5s"));
|
|
assert!(
|
|
name.starts_with(ownership::OWNED_NODE_NAME_PREFIX),
|
|
"{name}"
|
|
);
|
|
assert_eq!(
|
|
owned.as_deref(),
|
|
Some(ownership::OWNED_PROP_VALUE),
|
|
"carrier 1 must be on the live node, not just carrier 2"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn requested_in_range_is_honored() {
|
|
// The graph's requested quantum is produced verbatim when it fits.
|
|
assert_eq!(frames_to_produce(512, 8192), 512);
|
|
assert_eq!(frames_to_produce(1024, 8192), 1024);
|
|
assert_eq!(frames_to_produce(2048, 8192), 2048);
|
|
}
|
|
|
|
#[test]
|
|
fn requested_over_mapped_is_clamped() {
|
|
// Never write past the mapped slice, even if the graph asks for more.
|
|
assert_eq!(frames_to_produce(8192, 1024), 1024);
|
|
assert_eq!(frames_to_produce(2048, 2048), 2048);
|
|
}
|
|
|
|
#[test]
|
|
fn zero_requested_uses_safe_fallback() {
|
|
// requested()==0 (first buffers / non-driver cycles) → capped fallback,
|
|
// never the whole mapped slice (that's the over-pull crackle).
|
|
assert_eq!(frames_to_produce(0, 8192), 1024);
|
|
// ...still clamped to a small mapped slice.
|
|
assert_eq!(frames_to_produce(0, 256), 256);
|
|
}
|
|
|
|
#[test]
|
|
fn degenerate_zero_mapped_is_zero() {
|
|
// No buffer to write into → produce nothing, no panic / underflow.
|
|
assert_eq!(frames_to_produce(0, 0), 0);
|
|
assert_eq!(frames_to_produce(1024, 0), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn capture_size_larger_than_mapping_is_clamped() {
|
|
let mut samples = Vec::new();
|
|
for_each_capture_sample(&[1, 0, 2, 0, 3], usize::MAX, |sample| samples.push(sample));
|
|
assert_eq!(samples, vec![1, 2]);
|
|
}
|
|
|
|
#[test]
|
|
fn occupancy_is_reserved_before_frame_is_published() {
|
|
let rb = HeapRb::<i16>::new(8);
|
|
let (mut producer, mut consumer) = rb.split();
|
|
assert!(producer.try_push(7).is_ok());
|
|
|
|
let fill = AtomicUsize::new(1);
|
|
let dropped = AtomicU64::new(0);
|
|
publish_frame(&fill, &dropped, &mut producer, &[10, 11], || {
|
|
// Force the consumer to drain the old sample after the new frame's
|
|
// occupancy is reserved but before that frame is published.
|
|
assert_eq!(consumer.try_pop(), Some(7));
|
|
assert_eq!(fill.fetch_sub(1, Ordering::Relaxed), 3);
|
|
});
|
|
|
|
assert_eq!(fill.load(Ordering::Relaxed), 2);
|
|
assert_eq!(consumer.try_pop(), Some(10));
|
|
assert_eq!(fill.fetch_sub(1, Ordering::Relaxed), 2);
|
|
assert_eq!(consumer.try_pop(), Some(11));
|
|
assert_eq!(fill.fetch_sub(1, Ordering::Relaxed), 1);
|
|
assert_eq!(fill.load(Ordering::Relaxed), 0);
|
|
assert_eq!(dropped.load(Ordering::Relaxed), 0);
|
|
}
|
|
|
|
// --- drain_loop (A7: worker must not hang shutdown) ---
|
|
|
|
#[test]
|
|
fn drain_loop_exits_when_running_flips_even_with_sender_alive() {
|
|
// The exact A7 hang scenario: the frame sender is STILL alive (never
|
|
// dropped) when `running` goes false. A blocking `recv()` would park
|
|
// forever; `drain_loop` must wake within a poll interval and return.
|
|
let (tx, rx) = mpsc::channel::<Vec<i16>>();
|
|
let running = Arc::new(AtomicBool::new(true));
|
|
let r2 = running.clone();
|
|
let h = thread::spawn(move || drain_loop(&rx, &r2, |_| {}));
|
|
// Let it park in recv_timeout, then signal stop.
|
|
thread::sleep(Duration::from_millis(50));
|
|
running.store(false, Ordering::Relaxed);
|
|
// Wait comfortably longer than one poll interval; it must have exited.
|
|
thread::sleep(super::WORKER_POLL + Duration::from_millis(150));
|
|
assert!(
|
|
h.is_finished(),
|
|
"drain_loop must exit after running=false even while the sender is alive"
|
|
);
|
|
drop(tx); // keep tx alive until here so the test really covers the case
|
|
h.join().unwrap();
|
|
}
|
|
|
|
#[test]
|
|
fn drain_loop_returns_on_disconnect() {
|
|
// Sender dropped before we start → recv errors Disconnected → return.
|
|
// If this hangs, the test runner times out (i.e. it would fail loudly).
|
|
let (tx, rx) = mpsc::channel::<Vec<i16>>();
|
|
let running = Arc::new(AtomicBool::new(true));
|
|
drop(tx);
|
|
drain_loop(&rx, &running, |_| panic!("no frame should arrive"));
|
|
}
|
|
|
|
#[test]
|
|
fn drain_loop_delivers_frames_to_callback() {
|
|
let (tx, rx) = mpsc::channel::<Vec<i16>>();
|
|
let running = Arc::new(AtomicBool::new(true));
|
|
let r2 = running.clone();
|
|
let got = Arc::new(Mutex::new(Vec::new()));
|
|
let g2 = got.clone();
|
|
let h = thread::spawn(move || drain_loop(&rx, &r2, |f| g2.lock().unwrap().push(f)));
|
|
tx.send(vec![1, 2, 3]).unwrap();
|
|
tx.send(vec![4, 5]).unwrap();
|
|
thread::sleep(Duration::from_millis(50));
|
|
running.store(false, Ordering::Relaxed);
|
|
drop(tx);
|
|
h.join().unwrap();
|
|
let got = got.lock().unwrap();
|
|
assert_eq!(*got, vec![vec![1, 2, 3], vec![4, 5]]);
|
|
}
|
|
}
|