diff --git a/crates/e2e_test/src/reliant/grpc_lock_client.rs b/crates/e2e_test/src/reliant/grpc_lock_client.rs index fc772d520..1045dbf5e 100644 --- a/crates/e2e_test/src/reliant/grpc_lock_client.rs +++ b/crates/e2e_test/src/reliant/grpc_lock_client.rs @@ -21,7 +21,7 @@ use rustfs_lock::{ LockClient, LockError, LockId, LockInfo, LockRequest, LockResponse, LockStats, LockStatus, LockType, Result, types::{LockMetadata, LockPriority}, }; -use rustfs_protos::proto_gen::node_service::{GenerallyLockRequest, PingRequest}; +use rustfs_protos::proto_gen::node_service::{BatchGenerallyLockRequest, GenerallyLockRequest, PingRequest}; use tonic::Request; use tracing::{info, warn}; @@ -64,6 +64,44 @@ impl GrpcLockClient { suppress_contention_logs: false, } } + + fn build_lock_info(request: &LockRequest, lock_info_json: Option) -> LockInfo { + if let Some(lock_info_json) = lock_info_json { + match serde_json::from_str::(&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] @@ -89,46 +127,10 @@ impl LockClient for GrpcLockClient { // 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::(&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( @@ -138,6 +140,45 @@ impl LockClient for GrpcLockClient { } } + async fn acquire_locks_batch(&self, requests: &[LockRequest]) -> Result> { + 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::>>()?, + }); + + 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 { info!("grpc release for {}", lock_id); @@ -161,6 +202,31 @@ impl LockClient for GrpcLockClient { Ok(resp.success) } + async fn release_locks_batch(&self, lock_ids: &[LockId]) -> Result> { + 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::>>()?, + }); + + 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 { info!("grpc refresh for {}", lock_id); let refresh_request = Self::create_unlock_request(lock_id); diff --git a/crates/e2e_test/src/reliant/grpc_lock_server.rs b/crates/e2e_test/src/reliant/grpc_lock_server.rs index c1a927124..628152e5b 100644 --- a/crates/e2e_test/src/reliant/grpc_lock_server.rs +++ b/crates/e2e_test/src/reliant/grpc_lock_server.rs @@ -21,7 +21,8 @@ use rustfs_lock::{LockClient, LockRequest}; use rustfs_protos::{ models::PingBodyBuilder, proto_gen::node_service::{ - GenerallyLockRequest, GenerallyLockResponse, PingRequest, PingResponse, node_service_server::NodeService, + BatchGenerallyLockRequest, BatchGenerallyLockResponse, GenerallyLockRequest, GenerallyLockResponse, GenerallyLockResult, + PingRequest, PingResponse, node_service_server::NodeService, }, }; use std::pin::Pin; @@ -33,6 +34,22 @@ use tracing::debug; type ResponseStream = Pin> + Send>>; +fn lock_result_from_response(response: rustfs_lock::LockResponse) -> GenerallyLockResult { + GenerallyLockResult { + success: response.success, + error_info: response.error, + lock_info: response.lock_info.and_then(|info| serde_json::to_string(&info).ok()), + } +} + +fn lock_result_from_error(error: impl Into) -> GenerallyLockResult { + GenerallyLockResult { + success: false, + error_info: Some(error.into()), + lock_info: None, + } +} + /// Minimal NodeService implementation that only supports Lock RPCs /// Used for testing distributed lock scenarios with real gRPC #[derive(Debug)] @@ -187,6 +204,92 @@ impl NodeService for MinimalLockNodeService { } } + async fn lock_batch( + &self, + request: Request, + ) -> Result, Status> { + let request = request.into_inner(); + let mut results = vec![lock_result_from_error("request was not processed"); request.args.len()]; + let mut valid_requests = Vec::with_capacity(request.args.len()); + let mut valid_indices = Vec::with_capacity(request.args.len()); + + for (idx, arg) in request.args.iter().enumerate() { + match serde_json::from_str::(arg) { + Ok(args) => { + valid_requests.push(args); + valid_indices.push(idx); + } + Err(err) => { + results[idx] = lock_result_from_error(format!("can not decode args, err: {err}")); + } + } + } + + if !valid_requests.is_empty() { + match self.lock_client.acquire_locks_batch(&valid_requests).await { + Ok(batch_results) => { + for (result_idx, response) in batch_results.into_iter().enumerate() { + if let Some(request_idx) = valid_indices.get(result_idx) { + results[*request_idx] = lock_result_from_response(response); + } + } + } + Err(err) => { + for request_idx in valid_indices { + results[request_idx] = lock_result_from_error(format!("can not batch lock, err: {err}")); + } + } + } + } + + Ok(Response::new(BatchGenerallyLockResponse { results })) + } + + async fn un_lock_batch( + &self, + request: Request, + ) -> Result, Status> { + let request = request.into_inner(); + let mut results = vec![lock_result_from_error("request was not processed"); request.args.len()]; + let mut lock_ids = Vec::with_capacity(request.args.len()); + let mut valid_indices = Vec::with_capacity(request.args.len()); + + for (idx, arg) in request.args.iter().enumerate() { + match serde_json::from_str::(arg) { + Ok(args) => { + lock_ids.push(args.lock_id); + valid_indices.push(idx); + } + Err(err) => { + results[idx] = lock_result_from_error(format!("can not decode args, err: {err}")); + } + } + } + + if !lock_ids.is_empty() { + match self.lock_client.release_locks_batch(&lock_ids).await { + Ok(batch_results) => { + for (result_idx, success) in batch_results.into_iter().enumerate() { + if let Some(request_idx) = valid_indices.get(result_idx) { + results[*request_idx] = GenerallyLockResult { + success, + error_info: None, + lock_info: None, + }; + } + } + } + Err(err) => { + for request_idx in valid_indices { + results[request_idx] = lock_result_from_error(format!("can not batch unlock, err: {err}")); + } + } + } + } + + Ok(Response::new(BatchGenerallyLockResponse { results })) + } + // All other methods return unimplemented async fn heal_bucket( &self, diff --git a/crates/e2e_test/src/reliant/lock.rs b/crates/e2e_test/src/reliant/lock.rs index 7cae2c390..d7dfc536c 100644 --- a/crates/e2e_test/src/reliant/lock.rs +++ b/crates/e2e_test/src/reliant/lock.rs @@ -14,8 +14,10 @@ // limitations under the License. use super::{grpc_lock_client::GrpcLockClient, grpc_lock_server::spawn_lock_server}; -use rustfs_lock::client::local::LocalClient; -use rustfs_lock::{GlobalLockManager, LockError, LockInfo, LockResponse, LockStats, NamespaceLock, ObjectKey}; +use rustfs_lock::client::{LockClient, local::LocalClient}; +use rustfs_lock::{ + GlobalLockManager, LockError, LockInfo, LockRequest, LockResponse, LockStats, LockType, NamespaceLock, ObjectKey, +}; use std::sync::Arc; use std::time::Duration; @@ -223,6 +225,56 @@ async fn test_distributed_lock_2_nodes_grpc_read_survives_failed_node() { failing_handle.abort(); } +#[tokio::test] +async fn test_grpc_lock_client_batch_acquire_and_release() { + let manager = Arc::new(GlobalLockManager::new()); + let local_client: Arc = Arc::new(LocalClient::with_manager(manager)); + + let (addr, handle) = spawn_lock_server(local_client).await.expect("Failed to spawn server"); + tokio::time::sleep(Duration::from_millis(100)).await; + + let grpc_client = GrpcLockClient::new(addr); + let requests = vec![ + LockRequest::new(test_resource(), LockType::Exclusive, "owner-a").with_acquire_timeout(Duration::from_secs(2)), + LockRequest::new( + ObjectKey { + bucket: Arc::from("test-bucket"), + object: Arc::from("test-object-2"), + version: None, + }, + LockType::Exclusive, + "owner-a", + ) + .with_acquire_timeout(Duration::from_secs(2)), + ]; + + let responses = grpc_client + .acquire_locks_batch(&requests) + .await + .expect("batch acquire should succeed"); + assert_eq!(responses.len(), requests.len()); + assert!(responses.iter().all(|response| response.success)); + + let lock_ids = responses + .iter() + .map(|response| { + response + .lock_info + .as_ref() + .expect("batch response should include lock info") + .id + .clone() + }) + .collect::>(); + let released = grpc_client + .release_locks_batch(&lock_ids) + .await + .expect("batch release should succeed"); + assert_eq!(released, vec![true, true]); + + handle.abort(); +} + #[tokio::test] async fn test_distributed_lock_4_nodes_grpc_read_write_quorum_split_with_two_failed_nodes() { let manager1 = Arc::new(GlobalLockManager::new()); diff --git a/crates/ecstore/src/rpc/remote_locker.rs b/crates/ecstore/src/rpc/remote_locker.rs index e3f34ed4d..4e786a4d2 100644 --- a/crates/ecstore/src/rpc/remote_locker.rs +++ b/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) -> LockInfo { + if let Some(lock_info_json) = lock_info_json { + match serde_json::from_str::(&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::(&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> { + 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::>>()?, + }); + + 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 { 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> { + 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::>>()?, + }); + + 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 { info!("remote refresh for {}", lock_id); let refresh_request = Self::create_unlock_request(lock_id); diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index f76fad893..7a150245b 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/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, Vec>) { + let requests: Vec = 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![Vec::new(); requests.len()]; + let mut errors_by_object: Vec> = 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>) { + 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 = 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 = Arc::new(LocalClient::with_manager(manager1.clone())); + let client2: Arc = 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::(), 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 = Arc::new(LocalClient::with_manager(manager.clone())); + let failing_client: Arc = 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::(), 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 diff --git a/crates/ecstore/src/sets.rs b/crates/ecstore/src/sets.rs index d623d6cd4..48502a4b6 100644 --- a/crates/ecstore/src/sets.rs +++ b/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], + set_objects: &[DelObj], + dobjects: &[DeletedObject], + errs: Vec>, +) { + 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 = 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()) + ); + } +} diff --git a/crates/lock/src/client/mod.rs b/crates/lock/src/client/mod.rs index ee3f4786d..d16db4bb6 100644 --- a/crates/lock/src/client/mod.rs +++ b/crates/lock/src/client/mod.rs @@ -17,6 +17,7 @@ pub mod local; use crate::{LockId, LockInfo, LockRequest, LockResponse, LockStats, Result}; use async_trait::async_trait; +use futures::future::join_all; use std::sync::Arc; /// Lock client trait @@ -25,9 +26,25 @@ pub trait LockClient: Send + Sync + std::fmt::Debug { /// Acquire lock (generic method) async fn acquire_lock(&self, request: &LockRequest) -> Result; + /// Acquire multiple locks. Default implementation fans out to single-lock requests. + async fn acquire_locks_batch(&self, requests: &[LockRequest]) -> Result> { + Ok(join_all(requests.iter().map(|request| self.acquire_lock(request))) + .await + .into_iter() + .collect::>>()?) + } + /// Release lock async fn release(&self, lock_id: &LockId) -> Result; + /// Release multiple locks. Default implementation fans out to single-lock releases. + async fn release_locks_batch(&self, lock_ids: &[LockId]) -> Result> { + Ok(join_all(lock_ids.iter().map(|lock_id| self.release(lock_id))) + .await + .into_iter() + .collect::>>()?) + } + /// Refresh lock async fn refresh(&self, lock_id: &LockId) -> Result; diff --git a/crates/lock/src/distributed_lock.rs b/crates/lock/src/distributed_lock.rs index f461fffff..9928daffe 100644 --- a/crates/lock/src/distributed_lock.rs +++ b/crates/lock/src/distributed_lock.rs @@ -18,6 +18,7 @@ use crate::{ error::{LockError, Result}, types::{LockId, LockInfo, LockRequest, LockResponse, LockStatus, LockType}, }; +use futures::future::join_all; use std::sync::{Arc, LazyLock}; use std::time::Duration; use tokio::sync::mpsc; @@ -52,12 +53,13 @@ static UNLOCK_RUNTIME: LazyLock = LazyLock::new(|| { tokio::spawn(async move { while let Some(job) = rx.recv().await { // Best-effort release across all (LockId, client) entries. - let mut any_ok = false; - for (lock_id, client) in job.entries.into_iter() { - if client.release(&lock_id).await.unwrap_or(false) { - any_ok = true; - } - } + let results = join_all( + job.entries + .into_iter() + .map(|(lock_id, client)| async move { client.release(&lock_id).await.unwrap_or(false) }), + ) + .await; + let any_ok = results.into_iter().any(|released| released); if !any_ok { tracing::warn!("DistributedLockGuard background release failed for one or more entries"); @@ -142,7 +144,7 @@ impl DistributedLockGuard { let futures_iter = entries .into_iter() .map(|(lock_id, client)| async move { client.release(&lock_id).await.unwrap_or(false) }); - let _ = futures::future::join_all(futures_iter).await; + let _ = join_all(futures_iter).await; }); // Explicitly drop the JoinHandle to acknowledge detaching the task. drop(handle); @@ -411,8 +413,13 @@ impl DistributedLock { } else { // Rollback: release all locks that were successfully acquired let rollback_count = individual_locks.len(); - for (individual_lock_id, client) in &individual_locks { - if let Err(e) = client.release(individual_lock_id).await { + let rollback_results = join_all(individual_locks.iter().map(|(individual_lock_id, client)| async move { + (individual_lock_id, client.release(individual_lock_id).await) + })) + .await; + + for (individual_lock_id, result) in rollback_results { + if let Err(e) = result { tracing::warn!("Failed to rollback lock {} on client: {}", individual_lock_id, e); } } diff --git a/crates/lock/src/fast_lock/manager.rs b/crates/lock/src/fast_lock/manager.rs index d8c5c9295..4eca21210 100644 --- a/crates/lock/src/fast_lock/manager.rs +++ b/crates/lock/src/fast_lock/manager.rs @@ -138,7 +138,7 @@ impl FastObjectLockManager { shard_a.cmp(&shard_b).then_with(|| a.key.cmp(&b.key)) }); - // Try to use stack-allocated vectors for small batches, fallback to heap if needed + // Preserve shard order so every concurrent batch acquires locks in the same global order. let shard_groups = self.group_requests_by_shard(sorted_requests); // Choose strategy based on request type @@ -150,31 +150,28 @@ impl FastObjectLockManager { } /// Group requests by shard with proper fallback handling - fn group_requests_by_shard( - &self, - requests: Vec, - ) -> std::collections::HashMap> { - let mut shard_groups = std::collections::HashMap::new(); + fn group_requests_by_shard(&self, requests: Vec) -> Vec<(usize, Vec)> { + let mut shard_groups: Vec<(usize, Vec)> = Vec::new(); for request in requests { let shard_id = request.key.shard_index(self.shard_mask); - shard_groups.entry(shard_id).or_insert_with(Vec::new).push(request); + match shard_groups.last_mut() { + Some((last_shard_id, grouped_requests)) if *last_shard_id == shard_id => grouped_requests.push(request), + _ => shard_groups.push((shard_id, vec![request])), + } } shard_groups } /// Best effort acquisition (allows partial success) - async fn acquire_locks_best_effort( - &self, - shard_groups: &std::collections::HashMap>, - ) -> BatchLockResult { + async fn acquire_locks_best_effort(&self, shard_groups: &[(usize, Vec)]) -> BatchLockResult { let mut all_successful = Vec::new(); let mut all_failed = Vec::new(); let mut guards = Vec::new(); - for (&shard_id, requests) in shard_groups { - let shard = self.shards[shard_id].clone(); + for (shard_id, requests) in shard_groups { + let shard = self.shards[*shard_id].clone(); for request in requests { let key = request.key.clone(); @@ -212,16 +209,13 @@ impl FastObjectLockManager { } /// Two-phase commit for atomic acquisition - async fn acquire_locks_two_phase_commit( - &self, - shard_groups: &std::collections::HashMap>, - ) -> BatchLockResult { + async fn acquire_locks_two_phase_commit(&self, shard_groups: &[(usize, Vec)]) -> BatchLockResult { // Phase 1: Try to acquire all locks let mut acquired_guards = Vec::new(); let mut failed_locks = Vec::new(); - 'outer: for (&shard_id, requests) in shard_groups { - let shard = self.shards[shard_id].clone(); + 'outer: for (shard_id, requests) in shard_groups { + let shard = self.shards[*shard_id].clone(); for request in requests { match shard.acquire_lock(request).await { @@ -438,3 +432,48 @@ impl LockManager for FastObjectLockManager { false } } + +#[cfg(test)] +mod tests { + use super::*; + + fn make_request(manager: &FastObjectLockManager, shard_id: usize, suffix: usize) -> ObjectLockRequest { + let mut candidate = 0usize; + loop { + let object = format!("object-{shard_id}-{suffix}-{candidate}"); + let key = ObjectKey::new("bucket", object); + if key.shard_index(manager.shard_mask) == shard_id { + return ObjectLockRequest::new_write(key, "owner"); + } + candidate += 1; + } + } + + #[tokio::test] + async fn test_group_requests_by_shard_preserves_sorted_shard_order() { + let manager = FastObjectLockManager::new(); + let mut requests = vec![ + make_request(&manager, 3, 0), + make_request(&manager, 1, 0), + make_request(&manager, 2, 0), + make_request(&manager, 1, 1), + make_request(&manager, 3, 1), + ]; + + requests.sort_unstable_by(|a, b| { + let shard_a = a.key.shard_index(manager.shard_mask); + let shard_b = b.key.shard_index(manager.shard_mask); + shard_a.cmp(&shard_b).then_with(|| a.key.cmp(&b.key)) + }); + + let shard_groups = manager.group_requests_by_shard(requests); + let shard_ids: Vec<_> = shard_groups.iter().map(|(shard_id, _)| *shard_id).collect(); + + assert_eq!(shard_ids, vec![1, 2, 3]); + assert_eq!(shard_groups[0].1.len(), 2); + assert_eq!(shard_groups[1].1.len(), 1); + assert_eq!(shard_groups[2].1.len(), 2); + + manager.shutdown().await; + } +} diff --git a/crates/lock/src/namespace/tests.rs b/crates/lock/src/namespace/tests.rs index f8b48c882..1b11caa79 100644 --- a/crates/lock/src/namespace/tests.rs +++ b/crates/lock/src/namespace/tests.rs @@ -97,6 +97,37 @@ async fn test_namespace_lock_with_clients() { assert_eq!(lock.namespace(), "multi-client"); } +#[tokio::test] +async fn test_lock_client_default_batch_acquire_and_release() { + let manager = Arc::new(GlobalLockManager::new()); + let client = LocalClient::with_manager(manager); + let requests = vec![ + LockRequest::new(create_test_object_key("bucket", "object-a"), LockType::Exclusive, "owner-a") + .with_acquire_timeout(Duration::from_secs(1)), + LockRequest::new(create_test_object_key("bucket", "object-b"), LockType::Exclusive, "owner-a") + .with_acquire_timeout(Duration::from_secs(1)), + ]; + + let responses = client.acquire_locks_batch(&requests).await.unwrap(); + assert_eq!(responses.len(), requests.len()); + assert!(responses.iter().all(|response| response.success)); + + let lock_ids = responses + .iter() + .map(|response| { + response + .lock_info + .as_ref() + .expect("successful batch acquire should return lock info") + .id + .clone() + }) + .collect::>(); + let released = client.release_locks_batch(&lock_ids).await.unwrap(); + + assert_eq!(released, vec![true, true]); +} + #[tokio::test] async fn test_namespace_lock_get_resource_key() { let client = ClientFactory::create_local(); @@ -452,6 +483,45 @@ async fn test_namespace_lock_distributed_write_lock_fails_with_two_nodes_one_off ); } +#[tokio::test] +async fn test_namespace_lock_distributed_quorum_failure_rolls_back_successful_nodes() { + let manager1 = Arc::new(GlobalLockManager::new()); + let manager2 = Arc::new(GlobalLockManager::new()); + + let client1: Arc = Arc::new(LocalClient::with_manager(manager1.clone())); + let client2: Arc = Arc::new(LocalClient::with_manager(manager2.clone())); + let client3: Arc = Arc::new(FailingClient); + + let resource = create_test_object_key("bucket", "object"); + + let distributed_lock = NamespaceLock::with_clients_and_quorum("three-node".to_string(), vec![client1, client2, client3], 3); + let err = distributed_lock + .get_write_lock(resource.clone(), "owner-a", Duration::from_millis(100)) + .await + .expect_err("write lock should fail when quorum requires all three nodes"); + + let err_str = err.to_string().to_lowercase(); + assert!( + err_str.contains("quorum") || err_str.contains("not reached"), + "expected quorum error, got: {err}" + ); + + 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 guard1 = local_lock_1 + .get_write_lock(resource.clone(), "owner-b", Duration::from_millis(100)) + .await + .expect("quorum rollback should release node 1"); + let guard2 = local_lock_2 + .get_write_lock(resource, "owner-b", Duration::from_millis(100)) + .await + .expect("quorum rollback should release node 2"); + + drop(guard1); + drop(guard2); +} + #[tokio::test] async fn test_namespace_lock_distributed_even_node_read_write_quorum_split() { let manager1 = Arc::new(GlobalLockManager::new()); diff --git a/crates/protos/src/generated/proto_gen/node_service.rs b/crates/protos/src/generated/proto_gen/node_service.rs index 0efecb167..b71cc1393 100644 --- a/crates/protos/src/generated/proto_gen/node_service.rs +++ b/crates/protos/src/generated/proto_gen/node_service.rs @@ -658,6 +658,26 @@ pub struct GenerallyLockResponse { #[prost(string, optional, tag = "3")] pub lock_info: ::core::option::Option<::prost::alloc::string::String>, } +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct BatchGenerallyLockRequest { + #[prost(string, repeated, tag = "1")] + pub args: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct GenerallyLockResult { + #[prost(bool, tag = "1")] + pub success: bool, + #[prost(string, optional, tag = "2")] + pub error_info: ::core::option::Option<::prost::alloc::string::String>, + /// JSON serialized LockInfo + #[prost(string, optional, tag = "3")] + pub lock_info: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct BatchGenerallyLockResponse { + #[prost(message, repeated, tag = "1")] + pub results: ::prost::alloc::vec::Vec, +} #[derive(Clone, PartialEq, ::prost::Message)] pub struct Mss { #[prost(map = "string, string", tag = "1")] @@ -1776,6 +1796,36 @@ pub mod node_service_client { .insert(GrpcMethod::new("node_service.NodeService", "Refresh")); self.inner.unary(req, path, codec).await } + pub async fn lock_batch( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner + .ready() + .await + .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/LockBatch"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("node_service.NodeService", "LockBatch")); + self.inner.unary(req, path, codec).await + } + pub async fn un_lock_batch( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner + .ready() + .await + .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/UnLockBatch"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("node_service.NodeService", "UnLockBatch")); + self.inner.unary(req, path, codec).await + } pub async fn local_storage_info( &mut self, request: impl tonic::IntoRequest, @@ -2512,6 +2562,14 @@ pub mod node_service_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn lock_batch( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; + async fn un_lock_batch( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; async fn local_storage_info( &self, request: tonic::Request, @@ -3828,6 +3886,62 @@ pub mod node_service_server { }; Box::pin(fut) } + "/node_service.NodeService/LockBatch" => { + #[allow(non_camel_case_types)] + struct LockBatchSvc(pub Arc); + impl tonic::server::UnaryService for LockBatchSvc { + type Response = super::BatchGenerallyLockResponse; + type Future = BoxFuture, tonic::Status>; + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { ::lock_batch(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = LockBatchSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config(accept_compression_encodings, send_compression_encodings) + .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + "/node_service.NodeService/UnLockBatch" => { + #[allow(non_camel_case_types)] + struct UnLockBatchSvc(pub Arc); + impl tonic::server::UnaryService for UnLockBatchSvc { + type Response = super::BatchGenerallyLockResponse; + type Future = BoxFuture, tonic::Status>; + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { ::un_lock_batch(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = UnLockBatchSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config(accept_compression_encodings, send_compression_encodings) + .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/node_service.NodeService/LocalStorageInfo" => { #[allow(non_camel_case_types)] struct LocalStorageInfoSvc(pub Arc); diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index c3a20fd62..1c7111ceb 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -456,6 +456,20 @@ message GenerallyLockResponse { optional string lock_info = 3; // JSON serialized LockInfo } +message BatchGenerallyLockRequest { + repeated string args = 1; +} + +message GenerallyLockResult { + bool success = 1; + optional string error_info = 2; + optional string lock_info = 3; // JSON serialized LockInfo +} + +message BatchGenerallyLockResponse { + repeated GenerallyLockResult results = 1; +} + message Mss { map value = 1; } @@ -837,6 +851,8 @@ service NodeService { rpc UnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; rpc ForceUnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; rpc Refresh(GenerallyLockRequest) returns (GenerallyLockResponse) {}; + rpc LockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {}; + rpc UnLockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {}; /* -------------------------------peer rest service-------------------------- */ diff --git a/rustfs/src/storage/rpc/lock.rs b/rustfs/src/storage/rpc/lock.rs index 419457476..01e547190 100644 --- a/rustfs/src/storage/rpc/lock.rs +++ b/rustfs/src/storage/rpc/lock.rs @@ -14,6 +14,22 @@ use super::*; +fn lock_result_from_response(response: rustfs_lock::LockResponse) -> GenerallyLockResult { + GenerallyLockResult { + success: response.success, + error_info: response.error, + lock_info: response.lock_info.and_then(|info| serde_json::to_string(&info).ok()), + } +} + +fn lock_result_from_error(error: impl Into) -> GenerallyLockResult { + GenerallyLockResult { + success: false, + error_info: Some(error.into()), + lock_info: None, + } +} + impl NodeService { pub(super) async fn handle_refresh( &self, @@ -144,4 +160,92 @@ impl NodeService { })), } } + + pub(super) async fn handle_lock_batch( + &self, + request: Request, + ) -> Result, Status> { + let request = request.into_inner(); + let mut results = vec![lock_result_from_error("request was not processed"); request.args.len()]; + let mut valid_requests = Vec::with_capacity(request.args.len()); + let mut valid_indices = Vec::with_capacity(request.args.len()); + + for (idx, arg) in request.args.iter().enumerate() { + match serde_json::from_str::(arg) { + Ok(args) => { + valid_requests.push(args); + valid_indices.push(idx); + } + Err(err) => { + results[idx] = lock_result_from_error(format!("can not decode args, err: {err}")); + } + } + } + + if !valid_requests.is_empty() { + let lock_client = self.get_lock_client()?; + match lock_client.acquire_locks_batch(&valid_requests).await { + Ok(batch_results) => { + for (result_idx, response) in batch_results.into_iter().enumerate() { + if let Some(request_idx) = valid_indices.get(result_idx) { + results[*request_idx] = lock_result_from_response(response); + } + } + } + Err(err) => { + for request_idx in valid_indices { + results[request_idx] = lock_result_from_error(format!("can not batch lock, err: {err}")); + } + } + } + } + + Ok(Response::new(BatchGenerallyLockResponse { results })) + } + + pub(super) async fn handle_un_lock_batch( + &self, + request: Request, + ) -> Result, Status> { + let request = request.into_inner(); + let mut results = vec![lock_result_from_error("request was not processed"); request.args.len()]; + let mut lock_ids = Vec::with_capacity(request.args.len()); + let mut valid_indices = Vec::with_capacity(request.args.len()); + + for (idx, arg) in request.args.iter().enumerate() { + match serde_json::from_str::(arg) { + Ok(args) => { + lock_ids.push(args.lock_id); + valid_indices.push(idx); + } + Err(err) => { + results[idx] = lock_result_from_error(format!("can not decode args, err: {err}")); + } + } + } + + if !lock_ids.is_empty() { + let lock_client = self.get_lock_client()?; + match lock_client.release_locks_batch(&lock_ids).await { + Ok(batch_results) => { + for (result_idx, success) in batch_results.into_iter().enumerate() { + if let Some(request_idx) = valid_indices.get(result_idx) { + results[*request_idx] = GenerallyLockResult { + success, + error_info: None, + lock_info: None, + }; + } + } + } + Err(err) => { + for request_idx in valid_indices { + results[request_idx] = lock_result_from_error(format!("can not batch unlock, err: {err}")); + } + } + } + } + + Ok(Response::new(BatchGenerallyLockResponse { results })) + } } diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index c0989a9e1..a4bc7a814 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -386,6 +386,20 @@ impl Node for NodeService { self.handle_refresh(request).await } + async fn lock_batch( + &self, + request: Request, + ) -> Result, Status> { + self.handle_lock_batch(request).await + } + + async fn un_lock_batch( + &self, + request: Request, + ) -> Result, Status> { + self.handle_un_lock_batch(request).await + } + async fn local_storage_info( &self, _request: Request,