ferroui_system/threading/tasks/
task.rs

1//! `Task<TResult>`, `TaskStatus`, `TaskCompletionSource<TResult>` and `TaskCanceledException`
2//! (`System.Threading.Tasks`), single-threaded (docs/porting/reactive.md, section 5.8).
3
4use crate::ObservableError;
5use std::cell::{Cell, RefCell};
6use std::error::Error;
7use std::fmt;
8use std::future::Future;
9use std::pin::Pin;
10use std::rc::Rc;
11use std::sync::atomic::{AtomicU64, Ordering};
12use std::task::{Context, Poll, Waker};
13
14/// The state of a task (`TaskStatus`).
15#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
16pub enum TaskStatus {
17    /// Made, and not started (`Created`).
18    Created,
19    /// Waiting for its completion source or its future (`WaitingForActivation`).
20    WaitingForActivation,
21    /// Its future is being polled (`Running`).
22    Running,
23    /// Ended with a result (`RanToCompletion`).
24    RanToCompletion,
25    /// Ended with an error (`Faulted`).
26    Faulted,
27    /// Ended by a cancellation (`Canceled`).
28    Canceled,
29}
30
31/// The error of a canceled task (`TaskCanceledException`).
32#[derive(Clone, Debug, Default, PartialEq, Eq)]
33pub struct TaskCanceledError {
34    /// The id of the task, when the error names one (`TaskCanceledException.Task`).
35    pub task_id: Option<u64>,
36}
37
38impl fmt::Display for TaskCanceledError {
39    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
40        f.write_str("A task was canceled.")
41    }
42}
43
44impl Error for TaskCanceledError {}
45
46impl TaskCanceledError {
47    /// Whether `error` is a cancellation: a [`TaskCanceledError`].
48    pub fn is_cancellation(error: &ObservableError) -> bool {
49        error.downcast_ref::<TaskCanceledError>().is_some()
50    }
51}
52
53fn next_task_id() -> u64 {
54    static NEXT: AtomicU64 = AtomicU64::new(1);
55    NEXT.fetch_add(1, Ordering::Relaxed)
56}
57
58type Continuation = Box<dyn FnOnce()>;
59
60/// A continuation registered on a task that has not ended ([`Task::continue_with`]), for
61/// [`Task::remove_continuation`].
62#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
63pub struct ContinuationId(u64);
64
65struct TaskInner<T> {
66    id: u64,
67    status: Cell<TaskStatus>,
68    result: RefCell<Option<T>>,
69    error: RefCell<Option<ObservableError>>,
70    continuations: RefCell<Vec<(u64, Continuation)>>,
71    next_continuation: Cell<u64>,
72    wakers: RefCell<Vec<Waker>>,
73}
74
75/// The outcome of an asynchronous operation (`Task<TResult>`; `Task` is `Task<()>`): a shared
76/// handle that ends once, with a result, an error or a cancellation.
77///
78/// Continuations ([`continue_with`](Self::continue_with)) run synchronously when the task ends
79/// (`TaskContinuationOptions.ExecuteSynchronously`); a task is also a [`Future`] of its outcome.
80/// Tasks belong to the thread that made them, like the observables
81/// (docs/porting/reactive.md, section 5.8).
82pub struct Task<T> {
83    inner: Rc<TaskInner<T>>,
84}
85
86impl<T> Clone for Task<T> {
87    fn clone(&self) -> Self {
88        Self { inner: self.inner.clone() }
89    }
90}
91
92impl<T> fmt::Debug for Task<T> {
93    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
94        f.debug_struct("Task").field("id", &self.inner.id).field("status", &self.inner.status.get()).finish()
95    }
96}
97
98impl<T: 'static> Task<T> {
99    pub(crate) fn with_status(status: TaskStatus) -> Self {
100        Self {
101            inner: Rc::new(TaskInner {
102                id: next_task_id(),
103                status: Cell::new(status),
104                result: RefCell::new(None),
105                error: RefCell::new(None),
106                continuations: RefCell::new(Vec::new()),
107                next_continuation: Cell::new(0),
108                wakers: RefCell::new(Vec::new()),
109            }),
110        }
111    }
112
113    /// A task that has ended with `result` (`Task.FromResult`).
114    pub fn from_result(result: T) -> Self {
115        let task = Self::with_status(TaskStatus::WaitingForActivation);
116        task.try_complete(Ok(result));
117        task
118    }
119
120    /// A task that has ended with `error` (`Task.FromException`); a [`TaskCanceledError`] makes it
121    /// canceled.
122    pub fn from_exception(error: ObservableError) -> Self {
123        let task = Self::with_status(TaskStatus::WaitingForActivation);
124        task.try_complete(Err(error));
125        task
126    }
127
128    /// A canceled task (`Task.FromCanceled`).
129    pub fn from_canceled() -> Self {
130        let task = Self::with_status(TaskStatus::WaitingForActivation);
131        task.try_cancel();
132        task
133    }
134
135    /// Runs `future` on the current thread and gives its task (the counterpart of an `async`
136    /// method): see [`spawn_local`](super::spawn_local).
137    pub fn spawn_local(future: impl Future<Output = Result<T, ObservableError>> + 'static) -> Self {
138        super::local_executor::spawn_local(future)
139    }
140
141    /// The identity of the task (`Id`).
142    pub fn id(&self) -> u64 {
143        self.inner.id
144    }
145
146    /// The state of the task (`Status`).
147    pub fn status(&self) -> TaskStatus {
148        self.inner.status.get()
149    }
150
151    /// Whether the task has ended, in any way (`IsCompleted`).
152    pub fn is_completed(&self) -> bool {
153        matches!(self.status(), TaskStatus::RanToCompletion | TaskStatus::Faulted | TaskStatus::Canceled)
154    }
155
156    /// Whether the task has ended with a result (`IsCompletedSuccessfully`).
157    pub fn is_completed_successfully(&self) -> bool {
158        self.status() == TaskStatus::RanToCompletion
159    }
160
161    /// Whether the task has ended with an error (`IsFaulted`).
162    pub fn is_faulted(&self) -> bool {
163        self.status() == TaskStatus::Faulted
164    }
165
166    /// Whether the task has been canceled (`IsCanceled`).
167    pub fn is_canceled(&self) -> bool {
168        self.status() == TaskStatus::Canceled
169    }
170
171    /// The error of a faulted task (`Exception`, unwrapped from its `AggregateException`).
172    pub fn exception(&self) -> Option<ObservableError> {
173        self.inner.error.borrow().clone()
174    }
175
176    /// The outcome of an ended task: the result, the error of a faulted task, or a
177    /// [`TaskCanceledError`] (`Result`, `GetAwaiter().GetResult()`).
178    ///
179    /// # Panics
180    /// When the task has not ended: upstream blocks, which would deadlock the only thread.
181    pub fn result(&self) -> Result<T, ObservableError>
182    where
183        T: Clone,
184    {
185        match self.status() {
186            TaskStatus::RanToCompletion => Ok(self.inner.result.borrow().clone().expect("a completed task holds its result")),
187            TaskStatus::Faulted => Err(self.exception().expect("a faulted task holds its error")),
188            TaskStatus::Canceled => Err(Rc::new(TaskCanceledError { task_id: Some(self.id()) })),
189            _ => panic!("the result of a task that has not ended was asked for; await it or use continue_with"),
190        }
191    }
192
193    /// Runs `continuation` when the task ends, synchronously on the thread that ends it, or at once
194    /// when it has ended (`ContinueWith` with `TaskContinuationOptions.ExecuteSynchronously`). The
195    /// id removes a continuation that has not run ([`remove_continuation`](Self::remove_continuation),
196    /// the cancellation token of `ContinueWith`); `None` when it ran at once.
197    pub fn continue_with(&self, continuation: impl FnOnce(&Task<T>) + 'static) -> Option<ContinuationId> {
198        let task = self.clone();
199        if self.is_completed() {
200            continuation(&task);
201            return None;
202        }
203        let id = self.inner.next_continuation.get();
204        self.inner.next_continuation.set(id + 1);
205        self.inner.continuations.borrow_mut().push((id, Box::new(move || continuation(&task))));
206        Some(ContinuationId(id))
207    }
208
209    /// Removes a continuation that has not run; false when it has run or was removed.
210    pub fn remove_continuation(&self, id: ContinuationId) -> bool {
211        let removed = {
212            let mut continuations = self.inner.continuations.borrow_mut();
213            let position = continuations.iter().position(|(entry, _)| *entry == id.0);
214            position.map(|position| continuations.remove(position))
215        };
216        // Dropped outside of the borrow: what the continuation holds may reach the task.
217        removed.is_some()
218    }
219
220    /// Ends the task with `outcome` unless it has ended; an error that is a [`TaskCanceledError`]
221    /// cancels it. False when it had ended.
222    pub(crate) fn try_complete(&self, outcome: Result<T, ObservableError>) -> bool {
223        if self.is_completed() {
224            return false;
225        }
226        match outcome {
227            Ok(result) => {
228                *self.inner.result.borrow_mut() = Some(result);
229                self.inner.status.set(TaskStatus::RanToCompletion);
230            }
231            Err(error) if TaskCanceledError::is_cancellation(&error) => {
232                self.inner.status.set(TaskStatus::Canceled);
233            }
234            Err(error) => {
235                *self.inner.error.borrow_mut() = Some(error);
236                self.inner.status.set(TaskStatus::Faulted);
237            }
238        }
239        self.finish();
240        true
241    }
242
243    pub(crate) fn try_cancel(&self) -> bool {
244        if self.is_completed() {
245            return false;
246        }
247        self.inner.status.set(TaskStatus::Canceled);
248        self.finish();
249        true
250    }
251
252    fn finish(&self) {
253        let continuations = std::mem::take(&mut *self.inner.continuations.borrow_mut());
254        for (_, continuation) in continuations {
255            continuation();
256        }
257        let wakers = std::mem::take(&mut *self.inner.wakers.borrow_mut());
258        for waker in wakers {
259            waker.wake();
260        }
261    }
262}
263
264impl<T: Clone + 'static> Future for Task<T> {
265    type Output = Result<T, ObservableError>;
266
267    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
268        if self.is_completed() {
269            return Poll::Ready(self.result());
270        }
271        let mut wakers = self.inner.wakers.borrow_mut();
272        if !wakers.iter().any(|waker| waker.will_wake(cx.waker())) {
273            wakers.push(cx.waker().clone());
274        }
275        Poll::Pending
276    }
277}
278
279/// The producer side of a task that code ends by hand (`TaskCompletionSource<TResult>`).
280pub struct TaskCompletionSource<T> {
281    task: Task<T>,
282}
283
284impl<T: 'static> Default for TaskCompletionSource<T> {
285    fn default() -> Self {
286        Self::new()
287    }
288}
289
290impl<T> Clone for TaskCompletionSource<T> {
291    fn clone(&self) -> Self {
292        Self { task: self.task.clone() }
293    }
294}
295
296impl<T: 'static> TaskCompletionSource<T> {
297    /// A source with a task that has not ended.
298    pub fn new() -> Self {
299        Self { task: Task::with_status(TaskStatus::WaitingForActivation) }
300    }
301
302    /// The task of the source (`Task`).
303    pub fn task(&self) -> Task<T> {
304        self.task.clone()
305    }
306
307    /// Ends the task with `result` (`SetResult`).
308    ///
309    /// # Panics
310    /// When the task has ended (`InvalidOperationException`).
311    pub fn set_result(&self, result: T) {
312        assert!(self.try_set_result(result), "An attempt was made to transition a task to a final state when it had already completed.");
313    }
314
315    /// Ends the task with `result` unless it has ended (`TrySetResult`).
316    pub fn try_set_result(&self, result: T) -> bool {
317        self.task.try_complete(Ok(result))
318    }
319
320    /// Ends the task with `error` (`SetException`); a [`TaskCanceledError`] cancels it.
321    ///
322    /// # Panics
323    /// When the task has ended (`InvalidOperationException`).
324    pub fn set_exception(&self, error: ObservableError) {
325        assert!(self.try_set_exception(error), "An attempt was made to transition a task to a final state when it had already completed.");
326    }
327
328    /// Ends the task with `error` unless it has ended (`TrySetException`).
329    pub fn try_set_exception(&self, error: ObservableError) -> bool {
330        self.task.try_complete(Err(error))
331    }
332
333    /// Cancels the task (`SetCanceled`).
334    ///
335    /// # Panics
336    /// When the task has ended (`InvalidOperationException`).
337    pub fn set_canceled(&self) {
338        assert!(self.try_set_canceled(), "An attempt was made to transition a task to a final state when it had already completed.");
339    }
340
341    /// Cancels the task unless it has ended (`TrySetCanceled`).
342    pub fn try_set_canceled(&self) -> bool {
343        self.task.try_cancel()
344    }
345}
ferroui_system/threading/tasks/task.rs

ferroui_system/threading/tasks/
task.rs
