|
|
|
|
@ -10,6 +10,7 @@ use tokio::sync::watch;
|
|
|
|
|
|
|
|
|
|
pub struct Engine { |
|
|
|
|
rocks: rocksdb::DBWithThreadMode<rocksdb::MultiThreaded>, |
|
|
|
|
cache: rocksdb::Cache, |
|
|
|
|
old_cfs: Vec<String>, |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -62,6 +63,7 @@ impl DatabaseEngine for Arc<Engine> {
|
|
|
|
|
|
|
|
|
|
Ok(Arc::new(Engine { |
|
|
|
|
rocks: db, |
|
|
|
|
cache: rocksdb_cache, |
|
|
|
|
old_cfs: cfs, |
|
|
|
|
})) |
|
|
|
|
} |
|
|
|
|
@ -90,34 +92,36 @@ impl DatabaseEngine for Arc<Engine> {
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
fn memory_usage(&self) -> Result<String> { |
|
|
|
|
let stats = rocksdb::perf::get_memory_usage_stats(Some(&[&self.rocks]), None)?; |
|
|
|
|
Ok(format!("Approximate memory usage of all the mem-tables: {:.3} MB\n\ |
|
|
|
|
let stats = |
|
|
|
|
rocksdb::perf::get_memory_usage_stats(Some(&[&self.rocks]), Some(&[&self.cache]))?; |
|
|
|
|
Ok(format!( |
|
|
|
|
"Approximate memory usage of all the mem-tables: {:.3} MB\n\ |
|
|
|
|
Approximate memory usage of un-flushed mem-tables: {:.3} MB\n\ |
|
|
|
|
Approximate memory usage of all the table readers: {:.3} MB\n\ |
|
|
|
|
Approximate memory usage by cache: {:.3} MB", |
|
|
|
|
stats.mem_table_total as f64 / 1024.0 / 1024.0, |
|
|
|
|
stats.mem_table_unflushed as f64 / 1024.0 / 1024.0, |
|
|
|
|
stats.mem_table_readers_total as f64 / 1024.0 / 1024.0, |
|
|
|
|
stats.cache_total as f64 / 1024.0 / 1024.0 |
|
|
|
|
stats.mem_table_total as f64 / 1024.0 / 1024.0, |
|
|
|
|
stats.mem_table_unflushed as f64 / 1024.0 / 1024.0, |
|
|
|
|
stats.mem_table_readers_total as f64 / 1024.0 / 1024.0, |
|
|
|
|
stats.cache_total as f64 / 1024.0 / 1024.0 |
|
|
|
|
)) |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
impl RocksDbEngineTree<'_> { |
|
|
|
|
fn cf(&self) -> rocksdb::BoundColumnFamily<'_> { |
|
|
|
|
fn cf(&self) -> Arc<rocksdb::BoundColumnFamily<'_>> { |
|
|
|
|
self.db.rocks.cf_handle(self.name).unwrap() |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
impl Tree for RocksDbEngineTree<'_> { |
|
|
|
|
fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> { |
|
|
|
|
Ok(self.db.rocks.get_cf(self.cf(), key)?) |
|
|
|
|
Ok(self.db.rocks.get_cf(&self.cf(), key)?) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
fn insert(&self, key: &[u8], value: &[u8]) -> Result<()> { |
|
|
|
|
let lock = self.write_lock.read().unwrap(); |
|
|
|
|
|
|
|
|
|
let result = self.db.rocks.put_cf(self.cf(), key, value)?; |
|
|
|
|
let result = self.db.rocks.put_cf(&self.cf(), key, value)?; |
|
|
|
|
|
|
|
|
|
drop(lock); |
|
|
|
|
|
|
|
|
|
@ -146,21 +150,21 @@ impl Tree for RocksDbEngineTree<'_> {
|
|
|
|
|
|
|
|
|
|
fn insert_batch<'a>(&self, iter: &mut dyn Iterator<Item = (Vec<u8>, Vec<u8>)>) -> Result<()> { |
|
|
|
|
for (key, value) in iter { |
|
|
|
|
self.db.rocks.put_cf(self.cf(), key, value)?; |
|
|
|
|
self.db.rocks.put_cf(&self.cf(), key, value)?; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
Ok(()) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
fn remove(&self, key: &[u8]) -> Result<()> { |
|
|
|
|
Ok(self.db.rocks.delete_cf(self.cf(), key)?) |
|
|
|
|
Ok(self.db.rocks.delete_cf(&self.cf(), key)?) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a> { |
|
|
|
|
Box::new( |
|
|
|
|
self.db |
|
|
|
|
.rocks |
|
|
|
|
.iterator_cf(self.cf(), rocksdb::IteratorMode::Start) |
|
|
|
|
.iterator_cf(&self.cf(), rocksdb::IteratorMode::Start) |
|
|
|
|
.map(|(k, v)| (Vec::from(k), Vec::from(v))), |
|
|
|
|
) |
|
|
|
|
} |
|
|
|
|
@ -174,7 +178,7 @@ impl Tree for RocksDbEngineTree<'_> {
|
|
|
|
|
self.db |
|
|
|
|
.rocks |
|
|
|
|
.iterator_cf( |
|
|
|
|
self.cf(), |
|
|
|
|
&self.cf(), |
|
|
|
|
rocksdb::IteratorMode::From( |
|
|
|
|
from, |
|
|
|
|
if backwards { |
|
|
|
|
@ -191,9 +195,9 @@ impl Tree for RocksDbEngineTree<'_> {
|
|
|
|
|
fn increment(&self, key: &[u8]) -> Result<Vec<u8>> { |
|
|
|
|
let lock = self.write_lock.write().unwrap(); |
|
|
|
|
|
|
|
|
|
let old = self.db.rocks.get_cf(self.cf(), &key)?; |
|
|
|
|
let old = self.db.rocks.get_cf(&self.cf(), &key)?; |
|
|
|
|
let new = utils::increment(old.as_deref()).unwrap(); |
|
|
|
|
self.db.rocks.put_cf(self.cf(), key, &new)?; |
|
|
|
|
self.db.rocks.put_cf(&self.cf(), key, &new)?; |
|
|
|
|
|
|
|
|
|
drop(lock); |
|
|
|
|
Ok(new) |
|
|
|
|
@ -203,9 +207,9 @@ impl Tree for RocksDbEngineTree<'_> {
|
|
|
|
|
let lock = self.write_lock.write().unwrap(); |
|
|
|
|
|
|
|
|
|
for key in iter { |
|
|
|
|
let old = self.db.rocks.get_cf(self.cf(), &key)?; |
|
|
|
|
let old = self.db.rocks.get_cf(&self.cf(), &key)?; |
|
|
|
|
let new = utils::increment(old.as_deref()).unwrap(); |
|
|
|
|
self.db.rocks.put_cf(self.cf(), key, new)?; |
|
|
|
|
self.db.rocks.put_cf(&self.cf(), key, new)?; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
drop(lock); |
|
|
|
|
@ -221,7 +225,7 @@ impl Tree for RocksDbEngineTree<'_> {
|
|
|
|
|
self.db |
|
|
|
|
.rocks |
|
|
|
|
.iterator_cf( |
|
|
|
|
self.cf(), |
|
|
|
|
&self.cf(), |
|
|
|
|
rocksdb::IteratorMode::From(&prefix, rocksdb::Direction::Forward), |
|
|
|
|
) |
|
|
|
|
.map(|(k, v)| (Vec::from(k), Vec::from(v))) |
|
|
|
|
|