You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "BiteTheDDDDt (via GitHub)" <gi...@apache.org> on 2026/04/28 03:35:24 UTC
[PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
BiteTheDDDDt opened a new pull request, #62891:
URL: https://github.com/apache/doris/pull/62891
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: A delayed dependency wake_up could race with pipeline task close/finalize. The old state transition logic validated the current state and then stored the new state in separate operations, so a stale wake_up could overwrite FINISHED/FINALIZED back to RUNNABLE after finalize() had already released `_sink` and `_operators`, making the scheduler resubmit or inspect a released task.
This PR makes `_state_transition()` publish state changes with CAS. Terminal FINISHED/FINALIZED -> RUNNABLE remains legal under wake_up_early, but is an atomic no-op. The uncontended path succeeds with one CAS; only a real concurrent state change retries and revalidates the transition.
### Release note
None
### Check List (For Author)
- Test: Unit Test
- `./run-be-ut.sh --run --filter='PipelineTaskTest.TEST_STATE_TRANSITION:PipelineTaskTest.TEST_STATE_TRANSITION_KEEP_FINALIZED_ON_WAKE_UP_RACE'`
- `build-support/check-format.sh`
- `build-support/run-clang-tidy.sh` attempted; it reports pre-existing diagnostics in `pipeline_task.cpp`, `pipeline_task_test.cpp`, and `jni-util.h` in this environment.
- Behavior changed: No
- Does this need documentation: No
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
Posted by "BiteTheDDDDt (via GitHub)" <gi...@apache.org>.
BiteTheDDDDt commented on code in PR #62891:
URL: https://github.com/apache/doris/pull/62891#discussion_r3178147286
##########
be/src/exec/pipeline/pipeline_task.cpp:
##########
@@ -1076,12 +1085,18 @@ void PipelineTask::wake_up(Dependency* dep, std::unique_lock<std::mutex>& /* dep
Status PipelineTask::_state_transition(State new_state) {
const auto& table =
_wake_up_early ? WAKE_UP_EARLY_LEGAL_STATE_TRANSITION : LEGAL_STATE_TRANSITION;
- if (!table[(int)new_state].contains(_exec_state)) {
+ auto current_state = _exec_state.load();
+ if (!table[(int)new_state].contains(current_state)) {
Review Comment:
非wake up early的逻辑先不管
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
Posted by "BiteTheDDDDt (via GitHub)" <gi...@apache.org>.
BiteTheDDDDt commented on PR #62891:
URL: https://github.com/apache/doris/pull/62891#issuecomment-4333149326
/review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
Posted by "BiteTheDDDDt (via GitHub)" <gi...@apache.org>.
BiteTheDDDDt commented on PR #62891:
URL: https://github.com/apache/doris/pull/62891#issuecomment-4332258290
Addressed the review comments:
1. Replaced the timing-based `sleep_for(100ms)` synchronization in the UT with a debug-point callback plus `std::promise/std::future`, so the main thread deterministically waits until the wake-up thread is paused before running close/finalize. Debug-point state is now restored with `Defer`.
2. Switched `_exec_state` load/CAS to explicit acquire/acq_rel memory ordering. `_state_transition()` is not in the block/row hot path, but this keeps the required publication semantics without relying on default seq-cst.
3. Removed the production `sleep(1)` debug-point delay. The debug point now only runs the test callback when one is registered, so manually enabling the point without a callback is a no-op instead of throwing.
Rechecked:
- `./run-be-ut.sh --run --filter='PipelineTaskTest.TEST_STATE_TRANSITION:PipelineTaskTest.TEST_STATE_TRANSITION_KEEP_FINALIZED_ON_WAKE_UP_RACE'`
- `build-support/check-format.sh`
`build-support/run-clang-tidy.sh` was retried; it still fails in this environment on pre-existing diagnostics in these historical files / `jni-util.h` analysis, unrelated to the review changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
Posted by "BiteTheDDDDt (via GitHub)" <gi...@apache.org>.
BiteTheDDDDt commented on PR #62891:
URL: https://github.com/apache/doris/pull/62891#issuecomment-4349809769
/review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #62891:
URL: https://github.com/apache/doris/pull/62891#discussion_r3178059768
##########
be/src/exec/pipeline/pipeline_task.cpp:
##########
@@ -1076,12 +1085,18 @@ void PipelineTask::wake_up(Dependency* dep, std::unique_lock<std::mutex>& /* dep
Status PipelineTask::_state_transition(State new_state) {
const auto& table =
_wake_up_early ? WAKE_UP_EARLY_LEGAL_STATE_TRANSITION : LEGAL_STATE_TRANSITION;
- if (!table[(int)new_state].contains(_exec_state)) {
+ auto current_state = _exec_state.load();
+ if (!table[(int)new_state].contains(current_state)) {
Review Comment:
This still leaves a race where a delayed `wake_up()` can overwrite a terminal state. `wake_up()` is not protected by `_dependency_lifecycle_lock` unless it is synchronously invoked from `unblock_all_dependencies()`: a normal `Dependency::set_ready()` can call it from another thread. In `_state_transition(RUNNABLE)`, that thread can load `BLOCKED` here and pass validation, then `close()`/`finalize()` can take the lifecycle lock, publish `FINISHED`/`FINALIZED`, and clear `_sink`/`_operators`; after that this same `_state_transition()` can continue with `need_move == true` (computed before the terminal transition) and store `_exec_state = RUNNABLE`. `wake_up()` then sees a non-terminal state and resubmits a task whose resources have already been released, which is the crash this PR is trying to prevent. Please make the read/check/write of `_exec_state` atomic with respect to concurrent terminal transitions, for example by restoring a CAS/retry transition or by using one lock for ever
y `_state_transition()` caller that can race with `wake_up()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
Posted by "BiteTheDDDDt (via GitHub)" <gi...@apache.org>.
BiteTheDDDDt commented on PR #62891:
URL: https://github.com/apache/doris/pull/62891#issuecomment-4366080621
/review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
Posted by "BiteTheDDDDt (via GitHub)" <gi...@apache.org>.
BiteTheDDDDt commented on PR #62891:
URL: https://github.com/apache/doris/pull/62891#issuecomment-4366080485
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
Posted by "BiteTheDDDDt (via GitHub)" <gi...@apache.org>.
BiteTheDDDDt commented on PR #62891:
URL: https://github.com/apache/doris/pull/62891#issuecomment-4349810155
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #62891:
URL: https://github.com/apache/doris/pull/62891#discussion_r3151430076
##########
be/test/exec/pipeline/pipeline_task_test.cpp:
##########
@@ -557,6 +560,54 @@ TEST_F(PipelineTaskTest, TEST_STATE_TRANSITION) {
}
}
+TEST_F(PipelineTaskTest, TEST_STATE_TRANSITION_KEEP_FINALIZED_ON_WAKE_UP_RACE) {
+ 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);
+ task->_exec_time_slice = 10'000'000'000ULL;
+
+ std::vector<TScanRangeParams> scan_range;
+ int sender_id = 0;
+ TDataSink tsink;
+ EXPECT_TRUE(task->prepare(scan_range, sender_id, tsink).ok());
+ task->_wake_up_early = true;
+ task->_exec_state = PipelineTask::State::BLOCKED;
+ config::enable_debug_points = true;
+ DebugPoints::instance()->add("PipelineTask::_state_transition.before_runnable_cas");
+
+ std::thread wake_up_thread(
+ [&]() { EXPECT_TRUE(task->_state_transition(PipelineTask::State::RUNNABLE).ok()); });
+ std::this_thread::sleep_for(std::chrono::milliseconds(100));
+ EXPECT_TRUE(task->close(Status::OK()).ok());
+ EXPECT_TRUE(task->finalize().ok());
+ wake_up_thread.join();
+
+ EXPECT_EQ(task->_exec_state, PipelineTask::State::FINALIZED);
+ EXPECT_EQ(task->_sink, nullptr);
+ EXPECT_TRUE(task->_operators.empty());
+ DebugPoints::instance()->remove("PipelineTask::_state_transition.before_runnable_cas");
+ config::enable_debug_points = false;
Review Comment:
The test’s synchronization relies on a fixed `sleep_for(100ms)` to “let the wake_up thread reach the debug point”, which is timing-dependent and can be flaky under load/slow CI. Consider using an explicit synchronization mechanism (e.g., `std::promise/std::future`, `std::latch`, or a debug-point facility that signals when the point is hit) so the main thread deterministically waits until the wake-up thread is paused at the intended location. Also, `enable_debug_points` and the added debug point should be reset via a scoped guard/RAII to ensure cleanup even if an `EXPECT_*` fails mid-test (to avoid leaking debug-point state into later tests).
##########
be/src/exec/pipeline/pipeline_task.cpp:
##########
@@ -1074,29 +1074,47 @@ void PipelineTask::wake_up(Dependency* dep, std::unique_lock<std::mutex>& /* dep
}
Status PipelineTask::_state_transition(State new_state) {
- const auto& table =
- _wake_up_early ? WAKE_UP_EARLY_LEGAL_STATE_TRANSITION : LEGAL_STATE_TRANSITION;
- if (!table[(int)new_state].contains(_exec_state)) {
- return Status::InternalError(
- "Task state transition from {} to {} is not allowed! Task info: {}",
- _to_string(_exec_state), _to_string(new_state), debug_string());
- }
- // FINISHED/FINALIZED → RUNNABLE is legal under wake_up_early (delayed wake_up() arriving
- // after the task already terminated), but we must not actually move the state backwards
- // or update profile info (which would misleadingly show RUNNABLE for a terminated task).
- bool need_move = !((_exec_state == State::FINISHED || _exec_state == State::FINALIZED) &&
- new_state == State::RUNNABLE);
- if (need_move) {
- if (_exec_state != new_state) {
+ // State transition is a read-check-write sequence: validate the current state, then publish
+ // the next one. Use CAS so a concurrent close()/finalize() cannot be overwritten by a stale
+ // wake_up(); the uncontended path succeeds once, and retries only revalidate real races.
+ while (true) {
+ auto old_state = _exec_state.load();
+ const auto& table =
+ _wake_up_early ? WAKE_UP_EARLY_LEGAL_STATE_TRANSITION : LEGAL_STATE_TRANSITION;
+ if (!table[(int)new_state].contains(old_state)) {
+ return Status::InternalError(
+ "Task state transition from {} to {} is not allowed! Task info: {}",
+ _to_string(old_state), _to_string(new_state), debug_string());
+ }
+ // FINISHED/FINALIZED → RUNNABLE is legal under wake_up_early (delayed wake_up() arriving
+ // after the task already terminated), but it must be an atomic no-op because finalize()
+ // resets _sink/_operators.
+ if ((old_state == State::FINISHED || old_state == State::FINALIZED) &&
+ new_state == State::RUNNABLE) {
+ return Status::OK();
+ }
+ if (old_state == new_state) {
+ _task_profile->add_info_string("TaskState", _to_string(new_state));
+ _task_profile->add_info_string("BlockedByDependency",
+ _blocked_dep ? _blocked_dep->name() : "");
+ return Status::OK();
+ }
+
+ DBUG_EXECUTE_IF("PipelineTask::_state_transition.before_runnable_cas", {
+ if (new_state == State::RUNNABLE && old_state == State::BLOCKED) {
+ sleep(1);
+ }
+ });
+
+ if (_exec_state.compare_exchange_strong(old_state, new_state)) {
_state_change_watcher.reset();
_state_change_watcher.start();
+ _task_profile->add_info_string("TaskState", _to_string(new_state));
+ _task_profile->add_info_string("BlockedByDependency",
+ _blocked_dep ? _blocked_dep->name() : "");
+ return Status::OK();
}
- _task_profile->add_info_string("TaskState", _to_string(new_state));
- _task_profile->add_info_string("BlockedByDependency",
- _blocked_dep ? _blocked_dep->name() : "");
- _exec_state = new_state;
}
Review Comment:
`std::atomic::load()` / `compare_exchange_strong()` are currently using the default memory order (seq-cst). If `_state_transition()` is on a hot path, this can add unnecessary fencing costs. Consider specifying explicit memory ordering (e.g., `load(std::memory_order_acquire)` and `compare_exchange_strong(expected, desired, std::memory_order_acq_rel, std::memory_order_acquire)`), which typically preserves the needed happens-before for state publication while reducing overhead versus seq-cst.
##########
be/src/exec/pipeline/pipeline_task.cpp:
##########
@@ -1074,29 +1074,47 @@ void PipelineTask::wake_up(Dependency* dep, std::unique_lock<std::mutex>& /* dep
}
Status PipelineTask::_state_transition(State new_state) {
- const auto& table =
- _wake_up_early ? WAKE_UP_EARLY_LEGAL_STATE_TRANSITION : LEGAL_STATE_TRANSITION;
- if (!table[(int)new_state].contains(_exec_state)) {
- return Status::InternalError(
- "Task state transition from {} to {} is not allowed! Task info: {}",
- _to_string(_exec_state), _to_string(new_state), debug_string());
- }
- // FINISHED/FINALIZED → RUNNABLE is legal under wake_up_early (delayed wake_up() arriving
- // after the task already terminated), but we must not actually move the state backwards
- // or update profile info (which would misleadingly show RUNNABLE for a terminated task).
- bool need_move = !((_exec_state == State::FINISHED || _exec_state == State::FINALIZED) &&
- new_state == State::RUNNABLE);
- if (need_move) {
- if (_exec_state != new_state) {
+ // State transition is a read-check-write sequence: validate the current state, then publish
+ // the next one. Use CAS so a concurrent close()/finalize() cannot be overwritten by a stale
+ // wake_up(); the uncontended path succeeds once, and retries only revalidate real races.
+ while (true) {
+ auto old_state = _exec_state.load();
+ const auto& table =
+ _wake_up_early ? WAKE_UP_EARLY_LEGAL_STATE_TRANSITION : LEGAL_STATE_TRANSITION;
+ if (!table[(int)new_state].contains(old_state)) {
+ return Status::InternalError(
+ "Task state transition from {} to {} is not allowed! Task info: {}",
+ _to_string(old_state), _to_string(new_state), debug_string());
+ }
+ // FINISHED/FINALIZED → RUNNABLE is legal under wake_up_early (delayed wake_up() arriving
+ // after the task already terminated), but it must be an atomic no-op because finalize()
+ // resets _sink/_operators.
+ if ((old_state == State::FINISHED || old_state == State::FINALIZED) &&
+ new_state == State::RUNNABLE) {
+ return Status::OK();
+ }
+ if (old_state == new_state) {
+ _task_profile->add_info_string("TaskState", _to_string(new_state));
+ _task_profile->add_info_string("BlockedByDependency",
+ _blocked_dep ? _blocked_dep->name() : "");
+ return Status::OK();
+ }
+
+ DBUG_EXECUTE_IF("PipelineTask::_state_transition.before_runnable_cas", {
+ if (new_state == State::RUNNABLE && old_state == State::BLOCKED) {
+ sleep(1);
+ }
+ });
Review Comment:
The debug-point delay uses `sleep(1)`, which (a) makes the unit test incur a full 1s delay when enabled and (b) ties this code to a C/POSIX sleep API. Consider switching to `std::this_thread::sleep_for(...)` with a smaller duration (only needs to be “long enough” for the test to reach close/finalize) to speed up the test while keeping the race deterministic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
Posted by "BiteTheDDDDt (via GitHub)" <gi...@apache.org>.
BiteTheDDDDt commented on PR #62891:
URL: https://github.com/apache/doris/pull/62891#issuecomment-4344995299
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #62891:
URL: https://github.com/apache/doris/pull/62891#issuecomment-4384893834
PR approved by at least one committer and no changes requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [fix](be) Prevent finalized pipeline task resubmission [doris]
Posted by "BiteTheDDDDt (via GitHub)" <gi...@apache.org>.
BiteTheDDDDt merged PR #62891:
URL: https://github.com/apache/doris/pull/62891
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org