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    #[cfg(feature = "scheduled_events")]
137    shared_clock_input: triple_buffer::Input<SharedClock>,
138    profiler_tx: ProfilerTx,
139
140    #[cfg(feature = "musical_transport")]
141    proc_transport_state: ProcTransportState,
142
143    flags: FirewheelBitFlags,
144    shared_flags: Arc<SharedFlags>,
145    clamp_graph_inputs_below_amp: Option<f32>,
146
147    last_input_overflow_log_instant: Option<Instant>,
148    last_output_underflow_log_instant: Option<Instant>,
149
150    pub(crate) extra: ProcExtra,
151
152    /// If a panic occurs while processing, this flag is set to let the
153    /// main thread know that it shouldn't try spawning a new audio stream
154    /// with the shared `Arc<AtomicRefCell<FirewheelProcessorInner>>` object.
155    pub(crate) poisoned: bool,
156}
157
158pub(crate) struct FirewheelProcessorConfig {
159    pub flags: FirewheelBitFlags,
160    pub immediate_event_buffer_capacity: usize,
161    pub buffer_out_of_space_mode: BufferOutOfSpaceMode,
162    pub clamp_graph_inputs_below_amp: Option<f32>,
163    pub node_event_buffer_capacity: usize,
164    #[cfg(feature = "scheduled_events")]
165    pub scheduled_event_buffer_capacity: usize,
166}
167
168impl FirewheelProcessorInner {
169    /// Note, this method gets called on the main thread, not the audio thread.
170    pub(crate) fn new(
171        config: FirewheelProcessorConfig,
172        proc_channel: ProcessorChannel,
173        stream_info: &StreamInfo,
174    ) -> Self {
175        let FirewheelProcessorConfig {
176            flags,
177            immediate_event_buffer_capacity,
178            buffer_out_of_space_mode,
179            clamp_graph_inputs_below_amp,
180            node_event_buffer_capacity,
181            #[cfg(feature = "scheduled_events")]
182            scheduled_event_buffer_capacity,
183        } = config;
184
185        let ProcessorChannel {
186            shared_flags,
187            from_context_rx,
188            to_context_tx,
189            logger,
190            store,
191            profiler_tx,
192            #[cfg(feature = "scheduled_events")]
193            shared_clock_input,
194        } = proc_channel;
195
196        Self {
197            nodes: Arena::new(),
198            schedule_data: None,
199            from_graph_rx: from_context_rx,
200            to_graph_tx: to_context_tx,
201            event_scheduler: EventScheduler::new(
202                immediate_event_buffer_capacity,
203                #[cfg(feature = "scheduled_events")]
204                scheduled_event_buffer_capacity,
205                buffer_out_of_space_mode,
206            ),
207            proc_event_queue: Vec::with_capacity(node_event_buffer_capacity),
208            sample_rate: stream_info.sample_rate,
209            sample_rate_recip: stream_info.sample_rate_recip,
210            max_block_frames: stream_info.max_block_frames.get() as usize,
211            clock_samples: InstantSamples(0),
212            #[cfg(feature = "scheduled_events")]
213            shared_clock_input,
214            profiler_tx,
215            #[cfg(feature = "musical_transport")]
216            proc_transport_state: ProcTransportState::new(),
217            flags,
218            shared_flags,
219            clamp_graph_inputs_below_amp,
220            last_input_overflow_log_instant: None,
221            last_output_underflow_log_instant: None,
222            extra: ProcExtra {
223                scratch_buffers: ConstSequentialBuffer::new(
224                    stream_info.max_block_frames.get() as usize
225                ),
226                declick_values: DeclickValues::new(stream_info.declick_frames),
227                logger,
228                store,
229            },
230            poisoned: false,
231        }
232    }
233}
234
235pub(crate) struct NodeEntry {
236    pub processor: Box<dyn AudioNodeProcessor>,
237    pub bypass_declick: Declicker,
238    pub is_bypassed: bool,
239    pub is_first_process: bool,
240    pub in_place_buffers: bool,
241
242    event_data: NodeEventSchedulerData,
243}
244
245pub(crate) enum ContextToProcessorMsg {
246    EventGroup(Vec<NodeEvent>),
247    NewSchedule(Box<ScheduleHeapData>),
248    SetFlags(FirewheelBitFlags),
249    #[cfg(feature = "musical_transport")]
250    SetTransportState(Box<TransportState>),
251    #[cfg(feature = "scheduled_events")]
252    ClearScheduledEvents(SmallVec<[ClearScheduledEventsEvent; 1]>),
253}
254
255#[allow(clippy::enum_variant_names)]
256pub(crate) enum ProcessorToContextMsg {
257    DropEventGroup(Vec<NodeEvent>),
258    DropSchedule(Box<ScheduleHeapData>),
259    #[cfg(feature = "musical_transport")]
260    DropTransportState(Box<TransportState>),
261    #[cfg(feature = "scheduled_events")]
262    DropClearScheduledEvents(SmallVec<[ClearScheduledEventsEvent; 1]>),
263}
264
265#[cfg(feature = "scheduled_events")]
266pub(crate) struct ClearScheduledEventsEvent {
267    /// If `None`, then clear events for all nodes.
268    pub node_id: Option<NodeID>,
269    pub event_type: ClearScheduledEventsType,
270}
271
272#[cfg(feature = "scheduled_events")]
273#[derive(Clone)]
274pub(crate) struct SharedClock {
275    pub clock_samples: InstantSamples,
276    #[cfg(feature = "musical_transport")]
277    pub current_playhead: Option<InstantMusical>,
278    #[cfg(feature = "musical_transport")]
279    pub speed_multiplier: f64,
280    #[cfg(feature = "musical_transport")]
281    pub transport_is_playing: bool,
282    pub update_instant: Instant,
283}
284
285#[cfg(feature = "scheduled_events")]
286impl Default for SharedClock {
287    fn default() -> Self {
288        Self {
289            clock_samples: InstantSamples(0),
290            #[cfg(feature = "musical_transport")]
291            current_playhead: None,
292            #[cfg(feature = "musical_transport")]
293            speed_multiplier: 1.0,
294            #[cfg(feature = "musical_transport")]
295            transport_is_playing: false,
296            update_instant: Instant::now(),
297        }
298    }
299}
300
301/// How to handle event buffers on the audio thread running out of space.
302#[derive(Default, Debug, Clone, Copy, PartialEq, PartialOrd)]
303#[cfg_attr(feature = "bevy_reflect", derive(bevy_reflect::Reflect))]
304#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
305pub enum BufferOutOfSpaceMode {
306    #[default]
307    /// If an event buffer on the audio thread ran out of space to fit new
308    /// events, reallocate on the audio thread to fit the new items. If this
309    /// happens, it may cause underruns (audio glitches), and a warning will
310    /// be logged.
311    AllocateOnAudioThread,
312    /// If an event buffer on the audio thread ran out of space to fit new
313    /// events, then panic.
314    Panic,
315    /// If an event buffer on the audio thread ran out of space to fit new
316    /// events, drop those events to avoid allocating on the audio thread.
317    /// If this happens, a warning will be logged.
318    ///
319    /// (Not generally recommended, but the option is here if you want it.)
320    DropEvents,
321}
322
323#[derive(Default)]
324pub(crate) struct SharedFlags {
325    pub clipping_occurred: AtomicBool,
326}
firewheel_graph/processor.rs

firewheel_graph/
processor.rs
