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

@@ -116,6 +116,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/shm/shm_ubs.cpp

@@ -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;
+UbrTimerId g_shm_timer_id = nullptr;
 ShmList *g_shm_list = nullptr;
 static RETURN_CODE UbsShmInterfacesLoad(void);
 char hostname[MAX_HOST_NAME_DESC_LENGTH];
@@ -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 (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;
     }
 
@@ -418,50 +419,56 @@ void *UbsShmCallback(void* args)
         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;
+    {
+        LOCK_GUARD(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);
+        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
+    }
+    LOG(INFO) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << " success.";
 
-        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;
-        }
+    {
+        LOCK_GUARD(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;
+    }
+    LOG(INFO) << "Ubs free local shm=" << shm.name << " length=" << shm.len << " success.";
     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 (UNLIKELY(rc != UBRING_OK)) {
         LOG(ERROR) << "Start shm timer failed.";
         return UBRING_ERR;
     }
-    g_shm_timer_fd = timer_fd;
 
     return UBRING_OK;
 }
@@ -493,7 +500,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;