github-actions[bot] commented on code in PR #66385:
URL: https://github.com/apache/doris/pull/66385#discussion_r3704341402
##########
be/src/pipeline/task_scheduler.cpp:
##########
@@ -264,16 +264,23 @@ void _close_task(PipelineTask* task, PipelineTaskState
state, Status exec_status
void TaskScheduler::_do_work(size_t index) {
const auto& marker = _markers[index];
while (*marker) {
- auto* task = _task_queue->take(index);
- if (!task) {
+ auto task_holder = _task_queue->take(index);
+ if (!task_holder) {
continue;
}
- if (task->is_pipelineX() && task->is_running()) {
+ auto* task = task_holder.get();
+ if (task->is_pipelineX() && task->set_running(true)) {
static_cast<void>(_task_queue->push_back(task, index));
Review Comment:
[P1] Do not recycle a wake token across task-state epochs
Requeuing the losing entry preserves it without tying it to the dependency
epoch that created it. A delayed duplicate can sit in the queue while the owner
runs; if the owner then reaches EOS, `is_pending_finish()` registers an unready
finish dependency and the owner publishes `PENDING_FINISH` before clearing
`_running`. This stale entry can now claim the task, and the `PENDING_FINISH`
branch calls `_close_task()` without rechecking finish dependencies for
PipelineX, releasing async-writer/exchange state before the callback makes the
dependency ready. If the owner instead enters an ordinary blocked state, the
stale entry re-enters the same unready dependency and trips `_add_block_task`'s
duplicate DCHECK (or appends duplicate waiters in release builds). Please
coalesce or generation-tag pending wakes so an old token cannot satisfy a later
wait epoch, and add a latched regression covering both transitions.
##########
be/src/pipeline/task_scheduler.cpp:
##########
@@ -264,16 +264,23 @@ void _close_task(PipelineTask* task, PipelineTaskState
state, Status exec_status
void TaskScheduler::_do_work(size_t index) {
const auto& marker = _markers[index];
while (*marker) {
- auto* task = _task_queue->take(index);
- if (!task) {
+ auto task_holder = _task_queue->take(index);
+ if (!task_holder) {
continue;
}
- if (task->is_pipelineX() && task->is_running()) {
+ auto* task = task_holder.get();
+ if (task->is_pipelineX() && task->set_running(true)) {
Review Comment:
[P2] Claim or coalesce duplicates before touching task accounting
The atomic claim happens only after `take()` has updated this task's
`_queue_level`/`_core_id` and stopped its wait-worker watcher. When this branch
loses, it immediately pushes the same token again, so an idle worker can
hot-loop pop/push for the owner's full execution slice (or longer close) while
`push()` reads `_runtime`, increments `_schedule_time`, and restarts the same
plain stopwatch. Meanwhile the owner calls `update_statistics()`, which writes
`_runtime` and reads `_core_id`/`_queue_level`; per-core queue locks do not
synchronize these task fields, especially after steals. This leaves C++ data
races plus corrupted priority/profile accounting even though `execute()` is
serialized. Please claim/coalesce before per-task dequeue accounting and add a
latched multi-worker duplicate test.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
