github-actions[bot] commented on code in PR #66977:
URL: https://github.com/apache/doris/pull/66977#discussion_r3852926004
##########
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()) {
Review Comment:
[P1] Do not continue collection when cleanup cannot start
If `Thread::create()` fails (for example at the process thread/resource
limit), this constructor only logs and returns, but `record()` keeps allocating
a counter/bvar pair for every new table and `_cleanup_thread_callback()` is the
only production eviction caller. That restores the unbounded default-on growth
from the earlier thread on an explicit failure path. Please propagate/fail fast
on cleanup startup failure, or disable/bound collection with a fallback, and
cover the injected failure case.
##########
be/src/cloud/cloud_ms_backpressure_handler.cpp:
##########
@@ -216,25 +251,66 @@ double TableRpcQpsRegistry::get_qps(LoadRelatedRpc
rpc_type, int64_t table_id) c
return 0;
}
-void TableRpcQpsRegistry::cleanup_inactive_tables() {
- std::unique_lock lock(_mutex);
+size_t TableRpcQpsRegistry::cleanup_inactive_tables() {
+ const int64_t inactive_before_us =
+ MonotonicMicros() -
+
std::chrono::duration_cast<std::chrono::microseconds>(_inactive_timeout).count();
+ std::array<std::vector<int64_t>,
static_cast<size_t>(LoadRelatedRpc::COUNT)> candidates;
+
+ {
+ std::shared_lock lock(_mutex);
+ for (size_t idx = 0; idx < static_cast<size_t>(LoadRelatedRpc::COUNT);
++idx) {
+ for (const auto& [table_id, counter] : _counters[idx]) {
+ if (counter->last_record_time_us() <= inactive_before_us) {
+ candidates[idx].push_back(table_id);
+ }
+ }
+ }
+ }
+ size_t removed = 0;
+ std::unique_lock lock(_mutex);
for (size_t idx = 0; idx < static_cast<size_t>(LoadRelatedRpc::COUNT);
++idx) {
auto& counter_map = _counters[idx];
- for (auto it = counter_map.begin(); it != counter_map.end();) {
- // Remove counters with zero QPS for a long time
- if (it->second->get_qps() < 0.01) {
- it = counter_map.erase(it);
- } else {
- ++it;
+ for (int64_t table_id : candidates[idx]) {
+ auto it = counter_map.find(table_id);
+ if (it != counter_map.end() &&
+ it->second->last_record_time_us() <= inactive_before_us) {
+ counter_map.erase(it);
Review Comment:
[P2] Destroy evicted bvars after releasing the registry lock
`counter_map.erase(it)` synchronously destroys the counter's
`bvar::PerSecond`, backing `Adder`, and heap objects while the single
registry-wide write lock is held. The high-cardinality churn this cleanup is
meant to handle can therefore block every request-side `record()`/`get_qps()`
and MS_BUSY top-k scan for the whole destruction batch. Please detach the owned
counters under the lock and destroy them after unlocking (and consider
batching), with a concurrent high-cardinality latency test.
##########
be/src/cloud/cloud_throttle_state_machine.cpp:
##########
@@ -221,22 +233,29 @@ bool RpcThrottleCoordinator::report_ms_busy() {
<< ", cooldown=" << _params.upgrade_cooldown_ticks;
return true; // Should trigger upgrade
}
+
+ if (!_has_pending_upgrades) {
+ _ticks_since_last_ms_busy = -1;
+ }
return false; // Cooling down
}
-bool RpcThrottleCoordinator::tick(int ticks) {
+bool RpcThrottleCoordinator::tick(int64_t ticks) {
std::lock_guard lock(_mtx);
- // Increment tick counters
- if (_ticks_since_last_ms_busy >= 0) {
- _ticks_since_last_ms_busy += ticks;
- }
- if (_ticks_since_last_upgrade >= 0) {
- _ticks_since_last_upgrade += ticks;
+ // The upgrade counter is needed even without pending history to preserve
cooldown.
+ // Stop at the threshold because larger values do not change the decision.
+ advance_tick_counter(_ticks_since_last_upgrade, ticks,
_params.upgrade_cooldown_ticks);
Review Comment:
[P2] Preserve elapsed time when the cooldown grows
This caps the elapsed counter at the current cooldown. Because
`ms_backpressure_upgrade_interval_ms` is mutable, after a 10-second cap is
reached another 90 quiet seconds are discarded; if the cooldown is then raised
to 60 seconds, the next `MS_BUSY` sees only 10 and suppresses the upgrade for
50 more seconds even though 100 seconds have elapsed. Please retain elapsed
time independently of the mutable threshold (for example with the new `int64_t`
counter or a monotonic timestamp) and test an increase after spending extra
ticks beyond the old cap.
##########
be/src/cloud/cloud_ms_backpressure_handler.cpp:
##########
@@ -216,25 +251,66 @@ double TableRpcQpsRegistry::get_qps(LoadRelatedRpc
rpc_type, int64_t table_id) c
return 0;
}
-void TableRpcQpsRegistry::cleanup_inactive_tables() {
- std::unique_lock lock(_mutex);
+size_t TableRpcQpsRegistry::cleanup_inactive_tables() {
+ const int64_t inactive_before_us =
+ MonotonicMicros() -
+
std::chrono::duration_cast<std::chrono::microseconds>(_inactive_timeout).count();
+ std::array<std::vector<int64_t>,
static_cast<size_t>(LoadRelatedRpc::COUNT)> candidates;
+
+ {
+ std::shared_lock lock(_mutex);
+ for (size_t idx = 0; idx < static_cast<size_t>(LoadRelatedRpc::COUNT);
++idx) {
+ for (const auto& [table_id, counter] : _counters[idx]) {
+ if (counter->last_record_time_us() <= inactive_before_us) {
+ candidates[idx].push_back(table_id);
+ }
+ }
+ }
+ }
+ size_t removed = 0;
+ std::unique_lock lock(_mutex);
for (size_t idx = 0; idx < static_cast<size_t>(LoadRelatedRpc::COUNT);
++idx) {
auto& counter_map = _counters[idx];
- for (auto it = counter_map.begin(); it != counter_map.end();) {
- // Remove counters with zero QPS for a long time
- if (it->second->get_qps() < 0.01) {
- it = counter_map.erase(it);
- } else {
- ++it;
+ for (int64_t table_id : candidates[idx]) {
+ auto it = counter_map.find(table_id);
+ if (it != counter_map.end() &&
+ it->second->last_record_time_us() <= inactive_before_us) {
+ counter_map.erase(it);
+ ++removed;
}
}
}
+ return removed;
+}
+
+size_t TableRpcQpsRegistry::get_tracked_table_count(LoadRelatedRpc rpc_type)
const {
+ size_t idx = static_cast<size_t>(rpc_type);
+ if (idx >= static_cast<size_t>(LoadRelatedRpc::COUNT)) {
+ return 0;
+ }
+
+ std::shared_lock lock(_mutex);
+ return _counters[idx].size();
+}
+
+void TableRpcQpsRegistry::_cleanup_thread_callback() {
+ while (!_cleanup_stop_latch.wait_for(_cleanup_interval)) {
+ size_t removed = cleanup_inactive_tables();
Review Comment:
[P1] Bound throttle state along with inactive counters
This background pass bounds only `_counters`. With dry-run enabled by
default, rotating hot table IDs under persistent `MS_TOO_BUSY` still add real
entries to `TableRpcThrottler::_limiters`, `_current_limits`, and
`_upgrade_history`; every busy response resets the downgrade timer, so those
objects survive after their QPS counters are evicted and are later unwound only
one record per interval. Please coordinate eviction/capping across all
per-table throttle state and add a sustained rotating-table test that bounds
both memory and recovery time.
--
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]
