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 99653480bc8 branch-4.1: [fix](pipeline) Make PipelineTask::set_running
an unconditional exchange #67032 (#68929)
99653480bc8 is described below
commit 99653480bc8cdc0ec827e4174d95c47b9012d699
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Oct 12 18:05:20 2026 +0800
branch-4.1: [fix](pipeline) Make PipelineTask::set_running an unconditional
exchange #67032 (#68929)
Cherry-picked from #67032
Co-authored-by: Raghvendra Singh <[email protected]>
Co-authored-by: Claude Fable 5 <[email protected]>
---
be/src/exec/pipeline/pipeline_task.h | 12 ++++--
be/test/exec/pipeline/pipeline_task_test.cpp | 58 ++++++++++++++++++++++++++++
2 files changed, 67 insertions(+), 3 deletions(-)
diff --git a/be/src/exec/pipeline/pipeline_task.h
b/be/src/exec/pipeline/pipeline_task.h
index 5b5da9dca83..7c4b4f92057 100644
--- a/be/src/exec/pipeline/pipeline_task.h
+++ b/be/src/exec/pipeline/pipeline_task.h
@@ -153,9 +153,15 @@ public:
bool is_running() { return _running.load(); }
virtual bool set_running(bool running) {
- bool old_value = !running;
- _running.compare_exchange_weak(old_value, running);
- return old_value;
+ // This is a mutual-exclusion gate: TaskScheduler::_do_work skips a
task when
+ // set_running(true) returns true, and RevokableTask delegates here so
an
+ // in-flight revocation excludes its underlying task. It must be an
+ // unconditional exchange: compare_exchange_weak may fail spuriously
on LL/SC
+ // architectures (aarch64), and a spurious failure with _running ==
false
+ // returns "was not running" WITHOUT setting the flag — two workers
then
+ // execute the same task concurrently and race on its local state
+ // (shared hash table / _places).
+ return _running.exchange(running);
}
virtual RuntimeState* runtime_state() const { return _state; }
diff --git a/be/test/exec/pipeline/pipeline_task_test.cpp
b/be/test/exec/pipeline/pipeline_task_test.cpp
index 33b94ebfd83..104c381efb1 100644
--- a/be/test/exec/pipeline/pipeline_task_test.cpp
+++ b/be/test/exec/pipeline/pipeline_task_test.cpp
@@ -1857,4 +1857,62 @@ TEST_F(PipelineTaskTest, TEST_TERMINATE_RACE_FIX) {
config::enable_debug_points = false;
}
+TEST_F(PipelineTaskTest, TEST_SET_RUNNING_MUTUAL_EXCLUSION) {
+ 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;
+ auto task = std::make_shared<PipelineTask>(pip, task_id,
_runtime_state.get(), _context,
+ profile.get(),
shared_state_map, task_id);
+
+ // set_running(true) is the scheduler's double-execute gate (and
RevokableTask's mutual
+ // exclusion with its underlying task): at most one caller may observe
"was not running"
+ // until the matching set_running(false). The former compare_exchange_weak
implementation
+ // could fail spuriously on LL/SC architectures and report acquisition
without setting
+ // the flag, admitting two concurrent holders.
+ std::atomic<int> holders {0};
+ std::atomic<int64_t> acquired {0};
+ std::atomic<bool> violation {false};
+ constexpr int kThreads = 8;
+ constexpr int kItersPerThread = 200000;
+ std::vector<std::thread> threads;
+ threads.reserve(kThreads);
+ for (int t = 0; t < kThreads; ++t) {
+ threads.emplace_back([&]() {
+ for (int i = 0; i < kItersPerThread; ++i) {
+ if (!task->set_running(true)) {
+ if (holders.fetch_add(1) != 0) {
+ violation = true;
+ }
+ acquired.fetch_add(1, std::memory_order_relaxed);
+ holders.fetch_sub(1);
+ task->set_running(false);
+ }
+ }
+ });
+ }
+ for (auto& th : threads) {
+ th.join();
+ }
+ EXPECT_FALSE(violation.load()) << "two threads acquired the running gate
concurrently";
+ EXPECT_GT(acquired.load(), 0);
+ EXPECT_FALSE(task->is_running());
+}
+
} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
