evobench_tools/run/

run_job.rs

//! Running a benchmarking job

use std::{
    fs::{create_dir_all, remove_file},
    io::{Write, stderr},
    ops::Deref,
    os::unix::fs::symlink,
    path::{Path, PathBuf},
    sync::{Arc, Mutex},
};

use anyhow::{Result, bail};
use chrono::{DateTime, Local};
use cj_path_util::path_util::{AppendToPath, rename_tmp_path};
use itertools::Itertools;
use nix::unistd::getpid;
use regex::Regex;

use crate::{
    config_file::ron_to_string_pretty,
    ctx,
    git::GitHash,
    git_tags::GitTags,
    html_files::write_redirect_html_file,
    info,
    io_utils::output_capture_log::{CaptureOptions, OutputCaptureLog},
    run::{
        bench_tmp_dir::bench_tmp_dir,
        benchmarking_job::BenchmarkingJob,
        config::{RunConfig, ShareableConfig},
        dataset_dir_env_var::dataset_dir_for,
        env_vars::assert_evobench_env_var,
        key::{BenchmarkingJobParameters, RunParameters},
        output_directory::{
            post_process::compress_file_as,
            structure::{KeyDir, ReplaceBasePath, RunDir, ToPath},
        },
        run_queues::RunQueuesData,
        versioned_dataset_dir::VersionedDatasetDir,
        working_directory::{FetchTags, FetchedTags, WorkingDirectory},
    },
    serde_types::{date_and_time::DateTimeWithOffset, proper_dirname::ProperDirname},
    utillib::{
        arc::CloneArc,
        cleanup_daemon::{CleanupHandler, Deletion},
        escaped_display::AsEscapedString,
        into_arc_path::IntoArcPath,
        logging::{LogLevel, log_level},
    },
};

use super::{
    config::{BenchmarkingCommand, ScheduleCondition},
    working_directory_pool::{WorkingDirectoryId, WorkingDirectoryPool},
};

/// Get the string for the `COMMIT_TAGS` env var, e.g. "" or
/// "foo,v1.2.3". Wants to be assured that `git fetch --tags` was run
/// (see methods that return a `FetchedTags`).
pub fn get_commit_tags(
    working_dir: &WorkingDirectory,
    commit_id: &GitHash,
    re: &Regex,
    fetched_tags: FetchedTags,
) -> Result<String> {
    if fetched_tags != FetchedTags::Yes {
        bail!("need up to date tags, but got {fetched_tags:?}")
    }

    let git_working_dir = &working_dir.git_working_dir;

    let git_tags = GitTags::from_dir(git_working_dir)?;
    // (Huh, `let s = ` being required here makes
    // no sense to me. rustc 1.90.0)
    let s = git_tags
        .get_by_commit(&commit_id)
        .filter(|s| re.is_match(s))
        .join(",");
    Ok(s)
}

/// The context for running a job (information that should not be part
/// of `Key`). Only used for one run, as some fields
/// (e.g. `timestamp`) change. Independent of jobs: this context is
/// bundled before selecting the job to run.
pub struct JobRunner<'pool> {
    pub working_directory_pool: &'pool mut WorkingDirectoryPool,
    pub output_base_dir: &'pool Arc<Path>,
    /// The timestamp for this run.
    pub timestamp: DateTimeWithOffset,
    // Separate lifetime?
    pub shareable_config: &'pool ShareableConfig,
    // ditto?
    pub versioned_dataset_dir: &'pool VersionedDatasetDir,
    pub file_cleanup_handler: &'pool CleanupHandler,
}

impl<'pool> JobRunner<'pool> {
    pub fn timestamp_local(&self) -> DateTime<Local> {
        // (A little bit costly.)
        self.timestamp.to_datetime().into()
    }

    pub fn run_config(&self) -> &'pool RunConfig {
        &self.shareable_config.run_config
    }
}

pub struct JobRunnerJobData<'run_queues, 'j, 's> {
    pub job: &'j BenchmarkingJob,
    pub run_queues_data: &'s RunQueuesData<'run_queues>,
}

pub struct JobRunnerWithJob<'pool, 'run_queues, 'j, 's> {
    pub job_runner: JobRunner<'pool>,
    pub job_data: JobRunnerJobData<'run_queues, 'j, 's>,
}

impl<'run_queues, 'j, 's> JobRunnerJobData<'run_queues, 'j, 's> {
    /// Whether more job runs need to be done for the same commit, be
    /// it for the same job, or others.
    pub fn have_more_job_runs_for_same_commit(&self) -> bool {
        // Check if this the last run for the current job. `job` still
        // contains the count from before running it this time.
        if self.job.state.remaining_count > 1 {
            return true;
        }

        // Otherwise, look for *other* jobs than the current one. Not
        // so easy since jobs still don't contain an id? Except,
        // simply check if there is more than 1 entry.
        self.run_queues_data
            .jobs_by_commit_id(&self.job.public.run_parameters.commit_id)
            .len()
            > 1
    }
}

impl<'pool, 'run_queues, 'j, 's> JobRunnerWithJob<'pool, 'run_queues, 'j, 's> {
    pub fn run_job(
        &mut self,
        working_directory_id: WorkingDirectoryId,
        reason: &Option<String>,
        schedule_condition: &ScheduleCondition,
    ) -> Result<()> {
        // XX put that here, "for backwards compat", but could now use
        // something else for logging?
        let benchmarking_job_parameters = self.job_data.job.benchmarking_job_parameters();

        let BenchmarkingJobParameters {
            run_parameters,
            command,
        } = &benchmarking_job_parameters;
        let RunParameters {
            commit_id,
            custom_parameters,
        } = run_parameters.deref();

        let bench_tmp_dir = &bench_tmp_dir()?;
        info!(
            "bench_tmp_dir path, exists?: {:?}",
            (
                (&**bench_tmp_dir).as_escaped_string(),
                bench_tmp_dir.exists()
            )
        );

        let file_cleanup_handler = self.job_runner.file_cleanup_handler;

        // File for evobench library output
        let evobench_log;
        // File for other output, for optional use by target application
        let bench_output_log;
        {
            let pid = getpid();
            evobench_log = file_cleanup_handler.register_temporary_file(Deletion::file(
                bench_tmp_dir.append(format!("evobench-{pid}.log")),
            )?)?;

            bench_output_log = file_cleanup_handler.register_temporary_file(Deletion::file(
                bench_tmp_dir.append(format!("bench-output-{pid}.log")),
            )?)?;
        }

        // Remove any stale files from previous runs (we're not
        // removing all possible files (we leave files from other
        // processes alone (in case running multiple independent
        // daemons might be useful)), just those that would get in the
        // way).
        let _ = remove_file(&evobench_log);
        let _ = remove_file(&bench_output_log);

        let conf = self.job_runner.run_config();

        let (log_extraction, cleanup) = self
            .job_runner
            .working_directory_pool
            .process_in_working_directory(
                working_directory_id,
                &self.job_runner.timestamp,
                |mut working_directory| -> Result<(&ProperDirname, PathBuf)> {
                    // Have `checkout` always run git fetch to update
                    // the remote tags, to get them even if there have
                    // been past runs where they were not present yet;
                    // this is so that when the user changes the
                    // dataset directory and then makes a matching
                    // tag, we must have that matching tag. Also, for
                    // the release hack feature, we need to learn when
                    // the tag was added later. Thus always try to
                    // update from the git repository (failures
                    // leading to the working directory ending in
                    // error state and on repetition the job being
                    // aborted).
                    let fetched_tags = working_directory
                        .get()
                        .expect("not removed")
                        .checkout(commit_id.clone(), FetchTags::Always)?;

                    // Drop the lock on the pool
                    let working_directory = working_directory.into_inner().expect("not removed");

                    let dataset_dir = dataset_dir_for(
                        conf.versioned_datasets_base_dir.as_deref(),
                        &custom_parameters,
                        self.job_runner.versioned_dataset_dir,
                        &working_directory.git_working_dir,
                        commit_id,
                        fetched_tags.clone(),
                    )?;

                    let commit_tags = get_commit_tags(
                        &working_directory,
                        &commit_id,
                        &conf.commit_tags_regex,
                        fetched_tags,
                    )?;

                    let BenchmarkingCommand {
                        target_name,
                        subdir,
                        command,
                        arguments,
                        pre_exec_bash_code,
                    } = command.deref();

                    let mut command = pre_exec_bash_code
                        .to_run_with_pre_exec(conf)
                        .command(command, arguments);

                    let dir = working_directory
                        .git_working_dir
                        .working_dir_path_ref()
                        .append(subdir);
                    let dir_str = dir.as_escaped_string();

                    // for debugging info only:
                    let cmd_in_dir = format!("command {command:?} in directory {dir_str}");

                    info!(
                        "running {cmd_in_dir}, EVOBENCH_LOG={}...",
                        evobench_log.as_escaped_string()
                    );

                    let check = assert_evobench_env_var;

                    command
                        .envs(custom_parameters.btree_map())
                        .env(check("EVOBENCH_LOG"), &**evobench_log)
                        .env(check("BENCH_OUTPUT_LOG"), &**bench_output_log)
                        .env(check("COMMIT_ID"), commit_id.to_string())
                        .env(check("COMMIT_TAGS"), commit_tags)
                        .current_dir(&dir);
                    if let Some(dataset_dir) = &dataset_dir {
                        command.env(check("DATASET_DIR"), dataset_dir);
                    }

                    let command_output_file = OutputCaptureLog::create(
                        &working_directory
                            .working_directory_path()
                            .standard_log_path(&self.job_runner.timestamp)?,
                    )?;

                    // Add info header in YAML -- XX abstraction, and
                    // move to / merge with `command_log_file.rs`?
                    command_output_file
                        .write_str(&serde_yml::to_string(&benchmarking_job_parameters)?)?;
                    command_output_file.write_str("\n")?;

                    info!(
                        "bench_tmp_dir path, exists?: {:?}",
                        (
                            (&**bench_tmp_dir).as_escaped_string(),
                            bench_tmp_dir.exists()
                        )
                    );

                    let status = {
                        let mut other_files: Vec<Box<dyn Write + Send + 'static>> = vec![];
                        // Is it evil to use log_level() for this and not a
                        // function argument?
                        if log_level() >= LogLevel::Info {
                            other_files.push(Box::new(stderr()));
                        }
                        let other_files = Arc::new(Mutex::new(other_files));

                        command_output_file.run_with_capture(
                            command,
                            other_files,
                            CaptureOptions {
                                add_source_indicator: true,
                                add_timestamp: true,
                            },
                        )?
                    };

                    info!(
                        "bench_tmp_dir path, exists?: {:?}",
                        (
                            (&**bench_tmp_dir).as_escaped_string(),
                            bench_tmp_dir.exists()
                        )
                    );

                    if status.success() {
                        info!("running {cmd_in_dir} succeeded");

                        Ok((target_name, command_output_file.into_path()))
                    } else {
                        info!("running {cmd_in_dir} failed.");

                        info!(
                            "bench_tmp_dir path, exists?: {:?}",
                            (
                                (&**bench_tmp_dir).as_escaped_string(),
                                bench_tmp_dir.exists()
                            )
                        );

                        let last_part = command_output_file.last_part(3000)?;
                        if log_level() < LogLevel::Info {
                            let mut err = stderr().lock();
                            writeln!(err, "---- run_job: error in dir {dir_str}: -------")?;
                            err.write_all(last_part.as_bytes())?;
                            writeln!(err, "---- /run_job: error in dir {dir_str} -------")?;
                        }

                        bail!(
                            "benchmarking command {cmd_in_dir} gave \
                             error status {status}, last_part {}",
                            last_part.as_escaped_string()
                        )
                    }
                },
                Some(&benchmarking_job_parameters),
                "run_job",
                Some(&|| self.job_data.have_more_job_runs_for_same_commit()),
            )?;

        // Can clean up right away, we're not currently accessing the
        // working directory any longer, OK?
        self.job_runner
            .working_directory_pool
            .working_directory_cleanup(cleanup)?;

        // Move the results to the right location below the
        // `output_base_dir`, and create/update the extracts
        {
            // The directory holding the results of all of the runs for
            // the same "key" information
            let key_dir = KeyDir::from_base_target_params(
                self.job_runner.output_base_dir.clone_arc(),
                command.target_name.clone(),
                run_parameters,
            );

            // Maintain a "latest" subdirectory at the top of the output
            // dir tree: symlink the new entry into it, with timestamp as
            // name. Create the symlink first so that it is visible even
            // in case part of the actions creating the run output
            // directory fails. -- Also see "latest-redir" below.
            {
                let latest_dir = self.job_runner.output_base_dir.join("latest");
                create_dir_all(&latest_dir).map_err(ctx!("create_dir_all {latest_dir:?}"))?;

                // To get a relative path, simply use ".." as the base
                // dir.
                let key_dir_relative_from_latest_dir =
                    key_dir.replace_base_path("..".into_arc_path());

                let symlink_location = latest_dir.join(self.job_runner.timestamp.as_str());
                symlink(
                    key_dir_relative_from_latest_dir.to_path(),
                    &symlink_location,
                )
                .map_err(ctx!("creating symlink at {symlink_location:?}"))?;
            }

            // Inside the key_dir, make a dir for this particular run
            let run_dir: RunDir = key_dir
                .clone_arc()
                .append_subdir(self.job_runner.timestamp.clone());
            {
                let path = run_dir.to_path();
                create_dir_all(path).map_err(ctx!("create_dir_all {path:?}"))?;
            }

            // Maintain a "latest-redir" subdirectory at the top of
            // the output dir tree: make a new subdirectory with an
            // index.html containing a redirect to the actual
            // directory. This keeps the URL canonical, letting one
            // know what the context is, and allows to go to the
            // parent directory--hence redirect into the run dir, not
            // key dir.  -- Also see "latest" above.
            if let Some(output_dir_url) = &self.job_runner.run_config().output_dir.url {
                let latest_dir = self.job_runner.output_base_dir.join("latest-redir");
                let subdir: PathBuf = latest_dir.join(self.job_runner.timestamp.as_str());
                {
                    let path = &subdir;
                    create_dir_all(path).map_err(ctx!("create_dir_all {path:?}"))?;
                }

                // To get a relative path, simply use `output_dir_url` as the base
                // dir.
                let redirect_target = run_dir.replace_base_path(output_dir_url.into_arc_path());
                let url = redirect_target.to_path().to_string_lossy();
                write_redirect_html_file(&subdir.append("index.html"), &url)?;
            }

            // Move the result files to the run_dir and generate the
            // files with the extracts.
            {
                info!("moving files to {}", run_dir.to_path().as_escaped_string());

                // First try to compress the log file, here we check whether
                // it exists; before we expect to compress evobench.log
                // without checking its existence.
                if bench_output_log.exists() {
                    compress_file_as(&bench_output_log, run_dir.bench_output_log_path(), false)?;
                    drop(bench_output_log);
                }

                let evobench_log_tmp =
                    compress_file_as(&evobench_log, run_dir.evobench_log_path(), true)?;

                let (target_name, standard_log_tempfile) = log_extraction;
                compress_file_as(&standard_log_tempfile, run_dir.standard_log_path(), false)?;
                // It's OK to delete the original now, but we'll make use of
                // it for reading back.
                let standard_log_tempfile = file_cleanup_handler
                    .register_temporary_file(Deletion::file(standard_log_tempfile)?)?;

                {
                    let target = run_dir.append_str("schedule_condition.ron")?;
                    let target_str = target.as_escaped_string();
                    info!("saving context to {target_str}");
                    let schedule_condition_str = ron_to_string_pretty(&schedule_condition)?;
                    std::fs::write(&target, &schedule_condition_str)
                        .map_err(ctx!("saving to {target_str}"))?
                }

                {
                    let target = run_dir.append_str("reason.ron")?;
                    let target_str = target.as_escaped_string();
                    info!("saving context to {target_str}");
                    let s = ron_to_string_pretty(&reason)?;
                    std::fs::write(&target, &s).map_err(ctx!("saving to {target_str}"))?
                }

                let evobench_log_path = evobench_log.to_owned();
                run_dir.post_process_single(
                    Some(&evobench_log_path),
                    move || {
                        info!("evaluating the benchmark file succeeded");

                        drop(evobench_log);

                        rename_tmp_path(evobench_log_tmp)?;

                        info!("compressed benchmark file renamed");
                        Ok(())
                    },
                    // log extraction:
                    target_name,
                    &standard_log_tempfile,
                    &self.job_runner.run_config(),
                    // Do not omit generation of evobench.log stats
                    false,
                )?;
            }

            // Now that the results for the run are in the right place, we
            // can update the summaries for the key
            key_dir.generate_summaries_for_key_dir(
                // Do not omit generation of evobench.log stats
                false,
            )?;
        }

        Ok(())
    }
}