github-actions[bot] commented on code in PR #66977:
URL: https://github.com/apache/doris/pull/66977#discussion_r3859642655
##########
be/src/cloud/cloud_ms_backpressure_handler.cpp:
##########
@@ -124,42 +142,59 @@ double TableRpcQpsCounter::get_qps() const {
// ============== TableRpcQpsRegistry ==============
-TableRpcQpsRegistry::TableRpcQpsRegistry() = default;
+TableRpcQpsRegistry::TableRpcQpsRegistry()
+ : TableRpcQpsRegistry(kQpsRegistryCleanupInterval,
qps_registry_inactive_timeout()) {}
-void TableRpcQpsRegistry::record(LoadRelatedRpc rpc_type, int64_t table_id) {
- auto* counter = get_or_create_counter(rpc_type, table_id);
- if (counter) {
- counter->increment();
+TableRpcQpsRegistry::TableRpcQpsRegistry(std::chrono::milliseconds
cleanup_interval,
+ std::chrono::milliseconds
inactive_timeout)
+ : _cleanup_interval(cleanup_interval),
+ _inactive_timeout(inactive_timeout),
+ _cleanup_stop_latch(1) {
+ DORIS_CHECK_GT(_cleanup_interval.count(), 0);
+ DORIS_CHECK_GE(_inactive_timeout.count(), 0);
+
+ auto st = Thread::create(
+ "TableRpcQpsRegistry", "cleanup_thread", [this]() {
this->_cleanup_thread_callback(); },
+ &_cleanup_thread);
+ if (!st.ok()) {
+ LOG(WARNING) << "[ms-throttle] failed to create table QPS registry
cleanup thread: " << st;
}
}
-TableRpcQpsCounter* TableRpcQpsRegistry::get_or_create_counter(LoadRelatedRpc
rpc_type,
- int64_t
table_id) {
+TableRpcQpsRegistry::~TableRpcQpsRegistry() {
+ _cleanup_stop_latch.count_down();
+ if (_cleanup_thread) {
+ _cleanup_thread->join();
+ }
+}
+
+void TableRpcQpsRegistry::record(LoadRelatedRpc rpc_type, int64_t table_id) {
size_t idx = static_cast<size_t>(rpc_type);
if (idx >= static_cast<size_t>(LoadRelatedRpc::COUNT)) {
- return nullptr;
+ return;
}
{
std::shared_lock lock(_mutex);
auto it = _counters[idx].find(table_id);
if (it != _counters[idx].end()) {
- return it->second.get();
+ it->second->increment();
+ return;
}
}
std::unique_lock lock(_mutex);
// Double check after acquiring exclusive lock
auto it = _counters[idx].find(table_id);
if (it != _counters[idx].end()) {
- return it->second.get();
+ it->second->increment();
+ return;
}
auto counter = std::make_unique<TableRpcQpsCounter>(table_id, rpc_type,
config::ms_rpc_table_qps_window_sec);
- auto* ptr = counter.get();
+ counter->increment();
Review Comment:
[P2] Construct first-touch bvars outside the registry lock
After the shared-lock miss, this path holds the single registry-wide write
lock while allocating and constructing both the `bvar::Adder` and
`bvar::PerSecond`. With collection now enabled by default, rotating table IDs
serialize unrelated RPC types and block existing-table `record()`, `get_qps()`,
and MS_BUSY top-k scans behind every first-touch bvar construction. This is
distinct from the earlier eviction comment: current head moved bvar destruction
outside the lock, but create-side work remains inside it. Please build a
prospective counter before locking, double-check/insert under the lock, destroy
any race loser after unlocking, and add a concurrent first-touch
progress/latency test.
##########
be/src/cloud/cloud_ms_backpressure_handler.cpp:
##########
@@ -245,15 +329,35 @@ TableRpcThrottler::TableRpcThrottler() {
std::chrono::steady_clock::time_point
TableRpcThrottler::throttle(LoadRelatedRpc rpc_type,
int64_t
table_id) {
+ return throttle(rpc_type, table_id, false).wait_until;
+}
+
+TableRpcThrottleDecision TableRpcThrottler::throttle(LoadRelatedRpc rpc_type,
int64_t table_id,
+ bool dry_run) {
std::shared_lock lock(_mutex);
auto it = _limiters.find({rpc_type, table_id});
if (it == _limiters.end()) {
- return std::chrono::steady_clock::now();
+ return {.wait_until = std::chrono::steady_clock::now(), .dry_run =
dry_run};
}
- return it->second->reserve();
+ return {
+ .wait_until = it->second->reserve(),
Review Comment:
[P1] Bound the dry-run reservation horizon
Dry-run still calls `reserve()` here for every request but skips the
corresponding sleep. Once persistent MS_BUSY has driven a hot table to the
default 1-QPS floor, 100,000 dry-run calls/s advance `_next_allowed_time` by
100,000 seconds per wall-clock second, exhausting an int64 nanosecond
`steady_clock::time_point` in about 26 hours (about 11 days at 10,000 calls/s).
The next duration addition overflows, and an ordinary wrap makes the deadline
look elapsed, corrupting would-wait observations and any later enforcement. The
existing transition-debt thread does not cover permanent dry-run, because
resetting only when enforcement is enabled still leaves this path unbounded.
Please cap/rebase dry-run reservations or use a separately bounded observation
model, with a saturation-boundary test.
##########
be/src/cloud/cloud_meta_mgr.cpp:
##########
@@ -493,33 +494,54 @@ struct RpcRateLimitCtx {
int64_t table_id {-1}; // For table-level backpressure, passed from caller
};
-// Apply rate limiting before RPC (both host-level and table-level)
-void apply_rate_limit(MetaServiceRPC rpc, const RpcRateLimitCtx& ctx) {
- // Table-level rate limit (for load-related RPCs only)
- if (ctx.backpressure_handler && ctx.table_id > 0) {
- LoadRelatedRpc load_rpc = to_load_related_rpc(rpc);
- if (load_rpc != LoadRelatedRpc::COUNT) {
- auto wait_until = ctx.backpressure_handler->before_rpc(load_rpc,
ctx.table_id);
- auto now = std::chrono::steady_clock::now();
- if (wait_until > now) {
- auto wait_us =
-
std::chrono::duration_cast<std::chrono::microseconds>(wait_until - now)
- .count();
- if (wait_us > 0) {
- if (auto* recorder = get_throttle_wait_recorder(load_rpc);
- recorder != nullptr) {
- *recorder << wait_us;
- }
- bthread_usleep(wait_us);
- }
- }
- }
+void apply_table_level_rate_limit(MetaServiceRPC rpc, const RpcRateLimitCtx&
ctx) {
+ if (ctx.backpressure_handler == nullptr || ctx.table_id <= 0) {
+ return;
}
- // Host-level rate limit
- if (ctx.host_limiters) {
- ctx.host_limiters->limit(rpc);
+ const auto load_rpc = to_load_related_rpc(rpc);
+ if (load_rpc == LoadRelatedRpc::COUNT) {
+ return;
+ }
+
+ const auto decision = ctx.backpressure_handler->before_rpc(load_rpc,
ctx.table_id);
+ const auto now = std::chrono::steady_clock::now();
+ if (decision.wait_until <= now) {
+ return;
+ }
+
+ const auto wait_us =
+
std::chrono::duration_cast<std::chrono::microseconds>(decision.wait_until - now)
+ .count();
+ if (wait_us <= 0) {
+ return;
+ }
+
+ auto* recorder = get_throttle_wait_recorder(load_rpc);
+ DCHECK(recorder);
+ *recorder << wait_us;
Review Comment:
[P1] Keep dry-run waits within the recorder range
This insertion happens before the once-per-second log gate, but the pinned
brpc 1.4 `LatencyRecorder` forwards to an `IntRecorder` that [clamps samples
above `INT_MAX` and emits a WARNING on every
insertion](https://github.com/apache/brpc/blob/1.4.0/src/bvar/recorder.h#L218-L266);
its percentile path also narrows to `uint32_t`. `INT_MAX` microseconds is only
about 2,147 seconds of modeled wait. At the 1-QPS table floor, 1,000 unslept
dry-run requests/s cross that bound in about 2.15 seconds, after which every
request can warn and the metric is no longer representative. The host dry-run
recorder has the same record-before-log path. This occurs long before the
underlying limiter states overflow and needs a metric-aware bound. Please
saturate/encode the sample explicitly (with an overflow counter or wider
metric) before insertion and add host/table boundary tests that prove no
per-request warnings.
##########
be/src/cloud/cloud_ms_rpc_rate_limiters.cpp:
##########
@@ -139,10 +148,21 @@ int64_t HostLevelMSRpcRateLimiters::limit(MetaServiceRPC
rpc) {
}
auto limiter = _limiters[idx].load();
- if (limiter && limiter->limiter) {
- return limiter->limiter->add(1);
+ if (!limiter) {
+ return 0;
+ }
+ DCHECK(limiter->limiter);
+
+ auto result = dry_run ? limiter->limiter->reserve_with_config(1)
Review Comment:
[P1] Bound the host dry-run wait calculation
This new dry-run branch reserves every request without sleeping, so traffic
above the host QPS limit drives the shared bucket debt downward without bound.
`reserve()` later computes that debt as a `double` nanosecond wait and casts it
directly to `int64_t`; once the modeled wait exceeds about 292 years, the
conversion is out of range and has undefined behavior. At a supported 1-QPS
limit, 100,000 calls/s reaches that boundary in about 26 hours (and a 100-QPS
limit on a one-core BE in about 107 days), after which observations and any
later enforcement can become non-positive or otherwise invalid. Resetting only
when enforcement is enabled would address the existing transition-debt thread
but not permanent dry-run. Please cap/rebase observation debt or saturate
before conversion, with a boundary test.
##########
be/src/cloud/cloud_ms_backpressure_handler.cpp:
##########
@@ -413,7 +518,8 @@ void MSBackpressureHandler::_advance_time(int ticks) {
bool MSBackpressureHandler::on_ms_busy() {
g_ms_busy_count << 1;
- if (!config::enable_ms_backpressure_handling) {
+ if (!config::enable_ms_backpressure_handling &&
+ !config::enable_ms_backpressure_handling_dry_run) {
Review Comment:
[P1] Do not enable upgrades without a downgrade driver
The handler constructor only logs when its separate tick-thread
`Thread::create()` fails, but the new default-on dry-run path still records QPS
and applies SET_LIMIT/history on MS_BUSY. That thread is the only production
caller of `_advance_time()`, so after this explicit startup failure no limit
can ever reach `on_downgrade()` or be removed; dry-run observations remain
permanently queued, and actual enforcement can retain a stale limit
indefinitely. This is distinct from the existing registry-cleanup-thread
comment: that worker owns counter eviction, while this worker owns all adaptive
recovery. Please fail construction/startup, or disable/fail MS_BUSY upgrades
unless the tick driver is live, and cover the injected failure.
--
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]
