This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 6ff47acf fix(bthread): Synchronize keytable pool pointer access (#3638)
6ff47acf is described below
commit 6ff47acf554262f6fc2d0b2c6fe538b8208a6675
Author: darion-yaphet <[email protected]>
AuthorDate: Fri Oct 23 12:33:42 2026 +0800
fix(bthread): Synchronize keytable pool pointer access (#3638)
* fix(bthread): synchronize keytable pool access
Read pool pointers and the destroyed state under the existing rwlock. Keep
only a boolean availability hint between the read and write lock sections, then
recheck the pool state under the write lock.
Add coverage for concurrent borrowing, returning, and pool destruction,
including reclamation of tables returned after destruction.
* test(bthread): ensure keytable return after pool destruction
* test(bthread): isolate keytable pool test flag
---
src/bthread/key.cpp | 10 ++++--
test/bthread_key_unittest.cpp | 75 +++++++++++++++++++++++++++++++++++++++++++
2 files changed, 83 insertions(+), 2 deletions(-)
diff --git a/src/bthread/key.cpp b/src/bthread/key.cpp
index 8db23e08..74e718a4 100644
--- a/src/bthread/key.cpp
+++ b/src/bthread/key.cpp
@@ -322,10 +322,15 @@ private:
};
KeyTable* borrow_keytable(bthread_keytable_pool_t* pool) {
- if (pool != nullptr && (pool->list || pool->free_keytables)) {
+ if (pool != nullptr) {
KeyTable* p;
+ bool has_free_keytables;
{
pthread_rwlock_rdlock(&pool->rwlock);
+ if (pool->destroyed) {
+ pthread_rwlock_unlock(&pool->rwlock);
+ return nullptr;
+ }
auto list = (butil::ThreadLocal<bthread::KeyTableList>*)pool->list;
if (list) {
p = list->get()->remove_front();
@@ -334,9 +339,10 @@ KeyTable* borrow_keytable(bthread_keytable_pool_t* pool) {
return p;
}
}
+ has_free_keytables = pool->free_keytables != nullptr;
pthread_rwlock_unlock(&pool->rwlock);
}
- if (pool->free_keytables) {
+ if (has_free_keytables) {
pthread_rwlock_wrlock(&pool->rwlock);
if (pool->destroyed) {
pthread_rwlock_unlock(&pool->rwlock);
diff --git a/test/bthread_key_unittest.cpp b/test/bthread_key_unittest.cpp
index f0bfadec..a883ad1e 100644
--- a/test/bthread_key_unittest.cpp
+++ b/test/bthread_key_unittest.cpp
@@ -26,6 +26,7 @@
#include "butil/scoped_lock.h"
#include "butil/logging.h"
#include "bthread/bthread.h"
+#include "bthread/countdown_event.h"
#include "bthread/unstable.h"
using namespace bthread;
namespace bthread {
@@ -566,6 +567,80 @@ TEST(KeyTest, borrow_and_return_keytable_when_using_pool) {
ASSERT_EQ(0, bthread_keytable_pool_destroy(&test_pool));
}
+struct ConcurrentPoolData {
+ bthread_keytable_pool_t pool;
+ CountdownEvent ready;
+ CountdownEvent start;
+ CountdownEvent destroyed;
+ Counters counters;
+
+ explicit ConcurrentPoolData(int nworkers) : ready(nworkers) {}
+};
+
+static void* create_concurrent_pool_data(const void* arg) {
+ Counters* counters = const_cast<Counters*>(static_cast<const
Counters*>(arg));
+ counters->ncreate.fetch_add(1, butil::memory_order_relaxed);
+ return counters;
+}
+
+static void destroy_concurrent_pool_data(void* arg) {
+ static_cast<Counters*>(arg)->ndestroy.fetch_add(1,
butil::memory_order_relaxed);
+}
+
+static void* borrow_and_return_during_pool_destruction(void* arg) {
+ ConcurrentPoolData* data = static_cast<ConcurrentPoolData*>(arg);
+ // Keep a borrowed table across destruction as well as racing new borrows
+ // with destruction. The pool and key remain valid until all workers exit.
+ KeyTable* held = borrow_keytable(&data->pool);
+ EXPECT_NE(nullptr, held);
+ data->ready.signal();
+ data->start.wait();
+ for (int i = 0; i < 4096; ++i) {
+ KeyTable* kt = borrow_keytable(&data->pool);
+ return_keytable(&data->pool, kt);
+ }
+ data->destroyed.wait();
+ return_keytable(&data->pool, held);
+ return nullptr;
+}
+
+TEST(KeyTest, borrow_and_return_keytable_during_pool_destruction) {
+ GFLAGS_NAMESPACE::FlagSaver flag_saver;
+ bthread::FLAGS_borrow_from_globle_size = 20;
+ const size_t NWORKERS = 8;
+ const size_t NTABLES = NWORKERS * bthread::FLAGS_borrow_from_globle_size;
+ ConcurrentPoolData data(NWORKERS);
+ ASSERT_EQ(0, bthread_keytable_pool_init(&data.pool));
+ bthread_key_t key;
+ ASSERT_EQ(0, bthread_key_create(&key, destroy_concurrent_pool_data));
+ bthread_keytable_pool_reserve(&data.pool, NTABLES, key,
+ create_concurrent_pool_data, &data.counters);
+ ASSERT_EQ(NTABLES,
data.counters.ncreate.load(butil::memory_order_relaxed));
+
+ pthread_t workers[NWORKERS];
+ size_t nstarted = 0;
+ for (; nstarted < NWORKERS; ++nstarted) {
+ int rc = pthread_create(&workers[nstarted], nullptr,
+ borrow_and_return_during_pool_destruction,
&data);
+ EXPECT_EQ(0, rc);
+ if (rc != 0) {
+ data.ready.signal(NWORKERS - nstarted);
+ break;
+ }
+ }
+ data.ready.wait();
+ data.start.signal();
+ EXPECT_EQ(0, bthread_keytable_pool_destroy(&data.pool));
+ data.destroyed.signal();
+ for (size_t i = 0; i < nstarted; ++i) {
+ EXPECT_EQ(0, pthread_join(workers[i], nullptr));
+ }
+ EXPECT_EQ(nullptr, borrow_keytable(&data.pool));
+ EXPECT_EQ(NTABLES,
data.counters.ndestroy.load(butil::memory_order_relaxed));
+ EXPECT_EQ(0, bthread_key_delete(key));
+ EXPECT_EQ(0, pthread_rwlock_destroy(&data.pool.rwlock));
+}
+
// NOTE: lid is short for 'lock in dtor'.
butil::atomic<size_t> lid_seq(1);
std::vector<size_t> lid_seqs;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
