runnerd_core/admission/
recent.rs

1//! A job's recent runs, as admission and sizing read them (ADR 0070): each row of the history read once, in the order
2//! the runs finished, into a window of the job's last successful runs, the hold a run at its cap puts on it, and the
3//! runs past what they reserved. One rule for what each row says, whoever feeds it: the engine's predictor,
4//! `simulate` and sizing. Pure: no I/O, no clock; it must not keep anything the rows do not say, so a restart that
5//! reads the same rows decides the same.
6
7use std::collections::{BTreeMap, VecDeque};
8
9use super::Key;
10use crate::sizing::{Capped, History, Measure, Ran};
11
12/// Admission's window: a job's last this many successful runs, or `[sizing] min_jobs` when more. Under nearest rank
13/// a p95 of fewer than 20 values is their largest; 20 is the fewest at which it can leave one outlier out.
14pub const RECENT_RUNS: u64 = 20;
15
16/// Sizing's window is this many times admission's: a cap too small kills a job, a reservation too small only lets
17/// runs overlap, and a cold-cache run comes once in 20 to 50.
18pub const SIZING_TIMES: u64 = 3;
19
20/// A hold at the cap ends after this many successful runs in a row clear of their cap.
21pub const CLEAN_RUNS: u32 = 3;
22
23/// How a row ended, as the history keeps it apart.
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum Ending {
26    /// Completed, failed or lost: its peak counts.
27    Finished,
28    /// Cancelled: its peak may be cut short, so only a run that met its cap counts its peak.
29    Cancelled,
30}
31
32/// A run in a job's window.
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
34pub struct Counted {
35    /// When it finished, in ms.
36    pub at: u64,
37    /// Its peak, in bytes; at its cap or more when it met it, since the cap stopped it.
38    pub peak: u64,
39    /// Its CPU peak, millicores.
40    pub cpu_peak: u64,
41    /// The cap its container had.
42    pub cap: u64,
43    /// What it says about that cap.
44    pub capped: Capped,
45    /// Whether it completed with success.
46    pub succeeded: bool,
47}
48
49/// The latest run of a job that met its cap or was killed for memory, and how many successful runs clear of their
50/// cap came in a row after it.
51#[derive(Debug, Clone, Copy, PartialEq, Eq)]
52pub struct Hold {
53    /// When that run finished, in ms.
54    pub at: u64,
55    /// The clear successful runs after it, in a row.
56    pub clean: u32,
57}
58
59/// A run that peaked past what it reserved.
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61pub struct Overrun {
62    /// When it finished, in ms.
63    pub at: u64,
64    /// Its peak.
65    pub peak: u64,
66}
67
68/// One job's runs in one profile.
69#[derive(Debug, Clone)]
70pub struct JobRuns {
71    /// How many successful runs its window holds.
72    size: u64,
73    window: VecDeque<Counted>,
74    successes: u64,
75    sorted: Vec<u64>,
76    hold: Option<Hold>,
77    overruns: Vec<Overrun>,
78}
79
80impl JobRuns {
81    fn new(size: u64) -> JobRuns {
82        JobRuns {
83            size,
84            window: VecDeque::new(),
85            successes: 0,
86            sorted: Vec::new(),
87            hold: None,
88            overruns: Vec::new(),
89        }
90    }
91
92    /// The window's runs, oldest first.
93    pub fn window(&self) -> impl Iterator<Item = &Counted> {
94        self.window.iter()
95    }
96
97    /// How many runs the window has.
98    pub fn runs(&self) -> u64 {
99        self.window.len() as u64
100    }
101
102    /// How many of them succeeded.
103    pub fn succeeded(&self) -> u64 {
104        self.successes
105    }
106
107    /// Their peaks, sorted.
108    pub fn peaks(&self) -> &[u64] {
109        &self.sorted
110    }
111
112    /// The largest of them; 0 for none.
113    pub fn max(&self) -> u64 {
114        self.sorted.last().copied().unwrap_or(0)
115    }
116
117    /// The hold at the cap, if it has not ended by `CLEAN_RUNS` clear runs; `for_ms` ends it that long after its
118    /// run as well, `None` never.
119    pub fn held(&self, now: u64, for_ms: Option<u64>) -> Option<Hold> {
120        self.hold.filter(|hold| {
121            hold.clean < CLEAN_RUNS && for_ms.is_none_or(|ms| now < hold.at.saturating_add(ms))
122        })
123    }
124
125    /// The runs past what they reserved that finished at or after the window's start and less than `for_ms` before
126    /// `now`.
127    pub fn overruns(&self, now: u64, for_ms: u64) -> impl Iterator<Item = &Overrun> {
128        let start = self.start();
129        self.overruns
130            .iter()
131            .filter(move |overrun| overrun.at >= start && now < overrun.at.saturating_add(for_ms))
132    }
133
134    /// When the window starts: its oldest run's finish once it is full, 0 before.
135    fn start(&self) -> u64 {
136        if self.successes >= self.size {
137            self.window.front().map_or(0, |run| run.at)
138        } else {
139            0
140        }
141    }
142
143    fn push(&mut self, run: Counted) {
144        let size = self.size;
145        self.successes += u64::from(run.succeeded);
146        self.window.push_back(run);
147        while self.successes > size {
148            if self.window.pop_front().is_some_and(|old| old.succeeded) {
149                self.successes -= 1;
150            }
151        }
152        // A full window starts at its oldest successful run: failed runs before it never hold a place.
153        while self.successes == size && self.window.front().is_some_and(|old| !old.succeeded) {
154            self.window.pop_front();
155        }
156        self.sorted = self.window.iter().map(|run| run.peak).collect();
157        self.sorted.sort_unstable();
158        let start = self.start();
159        self.overruns.retain(|overrun| overrun.at >= start);
160    }
161}
162
163/// A job in one profile: a job of one name may run in several pools of its repository, whose runs say nothing of
164/// each other's (ADR 0058).
165#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
166pub struct InProfile {
167    /// The job.
168    pub key: Key,
169    /// The profile it ran in.
170    pub profile: String,
171}
172
173/// Every job's recent runs, with a window of `size` successful runs.
174#[derive(Debug, Clone)]
175pub struct Recent {
176    size: u64,
177    jobs: BTreeMap<InProfile, JobRuns>,
178    /// Each profile's jobs' windows together, sorted.
179    profiles: BTreeMap<String, Vec<u64>>,
180}
181
182impl Recent {
183    /// Nothing read yet, with windows of `size` successful runs, at least 1.
184    pub fn new(size: u64) -> Recent {
185        Recent {
186            size: size.max(1),
187            jobs: BTreeMap::new(),
188            profiles: BTreeMap::new(),
189        }
190    }
191
192    /// Reads a row of `profile` that ended at `at` (ms) under `cap`. Rows must come in the order they finished.
193    ///
194    /// A run sampled with page cache counted adds no peak, and only a kill for memory says anything of its cap. A
195    /// cancelled run adds its peak only when it met its cap: it got that far. A run past what it reserved is an
196    /// overrun, cancelled or not. A successful run clear of its cap and of what it reserved counts toward the end of
197    /// a hold; one that went past what it reserved starts the count again.
198    pub fn read(&mut self, run: &Ran, profile: &str, ending: Ending, cap: u64, at: u64) {
199        let size = self.size;
200        // A run sampled with page cache counted says nothing of its job but a kill for memory.
201        if run.measure == Measure::WithActiveFile && !run.oom_killed {
202            return;
203        }
204        let job = self
205            .jobs
206            .entry(InProfile {
207                key: Key::of(run),
208                profile: profile.to_owned(),
209            })
210            .or_insert_with(|| JobRuns::new(size));
211        if run.measure == Measure::WithActiveFile {
212            job.hold = Some(Hold { at, clean: 0 });
213            return;
214        }
215        let capped = run.capped(cap);
216        let overran = run
217            .reserved
218            .is_some_and(|reserved| run.memory_peak > reserved);
219        if capped.holds() {
220            job.hold = Some(Hold { at, clean: 0 });
221        } else if ending == Ending::Finished
222            && run.succeeded
223            && let Some(hold) = &mut job.hold
224        {
225            hold.clean = if overran { 0 } else { hold.clean + 1 };
226        }
227        if overran {
228            job.overruns.push(Overrun {
229                at,
230                peak: run.memory_peak,
231            });
232        }
233        if ending == Ending::Finished || capped.met() {
234            job.push(Counted {
235                at,
236                peak: if capped.met() {
237                    run.memory_peak.max(cap)
238                } else {
239                    run.memory_peak
240                },
241                cpu_peak: run.cpu_peak,
242                cap,
243                capped,
244                succeeded: ending == Ending::Finished && run.succeeded,
245            });
246            let mut peaks: Vec<u64> = self
247                .jobs
248                .iter()
249                .filter(|(job, _)| job.profile == profile)
250                .flat_map(|(_, runs)| runs.sorted.iter().copied())
251                .collect();
252            peaks.sort_unstable();
253            self.profiles.insert(profile.to_owned(), peaks);
254        }
255    }
256
257    /// Reads every row of `history` in the order they finished, each under the cap its container had, else
258    /// `cap_of(profile)`, else none.
259    pub fn read_history(&mut self, history: &History, cap_of: impl Fn(&str) -> Option<u64>) {
260        for (run, ending) in rows(history) {
261            let cap = run
262                .memory_cap
263                .or_else(|| cap_of(&run.profile))
264                .unwrap_or(u64::MAX);
265            self.read(run, &run.profile, ending, cap, finished(run));
266        }
267    }
268
269    /// The runs of `key` in `profile`, if it was seen.
270    pub fn job(&self, key: &Key, profile: &str) -> Option<&JobRuns> {
271        self.jobs.get(&InProfile {
272            key: key.clone(),
273            profile: profile.to_owned(),
274        })
275    }
276
277    /// Every job of `profile` and its runs.
278    pub fn jobs_of<'recent>(
279        &'recent self,
280        profile: &'recent str,
281    ) -> impl Iterator<Item = (&'recent Key, &'recent JobRuns)> {
282        self.jobs
283            .iter()
284            .filter(move |(job, _)| job.profile == profile)
285            .map(|(job, runs)| (&job.key, runs))
286    }
287
288    /// `profile`'s jobs' windows together, sorted.
289    pub fn profile(&self, profile: &str) -> &[u64] {
290        self.profiles.get(profile).map_or(&[], Vec::as_slice)
291    }
292}
293
294/// When a row ended, as the windows order it: its finish, or its queueing without one.
295pub fn finished(run: &Ran) -> u64 {
296    run.finished_at.unwrap_or(run.queued_at)
297}
298
299/// `history`'s rows, finished and cancelled, in the order they finished; ties in the store's order, finished ones
300/// first.
301pub fn rows(history: &History) -> Vec<(&Ran, Ending)> {
302    let mut rows: Vec<(&Ran, Ending)> = history
303        .jobs
304        .iter()
305        .map(|run| (run, Ending::Finished))
306        .chain(history.cancelled.iter().map(|run| (run, Ending::Cancelled)))
307        .collect();
308    rows.sort_by_key(|(run, _)| finished(run));
309    rows
310}
runnerd_core/admission/recent.rs

runnerd_core/admission/
recent.rs
