feat: add perf support in walltime

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "codspeed-runner"
-version = "3.5.0"
+version = "3.6.0-beta.2"
 edition = "2021"
 repository = "https://github.com/CodSpeedHQ/runner"
 publish = false
@@ -46,6 +46,17 @@ sysinfo = { version = "0.33.1", features = ["serde"] }
 indicatif = "0.17.8"
 console = "0.15.8"
 async-trait = "0.1.82"
+libc = "0.2.171"
+bincode = "1.3.3"
+object = "0.36.7"
+linux-perf-data = "0.11.0"
+debugid = "0.8.0"
+memmap2 = "0.9.5"
+nix = { version = "0.29.0", features = ["fs"] }
+futures = "0.3.31"
+
+[target.'cfg(target_os = "linux")'.dependencies]
+procfs = "0.17.0"
 
 [dev-dependencies]
 temp-env = { version = "0.3.6", features = ["async_closure"] }

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
-channel = "1.84.0"
+channel = "1.85.0"
 components = ["rustfmt", "clippy"]

src/run/runner/helpers/mod.rs

@@ -2,3 +2,4 @@ pub mod env;
 pub mod get_bench_command;
 pub mod profile_folder;
 pub mod run_command_with_log_pipe;
+pub mod setup;

src/run/runner/helpers/run_command_with_log_pipe.rs

@@ -1,12 +1,30 @@
 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;
 
-pub fn run_command_with_log_pipe(mut cmd: Command) -> Result<ExitStatus> {
+/// Run a command and log its output to stdout and stderr
+///
+/// # Arguments
+/// - `cmd`: The command to run.
+/// - `cb`: A callback function that takes the process ID and returns a result.
+///
+/// # Returns
+///
+/// The exit status of the command.
+///
+pub async fn run_command_with_log_pipe_and_callback<F, Fut>(
+    mut cmd: Command,
+    cb: F,
+) -> Result<ExitStatus>
+where
+    F: FnOnce(u32) -> Fut,
+    Fut: Future<Output = anyhow::Result<()>>,
+{
     fn log_tee(
         mut reader: impl Read,
         mut writer: impl Write,
@@ -46,5 +64,12 @@ pub fn run_command_with_log_pipe(mut cmd: Command) -> Result<ExitStatus> {
     thread::spawn(move || {
         log_tee(stderr, std::io::stderr(), Some("[stderr]")).unwrap();
     });
+
+    cb(process.id()).await?;
+
     process.wait().context("failed to wait for the process")
 }
+
+pub async fn run_command_with_log_pipe(cmd: Command) -> Result<ExitStatus> {
+    run_command_with_log_pipe_and_callback(cmd, async |_| Ok(())).await
+}

src/run/runner/helpers/setup.rs

@@ -0,0 +1,32 @@
+use crate::prelude::*;
+use log::{debug, info};
+use std::process::{Command, Stdio};
+
+/// Run a command with sudo if available
+pub fn run_with_sudo(command_args: &[&str]) -> Result<()> {
+    let use_sudo = Command::new("sudo")
+        // `sudo true` will fail if sudo does not exist or the current user does not have sudo privileges
+        .arg("true")
+        .stdout(Stdio::null())
+        .status()
+        .is_ok_and(|status| status.success());
+    let mut command_args: Vec<&str> = command_args.into();
+    if use_sudo {
+        command_args.insert(0, "sudo");
+    }
+
+    debug!("Running command: {}", command_args.join(" "));
+    let output = Command::new(command_args[0])
+        .args(&command_args[1..])
+        .stdout(Stdio::piped())
+        .output()
+        .map_err(|_| anyhow!("Failed to execute command: {}", command_args.join(" ")))?;
+
+    if !output.status.success() {
+        info!("stdout: {}", String::from_utf8_lossy(&output.stdout));
+        error!("stderr: {}", String::from_utf8_lossy(&output.stderr));
+        bail!("Failed to execute command: {}", command_args.join(" "));
+    }
+
+    Ok(())
+}

src/run/runner/mod.rs

@@ -30,12 +30,15 @@ pub const EXECUTOR_TARGET: &str = "executor";
 pub fn get_executor_from_mode(mode: &RunnerMode) -> Box<dyn Executor> {
     match mode {
         RunnerMode::Instrumentation => Box::new(ValgrindExecutor),
-        RunnerMode::Walltime => Box::new(WallTimeExecutor),
+        RunnerMode::Walltime => Box::new(WallTimeExecutor::new()),
     }
 }
 
 pub fn get_all_executors() -> Vec<Box<dyn Executor>> {
-    vec![Box::new(ValgrindExecutor), Box::new(WallTimeExecutor)]
+    vec![
+        Box::new(ValgrindExecutor),
+        Box::new(WallTimeExecutor::new()),
+    ]
 }
 
 pub fn get_run_data() -> Result<RunData> {

src/run/runner/valgrind/executor.rs

@@ -30,7 +30,7 @@ impl Executor for ValgrindExecutor {
         mongo_tracer: &Option<MongoTracer>,
     ) -> Result<()> {
         //TODO: add valgrind version check
-        measure::measure(config, &run_data.profile_folder, mongo_tracer)?;
+        measure::measure(config, &run_data.profile_folder, mongo_tracer).await?;
 
         Ok(())
     }
@@ -41,7 +41,7 @@ impl Executor for ValgrindExecutor {
         _system_info: &SystemInfo,
         run_data: &RunData,
     ) -> Result<()> {
-        harvest_perf_maps(&run_data.profile_folder)?;
+        harvest_perf_maps(&run_data.profile_folder).await?;
 
         Ok(())
     }

src/run/runner/valgrind/helpers/perf_maps.rs

@@ -1,15 +1,9 @@
 use crate::prelude::*;
-use lazy_static::lazy_static;
-use regex::Regex;
 use std::collections::HashSet;
 use std::fs;
-use std::path::Path;
+use std::path::{Path, PathBuf};
 
-lazy_static! {
-    static ref PERF_MAP_REGEX: Regex = Regex::new(r"perf-(\d+)\.map").unwrap();
-}
-
-pub fn harvest_perf_maps(profile_folder: &Path) -> Result<()> {
+pub async fn harvest_perf_maps(profile_folder: &Path) -> Result<()> {
     // Get profile files (files with .out extension)
     let profile_files = fs::read_dir(profile_folder)?
         .filter_map(|entry| entry.ok())
@@ -21,27 +15,31 @@ pub fn harvest_perf_maps(profile_folder: &Path) -> Result<()> {
         .iter()
         .filter_map(|path| path.file_stem())
         .map(|pid| pid.to_str().unwrap())
+        .filter_map(|pid| pid.parse().ok())
         .collect::<HashSet<_>>();
 
-    let perf_map_files = fs::read_dir("/tmp")?
-        .filter_map(|entry| entry.ok())
-        .map(|entry| entry.path())
-        .filter(|path| {
-            path.file_name()
-                .and_then(|name| name.to_str())
-                .and_then(|name| PERF_MAP_REGEX.captures(name))
-                .and_then(|captures| captures.get(1))
-                .map(|pid| pids.contains(pid.as_str()))
-                .unwrap_or(false)
-        });
+    harvest_perf_maps_for_pids(profile_folder, &pids).await
+}
+
+pub async fn harvest_perf_maps_for_pids(profile_folder: &Path, pids: &HashSet<i32>) -> Result<()> {
+    let perf_maps = pids
+        .iter()
+        .map(|pid| format!("perf-{}.map", pid))
+        .map(|file_name| {
+            (
+                PathBuf::from("/tmp").join(&file_name),
+                profile_folder.join(&file_name),
+            )
+        })
+        .filter(|(src_path, _)| src_path.exists())
+        .collect::<Vec<_>>();
+    debug!("Found {} perf maps", perf_maps.len());
 
-    for perf_map_file in perf_map_files {
-        let source_path = perf_map_file.clone();
-        let dest_path = profile_folder.join(perf_map_file.file_name().unwrap());
-        fs::copy(source_path, dest_path).map_err(|e| {
+    for (src_path, dst_path) in perf_maps {
+        fs::copy(&src_path, &dst_path).map_err(|e| {
             anyhow!(
-                "Failed to copy perf map file: {} to {}: {}",
-                perf_map_file.display(),
+                "Failed to copy perf map file: {:?} to {}: {}",
+                src_path.file_name(),
                 profile_folder.display(),
                 e
             )

src/run/runner/valgrind/measure.rs

@@ -73,7 +73,7 @@ fn create_run_script() -> anyhow::Result<TempPath> {
     Ok(script_file.into_temp_path())
 }
 
-pub fn measure(
+pub async fn measure(
     config: &Config,
     profile_folder: &Path,
     mongo_tracer: &Option<MongoTracer>,
@@ -132,6 +132,7 @@ pub fn measure(
 
     debug!("cmd: {:?}", cmd);
     let status = run_command_with_log_pipe(cmd)
+        .await
         .map_err(|e| anyhow!("failed to execute the benchmark process. {}", e))?;
     debug!(
         "Valgrind exit code = {:?}, Valgrind signal = {:?}",

src/run/runner/valgrind/mod.rs

@@ -1,4 +1,4 @@
 pub mod executor;
-mod helpers;
+pub mod helpers;
 mod measure;
 mod setup;

src/run/runner/valgrind/setup.rs

@@ -1,41 +1,8 @@
-use std::{
-    env,
-    process::{Command, Stdio},
-};
-
-use url::Url;
-
+use crate::run::runner::helpers::setup::run_with_sudo;
 use crate::{prelude::*, run::helpers::download_file, VALGRIND_CODSPEED_VERSION};
 use crate::{run::check_system::SystemInfo, VALGRIND_CODSPEED_DEB_VERSION};
-
-/// Run a command with sudo if available
-fn run_with_sudo(command_args: &[&str]) -> Result<()> {
-    let use_sudo = Command::new("sudo")
-        // `sudo true` will fail if sudo does not exist or the current user does not have sudo privileges
-        .arg("true")
-        .stdout(Stdio::null())
-        .status()
-        .is_ok_and(|status| status.success());
-    let mut command_args: Vec<&str> = command_args.into();
-    if use_sudo {
-        command_args.insert(0, "sudo");
-    }
-
-    debug!("Running command: {}", command_args.join(" "));
-    let output = Command::new(command_args[0])
-        .args(&command_args[1..])
-        .stdout(Stdio::piped())
-        .output()
-        .map_err(|_| anyhow!("Failed to execute command: {}", command_args.join(" ")))?;
-
-    if !output.status.success() {
-        info!("stdout: {}", String::from_utf8_lossy(&output.stdout));
-        error!("stderr: {}", String::from_utf8_lossy(&output.stderr));
-        bail!("Failed to execute command: {}", command_args.join(" "));
-    }
-
-    Ok(())
-}
+use std::{env, process::Command};
+use url::Url;
 
 fn get_codspeed_valgrind_filename(system_info: &SystemInfo) -> Result<String> {
     let (version, architecture) = match (

src/run/runner/wall_time/executor.rs

@@ -1,5 +1,5 @@
+use super::perf::PerfRunner;
 use crate::prelude::*;
-
 use crate::run::instruments::mongo_tracer::MongoTracer;
 use crate::run::runner::executor::Executor;
 use crate::run::runner::helpers::env::get_base_injected_env;
@@ -11,22 +11,49 @@ use async_trait::async_trait;
 use std::fs::canonicalize;
 use std::process::Command;
 
-pub struct WallTimeExecutor;
+pub struct WallTimeExecutor {
+    perf: Option<PerfRunner>,
+}
+
+impl WallTimeExecutor {
+    pub fn new() -> Self {
+        let use_perf = if cfg!(target_os = "linux") {
+            std::env::var("CODSPEED_USE_PERF").is_ok()
+        } else {
+            false
+        };
+        debug!("Running the cmd with perf: {}", use_perf);
+
+        Self {
+            perf: use_perf.then(PerfRunner::new),
+        }
+    }
+}
 
 #[async_trait(?Send)]
 impl Executor for WallTimeExecutor {
     fn name(&self) -> ExecutorName {
         ExecutorName::WallTime
     }
 
+    async fn setup(&self, _system_info: &SystemInfo) -> Result<()> {
+        if self.perf.is_some() {
+            PerfRunner::setup_environment()?;
+        }
+
+        Ok(())
+    }
+
     async fn run(
         &self,
         config: &Config,
         _system_info: &SystemInfo,
         run_data: &RunData,
         _mongo_tracer: &Option<MongoTracer>,
     ) -> Result<()> {
-        let mut cmd = Command::new("sh");
+        // IMPORTANT: Don't use `sh` here! We will use this pid to send signals to the
+        // spawned child process which won't work if we use a different shell.
+        let mut cmd = Command::new("bash");
 
         cmd.envs(get_base_injected_env(
             RunnerMode::Walltime,
@@ -38,13 +65,22 @@ impl Executor for WallTimeExecutor {
             cmd.current_dir(abs_cwd);
         }
 
-        cmd.args(["-c", get_bench_command(config)?.as_str()]);
+        let bench_cmd = get_bench_command(config)?;
+        let status = if let Some(perf) = &self.perf {
+            perf.run(cmd, &bench_cmd).await
+        } else {
+            cmd.args(["-c", &bench_cmd]);
+            debug!("cmd: {:?}", cmd);
+
+            run_command_with_log_pipe(cmd).await
+        };
+
+        let status =
+            status.map_err(|e| anyhow!("failed to execute the benchmark process. {}", e))?;
+        debug!("cmd exit status: {:?}", status);
 
-        debug!("cmd: {:?}", cmd);
-        let status = run_command_with_log_pipe(cmd)
-            .map_err(|e| anyhow!("failed to execute the benchmark process. {}", e))?;
         if !status.success() {
-            bail!("failed to execute the benchmark process");
+            bail!("failed to execute the benchmark process: {}", status);
         }
 
         Ok(())
@@ -54,8 +90,14 @@ impl Executor for WallTimeExecutor {
         &self,
         _config: &Config,
         _system_info: &SystemInfo,
-        _run_data: &RunData,
+        run_data: &RunData,
     ) -> Result<()> {
+        debug!("Copying files to the profile folder");
+
+        if let Some(perf) = &self.perf {
+            perf.save_files_to(&run_data.profile_folder).await?;
+        }
+
         Ok(())
     }
 }

src/run/runner/wall_time/mod.rs

@@ -1 +1,2 @@
 pub mod executor;
+pub mod perf;