refactor(cli): process-reports: use rayon and channels to paralle…

…lize report parsing
erichdongubler-mozilla · Oct 31, 2023 · c886ac2 · c886ac2
1 parent 29dc7dc
commit c886ac2
Show file tree

Hide file tree

Showing 3 changed files with 106 additions and 26 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/moz-webgpu-cts/Cargo.toml b/moz-webgpu-cts/Cargo.toml
@@ -21,6 +21,7 @@ log = { workspace = true }
 miette = { version = "5.10.0", features = ["fancy"] }
 natord = "1.0.9"
 path-dsl = "0.6.1"
+rayon = "1.8.0"
 regex = "1.9.5"
 serde = { version = "1.0.188", features = ["derive"] }
 serde_json = "1.0.107"

diff --git a/moz-webgpu-cts/src/main.rs b/moz-webgpu-cts/src/main.rs
@@ -22,7 +22,7 @@ use std::{
     io::{self, BufReader, BufWriter},
     path::{Path, PathBuf},
     process::ExitCode,
-    sync::Arc,
+    sync::{mpsc::channel, Arc},
 };
 
 use clap::Parser;
@@ -31,7 +31,7 @@ use format::lazy_format;
 use indexmap::{IndexMap, IndexSet};
 use miette::{miette, Diagnostic, IntoDiagnostic, NamedSource, Report, SourceSpan, WrapErr};
 use path_dsl::path;
-
+use rayon::prelude::{IntoParallelIterator, ParallelIterator};
 use regex::Regex;
 use wax::Glob;
 use whippit::{metadata::SectionHeader, reexport::chumsky::prelude::Rich};
@@ -344,30 +344,34 @@ fn run(cli: Cli) -> ExitCode {
 
             log::info!("gathering reported test outcomes for reconciliation with metadata…");
 
-            let exec_reports_iter = exec_report_paths.iter().map(|path| {
-                fs::File::open(path)
-                    .map(BufReader::new)
-                    .map_err(Report::msg)
-                    .wrap_err("failed to open file")
-                    .and_then(|reader| {
-                        serde_json::from_reader::<_, ExecutionReport>(reader)
-                            .into_diagnostic()
-                            .wrap_err("failed to parse JSON")
-                    })
-                    .map(|parsed| (path, parsed))
-                    .wrap_err_with(|| {
-                        format!(
-                            "failed to read WPT execution report from {}",
-                            path.display()
-                        )
-                    })
-                    .map_err(|e| {
-                        log::error!("{e:?}");
-                        AlreadyReportedToCommandline
-                    })
-            });
-
-            for res in exec_reports_iter {
+            let (exec_reports_sender, exec_reports_receiver) = channel();
+            exec_report_paths
+                .into_par_iter()
+                .for_each_with(exec_reports_sender, |sender, path| {
+                    let res = fs::File::open(&path)
+                        .map(BufReader::new)
+                        .map_err(Report::msg)
+                        .wrap_err("failed to open file")
+                        .and_then(|reader| {
+                            serde_json::from_reader::<_, ExecutionReport>(reader)
+                                .into_diagnostic()
+                                .wrap_err("failed to parse JSON")
+                        })
+                        .wrap_err_with(|| {
+                            format!(
+                                "failed to read WPT execution report from {}",
+                                path.display()
+                            )
+                        })
+                        .map(|parsed| (path, parsed))
+                        .map_err(|e| {
+                            log::error!("{e:?}");
+                            AlreadyReportedToCommandline
+                        });
+                    let _ = sender.send(res);
+                });
+
+            for res in exec_reports_receiver {
                 let (_path, exec_report) = match res {
                     Ok(ok) => ok,
                     Err(AlreadyReportedToCommandline) => return ExitCode::FAILURE,