slatedb 0.12.1

A cloud native embedded storage engine built on object storage.
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

use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering::SeqCst;

use crate::db_status::DbStatusManager;

/// Oracle is a trait that centralizes the generation & maintenance of various
/// sequence numbers. These sequence numbers are mostly related to the lifecycle
/// of a transaction commit.
pub(crate) trait Oracle: Send + Sync + 'static {
    /// The sequence number of the most recent write that has been fully committed.
    /// For reads with dirty=false, the maximum visible sequence number is capped
    /// at last_committed_seq.
    fn last_committed_seq(&self) -> u64;

    /// The sequence number of the most recent write that has been fully durable
    /// flushed to the remote storage.
    fn last_remote_persisted_seq(&self) -> u64;
}

pub(crate) struct DbOracle {
    // Fields are intentionally private so that all mutations go through methods
    // on this struct.
    last_seq: AtomicU64,
    last_committed_seq: AtomicU64,
    last_durable_seq: AtomicU64,
    status_reporter: DbStatusManager,
}

impl DbOracle {
    pub(crate) fn new(
        last_seq: u64,
        last_committed_seq: u64,
        last_durable_seq: u64,
        status_reporter: DbStatusManager,
    ) -> Self {
        Self {
            last_seq: AtomicU64::new(last_seq),
            last_committed_seq: AtomicU64::new(last_committed_seq),
            last_durable_seq: AtomicU64::new(last_durable_seq),
            status_reporter,
        }
    }

    pub(crate) fn next_seq(&self) -> u64 {
        self.last_seq.fetch_add(1, SeqCst) + 1
    }

    pub(crate) fn peek_next_seq(&self) -> u64 {
        self.last_seq.load(SeqCst) + 1
    }

    pub(crate) fn last_seq(&self) -> u64 {
        self.last_seq.load(SeqCst)
    }

    pub(crate) fn advance_last_seq(&self, seq: u64) {
        self.last_seq.fetch_max(seq, SeqCst);
    }

    pub(crate) fn advance_committed_seq(&self, seq: u64) {
        self.last_committed_seq.fetch_max(seq, SeqCst);
    }

    pub(crate) fn advance_durable_seq(&self, seq: u64) {
        self.last_durable_seq.fetch_max(seq, SeqCst);
        self.status_reporter.report_durable_seq(seq);
    }

    #[cfg(test)]
    pub(crate) fn set_durable_seq_unsafe(&self, value: u64) {
        self.last_durable_seq.store(value, SeqCst);
        self.status_reporter.report_durable_seq(value);
    }
}

impl Oracle for DbOracle {
    fn last_committed_seq(&self) -> u64 {
        self.last_committed_seq.load(SeqCst)
    }

    fn last_remote_persisted_seq(&self) -> u64 {
        self.last_durable_seq.load(SeqCst)
    }
}

pub(crate) struct DbReaderOracle {
    last_remote_persisted_seq: AtomicU64,
    status_reporter: DbStatusManager,
}

impl DbReaderOracle {
    /// for the read-only db instance (DbReader), only the last remote persisted sequence number
    /// is needed to be tracked, and last_seq and last_remote_persisted_seq are considered to be
    /// the same as last_committed_seq.
    pub(crate) fn new(last_remote_persisted_seq: u64, status_reporter: DbStatusManager) -> Self {
        Self {
            last_remote_persisted_seq: AtomicU64::new(last_remote_persisted_seq),
            status_reporter,
        }
    }

    pub(crate) fn advance_durable_seq(&self, seq: u64) {
        self.last_remote_persisted_seq.fetch_max(seq, SeqCst);
        self.status_reporter.report_durable_seq(seq);
    }
}

impl Oracle for DbReaderOracle {
    fn last_committed_seq(&self) -> u64 {
        self.last_remote_persisted_seq.load(SeqCst)
    }

    fn last_remote_persisted_seq(&self) -> u64 {
        self.last_remote_persisted_seq.load(SeqCst)
    }
}