perf(ecstore): batch delete object lock acquisition (#2374)

Co-authored-by: 安正超 <anzhengchao@gmail.com>

crates/ecstore/src/rpc/remote_locker.rs

@@ -19,7 +19,7 @@ use rustfs_lock::{
     types::{LockId, LockMetadata, LockPriority},
 };
 use rustfs_protos::proto_gen::node_service::node_service_client::NodeServiceClient;
-use rustfs_protos::proto_gen::node_service::{GenerallyLockRequest, PingRequest};
+use rustfs_protos::proto_gen::node_service::{BatchGenerallyLockRequest, GenerallyLockRequest, PingRequest};
 use tonic::Request;
 use tonic::service::interceptor::InterceptedService;
 use tonic::transport::Channel;
@@ -61,6 +61,44 @@ impl RemoteClient {
             .await
             .map_err(|err| LockError::internal(format!("can not get client, err: {err}")))
     }
+
+    fn build_lock_info(request: &LockRequest, lock_info_json: Option<String>) -> LockInfo {
+        if let Some(lock_info_json) = lock_info_json {
+            match serde_json::from_str::<LockInfo>(&lock_info_json) {
+                Ok(info) => info,
+                Err(e) => {
+                    warn!("Failed to deserialize lock_info from response: {}, using request data", e);
+                    LockInfo {
+                        id: request.lock_id.clone(),
+                        resource: request.resource.clone(),
+                        lock_type: request.lock_type,
+                        status: LockStatus::Acquired,
+                        owner: request.owner.clone(),
+                        acquired_at: std::time::SystemTime::now(),
+                        expires_at: std::time::SystemTime::now() + request.ttl,
+                        last_refreshed: std::time::SystemTime::now(),
+                        metadata: request.metadata.clone(),
+                        priority: request.priority,
+                        wait_start_time: None,
+                    }
+                }
+            }
+        } else {
+            LockInfo {
+                id: request.lock_id.clone(),
+                resource: request.resource.clone(),
+                lock_type: request.lock_type,
+                status: LockStatus::Acquired,
+                owner: request.owner.clone(),
+                acquired_at: std::time::SystemTime::now(),
+                expires_at: std::time::SystemTime::now() + request.ttl,
+                last_refreshed: std::time::SystemTime::now(),
+                metadata: request.metadata.clone(),
+                priority: request.priority,
+                wait_start_time: None,
+            }
+        }
+    }
 }
 
 #[async_trait]
@@ -86,46 +124,10 @@ impl LockClient for RemoteClient {
 
         // Check if the lock acquisition was successful
         if resp.success {
-            // Try to deserialize lock_info from response
-            let lock_info = if let Some(lock_info_json) = resp.lock_info {
-                match serde_json::from_str::<LockInfo>(&lock_info_json) {
-                    Ok(info) => info,
-                    Err(e) => {
-                        // If deserialization fails, fall back to constructing from request
-                        warn!("Failed to deserialize lock_info from response: {}, using request data", e);
-                        LockInfo {
-                            id: request.lock_id.clone(),
-                            resource: request.resource.clone(),
-                            lock_type: request.lock_type,
-                            status: LockStatus::Acquired,
-                            owner: request.owner.clone(),
-                            acquired_at: std::time::SystemTime::now(),
-                            expires_at: std::time::SystemTime::now() + request.ttl,
-                            last_refreshed: std::time::SystemTime::now(),
-                            metadata: request.metadata.clone(),
-                            priority: request.priority,
-                            wait_start_time: None,
-                        }
-                    }
-                }
-            } else {
-                // If lock_info is not provided, construct from request
-                LockInfo {
-                    id: request.lock_id.clone(),
-                    resource: request.resource.clone(),
-                    lock_type: request.lock_type,
-                    status: LockStatus::Acquired,
-                    owner: request.owner.clone(),
-                    acquired_at: std::time::SystemTime::now(),
-                    expires_at: std::time::SystemTime::now() + request.ttl,
-                    last_refreshed: std::time::SystemTime::now(),
-                    metadata: request.metadata.clone(),
-                    priority: request.priority,
-                    wait_start_time: None,
-                }
-            };
-
-            Ok(LockResponse::success(lock_info, std::time::Duration::ZERO))
+            Ok(LockResponse::success(
+                Self::build_lock_info(request, resp.lock_info),
+                std::time::Duration::ZERO,
+            ))
         } else {
             // Lock acquisition failed
             Ok(LockResponse::failure(
@@ -135,6 +137,45 @@ impl LockClient for RemoteClient {
         }
     }
 
+    async fn acquire_locks_batch(&self, requests: &[LockRequest]) -> Result<Vec<LockResponse>> {
+        let mut client = self.get_client().await?;
+        let req = Request::new(BatchGenerallyLockRequest {
+            args: requests
+                .iter()
+                .map(|request| {
+                    serde_json::to_string(request).map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))
+                })
+                .collect::<Result<Vec<_>>>()?,
+        });
+
+        let resp = client
+            .lock_batch(req)
+            .await
+            .map_err(|e| LockError::internal(e.to_string()))?
+            .into_inner();
+
+        Ok(requests
+            .iter()
+            .enumerate()
+            .map(|(idx, request)| match resp.results.get(idx) {
+                Some(result) if result.success => {
+                    LockResponse::success(Self::build_lock_info(request, result.lock_info.clone()), std::time::Duration::ZERO)
+                }
+                Some(result) => LockResponse::failure(
+                    result
+                        .error_info
+                        .clone()
+                        .unwrap_or_else(|| "Lock acquisition failed on remote server".to_string()),
+                    std::time::Duration::ZERO,
+                ),
+                None => LockResponse::failure(
+                    format!("Lock batch response missing entry for request index {idx}"),
+                    std::time::Duration::ZERO,
+                ),
+            })
+            .collect())
+    }
+
     async fn release(&self, lock_id: &LockId) -> Result<bool> {
         info!("remote release for {}", lock_id);
 
@@ -154,6 +195,31 @@ impl LockClient for RemoteClient {
         Ok(resp.success)
     }
 
+    async fn release_locks_batch(&self, lock_ids: &[LockId]) -> Result<Vec<bool>> {
+        let mut client = self.get_client().await?;
+        let req = Request::new(BatchGenerallyLockRequest {
+            args: lock_ids
+                .iter()
+                .map(|lock_id| {
+                    serde_json::to_string(&Self::create_unlock_request(lock_id))
+                        .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))
+                })
+                .collect::<Result<Vec<_>>>()?,
+        });
+
+        let resp = client
+            .un_lock_batch(req)
+            .await
+            .map_err(|e| LockError::internal(e.to_string()))?
+            .into_inner();
+
+        Ok(lock_ids
+            .iter()
+            .enumerate()
+            .map(|(idx, _)| resp.results.get(idx).map(|result| result.success).unwrap_or(false))
+            .collect())
+    }
+
     async fn refresh(&self, lock_id: &LockId) -> Result<bool> {
         info!("remote refresh for {}", lock_id);
         let refresh_request = Self::create_unlock_request(lock_id);

crates/ecstore/src/set_disk.rs

@@ -76,7 +76,7 @@ use rustfs_filemeta::{
 use rustfs_lock::LockClient;
 use rustfs_lock::fast_lock::types::LockResult;
 use rustfs_lock::local_lock::LocalLock;
-use rustfs_lock::{FastLockGuard, NamespaceLock, NamespaceLockGuard, NamespaceLockWrapper, ObjectKey};
+use rustfs_lock::{FastLockGuard, LockManager, NamespaceLock, NamespaceLockGuard, NamespaceLockWrapper, ObjectKey};
 use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem};
 use rustfs_rio::{EtagResolvable, HashReader, HashReaderMut, TryGetIndex as _};
 use rustfs_s3_common::EventName;
@@ -964,6 +964,132 @@ impl ObjectIO for SetDisks {
     }
 }
 
+impl SetDisks {
+    async fn acquire_dist_delete_object_locks_batch(
+        &self,
+        batch: &rustfs_lock::BatchLockRequest,
+    ) -> (HashMap<(String, String), String>, HashSet<String>, Vec<Vec<rustfs_lock::LockId>>) {
+        let requests: Vec<rustfs_lock::LockRequest> = batch
+            .requests
+            .iter()
+            .map(|req| {
+                rustfs_lock::LockRequest::new(req.key.clone(), rustfs_lock::LockType::Exclusive, self.locker_owner.clone())
+                    .with_acquire_timeout(get_lock_acquire_timeout())
+                    .with_ttl(rustfs_lock::fast_lock::DEFAULT_LOCK_TIMEOUT)
+            })
+            .collect();
+
+        let write_quorum = if self.lockers.len() > 1 {
+            (self.lockers.len() / 2) + 1
+        } else {
+            1
+        };
+
+        let client_results = join_all(self.lockers.iter().cloned().enumerate().map(|(client_idx, client)| {
+            let requests = requests.clone();
+            async move { (client_idx, client.acquire_locks_batch(&requests).await) }
+        }))
+        .await;
+
+        let mut lock_ids_by_object: Vec<Vec<(usize, rustfs_lock::LockId)>> = vec![Vec::new(); requests.len()];
+        let mut errors_by_object: Vec<Option<String>> = vec![None; requests.len()];
+
+        for (client_idx, result) in client_results {
+            match result {
+                Ok(responses) => {
+                    for (req_idx, request) in requests.iter().enumerate() {
+                        match responses.get(req_idx) {
+                            Some(response) if response.success => {
+                                if let Some(lock_info) = response.lock_info.as_ref() {
+                                    lock_ids_by_object[req_idx].push((client_idx, lock_info.id.clone()));
+                                } else if errors_by_object[req_idx].is_none() {
+                                    errors_by_object[req_idx] = Some(format!(
+                                        "missing distributed lock id for {}/{}",
+                                        request.resource.bucket, request.resource.object
+                                    ));
+                                }
+                            }
+                            Some(response) => {
+                                if errors_by_object[req_idx].is_none() {
+                                    errors_by_object[req_idx] = Some(
+                                        response
+                                            .error
+                                            .clone()
+                                            .unwrap_or_else(|| "distributed lock acquisition failed".to_string()),
+                                    );
+                                }
+                            }
+                            None => {
+                                if errors_by_object[req_idx].is_none() {
+                                    errors_by_object[req_idx] =
+                                        Some(format!("client {client_idx} returned incomplete batch lock response"));
+                                }
+                            }
+                        }
+                    }
+                }
+                Err(err) => {
+                    for error in errors_by_object.iter_mut().take(requests.len()) {
+                        if error.is_none() {
+                            *error = Some(format!("client {client_idx} batch lock request failed: {err}"));
+                        }
+                    }
+                }
+            }
+        }
+
+        let mut failed_map = HashMap::new();
+        let mut locked_objects = HashSet::new();
+        let mut held_lock_ids_by_client = vec![Vec::new(); self.lockers.len()];
+        let mut rollback_lock_ids_by_client = vec![Vec::new(); self.lockers.len()];
+
+        for (req_idx, req) in batch.requests.iter().enumerate() {
+            let success_count = lock_ids_by_object[req_idx].len();
+            if success_count >= write_quorum {
+                for (client_idx, lock_id) in lock_ids_by_object[req_idx].drain(..) {
+                    held_lock_ids_by_client[client_idx].push(lock_id);
+                }
+                locked_objects.insert(req.key.object.as_ref().to_string());
+            } else {
+                for (client_idx, lock_id) in lock_ids_by_object[req_idx].drain(..) {
+                    rollback_lock_ids_by_client[client_idx].push(lock_id);
+                }
+                failed_map.insert(
+                    (req.key.bucket.as_ref().to_string(), req.key.object.as_ref().to_string()),
+                    errors_by_object[req_idx].clone().unwrap_or_else(|| {
+                        format!("failed to acquire distributed delete lock quorum: {success_count}/{write_quorum}")
+                    }),
+                );
+            }
+        }
+
+        self.release_dist_delete_object_locks_batch(rollback_lock_ids_by_client).await;
+
+        (failed_map, locked_objects, held_lock_ids_by_client)
+    }
+
+    async fn release_dist_delete_object_locks_batch(&self, lock_ids_by_client: Vec<Vec<rustfs_lock::LockId>>) {
+        join_all(self.lockers.iter().cloned().enumerate().filter_map(|(client_idx, client)| {
+            let lock_ids = lock_ids_by_client.get(client_idx).cloned().unwrap_or_default();
+            if lock_ids.is_empty() {
+                None
+            } else {
+                Some(async move {
+                    if let Err(err) = client.release_locks_batch(&lock_ids).await {
+                        tracing::warn!(
+                            client_idx,
+                            lock_count = lock_ids.len(),
+                            "failed to release distributed delete locks in batch: {}",
+                            err
+                        );
+                    }
+                })
+            }
+        }))
+        .await;
+    }
+}
+
 #[async_trait::async_trait]
 impl StorageAPI for SetDisks {
     #[tracing::instrument(skip(self))]
@@ -1242,27 +1368,25 @@ impl ObjectOperations for SetDisks {
         }
 
         let mut failed_map = HashMap::new();
-        let mut batch_guards = Vec::with_capacity(batch.requests.len());
-
+        let mut _local_batch_guards: Vec<FastLockGuard> = Vec::with_capacity(batch.requests.len());
         let mut locked_objects = HashSet::new();
 
-        for req in batch.requests.iter() {
-            let ns_lock = match self.new_ns_lock(req.key.bucket.as_ref(), req.key.object.as_ref()).await {
-                Ok(ns_lock) => ns_lock,
-                Err(e) => {
-                    failed_map.insert((req.key.bucket.as_ref().to_string(), req.key.object.as_ref().to_string()), e.to_string());
-                    continue;
-                }
-            };
-            let _lock_guard = match ns_lock.get_write_lock(get_lock_acquire_timeout()).await {
-                Ok(lock_guard) => lock_guard,
-                Err(e) => {
-                    failed_map.insert((req.key.bucket.as_ref().to_string(), req.key.object.as_ref().to_string()), e.to_string());
-                    continue;
-                }
-            };
-            batch_guards.push(_lock_guard);
-            locked_objects.insert(req.key.object.as_ref().to_string());
+        let dist_erasure = is_dist_erasure().await;
+        let mut dist_batch_lock_ids = vec![Vec::new(); self.lockers.len()];
+
+        if dist_erasure {
+            (failed_map, locked_objects, dist_batch_lock_ids) = self.acquire_dist_delete_object_locks_batch(&batch).await;
+        } else {
+            let batch_result = self.local_lock_manager.acquire_locks_batch(batch).await;
+            _local_batch_guards = batch_result.guards;
+
+            for key in batch_result.successful_locks {
+                locked_objects.insert(key.object.as_ref().to_string());
+            }
+
+            for (key, err) in batch_result.failed_locks {
+                failed_map.insert((key.bucket.as_ref().to_string(), key.object.as_ref().to_string()), format!("{err:?}"));
+            }
         }
 
         // Mark failures for objects that could not be locked
@@ -1428,6 +1552,10 @@ impl ObjectOperations for SetDisks {
 
         // TODO: add_partial
 
+        if dist_erasure {
+            self.release_dist_delete_object_locks_batch(dist_batch_lock_ids).await;
+        }
+
         (del_objects, del_errs)
     }
 
@@ -4267,6 +4395,81 @@ mod tests {
         );
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn test_acquire_dist_delete_object_locks_batch_succeeds_with_two_healthy_lockers() {
+        let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+
+        let manager1 = Arc::new(rustfs_lock::GlobalLockManager::new());
+        let manager2 = Arc::new(rustfs_lock::GlobalLockManager::new());
+        let client1: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(manager1.clone()));
+        let client2: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(manager2.clone()));
+        let set_disks = make_test_set_disks(vec![client1, client2]).await;
+
+        let batch = rustfs_lock::BatchLockRequest::new(set_disks.locker_owner.as_str())
+            .with_all_or_nothing(false)
+            .add_write_lock(ObjectKey::new("bucket", "object-a"))
+            .add_write_lock(ObjectKey::new("bucket", "object-b"));
+
+        let (failed_map, locked_objects, held_lock_ids_by_client) =
+            set_disks.acquire_dist_delete_object_locks_batch(&batch).await;
+
+        assert!(failed_map.is_empty());
+        assert_eq!(locked_objects.len(), 2);
+        assert!(locked_objects.contains("object-a"));
+        assert!(locked_objects.contains("object-b"));
+        assert_eq!(held_lock_ids_by_client.iter().map(Vec::len).sum::<usize>(), batch.requests.len() * 2);
+
+        set_disks
+            .release_dist_delete_object_locks_batch(held_lock_ids_by_client)
+            .await;
+
+        let local_lock_1 = NamespaceLock::with_local_manager("node-1".to_string(), manager1);
+        let local_lock_2 = NamespaceLock::with_local_manager("node-2".to_string(), manager2);
+
+        let guard_1 = local_lock_1
+            .get_write_lock(ObjectKey::new("bucket", "object-a"), "owner-b", Duration::from_millis(100))
+            .await
+            .expect("released batch lock should free node 1");
+        let guard_2 = local_lock_2
+            .get_write_lock(ObjectKey::new("bucket", "object-b"), "owner-b", Duration::from_millis(100))
+            .await
+            .expect("released batch lock should free node 2");
+
+        drop(guard_1);
+        drop(guard_2);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn test_acquire_dist_delete_object_locks_batch_rolls_back_when_quorum_not_reached() {
+        let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+
+        let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+        let healthy_client: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(manager.clone()));
+        let failing_client: Arc<dyn LockClient> = Arc::new(FailingClient);
+        let set_disks = make_test_set_disks(vec![healthy_client, failing_client]).await;
+
+        let batch = rustfs_lock::BatchLockRequest::new(set_disks.locker_owner.as_str())
+            .with_all_or_nothing(false)
+            .add_write_lock(ObjectKey::new("bucket", "object-a"));
+
+        let (failed_map, locked_objects, held_lock_ids_by_client) =
+            set_disks.acquire_dist_delete_object_locks_batch(&batch).await;
+
+        assert!(locked_objects.is_empty());
+        assert!(failed_map.contains_key(&("bucket".to_string(), "object-a".to_string())));
+        assert_eq!(held_lock_ids_by_client.iter().map(Vec::len).sum::<usize>(), 0);
+
+        let local_lock = NamespaceLock::with_local_manager("node-1".to_string(), manager);
+        let guard = local_lock
+            .get_write_lock(ObjectKey::new("bucket", "object-a"), "owner-b", Duration::from_millis(100))
+            .await
+            .expect("quorum rollback should release the healthy node lock");
+
+        drop(guard);
+    }
+
     #[test]
     fn test_common_parity() {
         // Test common parity calculation

crates/ecstore/src/sets.rs

@@ -35,7 +35,10 @@ use crate::{
     },
     store_init::{check_format_erasure_values, get_format_erasure_in_quorum, load_format_erasure_all, save_format_file},
 };
-use futures::future::join_all;
+use futures::{
+    future::join_all,
+    stream::{FuturesUnordered, StreamExt},
+};
 use http::HeaderMap;
 use rustfs_common::heal_channel::HealOpts;
 use rustfs_common::{
@@ -336,6 +339,26 @@ struct DelObj {
     obj: ObjectToDelete,
 }
 
+fn apply_delete_objects_results(
+    del_objects: &mut [DeletedObject],
+    del_errs: &mut [Option<Error>],
+    set_objects: &[DelObj],
+    dobjects: &[DeletedObject],
+    errs: Vec<Option<Error>>,
+) {
+    for (i, err) in errs.into_iter().enumerate() {
+        let obj = set_objects
+            .get(i)
+            .expect("delete_objects should return errors aligned with input objects");
+
+        del_errs[obj.orig_idx] = err;
+        del_objects[obj.orig_idx] = dobjects
+            .get(i)
+            .expect("delete_objects should return objects aligned with input objects")
+            .clone();
+    }
+}
+
 #[async_trait::async_trait]
 impl ObjectIO for Sets {
     #[tracing::instrument(level = "debug", skip(self, object, h, opts))]
@@ -508,19 +531,30 @@ impl ObjectOperations for Sets {
             }
         }
 
-        // TODO: concurrency
+        let max_concurrent = set_obj_map.len().min(num_cpus::get()).max(1);
+        let semaphore = Arc::new(tokio::sync::Semaphore::new(max_concurrent));
+        let mut futures = FuturesUnordered::new();
+        let bucket = bucket.to_string();
+
         for (k, v) in set_obj_map {
             let disks = self.get_disks(k);
             let objs: Vec<ObjectToDelete> = v.iter().map(|v| v.obj.clone()).collect();
-            let (dobjects, errs) = disks.delete_objects(bucket, objs, opts.clone()).await;
+            let bucket = bucket.clone();
+            let opts = opts.clone();
+            let semaphore = semaphore.clone();
 
-            for (i, err) in errs.into_iter().enumerate() {
-                let obj = v.get(i).unwrap();
+            futures.push(async move {
+                let _permit = semaphore
+                    .acquire_owned()
+                    .await
+                    .expect("delete_objects semaphore should remain open");
+                let (dobjects, errs) = disks.delete_objects(&bucket, objs, opts).await;
+                (v, dobjects, errs)
+            });
+        }
 
-                del_errs[obj.orig_idx] = err;
-
-                del_objects[obj.orig_idx] = dobjects.get(i).unwrap().clone();
-            }
+        while let Some((v, dobjects, errs)) = futures.next().await {
+            apply_delete_objects_results(&mut del_objects, &mut del_errs, &v, &dobjects, errs);
         }
 
         (del_objects, del_errs)
@@ -1015,3 +1049,76 @@ fn new_heal_format_sets(
 
     (new_formats, current_disks_info)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_apply_delete_objects_results_preserves_original_order_for_out_of_order_batches() {
+        let mut del_objects = vec![DeletedObject::default(); 3];
+        let mut del_errs = vec![None, None, None];
+
+        let early_batch = vec![DelObj {
+            orig_idx: 1,
+            obj: ObjectToDelete {
+                object_name: "second".to_string(),
+                ..Default::default()
+            },
+        }];
+        let early_objects = vec![DeletedObject {
+            object_name: "second".to_string(),
+            found: true,
+            ..Default::default()
+        }];
+
+        let late_batch = vec![
+            DelObj {
+                orig_idx: 2,
+                obj: ObjectToDelete {
+                    object_name: "third".to_string(),
+                    ..Default::default()
+                },
+            },
+            DelObj {
+                orig_idx: 0,
+                obj: ObjectToDelete {
+                    object_name: "first".to_string(),
+                    ..Default::default()
+                },
+            },
+        ];
+        let late_objects = vec![
+            DeletedObject {
+                object_name: "third".to_string(),
+                found: true,
+                ..Default::default()
+            },
+            DeletedObject {
+                object_name: "first".to_string(),
+                found: true,
+                ..Default::default()
+            },
+        ];
+
+        apply_delete_objects_results(&mut del_objects, &mut del_errs, &early_batch, &early_objects, vec![None]);
+        apply_delete_objects_results(
+            &mut del_objects,
+            &mut del_errs,
+            &late_batch,
+            &late_objects,
+            vec![Some(Error::other("third failed")), None],
+        );
+
+        assert_eq!(del_objects[0].object_name, "first");
+        assert_eq!(del_objects[1].object_name, "second");
+        assert_eq!(del_objects[2].object_name, "third");
+
+        assert!(del_errs[0].is_none());
+        assert!(del_errs[1].is_none());
+        assert_eq!(
+            del_errs[2].as_ref().map(ToString::to_string),
+            Some(Error::other("third failed").to_string())
+        );
+    }
+}