rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-10-26 21:46:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

pageserver: clean up comms between pending IOs and awaiter

Browse Source 
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.
This commit is contained in:
Vlad Lazar
2024-11-04 14:07:20 +01:00
parent bde1694cb9
commit 0ad736e2a6
5 changed files with 52 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pageserver/src/tenant/storage_layer.rs
+36 -46
View File
@@ -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<Result<Bytes, std::io::Error>>,
)>,
pub(crate) img: Option<(
Lsn,
tokio::sync::oneshot::Receiver<Result<Bytes, std::io::Error>>,
tokio::sync::oneshot::Receiver<Result<OnDiskValue, std::io::Error>>,
)>,
pub(crate) situation: ValueReconstructSituation,
}
impl VectoredValueReconstructState {
fn get_cached_lsn(&self) -> Option<Lsn> {
self.img.as_ref().map(|img| img.0)
}
}
pub(crate) async fn collect_pending_ios(
self,
) -> Result<ValueReconstructState, PageReconstructError> {
let mut to = ValueReconstructState::default();
pub(crate) async fn convert(
_key: Key,
from: VectoredValueReconstructState,
) -> Result<ValueReconstructState, PageReconstructError> {
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<Lsn> {
// 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<Result<Bytes, std::io::Error>>,
value: tokio::sync::oneshot::Receiver<Result<OnDiskValue, std::io::Error>>,
) -> 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));
}
}
pageserver/src/tenant/storage_layer/delta_layer.rs
+4 -3
View File
@@ -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<Result<Bytes, std::io::Error>>,
sync::oneshot::Sender<Result<OnDiskValue, std::io::Error>>,
> = 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());
pageserver/src/tenant/storage_layer/image_layer.rs
+6 -7
View File
@@ -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<Result<Bytes, std::io::Error>>> =
Default::default();
let mut senders: HashMap<
(Key, Lsn),
oneshot::Sender<Result<OnDiskValue, std::io::Error>>,
> = 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());
pageserver/src/tenant/storage_layer/inmemory_layer.rs
+4 -2
View File
@@ -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<Key, Vec<ValueRead>> = HashMap::new();
let mut senders: HashMap<
(Key, Lsn),
tokio::sync::oneshot::Sender<Result<Bytes, std::io::Error>>,
tokio::sync::oneshot::Sender<Result<OnDiskValue, std::io::Error>>,
> = 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())));
}
}
}
pageserver/src/tenant/timeline.rs
+2 -2
View File
@@ -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));