Ubshm transport fix 2

docs/cn/ubring.md

@@ -187,7 +187,7 @@ UBRing 架构包含以下组件：
 
 ### 定时器管理
 
-UBRing 的连接管理和超时处理基于 bthread 定时器（`bthread_timer_add`/`bthread_timer_del`，由 `timer_mgr.cpp` 封装）。非阻塞的 `UbrTimerDel` 可在回调内自删，此时句柄会保留到该回调返回，因此并发的外部 `UbrTimerDelAndWait` 一定能等到它；销毁回调参数所属资源的外部路径用 `UbrTimerDelAndWait` 等待运行中回调退出；一次性定时器触发后自动清理句柄。进入关闭处理时会带上调用方看到的代际，并在 manager 锁内原子认领槽位，因此按旧快照发起的关闭（例如共享内存故障事件回调）不会误伤释放后被复用的新连接。定时器回调运行在进程全局的 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 定时线程上，必须快速返回。阻塞的共享内存清理不在这里进行：延迟清理与待回收映射的回调只仲裁清理所有权，并把任务投递给专用的清理 worker bthread（`ub_cleanup_worker.cpp`），由它把 UBS SDK 的 `unmap`/`deallocate` 调用移出定时线程。对一次性定时器，`UbrTimerDel` 返回 0 还额外保证 facade 已不再访问句柄槽位所在的存储（它会等待被定时线程派发但尚未竞争槽位的 wrapper 退出竞争），因此调用方可以释放承载该槽位的对象；延迟清理路径正是据此释放 `UbrCleanupCtl`。关闭检查定时器在链路空闲时按指数退避轮询（上限 `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; 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.
+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. Blocking shared-memory cleanup is therefore not done there: the delayed-clear and pending-unmap callbacks only settle the cleanup ownership and post a job to a dedicated cleanup worker bthread (`ub_cleanup_worker.cpp`), which runs the UBS SDK `unmap`/`deallocate` calls off the timer thread. For a one-shot timer, a `UbrTimerDel` that returns 0 additionally guarantees that the facade is done with the storage the handle slot lives in -- it waits out a wrapper the timer thread already dispatched -- so the caller may free the object that holds the slot; the delayed-clear path frees `UbrCleanupCtl` that way. 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

@@ -24,6 +24,7 @@
 #include <gflags/gflags.h>
 #include "butil/scoped_lock.h"
 #include "brpc/ubshm/timer/timer_mgr.h"
+#include "brpc/ubshm/ub_cleanup_worker.h"
 #include "brpc/ubshm/common/common.h"
 #include "brpc/ubshm/shm/shm_def.h"
 #include "brpc/ubshm/ub_ring_manager.h"
@@ -49,6 +50,11 @@ 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;
+// True while a drain step of g_shm_list is queued on (or running in) the
+// cleanup worker. It bounds the job backlog: the periodic timer only posts a
+// new step when the previous one finished, and the worker clears the flag at
+// the end of every step, successful or not.
+static butil::atomic<bool> g_shm_drain_pending(false);
 // 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.
@@ -434,30 +440,57 @@ void *UbsShmCallback(void* args, uint64_t)
         return nullptr;
     }
 
-    // Drain one node per fire and keep the SDK calls outside the lock, so
-    // a slow daemon cannot stall the timer thread for the whole list.
+    // The timer thread runs every timer in the process, so it must not call
+    // into the SDK: post the drain to the cleanup worker. At most one step may
+    // be in flight, otherwise a slow SDK call would pile up jobs for the same
+    // list head.
+    if (g_shm_drain_pending.exchange(true)) {
+        return nullptr;
+    }
+    if (BAIDU_UNLIKELY(!UbrCleanupWorker::PostShmDrain(shm_list, UbsShmDrainNode))) {
+        // The worker could not be started: keep the pre-worker behaviour and
+        // drain here rather than leaving the nodes in the list forever.
+        LOG(ERROR) << "Cleanup worker unavailable, draining shm on the timer thread.";
+        UbsShmDrainNode(shm_list);
+    }
+    return nullptr;
+}
+
+void UbsShmDrainNode(ShmList *shm_list)
+{
+    if (BAIDU_UNLIKELY(shm_list == nullptr)) {
+        LOG(ERROR) << "Shm list is null.";
+        g_shm_drain_pending.store(false);
+        return;
+    }
+
+    // Drain one node per step and keep the SDK calls outside the lock, so a
+    // slow daemon cannot stall other users of the list.
     SHM shm;
     {
         BAIDU_SCOPED_LOCK(shm_list->shm_lock);
         if (shm_list->head == nullptr) {
-            return nullptr;
+            g_shm_drain_pending.store(false);
+            return;
         }
         shm = shm_list->head->shm;
     }
     if (shm.addr == nullptr) {
         LOG(ERROR) << "Ubs input shm param is invalid, addr is NULL.";
         BAIDU_SCOPED_LOCK(shm_list->shm_lock);
         DeleteShmToList(shm_list);
-        return nullptr;
+        g_shm_drain_pending.store(false);
+        return;
     }
 
     int ret = ubsmem_shmem_unmap(shm.addr, shm.len);
     if (ret != UBSM_OK) {
-        if (ret == UBSM_ERR_NET) {
-            return nullptr;              // retried on the next fire
+        if (ret != UBSM_ERR_NET) {
+            LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " failed, ret=" << ret;
         }
-        LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " failed, ret=" << ret;
-        return nullptr;                  // node stays at head, retried
+        // The node stays at the head: the next timer fire retries it.
+        g_shm_drain_pending.store(false);
+        return;
     }
 
     {
@@ -469,7 +502,7 @@ void *UbsShmCallback(void* args, uint64_t)
     if (ret != UBSM_OK) {
         LOG(ERROR) << "Ubs delete shm=" << shm.name << " failed, ret=" << ret;
     }
-    return nullptr;
+    g_shm_drain_pending.store(false);
 }
 
 RETURN_CODE UbsShmAddTimer(ShmList *shm_list)
@@ -529,8 +562,11 @@ RETURN_CODE InitShmTimer(ShmList **shm_list)
 
 RETURN_CODE DestroyShmTimer(ShmList **shm_list)
 {
-    // Wait out a possibly running UbsShmCallback before tearing shm_list down.
+    // Wait out a possibly running UbsShmCallback, then the cleanup jobs it
+    // posted: the worker may be inside the SDK calls of a node that is freed
+    // right below, and must not race ubsmem_finalize either.
     UbrTimerDelAndWait(&g_shm_timer_id);
+    UbrCleanupWorker::DrainAndWait();
     if (BAIDU_UNLIKELY(shm_list == nullptr)) {
         LOG(ERROR) << "Shm list handle is null.";
         return UBRING_ERR;

src/brpc/ubshm/shm/shm_ubs.h

@@ -42,6 +42,11 @@ RETURN_CODE UbsShmLocalMmap(SHM *shm, int prot);
 void UbsMemLoggerPrint(int level, const char *msg);
 
 void *UbsShmCallback(void* args, uint64_t gen);
+// One drain step of the pending-unmap list: unmap the head node and deallocate
+// it. Called by the cleanup worker (and inline by UbsShmCallback when that
+// worker cannot be started), never by the timer thread directly. The head node
+// stays in the list when the unmap fails, so a later timer fire retries it.
+void UbsShmDrainNode(ShmList *shm_list);
 RETURN_CODE UbsShmAddTimer(ShmList *shm_list);
 RETURN_CODE InitShmTimer(ShmList **shm_list);
 // Takes the handle so it can be cleared after the list is freed: a caller that

src/brpc/ubshm/timer/timer_mgr.cpp

@@ -71,6 +71,14 @@ struct UbrTimerTask {
     butil::atomic<int> ref;
     butil::atomic<bool> join_pending;            // a DelAndWait is waiting
     butil::atomic<bool> done;                    // refs hit zero, joiner frees
+    // Set by whoever settles the handle slot: the one-shot wrapper (right
+    // after its slot CAS, before it runs the user callback), a deleter whose
+    // bthread_timer_del cancelled the task before dispatch, or the kDead
+    // path. Once true, no facade code -- including the dispatched wrapper --
+    // will ever touch `*slot` again, so the object storing the slot may be
+    // freed. The scheduled wrapper is kept alive by its own schedule
+    // reference while a deleter waits on this flag.
+    butil::atomic<bool> slot_retired;
 };
 
 namespace {
@@ -160,6 +168,13 @@ void UbrTimerOnFire(void* p) {
     // the slot is the single arbiter.
     UbrTimerId expected = task;
     const bool owned = task->slot->compare_exchange_strong(expected, nullptr);
+    // The slot storage is now retired: whatever the competition above
+    // decided, this wrapper will not dereference `slot' again. A one-shot
+    // UbrTimerDel that won the exchange (or that cancelled a task which had
+    // not been dispatched at all) is waiting for this store before it lets
+    // the caller free the storage, so it must happen before the potentially
+    // long user callback.
+    task->slot_retired.store(true);
     if (owned) {
         task->cb(task->arg, task->gen);
     }
@@ -208,6 +223,7 @@ RETURN_CODE TimerStartInternal(butil::atomic<UbrTimerId>* slot, uint64_t delay_u
     task->ref.store(3);                          // owner + schedule + starter
     task->join_pending.store(false);
     task->done.store(false);
+    task->slot_retired.store(false);
 
     // Publish the real task before scheduling so a delete or a DelAndWait
     // racing the start always has an object to act on or wait for.
@@ -293,19 +309,41 @@ int UbrTimerDel(butil::atomic<UbrTimerId>* slot) {
         bthread_usleep(1000);
     }
     if (task->state.load() == kDead) {
+        // The wrapper never ran and never will: this call is the one that
+        // settles the slot storage.
+        task->slot_retired.store(true);
         ReleaseRef(task);                        // owner; schedule/starter are
         return 1;                                // settled by the kDead path
     }
     bthread_timer_t id = task->id.load();
     if (id != 0 && bthread_timer_del(id) == 0) {
+        // Cancelled before dispatch: the timer thread will never call the
+        // wrapper, so nothing else can retire the slot storage.
+        task->slot_retired.store(true);
         ReleaseRef(task);                        // schedule: cancelled before dispatch
-    }                                            // ==1: dispatched, OnFire (owned==false)
-                                                 // releases it
+    } else if (!task->periodic) {
+        // One-shot already dispatched: the wrapper may be between its dispatch
+        // (which made bthread_timer_del return 1) and its slot CAS, so it
+        // still has to touch the storage before it returns. Wait it out here,
+        // while this caller still owns the task through the owner reference,
+        // so that returning 0 really means "the storage is not used anymore".
+        // The wrapper retires the slot before calling any user code and takes
+        // no lock doing so, hence this wait cannot deadlock, and it is bounded
+        // by the OS scheduling of the timer thread. A periodic task needs no
+        // such wait: its slot storage is the pooled UbrTrx, which the callers
+        // of the non-blocking delete never free (only UbrMgrFini does, after
+        // UbrTimerDelAndWait), and its retire-slot load happens after the user
+        // callback -- which may take the very lock this caller holds.
+        while (!task->slot_retired.load()) {
+            bthread_usleep(kTimerPollIntervalUs);
+        }
+    }                                            // ==1 periodic: dispatched/running;
+                                                 // OnFire releases the schedule ref
     ReleaseRef(task);                            // owner
     return 0;       // This call won the slot competition. For a one-shot timer,
-                    // the callback will not run. For a periodic timer, future
-                    // rearming is stopped, but an already dispatched or running
-                    // callback may still complete.
+                    // the callback will not run and the slot storage is unused.
+                    // For a periodic timer, future rearming is stopped, but an
+                    // already dispatched or running callback may still complete.
 }
 
 void UbrTimerDelAndWait(butil::atomic<UbrTimerId>* slot) {

src/brpc/ubshm/timer/timer_mgr.h

@@ -64,16 +64,27 @@ RETURN_CODE UbrTimerStartPeriodic(butil::atomic<UbrTimerId>* slot,
                                   UbrTimerBackoffFn backoff = nullptr);
 
 // Non-blocking delete, safe to call from inside the timer callback itself.
-// This function does not wait for an already running callback and does not
-// protect resources reachable from `arg` on its own.
+// This function does not wait for an already running user callback and does
+// not protect resources reachable from `arg` on its own.
 //
 // Returns 0 when this call wins the handle-slot competition.
-// - For a one-shot timer, the callback will not run.
+// - For a one-shot timer, the callback will not run. This call additionally
+//   waits out a wrapper that the timer thread already dispatched: such a
+//   wrapper may have been preempted between its dispatch and the moment it
+//   competes for the slot, and it still has to touch the slot once (to lose
+//   that competition) before it can return. When this call returns 0, the
+//   facade is therefore completely done with the storage `slot` points to,
+//   and the caller may free the object that stores the slot -- which is what
+//   the delayed-clear path does with UbrCleanupCtl. The wait only spins on a
+//   task-internal flag: the wrapper takes no lock and calls no user code
+//   while retiring the slot, so a caller holding any ubring lock can still
+//   wait, and no timer callback can be joined from its own dispatch.
 // - For a periodic timer, future rearming is stopped, but an already
-//   dispatched or running callback may still execute once more. Callers
-//   must not reclaim resources reachable from `arg` based on this return
-//   alone; use UbrTimerDelAndWait when teardown needs to wait for callbacks.
-//   A return of 0 therefore never means "the handle is free again".
+//   dispatched or running callback may still execute once more, and the task
+//   stays anchored in the slot until that callback returns. Callers must not
+//   reclaim resources reachable from `arg` based on this return alone; use
+//   UbrTimerDelAndWait when teardown needs to wait for callbacks. A return of
+//   0 therefore never means "the handle is free again".
 // A periodic callback that deletes its own timer falls into the periodic case
 // as well, but only marks it stopped: the handle slot stays anchored until the
 // callback returns, so a concurrent UbrTimerDelAndWait on the same slot really