ashlar-store 0.1.0

Documents: Arc snapshots, revisions, undo history, versioned formats with migrations, autosave
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102

//! Lock-free publishing of the latest document snapshot to other threads.
//!
//! The UI thread owns the [`crate::Store`]; an audio callback, a renderer or a background job reads
//! the newest snapshot from a [`SnapshotCell`] without taking a lock and without ever blocking the
//! writer (it is an `arc-swap` underneath).

use std::sync::Arc;

use arc_swap::ArcSwap;

use crate::revision::Revision;

/// One published state: the document and its revision.
#[derive(Debug)]
pub struct Published<D> {
    pub doc: Arc<D>,
    pub revision: Revision,
}

/// Holds the latest published snapshot.
pub struct SnapshotCell<D> {
    cur: ArcSwap<Published<D>>,
}

impl<D> std::fmt::Debug for SnapshotCell<D> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SnapshotCell").field("revision", &self.revision()).finish()
    }
}

impl<D> SnapshotCell<D> {
    pub fn new(doc: Arc<D>, revision: Revision) -> SnapshotCell<D> {
        SnapshotCell { cur: ArcSwap::from_pointee(Published { doc, revision }) }
    }

    /// Replace the published snapshot (writer side; one small allocation).
    pub fn publish(&self, doc: Arc<D>, revision: Revision) {
        self.cur.store(Arc::new(Published { doc, revision }));
    }

    /// The latest snapshot, to keep (a reference count increment, no lock).
    pub fn load(&self) -> Arc<Published<D>> {
        self.cur.load_full()
    }

    /// Read the latest snapshot in place without touching its reference count (the cheapest read,
    /// for real-time threads). Keep `f` short: the snapshot can't be freed while it runs.
    pub fn read<R>(&self, f: impl FnOnce(&Published<D>) -> R) -> R {
        let g = self.cur.load();
        f(&g)
    }

    /// Revision of the latest snapshot (to skip work when nothing changed).
    pub fn revision(&self) -> Revision {
        self.cur.load().revision
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn readers_see_latest_and_keep_old() {
        let r0 = Revision::next();
        let cell = SnapshotCell::new(Arc::new(1), r0);
        let old = cell.load();
        let r1 = Revision::next();
        cell.publish(Arc::new(2), r1);
        assert_eq!(*old.doc, 1, "a held snapshot stays valid");
        assert_eq!(cell.read(|p| *p.doc), 2);
        assert_eq!(cell.revision(), r1);
    }

    #[test]
    fn concurrent_readers_never_see_torn_state() {
        let cell = Arc::new(SnapshotCell::new(Arc::new((0u64, 0u64)), Revision::next()));
        let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let readers: Vec<_> = (0..4)
            .map(|_| {
                let cell = cell.clone();
                let stop = stop.clone();
                std::thread::spawn(move || {
                    let mut reads = 0u64;
                    while !stop.load(std::sync::atomic::Ordering::Relaxed) {
                        cell.read(|p| assert_eq!(p.doc.0, p.doc.1));
                        reads += 1;
                    }
                    reads
                })
            })
            .collect();
        for i in 1..=20_000u64 {
            cell.publish(Arc::new((i, i)), Revision::next());
        }
        stop.store(true, std::sync::atomic::Ordering::Relaxed);
        for r in readers {
            assert!(r.join().unwrap() > 0);
        }
        assert_eq!(cell.load().doc.0, 20_000);
    }
}