support for different key value backends
MODIFY
Cargo.lock
Cargo.lock
@@ -885,6 +885,16 @@ dependencies = [
"percent-encoding",
]
+[[package]]
+name = "fs2"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213"
+dependencies = [
+ "libc",
+ "winapi",
+]
+
[[package]]
name = "fs_extra"
version = "1.2.0"
@@ -1096,6 +1106,15 @@ dependencies = [
"tezos_messages",
]
+[[package]]
+name = "fxhash"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c31b6d751ae2c7f11320402d34e41349dd1016f8d5d45e48c4312bc8625af50c"
+dependencies = [
+ "byteorder",
+]
+
[[package]]
name = "generic-array"
version = "0.12.3"
@@ -2789,6 +2808,22 @@ version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c111b5bd5695e56cffe5129854aa230b39c93a305372fdbb2668ca2394eea9f8"
+[[package]]
+name = "sled"
+version = "0.34.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1d0132f3e393bcb7390c60bb45769498cf4550bcb7a21d7f95c02b69f6362cdc"
+dependencies = [
+ "crc32fast",
+ "crossbeam-epoch",
+ "crossbeam-utils",
+ "fs2",
+ "fxhash",
+ "libc",
+ "log",
+ "parking_lot",
+]
+
[[package]]
name = "slog"
version = "2.7.0"
@@ -2930,6 +2965,7 @@ dependencies = [
"rocksdb",
"serde 1.0.123",
"serde_json",
+ "sled",
"slog",
"slog-async",
"slog-term",

MODIFY
light_node/src/configuration.rs
light_node/src/configuration.rs
@@ -16,6 +16,7 @@ use rocksdb::ColumnFamilyDescriptor;
use shell::peer_manager::P2p;
use shell::PeerConnectionThreshold;
use storage::persistent::KeyValueSchema;
+use storage::KeyValueStoreBackend;
use tezos_api::environment;
use tezos_api::environment::TezosEnvironment;
use tezos_api::ffi::PatchContext;
@@ -111,6 +112,7 @@ pub struct Storage {
pub db_path: PathBuf,
pub tezos_data_dir: PathBuf,
pub action_store_backend: Vec<ContextActionStoreBackend>,
+ pub kv_store_backend: KeyValueStoreBackend,
pub compute_context_action_tree_hashes: bool,
pub patch_context: Option<PatchContext>,
}
@@ -865,6 +867,7 @@ impl Environment {
h
}
};
+
let action_store_backend = backends
.iter()
.map(|name| match name.as_str() {
@@ -880,6 +883,14 @@ impl Environment {
})
.collect();
+ let kv_store_backend = match args.value_of("kv-store-backend").unwrap_or("rocksdb")
+ {
+ "inmem" => KeyValueStoreBackend::InMem,
+ "sled" => KeyValueStoreBackend::Sled,
+ "btree" => KeyValueStoreBackend::BTreeMap,
+ _ => KeyValueStoreBackend::RocksDB,
+ };
+
crate::configuration::Storage {
tezos_data_dir: data_dir.clone(),
db,
@@ -888,6 +899,7 @@ impl Environment {
db_path,
compute_context_action_tree_hashes,
action_store_backend,
+ kv_store_backend,
patch_context: {
match args.value_of("sandbox-patch-context-json-file") {
Some(path) => {

MODIFY
light_node/src/main.rs
light_node/src/main.rs
@@ -35,6 +35,7 @@ use storage::persistent::{
};
use storage::ActionFileStorage;
use storage::ContextActionStorage;
+use storage::KeyValueStoreBackend;
use storage::{
check_database_compatibility, context::TezedgeContext, persistent::DBError,
resolve_storage_init_chain_data, BlockStorage, StorageInitInfo,
@@ -593,7 +594,13 @@ fn main() {
);
{
- let persistent_storage = PersistentStorage::new(kv, kv_context, kv_actions, commit_logs);
+ let persistent_storage = PersistentStorage::new(
+ kv,
+ kv_context,
+ kv_actions,
+ commit_logs,
+ env.storage.kv_store_backend.clone(),
+ );
let tezedge_context = TezedgeContext::new(
BlockStorage::new(&persistent_storage),
persistent_storage.merkle(),

MODIFY
storage/Cargo.toml
storage/Cargo.toml
@@ -23,6 +23,7 @@ snap = "1.0.4"
tezos_api = { path = "../tezos/api" }
tezos_context = { path = "../tezos/context" }
tezos_messages = { path = "../tezos/messages" }
+sled = "0.34.6"
bytes = "1.0.1"
[[bench]]

ADD
storage/src/backend/btree_map.rs
+use std::collections::{btree_map::Entry, BTreeMap, HashSet};
+
+use crate::merkle_storage::{ContextValue, EntryHash};
+use crate::persistent::database::RocksDBStats;
+use crate::storage_backend::{
+ StorageBackend as KVStoreTrait, StorageBackendError as KVStoreError,
+ StorageBackendStats as KVStoreStats,
+};
+
+/// In Memory Key Value Store implemented with [BTreeMap](std::collections::BTreeMap)
+#[derive(Debug)]
+pub struct KVStore<K: Ord, V> {
+ kv_map: BTreeMap<K, V>,
+}
+
+impl<K: Ord, V> Default for KVStore<K, V> {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl<K: Ord, V> KVStore<K, V> {
+ pub fn new() -> Self {
+ Self {
+ kv_map: BTreeMap::new(),
+ }
+ }
+}
+
+impl KVStoreTrait for KVStore<EntryHash, ContextValue> {
+ fn is_persisted(&self) -> bool {
+ false
+ }
+
+ /// put kv in map if key doesn't exist. If it does then return false.
+ fn put(&mut self, key: EntryHash, value: ContextValue) -> Result<bool, KVStoreError> {
+ match self.kv_map.entry(key) {
+ Entry::Vacant(entry) => {
+ entry.insert(value);
+ Ok(true)
+ }
+ _ => Ok(false),
+ }
+ }
+
+ fn merge(&mut self, key: EntryHash, value: ContextValue) -> Result<(), KVStoreError> {
+ self.kv_map.insert(key, value);
+ Ok(())
+ }
+
+ fn delete(&mut self, key: &EntryHash) -> Result<Option<ContextValue>, KVStoreError> {
+ Ok(self.kv_map.remove(key))
+ }
+
+ fn get(&self, key: &EntryHash) -> Result<Option<ContextValue>, KVStoreError> {
+ Ok(self.kv_map.get(key).cloned())
+ }
+
+ fn contains(&self, key: &EntryHash) -> Result<bool, KVStoreError> {
+ Ok(self.kv_map.contains_key(key))
+ }
+
+ fn get_mem_use_stats(&self) -> Result<RocksDBStats, KVStoreError> {
+ Ok(RocksDBStats {
+ mem_table_total: 0,
+ mem_table_unflushed: 0,
+ mem_table_readers_total: 0,
+ cache_total: 0,
+ })
+ }
+}
+
+pub type BTreeMapBackend = KVStore<EntryHash, ContextValue>;

ADD
storage/src/backend/in_memory_backend.rs
+// use rayon::prelude::*;
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock};
+
+use crate::merkle_storage::{ContextValue, EntryHash};
+use crate::persistent::database::RocksDBStats;
+use crate::storage_backend::{StorageBackend, StorageBackendError, StorageBackendStats};
+
+#[derive(Default)]
+pub struct InMemoryBackend {
+ inner: Arc<RwLock<HashMap<EntryHash, ContextValue>>>,
+ stats: StorageBackendStats,
+}
+
+impl InMemoryBackend {
+ pub fn new() -> Self {
+ InMemoryBackend {
+ inner: Arc::new(RwLock::new(HashMap::new())),
+ stats: Default::default(),
+ }
+ }
+}
+
+impl StorageBackend for InMemoryBackend {
+ fn is_persisted(&self) -> bool {
+ false
+ }
+
+ fn put(&mut self, key: EntryHash, value: ContextValue) -> Result<bool, StorageBackendError> {
+ let measurement = StorageBackendStats::from((&key, &value));
+ let mut w = self
+ .inner
+ .write()
+ .map_err(|e| StorageBackendError::GuardPoison {
+ error: format!("{}", e),
+ })?;
+
+ let was_added = w.insert(key, value).is_none();
+
+ if was_added {
+ self.stats += measurement;
+ }
+
+ Ok(was_added)
+ }
+
+ fn merge(&mut self, key: EntryHash, value: ContextValue) -> Result<(), StorageBackendError> {
+ let mut w = self
+ .inner
+ .write()
+ .map_err(|e| StorageBackendError::GuardPoison {
+ error: format!("{}", e),
+ })?;
+
+ w.insert(key, value);
+ Ok(())
+ }
+
+ fn delete(&mut self, key: &EntryHash) -> Result<Option<ContextValue>, StorageBackendError> {
+ let mut w = self
+ .inner
+ .write()
+ .map_err(|e| StorageBackendError::GuardPoison {
+ error: format!("{}", e),
+ })?;
+
+ Ok(w.remove(key))
+ }
+
+ fn get(&self, key: &EntryHash) -> Result<Option<ContextValue>, StorageBackendError> {
+ let db = self.inner.clone();
+ let r = db.read().map_err(|e| StorageBackendError::GuardPoison {
+ error: format!("{}", e),
+ })?;
+
+ match r.get(key) {
+ None => Ok(None),
+ Some(v) => Ok(Some(v.clone())),
+ }
+ }
+
+ fn contains(&self, key: &EntryHash) -> Result<bool, StorageBackendError> {
+ let db = self.inner.clone();
+ let r = db.read().map_err(|e| StorageBackendError::GuardPoison {
+ error: format!("{}", e),
+ })?;
+ Ok(r.contains_key(key))
+ }
+
+ fn get_mem_use_stats(&self) -> Result<RocksDBStats, StorageBackendError> {
+ Ok(RocksDBStats {
+ mem_table_total: 0,
+ mem_table_unflushed: 0,
+ mem_table_readers_total: 0,
+ cache_total: 0,
+ })
+ }
+}

ADD
storage/src/backend/mod.rs
+pub mod btree_map;
+pub mod in_memory_backend;
+pub mod rocksdb_backend;
+pub mod sled_backend;
+
+pub use btree_map::*;
+pub use in_memory_backend::*;
+pub use rocksdb_backend::*;
+pub use sled_backend::*;

ADD
storage/src/backend/rocksdb_backend.rs
+use crate::merkle_storage::{ContextValue, EntryHash};
+use crate::persistent::database::KeyValueStoreWithSchema;
+use crate::persistent::database::RocksDBStats;
+use crate::storage_backend::{StorageBackend, StorageBackendError, StorageBackendStats};
+use rocksdb::{WriteOptions, DB};
+use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
+use std::sync::Arc;
+
+pub struct RocksDBBackend {
+ column_name: &'static str,
+ inner: Arc<DB>,
+}
+
+impl RocksDBBackend {
+ pub fn new(db: Arc<DB>, column_name: &'static str) -> Self {
+ RocksDBBackend {
+ inner: db,
+ column_name,
+ }
+ }
+}
+
+impl RocksDBBackend {
+ fn default_write_options() -> WriteOptions {
+ let mut opts = WriteOptions::default();
+ opts.set_sync(false);
+ opts
+ }
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct RocksDBBackendStats {
+ mem_table_total: u64,
+ mem_table_unflushed: u64,
+ mem_table_readers_total: u64,
+ cache_total: u64,
+}
+
+impl StorageBackend for RocksDBBackend {
+ fn is_persisted(&self) -> bool {
+ true
+ }
+
+ fn put(&mut self, key: EntryHash, value: ContextValue) -> Result<bool, StorageBackendError> {
+ let cf = self.inner.cf_handle(&self.column_name).ok_or(
+ StorageBackendError::MissingColumnFamily {
+ name: &self.column_name,
+ },
+ )?;
+
+ self.inner
+ .put_cf_opt(
+ cf,
+ &key.as_ref()[..],
+ &value,
+ &Self::default_write_options(),
+ )
+ .map_err(StorageBackendError::from)
+ .map(|_| true)
+ }
+
+ fn merge(&mut self, key: EntryHash, value: ContextValue) -> Result<(), StorageBackendError> {
+ let cf = self.inner.cf_handle(&self.column_name).ok_or(
+ StorageBackendError::MissingColumnFamily {
+ name: &self.column_name,
+ },
+ )?;
+
+ self.inner
+ .merge_cf_opt(
+ cf,
+ &key.as_ref()[..],
+ &value,
+ &Self::default_write_options(),
+ )
+ .map_err(StorageBackendError::from)
+ }
+
+ /// Warning: always returns None.
+ fn delete(&mut self, key: &EntryHash) -> Result<Option<ContextValue>, StorageBackendError> {
+ let cf = self.inner.cf_handle(&self.column_name).ok_or(
+ StorageBackendError::MissingColumnFamily {
+ name: &self.column_name,
+ },
+ )?;
+
+ self.inner
+ .delete_cf_opt(cf, &key.as_ref()[..], &Self::default_write_options())
+ .map_err(StorageBackendError::from)?;
+ Ok(None)
+ }
+
+ fn get(&self, key: &EntryHash) -> Result<Option<ContextValue>, StorageBackendError> {
+ let cf = self.inner.cf_handle(&self.column_name).ok_or(
+ StorageBackendError::MissingColumnFamily {
+ name: &self.column_name,
+ },
+ )?;
+
+ let v = self
+ .inner
+ .get_cf(cf, &key.as_ref()[..])
+ .map_err(StorageBackendError::from)?
+ .map(|value| value);
+ Ok(v)
+ }
+
+ fn contains(&self, key: &EntryHash) -> Result<bool, StorageBackendError> {
+ self.get(key).map(|v| v.is_some())
+ }
+
+ fn get_mem_use_stats(&self) -> Result<RocksDBStats, StorageBackendError> {
+ // TODO - fix compilation problem
+ // let r = self.inner.get_mem_use_stats();
+ return Err(StorageBackendError::BackendError);
+ }
+}

ADD
storage/src/backend/sled_backend.rs
+use crate::merkle_storage::{ContextValue, EntryHash};
+use crate::persistent::database::RocksDBStats;
+use crate::storage_backend::{StorageBackend, StorageBackendError, StorageBackendStats};
+
+pub struct SledBackend {
+ inner: sled::Tree,
+}
+
+impl SledBackend {
+ pub fn new(db: sled::Tree) -> Self {
+ SledBackend { inner: db }
+ }
+}
+
+impl StorageBackend for SledBackend {
+ fn is_persisted(&self) -> bool {
+ true
+ }
+
+ fn put(&mut self, key: EntryHash, value: ContextValue) -> Result<bool, StorageBackendError> {
+ Ok(self
+ .inner
+ .insert(&key.as_ref()[..], value)
+ .map(|v| v.is_none())?)
+ }
+
+ fn merge(&mut self, key: EntryHash, value: ContextValue) -> Result<(), StorageBackendError> {
+ self.inner.insert(&key.as_ref()[..], value)?;
+ Ok(())
+ }
+
+ fn delete(&mut self, key: &EntryHash) -> Result<Option<ContextValue>, StorageBackendError> {
+ Ok(self.inner.remove(&key.as_ref()[..])?.map(|v| v.to_vec()))
+ }
+
+ fn get(&self, key: &EntryHash) -> Result<Option<ContextValue>, StorageBackendError> {
+ let r = self.inner.get(&key.as_ref()[..])?;
+
+ match r {
+ None => Err(StorageBackendError::BackendError),
+ Some(v) => Ok(Some(v.to_vec())),
+ }
+ }
+
+ fn contains(&self, key: &EntryHash) -> Result<bool, StorageBackendError> {
+ Ok(self.inner.contains_key(&key.as_ref()[..])?)
+ }
+
+ fn get_mem_use_stats(&self) -> Result<RocksDBStats, StorageBackendError> {
+ Ok(RocksDBStats {
+ mem_table_total: 0,
+ mem_table_unflushed: 0,
+ mem_table_readers_total: 0,
+ cache_total: 0,
+ })
+ }
+}

MODIFY
storage/src/lib.rs
storage/src/lib.rs
// SPDX-License-Identifier: MIT
#![forbid(unsafe_code)]
#![feature(const_fn)]
+#![feature(allocator_api)]
use std::convert::{TryFrom, TryInto};
use std::path::Path;
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
use failure::Fail;
use rocksdb::Cache;
@@ -24,6 +25,7 @@ use tezos_messages::p2p::binary_message::{BinaryMessage, MessageHash, MessageHas
use tezos_messages::p2p::encoding::prelude::BlockHeader;
use tezos_messages::Head;
+use crate::backend::RocksDBBackend;
pub use crate::block_meta_storage::{BlockMetaStorage, BlockMetaStorageKV, BlockMetaStorageReader};
pub use crate::block_storage::{
BlockAdditionalData, BlockAdditionalDataBuilder, BlockJsonData, BlockJsonDataBuilder,
@@ -49,6 +51,7 @@ pub use action_file_storage::ActionFileStorage;
pub mod action_file;
pub mod action_file_storage;
+pub mod backend;
pub mod block_meta_storage;
pub mod block_storage;
pub mod chain_meta_storage;
@@ -61,6 +64,7 @@ pub mod operations_storage;
pub mod persistent;
pub mod predecessor_storage;
pub mod skip_list;
+pub mod storage_backend;
pub mod system_storage;
/// Extension of block header with block hash
@@ -466,6 +470,14 @@ pub fn check_database_compatibility(
Ok(db_version_ok && chain_id_ok)
}
+#[derive(PartialEq, Debug, Clone)]
+pub enum KeyValueStoreBackend {
+ RocksDB,
+ InMem,
+ Sled,
+ BTreeMap,
+}
+
pub mod tests_common {
use std::path::{Path, PathBuf};
use std::sync::Arc;
@@ -560,6 +572,7 @@ pub mod tests_common {
Arc::new(kv_context),
Arc::new(kv_context_action),
Arc::new(clog),
+ KeyValueStoreBackend::RocksDB,
),
path,
remove_on_destroy,

MODIFY
storage/src/merkle_storage.rs
storage/src/merkle_storage.rs
@@ -49,11 +49,12 @@ use std::convert::TryInto;
use std::hash::Hash;
use std::sync::Arc;
use std::time::Instant;
+use std::ops::Deref;
use blake2::digest::{Update, VariableOutput};
use blake2::VarBlake2b;
use failure::Fail;
-use rocksdb::{Cache, ColumnFamilyDescriptor, WriteBatch};
+use rocksdb::{Cache, ColumnFamilyDescriptor};
use serde::Deserialize;
use serde::Serialize;
@@ -63,6 +64,7 @@ use crate::persistent;
use crate::persistent::database::RocksDBStats;
use crate::persistent::BincodeEncoded;
use crate::persistent::{default_table_options, KeyValueSchema, KeyValueStoreWithSchema};
+use crate::storage_backend::{StorageBackend, StorageBackendError};
const HASH_LEN: usize = 32;
@@ -126,7 +128,7 @@ enum Action {
Remove(RemoveAction),
}
-pub type MerkleStorageKV = dyn KeyValueStoreWithSchema<MerkleStorage> + Sync + Send;
+pub type MerkleStorageKV = dyn StorageBackend + Sync + Send;
pub type RefCnt = usize;
@@ -134,7 +136,7 @@ pub struct MerkleStorage {
/// tree with current staging area (currently checked out context)
current_stage_tree: Option<Tree>,
current_stage_tree_hash: Option<EntryHash>,
- db: Arc<MerkleStorageKV>,
+ db: Box<MerkleStorageKV>,
/// all entries in current staging area
staged: Vec<(EntryHash, RefCnt, Entry)>,
/// HashMap for looking up entry index in self.staged by hash
@@ -153,8 +155,12 @@ pub enum MerkleError {
DBError {
error: persistent::database::DBError,
},
+ #[fail(display = "KVStore error: {:?}", error)]
+ KVBackendError { error: StorageBackendError },
#[fail(display = "Serialization error: {:?}", error)]
SerializationError { error: bincode::Error },
+ #[fail(display = "Backend error: {:?}", error)]
+ StorageBackendError { error: StorageBackendError },
/// Internal unrecoverable bugs that should never occur
#[fail(display = "No root retrieved for this commit!")]
@@ -187,9 +193,9 @@ pub enum MerkleError {
HashError { error: FromBytesError },
}
-impl From<persistent::database::DBError> for MerkleError {
- fn from(error: persistent::database::DBError) -> Self {
- MerkleError::DBError { error }
+impl From<StorageBackendError> for MerkleError {
+ fn from(error: StorageBackendError) -> Self {
+ MerkleError::StorageBackendError { error }
}
}
@@ -355,7 +361,7 @@ fn hash_commit(commit: &Commit) -> Result<EntryHash, MerkleError> {
}
impl MerkleStorage {
- pub fn new(db: Arc<MerkleStorageKV>) -> Self {
+ pub fn new(db: Box<MerkleStorageKV>) -> Self {
MerkleStorage {
db,
staged: Vec::new(),
@@ -371,6 +377,10 @@ impl MerkleStorage {
}
}
+ pub fn has_persistent_backend(&self) -> bool {
+ self.db.is_persisted()
+ }
+
/// Get value from current staged root
pub fn get(&mut self, key: &ContextKey) -> Result<ContextValue, MerkleError> {
let instant = Instant::now();
@@ -1147,14 +1157,16 @@ impl MerkleStorage {
}
/// Persists an entry and its descendants from staged area to database on disk.
- fn persist_staged_entry_to_db(&self, entry: &Entry) -> Result<(), MerkleError> {
- let mut batch = WriteBatch::default(); // batch containing DB key values to persist
+ fn persist_staged_entry_to_db(&mut self, entry: &Entry) -> Result<(), MerkleError> {
+ let mut batch: Vec<(EntryHash, ContextValue)> = Vec::new();
// build list of entries to be persisted
self.get_entries_recursively(entry, &mut batch)?;
// atomically write all entries in one batch to DB
- self.db.write_batch(batch)?;
+ for (k, v) in batch {
+ self.db.put(k, v)?;
+ }
Ok(())
}
@@ -1163,11 +1175,11 @@ impl MerkleStorage {
fn get_entries_recursively(
&self,
entry: &Entry,
- batch: &mut WriteBatch,
+ batch: &mut Vec<(EntryHash, ContextValue)>,
) -> Result<(), MerkleError> {
// add entry to batch
- self.db
- .put_batch(batch, &self.hash_entry(entry)?, &bincode::serialize(entry)?)?;
+ // self.db.put(self.hash_entry(entry)?, bincode::serialize(entry)?)?;
+ batch.push((self.hash_entry(entry)?, bincode::serialize(entry)?));
match entry {
Entry::Blob(_) => Ok(()),
@@ -1407,6 +1419,7 @@ impl MerkleStorage {
#[allow(unused_must_use)]
mod tests {
// use hex;
+ use crate::backend::{BTreeMapBackend, InMemoryBackend, RocksDBBackend, SledBackend};
use assert_json_diff::assert_json_eq;
use rocksdb::{Options, DB};
use std::path::{Path, PathBuf};
@@ -1436,8 +1449,23 @@ mod tests {
open_db(get_db_name(db_name), &cache)
}
- fn get_storage(dn_name: &str, cache: &Cache) -> MerkleStorage {
- MerkleStorage::new(Arc::new(get_db(dn_name, &cache)))
+ fn get_storage(db_name: &str, cache: &Cache) -> MerkleStorage {
+ let default_backend = "rocksdb".to_string();
+ let backend = env::var("KVBACKEND").unwrap_or(default_backend);
+ let sled = sled::Config::new().path(db_name).open().unwrap();
+
+ match backend.as_str() {
+ "rocksdb" => MerkleStorage::new(Box::new(RocksDBBackend::new(
+ Arc::new(get_db(db_name, &cache)),
+ MerkleStorage::name(),
+ ))),
+ "sled" => MerkleStorage::new(Box::new(SledBackend::new(sled.deref().clone()))),
+ "btree" => MerkleStorage::new(Box::new(BTreeMapBackend::new())),
+ "inmem" => MerkleStorage::new(Box::new(InMemoryBackend::new())),
+ _ => {
+ panic!("unknown backend set")
+ }
+ }
}
fn clean_db(db_name: &str) {
@@ -1832,6 +1860,9 @@ mod tests {
let cache = Cache::new_lru_cache(32 * 1024 * 1024).unwrap();
let mut storage = get_storage(db_name, &cache);
+ if !storage.has_persistent_backend() {
+ return;
+ }
assert_eq!(
storage.get_history(&commit1, key_abc).unwrap(),
@@ -1987,6 +2018,9 @@ mod tests {
let cache = Cache::new_lru_cache(32 * 1024 * 1024).unwrap();
let mut storage = get_storage(db_name, &cache);
+ if !storage.has_persistent_backend() {
+ return;
+ }
storage.checkout(&commit1);
assert_eq!(storage.get(&key_abc).unwrap(), vec![1u8]);
assert_eq!(storage.get(&key_abx).unwrap(), vec![2u8]);
@@ -2018,6 +2052,9 @@ mod tests {
let cache = Cache::new_lru_cache(32 * 1024 * 1024).unwrap();
let mut storage = get_storage(db_name, &cache);
+ if !storage.has_persistent_backend() {
+ return;
+ }
assert_eq!(vec![2_u8], storage.get_history(&commit1, &key_abc).unwrap());
}
@@ -2028,15 +2065,25 @@ mod tests {
{
clean_db(db_name);
let cache = Cache::new_lru_cache(32 * 1024 * 1024).unwrap();
- get_storage(db_name, &cache);
+ MerkleStorage::new(Box::new(RocksDBBackend::new(
+ Arc::new(get_db(db_name, &cache)),
+ MerkleStorage::name(),
+ )));
}
let db = DB::open_for_read_only(&Options::default(), get_db_name(db_name), true).unwrap();
- let mut storage = MerkleStorage::new(Arc::new(db));
+ let mut storage = MerkleStorage::new(Box::new(RocksDBBackend::new(
+ Arc::new(db),
+ MerkleStorage::name(),
+ )));
storage.set(&vec!["a".to_string()], &vec![1u8]);
let res = storage.commit(0, "".to_string(), "".to_string());
+ println!("{:?}", res);
- assert!(matches!(res.err().unwrap(), MerkleError::DBError { .. }));
+ assert!(matches!(
+ res.err().unwrap(),
+ MerkleError::StorageBackendError { .. }
+ ));
}
// Test getting entire tree in string format for JSON RPC

MODIFY
storage/src/persistent/database.rs
storage/src/persistent/database.rs
@@ -12,10 +12,10 @@ use crate::persistent::schema::KeyValueSchema;
#[derive(Serialize, Debug, Clone)]
pub struct RocksDBStats {
- mem_table_total: u64,
- mem_table_unflushed: u64,
- mem_table_readers_total: u64,
- cache_total: u64,
+ pub mem_table_total: u64,
+ pub mem_table_unflushed: u64,
+ pub mem_table_readers_total: u64,
+ pub cache_total: u64,
}
/// Possible errors for schema

MODIFY
storage/src/persistent/mod.rs
storage/src/persistent/mod.rs
// SPDX-License-Identifier: MIT
use failure::Fail;
+use std::ops::Deref;
use std::path::Path;
use std::sync::{Arc, RwLock};
use crate::action_file::ActionFileError;
+use crate::KeyValueStoreBackend;
use crate::StorageError;
use derive_builder::Builder;
use rocksdb::{BlockBasedOptions, Cache, ColumnFamilyDescriptor, Options, DB};
@@ -15,6 +17,10 @@ pub use commit_log::{CommitLogError, CommitLogRef, CommitLogWithSchema, CommitLo
pub use database::{DBError, KeyValueStoreWithSchema};
pub use schema::{CommitLogDescriptor, CommitLogSchema, KeyValueSchema};
+use crate::backend::btree_map::BTreeMapBackend;
+use crate::backend::in_memory_backend::InMemoryBackend;
+use crate::backend::rocksdb_backend::RocksDBBackend;
+use crate::backend::sled_backend::SledBackend;
use crate::merkle_storage::MerkleStorage;
use crate::persistent::sequence::Sequences;
use tezos_context::channel::ContextActionMessage;
@@ -145,7 +151,22 @@ impl PersistentStorage {
db_context: Arc<DB>,
db_context_actions: Arc<DB>,
clog: Arc<CommitLogs>,
+ merkle_backend: KeyValueStoreBackend,
) -> Self {
+ let sled = sled::Config::new().path("/tmp/sled").open().unwrap();
+
+ let merkle = match merkle_backend {
+ KeyValueStoreBackend::RocksDB => MerkleStorage::new(Box::new(RocksDBBackend::new(
+ db_context.clone(),
+ MerkleStorage::name(),
+ ))),
+ KeyValueStoreBackend::InMem => MerkleStorage::new(Box::new(InMemoryBackend::new())),
+ KeyValueStoreBackend::Sled => {
+ MerkleStorage::new(Box::new(SledBackend::new(sled.deref().clone())))
+ }
+ KeyValueStoreBackend::BTreeMap => MerkleStorage::new(Box::new(BTreeMapBackend::new())),
+ };
+
let seq = Arc::new(Sequences::new(db.clone(), 1000));
Self {
clog,
@@ -153,7 +174,7 @@ impl PersistentStorage {
db_context: db_context.clone(),
db_context_actions,
seq,
- merkle: Arc::new(RwLock::new(MerkleStorage::new(db_context))),
+ merkle: Arc::new(RwLock::new(merkle)),
}
}

ADD
storage/src/storage_backend.rs
+use crate::persistent::database::{DBError, RocksDBStats};
+use failure::Fail;
+use serde::Serialize;
+use std::collections::HashSet;
+use std::mem;
+
+use crate::merkle_storage::{ContextValue, EntryHash};
+
+pub fn size_of_vec<T>(v: &Vec<T>) -> usize {
+ mem::size_of::<Vec<T>>() + mem::size_of::<T>() * v.capacity()
+}
+
+#[derive(Debug, Fail)]
+pub enum StorageBackendError {
+ #[fail(display = "RocksDB error: {}", error)]
+ RocksDBError { error: rocksdb::Error },
+ #[fail(display = "Column family {} is missing", name)]
+ MissingColumnFamily { name: &'static str },
+ #[fail(display = "Backend Error")]
+ BackendError,
+ #[fail(display = "SledDB error: {}", error)]
+ SledDBError { error: sled::Error },
+ #[fail(display = "Guard Poison {} ", error)]
+ GuardPoison { error: String },
+ #[fail(display = "Serialization error: {:?}", error)]
+ SerializationError { error: bincode::Error },
+ #[fail(display = "DBError error: {:?}", error)]
+ DBError { error: DBError },
+}
+
+impl From<rocksdb::Error> for StorageBackendError {
+ fn from(error: rocksdb::Error) -> Self {
+ StorageBackendError::RocksDBError { error }
+ }
+}
+
+impl From<sled::Error> for StorageBackendError {
+ fn from(error: sled::Error) -> Self {
+ StorageBackendError::SledDBError { error }
+ }
+}
+
+impl From<DBError> for StorageBackendError {
+ fn from(error: DBError) -> Self {
+ StorageBackendError::DBError { error }
+ }
+}
+
+impl From<bincode::Error> for StorageBackendError {
+ fn from(error: bincode::Error) -> Self {
+ StorageBackendError::SerializationError { error }
+ }
+}
+
+impl slog::Value for StorageBackendError {
+ fn serialize(
+ &self,
+ _record: &slog::Record,
+ key: slog::Key,
+ serializer: &mut dyn slog::Serializer,
+ ) -> slog::Result {
+ serializer.emit_arguments(key, &format_args!("{}", self))
+ }
+}
+
+pub trait StorageBackend: Send + Sync {
+ fn is_persisted(&self) -> bool;
+ fn get(&self, key: &EntryHash) -> Result<Option<ContextValue>, StorageBackendError>;
+ fn put(&mut self, key: EntryHash, value: ContextValue) -> Result<bool, StorageBackendError>;
+ fn merge(&mut self, key: EntryHash, value: ContextValue) -> Result<(), StorageBackendError>;
+ fn delete(&mut self, key: &EntryHash) -> Result<Option<ContextValue>, StorageBackendError>;
+ fn contains(&self, key: &EntryHash) -> Result<bool, StorageBackendError>;
+ fn get_mem_use_stats(&self) -> Result<RocksDBStats, StorageBackendError>;
+}
+
+#[derive(Debug, Default, Clone, Copy, Serialize)]
+pub struct StorageBackendStats {
+ pub key_bytes: usize,
+ pub value_bytes: usize,
+ pub reused_keys_bytes: usize,
+}
+
+impl StorageBackendStats {
+ /// increases `reused_keys_bytes` based on `key`
+ pub fn update_reused_keys(&mut self, list: &HashSet<EntryHash>) {
+ // TODO: bring back when MerklHash aka EntryHash will be allocated
+ // on stack
+ // self.reused_keys_bytes = list.capacity() * mem::size_of::<EntryHash>();
+ self.reused_keys_bytes = list.capacity() * 32;
+ }
+
+ pub fn total_as_bytes(&self) -> usize {
+ self.key_bytes + self.value_bytes + self.reused_keys_bytes
+ }
+}
+
+impl<'a> std::ops::Add<&'a Self> for StorageBackendStats {
+ type Output = Self;
+
+ fn add(self, other: &'a Self) -> Self::Output {
+ Self {
+ key_bytes: self.key_bytes + other.key_bytes,
+ value_bytes: self.value_bytes + other.value_bytes,
+ reused_keys_bytes: self.reused_keys_bytes + other.reused_keys_bytes,
+ }
+ }
+}
+
+impl std::ops::Add for StorageBackendStats {
+ type Output = Self;
+
+ fn add(self, other: Self) -> Self::Output {
+ self + &other
+ }
+}
+
+impl<'a> std::ops::AddAssign<&'a Self> for StorageBackendStats {
+ fn add_assign(&mut self, other: &'a Self) {
+ *self = *self + other;
+ }
+}
+
+impl std::ops::AddAssign for StorageBackendStats {
+ fn add_assign(&mut self, other: Self) {
+ *self = *self + other;
+ }
+}
+
+impl<'a> std::ops::Sub<&'a Self> for StorageBackendStats {
+ type Output = Self;
+
+ fn sub(self, other: &'a Self) -> Self::Output {
+ Self {
+ key_bytes: self.key_bytes - other.key_bytes,
+ value_bytes: self.value_bytes - other.value_bytes,
+ reused_keys_bytes: self.reused_keys_bytes - other.reused_keys_bytes,
+ }
+ }
+}
+
+impl std::ops::Sub for StorageBackendStats {
+ type Output = Self;
+
+ fn sub(self, other: Self) -> Self::Output {
+ self - &other
+ }
+}
+
+impl<'a> std::ops::SubAssign<&'a Self> for StorageBackendStats {
+ fn sub_assign(&mut self, other: &'a Self) {
+ *self = *self - other;
+ }
+}
+
+impl std::ops::SubAssign for StorageBackendStats {
+ fn sub_assign(&mut self, other: Self) {
+ *self = *self - other;
+ }
+}
+
+impl<'a> std::iter::Sum<&'a StorageBackendStats> for StorageBackendStats {
+ fn sum<I: Iterator<Item = &'a Self>>(iter: I) -> Self {
+ iter.fold(StorageBackendStats::default(), |acc, cur| acc + cur)
+ }
+}
+
+impl From<(&EntryHash, &ContextValue)> for StorageBackendStats {
+ fn from((entry_hash, value): (&EntryHash, &ContextValue)) -> Self {
+ StorageBackendStats {
+ // TODO: bring back when MerklHash aka EntryHash will be allocated
+ // on stack
+ // key_bytes: mem::size_of::<EntryHash>(),
+ key_bytes: entry_hash.as_ref().len(),
+ value_bytes: size_of_vec(&value),
+ reused_keys_bytes: 0,
+ }
+ }
+}

support for different key value backends