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