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 Arc,
10 atomic::{AtomicBool, AtomicUsize, Ordering},
11 mpsc,
12 },
13 task::{Context, Poll, Waker},
14 thread::ThreadId,
15 thread_local,
16};
17
18#[cfg(any(feature = "internal", test))]
19use crate::frame_clock::FrameClock;
20use crate::{
21 Applier, Command, FrameCallbackId, MutableStateInner, NodeError, RecomposeScopeInner, ScopeId,
22 collections::map::{HashMap, HashSet},
23 platform::{RuntimeScheduler, SchedulerRef},
24 state::{MutationPolicy, NeverEqual},
25};
26
27#[derive(Clone, Copy, PartialEq, Eq)]
28pub(crate) enum FrameCallbackKind {
29 Transient,
30 Perpetual,
31}
32
33enum UiMessage {
34 Task(Box<dyn FnOnce() + Send + 'static>),
35 Invoke { id: u64, value: Box<dyn Any + Send> },
36}
37
38type UiContinuation = Box<dyn Fn(Box<dyn Any>) -> bool + 'static>;
39type UiContinuationMap = HashMap<u64, UiContinuation>;
40
41struct TypedStateCell<T: Clone + 'static> {
42 inner: MutableStateInner<T>,
43}
44
45trait ScopeWatchCell {
46 fn unregister_scope(&self, scope_id: ScopeId);
47}
48
49impl<T: Clone + 'static> ScopeWatchCell for TypedStateCell<T> {
50 fn unregister_scope(&self, scope_id: ScopeId) {
51 self.inner.unregister_scope(scope_id);
52 }
53}
54
55struct StateArenaSlot {
56 generation: u32,
57 cell: Option<Rc<dyn Any>>,
58 watcher_cell: Option<Rc<dyn ScopeWatchCell>>,
59 lease: Option<Weak<StateHandleLease>>,
60}
61
62#[derive(Default)]
63struct StateArenaInner {
64 cells: Vec<StateArenaSlot>,
65 free: Vec<u32>,
66}
67
68#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
69pub struct StateArenaDebugStats {
70 pub cells_len: usize,
71 pub cells_cap: usize,
72 pub free_len: usize,
73 pub free_cap: usize,
74}
75
76#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
77pub struct RuntimeDebugStats {
78 pub node_updates_len: usize,
79 pub node_updates_cap: usize,
80 pub invalid_scopes_len: usize,
81 pub invalid_scopes_cap: usize,
82 pub scope_queue_len: usize,
83 pub scope_queue_cap: usize,
84 pub frame_callbacks_len: usize,
85 pub frame_callbacks_cap: usize,
86 pub local_tasks_len: usize,
87 pub local_tasks_cap: usize,
88 pub ui_conts_len: usize,
89 pub ui_conts_cap: usize,
90 pub tasks_len: usize,
91 pub tasks_cap: usize,
92 pub external_state_owners_len: usize,
93 pub external_state_owners_cap: usize,
94 pub ui_dispatcher_pending: usize,
95}
96
97#[derive(Default)]
98pub(crate) struct StateArena {
99 inner: RefCell<StateArenaInner>,
100}
101
102impl StateArena {
103 pub(crate) fn alloc<T: Clone + 'static>(&self, value: T, runtime: RuntimeHandle) -> StateId {
104 self.alloc_with_policy(value, runtime, Arc::new(NeverEqual))
105 }
106
107 pub(crate) fn alloc_with_policy<T: Clone + 'static>(
108 &self,
109 value: T,
110 runtime: RuntimeHandle,
111 policy: Arc<dyn MutationPolicy<T>>,
112 ) -> StateId {
113 let (slot, generation) = {
114 let mut inner = self.inner.borrow_mut();
115 loop {
116 let Some(slot) = inner.free.pop() else {
117 let slot = inner.cells.len() as u32;
118 inner.cells.push(StateArenaSlot {
119 generation: 0,
120 cell: None,
121 watcher_cell: None,
122 lease: None,
123 });
124 break (slot, 0);
125 };
126
127 let Some(entry) = inner.cells.get_mut(slot as usize) else {
128 continue;
129 };
130 if entry.cell.is_some() {
131 continue;
132 }
133
134 entry.watcher_cell = None;
135 entry.lease = None;
136 entry.generation = entry.generation.wrapping_add(1);
137 break (slot, entry.generation);
138 }
139 };
140 let id = StateId::new(slot, generation);
141 let inner = MutableStateInner::new_with_policy(value, runtime.clone(), policy);
142 inner.install_snapshot_observer(id);
143 let typed_cell = Rc::new(TypedStateCell { inner });
144 let cell: Rc<dyn Any> = typed_cell.clone();
145 let watcher_cell: Rc<dyn ScopeWatchCell> = typed_cell;
146 let mut arena = self.inner.borrow_mut();
147 let slot_entry = &mut arena.cells[slot as usize];
148 slot_entry.cell = Some(cell);
149 slot_entry.watcher_cell = Some(watcher_cell);
150 id
151 }
152
153 fn get_cell_opt(&self, id: StateId) -> Option<Rc<dyn Any>> {
154 self.inner
155 .borrow()
156 .cells
157 .get(id.slot_index())
158 .filter(|cell| cell.generation == id.generation())
159 .and_then(|cell| cell.cell.as_ref())
160 .cloned()
161 }
162
163 fn get_typed<T: Clone + 'static>(&self, id: StateId) -> Rc<TypedStateCell<T>> {
164 match self.get_cell_opt(id) {
165 None => panic!(
166 "state cell missing: slot={}, gen={}, expected={}",
167 id.slot(),
168 id.generation(),
169 std::any::type_name::<T>(),
170 ),
171 Some(cell) => Rc::downcast::<TypedStateCell<T>>(cell).unwrap_or_else(|_| {
172 panic!(
173 "state cell type mismatch: slot={}, gen={}, expected={}",
174 id.slot(),
175 id.generation(),
176 std::any::type_name::<T>(),
177 )
178 }),
179 }
180 }
181
182 fn get_typed_opt<T: Clone + 'static>(&self, id: StateId) -> Option<Rc<TypedStateCell<T>>> {
183 Rc::downcast::<TypedStateCell<T>>(self.get_cell_opt(id)?).ok()
184 }
185
186 pub(crate) fn with_typed<T: Clone + 'static, R>(
187 &self,
188 id: StateId,
189 f: impl FnOnce(&MutableStateInner<T>) -> R,
190 ) -> R {
191 let cell = self.get_typed::<T>(id);
192 f(&cell.inner)
193 }
194
195 pub(crate) fn with_typed_opt<T: Clone + 'static, R>(
196 &self,
197 id: StateId,
198 f: impl FnOnce(&MutableStateInner<T>) -> R,
199 ) -> Option<R> {
200 let cell = self.get_typed_opt::<T>(id)?;
201 Some(f(&cell.inner))
202 }
203
204 pub(crate) fn release(&self, id: StateId) {
205 let cell = {
206 let mut inner = self.inner.borrow_mut();
207 let Some(slot) = inner.cells.get_mut(id.slot_index()) else {
208 return;
209 };
210 if slot.generation != id.generation() {
211 return;
212 }
213 slot.lease = None;
214 slot.watcher_cell = None;
215 let cell = slot.cell.take();
216 if cell.is_some() {
217 inner.free.push(id.slot());
218 }
219 cell
220 };
221 drop(cell);
222 }
223
224 pub(crate) fn stats(&self) -> (usize, usize) {
225 let inner = self.inner.borrow();
226 (inner.cells.len(), inner.free.len())
227 }
228
229 pub(crate) fn debug_stats(&self) -> StateArenaDebugStats {
230 let inner = self.inner.borrow();
231 StateArenaDebugStats {
232 cells_len: inner.cells.len(),
233 cells_cap: inner.cells.capacity(),
234 free_len: inner.free.len(),
235 free_cap: inner.free.capacity(),
236 }
237 }
238
239 pub(crate) fn unregister_scope(&self, id: StateId, scope_id: ScopeId) {
240 let watcher_cell = {
241 let inner = self.inner.borrow();
242 inner
243 .cells
244 .get(id.slot_index())
245 .filter(|slot| slot.generation == id.generation())
246 .and_then(|slot| slot.watcher_cell.as_ref())
247 .cloned()
248 };
249 if let Some(watcher_cell) = watcher_cell {
250 watcher_cell.unregister_scope(scope_id);
251 }
252 }
253
254 pub(crate) fn register_lease(&self, id: StateId, lease: &Rc<StateHandleLease>) {
255 let mut inner = self.inner.borrow_mut();
256 let Some(slot) = inner.cells.get_mut(id.slot_index()) else {
257 return;
258 };
259 if slot.generation != id.generation() {
260 return;
261 }
262 slot.lease = Some(Rc::downgrade(lease));
263 }
264
265 pub(crate) fn retain_lease(&self, id: StateId) -> Option<Rc<StateHandleLease>> {
266 let inner = self.inner.borrow();
267 let slot = inner.cells.get(id.slot_index())?;
268 if slot.generation != id.generation() {
269 return None;
270 }
271 slot.lease.as_ref()?.upgrade()
272 }
273}
274
275#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
276pub struct StateId {
277 slot: u32,
278 generation: u32,
279}
280
281impl StateId {
282 const fn new(slot: u32, generation: u32) -> Self {
283 Self { slot, generation }
284 }
285
286 pub(crate) const fn slot(self) -> u32 {
287 self.slot
288 }
289
290 pub(crate) const fn slot_index(self) -> usize {
291 self.slot as usize
292 }
293
294 pub(crate) const fn generation(self) -> u32 {
295 self.generation
296 }
297}
298
299#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
300pub struct RuntimeId(u32);
301
302impl RuntimeId {
303 fn next() -> Self {
304 NEXT_RUNTIME_ID.with(|next| {
305 let id = next.get();
306 next.set(id.wrapping_add(1));
307 Self(id)
308 })
309 }
310}
311
312struct UiDispatcherInner {
313 scheduler: SchedulerRef,
314 tx: mpsc::Sender<UiMessage>,
315 pending: AtomicUsize,
316}
317
318#[cfg(not(target_arch = "wasm32"))]
328type UiDispatcherRef = Arc<UiDispatcherInner>;
329
330#[cfg(target_arch = "wasm32")]
332type UiDispatcherRef = Rc<UiDispatcherInner>;
333
334impl UiDispatcherInner {
335 fn new(scheduler: SchedulerRef, tx: mpsc::Sender<UiMessage>) -> Self {
336 Self {
337 scheduler,
338 tx,
339 pending: AtomicUsize::new(0),
340 }
341 }
342
343 fn post(&self, task: impl FnOnce() + Send + 'static) {
344 self.pending.fetch_add(1, Ordering::SeqCst);
345 if self.tx.send(UiMessage::Task(Box::new(task))).is_ok() {
346 self.scheduler.schedule_frame();
347 } else {
348 self.pending.fetch_sub(1, Ordering::SeqCst);
349 }
350 }
351
352 fn post_invoke(&self, id: u64, value: Box<dyn Any + Send>) {
353 self.pending.fetch_add(1, Ordering::SeqCst);
354 if self.tx.send(UiMessage::Invoke { id, value }).is_ok() {
355 self.scheduler.schedule_frame();
356 } else {
357 self.pending.fetch_sub(1, Ordering::SeqCst);
358 }
359 }
360
361 fn has_pending(&self) -> bool {
362 self.pending.load(Ordering::SeqCst) > 0
363 }
364}
365
366struct PendingGuard<'a> {
367 counter: &'a AtomicUsize,
368}
369
370impl<'a> PendingGuard<'a> {
371 fn new(counter: &'a AtomicUsize) -> Self {
372 Self { counter }
373 }
374}
375
376impl<'a> Drop for PendingGuard<'a> {
377 fn drop(&mut self) {
378 let mut current = self.counter.load(Ordering::SeqCst);
379 loop {
380 if current == 0 {
381 return;
382 }
383 match self.counter.compare_exchange(
384 current,
385 current - 1,
386 Ordering::SeqCst,
387 Ordering::SeqCst,
388 ) {
389 Ok(_) => return,
390 Err(next) => current = next,
391 }
392 }
393 }
394}
395
396#[derive(Clone)]
397pub struct UiDispatcher {
398 inner: UiDispatcherRef,
399}
400
401impl UiDispatcher {
402 fn new(inner: UiDispatcherRef) -> Self {
403 Self { inner }
404 }
405
406 pub fn post(&self, task: impl FnOnce() + Send + 'static) {
407 self.inner.post(task);
408 }
409
410 pub fn post_invoke<T>(&self, id: u64, value: T)
411 where
412 T: Send + 'static,
413 {
414 self.inner.post_invoke(id, Box::new(value));
415 }
416
417 pub fn has_pending(&self) -> bool {
418 self.inner.has_pending()
419 }
420}
421
422struct RuntimeInner {
423 scheduler: SchedulerRef,
424 needs_frame: RefCell<bool>,
425 node_updates: RefCell<Vec<Command>>,
426 invalid_scopes: RefCell<HashSet<ScopeId>>,
427 scope_queue: RefCell<Vec<(ScopeId, Weak<RecomposeScopeInner>)>>,
428 frame_callbacks: RefCell<VecDeque<FrameCallbackEntry>>,
429 next_frame_callback_id: Cell<u64>,
430 last_frame_time_nanos: Cell<Option<u64>>,
431 ui_dispatcher: UiDispatcherRef,
432 ui_rx: RefCell<mpsc::Receiver<UiMessage>>,
433 local_tasks: RefCell<VecDeque<Box<dyn FnOnce() + 'static>>>,
434 ui_conts: RefCell<UiContinuationMap>,
435 next_cont_id: Cell<u64>,
436 ui_thread_id: ThreadId,
437 tasks: RefCell<Vec<TaskEntry>>,
438 next_task_id: Cell<u64>,
439 state_arena: StateArena,
440 external_state_owners: RefCell<HashMap<StateId, Rc<StateHandleLease>>>,
441 live_recompose_scope_count: Cell<usize>,
442 runtime_id: RuntimeId,
443}
444
445struct TaskEntry {
446 id: u64,
447 label: String,
448 future: Pin<Box<dyn Future<Output = ()> + 'static>>,
449 runnable: Arc<AtomicBool>,
460 waker: Waker,
463}
464
465thread_local! {
466 static NEXT_TASK_LABEL: RefCell<Option<String>> = const { RefCell::new(None) };
467}
468
469pub fn label_next_ui_task(label: impl Into<String>) {
470 NEXT_TASK_LABEL.with(|held| *held.borrow_mut() = Some(label.into()));
471}
472
473impl RuntimeInner {
474 fn new(scheduler: SchedulerRef) -> Self {
475 let (tx, rx) = mpsc::channel();
476 let dispatcher = UiDispatcherRef::new(UiDispatcherInner::new(scheduler.clone(), tx));
477 Self {
478 scheduler,
479 needs_frame: RefCell::new(false),
480 node_updates: RefCell::new(Vec::new()),
481 invalid_scopes: RefCell::new(HashSet::default()),
482 scope_queue: RefCell::new(Vec::new()),
483 frame_callbacks: RefCell::new(VecDeque::new()),
484 next_frame_callback_id: Cell::new(1),
485 last_frame_time_nanos: Cell::new(None),
486 ui_dispatcher: dispatcher,
487 ui_rx: RefCell::new(rx),
488 local_tasks: RefCell::new(VecDeque::new()),
489 ui_conts: RefCell::new(UiContinuationMap::default()),
490 next_cont_id: Cell::new(1),
491 ui_thread_id: std::thread::current().id(),
492 tasks: RefCell::new(Vec::new()),
493 next_task_id: Cell::new(1),
494 state_arena: StateArena::default(),
495 external_state_owners: RefCell::new(HashMap::default()),
496 live_recompose_scope_count: Cell::new(0),
497 runtime_id: RuntimeId::next(),
498 }
499 }
500
501 fn schedule(&self) {
502 *self.needs_frame.borrow_mut() = true;
503 self.scheduler.schedule_frame();
504 }
505
506 fn enqueue_update(&self, command: Command) {
507 self.node_updates.borrow_mut().push(command);
508 self.schedule(); }
510
511 fn take_updates(&self) -> Vec<Command> {
512 self.node_updates.borrow_mut().drain(..).collect::<Vec<_>>()
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}