ferroui_base/threading/
dispatcher_queue.rs

1use std::sync::atomic::Ordering;
2use std::sync::Arc;
3
4use super::dispatcher::DispatcherState;
5use super::dispatcher_operation::OperationCore;
6use super::{CancellationToken, Dispatcher, DispatcherOperationStatus, DispatcherPriority, IDispatcherImpl};
7
8impl Dispatcher {
9    pub(crate) const MAXIMUM_INPUT_STARVATION_TIME_IN_FALLBACK_MODE: i64 = 50;
10    pub(crate) const MAXIMUM_INPUT_STARVATION_TIME_IN_EXPLICIT_PROCESSING_EXPLICIT_MODE: i64 = 50;
11
12    pub(crate) fn request_background_processing_locked(&self, st: &mut DispatcherState) {
13        // Dispatcher thread only.
14        let Some(impl_) = self.local_impl() else {
15            return;
16        };
17        if let Some(background) = impl_.as_explicit_background_processing() {
18            if st.explicit_background_processing_requested {
19                return;
20            }
21            st.explicit_background_processing_requested = true;
22            background.request_background_processing();
23        } else if st.due_time_for_background_processing.is_none() {
24            st.due_time_for_background_processing = Some(impl_.now() + 1);
25            self.update_os_timer_locked(st);
26        }
27    }
28
29    pub(crate) fn on_ready_for_explicit_background_processing(&self) {
30        self.lock().explicit_background_processing_requested = false;
31        self.execute_jobs_core(true);
32    }
33
34    /// Force-runs all dispatcher operations ignoring any pending OS events,
35    /// use with caution.
36    ///
37    /// `None` runs every active job.
38    pub fn run_jobs(&self, priority: Option<DispatcherPriority>) {
39        self.run_jobs_with_cancellation(priority, &CancellationToken::none());
40    }
41
42    pub(crate) fn run_jobs_with_cancellation(
43        &self,
44        priority: Option<DispatcherPriority>,
45        cancellation_token: &CancellationToken,
46    ) {
47        if self.lock().disabled_processing_count > 0 {
48            panic!("Cannot perform this operation while dispatcher processing is suspended.");
49        }
50
51        let mut priority = priority.unwrap_or(DispatcherPriority::MINIMUM_ACTIVE_VALUE);
52        if priority < DispatcherPriority::MINIMUM_ACTIVE_VALUE {
53            priority = DispatcherPriority::MINIMUM_ACTIVE_VALUE;
54        }
55        while !cancellation_token.is_cancellation_requested() {
56            let job = self.lock().queue.peek();
57            let Some(job) = job else {
58                return;
59            };
60            if job.priority() < priority {
61                return;
62            }
63            self.execute_job(&job);
64        }
65    }
66
67    pub(crate) fn execute_job(&self, job: &Arc<OperationCore>) {
68        let dead_local_operations = {
69            let mut st = self.lock();
70            if job.status() != DispatcherOperationStatus::Pending {
71                return;
72            }
73
74            st.queue.remove_item(job);
75            job.set_status(DispatcherOperationStatus::Executing);
76            std::mem::take(&mut st.dead_local_operations)
77        };
78        for operation_id in dead_local_operations {
79            self.discard_local_operation(operation_id);
80        }
81
82        job.execute();
83        // The backend might be firing timers with a low priority,
84        // so we manually check if our high priority timers are due for execution
85        self.promote_timers();
86    }
87
88    pub(crate) fn signaled(&self) {
89        self.lock().signaled = false;
90
91        self.execute_jobs_core(false);
92    }
93
94    pub(crate) fn execute_jobs_core(&self, from_explicit_background_processing_callback: bool) {
95        let mut background_job_execution_started_at: Option<i64> = None;
96        loop {
97            let (job, maximum_input_starvation_time) = {
98                let st = self.lock();
99                (st.queue.peek(), st.maximum_input_starvation_time)
100            };
101            let Some(impl_) = self.local_impl() else {
102                return;
103            };
104
105            let Some(job) = job else {
106                return;
107            };
108            if job.priority() < DispatcherPriority::MINIMUM_ACTIVE_VALUE {
109                return;
110            }
111
112            let pending_input = impl_.as_pending_input().filter(|pending| pending.can_query_pending_input());
113
114            // We don't stop for executing jobs queued with >Input priority
115            if job.priority() > DispatcherPriority::INPUT {
116                self.execute_job(&job);
117            }
118            // If platform supports pending input query, ask the platform if we can continue running low priority jobs
119            else if let Some(pending_input) = pending_input {
120                if !pending_input.has_pending_input() {
121                    self.execute_job(&job);
122                } else {
123                    self.request_background_processing_locked(&mut self.lock());
124                    return;
125                }
126            }
127            // We can't ask if the implementation has pending input, so we should let it to call us back
128            // Once it thinks that input is handled
129            else if impl_.as_explicit_background_processing().is_some()
130                && !from_explicit_background_processing_callback
131            {
132                self.request_background_processing_locked(&mut self.lock());
133                return;
134            }
135            // We can't check if there is pending input, but still need to enforce interactivity
136            // so we stop processing background jobs after some timeout and start a timer to continue later
137            else {
138                let started_at = *background_job_execution_started_at.get_or_insert_with(|| impl_.now());
139
140                if impl_.now() - started_at > maximum_input_starvation_time {
141                    self.request_background_processing_locked(&mut self.lock());
142                    return;
143                } else {
144                    self.execute_job(&job);
145                }
146            }
147
148            // Yield after each job once shutdown has started, so frames that were asked to exit
149            // can unwind before the next job runs; shutdown_impl then aborts whatever is left
150            if self.has_shutdown_started() {
151                // Dispatcher thread
152                let has_shutdown_finished = self.lock().has_shutdown_finished;
153                if !has_shutdown_finished && self.has_jobs_with_priority(DispatcherPriority::MINIMUM_ACTIVE_VALUE) {
154                    self.request_processing();
155                }
156                return;
157            }
158        }
159    }
160
161    pub(crate) fn request_processing(&self) -> bool {
162        self.request_processing_locked(&mut self.lock())
163    }
164
165    pub(crate) fn request_processing_locked(&self, st: &mut DispatcherState) -> bool {
166        if !self.check_access() {
167            Self::request_foreground_processing_locked(st, None);
168            return true;
169        }
170
171        let impl_ = self.local_impl();
172        if st.queue.max_priority() <= DispatcherPriority::INPUT {
173            let can_run_now = impl_
174                .as_ref()
175                .and_then(|impl_| impl_.as_pending_input())
176                .is_some_and(|pending| pending.can_query_pending_input() && !pending.has_pending_input());
177            if can_run_now {
178                Self::request_foreground_processing_locked(st, impl_.as_deref());
179            } else {
180                self.request_background_processing_locked(st);
181            }
182        } else {
183            Self::request_foreground_processing_locked(st, impl_.as_deref());
184        }
185        true
186    }
187
188    /// `impl_` is the platform implementation when called on the dispatcher
189    /// thread; other threads go through the thread-safe signal.
190    fn request_foreground_processing_locked(st: &mut DispatcherState, impl_: Option<&dyn IDispatcherImpl>) {
191        if !st.signaled {
192            st.signaled = true;
193            match impl_ {
194                Some(impl_) => impl_.signal(),
195                None => st.signal.signal(),
196            }
197        }
198    }
199
200    pub(crate) fn abort(&self, operation: &Arc<OperationCore>) -> bool {
201        let mut st = self.lock();
202        if operation.status() != DispatcherOperationStatus::Pending {
203            return false;
204        }
205        st.queue.remove_item(operation);
206        operation.set_status(DispatcherOperationStatus::Aborted);
207        true
208    }
209
210    /// Returns whether or not the priority was set.
211    pub(crate) fn set_priority(&self, operation: &Arc<OperationCore>, priority: DispatcherPriority) -> bool {
212        let mut notify = false;
213
214        let mut st = self.lock();
215        if operation.is_queued() {
216            st.queue.change_item_priority(operation, priority);
217            notify = true;
218
219            // Make sure we will wake up to process this operation.
220            self.request_processing_locked(&mut st);
221        }
222        notify
223    }
224
225    /// Whether a job with at least the given priority is queued.
226    pub fn has_jobs_with_priority(&self, priority: DispatcherPriority) -> bool {
227        self.lock().queue.max_priority() >= priority
228    }
229
230    /// Gets all pending jobs, in arrival order, without removing them.
231    ///
232    /// Only use between unit tests!
233    #[allow(dead_code)]
234    pub(crate) fn get_jobs(&self) -> Vec<Arc<OperationCore>> {
235        self.lock().queue.peek_all()
236    }
237
238    /// Clears all pending jobs.
239    ///
240    /// Only use between unit tests!
241    #[allow(dead_code)]
242    pub(crate) fn clear_jobs(&self) {
243        let jobs = self.lock().queue.clear();
244        // Released outside of the lock.
245        drop(jobs);
246    }
247
248    pub(crate) fn has_shutdown_started(&self) -> bool {
249        self.has_shutdown_started.load(Ordering::SeqCst)
250    }
251}
ferroui_base/threading/dispatcher_queue.rs

ferroui_base/threading/
dispatcher_queue.rs
