telar_reactive_core/runtime/
flush.rs

1use std::rc::Rc;
2
3use super::effects::run_effect;
4use super::{EffectId, FlushNotifyHandle, RUNTIME};
5
6const MAX_FLUSH_ITERATIONS: usize = 1_000;
7
8pub(crate) fn flush() {
9    RUNTIME.with(|rt| rt.borrow_mut().flushing = true);
10
11    // With the runtime shared across surfaces, a panic mid-effect must not leave `flushing` stuck true —
12    // that would wedge every surface's scheduling (schedule() early-returns while flushing). The guard
13    // clears it on any exit: normal return, the overflow panic below, or an unwind out of run_effect.
14    struct FlushGuard;
15    impl Drop for FlushGuard {
16        fn drop(&mut self) {
17            RUNTIME.with(|rt| rt.borrow_mut().flushing = false);
18        }
19    }
20
21    let (did_work, overflowed) = {
22        let _flush_guard = FlushGuard;
23        let mut did_work = false;
24        let mut overflowed = true;
25        for _ in 0..MAX_FLUSH_ITERATIONS {
26            // One epoch per drain, not one per flush. The dedup in `run_effect` is there so two writes to
27            // the same signal in one drain cost one run; an effect whose source is written by a *later*
28            // effect in the same cascade must still run again. Under a flush-wide epoch that re-run was
29            // scheduled, popped and skipped, and nothing rescheduled it — the effect stayed stale until
30            // some unrelated event forced it. A genuine write-read cycle is still caught by the
31            // iteration cap below.
32            RUNTIME.with(|rt| rt.borrow_mut().flush_epoch += 1);
33            // Drain memo_pending first (pure computations), then user effects. Pop minimum height first so producers run before consumers (topological order).
34            let memo_batch: Vec<EffectId> = RUNTIME.with(|rt| {
35                let mut rt = rt.borrow_mut();
36                let mut batch = Vec::new();
37                while let Some((_, id)) = rt.memo_pending.pop() {
38                    batch.push(id);
39                }
40                batch
41            });
42            let pending_batch = if memo_batch.is_empty() {
43                RUNTIME.with(|rt| {
44                    let mut rt = rt.borrow_mut();
45                    rt.pending_set.clear();
46                    std::mem::take(&mut rt.pending)
47                })
48            } else {
49                Vec::new()
50            };
51
52            if memo_batch.is_empty() && pending_batch.is_empty() {
53                overflowed = false;
54                break;
55            }
56            did_work = true;
57            for id in memo_batch {
58                run_effect(id);
59            }
60            for id in pending_batch {
61                run_effect(id);
62            }
63        }
64        (did_work, overflowed)
65    };
66
67    if overflowed {
68        panic!(
69            "reactive flush exceeded {MAX_FLUSH_ITERATIONS} iterations — \
70             likely an effect is writing to a signal it depends on"
71        );
72    }
73
74    // Notify flush observers (e.g. the runner's redraw waker) after `flushing` is cleared, so a callback
75    // that writes a signal can schedule and drive a fresh flush.
76    if did_work {
77        let cbs: smallvec::SmallVec<[Rc<dyn Fn()>; 2]> = RUNTIME.with(|rt| {
78            rt.borrow()
79                .flush_callbacks
80                .iter()
81                .map(|(_, cb)| Rc::clone(cb))
82                .collect()
83        });
84        for cb in cbs {
85            cb();
86        }
87    }
88}
89
90pub fn batch<R>(f: impl FnOnce() -> R) -> R {
91    // A panic inside `f` must not leave `batch_depth` unbalanced (it would suppress every future flush on
92    // the shared runtime). The guard decrements on any exit, including an unwind.
93    struct DepthGuard;
94    impl Drop for DepthGuard {
95        fn drop(&mut self) {
96            RUNTIME.with(|rt| {
97                let mut rt = rt.borrow_mut();
98                rt.batch_depth = rt.batch_depth.saturating_sub(1);
99            });
100        }
101    }
102    RUNTIME.with(|rt| rt.borrow_mut().batch_depth += 1);
103    let result = {
104        let _depth_guard = DepthGuard;
105        f()
106    };
107    let should_flush = RUNTIME.with(|rt| {
108        let rt = rt.borrow();
109        rt.batch_depth == 0 && (!rt.pending.is_empty() || !rt.memo_pending.is_empty())
110    });
111    if should_flush {
112        flush();
113    }
114    result
115}
116
117pub fn begin_batch() {
118    RUNTIME.with(|rt| rt.borrow_mut().batch_depth += 1);
119}
120
121pub fn end_batch() {
122    let should_flush = RUNTIME.with(|rt| {
123        let mut rt = rt.borrow_mut();
124        debug_assert!(rt.batch_depth > 0, "end_batch without begin_batch");
125        rt.batch_depth = rt.batch_depth.saturating_sub(1);
126        rt.batch_depth == 0 && (!rt.pending.is_empty() || !rt.memo_pending.is_empty())
127    });
128    if should_flush {
129        flush();
130    }
131}
132
133pub fn reset_runtime() {
134    RUNTIME.with(|cell| {
135        let old_ptr = cell.take_ptr();
136        // Install a fresh runtime BEFORE dropping the old one. Any re-entrant RUNTIME access during the old runtime's drop glue (drop_signal, etc.) will see the new empty runtime and return early — no borrow conflict, no double-free.
137        if !old_ptr.is_null() {
138            unsafe { drop(Box::from_raw(old_ptr)) };
139        }
140    });
141}
142
143pub fn set_flush_notify(f: impl Fn() + 'static) -> FlushNotifyHandle {
144    RUNTIME.with(|rt| {
145        let mut rt = rt.borrow_mut();
146        let id = rt.next_flush_callback_id;
147        rt.next_flush_callback_id += 1;
148        rt.flush_callbacks.push((id, Rc::new(f)));
149        FlushNotifyHandle { id }
150    })
151}
telar_reactive_core/runtime/flush.rs

telar_reactive_core/runtime/
flush.rs
