evobench_tools/run/

run_queues.rs

use std::{
    collections::{BTreeMap, BTreeSet, btree_map::Entry},
    ops::Neg,
    path::PathBuf,
    sync::Arc,
    time::SystemTime,
};

use anyhow::{Result, bail};
use chj_unix_util::polling_signals::PollingSignalsSender;
use chrono::{DateTime, Local};
use cj_path_util::path_util::AppendToPath;
use genawaiter::rc::Gen;
use itertools::{EitherOrBoth, Itertools};

use crate::{
    date_and_time::time_ranges::DateTimeRange,
    git::GitHash,
    info,
    key_val_fs::{
        key_val::{KeyValConfig, KeyValSync},
        queue::{Queue, QueueGetItemOptions, QueueItem, TimeKey},
    },
    run::{
        run_job::{JobRunnerJobData, JobRunnerWithJob},
        run_queue::JobStatus,
    },
    serde_types::{priority::Priority, proper_filename::ProperFilename},
    utillib::logging::{LogLevel, log_level},
};

use super::{
    benchmarking_job::BenchmarkingJob,
    config::{QueuesConfig, ScheduleCondition},
    global_app_state_dir::GlobalAppStateDir,
    run_context::RunContext,
    run_job::JobRunner,
    run_queue::{RunQueue, RunQueueData, RunQueueDataWithNext, RunQueueWithNext},
};

// Move, where?
pub fn get_now_chrono() -> DateTime<Local> {
    SystemTime::now().into()
}

#[ouroboros::self_referencing]
pub struct RunQueues {
    pub config: Arc<QueuesConfig>,
    // Checked to be at least 1, at most one is `Immediately`,
    // etc. (private field to prevent by-passing the constructor)
    #[borrows(config)]
    #[covariant]
    pipeline: Vec<RunQueue<'this>>,

    #[borrows(config)]
    #[covariant]
    erroneous_jobs_queue: Option<RunQueue<'this>>,

    #[borrows(config)]
    #[covariant]
    done_jobs_queue: Option<RunQueue<'this>>,
}

/// A loaded copy of the on-disk data, for on-the-fly
/// indexing/multiple traversal
pub struct RunQueuesData<'run_queues> {
    run_queues: &'run_queues RunQueues,
    pipeline_data: Vec<RunQueueData<'run_queues, 'run_queues>>,
    /// Value is (index in pipeline_data, index within its queue_data)
    jobs_by_commit_id: BTreeMap<GitHash, Vec<(usize, usize)>>,
}

impl RunQueues {
    /// Retuns all queues of all kinds; e.g. for migrating items.
    pub fn all_queues<'s>(&'s self) -> impl Iterator<Item = &'s RunQueue<'s>> {
        Gen::new(|co| async move {
            for queue in self.pipeline() {
                co.yield_(queue).await;
            }
            if let Some(queue) = self.borrow_erroneous_jobs_queue().as_ref() {
                co.yield_(queue).await;
            }
            if let Some(queue) = self.borrow_done_jobs_queue().as_ref() {
                co.yield_(queue).await;
            }
        })
        .into_iter()
    }

    pub fn pipeline(&self) -> &[RunQueue<'_>] {
        self.borrow_pipeline()
    }

    pub fn erroneous_jobs_queue(&self) -> Option<&RunQueue<'_>> {
        self.borrow_erroneous_jobs_queue().as_ref()
    }

    pub fn done_jobs_queue(&self) -> Option<&RunQueue<'_>> {
        self.borrow_done_jobs_queue().as_ref()
    }

    pub fn first(&self) -> &RunQueue<'_> {
        &self.pipeline()[0]
    }

    pub fn data<'run_queues>(&'run_queues self) -> Result<RunQueuesData<'run_queues>> {
        let pipeline_data: Vec<RunQueueData> = self
            .pipeline()
            .iter()
            .map(|rq| rq.data())
            .collect::<Result<_>>()?;
        let mut jobs_by_commit_id = BTreeMap::default();
        for (i, rqd) in pipeline_data.iter().enumerate() {
            for (j, job) in rqd.jobs().enumerate() {
                let commit_id = &job.public.run_parameters.commit_id;
                match jobs_by_commit_id.entry(commit_id.clone()) {
                    Entry::Vacant(vacant_entry) => {
                        vacant_entry.insert(vec![(i, j)]);
                    }
                    Entry::Occupied(mut occupied_entry) => {
                        occupied_entry.get_mut().push((i, j));
                    }
                }
            }
        }
        Ok(RunQueuesData {
            run_queues: self,
            pipeline_data,
            jobs_by_commit_id,
        })
    }

    /// Also returns the queue following the requested one, if any
    pub fn get_run_queue_with_next_by_name(
        &self,
        file_name: &ProperFilename,
    ) -> Option<RunQueueWithNext<'_, '_>> {
        let mut queues = self.pipeline().iter();
        while let Some(current) = queues.next() {
            if current.file_name == *file_name {
                let next = queues.next();
                return Some(RunQueueWithNext { current, next });
            }
        }
        None
    }

    /// Verify that the queue configuration is valid
    fn check_run_queues(&self) -> Result<()> {
        let pipeline = self.pipeline();
        let erroneous_jobs_queue = self.erroneous_jobs_queue();
        let done_jobs_queue = self.done_jobs_queue();
        if pipeline.is_empty() {
            bail!(
                "no queues defined -- need at least one, also \
                 suggested is to add a `Inactive` as the last"
            )
        }

        let mut check_seen = {
            let mut seen = BTreeSet::new();
            move |file_name: &ProperFilename| -> Result<()> {
                if seen.contains(file_name) {
                    bail!("duplicate queue name {file_name:?}")
                }
                seen.insert(file_name.clone());
                Ok(())
            }
        };

        let mut inactive_count = 0;
        for run_queue in pipeline {
            check_seen(&run_queue.file_name)?;
            match run_queue.schedule_condition {
                ScheduleCondition::Immediately { situation: _ } => (),
                ScheduleCondition::LocalNaiveTimeWindow {
                    priority: _,
                    situation: _,
                    stop_start,
                    repeatedly: _,
                    move_when_time_window_ends: _,
                    from: _,
                    to: _,
                } => {
                    if let Some(stop_start) = &stop_start {
                        if stop_start.is_empty() {
                            bail!(
                                "`LocalNaiveTimeWindow.stop_start` was given \
                                 but is the empty list, require at least a program name/path"
                            )
                        }
                    }
                }
                ScheduleCondition::Inactive => inactive_count += 1,
            }
        }
        if inactive_count > 1 {
            bail!("can have at most one `Inactive` queue");
        }
        if inactive_count > 0 {
            if *pipeline
                .last()
                .expect("checked in the if condition")
                .schedule_condition
                != ScheduleCondition::Inactive
            {
                bail!("`Inactive` queue must be the last in the pipeline")
            }
        }

        let mut check_extra_queue = |name: &str, run_queue: Option<&RunQueue>| -> Result<()> {
            if let Some(run_queue) = run_queue {
                check_seen(&run_queue.file_name)?;
                if !run_queue.schedule_condition.is_inactive() {
                    bail!("the `{name}` must be of kind `Inactive`")
                }
            }
            Ok(())
        };
        check_extra_queue("erroneous_jobs_queue", erroneous_jobs_queue)?;
        check_extra_queue("done_jobs_queue", done_jobs_queue)?;

        Ok(())
    }

    pub fn open(
        config: Arc<QueuesConfig>,
        create_dirs_if_not_exist: bool,
        global_app_state_dir: &GlobalAppStateDir,
        signal_change: Option<PollingSignalsSender>,
    ) -> Result<Self> {
        let run_queues_basedir =
            config.run_queues_basedir(create_dirs_if_not_exist, global_app_state_dir)?;

        fn make_run_queue<'this>(
            (filename, schedule_condition): &'this (ProperFilename, ScheduleCondition),
            run_queues_basedir: &PathBuf,
            create_dirs_if_not_exist: bool,
            signal_change: Option<PollingSignalsSender>,
        ) -> Result<RunQueue<'this>> {
            let run_queue_path = (&run_queues_basedir).append(filename.as_str());
            Ok(RunQueue {
                file_name: filename.clone(),
                schedule_condition,
                queue: Queue::<BenchmarkingJob>::open(
                    &run_queue_path,
                    KeyValConfig {
                        sync: KeyValSync::All,
                        create_dir_if_not_exists: create_dirs_if_not_exist,
                    },
                    signal_change,
                )?,
            })
        }

        let slf = Self::try_new(
            config,
            // pipeline:
            |config| -> Result<_> {
                let queues = config
                    .pipeline
                    .iter()
                    .map(|cfg| {
                        make_run_queue(
                            cfg,
                            &run_queues_basedir,
                            create_dirs_if_not_exist,
                            signal_change.clone(),
                        )
                    })
                    .collect::<Result<_>>()?;
                Ok(queues)
            },
            // erroneous_jobs_queue:
            |config| {
                if let Some(cfg) = config.erroneous_jobs_queue.as_ref() {
                    Ok(Some(make_run_queue(
                        cfg,
                        &run_queues_basedir,
                        create_dirs_if_not_exist,
                        signal_change.clone(),
                    )?))
                } else {
                    Ok(None)
                }
            },
            // done_jobs_queue:
            |config| {
                if let Some(cfg) = config.done_jobs_queue.as_ref() {
                    Ok(Some(make_run_queue(
                        cfg,
                        &run_queues_basedir,
                        create_dirs_if_not_exist,
                        signal_change.clone(),
                    )?))
                } else {
                    Ok(None)
                }
            },
        )?;

        slf.check_run_queues()?;

        Ok(slf)
    }
}

impl<'run_queues> RunQueuesData<'run_queues> {
    /// For things to be run from e.g. run_job
    /// (e.g. `regenerate_index_files` function)
    pub fn run_queues(&self) -> &'run_queues RunQueues {
        self.run_queues
    }

    /// Values are (index in pipeline_data, index within its queue_data)
    pub fn jobs_by_commit_id(&self, commit_id: &GitHash) -> &[(usize, usize)] {
        self.jobs_by_commit_id
            .get(commit_id)
            .map(|v| v.as_slice())
            .unwrap_or([].as_slice())
    }

    pub fn have_job_with_commit_id(&self, commit_id: &GitHash) -> bool {
        !self.jobs_by_commit_id(commit_id).is_empty()
    }

    /// Iterator over all entries for that commit id. Still efficient,
    /// since it just returns references to existing tuples.--Not
    /// actually used, might it be useful in the future?
    #[allow(unused)]
    fn entries_by_commit_id(
        &self,
        commit_id: &GitHash,
    ) -> impl Iterator<Item = &(TimeKey, BenchmarkingJob, Priority)> {
        self.jobs_by_commit_id(commit_id)
            .iter()
            .map(|(i, j)| self.pipeline_data[*i].entry(*j))
    }

    /// The `RunQueueData`s paired with their successor (still in the
    /// original, configured, order)
    pub fn run_queue_with_nexts<'s>(
        &'s self,
    ) -> impl Iterator<Item = RunQueueDataWithNext<'run_queues, 'run_queues, 's>> {
        self.pipeline_data
            .iter()
            .zip_longest(self.pipeline_data.iter().skip(1))
            .map(|either_or_both| match either_or_both {
                EitherOrBoth::Both(current, next) => RunQueueDataWithNext {
                    current,
                    next: Some(next),
                },
                EitherOrBoth::Left(current) => RunQueueDataWithNext {
                    current,
                    next: None,
                },
                EitherOrBoth::Right(_) => unreachable!("because the left sequence is longer"),
            })
    }

    /// All queues which are runnable at the given time, with their
    /// successor queue, and calculated time window if any
    fn active_queues<'s>(
        &'s self,
        reference_time: DateTime<Local>,
    ) -> impl Iterator<
        Item = (
            RunQueueDataWithNext<'run_queues, 'run_queues, 's>,
            Option<DateTimeRange<Local>>,
        ),
    > {
        self.run_queue_with_nexts().filter_map(move |rq| {
            if let Some(range) = rq
                .current
                .run_queue()
                .schedule_condition
                .is_runnable_at(reference_time)
            {
                Some((rq, range))
            } else {
                None
            }
        })
    }

    /// The most prioritized job across all runnable queues
    fn most_prioritized_job<'s, 'conf, 'r, 'rc>(
        &'s self,
        now: DateTime<Local>,
    ) -> Result<
        Option<(
            RunQueueDataWithNext<'run_queues, 'run_queues, 's>,
            Option<DateTimeRange<Local>>,
            QueueItem<'run_queues, BenchmarkingJob>,
            &'s BenchmarkingJob,
            Priority,
        )>,
    > {
        let verbose = log_level() >= LogLevel::Info;

        // Get the single most prioritized job from each queue (if
        // any), then of those the most prioritized one. Using
        // min_by_key since this takes the first of the equal jobs,
        // unlike max_by_key.
        if let Some(((key, job, prio), rq, dtr)) = self
            .active_queues(now)
            .filter_map(|(rq, dtr)| -> Option<_> {
                let entry = rq
                    .current
                    .entries()
                    .min_by_key(|(_, _, job_priority)| job_priority.neg())?;

                Some((entry, rq, dtr))
            })
            .min_by_key(|((_, _, job_priority), _, _)| job_priority.neg())
        {
            if let Some(item) = rq.current.run_queue().queue.get_item(
                key,
                QueueGetItemOptions {
                    verbose,
                    no_lock: true,
                    error_when_locked: false,
                    delete_first: false,
                },
            )? {
                Ok(Some((rq, dtr, item, job, *prio)))
            } else {
                info!("entry {key} has disappeared in the mean time, skipping it");
                Ok(None)
            }
        } else {
            Ok(None)
        }
    }

    /// Run the first or most prioritized job in the queues. Returns
    /// the job and its status after running it if one was found,
    /// None if all runnable queues are empty.
    ///
    /// This method needs to be run in a loop forever for daemon style
    /// processing. The reason this doesn't do the looping inside is
    /// to allow for a reload of the queue config and then
    /// queues. `current_stop_start`, if given, represents an active
    /// `stop_start` command that was run with `stop` and now needs a
    /// `start` when the next running action does not require the same
    /// command to be `stop`ed. Likewise, this method returns the
    /// active `stop_start` command, if any, by the time it
    /// returns. The caller should pass that back into this method on
    /// the next iteration. Using SliceOrBox to allow carrying it over
    /// a config reload. `now` should be the current time (at least is
    /// understood as such), get it via `get_now_chrono()` right
    /// before calling this method.
    pub fn run_next_job<'s, 'conf, 'r, 'rc>(
        &'s self,
        job_runner: JobRunner,
        run_context: &mut RunContext,
    ) -> Result<Option<(&'s BenchmarkingJob, JobStatus)>> {
        let timestamp_local = job_runner.timestamp_local();

        let job = self.most_prioritized_job(timestamp_local)?;

        if let Some((rqdwn, dtr, item, job, _)) = job {
            let rq = rqdwn.current.run_queue();
            run_context.stop_start_be(rq.schedule_condition.stop_start())?;
            if let Some(dtr) = dtr {
                run_context.running_job_in_windowed_queue(rq, dtr);
            }

            let working_directory_id;
            {
                let mut lock = job_runner
                    .working_directory_pool
                    .lock_mut("RunQueuesData.run_next_job")?;
                working_directory_id =
                    lock.get_a_working_directory_for(&job.public.run_parameters, self)?;

                lock.clear_current_working_directory()?;
            }

            let job_status = rqdwn.run_queue_with_next().run_job(
                &item,
                &mut JobRunnerWithJob {
                    job_runner,
                    job_data: JobRunnerJobData {
                        job,
                        run_queues_data: self,
                    },
                },
                self.run_queues.erroneous_jobs_queue(),
                self.run_queues.done_jobs_queue(),
                working_directory_id,
            )?;

            Ok(Some((job, job_status)))
        } else {
            run_context.stop_start_be(None)?;

            Ok(None)
        }
    }
}