servo_media_audio/
render_thread.rs

1/* This Source Code Form is subject to the terms of the Mozilla Public
2 * License, v. 2.0. If a copy of the MPL was not distributed with this
3 * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
4
5use std::sync::mpsc::{Receiver, Sender};
6
7use malloc_size_of_derive::MallocSizeOf;
8use servo_media_streams::{MediaSocket, MediaStreamId};
9
10use crate::analyser_node::AnalyserNode;
11use crate::audio_node::{AudioNodeEngine, AudioNodeInit, AudioNodeMessage, BlockInfo, ChannelInfo};
12use crate::biquad_filter_node::BiquadFilterNode;
13use crate::block::{Chunk, FRAMES_PER_BLOCK, Tick};
14use crate::buffer_source_node::AudioBufferSourceNode;
15use crate::channel_node::{ChannelMergerNode, ChannelSplitterNode};
16use crate::constant_source_node::ConstantSourceNode;
17use crate::context::{AudioContextOptions, ProcessingState, StateChangeResult};
18use crate::convolver_node::ConvolverNode;
19use crate::delay_node::DelayNode;
20use crate::gain_node::GainNode;
21use crate::graph::{AudioGraph, InputPort, NodeId, OutputPort, PortId};
22use crate::iir_filter_node::IIRFilterNode;
23use crate::media_element_source_node::MediaElementSourceNode;
24use crate::media_stream_destination_node::MediaStreamDestinationNode;
25use crate::media_stream_source_node::MediaStreamSourceNode;
26use crate::offline_sink::OfflineAudioSink;
27use crate::oscillator_node::OscillatorNode;
28use crate::panner_node::PannerNode;
29use crate::sink::{AudioSink, AudioSinkError};
30use crate::stereo_panner::StereoPannerNode;
31use crate::wave_shaper_node::WaveShaperNode;
32use crate::{AudioBackend, AudioStreamReader};
33
34pub type SinkEosCallback = Box<dyn Fn(Box<dyn AsRef<[f32]>>) + Send + Sync + 'static>;
35
36#[derive(MallocSizeOf)]
37pub enum AudioRenderThreadMsg {
38    CreateNode(AudioNodeInit, Sender<Option<NodeId>>, ChannelInfo),
39    ConnectPorts(PortId<OutputPort>, PortId<InputPort>),
40    MessageNode(NodeId, AudioNodeMessage),
41    Resume(Sender<StateChangeResult>),
42    Suspend(Sender<StateChangeResult>),
43    Close(Sender<StateChangeResult>),
44    SinkNeedData,
45    GetCurrentTime(Sender<f64>),
46
47    DisconnectAllFrom(NodeId),
48    DisconnectOutput(PortId<OutputPort>),
49    DisconnectBetween(NodeId, NodeId),
50    DisconnectTo(NodeId, PortId<InputPort>),
51    DisconnectOutputBetween(PortId<OutputPort>, NodeId),
52    DisconnectOutputBetweenTo(PortId<OutputPort>, PortId<InputPort>),
53
54    SetSinkEosCallback(#[ignore_malloc_size_of = "Fn"] SinkEosCallback),
55
56    SetMute(bool),
57}
58
59pub enum Sink {
60    RealTime(Box<dyn AudioSink>),
61    Offline(OfflineAudioSink),
62}
63
64impl AudioSink for Sink {
65    fn init(
66        &self,
67        sample_rate: f32,
68        sender: Sender<AudioRenderThreadMsg>,
69    ) -> Result<(), AudioSinkError> {
70        match *self {
71            Sink::RealTime(ref sink) => sink.init(sample_rate, sender),
72            Sink::Offline(ref sink) => {
73                sink.init(sample_rate, sender).unwrap();
74                Ok(())
75            },
76        }
77    }
78
79    fn init_stream(&self, _: u8, _: f32, _: Box<dyn MediaSocket>) -> Result<(), AudioSinkError> {
80        unreachable!("Sink should never be used for MediaStreamDestinationNode")
81    }
82
83    fn play(&self) -> Result<(), AudioSinkError> {
84        match *self {
85            Sink::RealTime(ref sink) => sink.play(),
86            Sink::Offline(ref sink) => {
87                sink.play().unwrap();
88                Ok(())
89            },
90        }
91    }
92
93    fn stop(&self) -> Result<(), AudioSinkError> {
94        match *self {
95            Sink::RealTime(ref sink) => sink.stop(),
96            Sink::Offline(ref sink) => {
97                sink.stop().unwrap();
98                Ok(())
99            },
100        }
101    }
102
103    fn has_enough_data(&self) -> bool {
104        match *self {
105            Sink::RealTime(ref sink) => sink.has_enough_data(),
106            Sink::Offline(ref sink) => sink.has_enough_data(),
107        }
108    }
109
110    fn push_data(&self, chunk: Chunk) -> Result<(), AudioSinkError> {
111        match *self {
112            Sink::RealTime(ref sink) => sink.push_data(chunk),
113            Sink::Offline(ref sink) => {
114                sink.push_data(chunk).unwrap();
115                Ok(())
116            },
117        }
118    }
119
120    fn set_eos_callback(
121        &self,
122        callback: Box<dyn Fn(Box<dyn AsRef<[f32]>>) + Send + Sync + 'static>,
123    ) {
124        match *self {
125            Sink::RealTime(ref sink) => sink.set_eos_callback(callback),
126            Sink::Offline(ref sink) => sink.set_eos_callback(callback),
127        }
128    }
129}
130
131pub type ReaderFactoryCallback =
132    dyn Fn(MediaStreamId, f32) -> Result<Box<dyn AudioStreamReader + Send>, AudioSinkError>;
133
134pub struct AudioRenderThread {
135    pub graph: AudioGraph,
136    pub sink: Sink,
137    pub sink_factory: Box<dyn Fn() -> Result<Box<dyn AudioSink + 'static>, AudioSinkError>>,
138    pub reader_factory: Box<ReaderFactoryCallback>,
139    pub state: ProcessingState,
140    pub sample_rate: f32,
141    pub current_time: f64,
142    pub current_frame: Tick,
143    pub muted: bool,
144}
145
146impl AudioRenderThread {
147    /// Initializes the AudioRenderThread object
148    ///
149    /// You must call .event_loop() on this to run it!
150    fn prepare_thread<B: AudioBackend>(
151        sender: Sender<AudioRenderThreadMsg>,
152        sample_rate: f32,
153        graph: AudioGraph,
154        options: AudioContextOptions,
155    ) -> Result<Self, AudioSinkError> {
156        let sink_factory = Box::new(|| B::make_sink().map(|s| Box::new(s) as Box<dyn AudioSink>));
157        let reader_factory = Box::new(|id, sample_rate| B::make_streamreader(id, sample_rate));
158        let sink = match options {
159            AudioContextOptions::RealTimeAudioContext(_) => Sink::RealTime(sink_factory()?),
160            AudioContextOptions::OfflineAudioContext(options) => Sink::Offline(
161                OfflineAudioSink::new(options.channels as usize, options.length),
162            ),
163        };
164
165        sink.init(sample_rate, sender)?;
166
167        Ok(Self {
168            graph,
169            sink,
170            sink_factory,
171            reader_factory,
172            state: ProcessingState::Suspended,
173            sample_rate,
174            current_time: 0.,
175            current_frame: Tick(0),
176            muted: false,
177        })
178    }
179
180    /// Start the audio render thread
181    ///
182    /// In case something fails, it will instead start a thread with a dummy backend
183    pub fn start<B: AudioBackend>(
184        event_queue: Receiver<AudioRenderThreadMsg>,
185        sender: Sender<AudioRenderThreadMsg>,
186        sample_rate: f32,
187        graph: AudioGraph,
188        options: AudioContextOptions,
189        init_sender: Sender<Result<(), AudioSinkError>>,
190    ) {
191        let mut thread = match Self::prepare_thread::<B>(sender, sample_rate, graph, options) {
192            Ok(thread) => {
193                let _ = init_sender.send(Ok(()));
194                thread
195            },
196            Err(e) => {
197                let _ = init_sender.send(Err(e));
198                return;
199            },
200        };
201
202        thread.event_loop(event_queue);
203    }
204
205    make_render_thread_state_change!(resume, Running, play);
206
207    make_render_thread_state_change!(suspend, Suspended, stop);
208
209    fn create_node(&mut self, node_type: AudioNodeInit, ch: ChannelInfo) -> Option<NodeId> {
210        let mut needs_listener = false;
211        let mut is_dest = false;
212        let node: Box<dyn AudioNodeEngine> = match node_type {
213            AudioNodeInit::AnalyserNode(sender) => Box::new(AnalyserNode::new(sender, ch)),
214            AudioNodeInit::AudioBufferSourceNode(options) => {
215                Box::new(AudioBufferSourceNode::new(options, ch))
216            },
217            AudioNodeInit::BiquadFilterNode(options) => {
218                Box::new(BiquadFilterNode::new(options, ch, self.sample_rate))
219            },
220            AudioNodeInit::DelayNode(options) => Box::new(DelayNode::new(options, ch)),
221            AudioNodeInit::GainNode(options) => Box::new(GainNode::new(options, ch)),
222            AudioNodeInit::StereoPannerNode(options) => {
223                Box::new(StereoPannerNode::new(options, ch))
224            },
225            AudioNodeInit::PannerNode(options) => {
226                needs_listener = true;
227                Box::new(PannerNode::new(options, ch))
228            },
229            AudioNodeInit::MediaStreamSourceNode(id) => {
230                let reader = (self.reader_factory)(id, self.sample_rate);
231                Box::new(MediaStreamSourceNode::new(reader.ok()?, ch))
232            },
233            AudioNodeInit::OscillatorNode(options) => Box::new(OscillatorNode::new(options, ch)),
234            AudioNodeInit::ChannelMergerNode(options) => {
235                Box::new(ChannelMergerNode::new(options, ch))
236            },
237            AudioNodeInit::ConstantSourceNode(options) => {
238                Box::new(ConstantSourceNode::new(options, ch))
239            },
240            AudioNodeInit::ConvolverNode(options) => Box::new(ConvolverNode::new(options, ch)),
241            AudioNodeInit::MediaStreamDestinationNode(socket) => {
242                is_dest = true;
243                Box::new(MediaStreamDestinationNode::new(
244                    socket,
245                    self.sample_rate,
246                    (self.sink_factory)().unwrap(),
247                    ch,
248                ))
249            },
250            AudioNodeInit::ChannelSplitterNode => Box::new(ChannelSplitterNode::new(ch)),
251            AudioNodeInit::WaveShaperNode(options) => Box::new(WaveShaperNode::new(options, ch)),
252            AudioNodeInit::MediaElementSourceNode => Box::new(MediaElementSourceNode::new(ch)),
253            AudioNodeInit::IIRFilterNode(options) => Box::new(IIRFilterNode::new(options, ch)),
254            _ => unimplemented!(),
255        };
256        let id = self.graph.add_node(node);
257        if needs_listener {
258            let listener = self.graph.listener_id().output(0);
259            self.graph.add_edge(listener, id.listener());
260        }
261        if is_dest {
262            self.graph.add_extra_dest(id);
263        }
264        Some(id)
265    }
266
267    fn connect_ports(&mut self, output: PortId<OutputPort>, input: PortId<InputPort>) {
268        self.graph.add_edge(output, input)
269    }
270
271    fn process(&mut self) -> Chunk {
272        if self.muted {
273            return Chunk::explicit_silence();
274        }
275
276        let info = BlockInfo {
277            sample_rate: self.sample_rate,
278            frame: self.current_frame,
279            time: self.current_time,
280        };
281        self.graph.process(&info)
282    }
283
284    fn set_mute(&mut self, val: bool) {
285        self.muted = val;
286    }
287
288    fn event_loop(&mut self, event_queue: Receiver<AudioRenderThreadMsg>) {
289        let sample_rate = self.sample_rate;
290        let handle_msg = move |context: &mut Self, msg: AudioRenderThreadMsg| -> bool {
291            let mut break_loop = false;
292            match msg {
293                AudioRenderThreadMsg::CreateNode(node_type, tx, ch) => {
294                    let _ = tx.send(context.create_node(node_type, ch));
295                },
296                AudioRenderThreadMsg::ConnectPorts(output, input) => {
297                    context.connect_ports(output, input);
298                },
299                AudioRenderThreadMsg::Resume(tx) => {
300                    let _ = tx.send(context.resume());
301                },
302                AudioRenderThreadMsg::Suspend(tx) => {
303                    let _ = tx.send(context.suspend());
304                },
305                AudioRenderThreadMsg::Close(tx) => {
306                    let _ = tx.send(context.suspend());
307                    break_loop = true;
308                },
309                AudioRenderThreadMsg::GetCurrentTime(response) => {
310                    response.send(context.current_time).unwrap()
311                },
312                AudioRenderThreadMsg::MessageNode(id, msg) => {
313                    context.graph.node_mut(id).message(msg, sample_rate)
314                },
315                AudioRenderThreadMsg::SinkNeedData => {
316                    // Do nothing. This will simply unblock the thread so we
317                    // can restart the non-blocking event loop.
318                },
319                AudioRenderThreadMsg::DisconnectAllFrom(id) => {
320                    context.graph.disconnect_all_from(id)
321                },
322                AudioRenderThreadMsg::DisconnectOutput(out) => context.graph.disconnect_output(out),
323                AudioRenderThreadMsg::DisconnectBetween(from, to) => {
324                    context.graph.disconnect_between(from, to)
325                },
326                AudioRenderThreadMsg::DisconnectTo(from, to) => {
327                    context.graph.disconnect_to(from, to)
328                },
329                AudioRenderThreadMsg::DisconnectOutputBetween(from, to) => {
330                    context.graph.disconnect_output_between(from, to)
331                },
332                AudioRenderThreadMsg::DisconnectOutputBetweenTo(from, to) => {
333                    context.graph.disconnect_output_between_to(from, to)
334                },
335                AudioRenderThreadMsg::SetSinkEosCallback(callback) => {
336                    context.sink.set_eos_callback(callback);
337                },
338                AudioRenderThreadMsg::SetMute(val) => {
339                    context.set_mute(val);
340                },
341            };
342
343            break_loop
344        };
345
346        loop {
347            if self.sink.has_enough_data() || self.state == ProcessingState::Suspended {
348                // If we are not processing audio or
349                // if we have already pushed enough data into the audio sink
350                // we wait for messages coming from the control thread or
351                // the audio sink. The audio sink will notify whenever it
352                // needs more data.
353                if event_queue.recv().is_ok_and(|msg| handle_msg(self, msg)) {
354                    break;
355                }
356            } else {
357                // If we have not pushed enough data into the audio sink yet,
358                // we process the control message queue
359                if event_queue
360                    .try_recv()
361                    .is_ok_and(|msg| handle_msg(self, msg))
362                {
363                    break;
364                }
365
366                if self.state == ProcessingState::Suspended {
367                    // Bail out if we just suspended processing.
368                    continue;
369                }
370
371                // push into the audio sink the result of processing a
372                // render quantum.
373                let data = self.process();
374                if self.sink.push_data(data).is_ok() {
375                    // increment current frame by the render quantum size.
376                    self.current_frame += FRAMES_PER_BLOCK;
377                    self.current_time = self.current_frame / self.sample_rate as f64;
378                } else {
379                    eprintln!("Could not push data to audio sink");
380                }
381            }
382        }
383    }
384}
servo_media_audio/render_thread.rs

servo_media_audio/
render_thread.rs
