1use std::collections::{BTreeMap, BTreeSet};
6use std::time::{Duration, SystemTime};
7
8use crate::api::Waiting;
9use crate::github::QueuedJob;
10
11pub mod engine;
12pub mod warm;
13
14pub const IDLE_TIMEOUT: Duration = Duration::from_secs(300);
17
18pub const CPU_PRIORITY_AFTER: Duration = Duration::from_secs(120);
21
22#[derive(Debug, Clone)]
24pub struct Repository {
25 pub name: String,
27 pub profile: String,
29 pub labels: Vec<String>,
31 pub paused: bool,
33 pub held_until: Option<SystemTime>,
35 pub queued: Vec<QueuedJob>,
37 pub runner_memory: Option<u64>,
40 pub runner_cpu: Option<u64>,
43 pub cpu_waiting_since: Option<SystemTime>,
45 pub warm: usize,
47}
48
49impl Repository {
50 fn starved_of_cpu(&self, now: SystemTime) -> bool {
52 self.cpu_waiting_since
53 .and_then(|since| now.duration_since(since).ok())
54 .is_some_and(|waited| waited >= CPU_PRIORITY_AFTER)
55 }
56}
57
58#[derive(Debug, Clone)]
60pub struct Profile {
61 pub memory: u64,
63 pub cpus: u64,
66 pub preparing: Option<Preparing>,
68 pub upkeep_due: bool,
71 pub slots: Vec<Slot>,
73}
74
75#[derive(Debug, Clone, Copy, PartialEq, Eq)]
77pub enum Preparing {
78 Reseeding,
80 PullingImage,
82}
83
84#[derive(Debug, Clone, PartialEq, Eq)]
86pub enum Slot {
87 Free,
89 Runner(RunnerSlot),
91 Closed,
94}
95
96#[derive(Debug, Clone, PartialEq, Eq)]
98pub struct RunnerSlot {
99 pub repository: String,
101 pub busy: bool,
103 pub started_at: SystemTime,
105 pub memory: u64,
107 pub cpu: u64,
109}
110
111#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
113pub struct SlotRef {
114 pub profile: String,
116 pub index: usize,
118}
119
120#[derive(Debug, Clone, PartialEq, Eq)]
122pub enum Decision {
123 Start {
125 slot: SlotRef,
127 repository: String,
129 memory: u64,
131 cpu: u64,
133 warm: bool,
135 replacing: Vec<SlotRef>,
138 },
139 HeldForCpu {
141 repository: String,
143 },
144 Stop {
147 slot: SlotRef,
149 },
150}
151
152#[derive(Debug, Clone, Default, PartialEq, Eq)]
154pub struct Plan {
155 pub decisions: Vec<Decision>,
157 pub waiting: BTreeMap<u64, Waiting>,
161 pub warm: BTreeSet<SlotRef>,
163}
164
165#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
167pub struct Budget {
168 pub memory: Option<u64>,
170 pub cpu: Option<u64>,
172 pub held: bool,
175}
176
177pub fn matches(job: &QueuedJob, labels: &[String]) -> bool {
180 job.labels.iter().all(|wanted| {
181 labels
182 .iter()
183 .any(|label| label.eq_ignore_ascii_case(wanted))
184 })
185}
186
187pub fn plan(
199 repositories: &[Repository],
200 profiles: &BTreeMap<String, Profile>,
201 budget: Budget,
202 now: SystemTime,
203) -> Plan {
204 let mut decided = Plan::default();
205 let mut wanted = BTreeMap::new();
206 let mut warm_wanted = Vec::new();
207 let mut jobs = Vec::new();
208 for repository in repositories {
209 if let Some(sorted) = sort_out(repository, profiles, now, &mut decided) {
210 wanted.insert(repository.name.as_str(), sorted.serve.len());
211 jobs.extend(sorted.serve.into_iter().map(|job| (job, repository)));
212 warm_wanted.push((repository, sorted.warm));
213 }
214 }
215 jobs.sort_by_key(|(id, _)| *id);
216
217 let mut warm = Warm::new(&decided.warm, profiles, budget.held);
218 let mut taken = taken_slots(profiles);
219 let mut started = Started::new(&taken, budget.memory, budget.cpu);
220 let mut left = Left {
221 memory: budget.memory,
222 cpu: budget.cpu,
223 };
224 let mut first_for_cpu = None;
226 for (job, repository) in jobs {
227 let Some(missing) = wanted.get_mut(repository.name.as_str()) else {
228 continue;
229 };
230 if *missing == 0 {
231 continue;
232 }
233 let (Some(profile), Some(taken)) = (
234 profiles.get(&repository.profile),
235 taken.get_mut(repository.profile.as_str()),
236 ) else {
237 continue;
238 };
239 if let Some(older) = first_for_cpu {
240 decided
241 .waiting
242 .insert(job, Waiting::BehindCpu { job: older });
243 continue;
244 }
245 let memory = repository.runner_memory.unwrap_or(profile.memory);
246 let cpu = repository.runner_cpu.unwrap_or(profile.cpus);
247 let free_index = taken.iter().position(|taken| !taken);
248 let fits = free_index.is_some() && left.fits(memory, cpu);
249 let room = if fits || profile.preparing.is_some() {
250 None
251 } else {
252 warm.make_room(repository, free_index.is_none(), left.short(memory, cpu))
253 };
254 let index = match (room.as_ref(), free_index) {
255 (Some(room), free) => free.or(room.slot).unwrap_or_default(),
256 (None, Some(index)) if fits => index,
257 (None, free) => {
258 let (reason, short) = why_waits(
259 &repository.profile,
260 profile,
261 free.is_some(),
262 memory,
263 cpu,
264 left,
265 &started,
266 );
267 decided.waiting.insert(job, reason);
268 if short == Short::Cpu {
269 let held = Decision::HeldForCpu {
270 repository: repository.name.clone(),
271 };
272 if !decided.decisions.contains(&held) {
273 decided.decisions.push(held);
274 }
275 if repository.starved_of_cpu(now) {
277 first_for_cpu = Some(job);
278 }
279 }
280 continue;
281 }
282 };
283 let replacing = room.map_or_else(Vec::new, |room| {
285 left.give(room.memory, room.cpu);
286 for slot in &room.stopped {
287 decided.warm.remove(slot);
288 decided
289 .decisions
290 .push(Decision::Stop { slot: slot.clone() });
291 }
292 room.stopped
293 });
294 taken[index] = true;
295 *missing -= 1;
296 started.note(&repository.profile, job);
297 left.take(memory, cpu);
298 decided
299 .decisions
300 .push(start(repository, index, memory, cpu, replacing));
301 }
302 if !budget.held {
303 start_warm(&warm_wanted, profiles, &mut taken, left, &mut decided);
304 }
305 decided
306}
307
308#[derive(Debug, Clone, Copy, PartialEq, Eq)]
310enum Short {
311 Slot,
312 Memory,
313 Cpu,
314}
315
316fn why_waits(
319 name: &str,
320 profile: &Profile,
321 free_slot: bool,
322 memory: u64,
323 cpu: u64,
324 left: Left,
325 started: &Started<'_>,
326) -> (Waiting, Short) {
327 let profile_name = || name.to_owned();
328 if !free_slot {
329 let reason = match profile.preparing {
330 Some(Preparing::Reseeding) => Waiting::Reseeding {
331 profile: profile_name(),
332 },
333 Some(Preparing::PullingImage) => Waiting::PullingImage {
334 profile: profile_name(),
335 },
336 None => started.for_slot(
337 name,
338 Waiting::NoSlot {
339 profile: profile_name(),
340 },
341 ),
342 };
343 return (reason, Short::Slot);
344 }
345 match (left.memory, left.cpu) {
346 (Some(free), _) if free < memory => (
347 started.for_memory(
348 memory,
349 Waiting::Memory {
350 needs: memory,
351 free,
352 },
353 ),
354 Short::Memory,
355 ),
356 (_, Some(free)) => (
357 started.for_cpu(cpu, Waiting::Cpu { needs: cpu, free }),
358 Short::Cpu,
359 ),
360 (_, None) => (
362 Waiting::NoSlot {
363 profile: profile_name(),
364 },
365 Short::Slot,
366 ),
367 }
368}
369
370fn start(
373 repository: &Repository,
374 index: usize,
375 memory: u64,
376 cpu: u64,
377 replacing: Vec<SlotRef>,
378) -> Decision {
379 Decision::Start {
380 slot: SlotRef {
381 profile: repository.profile.clone(),
382 index,
383 },
384 repository: repository.name.clone(),
385 memory,
386 cpu,
387 warm: false,
388 replacing,
389 }
390}
391
392fn start_warm(
396 wanted: &[(&Repository, usize)],
397 profiles: &BTreeMap<String, Profile>,
398 taken: &mut BTreeMap<&str, Vec<bool>>,
399 mut left: Left,
400 decided: &mut Plan,
401) {
402 let room_wanted = decided.waiting.values().any(|reason| {
403 matches!(
404 reason,
405 Waiting::NoSlot { .. }
406 | Waiting::Memory { .. }
407 | Waiting::Cpu { .. }
408 | Waiting::BehindCpu { .. }
409 | Waiting::BehindStart { .. }
410 )
411 });
412 if room_wanted {
413 return;
414 }
415 for round in 0.. {
416 let mut any = false;
417 for (repository, wants) in wanted {
418 if *wants <= round {
419 continue;
420 }
421 let (Some(profile), Some(taken)) = (
422 profiles.get(&repository.profile),
423 taken.get_mut(repository.profile.as_str()),
424 ) else {
425 continue;
426 };
427 let (memory, cpu) = (profile.memory, profile.cpus);
428 let Some(index) = taken.iter().position(|taken| !taken) else {
429 continue;
430 };
431 if !left.fits(memory, cpu) {
432 continue;
433 }
434 taken[index] = true;
435 left.take(memory, cpu);
436 any = true;
437 decided.decisions.push(Decision::Start {
438 slot: SlotRef {
439 profile: repository.profile.clone(),
440 index,
441 },
442 repository: repository.name.clone(),
443 memory,
444 cpu,
445 warm: true,
446 replacing: Vec::new(),
447 });
448 }
449 if !any {
450 break;
451 }
452 }
453}
454
455#[derive(Debug, Clone, Copy)]
457struct Left {
458 memory: Option<u64>,
459 cpu: Option<u64>,
460}
461
462impl Left {
463 fn fits(self, memory: u64, cpu: u64) -> bool {
464 self.memory.is_none_or(|free| free >= memory) && self.cpu.is_none_or(|free| free >= cpu)
465 }
466
467 fn short(self, memory: u64, cpu: u64) -> Left {
469 Left {
470 memory: self.memory.map(|free| memory.saturating_sub(free)),
471 cpu: self.cpu.map(|free| cpu.saturating_sub(free)),
472 }
473 }
474
475 fn take(&mut self, memory: u64, cpu: u64) {
476 self.memory = self.memory.map(|free| free.saturating_sub(memory));
477 self.cpu = self.cpu.map(|free| free.saturating_sub(cpu));
478 }
479
480 fn give(&mut self, memory: u64, cpu: u64) {
481 self.memory = self.memory.map(|free| free + memory);
482 self.cpu = self.cpu.map(|free| free + cpu);
483 }
484}
485
486struct Sorted {
488 serve: Vec<u64>,
489 warm: usize,
490}
491
492fn sort_out(
496 repository: &Repository,
497 profiles: &BTreeMap<String, Profile>,
498 now: SystemTime,
499 decided: &mut Plan,
500) -> Option<Sorted> {
501 let mut matching: Vec<&QueuedJob> = repository
502 .queued
503 .iter()
504 .filter(|job| matches(job, &repository.labels))
505 .collect();
506 let profile = profiles.get(&repository.profile);
507 let mut warm = repository.warm;
508 if repository.paused {
509 decided
510 .waiting
511 .extend(matching.iter().map(|job| (job.id, Waiting::Paused)));
512 matching.clear();
514 warm = 0;
515 }
516 let upkeep_due = profile.is_some_and(|profile| profile.upkeep_due);
518 if upkeep_due {
519 warm = 0;
520 }
521 let idle: Vec<(SlotRef, &RunnerSlot)> = profile
522 .into_iter()
523 .flat_map(|profile| profile.slots.iter().enumerate())
524 .filter_map(|(index, slot)| match slot {
525 Slot::Runner(runner) if runner.repository == repository.name && !runner.busy => Some((
526 SlotRef {
527 profile: repository.profile.clone(),
528 index,
529 },
530 runner,
531 )),
532 _ => None,
533 })
534 .collect();
535 let surplus = idle.len().saturating_sub(matching.len());
536 let stops: Vec<SlotRef> = idle
537 .iter()
538 .filter(|(_, runner)| {
539 upkeep_due
540 || now
541 .duration_since(runner.started_at)
542 .is_ok_and(|waited| waited >= IDLE_TIMEOUT)
543 })
544 .take(surplus.saturating_sub(warm))
545 .map(|(slot, _)| slot.clone())
546 .collect();
547 let kept: Vec<&SlotRef> = idle
549 .iter()
550 .map(|(slot, _)| slot)
551 .filter(|slot| !stops.contains(slot))
552 .collect();
553 decided
554 .warm
555 .extend(kept.iter().skip(matching.len()).map(|slot| (*slot).clone()));
556 let warm = warm.saturating_sub(kept.len().saturating_sub(matching.len()));
557 decided
558 .decisions
559 .extend(stops.into_iter().map(|slot| Decision::Stop { slot }));
560 let missing = matching.len().saturating_sub(idle.len());
561 let skip = matching.len() - missing;
563 decided.waiting.extend(
564 matching
565 .iter()
566 .take(skip)
567 .map(|job| (job.id, Waiting::RunnerStarting)),
568 );
569 let serve = matching.iter().skip(skip).map(|job| job.id);
570 match repository.held_until.filter(|until| *until > now) {
571 Some(until) => {
572 let until = crate::time::ms(until) / 1000;
573 decided
574 .waiting
575 .extend(serve.map(|job| (job, Waiting::Backoff { until })));
576 None
577 }
578 None => Some(Sorted {
579 serve: serve.collect(),
580 warm,
581 }),
582 }
583}
584
585struct Warm<'plan> {
587 runners: Vec<Option<(SlotRef, &'plan RunnerSlot)>>,
589}
590
591struct Room {
593 stopped: Vec<SlotRef>,
595 slot: Option<usize>,
597 memory: u64,
598 cpu: u64,
599}
600
601impl<'plan> Warm<'plan> {
602 fn new(
604 warm: &BTreeSet<SlotRef>,
605 profiles: &'plan BTreeMap<String, Profile>,
606 held: bool,
607 ) -> Warm<'plan> {
608 let runners = warm
609 .iter()
610 .filter(|_| !held)
611 .filter_map(
612 |slot| match profiles.get(&slot.profile)?.slots.get(slot.index)? {
613 Slot::Runner(runner) => Some(Some((slot.clone(), runner))),
614 Slot::Free | Slot::Closed => None,
615 },
616 )
617 .collect();
618 Warm { runners }
619 }
620
621 fn make_room(&mut self, repository: &Repository, slot: bool, short: Left) -> Option<Room> {
625 let (memory, cpu) = (short.memory.unwrap_or(0), short.cpu.unwrap_or(0));
626 let mut candidates: Vec<usize> = self
627 .runners
628 .iter()
629 .enumerate()
630 .filter_map(|(at, runner)| {
631 runner
632 .as_ref()
633 .filter(|(_, runner)| runner.repository != repository.name)
634 .map(|_| at)
635 })
636 .collect();
637 candidates.sort_by_key(|at| {
639 self.runners[*at]
640 .as_ref()
641 .is_some_and(|(slot, _)| slot.profile != repository.profile)
642 });
643 let mut room = Room {
644 stopped: Vec::new(),
645 slot: None,
646 memory: 0,
647 cpu: 0,
648 };
649 let mut picked = Vec::new();
650 for at in candidates {
651 let short = (slot && room.slot.is_none()) || room.memory < memory || room.cpu < cpu;
652 if !short {
653 break;
654 }
655 let Some((ref_, runner)) = &self.runners[at] else {
656 continue;
657 };
658 let gives_slot = ref_.profile == repository.profile;
659 let helps = (gives_slot && slot && room.slot.is_none())
661 || (room.memory < memory && runner.memory > 0)
662 || (room.cpu < cpu && runner.cpu > 0);
663 if !helps {
664 continue;
665 }
666 if gives_slot && room.slot.is_none() {
667 room.slot = Some(ref_.index);
668 }
669 room.memory += runner.memory;
670 room.cpu += runner.cpu;
671 room.stopped.push(ref_.clone());
672 picked.push(at);
673 }
674 if (slot && room.slot.is_none()) || room.memory < memory || room.cpu < cpu {
675 return None;
676 }
677 for at in picked {
678 self.runners[at] = None;
679 }
680 Some(room)
681 }
682}
683
684struct Started<'plan> {
687 free_slots: BTreeSet<&'plan str>,
689 memory: Option<u64>,
691 cpu: Option<u64>,
692 by_profile: BTreeMap<&'plan str, u64>,
694 first: Option<u64>,
695}
696
697impl<'plan> Started<'plan> {
698 fn new(
699 taken: &BTreeMap<&'plan str, Vec<bool>>,
700 memory: Option<u64>,
701 cpu: Option<u64>,
702 ) -> Started<'plan> {
703 Started {
704 free_slots: taken
705 .iter()
706 .filter(|(_, slots)| slots.contains(&false))
707 .map(|(name, _)| *name)
708 .collect(),
709 memory,
710 cpu,
711 by_profile: BTreeMap::new(),
712 first: None,
713 }
714 }
715
716 fn note(&mut self, profile: &'plan str, job: u64) {
718 self.by_profile.entry(profile).or_insert(job);
719 self.first.get_or_insert(job);
720 }
721
722 fn for_slot(&self, profile: &str, reason: Waiting) -> Waiting {
724 match self.by_profile.get(profile) {
725 Some(older) if self.free_slots.contains(profile) => {
726 Waiting::BehindStart { job: *older }
727 }
728 _ => reason,
729 }
730 }
731
732 fn for_memory(&self, needs: u64, reason: Waiting) -> Waiting {
734 Self::behind(self.first, self.memory, needs).unwrap_or(reason)
735 }
736
737 fn for_cpu(&self, needs: u64, reason: Waiting) -> Waiting {
739 Self::behind(self.first, self.cpu, needs).unwrap_or(reason)
740 }
741
742 fn behind(first: Option<u64>, free: Option<u64>, needs: u64) -> Option<Waiting> {
743 first
744 .filter(|_| free.is_some_and(|free| free >= needs))
745 .map(|job| Waiting::BehindStart { job })
746 }
747}
748
749fn taken_slots(profiles: &BTreeMap<String, Profile>) -> BTreeMap<&str, Vec<bool>> {
751 profiles
752 .iter()
753 .map(|(name, profile)| {
754 let taken = profile
755 .slots
756 .iter()
757 .map(|slot| profile.preparing.is_some() || *slot != Slot::Free)
758 .collect();
759 (name.as_str(), taken)
760 })
761 .collect()
762}
763
764#[cfg(test)]
765mod tests;