From aa054fda6a6501ee714516b551d6172ecc630ff9 Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Mon, 12 Oct 2026 15:45:56 +0800 Subject: [PATCH 1/2] fix(ubshm): keep the cleanup queue alive past process exit g_cleanup_queue was a file-scope MPSCQueue with a destructor, while the cleanup worker is a bthread that is never stopped: the bthread runtime itself is not destructed at exit (see the note in task_control.cpp), so the worker can still be inside Dequeue() when static destructors run. MPSCQueue's destructor drains the queue, turning that destruction into a second concurrent consumer of a single-consumer queue: _cur_dequeue_node is a plain pointer, so the destructor and the worker could pick and free the same node (double free) or interleave their advances (lost nodes). A timer callback firing during the destructor phase would additionally enqueue into a destroyed queue. No orderly teardown drains the queue at exit -- GlobalRelease only runs on init failure and in UT teardown -- so the residue the destructor would drain is really possible. Allocate the queue with new and never delete it, following the brpc convention for globals entangled with threads that never stop (g_task_control itself); the kernel reclaims it at exit. DrainAndWait observes g_cleanup_inflight, not the queue, so teardown semantics are unchanged. g_cleanup_inflight (trivially destructible atomics) and g_worker_mtx (a static initializer, nothing to destroy) keep their storage. Tested: cmake --build build --target brpc_ubring_unittest, and ./build/test/brpc_ubring_unittest (12/12 passed). --- src/brpc/ubshm/ub_cleanup_worker.cpp | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/src/brpc/ubshm/ub_cleanup_worker.cpp b/src/brpc/ubshm/ub_cleanup_worker.cpp index 151862aa5c..682e28a9c0 100644 --- a/src/brpc/ubshm/ub_cleanup_worker.cpp +++ b/src/brpc/ubshm/ub_cleanup_worker.cpp @@ -45,7 +45,18 @@ butil::atomic g_worker_unavailable(false); // the worker decrements it after the job returned, so DrainAndWait can wait // for quiescence without observing the queue itself. butil::atomic g_cleanup_inflight(0); -butil::MPSCQueue g_cleanup_queue; +// The queue is never destroyed. The cleanup worker bthread outlives every +// orderly teardown -- the bthread runtime itself is not destructed at exit +// (see the note in task_control.cpp), so the worker can still be inside +// Dequeue() while static destructors run. MPSCQueue's destructor drains the +// queue, which would make that destruction a *second* concurrent consumer +// of a single-consumer queue (_cur_dequeue_node is a plain pointer) and +// double-free nodes; a timer callback firing during the destructor phase +// would additionally enqueue into a destroyed queue. Leaking the queue +// follows the brpc convention for globals entangled with threads that +// never stop (g_task_control itself); the kernel reclaims it at exit. +butil::MPSCQueue* const g_cleanup_queue = + new butil::MPSCQueue(); // Serializes the posters (never taken by the worker) around the lazy start and // the enqueue; see Post(). pthread_mutex_t g_worker_mtx = PTHREAD_MUTEX_INITIALIZER; @@ -65,7 +76,7 @@ void RunTrxCleanup(const UbrCleanupJob& job) { void* UbrCleanupWorkerMain(void*) { UbrCleanupJob job; for (;;) { - if (!g_cleanup_queue.Dequeue(job)) { + if (!g_cleanup_queue->Dequeue(job)) { bthread_usleep(kCleanupIdlePollIntervalUs); continue; } @@ -110,7 +121,7 @@ bool Post(const UbrCleanupJob& job) { // Increment before enqueueing so a concurrent DrainAndWait can never // observe quiescence while this job is still on its way to the queue. g_cleanup_inflight.fetch_add(1); - g_cleanup_queue.Enqueue(job); + g_cleanup_queue->Enqueue(job); return true; } From 6befc10e4e5e9eee58574283fad4a7f2815e596f Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Mon, 12 Oct 2026 16:51:33 +0800 Subject: [PATCH 2/2] fix(ubshm): retry the cleanup worker start instead of disabling it A failed bthread_start_background in Post() latched g_worker_unavailable forever, so every later cleanup fell back to running inline: the SDK calls went back to the process-wide timer thread (UbsShmDrainNode from UbsShmCallback, the delayed trx cleanups from their callbacks), delaying unrelated RPC timeouts, backup requests and bthread wakeups -- exactly what the worker exists to avoid. bthread creation fails transiently (ENOMEM), so the degradation correlated with memory pressure, the moment where blocking the timer thread hurts most, and it never self-healed. Drop the terminal flag and retry the lazy start on every post made while the worker is not started. A failed attempt leaves g_worker_started false and fails only that post; the caller runs it inline as the safety net, and the next post retries the start. The posting mutex already serializes the start attempt with the enqueue, so a job is still only ever queued behind a start that succeeded -- the flag was redundant double insurance for orphaned jobs. The error is logged with LOG_EVERY_SECOND so a failure window cannot spam. Add the g_ub_test_worker_start_rc hook so the new UT can inject a start failure deterministically (no real bthread is created while it is set), and cover both review cases: repeated failures fail only their own post and queue nothing, and the first post after the injection is cleared retries the start, runs the job exactly once and lets DrainAndWait observe quiescence. The suite must observe the worker in its unstarted state, so it is declared ahead of UBShmEndpointTest, whose teardown posts cleanups, and probes for an already-started worker to skip rather than assert on a start that is never attempted. Tested: cmake --build build --target brpc_ubring_unittest, and ./build/test/brpc_ubring_unittest (13/13 passed). Reintroducing the latch locally makes the new UT fail, confirming it guards the retry. --- src/brpc/ubshm/ub_cleanup_worker.cpp | 57 +++++++++++++++++----------- src/brpc/ubshm/ub_cleanup_worker.h | 13 +++++-- 2 files changed, 43 insertions(+), 27 deletions(-) diff --git a/src/brpc/ubshm/ub_cleanup_worker.cpp b/src/brpc/ubshm/ub_cleanup_worker.cpp index 682e28a9c0..c53e351f24 100644 --- a/src/brpc/ubshm/ub_cleanup_worker.cpp +++ b/src/brpc/ubshm/ub_cleanup_worker.cpp @@ -28,6 +28,14 @@ namespace brpc { namespace ubring { +// Test hook, never set in production: when non-zero, the lazy worker start +// inside Post() skips the real bthread_start_background call and behaves as +// if it had returned this value, so a transient start failure (e.g. ENOMEM) +// and its recovery can be exercised deterministically. It is only consulted +// while the worker is not started; read under g_worker_mtx, written by tests +// before posting. +butil::atomic g_ub_test_worker_start_rc(0); + namespace { // Cleanup latency is tolerated in seconds (it only delays releasing memory of @@ -36,11 +44,6 @@ namespace { constexpr uint64_t kCleanupIdlePollIntervalUs = 100 * 1000; butil::atomic g_worker_started(false); -// Set when the worker bthread could not be created. Terminal: posting then -// fails forever, and every caller falls back to running its work inline -- the -// behaviour from before the worker existed. Never cleared, so no job can ever -// be queued behind a worker that will not run. -butil::atomic g_worker_unavailable(false); // Posted-but-not-finished jobs. A poster increments it before enqueueing and // the worker decrements it after the job returned, so DrainAndWait can wait // for quiescence without observing the queue itself. @@ -91,29 +94,37 @@ void* UbrCleanupWorkerMain(void*) { } bool Post(const UbrCleanupJob& job) { - // Serialize the posters so that "the worker is started" and "a job was - // enqueued" cannot be decided by different threads at the same time. - // Without this, a poster could observe g_worker_started == true (set by - // the thread that is starting the bthread) and enqueue a job while that - // start is about to fail: g_worker_unavailable is terminal, so the job - // would never be dequeued -- its UbrCleanupCtl reference would leak and - // g_cleanup_inflight would never return to zero, hanging every later - // DrainAndWait on a teardown path. Posting only happens on cleanup paths, - // so the mutex is free in practice. The worker itself never takes it. + // Serialize the posters (the worker itself never takes this mutex) so + // that at most one cleanup worker is ever started: two posters seeing + // "not started" at the same time would start two bthreads -- two + // concurrent consumers of a single-consumer queue. The start attempt and + // the enqueue also share one critical section, so a job is only ever + // queued behind a worker whose start succeeded. Posting only happens on + // cleanup paths, so the mutex is free in practice. BAIDU_SCOPED_LOCK(g_worker_mtx); - if (g_worker_unavailable.load()) { - return false; - } if (!g_worker_started.load()) { + // Retry the lazy start on every post instead of latching a failure: + // bthread_start_background can fail transiently (ENOMEM), and a + // permanent fallback would run every later SDK cleanup inline on the + // global timer thread -- exactly what the worker exists to avoid, + // and worst under the memory pressure that caused the failure. A + // failed attempt leaves g_worker_started false and fails only THIS + // post; the caller runs it inline as the safety net, and the next + // post retries the start. bthread_attr_t attr = BTHREAD_ATTR_NORMAL; bthread_attr_set_name(&attr, "UBCleanup"); bthread_t tid; - if (BAIDU_UNLIKELY(bthread_start_background( - &tid, &attr, UbrCleanupWorkerMain, nullptr) != 0)) { - LOG(ERROR) << "Fail to start the ubring cleanup worker bthread."; - // Terminal: never queue a job behind a worker that cannot run, so - // UbrCleanupWorker::DrainAndWait can never observe an orphaned job. - g_worker_unavailable.store(true); + // A non-zero test hook overrides the start result without creating + // a bthread; see g_ub_test_worker_start_rc. + const int injected_rc = g_ub_test_worker_start_rc.load( + butil::memory_order_relaxed); + const int rc = injected_rc != 0 ? injected_rc + : bthread_start_background(&tid, &attr, + UbrCleanupWorkerMain, nullptr); + if (BAIDU_UNLIKELY(rc != 0)) { + LOG_EVERY_SECOND(ERROR) << "Fail to start the ubring cleanup " + << "worker bthread, running the cleanup " + << "inline."; return false; } g_worker_started.store(true); diff --git a/src/brpc/ubshm/ub_cleanup_worker.h b/src/brpc/ubshm/ub_cleanup_worker.h index 005d973fc0..93e90863ad 100644 --- a/src/brpc/ubshm/ub_cleanup_worker.h +++ b/src/brpc/ubshm/ub_cleanup_worker.h @@ -67,15 +67,20 @@ struct UbrCleanupJob { // needs to be quiescent (DrainAndWait) before the shared memory it may touch // is finalized, and an idle worker performs no SDK call at all. Posting is // serialized, so a job can never be queued behind a worker start that failed. +// A failed start (bthread creation may fail transiently, e.g. ENOMEM) is +// retried on the next post; only the posts made while the start keeps failing +// run inline on the caller. class UbrCleanupWorker { public: - // Post a shared-memory drain step. Returns false when the worker cannot be - // started, in which case the caller must run `shm_step' inline. + // Post a shared-memory drain step. Returns false when the worker could not + // be started for this post (the start is retried on the next post), in + // which case the caller must run `shm_step' inline. static bool PostShmDrain(ShmList* shm_list, void (*shm_step)(ShmList* shm_list)); // Post a delayed trx cleanup, transferring the timer/callback reference of - // `ctl'. Returns false when the worker cannot be started, in which case - // the caller keeps the reference and must run the cleanup inline. + // `ctl'. Returns false when the worker could not be started for this post + // (the start is retried on the next post), in which case the caller keeps + // the reference and must run the cleanup inline. static bool PostTrxCleanup(UbrTrx* trx, uint64_t ubr_id, void (*work)(UbrTrx* trx, uint64_t ubr_id), UbrCleanupCtl* ctl);