Compare commits

...

7 Commits

Author SHA1 Message Date
John Spray 93e069ceae refactors 2023-12-11 15:29:37 +00:00
John Spray 90e27f1800 fix doc link 2023-12-07 10:29:36 +00:00
John Spray 0f533194d2 Avoid implict Weak ugprade in local_path 2023-12-07 10:05:06 +00:00
John Spray 5b181443b3 Merge remote-tracking branch 'upstream/main' into jcsp/layer-tenant-id 2023-12-07 09:49:12 +00:00
John Spray 41401ea2b8 pageserver: make Timeline::shutdown safe against Layer::drop 2023-12-05 17:41:28 +00:00
John Spray 2a8197b7ce pageserver: avoid storing path in layer
We can build it on-demand via Timeline.
2023-12-05 17:41:28 +00:00
John Spray d07bc7ba01 pageserver: don't store tenant/timeline ID on layers 2023-12-05 17:41:28 +00:00
10 changed files with 168 additions and 210 deletions
+2 -9
View File
@@ -3,7 +3,6 @@ use pageserver::repository::Key;
use pageserver::tenant::layer_map::LayerMap; use pageserver::tenant::layer_map::LayerMap;
use pageserver::tenant::storage_layer::LayerFileName; use pageserver::tenant::storage_layer::LayerFileName;
use pageserver::tenant::storage_layer::PersistentLayerDesc; use pageserver::tenant::storage_layer::PersistentLayerDesc;
use pageserver_api::shard::TenantShardId;
use rand::prelude::{SeedableRng, SliceRandom, StdRng}; use rand::prelude::{SeedableRng, SliceRandom, StdRng};
use std::cmp::{max, min}; use std::cmp::{max, min};
use std::fs::File; use std::fs::File;
@@ -11,7 +10,6 @@ use std::io::{BufRead, BufReader};
use std::path::PathBuf; use std::path::PathBuf;
use std::str::FromStr; use std::str::FromStr;
use std::time::Instant; use std::time::Instant;
use utils::id::{TenantId, TimelineId};
use utils::lsn::Lsn; use utils::lsn::Lsn;
@@ -211,13 +209,8 @@ fn bench_sequential(c: &mut Criterion) {
for i in 0..100_000 { for i in 0..100_000 {
let i32 = (i as u32) % 100; let i32 = (i as u32) % 100;
let zero = Key::from_hex("000000000000000000000000000000000000").unwrap(); let zero = Key::from_hex("000000000000000000000000000000000000").unwrap();
let layer = PersistentLayerDesc::new_img( let layer =
TenantShardId::unsharded(TenantId::generate()), PersistentLayerDesc::new_img(zero.add(10 * i32)..zero.add(10 * i32 + 1), Lsn(i), 0);
TimelineId::generate(),
zero.add(10 * i32)..zero.add(10 * i32 + 1),
Lsn(i),
0,
);
updates.insert_historic(layer); updates.insert_historic(layer);
} }
updates.flush(); updates.flush();
+2 -2
View File
@@ -310,8 +310,8 @@ pub async fn disk_usage_eviction_task_iteration_impl<U: Usage>(
.unwrap() .unwrap()
.as_micros(), .as_micros(),
partition, partition,
desc.tenant_shard_id, candidate.timeline.tenant_shard_id,
desc.timeline_id, candidate.timeline.timeline_id,
candidate.layer, candidate.layer,
); );
} }
@@ -1271,11 +1271,12 @@ impl RemoteTimelineClient {
let upload_result: anyhow::Result<()> = match &task.op { let upload_result: anyhow::Result<()> = match &task.op {
UploadOp::UploadLayer(ref layer, ref layer_metadata) => { UploadOp::UploadLayer(ref layer, ref layer_metadata) => {
let path = layer.local_path(); let path = layer.local_path_from_id(&self.tenant_shard_id, &self.timeline_id);
upload::upload_timeline_layer( upload::upload_timeline_layer(
self.conf, self.conf,
&self.storage_impl, &self.storage_impl,
path, &path,
layer_metadata, layer_metadata,
self.generation, self.generation,
) )
+3 -17
View File
@@ -24,7 +24,7 @@ use tracing::warn;
use utils::history_buffer::HistoryBufferWithDropCounter; use utils::history_buffer::HistoryBufferWithDropCounter;
use utils::rate_limit::RateLimit; use utils::rate_limit::RateLimit;
use utils::{id::TimelineId, lsn::Lsn}; use utils::lsn::Lsn;
pub use delta_layer::{DeltaLayer, DeltaLayerWriter, ValueRef}; pub use delta_layer::{DeltaLayer, DeltaLayerWriter, ValueRef};
pub use filename::{DeltaFileName, ImageFileName, LayerFileName}; pub use filename::{DeltaFileName, ImageFileName, LayerFileName};
@@ -301,31 +301,17 @@ pub trait AsLayerDesc {
} }
pub mod tests { pub mod tests {
use pageserver_api::shard::TenantShardId;
use super::*; use super::*;
impl From<DeltaFileName> for PersistentLayerDesc { impl From<DeltaFileName> for PersistentLayerDesc {
fn from(value: DeltaFileName) -> Self { fn from(value: DeltaFileName) -> Self {
PersistentLayerDesc::new_delta( PersistentLayerDesc::new_delta(value.key_range, value.lsn_range, 233)
TenantShardId::from([0; 18]),
TimelineId::from_array([0; 16]),
value.key_range,
value.lsn_range,
233,
)
} }
} }
impl From<ImageFileName> for PersistentLayerDesc { impl From<ImageFileName> for PersistentLayerDesc {
fn from(value: ImageFileName) -> Self { fn from(value: ImageFileName) -> Self {
PersistentLayerDesc::new_img( PersistentLayerDesc::new_img(value.key_range, value.lsn, 233)
TenantShardId::from([0; 18]),
TimelineId::from_array([0; 16]),
value.key_range,
value.lsn,
233,
)
} }
} }
@@ -84,17 +84,6 @@ pub struct Summary {
pub index_root_blk: u32, pub index_root_blk: u32,
} }
impl From<&DeltaLayer> for Summary {
fn from(layer: &DeltaLayer) -> Self {
Self::expected(
layer.desc.tenant_shard_id.tenant_id,
layer.desc.timeline_id,
layer.desc.key_range.clone(),
layer.desc.lsn_range.clone(),
)
}
}
impl Summary { impl Summary {
pub(super) fn expected( pub(super) fn expected(
tenant_id: TenantId, tenant_id: TenantId,
@@ -320,15 +309,9 @@ impl DeltaLayer {
.metadata() .metadata()
.context("get file metadata to determine size")?; .context("get file metadata to determine size")?;
// TODO(sharding): we must get the TenantShardId from the path instead of reading the Summary.
// we should also validate the path against the Summary, as both should contain the same tenant, timeline, key, lsn.
let tenant_shard_id = TenantShardId::unsharded(summary.tenant_id);
Ok(DeltaLayer { Ok(DeltaLayer {
path: path.to_path_buf(), path: path.to_path_buf(),
desc: PersistentLayerDesc::new_delta( desc: PersistentLayerDesc::new_delta(
tenant_shard_id,
summary.timeline_id,
summary.key_range, summary.key_range,
summary.lsn_range, summary.lsn_range,
metadata.len(), metadata.len(),
@@ -505,8 +488,6 @@ impl DeltaLayerWriterInner {
// set inner.file here. The first read will have to re-open it. // set inner.file here. The first read will have to re-open it.
let desc = PersistentLayerDesc::new_delta( let desc = PersistentLayerDesc::new_delta(
self.tenant_shard_id,
self.timeline_id,
self.key_start..key_end, self.key_start..key_end,
self.lsn_range.clone(), self.lsn_range.clone(),
metadata.len(), metadata.len(),
@@ -517,7 +498,7 @@ impl DeltaLayerWriterInner {
let layer = Layer::finish_creating(self.conf, timeline, desc, &self.path)?; let layer = Layer::finish_creating(self.conf, timeline, desc, &self.path)?;
trace!("created delta layer {}", layer.local_path()); trace!("created delta layer {}", self.path);
Ok(layer) Ok(layer)
} }
@@ -85,17 +85,6 @@ pub struct Summary {
// the 'values' part starts after the summary header, on block 1. // the 'values' part starts after the summary header, on block 1.
} }
impl From<&ImageLayer> for Summary {
fn from(layer: &ImageLayer) -> Self {
Self::expected(
layer.desc.tenant_shard_id.tenant_id,
layer.desc.timeline_id,
layer.desc.key_range.clone(),
layer.lsn,
)
}
}
impl Summary { impl Summary {
pub(super) fn expected( pub(super) fn expected(
tenant_id: TenantId, tenant_id: TenantId,
@@ -278,19 +267,9 @@ impl ImageLayer {
.metadata() .metadata()
.context("get file metadata to determine size")?; .context("get file metadata to determine size")?;
// TODO(sharding): we should get TenantShardId from path.
// OR, not at all: any layer we load from disk should also get reconciled with remote IndexPart.
let tenant_shard_id = TenantShardId::unsharded(summary.tenant_id);
Ok(ImageLayer { Ok(ImageLayer {
path: path.to_path_buf(), path: path.to_path_buf(),
desc: PersistentLayerDesc::new_img( desc: PersistentLayerDesc::new_img(summary.key_range, summary.lsn, metadata.len()), // Now we assume image layer ALWAYS covers the full range. This may change in the future.
tenant_shard_id,
summary.timeline_id,
summary.key_range,
summary.lsn,
metadata.len(),
), // Now we assume image layer ALWAYS covers the full range. This may change in the future.
lsn: summary.lsn, lsn: summary.lsn,
access_stats: LayerAccessStats::empty_will_record_residence_event_later(), access_stats: LayerAccessStats::empty_will_record_residence_event_later(),
inner: OnceCell::new(), inner: OnceCell::new(),
@@ -581,13 +560,7 @@ impl ImageLayerWriterInner {
.await .await
.context("get metadata to determine file size")?; .context("get metadata to determine file size")?;
let desc = PersistentLayerDesc::new_img( let desc = PersistentLayerDesc::new_img(self.key_range.clone(), self.lsn, metadata.len());
self.tenant_shard_id,
self.timeline_id,
self.key_range.clone(),
self.lsn,
metadata.len(),
);
// Note: Because we open the file in write-only mode, we cannot // Note: Because we open the file in write-only mode, we cannot
// reuse the same VirtualFile for reading later. That's why we don't // reuse the same VirtualFile for reading later. That's why we don't
@@ -599,7 +572,7 @@ impl ImageLayerWriterInner {
// FIXME: why not carry the virtualfile here, it supports renaming? // FIXME: why not carry the virtualfile here, it supports renaming?
let layer = Layer::finish_creating(self.conf, timeline, desc, &self.path)?; let layer = Layer::finish_creating(self.conf, timeline, desc, &self.path)?;
trace!("created image layer {}", layer.local_path()); trace!("created image layer {}", self.path);
Ok(layer) Ok(layer)
} }
+119 -72
View File
@@ -3,13 +3,15 @@ use camino::{Utf8Path, Utf8PathBuf};
use pageserver_api::models::{ use pageserver_api::models::{
HistoricLayerInfo, LayerAccessKind, LayerResidenceEventReason, LayerResidenceStatus, HistoricLayerInfo, LayerAccessKind, LayerResidenceEventReason, LayerResidenceStatus,
}; };
use pageserver_api::shard::ShardIndex; use pageserver_api::shard::{ShardIndex, TenantShardId};
use std::ops::Range; use std::ops::Range;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Weak}; use std::sync::{Arc, Weak};
use std::time::SystemTime; use std::time::SystemTime;
use tracing::Instrument; use tracing::Instrument;
use utils::id::TimelineId;
use utils::lsn::Lsn; use utils::lsn::Lsn;
use utils::sync::gate::GateError;
use utils::sync::heavier_once_cell; use utils::sync::heavier_once_cell;
use crate::config::PageServerConf; use crate::config::PageServerConf;
@@ -81,12 +83,7 @@ impl Layer {
file_name: LayerFileName, file_name: LayerFileName,
metadata: LayerFileMetadata, metadata: LayerFileMetadata,
) -> Self { ) -> Self {
let desc = PersistentLayerDesc::from_filename( let desc = PersistentLayerDesc::from_filename(file_name, metadata.file_size());
timeline.tenant_shard_id,
timeline.timeline_id,
file_name,
metadata.file_size(),
);
let access_stats = LayerAccessStats::for_loading_layer(LayerResidenceStatus::Evicted); let access_stats = LayerAccessStats::for_loading_layer(LayerResidenceStatus::Evicted);
@@ -100,7 +97,7 @@ impl Layer {
metadata.shard, metadata.shard,
))); )));
debug_assert!(owner.0.needs_download_blocking().unwrap().is_some()); debug_assert!(owner.0.needs_download_blocking(timeline).unwrap().is_some());
owner owner
} }
@@ -112,12 +109,7 @@ impl Layer {
file_name: LayerFileName, file_name: LayerFileName,
metadata: LayerFileMetadata, metadata: LayerFileMetadata,
) -> ResidentLayer { ) -> ResidentLayer {
let desc = PersistentLayerDesc::from_filename( let desc = PersistentLayerDesc::from_filename(file_name, metadata.file_size());
timeline.tenant_shard_id,
timeline.timeline_id,
file_name,
metadata.file_size(),
);
let access_stats = LayerAccessStats::for_loading_layer(LayerResidenceStatus::Resident); let access_stats = LayerAccessStats::for_loading_layer(LayerResidenceStatus::Resident);
@@ -144,7 +136,7 @@ impl Layer {
let downloaded = resident.expect("just initialized"); let downloaded = resident.expect("just initialized");
debug_assert!(owner.0.needs_download_blocking().unwrap().is_none()); debug_assert!(owner.0.needs_download_blocking(timeline).unwrap().is_none());
timeline timeline
.metrics .metrics
@@ -189,7 +181,7 @@ impl Layer {
let downloaded = resident.expect("just initialized"); let downloaded = resident.expect("just initialized");
// if the rename works, the path is as expected // if the rename works, the path is as expected
std::fs::rename(temp_path, owner.local_path()) std::fs::rename(temp_path, owner.local_path(timeline))
.with_context(|| format!("rename temporary file as correct path for {owner}"))?; .with_context(|| format!("rename temporary file as correct path for {owner}"))?;
Ok(ResidentLayer { downloaded, owner }) Ok(ResidentLayer { downloaded, owner })
@@ -309,8 +301,12 @@ impl Layer {
&self.0.access_stats &self.0.access_stats
} }
pub(crate) fn local_path(&self) -> &Utf8Path { fn local_path(&self, timeline: &Timeline) -> Utf8PathBuf {
&self.0.path self.0.local_path(timeline)
}
pub(crate) fn filename(&self) -> LayerFileName {
self.0.desc.filename()
} }
pub(crate) fn metadata(&self) -> LayerFileMetadata { pub(crate) fn metadata(&self) -> LayerFileMetadata {
@@ -402,13 +398,9 @@ impl ResidentOrWantedEvicted {
} }
struct LayerInner { struct LayerInner {
/// Only needed to check ondemand_download_behavior_treat_error_as_warn and creation of /// Only needed to check ondemand_download_behavior_treat_error_as_warn and in [`Self::local_path_from_id`]
/// [`Self::path`].
conf: &'static PageServerConf, conf: &'static PageServerConf,
/// Full path to the file; unclear if this should exist anymore.
path: Utf8PathBuf,
desc: PersistentLayerDesc, desc: PersistentLayerDesc,
/// Timeline access is needed for remote timeline client and metrics. /// Timeline access is needed for remote timeline client and metrics.
@@ -486,12 +478,32 @@ impl Drop for LayerInner {
return; return;
} }
let span = tracing::info_span!(parent: None, "layer_delete", tenant_id = %self.layer_desc().tenant_shard_id.tenant_id, shard_id=%self.layer_desc().tenant_shard_id.shard_slug(), timeline_id = %self.layer_desc().timeline_id); // We will only do I/O on drop if our Timeline still exists. Otherwise, we may safely
// leave garbage layers behind to be cleaned up the next time this Timeline is instantiated.
let Some(timeline) = self.timeline.upgrade() else {
// no need to nag that timeline is gone: under normal situation on
// task_mgr::remove_tenant_from_memory the timeline is gone before we get dropped.
LAYER_IMPL_METRICS.inc_deletes_failed(DeleteFailed::TimelineGone);
return;
};
// We will only do I/O during drop if our Timeline's layer_gate is open: this avoids
// the risk that we would race with Timeline::shutdown and end up doing I/O to a timeline
// path for which the Timeline object has been torn down already.
let _gate_guard = match timeline.layer_gate.enter() {
Ok(g) => g,
Err(GateError::GateClosed) => {
LAYER_IMPL_METRICS.inc_deletes_failed(DeleteFailed::TimelineGone);
return;
}
};
// If timeline is alive, we can construct a span with IDs for this function.
let span = tracing::info_span!(parent: None, "layer_delete", tenant_id = %timeline.tenant_shard_id.tenant_id, shard_id=%timeline.tenant_shard_id.shard_slug(), timeline_id = %timeline.timeline_id);
let path = self.local_path(&timeline);
let path = std::mem::take(&mut self.path);
let file_name = self.layer_desc().filename(); let file_name = self.layer_desc().filename();
let file_size = self.layer_desc().file_size; let file_size = self.layer_desc().file_size;
let timeline = self.timeline.clone();
let meta = self.metadata(); let meta = self.metadata();
let status = self.status.clone(); let status = self.status.clone();
@@ -519,32 +531,26 @@ impl Drop for LayerInner {
} }
}; };
if let Some(timeline) = timeline.upgrade() { if removed {
if removed { timeline.metrics.resident_physical_size_sub(file_size);
timeline.metrics.resident_physical_size_sub(file_size); }
} if let Some(remote_client) = timeline.remote_client.as_ref() {
if let Some(remote_client) = timeline.remote_client.as_ref() { let res = remote_client.schedule_deletion_of_unlinked(vec![(file_name, meta)]);
let res = remote_client.schedule_deletion_of_unlinked(vec![(file_name, meta)]);
if let Err(e) = res { if let Err(e) = res {
// test_timeline_deletion_with_files_stuck_in_upload_queue is good at // test_timeline_deletion_with_files_stuck_in_upload_queue is good at
// demonstrating this deadlock (without spawn_blocking): stop will drop // demonstrating this deadlock (without spawn_blocking): stop will drop
// queued items, which will have ResidentLayer's, and those drops would try // queued items, which will have ResidentLayer's, and those drops would try
// to re-entrantly lock the RemoteTimelineClient inner state. // to re-entrantly lock the RemoteTimelineClient inner state.
if !timeline.is_active() { if !timeline.is_active() {
tracing::info!("scheduling deletion on drop failed: {e:#}"); tracing::info!("scheduling deletion on drop failed: {e:#}");
} else {
tracing::warn!("scheduling deletion on drop failed: {e:#}");
}
LAYER_IMPL_METRICS.inc_deletes_failed(DeleteFailed::DeleteSchedulingFailed);
} else { } else {
LAYER_IMPL_METRICS.inc_completed_deletes(); tracing::warn!("scheduling deletion on drop failed: {e:#}");
} }
LAYER_IMPL_METRICS.inc_deletes_failed(DeleteFailed::DeleteSchedulingFailed);
} else {
LAYER_IMPL_METRICS.inc_completed_deletes();
} }
} else {
// no need to nag that timeline is gone: under normal situation on
// task_mgr::remove_tenant_from_memory the timeline is gone before we get dropped.
LAYER_IMPL_METRICS.inc_deletes_failed(DeleteFailed::TimelineGone);
} }
}); });
} }
@@ -560,10 +566,6 @@ impl LayerInner {
generation: Generation, generation: Generation,
shard: ShardIndex, shard: ShardIndex,
) -> Self { ) -> Self {
let path = conf
.timeline_path(&timeline.tenant_shard_id, &timeline.timeline_id)
.join(desc.filename().to_string());
let (inner, version) = if let Some(inner) = downloaded { let (inner, version) = if let Some(inner) = downloaded {
let version = inner.version; let version = inner.version;
let resident = ResidentOrWantedEvicted::Resident(inner); let resident = ResidentOrWantedEvicted::Resident(inner);
@@ -574,7 +576,6 @@ impl LayerInner {
LayerInner { LayerInner {
conf, conf,
path,
desc, desc,
timeline: Arc::downgrade(timeline), timeline: Arc::downgrade(timeline),
have_remote_client: timeline.remote_client.is_some(), have_remote_client: timeline.remote_client.is_some(),
@@ -590,6 +591,25 @@ impl LayerInner {
} }
} }
/// All call sites that need this function should already have a Timeline (e.g. from
/// upgrading the Self::timeline weak pointer) -- it doesn't make sense to try and
/// do anything with the local file if the Timeline isn't still alive.
fn local_path(&self, timeline: &Timeline) -> Utf8PathBuf {
self.local_path_from_id(&timeline.tenant_shard_id, &timeline.timeline_id)
}
/// Use this instead of `local_path` if you don't have a Timeline but do have its ID: this
/// is used by external callers such as [`crate::tenant::RemoteTimelineClient`]
pub(crate) fn local_path_from_id(
&self,
tenant_shard_id: &TenantShardId,
timeline_id: &TimelineId,
) -> Utf8PathBuf {
self.conf
.timeline_path(tenant_shard_id, timeline_id)
.join(self.desc.filename().to_string())
}
fn delete_on_drop(&self) { fn delete_on_drop(&self) {
let res = let res =
self.wanted_deleted self.wanted_deleted
@@ -683,7 +703,7 @@ impl LayerInner {
// check if we really need to be downloaded; could have been already downloaded by a // check if we really need to be downloaded; could have been already downloaded by a
// cancelled previous attempt. // cancelled previous attempt.
let needs_download = self let needs_download = self
.needs_download() .needs_download(&timeline)
.await .await
.map_err(DownloadError::PreStatFailed)?; .map_err(DownloadError::PreStatFailed)?;
@@ -833,12 +853,13 @@ impl LayerInner {
// block tenant::mgr::remove_tenant_from_memory. // block tenant::mgr::remove_tenant_from_memory.
let this: Arc<Self> = self.clone(); let this: Arc<Self> = self.clone();
let timeline_clone = timeline.clone();
crate::task_mgr::spawn( crate::task_mgr::spawn(
&tokio::runtime::Handle::current(), &tokio::runtime::Handle::current(),
crate::task_mgr::TaskKind::RemoteDownloadTask, crate::task_mgr::TaskKind::RemoteDownloadTask,
Some(self.desc.tenant_shard_id.tenant_id), Some(timeline.tenant_shard_id.tenant_id),
Some(self.desc.timeline_id), Some(timeline.timeline_id),
&task_name, &task_name,
false, false,
async move { async move {
@@ -894,7 +915,7 @@ impl LayerInner {
match rx.await { match rx.await {
Ok((Ok(()), permit)) => { Ok((Ok(()), permit)) => {
if let Some(reason) = self if let Some(reason) = self
.needs_download() .needs_download(&timeline_clone)
.await .await
.map_err(DownloadError::PostStatFailed)? .map_err(DownloadError::PostStatFailed)?
{ {
@@ -929,16 +950,26 @@ impl LayerInner {
} }
} }
async fn needs_download(&self) -> Result<Option<NeedsDownload>, std::io::Error> { async fn needs_download(
match tokio::fs::metadata(&self.path).await { &self,
timeline: &Timeline,
) -> Result<Option<NeedsDownload>, std::io::Error> {
let path = self.local_path(timeline);
match tokio::fs::metadata(path).await {
Ok(m) => Ok(self.is_file_present_and_good_size(&m).err()), Ok(m) => Ok(self.is_file_present_and_good_size(&m).err()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Some(NeedsDownload::NotFound)), Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Some(NeedsDownload::NotFound)),
Err(e) => Err(e), Err(e) => Err(e),
} }
} }
fn needs_download_blocking(&self) -> Result<Option<NeedsDownload>, std::io::Error> { fn needs_download_blocking(
match self.path.metadata() { &self,
timeline: &Timeline,
) -> Result<Option<NeedsDownload>, std::io::Error> {
let path = self.local_path(timeline);
match path.metadata() {
Ok(m) => Ok(self.is_file_present_and_good_size(&m).err()), Ok(m) => Ok(self.is_file_present_and_good_size(&m).err()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Some(NeedsDownload::NotFound)), Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Some(NeedsDownload::NotFound)),
Err(e) => Err(e), Err(e) => Err(e),
@@ -1004,7 +1035,10 @@ impl LayerInner {
// //
// FIXME: this is not true anymore, we can safely evict wanted deleted files. // FIXME: this is not true anymore, we can safely evict wanted deleted files.
} else if can_evict && evict { } else if can_evict && evict {
let span = tracing::info_span!(parent: None, "layer_evict", tenant_id = %self.desc.tenant_shard_id.tenant_id, shard_id = %self.desc.tenant_shard_id.shard_slug(), timeline_id = %self.desc.timeline_id, layer=%self, %version); // If timeline is alive, we can construct a span with IDs for this function.
let span = self.timeline.upgrade().map(|timeline| {
tracing::info_span!(parent: None, "layer_evict", tenant_id = %timeline.tenant_shard_id.tenant_id, shard_id=%timeline.tenant_shard_id.shard_slug(), timeline_id = %timeline.timeline_id)
});
// downgrade for queueing, in case there's a tear down already ongoing we should not // downgrade for queueing, in case there's a tear down already ongoing we should not
// hold it alive. // hold it alive.
@@ -1015,7 +1049,7 @@ impl LayerInner {
// drop while the `self.inner` is being locked, leading to a deadlock. // drop while the `self.inner` is being locked, leading to a deadlock.
crate::task_mgr::BACKGROUND_RUNTIME.spawn_blocking(move || { crate::task_mgr::BACKGROUND_RUNTIME.spawn_blocking(move || {
let _g = span.entered(); let _g = span.map(|s| s.entered());
// if LayerInner is already dropped here, do nothing because the delete on drop // if LayerInner is already dropped here, do nothing because the delete on drop
// has already ran while we were in queue // has already ran while we were in queue
@@ -1075,7 +1109,9 @@ impl LayerInner {
LayerResidenceEventReason::ResidenceChange, LayerResidenceEventReason::ResidenceChange,
); );
let res = match capture_mtime_and_remove(&self.path) { let local_path = self.local_path(&timeline);
let res = match capture_mtime_and_remove(&local_path) {
Ok(local_layer_mtime) => { Ok(local_layer_mtime) => {
let duration = SystemTime::now().duration_since(local_layer_mtime); let duration = SystemTime::now().duration_since(local_layer_mtime);
match duration { match duration {
@@ -1227,6 +1263,11 @@ impl DownloadedLayer {
owner: &Arc<LayerInner>, owner: &Arc<LayerInner>,
ctx: &RequestContext, ctx: &RequestContext,
) -> anyhow::Result<&'a LayerKind> { ) -> anyhow::Result<&'a LayerKind> {
let timeline = owner
.timeline
.upgrade()
.ok_or(DownloadError::TimelineShutdown)?;
let init = || async { let init = || async {
assert_eq!( assert_eq!(
Weak::as_ptr(&self.owner), Weak::as_ptr(&self.owner),
@@ -1236,23 +1277,23 @@ impl DownloadedLayer {
let res = if owner.desc.is_delta { let res = if owner.desc.is_delta {
let summary = Some(delta_layer::Summary::expected( let summary = Some(delta_layer::Summary::expected(
owner.desc.tenant_shard_id.tenant_id, timeline.tenant_shard_id.tenant_id,
owner.desc.timeline_id, timeline.timeline_id,
owner.desc.key_range.clone(), owner.desc.key_range.clone(),
owner.desc.lsn_range.clone(), owner.desc.lsn_range.clone(),
)); ));
delta_layer::DeltaLayerInner::load(&owner.path, summary, ctx) delta_layer::DeltaLayerInner::load(&owner.local_path(&timeline), summary, ctx)
.await .await
.map(|res| res.map(LayerKind::Delta)) .map(|res| res.map(LayerKind::Delta))
} else { } else {
let lsn = owner.desc.image_layer_lsn(); let lsn = owner.desc.image_layer_lsn();
let summary = Some(image_layer::Summary::expected( let summary = Some(image_layer::Summary::expected(
owner.desc.tenant_shard_id.tenant_id, timeline.tenant_shard_id.tenant_id,
owner.desc.timeline_id, timeline.timeline_id,
owner.desc.key_range.clone(), owner.desc.key_range.clone(),
lsn, lsn,
)); ));
image_layer::ImageLayerInner::load(&owner.path, lsn, summary, ctx) image_layer::ImageLayerInner::load(&owner.local_path(&timeline), lsn, summary, ctx)
.await .await
.map(|res| res.map(LayerKind::Image)) .map(|res| res.map(LayerKind::Image))
}; };
@@ -1376,8 +1417,14 @@ impl ResidentLayer {
} }
} }
pub(crate) fn local_path(&self) -> &Utf8Path { pub(crate) fn local_path_from_id(
&self.owner.0.path &self,
tenant_shard_id: &TenantShardId,
timeline_id: &TimelineId,
) -> Utf8PathBuf {
self.owner
.0
.local_path_from_id(tenant_shard_id, timeline_id)
} }
pub(crate) fn access_stats(&self) -> &LayerAccessStats { pub(crate) fn access_stats(&self) -> &LayerAccessStats {
@@ -1,7 +1,6 @@
use core::fmt::Display; use core::fmt::Display;
use pageserver_api::shard::TenantShardId;
use std::ops::Range; use std::ops::Range;
use utils::{id::TimelineId, lsn::Lsn}; use utils::lsn::Lsn;
use crate::repository::Key; use crate::repository::Key;
@@ -9,16 +8,11 @@ use super::{DeltaFileName, ImageFileName, LayerFileName};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
#[cfg(test)]
use utils::id::TenantId;
/// A unique identifier of a persistent layer. This is different from `LayerDescriptor`, which is only used in the /// A unique identifier of a persistent layer. This is different from `LayerDescriptor`, which is only used in the
/// benchmarks. This struct contains all necessary information to find the image / delta layer. It also provides /// benchmarks. This struct contains all necessary information to find the image / delta layer. It also provides
/// a unified way to generate layer information like file name. /// a unified way to generate layer information like file name.
#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)] #[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
pub struct PersistentLayerDesc { pub struct PersistentLayerDesc {
pub tenant_shard_id: TenantShardId,
pub timeline_id: TimelineId,
/// Range of keys that this layer covers /// Range of keys that this layer covers
pub key_range: Range<Key>, pub key_range: Range<Key>,
/// Inclusive start, exclusive end of the LSN range that this layer holds. /// Inclusive start, exclusive end of the LSN range that this layer holds.
@@ -57,8 +51,6 @@ impl PersistentLayerDesc {
#[cfg(test)] #[cfg(test)]
pub fn new_test(key_range: Range<Key>) -> Self { pub fn new_test(key_range: Range<Key>) -> Self {
Self { Self {
tenant_shard_id: TenantShardId::unsharded(TenantId::generate()),
timeline_id: TimelineId::generate(),
key_range, key_range,
lsn_range: Lsn(0)..Lsn(1), lsn_range: Lsn(0)..Lsn(1),
is_delta: false, is_delta: false,
@@ -66,16 +58,8 @@ impl PersistentLayerDesc {
} }
} }
pub fn new_img( pub fn new_img(key_range: Range<Key>, lsn: Lsn, file_size: u64) -> Self {
tenant_shard_id: TenantShardId,
timeline_id: TimelineId,
key_range: Range<Key>,
lsn: Lsn,
file_size: u64,
) -> Self {
Self { Self {
tenant_shard_id,
timeline_id,
key_range, key_range,
lsn_range: Self::image_layer_lsn_range(lsn), lsn_range: Self::image_layer_lsn_range(lsn),
is_delta: false, is_delta: false,
@@ -83,16 +67,8 @@ impl PersistentLayerDesc {
} }
} }
pub fn new_delta( pub fn new_delta(key_range: Range<Key>, lsn_range: Range<Lsn>, file_size: u64) -> Self {
tenant_shard_id: TenantShardId,
timeline_id: TimelineId,
key_range: Range<Key>,
lsn_range: Range<Lsn>,
file_size: u64,
) -> Self {
Self { Self {
tenant_shard_id,
timeline_id,
key_range, key_range,
lsn_range, lsn_range,
is_delta: true, is_delta: true,
@@ -100,23 +76,10 @@ impl PersistentLayerDesc {
} }
} }
pub fn from_filename( pub fn from_filename(filename: LayerFileName, file_size: u64) -> Self {
tenant_shard_id: TenantShardId,
timeline_id: TimelineId,
filename: LayerFileName,
file_size: u64,
) -> Self {
match filename { match filename {
LayerFileName::Image(i) => { LayerFileName::Image(i) => Self::new_img(i.key_range, i.lsn, file_size),
Self::new_img(tenant_shard_id, timeline_id, i.key_range, i.lsn, file_size) LayerFileName::Delta(d) => Self::new_delta(d.key_range, d.lsn_range, file_size),
}
LayerFileName::Delta(d) => Self::new_delta(
tenant_shard_id,
timeline_id,
d.key_range,
d.lsn_range,
file_size,
),
} }
} }
@@ -173,10 +136,6 @@ impl PersistentLayerDesc {
self.key_range.clone() self.key_range.clone()
} }
pub fn get_timeline_id(&self) -> TimelineId {
self.timeline_id
}
/// Does this layer only contain some data for the key-range (incremental), /// Does this layer only contain some data for the key-range (incremental),
/// or does it contain a version of every page? This is important to know /// or does it contain a version of every page? This is important to know
/// for garbage collecting old layers: an incremental layer depends on /// for garbage collecting old layers: an incremental layer depends on
@@ -192,9 +151,7 @@ impl PersistentLayerDesc {
pub fn dump(&self) { pub fn dump(&self) {
if self.is_delta { if self.is_delta {
println!( println!(
"----- delta layer for ten {} tli {} keys {}-{} lsn {}-{} is_incremental {} size {} ----", "----- delta layer keys {}-{} lsn {}-{} is_incremental {} size {} ----",
self.tenant_shard_id,
self.timeline_id,
self.key_range.start, self.key_range.start,
self.key_range.end, self.key_range.end,
self.lsn_range.start, self.lsn_range.start,
@@ -204,9 +161,7 @@ impl PersistentLayerDesc {
); );
} else { } else {
println!( println!(
"----- image layer for ten {} tli {} key {}-{} at {} is_incremental {} size {} ----", "----- image layer key {}-{} at {} is_incremental {} size {} ----",
self.tenant_shard_id,
self.timeline_id,
self.key_range.start, self.key_range.start,
self.key_range.end, self.key_range.end,
self.image_layer_lsn(), self.image_layer_lsn(),
+18 -5
View File
@@ -313,6 +313,10 @@ pub struct Timeline {
/// Gate to prevent shutdown completing while I/O is still happening to this timeline's data /// Gate to prevent shutdown completing while I/O is still happening to this timeline's data
pub(crate) gate: Gate, pub(crate) gate: Gate,
/// Gate to prevent shutdown completing until all Layers for this Timeline have finished
/// doing any background I/O such as deleting files on drop.
pub(crate) layer_gate: Gate,
/// Cancellation token scoped to this timeline: anything doing long-running work relating /// Cancellation token scoped to this timeline: anything doing long-running work relating
/// to the timeline should drop out when this token fires. /// to the timeline should drop out when this token fires.
pub(crate) cancel: CancellationToken, pub(crate) cancel: CancellationToken,
@@ -1002,8 +1006,15 @@ impl Timeline {
) )
.await; .await;
// Finally wait until any gate-holders are complete // Wait until any normal gate-holders such as page_service requests are complete
self.gate.close().await; self.gate.close().await;
// Drop our references to layers: this should permit all layers to be dropped, and any I/O
// in their drop() method to complete.
self.layers.write().await.clear();
// Wait until any Layer gate holders such as LayerInner::drop are complete
self.layer_gate.close().await;
} }
pub fn set_state(&self, new_state: TimelineState) { pub fn set_state(&self, new_state: TimelineState) {
@@ -1445,6 +1456,7 @@ impl Timeline {
cancel, cancel,
gate: Gate::new(format!("Timeline<{tenant_shard_id}/{timeline_id}>")), gate: Gate::new(format!("Timeline<{tenant_shard_id}/{timeline_id}>")),
layer_gate: Gate::new(format!("TimelineLayers<{tenant_shard_id}/{timeline_id}>")),
compaction_lock: tokio::sync::Mutex::default(), compaction_lock: tokio::sync::Mutex::default(),
gc_lock: tokio::sync::Mutex::default(), gc_lock: tokio::sync::Mutex::default(),
@@ -2176,7 +2188,7 @@ trait TraversalLayerExt {
impl TraversalLayerExt for Layer { impl TraversalLayerExt for Layer {
fn traversal_id(&self) -> TraversalId { fn traversal_id(&self) -> TraversalId {
self.local_path().to_string() self.filename().to_string()
} }
} }
@@ -2890,7 +2902,8 @@ impl Timeline {
let _g = span.entered(); let _g = span.entered();
let new_delta = let new_delta =
Handle::current().block_on(frozen_layer.write_to_disk(&self_clone, &ctx))?; Handle::current().block_on(frozen_layer.write_to_disk(&self_clone, &ctx))?;
let new_delta_path = new_delta.local_path().to_owned(); let new_delta_path = new_delta
.local_path_from_id(&self_clone.tenant_shard_id, &self_clone.timeline_id);
// Sync it to disk. // Sync it to disk.
// //
@@ -3134,7 +3147,7 @@ impl Timeline {
// and fsync them all in parallel. // and fsync them all in parallel.
let all_paths = image_layers let all_paths = image_layers
.iter() .iter()
.map(|layer| layer.local_path().to_owned()) .map(|layer| layer.local_path_from_id(&self.tenant_shard_id, &self.timeline_id))
.collect::<Vec<_>>(); .collect::<Vec<_>>();
par_fsync::par_fsync_async(&all_paths) par_fsync::par_fsync_async(&all_paths)
@@ -3683,7 +3696,7 @@ impl Timeline {
// FIXME: the writer already fsyncs all data, only rename needs to be fsynced here // FIXME: the writer already fsyncs all data, only rename needs to be fsynced here
let layer_paths: Vec<Utf8PathBuf> = new_layers let layer_paths: Vec<Utf8PathBuf> = new_layers
.iter() .iter()
.map(|l| l.local_path().to_owned()) .map(|l| l.local_path_from_id(&self.tenant_shard_id, &self.timeline_id))
.collect(); .collect();
// Fsync all the layer files and directory using multiple threads to // Fsync all the layer files and directory using multiple threads to
@@ -33,6 +33,11 @@ impl LayerManager {
} }
} }
pub(crate) fn clear(&mut self) {
self.layer_map = LayerMap::default();
self.layer_fmgr.clear();
}
pub(crate) fn get_from_desc(&self, desc: &PersistentLayerDesc) -> Layer { pub(crate) fn get_from_desc(&self, desc: &PersistentLayerDesc) -> Layer {
self.layer_fmgr.get_from_desc(desc) self.layer_fmgr.get_from_desc(desc)
} }
@@ -271,6 +276,10 @@ impl<T: AsLayerDesc + Clone> LayerFileManager<T> {
} }
} }
pub(crate) fn clear(&mut self) {
self.0.clear();
}
pub(crate) fn contains(&self, layer: &T) -> bool { pub(crate) fn contains(&self, layer: &T) -> bool {
self.0.contains_key(&layer.layer_desc().key()) self.0.contains_key(&layer.layer_desc().key())
} }