Rewrite cargo logs processing using an MPSC channel

Author: syl20bnr

crates/tracel-xtask/src/commands/test.rs

@@ -38,12 +38,14 @@ pub fn handle_command(args: TestCmdArgs) -> anyhow::Result<()> {
 }
 
 fn push_optional_args(cmd_args: &mut Vec<String>, args: &TestCmdArgs) {
+    // cargo options
     if let Some(jobs) = &args.jobs {
         cmd_args.extend(vec!["--jobs".to_string(), jobs.to_string()]);
     };
+    // test harness options
+    cmd_args.extend(vec!["--".to_string(), "--color=always".to_string()]);
     if let Some(threads) = &args.threads {
         cmd_args.extend(vec![
-            "--".to_string(),
             "--test-threads".to_string(),
             threads.to_string(),
         ]);

crates/tracel-xtask/src/utils/process.rs

@@ -2,7 +2,7 @@ use std::{
     collections::HashMap,
     io::{BufRead, BufReader},
     path::Path,
-    process::{Command, Stdio},
+    process::{Command, Stdio}, sync::mpsc, thread,
 };
 
 use anyhow::anyhow;
@@ -61,6 +61,7 @@ pub fn run_process_for_workspace<'a>(
         .iter()
         .for_each(|ex| args.extend(["--exclude", ex]));
     group_info!("Command line: cargo {}", args.join(" "));
+    // process
     let mut child = Command::new(name)
         .args(&args)
         .stdout(Stdio::piped())
@@ -75,53 +76,80 @@ pub fn run_process_for_workspace<'a>(
             ))
         })?;
 
-    let mut ignore_error = false;
-    let mut close_group = false;
+    // hanlde stdout and stderr in dedicated threads using a MPSC channel for synchronization
+    let (tx, rx) = mpsc::channel();
+    // stdout processing thread
     if let Some(stdout) = child.stdout.take() {
-        let reader = BufReader::new(stdout);
-        reader.lines().for_each(|line| {
-            if let Ok(line) = line {
-                println!("{}", line);
+        let tx = tx.clone();
+        thread::spawn(move || {
+            let reader = BufReader::new(stdout);
+            for line in reader.lines() {
+                if let Ok(line) = line {
+                    tx.send((line, false)).unwrap();
+                }
             }
         });
     }
+    // stderr processing thread
     if let Some(stderr) = child.stderr.take() {
-        let reader = BufReader::new(stderr);
-        reader.lines().for_each(|line| {
-            let mut skip_line = false;
-            if let Ok(line) = line {
-                if let Some(rx) = &group_rx {
-                    let cleaned_line = standardize_slashes(&remove_ansi_codes(&line));
-                    if let Some(caps) = rx.captures(&cleaned_line) {
-                        let crate_name = &caps[1];
-                        if close_group {
-                            endgroup!();
-                        }
-                        close_group = true;
-                        group!("{}: {}", group_name.unwrap_or("Group"), crate_name);
-                    }
+        let tx = tx.clone();
+        thread::spawn(move || {
+            let reader = BufReader::new(stderr);
+            for line in reader.lines() {
+                if let Ok(line) = line {
+                    tx.send((line, true)).unwrap();
                 }
-                if let Some(log) = ignore_log {
-                    if line.contains(log) {
-                        if let Some(msg) = ignore_msg {
-                            warn!("{}", msg);
-                        }
-                        ignore_error = true;
-                        skip_line = true;
-                    }
+            }
+        });
+    }
+    // Drop the sender once all the logs have been processed to close the channel
+    drop(tx);
+
+    // Process the stdout to inject log groups
+    let mut ignore_error = false;
+    let mut close_group = false;
+    for (line, is_stderr) in rx.iter() {
+        let mut skip_line = false;
+
+        if let Some(rx) = &group_rx {
+            let cleaned_line = standardize_slashes(&remove_ansi_codes(&line));
+            if let Some(caps) = rx.captures(&cleaned_line) {
+                let crate_name = &caps[1];
+                if close_group {
+                    endgroup!();
                 }
-                if !skip_line {
-                    eprintln!("{}", line);
+                close_group = true;
+                group!("{}: {}", group_name.unwrap_or("Group"), crate_name);
+            }
+        }
+
+        if let Some(log) = ignore_log {
+            if line.contains(log) {
+                if let Some(msg) = ignore_msg {
+                    warn!("{}", msg);
                 }
+                ignore_error = true;
+                skip_line = true;
             }
-        });
+        }
+
+        if !skip_line {
+            if is_stderr {
+                eprintln!("{}", line);
+            } else {
+                println!("{}", line);
+            }
+        }
     }
+
     if close_group {
         endgroup!();
     }
+
     let status = child
         .wait()
         .expect("Should be able to wait for the process to finish.");
+
     if status.success() || ignore_error {
         anyhow::Ok(())
     } else {