Stabilize timing-sensitive bthread tests and fix the butex interruption race

src/bthread/butex.cpp

@@ -729,9 +729,19 @@ static int butex_wait_from_pthread(TaskGroup* g, Butex* b, int expected_value,
         BT_LOOP_WHEN(task->current_waiter.exchange(
                          nullptr, butil::memory_order_acquire) == nullptr,
                      30/*nops before sched_yield*/);
+        // TaskGroup::interrupt() updates `interrupted` under `version_lock`.
+        // Keep the check, clear, and signal classification together so this
+        // epilogue neither races with nor consumes a concurrent interruption.
+        BAIDU_SCOPED_LOCK(task->version_lock);
         if (task->interrupted) {
             task->interrupted = false;
-            if (rc == 0) {
+            // If interrupted after enqueueing but before futex_wait_private,
+            // pw.sig is already signalled and futex_wait_private may report
+            // EWOULDBLOCK. This is an interruption, not a value mismatch on
+            // the user's butex. Preserve other errors (notably ETIMEDOUT).
+            if (rc == 0 || (errno == EWOULDBLOCK &&
+                            pw.sig.load(butil::memory_order_acquire) ==
+                            PTHREAD_SIGNALLED)) {
                 errno = EINTR;
                 return -1;
             }

test/brpc_socket_unittest.cpp

@@ -1819,15 +1819,20 @@ TEST_F(SocketTest, socket_buffer_options_before_accept) {
     brpc::SocketId id = brpc::INVALID_SOCKET_ID;
     ASSERT_EQ(0, brpc::Socket::Create(options, &id));
 
-    const int64_t start_time = butil::cpuwide_time_us();
-    while (messenger->ConnectionCount() < 1) {
+    // ConnectionCount includes reserved slots before the accepted sockets
+    // have been inserted into the connection map. Wait for publication, not
+    // just for the connection slot to be acquired.
+    std::vector<brpc::SocketId> connections;
+    int64_t deadline = butil::cpuwide_time_us() + 5000000L;
+    for (;;) {
+        messenger->ListConnections(&connections);
+        if (!connections.empty()) {
+            break;
+        }
+        ASSERT_LT(butil::cpuwide_time_us(), deadline)
+            << "Timed out waiting for the accepted socket to be published";
         bthread_usleep(1000);
-        ASSERT_LT(butil::cpuwide_time_us(), start_time + 1000000L)
-            << "Too long!";
     }
-
-    std::vector<brpc::SocketId> connections;
-    messenger->ListConnections(&connections);
     ASSERT_EQ(1ul, connections.size());
 
     {

test/bthread_butex_multi_tag_unittest.cpp

@@ -21,6 +21,8 @@
 #include "bthread/condition_variable.h"
 #include "bthread/countdown_event.h"
 #include "bthread/mutex.h"
+#include "bthread/task_group.h"
+#include "bthread/task_meta.h"
 
 DECLARE_int32(task_group_ntags);
 
@@ -33,7 +35,18 @@ int main(int argc, char* argv[]) {
 
 namespace {
 
-std::vector<bthread_tag_t> butex_wake_return(2, 0);
+// Observe registration, not just arrival immediately before a blocking call.
+void WaitForWaiter(bthread_t tid) {
+    auto* meta = bthread::TaskGroup::address_meta(tid);
+    int64_t deadline = butil::cpuwide_time_us() + 5000000L;
+    while (meta->current_waiter.load(butil::memory_order_acquire) == nullptr &&
+           butil::cpuwide_time_us() < deadline) {
+        bthread_usleep(1000);
+    }
+    ASSERT_NE(nullptr, meta->current_waiter.load(butil::memory_order_acquire));
+}
+
+std::vector<bthread_tag_t> butex_wake_return;
 
 void* butex_wake_func(void* arg) {
     auto mutex = static_cast<bthread::Mutex*>(arg);
@@ -45,19 +58,23 @@ void* butex_wake_func(void* arg) {
 }
 
 TEST(BthreadButexMultiTest, butex_wake) {
+    butex_wake_return.clear();
     bthread::Mutex mutex;
     mutex.lock();
     bthread_t tid1;
     bthread_attr_t attr = BTHREAD_ATTR_NORMAL;
     attr.tag = 1;
-    bthread_start_urgent(&tid1, &attr, butex_wake_func, &mutex);
+    ASSERT_EQ(0, bthread_start_urgent(&tid1, &attr, butex_wake_func, &mutex));
+    ASSERT_NO_FATAL_FAILURE(WaitForWaiter(tid1));
     mutex.unlock();
     bthread_join(tid1, nullptr);
-    ASSERT_EQ(butex_wake_return[0], butex_wake_return[1]);
+    ASSERT_EQ(2ul, butex_wake_return.size());
+    ASSERT_EQ(1, butex_wake_return[0]);
+    ASSERT_EQ(1, butex_wake_return[1]);
 }
 
-std::vector<bthread_tag_t> butex_wake_all_return1(2, 0);
-std::vector<bthread_tag_t> butex_wake_all_return2(2, 0);
+std::vector<bthread_tag_t> butex_wake_all_return1;
+std::vector<bthread_tag_t> butex_wake_all_return2;
 
 struct ButexWakeAllArgs {
     bthread::CountdownEvent* ev;
@@ -87,6 +104,8 @@ void* butex_wake_all_func2(void* arg) {
 }
 
 TEST(BthreadButexMultiTest, butex_wake_all) {
+    butex_wake_all_return1.clear();
+    butex_wake_all_return2.clear();
     bthread::CountdownEvent ev(2);
     bthread::CountdownEvent ack(2);
     ButexWakeAllArgs args{&ev, &ack};
@@ -98,15 +117,21 @@ TEST(BthreadButexMultiTest, butex_wake_all) {
     attr2.tag = 2;
     bthread_start_background(&tid2, &attr2, butex_wake_all_func2, &args);
     ack.wait();
+    ASSERT_NO_FATAL_FAILURE(WaitForWaiter(tid1));
+    ASSERT_NO_FATAL_FAILURE(WaitForWaiter(tid2));
     ev.signal(2);
     bthread_join(tid1, nullptr);
     bthread_join(tid2, nullptr);
-    ASSERT_EQ(butex_wake_all_return1[0], butex_wake_all_return1[1]);
-    ASSERT_EQ(butex_wake_all_return2[0], butex_wake_all_return2[1]);
+    ASSERT_EQ(2ul, butex_wake_all_return1.size());
+    ASSERT_EQ(2ul, butex_wake_all_return2.size());
+    ASSERT_EQ(1, butex_wake_all_return1[0]);
+    ASSERT_EQ(1, butex_wake_all_return1[1]);
+    ASSERT_EQ(2, butex_wake_all_return2[0]);
+    ASSERT_EQ(2, butex_wake_all_return2[1]);
 }
 
-std::vector<bthread_tag_t> butex_requeue_return1(2, 0);
-std::vector<bthread_tag_t> butex_requeue_return2(2, 0);
+std::vector<bthread_tag_t> butex_requeue_return1;
+std::vector<bthread_tag_t> butex_requeue_return2;
 
 struct ButexRequeueArgs {
     bthread::Mutex* mutex;
@@ -119,11 +144,11 @@ void* butex_requeue_func1(void* arg) {
     auto mutex = p->mutex;
     auto cond = p->cond;
     auto ack = p->ack;
-    butex_wake_all_return1.push_back(bthread_self_tag());
+    butex_requeue_return1.push_back(bthread_self_tag());
     std::unique_lock<bthread::Mutex> lk(*mutex);
     ack->signal();
     cond->wait(lk);
-    butex_wake_all_return1.push_back(bthread_self_tag());
+    butex_requeue_return1.push_back(bthread_self_tag());
     return nullptr;
 }
 
@@ -132,15 +157,17 @@ void* butex_requeue_func2(void* arg) {
     auto mutex = p->mutex;
     auto cond = p->cond;
     auto ack = p->ack;
-    butex_wake_all_return2.push_back(bthread_self_tag());
+    butex_requeue_return2.push_back(bthread_self_tag());
     std::unique_lock<bthread::Mutex> lk(*mutex);
     ack->signal();
     cond->wait(lk);
-    butex_wake_all_return2.push_back(bthread_self_tag());
+    butex_requeue_return2.push_back(bthread_self_tag());
     return nullptr;
 }
 
 TEST(BthreadButexMultiTest, butex_requeue) {
+    butex_requeue_return1.clear();
+    butex_requeue_return2.clear();
     bthread::Mutex mutex;
     bthread::ConditionVariable cond;
     bthread::CountdownEvent ack(2);
@@ -154,6 +181,8 @@ TEST(BthreadButexMultiTest, butex_requeue) {
     attr2.tag = 2;
     bthread_start_background(&tid2, &attr2, butex_requeue_func2, &args);
     ack.wait();
+    ASSERT_NO_FATAL_FAILURE(WaitForWaiter(tid1));
+    ASSERT_NO_FATAL_FAILURE(WaitForWaiter(tid2));
     {
         std::unique_lock<bthread::Mutex> lk(mutex);
         cond.notify_all();
@@ -164,8 +193,12 @@ TEST(BthreadButexMultiTest, butex_requeue) {
     }
     bthread_join(tid1, nullptr);
     bthread_join(tid2, nullptr);
-    ASSERT_EQ(butex_wake_all_return1[0], butex_wake_all_return1[1]);
-    ASSERT_EQ(butex_wake_all_return2[0], butex_wake_all_return2[1]);
+    ASSERT_EQ(2ul, butex_requeue_return1.size());
+    ASSERT_EQ(2ul, butex_requeue_return2.size());
+    ASSERT_EQ(1, butex_requeue_return1[0]);
+    ASSERT_EQ(1, butex_requeue_return1[1]);
+    ASSERT_EQ(2, butex_requeue_return2[0]);
+    ASSERT_EQ(2, butex_requeue_return2[1]);
 }
 
 }  // namespace