diff --git a/pageserver/src/tenant/layer_map.rs b/pageserver/src/tenant/layer_map.rs index 4c4cd90c99..167da28e11 100644 --- a/pageserver/src/tenant/layer_map.rs +++ b/pageserver/src/tenant/layer_map.rs @@ -552,12 +552,6 @@ impl LayerMap { where Pred: FnMut(&Arc) -> bool, { - if let Some(open) = &self.open_layer { - if pred(open) { - return Some(open.clone()); - } - } - self.frozen_layers.iter().rfind(|l| pred(l)).cloned() } @@ -853,11 +847,6 @@ impl LayerMap { pub async fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> { println!("Begin dump LayerMap"); - println!("open_layer:"); - if let Some(open_layer) = &self.open_layer { - open_layer.dump(verbose, ctx).await?; - } - println!("frozen_layers:"); for frozen_layer in self.frozen_layers.iter() { frozen_layer.dump(verbose, ctx).await?; diff --git a/pageserver/src/tenant/storage_layer/inmemory_layer.rs b/pageserver/src/tenant/storage_layer/inmemory_layer.rs index 066a803c06..87dcd08b0d 100644 --- a/pageserver/src/tenant/storage_layer/inmemory_layer.rs +++ b/pageserver/src/tenant/storage_layer/inmemory_layer.rs @@ -20,6 +20,7 @@ use pageserver_api::models::InMemoryLayerInfo; use pageserver_api::shard::TenantShardId; use std::collections::{BinaryHeap, HashMap, HashSet}; use std::sync::{Arc, OnceLock}; +use std::time::Instant; use tracing::*; use utils::{bin_ser::BeSer, id::TimelineId, lsn::Lsn, vec_map::VecMap}; // avoid binding to Write (conflicts with std::io::Write) @@ -30,7 +31,7 @@ use std::fmt::Write as _; use std::ops::Range; use std::sync::atomic::Ordering as AtomicOrdering; use std::sync::atomic::{AtomicU64, AtomicUsize}; -use tokio::sync::{RwLock, RwLockWriteGuard}; +use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard}; use super::{ DeltaLayerWriter, ResidentLayer, ValueReconstructSituation, ValueReconstructState, @@ -51,6 +52,8 @@ pub struct InMemoryLayer { /// start is inclusive. start_lsn: Lsn, + opened_at: Instant, + /// Frozen layers have an exclusive end LSN. /// Writes are only allowed when this is `None`. end_lsn: OnceLock, @@ -463,6 +466,7 @@ impl InMemoryLayer { timeline_id, tenant_shard_id, start_lsn, + opened_at: Instant::now(), end_lsn: OnceLock::new(), inner: RwLock::new(InMemoryLayerInner { index: HashMap::new(), @@ -618,4 +622,73 @@ impl InMemoryLayer { let delta_layer = delta_layer_writer.finish(Key::MAX, timeline).await?; Ok(delta_layer) } + + pub(crate) async fn should_roll(&self, write_size: u64, projected_lsn: Lsn) -> bool { + let inner = self.inner.read().await; + self.should_roll_locked(inner, write_size, projected_lsn) + } + + pub(crate) async fn try_should_roll( + &self, + write_size: u64, + projected_lsn: Lsn, + ) -> Option { + let inner = self.inner.try_read().ok()?; + Some(self.should_roll_locked(inner, write_size, projected_lsn)) + } + + fn should_roll_locked( + &self, + inner: RwLockReadGuard<'_, InMemoryLayerInner>, + write_size: u64, + projected_lsn: Lsn, + ) -> bool { + let arc_guard = self.tenant_conf.load(); + // TODO: This is quick and dirty. Should only merge the configs we need. + let tenant_conf = arc_guard + .tenant_conf + .merge(self.conf.default_tenant_conf.clone()); + + let layer_size = inner.file.len(); + let projected_layer_size = layer_size + write_size; + let distance = projected_lsn.widening_sub(self.start_lsn); + + // Rolling the open layer can be triggered by: + // 1. The distance from the last LSN we rolled at. This bounds the amount of WAL that + // the safekeepers need to store. For sharded tenants, we multiply by shard count to + // account for how writes are distributed across shards: we expect each node to consume + // 1/count of the LSN on average. + // 2. The size of the currently open layer. + // 3. The time since the last roll. It helps safekeepers to regard pageserver as caught + // up and suspend activity. + if distance + >= tenant_conf.checkpoint_distance as i128 + * self.tenant_shard_id.shard_count.count() as i128 + { + info!( + "Will roll layer at {} with layer size {} due to LSN distance ({})", + projected_lsn, layer_size, distance + ); + + true + } else if projected_layer_size >= tenant_conf.checkpoint_distance { + info!( + "Will roll layer at {} with layer size {} due to layer size ({})", + projected_lsn, layer_size, projected_layer_size + ); + + true + } else if distance > 0 && self.opened_at.elapsed() >= tenant_conf.checkpoint_timeout { + info!( + "Will roll layer at {} with layer size {} due to time since last flush ({:?})", + projected_lsn, + layer_size, + self.opened_at.elapsed() + ); + + true + } else { + false + } + } } diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index 59ce0cdb91..2f1217437f 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -223,6 +223,8 @@ pub struct Timeline { /// so that e.g. on-demand-download/eviction, and layer spreading, can operate just on `LayerFileManager`. pub(crate) layers: Arc>, + open_layer: tokio::sync::RwLock>>, + last_freeze_at: AtomicLsn, // Atomic would be more appropriate here. last_freeze_ts: RwLock, @@ -1184,13 +1186,8 @@ impl Timeline { return; }; - let Ok(layers_guard) = self.layers.try_read() else { - // Don't block if the layer lock is busy - return; - }; - - let Some(open_layer) = &layers_guard.layer_map().open_layer else { - // No open layer, no work to do. + let Some(open_layer) = &*self.open_layer.try_read() else { + // No open layer or the lock is held, no work to do. return; }; @@ -1235,7 +1232,6 @@ impl Timeline { } InMemoryLayerInfo::Open { .. } => { // Upgrade to a write lock and freeze the layer - drop(layers_guard); let mut layers_guard = self.layers.write().await; layers_guard .try_freeze_in_memory_layer(current_lsn, &self.last_freeze_at) @@ -1520,7 +1516,7 @@ impl Timeline { let guard = self.layers.read().await; let layer_map = guard.layer_map(); let mut in_memory_layers = Vec::with_capacity(layer_map.frozen_layers.len() + 1); - if let Some(open_layer) = &layer_map.open_layer { + if let Some(open_layer) = &*self.open_layer.read().await { in_memory_layers.push(open_layer.info()); } for frozen_layer in &layer_map.frozen_layers { @@ -2789,9 +2785,11 @@ impl Timeline { let guard = timeline.layers.read().await; let layers = guard.layer_map(); + // Check the open and frozen in-memory layers first, in order from newest // to oldest. - if let Some(open_layer) = &layers.open_layer { + // TODO: structure this so we don't grab the lock unless we need it + if let Some(open_layer) = &*self.open_layer.read().await { let start_lsn = open_layer.get_lsn_range().start; if cont_lsn > start_lsn { //info!("CHECKING for {} at {} on open layer {}", key, cont_lsn, open_layer.filename().display()); @@ -2989,10 +2987,19 @@ impl Timeline { let guard = timeline.layers.read().await; let layers = guard.layer_map(); - let in_memory_layer = layers.find_in_memory_layer(|l| { - let start_lsn = l.get_lsn_range().start; - cont_lsn > start_lsn - }); + let in_memory_layer = { + let open_layer = timeline.open_layer.read().await; + match open_layer.filter(|l| cont_lsn >= l.get_lsn_range().start).cloned() { + Some(l) => Some(l), + None => { + layers.find_in_memory_layer(|l| { + let start_lsn = l.get_lsn_range().start; + cont_lsn > start_lsn + }) + } + } + + }; match in_memory_layer { Some(l) => {