flams_system/building/
queueing.rs

1use either::Either;
2use flams_math_archives::{
3    backend::AnyBackend,
4    formats::{BuildTargetId, FormatOrTargets},
5    source_files::{FileState, SourceFile},
6    Archive, MathArchive,
7};
8use flams_utils::{triomphe::Arc, vecmap::VecSet};
9use ftml_ontology::utils::time::Eta;
10use ftml_uris::UriWithArchive;
11use parking_lot::RwLock;
12use std::collections::hash_map::Entry;
13
14use crate::building::{AtomicTaskState, QueueMessage};
15
16use super::{
17    queue::{Queue, QueueState, RunningQueue, TaskMap},
18    BuildStep, BuildStepI, BuildTask, BuildTaskId, Dependency, TaskState,
19};
20
21impl Queue {
22    #[allow(clippy::significant_drop_in_scrutinee)]
23    pub(super) fn sort(map: &TaskMap, state: &mut RunningQueue) {
24        let RunningQueue {
25            queue,
26            done,
27            blocked,
28            failed,
29            ..
30        } = state;
31        let mut tasks = map.map.values().cloned().collect::<Vec<_>>();
32        let mut weak = true;
33        while !tasks.is_empty() {
34            let mut changed = false;
35            for t in &tasks {
36                let mut has_failed = false;
37                let Some(step) = t.steps().iter().find(|s| {
38                    let state = s.0.state.get();
39                    if state == TaskState::Failed {
40                        has_failed = true;
41                        return false;
42                    }
43                    !matches!(state, TaskState::Done)
44                }) else {
45                    if has_failed {
46                        failed.push(t.clone());
47                    } else {
48                        done.push(t.clone());
49                    }
50                    continue;
51                };
52                let mut newstate = TaskState::Queued;
53                for d in step.0.requires.read().iter() {
54                    match d {
55                        Dependency::Resolved { task, strict, step } if *strict || weak => {
56                            match task
57                                .get_step(*step)
58                                .unwrap_or_else(|| unreachable!())
59                                .0
60                                .state
61                                .get()
62                            {
63                                TaskState::Done
64                                | TaskState::Queued
65                                | TaskState::Failed
66                                | TaskState::Running => (),
67                                TaskState::Blocked => {
68                                    newstate = TaskState::Blocked;
69                                }
70                                TaskState::None => {
71                                    newstate = TaskState::None;
72                                    break;
73                                }
74                            }
75                        }
76                        _ => (),
77                    }
78                }
79                let mut found = false;
80                if newstate == TaskState::None {
81                    continue;
82                }
83                changed = true;
84                for s in t.steps() {
85                    if s == step {
86                        found = true;
87                        s.0.state.set(newstate);
88                    } else if found {
89                        s.0.state.set(TaskState::Blocked);
90                    }
91                }
92                match newstate {
93                    TaskState::Blocked => blocked.push(t.clone()),
94                    TaskState::Queued => queue.push_back(t.clone()),
95                    _ => (),
96                }
97            }
98            if changed {
99                tasks.retain(|t| t.steps().iter().any(|s| s.0.state.get() == TaskState::None));
100            } else if weak {
101                weak = false;
102            } else {
103                let tasks = std::mem::take(&mut tasks);
104                for t in tasks {
105                    for s in t.steps() {
106                        s.0.state.set_if_is(TaskState::None, TaskState::Blocked);
107                    }
108                    blocked.push(t);
109                }
110            }
111        }
112    }
113
114    pub(super) fn get_next(&self) -> Option<(BuildTask, BuildTargetId)> {
115        loop {
116            match self.get_next_i() {
117                Ok(r) => return r,
118                Err(()) => std::thread::sleep(std::time::Duration::from_millis(100)),
119            }
120        }
121    }
122
123    #[cfg(feature = "tokio")]
124    pub(super) async fn get_next_async(&self) -> Option<(BuildTask, BuildTargetId)> {
125        loop {
126            match self.get_next_i() {
127                Ok(r) => return r,
128                Err(()) => tokio::time::sleep(std::time::Duration::from_millis(100)).await,
129            }
130        }
131    }
132
133    fn get_next_i(&self) -> Result<Option<(BuildTask, BuildTargetId)>, ()> {
134        let mut state = self.0.state.write();
135        let QueueState::Running(RunningQueue {
136            queue,
137            blocked,
138            running,
139            failed,
140            ..
141        }) = &mut *state
142        else {
143            unreachable!()
144        };
145        if queue.is_empty() && blocked.is_empty() && running.is_empty() {
146            return Ok(None);
147        }
148        if let Some((i, target)) = queue
149            .iter()
150            .enumerate()
151            .find_map(|(next, e)| Self::can_be_next(e).map(|t| (next, t)))
152        {
153            let Some(task) = queue.remove(i) else {
154                unreachable!()
155            };
156            task.get_step(target)
157                .unwrap_or_else(|| unreachable!())
158                .0
159                .state
160                .set(TaskState::Running);
161            running.push(task.clone());
162            return Ok(Some((task, target)));
163        }
164        if !running.is_empty() {
165            drop(state);
166            Err(())
167            //std::thread::sleep(std::time::Duration::from_secs(1));
168        } else if !blocked.is_empty() {
169            if let Some((i, target)) = blocked
170                .iter()
171                .enumerate()
172                .find_map(|(next, e)| Self::can_be_next(e).map(|t| (next, t)))
173            {
174                let task = blocked.remove(i);
175                for s in task.steps() {
176                    s.0.state.set_if_is(TaskState::Blocked, TaskState::Queued);
177                }
178                task.get_step(target)
179                    .unwrap_or_else(|| unreachable!())
180                    .0
181                    .state
182                    .set(TaskState::Running);
183                running.push(task.clone());
184                return Ok(Some((task, target)));
185            }
186            while let Some(t) = blocked.pop() {
187                for s in t.steps() {
188                    let state = s.0.state.get();
189                    if state != TaskState::Done {
190                        s.0.state.set(TaskState::Failed);
191                    }
192                }
193                self.0.sender.lazy_send(|| QueueMessage::TaskFailed {
194                    id: t.0.id,
195                    target: t.steps().last().expect("???").0.target,
196                    eta: Eta::default(),
197                });
198                failed.push(t);
199            }
200            Ok(None)
201        } else {
202            Ok(None)
203        }
204    }
205
206    fn can_be_next(e: &BuildTask) -> Option<BuildTargetId> {
207        let step =
208            e.0.steps.iter().find(|step| {
209                matches!(step.0.state.get(), TaskState::Queued | TaskState::Blocked)
210            })?;
211        for d in &step.0.requires.read().0 {
212            if let Dependency::Resolved { task, step, strict } = d {
213                if *strict
214                    && task
215                        .get_step(*step)
216                        .unwrap_or_else(|| unreachable!())
217                        .0
218                        .state
219                        .get()
220                        == TaskState::Running
221                {
222                    return None;
223                }
224            }
225        }
226        Some(step.0.target)
227    }
228
229    #[deprecated(note = "assumes local archives")]
230    pub(super) fn enqueue<'a, I: Iterator<Item = &'a SourceFile>>(
231        map: &mut TaskMap,
232        backend: &AnyBackend,
233        archive: &Archive,
234        target: FormatOrTargets,
235        stale_only: bool,
236        files: I,
237    ) -> usize {
238        let targets = match target {
239            FormatOrTargets::Format(f) => f.targets,
240            FormatOrTargets::Targets(t) => t,
241        };
242        let has_target = |f: &SourceFile, tgt: BuildTargetId| {
243            f.target_state
244                .iter()
245                .find_map(|(k, v)| if *k == tgt { Some(v) } else { None })
246                .is_some_and(|t| !stale_only || matches!(t, FileState::Stale(_) | FileState::New))
247        };
248        let should_queue = |f: &SourceFile| targets.iter().any(|t| has_target(f, *t));
249        let mut count = 0;
250
251        for f in files.filter(|f| should_queue(f)) {
252            let key = (archive.id().clone(), f.relative_path.clone());
253            let task = match map.map.entry(key) {
254                Entry::Vacant(e) => {
255                    count += 1;
256                    let steps = targets
257                        .iter()
258                        .filter(|t| has_target(f, **t))
259                        .map(|t| {
260                            BuildStep(Arc::new(BuildStepI {
261                                target: *t,
262                                state: AtomicTaskState::new(TaskState::None),
263                                //yields:RwLock::new(Vec::new()),
264                                requires: RwLock::new(VecSet::default()),
265                                dependents: RwLock::new(Vec::new()),
266                            }))
267                        })
268                        .collect::<Vec<_>>()
269                        .into_boxed_slice();
270                    map.total += steps.len();
271                    let id = map.counter;
272                    map.counter = map.counter.saturating_add(1);
273                    let task = BuildTask::new(
274                        BuildTaskId(id),
275                        archive.uri().clone(),
276                        steps,
277                        match archive {
278                            Archive::Local(archive) => {
279                                Either::Left(archive.source_dir().join(f.relative_path.as_ref()))
280                            }
281                            Archive::Ext(..) => todo!("foreign archives"),
282                        },
283                        f.relative_path.clone(),
284                    )
285                    .expect("this is a bug");
286                    e.insert(task.clone());
287                    task
288                }
289                Entry::Occupied(o) => {
290                    count += 1;
291                    for s in o.get().steps() {
292                        s.0.state.set(TaskState::None);
293                    }
294                    continue;
295                }
296            };
297            let spec = task.as_build_spec(backend);
298            if let FormatOrTargets::Format(fmt) = target {
299                for (f, d) in (fmt.dependencies)(spec) {
300                    if let Some(step) = task.get_step(f) {
301                        step.add_dependency(d.into());
302                    }
303                }
304                Self::process_dependencies(&task, map);
305            }
306        }
307        count
308    }
309
310    fn process_dependencies(task: &BuildTask, map: &mut TaskMap) {
311        for s in task.steps() {
312            let key = task.as_task_ref(s.0.target);
313            if let Some(v) = map.dependents.remove(&key) {
314                for (d, i) in v {
315                    if let Some(t) = d.get_step(s.0.target) {
316                        let mut deps = t.0.requires.write();
317                        for d in &mut deps.0 {
318                            if let Dependency::Physical {
319                                task: ref t,
320                                strict,
321                            } = d
322                            {
323                                if *t == key {
324                                    *d = Dependency::Resolved {
325                                        task: task.clone(),
326                                        step: i,
327                                        strict: *strict,
328                                    };
329                                }
330                            }
331                        }
332                    }
333                }
334            }
335            for dep in &mut s.0.requires.write().0 {
336                if let Dependency::Physical {
337                    task: ref deptask,
338                    strict,
339                } = dep
340                {
341                    if deptask.archive == *task.0.uri.archive_id()
342                        && deptask.rel_path == task.0.rel_path
343                    {
344                        continue;
345                        // TODO check for more
346                    }
347                    let key = (deptask.archive.clone(), deptask.rel_path.clone());
348                    if let Some(deptasks) = map.map.get(&key) {
349                        if let Some(step) =
350                            deptasks.steps().iter().find(|bt| bt.0.target == s.0.target)
351                        {
352                            step.0.dependents.write().push((task.0.id, s.0.target));
353                            *dep = Dependency::Resolved {
354                                task: deptasks.clone(),
355                                step: s.0.target,
356                                strict: *strict,
357                            };
358                            continue;
359                        }
360                        map.dependents
361                            .entry(deptask.clone())
362                            .or_default()
363                            .push((task.clone(), s.0.target));
364                    }
365                }
366            }
367        }
368    }
369}
flams_system/building/queueing.rs

flams_system/building/
queueing.rs
