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

servo_media_audio/
render_thread.rs
