fuchsia_audio_device/
audio_frame_stream.rs

1// Copyright 2023 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use fuchsia_inspect as inspect;
6use fuchsia_inspect_derive::{AttachError, Inspect};
7use fuchsia_sync::Mutex;
8use futures::stream::FusedStream;
9use futures::task::{Context, Poll};
10use futures::{FutureExt, Stream};
11use log::info;
12use std::pin::Pin;
13use std::sync::Arc;
14
15use crate::frame_vmo::{self, FrameReadResult};
16use crate::stream_config::{SoftStreamConfig, StreamConfigOrTask};
17use crate::types::{AudioStreamItem, Error, Result};
18
19/// A stream that produces audio frames.
20/// Frames are of constant length.
21/// Usually acquired via SoftStreamConfig::create_output()
22pub struct AudioFrameStream {
23    /// Handle to the VMO that is receiving the frames.
24    frame_vmo: Arc<Mutex<frame_vmo::FrameVmo>>,
25    /// The next frame number we should retrieve.
26    next_frame: usize,
27    /// Number of frames to return in a packet.
28    packet_frames: usize,
29    /// Vector that will be filled with a packet.
30    /// Replaced when stream produces a packet.
31    next_packet: std::cell::RefCell<Vec<u8>>,
32    /// SoftStreamConfig this is attached to, or the SoftStreamConfig::process_requests task
33    stream_task: StreamConfigOrTask,
34    /// Inspect node
35    inspect: inspect::Node,
36}
37
38impl AudioFrameStream {
39    pub fn new(stream: SoftStreamConfig) -> AudioFrameStream {
40        AudioFrameStream {
41            frame_vmo: stream.frame_vmo(),
42            next_frame: 0,
43            packet_frames: stream.packet_frames(),
44            next_packet: Vec::new().into(),
45            stream_task: StreamConfigOrTask::StreamConfig(Box::new(stream)),
46            inspect: Default::default(),
47        }
48    }
49
50    /// Start the requests task if not started, and poll the task.
51    fn poll_task(&mut self, cx: &mut Context<'_>) -> Poll<Result<()>> {
52        if let StreamConfigOrTask::Complete = &self.stream_task {
53            return Poll::Ready(Err(Error::InvalidState));
54        }
55        if let StreamConfigOrTask::Task(task) = &mut self.stream_task {
56            return task.poll_unpin(cx);
57        }
58        self.stream_task.start();
59        self.poll_task(cx)
60    }
61
62    /// Poll for changes to active channels. Returns Ready(true) when channels become active,
63    /// and Ready(false) when channels become inactive.
64    pub fn poll_active_channels(&self, cx: &mut Context<'_>) -> Poll<bool> {
65        self.frame_vmo.lock().poll_active_channels(cx)
66    }
67
68    /// Watch for active channel state changes. Resolves on active <-> inactive transitions.
69    pub fn watch_active_channels(&self) -> impl Future<Output = bool> + 'static {
70        let frame_vmo = self.frame_vmo.clone();
71        async move { futures::future::poll_fn(|cx| frame_vmo.lock().poll_active_channels(cx)).await }
72    }
73}
74
75impl Stream for AudioFrameStream {
76    type Item = Result<AudioStreamItem>;
77
78    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
79        if let Poll::Ready(r) = self.poll_task(cx) {
80            self.stream_task = StreamConfigOrTask::Complete;
81            return Poll::Ready(r.err().map(Result::Err));
82        }
83        if self.next_packet.borrow().len() == 0 {
84            if let Some(new_len) = self.frame_vmo.lock().bytecount_frames(self.packet_frames) {
85                self.next_packet.borrow_mut().resize(new_len, 0);
86            }
87        }
88        let result = {
89            let mut lock = self.frame_vmo.lock();
90            futures::ready!(lock.poll_read(
91                self.next_frame,
92                self.next_packet.borrow_mut().as_mut_slice(),
93                cx
94            ))
95        };
96
97        match result {
98            Ok(FrameReadResult::Frames { next_frame, missed }) => {
99                if missed > 0 && self.next_frame != 0 {
100                    info!("Missed {missed} frames due to slow polling");
101                }
102                self.next_frame = next_frame;
103                let vec_mut = self.next_packet.get_mut();
104                let bytes = vec_mut.len();
105                let frames = std::mem::replace(vec_mut, vec![0; bytes]);
106                Poll::Ready(Some(Ok(AudioStreamItem::Data(frames))))
107            }
108            Ok(FrameReadResult::AudioDisabled) => {
109                // Don't worry about missing frames when we restart
110                self.next_frame = 0;
111                Poll::Ready(Some(Ok(AudioStreamItem::AudioDisabled)))
112            }
113            Err(e) => Poll::Ready(Some(Err(e))),
114        }
115    }
116}
117
118impl FusedStream for AudioFrameStream {
119    fn is_terminated(&self) -> bool {
120        match self.stream_task {
121            StreamConfigOrTask::Complete => true,
122            _ => false,
123        }
124    }
125}
126
127impl Inspect for &mut AudioFrameStream {
128    fn iattach(
129        self,
130        parent: &fuchsia_inspect::Node,
131        name: impl AsRef<str>,
132    ) -> core::result::Result<(), AttachError> {
133        self.inspect = parent.create_child(name.as_ref());
134        if let StreamConfigOrTask::StreamConfig(o) = &mut self.stream_task {
135            return o.iattach(&self.inspect, "stream_config");
136        }
137        Ok(())
138    }
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144
145    use async_utils::PollExt;
146    use fidl_fuchsia_hardware_audio::*;
147    use fixture::fixture;
148    use fuchsia_async as fasync;
149    use futures::StreamExt;
150
151    use crate::stream_config::tests::with_audio_frame_stream;
152
153    const TEST_UNIQUE_ID: &[u8; 16] = &[5; 16];
154    const TEST_CLOCK_DOMAIN: u32 = 0x00010203;
155
156    #[fixture(with_audio_frame_stream)]
157    #[fuchsia::test]
158    #[rustfmt::skip]
159    fn soft_audio_out(mut exec: fasync::TestExecutor, stream_config: StreamConfigProxy, mut frame_stream: AudioFrameStream) {
160        let mut frame_fut = frame_stream.next();
161        // Poll the frame stream, which should start the processing of proxy requests.
162        exec.run_until_stalled(&mut frame_fut).expect_pending("no frames yet");
163
164        let result = exec.run_until_stalled(&mut stream_config.get_properties());
165        assert!(result.is_ready());
166        let props1 = match result {
167            Poll::Ready(Ok(v)) => v,
168            _ => panic!("stream config get properties error"),
169        };
170
171        assert_eq!(props1.unique_id.unwrap(),                *TEST_UNIQUE_ID);
172        assert_eq!(props1.is_input.unwrap(),                 false);
173        assert_eq!(props1.can_mute.unwrap(),                 false);
174        assert_eq!(props1.can_agc.unwrap(),                  false);
175        assert_eq!(props1.min_gain_db.unwrap(),              0f32);
176        assert_eq!(props1.max_gain_db.unwrap(),              0f32);
177        assert_eq!(props1.gain_step_db.unwrap(),             0f32);
178        assert_eq!(props1.plug_detect_capabilities.unwrap(), PlugDetectCapabilities::Hardwired);
179        assert_eq!(props1.manufacturer.unwrap(),             "Google");
180        assert_eq!(props1.product.unwrap(),                  "UnitTest");
181        assert_eq!(props1.clock_domain.unwrap(),             TEST_CLOCK_DOMAIN);
182
183        let result = exec.run_until_stalled(&mut stream_config.get_supported_formats());
184        assert!(result.is_ready());
185
186        let formats = match result {
187            Poll::Ready(Ok(v)) => v,
188            _ => panic!("get supported formats error"),
189        };
190
191        let first = formats.first().to_owned().expect("supported formats to be present");
192        let pcm = first.pcm_supported_formats.to_owned().expect("pcm format to be present");
193        assert_eq!(pcm.channel_sets.unwrap()[0].attributes.as_ref().unwrap().len(), 2usize);
194        assert_eq!(pcm.sample_formats.unwrap()[0],        SampleFormat::PcmSigned);
195        assert_eq!(pcm.bytes_per_sample.unwrap()[0],      2u8);
196        assert_eq!(pcm.valid_bits_per_sample.unwrap()[0], 16u8);
197        assert_eq!(pcm.frame_rates.unwrap()[0],           44100);
198
199        let (ring_buffer, server) = fidl::endpoints::create_proxy::<RingBufferMarker>();
200
201        let format = Format {
202            pcm_format: Some(fidl_fuchsia_hardware_audio::PcmFormat {
203                number_of_channels:      2u8,
204                sample_format:           SampleFormat::PcmSigned,
205                bytes_per_sample:        2u8,
206                valid_bits_per_sample:   16u8,
207                frame_rate:              44100,
208            }),
209            ..Default::default()
210        };
211
212        stream_config.create_ring_buffer(&format, server).expect("ring buffer error");
213
214        let props2 = match exec.run_until_stalled(&mut ring_buffer.get_properties()) {
215            Poll::Ready(Ok(v)) => v,
216            x => panic!("expected Ready Ok from get_properties, got {:?}", x),
217        };
218        assert_eq!(props2.needs_cache_flush_or_invalidate, Some(false));
219        assert!(props2.driver_transfer_bytes.unwrap() > 0);
220
221        let result = exec.run_until_stalled(&mut ring_buffer.get_vmo(88200, 0)); // 2 seconds.
222        assert!(result.is_ready());
223        let reply = match result {
224            Poll::Ready(Ok(Ok(v))) => v,
225            _ => panic!("ring buffer get vmo error"),
226        };
227        let audio_vmo = reply.1;
228
229        // Frames * bytes per sample * channels per sample.
230        let bytes_per_second: usize = 44100 * 2 * 2;
231        let vmo_size = audio_vmo.get_size().expect("size after getbuffer");
232        assert!(bytes_per_second <= vmo_size as usize);
233
234        // Put "audio" in buffer.
235        let mut sent_audio = Vec::new();
236        let mut x: u8 = 0x01;
237        sent_audio.resize_with(bytes_per_second, || {
238            x = x.wrapping_add(2);
239            x
240        });
241
242        assert_eq!(Ok(()), audio_vmo.write(&sent_audio, 0));
243
244        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(42));
245        let _ = exec.wake_expired_timers();
246        let start_time = exec.run_until_stalled(&mut ring_buffer.start());
247        if let Poll::Ready(s) = start_time {
248            assert_eq!(s.expect("start time error"), 42);
249        } else {
250            panic!("start error");
251        }
252
253        exec.run_until_stalled(&mut frame_fut).expect_pending("no frames until time passes");
254
255        // Run the ring buffer for a bit over half a second.
256        exec.set_fake_time(fasync::MonotonicInstant::after(zx::MonotonicDuration::from_millis(500)));
257        let _ = exec.wake_expired_timers();
258
259        let result = exec.run_until_stalled(&mut frame_fut);
260        assert!(result.is_ready());
261        let audio_recv = match result {
262            Poll::Ready(Some(Ok(AudioStreamItem::Data(v)))) => v,
263            x => panic!("expected Ready Ok from frame stream, got {:?}", x),
264        };
265
266        // We should receive exactly 100ms of audio
267        let expect_recv_bytes = bytes_per_second / 10;
268        assert_eq!(expect_recv_bytes, audio_recv.len());
269        assert_eq!(&sent_audio[0..expect_recv_bytes], &audio_recv);
270
271        let result = exec.run_until_stalled(&mut frame_fut);
272        assert!(result.is_ready());
273        let audio_recv = match result {
274            Poll::Ready(Some(Ok(AudioStreamItem::Data(v)))) => v,
275            x => panic!("expected Ready Ok from frame stream, got {:?}", x),
276        };
277
278        // We should receive exactly the next 100ms of audio
279        let expect_recv_bytes = bytes_per_second / 10;
280        assert_eq!(expect_recv_bytes, audio_recv.len());
281        assert_eq!(&sent_audio[expect_recv_bytes..expect_recv_bytes*2], &audio_recv);
282
283
284        let result = exec.run_until_stalled(&mut ring_buffer.stop());
285        assert!(result.is_ready());
286
287        // Watch gain only replies once.
288        let result = exec.run_until_stalled(&mut stream_config.watch_gain_state());
289        assert!(result.is_ready());
290        let result = exec.run_until_stalled(&mut stream_config.watch_gain_state());
291        assert!(!result.is_ready());
292
293        // Watch plug state only replies once.
294        let result = exec.run_until_stalled(&mut stream_config.watch_plug_state());
295        assert!(result.is_ready());
296        let result = exec.run_until_stalled(&mut stream_config.watch_plug_state());
297        assert!(!result.is_ready());
298    }
299
300    #[fixture(with_audio_frame_stream)]
301    #[fuchsia::test]
302    fn soft_audio_out_active_channels(
303        mut exec: fasync::TestExecutor,
304        stream_config: StreamConfigProxy,
305        mut frame_stream: AudioFrameStream,
306    ) {
307        let (ring_buffer, server) = fidl::endpoints::create_proxy::<RingBufferMarker>();
308        let format = Format {
309            pcm_format: Some(fidl_fuchsia_hardware_audio::PcmFormat {
310                number_of_channels: 2u8,
311                sample_format: SampleFormat::PcmSigned,
312                bytes_per_sample: 2u8,
313                valid_bits_per_sample: 16u8,
314                frame_rate: 44100,
315            }),
316            ..Default::default()
317        };
318
319        // Poll the frame stream, which should start the processing of proxy requests.
320        exec.run_until_stalled(&mut frame_stream.next()).expect_pending("no frames yet");
321
322        stream_config.create_ring_buffer(&format, server).expect("ring buffer error");
323
324        // watch_active_channels should resolve to true
325        let mut active_fut = Box::pin(frame_stream.watch_active_channels());
326        let res = exec.run_until_stalled(&mut active_fut);
327        assert_eq!(res, Poll::Ready(true));
328
329        let mut active_fut = Box::pin(frame_stream.watch_active_channels());
330        let res = exec.run_until_stalled(&mut active_fut);
331        assert_eq!(res, Poll::Pending);
332
333        // Disable channels via RingBuffer - this should be possible before starting
334        // the VMO.
335        let res = exec.run_until_stalled(&mut ring_buffer.set_active_channels(0));
336        assert!(res.is_ready());
337
338        // watch_active_channels should resolve to false
339        let res = exec.run_until_stalled(&mut active_fut);
340        assert_eq!(res, Poll::Ready(false));
341
342        assert!(res.is_ready());
343        let result = exec.run_until_stalled(&mut ring_buffer.get_vmo(88200, 0));
344        assert!(result.is_ready());
345        let reply = match result {
346            Poll::Ready(Ok(Ok(v))) => v,
347            x => panic!("ring buffer get vmo error: {x:?}"),
348        };
349        let audio_vmo = reply.1;
350
351        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(42));
352        let _ = exec.wake_expired_timers();
353        let start_time = exec.run_until_stalled(&mut ring_buffer.start());
354        assert!(start_time.is_ready());
355
356        // Create active watcher future and frame future
357        let mut active_fut = Box::pin(frame_stream.watch_active_channels());
358        let res = exec.run_until_stalled(&mut active_fut);
359        assert_eq!(res, Poll::Pending);
360
361        // Polling frames when channels are disabled yields AudioDisabled event
362        let mut frame_fut = frame_stream.next();
363        let res = exec.run_until_stalled(&mut frame_fut);
364        assert!(matches!(res, Poll::Ready(Some(Ok(AudioStreamItem::AudioDisabled)))));
365
366        // Polling frames again while still disabled returns Pending
367        let mut frame_fut = frame_stream.next();
368        exec.set_fake_time(fasync::MonotonicInstant::after(zx::MonotonicDuration::from_millis(
369            500,
370        )));
371        let _ = exec.wake_expired_timers();
372        let res = exec.run_until_stalled(&mut frame_fut);
373        assert!(res.is_pending());
374
375        // Put audio in buffer
376        let bytes_per_second: usize = 44100 * 2 * 2;
377        let sent_audio = vec![0x12; bytes_per_second];
378        assert_eq!(Ok(()), audio_vmo.write(&sent_audio, 0));
379
380        // Re-enable channels (bitmask 0b11 = 3 for 2 channels)
381        let res = exec.run_until_stalled(&mut ring_buffer.set_active_channels(3));
382        assert!(res.is_ready());
383
384        // watch_active_channels should resolve to true
385        let res = exec.run_until_stalled(&mut active_fut);
386        assert_eq!(res, Poll::Ready(true));
387
388        // frame_fut should now resolve with audio data
389        let res = exec.run_until_stalled(&mut frame_fut);
390        assert!(matches!(res, Poll::Ready(Some(Ok(AudioStreamItem::Data(_))))));
391    }
392}
fuchsia_audio_device/audio_frame_stream.rs

fuchsia_audio_device/
audio_frame_stream.rs
