jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-05 00:41:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge PR 1: coordinate TTS playback

Browse Source 
Co-authored-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
Signed-off-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
This commit is contained in:
Wren
2026-08-19 14:21:12 -04:00
parent bbd20fae75 8bfa4c0531
commit 203735f15a
6 changed files with 446 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files

236
desktop/src-tauri/src/huddle/tts.rs
View File

@@ -18,10 +18,9 @@
//! → cancel flag: a 10 ms barge-in monitor thread silences the player and //! → cancel flag: a 10 ms barge-in monitor thread silences the player and
//! releases tts_active on the flag's rising edge (~15 ms flag-to-silence, //! releases tts_active on the flag's rising edge (~15 ms flag-to-silence,
//! even mid-sentence while the worker is blocked in synth_chunk); the //! even mid-sentence while the worker is blocked in synth_chunk); the
//! worker then consumes the flag — drain queue + clear + play (un-pause). //! worker then consumes the flag and drains stale text. Every Player
//! Monitor clears and worker player mutations are serialized through the //! operation is serialized by `PlaybackCoordinator`; cancellation swaps in
//! `player_ops` mutex, with the flag re-checked under the lock — see the //! a fresh queue and drops the old Player after releasing the coordinator.
//! monitor block in `tts_worker` for the race this closes.
//! ``` //! ```
//! //!
//! Lookahead pipelining spans *items*, not just sentences within one item: //! Lookahead pipelining spans *items*, not just sentences within one item:
@@ -55,6 +54,9 @@ use super::preprocessing::preprocess_for_tts;
#[path = "tts_voice_transition.rs"] #[path = "tts_voice_transition.rs"]
mod voice_transition; mod voice_transition;
use voice_transition::*; use voice_transition::*;
#[path = "tts_playback.rs"]
mod playback;
use playback::*;
#[path = "tts_startup.rs"] #[path = "tts_startup.rs"]
mod startup; mod startup;
use startup::await_worker_startup; use startup::await_worker_startup;
@@ -102,10 +104,26 @@ const SYNTH_STEPS: usize = 1;
/// the leading waveform is important. /// the leading waveform is important.
const FADE_OUT_SAMPLES: usize = (SAMPLE_RATE as f64 * 0.008) as usize; const FADE_OUT_SAMPLES: usize = (SAMPLE_RATE as f64 * 0.008) as usize;
/// Length of the zero-sample cushion prepended when playback is idle, so the /// rodio 0.22.2 bootstraps `UniformSourceIterator` when a source is added to
/// OS audio device / rodio mixer has a fully-quiet ramp-up window before the /// the mixer (`conversions/uniform.rs:49-66`). Its empty queue's 512-sample
/// real onset hits. Continuously queued chunks receive no synthetic padding. /// span (`queue.rs::SourcesQueueInput::new`) can therefore retain placeholder
const SENTENCE_LEAD_IN_SAMPLES: usize = (SAMPLE_RATE as f64 * 0.020) as usize; /// format metadata until the next span. The lead-in covers that whole span,
/// rounded up to the next millisecond, while preserving the product's existing
/// 20 ms quiet ramp-up. Continuously queued chunks receive no synthetic padding.
const SAMPLES_PER_MS: usize = SAMPLE_RATE as usize / 1_000;
const PRODUCT_RAMP_UP_MS: usize = 20;
const PRODUCT_RAMP_UP_SAMPLES: usize = PRODUCT_RAMP_UP_MS * SAMPLES_PER_MS;
const RODIO_ADD_BOOTSTRAP_SPAN_SAMPLES: usize = 512;
const RODIO_ADD_BOOTSTRAP_CUSHION_MS: usize =
RODIO_ADD_BOOTSTRAP_SPAN_SAMPLES.div_ceil(SAMPLES_PER_MS);
const SENTENCE_LEAD_IN_SAMPLES: usize = {
let bootstrap_cushion = RODIO_ADD_BOOTSTRAP_CUSHION_MS * SAMPLES_PER_MS;
if PRODUCT_RAMP_UP_SAMPLES > bootstrap_cushion {
PRODUCT_RAMP_UP_SAMPLES
} else {
bootstrap_cushion
}
};
type WorkerControlState = ( type WorkerControlState = (
Arc<AtomicBool>, Arc<AtomicBool>,
@@ -332,7 +350,6 @@ fn tts_worker(
// ── 3. Initialise rodio output device ───────────────────────────────────── // ── 3. Initialise rodio output device ─────────────────────────────────────
use rodio::buffer::SamplesBuffer; use rodio::buffer::SamplesBuffer;
use rodio::Player;
let sink_handle = match super::audio_output::open_output_sink_by_name(output_device.as_deref()) let sink_handle = match super::audio_output::open_output_sink_by_name(output_device.as_deref())
{ {
@@ -360,28 +377,24 @@ fn tts_worker(
} }
}; };
// Single persistent Player for the lifetime of the worker — all sentence // One coordinator owns the current Player and every operation on it.
// buffers from all text items append here, and rodio plays them gaplessly. // Cancellation replaces its queue while the output stream and mixer stay
// Persistence is what enables cross-item pipelining: the worker never // alive, preserving cross-item pipelining without exposing Player handles.
// waits for one item to drain before synthesizing the next. let playback = Arc::new(PlaybackCoordinator::new(sink_handle.mixer()));
// playback_probe.install(Arc::clone(&playback));
// Shared (Arc) with the barge-in monitor thread below, which needs to
// silence it while this thread is blocked inside `synth_chunk`.
let player = Arc::new(Player::connect_new(sink_handle.mixer()));
playback_probe.install(Arc::clone(&player));
// Prime the audio output stream with a short silent buffer. // Prime the audio output stream with a short silent buffer.
// On macOS, CoreAudio initializes the output device lazily on first use. // On macOS, CoreAudio initializes the output device lazily on first use.
// Without this, the first real append races against device startup and // Without this, the first real append races against device startup and
// player.empty() returns true before audio has started draining — causing // playback.empty() returns true before audio has started draining — causing
// the first TTS message to be truncated after a few words. // the first TTS message to be truncated after a few words.
{ {
let silence = vec![0.0f32; SAMPLE_RATE as usize / 10]; // 100ms of silence let silence = vec![0.0f32; SAMPLE_RATE as usize / 10]; // 100ms of silence
player.append(SamplesBuffer::new(channels, rate, silence)); playback.append_untracked(SamplesBuffer::new(channels, rate, silence));
// Wait for the silent buffer to drain — this ensures the output stream // Wait for the silent buffer to drain — this ensures the output stream
// is fully initialized before the first real utterance. // is fully initialized before the first real utterance.
let deadline = std::time::Instant::now() + AUDIO_PRIME_TIMEOUT; let deadline = std::time::Instant::now() + AUDIO_PRIME_TIMEOUT;
while !player.empty() { while !playback.empty() {
if std::time::Instant::now() >= deadline { if std::time::Instant::now() >= deadline {
eprintln!("buzz-desktop: tts stage=startup status=failed reason=output_prime"); eprintln!("buzz-desktop: tts stage=startup status=failed reason=output_prime");
let _ = startup_tx.send(Err( let _ = startup_tx.send(Err(
@@ -397,16 +410,14 @@ fn tts_worker(
} }
eprintln!("buzz-desktop: tts stage=startup status=ready"); eprintln!("buzz-desktop: tts stage=startup status=ready");
let player_ops = Arc::clone(&playback_probe.player_ops);
let activity_frames = Arc::new(Mutex::new(VecDeque::<TtsSpeakerActivityFrame>::new())); let activity_frames = Arc::new(Mutex::new(VecDeque::<TtsSpeakerActivityFrame>::new()));
let monitor_stop = Arc::new(AtomicBool::new(false)); let monitor_stop = Arc::new(AtomicBool::new(false));
let monitor = spawn_tts_monitor(TtsMonitorState { let monitor = spawn_tts_monitor(TtsMonitorState {
player: Arc::clone(&player), playback: Arc::clone(&playback),
cancel: Arc::clone(&cancel), cancel: Arc::clone(&cancel),
voice_cancel: Arc::clone(&voice_cancel), voice_cancel: Arc::clone(&voice_cancel),
tts_active: Arc::clone(&tts_active), tts_active: Arc::clone(&tts_active),
stop: Arc::clone(&monitor_stop), stop: Arc::clone(&monitor_stop),
player_ops: Arc::clone(&player_ops),
activity_frames: Arc::clone(&activity_frames), activity_frames: Arc::clone(&activity_frames),
active_speaker: Arc::clone(&active_speaker), active_speaker: Arc::clone(&active_speaker),
speaker_cancel: Arc::clone(&speaker_cancel), speaker_cancel: Arc::clone(&speaker_cancel),
@@ -429,74 +440,72 @@ fn tts_worker(
// EXPERIMENTAL (latency bench): `Some(emit_frames)` = stream PCM deltas // EXPERIMENTAL (latency bench): `Some(emit_frames)` = stream PCM deltas
// out of Pocket as they are generated (see tts_streaming.rs). // out of Pocket as they are generated (see tts_streaming.rs).
let tts_streaming = streaming_emit_frames(); let tts_streaming = streaming_emit_frames();
// `first_append` = "no audio queued since the player last went idle".
// Flipped by `build_sentence_append_buffer` on the first real append; the
// idle branch below uses it to decide when to drop `tts_active` and to
// arm a fresh lead-in cushion for the next utterance.
let mut first_append = true;
let mut last_route_id = 0; let mut last_route_id = 0;
let mut deferred_text = VecDeque::new(); let mut deferred_text = VecDeque::new();
let append_audio = |prepared: PreparedModelAudio, let append_audio = |prepared: PreparedModelAudio,
route_id: u64, route_id: u64,
speaker_pubkey: Option<&str>, speaker_pubkey: Option<&str>,
speaker_generation: u64| { speaker_generation: u64| {
let _ops = lock_player_ops(&player_ops); let sample_count = prepared.sample_count;
if cancel.load(Ordering::Acquire) let chunk_index = prepared.chunk_index;
|| voice_cancel.load(Ordering::Acquire) let activity = speaker_pubkey.map(|pubkey| {
|| shutdown.load(Ordering::Acquire) build_tts_speaker_activity_frames(&prepared.buffer, pubkey, SAMPLE_RATE as usize)
{
let reason = if shutdown.load(Ordering::Acquire) {
"shutdown"
} else if cancel.load(Ordering::Acquire) {
"barge_in"
} else {
"voice_switch"
};
eprintln!(
"buzz-desktop: tts stage=synthesis status=cancelled reason={reason} route_id={route_id}"
);
return false;
}
let speaker_is_current = speaker_pubkey.is_none_or(|pubkey| {
current_speaker_generation(&speaker_generations, pubkey) == speaker_generation
}); });
if !speaker_is_current { let accepted = playback.append_if(
eprintln!( SamplesBuffer::new(channels, rate, prepared.buffer),
"buzz-desktop: tts stage=synthesis status=cancelled reason=speaker_removed route_id={route_id}" |player_empty| {
); if cancel.load(Ordering::Acquire)
|| voice_cancel.load(Ordering::Acquire)
|| shutdown.load(Ordering::Acquire)
{
let reason = if shutdown.load(Ordering::Acquire) {
"shutdown"
} else if cancel.load(Ordering::Acquire) {
"barge_in"
} else {
"voice_switch"
};
eprintln!(
"buzz-desktop: tts stage=synthesis status=cancelled reason={reason} route_id={route_id}"
);
return false;
}
if speaker_pubkey.is_some_and(|pubkey| {
current_speaker_generation(&speaker_generations, pubkey) != speaker_generation
}) {
eprintln!(
"buzz-desktop: tts stage=synthesis status=cancelled reason=speaker_removed route_id={route_id}"
);
return false;
}
if let Some(pubkey) = speaker_pubkey {
let mut active = active_speaker
.lock()
.unwrap_or_else(|error| error.into_inner());
if player_empty {
active.take();
}
if active
.as_deref()
.is_some_and(|current| !current.eq_ignore_ascii_case(pubkey))
{
return false;
}
active.get_or_insert_with(|| pubkey.to_ascii_lowercase());
activity_frames
.lock()
.unwrap_or_else(|error| error.into_inner())
.extend(activity.unwrap_or_default());
}
true
},
);
if !accepted {
return false; return false;
} }
if let Some(pubkey) = speaker_pubkey {
let mut active = active_speaker
.lock()
.unwrap_or_else(|error| error.into_inner());
if player.empty() {
active.take();
}
if active
.as_deref()
.is_some_and(|current| !current.eq_ignore_ascii_case(pubkey))
{
return false;
}
active.get_or_insert_with(|| pubkey.to_ascii_lowercase());
}
if let Some(pubkey) = speaker_pubkey {
activity_frames
.lock()
.unwrap_or_else(|error| error.into_inner())
.extend(build_tts_speaker_activity_frames(
&prepared.buffer,
pubkey,
SAMPLE_RATE as usize,
));
}
player.append(SamplesBuffer::new(channels, rate, prepared.buffer));
eprintln!( eprintln!(
"buzz-desktop: tts stage=player status=append_accepted route_id={route_id} chunk_index={} sample_count={}", "buzz-desktop: tts stage=player status=append_accepted route_id={route_id} chunk_index={chunk_index} sample_count={sample_count}"
prepared.chunk_index, prepared.sample_count
); );
// Set this only after append so STT remains open during synthesis.
tts_active.store(true, Ordering::Release); tts_active.store(true, Ordering::Release);
true true
}; };
@@ -509,9 +518,8 @@ fn tts_worker(
&speaker_generations, &speaker_generations,
&tts_active, &tts_active,
(&text_rx, &mut deferred_text, &mut no_current_text), (&text_rx, &mut deferred_text, &mut no_current_text),
Some((&player, &player_ops)), Some(&playback),
) { ) {
first_append = true;
continue; continue;
} }
if handle_cancel_or_shutdown( if handle_cancel_or_shutdown(
@@ -521,14 +529,13 @@ fn tts_worker(
(&text_rx, &mut deferred_text, &mut no_current_text), (&text_rx, &mut deferred_text, &mut no_current_text),
&voice_change_ack, &voice_change_ack,
None, None,
Some((&player, &player_ops)), Some(&playback),
) { ) {
if shutdown.load(Ordering::Acquire) { if shutdown.load(Ordering::Acquire) {
break; break;
} }
// Cancel consumed: queued audio cleared, queue drained. The next // Cancel consumed: queued audio cleared, queue drained. The next
// append starts a new utterance and needs its own lead-in cushion. // append starts a new utterance and needs its own lead-in cushion.
first_append = true;
continue; continue;
} }
@@ -555,7 +562,7 @@ fn tts_worker(
// Nothing queued. If playback has also finished, the agent // Nothing queued. If playback has also finished, the agent
// has gone quiet — release the mic gate and reset the // has gone quiet — release the mic gate and reset the
// lead-in so the next utterance gets a fresh cushion. // lead-in so the next utterance gets a fresh cushion.
if player.empty() && !first_append { playback.release_if_drained(|| {
tts_active.store(false, Ordering::Release); tts_active.store(false, Ordering::Release);
active_speaker active_speaker
.lock() .lock()
@@ -564,8 +571,7 @@ fn tts_worker(
eprintln!( eprintln!(
"buzz-desktop: tts stage=player status=drained route_id={last_route_id}" "buzz-desktop: tts stage=player status=drained route_id={last_route_id}"
); );
first_append = true; });
}
continue; continue;
} }
Err(mpsc::RecvTimeoutError::Disconnected) => break, Err(mpsc::RecvTimeoutError::Disconnected) => break,
@@ -582,12 +588,11 @@ fn tts_worker(
(&text_rx, &mut deferred_text, &mut queued_text), (&text_rx, &mut deferred_text, &mut queued_text),
&voice_change_ack, &voice_change_ack,
pending_route_id, pending_route_id,
Some((&player, &player_ops)), Some(&playback),
) { ) {
if shutdown.load(Ordering::Acquire) { if shutdown.load(Ordering::Acquire) {
break; break;
} }
first_append = true;
continue; continue;
} }
let Some(queued_text) = queued_text else { let Some(queued_text) = queued_text else {
@@ -607,7 +612,7 @@ fn tts_worker(
); );
continue; continue;
} }
if !player.empty() if !playback.empty()
&& queued_text && queued_text
.speaker_pubkey .speaker_pubkey
.as_deref() .as_deref()
@@ -639,18 +644,14 @@ fn tts_worker(
// release stale ownership before doing any potentially slow voice or // release stale ownership before doing any potentially slow voice or
// synthesis work. Serialize the drain decision with Stop and append so // synthesis work. Serialize the drain decision with Stop and append so
// those paths observe one coherent utterance boundary. // those paths observe one coherent utterance boundary.
{ playback.release_if_drained(|| {
let _ops = lock_player_ops(&player_ops); tts_active.store(false, Ordering::Release);
if player.empty() && !first_append { active_speaker
tts_active.store(false, Ordering::Release); .lock()
active_speaker .unwrap_or_else(|error| error.into_inner())
.lock() .take();
.unwrap_or_else(|error| error.into_inner()) eprintln!("buzz-desktop: tts stage=player status=drained route_id={last_route_id}");
.take(); });
eprintln!("buzz-desktop: tts stage=player status=drained route_id={last_route_id}");
first_append = true;
}
}
// From this point until the item finishes, an empty player can mean a // From this point until the item finishes, an empty player can mean a
// voice-preparation or synthesis gap rather than a drained utterance. // voice-preparation or synthesis gap rather than a drained utterance.
@@ -716,9 +717,8 @@ fn tts_worker(
(&text_rx, &mut deferred_text, &mut no_current_text), (&text_rx, &mut deferred_text, &mut no_current_text),
&voice_change_ack, &voice_change_ack,
Some(route_id), Some(route_id),
Some((&player, &player_ops)), Some(&playback),
) { ) {
first_append = true;
synthesis_outcome = "cancelled"; synthesis_outcome = "cancelled";
break; break;
} }
@@ -738,8 +738,7 @@ fn tts_worker(
emit_frames, emit_frames,
(&cancel, &voice_cancel, &shutdown), (&cancel, &voice_cancel, &shutdown),
StreamingPlayback { StreamingPlayback {
player: &player, playback: &playback,
first_append: &mut first_append,
route_id, route_id,
}, },
&mut |prepared| { &mut |prepared| {
@@ -791,9 +790,8 @@ fn tts_worker(
(&text_rx, &mut deferred_text, &mut no_current_text), (&text_rx, &mut deferred_text, &mut no_current_text),
&voice_change_ack, &voice_change_ack,
Some(route_id), Some(route_id),
Some((&player, &player_ops)), Some(&playback),
) { ) {
first_append = true;
synthesis_outcome = "cancelled"; synthesis_outcome = "cancelled";
break 'playback_chunks; break 'playback_chunks;
} }
@@ -817,25 +815,20 @@ fn tts_worker(
// synthesis that completed after cancellation so stale audio // synthesis that completed after cancellation so stale audio
// never reaches the player, while keeping buzz-voice's // never reaches the player, while keeping buzz-voice's
// extracted April engine API unchanged. // extracted April engine API unchanged.
first_append = true;
synthesis_outcome = "cancelled"; synthesis_outcome = "cancelled";
break 'playback_chunks; break 'playback_chunks;
} }
match synthesis { match synthesis {
Ok(samples) if !samples.is_empty() => { Ok(samples) if !samples.is_empty() => {
if let Some(prepared) = playback_audio.push( if let Some(prepared) = playback.prepare_audio(|first_append, empty| {
samples, playback_audio.push(samples, chunk_index, first_append, empty)
chunk_index, }) {
&mut first_append,
player.empty(),
) {
if !append_audio( if !append_audio(
prepared, prepared,
route_id, route_id,
speaker_pubkey.as_deref(), speaker_pubkey.as_deref(),
speaker_generation, speaker_generation,
) { ) {
first_append = true;
synthesis_outcome = "cancelled"; synthesis_outcome = "cancelled";
break 'playback_chunks; break 'playback_chunks;
} }
@@ -857,14 +850,15 @@ fn tts_worker(
} }
} }
} }
if let Some(prepared) = playback_audio.finish(&mut first_append, player.empty()) { if let Some(prepared) = playback
.prepare_audio(|first_append, empty| playback_audio.finish(first_append, empty))
{
if !append_audio( if !append_audio(
prepared, prepared,
route_id, route_id,
speaker_pubkey.as_deref(), speaker_pubkey.as_deref(),
speaker_generation, speaker_generation,
) { ) {
first_append = true;
synthesis_outcome = "cancelled"; synthesis_outcome = "cancelled";
break 'playback_chunks; break 'playback_chunks;
} }
@@ -884,8 +878,8 @@ fn tts_worker(
} }
} }
// Stop the barge-in monitor before exiting — it holds a Player clone, // Stop the barge-in monitor before exiting so an orphaned monitor cannot
// and an orphaned monitor would keep ticking against a dead pipeline. // keep ticking against a dead pipeline.
monitor_stop.store(true, Ordering::Release); monitor_stop.store(true, Ordering::Release);
if let Ok(handle) = monitor { if let Ok(handle) = monitor {
let _ = handle.join(); let _ = handle.join();

262
desktop/src-tauri/src/huddle/tts_playback.rs Normal file
View File

@@ -0,0 +1,262 @@
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use rodio::{mixer::Mixer, Player, Source};
/// Serializes every operation on the TTS player and owns the utterance-boundary
/// bookkeeping that must change atomically when playback is replaced.
///
/// Poison recovery is sound because `PlaybackState` has no partially-valid
/// representation: `Player` replacement is a single assignment, booleans are
/// independently valid at either value, and no mutable reference to the state
/// leaves the locked operation that created it.
pub(super) struct PlaybackCoordinator {
mixer: Mixer,
state: Mutex<PlaybackState>,
}
struct PlaybackState {
player: Player,
first_append: bool,
synthesis_in_flight: bool,
synthesis_generation: u64,
}
pub(super) struct SynthesisFlightGuard {
playback: Arc<PlaybackCoordinator>,
generation: u64,
}
impl Drop for SynthesisFlightGuard {
fn drop(&mut self) {
let mut state = self.playback.lock();
if state.synthesis_generation == self.generation {
state.synthesis_in_flight = false;
}
}
}
impl PlaybackCoordinator {
pub(super) fn new(mixer: &Mixer) -> Self {
Self {
mixer: mixer.clone(),
state: Mutex::new(PlaybackState {
player: Player::connect_new(mixer),
first_append: true,
synthesis_in_flight: false,
synthesis_generation: 0,
}),
}
}
fn lock(&self) -> MutexGuard<'_, PlaybackState> {
self.state.lock().unwrap_or_else(PoisonError::into_inner)
}
pub(super) fn append_if<S>(&self, source: S, authorize: impl FnOnce(bool) -> bool) -> bool
where
S: Source<Item = f32> + Send + 'static,
{
let mut state = self.lock();
if !authorize(state.player.empty()) {
return false;
}
state.player.append(source);
state.first_append = false;
true
}
pub(super) fn append_untracked<S>(&self, source: S)
where
S: Source<Item = f32> + Send + 'static,
{
self.lock().player.append(source);
}
pub(super) fn empty(&self) -> bool {
self.lock().player.empty()
}
pub(super) fn prepare_audio<R>(&self, prepare: impl FnOnce(&mut bool, bool) -> R) -> R {
let mut state = self.lock();
let empty = state.player.empty();
prepare(&mut state.first_append, empty)
}
pub(super) fn release_if_drained(&self, release: impl FnOnce()) -> bool {
let mut state = self.lock();
if !state.player.empty() || state.first_append {
return false;
}
release();
state.first_append = true;
true
}
pub(super) fn begin_synthesis(self: &Arc<Self>) -> SynthesisFlightGuard {
let generation = {
let mut state = self.lock();
state.synthesis_generation = state.synthesis_generation.wrapping_add(1);
state.synthesis_in_flight = true;
state.synthesis_generation
};
SynthesisFlightGuard {
playback: Arc::clone(self),
generation,
}
}
pub(super) fn with_playback_live<R>(&self, observe: impl FnOnce(bool) -> R) -> R {
let state = self.lock();
observe(!state.player.empty() || state.synthesis_in_flight)
}
/// Replace live playback with a fresh queue. The old player is dropped
/// after releasing the coordinator, so rodio's teardown cannot extend the
/// critical section. Concurrent cancel observers elect exactly one
/// replacement because replacement resets both liveness signals.
pub(super) fn cancel_if_live(&self, authorize: impl FnOnce() -> bool) -> bool {
let old_player = {
let mut state = self.lock();
if (state.player.empty() && !state.synthesis_in_flight) || !authorize() {
return false;
}
state.first_append = true;
state.synthesis_in_flight = false;
state.synthesis_generation = state.synthesis_generation.wrapping_add(1);
std::mem::replace(&mut state.player, Player::connect_new(&self.mixer))
};
drop(old_player);
true
}
}
#[cfg(test)]
mod tests {
use std::{
num::NonZero,
sync::{
atomic::{AtomicUsize, Ordering},
Arc, Barrier,
},
thread,
time::{Duration, Instant},
};
use rodio::buffer::SamplesBuffer;
use super::*;
fn coordinator() -> (Arc<PlaybackCoordinator>, rodio::mixer::MixerSource) {
let channels = NonZero::new(1).expect("nonzero channels");
let rate = NonZero::new(24_000).expect("nonzero rate");
let (mixer, source) = rodio::mixer::mixer(channels, rate);
(Arc::new(PlaybackCoordinator::new(&mixer)), source)
}
fn append_second(playback: &PlaybackCoordinator) {
playback.append_if(
SamplesBuffer::new(
NonZero::new(1).expect("nonzero channels"),
NonZero::new(24_000).expect("nonzero rate"),
vec![0.25; 24_000],
),
|_| true,
);
}
#[test]
fn cancel_replaces_playback_without_waiting_for_the_mixer() {
let (playback, _unpulled_source) = coordinator();
append_second(&playback);
let started = Instant::now();
assert!(playback.cancel_if_live(|| true));
assert!(started.elapsed() < Duration::from_millis(50));
assert!(playback.empty());
}
#[test]
fn concurrent_cancel_observers_elect_exactly_one_replacement() {
let (playback, _unpulled_source) = coordinator();
append_second(&playback);
let barrier = Arc::new(Barrier::new(3));
let replacements = Arc::new(AtomicUsize::new(0));
let mut threads = Vec::new();
for _ in 0..2 {
let playback = Arc::clone(&playback);
let barrier = Arc::clone(&barrier);
let replacements = Arc::clone(&replacements);
threads.push(thread::spawn(move || {
barrier.wait();
if playback.cancel_if_live(|| true) {
replacements.fetch_add(1, Ordering::Relaxed);
}
}));
}
barrier.wait();
for thread in threads {
thread.join().expect("cancel observer");
}
assert_eq!(replacements.load(Ordering::Relaxed), 1);
}
#[test]
fn append_and_cancel_are_one_serialized_public_operation() {
let (playback, _unpulled_source) = coordinator();
append_second(&playback);
let append_authorized = Arc::new(Barrier::new(2));
let release_append = Arc::new(Barrier::new(2));
let append_thread = {
let playback = Arc::clone(&playback);
let append_authorized = Arc::clone(&append_authorized);
let release_append = Arc::clone(&release_append);
thread::spawn(move || {
playback.append_if(
SamplesBuffer::new(
NonZero::new(1).expect("nonzero channels"),
NonZero::new(24_000).expect("nonzero rate"),
vec![0.5; 24_000],
),
|_| {
append_authorized.wait();
release_append.wait();
true
},
)
})
};
append_authorized.wait();
let cancel_thread = {
let playback = Arc::clone(&playback);
thread::spawn(move || playback.cancel_if_live(|| true))
};
release_append.wait();
assert!(append_thread.join().expect("append"));
assert!(cancel_thread.join().expect("cancel"));
assert!(
playback.empty(),
"cancel must replace the queue after append"
);
}
#[test]
fn cancellation_rearms_first_append_and_releases_activity_once() {
let (playback, _unpulled_source) = coordinator();
append_second(&playback);
assert!(playback.cancel_if_live(|| true));
assert!(!playback.release_if_drained(|| panic!("fresh replacement is not a drain")));
playback.prepare_audio(|first_append, starts_playback_chunk| {
assert!(*first_append, "replacement must rearm the first append");
assert!(
starts_playback_chunk,
"the first append after replacement must carry the onset cushion"
);
});
append_second(&playback);
assert!(!playback.release_if_drained(|| panic!("queued audio is not drained")));
}
}

50
desktop/src-tauri/src/huddle/tts_speaker_cancellation.rs
View File

@@ -1,12 +1,11 @@
use super::*; use super::*;
pub(super) struct TtsMonitorState { pub(super) struct TtsMonitorState {
pub(super) player: Arc<rodio::Player>, pub(super) playback: Arc<PlaybackCoordinator>,
pub(super) cancel: Arc<AtomicBool>, pub(super) cancel: Arc<AtomicBool>,
pub(super) voice_cancel: Arc<AtomicBool>, pub(super) voice_cancel: Arc<AtomicBool>,
pub(super) tts_active: Arc<AtomicBool>, pub(super) tts_active: Arc<AtomicBool>,
pub(super) stop: Arc<AtomicBool>, pub(super) stop: Arc<AtomicBool>,
pub(super) player_ops: Arc<Mutex<()>>,
pub(super) activity_frames: Arc<Mutex<VecDeque<TtsSpeakerActivityFrame>>>, pub(super) activity_frames: Arc<Mutex<VecDeque<TtsSpeakerActivityFrame>>>,
pub(super) active_speaker: ActiveSpeaker, pub(super) active_speaker: ActiveSpeaker,
pub(super) speaker_cancel: SpeakerCancellation, pub(super) speaker_cancel: SpeakerCancellation,
@@ -20,28 +19,24 @@ pub(super) fn spawn_tts_monitor(state: TtsMonitorState) -> std::io::Result<threa
let mut last_activity_pubkey: Option<String> = None; let mut last_activity_pubkey: Option<String> = None;
let mut next_activity_tick = Instant::now(); let mut next_activity_tick = Instant::now();
while !state.stop.load(Ordering::Acquire) { while !state.stop.load(Ordering::Acquire) {
if state.cancel.load(Ordering::Acquire) if (state.cancel.load(Ordering::Acquire)
|| state.voice_cancel.load(Ordering::Acquire) || state.voice_cancel.load(Ordering::Acquire))
&& state.playback.cancel_if_live(|| {
state.cancel.load(Ordering::Acquire)
|| state.voice_cancel.load(Ordering::Acquire)
})
{ {
let _ops = lock_player_ops(&state.player_ops); state
if state.cancel.load(Ordering::Acquire) .active_speaker
|| state.voice_cancel.load(Ordering::Acquire) .lock()
{ .unwrap_or_else(|error| error.into_inner())
state.player.clear(); .take();
state.player.play(); state.tts_active.store(false, Ordering::Release);
state
.active_speaker
.lock()
.unwrap_or_else(|error| error.into_inner())
.take();
state.tts_active.store(false, Ordering::Release);
}
} }
silence_cancelled_speaker( silence_cancelled_speaker(
&state.speaker_cancel, &state.speaker_cancel,
&state.active_speaker, &state.active_speaker,
&state.player, &state.playback,
&state.player_ops,
&state.tts_active, &state.tts_active,
); );
if let Some(ref app) = state.activity_app { if let Some(ref app) = state.activity_app {
@@ -94,8 +89,7 @@ pub(super) fn spawn_tts_monitor(state: TtsMonitorState) -> std::io::Result<threa
pub(super) fn silence_cancelled_speaker( pub(super) fn silence_cancelled_speaker(
cancellation: &SpeakerCancellation, cancellation: &SpeakerCancellation,
active_speaker: &ActiveSpeaker, active_speaker: &ActiveSpeaker,
player: &rodio::Player, playback: &PlaybackCoordinator,
player_ops: &Mutex<()>,
tts_active: &AtomicBool, tts_active: &AtomicBool,
) { ) {
let Some(cancelled) = cancellation let Some(cancelled) = cancellation
@@ -105,10 +99,7 @@ pub(super) fn silence_cancelled_speaker(
else { else {
return; return;
}; };
let _ops = lock_player_ops(player_ops); if playback.cancel_if_live(|| take_cancelled_active_speaker(&cancelled, active_speaker)) {
if take_cancelled_active_speaker(&cancelled, active_speaker) {
player.clear();
player.play();
tts_active.store(false, Ordering::Release); tts_active.store(false, Ordering::Release);
} }
} }
@@ -133,7 +124,7 @@ pub(super) fn consume_speaker_cancel(
generations: &SpeakerGenerations, generations: &SpeakerGenerations,
tts_active: &AtomicBool, tts_active: &AtomicBool,
text_state: CancelTextState<'_>, text_state: CancelTextState<'_>,
player: Option<(&rodio::Player, &Mutex<()>)>, playback: Option<&PlaybackCoordinator>,
) -> bool { ) -> bool {
let Some(cancelled) = cancellation let Some(cancelled) = cancellation
.lock() .lock()
@@ -145,11 +136,8 @@ pub(super) fn consume_speaker_cancel(
let (text_rx, deferred_text, current_text) = text_state; let (text_rx, deferred_text, current_text) = text_state;
retain_current_speaker_text(generations, deferred_text, current_text, text_rx); retain_current_speaker_text(generations, deferred_text, current_text, text_rx);
let mut cleared_player = false; let mut cleared_player = false;
if let Some((player, player_ops)) = player { if let Some(playback) = playback {
let _ops = lock_player_ops(player_ops); if playback.cancel_if_live(|| take_cancelled_active_speaker(&cancelled, active_speaker)) {
if take_cancelled_active_speaker(&cancelled, active_speaker) {
player.clear();
player.play();
tts_active.store(false, Ordering::Release); tts_active.store(false, Ordering::Release);
cleared_player = true; cleared_player = true;
} }

21
desktop/src-tauri/src/huddle/tts_streaming.rs
View File

@@ -27,8 +27,7 @@ pub(super) fn streaming_emit_frames() -> Option<usize> {
/// Playback context threaded through one streamed chunk. /// Playback context threaded through one streamed chunk.
pub(super) struct StreamingPlayback<'a> { pub(super) struct StreamingPlayback<'a> {
pub(super) player: &'a rodio::Player, pub(super) playback: &'a PlaybackCoordinator,
pub(super) first_append: &'a mut bool,
pub(super) route_id: u64, pub(super) route_id: u64,
} }
@@ -52,11 +51,7 @@ pub(super) fn synthesize_streaming(
append_audio: &mut dyn FnMut(PreparedModelAudio) -> bool, append_audio: &mut dyn FnMut(PreparedModelAudio) -> bool,
) -> Option<&'static str> { ) -> Option<&'static str> {
let (cancel, voice_cancel, shutdown) = signals; let (cancel, voice_cancel, shutdown) = signals;
let StreamingPlayback { let StreamingPlayback { playback, route_id } = playback;
player,
first_append,
route_id,
} = playback;
let mut playback_audio = PlaybackChunkAudio::new(); let mut playback_audio = PlaybackChunkAudio::new();
let mut delta_index = 0usize; let mut delta_index = 0usize;
let stream_result = engine.synth_chunk_streaming(text, style, emit_frames, &mut |samples| { let stream_result = engine.synth_chunk_streaming(text, style, emit_frames, &mut |samples| {
@@ -68,9 +63,9 @@ pub(super) fn synthesize_streaming(
} }
let chunk_index = delta_index; let chunk_index = delta_index;
delta_index += 1; delta_index += 1;
if let Some(prepared) = if let Some(prepared) = playback.prepare_audio(|first_append, empty| {
playback_audio.push(samples, chunk_index, first_append, player.empty()) playback_audio.push(samples, chunk_index, first_append, empty)
{ }) {
if !append_audio(prepared) { if !append_audio(prepared) {
return false; return false;
} }
@@ -79,9 +74,10 @@ pub(super) fn synthesize_streaming(
}); });
match stream_result { match stream_result {
Ok(true) => { Ok(true) => {
if let Some(prepared) = playback_audio.finish(first_append, player.empty()) { if let Some(prepared) = playback
.prepare_audio(|first_append, empty| playback_audio.finish(first_append, empty))
{
if !append_audio(prepared) { if !append_audio(prepared) {
*first_append = true;
return Some("cancelled"); return Some("cancelled");
} }
} }
@@ -91,7 +87,6 @@ pub(super) fn synthesize_streaming(
eprintln!( eprintln!(
"buzz-desktop: tts stage=synthesis status=cancelled reason=stream_callback route_id={route_id}" "buzz-desktop: tts stage=synthesis status=cancelled reason=stream_callback route_id={route_id}"
); );
*first_append = true;
Some("cancelled") Some("cancelled")
} }
Err(_) => { Err(_) => {

6
desktop/src-tauri/src/huddle/tts_tests/token_split.rs
View File

@@ -1,9 +1,11 @@
use super::*; use super::*;
/// The onset cushion covers 20 ms at the production sample rate. /// The onset cushion rounds rodio's 512-sample bootstrap span up to 22 ms at
/// the 24 kHz production sample rate (528 samples).
#[test] #[test]
fn chunk_lead_in_is_sane() { fn chunk_lead_in_is_sane() {
assert_eq!(SENTENCE_LEAD_IN_SAMPLES, 480, "20 ms × 24 kHz"); assert_eq!(RODIO_ADD_BOOTSTRAP_SPAN_SAMPLES, 512);
assert_eq!(SENTENCE_LEAD_IN_SAMPLES, 528, "22 ms × 24 kHz");
} }
/// Model-token splits remain contiguous: only the playback chunk as a whole /// Model-token splits remain contiguous: only the playback chunk as a whole

118
desktop/src-tauri/src/huddle/tts_voice_transition.rs
View File

@@ -5,10 +5,12 @@ use std::{
sync::{ sync::{
atomic::{AtomicBool, AtomicU64, Ordering}, atomic::{AtomicBool, AtomicU64, Ordering},
mpsc::{self, SyncSender}, mpsc::{self, SyncSender},
Arc, Mutex, MutexGuard, PoisonError, Arc, Mutex,
}, },
}; };
use super::{PlaybackCoordinator, SynthesisFlightGuard};
use crate::huddle::pocket::{load_voice_style, VoiceStyle, DEFAULT_VOICE, VOICE_FILE_EXT}; use crate::huddle::pocket::{load_voice_style, VoiceStyle, DEFAULT_VOICE, VOICE_FILE_EXT};
#[derive(Debug)] #[derive(Debug)]
@@ -32,51 +34,29 @@ pub(super) type CancelSignals<'a> = (&'a AtomicBool, &'a AtomicBool);
#[derive(Clone)] #[derive(Clone)]
pub(super) struct PlaybackProbe { pub(super) struct PlaybackProbe {
player: Arc<Mutex<Option<Arc<rodio::Player>>>>, playback: Arc<Mutex<Option<Arc<PlaybackCoordinator>>>>,
pub(super) player_ops: Arc<Mutex<()>>,
synthesis_in_flight: Arc<AtomicBool>,
}
pub(super) struct SynthesisFlightGuard {
playback_probe: PlaybackProbe,
}
impl Drop for SynthesisFlightGuard {
fn drop(&mut self) {
self.playback_probe.set_synthesis_in_flight(false);
}
} }
impl PlaybackProbe { impl PlaybackProbe {
pub(super) fn new() -> Self { pub(super) fn new() -> Self {
Self { Self {
player: Arc::new(Mutex::new(None)), playback: Arc::new(Mutex::new(None)),
player_ops: Arc::new(Mutex::new(())),
synthesis_in_flight: Arc::new(AtomicBool::new(false)),
} }
} }
pub(super) fn install(&self, player: Arc<rodio::Player>) { pub(super) fn install(&self, playback: Arc<PlaybackCoordinator>) {
self.player self.playback
.lock() .lock()
.unwrap_or_else(|error| error.into_inner()) .unwrap_or_else(|error| error.into_inner())
.replace(player); .replace(playback);
} }
pub(super) fn set_synthesis_in_flight(&self, in_flight: bool) { pub(super) fn begin_synthesis(&self) -> Option<SynthesisFlightGuard> {
let _ops = lock_player_ops(&self.player_ops); self.playback().map(|playback| playback.begin_synthesis())
self.synthesis_in_flight.store(in_flight, Ordering::Release);
} }
pub(super) fn begin_synthesis(&self) -> SynthesisFlightGuard { pub(super) fn playback(&self) -> Option<Arc<PlaybackCoordinator>> {
self.set_synthesis_in_flight(true); self.playback
SynthesisFlightGuard {
playback_probe: self.clone(),
}
}
fn player(&self) -> Option<Arc<rodio::Player>> {
self.player
.lock() .lock()
.unwrap_or_else(|error| error.into_inner()) .unwrap_or_else(|error| error.into_inner())
.clone() .clone()
@@ -199,19 +179,18 @@ pub(super) fn request_active_speaker_cancel(
playback_probe: &PlaybackProbe, playback_probe: &PlaybackProbe,
expected_speaker_pubkey: &str, expected_speaker_pubkey: &str,
) -> bool { ) -> bool {
let Some(player) = playback_probe.player() else { let Some(playback) = playback_probe.playback() else {
return false; return false;
}; };
let _ops = lock_player_ops(&playback_probe.player_ops); playback.with_playback_live(|playback_live| {
let playback_live = request_active_speaker_cancel_while_locked(
!player.empty() || playback_probe.synthesis_in_flight.load(Ordering::Acquire); generations,
request_active_speaker_cancel_while_locked( active_speaker,
generations, cancellation,
active_speaker, playback_live,
cancellation, expected_speaker_pubkey,
playback_live, )
expected_speaker_pubkey, })
)
} }
fn request_active_speaker_cancel_while_locked( fn request_active_speaker_cancel_while_locked(
@@ -475,10 +454,8 @@ fn log_cancelled_route(route_id: u64, reason: &str) {
/// Check for cancel or shutdown. Returns `true` if the caller should break/continue. /// Check for cancel or shutdown. Returns `true` if the caller should break/continue.
/// On cancel: drains the text queue and clears the cancel flag. /// On cancel: drains the text queue and clears the cancel flag.
/// ///
/// `player` pairs the Player with the `player_ops` mutex shared with the /// `playback` is the coordinator shared with the barge-in monitor; replacing
/// barge-in monitor thread; the cancel/shutdown clear runs under that lock so /// playback is serialized with append and with the monitor's stale observation.
/// it is serialized with the monitor's stale-branch re-check (see the monitor
/// block in `tts_worker`).
pub(super) fn handle_cancel_or_shutdown( pub(super) fn handle_cancel_or_shutdown(
cancel_signals: CancelSignals<'_>, cancel_signals: CancelSignals<'_>,
shutdown: &AtomicBool, shutdown: &AtomicBool,
@@ -486,7 +463,7 @@ pub(super) fn handle_cancel_or_shutdown(
text_state: CancelTextState<'_>, text_state: CancelTextState<'_>,
voice_change_ack: &VoiceChangeAck, voice_change_ack: &VoiceChangeAck,
active_route_id: Option<u64>, active_route_id: Option<u64>,
player: Option<(&rodio::Player, &Mutex<()>)>, playback: Option<&PlaybackCoordinator>,
) -> bool { ) -> bool {
let (cancel, voice_cancel) = cancel_signals; let (cancel, voice_cancel) = cancel_signals;
let (text_rx, deferred_text, current_text) = text_state; let (text_rx, deferred_text, current_text) = text_state;
@@ -495,9 +472,8 @@ pub(super) fn handle_cancel_or_shutdown(
"buzz-desktop: tts stage=cancellation reason=shutdown route_id={}", "buzz-desktop: tts stage=cancellation reason=shutdown route_id={}",
active_route_id.unwrap_or(0) active_route_id.unwrap_or(0)
); );
if let Some((p, ops)) = player { if let Some(playback) = playback {
let _ops = lock_player_ops(ops); playback.cancel_if_live(|| true);
p.clear();
} }
tts_active.store(false, Ordering::Release); tts_active.store(false, Ordering::Release);
return true; return true;
@@ -525,18 +501,11 @@ pub(super) fn handle_cancel_or_shutdown(
}) })
.flatten(); .flatten();
retain_cancelled_text(deferred_text, current_text, text_rx, preserve_generation); retain_cancelled_text(deferred_text, current_text, text_rx, preserve_generation);
if let Some((p, ops)) = player { if let Some(playback) = playback {
let _ops = lock_player_ops(ops); // Consume the flag at the coordinator serialization point: once
// `Player::clear()` removes queued sources AND pauses the player // released with `cancel == false`, a stale monitor observation
// (rodio 0.22 `clear()` ends with `self.pause()`). With one // cannot replace fresh post-cancel playback.
// persistent Player for the worker's lifetime, the un-pause is playback.cancel_if_live(|| true);
// mandatory: without `play()`, every append after a barge-in
// would queue silently forever.
p.clear();
p.play();
// Consume the flag under the lock: once released with
// `cancel == false`, the monitor's stale branch no-ops instead
// of clearing the fresh post-cancel utterance.
} }
tts_active.store(false, Ordering::Release); tts_active.store(false, Ordering::Release);
return true; return true;
@@ -544,16 +513,6 @@ pub(super) fn handle_cancel_or_shutdown(
false false
} }
/// Acquire the `player_ops` lock, recovering from poison.
///
/// The data under the mutex is `()` — it only serializes Player mutations —
/// so a panicked holder leaves nothing inconsistent to observe and recovery
/// is always safe. Without this, a worker panic would wedge the monitor (or
/// vice versa) on `unwrap()`.
pub(super) fn lock_player_ops(ops: &Mutex<()>) -> MutexGuard<'_, ()> {
ops.lock().unwrap_or_else(PoisonError::into_inner)
}
#[cfg(test)] #[cfg(test)]
mod speaker_generation_tests { mod speaker_generation_tests {
use super::*; use super::*;
@@ -562,16 +521,15 @@ mod speaker_generation_tests {
let channels = std::num::NonZero::new(1).expect("non-zero channels"); let channels = std::num::NonZero::new(1).expect("non-zero channels");
let sample_rate = std::num::NonZero::new(24_000).expect("non-zero sample rate"); let sample_rate = std::num::NonZero::new(24_000).expect("non-zero sample rate");
let (mixer, _mixer_source) = rodio::mixer::mixer(channels, sample_rate); let (mixer, _mixer_source) = rodio::mixer::mixer(channels, sample_rate);
let player = Arc::new(rodio::Player::connect_new(&mixer)); let playback = Arc::new(PlaybackCoordinator::new(&mixer));
if playback_live { if playback_live {
player.append(rodio::buffer::SamplesBuffer::new( playback.append_if(
channels, rodio::buffer::SamplesBuffer::new(channels, sample_rate, vec![0.0; 24_000]),
sample_rate, |_| true,
vec![0.0; 24_000], );
));
} }
let probe = PlaybackProbe::new(); let probe = PlaybackProbe::new();
probe.install(player); probe.install(playback);
probe probe
} }