This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new a310af513b6 branch-4.2: [fix](be) Copy PipelineTask::_sink under the
lifecycle lock in stop_if_finished (#69208) (#69270)
a310af513b6 is described below
commit a310af513b685d3608ed7f44aa3fd42717d59056
Author: Jerry Hu <[email protected]>
AuthorDate: Tue Oct 20 10:00:08 2026 +0800
branch-4.2: [fix](be) Copy PipelineTask::_sink under the lifecycle lock in
stop_if_finished (#69208) (#69270)
### What problem does this PR solve?
Issue Number: None
Related PR: #69208
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_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_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 `terminate()` are
called after the lock is released, because `terminate()` acquires the
same non-recursive mutex. A debug point inside the critical section lets
the unit test hold the lock deterministically.
This is the branch-4.2 backport of #69208. branch-4.2 still has a single
`_dependency_lock` instead of the `_dependency_lifecycle_lock` used on
master, and `stop_if_finished()` calls `terminate()` instead of
`unblock_all_dependencies()`. Only those two branch-specific call sites
are adapted; the fix intent and the test are unchanged.
### Release note
None
### Check List (For Author)
- Test: Unit Test
- Added `PipelineTaskTest.TEST_STOP_IF_FINISHED_EXCLUDES_FINALIZE`,
adapted from the original PR: it holds `stop_if_finished()` inside the
critical section, checks that a concurrent `finalize()` is blocked and
cannot reset `_sink`, and verifies that a finalized task is skipped
without touching the cleared `_sink`.
- All `PipelineTaskTest` cases pass: 23/23.
- `clang-format --dry-run --Werror` passes on the two changed files.
- 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 b364eaee3a7..fa12ce513d3 100644
--- a/be/src/exec/pipeline/pipeline_task.cpp
+++ b/be/src/exec/pipeline/pipeline_task.cpp
@@ -871,11 +871,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();
- terminate();
+ 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
terminate() because
+ // that function acquires it again.
+ std::unique_lock<std::mutex> lock(_dependency_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();
+ terminate();
}
}
diff --git a/be/test/exec/pipeline/pipeline_task_test.cpp
b/be/test/exec/pipeline/pipeline_task_test.cpp
index 104c381efb1..9e2b5e0892d 100644
--- a/be/test/exec/pipeline/pipeline_task_test.cpp
+++ b/be/test/exec/pipeline/pipeline_task_test.cpp
@@ -579,6 +579,85 @@ TEST_F(PipelineTaskTest, TEST_STATE_TRANSITION) {
}
}
+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]
