From 68a6914c953fe2bce2cd4c93dd9ea213a6c995d0 Mon Sep 17 00:00:00 2001 From: Sherlock Date: Thu, 24 Sep 2026 12:10:10 +0800 Subject: [PATCH] style(ubshm): use BAIDU macros, BAIDU_SCOPED_LOCK and butil::atomic Make the whole ubring shared-memory module follow the current brpc conventions, addressing review feedback: - Replace the module-local LIKELY/UNLIKELY aliases with BAIDU_LIKELY/ BAIDU_UNLIKELY and drop the aliases from common.h. - Use BAIDU_SCOPED_LOCK instead of the local GNU cleanup-attribute LOCK_GUARD. The pthread_mutex_t members keep their type, allocation and lifetime, so no static-init or placement-new change is needed. - Use butil::atomic instead of std::atomic. The timer handle slots and the trx cleanup slot are now real butil::atomic objects, so plain storage is never reinterpreted as an atomic: the timer API takes butil::atomic* and UbrTrx/UbrCleanupCtl hold atomic handles. UbrTrx slots are value-initialized with placement new so those atomics are constructed, replacing the per-acquisition memset. - Drop the two per-node INFO logs on the shm cleanup retry path in UbsShmCallback; DeleteShmToList still records each drained node. No behavioral change: the removed aliases expanded 1:1, BAIDU_SCOPED_LOCK locks the same pthread_mutex_t, butil::atomic is layout-identical to std::atomic, and placement-new value-initialization zeroes exactly what memset zeroed while also constructing the atomic members. --- src/brpc/ubshm/common/common.h | 13 +- src/brpc/ubshm/common/thread_lock.h | 2 +- src/brpc/ubshm/shm/shm_mgr.cpp | 20 ++-- src/brpc/ubshm/shm/shm_ubs.cpp | 29 ++--- src/brpc/ubshm/timer/timer_mgr.cpp | 46 ++++---- src/brpc/ubshm/timer/timer_mgr.h | 10 +- src/brpc/ubshm/ub_endpoint.cpp | 8 +- src/brpc/ubshm/ub_endpoint.h | 2 +- src/brpc/ubshm/ub_ring.cpp | 177 +++++++++++++--------------- src/brpc/ubshm/ub_ring.h | 24 ++-- src/brpc/ubshm/ub_ring_manager.cpp | 61 +++++----- src/brpc/ubshm/ubr_trx.h | 8 +- 12 files changed, 194 insertions(+), 206 deletions(-) diff --git a/src/brpc/ubshm/common/common.h b/src/brpc/ubshm/common/common.h index 504da07c28..56d2cb8706 100644 --- a/src/brpc/ubshm/common/common.h +++ b/src/brpc/ubshm/common/common.h @@ -24,9 +24,6 @@ #include "butil/compiler_specific.h" #include "butil/logging.h" -#define LIKELY(x) BAIDU_LIKELY(x) -#define UNLIKELY(x) BAIDU_UNLIKELY(x) - #ifndef UNREFERENCE_PARAM #define UNREFERENCE_PARAM(x) ((void)(x)) #endif @@ -42,11 +39,11 @@ #endif #ifdef __cplusplus -#include -using AtomicInt = std::atomic; -using AtomicBool = std::atomic; -using AtomicUintFast64 = std::atomic; -using AtomicUintFast8 = std::atomic; +#include "butil/atomicops.h" +using AtomicInt = butil::atomic; +using AtomicBool = butil::atomic; +using AtomicUintFast64 = butil::atomic; +using AtomicUintFast8 = butil::atomic; #define ATOMIC_INIT(var, value) var.store(value) #define ATOMIC_STORE(var, value) var.store(value) #define ATOMIC_LOAD(var) var.load() diff --git a/src/brpc/ubshm/common/thread_lock.h b/src/brpc/ubshm/common/thread_lock.h index 0233955168..bba5f80702 100644 --- a/src/brpc/ubshm/common/thread_lock.h +++ b/src/brpc/ubshm/common/thread_lock.h @@ -30,7 +30,7 @@ extern "C" { static inline void UnlockMutex(pthread_mutex_t **mtx) { - if (LIKELY(mtx != nullptr && *mtx != nullptr)) { + if (BAIDU_LIKELY(mtx != nullptr && *mtx != nullptr)) { pthread_mutex_unlock(*mtx); } else { LOG(ERROR) << "Invalid input for mtx."; diff --git a/src/brpc/ubshm/shm/shm_mgr.cpp b/src/brpc/ubshm/shm/shm_mgr.cpp index ea01113e54..3dc434756b 100644 --- a/src/brpc/ubshm/shm/shm_mgr.cpp +++ b/src/brpc/ubshm/shm/shm_mgr.cpp @@ -57,7 +57,7 @@ static bool CheckInputShmParam(SHM *shm) { } RETURN_CODE ShmMgrInit(void) { - if (UNLIKELY(FLAGS_ub_shm_type >= (int32_t)SHM_TYPE_UNSUPPORT || FLAGS_ub_shm_type <= (int32_t)SHM_TYPE_UB)) { + if (BAIDU_UNLIKELY(FLAGS_ub_shm_type >= (int32_t)SHM_TYPE_UNSUPPORT || FLAGS_ub_shm_type <= (int32_t)SHM_TYPE_UB)) { LOG(ERROR) << "Shm type config=" << FLAGS_ub_shm_type << " is not supported."; return UBRING_ERR; } @@ -88,7 +88,7 @@ void SetShmType(SHM_TYPE type) { } RETURN_CODE ShmLocalMalloc(SHM *shm) { - if (UNLIKELY(!CheckInputShmParam(shm))) { + if (BAIDU_UNLIKELY(!CheckInputShmParam(shm))) { LOG(ERROR) << "Input param shm is invalid."; return SHM_ERR_INPUT_INVALID; } @@ -110,11 +110,11 @@ RETURN_CODE ShmLocalMalloc(SHM *shm) { RETURN_CODE ShmLocalCalloc(SHM *shm) { RETURN_CODE rc = ShmLocalMalloc(shm); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Failed to alloc local shm."; return rc; } - if (UNLIKELY(shm->addr == nullptr)) { + if (BAIDU_UNLIKELY(shm->addr == nullptr)) { LOG(ERROR) << "Local shm=" << shm->name << " allocated with NULL address."; ShmFree(shm); return SHM_ERR; @@ -124,7 +124,7 @@ RETURN_CODE ShmLocalCalloc(SHM *shm) { } RETURN_CODE ShmLocalFree(SHM *shm) { - if (UNLIKELY(!CheckInputShmParam(shm))) { + if (BAIDU_UNLIKELY(!CheckInputShmParam(shm))) { LOG(ERROR) << "Input param shm is invalid."; return SHM_ERR_INPUT_INVALID; } @@ -145,7 +145,7 @@ RETURN_CODE ShmLocalFree(SHM *shm) { } RETURN_CODE ShmRemoteMalloc(SHM *shm) { - if (UNLIKELY(!CheckInputShmParam(shm))) { + if (BAIDU_UNLIKELY(!CheckInputShmParam(shm))) { LOG(ERROR) << "Input param shm is invalid."; return SHM_ERR_INPUT_INVALID; } @@ -166,7 +166,7 @@ RETURN_CODE ShmRemoteMalloc(SHM *shm) { } RETURN_CODE ShmRemoteFree(SHM *shm) { - if (UNLIKELY(!CheckInputShmParam(shm))) { + if (BAIDU_UNLIKELY(!CheckInputShmParam(shm))) { LOG(ERROR) << "Input param shm is invalid."; return SHM_ERR_INPUT_INVALID; } @@ -187,7 +187,7 @@ RETURN_CODE ShmRemoteFree(SHM *shm) { } RETURN_CODE ShmLocalMmap(SHM *shm, int prot) { - if (UNLIKELY(!CheckInputShmParam(shm))) { + if (BAIDU_UNLIKELY(!CheckInputShmParam(shm))) { LOG(ERROR) << "Input param shm is invalid."; return SHM_ERR_INPUT_INVALID; } @@ -208,7 +208,7 @@ RETURN_CODE ShmLocalMmap(SHM *shm, int prot) { } RETURN_CODE ShmMunmap(SHM *shm) { - if (UNLIKELY(!CheckInputShmParam(shm))) { + if (BAIDU_UNLIKELY(!CheckInputShmParam(shm))) { LOG(ERROR) << "Input param shm is invalid."; return SHM_ERR_INPUT_INVALID; } @@ -229,7 +229,7 @@ RETURN_CODE ShmMunmap(SHM *shm) { } RETURN_CODE ShmFree(SHM *shm) { - if (UNLIKELY(!CheckInputShmParam(shm))) { + if (BAIDU_UNLIKELY(!CheckInputShmParam(shm))) { LOG(ERROR) << "Input param shm is invalid."; return SHM_ERR_INPUT_INVALID; } diff --git a/src/brpc/ubshm/shm/shm_ubs.cpp b/src/brpc/ubshm/shm/shm_ubs.cpp index a4f27ee1d8..89f6702a6b 100644 --- a/src/brpc/ubshm/shm/shm_ubs.cpp +++ b/src/brpc/ubshm/shm/shm_ubs.cpp @@ -22,8 +22,8 @@ #include #include #include +#include "butil/scoped_lock.h" #include "brpc/ubshm/timer/timer_mgr.h" -#include "brpc/ubshm/common/thread_lock.h" #include "brpc/ubshm/common/common.h" #include "brpc/ubshm/shm/shm_def.h" #include "brpc/ubshm/ub_ring_manager.h" @@ -47,7 +47,7 @@ DEFINE_int32(ub_flying_io_timeout_s, 5, "Time in seconds to wait for stopping data sending and receiving " "when the link is disconnected."); char g_region_name[MAX_REGION_NAME_DESC_LENGTH] = {0}; -UbrTimerId g_shm_timer_id = nullptr; +butil::atomic g_shm_timer_id(nullptr); ShmList *g_shm_list = nullptr; static RETURN_CODE UbsShmInterfacesLoad(void); char hostname[MAX_HOST_NAME_DESC_LENGTH]; @@ -351,12 +351,12 @@ RETURN_CODE UbsShmInit(void) return UBRING_ERR; } - if (UNLIKELY(ubsmem_local_nid_query(&FLAGS_node_location) != UBSM_OK)) { + if (BAIDU_UNLIKELY(ubsmem_local_nid_query(&FLAGS_node_location) != UBSM_OK)) { LOG(ERROR) << "Get local nid failed."; return UBRING_ERR; } - if (UNLIKELY(ubsmem_shmem_faults_register(brpc::ubring::UBRingManager::UbEventCallback) != UBSM_OK)) { + if (BAIDU_UNLIKELY(ubsmem_shmem_faults_register(brpc::ubring::UBRingManager::UbEventCallback) != UBSM_OK)) { LOG(ERROR) << "Failed to register the ub event callback function."; return UBRING_ERR; } @@ -378,7 +378,7 @@ RETURN_CODE UbsShmInit(void) RETURN_CODE UbsShmFini(void) { // Stop the cleanup timer before finalizing the SDK it calls into. - if (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; } @@ -414,7 +414,7 @@ static void DeleteShmToList(ShmList* shm_list) void *UbsShmCallback(void* args) { ShmList *shm_list = (ShmList*)args; - if (UNLIKELY(shm_list == nullptr)) { + if (BAIDU_UNLIKELY(shm_list == nullptr)) { LOG(ERROR) << "Shm list is null."; return nullptr; } @@ -423,7 +423,7 @@ void *UbsShmCallback(void* args) // a slow daemon cannot stall the timer thread for the whole list. SHM shm; { - LOCK_GUARD(shm_list->shm_lock); + BAIDU_SCOPED_LOCK(shm_list->shm_lock); if (shm_list->head == nullptr) { return nullptr; } @@ -431,7 +431,7 @@ void *UbsShmCallback(void* args) } if (shm.addr == nullptr) { LOG(ERROR) << "Ubs input shm param is invalid, addr is NULL."; - LOCK_GUARD(shm_list->shm_lock); + BAIDU_SCOPED_LOCK(shm_list->shm_lock); DeleteShmToList(shm_list); return nullptr; } @@ -444,19 +444,16 @@ void *UbsShmCallback(void* args) LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " failed, ret=" << ret; return nullptr; // node stays at head, retried } - LOG(INFO) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " success."; { - LOCK_GUARD(shm_list->shm_lock); + BAIDU_SCOPED_LOCK(shm_list->shm_lock); DeleteShmToList(shm_list); } ret = ubsmem_shmem_deallocate(shm.name); if (ret != UBSM_OK) { LOG(ERROR) << "Ubs delete shm=" << shm.name << " failed, ret=" << ret; - return nullptr; } - LOG(INFO) << "Ubs free local shm=" << shm.name << " length=" << shm.len << " success."; return nullptr; } @@ -465,7 +462,7 @@ 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); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Start shm timer failed."; return UBRING_ERR; } @@ -521,11 +518,11 @@ RETURN_CODE DestroyShmTimer(ShmList *shm_list) RETURN_CODE IsExistInShmList(ShmList *shm_list, const SHM *shm) { - if (UNLIKELY(shm_list == nullptr || shm == nullptr)) { + if (BAIDU_UNLIKELY(shm_list == nullptr || shm == nullptr)) { LOG(ERROR) << "Shm list or shm is null."; return UBRING_ERR; } - LOCK_GUARD(shm_list->shm_lock); + BAIDU_SCOPED_LOCK(shm_list->shm_lock); ShmListNode *cur_node = shm_list->head; while (cur_node != nullptr) { @@ -556,7 +553,7 @@ RETURN_CODE AddShmToList(ShmList *shm_list, SHM *shm) } memcpy(&new_shm_node->shm, shm, sizeof(SHM)); - LOCK_GUARD(shm_list->shm_lock); + BAIDU_SCOPED_LOCK(shm_list->shm_lock); new_shm_node->next = nullptr; new_shm_node->prev = shm_list->tail; if (shm_list->tail) { diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index d4d4e6d5de..912e331c1e 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -15,10 +15,10 @@ // specific language governing permissions and limitations // under the License. -#include #include #include "bthread/bthread.h" // bthread_usleep #include "bthread/unstable.h" // bthread_timer_add/del +#include "butil/atomicops.h" #include "butil/time.h" #include "brpc/ubshm/timer/timer_mgr.h" @@ -46,18 +46,18 @@ enum UbrTimerState { // 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. struct UbrTimerTask { - UbrTimerId* slot; - std::atomic id; + butil::atomic* slot; + butil::atomic id; void* (*cb)(void*); void* arg; UbrTimerBackoffFn backoff; uint64_t interval_us; // timer thread only bool periodic; - std::atomic state; // kStarting/kScheduled/kDead - std::atomic stopped; - std::atomic ref; - std::atomic join_pending; // a DelAndWait is waiting - std::atomic done; // refs hit zero, joiner frees + butil::atomic state; // kStarting/kScheduled/kDead + butil::atomic stopped; + butil::atomic ref; + butil::atomic join_pending; // a DelAndWait is waiting + butil::atomic done; // refs hit zero, joiner frees }; namespace { @@ -117,9 +117,7 @@ void UbrTimerOnFire(void* p) { // RMW and seq_cst does not order the store-buffer case -- ownership of // the slot is the single arbiter. UbrTimerId expected = task; - const bool owned = - __atomic_compare_exchange_n(task->slot, &expected, (UbrTimerId) nullptr, - false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST); + const bool owned = task->slot->compare_exchange_strong(expected, nullptr); if (owned) { task->cb(task->arg); } @@ -129,20 +127,20 @@ void UbrTimerOnFire(void* p) { } } -UbrTimerTask* TakeOutTask(UbrTimerId* slot) { - return __atomic_exchange_n(slot, (UbrTimerId) nullptr, __ATOMIC_SEQ_CST); +UbrTimerTask* TakeOutTask(butil::atomic* slot) { + return slot->exchange(nullptr); } -RETURN_CODE TimerStartInternal(UbrTimerId* slot, uint64_t delay_us, +RETURN_CODE TimerStartInternal(butil::atomic* slot, uint64_t delay_us, uint64_t interval_us, void* (*cb)(void*), void* arg, UbrTimerBackoffFn backoff) { - if (UNLIKELY(slot == nullptr || cb == nullptr)) { + if (BAIDU_UNLIKELY(slot == nullptr || cb == nullptr)) { LOG(ERROR) << "Ubr timer start invalid argument, slot=" << slot; return UBRING_ERR; } UbrTimerTask* task = new (std::nothrow) UbrTimerTask(); - if (UNLIKELY(task == nullptr)) { + if (BAIDU_UNLIKELY(task == nullptr)) { LOG(ERROR) << "Fail to malloc ubring timer task."; return UBRING_ERR; } @@ -162,24 +160,20 @@ RETURN_CODE TimerStartInternal(UbrTimerId* slot, uint64_t delay_us, // 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. UbrTimerId expected = nullptr; - if (!__atomic_compare_exchange_n(slot, &expected, (UbrTimerId) task, false, - __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) { + if (!slot->compare_exchange_strong(expected, task)) { LOG(ERROR) << "Ubr timer start refused, slot already occupied"; delete task; // never published return UBRING_ERR; } bthread_timer_t id = 0; - if (UNLIKELY(bthread_timer_add( + if (BAIDU_UNLIKELY(bthread_timer_add( &id, butil::microseconds_from_now((int64_t)delay_us), UbrTimerOnFire, task) != 0)) { LOG(ERROR) << "Fail to add ubring timer"; task->state.store(kDead); // wake DelAndWait waiters expected = task; - const bool owned = - __atomic_compare_exchange_n(slot, &expected, (UbrTimerId) nullptr, - false, __ATOMIC_SEQ_CST, - __ATOMIC_SEQ_CST); + const bool owned = slot->compare_exchange_strong(expected, nullptr); ReleaseRef(task); // schedule, never ran if (owned) { ReleaseRef(task); // owner @@ -201,13 +195,13 @@ RETURN_CODE TimerStartInternal(UbrTimerId* slot, uint64_t delay_us, } // namespace -RETURN_CODE UbrTimerStart(UbrTimerId* slot, uint64_t delay_us, +RETURN_CODE UbrTimerStart(butil::atomic* slot, uint64_t delay_us, uint64_t interval_us, void* (*cb)(void*), void* arg, UbrTimerBackoffFn backoff) { return TimerStartInternal(slot, delay_us, interval_us, cb, arg, backoff); } -int UbrTimerDel(UbrTimerId* slot) { +int UbrTimerDel(butil::atomic* slot) { if (slot == nullptr) { return 1; } @@ -242,7 +236,7 @@ int UbrTimerDel(UbrTimerId* slot) { // callback may still complete. } -void UbrTimerDelAndWait(UbrTimerId* slot) { +void UbrTimerDelAndWait(butil::atomic* slot) { if (slot == nullptr) { return; } diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index f4d244eceb..24ef2e0b15 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -39,8 +39,10 @@ typedef uint64_t (*UbrTimerBackoffFn)(void* arg, uint64_t cur_interval_us); // re-arm itself after every run until deleted. One-shot timers release // their handle slot before running the callback, so the callback may free // the object that stores the slot; the task object itself is released -// automatically. -RETURN_CODE UbrTimerStart(UbrTimerId* slot, uint64_t delay_us, +// automatically. `slot' must point to a real butil::atomic +// object, so start and delete can race on one RMW without reinterpreting +// plain storage as an atomic. +RETURN_CODE UbrTimerStart(butil::atomic* slot, uint64_t delay_us, uint64_t interval_us, void* (*cb)(void*), void* arg, UbrTimerBackoffFn backoff = nullptr); @@ -59,7 +61,7 @@ RETURN_CODE UbrTimerStart(UbrTimerId* slot, uint64_t delay_us, // Returns 1 when this caller did not acquire timer ownership. The callback, // another deleter, or the scheduling path is responsible for settling the // timer resources, so this caller must not reclaim them. -int UbrTimerDel(UbrTimerId* slot); +int UbrTimerDel(butil::atomic* slot); // Delete and wait until a possibly running callback finished, so the // caller can free resources reachable from `arg'. Never call this on the @@ -68,7 +70,7 @@ int UbrTimerDel(UbrTimerId* slot); // 1) and cannot be waited for through the slot. The wait polls with // bthread_usleep, which degrades to ::usleep on plain pthread callers // (e.g. process-exit paths). -void UbrTimerDelAndWait(UbrTimerId* slot); +void UbrTimerDelAndWait(butil::atomic* slot); } // namespace ubring } // namespace brpc diff --git a/src/brpc/ubshm/ub_endpoint.cpp b/src/brpc/ubshm/ub_endpoint.cpp index 089ccef5bd..51003d56a9 100644 --- a/src/brpc/ubshm/ub_endpoint.cpp +++ b/src/brpc/ubshm/ub_endpoint.cpp @@ -533,7 +533,7 @@ void* UBShmEndpoint::ProcessHandshakeAtServer(void* arg) { } int result = snprintf(local_trx_shm.name, SHM_MAX_NAME_BUFF_LEN, "%s_%s", client_name, SERVER_SHM_NAME_SUFFIX); - if (UNLIKELY(result < 0)) { + if (BAIDU_UNLIKELY(result < 0)) { LOG(WARNING) << "Copy client shared memory name failed, ret=" << result; ub_transport->_ub_state = UBShmTransport::UB_OFF; } @@ -641,14 +641,14 @@ ssize_t UBShmEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata) { ssize_t nw = 0; errno = 0; nw = _ub_ring->UbrTrxWritev(vec, nvec); - if (UNLIKELY(nw == -1)) { + if (BAIDU_UNLIKELY(nw == -1)) { if (errno == EMSGSIZE) { LOG(ERROR) << "Non-blocking send msg failed, message is larger than ubring capacity."; } else { LOG(ERROR) << "Non-blocking send msg in failed, connection has been closed."; errno = EPIPE; } - } else if (UNLIKELY(nw == UBRING_RETRY)) { + } else if (BAIDU_UNLIKELY(nw == UBRING_RETRY)) { errno = EAGAIN; nw = -1; } @@ -833,7 +833,7 @@ int UBShmEndpoint::PollingModeInitialize(bthread_tag_t tag, } struct FnArgs { Poller* poller; - std::atomic* running; + butil::atomic* running; }; auto fn = [](void* p) -> void* { std::unique_ptr args(static_cast(p)); diff --git a/src/brpc/ubshm/ub_endpoint.h b/src/brpc/ubshm/ub_endpoint.h index a29a0927f7..c93fa330de 100644 --- a/src/brpc/ubshm/ub_endpoint.h +++ b/src/brpc/ubshm/ub_endpoint.h @@ -234,7 +234,7 @@ friend class Socket; struct BAIDU_CACHELINE_ALIGNMENT PollerGroup { PollerGroup() : pollers(FLAGS_ub_poller_num), running(false) {} std::vector pollers; - std::atomic running; + butil::atomic running; }; static std::vector _poller_groups; diff --git a/src/brpc/ubshm/ub_ring.cpp b/src/brpc/ubshm/ub_ring.cpp index 25b6f2749b..3ba1af8aa8 100644 --- a/src/brpc/ubshm/ub_ring.cpp +++ b/src/brpc/ubshm/ub_ring.cpp @@ -56,12 +56,12 @@ UBRing::~UBRing() RETURN_CODE UBRing::UbrTrxMapShm(SHM *local_shm, SHM *remote_shm) { RETURN_CODE rc = UbrTrxMapLocalShm(local_shm); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Trx map local shared memory failed."; return rc; } rc = UbrTrxMapRemoteShm(remote_shm); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Trx map remote shared memory failed."; return rc; } @@ -69,10 +69,10 @@ RETURN_CODE UBRing::UbrTrxMapShm(SHM *local_shm, SHM *remote_shm) } static void UbrDoAsynClearWork(UbrTrx *trx, uint64_t expect_ubr_id) { - if (UNLIKELY(UBRing::UbrTrxFreeShm(trx) != UBRING_OK)) { + if (BAIDU_UNLIKELY(UBRing::UbrTrxFreeShm(trx) != UBRING_OK)) { LOG(ERROR) << "Trx close, wait for local shm " << trx->local_shm.name << " free fail."; } - if (UNLIKELY(UBRingManager::ReleaseUbrTrxFromMgr(trx, expect_ubr_id) != UBRING_OK)) { + if (BAIDU_UNLIKELY(UBRingManager::ReleaseUbrTrxFromMgr(trx, expect_ubr_id) != UBRING_OK)) { LOG(ERROR) << "Trx close, release shm " << trx->local_shm.name << " trx failed."; } } @@ -88,7 +88,7 @@ static void UbrDoPassiveClearWork(UbrTrx *trx, uint64_t expect_ubr_id) { LOG(ERROR) << "Trx passive clear, delete local shm " << trx->local_shm.name << " failed. ret=" << rc; } - if (UNLIKELY(UBRingManager::ReleaseUbrTrxFromMgr(trx, expect_ubr_id) != UBRING_OK)) { + if (BAIDU_UNLIKELY(UBRingManager::ReleaseUbrTrxFromMgr(trx, expect_ubr_id) != UBRING_OK)) { LOG(ERROR) << "Trx passive clear, release shm " << trx->local_shm.name << " trx failed."; } } @@ -99,14 +99,14 @@ static void UbrDoPassiveClearWork(UbrTrx *trx, uint64_t expect_ubr_id) { // body, used directly when the timer cannot be started. static RETURN_CODE UbrScheduleClearTimer(UbrTrx *trx, void* (*cb)(void*), void (*work)(UbrTrx*, uint64_t)) { - if (UNLIKELY(trx == nullptr || trx->local_shm.addr == nullptr)) { + if (BAIDU_UNLIKELY(trx == nullptr || trx->local_shm.addr == nullptr)) { return UBRING_OK; // released trx, stale event } - if (__atomic_load_n(&trx->cleanup_ctl, __ATOMIC_SEQ_CST) != nullptr) { + if (trx->cleanup_ctl.load() != nullptr) { return UBRING_OK; // cleanup already scheduled } auto* ctl = new (std::nothrow) UbrCleanupCtl(); - if (UNLIKELY(ctl == nullptr)) { + if (BAIDU_UNLIKELY(ctl == nullptr)) { LOG(ERROR) << "Fail to malloc ubr cleanup ctl."; return UBRING_ERR; } @@ -119,8 +119,7 @@ static RETURN_CODE UbrScheduleClearTimer(UbrTrx *trx, void* (*cb)(void*), // TryPublishUnitCleanupCtl UbrCleanupCtl* expected = nullptr; - if (!__atomic_compare_exchange_n(&trx->cleanup_ctl, &expected, ctl, false, - __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) { + if (!trx->cleanup_ctl.compare_exchange_strong(expected, ctl)) { delete ctl; // another schedule won return UBRING_OK; } @@ -130,16 +129,14 @@ static RETURN_CODE UbrScheduleClearTimer(UbrTrx *trx, void* (*cb)(void*), // anchor: force close or the new occupant owns it now. Nothing // was armed yet -- just undo the trx-side publication. expected = ctl; - __atomic_compare_exchange_n(&trx->cleanup_ctl, &expected, - (UbrCleanupCtl*) nullptr, false, - __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST); + trx->cleanup_ctl.compare_exchange_strong(expected, nullptr); ctl->ReleaseRef(); // timer/callback reference, never armed ctl->ReleaseRef(); // starter reference return UBRING_OK; } RETURN_CODE rc = UbrTimerStart(&ctl->timer, (uint64_t)FLAGS_ub_flying_io_timeout_s * SEC_TO_USEC, 0, cb, ctl); - if (UNLIKELY(rc != UBRING_OK)) { + 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 @@ -156,24 +153,20 @@ static RETURN_CODE UbrScheduleClearTimer(UbrTrx *trx, void* (*cb)(void*), // close that this cleanup is owned (RUNNING/DONE), otherwise it // would fall into its no-ctl branch and clean the trx again. UbrCleanupCtl* published = ctl; - __atomic_compare_exchange_n(&trx->cleanup_ctl, &published, - (UbrCleanupCtl*) nullptr, false, - __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST); + trx->cleanup_ctl.compare_exchange_strong(published, nullptr); UBRingManager::DetachUnitCleanupCtl(trx->trx_mgr_index, ctl); } ctl->ReleaseRef(); // timer/callback reference ctl->ReleaseRef(); // starter reference return UBRING_ERR; } - if (UNLIKELY(ATOMIC_LOAD(trx->ubr_id) != ctl->ubr_id)) { + if (BAIDU_UNLIKELY(ATOMIC_LOAD(trx->ubr_id) != ctl->ubr_id)) { // Published onto a slot that was released and reused meanwhile. if (UbrTimerDel(&ctl->timer) == 0) { ctl->ReleaseRef(); // timer/callback reference } UbrCleanupCtl* published = ctl; - if (!__atomic_compare_exchange_n(&trx->cleanup_ctl, &published, - (UbrCleanupCtl*) nullptr, false, - __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) { + if (!trx->cleanup_ctl.compare_exchange_strong(published, nullptr)) { published = nullptr; } UBRingManager::DetachUnitCleanupCtl(trx->trx_mgr_index, ctl); @@ -186,7 +179,7 @@ static RETURN_CODE UbrScheduleClearTimer(UbrTrx *trx, void* (*cb)(void*), RETURN_CODE UBRing::UbrTrxClose() { RETURN_CODE close_check_rc = UbrTrxCloseCheck(_trx); - if (UNLIKELY(close_check_rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(close_check_rc != UBRING_OK)) { if (close_check_rc == UBRING_REENTRY) { LOG(INFO) << "Trx close skipped, already closing, local name=" << _trx->local_shm.name; return UBRING_OK; @@ -242,10 +235,10 @@ RETURN_CODE UBRing::UbrTrxClose() { if (_trx->ubr_tx.remote_rx_event_q.addr != nullptr) { ((UbrEventQMsg *)_trx->ubr_tx.remote_rx_event_q.addr)->flag = UBR_STATE_CLOSED; } - if (UNLIKELY(UbrTrxFreeShm(_trx) != UBRING_OK)) { + if (BAIDU_UNLIKELY(UbrTrxFreeShm(_trx) != UBRING_OK)) { LOG(WARNING) << "Force close, local shm " << _trx->local_shm.name << " free failed."; } - if (UNLIKELY(UBRingManager::ReleaseUbrTrxFromMgr(_trx, expect_ubr_id) != UBRING_OK)) { + if (BAIDU_UNLIKELY(UBRingManager::ReleaseUbrTrxFromMgr(_trx, expect_ubr_id) != UBRING_OK)) { LOG(WARNING) << "Force close, release trx " << _trx->local_shm.name << " failed."; } if (ctl != nullptr) { @@ -261,7 +254,7 @@ RETURN_CODE UBRing::UbrTrxClose() { } _trx->ubr_rx.trx_state = UBR_STATE_CLOSED; RETURN_CODE rc; - if (UNLIKELY((rc = ClearTrxResource(_trx)) != UBRING_OK)) { + if (BAIDU_UNLIKELY((rc = ClearTrxResource(_trx)) != UBRING_OK)) { if (rc == UBRING_REENTRY) { LOG(INFO) << "Trx close, peer is closing, trx local name=" << _trx->local_shm.name; return UBRING_OK; @@ -286,7 +279,7 @@ static uint64_t UbrCloseTimerBackoff(void* arg, uint64_t cur_interval_us) { auto* local_tx_event_q = (UbrEventQMsg *)trx->ubr_tx.local_tx_event_q.addr; const uint64_t in_io_id = ATOMIC_LOAD(trx->ubr_rx.in_io_id); const uint64_t out_io_id = ATOMIC_LOAD(trx->ubr_tx.out_io_id); - if (UNLIKELY(local_rx_event_q == nullptr)) { + if (BAIDU_UNLIKELY(local_rx_event_q == nullptr)) { return (uint64_t)FLAGS_ub_event_queue_timer_interval_us; } const bool has_traffic = (in_io_id != trx->close_chk_in_io_id) || @@ -308,7 +301,7 @@ static uint64_t UbrCloseTimerBackoff(void* arg, uint64_t cur_interval_us) { } RETURN_CODE UBRing::UbrAddCloseTimer() { - if (UNLIKELY(_trx == nullptr)) { + if (BAIDU_UNLIKELY(_trx == nullptr)) { LOG(ERROR) << "Trx add close timer failed, trx is null."; return UBRING_ERR; } @@ -319,7 +312,7 @@ RETURN_CODE UBRing::UbrAddCloseTimer() { RETURN_CODE rc = UbrTimerStart(&_trx->close_timer, 0, interval_us, UbrTrxCloseCallback, (void*)_trx, UbrCloseTimerBackoff); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Start ubr close timer failed, trx local name=" << _trx->local_shm.name; return UBRING_ERR; } @@ -327,12 +320,12 @@ RETURN_CODE UBRing::UbrAddCloseTimer() { } RETURN_CODE UBRing::UbrAddTimer() { - if (UNLIKELY(UbrAddCloseTimer() != UBRING_OK)) { + if (BAIDU_UNLIKELY(UbrAddCloseTimer() != UBRING_OK)) { LOG(ERROR) << "Ubr " << _trx->local_shm.name << " add closed timer failed."; return UBRING_ERR; } - if (UNLIKELY(UbrAddHBTimer() != UBRING_OK)) { + if (BAIDU_UNLIKELY(UbrAddHBTimer() != UBRING_OK)) { UbrTimerDelAndWait(&_trx->close_timer); LOG(ERROR) << "Ubr " << _trx->local_shm.name << " add heartbeat timer failed."; return UBRING_ERR; @@ -342,7 +335,7 @@ RETURN_CODE UBRing::UbrAddTimer() { void* UBRing::UbrTrxCloseCallback(void* args) { auto* trx = (UbrTrx*) args; - if (UNLIKELY(UBRing::UbrTrxCallbackCheck(trx) != UBRING_OK)) { + if (BAIDU_UNLIKELY(UBRing::UbrTrxCallbackCheck(trx) != UBRING_OK)) { return nullptr; } @@ -369,7 +362,7 @@ void* UBRing::UbrTrxCloseCallback(void* args) { } remote_rx_event_q->flag = UBR_STATE_CLOSED; RETURN_CODE clear_rc = ClearTrxResource(trx); - if (UNLIKELY(clear_rc != UBRING_OK && clear_rc != UBRING_REENTRY)) { + if (BAIDU_UNLIKELY(clear_rc != UBRING_OK && clear_rc != UBRING_REENTRY)) { LOG(ERROR) << "Trx close callback failed, " << trx->local_shm.name << " clear trx resource failed."; break; } @@ -378,7 +371,7 @@ void* UBRing::UbrTrxCloseCallback(void* args) { } RETURN_CODE UBRing::UbrAddHBTimer() { - if (UNLIKELY(_trx == nullptr)) { + if (BAIDU_UNLIKELY(_trx == nullptr)) { LOG(ERROR) << "Trx add heartbeat timer failed, trx is null."; return UBRING_ERR; } @@ -386,7 +379,7 @@ RETURN_CODE UBRing::UbrAddHBTimer() { const uint64_t interval_us = (uint64_t)FLAGS_ub_hb_timer_interval_s * SEC_TO_USEC; RETURN_CODE rc = UbrTimerStart(&_trx->hb_timer, 0, interval_us, UbrTrxHBCallback, (void*)_trx); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Start ubr heartbeat timer failed."; return UBRING_ERR; } @@ -395,7 +388,7 @@ RETURN_CODE UBRing::UbrAddHBTimer() { RETURN_CODE UBRing::UbrPassiveClearTrx(UbrTrx *trx) { RETURN_CODE passive_close_check_rc = UbrTrxCloseCheck(trx); - if (UNLIKELY(passive_close_check_rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(passive_close_check_rc != UBRING_OK)) { if (passive_close_check_rc == UBRING_REENTRY) { LOG(INFO) << "Passive close skipped, active close in progress, name=" << trx->local_shm.name; return ClearTrxResource(trx); @@ -414,7 +407,7 @@ RETURN_CODE UBRing::UbrPassiveClearTrx(UbrTrx *trx) { void* UBRing::UbrPassiveClearCallback(void* args) { auto* ctl = (UbrCleanupCtl*)args; - if (UNLIKELY(ctl == nullptr)) { + if (BAIDU_UNLIKELY(ctl == nullptr)) { LOG(ERROR) << "Trx passive clear callback failed, ctl is null."; return nullptr; } @@ -427,7 +420,7 @@ void* UBRing::UbrPassiveClearCallback(void* args) { return nullptr; } UbrTrx* trx = ctl->trx; - if (UNLIKELY(UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id))) { + if (BAIDU_UNLIKELY(UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id))) { UbrDoPassiveClearWork(trx, ctl->ubr_id); } ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE); @@ -437,13 +430,13 @@ void* UBRing::UbrPassiveClearCallback(void* args) { void* UBRing::UbrTrxHBCallback(void* args) { auto* trx = (UbrTrx*) args; - if (UNLIKELY(UbrTrxCallbackCheck(trx) != UBRING_OK)) { + if (BAIDU_UNLIKELY(UbrTrxCallbackCheck(trx) != UBRING_OK)) { return nullptr; } auto* local_data_status = (UbrDataStatusQMsg *)trx->ubr_tx.local_data_status_q.addr; auto* remote_data_status = (UbrDataStatusQMsg *)trx->ubr_rx.remote_data_status_q.addr; - if (UNLIKELY(local_data_status == nullptr || remote_data_status == nullptr)) { + if (BAIDU_UNLIKELY(local_data_status == nullptr || remote_data_status == nullptr)) { LOG(ERROR) << "Heartbeat error, datastatus is NULL."; return nullptr; } @@ -473,7 +466,7 @@ void* UBRing::UbrTrxHBCallback(void* args) { } RETURN_CODE UBRing::UbrAddAsynClearTimer(UbrTrx *trx) { - if (UNLIKELY(trx == nullptr)) { + if (BAIDU_UNLIKELY(trx == nullptr)) { LOG(ERROR) << "Trx add close timer failed, trx is null."; return UBRING_ERR; } @@ -483,7 +476,7 @@ RETURN_CODE UBRing::UbrAddAsynClearTimer(UbrTrx *trx) { void *UBRing::UbrAsynClearCallback(void *args) { auto* ctl = (UbrCleanupCtl*) args; - if (UNLIKELY(ctl == nullptr)) { + if (BAIDU_UNLIKELY(ctl == nullptr)) { LOG(ERROR) << "Trx close, ctl is null."; return nullptr; } @@ -496,7 +489,7 @@ void *UBRing::UbrAsynClearCallback(void *args) return nullptr; } UbrTrx* trx = ctl->trx; - if (UNLIKELY(UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id))) { + if (BAIDU_UNLIKELY(UBRingManager::IsUbrTrxSlotUsed(trx->trx_mgr_index, ctl->ubr_id))) { UbrDoAsynClearWork(trx, ctl->ubr_id); } ATOMIC_STORE(ctl->state, UBR_CLEANUP_DONE); @@ -506,7 +499,7 @@ void *UBRing::UbrAsynClearCallback(void *args) int UBRing::UbrTrxSend(const void *buf, uint32_t buf_len) { - if (UNLIKELY(CheckTrxSendPreCheck(_trx) != UBRING_OK)) { + if (BAIDU_UNLIKELY(CheckTrxSendPreCheck(_trx) != UBRING_OK)) { return UBRING_ERR; } // 1.2 Calculate space @@ -549,7 +542,7 @@ int UBRing::UbrTrxSend(const void *buf, uint32_t buf_len) int UBRing::UbrTrxRecv(void *buf, uint32_t buf_len) { RETURN_CODE rc = UBRING_OK; - if (UNLIKELY((rc = CheckTrxRecvParam(_trx, buf, buf_len)) != UBRING_OK)) { + if (BAIDU_UNLIKELY((rc = CheckTrxRecvParam(_trx, buf, buf_len)) != UBRING_OK)) { return (rc == UBR_NOT_CONNECTED) ? 0 : rc; } UbrMsgFormat *data_msg = (UbrMsgFormat *)_trx->ubr_rx.local_data_q.addr; @@ -564,7 +557,7 @@ int UBRing::UbrTrxRecv(void *buf, uint32_t buf_len) int UBRing::UbrTrxRecvBlockMode(uint8_t *dest, uint32_t buf_len) { RETURN_CODE rc = UBRING_OK; - if (UNLIKELY((rc = CheckTrxRecvParam(_trx, dest, buf_len)) != UBRING_OK)) { + if (BAIDU_UNLIKELY((rc = CheckTrxRecvParam(_trx, dest, buf_len)) != UBRING_OK)) { return (rc == UBR_NOT_CONNECTED) ? 0 : rc; } @@ -577,7 +570,7 @@ int UBRing::UbrTrxRecvBlockMode(uint8_t *dest, uint32_t buf_len) bool need_update_epoll_eof_pos = ubr_rx->read_pos == ubr_rx->ep_eof_pos; while (not_eof_encountered && remaining_len > 0) { - if (UNLIKELY(CheckTrxRecvPreCheck(_trx) != UBRING_OK)) { + if (BAIDU_UNLIKELY(CheckTrxRecvPreCheck(_trx) != UBRING_OK)) { return UBRING_ERR; } UbrMsgFormat *current_chunk = &data_msg[ubr_rx->read_pos]; @@ -594,7 +587,7 @@ int UBRing::UbrTrxRecvBlockMode(uint8_t *dest, uint32_t buf_len) } uint8_t chunk_msg_len = current_chunk->header[UBR_MSG_LEN_INDEX]; uint8_t cur_index = current_chunk->header[UBR_MSG_CUR_INDEX]; - if (UNLIKELY(!IsRecvChunkHeaderValid(chunk_msg_len, cur_index))) { + if (BAIDU_UNLIKELY(!IsRecvChunkHeaderValid(chunk_msg_len, cur_index))) { LOG(ERROR) << "Trx recv failed, invalid chunk header msg_len=" << (uint32_t)chunk_msg_len << " cur_index=" << (uint32_t)cur_index; errno = EBADMSG; @@ -607,7 +600,7 @@ int UBRing::UbrTrxRecvBlockMode(uint8_t *dest, uint32_t buf_len) total_copied += copy_len; remaining_len -= copy_len; current_chunk->header[UBR_MSG_CUR_INDEX] += (uint8_t)copy_len; - if (LIKELY(current_chunk->header[UBR_MSG_CUR_INDEX] == chunk_msg_len)) { + if (BAIDU_LIKELY(current_chunk->header[UBR_MSG_CUR_INDEX] == chunk_msg_len)) { current_chunk->header[UBR_MSG_FLAG_INDEX] = UBR_MSG_CHUNK_NONE; UpdateDataQTail(_trx); ubr_rx->read_pos = (ubr_rx->read_pos + 1) % ubr_rx->capacity; @@ -621,7 +614,7 @@ int UBRing::UbrTrxRecvBlockMode(uint8_t *dest, uint32_t buf_len) ssize_t UBRing::UbrTrxWritev(const struct iovec *iov, int iovcnt) { - if (UNLIKELY(CheckTrxSendPreCheck(_trx) != UBRING_OK)) { + if (BAIDU_UNLIKELY(CheckTrxSendPreCheck(_trx) != UBRING_OK)) { return UBRING_ERR; } @@ -678,7 +671,7 @@ ssize_t UBRing::UbrTrxWritev(const struct iovec *iov, int iovcnt) ssize_t UBRing::UbrTrxReadv(const struct iovec *iov, int iovcnt) { RETURN_CODE rc = UBRING_OK; - if (UNLIKELY((rc = CheckTrxRecvParam(_trx, iov, (uint32_t)iovcnt)) != UBRING_OK)) { + if (BAIDU_UNLIKELY((rc = CheckTrxRecvParam(_trx, iov, (uint32_t)iovcnt)) != UBRING_OK)) { return (rc == UBR_NOT_CONNECTED) ? 0 : rc; } UbrMsgFormat *data_msg = (UbrMsgFormat *)_trx->ubr_rx.local_data_q.addr; @@ -689,7 +682,7 @@ ssize_t UBRing::UbrTrxReadv(const struct iovec *iov, int iovcnt) return -1; } ssize_t nr = UbrTrxReadvBlockMode(iov, iovcnt); - if (UNLIKELY(nr == -1)) { + if (BAIDU_UNLIKELY(nr == -1)) { LOG(ERROR) << "Non-blocking readv msg in failed, connection has been closed."; errno = EPIPE; return -1; @@ -700,7 +693,7 @@ ssize_t UBRing::UbrTrxReadv(const struct iovec *iov, int iovcnt) ssize_t UBRing::UbrTrxReadvBlockMode(const struct iovec *iov, int iovcnt) { RETURN_CODE rc = UBRING_OK; - if (UNLIKELY((rc = CheckTrxRecvParam(_trx, iov, (uint32_t)iovcnt)) != UBRING_OK)) { + if (BAIDU_UNLIKELY((rc = CheckTrxRecvParam(_trx, iov, (uint32_t)iovcnt)) != UBRING_OK)) { return (rc == UBR_NOT_CONNECTED) ? 0 : rc; } @@ -720,15 +713,15 @@ ssize_t UBRing::UbrTrxReadvBlockMode(const struct iovec *iov, int iovcnt) RETURN_CODE UBRing::IsUbrTrxReadable(uint32_t ep_event) { - if (UNLIKELY(_trx == nullptr)) { + if (BAIDU_UNLIKELY(_trx == nullptr)) { LOG(ERROR) << "The trx to be checked is NULL."; return UBRING_ERR; } - if (UNLIKELY(_trx->local_shm.addr == nullptr)) { + if (BAIDU_UNLIKELY(_trx->local_shm.addr == nullptr)) { LOG(ERROR) << "The trx local_shm to be checked is NULL."; return UBRING_ERR; } - if (UNLIKELY(_trx->ubr_tx.trx_state != UBR_STATE_CONNECTED)) { + if (BAIDU_UNLIKELY(_trx->ubr_tx.trx_state != UBR_STATE_CONNECTED)) { return UBRING_ERR; } @@ -755,24 +748,24 @@ RETURN_CODE UBRing::IsUbrTrxReadable(uint32_t ep_event) RETURN_CODE UBRing::IsUbrTrxWriteable(uint32_t ep_event) { - if (UNLIKELY(_trx == nullptr)) { + if (BAIDU_UNLIKELY(_trx == nullptr)) { LOG(ERROR) << "The trx to be checked is NULL."; return UBRING_ERR; } - if (UNLIKELY(_trx->local_shm.addr == nullptr)) { + if (BAIDU_UNLIKELY(_trx->local_shm.addr == nullptr)) { LOG(ERROR) << "The trx local_shm to be checked is NULL."; return UBRING_ERR; } - if (UNLIKELY((UbrEventQMsg *)_trx->ubr_tx.local_tx_event_q.addr == nullptr)) { + if (BAIDU_UNLIKELY((UbrEventQMsg *)_trx->ubr_tx.local_tx_event_q.addr == nullptr)) { LOG(ERROR) << "The trx local_tx_event_q addr is NULL."; return UBRING_ERR; } - if (UNLIKELY((UbrEventQMsg *)_trx->ubr_tx.local_data_status_q.addr == nullptr)) { + if (BAIDU_UNLIKELY((UbrEventQMsg *)_trx->ubr_tx.local_data_status_q.addr == nullptr)) { LOG(ERROR) << "The trx local_data_status_q addr is NULL."; return UBRING_ERR; } - if (UNLIKELY(_trx->ubr_tx.trx_state != UBR_STATE_CONNECTED)) { + if (BAIDU_UNLIKELY(_trx->ubr_tx.trx_state != UBR_STATE_CONNECTED)) { LOG(ERROR) << "The trx is not connected state."; return UBRING_ERR; } @@ -816,13 +809,13 @@ RETURN_CODE UBRing::UbrTrxFreeShm(UbrTrx *trx) RETURN_CODE rc = UBRING_OK; rc = ShmMunmap(&trx->local_shm); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Trx close, local unmap " << trx->local_shm.name << " shm fail."; return UBRING_ERR; } rc = ShmFree(&trx->local_shm); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { if (rc != SHM_ERR_RESOURCE_ATTACHED && rc != SHM_ERR_NOT_FOUND) { LOG(ERROR) << "Wait for " << trx->local_shm.name << " local shm free fail."; return UBRING_ERR; @@ -843,7 +836,7 @@ RETURN_CODE UBRing::UbrTrxFreeShm(UbrTrx *trx) RETURN_CODE UBRing::UbrUnlinkLocalShm() { - if (UNLIKELY(_trx == nullptr)) { + if (BAIDU_UNLIKELY(_trx == nullptr)) { return UBRING_ERR; } RETURN_CODE rc = ShmFree(&_trx->local_shm); @@ -896,11 +889,11 @@ void UBRing::PrewriteUbrRx(UbrRx *rx) RETURN_CODE UBRing::UbrTrxMapLocalShm(SHM *local_shm) { - if (UNLIKELY(_trx == nullptr)) { + if (BAIDU_UNLIKELY(_trx == nullptr)) { LOG(ERROR) << "Trx map Shared memory failed, trx is null."; return UBRING_ERR; } - if (UNLIKELY(local_shm == nullptr || local_shm->addr == nullptr)) { + if (BAIDU_UNLIKELY(local_shm == nullptr || local_shm->addr == nullptr)) { LOG(ERROR) << "Trx map Shared memory failed, local_shm is null or addr is NULL."; return UBRING_ERR; } @@ -919,11 +912,11 @@ RETURN_CODE UBRing::UbrTrxMapLocalShm(SHM *local_shm) RETURN_CODE UBRing::UbrTrxMapRemoteShm(SHM *remote_shm) { - if (UNLIKELY(_trx == nullptr)) { + if (BAIDU_UNLIKELY(_trx == nullptr)) { LOG(ERROR) << "Trx map Shared memory failed, trx is null."; return UBRING_ERR; } - if (UNLIKELY(remote_shm == nullptr || remote_shm->addr == nullptr)) { + if (BAIDU_UNLIKELY(remote_shm == nullptr || remote_shm->addr == nullptr)) { LOG(ERROR) << "Trx map Shared memory failed, remote_shm is null or addr is NULL."; return UBRING_ERR; } @@ -943,7 +936,7 @@ RETURN_CODE UBRing::UbrTrxMapRemoteShm(SHM *remote_shm) RETURN_CODE UBRing::UbrServerTrxInit(SHM *local_shm, SHM *remote_shm) { RETURN_CODE rc = UbrTrxMapShm(local_shm, remote_shm); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) <<"Trx map shared memory failed."; return rc; } @@ -953,7 +946,7 @@ RETURN_CODE UBRing::UbrServerTrxInit(SHM *local_shm, SHM *remote_shm) _trx->ubr_rx.capacity = local_data_msg_cap; _trx->ubr_tx.capacity = remote_data_msg_cap; rc = UBRingManager::GetUbrDealMsgMaxCnt(_trx->ubr_rx.capacity, &_trx->ubr_rx.deal_msg_max_cnt); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Get ubring deal msg max cnt."; return rc; } @@ -963,7 +956,7 @@ RETURN_CODE UBRing::UbrServerTrxInit(SHM *local_shm, SHM *remote_shm) ((UbrDataStatusQMsg *)(_trx->ubr_tx.local_data_status_q.addr))->tail = remote_data_msg_cap - 1; ((UbrDataStatusQMsg *)(_trx->ubr_rx.remote_data_status_q.addr))->tail = local_data_msg_cap - 1; - if (UNLIKELY(UbrAddTimer() != UBRING_OK)) { + if (BAIDU_UNLIKELY(UbrAddTimer() != UBRING_OK)) { LOG(ERROR) << "Ubr add timer failed, local_name=" << local_shm->name; return UBRING_ERR; } @@ -982,26 +975,26 @@ RETURN_CODE UBRing::UbrServerTrxInit(SHM *local_shm, SHM *remote_shm) int UBRing::UbrAllocateServerShm(SHM* remote_trx_shm, SHM* local_trx_shm) { UbrSetSleepTask(UBR_TASK_ACCEPT_MAP_FRONT); - if (UNLIKELY((ShmRemoteMalloc(remote_trx_shm)) != UBRING_OK)) { + if (BAIDU_UNLIKELY((ShmRemoteMalloc(remote_trx_shm)) != UBRING_OK)) { LOG(ERROR) << "Trx apply remote shared memory failed."; return -1; } - if (UNLIKELY((ShmLocalCalloc(local_trx_shm)) != UBRING_OK)) { + if (BAIDU_UNLIKELY((ShmLocalCalloc(local_trx_shm)) != UBRING_OK)) { LOG(ERROR) << "Trx apply local shared memory failed."; ShmRemoteFree(remote_trx_shm); return -1; } UbrTrx **ubr_trx_ptr = &_trx; - if (UNLIKELY((UBRingManager::AcquireUbrTrxFromMgr(ubr_trx_ptr)) != UBRING_OK)) { + if (BAIDU_UNLIKELY((UBRingManager::AcquireUbrTrxFromMgr(ubr_trx_ptr)) != UBRING_OK)) { LOG(ERROR) << "Acquire ubrtrx failed."; ShmRemoteFree(remote_trx_shm); ShmLocalFree(local_trx_shm); return -1; } _trx->type = TCP_TRX; - if (UNLIKELY((UbrServerTrxInit(local_trx_shm, remote_trx_shm)) != UBRING_OK)) { + if (BAIDU_UNLIKELY((UbrServerTrxInit(local_trx_shm, remote_trx_shm)) != UBRING_OK)) { LOG(ERROR) << "Server trx init failed."; UbrTrxFreeShm(_trx); UBRingManager::ReleaseUbrTrxFromMgr(_trx, ATOMIC_LOAD(_trx->ubr_id)); @@ -1013,13 +1006,13 @@ int UBRing::UbrAllocateServerShm(SHM* remote_trx_shm, SHM* local_trx_shm) { int UBRing::UbrAllocateLocalShm(SHM *local_trx_shm, const char *shm_name) { - if (UNLIKELY((UBRingManager::AcquireUbrTrxFromMgr(&(_trx))) != UBRING_OK)) { + if (BAIDU_UNLIKELY((UBRingManager::AcquireUbrTrxFromMgr(&(_trx))) != UBRING_OK)) { LOG(ERROR) << "Acquire ubrtrx failed, local_name=" << shm_name; return -1; } _trx->type = TCP_TRX; - if (UNLIKELY((ApplyAndMapLocalShm(local_trx_shm, shm_name)) != UBRING_OK)) { + if (BAIDU_UNLIKELY((ApplyAndMapLocalShm(local_trx_shm, shm_name)) != UBRING_OK)) { LOG(ERROR) << "Trx apply or map local shared memory failed, local_name=" << shm_name; _trx = nullptr; return -1; @@ -1030,7 +1023,7 @@ int UBRing::UbrAllocateLocalShm(SHM *local_trx_shm, const char *shm_name) int UBRing::UbrMapRemoteShm(SHM *local_trx_shm, const char *local_name) { RETURN_CODE rc = UbrMapRemoteShmAddTimer(local_trx_shm, local_name); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Connect Trx failed, local shm name=" << local_trx_shm->name; return -1; } @@ -1056,19 +1049,19 @@ RETURN_CODE UBRing::UbrMapRemoteShmAddTimer(SHM *local_trx_shm, const char *loca SHM_NAME_PREFIX, local_name, SERVER_SHM_NAME_SUFFIX); - if (UNLIKELY(result < 0)) { + if (BAIDU_UNLIKELY(result < 0)) { LOG(ERROR) << "Copy server shared memory name failed, local_name=" << local_name << ", ret=" << result; return UBRING_ERR; } UbrSetSleepTask(UBR_TASK_CONNECT_MAP_FRONT); RETURN_CODE rc = ApplyAndMapRemoteShm(&remote_trx_shm); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Connect Trx map shared memory failed, remote shm=" << remote_trx_shm.name; return rc; } - if (UNLIKELY(UbrAddTimer() != UBRING_OK)) { + if (BAIDU_UNLIKELY(UbrAddTimer() != UBRING_OK)) { LOG(ERROR) << "Ubr add timer failed, local_name=" << local_name; ShmRemoteFree(&_trx->remote_shm); return UBRING_ERR; @@ -1090,7 +1083,7 @@ RETURN_CODE UBRing::UbrMapRemoteShmAddTimer(SHM *local_trx_shm, const char *loca RETURN_CODE UBRing::ApplyAndMapLocalShm(SHM *local_trx_shm, const char *local_name) { - if (UNLIKELY(_trx == nullptr || local_trx_shm == nullptr)) { + if (BAIDU_UNLIKELY(_trx == nullptr || local_trx_shm == nullptr)) { LOG(ERROR) << "Trx map Shared memory failed, trx is null, local_name=" << local_name; return UBRING_ERR; } @@ -1100,13 +1093,13 @@ RETURN_CODE UBRing::ApplyAndMapLocalShm(SHM *local_trx_shm, const char *local_na SHM_NAME_PREFIX, local_name, CLIENT_SHM_NAME_SUFFIX); - if (UNLIKELY(result < 0)) { + if (BAIDU_UNLIKELY(result < 0)) { LOG(ERROR) << "Copy client localTrx shared memory name failed, local_name=" << local_name << ", ret=" << result; return UBRING_ERR; } RETURN_CODE rc = ShmLocalCalloc(local_trx_shm); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Trx apply local shared memory failed, local shm name=" << local_trx_shm->name << ", rc=" << rc; if (rc == SHM_ERR_EXIST || rc == SHM_ERR_NOT_FOUND) { rc = UBR_ERR_ADDR_IN_USE; @@ -1115,7 +1108,7 @@ RETURN_CODE UBRing::ApplyAndMapLocalShm(SHM *local_trx_shm, const char *local_na return rc; } rc = UbrTrxMapLocalShm(local_trx_shm); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Trx map local shared memory failed, local shm name=" << local_trx_shm->name; ShmLocalFree(local_trx_shm); UBRingManager::ReleaseUbrTrxFromMgr(_trx, ATOMIC_LOAD(_trx->ubr_id)); @@ -1137,12 +1130,12 @@ RETURN_CODE UBRing::ApplyAndMapLocalShm(SHM *local_trx_shm, const char *local_na RETURN_CODE UBRing::ApplyAndMapRemoteShm(SHM *remote_trx_shm) { RETURN_CODE rc = ShmRemoteMalloc(remote_trx_shm); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Trx apply remote shared memory failed."; return rc; } rc = UbrTrxMapRemoteShm(remote_trx_shm); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Trx map shared memory failed."; ShmRemoteFree(remote_trx_shm); return rc; @@ -1173,13 +1166,13 @@ RETURN_CODE UBRing::WritevHasEnoughSpace(size_t buf_len) RETURN_CODE UBRing::UbrClearResourceCheck(UbrTrx *trx) { - if (UNLIKELY(trx == nullptr)) { + if (BAIDU_UNLIKELY(trx == nullptr)) { LOG(ERROR) << "Trx close failed, trx is null."; return UBRING_ERR; } UbrEventQMsg* local_tx_event_q = (UbrEventQMsg *)trx->ubr_tx.local_tx_event_q.addr; - if (UNLIKELY(local_tx_event_q == nullptr)) { + if (BAIDU_UNLIKELY(local_tx_event_q == nullptr)) { LOG(ERROR) << "Trx close failed, local_tx_event_q addr is NULL, trx local name=" << trx->local_shm.name; return UBRING_ERR; } @@ -1217,7 +1210,7 @@ RETURN_CODE UBRing::ClearTrxResource(UbrTrx *trx) RETURN_CODE UBRing::UbrTrxCloseCheck(UbrTrx *trx) { - if (UNLIKELY(trx == nullptr)) { + if (BAIDU_UNLIKELY(trx == nullptr)) { LOG(ERROR) << "Trx close failed, client trx is null."; return UBRING_ERR; } @@ -1227,7 +1220,7 @@ RETURN_CODE UBRing::UbrTrxCloseCheck(UbrTrx *trx) return UBRING_REENTRY; } - if (UNLIKELY(trx->ubr_tx.local_tx_event_q.addr == nullptr)) { + if (BAIDU_UNLIKELY(trx->ubr_tx.local_tx_event_q.addr == nullptr)) { LOG(ERROR) << "Trx close failed, local_tx_event_q addr is NULL, trx local name=" << trx->local_shm.name; return UBRING_ERR; } @@ -1242,7 +1235,7 @@ ssize_t UBRing::StartReadv(UbrTrx *trx, const struct iovec *iov, int iovcnt, siz UbrMsgFormat *data_msg = (UbrMsgFormat *)trx->ubr_rx.local_data_q.addr; bool not_eof_encountered = true; while (not_eof_encountered && remain_buf_len > 0) { - if (UNLIKELY(CheckTrxRecvPreCheck(trx) != UBRING_OK)) { + if (BAIDU_UNLIKELY(CheckTrxRecvPreCheck(trx) != UBRING_OK)) { return UBRING_ERR; } UbrMsgFormat *current_chunk = &data_msg[trx->ubr_rx.read_pos]; @@ -1259,7 +1252,7 @@ ssize_t UBRing::StartReadv(UbrTrx *trx, const struct iovec *iov, int iovcnt, siz } uint8_t chunk_msg_len = current_chunk->header[UBR_MSG_LEN_INDEX]; uint8_t cur_index = current_chunk->header[UBR_MSG_CUR_INDEX]; - if (UNLIKELY(!IsRecvChunkHeaderValid(chunk_msg_len, cur_index))) { + if (BAIDU_UNLIKELY(!IsRecvChunkHeaderValid(chunk_msg_len, cur_index))) { LOG(ERROR) << "Trx readv failed, invalid chunk header msg_len=" << (uint32_t)chunk_msg_len << " cur_index=" << (uint32_t)cur_index; errno = EBADMSG; diff --git a/src/brpc/ubshm/ub_ring.h b/src/brpc/ubshm/ub_ring.h index 0e47a04d4e..16dd59e21e 100644 --- a/src/brpc/ubshm/ub_ring.h +++ b/src/brpc/ubshm/ub_ring.h @@ -98,11 +98,11 @@ class UBRing : public butil::IReader { static inline RETURN_CODE CheckTrxConnectParam(const char *listener_name, const char *local_name) { - if (UNLIKELY(listener_name == nullptr)) { + if (BAIDU_UNLIKELY(listener_name == nullptr)) { LOG(ERROR) << "The request listener name is null."; return UBRING_ERR; } - if (UNLIKELY(local_name == nullptr)) { + if (BAIDU_UNLIKELY(local_name == nullptr)) { LOG(ERROR) << "The request trx shared memory name is null."; return UBRING_ERR; } @@ -119,7 +119,7 @@ class UBRing : public butil::IReader { static inline RETURN_CODE CheckTrxSendPreCheck(UbrTrx *trx) { - if (UNLIKELY(trx->ubr_tx.trx_state != UBR_STATE_CONNECTED)) { + if (BAIDU_UNLIKELY(trx->ubr_tx.trx_state != UBR_STATE_CONNECTED)) { LOG(ERROR) << "Trx send failed, trx is not connected state."; return UBRING_ERR; } @@ -128,25 +128,25 @@ class UBRing : public butil::IReader { } static RETURN_CODE CheckTrxRecvParam(UbrTrx *trx, const void *buf, uint32_t buf_len) { - if (UNLIKELY(trx == nullptr)) { + if (BAIDU_UNLIKELY(trx == nullptr)) { LOG(ERROR) << "Trx recv failed, trx is null."; return UBRING_ERR; } - if (UNLIKELY((UbrEventQMsg *)trx->ubr_rx.local_rx_event_q.addr == nullptr)) { + if (BAIDU_UNLIKELY((UbrEventQMsg *)trx->ubr_rx.local_rx_event_q.addr == nullptr)) { LOG(ERROR) << "Trx send failed, local_tx_event_q addr is NULL."; return UBRING_ERR; } - if (UNLIKELY(trx->ubr_rx.trx_state != UBR_STATE_CONNECTED)) { + if (BAIDU_UNLIKELY(trx->ubr_rx.trx_state != UBR_STATE_CONNECTED)) { LOG(ERROR) << "Trx recv failed, trx is not connected statep=" << trx->ubr_rx.trx_state; return UBR_NOT_CONNECTED; } - if (UNLIKELY(buf == nullptr)) { + if (BAIDU_UNLIKELY(buf == nullptr)) { LOG(ERROR) << "Trx recv failed, buf is null."; return UBRING_ERR; } - if (UNLIKELY(buf_len == 0)) { + if (BAIDU_UNLIKELY(buf_len == 0)) { LOG(ERROR) << "Trx recv failed, buf_len is 0."; return UBRING_ERR; } @@ -155,7 +155,7 @@ class UBRing : public butil::IReader { static inline RETURN_CODE CheckTrxRecvPreCheck(UbrTrx *trx) { - if (UNLIKELY(trx->ubr_rx.trx_state != UBR_STATE_CONNECTED)) { + if (BAIDU_UNLIKELY(trx->ubr_rx.trx_state != UBR_STATE_CONNECTED)) { LOG(ERROR) << "Trx recv failed, trx is not connected state."; return UBRING_ERR; } @@ -188,15 +188,15 @@ class UBRing : public butil::IReader { LOG(ERROR) << "Trx close callback failed, trx is null."; return UBRING_ERR; } - if (UNLIKELY(trx->local_shm.addr == nullptr)) { + if (BAIDU_UNLIKELY(trx->local_shm.addr == nullptr)) { LOG(ERROR) << "Trx close failed, local_shm addr is NULL."; return UBRING_ERR; } - if (UNLIKELY(trx->ubr_rx.local_rx_event_q.addr == nullptr)) { + if (BAIDU_UNLIKELY(trx->ubr_rx.local_rx_event_q.addr == nullptr)) { LOG(ERROR) << "Trx close failed, local_rx_event_q addr is NULL."; return UBRING_ERR; } - if (UNLIKELY(trx->ubr_tx.local_tx_event_q.addr == nullptr)) { + if (BAIDU_UNLIKELY(trx->ubr_tx.local_tx_event_q.addr == nullptr)) { LOG(ERROR) << "Trx close failed, local_tx_event_q addr is NULL."; return UBRING_ERR; } diff --git a/src/brpc/ubshm/ub_ring_manager.cpp b/src/brpc/ubshm/ub_ring_manager.cpp index 6abb485e8a..9084596bba 100644 --- a/src/brpc/ubshm/ub_ring_manager.cpp +++ b/src/brpc/ubshm/ub_ring_manager.cpp @@ -15,10 +15,12 @@ // specific language governing permissions and limitations // under the License. +#include #include #include "brpc/ubshm/ub_ring.h" #include "brpc/ubshm/ub_ring_manager.h" #include "butil/logging.h" +#include "butil/scoped_lock.h" namespace brpc { namespace ubring { @@ -41,11 +43,11 @@ uint64_t g_ub_event_cnt = 0; uint64_t g_ubr_listener_num = 0; RETURN_CODE UBRingManager::GetUbrDealMsgMaxCnt(const uint32_t capacity, uint32_t *deal_msg_max_cnt) { - if (UNLIKELY(deal_msg_max_cnt == nullptr)) { + if (BAIDU_UNLIKELY(deal_msg_max_cnt == nullptr)) { LOG(ERROR) << "Get update factor failed, deal_msg_max_cnt is null."; return UBRING_ERR; } - if (UNLIKELY(FLAGS_tail_update_after_read == 0)) { + if (BAIDU_UNLIKELY(FLAGS_tail_update_after_read == 0)) { LOG(ERROR) << "Get update factor failed, factor is 0."; return UBRING_ERR; } @@ -66,7 +68,7 @@ RETURN_CODE UBRingManager::UbrMgrDefault() RETURN_CODE UBRingManager::UbrMgrInit() { RETURN_CODE rc = UbrMgrDefault(); - if (UNLIKELY(rc != UBRING_OK)) { + if (BAIDU_UNLIKELY(rc != UBRING_OK)) { LOG(ERROR) << "Ubr manager set default values failed."; return rc; } @@ -79,7 +81,7 @@ RETURN_CODE UBRingManager::UbrMgrInit() { g_ubr_mgr.trx_mgr_unit_id = (uint64_t *)malloc(trx_mgr_id_size); size_t trx_mgr_ctl_size = g_ubr_mgr.trx_cap * sizeof(UbrCleanupCtl *); g_ubr_mgr.trx_mgr_unit_ctl = (UbrCleanupCtl **)malloc(trx_mgr_ctl_size); - if (UNLIKELY(g_ubr_mgr.trx_mgr == 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)) { @@ -105,7 +107,7 @@ void UBRingManager::UbrMgrFini() { while (busy) { busy = false; { - LOCK_GUARD(g_ubr_trx_mgr_mtx); + BAIDU_SCOPED_LOCK(g_ubr_trx_mgr_mtx); if (g_ubr_mgr.trx_mgr_unit_ctl != nullptr) { for (uint32_t i = 0; i < g_ubr_mgr.trx_cap; ++i) { UbrCleanupCtl* ctl = g_ubr_mgr.trx_mgr_unit_ctl[i]; @@ -127,7 +129,7 @@ void UBRingManager::UbrMgrFini() { } } { - LOCK_GUARD(g_ubr_trx_mgr_mtx); + BAIDU_SCOPED_LOCK(g_ubr_trx_mgr_mtx); if (g_ubr_mgr.trx_mgr_unit_ctl != nullptr) { for (uint32_t i = 0; i < g_ubr_mgr.trx_cap; ++i) { UbrCleanupCtl* ctl = g_ubr_mgr.trx_mgr_unit_ctl[i]; @@ -143,7 +145,7 @@ void UBRingManager::UbrMgrFini() { FREE_PTR(g_ubr_mgr.trx_mgr_unit_ctl); } { - LOCK_GUARD(g_ubr_listener_mgr_mtx); + BAIDU_SCOPED_LOCK(g_ubr_listener_mgr_mtx); } g_ubr_mgr.trx_num = 0; g_ubr_mgr.trx_cap = 0; @@ -151,17 +153,17 @@ void UBRingManager::UbrMgrFini() { } RETURN_CODE UBRingManager::AcquireUbrTrxFromMgr(UbrTrx **trx) { - if (UNLIKELY(trx == nullptr)) { + if (BAIDU_UNLIKELY(trx == nullptr)) { LOG(ERROR) << "Acquire trx failed, trx is null."; return UBRING_ERR; } - if (UNLIKELY(g_ubr_mgr.trx_mgr == nullptr)) { + if (BAIDU_UNLIKELY(g_ubr_mgr.trx_mgr == nullptr)) { LOG(ERROR) << "Acquire trx failed, trx_mgr is null."; return UBRING_ERR; } - LOCK_GUARD(g_ubr_trx_mgr_mtx); + BAIDU_SCOPED_LOCK(g_ubr_trx_mgr_mtx); if (g_ubr_mgr.trx_num >= g_ubr_mgr.trx_cap) { LOG(ERROR) << "Acquire trx failed, trx number is full."; return UBRING_ERR; @@ -169,9 +171,12 @@ RETURN_CODE UBRingManager::AcquireUbrTrxFromMgr(UbrTrx **trx) { for (uint32_t i = 0; i < g_ubr_mgr.trx_cap; ++i) { if (g_ubr_mgr.trx_mgr_unit_status[i] == UBR_MGR_UNIT_FREE) { - memset(&g_ubr_mgr.trx_mgr[i], 0, sizeof(UbrTrx)); - // The explicit re-initialization after memset is deliberate: it - // documents the per-acquisition invariants of these fields. + // Value-initialize the slot so its butil::atomic members are + // properly constructed while every other field is zeroed, which + // is what the previous memset did. + new (&g_ubr_mgr.trx_mgr[i]) UbrTrx(); + // The explicit re-initialization is deliberate: it documents the + // per-acquisition invariants of these fields. 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; @@ -202,23 +207,23 @@ RETURN_CODE UBRingManager::AcquireUbrTrxFromMgr(UbrTrx **trx) { RETURN_CODE UBRingManager::ReleaseUbrTrxFromMgr(UbrTrx *trx, uint64_t expect_ubr_id) { - if (UNLIKELY(trx == nullptr)) { + if (BAIDU_UNLIKELY(trx == nullptr)) { LOG(ERROR) << "Release trx failed, trx is null."; return UBRING_ERR; } - if (UNLIKELY(g_ubr_mgr.trx_mgr == nullptr)) { + if (BAIDU_UNLIKELY(g_ubr_mgr.trx_mgr == nullptr)) { LOG(ERROR) << "Release trx failed, trx_mgr is null."; return UBRING_ERR; } - LOCK_GUARD(g_ubr_trx_mgr_mtx); + BAIDU_SCOPED_LOCK(g_ubr_trx_mgr_mtx); uint32_t idx = trx->trx_mgr_index; if (g_ubr_mgr.trx_mgr_unit_status[idx] == UBR_MGR_UNIT_FREE) { LOG(INFO) << "Release trx already freed, name=" << trx->local_shm.name; return UBRING_OK; } - if (UNLIKELY(g_ubr_mgr.trx_mgr_unit_id[idx] != expect_ubr_id)) { + if (BAIDU_UNLIKELY(g_ubr_mgr.trx_mgr_unit_id[idx] != expect_ubr_id)) { // The slot was released and acquired again meanwhile; the stale // caller must not touch the new occupant. LOG(WARNING) << "Release stale trx refused, name=" << trx->local_shm.name; @@ -242,8 +247,8 @@ RETURN_CODE UBRingManager::ReleaseUbrTrxFromMgr(UbrTrx *trx, } UbrCleanupCtl* UBRingManager::SnapshotUnitCleanupCtl(uint32_t idx) { - LOCK_GUARD(g_ubr_trx_mgr_mtx); - if (UNLIKELY(g_ubr_mgr.trx_mgr_unit_ctl == nullptr || idx >= g_ubr_mgr.trx_cap)) { + 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; } UbrCleanupCtl* ctl = g_ubr_mgr.trx_mgr_unit_ctl[idx]; @@ -254,8 +259,8 @@ UbrCleanupCtl* UBRingManager::SnapshotUnitCleanupCtl(uint32_t idx) { } bool UBRingManager::IsUbrTrxSlotUsed(uint32_t idx, uint64_t expect_ubr_id) { - LOCK_GUARD(g_ubr_trx_mgr_mtx); - if (UNLIKELY(g_ubr_mgr.trx_mgr_unit_id == nullptr || + BAIDU_SCOPED_LOCK(g_ubr_trx_mgr_mtx); + if (BAIDU_UNLIKELY(g_ubr_mgr.trx_mgr_unit_id == nullptr || g_ubr_mgr.trx_mgr_unit_status == nullptr || idx >= g_ubr_mgr.trx_cap)) { return false; @@ -267,8 +272,8 @@ bool UBRingManager::IsUbrTrxSlotUsed(uint32_t idx, uint64_t expect_ubr_id) { bool UBRingManager::TryPublishUnitCleanupCtl(uint32_t idx, uint64_t expect_ubr_id, UbrCleanupCtl *ctl) { - LOCK_GUARD(g_ubr_trx_mgr_mtx); - if (UNLIKELY(g_ubr_mgr.trx_mgr_unit_ctl == nullptr || + BAIDU_SCOPED_LOCK(g_ubr_trx_mgr_mtx); + if (BAIDU_UNLIKELY(g_ubr_mgr.trx_mgr_unit_ctl == nullptr || g_ubr_mgr.trx_mgr_unit_status == nullptr || g_ubr_mgr.trx_mgr_unit_id == nullptr || idx >= g_ubr_mgr.trx_cap || @@ -283,8 +288,8 @@ bool UBRingManager::TryPublishUnitCleanupCtl(uint32_t idx, } bool UBRingManager::DetachUnitCleanupCtl(uint32_t idx, UbrCleanupCtl *ctl) { - LOCK_GUARD(g_ubr_trx_mgr_mtx); - if (UNLIKELY(g_ubr_mgr.trx_mgr_unit_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 || g_ubr_mgr.trx_mgr_unit_ctl[idx] != ctl)) { return false; @@ -320,7 +325,7 @@ void UBRingManager::LinkInfoFini(void) { return; } { - LOCK_GUARD(g_link_info_mgr_mtx); + BAIDU_SCOPED_LOCK(g_link_info_mgr_mtx); FREE_PTR(g_link_info_mgr.all_link_info); FREE_PTR(g_link_info_mgr.link_mgr_unit_status); } @@ -366,11 +371,11 @@ void UBRingManager::ReleaseLinkInfoFromMgr(UbrTrx *trx) { int32_t UBRingManager::UbEventCallback(const char *shm_name) { - if (UNLIKELY(shm_name == nullptr)) { + if (BAIDU_UNLIKELY(shm_name == nullptr)) { LOG(ERROR) << "Ub event callback failed, shm name is null."; return UBRING_ERR; } - if (UNLIKELY(g_ubr_mgr.trx_mgr == nullptr)) { + if (BAIDU_UNLIKELY(g_ubr_mgr.trx_mgr == nullptr)) { LOG(ERROR) << "Ub event callback failed, trx mgr is null."; return UBRING_ERR; } diff --git a/src/brpc/ubshm/ubr_trx.h b/src/brpc/ubshm/ubr_trx.h index 6a947d2d78..cc35f7223f 100644 --- a/src/brpc/ubshm/ubr_trx.h +++ b/src/brpc/ubshm/ubr_trx.h @@ -144,7 +144,7 @@ struct UbrCleanupCtl { TagUbrTrx* trx; // immutable uint64_t ubr_id; // immutable generation AtomicInt state; // UbrCleanupState - UbrTimerId timer; // delayed clear timer + butil::atomic timer; // delayed clear timer // References: timer/callback (held from a successful schedule until // the callback fully returned, or released by whoever cancels the // timer before it fires) + starter (until the schedule call @@ -167,9 +167,9 @@ typedef struct TagUbrTrx { UbrTrxType type; SHM local_shm; SHM remote_shm; - UbrTimerId close_timer; - UbrTimerId hb_timer; - UbrCleanupCtl* cleanup_ctl; + butil::atomic close_timer; + butil::atomic hb_timer; + butil::atomic cleanup_ctl; // 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;