refactor(meta-service): normalize compactor (#18373)

* refactor(meta-service): normalize state machine compactor

chore(meta-service): no need to check against prefix when list by prefix

chore(meta-service): SMV003.expire_cursor is not used and is removed

refactor: compactor acquire process

chore: remove useless trait bound

* M  src/meta/raft-store/src/leveled_store/leveled_map/acquire_compactor_test.rs

Author: drmingdrmer

src/meta/raft-store/src/leveled_store/db_builder.rs

@@ -140,7 +140,7 @@ impl DBBuilder {
         lm.freeze_writable();
 
         let mut compacter = lm.acquire_compactor().await;
-        let (sys_data, strm) = compacter.compact().await?;
+        let (sys_data, strm) = compacter.compact_into_stream().await?;
 
         self.append_kv_stream(strm).await?;
 

src/meta/raft-store/src/leveled_store/immutable_levels.rs

@@ -24,8 +24,6 @@ use crate::leveled_store::immutable::Immutable;
 use crate::leveled_store::level::Level;
 use crate::leveled_store::map_api::KVResultStream;
 use crate::leveled_store::map_api::MapKey;
-use crate::leveled_store::map_api::MapKeyDecode;
-use crate::leveled_store::map_api::MapKeyEncode;
 
 /// A readonly leveled map that owns the data.
 #[derive(Debug, Default, Clone)]
@@ -71,8 +69,6 @@ impl ImmutableLevels {
 impl<K> MapApiRO<K> for ImmutableLevels
 where
     K: MapKey,
-    K: MapKeyEncode,
-    K: MapKeyDecode,
     Level: MapApiRO<K>,
     Immutable: MapApiRO<K>,
 {

src/meta/raft-store/src/leveled_store/leveled_map/acquire_compactor_test.rs

@@ -17,47 +17,9 @@ use tokio::time::timeout;
 
 use crate::leveled_store::leveled_map::LeveledMap;
 
-#[tokio::test]
-async fn test_try_acquire_ok() -> anyhow::Result<()> {
-    let mut lm = LeveledMap::default();
-
-    let c = lm.try_acquire_compactor();
-    assert!(c.is_some(), "Ok to try get compactor");
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_try_acquire_fail() -> anyhow::Result<()> {
-    let mut lm = LeveledMap::default();
-
-    let _c = lm.acquire_compactor().await;
-
-    assert!(
-        lm.try_acquire_compactor().is_none(),
-        "can not get two compactor"
-    );
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_try_acquire_ok_after_previous_dropped() -> anyhow::Result<()> {
-    let mut lm = LeveledMap::default();
-
-    let _c = lm.acquire_compactor().await;
-
-    assert!(lm.try_acquire_compactor().is_none());
-
-    drop(_c);
-
-    assert!(lm.try_acquire_compactor().is_some());
-    Ok(())
-}
-
 #[tokio::test]
 async fn test_blocking_wait_timeout() -> anyhow::Result<()> {
-    let mut lm = LeveledMap::default();
+    let lm = LeveledMap::default();
 
     let _c = lm.acquire_compactor().await;
 
@@ -79,7 +41,7 @@ async fn test_blocking_wait_timeout() -> anyhow::Result<()> {
 
 #[tokio::test]
 async fn test_blocking_wait_ok() -> anyhow::Result<()> {
-    let mut lm = LeveledMap::default();
+    let lm = LeveledMap::default();
 
     let _c = lm.acquire_compactor().await;
 

src/meta/raft-store/src/leveled_store/leveled_map/compacting_data.rs

@@ -93,7 +93,7 @@ impl<'a> CompactingData<'a> {
     /// The stream Item is 2 items tuple of key, and value with seq.
     ///
     /// The exported stream contains encoded `String` key and rotbl value [`SeqMarked`]
-    pub async fn compact(
+    pub async fn compact_into_stream(
         &self,
     ) -> Result<(SysData, IOResultStream<(String, SeqMarked)>), io::Error> {
         fn with_context(e: io::Error, key: &impl fmt::Debug) -> io::Error {

src/meta/raft-store/src/leveled_store/leveled_map/compactor.rs

@@ -22,14 +22,16 @@ use rotbl::v001::SeqMarked;
 use crate::leveled_store::immutable_levels::ImmutableLevels;
 use crate::leveled_store::level_index::LevelIndex;
 use crate::leveled_store::leveled_map::compacting_data::CompactingData;
+use crate::leveled_store::leveled_map::compactor_acquirer::CompactorPermit;
 
 /// Compactor is responsible for compacting the immutable levels and db.
 ///
 /// Only one Compactor can be running at a time.
 pub struct Compactor {
-    /// When dropped, drop the Sender so that the [`LeveledMap`] will be notified.
-    #[allow(dead_code)]
-    pub(super) guard: tokio::sync::oneshot::Sender<()>,
+    /// Acquired permit for this compactor.
+    ///
+    /// This is used to ensure that only one compactor can run at a time.
+    pub(crate) _permit: CompactorPermit,
 
     /// In memory immutable levels.
     pub(super) immutable_levels: ImmutableLevels,
@@ -68,10 +70,10 @@ impl Compactor {
     /// and a stream contains `kv` and `expire` entries.
     ///
     /// The exported stream contains encoded `String` key and rotbl value [`SeqMarked`]
-    pub async fn compact(
+    pub async fn compact_into_stream(
         &mut self,
     ) -> Result<(SysData, IOResultStream<(String, SeqMarked)>), io::Error> {
         let compacting_data = CompactingData::new(&mut self.immutable_levels, self.db.as_ref());
-        compacting_data.compact().await
+        compacting_data.compact_into_stream().await
     }
 }

src/meta/raft-store/src/leveled_store/leveled_map/compactor_acquirer.rs

@@ -0,0 +1,40 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use tokio::sync::OwnedSemaphorePermit;
+use tokio::sync::Semaphore;
+
+/// Acquirer is used to acquire a permit for compaction, without holding lock to the state machine.
+pub struct CompactorAcquirer {
+    sem: Arc<Semaphore>,
+}
+
+impl CompactorAcquirer {
+    pub fn new(sem: Arc<Semaphore>) -> Self {
+        CompactorAcquirer { sem }
+    }
+
+    pub async fn acquire(self) -> CompactorPermit {
+        // Safe unwrap: it returns error only when semaphore is closed.
+        // This semaphore does not close.
+        let permit = self.sem.acquire_owned().await.unwrap();
+        CompactorPermit { _permit: permit }
+    }
+}
+
+pub struct CompactorPermit {
+    _permit: OwnedSemaphorePermit,
+}

src/meta/raft-store/src/leveled_store/leveled_map/mod.rs

@@ -12,21 +12,26 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
 use compactor::Compactor;
 use databend_common_meta_types::snapshot_db::DB;
 use databend_common_meta_types::sys_data::SysData;
 use log::info;
-use tokio::sync::oneshot::error::TryRecvError;
+use tokio::sync::Semaphore;
 
 use crate::leveled_store::immutable::Immutable;
 use crate::leveled_store::immutable_levels::ImmutableLevels;
 use crate::leveled_store::level::Level;
 use crate::leveled_store::level_index::LevelIndex;
+use crate::leveled_store::leveled_map::compactor_acquirer::CompactorAcquirer;
+use crate::leveled_store::leveled_map::compactor_acquirer::CompactorPermit;
 
 #[cfg(test)]
 mod acquire_compactor_test;
 pub mod compacting_data;
 pub mod compactor;
+pub mod compactor_acquirer;
 #[cfg(test)]
 mod leveled_map_test;
 mod map_api_impl;
@@ -37,12 +42,10 @@ mod map_api_impl;
 ///
 /// The top level is the newest and writable.
 /// Others are immutable.
-#[derive(Debug, Default)]
+#[derive(Debug)]
 pub struct LeveledMap {
-    /// Concurrency control: only one thread can set this field to Some and compact.
-    ///
-    /// The other should wait for the compaction to finish by blocking on the receiver.
-    current_compactor: Option<tokio::sync::oneshot::Receiver<()>>,
+    /// A semaphore that permits at most one compactor to run.
+    compaction_semaphore: Arc<Semaphore>,
 
     /// The top level is the newest and writable.
     writable: Level,
@@ -53,6 +56,18 @@ pub struct LeveledMap {
     persisted: Option<DB>,
 }
 
+impl Default for LeveledMap {
+    fn default() -> Self {
+        Self {
+            // Only one compactor is allowed.
+            compaction_semaphore: Arc::new(Semaphore::new(1)),
+            writable: Default::default(),
+            immutable_levels: Default::default(),
+            persisted: None,
+        }
+    }
+}
+
 impl AsRef<SysData> for LeveledMap {
     fn as_ref(&self) -> &SysData {
         self.writable.sys_data_ref()
@@ -157,50 +172,25 @@ impl LeveledMap {
         info!("compaction finished replacing the db");
     }
 
-    /// Try to get a singleton `Compactor` instance specific to `self`
-    /// if it is not currently in use by another thread.
-    ///
-    /// This method requires a mutable reference to prevent concurrent access to shared data,
-    /// such as `self.immediate_levels` and `self.persisted`, during the construction of the compactor.
-    pub(crate) fn try_acquire_compactor(&mut self) -> Option<Compactor> {
-        if let Some(rx) = &mut self.current_compactor {
-            match rx.try_recv() {
-                Err(TryRecvError::Closed) => {
-                    // Ok, released. Continue
-                }
-                Err(TryRecvError::Empty) => {
-                    // Another compactor still in use.
-                    return None;
-                }
-                Ok(_) => {
-                    unreachable!("it never send any value")
-                }
-            }
-        }
-
-        Some(self.new_compactor())
-    }
-
     /// Get a singleton `Compactor` instance specific to `self`.
     ///
     /// This method requires a mutable reference to prevent concurrent access to shared data,
     /// such as `self.immediate_levels` and `self.persisted`, during the construction of the compactor.
-    pub(crate) async fn acquire_compactor(&mut self) -> Compactor {
-        if let Some(rx) = self.current_compactor.take() {
-            let _ = rx.await;
-        }
+    pub(crate) async fn acquire_compactor(&self) -> Compactor {
+        let acquirer = self.new_compactor_acquirer();
 
-        self.new_compactor()
-    }
+        let permit = acquirer.acquire().await;
 
-    fn new_compactor(&mut self) -> Compactor {
-        let (tx, rx) = tokio::sync::oneshot::channel();
+        self.new_compactor(permit)
+    }
 
-        // current_compactor must be None, which is guaranteed by caller.
-        self.current_compactor = Some(rx);
+    pub(crate) fn new_compactor_acquirer(&self) -> CompactorAcquirer {
+        CompactorAcquirer::new(self.compaction_semaphore.clone())
+    }
 
+    pub(crate) fn new_compactor(&self, permit: CompactorPermit) -> Compactor {
         Compactor {
-            guard: tx,
+            _permit: permit,
             immutable_levels: self.immutable_levels.clone(),
             db: self.persisted.clone(),
             since: self.immutable_level_index(),

src/meta/raft-store/src/sm_v003/compact_immutable_levels_test.rs

@@ -149,7 +149,7 @@ async fn test_compact_3_level() -> anyhow::Result<()> {
 
     let mut compactor = lm.acquire_compactor().await;
 
-    let (sys_data, strm) = compactor.compact().await?;
+    let (sys_data, strm) = compactor.compact_into_stream().await?;
     assert_eq!(
         r#"{"last_applied":{"leader_id":{"term":3,"node_id":3},"index":3},"last_membership":{"log_id":{"leader_id":{"term":3,"node_id":3},"index":3},"membership":{"configs":[],"nodes":{}}},"nodes":{"3":{"name":"3","endpoint":{"addr":"3","port":3},"grpc_api_advertise_address":null}},"sequence":7}"#,
         serde_json::to_string(&sys_data).unwrap()
@@ -176,7 +176,7 @@ async fn test_export_2_level_with_meta() -> anyhow::Result<()> {
 
     let mut compactor = sm.acquire_compactor().await;
 
-    let (sys_data, strm) = compactor.compact().await?;
+    let (sys_data, strm) = compactor.compact_into_stream().await?;
     let got = strm
         .map_ok(|x| serde_json::to_string(&x).unwrap())
         .try_collect::<Vec<_>>()

src/meta/raft-store/src/sm_v003/compact_with_db_test.rs

@@ -269,7 +269,7 @@ async fn test_compact_output_3_level() -> anyhow::Result<()> {
 
     let mut compactor = lm.acquire_compactor().await;
 
-    let (sys_data, strm) = compactor.compact().await?;
+    let (sys_data, strm) = compactor.compact_into_stream().await?;
 
     assert_eq!(sys_data.curr_seq(), 7);
     assert_eq!(

src/meta/raft-store/src/sm_v003/sm_v003.rs

@@ -27,10 +27,11 @@ use openraft::entry::RaftEntry;
 
 use crate::applier::Applier;
 use crate::leveled_store::leveled_map::compactor::Compactor;
+use crate::leveled_store::leveled_map::compactor_acquirer::CompactorAcquirer;
+use crate::leveled_store::leveled_map::compactor_acquirer::CompactorPermit;
 use crate::leveled_store::leveled_map::LeveledMap;
 use crate::leveled_store::sys_data_api::SysDataApiRO;
 use crate::sm_v003::sm_v003_kv_api::SMV003KVApi;
-use crate::state_machine::ExpireKey;
 use crate::state_machine_api::StateMachineApi;
 
 type OnChange = Box<dyn Fn((String, Option<SeqV>, Option<SeqV>)) + Send + Sync>;
@@ -39,9 +40,6 @@ type OnChange = Box<dyn Fn((String, Option<SeqV>, Option<SeqV>)) + Send + Sync>;
 pub struct SMV003 {
     levels: LeveledMap,
 
-    /// The expiration key since which for next clean.
-    expire_cursor: ExpireKey,
-
     /// Callback when a change is applied to state machine
     pub(crate) on_change_applied: Option<OnChange>,
 }
@@ -50,7 +48,6 @@ impl fmt::Debug for SMV003 {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         f.debug_struct("SMV003")
             .field("levels", &self.levels)
-            .field("expire_cursor", &self.expire_cursor)
             .field(
                 "on_change_applied",
                 &self.on_change_applied.as_ref().map(|_x| "is_set"),
@@ -198,12 +195,18 @@ impl SMV003 {
         self.levels.freeze_writable();
     }
 
-    pub fn try_acquire_compactor(&mut self) -> Option<Compactor> {
-        self.map_mut().try_acquire_compactor()
+    /// A shortcut
+    pub async fn acquire_compactor(&self) -> Compactor {
+        let permit = self.new_compactor_acquirer().acquire().await;
+        self.new_compactor(permit)
+    }
+
+    pub fn new_compactor_acquirer(&self) -> CompactorAcquirer {
+        self.levels.new_compactor_acquirer()
     }
 
-    pub async fn acquire_compactor(&mut self) -> Compactor {
-        self.levels.acquire_compactor().await
+    pub fn new_compactor(&self, permit: CompactorPermit) -> Compactor {
+        self.levels.new_compactor(permit)
     }
 
     /// Replace all the state machine data with the given one.
@@ -220,9 +223,5 @@ impl SMV003 {
         );
 
         self.levels = level;
-
-        // The installed data may not clean up all expired keys, if it is built with an older state machine.
-        // So we need to reset the cursor then the next time applying a log it will clean up all expired.
-        self.expire_cursor = ExpireKey::new(0, 0);
     }
 }