This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 018ac2fdbc4 branch-4.1: [fix](be) Release workload group schedulers on
shutdown #65112 (#65186)
018ac2fdbc4 is described below
commit 018ac2fdbc484c3f44a6c4f2b871120f5b2be35a
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Jul 3 20:34:33 2026 +0800
branch-4.1: [fix](be) Release workload group schedulers on shutdown #65112
(#65186)
Cherry-picked from #65112
Co-authored-by: TengJianPing <[email protected]>
---
be/src/exec/pipeline/task_queue.cpp | 35 +++++++++++++++++-----
be/src/exec/pipeline/task_queue.h | 2 ++
be/src/exec/scan/scanner_scheduler.h | 8 +++--
.../time_sharing/time_sharing_task_executor.cpp | 7 +++--
be/src/runtime/exec_env_init.cpp | 8 ++++-
be/src/runtime/workload_group/workload_group.cpp | 13 ++++++++
be/src/runtime/workload_group/workload_group.h | 2 ++
.../workload_group/workload_group_manager.cpp | 7 +++++
.../workload_group/workload_group_manager.h | 2 ++
9 files changed, 71 insertions(+), 13 deletions(-)
diff --git a/be/src/exec/pipeline/task_queue.cpp
b/be/src/exec/pipeline/task_queue.cpp
index db20c7c12fd..126781e59af 100644
--- a/be/src/exec/pipeline/task_queue.cpp
+++ b/be/src/exec/pipeline/task_queue.cpp
@@ -21,6 +21,7 @@
#include <chrono> // IWYU pragma: keep
#include <memory>
#include <string>
+#include <vector>
#include "common/logging.h"
#include "exec/pipeline/pipeline_task.h"
@@ -38,6 +39,12 @@ PipelineTaskSPtr SubTaskQueue::try_take(bool is_steal) {
return task;
}
+std::queue<PipelineTaskSPtr> SubTaskQueue::take_all() {
+ std::queue<PipelineTaskSPtr> tasks;
+ tasks.swap(_queue);
+ return tasks;
+}
+
//////////////////// PriorityTaskQueue ////////////////////
PriorityTaskQueue::PriorityTaskQueue() : _closed(false) {
@@ -49,10 +56,24 @@ PriorityTaskQueue::PriorityTaskQueue() : _closed(false) {
}
void PriorityTaskQueue::close() {
- std::unique_lock<std::mutex> lock(_work_size_mutex);
- _closed = true;
- _wait_task.notify_all();
-
DorisMetrics::instance()->pipeline_task_queue_size->increment(-_total_task_size);
+ std::vector<std::queue<PipelineTaskSPtr>> tasks_to_release;
+ {
+ std::unique_lock<std::mutex> lock(_work_size_mutex);
+ _closed = true;
+ const auto total_task_size = _total_task_size.exchange(0);
+ _wait_task.notify_all();
+ DorisMetrics::instance()->pipeline_task_queue_size->increment(
+ -static_cast<int64_t>(total_task_size));
+ for (auto& sub_queue : _sub_queues) {
+ tasks_to_release.emplace_back(sub_queue.take_all());
+ }
+ }
+ for (auto& tasks : tasks_to_release) {
+ while (!tasks.empty()) {
+ tasks.front()->pop_out_runnable_queue();
+ tasks.pop();
+ }
+ }
}
PipelineTaskSPtr PriorityTaskQueue::_try_take_unprotected(bool is_steal) {
@@ -114,11 +135,11 @@ PipelineTaskSPtr PriorityTaskQueue::take(uint32_t
timeout_ms) {
}
Status PriorityTaskQueue::push(PipelineTaskSPtr task) {
+ auto level = _compute_level(task->get_runtime_ns());
+ std::unique_lock<std::mutex> lock(_work_size_mutex);
if (_closed) {
return Status::InternalError("WorkTaskQueue closed");
}
- auto level = _compute_level(task->get_runtime_ns());
- std::unique_lock<std::mutex> lock(_work_size_mutex);
// update empty queue's runtime, to avoid too high priority
if (_sub_queues[level].empty() &&
@@ -126,6 +147,7 @@ Status PriorityTaskQueue::push(PipelineTaskSPtr task) {
_sub_queues[level].adjust_runtime(_queue_level_min_vruntime);
}
+ task->put_in_runnable_queue();
_sub_queues[level].push_back(task);
_total_task_size++;
DorisMetrics::instance()->pipeline_task_queue_size->increment(1);
@@ -200,7 +222,6 @@ Status MultiCoreTaskQueue::push_back(PipelineTaskSPtr task)
{
Status MultiCoreTaskQueue::push_back(PipelineTaskSPtr task, int core_id) {
DCHECK(core_id < _core_size);
- task->put_in_runnable_queue();
return _prio_task_queues[core_id].push(task);
}
diff --git a/be/src/exec/pipeline/task_queue.h
b/be/src/exec/pipeline/task_queue.h
index 49796dbcbed..fc4ca1b80bb 100644
--- a/be/src/exec/pipeline/task_queue.h
+++ b/be/src/exec/pipeline/task_queue.h
@@ -57,6 +57,8 @@ public:
bool empty() { return _queue.empty(); }
+ std::queue<PipelineTaskSPtr> take_all();
+
private:
std::queue<PipelineTaskSPtr> _queue;
// depends on LEVEL_QUEUE_TIME_FACTOR
diff --git a/be/src/exec/scan/scanner_scheduler.h
b/be/src/exec/scan/scanner_scheduler.h
index 3635ffc4d11..f1939409bda 100644
--- a/be/src/exec/scan/scanner_scheduler.h
+++ b/be/src/exec/scan/scanner_scheduler.h
@@ -164,7 +164,9 @@ public:
}
void stop() override {
- _is_stop.store(true);
+ if (_is_stop.exchange(true)) {
+ return;
+ }
_scan_thread_pool->shutdown();
_scan_thread_pool->wait();
}
@@ -264,7 +266,9 @@ public:
}
void stop() override {
- _is_stop.store(true);
+ if (_is_stop.exchange(true)) {
+ return;
+ }
_task_executor->stop();
_task_executor->wait();
}
diff --git
a/be/src/exec/scan/task_executor/time_sharing/time_sharing_task_executor.cpp
b/be/src/exec/scan/task_executor/time_sharing/time_sharing_task_executor.cpp
index 138d2df197a..89847de4b29 100644
--- a/be/src/exec/scan/task_executor/time_sharing/time_sharing_task_executor.cpp
+++ b/be/src/exec/scan/task_executor/time_sharing/time_sharing_task_executor.cpp
@@ -274,9 +274,7 @@ Status TimeSharingTaskExecutor::init() {
}
TimeSharingTaskExecutor::~TimeSharingTaskExecutor() {
- if (!_stopped.exchange(true)) {
- stop();
- }
+ stop();
std::vector<std::shared_ptr<PrioritizedSplitRunner>> splits_to_destroy;
{
@@ -313,6 +311,9 @@ Status TimeSharingTaskExecutor::start() {
}
void TimeSharingTaskExecutor::stop() {
+ if (_stopped.exchange(true)) {
+ return;
+ }
// Why access to doris_metrics is safe here?
// Since DorisMetrics is a singleton, it will be destroyed only after
doris_main is exited.
// The shutdown/destroy of SplitThreadPool is guaranteed to take place
before doris_main exits by
diff --git a/be/src/runtime/exec_env_init.cpp b/be/src/runtime/exec_env_init.cpp
index d224ac78661..2bce2dc00c5 100644
--- a/be/src/runtime/exec_env_init.cpp
+++ b/be/src/runtime/exec_env_init.cpp
@@ -849,7 +849,8 @@ void ExecEnv::destroy() {
SAFE_STOP(_stream_load_recorder_manager);
// stop workload scheduler
SAFE_STOP(_workload_sched_mgr);
- // stop pipline step 2, cgroup execution
+ // Stop workload group execution threads before FragmentMgr. Running
pipeline tasks can still
+ // report status through FragmentMgr's async thread pool.
SAFE_STOP(_workload_group_manager);
SAFE_STOP(_external_scan_context_mgr);
@@ -862,6 +863,11 @@ void ExecEnv::destroy() {
_memtable_memory_limiter.reset();
_delta_writer_v2_pool.reset();
_load_stream_map_pool.reset();
+ // Workload group schedulers own the query pipeline, scan, and memtable
flush queues.
+ // Release them after fragment/load resources have stopped submitting
cleanup work.
+ if (_workload_group_manager) {
+ _workload_group_manager->destroy_schedulers();
+ }
SAFE_STOP(_write_cooldown_meta_executors);
// _id_manager must be destoried before tablet schema cache
diff --git a/be/src/runtime/workload_group/workload_group.cpp
b/be/src/runtime/workload_group/workload_group.cpp
index 888f85a7a44..fb70b203f82 100644
--- a/be/src/runtime/workload_group/workload_group.cpp
+++ b/be/src/runtime/workload_group/workload_group.cpp
@@ -745,6 +745,10 @@ int64_t WorkloadGroup::get_mem_used() {
void WorkloadGroup::try_stop_schedulers() {
std::lock_guard<std::shared_mutex> wlock(_task_sched_lock);
+ stop_schedulers_no_lock();
+}
+
+void WorkloadGroup::stop_schedulers_no_lock() {
if (_task_sched) {
_task_sched->stop();
}
@@ -768,6 +772,15 @@ void WorkloadGroup::try_stop_schedulers() {
}
}
+void WorkloadGroup::destroy_schedulers() {
+ std::lock_guard<std::shared_mutex> wlock(_task_sched_lock);
+ _task_sched.reset();
+ _scan_task_sched.reset();
+ _remote_scan_task_sched.reset();
+ _memtable_flush_pool.reset();
+ _cgroup_cpu_ctl.reset();
+}
+
void WorkloadGroup::update_memtable_flush_threads() {
if (_memtable_flush_pool == nullptr) {
return;
diff --git a/be/src/runtime/workload_group/workload_group.h
b/be/src/runtime/workload_group/workload_group.h
index 9b764664adb..2a310bf66a0 100644
--- a/be/src/runtime/workload_group/workload_group.h
+++ b/be/src/runtime/workload_group/workload_group.h
@@ -219,6 +219,8 @@ private:
void upsert_cgroup_cpu_ctl_no_lock(WorkloadGroupInfo* wg_info);
Status upsert_thread_pool_no_lock(WorkloadGroupInfo* wg_info,
std::shared_ptr<CgroupCpuCtl>
cg_cpu_ctl_ptr);
+ void stop_schedulers_no_lock();
+ void destroy_schedulers();
std::string _memory_debug_string() const;
diff --git a/be/src/runtime/workload_group/workload_group_manager.cpp
b/be/src/runtime/workload_group/workload_group_manager.cpp
index 6ea679eb625..019620a586d 100644
--- a/be/src/runtime/workload_group/workload_group_manager.cpp
+++ b/be/src/runtime/workload_group/workload_group_manager.cpp
@@ -944,6 +944,13 @@ void WorkloadGroupMgr::stop() {
}
}
+void WorkloadGroupMgr::destroy_schedulers() {
+ std::shared_lock<std::shared_mutex> r_lock(_group_mutex);
+ for (auto iter = _workload_groups.begin(); iter != _workload_groups.end();
iter++) {
+ iter->second->destroy_schedulers();
+ }
+}
+
Status WorkloadGroupMgr::create_internal_wg() {
TWorkloadGroupInfo twg_info;
twg_info.__set_id(INTERNAL_NORMAL_WG_ID);
diff --git a/be/src/runtime/workload_group/workload_group_manager.h
b/be/src/runtime/workload_group/workload_group_manager.h
index 4db43b50db1..5b2b1160d60 100644
--- a/be/src/runtime/workload_group/workload_group_manager.h
+++ b/be/src/runtime/workload_group/workload_group_manager.h
@@ -79,6 +79,8 @@ public:
void stop();
+ void destroy_schedulers();
+
void refresh_workload_group_memory_state();
void get_wg_resource_usage(Block* block);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
