chenBright commented on code in PR #3637:
URL: https://github.com/apache/brpc/pull/3637#discussion_r4336638663
##########
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:
butil::LinkNode::RemoveFromList() resets both links to this, so calling it
again on a detached node is safe. I've added a guard using the existing
signaled flag to make the lifecycle explicit and avoid the redundant removal
##########
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:
Fixed.
--
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]
