runnerd_core/
admission.rs

1//! Admission by measured peaks: a job's memory peak predicted from the job history (`Predictor`), and the history
2//! replayed through admission by caps and by predictions (`simulate`), to show what each would have done with the
3//! same jobs. Pure: no I/O, no clock.
4
5use std::collections::BTreeMap;
6
7use crate::api::Basis;
8use crate::api::Spread;
9use crate::config::{self, Config};
10use crate::sizing::{GRANULE, Ran, spread};
11
12/// A run that peaked within this percent of its cap may have been held back or killed there.
13const AT_CAP: u64 = 95;
14
15/// For this long after a run at its cap or killed for memory, in milliseconds, its job is admitted at its cap: 7
16/// days. After a run past what it reserved, its job reserves its largest peak for as long.
17pub const UNPREDICTABLE_FOR: u64 = 7 * 24 * 3600 * 1000;
18
19/// While the slice uses this percent of its `MemoryMax` or more, measured admission starts nothing.
20pub const HIGH_WATER: u64 = 90;
21
22/// A job as admission tells jobs apart: its repository, workflow and name.
23#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
24pub struct Key {
25    /// `owner/name`.
26    pub repository: String,
27    /// The workflow's name, when known.
28    pub workflow: Option<String>,
29    /// The job's name in the workflow.
30    pub name: String,
31}
32
33impl Key {
34    /// A finished run's key.
35    pub fn of(run: &Ran) -> Key {
36        Key {
37            repository: run.repository.clone(),
38            workflow: run.workflow.clone(),
39            name: run.name.clone(),
40        }
41    }
42}
43
44/// The memory admission reserves for a job, at most its cap, and what it goes by.
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub struct Prediction {
47    /// Bytes.
48    pub memory: u64,
49    /// Why that much.
50    pub basis: Basis,
51}
52
53/// The peaks of a set of runs, sorted, until when a run at its cap makes them unpredictable, and until when a run
54/// past what it reserved makes the largest of them, or that run's peak if more, the prediction.
55#[derive(Debug, Clone, Default)]
56struct Peaks {
57    sorted: Vec<u64>,
58    unpredictable_until: u64,
59    overran_until: u64,
60    /// The largest peak of a run past what it reserved, cancelled runs' included.
61    overran_peak: u64,
62}
63
64impl Peaks {
65    fn add(&mut self, peak: u64) {
66        let at = self.sorted.partition_point(|known| *known <= peak);
67        self.sorted.insert(at, peak);
68    }
69
70    fn runs(&self) -> u64 {
71        self.sorted.len() as u64
72    }
73
74    fn p95(&self) -> u64 {
75        spread(self.sorted.clone()).p95
76    }
77
78    fn max(&self) -> u64 {
79        self.sorted.last().copied().unwrap_or(0)
80    }
81}
82
83/// Predicts a job's peak from the runs it learned.
84#[derive(Debug, Clone)]
85pub struct Predictor {
86    margin: u32,
87    min_jobs: u32,
88    jobs: BTreeMap<Key, Peaks>,
89    profiles: BTreeMap<String, Peaks>,
90}
91
92impl Predictor {
93    /// A predictor that knows no run yet, with `[sizing]`'s margin and minimum runs.
94    pub fn new(rules: &config::Sizing) -> Predictor {
95        Predictor {
96            margin: rules.margin,
97            min_jobs: rules.min_jobs,
98            jobs: BTreeMap::new(),
99            profiles: BTreeMap::new(),
100        }
101    }
102
103    /// Learns a run of `profile` that finished at `at` (ms), having run under `cap`: its peak, and how it ended.
104    pub fn learn(&mut self, run: &Ran, profile: &str, cap: u64, at: u64) {
105        self.mark(run, cap, at);
106        let peak = run.memory_peak;
107        if let Some(job) = self.jobs.get_mut(&Key::of(run)) {
108            job.add(peak);
109        }
110        self.profiles
111            .entry(profile.to_owned())
112            .or_default()
113            .add(peak);
114    }
115
116    /// Learns only how a run ended, not its peak, as for a cancelled one: a kill for memory or a peak at its cap
117    /// makes its job unpredictable for `UNPREDICTABLE_FOR`, a peak past what it reserved makes its job reserve its
118    /// largest peak for as long.
119    pub fn mark(&mut self, run: &Ran, cap: u64, at: u64) {
120        let peak = run.memory_peak;
121        let job = self.jobs.entry(Key::of(run)).or_default();
122        if run.oom_killed || peak.saturating_mul(100) >= cap.saturating_mul(AT_CAP) {
123            job.unpredictable_until = job.unpredictable_until.max(at + UNPREDICTABLE_FOR);
124        }
125        if run.reserved.is_some_and(|reserved| peak > reserved) {
126            job.overran_until = job.overran_until.max(at + UNPREDICTABLE_FOR);
127            job.overran_peak = job.overran_peak.max(peak);
128        }
129    }
130
131    /// The memory to reserve for a run of `key` in `profile`, whose cap is `cap`, at `now` (ms).
132    pub fn predict(&self, key: &Key, profile: &str, cap: u64, now: u64) -> Prediction {
133        let job = self.jobs.get(key);
134        if let Some(until) = job
135            .map(|job| job.unpredictable_until)
136            .filter(|until| *until > now)
137        {
138            return Prediction {
139                memory: cap,
140                basis: Basis::Unpredictable {
141                    until: until / 1000,
142                },
143            };
144        }
145        let min_jobs = u64::from(self.min_jobs);
146        let (memory, basis) = match (job, self.profiles.get(profile)) {
147            // Seen only cancelled, so its peaks say nothing.
148            (Some(job), _) if job.runs() == 0 => (cap, Basis::Cap),
149            (Some(job), _) if job.overran_until > now => (
150                self.with_margin(job.max().max(job.overran_peak)),
151                Basis::Overran {
152                    until: job.overran_until / 1000,
153                },
154            ),
155            (Some(job), _) if job.runs() >= min_jobs => {
156                (self.with_margin(job.p95()), Basis::Job { runs: job.runs() })
157            }
158            (Some(job), Some(peers)) if peers.runs() >= min_jobs => (
159                self.with_margin(job.max().max(peers.p95())),
160                Basis::Profile { runs: peers.runs() },
161            ),
162            // A job never seen may be the profile's heaviest.
163            (None, Some(peers)) if peers.runs() >= min_jobs => (
164                self.with_margin(peers.max()),
165                Basis::ProfileMax { runs: peers.runs() },
166            ),
167            _ => (cap, Basis::Cap),
168        };
169        Prediction {
170            memory: memory.min(cap),
171            basis,
172        }
173    }
174
175    /// `peak` plus the margin, rounded up to 256M.
176    fn with_margin(&self, peak: u64) -> u64 {
177        let memory = peak.saturating_add(peak.saturating_mul(u64::from(self.margin)) / 100);
178        memory
179            .div_ceil(GRANULE)
180            .saturating_mul(GRANULE)
181            .max(GRANULE)
182    }
183}
184
185/// How a replay admits a queued job.
186#[derive(Debug, Clone, Copy, PartialEq, Eq)]
187pub enum Policy {
188    /// When a slot is free and the caps of the running jobs and its own fit the budget: what runnerd does now.
189    Caps,
190    /// As the daemon admits: when a slot is free, the running jobs (each at the larger of its prediction and its
191    /// peak) and the largest prediction among the repository's waiting jobs fit the budget, and the running jobs'
192    /// peaks are under `HIGH_WATER` percent of it. A started job then reserves its own prediction.
193    Measured,
194}
195
196/// What a replay did with the history's jobs.
197#[derive(Debug, Clone, PartialEq, Eq)]
198pub struct Replay {
199    /// The jobs replayed: finished ones of a configured profile.
200    pub jobs: u64,
201    /// Of those, the ones the policy never started: more than the budget holds.
202    pub never_started: u64,
203    /// Seconds from queued to started.
204    pub wait: Spread,
205    /// The mean of those, in seconds.
206    pub mean_wait: u64,
207    /// The most jobs that ran at once.
208    pub most_running: u64,
209    /// The most memory, in bytes, the running jobs' peaks and the mirror add up to at one moment: as if every
210    /// running job peaked at once, which is more than the slice ever holds. Not a bound on the daemon, which can
211    /// overlap more: it sees what jobs use, less than their peaks while they warm up, and the replay leaves out
212    /// start latency, seeds, shells and cancelled jobs.
213    pub worst_overlap: u64,
214    /// Seconds from the first job queued to the last one finished.
215    pub makespan: u64,
216    /// Jobs whose peak was above what admission reserved for them.
217    pub under_predicted: u64,
218    /// How many times the high-water mark held a job back that the predictions had room for.
219    pub held_back: u64,
220}
221
222/// A job the replay started.
223struct Running<'run> {
224    run: &'run Ran,
225    profile: &'run str,
226    ends: u64,
227    reserved: u64,
228    peak: u64,
229}
230
231/// Replays `jobs` through `policy`: each arrives when it was queued and runs as long as it ran, in a profile with
232/// `slots` (by name; the config's where absent) under the slice's `memory_max`. Jobs start as soon as the policy
233/// admits them, oldest first, a job that does not fit letting a later one past, as `scheduler::plan` does.
234/// Measured admission predicts from the runs finished before it decides, as the daemon would, and holds each
235/// running job at its peak for its whole run, which the daemon does not see.
236pub fn simulate(
237    config: &Config,
238    memory_max: Option<u64>,
239    jobs: &[Ran],
240    policy: Policy,
241    slots: &BTreeMap<String, u32>,
242) -> Replay {
243    let mirror = if config.caches.mirror.enabled {
244        config.caches.mirror.memory.0
245    } else {
246        0
247    };
248    let budget = memory_max.map(|max| max.saturating_sub(mirror));
249    let mut queue: Vec<&Ran> = jobs
250        .iter()
251        .filter(|run| config.profiles.contains_key(&run.profile))
252        .filter(|run| matches!((run.started_at, run.finished_at), (Some(start), Some(end)) if end >= start))
253        .collect();
254    queue.sort_by_key(|run| run.queued_at);
255    let mut replayer = Replayer {
256        config,
257        budget,
258        policy,
259        slots,
260        predictor: Predictor::new(&config.sizing),
261        running: Vec::new(),
262        waits: Vec::new(),
263        most_running: 0,
264        worst_overlap: 0,
265        last_end: 0,
266        under_predicted: 0,
267        held_back: 0,
268    };
269    let mut waiting: Vec<&Ran> = Vec::new();
270    let mut next = 0;
271    loop {
272        let arrival = queue.get(next).map(|run| run.queued_at);
273        let end = replayer.running.iter().map(|running| running.ends).min();
274        let Some(now) = [arrival, end].into_iter().flatten().min() else {
275            break;
276        };
277        replayer.finish(now);
278        while let Some(run) = queue.get(next).filter(|run| run.queued_at <= now) {
279            waiting.push(run);
280            next += 1;
281        }
282        let mut index = 0;
283        while let Some(run) = waiting.get(index).copied() {
284            if replayer.admit(run, &waiting, now) {
285                waiting.remove(index);
286            } else {
287                index += 1;
288            }
289        }
290        replayer.most_running = replayer.most_running.max(replayer.running.len() as u64);
291        let peaks: u64 = replayer.running.iter().map(|running| running.peak).sum();
292        replayer.worst_overlap = replayer.worst_overlap.max(peaks + mirror);
293    }
294    let waits = replayer.waits;
295    let first = queue.first().map_or(0, |run| run.queued_at);
296    Replay {
297        jobs: queue.len() as u64,
298        never_started: waiting.len() as u64,
299        mean_wait: waits.iter().sum::<u64>() / (waits.len() as u64).max(1),
300        wait: spread(waits),
301        most_running: replayer.most_running,
302        worst_overlap: replayer.worst_overlap,
303        makespan: replayer.last_end.saturating_sub(first) / 1000,
304        under_predicted: replayer.under_predicted,
305        held_back: replayer.held_back,
306    }
307}
308
309/// A replay's state between two moments.
310struct Replayer<'run> {
311    config: &'run Config,
312    budget: Option<u64>,
313    policy: Policy,
314    slots: &'run BTreeMap<String, u32>,
315    predictor: Predictor,
316    running: Vec<Running<'run>>,
317    waits: Vec<u64>,
318    most_running: u64,
319    worst_overlap: u64,
320    last_end: u64,
321    under_predicted: u64,
322    held_back: u64,
323}
324
325impl<'run> Replayer<'run> {
326    /// Ends the jobs that end by `now`; measured admission learns their peaks.
327    fn finish(&mut self, now: u64) {
328        let (ended, running): (Vec<_>, Vec<_>) = std::mem::take(&mut self.running)
329            .into_iter()
330            .partition(|running| running.ends <= now);
331        self.running = running;
332        for ended in ended {
333            let cap = self.config.profiles[ended.profile].memory.0;
334            let reserved = (self.policy == Policy::Measured).then_some(ended.reserved);
335            let run = Ran {
336                reserved,
337                ..ended.run.clone()
338            };
339            self.predictor.learn(&run, ended.profile, cap, ended.ends);
340            self.last_end = self.last_end.max(ended.ends);
341        }
342    }
343
344    /// What a running job takes of the budget: under measured admission, as the daemon counts it, the larger of
345    /// what it reserved and what it uses, here its peak for the whole run.
346    fn taken(&self) -> u64 {
347        self.running
348            .iter()
349            .map(|running| match self.policy {
350                Policy::Caps => running.reserved,
351                Policy::Measured => running.reserved.max(running.peak),
352            })
353            .sum()
354    }
355
356    /// Starts `run` at `now` if the policy admits it, `waiting` being every job not started yet; whether it did.
357    /// As in the daemon, the runner starts reserving the largest prediction among its repository's waiting jobs,
358    /// then its job's own.
359    fn admit(&mut self, run: &'run Ran, waiting: &[&Ran], now: u64) -> bool {
360        let Some((profile, limits)) = self.config.profiles.get_key_value(&run.profile) else {
361            return false;
362        };
363        let slots = self.slots.get(profile).copied().unwrap_or(limits.slots);
364        let in_profile = self
365            .running
366            .iter()
367            .filter(|running| running.profile == profile.as_str())
368            .count();
369        if in_profile >= slots as usize {
370            return false;
371        }
372        let cap = limits.memory.0;
373        let predict = |job: &Ran| {
374            self.predictor
375                .predict(&Key::of(job), profile, cap, now)
376                .memory
377        };
378        let (starting, reserved) = match self.policy {
379            Policy::Caps => (cap, cap),
380            Policy::Measured => (
381                waiting
382                    .iter()
383                    .filter(|job| job.repository == run.repository)
384                    .map(|job| predict(job))
385                    .max()
386                    .unwrap_or(0)
387                    .max(predict(run)),
388                predict(run),
389            ),
390        };
391        if let Some(budget) = self.budget {
392            if self.taken() + starting > budget {
393                return false;
394            }
395            let peaks: u64 = self.running.iter().map(|running| running.peak).sum();
396            if self.policy == Policy::Measured
397                && peaks.saturating_mul(100) >= budget.saturating_mul(HIGH_WATER)
398            {
399                self.held_back += 1;
400                return false;
401            }
402        }
403        // A job over its cap is killed at it.
404        let peak = run.memory_peak.min(cap);
405        self.under_predicted += u64::from(peak > reserved);
406        let ran_for = run
407            .finished_at
408            .unwrap_or_default()
409            .saturating_sub(run.started_at.unwrap_or_default());
410        self.waits.push(now.saturating_sub(run.queued_at) / 1000);
411        self.running.push(Running {
412            run,
413            profile,
414            ends: now + ran_for,
415            reserved,
416            peak,
417        });
418        true
419    }
420}
421
422#[cfg(test)]
423mod tests;
runnerd_core/admission.rs

runnerd_core/
admission.rs
