Skip to content

Fix bthread id butex state races #3689

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
songgaoye wants to merge 1 commit into
base: master
Choose a base branch
from
+111 −55
Open

Fix bthread id butex state races #3689

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 78 additions & 55 deletions src/bthread/id.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
// Date: Sun Aug 3 12:46:15 CST 2014

#include <deque>
#include "butil/atomicops.h"
#include "butil/logging.h"
#include "bthread/butex.h" // butex_*
#include "bthread/mutex.h"
Expand Down Expand Up @@ -107,6 +108,15 @@ struct PendingError {
PendingError() : id(INVALID_BTHREAD_ID), error_code(0), location(nullptr) {}
};

inline uint32_t LoadButex(const butil::atomic<int>* butex) {
return static_cast<uint32_t>(butex->load(butil::memory_order_relaxed));
}

inline void StoreButex(butil::atomic<int>* butex, uint32_t value,
butil::memory_order order = butil::memory_order_relaxed) {
butex->store(static_cast<int>(value), order);
}

struct BAIDU_CACHELINE_ALIGNMENT Id {
// first_ver ~ locked_ver - 1: unlocked versions
// locked_ver: locked
Expand All @@ -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<int>* butex;
butil::atomic<int>* join_butex;
SmallQueue<PendingError, 2> 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<uint32_t>();
join_butex = bthread::butex_create_checked<uint32_t>();
*butex = 0;
*join_butex = 0;
butex = bthread::butex_create_checked<butil::atomic<int> >();
join_butex = bthread::butex_create_checked<butil::atomic<int> >();
StoreButex(butex, 0);
StoreButex(join_butex, 0);
}

~Id() {
Expand Down Expand Up @@ -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
}
Expand All @@ -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<int>* butex = meta->butex;
bool valid = true;
void* data = nullptr;
int (*on_error)(bthread_id_t, void*, int) = nullptr;
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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<int>* 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;
Expand All @@ -366,23 +379,29 @@ 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<int>* 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;
}

} // namespace bthread

using bthread::LoadButex;
using bthread::StoreButex;

extern "C" {

int bthread_id_create(
Expand All @@ -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<int>* 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) {
Expand All @@ -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 &&
Expand Down Expand Up @@ -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<int>* 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.
Expand All @@ -491,20 +512,21 @@ int bthread_id_cancel(bthread_id_t id) {
if (!meta) {
return EINVAL;
}
uint32_t* butex = meta->butex;
butil::atomic<int>* 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;
Expand All @@ -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<int>* 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;
Expand All @@ -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<int>* 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;
Expand All @@ -569,7 +591,7 @@ int bthread_id_unlock(bthread_id_t id) {
if (!meta) {
return EINVAL;
}
uint32_t* butex = meta->butex;
butil::atomic<int>* 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);
Expand All @@ -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;
Expand All @@ -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.
Expand All @@ -611,23 +634,23 @@ 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<int>* butex = meta->butex;
butil::atomic<int>* join_butex = meta->join_butex;
const uint32_t id_ver = bthread::get_version(id);
meta->mutex.lock();
if (!meta->has_version(id_ver)) {
meta->mutex.unlock();
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();
Expand Down Expand Up @@ -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<int>* 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) {
Expand Down
33 changes: 33 additions & 0 deletions test/bthread_id_unittest.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@
// under the License.

#include <iostream>
#include <atomic>
#include <thread>
#include <gtest/gtest.h>
#include "butil/time.h"
#include "butil/macros.h"
Expand Down Expand Up @@ -112,6 +114,37 @@ TEST(BthreadIdTest, join_before_destroy) {
ASSERT_EQ(1UL, non_null_ret);
}

struct JoinRaceArgs {
bthread_id_t id;
std::atomic<bool> entered;
};

void* join_race_waiter(void* arg) {
JoinRaceArgs* args = static_cast<JoinRaceArgs*>(arg);
args->entered.store(true, std::memory_order_release);

@darion-yaphet darion-yaphet Oct 29, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The waiting thread sets entered to true before calling bthread_id_join().

Therefore, the following sequence may occur:

  1. The waiting thread sets entered = true and is then suspended by the scheduler;
  2. The main thread observes the flag and completes the locking and destruction of the ID;
  3. The waiting thread resumes, calls join() and immediately returns.

In this case, the test passes, but there is no target competition between the reading of the butex_wait() status and the writing of the destruction status.

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;
Expand Down
Loading