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

cranpose_core/
launched_effect.rs
