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