After each OfflineAudioContext render quantum, spin the event loop

Author: orottier

src/context/concrete_base.rs

@@ -13,7 +13,7 @@ use crate::spatial::AudioListenerParams;
 
 use crate::AudioListener;
 
-use crossbeam_channel::{Receiver, SendError, Sender};
+use crossbeam_channel::{SendError, Sender};
 use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
 use std::sync::{Arc, Mutex, RwLock, RwLockWriteGuard};
 
@@ -126,13 +126,11 @@ impl ConcreteBaseAudioContext {
         state: Arc<AtomicU8>,
         frames_played: Arc<AtomicU64>,
         render_channel: Sender<ControlMessage>,
-        event_channel: (Sender<EventDispatch>, Receiver<EventDispatch>),
+        event_send: Sender<EventDispatch>,
+        event_loop: EventLoop,
         offline: bool,
         node_id_consumer: llq::Consumer<AudioNodeId>,
     ) -> Self {
-        let event_loop = EventLoop::new();
-        let (event_send, event_recv) = event_channel;
-
         let audio_node_id_provider = AudioNodeIdProvider::new(node_id_consumer);
 
         let base_inner = ConcreteBaseAudioContextInner {
@@ -147,7 +145,7 @@ impl ConcreteBaseAudioContext {
             listener_params: None,
             offline,
             state,
-            event_loop: event_loop.clone(),
+            event_loop,
             event_send,
         };
         let base = Self {
@@ -216,9 +214,6 @@ impl ConcreteBaseAudioContext {
             crate::node::load_hrtf_processor(sample_rate as u32);
         }
 
-        // Boot the event loop thread that handles the events spawned by the render thread
-        event_loop.run(event_recv);
-
         base
     }
 

src/context/offline.rs

@@ -9,6 +9,7 @@ use crate::render::RenderThread;
 use crate::{assert_valid_sample_rate, RENDER_QUANTUM_SIZE};
 use crate::{Event, OfflineAudioCompletionEvent};
 
+use crate::events::EventLoop;
 use futures_channel::{mpsc, oneshot};
 use futures_util::SinkExt as _;
 
@@ -52,6 +53,8 @@ struct OfflineAudioContextRenderer {
     onstatechange_handler: Option<Box<dyn FnMut(Event) + Send + 'static>>,
     /// event handler for complete event
     oncomplete_handler: Option<Box<dyn FnOnce(OfflineAudioCompletionEvent) + Send + 'static>>,
+    /// event loop to run after each render quantum
+    event_loop: EventLoop,
 }
 
 impl BaseAudioContext for OfflineAudioContext {
@@ -103,6 +106,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_send, event_recv) = crossbeam_channel::unbounded();
+        let event_loop = EventLoop::new(event_recv);
 
         // setup the render 'thread', which will run inside the control thread
         let renderer = RenderThread::new(
@@ -121,7 +125,8 @@ impl OfflineAudioContext {
             state,
             frames_played,
             sender,
-            (event_send, event_recv),
+            event_send,
+            event_loop.clone(),
             true,
             node_id_consumer,
         );
@@ -135,6 +140,7 @@ impl OfflineAudioContext {
             resume_receiver,
             onstatechange_handler: None,
             oncomplete_handler: None,
+            event_loop,
         };
 
         Self {
@@ -169,13 +175,14 @@ impl OfflineAudioContext {
             suspend_callbacks,
             oncomplete_handler,
             mut onstatechange_handler,
+            event_loop,
             ..
         } = renderer;
 
         self.base.set_state(AudioContextState::Running);
         Self::emit_statechange(&mut onstatechange_handler);
 
-        let result = renderer.render_audiobuffer_sync(self.length, suspend_callbacks, self);
+        let result = renderer.render_audiobuffer_sync(self, suspend_callbacks, event_loop);
 
         self.base.set_state(AudioContextState::Closed);
         Self::emit_statechange(&mut onstatechange_handler);

src/context/online.rs

@@ -3,7 +3,7 @@ use std::error::Error;
 use std::sync::Mutex;
 
 use crate::context::{AudioContextState, BaseAudioContext, ConcreteBaseAudioContext};
-use crate::events::{EventDispatch, EventHandler, EventPayload, EventType};
+use crate::events::{EventDispatch, EventHandler, EventLoop, EventPayload, EventType};
 use crate::io::{self, AudioBackendManager, ControlThreadInit, NoneBackend, RenderThreadInit};
 use crate::media_devices::{enumerate_devices_sync, MediaDeviceInfoKind};
 use crate::media_streams::{MediaStream, MediaStreamTrack};
@@ -181,11 +181,13 @@ impl AudioContext {
     #[allow(clippy::needless_pass_by_value)]
     #[must_use]
     pub fn new(mut options: AudioContextOptions) -> Self {
+        // Log, but ignore invalid sinks
         if !is_valid_sink_id(&options.sink_id) {
             log::error!("NotFoundError: invalid sinkId {:?}", options.sink_id);
             options.sink_id = String::from("");
         }
 
+        // Set up the audio output thread
         let (control_thread_init, render_thread_init) = io::thread_init();
         let backend = io::build_output(options, render_thread_init.clone());
 
@@ -198,26 +200,37 @@ impl AudioContext {
             event_recv,
         } = control_thread_init;
 
+        // Construct the audio Graph and hand it to the render thread
         let (node_id_producer, node_id_consumer) = llq::Queue::new().split();
         let graph = Graph::new(node_id_producer);
         let message = ControlMessage::Startup { graph };
         ctrl_msg_send.send(message).unwrap();
 
+        // Set up the event loop thread that handles the events spawned by the render thread
+        let event_loop = EventLoop::new(event_recv);
+
+        // Put everything together in the BaseAudioContext (shared with offline context)
         let base = ConcreteBaseAudioContext::new(
             backend.sample_rate(),
             backend.number_of_channels(),
             state,
             frames_played,
             ctrl_msg_send,
-            (event_send, event_recv),
+            event_send,
+            event_loop.clone(),
             false,
             node_id_consumer,
         );
 
-        // setup AudioRenderCapacity for this context
+        // Setup AudioRenderCapacity for this context
         let base_clone = base.clone();
         let render_capacity = AudioRenderCapacity::new(base_clone, load_value_recv);
 
+        // As the final step, spawn a thread for the event loop. If we do this earlier we may miss
+        // event handling of the initial events that are emitted right after render thread
+        // construction.
+        event_loop.run_in_thread();
+
         Self {
             base,
             backend_manager: Mutex::new(backend),

src/events.rs

@@ -4,6 +4,7 @@ use crate::{AudioBuffer, AudioRenderCapacityEvent};
 use std::any::Any;
 use std::collections::HashMap;
 use std::hash::Hash;
+use std::ops::ControlFlow;
 use std::sync::{Arc, Mutex};
 
 use crossbeam_channel::Receiver;
@@ -120,52 +121,69 @@ pub(crate) enum EventHandler {
     Multiple(Box<dyn FnMut(EventPayload) + Send + 'static>),
 }
 
-#[derive(Clone, Default)]
+#[derive(Clone)]
 pub(crate) struct EventLoop {
+    event_recv: Receiver<EventDispatch>,
     event_handlers: Arc<Mutex<HashMap<EventType, EventHandler>>>,
 }
 
 impl EventLoop {
-    pub fn new() -> Self {
-        Self::default()
+    pub fn new(event_recv: Receiver<EventDispatch>) -> Self {
+        Self {
+            event_recv,
+            event_handlers: Default::default(),
+        }
     }
 
-    pub fn run(&self, event_channel: Receiver<EventDispatch>) {
-        log::debug!("Entering event loop");
-        let self_clone = self.clone();
+    fn handle_event(&self, mut event: EventDispatch) -> ControlFlow<()> {
+        // Terminate the event loop when the audio context is closing
+        let mut result = ControlFlow::Continue(());
+        if matches!(
+            event.payload,
+            EventPayload::AudioContextState(AudioContextState::Closed)
+        ) {
+            event.payload = EventPayload::None; // the statechange handler takes no argument
+            result = ControlFlow::Break(());
+        }
 
-        std::thread::spawn(move || {
-            // This thread is dedicated to event handling so we can block
-            for mut event in event_channel.iter() {
-                // Terminate the event loop when the audio context is closing
-                let mut terminate = false;
-                if matches!(
-                    event.payload,
-                    EventPayload::AudioContextState(AudioContextState::Closed)
-                ) {
-                    event.payload = EventPayload::None; // the statechange handler takes no argument
-                    terminate = true;
+        let mut event_handler_lock = self.event_handlers.lock().unwrap();
+        let callback_option = event_handler_lock.remove(&event.type_);
+        drop(event_handler_lock); // release Mutex while running callback
+
+        if let Some(callback) = callback_option {
+            match callback {
+                EventHandler::Once(f) => (f)(event.payload),
+                EventHandler::Multiple(mut f) => {
+                    (f)(event.payload);
+                    self.event_handlers
+                        .lock()
+                        .unwrap()
+                        .insert(event.type_, EventHandler::Multiple(f));
                 }
+            };
+        }
 
-                let mut event_handler_lock = self_clone.event_handlers.lock().unwrap();
-                let callback_option = event_handler_lock.remove(&event.type_);
-                drop(event_handler_lock); // release Mutex while running callback
-
-                if let Some(callback) = callback_option {
-                    match callback {
-                        EventHandler::Once(f) => (f)(event.payload),
-                        EventHandler::Multiple(mut f) => {
-                            (f)(event.payload);
-                            self_clone
-                                .event_handlers
-                                .lock()
-                                .unwrap()
-                                .insert(event.type_, EventHandler::Multiple(f));
-                        }
-                    };
-                }
+        result
+    }
 
-                if terminate {
+    pub fn handle_pending_events(&self) {
+        // try_iter will yield all pending events, but does not block
+        for event in self.event_recv.try_iter() {
+            self.handle_event(event);
+        }
+    }
+
+    pub fn run_in_thread(&self) {
+        log::debug!("Entering event thread");
+
+        // split borrows to help compiler
+        let self_clone = self.clone();
+
+        std::thread::spawn(move || {
+            // This thread is dedicated to event handling, so we can block
+            for event in self_clone.event_recv.iter() {
+                let result = self_clone.handle_event(event);
+                if result.is_break() {
                     break;
                 }
             }

src/render/thread.rs

@@ -17,7 +17,7 @@ use crate::buffer::{AudioBuffer, AudioBufferOptions};
 use crate::context::{
     AudioContextState, AudioNodeId, OfflineAudioContext, OfflineAudioContextCallback,
 };
-use crate::events::EventDispatch;
+use crate::events::{EventDispatch, EventLoop};
 use crate::message::ControlMessage;
 use crate::node::ChannelInterpretation;
 use crate::render::AudioWorkletGlobalScope;
@@ -225,10 +225,12 @@ impl RenderThread {
     // cf. https://webaudio.github.io/web-audio-api/#dom-offlineaudiocontext-startrendering
     pub fn render_audiobuffer_sync(
         mut self,
-        length: usize,
-        mut suspend_callbacks: Vec<(usize, Box<OfflineAudioContextCallback>)>,
         context: &mut OfflineAudioContext,
+        mut suspend_callbacks: Vec<(usize, Box<OfflineAudioContextCallback>)>,
+        event_loop: EventLoop,
     ) -> AudioBuffer {
+        let length = context.length();
+
         let options = AudioBufferOptions {
             number_of_channels: self.number_of_channels,
             length,
@@ -246,12 +248,14 @@ impl RenderThread {
             if suspend_callbacks.first().map(|&(q, _)| q) == Some(quantum) {
                 let callback = suspend_callbacks.remove(0).1;
                 (callback)(context);
-
-                // Handle addition/removal of nodes/edges
-                self.handle_control_messages();
             }
 
+            // Handle addition/removal of nodes/edges
+            self.handle_control_messages();
+
             self.render_offline_quantum(&mut buffer);
+
+            event_loop.handle_pending_events();
         }
 
         buffer