1use std::{
2 any::Any,
3 cell::{Cell, RefCell},
4 collections::VecDeque,
5 future::Future,
6 pin::Pin,
7 rc::{Rc, Weak},
8 sync::{
9 atomic::{AtomicBool, AtomicUsize, Ordering},
10 mpsc, Arc,
11 },
12 task::{Context, Poll, Waker},
13 thread::ThreadId,
14 thread_local,
15};
16
17#[cfg(any(feature = "internal", test))]
18use crate::frame_clock::FrameClock;
19use crate::{
20 collections::map::{HashMap, HashSet},
21 platform::{RuntimeScheduler, SchedulerRef},
22 state::{MutationPolicy, NeverEqual},
23 Applier, Command, FrameCallbackId, MutableStateInner, NodeError, RecomposeScopeInner, ScopeId,
24};
25
26#[derive(Clone, Copy, PartialEq, Eq)]
27pub(crate) enum FrameCallbackKind {
28 Transient,
29 Perpetual,
30}
31
32enum UiMessage {
33 Task(Box<dyn FnOnce() + Send + 'static>),
34 Invoke { id: u64, value: Box<dyn Any + Send> },
35}
36
37type UiContinuation = Box<dyn Fn(Box<dyn Any>) -> bool + 'static>;
38type UiContinuationMap = HashMap<u64, UiContinuation>;
39
40struct TypedStateCell<T: Clone + 'static> {
41 inner: MutableStateInner<T>,
42}
43
44trait ScopeWatchCell {
45 fn unregister_scope(&self, scope_id: ScopeId);
46}
47
48impl<T: Clone + 'static> ScopeWatchCell for TypedStateCell<T> {
49 fn unregister_scope(&self, scope_id: ScopeId) {
50 self.inner.unregister_scope(scope_id);
51 }
52}
53
54struct StateArenaSlot {
55 generation: u32,
56 cell: Option<Rc<dyn Any>>,
57 watcher_cell: Option<Rc<dyn ScopeWatchCell>>,
58 lease: Option<Weak<StateHandleLease>>,
59}
60
61#[derive(Default)]
62struct StateArenaInner {
63 cells: Vec<StateArenaSlot>,
64 free: Vec<u32>,
65}
66
67#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
68pub struct StateArenaDebugStats {
69 pub cells_len: usize,
70 pub cells_cap: usize,
71 pub free_len: usize,
72 pub free_cap: usize,
73}
74
75#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
76pub struct RuntimeDebugStats {
77 pub node_updates_len: usize,
78 pub node_updates_cap: usize,
79 pub invalid_scopes_len: usize,
80 pub invalid_scopes_cap: usize,
81 pub scope_queue_len: usize,
82 pub scope_queue_cap: usize,
83 pub frame_callbacks_len: usize,
84 pub frame_callbacks_cap: usize,
85 pub local_tasks_len: usize,
86 pub local_tasks_cap: usize,
87 pub ui_conts_len: usize,
88 pub ui_conts_cap: usize,
89 pub tasks_len: usize,
90 pub tasks_cap: usize,
91 pub external_state_owners_len: usize,
92 pub external_state_owners_cap: usize,
93 pub ui_dispatcher_pending: usize,
94}
95
96#[derive(Default)]
97pub(crate) struct StateArena {
98 inner: RefCell<StateArenaInner>,
99}
100
101impl StateArena {
102 pub(crate) fn alloc<T: Clone + 'static>(&self, value: T, runtime: RuntimeHandle) -> StateId {
103 self.alloc_with_policy(value, runtime, Arc::new(NeverEqual))
104 }
105
106 pub(crate) fn alloc_with_policy<T: Clone + 'static>(
107 &self,
108 value: T,
109 runtime: RuntimeHandle,
110 policy: Arc<dyn MutationPolicy<T>>,
111 ) -> StateId {
112 let (slot, generation) = {
113 let mut inner = self.inner.borrow_mut();
114 loop {
115 let Some(slot) = inner.free.pop() else {
116 let slot = inner.cells.len() as u32;
117 inner.cells.push(StateArenaSlot {
118 generation: 0,
119 cell: None,
120 watcher_cell: None,
121 lease: None,
122 });
123 break (slot, 0);
124 };
125
126 let Some(entry) = inner.cells.get_mut(slot as usize) else {
127 continue;
128 };
129 if entry.cell.is_some() {
130 continue;
131 }
132
133 entry.watcher_cell = None;
134 entry.lease = None;
135 entry.generation = entry.generation.wrapping_add(1);
136 break (slot, entry.generation);
137 }
138 };
139 let id = StateId::new(slot, generation);
140 let inner = MutableStateInner::new_with_policy(value, runtime.clone(), policy);
141 inner.install_snapshot_observer(id);
142 let typed_cell = Rc::new(TypedStateCell { inner });
143 let cell: Rc<dyn Any> = typed_cell.clone();
144 let watcher_cell: Rc<dyn ScopeWatchCell> = typed_cell;
145 let mut arena = self.inner.borrow_mut();
146 let slot_entry = &mut arena.cells[slot as usize];
147 slot_entry.cell = Some(cell);
148 slot_entry.watcher_cell = Some(watcher_cell);
149 id
150 }
151
152 fn get_cell_opt(&self, id: StateId) -> Option<Rc<dyn Any>> {
153 self.inner
154 .borrow()
155 .cells
156 .get(id.slot_index())
157 .filter(|cell| cell.generation == id.generation())
158 .and_then(|cell| cell.cell.as_ref())
159 .cloned()
160 }
161
162 fn get_typed<T: Clone + 'static>(&self, id: StateId) -> Rc<TypedStateCell<T>> {
163 match self.get_cell_opt(id) {
164 None => panic!(
165 "state cell missing: slot={}, gen={}, expected={}",
166 id.slot(),
167 id.generation(),
168 std::any::type_name::<T>(),
169 ),
170 Some(cell) => Rc::downcast::<TypedStateCell<T>>(cell).unwrap_or_else(|_| {
171 panic!(
172 "state cell type mismatch: slot={}, gen={}, expected={}",
173 id.slot(),
174 id.generation(),
175 std::any::type_name::<T>(),
176 )
177 }),
178 }
179 }
180
181 fn get_typed_opt<T: Clone + 'static>(&self, id: StateId) -> Option<Rc<TypedStateCell<T>>> {
182 Rc::downcast::<TypedStateCell<T>>(self.get_cell_opt(id)?).ok()
183 }
184
185 pub(crate) fn with_typed<T: Clone + 'static, R>(
186 &self,
187 id: StateId,
188 f: impl FnOnce(&MutableStateInner<T>) -> R,
189 ) -> R {
190 let cell = self.get_typed::<T>(id);
191 f(&cell.inner)
192 }
193
194 pub(crate) fn with_typed_opt<T: Clone + 'static, R>(
195 &self,
196 id: StateId,
197 f: impl FnOnce(&MutableStateInner<T>) -> R,
198 ) -> Option<R> {
199 let cell = self.get_typed_opt::<T>(id)?;
200 Some(f(&cell.inner))
201 }
202
203 pub(crate) fn release(&self, id: StateId) {
204 let cell = {
205 let mut inner = self.inner.borrow_mut();
206 let Some(slot) = inner.cells.get_mut(id.slot_index()) else {
207 return;
208 };
209 if slot.generation != id.generation() {
210 return;
211 }
212 slot.lease = None;
213 slot.watcher_cell = None;
214 let cell = slot.cell.take();
215 if cell.is_some() {
216 inner.free.push(id.slot());
217 }
218 cell
219 };
220 drop(cell);
221 }
222
223 pub(crate) fn stats(&self) -> (usize, usize) {
224 let inner = self.inner.borrow();
225 (inner.cells.len(), inner.free.len())
226 }
227
228 pub(crate) fn debug_stats(&self) -> StateArenaDebugStats {
229 let inner = self.inner.borrow();
230 StateArenaDebugStats {
231 cells_len: inner.cells.len(),
232 cells_cap: inner.cells.capacity(),
233 free_len: inner.free.len(),
234 free_cap: inner.free.capacity(),
235 }
236 }
237
238 pub(crate) fn unregister_scope(&self, id: StateId, scope_id: ScopeId) {
239 let watcher_cell = {
240 let inner = self.inner.borrow();
241 inner
242 .cells
243 .get(id.slot_index())
244 .filter(|slot| slot.generation == id.generation())
245 .and_then(|slot| slot.watcher_cell.as_ref())
246 .cloned()
247 };
248 if let Some(watcher_cell) = watcher_cell {
249 watcher_cell.unregister_scope(scope_id);
250 }
251 }
252
253 pub(crate) fn register_lease(&self, id: StateId, lease: &Rc<StateHandleLease>) {
254 let mut inner = self.inner.borrow_mut();
255 let Some(slot) = inner.cells.get_mut(id.slot_index()) else {
256 return;
257 };
258 if slot.generation != id.generation() {
259 return;
260 }
261 slot.lease = Some(Rc::downgrade(lease));
262 }
263
264 pub(crate) fn retain_lease(&self, id: StateId) -> Option<Rc<StateHandleLease>> {
265 let inner = self.inner.borrow();
266 let slot = inner.cells.get(id.slot_index())?;
267 if slot.generation != id.generation() {
268 return None;
269 }
270 slot.lease.as_ref()?.upgrade()
271 }
272}
273
274#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
275pub struct StateId {
276 slot: u32,
277 generation: u32,
278}
279
280impl StateId {
281 const fn new(slot: u32, generation: u32) -> Self {
282 Self { slot, generation }
283 }
284
285 pub(crate) const fn slot(self) -> u32 {
286 self.slot
287 }
288
289 pub(crate) const fn slot_index(self) -> usize {
290 self.slot as usize
291 }
292
293 pub(crate) const fn generation(self) -> u32 {
294 self.generation
295 }
296}
297
298#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
299pub struct RuntimeId(u32);
300
301impl RuntimeId {
302 fn next() -> Self {
303 NEXT_RUNTIME_ID.with(|next| {
304 let id = next.get();
305 next.set(id.wrapping_add(1));
306 Self(id)
307 })
308 }
309}
310
311struct UiDispatcherInner {
312 scheduler: SchedulerRef,
313 tx: mpsc::Sender<UiMessage>,
314 pending: AtomicUsize,
315}
316
317#[cfg(not(target_arch = "wasm32"))]
327type UiDispatcherRef = Arc<UiDispatcherInner>;
328
329#[cfg(target_arch = "wasm32")]
331type UiDispatcherRef = Rc<UiDispatcherInner>;
332
333impl UiDispatcherInner {
334 fn new(scheduler: SchedulerRef, tx: mpsc::Sender<UiMessage>) -> Self {
335 Self {
336 scheduler,
337 tx,
338 pending: AtomicUsize::new(0),
339 }
340 }
341
342 fn post(&self, task: impl FnOnce() + Send + 'static) {
343 self.pending.fetch_add(1, Ordering::SeqCst);
344 if self.tx.send(UiMessage::Task(Box::new(task))).is_ok() {
345 self.scheduler.schedule_frame();
346 } else {
347 self.pending.fetch_sub(1, Ordering::SeqCst);
348 }
349 }
350
351 fn post_invoke(&self, id: u64, value: Box<dyn Any + Send>) {
352 self.pending.fetch_add(1, Ordering::SeqCst);
353 if self.tx.send(UiMessage::Invoke { id, value }).is_ok() {
354 self.scheduler.schedule_frame();
355 } else {
356 self.pending.fetch_sub(1, Ordering::SeqCst);
357 }
358 }
359
360 fn has_pending(&self) -> bool {
361 self.pending.load(Ordering::SeqCst) > 0
362 }
363}
364
365struct PendingGuard<'a> {
366 counter: &'a AtomicUsize,
367}
368
369impl<'a> PendingGuard<'a> {
370 fn new(counter: &'a AtomicUsize) -> Self {
371 Self { counter }
372 }
373}
374
375impl<'a> Drop for PendingGuard<'a> {
376 fn drop(&mut self) {
377 let mut current = self.counter.load(Ordering::SeqCst);
378 loop {
379 if current == 0 {
380 return;
381 }
382 match self.counter.compare_exchange(
383 current,
384 current - 1,
385 Ordering::SeqCst,
386 Ordering::SeqCst,
387 ) {
388 Ok(_) => return,
389 Err(next) => current = next,
390 }
391 }
392 }
393}
394
395#[derive(Clone)]
396pub struct UiDispatcher {
397 inner: UiDispatcherRef,
398}
399
400impl UiDispatcher {
401 fn new(inner: UiDispatcherRef) -> Self {
402 Self { inner }
403 }
404
405 pub fn post(&self, task: impl FnOnce() + Send + 'static) {
406 self.inner.post(task);
407 }
408
409 pub fn post_invoke<T>(&self, id: u64, value: T)
410 where
411 T: Send + 'static,
412 {
413 self.inner.post_invoke(id, Box::new(value));
414 }
415
416 pub fn has_pending(&self) -> bool {
417 self.inner.has_pending()
418 }
419}
420
421struct RuntimeInner {
422 scheduler: SchedulerRef,
423 needs_frame: RefCell<bool>,
424 node_updates: RefCell<Vec<Command>>,
425 invalid_scopes: RefCell<HashSet<ScopeId>>,
426 scope_queue: RefCell<Vec<(ScopeId, Weak<RecomposeScopeInner>)>>,
427 frame_callbacks: RefCell<VecDeque<FrameCallbackEntry>>,
428 next_frame_callback_id: Cell<u64>,
429 last_frame_time_nanos: Cell<Option<u64>>,
430 ui_dispatcher: UiDispatcherRef,
431 ui_rx: RefCell<mpsc::Receiver<UiMessage>>,
432 local_tasks: RefCell<VecDeque<Box<dyn FnOnce() + 'static>>>,
433 ui_conts: RefCell<UiContinuationMap>,
434 next_cont_id: Cell<u64>,
435 ui_thread_id: ThreadId,
436 tasks: RefCell<Vec<TaskEntry>>,
437 next_task_id: Cell<u64>,
438 state_arena: StateArena,
439 external_state_owners: RefCell<HashMap<StateId, Rc<StateHandleLease>>>,
440 live_recompose_scope_count: Cell<usize>,
441 runtime_id: RuntimeId,
442}
443
444struct TaskEntry {
445 id: u64,
446 label: String,
447 future: Pin<Box<dyn Future<Output = ()> + 'static>>,
448 runnable: Arc<AtomicBool>,
459 waker: Waker,
462}
463
464thread_local! {
465 static NEXT_TASK_LABEL: RefCell<Option<String>> = const { RefCell::new(None) };
466}
467
468pub fn label_next_ui_task(label: impl Into<String>) {
469 NEXT_TASK_LABEL.with(|held| *held.borrow_mut() = Some(label.into()));
470}
471
472impl RuntimeInner {
473 fn new(scheduler: SchedulerRef) -> Self {
474 let (tx, rx) = mpsc::channel();
475 let dispatcher = UiDispatcherRef::new(UiDispatcherInner::new(scheduler.clone(), tx));
476 Self {
477 scheduler,
478 needs_frame: RefCell::new(false),
479 node_updates: RefCell::new(Vec::new()),
480 invalid_scopes: RefCell::new(HashSet::default()),
481 scope_queue: RefCell::new(Vec::new()),
482 frame_callbacks: RefCell::new(VecDeque::new()),
483 next_frame_callback_id: Cell::new(1),
484 last_frame_time_nanos: Cell::new(None),
485 ui_dispatcher: dispatcher,
486 ui_rx: RefCell::new(rx),
487 local_tasks: RefCell::new(VecDeque::new()),
488 ui_conts: RefCell::new(UiContinuationMap::default()),
489 next_cont_id: Cell::new(1),
490 ui_thread_id: std::thread::current().id(),
491 tasks: RefCell::new(Vec::new()),
492 next_task_id: Cell::new(1),
493 state_arena: StateArena::default(),
494 external_state_owners: RefCell::new(HashMap::default()),
495 live_recompose_scope_count: Cell::new(0),
496 runtime_id: RuntimeId::next(),
497 }
498 }
499
500 fn schedule(&self) {
501 *self.needs_frame.borrow_mut() = true;
502 self.scheduler.schedule_frame();
503 }
504
505 fn enqueue_update(&self, command: Command) {
506 self.node_updates.borrow_mut().push(command);
507 self.schedule(); }
509
510 fn take_updates(&self) -> Vec<Command> {
511 let updates = self.node_updates.borrow_mut().drain(..).collect::<Vec<_>>();
512 updates
513 }
514
515 fn has_updates(&self) -> bool {
516 !self.node_updates.borrow().is_empty() || self.has_invalid_scopes()
517 }
518
519 fn register_invalid_scope(&self, id: ScopeId, scope: Weak<RecomposeScopeInner>) {
520 let mut invalid = self.invalid_scopes.borrow_mut();
521 if invalid.insert(id) {
522 self.scope_queue.borrow_mut().push((id, scope));
523 self.schedule();
524 }
525 }
526
527 fn requeue_invalid_scope(&self, id: ScopeId, scope: Weak<RecomposeScopeInner>) {
528 if self.invalid_scopes.borrow().contains(&id) {
529 self.scope_queue.borrow_mut().push((id, scope));
530 self.schedule();
531 }
532 }
533
534 fn mark_scope_recomposed(&self, id: ScopeId) {
535 self.invalid_scopes.borrow_mut().remove(&id);
536 }
537
538 fn take_invalidated_scopes(&self) -> Vec<(ScopeId, Weak<RecomposeScopeInner>)> {
539 let mut queue = self.scope_queue.borrow_mut();
540 if queue.is_empty() {
541 return Vec::new();
542 }
543 let pending: Vec<_> = queue.drain(..).collect();
544 drop(queue);
545 let invalid = self.invalid_scopes.borrow();
546 pending
547 .into_iter()
548 .filter(|(id, _)| invalid.contains(id))
549 .collect()
550 }
551
552 fn has_invalid_scopes(&self) -> bool {
553 !self.invalid_scopes.borrow().is_empty()
554 }
555
556 fn increment_live_recompose_scope_count(&self) {
557 self.live_recompose_scope_count
558 .set(self.live_recompose_scope_count.get().saturating_add(1));
559 }
560
561 fn decrement_live_recompose_scope_count(&self) {
562 self.live_recompose_scope_count
563 .set(self.live_recompose_scope_count.get().saturating_sub(1));
564 }
565
566 fn live_recompose_scope_count(&self) -> usize {
567 self.live_recompose_scope_count.get()
568 }
569
570 fn has_frame_callbacks(&self) -> bool {
571 !self.frame_callbacks.borrow().is_empty()
572 }
573
574 fn has_transient_frame_callbacks(&self) -> bool {
575 self.frame_callbacks
576 .borrow()
577 .iter()
578 .any(|entry| entry.kind == FrameCallbackKind::Transient)
579 }
580
581 fn enqueue_ui_task(&self, task: Box<dyn FnOnce() + 'static>) {
586 self.local_tasks.borrow_mut().push_back(task);
587 self.schedule();
588 }
589
590 fn spawn_ui_task(&self, future: Pin<Box<dyn Future<Output = ()> + 'static>>) -> u64 {
591 let id = self.next_task_id.get();
592 self.next_task_id.set(id + 1);
593 let label = NEXT_TASK_LABEL
594 .with(|held| held.borrow_mut().take())
595 .unwrap_or_else(|| "unnamed".to_string());
596 let runnable = Arc::new(AtomicBool::new(true));
597 let waker = RuntimeTaskWaker::new(self, Arc::clone(&runnable)).into_waker();
598 self.tasks.borrow_mut().push(TaskEntry {
599 id,
600 label,
601 future,
602 runnable,
603 waker,
604 });
605 self.schedule();
606 id
607 }
608
609 fn cancel_task(&self, id: u64) {
610 let mut tasks = self.tasks.borrow_mut();
611 if tasks.iter().any(|entry| entry.id == id) {
612 tasks.retain(|entry| entry.id != id);
613 }
614 }
615
616 fn has_task(&self, id: u64) -> bool {
619 self.tasks
623 .try_borrow()
624 .map(|tasks| tasks.iter().any(|entry| entry.id == id))
625 .unwrap_or(true)
626 }
627
628 fn poll_async_tasks(&self) -> bool {
629 let mut tasks_ref = self.tasks.borrow_mut();
630 let tasks = std::mem::take(&mut *tasks_ref);
631 drop(tasks_ref);
632 let mut pending = Vec::with_capacity(tasks.len());
633 let mut made_progress = false;
634 for mut entry in tasks.into_iter() {
635 if !entry.runnable.swap(false, Ordering::AcqRel) {
639 pending.push(entry);
640 continue;
641 }
642 let mut cx = Context::from_waker(&entry.waker);
643 match entry.future.as_mut().poll(&mut cx) {
644 Poll::Ready(()) => {
645 made_progress = true;
646 }
647 Poll::Pending => {
648 pending.push(entry);
649 }
650 }
651 }
652 if !pending.is_empty() {
653 self.tasks.borrow_mut().extend(pending);
654 }
655 made_progress
656 }
657
658 fn drain_ui(&self) {
659 loop {
660 let mut executed = false;
661
662 {
663 let rx = &mut *self.ui_rx.borrow_mut();
664 for message in rx.try_iter() {
665 executed = true;
666 let _guard = PendingGuard::new(&self.ui_dispatcher.pending);
667 match message {
668 UiMessage::Task(task) => {
669 task();
670 }
671 UiMessage::Invoke { id, value } => {
672 self.invoke_ui_cont(id, value);
673 }
674 }
675 }
676 }
677
678 loop {
679 let task = {
680 let mut local = self.local_tasks.borrow_mut();
681 local.pop_front()
682 };
683
684 match task {
685 Some(task) => {
686 executed = true;
687 task();
688 }
689 None => break,
690 }
691 }
692
693 if self.poll_async_tasks() {
694 executed = true;
695 }
696
697 if !executed {
698 break;
699 }
700 }
701
702 self.clear_needs_frame_if_idle();
707 }
708
709 fn has_pending_ui(&self) -> bool {
710 let local_pending = self
711 .local_tasks
712 .try_borrow()
713 .map(|tasks| !tasks.is_empty())
714 .unwrap_or(true);
715
716 local_pending || self.ui_dispatcher.has_pending() || self.has_runnable_tasks()
717 }
718
719 fn has_runnable_tasks(&self) -> bool {
731 self.tasks
732 .try_borrow()
733 .map(|tasks| {
734 tasks
735 .iter()
736 .any(|task| task.runnable.load(Ordering::Acquire))
737 })
738 .unwrap_or(true)
739 }
740
741 fn register_ui_cont<T: 'static>(&self, f: impl FnOnce(T) + 'static) -> u64 {
742 debug_assert_eq!(
743 std::thread::current().id(),
744 self.ui_thread_id,
745 "UI continuation registered off the runtime thread",
746 );
747 let id = self.next_cont_id.get();
748 self.next_cont_id.set(id + 1);
749 let callback = RefCell::new(Some(f));
750 self.ui_conts.borrow_mut().insert(
751 id,
752 Box::new(move |value: Box<dyn Any>| {
753 let Ok(value) = value.downcast::<T>() else {
754 return false;
755 };
756 let Some(slot) = callback.borrow_mut().take() else {
757 return true;
758 };
759 slot(*value);
760 true
761 }),
762 );
763 id
764 }
765
766 fn invoke_ui_cont(&self, id: u64, value: Box<dyn Any + Send>) {
767 debug_assert_eq!(
768 std::thread::current().id(),
769 self.ui_thread_id,
770 "UI continuation invoked off the runtime thread",
771 );
772 let callback = { self.ui_conts.borrow_mut().remove(&id) };
773 if let Some(callback) = callback {
774 let value: Box<dyn Any> = value;
775 if !callback(value) {
776 self.ui_conts.borrow_mut().insert(id, callback);
777 }
778 }
779 }
780
781 fn cancel_ui_cont(&self, id: u64) {
782 self.ui_conts.borrow_mut().remove(&id);
783 }
784
785 fn register_frame_callback(
786 &self,
787 kind: FrameCallbackKind,
788 callback: Box<dyn FnOnce(u64) + 'static>,
789 ) -> FrameCallbackId {
790 let id = self.next_frame_callback_id.get();
791 self.next_frame_callback_id.set(id + 1);
792 self.frame_callbacks
793 .borrow_mut()
794 .push_back(FrameCallbackEntry {
795 id,
796 kind,
797 callback: Some(callback),
798 });
799 self.schedule();
800 id
801 }
802
803 fn cancel_frame_callback(&self, id: FrameCallbackId) {
804 let mut callbacks = self.frame_callbacks.borrow_mut();
805 if let Some(index) = callbacks.iter().position(|entry| entry.id == id) {
806 callbacks.remove(index);
807 }
808 drop(callbacks);
809 self.clear_needs_frame_if_idle();
810 }
811
812 fn clear_needs_frame_if_idle(&self) {
819 if !self.has_invalid_scopes()
820 && !self.has_updates()
821 && !self.has_frame_callbacks()
822 && !self.has_pending_ui()
823 {
824 *self.needs_frame.borrow_mut() = false;
825 }
826 }
827
828 fn drain_frame_callbacks(&self, frame_time_nanos: u64) {
829 self.last_frame_time_nanos.set(Some(frame_time_nanos));
830 let mut callbacks = self.frame_callbacks.borrow_mut();
831 let mut pending: Vec<Box<dyn FnOnce(u64) + 'static>> = Vec::with_capacity(callbacks.len());
832 while let Some(mut entry) = callbacks.pop_front() {
833 if let Some(callback) = entry.callback.take() {
834 pending.push(callback);
835 }
836 }
837 drop(callbacks);
838
839 if !pending.is_empty() {
844 let _ = crate::run_in_mutable_snapshot(|| {
845 for callback in pending {
846 callback(frame_time_nanos);
847 }
848 });
849 }
850
851 self.clear_needs_frame_if_idle();
852 }
853
854 fn debug_stats(&self) -> RuntimeDebugStats {
855 let node_updates = self.node_updates.borrow();
856 let invalid_scopes = self.invalid_scopes.borrow();
857 let scope_queue = self.scope_queue.borrow();
858 let frame_callbacks = self.frame_callbacks.borrow();
859 let local_tasks = self.local_tasks.borrow();
860 let ui_conts = self.ui_conts.borrow();
861 let tasks = self.tasks.borrow();
862 let external_state_owners = self.external_state_owners.borrow();
863
864 RuntimeDebugStats {
865 node_updates_len: node_updates.len(),
866 node_updates_cap: node_updates.capacity(),
867 invalid_scopes_len: invalid_scopes.len(),
868 invalid_scopes_cap: invalid_scopes.capacity(),
869 scope_queue_len: scope_queue.len(),
870 scope_queue_cap: scope_queue.capacity(),
871 frame_callbacks_len: frame_callbacks.len(),
872 frame_callbacks_cap: frame_callbacks.capacity(),
873 local_tasks_len: local_tasks.len(),
874 local_tasks_cap: local_tasks.capacity(),
875 ui_conts_len: ui_conts.len(),
876 ui_conts_cap: ui_conts.capacity(),
877 tasks_len: tasks.len(),
878 tasks_cap: tasks.capacity(),
879 external_state_owners_len: external_state_owners.len(),
880 external_state_owners_cap: external_state_owners.capacity(),
881 ui_dispatcher_pending: self.ui_dispatcher.pending.load(Ordering::SeqCst),
882 }
883 }
884}
885
886#[derive(Clone)]
887pub struct Runtime {
888 inner: Rc<RuntimeInner>,
889}
890
891impl Runtime {
892 pub fn new(scheduler: SchedulerRef) -> Self {
893 let inner = Rc::new(RuntimeInner::new(scheduler));
894 let runtime = Self { inner };
895 let handle = runtime.handle();
896 register_runtime_handle(&handle);
897 LAST_RUNTIME.with(|slot| *slot.borrow_mut() = Some(handle));
898 runtime
899 }
900
901 pub fn handle(&self) -> RuntimeHandle {
902 RuntimeHandle {
903 inner: Rc::downgrade(&self.inner),
904 dispatcher: UiDispatcher::new(self.inner.ui_dispatcher.clone()),
905 ui_thread_id: self.inner.ui_thread_id,
906 id: self.inner.runtime_id,
907 }
908 }
909
910 pub fn has_updates(&self) -> bool {
911 self.inner.has_updates()
912 }
913
914 pub fn needs_frame(&self) -> bool {
915 *self.inner.needs_frame.borrow() || self.inner.has_runnable_tasks()
924 }
925
926 pub fn set_needs_frame(&self, value: bool) {
927 *self.inner.needs_frame.borrow_mut() = value;
928 }
929
930 pub fn last_frame_time_nanos(&self) -> Option<u64> {
935 self.inner.last_frame_time_nanos.get()
936 }
937
938 #[cfg(any(feature = "internal", test))]
939 pub fn frame_clock(&self) -> FrameClock {
940 FrameClock::new(self.handle())
941 }
942}
943
944impl Drop for Runtime {
945 fn drop(&mut self) {
946 if Rc::strong_count(&self.inner) != 1 {
947 return;
948 }
949 unregister_runtime_handle(self.inner.runtime_id);
950 LAST_RUNTIME.with(|slot| {
951 let should_clear = slot
952 .borrow()
953 .as_ref()
954 .is_some_and(|handle| handle.id() == self.inner.runtime_id);
955 if should_clear {
956 *slot.borrow_mut() = None;
957 }
958 });
959 }
960}
961
962#[derive(Default)]
963pub struct DefaultScheduler;
964
965impl RuntimeScheduler for DefaultScheduler {
966 fn schedule_frame(&self) {}
967}
968
969#[cfg(test)]
970#[derive(Default)]
971pub struct TestScheduler;
972
973#[cfg(test)]
974impl RuntimeScheduler for TestScheduler {
975 fn schedule_frame(&self) {}
976}
977
978#[cfg(test)]
979pub struct TestRuntime {
980 runtime: Runtime,
981}
982
983#[cfg(test)]
984impl Default for TestRuntime {
985 fn default() -> Self {
986 Self::new()
987 }
988}
989
990#[cfg(test)]
991impl TestRuntime {
992 pub fn new() -> Self {
993 Self {
994 runtime: Runtime::new(Arc::new(TestScheduler)),
995 }
996 }
997
998 pub fn handle(&self) -> RuntimeHandle {
999 self.runtime.handle()
1000 }
1001}
1002
1003#[derive(Clone)]
1004pub struct RuntimeHandle {
1005 inner: Weak<RuntimeInner>,
1006 dispatcher: UiDispatcher,
1007 ui_thread_id: ThreadId,
1008 id: RuntimeId,
1009}
1010
1011pub struct TaskHandle {
1012 id: u64,
1013 runtime: RuntimeHandle,
1014}
1015
1016struct DeferredStateRelease {
1017 runtime: RuntimeHandle,
1018 id: StateId,
1019}
1020
1021pub(crate) struct StateHandleLease {
1022 id: StateId,
1023 runtime: RuntimeHandle,
1024}
1025
1026impl StateHandleLease {
1027 pub(crate) fn id(&self) -> StateId {
1028 self.id
1029 }
1030
1031 pub(crate) fn runtime(&self) -> RuntimeHandle {
1032 self.runtime.clone()
1033 }
1034}
1035
1036impl Drop for StateHandleLease {
1037 fn drop(&mut self) {
1038 defer_state_release(self.runtime.clone(), self.id);
1039 }
1040}
1041
1042impl RuntimeHandle {
1043 pub fn id(&self) -> RuntimeId {
1044 self.id
1045 }
1046
1047 pub(crate) fn alloc_state<T: Clone + 'static>(&self, value: T) -> Rc<StateHandleLease> {
1048 let id = self.with_state_arena(|arena| arena.alloc(value, self.clone()));
1049 let lease = Rc::new(StateHandleLease {
1050 id,
1051 runtime: self.clone(),
1052 });
1053 self.with_state_arena(|arena| arena.register_lease(id, &lease));
1054 lease
1055 }
1056
1057 pub(crate) fn alloc_state_with_policy<T: Clone + 'static>(
1058 &self,
1059 value: T,
1060 policy: Arc<dyn MutationPolicy<T>>,
1061 ) -> Rc<StateHandleLease> {
1062 let id =
1063 self.with_state_arena(|arena| arena.alloc_with_policy(value, self.clone(), policy));
1064 let lease = Rc::new(StateHandleLease {
1065 id,
1066 runtime: self.clone(),
1067 });
1068 self.with_state_arena(|arena| arena.register_lease(id, &lease));
1069 lease
1070 }
1071
1072 pub(crate) fn alloc_persistent_state<T: Clone + 'static>(
1073 &self,
1074 value: T,
1075 ) -> crate::MutableState<T> {
1076 let lease = self.alloc_state(value);
1077 if let Some(inner) = self.inner.upgrade() {
1078 inner
1079 .external_state_owners
1080 .borrow_mut()
1081 .insert(lease.id(), Rc::clone(&lease));
1082 }
1083 crate::MutableState::from_lease(&lease)
1084 }
1085
1086 pub(crate) fn retain_state_lease(&self, id: StateId) -> Option<Rc<StateHandleLease>> {
1087 self.with_state_arena(|arena| arena.retain_lease(id))
1088 }
1089
1090 pub(crate) fn with_state_arena<R>(&self, f: impl FnOnce(&StateArena) -> R) -> R {
1091 self.try_with_state_arena(f)
1092 .unwrap_or_else(|| panic!("runtime dropped"))
1093 }
1094
1095 pub(crate) fn try_with_state_arena<R>(&self, f: impl FnOnce(&StateArena) -> R) -> Option<R> {
1096 self.inner.upgrade().map(|inner| f(&inner.state_arena))
1097 }
1098
1099 fn release_state_immediate(&self, id: StateId) {
1100 if let Some(inner) = self.inner.upgrade() {
1101 inner.state_arena.release(id);
1102 }
1103 }
1104
1105 pub fn state_arena_stats(&self) -> (usize, usize) {
1106 self.try_with_state_arena(StateArena::stats)
1107 .unwrap_or_default()
1108 }
1109
1110 pub fn state_arena_debug_stats(&self) -> StateArenaDebugStats {
1111 self.try_with_state_arena(StateArena::debug_stats)
1112 .unwrap_or_default()
1113 }
1114
1115 pub fn debug_stats(&self) -> RuntimeDebugStats {
1116 self.inner
1117 .upgrade()
1118 .map(|inner| inner.debug_stats())
1119 .unwrap_or_default()
1120 }
1121
1122 pub fn live_ui_task_labels(&self) -> Vec<(u64, String)> {
1123 self.inner
1124 .upgrade()
1125 .map(|inner| {
1126 inner
1127 .tasks
1128 .borrow()
1129 .iter()
1130 .map(|entry| (entry.id, entry.label.clone()))
1131 .collect()
1132 })
1133 .unwrap_or_default()
1134 }
1135
1136 pub(crate) fn unregister_state_scope(&self, id: StateId, scope_id: ScopeId) {
1137 if let Some(inner) = self.inner.upgrade() {
1138 inner.state_arena.unregister_scope(id, scope_id);
1139 }
1140 }
1141
1142 pub fn schedule(&self) {
1143 if let Some(inner) = self.inner.upgrade() {
1144 inner.schedule();
1145 }
1146 }
1147
1148 pub(crate) fn enqueue_node_update(&self, command: Command) {
1149 if let Some(inner) = self.inner.upgrade() {
1150 inner.enqueue_update(command);
1151 }
1152 }
1153
1154 pub fn enqueue_ui_task(&self, task: Box<dyn FnOnce() + 'static>) {
1161 if let Some(inner) = self.inner.upgrade() {
1162 inner.enqueue_ui_task(task);
1163 } else {
1164 task();
1165 }
1166 }
1167
1168 pub fn spawn_ui<F>(&self, fut: F) -> Option<TaskHandle>
1169 where
1170 F: Future<Output = ()> + 'static,
1171 {
1172 self.inner.upgrade().map(|inner| {
1173 let id = inner.spawn_ui_task(Box::pin(fut));
1174 TaskHandle {
1175 id,
1176 runtime: self.clone(),
1177 }
1178 })
1179 }
1180
1181 pub fn cancel_task(&self, id: u64) {
1182 if let Some(inner) = self.inner.upgrade() {
1183 inner.cancel_task(id);
1184 }
1185 }
1186
1187 pub fn has_task(&self, id: u64) -> bool {
1189 self.inner
1190 .upgrade()
1191 .map(|inner| inner.has_task(id))
1192 .unwrap_or(false)
1193 }
1194
1195 pub fn post_ui(&self, task: impl FnOnce() + Send + 'static) {
1200 self.dispatcher.post(task);
1201 }
1202
1203 pub fn register_ui_cont<T: 'static>(&self, f: impl FnOnce(T) + 'static) -> Option<u64> {
1204 self.inner.upgrade().map(|inner| inner.register_ui_cont(f))
1205 }
1206
1207 pub fn cancel_ui_cont(&self, id: u64) {
1208 if let Some(inner) = self.inner.upgrade() {
1209 inner.cancel_ui_cont(id);
1210 }
1211 }
1212
1213 pub fn drain_ui(&self) {
1214 if let Some(inner) = self.inner.upgrade() {
1215 inner.drain_ui();
1216 }
1217 }
1218
1219 pub fn has_pending_ui(&self) -> bool {
1220 self.inner
1221 .upgrade()
1222 .map(|inner| inner.has_pending_ui())
1223 .unwrap_or_else(|| self.dispatcher.has_pending())
1224 }
1225
1226 pub fn register_frame_callback(
1227 &self,
1228 callback: impl FnOnce(u64) + 'static,
1229 ) -> Option<FrameCallbackId> {
1230 self.inner.upgrade().map(|inner| {
1231 inner.register_frame_callback(FrameCallbackKind::Transient, Box::new(callback))
1232 })
1233 }
1234
1235 pub fn register_perpetual_frame_callback(
1236 &self,
1237 callback: impl FnOnce(u64) + 'static,
1238 ) -> Option<FrameCallbackId> {
1239 self.inner.upgrade().map(|inner| {
1240 inner.register_frame_callback(FrameCallbackKind::Perpetual, Box::new(callback))
1241 })
1242 }
1243
1244 pub fn cancel_frame_callback(&self, id: FrameCallbackId) {
1245 if let Some(inner) = self.inner.upgrade() {
1246 inner.cancel_frame_callback(id);
1247 }
1248 }
1249
1250 pub fn drain_frame_callbacks(&self, frame_time_nanos: u64) {
1251 if let Some(inner) = self.inner.upgrade() {
1252 inner.drain_frame_callbacks(frame_time_nanos);
1253 }
1254 }
1255
1256 pub fn last_frame_time_nanos(&self) -> Option<u64> {
1259 self.inner
1260 .upgrade()
1261 .and_then(|inner| inner.last_frame_time_nanos.get())
1262 }
1263
1264 #[cfg(any(feature = "internal", test))]
1265 pub fn frame_clock(&self) -> FrameClock {
1266 FrameClock::new(self.clone())
1267 }
1268
1269 pub fn set_needs_frame(&self, value: bool) {
1270 if let Some(inner) = self.inner.upgrade() {
1271 *inner.needs_frame.borrow_mut() = value;
1272 }
1273 }
1274
1275 pub(crate) fn take_updates(&self) -> Vec<Command> {
1276 self.inner
1277 .upgrade()
1278 .map(|inner| inner.take_updates())
1279 .unwrap_or_default()
1280 }
1281
1282 pub fn has_updates(&self) -> bool {
1283 self.inner
1284 .upgrade()
1285 .map(|inner| inner.has_updates())
1286 .unwrap_or(false)
1287 }
1288
1289 pub(crate) fn mark_scope_recomposed(&self, id: ScopeId) {
1290 if let Some(inner) = self.inner.upgrade() {
1291 inner.mark_scope_recomposed(id);
1292 }
1293 }
1294
1295 pub(crate) fn register_invalid_scope(&self, id: ScopeId, scope: Weak<RecomposeScopeInner>) {
1296 if let Some(inner) = self.inner.upgrade() {
1297 inner.register_invalid_scope(id, scope);
1298 }
1299 }
1300
1301 pub(crate) fn requeue_invalid_scope(&self, id: ScopeId, scope: Weak<RecomposeScopeInner>) {
1302 if let Some(inner) = self.inner.upgrade() {
1303 inner.requeue_invalid_scope(id, scope);
1304 }
1305 }
1306
1307 pub(crate) fn take_invalidated_scopes(&self) -> Vec<(ScopeId, Weak<RecomposeScopeInner>)> {
1308 self.inner
1309 .upgrade()
1310 .map(|inner| inner.take_invalidated_scopes())
1311 .unwrap_or_default()
1312 }
1313
1314 pub fn has_invalid_scopes(&self) -> bool {
1315 self.inner
1316 .upgrade()
1317 .map(|inner| inner.has_invalid_scopes())
1318 .unwrap_or(false)
1319 }
1320
1321 pub(crate) fn increment_live_recompose_scope_count(&self) {
1322 if let Some(inner) = self.inner.upgrade() {
1323 inner.increment_live_recompose_scope_count();
1324 }
1325 }
1326
1327 pub(crate) fn decrement_live_recompose_scope_count(&self) {
1328 if let Some(inner) = self.inner.upgrade() {
1329 inner.decrement_live_recompose_scope_count();
1330 }
1331 }
1332
1333 fn live_recompose_scope_count(&self) -> usize {
1334 self.inner
1335 .upgrade()
1336 .map(|inner| inner.live_recompose_scope_count())
1337 .unwrap_or_default()
1338 }
1339
1340 #[doc(hidden)]
1341 pub fn debug_invalid_scope_ids(&self) -> Vec<usize> {
1342 self.inner
1343 .upgrade()
1344 .map(|inner| inner.invalid_scopes.borrow().iter().copied().collect())
1345 .unwrap_or_default()
1346 }
1347
1348 pub fn has_frame_callbacks(&self) -> bool {
1349 self.inner
1350 .upgrade()
1351 .map(|inner| inner.has_frame_callbacks())
1352 .unwrap_or(false)
1353 }
1354
1355 pub fn has_transient_frame_callbacks(&self) -> bool {
1356 self.inner
1357 .upgrade()
1358 .map(|inner| inner.has_transient_frame_callbacks())
1359 .unwrap_or(false)
1360 }
1361
1362 pub fn assert_ui_thread(&self) {
1363 debug_assert_eq!(
1364 std::thread::current().id(),
1365 self.ui_thread_id,
1366 "state mutated off the runtime's UI thread"
1367 );
1368 }
1369
1370 pub fn dispatcher(&self) -> UiDispatcher {
1371 self.dispatcher.clone()
1372 }
1373
1374 #[doc(hidden)]
1375 pub fn with_deferred_state_releases<R>(&self, f: impl FnOnce() -> R) -> R {
1376 let _scope = enter_state_teardown_scope();
1377 f()
1378 }
1379}
1380
1381impl TaskHandle {
1382 pub fn cancel(&self) {
1383 self.runtime.cancel_task(self.id);
1384 }
1385
1386 pub fn is_finished(&self) -> bool {
1388 !self.runtime.has_task(self.id)
1389 }
1390}
1391
1392pub(crate) struct FrameCallbackEntry {
1393 id: FrameCallbackId,
1394 kind: FrameCallbackKind,
1395 callback: Option<Box<dyn FnOnce(u64) + 'static>>,
1396}
1397
1398#[cfg(not(target_arch = "wasm32"))]
1410struct RuntimeTaskWaker {
1411 scheduler: SchedulerRef,
1412 runnable: Arc<AtomicBool>,
1413}
1414
1415#[cfg(target_arch = "wasm32")]
1416struct RuntimeTaskWaker {
1417 runtime_id: RuntimeId,
1418 runnable: Arc<AtomicBool>,
1419}
1420
1421impl RuntimeTaskWaker {
1422 #[cfg(not(target_arch = "wasm32"))]
1423 fn new(inner: &RuntimeInner, runnable: Arc<AtomicBool>) -> Self {
1424 let scheduler = inner.scheduler.clone();
1425 Self {
1426 scheduler,
1427 runnable,
1428 }
1429 }
1430
1431 #[cfg(target_arch = "wasm32")]
1432 fn new(inner: &RuntimeInner, runnable: Arc<AtomicBool>) -> Self {
1433 let runtime_id = inner.runtime_id;
1434 Self {
1435 runtime_id,
1436 runnable,
1437 }
1438 }
1439
1440 fn into_waker(self) -> Waker {
1441 futures_task::waker(Arc::new(self))
1442 }
1443}
1444
1445impl futures_task::ArcWake for RuntimeTaskWaker {
1446 #[cfg(not(target_arch = "wasm32"))]
1447 fn wake_by_ref(arc_self: &Arc<Self>) {
1448 arc_self.runnable.store(true, Ordering::Release);
1449 arc_self.scheduler.schedule_frame();
1450 }
1451
1452 #[cfg(target_arch = "wasm32")]
1453 fn wake_by_ref(arc_self: &Arc<Self>) {
1454 arc_self.runnable.store(true, Ordering::Release);
1455 REGISTERED_RUNTIMES.with(|registry| {
1456 if let Some(handle) = registry.borrow().get(&arc_self.runtime_id).cloned() {
1457 handle.schedule();
1458 }
1459 });
1460 }
1461}
1462
1463thread_local! {
1464 static NEXT_RUNTIME_ID: Cell<u32> = const { Cell::new(1) };
1465 static ACTIVE_RUNTIMES: RefCell<Vec<RuntimeHandle>> = const { RefCell::new(Vec::new()) };
1466 static LAST_RUNTIME: RefCell<Option<RuntimeHandle>> = const { RefCell::new(None) };
1467 static REGISTERED_RUNTIMES: RefCell<HashMap<RuntimeId, RuntimeHandle>> = RefCell::new(HashMap::default());
1468 static STATE_TEARDOWN_DEPTH: Cell<usize> = const { Cell::new(0) };
1469 static DEFERRED_STATE_RELEASES: RefCell<Vec<DeferredStateRelease>> = const { RefCell::new(Vec::new()) };
1470}
1471
1472#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1473pub struct RuntimeThreadLocalDebugStats {
1474 pub active_runtimes_len: usize,
1475 pub active_runtimes_cap: usize,
1476 pub registered_runtimes_len: usize,
1477 pub registered_runtimes_cap: usize,
1478 pub deferred_state_releases_len: usize,
1479 pub deferred_state_releases_cap: usize,
1480}
1481
1482pub fn current_runtime_handle() -> Option<RuntimeHandle> {
1487 if let Some(handle) = ACTIVE_RUNTIMES.with(|stack| stack.borrow().last().cloned()) {
1488 return Some(handle);
1489 }
1490 LAST_RUNTIME.with(|slot| slot.borrow().clone())
1491}
1492
1493pub(crate) fn runtime_handle_by_id(id: RuntimeId) -> Option<RuntimeHandle> {
1494 REGISTERED_RUNTIMES.with(|registry| registry.borrow().get(&id).cloned())
1495}
1496
1497pub(crate) fn live_recompose_scope_count() -> usize {
1498 REGISTERED_RUNTIMES.with(|registry| {
1499 registry
1500 .borrow()
1501 .values()
1502 .map(RuntimeHandle::live_recompose_scope_count)
1503 .sum()
1504 })
1505}
1506
1507pub fn debug_runtime_thread_local_stats() -> RuntimeThreadLocalDebugStats {
1508 let (active_runtimes_len, active_runtimes_cap) = ACTIVE_RUNTIMES.with(|stack| {
1509 let stack = stack.borrow();
1510 (stack.len(), stack.capacity())
1511 });
1512 let (registered_runtimes_len, registered_runtimes_cap) = REGISTERED_RUNTIMES.with(|registry| {
1513 let registry = registry.borrow();
1514 (registry.len(), registry.capacity())
1515 });
1516 let (deferred_state_releases_len, deferred_state_releases_cap) =
1517 DEFERRED_STATE_RELEASES.with(|releases| {
1518 let releases = releases.borrow();
1519 (releases.len(), releases.capacity())
1520 });
1521
1522 RuntimeThreadLocalDebugStats {
1523 active_runtimes_len,
1524 active_runtimes_cap,
1525 registered_runtimes_len,
1526 registered_runtimes_cap,
1527 deferred_state_releases_len,
1528 deferred_state_releases_cap,
1529 }
1530}
1531
1532fn register_runtime_handle(handle: &RuntimeHandle) {
1533 REGISTERED_RUNTIMES.with(|registry| {
1534 registry.borrow_mut().insert(handle.id(), handle.clone());
1535 });
1536}
1537
1538fn unregister_runtime_handle(id: RuntimeId) {
1539 REGISTERED_RUNTIMES.with(|registry| {
1540 registry.borrow_mut().remove(&id);
1541 });
1542}
1543
1544fn defer_state_release(runtime: RuntimeHandle, id: StateId) {
1545 let teardown_active = STATE_TEARDOWN_DEPTH.with(|depth| depth.get() > 0);
1546 if teardown_active {
1547 DEFERRED_STATE_RELEASES.with(|releases| {
1548 releases
1549 .borrow_mut()
1550 .push(DeferredStateRelease { runtime, id });
1551 });
1552 } else {
1553 runtime.release_state_immediate(id);
1554 }
1555}
1556
1557fn flush_deferred_state_releases() {
1558 DEFERRED_STATE_RELEASES.with(|releases| {
1559 let mut releases = releases.borrow_mut();
1560 while let Some(deferred) = releases.pop() {
1561 deferred.runtime.release_state_immediate(deferred.id);
1562 }
1563 });
1564}
1565
1566pub(crate) struct StateTeardownScope;
1567
1568pub(crate) fn enter_state_teardown_scope() -> StateTeardownScope {
1569 STATE_TEARDOWN_DEPTH.with(|depth| depth.set(depth.get() + 1));
1570 StateTeardownScope
1571}
1572
1573impl Drop for StateTeardownScope {
1574 fn drop(&mut self) {
1575 STATE_TEARDOWN_DEPTH.with(|depth| {
1576 let next = depth.get().saturating_sub(1);
1577 depth.set(next);
1578 if next == 0 {
1579 flush_deferred_state_releases();
1580 }
1581 });
1582 }
1583}
1584
1585pub(crate) fn push_active_runtime(handle: &RuntimeHandle) {
1586 register_runtime_handle(handle);
1587 ACTIVE_RUNTIMES.with(|stack| stack.borrow_mut().push(handle.clone()));
1588 LAST_RUNTIME.with(|slot| *slot.borrow_mut() = Some(handle.clone()));
1589}
1590
1591pub(crate) fn pop_active_runtime() {
1592 ACTIVE_RUNTIMES.with(|stack| {
1593 stack.borrow_mut().pop();
1594 });
1595}
1596
1597pub fn schedule_frame() {
1599 if let Some(handle) = current_runtime_handle() {
1600 handle.schedule();
1601 return;
1602 }
1603 log::debug!(
1604 target: "cranpose::runtime",
1605 "ignoring frame request without an active runtime",
1606 );
1607}
1608
1609pub fn schedule_node_update(
1611 update: impl FnOnce(&mut dyn Applier) -> Result<(), NodeError> + 'static,
1612) {
1613 if let Some(handle) = current_runtime_handle() {
1614 handle.enqueue_node_update(Command::callback(update));
1615 } else {
1616 drop(update);
1617 log::debug!(
1618 target: "cranpose::runtime",
1619 "ignoring node update request without an active runtime",
1620 );
1621 }
1622}
1623
1624#[cfg(test)]
1625mod tests {
1626 use super::*;
1627
1628 #[test]
1629 fn state_arena_alloc_skips_free_slot_outside_cell_storage() {
1630 let runtime = TestRuntime::new();
1631 let arena = StateArena::default();
1632 let first = arena.alloc(1_i32, runtime.handle());
1633 arena.inner.borrow_mut().free.push(u32::MAX);
1634
1635 let second = arena.alloc(2_i32, runtime.handle());
1636
1637 assert_eq!(first.slot(), 0);
1638 assert_eq!(second.slot(), 1);
1639 assert!(arena.get_typed_opt::<i32>(first).is_some());
1640 assert!(arena.get_typed_opt::<i32>(second).is_some());
1641 }
1642
1643 #[test]
1644 fn state_arena_alloc_skips_occupied_free_slot() {
1645 let runtime = TestRuntime::new();
1646 let arena = StateArena::default();
1647 let first = arena.alloc(1_i32, runtime.handle());
1648 arena.inner.borrow_mut().free.push(first.slot());
1649
1650 let second = arena.alloc(2_i32, runtime.handle());
1651
1652 assert_ne!(first.slot(), second.slot());
1653 assert_eq!(second.slot(), 1);
1654 assert!(arena.get_typed_opt::<i32>(first).is_some());
1655 assert!(arena.get_typed_opt::<i32>(second).is_some());
1656 }
1657
1658 #[test]
1659 fn state_arena_register_lease_ignores_stale_id() {
1660 let runtime = TestRuntime::new();
1661 let arena = StateArena::default();
1662 let stale_id = StateId::new(99, 0);
1663 let lease = Rc::new(StateHandleLease {
1664 id: stale_id,
1665 runtime: runtime.handle(),
1666 });
1667
1668 arena.register_lease(stale_id, &lease);
1669
1670 assert!(arena.retain_lease(stale_id).is_none());
1671 }
1672
1673 #[test]
1674 fn ui_continuation_type_mismatch_is_ignored_until_matching_payload() {
1675 let runtime = TestRuntime::new();
1676 let handle = runtime.handle();
1677 let received = Rc::new(Cell::new(None));
1678 let received_for_continuation = Rc::clone(&received);
1679 let cont_id = handle
1680 .register_ui_cont(move |value: u32| {
1681 received_for_continuation.set(Some(value));
1682 })
1683 .expect("test runtime is alive");
1684
1685 handle.dispatcher().post_invoke(cont_id, "wrong payload");
1686 handle.drain_ui();
1687
1688 assert_eq!(received.get(), None);
1689 assert_eq!(handle.debug_stats().ui_conts_len, 1);
1690
1691 handle.dispatcher().post_invoke(cont_id, 42_u32);
1692 handle.drain_ui();
1693
1694 assert_eq!(received.get(), Some(42));
1695 assert_eq!(handle.debug_stats().ui_conts_len, 0);
1696 }
1697
1698 #[test]
1699 fn ui_dispatcher_failed_send_does_not_leave_pending_work() {
1700 let runtime = Runtime::new(Arc::new(TestScheduler));
1701 let dispatcher = runtime.handle().dispatcher();
1702
1703 drop(runtime);
1704
1705 assert!(!dispatcher.has_pending());
1706 dispatcher.post(|| {});
1707 assert!(!dispatcher.has_pending());
1708 dispatcher.post_invoke(404, 12_u32);
1709 assert!(!dispatcher.has_pending());
1710 }
1711
1712 #[test]
1713 fn pending_guard_does_not_wrap_on_underflow() {
1714 let counter = AtomicUsize::new(0);
1715
1716 {
1717 let _guard = PendingGuard::new(&counter);
1718 }
1719
1720 assert_eq!(counter.load(Ordering::SeqCst), 0);
1721 }
1722
1723 #[test]
1724 fn pending_guard_decrements_pending_count() {
1725 let counter = AtomicUsize::new(2);
1726
1727 {
1728 let _guard = PendingGuard::new(&counter);
1729 }
1730
1731 assert_eq!(counter.load(Ordering::SeqCst), 1);
1732 }
1733
1734 #[test]
1735 fn schedule_frame_without_runtime_is_ignored() {
1736 let ok = std::thread::spawn(|| std::panic::catch_unwind(super::schedule_frame).is_ok())
1737 .join()
1738 .expect("test thread should join");
1739
1740 assert!(ok);
1741 }
1742
1743 #[test]
1744 fn schedule_node_update_without_runtime_is_ignored() {
1745 let ok = std::thread::spawn(|| {
1746 std::panic::catch_unwind(|| {
1747 super::schedule_node_update(|_| Ok(()));
1748 })
1749 .is_ok()
1750 })
1751 .join()
1752 .expect("test thread should join");
1753
1754 assert!(ok);
1755 }
1756}