Skip to content

style(ubshm): use BAIDU macros, BAIDU_SCOPED_LOCK and butil::atomic #57

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zchuango merged 1 commit into from
Sep 24, 2026
Merged

style(ubshm): use BAIDU macros, BAIDU_SCOPED_LOCK and butil::atomic #57

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 5 additions & 8 deletions src/brpc/ubshm/common/common.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -42,11 +39,11 @@
#endif

#ifdef __cplusplus
#include <atomic>
using AtomicInt = std::atomic<int>;
using AtomicBool = std::atomic<bool>;
using AtomicUintFast64 = std::atomic<uint_fast64_t>;
using AtomicUintFast8 = std::atomic<uint_fast8_t>;
#include "butil/atomicops.h"
using AtomicInt = butil::atomic<int>;
using AtomicBool = butil::atomic<bool>;
using AtomicUintFast64 = butil::atomic<uint_fast64_t>;
using AtomicUintFast8 = butil::atomic<uint_fast8_t>;
#define ATOMIC_INIT(var, value) var.store(value)
#define ATOMIC_STORE(var, value) var.store(value)
#define ATOMIC_LOAD(var) var.load()
Expand Down
2 changes: 1 addition & 1 deletion src/brpc/ubshm/common/thread_lock.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.";
Expand Down
20 changes: 10 additions & 10 deletions src/brpc/ubshm/shm/shm_mgr.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down Expand Up @@ -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;
}
Expand All @@ -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;
Expand All @@ -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;
}
Expand All @@ -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;
}
Expand All @@ -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;
}
Expand All @@ -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;
}
Expand All @@ -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;
}
Expand All @@ -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;
}
Expand Down
29 changes: 13 additions & 16 deletions src/brpc/ubshm/shm/shm_ubs.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,8 @@
#include <dlfcn.h>
#include <time.h>
#include <gflags/gflags.h>
#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"
Expand All @@ -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<UbrTimerId> g_shm_timer_id(nullptr);
ShmList *g_shm_list = nullptr;
static RETURN_CODE UbsShmInterfacesLoad(void);
char hostname[MAX_HOST_NAME_DESC_LENGTH];
Expand Down Expand Up @@ -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;
}
Expand All @@ -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;
}
Expand Down Expand Up @@ -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;
}
Expand All @@ -423,15 +423,15 @@ 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;
}
shm = shm_list->head->shm;
}
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;
}
Expand All @@ -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;
}

Expand All @@ -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;
}
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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) {
Expand Down
46 changes: 20 additions & 26 deletions src/brpc/ubshm/timer/timer_mgr.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,10 @@
// specific language governing permissions and limitations
// under the License.

#include <atomic>
#include <new>
#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"

Expand Down Expand Up @@ -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<bthread_timer_t> id;
butil::atomic<UbrTimerId>* slot;
butil::atomic<bthread_timer_t> id;
void* (*cb)(void*);
void* arg;
UbrTimerBackoffFn backoff;
uint64_t interval_us; // timer thread only
bool periodic;
std::atomic<int> state; // kStarting/kScheduled/kDead
std::atomic<bool> stopped;
std::atomic<int> ref;
std::atomic<bool> join_pending; // a DelAndWait is waiting
std::atomic<bool> done; // refs hit zero, joiner frees
butil::atomic<int> state; // kStarting/kScheduled/kDead
butil::atomic<bool> stopped;
butil::atomic<int> ref;
butil::atomic<bool> join_pending; // a DelAndWait is waiting
butil::atomic<bool> done; // refs hit zero, joiner frees
};

namespace {
Expand Down Expand Up @@ -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);
}
Expand All @@ -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<UbrTimerId>* slot) {
return slot->exchange(nullptr);
}

RETURN_CODE TimerStartInternal(UbrTimerId* slot, uint64_t delay_us,
RETURN_CODE TimerStartInternal(butil::atomic<UbrTimerId>* 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;
}
Expand All @@ -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
Expand All @@ -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<UbrTimerId>* 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<UbrTimerId>* slot) {
if (slot == nullptr) {
return 1;
}
Expand Down Expand Up @@ -242,7 +236,7 @@ int UbrTimerDel(UbrTimerId* slot) {
// callback may still complete.
}

void UbrTimerDelAndWait(UbrTimerId* slot) {
void UbrTimerDelAndWait(butil::atomic<UbrTimerId>* slot) {
if (slot == nullptr) {
return;
}
Expand Down
Loading
Loading