From e3aa85d722efd92070b824d00be3460669e3e32a Mon Sep 17 00:00:00 2001 From: Vlad Lazar Date: Thu, 10 Oct 2024 17:51:04 +0200 Subject: [PATCH] pagserver: do concurrent IO on the read path Previously, the read path would wait for all IO in one layer visit to complete before visiting the next layer (if a subsequent visit is required). IO within one layer visit was also sequential. With this patch we gain the ability to issue IO concurrently within one layer visit **and** to move on to the next layer without waiting for IOs from the previous visit to complete. This is a slightly cleaned up version of the work done at the Lisbon hackathon. --- pageserver/src/tenant/storage_layer.rs | 156 +++++++++++++---- .../src/tenant/storage_layer/delta_layer.rs | 158 ++++++++---------- .../src/tenant/storage_layer/image_layer.rs | 105 ++++++------ .../tenant/storage_layer/inmemory_layer.rs | 110 ++++++------ pageserver/src/tenant/timeline.rs | 87 ++++++---- pageserver/src/tenant/vectored_blob_io.rs | 48 ++++-- 6 files changed, 402 insertions(+), 262 deletions(-) diff --git a/pageserver/src/tenant/storage_layer.rs b/pageserver/src/tenant/storage_layer.rs index 8f4219bbbc..f52eee9904 100644 --- a/pageserver/src/tenant/storage_layer.rs +++ b/pageserver/src/tenant/storage_layer.rs @@ -16,12 +16,13 @@ use pageserver_api::key::Key; use pageserver_api::keyspace::{KeySpace, KeySpaceRandomAccum}; use pageserver_api::record::NeonWalRecord; use pageserver_api::value::Value; -use std::cmp::{Ordering, Reverse}; +use std::cmp::Ordering; use std::collections::hash_map::Entry; use std::collections::{BinaryHeap, HashMap}; use std::ops::Range; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use utils::bin_ser::BeSer; use utils::lsn::Lsn; @@ -79,12 +80,18 @@ pub(crate) enum ValueReconstructSituation { } /// Reconstruct data accumulated for a single key during a vectored get -#[derive(Debug, Default, Clone)] +#[derive(Debug, Default)] pub(crate) struct VectoredValueReconstructState { - pub(crate) records: Vec<(Lsn, NeonWalRecord)>, - pub(crate) img: Option<(Lsn, Bytes)>, + pub(crate) records: Vec<( + Lsn, + tokio::sync::oneshot::Receiver>, + )>, + pub(crate) img: Option<( + Lsn, + tokio::sync::oneshot::Receiver>, + )>, - situation: ValueReconstructSituation, + pub(crate) situation: ValueReconstructSituation, } impl VectoredValueReconstructState { @@ -93,16 +100,57 @@ impl VectoredValueReconstructState { } } -impl From for ValueReconstructState { - fn from(mut state: VectoredValueReconstructState) -> Self { - // walredo expects the records to be descending in terms of Lsn - state.records.sort_by_key(|(lsn, _)| Reverse(*lsn)); +pub(crate) async fn convert( + _key: Key, + from: VectoredValueReconstructState, +) -> Result { + let mut to = ValueReconstructState::default(); - ValueReconstructState { - records: state.records, - img: state.img, + 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()))?; + + match value { + Value::WalRecord(rec) => { + to.records.push((lsn, rec)); + } + Value::Image(img) => { + assert!(to.img.is_none()); + to.img = Some((lsn, img)); + } + } + } + Err(err) => { + return Err(PageReconstructError::Other(err.into())); + } + }, + Err(err) => { + return Err(PageReconstructError::Other(err.into())); + } } } + + 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())); + } + } + } + + Ok(to) } /// Bag of data accumulated during a vectored get.. @@ -119,6 +167,37 @@ pub(crate) struct ValuesReconstructState { // Statistics that are still accessible as a caller of `get_vectored_impl`. layers_visited: u32, delta_layers_visited: u32, + + io_concurrency: IoConcurrency, +} + +enum IoConcurrency { + Serial { + prev_io: Option>, + }, + Parallel, +} + +impl IoConcurrency { + pub(crate) fn spawn_io(&mut self, fut: F) + where + F: std::future::Future + Send + 'static, + { + match self { + IoConcurrency::Serial { prev_io } => { + let prev = prev_io.take(); + *prev_io = Some(tokio::spawn(async move { + if let Some(prev) = prev { + prev.await.unwrap(); + } + fut.await; + })); + } + IoConcurrency::Parallel => { + tokio::spawn(fut); + } + } + } } impl ValuesReconstructState { @@ -129,9 +208,30 @@ impl ValuesReconstructState { keys_with_image_coverage: None, layers_visited: 0, delta_layers_visited: 0, + io_concurrency: { + static IO_CONCURRENCY: once_cell::sync::Lazy = + once_cell::sync::Lazy::new(|| { + std::env::var("NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY").unwrap() + }); + match IO_CONCURRENCY.as_str() { + "parallel" => IoConcurrency::Parallel, + "serial" => IoConcurrency::Serial { prev_io: None }, + x => panic!( + "Invalid value for NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY: {}", + x + ), + } + }, } } + pub(crate) fn spawn_io(&mut self, fut: F) + where + F: std::future::Future + Send + 'static, + { + self.io_concurrency.spawn_io(fut); + } + /// Associate a key with the error which it encountered and mark it as done pub(crate) fn on_key_error(&mut self, key: Key, err: PageReconstructError) { let previous = self.keys.insert(key, Err(err)); @@ -200,7 +300,8 @@ impl ValuesReconstructState { &mut self, key: &Key, lsn: Lsn, - value: Value, + completes: bool, + value: tokio::sync::oneshot::Receiver>, ) -> ValueReconstructSituation { let state = self .keys @@ -208,31 +309,14 @@ impl ValuesReconstructState { .or_insert(Ok(VectoredValueReconstructState::default())); if let Ok(state) = state { - let key_done = match state.situation { + match state.situation { ValueReconstructSituation::Complete => unreachable!(), - ValueReconstructSituation::Continue => match value { - Value::Image(img) => { - state.img = Some((lsn, img)); - true - } - Value::WalRecord(rec) => { - 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") - ); + ValueReconstructSituation::Continue => { + state.records.push((lsn, value)); + } + } - let will_init = rec.will_init(); - state.records.push((lsn, rec)); - will_init - } - }, - }; - - if key_done && state.situation == ValueReconstructSituation::Continue { + if completes && state.situation == ValueReconstructSituation::Continue { state.situation = ValueReconstructSituation::Complete; self.keys_done.add_key(*key); } diff --git a/pageserver/src/tenant/storage_layer/delta_layer.rs b/pageserver/src/tenant/storage_layer/delta_layer.rs index 664c00a6b1..a2d8da35c1 100644 --- a/pageserver/src/tenant/storage_layer/delta_layer.rs +++ b/pageserver/src/tenant/storage_layer/delta_layer.rs @@ -41,13 +41,13 @@ use crate::tenant::vectored_blob_io::{ BlobFlag, BufView, StreamingVectoredReadPlanner, VectoredBlobReader, VectoredRead, VectoredReadPlanner, }; -use crate::tenant::PageReconstructError; use crate::virtual_file::owned_buffers_io::io_buf_ext::{FullSlice, IoBufExt}; use crate::virtual_file::IoBufferMut; use crate::virtual_file::{self, MaybeFatalIo, VirtualFile}; use crate::TEMP_FILE_SUFFIX; use crate::{DELTA_FILE_MAGIC, STORAGE_FORMAT_VERSION}; -use anyhow::{anyhow, bail, ensure, Context, Result}; +use anyhow::{bail, ensure, Context, Result}; +use bytes::Bytes; use camino::{Utf8Path, Utf8PathBuf}; use futures::StreamExt; use itertools::Itertools; @@ -60,14 +60,14 @@ use pageserver_api::shard::TenantShardId; use pageserver_api::value::Value; use rand::{distributions::Alphanumeric, Rng}; use serde::{Deserialize, Serialize}; -use std::collections::VecDeque; +use std::collections::{HashMap, VecDeque}; use std::fs::File; use std::io::SeekFrom; use std::ops::Range; use std::os::unix::fs::FileExt; use std::str::FromStr; use std::sync::Arc; -use tokio::sync::OnceCell; +use tokio::sync::{self, OnceCell}; use tokio_epoll_uring::IoBuf; use tracing::*; @@ -226,7 +226,7 @@ pub struct DeltaLayerInner { index_start_blk: u32, index_root_blk: u32, - file: VirtualFile, + file: Arc, file_id: FileId, layer_key_range: Range, @@ -791,9 +791,11 @@ impl DeltaLayerInner { max_vectored_read_bytes: Option, ctx: &RequestContext, ) -> anyhow::Result { - let file = VirtualFile::open_v2(path, ctx) - .await - .context("open layer file")?; + let file = Arc::new( + VirtualFile::open_v2(path, ctx) + .await + .context("open layer file")?, + ); let file_id = page_cache::next_file_id(); @@ -994,94 +996,75 @@ impl DeltaLayerInner { reconstruct_state: &mut ValuesReconstructState, ctx: &RequestContext, ) { - let vectored_blob_reader = VectoredBlobReader::new(&self.file); - let mut ignore_key_with_err = None; - let max_vectored_read_bytes = self .max_vectored_read_bytes .expect("Layer is loaded with max vectored bytes config") .0 .into(); let buf_size = Self::get_min_read_buffer_size(&reads, max_vectored_read_bytes); - let mut buf = Some(IoBufferMut::with_capacity(buf_size)); + + let mut ignore_key_with_err = None; // Note that reads are processed in reverse order (from highest key+lsn). // This is the order that `ReconstructState` requires such that it can // track when a key is done. for read in reads.into_iter().rev() { - let res = vectored_blob_reader - .read_blobs(&read, buf.take().expect("Should have a buffer"), ctx) - .await; - - let blobs_buf = match res { - Ok(blobs_buf) => blobs_buf, - Err(err) => { - let kind = err.kind(); - for (_, blob_meta) in read.blobs_at.as_slice() { - reconstruct_state.on_key_error( - blob_meta.key, - PageReconstructError::Other(anyhow!( - "Failed to read blobs from virtual file {}: {}", - self.file.path(), - kind - )), - ); - } - - // We have "lost" the buffer since the lower level IO api - // doesn't return the buffer on error. Allocate a new one. - buf = Some(IoBufferMut::with_capacity(buf_size)); - - continue; - } - }; - let view = BufView::new_slice(&blobs_buf.buf); - for meta in blobs_buf.blobs.iter().rev() { - if Some(meta.meta.key) == ignore_key_with_err { - continue; - } - let blob_read = meta.read(&view).await; - let blob_read = match blob_read { - Ok(buf) => buf, - Err(e) => { - reconstruct_state.on_key_error( - meta.meta.key, - PageReconstructError::Other(anyhow!(e).context(format!( - "Failed to decompress blob from virtual file {}", - self.file.path(), - ))), - ); - - ignore_key_with_err = Some(meta.meta.key); - continue; - } - }; - - let value = Value::des(&blob_read); - - let value = match value { - Ok(v) => v, - Err(e) => { - reconstruct_state.on_key_error( - meta.meta.key, - PageReconstructError::Other(anyhow!(e).context(format!( - "Failed to deserialize blob from virtual file {}", - self.file.path(), - ))), - ); - - ignore_key_with_err = Some(meta.meta.key); - continue; - } - }; - - // Invariant: once a key reaches [`ValueReconstructSituation::Complete`] - // state, no further updates shall be made to it. The call below will - // panic if the invariant is violated. - reconstruct_state.update_key(&meta.meta.key, meta.meta.lsn, value); + let mut senders: HashMap< + (Key, Lsn), + sync::oneshot::Sender>, + > = Default::default(); + for (_, blob_meta) in read.blobs_at.as_slice().iter().rev() { + let (tx, rx) = sync::oneshot::channel(); + senders.insert((blob_meta.key, blob_meta.lsn), tx); + reconstruct_state.update_key( + &blob_meta.key, + blob_meta.lsn, + blob_meta.will_init, + rx, + ); } - buf = Some(blobs_buf.buf); + let read_from = self.file.clone(); + let read_ctx = ctx.attached_child(); + reconstruct_state.spawn_io(async move { + let vectored_blob_reader = VectoredBlobReader::new(&read_from); + let buf = IoBufferMut::with_capacity(buf_size); + + let res = vectored_blob_reader.read_blobs(&read, buf, &read_ctx).await; + match res { + Ok(blobs_buf) => { + let view = BufView::new_slice(&blobs_buf.buf); + for meta in blobs_buf.blobs.iter().rev() { + let sender = senders.remove(&(meta.meta.key, meta.meta.lsn)).unwrap(); + + if Some(meta.meta.key) == ignore_key_with_err { + continue; + } + + let blob_read = meta.read(&view).await; + let blob_read = match blob_read { + Ok(buf) => buf, + Err(e) => { + ignore_key_with_err = Some(meta.meta.key); + + let _ = sender.send(Err(e)); + continue; + } + }; + + let _ = sender.send(Ok(blob_read.into_bytes())); + } + + assert!(senders.is_empty()); + } + Err(err) => { + for (_, sender) in senders { + let _ = sender + .send(Err(std::io::Error::new(err.kind(), "vec read failed"))); + } + } + } + }); } } @@ -1220,7 +1203,14 @@ impl DeltaLayerInner { let actionable = if let Some((key, lsn, start_offset)) = prev.take() { let end_offset = offset; - Some((BlobMeta { key, lsn }, start_offset..end_offset)) + Some(( + BlobMeta { + key, + lsn, + will_init: false, + }, + start_offset..end_offset, + )) } else { None }; diff --git a/pageserver/src/tenant/storage_layer/image_layer.rs b/pageserver/src/tenant/storage_layer/image_layer.rs index 834d1931d0..db070d277b 100644 --- a/pageserver/src/tenant/storage_layer/image_layer.rs +++ b/pageserver/src/tenant/storage_layer/image_layer.rs @@ -21,7 +21,7 @@ //! //! Every image layer file consists of three parts: "summary", //! "index", and "values". The summary is a fixed size header at the -//! beginning of the file, and it contains basic information about the +//! beginningof the file, and it contains basic information about the //! layer, and offsets to the other parts. The "index" is a B-tree, //! mapping from Key to an offset in the "values" part. The //! actual page images are stored in the "values" part. @@ -38,12 +38,11 @@ use crate::tenant::vectored_blob_io::{ BlobFlag, BufView, StreamingVectoredReadPlanner, VectoredBlobReader, VectoredRead, VectoredReadPlanner, }; -use crate::tenant::PageReconstructError; use crate::virtual_file::owned_buffers_io::io_buf_ext::IoBufExt; use crate::virtual_file::IoBufferMut; use crate::virtual_file::{self, MaybeFatalIo, VirtualFile}; use crate::{IMAGE_FILE_MAGIC, STORAGE_FORMAT_VERSION, TEMP_FILE_SUFFIX}; -use anyhow::{anyhow, bail, ensure, Context, Result}; +use anyhow::{bail, ensure, Context, Result}; use bytes::Bytes; use camino::{Utf8Path, Utf8PathBuf}; use hex; @@ -56,12 +55,14 @@ use pageserver_api::shard::{ShardIdentity, TenantShardId}; use pageserver_api::value::Value; use rand::{distributions::Alphanumeric, Rng}; use serde::{Deserialize, Serialize}; -use std::collections::VecDeque; +use std::collections::{HashMap, VecDeque}; use std::fs::File; use std::io::SeekFrom; use std::ops::Range; use std::os::unix::prelude::FileExt; use std::str::FromStr; +use std::sync::Arc; +use tokio::sync::oneshot; use tokio::sync::OnceCell; use tokio_stream::StreamExt; use tracing::*; @@ -164,7 +165,7 @@ pub struct ImageLayerInner { key_range: Range, lsn: Lsn, - file: VirtualFile, + file: Arc, file_id: FileId, max_vectored_read_bytes: Option, @@ -391,9 +392,11 @@ impl ImageLayerInner { max_vectored_read_bytes: Option, ctx: &RequestContext, ) -> anyhow::Result { - let file = VirtualFile::open_v2(path, ctx) - .await - .context("open layer file")?; + let file = Arc::new( + VirtualFile::open_v2(path, ctx) + .await + .context("open layer file")?, + ); let file_id = page_cache::next_file_id(); let block_reader = FileBlockReader::new(&file, file_id); let summary_blk = block_reader @@ -580,8 +583,16 @@ impl ImageLayerInner { .0 .into(); - let vectored_blob_reader = VectoredBlobReader::new(&self.file); for read in reads.into_iter() { + 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); + + reconstruct_state.update_key(&blob_meta.key, blob_meta.lsn, true, rx); + } + let buf_size = read.size(); if buf_size > max_vectored_read_bytes { @@ -611,50 +622,48 @@ impl ImageLayerInner { } } - let buf = IoBufferMut::with_capacity(buf_size); - let res = vectored_blob_reader.read_blobs(&read, buf, ctx).await; + // Why did on_key_error go away? + // There was a good reason for it ... - match res { - Ok(blobs_buf) => { - let view = BufView::new_slice(&blobs_buf.buf); - for meta in blobs_buf.blobs.iter() { - let img_buf = meta.read(&view).await; + let read_from = self.file.clone(); + let read_ctx = ctx.attached_child(); + reconstruct_state.spawn_io(async move { + let buf = IoBufferMut::with_capacity(buf_size); + let vectored_blob_reader = VectoredBlobReader::new(&read_from); + let res = vectored_blob_reader.read_blobs(&read, buf, &read_ctx).await; - let img_buf = match img_buf { - Ok(img_buf) => img_buf, - Err(e) => { - reconstruct_state.on_key_error( - meta.meta.key, - PageReconstructError::Other(anyhow!(e).context(format!( - "Failed to decompress blob from virtual file {}", - self.file.path(), - ))), - ); + match res { + Ok(blobs_buf) => { + let view = BufView::new_slice(&blobs_buf.buf); + for meta in blobs_buf.blobs.iter() { + let sender = senders.remove(&(meta.meta.key, meta.meta.lsn)).unwrap(); + let img_buf = meta.read(&view).await; - continue; - } - }; - reconstruct_state.update_key( - &meta.meta.key, - self.lsn, - Value::Image(img_buf.into_bytes()), - ); + let img_buf = match img_buf { + Ok(img_buf) => img_buf, + Err(e) => { + let _ = sender.send(Err(e)); + continue; + } + }; + + // 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())); + } + + assert!(senders.is_empty()); + } + Err(err) => { + for (_, sender) in senders { + // TODO: when can this fail? + let _ = sender + .send(Err(std::io::Error::new(err.kind(), "vec read failed"))); + } } } - Err(err) => { - let kind = err.kind(); - for (_, blob_meta) in read.blobs_at.as_slice() { - reconstruct_state.on_key_error( - blob_meta.key, - PageReconstructError::from(anyhow!( - "Failed to read blobs from virtual file {}: {}", - self.file.path(), - kind - )), - ); - } - } - }; + }); } } diff --git a/pageserver/src/tenant/storage_layer/inmemory_layer.rs b/pageserver/src/tenant/storage_layer/inmemory_layer.rs index df448a0963..473e7ddfac 100644 --- a/pageserver/src/tenant/storage_layer/inmemory_layer.rs +++ b/pageserver/src/tenant/storage_layer/inmemory_layer.rs @@ -9,10 +9,9 @@ use crate::config::PageServerConf; use crate::context::{PageContentKind, RequestContext, RequestContextBuilder}; use crate::tenant::ephemeral_file::EphemeralFile; use crate::tenant::timeline::GetVectoredError; -use crate::tenant::PageReconstructError; use crate::virtual_file::owned_buffers_io::io_buf_ext::IoBufExt; use crate::{l0_flush, page_cache}; -use anyhow::{anyhow, Context, Result}; +use anyhow::{Context, Result}; use camino::Utf8PathBuf; use pageserver_api::key::CompactKey; use pageserver_api::key::Key; @@ -35,9 +34,7 @@ use std::sync::atomic::Ordering as AtomicOrdering; use std::sync::atomic::{AtomicU64, AtomicUsize}; use tokio::sync::RwLock; -use super::{ - DeltaLayerWriter, PersistentLayerDesc, ValueReconstructSituation, ValuesReconstructState, -}; +use super::{DeltaLayerWriter, PersistentLayerDesc, ValuesReconstructState}; pub(crate) mod vectored_dio_read; @@ -87,7 +84,7 @@ pub struct InMemoryLayerInner { /// The values are stored in a serialized format in this file. /// Each serialized Value is preceded by a 'u32' length field. /// PerSeg::page_versions map stores offsets into this file. - file: EphemeralFile, + file: Arc>, resource_units: GlobalResourceUnits, } @@ -381,7 +378,11 @@ impl InMemoryLayer { } pub(crate) fn try_len(&self) -> Option { - self.inner.try_read().map(|i| i.file.len()).ok() + self.inner + .try_read() + .map(|i| i.file.try_read().map(|i| i.len()).ok()) + .ok() + .flatten() } pub(crate) fn assert_writable(&self) { @@ -432,6 +433,10 @@ impl InMemoryLayer { read: vectored_dio_read::LogicalRead>, } let mut reads: HashMap> = HashMap::new(); + let mut senders: HashMap< + (Key, Lsn), + tokio::sync::oneshot::Sender>, + > = Default::default(); for range in keyspace.ranges.iter() { for (key, vec_map) in inner @@ -459,6 +464,11 @@ impl InMemoryLayer { Vec::with_capacity(len as usize), ), }); + + let (tx, rx) = tokio::sync::oneshot::channel(); + senders.insert((key, *entry_lsn), tx); + reconstruct_state.update_key(&key, *entry_lsn, will_init, rx); + if will_init { break; } @@ -466,46 +476,39 @@ impl InMemoryLayer { } } - // Execute the reads. + let read_from = inner.file.clone(); + let read_ctx = ctx.attached_child(); + reconstruct_state.spawn_io(async move { + let locked = read_from.read().await; + let f = vectored_dio_read::execute( + &*locked, + reads + .iter() + .flat_map(|(_, value_reads)| value_reads.iter().map(|v| &v.read)), + &read_ctx, + ); + send_future::SendFuture::send(f) // https://github.com/rust-lang/rust/issues/96865 + .await; - let f = vectored_dio_read::execute( - &inner.file, - reads - .iter() - .flat_map(|(_, value_reads)| value_reads.iter().map(|v| &v.read)), - &ctx, - ); - send_future::SendFuture::send(f) // https://github.com/rust-lang/rust/issues/96865 - .await; - - // Process results into the reconstruct state - 'next_key: for (key, value_reads) in reads { - for ValueRead { entry_lsn, read } in value_reads { - match read.into_result().expect("we run execute() above") { - Err(e) => { - reconstruct_state.on_key_error(key, PageReconstructError::from(anyhow!(e))); - continue 'next_key; - } - Ok(value_buf) => { - let value = Value::des(&value_buf); - if let Err(e) = value { - reconstruct_state - .on_key_error(key, PageReconstructError::from(anyhow!(e))); - continue 'next_key; + for (key, value_reads) in reads { + for ValueRead { entry_lsn, read } in value_reads { + let sender = senders + .remove(&(key, entry_lsn)) + .expect("sender must exist"); + match read.into_result().expect("we run execute() above") { + Err(e) => { + let _ = sender + .send(Err(std::io::Error::new(e.kind(), "dio vec read failed"))); } - - let key_situation = - reconstruct_state.update_key(&key, entry_lsn, value.unwrap()); - if key_situation == ValueReconstructSituation::Complete { - // TODO: metric to see if we fetched more values than necessary - continue 'next_key; + Ok(value_buf) => { + let _ = sender.send(Ok(value_buf.into())); } - - // process the next value in the next iteration of the loop } } } - } + + assert!(senders.is_empty()); + }); reconstruct_state.on_lsn_advanced(&keyspace, self.start_lsn); @@ -600,7 +603,8 @@ impl InMemoryLayer { /// Get layer size. pub async fn size(&self) -> Result { let inner = self.inner.read().await; - Ok(inner.file.len()) + let locked = inner.file.try_read().expect("no contention"); + Ok(locked.len()) } /// Create a new, empty, in-memory layer @@ -614,9 +618,10 @@ impl InMemoryLayer { ) -> Result { trace!("initializing new empty InMemoryLayer for writing on timeline {timeline_id} at {start_lsn}"); - let file = - EphemeralFile::create(conf, tenant_shard_id, timeline_id, gate_guard, ctx).await?; - let key = InMemoryLayerFileId(file.page_cache_file_id()); + let file = Arc::new(tokio::sync::RwLock::new( + EphemeralFile::create(conf, tenant_shard_id, timeline_id, gate_guard, ctx).await?, + )); + let key = InMemoryLayerFileId(file.read().await.page_cache_file_id()); Ok(InMemoryLayer { file_id: key, @@ -648,7 +653,7 @@ impl InMemoryLayer { let mut inner = self.inner.write().await; self.assert_writable(); - let base_offset = inner.file.len(); + let base_offset = inner.file.read().await.len(); let SerializedBatch { raw, @@ -672,8 +677,13 @@ impl InMemoryLayer { } // Write the batch to the file - inner.file.write_raw(&raw, ctx).await?; - let new_size = inner.file.len(); + // FIXME: can't borrow arc + let new_size = { + let mut locked = inner.file.write().await; + locked.write_raw(&raw, ctx).await?; + locked.len() + }; + let expected_new_len = base_offset .checked_add(raw.len().into_u64()) // write_raw would error if we were to overflow u64. @@ -713,7 +723,7 @@ impl InMemoryLayer { pub(crate) async fn tick(&self) -> Option { let mut inner = self.inner.write().await; - let size = inner.file.len(); + let size = inner.file.read().await.len(); inner.resource_units.publish_size(size) } @@ -809,7 +819,7 @@ impl InMemoryLayer { match l0_flush_global_state { l0_flush::Inner::Direct { .. } => { - let file_contents = inner.file.load_to_io_buf(ctx).await?; + let file_contents = inner.file.read().await.load_to_io_buf(ctx).await?; let file_contents = file_contents.freeze(); for (key, vec_map) in inner.index.iter() { diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index 12919866a3..0474dc79e0 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -19,6 +19,7 @@ use camino::Utf8Path; use chrono::{DateTime, Utc}; use enumset::EnumSet; use fail::fail_point; +use futures::{stream::FuturesUnordered, StreamExt}; use handle::ShardTimelineId; use offload::OffloadError; use once_cell::sync::Lazy; @@ -71,7 +72,9 @@ use crate::{ config::AttachmentMode, layer_map::{LayerMap, SearchResult}, metadata::TimelineMetadata, - storage_layer::{inmemory_layer::IndexEntry, PersistentLayerDesc}, + storage_layer::{ + convert, inmemory_layer::IndexEntry, PersistentLayerDesc, ValueReconstructSituation, + }, }, walingest::WalLagCooldown, walredo, @@ -1136,22 +1139,38 @@ impl Timeline { let reconstruct_timer = crate::metrics::RECONSTRUCT_TIME .for_get_kind(get_kind) .start_timer(); - let mut results: BTreeMap> = BTreeMap::new(); let layers_visited = reconstruct_state.get_layers_visited(); + let futs = FuturesUnordered::new(); for (key, res) in std::mem::take(&mut reconstruct_state.keys) { - match res { - Err(err) => { - results.insert(key, Err(err)); - } - Ok(state) => { - let state = ValueReconstructState::from(state); + futs.push({ + let walredo_self = self.myself.upgrade().expect("&self method holds the arc"); + async move { + let state = res.expect("Read path is infallible"); + assert!(matches!( + state.situation, + ValueReconstructSituation::Complete + )); - let reconstruct_res = self.reconstruct_value(key, lsn, state).await; - results.insert(key, reconstruct_res); + let converted = match convert(key, state).await { + Ok(ok) => ok, + Err(err) => { + return (key, Err(err)); + } + }; + + ( + key, + walredo_self.reconstruct_value(key, lsn, converted).await, + ) } - } + }); } + + let results = futs + .collect::>>() + .await; + reconstruct_timer.stop_and_record(); // For aux file keys (v1 or v2) the vectored read path does not return an error @@ -5531,30 +5550,30 @@ impl Timeline { #[cfg(test)] pub(crate) async fn inspect_image_layers( self: &Arc, - lsn: Lsn, - ctx: &RequestContext, + _lsn: Lsn, + _ctx: &RequestContext, ) -> anyhow::Result> { - let mut all_data = Vec::new(); - let guard = self.layers.read().await; - for layer in guard.layer_map()?.iter_historic_layers() { - if !layer.is_delta() && layer.image_layer_lsn() == lsn { - let layer = guard.get_from_desc(&layer); - let mut reconstruct_data = ValuesReconstructState::default(); - layer - .get_values_reconstruct_data( - KeySpace::single(Key::MIN..Key::MAX), - lsn..Lsn(lsn.0 + 1), - &mut reconstruct_data, - ctx, - ) - .await?; - for (k, v) in reconstruct_data.keys { - all_data.push((k, v?.img.unwrap().1)); - } - } - } - all_data.sort(); - Ok(all_data) + // let mut all_data = Vec::new(); + // let guard = self.layers.read().await; + // for layer in guard.layer_map()?.iter_historic_layers() { + // if !layer.is_delta() && layer.image_layer_lsn() == lsn { + // let layer = guard.get_from_desc(&layer); + // let mut reconstruct_data = ValuesReconstructState::default(); + // layer + // .get_values_reconstruct_data( + // KeySpace::single(Key::MIN..Key::MAX), + // lsn..Lsn(lsn.0 + 1), + // &mut reconstruct_data, + // ctx, + // ) + // .await?; + // for (k, v) in reconstruct_data.keys { + // all_data.push((k, v?.img.unwrap().1)); + // } + // } + // } + // all_data.sort(); + Ok(Vec::new()) } /// Get all historic layer descriptors in the layer map diff --git a/pageserver/src/tenant/vectored_blob_io.rs b/pageserver/src/tenant/vectored_blob_io.rs index dfe2352310..fdafebc6a2 100644 --- a/pageserver/src/tenant/vectored_blob_io.rs +++ b/pageserver/src/tenant/vectored_blob_io.rs @@ -35,6 +35,7 @@ use crate::virtual_file::{self, VirtualFile}; pub struct BlobMeta { pub key: Key, pub lsn: Lsn, + pub will_init: bool, } /// A view into the vectored blobs read buffer. @@ -310,7 +311,8 @@ pub enum BlobFlag { /// * Iterate over the collected blobs and coalesce them into reads at the end pub struct VectoredReadPlanner { // Track all the blob offsets. Start offsets must be ordered. - blobs: BTreeMap>, + // Note: last bool is will_init + blobs: BTreeMap>, // Arguments for previous blob passed into [`VectoredReadPlanner::handle`] prev: Option<(Key, Lsn, u64, BlobFlag)>, @@ -371,12 +373,12 @@ impl VectoredReadPlanner { match flag { BlobFlag::None => { let blobs_for_key = self.blobs.entry(key).or_default(); - blobs_for_key.push((lsn, start_offset, end_offset)); + blobs_for_key.push((lsn, start_offset, end_offset, false)); } BlobFlag::ReplaceAll => { let blobs_for_key = self.blobs.entry(key).or_default(); blobs_for_key.clear(); - blobs_for_key.push((lsn, start_offset, end_offset)); + blobs_for_key.push((lsn, start_offset, end_offset, true)); } BlobFlag::Ignore => {} } @@ -387,11 +389,17 @@ impl VectoredReadPlanner { let mut reads = Vec::new(); for (key, blobs_for_key) in self.blobs { - for (lsn, start_offset, end_offset) in blobs_for_key { + for (lsn, start_offset, end_offset, will_init) in blobs_for_key { let extended = match &mut current_read_builder { - Some(read_builder) => { - read_builder.extend(start_offset, end_offset, BlobMeta { key, lsn }) - } + Some(read_builder) => read_builder.extend( + start_offset, + end_offset, + BlobMeta { + key, + lsn, + will_init, + }, + ), None => VectoredReadExtended::No, }; @@ -399,7 +407,11 @@ impl VectoredReadPlanner { let next_read_builder = ChunkedVectoredReadBuilder::new( start_offset, end_offset, - BlobMeta { key, lsn }, + BlobMeta { + key, + lsn, + will_init, + }, self.max_read_size, ); @@ -587,10 +599,20 @@ impl StreamingVectoredReadPlanner { start_offset: u64, end_offset: u64, is_last_blob_in_read: bool, + // destination: oneshot::Sender>, ) -> Option { match &mut self.read_builder { Some(read_builder) => { - let extended = read_builder.extend(start_offset, end_offset, BlobMeta { key, lsn }); + let extended = read_builder.extend( + start_offset, + end_offset, + BlobMeta { + key, + lsn, + // TODO(vlad): figure this out ... + will_init: false, + }, + ); assert_eq!(extended, VectoredReadExtended::Yes); } None => { @@ -598,7 +620,12 @@ impl StreamingVectoredReadPlanner { Some(ChunkedVectoredReadBuilder::new_streaming( start_offset, end_offset, - BlobMeta { key, lsn }, + BlobMeta { + key, + lsn, + // TODO(vlad): figure this out ... + will_init: false, + }, )) }; } @@ -923,6 +950,7 @@ mod tests { let meta = BlobMeta { key: Key::MIN, lsn: Lsn(0), + will_init: false, }; for (idx, (blob, offset)) in blobs.iter().zip(offsets.iter()).enumerate() {