Method handle_fifo_messages

src/executor/shared/fifo.rs:163–287

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

(
        &mut self,
        child: &mut std::process::Child,
        mut handle_cmd: impl AsyncFnMut(&FifoCommand) -> anyhow::Result<Option<FifoCommand>>,
    )

    ///
    /// Returns execution timestamps, benchmark data, and the exit status of the child process.
    pub async fn handle_fifo_messages(
        &mut self,
        child: &mut std::process::Child,
        mut handle_cmd: impl AsyncFnMut(&FifoCommand) -> anyhow::Result<Option<FifoCommand>>,
    ) -> 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();

        let mut integration = None;

        // Must match the clock used by the benchmarked process so timestamps
        // from both sides are comparable.
        let get_current_time = instrument_hooks_bindings::InstrumentHooks::current_timestamp;

        let mut benchmark_started = false;

        // Outer loop: continues until health check fails
        loop {
            // Inner loop: process commands until timeout/error
            loop {
                let result: Result<_, Elapsed> =
                    tokio::time::timeout(Duration::from_secs(1), self.recv_cmd()).await;
                let cmd = match result {
                    Ok(Ok(cmd)) => cmd,
                    Ok(Err(e)) => {
                        warn!("Failed to parse FIFO command: {e}");
                        break;
                    }
                    Err(_) => break, // Timeout
                };
                trace!("Received command: {cmd:?}");

                // Try executor-specific handler first
                if let Some(response) = handle_cmd(&cmd).await? {
                    self.send_cmd(response).await?;
                    continue;
                }

                // Fall through to shared implementation for standard commands
                match &cmd {
                    FifoCommand::CurrentBenchmark { pid, uri } => {
                        bench_order_by_timestamp.push((get_current_time(), uri.to_string()));
                        bench_pids.insert(*pid);
                        self.send_cmd(FifoCommand::Ack).await?;
                    }
                    FifoCommand::StartProfiler => {
                        if !benchmark_started {
                            benchmark_started = true;
                            markers.push(MarkerType::SampleStart(get_current_time()));
                        } else {
                            warn!("Received duplicate StartProfiler command, ignoring");
                        }
                        self.send_cmd(FifoCommand::Ack).await?;