1use std::collections::BTreeMap;
11use std::fmt::Write as _;
12
13use crate::api::{
14 Advice, ProfileSizing, Resize, Sizing, SizingChange, SizingRefused, SizingRevert, cores,
15 job_title,
16};
17use crate::config::{Admission, ByteSize, Config, Profile, cpu_budget};
18
19use super::{
20 BUSY, CpusSet, GRANULE, History, WAITED, Window, advise, busy, first_runner, spread, waits,
21};
22
23const RAISE_RUNS: u64 = 3;
26
27const BOUND: u64 = 70;
31
32const HEADROOM: u64 = 2;
36
37#[derive(Debug, Clone, Copy)]
39pub struct Context<'run> {
40 pub memory_max: Option<u64>,
42 pub cpus: Option<u64>,
44 pub window: Window,
46 pub last: &'run BTreeMap<String, u64>,
48 pub cpus_set: &'run [CpusSet],
51 pub running: &'run BTreeMap<String, u32>,
53}
54
55pub fn changes(config: &Config, history: &History, context: Context<'_>) -> Vec<SizingChange> {
74 next_look(config, history, context).changes
75}
76
77#[derive(Debug, Clone, Default, PartialEq, Eq)]
79pub struct Look {
80 pub changes: Vec<SizingChange>,
82 pub refused: Vec<SizingRefused>,
85}
86
87pub fn next_look(config: &Config, history: &History, context: Context<'_>) -> Look {
89 let rules = &config.sizing;
90 if !rules.auto {
91 return Look::default();
92 }
93 let window = context.window;
94 let sizing = advise(config, context.memory_max, history, window);
95 let now = window.now / 1000;
96 let mut budget = Budget::new(config, &sizing.profiles);
97 let mut changes = Vec::new();
98 let mut refused = Vec::new();
99 for advised in &sizing.profiles {
100 let Some(profile) = config.profiles.get(&advised.name) else {
101 continue;
102 };
103 let last = context.last.get(&advised.name).copied();
104 let since = last.map_or(u64::MAX, |at| now.saturating_sub(at));
105 if let Some(revert) = &advised.revert {
106 match reverted(config, profile, advised, revert, &sizing, &context, now) {
107 Revert::Write(change) => {
108 budget.take(&advised.name, profile, change.from, change.to);
109 changes.push(change);
110 continue;
111 }
112 Revert::Wait => continue,
113 Revert::Outside => {}
115 }
116 }
117 if since < rules.raise_after().as_secs() {
118 continue;
119 }
120 let lowering = if since >= rules.interval.as_secs() {
121 Lowering::Allowed
122 } else {
123 Lowering::Held
124 };
125 let fresh = Fresh::read(history, advised, last, window);
126 let mut reasons = Vec::new();
127 let memory = memory(config, profile, advised.advice, lowering);
128 let running = context.running.get(&advised.name).copied().unwrap_or(0);
129 let slots = match config.host.admission {
130 Admission::Caps => {
131 caps_slots(profile, advised, &fresh, memory, &context, &budget, running)
132 }
133 Admission::Measured => {
134 measured_slots(config, &sizing.profiles, advised, &fresh, &context, &budget)
135 }
136 };
137 let (slots, cpus) = cpus_and_slots(
138 config,
139 profile,
140 advised,
141 history,
142 fresh.start,
143 &context,
144 &budget,
145 slots,
146 lowering,
147 );
148 if memory != profile.memory.0 {
149 reasons.push(memory_reason(
150 &sizing,
151 advised,
152 memory,
153 slots,
154 config.sizing.max_step,
155 ));
156 }
157 let advised_slots = config.host.admission == Admission::Caps && memory < profile.memory.0;
159 if slots > profile.slots && !advised_slots {
160 reasons.push(fresh.waited(last.is_some()));
161 }
162 let cpus = cpus.taken(&advised.name, &mut reasons, &mut refused);
163 let from = Resize {
164 memory: profile.memory.0,
165 slots: profile.slots,
166 cpus: cpus.map(|_| profile.cpus),
167 };
168 let to = Resize {
169 memory,
170 slots,
171 cpus,
172 };
173 if to == from || slots == 0 || !within(profile, from, to) {
174 continue;
175 }
176 budget.take(&advised.name, profile, from, to);
177 changes.push(SizingChange {
178 at: now,
179 profile: advised.name.clone(),
180 from,
181 to,
182 reason: reasons.join("; "),
183 uncertain: false,
184 });
185 }
186 Look { changes, refused }
187}
188
189#[derive(Debug, Clone, Copy, PartialEq, Eq)]
191enum Lowering {
192 Allowed,
193 Held,
194}
195
196struct Budget {
198 caps: u64,
200 cpus: u64,
202 slots: BTreeMap<String, u32>,
204}
205
206impl Budget {
207 fn take(&mut self, name: &str, profile: &Profile, from: Resize, to: Resize) {
209 self.caps = self
210 .caps
211 .saturating_sub(from.memory.saturating_mul(u64::from(from.slots)))
212 .saturating_add(to.memory.saturating_mul(u64::from(to.slots)));
213 let millicores = |resize: Resize| {
214 u64::from(resize.cpus.unwrap_or(profile.cpus)) * 1000 * u64::from(resize.slots)
215 };
216 self.cpus = self
217 .cpus
218 .saturating_sub(millicores(from))
219 .saturating_add(millicores(to));
220 self.slots.insert(name.to_owned(), to.slots);
221 }
222
223 fn new(config: &Config, profiles: &[ProfileSizing]) -> Budget {
225 Budget {
226 caps: config.memory_caps().map_or(u64::MAX, |caps| caps.0),
227 cpus: config
228 .profiles
229 .values()
230 .map(|profile| u64::from(profile.cpus) * 1000 * u64::from(profile.slots))
231 .sum(),
232 slots: profiles
233 .iter()
234 .map(|advised| (advised.name.clone(), advised.slots))
235 .collect(),
236 }
237 }
238}
239
240struct Fresh {
243 start: u64,
245 jobs: u64,
246 wait: u64,
248 busy: u32,
249}
250
251impl Fresh {
252 fn read(
254 history: &History,
255 advised: &ProfileSizing,
256 last: Option<u64>,
257 window: Window,
258 ) -> Fresh {
259 let start = last.map_or(window.since, |at| at.saturating_mul(1000).max(window.since));
260 let jobs: Vec<_> = history
261 .jobs
262 .iter()
263 .filter(|job| job.profile == advised.name && job.queued_at >= start)
264 .collect();
265 Fresh {
266 start,
267 jobs: jobs.len() as u64,
268 wait: waits(&jobs).p95,
269 busy: busy(
270 history,
271 &advised.name,
272 advised.slots,
273 first_runner(history, start),
274 window.now,
275 ),
276 }
277 }
278
279 fn wants_slots(&self) -> bool {
282 self.jobs >= RAISE_RUNS && (self.wait >= WAITED || self.busy >= BUSY)
283 }
284
285 fn waited(&self, changed: bool) -> String {
287 format!(
288 "{} jobs {}waited {}s for a slot (p95) and slots were busy {}%; one more fits the slice beside the \
289 other profiles'",
290 self.jobs,
291 if changed {
292 "since its last change "
293 } else {
294 ""
295 },
296 self.wait,
297 self.busy
298 )
299 }
300}
301
302fn memory_reason(
306 sizing: &Sizing,
307 advised: &ProfileSizing,
308 memory: u64,
309 slots: u32,
310 max_step: u32,
311) -> String {
312 let made = ProfileSizing {
313 advice: match advised.advice {
314 Advice::Lower { .. } => Advice::Lower { memory, slots },
315 Advice::Raise { .. } => Advice::Raise { memory, slots },
316 other => other,
317 },
318 revert: advised.revert.clone().filter(|revert| revert.to == memory),
320 ..advised.clone()
321 };
322 let line = sizing.memory_line(&made);
323 match advised.advice.change() {
324 Some(advice) if advice.memory != memory && memory > advised.memory => format!(
325 "{line}; as far as its max_memory allows, short of {}",
326 ByteSize(advice.memory)
327 ),
328 Some(advice) if advice.memory != memory => format!(
329 "{line}; a step toward {}, at most {max_step}% a change",
330 ByteSize(advice.memory)
331 ),
332 _ => line,
333 }
334}
335
336fn within(profile: &Profile, from: Resize, to: Resize) -> bool {
339 profile.min_memory.is_none_or(|min| to.memory >= min.0)
340 && profile.max_memory.is_none_or(|max| to.memory <= max.0)
341 && (to.slots <= from.slots || profile.max_slots.is_none_or(|max| to.slots <= max))
342}
343
344enum Revert {
346 Write(SizingChange),
348 Wait,
350 Outside,
352}
353
354fn reverted(
358 config: &Config,
359 profile: &Profile,
360 advised: &ProfileSizing,
361 revert: &SizingRevert,
362 sizing: &Sizing,
363 context: &Context<'_>,
364 now: u64,
365) -> Revert {
366 let Advice::Raise { memory, slots } = advised.advice else {
367 return Revert::Outside;
368 };
369 let running = context.running.get(&advised.name).copied().unwrap_or(0);
370 let from = Resize {
371 memory: profile.memory.0,
372 slots: profile.slots,
373 cpus: None,
374 };
375 let to = Resize {
376 memory,
377 slots,
378 cpus: None,
379 };
380 if memory != revert.to || !within(profile, from, to) {
381 return Revert::Outside;
382 }
383 if config.host.admission == Admission::Caps && running > slots {
384 return Revert::Wait;
385 }
386 Revert::Write(SizingChange {
387 at: now,
388 profile: advised.name.clone(),
389 from,
390 to,
391 reason: sizing.memory_line(advised),
392 uncertain: false,
393 })
394}
395
396fn memory(config: &Config, profile: &Profile, advice: Advice, lowering: Lowering) -> u64 {
400 let current = profile.memory.0;
401 let step = current.saturating_mul(u64::from(config.sizing.max_step)) / 100;
402 match advice {
403 Advice::Lower { memory, .. } if lowering == Lowering::Allowed => {
404 let floor = current
405 .saturating_sub(step)
406 .div_ceil(GRANULE)
407 .saturating_mul(GRANULE);
408 let bound = profile.min_memory.map_or(0, |min| min.0);
409 memory.max(floor).max(bound).min(current)
410 }
411 Advice::Raise { memory, .. } => {
412 let bound = profile.max_memory.map_or(u64::MAX, |max| max.0);
413 memory.min(bound).max(current)
414 }
415 _ => current,
416 }
417}
418
419fn caps_slots(
424 profile: &Profile,
425 advised: &ProfileSizing,
426 fresh: &Fresh,
427 memory: u64,
428 context: &Context<'_>,
429 budget: &Budget,
430 running: u32,
431) -> u32 {
432 let current = profile.slots;
433 let max_slots = profile.max_slots.unwrap_or(u32::MAX);
434 let fit = |room: u64| u32::try_from(room / memory.max(1)).unwrap_or(u32::MAX);
435 if memory < profile.memory.0 && advised.wants_slots {
436 let share = profile.memory.0.saturating_mul(u64::from(current));
437 return fit(share).min(max_slots).max(current);
438 }
439 let Some(memory_max) = context.memory_max else {
440 return current;
441 };
442 if memory > profile.memory.0 {
443 let share = profile.memory.0.saturating_mul(u64::from(current));
444 let room = memory_max.saturating_sub(budget.caps.saturating_sub(share));
445 let slots = fit(room).min(current);
447 return if slots < running { 0 } else { slots };
448 }
449 if fresh.wants_slots()
450 && current < max_slots
451 && budget.caps.saturating_add(memory) <= memory_max
452 {
453 return current + 1;
454 }
455 current
456}
457
458fn measured_slots(
463 config: &Config,
464 profiles: &[ProfileSizing],
465 advised: &ProfileSizing,
466 fresh: &Fresh,
467 context: &Context<'_>,
468 budget: &Budget,
469) -> u32 {
470 let current = advised.slots;
471 let max_slots = config
472 .profiles
473 .get(&advised.name)
474 .and_then(|profile| profile.max_slots);
475 if !fresh.wants_slots() || max_slots.is_some_and(|max| max <= current) {
476 return current;
477 }
478 let Some(memory_max) = context.memory_max else {
479 return if max_slots.is_some() {
480 current + 1
481 } else {
482 current
483 };
484 };
485 let peaks: u64 = profiles
486 .iter()
487 .map(|other| {
488 let peak = if matches!(other.advice, Advice::TooFew { .. }) {
489 other.memory
490 } else {
491 other.peaks.memory.p95
492 };
493 let slots = if other.name == advised.name {
494 current + 1
495 } else {
496 budget
497 .slots
498 .get(&other.name)
499 .copied()
500 .unwrap_or(other.slots)
501 };
502 peak.saturating_mul(u64::from(slots))
503 })
504 .fold(config.mirror_memory(), u64::saturating_add);
505 if peaks <= memory_max {
506 current + 1
507 } else {
508 current
509 }
510}
511
512#[expect(
516 clippy::too_many_arguments,
517 reason = "each is one thing a CPU change reads; a struct for them would be read here only"
518)]
519fn cpus_and_slots(
520 config: &Config,
521 profile: &Profile,
522 advised: &ProfileSizing,
523 history: &History,
524 since: u64,
525 context: &Context<'_>,
526 budget: &Budget,
527 slots: u32,
528 lowering: Lowering,
529) -> (u32, CpuStep) {
530 let size = |slots: u32| {
531 cpus(
532 config, profile, advised, history, since, context, budget, slots, lowering,
533 )
534 };
535 let fit = |slots: u32, cpus: &CpuStep| {
537 let cpus = match cpus {
538 CpuStep::To(cpus) => cpus.to,
539 CpuStep::Keep | CpuStep::Refused(_) => profile.cpus,
540 };
541 if slots > profile.slots {
542 cpu_slots(config, profile, context, budget, cpus)
543 .map_or(slots, |fit| slots.min(fit.max(profile.slots)))
544 } else {
545 slots
546 }
547 };
548 let mut cpus = size(slots);
549 let fitted = fit(slots, &cpus);
550 if fitted < slots {
551 cpus = size(fitted);
552 }
553 (fit(fitted, &cpus), cpus)
554}
555
556fn cpu_slots(
560 config: &Config,
561 profile: &Profile,
562 context: &Context<'_>,
563 budget: &Budget,
564 cpus: u32,
565) -> Option<u32> {
566 let (Admission::Caps, Some(overcommit)) = (config.host.admission, config.host.cpu_overcommit)
567 else {
568 return None;
569 };
570 let Some(slice) = context.cpus else {
571 return Some(profile.slots);
572 };
573 let others = budget
574 .cpus
575 .saturating_sub(u64::from(profile.cpus) * 1000 * u64::from(profile.slots));
576 let room = cpu_budget(slice, overcommit).saturating_sub(others);
577 Some(u32::try_from(room / (u64::from(cpus.max(1)) * 1000)).unwrap_or(u32::MAX))
578}
579
580enum CpuStep {
582 Keep,
583 To(Cpus),
584 Refused(String),
586}
587
588impl CpuStep {
589 fn taken(
591 self,
592 profile: &str,
593 reasons: &mut Vec<String>,
594 refused: &mut Vec<SizingRefused>,
595 ) -> Option<u32> {
596 match self {
597 CpuStep::To(cpus) => {
598 reasons.push(cpus.reason);
599 Some(cpus.to)
600 }
601 CpuStep::Refused(reason) => {
602 refused.push(SizingRefused {
603 profile: profile.to_owned(),
604 reason,
605 });
606 None
607 }
608 CpuStep::Keep => None,
609 }
610 }
611}
612
613struct Cpus {
615 to: u32,
616 reason: String,
617}
618
619type JobKey<'run> = (&'run str, Option<&'run str>, &'run str);
621
622fn cpus_since(set: &[CpusSet], profile: &str, cpus: u32, now: u64) -> u64 {
625 let mut since = None;
626 let mut seen = false;
627 for event in set.iter().filter(|event| event.profile == profile) {
628 seen = true;
629 if event.cpus != cpus {
630 since = None;
631 } else if since.is_none() {
632 since = Some(event.at);
633 }
634 }
635 since.unwrap_or(if seen { now } else { 0 })
636}
637
638fn averages<'run>(
643 history: &'run History,
644 profile: &str,
645 cap: u64,
646 since: u64,
647) -> BTreeMap<JobKey<'run>, Vec<u64>> {
648 let mut jobs: BTreeMap<JobKey<'run>, Vec<u64>> = BTreeMap::new();
649 for job in &history.jobs {
650 if job.profile == profile
651 && job.cpu_cap == Some(cap)
652 && job.succeeded
653 && job.queued_at >= since
654 {
655 let averages = jobs
656 .entry((&job.repository, job.workflow.as_deref(), &job.name))
657 .or_default();
658 averages.extend(job.cpu_average.filter(|average| *average > 0));
659 }
660 }
661 jobs
662}
663
664fn job_named(job: JobKey<'_>) -> String {
666 let (repository, workflow, name) = job;
667 format!("{repository} {}", job_title(workflow, name, None))
668}
669
670#[expect(
678 clippy::too_many_arguments,
679 reason = "each is one thing a CPU change reads; a struct for them would be read here only"
680)]
681fn cpus(
682 config: &Config,
683 profile: &Profile,
684 advised: &ProfileSizing,
685 history: &History,
686 since: u64,
687 context: &Context<'_>,
688 budget: &Budget,
689 slots: u32,
690 lowering: Lowering,
691) -> CpuStep {
692 let current = profile.cpus;
693 let cap = u64::from(current) * 1000;
694 let since = since.max(cpus_since(
695 context.cpus_set,
696 &advised.name,
697 current,
698 context.window.now,
699 ));
700 let jobs = averages(history, &advised.name, cap, since);
701 let step = (current.saturating_mul(config.sizing.max_step) / 100).max(1);
702 let bound = jobs
703 .iter()
704 .filter(|(_, averages)| averages.len() as u64 >= RAISE_RUNS)
705 .map(|(job, averages)| (*job, averages.len(), spread(averages.clone()).p50))
706 .filter(|(_, _, median)| median.saturating_mul(100) >= cap.saturating_mul(BOUND))
707 .max_by_key(|(_, _, median)| *median);
708 if let Some((job, runs, median)) = bound {
709 let held = format!(
710 "{} runs of {} averaged {} of its {current} CPUs (median): the cap holds them back",
711 runs,
712 job_named(job),
713 cores(median)
714 );
715 let Some(slice) = context.cpus else {
716 return CpuStep::Refused(format!("{held}; the slice's CPUs could not be read"));
717 };
718 let slice_cpus = u32::try_from(slice / 1000).unwrap_or(u32::MAX);
719 let mut to = current.saturating_add(step).min(slice_cpus);
720 let mut why = format!("it has all {slice_cpus} of the slice's CPUs");
721 if let (Admission::Caps, Some(overcommit)) =
722 (config.host.admission, config.host.cpu_overcommit)
723 {
724 let others = budget
725 .cpus
726 .saturating_sub(cap.saturating_mul(u64::from(profile.slots)));
727 let room = cpu_budget(slice, overcommit).saturating_sub(others);
728 let fit = room / 1000 / u64::from(slots.max(1));
729 let fit = u32::try_from(fit).unwrap_or(u32::MAX);
730 if fit < to {
731 to = fit;
732 why = format!(
733 "{} CPUs for each of its {slots} slots do not fit the CPU budget beside the other profiles'",
734 current.saturating_add(1)
735 );
736 }
737 }
738 return if to > current {
739 CpuStep::To(Cpus { to, reason: held })
740 } else {
741 CpuStep::Refused(format!("{held}; {why}"))
742 };
743 }
744 let runs: usize = jobs.values().map(Vec::len).sum();
745 if lowering == Lowering::Held
747 || (runs as u64) < u64::from(config.sizing.min_jobs)
748 || jobs.values().any(Vec::is_empty)
749 {
750 return CpuStep::Keep;
751 }
752 let Some((job, heaviest)) = jobs
753 .iter()
754 .map(|(job, averages)| (*job, spread(averages.clone()).p95))
755 .max_by_key(|(_, p95)| *p95)
756 else {
757 return CpuStep::Keep;
758 };
759 let need = heaviest.saturating_mul(HEADROOM);
760 let target = u32::try_from(need.div_ceil(1000))
761 .unwrap_or(u32::MAX)
762 .max(1);
763 if target >= current {
764 return CpuStep::Keep;
765 }
766 let to = target.max(current.saturating_sub(step)).max(1);
767 let mut reason = format!(
768 "{runs} runs at its {current} CPUs: the heaviest job, {}, averaged {} CPUs (p95); twice that is {target}",
769 job_named(job),
770 cores(heaviest)
771 );
772 if to != target {
773 let _ = write!(
774 reason,
775 "; a step toward {target}, at most {}% a change",
776 config.sizing.max_step
777 );
778 }
779 CpuStep::To(Cpus { to, reason })
780}