salsa 0.28.5

A generic framework for on-demand, incrementalized computation (experimental)
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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143

#![cfg(all(feature = "inventory", not(feature = "shuttle")))]

//! Readers must not see a reusable slot as valid until its old memos have been cleared.

use std::hash::{Hash, Hasher};
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use crossbeam_channel::bounded;
use salsa::plumbing::{AsId, FromId};
use salsa::{Database, Setter};

#[salsa::db]
#[derive(Clone)]
struct TestDb {
    storage: salsa::Storage<Self>,
}

#[salsa::db]
impl Database for TestDb {}

#[salsa::input]
struct Input {
    #[returns(copy)]
    value: u32,
}

#[salsa::interned(revisions = 1)]
struct Value<'db> {
    #[returns(copy)]
    value: SameShard,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, salsa::SalsaValue)]
struct SameShard(u32);

impl Hash for SameShard {
    fn hash<H: Hasher>(&self, state: &mut H) {
        0u8.hash(state);
    }
}

#[test]
fn reuse_publishes_revision_after_memo_cleanup() {
    assert_unpublished_during_cleanup(false);
}

#[test]
fn reuse_publishes_durability_after_memo_cleanup() {
    assert_unpublished_during_cleanup(true);
}

#[test]
fn panic_during_memo_cleanup_leaves_slot_reusable() {
    let panic_once = AtomicBool::new(true);
    let mut db = database(move || {
        if panic_once.swap(false, Ordering::Relaxed) {
            panic!("memo cleanup panic");
        }
    });
    let input = Input::new(&db, 0);
    let old = intern(&db, input);
    let old_id = old.as_id();
    payload(&db, old);
    input.set_value(&mut db).to(1);

    assert!(catch_unwind(AssertUnwindSafe(|| intern(&db, input))).is_err());

    let replacement = intern(&db, input);
    assert_eq!(replacement.as_id(), old_id.next_generation().unwrap());
    assert_eq!(replacement.value(&db), SameShard(1));
    assert_eq!(payload(&db, replacement), "1");
}

fn assert_unpublished_during_cleanup(non_reusable: bool) {
    let (entered_send, entered_receive) = bounded(1);
    let (resume_send, resume_receive) = bounded::<()>(0);
    let mut db = database(move || {
        let _ = entered_send.send(());
        // Dropping the sender also releases a writer that arrives after the test times out.
        let _ = resume_receive.recv();
    });
    let input = Input::new(&db, 0);
    let old = intern(&db, input);
    let old_id = old.as_id();
    payload(&db, old);
    input.set_value(&mut db).to(1);

    let writer_db = db.clone();
    let writer = std::thread::spawn(move || {
        let replacement = if non_reusable {
            // Values interned outside queries are non-reusable.
            Value::new(&writer_db, SameShard(1))
        } else {
            intern(&writer_db, input)
        };
        assert_eq!(replacement.value(&writer_db), SameShard(1));
        replacement.as_id()
    });

    entered_receive
        .recv_timeout(Duration::from_secs(5))
        .expect("writer did not reach memo cleanup");
    let reader_db = db.clone();
    let (send, receive) = bounded(1);
    let reader = std::thread::spawn(move || {
        // Simulate a stale ID returned by a cached query without validating its producer.
        let stale = Value::from_id(old_id);
        let result = catch_unwind(AssertUnwindSafe(|| stale.value(&reader_db)));
        send.send(result.is_err()).unwrap();
    });

    // The reader must reject the slot even while the writer holds its shard lock. Release
    // the writer on timeout too, so a regression to locking fails instead of deadlocking.
    let rejected = receive.recv_timeout(Duration::from_secs(5));
    drop(resume_send);
    reader.join().unwrap();
    let new_id = writer.join().unwrap();

    assert!(rejected.expect("reader waited for the shard lock"));
    assert_eq!(new_id, old_id.next_generation().unwrap());
}

fn database(on_discard: impl Fn() + Send + Sync + 'static) -> TestDb {
    TestDb {
        storage: salsa::Storage::new(Some(Box::new(move |event| {
            if matches!(event.kind, salsa::EventKind::DidDiscard { .. }) {
                on_discard();
            }
        }))),
    }
}

#[salsa::tracked(returns(copy))]
fn intern(db: &dyn Database, input: Input) -> Value<'_> {
    Value::new(db, SameShard(input.value(db)))
}

#[salsa::tracked(returns(ref))]
fn payload<'db>(db: &'db dyn Database, value: Value<'db>) -> String {
    value.value(db).0.to_string()
}