chenBright commented on code in PR #3290: URL: https://github.com/apache/brpc/pull/3290#discussion_r3566618250
########## src/brpc/ubshm/timer/timer_mgr.cpp: ########## @@ -0,0 +1,468 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#define _GNU_SOURCE +#include <pthread.h> +#include <sched.h> +#include <errno.h> +#include <stdio.h> +#include <stdlib.h> +#include <unistd.h> +#include <atomic> +#include <sys/resource.h> +#include "brpc/ubshm/timer/timer_mgr.h" + +namespace brpc { +namespace ubring { + +int32_t g_epoll_fd = -1; +std::atomic<uint32_t> g_total_timer_num(0); +TimerFdCtx *g_timer_fd_ctx_map = NULL; +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 == NULL) { + return UBRING_OK; + } + + if (pthread_spin_lock(&g_timer_fd_ctx_map[fd].spinLock) != 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].spinLock); + return UBRING_OK; + } + + g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; + g_timer_fd_ctx_map[fd].cb = NULL; + g_timer_fd_ctx_map[fd].args = NULL; + 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].spinLock); + +#if defined(OS_LINUX) + epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, (int)fd, NULL); +#elif defined(OS_MACOSX) + struct kevent evt; + EV_SET(&evt, fd, EVFILT_TIMER, EV_DELETE, 0, 0, NULL); + kevent(g_epoll_fd, &evt, 1, NULL, 0, NULL); +#endif + + uint64_t exp = 0; + read((int)fd, &exp, sizeof(exp)); + + close((int)fd); + std::atomic_fetch_sub(&g_total_timer_num, 1); + 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; + } + + int ret = pthread_create(&g_epoll_execute_thread, NULL, TimerEpoll, NULL); + if (UNLIKELY(ret != 0)) { + LOG(ERROR) << "Failed to create thread err=" << ret; + return UBRING_ERR; + } + return UBRING_OK; +} + +static RETURN_CODE TimerSpinLocksInit(void) { + if (g_timer_fd_ctx_map == NULL) { + LOG(ERROR) << "Timer module is not fully initialized."; + return UBRING_ERR; + } + + for (uint32_t fd = 0; fd < g_max_system_fd; fd++) { + int ret = pthread_spin_init(&g_timer_fd_ctx_map[fd].spinLock, + PTHREAD_PROCESS_PRIVATE); + if (ret != EOK) { + LOG(ERROR) << "Failed to initialize spin lock for fd=" << fd; + for (uint32_t cleanupFd = 0; cleanupFd < fd; cleanupFd++) { + pthread_spin_destroy(&g_timer_fd_ctx_map[cleanupFd].spinLock); + } + return UBRING_ERR; + } + } + return UBRING_OK; +} + +static RETURN_CODE ExecuteCallback(int32_t timerFd) { + UnifiedCallback((void *)(&g_timer_fd_ctx_map[timerFd])); + 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; +} + +RETURN_CODE TimerInit(void) { + if (g_timer_module_initialized > 0) { + return UBRING_OK; + } + + 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; + } + g_max_system_fd = (uint32_t)rlim.rlim_cur; + + if (g_timer_fd_ctx_map == NULL) { + 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; + } + + 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 = NULL; + return UBRING_ERR; + } + } + + RETURN_CODE ret = StartTimeEpoll(); + if (ret != UBRING_OK) { + LOG(ERROR) << "Failed to start Timer Epoll. ret=" << ret; + if (LIKELY(g_timer_fd_ctx_map != NULL)) { + FREE_PTR(g_timer_fd_ctx_map); + } + return UBRING_ERR; + } + g_timer_module_initialized = 1; + return UBRING_OK; +} + +void *UnifiedCallback(void *args) { + TimerFdCtx *ctx = (TimerFdCtx *)args; + if (pthread_spin_lock(&ctx->spinLock) != 0) { + return NULL; + } + + if (ctx->status == TIMER_CONTEXT_NOT_USING) { + pthread_spin_unlock(&ctx->spinLock); + return NULL; + } + + void *(*cb)(void *) = ctx->cb; + void *cbArgs = ctx->args; + uint32_t fd = ctx->fd; + int isPeriodical = ctx->periodical; + ctx->status = TIMER_CONTEXT_CALLBACK_ONGOING; + + pthread_spin_unlock(&ctx->spinLock); + + cb(cbArgs); + + if (!isPeriodical) { + DeleteTimerInner(fd); + } + return NULL; +} + +void *TimerEpoll(void *args) { + UNREFERENCE_PARAM(args); +#if defined(OS_LINUX) + struct epoll_event readyEvents[MAX_TIMER]; +#elif defined(OS_MACOSX) + struct kevent readyEvents[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 readyNum = epoll_wait(g_epoll_fd, readyEvents, MAX_TIMER, Review Comment: Please update the variable naming to follow the snake_case style used in this codebase. For example, `readyNum` should be renamed to `ready_num`. Please apply the same naming convention to other variables as well. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
