This is an automated email from the ASF dual-hosted git repository.
HappenLee 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 feb9e04f784 [bugfix](cancel) cancel message is printed too many times
(#67344)
feb9e04f784 is described below
commit feb9e04f78490296c3393cbd594aef617af6b433
Author: yiguolei <[email protected]>
AuthorDate: Tue Sep 1 15:09:45 2026 +0800
[bugfix](cancel) cancel message is printed too many times (#67344)
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
When a query times out, every pending pipeline task can close with the
same error and call PipelineFragmentContext::cancel() before the
fragment task count is drained. Each call previously repeated fragment
cancellation logs, the full timeout task dump, stream-pipe cancellation,
and dependency wakeups, causing severe log amplification.
### 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 -->
---
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/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 +-
9 files changed, 48 insertions(+), 41 deletions(-)
diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp
b/be/src/exec/pipeline/pipeline_fragment_context.cpp
index 2fa064c8a68..f62bd073081 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp
@@ -204,10 +204,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))
@@ -241,7 +241,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);
@@ -2188,7 +2187,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;
@@ -2656,7 +2655,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());
Status submit_status =
_exec_env->fragment_mgr()->get_thread_pool()->submit_func([this,
req, ctx]() {
diff --git a/be/src/exec/pipeline/pipeline_fragment_context.h
b/be/src/exec/pipeline/pipeline_fragment_context.h
index 7243b0214d9..0d60375a9b0 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.h
+++ b/be/src/exec/pipeline/pipeline_fragment_context.h
@@ -84,8 +84,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; }
@@ -152,6 +150,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);
void _append_external_file_commit_data(const ReportStatusRequest& req,
TReportExecStatusParams* params)
const;
diff --git a/be/src/exec/pipeline/pipeline_task.cpp
b/be/src/exec/pipeline/pipeline_task.cpp
index 6c7f2d7f56e..515f10a6657 100644
--- a/be/src/exec/pipeline/pipeline_task.cpp
+++ b/be/src/exec/pipeline/pipeline_task.cpp
@@ -1071,7 +1071,7 @@ void PipelineTask::wake_up(Dependency* dep,
std::unique_lock<std::mutex>& /* dep
auto cancel_if_error = [&](const Status& st) {
if (!st.ok()) {
if (auto frag = fragment_context().lock()) {
- frag->cancel(st);
+ frag->get_query_ctx()->cancel(st);
}
}
};
diff --git a/be/src/exec/pipeline/task_scheduler.cpp
b/be/src/exec/pipeline/task_scheduler.cpp
index 51e890c7605..3bfbb965519 100644
--- a/be/src/exec/pipeline/task_scheduler.cpp
+++ b/be/src/exec/pipeline/task_scheduler.cpp
@@ -77,17 +77,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 ca2a700900b..d78c9571e21 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -665,7 +665,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;
}
}
@@ -1350,7 +1350,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 62643c618c6..00b59ae14f1 100644
--- a/be/src/runtime/query_context.cpp
+++ b/be/src/runtime/query_context.cpp
@@ -305,7 +305,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;
}
@@ -343,7 +343,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) {
@@ -366,24 +380,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 049709dbd71..5a79489b58f 100644
--- a/be/src/runtime/query_context.h
+++ b/be/src/runtime/query_context.h
@@ -136,11 +136,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 b089d772610..dfcb3380618 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 03d272eced1..4b11c042540 100644
--- a/be/src/runtime/workload_management/query_task_controller.h
+++ b/be/src/runtime/workload_management/query_task_controller.h
@@ -34,8 +34,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]
