diff --git a/crates/ecstore/src/io_support/bitrot.rs b/crates/ecstore/src/io_support/bitrot.rs index c52459a9d..2e046807d 100644 --- a/crates/ecstore/src/io_support/bitrot.rs +++ b/crates/ecstore/src/io_support/bitrot.rs @@ -120,26 +120,41 @@ struct BitrotReaderSource { impl BitrotReaderSource { async fn open(self) -> disk::error::Result> { - if let Some(data) = self.inline_data { - let mut rd = Cursor::new(data); - let offset = u64::try_from(self.offset).map_err(|_| DiskError::FileCorrupt)?; - rd.set_position(offset); - Ok(Some(ShardReader::InMemory(rd))) - } else if let Some(disk) = self.disk { - open_disk_reader( - &disk, - &self.bucket, - &self.path, - self.offset, - self.length, - self.use_mmap_read, - self.stage_metrics.map(|metrics| metrics.path), - ) + open_reader_source( + self.inline_data, + self.disk.as_ref(), + &self.bucket, + &self.path, + self.offset, + self.length, + self.use_mmap_read, + self.stage_metrics.map(|metrics| metrics.path), + ) + .await + } +} + +#[allow(clippy::too_many_arguments)] +async fn open_reader_source( + inline_data: Option, + disk: Option<&DiskStore>, + bucket: &str, + path: &str, + offset: usize, + length: usize, + use_mmap_read: bool, + metrics_path: Option<&'static str>, +) -> disk::error::Result> { + if let Some(data) = inline_data { + let mut reader = Cursor::new(data); + reader.set_position(u64::try_from(offset).map_err(|_| DiskError::FileCorrupt)?); + Ok(Some(ShardReader::InMemory(reader))) + } else if let Some(disk) = disk { + open_disk_reader(disk, bucket, path, offset, length, use_mmap_read, metrics_path) .await .map(Some) - } else { - Ok(None) - } + } else { + Ok(None) } } @@ -623,23 +638,22 @@ async fn create_bitrot_reader_from_bytes_with_stage_metrics( let reader_construction_start = stage_metrics_enabled.then(Instant::now); let (offset, length) = bitrot_encoded_range(offset, length, shard_size, checksum_algo.clone()); - let inline_source = inline_data.is_some(); - let source = BitrotReaderSource { - inline_data, - disk: disk.cloned(), - bucket: if inline_source { String::new() } else { bucket.to_string() }, - path: if inline_source { String::new() } else { path.to_string() }, - offset, - length, - use_mmap_read, - stage_metrics, - }; if let Some(metrics) = stage_metrics { record_get_stage_duration_if_enabled(metrics.path, metrics.reader_construction_stage, reader_construction_start); } let file_open_start = stage_metrics_enabled.then(Instant::now); - let reader = source.open().await?; + let reader = open_reader_source( + inline_data, + disk, + bucket, + path, + offset, + length, + use_mmap_read, + stage_metrics.map(|metrics| metrics.path), + ) + .await?; if let Some(metrics) = stage_metrics { record_get_stage_duration_if_enabled(metrics.path, metrics.file_open_stage, file_open_start); } diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 6f4577096..9a1a29ee6 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -2367,6 +2367,8 @@ pub struct SetDisks { pub default_parity_count: usize, pub set_index: usize, pub pool_index: usize, + /// Stable namespace shared by every object lock created for this set. + set_lock_namespace: Arc, pub format: FormatV3, disk_health_cache: Arc>>>, get_object_metadata_cache: moka::future::Cache>, @@ -2768,6 +2770,7 @@ impl SetDisks { instance_ctx: Arc, ) -> Arc { let ctx = instance_ctx; + let set_lock_namespace: Arc = format!("set-{pool_index}-{set_index}").into(); Arc::new(SetDisks { locker_owner, disks, @@ -2775,6 +2778,7 @@ impl SetDisks { default_parity_count, set_index, pool_index, + set_lock_namespace, format, set_endpoints, disk_health_cache: Arc::new(RwLock::new(Vec::new())), @@ -4934,6 +4938,28 @@ mod tests { ); } + #[tokio::test] + async fn new_ns_lock_reuses_the_set_namespace_allocation() { + let ctx = Arc::new(InstanceContext::new()); + ctx.update_erasure_type(SetupType::Erasure).await; + let set = make_test_set_disks_with_ctx(Vec::new(), ctx).await; + + assert_eq!(&*set.set_lock_namespace, "set-0-0"); + let before = Arc::strong_count(&set.set_lock_namespace); + let lock = set + .new_ns_lock("bucket", "object") + .await + .expect("namespace lock should be created"); + + assert_eq!( + Arc::strong_count(&set.set_lock_namespace), + before + 1, + "each lock should share the set namespace instead of formatting a new String" + ); + drop(lock); + assert_eq!(Arc::strong_count(&set.set_lock_namespace), before); + } + struct SetupTypeGuard { previous: SetupType, } diff --git a/crates/ecstore/src/set_disk/ops/locking.rs b/crates/ecstore/src/set_disk/ops/locking.rs index bfead3a53..e90772b7a 100644 --- a/crates/ecstore/src/set_disk/ops/locking.rs +++ b/crates/ecstore/src/set_disk/ops/locking.rs @@ -39,16 +39,9 @@ impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks { // Calculate quorum based on lockers count (majority) let lockers_count = self.lockers.len(); let write_quorum = if lockers_count > 1 { (lockers_count / 2) + 1 } else { 1 }; - NamespaceLock::with_clients_and_quorum( - format!("set-{}-{}", self.pool_index, self.set_index), - self.lockers.clone(), - write_quorum, - ) + NamespaceLock::with_clients_and_quorum_shared(self.set_lock_namespace.clone(), self.lockers.clone(), write_quorum) } else { - NamespaceLock::Local(LocalLock::new( - format!("set-{}-{}", self.pool_index, self.set_index), - self.local_lock_manager.clone(), - )) + NamespaceLock::with_local_manager_shared(self.set_lock_namespace.clone(), self.local_lock_manager.clone()) }; let resource = ObjectKey { diff --git a/crates/lock/src/distributed_lock.rs b/crates/lock/src/distributed_lock.rs index fbe741746..d1cbadf75 100644 --- a/crates/lock/src/distributed_lock.rs +++ b/crates/lock/src/distributed_lock.rs @@ -479,7 +479,7 @@ pub struct DistributedLock { /// Lock clients for this namespace clients: Vec>, /// Namespace identifier - namespace: String, + namespace: Arc, /// Quorum size for exclusive/write operations quorum: usize, } @@ -496,6 +496,11 @@ struct LockAcquireQuorumResult { impl DistributedLock { /// Create new distributed lock pub fn new(namespace: String, clients: Vec>, quorum: usize) -> Self { + Self::new_shared(namespace.into(), clients, quorum) + } + + /// Create a distributed lock that shares an existing namespace allocation. + pub(crate) fn new_shared(namespace: Arc, clients: Vec>, quorum: usize) -> Self { let q = if clients.len() <= 1 { 1 } else { diff --git a/crates/lock/src/local_lock.rs b/crates/lock/src/local_lock.rs index 9ef6e03fb..66f6c3de7 100644 --- a/crates/lock/src/local_lock.rs +++ b/crates/lock/src/local_lock.rs @@ -28,12 +28,17 @@ pub struct LocalLock { /// Global lock manager for fast local locks manager: Arc, /// Namespace identifier - namespace: String, + namespace: Arc, } impl LocalLock { /// Create new local lock pub fn new(namespace: String, manager: Arc) -> Self { + Self::new_shared(namespace.into(), manager) + } + + /// Create a local lock that shares an existing namespace allocation. + pub(crate) fn new_shared(namespace: Arc, manager: Arc) -> Self { Self { namespace, manager } } diff --git a/crates/lock/src/namespace/mod.rs b/crates/lock/src/namespace/mod.rs index 96fd73383..4a8854185 100644 --- a/crates/lock/src/namespace/mod.rs +++ b/crates/lock/src/namespace/mod.rs @@ -180,6 +180,11 @@ impl NamespaceLock { Self::Local(LocalLock::new(namespace, manager)) } + /// Create a local namespace lock that shares an existing namespace allocation. + pub fn with_local_manager_shared(namespace: Arc, manager: Arc) -> Self { + Self::Local(LocalLock::new_shared(namespace, manager)) + } + /// Create namespace lock with clients /// Uses DistributedLock with appropriate quorum pub fn with_clients(namespace: String, clients: Vec>) -> Self { @@ -195,6 +200,11 @@ impl NamespaceLock { Self::Distributed(DistributedLock::new(namespace, clients, quorum)) } + /// Create a namespace lock that shares an existing namespace allocation. + pub fn with_clients_and_quorum_shared(namespace: Arc, clients: Vec>, quorum: usize) -> Self { + Self::Distributed(DistributedLock::new_shared(namespace, clients, quorum)) + } + /// Get namespace identifier pub fn namespace(&self) -> &str { match self { diff --git a/crates/lock/src/namespace/tests.rs b/crates/lock/src/namespace/tests.rs index b1b99df5e..0ee5a81bb 100644 --- a/crates/lock/src/namespace/tests.rs +++ b/crates/lock/src/namespace/tests.rs @@ -356,6 +356,16 @@ async fn test_namespace_lock_with_local_manager() { assert_eq!(lock.namespace(), "local-ns"); } +#[tokio::test] +async fn namespace_lock_preserves_shared_namespace_storage() { + let namespace: Arc = Arc::from("shared-namespace"); + let namespace_ptr = Arc::as_ptr(&namespace); + let local = LocalLock::new_shared(namespace.clone(), Arc::new(GlobalLockManager::new())); + + assert_eq!(local.namespace(), namespace.as_ref()); + assert_eq!(local.namespace().as_ptr(), namespace_ptr.cast::()); +} + #[tokio::test] async fn test_namespace_lock_with_clients() { let clients = vec![ClientFactory::create_local(), ClientFactory::create_local()];