Skip to content

Commit 81f9ad8

Browse files
SWJTU-ZhangLeiSWJTU-ZhangLei
committed
[fix](be/ry) Fix S3RateLimter bvar rate_limit_exceed_req_num incorrectly
* Fix S3RateLimter bvar `rate_limit_exceed_req_num` incorrectly * Add more ut for S3RateLimter
1 parent 03962f2 commit 81f9ad8

File tree

5 files changed

+73
-42
lines changed

5 files changed

+73
-42
lines changed

‎be/src/util/s3_util.cpp

+4-14
Original file line numberDiff line numberDiff line change
@@ -105,16 +105,6 @@ constexpr char S3_REQUEST_TIMEOUT_MS[] = "AWS_REQUEST_TIMEOUT_MS";
105105
constexpr char S3_CONN_TIMEOUT_MS[] = "AWS_CONNECTION_TIMEOUT_MS";
106106
constexpr char S3_NEED_OVERRIDE_ENDPOINT[] = "AWS_NEED_OVERRIDE_ENDPOINT";
107107

108-
auto metric_func_factory(bvar::Adder<int64_t>& ns_bvar, bvar::Adder<int64_t>& req_num_bvar) {
109-
return [&](int64_t ns) {
110-
if (ns > 0) {
111-
ns_bvar << ns;
112-
} else {
113-
req_num_bvar << 1;
114-
}
115-
};
116-
}
117-
118108
} // namespace
119109

120110
bvar::Adder<int64_t> get_rate_limit_ns("get_rate_limit_ns");
@@ -193,12 +183,12 @@ S3ClientFactory::S3ClientFactory() {
193183
_ca_cert_file_path = get_valid_ca_cert_path();
194184
_rate_limiters = {
195185
std::make_unique<S3RateLimiterHolder>(
196-
S3RateLimitType::GET, config::s3_get_token_per_second,
197-
config::s3_get_bucket_tokens, config::s3_get_token_limit,
186+
config::s3_get_token_per_second, config::s3_get_bucket_tokens,
187+
config::s3_get_token_limit,
198188
metric_func_factory(get_rate_limit_ns, get_rate_limit_exceed_req_num)),
199189
std::make_unique<S3RateLimiterHolder>(
200-
S3RateLimitType::PUT, config::s3_put_token_per_second,
201-
config::s3_put_bucket_tokens, config::s3_put_token_limit,
190+
config::s3_put_token_per_second, config::s3_put_bucket_tokens,
191+
config::s3_put_token_limit,
202192
metric_func_factory(put_rate_limit_ns, put_rate_limit_exceed_req_num))};
203193
}
204194

‎cloud/src/recycler/s3_accessor.cpp

+4-17
Original file line numberDiff line numberDiff line change
@@ -49,20 +49,7 @@
4949
#include "recycler/s3_obj_client.h"
5050
#include "recycler/storage_vault_accessor.h"
5151

52-
namespace {
53-
auto metric_func_factory(bvar::Adder<int64_t>& ns_bvar, bvar::Adder<int64_t>& req_num_bvar) {
54-
return [&](int64_t ns) {
55-
if (ns > 0) {
56-
ns_bvar << ns;
57-
} else {
58-
req_num_bvar << 1;
59-
}
60-
};
61-
}
62-
} // namespace
63-
6452
namespace doris::cloud {
65-
6653
namespace s3_bvar {
6754
bvar::LatencyRecorder s3_get_latency("s3_get");
6855
bvar::LatencyRecorder s3_put_latency("s3_put");
@@ -84,12 +71,12 @@ bvar::Adder<int64_t> put_rate_limit_exceed_req_num("put_rate_limit_exceed_req_nu
8471
AccessorRateLimiter::AccessorRateLimiter()
8572
: _rate_limiters(
8673
{std::make_unique<S3RateLimiterHolder>(
87-
S3RateLimitType::GET, config::s3_get_token_per_second,
88-
config::s3_get_bucket_tokens, config::s3_get_token_limit,
74+
config::s3_get_token_per_second, config::s3_get_bucket_tokens,
75+
config::s3_get_token_limit,
8976
metric_func_factory(get_rate_limit_ns, get_rate_limit_exceed_req_num)),
9077
std::make_unique<S3RateLimiterHolder>(
91-
S3RateLimitType::PUT, config::s3_put_token_per_second,
92-
config::s3_put_bucket_tokens, config::s3_put_token_limit,
78+
config::s3_put_token_per_second, config::s3_put_bucket_tokens,
79+
config::s3_put_token_limit,
9380
metric_func_factory(put_rate_limit_ns,
9481
put_rate_limit_exceed_req_num))}) {}
9582

‎cloud/test/s3_rate_limiter_test.cpp

+50-5
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
// specific language governing permissions and limitations
1616
// under the License.
1717

18+
#include <bvar/bvar.h>
1819
#include <cpp/s3_rate_limiter.h>
1920
#include <gtest/gtest.h>
2021

@@ -43,15 +44,15 @@ int main(int argc, char** argv) {
4344
return RUN_ALL_TESTS();
4445
}
4546

46-
#define S3RateLimiterTest DISABLED_S3RateLimiterTest
47+
// #define S3RateLimiterTest DISABLED_S3RateLimiterTest
4748
TEST(S3RateLimiterTest, normal) {
4849
auto rate_limiter = doris::S3RateLimiter(1, 5, 10);
4950
std::atomic_int64_t failed;
5051
std::atomic_int64_t succ;
5152
std::atomic_int64_t sleep_thread_num;
5253
std::atomic_int64_t sleep;
5354
auto request_thread = [&]() {
54-
std::this_thread::sleep_for(std::chrono::milliseconds(500));
55+
std::this_thread::sleep_for(std::chrono::milliseconds(100));
5556
auto ms = rate_limiter.add(1);
5657
if (ms < 0) {
5758
failed++;
@@ -73,7 +74,51 @@ TEST(S3RateLimiterTest, normal) {
7374
}
7475
}
7576
}
76-
ASSERT_EQ(failed, 10);
77-
ASSERT_EQ(succ, 6);
78-
ASSERT_EQ(sleep_thread_num, 4);
77+
EXPECT_EQ(failed, 10);
78+
EXPECT_EQ(succ, 5);
79+
EXPECT_EQ(sleep_thread_num, 5);
80+
}
81+
82+
TEST(S3RateLimiterTest, BasicRateLimit) {
83+
auto rate_limiter = doris::S3RateLimiter(100, 100, 200);
84+
int64_t sleep_time = rate_limiter.add(50);
85+
EXPECT_EQ(sleep_time, 0);
86+
87+
sleep_time = rate_limiter.add(60);
88+
EXPECT_GT(sleep_time, 0);
89+
}
90+
91+
TEST(S3RateLimiterTest, BurstCapacity) {
92+
auto rate_limiter = doris::S3RateLimiter(125, 250, 1000);
93+
int64_t sleep_time = rate_limiter.add(250);
94+
EXPECT_EQ(sleep_time, 0);
95+
96+
sleep_time = rate_limiter.add(1);
97+
EXPECT_GT(sleep_time, 0);
98+
}
99+
100+
TEST(S3RateLimiterTest, ExceedLimit) {
101+
auto rate_limiter = doris::S3RateLimiter(125, 250, 250);
102+
int64_t sleep_time = rate_limiter.add(250);
103+
EXPECT_EQ(sleep_time, 0);
104+
105+
sleep_time = rate_limiter.add(1);
106+
EXPECT_EQ(sleep_time, -1);
107+
}
108+
109+
TEST(S3RateLimiterHolderTest, BvarMetric) {
110+
bvar::Adder<int64_t> rate_limit_ns("rate_limit_ns");
111+
bvar::Adder<int64_t> rate_limit_exceed_req_num("rate_limit_exceed_req_num");
112+
113+
auto rate_limiter_holder = doris::S3RateLimiterHolder(
114+
125, 250, 500, doris::metric_func_factory(rate_limit_ns, rate_limit_exceed_req_num));
115+
int64_t sleep_time = rate_limiter_holder.add(250);
116+
EXPECT_EQ(sleep_time, 0);
117+
EXPECT_EQ(rate_limit_ns.get_value(), 0);
118+
EXPECT_EQ(rate_limit_exceed_req_num.get_value(), 0);
119+
120+
sleep_time = rate_limiter_holder.add(1);
121+
EXPECT_GT(rate_limit_ns.get_value(), 0);
122+
EXPECT_EQ(rate_limit_exceed_req_num.get_value(), 1);
123+
EXPECT_GT(sleep_time, 0);
79124
}

‎common/cpp/s3_rate_limiter.cpp

+4-5
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@
2626
#if defined(__APPLE__)
2727
#include <ctime>
2828
#endif
29-
#define CURRENT_TIME std::chrono::system_clock::now()
3029

3130
namespace doris {
3231
// Just 10^6.
@@ -91,8 +90,8 @@ std::pair<size_t, double> S3RateLimiter::_update_remain_token(long now, size_t a
9190

9291
int64_t S3RateLimiter::add(size_t amount) {
9392
// Values obtained under lock to be checked after release
94-
auto time = CURRENT_TIME;
95-
auto time_nano_count = time.time_since_epoch().count();
93+
auto duration = std::chrono::system_clock::now().time_since_epoch();
94+
auto time_nano_count = std::chrono::duration_cast<std::chrono::nanoseconds>(duration).count();
9695
auto [count_value, tokens_value] = _update_remain_token(time_nano_count, amount);
9796

9897
if (_limit && count_value > _limit) {
@@ -110,8 +109,8 @@ int64_t S3RateLimiter::add(size_t amount) {
110109
return sleep_time_ns;
111110
}
112111

113-
S3RateLimiterHolder::S3RateLimiterHolder(S3RateLimitType type, size_t max_speed, size_t max_burst,
114-
size_t limit, std::function<void(int64_t)> metric_func)
112+
S3RateLimiterHolder::S3RateLimiterHolder(size_t max_speed, size_t max_burst, size_t limit,
113+
std::function<void(int64_t)> metric_func)
115114
: rate_limiter(std::make_unique<S3RateLimiter>(max_speed, max_burst, limit)),
116115
metric_func(std::move(metric_func)) {}
117116

‎common/cpp/s3_rate_limiter.h

+11-1
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
// under the License.
1717

1818
#pragma once
19+
#include <bvar/bvar.h>
1920

2021
#include <functional>
2122
#include <memory>
@@ -31,6 +32,15 @@ enum class S3RateLimitType : int {
3132
extern std::string to_string(S3RateLimitType type);
3233
extern S3RateLimitType string_to_s3_rate_limit_type(std::string_view value);
3334

35+
inline auto metric_func_factory(bvar::Adder<int64_t>& ns_bvar, bvar::Adder<int64_t>& req_num_bvar) {
36+
return [&](int64_t ns) {
37+
if (ns > 0) {
38+
ns_bvar << ns;
39+
req_num_bvar << 1;
40+
}
41+
};
42+
}
43+
3444
class S3RateLimiter {
3545
public:
3646
static constexpr size_t default_burst_seconds = 1;
@@ -57,7 +67,7 @@ class S3RateLimiter {
5767

5868
class S3RateLimiterHolder {
5969
public:
60-
S3RateLimiterHolder(S3RateLimitType type, size_t max_speed, size_t max_burst, size_t limit,
70+
S3RateLimiterHolder(size_t max_speed, size_t max_burst, size_t limit,
6171
std::function<void(int64_t)> metric_func);
6272
~S3RateLimiterHolder();
6373

0 commit comments

Comments
 (0)