ferroui_base/threading/
dispatcher.rs

1use std::any::Any;
2use std::cell::{Cell, RefCell};
3use std::collections::HashMap;
4use std::rc::Rc;
5use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
6use std::sync::{Arc, Mutex, MutexGuard, PoisonError, Weak};
7use std::thread::{self, ThreadId};
8
9use super::dispatcher_operation::LocalCallback;
10use super::dispatcher_priority_queue::DispatcherPriorityQueue;
11use super::dispatcher_task::LocalTask;
12use super::i_dispatcher_impl::{DetachedDispatcherImpl, DetachedDispatcherSignal};
13use super::{
14    DispatcherFrame, DispatcherPriority, DispatcherTimer, DispatcherUnhandledExceptionEventArgs,
15    DispatcherUnhandledExceptionFilterEventArgs, FerroSynchronizationContext, IDispatcher, IDispatcherImpl,
16    IDispatcherSignal,
17};
18use crate::utilities::HandlerList;
19
20/// Handler of [`Dispatcher::unhandled_exception`].
21pub type DispatcherUnhandledExceptionEventHandler = dyn for<'a> Fn(&DispatcherUnhandledExceptionEventArgs<'a>);
22
23/// Handler of [`Dispatcher::unhandled_exception_filter`].
24pub type DispatcherUnhandledExceptionFilterEventHandler =
25    dyn for<'a> Fn(&DispatcherUnhandledExceptionFilterEventArgs<'a>);
26
27/// The part of a dispatcher that never leaves its thread: everything that
28/// holds `Rc`-based or otherwise non-`Send` values. It lives in thread-local
29/// storage, so it needs no locking and cannot be reached from other threads.
30pub(crate) struct DispatcherLocal {
31    pub(crate) dispatcher_id: u64,
32    /// The platform implementation; it belongs to this thread.
33    pub(crate) impl_: RefCell<Rc<dyn IDispatcherImpl>>,
34    attached: Cell<bool>,
35    timer_token: Cell<u64>,
36    signaled_token: Cell<u64>,
37    background_token: Cell<Option<u64>>,
38    /// The per-priority synchronization contexts, created on demand.
39    pub(crate) priority_contexts: RefCell<Vec<Option<Arc<FerroSynchronizationContext>>>>,
40    /// The futures running on this dispatcher.
41    pub(crate) tasks: RefCell<HashMap<u64, LocalTask>>,
42    /// Callbacks of operations queued with the `*_local` methods.
43    pub(crate) jobs: RefCell<HashMap<u64, LocalCallback>>,
44    /// Results of completed local operations that have not been taken yet.
45    pub(crate) results: RefCell<HashMap<u64, Box<dyn Any>>>,
46    pub(crate) timers: RefCell<Vec<Rc<DispatcherTimer>>>,
47    pub(crate) timers_version: Cell<i64>,
48    pub(crate) shutdown_started: HandlerList<dyn Fn(&Arc<Dispatcher>)>,
49    pub(crate) shutdown_finished: HandlerList<dyn Fn(&Arc<Dispatcher>)>,
50    pub(crate) unhandled_exception: HandlerList<DispatcherUnhandledExceptionEventHandler>,
51    pub(crate) unhandled_exception_filter: HandlerList<DispatcherUnhandledExceptionFilterEventHandler>,
52    /// Addresses of panic payloads this dispatcher has already filtered and
53    /// that are still propagating through its frames.
54    pub(crate) seen_exceptions: RefCell<Vec<usize>>,
55    /// Number of operations currently executing (nested frames).
56    pub(crate) executing_depth: Cell<u32>,
57}
58
59impl DispatcherLocal {
60    fn new(dispatcher_id: u64) -> Self {
61        Self {
62            dispatcher_id,
63            impl_: RefCell::new(Rc::new(DetachedDispatcherImpl::new())),
64            attached: Cell::new(false),
65            timer_token: Cell::new(0),
66            signaled_token: Cell::new(0),
67            background_token: Cell::new(None),
68            priority_contexts: RefCell::new(Vec::new()),
69            tasks: RefCell::new(HashMap::new()),
70            jobs: RefCell::new(HashMap::new()),
71            results: RefCell::new(HashMap::new()),
72            timers: RefCell::new(Vec::new()),
73            timers_version: Cell::new(0),
74            shutdown_started: HandlerList::new(),
75            shutdown_finished: HandlerList::new(),
76            unhandled_exception: HandlerList::new(),
77            unhandled_exception_filter: HandlerList::new(),
78            seen_exceptions: RefCell::new(Vec::new()),
79            executing_depth: Cell::new(0),
80        }
81    }
82}
83
84/// Everything guarded by the dispatcher's instance lock.
85pub(crate) struct DispatcherState {
86    /// Wakes the dispatcher thread up; the only part of the platform
87    /// implementation that other threads touch.
88    pub(crate) signal: Arc<dyn IDispatcherSignal>,
89    pub(crate) supports_run_loops: bool,
90    pub(crate) initialized: bool,
91
92    // Queue
93    pub(crate) queue: DispatcherPriorityQueue,
94    pub(crate) signaled: bool,
95    pub(crate) explicit_background_processing_requested: bool,
96    pub(crate) maximum_input_starvation_time: i64,
97    /// Local operations that were discarded on another thread; their
98    /// thread-local storage is released by the dispatcher thread.
99    pub(crate) dead_local_operations: Vec<u64>,
100
101    // Timers
102    pub(crate) due_time_found: bool,
103    pub(crate) due_time_in_ms: i64,
104    pub(crate) due_time_for_timers: Option<i64>,
105    pub(crate) due_time_for_background_processing: Option<i64>,
106    pub(crate) os_timer_set_to: Option<i64>,
107
108    // Main loop
109    pub(crate) disabled_processing_count: i32,
110    pub(crate) has_shutdown_finished: bool,
111    pub(crate) starting_shutdown: bool,
112    pub(crate) frames: Vec<Arc<DispatcherFrame>>,
113}
114
115static NEXT_DISPATCHER_ID: AtomicU64 = AtomicU64::new(1);
116
117/// Provides services for managing work items on a thread.
118///
119/// # Threading
120///
121/// The object model is single-threaded and `Rc`-based; the dispatcher is the
122/// one entry point that can be used from other threads. A dispatcher is
123/// therefore handed out as `Arc<Dispatcher>` and is `Send + Sync`, and its
124/// API is split in two:
125///
126/// * [`post`](Self::post), [`invoke`](Self::invoke),
127///   [`invoke_async`](Self::invoke_async) and their variants can be called
128///   from **any thread**. Their callbacks (and results) must be `Send`.
129/// * [`post_local`](Self::post_local), [`invoke_local`](Self::invoke_local),
130///   [`invoke_async_local`](Self::invoke_async_local) and their variants
131///   accept callbacks that are not `Send` (they may capture `Rc`s, `Ref`s,
132///   ...). They can only be called **on the dispatcher thread**, which is
133///   checked at run time: calling them from another thread panics. The
134///   callback is kept in thread-local storage and never leaves the thread.
135///
136/// Timers ([`DispatcherTimer`]), the shutdown and unhandled-exception events
137/// and everything else that takes non-`Send` handlers is likewise bound to
138/// the dispatcher thread. So is the platform implementation
139/// ([`IDispatcherImpl`]): other threads only ever use its
140/// [`IDispatcherSignal`].
141///
142/// # Async
143///
144/// The dispatcher is also an executor for futures, which takes the place of
145/// the synchronization context of the reference implementation:
146/// [`invoke_async_task`](Self::invoke_async_task) and
147/// [`DispatcherTaskScheduler`](super::DispatcherTaskScheduler) run a future
148/// on the dispatcher thread, polling it from jobs queued at a fixed priority;
149/// [`resume`](Self::resume), [`yield_now`](Self::yield_now) and
150/// [`await_with_priority`](Self::await_with_priority) change the priority at
151/// which an `async` block continues. No async runtime is involved.
152///
153/// # Panics in callbacks
154///
155/// What the reference implementation does with exceptions is done here with
156/// panics: a panic in a posted callback goes through
157/// [`unhandled_exception_filter`](Self::unhandled_exception_filter) and
158/// [`unhandled_exception`](Self::unhandled_exception) and is re-raised on the
159/// dispatcher thread unless a handler marks it as handled; a panic in an
160/// `invoke`/`invoke_async` callback is captured and re-raised to whoever
161/// waits for the operation. On targets that abort on panic none of this
162/// applies.
163pub struct Dispatcher {
164    pub(crate) id: u64,
165    thread: ThreadId,
166    pub(crate) weak_self: Weak<Dispatcher>,
167    state: Mutex<DispatcherState>,
168    pub(crate) exit_all_frames_requested: AtomicBool,
169    pub(crate) has_shutdown_started: AtomicBool,
170}
171
172impl Dispatcher {
173    /// Creates the dispatcher of the current thread.
174    ///
175    /// # Panics
176    /// Panics when the thread already has a dispatcher or when `impl_`
177    /// belongs to a different thread.
178    pub(crate) fn new(impl_: Option<Rc<dyn IDispatcherImpl>>) -> Arc<Dispatcher> {
179        #[cfg(debug_assertions)]
180        {
181            use crate::LocatorExtensions;
182            let locator = crate::FerroLocator::current();
183            if locator.get_service::<dyn IDispatcherImpl>().is_some()
184                || locator.get_service::<dyn crate::platform::IPlatformThreadingInterface>().is_some()
185            {
186                panic!("Registering IDispatcherImpl or IPlatformThreadingInterface via locator is no longer valid");
187            }
188        }
189
190        if let Some(impl_) = &impl_ {
191            if !impl_.current_thread_is_loop_thread() {
192                panic!("IDispatcherImpl belongs to a different thread");
193            }
194        }
195
196        let id = NEXT_DISPATCHER_ID.fetch_add(1, Ordering::Relaxed);
197        let dispatcher = Arc::new_cyclic(|weak_self| Dispatcher {
198            id,
199            thread: current_thread_id(),
200            weak_self: weak_self.clone(),
201            state: Mutex::new(DispatcherState {
202                // Set by replace_implementation
203                signal: Arc::new(DetachedDispatcherSignal),
204                supports_run_loops: false,
205                initialized: false,
206                queue: DispatcherPriorityQueue::new(),
207                signaled: false,
208                explicit_background_processing_requested: false,
209                maximum_input_starvation_time: Self::MAXIMUM_INPUT_STARVATION_TIME_IN_FALLBACK_MODE,
210                dead_local_operations: Vec::new(),
211                due_time_found: false,
212                due_time_in_ms: 0,
213                due_time_for_timers: None,
214                due_time_for_background_processing: None,
215                os_timer_set_to: None,
216                disabled_processing_count: 0,
217                has_shutdown_finished: false,
218                starting_shutdown: false,
219                frames: Vec::new(),
220            }),
221            exit_all_frames_requested: AtomicBool::new(false),
222            has_shutdown_started: AtomicBool::new(false),
223        });
224
225        // The first created dispatcher becomes "UI thread one"
226        Self::register_current_thread_dispatcher(&dispatcher, Rc::new(DispatcherLocal::new(id)));
227
228        dispatcher.replace_implementation(impl_);
229        dispatcher
230    }
231
232    /// Whether the platform implementation can run nested event loops. When
233    /// this is `false`, [`push_frame`](Self::push_frame) and
234    /// [`main_loop`](Self::main_loop) are not available.
235    pub fn supports_run_loops(&self) -> bool {
236        self.lock().supports_run_loops
237    }
238
239    /// Checks that the current thread is the dispatcher thread.
240    #[inline]
241    pub fn check_access(&self) -> bool {
242        current_thread_id() == self.thread
243    }
244
245    /// Checks that the current thread is the dispatcher thread and panics if
246    /// not.
247    #[inline]
248    #[track_caller]
249    pub fn verify_access(&self) {
250        if !self.check_access() {
251            Self::throw_verify_access();
252        }
253    }
254
255    #[cold]
256    #[inline(never)]
257    #[track_caller]
258    fn throw_verify_access() -> ! {
259        panic!("The calling thread cannot access this object because a different thread owns it.");
260    }
261
262    /// The thread this dispatcher belongs to.
263    pub fn thread(&self) -> ThreadId {
264        self.thread
265    }
266
267    /// The platform implementation currently in use.
268    ///
269    /// # Panics
270    /// Panics when called from a thread other than the dispatcher thread:
271    /// the implementation belongs to that thread.
272    pub fn platform_impl(&self) -> Rc<dyn IDispatcherImpl> {
273        self.verify_access();
274        self.local_impl().unwrap_or_else(|| Rc::new(DetachedDispatcherImpl::new()))
275    }
276
277    /// The platform implementation, when called on the dispatcher thread
278    /// and the dispatcher has not been reset.
279    pub(crate) fn local_impl(&self) -> Option<Rc<dyn IDispatcherImpl>> {
280        self.try_local().map(|local| local.impl_.borrow().clone())
281    }
282
283    /// Takes the instance lock.
284    pub(crate) fn lock(&self) -> MutexGuard<'_, DispatcherState> {
285        self.state.lock().unwrap_or_else(PoisonError::into_inner)
286    }
287
288    /// A strong handle to this dispatcher.
289    pub(crate) fn to_arc(&self) -> Arc<Dispatcher> {
290        self.weak_self.upgrade().expect("the dispatcher is alive while it is borrowed")
291    }
292
293    /// The thread-local part of this dispatcher, when called on its thread
294    /// and it has not been reset.
295    pub(crate) fn try_local(&self) -> Option<Rc<DispatcherLocal>> {
296        if !self.check_access() {
297            return None;
298        }
299        Self::current_thread_local().filter(|local| local.dispatcher_id == self.id)
300    }
301
302    /// Releases the thread-local storage of a local operation that will not
303    /// run or whose result will not be read.
304    pub(crate) fn discard_local_operation(&self, operation_id: u64) {
305        if self.check_access() {
306            if let Some(local) = self.try_local() {
307                let job = local.jobs.borrow_mut().remove(&operation_id);
308                let result = local.results.borrow_mut().remove(&operation_id);
309                drop(job);
310                drop(result);
311            }
312        } else {
313            self.lock().dead_local_operations.push(operation_id);
314        }
315    }
316
317    pub(crate) fn replace_implementation(&self, impl_: Option<Rc<dyn IDispatcherImpl>>) {
318        if let Some(impl_) = &impl_ {
319            if !impl_.current_thread_is_loop_thread() {
320                panic!("IDispatcherImpl belongs to a different thread");
321            }
322        }
323        self.verify_access();
324        let Some(local) = self.try_local() else {
325            return;
326        };
327
328        let mut time_shift = None;
329        {
330            let mut st = self.lock();
331            let mut old_now = None;
332
333            if local.attached.get() {
334                // Not attached in the constructor
335                old_now = Some(local.impl_.borrow().now());
336                Self::detach_implementation(&local);
337            }
338
339            let impl_ = match impl_ {
340                Some(impl_) => {
341                    st.initialized = true;
342                    impl_
343                }
344                None => default_dispatcher_impl(),
345            };
346            *local.impl_.borrow_mut() = impl_.clone();
347            st.signal = impl_.signal_handle();
348            st.supports_run_loops = impl_.as_controlled().is_some();
349
350            let weak = self.weak_self.clone();
351            local.timer_token.set(impl_.timer().add(Rc::new(move |()| {
352                if let Some(dispatcher) = weak.upgrade() {
353                    dispatcher.on_os_timer();
354                }
355            })));
356            let weak = self.weak_self.clone();
357            local.signaled_token.set(impl_.signaled().add(Rc::new(move |()| {
358                if let Some(dispatcher) = weak.upgrade() {
359                    dispatcher.signaled();
360                }
361            })));
362            local.attached.set(true);
363
364            let background = impl_.as_explicit_background_processing();
365            st.maximum_input_starvation_time = if background.is_none() {
366                Self::MAXIMUM_INPUT_STARVATION_TIME_IN_FALLBACK_MODE
367            } else {
368                Self::MAXIMUM_INPUT_STARVATION_TIME_IN_EXPLICIT_PROCESSING_EXPLICIT_MODE
369            };
370            if let Some(background) = background {
371                let weak = self.weak_self.clone();
372                local.background_token.set(Some(background.ready_for_background_processing().add(Rc::new(
373                    move |()| {
374                        if let Some(dispatcher) = weak.upgrade() {
375                            dispatcher.on_ready_for_explicit_background_processing();
376                        }
377                    },
378                ))));
379            }
380
381            // All time comes from the implementation, so due times computed
382            // with the previous implementation's clock are moved to the new
383            // clock.
384            if let Some(old_now) = old_now {
385                let shift = impl_.now().wrapping_sub(old_now);
386                if shift != 0 {
387                    st.due_time_in_ms = st.due_time_in_ms.wrapping_add(shift);
388                    st.due_time_for_timers = st.due_time_for_timers.map(|due| due.wrapping_add(shift));
389                    st.due_time_for_background_processing =
390                        st.due_time_for_background_processing.map(|due| due.wrapping_add(shift));
391                    time_shift = Some(shift);
392                }
393            }
394
395            if st.signaled {
396                impl_.signal();
397            }
398            if st.explicit_background_processing_requested {
399                match background {
400                    Some(background) => background.request_background_processing(),
401                    None => {
402                        // The request cannot be repeated to an implementation
403                        // without explicit background processing; fall back to
404                        // the timer so that queued background jobs still run.
405                        st.explicit_background_processing_requested = false;
406                        self.request_background_processing_locked(&mut st);
407                    }
408                }
409            }
410
411            st.os_timer_set_to = None;
412            self.update_os_timer_locked(&mut st);
413        }
414
415        if let Some(shift) = time_shift {
416            for timer in local.timers.borrow().iter() {
417                timer.shift_due_time(shift);
418            }
419        }
420    }
421
422    /// Unsubscribes from the current implementation's events.
423    pub(crate) fn detach_implementation(local: &DispatcherLocal) {
424        if !local.attached.get() {
425            return;
426        }
427        let impl_ = local.impl_.borrow().clone();
428        impl_.timer().remove(local.timer_token.get());
429        impl_.signaled().remove(local.signaled_token.get());
430        if let Some(token) = local.background_token.take() {
431            if let Some(background) = impl_.as_explicit_background_processing() {
432                background.ready_for_background_processing().remove(token);
433            }
434        }
435        local.attached.set(false);
436    }
437
438    /// Replaces the implementation with one that ignores everything, without
439    /// touching the queue. Used while tearing a dispatcher down. `local` is
440    /// absent when the thread-local part is already gone.
441    pub(crate) fn detach_from_platform(st: &mut DispatcherState, local: Option<&DispatcherLocal>) {
442        if let Some(local) = local {
443            Self::detach_implementation(local);
444            let previous = local.impl_.replace(Rc::new(DetachedDispatcherImpl::new()));
445            drop(previous);
446        }
447        st.signal = Arc::new(DetachedDispatcherSignal);
448        st.supports_run_loops = false;
449    }
450}
451
452impl IDispatcher for Dispatcher {
453    fn check_access(&self) -> bool {
454        Dispatcher::check_access(self)
455    }
456
457    fn verify_access(&self) {
458        Dispatcher::verify_access(self)
459    }
460
461    fn post(&self, action: Box<dyn FnOnce() + Send>, priority: DispatcherPriority) {
462        Dispatcher::post(self, action, priority)
463    }
464}
465
466impl std::fmt::Debug for Dispatcher {
467    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
468        f.debug_struct("Dispatcher").field("id", &self.id).field("thread", &self.thread).finish()
469    }
470}
471
472/// The implementation used until the platform installs its own one.
473fn default_dispatcher_impl() -> Rc<dyn IDispatcherImpl> {
474    #[cfg(not(target_family = "wasm"))]
475    {
476        Rc::new(crate::platform::ManagedDispatcherImpl::new(None))
477    }
478    #[cfg(target_family = "wasm")]
479    {
480        // No clock and no blocking primitives: queue work until the platform
481        // implementation is installed, which then receives the pending
482        // signal and timer requests.
483        Rc::new(DetachedDispatcherImpl::new())
484    }
485}
486
487crate::ferro_thread_local! {
488    static THREAD_ID: ThreadId = thread::current().id();
489}
490
491/// The id of the calling thread, without the cost of `thread::current()`.
492#[inline]
493pub(crate) fn current_thread_id() -> ThreadId {
494    THREAD_ID.try_with(|id| *id).unwrap_or_else(|_| thread::current().id())
495}
ferroui_base/threading/dispatcher.rs

ferroui_base/threading/
dispatcher.rs
