forked from apache/doris
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathobj_storage_client.cpp
More file actions
154 lines (130 loc) · 5.85 KB
/
Copy pathobj_storage_client.cpp
File metadata and controls
154 lines (130 loc) · 5.85 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#include "recycler/obj_storage_client.h"
#include <chrono>
#include "common/config.h"
#include "cpp/sync_point.h"
#include "recycler/sync_executor.h"
using namespace std::chrono;
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;
size_t num_deleted = 0;
int error_count = 0;
size_t batch_count = 0;
auto start_time = steady_clock::now();
// Read max tasks per batch from config, validate to prevent overflow
int32_t config_val = config::recycler_max_tasks_per_batch;
size_t max_tasks_per_batch = 1000; // default value
if (config_val > 0) {
max_tasks_per_batch = static_cast<size_t>(config_val);
} else {
LOG(WARNING) << "recycler_max_tasks_per_batch=" << config_val
<< " is not positive, using default 1000";
}
while (true) {
// Create a new SyncExecutor for each batch
// Note: cancel lambda only takes effect within the current batch
SyncExecutor<int> batch_executor(
option.executor, fmt::format("delete batch under {}/{}", path.bucket, path.key),
[](const int& r) { return r != 0; });
std::vector<std::string> keys;
size_t tasks_in_batch = 0;
bool has_more = true;
// Collect tasks until reaching batch limit or no more files
while (tasks_in_batch < max_tasks_per_batch && has_more) {
auto obj = list_iter->next();
if (!obj.has_value()) {
has_more = false;
break;
}
if (expired_time > 0 && obj->mtime_s > expired_time) {
continue;
}
num_deleted++;
keys.emplace_back(std::move(obj->key));
// Submit a delete task when we have batch_size keys
if (keys.size() >= batch_size) {
batch_executor.add([this, &path, k = std::move(keys), option]() mutable {
return delete_objects(path.bucket, std::move(k), option).ret;
});
keys.clear();
tasks_in_batch++;
}
}
// Handle remaining keys (less than batch_size)
if (!keys.empty()) {
batch_executor.add([this, &path, k = std::move(keys), option]() mutable {
return delete_objects(path.bucket, std::move(k), option).ret;
});
tasks_in_batch++;
}
// Before exiting on empty batch, check if listing is valid
// Avoid silently treating listing failure as success
if (tasks_in_batch == 0) {
if (!list_iter->is_valid()) {
LOG(WARNING) << "list_iter invalid with no tasks collected";
ret = {-1};
}
break;
}
// Wait for current batch to complete
bool finished = true;
std::vector<int> rets = batch_executor.when_all(&finished);
batch_count++;
for (int r : rets) {
if (r != 0) {
error_count++;
}
}
// Log batch progress for monitoring long-running delete tasks
auto batch_elapsed = duration_cast<milliseconds>(steady_clock::now() - start_time).count();
LOG(INFO) << "delete objects under " << path.bucket << "/" << path.key << " batch "
<< batch_count << " completed"
<< ", tasks_in_batch=" << tasks_in_batch << ", total_deleted=" << num_deleted
<< ", elapsed=" << batch_elapsed << " ms";
// Check finished status: false means stop_token triggered, task timeout, or task invalid
if (!finished) {
LOG(WARNING) << "batch execution did not finish normally, stopping";
ret = {-1};
break;
}
// Check if list_iter is still valid (network errors, etc.)
if (!list_iter->is_valid()) {
LOG(WARNING) << "list_iter became invalid during iteration";
ret = {-1};
break;
}
// batch_executor goes out of scope, resources are automatically released
}
if (error_count > 0) {
LOG(WARNING) << "delete_objects_recursively completed with " << error_count << " errors";
ret = {-1};
}
auto elapsed = duration_cast<milliseconds>(steady_clock::now() - start_time).count();
LOG(INFO) << "delete objects under " << path.bucket << "/" << path.key
<< " finished, ret=" << ret.ret << ", total_batches=" << batch_count
<< ", num_deleted=" << num_deleted << ", error_count=" << error_count
<< ", cost=" << elapsed << " ms";
return ret;
}
} // namespace doris::cloud