shuyi_sound/mix/
queue.rs

1//! Commands from another thread: a fixed ring of plain data, one writer and one
2//! reader, with no lock, no allocation and no `unsafe`.
3//!
4//! A ring's slots are `AtomicU32`s and a command is written as words, so nothing
5//! has to be shared but atomics. The writer stores a slot's words and then the head
6//! with `Release`, the reader loads the head with `Acquire` and then the words, and
7//! the same the other way for the tail. Only loads and stores: they are what the
8//! smallest targets have.
9//!
10//! [`Mixer::remote`] makes the two ends. The [`Remote`] on the game's side decides
11//! which voice a play goes to, so it can answer [`Full`] on the spot, and the
12//! [`Inbox`] on the audio thread applies what came and says which voices ended.
13//!
14//! A [`Pcm`](crate::Pcm) or a source can't go through as words: a backend gives them
15//! to the mixer ([`Mixer::add_sound`], [`Mixer::add_stream`]) before the mixer goes
16//! to the audio thread, or across by a route of its own.
17
18use alloc::boxed::Box;
19use alloc::vec::Vec;
20use core::sync::atomic::{AtomicU32, Ordering};
21
22use super::{Full, Handle, Mixer, Params, Region, Shared, SoundId, StreamId, VoiceId};
23
24/// The words of a command.
25const COMMAND_WORDS: usize = 12;
26/// The words of an ended voice.
27const EVENT_WORDS: usize = 2;
28
29/// A fixed ring of slots of `width` words, for one writer and one reader.
30#[derive(Debug)]
31pub(super) struct Ring {
32    words: Box<[AtomicU32]>,
33    width: usize,
34    /// A power of two less one: a slot's index from a counter that runs on.
35    mask: u32,
36    /// How many slots were ever written. Only the writer stores it.
37    head: AtomicU32,
38    /// How many slots were ever read. Only the reader stores it.
39    tail: AtomicU32,
40}
41
42impl Ring {
43    /// A ring of at least `slots` slots, up to 2^31.
44    pub(super) fn new(slots: usize, width: usize) -> Self {
45        let slots = u32::try_from(slots)
46            .unwrap_or(1 << 31)
47            .clamp(1, 1 << 31)
48            .next_power_of_two();
49        let count = usize::try_from(slots).unwrap_or(0) * width;
50        Self {
51            words: (0..count).map(|_| AtomicU32::new(0)).collect(),
52            width,
53            mask: slots - 1,
54            head: AtomicU32::new(0),
55            tail: AtomicU32::new(0),
56        }
57    }
58
59    /// How many slots it has.
60    fn capacity(&self) -> u32 {
61        self.mask.wrapping_add(1)
62    }
63
64    /// How many slots are free, as the writer sees it.
65    pub(super) fn room(&self) -> usize {
66        let used = self
67            .head
68            .load(Ordering::Relaxed)
69            .wrapping_sub(self.tail.load(Ordering::Acquire));
70        usize::try_from(self.capacity().wrapping_sub(used)).unwrap_or(0)
71    }
72
73    /// The index of word `word` of the slot at counter `at`.
74    fn index(&self, at: u32, word: usize) -> usize {
75        usize::try_from(at & self.mask).unwrap_or(0) * self.width + word
76    }
77
78    /// Writes one slot, if one is free.
79    fn push(&self, slot: &[u32]) -> bool {
80        if self.room() == 0 {
81            return false;
82        }
83        let head = self.head.load(Ordering::Relaxed);
84        for (word, value) in slot.iter().enumerate() {
85            self.words[self.index(head, word)].store(*value, Ordering::Relaxed);
86        }
87        self.head.store(head.wrapping_add(1), Ordering::Release);
88        true
89    }
90
91    /// Reads one slot, if one is there.
92    fn pop(&self, slot: &mut [u32]) -> bool {
93        let tail = self.tail.load(Ordering::Relaxed);
94        if self.head.load(Ordering::Acquire) == tail {
95            return false;
96        }
97        for (word, value) in slot.iter_mut().enumerate() {
98            *value = self.words[self.index(tail, word)].load(Ordering::Relaxed);
99        }
100        self.tail.store(tail.wrapping_add(1), Ordering::Release);
101        true
102    }
103
104    /// Writes `count` one-word slots, each from `word`; the caller checked the
105    /// room. They appear to the reader together.
106    pub(super) fn write_words(&self, count: usize, mut word: impl FnMut(usize) -> u32) {
107        let head = self.head.load(Ordering::Relaxed);
108        let mut at = head;
109        for slot in 0..count {
110            self.words[self.index(at, 0)].store(word(slot), Ordering::Relaxed);
111            at = at.wrapping_add(1);
112        }
113        self.head.store(at, Ordering::Release);
114    }
115
116    /// Reads up to `count` one-word slots, handing each to `word`, and says how
117    /// many.
118    pub(super) fn read_words(&self, count: usize, mut word: impl FnMut(usize, u32)) -> usize {
119        let tail = self.tail.load(Ordering::Relaxed);
120        let ready = self.head.load(Ordering::Acquire).wrapping_sub(tail);
121        let frames = usize::try_from(ready).map_or(count, |ready| ready.min(count));
122        let mut at = tail;
123        for slot in 0..frames {
124            word(slot, self.words[self.index(at, 0)].load(Ordering::Relaxed));
125            at = at.wrapping_add(1);
126        }
127        self.tail.store(at, Ordering::Release);
128        frames
129    }
130}
131
132/// What the mixer can be told, as plain data.
133#[derive(Debug, Clone, Copy, PartialEq, Eq)]
134pub(super) enum Command {
135    Play(VoiceId, SoundId, Params),
136    Stream(VoiceId, StreamId, Params),
137    SetVolume(VoiceId, u16),
138    FadeTo(VoiceId, u16, u64),
139    SetBalance(VoiceId, u16, u16),
140    Stop(VoiceId),
141    Pause(VoiceId),
142    Resume(VoiceId),
143}
144
145fn pair(low: u16, high: u16) -> u32 {
146    u32::from(low) + (u32::from(high) << 16)
147}
148
149fn low(word: u32) -> u16 {
150    u16::try_from(word & 0xFFFF).unwrap_or(0)
151}
152
153fn high(word: u32) -> u16 {
154    u16::try_from(word >> 16).unwrap_or(0)
155}
156
157fn split(value: u64) -> (u32, u32) {
158    (
159        u32::try_from(value & 0xFFFF_FFFF).unwrap_or(0),
160        u32::try_from(value >> 32).unwrap_or(0),
161    )
162}
163
164fn join(low: u32, high: u32) -> u64 {
165    u64::from(low) + (u64::from(high) << 32)
166}
167
168/// What a play's words say past the voice.
169fn play_words(words: &mut [u32; COMMAND_WORDS], index: Handle, params: Params) {
170    words[3] = index.index;
171    words[4] = index.generation;
172    words[5] = u32::from(params.volume) + (u32::from(params.looped) << 16);
173    words[6] = params.rate;
174    words[7] = pair(params.balance.0, params.balance.1);
175    if let Some(region) = params.region {
176        (words[8], words[9]) = split(region.start());
177        (words[10], words[11]) = split(region.end());
178    }
179}
180
181fn play_params(words: &[u32; COMMAND_WORDS]) -> Params {
182    // No region leaves its words at nothing, which is no region.
183    let region = Region::new(join(words[8], words[9]), join(words[10], words[11]));
184    let mut params = Params::new()
185        .with_volume(low(words[5]))
186        .with_rate(words[6])
187        .with_balance(low(words[7]), high(words[7]));
188    params.looped = words[5] & (1 << 16) != 0;
189    params.region = region;
190    params
191}
192
193impl Command {
194    /// As a ring's words.
195    fn encode(self) -> [u32; COMMAND_WORDS] {
196        let mut words = [0; COMMAND_WORDS];
197        let voice = match self {
198            Self::Play(voice, sound, params) => {
199                words[0] = 0;
200                play_words(&mut words, sound.0, params);
201                voice
202            }
203            Self::Stream(voice, stream, params) => {
204                words[0] = 1;
205                play_words(&mut words, stream.0, params);
206                voice
207            }
208            Self::SetVolume(voice, volume) => {
209                words[0] = 2;
210                words[3] = u32::from(volume);
211                voice
212            }
213            Self::FadeTo(voice, volume, millis) => {
214                words[0] = 3;
215                words[3] = u32::from(volume);
216                (words[4], words[5]) = split(millis);
217                voice
218            }
219            Self::SetBalance(voice, left, right) => {
220                words[0] = 4;
221                words[3] = pair(left, right);
222                voice
223            }
224            Self::Stop(voice) => {
225                words[0] = 5;
226                voice
227            }
228            Self::Pause(voice) => {
229                words[0] = 6;
230                voice
231            }
232            Self::Resume(voice) => {
233                words[0] = 7;
234                voice
235            }
236        };
237        words[1] = voice.0.index;
238        words[2] = voice.0.generation;
239        words
240    }
241
242    /// From a ring's words, which [`encode`](Self::encode) wrote.
243    fn decode(words: &[u32; COMMAND_WORDS]) -> Option<Self> {
244        let voice = VoiceId(Handle {
245            index: words[1],
246            generation: words[2],
247        });
248        let handle = Handle {
249            index: words[3],
250            generation: words[4],
251        };
252        Some(match words[0] {
253            0 => Self::Play(voice, SoundId(handle), play_params(words)),
254            1 => Self::Stream(voice, StreamId(handle), play_params(words)),
255            2 => Self::SetVolume(voice, low(words[3])),
256            3 => Self::FadeTo(voice, low(words[3]), join(words[4], words[5])),
257            4 => Self::SetBalance(voice, low(words[3]), high(words[3])),
258            5 => Self::Stop(voice),
259            6 => Self::Pause(voice),
260            7 => Self::Resume(voice),
261            _ => return None,
262        })
263    }
264}
265
266/// Makes the two ends for `voices` voices, with the bank and the stream slots as
267/// they are: each one's generation and whether it is taken.
268pub(super) fn ends(
269    voices: usize,
270    sounds: Vec<(u32, bool)>,
271    streams: Vec<(u32, bool)>,
272    commands: usize,
273) -> (Remote, Inbox) {
274    let commands = Shared::new(Ring::new(commands, COMMAND_WORDS));
275    let ended = Shared::new(Ring::new(voices.max(1), EVENT_WORDS));
276    (
277        Remote {
278            commands: Shared::clone(&commands),
279            ended: Shared::clone(&ended),
280            voices: alloc::vec![(0, false); voices],
281            sounds,
282            streams,
283        },
284        Inbox { commands, ended },
285    )
286}
287
288/// The next tenant of a free place of `table`, which is taken from now on.
289fn reserve(table: &mut [(u32, bool)]) -> Result<Handle, Full> {
290    let index = table.iter().position(|place| !place.1).ok_or(Full)?;
291    let generation = table[index].0.wrapping_add(1);
292    table[index] = (generation, true);
293    Ok(Handle {
294        index: u32::try_from(index).map_err(|_| Full)?,
295        generation,
296    })
297}
298
299/// Frees `handle`'s place of `table`, if it is still its tenant's.
300fn release(table: &mut [(u32, bool)], handle: Handle) {
301    if let Some(place) = usize::try_from(handle.index)
302        .ok()
303        .and_then(|index| table.get_mut(index))
304        .filter(|place| place.0 == handle.generation)
305    {
306        place.1 = false;
307    }
308}
309
310/// The mixer as a game's thread sees it, when the mixer is on the audio thread:
311/// the same operations, none of which waits.
312///
313/// An operation is refused with [`Full`] when the queue has no room for it. Sized
314/// for what a frame sends, that doesn't happen.
315#[derive(Debug)]
316pub struct Remote {
317    commands: Shared<Ring>,
318    ended: Shared<Ring>,
319    /// Each voice's generation, and whether it is busy.
320    voices: Vec<(u32, bool)>,
321    /// The same of the bank's places and the stream slots, which this end hands out.
322    sounds: Vec<(u32, bool)>,
323    streams: Vec<(u32, bool)>,
324}
325
326impl Remote {
327    /// Hears which voices ended.
328    fn hear(&mut self) {
329        let mut slot = [0; EVENT_WORDS];
330        while self.ended.pop(&mut slot) {
331            let voice = usize::try_from(slot[0]).ok();
332            if let Some(entry) = voice.and_then(|voice| self.voices.get_mut(voice))
333                && entry.0 == slot[1]
334            {
335                entry.1 = false;
336            }
337        }
338    }
339
340    fn send(&self, command: Command) -> Result<(), Full> {
341        if self.commands.push(&command.encode()) {
342            Ok(())
343        } else {
344            Err(Full)
345        }
346    }
347
348    /// A free voice's next id, the voice left free until the play has gone.
349    fn free_voice(&mut self) -> Result<VoiceId, Full> {
350        self.hear();
351        let index = self.voices.iter().position(|voice| !voice.1).ok_or(Full)?;
352        Ok(VoiceId(Handle {
353            index: u32::try_from(index).map_err(|_| Full)?,
354            generation: self.voices[index].0.wrapping_add(1),
355        }))
356    }
357
358    /// Marks `voice` busy.
359    fn taken(&mut self, voice: VoiceId) {
360        if let Some(entry) = usize::try_from(voice.0.index)
361            .ok()
362            .and_then(|index| self.voices.get_mut(index))
363        {
364            *entry = (voice.0.generation, true);
365        }
366    }
367
368    /// A place in the bank, named now, so that a play can follow at once: a backend
369    /// hands the sound over to [`Mixer::put_sound`] by a route of its own, ahead of
370    /// that play. [`Full`] says the bank has none free.
371    pub fn reserve_sound(&mut self) -> Result<SoundId, Full> {
372        reserve(&mut self.sounds).map(SoundId)
373    }
374
375    /// Frees a place [`reserve_sound`](Self::reserve_sound) named.
376    pub fn release_sound(&mut self, sound: SoundId) {
377        release(&mut self.sounds, sound.0);
378    }
379
380    /// A stream slot, named now, as [`reserve_sound`](Self::reserve_sound) names a
381    /// place in the bank, for [`Mixer::put_stream`].
382    pub fn reserve_stream(&mut self) -> Result<StreamId, Full> {
383        reserve(&mut self.streams).map(StreamId)
384    }
385
386    /// Frees a slot [`reserve_stream`](Self::reserve_stream) named.
387    pub fn release_stream(&mut self, stream: StreamId) {
388        release(&mut self.streams, stream.0);
389    }
390
391    /// [`Mixer::play`], from here.
392    pub fn play(&mut self, sound: SoundId, params: Params) -> Result<VoiceId, Full> {
393        let voice = self.free_voice()?;
394        self.send(Command::Play(voice, sound, params))?;
395        self.taken(voice);
396        Ok(voice)
397    }
398
399    /// [`Mixer::play_stream`], from here.
400    pub fn play_stream(&mut self, stream: StreamId, params: Params) -> Result<VoiceId, Full> {
401        let voice = self.free_voice()?;
402        self.send(Command::Stream(voice, stream, params))?;
403        self.taken(voice);
404        Ok(voice)
405    }
406
407    /// [`Mixer::set_volume`], from here.
408    pub fn set_volume(&self, voice: VoiceId, volume: u16) -> Result<(), Full> {
409        self.send(Command::SetVolume(voice, volume))
410    }
411
412    /// [`Mixer::fade_to`], from here.
413    pub fn fade_to(&self, voice: VoiceId, volume: u16, millis: u64) -> Result<(), Full> {
414        self.send(Command::FadeTo(voice, volume, millis))
415    }
416
417    /// [`Mixer::set_balance`], from here.
418    pub fn set_balance(&self, voice: VoiceId, left: u16, right: u16) -> Result<(), Full> {
419        self.send(Command::SetBalance(voice, left, right))
420    }
421
422    /// [`Mixer::stop`], from here.
423    pub fn stop(&self, voice: VoiceId) -> Result<(), Full> {
424        self.send(Command::Stop(voice))
425    }
426
427    /// [`Mixer::pause`], from here.
428    pub fn pause(&self, voice: VoiceId) -> Result<(), Full> {
429        self.send(Command::Pause(voice))
430    }
431
432    /// [`Mixer::resume`], from here.
433    pub fn resume(&self, voice: VoiceId) -> Result<(), Full> {
434        self.send(Command::Resume(voice))
435    }
436
437    /// Whether `voice` is playing, as far as the audio thread has said: it goes
438    /// on saying so until the mixer has reported the voice over.
439    pub fn is_playing(&mut self, voice: VoiceId) -> bool {
440        self.hear();
441        usize::try_from(voice.0.index)
442            .ok()
443            .and_then(|index| self.voices.get(index))
444            .is_some_and(|entry| *entry == (voice.0.generation, true))
445    }
446}
447
448/// What the audio thread calls before each [`Mixer::render`].
449#[derive(Debug)]
450pub struct Inbox {
451    commands: Shared<Ring>,
452    ended: Shared<Ring>,
453}
454
455impl Inbox {
456    /// Applies every command that has come, and says which voices ended. It takes
457    /// no lock and allocates nothing.
458    pub fn apply(&mut self, mixer: &mut Mixer) {
459        self.apply_with(mixer, |_| {});
460    }
461
462    /// [`apply`](Self::apply), calling `before` ahead of each command. A backend's
463    /// own hand-over goes there: whatever was handed over before a command was sent
464    /// has arrived once that command has, so a play always finds its sound.
465    pub fn apply_with(&mut self, mixer: &mut Mixer, mut before: impl FnMut(&mut Mixer)) {
466        let mut slot = [0; COMMAND_WORDS];
467        while self.commands.pop(&mut slot) {
468            before(mixer);
469            if let Some(command) = Command::decode(&slot) {
470                mixer.apply(command);
471            }
472        }
473        for (index, voice) in (0u32..).zip(mixer.voices.iter_mut()) {
474            // Room for every voice at once, so the next pass is only a precaution.
475            if voice.ended && self.ended.push(&[index, voice.generation]) {
476                voice.ended = false;
477            }
478        }
479    }
480}
481
482#[cfg(test)]
483mod tests {
484    use super::*;
485    use proptest::prelude::*;
486
487    #[test]
488    fn a_ring_is_a_power_of_two_and_refuses_when_full() {
489        let ring = Ring::new(3, 2);
490        assert_eq!(ring.room(), 4);
491        for n in 0..4 {
492            assert!(ring.push(&[n, n + 10]));
493        }
494        assert_eq!(ring.room(), 0);
495        assert!(!ring.push(&[9, 9]));
496        let mut slot = [0; 2];
497        for n in 0..4 {
498            assert!(ring.pop(&mut slot));
499            assert_eq!(slot, [n, n + 10]);
500        }
501        assert!(!ring.pop(&mut slot));
502        assert_eq!(ring.room(), 4);
503    }
504
505    #[test]
506    fn a_ring_of_nothing_still_holds_one() {
507        let ring = Ring::new(0, 1);
508        assert_eq!(ring.room(), 1);
509    }
510
511    #[test]
512    fn a_rings_counters_wrap_past_their_width() {
513        let ring = Ring::new(4, 1);
514        ring.head.store(u32::MAX - 1, Ordering::Relaxed);
515        ring.tail.store(u32::MAX - 1, Ordering::Relaxed);
516        let mut slot = [0; 1];
517        for n in 0..16 {
518            assert!(ring.push(&[n]), "{n}");
519            assert!(ring.push(&[n + 100]), "{n}");
520            assert_eq!(ring.room(), 2);
521            assert!(ring.pop(&mut slot));
522            assert_eq!(slot[0], n);
523            assert!(ring.pop(&mut slot));
524            assert_eq!(slot[0], n + 100);
525            assert_eq!(ring.room(), 4);
526        }
527    }
528
529    #[test]
530    fn words_in_bulk_come_out_as_they_went_in() {
531        let ring = Ring::new(4, 1);
532        ring.write_words(3, |n| u32::try_from(n).unwrap_or(0) * 2);
533        let mut seen = Vec::new();
534        assert_eq!(ring.read_words(2, |n, word| seen.push((n, word))), 2);
535        assert_eq!(seen, [(0, 0), (1, 2)]);
536        assert_eq!(ring.room(), 3);
537        seen.clear();
538        assert_eq!(ring.read_words(9, |n, word| seen.push((n, word))), 1);
539        assert_eq!(seen, [(0, 4)]);
540        assert_eq!(ring.read_words(9, |_, _| panic!("nothing left")), 0);
541    }
542
543    fn handle(index: u32, generation: u32) -> Handle {
544        Handle { index, generation }
545    }
546
547    prop_compose! {
548        fn params()(
549            volume: u16, looped: bool, rate: u32, left: u16, right: u16,
550            region: Option<(u64, u64)>,
551        ) -> Params {
552            let mut params = Params::new()
553                .with_volume(volume)
554                .with_rate(rate)
555                .with_balance(left, right);
556            params.looped = looped;
557            params.region = region.and_then(|(a, b)| Region::new(a.min(b), a.max(b)));
558            params
559        }
560    }
561
562    fn command() -> impl Strategy<Value = Command> {
563        let voice = (any::<u32>(), any::<u32>()).prop_map(|(i, g)| VoiceId(handle(i, g)));
564        let id = (any::<u32>(), any::<u32>()).prop_map(|(i, g)| handle(i, g));
565        prop_oneof![
566            (voice.clone(), id.clone(), params()).prop_map(|(v, i, p)| Command::Play(
567                v,
568                SoundId(i),
569                p
570            )),
571            (voice.clone(), id, params()).prop_map(|(v, i, p)| Command::Stream(v, StreamId(i), p)),
572            (voice.clone(), any::<u16>()).prop_map(|(v, n)| Command::SetVolume(v, n)),
573            (voice.clone(), any::<u16>(), any::<u64>())
574                .prop_map(|(v, n, m)| Command::FadeTo(v, n, m)),
575            (voice.clone(), any::<u16>(), any::<u16>())
576                .prop_map(|(v, l, r)| Command::SetBalance(v, l, r)),
577            voice.clone().prop_map(Command::Stop),
578            voice.clone().prop_map(Command::Pause),
579            voice.prop_map(Command::Resume),
580        ]
581    }
582
583    proptest! {
584        /// Whatever a command is, its words give it back.
585        #[test]
586        fn commands_round_trip(command in command()) {
587            prop_assert_eq!(Command::decode(&command.encode()), Some(command));
588        }
589    }
590
591    #[test]
592    fn words_that_are_no_command_are_none() {
593        assert_eq!(Command::decode(&[8; COMMAND_WORDS]), None);
594    }
595}
shuyi_sound/mix/queue.rs

shuyi_sound/mix/
queue.rs
