You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by li...@apache.org on 2026/04/29 08:57:23 UTC
(doris) branch master updated: [fix](memtable) use CPU metrics delta for CPU pressure detection in adaptive flush controller (#62744)
This is an automated email from the ASF dual-hosted git repository.
liaoxin01 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 7ba4569729a [fix](memtable) use CPU metrics delta for CPU pressure detection in adaptive flush controller (#62744)
7ba4569729a is described below
commit 7ba4569729ad0becf7bc5def7b2c669d8cdcdd44
Author: hui lai <la...@selectdb.com>
AuthorDate: Wed Apr 29 16:57:15 2026 +0800
[fix](memtable) use CPU metrics delta for CPU pressure detection in adaptive flush controller (#62744)
### What problem does this PR solve?
Related PR: https://github.com/apache/doris/issues/60617
Problem Summary:
`AdaptiveThreadPoolController::is_cpu_busy()` previously used normalized
1-minute load average as a proxy for CPU usage. This could misclassify
IO wait pressure as CPU pressure and make adaptive flush thread
shrinking too aggressive.
This PR changes CPU pressure detection to use deltas of the existing
`doris_be_cpu` metrics from `SystemMetrics`. It aggregates the existing
CPU counters, computes CPU busy ratio from consecutive samples, and
treats `iowait` as idle time for this decision path.
---
be/src/common/metrics/system_metrics.cpp | 25 ++++++++
be/src/common/metrics/system_metrics.h | 11 ++++
be/src/storage/adaptive_thread_pool_controller.cpp | 35 +++++++++--
be/src/storage/adaptive_thread_pool_controller.h | 9 ++-
.../adaptive_thread_pool_controller_test.cpp | 69 ++++++++++++++++++++++
be/test/util/test_data/stat_cpu_busy_after | 1 +
be/test/util/test_data/stat_cpu_busy_before | 1 +
be/test/util/test_data/stat_cpu_iowait_after | 1 +
be/test/util/test_data/stat_cpu_recovery_after | 1 +
be/test/util/test_data/stat_cpu_regressed_after | 1 +
10 files changed, 148 insertions(+), 6 deletions(-)
diff --git a/be/src/common/metrics/system_metrics.cpp b/be/src/common/metrics/system_metrics.cpp
index 3dfd6186443..425abdd8d93 100644
--- a/be/src/common/metrics/system_metrics.cpp
+++ b/be/src/common/metrics/system_metrics.cpp
@@ -467,6 +467,17 @@ void SystemMetrics::_update_cpu_metrics() {
continue;
}
+ if (cpu_name == "cpu") {
+ AggregateCpuTime aggregate_cpu_time;
+ aggregate_cpu_time.total_time = values[0] + values[1] + values[2] + values[3] +
+ values[4] + values[5] + values[6] + values[7];
+ aggregate_cpu_time.idle_time = values[3] + values[4];
+ aggregate_cpu_time.initialized = aggregate_cpu_time.total_time > 0;
+ // Publish a consistent aggregate snapshot derived from one /proc/stat row.
+ std::lock_guard<std::mutex> lk(_aggregate_cpu_time_mutex);
+ _aggregate_cpu_time = aggregate_cpu_time;
+ }
+
for (int i = 0; i < CpuMetrics::cpu_num_metrics; ++i) {
it->second->metrics[i]->set_value(values[i]);
}
@@ -908,6 +919,20 @@ double SystemMetrics::get_load_average_1_min() {
}
}
+bool SystemMetrics::get_aggregate_cpu_time(int64_t* total_time, int64_t* idle_time) const {
+ DCHECK(total_time != nullptr);
+ DCHECK(idle_time != nullptr);
+
+ std::lock_guard<std::mutex> lk(_aggregate_cpu_time_mutex);
+ if (!_aggregate_cpu_time.initialized) {
+ return false;
+ }
+
+ *total_time = _aggregate_cpu_time.total_time;
+ *idle_time = _aggregate_cpu_time.idle_time;
+ return true;
+}
+
void SystemMetrics::get_network_traffic(std::map<std::string, int64_t>* send_map,
std::map<std::string, int64_t>* rcv_map) {
send_map->clear();
diff --git a/be/src/common/metrics/system_metrics.h b/be/src/common/metrics/system_metrics.h
index 52ca7299f70..a5b4ccceb7e 100644
--- a/be/src/common/metrics/system_metrics.h
+++ b/be/src/common/metrics/system_metrics.h
@@ -22,6 +22,7 @@
#include <map>
#include <memory>
+#include <mutex>
#include <set>
#include <string>
#include <vector>
@@ -59,6 +60,7 @@ public:
int64_t interval_sec, int64_t* send_rate, int64_t* rcv_rate);
double get_load_average_1_min();
+ bool get_aggregate_cpu_time(int64_t* total_time, int64_t* idle_time) const;
void update_max_disk_io_util_percent(const std::map<std::string, int64_t>& lst_value,
int64_t interval_sec);
@@ -99,6 +101,12 @@ private:
void get_cpu_name();
private:
+ struct AggregateCpuTime {
+ int64_t total_time = 0;
+ int64_t idle_time = 0;
+ bool initialized = false;
+ };
+
static const char* _s_hook_name;
std::map<std::string, CpuMetrics*> _cpu_metrics;
@@ -118,6 +126,9 @@ private:
MetricRegistry* _registry = nullptr;
std::shared_ptr<MetricEntity> _server_entity;
+ mutable std::mutex _aggregate_cpu_time_mutex;
+ AggregateCpuTime _aggregate_cpu_time;
+
IntGauge* max_disk_io_util_percent = nullptr;
IntGauge* max_network_send_bytes_rate = nullptr;
IntGauge* max_network_receive_bytes_rate = nullptr;
diff --git a/be/src/storage/adaptive_thread_pool_controller.cpp b/be/src/storage/adaptive_thread_pool_controller.cpp
index 8b8de54654d..bcb7b089389 100644
--- a/be/src/storage/adaptive_thread_pool_controller.cpp
+++ b/be/src/storage/adaptive_thread_pool_controller.cpp
@@ -252,6 +252,8 @@ int AdaptiveThreadPoolController::get_current_threads(const std::string& name) c
}
bool AdaptiveThreadPoolController::is_io_busy() {
+ std::lock_guard<std::mutex> lk(_metrics_state_mutex);
+
if (config::is_cloud_mode()) {
if (_s3_file_upload_pool == nullptr) return false;
int queue_size = _s3_file_upload_pool->get_queue_size();
@@ -275,14 +277,37 @@ bool AdaptiveThreadPoolController::is_io_busy() {
}
bool AdaptiveThreadPoolController::is_cpu_busy() {
+ std::lock_guard<std::mutex> lk(_metrics_state_mutex);
+
if (_system_metrics == nullptr) return false;
- double load_avg = _system_metrics->get_load_average_1_min();
- int num_cpus = std::thread::hardware_concurrency();
- if (num_cpus <= 0) return false;
+ int64_t total_time = 0;
+ int64_t idle_time = 0;
+ if (!_system_metrics->get_aggregate_cpu_time(&total_time, &idle_time)) {
+ return _last_cpu_busy;
+ }
+
+ if (_last_cpu_total_time < 0) {
+ _last_cpu_total_time = total_time;
+ _last_cpu_idle_time = idle_time;
+ _last_cpu_busy = false;
+ return false;
+ }
+
+ int64_t total_time_delta = total_time - _last_cpu_total_time;
+ int64_t idle_time_delta = idle_time - _last_cpu_idle_time;
+ if (total_time_delta <= 0 || idle_time_delta < 0 || idle_time_delta > total_time_delta) {
+ // Keep the previous baseline so a transient invalid sample does not
+ // poison the next interval's delta calculation.
+ return _last_cpu_busy;
+ }
- double cpu_usage_percent = (load_avg / num_cpus) * 100.0;
- return cpu_usage_percent > kCPUBusyThresholdPercent;
+ _last_cpu_total_time = total_time;
+ _last_cpu_idle_time = idle_time;
+ double cpu_busy_percent =
+ static_cast<double>(total_time_delta - idle_time_delta) * 100.0 / total_time_delta;
+ _last_cpu_busy = cpu_busy_percent > kCPUBusyThresholdPercent;
+ return _last_cpu_busy;
}
AdaptiveThreadPoolController::AdjustFunc AdaptiveThreadPoolController::make_flush_adjust_func(
diff --git a/be/src/storage/adaptive_thread_pool_controller.h b/be/src/storage/adaptive_thread_pool_controller.h
index 4a6f0096f62..0598531268c 100644
--- a/be/src/storage/adaptive_thread_pool_controller.h
+++ b/be/src/storage/adaptive_thread_pool_controller.h
@@ -56,7 +56,7 @@ struct TimerArg {
};
// AdaptiveThreadPoolController dynamically adjusts thread pool sizes based on
-// system load (IO utilisation, CPU load average, flush queue depth).
+// system load (IO utilisation, CPU utilisation, flush queue depth).
//
// Each registered pool group runs as a one-shot bthread_timer_add chain: the
// callback fires, adjusts the pool, then re-registers the next one-shot timer.
@@ -142,6 +142,7 @@ private:
ThreadPool* _s3_file_upload_pool = nullptr;
mutable std::mutex _mutex;
+ mutable std::mutex _metrics_state_mutex;
std::map<std::string, PoolGroup> _pool_groups;
// Last successfully computed IO-busy result. Returned as-is when the
@@ -151,6 +152,12 @@ private:
// For disk IO util calculation (used by is_io_busy).
std::map<std::string, int64_t> _last_disk_io_time;
int64_t _last_check_time_sec = 0;
+
+ // For CPU util calculation (used by is_cpu_busy). The counters come from
+ // SystemMetrics' existing cpu_* metrics and are compared as deltas.
+ bool _last_cpu_busy = false;
+ int64_t _last_cpu_total_time = -1;
+ int64_t _last_cpu_idle_time = -1;
};
} // namespace doris
diff --git a/be/test/storage/adaptive_thread_pool_controller_test.cpp b/be/test/storage/adaptive_thread_pool_controller_test.cpp
index b2d6464fcef..06d79629330 100644
--- a/be/test/storage/adaptive_thread_pool_controller_test.cpp
+++ b/be/test/storage/adaptive_thread_pool_controller_test.cpp
@@ -22,10 +22,23 @@
#include <thread>
#include "common/config.h"
+#include "common/metrics/metrics.h"
+#include "common/metrics/system_metrics.h"
+#include "testutil/test_util.h"
#include "util/threadpool.h"
namespace doris {
+extern const char* k_ut_stat_path;
+
+namespace {
+
+std::string get_stat_test_data_path(const std::string& file_name) {
+ return GetCurrentRunningDir() + "/util/test_data/" + file_name;
+}
+
+} // namespace
+
class AdaptiveThreadPoolControllerTest : public testing::Test {
protected:
void SetUp() override {
@@ -242,6 +255,62 @@ TEST_F(AdaptiveThreadPoolControllerTest, TestIoBusyCpuBusyWithNullMetrics) {
EXPECT_FALSE(controller.is_cpu_busy());
}
+TEST_F(AdaptiveThreadPoolControllerTest, TestCpuBusyUsesCpuMetricsDelta) {
+ MetricRegistry registry("test");
+ const std::string before_path = get_stat_test_data_path("stat_cpu_busy_before");
+ const std::string after_path = get_stat_test_data_path("stat_cpu_busy_after");
+ k_ut_stat_path = before_path.c_str();
+ SystemMetrics metrics(®istry, {}, {});
+ metrics.update();
+
+ AdaptiveThreadPoolController controller;
+ controller.init(&metrics, nullptr);
+
+ EXPECT_FALSE(controller.is_cpu_busy());
+ k_ut_stat_path = after_path.c_str();
+ metrics.update();
+ EXPECT_TRUE(controller.is_cpu_busy());
+}
+
+TEST_F(AdaptiveThreadPoolControllerTest, TestCpuBusyTreatsIoWaitAsIdle) {
+ MetricRegistry registry("test");
+ const std::string before_path = get_stat_test_data_path("stat_cpu_busy_before");
+ const std::string after_path = get_stat_test_data_path("stat_cpu_iowait_after");
+ k_ut_stat_path = before_path.c_str();
+ SystemMetrics metrics(®istry, {}, {});
+ metrics.update();
+
+ AdaptiveThreadPoolController controller;
+ controller.init(&metrics, nullptr);
+
+ EXPECT_FALSE(controller.is_cpu_busy());
+ k_ut_stat_path = after_path.c_str();
+ metrics.update();
+ EXPECT_FALSE(controller.is_cpu_busy());
+}
+
+TEST_F(AdaptiveThreadPoolControllerTest, TestCpuBusyInvalidSampleDoesNotAdvanceBaseline) {
+ MetricRegistry registry("test");
+ const std::string before_path = get_stat_test_data_path("stat_cpu_busy_before");
+ const std::string invalid_path = get_stat_test_data_path("stat_cpu_regressed_after");
+ const std::string recovery_path = get_stat_test_data_path("stat_cpu_recovery_after");
+ k_ut_stat_path = before_path.c_str();
+ SystemMetrics metrics(®istry, {}, {});
+ metrics.update();
+
+ AdaptiveThreadPoolController controller;
+ controller.init(&metrics, nullptr);
+
+ EXPECT_FALSE(controller.is_cpu_busy());
+ k_ut_stat_path = invalid_path.c_str();
+ metrics.update();
+ EXPECT_FALSE(controller.is_cpu_busy());
+
+ k_ut_stat_path = recovery_path.c_str();
+ metrics.update();
+ EXPECT_TRUE(controller.is_cpu_busy());
+}
+
// Test adjust function that uses controller's is_io_busy/is_cpu_busy
TEST_F(AdaptiveThreadPoolControllerTest, TestAdjustFuncWithControllerMethods) {
config::enable_adaptive_flush_threads = true;
diff --git a/be/test/util/test_data/stat_cpu_busy_after b/be/test/util/test_data/stat_cpu_busy_after
new file mode 100644
index 00000000000..2b5027df63d
--- /dev/null
+++ b/be/test/util/test_data/stat_cpu_busy_after
@@ -0,0 +1 @@
+cpu 250 0 200 820 0 0 0 0 0 0
diff --git a/be/test/util/test_data/stat_cpu_busy_before b/be/test/util/test_data/stat_cpu_busy_before
new file mode 100644
index 00000000000..bcf77b76c25
--- /dev/null
+++ b/be/test/util/test_data/stat_cpu_busy_before
@@ -0,0 +1 @@
+cpu 100 0 100 800 0 0 0 0 0 0
diff --git a/be/test/util/test_data/stat_cpu_iowait_after b/be/test/util/test_data/stat_cpu_iowait_after
new file mode 100644
index 00000000000..c58b33ea7ca
--- /dev/null
+++ b/be/test/util/test_data/stat_cpu_iowait_after
@@ -0,0 +1 @@
+cpu 120 0 110 810 260 0 0 0 0 0
diff --git a/be/test/util/test_data/stat_cpu_recovery_after b/be/test/util/test_data/stat_cpu_recovery_after
new file mode 100644
index 00000000000..3ecc01524b6
--- /dev/null
+++ b/be/test/util/test_data/stat_cpu_recovery_after
@@ -0,0 +1 @@
+cpu 200 0 200 819 0 0 0 0 0 0
diff --git a/be/test/util/test_data/stat_cpu_regressed_after b/be/test/util/test_data/stat_cpu_regressed_after
new file mode 100644
index 00000000000..111f0f30706
--- /dev/null
+++ b/be/test/util/test_data/stat_cpu_regressed_after
@@ -0,0 +1 @@
+cpu 90 0 90 770 0 0 0 0 0 0
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org