trine-kv 0.6.0

Embedded LSM MVCC key-value database.
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
144
145
146
147
148
149
150
151
152

#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
use super::{Arc, background_worker_loop, lock_poisoned};
use super::{
    BACKGROUND_MAINTENANCE_PROGRESS_WAIT, Db, Duration, KeyRange, MaintenanceBudget,
    MaintenanceRequest, Ordering, Result,
};

impl Db {
    pub(in crate::db) fn start_background_workers(&self) -> Result<()> {
        if !self.background_workers_enabled() {
            return Ok(());
        }

        #[cfg(all(target_arch = "wasm32", target_os = "unknown"))]
        {
            Err(crate::Error::unsupported_backend(
                "browser persistent background workers",
            ))
        }

        #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
        {
            for worker_index in 0..self.inner.options.background_worker_count {
                let inner = Arc::downgrade(&self.inner);
                let maintenance = Arc::clone(&self.inner.maintenance);
                let runtime_shutdown = self.inner.runtime_shutdown.clone();
                let worker = self.inner.runtime.spawn_background(
                    format!("trine-kv-maintenance-{worker_index}"),
                    move || background_worker_loop(&inner, &maintenance, &runtime_shutdown),
                )?;
                self.inner
                    .background_workers
                    .lock()
                    .map_err(|_| lock_poisoned("background worker registry"))?
                    .push(worker);
            }
            self.request_background_maintenance();

            Ok(())
        }
    }

    pub(in crate::db) fn background_workers_enabled(&self) -> bool {
        !self.inner.options.read_only
            && self.inner.options.background_worker_count != 0
            && self.inner.runtime.capabilities().background_threads()
            && self.inner.options.storage_mode.persistent_path().is_some()
    }

    #[cfg_attr(all(target_arch = "wasm32", target_os = "unknown"), allow(dead_code))]
    pub(in crate::db) fn request_background_maintenance(&self) {
        if self.background_workers_enabled() {
            self.inner.maintenance.request(MaintenanceRequest {
                flush: true,
                compaction: true,
            });
        }
    }

    pub(in crate::db) fn request_background_flush(&self) {
        if self.background_workers_enabled() {
            self.inner.maintenance.request(MaintenanceRequest {
                flush: true,
                compaction: false,
            });
        }
    }

    pub(in crate::db) fn request_background_compaction(&self) {
        if self.background_workers_enabled() {
            self.inner.maintenance.request(MaintenanceRequest {
                flush: false,
                compaction: true,
            });
        }
    }

    pub(in crate::db) fn take_background_maintenance_error(&self) -> Result<()> {
        if let Some(error) = self.inner.maintenance.take_error() {
            Err(error)
        } else {
            Ok(())
        }
    }

    pub(in crate::db) fn record_cooperative_maintenance_yield(&self) {
        self.inner
            .maintenance_cooperative_yields
            .fetch_add(1, Ordering::AcqRel);
    }

    pub(in crate::db) fn record_maintenance_budget_exhaustion(&self) {
        self.inner
            .maintenance_budget_exhaustions
            .fetch_add(1, Ordering::AcqRel);
    }

    pub(in crate::db) fn background_maintenance_budget(&self) -> MaintenanceBudget {
        MaintenanceBudget::new(
            self.inner.options.max_immutable_memtables,
            self.inner.options.max_l0_files.saturating_add(1),
        )
    }

    pub(in crate::db) fn background_flush_request_threshold(&self) -> usize {
        self.inner
            .options
            .max_immutable_memtables
            .saturating_sub(1)
            .max(3)
    }

    pub(in crate::db) const fn background_maintenance_progress_wait() -> Duration {
        BACKGROUND_MAINTENANCE_PROGRESS_WAIT
    }

    #[cfg_attr(all(target_arch = "wasm32", target_os = "unknown"), allow(dead_code))]
    pub(in crate::db) fn run_background_maintenance(
        &self,
        request: MaintenanceRequest,
    ) -> Result<()> {
        self.ensure_open()?;
        if self.inner.options.read_only {
            return Ok(());
        }

        let Some(path) = self.persistent_path() else {
            return Ok(());
        };
        let db_path = path.to_path_buf();
        let mut should_compact = request.compaction || self.l0_pressure_exceeded()?;
        let budget = self.background_maintenance_budget();

        if request.flush && self.has_immutable_memtables()? {
            let (flush_should_compact, _) =
                self.run_flush_once_with_budget(&db_path, false, budget)?;
            should_compact |= flush_should_compact;
        }

        if should_compact {
            self.run_compaction_once_with_budget(&db_path, &KeyRange::all(), true, budget)?;
        }
        if self.has_immutable_memtables()? {
            self.request_background_flush();
        }
        if self.l0_pressure_exceeded()? {
            self.request_background_compaction();
        }

        Ok(())
    }
}