You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by da...@apache.org on 2024/07/22 01:16:07 UTC
(doris) 19/26: [Fix]add min scan thread num for workload group's scan thread (#38096)
This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 7d70c1739c0bae681c01fb544e62f9eabaedc2d5
Author: wangbo <wa...@apache.org>
AuthorDate: Fri Jul 19 18:05:17 2024 +0800
[Fix]add min scan thread num for workload group's scan thread (#38096)
## Proposed changes
Set workload group's and non-workload group's remote scan min thread to
reduce thread num, prevent Be core for thread Exhaustion.
before:
<img width="582" alt="image"
src="https://github.com/user-attachments/assets/3a861191-c5a9-4b73-8a08-0aec0bed1cd5">
after:
<img width="522" alt="image"
src="https://github.com/user-attachments/assets/4024bbc8-d9d3-45bd-a895-07a6d87a6fd8">
---
be/src/common/config.cpp | 1 +
be/src/common/config.h | 1 +
be/src/runtime/workload_group/workload_group.cpp | 5 +-
be/src/util/s3_util.cpp | 2 +-
be/src/vec/exec/scan/scanner_scheduler.cpp | 60 +++++++++---------------
be/src/vec/exec/scan/scanner_scheduler.h | 13 +++--
6 files changed, 38 insertions(+), 44 deletions(-)
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index b152111011e..5222100170e 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -250,6 +250,7 @@ DEFINE_Validator(doris_scanner_thread_pool_thread_num, [](const int config) -> b
}
return true;
});
+DEFINE_Int32(doris_scanner_min_thread_pool_thread_num, "8");
DEFINE_Int32(remote_split_source_batch_size, "10240");
DEFINE_Int32(doris_max_remote_scanner_thread_pool_thread_num, "-1");
// number of olap scanner thread pool queue size
diff --git a/be/src/common/config.h b/be/src/common/config.h
index f4ed1decaa0..53261ab2fb9 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -299,6 +299,7 @@ DECLARE_mInt64(doris_blocking_priority_queue_wait_timeout_ms);
// number of scanner thread pool size for olap table
// and the min thread num of remote scanner thread pool
DECLARE_mInt32(doris_scanner_thread_pool_thread_num);
+DECLARE_mInt32(doris_scanner_min_thread_pool_thread_num);
// number of batch size to fetch the remote split source
DECLARE_mInt32(remote_split_source_batch_size);
// max number of remote scanner thread pool size
diff --git a/be/src/runtime/workload_group/workload_group.cpp b/be/src/runtime/workload_group/workload_group.cpp
index fd45093758e..f4d1e0d4f7e 100644
--- a/be/src/runtime/workload_group/workload_group.cpp
+++ b/be/src/runtime/workload_group/workload_group.cpp
@@ -316,7 +316,7 @@ WorkloadGroupInfo WorkloadGroupInfo::parse_topic_info(
}
// 11 min remote scan thread num
- int min_remote_scan_thread_num = vectorized::ScannerScheduler::get_remote_scan_thread_num();
+ int min_remote_scan_thread_num = config::doris_scanner_min_thread_pool_thread_num;
if (tworkload_group_info.__isset.min_remote_scan_thread_num &&
tworkload_group_info.min_remote_scan_thread_num > 0) {
min_remote_scan_thread_num = tworkload_group_info.min_remote_scan_thread_num;
@@ -415,7 +415,8 @@ void WorkloadGroup::upsert_task_scheduler(WorkloadGroupInfo* tg_info, ExecEnv* e
std::unique_ptr<vectorized::SimplifiedScanScheduler> remote_scan_scheduler =
std::make_unique<vectorized::SimplifiedScanScheduler>("RScan_" + tg_name,
cg_cpu_ctl_ptr);
- Status ret = remote_scan_scheduler->start(remote_max_thread_num, remote_max_thread_num,
+ Status ret = remote_scan_scheduler->start(remote_max_thread_num,
+ config::doris_scanner_min_thread_pool_thread_num,
remote_scan_thread_queue_size);
if (ret.ok()) {
_remote_scan_task_sched = std::move(remote_scan_scheduler);
diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp
index d7a83fa2cff..ffb93c2d9d9 100644
--- a/be/src/util/s3_util.cpp
+++ b/be/src/util/s3_util.cpp
@@ -257,7 +257,7 @@ std::shared_ptr<io::ObjStorageClient> S3ClientFactory::_create_s3_client(
aws_config.maxConnections = config::doris_scanner_thread_pool_thread_num;
#else
aws_config.maxConnections =
- ExecEnv::GetInstance()->scanner_scheduler()->remote_thread_pool_max_size();
+ ExecEnv::GetInstance()->scanner_scheduler()->remote_thread_pool_max_thread_num();
#endif
}
diff --git a/be/src/vec/exec/scan/scanner_scheduler.cpp b/be/src/vec/exec/scan/scanner_scheduler.cpp
index 4d07e66917d..351f5d4e275 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.cpp
+++ b/be/src/vec/exec/scan/scanner_scheduler.cpp
@@ -80,28 +80,33 @@ void ScannerScheduler::stop() {
_is_closed = true;
- _local_scan_thread_pool->shutdown();
- _remote_scan_thread_pool->shutdown();
_limited_scan_thread_pool->shutdown();
-
- _local_scan_thread_pool->join();
- _remote_scan_thread_pool->join();
_limited_scan_thread_pool->wait();
+ _local_scan_thread_pool->stop();
+ _remote_scan_thread_pool->stop();
+
LOG(INFO) << "ScannerScheduler stopped";
}
Status ScannerScheduler::init(ExecEnv* env) {
// 1. local scan thread pool
- _local_scan_thread_pool = std::make_unique<PriorityThreadPool>(
- config::doris_scanner_thread_pool_thread_num,
- config::doris_scanner_thread_pool_queue_size, "local_scan");
+ _local_scan_thread_pool =
+ std::make_unique<vectorized::SimplifiedScanScheduler>("local_scan", nullptr);
+ Status ret1 = _local_scan_thread_pool->start(config::doris_scanner_thread_pool_thread_num,
+ config::doris_scanner_thread_pool_thread_num,
+ config::doris_scanner_thread_pool_queue_size);
+ RETURN_IF_ERROR(ret1);
// 2. remote scan thread pool
- _remote_thread_pool_max_size = ScannerScheduler::get_remote_scan_thread_num();
+ _remote_thread_pool_max_thread_num = ScannerScheduler::get_remote_scan_thread_num();
int remote_scan_pool_queue_size = ScannerScheduler::get_remote_scan_thread_queue_size();
- _remote_scan_thread_pool = std::make_unique<PriorityThreadPool>(
- _remote_thread_pool_max_size, remote_scan_pool_queue_size, "RemoteScanThreadPool");
+ _remote_scan_thread_pool =
+ std::make_unique<vectorized::SimplifiedScanScheduler>("RemoteScanThreadPool", nullptr);
+ Status ret2 = _remote_scan_thread_pool->start(_remote_thread_pool_max_thread_num,
+ config::doris_scanner_min_thread_pool_thread_num,
+ remote_scan_pool_queue_size);
+ RETURN_IF_ERROR(ret2);
// 3. limited scan thread pool
RETURN_IF_ERROR(ThreadPoolBuilder("LimitedScanThreadPool")
@@ -127,9 +132,6 @@ void ScannerScheduler::submit(std::shared_ptr<ScannerContext> ctx,
return;
}
- // Submit scanners to thread pool
- // TODO(cmy): How to handle this "nice"?
- int nice = 1;
if (ctx->thread_token != nullptr) {
std::shared_ptr<ScannerDelegate> scanner_delegate = scan_task->scanner.lock();
if (scanner_delegate == nullptr) {
@@ -163,27 +165,13 @@ void ScannerScheduler::submit(std::shared_ptr<ScannerContext> ctx,
TabletStorageType type = scanner_delegate->_scanner->get_storage_type();
auto sumbit_task = [&]() {
bool is_local = type == TabletStorageType::STORAGE_TYPE_LOCAL;
- auto* scan_sched =
+ SimplifiedScanScheduler* scan_sched =
is_local ? ctx->get_simple_scan_scheduler() : ctx->get_remote_scan_scheduler();
- auto& thread_pool = is_local ? _local_scan_thread_pool : _remote_scan_thread_pool;
- if (scan_sched) {
- auto work_func = [scanner_ref = scan_task, ctx]() {
- auto status = [&] {
- RETURN_IF_CATCH_EXCEPTION(_scanner_scan(ctx, scanner_ref));
- return Status::OK();
- }();
-
- if (!status.ok()) {
- scanner_ref->set_status(status);
- ctx->append_block_to_queue(scanner_ref);
- }
- };
- SimplifiedScanTask simple_scan_task = {work_func, ctx};
- return scan_sched->submit_scan_task(simple_scan_task);
+ if (!scan_sched) { // query without workload group
+ scan_sched =
+ is_local ? _local_scan_thread_pool.get() : _remote_scan_thread_pool.get();
}
-
- PriorityThreadPool::Task task;
- task.work_function = [scanner_ref = scan_task, ctx]() {
+ auto work_func = [scanner_ref = scan_task, ctx]() {
auto status = [&] {
RETURN_IF_CATCH_EXCEPTION(_scanner_scan(ctx, scanner_ref));
return Status::OK();
@@ -194,10 +182,8 @@ void ScannerScheduler::submit(std::shared_ptr<ScannerContext> ctx,
ctx->append_block_to_queue(scanner_ref);
}
};
- task.priority = nice;
- return thread_pool->offer(task)
- ? Status::OK()
- : Status::InternalError("Scan thread pool had shutdown");
+ SimplifiedScanTask simple_scan_task = {work_func, ctx};
+ return scan_sched->submit_scan_task(simple_scan_task);
};
if (auto ret = sumbit_task(); !ret) {
diff --git a/be/src/vec/exec/scan/scanner_scheduler.h b/be/src/vec/exec/scan/scanner_scheduler.h
index 238afc15bf6..ddc61396e23 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.h
+++ b/be/src/vec/exec/scan/scanner_scheduler.h
@@ -39,6 +39,7 @@ namespace doris::vectorized {
class ScannerDelegate;
class ScanTask;
class ScannerContext;
+class SimplifiedScanScheduler;
// Responsible for the scheduling and execution of all Scanners of a BE node.
// Execution thread pool
@@ -63,7 +64,7 @@ public:
std::unique_ptr<ThreadPoolToken> new_limited_scan_pool_token(ThreadPool::ExecutionMode mode,
int max_concurrency);
- int remote_thread_pool_max_size() const { return _remote_thread_pool_max_size; }
+ int remote_thread_pool_max_thread_num() const { return _remote_thread_pool_max_thread_num; }
static int get_remote_scan_thread_num();
@@ -81,14 +82,14 @@ private:
// _local_scan_thread_pool is for local scan task(typically, olap scanner)
// _remote_scan_thread_pool is for remote scan task(cold data on s3, hdfs, etc.)
// _limited_scan_thread_pool is a special pool for queries with resource limit
- std::unique_ptr<PriorityThreadPool> _local_scan_thread_pool;
- std::unique_ptr<PriorityThreadPool> _remote_scan_thread_pool;
+ std::unique_ptr<vectorized::SimplifiedScanScheduler> _local_scan_thread_pool;
+ std::unique_ptr<vectorized::SimplifiedScanScheduler> _remote_scan_thread_pool;
std::unique_ptr<ThreadPool> _limited_scan_thread_pool;
// true is the scheduler is closed.
std::atomic_bool _is_closed = {false};
bool _is_init = false;
- int _remote_thread_pool_max_size;
+ int _remote_thread_pool_max_thread_num;
};
struct SimplifiedScanTask {
@@ -193,6 +194,10 @@ public:
}
}
+ int get_queue_size() { return _scan_thread_pool->get_queue_size(); }
+
+ int get_active_threads() { return _scan_thread_pool->num_active_threads(); }
+
std::vector<int> thread_debug_info() { return _scan_thread_pool->debug_info(); }
private:
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org