From e0325d4590284a0c32a4586738497eee6a462f94 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Mon, 29 Jun 2026 02:10:29 -0400 Subject: [PATCH] perf(recording): move single-file WAV disk I/O off the mixer path (A17) Recorder::write_frame ran on the playout mixer path and did a blocking write_all to disk per 20ms frame; slow/contended storage could stall the mixer and cause local playback underruns. Now the mixer thread only does the cheap mic-sum (extracted as the pure mix_with_mic helper) and try_sends the frame to a dedicated writer thread over a bounded sync_channel(256). The writer thread owns the WavWriter, writes queued frames, records the first write error then drains without writing, and patches the WAV size fields on channel close. A full queue DROPS the recording frame (counted + logged at 1 and every 256) rather than blocking call audio; a disconnected writer surfaces BrokenPipe. finalize() closes the channel, joins the thread, and returns the first write error or the finalize result (thread panic handled). Scope: single-file Recorder only; WavWriter unchanged so the multitrack recorder is untouched (its writer-thread offload is deferred as A17b). Public method signatures preserved -> no core/mod.rs changes. New end-to-end threaded WAV readback test + mix_with_mic helper tests; existing FIFO/mic-sum intent kept. No new deps, no wire change. Codex-implemented (gpt-5.5 xhigh), senior-reviewed. Co-Authored-By: Claude Opus 4.8 --- src/audio/recorder.rs | 194 ++++++++++++++++++++++++++++++++---------- 1 file changed, 151 insertions(+), 43 deletions(-) diff --git a/src/audio/recorder.rs b/src/audio/recorder.rs index 7c3615f..be71472 100644 --- a/src/audio/recorder.rs +++ b/src/audio/recorder.rs @@ -2,21 +2,26 @@ //! //! Records the **full call as you experienced it**: the mixed incoming audio //! (everyone you hear) summed with your own transmitted mic, into a single mono -//! WAV. Writing is driven by the playout mixer (one [`Recorder::write_frame`] -//! per produced 20ms frame, paced by the hardware clock); your mic arrives -//! separately from the capture thread via [`Recorder::push_mic`] and is buffered -//! in a small FIFO so the two independently-clocked streams stay roughly aligned. +//! WAV. Mixing/enqueue is driven by the playout mixer (one +//! [`Recorder::write_frame`] per produced 20ms frame, paced by the hardware +//! clock), while disk writes happen on a dedicated writer thread; your mic +//! arrives separately from the capture thread via [`Recorder::push_mic`] and is +//! buffered in a small FIFO so the two independently-clocked streams stay +//! roughly aligned. //! Minor clock drift just slowly grows/shrinks that FIFO (capped, so the lag //! between your voice and the recording is bounded) — harmless for a voice //! recording, no realtime crackle concern. //! //! No external crates: the WAV writer emits the 44-byte canonical header itself -//! and patches the two size fields on [`Recorder::finalize`]. +//! and patches the two size fields on the writer thread during +//! [`Recorder::finalize`]. use std::collections::VecDeque; use std::fs::{File, OpenOptions}; use std::io::{self, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; +use std::sync::mpsc::{self, SyncSender, TrySendError}; +use std::thread::{self, JoinHandle}; /// Capture sample rate (mono, 48kHz, matching the rest of the audio path). const SAMPLE_RATE: u32 = 48_000; @@ -25,6 +30,8 @@ const CHANNELS: u16 = 1; const RIFF_DATA_OVERHEAD: u64 = 36; const MAX_RIFF_DATA_BYTES: u64 = u32::MAX as u64 - RIFF_DATA_OVERHEAD; const MAX_NAME_ATTEMPTS: usize = 1_000; +const WRITER_QUEUE_FRAMES: usize = 256; +const DROP_LOG_INTERVAL_FRAMES: u64 = 256; /// Cap on buffered mic samples (~200ms). Bounds how far recording lag can drift /// if the capture clock runs persistently faster than playout — past this we drop @@ -118,13 +125,15 @@ impl WavWriter { } } -/// A live call recorder: a [`WavWriter`] plus a small mic FIFO that aligns your -/// transmitted mic with the playout mixer's incoming-mix frames. +/// A live call recorder: a writer-thread queue plus a small mic FIFO that aligns +/// your transmitted mic with the playout mixer's incoming-mix frames. pub struct Recorder { - writer: WavWriter, + frame_tx: SyncSender>, + writer_thread: JoinHandle>, /// Your transmitted mic samples, awaiting alignment with the next mix frame. mic_fifo: VecDeque, path: PathBuf, + dropped_frames: u64, } impl Recorder { @@ -142,10 +151,15 @@ impl Recorder { let path = dir.join(name); match OpenOptions::new().write(true).create_new(true).open(&path) { Ok(file) => { + let writer = WavWriter::from_file(file)?; + let (frame_tx, frame_rx) = mpsc::sync_channel(WRITER_QUEUE_FRAMES); + let writer_thread = thread::spawn(move || writer_thread_main(writer, frame_rx)); return Ok(Self { - writer: WavWriter::from_file(file)?, + frame_tx, + writer_thread, mic_fifo: VecDeque::new(), path, + dropped_frames: 0, }); } Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue, @@ -179,21 +193,73 @@ impl Recorder { /// treated as silence (you weren't transmitting), so quiet stretches record /// the incoming mix alone. pub fn write_frame(&mut self, mixed: &[i16]) -> io::Result<()> { - let mut out = Vec::with_capacity(mixed.len()); - for &m in mixed { - let mic = self.mic_fifo.pop_front().unwrap_or(0); - let sum = (m as i32 + mic as i32).clamp(i16::MIN as i32, i16::MAX as i32); - out.push(sum as i16); + let out = mix_with_mic(mixed, &mut self.mic_fifo); + match self.frame_tx.try_send(out) { + Ok(()) => Ok(()), + Err(TrySendError::Full(_)) => { + self.dropped_frames = self.dropped_frames.saturating_add(1); + if self.dropped_frames == 1 + || self.dropped_frames.is_multiple_of(DROP_LOG_INTERVAL_FRAMES) + { + crate::log_msg(&format!( + "recording: writer queue full; dropped {} frame(s)", + self.dropped_frames + )); + } + Ok(()) + } + Err(TrySendError::Disconnected(_)) => Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "recording writer thread stopped", + )), } - self.writer.write_samples(&out) } /// Finish the file, patching its size fields. Consumes the recorder. pub fn finalize(self) -> io::Result<()> { - self.writer.finalize() + let Self { + frame_tx, + writer_thread, + mic_fifo: _, + path: _, + dropped_frames: _, + } = self; + drop(frame_tx); + writer_thread + .join() + .unwrap_or_else(|_| Err(io::Error::other("recording writer thread panicked"))) } } +fn writer_thread_main(mut writer: WavWriter, frame_rx: mpsc::Receiver>) -> io::Result<()> { + let mut first_write_error = None; + + for frame in frame_rx { + if first_write_error.is_none() + && let Err(e) = writer.write_samples(&frame) + { + first_write_error = Some(e); + } + } + + let finalize_result = writer.finalize(); + if let Some(e) = first_write_error { + Err(e) + } else { + finalize_result + } +} + +fn mix_with_mic(mixed: &[i16], mic_fifo: &mut VecDeque) -> Vec { + let mut out = Vec::with_capacity(mixed.len()); + for &m in mixed { + let mic = mic_fifo.pop_front().unwrap_or(0); + let sum = (m as i32 + mic as i32).clamp(i16::MIN as i32, i16::MAX as i32); + out.push(sum as i16); + } + out +} + /// Civil date (year, month, day) from a count of days since the Unix epoch. /// Howard Hinnant's `civil_from_days`; valid across the whole practical range. fn civil_from_days(z: i64) -> (i64, u32, u32) { @@ -222,6 +288,23 @@ pub fn timestamp_filename(unix_secs: u64) -> String { #[cfg(test)] mod tests { use super::*; + use std::sync::atomic::{AtomicU64, Ordering}; + + static NEXT_TEMP_ID: AtomicU64 = AtomicU64::new(0); + + fn unique_temp_dir(prefix: &str) -> PathBuf { + let id = NEXT_TEMP_ID.fetch_add(1, Ordering::Relaxed); + std::env::temp_dir().join(format!("{prefix}-{}-{id}", std::process::id())) + } + + fn read_wav_samples(path: &Path) -> (Vec, Vec) { + let bytes = std::fs::read(path).unwrap(); + let samples = bytes[44..] + .chunks_exact(2) + .map(|sample| i16::from_le_bytes([sample[0], sample[1]])) + .collect(); + (bytes, samples) + } #[test] fn timestamp_filename_is_utc_and_padded() { @@ -236,10 +319,7 @@ mod tests { #[test] fn same_second_recordings_get_unique_files_without_truncation() { - let dir = std::env::temp_dir().join(format!( - "peerspeak-collision-{}", - std::process::id() - )); + let dir = unique_temp_dir("peerspeak-collision"); let _ = std::fs::remove_dir_all(&dir); std::fs::create_dir_all(&dir).unwrap(); @@ -258,6 +338,40 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + #[test] + fn recorder_thread_writes_mixed_samples_and_header_on_finalize() { + let dir = unique_temp_dir("peerspeak-recorder-thread"); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + + let mut recorder = Recorder::create(&dir, 1_700_000_123).unwrap(); + let path = recorder.path().to_path_buf(); + + recorder.push_mic(&[1000, i16::MAX, -1000, i16::MIN, 2222]); + recorder.write_frame(&[10, 20, -32700]).unwrap(); + recorder.push_mic(&[300, -300]); + recorder + .write_frame(&[0, 1000, i16::MAX, i16::MIN]) + .unwrap(); + recorder.finalize().unwrap(); + + let expected = vec![1010, i16::MAX, i16::MIN, i16::MIN, 3222, i16::MAX, i16::MIN]; + let expected_data_bytes = u32::try_from(expected.len() * 2).unwrap(); + let (bytes, samples) = read_wav_samples(&path); + + assert_eq!(&bytes[0..4], b"RIFF"); + assert_eq!(&bytes[8..12], b"WAVE"); + assert_eq!(&bytes[36..40], b"data"); + let riff = u32::from_le_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]); + let data = u32::from_le_bytes([bytes[40], bytes[41], bytes[42], bytes[43]]); + assert_eq!(data, expected_data_bytes); + assert_eq!(riff, RIFF_DATA_OVERHEAD as u32 + expected_data_bytes); + assert_eq!(bytes.len(), 44 + expected.len() * 2); + assert_eq!(samples, expected); + + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn wav_header_round_trips_sizes() { let dir = std::env::temp_dir(); @@ -300,38 +414,32 @@ mod tests { #[test] fn mic_is_summed_with_mix_when_present() { - let dir = std::env::temp_dir(); - let mut r = Recorder { - writer: WavWriter::new(&dir.join(format!("ps-sum-{}.wav", std::process::id()))) - .unwrap(), - mic_fifo: VecDeque::new(), - path: PathBuf::new(), - }; - r.push_mic(&[1000, 2000, 3000]); - // write_frame pops mic per-sample and sums; we can't read the file mid-stream, - // so assert the FIFO drains exactly by frame length. - r.write_frame(&[10, 20]).unwrap(); - assert_eq!(r.mic_fifo.len(), 1, "two samples consumed, one mic left"); - r.write_frame(&[0, 0]).unwrap(); + let mut mic_fifo = VecDeque::from([1000, 2000, 3000]); + + let first = mix_with_mic(&[10, 20], &mut mic_fifo); + assert_eq!(first, vec![1010, 2020]); + assert_eq!(mic_fifo.len(), 1, "two samples consumed, one mic left"); + + let second = mix_with_mic(&[0, 0], &mut mic_fifo); + assert_eq!(second, vec![3000, 0]); assert_eq!( - r.mic_fifo.len(), + mic_fifo.len(), 0, "remaining mic sample consumed; rest is silence" ); - let _ = r.finalize(); } #[test] fn mic_fifo_is_capped() { - let dir = std::env::temp_dir(); - let mut r = Recorder { - writer: WavWriter::new(&dir.join(format!("ps-cap-{}.wav", std::process::id()))) - .unwrap(), - mic_fifo: VecDeque::new(), - path: PathBuf::new(), - }; + let dir = unique_temp_dir("peerspeak-cap"); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + + let mut r = Recorder::create(&dir, 1_700_000_001).unwrap(); r.push_mic(&vec![5i16; MAX_MIC_FIFO * 2]); assert_eq!(r.mic_fifo.len(), MAX_MIC_FIFO, "FIFO is bounded to the cap"); - let _ = r.finalize(); + r.finalize().unwrap(); + + let _ = std::fs::remove_dir_all(&dir); } }