kitedb 0.2.18

High-performance embedded graph 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

//! MVCC Manager - coordinates MVCC components
//!
//! Mirrors src/mvcc/index.ts (MvccManager)

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
#[cfg(not(target_arch = "wasm32"))]
use std::thread;
#[cfg(not(target_arch = "wasm32"))]
use std::time::Duration;

use parking_lot::Mutex;

use crate::mvcc::{ConflictDetector, GarbageCollector, GcConfig, TxManager, VersionChainManager};
use crate::types::{Timestamp, TxId};

/// MVCC Manager - coordinates all MVCC components
pub struct MvccManager {
  pub tx_manager: Arc<Mutex<TxManager>>,
  pub version_chain: Arc<Mutex<VersionChainManager>>,
  pub conflict_detector: ConflictDetector,
  pub gc: Arc<Mutex<GarbageCollector>>,
  gc_stop: Arc<AtomicBool>,
  #[cfg(not(target_arch = "wasm32"))]
  gc_handle: Mutex<Option<thread::JoinHandle<()>>>,
  #[cfg(target_arch = "wasm32")]
  gc_handle: Mutex<()>,
}

impl MvccManager {
  /// Create a new MVCC manager
  pub fn new(initial_tx_id: TxId, initial_commit_ts: Timestamp, gc_config: GcConfig) -> Self {
    Self {
      tx_manager: Arc::new(Mutex::new(TxManager::with_initial(
        initial_tx_id,
        initial_commit_ts,
      ))),
      version_chain: Arc::new(Mutex::new(VersionChainManager::new())),
      conflict_detector: ConflictDetector::new(),
      gc: Arc::new(Mutex::new(GarbageCollector::with_config(gc_config))),
      gc_stop: Arc::new(AtomicBool::new(false)),
      #[cfg(not(target_arch = "wasm32"))]
      gc_handle: Mutex::new(None),
      #[cfg(target_arch = "wasm32")]
      gc_handle: Mutex::new(()),
    }
  }

  /// Initialize MVCC (starts background GC)
  #[cfg(not(target_arch = "wasm32"))]
  pub fn start(&self) {
    let mut handle_guard = self.gc_handle.lock();
    if handle_guard.is_some() {
      return;
    }

    self.gc_stop.store(false, Ordering::SeqCst);

    // Run immediately on start
    {
      let mut tx_mgr = self.tx_manager.lock();
      let mut vc = self.version_chain.lock();
      let mut gc = self.gc.lock();
      let _ = gc.run_gc(&mut tx_mgr, &mut vc);
    }

    let tx_mgr = self.tx_manager.clone();
    let vc = self.version_chain.clone();
    let gc = self.gc.clone();
    let stop_flag = self.gc_stop.clone();

    let handle = thread::spawn(move || loop {
      let interval_ms = {
        let gc = gc.lock();
        gc.config().interval_ms
      };

      thread::sleep(Duration::from_millis(interval_ms));

      if stop_flag.load(Ordering::SeqCst) {
        break;
      }

      let mut tx_mgr = tx_mgr.lock();
      let mut vc = vc.lock();
      let mut gc = gc.lock();
      let _ = gc.run_gc(&mut tx_mgr, &mut vc);
    });

    *handle_guard = Some(handle);
  }

  #[cfg(target_arch = "wasm32")]
  pub fn start(&self) {
    // No background threads on wasm; run one GC cycle and return.
    let mut tx_mgr = self.tx_manager.lock();
    let mut vc = self.version_chain.lock();
    let mut gc = self.gc.lock();
    let _ = gc.run_gc(&mut tx_mgr, &mut vc);
  }

  /// Shutdown MVCC (stop background GC)
  pub fn stop(&self) {
    self.gc_stop.store(true, Ordering::SeqCst);
    #[cfg(not(target_arch = "wasm32"))]
    {
      if let Some(handle) = self.gc_handle.lock().take() {
        let _ = handle.join();
      }
    }
  }
}

impl Drop for MvccManager {
  fn drop(&mut self) {
    self.stop();
  }
}