feat: add perf metadata; add python support

Author: not-matthias

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/wall_time/perf/metadata.rs

@@ -0,0 +1,29 @@
+// !!!!!!!!!!!!!!!!!!!!!!!!
+// !! DO NOT TOUCH BELOW !!
+// !!!!!!!!!!!!!!!!!!!!!!!!
+// Has to be in sync with `perf-parser`.
+//
+
+use std::{collections::HashMap, path::Path};
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize)]
+pub struct PerfMetadata {
+    /// Name and version of the integration
+    pub integration: (String, String),
+
+    /// The URIs of the benchmarks in the order they were executed.
+    pub bench_order_by_pid: HashMap<u32, Vec<String>>,
+
+    /// Modules that should be ignored and removed from the folded trace and callgraph (e.g. python interpreter)
+    pub ignored_modules: Vec<(String, u64, u64)>,
+}
+
+impl PerfMetadata {
+    pub fn save_to<P: AsRef<Path>>(&self, path: P) -> anyhow::Result<()> {
+        let file = std::fs::File::create(path.as_ref().join("perf.metadata"))?;
+        serde_json::to_writer(file, self)?;
+        Ok(())
+    }
+}

src/run/runner/wall_time/perf/mod.rs

@@ -1,19 +1,24 @@
 use crate::prelude::*;
 use crate::run::runner::helpers::run_command_with_log_pipe::run_command_with_log_pipe_and_callback;
 use crate::run::runner::helpers::setup::run_with_sudo;
+use crate::run::runner::valgrind::helpers::ignored_objects_path::get_objects_path_to_ignore;
+use crate::run::runner::valgrind::helpers::perf_maps::harvest_perf_maps_for_pids;
 use anyhow::Context;
 use fifo::{PerfFifo, RunnerFifo};
 use futures::stream::FuturesUnordered;
+use metadata::PerfMetadata;
 use perf_map::ProcessSymbols;
 use procfs::process::MMPermissions;
 use shared::Command as FifoCommand;
+use std::collections::HashSet;
 use std::path::PathBuf;
 use std::process::Command;
 use std::time::Duration;
 use std::{cell::OnceCell, collections::HashMap, process::ExitStatus};
 use tempfile::TempDir;
 use unwind_data::UnwindData;
 
+mod metadata;
 mod shared;
 pub use shared::*;
 
@@ -74,10 +79,21 @@ impl PerfRunner {
             .prefix(PERF_DATA_PREFIX)
             .tempfile_in(&self.perf_dir)?;
 
+        // Detect the mode based on the command to be executed
+        let cg_mode = if bench_cmd.contains("cargo") {
+            "dwarf"
+        } else if bench_cmd.contains("pytest") {
+            "fp"
+        } else {
+            warn!("Couldn't detect call graph mode for command: {}", bench_cmd);
+            "dwarf"
+        };
+        debug!("Using call graph mode: {}", cg_mode);
+
         cmd.args([
             "-c",
             &format!(
-                "perf record --quiet --user-callchains --freq=999 --switch-output --control=fifo:{},{} --delay=-1 -g --call-graph=dwarf --output={} -- {bench_cmd}",
+                "perf record --quiet --user-callchains --freq=999 --switch-output --control=fifo:{},{} --delay=-1 -g --call-graph={cg_mode} --output={} -- {bench_cmd}",
                 perf_fifo.ctl_fifo_path.to_string_lossy(),
                 perf_fifo.ack_fifo_path.to_string_lossy(),
                 perf_file.path().to_string_lossy()
@@ -109,12 +125,6 @@ impl PerfRunner {
     pub async fn save_files_to(&self, profile_folder: &PathBuf) -> anyhow::Result<()> {
         let start = std::time::Instant::now();
 
-        let bench_data = self
-            .benchmark_data
-            .get()
-            .expect("Benchmark order is not available");
-        bench_data.save_to(profile_folder).unwrap();
-
         // Copy the perf data files to the profile folder
         let copy_tasks = std::fs::read_dir(&self.perf_dir)?
             .filter_map(|entry| entry.ok())
@@ -141,16 +151,33 @@ impl PerfRunner {
                     let dst_path = profile_folder.join(dst_file_name);
                     tokio::fs::copy(src_path, dst_path).await?;
 
-                    Ok::<_, anyhow::Error>(())
+                    Ok::<_, anyhow::Error>(pid)
                 })
             })
             .collect::<FuturesUnordered<_>>();
+
+        let bench_data = self
+            .benchmark_data
+            .get()
+            .expect("Benchmark order is not available");
         assert_eq!(
             copy_tasks.len(),
             bench_data.bench_count(),
             "Benchmark count mismatch"
         );
-        futures::future::try_join_all(copy_tasks).await?;
+
+        // Harvest the perf maps generated by python. This will copy the perf
+        // maps from /tmp to the profile folder. We have to write our own perf
+        // maps to these files AFTERWARDS, otherwise it'll be overwritten!
+        let perf_map_pids = futures::future::try_join_all(copy_tasks)
+            .await?
+            .into_iter()
+            .filter_map(Result::ok)
+            .collect::<HashSet<_>>();
+        harvest_perf_maps_for_pids(profile_folder, &perf_map_pids).await?;
+
+        // Append perf maps, unwind info and other metadata
+        bench_data.save_to(profile_folder).unwrap();
 
         let elapsed = start.elapsed();
         debug!("Perf teardown took: {:?}", elapsed);
@@ -165,6 +192,7 @@ impl PerfRunner {
         let mut bench_order_by_pid = HashMap::<u32, Vec<String>>::new();
         let mut symbols_by_pid = HashMap::<u32, ProcessSymbols>::new();
         let mut unwind_data_by_pid = HashMap::<u32, Vec<UnwindData>>::new();
+        let mut integration = None;
 
         loop {
             let perf_ping = tokio::time::timeout(Duration::from_secs(1), perf_fifo.ping()).await;
@@ -230,11 +258,24 @@ impl PerfRunner {
                     perf_fifo.stop_events().await?;
                     runner_fifo.send_cmd(FifoCommand::Ack).await?;
                 }
+                FifoCommand::PingPerf => {
+                    if perf_fifo.ping().await.is_ok() {
+                        runner_fifo.send_cmd(FifoCommand::Ack).await?;
+                    } else {
+                        runner_fifo.send_cmd(FifoCommand::Err).await?;
+                    }
+                }
+                FifoCommand::SetIntegration { name, version } => {
+                    integration = Some((name, version));
+                    runner_fifo.send_cmd(FifoCommand::Ack).await?;
+                }
                 FifoCommand::Ack => unreachable!(),
+                FifoCommand::Err => unreachable!(),
             }
         }
 
         Ok(BenchmarkData {
+            integration: integration.context("Couldn't find integration metadata")?,
             bench_order_by_pid,
             symbols_by_pid,
             unwind_data_by_pid,
@@ -243,6 +284,9 @@ impl PerfRunner {
 }
 
 pub struct BenchmarkData {
+    /// Name and version of the integration
+    integration: (String, String),
+
     bench_order_by_pid: HashMap<u32, Vec<String>>,
     symbols_by_pid: HashMap<u32, ProcessSymbols>,
     unwind_data_by_pid: HashMap<u32, Vec<UnwindData>>,
@@ -260,11 +304,32 @@ impl BenchmarkData {
             }
         }
 
-        for (pid, orders) in &self.bench_order_by_pid {
-            let dst_file_name = format!("{}_.bench_order", pid);
-            let dst_path = path.as_ref().join(dst_file_name);
-            std::fs::write(dst_path, orders.join("\n"))?;
-        }
+        let metadata = PerfMetadata {
+            integration: self.integration.clone(),
+            bench_order_by_pid: self.bench_order_by_pid.clone(),
+            ignored_modules: {
+                let mut to_ignore = vec![];
+
+                // Check if any of the ignored modules has been loaded in the process
+                for ignore_path in get_objects_path_to_ignore() {
+                    for proc in self.symbols_by_pid.values() {
+                        if let Some(mapping) = proc.module_mapping(&ignore_path) {
+                            let (Some((base_addr, _)), Some((_, end_addr))) = (
+                                mapping.iter().min_by_key(|(base_addr, _)| base_addr),
+                                mapping.iter().max_by_key(|(_, end_addr)| end_addr),
+                            ) else {
+                                continue;
+                            };
+
+                            to_ignore.push((ignore_path.clone(), *base_addr, *end_addr));
+                        }
+                    }
+                }
+
+                to_ignore
+            },
+        };
+        metadata.save_to(&path).unwrap();
 
         Ok(())
     }

src/run/runner/wall_time/perf/perf_map.rs

@@ -79,17 +79,17 @@ impl ModuleSymbols {
 
 /// Represents all the modules inside a process and their symbols.
 pub struct ProcessSymbols {
-    modules_bounds: HashMap<PathBuf, Vec<(u64, u64)>>,
-    modules: HashMap<PathBuf, ModuleSymbols>,
     pid: u32,
+    module_mappings: HashMap<PathBuf, Vec<(u64, u64)>>,
+    modules: HashMap<PathBuf, ModuleSymbols>,
 }
 
 impl ProcessSymbols {
     pub fn new(pid: u32) -> Self {
         Self {
-            modules_bounds: HashMap::new(),
-            modules: HashMap::new(),
             pid,
+            module_mappings: HashMap::new(),
+            modules: HashMap::new(),
         }
     }
 
@@ -119,12 +119,21 @@ impl ProcessSymbols {
             }
         }
 
-        self.modules_bounds
+        self.module_mappings
             .entry(path.clone())
             .or_default()
             .push((start_addr, end_addr));
     }
 
+    pub fn module_mapping<P: AsRef<std::path::Path>>(
+        &self,
+        module_path: P,
+    ) -> Option<&[(u64, u64)]> {
+        self.module_mappings
+            .get(module_path.as_ref())
+            .map(|bounds| bounds.as_slice())
+    }
+
     pub fn save_to<P: AsRef<std::path::Path>>(&self, folder: P) -> anyhow::Result<()> {
         if self.modules.is_empty() {
             return Ok(());
@@ -133,7 +142,7 @@ impl ProcessSymbols {
         let symbols_path = folder.as_ref().join(format!("perf-{}.map", self.pid));
         for module in self.modules.values() {
             let Some((base_addr, _)) = self
-                .modules_bounds
+                .module_mappings
                 .get(&module.path)
                 .and_then(|bounds| bounds.iter().min_by_key(|(start, _)| start))
             else {

src/run/runner/wall_time/perf/shared.rs

@@ -12,6 +12,9 @@ pub enum Command {
     StartBenchmark,
     StopBenchmark,
     Ack,
+    PingPerf,
+    SetIntegration { name: String, version: String },
+    Err,
 }
 //
 // !!!!!!!!!!!!!!!!!!!!!!!!