wip

pageserver/src/tenant/layer_map.rs

@@ -552,12 +552,6 @@ impl LayerMap {
     where
         Pred: FnMut(&Arc<InMemoryLayer>) -> 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?;

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<Lsn>,
@@ -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<bool> {
+        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
+        }
+    }
 }

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<tokio::sync::RwLock<LayerManager>>,
 
+    open_layer: tokio::sync::RwLock<Option<Arc<InMemoryLayer>>>,
+
     last_freeze_at: AtomicLsn,
     // Atomic would be more appropriate here.
     last_freeze_ts: RwLock<Instant>,
@@ -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) => {