#![allow(clippy::type_complexity)]
use std::{
any::{Any, TypeId},
cell::{Cell, RefCell},
hash::{Hash, Hasher},
rc::{Rc, Weak},
sync::Arc,
};
use smallvec::SmallVec;
use crate::{
RecomposeScope, RecomposeScopeInner, ScopeId,
collections::map::{HashMap, HashSet},
hash::default as default_hash,
snapshot_v2::{
ReadObserver, StateObjectId, TransparentObserverMutableSnapshot, register_apply_observer,
},
state::StateObject,
};
type Executor = dyn Fn(Box<dyn FnOnce() + 'static>) + 'static;
trait ScopeChangedCallback: Fn(&dyn Any) + Any {}
impl<F: Fn(&dyn Any) + Any> ScopeChangedCallback for F {}
#[derive(Clone)]
pub struct SnapshotStateObserver {
inner: Rc<SnapshotStateObserverInner>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct SnapshotStateObserverDebugStats {
pub scopes_len: usize,
pub scopes_cap: usize,
pub fast_scopes_len: usize,
pub fast_scopes_cap: usize,
pub stateless_scope_count: usize,
pub observed_state_count: usize,
pub observed_state_capacity: usize,
}
impl SnapshotStateObserver {
pub fn new(on_changed_executor: impl Fn(Box<dyn FnOnce() + 'static>) + 'static) -> Self {
let inner = Rc::new(SnapshotStateObserverInner::new(on_changed_executor));
inner.set_self(Rc::downgrade(&inner));
Self { inner }
}
pub fn observe_reads<T, R>(
&self,
scope: T,
on_value_changed_for_scope: impl Fn(&T) + 'static,
block: impl FnOnce() -> R,
) -> R
where
T: Any + Clone + Eq + Hash + 'static,
{
self.inner
.observe_reads(scope, on_value_changed_for_scope, block)
}
pub fn begin_frame(&self) {
self.inner.begin_frame();
}
pub fn prune_dead_scopes(&self) {
self.inner.prune_dead_scopes();
}
pub fn with_no_observations<R>(&self, block: impl FnOnce() -> R) -> R {
self.inner.with_no_observations(block)
}
pub fn clear<T>(&self, scope: &T)
where
T: Any + Eq + Hash + 'static,
{
self.inner.clear(scope);
}
pub fn clear_if(&self, predicate: impl Fn(&dyn Any) -> bool) {
self.inner.clear_if(predicate);
}
pub fn clear_all(&self) {
self.inner.clear_all();
}
pub fn start(&self) {
let weak = Rc::downgrade(&self.inner);
self.inner.start(weak);
}
pub fn stop(&self) {
self.inner.stop();
}
pub fn debug_stats(&self) -> SnapshotStateObserverDebugStats {
self.inner.debug_stats()
}
#[cfg(test)]
pub fn notify_changes(&self, modified: &[Arc<dyn StateObject>]) {
self.inner.handle_apply(modified);
}
}
struct SnapshotStateObserverInner {
executor: Rc<Executor>,
owned_scopes: RefCell<HashMap<OwnedScopeIndexKey, OwnedScopeBucket>>,
fast_scopes: RefCell<HashMap<ScopeId, Rc<RefCell<ScopeEntry>>>>,
indexed_scopes: RefCell<HashMap<usize, Rc<RefCell<ScopeEntry>>>>,
observed_to_scopes: RefCell<HashMap<StateObjectId, HashSet<usize>>>,
pause_count: Rc<Cell<usize>>,
active_read_targets: Rc<RefCell<ReadObservationStack>>,
read_dispatcher: ReadObserver,
read_snapshot: RefCell<Option<Arc<TransparentObserverMutableSnapshot>>>,
apply_handle: RefCell<Option<crate::snapshot_v2::ObserverHandle>>,
weak_self: RefCell<Weak<SnapshotStateObserverInner>>,
frame_version: Cell<u64>,
next_entry_id: Cell<usize>,
}
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
struct OwnedScopeIndexKey {
type_id: TypeId,
value_hash: u64,
}
type OwnedScopeBucket = SmallVec<[Rc<RefCell<ScopeEntry>>; 1]>;
fn owned_scope_index_key<T>(scope: &T) -> OwnedScopeIndexKey
where
T: Any + Hash + 'static,
{
let mut hasher = default_hash::new();
scope.hash(&mut hasher);
OwnedScopeIndexKey {
type_id: TypeId::of::<T>(),
value_hash: hasher.finish(),
}
}
impl SnapshotStateObserverInner {
const MIN_RETAINED_SCOPE_CAPACITY: usize = 256;
fn new(on_changed_executor: impl Fn(Box<dyn FnOnce() + 'static>) + 'static) -> Self {
let pause_count = Rc::new(Cell::new(0));
let active_read_targets = Rc::new(RefCell::new(ReadObservationStack::default()));
let dispatcher_pause_count = Rc::clone(&pause_count);
let dispatcher_targets = Rc::clone(&active_read_targets);
let read_dispatcher: ReadObserver = Arc::new(move |state| {
if dispatcher_pause_count.get() > 0 {
return;
}
let observed = dispatcher_targets.borrow().last().cloned();
if let Some(observed) = observed {
observed.borrow_mut().insert(state);
}
});
Self {
executor: Rc::new(on_changed_executor),
owned_scopes: RefCell::new(HashMap::default()),
fast_scopes: RefCell::new(HashMap::default()),
indexed_scopes: RefCell::new(HashMap::default()),
observed_to_scopes: RefCell::new(HashMap::default()),
pause_count,
active_read_targets,
read_dispatcher,
read_snapshot: RefCell::new(None),
apply_handle: RefCell::new(None),
weak_self: RefCell::new(Weak::new()),
frame_version: Cell::new(0),
next_entry_id: Cell::new(0),
}
}
fn set_self(&self, weak: Weak<SnapshotStateObserverInner>) {
self.weak_self.replace(weak);
}
fn begin_frame(&self) {
let next = self.frame_version.get().wrapping_add(1);
self.frame_version.set(next);
self.prune_dead_scopes();
}
fn observe_reads<T, R>(
&self,
scope: T,
on_value_changed_for_scope: impl Fn(&T) + 'static,
block: impl FnOnce() -> R,
) -> R
where
T: Any + Clone + Eq + Hash + 'static,
{
let frame_version = self.frame_version.get();
let has_frame_version = frame_version != 0;
let existing_entry = self.find_scope_entry(&scope);
let on_changed = std::cell::LazyCell::new(|| {
let callback = move |scope_any: &dyn Any| {
if let Some(typed) = scope_any.downcast_ref::<T>() {
on_value_changed_for_scope(typed);
}
};
match existing_entry.as_ref() {
Some(entry) => entry.borrow_mut().callback_reusing(callback),
None => Rc::new(callback),
}
});
if let Some(entry) = existing_entry.as_ref() {
let already_observed = {
let mut entry_mut = entry.borrow_mut();
entry_mut.update_scope(scope.clone());
has_frame_version && entry_mut.last_seen_version == frame_version
};
let callback = on_changed.clone();
entry.borrow_mut().on_changed = callback;
if already_observed {
return block();
}
}
let observed = self.active_read_targets.borrow_mut().push();
struct ActiveObservationGuard {
stack: Rc<RefCell<ReadObservationStack>>,
}
impl Drop for ActiveObservationGuard {
fn drop(&mut self) {
let target = self.stack.borrow_mut().pop();
let discarded = target.replace(ObservedIds::new());
drop(discarded);
}
}
let _guard = ActiveObservationGuard {
stack: Rc::clone(&self.active_read_targets),
};
let result = self.run_with_read_observer(block);
if observed.borrow().is_empty() {
if existing_entry.is_some() {
self.clear(&scope);
}
return result;
}
let observed = {
let mut observed = observed.borrow_mut();
std::mem::replace(&mut *observed, ObservedIds::new())
};
let entry = existing_entry
.clone()
.unwrap_or_else(|| self.insert_scope_entry(scope.clone(), on_changed.clone()));
{
let mut entry_mut = entry.borrow_mut();
entry_mut.update(scope, Rc::clone(&on_changed));
entry_mut.last_seen_version = if has_frame_version {
frame_version
} else {
u64::MAX
};
}
self.replace_observed_ids(&entry, observed);
result
}
fn with_no_observations<R>(&self, block: impl FnOnce() -> R) -> R {
self.pause_count.set(self.pause_count.get() + 1);
let result = block();
self.pause_count
.set(self.pause_count.get().saturating_sub(1));
result
}
fn clear<T>(&self, scope: &T)
where
T: Any + Eq + Hash + 'static,
{
if let Some(rc_scope) = (scope as &dyn Any).downcast_ref::<RecomposeScope>() {
if let Some(entry) = self.fast_scopes.borrow_mut().remove(&rc_scope.id()) {
self.unregister_entry(&entry);
}
return;
}
let removed = self.remove_owned_scope_entry(scope);
if let Some(entry) = removed {
self.unregister_entry(&entry);
}
}
fn clear_if(&self, predicate: impl Fn(&dyn Any) -> bool) {
let removed_fast = {
let mut fast_scopes = self.fast_scopes.borrow_mut();
let removed_ids: Vec<_> = fast_scopes
.iter()
.filter(|(_, entry)| entry.borrow().matches_predicate(&predicate))
.map(|(scope_id, _)| *scope_id)
.collect();
removed_ids
.into_iter()
.filter_map(|scope_id| fast_scopes.remove(&scope_id))
.collect::<Vec<_>>()
};
let removed_owned =
{ self.partition_owned_scopes(|entry| entry.matches_predicate(&predicate)) };
for entry in removed_fast.into_iter().chain(removed_owned) {
self.unregister_entry(&entry);
}
}
fn clear_all(&self) {
self.fast_scopes.borrow_mut().clear();
self.owned_scopes.borrow_mut().clear();
self.indexed_scopes.borrow_mut().clear();
self.observed_to_scopes.borrow_mut().clear();
}
fn start(&self, weak_self: Weak<SnapshotStateObserverInner>) {
if self.apply_handle.borrow().is_some() {
return;
}
let handle = register_apply_observer(Rc::new(move |modified, _snapshot_id| {
if let Some(inner) = weak_self.upgrade() {
inner.handle_apply(modified);
}
}));
self.apply_handle.replace(Some(handle));
}
fn stop(&self) {
if let Some(handle) = self.apply_handle.borrow_mut().take() {
drop(handle);
}
}
fn find_scope_entry<T>(&self, scope: &T) -> Option<Rc<RefCell<ScopeEntry>>>
where
T: Any + Eq + Hash + 'static,
{
if let Some(scope) = (scope as &dyn Any).downcast_ref::<RecomposeScope>() {
return self.fast_scopes.borrow().get(&scope.id()).cloned();
}
self.find_owned_scope_entry(scope)
}
fn insert_scope_entry(
&self,
scope: impl Any + Clone + Eq + Hash + 'static,
on_changed: Rc<dyn ScopeChangedCallback>,
) -> Rc<RefCell<ScopeEntry>> {
let entry_id = self.next_entry_id.get();
self.next_entry_id.set(entry_id.wrapping_add(1));
let recompose_scope_id = (&scope as &dyn Any)
.downcast_ref::<RecomposeScope>()
.map(RecomposeScope::id);
let owned_scope_key = recompose_scope_id
.is_none()
.then(|| owned_scope_index_key(&scope));
let entry = Rc::new(RefCell::new(ScopeEntry::new(entry_id, scope, on_changed)));
self.indexed_scopes
.borrow_mut()
.insert(entry_id, Rc::clone(&entry));
if let Some(scope_id) = recompose_scope_id {
self.fast_scopes
.borrow_mut()
.insert(scope_id, Rc::clone(&entry));
} else if let Some(scope_key) = owned_scope_key {
self.owned_scopes
.borrow_mut()
.entry(scope_key)
.or_default()
.push(Rc::clone(&entry));
}
entry
}
fn prune_dead_scopes(&self) {
let removed_fast = {
let mut fast_scopes = self.fast_scopes.borrow_mut();
let removed_ids: Vec<_> = fast_scopes
.iter()
.filter(|(_, entry)| !entry.borrow().should_retain())
.map(|(scope_id, _)| *scope_id)
.collect();
let removed = removed_ids
.into_iter()
.filter_map(|scope_id| fast_scopes.remove(&scope_id))
.collect::<Vec<_>>();
shrink_map_if_sparse(&mut fast_scopes, Self::MIN_RETAINED_SCOPE_CAPACITY);
removed
};
let removed_owned = { self.partition_owned_scopes(|entry| !entry.should_retain()) };
for entry in removed_fast.into_iter().chain(removed_owned) {
self.unregister_entry(&entry);
}
}
fn find_owned_scope_entry<T>(&self, scope: &T) -> Option<Rc<RefCell<ScopeEntry>>>
where
T: Any + Eq + Hash + 'static,
{
let key = owned_scope_index_key(scope);
self.owned_scopes.borrow().get(&key).and_then(|bucket| {
bucket
.iter()
.find(|entry| entry.borrow().matches_scope(scope))
.cloned()
})
}
fn remove_owned_scope_entry<T>(&self, scope: &T) -> Option<Rc<RefCell<ScopeEntry>>>
where
T: Any + Eq + Hash + 'static,
{
let key = owned_scope_index_key(scope);
let mut owned_scopes = self.owned_scopes.borrow_mut();
let mut removed = None;
let mut remove_bucket = false;
if let Some(bucket) = owned_scopes.get_mut(&key)
&& let Some(index) = bucket
.iter()
.position(|entry| entry.borrow().matches_scope(scope))
{
removed = Some(bucket.remove(index));
remove_bucket = bucket.is_empty();
}
if remove_bucket {
owned_scopes.remove(&key);
}
shrink_map_if_sparse(&mut owned_scopes, Self::MIN_RETAINED_SCOPE_CAPACITY);
removed
}
fn partition_owned_scopes(
&self,
should_remove: impl Fn(&ScopeEntry) -> bool,
) -> Vec<Rc<RefCell<ScopeEntry>>> {
let mut owned_scopes = self.owned_scopes.borrow_mut();
let mut retained = HashMap::default();
let mut removed = Vec::new();
for (key, mut bucket) in owned_scopes.drain() {
let mut retained_bucket = OwnedScopeBucket::new();
for entry in bucket.drain(..) {
if should_remove(&entry.borrow()) {
removed.push(entry);
} else {
retained_bucket.push(entry);
}
}
if !retained_bucket.is_empty() {
retained.insert(key, retained_bucket);
}
}
*owned_scopes = retained;
shrink_map_if_sparse(&mut owned_scopes, Self::MIN_RETAINED_SCOPE_CAPACITY);
removed
}
fn debug_stats(&self) -> SnapshotStateObserverDebugStats {
let owned_scopes = self.owned_scopes.borrow();
let fast_scopes = self.fast_scopes.borrow();
let owned_scope_len = owned_scopes.values().map(SmallVec::len).sum::<usize>();
let owned_scope_cap =
owned_scopes.capacity() + owned_scopes.values().map(SmallVec::capacity).sum::<usize>();
let scopes_len = owned_scope_len + fast_scopes.len();
let scopes_cap = owned_scope_cap + fast_scopes.capacity();
let mut observed_state_count = 0;
let mut observed_state_capacity = 0;
let mut stateless_scope_count = 0;
for entry in owned_scopes
.values()
.flat_map(|bucket| bucket.iter())
.chain(fast_scopes.values())
{
let entry = entry.borrow();
observed_state_count += entry.observed.len();
observed_state_capacity += entry.observed.capacity();
stateless_scope_count += usize::from(entry.observed.is_empty());
}
SnapshotStateObserverDebugStats {
scopes_len,
scopes_cap,
fast_scopes_len: fast_scopes.len(),
fast_scopes_cap: fast_scopes.capacity(),
stateless_scope_count,
observed_state_count,
observed_state_capacity,
}
}
fn run_with_read_observer<R>(&self, block: impl FnOnce() -> R) -> R {
use crate::snapshot_v2::take_transparent_observer_mutable_snapshot_reusing;
let mut snapshot = take_transparent_observer_mutable_snapshot_reusing(
Some(self.read_dispatcher.clone()),
None,
self.read_snapshot.take(),
);
let result = snapshot.enter(block);
snapshot.dispose();
if Arc::get_mut(&mut snapshot).is_some() && !snapshot.has_pending_changes() {
self.read_snapshot.replace(Some(snapshot));
}
result
}
fn handle_apply(&self, modified: &[Arc<dyn StateObject>]) {
if modified.is_empty() {
return;
}
let mut seen_scope_ids: HashSet<usize> = HashSet::default();
let mut to_notify: Vec<Rc<RefCell<ScopeEntry>>> = Vec::new();
{
let observed_to_scopes = self.observed_to_scopes.borrow();
let indexed_scopes = self.indexed_scopes.borrow();
for state in modified {
if let Some(scope_ids) = observed_to_scopes.get(&state.object_id().as_usize()) {
let mut ordered_scope_ids: SmallVec<[usize; 8]> =
scope_ids.iter().copied().collect();
ordered_scope_ids.sort_unstable();
for scope_id in ordered_scope_ids {
if seen_scope_ids.insert(scope_id)
&& let Some(entry) = indexed_scopes.get(&scope_id)
{
to_notify.push(entry.clone());
}
}
}
}
}
if to_notify.is_empty() {
return;
}
for entry in to_notify {
let executor = self.executor.clone();
executor(Box::new(move || {
if let Ok(entry) = entry.try_borrow() {
entry.notify();
}
}));
}
}
fn replace_observed_ids(&self, entry: &Rc<RefCell<ScopeEntry>>, observed: ObservedIds) {
let (entry_id, previous) = {
let mut entry_mut = entry.borrow_mut();
let entry_id = entry_mut.id;
let previous = std::mem::replace(&mut entry_mut.observed, observed);
(entry_id, previous)
};
let entry_ref = entry.borrow();
if previous.iter().eq(entry_ref.observed.iter()) {
return;
}
self.unregister_observed_ids(entry_id, &previous);
self.register_observed_ids(entry_id, &entry_ref.observed);
}
fn register_observed_ids(&self, entry_id: usize, observed: &ObservedIds) {
let mut observed_to_scopes = self.observed_to_scopes.borrow_mut();
for state_id in observed.iter() {
let scope_ids = observed_to_scopes.entry(state_id).or_default();
scope_ids.insert(entry_id);
}
}
fn unregister_observed_ids(&self, entry_id: usize, observed: &ObservedIds) {
let mut observed_to_scopes = self.observed_to_scopes.borrow_mut();
let mut emptied = SmallVec::<[StateObjectId; MAX_OBSERVED_STATES]>::new();
for state_id in observed.iter() {
if let Some(scope_ids) = observed_to_scopes.get_mut(&state_id) {
scope_ids.remove(&entry_id);
if scope_ids.is_empty() {
emptied.push(state_id);
}
}
}
for state_id in emptied {
observed_to_scopes.remove(&state_id);
}
shrink_map_if_sparse(&mut observed_to_scopes, Self::MIN_RETAINED_SCOPE_CAPACITY);
}
fn unregister_entry(&self, entry: &Rc<RefCell<ScopeEntry>>) {
let (entry_id, observed) = {
let mut entry_mut = entry.borrow_mut();
let observed = std::mem::replace(&mut entry_mut.observed, ObservedIds::new());
(entry_mut.id, observed)
};
self.unregister_observed_ids(entry_id, &observed);
self.indexed_scopes.borrow_mut().remove(&entry_id);
}
}
fn shrink_map_if_sparse<K, V>(map: &mut HashMap<K, V>, min_retained_capacity: usize)
where
K: Eq + std::hash::Hash,
{
if map.capacity() <= map.len().max(min_retained_capacity).saturating_mul(4) {
return;
}
let retained = map.len().max(min_retained_capacity);
let mut rebuilt = HashMap::default();
rebuilt.reserve(retained);
rebuilt.extend(map.drain());
*map = rebuilt;
}
#[derive(Default)]
struct ReadObservationStack {
targets: Vec<Rc<RefCell<ObservedIds>>>,
depth: usize,
}
impl ReadObservationStack {
fn push(&mut self) -> Rc<RefCell<ObservedIds>> {
if self.depth == self.targets.len() {
self.targets.push(Rc::new(RefCell::new(ObservedIds::new())));
}
let target = Rc::clone(&self.targets[self.depth]);
self.depth += 1;
target
}
fn last(&self) -> Option<&Rc<RefCell<ObservedIds>>> {
self.depth.checked_sub(1).map(|index| &self.targets[index])
}
fn pop(&mut self) -> Rc<RefCell<ObservedIds>> {
self.depth -= 1;
Rc::clone(&self.targets[self.depth])
}
}
enum ObservedIds {
Small(SmallVec<[ObservedState; MAX_OBSERVED_STATES]>),
Large(HashMap<StateObjectId, Option<Rc<dyn Any>>>),
}
struct ObservedState {
id: StateObjectId,
_lease: Option<Rc<dyn Any>>,
}
impl ObservedIds {
fn new() -> Self {
ObservedIds::Small(SmallVec::new())
}
fn insert(&mut self, state: &dyn StateObject) {
let id = state.object_id().as_usize();
match self {
ObservedIds::Small(small) => {
if small.iter().any(|observed| observed.id == id) {
return;
}
if small.len() < MAX_OBSERVED_STATES {
small.push(ObservedState {
id,
_lease: state.observation_lease(),
});
} else {
let mut large =
HashMap::with_capacity_and_hasher(small.len() + 1, Default::default());
for observed in small.drain(..) {
large.insert(observed.id, observed._lease);
}
large.insert(id, state.observation_lease());
*self = ObservedIds::Large(large);
}
}
ObservedIds::Large(large) => {
large.entry(id).or_insert_with(|| state.observation_lease());
}
}
}
fn is_empty(&self) -> bool {
match self {
ObservedIds::Small(small) => small.is_empty(),
ObservedIds::Large(large) => large.is_empty(),
}
}
fn len(&self) -> usize {
match self {
ObservedIds::Small(small) => small.len(),
ObservedIds::Large(large) => large.len(),
}
}
fn capacity(&self) -> usize {
match self {
ObservedIds::Small(small) => small.capacity(),
ObservedIds::Large(large) => large.capacity(),
}
}
fn iter(&self) -> impl Iterator<Item = StateObjectId> + '_ {
let (small, large) = match self {
ObservedIds::Small(small) => (Some(small.as_slice()), None),
ObservedIds::Large(large) => (None, Some(large)),
};
small
.into_iter()
.flatten()
.map(|observed| observed.id)
.chain(large.into_iter().flat_map(|states| states.keys().copied()))
}
}
const MAX_OBSERVED_STATES: usize = 8;
enum ScopeStorage {
Owned(Box<dyn Any>),
RecomposeScope {
id: ScopeId,
weak: Weak<RecomposeScopeInner>,
},
}
struct ScopeEntry {
id: usize,
scope: ScopeStorage,
on_changed: Rc<dyn ScopeChangedCallback>,
observed: ObservedIds,
last_seen_version: u64,
}
impl ScopeEntry {
fn new<T>(id: usize, scope: T, on_changed: Rc<dyn ScopeChangedCallback>) -> Self
where
T: Any + 'static,
{
Self {
id,
scope: ScopeStorage::from_value(scope),
on_changed,
observed: ObservedIds::new(),
last_seen_version: u64::MAX,
}
}
fn callback_reusing<F: Fn(&dyn Any) + 'static>(
&mut self,
callback: F,
) -> Rc<dyn ScopeChangedCallback> {
if let Some(stored) = Rc::get_mut(&mut self.on_changed)
.and_then(|stored| (stored as &mut dyn Any).downcast_mut::<F>())
{
*stored = callback;
Rc::clone(&self.on_changed)
} else {
Rc::new(callback)
}
}
fn update<T>(&mut self, new_scope: T, on_changed: Rc<dyn ScopeChangedCallback>)
where
T: Any + 'static,
{
self.update_scope(new_scope);
self.on_changed = on_changed;
}
fn update_scope<T>(&mut self, new_scope: T)
where
T: Any + 'static,
{
if let ScopeStorage::Owned(stored) = &mut self.scope
&& let Some(stored) = stored.downcast_mut::<T>()
{
*stored = new_scope;
} else {
self.scope = ScopeStorage::from_value(new_scope);
}
}
fn matches_scope<T>(&self, scope: &T) -> bool
where
T: Any + Eq + 'static,
{
if let Some(scope) = (scope as &dyn Any).downcast_ref::<RecomposeScope>() {
return matches!(
&self.scope,
ScopeStorage::RecomposeScope { id, .. } if *id == scope.id()
);
}
match &self.scope {
ScopeStorage::Owned(stored) => stored
.downcast_ref::<T>()
.map(|stored| stored == scope)
.unwrap_or(false),
ScopeStorage::RecomposeScope { .. } => false,
}
}
fn matches_predicate(&self, predicate: &impl Fn(&dyn Any) -> bool) -> bool {
match &self.scope {
ScopeStorage::Owned(scope) => predicate(scope.as_ref()),
ScopeStorage::RecomposeScope { weak, .. } => weak
.upgrade()
.map(|inner| predicate(&RecomposeScope { inner }))
.unwrap_or(true),
}
}
fn should_retain(&self) -> bool {
match &self.scope {
ScopeStorage::Owned(_) => true,
ScopeStorage::RecomposeScope { weak, .. } => weak.upgrade().is_some(),
}
}
fn notify(&self) {
match &self.scope {
ScopeStorage::Owned(scope) => (self.on_changed)(scope.as_ref()),
ScopeStorage::RecomposeScope { weak, .. } => {
if let Some(inner) = weak.upgrade() {
(self.on_changed)(&RecomposeScope { inner });
}
}
}
}
}
impl ScopeStorage {
fn from_value<T>(value: T) -> Self
where
T: Any + 'static,
{
let any = &value as &dyn Any;
if let Some(scope) = any.downcast_ref::<RecomposeScope>() {
Self::RecomposeScope {
id: scope.id(),
weak: scope.downgrade(),
}
} else {
Self::Owned(Box::new(value))
}
}
}
#[cfg(test)]
mod tests {
use std::cell::Cell;
use super::*;
use crate::{
snapshot_v2::{TestRuntimeGuard, reset_runtime_for_tests, take_mutable_snapshot},
state::{NeverEqual, SnapshotMutableState},
};
fn reset_runtime() -> TestRuntimeGuard {
reset_runtime_for_tests()
}
#[derive(Clone, Eq, Hash, PartialEq)]
struct TestScope(&'static str);
#[test]
fn scope_update_reuses_storage_and_replaces_payload_and_callback() {
let first = Rc::new(String::from("first"));
let second = Rc::new(String::from("second"));
let delivered = Rc::new(RefCell::new(Vec::new()));
let mut entry = ScopeEntry::new(0, first.clone(), Rc::new(|_| panic!("stale callback")));
let ScopeStorage::Owned(stored) = &entry.scope else {
panic!("expected owned scope");
};
let address = stored.downcast_ref::<Rc<String>>().unwrap() as *const Rc<String>;
let received = delivered.clone();
entry.update(
second.clone(),
Rc::new(move |scope| {
received.borrow_mut().push(
scope
.downcast_ref::<Rc<String>>()
.unwrap()
.as_str()
.to_owned(),
);
}),
);
assert_eq!(Rc::strong_count(&first), 1);
assert_eq!(Rc::strong_count(&second), 2);
entry.notify();
assert_eq!(*delivered.borrow(), vec!["second"]);
let ScopeStorage::Owned(stored) = &entry.scope else {
panic!("expected owned scope");
};
assert_eq!(
stored.downcast_ref::<Rc<String>>().unwrap() as *const Rc<String>,
address
);
drop(entry);
assert_eq!(Rc::strong_count(&second), 1);
}
#[test]
fn reobservation_refreshes_captures_and_preserves_shared_callbacks() {
let _guard = reset_runtime();
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let delivered = Rc::new(RefCell::new(Vec::new()));
let callback = |generation| {
let delivered = delivered.clone();
move |scope: &TestScope| delivered.borrow_mut().push((generation, scope.0))
};
let scope = TestScope("callback");
let observer = SnapshotStateObserver::new(|callback| callback());
let read = || {
let _ = state.get();
};
observer.observe_reads(scope.clone(), callback(1), read);
let entry = observer.inner.find_scope_entry(&scope).unwrap();
let held = entry.borrow().on_changed.clone();
observer.observe_reads(scope.clone(), callback(2), read);
held(&scope);
entry.borrow().notify();
drop(held);
let allocation = Rc::as_ptr(&entry.borrow().on_changed);
observer.observe_reads(scope.clone(), callback(3), read);
assert!(std::ptr::addr_eq(
allocation,
Rc::as_ptr(&entry.borrow().on_changed)
));
entry.borrow().notify();
let received = delivered.clone();
observer.observe_reads(
scope,
move |scope| received.borrow_mut().push((4, scope.0)),
read,
);
entry.borrow().notify();
assert_eq!(
*delivered.borrow(),
[
(1, "callback"),
(2, "callback"),
(3, "callback"),
(4, "callback")
]
);
}
#[test]
fn reobservation_across_storage_thresholds_replaces_dependencies_and_callbacks() {
let _guard = reset_runtime();
let states: Vec<_> = (0..MAX_OBSERVED_STATES + 2)
.map(|_| SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual)))
.collect();
let notifications = Rc::new(RefCell::new(Vec::new()));
let observer = SnapshotStateObserver::new(|callback| callback());
observer.start();
for (generation, count) in [MAX_OBSERVED_STATES, MAX_OBSERVED_STATES + 1, 2, 0]
.into_iter()
.enumerate()
{
observer.begin_frame();
let recorded = notifications.clone();
observer.observe_reads(
TestScope("changing"),
move |scope| {
assert_eq!(scope.0, "changing");
recorded.borrow_mut().push(generation);
},
|| {
for state in states.iter().take(count) {
let _ = state.get();
let _ = state.get();
}
},
);
notifications.borrow_mut().clear();
for (index, state) in states.iter().enumerate() {
let snapshot = take_mutable_snapshot(None, None);
snapshot.enter(|| state.set(generation as i32));
snapshot.apply().check();
let expected = (index + 1).min(count);
assert_eq!(
*notifications.borrow(),
vec![generation; expected],
"count={count}, changed state={index}"
);
}
}
}
#[test]
fn reobservation_preserves_notifications_when_dependencies_repeat_or_change() {
let _guard = reset_runtime();
let states: Vec<_> = (0..3)
.map(|_| SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual)))
.collect();
let notifications = Rc::new(RefCell::new(Vec::new()));
let observer = SnapshotStateObserver::new(|callback| callback());
observer.start();
for (generation, indices) in [[0, 1], [0, 1], [1, 0], [1, 2], [1, 2]]
.into_iter()
.enumerate()
{
observer.begin_frame();
let received = notifications.clone();
observer.observe_reads(
TestScope("repeated"),
move |_| received.borrow_mut().push(generation),
|| {
for index in indices {
let _ = states[index].get();
}
},
);
for (index, state) in states.iter().enumerate() {
notifications.borrow_mut().clear();
let snapshot = take_mutable_snapshot(None, None);
snapshot.enter(|| state.set(generation as i32));
snapshot.apply().check();
assert_eq!(
*notifications.borrow(),
if indices.contains(&index) {
vec![generation]
} else {
vec![]
},
"generation={generation}, state={index}"
);
}
}
observer.clear(&TestScope("repeated"));
notifications.borrow_mut().clear();
for state in states {
observer.notify_changes(&[state]);
}
assert!(notifications.borrow().is_empty());
}
#[test]
fn stateless_scope_can_start_observing_and_replace_its_callback_before_the_block() {
let _guard = reset_runtime();
let observer = SnapshotStateObserver::new(|callback| callback());
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let observed: Arc<dyn StateObject> = state.clone();
let notifications = Rc::new(RefCell::new(Vec::new()));
let discarded = notifications.clone();
observer.observe_reads(
TestScope("changing"),
move |_| discarded.borrow_mut().push(0),
|| {},
);
assert_eq!(Rc::strong_count(¬ifications), 1);
assert_eq!(observer.debug_stats().scopes_len, 0);
for generation in 1..=2 {
observer.begin_frame();
let delivered = notifications.clone();
observer.observe_reads(
TestScope("changing"),
move |_| delivered.borrow_mut().push(generation),
|| {
observer.notify_changes(std::slice::from_ref(&observed));
let _ = state.get();
},
);
observer.notify_changes(std::slice::from_ref(&observed));
}
assert_eq!(*notifications.borrow(), vec![1, 2, 2]);
observer.clear(&TestScope("changing"));
assert_eq!(Rc::strong_count(¬ifications), 1);
}
#[test]
fn callback_captures_are_released_on_replacement_and_clear() {
let _guard = reset_runtime();
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let observer = SnapshotStateObserver::new(|callback| callback());
let owners = [Rc::new(Cell::new(0)), Rc::new(Cell::new(0))];
for owner in &owners {
let captured = owner.clone();
observer.observe_reads(
TestScope("owner"),
move |_| captured.set(captured.get() + 1),
|| {
let _ = state.get();
},
);
assert_eq!(Rc::strong_count(owner), 2);
}
assert_eq!(Rc::strong_count(&owners[0]), 1);
observer.clear(&TestScope("owner"));
assert_eq!(Rc::strong_count(&owners[1]), 1);
}
#[test]
fn notifies_scope_when_state_changes() {
let _guard = reset_runtime();
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let triggered = Rc::new(Cell::new(0));
let observer_trigger = triggered.clone();
let observer = SnapshotStateObserver::new(|callback| callback());
observer.start();
let scope = TestScope("scope");
observer.observe_reads(
scope.clone(),
move |_| {
observer_trigger.set(observer_trigger.get() + 1);
},
|| {
let _ = state.get();
},
);
let snapshot = take_mutable_snapshot(None, None);
snapshot.enter(|| {
state.set(1);
});
snapshot.apply().check();
assert_eq!(triggered.get(), 1);
observer.stop();
}
#[test]
fn clear_removes_scope_observation() {
let _guard = reset_runtime();
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let triggered = Rc::new(Cell::new(0));
let observer_trigger = triggered.clone();
let observer = SnapshotStateObserver::new(|callback| callback());
observer.start();
let scope = TestScope("scope");
observer.observe_reads(
scope.clone(),
move |_| {
observer_trigger.set(observer_trigger.get() + 1);
},
|| {
let _ = state.get();
},
);
observer.clear(&scope);
let snapshot = take_mutable_snapshot(None, None);
snapshot.enter(|| {
state.set(1);
});
snapshot.apply().check();
assert_eq!(triggered.get(), 0);
observer.stop();
}
#[test]
fn repeated_owned_scope_observations_reuse_the_same_entry() {
let _guard = reset_runtime();
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let observer = SnapshotStateObserver::new(|callback| callback());
let scope = TestScope("scope");
observer.observe_reads(
scope.clone(),
|_| {},
|| {
let _ = state.get();
},
);
observer.observe_reads(
scope,
|_| {},
|| {
let _ = state.get();
},
);
let stats = observer.debug_stats();
assert_eq!(stats.scopes_len, 1);
assert_eq!(stats.fast_scopes_len, 0);
}
#[test]
fn with_no_observations_skips_reads() {
let _guard = reset_runtime();
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let triggered = Rc::new(Cell::new(0));
let observer_trigger = triggered.clone();
let observer = SnapshotStateObserver::new(|callback| callback());
observer.start();
let scope = TestScope("scope");
observer.observe_reads(
scope.clone(),
move |_| {
observer_trigger.set(observer_trigger.get() + 1);
},
|| {
observer.with_no_observations(|| {
let _ = state.get();
});
},
);
let snapshot = take_mutable_snapshot(None, None);
snapshot.enter(|| {
state.set(1);
});
snapshot.apply().check();
assert_eq!(triggered.get(), 0);
observer.stop();
}
#[test]
fn recycled_observation_refreshes_snapshot_state() {
let _guard = reset_runtime();
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let observer = SnapshotStateObserver::new(|callback| callback());
let mut allocation = None;
for value in 1..=3 {
let parent = take_mutable_snapshot(None, None);
parent.enter(|| {
state.set(value);
let expected = crate::snapshot_v2::current_snapshot().unwrap();
observer.inner.run_with_read_observer(|| {
let crate::snapshot_v2::AnySnapshot::TransparentMutable(current) =
crate::snapshot_v2::current_snapshot().unwrap()
else {
panic!("expected an observation snapshot");
};
assert_eq!(current.snapshot_id(), expected.snapshot_id());
assert_eq!(current.invalid(), expected.invalid());
assert!(!current.is_disposed());
assert!(!current.has_pending_changes());
assert_eq!(state.get(), value);
let address = Arc::as_ptr(¤t) as usize;
assert_eq!(*allocation.get_or_insert(address), address);
});
});
parent.apply().check();
}
}
#[test]
fn recycled_observation_preserves_escaped_snapshots() {
let _guard = reset_runtime();
let observer = SnapshotStateObserver::new(|callback| callback());
let escaped = observer
.inner
.run_with_read_observer(|| crate::snapshot_v2::current_snapshot().unwrap());
let id = escaped.snapshot_id();
observer.inner.run_with_read_observer(|| {
let current = crate::snapshot_v2::current_snapshot().unwrap();
let crate::snapshot_v2::AnySnapshot::TransparentMutable(escaped) = &escaped else {
panic!("expected an observation snapshot");
};
assert!(!current.is_same_transparent(escaped));
assert_eq!(escaped.snapshot_id(), id);
assert!(escaped.is_disposed());
});
let weak = observer.inner.run_with_read_observer(|| {
let crate::snapshot_v2::AnySnapshot::TransparentMutable(current) =
crate::snapshot_v2::current_snapshot().unwrap()
else {
panic!("expected an observation snapshot");
};
Arc::downgrade(¤t)
});
assert!(weak.upgrade().is_none());
observer.inner.run_with_read_observer(|| {
assert!(weak.upgrade().is_none());
});
}
#[test]
fn recycled_observation_does_not_retain_written_state() {
let _guard = reset_runtime();
let observer = SnapshotStateObserver::new(|callback| callback());
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let owners = Arc::strong_count(&state);
observer.inner.run_with_read_observer(|| {
let current = crate::snapshot_v2::current_snapshot().unwrap();
current.record_write(state.clone());
});
assert_eq!(Arc::strong_count(&state), owners);
}
#[test]
fn nested_observe_reads_attributes_state_to_innermost_scope_only() {
let _guard = reset_runtime();
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let outer_state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let outer_triggered = Rc::new(Cell::new(0));
let inner_triggered = Rc::new(Cell::new(0));
let observer = SnapshotStateObserver::new(|callback| callback());
observer.start();
let outer_scope = TestScope("outer");
let inner_scope = TestScope("inner");
observer.observe_reads(
outer_scope.clone(),
{
let outer_triggered = Rc::clone(&outer_triggered);
move |_| outer_triggered.set(outer_triggered.get() + 1)
},
|| {
let _ = outer_state.get();
observer.observe_reads(
inner_scope.clone(),
{
let inner_triggered = Rc::clone(&inner_triggered);
move |_| inner_triggered.set(inner_triggered.get() + 1)
},
|| {
let _ = state.get();
},
);
},
);
let snapshot = take_mutable_snapshot(None, None);
snapshot.enter(|| {
state.set(1);
});
snapshot.apply().check();
assert_eq!(outer_triggered.get(), 0);
assert_eq!(inner_triggered.get(), 1);
let snapshot = take_mutable_snapshot(None, None);
snapshot.enter(|| outer_state.set(1));
snapshot.apply().check();
assert_eq!(outer_triggered.get(), 1);
assert_eq!(inner_triggered.get(), 1);
observer.stop();
}
#[test]
fn unwound_observation_does_not_leak_reads_into_reused_storage() {
let _guard = reset_runtime();
let abandoned = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let live = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let triggered = Rc::new(Cell::new(0));
let observer = SnapshotStateObserver::new(|callback| callback());
observer.start();
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
observer.observe_reads(
TestScope("abandoned"),
|_| {},
|| {
let _ = abandoned.get();
panic!("abandon observation");
},
);
}));
assert!(result.is_err());
observer.observe_reads(
TestScope("live"),
{
let triggered = Rc::clone(&triggered);
move |_| triggered.set(triggered.get() + 1)
},
|| {
let _ = live.get();
},
);
let snapshot = take_mutable_snapshot(None, None);
snapshot.enter(|| abandoned.set(1));
snapshot.apply().check();
assert_eq!(triggered.get(), 0);
let snapshot = take_mutable_snapshot(None, None);
snapshot.enter(|| live.set(1));
snapshot.apply().check();
assert_eq!(triggered.get(), 1);
observer.stop();
}
#[test]
fn clearing_one_scope_keeps_shared_state_registered_for_other_scope() {
let _guard = reset_runtime();
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let first_triggered = Rc::new(Cell::new(0));
let second_triggered = Rc::new(Cell::new(0));
let observer = SnapshotStateObserver::new(|callback| callback());
observer.start();
let first_scope = TestScope("first");
let second_scope = TestScope("second");
observer.observe_reads(
first_scope.clone(),
{
let first_triggered = Rc::clone(&first_triggered);
move |_| first_triggered.set(first_triggered.get() + 1)
},
|| {
let _ = state.get();
},
);
observer.observe_reads(
second_scope.clone(),
{
let second_triggered = Rc::clone(&second_triggered);
move |_| second_triggered.set(second_triggered.get() + 1)
},
|| {
let _ = state.get();
},
);
observer.clear(&first_scope);
let snapshot = take_mutable_snapshot(None, None);
snapshot.enter(|| {
state.set(1);
});
snapshot.apply().check();
assert_eq!(first_triggered.get(), 0);
assert_eq!(second_triggered.get(), 1);
observer.stop();
}
#[test]
fn shared_state_notifies_scopes_in_registration_order() {
let _guard = reset_runtime();
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let notifications = Rc::new(RefCell::new(Vec::new()));
let observer = SnapshotStateObserver::new(|callback| callback());
observer.start();
observer.observe_reads(
TestScope("first"),
{
let notifications = Rc::clone(¬ifications);
move |_| notifications.borrow_mut().push("first")
},
|| {
let _ = state.get();
},
);
observer.observe_reads(
TestScope("second"),
{
let notifications = Rc::clone(¬ifications);
move |_| notifications.borrow_mut().push("second")
},
|| {
let _ = state.get();
},
);
let snapshot = take_mutable_snapshot(None, None);
snapshot.enter(|| {
state.set(1);
});
snapshot.apply().check();
assert_eq!(notifications.borrow().as_slice(), &["first", "second"]);
observer.stop();
}
#[test]
fn stateless_recompose_scope_does_not_retain_observer_entry() {
let _guard = reset_runtime();
let observer = SnapshotStateObserver::new(|callback| callback());
let runtime = crate::TestRuntime::new();
let scope = RecomposeScope::new_for_test(runtime.handle());
observer.observe_reads(scope, |_| {}, || {});
let stats = observer.debug_stats();
assert_eq!(stats.scopes_len, 0);
assert_eq!(stats.fast_scopes_len, 0);
assert_eq!(stats.stateless_scope_count, 0);
}
#[test]
fn scope_that_stops_reading_state_is_removed_immediately() {
let _guard = reset_runtime();
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let observer = SnapshotStateObserver::new(|callback| callback());
let runtime = crate::TestRuntime::new();
let scope = RecomposeScope::new_for_test(runtime.handle());
let triggered = Rc::new(Cell::new(0));
let observer_trigger = Rc::clone(&triggered);
observer.observe_reads(
scope.clone(),
move |_| observer_trigger.set(observer_trigger.get() + 1),
|| {
let _ = state.get();
},
);
let after_stateful = observer.debug_stats();
assert_eq!(after_stateful.scopes_len, 1);
assert_eq!(after_stateful.fast_scopes_len, 1);
observer.observe_reads(scope, |_| {}, || {});
let after_stateless = observer.debug_stats();
assert_eq!(after_stateless.scopes_len, 0);
assert_eq!(after_stateless.fast_scopes_len, 0);
let snapshot = take_mutable_snapshot(None, None);
snapshot.enter(|| {
state.set(1);
});
snapshot.apply().check();
assert_eq!(triggered.get(), 0);
}
#[test]
fn begin_frame_prunes_dropped_recompose_scope_entries() {
let _guard = reset_runtime();
let state = SnapshotMutableState::new_in_arc(0, Arc::new(NeverEqual));
let observer = SnapshotStateObserver::new(|callback| callback());
let runtime = crate::TestRuntime::new();
let scope = RecomposeScope::new_for_test(runtime.handle());
observer.observe_reads(
scope.clone(),
|_| {},
|| {
let _ = state.get();
},
);
let before_prune = observer.debug_stats();
assert_eq!(before_prune.scopes_len, 1);
assert_eq!(before_prune.fast_scopes_len, 1);
drop(scope);
observer.begin_frame();
let after_prune = observer.debug_stats();
assert_eq!(after_prune.scopes_len, 0);
assert_eq!(after_prune.fast_scopes_len, 0);
}
}