fix(runner): use child process instead of pid/perf fifo for FIFO health check

Author: not-matthias

src/executor/helpers/run_command_with_log_pipe.rs

@@ -11,7 +11,7 @@ use std::thread;
 ///
 /// # Arguments
 /// - `cmd`: The command to run.
-/// - `cb`: A callback function that takes the process ID and returns a result.
+/// - `cb`: A callback function that takes the process and returns the exit status.
 ///
 /// # Returns
 ///
@@ -22,8 +22,8 @@ pub async fn run_command_with_log_pipe_and_callback<F, Fut>(
     cb: F,
 ) -> Result<ExitStatus>
 where
-    F: FnOnce(u32) -> Fut,
-    Fut: Future<Output = anyhow::Result<()>>,
+    F: FnOnce(std::process::Child) -> Fut,
+    Fut: Future<Output = anyhow::Result<ExitStatus>>,
 {
     fn log_tee(
         mut reader: impl Read,
@@ -93,11 +93,9 @@ where
         log_tee(stderr, std::io::stderr(), Some("[stderr]")).unwrap();
     });
 
-    cb(process.id()).await?;
-
-    process.wait().context("failed to wait for the process")
+    cb(process).await
 }
 
 pub async fn run_command_with_log_pipe(cmd: Command) -> Result<ExitStatus> {
-    run_command_with_log_pipe_and_callback(cmd, async |_| Ok(())).await
+    run_command_with_log_pipe_and_callback(cmd, |mut child| async move { Ok(child.wait()?) }).await
 }

src/executor/memory/executor.rs

@@ -98,15 +98,16 @@ impl Executor for MemoryExecutor {
         debug!("cmd: {cmd:?}");
 
         let runner_fifo = RunnerFifo::new()?;
-        let on_process_started = async |pid| -> anyhow::Result<()> {
-            let marker_result = Self::handle_fifo(runner_fifo, pid, ipc).await?;
+        let on_process_started = |mut child: std::process::Child| async move {
+            let (marker_result, exit_status) =
+                Self::handle_fifo(runner_fifo, ipc, &mut child).await?;
 
             // Directly write to the profile folder, to avoid having to define another field
             marker_result
                 .save_to(execution_context.profile_folder.join("results"))
                 .unwrap();
 
-            Ok(())
+            Ok(exit_status)
         };
 
         let status = run_command_with_log_pipe_and_callback(cmd, on_process_started).await?;
@@ -152,11 +153,9 @@ impl Executor for MemoryExecutor {
 impl MemoryExecutor {
     async fn handle_fifo(
         mut runner_fifo: RunnerFifo,
-        pid: u32,
         ipc: MemtrackIpcServer,
-    ) -> anyhow::Result<ExecutionTimestamps> {
-        debug!("handle_fifo called with PID {pid}");
-
+        child: &mut std::process::Child,
+    ) -> anyhow::Result<(ExecutionTimestamps, std::process::ExitStatus)> {
         // Accept the IPC connection from memtrack and get the sender it sends us
         // Use a timeout to prevent hanging if the process doesn't start properly
         // https://github.com/servo/ipc-channel/issues/261
@@ -170,15 +169,6 @@ impl MemoryExecutor {
         .context("Timeout waiting for IPC connection from memtrack process")??;
         let ipc_client = Rc::new(MemtrackIpcClient::from_accepted(memtrack_sender));
 
-        let ipc_client_health = Rc::clone(&ipc_client);
-        let health_check = async move || {
-            // Ping memtrack via IPC to check if it's still responding
-            match ipc_client_health.ping() {
-                Ok(()) => Ok(true),
-                Err(_) => Ok(false),
-            }
-        };
-
         let on_cmd = async move |cmd: &FifoCommand| {
             const INVALID_INTEGRATION_ERROR: &str = "This integration doesn't support memory profiling. Please update your integration to a version that supports memory profiling.";
 
@@ -234,9 +224,9 @@ impl MemoryExecutor {
             Ok(None)
         };
 
-        let (marker_result, _) = runner_fifo
-            .handle_fifo_messages(health_check, on_cmd)
-            .await?;
-        Ok(marker_result)
+        let (marker_result, _, exit_status) =
+            runner_fifo.handle_fifo_messages(child, on_cmd).await?;
+
+        Ok((marker_result, exit_status))
     }
 }

src/executor/shared/fifo.rs

@@ -131,17 +131,22 @@ impl RunnerFifo {
         Ok(())
     }
 
-    /// Handles all incoming FIFO messages until it's closed, or until the health check closure
-    /// returns `false` or an error.
+    /// Handles all incoming FIFO messages until it's closed, or until the child process exits.
     ///
     /// The `handle_cmd` callback is invoked first for each command. If it returns `Some(response)`,
     /// that response is sent and the shared implementation is skipped. If it returns `None`,
     /// the command falls through to the shared implementation for standard handling.
+    ///
+    /// Returns execution timestamps, benchmark data, and the exit status of the child process.
     pub async fn handle_fifo_messages(
         &mut self,
-        mut health_check: impl AsyncFnMut() -> anyhow::Result<bool>,
+        child: &mut std::process::Child,
         mut handle_cmd: impl AsyncFnMut(&FifoCommand) -> anyhow::Result<Option<FifoCommand>>,
-    ) -> anyhow::Result<(ExecutionTimestamps, FifoBenchmarkData)> {
+    ) -> anyhow::Result<(
+        ExecutionTimestamps,
+        FifoBenchmarkData,
+        std::process::ExitStatus,
+    )> {
         let mut bench_order_by_timestamp = Vec::<(u64, String)>::new();
         let mut bench_pids = HashSet::<pid_t>::new();
         let mut markers = Vec::<MarkerType>::new();
@@ -241,20 +246,24 @@ impl RunnerFifo {
                 }
             }
 
-            let is_alive = health_check().await?;
-            if !is_alive {
-                debug!("Process terminated, stopping the command handler");
-                break;
+            // Check if the process has exited using try_wait (non-blocking)
+            match child.try_wait() {
+                Ok(None) => {} // Still running, continue loop
+                Ok(Some(exit_status)) => {
+                    debug!(
+                        "Process terminated with status: {exit_status}, stopping the command handler"
+                    );
+                    let marker_result =
+                        ExecutionTimestamps::new(&bench_order_by_timestamp, &markers);
+                    let fifo_data = FifoBenchmarkData {
+                        integration,
+                        bench_pids,
+                    };
+                    return Ok((marker_result, fifo_data, exit_status));
+                }
+                Err(e) => return Err(anyhow::Error::from(e)),
             }
         }
-
-        let marker_result = ExecutionTimestamps::new(&bench_order_by_timestamp, &markers);
-        let fifo_data = FifoBenchmarkData {
-            integration,
-            bench_pids,
-        };
-
-        Ok((marker_result, fifo_data))
     }
 }
 

src/executor/wall_time/perf/mod.rs

@@ -31,10 +31,7 @@ use runner_shared::metadata::PerfMetadata;
 use runner_shared::unwind_data::UnwindData;
 use std::path::Path;
 use std::path::PathBuf;
-use std::sync::Arc;
-use std::time::Duration;
 use std::{cell::OnceCell, collections::HashMap, process::ExitStatus};
-use tokio::sync::Mutex;
 
 mod jit_dump;
 mod memory_mappings;
@@ -205,14 +202,15 @@ impl PerfRunner {
         let cmd = wrap_with_sudo(wrapped_builder)?.build();
         debug!("cmd: {cmd:?}");
 
-        let on_process_started = async |_| -> anyhow::Result<()> {
+        let on_process_started = |mut child: std::process::Child| async move {
             // If we output pipedata, we do not parse the perf map during teardown yet, so we need to parse memory
             // maps as we receive the `CurrentBenchmark` fifo commands.
-            let data = Self::handle_fifo(runner_fifo, perf_fifo, self.output_pipedata).await?;
+            let (data, exit_status) =
+                Self::handle_fifo(runner_fifo, perf_fifo, self.output_pipedata, &mut child).await?;
             self.benchmark_data.set(data).unwrap_or_else(|_| {
                 error!("Failed to set benchmark data in PerfRunner");
             });
-            Ok(())
+            Ok(exit_status)
         };
         run_command_with_log_pipe_and_callback(cmd, on_process_started).await
     }
@@ -248,37 +246,21 @@ impl PerfRunner {
 
     async fn handle_fifo(
         mut runner_fifo: RunnerFifo,
-        perf_fifo: PerfFifo,
+        mut perf_fifo: PerfFifo,
         parse_memory_maps: bool,
-    ) -> anyhow::Result<BenchmarkData> {
+        child: &mut std::process::Child,
+    ) -> anyhow::Result<(BenchmarkData, std::process::ExitStatus)> {
         let mut symbols_by_pid = HashMap::<pid_t, ProcessSymbols>::new();
         let mut unwind_data_by_pid = HashMap::<pid_t, Vec<UnwindData>>::new();
 
-        let perf_fifo = Arc::new(Mutex::new(perf_fifo));
-        let mut perf_ping_timeout = 5;
-        let health_check = async || {
-            let perf_ping = tokio::time::timeout(Duration::from_secs(perf_ping_timeout), async {
-                perf_fifo.lock().await.ping().await
-            })
-            .await;
-            if let Ok(Err(_)) | Err(_) = perf_ping {
-                debug!("Failed to ping perf FIFO, ending perf fifo loop");
-                return Ok(false);
-            }
-            // Perf has started successfully, we can decrease the timeout for future pings
-            perf_ping_timeout = 1;
-
-            Ok(true)
-        };
-
         let on_cmd = async |cmd: &FifoCommand| {
             #[allow(deprecated)]
             match cmd {
                 FifoCommand::StartBenchmark => {
-                    perf_fifo.lock().await.start_events().await?;
+                    perf_fifo.start_events().await?;
                 }
                 FifoCommand::StopBenchmark => {
-                    perf_fifo.lock().await.stop_events().await?;
+                    perf_fifo.stop_events().await?;
                 }
                 FifoCommand::CurrentBenchmark { pid, .. } => {
                     #[cfg(target_os = "linux")]
@@ -294,7 +276,7 @@ impl PerfRunner {
                     }
                 }
                 FifoCommand::PingPerf => {
-                    if perf_fifo.lock().await.ping().await.is_err() {
+                    if perf_fifo.ping().await.is_err() {
                         return Ok(Some(FifoCommand::Err));
                     }
                     return Ok(Some(FifoCommand::Ack));
@@ -310,15 +292,18 @@ impl PerfRunner {
             Ok(None)
         };
 
-        let (marker_result, fifo_data) = runner_fifo
-            .handle_fifo_messages(health_check, on_cmd)
-            .await?;
-        Ok(BenchmarkData {
-            fifo_data,
-            marker_result,
-            symbols_by_pid,
-            unwind_data_by_pid,
-        })
+        let (marker_result, fifo_data, exit_status) =
+            runner_fifo.handle_fifo_messages(child, on_cmd).await?;
+
+        Ok((
+            BenchmarkData {
+                fifo_data,
+                marker_result,
+                symbols_by_pid,
+                unwind_data_by_pid,
+            },
+            exit_status,
+        ))
     }
 
     fn get_perf_file_path<P: AsRef<Path>>(&self, profile_folder: P) -> PathBuf {