char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somdeslop: split the task lib into modules7e18c89

main
branchesmainall refs →
15.1 KiB463 linesraw
1use std::collections::{HashMap, HashSet};
2use std::path::PathBuf;
3use std::sync::Arc;
4use std::sync::atomic::{AtomicBool, Ordering};
5
6use serde::{Deserialize, Serialize};
7
8use crate::cache::Cache;
9use crate::env::EnvLoader;
10use crate::fingerprint::Fingerprinter;
11use crate::graph::{Action, Graph, shell_quote};
12use crate::{EnvMap, Error, Reporter, Result};
13
14#[derive(Clone, Default)]
15pub struct CancelToken(Arc<AtomicBool>);
16
17impl CancelToken {
18    pub fn new() -> Self {
19        CancelToken(Arc::new(AtomicBool::new(false)))
20    }
21    pub fn cancel(&self) {
22        self.0.store(true, Ordering::SeqCst);
23    }
24    pub fn is_cancelled(&self) -> bool {
25        self.0.load(Ordering::SeqCst)
26    }
27}
28
29pub struct RunOptions {
30    pub root: PathBuf,
31    pub cache: Option<Cache>,
32    pub force: bool,
33    pub extra_args: Vec<String>,
34    pub env: EnvMap,
35    pub reporter: Box<dyn Reporter>,
36    pub cancelled: CancelToken,
37}
38
39impl RunOptions {
40    pub fn new(root: PathBuf) -> Self {
41        RunOptions {
42            root,
43            cache: None,
44            force: false,
45            extra_args: Vec::new(),
46            env: EnvMap::new(),
47            reporter: Box::new(crate::NoopReporter),
48            cancelled: CancelToken::new(),
49        }
50    }
51}
52
53#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
54pub struct ReadyTask {
55    pub name: String,
56    pub action: Action,
57    /// Extra arguments; non-empty only for the run's target.
58    pub args: Vec<String>,
59    pub env: EnvMap,
60}
61
62impl ReadyTask {
63    pub fn display(&self) -> String {
64        let mut text = self.action.display();
65        for arg in &self.args {
66            text.push(' ');
67            text.push_str(&shell_quote(arg));
68        }
69        text
70    }
71}
72
73#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
74pub struct FailedTask {
75    pub name: String,
76    pub code: u8,
77}
78
79#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
80pub struct Report {
81    pub code: u8,
82    pub completed: Vec<String>,
83    pub failed: Vec<FailedTask>,
84    pub cached: Vec<String>,
85}
86
87impl Report {
88    pub fn ok(&self) -> bool {
89        self.code == 0
90    }
91}
92
93#[derive(Debug, Serialize, Deserialize)]
94pub enum Step {
95    Ready(Vec<ReadyTask>),
96    Done(Report),
97}
98
99struct Frame {
100    name: String,
101    claim: usize,
102    parent: Option<usize>,
103    dep_index: usize,
104    env: EnvMap,
105    ancestors: Vec<String>,
106    dep_outcomes: Vec<Option<u8>>,
107    active_children: HashSet<usize>,
108    deps_started: bool,
109    waiting_claim: Option<usize>,
110    dispatched: bool,
111    surfaced: bool,
112    done: bool,
113}
114
115struct Claim {
116    owner: Option<usize>,
117    outcome: Option<u8>,
118    fingerprint: Option<String>,
119    /// Whether the task's own action was dispatched (vs. completing
120    /// because a dependency failed). Only dispatched tasks are failures.
121    dispatched: bool,
122}
123
124pub struct Execution {
125    graph: Graph,
126    root: PathBuf,
127    cache: Option<Cache>,
128    force: bool,
129    extra_args: Vec<String>,
130    base_env: EnvMap,
131    reporter: Box<dyn Reporter>,
132    cancelled: CancelToken,
133    env: EnvLoader,
134    fingerprinter: Fingerprinter,
135    frames: Vec<Frame>,
136    claims: Vec<Claim>,
137    claim_by_name: HashMap<String, usize>,
138    root_frame: usize,
139    completed: Vec<String>,
140    cached: Vec<String>,
141}
142
143impl Execution {
144    pub fn new(graph: Graph, target: &str, opts: RunOptions) -> Result<Self> {
145        graph.get(target)?;
146        let root = opts.root.clone();
147        let mut execution = Execution {
148            fingerprinter: Fingerprinter::new(graph.clone(), root.clone()),
149            env: EnvLoader::new(root.clone()),
150            graph,
151            root,
152            cache: opts.cache,
153            force: opts.force,
154            extra_args: opts.extra_args,
155            base_env: opts.env,
156            reporter: opts.reporter,
157            cancelled: opts.cancelled,
158            frames: Vec::new(),
159            claims: Vec::new(),
160            claim_by_name: HashMap::new(),
161            root_frame: 0,
162            completed: Vec::new(),
163            cached: Vec::new(),
164        };
165        let base = execution.base_env.clone();
166        let root_frame = execution.create_frame(None, 0, target, &base, Vec::new())?;
167        execution.root_frame = root_frame;
168        Ok(execution)
169    }
170
171    fn create_frame(
172        &mut self,
173        parent: Option<usize>,
174        dep_index: usize,
175        name: &str,
176        inherited_env: &EnvMap,
177        ancestors: Vec<String>,
178    ) -> Result<usize> {
179        if let Some(start) = ancestors.iter().position(|a| a == name) {
180            let mut cycle = ancestors[start..].to_vec();
181            cycle.push(name.to_owned());
182            return Err(Error::Cycle(cycle.join(" -> ")));
183        }
184        let task = self.graph.get(name)?.clone();
185        let env = self.env.effective(&task, inherited_env)?;
186        let id = self.frames.len();
187        let (claim, waiting_claim) = match self.claim_by_name.get(name) {
188            Some(&claim) => (claim, Some(claim)),
189            None => {
190                let claim = self.claims.len();
191                self.claims.push(Claim {
192                    owner: Some(id),
193                    outcome: None,
194                    fingerprint: None,
195                    dispatched: false,
196                });
197                self.claim_by_name.insert(name.to_owned(), claim);
198                (claim, None)
199            }
200        };
201        self.frames.push(Frame {
202            name: name.to_owned(),
203            claim,
204            parent,
205            dep_index,
206            env,
207            ancestors,
208            dep_outcomes: vec![None; task.dependencies.len()],
209            active_children: HashSet::new(),
210            deps_started: false,
211            waiting_claim,
212            dispatched: false,
213            surfaced: false,
214            done: false,
215        });
216        Ok(id)
217    }
218
219    fn complete(&mut self, frame_id: usize, code: u8) {
220        let (claim, parent, dep_index) = {
221            let frame = &mut self.frames[frame_id];
222            if frame.done {
223                return;
224            }
225            frame.done = true;
226            (frame.claim, frame.parent, frame.dep_index)
227        };
228        if self.claims[claim].owner == Some(frame_id) && self.claims[claim].outcome.is_none() {
229            self.claims[claim].outcome = Some(code);
230            let name = self.frames[frame_id].name.clone();
231            self.completed.push(name);
232        }
233        if let Some(parent) = parent {
234            self.frames[parent].dep_outcomes[dep_index] = Some(code);
235            self.frames[parent].active_children.remove(&frame_id);
236        }
237    }
238
239    /// Advances every frame whose dependencies have resolved, marking
240    /// cached tasks complete and dispatchable tasks ready. Returns whether
241    /// anything changed.
242    fn advance(&mut self) -> Result<bool> {
243        let mut progressed = false;
244        let mut index = 0;
245        while index < self.frames.len() {
246            let frame = &self.frames[index];
247            if frame.done || frame.dispatched {
248                index += 1;
249                continue;
250            }
251            // A frame borrowing another frame's in-flight execution
252            // completes with that execution's outcome.
253            if let Some(claim) = frame.waiting_claim {
254                if let Some(code) = self.claims[claim].outcome {
255                    self.complete(index, code);
256                    progressed = true;
257                }
258                index += 1;
259                continue;
260            }
261            let task = self.graph.get(&frame.name)?.clone();
262
263            // Start dependencies.
264            if !frame.deps_started || !task.parallel && frame.active_children.is_empty() {
265                let next = if task.parallel {
266                    if frame.deps_started {
267                        None
268                    } else {
269                        Some(0..task.dependencies.len())
270                    }
271                } else {
272                    // Sequential: a failed dependency ends the task now.
273                    if let Some(code) = frame.dep_outcomes.iter().flatten().find(|code| **code != 0)
274                    {
275                        let code = *code;
276                        self.complete(index, code);
277                        progressed = true;
278                        index += 1;
279                        continue;
280                    }
281                    let started = frame.dep_outcomes.iter().filter(|o| o.is_some()).count()
282                        + frame.active_children.len();
283                    if started < task.dependencies.len() {
284                        Some(started..started + 1)
285                    } else {
286                        None
287                    }
288                };
289                if let Some(range) = next {
290                    self.frames[index].deps_started = true;
291                    let env = self.frames[index].env.clone();
292                    let mut child_ancestors = self.frames[index].ancestors.clone();
293                    child_ancestors.push(self.frames[index].name.clone());
294                    for dep_index in range {
295                        let dependency = task.dependencies[dep_index].clone();
296                        let child = self.create_frame(
297                            Some(index),
298                            dep_index,
299                            &dependency,
300                            &env,
301                            child_ancestors.clone(),
302                        )?;
303                        self.frames[index].active_children.insert(child);
304                    }
305                    progressed = true;
306                    index += 1;
307                    continue;
308                }
309            }
310            if !self.frames[index].active_children.is_empty() {
311                index += 1;
312                continue;
313            }
314            if task.parallel
315                && let Some(_failed) = frame.dep_outcomes.iter().flatten().find(|c| **c != 0)
316            {
317                self.complete(index, 1);
318                progressed = true;
319                index += 1;
320                continue;
321            }
322            if !task.parallel
323                && let Some(code) = frame.dep_outcomes.iter().flatten().find(|c| **c != 0)
324            {
325                let code = *code;
326                self.complete(index, code);
327                progressed = true;
328                index += 1;
329                continue;
330            }
331
332            // All dependencies succeeded: resolve this task's own action.
333            match task.action {
334                Action::None => {
335                    self.complete(index, 0);
336                    progressed = true;
337                }
338                _ => {
339                    if task.files.is_some() {
340                        let args: Vec<String> = if index == self.root_frame {
341                            self.extra_args.clone()
342                        } else {
343                            Vec::new()
344                        };
345                        let env = self.frames[index].env.clone();
346                        let name = self.frames[index].name.clone();
347                        let fingerprint = self.fingerprinter.fingerprint(&name, &args, &env)?;
348                        self.claims[self.frames[index].claim].fingerprint =
349                            Some(fingerprint.clone());
350                        let hit = !self.force
351                            && self.cache.as_ref().is_some_and(|cache| {
352                                cache
353                                    .manifest_matches(&self.root, &task, &fingerprint)
354                                    .unwrap_or(false)
355                            });
356                        if hit {
357                            self.reporter.cached(&name);
358                            self.cached.push(name);
359                            self.complete(index, 0);
360                            progressed = true;
361                            index += 1;
362                            continue;
363                        }
364                    }
365                    self.frames[index].dispatched = true;
366                    let claim = self.frames[index].claim;
367                    self.claims[claim].dispatched = true;
368                    progressed = true;
369                }
370            }
371            index += 1;
372        }
373        Ok(progressed)
374    }
375
376    pub fn step(&mut self) -> Result<Step> {
377        // Advance until nothing else resolves without driver input.
378        loop {
379            if !self.advance()? {
380                break;
381            }
382        }
383        if self.frames[self.root_frame].done {
384            return Ok(Step::Done(self.report(self.claim_code())));
385        }
386        if self.cancelled.is_cancelled() {
387            return Ok(Step::Done(self.report(1)));
388        }
389        let mut ready = Vec::new();
390        let root_name = self.frames[self.root_frame].name.clone();
391        for frame in self.frames.iter_mut() {
392            if frame.dispatched && !frame.surfaced && !frame.done {
393                frame.surfaced = true;
394                ready.push(ReadyTask {
395                    name: frame.name.clone(),
396                    action: self.graph.get(&frame.name).unwrap().action.clone(),
397                    args: if frame.name == root_name {
398                        self.extra_args.clone()
399                    } else {
400                        Vec::new()
401                    },
402                    env: frame.env.clone(),
403                });
404            }
405        }
406        Ok(Step::Ready(ready))
407    }
408
409    /// Records the outcome of a task handed out by `step`. On success the
410    /// task's outputs are committed to the cache.
411    pub fn finish(&mut self, name: &str, code: u8) {
412        let Some(frame_id) = self
413            .frames
414            .iter()
415            .position(|f| f.name == name && f.dispatched && !f.done)
416        else {
417            return;
418        };
419        if code == 0 && !self.cancelled.is_cancelled() {
420            let fingerprint = self.claims[self.frames[frame_id].claim].fingerprint.clone();
421            if let (Some(cache), Some(fingerprint)) = (&self.cache, fingerprint) {
422                let task = self.graph.get(name).unwrap().clone();
423                if let Err(error) =
424                    cache.record_at(&self.root, &task, &fingerprint, &*self.reporter, name)
425                {
426                    self.reporter.cache_error(name, &error.to_string());
427                }
428            }
429        }
430        self.complete(frame_id, code);
431    }
432
433    fn claim_code(&self) -> u8 {
434        let claim = self.frames[self.root_frame].claim;
435        self.claims[claim].outcome.unwrap_or(1)
436    }
437
438    fn report(&self, code: u8) -> Report {
439        let failed = self
440            .claims
441            .iter()
442            .enumerate()
443            .filter_map(|(index, claim)| {
444                let code = claim.outcome?;
445                if code == 0 || !claim.dispatched {
446                    return None;
447                }
448                let name = self
449                    .claim_by_name
450                    .iter()
451                    .find(|&(_, &claim_index)| claim_index == index)
452                    .map(|(name, _)| name.clone())?;
453                Some(FailedTask { name, code })
454            })
455            .collect();
456        Report {
457            code,
458            completed: self.completed.clone(),
459            failed,
460            cached: self.cached.clone(),
461        }
462    }
463}