Skip to main content

firewheel_graph/
context.rs

1use bevy_platform::sync::{
2    Arc,
3    atomic::{AtomicBool, Ordering},
4};
5use core::error::Error;
6use core::num::NonZeroU32;
7use core::time::Duration;
8use core::{any::Any, f64};
9use firewheel_core::node::{NodeError, ProcStore};
10use firewheel_core::{
11    StreamInfo,
12    channel_config::{ChannelConfig, ChannelCount},
13    diff::EventQueue,
14    dsp::declick::DeclickValues,
15    event::{NodeEvent, NodeEventType},
16    node::{AudioNode, DynAudioNode, NodeID},
17};
18use firewheel_core::{
19    dsp::volume::Volume,
20    log::{RealtimeLogger, RealtimeLoggerConfig, RealtimeLoggerMainThread},
21};
22use ringbuf::traits::{Consumer, Producer, Split};
23use smallvec::SmallVec;
24
25#[cfg(not(feature = "std"))]
26use num_traits::Float;
27
28#[cfg(feature = "scheduled_events")]
29use bevy_platform::time::Instant;
30#[cfg(feature = "scheduled_events")]
31use core::cell::RefCell;
32#[cfg(feature = "scheduled_events")]
33use firewheel_core::clock::{AudioClock, DurationSeconds};
34
35#[cfg(all(not(feature = "std"), feature = "musical_transport"))]
36use bevy_platform::prelude::Box;
37#[cfg(not(feature = "std"))]
38use bevy_platform::prelude::Vec;
39
40use crate::{
41    error::{ActivateError, RemoveNodeError},
42    processor::SharedFlags,
43};
44use crate::{
45    error::{AddEdgeError, UpdateError},
46    graph::{AudioGraph, Edge, EdgeID, NodeEntry, PortIdx},
47    processor::{
48        ContextToProcessorMsg, FirewheelProcessor, FirewheelProcessorInner, ProcessorToContextMsg,
49    },
50};
51use crate::{
52    error::{CompileGraphError, DeactivateError},
53    processor::{
54        BufferOutOfSpaceMode, FirewheelProcessorConfig, ProfilingData,
55        profiling::{ProfilerRx, ProfilerTx},
56    },
57};
58
59#[cfg(feature = "scheduled_events")]
60use crate::processor::{ClearScheduledEventsEvent, SharedClock};
61#[cfg(feature = "scheduled_events")]
62use firewheel_core::clock::EventInstant;
63
64#[cfg(feature = "musical_transport")]
65use firewheel_core::clock::TransportState;
66
67/// Information about the running audio stream.
68#[derive(Debug, Clone, Copy, PartialEq)]
69pub struct ActivateInfo {
70    /// The sample rate of the audio stream.
71    pub sample_rate: NonZeroU32,
72    /// The maximum number of frames that can appear in a single process cyle.
73    pub max_block_frames: NonZeroU32,
74    /// The number of input audio channels in the stream.
75    pub num_stream_in_channels: u32,
76    /// The number of output audio channels in the stream.
77    pub num_stream_out_channels: u32,
78    /// The latency of the input to output stream in seconds.
79    pub input_to_output_latency_seconds: f64,
80}
81
82/// The configuration of a Firewheel context.
83#[derive(Debug, Clone, Copy, PartialEq)]
84#[cfg_attr(feature = "bevy_reflect", derive(bevy_reflect::Reflect))]
85#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
86pub struct FirewheelConfig {
87    /// The number of input channels in the audio graph.
88    pub num_graph_inputs: ChannelCount,
89    /// The number of output channels in the audio graph.
90    pub num_graph_outputs: ChannelCount,
91
92    /// Extra configuration flags.
93    ///
94    /// By default, no flags are set.
95    pub flags: FirewheelFlags,
96
97    /// An initial capacity to allocate for the nodes in the audio graph.
98    ///
99    /// By default this is set to `128`.
100    pub initial_node_capacity: u32,
101    /// An initial capacity to allocate for the edges in the audio graph.
102    ///
103    /// By default this is set to `256`.
104    pub initial_edge_capacity: u32,
105    /// The amount of time in seconds to fade in/out when pausing/resuming
106    /// to avoid clicks and pops.
107    ///
108    /// By default this is set to `10.0 / 1_000.0`.
109    pub declick_seconds: f32,
110    /// The initial capacity for a group of events.
111    ///
112    /// By default this is set to `128`.
113    pub initial_event_group_capacity: u32,
114    /// The capacity of the engine's internal message channel.
115    ///
116    /// By default this is set to `64`.
117    pub channel_capacity: u32,
118    /// The maximum number of events that can be sent in a single call
119    /// to [`AudioNodeProcessor::process`].
120    ///
121    /// By default this is set to `128`.
122    ///
123    /// [`AudioNodeProcessor::process`]: firewheel_core::node::AudioNodeProcessor::process
124    pub event_queue_capacity: usize,
125    /// The maximum number of immediate events (events that do *NOT* have a
126    /// scheduled time component) that can be stored at once in the audio
127    /// thread.
128    ///
129    /// By default this is set to `512`.
130    pub immediate_event_capacity: usize,
131    /// The maximum number of scheduled events (events that have a scheduled
132    /// time component) that can be stored at once in the audio thread.
133    ///
134    /// This can be set to `0` to save some memory if you do not plan on using
135    /// scheduled events.
136    ///
137    /// This has no effect if the `scheduled_events` feature is disabled.
138    ///
139    /// By default this is set to `512`.
140    pub scheduled_event_capacity: usize,
141    /// How to handle event buffers on the audio thread running out of space.
142    ///
143    /// By default this is set to [`BufferOutOfSpaceMode::AllocateOnAudioThread`].
144    pub buffer_out_of_space_mode: BufferOutOfSpaceMode,
145
146    /// The configuration of the realtime safe logger.
147    pub logger_config: RealtimeLoggerConfig,
148
149    /// The initial number of slots to allocate for the [`ProcStore`].
150    ///
151    /// By default this is set to `8`.
152    pub proc_store_capacity: usize,
153
154    /// If `Some`, then inputs to the audio graph will be clamped to silence if the
155    /// max peak amplitude is less than the given volume. This can help improve the
156    /// performance of processing chains which use the graph inputs.
157    ///
158    /// If this is `None`, then no clamping will occur.
159    ///
160    /// Note, while this is functionally a noise gate, it is not a good noise gate,
161    /// and values above -70dB may cause audible clicking. If you need to increase
162    /// the threshold, it is recommended to instead use a dedicated noise gate node.
163    ///
164    /// By default this is set to `Some(Volume::Decibels(-70.0)`.
165    pub clamp_graph_inputs_below: Option<Volume>,
166}
167
168impl Default for FirewheelConfig {
169    fn default() -> Self {
170        Self {
171            num_graph_inputs: ChannelCount::ZERO,
172            num_graph_outputs: ChannelCount::STEREO,
173            flags: FirewheelFlags::default(),
174            initial_node_capacity: 128,
175            initial_edge_capacity: 256,
176            declick_seconds: DeclickValues::DEFAULT_FADE_SECONDS,
177            initial_event_group_capacity: 128,
178            channel_capacity: 64,
179            event_queue_capacity: 128,
180            immediate_event_capacity: 512,
181            scheduled_event_capacity: 512,
182            buffer_out_of_space_mode: BufferOutOfSpaceMode::AllocateOnAudioThread,
183            logger_config: RealtimeLoggerConfig::default(),
184            proc_store_capacity: 8,
185            clamp_graph_inputs_below: Some(Volume::Decibels(-70.0)),
186        }
187    }
188}
189
190/// Configuration flags for a [`FirewheelContext`]
191///
192/// Unlike [`FirewheelConfig`], these flags can be changed after the context has
193/// been created.
194#[derive(Default, Debug, Clone, Copy, PartialEq, Eq)]
195#[cfg_attr(feature = "bevy_reflect", derive(bevy_reflect::Reflect))]
196#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
197pub struct FirewheelFlags {
198    /// Hard clip all samples in the final output buffer to the range `[-1.0, 1.0]`.
199    ///
200    /// This usually isn't necessary since the OS itself generally hard clips the
201    /// output, but it is here if you need it.
202    ///
203    /// By default this is set to `false`.
204    pub hard_clip_outputs: bool,
205
206    /// Detect when a sample in the final output buffer falls outside the range
207    /// `[-1.0, 1.0]`. If a sample falls outside this range, then
208    /// [`FirewheelContext::clipping_occurred`] will return `true`.
209    ///
210    /// This check takes place before hard clipping if the
211    /// [`FirewheelFlags::hard_clip_outputs`] is set to `true`.
212    ///
213    /// By default this is set to `false`.
214    pub detect_clipping_on_output: bool,
215
216    /// Validate that all samples in the final output buffer are a valid finite
217    /// number. If a non finite number is detected, then the sample will will be
218    /// set to `0.0` and an error is logged.
219    ///
220    /// By default this is set to `false`.
221    pub validate_output_is_finite: bool,
222
223    /// Force all of a node's output buffers to be cleared before processing.
224    /// This shouldn't be necessary, but it can be used to debug nodes that
225    /// are misusing the silence flag feature.
226    ///
227    /// By default this is set to `false`.
228    pub force_clear_buffers: bool,
229
230    /// Enables performance profiling for engine bookkeeping operations on the
231    /// audio thread such as message handling, event sorting, event searching,
232    /// and final output processing.
233    ///
234    /// By default this is set to `false`.
235    pub profile_engine_bookkeeping: bool,
236
237    /// Enable per-node performance profiling.
238    ///
239    /// This has no effect when the `node_profiling` feature is disabled.
240    ///
241    /// By default this is set to `false`.
242    pub profile_nodes: bool,
243}
244
245bitflags::bitflags! {
246    #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
247    pub struct FirewheelBitFlags: u32 {
248        const HARD_CLIP_OUTPUTS = 1 << 0;
249        const DETECT_CLIPPING_ON_OUTPUT = 1 << 1;
250        const VALIDATE_OUTPUT_IS_FINITE = 1 << 2;
251        const FORCE_CLEAR_BUFFERS = 1 << 3;
252        const PROFILE_ENGINE_BOOKKEEPING = 1 << 4;
253        const PROFILE_NODES = 1 << 5;
254    }
255}
256
257impl From<FirewheelFlags> for FirewheelBitFlags {
258    fn from(value: FirewheelFlags) -> Self {
259        let mut b = Self::empty();
260        b.set(Self::HARD_CLIP_OUTPUTS, value.hard_clip_outputs);
261        b.set(
262            Self::DETECT_CLIPPING_ON_OUTPUT,
263            value.detect_clipping_on_output,
264        );
265        b.set(
266            Self::VALIDATE_OUTPUT_IS_FINITE,
267            value.validate_output_is_finite,
268        );
269        b.set(Self::FORCE_CLEAR_BUFFERS, value.force_clear_buffers);
270        b.set(
271            Self::PROFILE_ENGINE_BOOKKEEPING,
272            value.profile_engine_bookkeeping,
273        );
274        b.set(Self::PROFILE_NODES, value.profile_nodes);
275        b
276    }
277}
278
279pub(crate) struct ProcessorChannel {
280    pub(crate) shared_flags: Arc<SharedFlags>,
281    pub(crate) from_context_rx: ringbuf::HeapCons<ContextToProcessorMsg>,
282    pub(crate) to_context_tx: ringbuf::HeapProd<ProcessorToContextMsg>,
283    pub(crate) logger: RealtimeLogger,
284    pub(crate) store: ProcStore,
285    pub(crate) profiler_tx: ProfilerTx,
286    #[cfg(feature = "scheduled_events")]
287    pub(crate) shared_clock_input: triple_buffer::Input<SharedClock>,
288}
289
290/// A Firewheel context
291pub struct FirewheelContext {
292    graph: AudioGraph,
293
294    to_processor_tx: ringbuf::HeapProd<ContextToProcessorMsg>,
295    from_processor_rx: ringbuf::HeapCons<ProcessorToContextMsg>,
296    processor_drop_flag: Option<Arc<AtomicBool>>,
297    profiler_rx: ProfilerRx,
298    logger_rx: RealtimeLoggerMainThread,
299
300    pending_processor_channel: Option<ProcessorChannel>,
301    processor_drop_rx: Option<ringbuf::HeapCons<FirewheelProcessorInner>>,
302
303    #[cfg(feature = "scheduled_events")]
304    shared_clock_output: RefCell<triple_buffer::Output<SharedClock>>,
305
306    sample_rate: NonZeroU32,
307    sample_rate_recip: f64,
308    stream_info: Option<StreamInfo>,
309    shared_flags: Arc<SharedFlags>,
310
311    #[cfg(feature = "musical_transport")]
312    transport_state: Box<TransportState>,
313    #[cfg(feature = "musical_transport")]
314    transport_state_alloc_reuse: Option<Box<TransportState>>,
315
316    // Re-use the allocations for groups of events.
317    event_group_pool: Vec<Vec<NodeEvent>>,
318    event_group: Vec<NodeEvent>,
319    initial_event_group_capacity: usize,
320
321    #[cfg(feature = "scheduled_events")]
322    queued_clear_scheduled_events: Vec<ClearScheduledEventsEvent>,
323
324    config: FirewheelConfig,
325}
326
327impl FirewheelContext {
328    /// Create a new Firewheel context.
329    pub fn new(config: FirewheelConfig) -> Self {
330        let (to_processor_tx, from_context_rx) =
331            ringbuf::HeapRb::<ContextToProcessorMsg>::new(config.channel_capacity as usize).split();
332        let (to_context_tx, from_processor_rx) =
333            ringbuf::HeapRb::<ProcessorToContextMsg>::new(config.channel_capacity as usize * 2)
334                .split();
335
336        let initial_event_group_capacity = config.initial_event_group_capacity as usize;
337        let mut event_group_pool = Vec::with_capacity(16);
338        for _ in 0..3 {
339            event_group_pool.push(Vec::with_capacity(initial_event_group_capacity));
340        }
341
342        let graph = AudioGraph::new(&config);
343
344        #[cfg(feature = "scheduled_events")]
345        let (shared_clock_input, shared_clock_output) =
346            triple_buffer::triple_buffer(&SharedClock::default());
347
348        let (logger, logger_rx) = firewheel_core::log::realtime_logger(config.logger_config);
349        let (profiler_tx, profiler_rx) = crate::processor::profiling::profiler_channel(
350            config.initial_node_capacity as usize,
351            #[cfg(feature = "node_profiling")]
352            graph.graph_out_node(),
353        );
354        let shared_flags = Arc::new(SharedFlags::default());
355
356        let store = ProcStore::with_capacity(config.proc_store_capacity);
357
358        Self {
359            graph,
360            to_processor_tx,
361            from_processor_rx,
362            processor_drop_flag: None,
363            profiler_rx,
364            logger_rx,
365            pending_processor_channel: Some(ProcessorChannel {
366                shared_flags: Arc::clone(&shared_flags),
367                from_context_rx,
368                to_context_tx,
369                logger,
370                store,
371                profiler_tx,
372                #[cfg(feature = "scheduled_events")]
373                shared_clock_input,
374            }),
375            processor_drop_rx: None,
376            #[cfg(feature = "scheduled_events")]
377            shared_clock_output: RefCell::new(shared_clock_output),
378            sample_rate: NonZeroU32::new(44100).unwrap(),
379            sample_rate_recip: 44100.0f64.recip(),
380            stream_info: None,
381            shared_flags,
382            #[cfg(feature = "musical_transport")]
383            transport_state: Box::new(TransportState::default()),
384            #[cfg(feature = "musical_transport")]
385            transport_state_alloc_reuse: None,
386            event_group_pool,
387            event_group: Vec::with_capacity(initial_event_group_capacity),
388            initial_event_group_capacity,
389            #[cfg(feature = "scheduled_events")]
390            queued_clear_scheduled_events: Vec::new(),
391            config,
392        }
393    }
394
395    /// Try to modify the graph. If the given closure returns an error (or
396    /// if a cycle is detected), then any changes made to the graph inside
397    /// the closure will be reverted.
398    ///
399    /// Any custom error type can be used, though
400    /// [`ModifyGraphError`](crate::error::ModifyGraphError) is provided
401    /// for convenience.
402    pub fn try_modify_graph<E: Error>(
403        &mut self,
404        f: impl FnOnce(&mut Self) -> Result<(), E>,
405    ) -> Result<(), E> {
406        self.graph.begin_modify_guard();
407
408        let res = (f)(self);
409
410        self.graph.end_modify_guard(res.is_err());
411
412        res
413    }
414
415    /// Get an immutable reference to the processor store.
416    ///
417    /// If an audio stream is currently running, this will return `None`.
418    pub fn proc_store(&self) -> Option<&ProcStore> {
419        if let Some(proc_channel) = &self.pending_processor_channel {
420            Some(&proc_channel.store)
421        } else if let Some(processor) = self.processor_drop_rx.as_ref() {
422            if let Some(processor) = processor.last() {
423                if processor.poisoned {
424                    panic!("The audio thread has panicked!");
425                }
426
427                Some(&processor.extra.store)
428            } else {
429                None
430            }
431        } else {
432            None
433        }
434    }
435
436    /// Get a mutable reference to the processor store.
437    ///
438    /// If an audio stream is currently running, this will return `None`.
439    pub fn proc_store_mut(&mut self) -> Option<&mut ProcStore> {
440        if let Some(proc_channel) = &mut self.pending_processor_channel {
441            Some(&mut proc_channel.store)
442        } else if let Some(processor) = self.processor_drop_rx.as_mut() {
443            if let Some(processor) = processor.last_mut() {
444                if processor.poisoned {
445                    panic!("The audio thread has panicked!");
446                }
447
448                Some(&mut processor.extra.store)
449            } else {
450                None
451            }
452        } else {
453            None
454        }
455    }
456
457    /// Returns `true` if the context is currently active (the [`FirewheelProcessor`]
458    /// counterpart is still alive).
459    pub fn is_active(&self) -> bool {
460        if let Some(rx) = &self.processor_drop_rx {
461            rx.try_peek().is_none()
462        } else {
463            false
464        }
465    }
466
467    /// Activate the context with the given audio stream.
468    ///
469    /// Use [`FirewheelContext::is_active`] to check if the context is ready to
470    /// be activated.
471    ///
472    /// Note, in rare cases where the audio thread crashes without cleanly dropping
473    /// its contents, this may never succeed. Consider adding a timeout to avoid
474    /// deadlocking.
475    pub fn activate(&mut self, info: ActivateInfo) -> Result<FirewheelProcessor, ActivateError> {
476        let ActivateInfo {
477            sample_rate,
478            max_block_frames,
479            num_stream_in_channels,
480            num_stream_out_channels,
481            input_to_output_latency_seconds,
482        } = info;
483
484        if self.is_active() {
485            return Err(ActivateError::AlreadyActive);
486        }
487
488        let maybe_proc_channel = self.pending_processor_channel.take();
489
490        let prev_sample_rate = if maybe_proc_channel.is_some() {
491            sample_rate
492        } else {
493            self.sample_rate
494        };
495
496        let stream_info = StreamInfo {
497            sample_rate,
498            sample_rate_recip: (sample_rate.get() as f64).recip(),
499            prev_sample_rate,
500            max_block_frames,
501            num_stream_in_channels,
502            num_stream_out_channels,
503            input_to_output_latency_seconds,
504            declick_frames: NonZeroU32::new(
505                (self.config.declick_seconds * sample_rate.get() as f32).round() as u32,
506            )
507            .unwrap_or(NonZeroU32::MIN),
508        };
509
510        self.sample_rate = stream_info.sample_rate;
511        self.sample_rate_recip = stream_info.sample_rate_recip;
512
513        let schedule = self.graph.compile(&stream_info)?;
514
515        let (drop_tx, drop_rx) = ringbuf::HeapRb::<FirewheelProcessorInner>::new(1).split();
516
517        let processor = if let Some(proc_channel) = maybe_proc_channel {
518            FirewheelProcessorInner::new(
519                FirewheelProcessorConfig {
520                    flags: self.config.flags.into(),
521                    immediate_event_buffer_capacity: self.config.immediate_event_capacity,
522                    buffer_out_of_space_mode: self.config.buffer_out_of_space_mode,
523                    clamp_graph_inputs_below_amp: self
524                        .config
525                        .clamp_graph_inputs_below
526                        .map(|v| v.amp()),
527                    node_event_buffer_capacity: self.config.event_queue_capacity,
528                    #[cfg(feature = "scheduled_events")]
529                    scheduled_event_buffer_capacity: self.config.scheduled_event_capacity,
530                },
531                proc_channel,
532                &stream_info,
533            )
534        } else {
535            let mut processor = self.processor_drop_rx.as_mut().unwrap().try_pop().unwrap();
536
537            if processor.poisoned {
538                panic!("The audio thread has panicked!");
539            }
540
541            processor.new_stream(&stream_info);
542
543            processor
544        };
545
546        if self
547            .send_message_to_processor(ContextToProcessorMsg::NewSchedule(schedule))
548            .is_err()
549        {
550            panic!("Firewheel message channel is full!");
551        }
552
553        self.processor_drop_rx = Some(drop_rx);
554        self.stream_info = Some(stream_info);
555
556        let drop_flag = Arc::new(AtomicBool::new(false));
557        self.processor_drop_flag = Some(drop_flag.clone());
558
559        Ok(FirewheelProcessor::new(processor, drop_tx, drop_flag))
560    }
561
562    /// Request the context to be deactivated if it is active.
563    ///
564    /// This does not block the current thread. It might take a while for the
565    /// context to deactivate. Use [`FirewheelContext::is_active`] to check when
566    /// the context has been deactivated.
567    ///
568    /// Note, another way to deactivate the context is to drop the [`FirewheelProcessor`]
569    /// counterpart in the audio thread.
570    pub fn request_deactivate(&mut self) {
571        if let Some(flag) = &self.processor_drop_flag {
572            flag.store(true, Ordering::Relaxed);
573        }
574    }
575
576    /// If the context is active, request the context to be deactivated and wait
577    /// for the context to be deactivated before returning.
578    ///
579    /// If the `timeout` duration has been reached and the context is still not
580    /// deactivated, then an error is returned.
581    #[cfg(not(target_family = "wasm"))]
582    pub fn deactivate_blocking(&mut self, timeout: Duration) -> Result<(), DeactivateError> {
583        self.request_deactivate();
584
585        let now = bevy_platform::time::Instant::now();
586
587        while self.is_active() {
588            if now.elapsed() > timeout {
589                return Err(DeactivateError::TimedOut);
590            }
591
592            bevy_platform::thread::sleep(core::time::Duration::from_millis(1));
593        }
594
595        Ok(())
596    }
597
598    /// Information about the running audio stream.
599    ///
600    /// Returns `None` if the context is not currently active.
601    pub fn stream_info(&self) -> Option<&StreamInfo> {
602        if self.is_active() {
603            self.stream_info.as_ref()
604        } else {
605            None
606        }
607    }
608
609    /// Get the current time of the audio clock, without accounting for the delay
610    /// between when the clock was last updated and now.
611    ///
612    /// For most use cases you probably want to use [`FirewheelContext::audio_clock_corrected`]
613    /// instead, but this method is provided if needed.
614    ///
615    /// Note, due to the nature of audio processing, this clock is is *NOT* synced with
616    /// the system's time (`Instant::now`). (Instead it is based on the amount of data
617    /// that has been processed.) For applications where the timing of audio events is
618    /// critical (i.e. a rhythm game), sync the game to this audio clock instead of the
619    /// OS's clock (`Instant::now()`).
620    ///
621    /// Note, calling this method is not super cheap, so avoid calling it many
622    /// times within the same game loop iteration if possible.
623    #[cfg(feature = "scheduled_events")]
624    pub fn audio_clock(&self) -> AudioClock {
625        // Reading the latest value of the clock doesn't meaningfully mutate
626        // state, so treat it as an immutable operation with interior mutability.
627        //
628        // PANIC SAFETY: This struct is the only place this is ever borrowed, so this
629        // will never panic.
630        let mut clock_borrowed = self.shared_clock_output.borrow_mut();
631        let clock = clock_borrowed.read();
632
633        AudioClock {
634            samples: clock.clock_samples,
635            seconds: clock
636                .clock_samples
637                .to_seconds(self.sample_rate, self.sample_rate_recip),
638            #[cfg(feature = "musical_transport")]
639            musical: clock.current_playhead,
640            #[cfg(feature = "musical_transport")]
641            transport_is_playing: clock.transport_is_playing,
642            update_instant: self.is_active().then_some(clock.update_instant),
643        }
644    }
645
646    /// Get the current time of the audio clock.
647    ///
648    /// Unlike, [`FirewheelContext::audio_clock`], this method accounts for the delay
649    /// between when the audio clock was last updated and now, leading to a more
650    /// accurate result for games and other applications.
651    ///
652    /// If the delay could not be determined (i.e. an audio stream is not currently
653    /// running), then this will assume there was no delay between when the audio
654    /// clock was last updated and now.
655    ///
656    /// Note, due to the nature of audio processing, this clock is is *NOT* synced with
657    /// the system's time (`Instant::now`). (Instead it is based on the amount of data
658    /// that has been processed.) For applications where the timing of audio events is
659    /// critical (i.e. a rhythm game), sync the game to this audio clock instead of the
660    /// OS's clock (`Instant::now()`).
661    ///
662    /// Note, calling this method is not super cheap, so avoid calling it many
663    /// times within the same game loop iteration if possible.
664    #[cfg(feature = "scheduled_events")]
665    pub fn audio_clock_corrected(&self) -> AudioClock {
666        // Reading the latest value of the clock doesn't meaningfully mutate
667        // state, so treat it as an immutable operation with interior mutability.
668        //
669        // PANIC SAFETY: This struct is the only place this is ever borrowed, so this
670        // will never panic.
671        let mut clock_borrowed = self.shared_clock_output.borrow_mut();
672        let clock = clock_borrowed.read();
673
674        if !self.is_active() {
675            // The audio thread is not currently running, so just return the
676            // latest value of the clock.
677            return AudioClock {
678                samples: clock.clock_samples,
679                seconds: clock
680                    .clock_samples
681                    .to_seconds(self.sample_rate, self.sample_rate_recip),
682                #[cfg(feature = "musical_transport")]
683                musical: clock.current_playhead,
684                #[cfg(feature = "musical_transport")]
685                transport_is_playing: clock.transport_is_playing,
686                update_instant: None,
687            };
688        }
689
690        let update_instant = clock.update_instant;
691        let delay = update_instant.elapsed();
692
693        // Account for the delay between when the clock was last updated and now.
694        let delta_seconds = DurationSeconds(delay.as_secs_f64());
695
696        let samples = clock.clock_samples + delta_seconds.to_samples(self.sample_rate);
697
698        #[cfg(feature = "musical_transport")]
699        let musical = clock.current_playhead.map(|musical_time| {
700            if clock.transport_is_playing
701                && let Some(transport) = &self.transport_state.transport
702            {
703                transport.delta_seconds_from(musical_time, delta_seconds, clock.speed_multiplier)
704            } else {
705                musical_time
706            }
707        });
708
709        AudioClock {
710            samples,
711            seconds: samples.to_seconds(self.sample_rate, self.sample_rate_recip),
712            #[cfg(feature = "musical_transport")]
713            musical,
714            #[cfg(feature = "musical_transport")]
715            transport_is_playing: clock.transport_is_playing,
716            update_instant: Some(update_instant),
717        }
718    }
719
720    /// Get the instant the audio clock was last updated.
721    ///
722    /// If the audio thread is not currently running, or if the instant could not
723    /// be determined for any other reason, then this will return `None`.
724    ///
725    /// Note, calling this method is not super cheap, so avoid calling it many
726    /// times within the same game loop iteration if possible.
727    #[cfg(feature = "scheduled_events")]
728    pub fn audio_clock_instant(&self) -> Option<Instant> {
729        // Reading the latest value of the clock doesn't meaningfully mutate
730        // state, so treat it as an immutable operation with interior mutability.
731        //
732        // PANIC SAFETY: This struct is the only place this is ever borrowed, so this
733        // will never panic.
734        let mut clock_borrowed = self.shared_clock_output.borrow_mut();
735        let clock = clock_borrowed.read();
736
737        self.is_active().then_some(clock.update_instant)
738    }
739
740    /// Sync the state of the musical transport.
741    ///
742    /// If the message channel is full, then this will return an error.
743    #[cfg(feature = "musical_transport")]
744    pub fn sync_transport(&mut self, transport: &TransportState) -> Result<(), UpdateError> {
745        if &*self.transport_state != transport {
746            let transport_msg = if let Some(mut t) = self.transport_state_alloc_reuse.take() {
747                *t = transport.clone();
748                t
749            } else {
750                Box::new(transport.clone())
751            };
752
753            self.send_message_to_processor(ContextToProcessorMsg::SetTransportState(transport_msg))
754                .map_err(|(_, e)| e)?;
755
756            *self.transport_state = transport.clone();
757        }
758
759        Ok(())
760    }
761
762    /// Get the current transport state.
763    #[cfg(feature = "musical_transport")]
764    pub fn transport_state(&self) -> &TransportState {
765        &self.transport_state
766    }
767
768    /// Get the current transport state.
769    #[cfg(feature = "musical_transport")]
770    pub fn transport(&self) -> &TransportState {
771        &self.transport_state
772    }
773
774    /// The current configuration flags being used by this context.
775    pub fn flags(&self) -> &FirewheelFlags {
776        &self.config.flags
777    }
778
779    /// Set the configuration flags.
780    ///
781    /// This can be set while the context is active or inactive.
782    ///
783    /// If the message channel is full, then this will return an error.
784    pub fn set_flags(&mut self, flags: FirewheelFlags) -> Result<(), UpdateError> {
785        if self.config.flags == flags {
786            return Ok(());
787        }
788        self.config.flags = flags;
789
790        self.send_message_to_processor(ContextToProcessorMsg::SetFlags(flags.into()))
791            .map_err(|(_, e)| e)
792    }
793
794    /// Returns `true` if both the `FirewheelFlags::VALIDATE_OUTPUT_DOES_NOT_CLIP`
795    /// flag is set and a sample in the final output buffer fell outside the range
796    /// `[-1.0, 1.0]`.
797    ///
798    /// Calling this method resets the internal flag.
799    pub fn clipping_occurred(&self) -> bool {
800        self.shared_flags
801            .clipping_occurred
802            .swap(false, Ordering::Relaxed)
803    }
804
805    /// Retrieve the latest performance profiling data.
806    pub fn profiling_data(&mut self) -> &ProfilingData {
807        self.profiler_rx.fetch_info()
808    }
809
810    /// Update the firewheel context.
811    ///
812    /// This must be called regularly (i.e. once every frame).
813    pub fn update(&mut self) -> Result<(), UpdateError> {
814        self.logger_rx.flush(
815            |msg| {
816                #[cfg(feature = "tracing")]
817                tracing::error!("{}", msg);
818
819                #[cfg(all(feature = "log", not(feature = "tracing")))]
820                log::error!("{}", msg);
821
822                let _ = msg;
823            },
824            |msg| {
825                #[cfg(feature = "tracing")]
826                tracing::debug!("{}", msg);
827
828                #[cfg(all(feature = "log", not(feature = "tracing")))]
829                log::debug!("{}", msg);
830
831                let _ = msg;
832            },
833        );
834
835        firewheel_core::collector::GlobalRtGc::collect();
836
837        for msg in self.from_processor_rx.pop_iter() {
838            match msg {
839                ProcessorToContextMsg::DropEventGroup(mut event_group) => {
840                    event_group.clear();
841                    self.event_group_pool.push(event_group);
842                }
843                ProcessorToContextMsg::DropSchedule(schedule_data) => {
844                    self.graph.drop_old_schedule_data(schedule_data);
845                }
846                #[cfg(feature = "musical_transport")]
847                ProcessorToContextMsg::DropTransportState(transport_state) => {
848                    if self.transport_state_alloc_reuse.is_none() {
849                        self.transport_state_alloc_reuse = Some(transport_state);
850                    }
851                }
852                #[cfg(feature = "scheduled_events")]
853                ProcessorToContextMsg::DropClearScheduledEvents(msgs) => {
854                    let _ = msgs;
855                }
856            }
857        }
858
859        self.graph
860            .update(self.stream_info.as_ref(), &mut self.event_group);
861
862        if self.is_active() {
863            if self.graph.needs_compile() {
864                let schedule_data = self.graph.compile(self.stream_info.as_ref().unwrap())?;
865
866                if let Err((msg, e)) = self
867                    .send_message_to_processor(ContextToProcessorMsg::NewSchedule(schedule_data))
868                {
869                    let ContextToProcessorMsg::NewSchedule(schedule) = msg else {
870                        unreachable!();
871                    };
872
873                    self.graph.on_schedule_send_failed(schedule);
874
875                    return Err(e);
876                }
877            }
878
879            #[cfg(feature = "scheduled_events")]
880            if !self.queued_clear_scheduled_events.is_empty() {
881                let msgs: SmallVec<[ClearScheduledEventsEvent; 1]> =
882                    self.queued_clear_scheduled_events.drain(..).collect();
883
884                if let Err((msg, e)) = self
885                    .send_message_to_processor(ContextToProcessorMsg::ClearScheduledEvents(msgs))
886                {
887                    let ContextToProcessorMsg::ClearScheduledEvents(mut msgs) = msg else {
888                        unreachable!();
889                    };
890
891                    self.queued_clear_scheduled_events = msgs.drain(..).collect();
892
893                    return Err(e);
894                }
895            }
896
897            if !self.event_group.is_empty() {
898                let mut next_event_group = self
899                    .event_group_pool
900                    .pop()
901                    .unwrap_or_else(|| Vec::with_capacity(self.initial_event_group_capacity));
902                core::mem::swap(&mut next_event_group, &mut self.event_group);
903
904                if let Err((msg, e)) = self
905                    .send_message_to_processor(ContextToProcessorMsg::EventGroup(next_event_group))
906                {
907                    let ContextToProcessorMsg::EventGroup(mut event_group) = msg else {
908                        unreachable!();
909                    };
910
911                    core::mem::swap(&mut event_group, &mut self.event_group);
912                    self.event_group_pool.push(event_group);
913
914                    return Err(e);
915                }
916            }
917        } else {
918            self.stream_info = None;
919            self.graph.deactivate();
920        }
921
922        Ok(())
923    }
924
925    /// The ID of the graph input node
926    pub fn graph_in_node_id(&self) -> NodeID {
927        self.graph.graph_in_node()
928    }
929
930    /// The ID of the graph output node
931    pub fn graph_out_node_id(&self) -> NodeID {
932        self.graph.graph_out_node()
933    }
934
935    /// Add a node to the audio graph.
936    pub fn add_node<T: AudioNode + 'static>(
937        &mut self,
938        node: T,
939        config: Option<T::Configuration>,
940    ) -> Result<NodeID, NodeError> {
941        self.graph.add_node(node, config)
942    }
943
944    /// Add a node to the audio graph which implements the type-erased [`DynAudioNode`] trait.
945    pub fn add_dyn_node<T: DynAudioNode + 'static>(
946        &mut self,
947        node: T,
948    ) -> Result<NodeID, NodeError> {
949        self.graph.add_dyn_node(node)
950    }
951
952    /// Add a node to the audio graph with the given bypass state.
953    pub fn add_node_bypassed<T: AudioNode + 'static>(
954        &mut self,
955        node: T,
956        config: Option<T::Configuration>,
957        bypassed: bool,
958    ) -> Result<NodeID, NodeError> {
959        let node_id = self.add_node(node, config)?;
960        if bypassed {
961            self.queue_event_for(node_id, NodeEventType::SetBypassed(true));
962        }
963        Ok(node_id)
964    }
965
966    /// Add a node with the given bypass state to the audio graph which implements
967    /// the type-erased [`DynAudioNode`] trait.
968    pub fn add_dyn_node_bypassed<T: DynAudioNode + 'static>(
969        &mut self,
970        node: T,
971        bypassed: bool,
972    ) -> Result<NodeID, NodeError> {
973        let node_id = self.graph.add_dyn_node(node)?;
974        if bypassed {
975            self.queue_event_for(node_id, NodeEventType::SetBypassed(true));
976        }
977        Ok(node_id)
978    }
979
980    /// Remove the given node from the audio graph.
981    ///
982    /// This will automatically remove all edges from the graph that
983    /// were connected to this node.
984    ///
985    /// On success, this returns a list of all edges that were removed
986    /// from the graph as a result of removing this node.
987    ///
988    /// This will return an error if the ID is of the graph input or graph
989    /// output node.
990    pub fn remove_node(&mut self, node_id: NodeID) -> Result<SmallVec<[Edge; 4]>, RemoveNodeError> {
991        self.graph.remove_node(node_id, false)
992    }
993
994    /// Returns `true` if the node exists in the graph.
995    pub fn contains_node(&self, id: NodeID) -> bool {
996        self.graph.contains_node(id)
997    }
998
999    /// Get information about a node in the graph.
1000    ///
1001    /// If the node does not exist in the graph, then `None` will be returned.
1002    pub fn node_info(&self, id: NodeID) -> Option<&NodeEntry> {
1003        self.graph.node_info(id)
1004    }
1005
1006    /// Get the [`ChannelConfig`] of a node in the graph.
1007    ///
1008    /// If the node does not exist in the graph, then `None` will be returned.
1009    pub fn node_channel_config(&self, id: NodeID) -> Option<ChannelConfig> {
1010        self.graph.node_info(id).map(|n| n.info.channel_config)
1011    }
1012
1013    /// Get an immutable reference to the custom state of a node.
1014    ///
1015    /// If the node does not exist in the graph, then `None` will be returned.
1016    pub fn node_state<T: 'static>(&self, id: NodeID) -> Option<&T> {
1017        self.graph.node_state(id)
1018    }
1019
1020    /// Get a type-erased, immutable reference to the custom state of a node.
1021    ///
1022    /// If the node does not exist in the graph, then `None` will be returned.
1023    pub fn node_state_dyn(&self, id: NodeID) -> Option<&dyn Any> {
1024        self.graph.node_state_dyn(id)
1025    }
1026
1027    /// Get a mutable reference to the custom state of a node.
1028    ///
1029    /// If the node does not exist in the graph, then `None` will be returned.
1030    pub fn node_state_mut<T: 'static>(&mut self, id: NodeID) -> Option<&mut T> {
1031        self.graph.node_state_mut(id)
1032    }
1033
1034    /// Get a type-erased, mutable reference to the custom state of a node.
1035    ///
1036    /// If the node does not exist in the graph, then `None` will be returned.
1037    pub fn node_state_dyn_mut(&mut self, id: NodeID) -> Option<&mut dyn Any> {
1038        self.graph.node_state_dyn_mut(id)
1039    }
1040
1041    /// Get a list of all the existing nodes in the graph.
1042    pub fn nodes(&self) -> impl Iterator<Item = &NodeEntry> {
1043        self.graph.nodes()
1044    }
1045
1046    /// Get a list of all the existing edges in the graph.
1047    pub fn edges(&self) -> impl Iterator<Item = &Edge> {
1048        self.graph.edges()
1049    }
1050
1051    /// Set the number of input and output channels to and from the audio graph.
1052    ///
1053    /// Returns the list of edges that were removed.
1054    pub fn set_graph_channel_config(
1055        &mut self,
1056        channel_config: ChannelConfig,
1057    ) -> SmallVec<[Edge; 4]> {
1058        self.graph.set_graph_channel_config(channel_config, false)
1059    }
1060
1061    /// Add connections (edges) between two nodes to the graph.
1062    ///
1063    /// * `src_node` - The ID of the source node.
1064    /// * `dst_node` - The ID of the destination node.
1065    /// * `ports_src_dst` - The port indices for each connection to make,
1066    ///   where the first value in a tuple is the output port on `src_node`,
1067    ///   and the second value in that tuple is the input port on `dst_node`.
1068    /// * `check_for_cycles` - If `true`, then this will run a check to
1069    ///   see if adding these edges will create a cycle in the graph, and
1070    ///   return an error if it does. Note, checking for cycles can be quite
1071    ///   expensive, so avoid enabling this when calling this method many times
1072    ///   in a row.
1073    ///
1074    /// If successful, then this returns a list of edge IDs in order.
1075    ///
1076    /// If this returns an error, then the audio graph has not been
1077    /// modified.
1078    pub fn connect(
1079        &mut self,
1080        src_node: NodeID,
1081        dst_node: NodeID,
1082        ports_src_dst: &[(PortIdx, PortIdx)],
1083        check_for_cycles: bool,
1084    ) -> Result<SmallVec<[EdgeID; 4]>, AddEdgeError> {
1085        self.graph
1086            .connect(src_node, dst_node, ports_src_dst, check_for_cycles, false)
1087    }
1088
1089    /// Connect two nodes in the graph, connecting output port 0 to input port
1090    /// 0, output port 1 to input port 1, etc.
1091    ///
1092    /// If the number of output ports on `src_node` does not equal the number
1093    /// of input ports on `dst_node`, then only the first valid ports will be
1094    /// connected.
1095    ///
1096    /// If successful, then this returns a list of edge IDs in order.
1097    ///
1098    /// If this returns an error, then the audio graph has not been
1099    /// modified.
1100    pub fn auto_connect(
1101        &mut self,
1102        src_node: NodeID,
1103        dst_node: NodeID,
1104        check_for_cycles: bool,
1105    ) -> Result<SmallVec<[EdgeID; 4]>, AddEdgeError> {
1106        let num_src_out_ports = self
1107            .node_info(src_node)
1108            .ok_or(AddEdgeError::SrcNodeNotFound(src_node))?
1109            .info
1110            .channel_config
1111            .num_outputs
1112            .get();
1113        let num_dst_in_ports = self
1114            .node_info(dst_node)
1115            .ok_or(AddEdgeError::DstNodeNotFound(dst_node))?
1116            .info
1117            .channel_config
1118            .num_inputs
1119            .get();
1120        let num_connect_ports = num_src_out_ports.min(num_dst_in_ports);
1121
1122        let ports_src_dst: SmallVec<[(u32, u32); 4]> =
1123            (0..num_connect_ports).map(|i| (i, i)).collect();
1124
1125        self.graph
1126            .connect(src_node, dst_node, &ports_src_dst, check_for_cycles, false)
1127    }
1128
1129    /// Connect the first two output ports of a node to the first two input
1130    /// ports of a second node.
1131    ///
1132    /// * `src_node` - The ID of the source node.
1133    /// * `dst_node` - The ID of the destination node.
1134    /// * `check_for_cycles` - If `true`, then this will run a check to
1135    ///   see if adding these edges will create a cycle in the graph, and
1136    ///   return an error if it does. Note, checking for cycles can be quite
1137    ///   expensive, so avoid enabling this when calling this method many times
1138    ///   in a row.
1139    ///
1140    /// ## Behavior
1141    ///
1142    /// * If `num_out_ports_on_src_node >= 2 && num_in_ports_on_dst_node >= 2`,
1143    ///   then src port 0 will be connected to dst port 0, and src port 1 will be
1144    ///   connected to dst port 1.
1145    /// * If `num_out_ports_on_src_node == 1 && num_in_ports_on_dst_node >= 2`,
1146    ///   then src port 0 will be connected to both dst port 0 and dst port 1.
1147    /// * In all other cases, an error will be returned. (Note that converting
1148    ///   a stereo signal into a mono signal should be done with the
1149    ///   `StereoToMonoNode`.)
1150    ///
1151    /// If successful, then this returns a list of edge IDs in order.
1152    ///
1153    /// If this returns an error, then the audio graph has not been
1154    /// modified.
1155    pub fn connect_stereo(
1156        &mut self,
1157        src_node: NodeID,
1158        dst_node: NodeID,
1159        check_for_cycles: bool,
1160    ) -> Result<SmallVec<[EdgeID; 4]>, AddEdgeError> {
1161        let num_src_out_ports = self
1162            .node_info(src_node)
1163            .ok_or(AddEdgeError::SrcNodeNotFound(src_node))?
1164            .info
1165            .channel_config
1166            .num_outputs;
1167        let num_dst_in_ports = self
1168            .node_info(dst_node)
1169            .ok_or(AddEdgeError::DstNodeNotFound(dst_node))?
1170            .info
1171            .channel_config
1172            .num_inputs;
1173
1174        let ports_src_dst = if num_src_out_ports.get() >= 2 && num_dst_in_ports.get() >= 2 {
1175            &[(0, 0), (1, 1)]
1176        } else if num_src_out_ports.get() == 1 && num_dst_in_ports.get() >= 2 {
1177            &[(0, 0), (0, 1)]
1178        } else {
1179            return Err(if num_dst_in_ports.get() < 2 {
1180                AddEdgeError::InPortOutOfRange {
1181                    node: dst_node,
1182                    port_idx: 1,
1183                    num_in_ports: num_dst_in_ports,
1184                }
1185            } else {
1186                AddEdgeError::InPortOutOfRange {
1187                    node: src_node,
1188                    port_idx: 0,
1189                    num_in_ports: num_src_out_ports,
1190                }
1191            });
1192        };
1193
1194        self.graph
1195            .connect(src_node, dst_node, ports_src_dst, check_for_cycles, false)
1196    }
1197
1198    /// Remove connections (edges) between two nodes from the graph.
1199    ///
1200    /// * `src_node` - The ID of the source node.
1201    /// * `dst_node` - The ID of the destination node.
1202    /// * `ports_src_dst` - The port indices for each connection to make,
1203    ///   where the first value in a tuple is the output port on `src_node`,
1204    ///   and the second value in that tuple is the input port on `dst_node`.
1205    ///
1206    /// Returns the list of edges that were successfully removed.
1207    pub fn disconnect(
1208        &mut self,
1209        src_node: NodeID,
1210        dst_node: NodeID,
1211        ports_src_dst: &[(PortIdx, PortIdx)],
1212    ) -> SmallVec<[Edge; 4]> {
1213        self.graph.disconnect(src_node, dst_node, ports_src_dst)
1214    }
1215
1216    /// Remove all connections (edges) between two nodes in the graph.
1217    ///
1218    /// * `src_node` - The ID of the source node.
1219    /// * `dst_node` - The ID of the destination node.
1220    ///
1221    /// Returns the list of edges that were successfully removed.
1222    pub fn disconnect_all_between(
1223        &mut self,
1224        src_node: NodeID,
1225        dst_node: NodeID,
1226    ) -> SmallVec<[Edge; 4]> {
1227        self.graph.disconnect_all_between(src_node, dst_node)
1228    }
1229
1230    /// Remove a connection (edge) via the edge's unique ID.
1231    ///
1232    /// If the edge did not exist in this graph, then `None` will be returned.
1233    pub fn disconnect_by_edge_id(&mut self, edge_id: EdgeID) -> Option<Edge> {
1234        self.graph.disconnect_by_edge_id(edge_id, false)
1235    }
1236
1237    /// Get information about the given [Edge]
1238    pub fn edge(&self, edge_id: EdgeID) -> Option<&Edge> {
1239        self.graph.edge(edge_id)
1240    }
1241
1242    /// Runs a check to see if a cycle exists in the audio graph. If a cycle
1243    /// exists, an error is returned.
1244    ///
1245    /// Note, this method is expensive.
1246    pub fn cycle_detected(&mut self) -> Result<(), CompileGraphError> {
1247        if self.graph.cycle_detected() {
1248            Err(CompileGraphError::CycleDetected)
1249        } else {
1250            Ok(())
1251        }
1252    }
1253
1254    /// Queue an event to be sent to an audio node's processor.
1255    ///
1256    /// Note, this event will not be sent until the event queue is flushed
1257    /// in [`FirewheelContext::update`].
1258    pub fn queue_event(&mut self, event: NodeEvent) {
1259        if self.contains_node(event.node_id) {
1260            self.event_group.push(event);
1261        }
1262    }
1263
1264    /// Queue an event to be sent to an audio node's processor.
1265    ///
1266    /// Note, this event will not be sent until the event queue is flushed
1267    /// in [`FirewheelContext::update`].
1268    pub fn queue_event_for(&mut self, node_id: NodeID, event: NodeEventType) {
1269        self.queue_event(NodeEvent {
1270            node_id,
1271            #[cfg(feature = "scheduled_events")]
1272            time: None,
1273            event,
1274        });
1275    }
1276
1277    /// Queue a [`NodeEventType::SetBypassed`] event for the given node.
1278    pub fn queue_bypassed_for(&mut self, node_id: NodeID, bypassed: bool) {
1279        self.queue_event(NodeEvent {
1280            node_id,
1281            #[cfg(feature = "scheduled_events")]
1282            time: None,
1283            event: NodeEventType::SetBypassed(bypassed),
1284        });
1285    }
1286
1287    /// Queue an event at a certain time, to be sent to an audio node's processor.
1288    ///
1289    /// If `time` is `None`, then the event will occur as soon as the node's
1290    /// processor receives the event.
1291    ///
1292    /// Note, this event will not be sent until the event queue is flushed
1293    /// in [`FirewheelContext::update`].
1294    #[cfg(feature = "scheduled_events")]
1295    pub fn schedule_event_for(
1296        &mut self,
1297        node_id: NodeID,
1298        event: NodeEventType,
1299        time: Option<EventInstant>,
1300    ) {
1301        self.queue_event(NodeEvent {
1302            node_id,
1303            time,
1304            event,
1305        });
1306    }
1307
1308    /// Construct a [`ContextQueue`] for diffing.
1309    ///
1310    /// Returns `None` if the node does not exist in the graph.
1311    pub fn event_queue(&mut self, id: NodeID) -> ContextQueue<'_> {
1312        ContextQueue {
1313            context: self,
1314            id,
1315            #[cfg(feature = "scheduled_events")]
1316            time: None,
1317        }
1318    }
1319
1320    /// Construct a [`ContextQueue`] for diffing, along with the event instant that
1321    /// any pushed events will be scheduled for.
1322    ///
1323    /// Returns `None` if the node does not exist in the graph.
1324    #[cfg(feature = "scheduled_events")]
1325    pub fn event_queue_scheduled(
1326        &mut self,
1327        id: NodeID,
1328        time: Option<EventInstant>,
1329    ) -> ContextQueue<'_> {
1330        ContextQueue {
1331            context: self,
1332            id,
1333            time,
1334        }
1335    }
1336
1337    /// Cancel scheduled events for all nodes.
1338    ///
1339    /// This will clear all events that have been scheduled since the last call to
1340    /// [`FirewheelContext::update`]. Any events scheduled between then and the next call
1341    /// to [`FirewheelContext::update`] will not be canceled.
1342    ///
1343    /// This only takes effect once [`FirewheelContext::update`] is called.
1344    #[cfg(feature = "scheduled_events")]
1345    pub fn cancel_all_scheduled_events(&mut self, event_type: ClearScheduledEventsType) {
1346        self.queued_clear_scheduled_events
1347            .push(ClearScheduledEventsEvent {
1348                node_id: None,
1349                event_type,
1350            });
1351    }
1352
1353    /// Cancel scheduled events for a specific node.
1354    ///
1355    /// This will clear all events that have been scheduled since the last call to
1356    /// [`FirewheelContext::update`]. Any events scheduled between then and the next call
1357    /// to [`FirewheelContext::update`] will not be canceled.
1358    ///
1359    /// This only takes effect once [`FirewheelContext::update`] is called.
1360    #[cfg(feature = "scheduled_events")]
1361    pub fn cancel_scheduled_events_for(
1362        &mut self,
1363        node_id: NodeID,
1364        event_type: ClearScheduledEventsType,
1365    ) {
1366        self.queued_clear_scheduled_events
1367            .push(ClearScheduledEventsEvent {
1368                node_id: Some(node_id),
1369                event_type,
1370            });
1371    }
1372
1373    fn send_message_to_processor(
1374        &mut self,
1375        msg: ContextToProcessorMsg,
1376    ) -> Result<(), (ContextToProcessorMsg, UpdateError)> {
1377        self.to_processor_tx
1378            .try_push(msg)
1379            .map_err(|msg| (msg, UpdateError::MsgChannelFull))
1380    }
1381}
1382
1383impl Drop for FirewheelContext {
1384    fn drop(&mut self) {
1385        // Wait for the processor to be drop to avoid deallocating it on
1386        // the audio thread.
1387        #[cfg(not(target_family = "wasm"))]
1388        let _ = self.deactivate_blocking(core::time::Duration::from_secs(3));
1389
1390        #[cfg(target_family = "wasm")]
1391        self.request_deactivate();
1392
1393        // Make sure all node processors are dropped before node states in
1394        // order to be compatible with CLAP plugin hosting.
1395        if let Some(p) = &mut self.processor_drop_rx {
1396            p.clear();
1397        }
1398        self.from_processor_rx.clear();
1399        firewheel_core::collector::GlobalRtGc::collect();
1400    }
1401}
1402
1403/// An event queue acquired from [`FirewheelContext::event_queue`].
1404///
1405/// This can help reduce event queue allocations
1406/// when you have direct access to the context.
1407///
1408/// ```
1409/// # use firewheel_core::{diff::{Diff, PathBuilder}, node::NodeID};
1410/// # use firewheel_graph::{backend::AudioBackend, FirewheelContext, ContextQueue};
1411/// # fn context_queue<B: AudioBackend, D: Diff>(
1412/// #     context: &mut FirewheelContext,
1413/// #     node_id: NodeID,
1414/// #     params: &D,
1415/// #     baseline: &D,
1416/// # ) {
1417/// // Get a queue that will send events directly to the provided node.
1418/// let mut queue = context.event_queue(node_id);
1419/// // Perform diffing using this queue.
1420/// params.diff(baseline, PathBuilder::default(), &mut queue);
1421/// # }
1422/// ```
1423pub struct ContextQueue<'a> {
1424    pub context: &'a mut FirewheelContext,
1425    id: NodeID,
1426    #[cfg(feature = "scheduled_events")]
1427    time: Option<EventInstant>,
1428}
1429
1430impl ContextQueue<'_> {
1431    /// Send an event to set the bypass state of the node.
1432    pub fn push_bypassed(&mut self, bypassed: bool) {
1433        self.push(NodeEventType::SetBypassed(bypassed));
1434    }
1435}
1436
1437#[cfg(feature = "scheduled_events")]
1438impl ContextQueue<'_> {
1439    pub fn time(&self) -> Option<EventInstant> {
1440        self.time
1441    }
1442}
1443
1444impl EventQueue for ContextQueue<'_> {
1445    fn push(&mut self, data: NodeEventType) {
1446        self.context.queue_event(NodeEvent {
1447            event: data,
1448            #[cfg(feature = "scheduled_events")]
1449            time: self.time,
1450            node_id: self.id,
1451        });
1452    }
1453}
1454
1455/// The type of scheduled events to clear
1456#[cfg(feature = "scheduled_events")]
1457#[derive(Default, Debug, Clone, Copy, PartialEq)]
1458pub enum ClearScheduledEventsType {
1459    /// Clear both musical and non-musical scheduled events.
1460    #[default]
1461    All,
1462    /// Clear only non-musical scheduled events.
1463    NonMusicalOnly,
1464    /// Clear only musical scheduled events.
1465    MusicalOnly,
1466}