diff --git a/src/brpc/ubshm/ub_cleanup_worker.cpp b/src/brpc/ubshm/ub_cleanup_worker.cpp index 151862aa5c..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,16 +44,22 @@ 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. 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 +79,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; } @@ -80,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); @@ -110,7 +132,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; } 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);