telar_reactive_core/runtime/
mod.rs

1use std::cell::{Cell, Ref, RefCell, RefMut};
2use std::cmp::Reverse;
3use std::collections::BinaryHeap;
4use std::panic::Location;
5use std::rc::Rc;
6
7use rustc_hash::FxHashSet;
8
9mod effects;
10mod flush;
11mod signals;
12mod surface;
13
14pub(crate) use effects::{
15    current_observer, deregister_effect, is_alive, register_effect, register_pure_effect,
16    run_effect, schedule,
17};
18pub use flush::{batch, begin_batch, end_batch, reset_runtime, set_flush_notify};
19pub(crate) use signals::{
20    clone_signal, create_signal_storage, drop_signal, notify_signal, set_signal_value,
21    track_signal, update_signal_value, with_signal_value,
22};
23pub use surface::{
24    SurfaceEnterGuard, SurfaceHandle, current_surface, set_current_surface, set_surface_enter_hook,
25};
26
27pub(crate) type EffectId = usize;
28pub(crate) type SignalId = usize;
29
30pub(crate) struct EffectEntry {
31    pub(crate) callback: Box<dyn Fn()>,
32    // The surface active when this effect was registered; the flush re-enters it before running the
33    // callback so a cross-surface signal write resolves the effect against its own surface's world.
34    pub(crate) surface: SurfaceHandle,
35    pub(crate) is_pure: bool,
36    pub(crate) last_run_epoch: u64,
37    pub(crate) sources: Vec<SignalId>,
38    pub(crate) source_slots: Vec<usize>,
39    pub(crate) source_versions: Vec<u64>,
40    // Topological height; 0 = leaf (no tracked sources).
41    pub(crate) height: u32,
42    // Set when a memo schedules this effect. Memo dependencies are invisible to `sources` (they live in MemoInner.subscribers, not in a SignalStorage), so run_effect's version check must be bypassed once or an effect that also tracks an unchanged signal would be skipped forever.
43    pub(crate) memo_dirty: bool,
44}
45
46pub(crate) struct SignalStorage {
47    pub(crate) value: Box<dyn std::any::Any>,
48    pub(crate) version: u64,
49    pub(crate) subscribers: Vec<EffectId>,
50    // For each subscriber: the index in that effect's `source_slots` vec that records this signal.
51    pub(crate) observer_slots: Vec<usize>,
52    // Handle reference count; the slab slot is freed when it reaches zero.
53    pub(crate) ref_count: usize,
54}
55
56pub(crate) struct Runtime {
57    pub(crate) observer_stack: Vec<EffectId>,
58    pub(crate) effects: slab::Slab<EffectEntry>,
59    pub(crate) signals: slab::Slab<SignalStorage>,
60    pub(crate) batch_depth: usize,
61    pub(crate) pending: Vec<EffectId>,
62    pub(crate) memo_pending: BinaryHeap<(Reverse<u32>, EffectId)>,
63    pub(crate) pending_set: FxHashSet<EffectId>,
64    // Reused by notify_signal to copy a signal's subscribers out before scheduling, instead of allocating a fresh Vec per write.
65    subscriber_scratch: Vec<EffectId>,
66    flush_callbacks: Vec<(u64, Rc<dyn Fn()>)>,
67    next_flush_callback_id: u64,
68    pub(crate) flushing: bool,
69    flush_epoch: u64,
70}
71
72impl Runtime {
73    fn new() -> Self {
74        Runtime {
75            observer_stack: Vec::new(),
76            effects: slab::Slab::new(),
77            signals: slab::Slab::new(),
78            batch_depth: 0,
79            pending: Vec::new(),
80            memo_pending: BinaryHeap::new(),
81            pending_set: FxHashSet::default(),
82            subscriber_scratch: Vec::new(),
83            flush_callbacks: Vec::new(),
84            next_flush_callback_id: 0,
85            flushing: false,
86            flush_epoch: 0,
87        }
88    }
89}
90
91// RuntimeCell stores the runtime as a heap-allocated Box behind a raw pointer. Because *mut T has no Drop, Cell<*mut T> has no Drop, and this struct has no Drop either. That means thread_local! won't register a TLS destructor for RUNTIME — so dlclosing the dylib during hot reload no longer causes "double free or corruption" when the thread exits.
92struct RuntimeCell {
93    ptr: Cell<*mut RefCell<Runtime>>,
94    /// Where the last borrow to succeed was taken, so a collision can name what it collided with rather than
95    /// only itself. See [`crate::reentry`]. `Option<&'static Location>` has no `Drop`, so this keeps the cell
96    /// free of a TLS destructor.
97    last_borrow: Cell<Option<&'static Location<'static>>>,
98}
99
100impl RuntimeCell {
101    #[track_caller]
102    fn borrow_mut(&self) -> RefMut<'_, Runtime> {
103        self.enter(unsafe { (*self.ptr.get()).try_borrow_mut() }.ok())
104    }
105
106    #[track_caller]
107    fn borrow(&self) -> Ref<'_, Runtime> {
108        self.enter(unsafe { (*self.ptr.get()).try_borrow() }.ok())
109    }
110
111    /// Swaps in a fresh runtime and hands back the old pointer for the caller to drop. The recorded borrow
112    /// site goes with it: it names a call into a runtime that no longer exists.
113    fn take_ptr(&self) -> *mut RefCell<Runtime> {
114        self.last_borrow.set(None);
115        self.ptr
116            .replace(Box::into_raw(Box::new(RefCell::new(Runtime::new()))))
117    }
118
119    #[track_caller]
120    fn enter<G>(&self, borrowed: Option<G>) -> G {
121        let here = Location::caller();
122        match borrowed {
123            Some(guard) => {
124                self.last_borrow.set(Some(here));
125                guard
126            }
127            None => crate::reentry::borrow_collision("RUNTIME", self.last_borrow.get(), here),
128        }
129    }
130}
131
132thread_local! {
133    static RUNTIME: RuntimeCell = RuntimeCell {
134        ptr: Cell::new(Box::into_raw(Box::new(RefCell::new(Runtime::new())))),
135        last_borrow: const { Cell::new(None) },
136    };
137}
138
139pub struct FlushNotifyHandle {
140    id: u64,
141}
142
143impl Drop for FlushNotifyHandle {
144    fn drop(&mut self) {
145        deregister_flush_notify(self.id);
146    }
147}
148
149fn deregister_flush_notify(id: u64) {
150    RUNTIME.with(|rt| {
151        rt.borrow_mut()
152            .flush_callbacks
153            .retain(|(entry_id, _)| *entry_id != id);
154    });
155}
156
157#[cfg(test)]
158mod tests {
159    use super::*;
160
161    /// What a reentrant borrow used to say was `already borrowed: BorrowMutError`, over a backtrace of the
162    /// runtime's own frames: the call that collided is in there somewhere, the call it collided *with* never
163    /// is — and that second one is the whole of the diagnosis, because it is the operation still on the stack
164    /// that came back round.
165    #[test]
166    fn a_reentrant_runtime_borrow_names_both_call_sites() {
167        let quiet = std::panic::take_hook();
168        std::panic::set_hook(Box::new(|_| {}));
169        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
170            RUNTIME.with(|rt| {
171                let _held = rt.borrow_mut();
172                let _collides = rt.borrow_mut();
173            });
174        }));
175        std::panic::set_hook(quiet);
176
177        let payload = outcome.expect_err("the second borrow cannot succeed");
178        let message = payload
179            .downcast_ref::<String>()
180            .map(String::as_str)
181            .unwrap_or_default();
182        assert!(
183            message.contains("`RUNTIME` is already borrowed"),
184            "{message}"
185        );
186        assert_eq!(
187            message.matches(file!()).count(),
188            2,
189            "both sites are named, and both are in this file:\n{message}"
190        );
191    }
192}
telar_reactive_core/runtime/mod.rs

telar_reactive_core/runtime/
mod.rs
