chore: 添加删除并发限制

Cargo.lock

@@ -447,6 +447,7 @@ dependencies = [
  "http",
  "lazy_static",
  "netif",
+ "num_cpus",
  "openssl",
  "path-absolutize",
  "path-clean",

ecstore/Cargo.toml

@@ -38,13 +38,14 @@ base64-simd = "0.8.0"
 sha2 = "0.10.8"
 hex-simd = "0.8.0"
 path-clean = "1.0.1"
-tokio = { workspace = true, features = ["io-util"] }
+tokio = { workspace = true, features = ["io-util", "sync"] }
 tokio-stream = "0.1.15"
 tonic.workspace = true
 tower.workspace = true
 rmp = "0.8.14"
 byteorder = "1.5.0"
 xxhash-rust = { version = "0.8.12", features = ["xxh64"] }
+num_cpus = "1.16"
 
 [target.'cfg(not(windows))'.dependencies]
 openssl = "0.10.66"

ecstore/src/sets.rs

@@ -22,7 +22,7 @@ use crate::{
     utils::hash,
 };
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Sets {
     pub id: Uuid,
     // pub sets: Vec<Objects>,

ecstore/src/store.rs

@@ -24,7 +24,10 @@ use std::{
     time::Duration,
 };
 use time::OffsetDateTime;
-use tokio::{fs, sync::RwLock};
+use tokio::{
+    fs,
+    sync::{RwLock, Semaphore},
+};
 use tracing::{debug, info, warn};
 use uuid::Uuid;
 
@@ -391,62 +394,71 @@ impl ECStore {
         object: &str,
         opts: &ObjectOptions,
     ) -> Result<(PoolObjInfo, Vec<Error>)> {
-        let mut futures = Vec::new();
-
-        for pool in self.pools.iter() {
-            futures.push(pool.get_object_info(bucket, object, opts));
-        }
-
-        let results = join_all(futures).await;
-
-        let mut ress = Vec::new();
-
-        let mut i = 0;
-
-        // join_all结果跟输入顺序一致
-        for res in results {
-            let index = i;
-
-            match res {
-                Ok(r) => {
-                    ress.push(PoolObjInfo {
-                        index,
-                        object_info: r,
-                        err: None,
-                    });
-                }
-                Err(e) => {
-                    ress.push(PoolObjInfo {
-                        index,
-                        err: Some(e),
-                        ..Default::default()
-                    });
-                }
-            }
-            i += 1;
-        }
-
-        ress.sort_by(|a, b| {
-            let at = a.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
-            let bt = b.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
-
-            at.cmp(&bt)
-        });
-
-        for res in ress {
-            // check
-            if res.err.is_none() {
-                // TODO: let errs = self.poolsWithObject()
-                return Ok((res, Vec::new()));
-            }
-        }
-
-        let ret = PoolObjInfo::default();
-
-        Ok((ret, Vec::new()))
+        internal_get_pool_info_existing_with_opts(&self.pools, bucket, object, opts).await
     }
 }
 
+async fn internal_get_pool_info_existing_with_opts(
+    pools: &[Sets],
+    bucket: &str,
+    object: &str,
+    opts: &ObjectOptions,
+) -> Result<(PoolObjInfo, Vec<Error>)> {
+    let mut futures = Vec::new();
+
+    for pool in pools.iter() {
+        futures.push(pool.get_object_info(bucket, object, opts));
+    }
+
+    let results = join_all(futures).await;
+
+    let mut ress = Vec::new();
+
+    let mut i = 0;
+
+    // join_all结果跟输入顺序一致
+    for res in results {
+        let index = i;
+
+        match res {
+            Ok(r) => {
+                ress.push(PoolObjInfo {
+                    index,
+                    object_info: r,
+                    err: None,
+                });
+            }
+            Err(e) => {
+                ress.push(PoolObjInfo {
+                    index,
+                    err: Some(e),
+                    ..Default::default()
+                });
+            }
+        }
+        i += 1;
+    }
+
+    ress.sort_by(|a, b| {
+        let at = a.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
+        let bt = b.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
+
+        at.cmp(&bt)
+    });
+
+    for res in ress {
+        // check
+        if res.err.is_none() {
+            // TODO: let errs = self.poolsWithObject()
+            return Ok((res, Vec::new()));
+        }
+    }
+
+    let ret = PoolObjInfo::default();
+
+    Ok((ret, Vec::new()))
+}
+
 #[derive(Debug, Default)]
 pub struct PoolObjInfo {
     pub index: usize,
@@ -559,15 +571,29 @@ impl StorageAPI for ECStore {
             del_errs.push(None)
         }
 
-        // TODO: limte 限制并发数量
-        let opt = ObjectOptions::default();
-        // 取所有poolObjInfo
-        let mut futures = Vec::new();
-        for obj in objects.iter() {
-            futures.push(self.get_pool_info_existing_with_opts(bucket, &obj.object_name, &opt));
-        }
+        let mut jhs = Vec::new();
+        let semaphore = Arc::new(Semaphore::new(num_cpus::get()));
+        let pools = Arc::new(self.pools.clone());
 
-        let results = join_all(futures).await;
+        for obj in objects.iter() {
+            let (semaphore, pools, bucket, object_name, opt) = (
+                semaphore.clone(),
+                pools.clone(),
+                bucket.to_string(),
+                obj.object_name.to_string(),
+                ObjectOptions::default(),
+            );
+
+            let jh = tokio::spawn(async move {
+                let _permit = semaphore.acquire().await.unwrap();
+                internal_get_pool_info_existing_with_opts(pools.as_ref(), &bucket, &object_name, &opt).await
+            });
+            jhs.push(jh);
+        }
+        let mut results = Vec::new();
+        for jh in jhs {
+            results.push(jh.await.unwrap());
+        }
 
         // 记录pool Index 对应的objects pool_idx -> objects idx
         let mut pool_index_objects = HashMap::new();