ferroui_browser/
browser_single_threaded_dispatcher_impl.rs

1use crate::interop::thread_proxy::{self, RunOnThread};
2use ferroui_base::threading::{
3    DispatcherImplEvent, IDispatcherImpl, IDispatcherImplWithExplicitBackgroundProcessing, IDispatcherSignal,
4};
5use std::cell::{Cell, RefCell};
6use std::rc::Rc;
7use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
8use std::sync::Arc;
9use std::thread::{self, ThreadId};
10use std::time::Instant;
11use wasm_bindgen::prelude::*;
12
13#[wasm_bindgen(raw_module = "./ferroui.js")]
14extern "C" {
15    #[wasm_bindgen(js_namespace = SingleThreadedDispatcherHelper, js_name = signal)]
16    fn js_signal();
17
18    #[wasm_bindgen(js_namespace = SingleThreadedDispatcherHelper, js_name = requestBackgroundProcessing)]
19    fn js_request_background_processing();
20
21    #[wasm_bindgen(js_namespace = SingleThreadedDispatcherHelper, js_name = setTimer)]
22    fn js_set_timer(delay_ms: i32);
23
24    #[wasm_bindgen(js_namespace = SingleThreadedDispatcherHelper, js_name = clearTimer)]
25    fn js_clear_timer();
26}
27
28/// The calls into the page the dispatcher makes; replaced by a recorder in
29/// the tests.
30trait IScheduling {
31    fn signal(&self);
32    fn request_background_processing(&self);
33    fn set_timer(&self, delay_ms: i32);
34    fn clear_timer(&self);
35}
36
37struct PageScheduling;
38
39impl IScheduling for PageScheduling {
40    fn signal(&self) {
41        js_signal();
42    }
43
44    fn request_background_processing(&self) {
45        js_request_background_processing();
46    }
47
48    fn set_timer(&self, delay_ms: i32) {
49        js_set_timer(delay_ms);
50    }
51
52    fn clear_timer(&self) {
53        js_clear_timer();
54    }
55}
56
57ferroui_base::ferro_thread_local! {
58    static INSTANCE: RefCell<Option<Rc<BrowserSingleThreadedDispatcherImpl>>> = const { RefCell::new(None) };
59}
60
61fn instance() -> Option<Rc<BrowserSingleThreadedDispatcherImpl>> {
62    INSTANCE.with(|instance| instance.borrow().clone())
63}
64
65/// The thread of the dispatcher, as [`thread_proxy::current_thread`] names
66/// it; 0 until a dispatcher is created, and always in a build without
67/// threads.
68static LOOP_THREAD: AtomicUsize = AtomicUsize::new(0);
69
70/// Whether a wake-up raised on another thread is on its way to the thread of
71/// the dispatcher.
72static CROSS_THREAD_SIGNAL_PENDING: AtomicBool = AtomicBool::new(false);
73
74/// Dispatcher backend for the browser. The browser event loop is the only
75/// loop: wake-ups are posted as macrotasks and the dispatcher runs on the
76/// main thread, so its own state needs no locking. A module built with
77/// threads has other threads (the render worker), which wake the dispatcher
78/// through its signal handle. Pending input is deliberately
79/// not queried: the browser dispatches input between our tasks on its own,
80/// and gating low-priority jobs on a pending-input query starves them for as
81/// long as the pointer keeps moving.
82pub struct BrowserSingleThreadedDispatcherImpl {
83    thread: ThreadId,
84    clock: Instant,
85    signaled_flag: Cell<bool>,
86    background_requested: Cell<bool>,
87    timer_set: Cell<bool>,
88    signaled: DispatcherImplEvent,
89    timer: DispatcherImplEvent,
90    ready_for_background_processing: DispatcherImplEvent,
91    scheduling: Box<dyn IScheduling>,
92}
93
94/// Wakes the dispatcher up. On the thread of the dispatcher the handle
95/// forwards to the instance of that thread. On any other thread of a module
96/// built with threads it carries the wake-up across.
97///
98/// Not from upstream, whose browser dispatcher in the threaded mode is the
99/// managed one, woken through an event of the runtime. See
100/// `docs/porting/browser-render-worker.md`, section 3 and "B2.3".
101struct Signal;
102
103impl IDispatcherSignal for Signal {
104    fn signal(&self) {
105        if let Some(instance) = instance() {
106            IDispatcherImpl::signal(&*instance);
107        } else {
108            signal_from_another_thread(
109                &CROSS_THREAD_SIGNAL_PENDING,
110                LOOP_THREAD.load(Ordering::SeqCst),
111                thread_proxy::run_on_thread,
112            );
113        }
114    }
115}
116
117/// Carries a wake-up to `loop_thread`: a flag, so that the wake-ups raised
118/// before the thread has taken one are one, and a call queued for the thread
119/// that signals its dispatcher there. The call runs from the event loop of
120/// the thread, and what it does is what a signal raised on that thread does:
121/// it posts the task of the dispatcher. So a wake-up from another thread
122/// never runs dispatcher work in the middle of something else, not even
123/// while the thread waits for a frame.
124///
125/// Does nothing without a loop thread (no dispatcher yet, or a build without
126/// threads, where a thread that has no dispatcher has nobody to wake).
127fn signal_from_another_thread(pending: &'static AtomicBool, loop_thread: usize, run_on_thread: RunOnThread) {
128    if loop_thread == 0 {
129        return;
130    }
131    if pending.swap(true, Ordering::SeqCst) {
132        return;
133    }
134    let queued = run_on_thread(
135        loop_thread,
136        Box::new(move || {
137            // Cleared first: a wake-up raised from here on is queued again,
138            // and the signal below covers everything raised before.
139            pending.store(false, Ordering::SeqCst);
140            if let Some(instance) = instance() {
141                IDispatcherImpl::signal(&*instance);
142            }
143        }),
144    );
145    if !queued {
146        pending.store(false, Ordering::SeqCst);
147    }
148}
149
150impl BrowserSingleThreadedDispatcherImpl {
151    /// Creates the dispatcher backend of the current thread and makes it
152    /// the receiver of the callbacks of the page.
153    pub fn new() -> Rc<Self> {
154        Self::with_scheduling(Box::new(PageScheduling))
155    }
156
157    fn with_scheduling(scheduling: Box<dyn IScheduling>) -> Rc<Self> {
158        let this = Rc::new(Self {
159            thread: thread::current().id(),
160            clock: Instant::now(),
161            signaled_flag: Cell::new(false),
162            background_requested: Cell::new(false),
163            timer_set: Cell::new(false),
164            signaled: DispatcherImplEvent::new(),
165            timer: DispatcherImplEvent::new(),
166            ready_for_background_processing: DispatcherImplEvent::new(),
167            scheduling,
168        });
169        INSTANCE.with(|instance| *instance.borrow_mut() = Some(this.clone()));
170        LOOP_THREAD.store(thread_proxy::current_thread(), Ordering::SeqCst);
171        this
172    }
173}
174
175impl IDispatcherImpl for BrowserSingleThreadedDispatcherImpl {
176    fn current_thread_is_loop_thread(&self) -> bool {
177        thread::current().id() == self.thread
178    }
179
180    fn signal(&self) {
181        if self.signaled_flag.get() {
182            return;
183        }
184        self.signaled_flag.set(true);
185        self.scheduling.signal();
186    }
187
188    fn signal_handle(&self) -> Arc<dyn IDispatcherSignal> {
189        Arc::new(Signal)
190    }
191
192    fn signaled(&self) -> &DispatcherImplEvent {
193        &self.signaled
194    }
195
196    fn timer(&self) -> &DispatcherImplEvent {
197        &self.timer
198    }
199
200    fn now(&self) -> i64 {
201        self.clock.elapsed().as_millis() as i64
202    }
203
204    fn update_timer(&self, due_time_in_ms: Option<i64>) {
205        if self.timer_set.get() {
206            self.scheduling.clear_timer();
207            self.timer_set.set(false);
208        }
209
210        if let Some(due_time) = due_time_in_ms {
211            self.scheduling.set_timer((due_time - self.now()).clamp(0, i32::MAX as i64) as i32);
212            self.timer_set.set(true);
213        }
214    }
215
216    fn as_explicit_background_processing(&self) -> Option<&dyn IDispatcherImplWithExplicitBackgroundProcessing> {
217        Some(self)
218    }
219}
220
221impl IDispatcherImplWithExplicitBackgroundProcessing for BrowserSingleThreadedDispatcherImpl {
222    fn ready_for_background_processing(&self) -> &DispatcherImplEvent {
223        &self.ready_for_background_processing
224    }
225
226    fn request_background_processing(&self) {
227        if self.background_requested.get() {
228            return;
229        }
230        self.background_requested.set(true);
231        self.scheduling.request_background_processing();
232    }
233}
234
235/// The task posted by a signal runs.
236#[wasm_bindgen(js_name = BrowserSingleThreadedDispatcherImpl_OnSignaled)]
237pub fn on_signaled() {
238    let Some(impl_) = instance() else { return };
239    impl_.signaled_flag.set(false);
240    impl_.signaled.raise();
241}
242
243/// The task posted by a background processing request runs.
244#[wasm_bindgen(js_name = BrowserSingleThreadedDispatcherImpl_OnReadyForBackgroundProcessing)]
245pub fn on_ready_for_background_processing() {
246    let Some(impl_) = instance() else { return };
247    impl_.background_requested.set(false);
248    impl_.ready_for_background_processing.raise();
249}
250
251/// The timer of the dispatcher is due.
252#[wasm_bindgen(js_name = BrowserSingleThreadedDispatcherImpl_OnTimer)]
253pub fn on_timer() {
254    let Some(impl_) = instance() else { return };
255    impl_.timer_set.set(false);
256    impl_.timer.raise();
257}
258
259#[cfg(test)]
260mod tests {
261    use super::*;
262
263    #[derive(Default)]
264    struct Recorder {
265        calls: Rc<RefCell<Vec<String>>>,
266    }
267
268    impl IScheduling for Recorder {
269        fn signal(&self) {
270            self.calls.borrow_mut().push("signal".to_string());
271        }
272
273        fn request_background_processing(&self) {
274            self.calls.borrow_mut().push("background".to_string());
275        }
276
277        fn set_timer(&self, delay_ms: i32) {
278            self.calls.borrow_mut().push(format!("setTimer({})", delay_ms.min(1_000_000) / 1000));
279        }
280
281        fn clear_timer(&self) {
282            self.calls.borrow_mut().push("clearTimer".to_string());
283        }
284    }
285
286    fn create() -> (Rc<BrowserSingleThreadedDispatcherImpl>, Rc<RefCell<Vec<String>>>) {
287        let calls = Rc::new(RefCell::new(Vec::new()));
288        let impl_ = BrowserSingleThreadedDispatcherImpl::with_scheduling(Box::new(Recorder { calls: calls.clone() }));
289        (impl_, calls)
290    }
291
292    fn counter(event: &DispatcherImplEvent) -> Rc<Cell<u32>> {
293        let count = Rc::new(Cell::new(0));
294        let seen = count.clone();
295        event.add(Rc::new(move |_| seen.set(seen.get() + 1)));
296        count
297    }
298
299    #[test]
300    fn the_creating_thread_is_the_loop_thread() {
301        let (impl_, _) = create();
302
303        assert!(impl_.current_thread_is_loop_thread());
304    }
305
306    #[test]
307    fn signals_are_coalesced_until_the_posted_task_runs() {
308        let (impl_, calls) = create();
309        let signaled = counter(impl_.signaled());
310
311        IDispatcherImpl::signal(&*impl_);
312        IDispatcherImpl::signal(&*impl_);
313        assert_eq!(vec!["signal"], *calls.borrow());
314        assert_eq!(0, signaled.get());
315
316        on_signaled();
317        assert_eq!(1, signaled.get());
318
319        IDispatcherImpl::signal(&*impl_);
320        assert_eq!(vec!["signal", "signal"], *calls.borrow());
321    }
322
323    #[test]
324    fn the_signal_handle_wakes_the_instance_of_the_thread() {
325        let (impl_, calls) = create();
326
327        impl_.signal_handle().signal();
328        impl_.signal_handle().signal();
329
330        assert_eq!(vec!["signal"], *calls.borrow());
331    }
332
333    #[test]
334    fn a_signal_from_another_thread_is_carried_to_the_loop_thread_once() {
335        use crate::interop::thread_proxy::ThreadWork;
336        use std::sync::Mutex;
337
338        static PENDING: AtomicBool = AtomicBool::new(false);
339        static QUEUED: Mutex<Vec<(usize, ThreadWork)>> = Mutex::new(Vec::new());
340        fn queue(thread: usize, work: ThreadWork) -> bool {
341            QUEUED.lock().unwrap().push((thread, work));
342            true
343        }
344        fn refuse(_thread: usize, _work: ThreadWork) -> bool {
345            false
346        }
347        fn take() -> Vec<(usize, ThreadWork)> {
348            QUEUED.lock().unwrap().drain(..).collect()
349        }
350
351        let (_impl, calls) = create();
352
353        // Another thread has no dispatcher; its two wake-ups are one call for the loop thread.
354        std::thread::spawn(|| {
355            assert!(instance().is_none());
356            signal_from_another_thread(&PENDING, 7, queue);
357            signal_from_another_thread(&PENDING, 7, queue);
358        })
359        .join()
360        .unwrap();
361        let queued = take();
362        assert_eq!(vec![7], queued.iter().map(|(thread, _)| *thread).collect::<Vec<usize>>());
363        assert!(calls.borrow().is_empty());
364
365        // The call runs on the loop thread and signals the dispatcher there.
366        for (_, work) in queued {
367            work();
368        }
369        assert_eq!(vec!["signal"], *calls.borrow());
370        assert!(!PENDING.load(Ordering::SeqCst));
371
372        // A wake-up that could not be queued does not keep later ones back.
373        signal_from_another_thread(&PENDING, 7, refuse);
374        assert!(!PENDING.load(Ordering::SeqCst));
375
376        // Without a loop thread there is nobody to wake.
377        signal_from_another_thread(&PENDING, 0, queue);
378        assert!(take().is_empty());
379    }
380
381    #[test]
382    fn the_signal_handle_of_a_thread_without_a_dispatcher_is_lost_in_a_build_without_threads() {
383        let (impl_, calls) = create();
384        let handle = impl_.signal_handle();
385
386        std::thread::spawn(move || handle.signal()).join().unwrap();
387
388        assert!(calls.borrow().is_empty());
389        assert!(!CROSS_THREAD_SIGNAL_PENDING.load(Ordering::SeqCst));
390    }
391
392    #[test]
393    fn background_requests_are_coalesced_until_the_posted_task_runs() {
394        let (impl_, calls) = create();
395        let ready = counter(impl_.ready_for_background_processing());
396
397        impl_.request_background_processing();
398        impl_.request_background_processing();
399        assert_eq!(vec!["background"], *calls.borrow());
400
401        on_ready_for_background_processing();
402        assert_eq!(1, ready.get());
403
404        impl_.request_background_processing();
405        assert_eq!(vec!["background", "background"], *calls.borrow());
406    }
407
408    #[test]
409    fn the_timer_is_set_relative_to_now_and_replaced_by_the_next_update() {
410        let (impl_, calls) = create();
411
412        impl_.update_timer(Some(impl_.now() + 5_500));
413        impl_.update_timer(Some(impl_.now() + 2_500));
414        impl_.update_timer(None);
415        impl_.update_timer(None);
416
417        assert_eq!(vec!["setTimer(5)", "clearTimer", "setTimer(2)", "clearTimer"], *calls.borrow());
418    }
419
420    #[test]
421    fn a_due_time_in_the_past_fires_immediately() {
422        let (impl_, calls) = create();
423
424        impl_.update_timer(Some(-1_000));
425
426        assert_eq!(vec!["setTimer(0)"], *calls.borrow());
427    }
428
429    #[test]
430    fn a_fired_timer_is_not_cleared_by_the_next_update() {
431        let (impl_, calls) = create();
432        let timer = counter(impl_.timer());
433
434        impl_.update_timer(Some(impl_.now() + 1_500));
435        on_timer();
436        assert_eq!(1, timer.get());
437        impl_.update_timer(Some(impl_.now() + 3_500));
438
439        assert_eq!(vec!["setTimer(1)", "setTimer(3)"], *calls.borrow());
440    }
441
442    #[test]
443    fn explicit_background_processing_is_supported_and_a_loop_is_not() {
444        let (impl_, _) = create();
445
446        assert!(impl_.as_explicit_background_processing().is_some());
447        assert!(impl_.as_pending_input().is_none());
448        assert!(impl_.as_controlled().is_none());
449    }
450}
ferroui_browser/browser_single_threaded_dispatcher_impl.rs

ferroui_browser/
browser_single_threaded_dispatcher_impl.rs
