Copilot commented on code in PR #3637:
URL: https://github.com/apache/brpc/pull/3637#discussion_r4336435817
##########
src/bthread/sys_futex.cpp:
##########
@@ -22,29 +22,41 @@
#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>
#if defined(OS_MACOSX)
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.
+ // A notified waiter is already detached; otherwise remove it on
timeout
+ // or error.
+ RemoveFromList();
+ pthread_cond_destroy(&cond);
+ }
Review Comment:
This relies on `RemoveFromList()` being safe to call even if the node has
already been removed in `futex_wake_private` (i.e., double-remove is a no-op).
To avoid depending on that implicit contract, consider guarding the destructor
removal with an explicit “enqueued”/“linked” check (if provided by `LinkNode`),
or maintaining a small boolean flag that tracks whether the waiter is currently
in `simu_futex.waiters`.
##########
test/bthread_futex_unittest.cpp:
##########
@@ -115,6 +116,113 @@ 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) {
+ int notified = 0;
+ // 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) {
+ ++notified;
+ 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);
+ } else {
+ // Rescue a waiter that had not registered before the deadline.
+ word.store(1);
+ bthread::futex_wake_private(&word, INT_MAX);
+ }
Review Comment:
The “rescue” path can force the waiter to return `-1/EAGAIN` (by storing
`word=1` before the waiter reaches `futex_wait_private`), but the test always
asserts `arg.rc == 0`. This makes the test potentially flaky under slow
scheduling. Suggested fix: don’t mutate `word` in the rescue path; instead
treat `first_wake != 1` as a test failure with a clear assertion/message (or
keep retrying until either `first_wake==1` or `finished==true`, and then assert
the expected outcome based on `arg.rc/arg.error`).
##########
test/bthread_futex_unittest.cpp:
##########
@@ -115,6 +116,113 @@ 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) {
+ int notified = 0;
+ // 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) {
+ ++notified;
+ 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);
+ } else {
+ // Rescue a waiter that had not registered before the deadline.
+ word.store(1);
+ bthread::futex_wake_private(&word, INT_MAX);
Review Comment:
`INT_MAX` is used in the new tests but this file doesn’t include a header
that guarantees its definition. Add `#include <climits>` (or switch to
`std::numeric_limits<int>::max()` with `<limits>`) so the test compiles
portably.
##########
src/bthread/sys_futex.cpp:
##########
@@ -117,14 +137,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();
Review Comment:
This relies on `RemoveFromList()` being safe to call even if the node has
already been removed in `futex_wake_private` (i.e., double-remove is a no-op).
To avoid depending on that implicit contract, consider guarding the destructor
removal with an explicit “enqueued”/“linked” check (if provided by `LinkNode`),
or maintaining a small boolean flag that tracks whether the waiter is currently
in `simu_futex.waiters`.
##########
test/bthread_futex_unittest.cpp:
##########
@@ -115,6 +116,113 @@ 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) {
+ int notified = 0;
+ // 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) {
+ ++notified;
+ 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);
+ } else {
+ // Rescue a waiter that had not registered before the deadline.
+ word.store(1);
+ bthread::futex_wake_private(&word, INT_MAX);
+ }
+ // Join before checking results so a failed assertion cannot leave a
+ // waiter referring to this iteration's stack objects.
+ ASSERT_EQ(0, pthread_join(waiter, nullptr));
+ ASSERT_GE(first_wake, 0);
+ ASSERT_LE(first_wake, 1);
+ ASSERT_EQ(0, arg.rc) << "errno=" << arg.error;
Review Comment:
The “rescue” path can force the waiter to return `-1/EAGAIN` (by storing
`word=1` before the waiter reaches `futex_wait_private`), but the test always
asserts `arg.rc == 0`. This makes the test potentially flaky under slow
scheduling. Suggested fix: don’t mutate `word` in the rescue path; instead
treat `first_wake != 1` as a test failure with a clear assertion/message (or
keep retrying until either `first_wake==1` or `finished==true`, and then assert
the expected outcome based on `arg.rc/arg.error`).
##########
test/bthread_futex_unittest.cpp:
##########
@@ -115,6 +116,113 @@ 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) {
+ int notified = 0;
+ // 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) {
+ ++notified;
+ 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);
+ } else {
+ // Rescue a waiter that had not registered before the deadline.
+ word.store(1);
+ bthread::futex_wake_private(&word, INT_MAX);
+ }
+ // Join before checking results so a failed assertion cannot leave a
+ // waiter referring to this iteration's stack objects.
+ ASSERT_EQ(0, pthread_join(waiter, nullptr));
+ ASSERT_GE(first_wake, 0);
+ ASSERT_LE(first_wake, 1);
+ ASSERT_EQ(0, arg.rc) << "errno=" << arg.error;
+ if (first_wake == 1) {
+ if (expire_another_waiter) {
+ ASSERT_EQ(-1, expired_rc);
+ ASSERT_EQ(ETIMEDOUT, expired_errno);
+ }
+ ASSERT_EQ(0, second_wake) << "iteration=" << i;
+ }
+ }
+ ASSERT_GT(notified, 0);
+}
+
+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));
Review Comment:
`INT_MAX` is used in the new tests but this file doesn’t include a header
that guarantees its definition. Add `#include <climits>` (or switch to
`std::numeric_limits<int>::max()` with `<limits>`) so the test compiles
portably.
##########
test/bthread_futex_unittest.cpp:
##########
@@ -115,6 +116,113 @@ 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) {
+ int notified = 0;
+ // 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) {
+ ++notified;
+ 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);
+ } else {
+ // Rescue a waiter that had not registered before the deadline.
+ word.store(1);
+ bthread::futex_wake_private(&word, INT_MAX);
+ }
+ // Join before checking results so a failed assertion cannot leave a
+ // waiter referring to this iteration's stack objects.
+ ASSERT_EQ(0, pthread_join(waiter, nullptr));
+ ASSERT_GE(first_wake, 0);
+ ASSERT_LE(first_wake, 1);
+ ASSERT_EQ(0, arg.rc) << "errno=" << arg.error;
+ if (first_wake == 1) {
+ if (expire_another_waiter) {
+ ASSERT_EQ(-1, expired_rc);
+ ASSERT_EQ(ETIMEDOUT, expired_errno);
+ }
+ ASSERT_EQ(0, second_wake) << "iteration=" << i;
+ }
+ }
+ ASSERT_GT(notified, 0);
+}
+
+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));
Review Comment:
`INT_MAX` is used in the new tests but this file doesn’t include a header
that guarantees its definition. Add `#include <climits>` (or switch to
`std::numeric_limits<int>::max()` with `<limits>`) so the test compiles
portably.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
