dedicated database instances for context and context actions
MODIFY
light_node/src/configuration.rs
light_node/src/configuration.rs
// Copyright (c) SimpleStaking and Tezedge Contributors
// SPDX-License-Identifier: MIT
-use std::collections::HashMap;
use std::env;
use std::ffi::OsString;
use std::fs;
@@ -9,17 +8,27 @@ use std::io::{self, BufRead};
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::time::Duration;
+use std::{collections::HashMap, fmt::Debug};
use clap::{App, Arg};
+use rocksdb::ColumnFamilyDescriptor;
use shell::peer_manager::P2p;
use shell::PeerConnectionThreshold;
-use storage::persistent::{DbConfiguration, DbConfigurationBuilder};
+use storage;
+use storage::persistent::KeyValueSchema;
use tezos_api::environment;
use tezos_api::environment::TezosEnvironment;
use tezos_api::ffi::PatchContext;
use tezos_wrapper::TezosApiConnectionPoolConfiguration;
+const STORAGES_COUNT: usize = 3;
+const MINIMAL_THREAD_COUNT: usize = 1;
+const DB_STORAGE_VERSION: i64 = 16;
+const DB_CONTEXT_STORAGE_VERSION: i64 = 16;
+const DB_CONTEXT_ACTIONS_STORAGE_VERSION: i64 = 16;
+const LRU_CACHE_SIZE_32MB: usize = 32 * 1024 * 1024;
+
#[derive(Debug, Clone)]
pub struct Rpc {
pub listener_port: u16,
@@ -34,9 +43,71 @@ pub struct Logging {
pub file: Option<PathBuf>,
}
+pub trait ColumnFactory {
+ fn create(&self, cache: &rocksdb::Cache) -> Vec<ColumnFamilyDescriptor>;
+}
+#[derive(Debug, Clone)]
+pub struct DBTableInitializer {}
+
+#[derive(Debug, Clone)]
+pub struct ContextTableInitializer {}
+
+#[derive(Debug, Clone)]
+pub struct ContextActionsTableInitializer {}
+
+impl ColumnFactory for DBTableInitializer {
+ fn create(&self, cache: &rocksdb::Cache) -> Vec<ColumnFamilyDescriptor> {
+ vec![
+ storage::block_storage::BlockPrimaryIndex::descriptor(cache),
+ storage::block_storage::BlockByLevelIndex::descriptor(cache),
+ storage::block_storage::BlockByContextHashIndex::descriptor(cache),
+ storage::BlockMetaStorage::descriptor(cache),
+ storage::OperationsStorage::descriptor(cache),
+ storage::OperationsMetaStorage::descriptor(cache),
+ storage::SystemStorage::descriptor(cache),
+ storage::persistent::sequence::Sequences::descriptor(cache),
+ storage::MempoolStorage::descriptor(cache),
+ storage::ChainMetaStorage::descriptor(cache),
+ storage::PredecessorStorage::descriptor(cache),
+ ]
+ }
+}
+
+impl ColumnFactory for ContextTableInitializer {
+ fn create(&self, cache: &rocksdb::Cache) -> Vec<ColumnFamilyDescriptor> {
+ vec![
+ storage::SystemStorage::descriptor(cache),
+ storage::merkle_storage::MerkleStorage::descriptor(cache),
+ ]
+ }
+}
+
+impl ColumnFactory for ContextActionsTableInitializer {
+ fn create(&self, cache: &rocksdb::Cache) -> Vec<ColumnFamilyDescriptor> {
+ vec![
+ storage::SystemStorage::descriptor(cache),
+ storage::context_action_storage::ContextActionByBlockHashIndex::descriptor(cache),
+ storage::context_action_storage::ContextActionByContractIndex::descriptor(cache),
+ storage::context_action_storage::ContextActionByTypeIndex::descriptor(cache),
+ storage::context_action_storage::ContextActionStorage::descriptor(cache),
+ ]
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct RocksDBConfig<C: ColumnFactory> {
+ pub cache_size: usize,
+ pub expected_db_version: i64,
+ pub db_path: PathBuf,
+ pub columns: C,
+ pub threads: Option<usize>,
+}
+
#[derive(Debug, Clone)]
pub struct Storage {
- pub db_cfg: DbConfiguration,
+ pub db: RocksDBConfig<DBTableInitializer>,
+ pub db_context: RocksDBConfig<ContextTableInitializer>,
+ pub db_context_actions: RocksDBConfig<ContextActionsTableInitializer>,
pub db_path: PathBuf,
pub tezos_data_dir: PathBuf,
pub store_context_actions: bool,
@@ -689,28 +760,48 @@ impl Environment {
}
},
},
- storage: crate::configuration::Storage {
- tezos_data_dir: data_dir.clone(),
- db_cfg: {
- let mut db_cfg = DbConfigurationBuilder::default();
-
- if let Some(value) = args.value_of("db-cfg-max-threads") {
- let max_treads = value
- .parse::<usize>()
- .expect("Provided value cannot be converted to number");
- db_cfg.max_threads(Some(max_treads));
- }
-
- db_cfg.build().unwrap()
- },
- db_path: {
- let db_path = args
+ storage: {
+ let path = args
.value_of("bootstrap-db-path")
.unwrap_or("")
.parse::<PathBuf>()
.expect("Provided value cannot be converted to path");
- get_final_path(&data_dir, db_path)
- },
+ let db_path = get_final_path(&data_dir, path);
+
+ let threads_count = args.value_of("db-cfg-max-threads").map(|value| {
+ value
+ .parse::<usize>()
+ .map(|val| std::cmp::min(MINIMAL_THREAD_COUNT, val / STORAGES_COUNT))
+ .expect("Provided value cannot be converted to number")
+ });
+
+ let db = RocksDBConfig {
+ cache_size: LRU_CACHE_SIZE_32MB,
+ expected_db_version: DB_STORAGE_VERSION,
+ db_path: db_path.clone().join("db"),
+ columns: DBTableInitializer {},
+ threads: threads_count,
+ };
+ let db_context = RocksDBConfig {
+ cache_size: LRU_CACHE_SIZE_32MB,
+ expected_db_version: DB_CONTEXT_STORAGE_VERSION,
+ db_path: db_path.join("context"),
+ columns: ContextTableInitializer {},
+ threads: threads_count,
+ };
+ let db_context_actions = RocksDBConfig {
+ cache_size: LRU_CACHE_SIZE_32MB,
+ expected_db_version: DB_CONTEXT_ACTIONS_STORAGE_VERSION,
+ db_path: db_path.join("context_actions"),
+ columns: ContextActionsTableInitializer {},
+ threads: threads_count,
+ };
+ crate::configuration::Storage {
+ tezos_data_dir: data_dir.clone(),
+ db,
+ db_context,
+ db_context_actions,
+ db_path: db_path,
store_context_actions: args
.value_of("store-context-actions")
.unwrap_or("true")
@@ -728,8 +819,9 @@ impl Environment {
// validate valid json
if let Err(e) = serde_json::from_str::<
HashMap<String, serde_json::Value>,
- >(&content)
- {
+ >(
+ &content
+ ) {
panic!(
"Invalid json file: {}, reason: {}",
path.as_path().display().to_string(),
@@ -756,6 +848,7 @@ impl Environment {
}
}
},
+ }
},
identity: crate::configuration::Identity {
identity_json_file_path: {

MODIFY
light_node/src/main.rs
light_node/src/main.rs
@@ -6,8 +6,9 @@ use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
+use configuration::{ColumnFactory, RocksDBConfig};
use riker::actors::*;
-use rocksdb::Cache;
+use rocksdb::{Cache, DB};
use slog::{crit, debug, error, info, o, warn, Drain, Logger};
use logging::detailed_json;
@@ -23,15 +24,11 @@ use shell::mempool::init_mempool_state_storage;
use shell::mempool::mempool_prevalidator::MempoolPrevalidator;
use shell::peer_manager::PeerManager;
use shell::shell_channel::{ShellChannel, ShellChannelTopic, ShuttingDown};
-use storage::context::TezedgeContext;
-use storage::merkle_storage::MerkleStorage;
-use storage::persistent::sequence::Sequences;
-use storage::persistent::{open_cl, open_kv, CommitLogSchema, KeyValueSchema, PersistentStorage};
+use storage::persistent::DbConfiguration;
+use storage::persistent::{open_cl, open_kv, CommitLogSchema, PersistentStorage};
use storage::{
- block_storage, check_database_compatibility, context_action_storage,
- resolve_storage_init_chain_data, BlockMetaStorage, BlockStorage, ChainMetaStorage,
- ContextActionStorage, MempoolStorage, OperationsMetaStorage, OperationsStorage,
- PredecessorStorage, StorageInitInfo, SystemStorage,
+ check_database_compatibility, context::TezedgeContext, persistent::DBError,
+ resolve_storage_init_chain_data, BlockStorage, StorageInitInfo,
};
use tezos_api::environment;
use tezos_api::environment::TezosEnvironmentConfiguration;
@@ -467,6 +464,36 @@ fn block_on_actors(
});
}
+fn initialize_db<Factory: ColumnFactory>(
+ log: &Logger,
+ cache: &Cache,
+ config: &RocksDBConfig<Factory>,
+ env: &TezosEnvironmentConfiguration,
+) -> Result<Arc<DB>, DBError> {
+ let columns: Vec<_> = config.columns.create(cache);
+ let db = open_kv(
+ &config.db_path,
+ columns,
+ &DbConfiguration {
+ max_threads: config.threads,
+ },
+ )
+ .map(Arc::new)?;
+
+ match check_database_compatibility(db.clone(), config.expected_db_version, &env, &log) {
+ Ok(false) => Err(DBError::DatabaseIncompatibility {
+ name: format!(
+ "Database is incompatible with version {}",
+ config.expected_db_version
+ ),
+ }),
+ Err(e) => Err(DBError::DatabaseIncompatibility {
+ name: format!("Failed to verify database compatibility reason: '{}'", e),
+ }),
+ _ => Ok(db),
+ }
+}
+
fn main() {
// Parses config + cli args
let env = crate::configuration::Environment::from_args();
@@ -516,60 +543,49 @@ fn main() {
// create common RocksDB block cache to be shared among column families
// IMPORTANT: Cache object must live at least as long as DB (returned by open_kv)
- let cache = Cache::new_lru_cache(128 * 1024 * 1024).unwrap(); // 128 MB
-
- let schemas = vec![
- block_storage::BlockPrimaryIndex::descriptor(&cache),
- block_storage::BlockByLevelIndex::descriptor(&cache),
- block_storage::BlockByContextHashIndex::descriptor(&cache),
- BlockMetaStorage::descriptor(&cache),
- OperationsStorage::descriptor(&cache),
- OperationsMetaStorage::descriptor(&cache),
- context_action_storage::ContextActionByBlockHashIndex::descriptor(&cache),
- context_action_storage::ContextActionByContractIndex::descriptor(&cache),
- context_action_storage::ContextActionByTypeIndex::descriptor(&cache),
- ContextActionStorage::descriptor(&cache),
- MerkleStorage::descriptor(&cache),
- SystemStorage::descriptor(&cache),
- Sequences::descriptor(&cache),
- MempoolStorage::descriptor(&cache),
- ChainMetaStorage::descriptor(&cache),
- PredecessorStorage::descriptor(&cache),
+ let cache = [
+ Cache::new_lru_cache(env.storage.db.cache_size).unwrap(),
+ Cache::new_lru_cache(env.storage.db_context.cache_size).unwrap(),
+ Cache::new_lru_cache(env.storage.db_context_actions.cache_size).unwrap(),
];
- let rocks_db = match open_kv(&env.storage.db_path, schemas, &env.storage.db_cfg) {
- Ok(db) => Arc::new(db),
- Err(e) => shutdown_and_exit!(
- error!(log, "Failed to create RocksDB database at '{:?}'", &env.storage.db_path; "reason" => e),
- actor_system
- ),
- };
- debug!(log, "Loaded RocksDB database");
+ let storages: Result<Vec<Arc<DB>>, DBError> = vec![
+ initialize_db(&log, &cache[0], &env.storage.db, &tezos_env),
+ initialize_db(&log, &cache[1], &env.storage.db_context, &tezos_env),
+ initialize_db(&log, &cache[2], &env.storage.db_context_actions, &tezos_env),
+ ]
+ .into_iter()
+ .collect();
- match check_database_compatibility(rocks_db.clone(), DATABASE_VERSION, &tezos_env, &log) {
- Ok(false) => shutdown_and_exit!(
- crit!(log, "Database incompatibility detected"),
- actor_system
- ),
+ let storages = match storages {
Err(e) => shutdown_and_exit!(
- error!(log, "Failed to verify database compatibility"; "reason" => e),
+ crit!(
+ log,
+ "Failed to create initialize RocksDB databases '{:?}'",
+ e
+ ),
actor_system
),
- _ => (),
- }
-
- let schemas = vec![BlockStorage::descriptor()];
+ Ok(dbs) => dbs,
+ };
{
- let commit_logs = match open_cl(&env.storage.db_path, schemas) {
+ let commit_logs = match open_cl(&env.storage.db_path, vec![BlockStorage::descriptor()]) {
Ok(commit_logs) => Arc::new(commit_logs),
Err(e) => shutdown_and_exit!(
error!(log, "Failed to open commit logs"; "reason" => e),
actor_system
),
};
-
- let persistent_storage = PersistentStorage::new(rocks_db, commit_logs);
+ if let [kv, kv_context, kv_actions] = &storages[..] {
+ debug!(log, "Loaded RocksDB databases");
+
+ let persistent_storage = PersistentStorage::new(
+ kv.clone(),
+ kv_context.clone(),
+ kv_actions.clone(),
+ commit_logs,
+ );
let tezedge_context = TezedgeContext::new(
BlockStorage::new(&persistent_storage),
persistent_storage.merkle(),
@@ -596,5 +612,7 @@ fn main() {
actor_system
),
}
+ } else {
+ }
}
}

MODIFY
storage/src/block_meta_storage.rs
storage/src/block_meta_storage.rs
@@ -12,13 +12,13 @@ use slog::{warn, Logger};
use crypto::hash::{BlockHash, ChainId, HashType};
use tezos_messages::p2p::encoding::block_header::Level;
-use crate::num_from_slice;
use crate::persistent::database::{IteratorMode, IteratorWithSchema};
use crate::persistent::{
default_table_options, Decoder, Encoder, KeyValueSchema, KeyValueStoreWithSchema,
PersistentStorage, SchemaError,
};
use crate::predecessor_storage::{PredecessorKey, PredecessorStorage};
+use crate::{num_from_slice, persistent::StorageType};
use crate::{BlockHeaderWithHash, StorageError};
pub type BlockMetaStorageKV = dyn KeyValueStoreWithSchema<BlockMetaStorage> + Sync + Send;
@@ -52,7 +52,7 @@ impl BlockMetaStorage {
pub fn new(persistent_storage: &PersistentStorage) -> Self {
BlockMetaStorage {
- kv: persistent_storage.kv(),
+ kv: persistent_storage.kv(StorageType::Database),
predecessors_index: PredecessorStorage::new(persistent_storage),
}
}

MODIFY
storage/src/block_storage.rs
storage/src/block_storage.rs
@@ -13,7 +13,7 @@ use crypto::hash::{
use crate::persistent::{
BincodeEncoded, CommitLogSchema, CommitLogWithSchema, KeyValueSchema, KeyValueStoreWithSchema,
- Location, PersistentStorage,
+ Location, PersistentStorage, StorageType,
};
use crate::{BlockHeaderWithHash, Direction, IteratorMode, StorageError};
@@ -125,9 +125,11 @@ pub trait BlockStorageReader: Sync + Send {
impl BlockStorage {
pub fn new(persistent_storage: &PersistentStorage) -> Self {
Self {
- primary_index: BlockPrimaryIndex::new(persistent_storage.kv()),
- by_level_index: BlockByLevelIndex::new(persistent_storage.kv()),
- by_context_hash_index: BlockByContextHashIndex::new(persistent_storage.kv()),
+ primary_index: BlockPrimaryIndex::new(persistent_storage.kv(StorageType::Database)),
+ by_level_index: BlockByLevelIndex::new(persistent_storage.kv(StorageType::Database)),
+ by_context_hash_index: BlockByContextHashIndex::new(
+ persistent_storage.kv(StorageType::Database),
+ ),
clog: persistent_storage.clog(),
}
}

MODIFY
storage/src/chain_meta_storage.rs
storage/src/chain_meta_storage.rs
@@ -13,7 +13,7 @@ use tezos_messages::Head;
use crate::persistent::{
default_table_options, BincodeEncoded, Decoder, Encoder, KeyValueSchema,
- KeyValueStoreWithSchema, PersistentStorage, SchemaError,
+ KeyValueStoreWithSchema, PersistentStorage, SchemaError, StorageType,
};
use crate::StorageError;
@@ -59,7 +59,7 @@ pub struct ChainMetaStorage {
impl ChainMetaStorage {
pub fn new(persistent_storage: &PersistentStorage) -> Self {
Self {
- kv: persistent_storage.kv(),
+ kv: persistent_storage.kv(StorageType::Database),
}
}

MODIFY
storage/src/context_action_storage.rs
storage/src/context_action_storage.rs
@@ -18,7 +18,6 @@ use crypto::hash::{
use tezos_context::channel::ContextAction;
use tezos_messages::base::signature_public_key_hash::{ConversionError, SignaturePublicKeyHash};
-use crate::num_from_slice;
use crate::persistent::codec::range_from_idx_len;
use crate::persistent::sequence::{SequenceGenerator, SequenceNumber};
use crate::persistent::{
@@ -26,6 +25,7 @@ use crate::persistent::{
KeyValueStoreWithSchema, PersistentStorage, SchemaError,
};
use crate::StorageError;
+use crate::{num_from_slice, persistent::StorageType};
pub enum ContextHashType {
Block,
@@ -76,12 +76,13 @@ pub struct ContextActionStorage {
impl ContextActionStorage {
pub fn new(persistent_storage: &PersistentStorage) -> Self {
+ let storage = persistent_storage.kv(StorageType::ContextAction);
Self {
- kv: persistent_storage.kv(),
+ kv: persistent_storage.kv(StorageType::ContextAction),
generator: persistent_storage.seq().generator(Self::name()),
- context_by_block_index: ContextActionByBlockHashIndex::new(persistent_storage.kv()),
- context_by_contract_index: ContextActionByContractIndex::new(persistent_storage.kv()),
- context_by_type_index: ContextActionByTypeIndex::new(persistent_storage.kv()),
+ context_by_block_index: ContextActionByBlockHashIndex::new(storage.clone()),
+ context_by_contract_index: ContextActionByContractIndex::new(storage.clone()),
+ context_by_type_index: ContextActionByTypeIndex::new(storage),
}
}

MODIFY
storage/src/lib.rs
storage/src/lib.rs
@@ -493,7 +493,7 @@ pub mod tests_common {
let cache = Cache::new_lru_cache(128 * 1024 * 1024)?; // 128 MB
let kv = open_kv(
- &path,
+ path.join("db"),
vec![
block_storage::BlockPrimaryIndex::descriptor(&cache),
block_storage::BlockByLevelIndex::descriptor(&cache),
@@ -501,26 +501,43 @@ pub mod tests_common {
BlockMetaStorage::descriptor(&cache),
OperationsStorage::descriptor(&cache),
OperationsMetaStorage::descriptor(&cache),
- context_action_storage::ContextActionByBlockHashIndex::descriptor(&cache),
- context_action_storage::ContextActionByContractIndex::descriptor(&cache),
- context_action_storage::ContextActionByTypeIndex::descriptor(&cache),
- MerkleStorage::descriptor(&cache),
SystemStorage::descriptor(&cache),
Sequences::descriptor(&cache),
DatabaseBackedSkipList::descriptor(&cache),
Lane::descriptor(&cache),
ListValue::descriptor(&cache),
MempoolStorage::descriptor(&cache),
- ContextActionStorage::descriptor(&cache),
ChainMetaStorage::descriptor(&cache),
PredecessorStorage::descriptor(&cache),
],
&cfg,
)?;
+
+ let kv_context = open_kv(
+ path.join("context"),
+ vec![MerkleStorage::descriptor(&cache)],
+ &cfg,
+ )?;
+
+ let kv_context_action = open_kv(
+ path.join("context_actions"),
+ vec![
+ ContextActionStorage::descriptor(&cache),
+ context_action_storage::ContextActionByBlockHashIndex::descriptor(&cache),
+ context_action_storage::ContextActionByContractIndex::descriptor(&cache),
+ context_action_storage::ContextActionByTypeIndex::descriptor(&cache),
+ ],
+ &cfg,
+ )?;
let clog = open_cl(&path, vec![BlockStorage::descriptor()])?;
Ok(Self {
- persistent_storage: PersistentStorage::new(Arc::new(kv), Arc::new(clog)),
+ persistent_storage: PersistentStorage::new(
+ Arc::new(kv),
+ Arc::new(kv_context),
+ Arc::new(kv_context_action),
+ Arc::new(clog),
+ ),
path,
remove_on_destroy,
})

MODIFY
storage/src/mempool_storage.rs
storage/src/mempool_storage.rs
@@ -15,7 +15,7 @@ use tezos_messages::p2p::encoding::operation::OperationMessage;
use crate::persistent::{
BincodeEncoded, Decoder, Encoder, KeyValueSchema, KeyValueStoreWithSchema, PersistentStorage,
- SchemaError,
+ SchemaError, StorageType,
};
use crate::{num_from_slice, IteratorMode, StorageError};
@@ -65,7 +65,7 @@ pub struct MempoolStorage {
impl MempoolStorage {
pub fn new(persistent_storage: &PersistentStorage) -> Self {
Self {
- kv: persistent_storage.kv(),
+ kv: persistent_storage.kv(StorageType::Database),
}
}

MODIFY
storage/src/operations_meta_storage.rs
storage/src/operations_meta_storage.rs
@@ -10,12 +10,12 @@ use rocksdb::{Cache, ColumnFamilyDescriptor, MergeOperands};
use crypto::hash::{BlockHash, ChainId, HashType};
use tezos_messages::p2p::encoding::prelude::*;
-use crate::num_from_slice;
use crate::persistent::database::{IteratorMode, IteratorWithSchema};
use crate::persistent::{
default_table_options, Decoder, Encoder, KeyValueSchema, KeyValueStoreWithSchema,
PersistentStorage, SchemaError,
};
+use crate::{num_from_slice, persistent::StorageType};
use crate::{BlockHeaderWithHash, StorageError};
/// Convenience type for operation meta storage database
@@ -30,7 +30,7 @@ pub struct OperationsMetaStorage {
impl OperationsMetaStorage {
pub fn new(persistent_storage: &PersistentStorage) -> Self {
Self {
- kv: persistent_storage.kv(),
+ kv: persistent_storage.kv(StorageType::Database),
}
}

MODIFY
storage/src/operations_storage.rs
storage/src/operations_storage.rs
@@ -12,7 +12,7 @@ use tezos_messages::p2p::encoding::prelude::*;
use crate::persistent::{
default_table_options, Decoder, Encoder, KeyValueSchema, KeyValueStoreWithSchema,
- PersistentStorage, SchemaError,
+ PersistentStorage, SchemaError, StorageType,
};
use crate::StorageError;
@@ -35,7 +35,7 @@ pub struct OperationsStorage {
impl OperationsStorage {
pub fn new(persistent_storage: &PersistentStorage) -> Self {
Self {
- kv: persistent_storage.kv(),
+ kv: persistent_storage.kv(StorageType::Database),
}
}

MODIFY
storage/src/persistent/database.rs
storage/src/persistent/database.rs
@@ -27,6 +27,8 @@ pub enum DBError {
RocksDBError { error: Error },
#[fail(display = "Column family {} is missing", name)]
MissingColumnFamily { name: &'static str },
+ #[fail(display = "Database incompatibility {}", name)]
+ DatabaseIncompatibility { name: String },
}
impl From<SchemaError> for DBError {

MODIFY
storage/src/persistent/mod.rs
storage/src/persistent/mod.rs
@@ -26,7 +26,7 @@ pub mod sequence;
#[derive(Builder, Debug, Clone)]
pub struct DbConfiguration {
#[builder(default = "None")]
- max_threads: Option<usize>,
+ pub max_threads: Option<usize>,
}
impl Default for DbConfiguration {
@@ -116,7 +116,12 @@ where
#[derive(Clone)]
pub struct PersistentStorage {
/// key-value store
+ //TODO to be renamed to kv_db
kv: Arc<DB>,
+ /// context store
+ kv_context: Arc<DB>,
+ /// context actions store
+ kv_context_actions: Arc<DB>,
/// commit log store
clog: Arc<CommitLogs>,
/// autoincrement id generators
@@ -125,20 +130,37 @@ pub struct PersistentStorage {
merkle: Arc<RwLock<MerkleStorage>>,
}
+pub enum StorageType {
+ Database,
+ Context,
+ ContextAction,
+}
+
impl PersistentStorage {
- pub fn new(kv: Arc<DB>, clog: Arc<CommitLogs>) -> Self {
+ pub fn new(
+ kv: Arc<DB>,
+ kv_context: Arc<DB>,
+ kv_context_actions: Arc<DB>,
+ clog: Arc<CommitLogs>,
+ ) -> Self {
let seq = Arc::new(Sequences::new(kv.clone(), 1000));
Self {
clog,
- kv: kv.clone(),
+ kv,
+ kv_context: kv_context.clone(),
+ kv_context_actions,
seq,
- merkle: Arc::new(RwLock::new(MerkleStorage::new(kv))),
+ merkle: Arc::new(RwLock::new(MerkleStorage::new(kv_context))),
}
}
#[inline]
- pub fn kv(&self) -> Arc<DB> {
- self.kv.clone()
+ pub fn kv(&self, storage: StorageType) -> Arc<DB> {
+ match storage {
+ StorageType::Context => self.kv_context.clone(),
+ StorageType::ContextAction => self.kv_context_actions.clone(),
+ StorageType::Database => self.kv.clone(),
+ }
}
#[inline]

MODIFY
storage/src/predecessor_storage.rs
storage/src/predecessor_storage.rs
@@ -8,13 +8,13 @@ use serde::{Deserialize, Serialize};
use crypto::hash::BlockHash;
-use crate::block_meta_storage::Meta;
use crate::persistent::database::{IteratorMode, IteratorWithSchema};
use crate::persistent::{
default_table_options, BincodeEncoded, KeyValueSchema, KeyValueStoreWithSchema,
PersistentStorage,
};
use crate::StorageError;
+use crate::{block_meta_storage::Meta, persistent::StorageType};
pub type PredecessorsIndexStorageKV = dyn KeyValueStoreWithSchema<PredecessorStorage> + Sync + Send;
@@ -41,7 +41,7 @@ pub struct PredecessorStorage {
impl PredecessorStorage {
pub fn new(persistent_storage: &PersistentStorage) -> Self {
Self {
- kv: persistent_storage.kv(),
+ kv: persistent_storage.kv(StorageType::Database),
}
}

MODIFY
storage/tests/context_action_storage.rs
storage/tests/context_action_storage.rs
@@ -71,6 +71,11 @@ fn context_get_values_by_block_hash() -> Result<(), Error> {
storage.put_action(&block_hash_2, value_2_0)?;
storage.put_action(&block_hash_1, value_1_1)?;
storage.put_action(&block_hash_2, value_2_1)?;
+ tmp_storage
+ .storage()
+ .kv(persistent::StorageType::Context)
+ .flush()
+ .unwrap();
// block hash 1
let values = storage.get_by_block_hash(&block_hash_1)?;

MODIFY
storage/tests/lane.rs
storage/tests/lane.rs
use maplit::hashmap;
-use storage::skip_list::{Lane, SkipListError, TryExtend, TypedLane};
use storage::tests_common::TmpStorage;
+use storage::{
+ persistent::StorageType,
+ skip_list::{Lane, SkipListError, TryExtend, TypedLane},
+};
#[test]
fn lane_new() {
@@ -12,8 +15,8 @@ fn lane_new() {
let lane = Lane::new(
1,
0,
- tmp_storage.storage().kv(),
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage.storage().seq().generator("__lane:lane_new"),
);
assert_eq!(lane.level(), 0);
@@ -25,8 +28,8 @@ fn lane_higher() {
let lane = Lane::new(
2,
0,
- tmp_storage.storage().kv(),
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage.storage().seq().generator("__lane:lane_higher"),
);
let higher_lane = lane.higher_lane();
@@ -39,8 +42,8 @@ fn lane_lower() {
let lane = Lane::new(
3,
1,
- tmp_storage.storage().kv(),
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage.storage().seq().generator("__lane:lane_lower"),
);
let lower_lane = lane.lower_lane();
@@ -53,8 +56,8 @@ fn lane_lower_underflow() {
let lane = Lane::new(
4,
0,
- tmp_storage.storage().kv(),
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()
@@ -70,8 +73,8 @@ fn lane_put_get_values() {
let mut lane = Lane::new(
5,
0,
- tmp_storage.storage().kv(),
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()

MODIFY
storage/tests/skip_list.rs
storage/tests/skip_list.rs
@@ -11,7 +11,7 @@ use rand::{
};
use serde::{Deserialize, Serialize};
-use storage::persistent::BincodeEncoded;
+use storage::persistent::{BincodeEncoded, StorageType};
use storage::skip_list::{DatabaseBackedSkipList, TypedSkipList};
use storage::tests_common::TmpStorage;
@@ -41,7 +41,7 @@ fn list_new() {
let list: Box<dyn TypedSkipList<i32, i32>> = Box::new(
DatabaseBackedSkipList::new(
1,
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()
@@ -58,7 +58,7 @@ fn list_push() {
let mut list: Box<dyn TypedSkipList<i32, i32>> = Box::new(
DatabaseBackedSkipList::new(
2,
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()
@@ -88,7 +88,7 @@ fn list_check_first() {
let mut list: Box<dyn TypedSkipList<i32, i32>> = Box::new(
DatabaseBackedSkipList::new(
3,
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()
@@ -114,7 +114,7 @@ fn list_check_second() {
let mut list: Box<dyn TypedSkipList<i32, i32>> = Box::new(
DatabaseBackedSkipList::new(
4,
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()
@@ -143,7 +143,7 @@ fn list_check_bottom_lane() {
let mut list: Box<dyn TypedSkipList<i32, i32>> = Box::new(
DatabaseBackedSkipList::new(
5,
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()
@@ -173,7 +173,7 @@ pub fn list_check_faster_lane() {
let mut list: Box<dyn TypedSkipList<i32, i32>> = Box::new(
DatabaseBackedSkipList::new(
6,
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()
@@ -203,7 +203,7 @@ pub fn list_check_lane_traversal() {
let mut list: Box<dyn TypedSkipList<i32, i32>> = Box::new(
DatabaseBackedSkipList::new(
7,
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()
@@ -233,7 +233,7 @@ pub fn list_get_value_by_key() {
let mut list: Box<dyn TypedSkipList<i32, i32>> = Box::new(
DatabaseBackedSkipList::new(
8,
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()
@@ -265,7 +265,7 @@ pub fn list_get_values_by_prefix() -> Result<(), failure::Error> {
let mut list: Box<dyn TypedSkipList<String, i32>> = Box::new(
DatabaseBackedSkipList::new(
9,
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()
@@ -323,7 +323,7 @@ pub fn list_check_lane_order_traversal() {
let mut list: Box<dyn TypedSkipList<i32, i32>> = Box::new(
DatabaseBackedSkipList::new(
8,
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()
@@ -358,7 +358,7 @@ pub fn list_check_get_key() {
let mut list: Box<dyn TypedSkipList<i32, i32>> = Box::new(
DatabaseBackedSkipList::new(
8,
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()
@@ -386,7 +386,7 @@ pub fn skip_list_simulate_ledger() {
let list: Box<dyn TypedSkipList<u64, Operation>> = Box::new(
DatabaseBackedSkipList::new(
8,
- tmp_storage.storage().kv(),
+ tmp_storage.storage().kv(StorageType::Database),
tmp_storage
.storage()
.seq()

[TE-280] dedicated database instances for context and context actions

Signed-off-by: Mateusz Nowakowski <mateusz.nowakowski@simplestaking.com>