ashlar_store/
snapshot.rs

1//! Lock-free publishing of the latest document snapshot to other threads.
2//!
3//! The UI thread owns the [`crate::Store`]; an audio callback, a renderer or a background job reads
4//! the newest snapshot from a [`SnapshotCell`] without taking a lock and without ever blocking the
5//! writer (it is an `arc-swap` underneath).
6
7use std::sync::Arc;
8
9use arc_swap::ArcSwap;
10
11use crate::revision::Revision;
12
13/// One published state: the document and its revision.
14#[derive(Debug)]
15pub struct Published<D> {
16    pub doc: Arc<D>,
17    pub revision: Revision,
18}
19
20/// Holds the latest published snapshot.
21pub struct SnapshotCell<D> {
22    cur: ArcSwap<Published<D>>,
23}
24
25impl<D> std::fmt::Debug for SnapshotCell<D> {
26    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
27        f.debug_struct("SnapshotCell").field("revision", &self.revision()).finish()
28    }
29}
30
31impl<D> SnapshotCell<D> {
32    pub fn new(doc: Arc<D>, revision: Revision) -> SnapshotCell<D> {
33        SnapshotCell { cur: ArcSwap::from_pointee(Published { doc, revision }) }
34    }
35
36    /// Replace the published snapshot (writer side; one small allocation).
37    pub fn publish(&self, doc: Arc<D>, revision: Revision) {
38        self.cur.store(Arc::new(Published { doc, revision }));
39    }
40
41    /// The latest snapshot, to keep (a reference count increment, no lock).
42    pub fn load(&self) -> Arc<Published<D>> {
43        self.cur.load_full()
44    }
45
46    /// Read the latest snapshot in place without touching its reference count (the cheapest read,
47    /// for real-time threads). Keep `f` short: the snapshot can't be freed while it runs.
48    pub fn read<R>(&self, f: impl FnOnce(&Published<D>) -> R) -> R {
49        let g = self.cur.load();
50        f(&g)
51    }
52
53    /// Revision of the latest snapshot (to skip work when nothing changed).
54    pub fn revision(&self) -> Revision {
55        self.cur.load().revision
56    }
57}
58
59#[cfg(test)]
60mod tests {
61    use super::*;
62
63    #[test]
64    fn readers_see_latest_and_keep_old() {
65        let r0 = Revision::next();
66        let cell = SnapshotCell::new(Arc::new(1), r0);
67        let old = cell.load();
68        let r1 = Revision::next();
69        cell.publish(Arc::new(2), r1);
70        assert_eq!(*old.doc, 1, "a held snapshot stays valid");
71        assert_eq!(cell.read(|p| *p.doc), 2);
72        assert_eq!(cell.revision(), r1);
73    }
74
75    #[test]
76    fn concurrent_readers_never_see_torn_state() {
77        let cell = Arc::new(SnapshotCell::new(Arc::new((0u64, 0u64)), Revision::next()));
78        let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
79        let readers: Vec<_> = (0..4)
80            .map(|_| {
81                let cell = cell.clone();
82                let stop = stop.clone();
83                std::thread::spawn(move || {
84                    let mut reads = 0u64;
85                    while !stop.load(std::sync::atomic::Ordering::Relaxed) {
86                        cell.read(|p| assert_eq!(p.doc.0, p.doc.1));
87                        reads += 1;
88                    }
89                    reads
90                })
91            })
92            .collect();
93        for i in 1..=20_000u64 {
94            cell.publish(Arc::new((i, i)), Revision::next());
95        }
96        stop.store(true, std::sync::atomic::Ordering::Relaxed);
97        for r in readers {
98            assert!(r.join().unwrap() > 0);
99        }
100        assert_eq!(cell.load().doc.0, 20_000);
101    }
102}
ashlar_store/snapshot.rs

ashlar_store/
snapshot.rs
