From 571b3d43a69f8fa8629f88b7bcd0d4bb06d9e57f Mon Sep 17 00:00:00 2001 From: Roderick van Domburg Date: Fri, 31 Jul 2026 18:08:47 +0200 Subject: [PATCH 1/2] fix: don't re-invoke exhausted source during keep-alive silence SourcesQueueOutput::next() kept calling current.next() on the just-exhausted source every time the keep-alive silence budget ran out, instead of moving past it. Sources with side effects in next() (like EmptyCallback) fired repeatedly for as long as the queue stayed empty, instead of exactly once. Track exhaustion explicitly and stop calling next() on current once observed, while still checking the queue every frame so a newly appended source starts with no added latency. --- src/queue.rs | 89 +++++++++++++++++++++++++++++++++++++--------------- 1 file changed, 63 insertions(+), 26 deletions(-) diff --git a/src/queue.rs b/src/queue.rs index 840e4f50c..94c91c8f4 100644 --- a/src/queue.rs +++ b/src/queue.rs @@ -33,6 +33,7 @@ pub fn queue(keep_alive_if_empty: bool) -> (Arc, SourcesQueue let output = SourcesQueueOutput { current: Box::new(Empty::new()) as Box<_>, + current_exhausted: false, signal_after_end: None, input: input.clone(), samples_consumed_in_span: 0, @@ -114,6 +115,9 @@ pub struct SourcesQueueOutput { // The current iterator that produces samples. current: Box, + // Whether `current` has already reported exhaustion. + current_exhausted: bool, + // Signal this sender before picking from `next`. signal_after_end: Option>, @@ -123,8 +127,7 @@ pub struct SourcesQueueOutput { // Track samples consumed in the current span to detect mid-span endings. samples_consumed_in_span: usize, - // When a source ends mid-frame, this counts how many silence samples to inject - // to complete the frame before transitioning to the next source. + // Number of silence samples left to emit before consulting `current`/the queue again. silence_samples_remaining: usize, } @@ -216,33 +219,41 @@ impl Iterator for SourcesQueueOutput { } // Basic situation that will happen most of the time. - if let Some(sample) = self.current.next() { - return Some(sample); + if !self.current_exhausted { + if let Some(sample) = self.current.next() { + let channels = self.current.channels().get() as usize; + self.samples_consumed_in_span = (self.samples_consumed_in_span + 1) % channels; + return Some(sample); + } + self.current_exhausted = true; + + // Source ended - check if we ended mid-frame and need padding. + if self.samples_consumed_in_span > 0 { + let channels = self.current.channels().get() as usize; + // We're mid-frame - need to pad with silence to complete it. + self.silence_samples_remaining = channels - self.samples_consumed_in_span; + // Reset counter now since we're transitioning to a new span. + self.samples_consumed_in_span = 0; + // Continue loop - next iteration will inject silence. + continue; + } } - // Source ended - check if we ended mid-frame and need padding. - let channels = self.current.channels().get() as usize; - let incomplete_frame_samples = self.samples_consumed_in_span % channels; - if incomplete_frame_samples > 0 { - // We're mid-frame - need to pad with silence to complete it. - self.silence_samples_remaining = channels - incomplete_frame_samples; - // Reset counter now since we're transitioning to a new span. - self.samples_consumed_in_span = 0; - // Continue loop - next iteration will inject silence. + // `current` is exhausted: move to the next sound without calling `.next()` on it + // again. In order to avoid inlining this expensive operation, the code is in + // another function. + if self.go_next().is_ok() { + self.current_exhausted = false; continue; } - // Reset counter and move to next sound. - // In order to avoid inlining this expensive operation, the code is in another function. - self.samples_consumed_in_span = 0; - if self.go_next().is_err() { - if self.input.keep_alive_if_empty() { - self.silence_samples_remaining = self.current.channels().get() as usize; - continue; - } else { - return None; - } + if self.input.keep_alive_if_empty() { + // Emit one frame of silence, then re-check the queue on the very next call. + self.silence_samples_remaining = self.current.channels().get() as usize; + continue; } + + return None; } } @@ -253,8 +264,8 @@ impl Iterator for SourcesQueueOutput { } impl SourcesQueueOutput { - // Called when `current` is empty, and we must jump to the next element. - // Returns `Ok` if there is another sound should continue playing, or `Err` when there is not. + // Called when `current` is exhausted, and we must jump to the next element. + // Returns `Ok` if there is another sound queued, or `Err` when there is not. // // This method is separate so that it is not inlined. fn go_next(&mut self) -> Result<(), ()> { @@ -374,11 +385,37 @@ mod tests { } } + #[test] + fn exhausted_source_called_once_during_keep_alive() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + use crate::source::EmptyCallback; + + let (tx, mut rx) = queue::queue(true); + tx.append(SamplesBuffer::new(nz!(1), nz!(48000), vec![10.0, -10.0])); + + let calls = Arc::new(AtomicUsize::new(0)); + let calls_clone = calls.clone(); + tx.append(EmptyCallback::new(Box::new(move || { + calls_clone.fetch_add(1, Ordering::Relaxed); + }))); + + assert_eq!(rx.next(), Some(10.0)); + assert_eq!(rx.next(), Some(-10.0)); + + for _ in 0..10000 { + assert_eq!(rx.next(), Some(0.0)); + } + + assert_eq!(calls.load(Ordering::Relaxed), 1); + } + #[test] fn no_delay_when_added() { let (tx, mut rx) = queue::queue(true); - for _ in 0..500 { + for _ in 0..10000 { assert_eq!(rx.next(), Some(0.0)); } From 03e1f10dff5dabe4015ebe3448c7869c2e3c5a39 Mon Sep 17 00:00:00 2001 From: Roderick van Domburg Date: Sat, 1 Aug 2026 16:37:24 +0200 Subject: [PATCH 2/2] feat(queue): keep-alive with arbitrary sources and Empty format support - Introduce a format-aware Empty and allow queue keep-alive with an optional source (Option>) instead of a bool. - Add Mixer helpers (channels, sample_rate, silence) and Player::new_with_keep_alive/connect_new to use a silence keep-alive. - Cache peeked metadata in SourcesQueueOutput to correctly report channel/sample_rate while the queue is briefly empty. --- src/mixer.rs | 19 +++++- src/player.rs | 15 ++++- src/queue.rs | 137 +++++++++++++++++++++++++--------------- src/source/empty.rs | 33 ++++++++-- tests/channel_volume.rs | 2 +- 5 files changed, 146 insertions(+), 60 deletions(-) diff --git a/src/mixer.rs b/src/mixer.rs index d43675b4e..8001ee0cf 100644 --- a/src/mixer.rs +++ b/src/mixer.rs @@ -1,7 +1,7 @@ //! Mixer that plays multiple sounds at the same time. use crate::common::{ChannelCount, SampleRate}; -use crate::source::{SeekError, Source, UniformSourceIterator}; +use crate::source::{Empty, SeekError, Source, UniformSourceIterator}; use crate::Sample; use std::sync::Arc; use std::time::Duration; @@ -67,6 +67,23 @@ impl Mixer { // Ignore send errors (channel dropped means MixerSource was dropped) let _ = self.0.pending_tx.send(Box::new(uniform_source)); } + + /// The channel count sources are converted to when added to this mixer. + pub fn channels(&self) -> ChannelCount { + self.0.channels + } + + /// The sample rate sources are converted to when added to this mixer. + pub fn sample_rate(&self) -> SampleRate { + self.0.sample_rate + } + + /// A source in this mixer's format that never produces a sample. Useful as a + /// [`queue`](crate::queue::queue) keep-alive source, so idling sounds like this one won't + /// need source conversion set up until real content is appended. + pub fn silence(&self) -> Empty { + Empty::new_with_format(self.0.channels, self.0.sample_rate) + } } /// The output of the mixer. Implements `Source`. diff --git a/src/player.rs b/src/player.rs index c3547be00..016b90071 100644 --- a/src/player.rs +++ b/src/player.rs @@ -9,7 +9,7 @@ use dasp_sample::FromSample; use std::sync::mpsc::{Receiver, Sender}; use crate::mixer::Mixer; -use crate::source::SeekError; +use crate::source::{Empty, SeekError}; use crate::Float; use crate::{queue, source::Done, Source}; @@ -71,7 +71,7 @@ impl Player { /// Builds a new `Player`, beginning playback on a stream. #[inline] pub fn connect_new(mixer: &Mixer) -> Player { - let (sink, source) = Player::new(); + let (sink, source) = Player::new_with_keep_alive(mixer.silence()); mixer.add(source); sink } @@ -79,7 +79,16 @@ impl Player { /// Builds a new `Player`. #[inline] pub fn new() -> (Player, queue::SourcesQueueOutput) { - let (queue_tx, queue_rx) = queue::queue(true); + Player::new_with_keep_alive(Empty::new()) + } + + /// Builds a new `Player` that plays `keep_alive_with` (looping to silence in its format + /// once exhausted) instead of ending when it becomes empty. See [`queue::queue`]. + pub fn new_with_keep_alive(keep_alive_with: S) -> (Player, queue::SourcesQueueOutput) + where + S: Source + Send + 'static, + { + let (queue_tx, queue_rx) = queue::queue(Some(Box::new(keep_alive_with))); let sink = Player { queue_tx, diff --git a/src/queue.rs b/src/queue.rs index 94c91c8f4..e259e676e 100644 --- a/src/queue.rs +++ b/src/queue.rs @@ -1,5 +1,6 @@ //! Queue that plays sounds one after the other. +use std::cell::Cell; use std::collections::VecDeque; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; @@ -19,25 +20,37 @@ use std::sync::mpsc::{channel, Receiver, Sender}; /// The input can be used to add sounds to the end of the queue, while the output implements /// `Source` and plays the sounds. /// -/// The parameter indicates how the queue should behave if the queue becomes empty: +/// `keep_alive_with` controls what happens when the queue becomes empty: /// -/// - If you pass `true`, then the queue is infinite and will play a silence instead until you add -/// a new sound. -/// - If you pass `false`, then the queue will report that it has finished playing. -/// -pub fn queue(keep_alive_if_empty: bool) -> (Arc, SourcesQueueOutput) { +/// - `None`: the queue reports that it has finished playing. +/// - `Some(source)`: `source` plays first (if it produces any samples), then the queue stays +/// alive and plays silence in `source`'s format until a new sound is appended. Pass e.g. an +/// `Empty` in the target format to avoid setting up source conversion for a target that turns +/// out to match once a real source is appended - `Mixer::silence` builds one for a mixer's +/// format. +pub fn queue( + keep_alive_with: Option>, +) -> (Arc, SourcesQueueOutput) { let input = Arc::new(SourcesQueueInput { next_sounds: Mutex::new(VecDeque::new()), - keep_alive_if_empty: AtomicBool::new(keep_alive_if_empty), + keep_alive_if_empty: AtomicBool::new(keep_alive_with.is_some()), + metadata_dirty: AtomicBool::new(false), + }); + + let current = keep_alive_with.unwrap_or_else(|| Box::new(Empty::new())); + let peeked_metadata = Cell::new(PeekedMetadata { + channels: current.channels(), + sample_rate: current.sample_rate(), }); let output = SourcesQueueOutput { - current: Box::new(Empty::new()) as Box<_>, + current, current_exhausted: false, signal_after_end: None, input: input.clone(), samples_consumed_in_span: 0, silence_samples_remaining: 0, + peeked_metadata, }; (input, output) @@ -54,6 +67,9 @@ pub struct SourcesQueueInput { // See constructor. keep_alive_if_empty: AtomicBool, + + // Set on append/clear/pop. See `PeekedMetadata`. + metadata_dirty: AtomicBool, } impl SourcesQueueInput { @@ -67,6 +83,7 @@ impl SourcesQueueInput { .lock() .unwrap() .push_back((Box::new(source) as Box<_>, None)); + self.metadata_dirty.store(true, Ordering::Release); } /// Adds a new source to the end of the queue. @@ -85,6 +102,7 @@ impl SourcesQueueInput { .lock() .unwrap() .push_back((Box::new(source) as Box<_>, Some(tx))); + self.metadata_dirty.store(true, Ordering::Release); rx } @@ -106,10 +124,21 @@ impl SourcesQueueInput { let mut sounds = self.next_sounds.lock().unwrap(); let len = sounds.len(); sounds.clear(); + drop(sounds); + if len > 0 { + self.metadata_dirty.store(true, Ordering::Release); + } len } } +// Cached last peek at `next_sounds`. See `SourcesQueueInput::metadata_dirty`. +#[derive(Clone, Copy)] +struct PeekedMetadata { + channels: ChannelCount, + sample_rate: SampleRate, +} + /// The output of the queue. Implements `Source`. pub struct SourcesQueueOutput { // The current iterator that produces samples. @@ -129,6 +158,36 @@ pub struct SourcesQueueOutput { // Number of silence samples left to emit before consulting `current`/the queue again. silence_samples_remaining: usize, + + peeked_metadata: Cell, +} + +impl SourcesQueueOutput { + // Metadata to report while `current` is exhausted: the next queued non-exhausted + // source's format, or `current`'s own if there isn't one. + fn exhausted_metadata(&self) -> PeekedMetadata { + if !self.input.metadata_dirty.swap(false, Ordering::Acquire) { + return self.peeked_metadata.get(); + } + + let peeked = self + .input + .next_sounds + .lock() + .unwrap() + .iter() + .find(|(s, _)| !s.is_exhausted()) + .map(|(s, _)| (s.channels(), s.sample_rate())); + let (channels, sample_rate) = + peeked.unwrap_or((self.current.channels(), self.current.sample_rate())); + + let fresh = PeekedMetadata { + channels, + sample_rate, + }; + self.peeked_metadata.set(fresh); + fresh + } } impl Source for SourcesQueueOutput { @@ -145,44 +204,16 @@ impl Source for SourcesQueueOutput { #[inline] fn channels(&self) -> ChannelCount { if self.current.is_exhausted() && self.silence_samples_remaining == 0 { - // Skip exhausted sources at the head of the queue (e.g. an empty chain) and - // return the first non-exhausted source's metadata. This is critical: - // UniformSourceIterator queries metadata before pulling any samples, so we - // must report the upcoming source's format, not a preceding exhausted stub. - // - // If the queue is genuinely empty there is nothing to peek at. The stale value - // is returned below. This is corrected at the first span boundary after the - // new source begins playing. - if let Some((next, _)) = self - .input - .next_sounds - .lock() - .unwrap() - .iter() - .find(|(s, _)| !s.is_exhausted()) - { - return next.channels(); - } + return self.exhausted_metadata().channels; } - self.current.channels() } #[inline] fn sample_rate(&self) -> SampleRate { if self.current.is_exhausted() && self.silence_samples_remaining == 0 { - if let Some((next, _)) = self - .input - .next_sounds - .lock() - .unwrap() - .iter() - .find(|(s, _)| !s.is_exhausted()) - { - return next.sample_rate(); - } + return self.exhausted_metadata().sample_rate; } - self.current.sample_rate() } @@ -277,6 +308,7 @@ impl SourcesQueueOutput { let mut next = self.input.next_sounds.lock().unwrap(); next.pop_front().ok_or(())? }; + self.input.metadata_dirty.store(true, Ordering::Release); self.current = next; self.signal_after_end = signal_after_end; @@ -288,15 +320,19 @@ impl SourcesQueueOutput { mod tests { use crate::buffer::SamplesBuffer; use crate::math::nz; - use crate::source::{chain, SeekError, Source}; + use crate::source::{chain, Empty, SeekError, Source}; use crate::{queue, ChannelCount, Sample, SampleRate}; use std::time::Duration; + fn default_keep_alive() -> Option> { + Some(Box::new(Empty::new())) + } + #[test] #[ignore = "known limitation: metadata gap when queue is briefly empty after exhaustion"] fn metadata_gap_when_queue_briefly_empty() { let new_rate = nz!(48000); - let (tx, mut rx) = queue::queue(false); + let (tx, mut rx) = queue::queue(None); tx.append(SamplesBuffer::new(nz!(1), nz!(44100), vec![1.0])); assert_eq!(rx.next(), Some(1.0)); @@ -319,7 +355,7 @@ mod tests { let empty_chain_dummy_rate = chain(std::iter::empty::()).sample_rate(); assert_ne!(empty_chain_dummy_rate, source_rate); - let (tx, mut rx) = queue::queue(false); + let (tx, mut rx) = queue::queue(None); tx.append(chain(std::iter::empty::())); tx.append(SamplesBuffer::new(nz!(1), source_rate, vec![1.0, 2.0])); @@ -332,7 +368,7 @@ mod tests { #[test] fn basic() { - let (tx, mut rx) = queue::queue(false); + let (tx, mut rx) = queue::queue(None); tx.append(SamplesBuffer::new( nz!(1), @@ -362,13 +398,13 @@ mod tests { #[test] fn immediate_end() { - let (_, mut rx) = queue::queue(false); + let (_, mut rx) = queue::queue(None); assert_eq!(rx.next(), None); } #[test] fn keep_alive() { - let (tx, mut rx) = queue::queue(true); + let (tx, mut rx) = queue::queue(default_keep_alive()); tx.append(SamplesBuffer::new( nz!(1), nz!(48000), @@ -392,7 +428,7 @@ mod tests { use crate::source::EmptyCallback; - let (tx, mut rx) = queue::queue(true); + let (tx, mut rx) = queue::queue(default_keep_alive()); tx.append(SamplesBuffer::new(nz!(1), nz!(48000), vec![10.0, -10.0])); let calls = Arc::new(AtomicUsize::new(0)); @@ -413,7 +449,7 @@ mod tests { #[test] fn no_delay_when_added() { - let (tx, mut rx) = queue::queue(true); + let (tx, mut rx) = queue::queue(default_keep_alive()); for _ in 0..10000 { assert_eq!(rx.next(), Some(0.0)); @@ -432,7 +468,7 @@ mod tests { #[test] fn sample_rate_correct_after_stopped_source() { - let (tx, mut rx) = queue::queue(true); + let (tx, mut rx) = queue::queue(default_keep_alive()); let mut stopped_source = SamplesBuffer::new(nz!(1), nz!(48000), vec![0.0; 100]).stoppable(); stopped_source.stop(); @@ -451,7 +487,7 @@ mod tests { #[test] fn sample_rate_correct_after_skipped_source() { - let (tx, mut rx) = queue::queue(true); + let (tx, mut rx) = queue::queue(default_keep_alive()); let mut skipped_source = SamplesBuffer::new(nz!(1), nz!(48000), vec![0.0; 100]).skippable(); crate::source::Skippable::skip(&mut skipped_source); @@ -471,7 +507,7 @@ mod tests { #[test] fn channel_correct_on_first_append() { let (mixer_tx, mut mixer_rx) = crate::mixer::mixer(nz!(2), nz!(48000)); - let (tx, rx) = queue::queue(true); + let (tx, rx) = queue::queue(default_keep_alive()); assert_eq!(rx.channels(), nz!(1), "initial channels should be 1"); mixer_tx.add(rx); @@ -491,7 +527,8 @@ mod tests { #[test] fn append_updates_metadata() { for keep_alive in [false, true] { - let (tx, rx) = queue::queue(keep_alive); + let keep_alive_with = keep_alive.then(default_keep_alive).flatten(); + let (tx, rx) = queue::queue(keep_alive_with); assert_eq!( rx.channels(), nz!(1), diff --git a/src/source/empty.rs b/src/source/empty.rs index 308c02891..6f4133e80 100644 --- a/src/source/empty.rs +++ b/src/source/empty.rs @@ -6,15 +6,38 @@ use crate::math::nz; use crate::{Sample, Source}; /// An empty source. -#[derive(Debug, Default, Copy, Clone)] -pub struct Empty; +#[derive(Debug, Copy, Clone)] +pub struct Empty { + channels: ChannelCount, + sample_rate: SampleRate, +} + +impl Default for Empty { + #[inline] + fn default() -> Self { + Self { + channels: nz!(1), + sample_rate: crate::DEFAULT_SAMPLE_RATE, + } + } +} impl Empty { /// An empty source that immediately ends without ever returning a sample to /// play #[inline] pub fn new() -> Self { - Self + Self::default() + } + + /// Like [`Empty::new`], but reports `channels`/`sample_rate` instead of a default format. + /// Useful as a placeholder that won't need format conversion once given real content. + #[inline] + pub fn new_with_format(channels: ChannelCount, sample_rate: SampleRate) -> Self { + Self { + channels, + sample_rate, + } } } @@ -42,12 +65,12 @@ impl Source for Empty { #[inline] fn channels(&self) -> ChannelCount { - nz!(1) + self.channels } #[inline] fn sample_rate(&self) -> SampleRate { - crate::DEFAULT_SAMPLE_RATE + self.sample_rate } #[inline] diff --git a/tests/channel_volume.rs b/tests/channel_volume.rs index cfeb02948..7288bea34 100644 --- a/tests/channel_volume.rs +++ b/tests/channel_volume.rs @@ -24,7 +24,7 @@ fn channel_volume_without_queue() { #[test] fn channel_volume_with_queue() { let channel_volume = create_6_channel_source(); - let (controls, queue) = queue::queue(false); + let (controls, queue) = queue::queue(None); controls.append(channel_volume); assert_output_only_on_first_two_channels(queue, 6); }