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 let parent = current_snapshot();
463 match parent {
464 Some(AnySnapshot::TransparentMutable(transparent)) if transparent.can_reuse() => {
465 transparent
466 }
467 _ => {
468 let current = current_snapshot()
469 .unwrap_or_else(|| AnySnapshot::Global(GlobalSnapshot::get_or_create()));
470 let id = current.snapshot_id();
471 let invalid = current.invalid();
472 TransparentObserverMutableSnapshot::new(
473 id,
474 invalid,
475 read_observer,
476 write_observer,
477 None,
478 )
479 }
480 }
481}
482
483pub fn allocate_record_id() -> SnapshotId {
485 runtime::allocate_record_id()
486}
487
488pub(crate) fn peek_next_snapshot_id() -> SnapshotId {
489 runtime::peek_next_snapshot_id()
490}
491
492#[derive(Clone)]
493struct ObserverId(Rc<()>);
494
495impl ObserverId {
496 fn new() -> Self {
497 Self(Rc::new(()))
498 }
499}
500
501impl PartialEq for ObserverId {
502 fn eq(&self, other: &Self) -> bool {
503 Rc::ptr_eq(&self.0, &other.0)
504 }
505}
506
507impl Eq for ObserverId {}
508
509impl Hash for ObserverId {
510 fn hash<H: Hasher>(&self, state: &mut H) {
511 Rc::as_ptr(&self.0).hash(state);
512 }
513}
514
515thread_local! {
516 static APPLY_OBSERVERS: RefCell<HashMap<ObserverId, ApplyObserver>> = RefCell::new(HashMap::default());
517}
518
519thread_local! {
520 static LAST_WRITES: RefCell<HashMap<StateObjectId, SnapshotId>> = RefCell::new(HashMap::default());
521}
522
523thread_local! {
524 static EXTRA_STATE_OBJECTS: RefCell<crate::snapshot_weak_set::SnapshotWeakSet> = RefCell::new(crate::snapshot_weak_set::SnapshotWeakSet::new());
525}
526
527const UNUSED_RECORD_CLEANUP_INTERVAL: SnapshotId = 2;
528const UNUSED_RECORD_CLEANUP_BUSY_INTERVAL: SnapshotId = 1;
529const UNUSED_RECORD_CLEANUP_MIN_SIZE: usize = 64;
530
531thread_local! {
532 static LAST_UNUSED_RECORD_CLEANUP: Cell<SnapshotId> = const { Cell::new(0) };
533}
534
535#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
536pub struct SnapshotV2DebugStats {
537 pub apply_observers_len: usize,
538 pub apply_observers_cap: usize,
539 pub last_writes_len: usize,
540 pub last_writes_cap: usize,
541 pub extra_state_objects_len: usize,
542 pub extra_state_objects_cap: usize,
543 pub last_unused_record_cleanup: SnapshotId,
544}
545
546pub fn debug_snapshot_v2_stats() -> SnapshotV2DebugStats {
547 let (apply_observers_len, apply_observers_cap) = APPLY_OBSERVERS.with(|cell| {
548 let observers = cell.borrow();
549 (observers.len(), observers.capacity())
550 });
551 let (last_writes_len, last_writes_cap) = LAST_WRITES.with(|cell| {
552 let writes = cell.borrow();
553 (writes.len(), writes.capacity())
554 });
555 let SnapshotWeakSetDebugStats {
556 len: extra_state_objects_len,
557 capacity: extra_state_objects_cap,
558 } = EXTRA_STATE_OBJECTS.with(|cell| cell.borrow().debug_stats());
559 let last_unused_record_cleanup = LAST_UNUSED_RECORD_CLEANUP.with(|cell| cell.get());
560
561 SnapshotV2DebugStats {
562 apply_observers_len,
563 apply_observers_cap,
564 last_writes_len,
565 last_writes_cap,
566 extra_state_objects_len,
567 extra_state_objects_cap,
568 last_unused_record_cleanup,
569 }
570}
571
572pub fn register_apply_observer(observer: ApplyObserver) -> ObserverHandle {
576 let id = ObserverId::new();
577 APPLY_OBSERVERS.with(|cell| {
578 cell.borrow_mut().insert(id.clone(), observer);
579 });
580 ObserverHandle {
581 kind: ObserverKind::Apply,
582 id,
583 }
584}
585
586pub struct ObserverHandle {
590 kind: ObserverKind,
591 id: ObserverId,
592}
593
594enum ObserverKind {
595 Apply,
596}
597
598impl Drop for ObserverHandle {
599 fn drop(&mut self) {
600 match self.kind {
601 ObserverKind::Apply => {
602 APPLY_OBSERVERS.with(|cell| {
603 cell.borrow_mut().remove(&self.id);
604 });
605 }
606 }
607 }
608}
609
610pub(crate) fn notify_apply_observers(modified: &[Arc<dyn StateObject>], snapshot_id: SnapshotId) {
611 APPLY_OBSERVERS.with(|cell| {
612 let observers: Vec<ApplyObserver> = cell.borrow().values().cloned().collect();
613 for observer in observers.into_iter() {
614 observer(modified, snapshot_id);
615 }
616 });
617}
618
619pub(crate) fn set_last_write(id: StateObjectId, snapshot_id: SnapshotId) {
620 LAST_WRITES.with(|cell| {
621 cell.borrow_mut().insert(id, snapshot_id);
622 });
623}
624
625#[cfg(test)]
626pub(crate) fn clear_last_writes() {
627 LAST_WRITES.with(|cell| {
628 cell.borrow_mut().clear();
629 });
630}
631
632pub(crate) fn check_and_overwrite_unused_records_locked() {
633 EXTRA_STATE_OBJECTS.with(|cell| {
634 cell.borrow_mut()
635 .remove_if(|state| state.overwrite_unused_records());
636 });
637}
638
639pub(crate) fn maybe_check_and_overwrite_unused_records_locked(current_snapshot_id: SnapshotId) {
640 let should_run = EXTRA_STATE_OBJECTS.with(|cell| {
641 let set = cell.borrow();
642 if set.is_empty() {
643 return false;
644 }
645 let last_cleanup = LAST_UNUSED_RECORD_CLEANUP.with(|last| last.get());
646 let interval = if set.len() >= UNUSED_RECORD_CLEANUP_MIN_SIZE {
647 UNUSED_RECORD_CLEANUP_BUSY_INTERVAL
648 } else {
649 UNUSED_RECORD_CLEANUP_INTERVAL
650 };
651 current_snapshot_id.saturating_sub(last_cleanup) >= interval
652 });
653
654 if should_run {
655 LAST_UNUSED_RECORD_CLEANUP.with(|cell| cell.set(current_snapshot_id));
656 check_and_overwrite_unused_records_locked();
657 }
658}
659
660#[cfg(test)]
661pub(crate) fn clear_unused_record_cleanup_for_tests() {
662 LAST_UNUSED_RECORD_CLEANUP.with(|cell| cell.set(0));
663}
664
665pub(crate) fn optimistic_merges(
666 current_snapshot_id: SnapshotId,
667 base_parent_id: SnapshotId,
668 modified_objects: &[(StateObjectId, Arc<dyn StateObject>, SnapshotId)],
669 invalid_snapshots: &SnapshotIdSet,
670 applying_invalid: &SnapshotIdSet,
671) -> Option<HashMap<usize, Rc<StateRecord>>> {
672 if modified_objects.is_empty() {
673 return None;
674 }
675
676 let mut result: Option<HashMap<usize, Rc<StateRecord>>> = None;
677
678 for (_, state, writer_id) in modified_objects.iter() {
679 let head = state.first_record();
680
681 let current = match crate::state::readable_record_for(
682 &head,
683 current_snapshot_id,
684 invalid_snapshots,
685 ) {
686 Some(record) => record,
687 None => continue,
688 };
689
690 let (previous_opt, found_base) =
691 mutable::find_previous_record(&head, base_parent_id, applying_invalid);
692 let previous = previous_opt?;
693
694 if !found_base || previous.snapshot_id() == crate::state::PREEXISTING_SNAPSHOT_ID {
695 continue;
696 }
697
698 if Rc::ptr_eq(¤t, &previous) {
699 continue;
700 }
701
702 let applied = mutable::find_record_by_id(&head, *writer_id)?;
703
704 let merged = state.merge_records(
705 Rc::clone(&previous),
706 Rc::clone(¤t),
707 Rc::clone(&applied),
708 )?;
709
710 result
711 .get_or_insert_with(HashMap::default)
712 .insert(Rc::as_ptr(¤t) as usize, merged);
713 }
714
715 result
716}
717
718#[allow(clippy::arc_with_non_send_sync)]
724pub fn merge_read_observers(
725 a: Option<ReadObserver>,
726 b: Option<ReadObserver>,
727) -> Option<ReadObserver> {
728 match (a, b) {
729 (None, None) => None,
730 (Some(a), None) => Some(a),
731 (None, Some(b)) => Some(b),
732 (Some(a), Some(b)) => Some(Arc::new(move |state: &dyn StateObject| {
733 a(state);
734 b(state);
735 })),
736 }
737}
738
739#[allow(clippy::arc_with_non_send_sync)]
745pub fn merge_write_observers(
746 a: Option<WriteObserver>,
747 b: Option<WriteObserver>,
748) -> Option<WriteObserver> {
749 match (a, b) {
750 (None, None) => None,
751 (Some(a), None) => Some(a),
752 (None, Some(b)) => Some(b),
753 (Some(a), Some(b)) => Some(Arc::new(move |state: &dyn StateObject| {
754 a(state);
755 b(state);
756 })),
757 }
758}
759
760pub(crate) struct SnapshotState {
761 pub(crate) id: Cell<SnapshotId>,
762 pub(crate) invalid: RefCell<SnapshotIdSet>,
763 pub(crate) pin_handle: Cell<PinHandle>,
764 pub(crate) disposed: Cell<bool>,
765 pub(crate) read_observer: RefCell<Option<ReadObserver>>,
766 pub(crate) write_observer: RefCell<Option<WriteObserver>>,
767 #[allow(clippy::type_complexity)]
768 pub(crate) modified: RefCell<HashMap<StateObjectId, (Arc<dyn StateObject>, SnapshotId)>>,
769 on_dispose: RefCell<Option<Box<dyn FnOnce()>>>,
770 runtime_tracked: bool,
771 pending_children: RefCell<HashSet<SnapshotId>>,
772}
773
774impl SnapshotState {
775 pub(crate) fn new(
776 id: SnapshotId,
777 invalid: SnapshotIdSet,
778 read_observer: Option<ReadObserver>,
779 write_observer: Option<WriteObserver>,
780 runtime_tracked: bool,
781 ) -> Self {
782 Self::new_with_pinning(
783 id,
784 invalid,
785 read_observer,
786 write_observer,
787 runtime_tracked,
788 true,
789 )
790 }
791
792 pub(crate) fn new_with_pinning(
793 id: SnapshotId,
794 invalid: SnapshotIdSet,
795 read_observer: Option<ReadObserver>,
796 write_observer: Option<WriteObserver>,
797 runtime_tracked: bool,
798 should_pin: bool,
799 ) -> Self {
800 let pin_handle = if should_pin {
801 snapshot_pinning::track_pinning(id, &invalid)
802 } else {
803 snapshot_pinning::PinHandle::INVALID
804 };
805 Self {
806 id: Cell::new(id),
807 invalid: RefCell::new(invalid),
808 pin_handle: Cell::new(pin_handle),
809 disposed: Cell::new(false),
810 read_observer: RefCell::new(read_observer),
811 write_observer: RefCell::new(write_observer),
812 modified: RefCell::new(HashMap::default()),
813 on_dispose: RefCell::new(None),
814 runtime_tracked,
815 pending_children: RefCell::new(HashSet::default()),
816 }
817 }
818
819 pub(crate) fn record_read(&self, state: &dyn StateObject) {
820 if let Some(observer) = self.read_observer.borrow().as_ref() {
821 observer(state);
822 }
823 }
824
825 pub(crate) fn record_write(&self, state: Arc<dyn StateObject>, writer_id: SnapshotId) {
826 let state_id = state.object_id().as_usize();
827
828 let mut modified = self.modified.borrow_mut();
829
830 match modified.entry(state_id) {
831 std::collections::hash_map::Entry::Vacant(e) => {
832 if let Some(observer) = self.write_observer.borrow().as_ref() {
833 observer(&*state);
834 }
835 e.insert((state, writer_id));
836 }
837 std::collections::hash_map::Entry::Occupied(mut e) => {
838 e.insert((state, writer_id));
839 }
840 }
841 }
842
843 pub(crate) fn dispose(&self) {
844 if !self.disposed.replace(true) {
845 let pin_handle = self.pin_handle.get();
846 snapshot_pinning::release_pinning(pin_handle);
847 if let Some(cb) = self.on_dispose.borrow_mut().take() {
848 cb();
849 }
850 if self.runtime_tracked {
851 close_snapshot(self.id.get());
852 }
853 }
854 }
855
856 pub(crate) fn add_pending_child(&self, id: SnapshotId) {
857 self.pending_children.borrow_mut().insert(id);
858 }
859
860 pub(crate) fn remove_pending_child(&self, id: SnapshotId) {
861 self.pending_children.borrow_mut().remove(&id);
862 }
863
864 pub(crate) fn has_pending_children(&self) -> bool {
865 !self.pending_children.borrow().is_empty()
866 }
867
868 pub(crate) fn pending_children(&self) -> Vec<SnapshotId> {
869 self.pending_children.borrow().iter().copied().collect()
870 }
871
872 pub(crate) fn set_on_dispose<F>(&self, f: F)
873 where
874 F: FnOnce() + 'static,
875 {
876 *self.on_dispose.borrow_mut() = Some(Box::new(f));
877 }
878}
879
880#[cfg(test)]
881mod tests {
882 use super::*;
883
884 #[test]
885 fn apply_observer_ids_do_not_use_process_global_counter() {
886 let source = include_str!("mod.rs");
887 assert!(!source.contains(concat!("NEXT_", "OBSERVER_ID")));
888 assert!(!source.contains(concat!("Atomic", "Usize")));
889 }
890
891 #[test]
892 fn test_apply_result_is_success() {
893 assert!(SnapshotApplyResult::Success.is_success());
894 assert!(!SnapshotApplyResult::Failure.is_success());
895 }
896
897 #[test]
898 fn test_apply_result_is_failure() {
899 assert!(!SnapshotApplyResult::Success.is_failure());
900 assert!(SnapshotApplyResult::Failure.is_failure());
901 }
902
903 #[test]
904 fn test_apply_result_check_success() {
905 SnapshotApplyResult::Success.check();
906 }
907
908 #[test]
909 #[should_panic(expected = "Snapshot apply failed")]
910 fn test_apply_result_check_failure() {
911 SnapshotApplyResult::Failure.check();
912 }
913
914 #[test]
915 fn snapshot_enter_restores_current_snapshot_after_panic() {
916 let _guard = reset_runtime_for_tests();
917 let snapshot = take_mutable_snapshot(None, None);
918
919 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
920 snapshot.enter(|| panic!("snapshot body panic"));
921 }));
922
923 assert!(result.is_err());
924 assert!(
925 current_snapshot().is_none(),
926 "snapshot enter must restore the previous current snapshot while unwinding"
927 );
928 }
929
930 #[test]
931 fn test_merge_read_observers_both_none() {
932 let result = merge_read_observers(None, None);
933 assert!(result.is_none());
934 }
935
936 #[test]
937 fn test_merge_read_observers_one_some() {
938 let observer = Arc::new(|_: &dyn StateObject| {});
939 let result = merge_read_observers(Some(observer.clone()), None);
940 assert!(result.is_some());
941
942 let result = merge_read_observers(None, Some(observer));
943 assert!(result.is_some());
944 }
945
946 #[test]
947 fn test_merge_write_observers_both_none() {
948 let result = merge_write_observers(None, None);
949 assert!(result.is_none());
950 }
951
952 #[test]
953 fn test_merge_write_observers_one_some() {
954 let observer = Arc::new(|_: &dyn StateObject| {});
955 let result = merge_write_observers(Some(observer.clone()), None);
956 assert!(result.is_some());
957
958 let result = merge_write_observers(None, Some(observer));
959 assert!(result.is_some());
960 }
961
962 #[test]
963 fn test_current_snapshot_none_initially() {
964 set_current_snapshot(None);
965 assert!(current_snapshot().is_none());
966 }
967
968 struct TestStateObject {
969 id: usize,
970 }
971
972 impl TestStateObject {
973 fn new(id: usize) -> Arc<Self> {
974 Arc::new(Self { id })
975 }
976 }
977
978 impl StateObject for TestStateObject {
979 fn object_id(&self) -> crate::state::ObjectId {
980 crate::state::ObjectId(self.id)
981 }
982
983 fn first_record(&self) -> Rc<crate::state::StateRecord> {
984 unimplemented!("Not needed for observer tests")
985 }
986
987 fn try_readable_record(
988 &self,
989 _snapshot_id: SnapshotId,
990 _invalid: &SnapshotIdSet,
991 ) -> Option<Rc<crate::state::StateRecord>> {
992 None
993 }
994
995 fn readable_record(
996 &self,
997 _snapshot_id: SnapshotId,
998 _invalid: &SnapshotIdSet,
999 ) -> Rc<crate::state::StateRecord> {
1000 unimplemented!("Not needed for observer tests")
1001 }
1002
1003 fn prepend_state_record(&self, _record: Rc<crate::state::StateRecord>) {
1004 unimplemented!("Not needed for observer tests")
1005 }
1006
1007 fn promote_record(&self, _child_id: SnapshotId) -> Result<(), &'static str> {
1008 unimplemented!("Not needed for observer tests")
1009 }
1010
1011 fn as_any(&self) -> &dyn std::any::Any {
1012 self
1013 }
1014 }
1015
1016 #[test]
1017 fn test_apply_observer_receives_correct_modified_objects() {
1018 use std::sync::Mutex;
1019
1020 let received_count = Arc::new(Mutex::new(0));
1021 let received_snapshot_id = Arc::new(Mutex::new(0));
1022
1023 let received_count_clone = received_count.clone();
1024 let received_snapshot_id_clone = received_snapshot_id.clone();
1025
1026 let _handle = register_apply_observer(Rc::new(move |modified, snapshot_id| {
1027 *received_snapshot_id_clone.lock().unwrap() = snapshot_id;
1028 *received_count_clone.lock().unwrap() = modified.len();
1029 }));
1030
1031 let obj1: Arc<dyn StateObject> = TestStateObject::new(42);
1032 let obj2: Arc<dyn StateObject> = TestStateObject::new(99);
1033 let modified = vec![obj1, obj2];
1034
1035 notify_apply_observers(&modified, 123);
1036
1037 assert_eq!(*received_snapshot_id.lock().unwrap(), 123);
1038 assert_eq!(*received_count.lock().unwrap(), 2);
1039 }
1040
1041 #[test]
1042 fn test_apply_observer_receives_correct_snapshot_id() {
1043 use std::sync::Mutex;
1044
1045 let received_id = Arc::new(Mutex::new(0));
1046 let received_id_clone = received_id.clone();
1047
1048 let _handle = register_apply_observer(Rc::new(move |_, snapshot_id| {
1049 *received_id_clone.lock().unwrap() = snapshot_id;
1050 }));
1051
1052 notify_apply_observers(&[], 456);
1053
1054 assert_eq!(*received_id.lock().unwrap(), 456);
1055 }
1056
1057 #[test]
1058 fn test_multiple_apply_observers_all_called() {
1059 use std::sync::Mutex;
1060
1061 let call_count1 = Arc::new(Mutex::new(0));
1062 let call_count2 = Arc::new(Mutex::new(0));
1063 let call_count3 = Arc::new(Mutex::new(0));
1064
1065 let call_count1_clone = call_count1.clone();
1066 let call_count2_clone = call_count2.clone();
1067 let call_count3_clone = call_count3.clone();
1068
1069 let _handle1 = register_apply_observer(Rc::new(move |_, _| {
1070 *call_count1_clone.lock().unwrap() += 1;
1071 }));
1072
1073 let _handle2 = register_apply_observer(Rc::new(move |_, _| {
1074 *call_count2_clone.lock().unwrap() += 1;
1075 }));
1076
1077 let _handle3 = register_apply_observer(Rc::new(move |_, _| {
1078 *call_count3_clone.lock().unwrap() += 1;
1079 }));
1080
1081 notify_apply_observers(&[], 1);
1082
1083 assert_eq!(*call_count1.lock().unwrap(), 1);
1084 assert_eq!(*call_count2.lock().unwrap(), 1);
1085 assert_eq!(*call_count3.lock().unwrap(), 1);
1086
1087 notify_apply_observers(&[], 2);
1088
1089 assert_eq!(*call_count1.lock().unwrap(), 2);
1090 assert_eq!(*call_count2.lock().unwrap(), 2);
1091 assert_eq!(*call_count3.lock().unwrap(), 2);
1092 }
1093
1094 #[test]
1095 fn test_apply_observer_not_called_for_empty_modifications() {
1096 use std::sync::Mutex;
1097
1098 let call_count = Arc::new(Mutex::new(0));
1099 let call_count_clone = call_count.clone();
1100
1101 let _handle = register_apply_observer(Rc::new(move |modified, _| {
1102 *call_count_clone.lock().unwrap() += 1;
1103 assert_eq!(modified.len(), 0);
1104 }));
1105
1106 notify_apply_observers(&[], 1);
1107
1108 assert_eq!(*call_count.lock().unwrap(), 1);
1109 }
1110
1111 #[test]
1112 fn test_observer_handle_drop_removes_correct_observer() {
1113 use std::sync::Mutex;
1114
1115 let calls = Arc::new(Mutex::new(Vec::new()));
1116
1117 let calls1 = calls.clone();
1118 let handle1 = register_apply_observer(Rc::new(move |_, _| {
1119 calls1.lock().unwrap().push(1);
1120 }));
1121
1122 let calls2 = calls.clone();
1123 let handle2 = register_apply_observer(Rc::new(move |_, _| {
1124 calls2.lock().unwrap().push(2);
1125 }));
1126
1127 let calls3 = calls.clone();
1128 let handle3 = register_apply_observer(Rc::new(move |_, _| {
1129 calls3.lock().unwrap().push(3);
1130 }));
1131
1132 notify_apply_observers(&[], 1);
1133 let result = calls.lock().unwrap().clone();
1134 assert_eq!(result.len(), 3);
1135 assert!(result.contains(&1));
1136 assert!(result.contains(&2));
1137 assert!(result.contains(&3));
1138 calls.lock().unwrap().clear();
1139
1140 drop(handle2);
1141
1142 notify_apply_observers(&[], 2);
1143 let result = calls.lock().unwrap().clone();
1144 assert_eq!(result.len(), 2);
1145 assert!(result.contains(&1));
1146 assert!(result.contains(&3));
1147 assert!(!result.contains(&2));
1148 calls.lock().unwrap().clear();
1149
1150 drop(handle1);
1151
1152 notify_apply_observers(&[], 3);
1153 let result = calls.lock().unwrap().clone();
1154 assert_eq!(result.len(), 1);
1155 assert!(result.contains(&3));
1156 calls.lock().unwrap().clear();
1157
1158 drop(handle3);
1159
1160 notify_apply_observers(&[], 4);
1161 assert_eq!(calls.lock().unwrap().len(), 0);
1162 }
1163
1164 #[test]
1165 fn test_observer_handle_drop_in_different_orders() {
1166 use std::sync::Mutex;
1167
1168 {
1169 let calls = Arc::new(Mutex::new(Vec::new()));
1170
1171 let calls1 = calls.clone();
1172 let h1 = register_apply_observer(Rc::new(move |_, _| {
1173 calls1.lock().unwrap().push(1);
1174 }));
1175
1176 let calls2 = calls.clone();
1177 let h2 = register_apply_observer(Rc::new(move |_, _| {
1178 calls2.lock().unwrap().push(2);
1179 }));
1180
1181 let calls3 = calls.clone();
1182 let h3 = register_apply_observer(Rc::new(move |_, _| {
1183 calls3.lock().unwrap().push(3);
1184 }));
1185
1186 drop(h3);
1187 notify_apply_observers(&[], 1);
1188 let result = calls.lock().unwrap().clone();
1189 assert!(result.contains(&1) && result.contains(&2) && !result.contains(&3));
1190 calls.lock().unwrap().clear();
1191
1192 drop(h2);
1193 notify_apply_observers(&[], 2);
1194 let result = calls.lock().unwrap().clone();
1195 assert_eq!(result.len(), 1);
1196 assert!(result.contains(&1));
1197 calls.lock().unwrap().clear();
1198
1199 drop(h1);
1200 notify_apply_observers(&[], 3);
1201 assert_eq!(calls.lock().unwrap().len(), 0);
1202 }
1203
1204 {
1205 let calls = Arc::new(Mutex::new(Vec::new()));
1206
1207 let calls1 = calls.clone();
1208 let h1 = register_apply_observer(Rc::new(move |_, _| {
1209 calls1.lock().unwrap().push(1);
1210 }));
1211
1212 let calls2 = calls.clone();
1213 let h2 = register_apply_observer(Rc::new(move |_, _| {
1214 calls2.lock().unwrap().push(2);
1215 }));
1216
1217 let calls3 = calls.clone();
1218 let h3 = register_apply_observer(Rc::new(move |_, _| {
1219 calls3.lock().unwrap().push(3);
1220 }));
1221
1222 drop(h1);
1223 notify_apply_observers(&[], 1);
1224 let result = calls.lock().unwrap().clone();
1225 assert!(!result.contains(&1) && result.contains(&2) && result.contains(&3));
1226 calls.lock().unwrap().clear();
1227
1228 drop(h2);
1229 notify_apply_observers(&[], 2);
1230 let result = calls.lock().unwrap().clone();
1231 assert_eq!(result.len(), 1);
1232 assert!(result.contains(&3));
1233 calls.lock().unwrap().clear();
1234
1235 drop(h3);
1236 notify_apply_observers(&[], 3);
1237 assert_eq!(calls.lock().unwrap().len(), 0);
1238 }
1239 }
1240
1241 #[test]
1242 fn test_remaining_observers_still_work_after_drop() {
1243 use std::sync::Mutex;
1244
1245 let calls = Arc::new(Mutex::new(Vec::new()));
1246
1247 let calls1 = calls.clone();
1248 let handle1 = register_apply_observer(Rc::new(move |_, snapshot_id| {
1249 calls1.lock().unwrap().push((1, snapshot_id));
1250 }));
1251
1252 let calls2 = calls.clone();
1253 let handle2 = register_apply_observer(Rc::new(move |_, snapshot_id| {
1254 calls2.lock().unwrap().push((2, snapshot_id));
1255 }));
1256
1257 notify_apply_observers(&[], 100);
1258 assert_eq!(calls.lock().unwrap().len(), 2);
1259 calls.lock().unwrap().clear();
1260
1261 drop(handle1);
1262
1263 notify_apply_observers(&[], 200);
1264 assert_eq!(*calls.lock().unwrap(), vec![(2, 200)]);
1265 calls.lock().unwrap().clear();
1266
1267 let calls3 = calls.clone();
1268 let _handle3 = register_apply_observer(Rc::new(move |_, snapshot_id| {
1269 calls3.lock().unwrap().push((3, snapshot_id));
1270 }));
1271
1272 notify_apply_observers(&[], 300);
1273 let result = calls.lock().unwrap().clone();
1274 assert_eq!(result.len(), 2);
1275 assert!(result.contains(&(2, 300)));
1276 assert!(result.contains(&(3, 300)));
1277
1278 drop(handle2);
1279 }
1280
1281 #[test]
1282 fn test_observer_ids_are_unique() {
1283 use std::sync::Mutex;
1284
1285 let ids = Arc::new(Mutex::new(std::collections::HashSet::new()));
1286
1287 let mut handles = Vec::new();
1288
1289 for i in 0..100 {
1290 let ids_clone = ids.clone();
1291 let handle = register_apply_observer(Rc::new(move |_, _| {
1292 ids_clone.lock().unwrap().insert(i);
1293 }));
1294 handles.push(handle);
1295 }
1296
1297 notify_apply_observers(&[], 1);
1298 assert_eq!(ids.lock().unwrap().len(), 100);
1299
1300 for i in (0..100).step_by(2) {
1301 handles.remove(i / 2);
1302 }
1303
1304 ids.lock().unwrap().clear();
1305 notify_apply_observers(&[], 2);
1306 assert_eq!(ids.lock().unwrap().len(), 50);
1307 }
1308
1309 #[test]
1310 fn test_state_object_storage_in_modified_set() {
1311 use crate::state::StateObject;
1312
1313 struct TestState;
1314
1315 impl StateObject for TestState {
1316 fn object_id(&self) -> crate::state::ObjectId {
1317 crate::state::ObjectId(12345)
1318 }
1319
1320 fn first_record(&self) -> Rc<crate::state::StateRecord> {
1321 unimplemented!("Not needed for this test")
1322 }
1323
1324 fn try_readable_record(
1325 &self,
1326 _snapshot_id: SnapshotId,
1327 _invalid: &SnapshotIdSet,
1328 ) -> Option<Rc<crate::state::StateRecord>> {
1329 None
1330 }
1331
1332 fn readable_record(
1333 &self,
1334 _snapshot_id: SnapshotId,
1335 _invalid: &SnapshotIdSet,
1336 ) -> Rc<crate::state::StateRecord> {
1337 unimplemented!("Not needed for this test")
1338 }
1339
1340 fn prepend_state_record(&self, _record: Rc<crate::state::StateRecord>) {
1341 unimplemented!("Not needed for this test")
1342 }
1343
1344 fn promote_record(&self, _child_id: SnapshotId) -> Result<(), &'static str> {
1345 unimplemented!("Not needed for this test")
1346 }
1347
1348 fn as_any(&self) -> &dyn std::any::Any {
1349 self
1350 }
1351 }
1352
1353 let state = SnapshotState::new(1, SnapshotIdSet::new(), None, None, false);
1354
1355 let state_obj = Arc::new(TestState) as Arc<dyn StateObject>;
1356
1357 state.record_write(state_obj.clone(), 1);
1358
1359 let modified = state.modified.borrow();
1360 assert_eq!(modified.len(), 1);
1361 assert!(modified.contains_key(&12345));
1362
1363 let (stored, writer_id) = modified.get(&12345).unwrap();
1364 assert_eq!(stored.object_id().as_usize(), 12345);
1365 assert_eq!(*writer_id, 1);
1366 }
1367
1368 #[test]
1369 fn test_multiple_writes_to_same_state_object() {
1370 use crate::state::StateObject;
1371
1372 struct TestState;
1373
1374 impl StateObject for TestState {
1375 fn object_id(&self) -> crate::state::ObjectId {
1376 crate::state::ObjectId(99999)
1377 }
1378
1379 fn first_record(&self) -> Rc<crate::state::StateRecord> {
1380 unimplemented!()
1381 }
1382
1383 fn try_readable_record(
1384 &self,
1385 _snapshot_id: SnapshotId,
1386 _invalid: &SnapshotIdSet,
1387 ) -> Option<Rc<crate::state::StateRecord>> {
1388 None
1389 }
1390
1391 fn readable_record(
1392 &self,
1393 _snapshot_id: SnapshotId,
1394 _invalid: &SnapshotIdSet,
1395 ) -> Rc<crate::state::StateRecord> {
1396 unimplemented!()
1397 }
1398
1399 fn prepend_state_record(&self, _record: Rc<crate::state::StateRecord>) {
1400 unimplemented!()
1401 }
1402
1403 fn promote_record(&self, _child_id: SnapshotId) -> Result<(), &'static str> {
1404 unimplemented!()
1405 }
1406
1407 fn as_any(&self) -> &dyn std::any::Any {
1408 self
1409 }
1410 }
1411
1412 let state = SnapshotState::new(1, SnapshotIdSet::new(), None, None, false);
1413 let state_obj = Arc::new(TestState) as Arc<dyn StateObject>;
1414
1415 state.record_write(state_obj.clone(), 1);
1416 assert_eq!(state.modified.borrow().len(), 1);
1417
1418 state.record_write(state_obj.clone(), 2);
1419 let modified = state.modified.borrow();
1420 assert_eq!(modified.len(), 1);
1421 assert!(modified.contains_key(&99999));
1422 let (_, writer_id) = modified.get(&99999).unwrap();
1423 assert_eq!(*writer_id, 2);
1424 }
1425}