Ubshm transport fix 2

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` 一定能等到它；销毁回调参数所属资源的外部路径用 `UbrTimerDelAndWait` 等待运行中回调退出；一次性定时器触发后自动清理句柄。进入关闭处理时会带上调用方看到的代际，并在 manager 锁内原子认领槽位，因此按旧快照发起的关闭（例如共享内存故障事件回调）不会误伤释放后被复用的新连接。定时器回调运行在进程全局的 bthread 定时线程上，必须快速返回。阻塞的共享内存清理不在这里进行：延迟清理与待回收映射的回调只仲裁清理所有权，并把任务投递给专用的清理 worker bthread（`ub_cleanup_worker.cpp`），由它把 UBS SDK 的 `unmap`/`deallocate` 调用移出定时线程。对一次性定时器，`UbrTimerDel` 返回 0 还额外保证 facade 已不再访问句柄槽位所在的存储（它会等待被定时线程派发但尚未竞争槽位的 wrapper 退出竞争），因此调用方可以释放承载该槽位的对象；延迟清理路径正是据此释放 `UbrCleanupCtl`。关闭检查定时器在链路空闲时按指数退避轮询（上限 `ub_event_queue_timer_interval_max_us`，默认 10ms），有流量或正在关闭时恢复快速轮询。
 
 ## 参考资料
 

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; a periodic callback that deletes its own timer keeps its handle anchored until the callback returns, so a concurrent `UbrTimerDelAndWait` from a teardown path always waits for it. External teardown that frees resources reachable from the callback argument uses `UbrTimerDelAndWait` to wait out a possibly running callback; one-shot timers clean up their handle automatically when they fire. Entering a close carries the generation the caller saw and claims the pool slot atomically under the manager lock, so a close started from a stale snapshot (the shared-memory fault event, for instance) cannot hit a connection that reused the slot in the meantime. Timer callbacks run on the process-wide bthread timer thread and must return quickly. Blocking shared-memory cleanup is therefore not done there: the delayed-clear and pending-unmap callbacks only settle the cleanup ownership and post a job to a dedicated cleanup worker bthread (`ub_cleanup_worker.cpp`), which runs the UBS SDK `unmap`/`deallocate` calls off the timer thread. For a one-shot timer, a `UbrTimerDel` that returns 0 additionally guarantees that the facade is done with the storage the handle slot lives in -- it waits out a wrapper the timer thread already dispatched -- so the caller may free the object that holds the slot; the delayed-clear path frees `UbrCleanupCtl` that way. The close-check timer backs off exponentially while the link is idle (capped by `ub_event_queue_timer_interval_max_us`, 10ms by default) and returns to fast polling once there is traffic or a close in progress.
 
 ## References
 

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