Ubshm transport fix 2

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<int> 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<bool> 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<bool> 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<int> g_cleanup_inflight(0);
-butil::MPSCQueue<UbrCleanupJob> 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<UbrCleanupJob>* const g_cleanup_queue =
+        new butil::MPSCQueue<UbrCleanupJob>();
 // 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,37 +94,45 @@ 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);
     }
     // 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;
 }
 

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);