Skip to content

branch-4.1: [fix](rpc) Fix AutoReleaseClosure data race with callback reuse (#61782)#67340

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
linrrzqqq wants to merge 1 commit into
base:branch-4.1
Choose a base branch
from
+551 −157
Open

branch-4.1: [fix](rpc) Fix AutoReleaseClosure data race with callback reuse (#61782) #67340

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions be/src/exec/exchange/vdata_stream_sender.h
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -164,6 +164,7 @@ class Channel {

std::shared_ptr<ExchangeSendCallback<PTransmitDataResult>> get_send_callback(RpcInstance* ins,
bool eos) {
// here we reuse the callback because it's re-construction may be expensive due to many parameters' capture
if (!_send_callback) {
_send_callback = ExchangeSendCallback<PTransmitDataResult>::create_shared();
} else {
Expand Down
3 changes: 2 additions & 1 deletion be/src/exec/operator/exchange_sink_buffer.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -346,6 +346,7 @@ Status ExchangeSinkBuffer::_send_rpc(RpcInstance& instance_data) {
}
// The eos here only indicates that the current exchange sink has reached eos.
// However, the queue still contains data from other exchange sinks, so RPCs need to continue being sent.
// `_send_rpc` must be the LAST operation in this function, because it may reuse the callback!

@github-actionsgithub-actionsBotSep 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Keep the active handler alive across reentrant reuse

This call is not actually the last operation in the handler: on a synchronous failure from the nested send, lines 351-353 still use this lambda's captures. For another queued block on the same channel, _send_rpc() reuses the same ExchangeSendCallback and addSuccessHandler() assigns over _suc_fn while that very target is executing. If HTTP attachment/DNS/client setup then returns an error before launching the RPC, execution comes back here and calls _failed through a lambda whose stored target has already been destroyed. Please keep a local copy of the selected handler before invoking it (or defer handler replacement until it returns); the broadcast branch has the same pattern.

s = _send_rpc(ins);
if (!s) {
_failed(ins.id,
Expand DownExpand Up@@ -472,9 +473,9 @@ Status ExchangeSinkBuffer::_send_rpc(RpcInstance& instance_data) {
} else if (eos) {
_ended(ins);
}

// The eos here only indicates that the current exchange sink has reached eos.
// However, the queue still contains data from other exchange sinks, so RPCs need to continue being sent.
// `_send_rpc` must be the LAST operation in this function, because it may reuse the callback!
s = _send_rpc(ins);
if (!s) {
_failed(ins.id,
Expand Down
23 changes: 12 additions & 11 deletions be/src/exec/runtime_filter/runtime_filter.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -36,13 +36,12 @@ Status RuntimeFilter::_push_to_remote(RuntimeState* state, const TNetworkAddress

auto merge_filter_request = std::make_shared<PMergeFilterRequest>();
merge_filter_request->set_stage(_stage);
auto merge_filter_callback = DummyBrpcCallback<PMergeFilterResponse>::create_shared();
_merge_filter_callback = HandleErrorBrpcCallback<PMergeFilterResponse>::create_shared(

@github-actionsgithub-actionsBotSep 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Release the callback/controller after RPC completion

Storing this callback on the filter also keeps its brpc::Controller alive after Run(); the serialized filter is appended to that controller's request_attachment() below. In the pinned brpc 1.4.0 implementation the request attachment is cleared by controller reset/destruction, not by normal RPC completion, and there is no completion-path reset here. The same pattern appears in GlobalMergeContext::publish_callbacks, so successful merge/direct-publish RPCs retain an extra serialized bloom-filter buffer until a CTE reset or query teardown (potentially tens of MiB per filter during a long probe). Please use completion-scoped ownership like RuntimeFilterRelayRpcClosure, or otherwise release/reset each owner safely at the end of its callback.

state->query_options().ignore_runtime_filter_error ? std::weak_ptr<QueryContext> {}
: state->get_query_ctx_weak());
auto merge_filter_closure =
AutoReleaseClosure<PMergeFilterRequest, DummyBrpcCallback<PMergeFilterResponse>>::
create_unique(merge_filter_request, merge_filter_callback,
state->query_options().ignore_runtime_filter_error
? std::weak_ptr<QueryContext> {}
: state->get_query_ctx_weak());
AutoReleaseClosure<PMergeFilterRequest, HandleErrorBrpcCallback<PMergeFilterResponse>>::
create_unique(merge_filter_request, _merge_filter_callback);
void* data = nullptr;
int len = 0;

Expand All@@ -54,19 +53,21 @@ Status RuntimeFilter::_push_to_remote(RuntimeState* state, const TNetworkAddress
pfragment_instance_id->set_hi(BackendOptions::get_local_backend().id);
pfragment_instance_id->set_lo((int64_t)this);

merge_filter_callback->cntl_->set_timeout_ms(
_merge_filter_callback->cntl_->set_timeout_ms(
get_execution_rpc_timeout_ms(state->get_query_ctx()->execution_timeout()));
if (config::execution_ignore_eovercrowded) {
merge_filter_callback->cntl_->ignore_eovercrowded();
_merge_filter_callback->cntl_->ignore_eovercrowded();
}

RETURN_IF_ERROR(serialize(merge_filter_request.get(), &data, &len));

if (len > 0) {
DCHECK(data != nullptr);
merge_filter_callback->cntl_->request_attachment().append(data, len);
if (data == nullptr) {
return Status::InternalError(
"data is nullptr after serialization with len > 0, filter: {}", debug_string());
}
_merge_filter_callback->cntl_->request_attachment().append(data, len);
}

stub->merge_filter(merge_filter_closure->cntl_.get(), merge_filter_closure->request_.get(),
merge_filter_closure->response_.get(), merge_filter_closure.get());
// the closure will be released by brpc during closure->Run.
Expand Down
7 changes: 7 additions & 0 deletions be/src/exec/runtime_filter/runtime_filter.h
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -19,6 +19,8 @@

#include <gen_cpp/PaloInternalService_types.h>

#include <vector>

#include "common/exception.h"
#include "common/status.h"
#include "exec/runtime_filter/runtime_filter_definitions.h"
Expand All@@ -28,8 +30,11 @@

namespace doris {
#include "common/compile_check_begin.h"
class PMergeFilterResponse;
class RuntimeFilterWrapper;
class RuntimeProfile;
template <typename Response>
class HandleErrorBrpcCallback;

/// The runtimefilter is built in the join node.
/// The main purpose is to reduce the scanning amount of the
Expand DownExpand Up@@ -124,6 +129,8 @@ class RuntimeFilter {
// runtime filter type
RuntimeFilterType _runtime_filter_type = RuntimeFilterType::UNKNOWN_FILTER;

std::shared_ptr<HandleErrorBrpcCallback<PMergeFilterResponse>> _merge_filter_callback;

friend class RuntimeFilterProducer;
friend class RuntimeFilterConsumer;
friend class RuntimeFilterMerger;
Expand Down
73 changes: 50 additions & 23 deletions be/src/exec/runtime_filter/runtime_filter_mgr.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -43,6 +43,7 @@
#include "runtime/exec_env.h"
#include "runtime/memory/mem_tracker.h"
#include "runtime/query_context.h"
#include "runtime/runtime_profile.h"
#include "runtime/runtime_state.h"
#include "runtime/thread_context.h"
#include "util/brpc_client_cache.h"
Expand DownExpand Up@@ -76,6 +77,28 @@ std::vector<RuntimeFilterPublishTarget> build_runtime_filter_publish_targets(
return publish_targets;
}

class RuntimeFilterRelayRpcClosure final : public google::protobuf::Closure {
public:
RuntimeFilterRelayRpcClosure(std::shared_ptr<PPublishFilterRequestV2> request,
std::weak_ptr<QueryContext> query_ctx)
: _request(std::move(request)),
_callback(HandleErrorBrpcCallback<PPublishFilterResponse>::create_shared(
std::move(query_ctx))) {}

void Run() override {
std::unique_ptr<RuntimeFilterRelayRpcClosure> self(this);
_callback->call();
}

brpc::Controller* cntl() { return _callback->cntl_.get(); }
PPublishFilterRequestV2* request() { return _request.get(); }
PPublishFilterResponse* response() { return _callback->response_.get(); }

private:
std::shared_ptr<PPublishFilterRequestV2> _request;
std::shared_ptr<HandleErrorBrpcCallback<PPublishFilterResponse>> _callback;
};

Status send_runtime_filter_relay_rpc(const RuntimeFilterPublishTask& task,
const butil::IOBuf& request_attachment, int timeout_ms,
std::weak_ptr<QueryContext> query_ctx) {
Expand All@@ -88,21 +111,17 @@ Status send_runtime_filter_relay_rpc(const RuntimeFilterPublishTask& task,
task.receiver.addr.hostname(), task.receiver.addr.port());
}

auto closure =
AutoReleaseClosure<PPublishFilterRequestV2, DummyBrpcCallback<PPublishFilterResponse>>::
create_unique(std::make_shared<PPublishFilterRequestV2>(task.request),
DummyBrpcCallback<PPublishFilterResponse>::create_shared(),
query_ctx);
// brpc calls Run() exactly once; RuntimeFilterRelayRpcClosure deletes itself there.
auto* closure = new RuntimeFilterRelayRpcClosure(
std::make_shared<PPublishFilterRequestV2>(task.request), std::move(query_ctx));
if (!request_attachment.empty()) {
closure->cntl_->request_attachment().append(request_attachment);
closure->cntl()->request_attachment().append(request_attachment);
}
closure->cntl_->set_timeout_ms(timeout_ms);
closure->cntl()->set_timeout_ms(timeout_ms);
if (config::execution_ignore_eovercrowded) {
closure->cntl_->ignore_eovercrowded();
closure->cntl()->ignore_eovercrowded();
}
stub->apply_filterv2(closure->cntl_.get(), closure->request_.get(), closure->response_.get(),
closure.get());
closure.release();
stub->apply_filterv2(closure->cntl(), closure->request(), closure->response(), closure);
return Status::OK();
}

Expand DownExpand Up@@ -456,9 +475,9 @@ Status RuntimeFilterMergeControllerEntity::send_filter_size(std::shared_ptr<Quer
Status st = Status::OK();
// After all runtime filters' size are collected, we should send response to all producers.
if (cnt_val.merger->add_rf_size(request->filter_size())) {
auto ctx = query_ctx->ignore_runtime_filter_error() ? std::weak_ptr<QueryContext> {}
: query_ctx;
for (auto addr : cnt_val.source_addrs) {
cnt_val.sync_size_callbacks.resize(cnt_val.source_addrs.size());
for (size_t i = 0; i < cnt_val.source_addrs.size(); ++i) {
auto& addr = cnt_val.source_addrs[i];
std::shared_ptr<PBackendService_Stub> stub(
ExecEnv::GetInstance()->brpc_internal_client_cache()->get_client(addr));
if (stub == nullptr) {
Expand All@@ -471,10 +490,14 @@ Status RuntimeFilterMergeControllerEntity::send_filter_size(std::shared_ptr<Quer
auto sync_request = std::make_shared<PSyncFilterSizeRequest>();
sync_request->set_stage(cnt_val.stage);

auto closure = AutoReleaseClosure<PSyncFilterSizeRequest,
DummyBrpcCallback<PSyncFilterSizeResponse>>::
create_unique(sync_request,
DummyBrpcCallback<PSyncFilterSizeResponse>::create_shared(), ctx);
auto callback = HandleErrorBrpcCallback<PSyncFilterSizeResponse>::create_shared(
query_ctx->ignore_runtime_filter_error() ? std::weak_ptr<QueryContext> {}
: query_ctx->weak_from_this());
cnt_val.sync_size_callbacks[i] = callback;
auto closure = AutoReleaseClosure<
PSyncFilterSizeRequest,
HandleErrorBrpcCallback<PSyncFilterSizeResponse>>::create_unique(sync_request,
callback);

auto* pquery_id = closure->request_->mutable_query_id();
pquery_id->set_hi(query_ctx->query_id().hi);
Expand All@@ -487,7 +510,6 @@ Status RuntimeFilterMergeControllerEntity::send_filter_size(std::shared_ptr<Quer

closure->request_->set_filter_id(filter_id);
closure->request_->set_filter_size(cnt_val.merger->get_received_sum_size());

stub->sync_filter_size(closure->cntl_.get(), closure->request_.get(),
closure->response_.get(), closure.get());
closure.release();
Expand DownExpand Up@@ -669,11 +691,14 @@ Status RuntimeFilterMergeControllerEntity::_send_rf_to_target(
}

auto st = Status::OK();
for (auto& target : targets) {
cnt_val.publish_callbacks.resize(targets.size());

@github-actionsgithub-actionsBotSep 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Synchronize callback storage with recursive reset

merge() releases cnt_val.mtx before calling _send_rf_to_target(), so this resize/write is not synchronized with GlobalMergeContext::reset(), which takes that mutex and clears publish_callbacks. Waiting for the old PFC to be destroyed does not drain its fire-and-forget merge_filter RPC; an old merge handler can pass the stage check, drop the mutex, and still be publishing when the next recursive round resets the context. That permits concurrent clear()/resize() on the same vector (or lets an old stage repopulate the new stage's owner set), which is undefined behavior. Please serialize the whole publish setup with reset and revalidate the stage, or make callback ownership per-RPC/self-owned so reset never mutates shared in-flight storage.

for (size_t i = 0; i < targets.size(); ++i) {
auto& target = targets[i];
auto callback = HandleErrorBrpcCallback<PPublishFilterResponse>::create_shared(ctx);
cnt_val.publish_callbacks[i] = callback;
auto closure = AutoReleaseClosure<PPublishFilterRequestV2,
DummyBrpcCallback<PPublishFilterResponse>>::
create_unique(std::make_shared<PPublishFilterRequestV2>(apply_request),
DummyBrpcCallback<PPublishFilterResponse>::create_shared(), ctx);
HandleErrorBrpcCallback<PPublishFilterResponse>>::
create_unique(std::make_shared<PPublishFilterRequestV2>(apply_request), callback);

if (has_attachment) {
closure->cntl_->request_attachment().append(request_attachment);
Expand DownExpand Up@@ -717,6 +742,8 @@ Status GlobalMergeContext::reset(QueryContext* query_ctx) {
merger->increase_expected_producer_num(producer_size);
arrive_id.clear();
source_addrs.clear();
sync_size_callbacks.clear();
publish_callbacks.clear();
done = false;
stage++;
// Keep the Merger's own stage in sync for consistent debug output.
Expand Down
6 changes: 6 additions & 0 deletions be/src/exec/runtime_filter/runtime_filter_mgr.h
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -53,6 +53,9 @@ class RuntimeState;
class RuntimeFilterWrapper;
class QueryContext;
class ExecEnv;
template <typename Response>
class HandleErrorBrpcCallback;
class SyncSizeCallback;

struct RuntimeFilterPublishTarget {
PNetworkAddress addr;
Expand DownExpand Up@@ -94,6 +97,9 @@ struct GlobalMergeContext {
std::vector<TRuntimeFilterTargetParamsV2> targetv2_info;
std::unordered_set<UniqueId> arrive_id;
std::vector<PNetworkAddress> source_addrs;
std::vector<std::shared_ptr<HandleErrorBrpcCallback<PSyncFilterSizeResponse>>>
sync_size_callbacks;
std::vector<std::shared_ptr<HandleErrorBrpcCallback<PPublishFilterResponse>>> publish_callbacks;
std::atomic<bool> done = false;

// for represent the round number of recursive cte
Expand Down
63 changes: 8 additions & 55 deletions be/src/exec/runtime_filter/runtime_filter_producer.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -94,53 +94,6 @@ Status RuntimeFilterProducer::publish(RuntimeState* state, bool build_hash_table
return Status::OK();
}

class SyncSizeClosure : public AutoReleaseClosure<PSendFilterSizeRequest,
DummyBrpcCallback<PSendFilterSizeResponse>> {
std::shared_ptr<Dependency> _dependency;
// Should use weak ptr here, because when query context deconstructs, should also delete runtime filter
// context, it not the memory is not released. And rpc is in another thread, it will hold rf context
// after query context because the rpc is not returned.
std::weak_ptr<RuntimeFilterWrapper> _wrapper;
using Base =
AutoReleaseClosure<PSendFilterSizeRequest, DummyBrpcCallback<PSendFilterSizeResponse>>;
friend class RuntimeFilterProducer;
ENABLE_FACTORY_CREATOR(SyncSizeClosure);

void _process_if_rpc_failed() override {
Defer defer {[&]() {
Base::_process_if_rpc_failed();
((CountedFinishDependency*)_dependency.get())->sub();
}};
auto wrapper = _wrapper.lock();
if (!wrapper) {
return;
}

wrapper->set_state(RuntimeFilterWrapper::State::DISABLED, cntl_->ErrorText());
}

void _process_if_meet_error_status(const Status& status) override {
Defer defer {[&]() {
Base::_process_if_meet_error_status(status);
((CountedFinishDependency*)_dependency.get())->sub();
}};
auto wrapper = _wrapper.lock();
if (!wrapper) {
return;
}

wrapper->set_state(RuntimeFilterWrapper::State::DISABLED, status.to_string());
}

public:
SyncSizeClosure(std::shared_ptr<PSendFilterSizeRequest> req,
std::shared_ptr<DummyBrpcCallback<PSendFilterSizeResponse>> callback,
std::shared_ptr<Dependency> dependency,
std::shared_ptr<RuntimeFilterWrapper> wrapper,
std::weak_ptr<QueryContext> context)
: Base(req, callback, context), _dependency(std::move(dependency)), _wrapper(wrapper) {}
};

void RuntimeFilterProducer::latch_dependency(
const std::shared_ptr<CountedFinishDependency>& dependency) {
std::unique_lock<std::recursive_mutex> l(_rmtx);
Expand DownExpand Up@@ -198,14 +151,13 @@ Status RuntimeFilterProducer::send_size(RuntimeState* state, uint64_t local_filt

auto request = std::make_shared<PSendFilterSizeRequest>();
request->set_stage(_stage);

auto callback = DummyBrpcCallback<PSendFilterSizeResponse>::create_shared();
// when failed, will check `ignore_runtime_filter_error` in callback to decide cancel or not
_sync_size_callback = SyncSizeCallback::create_shared(_dependency, _wrapper,
state->get_query_ctx()->weak_from_this());
// RuntimeFilter maybe deconstructed before the rpc finished, so that could not use
// a raw pointer in closure. Has to use the context's shared ptr.
auto closure = SyncSizeClosure::create_unique(request, callback, _dependency, _wrapper,
state->query_options().ignore_runtime_filter_error
? std::weak_ptr<QueryContext> {}
: state->get_query_ctx_weak());
auto closure = AutoReleaseClosure<PSendFilterSizeRequest, SyncSizeCallback>::create_unique(
request, _sync_size_callback);
auto* pquery_id = request->mutable_query_id();
pquery_id->set_hi(state->get_query_ctx()->query_id().hi);
pquery_id->set_lo(state->get_query_ctx()->query_id().lo);
Expand All@@ -217,9 +169,10 @@ Status RuntimeFilterProducer::send_size(RuntimeState* state, uint64_t local_filt
request->set_filter_size(local_filter_size);
request->set_filter_id(_wrapper->filter_id());

callback->cntl_->set_timeout_ms(get_execution_rpc_timeout_ms(state->execution_timeout()));
_sync_size_callback->cntl_->set_timeout_ms(
get_execution_rpc_timeout_ms(state->execution_timeout()));
if (config::execution_ignore_eovercrowded) {
callback->cntl_->ignore_eovercrowded();
_sync_size_callback->cntl_->ignore_eovercrowded();
}

if (config::enable_debug_points &&
Expand Down
Loading
Loading