1use std::collections::BTreeMap;
11use std::fmt::Write as _;
12
13use crate::api::{
14 Advice, ProfileSizing, Resize, Sizing, SizingChange, SizingRefused, cores, job_title,
15};
16use crate::config::{Admission, ByteSize, Config, Profile, cpu_budget};
17
18use super::{
19 BUSY, CpusSet, GRANULE, History, WAITED, Window, advise, busy, first_runner, spread, waits,
20};
21
22const RAISE_RUNS: u64 = 3;
25
26const BOUND: u64 = 70;
30
31const HEADROOM: u64 = 2;
35
36#[derive(Debug, Clone, Copy)]
38pub struct Context<'run> {
39 pub memory_max: Option<u64>,
41 pub cpus: Option<u64>,
43 pub window: Window,
45 pub last: &'run BTreeMap<String, u64>,
47 pub cpus_set: &'run [CpusSet],
50 pub running: &'run BTreeMap<String, u32>,
52}
53
54pub fn changes(config: &Config, history: &History, context: Context<'_>) -> Vec<SizingChange> {
72 next_look(config, history, context).changes
73}
74
75#[derive(Debug, Clone, Default, PartialEq, Eq)]
77pub struct Look {
78 pub changes: Vec<SizingChange>,
80 pub refused: Vec<SizingRefused>,
83}
84
85pub fn next_look(config: &Config, history: &History, context: Context<'_>) -> Look {
87 let rules = &config.sizing;
88 if !rules.auto {
89 return Look::default();
90 }
91 let window = context.window;
92 let sizing = advise(config, context.memory_max, history, window);
93 let now = window.now / 1000;
94 let mut budget = Budget::new(config, &sizing.profiles);
95 let mut changes = Vec::new();
96 let mut refused = Vec::new();
97 for advised in &sizing.profiles {
98 let Some(profile) = config.profiles.get(&advised.name) else {
99 continue;
100 };
101 let last = context.last.get(&advised.name).copied();
102 let since = last.map_or(u64::MAX, |at| now.saturating_sub(at));
103 if since < rules.raise_after().as_secs() {
104 continue;
105 }
106 let lowering = if since >= rules.interval.as_secs() {
107 Lowering::Allowed
108 } else {
109 Lowering::Held
110 };
111 let fresh = Fresh::read(history, advised, last, window);
112 let mut reasons = Vec::new();
113 let memory = memory(
114 config,
115 profile,
116 advised.advice,
117 met_cap(history, &advised.name),
118 lowering,
119 );
120 let running = context.running.get(&advised.name).copied().unwrap_or(0);
121 let slots = match config.host.admission {
122 Admission::Caps => {
123 caps_slots(profile, advised, &fresh, memory, &context, &budget, running)
124 }
125 Admission::Measured => {
126 measured_slots(config, &sizing.profiles, advised, &fresh, &context, &budget)
127 }
128 };
129 let (slots, cpus) = cpus_and_slots(
130 config,
131 profile,
132 advised,
133 history,
134 fresh.start,
135 &context,
136 &budget,
137 slots,
138 lowering,
139 );
140 if memory != profile.memory.0 {
141 reasons.push(memory_reason(
142 &sizing,
143 advised,
144 memory,
145 slots,
146 config.sizing.max_step,
147 ));
148 }
149 let advised_slots = config.host.admission == Admission::Caps && memory < profile.memory.0;
151 if slots > profile.slots && !advised_slots {
152 reasons.push(fresh.waited(last.is_some()));
153 }
154 let cpus = cpus.taken(&advised.name, &mut reasons, &mut refused);
155 let from = Resize {
156 memory: profile.memory.0,
157 slots: profile.slots,
158 cpus: cpus.map(|_| profile.cpus),
159 };
160 let to = Resize {
161 memory,
162 slots,
163 cpus,
164 };
165 if to == from || slots == 0 || !within(profile, from, to) {
166 continue;
167 }
168 budget.take(&advised.name, profile, from, to);
169 changes.push(SizingChange {
170 at: now,
171 profile: advised.name.clone(),
172 from,
173 to,
174 reason: reasons.join("; "),
175 uncertain: false,
176 });
177 }
178 Look { changes, refused }
179}
180
181#[derive(Debug, Clone, Copy, PartialEq, Eq)]
183enum Lowering {
184 Allowed,
185 Held,
186}
187
188struct Budget {
190 caps: u64,
192 cpus: u64,
194 slots: BTreeMap<String, u32>,
196}
197
198impl Budget {
199 fn take(&mut self, name: &str, profile: &Profile, from: Resize, to: Resize) {
201 self.caps = self
202 .caps
203 .saturating_sub(from.memory.saturating_mul(u64::from(from.slots)))
204 .saturating_add(to.memory.saturating_mul(u64::from(to.slots)));
205 let millicores = |resize: Resize| {
206 u64::from(resize.cpus.unwrap_or(profile.cpus)) * 1000 * u64::from(resize.slots)
207 };
208 self.cpus = self
209 .cpus
210 .saturating_sub(millicores(from))
211 .saturating_add(millicores(to));
212 self.slots.insert(name.to_owned(), to.slots);
213 }
214
215 fn new(config: &Config, profiles: &[ProfileSizing]) -> Budget {
217 Budget {
218 caps: config.memory_caps().map_or(u64::MAX, |caps| caps.0),
219 cpus: config
220 .profiles
221 .values()
222 .map(|profile| u64::from(profile.cpus) * 1000 * u64::from(profile.slots))
223 .sum(),
224 slots: profiles
225 .iter()
226 .map(|advised| (advised.name.clone(), advised.slots))
227 .collect(),
228 }
229 }
230}
231
232struct Fresh {
235 start: u64,
237 jobs: u64,
238 wait: u64,
240 busy: u32,
241}
242
243impl Fresh {
244 fn read(
246 history: &History,
247 advised: &ProfileSizing,
248 last: Option<u64>,
249 window: Window,
250 ) -> Fresh {
251 let start = last.map_or(window.since, |at| at.saturating_mul(1000).max(window.since));
252 let jobs: Vec<_> = history
253 .jobs
254 .iter()
255 .filter(|job| job.profile == advised.name && job.queued_at >= start)
256 .collect();
257 Fresh {
258 start,
259 jobs: jobs.len() as u64,
260 wait: waits(&jobs).p95,
261 busy: busy(
262 history,
263 &advised.name,
264 advised.slots,
265 first_runner(history, start),
266 window.now,
267 ),
268 }
269 }
270
271 fn wants_slots(&self) -> bool {
274 self.jobs >= RAISE_RUNS && (self.wait >= WAITED || self.busy >= BUSY)
275 }
276
277 fn waited(&self, changed: bool) -> String {
279 format!(
280 "{} jobs {}waited {}s for a slot (p95) and slots were busy {}%; one more fits the slice beside the \
281 other profiles'",
282 self.jobs,
283 if changed {
284 "since its last change "
285 } else {
286 ""
287 },
288 self.wait,
289 self.busy
290 )
291 }
292}
293
294fn memory_reason(
298 sizing: &Sizing,
299 advised: &ProfileSizing,
300 memory: u64,
301 slots: u32,
302 max_step: u32,
303) -> String {
304 let made = ProfileSizing {
305 advice: match advised.advice {
306 Advice::Lower { .. } => Advice::Lower { memory, slots },
307 Advice::Raise { .. } => Advice::Raise { memory, slots },
308 other => other,
309 },
310 ..advised.clone()
311 };
312 let line = sizing.memory_line(&made);
313 match advised.advice.change() {
314 Some(advice) if advice.memory != memory => format!(
315 "{line}; a step toward {}, at most {max_step}% a change",
316 ByteSize(advice.memory)
317 ),
318 _ => line,
319 }
320}
321
322fn within(profile: &Profile, from: Resize, to: Resize) -> bool {
325 profile.min_memory.is_none_or(|min| to.memory >= min.0)
326 && profile.max_memory.is_none_or(|max| to.memory <= max.0)
327 && (to.slots <= from.slots || profile.max_slots.is_none_or(|max| to.slots <= max))
328}
329
330fn met_cap(history: &History, profile: &str) -> bool {
332 history
333 .jobs
334 .iter()
335 .chain(&history.cancelled)
336 .filter(|job| job.profile == profile)
337 .any(|job| job.oom_killed || job.memory_cap.is_some_and(|cap| job.met_cap(cap)))
338}
339
340fn memory(
343 config: &Config,
344 profile: &Profile,
345 advice: Advice,
346 met_cap: bool,
347 lowering: Lowering,
348) -> u64 {
349 let current = profile.memory.0;
350 let step = current.saturating_mul(u64::from(config.sizing.max_step)) / 100;
351 match advice {
352 Advice::Lower { memory, .. } if !met_cap && lowering == Lowering::Allowed => {
353 let floor = current
354 .saturating_sub(step)
355 .div_ceil(GRANULE)
356 .saturating_mul(GRANULE);
357 let bound = profile.min_memory.map_or(0, |min| min.0);
358 memory.max(floor).max(bound).min(current)
359 }
360 Advice::Raise { memory, .. } => {
361 let ceiling = current.saturating_add(step) / GRANULE * GRANULE;
362 let bound = profile.max_memory.map_or(u64::MAX, |max| max.0);
363 memory.min(ceiling).min(bound).max(current)
364 }
365 _ => current,
366 }
367}
368
369fn caps_slots(
374 profile: &Profile,
375 advised: &ProfileSizing,
376 fresh: &Fresh,
377 memory: u64,
378 context: &Context<'_>,
379 budget: &Budget,
380 running: u32,
381) -> u32 {
382 let current = profile.slots;
383 let max_slots = profile.max_slots.unwrap_or(u32::MAX);
384 let fit = |room: u64| u32::try_from(room / memory.max(1)).unwrap_or(u32::MAX);
385 if memory < profile.memory.0 && advised.wants_slots {
386 let share = profile.memory.0.saturating_mul(u64::from(current));
387 return fit(share).min(max_slots).max(current);
388 }
389 let Some(memory_max) = context.memory_max else {
390 return current;
391 };
392 if memory > profile.memory.0 {
393 let share = profile.memory.0.saturating_mul(u64::from(current));
394 let room = memory_max.saturating_sub(budget.caps.saturating_sub(share));
395 let slots = fit(room).min(current);
397 return if slots < running { 0 } else { slots };
398 }
399 if fresh.wants_slots()
400 && current < max_slots
401 && budget.caps.saturating_add(memory) <= memory_max
402 {
403 return current + 1;
404 }
405 current
406}
407
408fn measured_slots(
413 config: &Config,
414 profiles: &[ProfileSizing],
415 advised: &ProfileSizing,
416 fresh: &Fresh,
417 context: &Context<'_>,
418 budget: &Budget,
419) -> u32 {
420 let current = advised.slots;
421 let max_slots = config
422 .profiles
423 .get(&advised.name)
424 .and_then(|profile| profile.max_slots);
425 if !fresh.wants_slots() || max_slots.is_some_and(|max| max <= current) {
426 return current;
427 }
428 let Some(memory_max) = context.memory_max else {
429 return if max_slots.is_some() {
430 current + 1
431 } else {
432 current
433 };
434 };
435 let peaks: u64 = profiles
436 .iter()
437 .map(|other| {
438 let peak = if matches!(other.advice, Advice::TooFew { .. }) {
439 other.memory
440 } else {
441 other.peaks.memory.p95
442 };
443 let slots = if other.name == advised.name {
444 current + 1
445 } else {
446 budget
447 .slots
448 .get(&other.name)
449 .copied()
450 .unwrap_or(other.slots)
451 };
452 peak.saturating_mul(u64::from(slots))
453 })
454 .fold(config.mirror_memory(), u64::saturating_add);
455 if peaks <= memory_max {
456 current + 1
457 } else {
458 current
459 }
460}
461
462#[expect(
466 clippy::too_many_arguments,
467 reason = "each is one thing a CPU change reads; a struct for them would be read here only"
468)]
469fn cpus_and_slots(
470 config: &Config,
471 profile: &Profile,
472 advised: &ProfileSizing,
473 history: &History,
474 since: u64,
475 context: &Context<'_>,
476 budget: &Budget,
477 slots: u32,
478 lowering: Lowering,
479) -> (u32, CpuStep) {
480 let size = |slots: u32| {
481 cpus(
482 config, profile, advised, history, since, context, budget, slots, lowering,
483 )
484 };
485 let fit = |slots: u32, cpus: &CpuStep| {
487 let cpus = match cpus {
488 CpuStep::To(cpus) => cpus.to,
489 CpuStep::Keep | CpuStep::Refused(_) => profile.cpus,
490 };
491 if slots > profile.slots {
492 cpu_slots(config, profile, context, budget, cpus)
493 .map_or(slots, |fit| slots.min(fit.max(profile.slots)))
494 } else {
495 slots
496 }
497 };
498 let mut cpus = size(slots);
499 let fitted = fit(slots, &cpus);
500 if fitted < slots {
501 cpus = size(fitted);
502 }
503 (fit(fitted, &cpus), cpus)
504}
505
506fn cpu_slots(
510 config: &Config,
511 profile: &Profile,
512 context: &Context<'_>,
513 budget: &Budget,
514 cpus: u32,
515) -> Option<u32> {
516 let (Admission::Caps, Some(overcommit)) = (config.host.admission, config.host.cpu_overcommit)
517 else {
518 return None;
519 };
520 let Some(slice) = context.cpus else {
521 return Some(profile.slots);
522 };
523 let others = budget
524 .cpus
525 .saturating_sub(u64::from(profile.cpus) * 1000 * u64::from(profile.slots));
526 let room = cpu_budget(slice, overcommit).saturating_sub(others);
527 Some(u32::try_from(room / (u64::from(cpus.max(1)) * 1000)).unwrap_or(u32::MAX))
528}
529
530enum CpuStep {
532 Keep,
533 To(Cpus),
534 Refused(String),
536}
537
538impl CpuStep {
539 fn taken(
541 self,
542 profile: &str,
543 reasons: &mut Vec<String>,
544 refused: &mut Vec<SizingRefused>,
545 ) -> Option<u32> {
546 match self {
547 CpuStep::To(cpus) => {
548 reasons.push(cpus.reason);
549 Some(cpus.to)
550 }
551 CpuStep::Refused(reason) => {
552 refused.push(SizingRefused {
553 profile: profile.to_owned(),
554 reason,
555 });
556 None
557 }
558 CpuStep::Keep => None,
559 }
560 }
561}
562
563struct Cpus {
565 to: u32,
566 reason: String,
567}
568
569type JobKey<'run> = (&'run str, Option<&'run str>, &'run str);
571
572fn cpus_since(set: &[CpusSet], profile: &str, cpus: u32, now: u64) -> u64 {
575 let mut since = None;
576 let mut seen = false;
577 for event in set.iter().filter(|event| event.profile == profile) {
578 seen = true;
579 if event.cpus != cpus {
580 since = None;
581 } else if since.is_none() {
582 since = Some(event.at);
583 }
584 }
585 since.unwrap_or(if seen { now } else { 0 })
586}
587
588fn averages<'run>(
593 history: &'run History,
594 profile: &str,
595 cap: u64,
596 since: u64,
597) -> BTreeMap<JobKey<'run>, Vec<u64>> {
598 let mut jobs: BTreeMap<JobKey<'run>, Vec<u64>> = BTreeMap::new();
599 for job in &history.jobs {
600 if job.profile == profile
601 && job.cpu_cap == Some(cap)
602 && job.succeeded
603 && job.queued_at >= since
604 {
605 let averages = jobs
606 .entry((&job.repository, job.workflow.as_deref(), &job.name))
607 .or_default();
608 averages.extend(job.cpu_average.filter(|average| *average > 0));
609 }
610 }
611 jobs
612}
613
614fn job_named(job: JobKey<'_>) -> String {
616 let (repository, workflow, name) = job;
617 format!("{repository} {}", job_title(workflow, name, None))
618}
619
620#[expect(
628 clippy::too_many_arguments,
629 reason = "each is one thing a CPU change reads; a struct for them would be read here only"
630)]
631fn cpus(
632 config: &Config,
633 profile: &Profile,
634 advised: &ProfileSizing,
635 history: &History,
636 since: u64,
637 context: &Context<'_>,
638 budget: &Budget,
639 slots: u32,
640 lowering: Lowering,
641) -> CpuStep {
642 let current = profile.cpus;
643 let cap = u64::from(current) * 1000;
644 let since = since.max(cpus_since(
645 context.cpus_set,
646 &advised.name,
647 current,
648 context.window.now,
649 ));
650 let jobs = averages(history, &advised.name, cap, since);
651 let step = (current.saturating_mul(config.sizing.max_step) / 100).max(1);
652 let bound = jobs
653 .iter()
654 .filter(|(_, averages)| averages.len() as u64 >= RAISE_RUNS)
655 .map(|(job, averages)| (*job, averages.len(), spread(averages.clone()).p50))
656 .filter(|(_, _, median)| median.saturating_mul(100) >= cap.saturating_mul(BOUND))
657 .max_by_key(|(_, _, median)| *median);
658 if let Some((job, runs, median)) = bound {
659 let held = format!(
660 "{} runs of {} averaged {} of its {current} CPUs (median): the cap holds them back",
661 runs,
662 job_named(job),
663 cores(median)
664 );
665 let Some(slice) = context.cpus else {
666 return CpuStep::Refused(format!("{held}; the slice's CPUs could not be read"));
667 };
668 let slice_cpus = u32::try_from(slice / 1000).unwrap_or(u32::MAX);
669 let mut to = current.saturating_add(step).min(slice_cpus);
670 let mut why = format!("it has all {slice_cpus} of the slice's CPUs");
671 if let (Admission::Caps, Some(overcommit)) =
672 (config.host.admission, config.host.cpu_overcommit)
673 {
674 let others = budget
675 .cpus
676 .saturating_sub(cap.saturating_mul(u64::from(profile.slots)));
677 let room = cpu_budget(slice, overcommit).saturating_sub(others);
678 let fit = room / 1000 / u64::from(slots.max(1));
679 let fit = u32::try_from(fit).unwrap_or(u32::MAX);
680 if fit < to {
681 to = fit;
682 why = format!(
683 "{} CPUs for each of its {slots} slots do not fit the CPU budget beside the other profiles'",
684 current.saturating_add(1)
685 );
686 }
687 }
688 return if to > current {
689 CpuStep::To(Cpus { to, reason: held })
690 } else {
691 CpuStep::Refused(format!("{held}; {why}"))
692 };
693 }
694 let runs: usize = jobs.values().map(Vec::len).sum();
695 if lowering == Lowering::Held
697 || (runs as u64) < u64::from(config.sizing.min_jobs)
698 || jobs.values().any(Vec::is_empty)
699 {
700 return CpuStep::Keep;
701 }
702 let Some((job, heaviest)) = jobs
703 .iter()
704 .map(|(job, averages)| (*job, spread(averages.clone()).p95))
705 .max_by_key(|(_, p95)| *p95)
706 else {
707 return CpuStep::Keep;
708 };
709 let need = heaviest.saturating_mul(HEADROOM);
710 let target = u32::try_from(need.div_ceil(1000))
711 .unwrap_or(u32::MAX)
712 .max(1);
713 if target >= current {
714 return CpuStep::Keep;
715 }
716 let to = target.max(current.saturating_sub(step)).max(1);
717 let mut reason = format!(
718 "{runs} runs at its {current} CPUs: the heaviest job, {}, averaged {} CPUs (p95); twice that is {target}",
719 job_named(job),
720 cores(heaviest)
721 );
722 if to != target {
723 let _ = write!(
724 reason,
725 "; a step toward {target}, at most {}% a change",
726 config.sizing.max_step
727 );
728 }
729 CpuStep::To(Cpus { to, reason })
730}