github-actions[bot] commented on code in PR #65420:
URL: https://github.com/apache/doris/pull/65420#discussion_r3575720821
##########
be/src/util/s3_util.cpp:
##########
@@ -156,72 +159,318 @@ 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
Review Comment:
`get_current_num_cores()` is the raw online host CPU count, but this value
is now used to size `s3_*_per_core` limits. That bypasses the cgroup and
`--num_cores` cap that `CpuInfo::init()` stores in `num_cores_` and that other
per-core BE limiters use through `CpuInfo::num_cores()`. On a BE limited to 4
CPUs in a cgroup but running on a 64-core host, `s3_get_qps_per_core=10` would
configure 640 GET QPS instead of 40, and the byte limiters have the same
problem. Please base this on the Doris-available core count, or clamp the
refreshed count through the same cgroup/config logic before using it for S3
limits.
##########
be/src/util/s3_util.cpp:
##########
@@ -156,72 +159,318 @@ 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::get_current_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 get_s3_rate_limiter_cpu_cores_for_current_config() {
+ if (get_s3_qps_per_core(S3RateLimitType::GET) > 0 ||
+ get_s3_qps_per_core(S3RateLimitType::PUT) > 0 ||
+ get_s3_bytes_per_second_per_core(S3RateLimitType::GET) > 0 ||
+ get_s3_bytes_per_second_per_core(S3RateLimitType::PUT) > 0) {
+ return get_s3_rate_limiter_cpu_cores();
+ }
+ 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:
+ 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;
+ per_core = config::s3_get_qps_per_core;
+ max_qps = config::s3_get_qps_max;
+ break;
+ case S3RateLimitType::PUT:
+ 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;
+ per_core = config::s3_put_qps_per_core;
+ max_qps = config::s3_put_qps_max;
+ 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
bvar::Adder<int64_t> get_rate_limit_ns("get_rate_limit_ns");
bvar::Adder<int64_t>
get_rate_limit_exceed_req_num("get_rate_limit_exceed_req_num");
bvar::Adder<int64_t> put_rate_limit_ns("put_rate_limit_ns");
bvar::Adder<int64_t>
put_rate_limit_exceed_req_num("put_rate_limit_exceed_req_num");
+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");
+
+namespace {
+
+enum class S3RateLimiterKind {
+ QPS,
+ BYTES,
+};
-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};
+struct S3RateLimiterUpdateState {
+ S3RateLimiterConfig applied_config;
+ std::atomic<int64_t> applied_cpu_cores {0};
+ std::mutex update_lock;
-static std::atomic<bool> updating_get_limiter {false};
-static std::atomic<bool> updating_put_limiter {false};
+ void set_applied(const S3RateLimiterConfig& config, int64_t cpu_cores) {
+ applied_config = config;
+ applied_cpu_cores.store(cpu_cores, std::memory_order_release);
+ }
+};
+
+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(S3RateLimitType type, S3RateLimiterKind kind,
int64_t cpu_cores,
+ bool wait_for_concurrent_update = true) {
+ // Construction initializes update_states and must finish before taking an
update lock.
+ auto& factory = S3ClientFactory::instance();
+ auto& state = get_update_state(type, kind);
+ std::unique_lock update_guard(state.update_lock, std::defer_lock);
+ if (wait_for_concurrent_update) {
+ update_guard.lock();
+ } else if (!update_guard.try_lock()) {
+ return;
+ }
+
+ if (!wait_for_concurrent_update) {
+ DCHECK(kind == S3RateLimiterKind::QPS);
Review Comment:
`token_limit` is still part of the effective QPS limiter in per-core mode,
but this early return prevents the legacy checker from ever noticing
`s3_get_token_limit` or `s3_put_token_limit` changes once `s3_*_qps_per_core >=
0`. The hot path only rebuilds on CPU-count changes, and the new on-update
callbacks do not cover the token-limit configs, so an existing limiter can keep
its old limit until some unrelated per-core/max/CPU change resets it. Please
either register on-update handlers for the two token-limit configs or let this
checker continue comparing the token-limit field while per-core QPS is active.
--
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]
