From 5a1dce5e3d96d70ad4146871f37d0b3a7c851089 Mon Sep 17 00:00:00 2001 From: Vlad Lazar Date: Wed, 17 Apr 2024 14:14:58 +0100 Subject: [PATCH] pageserver: fix vectored read path cache handling --- libs/pageserver_api/src/keyspace.rs | 5 ++ pageserver/src/tenant/storage_layer.rs | 53 +++++++++++++++++-- .../src/tenant/storage_layer/delta_layer.rs | 12 +++-- .../tenant/storage_layer/inmemory_layer.rs | 2 + 4 files changed, 65 insertions(+), 7 deletions(-) diff --git a/libs/pageserver_api/src/keyspace.rs b/libs/pageserver_api/src/keyspace.rs index 78e4a3d735..da1313090f 100644 --- a/libs/pageserver_api/src/keyspace.rs +++ b/libs/pageserver_api/src/keyspace.rs @@ -178,6 +178,11 @@ impl KeySpace { pub fn overlaps(&self, range: &Range) -> bool { self.overlaps_at(range).is_some() } + + /// Check if the keyspace contains a key + pub fn contains(&self, key: &Key) -> bool { + self.overlaps(&(*key..key.next())) + } } /// diff --git a/pageserver/src/tenant/storage_layer.rs b/pageserver/src/tenant/storage_layer.rs index 9a2b086828..34c22ce961 100644 --- a/pageserver/src/tenant/storage_layer.rs +++ b/pageserver/src/tenant/storage_layer.rs @@ -138,6 +138,29 @@ impl ValuesReconstructState { } } + /// This function is called after reading a keyspace from a layer. + /// It checks if the read path has now moved past the cached Lsn for any keys. + /// + /// Implementation note: We intentionally iterate over the keys for which we've + /// already collected some reconstruct data. This avoids scaling complexity with + /// the size of the search space. + pub(crate) fn on_lsn_advanced(&mut self, keyspace: &KeySpace, advanced_to: Lsn) { + for (key, value) in self.keys.iter_mut() { + if !keyspace.contains(key) { + continue; + } + + if let Ok(state) = value { + if state.situation != ValueReconstructSituation::Complete + && state.get_cached_lsn() >= Some(advanced_to) + { + state.situation = ValueReconstructSituation::Complete; + self.keys_done.add_key(*key); + } + } + } + } + /// Update the state collected for a given key. /// Returns true if this was the last value needed for the key and false otherwise. /// @@ -162,11 +185,18 @@ impl ValuesReconstructState { true } Value::WalRecord(rec) => { - let reached_cache = - state.get_cached_lsn().map(|clsn| clsn + 1) == Some(lsn); + debug_assert!( + Some(lsn) > state.get_cached_lsn(), + "Attempt to collect a record below cached LSN for walredo: {} < {}", + lsn, + state + .get_cached_lsn() + .expect("Assertion can only fire if a cached lsn is present") + ); + let will_init = rec.will_init(); state.records.push((lsn, rec)); - will_init || reached_cache + will_init } }, }; @@ -349,11 +379,28 @@ impl ReadableLayer { ) -> Result<(), GetVectoredError> { match self { ReadableLayer::PersistentLayer(layer) => { + if keyspace.total_size() == 1 { + tracing::info!( + "Vectored path for {}: {} @ {:?}", + keyspace.start().unwrap(), + layer.local_path(), + lsn_range + ); + } + layer .get_values_reconstruct_data(keyspace, lsn_range, reconstruct_state, ctx) .await } ReadableLayer::InMemoryLayer(layer) => { + if keyspace.total_size() == 1 { + tracing::info!( + "Vectored path for {}: in mem layer @ {:?}", + keyspace.start().unwrap(), + lsn_range + ); + } + layer .get_values_reconstruct_data(keyspace, lsn_range.end, reconstruct_state, ctx) .await diff --git a/pageserver/src/tenant/storage_layer/delta_layer.rs b/pageserver/src/tenant/storage_layer/delta_layer.rs index c5b5e5c98f..5ed204c103 100644 --- a/pageserver/src/tenant/storage_layer/delta_layer.rs +++ b/pageserver/src/tenant/storage_layer/delta_layer.rs @@ -217,6 +217,7 @@ pub struct DeltaLayerInner { // values copied from summary index_start_blk: u32, index_root_blk: u32, + lsn_range: Range, file: VirtualFile, file_id: FileId, @@ -742,6 +743,7 @@ impl DeltaLayerInner { file_id, index_start_blk: actual_summary.index_start_blk, index_root_blk: actual_summary.index_root_blk, + lsn_range: actual_summary.lsn_range, max_vectored_read_bytes, })) } @@ -866,7 +868,7 @@ impl DeltaLayerInner { let data_end_offset = self.index_start_offset(); let reads = Self::plan_reads( - keyspace, + &keyspace, lsn_range, data_end_offset, index_reader, @@ -880,11 +882,13 @@ impl DeltaLayerInner { self.do_reads_and_update_state(reads, reconstruct_state) .await; + reconstruct_state.on_lsn_advanced(&keyspace, self.lsn_range.start); + Ok(()) } async fn plan_reads( - keyspace: KeySpace, + keyspace: &KeySpace, lsn_range: Range, data_end_offset: u64, index_reader: DiskBtreeReader, @@ -1532,7 +1536,7 @@ mod test { // Plan and validate let vectored_reads = DeltaLayerInner::plan_reads( - keyspace.clone(), + &keyspace, lsn_range.clone(), disk_offset, reader, @@ -1784,7 +1788,7 @@ mod test { let data_end_offset = inner.index_start_blk as u64 * PAGE_SZ as u64; let vectored_reads = DeltaLayerInner::plan_reads( - keyspace.clone(), + &keyspace, entries_meta.lsn_range.clone(), data_end_offset, index_reader, diff --git a/pageserver/src/tenant/storage_layer/inmemory_layer.rs b/pageserver/src/tenant/storage_layer/inmemory_layer.rs index 29751641b4..e851a72cfd 100644 --- a/pageserver/src/tenant/storage_layer/inmemory_layer.rs +++ b/pageserver/src/tenant/storage_layer/inmemory_layer.rs @@ -426,6 +426,8 @@ impl InMemoryLayer { } } + reconstruct_state.on_lsn_advanced(&keyspace, self.start_lsn); + Ok(()) } }