From 39eda3f8dfbbac965a3510c75ec3a896e252b4b3 Mon Sep 17 00:00:00 2001 From: lidongyang Date: Wed, 9 Sep 2026 10:55:50 +0800 Subject: [PATCH] [fix](be) Serialize bvar TLS agent lifetime updates ### What problem does this PR solve? Issue Number: N/A Related PR: #66977 Problem Summary: Concurrent destruction of a bvar combiner and TLS agent teardown can access the same std::weak_ptr at the same time. This violates the weak_ptr concurrency contract and can corrupt its shared ownership state, which was observed as an ASAN heap-use-after-free in TableRpcQpsRegistryTest.ConcurrentRecordAndCleanup. Serialize accesses to each agent's weak_ptr while using an atomic attachment flag for the reducer update hot path. Amplify the existing concurrent cleanup test to exercise repeated cross-thread combiner destruction. ### Release note None ### Check List (For Author) - Test: - ClangFormat 16 check - Build hygiene check - Applied all brpc patches in build order - GCC 15 C++17 syntax instantiation of the patched AgentCombiner - Behavior changed: No - Does this need documentation: No --- .../cloud_ms_backpressure_handler_test.cpp | 33 ++++++------ ...4.0-fix-agent-combiner-thread-safety.patch | 53 +++++++++++++------ 2 files changed, 56 insertions(+), 30 deletions(-) diff --git a/be/test/cloud/cloud_ms_backpressure_handler_test.cpp b/be/test/cloud/cloud_ms_backpressure_handler_test.cpp index c3b4fcea368566..8a21234e46e770 100644 --- a/be/test/cloud/cloud_ms_backpressure_handler_test.cpp +++ b/be/test/cloud/cloud_ms_backpressure_handler_test.cpp @@ -320,23 +320,26 @@ TEST_F(TableRpcQpsRegistryTest, CleanupThreadRunsIndependently) { } TEST_F(TableRpcQpsRegistryTest, ConcurrentRecordAndCleanup) { - TableRpcQpsRegistry registry(std::chrono::hours(1), std::chrono::milliseconds(0)); + for (int round = 0; round < 20; ++round) { + SCOPED_TRACE(round); + TableRpcQpsRegistry registry(std::chrono::hours(1), std::chrono::milliseconds(0)); - std::thread recorder([®istry]() { - for (int i = 0; i < 10000; ++i) { - registry.record(LoadRelatedRpc::PREPARE_ROWSET, i % 100); - } - }); - std::thread cleaner([®istry]() { - for (int i = 0; i < 1000; ++i) { - registry.cleanup_inactive_tables(); - } - }); - recorder.join(); - cleaner.join(); + std::thread recorder([®istry]() { + for (int i = 0; i < 10000; ++i) { + registry.record(LoadRelatedRpc::PREPARE_ROWSET, i % 100); + } + }); + std::thread cleaner([®istry]() { + for (int i = 0; i < 1000; ++i) { + registry.cleanup_inactive_tables(); + } + }); + recorder.join(); + cleaner.join(); - registry.record(LoadRelatedRpc::PREPARE_ROWSET, 100); - EXPECT_GE(registry.get_tracked_table_count(LoadRelatedRpc::PREPARE_ROWSET), 1); + registry.record(LoadRelatedRpc::PREPARE_ROWSET, 100); + EXPECT_GE(registry.get_tracked_table_count(LoadRelatedRpc::PREPARE_ROWSET), 1); + } } // ============== TableRpcThrottler Tests ============== diff --git a/thirdparty/patches/brpc-1.4.0-fix-agent-combiner-thread-safety.patch b/thirdparty/patches/brpc-1.4.0-fix-agent-combiner-thread-safety.patch index b7d0a9a93dfc79..91e5391b734907 100644 --- a/thirdparty/patches/brpc-1.4.0-fix-agent-combiner-thread-safety.patch +++ b/thirdparty/patches/brpc-1.4.0-fix-agent-combiner-thread-safety.patch @@ -27,7 +27,7 @@ index 6a6ab80..65ed9a1 100644 inline void load(T* out) { *out = _value.load(butil::memory_order_relaxed); } -@@ -153,24 +153,26 @@ private: +@@ -153,24 +153,37 @@ private: }; template @@ -47,11 +47,13 @@ index 6a6ab80..65ed9a1 100644 - Agent() : combiner(NULL) {} + struct Agent : public butil::LinkNode { ++ Agent() : _has_combiner(false) {} ++ ~Agent() { - if (combiner) { - combiner->commit_and_erase(this); - combiner = NULL; -+ self_shared_type c = combiner.lock(); ++ self_shared_type c = lock_combiner(); + if (NULL != c) { + c->commit_and_erase(this); } @@ -60,10 +62,20 @@ index 6a6ab80..65ed9a1 100644 - void reset(const ElementTp& val, self_type* c) { + + void reset(const ElementTp& val, const self_shared_type& c) { - combiner = c; +- combiner = c; ++ { ++ butil::AutoLock guard(_combiner_lock); ++ if (NULL != c) { ++ combiner = c; ++ _has_combiner.store(true, butil::memory_order_release); ++ } else { ++ _has_combiner.store(false, butil::memory_order_release); ++ combiner.reset(); ++ } ++ } element.store(val); } -@@ -181,11 +183,11 @@ friend class GlobalValue; +@@ -181,11 +194,11 @@ friend class GlobalValue; // void operator()(GlobalValue & global_value, // ElementTp & local_value) const { // if (test_for_merging(local_value)) { @@ -77,7 +89,7 @@ index 6a6ab80..65ed9a1 100644 // // *g and local_value are not changed provided // // merge_global is called from the thread owning // // the agent. -@@ -200,16 +202,23 @@ friend class GlobalValue; +@@ -200,16 +213,34 @@ friend class GlobalValue; // ... // } // }; @@ -90,7 +102,7 @@ index 6a6ab80..65ed9a1 100644 - element.merge_global(op, g); + void merge_global(const Op &op, self_shared_type c = NULL) { + if (NULL == c) { -+ c = combiner.lock(); ++ c = lock_combiner(); + } + if (NULL != c) { + GlobalValue g(this, c.get()); @@ -101,12 +113,23 @@ index 6a6ab80..65ed9a1 100644 - self_type *combiner; ElementContainer element; + private: -+ friend class AgentCombiner; ++ friend class AgentCombiner; ++ self_shared_type lock_combiner() const { ++ butil::AutoLock guard(_combiner_lock); ++ return combiner.lock(); ++ } ++ ++ bool has_combiner() const { ++ return _has_combiner.load(butil::memory_order_acquire); ++ } ++ // Serialize the same weak_ptr while keeping the update path lock-free. ++ mutable butil::Lock _combiner_lock; ++ butil::atomic _has_combiner; + self_weak_type combiner; }; typedef detail::AgentGroup AgentGroup; -@@ -231,7 +240,7 @@ friend class GlobalValue; +@@ -231,7 +262,7 @@ friend class GlobalValue; _id = -1; } } @@ -115,7 +138,7 @@ index 6a6ab80..65ed9a1 100644 // [Threadsafe] May be called from anywhere ResultTp combine_agents() const { ElementTp tls_value; -@@ -245,10 +254,10 @@ friend class GlobalValue; +@@ -245,10 +276,10 @@ friend class GlobalValue; return ret; } @@ -130,7 +153,7 @@ index 6a6ab80..65ed9a1 100644 // [Threadsafe] May be called from anywhere. ResultTp reset_all_agents() { -@@ -265,7 +274,7 @@ friend class GlobalValue; +@@ -265,7 +296,7 @@ friend class GlobalValue; } // Always called from the thread owning the agent. @@ -139,7 +162,7 @@ index 6a6ab80..65ed9a1 100644 if (NULL == agent) { return; } -@@ -279,7 +288,7 @@ friend class GlobalValue; +@@ -279,7 +310,7 @@ friend class GlobalValue; } // Always called from the thread owning the agent @@ -148,7 +171,7 @@ index 6a6ab80..65ed9a1 100644 if (NULL == agent) { return; } -@@ -290,7 +299,7 @@ friend class GlobalValue; +@@ -290,7 +321,7 @@ friend class GlobalValue; } // We need this function to be as fast as possible. @@ -157,12 +180,12 @@ index 6a6ab80..65ed9a1 100644 Agent* agent = AgentGroup::get_tls_agent(_id); if (!agent) { // Create the agent -@@ -300,10 +309,10 @@ friend class GlobalValue; +@@ -300,10 +331,10 @@ friend class GlobalValue; return NULL; } } - if (agent->combiner) { -+ if (!agent->combiner.expired()) { ++ if (agent->has_combiner()) { return agent; } - agent->reset(_element_identity, this); @@ -170,7 +193,7 @@ index 6a6ab80..65ed9a1 100644 // TODO: Is uniqueness-checking necessary here? { butil::AutoLock guard(_lock); -@@ -317,8 +326,7 @@ friend class GlobalValue; +@@ -317,8 +348,7 @@ friend class GlobalValue; // reseting agents is must because the agent object may be reused. // Set element to be default-constructed so that if it's non-pod, // internal allocations should be released.