rucc_driver/schedule.rs
1//! Running the jobs in a [`Plan`](crate::Plan) across threads, deterministically.
2//!
3//! Design: `spec/03-architecture.md` section 3.5, and section 3.7 for the determinism rule.
4//!
5//! `rucc a.c b.c c.c` compiles all three in one process on a shared
6//! [`Session`](rucc_session::Session), rather than the build system forking three processes
7//! that each re-read every header. That is the larger of the two levels of parallelism and it
8//! is the reason `Session` is thread-safe rather than merely convenient.
9//!
10//! The rule that makes this safe to have at all: **each unit of parallel work writes only to
11//! its own slot, and results are merged in index order, never in completion order.** Byte
12//! identical output is a requirement in `spec/02-the-goal.md`, and a scheduler that merges by
13//! whoever finishes first quietly gives it up. The API here makes that hard to get wrong,
14//! because the only thing a caller can do with a result is receive the whole vector back in
15//! input order.
16//!
17//! # Status
18//!
19//! The scheduler is real. The work it schedules is not, until M3. `spec/18-package-layout.md`
20//! section 18.3 has `rayon` down for this, and it will be needed for the per-function level
21//! inside a translation unit; for the per-file level a scoped thread per job is the whole
22//! implementation and it costs no dependency, so that is what this is.
23
24use std::num::NonZeroUsize;
25use std::sync::Mutex;
26use std::sync::atomic::{AtomicUsize, Ordering};
27
28/// How many jobs to run at once.
29#[derive(Debug, Clone, Copy, PartialEq, Eq)]
30pub enum Jobs {
31 /// One job at a time, in order. What `-j1` asks for, and what the determinism check in
32 /// CI compares against.
33 Serial,
34 /// At most this many at once.
35 Threads(NonZeroUsize),
36}
37
38impl Default for Jobs {
39 fn default() -> Jobs {
40 Jobs::available()
41 }
42}
43
44impl Jobs {
45 /// What the machine can do, or serial when it will not say.
46 #[must_use]
47 pub fn available() -> Jobs {
48 match std::thread::available_parallelism() {
49 Ok(n) if n.get() > 1 => Jobs::Threads(n),
50 _ => Jobs::Serial,
51 }
52 }
53
54 /// The number of workers this asks for.
55 #[must_use]
56 pub fn count(self) -> usize {
57 match self {
58 Jobs::Serial => 1,
59 Jobs::Threads(n) => n.get(),
60 }
61 }
62
63 /// Parses the argument of `-j`.
64 ///
65 /// A bare `-j` with no number means "as many as the machine has", which is what `make`
66 /// does and therefore what people expect.
67 ///
68 /// # Errors
69 ///
70 /// Returns the offending text when it is not a positive number.
71 pub fn parse(arg: &str) -> Result<Jobs, String> {
72 if arg.is_empty() {
73 return Ok(Jobs::available());
74 }
75 match arg.parse::<NonZeroUsize>() {
76 // One worker is the serial path, not a pool of one, so that `-j1` is exactly what
77 // the determinism check in CI compares against rather than merely equivalent.
78 Ok(n) if n.get() == 1 => Ok(Jobs::Serial),
79 Ok(n) => Ok(Jobs::Threads(n)),
80 Err(_) if arg == "0" => Err("-j0 asks for no workers at all".to_owned()),
81 Err(_) => Err(format!("`{arg}` is not a job count")),
82 }
83 }
84}
85
86/// Runs `work` over every item, and returns the results in input order.
87///
88/// The closure runs on several threads at once and may finish in any order. The vector that
89/// comes back does not depend on that order, on the thread count, or on timing, which is the
90/// property `spec/03-architecture.md` section 3.7 requires and the reason this function
91/// exists rather than each caller reaching for threads directly.
92///
93/// A panic in the closure propagates once every other job has finished, rather than leaving
94/// the compilation half done with no diagnostic.
95///
96/// # Panics
97///
98/// Panics if `work` panicked on any item.
99pub fn run<T, R, F>(jobs: Jobs, items: &[T], work: F) -> Vec<R>
100where
101 T: Sync,
102 R: Send,
103 F: Fn(usize, &T) -> R + Sync,
104{
105 if items.is_empty() {
106 return Vec::new();
107 }
108 let workers = jobs.count().min(items.len());
109 if workers <= 1 {
110 return items.iter().enumerate().map(|(i, t)| work(i, t)).collect();
111 }
112
113 // One slot per item, so no two threads ever write to the same place and the merge is
114 // just "take the slots in order". A channel would have been shorter and would have made
115 // the output depend on completion order.
116 let slots: Vec<Mutex<Option<R>>> = items.iter().map(|_| Mutex::new(None)).collect();
117 let next = AtomicUsize::new(0);
118
119 std::thread::scope(|scope| {
120 for _ in 0..workers {
121 scope.spawn(|| {
122 loop {
123 // Claiming the next index rather than splitting the work up front, because
124 // translation units differ in size by more than an order of magnitude and
125 // a static split leaves most of the machine idle behind the biggest file.
126 let i = next.fetch_add(1, Ordering::Relaxed);
127 let Some(item) = items.get(i) else { break };
128 let result = work(i, item);
129 *slots[i].lock().expect("a slot lock is only held to store one result") =
130 Some(result);
131 }
132 });
133 }
134 });
135
136 slots
137 .into_iter()
138 .map(|slot| {
139 slot.into_inner()
140 .expect("a slot lock is only held to store one result")
141 .expect("every index was claimed exactly once")
142 })
143 .collect()
144}
145
146#[cfg(test)]
147mod tests {
148 use std::sync::atomic::AtomicUsize;
149
150 use super::*;
151
152 #[test]
153 fn results_come_back_in_input_order_however_they_finish() {
154 // The first job is the slowest, so completion order is the reverse of input order.
155 // If the merge depended on completion order this test would fail, and so would the
156 // determinism check in CI, but much later and much less clearly.
157 let items: Vec<u64> = (0..8).collect();
158 let out = run(Jobs::Threads(NonZeroUsize::new(8).unwrap()), &items, |i, x| {
159 std::thread::sleep(std::time::Duration::from_millis((8 - i as u64) * 4));
160 x * 10
161 });
162 assert_eq!(out, vec![0, 10, 20, 30, 40, 50, 60, 70]);
163 }
164
165 #[test]
166 fn serial_and_parallel_give_the_same_answer() {
167 let items: Vec<usize> = (0..64).collect();
168 let serial = run(Jobs::Serial, &items, |i, x| i + x);
169 let parallel = run(Jobs::Threads(NonZeroUsize::new(4).unwrap()), &items, |i, x| i + x);
170 assert_eq!(serial, parallel);
171 }
172
173 #[test]
174 fn every_item_runs_exactly_once() {
175 let items: Vec<usize> = (0..500).collect();
176 let calls = AtomicUsize::new(0);
177 let out = run(Jobs::Threads(NonZeroUsize::new(16).unwrap()), &items, |_, x| {
178 calls.fetch_add(1, Ordering::Relaxed);
179 *x
180 });
181 assert_eq!(calls.load(Ordering::Relaxed), 500);
182 assert_eq!(out, items);
183 }
184
185 #[test]
186 fn more_workers_than_items_is_fine() {
187 let items = [1, 2];
188 let out = run(Jobs::Threads(NonZeroUsize::new(64).unwrap()), &items, |_, x| *x);
189 assert_eq!(out, vec![1, 2]);
190 }
191
192 #[test]
193 fn no_items_is_no_threads_and_no_results() {
194 let items: [u8; 0] = [];
195 let out: Vec<u8> = run(Jobs::available(), &items, |_, x| *x);
196 assert!(out.is_empty());
197 }
198
199 #[test]
200 fn dash_j_reads_the_way_make_reads_it() {
201 assert_eq!(Jobs::parse("1").unwrap(), Jobs::Serial);
202 assert_eq!(Jobs::parse("4").unwrap(), Jobs::Threads(NonZeroUsize::new(4).unwrap()));
203 assert_eq!(Jobs::parse("").unwrap(), Jobs::available());
204 assert!(Jobs::parse("0").is_err());
205 assert!(Jobs::parse("many").is_err());
206 }
207
208 #[test]
209 fn a_panicking_job_is_not_swallowed() {
210 // A compiler that loses an internal error and exits zero is worse than one that
211 // crashes, because the build carries on with a missing object.
212 let items = [0, 1, 2];
213 let r = std::panic::catch_unwind(|| {
214 run(Jobs::Threads(NonZeroUsize::new(3).unwrap()), &items, |_, x| {
215 assert!(*x != 1, "planted failure");
216 *x
217 })
218 });
219 assert!(r.is_err());
220 }
221}