Copilot commented on code in PR #61782:
URL: https://github.com/apache/doris/pull/61782#discussion_r3055431164
##########
be/src/exec/runtime_filter/runtime_filter.cpp:
##########
@@ -26,6 +28,34 @@
namespace doris {
#include "common/compile_check_begin.h"
+
+namespace {
+template <typename Response, typename Lock>
+class ManagedTrivialBrpcCallback final : public TrivialBrpcCallback<Response> {
+ ENABLE_FACTORY_CREATOR(ManagedTrivialBrpcCallback);
+
+public:
+ ManagedTrivialBrpcCallback(Lock* lock, std::vector<std::shared_ptr<void>>*
callbacks,
+ std::weak_ptr<QueryContext> context = {})
+ : TrivialBrpcCallback<Response>(std::move(context)),
+ _lock(lock),
+ _callbacks(callbacks) {}
+
+ void call() override {
+ TrivialBrpcCallback<Response>::call();
+ std::lock_guard<Lock> l(*_lock);
+ _callbacks->erase(
+ std::remove_if(_callbacks->begin(), _callbacks->end(),
+ [this](const auto& callback) { return
callback.get() == this; }),
+ _callbacks->end());
+ }
Review Comment:
`ManagedTrivialBrpcCallback` keeps raw pointers to `_rmtx` and
`_async_rpc_callbacks` and erases itself from `_async_rpc_callbacks` inside
`call()`. If the `RuntimeFilter` instance is destroyed concurrently with RPC
completion, the callback may run while the vector/mutex are being destructed,
which is undefined behavior. To make this safe, keep the removal bookkeeping in
a ref-counted shared state (owned by both the filter and callbacks) or
otherwise guarantee the `RuntimeFilter` outlives all in-flight callbacks.
##########
be/src/exec/runtime_filter/runtime_filter_producer.h:
##########
@@ -17,15 +17,71 @@
#pragma once
+#include <glog/logging.h>
+
#include <mutex>
#include "exec/pipeline/dependency.h"
#include "exec/runtime_filter/runtime_filter.h"
#include "runtime/query_context.h"
-#include "runtime/runtime_profile.h"
namespace doris {
#include "common/compile_check_begin.h"
+
+// Callback for sync-size RPCs. Handles errors (disable wrapper + sub
dependency) in call().
+class SyncSizeCallback : public TrivialBrpcCallback<PSendFilterSizeResponse> {
+ ENABLE_FACTORY_CREATOR(SyncSizeCallback);
+
+public:
+ SyncSizeCallback(std::shared_ptr<Dependency> dependency,
+ std::shared_ptr<RuntimeFilterWrapper> wrapper,
+ std::weak_ptr<QueryContext> context)
+ : _dependency(std::move(dependency)), _wrapper(wrapper),
_context(std::move(context)) {}
+
+ void call() override {
+ // On error: disable the wrapper and sub the dependency here, because
set_synced_size()
+ // will never be called (the merge node won't respond with a sync).
+ // On success: do NOT sub here. The merge node will respond with
sync_filter_size,
+ // which calls set_synced_size() -> _dependency->sub().
+ if (cntl_->Failed()) {
+ LOG(WARNING) << fmt::format("RPC meet failed: {}",
cntl_->ErrorText());
+ if (auto w = _wrapper.lock()) {
+ w->set_state(RuntimeFilterWrapper::State::DISABLED,
cntl_->ErrorText());
+ }
+ if (auto ctx = _context.lock()) {
+ if (!ctx->ignore_runtime_filter_error()) {
+ ctx->cancel(Status::NetworkError("RPC meet failed: {}",
cntl_->ErrorText()));
+ }
+ }
+ ((CountedFinishDependency*)_dependency.get())->sub();
+ return;
Review Comment:
`SyncSizeCallback::call()` uses a C-style downcast to
`CountedFinishDependency` (`(CountedFinishDependency*)_dependency.get()`).
Since `_dependency` is typed as `std::shared_ptr<Dependency>` in this callback,
the cast is unchecked and becomes UB if a different `Dependency` implementation
is passed in the future. Prefer storing a
`std::shared_ptr<CountedFinishDependency>` in the callback (or use
`static_pointer_cast` with a DCHECK/dynamic_cast in debug builds) to make the
type guarantee explicit.
##########
be/src/exec/runtime_filter/runtime_filter_producer.h:
##########
@@ -17,15 +17,71 @@
#pragma once
+#include <glog/logging.h>
+
#include <mutex>
#include "exec/pipeline/dependency.h"
#include "exec/runtime_filter/runtime_filter.h"
#include "runtime/query_context.h"
-#include "runtime/runtime_profile.h"
namespace doris {
#include "common/compile_check_begin.h"
+
+// Callback for sync-size RPCs. Handles errors (disable wrapper + sub
dependency) in call().
+class SyncSizeCallback : public TrivialBrpcCallback<PSendFilterSizeResponse> {
+ ENABLE_FACTORY_CREATOR(SyncSizeCallback);
+
+public:
Review Comment:
`SyncSizeCallback` derives from `TrivialBrpcCallback`, but this header does
not include `util/brpc_closure.h` (it only forward-declares
`TrivialBrpcCallback` in `runtime_filter.h`). This makes the header
non-self-contained and can fail to compile depending on include order (e.g.
`runtime_filter_mgr.cpp` includes this header before `util/brpc_closure.h`).
Add the missing include (or move `SyncSizeCallback` definition to a .cpp that
includes `util/brpc_closure.h`).
##########
be/src/exec/runtime_filter/runtime_filter_mgr.cpp:
##########
@@ -40,13 +41,42 @@
#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"
#include "util/brpc_closure.h"
namespace doris {
#include "common/compile_check_begin.h"
+
+namespace {
+template <typename Response, typename Lock>
+class ManagedTrivialBrpcCallback final : public TrivialBrpcCallback<Response> {
+ ENABLE_FACTORY_CREATOR(ManagedTrivialBrpcCallback);
+
+public:
+ ManagedTrivialBrpcCallback(Lock* lock, std::vector<std::shared_ptr<void>>*
callbacks,
+ std::weak_ptr<QueryContext> context = {})
+ : TrivialBrpcCallback<Response>(std::move(context)),
+ _lock(lock),
+ _callbacks(callbacks) {}
+
+ void call() override {
+ TrivialBrpcCallback<Response>::call();
+ std::lock_guard<Lock> l(*_lock);
+ _callbacks->erase(
+ std::remove_if(_callbacks->begin(), _callbacks->end(),
+ [this](const auto& callback) { return
callback.get() == this; }),
+ _callbacks->end());
+ }
Review Comment:
`ManagedTrivialBrpcCallback` stores raw pointers to the owner’s mutex/vector
(`_lock`, `_callbacks`) and uses them from the brpc completion thread. If the
owning object is being destroyed concurrently with RPC completion, these
pointers can dangle (e.g. vector/mutex destruct while `call()` is erasing),
leading to undefined behavior. Consider moving the lock+vector into a separate
`shared_ptr` state owned by both the entity and callbacks (or make the callback
removal conditional on successfully locking a weak_ptr to the owner/state).
##########
be/src/exec/sink/writer/vtablet_writer.cpp:
##########
@@ -94,6 +94,30 @@ class TExpr;
#include "common/compile_check_begin.h"
+namespace {
+template <typename Response>
+class ManagedTrivialBrpcCallback final : public TrivialBrpcCallback<Response> {
+ ENABLE_FACTORY_CREATOR(ManagedTrivialBrpcCallback);
+
+public:
+ ManagedTrivialBrpcCallback(std::mutex* lock,
std::vector<std::shared_ptr<void>>* callbacks)
+ : _lock(lock), _callbacks(callbacks) {}
+
+ void call() override {
+ TrivialBrpcCallback<Response>::call();
+ std::lock_guard<std::mutex> l(*_lock);
+ _callbacks->erase(
+ std::remove_if(_callbacks->begin(), _callbacks->end(),
+ [this](const auto& callback) { return
callback.get() == this; }),
+ _callbacks->end());
+ }
+
+private:
+ std::mutex* _lock;
+ std::vector<std::shared_ptr<void>>* _callbacks;
Review Comment:
This `ManagedTrivialBrpcCallback` stores raw pointers to `VNodeChannel`’s
`_cancel_callbacks_lock` and `_cancel_callbacks` and uses them asynchronously
in `call()`. If the `VNodeChannel` is destroyed while the cancel RPC is still
in flight, `call()` can run with dangling pointers (UB) or race with
destruction of the vector/mutex. Prefer a `shared_ptr`-owned state object for
callback bookkeeping (mutex + vector), or ensure the state outlives all
in-flight cancel callbacks.
```suggestion
struct CancelCallbackState {
std::mutex lock;
std::vector<std::shared_ptr<void>> callbacks;
};
template <typename Response>
class ManagedTrivialBrpcCallback final : public
TrivialBrpcCallback<Response> {
ENABLE_FACTORY_CREATOR(ManagedTrivialBrpcCallback);
public:
explicit ManagedTrivialBrpcCallback(std::shared_ptr<CancelCallbackState>
state)
: _state(std::move(state)) {}
void call() override {
TrivialBrpcCallback<Response>::call();
std::lock_guard<std::mutex> l(_state->lock);
_state->callbacks.erase(
std::remove_if(_state->callbacks.begin(),
_state->callbacks.end(),
[this](const auto& callback) { return
callback.get() == this; }),
_state->callbacks.end());
}
private:
std::shared_ptr<CancelCallbackState> _state;
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
