This is an automated email from the ASF dual-hosted git repository.
wasphin 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 b009adb42 Fix semaphore publication with release ordering (#3697)
b009adb42 is described below
commit b009adb42af89dd6a620c3fa41239fc2e2522e79
Author: Bright Chen <[email protected]>
AuthorDate: Sun Nov 1 22:01:12 2026 +0800
Fix semaphore publication with release ordering (#3697)
---
src/bthread/semaphore.cpp | 14 +++---
test/bthread_semaphore_unittest.cpp | 98 +++++++++++++++++++++++++++++++++++++
2 files changed, 106 insertions(+), 6 deletions(-)
diff --git a/src/bthread/semaphore.cpp b/src/bthread/semaphore.cpp
index 6bdf1a8a5..eaf3d69c4 100644
--- a/src/bthread/semaphore.cpp
+++ b/src/bthread/semaphore.cpp
@@ -110,16 +110,18 @@ static int bthread_sem_wait_impl(bthread_sem_t* sem,
const struct timespec* abst
static inline int bthread_sem_post(bthread_sem_t* sem, size_t num) {
if (num > 0) {
+ // Publish preceding writes to successful acquire-CAS waiters,
including
+ // those that consume permits without entering butex_wait().
((butil::atomic<unsigned>*)sem->butex)
- ->fetch_add(num, butil::memory_order_relaxed);
- const size_t sampling_range = nullptr != bthread::g_cp &&
sem->enable_csite ?
+ ->fetch_add(num, butil::memory_order_release);
+ size_t sampling_range = nullptr != bthread::g_cp && sem->enable_csite ?
bvar::is_collectable(&bthread::g_cp_sl) :
bvar::INVALID_SAMPLING_RANGE;
- const int64_t start_ns = bvar::is_sampling_range_valid(sampling_range)
?
- butil::cpuwide_time_ns() : -1;
+ int64_t start_ns = bvar::is_sampling_range_valid(sampling_range) ?
+ butil::cpuwide_time_ns() : -1;
bthread::butex_wake_n(sem->butex, num);
if (start_ns > 0) {
- const int64_t end_ns = butil::cpuwide_time_ns();
- const bthread_contention_site_t csite{end_ns - start_ns,
sampling_range};
+ int64_t end_ns = butil::cpuwide_time_ns();
+ bthread_contention_site_t csite{end_ns - start_ns, sampling_range};
bthread::submit_contention(csite, end_ns);
}
}
diff --git a/test/bthread_semaphore_unittest.cpp
b/test/bthread_semaphore_unittest.cpp
index 4e37985d8..88856dc14 100644
--- a/test/bthread_semaphore_unittest.cpp
+++ b/test/bthread_semaphore_unittest.cpp
@@ -16,6 +16,7 @@
// under the License.
#include <algorithm>
+#include <sched.h>
#include <gtest/gtest.h>
#include <bthread/bthread.h>
#include <bthread/butex.h>
@@ -134,6 +135,103 @@ TEST(SemaphoreTest, trywait) {
ASSERT_EQ(0, bthread_sem_destroy(&sem));
}
+typedef int (*SemWaitOp)(bthread_sem_t*);
+
+int sem_timedwait_fast(bthread_sem_t* sem) {
+ // An available permit must be consumed even with an expired deadline.
+ const timespec deadline = { -2, 0 };
+ return bthread_sem_timedwait(sem, &deadline);
+}
+
+struct SemPublicationArgs {
+ bthread_sem_t* sem;
+ SemWaitOp wait;
+ const int* payload;
+ int observed[16];
+ int rc;
+};
+
+void* sem_publication_waiter(void* arg) {
+ auto args = static_cast<SemPublicationArgs*>(arg);
+ auto value = reinterpret_cast<butil::atomic<unsigned>*>(args->sem->butex);
+ const int64_t deadline = butil::gettimeofday_us() + 10 * 1000 * 1000;
+ // Observe available permits without acquiring the payload. Each thread
+ // consumes only once, and the post supplies enough permits for all of
them.
+ // This keeps consumers off the blocking path and its additional locks.
+ while (value->load(butil::memory_order_relaxed) == 0) {
+ if (butil::gettimeofday_us() >= deadline) {
+ args->rc = ETIMEDOUT;
+ return nullptr;
+ }
+ sched_yield();
+ }
+ args->rc = args->wait(args->sem);
+ if (args->rc == 0) {
+ // Snapshot non-atomic data before join or any assertion can introduce
+ // synchronization. Only the successful semaphore wait may acquire it.
+ std::copy(args->payload, args->payload + ARRAY_SIZE(args->observed),
+ args->observed);
+ }
+ return nullptr;
+}
+
+void check_post_publication(bool batch) {
+ const SemWaitOp wait_ops[] = {
+ bthread_sem_wait, sem_timedwait_fast, bthread_sem_trywait
+ };
+ const size_t nwaiters = batch ? ARRAY_SIZE(wait_ops) : 1;
+ for (int round = 0; round < 1000; ++round) {
+ SCOPED_TRACE(round);
+ bthread_sem_t sem;
+ ASSERT_EQ(0, bthread_sem_init(&sem, 0));
+ int payload[16] = {};
+ SemPublicationArgs args[ARRAY_SIZE(wait_ops)] = {};
+ pthread_t threads[ARRAY_SIZE(wait_ops)];
+ size_t started = 0;
+ int create_rc = 0;
+ // Start every consumer before writing the payload, so pthread_create
+ // cannot publish it. Use pthreads to avoid bthread scheduler handoffs.
+ for (; started < nwaiters; ++started) {
+ args[started].sem = &sem;
+ args[started].wait = wait_ops[(round + started) %
ARRAY_SIZE(wait_ops)];
+ args[started].payload = payload;
+ create_rc = pthread_create(&threads[started], nullptr,
+ sem_publication_waiter, &args[started]);
+ if (create_rc != 0) {
+ break;
+ }
+ }
+ for (size_t i = 0; i < ARRAY_SIZE(payload); ++i) {
+ payload[i] = round + 1 + static_cast<int>(i);
+ }
+ // For a batch, later consumers must acquire the original publication
+ // through the release sequence extended by earlier consumers' CASes.
+ const int post_rc = batch ? bthread_sem_post_n(&sem, nwaiters)
+ : bthread_sem_post(&sem);
+ for (size_t i = 0; i < started; ++i) {
+ EXPECT_EQ(0, pthread_join(threads[i], nullptr));
+ }
+ EXPECT_EQ(0, bthread_sem_destroy(&sem));
+ ASSERT_EQ(0, create_rc);
+ ASSERT_EQ(0, post_rc);
+ for (size_t i = 0; i < nwaiters; ++i) {
+ ASSERT_EQ(0, args[i].rc) << "waiter=" << i;
+ for (size_t j = 0; j < ARRAY_SIZE(payload); ++j) {
+ ASSERT_EQ(payload[j], args[i].observed[j])
+ << "waiter=" << i << " index=" << j;
+ }
+ }
+ }
+}
+
+TEST(SemaphoreTest, post_publishes_data_to_fast_waiters) {
+ check_post_publication(false);
+}
+
+TEST(SemaphoreTest, post_n_publishes_data_to_fast_waiters) {
+ check_post_publication(true);
+}
+
struct SemButexWaitArgs {
bthread_sem_t* sem;
int expected_value;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
