diff --git a/src/bthread/semaphore.cpp b/src/bthread/semaphore.cpp index 7b9c22d555..6bdf1a8a5c 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*)sem->butex) + ((butil::atomic*)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 d0391dcfcd..4e37985d84 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 #include #include +#include +#include +#include +#include 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 nwoken{0}; +}; + +void* sem_butex_waiter(void* arg) { + auto args = static_cast(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(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*>(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(queued)), + static_cast(queued - remaining)); + ASSERT_EQ(0, bthread_sem_destroy(&sem)); + // More waiters than permits are needed to detect excessive wakeups. + if (static_cast(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) {