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 8d4aa518 Fix macOS futex wake accounting with per-waiter notifications
(#3637)
8d4aa518 is described below
commit 8d4aa518352d7dc51e5dbf9f37c12dd1482e3e2c
Author: Bright Chen <[email protected]>
AuthorDate: Thu Oct 22 21:21:24 2026 +0800
Fix macOS futex wake accounting with per-waiter notifications (#3637)
* Fix macOS futex wake accounting with per-waiter notifications
* Fix ut
* Guard macOS futex waiter cleanup and fix timeout test diagnostic
---
src/bthread/sys_futex.cpp | 76 ++++++++++++++++++++----------
test/bthread_futex_unittest.cpp | 101 ++++++++++++++++++++++++++++++++++++++++
2 files changed, 152 insertions(+), 25 deletions(-)
diff --git a/src/bthread/sys_futex.cpp b/src/bthread/sys_futex.cpp
index 010ddcad..b6ebdd60 100644
--- a/src/bthread/sys_futex.cpp
+++ b/src/bthread/sys_futex.cpp
@@ -22,6 +22,7 @@
#include "bthread/sys_futex.h"
#include "butil/scoped_lock.h"
#include "butil/atomicops.h"
+#include "butil/containers/linked_list.h"
#include <pthread.h>
#include <unordered_map>
@@ -29,22 +30,35 @@
namespace bthread {
-class SimuFutex {
-public:
- SimuFutex() : counts(0)
- , ref(0) {
- pthread_mutex_init(&lock, nullptr);
+struct SimuFutexWaiter : public butil::LinkNode<SimuFutexWaiter> {
+ SimuFutexWaiter() : signaled(false) {
pthread_cond_init(&cond, nullptr);
}
+ ~SimuFutexWaiter() {
+ // The futex mutex is still held when the stack waiter is destroyed.
+ // Notified waiters were detached by futex_wake_private. Remove only
+ // waiters that leave on timeout or error.
+ if (!signaled) {
+ RemoveFromList();
+ }
+ pthread_cond_destroy(&cond);
+ }
+
+ pthread_cond_t cond;
+ bool signaled;
+};
+
+struct SimuFutex {
+ SimuFutex() : ref(0) {
+ pthread_mutex_init(&lock, nullptr);
+ }
~SimuFutex() {
pthread_mutex_destroy(&lock);
- pthread_cond_destroy(&cond);
}
-public:
pthread_mutex_t lock;
- pthread_cond_t cond;
- int32_t counts;
+ // Only waiters that have not been notified are in this queue.
+ butil::LinkedList<SimuFutexWaiter> waiters;
int32_t ref;
};
@@ -70,20 +84,28 @@ int futex_wait_private(void* addr1, int expected, const
timespec* timeout) {
{
std::unique_lock<pthread_mutex_t> mu1(simu_futex.lock);
if (static_cast<butil::atomic<int>*>(addr1)->load() == expected) {
- ++simu_futex.counts;
+ SimuFutexWaiter waiter;
+ simu_futex.waiters.Append(&waiter);
+ timespec timeout_abs = {};
if (timeout) {
- timespec timeout_abs = butil::timespec_from_now(*timeout);
- if ((rc = pthread_cond_timedwait(&simu_futex.cond,
&simu_futex.lock, &timeout_abs)) != 0) {
- errno = rc;
- rc = -1;
- }
- } else {
- if ((rc = pthread_cond_wait(&simu_futex.cond,
&simu_futex.lock)) != 0) {
- errno = rc;
- rc = -1;
+ timeout_abs = butil::timespec_from_now(*timeout);
+ }
+ do {
+ if (timeout) {
+ rc = pthread_cond_timedwait(
+ &waiter.cond, &simu_futex.lock, &timeout_abs);
+ } else {
+ rc = pthread_cond_wait(&waiter.cond, &simu_futex.lock);
}
+ } while (rc == 0 && !waiter.signaled);
+ // A wake may win the mutex while a timed-out waiter is reacquiring
+ // it. Honor that notification so it cannot be counted and then
lost.
+ if (waiter.signaled) {
+ rc = 0;
+ } else if (rc != 0) {
+ errno = rc;
+ rc = -1;
}
- --simu_futex.counts;
} else {
errno = EAGAIN;
rc = -1;
@@ -117,14 +139,18 @@ int futex_wake_private(void* addr1, int nwake) {
int rc = 0;
{
std::unique_lock<pthread_mutex_t> mu1(simu_futex.lock);
- nwake = (nwake < simu_futex.counts)? nwake: simu_futex.counts;
- for (int i = 0; i < nwake; ++i) {
- if ((rc = pthread_cond_signal(&simu_futex.cond)) != 0) {
+ while (nwakedup < nwake && !simu_futex.waiters.empty()) {
+ SimuFutexWaiter* waiter = simu_futex.waiters.head()->value();
+ if ((rc = pthread_cond_signal(&waiter->cond)) != 0) {
errno = rc;
break;
- } else {
- ++nwakedup;
}
+ // Commit the notification to this waiter before releasing the
+ // mutex. Later wakes cannot count it again while it is
rescheduled,
+ // and another waiter's timeout cannot consume its notification.
+ waiter->signaled = true;
+ waiter->RemoveFromList();
+ ++nwakedup;
}
}
diff --git a/test/bthread_futex_unittest.cpp b/test/bthread_futex_unittest.cpp
index ce32356e..d08982b6 100644
--- a/test/bthread_futex_unittest.cpp
+++ b/test/bthread_futex_unittest.cpp
@@ -19,6 +19,7 @@
#include <unistd.h>
#include <stdio.h>
#include <signal.h>
+#include <pthread.h>
#include <gtest/gtest.h>
#include <vector>
#include "butil/time.h"
@@ -115,6 +116,106 @@ TEST(FutexTest, futex_wake_before_wait) {
ASSERT_EQ(ETIMEDOUT, errno);
}
+const int kWakeCountIterations = 256;
+const int64_t kWakeCountTimeoutUs = 5 * 1000 * 1000L;
+const useconds_t kWakeCountRetryIntervalUs = 50;
+
+struct WakeCountWaiterArg {
+ butil::atomic<int>* word;
+ butil::atomic<bool>* finished;
+ int64_t deadline_us;
+ int rc;
+ int error;
+};
+
+void* wake_count_waiter(void* void_arg) {
+ WakeCountWaiterArg* arg = static_cast<WakeCountWaiterArg*>(void_arg);
+ const int64_t remaining_us = arg->deadline_us - butil::cpuwide_time_us();
+ const timespec timeout = butil::microseconds_to_timespec(
+ remaining_us > 0 ? remaining_us : 0);
+ arg->rc = bthread::futex_wait_private(arg->word, 0, &timeout);
+ arg->error = errno;
+ arg->finished->store(true, butil::memory_order_release);
+ return nullptr;
+}
+
+// A successful wake confirms that the waiter was registered. The waiter never
+// waits again, so every subsequent wake must return zero, even before it runs.
+// Repeat to exercise both orders of reacquiring the simulated futex's mutex.
+void check_wake_count_after_notification(bool expire_another_waiter) {
+ // Share one deadline across all iterations and the waiter itself, so slow
+ // scheduling cannot renew the timeout on every iteration.
+ const int64_t deadline_us = butil::cpuwide_time_us() + kWakeCountTimeoutUs;
+ for (int i = 0; i < kWakeCountIterations; ++i) {
+ ASSERT_LT(butil::cpuwide_time_us(), deadline_us)
+ << "Timed out after " << i << " iterations";
+ butil::atomic<int> word(0);
+ butil::atomic<bool> finished(false);
+ WakeCountWaiterArg arg = { &word, &finished, deadline_us, -1, 0 };
+ pthread_t waiter;
+ ASSERT_EQ(0, pthread_create(&waiter, nullptr, wake_count_waiter,
&arg));
+
+ int first_wake = 0;
+ // A readiness flag published before futex_wait does not prove that the
+ // waiter is queued. Use the wake result, with a pause between retries.
+ while (butil::cpuwide_time_us() < deadline_us) {
+ first_wake = bthread::futex_wake_private(&word, 1);
+ if (first_wake != 0 || finished.load(butil::memory_order_acquire))
{
+ break;
+ }
+ usleep(kWakeCountRetryIntervalUs);
+ }
+
+ int expired_rc = 0;
+ int expired_errno = 0;
+ int second_wake = 0;
+ if (first_wake == 1) {
+ if (expire_another_waiter) {
+ const timespec timeout = { 0, 0 };
+ expired_rc = bthread::futex_wait_private(&word, 0, &timeout);
+ expired_errno = errno;
+ }
+ second_wake = bthread::futex_wake_private(&word, 1);
+ }
+ // The waiter has a timeout even if no wake succeeds. Join before
+ // checking results so a failed assertion cannot leave it referring to
+ // this iteration's stack objects.
+ ASSERT_EQ(0, pthread_join(waiter, nullptr));
+ ASSERT_EQ(1, first_wake)
+ << "No waiter was notified before the deadline: iteration=" << i
+ << " waiter_rc=" << arg.rc << " errno=" << arg.error;
+ ASSERT_EQ(0, arg.rc) << "errno=" << arg.error;
+ if (expire_another_waiter) {
+ ASSERT_EQ(-1, expired_rc);
+ ASSERT_EQ(ETIMEDOUT, expired_errno);
+ }
+ ASSERT_EQ(0, second_wake) << "iteration=" << i;
+ }
+}
+
+TEST(FutexTest, notified_waiter_is_not_counted_twice) {
+ check_wake_count_after_notification(false);
+}
+
+TEST(FutexTest, timeout_does_not_consume_another_waiters_notification) {
+ check_wake_count_after_notification(true);
+}
+
+TEST(FutexTest, timed_out_waiter_is_removed) {
+ butil::atomic<int> word(0);
+ const timespec timeout = { 0, 0 };
+ ASSERT_EQ(-1, bthread::futex_wait_private(&word, 0, &timeout));
+ ASSERT_EQ(ETIMEDOUT, errno);
+ ASSERT_EQ(0, bthread::futex_wake_private(&word, INT_MAX));
+}
+
+TEST(FutexTest, mismatched_value_does_not_queue_a_waiter) {
+ butil::atomic<int> word(1);
+ ASSERT_EQ(-1, bthread::futex_wait_private(&word, 0, nullptr));
+ ASSERT_EQ(EAGAIN, errno);
+ ASSERT_EQ(0, bthread::futex_wake_private(&word, INT_MAX));
+}
+
struct DummyWaiterArg {
butil::atomic<int>* lock;
butil::atomic<int>* registered;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
