shuyi_sound/mix/
stream.rs

1//! Streams: what a worker hands the mixer, and what a voice reads it through.
2//!
3//! A worker owns a decoder and a [`Feeder`]; the mixer owns the [`Feed`], the
4//! other end of a fixed ring of decoded frames. What the mixer holds of a stream
5//! is that ring and a small buffer, however long the stream is.
6//!
7//! The contract between them:
8//!
9//! - The worker keeps the ring full ahead of the mix ([`Feeder::fill_from`]), and
10//!   [`Feeder::close`]s it after the last frame.
11//! - A read of an empty ring that isn't closed is [`Error::Underrun`]. The voice
12//!   stays where it is and sounds nothing until frames come: an underrun delays a
13//!   stream, and loses and repeats nothing. [`Mixer::underrun_frames`] counts it.
14//! - The worker supplies a loop's frames too, as [`LoopedRegion`] does, so that the
15//!   way back to the start of a region is worked out and in the ring before the
16//!   mix reaches the end of it.
17//!
18//! [`Mixer::underrun_frames`]: super::Mixer::underrun_frames
19
20use alloc::boxed::Box;
21use core::sync::atomic::{AtomicBool, Ordering};
22
23use super::queue::Ring;
24use super::{Region, Shared};
25use crate::error::Error;
26use crate::pcm::{Channels, PcmFormat};
27use crate::source::Source;
28
29/// How many frames a voice holds of its stream.
30const BUFFER_FRAMES: usize = 128;
31/// How many samples a worker moves or discards at a time.
32const CHUNK_SAMPLES: usize = 512;
33
34/// A frame as one word of a ring: a mono frame has the sample twice.
35fn frame_word(left: i16, right: i16) -> u32 {
36    let [a, b] = left.to_le_bytes();
37    let [c, d] = right.to_le_bytes();
38    u32::from_le_bytes([a, b, c, d])
39}
40
41fn word_frame(word: u32) -> [i16; 2] {
42    let [a, b, c, d] = word.to_le_bytes();
43    [i16::from_le_bytes([a, b]), i16::from_le_bytes([c, d])]
44}
45
46/// A ring of `capacity` frames of `format` (rounded up to a power of two): the end
47/// a worker writes, and the end the mixer reads.
48pub fn feed(format: PcmFormat, capacity: usize) -> (Feeder, Feed) {
49    let ring = Shared::new(Ring::new(capacity, 1));
50    let closed = Shared::new(AtomicBool::new(false));
51    (
52        Feeder {
53            ring: Shared::clone(&ring),
54            closed: Shared::clone(&closed),
55            format,
56        },
57        Feed {
58            ring,
59            closed,
60            format,
61        },
62    )
63}
64
65/// The end of a [`feed`] a worker writes.
66#[derive(Debug)]
67pub struct Feeder {
68    ring: Shared<Ring>,
69    closed: Shared<AtomicBool>,
70    format: PcmFormat,
71}
72
73impl Feeder {
74    /// How many frames the ring has room for now.
75    pub fn room(&self) -> usize {
76        self.ring.room()
77    }
78
79    /// Writes as many whole frames of `samples` as there is room for, interleaved,
80    /// and says how many that was.
81    pub fn write(&mut self, samples: &[i16]) -> usize {
82        let channels = usize::from(self.format.channels.count());
83        let frames = (samples.len() / channels).min(self.ring.room());
84        self.ring.write_words(frames, |frame| {
85            let left = samples[frame * channels];
86            frame_word(left, samples[frame * channels + channels - 1])
87        });
88        frames
89    }
90
91    /// Moves frames from `source` until the ring has no room or `source` ends,
92    /// when it closes the ring, and says how many frames moved. This is the
93    /// decoder's work, for a worker: it is never done in a mix.
94    pub fn fill_from(&mut self, source: &mut impl Source) -> Result<usize, Error> {
95        if source.format().channels != self.format.channels {
96            return Err(Error::Channels(u32::from(source.format().channels.count())));
97        }
98        let channels = usize::from(self.format.channels.count());
99        let mut chunk = [0; CHUNK_SAMPLES];
100        let mut moved = 0;
101        loop {
102            let frames = self.ring.room().min(CHUNK_SAMPLES / channels);
103            if frames == 0 {
104                return Ok(moved);
105            }
106            let read = source.read(&mut chunk[..frames * channels])?;
107            if read == 0 {
108                self.close();
109                return Ok(moved);
110            }
111            moved += self.write(&chunk[..read * channels]);
112        }
113    }
114
115    /// There are no more frames than those written.
116    pub fn close(&mut self) {
117        self.closed.store(true, Ordering::Release);
118    }
119
120    /// Whether [`close`](Self::close) has been called, as
121    /// [`fill_from`](Self::fill_from) does at its source's end: nothing more to do.
122    pub fn is_closed(&self) -> bool {
123        self.closed.load(Ordering::Acquire)
124    }
125}
126
127/// The end of a [`feed`] the mixer reads, a [`Source`] that never waits.
128#[derive(Debug)]
129pub struct Feed {
130    ring: Shared<Ring>,
131    closed: Shared<AtomicBool>,
132    format: PcmFormat,
133}
134
135impl Source for Feed {
136    fn format(&self) -> PcmFormat {
137        self.format
138    }
139
140    fn total_frames(&self) -> Option<u64> {
141        None
142    }
143
144    fn read(&mut self, out: &mut [i16]) -> Result<usize, Error> {
145        let channels = usize::from(self.format.channels.count());
146        let room = crate::source::room(self.format, out)?;
147        // Before the ring: frames written before the close are there to see.
148        let closed = self.closed.load(Ordering::Acquire);
149        let frames = self.ring.read_words(room, |frame, word| {
150            let [left, right] = word_frame(word);
151            out[frame * channels] = left;
152            out[frame * channels + channels - 1] = right;
153        });
154        if frames == 0 && !closed {
155            return Err(Error::Underrun);
156        }
157        Ok(frames)
158    }
159
160    /// Nothing to do: the worker has put the frames a loop goes back to in the
161    /// ring already, right after the last, so the next read gives them.
162    fn rewind_to(&mut self, _frame: u64) -> Result<(), Error> {
163        Ok(())
164    }
165}
166
167/// A source that plays up to the end of a region, then from its start, for ever,
168/// for a worker to [`fill_from`](Feeder::fill_from). `restart` gives the source
169/// from its first frame again, however that is done (a seek, or a new decoder over
170/// the same bytes), and the frames before the region are read and dropped: what
171/// comes out is sample-exact at the join whatever the decoder's own pages are.
172#[derive(Debug)]
173pub struct LoopedRegion<S, F> {
174    source: S,
175    restart: F,
176    region: Region,
177    /// The next frame `source` gives.
178    at: u64,
179}
180
181impl<S: Source, F: FnMut() -> Result<S, Error>> LoopedRegion<S, F> {
182    /// `source`, from its first frame, looping `region`.
183    pub const fn new(source: S, region: Region, restart: F) -> Self {
184        Self {
185            source,
186            restart,
187            region,
188            at: 0,
189        }
190    }
191
192    /// Goes back to the start of the region.
193    fn back(&mut self) -> Result<(), Error> {
194        self.source = (self.restart)()?;
195        self.at = 0;
196        let channels = usize::from(self.source.format().channels.count());
197        let mut chunk = [0; CHUNK_SAMPLES];
198        while self.at < self.region.start() {
199            let left = self.region.start() - self.at;
200            let most = (CHUNK_SAMPLES / channels) as u64;
201            // At most a chunk's, which fits.
202            let frames = usize::try_from(left.min(most)).unwrap_or(0);
203            let read = self.source.read(&mut chunk[..frames * channels])?;
204            if read == 0 {
205                return Ok(());
206            }
207            self.at += read as u64;
208        }
209        Ok(())
210    }
211}
212
213impl<S: Source, F: FnMut() -> Result<S, Error>> Source for LoopedRegion<S, F> {
214    fn format(&self) -> PcmFormat {
215        self.source.format()
216    }
217
218    fn total_frames(&self) -> Option<u64> {
219        None
220    }
221
222    fn read(&mut self, out: &mut [i16]) -> Result<usize, Error> {
223        let channels = usize::from(self.format().channels.count());
224        let room = crate::source::room(self.format(), out)?;
225        let mut went_back = false;
226        loop {
227            if self.at >= self.region.end() {
228                self.back()?;
229                went_back = true;
230            }
231            let frames = usize::try_from(self.region.end().saturating_sub(self.at))
232                .map_or(room, |left| left.min(room));
233            let read = if frames == 0 {
234                0
235            } else {
236                self.source.read(&mut out[..frames * channels])?
237            };
238            if read > 0 {
239                self.at += read as u64;
240                return Ok(read);
241            }
242            // The file ends before the region does, which is where it comes round;
243            // a second time with nothing read means there is nothing to loop.
244            if went_back {
245                return Ok(0);
246            }
247            self.back()?;
248            went_back = true;
249        }
250    }
251}
252
253/// What a voice gets when it asks for its source's next frame.
254#[derive(Debug, Clone, Copy, PartialEq, Eq)]
255pub(super) enum Fetch {
256    /// The frame, left then right.
257    Frame([i16; 2]),
258    /// It has ended.
259    End,
260    /// It has no frame yet, and will.
261    Underrun,
262}
263
264/// A stream the mixer holds: its source and what a voice hasn't played of it.
265pub(super) struct Slot {
266    pub(super) source: Option<Box<dyn Source + Send>>,
267    pub(super) generation: u32,
268    /// A voice is playing it or has.
269    pub(super) used: bool,
270    /// A voice played it and is done with it: it waits to be taken away.
271    pub(super) retired: bool,
272    buffer: Box<[i16]>,
273    /// The next sample of `buffer`.
274    start: usize,
275    /// The end of what `buffer` holds.
276    end: usize,
277    /// The next frame the source gives.
278    read_at: u64,
279    /// How many samples a frame of the source has.
280    channels: usize,
281}
282
283impl Slot {
284    pub(super) fn new() -> Self {
285        Self {
286            source: None,
287            generation: 0,
288            used: false,
289            retired: false,
290            buffer: alloc::vec![0; BUFFER_FRAMES * 2].into_boxed_slice(),
291            start: 0,
292            end: 0,
293            read_at: 0,
294            channels: 2,
295        }
296    }
297
298    /// Holds `source`, ready to play from its start.
299    pub(super) fn hold(&mut self, source: Box<dyn Source + Send>) {
300        self.channels = match source.format().channels {
301            Channels::Mono => 1,
302            Channels::Stereo => 2,
303        };
304        self.source = Some(source);
305        self.generation = self.generation.wrapping_add(1);
306        self.used = false;
307        self.retired = false;
308        self.start = 0;
309        self.end = 0;
310        self.read_at = 0;
311    }
312
313    /// The source's next frame, with a loop's way back. No decoding: a source a
314    /// mixer reads is one that has its frames ready.
315    pub(super) fn next(&mut self, looped: bool, region: Option<Region>) -> Fetch {
316        if self.start >= self.end
317            && let Some(stop) = self.refill(looped, region)
318        {
319            return stop;
320        }
321        let frame = [
322            self.buffer[self.start],
323            self.buffer[self.start + self.channels - 1],
324        ];
325        self.start += self.channels;
326        Fetch::Frame(frame)
327    }
328
329    /// Fills the buffer, or says why it can't be.
330    fn refill(&mut self, looped: bool, region: Option<Region>) -> Option<Fetch> {
331        let source = self.source.as_mut()?;
332        let mut went_back = false;
333        loop {
334            if let Some(region) = region.filter(|region| self.read_at >= region.end()) {
335                match source.rewind_to(region.start()) {
336                    Ok(()) => self.read_at = region.start(),
337                    Err(Error::Underrun) => return Some(Fetch::Underrun),
338                    Err(_) => return Some(Fetch::End),
339                }
340                went_back = true;
341            }
342            let frames = region.map_or(BUFFER_FRAMES, |region| {
343                usize::try_from(region.end().saturating_sub(self.read_at))
344                    .map_or(BUFFER_FRAMES, |left| left.min(BUFFER_FRAMES))
345            });
346            match source.read(&mut self.buffer[..frames * self.channels]) {
347                Ok(0) => {
348                    // A source that ends before its region's end comes round there
349                    // too, and one with nothing after the way back has ended.
350                    if went_back || !(looped || region.is_some()) {
351                        return Some(Fetch::End);
352                    }
353                    let to = region.map_or(0, Region::start);
354                    match source.rewind_to(to) {
355                        Ok(()) => self.read_at = to,
356                        Err(Error::Underrun) => return Some(Fetch::Underrun),
357                        Err(_) => return Some(Fetch::End),
358                    }
359                    went_back = true;
360                }
361                Ok(read) => {
362                    self.read_at += read as u64;
363                    self.start = 0;
364                    self.end = read * self.channels;
365                    return None;
366                }
367                Err(Error::Underrun) => return Some(Fetch::Underrun),
368                Err(_) => return Some(Fetch::End),
369            }
370        }
371    }
372}
shuyi_sound/mix/stream.rs

shuyi_sound/mix/
stream.rs
