From de62260e32ebb051562d3b187376d0f8fc762ea4 Mon Sep 17 00:00:00 2001 From: chenBright Date: Sat, 15 Aug 2026 13:34:56 +0800 Subject: [PATCH] Fix bthread_join memory visibility with paired release/acquire on version_butex --- src/bthread/task_group.cpp | 27 +++++++++++++++++++-------- src/bthread/task_meta.h | 13 ++++++++++--- 2 files changed, 29 insertions(+), 11 deletions(-) diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index 777d7514a3..ec822958ca 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -520,9 +520,18 @@ void TaskGroup::task_runner(intptr_t skip_remained) { #ifdef BRPC_BTHREAD_TRACER tracing = TaskTracer::set_end_status_unsafe(m); #endif // BRPC_BTHREAD_TRACER - if (0 == ++*m->version_butex) { - ++*m->version_butex; + // Bump the version with a release store so that it pairs with the + // acquire load in TaskGroup::join(): all memory writes made by this + // bthread become visible to the joining thread. Accessing + // version_butex atomically also avoids a data race with the read in + // join(). This path is the only writer of version_butex at runtime. + auto* version = reinterpret_cast*>(m->version_butex); + uint32_t next_version = static_cast( + version->load(butil::memory_order_relaxed)) + 1; + if (0 == next_version) { + ++next_version; } + version->store(static_cast(next_version), butil::memory_order_release); } butex_wake_except(m->version_butex, 0); @@ -709,16 +718,18 @@ int TaskGroup::join(bthread_t tid, void** return_value) { return EINVAL; } const uint32_t expected_version = get_version(tid); - while (*m->version_butex == expected_version) { - if (butex_wait(m->version_butex, expected_version, NULL) < 0 && + // Acquire load pairs with the release store performed when the joined + // bthread ends (see the version bump above), ensuring all of its memory + // writes are visible after join() returns. This matches the semantic + // guarantee provided by pthread_join() across supported architectures. + auto* version = reinterpret_cast*>(m->version_butex); + const int expected_version_int = static_cast(expected_version); + while (version->load(butil::memory_order_acquire) == expected_version_int) { + if (butex_wait(m->version_butex, expected_version_int, NULL) < 0 && errno != EWOULDBLOCK && errno != EINTR) { return errno; } } - // Ensure all memory writes made by the joined bthread are visible to - // the joining thread after join returns. This matches the semantic - // guarantee provided by pthread_join() across supported architectures. - butil::atomic_thread_fence(butil::memory_order_acquire); if (return_value) { *return_value = NULL; } diff --git a/src/bthread/task_meta.h b/src/bthread/task_meta.h index 7c9e63790e..a00d0da2fd 100644 --- a/src/bthread/task_meta.h +++ b/src/bthread/task_meta.h @@ -85,10 +85,17 @@ struct TaskMeta { // Scheduling of the thread can be delayed. bool about_to_quit{false}; - // [Not Reset] guarantee visibility of version_butex. + // [Not Reset] Serializes the version bump at bthread end (in task_runner) + // with accessors that validate the version before touching other fields of + // this TaskMeta (get_attr/set_stopped/interrupt/set_butex_waiter/...). It + // makes their "check version then read/write field" sequence atomic w.r.t. + // the bump, so they never operate on a slot that got recycled in between. pthread_spinlock_t version_lock{}; - - // [Not Reset] only modified by one bthread at any time, no need to be atomic + + // [Not Reset] Backed by a butex (internally `butil::atomic`). The version + // bump at bthread end is published with a release store, and join() observes + // it with an acquire load so the joined bthread's prior writes are visible + // after join() returns. uint32_t* version_butex{NULL}; // The identifier. It does not have to be here, however many code is