You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "wangbo (via GitHub)" <gi...@apache.org> on 2024/02/25 05:57:16 UTC
[PR] [Improvement](executor)add remote scan thread pool [doris]
wangbo opened a new pull request, #31376:
URL: https://github.com/apache/doris/pull/31376
## Proposed changes
add remote scan thread pool for workload group.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on PR #31376:
URL: https://github.com/apache/doris/pull/31376#issuecomment-1963247726
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #31376:
URL: https://github.com/apache/doris/pull/31376#discussion_r1501755957
##########
be/src/runtime/task_group/task_group.h:
##########
@@ -121,6 +129,15 @@ class TaskGroup : public std::enable_shared_from_this<TaskGroup> {
int64_t gc_memory(int64_t need_free_mem, RuntimeProfile* profile);
+ void upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env);
+
+ void get_query_scheduler(doris::pipeline::TaskScheduler** exec_sched,
+ vectorized::SimplifiedScanScheduler** scan_sched,
+ ThreadPool** non_pipe_thread_pool,
+ vectorized::SimplifiedScanScheduler** remote_scan_sched);
+
+ void try_stop_task_scheduler();
Review Comment:
rename to try_stop_schedulers. because task_scheduler is for Pipeline task scheduler. It is very ambiguous
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #31376:
URL: https://github.com/apache/doris/pull/31376#issuecomment-1963305743
TeamCity be ut coverage result:
Function Coverage: 35.72% (8548/23929)
Line Coverage: 27.53% (69362/251973)
Region Coverage: 26.67% (35970/134855)
Branch Coverage: 23.48% (18385/78304)
Coverage Report: http://coverage.selectdb-in.cc/coverage/17e6bfc64aa051fdecbd202abb52a551dc52f780_17e6bfc64aa051fdecbd202abb52a551dc52f780/report/index.html
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #31376:
URL: https://github.com/apache/doris/pull/31376#discussion_r1501765685
##########
be/src/runtime/task_group/task_group.cpp:
##########
@@ -266,8 +273,167 @@ Status TaskGroupInfo::parse_topic_info(const TWorkloadGroupInfo& workload_group_
task_group_info->scan_thread_num = workload_group_info.scan_thread_num;
}
+ // 10 max remote scan thread num
+ task_group_info->max_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.max_remote_scan_thread_num &&
+ workload_group_info.max_remote_scan_thread_num > 0) {
+ task_group_info->max_remote_scan_thread_num =
+ workload_group_info.max_remote_scan_thread_num;
+ }
+
+ // 11 min remote scan thread num
+ task_group_info->min_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.min_remote_scan_thread_num &&
+ workload_group_info.min_remote_scan_thread_num > 0) {
+ task_group_info->min_remote_scan_thread_num =
+ workload_group_info.min_remote_scan_thread_num;
+ }
+
return Status::OK();
}
+void TaskGroup::upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env) {
+ uint64_t tg_id = tg_info->id;
+ std::string tg_name = tg_info->name;
+ int cpu_hard_limit = tg_info->cpu_hard_limit;
+ uint64_t cpu_shares = tg_info->cpu_share;
+ bool enable_cpu_hard_limit = tg_info->enable_cpu_hard_limit;
+ int scan_thread_num = tg_info->scan_thread_num;
+ int max_remote_scan_thread_num = tg_info->max_remote_scan_thread_num;
+ int min_remote_scan_thread_num = tg_info->min_remote_scan_thread_num;
+
+ std::lock_guard<std::shared_mutex> wlock(_task_sched_lock);
+ if (config::doris_cgroup_cpu_path != "" && _cgroup_cpu_ctl == nullptr) {
+ std::unique_ptr<CgroupCpuCtl> cgroup_cpu_ctl = std::make_unique<CgroupV1CpuCtl>(tg_id);
+ Status ret = cgroup_cpu_ctl->init();
+ if (ret.ok()) {
+ _cgroup_cpu_ctl = std::move(cgroup_cpu_ctl);
+ LOG(INFO) << "[upsert wg thread pool] cgroup init success";
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] cgroup init failed, gid= " << tg_id
+ << ", reason=" << ret.to_string();
+ }
+ }
+
+ CgroupCpuCtl* cg_cpu_ctl_ptr = _cgroup_cpu_ctl.get();
+
+ if (_task_sched == nullptr) {
+ int32_t executors_size = config::pipeline_executor_size;
+ if (executors_size <= 0) {
+ executors_size = CpuInfo::num_cores();
+ }
+ auto task_queue = std::make_shared<pipeline::MultiCoreTaskQueue>(executors_size);
+ std::unique_ptr<pipeline::TaskScheduler> pipeline_task_scheduler =
+ std::make_unique<pipeline::TaskScheduler>(
+ exec_env, exec_env->get_global_block_scheduler(), std::move(task_queue),
+ "Pipe_" + tg_name, cg_cpu_ctl_ptr);
+ Status ret = pipeline_task_scheduler->start();
+ if (ret.ok()) {
+ _task_sched = std::move(pipeline_task_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] task scheduler start failed, gid= " << tg_id;
+ }
+ }
+
+ if (_scan_task_sched == nullptr) {
+ std::unique_ptr<vectorized::SimplifiedScanScheduler> scan_scheduler =
+ std::make_unique<vectorized::SimplifiedScanScheduler>("Scan_" + tg_name,
+ cg_cpu_ctl_ptr);
+ Status ret = scan_scheduler->start();
+ if (ret.ok()) {
+ _scan_task_sched = std::move(scan_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] scan scheduler start failed, gid=" << tg_id;
+ }
+ }
+ if (scan_thread_num > 0 && _scan_task_sched) {
+ _scan_task_sched->reset_thread_num(scan_thread_num);
+ }
+
+ if (_remote_scan_task_sched == nullptr) {
+ 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();
+ if (ret.ok()) {
+ _remote_scan_task_sched = std::move(remote_scan_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] remote scan scheduler start failed, gid="
+ << tg_id;
+ }
+ }
+ if (max_remote_scan_thread_num > 0 && _remote_scan_task_sched) {
+ _remote_scan_task_sched->reset_max_thread_num(max_remote_scan_thread_num);
+ }
+ if (min_remote_scan_thread_num > 0 && _remote_scan_task_sched) {
+ _remote_scan_task_sched->reset_min_thread_num(min_remote_scan_thread_num);
+ }
+
+ if (_non_pipe_thread_pool == nullptr) {
+ std::unique_ptr<ThreadPool> thread_pool = nullptr;
+ auto ret = ThreadPoolBuilder("nonPip_" + tg_name)
+ .set_min_threads(1)
+ .set_max_threads(config::fragment_pool_thread_num_max)
+ .set_max_queue_size(config::fragment_pool_queue_size)
+ .set_cgroup_cpu_ctl(cg_cpu_ctl_ptr)
+ .build(&thread_pool);
+ if (!ret.ok()) {
+ LOG(INFO) << "[upsert wg thread pool] create non-pipline thread pool failed, gid="
+ << tg_id;
+ } else {
+ _non_pipe_thread_pool = std::move(thread_pool);
+ }
+ }
+
+ // step 6: update cgroup cpu if needed
+ if (_cgroup_cpu_ctl) {
+ if (enable_cpu_hard_limit) {
+ if (cpu_hard_limit > 0) {
+ _cgroup_cpu_ctl->update_cpu_hard_limit(cpu_hard_limit);
+ _cgroup_cpu_ctl->update_cpu_soft_limit(CPU_SOFT_LIMIT_DEFAULT_VALUE);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] enable cpu hard limit but value is illegal: "
+ << cpu_hard_limit << ", gid=" << tg_id;
+ }
+ } else {
+ if (config::enable_cgroup_cpu_soft_limit) {
+ _cgroup_cpu_ctl->update_cpu_soft_limit(cpu_shares);
+ _cgroup_cpu_ctl->update_cpu_hard_limit(
+ CPU_HARD_LIMIT_DEFAULT_VALUE); // disable cpu hard limit
+ }
+ }
+ _cgroup_cpu_ctl->get_cgroup_cpu_info(&(tg_info->cgroup_cpu_shares),
+ &(tg_info->cgroup_cpu_hard_limit));
+ }
+}
+
+void TaskGroup::get_query_scheduler(doris::pipeline::TaskScheduler** exec_sched,
+ vectorized::SimplifiedScanScheduler** scan_sched,
+ ThreadPool** non_pipe_thread_pool,
+ vectorized::SimplifiedScanScheduler** remote_scan_sched) {
+ std::shared_lock<std::shared_mutex> rlock(_task_sched_lock);
+ *exec_sched = _task_sched.get();
+ *scan_sched = _scan_task_sched.get();
+ *remote_scan_sched = _remote_scan_task_sched.get();
+ *non_pipe_thread_pool = _non_pipe_thread_pool.get();
+}
+
+void TaskGroup::try_stop_schedulers() {
Review Comment:
warning: method 'try_stop_schedulers' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/task_group/task_group.h:138:
```diff
- void try_stop_schedulers();
+ static void try_stop_schedulers();
```
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #31376:
URL: https://github.com/apache/doris/pull/31376#discussion_r1501763845
##########
be/src/runtime/task_group/task_group.cpp:
##########
@@ -266,8 +273,167 @@ Status TaskGroupInfo::parse_topic_info(const TWorkloadGroupInfo& workload_group_
task_group_info->scan_thread_num = workload_group_info.scan_thread_num;
}
+ // 10 max remote scan thread num
+ task_group_info->max_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.max_remote_scan_thread_num &&
+ workload_group_info.max_remote_scan_thread_num > 0) {
+ task_group_info->max_remote_scan_thread_num =
+ workload_group_info.max_remote_scan_thread_num;
+ }
+
+ // 11 min remote scan thread num
+ task_group_info->min_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.min_remote_scan_thread_num &&
+ workload_group_info.min_remote_scan_thread_num > 0) {
+ task_group_info->min_remote_scan_thread_num =
+ workload_group_info.min_remote_scan_thread_num;
+ }
+
return Status::OK();
}
+void TaskGroup::upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env) {
+ uint64_t tg_id = tg_info->id;
+ std::string tg_name = tg_info->name;
+ int cpu_hard_limit = tg_info->cpu_hard_limit;
+ uint64_t cpu_shares = tg_info->cpu_share;
+ bool enable_cpu_hard_limit = tg_info->enable_cpu_hard_limit;
+ int scan_thread_num = tg_info->scan_thread_num;
+ int max_remote_scan_thread_num = tg_info->max_remote_scan_thread_num;
+ int min_remote_scan_thread_num = tg_info->min_remote_scan_thread_num;
+
+ std::lock_guard<std::shared_mutex> wlock(_task_sched_lock);
+ if (config::doris_cgroup_cpu_path != "" && _cgroup_cpu_ctl == nullptr) {
+ std::unique_ptr<CgroupCpuCtl> cgroup_cpu_ctl = std::make_unique<CgroupV1CpuCtl>(tg_id);
+ Status ret = cgroup_cpu_ctl->init();
+ if (ret.ok()) {
+ _cgroup_cpu_ctl = std::move(cgroup_cpu_ctl);
+ LOG(INFO) << "[upsert wg thread pool] cgroup init success";
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] cgroup init failed, gid= " << tg_id
+ << ", reason=" << ret.to_string();
+ }
+ }
+
+ CgroupCpuCtl* cg_cpu_ctl_ptr = _cgroup_cpu_ctl.get();
+
+ if (_task_sched == nullptr) {
+ int32_t executors_size = config::pipeline_executor_size;
+ if (executors_size <= 0) {
+ executors_size = CpuInfo::num_cores();
+ }
+ auto task_queue = std::make_shared<pipeline::MultiCoreTaskQueue>(executors_size);
+ std::unique_ptr<pipeline::TaskScheduler> pipeline_task_scheduler =
+ std::make_unique<pipeline::TaskScheduler>(
+ exec_env, exec_env->get_global_block_scheduler(), std::move(task_queue),
+ "Exec_" + tg_name, cg_cpu_ctl_ptr);
+ Status ret = pipeline_task_scheduler->start();
+ if (ret.ok()) {
+ _task_sched = std::move(pipeline_task_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] task scheduler start failed, gid= " << tg_id;
+ }
+ }
+
+ if (_scan_task_sched == nullptr) {
+ std::unique_ptr<vectorized::SimplifiedScanScheduler> scan_scheduler =
+ std::make_unique<vectorized::SimplifiedScanScheduler>("Scan_" + tg_name,
+ cg_cpu_ctl_ptr);
+ Status ret = scan_scheduler->start();
+ if (ret.ok()) {
+ _scan_task_sched = std::move(scan_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] scan scheduler start failed, gid=" << tg_id;
+ }
+ }
+ if (scan_thread_num > 0 && _scan_task_sched) {
+ _scan_task_sched->reset_thread_num(scan_thread_num);
+ }
+
+ if (_remote_scan_task_sched == nullptr) {
+ 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();
+ if (ret.ok()) {
+ _remote_scan_task_sched = std::move(remote_scan_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] remote scan scheduler start failed, gid="
+ << tg_id;
+ }
+ }
+ if (max_remote_scan_thread_num > 0 && _remote_scan_task_sched) {
+ _remote_scan_task_sched->reset_max_thread_num(max_remote_scan_thread_num);
+ }
+ if (min_remote_scan_thread_num > 0 && _remote_scan_task_sched) {
+ _remote_scan_task_sched->reset_min_thread_num(min_remote_scan_thread_num);
+ }
+
+ if (_non_pipe_thread_pool == nullptr) {
+ std::unique_ptr<ThreadPool> thread_pool = nullptr;
+ auto ret = ThreadPoolBuilder("nonPip_" + tg_name)
+ .set_min_threads(1)
+ .set_max_threads(config::fragment_pool_thread_num_max)
+ .set_max_queue_size(config::fragment_pool_queue_size)
+ .set_cgroup_cpu_ctl(cg_cpu_ctl_ptr)
+ .build(&thread_pool);
+ if (!ret.ok()) {
+ LOG(INFO) << "[upsert wg thread pool] create non-pipline thread pool failed, gid="
+ << tg_id;
+ } else {
+ _non_pipe_thread_pool = std::move(thread_pool);
+ }
+ }
+
+ // step 6: update cgroup cpu if needed
+ if (_cgroup_cpu_ctl) {
+ if (enable_cpu_hard_limit) {
+ if (cpu_hard_limit > 0) {
+ _cgroup_cpu_ctl->update_cpu_hard_limit(cpu_hard_limit);
+ _cgroup_cpu_ctl->update_cpu_soft_limit(CPU_SOFT_LIMIT_DEFAULT_VALUE);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] enable cpu hard limit but value is illegal: "
+ << cpu_hard_limit << ", gid=" << tg_id;
+ }
+ } else {
+ if (config::enable_cgroup_cpu_soft_limit) {
+ _cgroup_cpu_ctl->update_cpu_soft_limit(cpu_shares);
+ _cgroup_cpu_ctl->update_cpu_hard_limit(
+ CPU_HARD_LIMIT_DEFAULT_VALUE); // disable cpu hard limit
+ }
+ }
+ _cgroup_cpu_ctl->get_cgroup_cpu_info(&(tg_info->cgroup_cpu_shares),
+ &(tg_info->cgroup_cpu_hard_limit));
+ }
+}
+
+void TaskGroup::get_query_scheduler(doris::pipeline::TaskScheduler** exec_sched,
+ vectorized::SimplifiedScanScheduler** scan_sched,
+ ThreadPool** non_pipe_thread_pool,
+ vectorized::SimplifiedScanScheduler** remote_scan_sched) {
+ std::shared_lock<std::shared_mutex> rlock(_task_sched_lock);
+ *exec_sched = _task_sched.get();
Review Comment:
the logic ```return global sched``` is in get_pipe_exec_scheduler/get_non_pipe_exec_thread_pool
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #31376:
URL: https://github.com/apache/doris/pull/31376#issuecomment-1962861784
PR approved by at least one committer and no changes requested.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei merged PR #31376:
URL: https://github.com/apache/doris/pull/31376
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #31376:
URL: https://github.com/apache/doris/pull/31376#discussion_r1501751586
##########
be/src/runtime/task_group/task_group.cpp:
##########
@@ -266,8 +273,167 @@ Status TaskGroupInfo::parse_topic_info(const TWorkloadGroupInfo& workload_group_
task_group_info->scan_thread_num = workload_group_info.scan_thread_num;
}
+ // 10 max remote scan thread num
+ task_group_info->max_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.max_remote_scan_thread_num &&
+ workload_group_info.max_remote_scan_thread_num > 0) {
+ task_group_info->max_remote_scan_thread_num =
+ workload_group_info.max_remote_scan_thread_num;
+ }
+
+ // 11 min remote scan thread num
+ task_group_info->min_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.min_remote_scan_thread_num &&
+ workload_group_info.min_remote_scan_thread_num > 0) {
+ task_group_info->min_remote_scan_thread_num =
+ workload_group_info.min_remote_scan_thread_num;
+ }
+
return Status::OK();
}
+void TaskGroup::upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env) {
Review Comment:
warning: method 'upsert_task_scheduler' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/task_group/task_group.h:131:
```diff
- void upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env);
+ static void upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env);
```
##########
be/src/vec/exec/scan/scanner_scheduler.h:
##########
@@ -148,11 +148,33 @@
}
}
+ void reset_max_thread_num(int thread_num) {
+ int max_thread_num = _scan_thread_pool->max_threads();
+
+ if (max_thread_num != thread_num) {
+ Status st = _scan_thread_pool->set_max_threads(thread_num);
+ if (!st.ok()) {
+ LOG(INFO) << "reset max thread num failed, sche name=" << _sched_name;
+ }
+ }
+ }
+
+ void reset_min_thread_num(int thread_num) {
Review Comment:
warning: method 'reset_min_thread_num' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void reset_min_thread_num(int thread_num) {
```
##########
be/src/vec/exec/scan/scanner_scheduler.h:
##########
@@ -148,11 +148,33 @@ class SimplifiedScanScheduler {
}
}
+ void reset_max_thread_num(int thread_num) {
Review Comment:
warning: method 'reset_max_thread_num' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void reset_max_thread_num(int thread_num) {
```
##########
be/src/runtime/task_group/task_group.cpp:
##########
@@ -266,8 +273,167 @@
task_group_info->scan_thread_num = workload_group_info.scan_thread_num;
}
+ // 10 max remote scan thread num
+ task_group_info->max_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.max_remote_scan_thread_num &&
+ workload_group_info.max_remote_scan_thread_num > 0) {
+ task_group_info->max_remote_scan_thread_num =
+ workload_group_info.max_remote_scan_thread_num;
+ }
+
+ // 11 min remote scan thread num
+ task_group_info->min_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.min_remote_scan_thread_num &&
+ workload_group_info.min_remote_scan_thread_num > 0) {
+ task_group_info->min_remote_scan_thread_num =
+ workload_group_info.min_remote_scan_thread_num;
+ }
+
return Status::OK();
}
+void TaskGroup::upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env) {
Review Comment:
warning: function 'upsert_task_scheduler' exceeds recommended size/complexity thresholds [readability-function-size]
```cpp
void TaskGroup::upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env) {
^
```
<details>
<summary>Additional context</summary>
**be/src/runtime/task_group/task_group.cpp:294:** 113 lines including whitespace and comments (threshold 80)
```cpp
void TaskGroup::upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env) {
^
```
</details>
##########
be/src/runtime/task_group/task_group.cpp:
##########
@@ -266,8 +273,167 @@
task_group_info->scan_thread_num = workload_group_info.scan_thread_num;
}
+ // 10 max remote scan thread num
+ task_group_info->max_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.max_remote_scan_thread_num &&
+ workload_group_info.max_remote_scan_thread_num > 0) {
+ task_group_info->max_remote_scan_thread_num =
+ workload_group_info.max_remote_scan_thread_num;
+ }
+
+ // 11 min remote scan thread num
+ task_group_info->min_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.min_remote_scan_thread_num &&
+ workload_group_info.min_remote_scan_thread_num > 0) {
+ task_group_info->min_remote_scan_thread_num =
+ workload_group_info.min_remote_scan_thread_num;
+ }
+
return Status::OK();
}
+void TaskGroup::upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env) {
+ uint64_t tg_id = tg_info->id;
+ std::string tg_name = tg_info->name;
+ int cpu_hard_limit = tg_info->cpu_hard_limit;
+ uint64_t cpu_shares = tg_info->cpu_share;
+ bool enable_cpu_hard_limit = tg_info->enable_cpu_hard_limit;
+ int scan_thread_num = tg_info->scan_thread_num;
+ int max_remote_scan_thread_num = tg_info->max_remote_scan_thread_num;
+ int min_remote_scan_thread_num = tg_info->min_remote_scan_thread_num;
+
+ std::lock_guard<std::shared_mutex> wlock(_task_sched_lock);
+ if (config::doris_cgroup_cpu_path != "" && _cgroup_cpu_ctl == nullptr) {
+ std::unique_ptr<CgroupCpuCtl> cgroup_cpu_ctl = std::make_unique<CgroupV1CpuCtl>(tg_id);
+ Status ret = cgroup_cpu_ctl->init();
+ if (ret.ok()) {
+ _cgroup_cpu_ctl = std::move(cgroup_cpu_ctl);
+ LOG(INFO) << "[upsert wg thread pool] cgroup init success";
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] cgroup init failed, gid= " << tg_id
+ << ", reason=" << ret.to_string();
+ }
+ }
+
+ CgroupCpuCtl* cg_cpu_ctl_ptr = _cgroup_cpu_ctl.get();
+
+ if (_task_sched == nullptr) {
+ int32_t executors_size = config::pipeline_executor_size;
+ if (executors_size <= 0) {
+ executors_size = CpuInfo::num_cores();
+ }
+ auto task_queue = std::make_shared<pipeline::MultiCoreTaskQueue>(executors_size);
+ std::unique_ptr<pipeline::TaskScheduler> pipeline_task_scheduler =
+ std::make_unique<pipeline::TaskScheduler>(
+ exec_env, exec_env->get_global_block_scheduler(), std::move(task_queue),
+ "Exec_" + tg_name, cg_cpu_ctl_ptr);
+ Status ret = pipeline_task_scheduler->start();
+ if (ret.ok()) {
+ _task_sched = std::move(pipeline_task_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] task scheduler start failed, gid= " << tg_id;
+ }
+ }
+
+ if (_scan_task_sched == nullptr) {
+ std::unique_ptr<vectorized::SimplifiedScanScheduler> scan_scheduler =
+ std::make_unique<vectorized::SimplifiedScanScheduler>("Scan_" + tg_name,
+ cg_cpu_ctl_ptr);
+ Status ret = scan_scheduler->start();
+ if (ret.ok()) {
+ _scan_task_sched = std::move(scan_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] scan scheduler start failed, gid=" << tg_id;
+ }
+ }
+ if (scan_thread_num > 0 && _scan_task_sched) {
+ _scan_task_sched->reset_thread_num(scan_thread_num);
+ }
+
+ if (_remote_scan_task_sched == nullptr) {
+ 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();
+ if (ret.ok()) {
+ _remote_scan_task_sched = std::move(remote_scan_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] remote scan scheduler start failed, gid="
+ << tg_id;
+ }
+ }
+ if (max_remote_scan_thread_num > 0 && _remote_scan_task_sched) {
+ _remote_scan_task_sched->reset_max_thread_num(max_remote_scan_thread_num);
+ }
+ if (min_remote_scan_thread_num > 0 && _remote_scan_task_sched) {
+ _remote_scan_task_sched->reset_min_thread_num(min_remote_scan_thread_num);
+ }
+
+ if (_non_pipe_thread_pool == nullptr) {
+ std::unique_ptr<ThreadPool> thread_pool = nullptr;
+ auto ret = ThreadPoolBuilder("nonPip_" + tg_name)
+ .set_min_threads(1)
+ .set_max_threads(config::fragment_pool_thread_num_max)
+ .set_max_queue_size(config::fragment_pool_queue_size)
+ .set_cgroup_cpu_ctl(cg_cpu_ctl_ptr)
+ .build(&thread_pool);
+ if (!ret.ok()) {
+ LOG(INFO) << "[upsert wg thread pool] create non-pipline thread pool failed, gid="
+ << tg_id;
+ } else {
+ _non_pipe_thread_pool = std::move(thread_pool);
+ }
+ }
+
+ // step 6: update cgroup cpu if needed
+ if (_cgroup_cpu_ctl) {
+ if (enable_cpu_hard_limit) {
+ if (cpu_hard_limit > 0) {
+ _cgroup_cpu_ctl->update_cpu_hard_limit(cpu_hard_limit);
+ _cgroup_cpu_ctl->update_cpu_soft_limit(CPU_SOFT_LIMIT_DEFAULT_VALUE);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] enable cpu hard limit but value is illegal: "
+ << cpu_hard_limit << ", gid=" << tg_id;
+ }
+ } else {
+ if (config::enable_cgroup_cpu_soft_limit) {
+ _cgroup_cpu_ctl->update_cpu_soft_limit(cpu_shares);
+ _cgroup_cpu_ctl->update_cpu_hard_limit(
+ CPU_HARD_LIMIT_DEFAULT_VALUE); // disable cpu hard limit
+ }
+ }
+ _cgroup_cpu_ctl->get_cgroup_cpu_info(&(tg_info->cgroup_cpu_shares),
+ &(tg_info->cgroup_cpu_hard_limit));
+ }
+}
+
+void TaskGroup::get_query_scheduler(doris::pipeline::TaskScheduler** exec_sched,
Review Comment:
warning: method 'get_query_scheduler' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/task_group/task_group.h:133:
```diff
- void get_query_scheduler(doris::pipeline::TaskScheduler** exec_sched,
+ static void get_query_scheduler(doris::pipeline::TaskScheduler** exec_sched,
```
##########
be/src/runtime/task_group/task_group.cpp:
##########
@@ -266,8 +273,167 @@
task_group_info->scan_thread_num = workload_group_info.scan_thread_num;
}
+ // 10 max remote scan thread num
+ task_group_info->max_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.max_remote_scan_thread_num &&
+ workload_group_info.max_remote_scan_thread_num > 0) {
+ task_group_info->max_remote_scan_thread_num =
+ workload_group_info.max_remote_scan_thread_num;
+ }
+
+ // 11 min remote scan thread num
+ task_group_info->min_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.min_remote_scan_thread_num &&
+ workload_group_info.min_remote_scan_thread_num > 0) {
+ task_group_info->min_remote_scan_thread_num =
+ workload_group_info.min_remote_scan_thread_num;
+ }
+
return Status::OK();
}
+void TaskGroup::upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env) {
+ uint64_t tg_id = tg_info->id;
+ std::string tg_name = tg_info->name;
+ int cpu_hard_limit = tg_info->cpu_hard_limit;
+ uint64_t cpu_shares = tg_info->cpu_share;
+ bool enable_cpu_hard_limit = tg_info->enable_cpu_hard_limit;
+ int scan_thread_num = tg_info->scan_thread_num;
+ int max_remote_scan_thread_num = tg_info->max_remote_scan_thread_num;
+ int min_remote_scan_thread_num = tg_info->min_remote_scan_thread_num;
+
+ std::lock_guard<std::shared_mutex> wlock(_task_sched_lock);
+ if (config::doris_cgroup_cpu_path != "" && _cgroup_cpu_ctl == nullptr) {
+ std::unique_ptr<CgroupCpuCtl> cgroup_cpu_ctl = std::make_unique<CgroupV1CpuCtl>(tg_id);
+ Status ret = cgroup_cpu_ctl->init();
+ if (ret.ok()) {
+ _cgroup_cpu_ctl = std::move(cgroup_cpu_ctl);
+ LOG(INFO) << "[upsert wg thread pool] cgroup init success";
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] cgroup init failed, gid= " << tg_id
+ << ", reason=" << ret.to_string();
+ }
+ }
+
+ CgroupCpuCtl* cg_cpu_ctl_ptr = _cgroup_cpu_ctl.get();
+
+ if (_task_sched == nullptr) {
+ int32_t executors_size = config::pipeline_executor_size;
+ if (executors_size <= 0) {
+ executors_size = CpuInfo::num_cores();
+ }
+ auto task_queue = std::make_shared<pipeline::MultiCoreTaskQueue>(executors_size);
+ std::unique_ptr<pipeline::TaskScheduler> pipeline_task_scheduler =
+ std::make_unique<pipeline::TaskScheduler>(
+ exec_env, exec_env->get_global_block_scheduler(), std::move(task_queue),
+ "Exec_" + tg_name, cg_cpu_ctl_ptr);
+ Status ret = pipeline_task_scheduler->start();
+ if (ret.ok()) {
+ _task_sched = std::move(pipeline_task_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] task scheduler start failed, gid= " << tg_id;
+ }
+ }
+
+ if (_scan_task_sched == nullptr) {
+ std::unique_ptr<vectorized::SimplifiedScanScheduler> scan_scheduler =
+ std::make_unique<vectorized::SimplifiedScanScheduler>("Scan_" + tg_name,
+ cg_cpu_ctl_ptr);
+ Status ret = scan_scheduler->start();
+ if (ret.ok()) {
+ _scan_task_sched = std::move(scan_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] scan scheduler start failed, gid=" << tg_id;
+ }
+ }
+ if (scan_thread_num > 0 && _scan_task_sched) {
+ _scan_task_sched->reset_thread_num(scan_thread_num);
+ }
+
+ if (_remote_scan_task_sched == nullptr) {
+ 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();
+ if (ret.ok()) {
+ _remote_scan_task_sched = std::move(remote_scan_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] remote scan scheduler start failed, gid="
+ << tg_id;
+ }
+ }
+ if (max_remote_scan_thread_num > 0 && _remote_scan_task_sched) {
+ _remote_scan_task_sched->reset_max_thread_num(max_remote_scan_thread_num);
+ }
+ if (min_remote_scan_thread_num > 0 && _remote_scan_task_sched) {
+ _remote_scan_task_sched->reset_min_thread_num(min_remote_scan_thread_num);
+ }
+
+ if (_non_pipe_thread_pool == nullptr) {
+ std::unique_ptr<ThreadPool> thread_pool = nullptr;
+ auto ret = ThreadPoolBuilder("nonPip_" + tg_name)
+ .set_min_threads(1)
+ .set_max_threads(config::fragment_pool_thread_num_max)
+ .set_max_queue_size(config::fragment_pool_queue_size)
+ .set_cgroup_cpu_ctl(cg_cpu_ctl_ptr)
+ .build(&thread_pool);
+ if (!ret.ok()) {
+ LOG(INFO) << "[upsert wg thread pool] create non-pipline thread pool failed, gid="
+ << tg_id;
+ } else {
+ _non_pipe_thread_pool = std::move(thread_pool);
+ }
+ }
+
+ // step 6: update cgroup cpu if needed
+ if (_cgroup_cpu_ctl) {
+ if (enable_cpu_hard_limit) {
+ if (cpu_hard_limit > 0) {
+ _cgroup_cpu_ctl->update_cpu_hard_limit(cpu_hard_limit);
+ _cgroup_cpu_ctl->update_cpu_soft_limit(CPU_SOFT_LIMIT_DEFAULT_VALUE);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] enable cpu hard limit but value is illegal: "
+ << cpu_hard_limit << ", gid=" << tg_id;
+ }
+ } else {
+ if (config::enable_cgroup_cpu_soft_limit) {
+ _cgroup_cpu_ctl->update_cpu_soft_limit(cpu_shares);
+ _cgroup_cpu_ctl->update_cpu_hard_limit(
+ CPU_HARD_LIMIT_DEFAULT_VALUE); // disable cpu hard limit
+ }
+ }
+ _cgroup_cpu_ctl->get_cgroup_cpu_info(&(tg_info->cgroup_cpu_shares),
+ &(tg_info->cgroup_cpu_hard_limit));
+ }
+}
+
+void TaskGroup::get_query_scheduler(doris::pipeline::TaskScheduler** exec_sched,
+ vectorized::SimplifiedScanScheduler** scan_sched,
+ ThreadPool** non_pipe_thread_pool,
+ vectorized::SimplifiedScanScheduler** remote_scan_sched) {
+ std::shared_lock<std::shared_mutex> rlock(_task_sched_lock);
+ *exec_sched = _task_sched.get();
+ *scan_sched = _scan_task_sched.get();
+ *remote_scan_sched = _remote_scan_task_sched.get();
+ *non_pipe_thread_pool = _non_pipe_thread_pool.get();
+}
+
+void TaskGroup::try_stop_task_scheduler() {
Review Comment:
warning: method 'try_stop_task_scheduler' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/task_group/task_group.h:138:
```diff
- void try_stop_task_scheduler();
+ static void try_stop_task_scheduler();
```
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #31376:
URL: https://github.com/apache/doris/pull/31376#issuecomment-1962857275
TeamCity be ut coverage result:
Function Coverage: 35.69% (8546/23943)
Line Coverage: 27.51% (69341/252021)
Region Coverage: 26.66% (35961/134868)
Branch Coverage: 23.48% (18384/78306)
Coverage Report: http://coverage.selectdb-in.cc/coverage/fa183583b0dbade9cbcd37b3d8b0b4e501a15dd1_fa183583b0dbade9cbcd37b3d8b0b4e501a15dd1/report/index.html
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #31376:
URL: https://github.com/apache/doris/pull/31376#discussion_r1501755687
##########
be/src/runtime/task_group/task_group.cpp:
##########
@@ -266,8 +273,167 @@ Status TaskGroupInfo::parse_topic_info(const TWorkloadGroupInfo& workload_group_
task_group_info->scan_thread_num = workload_group_info.scan_thread_num;
}
+ // 10 max remote scan thread num
+ task_group_info->max_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.max_remote_scan_thread_num &&
+ workload_group_info.max_remote_scan_thread_num > 0) {
+ task_group_info->max_remote_scan_thread_num =
+ workload_group_info.max_remote_scan_thread_num;
+ }
+
+ // 11 min remote scan thread num
+ task_group_info->min_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.min_remote_scan_thread_num &&
+ workload_group_info.min_remote_scan_thread_num > 0) {
+ task_group_info->min_remote_scan_thread_num =
+ workload_group_info.min_remote_scan_thread_num;
+ }
+
return Status::OK();
}
+void TaskGroup::upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env) {
+ uint64_t tg_id = tg_info->id;
+ std::string tg_name = tg_info->name;
+ int cpu_hard_limit = tg_info->cpu_hard_limit;
+ uint64_t cpu_shares = tg_info->cpu_share;
+ bool enable_cpu_hard_limit = tg_info->enable_cpu_hard_limit;
+ int scan_thread_num = tg_info->scan_thread_num;
+ int max_remote_scan_thread_num = tg_info->max_remote_scan_thread_num;
+ int min_remote_scan_thread_num = tg_info->min_remote_scan_thread_num;
+
+ std::lock_guard<std::shared_mutex> wlock(_task_sched_lock);
+ if (config::doris_cgroup_cpu_path != "" && _cgroup_cpu_ctl == nullptr) {
+ std::unique_ptr<CgroupCpuCtl> cgroup_cpu_ctl = std::make_unique<CgroupV1CpuCtl>(tg_id);
+ Status ret = cgroup_cpu_ctl->init();
+ if (ret.ok()) {
+ _cgroup_cpu_ctl = std::move(cgroup_cpu_ctl);
+ LOG(INFO) << "[upsert wg thread pool] cgroup init success";
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] cgroup init failed, gid= " << tg_id
+ << ", reason=" << ret.to_string();
+ }
+ }
+
+ CgroupCpuCtl* cg_cpu_ctl_ptr = _cgroup_cpu_ctl.get();
+
+ if (_task_sched == nullptr) {
+ int32_t executors_size = config::pipeline_executor_size;
+ if (executors_size <= 0) {
+ executors_size = CpuInfo::num_cores();
+ }
+ auto task_queue = std::make_shared<pipeline::MultiCoreTaskQueue>(executors_size);
+ std::unique_ptr<pipeline::TaskScheduler> pipeline_task_scheduler =
+ std::make_unique<pipeline::TaskScheduler>(
+ exec_env, exec_env->get_global_block_scheduler(), std::move(task_queue),
+ "Exec_" + tg_name, cg_cpu_ctl_ptr);
Review Comment:
Pipeline_ instead of Exec_
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #31376:
URL: https://github.com/apache/doris/pull/31376#issuecomment-1962842067
TeamCity be ut coverage result:
Function Coverage: 35.70% (8548/23943)
Line Coverage: 27.52% (69365/252021)
Region Coverage: 26.68% (35981/134868)
Branch Coverage: 23.49% (18391/78306)
Coverage Report: http://coverage.selectdb-in.cc/coverage/c8079ae76d38108645e7d2dca784e86152d1ecd4_c8079ae76d38108645e7d2dca784e86152d1ecd4/report/index.html
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #31376:
URL: https://github.com/apache/doris/pull/31376#issuecomment-1962861791
PR approved by anyone and no changes requested.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on PR #31376:
URL: https://github.com/apache/doris/pull/31376#issuecomment-1962846972
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #31376:
URL: https://github.com/apache/doris/pull/31376#discussion_r1501756222
##########
be/src/runtime/task_group/task_group.cpp:
##########
@@ -266,8 +273,167 @@ Status TaskGroupInfo::parse_topic_info(const TWorkloadGroupInfo& workload_group_
task_group_info->scan_thread_num = workload_group_info.scan_thread_num;
}
+ // 10 max remote scan thread num
+ task_group_info->max_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.max_remote_scan_thread_num &&
+ workload_group_info.max_remote_scan_thread_num > 0) {
+ task_group_info->max_remote_scan_thread_num =
+ workload_group_info.max_remote_scan_thread_num;
+ }
+
+ // 11 min remote scan thread num
+ task_group_info->min_remote_scan_thread_num = config::doris_scanner_thread_pool_thread_num;
+ if (workload_group_info.__isset.min_remote_scan_thread_num &&
+ workload_group_info.min_remote_scan_thread_num > 0) {
+ task_group_info->min_remote_scan_thread_num =
+ workload_group_info.min_remote_scan_thread_num;
+ }
+
return Status::OK();
}
+void TaskGroup::upsert_task_scheduler(taskgroup::TaskGroupInfo* tg_info, ExecEnv* exec_env) {
+ uint64_t tg_id = tg_info->id;
+ std::string tg_name = tg_info->name;
+ int cpu_hard_limit = tg_info->cpu_hard_limit;
+ uint64_t cpu_shares = tg_info->cpu_share;
+ bool enable_cpu_hard_limit = tg_info->enable_cpu_hard_limit;
+ int scan_thread_num = tg_info->scan_thread_num;
+ int max_remote_scan_thread_num = tg_info->max_remote_scan_thread_num;
+ int min_remote_scan_thread_num = tg_info->min_remote_scan_thread_num;
+
+ std::lock_guard<std::shared_mutex> wlock(_task_sched_lock);
+ if (config::doris_cgroup_cpu_path != "" && _cgroup_cpu_ctl == nullptr) {
+ std::unique_ptr<CgroupCpuCtl> cgroup_cpu_ctl = std::make_unique<CgroupV1CpuCtl>(tg_id);
+ Status ret = cgroup_cpu_ctl->init();
+ if (ret.ok()) {
+ _cgroup_cpu_ctl = std::move(cgroup_cpu_ctl);
+ LOG(INFO) << "[upsert wg thread pool] cgroup init success";
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] cgroup init failed, gid= " << tg_id
+ << ", reason=" << ret.to_string();
+ }
+ }
+
+ CgroupCpuCtl* cg_cpu_ctl_ptr = _cgroup_cpu_ctl.get();
+
+ if (_task_sched == nullptr) {
+ int32_t executors_size = config::pipeline_executor_size;
+ if (executors_size <= 0) {
+ executors_size = CpuInfo::num_cores();
+ }
+ auto task_queue = std::make_shared<pipeline::MultiCoreTaskQueue>(executors_size);
+ std::unique_ptr<pipeline::TaskScheduler> pipeline_task_scheduler =
+ std::make_unique<pipeline::TaskScheduler>(
+ exec_env, exec_env->get_global_block_scheduler(), std::move(task_queue),
+ "Exec_" + tg_name, cg_cpu_ctl_ptr);
+ Status ret = pipeline_task_scheduler->start();
+ if (ret.ok()) {
+ _task_sched = std::move(pipeline_task_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] task scheduler start failed, gid= " << tg_id;
+ }
+ }
+
+ if (_scan_task_sched == nullptr) {
+ std::unique_ptr<vectorized::SimplifiedScanScheduler> scan_scheduler =
+ std::make_unique<vectorized::SimplifiedScanScheduler>("Scan_" + tg_name,
+ cg_cpu_ctl_ptr);
+ Status ret = scan_scheduler->start();
+ if (ret.ok()) {
+ _scan_task_sched = std::move(scan_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] scan scheduler start failed, gid=" << tg_id;
+ }
+ }
+ if (scan_thread_num > 0 && _scan_task_sched) {
+ _scan_task_sched->reset_thread_num(scan_thread_num);
+ }
+
+ if (_remote_scan_task_sched == nullptr) {
+ 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();
+ if (ret.ok()) {
+ _remote_scan_task_sched = std::move(remote_scan_scheduler);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] remote scan scheduler start failed, gid="
+ << tg_id;
+ }
+ }
+ if (max_remote_scan_thread_num > 0 && _remote_scan_task_sched) {
+ _remote_scan_task_sched->reset_max_thread_num(max_remote_scan_thread_num);
+ }
+ if (min_remote_scan_thread_num > 0 && _remote_scan_task_sched) {
+ _remote_scan_task_sched->reset_min_thread_num(min_remote_scan_thread_num);
+ }
+
+ if (_non_pipe_thread_pool == nullptr) {
+ std::unique_ptr<ThreadPool> thread_pool = nullptr;
+ auto ret = ThreadPoolBuilder("nonPip_" + tg_name)
+ .set_min_threads(1)
+ .set_max_threads(config::fragment_pool_thread_num_max)
+ .set_max_queue_size(config::fragment_pool_queue_size)
+ .set_cgroup_cpu_ctl(cg_cpu_ctl_ptr)
+ .build(&thread_pool);
+ if (!ret.ok()) {
+ LOG(INFO) << "[upsert wg thread pool] create non-pipline thread pool failed, gid="
+ << tg_id;
+ } else {
+ _non_pipe_thread_pool = std::move(thread_pool);
+ }
+ }
+
+ // step 6: update cgroup cpu if needed
+ if (_cgroup_cpu_ctl) {
+ if (enable_cpu_hard_limit) {
+ if (cpu_hard_limit > 0) {
+ _cgroup_cpu_ctl->update_cpu_hard_limit(cpu_hard_limit);
+ _cgroup_cpu_ctl->update_cpu_soft_limit(CPU_SOFT_LIMIT_DEFAULT_VALUE);
+ } else {
+ LOG(INFO) << "[upsert wg thread pool] enable cpu hard limit but value is illegal: "
+ << cpu_hard_limit << ", gid=" << tg_id;
+ }
+ } else {
+ if (config::enable_cgroup_cpu_soft_limit) {
+ _cgroup_cpu_ctl->update_cpu_soft_limit(cpu_shares);
+ _cgroup_cpu_ctl->update_cpu_hard_limit(
+ CPU_HARD_LIMIT_DEFAULT_VALUE); // disable cpu hard limit
+ }
+ }
+ _cgroup_cpu_ctl->get_cgroup_cpu_info(&(tg_info->cgroup_cpu_shares),
+ &(tg_info->cgroup_cpu_hard_limit));
+ }
+}
+
+void TaskGroup::get_query_scheduler(doris::pipeline::TaskScheduler** exec_sched,
+ vectorized::SimplifiedScanScheduler** scan_sched,
+ ThreadPool** non_pipe_thread_pool,
+ vectorized::SimplifiedScanScheduler** remote_scan_sched) {
+ std::shared_lock<std::shared_mutex> rlock(_task_sched_lock);
+ *exec_sched = _task_sched.get();
Review Comment:
Should check if the task sched or scan task sched is valid. If it is not valid then should return the global sched thread pool.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #31376:
URL: https://github.com/apache/doris/pull/31376#discussion_r1501753544
##########
gensrc/thrift/FrontendService.thrift:
##########
@@ -412,7 +412,7 @@ struct TQueryStatistics {
struct TReportWorkloadRuntimeStatusParams {
1: optional i64 backend_id
- 2: map<string, TQueryStatistics> query_statistics_map
Review Comment:
add a new PR for branch-2.0 please.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #31376:
URL: https://github.com/apache/doris/pull/31376#issuecomment-1962824344
Thank you for your contribution to Apache Doris.
Don't know what should be done next? See [How to process your PR](https://cwiki.apache.org/confluence/display/DORIS/How+to+process+your+PR)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on PR #31376:
URL: https://github.com/apache/doris/pull/31376#issuecomment-1962824367
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #31376:
URL: https://github.com/apache/doris/pull/31376#issuecomment-1963800453
PR approved by at least one committer and no changes requested.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Improvement](executor)add remote scan thread pool [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on PR #31376:
URL: https://github.com/apache/doris/pull/31376#issuecomment-1962848077
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org