refactor(walltime): port PerfRunner to Profiler trait

PerfRunner becomes PerfProfiler implementing the Profiler trait. The
RunnerFifo loop moves out of the perf module and into WallTimeExecutor,
which now drives the trait callbacks (on_start_benchmark / on_stop_benchmark
/ on_ping / GetIntegrationMode) generically.

Profiler::wrap stashes the perf control fifo and output path on the
profiler instance; finalize harvests the perf.data artifacts and writes
walltime.metadata. The OnceCell<BenchmarkData> bridge is gone — fifo
data and timestamps are passed directly into finalize.

Executor::run now takes &mut self to allow profilers to hold per-run
state on the trait object. This cascades through run_executor,
orchestrator, and the Memory/Valgrind executors (no behavioral
change for those — they don't mutate self).

Author: not-matthias

src/executor/memory/executor.rs

@@ -93,7 +93,7 @@ impl Executor for MemoryExecutor {
     }
 
     async fn run(
-        &self,
+        &mut self,
         execution_context: &ExecutionContext,
         _mongo_tracer: &Option<MongoTracer>,
     ) -> Result<()> {

src/executor/mod.rs

@@ -106,7 +106,7 @@ pub trait Executor {
 
     /// Runs the executor
     async fn run(
-        &self,
+        &mut self,
         execution_context: &ExecutionContext,
         // TODO: use Instruments instead of directly passing the mongodb tracer
         mongo_tracer: &Option<MongoTracer>,
@@ -118,7 +118,7 @@ pub trait Executor {
 /// Run a single executor: setup → run → teardown → persist logs.
 /// Does NOT upload.
 pub async fn run_executor(
-    executor: &dyn Executor,
+    executor: &mut dyn Executor,
     orchestrator: &Orchestrator,
     execution_context: &ExecutionContext,
     setup_cache_dir: Option<&Path>,

src/executor/orchestrator.rs

@@ -157,7 +157,7 @@ impl Orchestrator {
             let config = self
                 .config
                 .executor_config_for_command(part.command, !part.uses_exec_harness);
-            let executor = get_executor_from_mode(part.mode);
+            let mut executor = get_executor_from_mode(part.mode);
             let profile_folder =
                 self.resolve_profile_folder(&executor.name(), run_part_index, total_parts)?;
 
@@ -167,7 +167,7 @@ impl Orchestrator {
                 activate_rolling_buffer(&part.label);
             }
 
-            run_executor(executor.as_ref(), self, &ctx, setup_cache_dir).await?;
+            run_executor(executor.as_mut(), self, &ctx, setup_cache_dir).await?;
 
             if !self.config.show_full_output {
                 deactivate_rolling_buffer();

src/executor/tests.rs

@@ -157,20 +157,19 @@ mod valgrind {
     use super::helpers::*;
     use crate::executor::valgrind::executor::ValgrindExecutor;
 
-    async fn get_valgrind_executor() -> (SemaphorePermit<'static>, &'static ValgrindExecutor) {
-        static VALGRIND_EXECUTOR: OnceCell<ValgrindExecutor> = OnceCell::const_new();
+    async fn get_valgrind_executor() -> (SemaphorePermit<'static>, ValgrindExecutor) {
+        static VALGRIND_SETUP: OnceCell<()> = OnceCell::const_new();
 
-        let executor = VALGRIND_EXECUTOR
+        VALGRIND_SETUP
             .get_or_init(|| async {
                 let executor = ValgrindExecutor;
                 let system_info = SystemInfo::new().unwrap();
                 executor.setup(&system_info, None).await.unwrap();
-                executor
             })
             .await;
         let _lock = acquire_bpf_instrumentation_lock().await;
 
-        (_lock, executor)
+        (_lock, ValgrindExecutor)
     }
 
     fn valgrind_config(command: &str) -> ExecutorConfig {
@@ -183,7 +182,7 @@ mod valgrind {
     #[apply(test_cases)]
     #[test_log::test(tokio::test)]
     async fn test_valgrind_executor(#[case] cmd: &str) {
-        let (_lock, executor) = get_valgrind_executor().await;
+        let (_lock, mut executor) = get_valgrind_executor().await;
 
         let config = valgrind_config(cmd);
         // Unset GITHUB_ACTIONS to force LocalProvider which supports repository_override
@@ -197,7 +196,7 @@ mod valgrind {
     #[apply(env_test_cases)]
     #[test_log::test(tokio::test)]
     async fn test_valgrind_executor_with_env(#[case] env_case: (&str, &str)) {
-        let (_lock, executor) = get_valgrind_executor().await;
+        let (_lock, mut executor) = get_valgrind_executor().await;
 
         let (env_var, env_value) = env_case;
         temp_env::async_with_vars(
@@ -252,7 +251,7 @@ mod walltime {
     #[rstest::rstest]
     #[test_log::test(tokio::test)]
     async fn test_walltime_executor(#[case] cmd: &str, #[values(false, true)] enable_perf: bool) {
-        let (_permit, executor) = get_walltime_executor().await;
+        let (_permit, mut executor) = get_walltime_executor().await;
 
         let config = walltime_config(cmd, enable_perf);
         // Unset GITHUB_ACTIONS to force LocalProvider which supports repository_override
@@ -270,7 +269,7 @@ mod walltime {
         #[case] env_case: (&str, &str),
         #[values(false, true)] enable_perf: bool,
     ) {
-        let (_permit, executor) = get_walltime_executor().await;
+        let (_permit, mut executor) = get_walltime_executor().await;
 
         let (env_var, env_value) = env_case;
         temp_env::async_with_vars(
@@ -289,7 +288,7 @@ mod walltime {
     #[rstest::rstest]
     #[test_log::test(tokio::test)]
     async fn test_walltime_executor_in_working_dir(#[values(false, true)] enable_perf: bool) {
-        let (_permit, executor) = get_walltime_executor().await;
+        let (_permit, mut executor) = get_walltime_executor().await;
 
         let cmd = r#"
 if [ "$(basename "$(pwd)")" != "within_sub_directory" ]; then
@@ -321,7 +320,7 @@ fi
     #[rstest::rstest]
     #[test_log::test(tokio::test)]
     async fn test_walltime_executor_fails(#[values(false, true)] enable_perf: bool) {
-        let (_permit, executor) = get_walltime_executor().await;
+        let (_permit, mut executor) = get_walltime_executor().await;
 
         let config = walltime_config("exit 1", enable_perf);
         // Unset GITHUB_ACTIONS to force LocalProvider which supports repository_override
@@ -361,7 +360,7 @@ fi
     async fn test_exec_harness(#[case] cmd: &str) {
         use exec_harness::walltime::WalltimeExecutionArgs;
 
-        let (_permit, executor) = get_walltime_executor().await;
+        let (_permit, mut executor) = get_walltime_executor().await;
 
         let walltime_args = WalltimeExecutionArgs {
             warmup_time: Some("0s".to_string()),
@@ -427,7 +426,7 @@ mod memory {
     #[apply(test_cases)]
     #[test_log::test(tokio::test)]
     async fn test_memory_executor(#[case] cmd: &str) {
-        let (_permit, _lock, executor) = get_memory_executor().await;
+        let (_permit, _lock, mut executor) = get_memory_executor().await;
 
         // Unset GITHUB_ACTIONS to force LocalProvider which supports repository_override
         temp_env::async_with_vars(&[("GITHUB_ACTIONS", None::<&str>)], async {
@@ -441,7 +440,7 @@ mod memory {
     #[apply(env_test_cases)]
     #[test_log::test(tokio::test)]
     async fn test_memory_executor_with_env(#[case] env_case: (&str, &str)) {
-        let (_permit, _lock, executor) = get_memory_executor().await;
+        let (_permit, _lock, mut executor) = get_memory_executor().await;
 
         let (env_var, env_value) = env_case;
         temp_env::async_with_vars(
@@ -473,7 +472,7 @@ fi
         );
         let config = memory_config(&cmd);
         let (execution_context, _temp_dir) = create_test_setup(config).await;
-        let (_permit, _lock, executor) = get_memory_executor().await;
+        let (_permit, _lock, mut executor) = get_memory_executor().await;
 
         temp_env::async_with_vars(&[("PATH", Some(&modified_path))], async {
             executor.run(&execution_context, &None).await.unwrap();

src/executor/valgrind/executor.rs

@@ -50,7 +50,7 @@ impl Executor for ValgrindExecutor {
     }
 
     async fn run(
-        &self,
+        &mut self,
         execution_context: &ExecutionContext,
         mongo_tracer: &Option<MongoTracer>,
     ) -> Result<()> {

src/executor/wall_time/executor.rs

@@ -1,21 +1,26 @@
 use super::helpers::validate_walltime_results;
 use super::isolation::wrap_with_isolation;
-use super::profiler::perf::PerfRunner;
+use super::profiler::Profiler;
+use super::profiler::perf::PerfProfiler;
 use crate::executor::Executor;
 use crate::executor::ExecutorConfig;
 use crate::executor::ToolStatus;
 use crate::executor::helpers::command::CommandBuilder;
 use crate::executor::helpers::env::{build_path_env, get_base_injected_env};
 use crate::executor::helpers::get_bench_command::get_bench_command;
 use crate::executor::helpers::run_command_with_log_pipe::run_command_with_log_pipe;
+use crate::executor::helpers::run_command_with_log_pipe::run_command_with_log_pipe_and_callback;
 use crate::executor::helpers::run_with_env::wrap_with_env;
 use crate::executor::helpers::run_with_sudo::wrap_with_sudo;
+use crate::executor::shared::fifo::RunnerFifo;
 use crate::executor::{ExecutionContext, ExecutorName, ExecutorSupport};
 use crate::instruments::mongo_tracer::MongoTracer;
 use crate::prelude::*;
 use crate::runner_mode::RunnerMode;
 use crate::system::{SupportedOs, SystemInfo};
 use async_trait::async_trait;
+use runner_shared::fifo::Command as FifoCommand;
+use runner_shared::fifo::IntegrationMode;
 use std::fs::canonicalize;
 use std::io::Write;
 use std::path::Path;
@@ -72,14 +77,17 @@ impl Drop for HookScriptsGuard {
 }
 
 pub struct WallTimeExecutor {
-    perf: Option<PerfRunner>,
+    profiler: Option<Box<dyn Profiler>>,
 }
 
 impl WallTimeExecutor {
     pub fn new() -> Self {
-        Self {
-            perf: cfg!(target_os = "linux").then(PerfRunner::new),
-        }
+        let profiler: Option<Box<dyn Profiler>> = if cfg!(target_os = "linux") {
+            Some(Box::new(PerfProfiler::new()))
+        } else {
+            None
+        };
+        Self { profiler }
     }
 
     fn walltime_bench_cmd(
@@ -122,9 +130,7 @@ impl Executor for WallTimeExecutor {
     }
 
     fn tool_status(&self) -> Option<ToolStatus> {
-        self.perf
-            .as_ref()
-            .map(|_| super::profiler::perf::setup::get_perf_status())
+        self.profiler.as_ref().and_then(|p| p.tool_status())
     }
 
     fn support_level(&self, system_info: &SystemInfo) -> ExecutorSupport {
@@ -136,33 +142,33 @@ impl Executor for WallTimeExecutor {
     }
 
     async fn setup(&self, system_info: &SystemInfo, setup_cache_dir: Option<&Path>) -> Result<()> {
-        if self.perf.is_some() {
-            return PerfRunner::setup_environment(system_info, setup_cache_dir).await;
+        if let Some(profiler) = &self.profiler {
+            profiler.setup(system_info, setup_cache_dir).await?;
         }
-
         Ok(())
     }
 
     async fn run(
-        &self,
+        &mut self,
         execution_context: &ExecutionContext,
         _mongo_tracer: &Option<MongoTracer>,
     ) -> Result<()> {
-        let status = {
-            let _guard = HookScriptsGuard::setup();
-
-            let (_env_file, _script_file, cmd_builder) =
-                WallTimeExecutor::walltime_bench_cmd(&execution_context.config, execution_context)?;
-            if let Some(perf) = &self.perf
-                && execution_context.config.enable_perf
-            {
-                perf.run(
+        let _guard = HookScriptsGuard::setup();
+
+        let (_env_file, _script_file, cmd_builder) =
+            WallTimeExecutor::walltime_bench_cmd(&execution_context.config, execution_context)?;
+
+        let status = match self.profiler.as_mut() {
+            Some(profiler) if execution_context.config.enable_perf => {
+                run_with_profiler(
+                    profiler.as_mut(),
                     cmd_builder,
                     &execution_context.config,
                     &execution_context.profile_folder,
                 )
                 .await
-            } else {
+            }
+            _ => {
                 let cmd_builder = if cfg!(target_os = "linux") {
                     wrap_with_sudo(cmd_builder)?
                 } else {
@@ -187,13 +193,6 @@ impl Executor for WallTimeExecutor {
     async fn teardown(&self, execution_context: &ExecutionContext) -> Result<()> {
         debug!("Copying files to the profile folder");
 
-        if let Some(perf) = &self.perf
-            && execution_context.config.enable_perf
-        {
-            perf.save_files_to(&execution_context.profile_folder)
-                .await?;
-        }
-
         validate_walltime_results(
             &execution_context.profile_folder,
             execution_context.config.allow_empty,
@@ -203,6 +202,55 @@ impl Executor for WallTimeExecutor {
     }
 }
 
+/// Drive a single benchmark run through a [`Profiler`]: wrap the command,
+/// spawn it, dispatch FIFO commands from the integration into the profiler's
+/// hooks, and finalize once the child exits.
+async fn run_with_profiler(
+    profiler: &mut dyn Profiler,
+    cmd_builder: CommandBuilder,
+    config: &ExecutorConfig,
+    profile_folder: &Path,
+) -> Result<std::process::ExitStatus> {
+    let wrapped = profiler.wrap(cmd_builder, config, profile_folder).await?;
+    let cmd = wrapped.build();
+    debug!("cmd: {cmd:?}");
+
+    let mut runner_fifo = RunnerFifo::new()?;
+
+    run_command_with_log_pipe_and_callback(cmd, async move |mut child| {
+        let on_cmd = async |c: &FifoCommand| match c {
+            FifoCommand::StartBenchmark => {
+                profiler.on_start_benchmark().await?;
+                Ok(None)
+            }
+            FifoCommand::StopBenchmark => {
+                profiler.on_stop_benchmark().await?;
+                Ok(None)
+            }
+            #[allow(deprecated)]
+            FifoCommand::PingProfiler => Ok(Some(if profiler.on_ping().await? {
+                FifoCommand::Ack
+            } else {
+                FifoCommand::Err
+            })),
+            FifoCommand::GetIntegrationMode => Ok(Some(FifoCommand::IntegrationModeResponse(
+                IntegrationMode::Walltime,
+            ))),
+            _ => Ok(None),
+        };
+
+        let (timestamps, fifo_data, exit_status) =
+            runner_fifo.handle_fifo_messages(&mut child, on_cmd).await?;
+
+        profiler
+            .finalize(fifo_data, timestamps, profile_folder)
+            .await?;
+
+        Ok(exit_status)
+    })
+    .await
+}
+
 #[cfg(test)]
 mod tests {
     use tempfile::NamedTempFile;