feat(meta-client-semaphore): generate semaphore sequence using timestamps to reduce conflicts (#17878)

### Context:

Previously, the semaphore sequence (`seq`) generation relied on a
generator key in the meta-service. This approach required a
compare-and-swap (CAS) loop, where:
1. A `seq` was generated from the generator key.
2. The semaphore permit key (derived from the `seq`) was inserted into
   the meta-service, before the generator key is updated again by other
   processes.

Under high concurrency, this method caused frequent conflicts during the
CAS loop, leading to repeated retries and increased latency.

### Changes:

- Replaced the existing sequence generation logic with a timestamp-based
  policy.
- The new approach directly uses the current timestamp as the `seq`
  value and attempts to insert it into the meta-service.
- If a duplicate key conflict occurs (e.g., due to concurrent requests),
  the process retries. However, the conflict rate is significantly
  reduced, as timestamp values are less likely to collide.

### Benefits:

- Drastically reduces the conflict rate during semaphore sequence
  generation, especially under high-concurrency scenarios.
- Most requests now require only a single RPC to insert the permit,
  minimizing latency.

Author: drmingdrmer

src/meta/binaries/metabench/main.rs

@@ -351,6 +351,10 @@ struct SemaphoreConfig {
     /// The capacity of resource in the semaphore.
     capacity: u64,
 
+    /// Whether to generate a sem seq with the current timestamp,
+    /// which reduce the conflict when enqueueing the permits.
+    time_based: bool,
+
     /// The ttl if lease is not extended.
     ttl_ms: Option<u64>,
 
@@ -363,6 +367,7 @@ impl Default for SemaphoreConfig {
         Self {
             semaphores: 1,
             capacity: 100,
+            time_based: false,
             ttl_ms: None,
             hold_ms: None,
         }
@@ -403,8 +408,10 @@ async fn benchmark_semaphore(
 
     let permit_str = format!("({sem_key}, id={id})");
 
-    let permit_res =
-        Semaphore::new_acquired(client.clone(), &sem_key, param.capacity, &id, param.ttl()).await;
+    let mut sem = Semaphore::new(client.clone(), &sem_key, param.capacity).await;
+    sem.set_time_based_seq(param.time_based);
+
+    let permit_res = sem.acquire(&id, param.ttl()).await;
 
     print_sem_res(i, format!("sem-acquired: {permit_str}",), &permit_res);
 

src/meta/client/src/grpc_client.rs

@@ -272,7 +272,7 @@ impl MetaGrpcClient {
         loop {
             let recv_res = req_rx.recv().await;
             let Some(mut worker_request) = recv_res else {
-                warn!("{} handle closed. worker quit", self);
+                info!("{} handle closed. worker quit", self);
                 return;
             };
 

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

@@ -16,6 +16,8 @@ use std::future::Future;
 use std::io;
 use std::sync::Arc;
 use std::time::Duration;
+use std::time::SystemTime;
+use std::time::UNIX_EPOCH;
 
 use codeq::Encode;
 use databend_common_base::runtime::spawn_named;
@@ -41,6 +43,19 @@ use crate::queue::PermitEvent;
 use crate::PermitEntry;
 use crate::PermitKey;
 
+#[derive(Clone, Debug)]
+pub(crate) enum SeqPolicy {
+    /// Use a sequence number generator to ensure globally unique sequence numbers.
+    ///
+    /// Retry if the seq generator is updated by other process.
+    GeneratorKey { generator_key: String },
+
+    /// Use the current timestamp as the sequence number.
+    ///
+    /// Retry if duplicated
+    TimeBased,
+}
+
 /// The acquirer is responsible for acquiring a semaphore permit.
 ///
 /// It is used to acquire a semaphore by creating a new [`PermitEntry`] in the meta-service.
@@ -61,12 +76,12 @@ pub(crate) struct Acquirer {
     /// For example, the [`Acquirer`] crashed abnormally.
     pub(crate) lease: Duration,
 
-    /// The key of the permit sequence number generator.
+    /// The way to generate sequence number for a [`PermitKey`].
     ///
     /// It is used to generate a globally unique sequence number for each [`Permit`].
     /// [`Permit`] will be sorted in meta-service by this sequence number
     /// and to determine the order to be acquired.
-    pub(crate) seq_generator_key: String,
+    pub(crate) seq_policy: SeqPolicy,
 
     /// The meta client to interact with the meta-service.
     pub(crate) meta_client: Arc<ClientHandle>,
@@ -88,9 +103,6 @@ pub(crate) struct Acquirer {
 impl Acquirer {
     /// Acquires a new semaphore permit and returns a [`Permit`] handle.
     pub async fn acquire(mut self) -> Result<Permit, AcquireError> {
-        let mut sleep_time = Duration::from_millis(10);
-        let max_sleep_time = Duration::from_secs(1);
-
         self.stat.start();
 
         let permit_entry = PermitEntry {
@@ -101,10 +113,90 @@ impl Acquirer {
             .encode_to_vec()
             .map_err(|e| conn_io_error(e, "encode semaphore entry").context(&self.ctx))?;
 
-        // The sem_key is the key of the semaphore entry.
-        let permit_key = loop {
+        // Step 1: Create a new semaphore entry with the key format `{prefix}/queue/{seq:020}`.
+
+        let permit_key = match self.seq_policy.clone() {
+            SeqPolicy::GeneratorKey { generator_key } => {
+                self.enqueue_permit_entry(&permit_entry, &generator_key)
+                    .await?
+            }
+            SeqPolicy::TimeBased => self.enqueue_time_based_permit_entry(&permit_entry).await?,
+        };
+
+        // Step 2: The sem entry is inserted, keep it alive by extending the lease.
+
+        let leaser_cancel_tx = self.spawn_extend_lease_task(permit_key.clone(), val_bytes);
+
+        // Step 3: Wait for the semaphore to be acquired or removed.
+
+        while let Some(sem_event) = self.permit_event_rx.recv().await {
+            info!(
+                "Acquirer({}): received semaphore event: {:?}",
+                self.ctx, sem_event
+            );
+
+            self.stat.on_receive_event(&sem_event);
+
+            match sem_event {
+                PermitEvent::Acquired((seq, _)) => {
+                    if seq == permit_key.seq {
+                        self.stat.on_acquire();
+
+                        info!(
+                            "{} acquired: {}->{}",
+                            self.ctx, permit_key, self.acquirer_id
+                        );
+                        break;
+                    }
+                }
+                PermitEvent::Removed((seq, _)) => {
+                    if seq == permit_key.seq {
+                        self.stat.on_remove();
+
+                        warn!(
+                            "semaphore removed before acquired: {}->{}",
+                            permit_key, self.acquirer_id
+                        );
+                        return Err(AcquireError::EarlyRemoved(EarlyRemoved::new(
+                            permit_key.clone(),
+                            permit_entry.clone(),
+                        )));
+                    }
+                }
+            }
+        }
+
+        let permit = Permit::new(
+            self.permit_event_rx,
+            permit_key,
+            permit_entry,
+            self.stat,
+            self.subscriber_cancel_tx,
+            leaser_cancel_tx,
+        );
+
+        Ok(permit)
+    }
+
+    /// Add a new semaphore entry to the `<prefix>/queue` in the meta-service.
+    ///
+    /// In this method, the sequence number is generated in meta-service, by updating a generator key.
+    /// The semaphore entry is inserted with this seq number as key.
+    async fn enqueue_permit_entry(
+        &mut self,
+        permit_entry: &PermitEntry,
+        seq_generator_key: &str,
+    ) -> Result<PermitKey, ConnectionClosed> {
+        let mut sleep_time = Duration::from_millis(50);
+        let max_sleep_time = Duration::from_secs(2);
+
+        let val_bytes = permit_entry
+            .encode_to_vec()
+            .map_err(|e| conn_io_error(e, "encode semaphore entry").context(&self.ctx))?;
+
+        loop {
             // Step 1: Get a new globally unique sequence number.
-            let sem_seq = self.next_global_unique_seq().await?;
+            let sem_seq = self.next_global_unique_seq(seq_generator_key).await?;
 
             self.stat.on_finish_get_seq();
 
@@ -121,7 +213,7 @@ impl Acquirer {
 
             let txn = pb::TxnRequest::default();
 
-            let cond = pb::TxnCondition::eq_seq(&self.seq_generator_key, sem_seq);
+            let cond = pb::TxnCondition::eq_seq(seq_generator_key, sem_seq);
             let op = pb::TxnOp::put_with_ttl(&sem_key_str, val_bytes.clone(), Some(self.lease));
             let txn = txn.push_if_then([cond], [op]);
 
@@ -144,7 +236,7 @@ impl Acquirer {
                     self.acquirer_id, sem_seq
                 );
                 self.stat.on_insert_seq(sem_key.seq);
-                break sem_key;
+                return Ok(sem_key);
             } else {
                 info!(
                     "acquire semaphore: enqueue conflict: acquirer: {}, sem_seq: {}; sleep {:?} and retry",
@@ -154,70 +246,76 @@ impl Acquirer {
                 tokio::time::sleep(sleep_time).await;
                 sleep_time = std::cmp::min(sleep_time * 3 / 2, max_sleep_time);
             }
-        };
+        }
+    }
 
-        // Step 3: The sem entry is inserted, keep it alive by extending the lease.
+    /// Add a new semaphore entry to the `<prefix>/queue` in the meta-service.
+    ///
+    /// In this method, the sequence number is generated with current timestamp,
+    /// which means it may not be globally unique, on which case it will be retried,
+    /// and it may not be in strict order, meaning more semaphore entries than the `capacity` may be acquired.
+    async fn enqueue_time_based_permit_entry(
+        &mut self,
+        permit_entry: &PermitEntry,
+    ) -> Result<PermitKey, ConnectionClosed> {
+        let val_bytes = permit_entry
+            .encode_to_vec()
+            .map_err(|e| conn_io_error(e, "encode semaphore entry").context(&self.ctx))?;
 
-        let leaser_cancel_tx = self.spawn_extend_lease_task(permit_key.clone(), val_bytes);
+        loop {
+            let sem_seq = SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_micros() as u64;
 
-        // Step 4: Wait for the semaphore to be acquired or removed.
+            self.stat.on_finish_get_seq();
 
-        while let Some(sem_event) = self.permit_event_rx.recv().await {
-            info!(
-                "Acquirer({}): received semaphore event: {:?}",
-                self.ctx, sem_event
-            );
+            // Create a new semaphore entry with the key format `{prefix}/queue/{seq:020}`.
+            let sem_key = PermitKey::new(self.prefix.clone(), sem_seq);
+            let sem_key_str = sem_key.format_key();
 
-            self.stat.on_receive_event(&sem_event);
+            let upsert = UpsertKV::insert(&sem_key_str, &val_bytes).with_ttl(self.lease);
 
-            match sem_event {
-                PermitEvent::Acquired((seq, _)) => {
-                    if seq == permit_key.seq {
-                        self.stat.on_acquire();
+            let res = self.meta_client.upsert_kv(upsert).await;
+            let resp = res.map_err(|e| {
+                conn_io_error(
+                    e,
+                    format!(
+                        "insert semaphore (seq={} entry={}) in transaction",
+                        &sem_key, &permit_entry
+                    ),
+                )
+                .context(&self.ctx)
+            })?;
 
-                        info!(
-                            "{} acquired: {}->{}",
-                            self.ctx, permit_key, self.acquirer_id
-                        );
-                        break;
-                    }
-                }
-                PermitEvent::Removed((seq, _)) => {
-                    if seq == permit_key.seq {
-                        self.stat.on_remove();
+            self.stat.on_finish_try_insert_seq();
 
-                        warn!(
-                            "semaphore removed before acquired: {}->{}",
-                            permit_key, self.acquirer_id
-                        );
-                        return Err(AcquireError::EarlyRemoved(EarlyRemoved::new(
-                            permit_key.clone(),
-                            permit_entry.clone(),
-                        )));
-                    }
-                }
+            if resp.result.is_some() {
+                info!(
+                    "acquire semaphore: enqueue done: acquirer_id: {}, sem_seq: {}",
+                    self.acquirer_id, sem_seq
+                );
+                self.stat.on_insert_seq(sem_key.seq);
+                return Ok(sem_key);
+            } else {
+                info!(
+                    "acquire semaphore: enqueue failed(duplicated key): acquirer: {}, sem_seq: {}; retry at once",
+                    self.acquirer_id, sem_seq
+                );
             }
         }
-
-        let permit = Permit::new(
-            self.permit_event_rx,
-            permit_key,
-            permit_entry,
-            self.stat,
-            self.subscriber_cancel_tx,
-            leaser_cancel_tx,
-        );
-
-        Ok(permit)
     }
 
     /// Gets a new globally unique sequence number by updating a key in the meta-service.
     ///
     /// This method uses the meta client to perform an upsert operation on the sequence
     /// generator key, which atomically increments the sequence number and returns the
     /// new value.
-    async fn next_global_unique_seq(&self) -> Result<u64, ConnectionClosed> {
-        let upsert = UpsertKV::update(&self.seq_generator_key, b"");
+    async fn next_global_unique_seq(
+        &self,
+        seq_generator_key: &str,
+    ) -> Result<u64, ConnectionClosed> {
+        let upsert = UpsertKV::update(seq_generator_key, b"");
 
         let res = self.meta_client.upsert_kv(upsert).await;
         let resp = res.map_err(|e| {

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

@@ -18,5 +18,6 @@ mod permit;
 mod stat;
 
 pub(crate) use acquirer::Acquirer;
+pub(crate) use acquirer::SeqPolicy;
 pub use permit::Permit;
 pub use stat::SharedAcquirerStat;

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

@@ -160,12 +160,14 @@ impl fmt::Display for SharedAcquirerStat {
         }
 
         if let Some(end) = window_end {
-            write!(
-                f,
-                ")=({:.1?}-{:.1?})",
-                window_start.unwrap().duration_since(base),
-                end.duration_since(base)
-            )?;
+            if let Some(start) = window_start {
+                write!(
+                    f,
+                    ")=({:.1?}-{:.1?})",
+                    start.duration_since(base),
+                    end.duration_since(base)
+                )?;
+            }
         }
 
         write!(f, "], ")?;