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 80ff138cfa1 [bugfix](memory&core) release memory quickly in runtime
filter rpc and do not reuse callback (#67755) (#67912)
80ff138cfa1 is described below
commit 80ff138cfa1532240c119a75a659fd6b2aaa1616
Author: yiguolei <[email protected]>
AuthorDate: Mon Sep 14 09:50:58 2026 +0800
[bugfix](memory&core) release memory quickly in runtime filter rpc and do
not reuse callback (#67755) (#67912)
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
None
- 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 -->
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
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 -->
---
be/src/exec/exchange/vdata_stream_sender.h | 9 +++------
be/src/exec/operator/exchange_sink_buffer.cpp | 19 +++++++++++++++++--
be/src/util/brpc_closure.h | 25 ++++++++++++++++++++++---
3 files changed, 42 insertions(+), 11 deletions(-)
diff --git a/be/src/exec/exchange/vdata_stream_sender.h
b/be/src/exec/exchange/vdata_stream_sender.h
index 9636c752d3d..173b3b5732b 100644
--- a/be/src/exec/exchange/vdata_stream_sender.h
+++ b/be/src/exec/exchange/vdata_stream_sender.h
@@ -164,12 +164,9 @@ public:
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 {
- _send_callback->cntl_->Reset();
- }
+ // Keep the latest callback alive because AutoReleaseClosure
intentionally holds only a
+ // weak_ptr. See ExchangeSinkBuffer::_send_rpc() for the callback
replacement sequence.
+ _send_callback =
ExchangeSendCallback<PTransmitDataResult>::create_shared();
_send_callback->init(ins, eos);
return _send_callback;
}
diff --git a/be/src/exec/operator/exchange_sink_buffer.cpp
b/be/src/exec/operator/exchange_sink_buffer.cpp
index 7f7b8981554..9251b356d7c 100644
--- a/be/src/exec/operator/exchange_sink_buffer.cpp
+++ b/be/src/exec/operator/exchange_sink_buffer.cpp
@@ -224,6 +224,23 @@ Status ExchangeSinkBuffer::add_block(Channel* channel,
BroadcastTransmitInfo&& r
}
Status ExchangeSinkBuffer::_send_rpc(RpcInstance& instance_data) {
+ // A successful callback may synchronously call _send_rpc() to send the
next queued packet.
+ // Therefore RPC-B can be started while RPC-A is still inside its
completion callback:
+ //
+ // RPC-A closure callback-A _send_rpc(RPC-B)
brpc
+ // | | |
|
+ // |-- call() ---------->| |
|
+ // | |-- success handler ---->|
|
+ // | | |-- create
callback-B |
+ // | | |-- send RPC-B
------>|
+ // | |<-----------------------|
|
+ // |<--------------------|
|
+ //
+ // Reusing callback-A for RPC-B would reset its Controller and reuse its
response while RPC-A
+ // is still on this stack. A later read by RPC-A could then observe
RPC-B's state or race with
+ // brpc writing RPC-B's response. Create a separate
callback/Controller/response for every RPC
+ // instead. AutoReleaseClosure::Run() locks callback-A's weak_ptr before
call(), so its local
+ // shared_ptr keeps callback-A alive when the channel member is replaced
by callback-B.
std::unique_lock<std::mutex> lock(*(instance_data.mutex));
auto& q_map = instance_data.package_queue;
@@ -346,7 +363,6 @@ 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!
s = _send_rpc(ins);
if (!s) {
_failed(ins.id,
@@ -475,7 +491,6 @@ 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!
s = _send_rpc(ins);
if (!s) {
_failed(ins.id,
diff --git a/be/src/util/brpc_closure.h b/be/src/util/brpc_closure.h
index c517b066a6d..672ee7c3e9d 100644
--- a/be/src/util/brpc_closure.h
+++ b/be/src/util/brpc_closure.h
@@ -102,8 +102,15 @@ private:
// Example:
// std::unique_ptr<AutoReleaseClosure> a(b);
// brpc_call(a.release());
-// the closure doesn't own the callback, so the callback MUST be kept alive
outside.
-// closure only keep a weak ref. if outside owner destroyed (like query
finish), the callback will be ignored.
+// The closure does not own the callback, so the callback MUST be kept alive
by its external owner
+// while its result is still needed. A callback may indirectly retain large
query-scoped objects,
+// such as QueryContext and operator state. Holding it strongly from an
in-flight RPC would extend
+// all of those objects' lifetimes until the RPC finishes, preventing a
cancelled query from
+// releasing its memory promptly.
+//
+// Keep only a weak reference here. If query cancellation or teardown destroys
the external owner
+// before a late RPC completion arrives, Run() still releases the
request/controller/response owned
+// by this closure, but skips callback business logic because that query no
longer needs the result.
template <typename Request, typename Callback>
class AutoReleaseClosure : public google::protobuf::Closure {
using ResponseType = typename Callback::ResponseType;
@@ -118,9 +125,21 @@ public:
~AutoReleaseClosure() override = default;
- // Will delete itself. all operations should be done in callback's call().
Run() only do one thing.
+ // Releases per-RPC resources, invokes the callback if it is still alive,
and then deletes itself.
void Run() override {
Defer defer {[&]() { delete this; }};
+
+ // The request attachment is no longer needed after brpc finishes the
RPC. It can contain a
+ // large serialized runtime filter, especially a Bloom filter. Since
the callback owner may
+ // retain cntl_ after this closure is deleted (and some callbacks may
also reuse it), keeping
+ // the attachment here would retain that memory until the next
Controller::Reset() or until
+ // the callback is destroyed.
+ //
+ // This must be cleared before call(): a callback may synchronously
start the next RPC and
+ // reuse the same Controller, in which case clearing it after call()
could erase the new
+ // RPC's request attachment.
+ cntl_->request_attachment().clear();
+
if (auto tmp = callback_.lock()) {
tmp->call();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
