Fix semaphore wakeup count to match posted permits

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
+__END_DECLS

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) {