starknet_patricia_storage: spawn a task or not based on config

crates/apollo_deployments/resources/app_configs/committer_config.json

@@ -10,7 +10,7 @@
     "committer_config.storage_config.inner_storage_config.enable_statistics": true,
     "committer_config.storage_config.inner_storage_config.max_background_jobs": 8,
     "committer_config.storage_config.inner_storage_config.use_mmap_reads": false,
-    "committer_config.storage_config.inner_storage_config.spawn_blocking_reads": false,
+    "committer_config.storage_config.inner_storage_config.spawn_blocking_reads": true,
     "committer_config.storage_config.inner_storage_config.max_subcompactions": 8,
     "committer_config.storage_config.inner_storage_config.max_write_buffers": 4,
     "committer_config.storage_config.inner_storage_config.num_threads": 8,

crates/apollo_deployments/resources/app_configs/replacer_committer_config.json

@@ -9,7 +9,7 @@
   "committer_config.storage_config.inner_storage_config.cache_size": 8589934592,
   "committer_config.storage_config.inner_storage_config.enable_statistics": true,
   "committer_config.storage_config.inner_storage_config.max_background_jobs": 8,
-  "committer_config.storage_config.inner_storage_config.spawn_blocking_reads": false,
+  "committer_config.storage_config.inner_storage_config.spawn_blocking_reads": true,
   "committer_config.storage_config.inner_storage_config.max_subcompactions": 8,
   "committer_config.storage_config.inner_storage_config.max_write_buffers": 4,
   "committer_config.storage_config.inner_storage_config.num_threads": 8,

crates/apollo_node/resources/config_schema.json

@@ -577,7 +577,7 @@
   "committer_config.storage_config.inner_storage_config.spawn_blocking_reads": {
     "description": "Whether to spawn blocking tasks for read operations",
     "privacy": "Public",
-    "value": false
+    "value": true
   },
   "committer_config.storage_config.inner_storage_config.write_buffer_size": {
     "description": "Amount of data to build up in memory before writing to disk",

crates/starknet_patricia_storage/src/rocksdb_storage.rs

@@ -119,7 +119,7 @@ impl RocksDbOptions {
 pub struct RocksDbStorage {
     db: Arc<DB>,
     options: Arc<RocksDbOptions>,
-    _config: RocksDbStorageConfig,
+    config: RocksDbStorageConfig,
 }
 
 /// Configuration for RocksDB storage.
@@ -170,7 +170,7 @@ impl Default for RocksDbStorageConfig {
             bloom_filter_bits: BLOOM_FILTER_NUM_BITS,
             enable_statistics: true,
             use_mmap_reads: false,
-            spawn_blocking_reads: false,
+            spawn_blocking_reads: true,
         }
     }
 }
@@ -260,7 +260,7 @@ impl RocksDbStorage {
     pub fn new(path: &Path, config: RocksDbStorageConfig) -> PatriciaStorageResult<Self> {
         let options = RocksDbOptions::from_config(&config);
         let db = Arc::new(DB::open(&options.db_options, path)?);
-        Ok(Self { db, options: Arc::new(options), _config: config })
+        Ok(Self { db, options: Arc::new(options), config })
     }
 }
 
@@ -289,27 +289,39 @@ impl Storage for RocksDbStorage {
     type Config = RocksDbStorageConfig;
 
     async fn get(&self, key: &DbKey) -> PatriciaStorageResult<Option<DbValue>> {
-        // TODO(Nimrod): Config should indicate whether to spawn a task or not.
-        let db = self.db.clone();
-        let key = key.clone();
-        Ok(spawn_blocking(move || db.get(&key.0).map(|opt| opt.map(DbValue))).await??)
+        if self.config.spawn_blocking_reads {
+            let db = self.db.clone();
+            let key = key.clone();
+            Ok(spawn_blocking(move || db.get(&key.0).map(|opt| opt.map(DbValue))).await??)
+        } else {
+            Ok(self.db.get(&key.0).map(|opt| opt.map(DbValue))?)
+        }
     }
 
     async fn set(&mut self, key: DbKey, value: DbValue) -> PatriciaStorageResult<()> {
         Ok(self.db.put_opt(&key.0, &value.0, &self.options.write_options)?)
     }
 
     async fn mget(&self, keys: &[&DbKey]) -> PatriciaStorageResult<Vec<Option<DbValue>>> {
-        // TODO(Nimrod): Config should indicate whether to spawn a task or not.
-        let db = self.db.clone();
-        let keys: Vec<Vec<u8>> = keys.iter().map(|k| k.0.clone()).collect();
-        spawn_blocking(move || {
-            db.multi_get(keys)
+        if self.config.spawn_blocking_reads {
+            let db = self.db.clone();
+            let keys: Vec<Vec<u8>> = keys.iter().map(|k| k.0.clone()).collect();
+            Ok(spawn_blocking(move || {
+                db.multi_get(keys)
+                    .into_iter()
+                    .map(|r| Ok(r?.map(DbValue)))
+                    .collect::<PatriciaStorageResult<_>>()
+            })
+            .await??)
+        } else {
+            let raw_keys = keys.iter().map(|k| k.0.as_slice());
+            Ok(self
+                .db
+                .multi_get(raw_keys)
                 .into_iter()
-                .map(|r| Ok(r?.map(DbValue)))
-                .collect::<PatriciaStorageResult<_>>()
-        })
-        .await?
+                .map(|r| r.map(|opt| opt.map(DbValue)).map_err(|e| e.into()))
+                .collect::<Result<Vec<_>, PatriciaStorageError>>()?)
+        }
     }
 
     async fn mset(&mut self, key_to_value: DbHashMap) -> PatriciaStorageResult<()> {