review list_objects

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<Self>,
         bucket: &str,
@@ -1524,27 +1527,6 @@ impl StorageAPI for ECStore {
     ) -> Result<ListObjectsV2Info> {
         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<Self>,

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::<StorageError>() {
+        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;

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_api::{ListObjectVersionsInfo, ListObjectsInfo, ObjectInfo, ObjectOptions};
-use crate::store_err::StorageError;
+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<Self>,
         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:
+        // cancel channel
-        let (tx, rx) = broadcast::channel(1);
+        let (cancel_tx, cancel_rx) = broadcast::channel(1);
+        let (err_tx, mut err_rx) = broadcast::channel::<Error>(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();
+        let cancel_rx1 = cancel_rx.resubscribe();
-        tokio::spawn(async move {
+        let err_tx1 = err_tx.clone();
-            if let Err(err) = store.list_merged(rx1, opts, sender).await {
+        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 cancel_rx2 = cancel_rx.resubscribe();
-        let res = gather_results(rx2, &o, &mut recv).await?;
 
-        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 {
+        let result = {
-            o: MetaCacheEntries(res.into_iter().map(Some).collect()),
+            // 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() {
+        // cancel call exit spawns
-        //     opts.base_dir = base_dir_from_prefix(&opts.prefix);
+        cancel_tx.send(true)?;
-        // }
 
-        // let objects = self.list_merged(&opts).await?;
+        // wait spawns exit
+        join_all(vec![job1, job2]).await;
 
-        // let info = ListObjectsInfo {
+        result
-        //     objects,
-        //     ..Default::default()
-        // };
-        // Ok(info)
     }
 
     // 读所有
@@ -486,16 +534,21 @@ impl ECStore {
     }
 }
 
-// TODO: FIXME: 异步
 async fn gather_results(
     _rx: B_Receiver<bool>,
-    opts: &ListPathOptions,
+    opts: ListPathOptions,
-    recv: &mut Receiver<MetaCacheEntry>,
+    recv: Receiver<MetaCacheEntry>,
-) -> Result<Vec<MetaCacheEntry>> {
+    results_tx: Sender<Result<MetaCacheEntriesSorted>>,
+) {
     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);
+    results_tx.send(Ok(results)).await;
-    Ok(results)
 }
 
 async fn select_from(

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)
+            }
         };
     }