refactor: avoid using RWlock, use semaphore instead (#18601)

* feat(meta-service): implement MVCC state machine architecture with controlled data access

NON-BREAKING CHANGE: State machine data access patterns changed from direct
mutable references to controlled access with closures

- Upgrade map-api to 0.3.0 and state-machine-api to 0.2.1 for MVCC support
- Replace sys_data_mut() direct access with with_sys_data() closures
- Add ApplierData and StateMachineView for MVCC operations
- Implement ViewReadonly and ScopedView traits for user and expire namespaces
- Refactor all state modifications to use controlled access patterns
- Update applier to serialize state machine operations with permits
- Migrate from direct field access to functional data manipulation

* chore: fix state-machine-api version

* chore: fix tests

* chore: fix lint

* chore: fix lint

* chore: move out name space, key, value

* chore remove unused MapApi impl

* chore: fix test

* refactor: avoid using RWlock, use semaphore instead

Author: drmingdrmer

Cargo.lock

@@ -374,7 +374,7 @@ logforth = { git = "https://github.com/datafuse-extras/logforth", branch = "main
     'fastrace',
 ] }
 lz4 = "1.24.0"
-map-api = { version = "0.2.5" }
+map-api = { version = "0.3.1" }
 maplit = "1.0.2"
 match-template = "0.0.1"
 md-5 = "0.10.5"
@@ -470,7 +470,7 @@ rustyline = "14"
 scroll = "0.12.0"
 self_cell = "1.2.0"
 semver = "1.0.14"
-seq-marked = { version = "0.3.1", features = ["seq-marked-serde", "seq-marked-bincode", "seqv-serde"] }
+seq-marked = { version = "0.3.3", features = ["seq-marked-serde", "seq-marked-bincode", "seqv-serde"] }
 serde = { version = "1.0.164", features = ["derive", "rc"] }
 serde_derive = "1"
 serde_ignored = "0.1.10"
@@ -496,7 +496,7 @@ socket2 = "0.5.3"
 span-map = { version = "0.2.0" }
 sqlx = { version = "0.8", features = ["mysql", "runtime-tokio"] }
 state = "0.6.0"
-state-machine-api = { version = "0.1.1" }
+state-machine-api = { version = "0.2.1" }
 stream-more = "0.1.3"
 strength_reduce = "0.2.4"
 stringslice = "0.2.0"
@@ -656,12 +656,12 @@ backtrace = { git = "https://github.com/rust-lang/backtrace-rs.git", rev = "7226
 color-eyre = { git = "https://github.com/eyre-rs/eyre.git", rev = "e5d92c3" }
 deltalake = { git = "https://github.com/delta-io/delta-rs", rev = "9954bff" }
 display-more = { git = "https://github.com/databendlabs/display-more", tag = "v0.2.0" }
-map-api = { git = "https://github.com/databendlabs/map-api", tag = "v0.2.7" }
+map-api = { git = "https://github.com/databendlabs/map-api", tag = "v0.3.1" }
 openraft = { git = "https://github.com/databendlabs/openraft", tag = "v0.10.0-alpha.9" }
 orc-rust = { git = "https://github.com/datafuse-extras/orc-rust", rev = "d82aa6d" }
 recursive = { git = "https://github.com/datafuse-extras/recursive.git", rev = "6af35a1" }
 sled = { git = "https://github.com/datafuse-extras/sled", tag = "v0.34.7-datafuse.1" }
-state-machine-api = { git = "https://github.com/databendlabs/state-machine-api.git", tag = "v0.1.1" }
+state-machine-api = { git = "https://github.com/databendlabs/state-machine-api.git", tag = "v0.2.1" }
 sub-cache = { git = "https://github.com/databendlabs/sub-cache", tag = "v0.2.1" }
 tantivy = { git = "https://github.com/datafuse-extras/tantivy", rev = "7502370" }
 tantivy-common = { git = "https://github.com/datafuse-extras/tantivy", rev = "7502370", package = "tantivy-common" }

Cargo.toml

@@ -201,8 +201,8 @@ async fn init_new_cluster(
     let mut sto = RaftStore::open(&raft_config).await?;
 
     let last_applied = {
-        let sm2 = sto.get_state_machine_write("get-last-applied").await;
-        *sm2.sys_data_ref().last_applied_ref()
+        let sm2 = sto.state_machine();
+        *sm2.sys_data().last_applied_ref()
     };
 
     let last_log_id = std::cmp::max(last_applied, max_log_id);
@@ -213,17 +213,15 @@ async fn init_new_cluster(
 
     // Update snapshot: Replace nodes set and membership config.
     {
-        let mut sm2 = sto
-            .get_state_machine_write("init_new_cluster-get-last-membership")
-            .await;
-
-        *sm2.sys_data_mut().nodes_mut() = nodes.clone();
+        let sm2 = sto.state_machine();
 
         // It must set membership to state machine because
         // the snapshot may contain more logs than the last_log_id.
         // In which case, logs will be purged upon startup.
-        *sm2.sys_data_mut().last_membership_mut() =
-            StoredMembership::new(last_applied, membership.clone());
+        sm2.with_sys_data(|s| {
+            *s.nodes_mut() = nodes.clone();
+            *s.last_membership_mut() = StoredMembership::new(last_applied, membership.clone());
+        });
     }
 
     // Build snapshot to persist state machine.

src/meta/control/src/import.rs

@@ -88,12 +88,15 @@ impl TestSuite {
         self.kv_write_read(&builder.build().await).await?;
         self.kv_delete(&builder.build().await).await?;
         self.kv_update(&builder.build().await).await?;
+
         self.kv_timeout(&builder.build().await).await?;
         self.kv_expire_sec_or_ms(&builder.build().await).await?;
         self.kv_upsert_with_ttl(&builder.build().await).await?;
+
         self.kv_meta(&builder.build().await).await?;
         self.kv_list(&builder.build().await).await?;
         self.kv_mget(&builder.build().await).await?;
+
         self.kv_txn_absent_seq_0(&builder.build().await).await?;
         self.kv_transaction(&builder.build().await).await?;
         self.kv_transaction_fetch_add_u64(&builder.build().await)

src/meta/kvapi-test-suite/src/kvapi_test_suite.rs

@@ -0,0 +1,134 @@
+// 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::io;
+use std::ops::RangeBounds;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::time::Duration;
+
+use futures_util::StreamExt;
+use futures_util::TryStreamExt;
+use map_api::mvcc;
+use map_api::mvcc::ViewReadonly;
+use map_api::IOResultStream;
+use seq_marked::InternalSeq;
+use seq_marked::SeqMarked;
+use state_machine_api::ExpireKey;
+use state_machine_api::MetaValue;
+use state_machine_api::UserKey;
+
+use crate::leveled_store::leveled_map::applier_acquirer::WriterPermit;
+use crate::leveled_store::leveled_map::LeveledMapData;
+use crate::leveled_store::types::Key;
+use crate::leveled_store::types::Namespace;
+use crate::leveled_store::types::Value;
+use crate::sm_v003::OnChange;
+
+pub(crate) type StateMachineView = mvcc::View<Namespace, Key, Value, Arc<LeveledMapData>>;
+
+pub(crate) struct ApplierData {
+    /// Hold a unique permit to serialize all apply operations to the state machine.
+    pub(crate) _permit: WriterPermit,
+
+    pub(crate) view: StateMachineView,
+
+    /// Since when to start cleaning expired keys.
+    pub(crate) cleanup_start_time: Arc<Mutex<Duration>>,
+
+    pub(crate) on_change_applied: Arc<Option<OnChange>>,
+}
+
+#[async_trait::async_trait]
+impl mvcc::ScopedViewReadonly<UserKey, MetaValue> for ApplierData {
+    fn base_seq(&self) -> InternalSeq {
+        self.view.base_seq()
+    }
+
+    async fn get(&self, key: UserKey) -> Result<SeqMarked<MetaValue>, io::Error> {
+        let got = self.view.get(Namespace::User, Key::User(key)).await?;
+        Ok(got.map(|x| x.into_user()))
+    }
+
+    async fn range<R>(
+        &self,
+        range: R,
+    ) -> Result<IOResultStream<(UserKey, SeqMarked<MetaValue>)>, io::Error>
+    where
+        R: RangeBounds<UserKey> + Send + Sync + Clone + 'static,
+    {
+        let start = range.start_bound().cloned();
+        let end = range.end_bound().cloned();
+
+        let start = start.map(Key::User);
+        let end = end.map(Key::User);
+
+        let strm = self.view.range(Namespace::User, (start, end)).await?;
+
+        Ok(strm
+            .map_ok(|(k, v)| (k.into_user(), v.map(|x| x.into_user())))
+            .boxed())
+    }
+}
+
+#[async_trait::async_trait]
+impl mvcc::ScopedView<UserKey, MetaValue> for ApplierData {
+    fn set(&mut self, key: UserKey, value: Option<MetaValue>) -> SeqMarked<()> {
+        self.view
+            .set(Namespace::User, Key::User(key), value.map(Value::User))
+    }
+}
+
+#[async_trait::async_trait]
+impl mvcc::ScopedViewReadonly<ExpireKey, String> for ApplierData {
+    fn base_seq(&self) -> InternalSeq {
+        self.view.base_seq()
+    }
+
+    async fn get(&self, key: ExpireKey) -> Result<SeqMarked<String>, io::Error> {
+        let got = self.view.get(Namespace::Expire, Key::Expire(key)).await?;
+        Ok(got.map(|x| x.into_expire()))
+    }
+
+    async fn range<R>(
+        &self,
+        range: R,
+    ) -> Result<IOResultStream<(ExpireKey, SeqMarked<String>)>, io::Error>
+    where
+        R: RangeBounds<ExpireKey> + Send + Sync + Clone + 'static,
+    {
+        let start = range.start_bound().cloned();
+        let end = range.end_bound().cloned();
+
+        let start = start.map(Key::Expire);
+        let end = end.map(Key::Expire);
+
+        let strm = self.view.range(Namespace::Expire, (start, end)).await?;
+
+        Ok(strm
+            .map_ok(|(k, v)| (k.into_expire(), v.map(|x| x.into_expire())))
+            .boxed())
+    }
+}
+
+#[async_trait::async_trait]
+impl mvcc::ScopedView<ExpireKey, String> for ApplierData {
+    fn set(&mut self, key: ExpireKey, value: Option<String>) -> SeqMarked<()> {
+        self.view.set(
+            Namespace::Expire,
+            Key::Expire(key),
+            value.map(Value::Expire),
+        )
+    }
+}