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 c79b3ffd Fix two races between butex_wait() and TaskGroup::interrupt()
(#3551)
c79b3ffd is described below
commit c79b3ffd67b599deb8c0badd0dc75c854a28d0fd
Author: Bright Chen <[email protected]>
AuthorDate: Tue Sep 22 10:38:21 2026 +0800
Fix two races between butex_wait() and TaskGroup::interrupt() (#3551)
* Fix two races between butex_wait() and TaskGroup::interrupt()
* Fix self-deadlock in wait_for_butex() when TimerThread::schedule() fails
---
src/bthread/butex.cpp | 132 +++++++++++++++++++++++++++++++++-----------
src/bthread/task_group.cpp | 4 +-
src/bthread/task_tracer.cpp | 40 ++++++++------
src/bthread/task_tracer.h | 1 +
4 files changed, 125 insertions(+), 52 deletions(-)
diff --git a/src/bthread/butex.cpp b/src/bthread/butex.cpp
index 63920ca9..790da145 100644
--- a/src/bthread/butex.cpp
+++ b/src/bthread/butex.cpp
@@ -573,26 +573,62 @@ void wait_for_butex(void* arg) {
BAIDU_SCOPED_LOCK(b->waiter_lock);
if (b->value.load(butil::memory_order_relaxed) != bw->expected_value) {
bw->waiter_state = WAITER_STATE_UNMATCHEDVALUE;
- } else if (bw->waiter_state == WAITER_STATE_READY/*1*/ &&
- !bw->task_meta->interrupted) {
- if (args->prepend) {
- b->waiters.Prepend(bw);
- } else {
- b->waiters.Append(bw);
- }
- bw->container.store(b, butil::memory_order_relaxed);
+ } else {
+ // Checking `interrupted` and publishing `bw->container` must be
+ // atomic with respect to TaskGroup::interrupt(), which sets
+ // `interrupted` and consumes `current_waiter` under the same
+ // `version_lock`. Otherwise interrupt() may consume `bw` in
between
+ // and its erase_from_butex() does nothing because `container` is
+ // still nullptr, leaving this bthread queued but never woken up.
+ // `container` cannot be published upfront: it must stay nullptr
until
+ // the bthread is off its stack, see the comment after this block.
+ bool suspend = false;
+ {
+ // Only the interrupted check, the enqueue and the container
+ // store belong under `version_lock', so that they are atomic
+ // w.r.t. TaskGroup::interrupt(). The tracer update is kept
here
+ // too because set_status_unsafe() takes no lock and this makes
+ // the transition to SUSPENDED atomic w.r.t. TraceImpl().
+ BAIDU_SCOPED_LOCK(bw->task_meta->version_lock);
+ if (bw->waiter_state == WAITER_STATE_READY/*1*/ &&
+ !bw->task_meta->interrupted) {
+ if (args->prepend) {
+ b->waiters.Prepend(bw);
+ } else {
+ b->waiters.Append(bw);
+ }
+ bw->container.store(b, butil::memory_order_relaxed);
#ifdef BRPC_BTHREAD_TRACER
- bw->control->_task_tracer.set_status(TASK_STATUS_SUSPENDED,
bw->task_meta);
+ TaskTracer::set_status_unsafe(TASK_STATUS_SUSPENDED,
bw->task_meta);
#endif // BRPC_BTHREAD_TRACER
- if (bw->abstime != nullptr) {
+ suspend = true;
+ }
+ }
+ if (suspend && bw->abstime != nullptr) {
bw->sleep_id = get_global_timer_thread()->schedule(
erase_from_butex_and_wakeup, bw, *bw->abstime);
if (!bw->sleep_id) { // TimerThread stopped.
+ // No timer will ever fire, so `bw` must not stay queued.
+ // CAUTION: erase_from_butex_and_wakeup() must NOT be
called
+ // here. It takes `waiter_lock` (already held) and then,
via
+ // TaskGroup::ready_to_run{,_remote}() ->
+ // TaskTracer::set_status(), `version_lock` as well. Both
are
+ // non-recursive, so calling it self-deadlocks.
errno = ESTOP;
- erase_from_butex_and_wakeup(bw);
+ bw->RemoveFromList();
+ bw->container.store(nullptr, butil::memory_order_relaxed);
+ bw->waiter_state = WAITER_STATE_TIMEDOUT;
+ suspend = false;
}
}
- return;
+ if (suspend) {
+ return;
+ }
+ // Not suspended: the bthread has already been switched out by
+ // set_remained()/sched() in butex_wait(), so nobody else will run
+ // it. Fall through to re-schedule it below. This covers value
+ // unmatched, already timed out (waiter_state != READY), already
+ // interrupted, and the ESTOP path above.
}
}
@@ -619,33 +655,64 @@ static int butex_wait_from_pthread(TaskGroup* g, Butex*
b, int expected_value,
TaskMeta* task = nullptr;
ButexPthreadWaiter pw;
pw.tid = 0;
+ // Initialize `container` to nullptr before `pw` is published to
+ // `current_waiter` below: a concurrent TaskGroup::interrupt() may consume
+ // `pw` and call erase_from_butex() on it, which must observe either
nullptr
+ // (a no-op) or a valid Butex, never stack garbage. `container` stays
nullptr
+ // until `pw` is queued.
+ pw.container.store(nullptr, butil::memory_order_relaxed);
pw.sig.store(PTHREAD_NOT_SIGNALLED, butil::memory_order_relaxed);
int rc = 0;
-
+
if (g) {
task = g->current_task();
task->current_waiter.store(&pw, butil::memory_order_release);
}
- b->waiter_lock.lock();
- if (b->value.load(butil::memory_order_relaxed) != expected_value) {
- b->waiter_lock.unlock();
- errno = EWOULDBLOCK;
- rc = -1;
- } else if (task != nullptr && task->interrupted) {
- b->waiter_lock.unlock();
- // Race with set and may consume multiple interruptions, which are OK.
- task->interrupted = false;
- errno = EINTR;
- rc = -1;
- } else {
- if (prepend) {
- b->waiters.Prepend(&pw);
+ bool queued = false;
+ {
+ BAIDU_SCOPED_LOCK(b->waiter_lock);
+ if (b->value.load(butil::memory_order_relaxed) != expected_value) {
+ errno = EWOULDBLOCK;
+ rc = -1;
+ } else if (task != nullptr) {
+ // Checking `interrupted` and publishing `container` must be atomic
+ // with respect to TaskGroup::interrupt(), which sets `interrupted`
+ // and consumes `current_waiter` under the same version_lock.
+ // Otherwise interrupt() may consume `pw' in between and its
+ // erase_from_butex() does nothing because `container' is still
+ // nullptr, leaving `pw` queued but never woken up.
+ BAIDU_SCOPED_LOCK(task->version_lock);
+ if (task->interrupted) {
+ // Already interrupted before queueing: consume it and return
+ // EINTR without blocking. An interruption that arrives after
+ // `pw` is queued is handled by the epilogue below instead.
+ // Race with set and may consume multiple interruptions,
+ // which are OK.
+ task->interrupted = false;
+ errno = EINTR;
+ rc = -1;
+ } else {
+ if (prepend) {
+ b->waiters.Prepend(&pw);
+ } else {
+ b->waiters.Append(&pw);
+ }
+ pw.container.store(b, butil::memory_order_relaxed);
+ queued = true;
+ }
} else {
- b->waiters.Append(&pw);
+ // A non-bthread pthread cannot be interrupted, so there is no race
+ // with interrupt() and no need to hold `version_lock` here.
+ if (prepend) {
+ b->waiters.Prepend(&pw);
+ } else {
+ b->waiters.Append(&pw);
+ }
+ pw.container.store(b, butil::memory_order_relaxed);
+ queued = true;
}
- pw.container.store(b, butil::memory_order_relaxed);
- b->waiter_lock.unlock();
-
+ }
+ if (queued) {
#ifdef SHOW_BTHREAD_BUTEX_WAITER_COUNT_IN_VARS
bvar::Adder<int64_t>& num_waiters = butex_waiter_count();
num_waiters << 1;
@@ -655,7 +722,8 @@ static int butex_wait_from_pthread(TaskGroup* g, Butex* b,
int expected_value,
num_waiters << -1;
#endif
}
- if (task) {
+
+ if (task != nullptr) {
// If current_waiter is nullptr, TaskGroup::interrupt() is running and
// using pw, spin until current_waiter != nullptr.
BT_LOOP_WHEN(task->current_waiter.exchange(
diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp
index 0503fa4f..3051c07a 100644
--- a/src/bthread/task_group.cpp
+++ b/src/bthread/task_group.cpp
@@ -1048,7 +1048,7 @@ int TaskGroup::usleep(TaskGroup** pg, uint64_t
timeout_us) {
bool erase_from_butex_because_of_interruption(ButexWaiter* bw);
static int interrupt_and_consume_waiters(
- bthread_t tid, ButexWaiter** pw, uint64_t* sleep_id) {
+ bthread_t tid, ButexWaiter** bw, uint64_t* sleep_id) {
TaskMeta* const m = TaskGroup::address_meta(tid);
if (m == nullptr) {
return EINVAL;
@@ -1056,7 +1056,7 @@ static int interrupt_and_consume_waiters(
const uint32_t given_ver = get_version(tid);
BAIDU_SCOPED_LOCK(m->version_lock);
if (given_ver == *m->version_butex) {
- *pw = m->current_waiter.exchange(nullptr, butil::memory_order_acquire);
+ *bw = m->current_waiter.exchange(nullptr, butil::memory_order_acquire);
*sleep_id = m->current_sleep;
m->current_sleep = 0; // only one stopper gets the sleep_id
m->interrupted = true;
diff --git a/src/bthread/task_tracer.cpp b/src/bthread/task_tracer.cpp
index 031ad793..afb1642d 100644
--- a/src/bthread/task_tracer.cpp
+++ b/src/bthread/task_tracer.cpp
@@ -161,27 +161,10 @@ bool TaskTracer::Init() {
}
void TaskTracer::set_status(TaskStatus s, TaskMeta* m) {
- CHECK_NE(TASK_STATUS_RUNNING, s) << "Use `set_running_status' instead";
- CHECK_NE(TASK_STATUS_END, s) << "Use `set_end_status_unsafe' instead";
-
bool tracing = false;
{
BAIDU_SCOPED_LOCK(m->version_lock);
- if (TASK_STATUS_UNKNOWN == m->status && TASK_STATUS_JUMPING == s) {
- // Do not update status for jumping when bthread is ending.
- return;
- }
-
- tracing = m->traced;
- // bthread is scheduled for the first time.
- if (TASK_STATUS_READY == s && nullptr == m->stack) {
- m->status = TASK_STATUS_FIRST_READY;
- } else {
- m->status = s;
- }
- if (TASK_STATUS_CREATED == s) {
- m->worker_tid = pthread_t{};
- }
+ tracing = set_status_unsafe(s, m);
}
// Make sure bthread does not jump stack when it is being traced.
@@ -190,6 +173,27 @@ void TaskTracer::set_status(TaskStatus s, TaskMeta* m) {
}
}
+bool TaskTracer::set_status_unsafe(TaskStatus s, TaskMeta* m) {
+ CHECK_NE(TASK_STATUS_RUNNING, s) << "Use `set_running_status' instead";
+ CHECK_NE(TASK_STATUS_END, s) << "Use `set_end_status_unsafe' instead";
+
+ if (TASK_STATUS_UNKNOWN == m->status && TASK_STATUS_JUMPING == s) {
+ // Do not update status for jumping when bthread is ending.
+ return false;
+ }
+
+ // A bthread is scheduled for the first time.
+ if (TASK_STATUS_READY == s && m->stack == nullptr) {
+ m->status = TASK_STATUS_FIRST_READY;
+ } else {
+ m->status = s;
+ }
+ if (TASK_STATUS_CREATED == s) {
+ m->worker_tid = pthread_t{};
+ }
+ return m->traced;
+}
+
void TaskTracer::set_running_status(pthread_t worker_tid, TaskMeta* m) {
BAIDU_SCOPED_LOCK(m->version_lock);
m->worker_tid = worker_tid;
diff --git a/src/bthread/task_tracer.h b/src/bthread/task_tracer.h
index 8844413a..2bf4099b 100644
--- a/src/bthread/task_tracer.h
+++ b/src/bthread/task_tracer.h
@@ -39,6 +39,7 @@ public:
bool Init();
// Set the status to `s'.
void set_status(TaskStatus s, TaskMeta* meta);
+ static bool set_status_unsafe(TaskStatus s, TaskMeta* meta);
static void set_running_status(pthread_t worker_tid, TaskMeta* meta);
static bool set_end_status_unsafe(TaskMeta* m);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
