1#![allow(clippy::type_complexity)]
3
4use crate::collections::map::{HashMap, HashSet};
5use crate::hash::default as default_hash;
6use crate::snapshot_v2::{register_apply_observer, ReadObserver, StateObjectId};
7use crate::state::StateObject;
8use crate::{RecomposeScope, RecomposeScopeInner, ScopeId};
9use smallvec::SmallVec;
10use std::any::{Any, TypeId};
11use std::cell::{Cell, RefCell};
12use std::hash::{Hash, Hasher};
13use std::rc::{Rc, Weak};
14use std::sync::Arc;
15
16type Executor = dyn Fn(Box<dyn FnOnce() + 'static>) + 'static;
18
19#[derive(Clone)]
32pub struct SnapshotStateObserver {
33 inner: Rc<SnapshotStateObserverInner>,
34}
35
36#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
37pub struct SnapshotStateObserverDebugStats {
38 pub scopes_len: usize,
39 pub scopes_cap: usize,
40 pub fast_scopes_len: usize,
41 pub fast_scopes_cap: usize,
42 pub stateless_scope_count: usize,
43 pub observed_state_count: usize,
44 pub observed_state_capacity: usize,
45}
46
47impl SnapshotStateObserver {
48 pub fn new(on_changed_executor: impl Fn(Box<dyn FnOnce() + 'static>) + 'static) -> Self {
50 let inner = Rc::new(SnapshotStateObserverInner::new(on_changed_executor));
51 inner.set_self(Rc::downgrade(&inner));
52 Self { inner }
53 }
54
55 pub fn observe_reads<T, R>(
61 &self,
62 scope: T,
63 on_value_changed_for_scope: impl Fn(&T) + 'static,
64 block: impl FnOnce() -> R,
65 ) -> R
66 where
67 T: Any + Clone + Eq + Hash + 'static,
68 {
69 self.inner
70 .observe_reads(scope, on_value_changed_for_scope, block)
71 }
72
73 pub fn begin_frame(&self) {
75 self.inner.begin_frame();
76 }
77
78 pub fn prune_dead_scopes(&self) {
80 self.inner.prune_dead_scopes();
81 }
82
83 pub fn with_no_observations<R>(&self, block: impl FnOnce() -> R) -> R {
85 self.inner.with_no_observations(block)
86 }
87
88 pub fn clear<T>(&self, scope: &T)
90 where
91 T: Any + Eq + Hash + 'static,
92 {
93 self.inner.clear(scope);
94 }
95
96 pub fn clear_if(&self, predicate: impl Fn(&dyn Any) -> bool) {
98 self.inner.clear_if(predicate);
99 }
100
101 pub fn clear_all(&self) {
103 self.inner.clear_all();
104 }
105
106 pub fn start(&self) {
108 let weak = Rc::downgrade(&self.inner);
109 self.inner.start(weak);
110 }
111
112 pub fn stop(&self) {
114 self.inner.stop();
115 }
116
117 pub fn debug_stats(&self) -> SnapshotStateObserverDebugStats {
118 self.inner.debug_stats()
119 }
120
121 #[cfg(test)]
123 pub fn notify_changes(&self, modified: &[Arc<dyn StateObject>]) {
124 self.inner.handle_apply(modified);
125 }
126}
127
128struct SnapshotStateObserverInner {
129 executor: Rc<Executor>,
143 owned_scopes: RefCell<HashMap<OwnedScopeIndexKey, OwnedScopeBucket>>,
144 fast_scopes: RefCell<HashMap<ScopeId, Rc<RefCell<ScopeEntry>>>>,
145 indexed_scopes: RefCell<HashMap<usize, Rc<RefCell<ScopeEntry>>>>,
146 observed_to_scopes: RefCell<HashMap<StateObjectId, HashSet<usize>>>,
147 pause_count: Rc<Cell<usize>>,
148 active_read_targets: Rc<RefCell<Vec<Rc<RefCell<ObservedIds>>>>>,
149 read_dispatcher: ReadObserver,
150 apply_handle: RefCell<Option<crate::snapshot_v2::ObserverHandle>>,
151 weak_self: RefCell<Weak<SnapshotStateObserverInner>>,
152 frame_version: Cell<u64>,
153 next_entry_id: Cell<usize>,
154}
155
156#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
157struct OwnedScopeIndexKey {
158 type_id: TypeId,
159 value_hash: u64,
160}
161
162type OwnedScopeBucket = SmallVec<[Rc<RefCell<ScopeEntry>>; 1]>;
163
164fn owned_scope_index_key<T>(scope: &T) -> OwnedScopeIndexKey
165where
166 T: Any + Hash + 'static,
167{
168 let mut hasher = default_hash::new();
169 scope.hash(&mut hasher);
170 OwnedScopeIndexKey {
171 type_id: TypeId::of::<T>(),
172 value_hash: hasher.finish(),
173 }
174}
175
176impl SnapshotStateObserverInner {
177 const MIN_RETAINED_SCOPE_CAPACITY: usize = 256;
178
179 fn new(on_changed_executor: impl Fn(Box<dyn FnOnce() + 'static>) + 'static) -> Self {
180 let pause_count = Rc::new(Cell::new(0));
181 let active_read_targets = Rc::new(RefCell::new(Vec::<Rc<RefCell<ObservedIds>>>::new()));
182 let dispatcher_pause_count = Rc::clone(&pause_count);
183 let dispatcher_targets = Rc::clone(&active_read_targets);
184 let read_dispatcher: ReadObserver = Arc::new(move |state| {
185 if dispatcher_pause_count.get() > 0 {
186 return;
187 }
188 let observed = dispatcher_targets.borrow().last().cloned();
189 if let Some(observed) = observed {
190 observed.borrow_mut().insert(state);
191 }
192 });
193
194 Self {
195 executor: Rc::new(on_changed_executor),
196 owned_scopes: RefCell::new(HashMap::default()),
197 fast_scopes: RefCell::new(HashMap::default()),
198 indexed_scopes: RefCell::new(HashMap::default()),
199 observed_to_scopes: RefCell::new(HashMap::default()),
200 pause_count,
201 active_read_targets,
202 read_dispatcher,
203 apply_handle: RefCell::new(None),
204 weak_self: RefCell::new(Weak::new()),
205 frame_version: Cell::new(0),
206 next_entry_id: Cell::new(0),
207 }
208 }
209
210 fn set_self(&self, weak: Weak<SnapshotStateObserverInner>) {
211 self.weak_self.replace(weak);
212 }
213
214 fn begin_frame(&self) {
215 let next = self.frame_version.get().wrapping_add(1);
216 self.frame_version.set(next);
217 self.prune_dead_scopes();
218 }
219
220 fn observe_reads<T, R>(
221 &self,
222 scope: T,
223 on_value_changed_for_scope: impl Fn(&T) + 'static,
224 block: impl FnOnce() -> R,
225 ) -> R
226 where
227 T: Any + Clone + Eq + Hash + 'static,
228 {
229 let frame_version = self.frame_version.get();
230 let has_frame_version = frame_version != 0;
231
232 let on_changed: Rc<dyn Fn(&dyn Any)> = {
233 let callback = Rc::new(on_value_changed_for_scope);
234 Rc::new(move |scope_any: &dyn Any| {
235 if let Some(typed) = scope_any.downcast_ref::<T>() {
236 callback(typed);
237 }
238 })
239 };
240
241 let existing_entry = self.find_scope_entry(&scope);
242 if let Some(entry) = existing_entry.as_ref() {
243 let already_observed = {
244 let mut entry_mut = entry.borrow_mut();
245 entry_mut.update(scope.clone(), on_changed.clone());
246 has_frame_version && entry_mut.last_seen_version == frame_version
247 };
248 if already_observed {
249 return block();
250 }
251 }
252
253 let observed = Rc::new(RefCell::new(ObservedIds::new()));
254 self.active_read_targets
255 .borrow_mut()
256 .push(Rc::clone(&observed));
257 struct ActiveObservationGuard {
258 stack: Rc<RefCell<Vec<Rc<RefCell<ObservedIds>>>>>,
259 }
260 impl Drop for ActiveObservationGuard {
261 fn drop(&mut self) {
262 self.stack.borrow_mut().pop();
263 }
264 }
265 let _guard = ActiveObservationGuard {
266 stack: Rc::clone(&self.active_read_targets),
267 };
268
269 let result = self.run_with_read_observer(block);
270
271 if observed.borrow().is_empty() {
272 if existing_entry.is_some() {
273 self.clear(&scope);
274 }
275 return result;
276 }
277
278 let observed = {
279 let mut observed = observed.borrow_mut();
280 std::mem::replace(&mut *observed, ObservedIds::new())
281 };
282 let entry = existing_entry
283 .unwrap_or_else(|| self.insert_scope_entry(scope.clone(), on_changed.clone()));
284 {
285 let mut entry_mut = entry.borrow_mut();
286 entry_mut.update(scope, on_changed);
287 entry_mut.last_seen_version = if has_frame_version {
288 frame_version
289 } else {
290 u64::MAX
291 };
292 }
293 self.replace_observed_ids(&entry, observed);
294
295 result
296 }
297
298 fn with_no_observations<R>(&self, block: impl FnOnce() -> R) -> R {
299 self.pause_count.set(self.pause_count.get() + 1);
300 let result = block();
301 self.pause_count
302 .set(self.pause_count.get().saturating_sub(1));
303 result
304 }
305
306 fn clear<T>(&self, scope: &T)
307 where
308 T: Any + Eq + Hash + 'static,
309 {
310 if let Some(rc_scope) = (scope as &dyn Any).downcast_ref::<RecomposeScope>() {
311 if let Some(entry) = self.fast_scopes.borrow_mut().remove(&rc_scope.id()) {
312 self.unregister_entry(&entry);
313 }
314 return;
315 }
316
317 let removed = self.remove_owned_scope_entry(scope);
318 if let Some(entry) = removed {
319 self.unregister_entry(&entry);
320 }
321 }
322
323 fn clear_if(&self, predicate: impl Fn(&dyn Any) -> bool) {
324 let removed_fast = {
325 let mut fast_scopes = self.fast_scopes.borrow_mut();
326 let removed_ids: Vec<_> = fast_scopes
327 .iter()
328 .filter(|(_, entry)| entry.borrow().matches_predicate(&predicate))
329 .map(|(scope_id, _)| *scope_id)
330 .collect();
331 removed_ids
332 .into_iter()
333 .filter_map(|scope_id| fast_scopes.remove(&scope_id))
334 .collect::<Vec<_>>()
335 };
336 let removed_owned =
337 { self.partition_owned_scopes(|entry| entry.matches_predicate(&predicate)) };
338
339 for entry in removed_fast.into_iter().chain(removed_owned) {
340 self.unregister_entry(&entry);
341 }
342 }
343
344 fn clear_all(&self) {
345 self.fast_scopes.borrow_mut().clear();
346 self.owned_scopes.borrow_mut().clear();
347 self.indexed_scopes.borrow_mut().clear();
348 self.observed_to_scopes.borrow_mut().clear();
349 }
350
351 fn start(&self, weak_self: Weak<SnapshotStateObserverInner>) {
352 if self.apply_handle.borrow().is_some() {
353 return;
354 }
355
356 let handle = register_apply_observer(Rc::new(move |modified, _snapshot_id| {
357 if let Some(inner) = weak_self.upgrade() {
358 inner.handle_apply(modified);
359 }
360 }));
361 self.apply_handle.replace(Some(handle));
362 }
363
364 fn stop(&self) {
365 if let Some(handle) = self.apply_handle.borrow_mut().take() {
366 drop(handle);
367 }
368 }
369
370 fn find_scope_entry<T>(&self, scope: &T) -> Option<Rc<RefCell<ScopeEntry>>>
371 where
372 T: Any + Eq + Hash + 'static,
373 {
374 if let Some(scope) = (scope as &dyn Any).downcast_ref::<RecomposeScope>() {
375 return self.fast_scopes.borrow().get(&scope.id()).cloned();
376 }
377
378 self.find_owned_scope_entry(scope)
379 }
380
381 fn insert_scope_entry(
382 &self,
383 scope: impl Any + Clone + Eq + Hash + 'static,
384 on_changed: Rc<dyn Fn(&dyn Any)>,
385 ) -> Rc<RefCell<ScopeEntry>> {
386 let entry_id = self.next_entry_id.get();
387 self.next_entry_id.set(entry_id.wrapping_add(1));
388 let recompose_scope_id = (&scope as &dyn Any)
389 .downcast_ref::<RecomposeScope>()
390 .map(RecomposeScope::id);
391 let owned_scope_key = recompose_scope_id
392 .is_none()
393 .then(|| owned_scope_index_key(&scope));
394 let entry = Rc::new(RefCell::new(ScopeEntry::new(entry_id, scope, on_changed)));
395 self.indexed_scopes
396 .borrow_mut()
397 .insert(entry_id, Rc::clone(&entry));
398 if let Some(scope_id) = recompose_scope_id {
399 self.fast_scopes
400 .borrow_mut()
401 .insert(scope_id, Rc::clone(&entry));
402 } else if let Some(scope_key) = owned_scope_key {
403 self.owned_scopes
404 .borrow_mut()
405 .entry(scope_key)
406 .or_default()
407 .push(Rc::clone(&entry));
408 }
409 entry
410 }
411
412 fn prune_dead_scopes(&self) {
413 let removed_fast = {
414 let mut fast_scopes = self.fast_scopes.borrow_mut();
415 let removed_ids: Vec<_> = fast_scopes
416 .iter()
417 .filter(|(_, entry)| !entry.borrow().should_retain())
418 .map(|(scope_id, _)| *scope_id)
419 .collect();
420 let removed = removed_ids
421 .into_iter()
422 .filter_map(|scope_id| fast_scopes.remove(&scope_id))
423 .collect::<Vec<_>>();
424 shrink_map_if_sparse(&mut fast_scopes, Self::MIN_RETAINED_SCOPE_CAPACITY);
425 removed
426 };
427
428 let removed_owned = { self.partition_owned_scopes(|entry| !entry.should_retain()) };
429
430 for entry in removed_fast.into_iter().chain(removed_owned) {
431 self.unregister_entry(&entry);
432 }
433 }
434
435 fn find_owned_scope_entry<T>(&self, scope: &T) -> Option<Rc<RefCell<ScopeEntry>>>
436 where
437 T: Any + Eq + Hash + 'static,
438 {
439 let key = owned_scope_index_key(scope);
440 self.owned_scopes.borrow().get(&key).and_then(|bucket| {
441 bucket
442 .iter()
443 .find(|entry| entry.borrow().matches_scope(scope))
444 .cloned()
445 })
446 }
447
448 fn remove_owned_scope_entry<T>(&self, scope: &T) -> Option<Rc<RefCell<ScopeEntry>>>
449 where
450 T: Any + Eq + Hash + 'static,
451 {
452 let key = owned_scope_index_key(scope);
453 let mut owned_scopes = self.owned_scopes.borrow_mut();
454 let mut removed = None;
455 let mut remove_bucket = false;
456 if let Some(bucket) = owned_scopes.get_mut(&key) {
457 if let Some(index) = bucket
458 .iter()
459 .position(|entry| entry.borrow().matches_scope(scope))
460 {
461 removed = Some(bucket.remove(index));
462 remove_bucket = bucket.is_empty();
463 }
464 }
465 if remove_bucket {
466 owned_scopes.remove(&key);
467 }
468 shrink_map_if_sparse(&mut owned_scopes, Self::MIN_RETAINED_SCOPE_CAPACITY);
469 removed
470 }
471
472 fn partition_owned_scopes(
473 &self,
474 should_remove: impl Fn(&ScopeEntry) -> bool,
475 ) -> Vec<Rc<RefCell<ScopeEntry>>> {
476 let mut owned_scopes = self.owned_scopes.borrow_mut();
477 let mut retained = HashMap::default();
478 let mut removed = Vec::new();
479 for (key, mut bucket) in owned_scopes.drain() {
480 let mut retained_bucket = OwnedScopeBucket::new();
481 for entry in bucket.drain(..) {
482 if should_remove(&entry.borrow()) {
483 removed.push(entry);
484 } else {
485 retained_bucket.push(entry);
486 }
487 }
488 if !retained_bucket.is_empty() {
489 retained.insert(key, retained_bucket);
490 }
491 }
492 *owned_scopes = retained;
493 shrink_map_if_sparse(&mut owned_scopes, Self::MIN_RETAINED_SCOPE_CAPACITY);
494 removed
495 }
496
497 fn debug_stats(&self) -> SnapshotStateObserverDebugStats {
498 let owned_scopes = self.owned_scopes.borrow();
499 let fast_scopes = self.fast_scopes.borrow();
500 let owned_scope_len = owned_scopes.values().map(SmallVec::len).sum::<usize>();
501 let owned_scope_cap =
502 owned_scopes.capacity() + owned_scopes.values().map(SmallVec::capacity).sum::<usize>();
503 let scopes_len = owned_scope_len + fast_scopes.len();
504 let scopes_cap = owned_scope_cap + fast_scopes.capacity();
505 let mut observed_state_count = 0;
506 let mut observed_state_capacity = 0;
507 let mut stateless_scope_count = 0;
508
509 for entry in owned_scopes
510 .values()
511 .flat_map(|bucket| bucket.iter())
512 .chain(fast_scopes.values())
513 {
514 let entry = entry.borrow();
515 observed_state_count += entry.observed.len();
516 observed_state_capacity += entry.observed.capacity();
517 stateless_scope_count += usize::from(entry.observed.is_empty());
518 }
519
520 SnapshotStateObserverDebugStats {
521 scopes_len,
522 scopes_cap,
523 fast_scopes_len: fast_scopes.len(),
524 fast_scopes_cap: fast_scopes.capacity(),
525 stateless_scope_count,
526 observed_state_count,
527 observed_state_capacity,
528 }
529 }
530
531 fn run_with_read_observer<R>(&self, block: impl FnOnce() -> R) -> R {
532 use crate::snapshot_v2::take_transparent_observer_mutable_snapshot;
535
536 let snapshot =
539 take_transparent_observer_mutable_snapshot(Some(self.read_dispatcher.clone()), None);
540 let result = snapshot.enter(block);
541 snapshot.dispose();
542 result
543 }
544
545 fn handle_apply(&self, modified: &[Arc<dyn StateObject>]) {
546 if modified.is_empty() {
547 return;
548 }
549
550 let mut seen_scope_ids: HashSet<usize> = HashSet::default();
551 let mut to_notify: Vec<Rc<RefCell<ScopeEntry>>> = Vec::new();
552 {
553 let observed_to_scopes = self.observed_to_scopes.borrow();
554 let indexed_scopes = self.indexed_scopes.borrow();
555 for state in modified {
556 if let Some(scope_ids) = observed_to_scopes.get(&state.object_id().as_usize()) {
557 let mut ordered_scope_ids: SmallVec<[usize; 8]> =
558 scope_ids.iter().copied().collect();
559 ordered_scope_ids.sort_unstable();
560 for scope_id in ordered_scope_ids {
561 if seen_scope_ids.insert(scope_id) {
562 if let Some(entry) = indexed_scopes.get(&scope_id) {
563 to_notify.push(entry.clone());
564 }
565 }
566 }
567 }
568 }
569 }
570
571 if to_notify.is_empty() {
572 return;
573 }
574
575 for entry in to_notify {
576 let executor = self.executor.clone();
577 executor(Box::new(move || {
578 if let Ok(entry) = entry.try_borrow() {
579 entry.notify();
580 }
581 }));
582 }
583 }
584
585 fn replace_observed_ids(&self, entry: &Rc<RefCell<ScopeEntry>>, observed: ObservedIds) {
586 let (entry_id, previous) = {
587 let mut entry_mut = entry.borrow_mut();
588 let entry_id = entry_mut.id;
589 let previous = std::mem::replace(&mut entry_mut.observed, observed);
590 (entry_id, previous)
591 };
592 self.unregister_observed_ids(entry_id, &previous);
593 let entry_ref = entry.borrow();
594 self.register_observed_ids(entry_id, &entry_ref.observed);
595 }
596
597 fn register_observed_ids(&self, entry_id: usize, observed: &ObservedIds) {
598 let mut observed_to_scopes = self.observed_to_scopes.borrow_mut();
599 for state_id in observed.iter() {
600 let scope_ids = observed_to_scopes.entry(state_id).or_default();
601 scope_ids.insert(entry_id);
602 }
603 }
604
605 fn unregister_observed_ids(&self, entry_id: usize, observed: &ObservedIds) {
606 let mut observed_to_scopes = self.observed_to_scopes.borrow_mut();
607 let mut emptied = SmallVec::<[StateObjectId; MAX_OBSERVED_STATES]>::new();
608 for state_id in observed.iter() {
609 if let Some(scope_ids) = observed_to_scopes.get_mut(&state_id) {
610 scope_ids.remove(&entry_id);
611 if scope_ids.is_empty() {
612 emptied.push(state_id);
613 }
614 }
615 }
616 for state_id in emptied {
617 observed_to_scopes.remove(&state_id);
618 }
619 shrink_map_if_sparse(&mut observed_to_scopes, Self::MIN_RETAINED_SCOPE_CAPACITY);
620 }
621
622 fn unregister_entry(&self, entry: &Rc<RefCell<ScopeEntry>>) {
623 let (entry_id, observed) = {
624 let mut entry_mut = entry.borrow_mut();
625 let observed = std::mem::replace(&mut entry_mut.observed, ObservedIds::new());
626 (entry_mut.id, observed)
627 };
628 self.unregister_observed_ids(entry_id, &observed);
629 self.indexed_scopes.borrow_mut().remove(&entry_id);
630 }
631}
632
633fn shrink_map_if_sparse<K, V>(map: &mut HashMap<K, V>, min_retained_capacity: usize)
634where
635 K: Eq + std::hash::Hash,
636{
637 if map.capacity() <= map.len().max(min_retained_capacity).saturating_mul(4) {
638 return;
639 }
640
641 let retained = map.len().max(min_retained_capacity);
642 let mut rebuilt = HashMap::default();
643 rebuilt.reserve(retained);
644 rebuilt.extend(map.drain());
645 *map = rebuilt;
646}
647
648enum ObservedIds {
649 Small(SmallVec<[ObservedState; MAX_OBSERVED_STATES]>),
650 Large(HashMap<StateObjectId, Option<Box<dyn Any>>>),
651}
652
653struct ObservedState {
654 id: StateObjectId,
655 _lease: Option<Box<dyn Any>>,
656}
657
658impl ObservedIds {
659 fn new() -> Self {
660 ObservedIds::Small(SmallVec::new())
661 }
662
663 fn insert(&mut self, state: &dyn StateObject) {
664 let id = state.object_id().as_usize();
665 match self {
666 ObservedIds::Small(small) => {
667 if small.iter().any(|observed| observed.id == id) {
668 return;
669 }
670 if small.len() < MAX_OBSERVED_STATES {
671 small.push(ObservedState {
672 id,
673 _lease: state.observation_lease(),
674 });
675 } else {
676 let mut large =
677 HashMap::with_capacity_and_hasher(small.len() + 1, Default::default());
678 for observed in small.drain(..) {
679 large.insert(observed.id, observed._lease);
680 }
681 large.insert(id, state.observation_lease());
682 *self = ObservedIds::Large(large);
683 }
684 }
685 ObservedIds::Large(large) => {
686 if !large.contains_key(&id) {
687 large.insert(id, state.observation_lease());
688 }
689 }
690 }
691 }
692
693 fn is_empty(&self) -> bool {
694 match self {
695 ObservedIds::Small(small) => small.is_empty(),
696 ObservedIds::Large(large) => large.is_empty(),
697 }
698 }
699
700 fn len(&self) -> usize {
701 match self {
702 ObservedIds::Small(small) => small.len(),
703 ObservedIds::Large(large) => large.len(),
704 }
705 }
706
707 fn capacity(&self) -> usize {
708 match self {
709 ObservedIds::Small(small) => small.capacity(),
710 ObservedIds::Large(large) => large.capacity(),
711 }
712 }
713
714 fn iter(&self) -> Box<dyn Iterator<Item = StateObjectId> + '_> {
715 match self {
716 ObservedIds::Small(small) => Box::new(small.iter().map(|observed| observed.id)),
717 ObservedIds::Large(large) => Box::new(large.keys().copied()),
718 }
719 }
720}
721
722const MAX_OBSERVED_STATES: usize = 8;
723
724enum ScopeStorage {
725 Owned(Box<dyn Any>),
726 RecomposeScope {
727 id: ScopeId,
728 weak: Weak<RecomposeScopeInner>,
729 },
730}
731
732struct ScopeEntry {
733 id: usize,
734 scope: ScopeStorage,
735 on_changed: Rc<dyn Fn(&dyn Any)>,
736 observed: ObservedIds,
737 last_seen_version: u64,
738}
739
740impl ScopeEntry {
741 fn new<T>(id: usize, scope: T, on_changed: Rc<dyn Fn(&dyn Any)>) -> Self
742 where
743 T: Any + 'static,
744 {
745 Self {
746 id,
747 scope: ScopeStorage::from_value(scope),
748 on_changed,
749 observed: ObservedIds::new(),
750 last_seen_version: u64::MAX,
751 }
752 }
753
754 fn update<T>(&mut self, new_scope: T, on_changed: Rc<dyn Fn(&dyn Any)>)
755 where
756 T: Any + 'static,
757 {
758 self.scope = ScopeStorage::from_value(new_scope);
759 self.on_changed = on_changed;
760 }
761
762 fn matches_scope<T>(&self, scope: &T) -> bool
763 where
764 T: Any + Eq + 'static,
765 {
766 if let Some(scope) = (scope as &dyn Any).downcast_ref::<RecomposeScope>() {
767 return matches!(
768 &self.scope,
769 ScopeStorage::RecomposeScope { id, .. } if *id == scope.id()
770 );
771 }
772
773 match &self.scope {
774 ScopeStorage::Owned(stored) => stored
775 .downcast_ref::<T>()
776 .map(|stored| stored == scope)
777 .unwrap_or(false),
778 ScopeStorage::RecomposeScope { .. } => false,
779 }
780 }
781
782 fn matches_predicate(&self, predicate: &impl Fn(&dyn Any) -> bool) -> bool {
783 match &self.scope {
784 ScopeStorage::Owned(scope) => predicate(scope.as_ref()),
785 ScopeStorage::RecomposeScope { weak, .. } => weak
786 .upgrade()
787 .map(|inner| predicate(&RecomposeScope { inner }))
788 .unwrap_or(true),
789 }
790 }
791
792 fn should_retain(&self) -> bool {
793 match &self.scope {
794 ScopeStorage::Owned(_) => true,
795 ScopeStorage::RecomposeScope { weak, .. } => weak.upgrade().is_some(),
796 }
797 }
798
799 fn notify(&self) {
800 match &self.scope {
801 ScopeStorage::Owned(scope) => (self.on_changed)(scope.as_ref()),
802 ScopeStorage::RecomposeScope { weak, .. } => {
803 if let Some(inner) = weak.upgrade() {
804 (self.on_changed)(&RecomposeScope { inner });
805 }
806 }
807 }
808 }
809}
810
811impl ScopeStorage {
812 fn from_value<T>(value: T) -> Self
813 where
814 T: Any + 'static,
815 {
816 let any = &value as &dyn Any;
817 if let Some(scope) = any.downcast_ref::<RecomposeScope>() {
818 Self::RecomposeScope {
819 id: scope.id(),
820 weak: scope.downgrade(),
821 }
822 } else {
823 Self::Owned(Box::new(value))
824 }
825 }
826}
827
828#[cfg(test)]
829mod tests {
830 use super::*;
831 use crate::snapshot_v2::take_mutable_snapshot;
832 use crate::snapshot_v2::{reset_runtime_for_tests, TestRuntimeGuard};
833 use crate::state::{NeverEqual, SnapshotMutableState};
834 use std::cell::Cell;
835
836 fn reset_runtime() -> TestRuntimeGuard {
837 reset_runtime_for_tests()
838 }
839
840 #[derive(Clone, Eq, Hash, PartialEq)]
841 struct TestScope(&'static str);
842
843 #[test]
844 fn notifies_scope_when_state_changes() {
845 let _guard = reset_runtime();
846
847 let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
848 let triggered = Rc::new(Cell::new(0));
849 let observer_trigger = triggered.clone();
850
851 let observer = SnapshotStateObserver::new(|callback| callback());
852 observer.start();
853
854 let scope = TestScope("scope");
855 observer.observe_reads(
856 scope.clone(),
857 move |_| {
858 observer_trigger.set(observer_trigger.get() + 1);
859 },
860 || {
861 let _ = state.get();
862 },
863 );
864
865 let snapshot = take_mutable_snapshot(None, None);
866 snapshot.enter(|| {
867 state.set(1);
868 });
869 snapshot.apply().check();
870
871 assert_eq!(triggered.get(), 1);
872 observer.stop();
873 }
874
875 #[test]
876 fn clear_removes_scope_observation() {
877 let _guard = reset_runtime();
878
879 let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
880 let triggered = Rc::new(Cell::new(0));
881 let observer_trigger = triggered.clone();
882
883 let observer = SnapshotStateObserver::new(|callback| callback());
884 observer.start();
885
886 let scope = TestScope("scope");
887 observer.observe_reads(
888 scope.clone(),
889 move |_| {
890 observer_trigger.set(observer_trigger.get() + 1);
891 },
892 || {
893 let _ = state.get();
894 },
895 );
896
897 observer.clear(&scope);
898
899 let snapshot = take_mutable_snapshot(None, None);
900 snapshot.enter(|| {
901 state.set(1);
902 });
903 snapshot.apply().check();
904
905 assert_eq!(triggered.get(), 0);
906 observer.stop();
907 }
908
909 #[test]
910 fn repeated_owned_scope_observations_reuse_the_same_entry() {
911 let _guard = reset_runtime();
912
913 let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
914 let observer = SnapshotStateObserver::new(|callback| callback());
915 let scope = TestScope("scope");
916
917 observer.observe_reads(
918 scope.clone(),
919 |_| {},
920 || {
921 let _ = state.get();
922 },
923 );
924 observer.observe_reads(
925 scope,
926 |_| {},
927 || {
928 let _ = state.get();
929 },
930 );
931
932 let stats = observer.debug_stats();
933 assert_eq!(stats.scopes_len, 1);
934 assert_eq!(stats.fast_scopes_len, 0);
935 }
936
937 #[test]
938 fn with_no_observations_skips_reads() {
939 let _guard = reset_runtime();
940
941 let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
942 let triggered = Rc::new(Cell::new(0));
943 let observer_trigger = triggered.clone();
944
945 let observer = SnapshotStateObserver::new(|callback| callback());
946 observer.start();
947
948 let scope = TestScope("scope");
949 observer.observe_reads(
950 scope.clone(),
951 move |_| {
952 observer_trigger.set(observer_trigger.get() + 1);
953 },
954 || {
955 observer.with_no_observations(|| {
956 let _ = state.get();
957 });
958 },
959 );
960
961 let snapshot = take_mutable_snapshot(None, None);
962 snapshot.enter(|| {
963 state.set(1);
964 });
965 snapshot.apply().check();
966
967 assert_eq!(triggered.get(), 0);
968 observer.stop();
969 }
970
971 #[test]
972 fn nested_observe_reads_attributes_state_to_innermost_scope_only() {
973 let _guard = reset_runtime();
974
975 let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
976 let outer_triggered = Rc::new(Cell::new(0));
977 let inner_triggered = Rc::new(Cell::new(0));
978
979 let observer = SnapshotStateObserver::new(|callback| callback());
980 observer.start();
981
982 let outer_scope = TestScope("outer");
983 let inner_scope = TestScope("inner");
984 observer.observe_reads(
985 outer_scope.clone(),
986 {
987 let outer_triggered = Rc::clone(&outer_triggered);
988 move |_| outer_triggered.set(outer_triggered.get() + 1)
989 },
990 || {
991 observer.observe_reads(
992 inner_scope.clone(),
993 {
994 let inner_triggered = Rc::clone(&inner_triggered);
995 move |_| inner_triggered.set(inner_triggered.get() + 1)
996 },
997 || {
998 let _ = state.get();
999 },
1000 );
1001 },
1002 );
1003
1004 let snapshot = take_mutable_snapshot(None, None);
1005 snapshot.enter(|| {
1006 state.set(1);
1007 });
1008 snapshot.apply().check();
1009
1010 assert_eq!(outer_triggered.get(), 0);
1011 assert_eq!(inner_triggered.get(), 1);
1012 observer.stop();
1013 }
1014
1015 #[test]
1016 fn clearing_one_scope_keeps_shared_state_registered_for_other_scope() {
1017 let _guard = reset_runtime();
1018
1019 let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
1020 let first_triggered = Rc::new(Cell::new(0));
1021 let second_triggered = Rc::new(Cell::new(0));
1022
1023 let observer = SnapshotStateObserver::new(|callback| callback());
1024 observer.start();
1025
1026 let first_scope = TestScope("first");
1027 let second_scope = TestScope("second");
1028 observer.observe_reads(
1029 first_scope.clone(),
1030 {
1031 let first_triggered = Rc::clone(&first_triggered);
1032 move |_| first_triggered.set(first_triggered.get() + 1)
1033 },
1034 || {
1035 let _ = state.get();
1036 },
1037 );
1038 observer.observe_reads(
1039 second_scope.clone(),
1040 {
1041 let second_triggered = Rc::clone(&second_triggered);
1042 move |_| second_triggered.set(second_triggered.get() + 1)
1043 },
1044 || {
1045 let _ = state.get();
1046 },
1047 );
1048
1049 observer.clear(&first_scope);
1050
1051 let snapshot = take_mutable_snapshot(None, None);
1052 snapshot.enter(|| {
1053 state.set(1);
1054 });
1055 snapshot.apply().check();
1056
1057 assert_eq!(first_triggered.get(), 0);
1058 assert_eq!(second_triggered.get(), 1);
1059 observer.stop();
1060 }
1061
1062 #[test]
1063 fn shared_state_notifies_scopes_in_registration_order() {
1064 let _guard = reset_runtime();
1065
1066 let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
1067 let notifications = Rc::new(RefCell::new(Vec::new()));
1068
1069 let observer = SnapshotStateObserver::new(|callback| callback());
1070 observer.start();
1071
1072 observer.observe_reads(
1073 TestScope("first"),
1074 {
1075 let notifications = Rc::clone(¬ifications);
1076 move |_| notifications.borrow_mut().push("first")
1077 },
1078 || {
1079 let _ = state.get();
1080 },
1081 );
1082 observer.observe_reads(
1083 TestScope("second"),
1084 {
1085 let notifications = Rc::clone(¬ifications);
1086 move |_| notifications.borrow_mut().push("second")
1087 },
1088 || {
1089 let _ = state.get();
1090 },
1091 );
1092
1093 let snapshot = take_mutable_snapshot(None, None);
1094 snapshot.enter(|| {
1095 state.set(1);
1096 });
1097 snapshot.apply().check();
1098
1099 assert_eq!(notifications.borrow().as_slice(), &["first", "second"]);
1100 observer.stop();
1101 }
1102
1103 #[test]
1104 fn stateless_recompose_scope_does_not_retain_observer_entry() {
1105 let _guard = reset_runtime();
1106
1107 let observer = SnapshotStateObserver::new(|callback| callback());
1108 let runtime = crate::TestRuntime::new();
1109 let scope = RecomposeScope::new_for_test(runtime.handle());
1110
1111 observer.observe_reads(scope, |_| {}, || {});
1112
1113 let stats = observer.debug_stats();
1114 assert_eq!(stats.scopes_len, 0);
1115 assert_eq!(stats.fast_scopes_len, 0);
1116 assert_eq!(stats.stateless_scope_count, 0);
1117 }
1118
1119 #[test]
1120 fn scope_that_stops_reading_state_is_removed_immediately() {
1121 let _guard = reset_runtime();
1122
1123 let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
1124 let observer = SnapshotStateObserver::new(|callback| callback());
1125 let runtime = crate::TestRuntime::new();
1126 let scope = RecomposeScope::new_for_test(runtime.handle());
1127 let triggered = Rc::new(Cell::new(0));
1128 let observer_trigger = Rc::clone(&triggered);
1129
1130 observer.observe_reads(
1131 scope.clone(),
1132 move |_| observer_trigger.set(observer_trigger.get() + 1),
1133 || {
1134 let _ = state.get();
1135 },
1136 );
1137
1138 let after_stateful = observer.debug_stats();
1139 assert_eq!(after_stateful.scopes_len, 1);
1140 assert_eq!(after_stateful.fast_scopes_len, 1);
1141
1142 observer.observe_reads(scope, |_| {}, || {});
1143
1144 let after_stateless = observer.debug_stats();
1145 assert_eq!(after_stateless.scopes_len, 0);
1146 assert_eq!(after_stateless.fast_scopes_len, 0);
1147
1148 let snapshot = take_mutable_snapshot(None, None);
1149 snapshot.enter(|| {
1150 state.set(1);
1151 });
1152 snapshot.apply().check();
1153
1154 assert_eq!(triggered.get(), 0);
1155 }
1156
1157 #[test]
1158 fn begin_frame_prunes_dropped_recompose_scope_entries() {
1159 let _guard = reset_runtime();
1160
1161 let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
1162 let observer = SnapshotStateObserver::new(|callback| callback());
1163 let runtime = crate::TestRuntime::new();
1164 let scope = RecomposeScope::new_for_test(runtime.handle());
1165
1166 observer.observe_reads(
1167 scope.clone(),
1168 |_| {},
1169 || {
1170 let _ = state.get();
1171 },
1172 );
1173
1174 let before_prune = observer.debug_stats();
1175 assert_eq!(before_prune.scopes_len, 1);
1176 assert_eq!(before_prune.fast_scopes_len, 1);
1177
1178 drop(scope);
1179 observer.begin_frame();
1180
1181 let after_prune = observer.debug_stats();
1182 assert_eq!(after_prune.scopes_len, 0);
1183 assert_eq!(after_prune.fast_scopes_len, 0);
1184 }
1185}