BaseAudioContext::state set by render thread to signal device readiness

Fixes #410

Except for suspending/resuming, I don't have a solution for that yet.

In summary:
- A context is instantiated in Suspended state
- Online context will change to Running at the first render quantum
- Online context will change to Closed when the render thread is dropped
  after a `close` call
- Online context suspend/resume state is still managed by the control
  thread - TODO
- Offline context will change to Running after startRendering

Author: orottier

src/context/concrete_base.rs

@@ -91,8 +91,8 @@ struct ConcreteBaseAudioContextInner {
     listener_params: Option<AudioListenerParams>,
     /// Denotes if this AudioContext is offline or not
     offline: bool,
-    /// Describes the current state of the `ConcreteBaseAudioContext`
-    state: AtomicU8,
+    /// Current state of the `ConcreteBaseAudioContext`, shared with the RenderThread
+    state: Arc<AtomicU8>,
     /// Stores the event handlers
     event_loop: EventLoop,
     /// Sender for events that will be handled by the EventLoop
@@ -150,6 +150,7 @@ impl ConcreteBaseAudioContext {
     pub(super) fn new(
         sample_rate: f32,
         max_channel_count: usize,
+        state: Arc<AtomicU8>,
         frames_played: Arc<AtomicU64>,
         render_channel: Sender<ControlMessage>,
         event_channel: Option<(Sender<EventDispatch>, Receiver<EventDispatch>)>,
@@ -175,7 +176,7 @@ impl ConcreteBaseAudioContext {
             queued_audio_listener_msgs: Mutex::new(Vec::new()),
             listener_params: None,
             offline,
-            state: AtomicU8::new(AudioContextState::Suspended as u8),
+            state,
             event_loop: event_loop.clone(),
             event_send,
         };
@@ -263,7 +264,6 @@ impl ConcreteBaseAudioContext {
         if self.state() != AudioContextState::Closed {
             let result = self.inner.render_channel.read().unwrap().send(msg);
             if result.is_err() {
-                self.set_state(AudioContextState::Closed);
                 log::warn!("Discarding control message - render thread is closed");
             }
         }
@@ -333,14 +333,14 @@ impl ConcreteBaseAudioContext {
     /// Returns state of current context
     #[must_use]
     pub(super) fn state(&self) -> AudioContextState {
-        self.inner.state.load(Ordering::SeqCst).into()
+        self.inner.state.load(Ordering::Acquire).into()
     }
 
     /// Updates state of current context
     pub(super) fn set_state(&self, state: AudioContextState) {
         let current_state = self.state();
         if current_state != state {
-            self.inner.state.store(state as u8, Ordering::SeqCst);
+            self.inner.state.store(state as u8, Ordering::Release);
             let _ = self.send_event(EventDispatch::state_change());
         }
     }

src/context/offline.rs

@@ -1,10 +1,10 @@
 //! The `OfflineAudioContext` type
 
-use std::sync::atomic::AtomicU64;
+use std::sync::atomic::{AtomicU64, AtomicU8};
 use std::sync::{Arc, Mutex};
 
 use crate::buffer::AudioBuffer;
-use crate::context::{BaseAudioContext, ConcreteBaseAudioContext};
+use crate::context::{AudioContextState, BaseAudioContext, ConcreteBaseAudioContext};
 use crate::render::RenderThread;
 use crate::{assert_valid_sample_rate, RENDER_QUANTUM_SIZE};
 
@@ -71,19 +71,23 @@ impl OfflineAudioContext {
         // track number of frames - synced from render thread to control thread
         let frames_played = Arc::new(AtomicU64::new(0));
         let frames_played_clone = Arc::clone(&frames_played);
+        let state = Arc::new(AtomicU8::new(AudioContextState::Suspended as u8));
+        let state_clone = Arc::clone(&state);
 
         // setup the render 'thread', which will run inside the control thread
         let renderer = RenderThread::new(
             sample_rate,
             number_of_channels,
             receiver,
+            state_clone,
             frames_played_clone,
         );
 
         // first, setup the base audio context
         let base = ConcreteBaseAudioContext::new(
             sample_rate,
             number_of_channels,
+            state,
             frames_played,
             sender,
             None,

src/context/online.rs

@@ -166,6 +166,7 @@ impl AudioContext {
         let backend = io::build_output(options, render_thread_init.clone());
 
         let ControlThreadInit {
+            state,
             frames_played,
             ctrl_msg_send,
             load_value_recv,
@@ -181,13 +182,13 @@ impl AudioContext {
         let base = ConcreteBaseAudioContext::new(
             backend.sample_rate(),
             backend.number_of_channels(),
+            state,
             frames_played,
             ctrl_msg_send,
             Some((event_send, event_recv)),
             false,
             node_id_consumer,
         );
-        base.set_state(AudioContextState::Running);
 
         // setup AudioRenderCapacity for this context
         let base_clone = base.clone();
@@ -255,9 +256,6 @@ impl AudioContext {
             return Ok(());
         }
 
-        // Temporarily set the state to Suspended, resume after the new backend is up
-        self.base().set_state(AudioContextState::Suspended);
-
         // Acquire exclusive lock on ctrl msg sender
         let ctrl_msg_send = self.base.lock_control_msg_sender();
 
@@ -286,6 +284,8 @@ impl AudioContext {
             graph_recv.recv().unwrap()
         };
 
+        backend_manager_guard.close();
+
         // hotswap the backend
         let options = AudioContextOptions {
             sample_rate: Some(self.sample_rate()),
@@ -304,10 +304,6 @@ impl AudioContext {
         let message = ControlMessage::Startup { graph };
         ctrl_msg_send.send(message).unwrap();
 
-        if original_state == AudioContextState::Running {
-            self.base().set_state(AudioContextState::Running);
-        }
-
         // flush the cached msgs
         pending_msgs
             .into_iter()
@@ -394,6 +390,8 @@ impl AudioContext {
     #[allow(clippy::missing_const_for_fn, clippy::unused_self)]
     pub fn suspend_sync(&self) {
         if self.backend_manager.lock().unwrap().suspend() {
+            // TODO: state should signal device readiness hence should be updated by the render
+            // thread, not the control thread.
             self.base().set_state(AudioContextState::Suspended);
         }
     }
@@ -413,6 +411,8 @@ impl AudioContext {
     #[allow(clippy::missing_const_for_fn, clippy::unused_self)]
     pub fn resume_sync(&self) {
         if self.backend_manager.lock().unwrap().resume() {
+            // TODO: state should signal device readiness hence should be updated by the render
+            // thread, not the control thread.
             self.base().set_state(AudioContextState::Running);
         }
     }
@@ -432,7 +432,6 @@ impl AudioContext {
     pub fn close_sync(&self) {
         self.backend_manager.lock().unwrap().close();
         self.render_capacity.stop();
-        self.base().set_state(AudioContextState::Closed);
     }
 
     /// Creates a [`MediaStreamAudioSourceNode`](node::MediaStreamAudioSourceNode) from a

src/io/cpal.rs

@@ -123,6 +123,7 @@ impl AudioBackendManager for CpalBackend {
         log::info!("Audio Output Host: cpal {:?}", host.id());
 
         let RenderThreadInit {
+            state,
             frames_played,
             ctrl_msg_recv,
             load_value_send,
@@ -190,6 +191,7 @@ impl AudioBackendManager for CpalBackend {
             sample_rate,
             preferred_config.channels as usize,
             ctrl_msg_recv.clone(),
+            Arc::clone(&state),
             Arc::clone(&frames_played),
         );
         renderer.set_event_channels(load_value_send.clone(), event_send.clone());
@@ -231,6 +233,7 @@ impl AudioBackendManager for CpalBackend {
                     sample_rate,
                     supported_config.channels as usize,
                     ctrl_msg_recv,
+                    state,
                     frames_played,
                 );
                 renderer.set_event_channels(load_value_send, event_send);

src/io/cubeb.rs

@@ -151,6 +151,7 @@ impl AudioBackendManager for CubebBackend {
         Self: Sized,
     {
         let RenderThreadInit {
+            state,
             frames_played,
             ctrl_msg_recv,
             load_value_send,
@@ -186,6 +187,7 @@ impl AudioBackendManager for CubebBackend {
             sample_rate,
             number_of_channels,
             ctrl_msg_recv,
+            state,
             frames_played,
         );
         renderer.set_event_channels(load_value_send, event_send);

src/io/mod.rs

@@ -1,12 +1,12 @@
 //! Audio input/output interfaces
 
-use std::sync::atomic::AtomicU64;
+use std::sync::atomic::{AtomicU64, AtomicU8};
 use std::sync::Arc;
 
 use crossbeam_channel::{Receiver, Sender};
 
 use crate::buffer::AudioBuffer;
-use crate::context::{AudioContextLatencyCategory, AudioContextOptions};
+use crate::context::{AudioContextLatencyCategory, AudioContextOptions, AudioContextState};
 use crate::events::EventDispatch;
 use crate::media_devices::MediaDeviceInfo;
 use crate::media_streams::{MediaStream, MediaStreamTrack};
@@ -26,6 +26,7 @@ mod microphone;
 
 #[derive(Debug)]
 pub(crate) struct ControlThreadInit {
+    pub state: Arc<AtomicU8>,
     pub frames_played: Arc<AtomicU64>,
     pub ctrl_msg_send: Sender<ControlMessage>,
     pub load_value_recv: Receiver<AudioRenderCapacityLoad>,
@@ -35,13 +36,17 @@ pub(crate) struct ControlThreadInit {
 
 #[derive(Clone, Debug)]
 pub(crate) struct RenderThreadInit {
+    pub state: Arc<AtomicU8>,
     pub frames_played: Arc<AtomicU64>,
     pub ctrl_msg_recv: Receiver<ControlMessage>,
     pub load_value_send: Sender<AudioRenderCapacityLoad>,
     pub event_send: Sender<EventDispatch>,
 }
 
 pub(crate) fn thread_init() -> (ControlThreadInit, RenderThreadInit) {
+    // Track audio context state - synced from render thread to control thread
+    let state = Arc::new(AtomicU8::new(AudioContextState::Suspended as u8));
+
     // Track number of frames - synced from render thread to control thread
     let frames_played = Arc::new(AtomicU64::new(0));
 
@@ -61,6 +66,7 @@ pub(crate) fn thread_init() -> (ControlThreadInit, RenderThreadInit) {
     let (event_send, event_recv) = crossbeam_channel::bounded(256);
 
     let control_thread_init = ControlThreadInit {
+        state: Arc::clone(&state),
         frames_played: Arc::clone(&frames_played),
         ctrl_msg_send,
         load_value_recv,
@@ -69,6 +75,7 @@ pub(crate) fn thread_init() -> (ControlThreadInit, RenderThreadInit) {
     };
 
     let render_thread_init = RenderThreadInit {
+        state,
         frames_played,
         ctrl_msg_recv,
         load_value_send,

src/io/none.rs

@@ -71,14 +71,20 @@ impl AudioBackendManager for NoneBackend {
         let sample_rate = options.sample_rate.unwrap_or(48000.);
 
         let RenderThreadInit {
+            state,
             frames_played,
             ctrl_msg_recv,
             load_value_send,
             event_send,
         } = render_thread_init;
 
-        let mut render_thread =
-            RenderThread::new(sample_rate, MAX_CHANNELS, ctrl_msg_recv, frames_played);
+        let mut render_thread = RenderThread::new(
+            sample_rate,
+            MAX_CHANNELS,
+            ctrl_msg_recv,
+            state,
+            frames_played,
+        );
         render_thread.set_event_channels(load_value_send, event_send);
         render_thread.spawn_garbage_collector_thread();
 

src/render/thread.rs

@@ -3,7 +3,7 @@
 use std::any::Any;
 use std::cell::Cell;
 
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -14,7 +14,9 @@ use futures::stream::StreamExt;
 
 use super::AudioRenderQuantum;
 use crate::buffer::{AudioBuffer, AudioBufferOptions};
-use crate::context::{AudioNodeId, OfflineAudioContext, OfflineAudioContextCallback};
+use crate::context::{
+    AudioContextState, AudioNodeId, OfflineAudioContext, OfflineAudioContextCallback,
+};
 use crate::events::EventDispatch;
 use crate::message::ControlMessage;
 use crate::node::ChannelInterpretation;
@@ -31,6 +33,7 @@ pub(crate) struct RenderThread {
     /// number of channels of the backend stream, i.e. sound card number of
     /// channels clamped to MAX_CHANNELS
     number_of_channels: usize,
+    state: Arc<AtomicU8>,
     frames_played: Arc<AtomicU64>,
     receiver: Option<Receiver<ControlMessage>>,
     buffer_offset: Option<(usize, AudioRenderQuantum)>,
@@ -66,13 +69,15 @@ impl RenderThread {
         sample_rate: f32,
         number_of_channels: usize,
         receiver: Receiver<ControlMessage>,
+        state: Arc<AtomicU8>,
         frames_played: Arc<AtomicU64>,
     ) -> Self {
         Self {
             graph: None,
             sample_rate,
             buffer_size: 0,
             number_of_channels,
+            state,
             frames_played,
             receiver: Some(receiver),
             buffer_offset: None,
@@ -150,13 +155,15 @@ impl RenderThread {
                     self.graph.as_mut().unwrap().mark_cycle_breaker(id);
                 }
                 Shutdown { sender } => {
+                    self.set_state(AudioContextState::Suspended);
                     let _ = sender.send(self.graph.take().unwrap());
                     self.receiver = None;
                     return; // no further handling of ctrl msgs
                 }
                 Startup { graph } => {
                     debug_assert!(self.graph.is_none());
                     self.graph = Some(graph);
+                    self.set_state(AudioContextState::Running);
                 }
                 NodeMessage { id, mut msg } => {
                     self.graph.as_mut().unwrap().route_message(id, msg.as_mut());
@@ -415,10 +422,23 @@ impl RenderThread {
             self.handle_control_messages();
         }
     }
+
+    fn set_state(&self, state: AudioContextState) {
+        self.state.store(state as u8, Ordering::Relaxed);
+        if let Some(sender) = self.event_sender.as_ref() {
+            sender.try_send(EventDispatch::state_change()).ok();
+        }
+    }
 }
 
 impl Drop for RenderThread {
     fn drop(&mut self) {
+        // If the audio graph is still there, this is an actual shutdown.
+        // When the graph is None, this means we are only changing the sink - not shutting down.
+        if self.graph.is_some() {
+            self.set_state(AudioContextState::Closed);
+        }
+
         if let Some(gc) = self.garbage_collector.as_mut() {
             gc.push(llq::Node::new(Box::new(TerminateGarbageCollectorThread)))
         }

tests/online.rs

@@ -106,10 +106,10 @@ fn test_none_sink_id() {
     assert_eq!(context.sink_id(), "none");
 
     context.suspend_sync();
-    assert_eq!(context.state(), AudioContextState::Suspended);
 
     // give event thread some time to pick up events
     std::thread::sleep(std::time::Duration::from_millis(20));
+    assert_eq!(context.state(), AudioContextState::Suspended);
     assert_eq!(state_changes.load(Ordering::Relaxed), 2); // suspended
 
     context.resume_sync();
@@ -120,6 +120,7 @@ fn test_none_sink_id() {
     assert_eq!(state_changes.load(Ordering::Relaxed), 3); // resumed
 
     context.close_sync();
+    std::thread::sleep(std::time::Duration::from_millis(20));
     assert_eq!(context.state(), AudioContextState::Closed);
 
     // give event thread some time to pick up events