diff --git a/ecstore/src/cache_value/metacache_set.rs b/ecstore/src/cache_value/metacache_set.rs index 2f719485d..6409bd78d 100644 --- a/ecstore/src/cache_value/metacache_set.rs +++ b/ecstore/src/cache_value/metacache_set.rs @@ -1,13 +1,9 @@ use std::{future::Future, pin::Pin, sync::Arc}; -use futures::{future::join_all, join}; +use futures::future::join_all; use tokio::{ spawn, - sync::{ - broadcast::Receiver as B_Receiver, - mpsc::{self}, - RwLock, - }, + sync::{broadcast::Receiver as B_Receiver, RwLock}, }; use tracing::error; @@ -17,7 +13,6 @@ use crate::{ DiskAPI, DiskStore, MetaCacheEntries, MetaCacheEntry, WalkDirOptions, }, error::{Error, Result}, - io::Writer, metacache::writer::MetacacheReader, }; @@ -306,7 +301,7 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - jobs.push(revjob); - let a = join_all(jobs).await; + let _ = join_all(jobs).await; Ok(()) } diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index a7d1ab0ac..2bf2d6093 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -157,6 +157,88 @@ impl SetDisks { disks } + + pub async fn get_online_disks_with_healing_and_info(&self, incl_healing: bool) -> (Vec, Vec, usize) { + let mut disks = self.get_disks_internal().await; + + let mut infos = Vec::with_capacity(disks.len()); + + let mut futures = Vec::with_capacity(disks.len()); + + let mut rng = thread_rng(); + disks.shuffle(&mut rng); + + let mut numbers: Vec = (0..disks.len()).collect(); + numbers.shuffle(&mut rand::thread_rng()); + + for &i in numbers.iter() { + let disk = disks[i].clone(); + futures.push(async move { + if let Some(disk) = disk { + disk.disk_info(&DiskInfoOptions::default()).await + } else { + Err(Error::new(DiskError::DiskNotFound)) + } + }); + } + + let results = join_all(futures).await; + for result in results { + match result { + Ok(res) => { + infos.push(res); + } + Err(err) => { + infos.push(DiskInfo { + error: err.to_string(), + ..Default::default() + }); + } + } + } + + let mut healing: usize = 0; + + let mut scanning_disks = Vec::new(); + let mut healing_disks = Vec::new(); + let mut scanning_infos = Vec::new(); + let mut healing_infos = Vec::new(); + + let mut new_disks = Vec::new(); + let mut new_infos = Vec::new(); + + for &i in numbers.iter() { + let (info, disk) = (infos[i].clone(), disks[i].clone()); + if !info.error.is_empty() || disk.is_none() { + continue; + } + + if info.healing { + healing += 1; + if incl_healing { + healing_disks.push(disk.unwrap()); + healing_infos.push(info); + } + + continue; + } + + if !info.healing { + new_disks.push(disk.unwrap()); + new_infos.push(info); + } else { + scanning_disks.push(disk.unwrap()); + scanning_infos.push(info); + } + } + + new_disks.extend(scanning_disks); + new_infos.extend(scanning_infos); + new_disks.extend(healing_disks); + new_infos.extend(healing_infos); + + (new_disks, new_infos, healing) + } async fn _get_local_disks(&self) -> Vec> { let mut disks = self.get_disks_internal().await; diff --git a/ecstore/src/store_list_objects.rs b/ecstore/src/store_list_objects.rs index fc7d47fe2..1e4851bf8 100644 --- a/ecstore/src/store_list_objects.rs +++ b/ecstore/src/store_list_objects.rs @@ -1,13 +1,20 @@ -use crate::disk::WalkDirOptions; +use crate::cache_value::metacache_set::{list_path_raw, ListPathRawOptions}; +use crate::disk::{DiskInfo, DiskStore, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, WalkDirOptions}; use crate::error::{Error, Result}; use crate::peer::is_reserved_or_invalid_bucket; +use crate::set_disk::SetDisks; use crate::store::check_list_objs_args; use crate::store_api::{ListObjectsInfo, ObjectInfo}; use crate::utils::path::{base_dir_from_prefix, SLASH_SEPARATOR}; use crate::{store::ECStore, store_api::ListObjectsV2Info}; use futures::future::join_all; -use std::collections::HashSet; +use rand::seq::SliceRandom; +use rand::thread_rng; +use std::collections::{HashMap, HashSet}; use std::io::ErrorKind; +use tokio::sync::broadcast::Receiver; +use tokio::sync::mpsc::Sender; +use tracing::error; const MAX_OBJECT_LIST: i32 = 1000; const MAX_DELETE_LIST: i32 = 1000; @@ -80,6 +87,8 @@ pub struct ListPathOptions { // Versioned is this a ListObjectVersions call. pub versioned: bool, + + pub stop_disk_at_limit: bool, } impl ListPathOptions { @@ -291,14 +300,169 @@ impl ECStore { } } +impl SetDisks { + pub async fn list_path(&self, rx: Receiver, opts: ListPathOptions, sender: Sender) -> Result<()> { + let (mut disks, infos, _) = self.get_online_disks_with_healing_and_info(true).await; + + let mut ask_disks = get_list_quorum(&opts.ask_disks, self.set_drive_count as i32); + if ask_disks == -1 { + let new_disks = get_quorum_disks(&disks, &infos, (disks.len() + 1) / 2); + if !new_disks.is_empty() { + disks = new_disks; + ask_disks = 1; + } else { + ask_disks = get_list_quorum("strict", self.set_drive_count as i32); + } + } + + if self.set_drive_count == 4 || ask_disks > disks.len() as i32 { + ask_disks = disks.len() as i32; + } + + let listing_quorum = ((ask_disks + 1) / 2) as usize; + + let mut fallback_disks = Vec::new(); + + if ask_disks > 0 && disks.len() > ask_disks as usize { + let mut rand = thread_rng(); + disks.shuffle(&mut rand); + + fallback_disks = disks.split_off(ask_disks as usize); + } + + let mut resolver = MetadataResolutionParams { + dir_quorum: listing_quorum, + obj_quorum: listing_quorum, + bucket: opts.bucket.clone(), + ..Default::default() + }; + + if opts.versioned { + resolver.requested_versions = 1; + } + + let limit = { + if opts.limit > 0 && opts.stop_disk_at_limit { + opts.limit + 4 + (opts.limit / 16) + } else { + 0 + } + }; + + let tx1 = sender.clone(); + let tx2 = sender.clone(); + + list_path_raw( + rx, + ListPathRawOptions { + disks: disks.iter().cloned().map(Some).collect(), + fallback_disks: fallback_disks.iter().cloned().map(Some).collect(), + bucket: opts.bucket, + path: opts.base_dir, + recursice: opts.recursive, + filter_prefix: opts.filter_prefix, + forward_to: opts.marker, + min_disks: listing_quorum, + per_disk_limit: limit, + agreed: Some(Box::new(move |entry: MetaCacheEntry| { + Box::pin({ + let value = tx1.clone(); + async move { + if let Err(err) = value.send(entry).await { + error!("list_path send fail {:?}", err); + } + } + }) + })), + partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { + Box::pin({ + let value = tx2.clone(); + let resolver = resolver.clone(); + async move { + if let Ok(Some(entry)) = entries.resolve(resolver) { + if let Err(err) = value.send(entry).await { + error!("list_path send fail {:?}", err); + } + } + } + }) + })), + finished: None, + ..Default::default() + }, + ) + .await + } +} + +fn get_list_quorum(quorum: &str, drive_count: i32) -> i32 { + match quorum { + "disk" => 1, + "reduced" => 2, + "optimal" => (drive_count + 1) / 2, + "auto" => -1, + _ => drive_count, // defaults to 'strict' + } +} + +fn get_quorum_disk_infos(disks: &[DiskStore], infos: &[DiskInfo], read_quorum: usize) -> (Vec, Vec) { + let common_mutations = calc_common_counter(infos, read_quorum); + let mut new_disks = Vec::new(); + let mut new_infos = Vec::new(); + + for (i, info) in infos.iter().enumerate() { + let mutations = info.metrics.total_deletes + info.metrics.total_writes; + if mutations >= common_mutations { + new_disks.push(disks[i].clone()); // Assuming StorageAPI derives Clone + new_infos.push(infos[i].clone()); // Assuming DiskInfo derives Clone + } + } + + (new_disks, new_infos) +} + +fn get_quorum_disks(disks: &[DiskStore], infos: &[DiskInfo], read_quorum: usize) -> Vec { + let (new_disks, _) = get_quorum_disk_infos(disks, infos, read_quorum); + new_disks +} + +fn calc_common_counter(infos: &[DiskInfo], read_quorum: usize) -> u64 { + let mut max = 0; + let mut common_count = 0; + let mut signature_map: HashMap = HashMap::new(); + + for info in infos { + if !info.error.is_empty() { + continue; + } + let mutations = info.metrics.total_deletes + info.metrics.total_writes; + *signature_map.entry(mutations).or_insert(0) += 1; + } + + for (&ops, &count) in &signature_map { + if max < count && common_count < ops { + max = count; + common_count = ops; + } + } + + if max < read_quorum { + return 0; + } + common_count +} + // list_path_raw #[cfg(test)] mod test { + use std::sync::Arc; + use crate::cache_value::metacache_set::list_path_raw; use crate::cache_value::metacache_set::ListPathRawOptions; use crate::disk::endpoint::Endpoint; use crate::disk::error::is_err_eof; + use crate::disk::format::FormatV3; use crate::disk::new_disk; use crate::disk::DiskAPI; use crate::disk::DiskOption; @@ -307,8 +471,14 @@ mod test { use crate::disk::WalkDirOptions; use crate::error::Error; use crate::metacache::writer::MetacacheReader; + use crate::set_disk::SetDisks; + use crate::store_list_objects::ListPathOptions; use futures::future::join_all; + use lock::namespace_lock::NsLockMap; use tokio::sync::broadcast; + use tokio::sync::mpsc; + use tokio::sync::RwLock; + use uuid::Uuid; #[tokio::test] async fn test_walk_dir() { @@ -418,4 +588,46 @@ mod test { .await .unwrap(); } + + #[tokio::test] + async fn test_list_path() { + let mut ep = Endpoint::try_from("/Users/weisd/project/weisd/s3-rustfs/target/volume/test").unwrap(); + ep.pool_idx = 0; + ep.set_idx = 0; + ep.disk_idx = 0; + ep.is_local = true; + + let disk = new_disk(&ep, &DiskOption::default()).await.expect("init disk fail"); + let _ = disk.set_disk_id(Some(Uuid::new_v4())).await; + + let set = SetDisks { + lockers: Vec::new(), + locker_owner: String::new(), + ns_mutex: Arc::new(RwLock::new(NsLockMap::new(false))), + disks: RwLock::new(vec![Some(disk)]), + set_endpoints: Vec::new(), + set_drive_count: 1, + default_parity_count: 0, + set_index: 0, + pool_index: 0, + format: FormatV3::new(1, 1), + }; + + let (_, rx) = broadcast::channel(1); + let bucket = "dada".to_owned(); + + let opts = ListPathOptions { + bucket, + recursive: true, + ..Default::default() + }; + + let (sender, mut recv) = mpsc::channel(10); + + set.list_path(rx, opts, sender).await.unwrap(); + + while let Some(entry) = recv.recv().await { + println!("get entry {:?}", entry.name) + } + } }