Cache deserialized immutable data on storage

Author: ndr-ds

Cargo.lock

@@ -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/Cargo.toml

@@ -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-base/src/lib.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/value_cache.rs linera-base/src/value_cache.rslinera-core/src/value_cache.rs renamed to linera-base/src/value_cache.rs

@@ -32,11 +32,12 @@ 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/actor.rs

@@ -43,10 +43,11 @@ use linera_views::{
 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},
 };
 
@@ -589,7 +590,7 @@ where
             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 @@ where
         }
 
         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 @@ where
         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;
@@ -1670,10 +1671,12 @@ where
         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/chain_worker/state.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/lib.rs

@@ -8,11 +8,10 @@ use linera_base::{
     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 @@ fn test_insert_single_certificate_value() {
     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 @@ fn test_insert_many_certificate_values_individually() {
     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 @@ fn test_reinsertion_of_values() {
     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 @@ fn test_eviction_occurs() {
         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
@@ -175,7 +174,7 @@ fn test_access_affects_eviction() {
             .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 @@ fn test_promotion_of_reinsertion() {
     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/unit_tests/value_cache_tests.rs

@@ -47,9 +47,9 @@ use crate::{
     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-core/src/worker.rs

@@ -81,7 +81,7 @@ use linera_service::{
     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;
@@ -1776,7 +1776,7 @@ impl RunnableWithStore for DatabaseToolJob<'_> {
             } => {
                 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 @@ impl RunnableWithStore for DatabaseToolJob<'_> {
             }
             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 @@ impl RunnableWithStore for DatabaseToolJob<'_> {
             }
             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 @@ impl RunnableWithStore for DatabaseToolJob<'_> {
             }
             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",