yiguolei commented on code in PR #49281:
URL: https://github.com/apache/doris/pull/49281#discussion_r2005375842
##########
be/src/pipeline/dependency.cpp:
##########
@@ -68,32 +68,42 @@ void Dependency::set_ready() {
local_block_task.swap(_blocked_task);
}
for (auto* task : local_block_task) {
- task->wake_up();
+ {
+ std::unique_lock<std::mutex> lc(_task_lock);
+ DCHECK_EQ(task->_blocked_dep, this)
+ << "dep : " << debug_string(0) << "task: " <<
debug_string();
+ task->_blocked_dep = nullptr;
+
THROW_IF_ERROR(task->_state_transition(PipelineTask::State::RUNNABLE));
+ }
+ task->wake_up(this);
}
}
-Dependency* Dependency::is_blocked_by(PipelineTask* task) {
+bool Dependency::is_blocked_by(PipelineTask* task) {
std::unique_lock<std::mutex> lc(_task_lock);
auto ready = _ready.load();
if (!ready && task) {
_add_block_task(task);
+ DCHECK_EQ(task->_blocked_dep, nullptr) << "task: " <<
task->debug_string();
+ task->_blocked_dep = this;
+ start_watcher();
+ THROW_IF_ERROR(task->_state_transition(PipelineTask::State::BLOCKED));
}
- return ready ? nullptr : this;
+ return !ready;
}
-Dependency* QueryGlobalDependency::is_blocked_by(PipelineTask* task) {
+bool QueryGlobalDependency::is_blocked_by(PipelineTask* task) {
if (task && task->wake_up_early()) {
Review Comment:
而且这种逻辑就很诡异，为什么只有QueryGlobalDependency 需要感知的到wakeup early，而不是所有的dependency
都感知呢？
--
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-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
