mirror of
https://github.com/bitcoinresearchkit/brk.git
synced 2026-07-26 10:18:10 -07:00
bindex: snapshot
This commit is contained in:
nym21
@@ -0,0 +1,174 @@
|
||||
// https://docs.rs/sanakirja/latest/sanakirja/index.html
|
||||
// https://pijul.org/posts/2021-02-06-rethinking-sanakirja/
|
||||
|
||||
use std::{
|
||||
collections::BTreeMap,
|
||||
fs::{self, File},
|
||||
io,
|
||||
path::{Component, Path, PathBuf},
|
||||
result::Result,
|
||||
};
|
||||
|
||||
use sanakirja::btree::{page, Db_};
|
||||
pub use sanakirja::*;
|
||||
|
||||
use crate::{DatabaseKey, DatabaseValue};
|
||||
|
||||
pub type UnitDatabase = Base<(), ()>;
|
||||
|
||||
///
|
||||
/// A simple wrapper around Sanakirja aatabase that acts as a very fast on disk BTreeMap.
|
||||
///
|
||||
/// The state of the tree is uncommited until `.export()` is called during which it is unsafe to stop the program.
|
||||
///
|
||||
pub struct Base<Key, Value>
|
||||
where
|
||||
Key: DatabaseKey,
|
||||
Value: DatabaseValue,
|
||||
{
|
||||
pathbuf: PathBuf,
|
||||
db: Db_<Key, Value, page::Page<Key, Value>>,
|
||||
txn: MutTxn<Env, ()>,
|
||||
}
|
||||
|
||||
const ROOT_DB: usize = 0;
|
||||
const PAGE_SIZE: u64 = 4096;
|
||||
|
||||
const DEFRAGMENT_RATIO_THRESHOLD: f64 = 0.5;
|
||||
|
||||
impl<Key, Value> Base<Key, Value>
|
||||
where
|
||||
Key: DatabaseKey,
|
||||
Value: DatabaseValue,
|
||||
{
|
||||
const KEY_SIZE: usize = size_of::<Key>();
|
||||
const VALUE_SIZE: usize = size_of::<Value>();
|
||||
const KEY_AND_VALUE_SIZE: usize = Self::KEY_SIZE + Self::VALUE_SIZE;
|
||||
|
||||
/// Open a database without a lock file where only one instance is safe to open.
|
||||
pub fn open(pathbuf: PathBuf) -> Result<Self, Error> {
|
||||
fs::create_dir_all(&pathbuf)?;
|
||||
|
||||
let env = unsafe { Env::new_nolock(Self::path_sanakirja_(&pathbuf), PAGE_SIZE, 1)? };
|
||||
|
||||
let mut txn = Env::mut_txn_begin(env)?;
|
||||
|
||||
let db = txn
|
||||
.root_db(ROOT_DB)
|
||||
.unwrap_or_else(|| unsafe { btree::create_db_(&mut txn).unwrap() });
|
||||
|
||||
Ok(Self { pathbuf, db, txn })
|
||||
}
|
||||
|
||||
pub fn path_sanakirja(&self) -> PathBuf {
|
||||
Self::path_sanakirja_(&self.pathbuf)
|
||||
}
|
||||
fn path_sanakirja_(path: &Path) -> PathBuf {
|
||||
path.join("sanakirja")
|
||||
}
|
||||
|
||||
pub fn path_self_defragmented(&self) -> PathBuf {
|
||||
let defragmented_path_opt: Option<Component> = self.pathbuf.components().last();
|
||||
let folder = match defragmented_path_opt {
|
||||
Some(Component::Normal(f)) => f.to_str().unwrap(),
|
||||
_ => unreachable!(),
|
||||
};
|
||||
let mut original_path = self.pathbuf.clone();
|
||||
original_path.pop();
|
||||
original_path.join(format!("{folder}-defragmented"))
|
||||
}
|
||||
|
||||
pub fn read_length(&self) -> usize {
|
||||
Self::read_length_(&self.pathbuf)
|
||||
}
|
||||
pub fn read_length_(path: &Path) -> usize {
|
||||
fs::read(Self::path_length(path))
|
||||
.map(|v| {
|
||||
let mut buf = [0_u8; 8];
|
||||
v.iter().enumerate().take(8).for_each(|(i, b)| {
|
||||
buf[i] = *b;
|
||||
});
|
||||
usize::from_le_bytes(buf)
|
||||
})
|
||||
.unwrap_or_default()
|
||||
}
|
||||
pub fn write_length(&self, len: usize) -> Result<(), io::Error> {
|
||||
Self::write_length_(&self.pathbuf, len)
|
||||
}
|
||||
pub fn write_length_(path: &Path, len: usize) -> Result<(), io::Error> {
|
||||
fs::write(Self::path_length(path), len.to_le_bytes())
|
||||
}
|
||||
fn path_length(path: &Path) -> PathBuf {
|
||||
path.join("length")
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn get(&self, key: &Key) -> Result<Option<&Value>, Error> {
|
||||
let option = btree::get(&self.txn, &self.db, key, None)?;
|
||||
if let Some((key_found, v)) = option {
|
||||
if key == key_found {
|
||||
return Ok(Some(v));
|
||||
}
|
||||
}
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
/// Iterate over key/value pairs from the database (disk)
|
||||
#[inline]
|
||||
#[allow(clippy::type_complexity)]
|
||||
pub fn iter(
|
||||
&self,
|
||||
) -> Result<btree::Iter<'_, MutTxn<Env, ()>, Key, Value, page::Page<Key, Value>>, Error> {
|
||||
btree::iter(&self.txn, &self.db, None)
|
||||
}
|
||||
|
||||
pub fn put(&mut self, key: &Key, value: &Value) -> Result<bool, Error> {
|
||||
btree::put(&mut self.txn, &mut self.db, key, value)
|
||||
}
|
||||
|
||||
pub fn del(&mut self, key: &Key, value: Option<&Value>) -> Result<bool, Error> {
|
||||
btree::del(&mut self.txn, &mut self.db, key, value)
|
||||
}
|
||||
|
||||
fn get_file_size_to_data_size_ratio(&self, len: usize) -> Result<f64, Error> {
|
||||
let data_bytes = (len * Self::KEY_AND_VALUE_SIZE) as f64;
|
||||
let file_bytes = File::open(&self.pathbuf)?.metadata()?.len() as f64;
|
||||
Ok(file_bytes / data_bytes)
|
||||
}
|
||||
|
||||
pub fn should_defragment(&self, len: usize) -> Result<bool, Error> {
|
||||
Ok(self.get_file_size_to_data_size_ratio(len)? >= DEFRAGMENT_RATIO_THRESHOLD)
|
||||
}
|
||||
|
||||
pub fn iter_collect(&self) -> Result<BTreeMap<&Key, &Value>, Error> {
|
||||
self.iter()?.collect::<_>()
|
||||
}
|
||||
|
||||
pub fn iter_collect_multi(&self) -> Result<BTreeMap<&Key, Vec<&Value>>, Error> {
|
||||
let mut tree: BTreeMap<_, Vec<_>> = BTreeMap::new();
|
||||
self.iter()?.try_for_each(|res| -> Result<(), Error> {
|
||||
let (key, value): (&Key, &Value) = res?;
|
||||
tree.entry(key).or_default().push(value);
|
||||
Ok(())
|
||||
})?;
|
||||
Ok(tree)
|
||||
}
|
||||
|
||||
pub fn commit(mut self, len: usize) -> Result<(), Error> {
|
||||
// dbg!(&self.pathbuf, len);
|
||||
// panic!();
|
||||
self.write_length(len)?;
|
||||
self.txn.set_root(ROOT_DB, self.db.db.into());
|
||||
self.txn.commit()
|
||||
}
|
||||
|
||||
pub fn destroy(self) -> io::Result<()> {
|
||||
let path = self.pathbuf.to_owned();
|
||||
drop(self);
|
||||
fs::remove_dir_all(&path)
|
||||
}
|
||||
|
||||
pub fn path(&self) -> &Path {
|
||||
&self.pathbuf
|
||||
}
|
||||
}
|
||||
+8
-340
@@ -1,346 +1,14 @@
|
||||
// https://docs.rs/sanakirja/latest/sanakirja/index.html
|
||||
// https://pijul.org/posts/2021-02-06-rethinking-sanakirja/
|
||||
|
||||
use core::panic;
|
||||
use std::{
|
||||
collections::{BTreeMap, BTreeSet},
|
||||
fmt::Debug,
|
||||
fs::{self, File},
|
||||
io, mem,
|
||||
path::{Path, PathBuf},
|
||||
result::Result,
|
||||
};
|
||||
|
||||
use sanakirja::btree::{page, Db_};
|
||||
pub use sanakirja::*;
|
||||
|
||||
///
|
||||
/// A simple wrapper around Sanakirja aatabase that acts as a very fast on disk BTreeMap.
|
||||
///
|
||||
/// The state of the tree is uncommited until `.export()` is called during which it is unsafe to stop the program.
|
||||
///
|
||||
pub struct Database<Key, Value>
|
||||
where
|
||||
Key: DatabaseKey,
|
||||
Value: DatabaseValue,
|
||||
{
|
||||
pathbuf: PathBuf,
|
||||
puts: BTreeMap<Key, Value>,
|
||||
dels: BTreeSet<Key>,
|
||||
len: usize,
|
||||
db: Db_<Key, Value, page::Page<Key, Value>>,
|
||||
txn: MutTxn<Env, ()>,
|
||||
}
|
||||
mod base;
|
||||
mod multi;
|
||||
mod traits;
|
||||
mod unique;
|
||||
|
||||
const ROOT_DB: usize = 0;
|
||||
const PAGE_SIZE: u64 = 4096;
|
||||
|
||||
pub type UnitDatabase = Database<(), ()>;
|
||||
|
||||
const DEFRAGMENT_RATIO_THRESHOLD: f64 = 0.5;
|
||||
|
||||
impl<Key, Value> Database<Key, Value>
|
||||
where
|
||||
Key: DatabaseKey,
|
||||
Value: DatabaseValue,
|
||||
{
|
||||
const KEY_SIZE: usize = size_of::<Key>();
|
||||
const VALUE_SIZE: usize = size_of::<Value>();
|
||||
const KEY_AND_VALUE_SIZE: usize = Self::KEY_SIZE + Self::VALUE_SIZE;
|
||||
|
||||
/// Open a database without a lock file where only one instance is safe to open.
|
||||
pub fn open(pathbuf: PathBuf) -> Result<Self, Error> {
|
||||
fs::create_dir_all(&pathbuf)?;
|
||||
|
||||
let env = unsafe { Env::new_nolock(Self::path_sanakirja_(&pathbuf), PAGE_SIZE, 1)? };
|
||||
|
||||
let mut txn = Env::mut_txn_begin(env)?;
|
||||
|
||||
let db = txn
|
||||
.root_db(ROOT_DB)
|
||||
.unwrap_or_else(|| unsafe { btree::create_db_(&mut txn).unwrap() });
|
||||
|
||||
Ok(Self {
|
||||
len: Self::read_length_(&pathbuf),
|
||||
pathbuf,
|
||||
puts: BTreeMap::default(),
|
||||
dels: BTreeSet::default(),
|
||||
db,
|
||||
txn,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn path_sanakirja(&self) -> PathBuf {
|
||||
Self::path_sanakirja_(&self.pathbuf)
|
||||
}
|
||||
fn path_sanakirja_(path: &Path) -> PathBuf {
|
||||
path.join("sanakirja")
|
||||
}
|
||||
|
||||
pub fn read_length(&self) -> usize {
|
||||
Self::read_length_(&self.pathbuf)
|
||||
}
|
||||
pub fn read_length_(path: &Path) -> usize {
|
||||
fs::read(Self::path_length(path))
|
||||
.map(|v| {
|
||||
let mut buf = [0_u8; 8];
|
||||
v.iter().enumerate().take(8).for_each(|(i, b)| {
|
||||
buf[i] = *b;
|
||||
});
|
||||
usize::from_le_bytes(buf)
|
||||
})
|
||||
.unwrap_or_default()
|
||||
}
|
||||
pub fn write_length(&self) -> Result<(), io::Error> {
|
||||
Self::write_length_(&self.pathbuf, self.len)
|
||||
}
|
||||
pub fn write_length_(path: &Path, len: usize) -> Result<(), io::Error> {
|
||||
fs::write(Self::path_length(path), len.to_le_bytes())
|
||||
}
|
||||
fn path_length(path: &Path) -> PathBuf {
|
||||
path.join("length")
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn get(&self, key: &Key) -> Option<&Value> {
|
||||
if let Some(cached_put) = self.get_from_ram(key) {
|
||||
return Some(cached_put);
|
||||
}
|
||||
|
||||
self.get_from_disk(key)
|
||||
}
|
||||
|
||||
/// Get only from the uncommited tree (ram) without checking the database (disk)
|
||||
#[inline]
|
||||
pub fn get_from_ram(&self, key: &Key) -> Option<&Value> {
|
||||
self.puts.get(key)
|
||||
}
|
||||
|
||||
/// Get mut only from the uncommited tree (ram) without checking the database (disk)
|
||||
#[inline]
|
||||
pub fn get_mut_from_ram(&mut self, key: &Key) -> Option<&mut Value> {
|
||||
self.puts.get_mut(key)
|
||||
}
|
||||
|
||||
/// Get only from the database (disk) without checking the uncommited tree (ram)
|
||||
#[inline]
|
||||
pub fn get_from_disk(&self, key: &Key) -> Option<&Value> {
|
||||
let option = btree::get(&self.txn, &self.db, key, None).unwrap();
|
||||
|
||||
if let Some((key_found, v)) = option {
|
||||
if key == key_found {
|
||||
return Some(v);
|
||||
}
|
||||
}
|
||||
|
||||
None
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn insert(&mut self, key: Key, value: Value) -> Option<Value> {
|
||||
self.dels.remove(&key);
|
||||
self.insert_to_ram(key, value)
|
||||
}
|
||||
|
||||
/// Insert without removing the key to the dels tree, so be sure that it hasn't added to the delete set
|
||||
#[inline]
|
||||
pub fn insert_to_ram(&mut self, key: Key, value: Value) -> Option<Value> {
|
||||
self.len += 1;
|
||||
self.puts.insert(key, value)
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn update(&mut self, key: Key, value: Value) -> Option<Value> {
|
||||
self.dels.insert(key.clone());
|
||||
self.puts.insert(key, value)
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn remove(&mut self, key: &Key) -> Option<Value> {
|
||||
self.len -= 1;
|
||||
self.puts.remove(key).or_else(|| {
|
||||
self.dels.insert(key.clone());
|
||||
None
|
||||
})
|
||||
}
|
||||
|
||||
/// Get only from the uncommited tree (ram) without checking the database (disk)
|
||||
#[inline]
|
||||
pub fn remove_from_ram(&mut self, key: &Key) -> Option<Value> {
|
||||
self.len -= 1;
|
||||
self.puts.remove(key)
|
||||
}
|
||||
|
||||
/// Add the key only to the dels tree without checking if it's present in the puts tree, only use if you are positive that you neither added nor updated an entry with this key
|
||||
#[inline]
|
||||
pub fn remove_later_from_disk(&mut self, key: &Key) {
|
||||
self.len -= 1;
|
||||
self.dels.insert(key.clone());
|
||||
}
|
||||
|
||||
/// Iterate over key/value pairs from the uncommited tree (ram)
|
||||
#[inline]
|
||||
pub fn iter_ram(&self) -> std::collections::btree_map::Iter<'_, Key, Value> {
|
||||
self.puts.iter()
|
||||
}
|
||||
|
||||
/// Iterate over key/value pairs from the database (disk)
|
||||
#[inline]
|
||||
pub fn iter_disk(
|
||||
&self,
|
||||
) -> btree::Iter<'_, MutTxn<Env, ()>, Key, Value, page::Page<Key, Value>> {
|
||||
btree::iter(&self.txn, &self.db, None).unwrap()
|
||||
}
|
||||
|
||||
/// Iterate over key/value pairs
|
||||
#[inline]
|
||||
pub fn iter_ram_then_disk(&self) -> impl Iterator<Item = (&Key, &Value)> {
|
||||
self.iter_ram().chain(self.iter_disk().map(|r| r.unwrap()))
|
||||
}
|
||||
|
||||
/// Collect a **clone** of all uncommited key/value pairs (ram)
|
||||
pub fn collect_ram(&self) -> BTreeMap<Key, Value> {
|
||||
self.puts.clone()
|
||||
}
|
||||
|
||||
/// Collect a **clone** of all key/value pairs from the database (disk)
|
||||
pub fn collect_disk(&self) -> BTreeMap<Key, Value> {
|
||||
self.iter_disk()
|
||||
.map(|r| r.unwrap())
|
||||
.map(|(key, value)| (key.clone(), value.clone()))
|
||||
.collect::<_>()
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn len(&self) -> usize {
|
||||
self.len
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.len == 0
|
||||
}
|
||||
|
||||
// pub fn export(self) -> Result<(), Error> {
|
||||
// self.boxed().boxed_export()
|
||||
// }
|
||||
|
||||
// pub fn boxed(self) -> Box<Self> {
|
||||
// Box::new(self)
|
||||
// }
|
||||
|
||||
pub fn get_file_size_to_data_ratio(&self) -> Result<f64, Error> {
|
||||
let data_bytes = (self.len() * Self::KEY_AND_VALUE_SIZE) as f64;
|
||||
let file_bytes = File::open(&self.pathbuf)?.metadata()?.len() as f64;
|
||||
Ok(file_bytes / data_bytes)
|
||||
}
|
||||
|
||||
/// Flush all puts and dels from the ram to disk with an option to defragment the database to save some disk space
|
||||
///
|
||||
/// /!\ Do not kill the program while this function is runnning /!\
|
||||
pub fn export(mut self) -> Result<(), Error> {
|
||||
let defragment = self.get_file_size_to_data_ratio()? >= DEFRAGMENT_RATIO_THRESHOLD;
|
||||
|
||||
if defragment {
|
||||
let mut btree = self.collect_disk();
|
||||
|
||||
let disk_len = btree.len();
|
||||
let dels_len = self.dels.len();
|
||||
let puts_len = self.puts.len();
|
||||
|
||||
let path = self.pathbuf.to_owned();
|
||||
self.dels.iter().for_each(|key| {
|
||||
btree.remove(key);
|
||||
});
|
||||
btree.append(&mut self.puts);
|
||||
|
||||
let len = btree.len();
|
||||
|
||||
if len != self.len {
|
||||
dbg!(len, self.len, path, disk_len, dels_len, puts_len);
|
||||
panic!("Len should be the same");
|
||||
}
|
||||
|
||||
self.destroy()?;
|
||||
|
||||
self = Self::open(path).unwrap();
|
||||
|
||||
if !self.is_empty() {
|
||||
panic!()
|
||||
}
|
||||
|
||||
self.len = len;
|
||||
self.puts = btree;
|
||||
}
|
||||
|
||||
self.write_length()?;
|
||||
|
||||
if self.dels.is_empty() && self.puts.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
mem::take(&mut self.dels)
|
||||
.into_iter()
|
||||
.try_for_each(|key| -> Result<(), Error> {
|
||||
btree::del(&mut self.txn, &mut self.db, &key, None)?;
|
||||
Ok(())
|
||||
})?;
|
||||
|
||||
mem::take(&mut self.puts).into_iter().try_for_each(
|
||||
|(key, value)| -> Result<(), Error> {
|
||||
btree::put(&mut self.txn, &mut self.db, &key, &value)?;
|
||||
Ok(())
|
||||
},
|
||||
)?;
|
||||
|
||||
self.txn.set_root(ROOT_DB, self.db.db.into());
|
||||
|
||||
self.txn.commit()
|
||||
}
|
||||
|
||||
pub fn destroy(self) -> io::Result<()> {
|
||||
let path = self.pathbuf.to_owned();
|
||||
|
||||
drop(self);
|
||||
|
||||
fs::remove_dir_all(&path)
|
||||
}
|
||||
}
|
||||
|
||||
pub trait AnyDatabase {
|
||||
fn export(self) -> Result<(), Error>;
|
||||
// fn boxed_export(self: Box<Self>) -> Result<(), Error>;
|
||||
fn destroy(self) -> io::Result<()>;
|
||||
}
|
||||
|
||||
impl<Key, Value> AnyDatabase for Database<Key, Value>
|
||||
where
|
||||
Key: DatabaseKey,
|
||||
Value: DatabaseValue,
|
||||
{
|
||||
fn export(self) -> Result<(), Error> {
|
||||
self.export()
|
||||
}
|
||||
|
||||
// fn boxed_export(self: Box<Self>) -> Result<(), Error> {
|
||||
// self.boxed_export()
|
||||
// }
|
||||
|
||||
fn destroy(self) -> io::Result<()> {
|
||||
self.destroy()
|
||||
}
|
||||
}
|
||||
|
||||
pub trait DatabaseKey
|
||||
where
|
||||
Self: Ord + Clone + Debug + Storable + Send + Sync,
|
||||
{
|
||||
}
|
||||
impl<T> DatabaseKey for T where T: Ord + Clone + Debug + Storable + Send + Sync {}
|
||||
|
||||
pub trait DatabaseValue
|
||||
where
|
||||
Self: Clone + Storable + PartialEq + Send + Sync,
|
||||
{
|
||||
}
|
||||
impl<T> DatabaseValue for T where T: Clone + Storable + PartialEq + Send + Sync {}
|
||||
pub use base::*;
|
||||
pub use multi::*;
|
||||
pub use traits::*;
|
||||
pub use unique::*;
|
||||
|
||||
+7
-7
@@ -1,16 +1,16 @@
|
||||
use snkrj::Database;
|
||||
use snkrj::DatabaseUnique;
|
||||
|
||||
fn main() {
|
||||
let path = std::env::temp_dir().join("./db");
|
||||
|
||||
let database: Database<i32, i32> = Database::open(path.clone()).unwrap();
|
||||
let _ = database.destroy();
|
||||
// let database: DatabaseUnique<i32, i32> = DatabaseUnique::open(path.clone()).unwrap();
|
||||
// let _ = database.destroy();
|
||||
|
||||
let mut database: Database<i32, i32> = Database::open(path.clone()).unwrap();
|
||||
let mut database: DatabaseUnique<i32, i32> = DatabaseUnique::open(path.clone()).unwrap();
|
||||
database.insert(64, 128);
|
||||
database.export().unwrap();
|
||||
|
||||
let mut database: Database<i32, i32> = Database::open(path).unwrap();
|
||||
let mut database: DatabaseUnique<i32, i32> = DatabaseUnique::open(path).unwrap();
|
||||
database.insert(1, 2);
|
||||
database.insert(128, 256);
|
||||
println!("iter_ram:");
|
||||
@@ -18,11 +18,11 @@ fn main() {
|
||||
println!("{:?}", pair);
|
||||
});
|
||||
println!("iter_disk:");
|
||||
database.iter_disk().for_each(|pair| {
|
||||
database.iter_disk().unwrap().for_each(|pair| {
|
||||
println!("{:?}", pair.unwrap());
|
||||
});
|
||||
println!("iter_ram_then_disk:");
|
||||
database.iter_ram_then_disk().for_each(|pair| {
|
||||
database.iter_ram_then_disk().unwrap().for_each(|pair| {
|
||||
println!("{:?}", pair);
|
||||
});
|
||||
database.export().unwrap();
|
||||
|
||||
@@ -0,0 +1,240 @@
|
||||
// https://docs.rs/sanakirja/latest/sanakirja/index.html
|
||||
// https://pijul.org/posts/2021-02-06-rethinking-sanakirja/
|
||||
|
||||
use core::panic;
|
||||
use std::{
|
||||
collections::{BTreeMap, BTreeSet},
|
||||
fs, mem,
|
||||
path::PathBuf,
|
||||
result::Result,
|
||||
};
|
||||
|
||||
use sanakirja::btree::page;
|
||||
pub use sanakirja::*;
|
||||
|
||||
use crate::{AnyDatabase, Base, DatabaseKey, DatabaseValue};
|
||||
|
||||
///
|
||||
/// A simple wrapper around Sanakirja aatabase that acts as a very fast on disk BTreeMap.
|
||||
///
|
||||
/// The state of the tree is uncommited until `.export()` is called during which it is unsafe to stop the program.
|
||||
///
|
||||
pub struct DatabaseMulti<Key, Value>
|
||||
where
|
||||
Key: DatabaseKey,
|
||||
Value: DatabaseValue,
|
||||
{
|
||||
puts: BTreeMap<Key, Vec<Value>>,
|
||||
dels: BTreeSet<Key>,
|
||||
len: usize,
|
||||
db: Base<Key, Value>,
|
||||
}
|
||||
|
||||
impl<Key, Value> DatabaseMulti<Key, Value>
|
||||
where
|
||||
Key: DatabaseKey,
|
||||
Value: DatabaseValue,
|
||||
{
|
||||
/// Open a database without a lock file where only one instance is safe to open.
|
||||
pub fn open(pathbuf: PathBuf) -> Result<Self, Error> {
|
||||
let db = Base::open(pathbuf)?;
|
||||
Ok(Self {
|
||||
len: db.read_length(),
|
||||
puts: BTreeMap::default(),
|
||||
dels: BTreeSet::default(),
|
||||
db,
|
||||
})
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn get(&self, key: &Key) -> Result<Option<&Value>, Error> {
|
||||
if let Some(cached_put) = self.get_uncommited(key) {
|
||||
return Ok(Some(cached_put));
|
||||
}
|
||||
|
||||
self.db.get(key)
|
||||
}
|
||||
|
||||
/// Get only from the uncommited tree (ram) without checking the database (disk)
|
||||
#[inline]
|
||||
pub fn get_uncommited(&self, key: &Key) -> Option<&Value> {
|
||||
self.puts.get(key).and_then(|v| v.first())
|
||||
}
|
||||
|
||||
/// Get mut only from the uncommited tree (ram) without checking the database (disk)
|
||||
#[inline]
|
||||
pub fn get_mut_uncommited(&mut self, key: &Key) -> Option<&mut Value> {
|
||||
self.puts.get_mut(key).and_then(|v| v.first_mut())
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn insert(&mut self, key: Key, value: Value) -> Option<Value> {
|
||||
self.dels.remove(&key);
|
||||
self.unchecked_insert(key, value)
|
||||
}
|
||||
|
||||
/// Insert without removing the key to the dels tree, so be sure that it hasn't added to the delete set
|
||||
#[inline]
|
||||
pub fn unchecked_insert(&mut self, key: Key, value: Value) -> Option<Value> {
|
||||
self.len += 1;
|
||||
self.puts.entry(key).or_default().push(value);
|
||||
None
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn update(&mut self, key: Key, value: Value) -> Option<Value> {
|
||||
todo!()
|
||||
// self.dels.insert(key.clone());
|
||||
// self.puts.insert(key, value)
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn remove(&mut self, key: &Key) -> Option<Value> {
|
||||
todo!()
|
||||
// self.len -= 1;
|
||||
// self.puts.remove(key).or_else(|| {
|
||||
// self.dels.insert(key.clone());
|
||||
// None
|
||||
// })
|
||||
}
|
||||
|
||||
/// Remove only from the uncommited tree (ram) without checking the database (disk)
|
||||
#[inline]
|
||||
pub fn remove_from_uncommited(&mut self, key: &Key) -> Option<Value> {
|
||||
todo!()
|
||||
// self.len -= 1;
|
||||
// self.puts.remove(key)
|
||||
}
|
||||
|
||||
/// Add the key only to the dels tree without checking if it's present in the puts tree, only use if you are positive that you neither added nor updated an entry with this key
|
||||
#[inline]
|
||||
pub fn remove_later_from_disk(&mut self, key: &Key) {
|
||||
todo!()
|
||||
// self.len -= 1;
|
||||
// self.dels.insert(key.clone());
|
||||
}
|
||||
|
||||
/// Iterate over key/value pairs from the uncommited tree (ram)
|
||||
#[inline]
|
||||
pub fn iter_ram(&self) -> std::collections::btree_map::Iter<'_, Key, Vec<Value>> {
|
||||
self.puts.iter()
|
||||
}
|
||||
|
||||
/// Iterate over key/value pairs from the database (disk)
|
||||
#[inline]
|
||||
#[allow(clippy::type_complexity)]
|
||||
pub fn iter_disk(
|
||||
&self,
|
||||
) -> Result<btree::Iter<'_, MutTxn<Env, ()>, Key, Value, page::Page<Key, Value>>, Error> {
|
||||
self.db.iter()
|
||||
}
|
||||
|
||||
/// Iterate over key/value pairs
|
||||
// #[inline]
|
||||
// pub fn iter_ram_then_disk(&self) -> Result<impl Iterator<Item = (&Key, &Value)>, Error> {
|
||||
// todo!();
|
||||
// // Ok(self.iter_ram().chain(self.iter_disk()?.map(|r| r.unwrap())))
|
||||
// }
|
||||
|
||||
/// Collect a **clone** of all uncommited key/value pairs (ram)
|
||||
pub fn collect_ram(&self) -> BTreeMap<Key, Value> {
|
||||
todo!()
|
||||
// self.puts.clone()
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn len(&self) -> usize {
|
||||
self.len
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.len == 0
|
||||
}
|
||||
|
||||
/// Flush all puts and dels from the ram to disk with an option to defragment the database to save some disk space
|
||||
///
|
||||
/// /!\ Do not kill the program while this function is runnning /!\
|
||||
pub fn export(mut self) -> Result<(), Error> {
|
||||
if self.dels.is_empty() && self.puts.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if self.db.should_defragment(self.len)? {
|
||||
let mut btree = self.db.iter_collect_multi()?;
|
||||
// TODO:
|
||||
// self.dels.iter().for_each(|key| {
|
||||
// btree.remove(key);
|
||||
// });
|
||||
self.puts.iter().for_each(|(key, values)| {
|
||||
// btree.insert(key, value);
|
||||
let vec = btree.entry(key).or_default();
|
||||
vec.extend(values.iter());
|
||||
});
|
||||
|
||||
let path_self_original = self.db.path().to_owned();
|
||||
let path_self_defragmented = self.db.path_self_defragmented();
|
||||
|
||||
let len = btree.values().map(|v| v.len()).sum::<usize>();
|
||||
|
||||
if len != self.len {
|
||||
dbg!(len, self.len, path_self_defragmented);
|
||||
panic!("Len should be the same");
|
||||
}
|
||||
|
||||
{
|
||||
let mut defragmented = Self::open(path_self_defragmented.clone()).unwrap();
|
||||
|
||||
btree
|
||||
.into_iter()
|
||||
.try_for_each(|(key, values)| -> Result<(), Error> {
|
||||
values
|
||||
.into_iter()
|
||||
.try_for_each(|value| -> Result<(), Error> {
|
||||
defragmented.db.put(key, value)?;
|
||||
Ok(())
|
||||
})?;
|
||||
Ok(())
|
||||
})?;
|
||||
|
||||
defragmented.len = len;
|
||||
defragmented.db.commit(self.len)?;
|
||||
}
|
||||
|
||||
drop(self);
|
||||
|
||||
fs::remove_dir_all(&path_self_original)?;
|
||||
fs::rename(&path_self_defragmented, &path_self_original)?;
|
||||
|
||||
Ok(())
|
||||
} else {
|
||||
mem::take(&mut self.dels)
|
||||
.into_iter()
|
||||
.try_for_each(|key| -> Result<(), Error> {
|
||||
self.db.del(&key, None)?;
|
||||
Ok(())
|
||||
})?;
|
||||
|
||||
mem::take(&mut self.puts).into_iter().try_for_each(
|
||||
|(key, vec): (Key, Vec<Value>)| -> Result<(), Error> {
|
||||
vec.into_iter().try_for_each(|value| -> Result<(), Error> {
|
||||
self.db.put(&key, &value)?;
|
||||
Ok(())
|
||||
})
|
||||
},
|
||||
)?;
|
||||
|
||||
self.db.commit(self.len)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<Key, Value> AnyDatabase for DatabaseMulti<Key, Value>
|
||||
where
|
||||
Key: DatabaseKey,
|
||||
Value: DatabaseValue,
|
||||
{
|
||||
fn export(self) -> Result<(), Error> {
|
||||
self.export()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
use std::fmt::Debug;
|
||||
|
||||
use sanakirja::Storable;
|
||||
|
||||
pub trait AnyDatabase {
|
||||
fn export(self) -> Result<(), sanakirja::Error>;
|
||||
// fn destroy(self) -> io::Result<()>;
|
||||
}
|
||||
|
||||
pub trait DatabaseKey
|
||||
where
|
||||
Self: Ord + Clone + Debug + Storable + Send + Sync,
|
||||
{
|
||||
const SIZE: usize = size_of::<Self>();
|
||||
const SIZE_SMALLER_THAN_TWO: bool = Self::SIZE < 2;
|
||||
|
||||
fn as_ne_byte(&self) -> u8 {
|
||||
let data: *const Self = self;
|
||||
let data: *const u8 = data as *const u8;
|
||||
let slice = unsafe { std::slice::from_raw_parts(data, Self::SIZE) };
|
||||
|
||||
*(if cfg!(target_endian = "big") {
|
||||
slice.last()
|
||||
} else {
|
||||
slice.first()
|
||||
})
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn as_ne_six_bits(&self) -> u8 {
|
||||
self.as_ne_byte() >> 2
|
||||
}
|
||||
|
||||
fn as_ne_two_bytes(&self) -> [u8; 2] {
|
||||
let data: *const Self = self;
|
||||
let data: *const u8 = data as *const u8;
|
||||
let slice = unsafe { std::slice::from_raw_parts(data, Self::SIZE) };
|
||||
|
||||
if Self::SIZE_SMALLER_THAN_TWO {
|
||||
panic!("Doesn't make sense")
|
||||
}
|
||||
|
||||
if cfg!(target_endian = "big") {
|
||||
let mut iter = slice.iter().rev();
|
||||
[*iter.next().unwrap(), *iter.next().unwrap()]
|
||||
} else {
|
||||
let mut iter = slice.iter();
|
||||
[*iter.next().unwrap(), *iter.next().unwrap()]
|
||||
}
|
||||
}
|
||||
}
|
||||
impl<T> DatabaseKey for T where T: Ord + Clone + Debug + Storable + Send + Sync {}
|
||||
|
||||
pub trait DatabaseValue
|
||||
where
|
||||
Self: Clone + Storable + PartialEq + Send + Sync,
|
||||
{
|
||||
}
|
||||
impl<T> DatabaseValue for T where T: Clone + Storable + PartialEq + Send + Sync {}
|
||||
@@ -0,0 +1,223 @@
|
||||
// https://docs.rs/sanakirja/latest/sanakirja/index.html
|
||||
// https://pijul.org/posts/2021-02-06-rethinking-sanakirja/
|
||||
|
||||
use core::panic;
|
||||
use std::{
|
||||
collections::{BTreeMap, BTreeSet},
|
||||
fs, mem,
|
||||
path::PathBuf,
|
||||
result::Result,
|
||||
};
|
||||
|
||||
use sanakirja::btree::page;
|
||||
pub use sanakirja::*;
|
||||
|
||||
use crate::{AnyDatabase, Base, DatabaseKey, DatabaseValue};
|
||||
|
||||
///
|
||||
/// A simple wrapper around Sanakirja aatabase that acts as a very fast on disk BTreeMap.
|
||||
///
|
||||
/// The state of the tree is uncommited until `.export()` is called during which it is unsafe to stop the program.
|
||||
///
|
||||
pub struct DatabaseUnique<Key, Value>
|
||||
where
|
||||
Key: DatabaseKey,
|
||||
Value: DatabaseValue,
|
||||
{
|
||||
puts: BTreeMap<Key, Value>,
|
||||
dels: BTreeSet<Key>,
|
||||
len: usize,
|
||||
db: Base<Key, Value>,
|
||||
}
|
||||
|
||||
impl<Key, Value> DatabaseUnique<Key, Value>
|
||||
where
|
||||
Key: DatabaseKey,
|
||||
Value: DatabaseValue,
|
||||
{
|
||||
/// Open a database without a lock file where only one instance is safe to open.
|
||||
pub fn open(pathbuf: PathBuf) -> Result<Self, Error> {
|
||||
let db = Base::open(pathbuf)?;
|
||||
Ok(Self {
|
||||
len: db.read_length(),
|
||||
puts: BTreeMap::default(),
|
||||
dels: BTreeSet::default(),
|
||||
db,
|
||||
})
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn get(&self, key: &Key) -> Result<Option<&Value>, Error> {
|
||||
if let Some(cached_put) = self.get_uncommited(key) {
|
||||
return Ok(Some(cached_put));
|
||||
}
|
||||
|
||||
self.db.get(key)
|
||||
}
|
||||
|
||||
/// Get only from the uncommited tree (ram) without checking the database (disk)
|
||||
#[inline]
|
||||
pub fn get_uncommited(&self, key: &Key) -> Option<&Value> {
|
||||
self.puts.get(key)
|
||||
}
|
||||
|
||||
/// Get mut only from the uncommited tree (ram) without checking the database (disk)
|
||||
#[inline]
|
||||
pub fn get_mut_uncommited(&mut self, key: &Key) -> Option<&mut Value> {
|
||||
self.puts.get_mut(key)
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn insert(&mut self, key: Key, value: Value) -> Option<Value> {
|
||||
self.dels.remove(&key);
|
||||
self.unchecked_insert(key, value)
|
||||
}
|
||||
|
||||
/// Insert without removing the key to the dels tree, so be sure that it hasn't added to the delete set
|
||||
#[inline]
|
||||
pub fn unchecked_insert(&mut self, key: Key, value: Value) -> Option<Value> {
|
||||
self.len += 1;
|
||||
self.puts.insert(key, value)
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn update(&mut self, key: Key, value: Value) -> Option<Value> {
|
||||
self.dels.insert(key.clone());
|
||||
self.puts.insert(key, value)
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn remove(&mut self, key: &Key) -> Option<Value> {
|
||||
self.len -= 1;
|
||||
self.puts.remove(key).or_else(|| {
|
||||
self.dels.insert(key.clone());
|
||||
None
|
||||
})
|
||||
}
|
||||
|
||||
/// Remove only from the uncommited tree (ram) without checking the database (disk)
|
||||
#[inline]
|
||||
pub fn remove_from_uncommited(&mut self, key: &Key) -> Option<Value> {
|
||||
self.len -= 1;
|
||||
self.puts.remove(key)
|
||||
}
|
||||
|
||||
/// Add the key only to the dels tree without checking if it's present in the puts tree, only use if you are positive that you neither added nor updated an entry with this key
|
||||
#[inline]
|
||||
pub fn remove_later_from_disk(&mut self, key: &Key) {
|
||||
self.len -= 1;
|
||||
self.dels.insert(key.clone());
|
||||
}
|
||||
|
||||
/// Iterate over key/value pairs from the uncommited tree (ram)
|
||||
#[inline]
|
||||
pub fn iter_ram(&self) -> std::collections::btree_map::Iter<'_, Key, Value> {
|
||||
self.puts.iter()
|
||||
}
|
||||
|
||||
/// Iterate over key/value pairs from the database (disk)
|
||||
#[inline]
|
||||
#[allow(clippy::type_complexity)]
|
||||
pub fn iter_disk(
|
||||
&self,
|
||||
) -> Result<btree::Iter<'_, MutTxn<Env, ()>, Key, Value, page::Page<Key, Value>>, Error> {
|
||||
self.db.iter()
|
||||
}
|
||||
|
||||
/// Iterate over key/value pairs
|
||||
#[inline]
|
||||
pub fn iter_ram_then_disk(&self) -> Result<impl Iterator<Item = (&Key, &Value)>, Error> {
|
||||
Ok(self.iter_ram().chain(self.iter_disk()?.map(|r| r.unwrap())))
|
||||
}
|
||||
|
||||
/// Collect a **clone** of all uncommited key/value pairs (ram)
|
||||
pub fn collect_ram(&self) -> BTreeMap<Key, Value> {
|
||||
self.puts.clone()
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn len(&self) -> usize {
|
||||
self.len
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.len == 0
|
||||
}
|
||||
|
||||
/// Flush all puts and dels from the ram to disk with an option to defragment the database to save some disk space
|
||||
///
|
||||
/// /!\ Do not kill the program while this function is runnning /!\
|
||||
pub fn export(mut self) -> Result<(), Error> {
|
||||
if self.dels.is_empty() && self.puts.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if self.db.should_defragment(self.len)? {
|
||||
let mut btree = self.db.iter_collect()?;
|
||||
self.dels.iter().for_each(|key| {
|
||||
btree.remove(key);
|
||||
});
|
||||
self.puts.iter().for_each(|(key, value)| {
|
||||
btree.insert(key, value);
|
||||
});
|
||||
|
||||
let path_self_original = self.db.path().to_owned();
|
||||
let path_self_defragmented = self.db.path_self_defragmented();
|
||||
|
||||
let len = btree.len();
|
||||
|
||||
if len != self.len {
|
||||
dbg!(len, self.len, path_self_defragmented);
|
||||
panic!("Len should be the same");
|
||||
}
|
||||
|
||||
{
|
||||
let mut defragmented = Self::open(path_self_defragmented.clone()).unwrap();
|
||||
|
||||
btree
|
||||
.into_iter()
|
||||
.try_for_each(|(key, value)| -> Result<(), Error> {
|
||||
defragmented.db.put(key, value)?;
|
||||
Ok(())
|
||||
})?;
|
||||
|
||||
defragmented.len = len;
|
||||
defragmented.db.commit(self.len)?;
|
||||
}
|
||||
|
||||
drop(self);
|
||||
|
||||
fs::remove_dir_all(&path_self_original)?;
|
||||
fs::rename(&path_self_defragmented, &path_self_original)?;
|
||||
|
||||
Ok(())
|
||||
} else {
|
||||
mem::take(&mut self.dels)
|
||||
.into_iter()
|
||||
.try_for_each(|key| -> Result<(), Error> {
|
||||
self.db.del(&key, None)?;
|
||||
Ok(())
|
||||
})?;
|
||||
|
||||
mem::take(&mut self.puts).into_iter().try_for_each(
|
||||
|(key, value)| -> Result<(), Error> {
|
||||
self.db.put(&key, &value)?;
|
||||
Ok(())
|
||||
},
|
||||
)?;
|
||||
|
||||
self.db.commit(self.len)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<Key, Value> AnyDatabase for DatabaseUnique<Key, Value>
|
||||
where
|
||||
Key: DatabaseKey,
|
||||
Value: DatabaseValue,
|
||||
{
|
||||
fn export(self) -> Result<(), Error> {
|
||||
self.export()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user