ferroui_base/threading/
dispatcher_frame.rs

1use std::panic::{catch_unwind, resume_unwind, AssertUnwindSafe};
2use std::sync::atomic::{AtomicBool, Ordering};
3use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
4
5use super::{CancellationTokenSource, Dispatcher, IControlledDispatcherImpl};
6
7#[derive(Default)]
8struct FrameState {
9    is_running: bool,
10    cancellation_token_source: Option<CancellationTokenSource>,
11}
12
13/// Represents an execution loop in the [`Dispatcher`].
14///
15/// Frames are shared handles: [`set_continue`](Self::set_continue) can be
16/// called from any thread.
17pub struct DispatcherFrame {
18    dispatcher: Arc<Dispatcher>,
19    exit_when_requested: bool,
20    continue_: AtomicBool,
21    state: Mutex<FrameState>,
22}
23
24impl DispatcherFrame {
25    /// Constructs a new instance of the frame for the current dispatcher.
26    /// The frame exits when all frames are requested to exit.
27    pub fn new() -> Arc<DispatcherFrame> {
28        Self::with_exit_when_requested(true)
29    }
30
31    /// Constructs a new instance of the frame for the current dispatcher.
32    ///
33    /// `exit_when_requested` indicates whether or not this frame will exit
34    /// when all frames are requested to exit. Dispatcher frames typically
35    /// break down into two categories:
36    ///
37    /// 1. Long running, general purpose frames, that exit only when told to.
38    ///    These frames should exit when requested.
39    /// 2. Short running, very specific frames that exit themselves when an
40    ///    important criteria is met. These frames may consider not exiting
41    ///    when requested in favor of waiting for their important criteria to
42    ///    be met. These frames should have a timeout associated with them.
43    pub fn with_exit_when_requested(exit_when_requested: bool) -> Arc<DispatcherFrame> {
44        let frame = Self::with_dispatcher(&Dispatcher::current_dispatcher(), exit_when_requested);
45        frame.dispatcher.verify_access();
46        frame
47    }
48
49    pub(crate) fn with_dispatcher(dispatcher: &Arc<Dispatcher>, exit_when_requested: bool) -> Arc<DispatcherFrame> {
50        Arc::new(DispatcherFrame {
51            dispatcher: dispatcher.clone(),
52            exit_when_requested,
53            continue_: AtomicBool::new(true),
54            state: Mutex::new(FrameState::default()),
55        })
56    }
57
58    /// The dispatcher this frame runs on.
59    pub fn dispatcher(&self) -> &Arc<Dispatcher> {
60        &self.dispatcher
61    }
62
63    fn lock(&self) -> MutexGuard<'_, FrameState> {
64        self.state.lock().unwrap_or_else(PoisonError::into_inner)
65    }
66
67    /// Indicates that this dispatcher frame should continue.
68    pub fn continue_(&self) -> bool {
69        // This method is free-threaded.
70
71        // First check if this frame wants to continue.
72        let mut should_continue = self.continue_.load(Ordering::SeqCst);
73        if should_continue {
74            // This frame wants to continue, so next check if it will
75            // respect the "exit requests" from the dispatcher.
76            if self.exit_when_requested {
77                let dispatcher = &self.dispatcher;
78
79                // This frame is willing to respect the "exit requests" of
80                // the dispatcher, so check them.
81                if dispatcher.exit_all_frames_requested() || dispatcher.has_shutdown_started() {
82                    should_continue = false;
83                }
84            }
85        }
86
87        should_continue
88    }
89
90    /// Sets whether this dispatcher frame should continue. Setting it to
91    /// `false` exits the loop the frame is running.
92    pub fn set_continue(&self, value: bool) {
93        // This method is free-threaded.
94        let to_cancel = {
95            let state = self.lock();
96            self.continue_.store(value, Ordering::SeqCst);
97            if !value {
98                state.cancellation_token_source.clone()
99            } else {
100                None
101            }
102        };
103        // Canceled outside of the lock: cancellation runs callbacks.
104        if let Some(source) = to_cancel {
105            source.cancel();
106        }
107    }
108
109    pub(crate) fn run(&self, impl_: &dyn IControlledDispatcherImpl) {
110        self.dispatcher.verify_access();
111
112        // Since the actual platform run loop is controlled by a cancellation token, we have an
113        // outer loop that restarts the platform one in case Continue was set to true after being set to false
114        loop {
115            let token = {
116                let mut state = self.lock();
117                if !self.continue_() {
118                    return;
119                }
120
121                if state.is_running {
122                    panic!("This frame is already running");
123                }
124
125                let source = CancellationTokenSource::new();
126                let token = source.token();
127                state.cancellation_token_source = Some(source);
128                state.is_running = true;
129                token
130            };
131
132            let result = catch_unwind(AssertUnwindSafe(|| {
133                // Wake up the dispatcher in case it has pending jobs
134                self.dispatcher.request_processing();
135                impl_.run_loop(token);
136            }));
137
138            // finally
139            let source = {
140                let mut state = self.lock();
141                state.is_running = false;
142                state.cancellation_token_source.take()
143            };
144            if let Some(source) = source {
145                source.cancel();
146            }
147            if let Err(payload) = result {
148                resume_unwind(payload);
149            }
150        }
151    }
152
153    pub(crate) fn maybe_exit_on_dispatcher_request(&self) {
154        if self.exit_when_requested {
155            let source = self.lock().cancellation_token_source.clone();
156            if let Some(source) = source {
157                source.cancel();
158            }
159        }
160    }
161}
ferroui_base/threading/dispatcher_frame.rs

ferroui_base/threading/
dispatcher_frame.rs
