linkrs-transaction 0.1.0

Transaction management for Linkrs
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

//! SSI (Serializable Snapshot Isolation) rw-dependency tracker.
//!
//! Instead of scanning all committed write sets (O(N)), this tracker maintains
//! per-resource read locks that enable O(1) dangerous-structure detection.

use std::collections::HashMap;

use linkrs_core::types::Timestamp;

use crate::types::{ResourceId, TransactionId, WriteSet};

pub(super) struct SsiTracker {
    /// Per-resource list of active readers: resource → Vec<(txn_id, start_ts)>
    read_locks: parking_lot::RwLock<HashMap<ResourceId, Vec<(TransactionId, Timestamp)>>>,
}

impl SsiTracker {
    pub(super) fn new() -> Self {
        Self {
            read_locks: parking_lot::RwLock::new(HashMap::new()),
        }
    }

    /// Register that `txn_id` read `resource` at `start_ts`.
    fn register_read(&self, txn_id: TransactionId, resource: ResourceId, start_ts: Timestamp) {
        self.read_locks
            .write()
            .entry(resource)
            .or_default()
            .push((txn_id, start_ts));
    }

    /// Register read locks for every entity in `read_set` (vertices, edges,
    /// schema and index resources), so phantom detection covers index reads.
    pub(super) fn register_txn_reads(
        &self,
        txn_id: TransactionId,
        start_ts: Timestamp,
        read_set: &WriteSet,
    ) {
        for vid in read_set.vertices.iter() {
            self.register_read(txn_id, ResourceId::Vertex(*vid), start_ts);
        }
        for edge in read_set.edges.iter() {
            self.register_read(txn_id, ResourceId::Edge(*edge), start_ts);
        }
        for resource in read_set.schema_resources.iter() {
            self.register_read(txn_id, ResourceId::Schema(resource.clone()), start_ts);
        }
        for resource in read_set.index_resources.iter() {
            self.register_read(txn_id, ResourceId::Index(resource.clone()), start_ts);
        }
    }

    /// Snapshot of the readers registered on `resource`.
    pub(super) fn readers(&self, resource: &ResourceId) -> Vec<(TransactionId, Timestamp)> {
        self.read_locks
            .read()
            .get(resource)
            .cloned()
            .unwrap_or_default()
    }

    /// Remove all read locks held by `txn_id` (on commit or abort).
    pub(super) fn unregister_reads(&self, txn_id: TransactionId) {
        let mut locks = self.read_locks.write();
        locks.retain(|_, entries| {
            entries.retain(|(id, _)| *id != txn_id);
            !entries.is_empty()
        });
    }

    /// Prune read locks older than `oldest_active_ts`.
    pub(super) fn prune(&self, oldest_active_ts: Timestamp) {
        let mut locks = self.read_locks.write();
        locks.retain(|_, entries| {
            entries.retain(|(_, ts)| *ts > oldest_active_ts);
            !entries.is_empty()
        });
    }
}