Browse Source
they have been disabled for a long time, and are unlikely to ever make a comebackmerge-requests/626/merge
Matthias Ahouansou
1 year ago
8 changed files with 7 additions and 648 deletions
@ -1,194 +0,0 @@
|
||||
use super::{super::Config, watchers::Watchers}; |
||||
use crossbeam::channel::{bounded, Sender as ChannelSender}; |
||||
use threadpool::ThreadPool; |
||||
|
||||
use crate::{Error, Result}; |
||||
use std::{ |
||||
future::Future, |
||||
pin::Pin, |
||||
sync::{Arc, Mutex}, |
||||
}; |
||||
|
||||
use super::{DatabaseEngine, Tree}; |
||||
|
||||
type TupleOfBytes = (Vec<u8>, Vec<u8>); |
||||
|
||||
pub struct Engine { |
||||
env: heed::Env, |
||||
iter_pool: Mutex<ThreadPool>, |
||||
} |
||||
|
||||
pub struct EngineTree { |
||||
engine: Arc<Engine>, |
||||
tree: Arc<heed::UntypedDatabase>, |
||||
watchers: Watchers, |
||||
} |
||||
|
||||
fn convert_error(error: heed::Error) -> Error { |
||||
Error::HeedError { |
||||
error: error.to_string(), |
||||
} |
||||
} |
||||
|
||||
impl DatabaseEngine for Engine { |
||||
fn open(config: &Config) -> Result<Arc<Self>> { |
||||
let mut env_builder = heed::EnvOpenOptions::new(); |
||||
env_builder.map_size(1024 * 1024 * 1024 * 1024); // 1 Terabyte
|
||||
env_builder.max_readers(126); |
||||
env_builder.max_dbs(128); |
||||
unsafe { |
||||
env_builder.flag(heed::flags::Flags::MdbWriteMap); |
||||
env_builder.flag(heed::flags::Flags::MdbMapAsync); |
||||
} |
||||
|
||||
Ok(Arc::new(Engine { |
||||
env: env_builder |
||||
.open(&config.database_path) |
||||
.map_err(convert_error)?, |
||||
iter_pool: Mutex::new(ThreadPool::new(10)), |
||||
})) |
||||
} |
||||
|
||||
fn open_tree(self: &Arc<Self>, name: &'static str) -> Result<Arc<dyn Tree>> { |
||||
// Creates the db if it doesn't exist already
|
||||
Ok(Arc::new(EngineTree { |
||||
engine: Arc::clone(self), |
||||
tree: Arc::new( |
||||
self.env |
||||
.create_database(Some(name)) |
||||
.map_err(convert_error)?, |
||||
), |
||||
watchers: Default::default(), |
||||
})) |
||||
} |
||||
|
||||
fn flush(self: &Arc<Self>) -> Result<()> { |
||||
self.env.force_sync().map_err(convert_error)?; |
||||
Ok(()) |
||||
} |
||||
} |
||||
|
||||
impl EngineTree { |
||||
fn iter_from_thread( |
||||
&self, |
||||
tree: Arc<heed::UntypedDatabase>, |
||||
from: Vec<u8>, |
||||
backwards: bool, |
||||
) -> Box<dyn Iterator<Item = TupleOfBytes> + Send + Sync> { |
||||
let (s, r) = bounded::<TupleOfBytes>(100); |
||||
let engine = Arc::clone(&self.engine); |
||||
|
||||
let lock = self.engine.iter_pool.lock().await; |
||||
if lock.active_count() < lock.max_count() { |
||||
lock.execute(move || { |
||||
iter_from_thread_work(tree, &engine.env.read_txn().unwrap(), from, backwards, &s); |
||||
}); |
||||
} else { |
||||
std::thread::spawn(move || { |
||||
iter_from_thread_work(tree, &engine.env.read_txn().unwrap(), from, backwards, &s); |
||||
}); |
||||
} |
||||
|
||||
Box::new(r.into_iter()) |
||||
} |
||||
} |
||||
|
||||
fn iter_from_thread_work( |
||||
tree: Arc<heed::UntypedDatabase>, |
||||
txn: &heed::RoTxn<'_>, |
||||
from: Vec<u8>, |
||||
backwards: bool, |
||||
s: &ChannelSender<(Vec<u8>, Vec<u8>)>, |
||||
) { |
||||
if backwards { |
||||
for (k, v) in tree.rev_range(txn, ..=&*from).unwrap().map(|r| r.unwrap()) { |
||||
if s.send((k.to_vec(), v.to_vec())).is_err() { |
||||
return; |
||||
} |
||||
} |
||||
} else { |
||||
if from.is_empty() { |
||||
for (k, v) in tree.iter(txn).unwrap().map(|r| r.unwrap()) { |
||||
if s.send((k.to_vec(), v.to_vec())).is_err() { |
||||
return; |
||||
} |
||||
} |
||||
} else { |
||||
for (k, v) in tree.range(txn, &*from..).unwrap().map(|r| r.unwrap()) { |
||||
if s.send((k.to_vec(), v.to_vec())).is_err() { |
||||
return; |
||||
} |
||||
} |
||||
} |
||||
} |
||||
} |
||||
|
||||
impl Tree for EngineTree { |
||||
fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> { |
||||
let txn = self.engine.env.read_txn().map_err(convert_error)?; |
||||
Ok(self |
||||
.tree |
||||
.get(&txn, &key) |
||||
.map_err(convert_error)? |
||||
.map(|s| s.to_vec())) |
||||
} |
||||
|
||||
fn insert(&self, key: &[u8], value: &[u8]) -> Result<()> { |
||||
let mut txn = self.engine.env.write_txn().map_err(convert_error)?; |
||||
self.tree |
||||
.put(&mut txn, &key, &value) |
||||
.map_err(convert_error)?; |
||||
txn.commit().map_err(convert_error)?; |
||||
self.watchers.wake(key); |
||||
Ok(()) |
||||
} |
||||
|
||||
fn remove(&self, key: &[u8]) -> Result<()> { |
||||
let mut txn = self.engine.env.write_txn().map_err(convert_error)?; |
||||
self.tree.delete(&mut txn, &key).map_err(convert_error)?; |
||||
txn.commit().map_err(convert_error)?; |
||||
Ok(()) |
||||
} |
||||
|
||||
fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + Send + 'a> { |
||||
self.iter_from(&[], false) |
||||
} |
||||
|
||||
fn iter_from( |
||||
&self, |
||||
from: &[u8], |
||||
backwards: bool, |
||||
) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + Send> { |
||||
self.iter_from_thread(Arc::clone(&self.tree), from.to_vec(), backwards) |
||||
} |
||||
|
||||
fn increment(&self, key: &[u8]) -> Result<Vec<u8>> { |
||||
let mut txn = self.engine.env.write_txn().map_err(convert_error)?; |
||||
|
||||
let old = self.tree.get(&txn, &key).map_err(convert_error)?; |
||||
let new = |
||||
crate::utils::increment(old.as_deref()).expect("utils::increment always returns Some"); |
||||
|
||||
self.tree |
||||
.put(&mut txn, &key, &&*new) |
||||
.map_err(convert_error)?; |
||||
|
||||
txn.commit().map_err(convert_error)?; |
||||
|
||||
Ok(new) |
||||
} |
||||
|
||||
fn scan_prefix<'a>( |
||||
&'a self, |
||||
prefix: Vec<u8>, |
||||
) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + Send + 'a> { |
||||
Box::new( |
||||
self.iter_from(&prefix, false) |
||||
.take_while(move |(key, _)| key.starts_with(&prefix)), |
||||
) |
||||
} |
||||
|
||||
fn watch_prefix<'a>(&'a self, prefix: &[u8]) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> { |
||||
self.watchers.watch(prefix) |
||||
} |
||||
} |
||||
@ -1,197 +0,0 @@
|
||||
use crate::{ |
||||
database::{ |
||||
abstraction::{watchers::Watchers, KeyValueDatabaseEngine, KvTree}, |
||||
Config, |
||||
}, |
||||
Result, |
||||
}; |
||||
use persy::{ByteVec, OpenOptions, Persy, Transaction, TransactionConfig, ValueMode}; |
||||
|
||||
use std::{future::Future, pin::Pin, sync::Arc}; |
||||
|
||||
use tracing::warn; |
||||
|
||||
pub struct Engine { |
||||
persy: Persy, |
||||
} |
||||
|
||||
impl KeyValueDatabaseEngine for Arc<Engine> { |
||||
fn open(config: &Config) -> Result<Self> { |
||||
let mut cfg = persy::Config::new(); |
||||
cfg.change_cache_size((config.db_cache_capacity_mb * 1024.0 * 1024.0) as u64); |
||||
|
||||
let persy = OpenOptions::new() |
||||
.create(true) |
||||
.config(cfg) |
||||
.open(&format!("{}/db.persy", config.database_path))?; |
||||
Ok(Arc::new(Engine { persy })) |
||||
} |
||||
|
||||
fn open_tree(&self, name: &'static str) -> Result<Arc<dyn KvTree>> { |
||||
// Create if it doesn't exist
|
||||
if !self.persy.exists_index(name)? { |
||||
let mut tx = self.persy.begin()?; |
||||
tx.create_index::<ByteVec, ByteVec>(name, ValueMode::Replace)?; |
||||
tx.prepare()?.commit()?; |
||||
} |
||||
|
||||
Ok(Arc::new(PersyTree { |
||||
persy: self.persy.clone(), |
||||
name: name.to_owned(), |
||||
watchers: Watchers::default(), |
||||
})) |
||||
} |
||||
|
||||
fn flush(&self) -> Result<()> { |
||||
Ok(()) |
||||
} |
||||
} |
||||
|
||||
pub struct PersyTree { |
||||
persy: Persy, |
||||
name: String, |
||||
watchers: Watchers, |
||||
} |
||||
|
||||
impl PersyTree { |
||||
fn begin(&self) -> Result<Transaction> { |
||||
Ok(self |
||||
.persy |
||||
.begin_with(TransactionConfig::new().set_background_sync(true))?) |
||||
} |
||||
} |
||||
|
||||
impl KvTree for PersyTree { |
||||
fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> { |
||||
let result = self |
||||
.persy |
||||
.get::<ByteVec, ByteVec>(&self.name, &ByteVec::from(key))? |
||||
.next() |
||||
.map(|v| (*v).to_owned()); |
||||
Ok(result) |
||||
} |
||||
|
||||
fn insert(&self, key: &[u8], value: &[u8]) -> Result<()> { |
||||
self.insert_batch(&mut Some((key.to_owned(), value.to_owned())).into_iter())?; |
||||
self.watchers.wake(key); |
||||
Ok(()) |
||||
} |
||||
|
||||
fn insert_batch<'a>(&self, iter: &mut dyn Iterator<Item = (Vec<u8>, Vec<u8>)>) -> Result<()> { |
||||
let mut tx = self.begin()?; |
||||
for (key, value) in iter { |
||||
tx.put::<ByteVec, ByteVec>( |
||||
&self.name, |
||||
ByteVec::from(key.clone()), |
||||
ByteVec::from(value), |
||||
)?; |
||||
} |
||||
tx.prepare()?.commit()?; |
||||
Ok(()) |
||||
} |
||||
|
||||
fn increment_batch<'a>(&self, iter: &mut dyn Iterator<Item = Vec<u8>>) -> Result<()> { |
||||
let mut tx = self.begin()?; |
||||
for key in iter { |
||||
let old = tx |
||||
.get::<ByteVec, ByteVec>(&self.name, &ByteVec::from(key.clone()))? |
||||
.next() |
||||
.map(|v| (*v).to_owned()); |
||||
let new = crate::utils::increment(old.as_deref()).unwrap(); |
||||
tx.put::<ByteVec, ByteVec>(&self.name, ByteVec::from(key), ByteVec::from(new))?; |
||||
} |
||||
tx.prepare()?.commit()?; |
||||
Ok(()) |
||||
} |
||||
|
||||
fn remove(&self, key: &[u8]) -> Result<()> { |
||||
let mut tx = self.begin()?; |
||||
tx.remove::<ByteVec, ByteVec>(&self.name, ByteVec::from(key), None)?; |
||||
tx.prepare()?.commit()?; |
||||
Ok(()) |
||||
} |
||||
|
||||
fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a> { |
||||
let iter = self.persy.range::<ByteVec, ByteVec, _>(&self.name, ..); |
||||
match iter { |
||||
Ok(iter) => Box::new(iter.filter_map(|(k, v)| { |
||||
v.into_iter() |
||||
.map(|val| ((*k).to_owned(), (*val).to_owned())) |
||||
.next() |
||||
})), |
||||
Err(e) => { |
||||
warn!("error iterating {:?}", e); |
||||
Box::new(std::iter::empty()) |
||||
} |
||||
} |
||||
} |
||||
|
||||
fn iter_from<'a>( |
||||
&'a self, |
||||
from: &[u8], |
||||
backwards: bool, |
||||
) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a> { |
||||
let range = if backwards { |
||||
self.persy |
||||
.range::<ByteVec, ByteVec, _>(&self.name, ..=ByteVec::from(from)) |
||||
} else { |
||||
self.persy |
||||
.range::<ByteVec, ByteVec, _>(&self.name, ByteVec::from(from)..) |
||||
}; |
||||
match range { |
||||
Ok(iter) => { |
||||
let map = iter.filter_map(|(k, v)| { |
||||
v.into_iter() |
||||
.map(|val| ((*k).to_owned(), (*val).to_owned())) |
||||
.next() |
||||
}); |
||||
if backwards { |
||||
Box::new(map.rev()) |
||||
} else { |
||||
Box::new(map) |
||||
} |
||||
} |
||||
Err(e) => { |
||||
warn!("error iterating with prefix {:?}", e); |
||||
Box::new(std::iter::empty()) |
||||
} |
||||
} |
||||
} |
||||
|
||||
fn increment(&self, key: &[u8]) -> Result<Vec<u8>> { |
||||
self.increment_batch(&mut Some(key.to_owned()).into_iter())?; |
||||
Ok(self.get(key)?.unwrap()) |
||||
} |
||||
|
||||
fn scan_prefix<'a>( |
||||
&'a self, |
||||
prefix: Vec<u8>, |
||||
) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a> { |
||||
let range_prefix = ByteVec::from(prefix.clone()); |
||||
let range = self |
||||
.persy |
||||
.range::<ByteVec, ByteVec, _>(&self.name, range_prefix..); |
||||
|
||||
match range { |
||||
Ok(iter) => { |
||||
let owned_prefix = prefix.clone(); |
||||
Box::new( |
||||
iter.take_while(move |(k, _)| (*k).starts_with(&owned_prefix)) |
||||
.filter_map(|(k, v)| { |
||||
v.into_iter() |
||||
.map(|val| ((*k).to_owned(), (*val).to_owned())) |
||||
.next() |
||||
}), |
||||
) |
||||
} |
||||
Err(e) => { |
||||
warn!("error scanning prefix {:?}", e); |
||||
Box::new(std::iter::empty()) |
||||
} |
||||
} |
||||
} |
||||
|
||||
fn watch_prefix<'a>(&'a self, prefix: &[u8]) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> { |
||||
self.watchers.watch(prefix) |
||||
} |
||||
} |
||||
@ -1,127 +0,0 @@
|
||||
use super::super::Config; |
||||
use crate::{utils, Result}; |
||||
use std::{future::Future, pin::Pin, sync::Arc}; |
||||
use tracing::warn; |
||||
|
||||
use super::{DatabaseEngine, Tree}; |
||||
|
||||
pub struct Engine(sled::Db); |
||||
|
||||
pub struct SledEngineTree(sled::Tree); |
||||
|
||||
impl DatabaseEngine for Engine { |
||||
fn open(config: &Config) -> Result<Arc<Self>> { |
||||
Ok(Arc::new(Engine( |
||||
sled::Config::default() |
||||
.path(&config.database_path) |
||||
.cache_capacity((config.db_cache_capacity_mb * 1024.0 * 1024.0) as u64) |
||||
.use_compression(true) |
||||
.open()?, |
||||
))) |
||||
} |
||||
|
||||
fn open_tree(self: &Arc<Self>, name: &'static str) -> Result<Arc<dyn Tree>> { |
||||
Ok(Arc::new(SledEngineTree(self.0.open_tree(name)?))) |
||||
} |
||||
|
||||
fn flush(self: &Arc<Self>) -> Result<()> { |
||||
Ok(()) // noop
|
||||
} |
||||
} |
||||
|
||||
impl Tree for SledEngineTree { |
||||
fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> { |
||||
Ok(self.0.get(key)?.map(|v| v.to_vec())) |
||||
} |
||||
|
||||
fn insert(&self, key: &[u8], value: &[u8]) -> Result<()> { |
||||
self.0.insert(key, value)?; |
||||
Ok(()) |
||||
} |
||||
|
||||
fn insert_batch<'a>(&self, iter: &mut dyn Iterator<Item = (Vec<u8>, Vec<u8>)>) -> Result<()> { |
||||
for (key, value) in iter { |
||||
self.0.insert(key, value)?; |
||||
} |
||||
|
||||
Ok(()) |
||||
} |
||||
|
||||
fn remove(&self, key: &[u8]) -> Result<()> { |
||||
self.0.remove(key)?; |
||||
Ok(()) |
||||
} |
||||
|
||||
fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a> { |
||||
Box::new( |
||||
self.0 |
||||
.iter() |
||||
.filter_map(|r| { |
||||
if let Err(e) = &r { |
||||
warn!("Error: {}", e); |
||||
} |
||||
r.ok() |
||||
}) |
||||
.map(|(k, v)| (k.to_vec().into(), v.to_vec().into())), |
||||
) |
||||
} |
||||
|
||||
fn iter_from( |
||||
&self, |
||||
from: &[u8], |
||||
backwards: bool, |
||||
) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)>> { |
||||
let iter = if backwards { |
||||
self.0.range(..=from) |
||||
} else { |
||||
self.0.range(from..) |
||||
}; |
||||
|
||||
let iter = iter |
||||
.filter_map(|r| { |
||||
if let Err(e) = &r { |
||||
warn!("Error: {}", e); |
||||
} |
||||
r.ok() |
||||
}) |
||||
.map(|(k, v)| (k.to_vec().into(), v.to_vec().into())); |
||||
|
||||
if backwards { |
||||
Box::new(iter.rev()) |
||||
} else { |
||||
Box::new(iter) |
||||
} |
||||
} |
||||
|
||||
fn increment(&self, key: &[u8]) -> Result<Vec<u8>> { |
||||
Ok(self |
||||
.0 |
||||
.update_and_fetch(key, utils::increment) |
||||
.map(|o| o.expect("increment always sets a value").to_vec())?) |
||||
} |
||||
|
||||
fn scan_prefix<'a>( |
||||
&'a self, |
||||
prefix: Vec<u8>, |
||||
) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a> { |
||||
let iter = self |
||||
.0 |
||||
.scan_prefix(prefix) |
||||
.filter_map(|r| { |
||||
if let Err(e) = &r { |
||||
warn!("Error: {}", e); |
||||
} |
||||
r.ok() |
||||
}) |
||||
.map(|(k, v)| (k.to_vec().into(), v.to_vec().into())); |
||||
|
||||
Box::new(iter) |
||||
} |
||||
|
||||
fn watch_prefix<'a>(&'a self, prefix: &[u8]) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> { |
||||
let prefix = prefix.to_vec(); |
||||
Box::pin(async move { |
||||
self.0.watch_prefix(prefix).await; |
||||
}) |
||||
} |
||||
} |
||||
Loading…
Reference in new issue