From 060ba38823d89ddba18461bffad166708204b5bb Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Mon, 24 Aug 2026 21:54:37 +0800 Subject: [PATCH 01/15] =?UTF-8?q?[fix]1=E3=80=81=E5=B0=86=E5=8E=9F?= =?UTF-8?q?=E6=9C=89=E7=9A=84=E8=AE=A1=E6=97=B6=E5=99=A8=E6=9B=BF=E6=8D=A2?= =?UTF-8?q?=E4=B8=BAbrpc=E7=9A=84bthread=EF=BC=8C=E4=BB=A5=E6=AD=A4?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=88=A0=E9=99=A4=E8=AE=A1=E6=97=B6?= =?UTF-8?q?=E5=99=A8=E5=8F=AF=E8=83=BD=E4=BC=9A=E6=97=A0=E9=99=90=E6=9C=9F?= =?UTF-8?q?=E9=98=BB=E5=A1=9E=E3=80=81fd=E9=87=8D=E7=94=A8=E7=AB=9E?= =?UTF-8?q?=E9=80=9F=E3=80=81=E8=AE=A1=E6=97=B6=E5=99=A8=E5=88=9D=E5=A7=8B?= =?UTF-8?q?=E5=8C=96=E5=8C=85=E5=90=AB=E4=B8=80=E4=B8=AA=E7=AB=9E=E9=80=89?= =?UTF-8?q?=E3=80=81=E9=BB=98=E8=AE=A4=E7=9A=84=E9=97=AD=E9=97=A8=E5=AE=9A?= =?UTF-8?q?=E6=97=B6=E5=99=A8=E8=BF=87=E4=BA=8E=E9=A2=91=E7=B9=81=E3=80=81?= =?UTF-8?q?=E6=89=80=E6=9C=89=E5=AE=9A=E6=97=B6=E5=99=A8=E5=9B=9E=E8=B0=83?= =?UTF-8?q?=E9=83=BD=E5=9C=A8=E5=90=8C=E4=B8=80P=E7=BA=BF=E7=A8=8B?= =?UTF-8?q?=E4=B8=8A=E4=B8=B2=E8=A1=8C=E3=80=81macOS=E6=A8=A1=E6=8B=9F?= =?UTF-8?q?=E5=9C=A8=E5=8A=9F=E8=83=BD=E4=B8=8A=E5=B9=B6=E4=B8=8D=E7=AD=89?= =?UTF-8?q?=E6=95=88=E3=80=822=E3=80=81=E7=A7=BB=E9=99=A4=E5=9B=BA?= =?UTF-8?q?=E5=AE=9A=E6=95=B0=E7=BB=84=EF=BC=8C=E6=94=B9=E7=94=A8unordered?= =?UTF-8?q?=5Fmap=EF=BC=9B3=E3=80=81=E4=B8=BAitimerspec=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E5=BD=92=E4=B8=80=E5=8C=96=EF=BC=9B4=E3=80=81=E8=A7=A3?= =?UTF-8?q?=E5=86=B3CqSidOp=E7=9A=84sid=E7=9B=B8=E5=90=8C=E8=80=8Cevent?= =?UTF-8?q?=E4=B8=8D=E5=90=8C=E6=97=B6=E6=B7=BB=E5=8A=A0=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.cpp | 488 ++++++----------------------- src/brpc/ubshm/timer/timer_mgr.h | 62 ++-- src/brpc/ubshm/ub_endpoint.cpp | 1 + 3 files changed, 125 insertions(+), 426 deletions(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index b5e0c9ef3b..e60bb02cd4 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -14,455 +14,167 @@ // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. - #define _GNU_SOURCE + #include -#include -#include -#include -#include -#include +#include +#include +#include #include -#include +#include #include "brpc/ubshm/timer/timer_mgr.h" namespace brpc { namespace ubring { -int32_t g_epoll_fd = -1; -std::atomic g_total_timer_num(0); -TimerFdCtx *g_timer_fd_ctx_map = nullptr; -uint32_t g_max_system_fd = 0; -static pthread_t g_epoll_execute_thread = 0; -static int32_t g_timer_module_initialized = 0; - -#if defined(OS_MACOSX) -static int timerfd_create_macosx(int clockid, int flags); -static int timerfd_settime_macosx(int fd, int flags, - const itimerspec *new_value, - itimerspec *old_value); -#endif - -static RETURN_CODE DeleteTimerInner(uint32_t fd) { - if (g_timer_fd_ctx_map == nullptr) { - return UBRING_OK; - } - - if (pthread_spin_lock(&g_timer_fd_ctx_map[fd].spin_lock) != 0) { - return UBRING_ERR; - } - - if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) { - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - return UBRING_OK; - } - - g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; - g_timer_fd_ctx_map[fd].cb = nullptr; - g_timer_fd_ctx_map[fd].args = nullptr; - g_timer_fd_ctx_map[fd].periodical = 0; - g_timer_fd_ctx_map[fd].fd = 0; - - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - -#if defined(OS_LINUX) - epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, (int)fd, nullptr); -#elif defined(OS_MACOSX) - struct kevent evt; - EV_SET(&evt, fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr); - kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr); -#endif - - uint64_t exp = 0; - read((int)fd, &exp, sizeof(exp)); - - close((int)fd); - std::atomic_fetch_sub(&g_total_timer_num, 1U); - return UBRING_OK; -} - -static RETURN_CODE StartTimeEpoll(void) { -#if defined(OS_LINUX) - g_epoll_fd = epoll_create1(0); -#elif defined(OS_MACOSX) - g_epoll_fd = kqueue(); -#endif - if (UNLIKELY(g_epoll_fd == -1)) { - LOG(ERROR) << "Failed to create epoll/kqueue. errno=" << errno; - return UBRING_ERR; - } +std::unordered_map g_timer_ctx_map; +std::mutex g_timer_ctx_mutex; +std::atomic g_total_timer_num; - int ret = pthread_create(&g_epoll_execute_thread, nullptr, TimerEpoll, nullptr); - if (UNLIKELY(ret != 0)) { - LOG(ERROR) << "Failed to create thread err=" << ret; - return UBRING_ERR; - } - return UBRING_OK; -} +static std::atomic g_timer_id_counter(0); -static RETURN_CODE TimerSpinLocksInit(void) { - if (g_timer_fd_ctx_map == nullptr) { - LOG(ERROR) << "Timer module is not fully initialized."; - return UBRING_ERR; +static void normalize_timespec(itimerspec *spec) { + if (spec->it_interval.tv_nsec >= 1000000000L) { + spec->it_interval.tv_sec += spec->it_interval.tv_nsec / 1000000000L; + spec->it_interval.tv_nsec %= 1000000000L; } - - for (uint32_t fd = 0; fd < g_max_system_fd; fd++) { - int ret = pthread_spin_init(&g_timer_fd_ctx_map[fd].spin_lock, - PTHREAD_PROCESS_PRIVATE); - if (ret != EOK) { - LOG(ERROR) << "Failed to initialize spin lock for fd=" << fd; - for (uint32_t cleanup_fd = 0; cleanup_fd < fd; cleanup_fd++) { - pthread_spin_destroy(&g_timer_fd_ctx_map[cleanup_fd].spin_lock); - } - return UBRING_ERR; - } + if (spec->it_value.tv_nsec >= 1000000000L) { + spec->it_value.tv_sec += spec->it_value.tv_nsec / 1000000000L; + spec->it_value.tv_nsec %= 1000000000L; } - return UBRING_OK; -} - -static RETURN_CODE ExecuteCallback(int32_t timer_fd) { - UnifiedCallback((void *)(&g_timer_fd_ctx_map[timer_fd])); - return UBRING_OK; } -static RETURN_CODE TimerCtxMapCompletion(void) { - memset(g_timer_fd_ctx_map, 0, sizeof(TimerFdCtx) * g_max_system_fd); - RETURN_CODE ret = TimerSpinLocksInit(); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed to init spin locks for timer module."; - return UBRING_ERR; - } - return UBRING_OK; +int TimerInit() { + return 0; } -RETURN_CODE TimerInit(void) { - if (g_timer_module_initialized > 0) { - return UBRING_OK; +void TimerModuleDestroy() { + std::lock_guard lock(g_timer_ctx_mutex); + for (auto & pair: g_timer_ctx_map) { + bthread_timer_del(pair.second.timer_id); + pthread_spin_destroy(&pair.second.spin_lock); } - + g_timer_ctx_map.clear(); g_total_timer_num.store(0); +} - struct rlimit rlim; - if (getrlimit(RLIMIT_NOFILE, &rlim) != UBRING_OK) { - LOG(ERROR) << "Failed to get fd"; - return UBRING_ERR; +int32_t TimerStart(const itimerspec *time, TimerCallback cb, void *args) { + if (cb==nullptr) { + // LOG(ERROR) << "Timer callback is nullptr"; + return -1; } - g_max_system_fd = (uint32_t)rlim.rlim_cur; - if (g_timer_fd_ctx_map == nullptr) { - g_timer_fd_ctx_map = (TimerFdCtx *)malloc(sizeof(TimerFdCtx) * g_max_system_fd); - if (UNLIKELY(!g_timer_fd_ctx_map)) { - LOG(ERROR) << "Fail to malloc space for timer modules. errno=%d", errno; - return UBRING_ERR; - } + TimerContext ctx{}; + ctx.cb = cb; + ctx.args = args; + ctx.periodical = (time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0) ? 1 : 0; + ctx.interval = time->it_interval; + pthread_spin_init(&ctx.spin_lock, PTHREAD_PROCESS_PRIVATE); - RETURN_CODE ret = TimerCtxMapCompletion(); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed to init main data structure of Time Module. ret=" << ret; - free(g_timer_fd_ctx_map); - g_timer_fd_ctx_map = nullptr; - return UBRING_ERR; - } - } + uint64_t timer_id = g_timer_id_counter.fetch_add(1); - RETURN_CODE ret = StartTimeEpoll(); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed to start Timer Epoll. ret=" << ret; - if (LIKELY(g_timer_fd_ctx_map != nullptr)) { - FREE_PTR(g_timer_fd_ctx_map); - } - return UBRING_ERR; - } - g_timer_module_initialized = 1; - return UBRING_OK; -} + itimerspec normalized_time = *time; + normalize_timespec(&normalized_time); + timespec abstime = normalized_time.it_value; -void *UnifiedCallback(void *args) { - TimerFdCtx *ctx = (TimerFdCtx *)args; - if (pthread_spin_lock(&ctx->spin_lock) != 0) { - return nullptr; + int ret = bthread_timer_add(&ctx.timer_id, abstime, TimerCallbackWrapper, reinterpret_cast(timer_id)); + if (ret != 0) { + // LOG(ERROR) << "Failed to add bthread timer, ret=" << ret; + pthread_spin_destroy(&ctx.spin_lock); + return -1; } - if (ctx->status == TIMER_CONTEXT_NOT_USING) { - pthread_spin_unlock(&ctx->spin_lock); - return nullptr; + { + std::lock_guard lock(g_timer_ctx_mutex); + g_timer_ctx_map[timer_id] = ctx; } - void *(*cb)(void *) = ctx->cb; - void *cb_args = ctx->args; - uint32_t fd = ctx->fd; - int is_periodical = ctx->periodical; - ctx->status = TIMER_CONTEXT_CALLBACK_ONGOING; - - pthread_spin_unlock(&ctx->spin_lock); - - cb(cb_args); - - if (!is_periodical) { - DeleteTimerInner(fd); - } - return nullptr; + std::atomic_fetch_add(&g_total_timer_num, 1U); + return static_cast(timer_id); } -void *TimerEpoll(void *args) { - UNREFERENCE_PARAM(args); -#if defined(OS_LINUX) - struct epoll_event ready_events[MAX_TIMER]; -#elif defined(OS_MACOSX) - struct kevent ready_events[MAX_TIMER]; -#endif - - while (1) { - if (g_timer_module_initialized <= 0) { - LOG(ERROR) << "The Timer module is not initialized."; - break; - } - -#if defined(OS_LINUX) - int32_t ready_num = epoll_wait(g_epoll_fd, ready_events, MAX_TIMER, - TIMER_EPOLL_WAIT_TIMEOUT); -#elif defined(OS_MACOSX) - struct timespec timeout = {0, TIMER_EPOLL_WAIT_TIMEOUT * 1000000}; - int32_t ready_num = kevent(g_epoll_fd, nullptr, 0, ready_events, MAX_TIMER, &timeout); -#endif - - if (UNLIKELY(ready_num == -1)) { - errno_t err = errno; - if (err == EINTR) { - LOG_EVERY_SECOND(WARNING) << "Epoll/Kqueue wait was interrupted. errno=" << err; - continue; - } else if (err == EBADF) { - LOG(WARNING) << "The Timer module is destroyed."; - break; - } - LOG(ERROR) << "Epoll/Kqueue wait internal error. errno=" << err; - break; - } - - for (int32_t i = 0; i < ready_num; i++) { -#if defined(OS_LINUX) - struct epoll_event *event = &ready_events[i]; - int32_t timer_fd = event->data.fd; -#elif defined(OS_MACOSX) - struct kevent *event = &ready_events[i]; - int32_t timer_fd = event->ident; -#endif - - uint64_t exp = 0; - if (read(timer_fd, &exp, sizeof(exp)) < 0) { - if (errno != EBADF) { - LOG(ERROR) << "Failed to read timerfd=" << timer_fd << " errno=" << errno; - } - continue; - } - if (TimerFdCtxValidate((uint32_t)timer_fd) != UBRING_OK) { - continue; - } - - RETURN_CODE ret = ExecuteCallback(timer_fd); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed execute callback ret=" << ret; - DeleteTimerInner((uint32_t)timer_fd); - continue; - } - } - } - return nullptr; +uint32_t GetActiveTimerNum() { + return std::atomic_load(&g_total_timer_num); } -void DeleteTimerSafe(uint32_t fd) { - if (g_timer_fd_ctx_map == nullptr) { - return; - } +void DeleteTimerSafe(uint64_t timer_id) { + { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + return; + } - if (pthread_spin_lock(&g_timer_fd_ctx_map[fd].spin_lock) != 0) { - return; - } + bthread_timer_del(it->second.timer_id); - if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) { - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - return; + pthread_spin_destroy(&it->second.spin_lock); + g_timer_ctx_map.erase(it); } - - g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; - g_timer_fd_ctx_map[fd].cb = nullptr; - g_timer_fd_ctx_map[fd].args = nullptr; - g_timer_fd_ctx_map[fd].periodical = 0; - g_timer_fd_ctx_map[fd].fd = 0; - - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - -#if defined(OS_LINUX) - epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, (int)fd, nullptr); -#elif defined(OS_MACOSX) - struct kevent evt; - EV_SET(&evt, fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr); - kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr); -#endif - - uint64_t exp = 0; - read((int)fd, &exp, sizeof(exp)); - - close((int)fd); std::atomic_fetch_sub(&g_total_timer_num, 1U); } -void DeleteTimer(uint32_t fd) { - if (g_timer_fd_ctx_map == nullptr) { - LOG(WARNING) << "The timer is not initialized."; +void DeleteTimer(uint64_t timer_id) { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + LOG(WARNING) << "Timer id=" << timer_id << " not found"; return; } - - g_timer_fd_ctx_map[fd].periodical = 0; + it->second.periodical = 0; } -int32_t TimerStart(const itimerspec *time, void *(*cb)(void *), void *args) { - if (g_epoll_fd == -1) { - LOG(ERROR) << "Timer epoll/kqueue encountered internal error."; - return -1; - } +void TimerCallbackWrapper(void *arg) { + auto timer_id = reinterpret_cast(arg); -#if defined(OS_LINUX) - int timer_fd = timerfd_create(CLOCK_MONOTONIC, 0); -#elif defined(OS_MACOSX) - int timer_fd = timerfd_create_macosx(CLOCK_MONOTONIC, 0); -#endif - - if (UNLIKELY(timer_fd >= (int)g_max_system_fd || timer_fd == -1)) { - LOG(ERROR) << "Failed to create timerfd=" << timer_fd << " errno=" << errno; - return -1; - } - - g_timer_fd_ctx_map[timer_fd].status = TIMER_CONTEXT_EPOLL_WAITING; - g_timer_fd_ctx_map[timer_fd].cb = cb; - g_timer_fd_ctx_map[timer_fd].args = args; - g_timer_fd_ctx_map[timer_fd].fd = (uint32_t)timer_fd; - - if (LIKELY(time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0)) { - g_timer_fd_ctx_map[timer_fd].periodical = 1; + TimerContext *ctx = nullptr; + { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it==g_timer_ctx_map.end()) { + return; + } + ctx = &it->second; } -#if defined(OS_LINUX) - struct epoll_event event = { - .events = EPOLLIN, - .data = {.fd = timer_fd} - }; - - int32_t ret = epoll_ctl(g_epoll_fd, EPOLL_CTL_ADD, timer_fd, &event); -#elif defined(OS_MACOSX) - struct kevent event; - uint64_t timeout_nsec = time->it_value.tv_sec * 1000000000ULL + time->it_value.tv_nsec; - uint64_t interval_nsec = time->it_interval.tv_sec * 1000000000ULL + time->it_interval.tv_nsec; - EV_SET(&event, timer_fd, EVFILT_TIMER, EV_ADD | EV_ENABLE, 0, - timeout_nsec / 1000000, nullptr); - int32_t ret = kevent(g_epoll_fd, &event, 1, nullptr, 0, nullptr); -#endif - - if (UNLIKELY(ret != 0)) { - CloseTimerFd(timer_fd); - LOG(ERROR) << "Failed to add event to epoll/kqueue. errno=" << errno; - return -1; + if (ctx == nullptr || ctx->cb == nullptr) { + return; } - std::atomic_fetch_add(&g_total_timer_num, 1U); + void *cb_args = ctx->args; + auto is_periodical = ctx->periodical; + timespec interval = ctx->interval; -#if defined(OS_LINUX) - ret = timerfd_settime(timer_fd, 0, time, nullptr); -#elif defined(OS_MACOSX) - ret = timerfd_settime_macosx(timer_fd, 0, time, nullptr); -#endif - - if (UNLIKELY(ret != 0)) { -#if defined(OS_LINUX) - if (epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, timer_fd, nullptr) != 0) { -#elif defined(OS_MACOSX) - struct kevent evt; - EV_SET(&evt, timer_fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr); - if (kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr) != 0) { -#endif - LOG(ERROR) << "Failed to delete the timer fd=" << timer_fd << " with errno=" << errno; + if (!is_periodical) { + { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it!=g_timer_ctx_map.end()) { + pthread_spin_destroy(&it->second.spin_lock); + g_timer_ctx_map.erase(it); + } } - CloseTimerFd(timer_fd); std::atomic_fetch_sub(&g_total_timer_num, 1U); - LOG(ERROR) << "Failed to set timer"; - return -1; } - return timer_fd; -} + ctx->cb(cb_args); -uint32_t GetActiveTimerNum(void) { - return std::atomic_load(&g_total_timer_num); -} + if (is_periodical) { + auto now = std::chrono::steady_clock::now(); + auto interval_duration = std::chrono::seconds(interval.tv_sec) + std::chrono::nanoseconds(interval.tv_nsec); + auto future_time = now + interval_duration; -void CloseTimerFd(int fd) { - g_timer_fd_ctx_map[fd].cb = nullptr; - g_timer_fd_ctx_map[fd].args = nullptr; - g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; - g_timer_fd_ctx_map[fd].fd = 0; - g_timer_fd_ctx_map[fd].periodical = 0; - if (close((int)fd) != 0) { - LOG(ERROR) << "Failed to close timer fd=" << fd << " errno=" << errno; - return; - } -} + auto future_time_point = std::chrono::time_point_cast(future_time); + timespec abstime{}; + abstime.tv_sec=std::chrono::duration_cast(future_time_point.time_since_epoch()).count(); + abstime.tv_nsec = future_time_point.time_since_epoch().count() % 1000000000; -void TimerModuleDestroy(void) { - uint32_t max_fd = g_max_system_fd; - if (g_timer_fd_ctx_map) { - for (uint32_t fd = 0; fd < max_fd; fd++) { - if (g_timer_fd_ctx_map[fd].status != TIMER_CONTEXT_NOT_USING) { - DeleteTimerSafe(fd); - } - } - } - close(g_epoll_fd); - g_epoll_fd = -1; - g_total_timer_num = 0; - g_timer_module_initialized = 0; - int32_t ret = pthread_join(g_epoll_execute_thread, nullptr); - if (ret != EOK) { - LOG(ERROR) << "Failed to join pthread, during destroying timer module. ret=" << ret; - return; + bthread_timer_add(&ctx->timer_id, abstime, TimerCallbackWrapper, reinterpret_cast(timer_id)); } } -RETURN_CODE TimerFdCtxValidate(uint32_t fd) { - if (fd >= g_max_system_fd) { - LOG(ERROR) << "TimerFd=" << fd << " is out of range=" << g_max_system_fd; - return UBRING_ERR; - } - if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) { - LOG(ERROR) << "TimerFd=" << fd << " has wrong status=" << g_timer_fd_ctx_map[fd].status; - return UBRING_ERR; - } - if (g_timer_fd_ctx_map[fd].cb == nullptr) { - LOG(ERROR) << "The callback is not set."; - return UBRING_ERR; - } - - return UBRING_OK; -} -#if defined(OS_MACOSX) -static int timerfd_create_macosx(int clockid, int flags) { - int pipefd[2]; - if (pipe(pipefd) == -1) { - return -1; - } - return pipefd[0]; -} -static int timerfd_settime_macosx(int fd, int flags, - const itimerspec *new_value, - itimerspec *old_value) { - if (old_value != nullptr) { - memset(old_value, 0, sizeof(itimerspec)); - } - return 0; -} -#endif } // namespace ubring } // namespace brpc \ No newline at end of file diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index 1b42caef04..cb0ac856c9 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -18,56 +18,42 @@ #ifndef BRPC_TIMER_MGR_H #define BRPC_TIMER_MGR_H #include -#include +#include +#include +#include +#include +#include "bthread/types.h" +#include "bthread/unstable.h" #include "brpc/ubshm/common/common.h" -#if defined(OS_LINUX) -#include -#include -#elif defined(OS_MACOSX) -#include -#include -#include -#endif - -#define MAX_TIMER 1024 -#define TIMER_EPOLL_WAIT_TIMEOUT 1000 - -#if defined(OS_MACOSX) -struct itimerspec -{ - struct timespec it_interval; - struct timespec it_value; -}; -#endif namespace brpc { namespace ubring { -typedef enum { - TIMER_CONTEXT_NOT_USING, - TIMER_CONTEXT_EPOLL_WAITING, - TIMER_CONTEXT_CALLBACK_ONGOING -} TimerFdCtxStatus; -typedef struct { - void *(*cb)(void*); +typedef void * (*TimerCallback)(void *); + +struct TimerContext{ + TimerCallback cb; void *args; - uint32_t fd; - TimerFdCtxStatus status; uint32_t periodical; + timespec interval; + bthread_timer_t timer_id; pthread_spinlock_t spin_lock; -} TimerFdCtx; +}; + +extern std::unordered_map g_timer_ctx_map; +extern std::mutex g_timer_ctx_mutex; +extern std::atomic g_total_timer_num; -RETURN_CODE TimerInit(void); + +int TimerInit(void); void TimerModuleDestroy(void); -void *UnifiedCallback(void *args); -void *TimerEpoll(void *args); -int32_t TimerStart(const itimerspec *time, void *(*cb)(void *), void *args); +int32_t TimerStart(const itimerspec *time, TimerCallback cb, void *args); uint32_t GetActiveTimerNum(void); -void CloseTimerFd(int fd); -void DeleteTimerSafe(uint32_t fd); -void DeleteTimer(uint32_t fd); -RETURN_CODE TimerFdCtxValidate(uint32_t fd); +void DeleteTimerSafe(uint64_t timer_id); +void DeleteTimer(uint64_t timer_id); + +void TimerCallbackWrapper(void *arg); } } #endif //BRPC_TIMER_MGR_H \ No newline at end of file diff --git a/src/brpc/ubshm/ub_endpoint.cpp b/src/brpc/ubshm/ub_endpoint.cpp index 45794fdc6e..49fe732209 100644 --- a/src/brpc/ubshm/ub_endpoint.cpp +++ b/src/brpc/ubshm/ub_endpoint.cpp @@ -849,6 +849,7 @@ int UBShmEndpoint::PollingModeInitialize(bthread_tag_t tag, while (running->load(std::memory_order_relaxed)) { while (poller->op_queue.Dequeue(op)) { if (op.type == CqSidOp::ADD) { + cq_sids.erase(op); cq_sids.emplace(op); } else if (op.type == CqSidOp::REMOVE) { cq_sids.erase(op); From ccaecae50fe7429a2da56f8b857967c959c1afec Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Mon, 24 Aug 2026 21:54:37 +0800 Subject: [PATCH 02/15] =?UTF-8?q?[fix]1=E3=80=81=E5=B0=86=E5=8E=9F?= =?UTF-8?q?=E6=9C=89=E7=9A=84=E8=AE=A1=E6=97=B6=E5=99=A8=E6=9B=BF=E6=8D=A2?= =?UTF-8?q?=E4=B8=BAbrpc=E7=9A=84bthread=EF=BC=8C=E4=BB=A5=E6=AD=A4?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=88=A0=E9=99=A4=E8=AE=A1=E6=97=B6?= =?UTF-8?q?=E5=99=A8=E5=8F=AF=E8=83=BD=E4=BC=9A=E6=97=A0=E9=99=90=E6=9C=9F?= =?UTF-8?q?=E9=98=BB=E5=A1=9E=E3=80=81fd=E9=87=8D=E7=94=A8=E7=AB=9E?= =?UTF-8?q?=E9=80=9F=E3=80=81=E8=AE=A1=E6=97=B6=E5=99=A8=E5=88=9D=E5=A7=8B?= =?UTF-8?q?=E5=8C=96=E5=8C=85=E5=90=AB=E4=B8=80=E4=B8=AA=E7=AB=9E=E9=80=89?= =?UTF-8?q?=E3=80=81=E9=BB=98=E8=AE=A4=E7=9A=84=E9=97=AD=E9=97=A8=E5=AE=9A?= =?UTF-8?q?=E6=97=B6=E5=99=A8=E8=BF=87=E4=BA=8E=E9=A2=91=E7=B9=81=E3=80=81?= =?UTF-8?q?=E6=89=80=E6=9C=89=E5=AE=9A=E6=97=B6=E5=99=A8=E5=9B=9E=E8=B0=83?= =?UTF-8?q?=E9=83=BD=E5=9C=A8=E5=90=8C=E4=B8=80P=E7=BA=BF=E7=A8=8B?= =?UTF-8?q?=E4=B8=8A=E4=B8=B2=E8=A1=8C=E3=80=81macOS=E6=A8=A1=E6=8B=9F?= =?UTF-8?q?=E5=9C=A8=E5=8A=9F=E8=83=BD=E4=B8=8A=E5=B9=B6=E4=B8=8D=E7=AD=89?= =?UTF-8?q?=E6=95=88=E3=80=822=E3=80=81=E7=A7=BB=E9=99=A4=E5=9B=BA?= =?UTF-8?q?=E5=AE=9A=E6=95=B0=E7=BB=84=EF=BC=8C=E6=94=B9=E7=94=A8unordered?= =?UTF-8?q?=5Fmap=EF=BC=9B3=E3=80=81=E4=B8=BAitimerspec=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E5=BD=92=E4=B8=80=E5=8C=96=EF=BC=9B4=E3=80=81=E8=A7=A3?= =?UTF-8?q?=E5=86=B3CqSidOp=E7=9A=84sid=E7=9B=B8=E5=90=8C=E8=80=8Cevent?= =?UTF-8?q?=E4=B8=8D=E5=90=8C=E6=97=B6=E6=B7=BB=E5=8A=A0=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.cpp | 488 ++++++----------------------- src/brpc/ubshm/timer/timer_mgr.h | 62 ++-- src/brpc/ubshm/ub_endpoint.cpp | 1 + 3 files changed, 125 insertions(+), 426 deletions(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index b5e0c9ef3b..e60bb02cd4 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -14,455 +14,167 @@ // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. - #define _GNU_SOURCE + #include -#include -#include -#include -#include -#include +#include +#include +#include #include -#include +#include #include "brpc/ubshm/timer/timer_mgr.h" namespace brpc { namespace ubring { -int32_t g_epoll_fd = -1; -std::atomic g_total_timer_num(0); -TimerFdCtx *g_timer_fd_ctx_map = nullptr; -uint32_t g_max_system_fd = 0; -static pthread_t g_epoll_execute_thread = 0; -static int32_t g_timer_module_initialized = 0; - -#if defined(OS_MACOSX) -static int timerfd_create_macosx(int clockid, int flags); -static int timerfd_settime_macosx(int fd, int flags, - const itimerspec *new_value, - itimerspec *old_value); -#endif - -static RETURN_CODE DeleteTimerInner(uint32_t fd) { - if (g_timer_fd_ctx_map == nullptr) { - return UBRING_OK; - } - - if (pthread_spin_lock(&g_timer_fd_ctx_map[fd].spin_lock) != 0) { - return UBRING_ERR; - } - - if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) { - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - return UBRING_OK; - } - - g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; - g_timer_fd_ctx_map[fd].cb = nullptr; - g_timer_fd_ctx_map[fd].args = nullptr; - g_timer_fd_ctx_map[fd].periodical = 0; - g_timer_fd_ctx_map[fd].fd = 0; - - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - -#if defined(OS_LINUX) - epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, (int)fd, nullptr); -#elif defined(OS_MACOSX) - struct kevent evt; - EV_SET(&evt, fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr); - kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr); -#endif - - uint64_t exp = 0; - read((int)fd, &exp, sizeof(exp)); - - close((int)fd); - std::atomic_fetch_sub(&g_total_timer_num, 1U); - return UBRING_OK; -} - -static RETURN_CODE StartTimeEpoll(void) { -#if defined(OS_LINUX) - g_epoll_fd = epoll_create1(0); -#elif defined(OS_MACOSX) - g_epoll_fd = kqueue(); -#endif - if (UNLIKELY(g_epoll_fd == -1)) { - LOG(ERROR) << "Failed to create epoll/kqueue. errno=" << errno; - return UBRING_ERR; - } +std::unordered_map g_timer_ctx_map; +std::mutex g_timer_ctx_mutex; +std::atomic g_total_timer_num; - int ret = pthread_create(&g_epoll_execute_thread, nullptr, TimerEpoll, nullptr); - if (UNLIKELY(ret != 0)) { - LOG(ERROR) << "Failed to create thread err=" << ret; - return UBRING_ERR; - } - return UBRING_OK; -} +static std::atomic g_timer_id_counter(0); -static RETURN_CODE TimerSpinLocksInit(void) { - if (g_timer_fd_ctx_map == nullptr) { - LOG(ERROR) << "Timer module is not fully initialized."; - return UBRING_ERR; +static void normalize_timespec(itimerspec *spec) { + if (spec->it_interval.tv_nsec >= 1000000000L) { + spec->it_interval.tv_sec += spec->it_interval.tv_nsec / 1000000000L; + spec->it_interval.tv_nsec %= 1000000000L; } - - for (uint32_t fd = 0; fd < g_max_system_fd; fd++) { - int ret = pthread_spin_init(&g_timer_fd_ctx_map[fd].spin_lock, - PTHREAD_PROCESS_PRIVATE); - if (ret != EOK) { - LOG(ERROR) << "Failed to initialize spin lock for fd=" << fd; - for (uint32_t cleanup_fd = 0; cleanup_fd < fd; cleanup_fd++) { - pthread_spin_destroy(&g_timer_fd_ctx_map[cleanup_fd].spin_lock); - } - return UBRING_ERR; - } + if (spec->it_value.tv_nsec >= 1000000000L) { + spec->it_value.tv_sec += spec->it_value.tv_nsec / 1000000000L; + spec->it_value.tv_nsec %= 1000000000L; } - return UBRING_OK; -} - -static RETURN_CODE ExecuteCallback(int32_t timer_fd) { - UnifiedCallback((void *)(&g_timer_fd_ctx_map[timer_fd])); - return UBRING_OK; } -static RETURN_CODE TimerCtxMapCompletion(void) { - memset(g_timer_fd_ctx_map, 0, sizeof(TimerFdCtx) * g_max_system_fd); - RETURN_CODE ret = TimerSpinLocksInit(); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed to init spin locks for timer module."; - return UBRING_ERR; - } - return UBRING_OK; +int TimerInit() { + return 0; } -RETURN_CODE TimerInit(void) { - if (g_timer_module_initialized > 0) { - return UBRING_OK; +void TimerModuleDestroy() { + std::lock_guard lock(g_timer_ctx_mutex); + for (auto & pair: g_timer_ctx_map) { + bthread_timer_del(pair.second.timer_id); + pthread_spin_destroy(&pair.second.spin_lock); } - + g_timer_ctx_map.clear(); g_total_timer_num.store(0); +} - struct rlimit rlim; - if (getrlimit(RLIMIT_NOFILE, &rlim) != UBRING_OK) { - LOG(ERROR) << "Failed to get fd"; - return UBRING_ERR; +int32_t TimerStart(const itimerspec *time, TimerCallback cb, void *args) { + if (cb==nullptr) { + // LOG(ERROR) << "Timer callback is nullptr"; + return -1; } - g_max_system_fd = (uint32_t)rlim.rlim_cur; - if (g_timer_fd_ctx_map == nullptr) { - g_timer_fd_ctx_map = (TimerFdCtx *)malloc(sizeof(TimerFdCtx) * g_max_system_fd); - if (UNLIKELY(!g_timer_fd_ctx_map)) { - LOG(ERROR) << "Fail to malloc space for timer modules. errno=%d", errno; - return UBRING_ERR; - } + TimerContext ctx{}; + ctx.cb = cb; + ctx.args = args; + ctx.periodical = (time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0) ? 1 : 0; + ctx.interval = time->it_interval; + pthread_spin_init(&ctx.spin_lock, PTHREAD_PROCESS_PRIVATE); - RETURN_CODE ret = TimerCtxMapCompletion(); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed to init main data structure of Time Module. ret=" << ret; - free(g_timer_fd_ctx_map); - g_timer_fd_ctx_map = nullptr; - return UBRING_ERR; - } - } + uint64_t timer_id = g_timer_id_counter.fetch_add(1); - RETURN_CODE ret = StartTimeEpoll(); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed to start Timer Epoll. ret=" << ret; - if (LIKELY(g_timer_fd_ctx_map != nullptr)) { - FREE_PTR(g_timer_fd_ctx_map); - } - return UBRING_ERR; - } - g_timer_module_initialized = 1; - return UBRING_OK; -} + itimerspec normalized_time = *time; + normalize_timespec(&normalized_time); + timespec abstime = normalized_time.it_value; -void *UnifiedCallback(void *args) { - TimerFdCtx *ctx = (TimerFdCtx *)args; - if (pthread_spin_lock(&ctx->spin_lock) != 0) { - return nullptr; + int ret = bthread_timer_add(&ctx.timer_id, abstime, TimerCallbackWrapper, reinterpret_cast(timer_id)); + if (ret != 0) { + // LOG(ERROR) << "Failed to add bthread timer, ret=" << ret; + pthread_spin_destroy(&ctx.spin_lock); + return -1; } - if (ctx->status == TIMER_CONTEXT_NOT_USING) { - pthread_spin_unlock(&ctx->spin_lock); - return nullptr; + { + std::lock_guard lock(g_timer_ctx_mutex); + g_timer_ctx_map[timer_id] = ctx; } - void *(*cb)(void *) = ctx->cb; - void *cb_args = ctx->args; - uint32_t fd = ctx->fd; - int is_periodical = ctx->periodical; - ctx->status = TIMER_CONTEXT_CALLBACK_ONGOING; - - pthread_spin_unlock(&ctx->spin_lock); - - cb(cb_args); - - if (!is_periodical) { - DeleteTimerInner(fd); - } - return nullptr; + std::atomic_fetch_add(&g_total_timer_num, 1U); + return static_cast(timer_id); } -void *TimerEpoll(void *args) { - UNREFERENCE_PARAM(args); -#if defined(OS_LINUX) - struct epoll_event ready_events[MAX_TIMER]; -#elif defined(OS_MACOSX) - struct kevent ready_events[MAX_TIMER]; -#endif - - while (1) { - if (g_timer_module_initialized <= 0) { - LOG(ERROR) << "The Timer module is not initialized."; - break; - } - -#if defined(OS_LINUX) - int32_t ready_num = epoll_wait(g_epoll_fd, ready_events, MAX_TIMER, - TIMER_EPOLL_WAIT_TIMEOUT); -#elif defined(OS_MACOSX) - struct timespec timeout = {0, TIMER_EPOLL_WAIT_TIMEOUT * 1000000}; - int32_t ready_num = kevent(g_epoll_fd, nullptr, 0, ready_events, MAX_TIMER, &timeout); -#endif - - if (UNLIKELY(ready_num == -1)) { - errno_t err = errno; - if (err == EINTR) { - LOG_EVERY_SECOND(WARNING) << "Epoll/Kqueue wait was interrupted. errno=" << err; - continue; - } else if (err == EBADF) { - LOG(WARNING) << "The Timer module is destroyed."; - break; - } - LOG(ERROR) << "Epoll/Kqueue wait internal error. errno=" << err; - break; - } - - for (int32_t i = 0; i < ready_num; i++) { -#if defined(OS_LINUX) - struct epoll_event *event = &ready_events[i]; - int32_t timer_fd = event->data.fd; -#elif defined(OS_MACOSX) - struct kevent *event = &ready_events[i]; - int32_t timer_fd = event->ident; -#endif - - uint64_t exp = 0; - if (read(timer_fd, &exp, sizeof(exp)) < 0) { - if (errno != EBADF) { - LOG(ERROR) << "Failed to read timerfd=" << timer_fd << " errno=" << errno; - } - continue; - } - if (TimerFdCtxValidate((uint32_t)timer_fd) != UBRING_OK) { - continue; - } - - RETURN_CODE ret = ExecuteCallback(timer_fd); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed execute callback ret=" << ret; - DeleteTimerInner((uint32_t)timer_fd); - continue; - } - } - } - return nullptr; +uint32_t GetActiveTimerNum() { + return std::atomic_load(&g_total_timer_num); } -void DeleteTimerSafe(uint32_t fd) { - if (g_timer_fd_ctx_map == nullptr) { - return; - } +void DeleteTimerSafe(uint64_t timer_id) { + { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + return; + } - if (pthread_spin_lock(&g_timer_fd_ctx_map[fd].spin_lock) != 0) { - return; - } + bthread_timer_del(it->second.timer_id); - if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) { - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - return; + pthread_spin_destroy(&it->second.spin_lock); + g_timer_ctx_map.erase(it); } - - g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; - g_timer_fd_ctx_map[fd].cb = nullptr; - g_timer_fd_ctx_map[fd].args = nullptr; - g_timer_fd_ctx_map[fd].periodical = 0; - g_timer_fd_ctx_map[fd].fd = 0; - - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - -#if defined(OS_LINUX) - epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, (int)fd, nullptr); -#elif defined(OS_MACOSX) - struct kevent evt; - EV_SET(&evt, fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr); - kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr); -#endif - - uint64_t exp = 0; - read((int)fd, &exp, sizeof(exp)); - - close((int)fd); std::atomic_fetch_sub(&g_total_timer_num, 1U); } -void DeleteTimer(uint32_t fd) { - if (g_timer_fd_ctx_map == nullptr) { - LOG(WARNING) << "The timer is not initialized."; +void DeleteTimer(uint64_t timer_id) { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + LOG(WARNING) << "Timer id=" << timer_id << " not found"; return; } - - g_timer_fd_ctx_map[fd].periodical = 0; + it->second.periodical = 0; } -int32_t TimerStart(const itimerspec *time, void *(*cb)(void *), void *args) { - if (g_epoll_fd == -1) { - LOG(ERROR) << "Timer epoll/kqueue encountered internal error."; - return -1; - } +void TimerCallbackWrapper(void *arg) { + auto timer_id = reinterpret_cast(arg); -#if defined(OS_LINUX) - int timer_fd = timerfd_create(CLOCK_MONOTONIC, 0); -#elif defined(OS_MACOSX) - int timer_fd = timerfd_create_macosx(CLOCK_MONOTONIC, 0); -#endif - - if (UNLIKELY(timer_fd >= (int)g_max_system_fd || timer_fd == -1)) { - LOG(ERROR) << "Failed to create timerfd=" << timer_fd << " errno=" << errno; - return -1; - } - - g_timer_fd_ctx_map[timer_fd].status = TIMER_CONTEXT_EPOLL_WAITING; - g_timer_fd_ctx_map[timer_fd].cb = cb; - g_timer_fd_ctx_map[timer_fd].args = args; - g_timer_fd_ctx_map[timer_fd].fd = (uint32_t)timer_fd; - - if (LIKELY(time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0)) { - g_timer_fd_ctx_map[timer_fd].periodical = 1; + TimerContext *ctx = nullptr; + { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it==g_timer_ctx_map.end()) { + return; + } + ctx = &it->second; } -#if defined(OS_LINUX) - struct epoll_event event = { - .events = EPOLLIN, - .data = {.fd = timer_fd} - }; - - int32_t ret = epoll_ctl(g_epoll_fd, EPOLL_CTL_ADD, timer_fd, &event); -#elif defined(OS_MACOSX) - struct kevent event; - uint64_t timeout_nsec = time->it_value.tv_sec * 1000000000ULL + time->it_value.tv_nsec; - uint64_t interval_nsec = time->it_interval.tv_sec * 1000000000ULL + time->it_interval.tv_nsec; - EV_SET(&event, timer_fd, EVFILT_TIMER, EV_ADD | EV_ENABLE, 0, - timeout_nsec / 1000000, nullptr); - int32_t ret = kevent(g_epoll_fd, &event, 1, nullptr, 0, nullptr); -#endif - - if (UNLIKELY(ret != 0)) { - CloseTimerFd(timer_fd); - LOG(ERROR) << "Failed to add event to epoll/kqueue. errno=" << errno; - return -1; + if (ctx == nullptr || ctx->cb == nullptr) { + return; } - std::atomic_fetch_add(&g_total_timer_num, 1U); + void *cb_args = ctx->args; + auto is_periodical = ctx->periodical; + timespec interval = ctx->interval; -#if defined(OS_LINUX) - ret = timerfd_settime(timer_fd, 0, time, nullptr); -#elif defined(OS_MACOSX) - ret = timerfd_settime_macosx(timer_fd, 0, time, nullptr); -#endif - - if (UNLIKELY(ret != 0)) { -#if defined(OS_LINUX) - if (epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, timer_fd, nullptr) != 0) { -#elif defined(OS_MACOSX) - struct kevent evt; - EV_SET(&evt, timer_fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr); - if (kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr) != 0) { -#endif - LOG(ERROR) << "Failed to delete the timer fd=" << timer_fd << " with errno=" << errno; + if (!is_periodical) { + { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it!=g_timer_ctx_map.end()) { + pthread_spin_destroy(&it->second.spin_lock); + g_timer_ctx_map.erase(it); + } } - CloseTimerFd(timer_fd); std::atomic_fetch_sub(&g_total_timer_num, 1U); - LOG(ERROR) << "Failed to set timer"; - return -1; } - return timer_fd; -} + ctx->cb(cb_args); -uint32_t GetActiveTimerNum(void) { - return std::atomic_load(&g_total_timer_num); -} + if (is_periodical) { + auto now = std::chrono::steady_clock::now(); + auto interval_duration = std::chrono::seconds(interval.tv_sec) + std::chrono::nanoseconds(interval.tv_nsec); + auto future_time = now + interval_duration; -void CloseTimerFd(int fd) { - g_timer_fd_ctx_map[fd].cb = nullptr; - g_timer_fd_ctx_map[fd].args = nullptr; - g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; - g_timer_fd_ctx_map[fd].fd = 0; - g_timer_fd_ctx_map[fd].periodical = 0; - if (close((int)fd) != 0) { - LOG(ERROR) << "Failed to close timer fd=" << fd << " errno=" << errno; - return; - } -} + auto future_time_point = std::chrono::time_point_cast(future_time); + timespec abstime{}; + abstime.tv_sec=std::chrono::duration_cast(future_time_point.time_since_epoch()).count(); + abstime.tv_nsec = future_time_point.time_since_epoch().count() % 1000000000; -void TimerModuleDestroy(void) { - uint32_t max_fd = g_max_system_fd; - if (g_timer_fd_ctx_map) { - for (uint32_t fd = 0; fd < max_fd; fd++) { - if (g_timer_fd_ctx_map[fd].status != TIMER_CONTEXT_NOT_USING) { - DeleteTimerSafe(fd); - } - } - } - close(g_epoll_fd); - g_epoll_fd = -1; - g_total_timer_num = 0; - g_timer_module_initialized = 0; - int32_t ret = pthread_join(g_epoll_execute_thread, nullptr); - if (ret != EOK) { - LOG(ERROR) << "Failed to join pthread, during destroying timer module. ret=" << ret; - return; + bthread_timer_add(&ctx->timer_id, abstime, TimerCallbackWrapper, reinterpret_cast(timer_id)); } } -RETURN_CODE TimerFdCtxValidate(uint32_t fd) { - if (fd >= g_max_system_fd) { - LOG(ERROR) << "TimerFd=" << fd << " is out of range=" << g_max_system_fd; - return UBRING_ERR; - } - if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) { - LOG(ERROR) << "TimerFd=" << fd << " has wrong status=" << g_timer_fd_ctx_map[fd].status; - return UBRING_ERR; - } - if (g_timer_fd_ctx_map[fd].cb == nullptr) { - LOG(ERROR) << "The callback is not set."; - return UBRING_ERR; - } - - return UBRING_OK; -} -#if defined(OS_MACOSX) -static int timerfd_create_macosx(int clockid, int flags) { - int pipefd[2]; - if (pipe(pipefd) == -1) { - return -1; - } - return pipefd[0]; -} -static int timerfd_settime_macosx(int fd, int flags, - const itimerspec *new_value, - itimerspec *old_value) { - if (old_value != nullptr) { - memset(old_value, 0, sizeof(itimerspec)); - } - return 0; -} -#endif } // namespace ubring } // namespace brpc \ No newline at end of file diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index 1b42caef04..cb0ac856c9 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -18,56 +18,42 @@ #ifndef BRPC_TIMER_MGR_H #define BRPC_TIMER_MGR_H #include -#include +#include +#include +#include +#include +#include "bthread/types.h" +#include "bthread/unstable.h" #include "brpc/ubshm/common/common.h" -#if defined(OS_LINUX) -#include -#include -#elif defined(OS_MACOSX) -#include -#include -#include -#endif - -#define MAX_TIMER 1024 -#define TIMER_EPOLL_WAIT_TIMEOUT 1000 - -#if defined(OS_MACOSX) -struct itimerspec -{ - struct timespec it_interval; - struct timespec it_value; -}; -#endif namespace brpc { namespace ubring { -typedef enum { - TIMER_CONTEXT_NOT_USING, - TIMER_CONTEXT_EPOLL_WAITING, - TIMER_CONTEXT_CALLBACK_ONGOING -} TimerFdCtxStatus; -typedef struct { - void *(*cb)(void*); +typedef void * (*TimerCallback)(void *); + +struct TimerContext{ + TimerCallback cb; void *args; - uint32_t fd; - TimerFdCtxStatus status; uint32_t periodical; + timespec interval; + bthread_timer_t timer_id; pthread_spinlock_t spin_lock; -} TimerFdCtx; +}; + +extern std::unordered_map g_timer_ctx_map; +extern std::mutex g_timer_ctx_mutex; +extern std::atomic g_total_timer_num; -RETURN_CODE TimerInit(void); + +int TimerInit(void); void TimerModuleDestroy(void); -void *UnifiedCallback(void *args); -void *TimerEpoll(void *args); -int32_t TimerStart(const itimerspec *time, void *(*cb)(void *), void *args); +int32_t TimerStart(const itimerspec *time, TimerCallback cb, void *args); uint32_t GetActiveTimerNum(void); -void CloseTimerFd(int fd); -void DeleteTimerSafe(uint32_t fd); -void DeleteTimer(uint32_t fd); -RETURN_CODE TimerFdCtxValidate(uint32_t fd); +void DeleteTimerSafe(uint64_t timer_id); +void DeleteTimer(uint64_t timer_id); + +void TimerCallbackWrapper(void *arg); } } #endif //BRPC_TIMER_MGR_H \ No newline at end of file diff --git a/src/brpc/ubshm/ub_endpoint.cpp b/src/brpc/ubshm/ub_endpoint.cpp index 31539fda85..f6e0ef7ba0 100644 --- a/src/brpc/ubshm/ub_endpoint.cpp +++ b/src/brpc/ubshm/ub_endpoint.cpp @@ -848,6 +848,7 @@ int UBShmEndpoint::PollingModeInitialize(bthread_tag_t tag, while (running->load(std::memory_order_relaxed)) { while (poller->op_queue.Dequeue(op)) { if (op.type == PollerSidOp::ADD) { + poller_sids.erase(op); poller_sids.emplace(op); } else if (op.type == PollerSidOp::REMOVE) { poller_sids.erase(op); From a40c675b6f85f9192173a659c272a71cb422f0dd Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Tue, 25 Aug 2026 18:02:27 +0800 Subject: [PATCH 03/15] =?UTF-8?q?[fix]1=E3=80=81=E5=B0=86=E5=8E=9F?= =?UTF-8?q?=E6=9C=89=E7=9A=84=E8=AE=A1=E6=97=B6=E5=99=A8=E6=9B=BF=E6=8D=A2?= =?UTF-8?q?=E4=B8=BAbrpc=E7=9A=84bthread=EF=BC=8C=E4=BB=A5=E6=AD=A4?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=88=A0=E9=99=A4=E8=AE=A1=E6=97=B6?= =?UTF-8?q?=E5=99=A8=E5=8F=AF=E8=83=BD=E4=BC=9A=E6=97=A0=E9=99=90=E6=9C=9F?= =?UTF-8?q?=E9=98=BB=E5=A1=9E=E3=80=81fd=E9=87=8D=E7=94=A8=E7=AB=9E?= =?UTF-8?q?=E9=80=9F=E3=80=81=E8=AE=A1=E6=97=B6=E5=99=A8=E5=88=9D=E5=A7=8B?= =?UTF-8?q?=E5=8C=96=E5=8C=85=E5=90=AB=E4=B8=80=E4=B8=AA=E7=AB=9E=E9=80=89?= =?UTF-8?q?=E3=80=81=E9=BB=98=E8=AE=A4=E7=9A=84=E9=97=AD=E9=97=A8=E5=AE=9A?= =?UTF-8?q?=E6=97=B6=E5=99=A8=E8=BF=87=E4=BA=8E=E9=A2=91=E7=B9=81=E3=80=81?= =?UTF-8?q?=E6=89=80=E6=9C=89=E5=AE=9A=E6=97=B6=E5=99=A8=E5=9B=9E=E8=B0=83?= =?UTF-8?q?=E9=83=BD=E5=9C=A8=E5=90=8C=E4=B8=80P=E7=BA=BF=E7=A8=8B?= =?UTF-8?q?=E4=B8=8A=E4=B8=B2=E8=A1=8C=E3=80=81macOS=E6=A8=A1=E6=8B=9F?= =?UTF-8?q?=E5=9C=A8=E5=8A=9F=E8=83=BD=E4=B8=8A=E5=B9=B6=E4=B8=8D=E7=AD=89?= =?UTF-8?q?=E6=95=88=E3=80=822=E3=80=81=E7=A7=BB=E9=99=A4=E5=9B=BA?= =?UTF-8?q?=E5=AE=9A=E6=95=B0=E7=BB=84=EF=BC=8C=E6=94=B9=E7=94=A8unordered?= =?UTF-8?q?=5Fmap=EF=BC=9B3=E3=80=81=E4=B8=BAitimerspec=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E5=BD=92=E4=B8=80=E5=8C=96=EF=BC=9B4=E3=80=81=E8=A7=A3?= =?UTF-8?q?=E5=86=B3CqSidOp=E7=9A=84sid=E7=9B=B8=E5=90=8C=E8=80=8Cevent?= =?UTF-8?q?=E4=B8=8D=E5=90=8C=E6=97=B6=E6=B7=BB=E5=8A=A0=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.h | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index cb0ac856c9..088cde00b9 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -26,6 +26,14 @@ #include "bthread/unstable.h" #include "brpc/ubshm/common/common.h" +#if defined(OS_MACOSX) +struct itimerspec +{ + struct timespec it_interval; + struct timespec it_value; +}; +#endif + namespace brpc { namespace ubring { From 3858f5e8bfe77f02adf28458887a30c9f9d01bee Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Thu, 27 Aug 2026 09:52:10 +0800 Subject: [PATCH 04/15] =?UTF-8?q?[fix]=E4=BF=AE=E5=A4=8D=E9=AD=94=E9=AC=BC?= =?UTF-8?q?=E6=95=B0=E5=AD=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.cpp | 12 ++++++------ src/brpc/ubshm/timer/timer_mgr.h | 2 ++ 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index e60bb02cd4..cc17af7b8e 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -34,13 +34,13 @@ std::atomic g_total_timer_num; static std::atomic g_timer_id_counter(0); static void normalize_timespec(itimerspec *spec) { - if (spec->it_interval.tv_nsec >= 1000000000L) { - spec->it_interval.tv_sec += spec->it_interval.tv_nsec / 1000000000L; - spec->it_interval.tv_nsec %= 1000000000L; + if (spec->it_interval.tv_nsec >= NS_PER_SEC) { + spec->it_interval.tv_sec += spec->it_interval.tv_nsec / NS_PER_SEC; + spec->it_interval.tv_nsec %= NS_PER_SEC; } - if (spec->it_value.tv_nsec >= 1000000000L) { - spec->it_value.tv_sec += spec->it_value.tv_nsec / 1000000000L; - spec->it_value.tv_nsec %= 1000000000L; + if (spec->it_value.tv_nsec >= NS_PER_SEC) { + spec->it_value.tv_sec += spec->it_value.tv_nsec / NS_PER_SEC; + spec->it_value.tv_nsec %= NS_PER_SEC; } } diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index 088cde00b9..10808ee1c2 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -37,6 +37,8 @@ struct itimerspec namespace brpc { namespace ubring { +constexpr long long NS_PER_SEC = 1000000000LL; + typedef void * (*TimerCallback)(void *); struct TimerContext{ From b1bf4f754420d24f2aee8c7de21c949dffefc634 Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Mon, 24 Aug 2026 21:54:37 +0800 Subject: [PATCH 05/15] =?UTF-8?q?[fix]1=E3=80=81=E5=B0=86=E5=8E=9F?= =?UTF-8?q?=E6=9C=89=E7=9A=84=E8=AE=A1=E6=97=B6=E5=99=A8=E6=9B=BF=E6=8D=A2?= =?UTF-8?q?=E4=B8=BAbrpc=E7=9A=84bthread=EF=BC=8C=E4=BB=A5=E6=AD=A4?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=88=A0=E9=99=A4=E8=AE=A1=E6=97=B6?= =?UTF-8?q?=E5=99=A8=E5=8F=AF=E8=83=BD=E4=BC=9A=E6=97=A0=E9=99=90=E6=9C=9F?= =?UTF-8?q?=E9=98=BB=E5=A1=9E=E3=80=81fd=E9=87=8D=E7=94=A8=E7=AB=9E?= =?UTF-8?q?=E9=80=9F=E3=80=81=E8=AE=A1=E6=97=B6=E5=99=A8=E5=88=9D=E5=A7=8B?= =?UTF-8?q?=E5=8C=96=E5=8C=85=E5=90=AB=E4=B8=80=E4=B8=AA=E7=AB=9E=E9=80=89?= =?UTF-8?q?=E3=80=81=E9=BB=98=E8=AE=A4=E7=9A=84=E9=97=AD=E9=97=A8=E5=AE=9A?= =?UTF-8?q?=E6=97=B6=E5=99=A8=E8=BF=87=E4=BA=8E=E9=A2=91=E7=B9=81=E3=80=81?= =?UTF-8?q?=E6=89=80=E6=9C=89=E5=AE=9A=E6=97=B6=E5=99=A8=E5=9B=9E=E8=B0=83?= =?UTF-8?q?=E9=83=BD=E5=9C=A8=E5=90=8C=E4=B8=80P=E7=BA=BF=E7=A8=8B?= =?UTF-8?q?=E4=B8=8A=E4=B8=B2=E8=A1=8C=E3=80=81macOS=E6=A8=A1=E6=8B=9F?= =?UTF-8?q?=E5=9C=A8=E5=8A=9F=E8=83=BD=E4=B8=8A=E5=B9=B6=E4=B8=8D=E7=AD=89?= =?UTF-8?q?=E6=95=88=E3=80=822=E3=80=81=E7=A7=BB=E9=99=A4=E5=9B=BA?= =?UTF-8?q?=E5=AE=9A=E6=95=B0=E7=BB=84=EF=BC=8C=E6=94=B9=E7=94=A8unordered?= =?UTF-8?q?=5Fmap=EF=BC=9B3=E3=80=81=E4=B8=BAitimerspec=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E5=BD=92=E4=B8=80=E5=8C=96=EF=BC=9B4=E3=80=81=E8=A7=A3?= =?UTF-8?q?=E5=86=B3CqSidOp=E7=9A=84sid=E7=9B=B8=E5=90=8C=E8=80=8Cevent?= =?UTF-8?q?=E4=B8=8D=E5=90=8C=E6=97=B6=E6=B7=BB=E5=8A=A0=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.cpp | 488 ++++++----------------------- src/brpc/ubshm/timer/timer_mgr.h | 62 ++-- src/brpc/ubshm/ub_endpoint.cpp | 1 + 3 files changed, 125 insertions(+), 426 deletions(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index b5e0c9ef3b..e60bb02cd4 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -14,455 +14,167 @@ // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. - #define _GNU_SOURCE + #include -#include -#include -#include -#include -#include +#include +#include +#include #include -#include +#include #include "brpc/ubshm/timer/timer_mgr.h" namespace brpc { namespace ubring { -int32_t g_epoll_fd = -1; -std::atomic g_total_timer_num(0); -TimerFdCtx *g_timer_fd_ctx_map = nullptr; -uint32_t g_max_system_fd = 0; -static pthread_t g_epoll_execute_thread = 0; -static int32_t g_timer_module_initialized = 0; - -#if defined(OS_MACOSX) -static int timerfd_create_macosx(int clockid, int flags); -static int timerfd_settime_macosx(int fd, int flags, - const itimerspec *new_value, - itimerspec *old_value); -#endif - -static RETURN_CODE DeleteTimerInner(uint32_t fd) { - if (g_timer_fd_ctx_map == nullptr) { - return UBRING_OK; - } - - if (pthread_spin_lock(&g_timer_fd_ctx_map[fd].spin_lock) != 0) { - return UBRING_ERR; - } - - if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) { - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - return UBRING_OK; - } - - g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; - g_timer_fd_ctx_map[fd].cb = nullptr; - g_timer_fd_ctx_map[fd].args = nullptr; - g_timer_fd_ctx_map[fd].periodical = 0; - g_timer_fd_ctx_map[fd].fd = 0; - - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - -#if defined(OS_LINUX) - epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, (int)fd, nullptr); -#elif defined(OS_MACOSX) - struct kevent evt; - EV_SET(&evt, fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr); - kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr); -#endif - - uint64_t exp = 0; - read((int)fd, &exp, sizeof(exp)); - - close((int)fd); - std::atomic_fetch_sub(&g_total_timer_num, 1U); - return UBRING_OK; -} - -static RETURN_CODE StartTimeEpoll(void) { -#if defined(OS_LINUX) - g_epoll_fd = epoll_create1(0); -#elif defined(OS_MACOSX) - g_epoll_fd = kqueue(); -#endif - if (UNLIKELY(g_epoll_fd == -1)) { - LOG(ERROR) << "Failed to create epoll/kqueue. errno=" << errno; - return UBRING_ERR; - } +std::unordered_map g_timer_ctx_map; +std::mutex g_timer_ctx_mutex; +std::atomic g_total_timer_num; - int ret = pthread_create(&g_epoll_execute_thread, nullptr, TimerEpoll, nullptr); - if (UNLIKELY(ret != 0)) { - LOG(ERROR) << "Failed to create thread err=" << ret; - return UBRING_ERR; - } - return UBRING_OK; -} +static std::atomic g_timer_id_counter(0); -static RETURN_CODE TimerSpinLocksInit(void) { - if (g_timer_fd_ctx_map == nullptr) { - LOG(ERROR) << "Timer module is not fully initialized."; - return UBRING_ERR; +static void normalize_timespec(itimerspec *spec) { + if (spec->it_interval.tv_nsec >= 1000000000L) { + spec->it_interval.tv_sec += spec->it_interval.tv_nsec / 1000000000L; + spec->it_interval.tv_nsec %= 1000000000L; } - - for (uint32_t fd = 0; fd < g_max_system_fd; fd++) { - int ret = pthread_spin_init(&g_timer_fd_ctx_map[fd].spin_lock, - PTHREAD_PROCESS_PRIVATE); - if (ret != EOK) { - LOG(ERROR) << "Failed to initialize spin lock for fd=" << fd; - for (uint32_t cleanup_fd = 0; cleanup_fd < fd; cleanup_fd++) { - pthread_spin_destroy(&g_timer_fd_ctx_map[cleanup_fd].spin_lock); - } - return UBRING_ERR; - } + if (spec->it_value.tv_nsec >= 1000000000L) { + spec->it_value.tv_sec += spec->it_value.tv_nsec / 1000000000L; + spec->it_value.tv_nsec %= 1000000000L; } - return UBRING_OK; -} - -static RETURN_CODE ExecuteCallback(int32_t timer_fd) { - UnifiedCallback((void *)(&g_timer_fd_ctx_map[timer_fd])); - return UBRING_OK; } -static RETURN_CODE TimerCtxMapCompletion(void) { - memset(g_timer_fd_ctx_map, 0, sizeof(TimerFdCtx) * g_max_system_fd); - RETURN_CODE ret = TimerSpinLocksInit(); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed to init spin locks for timer module."; - return UBRING_ERR; - } - return UBRING_OK; +int TimerInit() { + return 0; } -RETURN_CODE TimerInit(void) { - if (g_timer_module_initialized > 0) { - return UBRING_OK; +void TimerModuleDestroy() { + std::lock_guard lock(g_timer_ctx_mutex); + for (auto & pair: g_timer_ctx_map) { + bthread_timer_del(pair.second.timer_id); + pthread_spin_destroy(&pair.second.spin_lock); } - + g_timer_ctx_map.clear(); g_total_timer_num.store(0); +} - struct rlimit rlim; - if (getrlimit(RLIMIT_NOFILE, &rlim) != UBRING_OK) { - LOG(ERROR) << "Failed to get fd"; - return UBRING_ERR; +int32_t TimerStart(const itimerspec *time, TimerCallback cb, void *args) { + if (cb==nullptr) { + // LOG(ERROR) << "Timer callback is nullptr"; + return -1; } - g_max_system_fd = (uint32_t)rlim.rlim_cur; - if (g_timer_fd_ctx_map == nullptr) { - g_timer_fd_ctx_map = (TimerFdCtx *)malloc(sizeof(TimerFdCtx) * g_max_system_fd); - if (UNLIKELY(!g_timer_fd_ctx_map)) { - LOG(ERROR) << "Fail to malloc space for timer modules. errno=%d", errno; - return UBRING_ERR; - } + TimerContext ctx{}; + ctx.cb = cb; + ctx.args = args; + ctx.periodical = (time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0) ? 1 : 0; + ctx.interval = time->it_interval; + pthread_spin_init(&ctx.spin_lock, PTHREAD_PROCESS_PRIVATE); - RETURN_CODE ret = TimerCtxMapCompletion(); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed to init main data structure of Time Module. ret=" << ret; - free(g_timer_fd_ctx_map); - g_timer_fd_ctx_map = nullptr; - return UBRING_ERR; - } - } + uint64_t timer_id = g_timer_id_counter.fetch_add(1); - RETURN_CODE ret = StartTimeEpoll(); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed to start Timer Epoll. ret=" << ret; - if (LIKELY(g_timer_fd_ctx_map != nullptr)) { - FREE_PTR(g_timer_fd_ctx_map); - } - return UBRING_ERR; - } - g_timer_module_initialized = 1; - return UBRING_OK; -} + itimerspec normalized_time = *time; + normalize_timespec(&normalized_time); + timespec abstime = normalized_time.it_value; -void *UnifiedCallback(void *args) { - TimerFdCtx *ctx = (TimerFdCtx *)args; - if (pthread_spin_lock(&ctx->spin_lock) != 0) { - return nullptr; + int ret = bthread_timer_add(&ctx.timer_id, abstime, TimerCallbackWrapper, reinterpret_cast(timer_id)); + if (ret != 0) { + // LOG(ERROR) << "Failed to add bthread timer, ret=" << ret; + pthread_spin_destroy(&ctx.spin_lock); + return -1; } - if (ctx->status == TIMER_CONTEXT_NOT_USING) { - pthread_spin_unlock(&ctx->spin_lock); - return nullptr; + { + std::lock_guard lock(g_timer_ctx_mutex); + g_timer_ctx_map[timer_id] = ctx; } - void *(*cb)(void *) = ctx->cb; - void *cb_args = ctx->args; - uint32_t fd = ctx->fd; - int is_periodical = ctx->periodical; - ctx->status = TIMER_CONTEXT_CALLBACK_ONGOING; - - pthread_spin_unlock(&ctx->spin_lock); - - cb(cb_args); - - if (!is_periodical) { - DeleteTimerInner(fd); - } - return nullptr; + std::atomic_fetch_add(&g_total_timer_num, 1U); + return static_cast(timer_id); } -void *TimerEpoll(void *args) { - UNREFERENCE_PARAM(args); -#if defined(OS_LINUX) - struct epoll_event ready_events[MAX_TIMER]; -#elif defined(OS_MACOSX) - struct kevent ready_events[MAX_TIMER]; -#endif - - while (1) { - if (g_timer_module_initialized <= 0) { - LOG(ERROR) << "The Timer module is not initialized."; - break; - } - -#if defined(OS_LINUX) - int32_t ready_num = epoll_wait(g_epoll_fd, ready_events, MAX_TIMER, - TIMER_EPOLL_WAIT_TIMEOUT); -#elif defined(OS_MACOSX) - struct timespec timeout = {0, TIMER_EPOLL_WAIT_TIMEOUT * 1000000}; - int32_t ready_num = kevent(g_epoll_fd, nullptr, 0, ready_events, MAX_TIMER, &timeout); -#endif - - if (UNLIKELY(ready_num == -1)) { - errno_t err = errno; - if (err == EINTR) { - LOG_EVERY_SECOND(WARNING) << "Epoll/Kqueue wait was interrupted. errno=" << err; - continue; - } else if (err == EBADF) { - LOG(WARNING) << "The Timer module is destroyed."; - break; - } - LOG(ERROR) << "Epoll/Kqueue wait internal error. errno=" << err; - break; - } - - for (int32_t i = 0; i < ready_num; i++) { -#if defined(OS_LINUX) - struct epoll_event *event = &ready_events[i]; - int32_t timer_fd = event->data.fd; -#elif defined(OS_MACOSX) - struct kevent *event = &ready_events[i]; - int32_t timer_fd = event->ident; -#endif - - uint64_t exp = 0; - if (read(timer_fd, &exp, sizeof(exp)) < 0) { - if (errno != EBADF) { - LOG(ERROR) << "Failed to read timerfd=" << timer_fd << " errno=" << errno; - } - continue; - } - if (TimerFdCtxValidate((uint32_t)timer_fd) != UBRING_OK) { - continue; - } - - RETURN_CODE ret = ExecuteCallback(timer_fd); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed execute callback ret=" << ret; - DeleteTimerInner((uint32_t)timer_fd); - continue; - } - } - } - return nullptr; +uint32_t GetActiveTimerNum() { + return std::atomic_load(&g_total_timer_num); } -void DeleteTimerSafe(uint32_t fd) { - if (g_timer_fd_ctx_map == nullptr) { - return; - } +void DeleteTimerSafe(uint64_t timer_id) { + { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + return; + } - if (pthread_spin_lock(&g_timer_fd_ctx_map[fd].spin_lock) != 0) { - return; - } + bthread_timer_del(it->second.timer_id); - if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) { - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - return; + pthread_spin_destroy(&it->second.spin_lock); + g_timer_ctx_map.erase(it); } - - g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; - g_timer_fd_ctx_map[fd].cb = nullptr; - g_timer_fd_ctx_map[fd].args = nullptr; - g_timer_fd_ctx_map[fd].periodical = 0; - g_timer_fd_ctx_map[fd].fd = 0; - - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - -#if defined(OS_LINUX) - epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, (int)fd, nullptr); -#elif defined(OS_MACOSX) - struct kevent evt; - EV_SET(&evt, fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr); - kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr); -#endif - - uint64_t exp = 0; - read((int)fd, &exp, sizeof(exp)); - - close((int)fd); std::atomic_fetch_sub(&g_total_timer_num, 1U); } -void DeleteTimer(uint32_t fd) { - if (g_timer_fd_ctx_map == nullptr) { - LOG(WARNING) << "The timer is not initialized."; +void DeleteTimer(uint64_t timer_id) { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + LOG(WARNING) << "Timer id=" << timer_id << " not found"; return; } - - g_timer_fd_ctx_map[fd].periodical = 0; + it->second.periodical = 0; } -int32_t TimerStart(const itimerspec *time, void *(*cb)(void *), void *args) { - if (g_epoll_fd == -1) { - LOG(ERROR) << "Timer epoll/kqueue encountered internal error."; - return -1; - } +void TimerCallbackWrapper(void *arg) { + auto timer_id = reinterpret_cast(arg); -#if defined(OS_LINUX) - int timer_fd = timerfd_create(CLOCK_MONOTONIC, 0); -#elif defined(OS_MACOSX) - int timer_fd = timerfd_create_macosx(CLOCK_MONOTONIC, 0); -#endif - - if (UNLIKELY(timer_fd >= (int)g_max_system_fd || timer_fd == -1)) { - LOG(ERROR) << "Failed to create timerfd=" << timer_fd << " errno=" << errno; - return -1; - } - - g_timer_fd_ctx_map[timer_fd].status = TIMER_CONTEXT_EPOLL_WAITING; - g_timer_fd_ctx_map[timer_fd].cb = cb; - g_timer_fd_ctx_map[timer_fd].args = args; - g_timer_fd_ctx_map[timer_fd].fd = (uint32_t)timer_fd; - - if (LIKELY(time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0)) { - g_timer_fd_ctx_map[timer_fd].periodical = 1; + TimerContext *ctx = nullptr; + { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it==g_timer_ctx_map.end()) { + return; + } + ctx = &it->second; } -#if defined(OS_LINUX) - struct epoll_event event = { - .events = EPOLLIN, - .data = {.fd = timer_fd} - }; - - int32_t ret = epoll_ctl(g_epoll_fd, EPOLL_CTL_ADD, timer_fd, &event); -#elif defined(OS_MACOSX) - struct kevent event; - uint64_t timeout_nsec = time->it_value.tv_sec * 1000000000ULL + time->it_value.tv_nsec; - uint64_t interval_nsec = time->it_interval.tv_sec * 1000000000ULL + time->it_interval.tv_nsec; - EV_SET(&event, timer_fd, EVFILT_TIMER, EV_ADD | EV_ENABLE, 0, - timeout_nsec / 1000000, nullptr); - int32_t ret = kevent(g_epoll_fd, &event, 1, nullptr, 0, nullptr); -#endif - - if (UNLIKELY(ret != 0)) { - CloseTimerFd(timer_fd); - LOG(ERROR) << "Failed to add event to epoll/kqueue. errno=" << errno; - return -1; + if (ctx == nullptr || ctx->cb == nullptr) { + return; } - std::atomic_fetch_add(&g_total_timer_num, 1U); + void *cb_args = ctx->args; + auto is_periodical = ctx->periodical; + timespec interval = ctx->interval; -#if defined(OS_LINUX) - ret = timerfd_settime(timer_fd, 0, time, nullptr); -#elif defined(OS_MACOSX) - ret = timerfd_settime_macosx(timer_fd, 0, time, nullptr); -#endif - - if (UNLIKELY(ret != 0)) { -#if defined(OS_LINUX) - if (epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, timer_fd, nullptr) != 0) { -#elif defined(OS_MACOSX) - struct kevent evt; - EV_SET(&evt, timer_fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr); - if (kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr) != 0) { -#endif - LOG(ERROR) << "Failed to delete the timer fd=" << timer_fd << " with errno=" << errno; + if (!is_periodical) { + { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it!=g_timer_ctx_map.end()) { + pthread_spin_destroy(&it->second.spin_lock); + g_timer_ctx_map.erase(it); + } } - CloseTimerFd(timer_fd); std::atomic_fetch_sub(&g_total_timer_num, 1U); - LOG(ERROR) << "Failed to set timer"; - return -1; } - return timer_fd; -} + ctx->cb(cb_args); -uint32_t GetActiveTimerNum(void) { - return std::atomic_load(&g_total_timer_num); -} + if (is_periodical) { + auto now = std::chrono::steady_clock::now(); + auto interval_duration = std::chrono::seconds(interval.tv_sec) + std::chrono::nanoseconds(interval.tv_nsec); + auto future_time = now + interval_duration; -void CloseTimerFd(int fd) { - g_timer_fd_ctx_map[fd].cb = nullptr; - g_timer_fd_ctx_map[fd].args = nullptr; - g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; - g_timer_fd_ctx_map[fd].fd = 0; - g_timer_fd_ctx_map[fd].periodical = 0; - if (close((int)fd) != 0) { - LOG(ERROR) << "Failed to close timer fd=" << fd << " errno=" << errno; - return; - } -} + auto future_time_point = std::chrono::time_point_cast(future_time); + timespec abstime{}; + abstime.tv_sec=std::chrono::duration_cast(future_time_point.time_since_epoch()).count(); + abstime.tv_nsec = future_time_point.time_since_epoch().count() % 1000000000; -void TimerModuleDestroy(void) { - uint32_t max_fd = g_max_system_fd; - if (g_timer_fd_ctx_map) { - for (uint32_t fd = 0; fd < max_fd; fd++) { - if (g_timer_fd_ctx_map[fd].status != TIMER_CONTEXT_NOT_USING) { - DeleteTimerSafe(fd); - } - } - } - close(g_epoll_fd); - g_epoll_fd = -1; - g_total_timer_num = 0; - g_timer_module_initialized = 0; - int32_t ret = pthread_join(g_epoll_execute_thread, nullptr); - if (ret != EOK) { - LOG(ERROR) << "Failed to join pthread, during destroying timer module. ret=" << ret; - return; + bthread_timer_add(&ctx->timer_id, abstime, TimerCallbackWrapper, reinterpret_cast(timer_id)); } } -RETURN_CODE TimerFdCtxValidate(uint32_t fd) { - if (fd >= g_max_system_fd) { - LOG(ERROR) << "TimerFd=" << fd << " is out of range=" << g_max_system_fd; - return UBRING_ERR; - } - if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) { - LOG(ERROR) << "TimerFd=" << fd << " has wrong status=" << g_timer_fd_ctx_map[fd].status; - return UBRING_ERR; - } - if (g_timer_fd_ctx_map[fd].cb == nullptr) { - LOG(ERROR) << "The callback is not set."; - return UBRING_ERR; - } - - return UBRING_OK; -} -#if defined(OS_MACOSX) -static int timerfd_create_macosx(int clockid, int flags) { - int pipefd[2]; - if (pipe(pipefd) == -1) { - return -1; - } - return pipefd[0]; -} -static int timerfd_settime_macosx(int fd, int flags, - const itimerspec *new_value, - itimerspec *old_value) { - if (old_value != nullptr) { - memset(old_value, 0, sizeof(itimerspec)); - } - return 0; -} -#endif } // namespace ubring } // namespace brpc \ No newline at end of file diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index 1b42caef04..cb0ac856c9 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -18,56 +18,42 @@ #ifndef BRPC_TIMER_MGR_H #define BRPC_TIMER_MGR_H #include -#include +#include +#include +#include +#include +#include "bthread/types.h" +#include "bthread/unstable.h" #include "brpc/ubshm/common/common.h" -#if defined(OS_LINUX) -#include -#include -#elif defined(OS_MACOSX) -#include -#include -#include -#endif - -#define MAX_TIMER 1024 -#define TIMER_EPOLL_WAIT_TIMEOUT 1000 - -#if defined(OS_MACOSX) -struct itimerspec -{ - struct timespec it_interval; - struct timespec it_value; -}; -#endif namespace brpc { namespace ubring { -typedef enum { - TIMER_CONTEXT_NOT_USING, - TIMER_CONTEXT_EPOLL_WAITING, - TIMER_CONTEXT_CALLBACK_ONGOING -} TimerFdCtxStatus; -typedef struct { - void *(*cb)(void*); +typedef void * (*TimerCallback)(void *); + +struct TimerContext{ + TimerCallback cb; void *args; - uint32_t fd; - TimerFdCtxStatus status; uint32_t periodical; + timespec interval; + bthread_timer_t timer_id; pthread_spinlock_t spin_lock; -} TimerFdCtx; +}; + +extern std::unordered_map g_timer_ctx_map; +extern std::mutex g_timer_ctx_mutex; +extern std::atomic g_total_timer_num; -RETURN_CODE TimerInit(void); + +int TimerInit(void); void TimerModuleDestroy(void); -void *UnifiedCallback(void *args); -void *TimerEpoll(void *args); -int32_t TimerStart(const itimerspec *time, void *(*cb)(void *), void *args); +int32_t TimerStart(const itimerspec *time, TimerCallback cb, void *args); uint32_t GetActiveTimerNum(void); -void CloseTimerFd(int fd); -void DeleteTimerSafe(uint32_t fd); -void DeleteTimer(uint32_t fd); -RETURN_CODE TimerFdCtxValidate(uint32_t fd); +void DeleteTimerSafe(uint64_t timer_id); +void DeleteTimer(uint64_t timer_id); + +void TimerCallbackWrapper(void *arg); } } #endif //BRPC_TIMER_MGR_H \ No newline at end of file diff --git a/src/brpc/ubshm/ub_endpoint.cpp b/src/brpc/ubshm/ub_endpoint.cpp index 31539fda85..f6e0ef7ba0 100644 --- a/src/brpc/ubshm/ub_endpoint.cpp +++ b/src/brpc/ubshm/ub_endpoint.cpp @@ -848,6 +848,7 @@ int UBShmEndpoint::PollingModeInitialize(bthread_tag_t tag, while (running->load(std::memory_order_relaxed)) { while (poller->op_queue.Dequeue(op)) { if (op.type == PollerSidOp::ADD) { + poller_sids.erase(op); poller_sids.emplace(op); } else if (op.type == PollerSidOp::REMOVE) { poller_sids.erase(op); From 72294b6c10ad2c663108598b223e6e0b170ef3cc Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Tue, 25 Aug 2026 18:02:27 +0800 Subject: [PATCH 06/15] =?UTF-8?q?[fix]1=E3=80=81=E5=B0=86=E5=8E=9F?= =?UTF-8?q?=E6=9C=89=E7=9A=84=E8=AE=A1=E6=97=B6=E5=99=A8=E6=9B=BF=E6=8D=A2?= =?UTF-8?q?=E4=B8=BAbrpc=E7=9A=84bthread=EF=BC=8C=E4=BB=A5=E6=AD=A4?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=88=A0=E9=99=A4=E8=AE=A1=E6=97=B6?= =?UTF-8?q?=E5=99=A8=E5=8F=AF=E8=83=BD=E4=BC=9A=E6=97=A0=E9=99=90=E6=9C=9F?= =?UTF-8?q?=E9=98=BB=E5=A1=9E=E3=80=81fd=E9=87=8D=E7=94=A8=E7=AB=9E?= =?UTF-8?q?=E9=80=9F=E3=80=81=E8=AE=A1=E6=97=B6=E5=99=A8=E5=88=9D=E5=A7=8B?= =?UTF-8?q?=E5=8C=96=E5=8C=85=E5=90=AB=E4=B8=80=E4=B8=AA=E7=AB=9E=E9=80=89?= =?UTF-8?q?=E3=80=81=E9=BB=98=E8=AE=A4=E7=9A=84=E9=97=AD=E9=97=A8=E5=AE=9A?= =?UTF-8?q?=E6=97=B6=E5=99=A8=E8=BF=87=E4=BA=8E=E9=A2=91=E7=B9=81=E3=80=81?= =?UTF-8?q?=E6=89=80=E6=9C=89=E5=AE=9A=E6=97=B6=E5=99=A8=E5=9B=9E=E8=B0=83?= =?UTF-8?q?=E9=83=BD=E5=9C=A8=E5=90=8C=E4=B8=80P=E7=BA=BF=E7=A8=8B?= =?UTF-8?q?=E4=B8=8A=E4=B8=B2=E8=A1=8C=E3=80=81macOS=E6=A8=A1=E6=8B=9F?= =?UTF-8?q?=E5=9C=A8=E5=8A=9F=E8=83=BD=E4=B8=8A=E5=B9=B6=E4=B8=8D=E7=AD=89?= =?UTF-8?q?=E6=95=88=E3=80=822=E3=80=81=E7=A7=BB=E9=99=A4=E5=9B=BA?= =?UTF-8?q?=E5=AE=9A=E6=95=B0=E7=BB=84=EF=BC=8C=E6=94=B9=E7=94=A8unordered?= =?UTF-8?q?=5Fmap=EF=BC=9B3=E3=80=81=E4=B8=BAitimerspec=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E5=BD=92=E4=B8=80=E5=8C=96=EF=BC=9B4=E3=80=81=E8=A7=A3?= =?UTF-8?q?=E5=86=B3CqSidOp=E7=9A=84sid=E7=9B=B8=E5=90=8C=E8=80=8Cevent?= =?UTF-8?q?=E4=B8=8D=E5=90=8C=E6=97=B6=E6=B7=BB=E5=8A=A0=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.h | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index cb0ac856c9..088cde00b9 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -26,6 +26,14 @@ #include "bthread/unstable.h" #include "brpc/ubshm/common/common.h" +#if defined(OS_MACOSX) +struct itimerspec +{ + struct timespec it_interval; + struct timespec it_value; +}; +#endif + namespace brpc { namespace ubring { From 02adaf772cfee53ee79676653a8321ba0e91ae36 Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Thu, 27 Aug 2026 09:52:10 +0800 Subject: [PATCH 07/15] =?UTF-8?q?[fix]=E4=BF=AE=E5=A4=8D=E9=AD=94=E9=AC=BC?= =?UTF-8?q?=E6=95=B0=E5=AD=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.cpp | 12 ++++++------ src/brpc/ubshm/timer/timer_mgr.h | 2 ++ 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index e60bb02cd4..cc17af7b8e 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -34,13 +34,13 @@ std::atomic g_total_timer_num; static std::atomic g_timer_id_counter(0); static void normalize_timespec(itimerspec *spec) { - if (spec->it_interval.tv_nsec >= 1000000000L) { - spec->it_interval.tv_sec += spec->it_interval.tv_nsec / 1000000000L; - spec->it_interval.tv_nsec %= 1000000000L; + if (spec->it_interval.tv_nsec >= NS_PER_SEC) { + spec->it_interval.tv_sec += spec->it_interval.tv_nsec / NS_PER_SEC; + spec->it_interval.tv_nsec %= NS_PER_SEC; } - if (spec->it_value.tv_nsec >= 1000000000L) { - spec->it_value.tv_sec += spec->it_value.tv_nsec / 1000000000L; - spec->it_value.tv_nsec %= 1000000000L; + if (spec->it_value.tv_nsec >= NS_PER_SEC) { + spec->it_value.tv_sec += spec->it_value.tv_nsec / NS_PER_SEC; + spec->it_value.tv_nsec %= NS_PER_SEC; } } diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index 088cde00b9..10808ee1c2 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -37,6 +37,8 @@ struct itimerspec namespace brpc { namespace ubring { +constexpr long long NS_PER_SEC = 1000000000LL; + typedef void * (*TimerCallback)(void *); struct TimerContext{ From 3bf7a9556ed1828577a1e7aa18e1abde244e7bab Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Thu, 27 Aug 2026 15:56:07 +0800 Subject: [PATCH 08/15] =?UTF-8?q?[fix]callback/delete=E7=94=9F=E5=91=BD?= =?UTF-8?q?=E5=91=A8=E6=9C=9F=E7=AB=9E=E6=80=81=E3=80=81del=E8=BF=94?= =?UTF-8?q?=E5=9B=9E1=E6=97=B6=E6=AD=A3=E7=A1=AE=E5=A4=84=E7=90=86?= =?UTF-8?q?=E3=80=81add=E4=B8=AD=E7=9A=84abstime=E6=AD=A3=E7=A1=AE?= =?UTF-8?q?=E8=B5=8B=E5=80=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.cpp | 167 +++++++++++++++-------------- src/brpc/ubshm/timer/timer_mgr.h | 7 +- 2 files changed, 91 insertions(+), 83 deletions(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index cc17af7b8e..d1ad0b8b54 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -21,97 +21,125 @@ #include #include #include +#include #include +#include #include "brpc/ubshm/timer/timer_mgr.h" namespace brpc { namespace ubring { - -std::unordered_map g_timer_ctx_map; +std::unordered_map > g_timer_ctx_map; std::mutex g_timer_ctx_mutex; -std::atomic g_total_timer_num; +std::atomic g_total_timer_num; -static std::atomic g_timer_id_counter(0); +static std::atomic g_timer_id_counter(1); -static void normalize_timespec(itimerspec *spec) { - if (spec->it_interval.tv_nsec >= NS_PER_SEC) { - spec->it_interval.tv_sec += spec->it_interval.tv_nsec / NS_PER_SEC; - spec->it_interval.tv_nsec %= NS_PER_SEC; - } - if (spec->it_value.tv_nsec >= NS_PER_SEC) { - spec->it_value.tv_sec += spec->it_value.tv_nsec / NS_PER_SEC; - spec->it_value.tv_nsec %= NS_PER_SEC; +static timespec get_current_realtime() { + timespec ts{}; + clock_gettime(CLOCK_REALTIME, &ts); + return ts; +} + +static timespec add_timespec(const timespec &base, const timespec &offset) { + timespec result{}; + result.tv_sec = base.tv_sec + offset.tv_sec; + result.tv_nsec = base.tv_nsec + offset.tv_nsec; + if (result.tv_nsec >= NS_PER_SEC) { + result.tv_sec += result.tv_nsec / NS_PER_SEC; + result.tv_nsec %= NS_PER_SEC; } + return result; } +static std::shared_ptr find_context(uint64_t timer_id) { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + return nullptr; + } + return it->second; +} int TimerInit() { return 0; } void TimerModuleDestroy() { - std::lock_guard lock(g_timer_ctx_mutex); - for (auto & pair: g_timer_ctx_map) { - bthread_timer_del(pair.second.timer_id); - pthread_spin_destroy(&pair.second.spin_lock); + std::vector > contexts; + contexts.reserve(g_timer_ctx_map.size()); + + { + std::lock_guard lock(g_timer_ctx_mutex); + for (auto &pair: g_timer_ctx_map) { + pair.second->periodical = 0; + bthread_timer_del(pair.second->timer_id); + contexts.push_back(pair.second); + } + g_timer_ctx_map.clear(); + g_total_timer_num.store(0); + } + + for (auto &ctx: contexts) { + ctx->self_ref.reset(); } - g_timer_ctx_map.clear(); - g_total_timer_num.store(0); } int32_t TimerStart(const itimerspec *time, TimerCallback cb, void *args) { - if (cb==nullptr) { - // LOG(ERROR) << "Timer callback is nullptr"; + if (cb == nullptr) { + LOG(ERROR) << "Timer callback is nullptr"; return -1; } - TimerContext ctx{}; - ctx.cb = cb; - ctx.args = args; - ctx.periodical = (time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0) ? 1 : 0; - ctx.interval = time->it_interval; - pthread_spin_init(&ctx.spin_lock, PTHREAD_PROCESS_PRIVATE); + auto ctx = std::make_shared(); + ctx->cb = cb; + ctx->args = args; + ctx->periodical = (time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0) ? 1 : 0; + ctx->interval = time->it_interval; + ctx->self_ref = ctx; uint64_t timer_id = g_timer_id_counter.fetch_add(1); - itimerspec normalized_time = *time; - normalize_timespec(&normalized_time); - timespec abstime = normalized_time.it_value; - - int ret = bthread_timer_add(&ctx.timer_id, abstime, TimerCallbackWrapper, reinterpret_cast(timer_id)); - if (ret != 0) { - // LOG(ERROR) << "Failed to add bthread timer, ret=" << ret; - pthread_spin_destroy(&ctx.spin_lock); - return -1; - } + timespec abstime = add_timespec(get_current_realtime(), time->it_value); { std::lock_guard lock(g_timer_ctx_mutex); g_timer_ctx_map[timer_id] = ctx; + ++g_total_timer_num; + int ret = bthread_timer_add(&ctx->timer_id, abstime, TimerCallbackWrapper, reinterpret_cast(timer_id)); + if (ret != 0) { + LOG(ERROR) << "Failed to add bthread timer, ret=" << ret; + g_timer_ctx_map.erase(timer_id); + --g_total_timer_num; + return -1; + } } - std::atomic_fetch_add(&g_total_timer_num, 1U); return static_cast(timer_id); } uint32_t GetActiveTimerNum() { - return std::atomic_load(&g_total_timer_num); + return g_total_timer_num.load(); } void DeleteTimerSafe(uint64_t timer_id) { + int ret = 0; { std::lock_guard lock(g_timer_ctx_mutex); auto it = g_timer_ctx_map.find(timer_id); if (it == g_timer_ctx_map.end()) { return; } - - bthread_timer_del(it->second.timer_id); - - pthread_spin_destroy(&it->second.spin_lock); - g_timer_ctx_map.erase(it); + auto ctx = it->second; + ctx->periodical = 0; + ret = bthread_timer_del(ctx->timer_id); + if (ret == 0) { + g_timer_ctx_map.erase(it); + ctx->self_ref.reset(); + } + } + if (ret == 0) { + --g_total_timer_num; } - std::atomic_fetch_sub(&g_total_timer_num, 1U); } void DeleteTimer(uint64_t timer_id) { @@ -121,60 +149,39 @@ void DeleteTimer(uint64_t timer_id) { LOG(WARNING) << "Timer id=" << timer_id << " not found"; return; } - it->second.periodical = 0; + it->second->periodical = 0; } void TimerCallbackWrapper(void *arg) { auto timer_id = reinterpret_cast(arg); - - TimerContext *ctx = nullptr; - { - std::lock_guard lock(g_timer_ctx_mutex); - auto it = g_timer_ctx_map.find(timer_id); - if (it==g_timer_ctx_map.end()) { - return; - } - ctx = &it->second; + auto ctx = find_context(timer_id); + if (ctx == nullptr) { + LOG(ERROR) << "timer_id is not found, timer_id=" << timer_id; + return; } - - if (ctx == nullptr || ctx->cb == nullptr) { + if (ctx->cb == nullptr) { + LOG(ERROR) << "Timer callback is nullptr"; return; } void *cb_args = ctx->args; - auto is_periodical = ctx->periodical; timespec interval = ctx->interval; - if (!is_periodical) { + if (ctx->periodical == 0) { { std::lock_guard lock(g_timer_ctx_mutex); - auto it = g_timer_ctx_map.find(timer_id); - if (it!=g_timer_ctx_map.end()) { - pthread_spin_destroy(&it->second.spin_lock); - g_timer_ctx_map.erase(it); - } + g_timer_ctx_map.erase(timer_id); + --g_total_timer_num; } - std::atomic_fetch_sub(&g_total_timer_num, 1U); + ctx->self_ref.reset(); } ctx->cb(cb_args); - if (is_periodical) { - auto now = std::chrono::steady_clock::now(); - auto interval_duration = std::chrono::seconds(interval.tv_sec) + std::chrono::nanoseconds(interval.tv_nsec); - auto future_time = now + interval_duration; - - auto future_time_point = std::chrono::time_point_cast(future_time); - timespec abstime{}; - abstime.tv_sec=std::chrono::duration_cast(future_time_point.time_since_epoch()).count(); - abstime.tv_nsec = future_time_point.time_since_epoch().count() % 1000000000; - + if (ctx->periodical == 1) { + timespec abstime = add_timespec(get_current_realtime(), interval); bthread_timer_add(&ctx->timer_id, abstime, TimerCallbackWrapper, reinterpret_cast(timer_id)); } } - - - - -} // namespace ubring -} // namespace brpc \ No newline at end of file +} // namespace ubring +} // namespace brpc diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index 10808ee1c2..4f1236c79d 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -21,6 +21,7 @@ #include #include #include +#include #include #include "bthread/types.h" #include "bthread/unstable.h" @@ -47,12 +48,12 @@ struct TimerContext{ uint32_t periodical; timespec interval; bthread_timer_t timer_id; - pthread_spinlock_t spin_lock; + std::shared_ptr self_ref; }; -extern std::unordered_map g_timer_ctx_map; +extern std::unordered_map> g_timer_ctx_map; extern std::mutex g_timer_ctx_mutex; -extern std::atomic g_total_timer_num; +extern std::atomic g_total_timer_num; int TimerInit(void); From fc55e43f9c1aecc11fdc2c6b1016ffdd47bb571f Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Thu, 27 Aug 2026 21:12:16 +0800 Subject: [PATCH 09/15] =?UTF-8?q?[fix]timer=5Fmgr.h=20=E2=80=94=20?= =?UTF-8?q?=E9=87=8D=E6=9E=84=20TimerContext=EF=BC=9A=E5=88=A0=E9=99=A4=20?= =?UTF-8?q?self=5Fref=E3=80=81=20periodical=20=E6=94=B9=E4=B8=BA=20bool?= =?UTF-8?q?=EF=BC=8C=E4=B8=94=E6=9E=84=E9=80=A0=E5=90=8E=E4=B8=8D=E5=8F=AF?= =?UTF-8?q?=E5=8F=98=E3=80=81=E6=96=B0=E5=A2=9E=E5=90=8C=E6=AD=A5=E5=AD=97?= =?UTF-8?q?=E6=AE=B5=EF=BC=9Astd::mutex=20mtx=E3=80=81std::condition=5Fvar?= =?UTF-8?q?iable=20cv=E3=80=81bool=20stopped=EF=BC=88=E7=A1=AC=E5=88=A0?= =?UTF-8?q?=E9=99=A4=EF=BC=89=E3=80=81bool=20no=5Freschedule=EF=BC=88?= =?UTF-8?q?=E8=BD=AF=E5=88=A0=E9=99=A4=EF=BC=89=E3=80=81int=20running?= =?UTF-8?q?=EF=BC=88=E5=9C=A8=E9=80=94=E5=9B=9E=E8=B0=83=E8=AE=A1=E6=95=B0?= =?UTF-8?q?=EF=BC=89=20timer=5Fmgr.cpp=20=E2=80=94=20=E9=87=8D=E5=86=99?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E9=80=BB=E8=BE=91=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.cpp | 132 ++++++++++++++++++++--------- src/brpc/ubshm/timer/timer_mgr.h | 14 ++- 2 files changed, 103 insertions(+), 43 deletions(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index d1ad0b8b54..0755667b09 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -24,6 +24,7 @@ #include #include #include +#include "bthread/bthread.h" #include "brpc/ubshm/timer/timer_mgr.h" namespace brpc { @@ -60,19 +61,65 @@ static std::shared_ptr find_context(uint64_t timer_id) { return it->second; } +static void remove_timer_from_map(uint64_t timer_id) { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + return; + } + g_timer_ctx_map.erase(it); + --g_total_timer_num; +} + +struct TimerCallbackArgs { + std::shared_ptr ctx; + uint64_t timer_id; +}; + +static void RunTimerCallback(std::shared_ptr ctx, uint64_t timer_id) { + if (ctx->cb != nullptr) { + ctx->cb(ctx->args); + } + + bool need_remove = true; + { + std::lock_guard lock(ctx->mtx); + if (ctx->periodical && !ctx->stopped && !ctx->no_reschedule) { + timespec abstime = add_timespec(get_current_realtime(), ctx->interval); + if (bthread_timer_add(&ctx->timer_id, abstime, TimerCallbackWrapper, + reinterpret_cast(timer_id)) == 0) { + need_remove = false; + } + } + } + + if (need_remove) { + remove_timer_from_map(timer_id); + } + + { + std::lock_guard lock(ctx->mtx); + --ctx->running; + } + ctx->cv.notify_all(); +} + +static void *TimerCallbackWorker(void *arg) { + std::unique_ptr holder(static_cast(arg)); + RunTimerCallback(holder->ctx, holder->timer_id); + return nullptr; +} + int TimerInit() { return 0; } void TimerModuleDestroy() { std::vector > contexts; - contexts.reserve(g_timer_ctx_map.size()); - { std::lock_guard lock(g_timer_ctx_mutex); + contexts.reserve(g_timer_ctx_map.size()); for (auto &pair: g_timer_ctx_map) { - pair.second->periodical = 0; - bthread_timer_del(pair.second->timer_id); contexts.push_back(pair.second); } g_timer_ctx_map.clear(); @@ -80,7 +127,13 @@ void TimerModuleDestroy() { } for (auto &ctx: contexts) { - ctx->self_ref.reset(); + { + std::lock_guard lock(ctx->mtx); + ctx->stopped = true; + bthread_timer_del(ctx->timer_id); + } + std::unique_lock lock(ctx->mtx); + ctx->cv.wait(lock, [&ctx] { return ctx->running == 0; }); } } @@ -93,9 +146,8 @@ int32_t TimerStart(const itimerspec *time, TimerCallback cb, void *args) { auto ctx = std::make_shared(); ctx->cb = cb; ctx->args = args; - ctx->periodical = (time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0) ? 1 : 0; + ctx->periodical = (time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0); ctx->interval = time->it_interval; - ctx->self_ref = ctx; uint64_t timer_id = g_timer_id_counter.fetch_add(1); @@ -122,34 +174,40 @@ uint32_t GetActiveTimerNum() { } void DeleteTimerSafe(uint64_t timer_id) { - int ret = 0; + std::shared_ptr ctx; { std::lock_guard lock(g_timer_ctx_mutex); auto it = g_timer_ctx_map.find(timer_id); if (it == g_timer_ctx_map.end()) { return; } - auto ctx = it->second; - ctx->periodical = 0; - ret = bthread_timer_del(ctx->timer_id); - if (ret == 0) { - g_timer_ctx_map.erase(it); - ctx->self_ref.reset(); - } + ctx = it->second; } - if (ret == 0) { - --g_total_timer_num; + + { + std::lock_guard lock(ctx->mtx); + ctx->stopped = true; + bthread_timer_del(ctx->timer_id); } + std::unique_lock lock(ctx->mtx); + ctx->cv.wait(lock, [&ctx] { return ctx->running == 0; }); + + remove_timer_from_map(timer_id); } void DeleteTimer(uint64_t timer_id) { - std::lock_guard lock(g_timer_ctx_mutex); - auto it = g_timer_ctx_map.find(timer_id); - if (it == g_timer_ctx_map.end()) { - LOG(WARNING) << "Timer id=" << timer_id << " not found"; - return; + std::shared_ptr ctx; + { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + LOG(WARNING) << "Timer id=" << timer_id << " not found"; + return; + } + ctx = it->second; } - it->second->periodical = 0; + std::lock_guard lock(ctx->mtx); + ctx->no_reschedule = true; } void TimerCallbackWrapper(void *arg) { @@ -159,28 +217,20 @@ void TimerCallbackWrapper(void *arg) { LOG(ERROR) << "timer_id is not found, timer_id=" << timer_id; return; } - if (ctx->cb == nullptr) { - LOG(ERROR) << "Timer callback is nullptr"; - return; - } - - void *cb_args = ctx->args; - timespec interval = ctx->interval; - if (ctx->periodical == 0) { - { - std::lock_guard lock(g_timer_ctx_mutex); - g_timer_ctx_map.erase(timer_id); - --g_total_timer_num; + { + std::lock_guard lock(ctx->mtx); + if (ctx->stopped) { + return; } - ctx->self_ref.reset(); + ++ctx->running; } - ctx->cb(cb_args); - - if (ctx->periodical == 1) { - timespec abstime = add_timespec(get_current_realtime(), interval); - bthread_timer_add(&ctx->timer_id, abstime, TimerCallbackWrapper, reinterpret_cast(timer_id)); + auto *holder = new TimerCallbackArgs{ctx, timer_id}; + bthread_t tid; + if (bthread_start_background(&tid, nullptr, TimerCallbackWorker, holder) != 0) { + delete holder; + RunTimerCallback(ctx, timer_id); } } } // namespace ubring diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index 4f1236c79d..8f335b00c1 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -23,6 +23,7 @@ #include #include #include +#include #include "bthread/types.h" #include "bthread/unstable.h" #include "brpc/ubshm/common/common.h" @@ -43,12 +44,21 @@ constexpr long long NS_PER_SEC = 1000000000LL; typedef void * (*TimerCallback)(void *); struct TimerContext{ + TimerContext() + : cb(nullptr), args(nullptr), periodical(false), + interval(), timer_id(0), stopped(false), no_reschedule(false), + running(0) {} + TimerCallback cb; void *args; - uint32_t periodical; + bool periodical; timespec interval; bthread_timer_t timer_id; - std::shared_ptr self_ref; + std::mutex mtx; + std::condition_variable cv; + bool stopped; + bool no_reschedule; + int running; }; extern std::unordered_map> g_timer_ctx_map; From a18faa9be8bec8c72587c73d4b0821ab44d2a80b Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Fri, 28 Aug 2026 10:50:23 +0800 Subject: [PATCH 10/15] =?UTF-8?q?[fix]=E5=9C=A8=E8=87=AA=E5=B7=B1=E7=9A=84?= =?UTF-8?q?=E5=9B=9E=E8=B0=83=E9=87=8C=E7=A1=AC=E5=88=A0=E8=87=AA=E5=B7=B1?= =?UTF-8?q?=E7=9A=84=E5=AE=9A=E6=97=B6=E5=99=A8=E9=80=A0=E6=88=90=E6=AD=BB?= =?UTF-8?q?=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.cpp | 44 ++++++++++++++++++++++-------- src/brpc/ubshm/timer/timer_mgr.h | 10 ++++--- 2 files changed, 38 insertions(+), 16 deletions(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index 0755667b09..021689b39a 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -23,6 +23,7 @@ #include #include #include +#include #include #include "bthread/bthread.h" #include "brpc/ubshm/timer/timer_mgr.h" @@ -77,13 +78,18 @@ struct TimerCallbackArgs { }; static void RunTimerCallback(std::shared_ptr ctx, uint64_t timer_id) { + { + std::lock_guard lock(ctx->mtx); + ctx->worker_tid = bthread_self(); + } + if (ctx->cb != nullptr) { ctx->cb(ctx->args); } bool need_remove = true; { - std::lock_guard lock(ctx->mtx); + std::lock_guard lock(ctx->mtx); if (ctx->periodical && !ctx->stopped && !ctx->no_reschedule) { timespec abstime = add_timespec(get_current_realtime(), ctx->interval); if (bthread_timer_add(&ctx->timer_id, abstime, TimerCallbackWrapper, @@ -98,8 +104,9 @@ static void RunTimerCallback(std::shared_ptr ctx, uint64_t timer_i } { - std::lock_guard lock(ctx->mtx); + std::lock_guard lock(ctx->mtx); --ctx->running; + ctx->worker_tid = 0; } ctx->cv.notify_all(); } @@ -122,19 +129,22 @@ void TimerModuleDestroy() { for (auto &pair: g_timer_ctx_map) { contexts.push_back(pair.second); } - g_timer_ctx_map.clear(); - g_total_timer_num.store(0); } - for (auto &ctx: contexts) { { - std::lock_guard lock(ctx->mtx); + std::lock_guard lock(ctx->mtx); ctx->stopped = true; bthread_timer_del(ctx->timer_id); } - std::unique_lock lock(ctx->mtx); + std::unique_lock lock(ctx->mtx); ctx->cv.wait(lock, [&ctx] { return ctx->running == 0; }); } + + { + std::lock_guard lock(g_timer_ctx_mutex); + g_timer_ctx_map.clear(); + g_total_timer_num.store(0); + } } int32_t TimerStart(const itimerspec *time, TimerCallback cb, void *args) { @@ -184,12 +194,18 @@ void DeleteTimerSafe(uint64_t timer_id) { ctx = it->second; } + bool self = false; { - std::lock_guard lock(ctx->mtx); + std::lock_guard lock(ctx->mtx); ctx->stopped = true; bthread_timer_del(ctx->timer_id); + self = (bthread_self() != 0 && bthread_self() == ctx->worker_tid); + } + if (self) { + return; } - std::unique_lock lock(ctx->mtx); + + std::unique_lock lock(ctx->mtx); ctx->cv.wait(lock, [&ctx] { return ctx->running == 0; }); remove_timer_from_map(timer_id); @@ -206,7 +222,7 @@ void DeleteTimer(uint64_t timer_id) { } ctx = it->second; } - std::lock_guard lock(ctx->mtx); + std::lock_guard lock(ctx->mtx); ctx->no_reschedule = true; } @@ -219,14 +235,18 @@ void TimerCallbackWrapper(void *arg) { } { - std::lock_guard lock(ctx->mtx); + std::lock_guard lock(ctx->mtx); if (ctx->stopped) { return; } ++ctx->running; } - auto *holder = new TimerCallbackArgs{ctx, timer_id}; + auto *holder = new (std::nothrow) TimerCallbackArgs{ctx, timer_id}; + if (holder == nullptr) { + RunTimerCallback(ctx, timer_id); + return; + } bthread_t tid; if (bthread_start_background(&tid, nullptr, TimerCallbackWorker, holder) != 0) { delete holder; diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index 8f335b00c1..474594149f 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -23,9 +23,10 @@ #include #include #include -#include #include "bthread/types.h" #include "bthread/unstable.h" +#include "bthread/mutex.h" +#include "bthread/condition_variable.h" #include "brpc/ubshm/common/common.h" #if defined(OS_MACOSX) @@ -47,18 +48,19 @@ struct TimerContext{ TimerContext() : cb(nullptr), args(nullptr), periodical(false), interval(), timer_id(0), stopped(false), no_reschedule(false), - running(0) {} + running(0), worker_tid(0) {} TimerCallback cb; void *args; bool periodical; timespec interval; bthread_timer_t timer_id; - std::mutex mtx; - std::condition_variable cv; + bthread::Mutex mtx; + bthread::ConditionVariable cv; bool stopped; bool no_reschedule; int running; + bthread_t worker_tid; }; extern std::unordered_map> g_timer_ctx_map; From cc0e52220938920d74af5e4111a354b115e54b87 Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Fri, 28 Aug 2026 14:35:16 +0800 Subject: [PATCH 11/15] =?UTF-8?q?[fix]DeleteTimerSafe=E9=9D=9E=E9=98=BB?= =?UTF-8?q?=E5=A1=9E=E9=98=B2=E6=AD=A2=E6=AD=BB=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.cpp | 54 ++++++++++++------------------ src/brpc/ubshm/timer/timer_mgr.h | 10 ++---- 2 files changed, 23 insertions(+), 41 deletions(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index 021689b39a..9563910f8b 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -25,7 +25,7 @@ #include #include #include -#include "bthread/bthread.h" +#include "bthread/bthread.h" // bthread_start_background #include "brpc/ubshm/timer/timer_mgr.h" namespace brpc { @@ -78,18 +78,13 @@ struct TimerCallbackArgs { }; static void RunTimerCallback(std::shared_ptr ctx, uint64_t timer_id) { - { - std::lock_guard lock(ctx->mtx); - ctx->worker_tid = bthread_self(); - } - if (ctx->cb != nullptr) { ctx->cb(ctx->args); } bool need_remove = true; { - std::lock_guard lock(ctx->mtx); + std::lock_guard lock(ctx->mtx); if (ctx->periodical && !ctx->stopped && !ctx->no_reschedule) { timespec abstime = add_timespec(get_current_realtime(), ctx->interval); if (bthread_timer_add(&ctx->timer_id, abstime, TimerCallbackWrapper, @@ -102,13 +97,6 @@ static void RunTimerCallback(std::shared_ptr ctx, uint64_t timer_i if (need_remove) { remove_timer_from_map(timer_id); } - - { - std::lock_guard lock(ctx->mtx); - --ctx->running; - ctx->worker_tid = 0; - } - ctx->cv.notify_all(); } static void *TimerCallbackWorker(void *arg) { @@ -130,14 +118,11 @@ void TimerModuleDestroy() { contexts.push_back(pair.second); } } + for (auto &ctx: contexts) { - { - std::lock_guard lock(ctx->mtx); - ctx->stopped = true; - bthread_timer_del(ctx->timer_id); - } - std::unique_lock lock(ctx->mtx); - ctx->cv.wait(lock, [&ctx] { return ctx->running == 0; }); + std::lock_guard lock(ctx->mtx); + ctx->stopped = true; + bthread_timer_del(ctx->timer_id); } { @@ -183,34 +168,35 @@ uint32_t GetActiveTimerNum() { return g_total_timer_num.load(); } +// Hard delete: stop the timer so it never fires again. Non-blocking — it does +// NOT wait for an in-flight callback; the caller is responsible for keeping +// `args` alive until any already-running callback has returned. void DeleteTimerSafe(uint64_t timer_id) { std::shared_ptr ctx; { std::lock_guard lock(g_timer_ctx_mutex); auto it = g_timer_ctx_map.find(timer_id); if (it == g_timer_ctx_map.end()) { + LOG(WARNING) << "Timer id=" << timer_id << " not found"; return; } ctx = it->second; } - bool self = false; { - std::lock_guard lock(ctx->mtx); + // Mark stopped and unschedule under ctx->mtx so this serializes with a + // concurrent reschedule (RunTimerCallback). After this scope no new + // firing can start a callback. + std::lock_guard lock(ctx->mtx); ctx->stopped = true; bthread_timer_del(ctx->timer_id); - self = (bthread_self() != 0 && bthread_self() == ctx->worker_tid); } - if (self) { - return; - } - - std::unique_lock lock(ctx->mtx); - ctx->cv.wait(lock, [&ctx] { return ctx->running == 0; }); remove_timer_from_map(timer_id); } +// Soft delete: stop future rescheduling but let the pending firing (if any) +// run its callback once more, which then cleans itself up. void DeleteTimer(uint64_t timer_id) { std::shared_ptr ctx; { @@ -222,7 +208,7 @@ void DeleteTimer(uint64_t timer_id) { } ctx = it->second; } - std::lock_guard lock(ctx->mtx); + std::lock_guard lock(ctx->mtx); ctx->no_reschedule = true; } @@ -235,20 +221,22 @@ void TimerCallbackWrapper(void *arg) { } { - std::lock_guard lock(ctx->mtx); + std::lock_guard lock(ctx->mtx); if (ctx->stopped) { + // Hard-deleted while this firing was in flight; don't run. return; } - ++ctx->running; } auto *holder = new (std::nothrow) TimerCallbackArgs{ctx, timer_id}; if (holder == nullptr) { + // Allocation failed (OOM); run inline so the callback is not dropped. RunTimerCallback(ctx, timer_id); return; } bthread_t tid; if (bthread_start_background(&tid, nullptr, TimerCallbackWorker, holder) != 0) { + // Extremely unlikely (ENOMEM); fall back to running inline. delete holder; RunTimerCallback(ctx, timer_id); } diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index 474594149f..f1a1e2fca9 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -25,8 +25,6 @@ #include #include "bthread/types.h" #include "bthread/unstable.h" -#include "bthread/mutex.h" -#include "bthread/condition_variable.h" #include "brpc/ubshm/common/common.h" #if defined(OS_MACOSX) @@ -47,20 +45,16 @@ typedef void * (*TimerCallback)(void *); struct TimerContext{ TimerContext() : cb(nullptr), args(nullptr), periodical(false), - interval(), timer_id(0), stopped(false), no_reschedule(false), - running(0), worker_tid(0) {} + interval(), timer_id(0), stopped(false), no_reschedule(false) {} TimerCallback cb; void *args; bool periodical; timespec interval; bthread_timer_t timer_id; - bthread::Mutex mtx; - bthread::ConditionVariable cv; + std::mutex mtx; bool stopped; bool no_reschedule; - int running; - bthread_t worker_tid; }; extern std::unordered_map> g_timer_ctx_map; From cd6fcf840237eee8a1c20a8dec2b54972df84260 Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Fri, 28 Aug 2026 14:44:53 +0800 Subject: [PATCH 12/15] =?UTF-8?q?[fix]DeleteTimerSafe=E9=9D=9E=E9=98=BB?= =?UTF-8?q?=E5=A1=9E=E9=98=B2=E6=AD=A2=E6=AD=BB=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.cpp | 13 +------------ 1 file changed, 1 insertion(+), 12 deletions(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index 9563910f8b..7246c827ce 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -25,7 +25,7 @@ #include #include #include -#include "bthread/bthread.h" // bthread_start_background +#include "bthread/bthread.h" #include "brpc/ubshm/timer/timer_mgr.h" namespace brpc { @@ -168,9 +168,6 @@ uint32_t GetActiveTimerNum() { return g_total_timer_num.load(); } -// Hard delete: stop the timer so it never fires again. Non-blocking — it does -// NOT wait for an in-flight callback; the caller is responsible for keeping -// `args` alive until any already-running callback has returned. void DeleteTimerSafe(uint64_t timer_id) { std::shared_ptr ctx; { @@ -184,9 +181,6 @@ void DeleteTimerSafe(uint64_t timer_id) { } { - // Mark stopped and unschedule under ctx->mtx so this serializes with a - // concurrent reschedule (RunTimerCallback). After this scope no new - // firing can start a callback. std::lock_guard lock(ctx->mtx); ctx->stopped = true; bthread_timer_del(ctx->timer_id); @@ -195,8 +189,6 @@ void DeleteTimerSafe(uint64_t timer_id) { remove_timer_from_map(timer_id); } -// Soft delete: stop future rescheduling but let the pending firing (if any) -// run its callback once more, which then cleans itself up. void DeleteTimer(uint64_t timer_id) { std::shared_ptr ctx; { @@ -223,20 +215,17 @@ void TimerCallbackWrapper(void *arg) { { std::lock_guard lock(ctx->mtx); if (ctx->stopped) { - // Hard-deleted while this firing was in flight; don't run. return; } } auto *holder = new (std::nothrow) TimerCallbackArgs{ctx, timer_id}; if (holder == nullptr) { - // Allocation failed (OOM); run inline so the callback is not dropped. RunTimerCallback(ctx, timer_id); return; } bthread_t tid; if (bthread_start_background(&tid, nullptr, TimerCallbackWorker, holder) != 0) { - // Extremely unlikely (ENOMEM); fall back to running inline. delete holder; RunTimerCallback(ctx, timer_id); } From 4ff1f1f5dd4263c5c613e0199872e8cc9b927b1e Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Fri, 28 Aug 2026 16:52:41 +0800 Subject: [PATCH 13/15] =?UTF-8?q?[fix]DeleteTimerSafe=E5=90=8E=E4=B8=8D?= =?UTF-8?q?=E8=B0=83=E7=94=A8ctx->cb(ctx->args);?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index 7246c827ce..8b76c2b075 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -78,7 +78,7 @@ struct TimerCallbackArgs { }; static void RunTimerCallback(std::shared_ptr ctx, uint64_t timer_id) { - if (ctx->cb != nullptr) { + if (ctx->cb != nullptr && !ctx->stopped) { ctx->cb(ctx->args); } From 998630240ef650947a1001513685e1dfcef268bd Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Sat, 29 Aug 2026 09:34:27 +0800 Subject: [PATCH 14/15] =?UTF-8?q?[fix]=E6=96=B0=E5=A2=9E=E9=9D=9E=E9=98=BB?= =?UTF-8?q?=E5=A1=9EStopTimer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.cpp | 51 ++++++++++--- src/brpc/ubshm/timer/timer_mgr.h | 12 ++- src/brpc/ubshm/ub_ring.cpp | 114 ++++++++++++++++++++--------- src/brpc/ubshm/ubr_trx.h | 1 + 4 files changed, 130 insertions(+), 48 deletions(-) diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index 8b76c2b075..926fdb9fe1 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -78,13 +78,24 @@ struct TimerCallbackArgs { }; static void RunTimerCallback(std::shared_ptr ctx, uint64_t timer_id) { - if (ctx->cb != nullptr && !ctx->stopped) { + bool should_run_cb = false; + { + BAIDU_SCOPED_LOCK(ctx->mtx); + if (!ctx->stopped&&ctx->cb!=nullptr) { + should_run_cb = true; + } + } + + if (should_run_cb) { ctx->cb(ctx->args); } bool need_remove = true; { - std::lock_guard lock(ctx->mtx); + BAIDU_SCOPED_LOCK(ctx->mtx); + --ctx->running; + ctx->worker_tid = 0; + ctx->cv.notify_all(); if (ctx->periodical && !ctx->stopped && !ctx->no_reschedule) { timespec abstime = add_timespec(get_current_realtime(), ctx->interval); if (bthread_timer_add(&ctx->timer_id, abstime, TimerCallbackWrapper, @@ -120,7 +131,7 @@ void TimerModuleDestroy() { } for (auto &ctx: contexts) { - std::lock_guard lock(ctx->mtx); + BAIDU_SCOPED_LOCK(ctx->mtx); ctx->stopped = true; bthread_timer_del(ctx->timer_id); } @@ -168,6 +179,23 @@ uint32_t GetActiveTimerNum() { return g_total_timer_num.load(); } +void StopTimer(uint64_t timer_id) { + std::shared_ptr ctx; + { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + return; + } + ctx = it->second; + } + + BAIDU_SCOPED_LOCK(ctx->mtx); + ctx->stopped = true; + ctx->no_reschedule = true; + bthread_timer_del(ctx->timer_id); +} + void DeleteTimerSafe(uint64_t timer_id) { std::shared_ptr ctx; { @@ -180,12 +208,16 @@ void DeleteTimerSafe(uint64_t timer_id) { ctx = it->second; } - { - std::lock_guard lock(ctx->mtx); - ctx->stopped = true; - bthread_timer_del(ctx->timer_id); + std::unique_lock lock(ctx->mtx); + ctx->stopped = true; + ctx->no_reschedule = true; + bthread_timer_del(ctx->timer_id); + + while (ctx->running > 0) { + ctx->cv.wait(lock); } + lock.unlock(); remove_timer_from_map(timer_id); } @@ -200,7 +232,7 @@ void DeleteTimer(uint64_t timer_id) { } ctx = it->second; } - std::lock_guard lock(ctx->mtx); + BAIDU_SCOPED_LOCK(ctx->mtx); ctx->no_reschedule = true; } @@ -213,10 +245,11 @@ void TimerCallbackWrapper(void *arg) { } { - std::lock_guard lock(ctx->mtx); + BAIDU_SCOPED_LOCK(ctx->mtx); if (ctx->stopped) { return; } + ++ctx->running; } auto *holder = new (std::nothrow) TimerCallbackArgs{ctx, timer_id}; diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index f1a1e2fca9..70261aba69 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -42,19 +42,22 @@ constexpr long long NS_PER_SEC = 1000000000LL; typedef void * (*TimerCallback)(void *); -struct TimerContext{ +struct TimerContext { TimerContext() - : cb(nullptr), args(nullptr), periodical(false), - interval(), timer_id(0), stopped(false), no_reschedule(false) {} + : cb(nullptr), args(nullptr), periodical(false), interval(), timer_id(0), + stopped(false), no_reschedule(false), running(0), worker_tid(0) {} TimerCallback cb; void *args; bool periodical; timespec interval; bthread_timer_t timer_id; - std::mutex mtx; + bthread::Mutex mtx; + bthread::ConditionVariable cv; bool stopped; bool no_reschedule; + int running; + bthread_t worker_tid; }; extern std::unordered_map> g_timer_ctx_map; @@ -67,6 +70,7 @@ void TimerModuleDestroy(void); int32_t TimerStart(const itimerspec *time, TimerCallback cb, void *args); uint32_t GetActiveTimerNum(void); +void StopTimer(uint64_t timer_id); void DeleteTimerSafe(uint64_t timer_id); void DeleteTimer(uint64_t timer_id); diff --git a/src/brpc/ubshm/ub_ring.cpp b/src/brpc/ubshm/ub_ring.cpp index cdc346edf4..bbeb8889ed 100644 --- a/src/brpc/ubshm/ub_ring.cpp +++ b/src/brpc/ubshm/ub_ring.cpp @@ -19,8 +19,11 @@ #include #include #include +#include #include "bthread/bthread.h" #include "butil/logging.h" +#include "butil/atomicops.h" +#include "butil/synchronization/lock.h" #include "brpc/ubshm/ub_ring.h" #include "brpc/ubshm/ub_ring_manager.h" #include "brpc/ubshm/shm/shm_ipc.h" @@ -218,40 +221,90 @@ RETURN_CODE UBRing::UbrAddHBTimer() { return UBRING_OK; } +static std::vector g_cleanup_queue; +static butil::Mutex g_cleanup_queue_mutex; +static butil::atomic g_cleanup_thread_running; +static bthread_t g_cleanup_thread_tid; +constexpr uint64_t SLEEP_US = 1000000; + +static void* CleanupOwnerThread(void*) { + while (g_cleanup_thread_running.load(butil::memory_order_acquire) == 1) { + std::vector queue_to_process; + { + BAIDU_SCOPED_LOCK(g_cleanup_queue_mutex); + if (g_cleanup_queue.empty()) { + bthread_usleep(SLEEP_US); + continue; + } + queue_to_process.swap(g_cleanup_queue); + } + + for (UbrTrx *trx : queue_to_process) { + if (trx == nullptr) { + continue; + } + + DeleteTimerSafe(trx->timer_fd); + DeleteTimerSafe(trx->hb_timer_fd); + + bthread_usleep(SLEEP_US); + + if (UNLIKELY(UBRing::UbrTrxFreeShm(trx) != UBRING_OK)) { + LOG(ERROR) << "Cleanup owner: free shm failed for " << trx->local_shm.name; + } + + if (UNLIKELY(UBRingManager::ReleaseUbrTrxFromMgr(trx) != UBRING_OK)) { + LOG(ERROR) << "Cleanup owner: release rtx failed for " << trx->local_shm.name; + } + } + } + return nullptr; +} + +static void InitCleanupOwner() { + if (g_cleanup_thread_running.load(butil::memory_order_acquire) != 0) { + return; + } + g_cleanup_thread_running.store(1, butil::memory_order_release); + bthread_t tid; + if (bthread_start_background(&tid,nullptr,CleanupOwnerThread,nullptr) != 0) { + LOG(ERROR) << "Cleanup owner: start background failed."; + g_cleanup_thread_running.store(0, butil::memory_order_release); + }else { + g_cleanup_thread_tid = tid; + } +} + +static void RequestTrxCleanup(UbrTrx *trx) { + int expected = 0; + if (!ATOMIC_COMPARE_EXCHANGE_STRONG(trx->cleanup_pending,expected,1)) { + LOG(ERROR) << "Cleanup already pending for trx, name=" << trx->local_shm.name; + return; + } + + InitCleanupOwner(); + { + BAIDU_SCOPED_LOCK(g_cleanup_queue_mutex); + g_cleanup_queue.push_back(trx); + } +} + RETURN_CODE UBRing::UbrPassiveClearTrx(UbrTrx *trx, int fd, PASSIVE_DISC_TYPE type) { RETURN_CODE passive_close_check_rc = UbrTrxCloseCheck(trx); if (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; - uint64_t start_time = GetCurNanoSeconds(); - return ClearTrxResource(trx, start_time, UBR_CALL_BACK_CLOSE); + return UBRING_OK; } return UBRING_ERR; } trx->ubr_tx.trx_state = UBR_STATE_CLOSED; trx->ubr_rx.trx_state = UBR_STATE_CLOSED; - DeleteTimerSafe((uint32_t)trx->timer_fd); - const char *type_name = nullptr; - if (type == UBR_HEARTBEAT) { - DeleteTimer((uint32_t)trx->hb_timer_fd); - type_name = "Trx heartbeat"; - } else if (type == UBR_UB_EVENT) { - DeleteTimerSafe((uint32_t)trx->hb_timer_fd); - type_name = "Ub event callback"; - } - constexpr int64_t kMicrosecondsPerSecond = 1000000LL; - bthread_usleep(FLAGS_ub_flying_io_timeout_s * kMicrosecondsPerSecond); - - int rc = ShmLocalFree(&trx->remote_shm); - if (rc != UBRING_OK) { - LOG(ERROR) << type_name << ", delete remote shm failed. ret=" << rc; - } - rc = ShmLocalFree(&trx->local_shm); - if (rc != UBRING_OK) { - LOG(ERROR) << type_name << ", delete local shm failed. ret=" << rc; - } - UBRingManager::ReleaseUbrTrxFromMgr(trx); + StopTimer(trx->timer_fd); + StopTimer(trx->hb_timer_fd); + + RequestTrxCleanup(trx); return UBRING_OK; } @@ -1006,12 +1059,8 @@ RETURN_CODE UBRing::UbrClearResourceCheck(UbrTrx *trx, uint64_t start_time, UbrC local_tx_event_q->flag = UBR_STATE_CLOSING; } - if (close_type == UBR_SEND_CLOSE) { - DeleteTimerSafe((uint32_t)trx->timer_fd); - } else { - DeleteTimer((uint32_t)trx->timer_fd); - } - DeleteTimerSafe((uint32_t)trx->hb_timer_fd); + StopTimer(trx->timer_fd); + StopTimer(trx->hb_timer_fd); if (local_tx_event_q->flag == UBR_STATE_CLOSING) { local_tx_event_q->flag = UBR_STATE_CLOSED; @@ -1028,12 +1077,7 @@ RETURN_CODE UBRing::ClearTrxResource(UbrTrx *trx, uint64_t start_time, UbrCloseT return rc; } - rc = UbrAddAsynClearTimer(trx); - if (rc != UBRING_OK) { - LOG(ERROR) << "Trx close, add " << trx->local_shm.name << " close clear timer failed."; - return UBRING_ERR; - } - + RequestTrxCleanup(trx); return UBRING_OK; } diff --git a/src/brpc/ubshm/ubr_trx.h b/src/brpc/ubshm/ubr_trx.h index b4e89645ff..9ff8c4c66b 100644 --- a/src/brpc/ubshm/ubr_trx.h +++ b/src/brpc/ubshm/ubr_trx.h @@ -140,6 +140,7 @@ typedef struct TagUbrTrx { int clear_timer_fd; AtomicInt close_cnt; AtomicInt close_state; + AtomicInt cleanup_pending; } UbrTrx; typedef struct TagFileLock { From 3fdf6fa0b9ba18428c01225c2166aeeb2e4854ef Mon Sep 17 00:00:00 2001 From: Sherlock0203 <1297399478@qq.com> Date: Sat, 29 Aug 2026 09:39:53 +0800 Subject: [PATCH 15/15] =?UTF-8?q?[fix]=E6=96=B0=E5=A2=9E=E9=9D=9E=E9=98=BB?= =?UTF-8?q?=E5=A1=9EStopTimer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/brpc/ubshm/timer/timer_mgr.h | 1 + 1 file changed, 1 insertion(+) diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index 70261aba69..a1314e7511 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -25,6 +25,7 @@ #include #include "bthread/types.h" #include "bthread/unstable.h" +#include "bthread/bthread.h" #include "brpc/ubshm/common/common.h" #if defined(OS_MACOSX)