diff --git a/Cargo.lock b/Cargo.lock index 22b19bba1..b8c76b3e2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -447,6 +447,7 @@ dependencies = [ "http", "lazy_static", "netif", + "num_cpus", "openssl", "path-absolutize", "path-clean", diff --git a/ecstore/Cargo.toml b/ecstore/Cargo.toml index ada0fc7a5..75f503ef4 100644 --- a/ecstore/Cargo.toml +++ b/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" diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index 8698b2103..9bbe781a9 100644 --- a/ecstore/src/sets.rs +++ b/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, diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 44473c3b0..74a4f0c45 100644 --- a/ecstore/src/store.rs +++ b/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)> { - 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)> { + 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();