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 d8d8152183b [branch-4.1](cherry-pick) cancel message is printed too
many times (#67548)
d8d8152183b is described below
commit d8d8152183b81caa4e94aa79908f7b62bde15b92
Author: yiguolei <[email protected]>
AuthorDate: Mon Sep 7 09:15:36 2026 +0800
[branch-4.1](cherry-pick) cancel message is printed too many times (#67548)
### What problem does this PR solve?
Issue Number: close #xxx
pick #56822 #67344
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---------
Co-authored-by: Gabriel <[email protected]>
---
be/src/exec/pipeline/dependency.cpp | 4 +--
be/src/exec/pipeline/pipeline_fragment_context.cpp | 15 ++++-----
be/src/exec/pipeline/pipeline_fragment_context.h | 10 ++++--
be/src/exec/pipeline/pipeline_task.cpp | 2 +-
be/src/exec/pipeline/pipeline_task.h | 4 +--
be/src/exec/pipeline/task_scheduler.cpp | 10 ++++--
be/src/runtime/fragment_mgr.cpp | 4 +--
be/src/runtime/query_context.cpp | 36 ++++++++++------------
be/src/runtime/query_context.h | 5 +--
.../workload_management/query_task_controller.cpp | 4 +--
.../workload_management/query_task_controller.h | 3 +-
11 files changed, 52 insertions(+), 45 deletions(-)
diff --git a/be/src/exec/pipeline/dependency.cpp
b/be/src/exec/pipeline/dependency.cpp
index 014f3182183..c8f9d3d3ba2 100644
--- a/be/src/exec/pipeline/dependency.cpp
+++ b/be/src/exec/pipeline/dependency.cpp
@@ -79,7 +79,7 @@ void Dependency::set_ready() {
for (auto task : local_block_task) {
if (auto t = task.lock()) {
std::unique_lock<std::mutex> lc(_task_lock);
- THROW_IF_ERROR(t->wake_up(this));
+ THROW_IF_ERROR(t->wake_up(this, lc));
}
}
}
@@ -90,7 +90,7 @@ Dependency*
Dependency::is_blocked_by(std::shared_ptr<PipelineTask> task) {
if (!ready && task) {
_add_block_task(task);
start_watcher();
- THROW_IF_ERROR(task->blocked(this));
+ THROW_IF_ERROR(task->blocked(this, lc));
}
return ready ? nullptr : this;
}
diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp
b/be/src/exec/pipeline/pipeline_fragment_context.cpp
index 4193927e395..06b617b8339 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp
@@ -203,10 +203,10 @@ bool PipelineFragmentContext::notify_close() {
return all_closed;
}
-// Must not add lock in this method. Because it will call query ctx cancel. And
-// QueryCtx cancel will call fragment ctx cancel. And Also Fragment ctx's
running
-// Method like exchange sink buffer will call query ctx cancel. If we add lock
here
-// There maybe dead lock.
+// QueryContext is the sole cancellation entry point and invokes this method
to apply the accepted
+// cancellation to this fragment. Do not call QueryContext::cancel() from
here: doing so would make
+// cancellation bidirectional and allow every task closed with the query error
to repeat the whole
+// fragment's timeout diagnostics and dependency-unblocking work.
void PipelineFragmentContext::cancel(const Status reason) {
LOG_INFO("PipelineFragmentContext::cancel")
.tag("query_id", print_id(_query_id))
@@ -240,7 +240,6 @@ void PipelineFragmentContext::cancel(const Status reason) {
_query_ctx->set_first_error_msg(first_error_msg);
}
- _query_ctx->cancel(reason, _fragment_id);
if (!reason.is<ErrorCode::LIMIT_REACH>() &&
!reason.is<ErrorCode::FINISHED>()) {
for (auto& id : _fragment_instance_ids) {
LOG(WARNING) << "PipelineFragmentContext cancel instance: " <<
print_id(id);
@@ -1826,7 +1825,7 @@ Status PipelineFragmentContext::submit() {
DBUG_EXECUTE_IF("PipelineFragmentContext.submit.failed",
{ st =
Status::Aborted("PipelineFragmentContext.submit.failed"); });
if (!st) {
- cancel(Status::InternalError("submit context to executor
fail"));
+ _query_ctx->cancel(Status::InternalError("submit context to
executor fail"));
std::lock_guard<std::mutex> l(_task_mutex);
_total_tasks = submit_tasks;
break;
@@ -2287,7 +2286,9 @@ Status PipelineFragmentContext::send_report(bool done) {
.runtime_state = _runtime_state.get(),
.load_error_url = load_eror_url,
.first_error_msg = first_error_msg,
- .cancel_fn = [this](const Status& reason) {
cancel(reason); }};
+ .cancel_fn = [query_ctx = _query_ctx](const
Status& reason) {
+ query_ctx->cancel(reason);
+ }};
auto ctx =
std::dynamic_pointer_cast<PipelineFragmentContext>(shared_from_this());
return _exec_env->fragment_mgr()->get_thread_pool()->submit_func([this,
req, ctx]() {
SCOPED_ATTACH_TASK(ctx->get_query_ctx()->query_mem_tracker());
diff --git a/be/src/exec/pipeline/pipeline_fragment_context.h
b/be/src/exec/pipeline/pipeline_fragment_context.h
index f8a1bfea229..f5cb2fc22f9 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.h
+++ b/be/src/exec/pipeline/pipeline_fragment_context.h
@@ -81,8 +81,6 @@ public:
void set_is_report_success(bool is_report_success) { _is_report_success =
is_report_success; }
- void cancel(const Status reason);
-
bool notify_close();
TUniqueId get_query_id() const { return _query_id; }
@@ -149,6 +147,14 @@ public:
}
private:
+ // QueryContext is the sole entry point for query cancellation. Keep
fragment cancellation
+ // private so callers cannot bypass QueryContext's first-error-wins guard
and repeatedly run
+ // expensive fragment-local cleanup (for example, timeout diagnostics and
task unblocking).
+ // QueryContext::cancel() calls this method only to propagate the accepted
query cancellation
+ // to each fragment; this method must not call QueryContext::cancel() back.
+ friend void QueryContext::cancel(Status new_status);
+ void cancel(const Status reason);
+
void _coordinator_callback(const ReportStatusRequest& req);
std::string _to_http_path(const std::string& file_name) const;
diff --git a/be/src/exec/pipeline/pipeline_task.cpp
b/be/src/exec/pipeline/pipeline_task.cpp
index c898024c915..b364eaee3a7 100644
--- a/be/src/exec/pipeline/pipeline_task.cpp
+++ b/be/src/exec/pipeline/pipeline_task.cpp
@@ -1055,7 +1055,7 @@ Status PipelineTask::revoke_memory(const
std::shared_ptr<SpillContext>& spill_co
return Status::OK();
}
-Status PipelineTask::wake_up(Dependency* dep) {
+Status PipelineTask::wake_up(Dependency* dep, std::unique_lock<std::mutex>& /*
dep_lock */) {
// call by dependency
DCHECK_EQ(_blocked_dep, dep) << "dep : " << dep->debug_string(0) << "task:
" << debug_string();
_blocked_dep = nullptr;
diff --git a/be/src/exec/pipeline/pipeline_task.h
b/be/src/exec/pipeline/pipeline_task.h
index 68d2195e559..5b5da9dca83 100644
--- a/be/src/exec/pipeline/pipeline_task.h
+++ b/be/src/exec/pipeline/pipeline_task.h
@@ -119,7 +119,7 @@ public:
return _op_shared_states[id].get();
}
- Status wake_up(Dependency* dep);
+ Status wake_up(Dependency* dep, std::unique_lock<std::mutex>& /* dep_lock
*/);
DataSinkOperatorPtr sink() const { return _sink; }
@@ -173,7 +173,7 @@ public:
[[nodiscard]] size_t get_revocable_size() const;
[[nodiscard]] Status revoke_memory(const std::shared_ptr<SpillContext>&
spill_context);
- Status blocked(Dependency* dependency) {
+ Status blocked(Dependency* dependency, std::unique_lock<std::mutex>& /*
dep_lock */) {
DCHECK_EQ(_blocked_dep, nullptr) << "task: " << debug_string();
_blocked_dep = dependency;
return _state_transition(PipelineTask::State::BLOCKED);
diff --git a/be/src/exec/pipeline/task_scheduler.cpp
b/be/src/exec/pipeline/task_scheduler.cpp
index 1a052fc1ffa..3a1274defcb 100644
--- a/be/src/exec/pipeline/task_scheduler.cpp
+++ b/be/src/exec/pipeline/task_scheduler.cpp
@@ -81,17 +81,21 @@ void close_task(PipelineTask* task, Status exec_status,
PipelineFragmentContext*
// task finished.
SCOPED_ATTACH_TASK(task->runtime_state());
if (!exec_status.ok()) {
- ctx->cancel(exec_status);
+ // Always enter cancellation through QueryContext. The status passed
while closing a task
+ // may be the query's existing cancellation status rather than a new
task failure; the
+ // QueryContext first-error-wins guard makes that case a no-op and
prevents repeated
+ // fragment-local cancellation work.
+ ctx->get_query_ctx()->cancel(exec_status);
LOG(WARNING) << fmt::format("Pipeline task failed. query_id: {}
reason: {}",
print_id(ctx->get_query_id()),
exec_status.to_string());
}
Status status = task->close(exec_status);
if (!status.ok()) {
- ctx->cancel(status);
+ ctx->get_query_ctx()->cancel(status);
}
status = task->finalize();
if (!status.ok()) {
- ctx->cancel(status);
+ ctx->get_query_ctx()->cancel(status);
}
}
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index c8efc4b8ed5..55aad40baac 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -655,7 +655,7 @@ Status FragmentMgr::exec_plan_fragment(const
TPipelineFragmentParams& params,
prepare_st =
Status::Aborted("FragmentMgr.exec_plan_fragment.prepare_failed");
});
if (!prepare_st.ok()) {
- query_ctx->cancel(prepare_st, params.fragment_id);
+ query_ctx->cancel(prepare_st);
return prepare_st;
}
}
@@ -1322,7 +1322,7 @@ Status FragmentMgr::rerun_fragment(const
std::shared_ptr<brpc::ClosureGuard>& gu
ASSIGN_STATUS_IF_CATCH_EXCEPTION(prepare_st =
context->prepare(_thread_pool.get()),
prepare_st);
if (!prepare_st.ok()) {
- q_ctx->cancel(prepare_st, info.params.fragment_id);
+ q_ctx->cancel(prepare_st);
return prepare_st;
}
diff --git a/be/src/runtime/query_context.cpp b/be/src/runtime/query_context.cpp
index aaab0ad698c..c5e9b277093 100644
--- a/be/src/runtime/query_context.cpp
+++ b/be/src/runtime/query_context.cpp
@@ -315,7 +315,7 @@ void QueryContext::set_memory_sufficient(bool sufficient) {
}
}
-void QueryContext::cancel(Status new_status, int fragment_id) {
+void QueryContext::cancel(Status new_status) {
if (!_exec_status.update(new_status)) {
return;
}
@@ -353,7 +353,21 @@ void QueryContext::cancel(Status new_status, int
fragment_id) {
}
set_ready_to_execute(new_status);
- cancel_all_pipeline_context(new_status, fragment_id);
+
+ // Copy the fragment contexts under the map lock, then cancel them after
releasing it. Fragment
+ // cancellation may take task-level locks and must not run while holding
the query map lock.
+ std::vector<std::weak_ptr<PipelineFragmentContext>> ctx_to_cancel;
+ {
+ std::lock_guard<std::mutex> lock(_pipeline_map_write_lock);
+ for (auto& entry : _fragment_id_to_pipeline_ctx) {
+ ctx_to_cancel.push_back(entry.second);
+ }
+ }
+ for (auto& f_context : ctx_to_cancel) {
+ if (auto pipeline_ctx = f_context.lock()) {
+ pipeline_ctx->cancel(new_status);
+ }
+ }
}
void QueryContext::set_load_error_url(std::string error_url) {
@@ -376,24 +390,6 @@ std::string QueryContext::get_first_error_msg() {
return _first_error_msg;
}
-void QueryContext::cancel_all_pipeline_context(const Status& reason, int
fragment_id) {
- std::vector<std::weak_ptr<PipelineFragmentContext>> ctx_to_cancel;
- {
- std::lock_guard<std::mutex> lock(_pipeline_map_write_lock);
- for (auto& [f_id, f_context] : _fragment_id_to_pipeline_ctx) {
- if (fragment_id == f_id) {
- continue;
- }
- ctx_to_cancel.push_back(f_context);
- }
- }
- for (auto& f_context : ctx_to_cancel) {
- if (auto pipeline_ctx = f_context.lock()) {
- pipeline_ctx->cancel(reason);
- }
- }
-}
-
std::string QueryContext::print_all_pipeline_context() {
std::vector<std::weak_ptr<PipelineFragmentContext>> ctx_to_print;
fmt::memory_buffer debug_string_buffer;
diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h
index e0368def54e..de6b416cf14 100644
--- a/be/src/runtime/query_context.h
+++ b/be/src/runtime/query_context.h
@@ -135,11 +135,12 @@ public:
[[nodiscard]] bool is_cancelled() const { return !_exec_status.ok(); }
- void cancel_all_pipeline_context(const Status& reason, int fragment_id =
-1);
std::string print_all_pipeline_context();
void set_pipeline_context(const int fragment_id,
std::shared_ptr<PipelineFragmentContext>
pip_ctx);
- void cancel(Status new_status, int fragment_id = -1);
+ // The sole entry point for query cancellation. Only the first error is
accepted; it is then
+ // propagated to every PipelineFragmentContext for fragment-local cleanup.
+ void cancel(Status new_status);
[[nodiscard]] Status exec_status() { return _exec_status.status(); }
diff --git a/be/src/runtime/workload_management/query_task_controller.cpp
b/be/src/runtime/workload_management/query_task_controller.cpp
index de6b3e8a24f..8708429b2bd 100644
--- a/be/src/runtime/workload_management/query_task_controller.cpp
+++ b/be/src/runtime/workload_management/query_task_controller.cpp
@@ -39,12 +39,12 @@ bool QueryTaskController::is_cancelled() const {
return query_ctx->is_cancelled();
}
-bool QueryTaskController::cancel_impl(const Status& reason, int fragment_id) {
+bool QueryTaskController::cancel_impl(const Status& reason) {
auto query_ctx = query_ctx_.lock();
if (query_ctx == nullptr) {
return false;
}
- query_ctx->cancel(reason, fragment_id);
+ query_ctx->cancel(reason);
return true;
}
diff --git a/be/src/runtime/workload_management/query_task_controller.h
b/be/src/runtime/workload_management/query_task_controller.h
index 854a2c05cb6..7fe354681b8 100644
--- a/be/src/runtime/workload_management/query_task_controller.h
+++ b/be/src/runtime/workload_management/query_task_controller.h
@@ -35,8 +35,7 @@ public:
~QueryTaskController() override = default;
bool is_cancelled() const override;
- bool cancel_impl(const Status& reason, int fragment_id);
- bool cancel_impl(const Status& reason) override { return
cancel_impl(reason, -1); }
+ bool cancel_impl(const Status& reason) override;
bool is_pure_load_task() const override;
int32_t get_slot_count() const override;
void disable_reserve_memory() override;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
