wip: try make process async + parallel callback

Author: not-matthias

Cargo.lock

@@ -29,7 +29,7 @@ serde = { version = "1.0.192", features = ["derive"] }
 serde_json = { version = "1.0.108", features = ["preserve_order"] }
 url = "2.4.1"
 sha256 = "1.4.0"
-tokio = { version = "1", features = ["macros", "rt"] }
+tokio = { version = "1", features = ["macros", "rt","process"] }
 tokio-tar = "0.3.1"
 md5 = "0.7.0"
 base64 = "0.21.0"

Cargo.toml

@@ -2,10 +2,10 @@ use std::{
     env,
     io::Read,
     path::{Path, PathBuf},
-    process::{Child, Command, Stdio},
+    process::{Child, Stdio},
     str::FromStr,
-    thread,
 };
+use tokio::process::Command;
 
 use reqwest::Client;
 use tokio::fs;
@@ -150,21 +150,22 @@ impl MongoTracer {
         } else {
             info!("No MongoDB URI provided, user will have to provide it dynamically through the CodSpeed integration");
         }
-        let mut process = command
+        let process = command
             .stdout(Stdio::piped())
             .stderr(Stdio::piped())
             .spawn()?;
 
-        let process_stdout = process.stdout.take().expect("error taking child stdout");
-        let process_stderr = process.stderr.take().expect("error taking child stderr");
-        thread::spawn(move || {
-            dump_tracer_log(process_stdout).expect("error communicating with child stdout")
-        });
-        thread::spawn(move || {
-            dump_tracer_log(process_stderr).expect("error communicating with child stderr")
-        });
+        // let process_stdout = process.stdout.take().expect("error taking child stdout");
+        // let process_stderr = process.stderr.take().expect("error taking child stderr");
+        // FIXME: Add this again
+        // thread::spawn(move || {
+        //     dump_tracer_log(process_stdout).expect("error communicating with child stdout")
+        // });
+        // thread::spawn(move || {
+        //     dump_tracer_log(process_stderr).expect("error communicating with child stderr")
+        // });
 
-        self.process = Some(process);
+        // self.process = Some(process);
 
         Ok(())
     }
@@ -243,7 +244,7 @@ pub async fn install_mongodb_tracer() -> Result<()> {
     let output = Command::new("bash")
         .arg(installer_path.to_str().unwrap())
         .stdout(Stdio::piped())
-        .output()
+        .output().await
         .map_err(|_| anyhow!("Failed to install mongo-tracer"))?;
 
     if !output.status.success() {

src/run/instruments/mongo_tracer.rs

@@ -1,11 +1,7 @@
-use crate::local_logger::suspend_progress_bar;
 use crate::prelude::*;
-use crate::run::runner::EXECUTOR_TARGET;
 use std::future::Future;
-use std::io::{Read, Write};
-use std::process::Command;
 use std::process::ExitStatus;
-use std::thread;
+use tokio::process::Command;
 
 /// Run a command and log its output to stdout and stderr
 ///
@@ -25,49 +21,52 @@ where
     F: FnOnce(u32) -> Fut,
     Fut: Future<Output = anyhow::Result<()>>,
 {
-    fn log_tee(
-        mut reader: impl Read,
-        mut writer: impl Write,
-        log_prefix: Option<&str>,
-    ) -> Result<()> {
-        let prefix = log_prefix.unwrap_or("");
-        let mut buffer = [0; 1024];
-        loop {
-            let bytes_read = reader.read(&mut buffer)?;
-            if bytes_read == 0 {
-                break;
-            }
-            suspend_progress_bar(|| {
-                writer.write_all(&buffer[..bytes_read]).unwrap();
-                trace!(
-                    target: EXECUTOR_TARGET,
-                    "{}{}",
-                    prefix,
-                    String::from_utf8_lossy(&buffer[..bytes_read])
-                );
-            });
-        }
-        Ok(())
-    }
+    // fn log_tee(
+    //     mut reader: impl Read,
+    //     mut writer: impl Write,
+    //     log_prefix: Option<&str>,
+    // ) -> Result<()> {
+    //     let prefix = log_prefix.unwrap_or("");
+    //     let mut buffer = [0; 1024];
+    //     loop {
+    //         let bytes_read = reader.read(&mut buffer)?;
+    //         if bytes_read == 0 {
+    //             break;
+    //         }
+    //         suspend_progress_bar(|| {
+    //             writer.write_all(&buffer[..bytes_read]).unwrap();
+    //             trace!(
+    //                 target: EXECUTOR_TARGET,
+    //                 "{}{}",
+    //                 prefix,
+    //                 String::from_utf8_lossy(&buffer[..bytes_read])
+    //             );
+    //         });
+    //     }
+    //     Ok(())
+    // }
 
     let mut process = cmd
         .stdout(std::process::Stdio::piped())
         .stderr(std::process::Stdio::piped())
         .spawn()
         .context("failed to spawn the process")?;
-    let stdout = process.stdout.take().expect("unable to get stdout");
-    let stderr = process.stderr.take().expect("unable to get stderr");
-    thread::spawn(move || {
-        log_tee(stdout, std::io::stdout(), None).unwrap();
-    });
+    // let stdout = process.stdout.take().expect("unable to get stdout");
+    // let stderr = process.stderr.take().expect("unable to get stderr");
+    // thread::spawn(move || {
+    //     log_tee(stdout, std::io::stdout(), None).unwrap();
+    // });
 
-    thread::spawn(move || {
-        log_tee(stderr, std::io::stderr(), Some("[stderr]")).unwrap();
-    });
+    // thread::spawn(move || {
+    //     log_tee(stderr, std::io::stderr(), Some("[stderr]")).unwrap();
+    // });
 
-    cb(process.id()).await?;
+    let (_, exit_status) = tokio::join!(
+        cb(process.id().unwrap()),
+        process.wait()
+    );
 
-    process.wait().context("failed to wait for the process")
+    exit_status.context("failed to wait for the process")
 }
 
 pub async fn run_command_with_log_pipe(cmd: Command) -> Result<ExitStatus> {

src/run/runner/helpers/run_command_with_log_pipe.rs

@@ -13,7 +13,8 @@ use std::io::Write;
 use std::os::unix::fs::PermissionsExt;
 use std::os::unix::process::ExitStatusExt;
 use std::path::Path;
-use std::{env::consts::ARCH, process::Command};
+use std::{env::consts::ARCH};
+use tokio::process::Command;
 use tempfile::TempPath;
 
 lazy_static! {

src/run/runner/valgrind/measure.rs

@@ -9,7 +9,7 @@ use crate::run::runner::{ExecutorName, RunData, RunnerMode};
 use crate::run::{check_system::SystemInfo, config::Config};
 use async_trait::async_trait;
 use std::fs::canonicalize;
-use std::process::Command;
+use tokio::process::Command;
 
 pub struct WallTimeExecutor {
     perf: Option<PerfRunner>,

src/run/runner/wall_time/executor.rs

@@ -13,7 +13,7 @@ use perf_map::ProcessSymbols;
 use shared::Command as FifoCommand;
 use std::collections::HashSet;
 use std::path::PathBuf;
-use std::process::Command;
+use tokio::process::Command;
 use std::time::Duration;
 use std::{cell::OnceCell, collections::HashMap, process::ExitStatus};
 use tempfile::TempDir;