This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 8ef336ea Stabilize timing-sensitive bthread tests and fix the butex
interruption race (#3545)
8ef336ea is described below
commit 8ef336ea6cefcf16d4b1264f1f70af2cc677846b
Author: Bright Chen <[email protected]>
AuthorDate: Thu Sep 24 17:53:40 2026 +0800
Stabilize timing-sensitive bthread tests and fix the butex interruption
race (#3545)
* Stabilize timing-sensitive bthread tests and fix the butex interruption
race
* Fix pthread_kill get ESRCH
* Use a non-fatal assertion
* Fix some issues
* Check and set task->interrupted under task->version_lock
* Fix butex_requeue UT
---
src/bthread/butex.cpp | 12 +-
test/brpc_socket_unittest.cpp | 19 +-
test/bthread_butex_multi_tag_unittest.cpp | 63 ++++-
test/bthread_butex_unittest.cpp | 387 +++++++++++++++++---------
test/bthread_cond_unittest.cpp | 2 +-
test/bthread_fd_unittest.cpp | 250 ++++++++++-------
test/bthread_futex_unittest.cpp | 117 ++++++--
test/bthread_mutex_unittest.cpp | 10 +-
test/bthread_rwlock_unittest.cpp | 83 +++---
test/bthread_timer_thread_unittest.cpp | 128 ++++++---
test/bthread_unittest.cpp | 26 +-
test/bthread_work_stealing_queue_unittest.cpp | 3 +-
12 files changed, 735 insertions(+), 365 deletions(-)
diff --git a/src/bthread/butex.cpp b/src/bthread/butex.cpp
index 790da145..df0562bd 100644
--- a/src/bthread/butex.cpp
+++ b/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;
}
diff --git a/test/brpc_socket_unittest.cpp b/test/brpc_socket_unittest.cpp
index 30e4e763..6c3211c1 100644
--- a/test/brpc_socket_unittest.cpp
+++ b/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());
{
diff --git a/test/bthread_butex_multi_tag_unittest.cpp
b/test/bthread_butex_multi_tag_unittest.cpp
index e41fefdd..31be2afe 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<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
diff --git a/test/bthread_butex_unittest.cpp b/test/bthread_butex_unittest.cpp
index 924f30cb..a626e3b1 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 <condition_variable>
+#include <mutex>
#include <gtest/gtest.h>
#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<uint32_t>();
ASSERT_TRUE(butex);
@@ -43,6 +46,7 @@ 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 {
@@ -52,7 +56,14 @@ struct JoinSleepArg {
};
void* sleeper(void* arg) {
- bthread_usleep((uint64_t)arg);
+ JoinSleepArg* a = static_cast<JoinSleepArg*>(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<int64_t>(a->sleep_us));
+ a->finished.store(true, butil::memory_order_release);
return nullptr;
}
@@ -133,14 +144,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;
@@ -149,9 +163,9 @@ void* waiter(void* arg) {
TEST(ButexTest, sanity) {
const size_t N = 5;
WaiterArg args[N * 4];
- pthread_t t1, t2;
- butil::atomic<int>* b1 =
- bthread::butex_create_checked<butil::atomic<int> >();
+ pthread_t pthreads[2 * N];
+ bthread_t bthreads[2 * N];
+ butil::atomic<int>* b1 = bthread::butex_create_checked<butil::atomic<int>
>();
ASSERT_TRUE(b1);
bthread::butex_destroy(b1);
@@ -159,17 +173,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) {
@@ -179,34 +193,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<std::mutex> lock(_mutex);
+ _cond.wait(lock, [this] { return _open; });
+ }
+
+ void signal() {
+ std::lock_guard<std::mutex> 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<ButexWaitArg*>(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);
@@ -221,11 +282,11 @@ TEST(ButexTest, wait_without_stop) {
int* butex = bthread::butex_create_checked<int>();
*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();
@@ -233,7 +294,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);
}
@@ -241,23 +303,24 @@ TEST(ButexTest, wait_without_stop) {
TEST(ButexTest, stop_after_running) {
int* butex = bthread::butex_create_checked<int>();
*butex = 7;
- 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);
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 };
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));
// ASSERT_TRUE(bthread::get_task_control()->
// timer_thread()._idset.empty());
ASSERT_EQ(EINVAL, bthread_stop(th));
- }
+ }
bthread::butex_destroy(butex);
}
@@ -267,13 +330,18 @@ TEST(ButexTest, stop_before_running) {
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 };
+ 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));
ASSERT_EQ(EINVAL, bthread_stop(th));
@@ -281,132 +349,168 @@ TEST(ButexTest, stop_before_running) {
bthread::butex_destroy(butex);
}
+struct JoinWaiterArg {
+ bthread_t tid;
+ TestGate entering_join;
+ butil::atomic<bool> finished{false};
+};
+
void* join_the_waiter(void* arg) {
- EXPECT_EQ(0, bthread_join((bthread_t)arg, nullptr));
+ JoinWaiterArg* a = static_cast<JoinWaiterArg*>(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<int>();
*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<StopSleepArg*>(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";
+ }
}
- // ASSERT_TRUE(bthread::get_task_control()->
- // timer_thread()._idset.empty());
+ EXPECT_EQ(0, bthread_stop(th));
+ before_sleep.signal();
+ arg.allow_exit.signal();
+ if (stop_before_sleep) {
+ bthread_flush();
+ }
+ 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<bool> 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<SignalArg*>(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<SignalArg*>(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 +518,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<int>* butex =
@@ -423,23 +527,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 85181691..5fd0a70d 100644
--- a/test/bthread_cond_unittest.cpp
+++ b/test/bthread_cond_unittest.cpp
@@ -145,7 +145,6 @@ std::atomic<int> 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 02c579c1..a730f2d7 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<bool> done{false};
+ int result = 0;
+ int error = 0;
+};
+
void* epoll_waiter(void* arg) {
+ EpollWaitArg* a = static_cast<EpollWaitArg*>(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,74 @@ 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<FDWaitArg*>(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);
+ }
+ // Keep cleanup reachable even if the waiter did not register in time.
+ EXPECT_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 +621,76 @@ 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<sockaddr*>(&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<sockaddr*>(&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<sockaddr*>(&address), length, &deadline);
+ int error = errno;
+ ASSERT_EQ(-1, rc);
+ // tcp_abort_on_overflow may turn an accept-queue overflow into an RST.
+ ASSERT_TRUE(error == ETIMEDOUT || error == ECONNREFUSED)
+ << "errno=" << 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 a9742755..ce32356e 100644
--- a/test/bthread_futex_unittest.cpp
+++ b/test/bthread_futex_unittest.cpp
@@ -20,6 +20,7 @@
#include <stdio.h>
#include <signal.h>
#include <gtest/gtest.h>
+#include <vector>
#include "butil/time.h"
#include "butil/macros.h"
#include "butil/errno.h"
@@ -30,7 +31,7 @@
#include <bthread/processor.h>
namespace {
-volatile bool stop = false;
+butil::atomic<bool> stop(false);
butil::atomic<int> 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<int> 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;
@@ -112,31 +115,91 @@ TEST(FutexTest, futex_wake_before_wait) {
ASSERT_EQ(ETIMEDOUT, errno);
}
-void* dummy_waiter(void* lock) {
- bthread::futex_wait_private(lock, 0, nullptr);
+struct DummyWaiterArg {
+ butil::atomic<int>* lock;
+ butil::atomic<int>* registered;
+ butil::atomic<bool>* cleaning_up;
+};
+
+void* dummy_waiter(void* void_arg) {
+ DummyWaiterArg* arg = static_cast<DummyWaiterArg*>(void_arg);
+ timespec timeout = butil::seconds_to_timespec(10);
+ // Publish readiness before entering the wait so the controller does not
+ // start measuring wakeups while threads are still being created.
+ arg->registered->fetch_add(1, butil::memory_order_release);
+ int rc;
+ do {
+ rc = bthread::futex_wait_private(arg->lock, 0, &timeout);
+ } while (rc != 0 && errno == EINTR);
+ if (arg->cleaning_up->load(butil::memory_order_acquire) &&
+ rc == -1 && errno == EWOULDBLOCK) {
+ return nullptr;
+ }
+ 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<int> lock1(0);
+ butil::atomic<int> registered(0);
+ butil::atomic<bool> cleaning_up(false);
+ DummyWaiterArg arg = { &lock1, ®istered, &cleaning_up };
+ std::vector<pthread_t> threads;
+ for (size_t i = 0; i < 1000; ++i) {
+ pthread_t th;
+ if (pthread_create(&th, nullptr, dummy_waiter, &arg) != 0) {
+ break;
+ }
+ threads.push_back(th);
+ }
+ ASSERT_FALSE(threads.empty());
+ size_t N = threads.size();
+ const int64_t registration_deadline =
+ butil::cpuwide_time_us() + 10000000L;
+ while (registered.load(butil::memory_order_acquire) !=
+ static_cast<int>(N) &&
+ butil::cpuwide_time_us() < registration_deadline) {
+ usleep(1000);
+ }
+ const bool all_registered =
+ registered.load(butil::memory_order_acquire) == static_cast<int>(N);
+
int nwakeup = 0;
+ int64_t wake_ns = 0;
+ const int64_t wake_deadline = butil::cpuwide_time_us() + 10000000L;
butil::Timer tm;
- tm.start();
- for (size_t i = 0; i < N; ++i) {
- nwakeup += bthread::futex_wake_private(&lock1, 1);
+ if (all_registered) {
+ while (static_cast<size_t>(nwakeup) < N &&
+ butil::cpuwide_time_us() < wake_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);
+ }
+ }
+ }
+ // Also release waiters if a wake assertion fails; a wake alone is not
+ // persistent, and a worker that races with this store gets EWOULDBLOCK.
+ cleaning_up.store(true, butil::memory_order_release);
+ lock1.store(1);
+ bthread::futex_wake_private(&lock1, INT_MAX);
+ for (pthread_t th : threads) {
+ EXPECT_EQ(0, pthread_join(th, nullptr));
+ }
+ EXPECT_TRUE(all_registered)
+ << "Timed out waiting for all futex waiters to register";
+ EXPECT_EQ(N, static_cast<size_t>(nwakeup));
+ if (nwakeup != 0) {
+ printf("N=%lu, futex_wake a thread = %" PRId64 "ns\n", N,
+ wake_ns / nwakeup);
}
- tm.stop();
- printf("N=%lu, futex_wake a thread = %" PRId64 "ns\n", N, tm.n_elapsed() /
N);
- ASSERT_EQ(N, (size_t)nwakeup);
- 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 +254,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 98494421..eca329e0 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<butil::atomic<unsigned>*>(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.
+ EXPECT_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 3747c29f..9031a579 100644
--- a/test/bthread_rwlock_unittest.cpp
+++ b/test/bthread_rwlock_unittest.cpp
@@ -21,6 +21,8 @@
#include <bthread/rwlock.h>
#include <bthread/condition_variable.h>
#include <functional>
+#include "bthread/task_group.h"
+#include "bthread/task_meta.h"
#include <vector>
namespace {
@@ -210,8 +212,21 @@ TEST(RWLockTest, cpp_wrapper) {
}
}
-bool g_started = false;
-bool g_stopped = false;
+butil::atomic<bool> g_started(false);
+butil::atomic<bool> g_stopped(false);
+
+// Read only synchronized predicates; deadlines bound failures, not latency.
+template <typename Predicate>
+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<butil::atomic<unsigned>*>(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<butil::atomic<unsigned>*>(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 @@ public:
}
}
+ 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 @@ private:
#define CHECK_RWLOCK_LOCKED_VALUE_EQUAL(mutex_name, value, expected_value) \
{ \
std::unique_lock<bthread::Mutex> 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<bthread::Mutex> 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 c784cf25..13153fe4 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 <typename Predicate>
+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 @@ public:
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,14 @@ public:
{
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);
+ // Keep a generous bound to catch a stalled timer without depending on
+ // narrow scheduler-latency assumptions.
+ EXPECT_LT(diff, 10000000L);
}
void expect_not_run() {
- EXPECT_TRUE(_run_times.empty());
+ EXPECT_FALSE(_started.load(butil::memory_order_acquire));
}
static void routine(void *arg)
@@ -98,12 +117,26 @@ public:
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<int> _sleep_ms;
+ butil::atomic<bool> _started{false};
+ butil::atomic<bool> _finished{false};
std::vector<timespec> _run_times;
};
@@ -111,46 +144,43 @@ TEST(TimerThreadTest, RunTasks) {
bthread::TimerThread timer_thread;
ASSERT_EQ(0, timer_thread.start(nullptr));
- timespec _2s_later = butil::seconds_from_now(2);
- TimeKeeper keeper1(_2s_later, "keeper1");
+ timespec time_20ms_later = butil::milliseconds_from_now(20);
+ TimeKeeper keeper1(time_20ms_later, "keeper1");
keeper1.schedule(&timer_thread);
- TimeKeeper keeper2(_2s_later, "keeper2"); // same time with keeper1
+ timespec time_1h_later = butil::seconds_from_now(3600);
+ TimeKeeper keeper2(time_1h_later, "keeper2");
keeper2.schedule(&timer_thread);
- timespec _1s_later = butil::seconds_from_now(1);
- TimeKeeper keeper3(_1s_later, "keeper3");
+ timespec time_10ms_later = butil::milliseconds_from_now(10);
+ TimeKeeper keeper3(time_10ms_later, "keeper3");
keeper3.schedule(&timer_thread);
- timespec _10s_later = butil::seconds_from_now(10);
- TimeKeeper keeper4(_10s_later, "keeper4");
+ TimeKeeper keeper4(time_1h_later, "keeper4");
keeper4.schedule(&timer_thread);
- TimeKeeper keeper5(_10s_later, "keeper5");
+ TimeKeeper keeper5(time_1h_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();
timer_thread.stop_and_join();
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);
+ // tasks scheduled an hour later. Allow for CI runner scheduling delays.
+ ASSERT_LT(tm.m_elapsed(), 10000);
// verify all runs in expected time range.
keeper1.expect_first_run();
@@ -215,20 +245,22 @@ TEST(TimerThreadTest, schedule_and_unschedule_in_task) {
bthread::TimerThread timer_thread;
timespec past_time = { 0, 0 };
timespec future_time = { std::numeric_limits<int>::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
+ // Preserve timer-thread cleanup even when a callback starts late.
+ EXPECT_TRUE(keeper3.wait_started());
+ EXPECT_TRUE(keeper5.wait_started());
TestTask test_task1(&timer_thread, &keeper1, &keeper2, 0);
timer_thread.schedule(TestTask::routine, &test_task1, past_time);
@@ -236,7 +268,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 +277,8 @@ TEST(TimerThreadTest, schedule_and_unschedule_in_task) {
// wake up keeper5 to let test_task1/2 run.
keeper5.wakeup();
- sleep(1);
+ EXPECT_TRUE(keeper2.wait_started());
+ EXPECT_TRUE(keeper4.wait_started());
timer_thread.stop_and_join();
timespec finish_time;
@@ -293,9 +325,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 +340,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<int64_t>(kBatch);
+ }));
// Now unschedule the far tasks: they become dead-in-heap, exactly the
// case that used to linger until run_time.
@@ -319,11 +354,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 +396,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 +413,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 c86bc83e..6a90ea56 100644
--- a/test/bthread_unittest.cpp
+++ b/test/bthread_unittest.cpp
@@ -501,21 +501,33 @@ 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);
+ // bthread_stop is persistent, so cleanup is safe even before registration.
+ EXPECT_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 c6738e82..6b33d538 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<bool> 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<value_type> q;
ASSERT_EQ(0, q.init(CAP));
pthread_t rth[8];
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
