This is an automated email from the ASF dual-hosted git repository.
mrhhsg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new faa3fdac16e [fix](be) Copy PipelineTask::_sink under the lifecycle
lock in stop_if_finished (#69208)
faa3fdac16e is described below
commit faa3fdac16effc08881b099ff1c9f69c61eb5c3f
Author: Jerry Hu <[email protected]>
AuthorDate: Sat Oct 17 06:55:02 2026 +0800
[fix](be) Copy PipelineTask::_sink under the lifecycle lock in
stop_if_finished (#69208)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: `PipelineTask::stop_if_finished()` is called by
`FragmentMgr::cancel_worker()` (through
`PipelineFragmentContext::clear_finished_tasks()`) on the background
cancel thread. It copied the `_sink` shared_ptr without any lock. At the
same time a scheduler worker can run `PipelineTask::finalize()` for the
same task, which resets `_sink` while holding
`_dependency_lifecycle_lock`. Copying and resetting the same shared_ptr
instance from two threads without synchronization is a data race
(undefined behavior): the cancel thread can observe a torn
pointer/control block and then call `is_finished()` on an object that is
being released. Every other cross-thread reader of `_sink`
(`debug_string()`, the hybrid scheduler submit path) is already
synchronized; `stop_if_finished()` was the only one left.
Fix: take `_dependency_lifecycle_lock` in `stop_if_finished()`, return
if the task is already finalized, and copy `_sink` under the lock,
following the pattern of `debug_string()`. `is_finished()` and
`unblock_all_dependencies()` are called after the lock is released,
because `unblock_all_dependencies()` acquires the same non-recursive
mutex. A debug point inside the critical section lets the unit test hold
the lock deterministically.
### Release note
None
### Check List (For Author)
- Test: Unit Test
- Added `PipelineTaskTest.TEST_STOP_IF_FINISHED_EXCLUDES_FINALIZE`: it
holds `stop_if_finished()` inside the critical section and checks that a
concurrent `finalize()` is blocked and cannot reset `_sink`, and that a
finalized task is skipped. It fails without the lock and passes with it
(200 repeated runs). All `PipelineTaskTest` cases pass.
- No regression test: the race window is a few instructions wide and
cannot be triggered reliably from SQL.
- Behavior changed: No
- Does this need documentation: No
---
be/src/exec/pipeline/pipeline_task.cpp | 24 +++++++--
be/test/exec/pipeline/pipeline_task_test.cpp | 79 ++++++++++++++++++++++++++++
2 files changed, 99 insertions(+), 4 deletions(-)
diff --git a/be/src/exec/pipeline/pipeline_task.cpp
b/be/src/exec/pipeline/pipeline_task.cpp
index aa5d392c19e..e614fa92816 100644
--- a/be/src/exec/pipeline/pipeline_task.cpp
+++ b/be/src/exec/pipeline/pipeline_task.cpp
@@ -883,11 +883,27 @@ void PipelineTask::stop_if_finished() {
return;
}
SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(fragment->get_query_ctx()->query_mem_tracker());
- if (auto sink = _sink) {
- if (sink->is_finished(_state)) {
- set_wake_up_early();
- unblock_all_dependencies();
+ DataSinkOperatorPtr sink;
+ {
+ // This runs on FragmentMgr's cancel thread while a scheduler worker
may be in finalize(),
+ // which resets _sink under the same lock. Copy the shared_ptr under
the lock so the two
+ // threads never access it concurrently. The lock is released before
+ // unblock_all_dependencies() because that function acquires it again.
+ std::unique_lock<std::mutex> lock(_dependency_lifecycle_lock);
+
DBUG_EXECUTE_IF("PipelineTask::stop_if_finished.holding_lifecycle_lock", {
+ if (dp->callback.has_value()) {
+ DBUG_RUN_CALLBACK();
+ }
+ });
+ if (is_finalized()) {
+ return;
}
+ sink = _sink;
+ }
+ DCHECK(sink);
+ if (sink->is_finished(_state)) {
+ set_wake_up_early();
+ unblock_all_dependencies();
}
}
diff --git a/be/test/exec/pipeline/pipeline_task_test.cpp
b/be/test/exec/pipeline/pipeline_task_test.cpp
index b91c4c6917e..12bc33ad4e3 100644
--- a/be/test/exec/pipeline/pipeline_task_test.cpp
+++ b/be/test/exec/pipeline/pipeline_task_test.cpp
@@ -714,6 +714,85 @@ TEST_F(PipelineTaskTest,
TEST_WAKE_UP_SUBMIT_PROTECTED_FROM_FINALIZE) {
EXPECT_TRUE(task->_operators.empty());
}
+TEST_F(PipelineTaskTest, TEST_STOP_IF_FINISHED_EXCLUDES_FINALIZE) {
+ auto num_instances = 1;
+ auto pip_id = 0;
+ auto task_id = 0;
+ auto pip = std::make_shared<Pipeline>(pip_id, num_instances,
num_instances);
+ OperatorPtr source_op;
+ source_op.reset(new DummyOperator());
+ EXPECT_TRUE(pip->add_operator(source_op, num_instances).ok());
+
+ int op_id = 1;
+ int node_id = 2;
+ int dest_id = 3;
+ DataSinkOperatorPtr sink_op;
+ sink_op.reset(new DummySinkOperatorX(op_id, node_id, dest_id));
+ EXPECT_TRUE(pip->set_sink(sink_op).ok());
+
+ auto profile = std::make_shared<RuntimeProfile>("Pipeline : " +
std::to_string(pip_id));
+ std::map<int,
+ std::pair<std::shared_ptr<BasicSharedState>,
std::vector<std::shared_ptr<Dependency>>>>
+ shared_state_map;
+ _runtime_state->resize_op_id_to_local_state(-1);
+ auto task = std::make_shared<PipelineTask>(pip, task_id,
_runtime_state.get(), _context,
+ profile.get(),
shared_state_map, task_id);
+
+ std::vector<TScanRangeParams> scan_range;
+ int sender_id = 0;
+ TDataSink tsink;
+ EXPECT_TRUE(task->prepare(scan_range, sender_id, tsink).ok());
+ // A closed but not yet finalized task is what FragmentMgr's cancel thread
usually scans.
+ EXPECT_TRUE(task->close(Status::OK()).ok());
+
+ auto origin_enable_debug_points = config::enable_debug_points;
+ config::enable_debug_points = true;
+ Defer debug_point_cleanup {[&]() {
+
DebugPoints::instance()->remove("PipelineTask::stop_if_finished.holding_lifecycle_lock");
+ config::enable_debug_points = origin_enable_debug_points;
+ }};
+
+ std::promise<void> lock_held_promise;
+ auto lock_held = lock_held_promise.get_future();
+ std::promise<void> release_promise;
+ auto release = release_promise.get_future();
+ DebugPoints::instance()->add_with_callback(
+ "PipelineTask::stop_if_finished.holding_lifecycle_lock",
std::function<void()>([&]() {
+ lock_held_promise.set_value();
+ release.wait();
+ }));
+
+ // Simulates FragmentMgr::cancel_worker -> clear_finished_tasks ->
stop_if_finished.
+ std::thread stop_thread([&]() { task->stop_if_finished(); });
+ EXPECT_EQ(lock_held.wait_for(std::chrono::seconds(10)),
std::future_status::ready);
+
+ // Simulates the scheduler worker finalizing the task. It must not reset
_sink while
+ // stop_if_finished() is reading it.
+ std::promise<void> finalize_started_promise;
+ auto finalize_started = finalize_started_promise.get_future();
+ auto finalize_future = std::async(std::launch::async, [&]() {
+ finalize_started_promise.set_value();
+ return task->finalize();
+ });
+ EXPECT_EQ(finalize_started.wait_for(std::chrono::seconds(10)),
std::future_status::ready);
+ EXPECT_EQ(finalize_future.wait_for(std::chrono::milliseconds(100)),
+ std::future_status::timeout);
+ EXPECT_NE(task->_sink, nullptr);
+
+ release_promise.set_value();
+ stop_thread.join();
+ EXPECT_TRUE(finalize_future.get().ok());
+
+ EXPECT_EQ(task->_exec_state, PipelineTask::State::FINALIZED);
+ EXPECT_EQ(task->_sink, nullptr);
+ EXPECT_FALSE(task->_wake_up_early);
+
+ // A finalized task is skipped without touching the cleared _sink.
+
DebugPoints::instance()->remove("PipelineTask::stop_if_finished.holding_lifecycle_lock");
+ task->stop_if_finished();
+ EXPECT_FALSE(task->_wake_up_early);
+}
+
TEST_F(PipelineTaskTest,
TEST_CLOSED_TASK_REJECTS_HYBRID_SUBMIT_BEFORE_FINALIZE) {
auto num_instances = 1;
auto pip_id = 0;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
