diff --git a/native/src/audio/engine.rs b/native/src/audio/engine.rs index 0a93f28..d9c0498 100644 --- a/native/src/audio/engine.rs +++ b/native/src/audio/engine.rs @@ -36,18 +36,21 @@ const BLOCK: usize = 480; // 10 ms at 48 kHz const FRAME: usize = 960; // 20 ms at 48 kHz const MAX_DECODED: usize = 5760; // 120 ms at 48 kHz const MAX_CONCEALED_PACKETS: u16 = 5; +const MAX_PENDING_PLAYBACK_MS: usize = 1_000; +const CLEAR_ACK_TIMEOUT: Duration = Duration::from_secs(1); const LEVEL_INTERVAL: Duration = Duration::from_millis(50); +const MAX_BLOCKS_PER_SERVICE: usize = 2; const SERVICE_INTERVAL: Duration = Duration::from_millis(5); #[derive(Default)] pub struct EngineControl { pub muted: AtomicBool, pub stop: AtomicBool, + pub clear: AtomicBool, } pub struct Engine { pub control: Arc, - playback: Arc, thread: Option>, } @@ -66,7 +69,6 @@ impl Engine { .take() .ok_or_else(|| anyhow::anyhow!("audio already started"))?; let state = State::new(io.input.rate, io.output.rate)?; - let playback = io.control.clone(); let thread_control = control.clone(); let thread = std::thread::Builder::new() .name("audio-engine".into()) @@ -88,7 +90,6 @@ impl Engine { })?; Ok(Self { control, - playback, thread: Some(thread), }) } @@ -98,7 +99,7 @@ impl Engine { } pub fn clear_output(&self) { - self.playback.clear.store(true, Ordering::Release); + self.control.clear.store(true, Ordering::Release); } } @@ -153,6 +154,7 @@ struct State { pending: Vec, expected_sequence: Option, output_rate: u32, + clear_started: Option, mic: Meter, speaker: Meter, scratch: Vec, @@ -185,13 +187,18 @@ impl State { Ok(Self { capture: Converter::new(input_rate, RATE)?, reference: Converter::new(output_rate, RATE)?, - playback: Converter::new(RATE, output_rate)?, + playback: Converter::new_bounded( + RATE, + output_rate, + (output_rate as usize * MAX_PENDING_PLAYBACK_MS) / 1000, + )?, apm, encoder, decoder, pending: Vec::with_capacity(FRAME), expected_sequence: None, output_rate, + clear_started: None, mic: Meter::new(), speaker: Meter::new(), scratch: vec![0.0; 8192], @@ -210,6 +217,10 @@ impl State { let mut last_levels = Instant::now(); let mut was_muted = control.muted.load(Ordering::Acquire); while !control.stop.load(Ordering::Acquire) { + // Service incoming audio before DSP work: a capture backlog must not delay + // speech until the transcript (which arrives on another thread) is complete. + self.service_playback(&mut rings.playback, incoming, control, &io.control)?; + // Speaker reference first so capture blocks see an up-to-date echo model. loop { let read = rings.reference.pop_slice(&mut self.scratch); @@ -220,7 +231,10 @@ impl State { } let mut block = [0.0f32; BLOCK]; let mut processed = [0.0f32; BLOCK]; - while self.reference.take_exact(&mut block) { + for _ in 0..MAX_BLOCKS_PER_SERVICE { + if !self.reference.take_exact(&mut block) { + break; + } self.apm .process_render_f32(&[&block], &mut [&mut processed]) .map_err(|_| anyhow::anyhow!("echo reference processing failed"))?; @@ -242,7 +256,10 @@ impl State { let playback_delay_ms = rings.playback.occupied_len() as f64 * 1000.0 / f64::from(self.output_rate); let delay = (playback_delay_ms + 20.0).clamp(0.0, 500.0) as i32; - while self.capture.take_exact(&mut block) { + for _ in 0..MAX_BLOCKS_PER_SERVICE { + if !self.capture.take_exact(&mut block) { + break; + } let _ = self.apm.set_stream_delay_ms(delay); self.apm .process_capture_f32(&[&block], &mut [&mut processed]) @@ -264,15 +281,6 @@ impl State { } } - // Playback. - while let Ok(packet) = incoming.try_recv() { - self.receive(packet)?; - } - if self.playback.available() > 0 { - let samples: Vec = self.playback.drain_all().collect(); - rings.playback.push_slice(&samples); - } - if last_levels.elapsed() >= LEVEL_INTERVAL { last_levels = Instant::now(); emitter.emit(serde_json::json!({ @@ -288,6 +296,53 @@ impl State { Ok(()) } + fn service_playback( + &mut self, + sink: &mut ringbuf::HeapProd, + incoming: &std_mpsc::Receiver, + control: &EngineControl, + playback: &super::io::PlaybackControl, + ) -> anyhow::Result<()> { + if control.clear.swap(false, Ordering::AcqRel) { + self.playback.clear(); + if !playback.clear.swap(true, Ordering::AcqRel) { + self.clear_started = Some(Instant::now()); + } + } + // Leave the producer idle until the renderer acknowledges clearing its ring. + if playback.clear.load(Ordering::Acquire) { + if self + .clear_started + .is_some_and(|started| started.elapsed() >= CLEAR_ACK_TIMEOUT) + && playback.clear.load(Ordering::Acquire) + { + anyhow::bail!("speaker did not acknowledge playback clear within 1 second"); + } + return Ok(()); + } + self.clear_started = None; + // Flush after every packet, so a large batch moves into the speaker ring as it is + // decoded instead of overflowing the pending cap while the ring still has room. + for _ in 0..32 { + let Ok(packet) = incoming.try_recv() else { + break; + }; + self.receive(packet)?; + self.flush_playback(sink); + } + self.flush_playback(sink); + Ok(()) + } + + fn flush_playback(&mut self, sink: &mut ringbuf::HeapProd) { + let capacity = sink.vacant_len().min(self.playback.available()); + if capacity > 0 { + let samples: Vec = self.playback.drain(capacity).collect(); + let written = sink.push_slice(&samples); + debug_assert_eq!(written, samples.len()); + } + } + fn receive(&mut self, packet: IncomingPacket) -> anyhow::Result<()> { let mut decoded = [0.0f32; MAX_DECODED]; if let Some(expected) = self.expected_sequence { @@ -326,3 +381,161 @@ impl State { self.playback.push(samples) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn speaker_backpressure_preserves_every_sample_in_order() { + use ringbuf::traits::Split; + let mut state = State::new(RATE, RATE).unwrap(); + let (mut prod, mut cons) = ringbuf::HeapRb::::new(2).split(); + state.playback.push(&[1.0, 2.0, 3.0, 4.0, 5.0]).unwrap(); + state.flush_playback(&mut prod); + assert_eq!(state.playback.available(), 3); + // A second service while full must not consume queued samples. + state.flush_playback(&mut prod); + assert_eq!(state.playback.available(), 3); + let mut actual = Vec::new(); + while state.playback.available() > 0 || !cons.is_empty() { + while let Some(sample) = cons.try_pop() { + actual.push(sample); + } + state.flush_playback(&mut prod); + } + assert_eq!(actual, vec![1.0, 2.0, 3.0, 4.0, 5.0]); + } + + #[test] + fn clear_discards_overflow_and_waits_for_the_renderer_before_refilling() { + use ringbuf::traits::Split; + let mut state = State::new(RATE, RATE).unwrap(); + let control = EngineControl::default(); + let playback = Arc::new(super::super::io::PlaybackControl::default()); + let (mut prod, cons) = ringbuf::HeapRb::::new(2).split(); + let (reference, _) = ringbuf::HeapRb::::new(20).split(); + let mut renderer = super::super::io::Renderer::new(cons, reference, playback.clone(), 10); + let (_tx, incoming) = std_mpsc::channel(); + state.playback.push(&[1.0, 2.0, 3.0, 4.0, 5.0]).unwrap(); + state.flush_playback(&mut prod); + assert_eq!(state.playback.available(), 3); + for _ in 0..2 { + control.clear.store(true, Ordering::Release); + state + .service_playback(&mut prod, &incoming, &control, &playback) + .unwrap(); + assert_eq!(state.playback.available(), 0); + assert!(playback.clear.load(Ordering::Acquire)); + } + state.playback.push(&[6.0, 7.0]).unwrap(); + state + .service_playback(&mut prod, &incoming, &control, &playback) + .unwrap(); + assert_eq!(state.playback.available(), 2); + let mut out = [0.0; 2]; + renderer.render(&mut out); + assert_eq!(out, [0.0; 2]); + assert!(!playback.clear.load(Ordering::Acquire)); + state + .service_playback(&mut prod, &incoming, &control, &playback) + .unwrap(); + renderer.render(&mut out); + assert_eq!(out, [6.0, 7.0]); + assert_eq!(state.playback.available(), 0); + } + + #[test] + fn a_full_batch_of_packets_reaches_the_speaker_ring_without_dropping() { + use ringbuf::traits::Observer; + use ringbuf::traits::Split; + let mut state = State::new(RATE, RATE).unwrap(); + let control = EngineControl::default(); + let playback = super::super::io::PlaybackControl::default(); + let (mut sink, _cons) = ringbuf::HeapRb::::new(RATE as usize * 4).split(); + let (tx, incoming) = std_mpsc::channel(); + let mut encoder = + opus::Encoder::new(RATE, opus::Channels::Mono, opus::Application::Voip).unwrap(); + let packet = vec![0.1f32; FRAME * 2]; // 40 ms + for sequence in 0..32u16 { + let payload = encoder.encode_vec_float(&packet, 4000).unwrap(); + tx.send(IncomingPacket { sequence, payload }).unwrap(); + } + state + .service_playback(&mut sink, &incoming, &control, &playback) + .unwrap(); + // 32 x 40 ms is more than the 1 s pending cap, but fits the 4 s speaker ring. + assert_eq!( + sink.occupied_len() + state.playback.available(), + 32 * packet.len() + ); + } + + #[test] + fn playback_pending_audio_is_bounded_and_keeps_the_newest_samples() { + let mut state = State::new(RATE, RATE).unwrap(); + let samples: Vec = (0..RATE as usize + 10) + .map(|sample| sample as f32) + .collect(); + state.playback.push(&samples).unwrap(); + assert_eq!(state.playback.available(), RATE as usize); + assert_eq!(state.playback.drain(1).next(), Some(10.0)); + } + + #[test] + fn clear_ack_timeout_fails_instead_of_leaving_audio_silently_stalled() { + use ringbuf::traits::Split; + let mut state = State::new(RATE, RATE).unwrap(); + let control = EngineControl::default(); + let playback = Arc::new(super::super::io::PlaybackControl::default()); + playback.clear.store(true, Ordering::Release); + state.clear_started = Some(Instant::now() - CLEAR_ACK_TIMEOUT - Duration::from_millis(1)); + let (mut sink, _) = ringbuf::HeapRb::::new(2).split(); + let (_tx, incoming) = std_mpsc::channel(); + + let error = state + .service_playback(&mut sink, &incoming, &control, &playback) + .unwrap_err(); + assert!( + error + .to_string() + .contains("did not acknowledge playback clear") + ); + } + + // Offline: no microphone, devices, credentials or network. Run explicitly to + // check the DSP budget when evaluating a helper build on a target machine. + #[test] + #[ignore = "manual DSP timing diagnostic"] + fn one_second_dsp_budget() { + let mut state = State::new(RATE, RATE).unwrap(); + let input: Vec = (0..BLOCK) + .map(|i| ((i as f32 / RATE as f32) * 440.0 * std::f32::consts::TAU).sin() * 0.1) + .collect(); + let mut processed = [0.0; BLOCK]; + let mut encoded = [0u8; 1275]; + let started = Instant::now(); + for _ in 0..100 { + state + .apm + .process_render_f32(&[&input], &mut [&mut processed]) + .unwrap(); + state + .apm + .process_capture_f32(&[&input], &mut [&mut processed]) + .unwrap(); + state.pending.extend_from_slice(&processed); + if state.pending.len() >= FRAME { + state + .encoder + .encode_float(&state.pending[..FRAME], &mut encoded) + .unwrap(); + state.pending.drain(..FRAME); + } + } + eprintln!( + "DSP processing for 1000 ms of audio: {} ms", + started.elapsed().as_millis() + ); + } +} diff --git a/native/src/audio/io.rs b/native/src/audio/io.rs index ea5ee5b..2704619 100644 --- a/native/src/audio/io.rs +++ b/native/src/audio/io.rs @@ -314,16 +314,18 @@ where } /// Renders mono speaker samples, handling prebuffering after underruns and clear requests. -struct Renderer { +pub(super) struct Renderer { playback: HeapCons, reference: HeapProd, control: Arc, starved: bool, prebuffer: usize, + waited: usize, + max_wait: usize, } impl Renderer { - fn new( + pub(super) fn new( playback: HeapCons, reference: HeapProd, control: Arc, @@ -336,16 +338,31 @@ impl Renderer { starved: true, // 60 ms of jitter absorption before speech starts playing. prebuffer: rate as usize * 60 / 1000, + waited: 0, + // A short response or tail must not wait for another utterance to reach 60 ms. + max_wait: rate as usize / 10, } } - fn render(&mut self, out: &mut [f32]) { - if self.control.clear.swap(false, Ordering::AcqRel) { + pub(super) fn render(&mut self, out: &mut [f32]) { + if self.control.clear.load(Ordering::Acquire) { self.playback.clear(); self.starved = true; + self.waited = 0; + // The engine must not refill the ring until it has been emptied. + self.control.clear.store(false, Ordering::Release); } - if self.starved && self.playback.occupied_len() >= self.prebuffer { - self.starved = false; + if self.starved { + let available = self.playback.occupied_len(); + if available > 0 { + self.waited = self.waited.saturating_add(out.len()); + if available >= self.prebuffer || self.waited >= self.max_wait { + self.starved = false; + self.waited = 0; + } + } else { + self.waited = 0; + } } let read = if self.starved { 0 @@ -543,3 +560,42 @@ fn run_virtual_output( let _ = writer.finalize(); } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn short_response_plays_without_waiting_for_the_next_turn() { + let (mut prod, cons) = HeapRb::::new(200).split(); + let (reference, _) = HeapRb::::new(200).split(); + let mut renderer = + Renderer::new(cons, reference, Arc::new(PlaybackControl::default()), 1000); + prod.push_slice(&[0.5; 20]); + let mut out = [0.0; 10]; + for _ in 0..9 { + renderer.render(&mut out); + assert_eq!(out, [0.0; 10]); + } + renderer.render(&mut out); + assert_eq!(out, [0.5; 10]); + renderer.render(&mut out); + assert_eq!(out, [0.5; 10]); + } + + #[test] + fn normal_prebuffer_starts_immediately_and_clear_discards_it() { + let (mut prod, cons) = HeapRb::::new(200).split(); + let (reference, _) = HeapRb::::new(200).split(); + let control = Arc::new(PlaybackControl::default()); + let mut renderer = Renderer::new(cons, reference, control.clone(), 1000); + prod.push_slice(&[0.5; 60]); + let mut out = [0.0; 10]; + renderer.render(&mut out); + assert_eq!(out, [0.5; 10]); + control.clear.store(true, Ordering::Release); + renderer.render(&mut out); + assert_eq!(out, [0.0; 10]); + assert!(renderer.playback.is_empty()); + } +} diff --git a/native/src/audio/resample.rs b/native/src/audio/resample.rs index 8528d6a..d9a2d8d 100644 --- a/native/src/audio/resample.rs +++ b/native/src/audio/resample.rs @@ -8,6 +8,7 @@ use rubato::audioadapter_buffers::direct::InterleavedSlice; pub struct Converter { inner: Option, output: VecDeque, + output_limit: Option, } struct Inner { @@ -40,12 +41,36 @@ impl Converter { Ok(Self { inner, output: VecDeque::new(), + output_limit: None, }) } + pub fn new_bounded(from: u32, to: u32, max_output: usize) -> anyhow::Result { + let mut converter = Self::new(from, to)?; + converter.output_limit = Some(max_output); + Ok(converter) + } + + fn append_output(output: &mut VecDeque, limit: Option, samples: &[f32]) { + let Some(limit) = limit else { + output.extend(samples); + return; + }; + let skip = samples.len().saturating_sub(limit); + let samples = &samples[skip..]; + let excess = output + .len() + .saturating_add(samples.len()) + .saturating_sub(limit); + if excess > 0 { + output.drain(..excess); + } + output.extend(samples); + } + pub fn push(&mut self, samples: &[f32]) -> anyhow::Result<()> { let Some(inner) = &mut self.inner else { - self.output.extend(samples); + Self::append_output(&mut self.output, self.output_limit, samples); return Ok(()); }; inner.input.extend(samples); @@ -62,7 +87,11 @@ impl Converter { .process_into_buffer(&input, &mut output, None) .map_err(|error| anyhow::anyhow!("resampling failed: {error}"))?; inner.input.drain(..consumed); - self.output.extend(&inner.scratch[..produced]); + Self::append_output( + &mut self.output, + self.output_limit, + &inner.scratch[..produced], + ); } Ok(()) } @@ -71,6 +100,14 @@ impl Converter { self.output.len() } + pub fn clear(&mut self) { + self.output.clear(); + if let Some(inner) = &mut self.inner { + inner.input.clear(); + inner.resampler.reset(); + } + } + /// Takes exactly `out.len()` samples when enough are buffered. pub fn take_exact(&mut self, out: &mut [f32]) -> bool { if self.output.len() < out.len() { @@ -83,7 +120,68 @@ impl Converter { true } - pub fn drain_all(&mut self) -> std::collections::vec_deque::Drain<'_, f32> { - self.output.drain(..) + /// Drain only samples the speaker ring can accept, preserving any overflow. + pub fn drain(&mut self, limit: usize) -> std::collections::vec_deque::Drain<'_, f32> { + self.output.drain(..limit.min(self.output.len())) + } +} + +#[cfg(test)] +mod tests { + use super::Converter; + + #[test] + fn partial_playback_drain_preserves_order_and_overflow() { + let mut converter = Converter::new(48_000, 48_000).unwrap(); + converter.push(&[1.0, 2.0, 3.0, 4.0]).unwrap(); + assert_eq!(converter.drain(2).collect::>(), vec![1.0, 2.0]); + assert_eq!(converter.available(), 2); + assert_eq!(converter.drain(8).collect::>(), vec![3.0, 4.0]); + } + + #[test] + fn full_speaker_ring_does_not_discard_pending_audio() { + let mut converter = Converter::new(48_000, 48_000).unwrap(); + converter.push(&[1.0, 2.0]).unwrap(); + assert_eq!(converter.drain(0).count(), 0); + assert_eq!(converter.available(), 2); + } + + #[test] + fn bounded_output_drops_oldest_samples() { + let mut converter = Converter::new_bounded(48_000, 48_000, 4).unwrap(); + converter.push(&[1.0, 2.0, 3.0, 4.0]).unwrap(); + converter.push(&[5.0, 6.0]).unwrap(); + assert_eq!(converter.available(), 4); + assert_eq!( + converter.drain(usize::MAX).collect::>(), + vec![3.0, 4.0, 5.0, 6.0] + ); + } + + #[test] + fn bounded_output_stays_within_limit_after_resampling() { + let mut converter = Converter::new_bounded(48_000, 44_100, 32).unwrap(); + converter.push(&[0.25; 4_800]).unwrap(); + assert_eq!(converter.available(), 32); + } + + #[test] + fn clear_discards_output_partial_input_and_resampler_history() { + for rate in [48_000, 44_100] { + let mut converter = Converter::new(48_000, rate).unwrap(); + converter.push(&[1.0; 4_801]).unwrap(); + assert!(converter.available() > 0); + converter.clear(); + assert_eq!(converter.available(), 0); + let mut fresh = Converter::new(48_000, rate).unwrap(); + let samples = [0.25; 4_800]; + converter.push(&samples).unwrap(); + fresh.push(&samples).unwrap(); + assert_eq!( + converter.drain(usize::MAX).collect::>(), + fresh.drain(usize::MAX).collect::>() + ); + } } } diff --git a/native/src/main.rs b/native/src/main.rs index 5749a68..510e939 100644 --- a/native/src/main.rs +++ b/native/src/main.rs @@ -204,7 +204,7 @@ impl Session { emitter.emit(serde_json::json!({ "type": "muted", "muted": muted })); } Command::Clear {} => { - // Speaker queue is owned by the output endpoint; the engine forwards the flag. + // The engine clears its pending audio before asking the endpoint to clear. if let Some(engine) = &self.engine { engine.clear_output(); }