refactor(meta): break hummock manager Versioning into smaller structs

src/common/metrics/src/monitor/rwlock.rs

@@ -16,31 +16,34 @@ use prometheus::HistogramVec;
 use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
 
 pub struct MonitoredRwLock<T> {
+    // labels: [lock_name, lock_type]
     metrics: HistogramVec,
     inner: RwLock<T>,
+    lock_name: &'static str,
 }
 
 impl<T> MonitoredRwLock<T> {
-    pub fn new(metrics: HistogramVec, val: T) -> Self {
+    pub fn new(metrics: HistogramVec, val: T, lock_name: &'static str) -> Self {
         Self {
             metrics,
             inner: RwLock::new(val),
+            lock_name,
         }
     }
 
-    pub async fn read<'a, 'b>(
-        &'a self,
-        label_values: &'b [&'static str],
-    ) -> RwLockReadGuard<'a, T> {
-        let _timer = self.metrics.with_label_values(label_values).start_timer();
+    pub async fn read(&self) -> RwLockReadGuard<'_, T> {
+        let _timer = self
+            .metrics
+            .with_label_values(&[self.lock_name, "read"])
+            .start_timer();
         self.inner.read().await
     }
 
-    pub async fn write<'a, 'b>(
-        &'a self,
-        label_values: &'b [&'static str],
-    ) -> RwLockWriteGuard<'a, T> {
-        let _timer = self.metrics.with_label_values(label_values).start_timer();
+    pub async fn write(&self) -> RwLockWriteGuard<'_, T> {
+        let _timer = self
+            .metrics
+            .with_label_values(&[self.lock_name, "write"])
+            .start_timer();
         self.inner.write().await
     }
 }

src/meta/src/hummock/manager/checkpoint.rs

@@ -17,7 +17,6 @@ use std::ops::Bound::{Excluded, Included};
 use std::ops::{Deref, DerefMut};
 use std::sync::atomic::Ordering;
 
-use function_name::named;
 use risingwave_hummock_sdk::compaction_group::hummock_version_ext::{
     object_size_map, summarize_group_deltas,
 };
@@ -32,7 +31,6 @@ use tracing::warn;
 
 use crate::hummock::error::Result;
 use crate::hummock::manager::versioning::Versioning;
-use crate::hummock::manager::{read_lock, write_lock};
 use crate::hummock::metrics_utils::{trigger_gc_stat, trigger_split_stat};
 use crate::hummock::HummockManager;
 
@@ -122,11 +120,10 @@ impl HummockManager {
     /// Returns the diff between new and old checkpoint id.
     /// Note that this method must not be called concurrently, because internally it doesn't hold
     /// lock throughout the method.
-    #[named]
     pub async fn create_version_checkpoint(&self, min_delta_log_num: u64) -> Result<u64> {
         let timer = self.metrics.version_checkpoint_latency.start_timer();
         // 1. hold read lock and create new checkpoint
-        let versioning_guard = read_lock!(self, versioning).await;
+        let versioning_guard = self.versioning.read().await;
         let versioning: &Versioning = versioning_guard.deref();
         let current_version: &HummockVersion = &versioning.current_version;
         let old_checkpoint: &HummockVersionCheckpoint = &versioning.checkpoint;
@@ -137,9 +134,10 @@ impl HummockManager {
         }
         if cfg!(test) && new_checkpoint_id == old_checkpoint_id {
             drop(versioning_guard);
-            let mut versioning = write_lock!(self, versioning).await;
-            versioning.mark_objects_for_deletion();
-            let min_pinned_version_id = versioning.min_pinned_version_id();
+            let versioning = self.versioning.read().await;
+            let context_info = self.context_info.read().await;
+            versioning.mark_objects_for_deletion(&context_info, &self.delete_object_tracker);
+            let min_pinned_version_id = context_info.min_pinned_version_id();
             trigger_gc_stat(&self.metrics, &versioning.checkpoint, min_pinned_version_id);
             return Ok(0);
         }
@@ -224,16 +222,17 @@ impl HummockManager {
             }
         }
         // 3. hold write lock and update in memory state
-        let mut versioning_guard = write_lock!(self, versioning).await;
+        let mut versioning_guard = self.versioning.write().await;
         let versioning = versioning_guard.deref_mut();
+        let context_info = self.context_info.read().await;
         assert!(new_checkpoint.version.id > versioning.checkpoint.version.id);
         versioning.checkpoint = new_checkpoint;
         // Not delete stale objects when archive is enabled
         if !self.env.opts.enable_hummock_data_archive {
-            versioning.mark_objects_for_deletion();
+            versioning.mark_objects_for_deletion(&context_info, &self.delete_object_tracker);
         }
 
-        let min_pinned_version_id = versioning.min_pinned_version_id();
+        let min_pinned_version_id = context_info.min_pinned_version_id();
         trigger_gc_stat(&self.metrics, &versioning.checkpoint, min_pinned_version_id);
         trigger_split_stat(&self.metrics, &versioning.current_version);
         drop(versioning_guard);
@@ -260,9 +259,8 @@ impl HummockManager {
         self.pause_version_checkpoint.load(Ordering::Relaxed)
     }
 
-    #[named]
     pub async fn get_checkpoint_version(&self) -> HummockVersion {
-        let versioning_guard = read_lock!(self, versioning).await;
+        let versioning_guard = self.versioning.read().await;
         versioning_guard.checkpoint.version.clone()
     }
 }

src/meta/src/hummock/manager/compaction.rs

@@ -15,7 +15,6 @@
 use std::collections::{BTreeMap, HashMap, HashSet};
 use std::sync::Arc;
 
-use function_name::named;
 use futures::future::Shared;
 use itertools::Itertools;
 use risingwave_hummock_sdk::{CompactionGroupId, HummockCompactionTaskId};
@@ -37,7 +36,7 @@ use tokio::sync::oneshot::Receiver as OneShotReceiver;
 use crate::hummock::compaction::selector::level_selector::PickerInfo;
 use crate::hummock::compaction::selector::DynamicLevelSelectorCore;
 use crate::hummock::compaction::{CompactStatus, CompactionDeveloperConfig, CompactionSelector};
-use crate::hummock::manager::{init_selectors, read_lock};
+use crate::hummock::manager::init_selectors;
 use crate::hummock::HummockManager;
 
 const MAX_SKIP_TIMES: usize = 8;
@@ -54,17 +53,12 @@ pub struct Compaction {
 }
 
 impl HummockManager {
-    #[named]
     pub async fn get_assigned_compact_task_num(&self) -> u64 {
-        read_lock!(self, compaction)
-            .await
-            .compact_task_assignment
-            .len() as u64
+        self.compaction.read().await.compact_task_assignment.len() as u64
     }
 
-    #[named]
     pub async fn list_all_tasks_ids(&self) -> Vec<HummockCompactionTaskId> {
-        let compaction = read_lock!(self, compaction).await;
+        let compaction = self.compaction.read().await;
 
         compaction
             .compaction_statuses
@@ -77,11 +71,10 @@ impl HummockManager {
             .collect_vec()
     }
 
-    #[named]
     pub async fn list_compaction_status(
         &self,
     ) -> (Vec<PbCompactStatus>, Vec<CompactTaskAssignment>) {
-        let compaction = read_lock!(self, compaction).await;
+        let compaction = self.compaction.read().await;
         (
             compaction.compaction_statuses.values().map_into().collect(),
             compaction
@@ -92,14 +85,13 @@ impl HummockManager {
         )
     }
 
-    #[named]
     pub async fn get_compaction_scores(
         &self,
         compaction_group_id: CompactionGroupId,
     ) -> Vec<PickerInfo> {
         let (status, levels, group) = {
-            let compaction = read_lock!(self, compaction).await;
-            let versioning = read_lock!(self, versioning).await;
+            let compaction = self.compaction.read().await;
+            let versioning = self.versioning.read().await;
             let config_manager = self.compaction_group_manager.read().await;
             match (
                 compaction.compaction_statuses.get(&compaction_group_id),