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"))]
319type UiDispatcherRef = Arc<UiDispatcherInner>;
320
321#[cfg(target_arch = "wasm32")]
322type UiDispatcherRef = Rc<UiDispatcherInner>;
323
324impl UiDispatcherInner {
325 fn new(scheduler: SchedulerRef, tx: mpsc::Sender<UiMessage>) -> Self {
326 Self {
327 scheduler,
328 tx,
329 pending: AtomicUsize::new(0),
330 }
331 }
332
333 fn post(&self, task: impl FnOnce() + Send + 'static) {
334 self.pending.fetch_add(1, Ordering::SeqCst);
335 if self.tx.send(UiMessage::Task(Box::new(task))).is_ok() {
336 self.scheduler.schedule_frame();
337 } else {
338 self.pending.fetch_sub(1, Ordering::SeqCst);
339 }
340 }
341
342 fn post_invoke(&self, id: u64, value: Box<dyn Any + Send>) {
343 self.pending.fetch_add(1, Ordering::SeqCst);
344 if self.tx.send(UiMessage::Invoke { id, value }).is_ok() {
345 self.scheduler.schedule_frame();
346 } else {
347 self.pending.fetch_sub(1, Ordering::SeqCst);
348 }
349 }
350
351 fn has_pending(&self) -> bool {
352 self.pending.load(Ordering::SeqCst) > 0
353 }
354}
355
356struct PendingGuard<'a> {
357 counter: &'a AtomicUsize,
358}
359
360impl<'a> PendingGuard<'a> {
361 fn new(counter: &'a AtomicUsize) -> Self {
362 Self { counter }
363 }
364}
365
366impl<'a> Drop for PendingGuard<'a> {
367 fn drop(&mut self) {
368 let mut current = self.counter.load(Ordering::SeqCst);
369 loop {
370 if current == 0 {
371 return;
372 }
373 match self.counter.compare_exchange(
374 current,
375 current - 1,
376 Ordering::SeqCst,
377 Ordering::SeqCst,
378 ) {
379 Ok(_) => return,
380 Err(next) => current = next,
381 }
382 }
383 }
384}
385
386#[derive(Clone)]
387pub struct UiDispatcher {
388 inner: UiDispatcherRef,
389}
390
391impl UiDispatcher {
392 fn new(inner: UiDispatcherRef) -> Self {
393 Self { inner }
394 }
395
396 pub fn post(&self, task: impl FnOnce() + Send + 'static) {
397 self.inner.post(task);
398 }
399
400 pub fn post_invoke<T>(&self, id: u64, value: T)
401 where
402 T: Send + 'static,
403 {
404 self.inner.post_invoke(id, Box::new(value));
405 }
406
407 pub fn has_pending(&self) -> bool {
408 self.inner.has_pending()
409 }
410}
411
412struct RuntimeInner {
413 scheduler: SchedulerRef,
414 needs_frame: RefCell<bool>,
415 node_updates: RefCell<Vec<Command>>,
416 invalid_scopes: RefCell<HashSet<ScopeId>>,
417 scope_queue: RefCell<Vec<(ScopeId, Weak<RecomposeScopeInner>)>>,
418 frame_callbacks: RefCell<VecDeque<FrameCallbackEntry>>,
419 next_frame_callback_id: Cell<u64>,
420 last_frame_time_nanos: Cell<Option<u64>>,
421 ui_dispatcher: UiDispatcherRef,
422 ui_rx: RefCell<mpsc::Receiver<UiMessage>>,
423 local_tasks: RefCell<VecDeque<Box<dyn FnOnce() + 'static>>>,
424 ui_conts: RefCell<UiContinuationMap>,
425 next_cont_id: Cell<u64>,
426 ui_thread_id: ThreadId,
427 tasks: RefCell<Vec<TaskEntry>>,
428 next_task_id: Cell<u64>,
429 state_arena: StateArena,
430 external_state_owners: RefCell<HashMap<StateId, Rc<StateHandleLease>>>,
431 live_recompose_scope_count: Cell<usize>,
432 runtime_id: RuntimeId,
433}
434
435struct TaskEntry {
436 id: u64,
437 label: String,
438 future: Pin<Box<dyn Future<Output = ()> + 'static>>,
439 runnable: Arc<AtomicBool>,
440 waker: Waker,
441}
442
443thread_local! {
444 static NEXT_TASK_LABEL: RefCell<Option<String>> = const { RefCell::new(None) };
445}
446
447pub fn label_next_ui_task(label: impl Into<String>) {
448 NEXT_TASK_LABEL.with(|held| *held.borrow_mut() = Some(label.into()));
449}
450
451impl RuntimeInner {
452 fn new(scheduler: SchedulerRef) -> Self {
453 let (tx, rx) = mpsc::channel();
454 let dispatcher = UiDispatcherRef::new(UiDispatcherInner::new(scheduler.clone(), tx));
455 Self {
456 scheduler,
457 needs_frame: RefCell::new(false),
458 node_updates: RefCell::new(Vec::new()),
459 invalid_scopes: RefCell::new(HashSet::default()),
460 scope_queue: RefCell::new(Vec::new()),
461 frame_callbacks: RefCell::new(VecDeque::new()),
462 next_frame_callback_id: Cell::new(1),
463 last_frame_time_nanos: Cell::new(None),
464 ui_dispatcher: dispatcher,
465 ui_rx: RefCell::new(rx),
466 local_tasks: RefCell::new(VecDeque::new()),
467 ui_conts: RefCell::new(UiContinuationMap::default()),
468 next_cont_id: Cell::new(1),
469 ui_thread_id: std::thread::current().id(),
470 tasks: RefCell::new(Vec::new()),
471 next_task_id: Cell::new(1),
472 state_arena: StateArena::default(),
473 external_state_owners: RefCell::new(HashMap::default()),
474 live_recompose_scope_count: Cell::new(0),
475 runtime_id: RuntimeId::next(),
476 }
477 }
478
479 fn schedule(&self) {
480 *self.needs_frame.borrow_mut() = true;
481 self.scheduler.schedule_frame();
482 }
483
484 fn enqueue_update(&self, command: Command) {
485 self.node_updates.borrow_mut().push(command);
486 self.schedule();
487 }
488
489 fn take_updates(&self) -> Vec<Command> {
490 self.node_updates.borrow_mut().drain(..).collect::<Vec<_>>()
491 }
492
493 fn has_updates(&self) -> bool {
494 !self.node_updates.borrow().is_empty() || self.has_invalid_scopes()
495 }
496
497 fn register_invalid_scope(&self, id: ScopeId, scope: Weak<RecomposeScopeInner>) {
498 let mut invalid = self.invalid_scopes.borrow_mut();
499 if invalid.insert(id) {
500 self.scope_queue.borrow_mut().push((id, scope));
501 self.schedule();
502 }
503 }
504
505 fn requeue_invalid_scope(&self, id: ScopeId, scope: Weak<RecomposeScopeInner>) {
506 if self.invalid_scopes.borrow().contains(&id) {
507 self.scope_queue.borrow_mut().push((id, scope));
508 self.schedule();
509 }
510 }
511
512 fn mark_scope_recomposed(&self, id: ScopeId) {
513 self.invalid_scopes.borrow_mut().remove(&id);
514 }
515
516 fn take_invalidated_scopes(&self) -> Vec<(ScopeId, Weak<RecomposeScopeInner>)> {
517 let mut queue = self.scope_queue.borrow_mut();
518 if queue.is_empty() {
519 return Vec::new();
520 }
521 let pending: Vec<_> = queue.drain(..).collect();
522 drop(queue);
523 let invalid = self.invalid_scopes.borrow();
524 pending
525 .into_iter()
526 .filter(|(id, _)| invalid.contains(id))
527 .collect()
528 }
529
530 fn has_invalid_scopes(&self) -> bool {
531 !self.invalid_scopes.borrow().is_empty()
532 }
533
534 fn increment_live_recompose_scope_count(&self) {
535 self.live_recompose_scope_count
536 .set(self.live_recompose_scope_count.get().saturating_add(1));
537 }
538
539 fn decrement_live_recompose_scope_count(&self) {
540 self.live_recompose_scope_count
541 .set(self.live_recompose_scope_count.get().saturating_sub(1));
542 }
543
544 fn live_recompose_scope_count(&self) -> usize {
545 self.live_recompose_scope_count.get()
546 }
547
548 fn has_frame_callbacks(&self) -> bool {
549 !self.frame_callbacks.borrow().is_empty()
550 }
551
552 fn has_transient_frame_callbacks(&self) -> bool {
553 self.frame_callbacks
554 .borrow()
555 .iter()
556 .any(|entry| entry.kind == FrameCallbackKind::Transient)
557 }
558
559 fn enqueue_ui_task(&self, task: Box<dyn FnOnce() + 'static>) {
560 self.local_tasks.borrow_mut().push_back(task);
561 self.schedule();
562 }
563
564 fn spawn_ui_task(&self, future: Pin<Box<dyn Future<Output = ()> + 'static>>) -> u64 {
565 let id = self.next_task_id.get();
566 self.next_task_id.set(id + 1);
567 let label = NEXT_TASK_LABEL
568 .with(|held| held.borrow_mut().take())
569 .unwrap_or_else(|| "unnamed".to_string());
570 let runnable = Arc::new(AtomicBool::new(true));
571 let waker = RuntimeTaskWaker::new(self, Arc::clone(&runnable)).into_waker();
572 self.tasks.borrow_mut().push(TaskEntry {
573 id,
574 label,
575 future,
576 runnable,
577 waker,
578 });
579 self.schedule();
580 id
581 }
582
583 fn cancel_task(&self, id: u64) {
584 let mut tasks = self.tasks.borrow_mut();
585 if tasks.iter().any(|entry| entry.id == id) {
586 tasks.retain(|entry| entry.id != id);
587 }
588 }
589
590 fn has_task(&self, id: u64) -> bool {
591 self.tasks
592 .try_borrow()
593 .map(|tasks| tasks.iter().any(|entry| entry.id == id))
594 .unwrap_or(true)
595 }
596
597 fn poll_async_tasks(&self) -> bool {
598 let mut tasks_ref = self.tasks.borrow_mut();
599 let tasks = std::mem::take(&mut *tasks_ref);
600 drop(tasks_ref);
601 let mut pending = Vec::with_capacity(tasks.len());
602 let mut made_progress = false;
603 for mut entry in tasks.into_iter() {
604 if !entry.runnable.swap(false, Ordering::AcqRel) {
605 pending.push(entry);
606 continue;
607 }
608 let mut cx = Context::from_waker(&entry.waker);
609 match entry.future.as_mut().poll(&mut cx) {
610 Poll::Ready(()) => {
611 made_progress = true;
612 }
613 Poll::Pending => {
614 pending.push(entry);
615 }
616 }
617 }
618 if !pending.is_empty() {
619 self.tasks.borrow_mut().extend(pending);
620 }
621 made_progress
622 }
623
624 fn drain_ui(&self) {
625 loop {
626 let mut executed = false;
627
628 {
629 let rx = &mut *self.ui_rx.borrow_mut();
630 for message in rx.try_iter() {
631 executed = true;
632 let _guard = PendingGuard::new(&self.ui_dispatcher.pending);
633 match message {
634 UiMessage::Task(task) => {
635 task();
636 }
637 UiMessage::Invoke { id, value } => {
638 self.invoke_ui_cont(id, value);
639 }
640 }
641 }
642 }
643
644 loop {
645 let task = {
646 let mut local = self.local_tasks.borrow_mut();
647 local.pop_front()
648 };
649
650 match task {
651 Some(task) => {
652 executed = true;
653 task();
654 }
655 None => break,
656 }
657 }
658
659 if self.poll_async_tasks() {
660 executed = true;
661 }
662
663 if !executed {
664 break;
665 }
666 }
667
668 self.clear_needs_frame_if_idle();
669 }
670
671 fn has_pending_ui(&self) -> bool {
672 let local_pending = self
673 .local_tasks
674 .try_borrow()
675 .map(|tasks| !tasks.is_empty())
676 .unwrap_or(true);
677
678 local_pending || self.ui_dispatcher.has_pending() || self.has_runnable_tasks()
679 }
680
681 fn has_runnable_tasks(&self) -> bool {
682 self.tasks
683 .try_borrow()
684 .map(|tasks| {
685 tasks
686 .iter()
687 .any(|task| task.runnable.load(Ordering::Acquire))
688 })
689 .unwrap_or(true)
690 }
691
692 fn register_ui_cont<T: 'static>(&self, f: impl FnOnce(T) + 'static) -> u64 {
693 debug_assert_eq!(
694 std::thread::current().id(),
695 self.ui_thread_id,
696 "UI continuation registered off the runtime thread",
697 );
698 let id = self.next_cont_id.get();
699 self.next_cont_id.set(id + 1);
700 let callback = RefCell::new(Some(f));
701 self.ui_conts.borrow_mut().insert(
702 id,
703 Box::new(move |value: Box<dyn Any>| {
704 let Ok(value) = value.downcast::<T>() else {
705 return false;
706 };
707 let Some(slot) = callback.borrow_mut().take() else {
708 return true;
709 };
710 slot(*value);
711 true
712 }),
713 );
714 id
715 }
716
717 fn invoke_ui_cont(&self, id: u64, value: Box<dyn Any + Send>) {
718 debug_assert_eq!(
719 std::thread::current().id(),
720 self.ui_thread_id,
721 "UI continuation invoked off the runtime thread",
722 );
723 let callback = { self.ui_conts.borrow_mut().remove(&id) };
724 if let Some(callback) = callback {
725 let value: Box<dyn Any> = value;
726 if !callback(value) {
727 self.ui_conts.borrow_mut().insert(id, callback);
728 }
729 }
730 }
731
732 fn cancel_ui_cont(&self, id: u64) {
733 self.ui_conts.borrow_mut().remove(&id);
734 }
735
736 fn register_frame_callback(
737 &self,
738 kind: FrameCallbackKind,
739 callback: Box<dyn FnOnce(u64) + 'static>,
740 ) -> FrameCallbackId {
741 let id = self.next_frame_callback_id.get();
742 self.next_frame_callback_id.set(id + 1);
743 self.frame_callbacks
744 .borrow_mut()
745 .push_back(FrameCallbackEntry {
746 id,
747 kind,
748 callback: Some(callback),
749 });
750 self.schedule();
751 id
752 }
753
754 fn cancel_frame_callback(&self, id: FrameCallbackId) {
755 let mut callbacks = self.frame_callbacks.borrow_mut();
756 if let Some(index) = callbacks.iter().position(|entry| entry.id == id) {
757 callbacks.remove(index);
758 }
759 drop(callbacks);
760 self.clear_needs_frame_if_idle();
761 }
762
763 fn clear_needs_frame_if_idle(&self) {
764 if !self.has_invalid_scopes()
765 && !self.has_updates()
766 && !self.has_frame_callbacks()
767 && !self.has_pending_ui()
768 {
769 *self.needs_frame.borrow_mut() = false;
770 }
771 }
772
773 fn drain_frame_callbacks(&self, frame_time_nanos: u64) {
774 self.last_frame_time_nanos.set(Some(frame_time_nanos));
775 let mut callbacks = self.frame_callbacks.borrow_mut();
776 let mut pending: Vec<Box<dyn FnOnce(u64) + 'static>> = Vec::with_capacity(callbacks.len());
777 while let Some(mut entry) = callbacks.pop_front() {
778 if let Some(callback) = entry.callback.take() {
779 pending.push(callback);
780 }
781 }
782 drop(callbacks);
783
784 if !pending.is_empty() {
785 let _ = crate::run_in_mutable_snapshot(|| {
786 for callback in pending {
787 callback(frame_time_nanos);
788 }
789 });
790 }
791
792 self.clear_needs_frame_if_idle();
793 }
794
795 fn debug_stats(&self) -> RuntimeDebugStats {
796 let node_updates = self.node_updates.borrow();
797 let invalid_scopes = self.invalid_scopes.borrow();
798 let scope_queue = self.scope_queue.borrow();
799 let frame_callbacks = self.frame_callbacks.borrow();
800 let local_tasks = self.local_tasks.borrow();
801 let ui_conts = self.ui_conts.borrow();
802 let tasks = self.tasks.borrow();
803 let external_state_owners = self.external_state_owners.borrow();
804
805 RuntimeDebugStats {
806 node_updates_len: node_updates.len(),
807 node_updates_cap: node_updates.capacity(),
808 invalid_scopes_len: invalid_scopes.len(),
809 invalid_scopes_cap: invalid_scopes.capacity(),
810 scope_queue_len: scope_queue.len(),
811 scope_queue_cap: scope_queue.capacity(),
812 frame_callbacks_len: frame_callbacks.len(),
813 frame_callbacks_cap: frame_callbacks.capacity(),
814 local_tasks_len: local_tasks.len(),
815 local_tasks_cap: local_tasks.capacity(),
816 ui_conts_len: ui_conts.len(),
817 ui_conts_cap: ui_conts.capacity(),
818 tasks_len: tasks.len(),
819 tasks_cap: tasks.capacity(),
820 external_state_owners_len: external_state_owners.len(),
821 external_state_owners_cap: external_state_owners.capacity(),
822 ui_dispatcher_pending: self.ui_dispatcher.pending.load(Ordering::SeqCst),
823 }
824 }
825}
826
827#[derive(Clone)]
828pub struct Runtime {
829 inner: Rc<RuntimeInner>,
830}
831
832impl Runtime {
833 pub fn new(scheduler: SchedulerRef) -> Self {
834 let inner = Rc::new(RuntimeInner::new(scheduler));
835 let runtime = Self { inner };
836 let handle = runtime.handle();
837 register_runtime_handle(&handle);
838 LAST_RUNTIME.with(|slot| *slot.borrow_mut() = Some(handle));
839 runtime
840 }
841
842 pub fn handle(&self) -> RuntimeHandle {
843 RuntimeHandle {
844 inner: Rc::downgrade(&self.inner),
845 dispatcher: UiDispatcher::new(self.inner.ui_dispatcher.clone()),
846 ui_thread_id: self.inner.ui_thread_id,
847 id: self.inner.runtime_id,
848 }
849 }
850
851 pub fn has_updates(&self) -> bool {
852 self.inner.has_updates()
853 }
854
855 pub fn needs_frame(&self) -> bool {
856 *self.inner.needs_frame.borrow() || self.inner.has_runnable_tasks()
857 }
858
859 pub fn set_needs_frame(&self, value: bool) {
860 *self.inner.needs_frame.borrow_mut() = value;
861 }
862
863 pub fn last_frame_time_nanos(&self) -> Option<u64> {
868 self.inner.last_frame_time_nanos.get()
869 }
870
871 #[cfg(any(feature = "internal", test))]
872 pub fn frame_clock(&self) -> FrameClock {
873 FrameClock::new(self.handle())
874 }
875}
876
877impl Drop for Runtime {
878 fn drop(&mut self) {
879 if Rc::strong_count(&self.inner) != 1 {
880 return;
881 }
882 unregister_runtime_handle(self.inner.runtime_id);
883 LAST_RUNTIME.with(|slot| {
884 let should_clear = slot
885 .borrow()
886 .as_ref()
887 .is_some_and(|handle| handle.id() == self.inner.runtime_id);
888 if should_clear {
889 *slot.borrow_mut() = None;
890 }
891 });
892 }
893}
894
895#[derive(Default)]
896pub struct DefaultScheduler;
897
898impl RuntimeScheduler for DefaultScheduler {
899 fn schedule_frame(&self) {}
900}
901
902#[cfg(test)]
903#[derive(Default)]
904pub struct TestScheduler;
905
906#[cfg(test)]
907impl RuntimeScheduler for TestScheduler {
908 fn schedule_frame(&self) {}
909}
910
911#[cfg(test)]
912pub struct TestRuntime {
913 runtime: Runtime,
914}
915
916#[cfg(test)]
917impl Default for TestRuntime {
918 fn default() -> Self {
919 Self::new()
920 }
921}
922
923#[cfg(test)]
924impl TestRuntime {
925 pub fn new() -> Self {
926 Self {
927 runtime: Runtime::new(Arc::new(TestScheduler)),
928 }
929 }
930
931 pub fn handle(&self) -> RuntimeHandle {
932 self.runtime.handle()
933 }
934}
935
936#[derive(Clone)]
937pub struct RuntimeHandle {
938 inner: Weak<RuntimeInner>,
939 dispatcher: UiDispatcher,
940 ui_thread_id: ThreadId,
941 id: RuntimeId,
942}
943
944pub struct TaskHandle {
945 id: u64,
946 runtime: RuntimeHandle,
947}
948
949struct DeferredStateRelease {
950 runtime: RuntimeHandle,
951 id: StateId,
952}
953
954pub(crate) struct StateHandleLease {
955 id: StateId,
956 runtime: RuntimeHandle,
957}
958
959impl StateHandleLease {
960 pub(crate) fn id(&self) -> StateId {
961 self.id
962 }
963
964 pub(crate) fn runtime(&self) -> RuntimeHandle {
965 self.runtime.clone()
966 }
967}
968
969impl Drop for StateHandleLease {
970 fn drop(&mut self) {
971 defer_state_release(self.runtime.clone(), self.id);
972 }
973}
974
975impl RuntimeHandle {
976 pub fn id(&self) -> RuntimeId {
977 self.id
978 }
979
980 pub(crate) fn alloc_state<T: Clone + 'static>(&self, value: T) -> Rc<StateHandleLease> {
981 let id = self.with_state_arena(|arena| arena.alloc(value, self.clone()));
982 let lease = Rc::new(StateHandleLease {
983 id,
984 runtime: self.clone(),
985 });
986 self.with_state_arena(|arena| arena.register_lease(id, &lease));
987 lease
988 }
989
990 pub(crate) fn alloc_state_with_policy<T: Clone + 'static>(
991 &self,
992 value: T,
993 policy: Arc<dyn MutationPolicy<T>>,
994 ) -> Rc<StateHandleLease> {
995 let id =
996 self.with_state_arena(|arena| arena.alloc_with_policy(value, self.clone(), policy));
997 let lease = Rc::new(StateHandleLease {
998 id,
999 runtime: self.clone(),
1000 });
1001 self.with_state_arena(|arena| arena.register_lease(id, &lease));
1002 lease
1003 }
1004
1005 pub(crate) fn alloc_persistent_state<T: Clone + 'static>(
1006 &self,
1007 value: T,
1008 ) -> crate::MutableState<T> {
1009 let lease = self.alloc_state(value);
1010 if let Some(inner) = self.inner.upgrade() {
1011 inner
1012 .external_state_owners
1013 .borrow_mut()
1014 .insert(lease.id(), Rc::clone(&lease));
1015 }
1016 crate::MutableState::from_lease(&lease)
1017 }
1018
1019 pub(crate) fn retain_state_lease(&self, id: StateId) -> Option<Rc<StateHandleLease>> {
1020 self.with_state_arena(|arena| arena.retain_lease(id))
1021 }
1022
1023 pub(crate) fn with_state_arena<R>(&self, f: impl FnOnce(&StateArena) -> R) -> R {
1024 self.try_with_state_arena(f)
1025 .unwrap_or_else(|| panic!("runtime dropped"))
1026 }
1027
1028 pub(crate) fn try_with_state_arena<R>(&self, f: impl FnOnce(&StateArena) -> R) -> Option<R> {
1029 self.inner.upgrade().map(|inner| f(&inner.state_arena))
1030 }
1031
1032 fn release_state_immediate(&self, id: StateId) {
1033 if let Some(inner) = self.inner.upgrade() {
1034 inner.state_arena.release(id);
1035 }
1036 }
1037
1038 pub fn state_arena_stats(&self) -> (usize, usize) {
1039 self.try_with_state_arena(StateArena::stats)
1040 .unwrap_or_default()
1041 }
1042
1043 pub fn state_arena_debug_stats(&self) -> StateArenaDebugStats {
1044 self.try_with_state_arena(StateArena::debug_stats)
1045 .unwrap_or_default()
1046 }
1047
1048 pub fn debug_stats(&self) -> RuntimeDebugStats {
1049 self.inner
1050 .upgrade()
1051 .map(|inner| inner.debug_stats())
1052 .unwrap_or_default()
1053 }
1054
1055 pub fn live_ui_task_labels(&self) -> Vec<(u64, String)> {
1056 self.inner
1057 .upgrade()
1058 .map(|inner| {
1059 inner
1060 .tasks
1061 .borrow()
1062 .iter()
1063 .map(|entry| (entry.id, entry.label.clone()))
1064 .collect()
1065 })
1066 .unwrap_or_default()
1067 }
1068
1069 pub(crate) fn unregister_state_scope(&self, id: StateId, scope_id: ScopeId) {
1070 if let Some(inner) = self.inner.upgrade() {
1071 inner.state_arena.unregister_scope(id, scope_id);
1072 }
1073 }
1074
1075 pub fn schedule(&self) {
1076 if let Some(inner) = self.inner.upgrade() {
1077 inner.schedule();
1078 }
1079 }
1080
1081 pub(crate) fn enqueue_node_update(&self, command: Command) {
1082 if let Some(inner) = self.inner.upgrade() {
1083 inner.enqueue_update(command);
1084 }
1085 }
1086
1087 pub fn enqueue_ui_task(&self, task: Box<dyn FnOnce() + 'static>) {
1094 if let Some(inner) = self.inner.upgrade() {
1095 inner.enqueue_ui_task(task);
1096 } else {
1097 task();
1098 }
1099 }
1100
1101 pub fn spawn_ui<F>(&self, fut: F) -> Option<TaskHandle>
1102 where
1103 F: Future<Output = ()> + 'static,
1104 {
1105 self.inner.upgrade().map(|inner| {
1106 let id = inner.spawn_ui_task(Box::pin(fut));
1107 TaskHandle {
1108 id,
1109 runtime: self.clone(),
1110 }
1111 })
1112 }
1113
1114 pub fn cancel_task(&self, id: u64) {
1115 if let Some(inner) = self.inner.upgrade() {
1116 inner.cancel_task(id);
1117 }
1118 }
1119
1120 pub fn has_task(&self, id: u64) -> bool {
1122 self.inner
1123 .upgrade()
1124 .map(|inner| inner.has_task(id))
1125 .unwrap_or(false)
1126 }
1127
1128 pub fn post_ui(&self, task: impl FnOnce() + Send + 'static) {
1133 self.dispatcher.post(task);
1134 }
1135
1136 pub fn register_ui_cont<T: 'static>(&self, f: impl FnOnce(T) + 'static) -> Option<u64> {
1137 self.inner.upgrade().map(|inner| inner.register_ui_cont(f))
1138 }
1139
1140 pub fn cancel_ui_cont(&self, id: u64) {
1141 if let Some(inner) = self.inner.upgrade() {
1142 inner.cancel_ui_cont(id);
1143 }
1144 }
1145
1146 pub fn drain_ui(&self) {
1147 if let Some(inner) = self.inner.upgrade() {
1148 inner.drain_ui();
1149 }
1150 }
1151
1152 pub fn has_pending_ui(&self) -> bool {
1153 self.inner
1154 .upgrade()
1155 .map(|inner| inner.has_pending_ui())
1156 .unwrap_or_else(|| self.dispatcher.has_pending())
1157 }
1158
1159 pub fn register_frame_callback(
1160 &self,
1161 callback: impl FnOnce(u64) + 'static,
1162 ) -> Option<FrameCallbackId> {
1163 self.inner.upgrade().map(|inner| {
1164 inner.register_frame_callback(FrameCallbackKind::Transient, Box::new(callback))
1165 })
1166 }
1167
1168 pub fn register_perpetual_frame_callback(
1169 &self,
1170 callback: impl FnOnce(u64) + 'static,
1171 ) -> Option<FrameCallbackId> {
1172 self.inner.upgrade().map(|inner| {
1173 inner.register_frame_callback(FrameCallbackKind::Perpetual, Box::new(callback))
1174 })
1175 }
1176
1177 pub fn cancel_frame_callback(&self, id: FrameCallbackId) {
1178 if let Some(inner) = self.inner.upgrade() {
1179 inner.cancel_frame_callback(id);
1180 }
1181 }
1182
1183 pub fn drain_frame_callbacks(&self, frame_time_nanos: u64) {
1184 if let Some(inner) = self.inner.upgrade() {
1185 inner.drain_frame_callbacks(frame_time_nanos);
1186 }
1187 }
1188
1189 pub fn last_frame_time_nanos(&self) -> Option<u64> {
1192 self.inner
1193 .upgrade()
1194 .and_then(|inner| inner.last_frame_time_nanos.get())
1195 }
1196
1197 #[cfg(any(feature = "internal", test))]
1198 pub fn frame_clock(&self) -> FrameClock {
1199 FrameClock::new(self.clone())
1200 }
1201
1202 pub fn set_needs_frame(&self, value: bool) {
1203 if let Some(inner) = self.inner.upgrade() {
1204 *inner.needs_frame.borrow_mut() = value;
1205 }
1206 }
1207
1208 pub(crate) fn take_updates(&self) -> Vec<Command> {
1209 self.inner
1210 .upgrade()
1211 .map(|inner| inner.take_updates())
1212 .unwrap_or_default()
1213 }
1214
1215 pub fn has_updates(&self) -> bool {
1216 self.inner
1217 .upgrade()
1218 .map(|inner| inner.has_updates())
1219 .unwrap_or(false)
1220 }
1221
1222 pub(crate) fn mark_scope_recomposed(&self, id: ScopeId) {
1223 if let Some(inner) = self.inner.upgrade() {
1224 inner.mark_scope_recomposed(id);
1225 }
1226 }
1227
1228 pub(crate) fn register_invalid_scope(&self, id: ScopeId, scope: Weak<RecomposeScopeInner>) {
1229 if let Some(inner) = self.inner.upgrade() {
1230 inner.register_invalid_scope(id, scope);
1231 }
1232 }
1233
1234 pub(crate) fn requeue_invalid_scope(&self, id: ScopeId, scope: Weak<RecomposeScopeInner>) {
1235 if let Some(inner) = self.inner.upgrade() {
1236 inner.requeue_invalid_scope(id, scope);
1237 }
1238 }
1239
1240 pub(crate) fn take_invalidated_scopes(&self) -> Vec<(ScopeId, Weak<RecomposeScopeInner>)> {
1241 self.inner
1242 .upgrade()
1243 .map(|inner| inner.take_invalidated_scopes())
1244 .unwrap_or_default()
1245 }
1246
1247 pub fn has_invalid_scopes(&self) -> bool {
1248 self.inner
1249 .upgrade()
1250 .map(|inner| inner.has_invalid_scopes())
1251 .unwrap_or(false)
1252 }
1253
1254 pub(crate) fn increment_live_recompose_scope_count(&self) {
1255 if let Some(inner) = self.inner.upgrade() {
1256 inner.increment_live_recompose_scope_count();
1257 }
1258 }
1259
1260 pub(crate) fn decrement_live_recompose_scope_count(&self) {
1261 if let Some(inner) = self.inner.upgrade() {
1262 inner.decrement_live_recompose_scope_count();
1263 }
1264 }
1265
1266 fn live_recompose_scope_count(&self) -> usize {
1267 self.inner
1268 .upgrade()
1269 .map(|inner| inner.live_recompose_scope_count())
1270 .unwrap_or_default()
1271 }
1272
1273 #[doc(hidden)]
1274 pub fn debug_invalid_scope_ids(&self) -> Vec<usize> {
1275 self.inner
1276 .upgrade()
1277 .map(|inner| inner.invalid_scopes.borrow().iter().copied().collect())
1278 .unwrap_or_default()
1279 }
1280
1281 pub fn has_frame_callbacks(&self) -> bool {
1282 self.inner
1283 .upgrade()
1284 .map(|inner| inner.has_frame_callbacks())
1285 .unwrap_or(false)
1286 }
1287
1288 pub fn has_transient_frame_callbacks(&self) -> bool {
1289 self.inner
1290 .upgrade()
1291 .map(|inner| inner.has_transient_frame_callbacks())
1292 .unwrap_or(false)
1293 }
1294
1295 pub fn assert_ui_thread(&self) {
1296 debug_assert_eq!(
1297 std::thread::current().id(),
1298 self.ui_thread_id,
1299 "state mutated off the runtime's UI thread"
1300 );
1301 }
1302
1303 pub fn dispatcher(&self) -> UiDispatcher {
1304 self.dispatcher.clone()
1305 }
1306
1307 #[doc(hidden)]
1308 pub fn with_deferred_state_releases<R>(&self, f: impl FnOnce() -> R) -> R {
1309 let _scope = enter_state_teardown_scope();
1310 f()
1311 }
1312}
1313
1314impl TaskHandle {
1315 pub fn cancel(&self) {
1316 self.runtime.cancel_task(self.id);
1317 }
1318
1319 pub fn is_finished(&self) -> bool {
1321 !self.runtime.has_task(self.id)
1322 }
1323}
1324
1325pub(crate) struct FrameCallbackEntry {
1326 id: FrameCallbackId,
1327 kind: FrameCallbackKind,
1328 callback: Option<Box<dyn FnOnce(u64) + 'static>>,
1329}
1330
1331#[cfg(not(target_arch = "wasm32"))]
1332struct RuntimeTaskWaker {
1333 scheduler: SchedulerRef,
1334 runnable: Arc<AtomicBool>,
1335}
1336
1337#[cfg(target_arch = "wasm32")]
1338struct RuntimeTaskWaker {
1339 runtime_id: RuntimeId,
1340 runnable: Arc<AtomicBool>,
1341}
1342
1343impl RuntimeTaskWaker {
1344 #[cfg(not(target_arch = "wasm32"))]
1345 fn new(inner: &RuntimeInner, runnable: Arc<AtomicBool>) -> Self {
1346 let scheduler = inner.scheduler.clone();
1347 Self {
1348 scheduler,
1349 runnable,
1350 }
1351 }
1352
1353 #[cfg(target_arch = "wasm32")]
1354 fn new(inner: &RuntimeInner, runnable: Arc<AtomicBool>) -> Self {
1355 let runtime_id = inner.runtime_id;
1356 Self {
1357 runtime_id,
1358 runnable,
1359 }
1360 }
1361
1362 fn into_waker(self) -> Waker {
1363 futures_task::waker(Arc::new(self))
1364 }
1365}
1366
1367impl futures_task::ArcWake for RuntimeTaskWaker {
1368 #[cfg(not(target_arch = "wasm32"))]
1369 fn wake_by_ref(arc_self: &Arc<Self>) {
1370 arc_self.runnable.store(true, Ordering::Release);
1371 arc_self.scheduler.schedule_frame();
1372 }
1373
1374 #[cfg(target_arch = "wasm32")]
1375 fn wake_by_ref(arc_self: &Arc<Self>) {
1376 arc_self.runnable.store(true, Ordering::Release);
1377 REGISTERED_RUNTIMES.with(|registry| {
1378 if let Some(handle) = registry.borrow().get(&arc_self.runtime_id).cloned() {
1379 handle.schedule();
1380 }
1381 });
1382 }
1383}
1384
1385thread_local! {
1386 static NEXT_RUNTIME_ID: Cell<u32> = const { Cell::new(1) };
1387 static ACTIVE_RUNTIMES: RefCell<Vec<RuntimeHandle>> = const { RefCell::new(Vec::new()) };
1388 static LAST_RUNTIME: RefCell<Option<RuntimeHandle>> = const { RefCell::new(None) };
1389 static REGISTERED_RUNTIMES: RefCell<HashMap<RuntimeId, RuntimeHandle>> = RefCell::new(HashMap::default());
1390 static STATE_TEARDOWN_DEPTH: Cell<usize> = const { Cell::new(0) };
1391 static DEFERRED_STATE_RELEASES: RefCell<Vec<DeferredStateRelease>> = const { RefCell::new(Vec::new()) };
1392}
1393
1394#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1395pub struct RuntimeThreadLocalDebugStats {
1396 pub active_runtimes_len: usize,
1397 pub active_runtimes_cap: usize,
1398 pub registered_runtimes_len: usize,
1399 pub registered_runtimes_cap: usize,
1400 pub deferred_state_releases_len: usize,
1401 pub deferred_state_releases_cap: usize,
1402}
1403
1404pub fn current_runtime_handle() -> Option<RuntimeHandle> {
1409 if let Some(handle) = ACTIVE_RUNTIMES.with(|stack| stack.borrow().last().cloned()) {
1410 return Some(handle);
1411 }
1412 LAST_RUNTIME.with(|slot| slot.borrow().clone())
1413}
1414
1415pub(crate) fn runtime_handle_by_id(id: RuntimeId) -> Option<RuntimeHandle> {
1416 REGISTERED_RUNTIMES.with(|registry| registry.borrow().get(&id).cloned())
1417}
1418
1419pub(crate) fn live_recompose_scope_count() -> usize {
1420 REGISTERED_RUNTIMES.with(|registry| {
1421 registry
1422 .borrow()
1423 .values()
1424 .map(RuntimeHandle::live_recompose_scope_count)
1425 .sum()
1426 })
1427}
1428
1429pub fn debug_runtime_thread_local_stats() -> RuntimeThreadLocalDebugStats {
1430 let (active_runtimes_len, active_runtimes_cap) = ACTIVE_RUNTIMES.with(|stack| {
1431 let stack = stack.borrow();
1432 (stack.len(), stack.capacity())
1433 });
1434 let (registered_runtimes_len, registered_runtimes_cap) = REGISTERED_RUNTIMES.with(|registry| {
1435 let registry = registry.borrow();
1436 (registry.len(), registry.capacity())
1437 });
1438 let (deferred_state_releases_len, deferred_state_releases_cap) =
1439 DEFERRED_STATE_RELEASES.with(|releases| {
1440 let releases = releases.borrow();
1441 (releases.len(), releases.capacity())
1442 });
1443
1444 RuntimeThreadLocalDebugStats {
1445 active_runtimes_len,
1446 active_runtimes_cap,
1447 registered_runtimes_len,
1448 registered_runtimes_cap,
1449 deferred_state_releases_len,
1450 deferred_state_releases_cap,
1451 }
1452}
1453
1454fn register_runtime_handle(handle: &RuntimeHandle) {
1455 REGISTERED_RUNTIMES.with(|registry| {
1456 registry.borrow_mut().insert(handle.id(), handle.clone());
1457 });
1458}
1459
1460fn unregister_runtime_handle(id: RuntimeId) {
1461 REGISTERED_RUNTIMES.with(|registry| {
1462 registry.borrow_mut().remove(&id);
1463 });
1464}
1465
1466fn defer_state_release(runtime: RuntimeHandle, id: StateId) {
1467 let teardown_active = STATE_TEARDOWN_DEPTH.with(|depth| depth.get() > 0);
1468 if teardown_active {
1469 DEFERRED_STATE_RELEASES.with(|releases| {
1470 releases
1471 .borrow_mut()
1472 .push(DeferredStateRelease { runtime, id });
1473 });
1474 } else {
1475 runtime.release_state_immediate(id);
1476 }
1477}
1478
1479fn flush_deferred_state_releases() {
1480 DEFERRED_STATE_RELEASES.with(|releases| {
1481 let mut releases = releases.borrow_mut();
1482 while let Some(deferred) = releases.pop() {
1483 deferred.runtime.release_state_immediate(deferred.id);
1484 }
1485 });
1486}
1487
1488pub(crate) struct StateTeardownScope;
1489
1490pub(crate) fn enter_state_teardown_scope() -> StateTeardownScope {
1491 STATE_TEARDOWN_DEPTH.with(|depth| depth.set(depth.get() + 1));
1492 StateTeardownScope
1493}
1494
1495impl Drop for StateTeardownScope {
1496 fn drop(&mut self) {
1497 STATE_TEARDOWN_DEPTH.with(|depth| {
1498 let next = depth.get().saturating_sub(1);
1499 depth.set(next);
1500 if next == 0 {
1501 flush_deferred_state_releases();
1502 }
1503 });
1504 }
1505}
1506
1507pub(crate) fn push_active_runtime(handle: &RuntimeHandle) {
1508 register_runtime_handle(handle);
1509 ACTIVE_RUNTIMES.with(|stack| stack.borrow_mut().push(handle.clone()));
1510 LAST_RUNTIME.with(|slot| *slot.borrow_mut() = Some(handle.clone()));
1511}
1512
1513pub(crate) fn pop_active_runtime() {
1514 ACTIVE_RUNTIMES.with(|stack| {
1515 stack.borrow_mut().pop();
1516 });
1517}
1518
1519pub fn schedule_frame() {
1521 if let Some(handle) = current_runtime_handle() {
1522 handle.schedule();
1523 return;
1524 }
1525 log::debug!(
1526 target: "cranpose::runtime",
1527 "ignoring frame request without an active runtime",
1528 );
1529}
1530
1531pub fn schedule_node_update(
1533 update: impl FnOnce(&mut dyn Applier) -> Result<(), NodeError> + 'static,
1534) {
1535 if let Some(handle) = current_runtime_handle() {
1536 handle.enqueue_node_update(Command::callback(update));
1537 } else {
1538 drop(update);
1539 log::debug!(
1540 target: "cranpose::runtime",
1541 "ignoring node update request without an active runtime",
1542 );
1543 }
1544}
1545
1546#[cfg(test)]
1547mod tests {
1548 use super::*;
1549
1550 #[test]
1551 fn state_arena_alloc_skips_free_slot_outside_cell_storage() {
1552 let runtime = TestRuntime::new();
1553 let arena = StateArena::default();
1554 let first = arena.alloc(1_i32, runtime.handle());
1555 arena.inner.borrow_mut().free.push(u32::MAX);
1556
1557 let second = arena.alloc(2_i32, runtime.handle());
1558
1559 assert_eq!(first.slot(), 0);
1560 assert_eq!(second.slot(), 1);
1561 assert!(arena.get_typed_opt::<i32>(first).is_some());
1562 assert!(arena.get_typed_opt::<i32>(second).is_some());
1563 }
1564
1565 #[test]
1566 fn state_arena_alloc_skips_occupied_free_slot() {
1567 let runtime = TestRuntime::new();
1568 let arena = StateArena::default();
1569 let first = arena.alloc(1_i32, runtime.handle());
1570 arena.inner.borrow_mut().free.push(first.slot());
1571
1572 let second = arena.alloc(2_i32, runtime.handle());
1573
1574 assert_ne!(first.slot(), second.slot());
1575 assert_eq!(second.slot(), 1);
1576 assert!(arena.get_typed_opt::<i32>(first).is_some());
1577 assert!(arena.get_typed_opt::<i32>(second).is_some());
1578 }
1579
1580 #[test]
1581 fn state_arena_register_lease_ignores_stale_id() {
1582 let runtime = TestRuntime::new();
1583 let arena = StateArena::default();
1584 let stale_id = StateId::new(99, 0);
1585 let lease = Rc::new(StateHandleLease {
1586 id: stale_id,
1587 runtime: runtime.handle(),
1588 });
1589
1590 arena.register_lease(stale_id, &lease);
1591
1592 assert!(arena.retain_lease(stale_id).is_none());
1593 }
1594
1595 #[test]
1596 fn ui_continuation_type_mismatch_is_ignored_until_matching_payload() {
1597 let runtime = TestRuntime::new();
1598 let handle = runtime.handle();
1599 let received = Rc::new(Cell::new(None));
1600 let received_for_continuation = Rc::clone(&received);
1601 let cont_id = handle
1602 .register_ui_cont(move |value: u32| {
1603 received_for_continuation.set(Some(value));
1604 })
1605 .expect("test runtime is alive");
1606
1607 handle.dispatcher().post_invoke(cont_id, "wrong payload");
1608 handle.drain_ui();
1609
1610 assert_eq!(received.get(), None);
1611 assert_eq!(handle.debug_stats().ui_conts_len, 1);
1612
1613 handle.dispatcher().post_invoke(cont_id, 42_u32);
1614 handle.drain_ui();
1615
1616 assert_eq!(received.get(), Some(42));
1617 assert_eq!(handle.debug_stats().ui_conts_len, 0);
1618 }
1619
1620 #[test]
1621 fn ui_dispatcher_failed_send_does_not_leave_pending_work() {
1622 let runtime = Runtime::new(Arc::new(TestScheduler));
1623 let dispatcher = runtime.handle().dispatcher();
1624
1625 drop(runtime);
1626
1627 assert!(!dispatcher.has_pending());
1628 dispatcher.post(|| {});
1629 assert!(!dispatcher.has_pending());
1630 dispatcher.post_invoke(404, 12_u32);
1631 assert!(!dispatcher.has_pending());
1632 }
1633
1634 #[test]
1635 fn pending_guard_does_not_wrap_on_underflow() {
1636 let counter = AtomicUsize::new(0);
1637
1638 {
1639 let _guard = PendingGuard::new(&counter);
1640 }
1641
1642 assert_eq!(counter.load(Ordering::SeqCst), 0);
1643 }
1644
1645 #[test]
1646 fn pending_guard_decrements_pending_count() {
1647 let counter = AtomicUsize::new(2);
1648
1649 {
1650 let _guard = PendingGuard::new(&counter);
1651 }
1652
1653 assert_eq!(counter.load(Ordering::SeqCst), 1);
1654 }
1655
1656 #[test]
1657 fn schedule_frame_without_runtime_is_ignored() {
1658 let ok = std::thread::spawn(|| std::panic::catch_unwind(super::schedule_frame).is_ok())
1659 .join()
1660 .expect("test thread should join");
1661
1662 assert!(ok);
1663 }
1664
1665 #[test]
1666 fn schedule_node_update_without_runtime_is_ignored() {
1667 let ok = std::thread::spawn(|| {
1668 std::panic::catch_unwind(|| {
1669 super::schedule_node_update(|_| Ok(()));
1670 })
1671 .is_ok()
1672 })
1673 .join()
1674 .expect("test thread should join");
1675
1676 assert!(ok);
1677 }
1678}