diff --git a/src/bthread/butex.cpp b/src/bthread/butex.cpp index 63920ca9eb..35ea5e35ac 100644 --- a/src/bthread/butex.cpp +++ b/src/bthread/butex.cpp @@ -663,7 +663,13 @@ static int butex_wait_from_pthread(TaskGroup* g, Butex* b, int expected_value, 30/*nops before sched_yield*/); 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; } diff --git a/test/brpc_socket_unittest.cpp b/test/brpc_socket_unittest.cpp index 740a17289c..026331b704 100644 --- a/test/brpc_socket_unittest.cpp +++ b/test/brpc_socket_unittest.cpp @@ -1634,15 +1634,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 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 connections; - messenger->ListConnections(&connections); ASSERT_EQ(1ul, connections.size()); { diff --git a/test/bthread_butex_multi_tag_unittest.cpp b/test/bthread_butex_multi_tag_unittest.cpp index e41fefddaf..43e46ba40e 100644 --- a/test/bthread_butex_multi_tag_unittest.cpp +++ b/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 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 butex_wake_return; void* butex_wake_func(void* arg) { auto mutex = static_cast(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 butex_wake_all_return1(2, 0); -std::vector butex_wake_all_return2(2, 0); +std::vector butex_wake_all_return1; +std::vector 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 butex_requeue_return1(2, 0); -std::vector butex_requeue_return2(2, 0); +std::vector butex_requeue_return1; +std::vector 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 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 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); @@ -164,8 +191,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 diff --git a/test/bthread_butex_unittest.cpp b/test/bthread_butex_unittest.cpp index 730eac7b4e..f0876083c0 100644 --- a/test/bthread_butex_unittest.cpp +++ b/test/bthread_butex_unittest.cpp @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +#include +#include #include #include "butil/atomicops.h" #include "butil/time.h" @@ -35,6 +37,7 @@ inline TaskControl* get_task_control() { } // namespace bthread namespace { + TEST(ButexTest, wait_on_already_timedout_butex) { uint32_t* butex = bthread::butex_create_checked(); ASSERT_TRUE(butex); @@ -43,27 +46,35 @@ TEST(ButexTest, wait_on_already_timedout_butex) { *butex = 1; ASSERT_EQ(-1, bthread::butex_wait(butex, 1, &now)); ASSERT_EQ(ETIMEDOUT, errno); + bthread::butex_destroy(butex); } +struct JoinSleepArg { + bthread_t tid = 0; + uint64_t sleep_us = 0; + butil::atomic finished{false}; +}; + void* sleeper(void* arg) { - bthread_usleep((uint64_t)arg); + JoinSleepArg* a = static_cast(arg); + butil::Timer tm; + tm.start(); + EXPECT_EQ(0, bthread_usleep(a->sleep_us)); + tm.stop(); + // Sleep may finish late under load, but must not finish early. + EXPECT_GE(tm.u_elapsed(), static_cast(a->sleep_us)); + a->finished.store(true, butil::memory_order_release); return nullptr; } void* joiner(void* arg) { - const long t1 = butil::gettimeofday_us(); - for (bthread_t* th = (bthread_t*)arg; *th; ++th) { - if (0 != bthread_join(*th, nullptr)) { - LOG(FATAL) << "fail to join thread_" << th - (bthread_t*)arg; - } - long elp = butil::gettimeofday_us() - t1; - EXPECT_LE(labs(elp - (th - (bthread_t*)arg + 1) * 100000L), 15000L) - << "timeout when joining thread_" << th - (bthread_t*)arg; - LOG(INFO) << "Joined thread " << *th << " at " << elp << "us [" - << bthread_self() << "]"; + JoinSleepArg* args = static_cast(arg); + for (JoinSleepArg* a = args; a->tid; ++a) { + EXPECT_EQ(0, bthread_join(a->tid, nullptr)); + EXPECT_TRUE(a->finished.load(butil::memory_order_acquire)); } - for (bthread_t* th = (bthread_t*)arg; *th; ++th) { - EXPECT_EQ(0, bthread_join(*th, nullptr)); + for (JoinSleepArg* a = args; a->tid; ++a) { + EXPECT_EQ(0, bthread_join(a->tid, nullptr)); } return nullptr; } @@ -86,21 +97,20 @@ TEST(ButexTest, with_or_without_array_zero) { TEST(ButexTest, join) { const size_t N = 6; const size_t M = 6; - bthread_t th[N+1]; + JoinSleepArg args[N+1]; bthread_t jth[M]; pthread_t pth[M]; for (size_t i = 0; i < N; ++i) { bthread_attr_t attr = (i == 0 ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL); - ASSERT_EQ(0, bthread_start_urgent( - &th[i], &attr, sleeper, - (void*)(100000L/*100ms*/ * (i + 1)))); + args[i].sleep_us = 100000L/*100ms*/ * (i + 1); + ASSERT_EQ(0, bthread_start_urgent(&args[i].tid, &attr, sleeper, &args[i])); } - th[N] = 0; // joiner will join tids in `th' until seeing 0. + // The last argument's zero tid terminates the joiner's iteration. for (size_t i = 0; i < M; ++i) { - ASSERT_EQ(0, bthread_start_urgent(&jth[i], nullptr, joiner, th)); + ASSERT_EQ(0, bthread_start_urgent(&jth[i], nullptr, joiner, args)); } for (size_t i = 0; i < M; ++i) { - ASSERT_EQ(0, pthread_create(&pth[i], nullptr, joiner, th)); + ASSERT_EQ(0, pthread_create(&pth[i], nullptr, joiner, args)); } for (size_t i = 0; i < M; ++i) { @@ -122,14 +132,17 @@ struct WaiterArg { void* waiter(void* arg) { WaiterArg * wa = (WaiterArg*)arg; - const long t1 = butil::gettimeofday_us(); - const int rc = bthread::butex_wait( - wa->butex, wa->expected_value, wa->ptimeout); - const long t2 = butil::gettimeofday_us(); + long t1 = butil::gettimeofday_us(); + int rc = bthread::butex_wait(wa->butex, wa->expected_value, wa->ptimeout); + int saved_errno = errno; + long t2 = butil::gettimeofday_us(); if (rc == 0) { EXPECT_EQ(wa->expected_result, 0) << bthread_self(); } else { - EXPECT_EQ(wa->expected_result, errno) << bthread_self(); + EXPECT_EQ(wa->expected_result, saved_errno) << bthread_self(); + if (saved_errno == ETIMEDOUT && wa->ptimeout) { + EXPECT_GE(t2, butil::timespec_to_microseconds(*wa->ptimeout)); + } } LOG(INFO) << "after wait, time=" << (t2-t1) << "us"; return nullptr; @@ -138,9 +151,9 @@ void* waiter(void* arg) { TEST(ButexTest, sanity) { const size_t N = 5; WaiterArg args[N * 4]; - pthread_t t1, t2; - butil::atomic* b1 = - bthread::butex_create_checked >(); + pthread_t pthreads[2 * N]; + bthread_t bthreads[2 * N]; + butil::atomic* b1 = bthread::butex_create_checked >(); ASSERT_TRUE(b1); bthread::butex_destroy(b1); @@ -148,17 +161,17 @@ TEST(ButexTest, sanity) { *b1 = 1; ASSERT_EQ(0, bthread::butex_wake(b1)); - WaiterArg *unmatched_arg = new WaiterArg; - unmatched_arg->expected_value = *b1 + 1; - unmatched_arg->expected_result = EWOULDBLOCK; - unmatched_arg->butex = b1; - unmatched_arg->ptimeout = nullptr; - pthread_create(&t2, nullptr, waiter, unmatched_arg); - bthread_t th; - ASSERT_EQ(0, bthread_start_urgent(&th, nullptr, waiter, unmatched_arg)); - - const timespec abstime = butil::seconds_from_now(1); - for (size_t i = 0; i < 4*N; ++i) { + WaiterArg unmatched_arg = { EWOULDBLOCK, *b1 + 1, b1, nullptr }; + pthread_t unmatched_pthread; + bthread_t unmatched_bthread; + ASSERT_EQ(0, pthread_create(&unmatched_pthread, nullptr, waiter, &unmatched_arg)); + ASSERT_EQ(0, bthread_start_urgent( + &unmatched_bthread, nullptr, waiter, &unmatched_arg)); + ASSERT_EQ(0, pthread_join(unmatched_pthread, nullptr)); + ASSERT_EQ(0, bthread_join(unmatched_bthread, nullptr)); + + timespec abstime = butil::seconds_from_now(1); + for (size_t i = 0; i < 4 * N; ++i) { args[i].expected_value = *b1; args[i].butex = b1; if ((i % 2) == 0) { @@ -168,34 +181,81 @@ TEST(ButexTest, sanity) { args[i].expected_result = ETIMEDOUT; args[i].ptimeout = &abstime; } - if (i < 2*N) { - pthread_create(&t1, nullptr, waiter, &args[i]); + if (i < 2*N) { + ASSERT_EQ(0, pthread_create(&pthreads[i], nullptr, waiter, &args[i])); } else { - ASSERT_EQ(0, bthread_start_urgent(&th, nullptr, waiter, &args[i])); + ASSERT_EQ(0, bthread_start_urgent(&bthreads[i - 2*N], nullptr, waiter, &args[i])); } } - - sleep(2); - for (size_t i = 0; i < 2*N; ++i) { - ASSERT_EQ(1, bthread::butex_wake(b1)); + + // Join timed waiters before waking anyone, rather than assuming all + // timeout callbacks have completed after a fixed sleep. + for (size_t i = 1; i < 2 * N; i += 2) { + ASSERT_EQ(0, pthread_join(pthreads[i], nullptr)); + ASSERT_EQ(0, bthread_join(bthreads[i], nullptr)); + } + size_t nwoken = 0; + int64_t deadline = butil::cpuwide_time_us() + 5000000L; + while (nwoken < 2 * N) { + int rc = bthread::butex_wake(b1); + ASSERT_GE(rc, 0); + ASSERT_LE(rc, 1); + nwoken += rc; + if (rc == 0) { + ASSERT_LT(butil::cpuwide_time_us(), deadline) + << "Timed out waiting for untimed waiters to register"; + bthread_usleep(1000); + } } ASSERT_EQ(0, bthread::butex_wake(b1)); - sleep(1); + for (size_t i = 0; i < 2 * N; i += 2) { + ASSERT_EQ(0, pthread_join(pthreads[i], nullptr)); + ASSERT_EQ(0, bthread_join(bthreads[i], nullptr)); + } bthread::butex_destroy(b1); } +// A gate deliberately using pthread-backed synchronization: stopping a bthread +// must not consume its pending interruption while it waits at the test gate. +class TestGate { +public: + void wait() { + std::unique_lock lock(_mutex); + _cond.wait(lock, [this] { return _open; }); + } + + void signal() { + std::lock_guard lock(_mutex); + _open = true; + _cond.notify_all(); + } + +private: + std::mutex _mutex; + std::condition_variable _cond; + bool _open = false; +}; struct ButexWaitArg { int* butex; int expected_val; long wait_msec; int error_code; + TestGate* before_wait; + TestGate* entering_wait; }; void* wait_butex(void* void_arg) { ButexWaitArg* arg = static_cast(void_arg); - const timespec ts = butil::milliseconds_from_now(arg->wait_msec); - int rc = bthread::butex_wait(arg->butex, arg->expected_val, &ts); + if (arg->before_wait) { + arg->before_wait->wait(); + } + if (arg->entering_wait) { + arg->entering_wait->signal(); + } + timespec ts = butil::milliseconds_from_now(arg->wait_msec); + int rc = bthread::butex_wait(arg->butex, arg->expected_val, + arg->wait_msec < 0 ? nullptr : &ts); int saved_errno = errno; if (arg->error_code) { EXPECT_EQ(-1, rc); @@ -210,11 +270,11 @@ TEST(ButexTest, wait_without_stop) { int* butex = bthread::butex_create_checked(); *butex = 7; butil::Timer tm; - const long WAIT_MSEC = 500; + long WAIT_MSEC = 500; for (int i = 0; i < 2; ++i) { - const bthread_attr_t attr = - (i == 0 ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL); - ButexWaitArg arg = { butex, *butex, WAIT_MSEC, ETIMEDOUT }; + bthread_attr_t attr = (i == 0 ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL); + ButexWaitArg arg = { butex, *butex, WAIT_MSEC, ETIMEDOUT, + nullptr, nullptr }; bthread_t th; tm.start(); @@ -222,7 +282,8 @@ TEST(ButexTest, wait_without_stop) { ASSERT_EQ(0, bthread_join(th, nullptr)); tm.stop(); - ASSERT_LT(labs(tm.m_elapsed() - WAIT_MSEC), 250); + // Timer delivery and rescheduling may be arbitrarily delayed by load. + ASSERT_GE(tm.m_elapsed(), WAIT_MSEC); } bthread::butex_destroy(butex); } @@ -230,183 +291,211 @@ TEST(ButexTest, wait_without_stop) { TEST(ButexTest, stop_after_running) { int* butex = bthread::butex_create_checked(); *butex = 7; - butil::Timer tm; - const long WAIT_MSEC = 500; - const long SLEEP_MSEC = 10; - for (int i = 0; i < 2; ++i) { - const bthread_attr_t attr = - (i == 0 ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL); + // Repetition also covers interruption between waiter registration and + // the pthread's futex wait, which must not leak EWOULDBLOCK to the caller. + for (int i = 0; i < 100; ++i) { + bthread_attr_t attr = (i % 2 == 0 ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL); bthread_t th; - ButexWaitArg arg = { butex, *butex, WAIT_MSEC, EINTR }; + TestGate entering_wait; + // No timeout may race with stop. The waiter must report EINTR. + ButexWaitArg arg = { butex, *butex, -1, EINTR, + nullptr, &entering_wait }; - tm.start(); ASSERT_EQ(0, bthread_start_urgent(&th, &attr, wait_butex, &arg)); - ASSERT_EQ(0, bthread_usleep(SLEEP_MSEC * 1000L)); - ASSERT_EQ(0, bthread_stop(th)); + entering_wait.wait(); + EXPECT_EQ(0, bthread_stop(th)); ASSERT_EQ(0, bthread_join(th, nullptr)); - tm.stop(); - - ASSERT_LT(labs(tm.m_elapsed() - SLEEP_MSEC), 25); - // ASSERT_TRUE(bthread::get_task_control()-> - // timer_thread()._idset.empty()); ASSERT_EQ(EINVAL, bthread_stop(th)); - } + } bthread::butex_destroy(butex); } TEST(ButexTest, stop_before_running) { int* butex = bthread::butex_create_checked(); *butex = 7; - butil::Timer tm; - const long WAIT_MSEC = 500; for (int i = 0; i < 2; ++i) { - const bthread_attr_t attr = + bthread_attr_t attr = (i == 0 ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL) | BTHREAD_NOSIGNAL; bthread_t th; - ButexWaitArg arg = { butex, *butex, WAIT_MSEC, EINTR }; - - tm.start(); + TestGate before_wait; + ButexWaitArg arg = { butex, *butex, -1, EINTR, + &before_wait, nullptr }; + ASSERT_EQ(0, bthread_start_background(&th, &attr, wait_butex, &arg)); - ASSERT_EQ(0, bthread_stop(th)); + EXPECT_EQ(0, bthread_stop(th)); + // NOSIGNAL suppresses notification, but does not prevent a worker + // from picking up the task. The gate enforces stop-before-wait. + before_wait.signal(); bthread_flush(); ASSERT_EQ(0, bthread_join(th, nullptr)); - tm.stop(); - - ASSERT_LT(tm.m_elapsed(), 5); - // ASSERT_TRUE(bthread::get_task_control()-> - // timer_thread()._idset.empty()); ASSERT_EQ(EINVAL, bthread_stop(th)); } bthread::butex_destroy(butex); } +struct JoinWaiterArg { + bthread_t tid; + TestGate entering_join; + butil::atomic finished{false}; +}; + void* join_the_waiter(void* arg) { - EXPECT_EQ(0, bthread_join((bthread_t)arg, nullptr)); + JoinWaiterArg* a = static_cast(arg); + a->entering_join.signal(); + EXPECT_EQ(0, bthread_join(a->tid, nullptr)); + a->finished.store(true, butil::memory_order_release); return nullptr; } TEST(ButexTest, join_cant_be_wakeup) { - const long WAIT_MSEC = 100; int* butex = bthread::butex_create_checked(); *butex = 7; - butil::Timer tm; - ButexWaitArg arg = { butex, *butex, 1000, EINTR }; - for (int i = 0; i < 2; ++i) { - const bthread_attr_t attr = + bthread_attr_t attr = (i == 0 ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL); - tm.start(); bthread_t th, th2; + TestGate entering_wait; + ButexWaitArg arg = { butex, *butex, -1, EINTR, + nullptr, &entering_wait }; ASSERT_EQ(0, bthread_start_urgent(&th, nullptr, wait_butex, &arg)); - ASSERT_EQ(0, bthread_start_urgent(&th2, &attr, join_the_waiter, (void*)th)); - ASSERT_EQ(0, bthread_stop(th2)); - ASSERT_EQ(0, bthread_usleep(WAIT_MSEC / 2 * 1000L)); - ASSERT_TRUE(bthread::TaskGroup::exists(th)); - ASSERT_TRUE(bthread::TaskGroup::exists(th2)); - ASSERT_EQ(0, bthread_usleep(WAIT_MSEC / 2 * 1000L)); - ASSERT_EQ(0, bthread_stop(th)); + // Start the target before a pthread-stack joiner can block its worker. + entering_wait.wait(); + JoinWaiterArg join_arg; + join_arg.tid = th; + ASSERT_EQ(0, bthread_start_urgent(&th2, &attr, join_the_waiter, &join_arg)); + join_arg.entering_join.wait(); + EXPECT_EQ(0, bthread_stop(th2)); + // Give the interrupted joiner a chance to run. There is no deadline + // on the target waiter, so scheduler delays cannot make it exit. + EXPECT_EQ(0, bthread_usleep(50000)); + EXPECT_FALSE(join_arg.finished.load(butil::memory_order_acquire)); + EXPECT_TRUE(bthread::TaskGroup::exists(th)); + EXPECT_TRUE(bthread::TaskGroup::exists(th2)); + EXPECT_EQ(0, bthread_stop(th)); ASSERT_EQ(0, bthread_join(th2, nullptr)); ASSERT_EQ(0, bthread_join(th, nullptr)); - tm.stop(); - ASSERT_LT(tm.m_elapsed(), WAIT_MSEC + 15); + EXPECT_TRUE(join_arg.finished.load(butil::memory_order_acquire)); ASSERT_EQ(EINVAL, bthread_stop(th)); ASSERT_EQ(EINVAL, bthread_stop(th2)); } bthread::butex_destroy(butex); } -TEST(ButexTest, stop_after_slept) { +struct StopSleepArg { + bool pthread_task; + TestGate* before_sleep; + TestGate entering_sleep; + TestGate allow_exit; +}; + +void* stoppable_sleeper(void* arg) { + StopSleepArg* a = static_cast(arg); + if (a->before_sleep) { + a->before_sleep->wait(); + } + // Pthread-stack tasks use native usleep, which bthread_stop cannot + // interrupt. Normal bthreads must return ESTOP rather than time out. + int64_t sleep_us = a->pthread_task ? 100000L : 60000000L; + a->entering_sleep.signal(); butil::Timer tm; - const long SLEEP_MSEC = 100; - const long WAIT_MSEC = 10; - - for (int i = 0; i < 2; ++i) { - const bthread_attr_t attr = - (i == 0 ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL); - tm.start(); - bthread_t th; - ASSERT_EQ(0, bthread_start_urgent( - &th, &attr, sleeper, (void*)(SLEEP_MSEC*1000L))); - ASSERT_EQ(0, bthread_usleep(WAIT_MSEC * 1000L)); - ASSERT_EQ(0, bthread_stop(th)); - ASSERT_EQ(0, bthread_join(th, nullptr)); - tm.stop(); - if (attr.stack_type == BTHREAD_STACKTYPE_PTHREAD) { - ASSERT_LT(labs(tm.m_elapsed() - SLEEP_MSEC), 15); - } else { - ASSERT_LT(labs(tm.m_elapsed() - WAIT_MSEC), 15); - } - // ASSERT_TRUE(bthread::get_task_control()-> - // timer_thread()._idset.empty()); - ASSERT_EQ(EINVAL, bthread_stop(th)); + tm.start(); + int rc = bthread_usleep(sleep_us); + int saved_errno = errno; + tm.stop(); + if (a->pthread_task) { + EXPECT_EQ(0, rc); + EXPECT_GE(tm.u_elapsed(), sleep_us); + } else { + EXPECT_EQ(-1, rc); + EXPECT_EQ(ESTOP, saved_errno); } + // Keep the task alive even if the controller is descheduled longer than + // the native sleep, so stop() cannot race with task destruction. + a->allow_exit.wait(); + return nullptr; } -TEST(ButexTest, stop_just_when_sleeping) { - butil::Timer tm; - const long SLEEP_MSEC = 100; - +void TestStopSleep(bool stop_before_sleep, bool wait_until_sleeping) { for (int i = 0; i < 2; ++i) { - const bthread_attr_t attr = - (i == 0 ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL); - tm.start(); + bthread_attr_t attr = (i == 0 ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL); + StopSleepArg arg; + arg.pthread_task = (i == 0); + TestGate before_sleep; + arg.before_sleep = stop_before_sleep ? &before_sleep : nullptr; bthread_t th; - ASSERT_EQ(0, bthread_start_urgent( - &th, &attr, sleeper, (void*)(SLEEP_MSEC*1000L))); - ASSERT_EQ(0, bthread_stop(th)); - ASSERT_EQ(0, bthread_join(th, nullptr)); - tm.stop(); - if (attr.stack_type == BTHREAD_STACKTYPE_PTHREAD) { - ASSERT_LT(labs(tm.m_elapsed() - SLEEP_MSEC), 15); + if (stop_before_sleep) { + attr = attr | BTHREAD_NOSIGNAL; + ASSERT_EQ(0, bthread_start_background( + &th, &attr, stoppable_sleeper, &arg)); } else { - ASSERT_LT(tm.m_elapsed(), 15); + ASSERT_EQ(0, bthread_start_urgent(&th, &attr, stoppable_sleeper, &arg)); + arg.entering_sleep.wait(); + if (wait_until_sleeping && !arg.pthread_task) { + // Observe timer registration instead of assuming a fixed + // delay is enough for the worker to enter sleep. + bthread::TaskMeta* meta = bthread::TaskGroup::address_meta(th); + int64_t deadline = butil::cpuwide_time_us() + 5000000L; + bool sleeping = false; + do { + pthread_spin_lock(&meta->version_lock); + sleeping = (meta->current_sleep != 0); + pthread_spin_unlock(&meta->version_lock); + if (sleeping) { + break; + } + bthread_usleep(1000); + } while (butil::cpuwide_time_us() < deadline); + EXPECT_TRUE(sleeping) << "Timed out waiting for sleep registration"; + } + } + EXPECT_EQ(0, bthread_stop(th)); + before_sleep.signal(); + arg.allow_exit.signal(); + if (stop_before_sleep) { + bthread_flush(); } - // ASSERT_TRUE(bthread::get_task_control()-> - // timer_thread()._idset.empty()); + ASSERT_EQ(0, bthread_join(th, nullptr)); ASSERT_EQ(EINVAL, bthread_stop(th)); } } +TEST(ButexTest, stop_after_slept) { + TestStopSleep(false, true); +} + +TEST(ButexTest, stop_just_when_sleeping) { + TestStopSleep(false, false); +} + TEST(ButexTest, stop_before_sleeping) { - butil::Timer tm; - const long SLEEP_MSEC = 100; + TestStopSleep(true, false); +} - for (int i = 0; i < 2; ++i) { - bthread_t th; - const bthread_attr_t attr = - (i == 0 ? BTHREAD_ATTR_PTHREAD : BTHREAD_ATTR_NORMAL) | BTHREAD_NOSIGNAL; - - tm.start(); - ASSERT_EQ(0, bthread_start_background(&th, &attr, sleeper, - (void*)(SLEEP_MSEC*1000L))); - ASSERT_EQ(0, bthread_stop(th)); - bthread_flush(); - ASSERT_EQ(0, bthread_join(th, nullptr)); - tm.stop(); +struct SignalArg { + pthread_t waiter; + WaiterArg* wait_arg; + butil::atomic stop{false}; +}; - if (attr.stack_type == BTHREAD_STACKTYPE_PTHREAD) { - ASSERT_LT(labs(tm.m_elapsed() - SLEEP_MSEC), 10); - } else { - ASSERT_LT(tm.m_elapsed(), 10); - } - // ASSERT_TRUE(bthread::get_task_control()-> - // timer_thread()._idset.empty()); - ASSERT_EQ(EINVAL, bthread_stop(th)); - } +void* signal_waiter(void* arg) { + SignalArg* a = static_cast(arg); + waiter(a->wait_arg); + a->stop.store(true); + return nullptr; } void* trigger_signal(void* arg) { - pthread_t * th = (pthread_t*)arg; - const long t1 = butil::gettimeofday_us(); - for (size_t i = 0; i < 50; ++i) { + SignalArg* a = static_cast(arg); + long t1 = butil::gettimeofday_us(); + for (size_t i = 0; i < 50 && !a->stop.load(); ++i) { usleep(100000); - if (bthread::interrupt_pthread(*th) == ESRCH) { + if (a->stop.load() || bthread::interrupt_pthread(a->waiter) == ESRCH) { LOG(INFO) << "waiter thread end, trigger count=" << i; break; } } - const long t2 = butil::gettimeofday_us(); + long t2 = butil::gettimeofday_us(); LOG(INFO) << "trigger signal thread end, elapsed=" << (t2-t1) << "us"; return nullptr; } @@ -414,7 +503,7 @@ void* trigger_signal(void* arg) { TEST(ButexTest, wait_with_signal_triggered) { butil::Timer tm; - const int64_t WAIT_MSEC = 500; + int64_t WAIT_MSEC = 500; WaiterArg waiter_args; pthread_t waiter_th, tigger_th; butil::atomic* butex = @@ -423,23 +512,34 @@ TEST(ButexTest, wait_with_signal_triggered) { *butex = 1; ASSERT_EQ(0, bthread::butex_wake(butex)); - const timespec abstime = butil::milliseconds_from_now(WAIT_MSEC); + timespec abstime = butil::milliseconds_from_now(WAIT_MSEC); waiter_args.expected_value = *butex; waiter_args.butex = butex; waiter_args.expected_result = ETIMEDOUT; waiter_args.ptimeout = &abstime; tm.start(); - pthread_create(&waiter_th, nullptr, waiter, &waiter_args); - pthread_create(&tigger_th, nullptr, trigger_signal, &waiter_th); - + SignalArg signal_arg; + signal_arg.wait_arg = &waiter_args; + ASSERT_EQ(0, pthread_create(&waiter_th, nullptr, signal_waiter, &signal_arg)); + signal_arg.waiter = waiter_th; + int signal_rc = pthread_create( + &tigger_th, nullptr, trigger_signal, &signal_arg); + EXPECT_EQ(0, signal_rc); + // Keep the target joinable until the signalling thread has stopped. + // pthread_kill on a pthread_t whose lifetime ended at join is undefined. + if (signal_rc == 0) { + EXPECT_EQ(0, pthread_join(tigger_th, nullptr)); + } ASSERT_EQ(0, pthread_join(waiter_th, nullptr)); tm.stop(); - auto wait_elapsed_ms = tm.m_elapsed();; + auto wait_elapsed_ms = tm.m_elapsed(); LOG(INFO) << "waiter thread end, elapsed " << wait_elapsed_ms << " ms"; - ASSERT_LT(labs(wait_elapsed_ms - WAIT_MSEC), 250); + // The timeout is absolute and starts before the worker is scheduled. + // Check the deadline itself rather than a narrow elapsed-time window. + EXPECT_GE(butil::gettimeofday_us(), + butil::timespec_to_microseconds(abstime)); - ASSERT_EQ(0, pthread_join(tigger_th, nullptr)); bthread::butex_destroy(butex); } diff --git a/test/bthread_cond_unittest.cpp b/test/bthread_cond_unittest.cpp index 85181691e5..5fd0a70d73 100644 --- a/test/bthread_cond_unittest.cpp +++ b/test/bthread_cond_unittest.cpp @@ -145,7 +145,6 @@ std::atomic WrapperArg::wake_time{0}; void* cv_signaler(void* void_arg) { WrapperArg* a = (WrapperArg*)void_arg; - signal_start_time = butil::gettimeofday_us(); while (!stop) { bthread_usleep(SIGNAL_INTERVAL_US); a->cond.notify_one(); @@ -224,6 +223,7 @@ TEST(CondTest, cpp_wrapper) { TEST(CondTest, cpp_wrapper2) { stop = false; + WrapperArg::wake_time = 0; bthread::ConditionVariable cond; pthread_t bmutex_waiter_threads[8]; pthread_t mutex_waiter_threads[8]; diff --git a/test/bthread_fd_unittest.cpp b/test/bthread_fd_unittest.cpp index 02c579c168..9a117835b8 100644 --- a/test/bthread_fd_unittest.cpp +++ b/test/bthread_fd_unittest.cpp @@ -411,39 +411,69 @@ TEST(FDTest, add_existing_fd) { #endif } +struct EpollWaitArg { + int epfd; + butil::atomic done{false}; + int result = 0; + int error = 0; +}; + void* epoll_waiter(void* arg) { + EpollWaitArg* a = static_cast(arg); #if defined(OS_LINUX) epoll_event e; - if (1 == epoll_wait((int)(intptr_t)arg, &e, 1, -1)) { - std::cout << e.events << std::endl; - } + a->result = epoll_wait(a->epfd, &e, 1, 10000); #elif defined(OS_MACOSX) struct kevent e; - if (1 == kevent((int)(intptr_t)arg, nullptr, 0, &e, 1, nullptr)) { - std::cout << e.flags << std::endl; - } + timespec timeout = {10, 0}; + a->result = kevent(a->epfd, nullptr, 0, &e, 1, &timeout); #endif - std::cout << pthread_self() << " quits" << std::endl; + a->error = errno; + a->done.store(true, butil::memory_order_release); return nullptr; } TEST(FDTest, interrupt_pthread) { #if defined(OS_LINUX) - const int epfd = epoll_create(1024); + butil::fd_guard epfd(epoll_create(1024)); #elif defined(OS_MACOSX) - const int epfd = kqueue(); + butil::fd_guard epfd(kqueue()); #endif - pthread_t th, th2; - ASSERT_EQ(0, pthread_create(&th, nullptr, epoll_waiter, (void*)(intptr_t)epfd)); - ASSERT_EQ(0, pthread_create(&th2, nullptr, epoll_waiter, (void*)(intptr_t)epfd)); - bthread_usleep(100000L); - std::cout << "wake up " << th << std::endl; - bthread::interrupt_pthread(th); - bthread_usleep(100000L); - std::cout << "wake up " << th2 << std::endl; - bthread::interrupt_pthread(th2); - pthread_join(th, nullptr); - pthread_join(th2, nullptr); + ASSERT_GE(epfd, 0); + EpollWaitArg args[2]; + pthread_t threads[2]; + size_t started = 0; + for (; started < ARRAY_SIZE(threads); ++started) { + args[started].epfd = epfd; + int rc = pthread_create(&threads[started], nullptr, + epoll_waiter, &args[started]); + EXPECT_EQ(0, rc); + if (rc != 0) { + break; + } + } + int64_t deadline = butil::cpuwide_time_us() + 15000000L; + for (size_t i = 0; i < started; ++i) { + // Signals are not persistent. Retry until the syscall observes one; + // keep the pthread joinable until all signalling is finished. + while (!args[i].done.load(butil::memory_order_acquire) && + butil::cpuwide_time_us() < deadline) { + int rc = bthread::interrupt_pthread(threads[i]); + if (rc != 0) { + // The waiter may finish between the check above and the + // signal, in which case interruption is no longer needed. + // Why it stopped waiting is checked on args[i] below. + EXPECT_EQ(ESRCH, rc) << berror(rc); + break; + } + bthread_usleep(1000); + } + EXPECT_EQ(0, pthread_join(threads[i], nullptr)); + } + for (size_t i = 0; i < started; ++i) { + ASSERT_EQ(-1, args[i].result); + ASSERT_EQ(EINTR, args[i].error); + } } void* close_the_fd(void* arg) { @@ -484,49 +514,73 @@ TEST(FDTest, invalid_epoll_events) { ASSERT_EQ(0, bthread_fd_wait(fds[0], EVFILT_READ)); #endif tm.stop(); - ASSERT_LT(tm.m_elapsed(), 20); + // Successful readiness, not scheduler latency, is the contract. ASSERT_EQ(0, bthread_join(th, nullptr)); ASSERT_EQ(0, bthread_close(fds[0])); } +struct FDWaitArg { + int fd; + int timeout_ms; + int result = 0; + int error = 0; +}; + void* wait_for_the_fd(void* arg) { - timespec ts = butil::milliseconds_from_now(50); + FDWaitArg* a = static_cast(arg); + timespec ts = butil::milliseconds_from_now(a->timeout_ms); #if defined(OS_LINUX) - bthread_fd_timedwait(*(int*)arg, EPOLLIN, &ts); + a->result = bthread_fd_timedwait(a->fd, EPOLLIN, &ts); #elif defined(OS_MACOSX) - bthread_fd_timedwait(*(int*)arg, EVFILT_READ, &ts); + a->result = bthread_fd_timedwait(a->fd, EVFILT_READ, &ts); #endif + a->error = errno; + if (a->result == -1 && a->error == ETIMEDOUT) { + EXPECT_GE(butil::gettimeofday_us(), butil::timespec_to_microseconds(ts)); + } return nullptr; } TEST(FDTest, timeout) { int fds[2]; ASSERT_EQ(0, pipe(fds)); + FDWaitArg args[2]; + for (auto& arg : args) { + arg.fd = fds[0]; + arg.timeout_ms = 50; + } pthread_t th; - ASSERT_EQ(0, pthread_create(&th, nullptr, wait_for_the_fd, &fds[0])); + ASSERT_EQ(0, pthread_create(&th, nullptr, wait_for_the_fd, &args[0])); bthread_t bth; - ASSERT_EQ(0, bthread_start_urgent(&bth, nullptr, wait_for_the_fd, &fds[0])); - butil::Timer tm; - tm.start(); + ASSERT_EQ(0, bthread_start_urgent(&bth, nullptr, wait_for_the_fd, &args[1])); ASSERT_EQ(0, pthread_join(th, nullptr)); ASSERT_EQ(0, bthread_join(bth, nullptr)); - tm.stop(); - ASSERT_LT(tm.m_elapsed(), 80); ASSERT_EQ(0, bthread_close(fds[0])); ASSERT_EQ(0, bthread_close(fds[1])); + for (auto& arg : args) { + ASSERT_EQ(-1, arg.result); + ASSERT_EQ(ETIMEDOUT, arg.error); + } } TEST(FDTest, close_should_wakeup_waiter) { int fds[2]; ASSERT_EQ(0, pipe(fds)); + FDWaitArg arg; + arg.fd = fds[0]; + arg.timeout_ms = 10000; bthread_t bth; - ASSERT_EQ(0, bthread_start_urgent(&bth, nullptr, wait_for_the_fd, &fds[0])); - butil::Timer tm; - tm.start(); + ASSERT_EQ(0, bthread_start_urgent(&bth, nullptr, wait_for_the_fd, &arg)); + auto* meta = bthread::TaskGroup::address_meta(bth); + 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)); ASSERT_EQ(0, bthread_close(fds[0])); ASSERT_EQ(0, bthread_join(bth, nullptr)); - tm.stop(); - ASSERT_LT(tm.m_elapsed(), 5); + ASSERT_EQ(0, arg.result) << "errno=" << arg.error; // Launch again, should quit soon due to EBADF #if defined(OS_LINUX) @@ -566,73 +620,74 @@ TEST(FDTest, double_close) { ASSERT_EQ(ec, errno); } -const char* g_hostname1 = "github.com"; -const char* g_hostname2 = "baidu.com"; -TEST(FDTest, bthread_connect) { - butil::EndPoint ep1; - butil::EndPoint ep2; - ASSERT_EQ(0, butil::hostname2endpoint(g_hostname1, 80, &ep1)); - ASSERT_EQ(0, butil::hostname2endpoint(g_hostname2, 80, &ep2)); - - { - struct sockaddr_storage serv_addr{}; - socklen_t serv_addr_size = 0; - ASSERT_EQ(0, endpoint2sockaddr(ep1, &serv_addr, &serv_addr_size)); - butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM, 0)); - ASSERT_LE(0, sockfd); - bool is_blocking = butil::is_blocking(sockfd); - ASSERT_LE(0, sockfd); - ASSERT_EQ(0, bthread_connect(sockfd, (struct sockaddr*) &serv_addr, serv_addr_size)); - ASSERT_EQ(is_blocking, butil::is_blocking(sockfd)); +// Local listeners keep connect tests independent of DNS, Internet latency, +// and the assumption that a connection cannot complete within one millisecond. +void TestLocalConnect(bool timed) { + butil::EndPoint endpoint; + ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:0", &endpoint)); + butil::fd_guard listener(butil::tcp_listen(endpoint)); + ASSERT_GE(listener, 0); + ASSERT_EQ(0, butil::get_local_side(listener, &endpoint)); + struct sockaddr_storage address{}; + socklen_t length = 0; + ASSERT_EQ(0, endpoint2sockaddr(endpoint, &address, &length)); + butil::fd_guard client(socket(address.ss_family, SOCK_STREAM, 0)); + ASSERT_GE(client, 0); + bool was_blocking = butil::is_blocking(client); + timespec deadline = butil::seconds_from_now(10); + int rc = bthread_timed_connect( + client, reinterpret_cast(&address), length, + timed ? &deadline : nullptr); + ASSERT_EQ(0, rc) << "errno=" << errno; + ASSERT_EQ(was_blocking, butil::is_blocking(client)); + ASSERT_EQ(0, butil::is_connected(client)); + // The handshake does not require a concurrent accept thread. + butil::fd_guard accepted(accept(listener, nullptr, nullptr)); + ASSERT_GE(accepted, 0); +} - } +TEST(FDTest, bthread_connect) { + TestLocalConnect(false); + TestLocalConnect(true); +} - { - struct sockaddr_storage serv_addr{}; - socklen_t serv_addr_size = 0; - ASSERT_EQ(0, endpoint2sockaddr(ep2, &serv_addr, &serv_addr_size)); - butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM, 0)); - ASSERT_LE(0, sockfd); - bool is_blocking = butil::is_blocking(sockfd); - // In most cases, 1 millisecond will result in a connection timeout. - timespec abstime = butil::milliseconds_from_now(1); - const int rc = bthread_timed_connect( - sockfd, (struct sockaddr*) &serv_addr, - serv_addr_size, &abstime); - ASSERT_EQ(-1, rc); - ASSERT_EQ(ETIMEDOUT, errno); - ASSERT_EQ(is_blocking, butil::is_blocking(sockfd)); - } +#if defined(OS_LINUX) +TEST(FDTest, connect_timeout_with_full_accept_queue) { + butil::EndPoint endpoint; + ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:0", &endpoint)); + butil::fd_guard listener(butil::tcp_listen(endpoint)); + ASSERT_GE(listener, 0); + // Linux permits one queued connection for backlog=0. Leave it unaccepted + // so the next handshake cannot complete; no external network is needed. + ASSERT_EQ(0, listen(listener, 0)); + ASSERT_EQ(0, butil::get_local_side(listener, &endpoint)); + sockaddr_storage address{}; + socklen_t length = 0; + ASSERT_EQ(0, endpoint2sockaddr(endpoint, &address, &length)); + butil::fd_guard queued(socket(AF_INET, SOCK_STREAM, 0)); + ASSERT_GE(queued, 0); + timespec setup_deadline = butil::seconds_from_now(5); + ASSERT_EQ(0, bthread_timed_connect(queued, + reinterpret_cast(&address), length, &setup_deadline)); + butil::fd_guard client(socket(AF_INET, SOCK_STREAM, 0)); + ASSERT_GE(client, 0); + timespec deadline = butil::milliseconds_from_now(50); + int rc = bthread_timed_connect(client, + reinterpret_cast(&address), length, &deadline); + int error = errno; + ASSERT_EQ(-1, rc); + ASSERT_EQ(ETIMEDOUT, error); + EXPECT_TRUE(butil::is_blocking(client)); } +#endif void TestConnectInterruptImpl(bool timed) { - butil::EndPoint ep; - ASSERT_EQ(0, butil::hostname2endpoint(g_hostname1, 80, &ep)); - struct sockaddr_storage serv_addr{}; - socklen_t serv_addr_size = 0; - ASSERT_EQ(0, endpoint2sockaddr(ep, &serv_addr, &serv_addr_size)); - butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM, 0)); - ASSERT_GE(sockfd, 0); - - int rc; - if (timed) { - int64_t start_ms = butil::cpuwide_time_ms(); - butil::tcp_connect(ep, nullptr); - int64_t connect_ms = butil::cpuwide_time_ms() - start_ms; - LOG(INFO) << "Connect to " << ep << ", cost " << connect_ms << "ms"; - - timespec abstime = butil::milliseconds_from_now(connect_ms * 10); - rc = bthread_timed_connect( - sockfd, (struct sockaddr*) &serv_addr, - serv_addr_size, &abstime); - } else { - rc = bthread_timed_connect( - sockfd, (struct sockaddr*) &serv_addr, - serv_addr_size, nullptr); + // Stop must precede connect even if the task starts on another worker + // immediately. Yield does not consume the pending interruption. + while (!bthread_stopped(bthread_self())) { + bthread_yield(); } - ASSERT_EQ(0, rc) << "errno=" << errno; - ASSERT_EQ(0, butil::is_connected(sockfd)); - + TestLocalConnect(timed); } void* ConnectThread(void* arg) { diff --git a/test/bthread_futex_unittest.cpp b/test/bthread_futex_unittest.cpp index a9742755c9..1a8d564656 100644 --- a/test/bthread_futex_unittest.cpp +++ b/test/bthread_futex_unittest.cpp @@ -20,6 +20,7 @@ #include #include #include +#include #include "butil/time.h" #include "butil/macros.h" #include "butil/errno.h" @@ -30,7 +31,7 @@ #include namespace { -volatile bool stop = false; +butil::atomic stop(false); butil::atomic nthread(0); @@ -57,21 +58,26 @@ void* read_thread(void* arg) { } ++nthread; - bthread::futex_wait_private(m/*lock1*/, 0/*consumed_njob*/, nullptr); + // A stop between the loop condition and futex_wait must not leave + // this worker asleep forever. Periodically recheck the stop flag. + timespec timeout = butil::milliseconds_to_timespec(100); + bthread::futex_wait_private(m/*lock1*/, 0/*consumed_njob*/, &timeout); --nthread; } return new int(njob); } TEST(FutexTest, rdlock_performance) { - const size_t N = 100000; + stop = false; + nthread = 0; + size_t N = 100000; butil::atomic lock1(0); pthread_t rth[8]; for (size_t i = 0; i < ARRAY_SIZE(rth); ++i) { ASSERT_EQ(0, pthread_create(&rth[i], nullptr, read_thread, &lock1)); } - const int64_t t1 = butil::cpuwide_time_ns(); + int64_t t1 = butil::cpuwide_time_ns(); for (size_t i = 0; i < N; ++i) { if (nthread) { lock1.fetch_add(1); @@ -83,14 +89,11 @@ TEST(FutexTest, rdlock_performance) { } } } - const int64_t t2 = butil::cpuwide_time_ns(); + int64_t t2 = butil::cpuwide_time_ns(); bthread_usleep(3000000); stop = true; - for (int i = 0; i < 10; ++i) { - bthread::futex_wake_private(&lock1, INT_MAX); - sched_yield(); - } + bthread::futex_wake_private(&lock1, INT_MAX); int njob = 0; int* res; @@ -113,30 +116,55 @@ TEST(FutexTest, futex_wake_before_wait) { } void* dummy_waiter(void* lock) { - bthread::futex_wait_private(lock, 0, nullptr); + timespec timeout = butil::seconds_to_timespec(10); + int rc; + do { + rc = bthread::futex_wait_private(lock, 0, &timeout); + } while (rc != 0 && errno == EINTR); + EXPECT_EQ(0, rc); return nullptr; } TEST(FutexTest, futex_wake_many_waiters_perf) { - int lock1 = 0; - size_t N = 0; - pthread_t th; - for (; N < 1000 && !pthread_create(&th, nullptr, dummy_waiter, &lock1); ++N) {} - - sleep(1); + butil::atomic lock1(0); + std::vector threads; + for (size_t i = 0; i < 1000; ++i) { + pthread_t th; + if (pthread_create(&th, nullptr, dummy_waiter, &lock1) != 0) { + break; + } + threads.push_back(th); + } + ASSERT_FALSE(threads.empty()); + size_t N = threads.size(); int nwakeup = 0; + int64_t wake_ns = 0; + int64_t deadline = butil::cpuwide_time_us() + 5000000L; butil::Timer tm; - tm.start(); - for (size_t i = 0; i < N; ++i) { - nwakeup += bthread::futex_wake_private(&lock1, 1); + while (static_cast(nwakeup) < N && + butil::cpuwide_time_us() < deadline) { + tm.start(); + int rc = bthread::futex_wake_private(&lock1, 1); + tm.stop(); + EXPECT_GE(rc, 0); + if (rc > 0) { + nwakeup += rc; + wake_ns += tm.n_elapsed(); + } else { + usleep(1000); + } } - tm.stop(); - printf("N=%lu, futex_wake a thread = %" PRId64 "ns\n", N, tm.n_elapsed() / N); - ASSERT_EQ(N, (size_t)nwakeup); + // Also release late waiters on failure; a wake alone is not persistent. + lock1.store(1); + bthread::futex_wake_private(&lock1, INT_MAX); + for (pthread_t th : threads) { + EXPECT_EQ(0, pthread_join(th, nullptr)); + } + ASSERT_EQ(N, static_cast(nwakeup)); + printf("N=%lu, futex_wake a thread = %" PRId64 "ns\n", N, wake_ns / N); - sleep(2); - const size_t REP = 10000; + size_t REP = 10000; nwakeup = 0; tm.start(); for (size_t i = 0; i < REP; ++i) { @@ -191,7 +219,7 @@ void* batch_waker(void* lock) { TEST(FutexTest, many_futex_wake_nop_perf) { pthread_t th[8]; - int lock1; + int lock1 = 0; std::cout << "[Direct wake]" << std::endl; for (size_t i = 0; i < ARRAY_SIZE(th); ++i) { ASSERT_EQ(0, pthread_create(&th[i], nullptr, waker, &lock1)); diff --git a/test/bthread_mutex_unittest.cpp b/test/bthread_mutex_unittest.cpp index 9849442160..65a543ad94 100644 --- a/test/bthread_mutex_unittest.cpp +++ b/test/bthread_mutex_unittest.cpp @@ -52,8 +52,14 @@ TEST(MutexTest, sanity) { ASSERT_EQ(1u, *get_butex(m)); bthread_t th1; ASSERT_EQ(0, bthread_start_urgent(&th1, nullptr, locker, &m)); - usleep(5000); // wait for locker to run. - ASSERT_EQ(257u, *get_butex(m)); // contention + auto* state = reinterpret_cast*>(get_butex(m)); + int64_t deadline = butil::cpuwide_time_us() + 5000000L; + while (state->load(butil::memory_order_relaxed) != 257u && + butil::cpuwide_time_us() < deadline) { + usleep(1000); + } + // Keep cleanup reachable even if the worker did not run in time. + ASSERT_EQ(257u, state->load(butil::memory_order_relaxed)); ASSERT_EQ(0, bthread_mutex_unlock(&m)); ASSERT_EQ(0, bthread_join(th1, nullptr)); ASSERT_EQ(0u, *get_butex(m)); diff --git a/test/bthread_rwlock_unittest.cpp b/test/bthread_rwlock_unittest.cpp index 3747c29f2a..9031a5798e 100644 --- a/test/bthread_rwlock_unittest.cpp +++ b/test/bthread_rwlock_unittest.cpp @@ -21,6 +21,8 @@ #include #include #include +#include "bthread/task_group.h" +#include "bthread/task_meta.h" #include namespace { @@ -210,8 +212,21 @@ TEST(RWLockTest, cpp_wrapper) { } } -bool g_started = false; -bool g_stopped = false; +butil::atomic g_started(false); +butil::atomic g_stopped(false); + +// Read only synchronized predicates; deadlines bound failures, not latency. +template +bool WaitForRWLockState(Predicate predicate) { + int64_t deadline = butil::cpuwide_time_us() + 5000000L; + while (!predicate()) { + if (butil::cpuwide_time_us() >= deadline) { + return false; + } + bthread_usleep(1000); + } + return true; +} void read_op(bthread_rwlock_t* rw, int64_t sleep_us) { ASSERT_EQ(0, bthread_rwlock_rdlock(rw)); @@ -334,18 +349,19 @@ TEST(RWLockTest, writer_priority) { WriterPriorityArgs r2arg {&rw, &order, -1, 0}; // (2) Start a writer; it should park inside wrlock() because the read - // lock is held. Sleep long enough for it to fetch_add into - // writer_wait_count and reach the butex_wait on `lock_word'. + // lock is held. Observe its registration before starting a reader. bthread_t wth; ASSERT_EQ(0, bthread_start_urgent(&wth, nullptr, wp_writer_fn, &warg)); - bthread_usleep(50 * 1000); + EXPECT_TRUE(WaitForRWLockState([&] { + return reinterpret_cast*>(rw.writer_wait_count) + ->load(butil::memory_order_relaxed) == 1; + })); // (3) Now spawn a fresh reader. By writer-priority it MUST observe // writer_wait_count > 0 and park on it (NOT join the active read // lock). bthread_t r2th; ASSERT_EQ(0, bthread_start_urgent(&r2th, nullptr, wp_reader_fn, &r2arg)); - bthread_usleep(50 * 1000); // (4) Release the original read lock. The writer should win the race // and complete BEFORE the queued reader. @@ -396,14 +412,9 @@ TEST(RWLockTest, wrlock_failure_does_not_leak_writer_count) { // so a new reader MUST acquire the lock immediately. ASSERT_EQ(0, bthread_rwlock_unlock(&rw)); - timespec ts = butil::milliseconds_from_now(500); - butil::Timer t; - t.start(); - ASSERT_EQ(0, bthread_rwlock_timedrdlock(&rw, &ts)); - t.stop(); - EXPECT_LT(t.m_elapsed(), 100) - << "Reader was blocked for " << t.m_elapsed() << "ms; " - << "writer_wait_count was likely leaked by the cleanup path."; + EXPECT_EQ(0u, reinterpret_cast*>(rw.writer_wait_count) + ->load(butil::memory_order_relaxed)); + ASSERT_EQ(0, bthread_rwlock_tryrdlock(&rw)); ASSERT_EQ(0, bthread_rwlock_unlock(&rw)); ASSERT_EQ(0, bthread_rwlock_destroy(&rw)); @@ -518,17 +529,14 @@ TEST(RWLockTest, no_writer_starvation) { // Let the readers ramp up and saturate the lock. bthread_usleep(50 * 1000); - // A single writer must succeed within a generous budget. - butil::Timer t; - t.start(); - ASSERT_EQ(0, bthread_rwlock_wrlock(&rw)); - t.stop(); - - EXPECT_LT(t.m_elapsed(), 1000) - << "Writer starved for " << t.m_elapsed() << "ms under " - << R << " concurrent readers; writer-priority is broken."; - - ASSERT_EQ(0, bthread_rwlock_unlock(&rw)); + // A timed acquisition also makes the failure path reachable when the + // writer really starves, so readers can be stopped and joined safely. + timespec deadline = butil::seconds_from_now(10); + int rc = bthread_rwlock_timedwrlock(&rw, &deadline); + EXPECT_EQ(0, rc) << "Writer starved under concurrent readers"; + if (rc == 0) { + EXPECT_EQ(0, bthread_rwlock_unlock(&rw)); + } g_stopped = true; for (int i = 0; i < R; ++i) { @@ -726,6 +734,16 @@ class RwlockTaskRunner { } } + void WaitUntilBlocked() { + for (bthread_t tid : _tids) { + auto* meta = bthread::TaskGroup::address_meta(tid); + EXPECT_TRUE(WaitForRWLockState([&] { + return meta->current_waiter.load(butil::memory_order_acquire) + != nullptr; + })); + } + } + void Join() { for (size_t i = 0; i < _tids.size(); ++i) { bthread_join(_tids[i], nullptr); @@ -745,7 +763,7 @@ class RwlockTaskRunner { #define CHECK_RWLOCK_LOCKED_VALUE_EQUAL(mutex_name, value, expected_value) \ { \ std::unique_lock lock(mutex_name); \ - ASSERT_EQ(value, expected_value); \ + EXPECT_EQ(value, expected_value); \ } void RwlockLockingThreadFunc(bthread_rwlock_t* rw, bool read_lock, @@ -854,7 +872,10 @@ TEST(RWLockTest, boost_style_only_one_writer_permitted) { } task_runner.RunTask(); - bthread_usleep(200 * 1000); + EXPECT_TRUE(WaitForRWLockState([&] { + std::unique_lock lk(unblocked_count_mutex); + return unblocked_count >= 1; + })); CHECK_RWLOCK_LOCKED_VALUE_EQUAL(unblocked_count_mutex, unblocked_count, 1u); @@ -905,7 +926,7 @@ TEST(RWLockTest, boost_style_reader_blocks_writer) { &simultaneous_running_count, &max_simultaneous_running)); writer_runner.RunTask(); - bthread_usleep(100 * 1000); + writer_runner.WaitUntilBlocked(); CHECK_RWLOCK_LOCKED_VALUE_EQUAL(unblocked_count_mutex, unblocked_count, 1u); finish_lock.unlock(); @@ -945,7 +966,7 @@ TEST(RWLockTest, boost_style_unlocking_writer_unblocks_all_readers) { } task_runner.RunTask(); - bthread_usleep(100 * 1000); + task_runner.WaitUntilBlocked(); CHECK_RWLOCK_LOCKED_VALUE_EQUAL(unblocked_count_mutex, unblocked_count, 0u); ASSERT_EQ(0, bthread_rwlock_unlock(&rw)); @@ -1015,7 +1036,7 @@ TEST(RWLockTest, boost_style_unlocking_last_reader_only_unblocks_one_writer) { } writer_runner.RunTask(); - bthread_usleep(100 * 1000); + writer_runner.WaitUntilBlocked(); CHECK_RWLOCK_LOCKED_VALUE_EQUAL(unblocked_count_mutex, unblocked_count, reader_count); @@ -1027,7 +1048,7 @@ TEST(RWLockTest, boost_style_unlocking_last_reader_only_unblocks_one_writer) { unblocked_condition.wait(lk); } } - bthread_usleep(100 * 1000); + writer_runner.WaitUntilBlocked(); CHECK_RWLOCK_LOCKED_VALUE_EQUAL(unblocked_count_mutex, unblocked_count, reader_count + 1); diff --git a/test/bthread_timer_thread_unittest.cpp b/test/bthread_timer_thread_unittest.cpp index c784cf252d..f09ddc8b99 100644 --- a/test/bthread_timer_thread_unittest.cpp +++ b/test/bthread_timer_thread_unittest.cpp @@ -27,6 +27,20 @@ namespace { +// A generous deadline bounds failures, not the scheduler's response time. +// Predicates must read atomics or otherwise synchronize with the timer thread. +template +bool WaitUntil(Predicate predicate) { + int64_t deadline = butil::cpuwide_time_us() + 10000000L; + while (!predicate()) { + if (butil::cpuwide_time_us() >= deadline) { + return false; + } + usleep(1000); + } + return true; +} + long timespec_diff_us(const timespec& ts1, const timespec& ts2) { return (ts1.tv_sec - ts2.tv_sec) * 1000000L + (ts1.tv_nsec - ts2.tv_nsec) / 1000; @@ -53,16 +67,18 @@ class TimeKeeper { timespec current_time; clock_gettime(CLOCK_REALTIME, ¤t_time); if (_name) { - LOG(INFO) << "Run `" << _name << "' task_id=" << _task_id; + LOG(INFO) << "Run `" << _name << "'"; } else { - LOG(INFO) << "Run task_id=" << _task_id; + LOG(INFO) << "Run timer task"; } _run_times.push_back(current_time); - const int saved_sleep_ms = _sleep_ms; + _started.store(true, butil::memory_order_release); + int saved_sleep_ms = _sleep_ms.load(); if (saved_sleep_ms > 0) { timespec timeout = butil::milliseconds_to_timespec(saved_sleep_ms); bthread::futex_wait_private(&_sleep_ms, saved_sleep_ms, &timeout); } + _finished.store(true, butil::memory_order_release); } void wakeup() { @@ -85,11 +101,11 @@ class TimeKeeper { { ASSERT_TRUE(!_run_times.empty()); long diff = timespec_diff_us(_run_times[0], expect_run_time); - EXPECT_LE(labs(diff), 50000); + EXPECT_GE(diff, 0); } void expect_not_run() { - EXPECT_TRUE(_run_times.empty()); + EXPECT_FALSE(_started.load(butil::memory_order_acquire)); } static void routine(void *arg) @@ -98,12 +114,26 @@ class TimeKeeper { keeper->run(); } + bool wait_finished() { + return WaitUntil([this] { + return _finished.load(butil::memory_order_acquire); + }); + } + + bool wait_started() { + return WaitUntil([this] { + return _started.load(butil::memory_order_acquire); + }); + } + timespec _expect_run_time; bthread::TimerThread::TaskId _task_id; private: const char* _name; - int _sleep_ms; + butil::atomic _sleep_ms; + butil::atomic _started{false}; + butil::atomic _finished{false}; std::vector _run_times; }; @@ -111,38 +141,35 @@ TEST(TimerThreadTest, RunTasks) { bthread::TimerThread timer_thread; ASSERT_EQ(0, timer_thread.start(nullptr)); - timespec _2s_later = butil::seconds_from_now(2); + timespec _2s_later = butil::milliseconds_from_now(20); TimeKeeper keeper1(_2s_later, "keeper1"); keeper1.schedule(&timer_thread); - TimeKeeper keeper2(_2s_later, "keeper2"); // same time with keeper1 + TimeKeeper keeper2(butil::seconds_from_now(3600), "keeper2"); keeper2.schedule(&timer_thread); - timespec _1s_later = butil::seconds_from_now(1); + timespec _1s_later = butil::milliseconds_from_now(10); TimeKeeper keeper3(_1s_later, "keeper3"); keeper3.schedule(&timer_thread); - timespec _10s_later = butil::seconds_from_now(10); + timespec _10s_later = butil::seconds_from_now(3600); TimeKeeper keeper4(_10s_later, "keeper4"); keeper4.schedule(&timer_thread); TimeKeeper keeper5(_10s_later, "keeper5"); keeper5.schedule(&timer_thread); - // sleep 1 second, and unschedule task2 - LOG(INFO) << "Sleep 1s"; - sleep(1); - timer_thread.unschedule(keeper2._task_id); - timer_thread.unschedule(keeper4._task_id); + ASSERT_EQ(0, timer_thread.unschedule(keeper2._task_id)); + ASSERT_EQ(0, timer_thread.unschedule(keeper4._task_id)); timespec old_time = { 0, 0 }; TimeKeeper keeper6(old_time, "keeper6"); + timespec keeper6_addtime = butil::seconds_from_now(0); keeper6.schedule(&timer_thread); - const timespec keeper6_addtime = butil::seconds_from_now(0); - // sleep 10 seconds and stop. - LOG(INFO) << "Sleep 2s"; - sleep(2); + ASSERT_TRUE(keeper1.wait_started()); + ASSERT_TRUE(keeper3.wait_started()); + ASSERT_TRUE(keeper6.wait_started()); LOG(INFO) << "Stop timer_thread"; butil::Timer tm; tm.start(); @@ -150,7 +177,7 @@ TEST(TimerThreadTest, RunTasks) { tm.stop(); // stop_and_join() should wake the timer thread instead of waiting for the // tasks scheduled 10 seconds later. Allow for CI runner scheduling delays. - ASSERT_LT(tm.m_elapsed(), 1000); + ASSERT_LT(tm.m_elapsed(), 10000); // verify all runs in expected time range. keeper1.expect_first_run(); @@ -215,20 +242,21 @@ TEST(TimerThreadTest, schedule_and_unschedule_in_task) { bthread::TimerThread timer_thread; timespec past_time = { 0, 0 }; timespec future_time = { std::numeric_limits::max(), 0 }; - const timespec _500ms_after = butil::milliseconds_from_now(500); + timespec _500ms_after = butil::milliseconds_from_now(500); TimeKeeper keeper1(future_time, "keeper1"); TimeKeeper keeper2(past_time, "keeper2"); TimeKeeper keeper3(past_time, "keeper3"); TimeKeeper keeper4(past_time, "keeper4"); - TimeKeeper keeper5(_500ms_after, "keeper5", 10000/*10s*/); + TimeKeeper keeper5(_500ms_after, "keeper5", 60000); ASSERT_EQ(0, timer_thread.start(nullptr)); keeper1.schedule(&timer_thread); // start keeper1 - keeper3.schedule(&timer_thread); // start keeper3 timespec keeper3_addtime = butil::seconds_from_now(0); + keeper3.schedule(&timer_thread); // start keeper3 keeper5.schedule(&timer_thread); // start keeper5 - sleep(1); // let keeper1/3/5 run + ASSERT_TRUE(keeper3.wait_started()); + ASSERT_TRUE(keeper5.wait_started()); TestTask test_task1(&timer_thread, &keeper1, &keeper2, 0); timer_thread.schedule(TestTask::routine, &test_task1, past_time); @@ -236,7 +264,6 @@ TEST(TimerThreadTest, schedule_and_unschedule_in_task) { TestTask test_task2(&timer_thread, &keeper3, &keeper4, -1); timer_thread.schedule(TestTask::routine, &test_task2, past_time); - sleep(1); // test_task1/2 should be both blocked by keeper5. keeper2.expect_not_run(); keeper4.expect_not_run(); @@ -246,7 +273,8 @@ TEST(TimerThreadTest, schedule_and_unschedule_in_task) { // wake up keeper5 to let test_task1/2 run. keeper5.wakeup(); - sleep(1); + ASSERT_TRUE(keeper2.wait_started()); + ASSERT_TRUE(keeper4.wait_started()); timer_thread.stop_and_join(); timespec finish_time; @@ -293,9 +321,9 @@ TEST(TimerThreadTest, sweep_unscheduled_tasks_in_heap) { ASSERT_EQ(0, timer_thread.start(nullptr)); // Run far enough in the future that these tasks never fire on their own. - const timespec far = butil::seconds_from_now(100000); - const size_t kBatch = 2000; - const size_t kRounds = 20; + timespec far = butil::seconds_from_now(100000); + size_t kBatch = 2000; + size_t kRounds = 20; int64_t max_pending = 0; for (size_t r = 0; r < kRounds; ++r) { @@ -308,7 +336,10 @@ TEST(TimerThreadTest, sweep_unscheduled_tasks_in_heap) { // buckets, so the far tasks above land in the heap (alive). timer_thread.schedule(noop_routine, nullptr, butil::milliseconds_from_now(1)); - usleep(20000); // let the timer thread consume the buckets + ASSERT_TRUE(WaitUntil([&] { + return timer_thread._npending.load(butil::memory_order_relaxed) >= + static_cast(kBatch); + })); // Now unschedule the far tasks: they become dead-in-heap, exactly the // case that used to linger until run_time. @@ -319,11 +350,18 @@ TEST(TimerThreadTest, sweep_unscheduled_tasks_in_heap) { // sweep that reclaims the dead tasks. timer_thread.schedule(noop_routine, nullptr, butil::milliseconds_from_now(1)); - usleep(20000); + // A callback acknowledgement ensures a timer pass completed. Heap + // reclamation is checked across rounds below (sweeps are amortized). + TimeKeeper consumed(butil::seconds_from_now(0)); + consumed.schedule(&timer_thread); + if (!consumed.wait_finished()) { + timer_thread.stop_and_join(); + FAIL() << "Timer did not consume the wakeup task"; + } // Read the internal heap size directly (brpc tests are built with // -fno-access-control, so no public accessor is needed). - const int64_t pending = + int64_t pending = timer_thread._npending.load(butil::memory_order_relaxed); LOG(INFO) << "round=" << r << " pending=" << pending; max_pending = std::max(max_pending, pending); @@ -354,13 +392,15 @@ TEST(TimerThreadTest, periodic_wakeup_drains_buckets) { // (with even later run_times) are never the "earliest" and thus never wake // the timer via schedule() -- only the periodic wakeup can drain them. timer_thread.schedule(noop_routine, nullptr, butil::seconds_from_now(3600)); - usleep(100000); // let the anchor be consumed into the heap + EXPECT_TRUE(WaitUntil([&] { + return timer_thread._npending.load(butil::memory_order_relaxed) == 1; + })); // Only the anchor is in the heap so far. ASSERT_EQ(1, timer_thread._npending.load(butil::memory_order_relaxed)); // Pile far tasks with strictly-increasing run_times into the buckets. None // of these wake the timer. - const int kN = 2000; + int kN = 2000; for (int i = 0; i < kN; ++i) { timer_thread.schedule(noop_routine, nullptr, butil::seconds_from_now(3600 + 1 + i)); @@ -369,8 +409,10 @@ TEST(TimerThreadTest, periodic_wakeup_drains_buckets) { // Without the periodic wakeup the timer would stay asleep (nearest task is // an hour away) and these would sit in the buckets, unconsumed. With it, // they are pulled into the heap within a few wakeup intervals. - usleep(300000); // several 50ms intervals - const int64_t pending = + EXPECT_TRUE(WaitUntil([&] { + return timer_thread._npending.load(butil::memory_order_relaxed) == kN + 1; + })); + int64_t pending = timer_thread._npending.load(butil::memory_order_relaxed); LOG(INFO) << "pending after bucket fill = " << pending << " (scheduled " << kN << " + 1 anchor)"; diff --git a/test/bthread_unittest.cpp b/test/bthread_unittest.cpp index c86bc83e8a..635b7e9a9a 100644 --- a/test/bthread_unittest.cpp +++ b/test/bthread_unittest.cpp @@ -501,21 +501,32 @@ TEST_F(BthreadTest, start_latency_when_high_idle) { } void* sleep_for_awhile_with_sleep(void* arg) { - bthread_usleep((intptr_t)arg); + int rc = bthread_usleep((intptr_t)arg); + int error = errno; + EXPECT_EQ(-1, rc); + EXPECT_EQ(ESTOP, error); return nullptr; } TEST_F(BthreadTest, stop_sleep) { bthread_t th; ASSERT_EQ(0, bthread_start_urgent( - &th, nullptr, sleep_for_awhile_with_sleep, (void*)1000000L)); - butil::Timer tm; - tm.start(); - bthread_usleep(10000); + &th, nullptr, sleep_for_awhile_with_sleep, (void*)60000000L)); + auto* meta = bthread::TaskGroup::address_meta(th); + int64_t deadline = butil::cpuwide_time_us() + 5000000L; + bool sleeping = false; + do { + pthread_spin_lock(&meta->version_lock); + sleeping = (meta->current_sleep != 0); + pthread_spin_unlock(&meta->version_lock); + if (sleeping) { + break; + } + bthread_usleep(1000); + } while (butil::cpuwide_time_us() < deadline); + ASSERT_TRUE(sleeping); ASSERT_EQ(0, bthread_stop(th)); ASSERT_EQ(0, bthread_join(th, nullptr)); - tm.stop(); - ASSERT_LE(labs(tm.m_elapsed() - 10), 10); } TEST_F(BthreadTest, bthread_exit) { diff --git a/test/bthread_work_stealing_queue_unittest.cpp b/test/bthread_work_stealing_queue_unittest.cpp index c6738e8208..6b33d53834 100644 --- a/test/bthread_work_stealing_queue_unittest.cpp +++ b/test/bthread_work_stealing_queue_unittest.cpp @@ -25,7 +25,7 @@ namespace { typedef size_t value_type; -bool g_stop = false; +butil::atomic g_stop(false); const size_t N = 1024*512; const size_t CAP = 8; pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; @@ -85,6 +85,7 @@ void* pop_thread(void* arg) { TEST(WSQTest, sanity) { + g_stop = false; bthread::WorkStealingQueue q; ASSERT_EQ(0, q.init(CAP)); pthread_t rth[8];