From 0ad736e2a6fb0344c9fbf3332729c58b9e936f38 Mon Sep 17 00:00:00 2001 From: Vlad Lazar Date: Thu, 10 Oct 2024 19:11:01 +0200 Subject: [PATCH] pageserver: clean up comms between pending IOs and awaiter Previously, each pending IO sent a stupid buffer which was just what it read from the layer file for the key. This made the awaiter code confusing because on disk images in layer files don't keep the enum wrapper, but the ones in delta layers do. This commit introduces a type to make this a bit easier and cleans up the IO awaiting code a bit. We also avoid some rather silly serialize, deserialize dance. --- pageserver/src/tenant/storage_layer.rs | 82 ++++++++----------- .../src/tenant/storage_layer/delta_layer.rs | 7 +- .../src/tenant/storage_layer/image_layer.rs | 13 ++- .../tenant/storage_layer/inmemory_layer.rs | 6 +- pageserver/src/tenant/timeline.rs | 4 +- 5 files changed, 52 insertions(+), 60 deletions(-) diff --git a/pageserver/src/tenant/storage_layer.rs b/pageserver/src/tenant/storage_layer.rs index 5534239614..e75076f158 100644 --- a/pageserver/src/tenant/storage_layer.rs +++ b/pageserver/src/tenant/storage_layer.rs @@ -79,40 +79,43 @@ pub(crate) enum ValueReconstructSituation { Continue, } +/// On disk representation of a value loaded in a buffer +#[derive(Debug)] +pub(crate) enum OnDiskValue { + /// Unencoded [`Value::Image`] + RawImage(Bytes), + /// Encoded [`Value`]. Can deserialize into an image or a WAL record + WalRecordOrImage(Bytes), +} + /// Reconstruct data accumulated for a single key during a vectored get #[derive(Debug, Default)] pub(crate) struct VectoredValueReconstructState { - pub(crate) records: Vec<( + pub(crate) on_disk_values: Vec<( Lsn, - tokio::sync::oneshot::Receiver>, - )>, - pub(crate) img: Option<( - Lsn, - tokio::sync::oneshot::Receiver>, + tokio::sync::oneshot::Receiver>, )>, pub(crate) situation: ValueReconstructSituation, } impl VectoredValueReconstructState { - fn get_cached_lsn(&self) -> Option { - self.img.as_ref().map(|img| img.0) - } -} + pub(crate) async fn collect_pending_ios( + self, + ) -> Result { + let mut to = ValueReconstructState::default(); -pub(crate) async fn convert( - _key: Key, - from: VectoredValueReconstructState, -) -> Result { - let mut to = ValueReconstructState::default(); - - for (lsn, fut) in from.records { - match fut.await { - Ok(res) => match res { - Ok(bytes) => { - let value = Value::des(&bytes) - .map_err(|err| PageReconstructError::Other(err.into()))?; + for (lsn, fut) in self.on_disk_values { + // TODO: IO futures are not failable - we could expect + let res = fut + .await + .map_err(|err| PageReconstructError::Other(err.into()))?; + let on_disk_value = res.map_err(|err| PageReconstructError::Other(err.into()))?; + match on_disk_value { + OnDiskValue::WalRecordOrImage(buf) => { + let value = + Value::des(&buf).map_err(|err| PageReconstructError::Other(err.into()))?; match value { Value::WalRecord(rec) => { to.records.push((lsn, rec)); @@ -123,34 +126,21 @@ pub(crate) async fn convert( } } } - Err(err) => { - return Err(PageReconstructError::Other(err.into())); + OnDiskValue::RawImage(img) => { + assert!(to.img.is_none()); + to.img = Some((lsn, img)); } - }, - Err(err) => { - return Err(PageReconstructError::Other(err.into())); } } + + Ok(to) } - if to.img.is_none() && from.img.is_some() { - let (lsn, fut) = from.img.expect("Has an image"); - match fut.await { - Ok(res) => match res { - Ok(bytes) => { - to.img = Some((lsn, bytes)); - } - Err(err) => { - return Err(PageReconstructError::Other(err.into())); - } - }, - Err(err) => { - return Err(PageReconstructError::Other(err.into())); - } - } + fn get_cached_lsn(&self) -> Option { + // self.img.as_ref().map(|img| img.0) + // TODO: rip this out + Some(Lsn(0)) } - - Ok(to) } /// Bag of data accumulated during a vectored get.. @@ -291,7 +281,7 @@ impl ValuesReconstructState { key: &Key, lsn: Lsn, completes: bool, - value: tokio::sync::oneshot::Receiver>, + value: tokio::sync::oneshot::Receiver>, ) -> ValueReconstructSituation { let state = self .keys @@ -302,7 +292,7 @@ impl ValuesReconstructState { match state.situation { ValueReconstructSituation::Complete => unreachable!(), ValueReconstructSituation::Continue => { - state.records.push((lsn, value)); + state.on_disk_values.push((lsn, value)); } } diff --git a/pageserver/src/tenant/storage_layer/delta_layer.rs b/pageserver/src/tenant/storage_layer/delta_layer.rs index a2d8da35c1..03179414de 100644 --- a/pageserver/src/tenant/storage_layer/delta_layer.rs +++ b/pageserver/src/tenant/storage_layer/delta_layer.rs @@ -77,7 +77,7 @@ use utils::{ lsn::Lsn, }; -use super::{AsLayerDesc, LayerName, PersistentLayerDesc, ValuesReconstructState}; +use super::{AsLayerDesc, LayerName, OnDiskValue, PersistentLayerDesc, ValuesReconstructState}; /// /// Header stored in the beginning of the file @@ -1011,7 +1011,7 @@ impl DeltaLayerInner { for read in reads.into_iter().rev() { let mut senders: HashMap< (Key, Lsn), - sync::oneshot::Sender>, + sync::oneshot::Sender>, > = Default::default(); for (_, blob_meta) in read.blobs_at.as_slice().iter().rev() { let (tx, rx) = sync::oneshot::channel(); @@ -1052,7 +1052,8 @@ impl DeltaLayerInner { } }; - let _ = sender.send(Ok(blob_read.into_bytes())); + let _ = sender + .send(Ok(OnDiskValue::WalRecordOrImage(blob_read.into_bytes()))); } assert!(senders.is_empty()); diff --git a/pageserver/src/tenant/storage_layer/image_layer.rs b/pageserver/src/tenant/storage_layer/image_layer.rs index bf3e43c5d5..4ea2dca06d 100644 --- a/pageserver/src/tenant/storage_layer/image_layer.rs +++ b/pageserver/src/tenant/storage_layer/image_layer.rs @@ -74,7 +74,7 @@ use utils::{ }; use super::layer_name::ImageLayerName; -use super::{AsLayerDesc, LayerName, PersistentLayerDesc, ValuesReconstructState}; +use super::{AsLayerDesc, LayerName, OnDiskValue, PersistentLayerDesc, ValuesReconstructState}; /// /// Header stored in the beginning of the file @@ -584,8 +584,10 @@ impl ImageLayerInner { .into(); for read in reads.into_iter() { - let mut senders: HashMap<(Key, Lsn), oneshot::Sender>> = - Default::default(); + let mut senders: HashMap< + (Key, Lsn), + oneshot::Sender>, + > = Default::default(); for (_, blob_meta) in read.blobs_at.as_slice() { let (tx, rx) = oneshot::channel(); senders.insert((blob_meta.key, blob_meta.lsn), tx); @@ -644,10 +646,7 @@ impl ImageLayerInner { } }; - // TODO: this is silly - sort it out - let image_serialized = Value::ser(&Value::Image(img_buf.into_bytes())) - .expect("stupid but correct"); - let _ = sender.send(Ok(image_serialized.into())); + let _ = sender.send(Ok(OnDiskValue::RawImage(img_buf.into_bytes()))); } assert!(senders.is_empty()); diff --git a/pageserver/src/tenant/storage_layer/inmemory_layer.rs b/pageserver/src/tenant/storage_layer/inmemory_layer.rs index 473e7ddfac..cc7284d5bb 100644 --- a/pageserver/src/tenant/storage_layer/inmemory_layer.rs +++ b/pageserver/src/tenant/storage_layer/inmemory_layer.rs @@ -8,6 +8,7 @@ use crate::assert_u64_eq_usize::{u64_to_usize, U64IsUsize, UsizeIsU64}; use crate::config::PageServerConf; use crate::context::{PageContentKind, RequestContext, RequestContextBuilder}; use crate::tenant::ephemeral_file::EphemeralFile; +use crate::tenant::storage_layer::OnDiskValue; use crate::tenant::timeline::GetVectoredError; use crate::virtual_file::owned_buffers_io::io_buf_ext::IoBufExt; use crate::{l0_flush, page_cache}; @@ -435,7 +436,7 @@ impl InMemoryLayer { let mut reads: HashMap> = HashMap::new(); let mut senders: HashMap< (Key, Lsn), - tokio::sync::oneshot::Sender>, + tokio::sync::oneshot::Sender>, > = Default::default(); for range in keyspace.ranges.iter() { @@ -501,7 +502,8 @@ impl InMemoryLayer { .send(Err(std::io::Error::new(e.kind(), "dio vec read failed"))); } Ok(value_buf) => { - let _ = sender.send(Ok(value_buf.into())); + let _ = + sender.send(Ok(OnDiskValue::WalRecordOrImage(value_buf.into()))); } } } diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index 0474dc79e0..6db1107bb2 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -73,7 +73,7 @@ use crate::{ layer_map::{LayerMap, SearchResult}, metadata::TimelineMetadata, storage_layer::{ - convert, inmemory_layer::IndexEntry, PersistentLayerDesc, ValueReconstructSituation, + inmemory_layer::IndexEntry, PersistentLayerDesc, ValueReconstructSituation, }, }, walingest::WalLagCooldown, @@ -1152,7 +1152,7 @@ impl Timeline { ValueReconstructSituation::Complete )); - let converted = match convert(key, state).await { + let converted = match state.collect_pending_ios().await { Ok(ok) => ok, Err(err) => { return (key, Err(err));