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

qframe/runtime/
task.rs
