This is an automated email from the ASF dual-hosted git repository.
wwbmmm 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 6715a9230 Fix semaphore wakeup count to match posted permits (#3673)
6715a9230 is described below
commit 6715a92307d94460de059c060c70755622db6a18
Author: Bright Chen <[email protected]>
AuthorDate: Sat Oct 31 14:46:31 2026 +0800
Fix semaphore wakeup count to match posted permits (#3673)
* Fix semaphore wakeup count to match posted permits
* Fix semaphore wake-count test race by using bthread-only waiters
---
src/bthread/semaphore.cpp | 6 +--
test/bthread_semaphore_unittest.cpp | 90 +++++++++++++++++++++++++++++++++++++
2 files changed, 93 insertions(+), 3 deletions(-)
diff --git a/src/bthread/semaphore.cpp b/src/bthread/semaphore.cpp
index 7b9c22d55..6bdf1a8a5 100644
--- a/src/bthread/semaphore.cpp
+++ b/src/bthread/semaphore.cpp
@@ -110,13 +110,13 @@ 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) {
- unsigned n = ((butil::atomic<unsigned>*)sem->butex)
+ ((butil::atomic<unsigned>*)sem->butex)
->fetch_add(num, butil::memory_order_relaxed);
const 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;
- bthread::butex_wake_n(sem->butex, n);
+ 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};
@@ -170,4 +170,4 @@ int bthread_sem_post_n(bthread_sem_t* sem, size_t n) {
return bthread::bthread_sem_post(sem, n);
}
-__END_DECLS
\ No newline at end of file
+__END_DECLS
diff --git a/test/bthread_semaphore_unittest.cpp
b/test/bthread_semaphore_unittest.cpp
index d0391dcfc..4e37985d8 100644
--- a/test/bthread_semaphore_unittest.cpp
+++ b/test/bthread_semaphore_unittest.cpp
@@ -15,8 +15,13 @@
// specific language governing permissions and limitations
// under the License.
+#include <algorithm>
#include <gtest/gtest.h>
#include <bthread/bthread.h>
+#include <bthread/butex.h>
+#include <bthread/countdown_event.h>
+#include <butil/atomicops.h>
+#include <butil/time.h>
namespace {
@@ -129,6 +134,91 @@ TEST(SemaphoreTest, trywait) {
ASSERT_EQ(0, bthread_sem_destroy(&sem));
}
+struct SemButexWaitArgs {
+ bthread_sem_t* sem;
+ int expected_value;
+ bthread::CountdownEvent* ready;
+ butil::atomic<int> nwoken{0};
+};
+
+void* sem_butex_waiter(void* arg) {
+ auto args = static_cast<SemButexWaitArgs*>(arg);
+ args->ready->signal();
+ const timespec deadline = butil::seconds_from_now(10);
+ // Wait only once so that excess wakeups cannot be hidden by sem_wait's
+ // retry loop. Do not consume the posted permits.
+ const int rc = bthread::butex_wait(args->sem->butex, args->expected_value,
+ &deadline);
+ if (rc == 0) {
+ args->nwoken.fetch_add(1, butil::memory_order_relaxed);
+ } else {
+ // The post may race with a waiter that has not yet been queued.
+ EXPECT_EQ(EWOULDBLOCK, errno);
+ }
+ return nullptr;
+}
+
+void check_post_wakeup_count(unsigned initial_value, size_t num, bool batch) {
+ SCOPED_TRACE(initial_value);
+ const int N = 16;
+ // Retry if too few waiters were queued before posting. This avoids relying
+ // on a fixed sleep to establish that the waiters have entered butex_wait.
+ for (int attempt = 0; attempt < 10; ++attempt) {
+ bthread_sem_t sem;
+ ASSERT_EQ(0, bthread_sem_init(&sem, initial_value));
+ bthread::CountdownEvent ready(N);
+ SemButexWaitArgs args{&sem, static_cast<int>(initial_value), &ready};
+ // A queued pthread woken before entering futex_wait can still return
+ // EWOULDBLOCK, which is indistinguishable from a late arrival. Use
only
+ // bthreads so successful waits account for every counted wakeup.
+ bthread_t bthreads[N];
+ for (int i = 0; i < N; ++i) {
+ ASSERT_EQ(0, bthread_start_background(
+ &bthreads[i], nullptr, sem_butex_waiter, &args));
+ }
+ EXPECT_EQ(0, ready.timed_wait(butil::seconds_from_now(10)));
+
+ EXPECT_EQ(0, batch ? bthread_sem_post_n(&sem, num)
+ : bthread_sem_post(&sem));
+ auto value = reinterpret_cast<butil::atomic<unsigned>*>(sem.butex);
+ EXPECT_EQ(initial_value + num,
value->load(butil::memory_order_relaxed));
+ // Also change the value for post_n(0), so late arrivals cannot enqueue
+ // after cleanup and block forever.
+ value->fetch_add(1, butil::memory_order_relaxed);
+ const int remaining = bthread::butex_wake_all(sem.butex);
+ for (int i = 0; i < N; ++i) {
+ EXPECT_EQ(0, bthread_join(bthreads[i], nullptr));
+ }
+ const int queued = args.nwoken.load(butil::memory_order_relaxed);
+ // Successful waits include both post and cleanup wakeups; late
+ // arrivals that returned EWOULDBLOCK are excluded.
+ EXPECT_EQ(std::min(num, static_cast<size_t>(queued)),
+ static_cast<size_t>(queued - remaining));
+ ASSERT_EQ(0, bthread_sem_destroy(&sem));
+ // More waiters than permits are needed to detect excessive wakeups.
+ if (static_cast<size_t>(queued) > num) {
+ return;
+ }
+ }
+ FAIL() << "Too few waiters were queued to check the wakeup limit";
+}
+
+TEST(SemaphoreTest, post_wakes_one_waiter) {
+ check_post_wakeup_count(0, 1, false);
+ check_post_wakeup_count(4, 1, false);
+}
+
+TEST(SemaphoreTest, post_n_wakes_requested_waiters) {
+ check_post_wakeup_count(0, 3, true);
+ check_post_wakeup_count(1, 3, true);
+ check_post_wakeup_count(8, 3, true);
+}
+
+TEST(SemaphoreTest, post_n_zero_does_not_wake_waiters) {
+ check_post_wakeup_count(0, 0, true);
+ check_post_wakeup_count(5, 0, true);
+}
+
bool g_stopped = false;
void wait_op(bthread_sem_t* sem, int64_t sleep_us) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
