0AyanamiRei commented on code in PR #65420:
URL: https://github.com/apache/doris/pull/65420#discussion_r3613011235
##########
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:
Thanks. We accept the bounded transient overlap between retired and newly
published limiter generations during a rare dynamic limiter reset, and do not
plan to address it in this PR.
##########
be/src/util/s3_rate_limiter_manager.cpp:
##########
@@ -0,0 +1,204 @@
+// 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 "util/s3_rate_limiter_manager.h"
+
+#include <algorithm>
+#include <limits>
+#include <thread>
+
+#include "common/config.h"
+#include "common/logging.h"
+#include "util/cgroup_util.h"
+
+namespace doris {
+
+bvar::Adder<int64_t>
s3_get_bytes_rate_limit_sleep_ns("s3_get_bytes_rate_limit_sleep_ns");
+bvar::Adder<int64_t>
s3_get_bytes_rate_limit_sleep_count("s3_get_bytes_rate_limit_sleep_count");
+bvar::Adder<int64_t>
s3_put_bytes_rate_limit_sleep_ns("s3_put_bytes_rate_limit_sleep_ns");
+bvar::Adder<int64_t>
s3_put_bytes_rate_limit_sleep_count("s3_put_bytes_rate_limit_sleep_count");
+
+namespace {
+
+std::function<void(int64_t)> bytes_rate_limiter_metric_func(S3RateLimitType
type) {
+ switch (type) {
+ case S3RateLimitType::GET:
+ return metric_func_factory(s3_get_bytes_rate_limit_sleep_ns,
+ s3_get_bytes_rate_limit_sleep_count);
+ case S3RateLimitType::PUT:
+ return metric_func_factory(s3_put_bytes_rate_limit_sleep_ns,
+ s3_put_bytes_rate_limit_sleep_count);
+ default:
+ return [](int64_t) {};
+ }
+}
+
+// min(per_core * cores, cap) with overflow protection; cap == 0 means no cap.
+int64_t cap_multiply(int64_t per_core, int64_t cores, int64_t cap) {
+ cap = cap > 0 ? cap : std::numeric_limits<int64_t>::max();
+ if (per_core > cap / cores) {
+ return cap;
+ }
+ return per_core * cores;
+}
+
+size_t index_of(S3RateLimitType type) {
+ CHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT) <<
to_string(type);
+ return static_cast<size_t>(type);
+}
+
+} // namespace
+
+S3EffectiveRateLimit resolve_s3_rate_limit(S3RateLimitType type, int64_t
cores) {
+ const bool is_get = type == S3RateLimitType::GET;
+ const int64_t qps_per_core = is_get ? config::s3_get_qps_per_core :
config::s3_put_qps_per_core;
+ const int64_t qps_max = is_get ? config::s3_get_qps_max :
config::s3_put_qps_max;
+ const int64_t bytes_per_core = is_get ?
config::s3_get_bytes_per_second_per_core
+ :
config::s3_put_bytes_per_second_per_core;
+ const int64_t bytes_max =
+ is_get ? config::s3_get_bytes_per_second_max :
config::s3_put_bytes_per_second_max;
+ cores = std::max<int64_t>(1, cores);
+
+ S3EffectiveRateLimit limit;
+ if (qps_per_core < 0) {
+ // Unset: the legacy absolute configs stay in charge, bit-for-bit
compatible.
+ limit.qps = is_get ? config::s3_get_token_per_second :
config::s3_put_token_per_second;
+ limit.burst = is_get ? config::s3_get_bucket_tokens :
config::s3_put_bucket_tokens;
+ limit.count_limit = is_get ? config::s3_get_token_limit :
config::s3_put_token_limit;
+ } else if (qps_per_core > 0) {
+ limit.qps = cap_multiply(qps_per_core, cores, qps_max);
+ limit.burst = limit.qps; // burst = 1 second worth of quota
+ } // qps_per_core == 0: QPS limiting disabled,
all fields stay 0.
+
+ if (bytes_per_core > 0) {
+ limit.bytes_per_second = cap_multiply(bytes_per_core, cores,
bytes_max);
+ }
+ return limit;
+}
+
+int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t
max_burst, size_t limit) {
+ if (type == S3RateLimitType::UNKNOWN) {
+ return -1;
+ }
+ return
S3RateLimiterManager::instance().qps_limiter(type)->reset(max_speed, max_burst,
limit);
+}
+
+int64_t s3_rate_limiter_cpu_cores() {
+ if (int64_t overridden = config::s3_rate_limiter_cpu_cores; overridden >
0) {
+ return overridden;
+ }
+ int physical = static_cast<int>(std::thread::hardware_concurrency());
+ // Re-read the cgroup quota on every call: serverless BEs can be resized
in place,
+ // and the daemon refresh thread picks the change up through here.
+ int limited = CGroupUtil::get_cgroup_limited_cpu_number(physical);
+ return std::max(1, limited);
+}
+
+S3RateLimiterManager::S3RateLimiterManager() {
+ const int64_t cores = s3_rate_limiter_cpu_cores();
+ for (auto type : {S3RateLimitType::GET, S3RateLimitType::PUT}) {
+ auto limit = resolve_s3_rate_limit(type, cores);
+ _qps_limiters[index_of(type)] = std::make_unique<S3RateLimiterHolder>(
+ limit.qps, limit.burst, limit.count_limit,
s3_rate_limiter_metric_func(type));
+ _bytes_limiters[index_of(type)] =
std::make_unique<S3RateLimiterHolder>(
+ limit.bytes_per_second, limit.bytes_per_second, 0,
+ bytes_rate_limiter_metric_func(type));
+ }
+}
+
+S3RateLimiterManager& S3RateLimiterManager::instance() {
+ static S3RateLimiterManager ret;
+ return ret;
+}
+
+S3RateLimiterHolder* S3RateLimiterManager::qps_limiter(S3RateLimitType type) {
+ return _qps_limiters[index_of(type)].get();
+}
+
+S3RateLimiterHolder* S3RateLimiterManager::bytes_limiter(S3RateLimitType type)
{
+ return _bytes_limiters[index_of(type)].get();
+}
+
+void S3RateLimiterManager::refresh() {
+ std::lock_guard guard(_refresh_lock);
+ const int64_t cores = s3_rate_limiter_cpu_cores();
+ for (auto type : {S3RateLimitType::GET, S3RateLimitType::PUT}) {
+ const auto limit = resolve_s3_rate_limit(type, cores);
+
+ auto* qps = qps_limiter(type);
+ if (qps->get_max_speed() != static_cast<size_t>(limit.qps) ||
+ qps->get_max_burst() != static_cast<size_t>(limit.burst) ||
+ qps->get_limit() != static_cast<size_t>(limit.count_limit)) {
+ qps->reset(limit.qps, limit.burst, limit.count_limit);
+ LOG(INFO) << "reset S3 " << to_string(type) << " QPS rate limiter,
qps=" << limit.qps
+ << ", burst=" << limit.burst << ", count_limit=" <<
limit.count_limit
+ << ", cores=" << cores;
+ }
+
+ auto* bytes = bytes_limiter(type);
+ if (bytes->get_max_speed() !=
static_cast<size_t>(limit.bytes_per_second)) {
+ bytes->reset(limit.bytes_per_second, limit.bytes_per_second, 0);
+ LOG(INFO) << "reset S3 " << to_string(type)
+ << " bytes rate limiter, bytes_per_second=" <<
limit.bytes_per_second
+ << ", cores=" << cores;
+ }
+ }
+}
+
+S3RateLimitGuard::S3RateLimitGuard(S3RateLimitType type, size_t
estimated_bytes) {
+ if (!config::enable_s3_rate_limiter) {
+ return;
+ }
+ auto& mgr = S3RateLimiterManager::instance();
+
+ auto* qps = mgr.qps_limiter(type);
+ if (qps->is_enabled() &&
+ apply_s3_rate_limit(type, qps, config::s3_rate_limiter_log_interval) <
0) {
+ _ok = false;
+ return;
+ }
+
+ if (estimated_bytes == 0) {
+ return;
+ }
+ auto* bytes = mgr.bytes_limiter(type);
+ if (!bytes->is_enabled()) {
+ return;
+ }
+ // Clamp the reservation to 1 second worth of bandwidth so a single
oversized IO
+ // (e.g. a whole-file read_at) cannot create unbounded upfront debt. The
clamped
+ // remainder is intentionally not accounted; effective quotas below the
single-IO
+ // upper bound are excluded by the config contract (see config.cpp).
+ _reserved = std::min(estimated_bytes, bytes->get_max_speed());
Review Comment:
Thanks. Strict accounting for an I/O larger than one second of quota needs a
separate design discussion, since simply removing the clamp would reintroduce
unbounded limiter sleeps. We will discuss this separately and will not address
it in this PR.
##########
be/src/io/fs/s3_obj_storage_client.cpp:
##########
@@ -173,7 +175,9 @@ ObjectStorageResponse S3ObjStorageClient::put_object(const
ObjectStoragePathOpti
MonotonicStopWatch watch;
watch.start();
auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(
- s3_put_rate_limit([&]() { return _client->PutObject(request); }),
+ s3_put_rate_limit(
Review Comment:
This is intentional. The limiter accounts bytes once per logical top-level
object-storage operation; SDK-internal retry wire bytes are outside this
contract. The PR description and decorator comments now state this
logical-request boundary explicitly, so we do not plan a change here.
##########
be/src/io/fs/rate_limited_obj_storage_client.h:
##########
@@ -0,0 +1,70 @@
+// 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.
+
+#pragma once
+
+#include <memory>
+
+#include "io/fs/obj_storage_client.h"
+
+namespace doris::io {
+
+// Decorator that applies the process-wide S3 GET/PUT QPS and bandwidth rate
limiters
+// in front of any ObjStorageClient. This is the single place where rate
limiting is
+// wired into the object storage path: provider clients (S3, Azure, future
GCP, ...)
+// contain no rate limiting code, and S3ClientFactory decides at construction
time
+// whether to wrap a client (internal storage-vault buckets) or return it bare
+// (external buckets: S3 load, TVF, external catalogs in cloud mode).
+//
+// Each public API call is charged once against the QPS bucket, and
data-carrying
+// calls additionally reserve their payload size from the bytes bucket
(reconciled
+// with the actually transferred size for reads). Note that APIs which
internally
+// paginate (list_objects, delete_objects_recursively) are charged once per
logical
Review Comment:
Thanks for pointing this out. Charging once per logical object-storage
operation is intentional for the current scope.
In cloud mode, rate limiting is applied only to internal storage-vault
buckets. BE traffic to those buckets uses the 1:1 data paths such as GET, PUT,
UploadPart, and HEAD. BE does not delete vault data; deletion is owned by the
separate recycler process, which is outside this PR target. BE also has no
current LIST scenario for internal vault buckets. In addition, a single-tablet
prefix exceeding one page (1,000 objects) is rare.
If a future BE feature introduces high-frequency LIST traffic against
internal buckets, such as a warmup scan, we can have the factory inject a
shared/no-op provider policy and add a one-line hook in each pagination loop
without changing the public ObjStorageClient interface. We therefore accept
logical-call QPS accounting for this PR and do not plan a change here.
##########
be/src/io/fs/rate_limited_obj_storage_client.cpp:
##########
@@ -0,0 +1,139 @@
+// 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 "io/fs/rate_limited_obj_storage_client.h"
+
+#include "common/status.h"
+#include "util/s3_rate_limiter_manager.h"
+
+namespace doris::io {
+namespace {
+
+ObjectStorageResponse rate_limited_response(S3RateLimitType type) {
+ return {.status =
convert_to_obj_response(Status::Error<ErrorCode::EXCEEDED_LIMIT, false>(
+ "s3 {} request exceeds request limit, rejected by BE rate
limiter",
+ to_string(type))),
+ .http_code = 429};
Review Comment:
Handled in 3e1c0d3e05d. Rejected admissions now atomically roll back both
the current count and the token charge, so retrying a rejected local admission
no longer accumulates count or rate debt. We keep the shared 429 path because
it also represents transient bytes-reservation saturation, where retry is
expected.
##########
be/src/util/s3_rate_limiter_manager.cpp:
##########
@@ -0,0 +1,204 @@
+// 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 "util/s3_rate_limiter_manager.h"
+
+#include <algorithm>
+#include <limits>
+#include <thread>
+
+#include "common/config.h"
+#include "common/logging.h"
+#include "util/cgroup_util.h"
+
+namespace doris {
+
+bvar::Adder<int64_t>
s3_get_bytes_rate_limit_sleep_ns("s3_get_bytes_rate_limit_sleep_ns");
+bvar::Adder<int64_t>
s3_get_bytes_rate_limit_sleep_count("s3_get_bytes_rate_limit_sleep_count");
+bvar::Adder<int64_t>
s3_put_bytes_rate_limit_sleep_ns("s3_put_bytes_rate_limit_sleep_ns");
+bvar::Adder<int64_t>
s3_put_bytes_rate_limit_sleep_count("s3_put_bytes_rate_limit_sleep_count");
+
+namespace {
+
+std::function<void(int64_t)> bytes_rate_limiter_metric_func(S3RateLimitType
type) {
+ switch (type) {
+ case S3RateLimitType::GET:
+ return metric_func_factory(s3_get_bytes_rate_limit_sleep_ns,
+ s3_get_bytes_rate_limit_sleep_count);
+ case S3RateLimitType::PUT:
+ return metric_func_factory(s3_put_bytes_rate_limit_sleep_ns,
+ s3_put_bytes_rate_limit_sleep_count);
+ default:
+ return [](int64_t) {};
+ }
+}
+
+// min(per_core * cores, cap) with overflow protection; cap == 0 means no cap.
+int64_t cap_multiply(int64_t per_core, int64_t cores, int64_t cap) {
+ cap = cap > 0 ? cap : std::numeric_limits<int64_t>::max();
+ if (per_core > cap / cores) {
+ return cap;
+ }
+ return per_core * cores;
+}
+
+size_t index_of(S3RateLimitType type) {
+ CHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT) <<
to_string(type);
+ return static_cast<size_t>(type);
+}
+
+} // namespace
+
+S3EffectiveRateLimit resolve_s3_rate_limit(S3RateLimitType type, int64_t
cores) {
+ const bool is_get = type == S3RateLimitType::GET;
+ const int64_t qps_per_core = is_get ? config::s3_get_qps_per_core :
config::s3_put_qps_per_core;
+ const int64_t qps_max = is_get ? config::s3_get_qps_max :
config::s3_put_qps_max;
+ const int64_t bytes_per_core = is_get ?
config::s3_get_bytes_per_second_per_core
+ :
config::s3_put_bytes_per_second_per_core;
+ const int64_t bytes_max =
+ is_get ? config::s3_get_bytes_per_second_max :
config::s3_put_bytes_per_second_max;
+ cores = std::max<int64_t>(1, cores);
+
+ S3EffectiveRateLimit limit;
+ if (qps_per_core < 0) {
+ // Unset: the legacy absolute configs stay in charge, bit-for-bit
compatible.
+ limit.qps = is_get ? config::s3_get_token_per_second :
config::s3_put_token_per_second;
+ limit.burst = is_get ? config::s3_get_bucket_tokens :
config::s3_put_bucket_tokens;
+ limit.count_limit = is_get ? config::s3_get_token_limit :
config::s3_put_token_limit;
+ } else if (qps_per_core > 0) {
+ limit.qps = cap_multiply(qps_per_core, cores, qps_max);
+ limit.burst = limit.qps; // burst = 1 second worth of quota
+ } // qps_per_core == 0: QPS limiting disabled,
all fields stay 0.
+
+ if (bytes_per_core > 0) {
+ limit.bytes_per_second = cap_multiply(bytes_per_core, cores,
bytes_max);
+ }
+ return limit;
+}
+
+int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t
max_burst, size_t limit) {
+ if (type == S3RateLimitType::UNKNOWN) {
+ return -1;
+ }
+ return
S3RateLimiterManager::instance().qps_limiter(type)->reset(max_speed, max_burst,
limit);
+}
+
+int64_t s3_rate_limiter_cpu_cores() {
+ if (int64_t overridden = config::s3_rate_limiter_cpu_cores; overridden >
0) {
+ return overridden;
+ }
+ int physical = static_cast<int>(std::thread::hardware_concurrency());
+ // Re-read the cgroup quota on every call: serverless BEs can be resized
in place,
+ // and the daemon refresh thread picks the change up through here.
+ int limited = CGroupUtil::get_cgroup_limited_cpu_number(physical);
+ return std::max(1, limited);
+}
+
+S3RateLimiterManager::S3RateLimiterManager() {
+ const int64_t cores = s3_rate_limiter_cpu_cores();
+ for (auto type : {S3RateLimitType::GET, S3RateLimitType::PUT}) {
+ auto limit = resolve_s3_rate_limit(type, cores);
+ _qps_limiters[index_of(type)] = std::make_unique<S3RateLimiterHolder>(
+ limit.qps, limit.burst, limit.count_limit,
s3_rate_limiter_metric_func(type));
+ _bytes_limiters[index_of(type)] =
std::make_unique<S3RateLimiterHolder>(
+ limit.bytes_per_second, limit.bytes_per_second, 0,
+ bytes_rate_limiter_metric_func(type));
+ }
+}
+
+S3RateLimiterManager& S3RateLimiterManager::instance() {
+ static S3RateLimiterManager ret;
+ return ret;
+}
+
+S3RateLimiterHolder* S3RateLimiterManager::qps_limiter(S3RateLimitType type) {
+ return _qps_limiters[index_of(type)].get();
+}
+
+S3RateLimiterHolder* S3RateLimiterManager::bytes_limiter(S3RateLimitType type)
{
+ return _bytes_limiters[index_of(type)].get();
+}
+
+void S3RateLimiterManager::refresh() {
+ std::lock_guard guard(_refresh_lock);
+ const int64_t cores = s3_rate_limiter_cpu_cores();
+ for (auto type : {S3RateLimitType::GET, S3RateLimitType::PUT}) {
+ const auto limit = resolve_s3_rate_limit(type, cores);
+
+ auto* qps = qps_limiter(type);
+ if (qps->get_max_speed() != static_cast<size_t>(limit.qps) ||
+ qps->get_max_burst() != static_cast<size_t>(limit.burst) ||
+ qps->get_limit() != static_cast<size_t>(limit.count_limit)) {
+ qps->reset(limit.qps, limit.burst, limit.count_limit);
+ LOG(INFO) << "reset S3 " << to_string(type) << " QPS rate limiter,
qps=" << limit.qps
+ << ", burst=" << limit.burst << ", count_limit=" <<
limit.count_limit
+ << ", cores=" << cores;
+ }
+
+ auto* bytes = bytes_limiter(type);
+ if (bytes->get_max_speed() !=
static_cast<size_t>(limit.bytes_per_second)) {
+ bytes->reset(limit.bytes_per_second, limit.bytes_per_second, 0);
+ LOG(INFO) << "reset S3 " << to_string(type)
+ << " bytes rate limiter, bytes_per_second=" <<
limit.bytes_per_second
+ << ", cores=" << cores;
+ }
+ }
+}
+
+S3RateLimitGuard::S3RateLimitGuard(S3RateLimitType type, size_t
estimated_bytes) {
+ if (!config::enable_s3_rate_limiter) {
+ return;
+ }
+ auto& mgr = S3RateLimiterManager::instance();
+
+ auto* qps = mgr.qps_limiter(type);
+ if (qps->is_enabled() &&
+ apply_s3_rate_limit(type, qps, config::s3_rate_limiter_log_interval) <
0) {
+ _ok = false;
+ return;
+ }
+
+ if (estimated_bytes == 0) {
+ return;
+ }
+ auto* bytes = mgr.bytes_limiter(type);
+ if (!bytes->is_enabled()) {
+ return;
+ }
+ // Clamp the reservation to 1 second worth of bandwidth so a single
oversized IO
+ // (e.g. a whole-file read_at) cannot create unbounded upfront debt. The
clamped
+ // remainder is intentionally not accounted; effective quotas below the
single-IO
+ // upper bound are excluded by the config contract (see config.cpp).
+ _reserved = std::min(estimated_bytes, bytes->get_max_speed());
+ if (_reserved > 0) {
+ // Debt model: may sleep, never rejects (count limit is 0). Pin the
charged
+ // bucket generation for settle().
+ _charged_bucket = bytes->charge(_reserved);
Review Comment:
Thanks. This requires a limiter refresh to land exactly between reservation
sizing and charging, while reset itself only occurs when the effective config
or CPU allocation changes. We consider this transient race extremely unlikely
and do not plan to add the extra synchronization/API complexity in this PR.
##########
common/cpp/token_bucket_rate_limiter.cpp:
##########
@@ -124,33 +124,58 @@ int64_t TokenBucketRateLimiter::add(size_t amount) {
return sleep_time_ns;
}
+void TokenBucketRateLimiter::refund(size_t amount) {
Review Comment:
Thanks. We will discuss the refill/debit/refund ordering separately and
handle it outside this PR.
--
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]
