fix(ubshm): wait for self-deleted timers and serialize pool teardown

docs/cn/ubring.md

@@ -187,7 +187,7 @@ UBRing 架构包含以下组件：
 
 ### 定时器管理
 
-UBRing 的连接管理和超时处理基于 bthread 定时器（`bthread_timer_add`/`bthread_timer_del`，由 `timer_mgr.cpp` 封装）。非阻塞的 `UbrTimerDel` 可在回调内自删；销毁回调参数所属资源的外部路径用 `UbrTimerDelAndWait` 等待运行中回调退出；一次性定时器触发后自动清理句柄。定时器回调运行在进程全局的 bthread 定时线程上，必须快速返回。关闭检查定时器在链路空闲时按指数退避轮询（上限 `ub_event_queue_timer_interval_max_us`，默认 10ms），有流量或正在关闭时恢复快速轮询。
+UBRing 的连接管理和超时处理基于 bthread 定时器（`bthread_timer_add`/`bthread_timer_del`，由 `timer_mgr.cpp` 封装）。非阻塞的 `UbrTimerDel` 可在回调内自删，此时句柄会保留到该回调返回，因此并发的外部 `UbrTimerDelAndWait` 一定能等到它；销毁回调参数所属资源的外部路径用 `UbrTimerDelAndWait` 等待运行中回调退出；一次性定时器触发后自动清理句柄。进入关闭处理时会带上调用方看到的代际，并在 manager 锁内原子认领槽位，因此按旧快照发起的关闭（例如共享内存故障事件回调）不会误伤释放后被复用的新连接。定时器回调运行在进程全局的 bthread 定时线程上，必须快速返回。关闭检查定时器在链路空闲时按指数退避轮询（上限 `ub_event_queue_timer_interval_max_us`，默认 10ms），有流量或正在关闭时恢复快速轮询。
 
 ## 参考资料
 

docs/en/ubring.md

@@ -186,7 +186,7 @@ The shared memory manager (`shm_mgr.cpp`) provides a unified interface for diffe
 
 ### Timer Management
 
-UBRing connection management and timeout handling are built on bthread timers (`bthread_timer_add`/`bthread_timer_del`, wrapped in `timer_mgr.cpp`). The non-blocking `UbrTimerDel` is safe to call from within a timer callback; external teardown that frees resources reachable from the callback argument uses `UbrTimerDelAndWait` to wait out a possibly running callback; one-shot timers clean up their handle automatically when they fire. Timer callbacks run on the process-wide bthread timer thread and must return quickly. The close-check timer backs off exponentially while the link is idle (capped by `ub_event_queue_timer_interval_max_us`, 10ms by default) and returns to fast polling once there is traffic or a close in progress.
+UBRing connection management and timeout handling are built on bthread timers (`bthread_timer_add`/`bthread_timer_del`, wrapped in `timer_mgr.cpp`). The non-blocking `UbrTimerDel` is safe to call from within a timer callback; a periodic callback that deletes its own timer keeps its handle anchored until the callback returns, so a concurrent `UbrTimerDelAndWait` from a teardown path always waits for it. External teardown that frees resources reachable from the callback argument uses `UbrTimerDelAndWait` to wait out a possibly running callback; one-shot timers clean up their handle automatically when they fire. Entering a close carries the generation the caller saw and claims the pool slot atomically under the manager lock, so a close started from a stale snapshot (the shared-memory fault event, for instance) cannot hit a connection that reused the slot in the meantime. Timer callbacks run on the process-wide bthread timer thread and must return quickly. The close-check timer backs off exponentially while the link is idle (capped by `ub_event_queue_timer_interval_max_us`, 10ms by default) and returns to fast polling once there is traffic or a close in progress.
 
 ## References
 

src/brpc/ubshm/shm/shm_ubs.cpp

@@ -49,6 +49,10 @@ DEFINE_int32(ub_flying_io_timeout_s, 5,
 char g_region_name[MAX_REGION_NAME_DESC_LENGTH] = {0};
 butil::atomic<UbrTimerId> g_shm_timer_id(nullptr);
 ShmList *g_shm_list = nullptr;
+// Set by UbsShmFini, cleared by UbsShmInit: ShmMgrFini may run more than once,
+// and the SDK finalize plus the shm list teardown must happen exactly once per
+// init.
+static bool g_ubs_shm_finalized = false;
 static RETURN_CODE UbsShmInterfacesLoad(void);
 char hostname[MAX_HOST_NAME_DESC_LENGTH];
 
@@ -314,6 +318,9 @@ void UbsMemLoggerPrint(int level, const char *msg)
 
 RETURN_CODE UbsShmInit(void)
 {
+    // A new init makes UbsShmFini run its teardown again.
+    g_ubs_shm_finalized = false;
+
     // load libubsm_sdk.so and get function pointer
     RETURN_CODE ret_code = UbsShmInterfacesLoad();
     if (ret_code != UBRING_OK) {
@@ -377,8 +384,16 @@ RETURN_CODE UbsShmInit(void)
 
 RETURN_CODE UbsShmFini(void)
 {
+    // Idempotent: ShmMgrFini is allowed to run more than once (GlobalRelease
+    // plus an explicit caller), and finalizing the SDK or tearing the shm list
+    // down twice would touch freed memory.
+    if (g_ubs_shm_finalized) {
+        return UBRING_OK;
+    }
+    g_ubs_shm_finalized = true;
+
     // Stop the cleanup timer before finalizing the SDK it calls into.
-    if (BAIDU_UNLIKELY(DestroyShmTimer(g_shm_list) != UBRING_OK)) {
+    if (BAIDU_UNLIKELY(DestroyShmTimer(&g_shm_list) != UBRING_OK)) {
         LOG(ERROR) << "Ubs shm list finalize failed.";
         return UBRING_ERR;
     }
@@ -459,9 +474,26 @@ void *UbsShmCallback(void* args, uint64_t)
 
 RETURN_CODE UbsShmAddTimer(ShmList *shm_list)
 {
-    const uint64_t timer_interval_us = (uint64_t)FLAGS_ub_flying_io_timeout_s * SEC_TO_USEC;
-    RETURN_CODE rc = UbrTimerStart(&g_shm_timer_id, 0, timer_interval_us,
-                                   UbsShmCallback, (void*)shm_list, 0);
+    // This timer is periodic -- it drains one pending unmap per fire -- while
+    // the same flag is also the one-shot delay of the delayed-clear path, where
+    // 0 legitimately means "do not wait for in-flight IO". A periodic timer
+    // needs a positive period, so fall back to a drain period of one second
+    // instead of either refusing to start (which would break UBS shm init for a
+    // valid tuning of the other use) or degrading to a single drain.
+    uint64_t period_s = (uint64_t)FLAGS_ub_flying_io_timeout_s;
+    if (BAIDU_UNLIKELY(FLAGS_ub_flying_io_timeout_s <= 0)) {
+        LOG(WARNING) << "ub_flying_io_timeout_s=" << FLAGS_ub_flying_io_timeout_s
+                     << " disables the delayed-clear wait; using a 1s drain period "
+                     << "for the shm cleanup timer.";
+        period_s = 1;
+    }
+    const uint64_t timer_interval_us = period_s * SEC_TO_USEC;
+    // The generation argument is the caller's staleness guard; UbsShmCallback
+    // ignores it and never frees `shm_list', whose lifetime is settled by
+    // DestroyShmTimer waiting through UbrTimerDelAndWait before it tears the
+    // list down. 0 is therefore deliberate here, not a missing generation.
+    RETURN_CODE rc = UbrTimerStartPeriodic(&g_shm_timer_id, 0, timer_interval_us,
+                                           UbsShmCallback, (void*)shm_list, 0);
     if (BAIDU_UNLIKELY(rc != UBRING_OK)) {
         LOG(ERROR) << "Start shm timer failed.";
         return UBRING_ERR;
@@ -495,24 +527,33 @@ RETURN_CODE InitShmTimer(ShmList **shm_list)
     return UBRING_OK;
 }
 
-RETURN_CODE DestroyShmTimer(ShmList *shm_list)
+RETURN_CODE DestroyShmTimer(ShmList **shm_list)
 {
     // Wait out a possibly running UbsShmCallback before tearing shm_list down.
     UbrTimerDelAndWait(&g_shm_timer_id);
-    if (shm_list == nullptr) {
-        LOG(WARNING) << "Shm list is null.";
+    if (BAIDU_UNLIKELY(shm_list == nullptr)) {
+        LOG(ERROR) << "Shm list handle is null.";
         return UBRING_ERR;
     }
-    ShmListNode* current = shm_list->head;
+    if (*shm_list == nullptr) {
+        return UBRING_OK;                    // already destroyed, idempotent
+    }
+
+    // Own the pointer for the whole teardown, and clear the caller's handle
+    // (FREE_PTR does that for us) before returning: leaving it dangling lets a
+    // second ShmMgrFini/AddShmToList use freed memory.
+    ShmList* list = *shm_list;
+    ShmListNode* current = list->head;
     ShmListNode* next;
 
     while (current != nullptr) {
         next = current->next;
         free(current);
         current = next;
     }
-    pthread_mutex_destroy(&shm_list->shm_lock);
-    FREE_PTR(shm_list);
+    pthread_mutex_destroy(&list->shm_lock);
+    FREE_PTR(*shm_list);
+    LOG(INFO) << "Ubs shm list destroyed.";
     return UBRING_OK;
 }
 

src/brpc/ubshm/shm/shm_ubs.h

@@ -44,7 +44,10 @@ void UbsMemLoggerPrint(int level, const char *msg);
 void *UbsShmCallback(void* args, uint64_t gen);
 RETURN_CODE UbsShmAddTimer(ShmList *shm_list);
 RETURN_CODE InitShmTimer(ShmList **shm_list);
-RETURN_CODE DestroyShmTimer(ShmList *shm_list);
+// Takes the handle so it can be cleared after the list is freed: a caller that
+// keeps the pointer (g_shm_list) must not be left with a dangling one. Calling
+// it again on a cleared handle is a no-op.
+RETURN_CODE DestroyShmTimer(ShmList **shm_list);
 RETURN_CODE AddShmToList(ShmList *shm_list, SHM *shm);
 RETURN_CODE IsExistInShmList(ShmList *shm_list, const SHM *shm);
 }

src/brpc/ubshm/timer/timer_mgr.cpp

@@ -33,6 +33,11 @@ enum UbrTimerState {
     kDead = 2                                    // scheduling failed
 };
 
+// Granularity of the two join waits in UbrTimerDelAndWait. They only run on
+// teardown paths and only for the short window in which a timer is being
+// started or a callback is finishing.
+constexpr uint64_t kTimerPollIntervalUs = 100;
+
 }  // namespace
 
 // Reference rules: one "owner" ref for the handle slot, one "schedule" ref
@@ -45,6 +50,13 @@ enum UbrTimerState {
 // the callback so that the callback may free the object storing the slot.
 // All atomics are seq_cst so no interleaving can release a ref twice or
 // free the task while a callback or the starter still touches it.
+//
+// A periodic callback that deletes its own timer does NOT take the task out
+// of the slot: it only marks the task stopped, and the firing callback
+// retires the slot right before it returns (see UbrTimerOnFire). Keeping the
+// task anchored for the whole callback is what lets a concurrent
+// UbrTimerDelAndWait still find it and wait for the callback to finish,
+// instead of returning as if no callback were in flight.
 struct UbrTimerTask {
     butil::atomic<UbrTimerId>* slot;
     butil::atomic<bthread_timer_t> id;
@@ -63,6 +75,11 @@ struct UbrTimerTask {
 
 namespace {
 
+// The task whose callback is running on this thread, if any. It lets
+// UbrTimerDel recognize a delete issued by the callback's own dispatch and
+// leave the task anchored until the callback returns.
+thread_local UbrTimerTask* t_running_task = nullptr;
+
 void ReleaseRef(UbrTimerTask* task) {
     if (task->ref.fetch_sub(1) == 1) {
         if (task->join_pending.load()) {
@@ -78,17 +95,41 @@ void UbrTimerOnFire(void* p) {
 
     if (task->periodic) {
         if (!task->stopped.load()) {
+            // Publish the running task so an UbrTimerDel issued by this very
+            // callback leaves the slot anchored for a concurrent waiter.
+            UbrTimerTask* prev_running = t_running_task;
+            t_running_task = task;
             task->cb(task->arg, task->gen);
+            t_running_task = prev_running;
         }
         // Claim the next schedule's ref before re-reading `stopped' so a
         // racing delete can neither free the task nor orphan a re-arm.
         task->ref.fetch_add(1);
         if (task->stopped.load()) {
-            ReleaseRef(task);
+            // Retire the slot here, on the callback's own exit: a self-delete
+            // deliberately left the task anchored. A deleter that took the
+            // slot instead has already consumed the owner reference.
+            UbrTimerId anchored = task;
+            if (task->slot->compare_exchange_strong(anchored, nullptr)) {
+                ReleaseRef(task);                // owner
+            }
+            ReleaseRef(task);                    // claimed next schedule
         } else {
             uint64_t interval = task->interval_us;
             if (task->backoff != nullptr) {
-                interval = task->backoff(task->arg, interval);
+                const uint64_t next = task->backoff(task->arg, interval);
+                if (BAIDU_UNLIKELY(next == 0)) {
+                    // Re-arming for "now" would let this one timer monopolize
+                    // the process-wide timer thread. A back-off (or a flag it
+                    // reads) returning 0 is a bug, so keep the previous
+                    // interval instead of honouring it. Guarded here rather
+                    // than in each back-off function, because any of them can
+                    // make this mistake.
+                    LOG_EVERY_SECOND(ERROR) << "Ubr timer back-off returned 0, keeping interval_us="
+                                            << interval;
+                } else {
+                    interval = next;
+                }
                 task->interval_us = interval;
             }
             bthread_timer_t id = 0;
@@ -133,13 +174,21 @@ UbrTimerTask* TakeOutTask(butil::atomic<UbrTimerId>* slot) {
 }
 
 RETURN_CODE TimerStartInternal(butil::atomic<UbrTimerId>* slot, uint64_t delay_us,
-                               uint64_t interval_us, void* (*cb)(void*, uint64_t),
+                               uint64_t interval_us, bool periodic,
+                               void* (*cb)(void*, uint64_t),
                                void* arg, uint64_t gen,
                                UbrTimerBackoffFn backoff) {
     if (BAIDU_UNLIKELY(slot == nullptr || cb == nullptr)) {
         LOG(ERROR) << "Ubr timer start invalid argument, slot=" << slot;
         return UBRING_ERR;
     }
+    if (BAIDU_UNLIKELY(periodic && interval_us == 0)) {
+        // Rejecting beats silently arming a one-shot: a caller that asks for a
+        // periodic timer and gets a single fire loses whatever the timer was
+        // watching for the rest of the connection's life.
+        LOG(ERROR) << "Ubr periodic timer start requires a positive interval.";
+        return UBRING_ERR;
+    }
 
     UbrTimerTask* task = new (std::nothrow) UbrTimerTask();
     if (BAIDU_UNLIKELY(task == nullptr)) {
@@ -153,7 +202,7 @@ RETURN_CODE TimerStartInternal(butil::atomic<UbrTimerId>* slot, uint64_t delay_u
     task->gen = gen;
     task->backoff = backoff;
     task->interval_us = interval_us;
-    task->periodic = (interval_us > 0);
+    task->periodic = periodic;
     task->state.store(kStarting);
     task->stopped.store(false);
     task->ref.store(3);                          // owner + schedule + starter
@@ -199,15 +248,35 @@ RETURN_CODE TimerStartInternal(butil::atomic<UbrTimerId>* slot, uint64_t delay_u
 }  // namespace
 
 RETURN_CODE UbrTimerStart(butil::atomic<UbrTimerId>* slot, uint64_t delay_us,
-                          uint64_t interval_us, void* (*cb)(void*, uint64_t),
-                          void* arg, uint64_t gen, UbrTimerBackoffFn backoff) {
-    return TimerStartInternal(slot, delay_us, interval_us, cb, arg, gen, backoff);
+                          void* (*cb)(void*, uint64_t), void* arg, uint64_t gen) {
+    return TimerStartInternal(slot, delay_us, 0, false, cb, arg, gen, nullptr);
+}
+
+RETURN_CODE UbrTimerStartPeriodic(butil::atomic<UbrTimerId>* slot,
+                                  uint64_t delay_us, uint64_t interval_us,
+                                  void* (*cb)(void*, uint64_t), void* arg,
+                                  uint64_t gen, UbrTimerBackoffFn backoff) {
+    return TimerStartInternal(slot, delay_us, interval_us, true, cb, arg, gen,
+                              backoff);
 }
 
 int UbrTimerDel(butil::atomic<UbrTimerId>* slot) {
     if (slot == nullptr) {
         return 1;
     }
+    // A periodic callback deleting its own timer must not pull the task out of
+    // the slot: the anchored task is what lets a concurrent UbrTimerDelAndWait
+    // find the running callback and wait for it. Mark it stopped instead and
+    // let UbrTimerOnFire retire the slot when the callback returns. Deleting a
+    // sibling timer from inside a callback is unaffected: the slots differ, so
+    // the pointer comparison below fails and the normal path runs.
+    if (t_running_task != nullptr && slot->load() == t_running_task) {
+        t_running_task->stopped.store(true);
+        // 0 means "the timer is stopped", not "the handle slot is now free":
+        // the slot is retired by UbrTimerOnFire when this callback returns, and
+        // a concurrent UbrTimerDelAndWait may still be joining the task.
+        return 0;
+    }
     // Take the ownership of the slot first: after this exchange every
     // dereference below is safe (the task cannot be freed while we hold
     // the owner reference the slot used to anchor).
@@ -243,16 +312,29 @@ void UbrTimerDelAndWait(butil::atomic<UbrTimerId>* slot) {
     if (slot == nullptr) {
         return;
     }
+    // Called by the callback of that very timer: joining it would wait for
+    // ourselves, so degrade to the non-blocking delete (UbrTimerDel marks the
+    // task stopped and lets UbrTimerOnFire retire the slot on callback exit).
+    // Callers therefore do not need to know whether they run on the timer
+    // thread.
+    if (t_running_task != nullptr && slot->load() == t_running_task) {
+        UbrTimerDel(slot);
+        return;
+    }
     UbrTimerTask* task = TakeOutTask(slot);
     if (task == nullptr) {
         return;
     }
     task->join_pending.store(true);
     task->stopped.store(true);
     // A start still in flight cannot be cancelled yet; wait for the
-    // starter to schedule it or mark it dead.
+    // starter to schedule it or mark it dead, polling at
+    // kTimerPollIntervalUs. DelAndWait runs on teardown paths (force close,
+    // UbrMgrFini, connect failure) where a sub-millisecond delay is
+    // irrelevant, so polling keeps the facade free of a second wait primitive
+    // and works identically from bthreads and plain pthreads.
     while (task->state.load() == kStarting) {
-        bthread_usleep(1000);
+        bthread_usleep(kTimerPollIntervalUs);
     }
     if (task->state.load() == kScheduled) {
         bthread_timer_t id = task->id.load();
@@ -262,7 +344,7 @@ void UbrTimerDelAndWait(butil::atomic<UbrTimerId>* slot) {
     }
     ReleaseRef(task);                            // owner reference
     while (!task->done.load()) {
-        bthread_usleep(1000);
+        bthread_usleep(kTimerPollIntervalUs);
     }
     task->join_pending.store(false);
     delete task;