branch-4.0: [pipeline](conf) make blocking scheduler configurable #57354 (#57392)

Cherry-picked from #57354

Co-authored-by: Gabriel <[email protected]>

Author: github-actions[bot]

be/src/common/config.cpp

@@ -1091,6 +1091,7 @@ DEFINE_Bool(enable_graceful_exit_check, "false");
 DEFINE_Bool(enable_debug_points, "false");
 
 DEFINE_Int32(pipeline_executor_size, "0");
+DEFINE_Int32(blocking_pipeline_executor_size, "0");
 DEFINE_Bool(enable_workload_group_for_scan, "false");
 DEFINE_mInt64(workload_group_scan_task_wait_timeout_ms, "10000");
 

be/src/common/config.h

@@ -1122,6 +1122,7 @@ DECLARE_Bool(enable_graceful_exit_check);
 DECLARE_Bool(enable_debug_points);
 
 DECLARE_Int32(pipeline_executor_size);
+DECLARE_Int32(blocking_pipeline_executor_size);
 
 // block file cache
 DECLARE_Bool(enable_file_cache);

be/src/pipeline/task_scheduler.h

@@ -80,10 +80,11 @@ class TaskScheduler {
 
 class HybridTaskScheduler MOCK_REMOVE(final) : public TaskScheduler {
 public:
-    HybridTaskScheduler(int core_num, std::string name,
+    HybridTaskScheduler(int exec_thread_num, int blocking_exec_thread_num, std::string name,
                         std::shared_ptr<CgroupCpuCtl> cgroup_cpu_ctl)
-            : _blocking_scheduler(core_num * 2, name + "_blocking_scheduler", cgroup_cpu_ctl),
-              _simple_scheduler(core_num, name + "_simple_scheduler", cgroup_cpu_ctl) {}
+            : _blocking_scheduler(blocking_exec_thread_num, name + "_blocking_scheduler",
+                                  cgroup_cpu_ctl),
+              _simple_scheduler(exec_thread_num, name + "_simple_scheduler", cgroup_cpu_ctl) {}
 
     Status submit(PipelineTaskSPtr task) override;
 

be/src/runtime/workload_group/workload_group.cpp

@@ -414,6 +414,10 @@ WorkloadGroupInfo WorkloadGroupInfo::parse_topic_info(
     if (exec_thread_num <= 0) {
         exec_thread_num = CpuInfo::num_cores();
     }
+    int blocking_exec_thread_num = config::blocking_pipeline_executor_size;
+    if (blocking_exec_thread_num <= 0) {
+        blocking_exec_thread_num = CpuInfo::num_cores() * 2;
+    }
 
     int num_disk = 1;
     int num_cpus = 1;
@@ -484,6 +488,7 @@ WorkloadGroupInfo WorkloadGroupInfo::parse_topic_info(
             .total_query_slot_count = total_query_slot_count,
             .slot_mem_policy = slot_mem_policy,
             .pipeline_exec_thread_num = exec_thread_num,
+            .blocking_pipeline_exec_thread_num = blocking_exec_thread_num,
             .max_flush_thread_num = max_flush_thread_num,
             .min_flush_thread_num = min_flush_thread_num};
 }
@@ -522,6 +527,7 @@ Status WorkloadGroup::upsert_thread_pool_no_lock(WorkloadGroupInfo* wg_info,
     uint64_t wg_id = wg_info->id;
     std::string wg_name = wg_info->name;
     int pipeline_exec_thread_num = wg_info->pipeline_exec_thread_num;
+    int blocking_exec_thread_num = wg_info->blocking_pipeline_exec_thread_num;
     int scan_thread_num = wg_info->scan_thread_num;
     int max_remote_scan_thread_num = wg_info->max_remote_scan_thread_num;
     int min_remote_scan_thread_num = wg_info->min_remote_scan_thread_num;
@@ -532,6 +538,7 @@ Status WorkloadGroup::upsert_thread_pool_no_lock(WorkloadGroupInfo* wg_info,
     if (_task_sched == nullptr) {
         std::unique_ptr<pipeline::TaskScheduler> pipeline_task_scheduler =
                 std::make_unique<pipeline::HybridTaskScheduler>(pipeline_exec_thread_num,
+                                                                blocking_exec_thread_num,
                                                                 "p_" + wg_name, cg_cpu_ctl_ptr);
         Status ret = pipeline_task_scheduler->start();
         if (ret.ok()) {

be/src/runtime/workload_group/workload_group.h

@@ -285,6 +285,7 @@ struct WorkloadGroupInfo {
     int cgroup_cpu_hard_limit = 0;
     const bool valid = true;
     const int pipeline_exec_thread_num = 0;
+    const int blocking_pipeline_exec_thread_num = 0;
     const int max_flush_thread_num = 0;
     const int min_flush_thread_num = 0;