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