From df737f591601a30dccae4a4bb0bee84e30a8a43b Mon Sep 17 00:00:00 2001 From: "kaijian.ding" Date: Wed, 8 Jul 2026 00:06:47 +0800 Subject: [PATCH 1/7] [BugFix] Fix bRPC stub cache clean timer leak Signed-off-by: kaijian.ding --- be/src/common/brpc/brpc_stub_cache.cpp | 223 +++++++++++-------- be/src/common/brpc/brpc_stub_cache.h | 119 +++++++++- be/test/common/brpc/brpc_stub_cache_test.cpp | 156 +++++++++++++ 3 files changed, 391 insertions(+), 107 deletions(-) diff --git a/be/src/common/brpc/brpc_stub_cache.cpp b/be/src/common/brpc/brpc_stub_cache.cpp index db146ef3823bec..00a1dd59cbb19e 100644 --- a/be/src/common/brpc/brpc_stub_cache.cpp +++ b/be/src/common/brpc/brpc_stub_cache.cpp @@ -16,6 +16,7 @@ #include "base/failpoint/fail_point.h" #include "base/metrics.h" +#include "base/time/time.h" #include "common/config_network_fwd.h" #include "gen_cpp/internal_service.pb.h" #ifndef __APPLE__ @@ -40,8 +41,65 @@ std::mutex& singleton_cache_mutex() { return mutex; } +inline int64_t absolute_deadline_us(int64_t ttl_seconds) { + return butil::gettimeofday_us() + ttl_seconds * 1000 * 1000; +} } // namespace +template +void wait_clean_tasks_terminate(CacheT* cache, ExtractFn extract) { + std::vector>> tasks; + BthreadTimer* timer = nullptr; + { + std::lock_guard l(cache->_lock); + cache->_stopping = true; + timer = cache->_timer; + cache->_timer = nullptr; + for (auto& stub : cache->_stub_map) { + tasks.push_back(extract(stub.second)); + } + cache->_stub_map.clear(); + } + if (timer != nullptr) { + for (auto& task : tasks) { + task->unschedule_and_join(timer); + } + } +} + +template +void reschedule_if_ttl_shrunk_locked(CacheT* cache, EntryT& entry, const butil::EndPoint& endpoint, GetTaskFn get_task, + SetTaskFn set_task) { + auto& existing_task = get_task(entry); + int64_t new_deadline = absolute_deadline_us(config::brpc_stub_expire_s); + if (new_deadline >= existing_task->deadline_locked()) { + existing_task->renew_deadline_locked(new_deadline); + return; + } + // TTL was shrunk. Build a fresh task; reusing the existing one would re-arm the same + // BthreadTimerTask and trip its one-shot std::latch on the next fire. The old task is left + // in the timer queue and will fire at its original deadline; when it does, the map entry + // already points at the new task, so its erase drops the new entry harmlessly; the next call recreates it. + auto new_task = std::make_shared>(cache, endpoint, config::brpc_stub_expire_s); + new_task->renew_deadline_locked(new_deadline); + timespec tm = butil::microseconds_to_timespec(new_deadline); + auto status = cache->_timer->schedule(new_task.get(), tm); + if (status.ok()) { + set_task(entry, std::move(new_task)); + } else { + LOG(WARNING) << "Failed to reschedule brpc cleanup task: " << endpoint; + } +} + +template +void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer) { + DCHECK(timer != nullptr); + std::lock_guard l(cache->_lock); + DCHECK(cache->_stub_map.empty() || cache->_timer == timer); + cache->_stopping = false; + cache->_timer = timer; +} + struct BrpcStubCache::Metrics { Metrics(MetricRegistry* metric_registry, BrpcStubCache* cache) : registry(metric_registry), cache(cache) { DCHECK(registry != nullptr); @@ -71,21 +129,15 @@ BrpcStubCache::BrpcStubCache(BthreadTimer* timer, MetricRegistry* metric_registr BrpcStubCache::~BrpcStubCache() { _metrics.reset(); + wait_clean_tasks_terminate(this, [](const std::shared_ptr& pool) { return pool->_cleanup_task; }); +} - std::vector> pools_to_cleanup; - { - std::lock_guard l(_lock); - - for (auto& stub : _stub_map) { - pools_to_cleanup.push_back(stub.second); - } - } - - for (auto& pool : pools_to_cleanup) { - (void)_timer->unschedule(pool->_cleanup_task.get()); +void BrpcStubCache::replace_cleanup_task_locked(const butil::EndPoint& endpoint, + std::shared_ptr> task) { + auto pool = _stub_map.seek(endpoint); + if (pool != nullptr) { + (*pool)->_cleanup_task = std::move(task); } - std::lock_guard l(_lock); - _stub_map.clear(); } std::shared_ptr BrpcStubCache::get_stub(const butil::EndPoint& endpoint) { @@ -94,17 +146,26 @@ std::shared_ptr BrpcStubCache::get_stub(const auto stub_pool = _stub_map.seek(endpoint); if (stub_pool == nullptr) { auto new_pool = std::make_shared(); - new_pool->_cleanup_task = std::make_shared>(this, endpoint); + new_pool->_cleanup_task = + std::make_shared>(this, endpoint, config::brpc_stub_expire_s); + new_pool->_cleanup_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s)); _stub_map.insert(endpoint, new_pool); stub_pool = _stub_map.seek(endpoint); - } - if (_timer->unschedule((*stub_pool)->_cleanup_task.get()) != TIMER_TASK_RUNNING) { - timespec tm = butil::seconds_from_now(config::brpc_stub_expire_s); + timespec tm = butil::microseconds_to_timespec((*stub_pool)->_cleanup_task->deadline_locked()); auto status = _timer->schedule((*stub_pool)->_cleanup_task.get(), tm); if (!status.ok()) { LOG(WARNING) << "Failed to schedule brpc cleanup task: " << endpoint; } + } else { + reschedule_if_ttl_shrunk_locked( + this, *stub_pool, endpoint, + [](std::shared_ptr& pool) -> std::shared_ptr>& { + return pool->_cleanup_task; + }, + [](std::shared_ptr& pool, std::shared_ptr> task) { + pool->_cleanup_task = std::move(task); + }); } return (*stub_pool)->get_or_create(endpoint); @@ -134,13 +195,6 @@ std::shared_ptr BrpcStubCache::get_stub(const return get_stub(endpoint); } -void BrpcStubCache::cleanup_expired(const butil::EndPoint& endpoint) { - std::lock_guard l(_lock); - - LOG(INFO) << "cleanup brpc stub, endpoint:" << endpoint; - _stub_map.erase(endpoint); -} - BrpcStubCache::StubPool::StubPool() { _stubs.reserve(config::brpc_max_connections_per_server); } @@ -190,30 +244,18 @@ HttpBrpcStubCache::~HttpBrpcStubCache() { } void HttpBrpcStubCache::bind_timer(BthreadTimer* timer) { - DCHECK(timer != nullptr); - std::lock_guard l(_lock); - DCHECK(_stub_map.empty() || _timer == timer); - _timer = timer; + reset_state_for_rebind(this, timer); } void HttpBrpcStubCache::shutdown() { - std::vector>> task_to_cleanup; - BthreadTimer* timer = nullptr; + wait_clean_tasks_terminate(this, [](const StubEntry& entry) { return entry.cleanup_task; }); +} - { - std::lock_guard l(_lock); - timer = _timer; - _timer = nullptr; - for (auto& stub : _stub_map) { - task_to_cleanup.push_back(stub.second.second); - } - _stub_map.clear(); - } - - if (timer != nullptr) { - for (auto& task : task_to_cleanup) { - timer->unschedule(task.get()); - } +void HttpBrpcStubCache::replace_cleanup_task_locked(const butil::EndPoint& endpoint, + std::shared_ptr> task) { + auto entry = _stub_map.seek(endpoint); + if (entry != nullptr) { + entry->cleanup_task = std::move(task); } } @@ -243,33 +285,34 @@ StatusOr> HttpBrpcStubCache::g auto stub_pair_ptr = _stub_map.seek(endpoint); if (stub_pair_ptr == nullptr) { // create - auto new_task = std::make_shared>(this, endpoint); + auto new_task = + std::make_shared>(this, endpoint, config::brpc_stub_expire_s); auto stub = std::make_shared(endpoint, "http"); if (!stub->reset_channel().ok()) { return Status::RuntimeError("init http brpc channel error on " + taddr.hostname + ":" + std::to_string(taddr.port)); } - _stub_map.insert(endpoint, std::make_pair(stub, new_task)); + new_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s)); + _stub_map.insert(endpoint, StubEntry{stub, new_task}); stub_pair_ptr = _stub_map.seek(endpoint); - } - // schedule clean up task - if (_timer->unschedule((*stub_pair_ptr).second.get()) != TIMER_TASK_RUNNING) { - timespec tm = butil::seconds_from_now(config::brpc_stub_expire_s); - auto status = _timer->schedule((*stub_pair_ptr).second.get(), tm); + timespec tm = butil::microseconds_to_timespec(stub_pair_ptr->cleanup_task->deadline_locked()); + auto status = _timer->schedule(stub_pair_ptr->cleanup_task.get(), tm); if (!status.ok()) { - LOG(WARNING) << "Failed to schedule http brpc cleanup task: " << endpoint; + LOG(WARNING) << "Failed to schedule brpc cleanup task: " << endpoint; } + } else { + reschedule_if_ttl_shrunk_locked( + this, *stub_pair_ptr, endpoint, + [](StubEntry& entry) -> std::shared_ptr>& { + return entry.cleanup_task; + }, + [](StubEntry& entry, std::shared_ptr> task) { + entry.cleanup_task = std::move(task); + }); } - return (*stub_pair_ptr).first; -} - -void HttpBrpcStubCache::cleanup_expired(const butil::EndPoint& endpoint) { - std::lock_guard l(_lock); - - LOG(INFO) << "cleanup http brpc stub, endpoint:" << endpoint; - _stub_map.erase(endpoint); + return stub_pair_ptr->stub; } #ifndef __APPLE__ @@ -298,30 +341,18 @@ LakeServiceBrpcStubCache::~LakeServiceBrpcStubCache() { } void LakeServiceBrpcStubCache::bind_timer(BthreadTimer* timer) { - DCHECK(timer != nullptr); - std::lock_guard l(_lock); - DCHECK(_stub_map.empty() || _timer == timer); - _timer = timer; + reset_state_for_rebind(this, timer); } void LakeServiceBrpcStubCache::shutdown() { - std::vector>> task_to_cleanup; - BthreadTimer* timer = nullptr; + wait_clean_tasks_terminate(this, [](const StubEntry& entry) { return entry.cleanup_task; }); +} - { - std::lock_guard l(_lock); - timer = _timer; - _timer = nullptr; - for (auto& stub : _stub_map) { - task_to_cleanup.push_back(stub.second.second); - } - _stub_map.clear(); - } - - if (timer != nullptr) { - for (auto& task : task_to_cleanup) { - timer->unschedule(task.get()); - } +void LakeServiceBrpcStubCache::replace_cleanup_task_locked( + const butil::EndPoint& endpoint, std::shared_ptr> task) { + auto entry = _stub_map.seek(endpoint); + if (entry != nullptr) { + entry->cleanup_task = std::move(task); } } @@ -350,32 +381,34 @@ StatusOr> LakeServiceBrp if (stub_pair_ptr == nullptr) { // create auto stub = std::make_shared(endpoint, ""); - auto new_task = std::make_shared>(this, endpoint); + auto new_task = std::make_shared>(this, endpoint, + config::brpc_stub_expire_s); if (!stub->reset_channel().ok()) { return Status::RuntimeError("init lakeService brpc channel error on " + host + ":" + std::to_string(port)); } - _stub_map.insert(endpoint, std::make_pair(stub, new_task)); + new_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s)); + _stub_map.insert(endpoint, StubEntry{stub, new_task}); stub_pair_ptr = _stub_map.seek(endpoint); - } - // schedule clean up task - if (_timer->unschedule((*stub_pair_ptr).second.get()) != TIMER_TASK_RUNNING) { - timespec tm = butil::seconds_from_now(config::brpc_stub_expire_s); - auto status = _timer->schedule((*stub_pair_ptr).second.get(), tm); + timespec tm = butil::microseconds_to_timespec(stub_pair_ptr->cleanup_task->deadline_locked()); + auto status = _timer->schedule(stub_pair_ptr->cleanup_task.get(), tm); if (!status.ok()) { - LOG(WARNING) << "Failed to schedule lake brpc cleanup task: " << endpoint; + LOG(WARNING) << "Failed to schedule brpc cleanup task: " << endpoint; } + } else { + reschedule_if_ttl_shrunk_locked( + this, *stub_pair_ptr, endpoint, + [](StubEntry& entry) -> std::shared_ptr>& { + return entry.cleanup_task; + }, + [](StubEntry& entry, std::shared_ptr> task) { + entry.cleanup_task = std::move(task); + }); } - return (*stub_pair_ptr).first; + return stub_pair_ptr->stub; } -void LakeServiceBrpcStubCache::cleanup_expired(const butil::EndPoint& endpoint) { - std::lock_guard l(_lock); - - LOG(INFO) << "cleanup lake service brpc stub, endpoint:" << endpoint; - _stub_map.erase(endpoint); -} #endif } // namespace starrocks diff --git a/be/src/common/brpc/brpc_stub_cache.h b/be/src/common/brpc/brpc_stub_cache.h index 2ba92f57289e5f..7fc428b542d29e 100644 --- a/be/src/common/brpc/brpc_stub_cache.h +++ b/be/src/common/brpc/brpc_stub_cache.h @@ -41,8 +41,10 @@ #include "base/brpc/brpc.h" #include "base/concurrency/spinlock.h" #include "base/network/network_util.h" +#include "base/time/time.h" #include "common/brpc/internal_service_recoverable_stub.h" #include "common/bthread_timer.h" +#include "common/logging.h" #include "common/statusor.h" #include "gen_cpp/Types_types.h" // TNetworkAddress @@ -53,18 +55,56 @@ namespace starrocks { class MetricRegistry; +class BrpcStubCacheTest; constexpr int TIMER_TASK_RUNNING = 1; template -class EndpointCleanupTask : public BthreadLightTimerTask { +class EndpointCleanupTask : public BthreadTimerTask { public: - EndpointCleanupTask(StubCacheT* cache, const butil::EndPoint& endpoint) : _cache(cache), _endpoint(endpoint){}; - void Run() override { _cache->cleanup_expired(_endpoint); } + // ttl_seconds is the cache-wide expire window (config::brpc_stub_expire_s). + EndpointCleanupTask(StubCacheT* cache, const butil::EndPoint& endpoint, int64_t ttl_seconds) + : _cache(cache), _endpoint(endpoint), _ttl_seconds(ttl_seconds) {} + // The actual cleanup/renewal decision must run while the cache's _lock is held so + // that _stopping and _deadline are observed atomically with the cache state. + void Run() override { + std::lock_guard l(_cache->_lock); + if (_cache->_stopping) { + return; + } + int64_t now_us = butil::gettimeofday_us(); + if (now_us >= _deadline) { + LOG(INFO) << "cleanup brpc stub, endpoint:" << _endpoint << ", idle for " << (now_us - _deadline) / 1000 + << "ms past deadline"; + _cache->_stub_map.erase(_endpoint); + return; + } + auto new_task = std::make_shared>(_cache, _endpoint, _ttl_seconds); + new_task->_deadline = _deadline; + _cache->replace_cleanup_task_locked(_endpoint, new_task); + timespec tm = butil::microseconds_to_timespec(_deadline); + auto status = _cache->_timer->schedule(new_task.get(), tm); + if (!status.ok()) { + LOG(WARNING) << "Failed to reschedule brpc cleanup task: " << _endpoint; + // Drop the entry; the next get_*_stub() will recreate it with a fresh task. + _cache->_stub_map.erase(_endpoint); + } + } + + // Reset the absolute deadline (in butil::gettimeofday_us() units) used by the next + // Run() invocation to decide between evict and reschedule. Caller must hold the + // cache lock. + void renew_deadline_locked(int64_t new_deadline) { _deadline = new_deadline; } + int64_t deadline_locked() const { return _deadline; } private: StubCacheT* _cache; butil::EndPoint _endpoint; + // Absolute deadline (in butil::gettimeofday_us() units) used to decide whether a + // firing task should evict the stub or simply reschedule itself. Read/written only + // under the cache's _lock, so it does not need to be atomic. + int64_t _deadline{0}; + int64_t _ttl_seconds{0}; }; class BrpcStubCache { @@ -75,9 +115,25 @@ class BrpcStubCache { std::shared_ptr get_stub(const butil::EndPoint& endpoint); std::shared_ptr get_stub(const TNetworkAddress& taddr); std::shared_ptr get_stub(const std::string& host, int port); - void cleanup_expired(const butil::EndPoint& endpoint); private: + friend class EndpointCleanupTask; + + template + friend void wait_clean_tasks_terminate(CacheT* cache, ExtractFn extract); + + template + friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); + + template + friend void reschedule_if_ttl_shrunk_locked(CacheT* cache, EntryT& entry, const butil::EndPoint& endpoint, + GetTaskFn get_task, SetTaskFn set_task); + + friend class starrocks::BrpcStubCacheTest; + + void replace_cleanup_task_locked(const butil::EndPoint& endpoint, + std::shared_ptr> task); + struct Metrics; struct StubPool { StubPool(); @@ -93,6 +149,7 @@ class BrpcStubCache { butil::FlatMap> _stub_map; BthreadTimer* _timer; std::unique_ptr _metrics; + bool _stopping{false}; }; class HttpBrpcStubCache { @@ -103,19 +160,38 @@ class HttpBrpcStubCache { static void initialize(BthreadTimer* timer); static HttpBrpcStubCache* getInstance(); StatusOr> get_http_stub(const TNetworkAddress& taddr); - void cleanup_expired(const butil::EndPoint& endpoint); void shutdown(); private: explicit HttpBrpcStubCache(BthreadTimer* timer); ~HttpBrpcStubCache(); void bind_timer(BthreadTimer* timer); + friend class EndpointCleanupTask; + + template + friend void wait_clean_tasks_terminate(CacheT* cache, ExtractFn extract); + + template + friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); + + template + friend void reschedule_if_ttl_shrunk_locked(CacheT* cache, EntryT& entry, const butil::EndPoint& endpoint, + GetTaskFn get_task, SetTaskFn set_task); + + friend class starrocks::BrpcStubCacheTest; + + void replace_cleanup_task_locked(const butil::EndPoint& endpoint, + std::shared_ptr> task); + + struct StubEntry { + std::shared_ptr stub; + std::shared_ptr> cleanup_task; + }; SpinLock _lock; - butil::FlatMap, - std::shared_ptr>>> - _stub_map; + butil::FlatMap _stub_map; BthreadTimer* _timer; + bool _stopping{false}; }; #ifndef __APPLE__ @@ -127,19 +203,38 @@ class LakeServiceBrpcStubCache { static void initialize(BthreadTimer* timer); static LakeServiceBrpcStubCache* getInstance(); StatusOr> get_stub(const std::string& host, int port); - void cleanup_expired(const butil::EndPoint& endpoint); void shutdown(); private: explicit LakeServiceBrpcStubCache(BthreadTimer* timer); ~LakeServiceBrpcStubCache(); void bind_timer(BthreadTimer* timer); + friend class EndpointCleanupTask; + + template + friend void wait_clean_tasks_terminate(CacheT* cache, ExtractFn extract); + + template + friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); + + template + friend void reschedule_if_ttl_shrunk_locked(CacheT* cache, EntryT& entry, const butil::EndPoint& endpoint, + GetTaskFn get_task, SetTaskFn set_task); + + friend class starrocks::BrpcStubCacheTest; + + void replace_cleanup_task_locked(const butil::EndPoint& endpoint, + std::shared_ptr> task); + + struct StubEntry { + std::shared_ptr stub; + std::shared_ptr> cleanup_task; + }; SpinLock _lock; - butil::FlatMap, - std::shared_ptr>>> - _stub_map; + butil::FlatMap _stub_map; BthreadTimer* _timer; + bool _stopping{false}; }; #endif diff --git a/be/test/common/brpc/brpc_stub_cache_test.cpp b/be/test/common/brpc/brpc_stub_cache_test.cpp index 4ab56fe524dd65..2b548ea3ad22ba 100644 --- a/be/test/common/brpc/brpc_stub_cache_test.cpp +++ b/be/test/common/brpc/brpc_stub_cache_test.cpp @@ -181,6 +181,162 @@ TEST_F(BrpcStubCacheTest, test_http_cleanup) { ASSERT_NE(*stub3, *stub1); } +// Regression test: destroying BrpcStubCache while a cleanup task is scheduled +// (and possibly firing) must join the task before the cache state is torn down. +TEST_F(BrpcStubCacheTest, test_destructor_joins_inflight_cleanup_tasks) { + config::brpc_stub_expire_s = 1; + auto cache = std::make_unique(_timer.get()); + TNetworkAddress address; + address.hostname = "127.0.0.1"; + address.port = 123; + auto stub = cache->get_stub(address); + ASSERT_NE(nullptr, stub); + + // Trigger ~BrpcStubCache() while the cleanup task is (or will be) in flight. + // Drain the cache and assert the unique_ptr release returns cleanly. + cache.reset(); + + // Reacquire the endpoint through a fresh cache; the slot must have been + // cleanly torn down without leaking the previous task. + auto cache2 = std::make_unique(_timer.get()); + auto fresh_stub = cache2->get_stub(address); + ASSERT_NE(nullptr, fresh_stub); + cache2.reset(); +} + +// Regression test for the lazy-reschedule fix: while an endpoint is being +// accessed within the expire window, the timer fires, sees the stub is still +// active (idle < brpc_stub_expire_s), and reschedules instead of evicting, so the +// stub must survive across multiple timer periods. +TEST_F(BrpcStubCacheTest, test_active_access_keeps_stub_alive_across_expire_window) { + config::brpc_stub_expire_s = 2; + BrpcStubCache cache(_timer.get()); + TNetworkAddress address; + address.hostname = "127.0.0.1"; + address.port = 123; + auto stub1 = cache.get_stub(address); + ASSERT_NE(nullptr, stub1); + + // Repeatedly access within the window; each access refreshes the last-access + // time, and the single scheduled timer reschedules instead of evicting. + for (int i = 0; i < 3; ++i) { + sleep(1); + auto stub = cache.get_stub(address); + ASSERT_EQ(stub1, stub) << "stub must not be evicted while being accessed"; + } +} + +TEST_F(BrpcStubCacheTest, test_http_active_access_keeps_stub_alive_across_expire_window) { + config::brpc_stub_expire_s = 2; + HttpBrpcStubCache cache(_timer.get()); + TNetworkAddress address; + address.hostname = "127.0.0.1"; + address.port = 123; + auto stub1 = cache.get_http_stub(address); + ASSERT_NE(nullptr, *stub1); + + for (int i = 0; i < 3; ++i) { + sleep(1); + auto stub = cache.get_http_stub(address); + ASSERT_NE(nullptr, *stub); + ASSERT_EQ(*stub1, *stub) << "http stub must not be evicted while being accessed"; + } +} + +#ifndef __APPLE__ +TEST_F(BrpcStubCacheTest, test_lake_active_access_keeps_stub_alive_across_expire_window) { + config::brpc_stub_expire_s = 2; + LakeServiceBrpcStubCache cache(_timer.get()); + std::string hostname = "127.0.0.1"; + int32_t port = 123; + auto stub1 = cache.get_stub(hostname, port); + ASSERT_TRUE(stub1.ok()); + ASSERT_NE(nullptr, *stub1); + + for (int i = 0; i < 3; ++i) { + sleep(1); + auto stub = cache.get_stub(hostname, port); + ASSERT_TRUE(stub.ok()); + ASSERT_NE(nullptr, *stub); + ASSERT_EQ(*stub1, *stub) << "lake stub must not be evicted while being accessed"; + } +} +#endif + +// Reducing brpc_stub_expire_s between get_*_stub() calls must replace the cleanup task: a fresh +// task is scheduled with the new earlier deadline, while the previous task is left in the timer +// queue and will fire at its original (later) deadline. That late fire ends up calling +// _cache->_stub_map.erase() on the new entry -- safe, because the next get_*_stub() recreates it. +// The check below: the task pointer stored in the map must change after a TTL shrink. +TEST_F(BrpcStubCacheTest, test_ttl_shrink_replaces_cleanup_task) { + config::brpc_stub_expire_s = 3600; + BrpcStubCache cache(_timer.get()); + butil::EndPoint endpoint; + ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:123", &endpoint)); + ASSERT_NE(nullptr, cache.get_stub(endpoint)); + + auto* pool_before = cache._stub_map.seek(endpoint); + ASSERT_NE(nullptr, pool_before); + auto* task_before = (*pool_before)->_cleanup_task.get(); + + config::brpc_stub_expire_s = 1; + ASSERT_NE(nullptr, cache.get_stub(endpoint)); + + auto* pool_after = cache._stub_map.seek(endpoint); + ASSERT_NE(nullptr, pool_after); + auto* task_after = (*pool_after)->_cleanup_task.get(); + ASSERT_NE(task_before, task_after) << "shrinking TTL must install a new cleanup task"; +} + +TEST_F(BrpcStubCacheTest, test_http_ttl_shrink_replaces_cleanup_task) { + config::brpc_stub_expire_s = 3600; + HttpBrpcStubCache cache(_timer.get()); + TNetworkAddress address; + address.hostname = "127.0.0.1"; + address.port = 123; + auto stub = cache.get_http_stub(address); + ASSERT_TRUE(stub.ok()); + + butil::EndPoint endpoint; + ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:123", &endpoint)); + auto* entry_before = cache._stub_map.seek(endpoint); + ASSERT_NE(nullptr, entry_before); + auto* task_before = entry_before->cleanup_task.get(); + + config::brpc_stub_expire_s = 1; + auto stub2 = cache.get_http_stub(address); + ASSERT_TRUE(stub2.ok()); + + auto* entry_after = cache._stub_map.seek(endpoint); + ASSERT_NE(nullptr, entry_after); + auto* task_after = entry_after->cleanup_task.get(); + ASSERT_NE(task_before, task_after) << "shrinking TTL must install a new cleanup task"; +} + +#ifndef __APPLE__ +TEST_F(BrpcStubCacheTest, test_lake_ttl_shrink_replaces_cleanup_task) { + config::brpc_stub_expire_s = 3600; + LakeServiceBrpcStubCache cache(_timer.get()); + auto stub = cache.get_stub("127.0.0.1", 123); + ASSERT_TRUE(stub.ok()); + + butil::EndPoint endpoint; + ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:123", &endpoint)); + auto* entry_before = cache._stub_map.seek(endpoint); + ASSERT_NE(nullptr, entry_before); + auto* task_before = entry_before->cleanup_task.get(); + + config::brpc_stub_expire_s = 1; + auto stub2 = cache.get_stub("127.0.0.1", 123); + ASSERT_TRUE(stub2.ok()); + + auto* entry_after = cache._stub_map.seek(endpoint); + ASSERT_NE(nullptr, entry_after); + auto* task_after = entry_after->cleanup_task.get(); + ASSERT_NE(task_before, task_after) << "shrinking TTL must install a new cleanup task"; +} +#endif + TEST_F(BrpcStubCacheTest, http_singleton_reinitialize_rebinds_pipeline_timer) { auto timer2 = std::make_unique(); ASSERT_OK(timer2->start()); From 00f8a6ae9c4eb3d7594bfbcd77fc51ffc44702e8 Mon Sep 17 00:00:00 2001 From: "kaijian.ding" Date: Thu, 9 Jul 2026 21:57:04 +0800 Subject: [PATCH 2/7] Avoid rescheduling BRPC cleanup tasks Signed-off-by: kaijian.ding --- be/src/common/brpc/brpc_stub_cache.cpp | 51 ++------------------------ be/src/common/brpc/brpc_stub_cache.h | 12 ------ 2 files changed, 3 insertions(+), 60 deletions(-) diff --git a/be/src/common/brpc/brpc_stub_cache.cpp b/be/src/common/brpc/brpc_stub_cache.cpp index 00a1dd59cbb19e..2c7affdbc63630 100644 --- a/be/src/common/brpc/brpc_stub_cache.cpp +++ b/be/src/common/brpc/brpc_stub_cache.cpp @@ -67,30 +67,6 @@ void wait_clean_tasks_terminate(CacheT* cache, ExtractFn extract) { } } -template -void reschedule_if_ttl_shrunk_locked(CacheT* cache, EntryT& entry, const butil::EndPoint& endpoint, GetTaskFn get_task, - SetTaskFn set_task) { - auto& existing_task = get_task(entry); - int64_t new_deadline = absolute_deadline_us(config::brpc_stub_expire_s); - if (new_deadline >= existing_task->deadline_locked()) { - existing_task->renew_deadline_locked(new_deadline); - return; - } - // TTL was shrunk. Build a fresh task; reusing the existing one would re-arm the same - // BthreadTimerTask and trip its one-shot std::latch on the next fire. The old task is left - // in the timer queue and will fire at its original deadline; when it does, the map entry - // already points at the new task, so its erase drops the new entry harmlessly; the next call recreates it. - auto new_task = std::make_shared>(cache, endpoint, config::brpc_stub_expire_s); - new_task->renew_deadline_locked(new_deadline); - timespec tm = butil::microseconds_to_timespec(new_deadline); - auto status = cache->_timer->schedule(new_task.get(), tm); - if (status.ok()) { - set_task(entry, std::move(new_task)); - } else { - LOG(WARNING) << "Failed to reschedule brpc cleanup task: " << endpoint; - } -} - template void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer) { DCHECK(timer != nullptr); @@ -158,14 +134,7 @@ std::shared_ptr BrpcStubCache::get_stub(const LOG(WARNING) << "Failed to schedule brpc cleanup task: " << endpoint; } } else { - reschedule_if_ttl_shrunk_locked( - this, *stub_pool, endpoint, - [](std::shared_ptr& pool) -> std::shared_ptr>& { - return pool->_cleanup_task; - }, - [](std::shared_ptr& pool, std::shared_ptr> task) { - pool->_cleanup_task = std::move(task); - }); + (*stub_pool)->_cleanup_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s)); } return (*stub_pool)->get_or_create(endpoint); @@ -302,14 +271,7 @@ StatusOr> HttpBrpcStubCache::g LOG(WARNING) << "Failed to schedule brpc cleanup task: " << endpoint; } } else { - reschedule_if_ttl_shrunk_locked( - this, *stub_pair_ptr, endpoint, - [](StubEntry& entry) -> std::shared_ptr>& { - return entry.cleanup_task; - }, - [](StubEntry& entry, std::shared_ptr> task) { - entry.cleanup_task = std::move(task); - }); + stub_pair_ptr->cleanup_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s)); } return stub_pair_ptr->stub; @@ -396,14 +358,7 @@ StatusOr> LakeServiceBrp LOG(WARNING) << "Failed to schedule brpc cleanup task: " << endpoint; } } else { - reschedule_if_ttl_shrunk_locked( - this, *stub_pair_ptr, endpoint, - [](StubEntry& entry) -> std::shared_ptr>& { - return entry.cleanup_task; - }, - [](StubEntry& entry, std::shared_ptr> task) { - entry.cleanup_task = std::move(task); - }); + stub_pair_ptr->cleanup_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s)); } return stub_pair_ptr->stub; diff --git a/be/src/common/brpc/brpc_stub_cache.h b/be/src/common/brpc/brpc_stub_cache.h index 7fc428b542d29e..3830d248a9268b 100644 --- a/be/src/common/brpc/brpc_stub_cache.h +++ b/be/src/common/brpc/brpc_stub_cache.h @@ -125,10 +125,6 @@ class BrpcStubCache { template friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); - template - friend void reschedule_if_ttl_shrunk_locked(CacheT* cache, EntryT& entry, const butil::EndPoint& endpoint, - GetTaskFn get_task, SetTaskFn set_task); - friend class starrocks::BrpcStubCacheTest; void replace_cleanup_task_locked(const butil::EndPoint& endpoint, @@ -174,10 +170,6 @@ class HttpBrpcStubCache { template friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); - template - friend void reschedule_if_ttl_shrunk_locked(CacheT* cache, EntryT& entry, const butil::EndPoint& endpoint, - GetTaskFn get_task, SetTaskFn set_task); - friend class starrocks::BrpcStubCacheTest; void replace_cleanup_task_locked(const butil::EndPoint& endpoint, @@ -217,10 +209,6 @@ class LakeServiceBrpcStubCache { template friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); - template - friend void reschedule_if_ttl_shrunk_locked(CacheT* cache, EntryT& entry, const butil::EndPoint& endpoint, - GetTaskFn get_task, SetTaskFn set_task); - friend class starrocks::BrpcStubCacheTest; void replace_cleanup_task_locked(const butil::EndPoint& endpoint, From 77bc3ae9de60aa4f0542641a6ea3857883343b57 Mon Sep 17 00:00:00 2001 From: "kaijian.ding" Date: Fri, 10 Jul 2026 11:28:34 +0800 Subject: [PATCH 3/7] Remove obsolete BRPC TTL shrink tests Signed-off-by: kaijian.ding --- be/test/common/brpc/brpc_stub_cache_test.cpp | 74 -------------------- 1 file changed, 74 deletions(-) diff --git a/be/test/common/brpc/brpc_stub_cache_test.cpp b/be/test/common/brpc/brpc_stub_cache_test.cpp index 2b548ea3ad22ba..8fd95e98dbaa92 100644 --- a/be/test/common/brpc/brpc_stub_cache_test.cpp +++ b/be/test/common/brpc/brpc_stub_cache_test.cpp @@ -263,80 +263,6 @@ TEST_F(BrpcStubCacheTest, test_lake_active_access_keeps_stub_alive_across_expire } #endif -// Reducing brpc_stub_expire_s between get_*_stub() calls must replace the cleanup task: a fresh -// task is scheduled with the new earlier deadline, while the previous task is left in the timer -// queue and will fire at its original (later) deadline. That late fire ends up calling -// _cache->_stub_map.erase() on the new entry -- safe, because the next get_*_stub() recreates it. -// The check below: the task pointer stored in the map must change after a TTL shrink. -TEST_F(BrpcStubCacheTest, test_ttl_shrink_replaces_cleanup_task) { - config::brpc_stub_expire_s = 3600; - BrpcStubCache cache(_timer.get()); - butil::EndPoint endpoint; - ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:123", &endpoint)); - ASSERT_NE(nullptr, cache.get_stub(endpoint)); - - auto* pool_before = cache._stub_map.seek(endpoint); - ASSERT_NE(nullptr, pool_before); - auto* task_before = (*pool_before)->_cleanup_task.get(); - - config::brpc_stub_expire_s = 1; - ASSERT_NE(nullptr, cache.get_stub(endpoint)); - - auto* pool_after = cache._stub_map.seek(endpoint); - ASSERT_NE(nullptr, pool_after); - auto* task_after = (*pool_after)->_cleanup_task.get(); - ASSERT_NE(task_before, task_after) << "shrinking TTL must install a new cleanup task"; -} - -TEST_F(BrpcStubCacheTest, test_http_ttl_shrink_replaces_cleanup_task) { - config::brpc_stub_expire_s = 3600; - HttpBrpcStubCache cache(_timer.get()); - TNetworkAddress address; - address.hostname = "127.0.0.1"; - address.port = 123; - auto stub = cache.get_http_stub(address); - ASSERT_TRUE(stub.ok()); - - butil::EndPoint endpoint; - ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:123", &endpoint)); - auto* entry_before = cache._stub_map.seek(endpoint); - ASSERT_NE(nullptr, entry_before); - auto* task_before = entry_before->cleanup_task.get(); - - config::brpc_stub_expire_s = 1; - auto stub2 = cache.get_http_stub(address); - ASSERT_TRUE(stub2.ok()); - - auto* entry_after = cache._stub_map.seek(endpoint); - ASSERT_NE(nullptr, entry_after); - auto* task_after = entry_after->cleanup_task.get(); - ASSERT_NE(task_before, task_after) << "shrinking TTL must install a new cleanup task"; -} - -#ifndef __APPLE__ -TEST_F(BrpcStubCacheTest, test_lake_ttl_shrink_replaces_cleanup_task) { - config::brpc_stub_expire_s = 3600; - LakeServiceBrpcStubCache cache(_timer.get()); - auto stub = cache.get_stub("127.0.0.1", 123); - ASSERT_TRUE(stub.ok()); - - butil::EndPoint endpoint; - ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:123", &endpoint)); - auto* entry_before = cache._stub_map.seek(endpoint); - ASSERT_NE(nullptr, entry_before); - auto* task_before = entry_before->cleanup_task.get(); - - config::brpc_stub_expire_s = 1; - auto stub2 = cache.get_stub("127.0.0.1", 123); - ASSERT_TRUE(stub2.ok()); - - auto* entry_after = cache._stub_map.seek(endpoint); - ASSERT_NE(nullptr, entry_after); - auto* task_after = entry_after->cleanup_task.get(); - ASSERT_NE(task_before, task_after) << "shrinking TTL must install a new cleanup task"; -} -#endif - TEST_F(BrpcStubCacheTest, http_singleton_reinitialize_rebinds_pipeline_timer) { auto timer2 = std::make_unique(); ASSERT_OK(timer2->start()); From 4b7827160f795c02317850bbd84a2b8aef70607a Mon Sep 17 00:00:00 2001 From: "kaijian.ding" Date: Fri, 10 Jul 2026 17:59:47 +0800 Subject: [PATCH 4/7] [BugFix] Remove stub cache entry on first cleanup schedule failure Signed-off-by: kaijian.ding --- be/src/common/brpc/brpc_stub_cache.cpp | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/be/src/common/brpc/brpc_stub_cache.cpp b/be/src/common/brpc/brpc_stub_cache.cpp index 2c7affdbc63630..0157bc39adb2c4 100644 --- a/be/src/common/brpc/brpc_stub_cache.cpp +++ b/be/src/common/brpc/brpc_stub_cache.cpp @@ -132,6 +132,8 @@ std::shared_ptr BrpcStubCache::get_stub(const auto status = _timer->schedule((*stub_pool)->_cleanup_task.get(), tm); if (!status.ok()) { LOG(WARNING) << "Failed to schedule brpc cleanup task: " << endpoint; + _stub_map.erase(endpoint); + return new_pool->get_or_create(endpoint); } } else { (*stub_pool)->_cleanup_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s)); @@ -269,6 +271,8 @@ StatusOr> HttpBrpcStubCache::g auto status = _timer->schedule(stub_pair_ptr->cleanup_task.get(), tm); if (!status.ok()) { LOG(WARNING) << "Failed to schedule brpc cleanup task: " << endpoint; + _stub_map.erase(endpoint); + return stub; } } else { stub_pair_ptr->cleanup_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s)); @@ -356,6 +360,8 @@ StatusOr> LakeServiceBrp auto status = _timer->schedule(stub_pair_ptr->cleanup_task.get(), tm); if (!status.ok()) { LOG(WARNING) << "Failed to schedule brpc cleanup task: " << endpoint; + _stub_map.erase(endpoint); + return stub; } } else { stub_pair_ptr->cleanup_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s)); From 9f3cd2561225cdabf10d9413c7b21e4c25086294 Mon Sep 17 00:00:00 2001 From: "kaijian.ding" Date: Mon, 13 Jul 2026 22:39:23 +0800 Subject: [PATCH 5/7] Remove BrpcStubCacheTest friend declaration from production header Replace white-box _timer assertions with behavioral assertions via public API in the two singleton-reinitialize tests. Signed-off-by: kaijian.ding --- be/src/common/brpc/brpc_stub_cache.h | 7 ------- be/test/common/brpc/brpc_stub_cache_test.cpp | 8 ++------ 2 files changed, 2 insertions(+), 13 deletions(-) diff --git a/be/src/common/brpc/brpc_stub_cache.h b/be/src/common/brpc/brpc_stub_cache.h index 3830d248a9268b..d2bccfa8b74a14 100644 --- a/be/src/common/brpc/brpc_stub_cache.h +++ b/be/src/common/brpc/brpc_stub_cache.h @@ -55,7 +55,6 @@ namespace starrocks { class MetricRegistry; -class BrpcStubCacheTest; constexpr int TIMER_TASK_RUNNING = 1; @@ -125,8 +124,6 @@ class BrpcStubCache { template friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); - friend class starrocks::BrpcStubCacheTest; - void replace_cleanup_task_locked(const butil::EndPoint& endpoint, std::shared_ptr> task); @@ -170,8 +167,6 @@ class HttpBrpcStubCache { template friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); - friend class starrocks::BrpcStubCacheTest; - void replace_cleanup_task_locked(const butil::EndPoint& endpoint, std::shared_ptr> task); @@ -209,8 +204,6 @@ class LakeServiceBrpcStubCache { template friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); - friend class starrocks::BrpcStubCacheTest; - void replace_cleanup_task_locked(const butil::EndPoint& endpoint, std::shared_ptr> task); diff --git a/be/test/common/brpc/brpc_stub_cache_test.cpp b/be/test/common/brpc/brpc_stub_cache_test.cpp index 8fd95e98dbaa92..a7bd080a3ff2e3 100644 --- a/be/test/common/brpc/brpc_stub_cache_test.cpp +++ b/be/test/common/brpc/brpc_stub_cache_test.cpp @@ -270,7 +270,6 @@ TEST_F(BrpcStubCacheTest, http_singleton_reinitialize_rebinds_pipeline_timer) { HttpBrpcStubCache::initialize(_timer.get()); auto* cache = HttpBrpcStubCache::getInstance(); ASSERT_NE(nullptr, cache); - ASSERT_EQ(_timer.get(), cache->_timer); TNetworkAddress address; address.hostname = "127.0.0.1"; @@ -280,10 +279,9 @@ TEST_F(BrpcStubCacheTest, http_singleton_reinitialize_rebinds_pipeline_timer) { ASSERT_NE(nullptr, *stub); cache->shutdown(); - ASSERT_EQ(nullptr, cache->_timer); + ASSERT_FALSE(cache->get_http_stub(address).ok()); HttpBrpcStubCache::initialize(timer2.get()); - ASSERT_EQ(timer2.get(), cache->_timer); auto rebound_stub = cache->get_http_stub(address); ASSERT_TRUE(rebound_stub.ok()); @@ -300,17 +298,15 @@ TEST_F(BrpcStubCacheTest, lake_singleton_reinitialize_rebinds_pipeline_timer) { LakeServiceBrpcStubCache::initialize(_timer.get()); auto* cache = LakeServiceBrpcStubCache::getInstance(); ASSERT_NE(nullptr, cache); - ASSERT_EQ(_timer.get(), cache->_timer); auto stub = cache->get_stub("127.0.0.1", 123); ASSERT_TRUE(stub.ok()); ASSERT_NE(nullptr, *stub); cache->shutdown(); - ASSERT_EQ(nullptr, cache->_timer); + ASSERT_FALSE(cache->get_stub("127.0.0.1", 123).ok()); LakeServiceBrpcStubCache::initialize(timer2.get()); - ASSERT_EQ(timer2.get(), cache->_timer); auto rebound_stub = cache->get_stub("127.0.0.1", 123); ASSERT_TRUE(rebound_stub.ok()); From 0d0722f190b18af3c76b7db7647083c3d5b3be76 Mon Sep 17 00:00:00 2001 From: "kaijian.ding" Date: Tue, 14 Jul 2026 14:10:23 +0800 Subject: [PATCH 6/7] [BugFix] Fix UAF in brpc stub cache cleanup task on reinitialize Signed-off-by: kaijian.ding --- be/src/common/brpc/brpc_stub_cache.cpp | 21 ++++++++++----- be/src/common/brpc/brpc_stub_cache.h | 36 +++++++++++++++++++++++--- 2 files changed, 47 insertions(+), 10 deletions(-) diff --git a/be/src/common/brpc/brpc_stub_cache.cpp b/be/src/common/brpc/brpc_stub_cache.cpp index 0157bc39adb2c4..2163a3a2905dab 100644 --- a/be/src/common/brpc/brpc_stub_cache.cpp +++ b/be/src/common/brpc/brpc_stub_cache.cpp @@ -108,12 +108,15 @@ BrpcStubCache::~BrpcStubCache() { wait_clean_tasks_terminate(this, [](const std::shared_ptr& pool) { return pool->_cleanup_task; }); } -void BrpcStubCache::replace_cleanup_task_locked(const butil::EndPoint& endpoint, - std::shared_ptr> task) { +bool BrpcStubCache::replace_cleanup_task_locked( + const butil::EndPoint& endpoint, + std::shared_ptr> task) { auto pool = _stub_map.seek(endpoint); if (pool != nullptr) { (*pool)->_cleanup_task = std::move(task); + return true; } + return false; } std::shared_ptr BrpcStubCache::get_stub(const butil::EndPoint& endpoint) { @@ -222,12 +225,15 @@ void HttpBrpcStubCache::shutdown() { wait_clean_tasks_terminate(this, [](const StubEntry& entry) { return entry.cleanup_task; }); } -void HttpBrpcStubCache::replace_cleanup_task_locked(const butil::EndPoint& endpoint, - std::shared_ptr> task) { +bool HttpBrpcStubCache::replace_cleanup_task_locked( + const butil::EndPoint& endpoint, + std::shared_ptr> task) { auto entry = _stub_map.seek(endpoint); if (entry != nullptr) { entry->cleanup_task = std::move(task); + return true; } + return false; } StatusOr> HttpBrpcStubCache::get_http_stub( @@ -314,12 +320,15 @@ void LakeServiceBrpcStubCache::shutdown() { wait_clean_tasks_terminate(this, [](const StubEntry& entry) { return entry.cleanup_task; }); } -void LakeServiceBrpcStubCache::replace_cleanup_task_locked( - const butil::EndPoint& endpoint, std::shared_ptr> task) { +bool LakeServiceBrpcStubCache::replace_cleanup_task_locked( + const butil::EndPoint& endpoint, + std::shared_ptr> task) { auto entry = _stub_map.seek(endpoint); if (entry != nullptr) { entry->cleanup_task = std::move(task); + return true; } + return false; } DEFINE_FAIL_POINT(get_stub_return_nullptr); diff --git a/be/src/common/brpc/brpc_stub_cache.h b/be/src/common/brpc/brpc_stub_cache.h index d2bccfa8b74a14..9caea3b7362edb 100644 --- a/be/src/common/brpc/brpc_stub_cache.h +++ b/be/src/common/brpc/brpc_stub_cache.h @@ -71,6 +71,13 @@ class EndpointCleanupTask : public BthreadTimerTask { if (_cache->_stopping) { return; } + // Make sure this task is still the authoritative cleanup task for the + // endpoint before rescheduling. If shutdown() cleared the cache or a new + // entry was created for the same endpoint, this task is stale and must + // not schedule anything. + if (!_cache->is_cleanup_task_owner_locked(_endpoint, this)) { + return; + } int64_t now_us = butil::gettimeofday_us(); if (now_us >= _deadline) { LOG(INFO) << "cleanup brpc stub, endpoint:" << _endpoint << ", idle for " << (now_us - _deadline) / 1000 @@ -80,7 +87,9 @@ class EndpointCleanupTask : public BthreadTimerTask { } auto new_task = std::make_shared>(_cache, _endpoint, _ttl_seconds); new_task->_deadline = _deadline; - _cache->replace_cleanup_task_locked(_endpoint, new_task); + if (!_cache->replace_cleanup_task_locked(_endpoint, new_task)) { + return; + } timespec tm = butil::microseconds_to_timespec(_deadline); auto status = _cache->_timer->schedule(new_task.get(), tm); if (!status.ok()) { @@ -124,7 +133,13 @@ class BrpcStubCache { template friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); - void replace_cleanup_task_locked(const butil::EndPoint& endpoint, + bool is_cleanup_task_owner_locked( + const butil::EndPoint& endpoint, const EndpointCleanupTask* task) const { + auto pool = _stub_map.seek(endpoint); + return pool != nullptr && (*pool)->_cleanup_task.get() == task; + } + + bool replace_cleanup_task_locked(const butil::EndPoint& endpoint, std::shared_ptr> task); struct Metrics; @@ -167,7 +182,13 @@ class HttpBrpcStubCache { template friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); - void replace_cleanup_task_locked(const butil::EndPoint& endpoint, + bool is_cleanup_task_owner_locked( + const butil::EndPoint& endpoint, const EndpointCleanupTask* task) const { + auto entry = _stub_map.seek(endpoint); + return entry != nullptr && entry->cleanup_task.get() == task; + } + + bool replace_cleanup_task_locked(const butil::EndPoint& endpoint, std::shared_ptr> task); struct StubEntry { @@ -204,7 +225,14 @@ class LakeServiceBrpcStubCache { template friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); - void replace_cleanup_task_locked(const butil::EndPoint& endpoint, + bool is_cleanup_task_owner_locked( + const butil::EndPoint& endpoint, + const EndpointCleanupTask* task) const { + auto entry = _stub_map.seek(endpoint); + return entry != nullptr && entry->cleanup_task.get() == task; + } + + bool replace_cleanup_task_locked(const butil::EndPoint& endpoint, std::shared_ptr> task); struct StubEntry { From 4b8ba5c04a5a3a96dcab65a26b22bfeb15a3d09b Mon Sep 17 00:00:00 2001 From: wanpengfei-git Date: Tue, 14 Jul 2026 06:12:35 +0000 Subject: [PATCH 7/7] style: auto-fix clang-format issues --- be/src/common/brpc/brpc_stub_cache.cpp | 13 +++++-------- be/src/common/brpc/brpc_stub_cache.h | 13 ++++++------- 2 files changed, 11 insertions(+), 15 deletions(-) diff --git a/be/src/common/brpc/brpc_stub_cache.cpp b/be/src/common/brpc/brpc_stub_cache.cpp index 2163a3a2905dab..344c3488ee4082 100644 --- a/be/src/common/brpc/brpc_stub_cache.cpp +++ b/be/src/common/brpc/brpc_stub_cache.cpp @@ -108,9 +108,8 @@ BrpcStubCache::~BrpcStubCache() { wait_clean_tasks_terminate(this, [](const std::shared_ptr& pool) { return pool->_cleanup_task; }); } -bool BrpcStubCache::replace_cleanup_task_locked( - const butil::EndPoint& endpoint, - std::shared_ptr> task) { +bool BrpcStubCache::replace_cleanup_task_locked(const butil::EndPoint& endpoint, + std::shared_ptr> task) { auto pool = _stub_map.seek(endpoint); if (pool != nullptr) { (*pool)->_cleanup_task = std::move(task); @@ -225,9 +224,8 @@ void HttpBrpcStubCache::shutdown() { wait_clean_tasks_terminate(this, [](const StubEntry& entry) { return entry.cleanup_task; }); } -bool HttpBrpcStubCache::replace_cleanup_task_locked( - const butil::EndPoint& endpoint, - std::shared_ptr> task) { +bool HttpBrpcStubCache::replace_cleanup_task_locked(const butil::EndPoint& endpoint, + std::shared_ptr> task) { auto entry = _stub_map.seek(endpoint); if (entry != nullptr) { entry->cleanup_task = std::move(task); @@ -321,8 +319,7 @@ void LakeServiceBrpcStubCache::shutdown() { } bool LakeServiceBrpcStubCache::replace_cleanup_task_locked( - const butil::EndPoint& endpoint, - std::shared_ptr> task) { + const butil::EndPoint& endpoint, std::shared_ptr> task) { auto entry = _stub_map.seek(endpoint); if (entry != nullptr) { entry->cleanup_task = std::move(task); diff --git a/be/src/common/brpc/brpc_stub_cache.h b/be/src/common/brpc/brpc_stub_cache.h index 9caea3b7362edb..da3833fb6471d8 100644 --- a/be/src/common/brpc/brpc_stub_cache.h +++ b/be/src/common/brpc/brpc_stub_cache.h @@ -133,8 +133,8 @@ class BrpcStubCache { template friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); - bool is_cleanup_task_owner_locked( - const butil::EndPoint& endpoint, const EndpointCleanupTask* task) const { + bool is_cleanup_task_owner_locked(const butil::EndPoint& endpoint, + const EndpointCleanupTask* task) const { auto pool = _stub_map.seek(endpoint); return pool != nullptr && (*pool)->_cleanup_task.get() == task; } @@ -182,8 +182,8 @@ class HttpBrpcStubCache { template friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); - bool is_cleanup_task_owner_locked( - const butil::EndPoint& endpoint, const EndpointCleanupTask* task) const { + bool is_cleanup_task_owner_locked(const butil::EndPoint& endpoint, + const EndpointCleanupTask* task) const { auto entry = _stub_map.seek(endpoint); return entry != nullptr && entry->cleanup_task.get() == task; } @@ -225,9 +225,8 @@ class LakeServiceBrpcStubCache { template friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer); - bool is_cleanup_task_owner_locked( - const butil::EndPoint& endpoint, - const EndpointCleanupTask* task) const { + bool is_cleanup_task_owner_locked(const butil::EndPoint& endpoint, + const EndpointCleanupTask* task) const { auto entry = _stub_map.seek(endpoint); return entry != nullptr && entry->cleanup_task.get() == task; }