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 1/4] =?UTF-8?q?[fix]1=E3=80=81=E5=B0=86=E5=8E=9F=E6=9C=89?= =?UTF-8?q?=E7=9A=84=E8=AE=A1=E6=97=B6=E5=99=A8=E6=9B=BF=E6=8D=A2=E4=B8=BA?= =?UTF-8?q?brpc=E7=9A=84bthread=EF=BC=8C=E4=BB=A5=E6=AD=A4=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=EF=BC=9A=E5=88=A0=E9=99=A4=E8=AE=A1=E6=97=B6=E5=99=A8?= =?UTF-8?q?=E5=8F=AF=E8=83=BD=E4=BC=9A=E6=97=A0=E9=99=90=E6=9C=9F=E9=98=BB?= =?UTF-8?q?=E5=A1=9E=E3=80=81fd=E9=87=8D=E7=94=A8=E7=AB=9E=E9=80=9F?= =?UTF-8?q?=E3=80=81=E8=AE=A1=E6=97=B6=E5=99=A8=E5=88=9D=E5=A7=8B=E5=8C=96?= =?UTF-8?q?=E5=8C=85=E5=90=AB=E4=B8=80=E4=B8=AA=E7=AB=9E=E9=80=89=E3=80=81?= =?UTF-8?q?=E9=BB=98=E8=AE=A4=E7=9A=84=E9=97=AD=E9=97=A8=E5=AE=9A=E6=97=B6?= =?UTF-8?q?=E5=99=A8=E8=BF=87=E4=BA=8E=E9=A2=91=E7=B9=81=E3=80=81=E6=89=80?= =?UTF-8?q?=E6=9C=89=E5=AE=9A=E6=97=B6=E5=99=A8=E5=9B=9E=E8=B0=83=E9=83=BD?= =?UTF-8?q?=E5=9C=A8=E5=90=8C=E4=B8=80P=E7=BA=BF=E7=A8=8B=E4=B8=8A?= =?UTF-8?q?=E4=B8=B2=E8=A1=8C=E3=80=81macOS=E6=A8=A1=E6=8B=9F=E5=9C=A8?= =?UTF-8?q?=E5=8A=9F=E8=83=BD=E4=B8=8A=E5=B9=B6=E4=B8=8D=E7=AD=89=E6=95=88?= =?UTF-8?q?=E3=80=822=E3=80=81=E7=A7=BB=E9=99=A4=E5=9B=BA=E5=AE=9A?= =?UTF-8?q?=E6=95=B0=E7=BB=84=EF=BC=8C=E6=94=B9=E7=94=A8unordered=5Fmap?= =?UTF-8?q?=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 2/4] =?UTF-8?q?[fix]1=E3=80=81=E5=B0=86=E5=8E=9F=E6=9C=89?= =?UTF-8?q?=E7=9A=84=E8=AE=A1=E6=97=B6=E5=99=A8=E6=9B=BF=E6=8D=A2=E4=B8=BA?= =?UTF-8?q?brpc=E7=9A=84bthread=EF=BC=8C=E4=BB=A5=E6=AD=A4=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=EF=BC=9A=E5=88=A0=E9=99=A4=E8=AE=A1=E6=97=B6=E5=99=A8?= =?UTF-8?q?=E5=8F=AF=E8=83=BD=E4=BC=9A=E6=97=A0=E9=99=90=E6=9C=9F=E9=98=BB?= =?UTF-8?q?=E5=A1=9E=E3=80=81fd=E9=87=8D=E7=94=A8=E7=AB=9E=E9=80=9F?= =?UTF-8?q?=E3=80=81=E8=AE=A1=E6=97=B6=E5=99=A8=E5=88=9D=E5=A7=8B=E5=8C=96?= =?UTF-8?q?=E5=8C=85=E5=90=AB=E4=B8=80=E4=B8=AA=E7=AB=9E=E9=80=89=E3=80=81?= =?UTF-8?q?=E9=BB=98=E8=AE=A4=E7=9A=84=E9=97=AD=E9=97=A8=E5=AE=9A=E6=97=B6?= =?UTF-8?q?=E5=99=A8=E8=BF=87=E4=BA=8E=E9=A2=91=E7=B9=81=E3=80=81=E6=89=80?= =?UTF-8?q?=E6=9C=89=E5=AE=9A=E6=97=B6=E5=99=A8=E5=9B=9E=E8=B0=83=E9=83=BD?= =?UTF-8?q?=E5=9C=A8=E5=90=8C=E4=B8=80P=E7=BA=BF=E7=A8=8B=E4=B8=8A?= =?UTF-8?q?=E4=B8=B2=E8=A1=8C=E3=80=81macOS=E6=A8=A1=E6=8B=9F=E5=9C=A8?= =?UTF-8?q?=E5=8A=9F=E8=83=BD=E4=B8=8A=E5=B9=B6=E4=B8=8D=E7=AD=89=E6=95=88?= =?UTF-8?q?=E3=80=822=E3=80=81=E7=A7=BB=E9=99=A4=E5=9B=BA=E5=AE=9A?= =?UTF-8?q?=E6=95=B0=E7=BB=84=EF=BC=8C=E6=94=B9=E7=94=A8unordered=5Fmap?= =?UTF-8?q?=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 3/4] =?UTF-8?q?[fix]1=E3=80=81=E5=B0=86=E5=8E=9F=E6=9C=89?= =?UTF-8?q?=E7=9A=84=E8=AE=A1=E6=97=B6=E5=99=A8=E6=9B=BF=E6=8D=A2=E4=B8=BA?= =?UTF-8?q?brpc=E7=9A=84bthread=EF=BC=8C=E4=BB=A5=E6=AD=A4=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=EF=BC=9A=E5=88=A0=E9=99=A4=E8=AE=A1=E6=97=B6=E5=99=A8?= =?UTF-8?q?=E5=8F=AF=E8=83=BD=E4=BC=9A=E6=97=A0=E9=99=90=E6=9C=9F=E9=98=BB?= =?UTF-8?q?=E5=A1=9E=E3=80=81fd=E9=87=8D=E7=94=A8=E7=AB=9E=E9=80=9F?= =?UTF-8?q?=E3=80=81=E8=AE=A1=E6=97=B6=E5=99=A8=E5=88=9D=E5=A7=8B=E5=8C=96?= =?UTF-8?q?=E5=8C=85=E5=90=AB=E4=B8=80=E4=B8=AA=E7=AB=9E=E9=80=89=E3=80=81?= =?UTF-8?q?=E9=BB=98=E8=AE=A4=E7=9A=84=E9=97=AD=E9=97=A8=E5=AE=9A=E6=97=B6?= =?UTF-8?q?=E5=99=A8=E8=BF=87=E4=BA=8E=E9=A2=91=E7=B9=81=E3=80=81=E6=89=80?= =?UTF-8?q?=E6=9C=89=E5=AE=9A=E6=97=B6=E5=99=A8=E5=9B=9E=E8=B0=83=E9=83=BD?= =?UTF-8?q?=E5=9C=A8=E5=90=8C=E4=B8=80P=E7=BA=BF=E7=A8=8B=E4=B8=8A?= =?UTF-8?q?=E4=B8=B2=E8=A1=8C=E3=80=81macOS=E6=A8=A1=E6=8B=9F=E5=9C=A8?= =?UTF-8?q?=E5=8A=9F=E8=83=BD=E4=B8=8A=E5=B9=B6=E4=B8=8D=E7=AD=89=E6=95=88?= =?UTF-8?q?=E3=80=822=E3=80=81=E7=A7=BB=E9=99=A4=E5=9B=BA=E5=AE=9A?= =?UTF-8?q?=E6=95=B0=E7=BB=84=EF=BC=8C=E6=94=B9=E7=94=A8unordered=5Fmap?= =?UTF-8?q?=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 4/4] =?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{