refactor: rm pseudo lock in timeline

Signed-off-by: Alex Chi Z <chi@neon.tech>

Cargo.lock

@@ -110,6 +110,12 @@ dependencies = [
  "backtrace",
 ]
 
+[[package]]
+name = "arc-swap"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bddcadddf5e9015d310179a59bb28c4d4b9920ad0f11e8e14dbadf654890c9a6"
+
 [[package]]
 name = "archery"
 version = "0.5.0"
@@ -2542,6 +2548,7 @@ name = "pageserver"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "arc-swap",
  "async-stream",
  "async-trait",
  "byteorder",

Cargo.toml

@@ -32,6 +32,7 @@ license = "Apache-2.0"
 ## All dependency versions, used in the project
 [workspace.dependencies]
 anyhow = { version = "1.0", features = ["backtrace"] }
+arc-swap = "1.6"
 async-stream = "0.3"
 async-trait = "0.1"
 atty = "0.2.14"

pageserver/Cargo.toml

@@ -12,6 +12,7 @@ testing = ["fail/failpoints"]
 
 [dependencies]
 anyhow.workspace = true
+arc-swap.workspace = true
 async-stream.workspace = true
 async-trait.workspace = true
 byteorder.workspace = true

pageserver/src/page_service.rs

@@ -489,7 +489,9 @@ impl PageServerHandler {
         // Create empty timeline
         info!("creating new timeline");
         let tenant = get_active_tenant_with_timeout(tenant_id, &ctx).await?;
-        let timeline = tenant.create_empty_timeline(timeline_id, base_lsn, pg_version, &ctx)?;
+        let timeline = tenant
+            .create_empty_timeline(timeline_id, base_lsn, pg_version, &ctx)
+            .await?;
 
         // TODO mark timeline as not ready until it reaches end_lsn.
         // We might have some wal to import as well, and we should prevent compute

pageserver/src/tenant.rs

@@ -87,6 +87,7 @@ pub mod disk_btree;
 pub(crate) mod ephemeral_file;
 pub mod layer_cache;
 pub mod layer_map;
+pub mod layer_map_mgr;
 pub mod manifest;
 
 pub mod metadata;
@@ -556,17 +557,10 @@ impl Tenant {
                 .context("failed to reconcile with remote")?
         }
 
+        let layers = timeline.layer_mgr.read();
         // Sanity check: a timeline should have some content.
         anyhow::ensure!(
-            ancestor.is_some()
-                || timeline
-                    .layers
-                    .read()
-                    .await
-                    .0
-                    .iter_historic_layers()
-                    .next()
-                    .is_some(),
+            ancestor.is_some() || layers.iter_historic_layers().next().is_some(),
             "Timeline has no ancestor and no layer files"
         );
 
@@ -1247,7 +1241,7 @@ impl Tenant {
     /// For tests, use `DatadirModification::init_empty_test_timeline` + `commit` to setup the
     /// minimum amount of keys required to get a writable timeline.
     /// (Without it, `put` might fail due to `repartition` failing.)
-    pub fn create_empty_timeline(
+    pub async fn create_empty_timeline(
         &self,
         new_timeline_id: TimelineId,
         initdb_lsn: Lsn,
@@ -1259,9 +1253,11 @@ impl Tenant {
             "Cannot create empty timelines on inactive tenant"
         );
 
-        let timelines = self.timelines.lock().unwrap();
-        let timeline_uninit_mark = self.create_timeline_uninit_mark(new_timeline_id, &timelines)?;
-        drop(timelines);
+        let timeline_uninit_mark = {
+            let timelines: MutexGuard<'_, HashMap<TimelineId, Arc<Timeline>>> =
+                self.timelines.lock().unwrap();
+            self.create_timeline_uninit_mark(new_timeline_id, &timelines)?
+        };
 
         let new_metadata = TimelineMetadata::new(
             // Initialize disk_consistent LSN to 0, The caller must import some data to
@@ -1281,6 +1277,7 @@ impl Tenant {
             initdb_lsn,
             None,
         )
+        .await
     }
 
     /// Helper for unit tests to create an emtpy timeline.
@@ -1296,7 +1293,9 @@ impl Tenant {
         pg_version: u32,
         ctx: &RequestContext,
     ) -> anyhow::Result<Arc<Timeline>> {
-        let uninit_tl = self.create_empty_timeline(new_timeline_id, initdb_lsn, pg_version, ctx)?;
+        let uninit_tl = self
+            .create_empty_timeline(new_timeline_id, initdb_lsn, pg_version, ctx)
+            .await?;
         let tline = uninit_tl.raw_timeline().expect("we just created it");
         assert_eq!(tline.get_last_record_lsn(), Lsn(0));
 
@@ -2756,13 +2755,15 @@ impl Tenant {
             src_timeline.pg_version,
         );
 
-        let uninitialized_timeline = self.prepare_new_timeline(
-            dst_id,
-            &metadata,
-            timeline_uninit_mark,
-            start_lsn + 1,
-            Some(Arc::clone(src_timeline)),
-        )?;
+        let uninitialized_timeline = self
+            .prepare_new_timeline(
+                dst_id,
+                &metadata,
+                timeline_uninit_mark,
+                start_lsn + 1,
+                Some(Arc::clone(src_timeline)),
+            )
+            .await?;
 
         let new_timeline = uninitialized_timeline.finish_creation()?;
 
@@ -2840,13 +2841,15 @@ impl Tenant {
             pgdata_lsn,
             pg_version,
         );
-        let raw_timeline = self.prepare_new_timeline(
-            timeline_id,
-            &new_metadata,
-            timeline_uninit_mark,
-            pgdata_lsn,
-            None,
-        )?;
+        let raw_timeline = self
+            .prepare_new_timeline(
+                timeline_id,
+                &new_metadata,
+                timeline_uninit_mark,
+                pgdata_lsn,
+                None,
+            )
+            .await?;
 
         let tenant_id = raw_timeline.owning_tenant.tenant_id;
         let unfinished_timeline = raw_timeline.raw_timeline()?;
@@ -2899,7 +2902,7 @@ impl Tenant {
     /// at 'disk_consistent_lsn'. After any initial data has been imported, call
     /// `finish_creation` to insert the Timeline into the timelines map and to remove the
     /// uninit mark file.
-    fn prepare_new_timeline(
+    async fn prepare_new_timeline(
         &self,
         new_timeline_id: TimelineId,
         new_metadata: &TimelineMetadata,
@@ -2926,7 +2929,7 @@ impl Tenant {
             .create_timeline_struct(new_timeline_id, new_metadata, ancestor, remote_client, None)
             .context("Failed to create timeline data structure")?;
 
-        timeline_struct.init_empty_layer_map(start_lsn);
+        timeline_struct.init_empty_layer_map(start_lsn).await?;
 
         if let Err(e) =
             self.create_timeline_files(&uninit_mark.timeline_path, new_timeline_id, new_metadata)
@@ -3623,7 +3626,10 @@ mod tests {
             .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
             .await?;
 
-        match tenant.create_empty_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx) {
+        match tenant
+            .create_empty_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
+            .await
+        {
             Ok(_) => panic!("duplicate timeline creation should fail"),
             Err(e) => assert_eq!(
                 e.to_string(),
@@ -4423,8 +4429,9 @@ mod tests {
             .await;
 
         let initdb_lsn = Lsn(0x20);
-        let utline =
-            tenant.create_empty_timeline(TIMELINE_ID, initdb_lsn, DEFAULT_PG_VERSION, &ctx)?;
+        let utline = tenant
+            .create_empty_timeline(TIMELINE_ID, initdb_lsn, DEFAULT_PG_VERSION, &ctx)
+            .await?;
         let tline = utline.raw_timeline().unwrap();
 
         // Spawn flush loop now so that we can set the `expect_initdb_optimization`

pageserver/src/tenant/layer_cache.rs

@@ -24,10 +24,6 @@ pub struct LayerCache {
     mapping: Mutex<HashMap<PersistentLayerKey, Arc<dyn PersistentLayer>>>,
 }
 
-pub struct LayerInUseWrite(tokio::sync::OwnedRwLockWriteGuard<()>);
-
-pub struct LayerInUseRead(tokio::sync::OwnedRwLockReadGuard<()>);
-
 #[derive(Clone)]
 pub struct DeleteGuard(Arc<tokio::sync::OwnedMutexGuard<()>>);
 
@@ -46,18 +42,6 @@ impl LayerCache {
         guard.get(&desc.key()).expect("not found").clone()
     }
 
-    /// This function is to mock the original behavior of `layers` lock in `Timeline`. Can be removed after we ensure
-    /// we won't delete files that are being read.
-    pub async fn layer_in_use_write(&self) -> LayerInUseWrite {
-        LayerInUseWrite(self.layers_operation_lock.clone().write_owned().await)
-    }
-
-    /// This function is to mock the original behavior of `layers` lock in `Timeline`. Can be removed after we ensure
-    /// we won't delete files that are being read.
-    pub async fn layer_in_use_read(&self) -> LayerInUseRead {
-        LayerInUseRead(self.layers_operation_lock.clone().read_owned().await)
-    }
-
     /// Ensures only one of compaction / gc can happen at a time.
     pub async fn delete_guard(&self) -> DeleteGuard {
         DeleteGuard(Arc::new(

pageserver/src/tenant/layer_map.rs

@@ -66,7 +66,7 @@ use super::storage_layer::PersistentLayerDesc;
 ///
 /// LayerMap tracks what layers exist on a timeline.
 ///
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct LayerMap {
     //
     // 'open_layer' holds the current InMemoryLayer that is accepting new

pageserver/src/tenant/layer_map/historic_layer_coverage.rs

@@ -72,6 +72,7 @@ impl From<&PersistentLayerDesc> for LayerKey {
 /// Allows answering layer map queries very efficiently,
 /// but doesn't allow retroactive insertion, which is
 /// sometimes necessary. See BufferedHistoricLayerCoverage.
+#[derive(Clone)]
 pub struct HistoricLayerCoverage<Value> {
     /// The latest state
     head: LayerCoverageTuple<Value>,
@@ -425,6 +426,7 @@ fn test_persistent_overlapping() {
 ///
 /// See this for more on persistent and retroactive techniques:
 /// https://www.youtube.com/watch?v=WqCWghETNDc&t=581s
+#[derive(Clone)]
 pub struct BufferedHistoricLayerCoverage<Value> {
     /// A persistent layer map that we rebuild when we need to retroactively update
     historic_coverage: HistoricLayerCoverage<Value>,

pageserver/src/tenant/layer_map/layer_coverage.rs

@@ -15,6 +15,7 @@ use rpds::RedBlackTreeMapSync;
 ///
 /// NOTE The struct is parameterized over Value for easier
 ///      testing, but in practice it's some sort of layer.
+#[derive(Clone)]
 pub struct LayerCoverage<Value> {
     /// For every change in coverage (as we sweep the key space)
     /// we store (lsn.end, value).
@@ -139,6 +140,7 @@ impl<Value: Clone> LayerCoverage<Value> {
 }
 
 /// Image and delta coverage at a specific LSN.
+#[derive(Clone)]
 pub struct LayerCoverageTuple<Value> {
     pub image_coverage: LayerCoverage<Value>,
     pub delta_coverage: LayerCoverage<Value>,

pageserver/src/tenant/layer_map_mgr.rs

@@ -0,0 +1,146 @@
+//! This module implements `LayerMapMgr`, which manages a layer map object and provides lock-free access to the state.
+//!
+//! A common usage pattern is as follows:
+//!
+//! ```ignore
+//! async fn compaction(&self) {
+//!     // Get the current state.
+//!     let state = self.layer_map_mgr.read();
+//!     // No lock held at this point. Do compaction based on the state. This part usually incurs I/O operations and may
+//!     // take a long time.
+//!     let compaction_result = self.do_compaction(&state).await?;
+//!     // Update the state.
+//!     self.layer_map_mgr.update(|mut state| async move {
+//!         // do updates to the state, return it.
+//!         Ok(state)
+//!     }).await?;
+//! }
+//! ```
+use anyhow::Result;
+use arc_swap::ArcSwap;
+use futures::Future;
+use std::sync::Arc;
+
+use super::layer_map::LayerMap;
+
+/// Manages the storage state. Provide utility functions to modify the layer map and get an immutable reference to the
+/// layer map.
+pub struct LayerMapMgr {
+    layer_map: ArcSwap<LayerMap>,
+    state_lock: tokio::sync::Mutex<()>,
+}
+
+impl LayerMapMgr {
+    /// Get the current state of the layer map.
+    pub fn read(&self) -> Arc<LayerMap> {
+        // TODO: it is possible to use `load` to reduce the overhead of cloning the Arc, but read path usually involves
+        // disk reads and layer mapping fetching, and therefore it's not a big deal to use a more optimized version
+        // here.
+        self.layer_map.load_full()
+    }
+
+    /// Clone the layer map for modification.
+    fn clone_for_write(&self, _state_lock_witness: &tokio::sync::MutexGuard<'_, ()>) -> LayerMap {
+        (**self.layer_map.load()).clone()
+    }
+
+    pub fn new(layer_map: LayerMap) -> Self {
+        Self {
+            layer_map: ArcSwap::new(Arc::new(layer_map)),
+            state_lock: tokio::sync::Mutex::new(()),
+        }
+    }
+
+    /// Update the layer map.
+    pub async fn update<O, F>(&self, operation: O) -> Result<()>
+    where
+        O: FnOnce(LayerMap) -> F,
+        F: Future<Output = Result<LayerMap>>,
+    {
+        let state_lock = self.state_lock.lock().await;
+        let state = self.clone_for_write(&state_lock);
+        let new_state = operation(state).await?;
+        self.layer_map.store(Arc::new(new_state));
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use utils::{
+        id::{TenantId, TimelineId},
+        lsn::Lsn,
+    };
+
+    use crate::{repository::Key, tenant::storage_layer::PersistentLayerDesc};
+
+    use super::*;
+
+    #[tokio::test]
+    async fn test_layer_map_manage() -> Result<()> {
+        let mgr = LayerMapMgr::new(Default::default());
+        mgr.update(|mut map| async move {
+            let mut updates = map.batch_update();
+            updates.insert_historic(PersistentLayerDesc::new_img(
+                TenantId::generate(),
+                TimelineId::generate(),
+                Key::from_i128(0)..Key::from_i128(1),
+                Lsn(0),
+                false,
+                0,
+            ));
+            updates.flush();
+            Ok(map)
+        })
+        .await?;
+
+        let ref_1 = mgr.read();
+
+        mgr.update(|mut map| async move {
+            let mut updates = map.batch_update();
+            updates.insert_historic(PersistentLayerDesc::new_img(
+                TenantId::generate(),
+                TimelineId::generate(),
+                Key::from_i128(1)..Key::from_i128(2),
+                Lsn(0),
+                false,
+                0,
+            ));
+            updates.flush();
+            Ok(map)
+        })
+        .await?;
+
+        let ref_2 = mgr.read();
+
+        // Modification should not be visible to the old reference.
+        assert_eq!(
+            ref_1
+                .search(Key::from_i128(0), Lsn(1))
+                .unwrap()
+                .layer
+                .key_range,
+            Key::from_i128(0)..Key::from_i128(1)
+        );
+        assert!(ref_1.search(Key::from_i128(1), Lsn(1)).is_none());
+
+        // Modification should be visible to the new reference.
+        assert_eq!(
+            ref_2
+                .search(Key::from_i128(0), Lsn(1))
+                .unwrap()
+                .layer
+                .key_range,
+            Key::from_i128(0)..Key::from_i128(1)
+        );
+        assert_eq!(
+            ref_2
+                .search(Key::from_i128(1), Lsn(1))
+                .unwrap()
+                .layer
+                .key_range,
+            Key::from_i128(1)..Key::from_i128(2)
+        );
+        Ok(())
+    }
+}

pageserver/src/tenant/storage_layer.rs

@@ -41,8 +41,6 @@ pub use inmemory_layer::InMemoryLayer;
 pub use layer_desc::{PersistentLayerDesc, PersistentLayerKey};
 pub use remote_layer::RemoteLayer;
 
-use super::layer_map::BatchedUpdates;
-
 pub fn range_overlaps<T>(a: &Range<T>, b: &Range<T>) -> bool
 where
     T: PartialOrd<T>,
@@ -176,16 +174,9 @@ impl LayerAccessStats {
     /// Create an empty stats object and record a [`LayerLoad`] event with the given residence status.
     ///
     /// See [`record_residence_event`] for why you need to do this while holding the layer map lock.
-    pub(crate) fn for_loading_layer(
-        layer_map_lock_held_witness: &BatchedUpdates<'_>,
-        status: LayerResidenceStatus,
-    ) -> Self {
+    pub(crate) fn for_loading_layer(status: LayerResidenceStatus) -> Self {
         let new = LayerAccessStats(Mutex::new(LayerAccessStatsLocked::default()));
-        new.record_residence_event(
-            layer_map_lock_held_witness,
-            status,
-            LayerResidenceEventReason::LayerLoad,
-        );
+        new.record_residence_event(status, LayerResidenceEventReason::LayerLoad);
         new
     }
 
@@ -196,7 +187,6 @@ impl LayerAccessStats {
     /// See [`record_residence_event`] for why you need to do this while holding the layer map lock.
     pub(crate) fn clone_for_residence_change(
         &self,
-        layer_map_lock_held_witness: &BatchedUpdates<'_>,
         new_status: LayerResidenceStatus,
     ) -> LayerAccessStats {
         let clone = {
@@ -204,11 +194,7 @@ impl LayerAccessStats {
             inner.clone()
         };
         let new = LayerAccessStats(Mutex::new(clone));
-        new.record_residence_event(
-            layer_map_lock_held_witness,
-            new_status,
-            LayerResidenceEventReason::ResidenceChange,
-        );
+        new.record_residence_event(new_status, LayerResidenceEventReason::ResidenceChange);
         new
     }
 
@@ -228,7 +214,6 @@ impl LayerAccessStats {
     ///
     pub(crate) fn record_residence_event(
         &self,
-        _layer_map_lock_held_witness: &BatchedUpdates<'_>,
         status: LayerResidenceStatus,
         reason: LayerResidenceEventReason,
     ) {

pageserver/src/tenant/storage_layer/remote_layer.rs

@@ -4,7 +4,6 @@
 use crate::config::PageServerConf;
 use crate::context::RequestContext;
 use crate::repository::Key;
-use crate::tenant::layer_map::BatchedUpdates;
 use crate::tenant::remote_timeline_client::index::LayerFileMetadata;
 use crate::tenant::storage_layer::{Layer, ValueReconstructResult, ValueReconstructState};
 use anyhow::{bail, Result};
@@ -220,7 +219,6 @@ impl RemoteLayer {
     /// Create a Layer struct representing this layer, after it has been downloaded.
     pub fn create_downloaded_layer(
         &self,
-        layer_map_lock_held_witness: &BatchedUpdates<'_>,
         conf: &'static PageServerConf,
         file_size: u64,
     ) -> Arc<dyn PersistentLayer> {
@@ -232,10 +230,8 @@ impl RemoteLayer {
                 self.desc.tenant_id,
                 &fname,
                 file_size,
-                self.access_stats.clone_for_residence_change(
-                    layer_map_lock_held_witness,
-                    LayerResidenceStatus::Resident,
-                ),
+                self.access_stats
+                    .clone_for_residence_change(LayerResidenceStatus::Resident),
             ))
         } else {
             let fname = self.desc.image_file_name();
@@ -245,10 +241,8 @@ impl RemoteLayer {
                 self.desc.tenant_id,
                 &fname,
                 file_size,
-                self.access_stats.clone_for_residence_change(
-                    layer_map_lock_held_witness,
-                    LayerResidenceStatus::Resident,
-                ),
+                self.access_stats
+                    .clone_for_residence_change(LayerResidenceStatus::Resident),
             ))
         }
     }

pageserver/src/tenant/timeline.rs

@@ -81,6 +81,7 @@ use self::walreceiver::{WalReceiver, WalReceiverConf};
 use super::config::TenantConf;
 use super::layer_cache::{DeleteGuard, LayerCache};
 use super::layer_map::BatchedUpdates;
+use super::layer_map_mgr::LayerMapMgr;
 use super::remote_timeline_client::index::IndexPart;
 use super::remote_timeline_client::RemoteTimelineClient;
 use super::storage_layer::{
@@ -118,26 +119,6 @@ impl PartialOrd for Hole {
     }
 }
 
-pub struct LayerMapping(());
-
-impl LayerMapping {
-    pub(crate) fn new() -> Self {
-        Self(())
-    }
-}
-
-/// Temporary function for immutable storage state refactor, ensures we are dropping mutex guard instead of other things.
-/// Can be removed after all refactors are done.
-fn drop_rlock<T>(rlock: tokio::sync::RwLockReadGuard<'_, T>) {
-    drop(rlock)
-}
-
-/// Temporary function for immutable storage state refactor, ensures we are dropping mutex guard instead of other things.
-/// Can be removed after all refactors are done.
-fn drop_wlock<T>(rlock: tokio::sync::RwLockWriteGuard<'_, T>) {
-    drop(rlock)
-}
-
 pub struct Timeline {
     pub(super) conf: &'static PageServerConf,
     tenant_conf: Arc<RwLock<TenantConfOpt>>,
@@ -149,8 +130,7 @@ pub struct Timeline {
 
     pub pg_version: u32,
 
-    pub(crate) layers: tokio::sync::RwLock<(LayerMap, LayerMapping)>,
-
+    pub(super) layer_mgr: LayerMapMgr,
     pub(super) lcache: LayerCache,
 
     /// Set of key ranges which should be covered by image layers to
@@ -617,8 +597,7 @@ impl Timeline {
     /// This method makes no distinction between local and remote layers.
     /// Hence, the result **does not represent local filesystem usage**.
     pub async fn layer_size_sum(&self) -> u64 {
-        let guard = self.layers.read().await;
-        let (layer_map, _) = &*guard;
+        let layer_map = self.layer_mgr.read();
         let mut size = 0;
         for l in layer_map.iter_historic_layers() {
             size += l.file_size();
@@ -928,8 +907,7 @@ impl Timeline {
     pub async fn check_checkpoint_distance(self: &Arc<Timeline>) -> anyhow::Result<()> {
         let last_lsn = self.get_last_record_lsn();
         let open_layer_size = {
-            let guard = self.layers.read().await;
-            let (layers, _) = &*guard;
+            let layers = self.layer_mgr.read();
             let Some(open_layer) = layers.open_layer.as_ref() else {
                 return Ok(());
             };
@@ -1060,8 +1038,7 @@ impl Timeline {
     }
 
     pub async fn layer_map_info(&self, reset: LayerAccessStatsReset) -> LayerMapInfo {
-        let guard = self.layers.read().await;
-        let (layer_map, _) = &*guard;
+        let layer_map = self.layer_mgr.read();
         let mut in_memory_layers = Vec::with_capacity(layer_map.frozen_layers.len() + 1);
         if let Some(open_layer) = &layer_map.open_layer {
             in_memory_layers.push(open_layer.info());
@@ -1181,25 +1158,33 @@ impl Timeline {
             );
         }
 
-        // start the batch update
-        let mut guard = self.layers.write().await;
-        let (layer_map, _) = &mut *guard;
-        let mut batch_updates = layer_map.batch_update();
-
         let mut results = Vec::with_capacity(layers_to_evict.len());
 
-        for l in layers_to_evict.iter() {
-            let res = if cancel.is_cancelled() {
-                None
-            } else {
-                Some(self.evict_layer_batch_impl(&layer_removal_guard, l, &mut batch_updates))
-            };
-            results.push(res);
-        }
+        // start the batch update
+        self.layer_mgr
+            .update(|mut layer_map| async {
+                let mut batch_updates = layer_map.batch_update();
 
-        // commit the updates & release locks
-        batch_updates.flush();
-        drop_wlock(guard);
+                for l in layers_to_evict.iter() {
+                    let res = if cancel.is_cancelled() {
+                        None
+                    } else {
+                        Some(self.evict_layer_batch_impl(
+                            &layer_removal_guard,
+                            l,
+                            &mut batch_updates,
+                        ))
+                    };
+                    results.push(res);
+                }
+
+                // commit the updates & release locks
+                batch_updates.flush();
+                Ok(layer_map)
+            })
+            .await?;
+
+        // drop(guard);
         drop(layer_removal_guard);
 
         assert_eq!(results.len(), layers_to_evict.len());
@@ -1246,7 +1231,7 @@ impl Timeline {
                 &layer_metadata,
                 local_layer
                     .access_stats()
-                    .clone_for_residence_change(batch_updates, LayerResidenceStatus::Evicted),
+                    .clone_for_residence_change(LayerResidenceStatus::Evicted),
             ),
             LayerFileName::Delta(delta_name) => RemoteLayer::new_delta(
                 self.tenant_id,
@@ -1255,7 +1240,7 @@ impl Timeline {
                 &layer_metadata,
                 local_layer
                     .access_stats()
-                    .clone_for_residence_change(batch_updates, LayerResidenceStatus::Evicted),
+                    .clone_for_residence_change(LayerResidenceStatus::Evicted),
             ),
         });
 
@@ -1432,7 +1417,7 @@ impl Timeline {
                 timeline_id,
                 tenant_id,
                 pg_version,
-                layers: tokio::sync::RwLock::new((LayerMap::default(), LayerMapping::new())),
+                layer_mgr: LayerMapMgr::new(LayerMap::default()),
                 lcache: LayerCache::new(myself.clone()),
                 wanted_image_layers: Mutex::new(None),
 
@@ -1612,133 +1597,139 @@ impl Timeline {
     ///
     /// Initialize with an empty layer map. Used when creating a new timeline.
     ///
-    pub(super) fn init_empty_layer_map(&self, start_lsn: Lsn) {
-        let mut layers = self.layers.try_write().expect(
-            "in the context where we call this function, no other task has access to the object",
-        );
-        layers.0.next_open_layer_at = Some(Lsn(start_lsn.0));
+    pub(super) async fn init_empty_layer_map(&self, start_lsn: Lsn) -> anyhow::Result<()> {
+        self.layer_mgr
+            .update(|mut layers| async move {
+                layers.next_open_layer_at = Some(Lsn(start_lsn.0));
+                Ok(layers)
+            })
+            .await
     }
 
     ///
     /// Scan the timeline directory to populate the layer map.
     ///
     pub(super) async fn load_layer_map(&self, disk_consistent_lsn: Lsn) -> anyhow::Result<()> {
-        let mut guard = self.layers.write().await;
-        let (layers, _) = &mut *guard;
-        let mut updates = layers.batch_update();
-        let mut num_layers = 0;
+        self.layer_mgr
+            .update(|mut layers| async move {
+                let mut updates = layers.batch_update();
+                let mut num_layers = 0;
 
-        let timer = self.metrics.load_layer_map_histo.start_timer();
+                let timer = self.metrics.load_layer_map_histo.start_timer();
 
-        // Scan timeline directory and create ImageFileName and DeltaFilename
-        // structs representing all files on disk
-        let timeline_path = self.conf.timeline_path(&self.timeline_id, &self.tenant_id);
-        // total size of layer files in the current timeline directory
-        let mut total_physical_size = 0;
+                // Scan timeline directory and create ImageFileName and DeltaFilename
+                // structs representing all files on disk
+                let timeline_path = self.conf.timeline_path(&self.timeline_id, &self.tenant_id);
+                // total size of layer files in the current timeline directory
+                let mut total_physical_size = 0;
 
-        for direntry in fs::read_dir(timeline_path)? {
-            let direntry = direntry?;
-            let direntry_path = direntry.path();
-            let fname = direntry.file_name();
-            let fname = fname.to_string_lossy();
+                for direntry in fs::read_dir(timeline_path)? {
+                    let direntry = direntry?;
+                    let direntry_path = direntry.path();
+                    let fname = direntry.file_name();
+                    let fname = fname.to_string_lossy();
 
-            if let Some(imgfilename) = ImageFileName::parse_str(&fname) {
-                // create an ImageLayer struct for each image file.
-                if imgfilename.lsn > disk_consistent_lsn {
-                    warn!(
+                    if let Some(imgfilename) = ImageFileName::parse_str(&fname) {
+                        // create an ImageLayer struct for each image file.
+                        if imgfilename.lsn > disk_consistent_lsn {
+                            warn!(
                         "found future image layer {} on timeline {} disk_consistent_lsn is {}",
                         imgfilename, self.timeline_id, disk_consistent_lsn
                     );
 
-                    rename_to_backup(&direntry_path)?;
-                    continue;
-                }
+                            rename_to_backup(&direntry_path)?;
+                            continue;
+                        }
 
-                let file_size = direntry_path.metadata()?.len();
+                        let file_size = direntry_path.metadata()?.len();
 
-                let layer = ImageLayer::new(
-                    self.conf,
-                    self.timeline_id,
-                    self.tenant_id,
-                    &imgfilename,
-                    file_size,
-                    LayerAccessStats::for_loading_layer(&updates, LayerResidenceStatus::Resident),
-                );
+                        let layer = ImageLayer::new(
+                            self.conf,
+                            self.timeline_id,
+                            self.tenant_id,
+                            &imgfilename,
+                            file_size,
+                            LayerAccessStats::for_loading_layer(LayerResidenceStatus::Resident),
+                        );
 
-                trace!("found layer {}", layer.path().display());
-                total_physical_size += file_size;
-                updates.insert_historic(layer.layer_desc().clone());
-                self.lcache.populate_local_when_init(Arc::new(layer));
-                num_layers += 1;
-            } else if let Some(deltafilename) = DeltaFileName::parse_str(&fname) {
-                // Create a DeltaLayer struct for each delta file.
-                // The end-LSN is exclusive, while disk_consistent_lsn is
-                // inclusive. For example, if disk_consistent_lsn is 100, it is
-                // OK for a delta layer to have end LSN 101, but if the end LSN
-                // is 102, then it might not have been fully flushed to disk
-                // before crash.
-                if deltafilename.lsn_range.end > disk_consistent_lsn + 1 {
-                    warn!(
+                        trace!("found layer {}", layer.path().display());
+                        total_physical_size += file_size;
+                        updates.insert_historic(layer.layer_desc().clone());
+                        self.lcache.populate_local_when_init(Arc::new(layer));
+                        num_layers += 1;
+                    } else if let Some(deltafilename) = DeltaFileName::parse_str(&fname) {
+                        // Create a DeltaLayer struct for each delta file.
+                        // The end-LSN is exclusive, while disk_consistent_lsn is
+                        // inclusive. For example, if disk_consistent_lsn is 100, it is
+                        // OK for a delta layer to have end LSN 101, but if the end LSN
+                        // is 102, then it might not have been fully flushed to disk
+                        // before crash.
+                        if deltafilename.lsn_range.end > disk_consistent_lsn + 1 {
+                            warn!(
                         "found future delta layer {} on timeline {} disk_consistent_lsn is {}",
                         deltafilename, self.timeline_id, disk_consistent_lsn
                     );
 
-                    rename_to_backup(&direntry_path)?;
-                    continue;
-                }
+                            rename_to_backup(&direntry_path)?;
+                            continue;
+                        }
 
-                let file_size = direntry_path.metadata()?.len();
+                        let file_size = direntry_path.metadata()?.len();
 
-                let layer = DeltaLayer::new(
-                    self.conf,
-                    self.timeline_id,
-                    self.tenant_id,
-                    &deltafilename,
-                    file_size,
-                    LayerAccessStats::for_loading_layer(&updates, LayerResidenceStatus::Resident),
-                );
+                        let layer = DeltaLayer::new(
+                            self.conf,
+                            self.timeline_id,
+                            self.tenant_id,
+                            &deltafilename,
+                            file_size,
+                            LayerAccessStats::for_loading_layer(LayerResidenceStatus::Resident),
+                        );
 
-                trace!("found layer {}", layer.path().display());
-                total_physical_size += file_size;
-                updates.insert_historic(layer.layer_desc().clone());
-                self.lcache.populate_local_when_init(Arc::new(layer));
-                num_layers += 1;
-            } else if fname == METADATA_FILE_NAME || fname.ends_with(".old") {
-                // ignore these
-            } else if remote_timeline_client::is_temp_download_file(&direntry_path) {
-                info!(
+                        trace!("found layer {}", layer.path().display());
+                        total_physical_size += file_size;
+                        updates.insert_historic(layer.layer_desc().clone());
+                        self.lcache.populate_local_when_init(Arc::new(layer));
+                        num_layers += 1;
+                    } else if fname == METADATA_FILE_NAME || fname.ends_with(".old") {
+                        // ignore these
+                    } else if remote_timeline_client::is_temp_download_file(&direntry_path) {
+                        info!(
                     "skipping temp download file, reconcile_with_remote will resume / clean up: {}",
                     fname
                 );
-            } else if is_ephemeral_file(&fname) {
-                // Delete any old ephemeral files
-                trace!("deleting old ephemeral file in timeline dir: {}", fname);
-                fs::remove_file(&direntry_path)?;
-            } else if is_temporary(&direntry_path) {
-                info!("removing temp timeline file at {}", direntry_path.display());
-                fs::remove_file(&direntry_path).with_context(|| {
-                    format!(
-                        "failed to remove temp download file at {}",
-                        direntry_path.display()
-                    )
-                })?;
-            } else {
-                warn!("unrecognized filename in timeline dir: {}", fname);
-            }
-        }
+                    } else if is_ephemeral_file(&fname) {
+                        // Delete any old ephemeral files
+                        trace!("deleting old ephemeral file in timeline dir: {}", fname);
+                        fs::remove_file(&direntry_path)?;
+                    } else if is_temporary(&direntry_path) {
+                        info!("removing temp timeline file at {}", direntry_path.display());
+                        fs::remove_file(&direntry_path).with_context(|| {
+                            format!(
+                                "failed to remove temp download file at {}",
+                                direntry_path.display()
+                            )
+                        })?;
+                    } else {
+                        warn!("unrecognized filename in timeline dir: {}", fname);
+                    }
+                }
 
-        updates.flush();
-        layers.next_open_layer_at = Some(Lsn(disk_consistent_lsn.0) + 1);
+                updates.flush();
+                layers.next_open_layer_at = Some(Lsn(disk_consistent_lsn.0) + 1);
 
-        info!(
-            "loaded layer map with {} layers at {}, total physical size: {}",
-            num_layers, disk_consistent_lsn, total_physical_size
-        );
-        self.metrics
-            .resident_physical_size_gauge
-            .set(total_physical_size);
+                info!(
+                    "loaded layer map with {} layers at {}, total physical size: {}",
+                    num_layers, disk_consistent_lsn, total_physical_size
+                );
+                self.metrics
+                    .resident_physical_size_gauge
+                    .set(total_physical_size);
 
-        timer.stop_and_record();
+                timer.stop_and_record();
+
+                Ok(layers)
+            })
+            .await?;
 
         Ok(())
     }
@@ -1755,9 +1746,9 @@ impl Timeline {
 
         // We're holding a layer map lock for a while but this
         // method is only called during init so it's fine.
-        let mut guard = self.layers.write().await;
-        let (layer_map, _) = &mut *guard;
-        let mut updates = layer_map.batch_update();
+        let mut layers_to_remove = vec![];
+        let mut layers_to_insert = vec![];
+
         for remote_layer_name in &index_part.timeline_layers {
             let local_layer = local_only_layers.remove(remote_layer_name);
 
@@ -1801,7 +1792,7 @@ impl Timeline {
                         anyhow::bail!("could not rename file {local_layer_path:?}: {err:?}");
                     } else {
                         self.metrics.resident_physical_size_gauge.sub(local_size);
-                        updates.remove_historic(local_layer.layer_desc().clone());
+                        layers_to_remove.push(local_layer.layer_desc().clone());
                         self.lcache.remove_local_when_init(local_layer);
                         // fall-through to adding the remote layer
                     }
@@ -1834,14 +1825,10 @@ impl Timeline {
                         self.timeline_id,
                         imgfilename,
                         &remote_layer_metadata,
-                        LayerAccessStats::for_loading_layer(
-                            &updates,
-                            LayerResidenceStatus::Evicted,
-                        ),
+                        LayerAccessStats::for_loading_layer(LayerResidenceStatus::Evicted),
                     );
                     let remote_layer = Arc::new(remote_layer);
-
-                    updates.insert_historic(remote_layer.layer_desc().clone());
+                    layers_to_insert.push(remote_layer.layer_desc().clone());
                     self.lcache.populate_remote_when_init(remote_layer);
                 }
                 LayerFileName::Delta(deltafilename) => {
@@ -1863,19 +1850,29 @@ impl Timeline {
                         self.timeline_id,
                         deltafilename,
                         &remote_layer_metadata,
-                        LayerAccessStats::for_loading_layer(
-                            &updates,
-                            LayerResidenceStatus::Evicted,
-                        ),
+                        LayerAccessStats::for_loading_layer(LayerResidenceStatus::Evicted),
                     );
                     let remote_layer = Arc::new(remote_layer);
-                    updates.insert_historic(remote_layer.layer_desc().clone());
+                    layers_to_insert.push(remote_layer.layer_desc().clone());
                     self.lcache.populate_remote_when_init(remote_layer);
                 }
             }
         }
 
-        updates.flush();
+        let operation = |mut layer_map: LayerMap| async {
+            let mut updates = layer_map.batch_update();
+            for layer in layers_to_remove {
+                updates.remove_historic(layer);
+            }
+            for layer in layers_to_insert {
+                updates.insert_historic(layer);
+            }
+            updates.flush();
+            Ok(layer_map)
+        };
+
+        self.layer_mgr.update(operation).await?;
+
         Ok(local_only_layers)
     }
 
@@ -1910,8 +1907,7 @@ impl Timeline {
         let disk_consistent_lsn = up_to_date_metadata.disk_consistent_lsn();
 
         let local_layers = {
-            let guard = self.layers.read().await;
-            let (layers, _) = &*guard;
+            let layers = self.layer_mgr.read();
             layers
                 .iter_historic_layers()
                 .map(|l| (l.filename(), self.lcache.get_from_desc(&l)))
@@ -2287,8 +2283,7 @@ impl Timeline {
     }
 
     async fn find_layer(&self, layer_file_name: &str) -> Option<Arc<dyn PersistentLayer>> {
-        let guard = self.layers.read().await;
-        let (layers, _) = &*guard;
+        let layers = self.layer_mgr.read();
         for historic_layer in layers.iter_historic_layers() {
             let historic_layer_name = historic_layer.filename().file_name();
             if layer_file_name == historic_layer_name {
@@ -2508,8 +2503,7 @@ impl Timeline {
             #[allow(clippy::never_loop)] // see comment at bottom of this loop
             'layer_map_search: loop {
                 let remote_layer = {
-                    let guard = timeline.layers.read().await;
-                    let (layers, _) = &*guard;
+                    let layers = timeline.layer_mgr.read();
 
                     // Check the open and frozen in-memory layers first, in order from newest
                     // to oldest.
@@ -2693,8 +2687,7 @@ impl Timeline {
     /// Get a handle to the latest layer for appending.
     ///
     async fn get_layer_for_write(&self, lsn: Lsn) -> anyhow::Result<Arc<InMemoryLayer>> {
-        let mut guard = self.layers.write().await;
-        let (layers, _) = &mut *guard;
+        let layers = self.layer_mgr.read();
 
         ensure!(lsn.is_aligned());
 
@@ -2735,8 +2728,13 @@ impl Timeline {
                 InMemoryLayer::create(self.conf, self.timeline_id, self.tenant_id, start_lsn)?;
             let layer_rc = Arc::new(new_layer);
 
-            layers.open_layer = Some(Arc::clone(&layer_rc));
-            layers.next_open_layer_at = None;
+            self.layer_mgr
+                .update(|mut layers| async {
+                    layers.open_layer = Some(Arc::clone(&layer_rc));
+                    layers.next_open_layer_at = None;
+                    Ok(layers)
+                })
+                .await?;
 
             layer = layer_rc;
         }
@@ -2771,22 +2769,28 @@ impl Timeline {
         } else {
             Some(self.write_lock.lock().await)
         };
-        let mut guard = self.layers.write().await;
-        let (layers, _) = &mut *guard;
+        let layers = self.layer_mgr.read();
+
         if let Some(open_layer) = &layers.open_layer {
             let open_layer_rc = Arc::clone(open_layer);
             // Does this layer need freezing?
             let end_lsn = Lsn(self.get_last_record_lsn().0 + 1);
             open_layer.freeze(end_lsn);
 
-            // The layer is no longer open, update the layer map to reflect this.
-            // We will replace it with on-disk historics below.
-            layers.frozen_layers.push_back(open_layer_rc);
-            layers.open_layer = None;
-            layers.next_open_layer_at = Some(end_lsn);
+            self.layer_mgr
+                .update(|mut layers| async move {
+                    // The layer is no longer open, update the layer map to reflect this.
+                    // We will replace it with on-disk historics below.
+                    layers.frozen_layers.push_back(open_layer_rc);
+                    layers.open_layer = None;
+                    layers.next_open_layer_at = Some(end_lsn);
+                    Ok(layers)
+                })
+                .await
+                .unwrap();
+
             self.last_freeze_at.store(end_lsn);
         }
-        drop_wlock(guard);
     }
 
     /// Layer flusher task's main loop.
@@ -2810,8 +2814,7 @@ impl Timeline {
             let flush_counter = *layer_flush_start_rx.borrow();
             let result = loop {
                 let layer_to_flush = {
-                    let guard = self.layers.read().await;
-                    let (layers, _) = &*guard;
+                    let layers = self.layer_mgr.read();
                     layers.frozen_layers.front().cloned()
                     // drop 'layers' lock to allow concurrent reads and writes
                 };
@@ -2936,14 +2939,19 @@ impl Timeline {
         // The new on-disk layers are now in the layer map. We can remove the
         // in-memory layer from the map now.
         {
-            let mut layers = self.layers.write().await;
-            let l = layers.0.frozen_layers.pop_front();
+            self.layer_mgr
+                .update(|mut layers| async {
+                    let l = layers.frozen_layers.pop_front();
 
-            // Only one thread may call this function at a time (for this
-            // timeline). If two threads tried to flush the same frozen
-            // layer to disk at the same time, that would not work.
-            assert!(layer_map::compare_arced_layers(&l.unwrap(), &frozen_layer));
+                    // Only one thread may call this function at a time (for this
+                    // timeline). If two threads tried to flush the same frozen
+                    // layer to disk at the same time, that would not work.
+                    assert!(layer_map::compare_arced_layers(&l.unwrap(), &frozen_layer));
 
+                    Ok(layers)
+                })
+                .await
+                .unwrap();
             // release lock on 'layers'
         }
 
@@ -3073,17 +3081,19 @@ impl Timeline {
 
         // Add it to the layer map
         let l = Arc::new(new_delta);
-        let mut guard = self.layers.write().await;
-        let (layers, _) = &mut *guard;
-        let mut batch_updates = layers.batch_update();
-        l.access_stats().record_residence_event(
-            &batch_updates,
-            LayerResidenceStatus::Resident,
-            LayerResidenceEventReason::LayerCreate,
-        );
-        batch_updates.insert_historic(l.layer_desc().clone());
-        self.lcache.create_new_layer(l);
-        batch_updates.flush();
+        self.layer_mgr
+            .update(|mut layers| async {
+                let mut batch_updates = layers.batch_update();
+                l.access_stats().record_residence_event(
+                    LayerResidenceStatus::Resident,
+                    LayerResidenceEventReason::LayerCreate,
+                );
+                batch_updates.insert_historic(l.layer_desc().clone());
+                self.lcache.create_new_layer(l);
+                batch_updates.flush();
+                Ok(layers)
+            })
+            .await?;
 
         // update the timeline's physical size
         self.metrics.resident_physical_size_gauge.add(sz);
@@ -3132,8 +3142,7 @@ impl Timeline {
     ) -> anyhow::Result<bool> {
         let threshold = self.get_image_creation_threshold();
 
-        let guard = self.layers.read().await;
-        let (layers, _) = &*guard;
+        let layers = self.layer_mgr.read();
 
         let mut max_deltas = 0;
         {
@@ -3311,34 +3320,38 @@ impl Timeline {
 
         let mut layer_paths_to_upload = HashMap::with_capacity(image_layers.len());
 
-        let mut guard = self.layers.write().await;
-        let (layers, _) = &mut *guard;
-        let mut updates = layers.batch_update();
-        let timeline_path = self.conf.timeline_path(&self.timeline_id, &self.tenant_id);
+        self.layer_mgr
+            .update(|mut layers| async {
+                let mut updates = layers.batch_update();
+                let timeline_path = self.conf.timeline_path(&self.timeline_id, &self.tenant_id);
 
-        for l in image_layers {
-            let path = l.filename();
-            let metadata = timeline_path
-                .join(path.file_name())
-                .metadata()
-                .with_context(|| format!("reading metadata of layer file {}", path.file_name()))?;
+                for l in image_layers {
+                    let path = l.filename();
+                    let metadata = timeline_path
+                        .join(path.file_name())
+                        .metadata()
+                        .with_context(|| {
+                            format!("reading metadata of layer file {}", path.file_name())
+                        })?;
 
-            layer_paths_to_upload.insert(path, LayerFileMetadata::new(metadata.len()));
+                    layer_paths_to_upload.insert(path, LayerFileMetadata::new(metadata.len()));
 
-            self.metrics
-                .resident_physical_size_gauge
-                .add(metadata.len());
-            let l = Arc::new(l);
-            l.access_stats().record_residence_event(
-                &updates,
-                LayerResidenceStatus::Resident,
-                LayerResidenceEventReason::LayerCreate,
-            );
-            updates.insert_historic(l.layer_desc().clone());
-            self.lcache.create_new_layer(l);
-        }
-        updates.flush();
-        drop_wlock(guard);
+                    self.metrics
+                        .resident_physical_size_gauge
+                        .add(metadata.len());
+                    let l = Arc::new(l);
+                    l.access_stats().record_residence_event(
+                        LayerResidenceStatus::Resident,
+                        LayerResidenceEventReason::LayerCreate,
+                    );
+                    updates.insert_historic(l.layer_desc().clone());
+                    self.lcache.create_new_layer(l);
+                }
+                updates.flush();
+                Ok(layers)
+            })
+            .await?;
+        // drop(guard);
         timer.stop_and_record();
 
         Ok(layer_paths_to_upload)
@@ -3381,8 +3394,7 @@ impl Timeline {
         target_file_size: u64,
         ctx: &RequestContext,
     ) -> Result<CompactLevel0Phase1Result, CompactionError> {
-        let guard = self.layers.read().await;
-        let (layers, _) = &*guard;
+        let layers = self.layer_mgr.read();
         let mut level0_deltas = layers.get_level0_deltas()?;
 
         // Only compact if enough layers have accumulated.
@@ -3445,7 +3457,7 @@ impl Timeline {
             .map(|l| self.lcache.get_from_desc(l))
             .collect_vec();
 
-        drop_rlock(guard);
+        // drop(guard);
 
         if !remotes.is_empty() {
             // caller is holding the lock to layer_removal_cs, and we don't want to download while
@@ -3512,8 +3524,7 @@ impl Timeline {
         // Determine N largest holes where N is number of compacted layers.
         let max_holes = deltas_to_compact.len();
         let last_record_lsn = self.get_last_record_lsn();
-        let guard = self.layers.read().await; // Is'n it better to hold original layers lock till here?
-        let (layers, _) = &*guard;
+        let layers = self.layer_mgr.read();
         let min_hole_range = (target_file_size / page_cache::PAGE_SZ as u64) as i128;
         let min_hole_coverage_size = 3; // TODO: something more flexible?
 
@@ -3546,7 +3557,7 @@ impl Timeline {
             }
             prev = Some(next_key.next());
         }
-        drop_rlock(guard);
+        // drop(guard);
         let mut holes = heap.into_vec();
         holes.sort_unstable_by_key(|hole| hole.key_range.start);
         let mut next_hole = 0; // index of next hole in holes vector
@@ -3750,52 +3761,54 @@ impl Timeline {
                 .context("wait for layer upload ops to complete")?;
         }
 
-        let mut guard = self.layers.write().await;
-        let (layers, _) = &mut *guard;
-        let mut updates = layers.batch_update();
-        let mut new_layer_paths = HashMap::with_capacity(new_layers.len());
-        for l in new_layers {
-            let new_delta_path = l.path();
-
-            let metadata = new_delta_path.metadata().with_context(|| {
-                format!(
-                    "read file metadata for new created layer {}",
-                    new_delta_path.display()
-                )
-            })?;
-
-            if let Some(remote_client) = &self.remote_client {
-                remote_client.schedule_layer_file_upload(
-                    &l.filename(),
-                    &LayerFileMetadata::new(metadata.len()),
-                )?;
-            }
-
-            // update the timeline's physical size
-            self.metrics
-                .resident_physical_size_gauge
-                .add(metadata.len());
-
-            new_layer_paths.insert(new_delta_path, LayerFileMetadata::new(metadata.len()));
-            let x: Arc<dyn PersistentLayer + 'static> = Arc::new(l);
-            x.access_stats().record_residence_event(
-                &updates,
-                LayerResidenceStatus::Resident,
-                LayerResidenceEventReason::LayerCreate,
-            );
-            updates.insert_historic(x.layer_desc().clone());
-            self.lcache.create_new_layer(x);
-        }
-
-        // Now that we have reshuffled the data to set of new delta layers, we can
-        // delete the old ones
         let mut layer_names_to_delete = Vec::with_capacity(deltas_to_compact.len());
-        for l in deltas_to_compact {
-            layer_names_to_delete.push(l.filename());
-            self.delete_historic_layer(layer_removal_cs.clone(), l, &mut updates)?;
-        }
-        updates.flush();
-        drop_wlock(guard);
+        self.layer_mgr
+            .update(|mut layers| async {
+                let mut updates = layers.batch_update();
+                let mut new_layer_paths = HashMap::with_capacity(new_layers.len());
+                for l in new_layers {
+                    let new_delta_path = l.path();
+
+                    let metadata = new_delta_path.metadata().with_context(|| {
+                        format!(
+                            "read file metadata for new created layer {}",
+                            new_delta_path.display()
+                        )
+                    })?;
+
+                    if let Some(remote_client) = &self.remote_client {
+                        remote_client.schedule_layer_file_upload(
+                            &l.filename(),
+                            &LayerFileMetadata::new(metadata.len()),
+                        )?;
+                    }
+
+                    // update the timeline's physical size
+                    self.metrics
+                        .resident_physical_size_gauge
+                        .add(metadata.len());
+
+                    new_layer_paths.insert(new_delta_path, LayerFileMetadata::new(metadata.len()));
+                    let x: Arc<dyn PersistentLayer + 'static> = Arc::new(l);
+                    x.access_stats().record_residence_event(
+                        LayerResidenceStatus::Resident,
+                        LayerResidenceEventReason::LayerCreate,
+                    );
+                    updates.insert_historic(x.layer_desc().clone());
+                    self.lcache.create_new_layer(x);
+                }
+
+                // Now that we have reshuffled the data to set of new delta layers, we can
+                // delete the old ones
+                for l in deltas_to_compact {
+                    layer_names_to_delete.push(l.filename());
+                    self.delete_historic_layer(layer_removal_cs.clone(), l, &mut updates)?;
+                }
+                updates.flush();
+                Ok(layers)
+            })
+            .await?;
+        // drop(guard);
 
         // Also schedule the deletions in remote storage
         if let Some(remote_client) = &self.remote_client {
@@ -4011,8 +4024,7 @@ impl Timeline {
         // 4. newer on-disk image layers cover the layer's whole key range
         //
         // TODO holding a write lock is too agressive and avoidable
-        let mut guard = self.layers.write().await;
-        let (layers, _) = &mut *guard;
+        let layers = self.layer_mgr.read();
         'outer: for l in layers.iter_historic_layers() {
             result.layers_total += 1;
 
@@ -4111,37 +4123,42 @@ impl Timeline {
             .unwrap()
             .replace((new_gc_cutoff, wanted_image_layers.to_keyspace()));
 
-        let mut updates = layers.batch_update();
-        if !layers_to_remove.is_empty() {
-            // Persist the new GC cutoff value in the metadata file, before
-            // we actually remove anything.
-            self.update_metadata_file(self.disk_consistent_lsn.load(), HashMap::new())?;
+        self.layer_mgr
+            .update(|mut layers| async move {
+                let mut updates = layers.batch_update();
+                if !layers_to_remove.is_empty() {
+                    // Persist the new GC cutoff value in the metadata file, before
+                    // we actually remove anything.
+                    self.update_metadata_file(self.disk_consistent_lsn.load(), HashMap::new())?;
 
-            // Actually delete the layers from disk and remove them from the map.
-            // (couldn't do this in the loop above, because you cannot modify a collection
-            // while iterating it. BTreeMap::retain() would be another option)
-            let mut layer_names_to_delete = Vec::with_capacity(layers_to_remove.len());
-            {
-                for doomed_layer in layers_to_remove {
-                    layer_names_to_delete.push(doomed_layer.filename());
-                    self.delete_historic_layer(
-                        layer_removal_cs.clone(),
-                        doomed_layer,
-                        &mut updates,
-                    )?; // FIXME: schedule succeeded deletions before returning?
-                    result.layers_removed += 1;
+                    // Actually delete the layers from disk and remove them from the map.
+                    // (couldn't do this in the loop above, because you cannot modify a collection
+                    // while iterating it. BTreeMap::retain() would be another option)
+                    let mut layer_names_to_delete = Vec::with_capacity(layers_to_remove.len());
+                    {
+                        for doomed_layer in layers_to_remove {
+                            layer_names_to_delete.push(doomed_layer.filename());
+                            self.delete_historic_layer(
+                                layer_removal_cs.clone(),
+                                doomed_layer,
+                                &mut updates,
+                            )?; // FIXME: schedule succeeded deletions before returning?
+                            result.layers_removed += 1;
+                        }
+                    }
+
+                    if result.layers_removed != 0 {
+                        fail_point!("after-timeline-gc-removed-layers");
+                    }
+
+                    if let Some(remote_client) = &self.remote_client {
+                        remote_client.schedule_layer_file_deletion(&layer_names_to_delete)?;
+                    }
                 }
-            }
-
-            if result.layers_removed != 0 {
-                fail_point!("after-timeline-gc-removed-layers");
-            }
-
-            if let Some(remote_client) = &self.remote_client {
-                remote_client.schedule_layer_file_deletion(&layer_names_to_delete)?;
-            }
-        }
-        updates.flush();
+                updates.flush();
+                Ok(layers)
+            })
+            .await?;
 
         info!(
             "GC completed removing {} layers, cutoff {}",
@@ -4312,11 +4329,7 @@ impl Timeline {
 
                     // Download complete. Replace the RemoteLayer with the corresponding
                     // Delta- or ImageLayer in the layer map.
-                    let mut guard = self_clone.layers.write().await;
-                    let (layers, _) = &mut *guard;
-                    let updates = layers.batch_update();
-                    let new_layer =
-                        remote_layer.create_downloaded_layer(&updates, self_clone.conf, *size);
+                    let new_layer = remote_layer.create_downloaded_layer(self_clone.conf, *size);
                     {
                         let l: Arc<dyn PersistentLayer> = remote_layer.clone();
                         let failure = match self_clone.lcache.replace_and_verify(l, new_layer) {
@@ -4347,8 +4360,7 @@ impl Timeline {
                                 .store(true, Relaxed);
                         }
                     }
-                    updates.flush();
-                    drop_wlock(guard);
+                    // drop(guard);
 
                     info!("on-demand download successful");
 
@@ -4447,8 +4459,7 @@ impl Timeline {
     ) {
         let mut downloads = Vec::new();
         {
-            let guard = self.layers.read().await;
-            let (layers, _) = &*guard;
+            let layers = self.layer_mgr.read();
             layers
                 .iter_historic_layers()
                 .map(|l| self.lcache.get_from_desc(&l))
@@ -4552,8 +4563,7 @@ impl LocalLayerInfoForDiskUsageEviction {
 
 impl Timeline {
     pub(crate) async fn get_local_layers_for_disk_usage_eviction(&self) -> DiskUsageEvictionInfo {
-        let guard = self.layers.read().await;
-        let (layers, _) = &*guard;
+        let layers = self.layer_mgr.read();
 
         let mut max_layer_size: Option<u64> = None;
         let mut resident_layers = Vec::new();

pageserver/src/tenant/timeline/eviction_task.rs

@@ -197,8 +197,7 @@ impl Timeline {
         // We don't want to hold the layer map lock during eviction.
         // So, we just need to deal with this.
         let candidates: Vec<Arc<dyn PersistentLayer>> = {
-            let guard = self.layers.read().await;
-            let (layers, _) = &*guard;
+            let layers = self.layer_mgr.read();
             let mut candidates = Vec::new();
             for hist_layer in layers.iter_historic_layers() {
                 let hist_layer = self.lcache.get_from_desc(&hist_layer);