From ed9a32f5c46b5e84fc399b7313ce033cf6dd3e94 Mon Sep 17 00:00:00 2001 From: songgaoye Date: Wed, 28 Oct 2026 23:26:27 +0800 Subject: [PATCH 1/2] Fix bthread id butex state races --- src/bthread/id.cpp | 133 ++++++++++++++++++++--------------- test/bthread_id_unittest.cpp | 33 +++++++++ 2 files changed, 111 insertions(+), 55 deletions(-) diff --git a/src/bthread/id.cpp b/src/bthread/id.cpp index 9a97dc8f95..8f5a066496 100644 --- a/src/bthread/id.cpp +++ b/src/bthread/id.cpp @@ -20,6 +20,7 @@ // Date: Sun Aug 3 12:46:15 CST 2014 #include +#include "butil/atomicops.h" #include "butil/logging.h" #include "bthread/butex.h" // butex_* #include "bthread/mutex.h" @@ -107,6 +108,15 @@ struct PendingError { PendingError() : id(INVALID_BTHREAD_ID), error_code(0), location(nullptr) {} }; +inline uint32_t LoadButex(const butil::atomic* butex) { + return static_cast(butex->load(butil::memory_order_relaxed)); +} + +inline void StoreButex(butil::atomic* butex, uint32_t value, + butil::memory_order order = butil::memory_order_relaxed) { + butex->store(static_cast(value), order); +} + struct BAIDU_CACHELINE_ALIGNMENT Id { // first_ver ~ locked_ver - 1: unlocked versions // locked_ver: locked @@ -119,17 +129,17 @@ struct BAIDU_CACHELINE_ALIGNMENT Id { int (*on_error)(bthread_id_t, void*, int); int (*on_error2)(bthread_id_t, void*, int, const std::string&); const char *lock_location; - uint32_t* butex; - uint32_t* join_butex; + butil::atomic* butex; + butil::atomic* join_butex; SmallQueue pending_q; Id() { // Although value of the butex(as version part of bthread_id_t) // does not matter, we set it to 0 to make program more deterministic. - butex = bthread::butex_create_checked(); - join_butex = bthread::butex_create_checked(); - *butex = 0; - *join_butex = 0; + butex = bthread::butex_create_checked >(); + join_butex = bthread::butex_create_checked >(); + StoreButex(butex, 0); + StoreButex(join_butex, 0); } ~Id() { @@ -179,7 +189,7 @@ inline bool id_exists_with_true_negatives(bthread_id_t id) { uint32_t id_value(bthread_id_t id) { Id* const meta = address_resource(get_slot(id)); if (meta != nullptr) { - return *meta->butex; + return LoadButex(meta->butex); } return 0; // valid version never be zero } @@ -199,7 +209,7 @@ void id_status(bthread_id_t id, std::ostream &os) { return; } const uint32_t id_ver = bthread::get_version(id); - uint32_t* butex = meta->butex; + butil::atomic* butex = meta->butex; bool valid = true; void* data = nullptr; int (*on_error)(bthread_id_t, void*, int) = nullptr; @@ -229,7 +239,7 @@ void id_status(bthread_id_t id, std::ostream &os) { meta->pending_q.push(front); pending_q.push(front); } - butex_value = *butex; + butex_value = LoadButex(butex); } else { valid = false; } @@ -333,16 +343,19 @@ static int id_create_impl( meta->on_error = on_error; meta->on_error2 = on_error2; CHECK(meta->pending_q.empty()); - uint32_t* butex = meta->butex; - if (0 == *butex || *butex + ID_MAX_RANGE + 2 < *butex) { + butil::atomic* butex = meta->butex; + uint32_t butex_value = LoadButex(butex); + if (0 == butex_value || + butex_value + ID_MAX_RANGE + 2 < butex_value) { // Skip 0 so that bthread_id_t is never 0 // avoid overflow to make comparisons simpler. - *butex = 1; + StoreButex(butex, 1); + butex_value = 1; } - *meta->join_butex = *butex; - meta->first_ver = *butex; - meta->locked_ver = *butex + 1; - *id = make_id(*butex, slot); + StoreButex(meta->join_butex, butex_value); + meta->first_ver = butex_value; + meta->locked_ver = butex_value + 1; + *id = make_id(butex_value, slot); return 0; } return ENOMEM; @@ -366,16 +379,19 @@ static int id_create_ranged_impl( meta->on_error = on_error; meta->on_error2 = on_error2; CHECK(meta->pending_q.empty()); - uint32_t* butex = meta->butex; - if (0 == *butex || *butex + ID_MAX_RANGE + 2 < *butex) { + butil::atomic* butex = meta->butex; + uint32_t butex_value = LoadButex(butex); + if (0 == butex_value || + butex_value + ID_MAX_RANGE + 2 < butex_value) { // Skip 0 so that bthread_id_t is never 0 // avoid overflow to make comparisons simpler. - *butex = 1; + StoreButex(butex, 1); + butex_value = 1; } - *meta->join_butex = *butex; - meta->first_ver = *butex; - meta->locked_ver = *butex + range; - *id = make_id(*butex, slot); + StoreButex(meta->join_butex, butex_value); + meta->first_ver = butex_value; + meta->locked_ver = butex_value + range; + *id = make_id(butex_value, slot); return 0; } return ENOMEM; @@ -383,6 +399,9 @@ static int id_create_ranged_impl( } // namespace bthread +using bthread::LoadButex; +using bthread::StoreButex; + extern "C" { int bthread_id_create( @@ -409,11 +428,11 @@ int bthread_id_lock_and_reset_range_verbose( return EINVAL; } const uint32_t id_ver = bthread::get_version(id); - uint32_t* butex = meta->butex; + butil::atomic* butex = meta->butex; bool ever_contended = false; meta->mutex.lock(); while (meta->has_version(id_ver)) { - if (*butex == meta->first_ver) { + if (LoadButex(butex) == meta->first_ver) { // contended locker always wakes up the butex at unlock. meta->lock_location = location; if (range == 0) { @@ -429,15 +448,16 @@ int bthread_id_lock_and_reset_range_verbose( } else { meta->locked_ver = meta->first_ver + range; } - *butex = (ever_contended ? meta->contended_ver() : meta->locked_ver); + StoreButex(butex, ever_contended ? meta->contended_ver() + : meta->locked_ver); meta->mutex.unlock(); if (pdata) { *pdata = meta->data; } return 0; - } else if (*butex != meta->unlockable_ver()) { - *butex = meta->contended_ver(); - uint32_t expected_ver = *butex; + } else if (LoadButex(butex) != meta->unlockable_ver()) { + const uint32_t expected_ver = meta->contended_ver(); + StoreButex(butex, expected_ver); meta->mutex.unlock(); ever_contended = true; if (bthread::butex_wait(butex, expected_ver, nullptr) < 0 && @@ -465,19 +485,20 @@ int bthread_id_about_to_destroy(bthread_id_t id) { return EINVAL; } const uint32_t id_ver = bthread::get_version(id); - uint32_t* butex = meta->butex; + butil::atomic* butex = meta->butex; meta->mutex.lock(); if (!meta->has_version(id_ver)) { meta->mutex.unlock(); return EINVAL; } - if (*butex == meta->first_ver) { + if (LoadButex(butex) == meta->first_ver) { meta->mutex.unlock(); LOG(FATAL) << "bthread_id=" << id.value << " is not locked!"; return EPERM; } - const bool contended = (*butex == meta->contended_ver()); - *butex = meta->unlockable_ver(); + const bool contended = + (LoadButex(butex) == meta->contended_ver()); + StoreButex(butex, meta->unlockable_ver(), butil::memory_order_release); meta->mutex.unlock(); if (contended) { // wake up all waiting lockers. @@ -491,20 +512,21 @@ int bthread_id_cancel(bthread_id_t id) { if (!meta) { return EINVAL; } - uint32_t* butex = meta->butex; + butil::atomic* butex = meta->butex; const uint32_t id_ver = bthread::get_version(id); meta->mutex.lock(); if (!meta->has_version(id_ver)) { meta->mutex.unlock(); return EINVAL; } - if (*butex != meta->first_ver) { + if (LoadButex(butex) != meta->first_ver) { meta->mutex.unlock(); return EPERM; } - *butex = meta->end_ver(); - meta->first_ver = *butex; - meta->locked_ver = *butex; + const uint32_t next_ver = meta->end_ver(); + StoreButex(butex, next_ver, butil::memory_order_release); + meta->first_ver = next_ver; + meta->locked_ver = next_ver; meta->mutex.unlock(); return_resource(bthread::get_slot(id)); return 0; @@ -518,11 +540,11 @@ int bthread_id_join(bthread_id_t id) { return EINVAL; } const uint32_t id_ver = bthread::get_version(id); - uint32_t* join_butex = meta->join_butex; + butil::atomic* join_butex = meta->join_butex; while (1) { meta->mutex.lock(); const bool has_ver = meta->has_version(id_ver); - const uint32_t expected_ver = *join_butex; + const uint32_t expected_ver = LoadButex(join_butex); meta->mutex.unlock(); if (!has_ver) { break; @@ -540,18 +562,18 @@ int bthread_id_trylock(bthread_id_t id, void** pdata) { if (!meta) { return EINVAL; } - uint32_t* butex = meta->butex; + butil::atomic* butex = meta->butex; const uint32_t id_ver = bthread::get_version(id); meta->mutex.lock(); if (!meta->has_version(id_ver)) { meta->mutex.unlock(); return EINVAL; } - if (*butex != meta->first_ver) { + if (LoadButex(butex) != meta->first_ver) { meta->mutex.unlock(); return EBUSY; } - *butex = meta->locked_ver; + StoreButex(butex, meta->locked_ver); meta->mutex.unlock(); if (pdata != nullptr) { *pdata = meta->data; @@ -569,7 +591,7 @@ int bthread_id_unlock(bthread_id_t id) { if (!meta) { return EINVAL; } - uint32_t* butex = meta->butex; + butil::atomic* butex = meta->butex; // Release fence makes sure all changes made before signal visible to // woken-up waiters. const uint32_t id_ver = bthread::get_version(id); @@ -579,7 +601,7 @@ int bthread_id_unlock(bthread_id_t id) { LOG(FATAL) << "Invalid bthread_id=" << id.value; return EINVAL; } - if (*butex == meta->first_ver) { + if (LoadButex(butex) == meta->first_ver) { meta->mutex.unlock(); LOG(FATAL) << "bthread_id=" << id.value << " is not locked!"; return EPERM; @@ -595,8 +617,9 @@ int bthread_id_unlock(bthread_id_t id) { front.error_text); } } else { - const bool contended = (*butex == meta->contended_ver()); - *butex = meta->first_ver; + const bool contended = + (LoadButex(butex) == meta->contended_ver()); + StoreButex(butex, meta->first_ver, butil::memory_order_release); meta->mutex.unlock(); if (contended) { // We may wake up already-reused id, but that's OK. @@ -611,8 +634,8 @@ int bthread_id_unlock_and_destroy(bthread_id_t id) { if (!meta) { return EINVAL; } - uint32_t* butex = meta->butex; - uint32_t* join_butex = meta->join_butex; + butil::atomic* butex = meta->butex; + butil::atomic* join_butex = meta->join_butex; const uint32_t id_ver = bthread::get_version(id); meta->mutex.lock(); if (!meta->has_version(id_ver)) { @@ -620,14 +643,14 @@ int bthread_id_unlock_and_destroy(bthread_id_t id) { LOG(FATAL) << "Invalid bthread_id=" << id.value; return EINVAL; } - if (*butex == meta->first_ver) { + if (LoadButex(butex) == meta->first_ver) { meta->mutex.unlock(); LOG(FATAL) << "bthread_id=" << id.value << " is not locked!"; return EPERM; } const uint32_t next_ver = meta->end_ver(); - *butex = next_ver; - *join_butex = next_ver; + StoreButex(butex, next_ver, butil::memory_order_release); + StoreButex(join_butex, next_ver, butil::memory_order_release); meta->first_ver = next_ver; meta->locked_ver = next_ver; meta->pending_q.clear(); @@ -711,14 +734,14 @@ int bthread_id_error2_verbose(bthread_id_t id, int error_code, return EINVAL; } const uint32_t id_ver = bthread::get_version(id); - uint32_t* butex = meta->butex; + butil::atomic* butex = meta->butex; meta->mutex.lock(); if (!meta->has_version(id_ver)) { meta->mutex.unlock(); return EINVAL; } - if (*butex == meta->first_ver) { - *butex = meta->locked_ver; + if (LoadButex(butex) == meta->first_ver) { + StoreButex(butex, meta->locked_ver); meta->lock_location = location; meta->mutex.unlock(); if (meta->on_error) { diff --git a/test/bthread_id_unittest.cpp b/test/bthread_id_unittest.cpp index 59352ba5d3..e87d877254 100644 --- a/test/bthread_id_unittest.cpp +++ b/test/bthread_id_unittest.cpp @@ -16,6 +16,8 @@ // under the License. #include +#include +#include #include #include "butil/time.h" #include "butil/macros.h" @@ -112,6 +114,37 @@ TEST(BthreadIdTest, join_before_destroy) { ASSERT_EQ(1UL, non_null_ret); } +struct JoinRaceArgs { + bthread_id_t id; + std::atomic entered; +}; + +void* join_race_waiter(void* arg) { + JoinRaceArgs* args = static_cast(arg); + args->entered.store(true, std::memory_order_release); + EXPECT_EQ(0, bthread_id_join(args->id)); + return nullptr; +} + +TEST(BthreadIdTest, join_races_with_destroy) { + for (int i = 0; i < 100; ++i) { + bthread_id_t id; + ASSERT_EQ(0, bthread_id_create(&id, nullptr, nullptr)); + JoinRaceArgs args; + args.id = id; + args.entered.store(false, std::memory_order_relaxed); + pthread_t waiter; + ASSERT_EQ(0, pthread_create(&waiter, nullptr, join_race_waiter, &args)); + while (!args.entered.load(std::memory_order_acquire)) { + std::this_thread::yield(); + } + void* data = nullptr; + ASSERT_EQ(0, bthread_id_trylock(id, &data)); + ASSERT_EQ(0, bthread_id_unlock_and_destroy(id)); + ASSERT_EQ(0, pthread_join(waiter, nullptr)); + } +} + struct OnResetArg { bthread_id_t id; int error_code; From 73f925cae278df4a6aa2f55efa80d50ce1351449 Mon Sep 17 00:00:00 2001 From: songgaoye Date: Thu, 29 Oct 2026 23:28:40 +0800 Subject: [PATCH 2/2] Make bthread_id_join destroy race test deterministic --- src/bthread/id.cpp | 11 ++++++++ test/bthread_id_unittest.cpp | 50 ++++++++++++++++++++++++------------ 2 files changed, 44 insertions(+), 17 deletions(-) diff --git a/src/bthread/id.cpp b/src/bthread/id.cpp index 8f5a066496..c1ce086106 100644 --- a/src/bthread/id.cpp +++ b/src/bthread/id.cpp @@ -30,6 +30,11 @@ namespace bthread { +#if defined(UNIT_TEST) +void (*g_bthread_id_join_before_wait_for_test)(void*) = nullptr; +void* g_bthread_id_join_before_wait_arg_for_test = nullptr; +#endif + // This queue reduces the chance to allocate memory for deque template class SmallQueue { @@ -549,6 +554,12 @@ int bthread_id_join(bthread_id_t id) { if (!has_ver) { break; } +#if defined(UNIT_TEST) + if (bthread::g_bthread_id_join_before_wait_for_test != nullptr) { + bthread::g_bthread_id_join_before_wait_for_test( + bthread::g_bthread_id_join_before_wait_arg_for_test); + } +#endif if (bthread::butex_wait(join_butex, expected_ver, nullptr) < 0 && errno != EWOULDBLOCK && errno != EINTR) { return errno; diff --git a/test/bthread_id_unittest.cpp b/test/bthread_id_unittest.cpp index e87d877254..4f4a8ef791 100644 --- a/test/bthread_id_unittest.cpp +++ b/test/bthread_id_unittest.cpp @@ -28,6 +28,8 @@ namespace bthread { void id_status(bthread_id_t, std::ostream &); uint32_t id_value(bthread_id_t id); +extern void (*g_bthread_id_join_before_wait_for_test)(void*); +extern void* g_bthread_id_join_before_wait_arg_for_test; } namespace { @@ -116,33 +118,47 @@ TEST(BthreadIdTest, join_before_destroy) { struct JoinRaceArgs { bthread_id_t id; - std::atomic entered; + std::atomic join_butex_read; + std::atomic allow_wait; }; +void pause_join_before_wait(void* arg) { + JoinRaceArgs* args = static_cast(arg); + args->join_butex_read.store(true, std::memory_order_release); + while (!args->allow_wait.load(std::memory_order_acquire)) { + std::this_thread::yield(); + } +} + void* join_race_waiter(void* arg) { JoinRaceArgs* args = static_cast(arg); - args->entered.store(true, std::memory_order_release); EXPECT_EQ(0, bthread_id_join(args->id)); return nullptr; } TEST(BthreadIdTest, join_races_with_destroy) { - for (int i = 0; i < 100; ++i) { - bthread_id_t id; - ASSERT_EQ(0, bthread_id_create(&id, nullptr, nullptr)); - JoinRaceArgs args; - args.id = id; - args.entered.store(false, std::memory_order_relaxed); - pthread_t waiter; - ASSERT_EQ(0, pthread_create(&waiter, nullptr, join_race_waiter, &args)); - while (!args.entered.load(std::memory_order_acquire)) { - std::this_thread::yield(); - } - void* data = nullptr; - ASSERT_EQ(0, bthread_id_trylock(id, &data)); - ASSERT_EQ(0, bthread_id_unlock_and_destroy(id)); - ASSERT_EQ(0, pthread_join(waiter, nullptr)); + bthread_id_t id; + ASSERT_EQ(0, bthread_id_create(&id, nullptr, nullptr)); + JoinRaceArgs args; + args.id = id; + args.join_butex_read.store(false, std::memory_order_relaxed); + args.allow_wait.store(false, std::memory_order_relaxed); + bthread::g_bthread_id_join_before_wait_for_test = pause_join_before_wait; + bthread::g_bthread_id_join_before_wait_arg_for_test = &args; + + pthread_t waiter; + ASSERT_EQ(0, pthread_create(&waiter, nullptr, join_race_waiter, &args)); + while (!args.join_butex_read.load(std::memory_order_acquire)) { + std::this_thread::yield(); } + void* data = nullptr; + ASSERT_EQ(0, bthread_id_trylock(id, &data)); + ASSERT_EQ(0, bthread_id_unlock_and_destroy(id)); + args.allow_wait.store(true, std::memory_order_release); + ASSERT_EQ(0, pthread_join(waiter, nullptr)); + + bthread::g_bthread_id_join_before_wait_for_test = nullptr; + bthread::g_bthread_id_join_before_wait_arg_for_test = nullptr; } struct OnResetArg {