diff --git a/be/src/cloud/cloud_delta_writer.cpp b/be/src/cloud/cloud_delta_writer.cpp index 9e7641aca1fb4f..bb6fdcf26b9f67 100644 --- a/be/src/cloud/cloud_delta_writer.cpp +++ b/be/src/cloud/cloud_delta_writer.cpp @@ -99,8 +99,9 @@ Status CloudDeltaWriter::write(const Block* block, const TabletAddRowsPayload& r CHECK(_is_init || _is_cancelled); { SCOPED_TIMER(_wait_flush_limit_timer); - while (_memtable_writer->flush_running_count() >= - config::memtable_flush_running_count_limit) { + auto* adaptive_controller = + ExecEnv::GetInstance()->storage_engine().adaptive_thread_controller(); + while (adaptive_controller->is_io_busy()) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); } } diff --git a/be/src/load/delta_writer/delta_writer.cpp b/be/src/load/delta_writer/delta_writer.cpp index d831df4c8a3d90..abcda75afa4762 100644 --- a/be/src/load/delta_writer/delta_writer.cpp +++ b/be/src/load/delta_writer/delta_writer.cpp @@ -178,8 +178,10 @@ Status DeltaWriter::write(const Block* block, const TabletAddRowsPayload& rows, } { SCOPED_TIMER(_wait_flush_limit_timer); - while (_memtable_writer->flush_running_count() >= - config::memtable_flush_running_count_limit) { + const auto effective_flush_running_count_limit = + config::memtable_flush_running_count_limit * + (_req.write_req_type == WriteRequestType::GROUP ? 2 : 1); + while (_memtable_writer->flush_running_count() >= effective_flush_running_count_limit) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); } } diff --git a/be/src/storage/storage_engine.cpp b/be/src/storage/storage_engine.cpp index 2093a17e663ae8..80f92e4b14435f 100644 --- a/be/src/storage/storage_engine.cpp +++ b/be/src/storage/storage_engine.cpp @@ -140,15 +140,17 @@ int64_t BaseStorageEngine::memory_limitation_bytes_per_thread_for_schema_change( } void BaseStorageEngine::_start_adaptive_thread_controller() { - if (!config::enable_adaptive_flush_threads) { - return; - } - auto* system_metrics = DorisMetrics::instance()->system_metrics(); auto* s3_upload_pool = ExecEnv::GetInstance()->s3_file_upload_thread_pool(); _adaptive_thread_controller.init(system_metrics, s3_upload_pool); + // Cloud write backpressure also uses is_io_busy(), so initialize the IO signal even when + // adaptive flush-thread adjustment is disabled. + if (!config::enable_adaptive_flush_threads) { + return; + } + if (_memtable_flush_executor) { auto* flush_pool = _memtable_flush_executor->flush_pool(); auto* high_prio_pool = _memtable_flush_executor->high_prio_flush_pool(); diff --git a/be/test/storage/adaptive_thread_pool_controller_test.cpp b/be/test/storage/adaptive_thread_pool_controller_test.cpp index 6c4e42fea05897..38767812d48cda 100644 --- a/be/test/storage/adaptive_thread_pool_controller_test.cpp +++ b/be/test/storage/adaptive_thread_pool_controller_test.cpp @@ -22,10 +22,12 @@ #include #include +#include "cloud/config.h" #include "common/config.h" #include "common/metrics/metrics.h" #include "common/metrics/system_metrics.h" #include "testutil/test_util.h" +#include "util/countdown_latch.h" #include "util/threadpool.h" namespace doris { @@ -44,6 +46,8 @@ class AdaptiveThreadPoolControllerTest : public testing::Test { protected: void SetUp() override { _original_enable_adaptive = config::enable_adaptive_flush_threads; + _original_deploy_mode = config::deploy_mode; + _original_cloud_unique_id = config::cloud_unique_id; int num_cpus = std::thread::hardware_concurrency(); if (num_cpus <= 0) num_cpus = 1; @@ -64,11 +68,15 @@ class AdaptiveThreadPoolControllerTest : public testing::Test { void TearDown() override { config::enable_adaptive_flush_threads = _original_enable_adaptive; + config::deploy_mode = _original_deploy_mode; + config::cloud_unique_id = _original_cloud_unique_id; if (_pool) _pool->shutdown(); if (_pool2) _pool2->shutdown(); } bool _original_enable_adaptive; + std::string _original_deploy_mode; + std::string _original_cloud_unique_id; std::unique_ptr _pool; std::unique_ptr _pool2; }; @@ -260,6 +268,44 @@ TEST_F(AdaptiveThreadPoolControllerTest, TestIoBusyCpuBusyWithNullMetrics) { EXPECT_FALSE(controller.is_cpu_busy()); } +TEST_F(AdaptiveThreadPoolControllerTest, TestCloudIoBusyUsesS3UploadQueue) { + config::deploy_mode = "cloud"; + config::cloud_unique_id.clear(); + + std::unique_ptr upload_pool; + ASSERT_TRUE(ThreadPoolBuilder("TestS3UploadPool") + .set_min_threads(1) + .set_max_threads(1) + .set_max_queue_size(AdaptiveThreadPoolController::kS3QueueBusyThreshold + 1) + .build(&upload_pool) + .ok()); + + CountDownLatch task_started(1); + CountDownLatch release_task(1); + ASSERT_TRUE(upload_pool + ->submit_func([&]() { + task_started.count_down(); + release_task.wait(); + }) + .ok()); + EXPECT_TRUE(task_started.wait_for(std::chrono::seconds(5))); + + for (int i = 0; i < AdaptiveThreadPoolController::kS3QueueBusyThreshold; ++i) { + EXPECT_TRUE(upload_pool->submit_func([]() {}).ok()); + } + + AdaptiveThreadPoolController controller; + controller.init(nullptr, upload_pool.get()); + EXPECT_FALSE(controller.is_io_busy()); + + EXPECT_TRUE(upload_pool->submit_func([]() {}).ok()); + EXPECT_TRUE(controller.is_io_busy()); + + release_task.count_down(); + upload_pool->shutdown(); + upload_pool->wait(); +} + TEST_F(AdaptiveThreadPoolControllerTest, TestCpuBusyUsesCpuMetricsDelta) { MetricRegistry registry("test"); const std::string before_path = get_stat_test_data_path("stat_cpu_busy_before");