From 938df62167303432da6da6d996003f431f2fb7ef Mon Sep 17 00:00:00 2001 From: weisd Date: Tue, 24 Dec 2024 17:01:37 +0800 Subject: [PATCH] review list_objects --- ecstore/src/store.rs | 26 +------ ecstore/src/store_err.rs | 11 +++ ecstore/src/store_list_objects.rs | 125 +++++++++++++++++++++--------- rustfs/src/storage/error.rs | 3 + 4 files changed, 107 insertions(+), 58 deletions(-) diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 0aa548832..1d84e39c0 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -1511,7 +1511,10 @@ impl StorageAPI for ECStore { Err(Error::new(StorageError::ObjectNotFound(bucket.to_owned(), object.to_owned()))) } - // TODO: review + // @continuation_token marker + // @start_after as marker when continuation_token empty + // @delimiter default="/", empty when recursive + // @max_keys limit async fn list_objects_v2( self: Arc, bucket: &str, @@ -1524,27 +1527,6 @@ impl StorageAPI for ECStore { ) -> Result { self.inner_list_objects_v2(bucket, prefix, continuation_token, delimiter, max_keys, fetch_owner, start_after) .await - - // let opts = ListPathOptions { - // bucket: bucket.to_string(), - // limit: max_keys, - // prefix: prefix.to_owned(), - // ..Default::default() - // }; - - // let info = self.list_path(&opts, delimiter).await?; - - // // warn!("list_objects_v2 info {:?}", info); - - // let v2 = ListObjectsV2Info { - // is_truncated: info.is_truncated, - // continuation_token: continuation_token.to_owned(), - // next_continuation_token: info.next_marker, - // objects: info.objects, - // prefixes: info.prefixes, - // }; - - // Ok(v2) } async fn list_object_versions( self: Arc, diff --git a/ecstore/src/store_err.rs b/ecstore/src/store_err.rs index e0dabb870..a0fc12cc4 100644 --- a/ecstore/src/store_err.rs +++ b/ecstore/src/store_err.rs @@ -52,6 +52,9 @@ pub enum StorageError { #[error("Object not found: {0}/{1}")] ObjectNotFound(String, String), + #[error("volume not found: {0}")] + VolumeNotFound(String), + #[error("Version not found: {0}/{1}-{2}")] VersionNotFound(String, String, String), @@ -193,6 +196,14 @@ pub fn is_err_bucket_exists(err: &Error) -> bool { } } +pub fn is_err_bucket_not_found(err: &Error) -> bool { + if let Some(e) = err.downcast_ref::() { + matches!(e, StorageError::VolumeNotFound(_)) || matches!(e, StorageError::BucketNotFound(_)) + } else { + false + } +} + pub fn is_err_object_not_found(err: &Error) -> bool { if is_err_file_not_found(err) { return true; diff --git a/ecstore/src/store_list_objects.rs b/ecstore/src/store_list_objects.rs index 02c3e2ed8..95c2dd9d5 100644 --- a/ecstore/src/store_list_objects.rs +++ b/ecstore/src/store_list_objects.rs @@ -6,9 +6,10 @@ use crate::file_meta::merge_file_meta_versions; use crate::peer::is_reserved_or_invalid_bucket; use crate::set_disk::SetDisks; use crate::store::check_list_objs_args; -use crate::store_api::{ListObjectVersionsInfo, ListObjectsInfo, ObjectInfo}; -use crate::store_err::StorageError; +use crate::store_api::{ListObjectVersionsInfo, ListObjectsInfo, ObjectInfo, ObjectOptions}; +use crate::store_err::{is_err_bucket_exists, is_err_bucket_not_found, StorageError}; use crate::utils::path::{self, base_dir_from_prefix, SLASH_SEPARATOR}; +use crate::StorageAPI; use crate::{store::ECStore, store_api::ListObjectsV2Info}; use futures::future::join_all; use rand::seq::SliceRandom; @@ -29,7 +30,7 @@ const METACACHE_SHARE_PREFIX: bool = false; pub fn max_keys_plus_one(max_keys: i32, add_one: bool) -> i32 { let mut max_keys = max_keys; - if max_keys > MAX_OBJECT_LIST { + if !(0..=MAX_OBJECT_LIST).contains(&max_keys) { max_keys = MAX_OBJECT_LIST; } if add_one { @@ -115,6 +116,10 @@ impl ListPathOptions { impl ECStore { #[allow(clippy::too_many_arguments)] + // @continuation_token marker + // @start_after as marker when continuation_token empty + // @delimiter default="/", empty when recursive + // @max_keys limit pub async fn inner_list_objects_v2( self: Arc, bucket: &str, @@ -162,6 +167,33 @@ impl ECStore { ..Default::default() }; + // use get + if !opts.prefix.is_empty() && opts.limit == 1 && opts.marker.is_empty() { + match self + .get_object_info( + &opts.bucket, + &opts.prefix, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + { + Ok(res) => { + return Ok(ListObjectsInfo { + objects: vec![res], + ..Default::default() + }); + } + Err(err) => { + if is_err_bucket_not_found(&err) { + return Err(err); + } + } + }; + }; + let mut err_eof = false; let has_merged = match self.list_path(&opts).await { Ok(res) => Some(res), @@ -367,44 +399,60 @@ impl ECStore { o.create = false; } - // FIXME:TODO: - let (tx, rx) = broadcast::channel(1); + // cancel channel + let (cancel_tx, cancel_rx) = broadcast::channel(1); + let (err_tx, mut err_rx) = broadcast::channel::(1); - let (sender, mut recv) = mpsc::channel(o.limit as usize); + let (sender, recv) = mpsc::channel(o.limit as usize); let store = self.clone(); let opts = o.clone(); - let rx1 = rx.resubscribe(); - tokio::spawn(async move { - if let Err(err) = store.list_merged(rx1, opts, sender).await { + let cancel_rx1 = cancel_rx.resubscribe(); + let err_tx1 = err_tx.clone(); + let job1 = tokio::spawn(async move { + let mut opts = opts; + opts.stop_disk_at_limit = true; + if let Err(err) = store.list_merged(cancel_rx1, opts, sender).await { error!("list_merged err {:?}", err); + let _ = err_tx1.send(err); } }); - let rx2 = rx.resubscribe(); - let res = gather_results(rx2, &o, &mut recv).await?; + let cancel_rx2 = cancel_rx.resubscribe(); - tx.send(true)?; + let (result_tx, mut result_rx) = mpsc::channel(1); - // TODO: recv list_merged err + let job2 = tokio::spawn(gather_results(cancel_rx2, o, recv, result_tx)); - Ok(MetaCacheEntriesSorted { - o: MetaCacheEntries(res.into_iter().map(Some).collect()), - }) + let result = { + // receiver result + tokio::select! { + res = err_rx.recv() =>{ - // let mut opts = opts.clone(); + match res{ + Ok(o) => { + error!("list_path err_rx.recv() ok {:?}", &o); + Err(o) + }, + Err(err) => { + error!("list_path err_rx.recv() err {:?}", &err); + Err(Error::new(err)) + }, + } + }, + Some(result) = result_rx.recv()=>{ + result + } + } + }; - // if opts.base_dir.is_empty() { - // opts.base_dir = base_dir_from_prefix(&opts.prefix); - // } + // cancel call exit spawns + cancel_tx.send(true)?; - // let objects = self.list_merged(&opts).await?; + // wait spawns exit + join_all(vec![job1, job2]).await; - // let info = ListObjectsInfo { - // objects, - // ..Default::default() - // }; - // Ok(info) + result } // 读所有 @@ -486,16 +534,21 @@ impl ECStore { } } -// TODO: FIXME: 异步 async fn gather_results( _rx: B_Receiver, - opts: &ListPathOptions, - recv: &mut Receiver, -) -> Result> { + opts: ListPathOptions, + recv: Receiver, + results_tx: Sender>, +) { let mut returned = false; - let mut results = Vec::new(); + let mut results = MetaCacheEntriesSorted { + o: MetaCacheEntries(Vec::new()), + }; + + let mut recv = recv; + // let mut entrys = Vec::new(); while let Some(mut entry) = recv.recv().await { - // warn!("gather_results entry {}", &entry.name); + // warn!("gather_entrys entry {}", &entry.name); if returned { continue; } @@ -514,16 +567,16 @@ async fn gather_results( } // TODO: other - if opts.limit > 0 && results.len() >= opts.limit as usize { + if opts.limit > 0 && results.o.0.len() >= opts.limit as usize { returned = true; continue; } - results.push(entry); + results.o.0.push(Some(entry)); + // entrys.push(entry); } - // warn!("gather_results results {:?}", &results); - Ok(results) + results_tx.send(Ok(results)).await; } async fn select_from( diff --git a/rustfs/src/storage/error.rs b/rustfs/src/storage/error.rs index 53aad658c..72873dd6f 100644 --- a/rustfs/src/storage/error.rs +++ b/rustfs/src/storage/error.rs @@ -62,6 +62,9 @@ pub fn to_s3_error(err: Error) -> S3Error { s3_error!(SlowDown, "Storage resources are insufficient for the write operation") } StorageError::DecommissionNotStarted => s3_error!(InvalidArgument, "Decommission Not Started"), + StorageError::VolumeNotFound(bucket) => { + s3_error!(NoSuchBucket, "bucket not found {}", bucket) + } }; }