This is an automated email from the ASF dual-hosted git repository.
yiguolei 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 0ca0f1968d7 [fix](pipeline) Make PipelineTask::set_running an
unconditional exchange (#67032)
0ca0f1968d7 is described below
commit 0ca0f1968d7e6571d8c3a541bc5f7179282c2caa
Author: Raghvendra Singh <[email protected]>
AuthorDate: Mon Oct 12 09:03:20 2026 +0530
[fix](pipeline) Make PipelineTask::set_running an unconditional exchange
(#67032)
### What problem does this PR solve?
Issue Number: close #67031
Problem Summary:
`PipelineTask::set_running` is the scheduler's double-execute gate
(`TaskScheduler::_do_work` re-queues a task when it returns true) and,
via `RevokableTask`'s delegation, the mutual exclusion between a task
and its in-flight spill revocation. It was implemented as a single
`compare_exchange_weak` attempt. Weak CAS may fail spuriously on LL/SC
architectures (aarch64): on a spurious failure with `_running == false`
the caller is told "was not running" while the flag is never set, so a
second worker acquires the same task and both execute it concurrently —
one thread's `reset_hash_table`/`close()` then races the other's
`sink_impl` on shared local state (production SIGSEGV @0x0 in
`AggSinkOperatorX::sink_impl` on Graviton BEs, including two workers
crashing in the same second attached to the same query). x86 compiles
weak CAS as never-spurious, so this only reproduces on ARM.
`std::atomic::exchange` keeps the exact intended semantics (return the
old value, always store the new one) and cannot fail spuriously. Every
other `compare_exchange_weak` in be/src sits in a retry loop; this was
the only no-retry gate.
Adds `TEST_SET_RUNNING_MUTUAL_EXCLUSION`: 8 threads hammer
acquire/release and assert at most one concurrent holder (catches the
old implementation probabilistically on ARM hardware; a hard invariant
for the new one).
### Release note
Fix a BE crash on aarch64 where two pipeline workers could execute the
same task concurrently under spill/memory-pressure churn.
### Check List (For Author)
- Test
- [x] Unit Test
- Behavior changed:
- [x] No.
- Does this need documentation?
- [x] No.
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 dbe5d57f23d..a6e62568c57 100644
--- a/be/src/exec/pipeline/pipeline_task.h
+++ b/be/src/exec/pipeline/pipeline_task.h
@@ -157,9 +157,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 cd00dff86d9..34f8df73810 100644
--- a/be/test/exec/pipeline/pipeline_task_test.cpp
+++ b/be/test/exec/pipeline/pipeline_task_test.cpp
@@ -1992,4 +1992,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]
