From 1bf14ba08f0b8ebba2fea28e3cd5d2a66d0f030f Mon Sep 17 00:00:00 2001 From: Mollusk Date: Mon, 29 Jun 2026 14:53:52 -0400 Subject: [PATCH] perf(recording): move multitrack stem disk I/O off the mixer path (A17b) The multitrack recorder wrote every per-stem WAV frame (and the potentially large late-joiner back-pad) inline on the caller thread while holding the recorder mutex, so a slow/contended disk stalled the playout mixer (local underruns) and the events loop. This is the multitrack counterpart to A17 (e0325d4), which moved the single-file recorder's writes off the mixer path. Design: the front (MultitrackRecorder) now keeps only cheap in-memory state (known-peer set, mic FIFO, a pending-cycle builder) and on each end_cycle assembles ONE whole-cycle batch (new peers + mic frame + optional mix frame + the map of peer frames written this cycle) and try_sends it over a bounded sync_channel(256) to a dedicated writer thread. The writer thread owns every WavWriter, is authoritative for its own cycle count, back-pads a brand-new peer by cycles_written*frame_samples, fills absent peer/mix frames with silence, latches the first write/create error then drains, and finalizes all headers on channel close. The unit of hand-off is a whole cycle, not a track: the writer appends exactly frame_samples to every existing track per applied batch, and a full queue DROPS the entire batch (counted + logged at 1 and every 256). So a dropped cycle omits the same 20ms from every stem at once and all tracks stay equal-length and sample-aligned by construction even under disk back-pressure. On drop the batch's new-peer announcements are rolled back out of the known set so they re-announce (and correctly re-back-pad) on the next applied cycle. Public method signatures are unchanged -> zero core/mod.rs edits. The WAV/file format is unchanged (no wire/on-disk change), no new deps (std::sync::mpsc + std::thread, as A17). Writer logic is factored behind a generic SampleWriter seam so the apply-batch alignment invariant is unit-tested without spawning the thread; new tests cover the back-pad-on-apply invariant, the dropped-cycle equal-length property, and async create-error surfacing at finalize. The three existing end-to-end tests pass unchanged (now exercising the threaded path). 496 lib tests, clippy --all-targets clean, release builds. Codex-implemented (gpt-5.5 xhigh), senior-reviewed. Co-Authored-By: Claude Opus 4.8 --- src/audio/multitrack.rs | 502 +++++++++++++++++++++++++++++++++------- 1 file changed, 418 insertions(+), 84 deletions(-) diff --git a/src/audio/multitrack.rs b/src/audio/multitrack.rs index 768d29a..a2788a7 100644 --- a/src/audio/multitrack.rs +++ b/src/audio/multitrack.rs @@ -12,9 +12,11 @@ //! This module is pure plumbing over [`WavWriter`]: no audio decode, no //! networking, no realtime work. The mixer (a non-RT task) drives it. -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::io; use std::path::{Path, PathBuf}; +use std::sync::mpsc::{self, SyncSender, TrySendError}; +use std::thread::{self, JoinHandle}; use iroh::EndpointId; @@ -24,6 +26,8 @@ use crate::core::jitter::FRAME_SAMPLES; /// Cap on the silence chunk written at once when pre-padding a late joiner, so a /// long-running call can't trigger a single multi-hundred-MB allocation. const SILENCE_CHUNK: usize = FRAME_SAMPLES * 256; +const WRITER_QUEUE_CYCLES: usize = 256; +const DROP_LOG_INTERVAL_CYCLES: u64 = 256; /// Cap on buffered mic samples (~200ms @ 48kHz). Bounds how far the mic track /// can drift if the capture clock runs ahead of the mixer cycle; past it the @@ -56,41 +60,6 @@ pub fn create_session_dir(base: &Path, now_unix_secs: u64) -> io::Result io::Result { - Ok(Self { - writer: WavWriter::new(path)?, - written_this_cycle: false, - }) - } - - /// Append `frame` fitted to exactly `frame_samples` (zero-padded if short), - /// and mark the track as written for this cycle. - fn write_frame(&mut self, frame: &[i16], frame_samples: usize) -> io::Result<()> { - self.writer.write_samples(&fit(frame, frame_samples))?; - self.written_this_cycle = true; - Ok(()) - } - - /// Append `samples` of silence (no cycle-marking — used for padding). - fn write_silence(&mut self, samples: usize) -> io::Result<()> { - let mut remaining = samples; - while remaining > 0 { - let n = remaining.min(SILENCE_CHUNK); - self.writer.write_samples(&vec![0i16; n])?; - remaining -= n; - } - Ok(()) - } -} - /// Return `frame` resized to exactly `n` samples: truncated if longer (shouldn't /// happen — Opus frames are uniform), zero-padded if shorter. fn fit(frame: &[i16], n: usize) -> Vec { @@ -126,41 +95,210 @@ pub fn track_filename(name: &str, id: &EndpointId) -> String { format!("{slug}-{short}.wav") } -/// A live multitrack recording: per-peer stems + your mic, plus an optional -/// mixed track, all under one session directory and clocked together. -pub struct MultitrackRecorder { - dir: PathBuf, - frame_samples: usize, - /// Cycles recorded so far = the shared length (in frames) of every track. - cycles: u64, - peers: HashMap, - /// Your mic track. Fed asynchronously from the capture thread via - /// [`push_mic`](MultitrackRecorder::push_mic) into `mic_fifo`, then drained - /// one frame per `end_cycle` so it aligns with the cycle clock. - mic: WavWriter, - mic_fifo: VecDeque, - /// Present in "Both" mode (stems + mixed), absent in "stems only". - mix: Option, +#[derive(Default)] +struct PendingCycle { + new_peers: Vec, + peer_frames: HashMap>, + mix_frame: Option>, } -impl MultitrackRecorder { - /// Create a recording in `dir` (which must already exist). `with_mix` adds - /// the convenience mixed track (`mix.wav`). Your mic is always `me.wav`. - pub fn create(dir: &Path, frame_samples: usize, with_mix: bool) -> io::Result { +struct NewPeer { + id: EndpointId, + filename: String, +} + +struct CycleBatch { + new_peers: Vec, + mic_frame: Vec, + mix_frame: Option>, + peer_frames: HashMap>, +} + +trait SampleWriter { + fn write_samples(&mut self, samples: &[i16]) -> io::Result<()>; + fn finalize(self) -> io::Result<()>; +} + +impl SampleWriter for WavWriter { + fn write_samples(&mut self, samples: &[i16]) -> io::Result<()> { + WavWriter::write_samples(self, samples) + } + + fn finalize(self) -> io::Result<()> { + WavWriter::finalize(self) + } +} + +struct WriterState { + dir: PathBuf, + frame_samples: usize, + peers: HashMap, + mic: W, + mix: Option, + cycles_written: u64, +} + +impl WriterState { + fn create(dir: &Path, frame_samples: usize, with_mix: bool) -> io::Result { let mic = WavWriter::new(&dir.join("me.wav"))?; let mix = if with_mix { - Some(Track::create(&dir.join("mix.wav"))?) + Some(WavWriter::new(&dir.join("mix.wav"))?) } else { None }; Ok(Self { dir: dir.to_path_buf(), frame_samples, - cycles: 0, peers: HashMap::new(), mic, - mic_fifo: VecDeque::new(), mix, + cycles_written: 0, + }) + } +} + +impl WriterState { + fn apply_batch(&mut self, batch: &CycleBatch, mut create_peer: F) -> io::Result<()> + where + F: FnMut(&Path) -> io::Result, + { + for peer in &batch.new_peers { + if !self.peers.contains_key(&peer.id) { + let writer = create_peer(&self.dir.join(&peer.filename))?; + self.peers.insert(peer.id, writer); + let pad = self.back_pad_samples()?; + let writer = self.peers.get_mut(&peer.id).unwrap(); + Self::write_silence(writer, pad)?; + } + } + + self.mic.write_samples(&batch.mic_frame)?; + if let Some(mix) = self.mix.as_mut() { + if let Some(frame) = batch.mix_frame.as_deref() { + mix.write_samples(frame)?; + } else { + Self::write_silence(mix, self.frame_samples)?; + } + } + + let silence = vec![0i16; self.frame_samples]; + for (id, writer) in &mut self.peers { + let frame = batch + .peer_frames + .get(id) + .map(Vec::as_slice) + .unwrap_or(&silence); + writer.write_samples(frame)?; + } + self.cycles_written += 1; + Ok(()) + } + + fn back_pad_samples(&self) -> io::Result { + let cycles = usize::try_from(self.cycles_written) + .map_err(|_| io::Error::other("multitrack recording too long"))?; + cycles + .checked_mul(self.frame_samples) + .ok_or_else(|| io::Error::other("multitrack recording too long")) + } + + fn write_silence(writer: &mut W, samples: usize) -> io::Result<()> { + let mut remaining = samples; + let silence = vec![0i16; remaining.min(SILENCE_CHUNK)]; + while remaining > 0 { + let n = remaining.min(silence.len()); + writer.write_samples(&silence[..n])?; + remaining -= n; + } + Ok(()) + } + + fn finalize(self) -> io::Result<()> { + let mut first_finalize_error = None; + record_first_error(&mut first_finalize_error, self.mic.finalize()); + if let Some(mix) = self.mix { + record_first_error(&mut first_finalize_error, mix.finalize()); + } + for writer in self.peers.into_values() { + record_first_error(&mut first_finalize_error, writer.finalize()); + } + if let Some(e) = first_finalize_error { + Err(e) + } else { + Ok(()) + } + } +} + +fn record_first_error(slot: &mut Option, result: io::Result<()>) { + if slot.is_none() + && let Err(e) = result + { + *slot = Some(e); + } +} + +/// Applies whole-cycle batches on the writer thread. Each applied batch appends +/// exactly `frame_samples` to every existing track, and a dropped batch never +/// reaches this loop for any track, so stem lengths stay equal even when the +/// bounded queue applies back-pressure. +fn writer_thread_main( + mut state: WriterState, + batch_rx: mpsc::Receiver, +) -> io::Result<()> { + let mut first_write_error = None; + + for batch in batch_rx { + if first_write_error.is_none() + && let Err(e) = state.apply_batch(&batch, WavWriter::new) + { + first_write_error = Some(e); + } + } + + let finalize_result = state.finalize(); + if let Some(e) = first_write_error { + Err(e) + } else { + finalize_result + } +} + +/// A live multitrack recording: per-peer stems + your mic, plus an optional +/// mixed track, all under one session directory and clocked together. +pub struct MultitrackRecorder { + dir: PathBuf, + frame_samples: usize, + known_peers: HashSet, + /// Your mic track. Fed asynchronously from the capture thread via + /// [`push_mic`](MultitrackRecorder::push_mic) into `mic_fifo`, then drained + /// one frame per `end_cycle` so it aligns with the cycle clock. + mic_fifo: VecDeque, + /// Present in "Both" mode (stems + mixed), absent in "stems only". + with_mix: bool, + batch_tx: SyncSender, + writer_thread: JoinHandle>, + dropped_cycles: u64, + pending: PendingCycle, +} + +impl MultitrackRecorder { + /// Create a recording in `dir` (which must already exist). `with_mix` adds + /// the convenience mixed track (`mix.wav`). Your mic is always `me.wav`. + pub fn create(dir: &Path, frame_samples: usize, with_mix: bool) -> io::Result { + let writer_state = WriterState::create(dir, frame_samples, with_mix)?; + let (batch_tx, batch_rx) = mpsc::sync_channel(WRITER_QUEUE_CYCLES); + let writer_thread = thread::spawn(move || writer_thread_main(writer_state, batch_rx)); + Ok(Self { + dir: dir.to_path_buf(), + frame_samples, + known_peers: HashSet::new(), + mic_fifo: VecDeque::new(), + with_mix, + batch_tx, + writer_thread, + dropped_cycles: 0, + pending: PendingCycle::default(), }) } @@ -173,12 +311,14 @@ impl MultitrackRecorder { /// so it aligns with the others. Idempotent: a peer already tracked is left /// as-is (re-announce / name change doesn't restart their file). pub fn add_peer(&mut self, id: EndpointId, name: &str) -> io::Result<()> { - if self.peers.contains_key(&id) { + if self.known_peers.contains(&id) { return Ok(()); } - let mut track = Track::create(&self.dir.join(track_filename(name, &id)))?; - track.write_silence(self.cycles as usize * self.frame_samples)?; - self.peers.insert(id, track); + self.known_peers.insert(id); + self.pending.new_peers.push(NewPeer { + id, + filename: track_filename(name, &id), + }); Ok(()) } @@ -186,11 +326,13 @@ impl MultitrackRecorder { /// registered yet (write raced ahead of the join event), auto-register it /// with an id-only name so no audio is dropped. pub fn write_peer(&mut self, id: EndpointId, frame: &[i16]) -> io::Result<()> { - if !self.peers.contains_key(&id) { + if !self.known_peers.contains(&id) { self.add_peer(id, "")?; } - let fs = self.frame_samples; - self.peers.get_mut(&id).unwrap().write_frame(frame, fs) + self.pending + .peer_frames + .insert(id, fit(frame, self.frame_samples)); + Ok(()) } /// Buffer a frame of your transmitted mic audio (called from the capture @@ -215,9 +357,8 @@ impl MultitrackRecorder { /// Record the finished mixed-bus frame for the current cycle (no-op in /// stems-only mode). pub fn write_mix(&mut self, frame: &[i16]) -> io::Result<()> { - let fs = self.frame_samples; - if let Some(mix) = self.mix.as_mut() { - mix.write_frame(frame, fs)?; + if self.with_mix { + self.pending.mix_frame = Some(fit(frame, self.frame_samples)); } Ok(()) } @@ -230,28 +371,59 @@ impl MultitrackRecorder { // Mic: always one frame per cycle, drained from the FIFO (silence on // underrun), so it tracks the cycle clock like the peer stems. let mic_frame = self.drain_mic(fs); - self.mic.write_samples(&mic_frame)?; - // Peers + the optional mix track: pad any not written this cycle. - for track in self.peers.values_mut().chain(self.mix.as_mut()) { - if !track.written_this_cycle { - track.write_silence(fs)?; + let mut pending = std::mem::take(&mut self.pending); + pending.new_peers.sort_by(|a, b| { + a.filename + .cmp(&b.filename) + .then_with(|| a.id.to_string().cmp(&b.id.to_string())) + }); + let batch = CycleBatch { + new_peers: pending.new_peers, + mic_frame, + mix_frame: if self.with_mix { pending.mix_frame } else { None }, + peer_frames: pending.peer_frames, + }; + match self.batch_tx.try_send(batch) { + Ok(()) => Ok(()), + Err(TrySendError::Full(batch)) => { + for peer in &batch.new_peers { + self.known_peers.remove(&peer.id); + } + self.dropped_cycles = self.dropped_cycles.saturating_add(1); + if self.dropped_cycles == 1 + || self.dropped_cycles.is_multiple_of(DROP_LOG_INTERVAL_CYCLES) + { + crate::log_msg(&format!( + "multitrack recording: writer queue full; dropped {} cycle(s)", + self.dropped_cycles + )); + } + Ok(()) } - track.written_this_cycle = false; + Err(TrySendError::Disconnected(_)) => Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "multitrack writer thread stopped", + )), } - self.cycles += 1; - Ok(()) } /// Finalize every track's WAV header. Consumes the recorder. pub fn finalize(self) -> io::Result<()> { - self.mic.finalize()?; - if let Some(mix) = self.mix { - mix.writer.finalize()?; - } - for (_, track) in self.peers { - track.writer.finalize()?; - } - Ok(()) + let Self { + dir: _, + frame_samples: _, + known_peers: _, + mic_fifo: _, + with_mix: _, + batch_tx, + writer_thread, + dropped_cycles: _, + pending: _, + } = self; + drop(batch_tx); + writer_thread + .join() + .unwrap_or_else(|_| Err(io::Error::other("multitrack writer thread panicked"))) } } @@ -277,6 +449,51 @@ mod tests { d } + #[derive(Default)] + struct TestWriter { + samples: Vec, + } + + impl SampleWriter for TestWriter { + fn write_samples(&mut self, samples: &[i16]) -> io::Result<()> { + self.samples.extend_from_slice(samples); + Ok(()) + } + + fn finalize(self) -> io::Result<()> { + Ok(()) + } + } + + fn test_writer_state(frame_samples: usize, with_mix: bool) -> WriterState { + WriterState { + dir: PathBuf::new(), + frame_samples, + peers: HashMap::new(), + mic: TestWriter::default(), + mix: if with_mix { + Some(TestWriter::default()) + } else { + None + }, + cycles_written: 0, + } + } + + fn test_batch( + new_peers: Vec, + mic_frame: Vec, + mix_frame: Option>, + peer_frames: Vec<(EndpointId, Vec)>, + ) -> CycleBatch { + CycleBatch { + new_peers, + mic_frame, + mix_frame, + peer_frames: peer_frames.into_iter().collect(), + } + } + #[test] fn fit_pads_and_truncates() { assert_eq!(fit(&[1, 2], 4), vec![1, 2, 0, 0]); @@ -311,6 +528,104 @@ mod tests { let _ = std::fs::remove_dir_all(&base); } + #[test] + fn apply_batch_advances_existing_tracks_and_back_pads_late_peer() { + let frame = 3; + let early = an_id(); + let late = an_id(); + let mut state = test_writer_state(frame, true); + state.cycles_written = 2; + state.mic.samples = vec![8; 2 * frame]; + state.mix.as_mut().unwrap().samples = vec![6; 2 * frame]; + state.peers.insert( + early, + TestWriter { + samples: vec![1; 2 * frame], + }, + ); + + let batch = test_batch( + vec![NewPeer { + id: late, + filename: "late.wav".to_string(), + }], + vec![9; frame], + None, + vec![(early, vec![2; frame]), (late, vec![7; frame])], + ); + state + .apply_batch(&batch, |_| Ok(TestWriter::default())) + .unwrap(); + + assert_eq!(state.cycles_written, 3); + assert_eq!(state.mic.samples.len(), 3 * frame); + assert_eq!(state.mix.as_ref().unwrap().samples.len(), 3 * frame); + assert_eq!(&state.mix.as_ref().unwrap().samples[2 * frame..], &[0, 0, 0]); + assert_eq!(state.peers.get(&early).unwrap().samples.len(), 3 * frame); + assert_eq!(&state.peers.get(&early).unwrap().samples[2 * frame..], &[2, 2, 2]); + assert_eq!( + state.peers.get(&late).unwrap().samples, + vec![0, 0, 0, 0, 0, 0, 7, 7, 7], + "late peer is back-padded by completed cycles before this batch" + ); + } + + #[test] + fn skipped_batches_keep_all_tracks_equal_length() { + let frame = 2; + let p1 = an_id(); + let p2 = an_id(); + let mut state = test_writer_state(frame, true); + + let first = test_batch( + vec![ + NewPeer { + id: p1, + filename: "p1.wav".to_string(), + }, + NewPeer { + id: p2, + filename: "p2.wav".to_string(), + }, + ], + vec![1; frame], + Some(vec![5; frame]), + vec![(p1, vec![10; frame]), (p2, vec![20; frame])], + ); + state + .apply_batch(&first, |_| Ok(TestWriter::default())) + .unwrap(); + + let _dropped_cycle = test_batch( + Vec::new(), + vec![2; frame], + Some(vec![6; frame]), + vec![(p1, vec![11; frame])], + ); + + let after_drop = test_batch( + Vec::new(), + vec![3; frame], + None, + vec![(p1, vec![12; frame])], + ); + state + .apply_batch(&after_drop, |_| Ok(TestWriter::default())) + .unwrap(); + + let expected = 2 * frame; + assert_eq!(state.cycles_written, 2); + assert_eq!(state.mic.samples.len(), expected); + assert_eq!(state.mix.as_ref().unwrap().samples.len(), expected); + assert_eq!(state.peers.get(&p1).unwrap().samples.len(), expected); + assert_eq!(state.peers.get(&p2).unwrap().samples.len(), expected); + assert_eq!( + &state.peers.get(&p2).unwrap().samples[frame..], + &[0, 0], + "peer absent from an applied batch gets silence for that cycle" + ); + } + #[test] fn all_tracks_equal_length_after_n_cycles() { let dir = tmpdir("equal"); @@ -406,4 +721,23 @@ mod tests { "no mix track in stems-only mode" ); } + + #[cfg(unix)] + #[test] + fn async_peer_create_error_surfaces_at_finalize() { + use std::os::unix::fs::PermissionsExt; + + let dir = tmpdir("asyncerr"); + let mut rec = MultitrackRecorder::create(&dir, 4, false).unwrap(); + std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o500)).unwrap(); + + rec.add_peer(an_id(), "blocked").unwrap(); + rec.end_cycle().unwrap(); + let result = rec.finalize(); + + std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).unwrap(); + let err = result.unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::PermissionDenied); + let _ = std::fs::remove_dir_all(&dir); + } }