audio_streams_conformance_test: add playback test for NoopStream

The first version of audio_streams_conformance_test.
It is used to test if the implementation of
audio_streams::AsyncPlaybackBufferStream is correct.

This version only supports the playback test of NoopStream.
It prints the following information for NoopStream.
==

Playback Source: NoopStream
Channels: 2
Format: S16LE
Sample rate: 48000 frames/s
Buffer size: 240 frames
Iterations: 10

Cold start latency: 3.89µs
Records count: 10
[Step] min: 4.94 ms, max: 5.15 ms, average: 5.01 ms,
standard deviation: 0.06 ms.
[Linear Regression] rate: 47916.19 frames/s, standard error: 2.0

BUG=b:238038707
TEST=cargo run

Change-Id: Ifca7dfd35473ffd75856a27e2c6aa1555eba7576
Reviewed-on: https://chromium-review.googlesource.com/c/crosvm/crosvm/+/3805090
Tested-by: Judy Hsiao <judyhsiao@google.com>
Reviewed-by: Chih-Yang Hsia <paulhsia@chromium.org>
Commit-Queue: Chih-Yang Hsia <paulhsia@chromium.org>
Auto-Submit: Judy Hsiao <judyhsiao@google.com>

Cargo.lock

@@ -167,6 +167,7 @@ dependencies = [
  "async-trait",
  "futures",
  "remain",
+ "serde",
  "thiserror",
 ]
 

Cargo.toml

@@ -103,7 +103,8 @@ exclude = [
     "common/sync",
     "tube_transporter",
     "win_util",
-    "tools/examples/baremetal"
+    "tools/examples/baremetal",
+    "tools/audio_streams_conformance_test",
 ]
 
 [features]

common/audio_streams/Cargo.lock

@@ -20,6 +20,7 @@ dependencies = [
  "async-trait",
  "futures",
  "remain",
+ "serde",
  "thiserror",
 ]
 
@@ -165,6 +166,26 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "serde"
+version = "1.0.144"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0f747710de3dcd43b88c9168773254e809d8ddbdf9653b84e2554ab219f17860"
+dependencies = [
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_derive"
+version = "1.0.144"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "94ed3a816fb1d101812f83e789f888322c34e291f894f19590dc310963e87a00"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "slab"
 version = "0.4.7"

common/audio_streams/Cargo.toml

@@ -12,3 +12,4 @@ async-trait = "0.1.36"
 remain = "0.2"
 thiserror = "1.0.20"
 futures = "0.3"
+serde = { version = "1.0", features = ["derive"] }

common/audio_streams/src/audio_streams.rs

@@ -61,9 +61,10 @@ pub use async_api::AsyncStream;
 pub use async_api::AudioStreamsExecutor;
 use async_trait::async_trait;
 use remain::sorted;
+use serde::Serialize;
 use thiserror::Error;
 
-#[derive(Copy, Clone, Debug, PartialEq)]
+#[derive(Copy, Clone, Debug, PartialEq, Serialize)]
 pub enum SampleFormat {
     U8,
     S16LE,
@@ -95,6 +96,27 @@ impl Display for SampleFormat {
     }
 }
 
+impl FromStr for SampleFormat {
+    type Err = SampleFormatError;
+    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
+        match s {
+            "U8" => Ok(SampleFormat::U8),
+            "S16_LE" => Ok(SampleFormat::S16LE),
+            "S24_LE" => Ok(SampleFormat::S24LE),
+            "S32_LE" => Ok(SampleFormat::S32LE),
+            _ => Err(SampleFormatError::InvalidSampleFormat),
+        }
+    }
+}
+
+/// Errors that are possible from a `SampleFormat`.
+#[sorted]
+#[derive(Error, Debug)]
+pub enum SampleFormatError {
+    #[error("Must be in [U8, S16_LE, S24_LE, S32_LE]")]
+    InvalidSampleFormat,
+}
+
 /// Valid directions of an audio stream.
 #[derive(Copy, Clone, Debug, PartialEq)]
 pub enum StreamDirection {

common/cros_asyncv2/Cargo.lock

@@ -56,6 +56,7 @@ dependencies = [
  "async-trait",
  "futures",
  "remain",
+ "serde",
  "thiserror",
 ]
 

tools/audio_streams_conformance_test/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "audio_streams_conformance_test"
+version = "0.1.0"
+authors = ["The Chromium OS Authors"]
+edition = "2021"
+
+[[bin]]
+name = "audio_streams_conformance_test"
+path = "src/main.rs"
+
+[dependencies]
+argh = "*"
+audio_streams = { path = "../../common/audio_streams" } # provided by ebuild
+cros_async = { path = "../../cros_async" } # provided by ebuild
+num = "*"
+remain = "0.2"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "*"
+thiserror = "1.0.20"

tools/audio_streams_conformance_test/src/args.rs

@@ -0,0 +1,117 @@
+// Copyright 2022 The ChromiumOS Authors.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+use std::{fmt, str::FromStr};
+
+use argh::FromArgs;
+use audio_streams::*;
+use serde::Serialize;
+
+use super::error::Error;
+
+// maybe use StreamSourceGenerator directly
+#[derive(Copy, Clone, Debug, PartialEq, Serialize)]
+pub enum StreamSourceEnum {
+    NoopStream,
+}
+
+impl fmt::Display for StreamSourceEnum {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            StreamSourceEnum::NoopStream => write!(f, "noop"),
+        }
+    }
+}
+
+impl FromStr for StreamSourceEnum {
+    type Err = Error;
+    fn from_str(s: &str) -> ::std::result::Result<StreamSourceEnum, Self::Err> {
+        match s {
+            "noop" => Ok(StreamSourceEnum::NoopStream),
+            _ => Err(Error::InvalidStreamSuorce(s.to_owned())),
+        }
+    }
+}
+
+fn default_channels() -> usize {
+    2
+}
+
+fn default_sample_format() -> SampleFormat {
+    SampleFormat::S16LE
+}
+
+fn default_rate() -> u32 {
+    48000
+}
+
+fn default_buffer_frames() -> usize {
+    240
+}
+
+fn default_iterations() -> usize {
+    10
+}
+
+fn default_stream_source() -> StreamSourceEnum {
+    StreamSourceEnum::NoopStream
+}
+
+#[derive(Copy, Clone, Debug, FromArgs, Serialize)]
+/// test test
+pub struct Args {
+    /// the StreamSource to use for playback. (default: noop).
+    #[argh(
+        option,
+        short = 'P',
+        default = "default_stream_source()",
+        from_str_fn(StreamSourceEnum::from_str)
+    )]
+    pub playback_source: StreamSourceEnum,
+    /// the channel numbers. (default: 2)
+    #[argh(option, short = 'c', default = "default_channels()")]
+    pub channels: usize,
+    /// format. Must be in [U8, S16_LE, S24_LE, S32_LE]. (default:S16_LE)
+    #[argh(
+        option,
+        short = 'f',
+        default = "default_sample_format()",
+        from_str_fn(SampleFormat::from_str)
+    )]
+    pub format: SampleFormat,
+    /// sample rate. (default: 48000)
+    #[argh(option, short = 'r', default = "default_rate()")]
+    pub rate: u32,
+    /// block buffer size (frames) of each write. (default: 240).
+    #[argh(option, short = 'b', default = "default_buffer_frames()")]
+    pub buffer_frames: usize,
+    /// the iterations to fill in the audio buffer. default: 10)
+    #[argh(option, default = "default_iterations()")]
+    pub iterations: usize,
+    /// whether or not to print in json format
+    #[argh(switch)]
+    pub json: bool,
+}
+
+impl fmt::Display for Args {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(
+            f,
+            r#"
+Playback Source: {:?}
+Channels: {}
+Format: {:?}
+Sample rate: {} frames/s
+Buffer size: {} frames
+Iterations: {}
+          "#,
+            self.playback_source,
+            self.channels,
+            self.format,
+            self.rate,
+            self.buffer_frames,
+            self.iterations
+        )
+    }
+}

tools/audio_streams_conformance_test/src/error.rs

@@ -0,0 +1,32 @@
+// Copyright 2022 The ChromiumOS Authors.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+use audio_streams::BoxError;
+use remain::sorted;
+use std::io;
+use thiserror::Error;
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+#[sorted]
+#[derive(Error, Debug)]
+pub enum Error {
+    /// Creating stream failed.
+    #[error(transparent)]
+    CreateStream(BoxError),
+    #[error(transparent)]
+    FetchBuffer(BoxError),
+    #[error("failed to generate stream source: {0}")]
+    GenerateStreamSource(BoxError),
+    #[error("invalid stream source: `{0}`")]
+    InvalidStreamSuorce(String),
+    #[error("mismatched x[] and y[] for linear regression")]
+    MismatchedSamples,
+    #[error("do not have enough samples")]
+    NotEnoughSamples,
+    #[error(transparent)]
+    SerdeError(#[from] serde_json::Error),
+    #[error(transparent)]
+    WriteBuffer(io::Error),
+}

tools/audio_streams_conformance_test/src/main.rs

@@ -0,0 +1,75 @@
+// Copyright 2022 The ChromiumOS Authors.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+use std::{io, time::Instant};
+
+use audio_streams::*;
+use cros_async::Executor;
+
+mod args;
+mod error;
+mod performance_data;
+
+use crate::{
+    args::*,
+    error::{Error, Result},
+    performance_data::*,
+};
+
+async fn run_playback(ex: &Executor, args: &Args) -> Result<PerformanceData> {
+    let mut data = PerformanceData::default();
+    let generator: Box<dyn StreamSourceGenerator> = match args.playback_source {
+        StreamSourceEnum::NoopStream => Box::new(NoopStreamSourceGenerator::new()),
+    };
+    let num_channels = args.channels;
+    let format = args.format;
+    let frame_rate = args.rate;
+    let buffer_size = args.buffer_frames;
+    let iterations = args.iterations;
+
+    let mut stream_source = generator.generate().map_err(Error::GenerateStreamSource)?;
+    let start = Instant::now();
+    let (_, mut stream) = stream_source
+        .new_async_playback_stream(num_channels, format, frame_rate, buffer_size, ex)
+        .map_err(Error::CreateStream)?;
+    data.cold_start = start.elapsed();
+    let frame_size = args.format.sample_bytes() * args.channels;
+
+    let start = Instant::now();
+    let mut frames_played = 0;
+    for _ in 0..iterations {
+        let mut stream_buffer = stream
+            .next_playback_buffer(ex)
+            .await
+            .map_err(Error::FetchBuffer)?;
+        let bytes = stream_buffer
+            .copy_from(&mut io::repeat(0))
+            .map_err(Error::WriteBuffer)?;
+        stream_buffer.commit().await;
+        frames_played += bytes / frame_size;
+        data.records
+            .push(BufferConsumptionRecord::new(frames_played, start.elapsed()));
+    }
+    Ok(data)
+}
+
+fn main() -> Result<()> {
+    let args: Args = argh::from_env();
+    let ex = Executor::new().expect("Failed to create an executor");
+    let done = run_playback(&ex, &args);
+
+    match ex.run_until(done) {
+        Ok(Ok(data)) => {
+            let report = data.gen_report(args)?;
+            if args.json {
+                println!("{}", serde_json::to_string(&report)?);
+            } else {
+                print!("{}", report);
+            }
+        }
+        Ok(Err(e)) => eprintln!("{}", e),
+        Err(e) => eprintln!("Error happened in executor: {}", e),
+    }
+    Ok(())
+}

tools/audio_streams_conformance_test/src/performance_data.rs

@@ -0,0 +1,224 @@
+// Copyright 2022 The ChromiumOS Authors.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+use std::{fmt, time::Duration};
+
+use num::integer::Roots;
+use serde::Serialize;
+
+use crate::{args::Args, error::*};
+
+const NANOS_PER_MICROS: f32 = 1_000_000.0;
+
+/// `PerformanceReport` is the estimated buffer consumption rate and error term
+/// derived by the linear regression of `BufferConsumptionRecord`.
+#[derive(Debug, Serialize)]
+pub struct PerformanceReport {
+    args: Args,
+    cold_start_latency: Duration,
+    record_count: usize,
+    rate: EstimatedRate,
+    /// {min, max, avg, stddev}_time for per "next_buffer + zero write + commit" call
+    min_time: Duration,
+    max_time: Duration,
+    avg_time: Duration,
+    stddev_time: Duration,
+    /// How many times that consumed frames are different from buffer_frames.
+    mismatched_frame_count: u32,
+}
+
+impl fmt::Display for PerformanceReport {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        if self.mismatched_frame_count != 0 {
+            eprint!(
+                "[Error] {} consumed buffers size != {} frames",
+                self.mismatched_frame_count, self.args.buffer_frames
+            );
+        }
+        write!(
+            f,
+            r#"{}
+Cold start latency: {:?}
+Records count: {}
+[Step] min: {:.2} ms, max: {:.2} ms, average: {:.2} ms, standard deviation: {:.2} ms.
+{}
+"#,
+            self.args,
+            self.cold_start_latency,
+            self.record_count,
+            to_micros(self.min_time),
+            to_micros(self.max_time),
+            to_micros(self.avg_time),
+            to_micros(self.stddev_time),
+            self.rate,
+        )
+    }
+}
+
+/// `BufferConsumptionRecord` records the timestamp and the
+/// accumulated number of consumed frames at every stream buffer commit.
+/// It is used to compute the buffer consumption rate.
+#[derive(Debug, Default)]
+pub struct BufferConsumptionRecord {
+    pub ts: Duration,
+    pub frames: usize,
+}
+
+impl BufferConsumptionRecord {
+    pub fn new(frames: usize, ts: Duration) -> Self {
+        Self { ts, frames }
+    }
+}
+
+#[derive(Debug, Serialize, PartialEq)]
+pub struct EstimatedRate {
+    /// linear coefficients of LINEST(frames,timestamps).
+    rate: f64,
+    /// STEYX(frames, timestamps).
+    error: f64,
+}
+
+impl EstimatedRate {
+    fn new(rate: f64, error: f64) -> Self {
+        Self { rate, error }
+    }
+}
+
+impl fmt::Display for EstimatedRate {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(
+            f,
+            "[Linear Regression] rate: {:.2} frames/s, standard error: {:.2} ",
+            self.rate, self.error
+        )
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct PerformanceData {
+    pub cold_start: Duration,
+    pub records: Vec<BufferConsumptionRecord>,
+}
+
+fn to_micros(t: Duration) -> f32 {
+    t.as_nanos() as f32 / NANOS_PER_MICROS
+}
+
+fn linear_regression(x: &[f64], y: &[f64]) -> Result<EstimatedRate> {
+    if x.len() != y.len() {
+        return Err(Error::MismatchedSamples);
+    }
+
+    if x.len() <= 2 {
+        return Err(Error::NotEnoughSamples);
+    }
+
+    /* hat(y_i) = b(x_i) + a */
+    let x_sum: f64 = x.iter().sum();
+    let x_average = x_sum / x.len() as f64;
+    // sum(x_i * x_i)
+    let x_square_sum: f64 = x.iter().map(|&xi| xi * xi).sum();
+    // sum(x_i * y_i)
+    let x_y_sum: f64 = x.iter().zip(y.iter()).map(|(&xi, &yi)| xi * yi).sum();
+
+    let y_sum: f64 = y.iter().sum();
+
+    let y_square_sum: f64 = y.iter().map(|yi| yi * yi).sum();
+    /* b = (n * sum(x * y) - sum(x) * sum(y)) / (n * sum(x ^ 2) - sum(x) ^ 2)
+    = (sum(x * y) - avg(x) * sum(y)) / (sum(x ^ 2) - avg(x) * sum(x)) */
+    let b = (x_y_sum - x_average * y_sum) / (x_square_sum - x_average * x_sum);
+    let n = y.len() as f64;
+    /* err = sqrt(sum((y_i - hat(y_i)) ^ 2) / n) */
+    let err: f64 = ((n * y_square_sum - y_sum * y_sum - b * b * (n * x_square_sum - x_sum * x_sum))
+        as f64
+        / (n * (n - 2.0)))
+        .sqrt();
+
+    Ok(EstimatedRate::new(b, err))
+}
+
+impl PerformanceData {
+    pub fn gen_report(&self, args: Args) -> Result<PerformanceReport> {
+        let time_records: Vec<f64> = self
+            .records
+            .iter()
+            .map(|record| record.ts.as_secs_f64())
+            .collect();
+
+        let frames: Vec<f64> = self
+            .records
+            .iter()
+            .map(|record| record.frames as f64)
+            .collect();
+
+        let mut steps = Vec::new();
+        let mut mismatched_frame_count = 0;
+        for i in 1..frames.len() {
+            let time_diff = self.records[i].ts - self.records[i - 1].ts;
+            steps.push(time_diff);
+
+            let frame_diff = self.records[i].frames - self.records[i - 1].frames;
+            if frame_diff != args.buffer_frames {
+                mismatched_frame_count += 1;
+            }
+        }
+        let avg_time = steps
+            .iter()
+            .sum::<Duration>()
+            .checked_div(steps.len() as u32)
+            .ok_or(Error::NotEnoughSamples)?;
+        let stddev_time = (steps
+            .iter()
+            .map(|x| {
+                x.as_nanos().abs_diff(avg_time.as_nanos())
+                    * x.as_nanos().abs_diff(avg_time.as_nanos())
+            })
+            .sum::<u128>()
+            / steps.len() as u128)
+            .sqrt();
+
+        let rate = linear_regression(&time_records, &frames)?;
+        let min_time = steps.iter().min().unwrap().to_owned();
+        let max_time = steps.iter().max().unwrap().to_owned();
+
+        Ok(PerformanceReport {
+            args,
+            cold_start_latency: self.cold_start,
+            record_count: self.records.len(),
+            rate,
+            min_time,
+            max_time,
+            avg_time,
+            stddev_time: Duration::from_nanos(stddev_time as u64),
+            mismatched_frame_count,
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test1() {
+        let xs: Vec<f64> = vec![1.0, 2.0, 3.0, 4.0, 5.0];
+        let ys: Vec<f64> = vec![1.0, 2.0, 3.0, 4.0, 5.0];
+
+        assert_eq!(
+            EstimatedRate::new(1.0, 0.0),
+            linear_regression(&xs, &ys).expect("test1 should pass")
+        );
+    }
+
+    #[test]
+    fn test2() {
+        let xs: Vec<f64> = vec![1.0, 2.0, 3.0, 4.0, 5.0];
+        let ys: Vec<f64> = vec![2.0, 4.0, 5.0, 4.0, 5.0];
+
+        assert_eq!(
+            EstimatedRate::new(0.6, 0.8944271909999159),
+            linear_regression(&xs, &ys).expect("test2 should pass")
+        );
+    }
+}