From 48ad166bccf87d6adaeb7c21bd6e3db121195eea Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Fri, 9 Oct 2026 16:06:18 +0800 Subject: [PATCH 1/6] fix(ubshm): retire the slot storage before a deleted one-shot timer returns UbrTimerDel wins the handle-slot exchange and reports success for a one-shot timer, but a wrapper that the timer thread already dispatched may still be between its dispatch and its own (losing) slot exchange. The caller of UbrTimerDel treats return 0 as "the callback will not run" and frees the object that stores the slot -- UbrCleanupCtl on the delayed-clear path -- so that late wrapper access is a use-after-free. Track the slot retirement on the task: the one-shot wrapper sets slot_retired right after its slot CAS and before running the user callback, and UbrTimerDel waits for it when bthread_timer_del reports the task as dispatched. Callers that win the cancellation get the same guarantee from the delete itself. The wait only spins on a task-internal flag, so it is safe to do under the trx manager lock that the force-close, pool-acquire and UbrMgrFini callers hold, and it is limited to the one-shot path (a periodic task's slot storage is pool memory that the non-blocking callers never free). --- src/brpc/ubshm/timer/timer_mgr.cpp | 48 ++++++++++++++++++++++++++---- src/brpc/ubshm/timer/timer_mgr.h | 25 +++++++++++----- 2 files changed, 61 insertions(+), 12 deletions(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index 344c365a9f..8eb864dd00 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -71,6 +71,14 @@ struct UbrTimerTask { butil::atomic ref; butil::atomic join_pending; // a DelAndWait is waiting butil::atomic 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 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* 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* 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* slot) { diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index b8d3676ab6..1c35a22499 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -64,16 +64,27 @@ RETURN_CODE UbrTimerStartPeriodic(butil::atomic* 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 From 0552d8fa708e722e369d77b843172ef38e12eb4d Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Fri, 9 Oct 2026 16:06:25 +0800 Subject: [PATCH 2/6] fix(ubshm): claim forced cleanup and cleanup publication atomically The force close used to snapshot the pool slot's cleanup control object under the manager lock and then re-verify "still no ctl" out of the lock. A concurrent SDK-fault callback could publish a delayed cleanup for the same generation in between, so both the force close and the delayed-clear callback would run the cleanup of the same shared memory (double unmap / stale event queues). Replace the two steps with ClaimTrxCleanupForced, which snapshots the ctl and, when there is none, claims the cleanup for the generation inside one critical section, and make TryPublishUnitCleanupCtl refuse to publish once that claim happened (new per-acquisition trx->cleanup_forced flag, reset by the next AcquireUbrTrxFromMgr). Publication and claim are now totally ordered by the manager lock, so exactly one of them owns the cleanup. An existing publication keeps the current ctl->state arbitration untouched. --- src/brpc/ubshm/ub_ring.cpp | 17 +++++++----- src/brpc/ubshm/ub_ring_manager.cpp | 43 +++++++++++++++++++++++++----- src/brpc/ubshm/ub_ring_manager.h | 36 ++++++++++++++++++++++--- src/brpc/ubshm/ubr_trx.h | 9 +++++++ 4 files changed, 88 insertions(+), 17 deletions(-) diff --git a/src/brpc/ubshm/ub_ring.cpp b/src/brpc/ubshm/ub_ring.cpp index 1d551e3315..0ec3147a19 100644 --- a/src/brpc/ubshm/ub_ring.cpp +++ b/src/brpc/ubshm/ub_ring.cpp @@ -241,13 +241,16 @@ RETURN_CODE UBRing::UbrTrxClose() { // to an already running delayed-clear callback otherwise. UbrTimerDelAndWait(&_trx->close_timer); UbrTimerDelAndWait(&_trx->hb_timer); - UbrCleanupCtl* ctl = UBRingManager::SnapshotUnitCleanupCtl(_trx->trx_mgr_index); - if (ctl != nullptr && ctl->ubr_id != expect_ubr_id) { - ctl->ReleaseRef(); // snapshot reference - ctl = nullptr; // slot reused, not ours - } + // Snapshot the delayed cleanup and, when there is none, claim the + // cleanup for this force close in one critical section. Splitting + // the two would let a concurrent SDK-fault callback publish a new + // cleanup after our empty snapshot, leaving two paths running the + // cleanup of the same shared memory. + UbrCleanupCtl* ctl = nullptr; + const UbrCleanupClaim claim = UBRingManager::ClaimTrxCleanupForced( + _trx->trx_mgr_index, expect_ubr_id, &ctl); bool cleanup_owned = false; - if (ctl != nullptr) { + if (claim == CLAIM_HAS_CTL) { int expected = UBR_CLEANUP_PENDING; if (ATOMIC_COMPARE_EXCHANGE_STRONG(ctl->state, expected, UBR_CLEANUP_RUNNING)) { cleanup_owned = true; @@ -255,7 +258,7 @@ RETURN_CODE UBRing::UbrTrxClose() { ctl->ReleaseRef(); // timer/callback reference } } - } else if (ATOMIC_LOAD(_trx->ubr_id) == expect_ubr_id) { + } else if (claim == CLAIM_OWNED_NULL) { cleanup_owned = true; } if (cleanup_owned) { diff --git a/src/brpc/ubshm/ub_ring_manager.cpp b/src/brpc/ubshm/ub_ring_manager.cpp index 89529e37e6..32e9332d92 100644 --- a/src/brpc/ubshm/ub_ring_manager.cpp +++ b/src/brpc/ubshm/ub_ring_manager.cpp @@ -251,6 +251,7 @@ RETURN_CODE UBRingManager::AcquireUbrTrxFromMgr(UbrTrx **trx) { g_ubr_mgr.trx_mgr[i].close_timer = nullptr; g_ubr_mgr.trx_mgr[i].hb_timer = nullptr; g_ubr_mgr.trx_mgr[i].cleanup_ctl = nullptr; + g_ubr_mgr.trx_mgr[i].cleanup_forced.store(false); // Retire the previous acquisition's cleanup control object. UbrCleanupCtl* old_ctl = g_ubr_mgr.trx_mgr_unit_ctl[i]; g_ubr_mgr.trx_mgr_unit_ctl[i] = nullptr; @@ -329,16 +330,45 @@ RETURN_CODE UBRingManager::ReleaseUbrTrxFromMgr(UbrTrx *trx, return UBRING_OK; } -UbrCleanupCtl* UBRingManager::SnapshotUnitCleanupCtl(uint32_t idx) { +UbrCleanupClaim UBRingManager::ClaimTrxCleanupForced( + uint32_t idx, uint64_t expect_ubr_id, UbrCleanupCtl** out_ctl) { + if (out_ctl != nullptr) { + *out_ctl = nullptr; + } BAIDU_SCOPED_LOCK(g_ubr_trx_mgr_mtx); - if (BAIDU_UNLIKELY(g_ubr_mgr.trx_mgr_unit_ctl == nullptr || idx >= g_ubr_mgr.trx_cap)) { - return nullptr; + if (BAIDU_UNLIKELY(g_ubr_mgr.trx_mgr == nullptr || + g_ubr_mgr.trx_mgr_unit_status == nullptr || + g_ubr_mgr.trx_mgr_unit_id == nullptr || + g_ubr_mgr.trx_mgr_unit_ctl == nullptr || + idx >= g_ubr_mgr.trx_cap)) { + return CLAIM_NOT_OURS; + } + // Same generation check as TryClaimTrxClose: a caller that snapshotted the + // slot before a release/reuse must not claim the new occupant's cleanup. + if (g_ubr_mgr.trx_mgr_unit_status[idx] != UBR_MGR_UNIT_USED || + g_ubr_mgr.trx_mgr_unit_id[idx] != expect_ubr_id) { + return CLAIM_NOT_OURS; } UbrCleanupCtl* ctl = g_ubr_mgr.trx_mgr_unit_ctl[idx]; if (ctl != nullptr) { + // An anchored ctl always belongs to the current generation: the + // publication checked it and the acquire retires the previous one. + // Keep the explicit check as a belt-and-braces guard. + if (BAIDU_UNLIKELY(ctl->ubr_id != expect_ubr_id)) { + return CLAIM_NOT_OURS; + } ctl->ref.fetch_add(1); // snapshot reference - } - return ctl; + if (out_ctl != nullptr) { + *out_ctl = ctl; + } + return CLAIM_HAS_CTL; + } + // No cleanup was published: this call owns the forced cleanup, and + // refusing later publications keeps it the only owner. TryPublishUnitCleanupCtl + // checks the flag in this same critical section, so a publication either + // happened before (we would have seen its ctl above) or is refused. + g_ubr_mgr.trx_mgr[idx].cleanup_forced.store(true); + return CLAIM_OWNED_NULL; } bool UBRingManager::IsUbrTrxSlotUsed(uint32_t idx, uint64_t expect_ubr_id) { @@ -425,8 +455,9 @@ bool UBRingManager::TryPublishUnitCleanupCtl(uint32_t idx, idx >= g_ubr_mgr.trx_cap || g_ubr_mgr.trx_mgr_unit_status[idx] != UBR_MGR_UNIT_USED || g_ubr_mgr.trx_mgr_unit_id[idx] != expect_ubr_id || + g_ubr_mgr.trx_mgr[idx].cleanup_forced.load() || g_ubr_mgr.trx_mgr_unit_ctl[idx] != nullptr)) { - return false; // shutting down / released / reused / already anchored + return false; // shutting down / released / reused / forced / already anchored } // Publish on the trx and anchor it in the pool slot inside one critical // section: a force close that snapshots the slot (also under this lock) diff --git a/src/brpc/ubshm/ub_ring_manager.h b/src/brpc/ubshm/ub_ring_manager.h index 298bbf218e..b171fb03f7 100644 --- a/src/brpc/ubshm/ub_ring_manager.h +++ b/src/brpc/ubshm/ub_ring_manager.h @@ -31,6 +31,19 @@ typedef enum { UBR_MGR_UNIT_USED = 1 } UbrMgrUnitStatus; +// Result of UBRingManager::ClaimTrxCleanupForced. +typedef enum { + // The slot is not (or no longer) used by the requested generation: the + // caller must not touch it and must not run the cleanup. + CLAIM_NOT_OURS = 0, + // A delayed cleanup is published for that generation and was handed back + // in *out_ctl with a snapshot reference the caller must release. + CLAIM_HAS_CTL = 1, + // No delayed cleanup existed; the forced claim took the cleanup ownership + // and later publications for that generation are refused. + CLAIM_OWNED_NULL = 2 +} UbrCleanupClaim; + typedef struct TagUbrMgr { uint32_t trx_num; uint32_t trx_cap; @@ -95,9 +108,23 @@ class UBRingManager { static RETURN_CODE ReleaseUbrTrxFromMgr(UbrTrx *trx, uint64_t expect_ubr_id); - // Snapshot the cleanup control object of a pool slot. The caller - // receives a reference and must ReleaseRef it on every exit path. - static UbrCleanupCtl* SnapshotUnitCleanupCtl(uint32_t idx); + // Snapshot the delayed cleanup control object of a pool slot and, when + // there is none, claim the cleanup of `expect_ubr_id' for the force close + // that will run it. Both the snapshot and the claim happen under the + // manager lock, i.e. in the same critical section as + // TryPublishUnitCleanupCtl, so the two are totally ordered: either the + // publication ran first (CLAIM_HAS_CTL, and the existing ctl->state + // arbitration decides who runs the cleanup) or the claim ran first + // (CLAIM_OWNED_NULL, and every later publication for this generation is + // refused through trx->cleanup_forced). Splitting the two steps -- a + // snapshot, then an out-of-lock re-check that no ctl appeared -- leaves a + // window in which a concurrent SDK-fault callback publishes a new cleanup + // after an empty snapshot and both paths run the cleanup. + // `*out_ctl' is only written on CLAIM_HAS_CTL, and carries a snapshot + // reference the caller must ReleaseRef. + static UbrCleanupClaim ClaimTrxCleanupForced(uint32_t idx, + uint64_t expect_ubr_id, + UbrCleanupCtl** out_ctl); // Under the manager lock, confirm the pool slot is still used by the // given generation (not released or reused meanwhile). @@ -121,7 +148,8 @@ class UBRingManager { // other. On success the ctl gains the manager anchor reference (released // when the anchor is detached or the slot is retired); returns false -- // leaving the slot and the reference counts untouched -- when the trx was - // released, reused, or a cleanup is already anchored. + // released, reused, a force close already claimed the cleanup, or a + // cleanup is already anchored. static bool TryPublishUnitCleanupCtl(uint32_t idx, uint64_t expect_ubr_id, UbrCleanupCtl *ctl); diff --git a/src/brpc/ubshm/ubr_trx.h b/src/brpc/ubshm/ubr_trx.h index cc35f7223f..5a489117a7 100644 --- a/src/brpc/ubshm/ubr_trx.h +++ b/src/brpc/ubshm/ubr_trx.h @@ -170,6 +170,15 @@ typedef struct TagUbrTrx { butil::atomic close_timer; butil::atomic hb_timer; butil::atomic cleanup_ctl; + // True once a force close claimed the cleanup of this acquisition while no + // delayed cleanup was published (see + // UBRingManager::ClaimTrxCleanupForced). The forced claim and the + // publication of a delayed cleanup are serialized by the manager lock, so + // a later publication sees the flag and refuses: only one of the two ever + // owns the cleanup. Reset by AcquireUbrTrxFromMgr together with the other + // per-acquisition fields; a stale true on a FREE slot is harmless because + // a publication is already refused while the slot is not USED. + butil::atomic cleanup_forced; // Last io ids seen by the close-check timer, used to reset its // back-off polling interval when the link has traffic. uint64_t close_chk_in_io_id; From e47d0048e631b4df369afcaee9fec12be74d9c84 Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Fri, 9 Oct 2026 16:06:33 +0800 Subject: [PATCH 3/6] fix(ubshm): offload ubs sdk cleanup from the global timer thread Timers are bthread based, so every callback in the process runs on one global timer thread. UbsShmCallback and the delayed-clear callbacks still called the UBS SDK there (ubsmem_shmem_unmap/deallocate), and those calls can wait on the SDK daemon with no latency bound, so a slow or failed shm release delayed every other timer in the process -- RPC timeouts, backup requests, bthread_usleep wakeups and ubring's own heartbeat/close checks. Add a lazily started cleanup worker bthread with an MPSC job queue. The timer callbacks now only arbitrate cleanup ownership and post a job (shared-memory drain or trx cleanup); the worker runs the blocking part, and takes over the timer/callback reference of the cleanup control object. Callers run the work inline only when the worker cannot be started. The drain keeps the previous retry semantics (the node stays in the list on unmap failure and the periodic timer retries it) and at most one drain step is in flight, and DestroyShmTimer drains the worker before it frees the list and finalizes the SDK. UbrMgrFini already waits for the control-object references, so it needs no change. --- src/brpc/ubshm/shm/shm_ubs.cpp | 56 ++++++++-- src/brpc/ubshm/shm/shm_ubs.h | 5 + src/brpc/ubshm/ub_cleanup_worker.cpp | 147 +++++++++++++++++++++++++++ src/brpc/ubshm/ub_cleanup_worker.h | 90 ++++++++++++++++ src/brpc/ubshm/ub_ring.cpp | 67 +++++++----- 5 files changed, 331 insertions(+), 34 deletions(-) create mode 100644 src/brpc/ubshm/ub_cleanup_worker.cpp create mode 100644 src/brpc/ubshm/ub_cleanup_worker.h diff --git a/src/brpc/ubshm/shm/shm_ubs.cpp b/src/brpc/ubshm/shm/shm_ubs.cpp index aa4086619d..699c26dd44 100644 --- a/src/brpc/ubshm/shm/shm_ubs.cpp +++ b/src/brpc/ubshm/shm/shm_ubs.cpp @@ -24,6 +24,7 @@ #include #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 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 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,13 +440,38 @@ 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; } @@ -448,16 +479,18 @@ void *UbsShmCallback(void* args, uint64_t) 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; diff --git a/src/brpc/ubshm/shm/shm_ubs.h b/src/brpc/ubshm/shm/shm_ubs.h index 02ed51cd46..562e1f52c2 100644 --- a/src/brpc/ubshm/shm/shm_ubs.h +++ b/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 diff --git a/src/brpc/ubshm/ub_cleanup_worker.cpp b/src/brpc/ubshm/ub_cleanup_worker.cpp new file mode 100644 index 0000000000..09020645b2 --- /dev/null +++ b/src/brpc/ubshm/ub_cleanup_worker.cpp @@ -0,0 +1,147 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include +#include "butil/atomicops.h" +#include "butil/containers/mpsc_queue.h" +#include "butil/logging.h" +#include "bthread/bthread.h" +#include "brpc/ubshm/ub_cleanup_worker.h" +#include "brpc/ubshm/ub_ring_manager.h" + +namespace brpc { +namespace ubring { + +namespace { + +// Cleanup latency is tolerated in seconds (it only delays releasing memory of +// an already dead link), so the idle worker polls instead of spinning and the +// module needs no extra wakeup primitive. +constexpr uint64_t kCleanupIdlePollIntervalUs = 100 * 1000; + +butil::atomic g_worker_started(false); +// Set when the worker bthread could not be created. Posting then fails +// forever, and every caller falls back to running its work inline -- the +// behaviour from before the worker existed. +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; + +void RunTrxCleanup(const UbrCleanupJob& job) { + UbrCleanupCtl* ctl = job.ctl; + // The slot may have been released and reused while the job was queued; the + // cleanup body must then be skipped, but the control object still has to + // be settled (exactly as the timer callback did before the offload). + if (UBRingManager::IsUbrTrxSlotUsed(job.trx->trx_mgr_index, job.ubr_id)) { + job.work(job.trx, job.ubr_id); + } + ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE); + ctl->ReleaseRef(); // timer/callback reference +} + +void* UbrCleanupWorkerMain(void*) { + UbrCleanupJob job; + for (;;) { + if (!g_cleanup_queue.Dequeue(job)) { + bthread_usleep(kCleanupIdlePollIntervalUs); + continue; + } + if (job.type == UbrCleanupJob::SHM_DRAIN) { + job.shm_step(job.shm_list); + } else { + RunTrxCleanup(job); + } + g_cleanup_inflight.fetch_sub(1); + } + return nullptr; +} + +bool EnsureStarted() { + if (g_worker_unavailable.load()) { + return false; + } + if (g_worker_started.load()) { + return true; + } + bool expected = false; + if (!g_worker_started.compare_exchange_strong(expected, true)) { + return true; // another poster is starting it + } + 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."; + g_worker_unavailable.store(true); + g_worker_started.store(false); + return false; + } + return true; +} + +bool Post(const UbrCleanupJob& job) { + if (!EnsureStarted()) { + return false; + } + // 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); + return true; +} + +} // namespace + +bool UbrCleanupWorker::PostShmDrain(ShmList* shm_list, + void (*shm_step)(ShmList* shm_list)) { + if (BAIDU_UNLIKELY(shm_list == nullptr || shm_step == nullptr)) { + return false; + } + UbrCleanupJob job; + job.type = UbrCleanupJob::SHM_DRAIN; + job.shm_list = shm_list; + job.shm_step = shm_step; + return Post(job); +} + +bool UbrCleanupWorker::PostTrxCleanup(UbrTrx* trx, uint64_t ubr_id, + void (*work)(UbrTrx*, uint64_t), + UbrCleanupCtl* ctl) { + if (BAIDU_UNLIKELY(trx == nullptr || work == nullptr || ctl == nullptr)) { + return false; + } + UbrCleanupJob job; + job.type = UbrCleanupJob::TRX_CLEANUP; + job.trx = trx; + job.ubr_id = ubr_id; + job.work = work; + job.ctl = ctl; + return Post(job); +} + +void UbrCleanupWorker::DrainAndWait() { + while (g_cleanup_inflight.load() > 0) { + bthread_usleep(1000); + } +} + +} // namespace ubring +} // namespace brpc diff --git a/src/brpc/ubshm/ub_cleanup_worker.h b/src/brpc/ubshm/ub_cleanup_worker.h new file mode 100644 index 0000000000..3c6bbae420 --- /dev/null +++ b/src/brpc/ubshm/ub_cleanup_worker.h @@ -0,0 +1,90 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#ifndef BRPC_UB_CLEANUP_WORKER_H +#define BRPC_UB_CLEANUP_WORKER_H + +#include +#include "brpc/ubshm/common/common.h" +#include "brpc/ubshm/shm/shm_def.h" + +namespace brpc { +namespace ubring { + +struct TagUbrTrx; +typedef struct TagUbrTrx UbrTrx; +struct UbrCleanupCtl; + +// One unit of ubring cleanup work. +// +// The process-wide bthread timer thread runs every timer in the process (RPC +// timeouts, backup requests, bthread_usleep wakeups, ubring heartbeats and +// close checks), so a callback that calls into the UBS SDK -- which may wait +// on a network round trip to the SDK daemon -- delays all of them. The timer +// callbacks therefore only arbitrate ownership and post a job here; the +// blocking part runs on the dedicated cleanup worker. +struct UbrCleanupJob { + enum Type { + SHM_DRAIN = 0, // release one pending shared-memory unmap + TRX_CLEANUP = 1 // run the delayed cleanup of one trx generation + }; + + Type type = SHM_DRAIN; + + // SHM_DRAIN: `shm_step' performs the blocking SDK calls and settles the + // producer's retry bookkeeping. It is a function pointer so this worker + // stays independent of the shm backend, and so its scheduling can be + // unit-tested without the SDK. + void (*shm_step)(ShmList* shm_list) = nullptr; + ShmList* shm_list = nullptr; + + // TRX_CLEANUP: the job takes over the timer/callback reference the + // scheduling callback held on `ctl' and releases it when done. `work' is + // the cleanup body (used by UbrDoAsynClearWork / UbrDoPassiveClearWork) + // and only runs if the slot still belongs to `ubr_id'. + UbrTrx* trx = nullptr; + uint64_t ubr_id = 0; + void (*work)(UbrTrx* trx, uint64_t ubr_id) = nullptr; + UbrCleanupCtl* ctl = nullptr; +}; + +// A lazily started, process-wide bthread that drains UbrCleanupJob's. It is +// never stopped during a normal run: after a ShmMgrFini cycle the worker only +// needs to be quiescent (DrainAndWait) before the shared memory it may touch +// is finalized, and an idle worker performs no SDK call at all. +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. + 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. + static bool PostTrxCleanup(UbrTrx* trx, uint64_t ubr_id, + void (*work)(UbrTrx* trx, uint64_t ubr_id), + UbrCleanupCtl* ctl); + + // Wait until every posted job finished. Bounded by one in-flight SDK call; + // returns immediately when the worker never started. + static void DrainAndWait(); +}; + +} // namespace ubring +} // namespace brpc + +#endif //BRPC_UB_CLEANUP_WORKER_H diff --git a/src/brpc/ubshm/ub_ring.cpp b/src/brpc/ubshm/ub_ring.cpp index 0ec3147a19..7930ad74b3 100644 --- a/src/brpc/ubshm/ub_ring.cpp +++ b/src/brpc/ubshm/ub_ring.cpp @@ -23,6 +23,7 @@ #include "bthread/bthread.h" #include "butil/logging.h" #include "brpc/ubshm/ub_ring.h" +#include "brpc/ubshm/ub_cleanup_worker.h" #include "brpc/ubshm/ub_ring_manager.h" #include "brpc/ubshm/shm/shm_ipc.h" @@ -159,26 +160,32 @@ static RETURN_CODE UbrScheduleClearTimer(UbrTrx *trx, uint64_t expect_ubr_id, expect_ubr_id); if (BAIDU_UNLIKELY(rc != UBRING_OK)) { // The timer was never scheduled: this path owns the manager, - // timer/callback and starter references. Roll the schedule back - // and run the cleanup inline so the trx does not end up with - // neither timers nor a queued cleanup. If force close claimed the - // ownership meanwhile, leave the manager anchor to it. + // timer/callback and starter references. Hand the cleanup to the + // worker -- the caller may be the timer thread -- so the trx does not + // end up with neither timers nor a queued cleanup, and run it inline + // only when the worker cannot be started. int state_expected = UBR_CLEANUP_PENDING; if (ATOMIC_COMPARE_EXCHANGE_STRONG(ctl->state, state_expected, UBR_CLEANUP_RUNNING)) { - // Gate on the slot still being used by this generation: force - // close may have claimed the cleanup and released the slot (which - // already freed the trx resources) after we were anchored. - if (UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id)) { - work(trx, ctl->ubr_id); + if (!UbrCleanupWorker::PostTrxCleanup(trx, ctl->ubr_id, work, ctl)) { + // Gate on the slot still being used by this generation: force + // close may have claimed the cleanup and released the slot + // (which already freed the trx resources) after we were + // anchored. + if (UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id)) { + work(trx, ctl->ubr_id); + } + ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE); + ctl->ReleaseRef(); // timer/callback reference } - ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE); - // Detach only after DONE: while the inline cleanup runs, the - // anchor and cleanup_ctl must keep telling a concurrent force - // close that this cleanup is owned (RUNNING/DONE), otherwise it - // would fall into its no-ctl branch and clean the trx again. - UBRingManager::DetachUnitCleanupCtl(trx->trx_mgr_index, ctl); + // When the job was posted, the worker owns the timer/callback + // reference and settles state/anchor exactly like the timer-fired + // path; the manager anchor keeps the control object alive until + // the slot is acquired again or UbrMgrFini retires it. + } else { + // A force close already owns the cleanup. It could not take the + // reference through the never-armed timer, so release it here. + ctl->ReleaseRef(); // timer/callback reference } - ctl->ReleaseRef(); // timer/callback reference ctl->ReleaseRef(); // starter reference return UBRING_ERR; } @@ -541,12 +548,19 @@ void* UBRing::UbrPassiveClearCallback(void* args, uint64_t) { ctl->ReleaseRef(); return nullptr; } + // The ownership arbitration above is settled; hand the blocking work to + // the cleanup worker instead of running it on the timer thread, which + // serves every timer in the process. The job takes over the timer/callback + // reference. UbrTrx* trx = ctl->trx; - if (BAIDU_UNLIKELY(UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id))) { - UbrDoPassiveClearWork(trx, ctl->ubr_id); + if (!UbrCleanupWorker::PostTrxCleanup(trx, ctl->ubr_id, + UbrDoPassiveClearWork, ctl)) { + if (BAIDU_UNLIKELY(UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id))) { + UbrDoPassiveClearWork(trx, ctl->ubr_id); + } + ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE); + ctl->ReleaseRef(); // timer/callback reference } - ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE); - ctl->ReleaseRef(); // timer/callback reference return nullptr; } @@ -618,12 +632,17 @@ void *UBRing::UbrAsynClearCallback(void *args, uint64_t) ctl->ReleaseRef(); return nullptr; } + // Same offload as the passive path: the timer thread only arbitrates + // ownership, the worker does the SDK-backed cleanup. UbrTrx* trx = ctl->trx; - if (BAIDU_UNLIKELY(UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id))) { - UbrDoAsynClearWork(trx, ctl->ubr_id); + if (!UbrCleanupWorker::PostTrxCleanup(trx, ctl->ubr_id, + UbrDoAsynClearWork, ctl)) { + if (BAIDU_UNLIKELY(UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id))) { + UbrDoAsynClearWork(trx, ctl->ubr_id); + } + ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE); + ctl->ReleaseRef(); // timer/callback reference } - ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE); - ctl->ReleaseRef(); // timer/callback reference return nullptr; } From efcebda8b6544f7a226219629f6df8c830302528 Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Fri, 9 Oct 2026 17:05:26 +0800 Subject: [PATCH 4/6] fix(ubshm): serialize cleanup worker posting with the lazy start Post() was not serialized with the lazy start of the worker bthread. Two posters could decide "the worker is started" and "enqueue the job" independently: one thread won the start CAS and was about to see bthread_start_background fail, while another read g_worker_started == true and enqueued a job. Since g_worker_unavailable is terminal and no worker ever runs, that job is never dequeued -- its UbrCleanupCtl reference leaks, the pool slot is never released, and g_cleanup_inflight never returns to zero, so every later UbrCleanupWorker::DrainAndWait (DestroyShmTimer on a teardown path) hangs forever. Take a small mutex in Post() around the unavailable check, the start attempt and the enqueue, so a job can only ever be queued behind a worker that really started. Posting only happens on cleanup paths, the worker never takes the mutex, and a failed start now returns false (callers keep running the work inline) without having enqueued anything. --- src/brpc/ubshm/ub_cleanup_worker.cpp | 60 +++++++++++++++------------- src/brpc/ubshm/ub_cleanup_worker.h | 3 +- 2 files changed, 35 insertions(+), 28 deletions(-) diff --git a/src/brpc/ubshm/ub_cleanup_worker.cpp b/src/brpc/ubshm/ub_cleanup_worker.cpp index 09020645b2..151862aa5c 100644 --- a/src/brpc/ubshm/ub_cleanup_worker.cpp +++ b/src/brpc/ubshm/ub_cleanup_worker.cpp @@ -15,10 +15,12 @@ // specific language governing permissions and limitations // under the License. +#include #include #include "butil/atomicops.h" #include "butil/containers/mpsc_queue.h" #include "butil/logging.h" +#include "butil/scoped_lock.h" #include "bthread/bthread.h" #include "brpc/ubshm/ub_cleanup_worker.h" #include "brpc/ubshm/ub_ring_manager.h" @@ -34,15 +36,19 @@ namespace { constexpr uint64_t kCleanupIdlePollIntervalUs = 100 * 1000; butil::atomic g_worker_started(false); -// Set when the worker bthread could not be created. Posting then fails -// forever, and every caller falls back to running its work inline -- the -// behaviour from before the worker existed. +// 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; +// 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; void RunTrxCleanup(const UbrCleanupJob& job) { UbrCleanupCtl* ctl = job.ctl; @@ -73,33 +79,33 @@ void* UbrCleanupWorkerMain(void*) { return nullptr; } -bool EnsureStarted() { +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. + BAIDU_SCOPED_LOCK(g_worker_mtx); if (g_worker_unavailable.load()) { return false; } - if (g_worker_started.load()) { - return true; - } - bool expected = false; - if (!g_worker_started.compare_exchange_strong(expected, true)) { - return true; // another poster is starting it - } - 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."; - g_worker_unavailable.store(true); - g_worker_started.store(false); - return false; - } - return true; -} - -bool Post(const UbrCleanupJob& job) { - if (!EnsureStarted()) { - return false; + if (!g_worker_started.load()) { + 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); + 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. diff --git a/src/brpc/ubshm/ub_cleanup_worker.h b/src/brpc/ubshm/ub_cleanup_worker.h index 3c6bbae420..005d973fc0 100644 --- a/src/brpc/ubshm/ub_cleanup_worker.h +++ b/src/brpc/ubshm/ub_cleanup_worker.h @@ -65,7 +65,8 @@ struct UbrCleanupJob { // A lazily started, process-wide bthread that drains UbrCleanupJob's. It is // never stopped during a normal run: after a ShmMgrFini cycle the worker only // needs to be quiescent (DrainAndWait) before the shared memory it may touch -// is finalized, and an idle worker performs no SDK call at all. +// 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. class UbrCleanupWorker { public: // Post a shared-memory drain step. Returns false when the worker cannot be From ba0c413e89bbc49273ed74c0c8fecee17953d47c Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Fri, 9 Oct 2026 17:05:35 +0800 Subject: [PATCH 5/6] style(ubshm): prefix the cleanup claim results with UBR_CLEANUP_CLAIM The UbrCleanupClaim members were the only constants in ub_ring_manager.h without the module's UBR_ prefix, next to UBR_MGR_UNIT_* and UBR_CLEANUP_*. Rename them to UBR_CLEANUP_CLAIM_NOT_OURS / _HAS_CTL / _OWNED_NULL. --- src/brpc/ubshm/ub_ring.cpp | 4 ++-- src/brpc/ubshm/ub_ring_manager.cpp | 10 +++++----- src/brpc/ubshm/ub_ring_manager.h | 24 ++++++++++++------------ 3 files changed, 19 insertions(+), 19 deletions(-) diff --git a/src/brpc/ubshm/ub_ring.cpp b/src/brpc/ubshm/ub_ring.cpp index 7930ad74b3..17be3feb21 100644 --- a/src/brpc/ubshm/ub_ring.cpp +++ b/src/brpc/ubshm/ub_ring.cpp @@ -257,7 +257,7 @@ RETURN_CODE UBRing::UbrTrxClose() { const UbrCleanupClaim claim = UBRingManager::ClaimTrxCleanupForced( _trx->trx_mgr_index, expect_ubr_id, &ctl); bool cleanup_owned = false; - if (claim == CLAIM_HAS_CTL) { + if (claim == UBR_CLEANUP_CLAIM_HAS_CTL) { int expected = UBR_CLEANUP_PENDING; if (ATOMIC_COMPARE_EXCHANGE_STRONG(ctl->state, expected, UBR_CLEANUP_RUNNING)) { cleanup_owned = true; @@ -265,7 +265,7 @@ RETURN_CODE UBRing::UbrTrxClose() { ctl->ReleaseRef(); // timer/callback reference } } - } else if (claim == CLAIM_OWNED_NULL) { + } else if (claim == UBR_CLEANUP_CLAIM_OWNED_NULL) { cleanup_owned = true; } if (cleanup_owned) { diff --git a/src/brpc/ubshm/ub_ring_manager.cpp b/src/brpc/ubshm/ub_ring_manager.cpp index 32e9332d92..c5c654f98b 100644 --- a/src/brpc/ubshm/ub_ring_manager.cpp +++ b/src/brpc/ubshm/ub_ring_manager.cpp @@ -341,13 +341,13 @@ UbrCleanupClaim UBRingManager::ClaimTrxCleanupForced( g_ubr_mgr.trx_mgr_unit_id == nullptr || g_ubr_mgr.trx_mgr_unit_ctl == nullptr || idx >= g_ubr_mgr.trx_cap)) { - return CLAIM_NOT_OURS; + return UBR_CLEANUP_CLAIM_NOT_OURS; } // Same generation check as TryClaimTrxClose: a caller that snapshotted the // slot before a release/reuse must not claim the new occupant's cleanup. if (g_ubr_mgr.trx_mgr_unit_status[idx] != UBR_MGR_UNIT_USED || g_ubr_mgr.trx_mgr_unit_id[idx] != expect_ubr_id) { - return CLAIM_NOT_OURS; + return UBR_CLEANUP_CLAIM_NOT_OURS; } UbrCleanupCtl* ctl = g_ubr_mgr.trx_mgr_unit_ctl[idx]; if (ctl != nullptr) { @@ -355,20 +355,20 @@ UbrCleanupClaim UBRingManager::ClaimTrxCleanupForced( // publication checked it and the acquire retires the previous one. // Keep the explicit check as a belt-and-braces guard. if (BAIDU_UNLIKELY(ctl->ubr_id != expect_ubr_id)) { - return CLAIM_NOT_OURS; + return UBR_CLEANUP_CLAIM_NOT_OURS; } ctl->ref.fetch_add(1); // snapshot reference if (out_ctl != nullptr) { *out_ctl = ctl; } - return CLAIM_HAS_CTL; + return UBR_CLEANUP_CLAIM_HAS_CTL; } // No cleanup was published: this call owns the forced cleanup, and // refusing later publications keeps it the only owner. TryPublishUnitCleanupCtl // checks the flag in this same critical section, so a publication either // happened before (we would have seen its ctl above) or is refused. g_ubr_mgr.trx_mgr[idx].cleanup_forced.store(true); - return CLAIM_OWNED_NULL; + return UBR_CLEANUP_CLAIM_OWNED_NULL; } bool UBRingManager::IsUbrTrxSlotUsed(uint32_t idx, uint64_t expect_ubr_id) { diff --git a/src/brpc/ubshm/ub_ring_manager.h b/src/brpc/ubshm/ub_ring_manager.h index b171fb03f7..711c06e339 100644 --- a/src/brpc/ubshm/ub_ring_manager.h +++ b/src/brpc/ubshm/ub_ring_manager.h @@ -35,13 +35,13 @@ typedef enum { typedef enum { // The slot is not (or no longer) used by the requested generation: the // caller must not touch it and must not run the cleanup. - CLAIM_NOT_OURS = 0, + UBR_CLEANUP_CLAIM_NOT_OURS = 0, // A delayed cleanup is published for that generation and was handed back // in *out_ctl with a snapshot reference the caller must release. - CLAIM_HAS_CTL = 1, + UBR_CLEANUP_CLAIM_HAS_CTL = 1, // No delayed cleanup existed; the forced claim took the cleanup ownership // and later publications for that generation are refused. - CLAIM_OWNED_NULL = 2 + UBR_CLEANUP_CLAIM_OWNED_NULL = 2 } UbrCleanupClaim; typedef struct TagUbrMgr { @@ -113,15 +113,15 @@ class UBRingManager { // that will run it. Both the snapshot and the claim happen under the // manager lock, i.e. in the same critical section as // TryPublishUnitCleanupCtl, so the two are totally ordered: either the - // publication ran first (CLAIM_HAS_CTL, and the existing ctl->state - // arbitration decides who runs the cleanup) or the claim ran first - // (CLAIM_OWNED_NULL, and every later publication for this generation is - // refused through trx->cleanup_forced). Splitting the two steps -- a - // snapshot, then an out-of-lock re-check that no ctl appeared -- leaves a - // window in which a concurrent SDK-fault callback publishes a new cleanup - // after an empty snapshot and both paths run the cleanup. - // `*out_ctl' is only written on CLAIM_HAS_CTL, and carries a snapshot - // reference the caller must ReleaseRef. + // publication ran first (UBR_CLEANUP_CLAIM_HAS_CTL, and the existing + // ctl->state arbitration decides who runs the cleanup) or the claim ran + // first (UBR_CLEANUP_CLAIM_OWNED_NULL, and every later publication for + // this generation is refused through trx->cleanup_forced). Splitting the + // two steps -- a snapshot, then an out-of-lock re-check that no ctl + // appeared -- leaves a window in which a concurrent SDK-fault callback + // publishes a new cleanup after an empty snapshot and both paths run the + // cleanup. `*out_ctl' is only written on UBR_CLEANUP_CLAIM_HAS_CTL, and + // carries a snapshot reference the caller must ReleaseRef. static UbrCleanupClaim ClaimTrxCleanupForced(uint32_t idx, uint64_t expect_ubr_id, UbrCleanupCtl** out_ctl); From cf25cd68f342108c9dd6caaab7d36e71a8f20646 Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Fri, 9 Oct 2026 17:05:43 +0800 Subject: [PATCH 6/6] docs(ubshm): document the cleanup worker and the one-shot delete contract The Timer Management sections still implied all cleanup work runs on the timer thread. Document that the delayed-clear and pending-unmap callbacks only arbitrate ownership and post a job to the cleanup worker, and that a one-shot UbrTimerDel returning 0 now also guarantees the facade is done with the storage the handle slot lives in, so the caller may free the object holding it (UbrCleanupCtl on the delayed-clear path). Also refresh the ordering comment above UbrTrxCloseCallback: with the cleanup body on the worker, the guarantee rests on the generation gate the worker re-applies under the manager lock (IsUbrTrxSlotUsed) plus the per-trx UbrStopTrxTimer/UbrTimerDelAndWait waits, not on the timer thread serializing the cleanup itself. --- docs/cn/ubring.md | 2 +- docs/en/ubring.md | 2 +- src/brpc/ubshm/ub_ring.cpp | 14 ++++++++++---- 3 files changed, 12 insertions(+), 6 deletions(-) diff --git a/docs/cn/ubring.md b/docs/cn/ubring.md index f4c7c0b841..182de6cba6 100644 --- a/docs/cn/ubring.md +++ b/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），有流量或正在关闭时恢复快速轮询。 ## 参考资料 diff --git a/docs/en/ubring.md b/docs/en/ubring.md index 9247360e04..8cb64df6dc 100644 --- a/docs/en/ubring.md +++ b/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 diff --git a/src/brpc/ubshm/ub_ring.cpp b/src/brpc/ubshm/ub_ring.cpp index 17be3feb21..98f845702e 100644 --- a/src/brpc/ubshm/ub_ring.cpp +++ b/src/brpc/ubshm/ub_ring.cpp @@ -418,10 +418,16 @@ RETURN_CODE UBRing::UbrAddTimer() { // A slot cannot be released and reused underneath a callback that already // passed that check, and the guarantee is a synchronization one, not a timing // one. bthread dispatches all timer callbacks from a single global timer -// thread, so two per-trx callbacks never run concurrently, and a cleanup that -// runs from a one-shot timer callback (the delayed-clear path) is serialized -// after every per-trx callback that was dispatched before it. Entering a close -// claims the slot for the generation under g_ubr_trx_mgr_mtx +// thread, so two per-trx callbacks never run concurrently, and the delayed +// clear arbitration (the ctl->state compare-exchange in the clear callbacks) +// is serialized after every per-trx callback that was dispatched before it. +// The cleanup body itself no longer runs on the timer thread but on the +// cleanup worker (UbrCleanupWorker): it is kept away from a released or reused +// slot by the generation gate it re-applies under g_ubr_trx_mgr_mtx right +// before running the work (IsUbrTrxSlotUsed), and by the scheduling paths that +// wait for the per-trx callbacks through UbrStopTrxTimer -> UbrTimerDelAndWait +// before they may clear the trx or hand its slot on. Entering a close claims +// the slot for the generation under g_ubr_trx_mgr_mtx // (UBRingManager::TryClaimTrxClose), so a caller working from a stale snapshot // -- the faulty-shm event, which looks the slot up before the claim -- cannot // touch the next occupant. Teardown that happens on any other thread then waits