Cache deserialized immutable data on storage

linera-base/Cargo.toml

@@ -49,7 +49,9 @@ hex.workspace = true
 k256.workspace = true
 linera-kywasmtime = { workspace = true, optional = true }
 linera-witty = { workspace = true, features = ["macros"] }
+lru.workspace = true
 prometheus = { workspace = true, optional = true }
+quick_cache.workspace = true
 proptest = { workspace = true, optional = true, features = ["alloc"] }
 rand.workspace = true
 reqwest = { workspace = true, optional = true }

linera-base/src/lib.rs

@@ -39,6 +39,7 @@ pub mod time;
 #[cfg(test)]
 mod unit_tests;
 pub mod util;
+pub mod value_cache;
 pub mod vm;
 
 pub use graphql::BcsHexParseError;

linera-core/src/value_cache.rs → linera-base/src/value_cache.rs

@@ -3,26 +3,24 @@
 
 //! Concurrent caches for values.
 
-#[cfg(test)]
-#[path = "unit_tests/value_cache_tests.rs"]
-mod unit_tests;
-
 #[cfg(with_metrics)]
 use std::any::type_name;
 use std::{borrow::Cow, hash::Hash, num::NonZeroUsize, sync::Mutex};
 
-use linera_base::{crypto::CryptoHash, hashed::Hashed};
 use lru::LruCache;
 use quick_cache::sync::Cache;
 
+use crate::{crypto::CryptoHash, hashed::Hashed};
+
 /// A counter metric for the number of cache hits in the [`ValueCache`].
 #[cfg(with_metrics)]
 mod metrics {
     use std::sync::LazyLock;
 
-    use linera_base::prometheus_util::register_int_counter_vec;
     use prometheus::IntCounterVec;
 
+    use crate::prometheus_util::register_int_counter_vec;
+
     pub static CACHE_HIT_COUNT: LazyLock<IntCounterVec> = LazyLock::new(|| {
         register_int_counter_vec(
             "value_cache_hit",
@@ -65,11 +63,27 @@ where
         }
     }
 
+    /// Inserts a value into the cache with the given key.
+    /// Returns `true` if the value was newly inserted, `false` if it already existed.
+    pub fn insert(&self, key: K, value: V) -> bool {
+        if self.contains_key(&key) {
+            false
+        } else {
+            self.cache.insert(key, value);
+            true
+        }
+    }
+
     /// Returns a `V` from the cache, if present.
     pub fn get(&self, key: &K) -> Option<V> {
         Self::track_cache_usage(self.cache.get(key))
     }
 
+    /// Returns `true` if the cache contains the given key.
+    pub fn contains_key(&self, key: &K) -> bool {
+        self.cache.get(key).is_some()
+    }
+
     fn track_cache_usage(maybe_value: Option<V>) -> Option<V> {
         #[cfg(with_metrics)]
         {
@@ -94,7 +108,7 @@ impl<T: Clone> ValueCache<CryptoHash, Hashed<T>> {
     /// inserted in the cache.
     ///
     /// Returns [`true`] if the value was not already present in the cache.
-    pub fn insert(&self, value: Cow<Hashed<T>>) -> bool {
+    pub fn insert_hashed(&self, value: Cow<Hashed<T>>) -> bool {
         let hash = (*value).hash();
         if self.cache.get(&hash).is_some() {
             false
@@ -108,7 +122,7 @@ impl<T: Clone> ValueCache<CryptoHash, Hashed<T>> {
     ///
     /// The `values` are wrapped in [`Cow`]s so that each `value` is only cloned if it
     /// needs to be inserted in the cache.
-    #[cfg(test)]
+    #[cfg(with_testing)]
     pub fn insert_all<'a>(&self, values: impl IntoIterator<Item = Cow<'a, Hashed<T>>>)
     where
         T: 'a,
@@ -122,7 +136,7 @@ impl<T: Clone> ValueCache<CryptoHash, Hashed<T>> {
     }
 }
 
-#[cfg(test)]
+#[cfg(with_testing)]
 impl<K, V> ValueCache<K, V>
 where
     K: Hash + Eq + Clone,
@@ -137,6 +151,11 @@ where
     pub fn len(&self) -> usize {
         self.cache.len()
     }
+
+    /// Returns [`true`] if the cache is empty.
+    pub fn is_empty(&self) -> bool {
+        self.cache.len() == 0
+    }
 }
 
 /// A cache for values that need to be "parked" temporarily and taken out for exclusive use.

linera-core/src/chain_worker/actor.rs

@@ -14,7 +14,7 @@
 use linera_base::{
     crypto::{CryptoHash, ValidatorPublicKey},
     data_types::{ApplicationDescription, Blob, BlockHeight, Epoch, TimeDelta, Timestamp},
     hashed::Hashed,
     identifiers::{ApplicationId, BlobId, ChainId, StreamId},
     time::Instant,
 };
@@ -28,15 +28,16 @@
     ServiceRuntimeEndpoint, ServiceSyncRuntime,
 };
 use linera_storage::{Clock as _, Storage};
 use linera_views::context::InactiveContext;
 use tokio::sync::{mpsc, oneshot, OwnedRwLockReadGuard};
 use tracing::{debug, instrument, trace, Instrument as _};
 
+use linera_base::value_cache::{ParkingCache, ValueCache};
+
 use super::{config::ChainWorkerConfig, state::ChainWorkerState, DeliveryNotifier};
 use crate::{
     chain_worker::BlockOutcome,
     data_types::{ChainInfoQuery, ChainInfoResponse},
-    value_cache::{ParkingCache, ValueCache},
     worker::{NetworkActions, WorkerError},
 };
 

linera-core/src/chain_worker/state.rs

@@ -17,7 +17,7 @@
     data_types::{
         ApplicationDescription, ArithmeticError, Blob, BlockHeight, Epoch, Round, Timestamp,
     },
     ensure,
     hashed::Hashed,
     identifiers::{AccountOwner, ApplicationId, BlobId, BlobType, ChainId, EventId, StreamId},
 };
@@ -40,13 +40,14 @@
     context::{Context, InactiveContext},
     views::{ClonableView, ReplaceContext as _, RootView as _, View as _},
 };
 use tokio::sync::{oneshot, OwnedRwLockReadGuard, RwLock, RwLockWriteGuard};
 use tracing::{debug, instrument, trace, warn};
 
+use linera_base::value_cache::{ParkingCache, ValueCache};
+
 use super::{ChainWorkerConfig, ChainWorkerRequest, DeliveryNotifier, EventSubscriptionsResult};
 use crate::{
     data_types::{ChainInfo, ChainInfoQuery, ChainInfoResponse, CrossChainRequest},
-    value_cache::{ParkingCache, ValueCache},
     worker::{NetworkActions, Notification, Reason, WorkerError},
 };
 
@@ -586,10 +587,10 @@
                 }
             };
 
             for block in blocks {
                 let hashed_block = block.into_inner();
                 let height = hashed_block.inner().header.height;
-                self.block_values.insert(Cow::Owned(hashed_block.clone()));
+                self.block_values.insert_hashed(Cow::Owned(hashed_block.clone()));
                 height_to_blocks.insert(height, hashed_block);
             }
         }
@@ -782,7 +783,7 @@
         }
 
         self.block_values
-            .insert(Cow::Borrowed(certificate.inner().inner()));
+            .insert_hashed(Cow::Borrowed(certificate.inner().inner()));
         let required_blob_ids = block.required_blob_ids();
         let maybe_blobs = self
             .maybe_get_required_blobs(required_blob_ids, Some(&block.created_blobs()))
@@ -1027,7 +1028,7 @@
         self.save().await?;
 
         self.block_values
-            .insert(Cow::Owned(certificate.into_inner().into_inner()));
+            .insert_hashed(Cow::Owned(certificate.into_inner().into_inner()));
 
         self.register_delivery_notifier(height, &actions, notify_when_messages_are_delivered)
             .await;
@@ -1454,7 +1455,7 @@
                             .clone_with_base_key(ctx.base_key().bytes.clone())
                     })
                     .await
                     .query_application(context, query, self.service_runtime_endpoint.as_mut())
                     .await
                     .with_execution_context(ChainExecutionContext::Query)?;
                 self.execution_state_cache
@@ -1670,10 +1671,12 @@
         match manager.create_vote(proposal, block, key_pair, local_time, blobs)? {
             // Cache the value we voted on, so the client doesn't have to send it again.
             Some(Either::Left(vote)) => {
-                self.block_values.insert(Cow::Borrowed(vote.value.inner()));
+                self.block_values
+                    .insert_hashed(Cow::Borrowed(vote.value.inner()));
             }
             Some(Either::Right(vote)) => {
-                self.block_values.insert(Cow::Borrowed(vote.value.inner()));
+                self.block_values
+                    .insert_hashed(Cow::Borrowed(vote.value.inner()));
             }
             None => (),
         }

linera-core/src/lib.rs

@@ -23,7 +23,10 @@ pub mod test_utils;
 pub mod worker;
 
 pub(crate) mod updater;
-mod value_cache;
+
+#[cfg(test)]
+#[path = "unit_tests/value_cache_tests.rs"]
+mod value_cache_tests;
 
 pub use local_node::LocalNodeError;
 pub use updater::DEFAULT_QUORUM_GRACE_PERIOD;

linera-core/src/unit_tests/value_cache_tests.rs

@@ -8,11 +8,10 @@
     data_types::{BlockHeight, Epoch},
     hashed::Hashed,
     identifiers::ChainId,
+    value_cache::ValueCache,
 };
 use linera_chain::types::Timeout;
 
-use super::ValueCache;
-
 /// Test cache size for unit tests.
 const TEST_CACHE_SIZE: usize = 10;
 
@@ -33,7 +32,7 @@
     let value = create_dummy_certificate_value(0);
     let hash = value.hash();
 
-    assert!(cache.insert(Cow::Borrowed(&value)));
+    assert!(cache.insert_hashed(Cow::Borrowed(&value)));
     assert!(cache.contains(&hash));
     assert_eq!(cache.get(&hash), Some(value));
     assert_eq!(cache.len(), 1);
@@ -46,7 +45,7 @@
     let values = create_dummy_certificate_values(0..(TEST_CACHE_SIZE as u64)).collect::<Vec<_>>();
 
     for value in &values {
-        assert!(cache.insert(Cow::Borrowed(value)));
+        assert!(cache.insert_hashed(Cow::Borrowed(value)));
     }
 
     for value in &values {
@@ -82,7 +81,7 @@
     cache.insert_all(values.iter().map(Cow::Borrowed));
 
     for value in &values {
-        assert!(!cache.insert(Cow::Borrowed(value)));
+        assert!(!cache.insert_hashed(Cow::Borrowed(value)));
     }
 
     for value in &values {
@@ -105,7 +104,7 @@
         create_dummy_certificate_values(0..((TEST_CACHE_SIZE as u64) * 2)).collect::<Vec<_>>();
 
     for value in &values {
-        cache.insert(Cow::Borrowed(value));
+        cache.insert_hashed(Cow::Borrowed(value));
     }
 
     // Cache size should be bounded by capacity
@@ -143,7 +142,7 @@
         TEST_CACHE_SIZE
     );
 
     // Exactly one value should have been evicted
     let present_count = values.iter().filter(|v| cache.contains(&v.hash())).count();
     assert_eq!(
         present_count, TEST_CACHE_SIZE,
@@ -175,7 +174,7 @@
             .collect::<Vec<_>>();
 
     for value in &extra_values {
-        cache.insert(Cow::Borrowed(value));
+        cache.insert_hashed(Cow::Borrowed(value));
     }
 
     // The frequently accessed first value should still be present
@@ -195,15 +194,15 @@
     cache.insert_all(values.iter().map(Cow::Borrowed));
 
     // Re-insert the first value (this should "promote" it)
-    assert!(!cache.insert(Cow::Borrowed(&values[0])));
+    assert!(!cache.insert_hashed(Cow::Borrowed(&values[0])));
 
     // Insert additional values to trigger eviction
     let extra_values =
         create_dummy_certificate_values((TEST_CACHE_SIZE as u64)..((TEST_CACHE_SIZE as u64) + 3))
             .collect::<Vec<_>>();
 
     for value in &extra_values {
-        cache.insert(Cow::Borrowed(value));
+        cache.insert_hashed(Cow::Borrowed(value));
     }
 
     // The re-inserted first value should still be present

linera-core/src/worker.rs

@@ -16,7 +16,7 @@
     },
     doc_scalar,
     hashed::Hashed,
     identifiers::{AccountOwner, ApplicationId, BlobId, ChainId, EventId, StreamId},
     time::Instant,
     util::traits::DynError,
 };
@@ -46,10 +46,10 @@
     },
     data_types::{ChainInfoQuery, ChainInfoResponse, CrossChainRequest},
     join_set_ext::{JoinSet, JoinSetExt},
     notifier::Notifier,
-    value_cache::{ParkingCache, ValueCache},
     CHAIN_INFO_MAX_RECEIVED_LOG_ENTRIES,
 };
+use linera_base::value_cache::{ParkingCache, ValueCache};
 
 #[cfg(test)]
 #[path = "unit_tests/worker_tests.rs"]

linera-service/src/cli/main.rs

@@ -81,7 +81,7 @@
     task_processor::TaskProcessor,
     util,
 };
-use linera_storage::{DbStorage, Storage};
+use linera_storage::{DbStorage, Storage, StorageCacheConfig};
 use linera_views::store::{KeyValueDatabase, KeyValueStore};
 use options::Options;
 use serde_json::Value;
@@ -1772,11 +1772,11 @@
                 }
             }
             DatabaseToolCommand::Initialize {
                 genesis_config_path,
             } => {
                 let genesis_config: GenesisConfig = util::read_json(genesis_config_path)?;
                 let mut storage =
-                    DbStorage::<D, _>::maybe_create_and_connect(&config, &namespace, None).await?;
+                    DbStorage::<D, _>::maybe_create_and_connect(&config, &namespace, None, StorageCacheConfig::default()).await?;
                 genesis_config.initialize_storage(&mut storage).await?;
                 info!(
                     "Namespace {namespace} was initialized in {} ms",
@@ -1796,7 +1796,7 @@
             }
             DatabaseToolCommand::ListBlobIds => {
                 let storage =
-                    DbStorage::<D, _>::maybe_create_and_connect(&config, &namespace, None).await?;
+                    DbStorage::<D, _>::maybe_create_and_connect(&config, &namespace, None, StorageCacheConfig::default()).await?;
                 let blob_ids = storage.list_blob_ids().await?;
                 info!("Blob IDs listed in {} ms", start_time.elapsed().as_millis());
                 info!("The list of blob IDs is:");
@@ -1806,7 +1806,7 @@
             }
             DatabaseToolCommand::ListChainIds => {
                 let storage =
-                    DbStorage::<D, _>::maybe_create_and_connect(&config, &namespace, None).await?;
+                    DbStorage::<D, _>::maybe_create_and_connect(&config, &namespace, None, StorageCacheConfig::default()).await?;
                 let chain_ids = storage.list_chain_ids().await?;
                 info!(
                     "Chain IDs listed in {} ms",
@@ -1819,7 +1819,7 @@
             }
             DatabaseToolCommand::ListEventIds => {
                 let storage =
-                    DbStorage::<D, _>::maybe_create_and_connect(&config, &namespace, None).await?;
+                    DbStorage::<D, _>::maybe_create_and_connect(&config, &namespace, None, StorageCacheConfig::default()).await?;
                 let event_ids = storage.list_event_ids().await?;
                 info!(
                     "Event IDs listed in {} ms",

linera-service/src/cli/options.rs

@@ -132,8 +132,10 @@ impl Options {
         debug!("Running command using storage configuration: {storage_config}");
         let store_config =
             storage_config.add_common_storage_options(&self.common_storage_options)?;
+        let db_storage_cache_config = self.common_storage_options.db_storage_cache_config();
         let output = Box::pin(store_config.run_with_storage(
             self.wasm_runtime.with_wasm_default(),
+            db_storage_cache_config,
             self.application_logs,
             job,
         ))

linera-service/src/exporter/main.rs

@@ -183,10 +183,11 @@ impl ExporterOptions {
                 .storage_config
                 .add_common_storage_options(&self.common_storage_options)
                 .unwrap();
+            let db_storage_cache_config = self.common_storage_options.db_storage_cache_config();
             // Exporters are part of validator infrastructure and should not output contract logs.
             let allow_application_logs = false;
             store_config
-                .run_with_storage(None, allow_application_logs, context)
+                .run_with_storage(None, db_storage_cache_config, allow_application_logs, context)
                 .boxed()
                 .await
         };

linera-service/src/proxy/main.rs

@@ -506,11 +506,13 @@ impl ProxyOptions {
         let store_config = self
             .storage_config
             .add_common_storage_options(&self.common_storage_options)?;
+        let db_storage_cache_config = self.common_storage_options.db_storage_cache_config();
         // Proxies are part of validator infrastructure and should not output contract logs.
         let allow_application_logs = false;
         store_config
             .run_with_storage(
                 None,
+                db_storage_cache_config,
                 allow_application_logs,
                 ProxyContext::from_options(self)?,
             )

linera-service/src/server.rs

@@ -561,10 +561,11 @@ async fn run(options: ServerOptions) {
             let store_config = storage_config
                 .add_common_storage_options(&common_storage_options)
                 .unwrap();
+            let db_storage_cache_config = common_storage_options.db_storage_cache_config();
             // Validators should not output contract logs.
             let allow_application_logs = false;
             store_config
-                .run_with_storage(wasm_runtime, allow_application_logs, job)
+                .run_with_storage(wasm_runtime, db_storage_cache_config, allow_application_logs, job)
                 .boxed()
                 .await
                 .unwrap()