refactor: add more log to semaphore, add benchmark CLI (#17875)

- Upgrade sub crate `watcher` and `display-more`.

- Add semaphore benchmark:
```
  databend-metabench \
      --rpc 'semaphore:{"semaphores":10,"capacity":5,"ttl_ms":3000,"hold_ms":1000}' \
      --client 200 \
      --number 100 \
      --prefix "foo" \
      --grpc-api-address 127.0.0.1:$grpc_port
  ```

- Log the proposing time of a raft-log in human-readable format.

- Collect time profiling data of a semaphore and log it when a semaphore
  Permit is acquired. It includes the time spent on get/insert
  permit-seq CAS loop, and the time spent on waiting for `Acquired`
  event to be received.
  ```
  "AcquirerStat{ latencies: {
      seq(get/insert): 3/2: [100.1ms/300.1ms, 200.1ms/400.1ms, 250.1ms/-],
      events(my_seq:4): 3 [(A-3, **A-4**)=(0.0ns-7.9ms), (D-5)=(200.0ms-200.0ms)],
      total-acquired: 700.1ms } }"
  ```

- Log more detailed add/remove watcher events in meta-service.

Author: drmingdrmer

Cargo.lock

@@ -523,7 +523,7 @@ url = "2.5.4"
 uuid = { version = "1.10.0", features = ["std", "serde", "v4", "v7"] }
 volo-thrift = "0.10"
 walkdir = "2.3.2"
-watcher = { version = "0.1.1" }
+watcher = { version = "0.2.0" }
 wiremock = "0.6"
 wkt = "0.11.1"
 xorf = { version = "0.11.0", default-features = false, features = ["binary-fuse"] }
@@ -636,7 +636,7 @@ async-recursion = { git = "https://github.com/datafuse-extras/async-recursion.gi
 backtrace = { git = "https://github.com/rust-lang/backtrace-rs.git", rev = "72265be" }
 color-eyre = { git = "https://github.com/eyre-rs/eyre.git", rev = "e5d92c3" }
 deltalake = { git = "https://github.com/delta-io/delta-rs", rev = "c149502" }
-display-more = { git = "https://github.com/databendlabs/display-more", tag = "v0.1.2" }
+display-more = { git = "https://github.com/databendlabs/display-more", tag = "v0.1.3" }
 jsonb = { git = "https://github.com/databendlabs/jsonb", rev = "dcaf261" }
 map-api = { git = "https://github.com/databendlabs/map-api", tag = "v0.2.3" }
 openai_api_rust = { git = "https://github.com/datafuse-extras/openai-api", rev = "819a0ed" }
@@ -647,5 +647,5 @@ sled = { git = "https://github.com/datafuse-extras/sled", tag = "v0.34.7-datafus
 tantivy = { git = "https://github.com/datafuse-extras/tantivy", rev = "7502370" }
 tantivy-common = { git = "https://github.com/datafuse-extras/tantivy", rev = "7502370", package = "tantivy-common" }
 tantivy-jieba = { git = "https://github.com/datafuse-extras/tantivy-jieba", rev = "0e300e9" }
-watcher = { git = "https://github.com/databendlabs/watcher", tag = "v0.1.1" }
+watcher = { git = "https://github.com/databendlabs/watcher", tag = "v0.2.0" }
 xorfilter-rs = { git = "https://github.com/datafuse-extras/xorfilter", tag = "databend-alpha.4" }

Cargo.toml

@@ -133,7 +133,7 @@ where
             if let Ok(itm) = check_res {
                 return Ok(itm);
             } else {
-                warn!("RaftNetwork check reused item failed: {:?}", key);
+                warn!("Pool check reused item failed: {:?}", key);
                 // mark broken conn as deleted
                 *guard = None;
             }

src/common/base/src/containers/pool.rs

@@ -31,6 +31,7 @@ databend-common-meta-client = { workspace = true }
 databend-common-meta-control = { workspace = true }
 databend-common-meta-kvapi = { workspace = true }
 databend-common-meta-raft-store = { workspace = true }
+databend-common-meta-semaphore = { workspace = true }
 databend-common-meta-sled-store = { workspace = true }
 databend-common-meta-store = { workspace = true }
 databend-common-meta-types = { workspace = true }

src/meta/binaries/Cargo.toml

@@ -18,6 +18,7 @@ use std::collections::BTreeMap;
 use std::fmt::Debug;
 use std::fmt::Display;
 use std::sync::Arc;
+use std::time::Duration;
 use std::time::Instant;
 
 use chrono::Utc;
@@ -42,6 +43,7 @@ use databend_common_meta_client::required;
 use databend_common_meta_client::ClientHandle;
 use databend_common_meta_client::MetaGrpcClient;
 use databend_common_meta_kvapi::kvapi::KVApi;
+use databend_common_meta_semaphore::Semaphore;
 use databend_common_meta_types::MatchSeq;
 use databend_common_meta_types::Operation;
 use databend_common_meta_types::TxnRequest;
@@ -52,6 +54,7 @@ use databend_common_tracing::StderrConfig;
 use databend_meta::version::METASRV_COMMIT_VERSION;
 use serde::Deserialize;
 use serde::Serialize;
+use tokio::time::sleep;
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Parser)]
 #[clap(about, version = & * * METASRV_COMMIT_VERSION, author)]
@@ -152,6 +155,8 @@ async fn main() {
                     benchmark_get_table(&client, prefix, client_num, i).await;
                 } else if cmd == "table_copy_file" {
                     benchmark_table_copy_file(&client, prefix, client_num, i, &param).await;
+                } else if cmd == "semaphore" {
+                    benchmark_semaphore(&client, prefix, client_num, i, &param).await;
                 } else {
                     unreachable!("Invalid config.rpc: {}", rpc);
                 }
@@ -339,6 +344,91 @@ async fn benchmark_table_copy_file(
     res.unwrap();
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+struct SemaphoreConfig {
+    semaphores: u64,
+
+    /// The capacity of resource in the semaphore.
+    capacity: u64,
+
+    /// The ttl if lease is not extended.
+    ttl_ms: Option<u64>,
+
+    /// The time a permit is held by the application for simulation
+    hold_ms: Option<u64>,
+}
+
+impl Default for SemaphoreConfig {
+    fn default() -> Self {
+        Self {
+            semaphores: 1,
+            capacity: 100,
+            ttl_ms: None,
+            hold_ms: None,
+        }
+    }
+}
+
+impl SemaphoreConfig {
+    pub fn ttl(&self) -> Duration {
+        Duration::from_millis(self.ttl_ms.unwrap_or(3_000))
+    }
+
+    pub fn hold(&self) -> Duration {
+        Duration::from_millis(self.hold_ms.unwrap_or(100))
+    }
+}
+
+/// Benchmark semaphore acquire.
+///
+/// - `key_prefix` is used to distribut the load to separate key spaces.
+/// - `client_num` is number of concurrent clients.
+/// - `i` is the index of the current client.
+/// - `param` is a json string of bench specific config.
+async fn benchmark_semaphore(
+    client: &Arc<ClientHandle>,
+    key_prefix: u64,
+    client_num: u64,
+    i: u64,
+    param: &str,
+) {
+    let param = if param.is_empty() {
+        SemaphoreConfig::default()
+    } else {
+        serde_json::from_str(param).unwrap()
+    };
+
+    let sem_key = format!("sem-{}-{}", key_prefix, client_num % param.semaphores);
+    let id = format!("cli-{client_num}-{i}th");
+
+    let permit_str = format!("({sem_key}, id={id})");
+
+    let permit_res =
+        Semaphore::new_acquired(client.clone(), &sem_key, param.capacity, &id, param.ttl()).await;
+
+    print_sem_res(i, format!("sem-acquired: {permit_str}",), &permit_res);
+
+    let permit = match permit_res {
+        Ok(permit) => permit,
+        Err(e) => {
+            println!("ERROR: Failed to acquire semaphore: {permit_str}: {}", e);
+            return;
+        }
+    };
+
+    sleep(param.hold()).await;
+
+    print_sem_res(
+        i,
+        format!("sem-released: {permit_str}, {}", permit.stat()),
+        &permit,
+    );
+
+    fn print_sem_res<D: Debug>(i: u64, typ: impl Display, res: &D) {
+        println!("{:>10}-th {} result: {:?}", i, typ, res);
+    }
+}
+
 fn print_res<D: Debug>(i: u64, typ: impl Display, res: &D) {
     if i % 100 == 0 {
         println!("{:>10}-th {} result: {:?}", i, typ, res);

src/meta/binaries/metabench/main.rs

@@ -96,7 +96,11 @@ where SM: StateMachineApi + 'static
         let log_id = &entry.log_id;
         let log_time_ms = Self::get_log_time(entry);
 
-        debug!("apply: entry: {}, log_time_ms: {}", entry, log_time_ms);
+        debug!(
+            "apply: entry: {}, log_time_ms: {}",
+            entry,
+            Duration::from_millis(log_time_ms).display_unix_timestamp_short()
+        );
 
         self.cmd_ctx = CmdContext::from_millis(log_time_ms);
 
@@ -544,7 +548,10 @@ where SM: StateMachineApi + 'static
             return Ok(());
         }
 
-        debug!("to clean expired kvs, log_time_ts: {}", log_time_ms);
+        debug!(
+            "to clean expired kvs, log_time_ts: {}",
+            Duration::from_millis(log_time_ms).display_unix_timestamp_short()
+        );
 
         let mut to_clean = vec![];
         let mut strm = self.sm.list_expire_index(log_time_ms).await?;
@@ -608,7 +615,7 @@ where SM: StateMachineApi + 'static
                 Some(ms) => {
                     debug!(
                         "apply: raft-log time: {}",
-                        Duration::from_millis(ms).display_unix_timestamp()
+                        Duration::from_millis(ms).display_unix_timestamp_short()
                     );
                     ms
                 }

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

@@ -15,6 +15,7 @@
 use std::future;
 use std::io;
 use std::ops::RangeBounds;
+use std::time::Duration;
 
 use databend_common_meta_types::CmdContext;
 use databend_common_meta_types::Expirable;
@@ -23,6 +24,7 @@ use databend_common_meta_types::Operation;
 use databend_common_meta_types::SeqV;
 use databend_common_meta_types::SeqValue;
 use databend_common_meta_types::UpsertKV;
+use display_more::DisplayUnixTimeStampExt;
 use futures_util::StreamExt;
 use futures_util::TryStreamExt;
 use log::debug;
@@ -71,6 +73,13 @@ pub trait StateMachineApiExt: StateMachineApi {
 
         let expire_ms = kv_meta.expires_at_ms();
         if expire_ms < self.get_expire_cursor().time_ms {
+            warn!(
+                "upsert_kv_primary_index: expired key inserted: {} < expire-cursor: {}; key: {}",
+                Duration::from_millis(expire_ms).display_unix_timestamp_short(),
+                Duration::from_millis(self.get_expire_cursor().time_ms)
+                    .display_unix_timestamp_short(),
+                upsert_kv.key
+            );
             // The record has expired, delete it at once.
             //
             // Note that it must update first then delete,

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

@@ -21,6 +21,7 @@ databend-common-meta-kvapi = { workspace = true }
 databend-common-meta-types = { workspace = true }
 display-more = { workspace = true }
 futures = { workspace = true }
+itertools = { workspace = true }
 log = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }

src/meta/semaphore/Cargo.toml

@@ -27,13 +27,13 @@ use databend_common_meta_types::SeqValue;
 use databend_common_meta_types::UpsertKV;
 use databend_common_meta_types::With;
 use futures::FutureExt;
-use log::debug;
 use log::info;
 use log::warn;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 
 use crate::acquirer::Permit;
+use crate::acquirer::SharedAcquirerStat;
 use crate::errors::AcquireError;
 use crate::errors::ConnectionClosed;
 use crate::errors::EarlyRemoved;
@@ -78,6 +78,9 @@ pub(crate) struct Acquirer {
     /// This task subscribes to the watch stream and forwards relevant state changes through this channel.
     pub(crate) permit_event_rx: mpsc::Receiver<PermitEvent>,
 
+    /// The stat about the process of acquiring the semaphore.
+    pub(crate) stat: SharedAcquirerStat,
+
     /// The context information of this acquirer instance, used for logging.
     pub(crate) ctx: String,
 }
@@ -88,6 +91,8 @@ impl Acquirer {
         let mut sleep_time = Duration::from_millis(10);
         let max_sleep_time = Duration::from_secs(1);
 
+        self.stat.start();
+
         let permit_entry = PermitEntry {
             id: self.acquirer_id.clone(),
             permits: 1,
@@ -101,6 +106,8 @@ impl Acquirer {
             // Step 1: Get a new globally unique sequence number.
             let sem_seq = self.next_global_unique_seq().await?;
 
+            self.stat.on_finish_get_seq();
+
             // Step 2: Create a new semaphore entry with the key format `{prefix}/queue/{seq:020}`.
             //         We use a transaction to ensure the entry is only inserted if the sequence number
             //         hasn't changed since we obtained it. This guarantees that semaphore entries are
@@ -129,12 +136,18 @@ impl Acquirer {
                 .context(&self.ctx)
             })?;
 
+            self.stat.on_finish_try_insert_seq();
+
             if txn_reply.success {
-                info!("acquire semaphore: {} -> {}", self.acquirer_id, sem_seq);
+                info!(
+                    "acquire semaphore: enqueue done: acquirer_id: {}, sem_seq: {}",
+                    self.acquirer_id, sem_seq
+                );
+                self.stat.on_insert_seq(sem_key.seq);
                 break sem_key;
             } else {
                 info!(
-                    "acquire semaphore failed: {} -> {}; sleep {:?} and retry",
+                    "acquire semaphore: enqueue conflict: acquirer: {}, sem_seq: {}; sleep {:?} and retry",
                     self.acquirer_id, sem_seq, sleep_time
                 );
 
@@ -155,10 +168,14 @@ impl Acquirer {
                 self.ctx, sem_event
             );
 
+            self.stat.on_receive_event(&sem_event);
+
             match sem_event {
                 PermitEvent::Acquired((seq, _)) => {
                     if seq == permit_key.seq {
-                        debug!(
+                        self.stat.on_acquire();
+
+                        info!(
                             "{} acquired: {}->{}",
                             self.ctx, permit_key, self.acquirer_id
                         );
@@ -167,6 +184,8 @@ impl Acquirer {
                 }
                 PermitEvent::Removed((seq, _)) => {
                     if seq == permit_key.seq {
+                        self.stat.on_remove();
+
                         warn!(
                             "semaphore removed before acquired: {}->{}",
                             permit_key, self.acquirer_id
@@ -180,15 +199,16 @@ impl Acquirer {
             }
         }
 
-        let guard = Permit::new(
+        let permit = Permit::new(
             self.permit_event_rx,
             permit_key,
             permit_entry,
+            self.stat,
             self.subscriber_cancel_tx,
             leaser_cancel_tx,
         );
 
-        Ok(guard)
+        Ok(permit)
     }
 
     /// Gets a new globally unique sequence number by updating a key in the meta-service.

src/meta/semaphore/src/acquirer/acquirer.rs

@@ -15,6 +15,8 @@
 #[allow(clippy::module_inception)]
 mod acquirer;
 mod permit;
+mod stat;
 
 pub(crate) use acquirer::Acquirer;
 pub use permit::Permit;
+pub use stat::SharedAcquirerStat;