diff --git a/src/mixer.rs b/src/mixer.rs index d43675b4..8001ee0c 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 c3547be0..016b9007 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 840e4f50..e259e676 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,24 +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) @@ -53,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 { @@ -66,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. @@ -84,6 +102,7 @@ impl SourcesQueueInput { .lock() .unwrap() .push_back((Box::new(source) as Box<_>, Some(tx))); + self.metadata_dirty.store(true, Ordering::Release); rx } @@ -105,15 +124,29 @@ 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. current: Box, + // Whether `current` has already reported exhaustion. + current_exhausted: bool, + // Signal this sender before picking from `next`. signal_after_end: Option>, @@ -123,9 +156,38 @@ 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, + + 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 { @@ -142,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() } @@ -216,33 +250,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 +295,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<(), ()> { @@ -266,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; @@ -277,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)); @@ -308,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])); @@ -321,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), @@ -351,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), @@ -374,11 +421,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(default_keep_alive()); + 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); + let (tx, mut rx) = queue::queue(default_keep_alive()); - for _ in 0..500 { + for _ in 0..10000 { assert_eq!(rx.next(), Some(0.0)); } @@ -395,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(); @@ -414,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); @@ -434,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); @@ -454,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 308c0289..6f4133e8 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 cfeb0294..7288bea3 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); }