diff --git a/docs/configuration/node-config.md b/docs/configuration/node-config.md index cbe3f8b0022..40d92944a56 100644 --- a/docs/configuration/node-config.md +++ b/docs/configuration/node-config.md @@ -313,6 +313,11 @@ This section contains the configuration options for a Searcher. | `request_timeout_secs` | The time before a search request is cancelled. This should match the timeout of the stack calling into quickwit if there is one set. | `30` | | `use_metastore_read_replica` | If true, routes read-only metastore requests from searchers, including DataFusion when enabled, to nodes running the `metastore_read_replica` service. Searchers require at least one `metastore_read_replica` node at startup and do not fall back to the primary metastore. | `false` | +These in-memory cache capacities bound the memory held by cached entries, keys included, not just +the size of the cached values. `partial_request_cache_capacity` is the most affected by this, since +its keys embed the search request itself: at an unchanged setting it holds somewhat fewer entries +than a value-only budget would suggest. + ### Searcher split cache configuration This section contains the configuration options for the on-disk searcher split cache. Files are stored in the data directory under `searcher-split-cache/`. diff --git a/quickwit/quickwit-config/src/node_config/mod.rs b/quickwit/quickwit-config/src/node_config/mod.rs index 5e02be1af39..e596e23298a 100644 --- a/quickwit/quickwit-config/src/node_config/mod.rs +++ b/quickwit/quickwit-config/src/node_config/mod.rs @@ -561,6 +561,14 @@ impl CacheConfig { } } + pub fn with_capacity_and_policy(capacity: ByteSize, policy: CachePolicy) -> Self { + CacheConfig { + capacity: Some(capacity), + policy: Some(policy), + virtual_caches: Vec::new(), + } + } + pub fn capacity(&self) -> ByteSize { // this should always be there self.capacity.unwrap_or_default() diff --git a/quickwit/quickwit-search/src/leaf_cache.rs b/quickwit/quickwit-search/src/leaf_cache.rs index 39758b08956..5fa97b0f49e 100644 --- a/quickwit/quickwit-search/src/leaf_cache.rs +++ b/quickwit/quickwit-search/src/leaf_cache.rs @@ -19,7 +19,7 @@ use quickwit_config::CacheConfig; use quickwit_proto::search::{ CountHits, LeafResourceStats, LeafSearchResponse, SearchRequest, SplitIdAndFooterOffsets, }; -use quickwit_storage::{MemorySizedCache, OwnedBytes}; +use quickwit_storage::{MemUsage, MemorySizedCache, OwnedBytes}; use tantivy::index::SegmentId; /// A cache to memoize `leaf_search_single_split` results. @@ -118,6 +118,14 @@ impl CacheKey { } } +impl MemUsage for CacheKey { + fn heap_mem_usage(&self) -> usize { + // `SearchRequest` cannot easily implement `MemUsage`, but `encoded_len()` is a good proxy + // for its memory footprint. + self.split_id.heap_mem_usage() + self.request.encoded_len() + } +} + /// A (half-open) range bounded inclusively below and exclusively above [start..end). #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)] struct HalfOpenRange { @@ -249,7 +257,45 @@ mod tests { SplitIdAndFooterOffsets, }; - use super::LeafSearchCache; + use super::{CacheKey, LeafSearchCache, MemUsage}; + + #[test] + fn test_cache_key_mem_usage_accounts_the_request() { + let split_info = SplitIdAndFooterOffsets { + split_id: "01H4V2E9K7XQZ8N3M5P6R7S8T9".to_string(), + split_footer_start: 0, + split_footer_end: 100, + timestamp_start: None, + timestamp_end: None, + num_docs: 0, + }; + let short_request = SearchRequest { + index_id_patterns: vec!["test-idx".to_string()], + query_ast: "short".to_string(), + max_hits: 10, + ..Default::default() + }; + let padding = "x".repeat(10_000); + let long_request = SearchRequest { + query_ast: format!("short{padding}"), + ..short_request.clone() + }; + + let short_key = CacheKey::from_split_meta_and_request(split_info.clone(), short_request); + let long_key = CacheKey::from_split_meta_and_request(split_info, long_request); + + // The split id and the request are both owned outside the struct itself. + assert!(short_key.mem_usage() > size_of::()); + // A longer query AST is charged for, roughly byte for byte. `encoded_len` is a proxy, so + // only the order of magnitude is pinned here: the extra length must show up, and must not + // be wildly over-counted. + let delta = long_key.mem_usage() - short_key.mem_usage(); + assert!( + delta >= padding.len() && delta <= padding.len() + 16, + "a {} byte longer query AST grew the key by {delta} bytes", + padding.len() + ); + } #[test] fn test_leaf_search_cache_no_timestamp() { diff --git a/quickwit/quickwit-storage/src/cache/base_cache.rs b/quickwit/quickwit-storage/src/cache/base_cache.rs index 810ef341c65..b85d7b3ea4c 100644 --- a/quickwit/quickwit-storage/src/cache/base_cache.rs +++ b/quickwit/quickwit-storage/src/cache/base_cache.rs @@ -26,7 +26,8 @@ use tokio::time::Instant; use tracing::{error, warn}; use crate::OwnedBytes; -use crate::cache::stored_item::{StoredItem, ValueLen}; +use crate::cache::mem_usage::MemUsage; +use crate::cache::stored_item::{KeyedEntry, StoredItem, ValueLen}; use crate::metrics::SingleCacheMetrics; /// We do not evict anything that has been accessed in the last 60s. @@ -76,7 +77,7 @@ pub(crate) enum AnyCache { TinyLfu(TinyLfu), } -impl +impl AnyCache { pub fn from_policy_and_capacity( @@ -142,7 +143,7 @@ impl Drop for Lru { } } -impl Lru { +impl Lru { /// Creates a new NeedMutSliceCache with the given capacity. fn with_capacity(capacity: Capacity, cache_metrics: SingleCacheMetrics) -> Self { Lru { @@ -185,7 +186,11 @@ impl Lru { let item_opt = self.lru_cache.get_mut(cache_key); if let Some(item) = item_opt { self.cache_metrics.hits_num_items.inc(); - self.cache_metrics.hits_num_bytes.inc_by(item.len() as u64); + // Hits are measured in payload bytes: this counts what the caller got instead of + // going to storage, so the key is deliberately excluded. + self.cache_metrics + .hits_num_bytes + .inc_by(item.payload_num_bytes() as u64); Some(item.payload()) } else { self.cache_metrics.misses_num_items.inc(); @@ -194,28 +199,31 @@ impl Lru { } /// Attempt to put the given amount of data in the cache. - /// This may fail silently if the owned_bytes slice is larger than the cache - /// capacity. + /// This may fail silently if the key and the owned_bytes slice together are larger than the + /// cache capacity. fn put(&mut self, key: K, bytes: V) { - if self.capacity.exceeds_capacity(bytes.len()) { - // The value does not fit in the cache. We simply don't store it. + let key_mem_usage = key.mem_usage(); + let entry_num_bytes = key_mem_usage + bytes.len(); + if self.capacity.exceeds_capacity(entry_num_bytes) { + // The entry does not fit in the cache. We simply don't store it. if self.capacity != Capacity::InBytes(0) { warn!( capacity_in_bytes = ?self.capacity, len = bytes.len(), - "Downloaded a byte slice larger than the cache capacity." + key_mem_usage, + "Downloaded a cache entry (key + value) larger than the cache capacity." ); } return; } if let Some(previous_data) = self.lru_cache.pop(&key) { - self.drop_item(previous_data.len() as u64); + self.drop_item(previous_data.entry_num_bytes() as u64); } let now = Instant::now(); while self .capacity - .exceeds_capacity(self.num_bytes as usize + bytes.len()) + .exceeds_capacity(self.num_bytes as usize + entry_num_bytes) { if let Some((_, candidate_for_eviction)) = self.lru_cache.peek_lru() { let time_since_last_access = @@ -227,8 +235,8 @@ impl Lru { return; } } - if let Some((_, bytes)) = self.lru_cache.pop_lru() { - self.drop_item(bytes.len() as u64); + if let Some((_, stored_item)) = self.lru_cache.pop_lru() { + self.drop_item(stored_item.entry_num_bytes() as u64); } else { error!( "Logical error. Even after removing all of the items in the cache the \ @@ -238,8 +246,9 @@ impl Lru { return; } } - self.record_item(bytes.len() as u64); - self.lru_cache.put(key, StoredItem::new(bytes, now)); + self.record_item(entry_num_bytes as u64); + self.lru_cache + .put(key, StoredItem::new(bytes, key_mem_usage, now)); } } @@ -247,8 +256,13 @@ impl Lru { // readme says. While both are clearly distinct (one being clock-based, the other being fifo // based), they are not too disimilar in term of strenght/weaknesses. pub struct S3Fifo { - cache: - QuickCache, + cache: QuickCache< + K, + KeyedEntry, + QuickCacheWeighter, + quick_cache::DefaultHashBuilder, + QuickCacheLifecycle, + >, capacity: u64, cache_metrics: SingleCacheMetrics, } @@ -266,9 +280,12 @@ impl Drop for S3Fifo { } struct QuickCacheWeighter; -impl quick_cache::Weighter for QuickCacheWeighter { - fn weight(&self, _key: &K, value: &V) -> u64 { - value.len() as u64 +impl quick_cache::Weighter> for QuickCacheWeighter { + // The key footprint is carried by the entry itself, so the key is not needed here. This also + // keeps `Weighter` free of a `K: MemUsage` bound, which `Drop for S3Fifo` would otherwise have + // to carry all the way up to `MemorySizedCache`'s declaration. + fn weight(&self, _key: &K, entry: &KeyedEntry) -> u64 { + entry.entry_num_bytes() as u64 } } @@ -278,16 +295,16 @@ struct QuickCacheQueryEffect { count: u64, bytes: u64, } -impl quick_cache::Lifecycle for QuickCacheLifecycle { +impl quick_cache::Lifecycle> for QuickCacheLifecycle { type RequestState = QuickCacheQueryEffect; - fn on_evict(&self, state: &mut Self::RequestState, _key: K, val: V) { + fn on_evict(&self, state: &mut Self::RequestState, _key: K, entry: KeyedEntry) { state.count += 1; - state.bytes += val.len() as u64; + state.bytes += entry.entry_num_bytes() as u64; } } -impl S3Fifo { +impl S3Fifo { /// Creates a new NeedMutSliceCache with the given capacity. fn with_capacity(capacity: u64, cache_metrics: SingleCacheMetrics) -> Self { S3Fifo { @@ -308,11 +325,15 @@ impl S3Fifo { K: Borrow, Q: Hash + Eq + ?Sized, { - let item_opt = self.cache.get(cache_key); - if let Some(item) = item_opt { + let entry_opt = self.cache.get(cache_key); + if let Some(entry) = entry_opt { self.cache_metrics.hits_num_items.inc(); - self.cache_metrics.hits_num_bytes.inc_by(item.len() as u64); - Some(item.clone()) + // Hits are measured in payload bytes: this counts what the caller got instead of + // going to storage, so the key is deliberately excluded. + self.cache_metrics + .hits_num_bytes + .inc_by(entry.payload_num_bytes() as u64); + Some(entry.value().clone()) } else { self.cache_metrics.misses_num_items.inc(); None @@ -320,16 +341,20 @@ impl S3Fifo { } /// Attempt to put the given amount of data in the cache. - /// This may fail silently if the owned_bytes slice is larger than the cache - /// capacity. + /// This may fail silently if the key and the owned_bytes slice together are larger than the + /// cache capacity. fn put(&mut self, key: K, value: V) { - if self.capacity < value.len() as u64 { - // The value does not fit in the cache. We simply don't store it. + // Measured before `key` is moved into the cache below. + let key_mem_usage = key.mem_usage(); + let entry_num_bytes = (key_mem_usage + value.len()) as u64; + if self.capacity < entry_num_bytes { + // The entry does not fit in the cache. We simply don't store it. if self.capacity != 0 { warn!( capacity_in_bytes = ?self.capacity, len = value.len(), - "Downloaded a byte slice larger than the cache capacity." + key_mem_usage, + "Downloaded a cache entry (key + value) larger than the cache capacity." ); } return; @@ -338,9 +363,10 @@ impl S3Fifo { self.cache_metrics.in_cache_count.inc(); self.cache_metrics .in_cache_num_bytes - .inc_by(value.len() as f64); + .inc_by(entry_num_bytes as f64); let mut evicted = QuickCacheQueryEffect::default(); - self.cache.insert_with_lifecycle(key, value, &mut evicted); + self.cache + .insert_with_lifecycle(key, KeyedEntry::new(value, key_mem_usage), &mut evicted); self.cache_metrics .in_cache_count .dec_by(evicted.count as f64); @@ -354,19 +380,21 @@ impl S3Fifo { // We don't make this value Clone to ensure each item is dropped only once struct CapacityTracker { - item: V, + // Moka hands us no key on drop, hence the memorized key footprint inside `KeyedEntry`. + entry: KeyedEntry, cache_metrics: Weak, } impl Drop for CapacityTracker { fn drop(&mut self) { if let Some(cache_metrics) = self.cache_metrics.upgrade() { + let entry_num_bytes = self.entry.entry_num_bytes(); cache_metrics.in_cache_count.dec(); cache_metrics .in_cache_num_bytes - .dec_by(self.item.len() as f64); + .dec_by(entry_num_bytes as f64); cache_metrics.evict_num_items.inc(); - cache_metrics.evict_num_bytes.inc_by(self.item.len() as u64); + cache_metrics.evict_num_bytes.inc_by(entry_num_bytes as u64); } } } @@ -395,7 +423,7 @@ impl Drop for TinyLfu { } } -impl +impl TinyLfu { /// Creates a new NeedMutSliceCache with the given capacity. @@ -403,8 +431,9 @@ impl>| { - v.item.len().try_into().unwrap_or(u32::MAX) + v.entry.entry_num_bytes().try_into().unwrap_or(u32::MAX) }) .build(), capacity, @@ -420,10 +449,12 @@ impl()`, field contributions compose without double-counting inline bytes. + fn heap_mem_usage(&self) -> usize; + + /// `size_of::()` plus every byte this value transitively owns. + /// + /// Counting `size_of::()` is deliberate: a key stored inside a cache lives in a + /// heap-allocated node, so its inline bytes are resident too. + fn mem_usage(&self) -> usize + where Self: Sized { + size_of::() + self.heap_mem_usage() + } +} + +impl MemUsage for String { + fn heap_mem_usage(&self) -> usize { + self.capacity() + } +} + +impl MemUsage for PathBuf { + fn heap_mem_usage(&self) -> usize { + self.capacity() + } +} + +/// A borrow owns nothing, however large the pointee is. +impl MemUsage for &T { + fn heap_mem_usage(&self) -> usize { + 0 + } +} + +impl MemUsage for (A, B) { + fn heap_mem_usage(&self) -> usize { + self.0.heap_mem_usage() + self.1.heap_mem_usage() + } +} + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + + use super::MemUsage; + + #[test] + fn test_string_mem_usage() { + assert_eq!(String::new().mem_usage(), size_of::()); + assert_eq!("hello".to_string().mem_usage(), size_of::() + 5); + } + + #[test] + fn test_string_mem_usage_counts_spare_capacity() { + let mut string = String::with_capacity(100); + string.push_str("hello"); + // We charge the allocation, not the length. + assert_eq!(string.mem_usage(), size_of::() + 100); + } + + #[test] + fn test_path_buf_mem_usage() { + let path_buf = PathBuf::from("/tmp/split.split"); + assert_eq!( + path_buf.mem_usage(), + size_of::() + path_buf.capacity() + ); + } + + #[test] + fn test_str_ref_mem_usage() { + // A borrow owns nothing, however long the pointee is. + assert_eq!("hello".mem_usage(), size_of::<&str>()); + assert_eq!("hello world, at length".mem_usage(), size_of::<&str>()); + } + + #[test] + fn test_tuple_mem_usage() { + let key = ("abc".to_string(), "de".to_string()); + assert_eq!(key.mem_usage(), size_of::<(String, String)>() + 3 + 2); + assert_eq!(key.heap_mem_usage(), 5); + } + + #[test] + fn test_heap_mem_usage_excludes_inline_size() { + assert_eq!("hello".to_string().heap_mem_usage(), 5); + assert_eq!(String::new().heap_mem_usage(), 0); + assert_eq!(MemUsage::heap_mem_usage(&"borrowed"), 0); + } + + #[test] + fn test_nested_struct_mem_usage_does_not_double_count() { + struct Key { + name: String, + path: PathBuf, + } + impl MemUsage for Key { + fn heap_mem_usage(&self) -> usize { + self.name.heap_mem_usage() + self.path.heap_mem_usage() + } + } + let key = Key { + name: "hello".to_string(), + path: PathBuf::from("/tmp/split.split"), + }; + assert_eq!(key.mem_usage(), size_of::() + 5 + key.path.capacity()); + } +} diff --git a/quickwit/quickwit-storage/src/cache/memory_sized_cache.rs b/quickwit/quickwit-storage/src/cache/memory_sized_cache.rs index 707d633501f..0cfdded69fd 100644 --- a/quickwit/quickwit-storage/src/cache/memory_sized_cache.rs +++ b/quickwit/quickwit-storage/src/cache/memory_sized_cache.rs @@ -22,6 +22,7 @@ use quickwit_config::CacheConfig; use crate::OwnedBytes; use crate::cache::base_cache::{AnyCache, FakeCacheEntry}; +use crate::cache::mem_usage::MemUsage; use crate::cache::slice_address::{SliceAddress, SliceAddressKey, SliceAddressRef}; use crate::metrics::CacheMetrics; @@ -30,7 +31,7 @@ struct CacheState { virtual_caches: Vec>, } -impl CacheState { +impl CacheState { fn from_config(cache_config: &CacheConfig, cache_counters: &'static CacheMetrics) -> Self { let cache = AnyCache::from_policy_and_capacity( cache_config.policy(), @@ -91,7 +92,7 @@ pub struct MemorySizedCache { inner: Mutex>, } -impl MemorySizedCache { +impl MemorySizedCache { /// Creates an slice cache with the given capacity. pub fn from_config(cache_config: &CacheConfig, cache_counters: &'static CacheMetrics) -> Self { MemorySizedCache { @@ -117,8 +118,9 @@ impl MemorySizedCache { } /// Attempt to put the given amount of data in the cache. - /// This may fail silently if the owned_bytes slice is larger than the cache - /// capacity. + /// + /// An entry is charged for its key as well as its value, so this may fail silently if the key + /// and the owned_bytes slice together are larger than the cache capacity. pub fn put(&self, val: K, bytes: OwnedBytes) { self.inner.lock().unwrap().put(val, bytes); } @@ -132,8 +134,9 @@ impl MemorySizedCache { } /// Attempt to put the given amount of data in the cache. - /// This may fail silently if the owned_bytes slice is larger than the cache - /// capacity. + /// + /// An entry is charged for its key as well as its value, so this may fail silently if the key + /// and the owned_bytes slice together are larger than the cache capacity. pub fn put_slice(&self, path: PathBuf, byte_range: Range, bytes: OwnedBytes) { let slice_address = SliceAddress { path, byte_range }; self.put(slice_address, bytes); @@ -142,19 +145,27 @@ impl MemorySizedCache { #[cfg(test)] mod tests { + use std::mem::size_of; + use bytesize::ByteSize; + use quickwit_config::CachePolicy; use super::*; use crate::cache::base_cache::LRU_MIN_TIME_SINCE_LAST_ACCESS; use crate::metrics::CACHE_METRICS_FOR_TESTS; + /// Memory charged for one of the single-character `String` keys used below: the inline size of + /// a `String` plus its one byte of heap. Entries are charged their key on top of their value, + /// so capacities in these tests are expressed relative to it. + const KEY_COST: usize = size_of::() + 1; + #[tokio::test] async fn test_cache_edge_condition() { tokio::time::pause(); - let cache = MemorySizedCache::::from_config( - &ByteSize::b(5).into(), - &CACHE_METRICS_FOR_TESTS, - ); + // Room for the two first entries ("abc" and "de") and their keys, exactly. + let capacity = ByteSize::b((2 * KEY_COST + 5) as u64); + let cache = + MemorySizedCache::::from_config(&capacity.into(), &CACHE_METRICS_FOR_TESTS); { let data = OwnedBytes::new(&b"abc"[..]); cache.put("3".to_string(), data); @@ -184,7 +195,8 @@ mod tests { } tokio::time::advance(LRU_MIN_TIME_SINCE_LAST_ACCESS.mul_f32(1.1f32)).await; { - let data = OwnedBytes::new(&b"klmnop"[..]); + // Large enough that even alone with its key it overflows the whole cache. + let data = OwnedBytes::new(vec![0u8; capacity.as_u64() as usize - KEY_COST + 1]); cache.put("6".to_string(), data); // The entry put should have been dismissed as it is too large for the cache assert!(cache.get(&"6".to_string()).is_none()); @@ -193,6 +205,59 @@ mod tests { } } + #[test] + fn test_cache_charges_the_key() { + let data = OwnedBytes::new(&b"abc"[..]); + // A capacity covering the value alone is not enough: the key is charged too. + let value_only_cache = MemorySizedCache::::from_config( + &ByteSize::b(data.len() as u64).into(), + &CACHE_METRICS_FOR_TESTS, + ); + value_only_cache.put("3".to_string(), data.clone()); + assert!(value_only_cache.get(&"3".to_string()).is_none()); + + // Room for the key on top of the value, and the very same entry fits. + let key_and_value_cache = MemorySizedCache::::from_config( + &ByteSize::b((KEY_COST + data.len()) as u64).into(), + &CACHE_METRICS_FOR_TESTS, + ); + key_and_value_cache.put("3".to_string(), data); + assert_eq!( + key_and_value_cache.get(&"3".to_string()).unwrap(), + &b"abc"[..] + ); + } + + #[test] + fn test_every_policy_charges_the_key() { + const NUM_ENTRIES: usize = 500; + const KEY_PADDING_LEN: usize = 8_000; + let capacity = ByteSize::mb(1); + let keys: Vec = (0..NUM_ENTRIES) + .map(|i| format!("{i:08}{}", "k".repeat(KEY_PADDING_LEN))) + .collect(); + // How many entries the budget can hold once keys are charged. + let expected_num_entries = + capacity.as_u64() as usize / (keys[0].mem_usage() + "0123456789abcdef".len()); + + for policy in [CachePolicy::Lru, CachePolicy::S3Fifo, CachePolicy::TinyLfu] { + let cache = MemorySizedCache::::from_config( + &CacheConfig::with_capacity_and_policy(capacity, policy), + &CACHE_METRICS_FOR_TESTS, + ); + for key in &keys { + cache.put(key.clone(), OwnedBytes::new(&b"0123456789abcdef"[..])); + } + + let num_entries = keys.iter().filter(|key| cache.get(*key).is_some()).count(); + assert_eq!( + num_entries, expected_num_entries, + "{policy:?} should have held only the entries whose keys and values fit in \ + {capacity}" + ); + } + } + #[test] fn test_cache_edge_unlimited_capacity() { let cache = MemorySizedCache::with_infinite_capacity(&CACHE_METRICS_FOR_TESTS); diff --git a/quickwit/quickwit-storage/src/cache/mod.rs b/quickwit/quickwit-storage/src/cache/mod.rs index 6a2bf38abf8..1b6079cb32a 100644 --- a/quickwit/quickwit-storage/src/cache/mod.rs +++ b/quickwit/quickwit-storage/src/cache/mod.rs @@ -14,6 +14,7 @@ mod base_cache; mod byte_range_cache; +mod mem_usage; mod memory_sized_cache; mod quickwit_cache; mod slice_address; @@ -29,6 +30,7 @@ pub use quickwit_cache::QuickwitCache; pub use storage_with_cache::StorageWithCache; pub use self::byte_range_cache::{ByteRangeCache, FileByteRangeCache}; +pub use self::mem_usage::MemUsage; pub use self::memory_sized_cache::MemorySizedCache; use crate::{OwnedBytes, Storage}; diff --git a/quickwit/quickwit-storage/src/cache/slice_address.rs b/quickwit/quickwit-storage/src/cache/slice_address.rs index b64e59d2148..d477dfba0eb 100644 --- a/quickwit/quickwit-storage/src/cache/slice_address.rs +++ b/quickwit/quickwit-storage/src/cache/slice_address.rs @@ -17,12 +17,20 @@ use std::hash::{Hash, Hasher}; use std::ops::Range; use std::path::{Path, PathBuf}; +use crate::cache::mem_usage::MemUsage; + #[derive(Hash, Clone, Debug, Eq, PartialEq)] pub struct SliceAddress { pub path: PathBuf, pub byte_range: Range, } +impl MemUsage for SliceAddress { + fn heap_mem_usage(&self) -> usize { + self.path.heap_mem_usage() + } +} + // ------------------------------------------------------------ // The following struct exists to make it possible to // fetch a slice from a cache without cloning PathBuf. diff --git a/quickwit/quickwit-storage/src/cache/stored_item.rs b/quickwit/quickwit-storage/src/cache/stored_item.rs index 8d1af95c8fd..42b303f19bb 100644 --- a/quickwit/quickwit-storage/src/cache/stored_item.rs +++ b/quickwit/quickwit-storage/src/cache/stored_item.rs @@ -15,18 +15,57 @@ use tantivy::directory::OwnedBytes; use tokio::time::Instant; +/// A cached value together with the memory charged for the key it is stored under. +/// +/// The key itself lives in the backing cache's own map, out of reach of the value, so every +/// backend memorizes the key footprint at insertion time rather than recomputing it from the key +/// on the way out. That serves two purposes: it keeps `MemUsage::mem_usage` off the eviction path, +/// and it guarantees an entry is debited exactly what it was credited — any divergence would make +/// a cache's `num_bytes` counter drift, and underflow on the way down. +#[derive(Clone)] +pub(super) struct KeyedEntry { + value: V, + key_mem_usage: usize, +} + +impl KeyedEntry { + pub fn new(value: V, key_mem_usage: usize) -> Self { + KeyedEntry { + value, + key_mem_usage, + } + } + + pub fn value(&self) -> &V { + &self.value + } +} + +impl KeyedEntry { + /// Number of bytes of payload, as served to a caller on a cache hit. + pub fn payload_num_bytes(&self) -> usize { + self.value.len() + } + + /// Number of bytes this entry occupies, key included. This is what a cache capacity is + /// enforced on. + pub fn entry_num_bytes(&self) -> usize { + self.key_mem_usage + self.value.len() + } +} + /// It is a bit overkill to put this in its own module, but I /// wanted to ensure that no one would access payload without updating `last_access_time`. pub(super) struct StoredItem { last_access_time: Instant, - payload: V, + entry: KeyedEntry, } impl StoredItem { - pub fn new(payload: V, now: Instant) -> Self { + pub fn new(payload: V, key_mem_usage: usize, now: Instant) -> Self { StoredItem { last_access_time: now, - payload, + entry: KeyedEntry::new(payload, key_mem_usage), } } } @@ -34,11 +73,18 @@ impl StoredItem { impl StoredItem { pub fn payload(&mut self) -> V { self.last_access_time = Instant::now(); - self.payload.clone() + self.entry.value().clone() + } + + /// Number of bytes of payload, as served to a caller on a cache hit. + pub fn payload_num_bytes(&self) -> usize { + self.entry.payload_num_bytes() } - pub fn len(&self) -> usize { - self.payload.len() + /// Number of bytes this entry occupies, key included. This is what the cache capacity is + /// enforced on. + pub fn entry_num_bytes(&self) -> usize { + self.entry.entry_num_bytes() } pub fn last_access_time(&self) -> Instant { diff --git a/quickwit/quickwit-storage/src/lib.rs b/quickwit/quickwit-storage/src/lib.rs index 73ef173eee2..4effe97f76e 100644 --- a/quickwit/quickwit-storage/src/lib.rs +++ b/quickwit/quickwit-storage/src/lib.rs @@ -66,7 +66,7 @@ pub use self::bundle_storage::{ #[cfg(any(test, feature = "testsuite"))] pub use self::cache::MockStorageCache; pub use self::cache::{ - ByteRangeCache, FileByteRangeCache, MemorySizedCache, QuickwitCache, StorageCache, + ByteRangeCache, FileByteRangeCache, MemUsage, MemorySizedCache, QuickwitCache, StorageCache, wrap_storage_with_cache, }; pub use self::counting_storage::{CountingStorage, DownloadCounters};