perf(get): reuse reader paths and lock namespaces (#6015)

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/io_support/bitrot.rs

@@ -120,26 +120,41 @@ struct BitrotReaderSource {
 
 impl BitrotReaderSource {
     async fn open(self) -> disk::error::Result<Option<BoxedObjectReader>> {
-        if let Some(data) = self.inline_data {
+        open_reader_source(
-            let mut rd = Cursor::new(data);
+            self.inline_data,
-            let offset = u64::try_from(self.offset).map_err(|_| DiskError::FileCorrupt)?;
+            self.disk.as_ref(),
-            rd.set_position(offset);
+            &self.bucket,
-            Ok(Some(ShardReader::InMemory(rd)))
+            &self.path,
-        } else if let Some(disk) = self.disk {
+            self.offset,
-            open_disk_reader(
+            self.length,
-                &disk,
+            self.use_mmap_read,
-                &self.bucket,
+            self.stage_metrics.map(|metrics| metrics.path),
-                &self.path,
+        )
-                self.offset,
+        .await
-                self.length,
+    }
-                self.use_mmap_read,
+}
-                self.stage_metrics.map(|metrics| metrics.path),
+
-            )
+#[allow(clippy::too_many_arguments)]
+async fn open_reader_source(
+    inline_data: Option<Bytes>,
+    disk: Option<&DiskStore>,
+    bucket: &str,
+    path: &str,
+    offset: usize,
+    length: usize,
+    use_mmap_read: bool,
+    metrics_path: Option<&'static str>,
+) -> disk::error::Result<Option<BoxedObjectReader>> {
+    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 {
+    } else {
-            Ok(None)
+        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);
     }

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<str>,
     pub format: FormatV3,
     disk_health_cache: Arc<RwLock<Vec<Option<DiskHealthEntry>>>>,
     get_object_metadata_cache: moka::future::Cache<GetObjectMetadataCacheKey, Arc<GetObjectMetadataCacheEntry>>,
@@ -2768,6 +2770,7 @@ impl SetDisks {
         instance_ctx: Arc<InstanceContext>,
     ) -> Arc<Self> {
         let ctx = instance_ctx;
+        let set_lock_namespace: Arc<str> = 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,
     }

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(
+            NamespaceLock::with_clients_and_quorum_shared(self.set_lock_namespace.clone(), self.lockers.clone(), write_quorum)
-                format!("set-{}-{}", self.pool_index, self.set_index),
-                self.lockers.clone(),
-                write_quorum,
-            )
         } else {
-            NamespaceLock::Local(LocalLock::new(
+            NamespaceLock::with_local_manager_shared(self.set_lock_namespace.clone(), self.local_lock_manager.clone())
-                format!("set-{}-{}", self.pool_index, self.set_index),
-                self.local_lock_manager.clone(),
-            ))
         };
 
         let resource = ObjectKey {

crates/lock/src/distributed_lock.rs

@@ -479,7 +479,7 @@ pub struct DistributedLock {
     /// Lock clients for this namespace
     clients: Vec<Arc<dyn LockClient>>,
     /// Namespace identifier
-    namespace: String,
+    namespace: Arc<str>,
     /// 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<Arc<dyn LockClient>>, 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<str>, clients: Vec<Arc<dyn LockClient>>, quorum: usize) -> Self {
         let q = if clients.len() <= 1 {
             1
         } else {

crates/lock/src/local_lock.rs

@@ -28,12 +28,17 @@ pub struct LocalLock {
     /// Global lock manager for fast local locks
     manager: Arc<GlobalLockManager>,
     /// Namespace identifier
-    namespace: String,
+    namespace: Arc<str>,
 }
 
 impl LocalLock {
     /// Create new local lock
     pub fn new(namespace: String, manager: Arc<GlobalLockManager>) -> Self {
+        Self::new_shared(namespace.into(), manager)
+    }
+
+    /// Create a local lock that shares an existing namespace allocation.
+    pub(crate) fn new_shared(namespace: Arc<str>, manager: Arc<GlobalLockManager>) -> Self {
         Self { namespace, manager }
     }
 

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<str>, manager: Arc<crate::GlobalLockManager>) -> 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<Arc<dyn LockClient>>) -> 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<str>, clients: Vec<Arc<dyn LockClient>>, quorum: usize) -> Self {
+        Self::Distributed(DistributedLock::new_shared(namespace, clients, quorum))
+    }
+
     /// Get namespace identifier
     pub fn namespace(&self) -> &str {
         match self {

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<str> = 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::<u8>());
+}
+
 #[tokio::test]
 async fn test_namespace_lock_with_clients() {
     let clients = vec![ClientFactory::create_local(), ClientFactory::create_local()];