ferroui_base/threading/
dispatcher_invoke.rs

1use std::any::Any;
2use std::fmt;
3use std::future::Future;
4use std::panic::{catch_unwind, resume_unwind, AssertUnwindSafe};
5use std::sync::Arc;
6use std::time::Duration;
7
8use super::dispatcher_operation::{box_local_callback, OperationCore, SendCallback, WaitOutcome};
9use super::dispatcher_task::{wrap_local_future, wrap_send_future};
10use super::{
11    CancellationToken, Dispatcher, DispatcherOperation, DispatcherOperationStatus, DispatcherPriority,
12    DispatcherPriorityAwaitable, DispatcherPriorityTaskAwaitable, DispatcherTask, FerroSynchronizationContext,
13    OperationCanceledError,
14};
15
16/// The error of a synchronous invoke with a timeout.
17#[derive(Clone, Copy, Debug, PartialEq, Eq)]
18pub enum DispatcherInvokeError {
19    /// The operation was aborted before it ran: its cancellation token was
20    /// canceled, or the dispatcher has shut down.
21    Canceled,
22    /// The operation did not start within the timeout and was aborted.
23    Timeout,
24}
25
26impl fmt::Display for DispatcherInvokeError {
27    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
28        match self {
29            Self::Canceled => f.write_str("The dispatcher operation was canceled"),
30            Self::Timeout => f.write_str("The dispatcher operation timed out"),
31        }
32    }
33}
34
35impl std::error::Error for DispatcherInvokeError {}
36
37/// Synchronous execution. The callbacks of [`invoke`](Dispatcher::invoke)
38/// and its variants can come from any thread and must be `Send`; the
39/// `invoke_local` variants accept any callback but can only be called on the
40/// dispatcher thread.
41impl Dispatcher {
42    /// Executes the specified callback synchronously on the thread that the
43    /// dispatcher was created on, at [`DispatcherPriority::SEND`].
44    ///
45    /// Returns `Err` when the operation was aborted before it ran (for
46    /// instance because the dispatcher has shut down).
47    ///
48    /// # Panics
49    /// Re-raises the panic of the callback.
50    pub fn invoke<R: Send + 'static>(
51        &self,
52        callback: impl FnOnce() -> R + Send + 'static,
53    ) -> Result<R, OperationCanceledError> {
54        self.invoke_with_cancellation(callback, DispatcherPriority::SEND, &CancellationToken::none())
55    }
56
57    /// Executes the specified callback synchronously with the specified
58    /// priority on the thread that the dispatcher was created on.
59    pub fn invoke_with_priority<R: Send + 'static>(
60        &self,
61        callback: impl FnOnce() -> R + Send + 'static,
62        priority: DispatcherPriority,
63    ) -> Result<R, OperationCanceledError> {
64        self.invoke_with_cancellation(callback, priority, &CancellationToken::none())
65    }
66
67    /// Executes the specified callback synchronously with the specified
68    /// priority on the thread that the dispatcher was created on.
69    ///
70    /// `cancellation_token` cancels the operation as long as it has not
71    /// started. Returns `Err` when the token is already canceled.
72    pub fn invoke_with_cancellation<R: Send + 'static>(
73        &self,
74        callback: impl FnOnce() -> R + Send + 'static,
75        priority: DispatcherPriority,
76        cancellation_token: &CancellationToken,
77    ) -> Result<R, OperationCanceledError> {
78        DispatcherPriority::validate(priority, "priority");
79
80        // Fast-Path: if on the same thread, and invoking at Send priority,
81        // and the cancellation token is not already canceled, then just
82        // call the callback directly.
83        if !cancellation_token.is_cancellation_requested() && priority == DispatcherPriority::SEND && self.check_access()
84        {
85            let _restore = FerroSynchronizationContext::ensure_with_dispatcher(self, priority);
86            return Ok(callback());
87        }
88
89        // Slow-Path: go through the queue.
90        let operation = DispatcherOperation::new_send(&self.to_arc(), priority, callback);
91        self.invoke_impl(&operation, cancellation_token, None).map_err(|_| OperationCanceledError)
92    }
93
94    /// Executes the specified callback synchronously with the specified
95    /// priority on the thread that the dispatcher was created on, giving up
96    /// when the operation has not started within `timeout`.
97    ///
98    /// The timeout only applies to the time the operation spends in the
99    /// queue: once it has started, the call waits for it to complete. When
100    /// the operation is still pending after `timeout` it is aborted and
101    /// [`DispatcherInvokeError::Timeout`] is returned.
102    ///
103    /// On the dispatcher thread the queue is pumped while waiting and the
104    /// timeout is measured with the dispatcher clock.
105    pub fn invoke_with_timeout<R: Send + 'static>(
106        &self,
107        callback: impl FnOnce() -> R + Send + 'static,
108        priority: DispatcherPriority,
109        cancellation_token: &CancellationToken,
110        timeout: Duration,
111    ) -> Result<R, DispatcherInvokeError> {
112        DispatcherPriority::validate(priority, "priority");
113
114        // Fast-Path, see invoke_with_cancellation.
115        if !cancellation_token.is_cancellation_requested() && priority == DispatcherPriority::SEND && self.check_access()
116        {
117            let _restore = FerroSynchronizationContext::ensure_with_dispatcher(self, priority);
118            return Ok(callback());
119        }
120
121        // Slow-Path: go through the queue.
122        let operation = DispatcherOperation::new_send(&self.to_arc(), priority, callback);
123        self.invoke_impl(&operation, cancellation_token, Some(timeout))
124    }
125
126    /// Like [`invoke`](Self::invoke) for callbacks that are not `Send`.
127    ///
128    /// # Panics
129    /// Panics when called from a thread other than the dispatcher thread.
130    pub fn invoke_local<R: 'static>(&self, callback: impl FnOnce() -> R + 'static) -> Result<R, OperationCanceledError> {
131        self.invoke_local_with_cancellation(callback, DispatcherPriority::SEND, &CancellationToken::none())
132    }
133
134    /// Like [`invoke_with_priority`](Self::invoke_with_priority) for
135    /// callbacks that are not `Send`.
136    ///
137    /// # Panics
138    /// Panics when called from a thread other than the dispatcher thread.
139    pub fn invoke_local_with_priority<R: 'static>(
140        &self,
141        callback: impl FnOnce() -> R + 'static,
142        priority: DispatcherPriority,
143    ) -> Result<R, OperationCanceledError> {
144        self.invoke_local_with_cancellation(callback, priority, &CancellationToken::none())
145    }
146
147    /// Like [`invoke_with_cancellation`](Self::invoke_with_cancellation) for
148    /// callbacks that are not `Send`.
149    ///
150    /// # Panics
151    /// Panics when called from a thread other than the dispatcher thread.
152    pub fn invoke_local_with_cancellation<R: 'static>(
153        &self,
154        callback: impl FnOnce() -> R + 'static,
155        priority: DispatcherPriority,
156        cancellation_token: &CancellationToken,
157    ) -> Result<R, OperationCanceledError> {
158        self.verify_access();
159        DispatcherPriority::validate(priority, "priority");
160
161        // Fast-Path, see invoke_with_cancellation.
162        if !cancellation_token.is_cancellation_requested() && priority == DispatcherPriority::SEND {
163            let _restore = FerroSynchronizationContext::ensure_with_dispatcher(self, priority);
164            return Ok(callback());
165        }
166
167        // Slow-Path: go through the queue.
168        let operation = DispatcherOperation::new_local(&self.to_arc(), priority, callback);
169        self.invoke_impl(&operation, cancellation_token, None).map_err(|_| OperationCanceledError)
170    }
171
172    /// Like [`invoke_with_timeout`](Self::invoke_with_timeout) for callbacks
173    /// that are not `Send`.
174    ///
175    /// # Panics
176    /// Panics when called from a thread other than the dispatcher thread.
177    pub fn invoke_local_with_timeout<R: 'static>(
178        &self,
179        callback: impl FnOnce() -> R + 'static,
180        priority: DispatcherPriority,
181        cancellation_token: &CancellationToken,
182        timeout: Duration,
183    ) -> Result<R, DispatcherInvokeError> {
184        self.verify_access();
185        DispatcherPriority::validate(priority, "priority");
186
187        // Fast-Path, see invoke_with_cancellation.
188        if !cancellation_token.is_cancellation_requested() && priority == DispatcherPriority::SEND {
189            let _restore = FerroSynchronizationContext::ensure_with_dispatcher(self, priority);
190            return Ok(callback());
191        }
192
193        // Slow-Path: go through the queue.
194        let operation = DispatcherOperation::new_local(&self.to_arc(), priority, callback);
195        self.invoke_impl(&operation, cancellation_token, Some(timeout))
196    }
197
198    fn invoke_impl<R: 'static>(
199        &self,
200        operation: &DispatcherOperation<R>,
201        cancellation_token: &CancellationToken,
202        timeout: Option<Duration>,
203    ) -> Result<R, DispatcherInvokeError> {
204        if cancellation_token.is_cancellation_requested() {
205            operation.core().set_status(DispatcherOperationStatus::Aborted);
206            operation.core().call_abort_callbacks();
207            return Err(DispatcherInvokeError::Canceled);
208        }
209
210        debug_assert!(operation.priority() != DispatcherPriority::SEND || !self.check_access()); // should be handled by caller
211
212        // This operation must be queued since it was invoked either to
213        // another thread, or at a priority other than Send.
214        self.invoke_async_impl(operation.core(), cancellation_token);
215
216        // We have already registered with the cancellation token to abort
217        // the operation when it is canceled, and the wait below aborts it
218        // when the timeout expires while it is still pending.  If the
219        // operation has already started when the timeout expires, we still
220        // wait for it to complete.  This is different than simply waiting on
221        // the operation with a timeout because we are the ones queuing the
222        // dispatcher operation, not the caller.  We can't leave the operation
223        // in a state that it might execute if we return that it did not
224        // invoke.
225        match operation.core().wait_with_limit(timeout, true) {
226            // The operation was aborted because of the timeout.
227            WaitOutcome::TimedOut => Err(DispatcherInvokeError::Timeout),
228            // Completed, or canceled for some other reason.
229            WaitOutcome::Finished => operation.take_result().map_err(|_| DispatcherInvokeError::Canceled),
230        }
231    }
232
233    /// Runs `action` on the dispatcher thread and waits for it. A panic of
234    /// the action goes through the unhandled-exception events of the
235    /// dispatcher and is re-raised on the dispatcher thread when it is not
236    /// handled there.
237    ///
238    /// `None` stands for [`DispatcherPriority::SEND`].
239    pub(crate) fn send(&self, action: impl FnOnce() + Send + 'static, priority: Option<DispatcherPriority>) {
240        let priority = priority.unwrap_or(DispatcherPriority::SEND);
241
242        if priority == DispatcherPriority::SEND && self.check_access() {
243            self.send_inline(action, priority);
244        } else {
245            let callback: SendCallback = Box::new(move || {
246                action();
247                Box::new(()) as Box<dyn Any + Send>
248            });
249            let operation = OperationCore::new_send(&self.to_arc(), priority, true, callback);
250            self.invoke_async_impl(&operation, &CancellationToken::none());
251            operation.wait();
252        }
253    }
254
255    /// Like [`send`](Self::send) for actions that are not `Send`.
256    ///
257    /// # Panics
258    /// Panics when called from a thread other than the dispatcher thread.
259    pub(crate) fn send_local(&self, action: impl FnOnce() + 'static, priority: Option<DispatcherPriority>) {
260        self.verify_access();
261        let priority = priority.unwrap_or(DispatcherPriority::SEND);
262
263        if priority == DispatcherPriority::SEND {
264            self.send_inline(action, priority);
265        } else {
266            let operation = OperationCore::new_local(&self.to_arc(), priority, true, box_local_callback(action));
267            self.invoke_async_impl(&operation, &CancellationToken::none());
268            operation.wait();
269        }
270    }
271
272    fn send_inline(&self, action: impl FnOnce(), priority: DispatcherPriority) {
273        let result = catch_unwind(AssertUnwindSafe(|| {
274            let _restore = FerroSynchronizationContext::ensure_with_dispatcher(self, priority);
275            action();
276        }));
277        if let Err(payload) = result {
278            if !self.try_catch_when(&payload) {
279                self.forget_exception_if_outermost(&payload);
280                resume_unwind(payload);
281            }
282        }
283    }
284}
285
286/// Asynchronous execution; see [`Dispatcher`] for the `Send`/local split.
287impl Dispatcher {
288    /// Executes the specified callback asynchronously on the thread that the
289    /// dispatcher was created on, at [`DispatcherPriority::DEFAULT`].
290    ///
291    /// The returned operation can be awaited, waited for, aborted and
292    /// re-prioritized.
293    pub fn invoke_async<R: Send + 'static>(
294        &self,
295        callback: impl FnOnce() -> R + Send + 'static,
296    ) -> DispatcherOperation<R> {
297        self.invoke_async_with_cancellation(callback, DispatcherPriority::DEFAULT, &CancellationToken::none())
298    }
299
300    /// Executes the specified callback asynchronously with the specified
301    /// priority on the thread that the dispatcher was created on.
302    pub fn invoke_async_with_priority<R: Send + 'static>(
303        &self,
304        callback: impl FnOnce() -> R + Send + 'static,
305        priority: DispatcherPriority,
306    ) -> DispatcherOperation<R> {
307        self.invoke_async_with_cancellation(callback, priority, &CancellationToken::none())
308    }
309
310    /// Executes the specified callback asynchronously with the specified
311    /// priority on the thread that the dispatcher was created on.
312    ///
313    /// `cancellation_token` aborts the operation as long as it has not
314    /// started.
315    pub fn invoke_async_with_cancellation<R: Send + 'static>(
316        &self,
317        callback: impl FnOnce() -> R + Send + 'static,
318        priority: DispatcherPriority,
319        cancellation_token: &CancellationToken,
320    ) -> DispatcherOperation<R> {
321        DispatcherPriority::validate(priority, "priority");
322
323        let operation = DispatcherOperation::new_send(&self.to_arc(), priority, callback);
324        self.invoke_async_impl(operation.core(), cancellation_token);
325
326        operation
327    }
328
329    /// Like [`invoke_async`](Self::invoke_async) for callbacks that are not
330    /// `Send`.
331    ///
332    /// # Panics
333    /// Panics when called from a thread other than the dispatcher thread.
334    pub fn invoke_async_local<R: 'static>(&self, callback: impl FnOnce() -> R + 'static) -> DispatcherOperation<R> {
335        self.invoke_async_local_with_cancellation(callback, DispatcherPriority::DEFAULT, &CancellationToken::none())
336    }
337
338    /// Like [`invoke_async_with_priority`](Self::invoke_async_with_priority)
339    /// for callbacks that are not `Send`.
340    ///
341    /// # Panics
342    /// Panics when called from a thread other than the dispatcher thread.
343    pub fn invoke_async_local_with_priority<R: 'static>(
344        &self,
345        callback: impl FnOnce() -> R + 'static,
346        priority: DispatcherPriority,
347    ) -> DispatcherOperation<R> {
348        self.invoke_async_local_with_cancellation(callback, priority, &CancellationToken::none())
349    }
350
351    /// Like
352    /// [`invoke_async_with_cancellation`](Self::invoke_async_with_cancellation)
353    /// for callbacks that are not `Send`.
354    ///
355    /// # Panics
356    /// Panics when called from a thread other than the dispatcher thread.
357    pub fn invoke_async_local_with_cancellation<R: 'static>(
358        &self,
359        callback: impl FnOnce() -> R + 'static,
360        priority: DispatcherPriority,
361        cancellation_token: &CancellationToken,
362    ) -> DispatcherOperation<R> {
363        self.verify_access();
364        DispatcherPriority::validate(priority, "priority");
365
366        let operation = DispatcherOperation::new_local(&self.to_arc(), priority, callback);
367        self.invoke_async_impl(operation.core(), cancellation_token);
368
369        operation
370    }
371
372    pub(crate) fn invoke_async_impl(&self, operation: &Arc<OperationCore>, cancellation_token: &CancellationToken) {
373        let mut succeeded = false;
374
375        // Could be a non-dispatcher thread, lock to read
376        {
377            let mut st = self.lock();
378            if !cancellation_token.is_cancellation_requested() && !st.has_shutdown_finished {
379                // Add the operation to the work queue
380                st.queue.enqueue(operation.priority(), operation.clone());
381
382                // Make sure we will wake up to process this operation.
383                succeeded = self.request_processing_locked(&mut st);
384
385                if !succeeded {
386                    // Dequeue the item since we failed to request
387                    // processing for it.  Note we will mark it aborted
388                    // below.
389                    st.queue.remove_item(operation);
390                }
391            }
392        }
393
394        if succeeded {
395            // We have enqueued the operation.  Register a callback
396            // with the cancellation token to abort the operation
397            // when cancellation is requested.
398            if cancellation_token.can_be_canceled() {
399                let to_abort = operation.clone();
400                let registration = Arc::new(cancellation_token.register(move || {
401                    to_abort.abort();
402                }));
403
404                // Revoke the cancellation when the operation is done.
405                let revoke = registration.clone();
406                operation.add_aborted(Arc::new(move || revoke.dispose()));
407                operation.add_completed(Arc::new(move || registration.dispose()));
408            }
409        } else {
410            // We failed to enqueue the operation, and the caller that
411            // created the operation does not expose it before we return,
412            // so it is safe to modify the operation outside of the lock.
413            // Just mark the operation as aborted, which we can safely
414            // return to the user.
415            operation.set_status(DispatcherOperationStatus::Aborted);
416            operation.call_abort_callbacks();
417        }
418    }
419
420    /// Executes the specified callback on the dispatcher thread and then
421    /// runs the future it returns there, at
422    /// [`DispatcherPriority::DEFAULT`].
423    ///
424    /// The returned task completes when the future has completed; see
425    /// [`invoke_async_task_with_priority`](Self::invoke_async_task_with_priority).
426    pub fn invoke_async_task<F, Fut>(&self, callback: F) -> DispatcherTask<Fut::Output>
427    where
428        F: FnOnce() -> Fut + Send + 'static,
429        Fut: Future + 'static,
430        Fut::Output: Send + 'static,
431    {
432        self.invoke_async_task_with_priority(callback, DispatcherPriority::DEFAULT)
433    }
434
435    /// Executes the specified callback with the specified priority on the
436    /// dispatcher thread and then runs the future it returns there.
437    ///
438    /// The callback can come from any thread; the future it creates never
439    /// leaves the dispatcher thread and does not have to be `Send`. The
440    /// future is polled for the first time right after the callback, inside
441    /// the same job, and afterwards from jobs of the same priority.
442    ///
443    /// The returned task completes when the future has completed. It is
444    /// canceled when the operation is aborted before the callback ran or the
445    /// dispatcher shuts down before the future is done, and it fails with
446    /// the panic of the callback or the future.
447    pub fn invoke_async_task_with_priority<F, Fut>(
448        &self,
449        callback: F,
450        priority: DispatcherPriority,
451    ) -> DispatcherTask<Fut::Output>
452    where
453        F: FnOnce() -> Fut + Send + 'static,
454        Fut: Future + 'static,
455        Fut::Output: Send + 'static,
456    {
457        DispatcherPriority::validate(priority, "priority");
458        let dispatcher = self.to_arc();
459        let (task, completion, wrap) = wrap_send_future::<Fut>(&dispatcher);
460        let target = dispatcher.clone();
461        // When the operation is aborted its callback is dropped, which
462        // cancels the task.
463        drop(self.invoke_async_with_priority(
464            move || match catch_unwind(AssertUnwindSafe(callback)) {
465                Ok(future) => target.start_task(priority, wrap(future), completion, true),
466                Err(payload) => completion.fail(payload),
467            },
468            priority,
469        ));
470        task
471    }
472
473    /// Like [`invoke_async_task`](Self::invoke_async_task) for callbacks and
474    /// results that are not `Send`.
475    ///
476    /// # Panics
477    /// Panics when called from a thread other than the dispatcher thread.
478    pub fn invoke_async_task_local<F, Fut>(&self, callback: F) -> DispatcherTask<Fut::Output>
479    where
480        F: FnOnce() -> Fut + 'static,
481        Fut: Future + 'static,
482        Fut::Output: 'static,
483    {
484        self.invoke_async_task_local_with_priority(callback, DispatcherPriority::DEFAULT)
485    }
486
487    /// Like
488    /// [`invoke_async_task_with_priority`](Self::invoke_async_task_with_priority)
489    /// for callbacks and results that are not `Send`.
490    ///
491    /// # Panics
492    /// Panics when called from a thread other than the dispatcher thread.
493    pub fn invoke_async_task_local_with_priority<F, Fut>(
494        &self,
495        callback: F,
496        priority: DispatcherPriority,
497    ) -> DispatcherTask<Fut::Output>
498    where
499        F: FnOnce() -> Fut + 'static,
500        Fut: Future + 'static,
501        Fut::Output: 'static,
502    {
503        self.verify_access();
504        DispatcherPriority::validate(priority, "priority");
505        let dispatcher = self.to_arc();
506        let (task, completion, wrap) = wrap_local_future::<Fut>(&dispatcher);
507        let target = dispatcher.clone();
508        drop(self.invoke_async_local_with_priority(
509            move || match catch_unwind(AssertUnwindSafe(callback)) {
510                Ok(future) => target.start_task(priority, wrap(future), completion, true),
511                Err(payload) => completion.fail(payload),
512            },
513            priority,
514        ));
515        task
516    }
517
518    /// Posts an action that will be invoked on the dispatcher thread.
519    ///
520    /// Can be called from any thread. A panic raised by the action goes
521    /// through the unhandled-exception events of the dispatcher.
522    pub fn post(&self, action: impl FnOnce() + Send + 'static, priority: DispatcherPriority) {
523        let callback: SendCallback = Box::new(move || {
524            action();
525            Box::new(()) as Box<dyn Any + Send>
526        });
527        let operation = OperationCore::new_send(&self.to_arc(), priority, true, callback);
528        self.invoke_async_impl(&operation, &CancellationToken::none());
529    }
530
531    /// Posts an action that is not `Send` to be invoked on the dispatcher
532    /// thread.
533    ///
534    /// # Panics
535    /// Panics when called from a thread other than the dispatcher thread.
536    pub fn post_local(&self, action: impl FnOnce() + 'static, priority: DispatcherPriority) {
537        let operation = OperationCore::new_local(&self.to_arc(), priority, true, box_local_callback(action));
538        self.invoke_async_impl(&operation, &CancellationToken::none());
539    }
540
541    /// Returns a future that awaits `task` and then resumes the awaiting
542    /// code from a job of this dispatcher with the given priority, yielding
543    /// the output of `task`.
544    pub fn await_with_priority<F: Future>(
545        &self,
546        task: F,
547        priority: DispatcherPriority,
548    ) -> DispatcherPriorityTaskAwaitable<F> {
549        DispatcherPriority::validate(priority, "priority");
550        DispatcherPriorityTaskAwaitable::new(self.to_arc(), task, priority)
551    }
552
553    /// Returns a future that resumes on this dispatcher's thread at
554    /// [`DispatcherPriority::BACKGROUND`] when awaited.
555    pub fn resume(&self) -> DispatcherPriorityAwaitable {
556        self.resume_with_priority(DispatcherPriority::BACKGROUND)
557    }
558
559    /// Returns a future that resumes on this dispatcher's thread at the
560    /// given priority when awaited.
561    pub fn resume_with_priority(&self, priority: DispatcherPriority) -> DispatcherPriorityAwaitable {
562        DispatcherPriority::validate(priority, "priority");
563        DispatcherPriorityAwaitable::new(self.to_arc(), priority)
564    }
565
566    /// Returns a future that yields to the current dispatcher, letting it
567    /// process other work, and resumes at [`DispatcherPriority::BACKGROUND`].
568    pub fn yield_now() -> DispatcherPriorityAwaitable {
569        Self::yield_with_priority(DispatcherPriority::BACKGROUND)
570    }
571
572    /// Returns a future that yields to the current dispatcher and resumes at
573    /// the given priority.
574    pub fn yield_with_priority(priority: DispatcherPriority) -> DispatcherPriorityAwaitable {
575        Self::current_dispatcher().resume_with_priority(priority)
576    }
577}
ferroui_base/threading/dispatcher_invoke.rs

ferroui_base/threading/
dispatcher_invoke.rs
