github-actions[bot] commented on code in PR #65420:
URL: https://github.com/apache/doris/pull/65420#discussion_r3593637405
##########
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:
[P1] Charge bytes at the retry-attempt boundary. This admission runs once
around `PutObject`, but the factory installs
`S3CustomRetryStrategy(max_s3_client_retry)` (default 10), so a retryable 1 GiB
PUT can send the payload twice while consuming only 1 GiB of tokens. Azure's
`UploadFrom`/`StageBlock` wrappers have the same per-operation placement around
a client with `Retry.MaxRetries`. That lets failure-time wire throughput exceed
the advertised bytes-per-second cap by the retry multiplier. Please acquire in
each SDK retry attempt (Azure can use a per-retry policy), or explicitly
define/rename this as a logical-request-byte limiter and test that contract.
##########
be/src/common/config.cpp:
##########
@@ -2197,6 +2240,7 @@ bool init(const char* conf_file, bool fill_conf_map, bool
must_exist, bool set_t
}
\
TYPE& ref_conf_value = *reinterpret_cast<TYPE*>((FIELD).storage);
\
TYPE old_value = ref_conf_value;
\
+ ref_conf_value = new_value;
\
Review Comment:
[P1] Validate the candidate before publishing it. `DEFINE_VALIDATOR`
installs a zero-argument closure that re-reads this global field, while numeric
`set_config()` calls and runtime readers are unsynchronized. Two concurrent
`/api/update_config` calls can therefore validate each other's values or let a
failed call restore its stale `old_value` over a successful update. A rejected
`variant_storage_parse_mode=3` is also briefly visible to the load path, whose
default arm `CHECK`s and can terminate the BE. Please pass `new_value` directly
to the validator and publish it only after validation succeeds (and serialize
same-field publication/bookkeeping), with a barrier-based concurrency test.
##########
be/src/util/s3_util.cpp:
##########
@@ -156,67 +160,276 @@ constexpr char S3_NEED_OVERRIDE_ENDPOINT[] =
"AWS_NEED_OVERRIDE_ENDPOINT";
constexpr char S3_ROLE_ARN[] = "AWS_ROLE_ARN";
constexpr char S3_EXTERNAL_ID[] = "AWS_EXTERNAL_ID";
constexpr char S3_CREDENTIALS_PROVIDER_TYPE[] =
"AWS_CREDENTIALS_PROVIDER_TYPE";
+
+#ifdef BE_TEST
+std::atomic<int64_t> s3_rate_limiter_cpu_cores_for_test {0};
+#endif
+
+int64_t get_s3_rate_limiter_cpu_cores() {
+#ifdef BE_TEST
+ const int64_t cpu_cores_for_test =
+ s3_rate_limiter_cpu_cores_for_test.load(std::memory_order_relaxed);
+ if (cpu_cores_for_test > 0) {
+ return cpu_cores_for_test;
+ }
+#endif
+ return CpuInfo::num_cores();
+}
+
+int64_t get_s3_qps_per_core(S3RateLimitType type) {
+ switch (type) {
+ case S3RateLimitType::GET:
+ return config::s3_get_qps_per_core;
+ case S3RateLimitType::PUT:
+ return config::s3_put_qps_per_core;
+ default:
+ CHECK(false) << "unknown S3 rate limiter type: " << to_string(type);
+ return -1;
+ }
+}
+
+int64_t get_s3_bytes_per_second_per_core(S3RateLimitType type) {
+ switch (type) {
+ case S3RateLimitType::GET:
+ return config::s3_get_bytes_per_second_per_core;
+ case S3RateLimitType::PUT:
+ return config::s3_put_bytes_per_second_per_core;
+ default:
+ CHECK(false) << "unknown S3 rate limiter type: " << to_string(type);
+ return -1;
+ }
+}
+
+int64_t compute_s3_rate_limit_from_core(int64_t per_core, int64_t cores,
int64_t max_rate) {
+ max_rate = max_rate > 0 ? max_rate : std::numeric_limits<int64_t>::max();
+ if (per_core > max_rate / cores) {
+ return max_rate;
+ }
+ return per_core * cores;
+}
+
+S3RateLimiterConfig
get_effective_s3_rate_limiter_config_with_cpu_cores(S3RateLimitType type,
+
int64_t cpu_cores) {
+ S3RateLimiterConfig result;
+ int64_t per_core;
+ int64_t max_qps;
+
+ switch (type) {
+ case S3RateLimitType::GET:
+ per_core = config::s3_get_qps_per_core;
+ max_qps = config::s3_get_qps_max;
+ if (per_core < 0) {
+ result.token_limit = config::s3_get_token_limit;
+ result.token_per_second = config::s3_get_token_per_second;
+ result.bucket_tokens = config::s3_get_bucket_tokens;
+ }
+ break;
+ case S3RateLimitType::PUT:
+ per_core = config::s3_put_qps_per_core;
+ max_qps = config::s3_put_qps_max;
+ if (per_core < 0) {
+ result.token_limit = config::s3_put_token_limit;
+ result.token_per_second = config::s3_put_token_per_second;
+ result.bucket_tokens = config::s3_put_bucket_tokens;
+ }
+ break;
+ default:
+ CHECK(false) << "unknown S3 rate limiter type: " << to_string(type);
+ return result;
+ }
+
+ if (per_core >= 0) {
+ result.token_per_second = compute_s3_rate_limit_from_core(per_core,
cpu_cores, max_qps);
+ result.bucket_tokens = result.token_per_second;
+ }
+ return result;
+}
+
+S3RateLimiterConfig
get_effective_s3_bytes_rate_limiter_config_with_cpu_cores(S3RateLimitType type,
+
int64_t cpu_cores) {
+ int64_t per_core = 0;
+ int64_t max_bytes_per_second = 0;
+
+ switch (type) {
+ case S3RateLimitType::GET:
+ per_core = config::s3_get_bytes_per_second_per_core;
+ max_bytes_per_second = config::s3_get_bytes_per_second_max;
+ break;
+ case S3RateLimitType::PUT:
+ per_core = config::s3_put_bytes_per_second_per_core;
+ max_bytes_per_second = config::s3_put_bytes_per_second_max;
+ break;
+ default:
+ CHECK(false) << "unknown S3 rate limiter type: " << to_string(type);
+ }
+
+ S3RateLimiterConfig result;
+ if (per_core > 0) {
+ result.token_per_second =
+ compute_s3_rate_limit_from_core(per_core, cpu_cores,
max_bytes_per_second);
+ result.bucket_tokens = result.token_per_second;
+ }
+ return result;
+}
} // namespace
-static std::atomic<int64_t> last_s3_get_token_bucket_tokens {0};
-static std::atomic<int64_t> last_s3_get_token_limit {0};
-static std::atomic<int64_t> last_s3_get_token_per_second {0};
-static std::atomic<int64_t> last_s3_put_token_per_second {0};
-static std::atomic<int64_t> last_s3_put_token_bucket_tokens {0};
-static std::atomic<int64_t> last_s3_put_token_limit {0};
+bvar::Adder<int64_t> get_bytes_rate_limit_ns("get_bytes_rate_limit_ns");
+bvar::Adder<int64_t>
get_bytes_rate_limit_exceed_req_num("get_bytes_rate_limit_exceed_req_num");
+bvar::Adder<int64_t> put_bytes_rate_limit_ns("put_bytes_rate_limit_ns");
+bvar::Adder<int64_t>
put_bytes_rate_limit_exceed_req_num("put_bytes_rate_limit_exceed_req_num");
-static std::atomic<bool> updating_get_limiter {false};
-static std::atomic<bool> updating_put_limiter {false};
+namespace {
+
+enum class S3RateLimiterKind {
+ QPS,
+ BYTES,
+};
+
+struct S3RateLimiterUpdateState {
+ S3RateLimiterConfig applied_config;
+ std::mutex update_lock;
+};
+
+std::array<std::array<S3RateLimiterUpdateState, 2>, 2> update_states;
+
+S3RateLimiterUpdateState& get_update_state(S3RateLimitType type,
S3RateLimiterKind kind) {
+ DCHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT);
+ return update_states[static_cast<size_t>(kind)][static_cast<size_t>(type)];
+}
+
+void update_s3_rate_limiter(S3ClientFactory& factory, S3RateLimitType type,
S3RateLimiterKind kind,
+ int64_t cpu_cores) {
+ auto& state = get_update_state(type, kind);
+ std::lock_guard update_guard(state.update_lock);
+
+ const int64_t per_core = kind == S3RateLimiterKind::BYTES
+ ? get_s3_bytes_per_second_per_core(type)
+ : get_s3_qps_per_core(type);
+ const auto current =
+ kind == S3RateLimiterKind::BYTES
+ ?
get_effective_s3_bytes_rate_limiter_config_with_cpu_cores(type, cpu_cores)
+ :
get_effective_s3_rate_limiter_config_with_cpu_cores(type, cpu_cores);
+ const bool config_changed = state.applied_config.token_per_second !=
current.token_per_second ||
+ state.applied_config.bucket_tokens !=
current.bucket_tokens ||
+ state.applied_config.token_limit !=
current.token_limit;
+ if (!config_changed) {
+ return;
+ }
+
+ auto* limiter = kind == S3RateLimiterKind::BYTES ?
factory.bytes_rate_limiter(type)
+ :
factory.rate_limiter(type);
+ const int ret =
+ limiter->reset(current.token_per_second, current.bucket_tokens,
current.token_limit);
+ if (ret != 0) {
+ LOG(WARNING) << "Failed to reset S3 " << to_string(type)
+ << (kind == S3RateLimiterKind::BYTES ? " bytes" : " QPS")
+ << " rate limiter, error code: " << ret;
+ return;
+ }
+
+ const int64_t logged_cpu_cores = per_core > 0 ? cpu_cores : 1;
+ if (kind == S3RateLimiterKind::BYTES) {
+ LOG(INFO) << "Reset S3 " << to_string(type)
+ << " bytes rate limiter, bytes_per_second=" <<
current.token_per_second
+ << ", burst_bytes=" << current.bucket_tokens
+ << ", cpu_cores=" << logged_cpu_cores;
+ } else if (per_core >= 0) {
+ LOG(INFO) << "Reset S3 " << to_string(type)
+ << " QPS rate limiter, qps=" << current.token_per_second
+ << ", burst=" << current.bucket_tokens << ", limit=" <<
current.token_limit
+ << ", cpu_cores=" << logged_cpu_cores;
+ }
+
+ state.applied_config = current;
+}
+
+} // namespace
S3RateLimiterHolder* S3ClientFactory::rate_limiter(S3RateLimitType type) {
CHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT) <<
to_string(type);
return _rate_limiters[static_cast<size_t>(type)].get();
}
-template <S3RateLimitType LimiterType>
-void update_rate_limiter_if_changed(int64_t current_tps, int64_t
current_bucket,
- int64_t current_limit,
std::atomic<int64_t>& last_tps,
- std::atomic<int64_t>& last_bucket,
- std::atomic<int64_t>& last_limit,
- std::atomic<bool>& updating_flag, const
char* limiter_name) {
- if (last_tps.load(std::memory_order_relaxed) != current_tps ||
- last_bucket.load(std::memory_order_relaxed) != current_bucket ||
- last_limit.load(std::memory_order_relaxed) != current_limit) {
- bool expected = false;
- if (!updating_flag.compare_exchange_strong(expected, true,
std::memory_order_acq_rel)) {
- return;
- }
- if (last_tps.load(std::memory_order_acquire) != current_tps ||
- last_bucket.load(std::memory_order_acquire) != current_bucket ||
- last_limit.load(std::memory_order_acquire) != current_limit) {
- int ret =
- reset_s3_rate_limiter(LimiterType, current_tps,
current_bucket, current_limit);
-
- if (ret == 0) {
- last_tps.store(current_tps, std::memory_order_release);
- last_bucket.store(current_bucket, std::memory_order_release);
- last_limit.store(current_limit, std::memory_order_release);
- } else {
- LOG(WARNING) << "Failed to reset S3 " << limiter_name
- << " rate limiter, error code: " << ret;
- }
- }
+S3RateLimiterHolder* S3ClientFactory::bytes_rate_limiter(S3RateLimitType type)
{
+ CHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT) <<
to_string(type);
+ return _bytes_rate_limiters[static_cast<size_t>(type)].get();
+}
- updating_flag.store(false, std::memory_order_release);
+bool acquire_s3_rate_limit(S3RateLimitType type, size_t bytes) {
+ if (!config::enable_s3_rate_limiter) {
+ return true;
+ }
+
Review Comment:
[P2] Skip a disabled QPS holder too. `s3_*_qps_per_core=0` intentionally
publishes an all-zero holder, but this call still enters `add_with_config(1)`:
it takes the holder shared mutex, bumps the `shared_ptr` references, takes the
limiter spinlock, increments `_count`, and runs the callback although it cannot
throttle or reject. In a supported bytes-only setup this serializes every
S3/Azure operation, including zero-byte metadata requests, on a no-op QPS
limiter. The existing `is_enabled()` guard fixes only the symmetric
disabled-bytes case. Please gate this call on the active QPS holder's
`is_enabled()` state and add a bytes-only fast-path test.
##########
be/test/common/config_test.cpp:
##########
@@ -32,8 +32,10 @@ namespace doris {
using namespace config;
class ConfigTest : public testing::Test {
- void SetUp() override { config::Register::_s_field_map->clear(); }
- void TearDown() override { config::Register::_s_field_map->clear(); }
+ void SetUp() override {
_saved_fields.swap(*config::Register::_s_field_map); }
+ void TearDown() override {
_saved_fields.swap(*config::Register::_s_field_map); }
Review Comment:
[P2] Restore the matching config snapshot too. `DumpAllConfigs` fills the
process-global `full_conf_map` with these fixture-only `cfg_*` keys, but this
teardown swaps back only `Register::_s_field_map`. Afterward
`get_config_info()` joins the restored production registry against the stale
test map and returns no production entries, making later tests/order-dependent
helpers observe an inconsistent config state. Please save/restore
`full_conf_map` (including its initially-null state) together with the
registry, or rebuild it after teardown.
--
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]
