Properly emit events from the render thread for offline rendering

Author: orottier

src/context/offline.rs

@@ -102,7 +102,7 @@ impl OfflineAudioContext {
 
         // Communication channel for events from the render thread to the control thread.
         // Use an unbounded channel because we do not require real-time safety.
-        let event_channel = crossbeam_channel::unbounded();
+        let (event_send, event_recv) = crossbeam_channel::unbounded();
 
         // setup the render 'thread', which will run inside the control thread
         let renderer = RenderThread::new(
@@ -111,6 +111,7 @@ impl OfflineAudioContext {
             receiver,
             state_clone,
             frames_played_clone,
+            event_send.clone(),
         );
 
         // first, setup the base audio context
@@ -120,7 +121,7 @@ impl OfflineAudioContext {
             state,
             frames_played,
             sender,
-            event_channel,
+            (event_send, event_recv),
             true,
             node_id_consumer,
         );

src/io/cpal.rs

@@ -193,8 +193,9 @@ impl AudioBackendManager for CpalBackend {
             ctrl_msg_recv.clone(),
             Arc::clone(&state),
             Arc::clone(&frames_played),
+            event_send.clone(),
         );
-        renderer.set_event_channels(load_value_send.clone(), event_send.clone());
+        renderer.set_load_value_sender(load_value_send.clone());
         renderer.spawn_garbage_collector_thread();
 
         log::debug!(
@@ -235,8 +236,9 @@ impl AudioBackendManager for CpalBackend {
                     ctrl_msg_recv,
                     state,
                     frames_played,
+                    event_send,
                 );
-                renderer.set_event_channels(load_value_send, event_send);
+                renderer.set_load_value_sender(load_value_send);
                 renderer.spawn_garbage_collector_thread();
 
                 let spawned = spawn_output_stream(

src/io/cubeb.rs

@@ -191,8 +191,9 @@ impl AudioBackendManager for CubebBackend {
             ctrl_msg_recv,
             state,
             frames_played,
+            event_send,
         );
-        renderer.set_event_channels(load_value_send, event_send);
+        renderer.set_load_value_sender(load_value_send);
         renderer.spawn_garbage_collector_thread();
 
         let params = cubeb::StreamParamsBuilder::new()

src/io/none.rs

@@ -94,8 +94,9 @@ impl AudioBackendManager for NoneBackend {
             ctrl_msg_recv,
             state,
             frames_played,
+            event_send,
         );
-        render_thread.set_event_channels(load_value_send, event_send);
+        render_thread.set_load_value_sender(load_value_send);
         render_thread.spawn_garbage_collector_thread();
 
         // Use a bounded channel for real-time safety. A maximum of 32 control messages (resume,

src/render/graph.rs

@@ -750,7 +750,7 @@ mod tests {
             current_time: 0.,
             sample_rate: 48000.,
             node_id: std::cell::Cell::new(AudioNodeId(0)),
-            event_sender: None,
+            event_sender: crossbeam_channel::unbounded().0,
         };
         graph.render(&scope);
 
@@ -807,7 +807,7 @@ mod tests {
             current_time: 0.,
             sample_rate: 48000.,
             node_id: std::cell::Cell::new(AudioNodeId(0)),
-            event_sender: None,
+            event_sender: crossbeam_channel::unbounded().0,
         };
 
         // render twice
@@ -875,7 +875,7 @@ mod tests {
             current_time: 0.,
             sample_rate: 48000.,
             node_id: std::cell::Cell::new(AudioNodeId(0)),
-            event_sender: None,
+            event_sender: crossbeam_channel::unbounded().0,
         };
 
         // render twice
@@ -924,7 +924,7 @@ mod tests {
             current_time: 0.,
             sample_rate: 48000.,
             node_id: std::cell::Cell::new(AudioNodeId(0)),
-            event_sender: None,
+            event_sender: crossbeam_channel::unbounded().0,
         };
         graph.render(&scope);
 

src/render/processor.rs

@@ -23,7 +23,7 @@ pub struct AudioWorkletGlobalScope {
     pub sample_rate: f32,
 
     pub(crate) node_id: Cell<AudioNodeId>,
-    pub(crate) event_sender: Option<Sender<EventDispatch>>,
+    pub(crate) event_sender: Sender<EventDispatch>,
 }
 
 impl std::fmt::Debug for AudioWorkletGlobalScope {
@@ -44,17 +44,17 @@ impl AudioWorkletGlobalScope {
     /// [`MessagePort`](https://webaudio.github.io/web-audio-api/#dom-audioworkletprocessor-port)
     /// `postMessage` functionality of the AudioWorkletProcessor.
     pub fn post_message(&self, msg: Box<dyn Any + Send + 'static>) {
-        if let Some(sender) = self.event_sender.as_ref() {
-            // sending could fail if the channel is saturated or the main thread is shutting down
-            let _ = sender.try_send(EventDispatch::message(self.node_id.get(), msg));
-        }
+        // sending could fail if the channel is saturated or the main thread is shutting down
+        let _ = self
+            .event_sender
+            .try_send(EventDispatch::message(self.node_id.get(), msg));
     }
 
     pub(crate) fn send_ended_event(&self) {
-        if let Some(sender) = self.event_sender.as_ref() {
-            // sending could fail if the channel is saturated or the main thread is shutting down
-            let _ = sender.try_send(EventDispatch::ended(self.node_id.get()));
-        }
+        // sending could fail if the channel is saturated or the main thread is shutting down
+        let _ = self
+            .event_sender
+            .try_send(EventDispatch::ended(self.node_id.get()));
     }
 
     pub(crate) fn report_error(&self, error: Box<dyn Any + Send>) {
@@ -73,16 +73,16 @@ impl AudioWorkletGlobalScope {
             &message
         );
 
-        if let Some(sender) = self.event_sender.as_ref() {
-            let event = ErrorEvent {
-                message,
-                error,
-                event: Event {
-                    type_: "ErrorEvent",
-                },
-            };
-            let _ = sender.try_send(EventDispatch::processor_error(self.node_id.get(), event));
-        }
+        let event = ErrorEvent {
+            message,
+            error,
+            event: Event {
+                type_: "ErrorEvent",
+            },
+        };
+        let _ = self
+            .event_sender
+            .try_send(EventDispatch::processor_error(self.node_id.get(), event));
     }
 }
 

src/render/thread.rs

@@ -39,7 +39,7 @@ pub(crate) struct RenderThread {
     receiver: Option<Receiver<ControlMessage>>,
     buffer_offset: Option<(usize, AudioRenderQuantum)>,
     load_value_sender: Option<Sender<AudioRenderCapacityLoad>>,
-    event_sender: Option<Sender<EventDispatch>>,
+    event_sender: Sender<EventDispatch>,
     garbage_collector: Option<llq::Producer<Box<dyn Any + Send>>>,
 }
 
@@ -72,6 +72,7 @@ impl RenderThread {
         receiver: Receiver<ControlMessage>,
         state: Arc<AtomicU8>,
         frames_played: Arc<AtomicU64>,
+        event_sender: Sender<EventDispatch>,
     ) -> Self {
         Self {
             graph: None,
@@ -84,18 +85,16 @@ impl RenderThread {
             receiver: Some(receiver),
             buffer_offset: None,
             load_value_sender: None,
-            event_sender: None,
+            event_sender,
             garbage_collector: None,
         }
     }
 
-    pub(crate) fn set_event_channels(
+    pub(crate) fn set_load_value_sender(
         &mut self,
         load_value_sender: Sender<AudioRenderCapacityLoad>,
-        event_sender: Sender<EventDispatch>,
     ) {
         self.load_value_sender = Some(load_value_sender);
-        self.event_sender = Some(event_sender);
     }
 
     pub(crate) fn spawn_garbage_collector_thread(&mut self) {
@@ -174,14 +173,12 @@ impl RenderThread {
                     }
                 }
                 RunDiagnostics { mut buffer } => {
-                    if let Some(sender) = self.event_sender.as_ref() {
-                        use std::io::Write;
-                        writeln!(&mut buffer, "{:#?}", &self).ok();
-                        writeln!(&mut buffer, "{:?}", &self.graph).ok();
-                        sender
-                            .try_send(EventDispatch::diagnostics(buffer))
-                            .expect("Unable to send diagnostics - channel is full");
-                    }
+                    use std::io::Write;
+                    writeln!(&mut buffer, "{:#?}", &self).ok();
+                    writeln!(&mut buffer, "{:?}", &self.graph).ok();
+                    self.event_sender
+                        .try_send(EventDispatch::diagnostics(buffer))
+                        .expect("Unable to send diagnostics - channel is full");
                 }
                 Suspend { notify } => {
                     self.suspended = true;
@@ -460,9 +457,9 @@ impl RenderThread {
 
     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(state)).ok();
-        }
+        self.event_sender
+            .try_send(EventDispatch::state_change(state))
+            .ok();
     }
 }