qframe/runtime/
task.rs

1//! Background tasks with progress, cancellation and an outcome, and the model that tracks them
2//! for display.
3//!
4//! A [`Task`] runs on its own thread and talks to the application only through messages, so
5//! drawing never waits for it. The application keeps a [`Tasks`] model up to date from
6//! [`TaskEvent`]s and shows it, for example with [`TaskList`](crate::widgets::TaskList).
7
8use std::collections::{HashMap, HashSet};
9use std::io;
10use std::panic::{AssertUnwindSafe, catch_unwind};
11use std::sync::atomic::{AtomicU64, Ordering};
12use std::sync::mpsc::Sender;
13use std::sync::{Arc, Condvar, Mutex, MutexGuard, PoisonError};
14use std::time::{Duration, Instant};
15
16/// Identifies one task for its whole life. Ids are unique within the process.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
18pub struct TaskId(u64);
19
20impl TaskId {
21    fn next() -> Self {
22        static NEXT: AtomicU64 = AtomicU64::new(1);
23        Self(NEXT.fetch_add(1, Ordering::Relaxed))
24    }
25}
26
27/// How a task ended.
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub enum TaskOutcome {
30    /// The work returned `Ok`; its message was delivered just before this outcome.
31    Done,
32    /// The work returned `Err` with this reason, or panicked.
33    Failed(String),
34    /// [`Command::cancel_task`](crate::runtime::Command::cancel_task) asked it to stop; its
35    /// result was dropped.
36    Cancelled,
37}
38
39/// What happened to a task, delivered through [`Task::on_event`].
40#[derive(Debug, Clone, PartialEq)]
41pub enum TaskEvent {
42    /// The task was started with this label.
43    Started {
44        /// The task.
45        id: TaskId,
46        /// Its label.
47        label: String,
48    },
49    /// The task reported progress. `None` fields keep their previous value.
50    Progress {
51        /// The task.
52        id: TaskId,
53        /// Completed share from 0 to 1.
54        fraction: Option<f32>,
55        /// A short note about the current step.
56        note: Option<String>,
57    },
58    /// The task ended.
59    Finished {
60        /// The task.
61        id: TaskId,
62        /// How.
63        outcome: TaskOutcome,
64    },
65}
66
67impl TaskEvent {
68    /// The task the event is about.
69    #[must_use]
70    pub fn id(&self) -> TaskId {
71        match self {
72            Self::Started { id, .. } | Self::Progress { id, .. } | Self::Finished { id, .. } => *id,
73        }
74    }
75}
76
77type Work<Msg> = Box<dyn FnOnce(&TaskCx<Msg>) -> Result<Msg, String> + Send>;
78type EventMessage<Msg> = Arc<dyn Fn(TaskEvent) -> Msg + Send + Sync>;
79
80/// Work to run in the background with progress, cancellation and an outcome.
81///
82/// ```
83/// use std::time::Duration;
84/// use qframe::runtime::{Command, Task, TaskEvent};
85///
86/// enum Msg {
87///     Task(TaskEvent),
88///     Built(String),
89/// }
90///
91/// let task = Task::new("Build image", |cx| {
92///     for step in 0..4 {
93///         if !cx.sleep(Duration::from_millis(300)) {
94///             return Err("stopped".into());
95///         }
96///         cx.progress((step + 1) as f32 / 4.0);
97///     }
98///     Ok(Msg::Built("sha256:4f2a".into()))
99/// })
100/// .on_event(Msg::Task);
101/// let id = task.id(); // keep it to cancel the task later
102/// let command: Command<Msg> = Command::task(task);
103/// # let _ = (id, command);
104/// ```
105pub struct Task<Msg> {
106    id: TaskId,
107    label: String,
108    work: Work<Msg>,
109    on_event: Option<EventMessage<Msg>>,
110}
111
112impl<Msg: Send + 'static> Task<Msg> {
113    /// A task labelled `label` running `work`. `Ok` delivers its message; `Err` fails the task
114    /// with a reason.
115    #[must_use]
116    pub fn new(
117        label: impl Into<String>,
118        work: impl FnOnce(&TaskCx<Msg>) -> Result<Msg, String> + Send + 'static,
119    ) -> Self {
120        Self { id: TaskId::next(), label: label.into(), work: Box::new(work), on_event: None }
121    }
122
123    /// Turns start, progress and the outcome into messages, e.g. to update a [`Tasks`] model.
124    #[must_use]
125    pub fn on_event(mut self, message: impl Fn(TaskEvent) -> Msg + Send + Sync + 'static) -> Self {
126        self.on_event = Some(Arc::new(message));
127        self
128    }
129
130    /// The task's id, known before it starts.
131    #[must_use]
132    pub fn id(&self) -> TaskId {
133        self.id
134    }
135
136    /// The task's label.
137    #[must_use]
138    pub fn label(&self) -> &str {
139        &self.label
140    }
141}
142
143/// A message from a background thread to the event loop.
144pub(crate) enum Delivery<Msg> {
145    /// Apply this message.
146    Message(Msg),
147    /// A piece of background work ended.
148    Ended,
149}
150
151/// Time as tasks see it: the real clock, or the test harness's fake clock that only moves when
152/// the test advances it.
153pub(crate) struct TaskClock {
154    fake: bool,
155    state: Mutex<ClockState>,
156    changed: Condvar,
157}
158
159#[derive(Default)]
160struct ClockState {
161    now: Duration,
162    /// Tasks that are working rather than sleeping.
163    busy: usize,
164    cancelled: HashSet<TaskId>,
165    /// Fake clock only: tasks asleep and when they wake. Whoever wakes a sleeper (the clock
166    /// moving or a cancel) counts it as busy right away, so settling never misses it.
167    sleeping: HashMap<TaskId, Duration>,
168    /// Fake clock only: where each task is in time. A task woken by a big clock jump carries on
169    /// from the moment its sleep ended, so a loop of sleeps keeps its rhythm.
170    task_time: HashMap<TaskId, Duration>,
171}
172
173/// How long the harness waits for tasks to reach a sleep before it gives up.
174const SETTLE_LIMIT: Duration = Duration::from_secs(10);
175
176impl TaskClock {
177    pub(crate) fn new(fake: bool) -> Arc<Self> {
178        Arc::new(Self { fake, state: Mutex::new(ClockState::default()), changed: Condvar::new() })
179    }
180
181    fn lock(&self) -> MutexGuard<'_, ClockState> {
182        self.state.lock().unwrap_or_else(PoisonError::into_inner)
183    }
184
185    /// Asks task `id` to stop; its sleeps return at once.
186    pub(crate) fn cancel(&self, id: TaskId) {
187        let mut state = self.lock();
188        state.cancelled.insert(id);
189        if state.sleeping.remove(&id).is_some() {
190            state.busy += 1;
191            let now = state.now;
192            state.task_time.insert(id, now);
193        }
194        drop(state);
195        self.changed.notify_all();
196    }
197
198    /// Moves the fake clock to `now` and waits until every task is asleep or finished.
199    ///
200    /// # Panics
201    ///
202    /// Panics when a task keeps working for longer than ten seconds, which in a test means it
203    /// never sleeps or blocks forever.
204    pub(crate) fn settle(&self, now: Duration) {
205        let mut state = self.lock();
206        state.now = state.now.max(now);
207        let current = state.now;
208        let due: Vec<(TaskId, Duration)> =
209            state.sleeping.iter().filter(|(_, until)| **until <= current).map(|(id, until)| (*id, *until)).collect();
210        for (id, until) in due {
211            state.sleeping.remove(&id);
212            state.task_time.insert(id, until);
213            state.busy += 1;
214        }
215        self.changed.notify_all();
216        let started = Instant::now();
217        while state.busy > 0 {
218            let waited = started.elapsed();
219            assert!(waited < SETTLE_LIMIT, "a background task kept working for {SETTLE_LIMIT:?} without sleeping");
220            state = self.changed.wait_timeout(state, SETTLE_LIMIT - waited).unwrap_or_else(PoisonError::into_inner).0;
221        }
222    }
223
224    fn is_cancelled(&self, id: TaskId) -> bool {
225        self.lock().cancelled.contains(&id)
226    }
227
228    fn begin(&self, id: TaskId) {
229        let mut state = self.lock();
230        state.busy += 1;
231        let now = state.now;
232        state.task_time.insert(id, now);
233    }
234
235    fn end(&self, id: TaskId) {
236        let mut state = self.lock();
237        state.busy = state.busy.saturating_sub(1);
238        state.cancelled.remove(&id);
239        state.task_time.remove(&id);
240        drop(state);
241        self.changed.notify_all();
242    }
243
244    /// Sleeps task `id` for `duration`; returns `false` when it was cancelled.
245    fn sleep(&self, id: TaskId, duration: Duration) -> bool {
246        let mut state = self.lock();
247        if self.fake {
248            let until = state.task_time.get(&id).copied().unwrap_or(state.now) + duration;
249            if until <= state.now {
250                state.task_time.insert(id, until);
251            } else if !state.cancelled.contains(&id) {
252                state.sleeping.insert(id, until);
253                state.busy = state.busy.saturating_sub(1);
254                self.changed.notify_all();
255                while state.sleeping.contains_key(&id) {
256                    state = self.changed.wait(state).unwrap_or_else(PoisonError::into_inner);
257                }
258            }
259        } else {
260            let deadline = Instant::now() + duration;
261            while !state.cancelled.contains(&id) {
262                let left = deadline.saturating_duration_since(Instant::now());
263                if left.is_zero() {
264                    break;
265                }
266                state = self.changed.wait_timeout(state, left).unwrap_or_else(PoisonError::into_inner).0;
267            }
268        }
269        !state.cancelled.contains(&id)
270    }
271}
272
273/// What running work can do: report progress, send messages, notice cancellation and sleep.
274pub struct TaskCx<Msg> {
275    id: TaskId,
276    clock: Arc<TaskClock>,
277    sender: Sender<Delivery<Msg>>,
278    on_event: Option<EventMessage<Msg>>,
279}
280
281impl<Msg: Send + 'static> TaskCx<Msg> {
282    /// The running task.
283    #[must_use]
284    pub fn id(&self) -> TaskId {
285        self.id
286    }
287
288    /// Reports the completed share, from 0 to 1.
289    pub fn progress(&self, fraction: f32) {
290        self.event(TaskEvent::Progress { id: self.id, fraction: Some(fraction.clamp(0.0, 1.0)), note: None });
291    }
292
293    /// Describes the current step, e.g. "pushing layers".
294    pub fn note(&self, note: impl Into<String>) {
295        self.event(TaskEvent::Progress { id: self.id, fraction: None, note: Some(note.into()) });
296    }
297
298    /// Delivers `message` to the application while the work goes on, e.g. a log line.
299    pub fn send(&self, message: Msg) {
300        let _ = self.sender.send(Delivery::Message(message));
301    }
302
303    /// Whether the application asked this task to stop. Long work should check it and return.
304    #[must_use]
305    pub fn is_cancelled(&self) -> bool {
306        self.clock.is_cancelled(self.id)
307    }
308
309    /// Waits `duration`, waking early when the task is cancelled. Returns `false` when it was
310    /// cancelled. In tests the harness's fake clock decides when the sleep ends.
311    #[must_use]
312    pub fn sleep(&self, duration: Duration) -> bool {
313        self.clock.sleep(self.id, duration)
314    }
315
316    fn event(&self, event: TaskEvent) {
317        if let Some(message) = &self.on_event {
318            self.send(message(event));
319        }
320    }
321}
322
323/// Starts a thread named `name` running `run`. A failed start drops `run`.
324pub(crate) type Spawner = fn(String, Box<dyn FnOnce() + Send>) -> io::Result<()>;
325
326/// The [`Spawner`] of the runtime: a real thread.
327pub(crate) fn spawn_thread(name: String, run: Box<dyn FnOnce() + Send>) -> io::Result<()> {
328    std::thread::Builder::new().name(name).spawn(run).map(drop)
329}
330
331/// The reason of a task whose thread could not start.
332const NO_THREAD: &str = "could not start a thread";
333
334/// Starts `task` on a thread. The `Started` message is returned for the caller to apply at once.
335/// When no thread can start, the task fails at once and still ends.
336pub(crate) fn spawn<Msg: Send + 'static>(
337    task: Task<Msg>,
338    clock: &Arc<TaskClock>,
339    sender: &Sender<Delivery<Msg>>,
340    spawner: Spawner,
341) -> Option<Msg> {
342    let Task { id, label, work, on_event } = task;
343    let started = on_event.as_ref().map(|message| message(TaskEvent::Started { id, label: label.clone() }));
344    let failed = on_event.clone();
345    let cx = TaskCx { id, clock: Arc::clone(clock), sender: sender.clone(), on_event };
346    clock.begin(id);
347    let run = Box::new(move || {
348        let result = catch_unwind(AssertUnwindSafe(|| work(&cx)))
349            .unwrap_or_else(|_| Err(format!("the task `{label}` panicked")));
350        let outcome = match result {
351            _ if cx.is_cancelled() => TaskOutcome::Cancelled,
352            Ok(message) => {
353                cx.send(message);
354                TaskOutcome::Done
355            }
356            Err(reason) => TaskOutcome::Failed(reason),
357        };
358        cx.event(TaskEvent::Finished { id, outcome });
359        let _ = cx.sender.send(Delivery::Ended);
360        cx.clock.end(id);
361    });
362    if spawner(format!("quvyta-task-{}", id.0), run).is_err() {
363        // The work went with the closure; report the failure so the task never stays running.
364        clock.end(id);
365        if let Some(message) = failed {
366            let outcome = TaskOutcome::Failed(NO_THREAD.to_owned());
367            let _ = sender.send(Delivery::Message(message(TaskEvent::Finished { id, outcome })));
368        }
369        let _ = sender.send(Delivery::Ended);
370    }
371    started
372}
373
374/// One task as the application shows it.
375#[derive(Debug, Clone, PartialEq)]
376pub struct TaskEntry {
377    /// The task.
378    pub id: TaskId,
379    /// Its label.
380    pub label: String,
381    /// Completed share from 0 to 1, when the task reports one.
382    pub fraction: Option<f32>,
383    /// The latest note.
384    pub note: Option<String>,
385    /// How it ended; `None` while running.
386    pub outcome: Option<TaskOutcome>,
387}
388
389/// Tasks an application shows, kept up to date with [`Tasks::apply`]. Newest last.
390#[derive(Debug, Clone, Default, PartialEq)]
391pub struct Tasks {
392    entries: Vec<TaskEntry>,
393}
394
395impl Tasks {
396    /// No tasks.
397    #[must_use]
398    pub fn new() -> Self {
399        Self::default()
400    }
401
402    /// Records `event`.
403    pub fn apply(&mut self, event: &TaskEvent) {
404        match event {
405            TaskEvent::Started { id, label } => {
406                self.entries.retain(|entry| entry.id != *id);
407                self.entries.push(TaskEntry {
408                    id: *id,
409                    label: label.clone(),
410                    fraction: None,
411                    note: None,
412                    outcome: None,
413                });
414            }
415            TaskEvent::Progress { id, fraction, note } => {
416                if let Some(entry) = self.entries.iter_mut().find(|entry| entry.id == *id) {
417                    entry.fraction = fraction.or(entry.fraction);
418                    if note.is_some() {
419                        entry.note.clone_from(note);
420                    }
421                }
422            }
423            TaskEvent::Finished { id, outcome } => {
424                if let Some(entry) = self.entries.iter_mut().find(|entry| entry.id == *id) {
425                    entry.outcome = Some(outcome.clone());
426                }
427            }
428        }
429    }
430
431    /// Every task, oldest first.
432    #[must_use]
433    pub fn entries(&self) -> &[TaskEntry] {
434        &self.entries
435    }
436
437    /// The task `id`.
438    #[must_use]
439    pub fn get(&self, id: TaskId) -> Option<&TaskEntry> {
440        self.entries.iter().find(|entry| entry.id == id)
441    }
442
443    /// How many tasks are still running.
444    #[must_use]
445    pub fn running(&self) -> usize {
446        self.entries.iter().filter(|entry| entry.outcome.is_none()).count()
447    }
448
449    /// Forgets finished tasks.
450    pub fn clear_finished(&mut self) {
451        self.entries.retain(|entry| entry.outcome.is_none());
452    }
453}
454
455#[cfg(test)]
456mod tests {
457    use super::*;
458    use crate::runtime::engine::{Engine, TaskMode};
459    use crate::runtime::{App, Command, Harness};
460    use crate::widget::View;
461    use crate::widgets::Text;
462
463    #[derive(Default)]
464    struct Pipeline {
465        tasks: Tasks,
466        built: Option<String>,
467        lines: Vec<String>,
468        build: Option<TaskId>,
469    }
470
471    enum Msg {
472        Build,
473        Cancel,
474        Fail,
475        Panic,
476        Task(TaskEvent),
477        Built(String),
478        Line(String),
479    }
480
481    impl App for Pipeline {
482        type Msg = Msg;
483        fn update(&mut self, msg: Msg) -> Command<Msg> {
484            match msg {
485                Msg::Build => {
486                    let task = Task::new("Build image", |cx| {
487                        cx.note("resolving layers");
488                        for step in 0..4 {
489                            if !cx.sleep(Duration::from_millis(100)) {
490                                return Err("stopped".into());
491                            }
492                            cx.progress((step + 1) as f32 / 4.0);
493                            cx.send(Msg::Line(format!("layer {step}")));
494                        }
495                        Ok(Msg::Built("sha256:4f2a".into()))
496                    })
497                    .on_event(Msg::Task);
498                    self.build = Some(task.id());
499                    return Command::task(task);
500                }
501                Msg::Cancel => return self.build.map_or_else(Command::none, Command::cancel_task),
502                Msg::Fail => {
503                    return Command::task(
504                        Task::new("Sync registry", |cx| {
505                            let _ = cx.sleep(Duration::from_millis(50));
506                            Err("registry timed out".into())
507                        })
508                        .on_event(Msg::Task),
509                    );
510                }
511                Msg::Panic => {
512                    return Command::task(
513                        Task::new("Broken", |_| -> Result<Msg, String> { panic!("boom") }).on_event(Msg::Task),
514                    );
515                }
516                Msg::Task(event) => self.tasks.apply(&event),
517                Msg::Built(digest) => self.built = Some(digest),
518                Msg::Line(line) => self.lines.push(line),
519            }
520            Command::none()
521        }
522        fn view(&self, ui: &mut View<'_, Msg>) {
523            ui.add(Text::new(format!("running {}", self.tasks.running())));
524        }
525    }
526
527    #[test]
528    fn progress_follows_the_fake_clock_and_completes() {
529        let mut h = Harness::new(Pipeline::default(), 20, 1);
530        h.send(Msg::Build);
531        assert_eq!(h.screen(), "running 1\n");
532        let entry = h.app().tasks.entries()[0].clone();
533        assert_eq!(entry.label, "Build image");
534        assert_eq!(entry.note.as_deref(), Some("resolving layers"));
535        assert_eq!(entry.fraction, None);
536        h.advance(Duration::from_millis(100));
537        assert_eq!(h.app().tasks.entries()[0].fraction, Some(0.25));
538        assert_eq!(h.app().lines, ["layer 0"]);
539        h.advance(Duration::from_millis(250));
540        assert_eq!(h.app().tasks.entries()[0].fraction, Some(0.75));
541        h.advance(Duration::from_millis(100));
542        assert_eq!(h.app().built.as_deref(), Some("sha256:4f2a"));
543        assert_eq!(h.app().tasks.entries()[0].outcome, Some(TaskOutcome::Done));
544        assert_eq!(h.screen(), "running 0\n");
545    }
546
547    #[test]
548    fn cancelling_wakes_the_sleep_and_drops_the_result() {
549        let mut h = Harness::new(Pipeline::default(), 20, 1);
550        h.send(Msg::Build).advance(Duration::from_millis(150)).send(Msg::Cancel);
551        let entry = &h.app().tasks.entries()[0];
552        assert_eq!(entry.outcome, Some(TaskOutcome::Cancelled));
553        assert_eq!(entry.fraction, Some(0.25));
554        assert!(h.app().built.is_none());
555    }
556
557    fn no_thread(_: String, _: Box<dyn FnOnce() + Send>) -> io::Result<()> {
558        Err(io::Error::other("no threads left"))
559    }
560
561    #[test]
562    fn a_task_whose_thread_cannot_start_fails_and_ends() {
563        let mut engine = Engine::new(Pipeline::default(), crate::env::Env::builtin(), TaskMode::Threads);
564        engine.spawner = no_thread;
565        engine.update(Msg::Build);
566        assert_eq!(engine.poll_tasks(), 2, "Finished, then Ended");
567        let entry = &engine.app.tasks.entries()[0];
568        assert_eq!(entry.outcome, Some(TaskOutcome::Failed("could not start a thread".into())));
569        assert_eq!((engine.app.tasks.running(), engine.pending_tasks), (0, 0));
570        assert!(engine.app.built.is_none());
571    }
572
573    #[test]
574    fn failures_and_panics_become_outcomes() {
575        let mut h = Harness::new(Pipeline::default(), 20, 1);
576        h.send(Msg::Fail).send(Msg::Panic);
577        assert_eq!(h.app().tasks.running(), 1, "the failing task still sleeps");
578        assert_eq!(h.app().tasks.entries()[1].outcome, Some(TaskOutcome::Failed("the task `Broken` panicked".into())));
579        h.advance(Duration::from_millis(50));
580        assert_eq!(h.app().tasks.entries()[0].outcome, Some(TaskOutcome::Failed("registry timed out".into())));
581        let mut tasks = h.app().tasks.clone();
582        tasks.clear_finished();
583        assert!(tasks.entries().is_empty());
584    }
585}
qframe/runtime/task.rs

qframe/runtime/
task.rs
