refactor(cli): process-reports: use rayon and channels to parallelize report parsing

Author: ErichDonGubler

Cargo.lock

@@ -24,6 +24,7 @@ log = "0.4.20"
 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"

Cargo.toml

@@ -21,7 +21,7 @@ use std::{
     num::NonZeroU64,
     path::{Path, PathBuf},
     process::ExitCode,
-    sync::Arc,
+    sync::{mpsc::channel, Arc},
 };
 
 use clap::Parser;
@@ -30,6 +30,7 @@ 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::{
@@ -347,30 +348,34 @@ fn run(cli: Cli) -> ExitCode {
 
             log::info!("gathering reported test outcomes for comparison to 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,