The runner runs up to max_jobs jobs at once within a memory and vCPU budget, asking the hub only for jobs that fit (a lone job may use the whole budget); preparing workspaces stays one at a time bb643c252515

max_jobs = 3                 # jobs at once (default 1)

memory_mib = 24576           # what they may use together; a lone job may use it all

cpus = 8

    /// Memory the runner has free for another job, in MiB: the hub hands

    /// out only a job whose sandbox fits. Absent: no limit.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub free_memory_mib: Option<u32>,

    /// vCPUs the runner has free for another job. Absent: no limit.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub free_cpus: Option<u32>,

    /// Jobs run at once (default 1).

    #[serde(default)]

    pub max_jobs: Option<usize>,

    /// Memory (MiB) and vCPUs the jobs running at once may use together.

    #[serde(default)]

    pub memory_mib: Option<u32>,

    #[serde(default)]

    pub cpus: Option<u32>,

/// What the runner may run at once (flow.toml `[runner]`): at most

/// `max_jobs` jobs, whose sandboxes together fit `memory_mib` and `cpus`.

#[derive(Debug, Default, PartialEq, Eq)]

pub struct Budget {

    pub max_jobs: usize,

    pub memory_mib: Option<u32>,

    pub cpus: Option<u32>,

}

/// What is in use now.

#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]

pub struct Used {

    pub jobs: usize,

    pub memory_mib: u32,

    pub cpus: u32,

}

impl Budget {

    pub fn of(cfg: &RunnerConfig) -> Budget {

        Budget { max_jobs: cfg.max_jobs.unwrap_or(1).max(1), memory_mib: cfg.memory_mib, cpus: cfg.cpus }

    }

    /// The room to ask the hub for, or `None` when no job more can start.

    /// A lone job may use the whole budget even if it asks for more, so a

    /// large sandbox is never starved.

    pub fn room(&self, used: Used) -> Option<(Option<u32>, Option<u32>)> {

        if used.jobs >= self.max_jobs {

            return None;

        }

        if used.jobs == 0 {

            return Some((None, None));

        }

        let mem = self.memory_mib.map(|m| m.saturating_sub(used.memory_mib));

        let cpus = self.cpus.map(|c| c.saturating_sub(used.cpus));

        if mem == Some(0) || cpus == Some(0) {

            return None;

        }

        Some((mem, cpus))

    }

}

    let budget = Budget::of(&cfg);

    eprintln!("flow runner {name}: polling {} (up to {} jobs at once)", cfg.hub, budget.max_jobs);

    let used = std::sync::Mutex::new(Used::default());

    std::thread::scope(|scope| loop {

        let room = budget.room(*used.lock().unwrap());

        let Some((free_memory_mib, free_cpus)) = room else {

            std::thread::sleep(Duration::from_millis(500));

            continue;

        };

        let req = ClaimRequest { runner: name.clone(), runtimes: vec!["microvm".into()], free_memory_mib, free_cpus };

                let (mem, cpus) = (claim.sandbox.memory_mib, claim.sandbox.cpus);

                {

                    let mut u = used.lock().unwrap();

                    u.jobs += 1;

                    u.memory_mib += mem;

                    u.cpus += cpus;

                let (home, hub, cfg, used) = (home, &hub, &cfg, &used);

                scope.spawn(move || {

                    let id = claim.job_id;

                    eprintln!("job {id}: {} {} on {}", claim.repo, claim.job.name, claim.heads.first().map(|h| &h[..12]).unwrap_or("-"));

                    let finish = run_claim(home, hub, cfg, &claim).unwrap_or_else(|e| Finish {

                        exit_code: None,

                        duration_ms: 0,

                        receipt: None,

                        error: Some(format!("{e:#}")),

                    });

                    eprintln!(

                        "job {id}: exit {:?}{}",

                        finish.exit_code,

                        finish.error.as_deref().map(|e| format!(" ({e})")).unwrap_or_default()

                    );

                    if let Err(e) = hub.finish(id, &finish) {

                        eprintln!("job {id}: reporting failed: {e:#}");

                    }

                    let mut u = used.lock().unwrap();

                    u.jobs -= 1;

                    u.memory_mib -= mem;

                    u.cpus -= cpus;

                });

    })

/// Held while a job's workspace is prepared (see `run_claim`).

static PREPARE: std::sync::Mutex<()> = std::sync::Mutex::new(());

    // Preparing a workspace (the cached repositories, the export, the

    // image) is one job at a time: fetches into a cached repository must not

    // overlap, and the first clone passes its token through the process

    // environment. Only the microVMs run side by side.

    let mut preparing = Some(PREPARE.lock().unwrap_or_else(|p| p.into_inner()));

        preparing.take();

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn the_budget_admits_jobs_that_fit() {

        let b = Budget { max_jobs: 3, memory_mib: Some(24_576), cpus: Some(8) };

        let used = |jobs, memory_mib, cpus| Used { jobs, memory_mib, cpus };

        assert_eq!(b.room(used(0, 0, 0)), Some((None, None)), "a lone job may take everything");

        assert_eq!(b.room(used(1, 8192, 4)), Some((Some(16_384), Some(4))));

        assert_eq!(b.room(used(2, 16_384, 8)), None, "no vCPU left");

        assert_eq!(b.room(used(3, 1024, 1)), None, "at max_jobs");

        let one = Budget { max_jobs: 1, memory_mib: None, cpus: None };

        assert_eq!(one.room(used(0, 0, 0)), Some((None, None)));

        assert_eq!(one.room(used(1, 0, 0)), None, "the default runs one job at a time");

        let open = Budget { max_jobs: 2, memory_mib: None, cpus: None };

        assert_eq!(open.room(used(1, 8192, 4)), Some((None, None)), "no limits configured");

    }