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));