perf(lock): Use global lock manager, instead of one per request (#1848)

Co-authored-by: loverustfs <hello@rustfs.com>
Co-authored-by: houseme <housemecn@gmail.com>

crates/ecstore/src/set_disk.rs

@@ -147,6 +147,7 @@ pub struct SetDisks {
     pub format: FormatV3,
     disk_health_cache: Arc<RwLock<Vec<Option<DiskHealthEntry>>>>,
     pub lockers: Vec<Arc<dyn LockClient>>,
+    local_lock_manager: Arc<rustfs_lock::GlobalLockManager>,
 }
 
 #[derive(Clone, Debug)]
@@ -189,6 +190,7 @@ impl SetDisks {
             set_endpoints,
             disk_health_cache: Arc::new(RwLock::new(Vec::new())),
             lockers,
+            local_lock_manager: rustfs_lock::get_global_lock_manager(),
         })
     }
 
@@ -801,7 +803,7 @@ impl StorageAPI for SetDisks {
         } else {
             NamespaceLock::Local(LocalLock::new(
                 format!("set-{}-{}", self.pool_index, self.set_index),
-                Arc::new(rustfs_lock::GlobalLockManager::new()),
+                self.local_lock_manager.clone(),
             ))
         };
 
@@ -2993,15 +2995,17 @@ impl StorageAPI for SetDisks {
         }
 
         // Heal the object.
-        let (result, err) = self.heal_object(bucket, object, version_id, opts).await?;
+        // Pass no_lock=true since we already obtained write lock (or are already called with no_lock=true)
+        let mut inner_opts = *opts;
+        inner_opts.no_lock = true;
+        let (result, err) = self.heal_object(bucket, object, version_id, &inner_opts).await?;
         if let Some(err) = err.as_ref() {
             match err {
                 &DiskError::FileCorrupt if opts.scan_mode != HealScanMode::Deep => {
                     // Instead of returning an error when a bitrot error is detected
                     // during a normal heal scan, heal again with bitrot flag enabled.
-                    let mut opts = *opts;
-                    opts.scan_mode = HealScanMode::Deep;
-                    let (result, err) = self.heal_object(bucket, object, version_id, &opts).await?;
+                    inner_opts.scan_mode = HealScanMode::Deep;
+                    let (result, err) = self.heal_object(bucket, object, version_id, &inner_opts).await?;
                     return Ok((result, err.map(|e| e.into())));
                 }
                 _ => {}