runnerd_core/scheduler.rs
1//! Which runners to start and stop, as a pure function of queued jobs, slots and the memory and CPU budgets (`plan`), and
2//! the loop that applies those decisions with GitHub and Docker (`engine`). `plan` does no I/O and reads no clock.
3
4use std::collections::BTreeMap;
5use std::time::{Duration, SystemTime};
6
7use crate::github::QueuedJob;
8
9pub mod engine;
10
11/// How long a started runner may wait for a job before it counts as surplus: covers the container's start and
12/// GitHub's assignment.
13pub const IDLE_TIMEOUT: Duration = Duration::from_secs(300);
14
15/// Once a job has waited this long for CPU, no younger job takes CPU until it fits: smaller jobs arriving all the
16/// time would otherwise take every CPU that frees up before it.
17pub const CPU_PRIORITY_AFTER: Duration = Duration::from_secs(120);
18
19/// A repository as the planner sees it.
20#[derive(Debug, Clone)]
21pub struct Repository {
22 /// `owner/name` on GitHub.
23 pub name: String,
24 /// The profile its jobs run in.
25 pub profile: String,
26 /// The labels its runners register with.
27 pub labels: Vec<String>,
28 /// Paused repositories get no new runners; their running jobs finish.
29 pub paused: bool,
30 /// No runner starts for this repository before then, after runners failed to take a job.
31 pub held_until: Option<SystemTime>,
32 /// Its queued jobs at the last poll.
33 pub queued: Vec<QueuedJob>,
34 /// What a new runner of it takes of the budget: `None`, its profile's `memory`. Under measured admission, the
35 /// largest predicted peak of its queued jobs, since GitHub may give the runner any of them.
36 pub runner_memory: Option<u64>,
37 /// What a new runner of it takes of the CPU budget, in millicores: `None`, its profile's `cpus`. Under measured
38 /// admission, the largest predicted average CPU of its queued jobs.
39 pub runner_cpu: Option<u64>,
40 /// Since when its next job waits for CPU that a slot and memory have room for.
41 pub cpu_waiting_since: Option<SystemTime>,
42}
43
44impl Repository {
45 /// Whether its next job has waited `CPU_PRIORITY_AFTER` or more for CPU at `now`.
46 fn starved_of_cpu(&self, now: SystemTime) -> bool {
47 self.cpu_waiting_since
48 .and_then(|since| now.duration_since(since).ok())
49 .is_some_and(|waited| waited >= CPU_PRIORITY_AFTER)
50 }
51}
52
53/// A profile's slots: one concurrent job each.
54#[derive(Debug, Clone)]
55pub struct Profile {
56 /// Memory a slot's container may use.
57 pub memory: u64,
58 /// CPU a slot's container may use, in millicores.
59 pub cpus: u64,
60 /// While the profile's snapshot is rebuilt or its image pulled, none of its slots starts.
61 pub preparing: bool,
62 /// Index 0 is slot 1.
63 pub slots: Vec<Slot>,
64}
65
66/// What a slot holds.
67#[derive(Debug, Clone, PartialEq, Eq)]
68pub enum Slot {
69 /// Nothing: a runner may start here.
70 Free,
71 /// A runner, waiting for its job or running it.
72 Runner(RunnerSlot),
73 /// Free, but over the profile's number of slots, which a reload lowered while its runners ran: nothing starts
74 /// here.
75 Closed,
76}
77
78/// A slot whose container runs a runner registered for one job.
79#[derive(Debug, Clone, PartialEq, Eq)]
80pub struct RunnerSlot {
81 /// `owner/name` the runner is registered on.
82 pub repository: String,
83 /// Whether GitHub reports the runner running a job.
84 pub busy: bool,
85 /// When its container started.
86 pub started_at: SystemTime,
87}
88
89/// Where slots are, by profile name and slot index (0 is slot 1).
90#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
91pub struct SlotRef {
92 /// The profile's name.
93 pub profile: String,
94 /// 0 is slot 1.
95 pub index: usize,
96}
97
98/// What `plan` decides for one slot.
99#[derive(Debug, Clone, PartialEq, Eq)]
100pub enum Decision {
101 /// Register a runner for the repository and start it in the slot.
102 Start {
103 /// The free slot.
104 slot: SlotRef,
105 /// `owner/name` the runner registers on.
106 repository: String,
107 /// What it takes of the budget.
108 memory: u64,
109 /// What it takes of the CPU budget, in millicores.
110 cpu: u64,
111 },
112 /// Hold the repository's next job: a slot and memory had room for its runner, the CPU budget did not.
113 HeldForCpu {
114 /// `owner/name`.
115 repository: String,
116 },
117 /// Remove the slot's runner, which waited `IDLE_TIMEOUT` for a job no longer queued.
118 Stop {
119 /// The slot whose runner stops.
120 slot: SlotRef,
121 },
122}
123
124/// Whether a job may run on a repository's runners: every label it asks for is one of theirs. GitHub compares
125/// labels without regard to case.
126pub fn matches(job: &QueuedJob, labels: &[String]) -> bool {
127 job.labels.iter().all(|wanted| {
128 labels
129 .iter()
130 .any(|label| label.eq_ignore_ascii_case(wanted))
131 })
132}
133
134/// Decides which slots start a runner and which idle runners stop.
135///
136/// Each repository gets a runner per matching queued job, less its runners still waiting for a job. Jobs are
137/// served oldest first across repositories, each in the lowest free slot of its repository's profile, while what
138/// the runner takes (`Repository::runner_memory`, by default the slot's memory) fits `free_memory` and its CPU
139/// (`Repository::runner_cpu`, by default the slot's `cpus`) fits `free_cpu` (`None`: no budget). A repository whose
140/// job had a slot and memory but not CPU is held once (`Decision::HeldForCpu`). An idle runner stops once no queued
141/// job is left for it and it waited `IDLE_TIMEOUT`. Once a held repository waited `CPU_PRIORITY_AFTER`, no younger
142/// job starts until its own fits.
143pub fn plan(
144 repositories: &[Repository],
145 profiles: &BTreeMap<String, Profile>,
146 free_memory: Option<u64>,
147 free_cpu: Option<u64>,
148 now: SystemTime,
149) -> Vec<Decision> {
150 let mut decisions = Vec::new();
151 let mut wanted = BTreeMap::new();
152 let mut jobs = Vec::new();
153 for repository in repositories {
154 let held = repository.held_until.is_some_and(|until| until > now);
155 let matching: Vec<&QueuedJob> = if repository.paused {
156 Vec::new()
157 } else {
158 repository
159 .queued
160 .iter()
161 .filter(|job| matches(job, &repository.labels))
162 .collect()
163 };
164 let idle: Vec<(SlotRef, &RunnerSlot)> = profiles
165 .get(&repository.profile)
166 .into_iter()
167 .flat_map(|profile| profile.slots.iter().enumerate())
168 .filter_map(|(index, slot)| match slot {
169 Slot::Runner(runner) if runner.repository == repository.name && !runner.busy => {
170 Some((
171 SlotRef {
172 profile: repository.profile.clone(),
173 index,
174 },
175 runner,
176 ))
177 }
178 _ => None,
179 })
180 .collect();
181 let surplus = idle.len().saturating_sub(matching.len());
182 decisions.extend(
183 idle.iter()
184 .filter(|(_, runner)| {
185 now.duration_since(runner.started_at)
186 .is_ok_and(|waited| waited >= IDLE_TIMEOUT)
187 })
188 .take(surplus)
189 .map(|(slot, _)| Decision::Stop { slot: slot.clone() }),
190 );
191 if !held {
192 let missing = matching.len().saturating_sub(idle.len());
193 wanted.insert(repository.name.as_str(), missing);
194 // The oldest jobs are the ones still waiting once the idle runners take theirs.
195 let skip = matching.len() - missing;
196 jobs.extend(matching.iter().skip(skip).map(|job| (job.id, repository)));
197 }
198 }
199 jobs.sort_by_key(|(id, _)| *id);
200
201 let mut taken = taken_slots(profiles);
202 let mut free_memory = free_memory;
203 let mut free_cpu = free_cpu;
204 for (_, repository) in jobs {
205 let Some(missing) = wanted.get_mut(repository.name.as_str()) else {
206 continue;
207 };
208 if *missing == 0 {
209 continue;
210 }
211 let (Some(profile), Some(taken)) = (
212 profiles.get(&repository.profile),
213 taken.get_mut(repository.profile.as_str()),
214 ) else {
215 continue;
216 };
217 let memory = repository.runner_memory.unwrap_or(profile.memory);
218 if free_memory.is_some_and(|free| free < memory) {
219 continue;
220 }
221 let Some(index) = taken.iter().position(|taken| !taken) else {
222 continue;
223 };
224 let cpu = repository.runner_cpu.unwrap_or(profile.cpus);
225 if free_cpu.is_some_and(|free| free < cpu) {
226 let held = Decision::HeldForCpu {
227 repository: repository.name.clone(),
228 };
229 if !decisions.contains(&held) {
230 decisions.push(held);
231 }
232 // It waited long enough: no younger job takes CPU until it fits.
233 if repository.starved_of_cpu(now) {
234 break;
235 }
236 continue;
237 }
238 taken[index] = true;
239 *missing -= 1;
240 free_memory = free_memory.map(|free| free - memory);
241 free_cpu = free_cpu.map(|free| free - cpu);
242 decisions.push(Decision::Start {
243 slot: SlotRef {
244 profile: repository.profile.clone(),
245 index,
246 },
247 repository: repository.name.clone(),
248 memory,
249 cpu,
250 });
251 }
252 decisions
253}
254
255/// Whether each slot of each profile is taken, by profile name: by a runner, closed, or its profile preparing.
256fn taken_slots(profiles: &BTreeMap<String, Profile>) -> BTreeMap<&str, Vec<bool>> {
257 profiles
258 .iter()
259 .map(|(name, profile)| {
260 let taken = profile
261 .slots
262 .iter()
263 .map(|slot| profile.preparing || *slot != Slot::Free)
264 .collect();
265 (name.as_str(), taken)
266 })
267 .collect()
268}
269
270#[cfg(test)]
271mod tests;