Journal fixed lru cache #498
Conversation
Codecov ReportAttention: Patch coverage is
@@ Coverage Diff @@
## main #498 +/- ##
==========================================
+ Coverage 94.34% 94.37% +0.02%
==========================================
Files 109 109
Lines 30217 30282 +65
==========================================
+ Hits 28509 28579 +70
+ Misses 1708 1703 -5
... and 2 files with indirect coverage changes Continue to review full report in Codecov by Sentry.
|
| //blobs: BTreeMap<u64, B>, | ||
| min_blob_index: u64, | ||
| newest_index_blob: (u64, B), | ||
| lru_cache: Arc<tokio::sync::Mutex<LruCache<u64, B>>>, |
There was a problem hiding this comment.
Why does this need the Arc<> ? (& Mutex < > ?)
| @@ -84,7 +91,10 @@ pub struct Journal<B: Blob, E: Storage<B>, A: Array> { | |||
|
|
|||
| // Blobs are stored in a BTreeMap to ensure they are always iterated in order of their indices. | |||
| @@ -24,6 +24,8 @@ futures-util = { workspace = true } | |||
| tracing = { workspace = true } | |||
| zstd = { workspace = true } | |||
| rangemap = "1.5.1" | |||
| lru = "0.13.0" | |||
| tokio = { workspace = true} | |||
There was a problem hiding this comment.
I remember team said we should not import tokio. This is currently used to open the discussion.
When removing Arc<Mutex<>> implies to change the read(&self) trait method signature to accept read(&mut self), this creates chained changes and not sure this is expected.
There was a problem hiding this comment.
Ah I see, this introduces an internal mutability concern. I think we can just change the API to read(&mut self) for now?
There was a problem hiding this comment.
That is correct. We shouldn't be importing tokio here.
We have a few options:
- make
read(&mut self)(gross) - use a cache that doesn't require a mutable reference (either external crate or write our own)
- refactor code to avoid holding cache references over
await
If we aren't thoughtful with this cache integration, we could accidentally prohibit concurrent reads (even if we keep read(&self) and use a cache with interior mutability, it may lock on all requests to update the cache state).
A concurrent_lru could look something like this (we shouldn't use this crate but to give you some idea of a more compatible interface with our goals): https://docs.rs/concurrent_lru/latest/concurrent_lru/.
| } | ||
| let items_per_blob = self.cfg.items_per_blob; | ||
| let blob = self.get_blob(blob_index).await?; | ||
| let item_index = item_position % items_per_blob; |
There was a problem hiding this comment.
| let item_index = item_position % items_per_blob; | |
| let item_index = item_position % self.cfg.items_per_blob; |
then drop L309
| let offset = item_index * Self::CHUNK_SIZE as u64; | ||
| let mut buf = vec![0u8; Self::CHUNK_SIZE]; | ||
| blob.read_at(&mut buf, offset).await?; | ||
|
|
||
| // Verify integrity |
There was a problem hiding this comment.
while you're here can you delete this comment which is simply stating the obvious?
| @@ -37,4 +37,6 @@ pub enum Error { | |||
| InvalidItem(u64), | |||
| #[error("invalid rewind: {0}")] | |||
| InvalidRewind(u64), | |||
| #[error("invalid index: {0}")] | |||
| InvalidIndex(u64), | |||
There was a problem hiding this comment.
I wouldn't expose the fact that we index blobs using integers to the external API like this.
Closes #452
This makes changes to use LRU Cache with fixed length item journal.