firewheel_graph/
processor.rs

1use audioadapter::{Adapter, AdapterMut};
2use bevy_platform::sync::{
3    Arc,
4    atomic::{AtomicBool, Ordering},
5};
6use core::num::NonZeroU32;
7use ringbuf::traits::Producer;
8use thunderdome::Arena;
9
10#[cfg(not(feature = "std"))]
11use bevy_platform::prelude::{Box, Vec};
12
13use bevy_platform::time::Instant;
14
15use firewheel_core::{
16    StreamInfo,
17    clock::InstantSamples,
18    dsp::{
19        buffer::ConstSequentialBuffer,
20        declick::{DeclickValues, Declicker},
21    },
22    event::{NodeEvent, ProcEventsIndex},
23    node::{AudioNodeProcessor, ProcExtra},
24};
25
26use crate::{
27    backend::BackendProcessInfo,
28    context::{FirewheelBitFlags, ProcessorChannel},
29    graph::ScheduleHeapData,
30    processor::{
31        event_scheduler::{EventScheduler, NodeEventSchedulerData},
32        profiling::ProfilerTx,
33    },
34};
35
36pub use profiling::ProfilingData;
37
38#[cfg(feature = "scheduled_events")]
39use crate::context::ClearScheduledEventsType;
40#[cfg(feature = "scheduled_events")]
41use firewheel_core::node::NodeID;
42#[cfg(feature = "scheduled_events")]
43use smallvec::SmallVec;
44
45#[cfg(feature = "musical_transport")]
46use firewheel_core::clock::{InstantMusical, TransportState};
47
48mod event_scheduler;
49mod handle_messages;
50mod process;
51pub(crate) mod profiling;
52
53#[cfg(feature = "musical_transport")]
54mod transport;
55#[cfg(feature = "musical_transport")]
56use transport::ProcTransportState;
57
58pub struct FirewheelProcessor {
59    inner: Option<FirewheelProcessorInner>,
60    drop_tx: ringbuf::HeapProd<FirewheelProcessorInner>,
61    drop_flag: Arc<AtomicBool>,
62}
63
64impl Drop for FirewheelProcessor {
65    fn drop(&mut self) {
66        self.drop_inner();
67    }
68}
69
70impl FirewheelProcessor {
71    pub(crate) fn new(
72        processor: FirewheelProcessorInner,
73        drop_tx: ringbuf::HeapProd<FirewheelProcessorInner>,
74        drop_flag: Arc<AtomicBool>,
75    ) -> Self {
76        Self {
77            inner: Some(processor),
78            drop_tx,
79            drop_flag,
80        }
81    }
82
83    pub fn process(
84        &mut self,
85        input: &dyn Adapter<f32>,
86        output: &mut dyn AdapterMut<f32>,
87        info: BackendProcessInfo,
88    ) {
89        self.poll_drop_flag();
90
91        if let Some(inner) = &mut self.inner {
92            inner.process(input, output, info);
93        } else {
94            output.fill_frames_with(0, info.frames, &0.0);
95        }
96    }
97
98    fn poll_drop_flag(&mut self) {
99        if self.inner.is_some() && self.drop_flag.load(Ordering::Relaxed) {
100            self.drop_inner();
101        }
102    }
103
104    fn drop_inner(&mut self) {
105        let Some(mut inner) = self.inner.take() else {
106            return;
107        };
108
109        inner.stream_stopped();
110
111        // TODO: Remove this feature gate if `bevy_platform` implements this.
112        #[cfg(feature = "std")]
113        if std::thread::panicking() {
114            inner.poisoned = true;
115        }
116
117        let _ = self.drop_tx.try_push(inner);
118    }
119}
120
121pub(crate) struct FirewheelProcessorInner {
122    nodes: Arena<NodeEntry>,
123    schedule_data: Option<Box<ScheduleHeapData>>,
124
125    from_graph_rx: ringbuf::HeapCons<ContextToProcessorMsg>,
126    to_graph_tx: ringbuf::HeapProd<ProcessorToContextMsg>,
127
128    event_scheduler: EventScheduler,
129    proc_event_queue: Vec<ProcEventsIndex>,
130
131    sample_rate: NonZeroU32,
132    sample_rate_recip: f64,
133    max_block_frames: usize,
134
135    clock_samples: InstantSamples,
136    shared_clock_input: triple_buffer::Input<SharedClock>,
137    profiler_tx: ProfilerTx,
138
139    #[cfg(feature = "musical_transport")]
140    proc_transport_state: ProcTransportState,
141
142    flags: FirewheelBitFlags,
143    shared_flags: Arc<SharedFlags>,
144    clamp_graph_inputs_below_amp: Option<f32>,
145
146    last_input_overflow_log_instant: Option<Instant>,
147    last_output_underflow_log_instant: Option<Instant>,
148
149    pub(crate) extra: ProcExtra,
150
151    /// If a panic occurs while processing, this flag is set to let the
152    /// main thread know that it shouldn't try spawning a new audio stream
153    /// with the shared `Arc<AtomicRefCell<FirewheelProcessorInner>>` object.
154    pub(crate) poisoned: bool,
155}
156
157pub(crate) struct FirewheelProcessorConfig {
158    pub flags: FirewheelBitFlags,
159    pub immediate_event_buffer_capacity: usize,
160    pub buffer_out_of_space_mode: BufferOutOfSpaceMode,
161    pub clamp_graph_inputs_below_amp: Option<f32>,
162    pub node_event_buffer_capacity: usize,
163    #[cfg(feature = "scheduled_events")]
164    pub scheduled_event_buffer_capacity: usize,
165}
166
167impl FirewheelProcessorInner {
168    /// Note, this method gets called on the main thread, not the audio thread.
169    pub(crate) fn new(
170        config: FirewheelProcessorConfig,
171        proc_channel: ProcessorChannel,
172        stream_info: &StreamInfo,
173    ) -> Self {
174        let FirewheelProcessorConfig {
175            flags,
176            immediate_event_buffer_capacity,
177            buffer_out_of_space_mode,
178            clamp_graph_inputs_below_amp,
179            node_event_buffer_capacity,
180            #[cfg(feature = "scheduled_events")]
181            scheduled_event_buffer_capacity,
182        } = config;
183
184        let ProcessorChannel {
185            shared_flags,
186            from_context_rx,
187            to_context_tx,
188            logger,
189            store,
190            profiler_tx,
191            shared_clock_input,
192        } = proc_channel;
193
194        Self {
195            nodes: Arena::new(),
196            schedule_data: None,
197            from_graph_rx: from_context_rx,
198            to_graph_tx: to_context_tx,
199            event_scheduler: EventScheduler::new(
200                immediate_event_buffer_capacity,
201                #[cfg(feature = "scheduled_events")]
202                scheduled_event_buffer_capacity,
203                buffer_out_of_space_mode,
204            ),
205            proc_event_queue: Vec::with_capacity(node_event_buffer_capacity),
206            sample_rate: stream_info.sample_rate,
207            sample_rate_recip: stream_info.sample_rate_recip,
208            max_block_frames: stream_info.max_block_frames.get() as usize,
209            clock_samples: InstantSamples(0),
210            shared_clock_input,
211            profiler_tx,
212            #[cfg(feature = "musical_transport")]
213            proc_transport_state: ProcTransportState::new(),
214            flags,
215            shared_flags,
216            clamp_graph_inputs_below_amp,
217            last_input_overflow_log_instant: None,
218            last_output_underflow_log_instant: None,
219            extra: ProcExtra {
220                scratch_buffers: ConstSequentialBuffer::new(
221                    stream_info.max_block_frames.get() as usize
222                ),
223                declick_values: DeclickValues::new(stream_info.declick_frames),
224                logger,
225                store,
226            },
227            poisoned: false,
228        }
229    }
230}
231
232pub(crate) struct NodeEntry {
233    pub processor: Box<dyn AudioNodeProcessor>,
234    pub bypass_declick: Declicker,
235    pub is_bypassed: bool,
236    pub is_first_process: bool,
237    pub in_place_buffers: bool,
238
239    event_data: NodeEventSchedulerData,
240}
241
242pub(crate) enum ContextToProcessorMsg {
243    EventGroup(Vec<NodeEvent>),
244    NewSchedule(Box<ScheduleHeapData>),
245    SetFlags(FirewheelBitFlags),
246    #[cfg(feature = "musical_transport")]
247    SetTransportState(Box<TransportState>),
248    #[cfg(feature = "scheduled_events")]
249    ClearScheduledEvents(SmallVec<[ClearScheduledEventsEvent; 1]>),
250}
251
252#[allow(clippy::enum_variant_names)]
253pub(crate) enum ProcessorToContextMsg {
254    DropEventGroup(Vec<NodeEvent>),
255    DropSchedule(Box<ScheduleHeapData>),
256    #[cfg(feature = "musical_transport")]
257    DropTransportState(Box<TransportState>),
258    #[cfg(feature = "scheduled_events")]
259    DropClearScheduledEvents(SmallVec<[ClearScheduledEventsEvent; 1]>),
260}
261
262#[cfg(feature = "scheduled_events")]
263pub(crate) struct ClearScheduledEventsEvent {
264    /// If `None`, then clear events for all nodes.
265    pub node_id: Option<NodeID>,
266    pub event_type: ClearScheduledEventsType,
267}
268
269#[derive(Clone)]
270pub(crate) struct SharedClock {
271    pub clock_samples: InstantSamples,
272    #[cfg(feature = "scheduled_events")]
273    pub update_instant: Instant,
274    #[cfg(feature = "musical_transport")]
275    pub current_playhead: Option<InstantMusical>,
276    #[cfg(feature = "musical_transport")]
277    pub speed_multiplier: f64,
278    #[cfg(feature = "musical_transport")]
279    pub transport_is_playing: bool,
280}
281
282impl Default for SharedClock {
283    fn default() -> Self {
284        Self {
285            clock_samples: InstantSamples(0),
286            #[cfg(feature = "scheduled_events")]
287            update_instant: Instant::now(),
288            #[cfg(feature = "musical_transport")]
289            current_playhead: None,
290            #[cfg(feature = "musical_transport")]
291            speed_multiplier: 1.0,
292            #[cfg(feature = "musical_transport")]
293            transport_is_playing: false,
294        }
295    }
296}
297
298/// How to handle event buffers on the audio thread running out of space.
299#[derive(Default, Debug, Clone, Copy, PartialEq, PartialOrd)]
300#[cfg_attr(feature = "bevy_reflect", derive(bevy_reflect::Reflect))]
301#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
302pub enum BufferOutOfSpaceMode {
303    #[default]
304    /// If an event buffer on the audio thread ran out of space to fit new
305    /// events, reallocate on the audio thread to fit the new items. If this
306    /// happens, it may cause underruns (audio glitches), and a warning will
307    /// be logged.
308    AllocateOnAudioThread,
309    /// If an event buffer on the audio thread ran out of space to fit new
310    /// events, then panic.
311    Panic,
312    /// If an event buffer on the audio thread ran out of space to fit new
313    /// events, drop those events to avoid allocating on the audio thread.
314    /// If this happens, a warning will be logged.
315    ///
316    /// (Not generally recommended, but the option is here if you want it.)
317    DropEvents,
318}
319
320#[derive(Default)]
321pub(crate) struct SharedFlags {
322    pub clipping_occurred: AtomicBool,
323}
firewheel_graph/processor.rs

firewheel_graph/
processor.rs
