sollhui commented on code in PR #65420:
URL: https://github.com/apache/doris/pull/65420#discussion_r3596869021
##########
be/src/io/fs/s3_obj_storage_client.cpp:
##########
@@ -78,26 +78,36 @@ inline ::Aws::Client::AWSError<::Aws::S3::S3Errors>
s3_error_factory() {
}
template <typename Func>
-auto s3_rate_limit(doris::S3RateLimitType op, Func callback) ->
decltype(callback()) {
+auto s3_rate_limit(bool is_internal_bucket, doris::S3RateLimitType op, Func
callback)
+ -> decltype(callback()) {
using T = decltype(callback());
- if (!doris::config::enable_s3_rate_limiter) {
- return callback();
- }
- auto sleep_duration = doris::apply_s3_rate_limit(op);
- if (sleep_duration < 0) {
+ if (!doris::acquire_s3_qps_rate_limit(is_internal_bucket, op)) {
return T(s3_error_factory());
}
return callback();
}
template <typename Func>
-auto s3_get_rate_limit(Func callback) -> decltype(callback()) {
- return s3_rate_limit(doris::S3RateLimitType::GET, std::move(callback));
+auto s3_get_rate_limit(bool is_internal_bucket, Func callback) ->
decltype(callback()) {
+ return s3_rate_limit(is_internal_bucket, doris::S3RateLimitType::GET,
std::move(callback));
}
template <typename Func>
-auto s3_put_rate_limit(Func callback) -> decltype(callback()) {
- return s3_rate_limit(doris::S3RateLimitType::PUT, std::move(callback));
+auto s3_put_rate_limit(bool is_internal_bucket, Func callback) ->
decltype(callback()) {
+ return s3_rate_limit(is_internal_bucket, doris::S3RateLimitType::PUT,
std::move(callback));
+}
+
+template <typename Request>
+void set_s3_bytes_rate_limit_handler(Request& request, bool is_internal_bucket,
+ doris::S3RateLimitType type, size_t
bytes) {
+ request.SetRequestSignedHandler([is_internal_bucket, type,
+ bytes](const Aws::Http::HttpRequest&) {
+ if (!doris::acquire_s3_bytes_rate_limit(is_internal_bucket, type,
bytes)) {
Review Comment:
[P1] Avoid blocking after signing. In the bundled aws-sdk-cpp 1.11.219,
`AttemptOneRequest()` calls `SignRequest()`, then invokes
`GetRequestSignedHandler()`, and only then calls `MakeRequest()`.
`acquire_s3_bytes_rate_limit()` can sleep for a long interval under a low
bytes-per-second setting, so the request may reach S3 with a stale SigV4
timestamp; each retry repeats the same sign-then-sleep sequence. Please charge
the first attempt before entering the SDK and use `SetRequestRetryHandler`
(which runs before the next attempt is signed), or use another
pre-sign/transport hook. The test should drive the real SDK retry loop rather
than manually invoking the signed handler.
##########
common/cpp/token_bucket_rate_limiter.cpp:
##########
@@ -127,30 +127,35 @@ int64_t TokenBucketRateLimiter::add(size_t amount) {
TokenBucketRateLimiterHolder::TokenBucketRateLimiterHolder(size_t max_speed,
size_t max_burst,
size_t limit,
std::function<void(int64_t)> metric_func)
- : rate_limiter(std::make_unique<TokenBucketRateLimiter>(max_speed,
max_burst, limit)),
+ : rate_limiter(std::make_shared<TokenBucketRateLimiter>(max_speed,
max_burst, limit)),
+ _enabled(max_speed > 0 || limit > 0),
metric_func(std::move(metric_func)) {}
int64_t TokenBucketRateLimiterHolder::add(size_t amount) {
return add_with_config(amount).sleep_duration;
}
TokenBucketRateLimiterResult
TokenBucketRateLimiterHolder::add_with_config(size_t amount) {
- TokenBucketRateLimiterResult result;
+ std::shared_ptr<TokenBucketRateLimiter> limiter;
{
std::shared_lock read {rate_limiter_rw_lock};
- result = {.sleep_duration = rate_limiter->add(amount),
- .max_speed = rate_limiter->get_max_speed(),
- .max_burst = rate_limiter->get_max_burst(),
- .limit = rate_limiter->get_limit()};
+ limiter = rate_limiter;
}
+ TokenBucketRateLimiterResult result = {.sleep_duration =
limiter->add(amount),
+ .max_speed =
limiter->get_max_speed(),
+ .max_burst =
limiter->get_max_burst(),
+ .limit = limiter->get_limit()};
metric_func(result.sleep_duration);
return result;
}
int TokenBucketRateLimiterHolder::reset(size_t max_speed, size_t max_burst,
size_t limit) {
+ auto new_rate_limiter =
+ std::make_shared<TokenBucketRateLimiter>(max_speed, max_burst,
limit);
{
std::unique_lock write {rate_limiter_rw_lock};
- rate_limiter = std::make_unique<TokenBucketRateLimiter>(max_speed,
max_burst, limit);
+ rate_limiter = std::move(new_rate_limiter);
Review Comment:
[P1] Preserve outstanding rate/debt across reset. `add_with_config()`
snapshots the old limiter and releases the holder lock before `limiter->add()`
sleeps, while `reset()` immediately publishes a fresh full bucket. If the rate
is lowered during an old-generation sleep, retired requests still wake on the
old schedule while new requests consume the new bucket, so traffic can exceed
the newly configured cap. Please carry debt/reservations into the new
generation or gate new admissions until retired waiters drain, and add a
barrier-based reset-with-waiters test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
