cranpose_core/
launched_effect.rs

1#[cfg(not(target_arch = "wasm32"))]
2use std::cell::{Cell, RefCell};
3#[cfg(not(target_arch = "wasm32"))]
4use std::rc::Rc;
5use std::{
6    future::Future,
7    pin::Pin,
8    sync::{
9        Arc,
10        atomic::{AtomicBool, Ordering},
11    },
12};
13
14use crate::{Key, RuntimeHandle, TaskHandle, effect_key::EffectKey, with_current_composer};
15
16#[derive(Default)]
17struct LaunchedEffectState {
18    key: Option<EffectKey>,
19    cancel: Option<LaunchedEffectCancellation>,
20}
21
22struct LaunchedEffectCancellation {
23    #[cfg(not(target_arch = "wasm32"))]
24    runtime: RuntimeHandle,
25    active: Arc<AtomicBool>,
26    #[cfg(not(target_arch = "wasm32"))]
27    continuations: Rc<RefCell<Vec<u64>>>,
28}
29
30/// Where an effect was written, carried onto the task it spawns.
31///
32/// A leak report that can only say "one task is still queued" names nothing:
33/// every run prints the same number and no run says which effect it was. This
34/// is what turns that count into a place to look, and it stays a pair of plain
35/// fields so carrying it costs nothing until a task is actually spawned.
36#[derive(Clone, Copy, Debug, PartialEq, Eq)]
37pub struct TaskSite {
38    pub file: &'static str,
39    pub line: u32,
40}
41
42impl TaskSite {
43    /// The site a macro captured with `file!()` and `line!()`.
44    pub const fn new(file: &'static str, line: u32) -> TaskSite {
45        TaskSite { file, line }
46    }
47}
48
49impl Default for TaskSite {
50    fn default() -> TaskSite {
51        TaskSite::new("unknown", 0)
52    }
53}
54
55impl std::fmt::Display for TaskSite {
56    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
57        write!(formatter, "{}:{}", self.file, self.line)
58    }
59}
60
61impl From<&'static std::panic::Location<'static>> for TaskSite {
62    fn from(location: &'static std::panic::Location<'static>) -> TaskSite {
63        TaskSite::new(location.file(), location.line())
64    }
65}
66
67#[derive(Default)]
68struct LaunchedEffectAsyncState {
69    key: Option<EffectKey>,
70    cancel: Option<LaunchedEffectCancellation>,
71    task: Option<TaskHandle>,
72    site: TaskSite,
73}
74
75impl LaunchedEffectState {
76    fn should_run(&self, key: &EffectKey) -> bool {
77        match &self.key {
78            Some(current) => key.differs_from(current),
79            None => true,
80        }
81    }
82
83    fn set_key(&mut self, key: EffectKey) {
84        self.key = Some(key);
85    }
86
87    fn launch(
88        &mut self,
89        runtime: RuntimeHandle,
90        effect: impl FnOnce(LaunchedEffectScope) + 'static,
91    ) {
92        self.cancel_current();
93        let active = Arc::new(AtomicBool::new(true));
94        #[cfg(not(target_arch = "wasm32"))]
95        let continuations = Rc::new(RefCell::new(Vec::new()));
96        self.cancel = Some(LaunchedEffectCancellation {
97            #[cfg(not(target_arch = "wasm32"))]
98            runtime: runtime.clone(),
99            active: Arc::clone(&active),
100            #[cfg(not(target_arch = "wasm32"))]
101            continuations: Rc::clone(&continuations),
102        });
103        let scope = LaunchedEffectScope {
104            active: Arc::clone(&active),
105            runtime: runtime.clone(),
106            #[cfg(not(target_arch = "wasm32"))]
107            continuations,
108        };
109        runtime.enqueue_ui_task(Box::new(move || effect(scope)));
110    }
111
112    fn cancel_current(&mut self) {
113        if let Some(cancel) = self.cancel.take() {
114            cancel.cancel();
115        }
116    }
117}
118
119impl LaunchedEffectCancellation {
120    fn cancel(&self) {
121        self.active.store(false, Ordering::SeqCst);
122        #[cfg(not(target_arch = "wasm32"))]
123        {
124            let mut pending = self.continuations.borrow_mut();
125            for id in pending.drain(..) {
126                self.runtime.cancel_ui_cont(id);
127            }
128        }
129    }
130}
131
132impl LaunchedEffectAsyncState {
133    fn should_run(&self, key: &EffectKey) -> bool {
134        match &self.key {
135            Some(current) => key.differs_from(current),
136            None => true,
137        }
138    }
139
140    fn set_key(&mut self, key: EffectKey) {
141        self.key = Some(key);
142    }
143
144    fn set_site(&mut self, site: TaskSite) {
145        self.site = site;
146    }
147
148    fn launch(
149        &mut self,
150        runtime: RuntimeHandle,
151        mk_future: impl FnOnce(LaunchedEffectScope) -> Pin<Box<dyn Future<Output = ()>>> + 'static,
152    ) {
153        self.cancel_current();
154        let active = Arc::new(AtomicBool::new(true));
155        #[cfg(not(target_arch = "wasm32"))]
156        let continuations = Rc::new(RefCell::new(Vec::new()));
157        self.cancel = Some(LaunchedEffectCancellation {
158            #[cfg(not(target_arch = "wasm32"))]
159            runtime: runtime.clone(),
160            active: Arc::clone(&active),
161            #[cfg(not(target_arch = "wasm32"))]
162            continuations: Rc::clone(&continuations),
163        });
164        let scope = LaunchedEffectScope {
165            active: Arc::clone(&active),
166            runtime: runtime.clone(),
167            #[cfg(not(target_arch = "wasm32"))]
168            continuations,
169        };
170        let future = mk_future(scope.clone());
171        let active_flag = Arc::clone(&scope.active);
172        crate::label_next_ui_task(self.site.to_string());
173        match runtime.spawn_ui(async move {
174            future.await;
175            active_flag.store(false, Ordering::SeqCst);
176        }) {
177            Some(handle) => {
178                self.task = Some(handle);
179            }
180            None => {
181                active.store(false, Ordering::SeqCst);
182                self.cancel = None;
183            }
184        }
185    }
186
187    fn cancel_current(&mut self) {
188        if let Some(handle) = self.task.take() {
189            handle.cancel();
190        }
191        if let Some(cancel) = self.cancel.take() {
192            cancel.cancel();
193        }
194    }
195}
196
197impl Drop for LaunchedEffectState {
198    fn drop(&mut self) {
199        self.cancel_current();
200    }
201}
202
203impl Drop for LaunchedEffectAsyncState {
204    fn drop(&mut self) {
205        self.cancel_current();
206    }
207}
208
209#[derive(Clone)]
210pub struct LaunchedEffectScope {
211    active: Arc<AtomicBool>,
212    runtime: RuntimeHandle,
213    #[cfg(not(target_arch = "wasm32"))]
214    continuations: Rc<RefCell<Vec<u64>>>,
215}
216
217impl LaunchedEffectScope {
218    #[cfg(not(target_arch = "wasm32"))]
219    fn track_continuation(&self, id: u64) {
220        self.continuations.borrow_mut().push(id);
221    }
222
223    #[cfg(not(target_arch = "wasm32"))]
224    fn release_continuation(&self, id: u64) {
225        let mut continuations = self.continuations.borrow_mut();
226        if let Some(index) = continuations.iter().position(|entry| *entry == id) {
227            continuations.remove(index);
228        }
229    }
230
231    pub fn is_active(&self) -> bool {
232        self.active.load(Ordering::SeqCst)
233    }
234
235    pub fn runtime(&self) -> RuntimeHandle {
236        self.runtime.clone()
237    }
238
239    /// Runs a follow-up `LaunchedEffect` task on the UI thread.
240    ///
241    /// The provided closure executes on the runtime thread and may freely
242    /// capture `Rc`/`RefCell` state. This must only be called from the UI
243    /// thread, typically inside another effect callback.
244    pub fn launch(&self, task: impl FnOnce(LaunchedEffectScope) + 'static) {
245        if !self.is_active() {
246            return;
247        }
248        let scope = self.clone();
249        self.runtime.enqueue_ui_task(Box::new(move || {
250            if scope.is_active() {
251                task(scope);
252            }
253        }));
254    }
255
256    /// Posts UI-only work that will execute on the runtime thread.
257    ///
258    /// The closure never crosses threads, so it may capture non-`Send` values.
259    /// Callers must invoke this from the UI thread.
260    pub fn post_ui(&self, task: impl FnOnce() + 'static) {
261        if !self.is_active() {
262            return;
263        }
264        let active = Arc::clone(&self.active);
265        self.runtime.enqueue_ui_task(Box::new(move || {
266            if active.load(Ordering::SeqCst) {
267                task();
268            }
269        }));
270    }
271
272    /// Runs background work and delivers results to the UI.
273    ///
274    /// On native targets, `work` runs on a worker thread and its future is
275    /// driven to completion there. On WASM, `work` runs as a task on the
276    /// browser event loop. The `on_ui` continuation always runs on the runtime
277    /// thread, so it may capture `Rc`/`RefCell` state safely.
278    #[cfg(not(target_arch = "wasm32"))]
279    pub fn launch_background<T, Work, Ui, Fut>(&self, work: Work, on_ui: Ui)
280    where
281        T: Send + 'static,
282        Work: FnOnce(CancelToken) -> Fut + Send + 'static,
283        Fut: Future<Output = T> + Send + 'static,
284        Ui: FnOnce(T) + 'static,
285    {
286        if !self.is_active() {
287            return;
288        }
289        let dispatcher = self.runtime.dispatcher();
290        let active_for_thread = Arc::clone(&self.active);
291        let continuation_scope = self.clone();
292        let continuation_active = Arc::clone(&self.active);
293        let id_cell = Rc::new(Cell::new(0));
294        let id_for_closure = Rc::clone(&id_cell);
295        let continuation = move |value: T| {
296            let id = id_for_closure.get();
297            continuation_scope.release_continuation(id);
298            if continuation_active.load(Ordering::SeqCst) {
299                on_ui(value);
300            }
301        };
302
303        let Some(cont_id) = self.runtime.register_ui_cont(continuation) else {
304            return;
305        };
306        id_cell.set(cont_id);
307        self.track_continuation(cont_id);
308
309        std::thread::spawn(move || {
310            let token = CancelToken::new(Arc::clone(&active_for_thread));
311            let value = pollster::block_on(work(token.clone()));
312            if token.is_cancelled() {
313                return;
314            }
315            dispatcher.post_invoke(cont_id, value);
316        });
317    }
318
319    /// Runs background work and delivers results to the UI.
320    ///
321    /// On native targets, `work` runs on a worker thread and its future is
322    /// driven to completion there. On WASM, `work` runs as a task on the
323    /// browser event loop. The `on_ui` continuation always runs on the runtime
324    /// thread, so it may capture `Rc`/`RefCell` state safely.
325    #[cfg(target_arch = "wasm32")]
326    pub fn launch_background<T, Work, Ui, Fut>(&self, work: Work, on_ui: Ui)
327    where
328        T: 'static,
329        Work: FnOnce(CancelToken) -> Fut + 'static,
330        Fut: Future<Output = T> + 'static,
331        Ui: FnOnce(T) + 'static,
332    {
333        if !self.is_active() {
334            return;
335        }
336        let active_for_task = Arc::clone(&self.active);
337        let scope = self.clone();
338        wasm_bindgen_futures::spawn_local(async move {
339            let token = CancelToken::new(Arc::clone(&active_for_task));
340            let value = work(token.clone()).await;
341            if token.is_cancelled() {
342                return;
343            }
344            scope.post_ui(move || {
345                if token.is_active() {
346                    on_ui(value);
347                }
348            });
349        });
350    }
351}
352
353#[derive(Clone)]
354/// Cooperative cancellation token passed into background `LaunchedEffect` work.
355///
356/// The token flips to "cancelled" when the associated scope leaves composition.
357/// Callers should periodically check [`CancelToken::is_cancelled`] in long-running
358/// operations and exit early; blocking I/O will not be interrupted automatically.
359pub struct CancelToken {
360    active: Arc<AtomicBool>,
361}
362
363impl CancelToken {
364    fn new(active: Arc<AtomicBool>) -> Self {
365        Self { active }
366    }
367
368    /// Returns `true` once the associated scope has been cancelled.
369    pub fn is_cancelled(&self) -> bool {
370        !self.active.load(Ordering::SeqCst)
371    }
372
373    /// Returns whether the scope is still active.
374    pub fn is_active(&self) -> bool {
375        self.active.load(Ordering::SeqCst)
376    }
377}
378
379pub fn __launched_effect_impl<K, F>(group_key: Key, keys: K, effect: F)
380where
381    K: PartialEq + 'static,
382    F: FnOnce(LaunchedEffectScope) + 'static,
383{
384    with_current_composer(|composer| {
385        composer.with_group(group_key, |composer| {
386            let key = EffectKey::new(keys);
387            let state = composer.remember_effect::<LaunchedEffectState>();
388            if state.with(|state| state.should_run(&key)) {
389                state.update(|state| state.set_key(key));
390                let runtime = composer.runtime_handle();
391                let state_for_effect = state.clone();
392                let mut effect_opt = Some(effect);
393                composer.register_side_effect(move || {
394                    if let Some(effect) = effect_opt.take() {
395                        state_for_effect.update(|state| state.launch(runtime.clone(), effect));
396                    }
397                });
398            }
399        });
400    });
401}
402
403#[macro_export]
404macro_rules! LaunchedEffect {
405    ($keys:expr, $effect:expr) => {
406        $crate::__launched_effect_impl(
407            $crate::location_key(file!(), line!(), column!()),
408            $keys,
409            $effect,
410        )
411    };
412}
413
414/// `site` is where the effect was written. It travels onto the task the effect
415/// spawns, so a task the runtime is still holding says which effect started it.
416pub fn __launched_effect_async_impl<K, F>(group_key: Key, site: TaskSite, keys: K, mk_future: F)
417where
418    K: PartialEq + 'static,
419    F: FnOnce(LaunchedEffectScope) -> Pin<Box<dyn Future<Output = ()>>> + 'static,
420{
421    with_current_composer(|composer| {
422        composer.with_group(group_key, |composer| {
423            let key = EffectKey::new(keys);
424            let state = composer.remember_effect::<LaunchedEffectAsyncState>();
425            if state.with(|state| state.should_run(&key)) {
426                state.update(|state| {
427                    state.set_key(key);
428                    state.set_site(site);
429                });
430                let runtime = composer.runtime_handle();
431                let state_for_effect = state.clone();
432                let mut mk_future_opt = Some(mk_future);
433                composer.register_side_effect(move || {
434                    if let Some(mk_future) = mk_future_opt.take() {
435                        state_for_effect.update(|state| {
436                            state.launch(runtime.clone(), mk_future);
437                        });
438                    }
439                });
440            }
441        });
442    });
443}
444
445#[macro_export]
446macro_rules! LaunchedEffectAsync {
447    ($keys:expr, $future:expr) => {
448        $crate::__launched_effect_async_impl(
449            $crate::location_key(file!(), line!(), column!()),
450            $crate::TaskSite::new(file!(), line!()),
451            $keys,
452            $future,
453        )
454    };
455}
cranpose_core/launched_effect.rs

cranpose_core/
launched_effect.rs
