ferroui_base/threading/
ferro_synchronization_context.rs

1use std::cell::{Cell, RefCell};
2use std::future::Future;
3use std::sync::atomic::{AtomicBool, Ordering};
4use std::sync::Arc;
5
6use super::dispatcher_task::{wrap_local_future, wrap_send_future};
7use super::{Dispatcher, DispatcherPriority, DispatcherTask};
8use crate::reactive::IDisposable;
9
10static AUTO_INSTALL: AtomicBool = AtomicBool::new(true);
11
12crate::ferro_thread_local! {
13    static CURRENT: RefCell<Option<Arc<FerroSynchronizationContext>>> = const { RefCell::new(None) };
14}
15
16/// A dispatcher together with the priority at which work handed to it is
17/// queued.
18///
19/// The *current* context of a thread tells asynchronous code where and at
20/// which priority to continue. While the dispatcher executes a job, the
21/// current context is the dispatcher's context for the priority of that job;
22/// futures started without an explicit priority (see
23/// [`Dispatcher::to_task_scheduler`]) therefore continue at the priority of
24/// the job that started them.
25///
26/// What the reference implementation additionally does in its context -
27/// replacing the blocking wait of the runtime with one that does not pump
28/// messages - has no counterpart: blocking primitives never run a message
29/// pump here.
30pub struct FerroSynchronizationContext {
31    dispatcher: Arc<Dispatcher>,
32    priority: DispatcherPriority,
33}
34
35impl FerroSynchronizationContext {
36    /// Creates a context for the current thread's dispatcher at
37    /// [`DispatcherPriority::DEFAULT`].
38    #[allow(clippy::new_ret_no_self)]
39    pub fn new() -> Arc<Self> {
40        Self::with_dispatcher(&Dispatcher::current_dispatcher(), DispatcherPriority::DEFAULT)
41    }
42
43    /// Creates a context for the current thread's dispatcher.
44    pub fn with_priority(priority: DispatcherPriority) -> Arc<Self> {
45        Self::with_dispatcher(&Dispatcher::current_dispatcher(), priority)
46    }
47
48    pub fn with_dispatcher(dispatcher: &Arc<Dispatcher>, priority: DispatcherPriority) -> Arc<Self> {
49        Arc::new(Self { dispatcher: dispatcher.clone(), priority })
50    }
51
52    /// The priority at which this context queues work.
53    pub fn priority(&self) -> DispatcherPriority {
54        self.priority
55    }
56
57    pub fn dispatcher(&self) -> &Arc<Dispatcher> {
58        &self.dispatcher
59    }
60
61    /// The context of the calling thread.
62    pub fn current() -> Option<Arc<FerroSynchronizationContext>> {
63        CURRENT.try_with(|current| current.borrow().clone()).ok().flatten()
64    }
65
66    /// Sets the context of the calling thread and returns the previous one.
67    pub fn set_current(context: Option<Arc<FerroSynchronizationContext>>) -> Option<Arc<FerroSynchronizationContext>> {
68        CURRENT.try_with(|current| current.replace(context)).ok().flatten()
69    }
70
71    /// Controls whether [`install_if_needed`](Self::install_if_needed) does
72    /// anything.
73    pub fn auto_install() -> bool {
74        AUTO_INSTALL.load(Ordering::Relaxed)
75    }
76
77    pub fn set_auto_install(value: bool) {
78        AUTO_INSTALL.store(value, Ordering::Relaxed);
79    }
80
81    /// Makes the [`DispatcherPriority::NORMAL`] context of the current
82    /// thread's dispatcher the current context, unless the thread already
83    /// has one.
84    pub fn install_if_needed() {
85        if !Self::auto_install() || Self::current().is_some() {
86            return;
87        }
88
89        Self::set_current(Some(Dispatcher::current_dispatcher().get_context_with_priority(DispatcherPriority::NORMAL)));
90    }
91
92    /// Queues `callback` on the dispatcher at the priority of this context.
93    pub fn post(&self, callback: impl FnOnce() + Send + 'static) {
94        self.dispatcher.post(callback, self.priority);
95    }
96
97    /// Like [`post`](Self::post) for callbacks that are not `Send`.
98    ///
99    /// # Panics
100    /// Panics when called from a thread other than the dispatcher thread.
101    pub fn post_local(&self, callback: impl FnOnce() + 'static) {
102        self.dispatcher.post_local(callback, self.priority);
103    }
104
105    /// Runs `callback` on the dispatcher thread and waits for it.
106    pub fn send(&self, callback: impl FnOnce() + Send + 'static) {
107        if self.dispatcher.check_access() {
108            // Same-thread, use send priority to avoid any reentrancy.
109            self.dispatcher.send(callback, Some(DispatcherPriority::SEND));
110        } else {
111            self.dispatcher.send(callback, Some(self.priority));
112        }
113    }
114
115    /// Like [`send`](Self::send) for callbacks that are not `Send`: runs
116    /// `callback` right away.
117    ///
118    /// # Panics
119    /// Panics when called from a thread other than the dispatcher thread.
120    pub fn send_local(&self, callback: impl FnOnce() + 'static) {
121        // Same-thread, use send priority to avoid any reentrancy.
122        self.dispatcher.send_local(callback, Some(DispatcherPriority::SEND));
123    }
124
125    /// The scheduler that runs futures on the dispatcher at the priority of
126    /// this context.
127    pub fn to_task_scheduler(self: &Arc<Self>) -> DispatcherTaskScheduler {
128        DispatcherTaskScheduler { context: self.clone() }
129    }
130
131    /// Makes sure that the current context of the calling thread is the one
132    /// of the current dispatcher for `priority`. Disposing or dropping the
133    /// returned value restores the previous context.
134    pub fn ensure(priority: DispatcherPriority) -> RestoreContext {
135        Self::ensure_with_dispatcher(&Dispatcher::current_dispatcher(), priority)
136    }
137
138    /// Makes sure that the current context of the calling thread is the one
139    /// of `dispatcher` for `priority`. Disposing or dropping the returned
140    /// value restores the previous context.
141    ///
142    /// # Panics
143    /// Panics when the context has to change and the calling thread is not
144    /// the dispatcher thread.
145    pub fn ensure_with_dispatcher(dispatcher: &Dispatcher, priority: DispatcherPriority) -> RestoreContext {
146        let unchanged = CURRENT
147            .try_with(|current| current.borrow().as_ref().is_some_and(|context| context.priority == priority))
148            .unwrap_or(true);
149        if unchanged {
150            return RestoreContext::default();
151        }
152        dispatcher.verify_access();
153        let old_context = Self::set_current(Some(dispatcher.get_context_with_priority(priority)));
154        RestoreContext { old_context: Cell::new(Some(old_context)) }
155    }
156}
157
158/// The context as a `System.Threading.SynchronizationContext` of `ferroui-system`, for the code
159/// that is not tied to the user interface (the reactive libraries, the tasks of `ferroui-system`):
160/// each member is the method of the same name above.
161impl ferroui_system::threading::SynchronizationContext for FerroSynchronizationContext {
162    fn post(&self, work: Box<dyn FnOnce() + Send>) {
163        FerroSynchronizationContext::post(self, work);
164    }
165
166    fn send(&self, work: Box<dyn FnOnce() + Send>) {
167        FerroSynchronizationContext::send(self, work);
168    }
169
170    fn post_local(&self, work: Box<dyn FnOnce()>) {
171        FerroSynchronizationContext::post_local(self, work);
172    }
173
174    fn send_local(&self, work: Box<dyn FnOnce()>) {
175        FerroSynchronizationContext::send_local(self, work);
176    }
177}
178
179/// Restores the previous current context; see
180/// [`FerroSynchronizationContext::ensure`].
181///
182/// The context is restored when the value is disposed or dropped, whichever
183/// comes first.
184#[derive(Default)]
185pub struct RestoreContext {
186    /// `Some` while the context still has to be restored.
187    old_context: Cell<Option<Option<Arc<FerroSynchronizationContext>>>>,
188}
189
190impl IDisposable for RestoreContext {
191    fn dispose(&self) {
192        if let Some(old_context) = self.old_context.take() {
193            FerroSynchronizationContext::set_current(old_context);
194        }
195    }
196}
197
198impl Drop for RestoreContext {
199    fn drop(&mut self) {
200        self.dispose();
201    }
202}
203
204/// Runs futures on a [`Dispatcher`] at a fixed priority: every poll of a
205/// future started here is a dispatcher job of that priority.
206///
207/// Obtained from [`Dispatcher::to_task_scheduler`] or
208/// [`FerroSynchronizationContext::to_task_scheduler`].
209#[derive(Clone)]
210pub struct DispatcherTaskScheduler {
211    context: Arc<FerroSynchronizationContext>,
212}
213
214impl DispatcherTaskScheduler {
215    pub fn dispatcher(&self) -> &Arc<Dispatcher> {
216        &self.context.dispatcher
217    }
218
219    pub fn priority(&self) -> DispatcherPriority {
220        self.context.priority
221    }
222
223    /// Runs `future` on the dispatcher thread. Can be called from any
224    /// thread.
225    pub fn spawn<Fut>(&self, future: Fut) -> DispatcherTask<Fut::Output>
226    where
227        Fut: Future + Send + 'static,
228        Fut::Output: Send + 'static,
229    {
230        let dispatcher = self.context.dispatcher.clone();
231        let priority = self.context.priority;
232        let (task, completion, wrap) = wrap_send_future::<Fut>(&dispatcher);
233        let target = dispatcher.clone();
234        // When the job is aborted its closure is dropped, which cancels the task.
235        dispatcher.post(move || target.start_task(priority, wrap(future), completion, true), priority);
236        task
237    }
238
239    /// Runs a future that is not `Send` on the dispatcher thread.
240    ///
241    /// # Panics
242    /// Panics when called from a thread other than the dispatcher thread.
243    pub fn spawn_local<Fut>(&self, future: Fut) -> DispatcherTask<Fut::Output>
244    where
245        Fut: Future + 'static,
246        Fut::Output: 'static,
247    {
248        let dispatcher = &self.context.dispatcher;
249        dispatcher.verify_access();
250        let (task, completion, wrap) = wrap_local_future::<Fut>(dispatcher);
251        dispatcher.start_task(self.context.priority, wrap(future), completion, false);
252        task
253    }
254
255    /// Starts a future that is not `Send` on the dispatcher thread the way
256    /// calling an asynchronous method does: the future is polled for the
257    /// first time before this returns, so it runs up to its first pending
258    /// await right away (and the returned task is already complete if there
259    /// is none). Afterwards it is resumed from dispatcher jobs at the
260    /// priority of the context.
261    ///
262    /// # Panics
263    /// Panics when called from a thread other than the dispatcher thread.
264    pub fn start_local<Fut>(&self, future: Fut) -> DispatcherTask<Fut::Output>
265    where
266        Fut: Future + 'static,
267        Fut::Output: 'static,
268    {
269        let dispatcher = &self.context.dispatcher;
270        dispatcher.verify_access();
271        let (task, completion, wrap) = wrap_local_future::<Fut>(dispatcher);
272        dispatcher.start_task(self.context.priority, wrap(future), completion, true);
273        task
274    }
275}
276
277impl Dispatcher {
278    /// The context of this dispatcher for `priority`. On the dispatcher
279    /// thread the contexts are cached, one per priority.
280    pub(crate) fn get_context_with_priority(&self, priority: DispatcherPriority) -> Arc<FerroSynchronizationContext> {
281        DispatcherPriority::validate(priority, "priority");
282        let Some(local) = self.try_local() else {
283            return FerroSynchronizationContext::with_dispatcher(&self.to_arc(), priority);
284        };
285        let index = (priority.value() - DispatcherPriority::MIN_VALUE.value()) as usize;
286        let mut contexts = local.priority_contexts.borrow_mut();
287        if contexts.is_empty() {
288            let count = (DispatcherPriority::MAX_VALUE.value() - DispatcherPriority::MIN_VALUE.value() + 1) as usize;
289            contexts.resize(count, None);
290        }
291        contexts[index]
292            .get_or_insert_with(|| FerroSynchronizationContext::with_dispatcher(&self.to_arc(), priority))
293            .clone()
294    }
295
296    /// Gets a scheduler which runs futures on this dispatcher. The priority
297    /// is taken from the current [`FerroSynchronizationContext`] if one is
298    /// available. Otherwise, [`DispatcherPriority::DEFAULT`] is used.
299    pub fn to_task_scheduler(&self) -> DispatcherTaskScheduler {
300        match FerroSynchronizationContext::current() {
301            Some(context) => self.to_task_scheduler_with_priority(context.priority()),
302            None => self.to_task_scheduler_with_priority(DispatcherPriority::DEFAULT),
303        }
304    }
305
306    /// Gets a scheduler which runs futures on this dispatcher with the
307    /// specified priority.
308    pub fn to_task_scheduler_with_priority(&self, priority: DispatcherPriority) -> DispatcherTaskScheduler {
309        self.get_context_with_priority(priority).to_task_scheduler()
310    }
311}
ferroui_base/threading/ferro_synchronization_context.rs

ferroui_base/threading/
ferro_synchronization_context.rs
