make use of atomic rocksdb write_batch functionality
MODIFY
storage/src/backend/btree_map.rs
storage/src/backend/btree_map.rs
@@ -43,6 +43,13 @@ impl KVStoreTrait for KVStore<EntryHash, ContextValue> {
}
}
+ fn put_batch(&mut self, batch: Vec<(EntryHash, ContextValue)>) -> Result<(), KVStoreError> {
+ for (k, v) in batch.into_iter() {
+ self.put(&k, v)?;
+ }
+ Ok(())
+ }
+
fn merge(&mut self, key: &EntryHash, value: ContextValue) -> Result<(), KVStoreError> {
self.kv_map.insert(key.clone(), value);
Ok(())

MODIFY
storage/src/backend/in_memory_backend.rs
storage/src/backend/in_memory_backend.rs
@@ -46,6 +46,16 @@ impl StorageBackend for InMemoryBackend {
Ok(was_added)
}
+ fn put_batch(
+ &mut self,
+ batch: Vec<(EntryHash, ContextValue)>,
+ ) -> Result<(), StorageBackendError> {
+ for (k, v) in batch.into_iter() {
+ self.put(&k, v)?;
+ }
+ Ok(())
+ }
+
fn merge(&mut self, key: &EntryHash, value: ContextValue) -> Result<(), StorageBackendError> {
let mut w = self
.inner

MODIFY
storage/src/backend/rocksdb_backend.rs
storage/src/backend/rocksdb_backend.rs
use crate::merkle_storage::{ContextValue, EntryHash};
use crate::persistent::database::GetInMemStats;
+use crate::persistent::database::KeyValueStoreWithSchema;
use crate::persistent::database::RocksDBStats;
use crate::storage_backend::{StorageBackend, StorageBackendError};
+use crate::MerkleStorage;
+use rocksdb::WriteBatch;
use rocksdb::{WriteOptions, DB};
use serde::{Deserialize, Serialize};
+use std::ops::Deref;
use std::sync::Arc;
pub struct RocksDBBackend {
@@ -62,6 +66,23 @@ impl StorageBackend for RocksDBBackend {
.map(|_| true)
}
+ fn put_batch(
+ &mut self,
+ batch: Vec<(EntryHash, ContextValue)>,
+ ) -> Result<(), StorageBackendError> {
+ let mut rocksb_batch = WriteBatch::default(); // batch containing DB key values to persist
+
+ for (k, v) in batch.iter() {
+ (self.inner.deref() as &dyn KeyValueStoreWithSchema<MerkleStorage>).put_batch(
+ &mut rocksb_batch,
+ &k,
+ &v,
+ )?;
+ }
+
+ Ok(())
+ }
+
fn merge(&mut self, key: &EntryHash, value: ContextValue) -> Result<(), StorageBackendError> {
let cf = self.inner.cf_handle(&self.column_name).ok_or(
StorageBackendError::MissingColumnFamily {

MODIFY
storage/src/backend/sled_backend.rs
storage/src/backend/sled_backend.rs
@@ -27,6 +27,16 @@ impl StorageBackend for SledBackend {
.map(|v| v.is_none())?)
}
+ fn put_batch(
+ &mut self,
+ batch: Vec<(EntryHash, ContextValue)>,
+ ) -> Result<(), StorageBackendError> {
+ for (k, v) in batch.into_iter() {
+ self.put(&k, v)?;
+ }
+ Ok(())
+ }
+
fn merge(&mut self, key: &EntryHash, value: ContextValue) -> Result<(), StorageBackendError> {
self.inner.insert(&key.as_ref()[..], value)?;
Ok(())

MODIFY
storage/src/merkle_storage.rs
storage/src/merkle_storage.rs
@@ -1162,10 +1162,8 @@ impl MerkleStorage {
// build list of entries to be persisted
self.get_entries_recursively(entry, &mut batch)?;
- // atomically write all entries in one batch to DB
- for (k, v) in batch {
- self.db.put(&k, v)?;
- }
+ // write all entries at once (depends on backend)
+ self.db.put_batch(batch)?;
Ok(())
}

MODIFY
storage/src/storage_backend.rs
storage/src/storage_backend.rs
@@ -70,6 +70,10 @@ 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 put_batch(
+ &mut self,
+ batch: Vec<(EntryHash, ContextValue)>,
+ ) -> Result<(), 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>;

make use of atomic rocksdb write_batch functionality