From 9b008ed6bbbdbacf0cb0823045f3ec73cbd0ec8d Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 20 Oct 2026 22:45:23 -0400 Subject: [PATCH] berd-call: wait for playback room instead of failing long Siri replies Siri renders faster than playback drains, and the native callback failed the whole reply once 60 seconds of audio were pending. It now waits for room when the bound is full and keeps it as a memory cap. Signed-off-by: John Tennant Co-Authored-By: Claude Opus 5.5 --- src-tauri/crates/berd-call/src/siri.rs | 174 +++++++++++++++++++------ 1 file changed, 131 insertions(+), 43 deletions(-) diff --git a/src-tauri/crates/berd-call/src/siri.rs b/src-tauri/crates/berd-call/src/siri.rs index a20402a69..9e3145c1c 100644 --- a/src-tauri/crates/berd-call/src/siri.rs +++ b/src-tauri/crates/berd-call/src/siri.rs @@ -484,21 +484,26 @@ unsafe extern "C" fn receive_pcm( let frames = unsafe { std::slice::from_raw_parts(samples, frame_count as usize) }; // SAFETY: The native call is scoped to the lifetime of this context. let context = unsafe { &*(context.cast::>()) }; - if !context.active.load(Ordering::SeqCst) || context.callback_cancelled.load(Ordering::SeqCst) { - return false; - } - let Ok(mut pending) = context.frames.lock() else { - context.queue_overflowed.store(true, Ordering::SeqCst); - context.callback_cancelled.store(true, Ordering::SeqCst); - return false; + // Siri renders faster than playback drains, so a reply longer than the + // bound waits here for room; shorter ones never wait on playback. + let mut pending = loop { + if !context.active.load(Ordering::SeqCst) + || context.callback_cancelled.load(Ordering::SeqCst) + { + return false; + } + let Ok(pending) = context.frames.lock() else { + context.queue_overflowed.store(true, Ordering::SeqCst); + context.callback_cancelled.store(true, Ordering::SeqCst); + return false; + }; + if pending.is_empty() || pending.len() + frames.len() <= MAX_PENDING_PCM_FRAMES { + break pending; + } + drop(pending); + std::thread::sleep(SYNTHESIS_POLL_INTERVAL); }; - if pending - .len() - .checked_add(frames.len()) - .filter(|total| *total <= MAX_PENDING_PCM_FRAMES) - .is_none() - || pending.try_reserve(frames.len()).is_err() - { + if pending.try_reserve(frames.len()).is_err() { context.queue_overflowed.store(true, Ordering::SeqCst); context.callback_cancelled.store(true, Ordering::SeqCst); return false; @@ -709,7 +714,7 @@ mod tests { #[cfg(target_os = "macos")] use std::collections::VecDeque; #[cfg(target_os = "macos")] - use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; #[cfg(target_os = "macos")] use std::sync::{mpsc, Arc, Mutex}; use std::time::Duration; @@ -1000,41 +1005,100 @@ mod tests { #[test] #[cfg(target_os = "macos")] - fn native_pcm_callback_fails_instead_of_exceeding_its_memory_bound() { + fn a_reply_longer_than_the_pcm_bound_waits_for_playback_instead_of_failing() { let active = AtomicBool::new(true); let callback_cancelled = AtomicBool::new(false); let queue_overflowed = AtomicBool::new(false); - let pending_frames = Mutex::new(VecDeque::from(vec![0.0; MAX_PENDING_PCM_FRAMES])); + let pending_frames = Mutex::new(VecDeque::new()); let (sender, receiver) = mpsc::sync_channel(1); - let mut context = CallbackContext { - active: &active, - callback_cancelled: &callback_cancelled, - queue_overflowed: &queue_overflowed, - frames: &pending_frames, - notification: sender, - }; - let sample = 0.25_f32; - // SAFETY: The sample and callback context remain live for this call. - assert!(!unsafe { - receive_pcm( - &sample, - 1, - (&mut context as *mut CallbackContext<'_>).cast(), + let reply_frames = MAX_PENDING_PCM_FRAMES * 3 / 2; + let rendered = AtomicUsize::new(0); + let mut played = 0; + std::thread::scope(|scope| { + scope.spawn(|| { + let mut context = CallbackContext { + active: &active, + callback_cancelled: &callback_cancelled, + queue_overflowed: &queue_overflowed, + frames: &pending_frames, + notification: sender, + }; + let chunk = [0.25_f32; 4_096]; + while rendered.load(Ordering::SeqCst) < reply_frames { + // SAFETY: The chunk and callback context remain live for this call. + if !unsafe { + receive_pcm( + chunk.as_ptr(), + chunk.len() as u32, + (&mut context as *mut CallbackContext<'_>).cast(), + ) + } { + return; + } + rendered.fetch_add(chunk.len(), Ordering::SeqCst); + } + }); + receive_pcm_until_complete( + receiver, + &callback_cancelled, + &queue_overflowed, + &pending_frames, + Duration::from_millis(1), + Duration::from_secs(5), + &mut |event| { + if let TtsSynthesisEvent::Frames(frames) = event { + // Playback starts only once Siri has rendered a full bound ahead. + let deadline = Instant::now() + Duration::from_secs(5); + while played == 0 + && rendered.load(Ordering::SeqCst) < MAX_PENDING_PCM_FRAMES + && Instant::now() < deadline + { + std::thread::sleep(Duration::from_millis(1)); + } + played += frames.len(); + } + Ok(()) + }, ) + .unwrap(); + }); + assert!(played >= reply_frames); + assert!(!queue_overflowed.load(Ordering::SeqCst)); + } + + #[test] + #[cfg(target_os = "macos")] + fn a_full_pcm_queue_releases_its_callback_when_synthesis_is_cancelled() { + let active = AtomicBool::new(true); + let callback_cancelled = AtomicBool::new(false); + let queue_overflowed = AtomicBool::new(false); + let pending_frames = Mutex::new(VecDeque::from(vec![0.0; MAX_PENDING_PCM_FRAMES])); + let (sender, _receiver) = mpsc::sync_channel(1); + std::thread::scope(|scope| { + let callback = scope.spawn(|| { + let mut context = CallbackContext { + active: &active, + callback_cancelled: &callback_cancelled, + queue_overflowed: &queue_overflowed, + frames: &pending_frames, + notification: sender, + }; + let sample = 0.25_f32; + // SAFETY: The sample and callback context remain live for this call. + unsafe { + receive_pcm( + &sample, + 1, + (&mut context as *mut CallbackContext<'_>).cast(), + ) + } + }); + std::thread::sleep(Duration::from_millis(50)); + assert!(!callback.is_finished(), "a full queue waits for playback"); + active.store(false, Ordering::SeqCst); + assert!(!callback.join().unwrap()); }); - let error = receive_pcm_until_complete( - receiver, - &callback_cancelled, - &queue_overflowed, - &pending_frames, - Duration::from_millis(1), - Duration::from_secs(1), - &mut |_| Ok(()), - ) - .unwrap_err(); - assert_eq!(error, "Siri synthesis exceeded its bounded PCM queue"); assert_eq!(pending_frames.lock().unwrap().len(), MAX_PENDING_PCM_FRAMES); - assert!(callback_cancelled.load(Ordering::SeqCst)); } #[test] @@ -1117,4 +1181,28 @@ mod tests { assert_eq!(outcome, TtsOutcome::Cancelled); assert!(started.elapsed() < Duration::from_secs(2)); } + + #[test] + #[cfg(target_os = "macos")] + #[ignore = "requires BERD_SIRI_TEST_VOICE and invokes private sirittsd synthesis"] + fn a_reply_longer_than_the_pcm_bound_synthesizes_completely() { + let voice = std::env::var("BERD_SIRI_TEST_VOICE").unwrap(); + let language = std::env::var("BERD_SIRI_TEST_LANGUAGE").unwrap_or_else(|_| "en-US".into()); + let backend = SiriTts::new(&language, &voice, 1.0).unwrap(); + let text = "This sentence is one of many that together run well past a minute of speech. " + .repeat(30); + let mut frames = 0; + let outcome = backend + .synthesize(&text, &AtomicBool::new(true), &mut |chunk| { + // Hold the first chunk, as playback would, while Siri renders ahead. + if frames == 0 { + std::thread::sleep(Duration::from_secs(20)); + } + frames += chunk.len(); + Ok(()) + }) + .unwrap(); + assert_eq!(outcome, TtsOutcome::Completed); + assert!(frames > MAX_PENDING_PCM_FRAMES); + } }