add util test

use atomic

use value

cloud/src/common/config.h

@@ -62,6 +62,8 @@ CONF_Strings(recycle_whitelist, ""); // Comma seprated list
 CONF_Strings(recycle_blacklist, ""); // Comma seprated list
 CONF_mInt32(instance_recycler_worker_pool_size, "1");
 CONF_Bool(enable_checker, "false");
+// The parallelism for parallel recycle operation
+CONF_Int32(recycle_pool_parallelism, "10");
 // Currently only used for recycler test
 CONF_Bool(enable_inverted_check, "false");
 // interval for scanning instances to do checks and inspections

cloud/src/recycler/azure_obj_client.cpp

@@ -209,7 +209,8 @@ std::unique_ptr<ObjectListIterator> AzureObjClient::list_objects(ObjectStoragePa
 // You can find out the num in https://learn.microsoft.com/en-us/rest/api/storageservices/blob-batch?tabs=microsoft-entra-id
 // > Each batch request supports a maximum of 256 subrequests.
 ObjectStorageResponse AzureObjClient::delete_objects(const std::string& bucket,
-                                                     std::vector<std::string> keys) {
+                                                     std::vector<std::string> keys,
+                                                     ObjClientOptions option) {
     if (keys.empty()) {
         return {0};
     }
@@ -275,8 +276,9 @@ ObjectStorageResponse AzureObjClient::delete_object(ObjectStoragePathRef path) {
 }
 
 ObjectStorageResponse AzureObjClient::delete_objects_recursively(ObjectStoragePathRef path,
+                                                                 ObjClientOptions option,
                                                                  int64_t expiration_time) {
-    return delete_objects_recursively_(path, expiration_time, BlobBatchMaxOperations);
+    return delete_objects_recursively_(path, option, expiration_time, BlobBatchMaxOperations);
 }
 
 ObjectStorageResponse AzureObjClient::get_life_cycle(const std::string& bucket,

cloud/src/recycler/azure_obj_client.h

@@ -38,12 +38,13 @@ class AzureObjClient final : public ObjStorageClient {
 
     std::unique_ptr<ObjectListIterator> list_objects(ObjectStoragePathRef path) override;
 
-    ObjectStorageResponse delete_objects(const std::string& bucket,
-                                         std::vector<std::string> keys) override;
+    ObjectStorageResponse delete_objects(const std::string& bucket, std::vector<std::string> keys,
+                                         ObjClientOptions option) override;
 
     ObjectStorageResponse delete_object(ObjectStoragePathRef path) override;
 
     ObjectStorageResponse delete_objects_recursively(ObjectStoragePathRef path,
+                                                     ObjClientOptions option,
                                                      int64_t expiration_time = 0) override;
 
     ObjectStorageResponse get_life_cycle(const std::string& bucket,

cloud/src/recycler/obj_storage_client.cpp

@@ -18,17 +18,24 @@
 #include "recycler/obj_storage_client.h"
 
 #include "cpp/sync_point.h"
+#include "recycler/sync_executor.h"
+#include "recycler/util.h"
 
 namespace doris::cloud {
 
 ObjectStorageResponse ObjStorageClient::delete_objects_recursively_(ObjectStoragePathRef path,
+                                                                    const ObjClientOptions& option,
                                                                     int64_t expired_time,
                                                                     size_t batch_size) {
     TEST_SYNC_POINT_CALLBACK("ObjStorageClient::delete_objects_recursively_", &batch_size);
     auto list_iter = list_objects(path);
 
     ObjectStorageResponse ret;
     std::vector<std::string> keys;
+    SyncExecutor<int> concurrent_delete_executor(
+            option.executor,
+            fmt::format("delete objects under bucket {}, path {}", path.bucket, path.key),
+            [](const int& ret) { return ret != 0; });
 
     for (auto obj = list_iter->next(); obj.has_value(); obj = list_iter->next()) {
         if (expired_time > 0 && obj->mtime_s > expired_time) {
@@ -39,20 +46,31 @@ ObjectStorageResponse ObjStorageClient::delete_objects_recursively_(ObjectStorag
         if (keys.size() < batch_size) {
             continue;
         }
-
-        ret = delete_objects(path.bucket, std::move(keys));
-        if (ret.ret != 0) {
-            return ret;
-        }
+        concurrent_delete_executor.add([this, &path, k = std::move(keys), option]() mutable {
+            return delete_objects(path.bucket, std::move(k), option).ret;
+        });
     }
 
     if (!list_iter->is_valid()) {
+        bool finished;
+        concurrent_delete_executor.when_all(&finished);
         return {-1};
     }
 
     if (!keys.empty()) {
-        return delete_objects(path.bucket, std::move(keys));
+        concurrent_delete_executor.add([this, &path, k = std::move(keys), option]() mutable {
+            return delete_objects(path.bucket, std::move(k), option).ret;
+        });
     }
+    bool finished = true;
+    std::vector<int> rets = concurrent_delete_executor.when_all(&finished);
+    for (int r : rets) {
+        if (r != 0) {
+            ret = -1;
+        }
+    }
+
+    ret = finished ? ret : -1;
 
     return ret;
 }

cloud/src/recycler/obj_storage_client.h

@@ -51,6 +51,12 @@ class ObjectListIterator {
     virtual std::optional<ObjectMeta> next() = 0;
 };
 
+class SimpleThreadPool;
+struct ObjClientOptions {
+    bool prefetch {true};
+    std::shared_ptr<SimpleThreadPool> executor;
+};
+
 class ObjStorageClient {
 public:
     ObjStorageClient() = default;
@@ -71,14 +77,16 @@ class ObjStorageClient {
 
     // According to the bucket and prefix specified by the user, it performs batch deletion based on the object names in the object array.
     virtual ObjectStorageResponse delete_objects(const std::string& bucket,
-                                                 std::vector<std::string> keys) = 0;
+                                                 std::vector<std::string> keys,
+                                                 ObjClientOptions option) = 0;
 
     // Delete the file named key in the object storage bucket.
     virtual ObjectStorageResponse delete_object(ObjectStoragePathRef path) = 0;
 
     // According to the prefix, recursively delete all objects under the prefix.
     // If `expiration_time` > 0, only delete objects with mtime earlier than `expiration_time`.
     virtual ObjectStorageResponse delete_objects_recursively(ObjectStoragePathRef path,
+                                                             ObjClientOptions option,
                                                              int64_t expiration_time = 0) = 0;
 
     // Get the objects' expiration time on the bucket
@@ -91,6 +99,7 @@ class ObjStorageClient {
 
 protected:
     ObjectStorageResponse delete_objects_recursively_(ObjectStoragePathRef path,
+                                                      const ObjClientOptions& option,
                                                       int64_t expiration_time, size_t batch_size);
 };