Replace custom timer with bthread timer and improve lifecycle safety

docs/cn/ubring.md

@@ -187,7 +187,7 @@ UBRing 架构包含以下组件：
 
 ### 定时器管理
 
-UBRing 使用高精度定时器系统 (`timer_mgr.cpp`) 进行连接管理和超时处理，支持 epoll（Linux）和 kqueue（macOS）。
+UBRing 的连接管理和超时处理基于 bthread 定时器（`bthread_timer_add`/`bthread_timer_del`，由 `timer_mgr.cpp` 封装）。非阻塞的 `UbrTimerDel` 可在回调内自删；销毁回调参数所属资源的外部路径用 `UbrTimerDelAndWait` 等待运行中回调退出；一次性定时器触发后自动清理句柄。定时器回调运行在进程全局的 bthread 定时线程上，必须快速返回。关闭检查定时器在链路空闲时按指数退避轮询（上限 `ub_event_queue_timer_interval_max_us`，默认 10ms），有流量或正在关闭时恢复快速轮询。
 
 ## 参考资料
 

docs/en/ubring.md

@@ -186,7 +186,7 @@ The shared memory manager (`shm_mgr.cpp`) provides a unified interface for diffe
 
 ### Timer Management
 
-UBRing uses a high-precision timer system (`timer_mgr.cpp`) for connection management and timeout handling, supporting both epoll (Linux) and kqueue (macOS).
+UBRing connection management and timeout handling are built on bthread timers (`bthread_timer_add`/`bthread_timer_del`, wrapped in `timer_mgr.cpp`). The non-blocking `UbrTimerDel` is safe to call from within a timer callback; external teardown that frees resources reachable from the callback argument uses `UbrTimerDelAndWait` to wait out a possibly running callback; one-shot timers clean up their handle automatically when they fire. Timer callbacks run on the process-wide bthread timer thread and must return quickly. The close-check timer backs off exponentially while the link is idle (capped by `ub_event_queue_timer_interval_max_us`, 10ms by default) and returns to fast polling once there is traffic or a close in progress.
 
 ## References
 

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 <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()
@@ -116,6 +113,7 @@ static inline int Copy64Byte(int8_t *dst, int8_t *src) {
 
 #define SEC_TO_NSEC 1000000000
 #define MSEC_TO_NSEC 1000000
+#define SEC_TO_USEC 1000000
 #define USEC_TO_NSEC 1000
 #define MSEC_TO_SEC 1000
 #define MAX_IP_PORT_STR_LEN 23

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.";

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;
     }

src/brpc/ubshm/shm/shm_ubs.cpp

@@ -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"
@@ -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};
-int g_shm_timer_fd = 0;
+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];
@@ -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;
     }
@@ -377,14 +377,15 @@ RETURN_CODE UbsShmInit(void)
 
 RETURN_CODE UbsShmFini(void)
 {
-    int ret = ubsmem_finalize();
-    if (ret != UBSM_OK) {
-        LOG(ERROR) << "Ubs shm finalize fail, ret=" << ret;
+    // Stop the cleanup timer before finalizing the SDK it calls into.
+    if (BAIDU_UNLIKELY(DestroyShmTimer(g_shm_list) != UBRING_OK)) {
+        LOG(ERROR) << "Ubs shm list finalize failed.";
         return UBRING_ERR;
     }
 
-    if (UNLIKELY(DestroyShmTimer(g_shm_list) != UBRING_OK)) {
-        LOG(ERROR) << "Ubs shm list finalize failed.";
+    int ret = ubsmem_finalize();
+    if (ret != UBSM_OK) {
+        LOG(ERROR) << "Ubs shm finalize fail, ret=" << ret;
         return UBRING_ERR;
     }
 
@@ -413,55 +414,58 @@ 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;
     }
 
-    LOCK_GUARD(shm_list->shm_lock);
-    while (shm_list->head != nullptr) {
-        SHM shm = shm_list->head->shm;
-        if (shm.addr == nullptr) {
-            LOG(ERROR) << "Ubs input shm param is invalid, addr is NULL.";
+    // 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.
+    SHM shm;
+    {
+        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.";
+        BAIDU_SCOPED_LOCK(shm_list->shm_lock);
+        DeleteShmToList(shm_list);
+        return nullptr;
+    }
 
-        int ret = ubsmem_shmem_unmap(shm.addr, shm.len);
-        if (ret != UBSM_OK) {
-            if (ret == UBSM_ERR_NET) {
-                return nullptr;
-            }
-            LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " failed, ret=" << ret;
-            return nullptr;
+    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
         }
-        LOG(INFO) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " success.";
+        LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " failed, ret=" << ret;
+        return nullptr;                  // node stays at head, retried
+    }
 
-        ret = ubsmem_shmem_deallocate(shm.name);
-        if (ret != UBSM_OK) {
-            DeleteShmToList(shm_list);
-            LOG(ERROR) << "Ubs delete shm=" << shm.name << " failed, ret=" << ret;
-            return nullptr;
-        }
+    {
+        BAIDU_SCOPED_LOCK(shm_list->shm_lock);
         DeleteShmToList(shm_list);
-        LOG(INFO) << "Ubs free local shm=" << shm.name << " length=" << shm.len << " success.";
     }
 
+    ret = ubsmem_shmem_deallocate(shm.name);
+    if (ret != UBSM_OK) {
+        LOG(ERROR) << "Ubs delete shm=" << shm.name << " failed, ret=" << ret;
+    }
     return nullptr;
 }
 
 RETURN_CODE UbsShmAddTimer(ShmList *shm_list)
 {
-    const uint32_t timer_interval_s = FLAGS_ub_flying_io_timeout_s;
-    itimerspec time_spec = {
-        .it_interval = {.tv_sec = timer_interval_s, .tv_nsec = 0},
-        .it_value = {.tv_sec = 0, .tv_nsec = 1}
-    };
-    int timer_fd = TimerStart(&time_spec, UbsShmCallback, (void*)shm_list);
-    if (UNLIKELY(timer_fd == -1)) {
+    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 (BAIDU_UNLIKELY(rc != UBRING_OK)) {
         LOG(ERROR) << "Start shm timer failed.";
         return UBRING_ERR;
     }
-    g_shm_timer_fd = timer_fd;
 
     return UBRING_OK;
 }
@@ -493,7 +497,8 @@ RETURN_CODE InitShmTimer(ShmList **shm_list)
 
 RETURN_CODE DestroyShmTimer(ShmList *shm_list)
 {
-    DeleteTimerSafe((uint32_t)g_shm_timer_fd);
+    // Wait out a possibly running UbsShmCallback before tearing shm_list down.
+    UbrTimerDelAndWait(&g_shm_timer_id);
     if (shm_list == nullptr) {
         LOG(WARNING) << "Shm list is null.";
         return UBRING_ERR;
@@ -513,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) {
@@ -548,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) {