1#![allow(clippy::arc_with_non_send_sync)]
22
23use std::{
24 cell::{Cell, RefCell},
25 hash::{Hash, Hasher},
26 rc::Rc,
27 sync::{Arc, Weak},
28};
29
30use crate::{
31 collections::map::{HashMap, HashSet},
32 snapshot_id_set::{SnapshotId, SnapshotIdSet},
33 snapshot_pinning::{self, PinHandle},
34 snapshot_weak_set::SnapshotWeakSetDebugStats,
35 state::{StateObject, StateRecord},
36};
37
38mod global;
39mod mutable;
40mod nested;
41mod readonly;
42mod runtime;
43mod transparent;
44
45#[cfg(test)]
46mod integration_tests;
47
48pub use global::{GlobalSnapshot, advance_global_snapshot};
49pub use mutable::MutableSnapshot;
50pub use nested::{NestedMutableSnapshot, NestedReadonlySnapshot};
51pub use readonly::ReadonlySnapshot;
52#[cfg(test)]
53pub(crate) use runtime::{TestRuntimeGuard, reset_runtime_for_tests};
54pub(crate) use runtime::{allocate_snapshot, close_snapshot, with_runtime};
55pub use transparent::{TransparentObserverMutableSnapshot, TransparentObserverSnapshot};
56
57pub type ReadObserver = Arc<dyn Fn(&dyn StateObject) + 'static>;
59
60pub type WriteObserver = Arc<dyn Fn(&dyn StateObject) + 'static>;
62
63pub type ApplyObserver = Rc<dyn Fn(&[Arc<dyn StateObject>], SnapshotId) + 'static>;
65
66#[derive(Debug, Clone, PartialEq, Eq)]
68pub enum SnapshotApplyResult {
69 Success,
71 Failure,
73}
74
75impl SnapshotApplyResult {
76 pub fn is_success(&self) -> bool {
78 matches!(self, SnapshotApplyResult::Success)
79 }
80
81 pub fn is_failure(&self) -> bool {
83 matches!(self, SnapshotApplyResult::Failure)
84 }
85
86 #[track_caller]
88 pub fn check(&self) {
89 if self.is_failure() {
90 panic!("Snapshot apply failed");
91 }
92 }
93}
94
95pub type StateObjectId = usize;
97
98#[derive(Clone)]
103pub enum AnySnapshot {
104 Readonly(Arc<ReadonlySnapshot>),
105 Mutable(Arc<MutableSnapshot>),
106 NestedReadonly(Arc<NestedReadonlySnapshot>),
107 NestedMutable(Arc<NestedMutableSnapshot>),
108 Global(Arc<GlobalSnapshot>),
109 TransparentMutable(Arc<TransparentObserverMutableSnapshot>),
110 TransparentReadonly(Arc<TransparentObserverSnapshot>),
111}
112
113#[derive(Clone)]
120pub enum AnyMutableSnapshot {
121 Root(Arc<MutableSnapshot>),
122 Nested(Arc<NestedMutableSnapshot>),
123}
124
125impl AnyMutableSnapshot {
126 pub fn snapshot_id(&self) -> SnapshotId {
128 match self {
129 AnyMutableSnapshot::Root(s) => s.snapshot_id(),
130 AnyMutableSnapshot::Nested(s) => s.snapshot_id(),
131 }
132 }
133
134 pub fn invalid(&self) -> SnapshotIdSet {
136 match self {
137 AnyMutableSnapshot::Root(s) => s.invalid(),
138 AnyMutableSnapshot::Nested(s) => s.invalid(),
139 }
140 }
141
142 pub fn enter<T>(&self, f: impl FnOnce() -> T) -> T {
144 match self {
145 AnyMutableSnapshot::Root(s) => s.enter(f),
146 AnyMutableSnapshot::Nested(s) => s.enter(f),
147 }
148 }
149
150 pub fn apply(&self) -> SnapshotApplyResult {
152 match self {
153 AnyMutableSnapshot::Root(s) => s.apply(),
154 AnyMutableSnapshot::Nested(s) => s.apply(),
155 }
156 }
157
158 pub fn dispose(&self) {
160 match self {
161 AnyMutableSnapshot::Root(s) => s.dispose(),
162 AnyMutableSnapshot::Nested(s) => s.dispose(),
163 }
164 }
165}
166
167impl AnySnapshot {
168 pub fn snapshot_id(&self) -> SnapshotId {
170 match self {
171 AnySnapshot::Readonly(s) => s.snapshot_id(),
172 AnySnapshot::Mutable(s) => s.snapshot_id(),
173 AnySnapshot::NestedReadonly(s) => s.snapshot_id(),
174 AnySnapshot::NestedMutable(s) => s.snapshot_id(),
175 AnySnapshot::Global(s) => s.snapshot_id(),
176 AnySnapshot::TransparentMutable(s) => s.snapshot_id(),
177 AnySnapshot::TransparentReadonly(s) => s.snapshot_id(),
178 }
179 }
180
181 pub fn invalid(&self) -> SnapshotIdSet {
183 match self {
184 AnySnapshot::Readonly(s) => s.invalid(),
185 AnySnapshot::Mutable(s) => s.invalid(),
186 AnySnapshot::NestedReadonly(s) => s.invalid(),
187 AnySnapshot::NestedMutable(s) => s.invalid(),
188 AnySnapshot::Global(s) => s.invalid(),
189 AnySnapshot::TransparentMutable(s) => s.invalid(),
190 AnySnapshot::TransparentReadonly(s) => s.invalid(),
191 }
192 }
193
194 pub fn is_valid(&self, id: SnapshotId) -> bool {
196 let snapshot_id = self.snapshot_id();
197 id <= snapshot_id && !self.invalid().get(id)
198 }
199
200 pub fn read_only(&self) -> bool {
202 match self {
203 AnySnapshot::Readonly(_) => true,
204 AnySnapshot::Mutable(_) => false,
205 AnySnapshot::NestedReadonly(_) => true,
206 AnySnapshot::NestedMutable(_) => false,
207 AnySnapshot::Global(_) => false,
208 AnySnapshot::TransparentMutable(_) => false,
209 AnySnapshot::TransparentReadonly(_) => true,
210 }
211 }
212
213 pub fn root(&self) -> AnySnapshot {
215 match self {
216 AnySnapshot::Readonly(s) => AnySnapshot::Readonly(s.root_readonly()),
217 AnySnapshot::Mutable(s) => AnySnapshot::Mutable(s.root_mutable()),
218 AnySnapshot::NestedReadonly(s) => AnySnapshot::NestedReadonly(s.root_nested_readonly()),
219 AnySnapshot::NestedMutable(s) => AnySnapshot::Mutable(s.root_mutable()),
220 AnySnapshot::Global(s) => AnySnapshot::Global(s.root_global()),
221 AnySnapshot::TransparentMutable(s) => {
222 AnySnapshot::TransparentMutable(s.root_transparent_mutable())
223 }
224 AnySnapshot::TransparentReadonly(s) => {
225 AnySnapshot::TransparentReadonly(s.root_transparent_readonly())
226 }
227 }
228 }
229
230 pub fn is_same_transparent(&self, other: &Arc<TransparentObserverMutableSnapshot>) -> bool {
232 matches!(self, AnySnapshot::TransparentMutable(snapshot) if Arc::ptr_eq(snapshot, other))
233 }
234
235 pub fn is_same_transparent_mutable(
237 &self,
238 other: &Arc<TransparentObserverMutableSnapshot>,
239 ) -> bool {
240 self.is_same_transparent(other)
241 }
242
243 pub fn is_same_transparent_readonly(&self, other: &Arc<TransparentObserverSnapshot>) -> bool {
245 matches!(self, AnySnapshot::TransparentReadonly(snapshot) if Arc::ptr_eq(snapshot, other))
246 }
247
248 pub fn enter<T>(&self, f: impl FnOnce() -> T) -> T {
250 match self {
251 AnySnapshot::Readonly(s) => s.enter(f),
252 AnySnapshot::Mutable(s) => s.enter(f),
253 AnySnapshot::NestedReadonly(s) => s.enter(f),
254 AnySnapshot::NestedMutable(s) => s.enter(f),
255 AnySnapshot::Global(s) => s.enter(f),
256 AnySnapshot::TransparentMutable(s) => s.enter(f),
257 AnySnapshot::TransparentReadonly(s) => s.enter(f),
258 }
259 }
260
261 pub fn take_nested_snapshot(&self, read_observer: Option<ReadObserver>) -> AnySnapshot {
263 match self {
264 AnySnapshot::Readonly(s) => {
265 AnySnapshot::Readonly(s.take_nested_snapshot(read_observer))
266 }
267 AnySnapshot::Mutable(s) => AnySnapshot::Readonly(s.take_nested_snapshot(read_observer)),
268 AnySnapshot::NestedReadonly(s) => {
269 AnySnapshot::NestedReadonly(s.take_nested_snapshot(read_observer))
270 }
271 AnySnapshot::NestedMutable(s) => {
272 AnySnapshot::Readonly(s.take_nested_snapshot(read_observer))
273 }
274 AnySnapshot::Global(s) => AnySnapshot::Readonly(s.take_nested_snapshot(read_observer)),
275 AnySnapshot::TransparentMutable(s) => {
276 AnySnapshot::Readonly(s.take_nested_snapshot(read_observer))
277 }
278 AnySnapshot::TransparentReadonly(s) => {
279 AnySnapshot::TransparentReadonly(s.take_nested_snapshot(read_observer))
280 }
281 }
282 }
283
284 pub fn has_pending_changes(&self) -> bool {
286 match self {
287 AnySnapshot::Readonly(s) => s.has_pending_changes(),
288 AnySnapshot::Mutable(s) => s.has_pending_changes(),
289 AnySnapshot::NestedReadonly(s) => s.has_pending_changes(),
290 AnySnapshot::NestedMutable(s) => s.has_pending_changes(),
291 AnySnapshot::Global(s) => s.has_pending_changes(),
292 AnySnapshot::TransparentMutable(s) => s.has_pending_changes(),
293 AnySnapshot::TransparentReadonly(s) => s.has_pending_changes(),
294 }
295 }
296
297 pub fn dispose(&self) {
299 match self {
300 AnySnapshot::Readonly(s) => s.dispose(),
301 AnySnapshot::Mutable(s) => s.dispose(),
302 AnySnapshot::NestedReadonly(s) => s.dispose(),
303 AnySnapshot::NestedMutable(s) => s.dispose(),
304 AnySnapshot::Global(s) => s.dispose(),
305 AnySnapshot::TransparentMutable(s) => s.dispose(),
306 AnySnapshot::TransparentReadonly(s) => s.dispose(),
307 }
308 }
309
310 pub fn is_disposed(&self) -> bool {
312 match self {
313 AnySnapshot::Readonly(s) => s.is_disposed(),
314 AnySnapshot::Mutable(s) => s.is_disposed(),
315 AnySnapshot::NestedReadonly(s) => s.is_disposed(),
316 AnySnapshot::NestedMutable(s) => s.is_disposed(),
317 AnySnapshot::Global(s) => s.is_disposed(),
318 AnySnapshot::TransparentMutable(s) => s.is_disposed(),
319 AnySnapshot::TransparentReadonly(s) => s.is_disposed(),
320 }
321 }
322
323 pub fn record_read(&self, state: &dyn StateObject) {
325 match self {
326 AnySnapshot::Readonly(s) => s.record_read(state),
327 AnySnapshot::Mutable(s) => s.record_read(state),
328 AnySnapshot::NestedReadonly(s) => s.record_read(state),
329 AnySnapshot::NestedMutable(s) => s.record_read(state),
330 AnySnapshot::Global(s) => s.record_read(state),
331 AnySnapshot::TransparentMutable(s) => s.record_read(state),
332 AnySnapshot::TransparentReadonly(s) => s.record_read(state),
333 }
334 }
335
336 pub fn record_write(&self, state: Arc<dyn StateObject>) {
338 match self {
339 AnySnapshot::Readonly(s) => s.record_write(state),
340 AnySnapshot::Mutable(s) => s.record_write(state),
341 AnySnapshot::NestedReadonly(s) => s.record_write(state),
342 AnySnapshot::NestedMutable(s) => s.record_write(state),
343 AnySnapshot::Global(s) => s.record_write(state),
344 AnySnapshot::TransparentMutable(s) => s.record_write(state),
345 AnySnapshot::TransparentReadonly(s) => s.record_write(state),
346 }
347 }
348
349 pub fn apply(&self) -> SnapshotApplyResult {
351 match self {
352 AnySnapshot::Mutable(s) => s.apply(),
353 AnySnapshot::NestedMutable(s) => s.apply(),
354 AnySnapshot::Global(s) => s.apply(),
355 AnySnapshot::TransparentMutable(s) => s.apply(),
356 _ => panic!("Cannot apply a read-only snapshot"),
357 }
358 }
359
360 pub fn take_nested_mutable_snapshot(
362 &self,
363 read_observer: Option<ReadObserver>,
364 write_observer: Option<WriteObserver>,
365 ) -> AnySnapshot {
366 match self {
367 AnySnapshot::Mutable(s) => AnySnapshot::NestedMutable(
368 s.take_nested_mutable_snapshot(read_observer, write_observer),
369 ),
370 AnySnapshot::NestedMutable(s) => AnySnapshot::NestedMutable(
371 s.take_nested_mutable_snapshot(read_observer, write_observer),
372 ),
373 AnySnapshot::Global(s) => {
374 AnySnapshot::Mutable(s.take_nested_mutable_snapshot(read_observer, write_observer))
375 }
376 AnySnapshot::TransparentMutable(s) => AnySnapshot::TransparentMutable(
377 s.take_nested_mutable_snapshot(read_observer, write_observer),
378 ),
379 _ => panic!("Cannot take nested mutable snapshot from read-only snapshot"),
380 }
381 }
382}
383
384thread_local! {
385 static CURRENT_SNAPSHOT: RefCell<Option<AnySnapshot>> = const { RefCell::new(None) };
386}
387
388pub fn current_snapshot() -> Option<AnySnapshot> {
390 CURRENT_SNAPSHOT
391 .try_with(|cell| cell.borrow().clone())
392 .unwrap_or(None)
393}
394
395pub(crate) fn set_current_snapshot(snapshot: Option<AnySnapshot>) {
396 let _ = CURRENT_SNAPSHOT.try_with(|cell| {
397 *cell.borrow_mut() = snapshot;
398 });
399}
400
401struct CurrentSnapshotGuard {
402 previous: Option<AnySnapshot>,
403}
404
405impl CurrentSnapshotGuard {
406 fn enter(snapshot: AnySnapshot) -> Self {
407 let previous = current_snapshot();
408 set_current_snapshot(Some(snapshot));
409 Self { previous }
410 }
411}
412
413impl Drop for CurrentSnapshotGuard {
414 fn drop(&mut self) {
415 set_current_snapshot(self.previous.take());
416 }
417}
418
419pub(crate) fn enter_snapshot_scope<T>(snapshot: AnySnapshot, f: impl FnOnce() -> T) -> T {
420 let _guard = CurrentSnapshotGuard::enter(snapshot);
421 f()
422}
423
424pub fn take_mutable_snapshot(
433 read_observer: Option<ReadObserver>,
434 write_observer: Option<WriteObserver>,
435) -> AnyMutableSnapshot {
436 match current_snapshot() {
437 Some(AnySnapshot::Mutable(parent)) => AnyMutableSnapshot::Nested(
438 parent.take_nested_mutable_snapshot(read_observer, write_observer),
439 ),
440 Some(AnySnapshot::NestedMutable(parent)) => AnyMutableSnapshot::Nested(
441 parent.take_nested_mutable_snapshot(read_observer, write_observer),
442 ),
443 _ => AnyMutableSnapshot::Root(
444 GlobalSnapshot::get_or_create()
445 .take_nested_mutable_snapshot(read_observer, write_observer),
446 ),
447 }
448}
449
450pub fn take_transparent_observer_mutable_snapshot(
459 read_observer: Option<ReadObserver>,
460 write_observer: Option<WriteObserver>,
461) -> Arc<TransparentObserverMutableSnapshot> {
462 take_transparent_observer_mutable_snapshot_reusing(read_observer, write_observer, None)
463}
464
465pub(crate) fn take_transparent_observer_mutable_snapshot_reusing(
466 read_observer: Option<ReadObserver>,
467 write_observer: Option<WriteObserver>,
468 recycled: Option<Arc<TransparentObserverMutableSnapshot>>,
469) -> Arc<TransparentObserverMutableSnapshot> {
470 let parent = current_snapshot();
471 match parent {
472 Some(AnySnapshot::TransparentMutable(transparent)) if transparent.can_reuse() => {
473 transparent
474 }
475 _ => {
476 let current = current_snapshot()
477 .unwrap_or_else(|| AnySnapshot::Global(GlobalSnapshot::get_or_create()));
478 let id = current.snapshot_id();
479 let invalid = current.invalid();
480 TransparentObserverMutableSnapshot::new_reusing(
481 recycled,
482 id,
483 invalid,
484 read_observer,
485 write_observer,
486 None,
487 )
488 }
489 }
490}
491
492pub fn allocate_record_id() -> SnapshotId {
494 runtime::allocate_record_id()
495}
496
497pub(crate) fn peek_next_snapshot_id() -> SnapshotId {
498 runtime::peek_next_snapshot_id()
499}
500
501#[derive(Clone)]
502struct ObserverId(Rc<()>);
503
504impl ObserverId {
505 fn new() -> Self {
506 Self(Rc::new(()))
507 }
508}
509
510impl PartialEq for ObserverId {
511 fn eq(&self, other: &Self) -> bool {
512 Rc::ptr_eq(&self.0, &other.0)
513 }
514}
515
516impl Eq for ObserverId {}
517
518impl Hash for ObserverId {
519 fn hash<H: Hasher>(&self, state: &mut H) {
520 Rc::as_ptr(&self.0).hash(state);
521 }
522}
523
524thread_local! {
525 static APPLY_OBSERVERS: RefCell<HashMap<ObserverId, ApplyObserver>> = RefCell::new(HashMap::default());
526}
527
528thread_local! {
529 static LAST_WRITES: RefCell<HashMap<StateObjectId, SnapshotId>> = RefCell::new(HashMap::default());
530}
531
532thread_local! {
533 static EXTRA_STATE_OBJECTS: RefCell<crate::snapshot_weak_set::SnapshotWeakSet> = RefCell::new(crate::snapshot_weak_set::SnapshotWeakSet::new());
534}
535
536const UNUSED_RECORD_CLEANUP_INTERVAL: SnapshotId = 2;
537const UNUSED_RECORD_CLEANUP_BUSY_INTERVAL: SnapshotId = 1;
538const UNUSED_RECORD_CLEANUP_MIN_SIZE: usize = 64;
539
540thread_local! {
541 static LAST_UNUSED_RECORD_CLEANUP: Cell<SnapshotId> = const { Cell::new(0) };
542}
543
544#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
545pub struct SnapshotV2DebugStats {
546 pub apply_observers_len: usize,
547 pub apply_observers_cap: usize,
548 pub last_writes_len: usize,
549 pub last_writes_cap: usize,
550 pub extra_state_objects_len: usize,
551 pub extra_state_objects_cap: usize,
552 pub last_unused_record_cleanup: SnapshotId,
553}
554
555pub fn debug_snapshot_v2_stats() -> SnapshotV2DebugStats {
556 let (apply_observers_len, apply_observers_cap) = APPLY_OBSERVERS.with(|cell| {
557 let observers = cell.borrow();
558 (observers.len(), observers.capacity())
559 });
560 let (last_writes_len, last_writes_cap) = LAST_WRITES.with(|cell| {
561 let writes = cell.borrow();
562 (writes.len(), writes.capacity())
563 });
564 let SnapshotWeakSetDebugStats {
565 len: extra_state_objects_len,
566 capacity: extra_state_objects_cap,
567 } = EXTRA_STATE_OBJECTS.with(|cell| cell.borrow().debug_stats());
568 let last_unused_record_cleanup = LAST_UNUSED_RECORD_CLEANUP.with(|cell| cell.get());
569
570 SnapshotV2DebugStats {
571 apply_observers_len,
572 apply_observers_cap,
573 last_writes_len,
574 last_writes_cap,
575 extra_state_objects_len,
576 extra_state_objects_cap,
577 last_unused_record_cleanup,
578 }
579}
580
581pub fn register_apply_observer(observer: ApplyObserver) -> ObserverHandle {
585 let id = ObserverId::new();
586 APPLY_OBSERVERS.with(|cell| {
587 cell.borrow_mut().insert(id.clone(), observer);
588 });
589 ObserverHandle {
590 kind: ObserverKind::Apply,
591 id,
592 }
593}
594
595pub struct ObserverHandle {
599 kind: ObserverKind,
600 id: ObserverId,
601}
602
603enum ObserverKind {
604 Apply,
605}
606
607impl Drop for ObserverHandle {
608 fn drop(&mut self) {
609 match self.kind {
610 ObserverKind::Apply => {
611 APPLY_OBSERVERS.with(|cell| {
612 cell.borrow_mut().remove(&self.id);
613 });
614 }
615 }
616 }
617}
618
619pub(crate) fn notify_apply_observers(modified: &[Arc<dyn StateObject>], snapshot_id: SnapshotId) {
620 APPLY_OBSERVERS.with(|cell| {
621 let observers: Vec<ApplyObserver> = cell.borrow().values().cloned().collect();
622 for observer in observers.into_iter() {
623 observer(modified, snapshot_id);
624 }
625 });
626}
627
628pub(crate) fn set_last_write(id: StateObjectId, snapshot_id: SnapshotId) {
629 LAST_WRITES.with(|cell| {
630 cell.borrow_mut().insert(id, snapshot_id);
631 });
632}
633
634#[cfg(test)]
635pub(crate) fn clear_last_writes() {
636 LAST_WRITES.with(|cell| {
637 cell.borrow_mut().clear();
638 });
639}
640
641pub(crate) fn check_and_overwrite_unused_records_locked() {
642 EXTRA_STATE_OBJECTS.with(|cell| {
643 cell.borrow_mut()
644 .remove_if(|state| state.overwrite_unused_records());
645 });
646}
647
648pub(crate) fn maybe_check_and_overwrite_unused_records_locked(current_snapshot_id: SnapshotId) {
649 let should_run = EXTRA_STATE_OBJECTS.with(|cell| {
650 let set = cell.borrow();
651 if set.is_empty() {
652 return false;
653 }
654 let last_cleanup = LAST_UNUSED_RECORD_CLEANUP.with(|last| last.get());
655 let interval = if set.len() >= UNUSED_RECORD_CLEANUP_MIN_SIZE {
656 UNUSED_RECORD_CLEANUP_BUSY_INTERVAL
657 } else {
658 UNUSED_RECORD_CLEANUP_INTERVAL
659 };
660 current_snapshot_id.saturating_sub(last_cleanup) >= interval
661 });
662
663 if should_run {
664 LAST_UNUSED_RECORD_CLEANUP.with(|cell| cell.set(current_snapshot_id));
665 check_and_overwrite_unused_records_locked();
666 }
667}
668
669#[cfg(test)]
670pub(crate) fn clear_unused_record_cleanup_for_tests() {
671 LAST_UNUSED_RECORD_CLEANUP.with(|cell| cell.set(0));
672}
673
674pub(crate) fn optimistic_merges(
675 current_snapshot_id: SnapshotId,
676 base_parent_id: SnapshotId,
677 modified_objects: &[(StateObjectId, Arc<dyn StateObject>, SnapshotId)],
678 invalid_snapshots: &SnapshotIdSet,
679 applying_invalid: &SnapshotIdSet,
680) -> Option<HashMap<usize, Rc<StateRecord>>> {
681 if modified_objects.is_empty() {
682 return None;
683 }
684
685 let mut result: Option<HashMap<usize, Rc<StateRecord>>> = None;
686
687 for (_, state, writer_id) in modified_objects.iter() {
688 let head = state.first_record();
689
690 let current = match crate::state::readable_record_for(
691 &head,
692 current_snapshot_id,
693 invalid_snapshots,
694 ) {
695 Some(record) => record,
696 None => continue,
697 };
698
699 let (previous_opt, found_base) =
700 mutable::find_previous_record(&head, base_parent_id, applying_invalid);
701 let previous = previous_opt?;
702
703 if !found_base || previous.snapshot_id() == crate::state::PREEXISTING_SNAPSHOT_ID {
704 continue;
705 }
706
707 if Rc::ptr_eq(¤t, &previous) {
708 continue;
709 }
710
711 let applied = mutable::find_record_by_id(&head, *writer_id)?;
712
713 let merged = state.merge_records(
714 Rc::clone(&previous),
715 Rc::clone(¤t),
716 Rc::clone(&applied),
717 )?;
718
719 result
720 .get_or_insert_with(HashMap::default)
721 .insert(Rc::as_ptr(¤t) as usize, merged);
722 }
723
724 result
725}
726
727#[allow(clippy::arc_with_non_send_sync)]
728fn merge_observers(a: Option<ReadObserver>, b: Option<ReadObserver>) -> Option<ReadObserver> {
729 match (a, b) {
730 (None, None) => None,
731 (Some(a), None) => Some(a),
732 (None, Some(b)) => Some(b),
733 (Some(a), Some(b)) => Some(Arc::new(move |state: &dyn StateObject| {
734 a(state);
735 b(state);
736 })),
737 }
738}
739
740pub fn merge_read_observers(
746 a: Option<ReadObserver>,
747 b: Option<ReadObserver>,
748) -> Option<ReadObserver> {
749 merge_observers(a, b)
750}
751
752pub fn merge_write_observers(
758 a: Option<WriteObserver>,
759 b: Option<WriteObserver>,
760) -> Option<WriteObserver> {
761 merge_observers(a, b)
762}
763
764pub(crate) struct SnapshotState {
765 pub(crate) id: Cell<SnapshotId>,
766 pub(crate) invalid: RefCell<SnapshotIdSet>,
767 pub(crate) pin_handle: Cell<PinHandle>,
768 pub(crate) disposed: Cell<bool>,
769 pub(crate) read_observer: RefCell<Option<ReadObserver>>,
770 pub(crate) write_observer: RefCell<Option<WriteObserver>>,
771 #[allow(clippy::type_complexity)]
772 pub(crate) modified: RefCell<HashMap<StateObjectId, (Arc<dyn StateObject>, SnapshotId)>>,
773 on_dispose: RefCell<Option<Box<dyn FnOnce()>>>,
774 runtime_tracked: bool,
775 pending_children: RefCell<HashSet<SnapshotId>>,
776}
777
778impl SnapshotState {
779 pub(crate) fn new(
780 id: SnapshotId,
781 invalid: SnapshotIdSet,
782 read_observer: Option<ReadObserver>,
783 write_observer: Option<WriteObserver>,
784 runtime_tracked: bool,
785 ) -> Self {
786 Self::new_with_pinning(
787 id,
788 invalid,
789 read_observer,
790 write_observer,
791 runtime_tracked,
792 true,
793 )
794 }
795
796 pub(crate) fn new_with_pinning(
797 id: SnapshotId,
798 invalid: SnapshotIdSet,
799 read_observer: Option<ReadObserver>,
800 write_observer: Option<WriteObserver>,
801 runtime_tracked: bool,
802 should_pin: bool,
803 ) -> Self {
804 let pin_handle = if should_pin {
805 snapshot_pinning::track_pinning(id, &invalid)
806 } else {
807 snapshot_pinning::PinHandle::INVALID
808 };
809 Self {
810 id: Cell::new(id),
811 invalid: RefCell::new(invalid),
812 pin_handle: Cell::new(pin_handle),
813 disposed: Cell::new(false),
814 read_observer: RefCell::new(read_observer),
815 write_observer: RefCell::new(write_observer),
816 modified: RefCell::new(HashMap::default()),
817 on_dispose: RefCell::new(None),
818 runtime_tracked,
819 pending_children: RefCell::new(HashSet::default()),
820 }
821 }
822
823 pub(crate) fn record_read(&self, state: &dyn StateObject) {
824 if let Some(observer) = self.read_observer.borrow().as_ref() {
825 observer(state);
826 }
827 }
828
829 pub(crate) fn record_write(&self, state: Arc<dyn StateObject>, writer_id: SnapshotId) {
830 let state_id = state.object_id().as_usize();
831
832 let mut modified = self.modified.borrow_mut();
833
834 match modified.entry(state_id) {
835 std::collections::hash_map::Entry::Vacant(e) => {
836 if let Some(observer) = self.write_observer.borrow().as_ref() {
837 observer(&*state);
838 }
839 e.insert((state, writer_id));
840 }
841 std::collections::hash_map::Entry::Occupied(mut e) => {
842 e.insert((state, writer_id));
843 }
844 }
845 }
846
847 pub(crate) fn dispose(&self) {
848 if !self.disposed.replace(true) {
849 let pin_handle = self.pin_handle.get();
850 snapshot_pinning::release_pinning(pin_handle);
851 if let Some(cb) = self.on_dispose.borrow_mut().take() {
852 cb();
853 }
854 if self.runtime_tracked {
855 close_snapshot(self.id.get());
856 }
857 }
858 }
859
860 pub(crate) fn add_pending_child(&self, id: SnapshotId) {
861 self.pending_children.borrow_mut().insert(id);
862 }
863
864 pub(crate) fn remove_pending_child(&self, id: SnapshotId) {
865 self.pending_children.borrow_mut().remove(&id);
866 }
867
868 pub(crate) fn has_pending_children(&self) -> bool {
869 !self.pending_children.borrow().is_empty()
870 }
871
872 pub(crate) fn pending_children(&self) -> Vec<SnapshotId> {
873 self.pending_children.borrow().iter().copied().collect()
874 }
875
876 pub(crate) fn set_on_dispose<F>(&self, f: F)
877 where
878 F: FnOnce() + 'static,
879 {
880 *self.on_dispose.borrow_mut() = Some(Box::new(f));
881 }
882}
883
884pub(crate) trait NestedMutableHost {
885 fn snapshot_state(&self) -> &SnapshotState;
886 fn nested_count(&self) -> &Cell<usize>;
887}
888
889pub(crate) fn clear_nested_child_on_dispose<P>(
890 parent: &Arc<P>,
891 child_id: SnapshotId,
892) -> impl FnOnce() + 'static
893where
894 P: NestedMutableHost + 'static,
895{
896 let weak = Arc::downgrade(parent);
897 move || {
898 if let Some(parent) = weak.upgrade() {
899 let nested_count = parent.nested_count();
900 if nested_count.get() > 0 {
901 nested_count.set(nested_count.get().saturating_sub(1));
902 }
903 let state = parent.snapshot_state();
904 let new_invalid = state.invalid.borrow().clone().clear(child_id);
905 state.invalid.replace(new_invalid);
906 state.remove_pending_child(child_id);
907 }
908 }
909}
910
911pub(crate) fn allocate_nested_mutable_snapshot<P>(
912 parent: &Arc<P>,
913 root: Weak<MutableSnapshot>,
914 read_observer: Option<ReadObserver>,
915 write_observer: Option<WriteObserver>,
916) -> Arc<NestedMutableSnapshot>
917where
918 P: NestedMutableHost + 'static,
919{
920 let state = parent.snapshot_state();
921 let merged_read = merge_read_observers(read_observer, state.read_observer.borrow().clone());
922 let merged_write = merge_write_observers(write_observer, state.write_observer.borrow().clone());
923
924 let parent_id = state.id.get();
925 let current_invalid = state.invalid.borrow().clone();
926
927 let (new_id, _runtime_invalid) = allocate_snapshot();
928
929 let parent_invalid_with_child = current_invalid.set(new_id);
930 state.invalid.replace(parent_invalid_with_child);
931
932 let invalid = current_invalid.add_range(parent_id + 1, new_id);
933
934 let nested = NestedMutableSnapshot::new(
935 new_id,
936 invalid,
937 merged_read,
938 merged_write,
939 root,
940 state.id.get(),
941 );
942
943 let nested_count = parent.nested_count();
944 nested_count.set(nested_count.get() + 1);
945 state.add_pending_child(new_id);
946
947 nested.set_on_dispose(clear_nested_child_on_dispose(parent, new_id));
948
949 nested
950}
951
952#[cfg(test)]
953mod tests {
954 use std::sync::Mutex;
955
956 use super::*;
957
958 #[test]
959 fn apply_observer_ids_do_not_use_process_global_counter() {
960 let source = include_str!("mod.rs");
961 assert!(!source.contains(concat!("NEXT_", "OBSERVER_ID")));
962 assert!(!source.contains(concat!("Atomic", "Usize")));
963 }
964
965 #[test]
966 fn test_apply_result_is_success() {
967 assert!(SnapshotApplyResult::Success.is_success());
968 assert!(!SnapshotApplyResult::Failure.is_success());
969 }
970
971 #[test]
972 fn test_apply_result_is_failure() {
973 assert!(!SnapshotApplyResult::Success.is_failure());
974 assert!(SnapshotApplyResult::Failure.is_failure());
975 }
976
977 #[test]
978 fn test_apply_result_check_success() {
979 SnapshotApplyResult::Success.check();
980 }
981
982 #[test]
983 #[should_panic(expected = "Snapshot apply failed")]
984 fn test_apply_result_check_failure() {
985 SnapshotApplyResult::Failure.check();
986 }
987
988 #[test]
989 fn snapshot_enter_restores_current_snapshot_after_panic() {
990 let _guard = reset_runtime_for_tests();
991 let snapshot = take_mutable_snapshot(None, None);
992
993 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
994 snapshot.enter(|| panic!("snapshot body panic"));
995 }));
996
997 assert!(result.is_err());
998 assert!(
999 current_snapshot().is_none(),
1000 "snapshot enter must restore the previous current snapshot while unwinding"
1001 );
1002 }
1003
1004 #[test]
1005 fn test_merge_read_observers_both_none() {
1006 let result = merge_read_observers(None, None);
1007 assert!(result.is_none());
1008 }
1009
1010 #[test]
1011 fn test_merge_read_observers_one_some() {
1012 let observer = Arc::new(|_: &dyn StateObject| {});
1013 let result = merge_read_observers(Some(observer.clone()), None);
1014 assert!(result.is_some());
1015
1016 let result = merge_read_observers(None, Some(observer));
1017 assert!(result.is_some());
1018 }
1019
1020 #[test]
1021 fn test_merge_write_observers_both_none() {
1022 let result = merge_write_observers(None, None);
1023 assert!(result.is_none());
1024 }
1025
1026 #[test]
1027 fn test_merge_write_observers_one_some() {
1028 let observer = Arc::new(|_: &dyn StateObject| {});
1029 let result = merge_write_observers(Some(observer.clone()), None);
1030 assert!(result.is_some());
1031
1032 let result = merge_write_observers(None, Some(observer));
1033 assert!(result.is_some());
1034 }
1035
1036 #[test]
1037 fn test_current_snapshot_none_initially() {
1038 set_current_snapshot(None);
1039 assert!(current_snapshot().is_none());
1040 }
1041
1042 struct TestStateObject {
1043 id: usize,
1044 }
1045
1046 impl TestStateObject {
1047 fn new(id: usize) -> Arc<Self> {
1048 Arc::new(Self { id })
1049 }
1050 }
1051
1052 impl StateObject for TestStateObject {
1053 fn object_id(&self) -> crate::state::ObjectId {
1054 crate::state::ObjectId(self.id)
1055 }
1056
1057 fn first_record(&self) -> Rc<crate::state::StateRecord> {
1058 unimplemented!("Not needed for observer tests")
1059 }
1060
1061 fn try_readable_record(
1062 &self,
1063 _snapshot_id: SnapshotId,
1064 _invalid: &SnapshotIdSet,
1065 ) -> Option<Rc<crate::state::StateRecord>> {
1066 None
1067 }
1068
1069 fn readable_record(
1070 &self,
1071 _snapshot_id: SnapshotId,
1072 _invalid: &SnapshotIdSet,
1073 ) -> Rc<crate::state::StateRecord> {
1074 unimplemented!("Not needed for observer tests")
1075 }
1076
1077 fn prepend_state_record(&self, _record: Rc<crate::state::StateRecord>) {
1078 unimplemented!("Not needed for observer tests")
1079 }
1080
1081 fn promote_record(&self, _child_id: SnapshotId) -> Result<(), &'static str> {
1082 unimplemented!("Not needed for observer tests")
1083 }
1084
1085 fn as_any(&self) -> &dyn std::any::Any {
1086 self
1087 }
1088 }
1089
1090 #[test]
1091 fn test_apply_observer_receives_correct_modified_objects() {
1092 use std::sync::Mutex;
1093
1094 let received_count = Arc::new(Mutex::new(0));
1095 let received_snapshot_id = Arc::new(Mutex::new(0));
1096
1097 let received_count_clone = received_count.clone();
1098 let received_snapshot_id_clone = received_snapshot_id.clone();
1099
1100 let _handle = register_apply_observer(Rc::new(move |modified, snapshot_id| {
1101 *received_snapshot_id_clone.lock().unwrap() = snapshot_id;
1102 *received_count_clone.lock().unwrap() = modified.len();
1103 }));
1104
1105 let obj1: Arc<dyn StateObject> = TestStateObject::new(42);
1106 let obj2: Arc<dyn StateObject> = TestStateObject::new(99);
1107 let modified = vec![obj1, obj2];
1108
1109 notify_apply_observers(&modified, 123);
1110
1111 assert_eq!(*received_snapshot_id.lock().unwrap(), 123);
1112 assert_eq!(*received_count.lock().unwrap(), 2);
1113 }
1114
1115 #[test]
1116 fn test_apply_observer_receives_correct_snapshot_id() {
1117 use std::sync::Mutex;
1118
1119 let received_id = Arc::new(Mutex::new(0));
1120 let received_id_clone = received_id.clone();
1121
1122 let _handle = register_apply_observer(Rc::new(move |_, snapshot_id| {
1123 *received_id_clone.lock().unwrap() = snapshot_id;
1124 }));
1125
1126 notify_apply_observers(&[], 456);
1127
1128 assert_eq!(*received_id.lock().unwrap(), 456);
1129 }
1130
1131 #[test]
1132 fn test_multiple_apply_observers_all_called() {
1133 use std::sync::Mutex;
1134
1135 let call_count1 = Arc::new(Mutex::new(0));
1136 let call_count2 = Arc::new(Mutex::new(0));
1137 let call_count3 = Arc::new(Mutex::new(0));
1138
1139 let call_count1_clone = call_count1.clone();
1140 let call_count2_clone = call_count2.clone();
1141 let call_count3_clone = call_count3.clone();
1142
1143 let _handle1 = register_apply_observer(Rc::new(move |_, _| {
1144 *call_count1_clone.lock().unwrap() += 1;
1145 }));
1146
1147 let _handle2 = register_apply_observer(Rc::new(move |_, _| {
1148 *call_count2_clone.lock().unwrap() += 1;
1149 }));
1150
1151 let _handle3 = register_apply_observer(Rc::new(move |_, _| {
1152 *call_count3_clone.lock().unwrap() += 1;
1153 }));
1154
1155 notify_apply_observers(&[], 1);
1156
1157 assert_eq!(*call_count1.lock().unwrap(), 1);
1158 assert_eq!(*call_count2.lock().unwrap(), 1);
1159 assert_eq!(*call_count3.lock().unwrap(), 1);
1160
1161 notify_apply_observers(&[], 2);
1162
1163 assert_eq!(*call_count1.lock().unwrap(), 2);
1164 assert_eq!(*call_count2.lock().unwrap(), 2);
1165 assert_eq!(*call_count3.lock().unwrap(), 2);
1166 }
1167
1168 #[test]
1169 fn test_apply_observer_not_called_for_empty_modifications() {
1170 use std::sync::Mutex;
1171
1172 let call_count = Arc::new(Mutex::new(0));
1173 let call_count_clone = call_count.clone();
1174
1175 let _handle = register_apply_observer(Rc::new(move |modified, _| {
1176 *call_count_clone.lock().unwrap() += 1;
1177 assert_eq!(modified.len(), 0);
1178 }));
1179
1180 notify_apply_observers(&[], 1);
1181
1182 assert_eq!(*call_count.lock().unwrap(), 1);
1183 }
1184
1185 fn register_counting_observer(calls: &Arc<Mutex<Vec<i32>>>, tag: i32) -> ObserverHandle {
1186 let calls = calls.clone();
1187 register_apply_observer(Rc::new(move |_, _| {
1188 calls.lock().unwrap().push(tag);
1189 }))
1190 }
1191
1192 #[test]
1193 fn test_observer_handle_drop_removes_correct_observer() {
1194 let calls = Arc::new(Mutex::new(Vec::new()));
1195
1196 let handle1 = register_counting_observer(&calls, 1);
1197 let handle2 = register_counting_observer(&calls, 2);
1198 let handle3 = register_counting_observer(&calls, 3);
1199
1200 notify_apply_observers(&[], 1);
1201 let result = calls.lock().unwrap().clone();
1202 assert_eq!(result.len(), 3);
1203 assert!(result.contains(&1));
1204 assert!(result.contains(&2));
1205 assert!(result.contains(&3));
1206 calls.lock().unwrap().clear();
1207
1208 drop(handle2);
1209
1210 notify_apply_observers(&[], 2);
1211 let result = calls.lock().unwrap().clone();
1212 assert_eq!(result.len(), 2);
1213 assert!(result.contains(&1));
1214 assert!(result.contains(&3));
1215 assert!(!result.contains(&2));
1216 calls.lock().unwrap().clear();
1217
1218 drop(handle1);
1219
1220 notify_apply_observers(&[], 3);
1221 let result = calls.lock().unwrap().clone();
1222 assert_eq!(result.len(), 1);
1223 assert!(result.contains(&3));
1224 calls.lock().unwrap().clear();
1225
1226 drop(handle3);
1227
1228 notify_apply_observers(&[], 4);
1229 assert_eq!(calls.lock().unwrap().len(), 0);
1230 }
1231
1232 #[test]
1233 fn test_observer_handle_drop_in_different_orders() {
1234 {
1235 let calls = Arc::new(Mutex::new(Vec::new()));
1236
1237 let h1 = register_counting_observer(&calls, 1);
1238 let h2 = register_counting_observer(&calls, 2);
1239 let h3 = register_counting_observer(&calls, 3);
1240
1241 drop(h3);
1242 notify_apply_observers(&[], 1);
1243 let result = calls.lock().unwrap().clone();
1244 assert!(result.contains(&1) && result.contains(&2) && !result.contains(&3));
1245 calls.lock().unwrap().clear();
1246
1247 drop(h2);
1248 notify_apply_observers(&[], 2);
1249 let result = calls.lock().unwrap().clone();
1250 assert_eq!(result.len(), 1);
1251 assert!(result.contains(&1));
1252 calls.lock().unwrap().clear();
1253
1254 drop(h1);
1255 notify_apply_observers(&[], 3);
1256 assert_eq!(calls.lock().unwrap().len(), 0);
1257 }
1258
1259 {
1260 let calls = Arc::new(Mutex::new(Vec::new()));
1261
1262 let h1 = register_counting_observer(&calls, 1);
1263 let h2 = register_counting_observer(&calls, 2);
1264 let h3 = register_counting_observer(&calls, 3);
1265
1266 drop(h1);
1267 notify_apply_observers(&[], 1);
1268 let result = calls.lock().unwrap().clone();
1269 assert!(!result.contains(&1) && result.contains(&2) && result.contains(&3));
1270 calls.lock().unwrap().clear();
1271
1272 drop(h2);
1273 notify_apply_observers(&[], 2);
1274 let result = calls.lock().unwrap().clone();
1275 assert_eq!(result.len(), 1);
1276 assert!(result.contains(&3));
1277 calls.lock().unwrap().clear();
1278
1279 drop(h3);
1280 notify_apply_observers(&[], 3);
1281 assert_eq!(calls.lock().unwrap().len(), 0);
1282 }
1283 }
1284
1285 #[test]
1286 fn test_remaining_observers_still_work_after_drop() {
1287 use std::sync::Mutex;
1288
1289 let calls = Arc::new(Mutex::new(Vec::new()));
1290
1291 let calls1 = calls.clone();
1292 let handle1 = register_apply_observer(Rc::new(move |_, snapshot_id| {
1293 calls1.lock().unwrap().push((1, snapshot_id));
1294 }));
1295
1296 let calls2 = calls.clone();
1297 let handle2 = register_apply_observer(Rc::new(move |_, snapshot_id| {
1298 calls2.lock().unwrap().push((2, snapshot_id));
1299 }));
1300
1301 notify_apply_observers(&[], 100);
1302 assert_eq!(calls.lock().unwrap().len(), 2);
1303 calls.lock().unwrap().clear();
1304
1305 drop(handle1);
1306
1307 notify_apply_observers(&[], 200);
1308 assert_eq!(*calls.lock().unwrap(), vec![(2, 200)]);
1309 calls.lock().unwrap().clear();
1310
1311 let calls3 = calls.clone();
1312 let _handle3 = register_apply_observer(Rc::new(move |_, snapshot_id| {
1313 calls3.lock().unwrap().push((3, snapshot_id));
1314 }));
1315
1316 notify_apply_observers(&[], 300);
1317 let result = calls.lock().unwrap().clone();
1318 assert_eq!(result.len(), 2);
1319 assert!(result.contains(&(2, 300)));
1320 assert!(result.contains(&(3, 300)));
1321
1322 drop(handle2);
1323 }
1324
1325 #[test]
1326 fn test_observer_ids_are_unique() {
1327 use std::sync::Mutex;
1328
1329 let ids = Arc::new(Mutex::new(std::collections::HashSet::new()));
1330
1331 let mut handles = Vec::new();
1332
1333 for i in 0..100 {
1334 let ids_clone = ids.clone();
1335 let handle = register_apply_observer(Rc::new(move |_, _| {
1336 ids_clone.lock().unwrap().insert(i);
1337 }));
1338 handles.push(handle);
1339 }
1340
1341 notify_apply_observers(&[], 1);
1342 assert_eq!(ids.lock().unwrap().len(), 100);
1343
1344 for i in (0..100).step_by(2) {
1345 handles.remove(i / 2);
1346 }
1347
1348 ids.lock().unwrap().clear();
1349 notify_apply_observers(&[], 2);
1350 assert_eq!(ids.lock().unwrap().len(), 50);
1351 }
1352
1353 #[test]
1354 fn test_state_object_storage_in_modified_set() {
1355 let state = SnapshotState::new(1, SnapshotIdSet::new(), None, None, false);
1356
1357 let state_obj = TestStateObject::new(12345) as Arc<dyn StateObject>;
1358
1359 state.record_write(state_obj.clone(), 1);
1360
1361 let modified = state.modified.borrow();
1362 assert_eq!(modified.len(), 1);
1363 assert!(modified.contains_key(&12345));
1364
1365 let (stored, writer_id) = modified.get(&12345).unwrap();
1366 assert_eq!(stored.object_id().as_usize(), 12345);
1367 assert_eq!(*writer_id, 1);
1368 }
1369
1370 #[test]
1371 fn test_multiple_writes_to_same_state_object() {
1372 let state = SnapshotState::new(1, SnapshotIdSet::new(), None, None, false);
1373 let state_obj = TestStateObject::new(99999) as Arc<dyn StateObject>;
1374
1375 state.record_write(state_obj.clone(), 1);
1376 assert_eq!(state.modified.borrow().len(), 1);
1377
1378 state.record_write(state_obj.clone(), 2);
1379 let modified = state.modified.borrow();
1380 assert_eq!(modified.len(), 1);
1381 assert!(modified.contains_key(&99999));
1382 let (_, writer_id) = modified.get(&99999).unwrap();
1383 assert_eq!(*writer_id, 2);
1384 }
1385}