Print runtime benchmark results interactively

Author: Kobzol

collector/benchlib/src/benchmark.rs

@@ -1,23 +1,22 @@
 use crate::cli::{parse_cli, Args, BenchmarkArgs};
+use crate::comm::messages::{BenchmarkMeasurement, BenchmarkMessage, BenchmarkStats};
+use crate::comm::output_message;
 use crate::measure::benchmark_function;
-use crate::messages::BenchmarkResult;
 use crate::process::raise_process_priority;
-use log::LevelFilter;
 use std::collections::HashMap;
 
 /// Create a new benchmark suite. Use the closure argument to define benchmarks.
 pub fn benchmark_suite<F: FnOnce(&mut BenchmarkSuite)>(define_func: F) {
-    env_logger::Builder::from_default_env()
-        .filter_level(LevelFilter::Info)
-        .init();
+    env_logger::init();
+
     let mut suite = BenchmarkSuite::new();
     define_func(&mut suite);
     suite.run().expect("Benchmark suite has failed");
 }
 
 /// Type-erased function that performs a benchmark.
 struct BenchmarkWrapper {
-    func: Box<dyn Fn() -> anyhow::Result<BenchmarkResult>>,
+    func: Box<dyn Fn() -> anyhow::Result<BenchmarkMeasurement>>,
 }
 
 type BenchmarkMap = HashMap<&'static str, BenchmarkWrapper>;
@@ -80,16 +79,25 @@ fn run_benchmark(args: BenchmarkArgs, benchmarks: BenchmarkMap) -> anyhow::Resul
         .collect();
     items.sort_unstable_by_key(|item| item.0);
 
-    let mut results: Vec<BenchmarkResult> = Vec::with_capacity(items.len());
+    let mut stdout = std::io::stdout().lock();
+
     for (name, def) in items {
+        let mut measurements: Vec<BenchmarkMeasurement> =
+            Vec::with_capacity(args.iterations as usize);
         for i in 0..args.iterations {
-            let result = (def.func)()?;
-            log::info!("Benchmark (run {i}) `{name}` completed: {result:?}");
-            results.push(result);
+            let measurement = (def.func)()?;
+            log::info!("Benchmark (run {i}) `{name}` completed: {measurement:?}");
+            measurements.push(measurement);
         }
+        output_message(
+            &mut stdout,
+            BenchmarkMessage::Stats(BenchmarkStats {
+                name: name.to_string(),
+                measurements,
+            }),
+        )?;
     }
 
-    println!("{}", serde_json::to_string(&results)?);
     Ok(())
 }
 

collector/benchlib/src/comm/messages.rs

@@ -0,0 +1,27 @@
+//! This module defines messages that are exchanged between a binary that uses `benchlib` and
+//! the `collector` crate.
+
+use std::time::Duration;
+
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub enum BenchmarkMessage {
+    Stats(BenchmarkStats),
+}
+
+/// Results of several measurements of a single benchmark.
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub struct BenchmarkStats {
+    pub name: String,
+    pub measurements: Vec<BenchmarkMeasurement>,
+}
+
+/// Results of a single benchmark execution.
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub struct BenchmarkMeasurement {
+    pub cycles: u64,
+    pub instructions: u64,
+    pub branch_misses: u64,
+    pub cache_misses: u64,
+    pub cache_references: u64,
+    pub wall_time: Duration,
+}

collector/benchlib/src/comm/mod.rs

@@ -0,0 +1,40 @@
+use crate::comm::messages::BenchmarkMessage;
+use std::io::{BufRead, BufReader, Read, Write};
+
+pub mod messages;
+
+/// Messages are communicated as line-delimited JSON.
+pub fn output_message<W: Write>(mut sink: W, message: BenchmarkMessage) -> anyhow::Result<()> {
+    serde_json::to_writer(&mut sink, &message)?;
+    sink.write_all(b"\n")?;
+    Ok(())
+}
+
+pub struct MessageReader<R> {
+    inner: BufReader<R>,
+    line: String,
+}
+
+impl<R: Read> MessageReader<R> {
+    pub fn new(inner: R) -> Self {
+        Self {
+            inner: BufReader::new(inner),
+            line: Default::default(),
+        }
+    }
+}
+
+impl<R: Read> Iterator for MessageReader<R> {
+    type Item = anyhow::Result<BenchmarkMessage>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.inner.read_line(&mut self.line) {
+            Ok(0) => None,
+            Ok(_) => match serde_json::from_str(&self.line) {
+                Ok(value) => Some(Ok(value)),
+                Err(error) => Some(Err(error.into())),
+            },
+            Err(error) => Some(Err(error.into())),
+        }
+    }
+}

collector/benchlib/src/lib.rs

@@ -15,6 +15,6 @@
 
 pub mod benchmark;
 mod cli;
+pub mod comm;
 pub mod measure;
-pub mod messages;
 pub mod process;

collector/benchlib/src/measure/perf_counter/unix.rs

@@ -1,5 +1,5 @@
 use crate::benchmark::black_box;
-use crate::messages::BenchmarkResult;
+use crate::comm::messages::BenchmarkMeasurement;
 use perf_event::events::Hardware;
 use perf_event::{Builder, Counter, Group};
 use std::time::Instant;
@@ -18,7 +18,7 @@ struct Counters {
 pub fn benchmark_function<F: Fn() -> Bench + 'static, R, Bench: FnOnce() -> R + 'static>(
     name: &'static str,
     benchmark_constructor: F,
-) -> anyhow::Result<BenchmarkResult> {
+) -> anyhow::Result<BenchmarkMeasurement> {
     let mut group = create_group()?;
     let counters = prepare_counters(&mut group)?;
 
@@ -48,8 +48,7 @@ pub fn benchmark_function<F: Fn() -> Bench + 'static, R, Bench: FnOnce() -> R +
     // Try to avoid optimizing the result out.
     black_box(output);
 
-    let result = BenchmarkResult {
-        name: String::from(name),
+    let result = BenchmarkMeasurement {
         cycles: measurement[&counters.cycles],
         instructions: measurement[&counters.instructions],
         branch_misses: measurement[&counters.branch_misses],

collector/benchlib/src/measure/perf_counter/windows.rs

@@ -1,8 +1,8 @@
-use crate::messages::BenchmarkResult;
+use crate::messages::BenchmarkStats;
 
 pub fn benchmark_function<F: FnOnce() -> R, R>(
     _name: &'static str,
     _func: F,
-) -> anyhow::Result<BenchmarkResult> {
+) -> anyhow::Result<BenchmarkStats> {
     panic!("Runtime benchmarking is not supported on Windows");
 }

collector/benchlib/src/messages.rs

@@ -1,17 +1,24 @@
 use crate::benchmark::profile::Profile;
 use crate::toolchain::{get_local_toolchain, LocalToolchain};
 use benchlib::benchmark::passes_filter;
-use benchlib::messages::BenchmarkResult;
+use benchlib::comm::messages::BenchmarkMessage;
 use cargo_metadata::Message;
+use std::io::{BufRead, BufReader};
 use std::path::{Path, PathBuf};
-use std::process::Command;
+use std::process::{Command, Stdio};
 
 #[derive(Debug)]
 struct BenchmarkBinary {
     path: PathBuf,
     benchmark_names: Vec<String>,
 }
 
+impl BenchmarkBinary {
+    fn name(&self) -> &str {
+        self.path.file_name().unwrap().to_str().unwrap()
+    }
+}
+
 #[derive(Debug)]
 struct BenchmarkDatabase {
     binaries: Vec<BenchmarkBinary>,
@@ -73,25 +80,40 @@ pub fn bench_runtime(
         total_benchmark_count - filtered
     );
 
+    let mut benchmark_index = 0;
     for binary in benchmark_db.binaries {
-        let name = binary.path.file_name().and_then(|s| s.to_str()).unwrap();
-
-        let data: Vec<BenchmarkResult> = execute_runtime_binary(&binary.path, name, &filter)?;
-        // TODO: do something with the result
-        println!("{name}: {:?}", data);
+        for message in execute_runtime_benchmark(&binary.path, &filter)? {
+            let message = message.map_err(|err| {
+                anyhow::anyhow!(
+                    "Cannot parse BenchmarkMessage from benchmark {}: {err:?}",
+                    binary.path.display()
+                )
+            })?;
+            match message {
+                BenchmarkMessage::Stats(stats) => {
+                    benchmark_index += 1;
+                    println!(
+                        "Finished {}/{} ({}/{})",
+                        binary.name(),
+                        stats.name,
+                        benchmark_index,
+                        filtered
+                    );
+                }
+            }
+        }
     }
 
     Ok(())
 }
 
-/// Execute a single runtime benchmark suite defined in a binary crate located in
+/// Starts executing a single runtime benchmark suite defined in a binary crate located in
 /// `runtime-benchmarks`. The binary is expected to use benchlib's `BenchmarkSuite` to execute
-/// a set of runtime benchmarks and return a list of `BenchmarkResult`s encoded as JSON.
-fn execute_runtime_binary(
+/// a set of runtime benchmarks and print `BenchmarkMessage`s encoded as JSON, one per line.
+fn execute_runtime_benchmark(
     binary: &Path,
-    name: &str,
     filter: &BenchmarkFilter,
-) -> anyhow::Result<Vec<BenchmarkResult>> {
+) -> anyhow::Result<impl Iterator<Item = anyhow::Result<BenchmarkMessage>>> {
     // Turn off ASLR
     let mut command = Command::new("setarch");
     command.arg(std::env::consts::ARCH);
@@ -106,19 +128,16 @@ fn execute_runtime_binary(
         command.args(&["--include", include]);
     }
 
-    let result = command.output()?;
-
-    if !result.status.success() {
-        return Err(anyhow::anyhow!(
-            "Failed to run runtime benchmark {name}\n{}\n{}",
-            String::from_utf8_lossy(&result.stdout),
-            String::from_utf8_lossy(&result.stderr)
-        ));
-    }
-
-    log::info!("Successfully ran runtime benchmark {name}");
+    command.stdout(Stdio::piped());
+    let mut child = command.spawn()?;
+    let stdout = child.stdout.take().unwrap();
 
-    Ok(serde_json::from_slice(&result.stdout)?)
+    let reader = BufReader::new(stdout);
+    let iterator = reader.lines().map(|line| {
+        line.and_then(|line| Ok(serde_json::from_str::<BenchmarkMessage>(&line)?))
+            .map_err(|err| err.into())
+    });
+    Ok(iterator)
 }
 
 /// Compiles all runtime benchmarks and returns the stdout output of Cargo.