Copilot commented on code in PR #3509:
URL: https://github.com/apache/brpc/pull/3509#discussion_r3892895153
##########
src/brpc/ubshm/timer/timer_mgr.h:
##########
@@ -15,59 +15,58 @@
// specific language governing permissions and limitations
// under the License.
+// bthread based timer facade for the ubring module. Callbacks run on the
+// process-wide bthread timer thread and must return quickly.
+
#ifndef BRPC_TIMER_MGR_H
#define BRPC_TIMER_MGR_H
-#include <pthread.h>
-#include <time.h>
-#include "brpc/ubshm/common/common.h"
-#if defined(OS_LINUX)
-#include <sys/epoll.h>
-#include <sys/timerfd.h>
-#elif defined(OS_MACOSX)
-#include <sys/types.h>
-#include <sys/event.h>
-#include <sys/time.h>
-#endif
-
-#define MAX_TIMER 1024
-#define TIMER_EPOLL_WAIT_TIMEOUT 1000
+#include <stdint.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 {
-typedef enum {
- TIMER_CONTEXT_NOT_USING,
- TIMER_CONTEXT_EPOLL_WAITING,
- TIMER_CONTEXT_CALLBACK_ONGOING
-} TimerFdCtxStatus;
-typedef struct {
- void *(*cb)(void*);
- void *args;
- uint32_t fd;
- TimerFdCtxStatus status;
- uint32_t periodical;
- pthread_spinlock_t spin_lock;
-} TimerFdCtx;
+// Opaque timer handle. nullptr means "not started" (or already deleted /
+// fired for one-shot timers).
+typedef struct UbrTimerTask* UbrTimerId;
+
+// Maps the current re-arm interval of a periodic timer to the next one.
+// Runs on the timer thread only.
+typedef uint64_t (*UbrTimerBackoffFn)(void* arg, uint64_t cur_interval_us);
+
+// Schedule `cb(arg)' to run after `delay_us' and, when `interval_us' > 0,
+// re-arm itself after every run until deleted. One-shot timers release
+// their handle slot before running the callback, so the callback may free
+// the object that stores the slot; the task object itself is released
+// automatically.
+RETURN_CODE UbrTimerStart(UbrTimerId* slot, uint64_t delay_us,
+ uint64_t interval_us, void* (*cb)(void*),
+ void* arg,
+ UbrTimerBackoffFn backoff = nullptr);
+
+// Non-blocking delete, safe from inside the timer callback itself. Does
+// not wait for a running callback and does not protect `arg' on its own.
+// Returns 0 when the call won the slot competition: a one-shot callback
+// is guaranteed never to run, and the caller consumes any per-task
+// resources it tracks for this timer (ownership of them transfers to the
+// caller); for a periodic timer an already-started callback is not
+// interrupted. Returns 1 when the callback has been dispatched (it
+// consumes those resources itself on every exit) or its fate is still
+// being settled by the scheduler -- the caller must not consume anything
+// then.
Review Comment:
The `UbrTimerDel` contract comment says it returns 1 once the callback has
been dispatched, but the implementation can still suppress a one-shot callback
after dispatch by winning the slot exchange. Clarifying the comment is
important so callers know when it is safe to reclaim resources referenced by
`arg` (especially for periodic timers where an in-flight callback may still
run).
##########
src/brpc/ubshm/timer/timer_mgr.cpp:
##########
@@ -15,454 +15,259 @@
// 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 <new>
+#include "bthread/bthread.h" // bthread_usleep
+#include "bthread/unstable.h" // bthread_timer_add/del
+#include "butil/time.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 = 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;
- }
-
- 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 RETURN_CODE TimerSpinLocksInit(void) {
- if (g_timer_fd_ctx_map == nullptr) {
- 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].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;
- }
- }
- 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;
-}
-
-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 == 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;
- }
-
- 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;
- }
- }
-
- 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);
+namespace {
+
+enum UbrTimerState {
+ kStarting = 0, // published, not scheduled
yet
+ kScheduled = 1,
+ kDead = 2 // scheduling failed
+};
+
+} // namespace
+
+// Reference rules: one "owner" ref for the handle slot, one "schedule" ref
+// per pending/running bthread schedule, plus one ref held by the starter
+// until its post-schedule bookkeeping is done. The schedule ref is
+// consumed by the firing callback or by the deleter whose
+// bthread_timer_del returned 0 (cancelled before run); the owner ref is
+// consumed by whoever takes the task out of *slot -- a deleter, or the
+// one-shot firing callback itself, which exits the slot BEFORE running
+// the callback so that the callback may free the object storing the slot.
+// All atomics are seq_cst so no interleaving can release a ref twice or
+// free the task while a callback or the starter still touches it.
+struct UbrTimerTask {
+ UbrTimerId* slot;
+ std::atomic<bthread_timer_t> id;
+ void* (*cb)(void*);
+ void* arg;
+ UbrTimerBackoffFn backoff;
+ uint64_t interval_us; // timer thread only
+ bool periodic;
+ std::atomic<int> state; // kStarting/kScheduled/kDead
+ std::atomic<bool> stopped;
+ std::atomic<int> ref;
+ std::atomic<bool> join_pending; // a DelAndWait is waiting
+ std::atomic<bool> done; // refs hit zero, joiner frees
+};
+
+namespace {
+
+void ReleaseRef(UbrTimerTask* task) {
+ if (task->ref.fetch_sub(1) == 1) {
+ if (task->join_pending.load()) {
+ task->done.store(true); // joiner frees the task
+ } else {
+ delete task;
}
- return UBRING_ERR;
- }
- g_timer_module_initialized = 1;
- return UBRING_OK;
-}
-
-void *UnifiedCallback(void *args) {
- TimerFdCtx *ctx = (TimerFdCtx *)args;
- if (pthread_spin_lock(&ctx->spin_lock) != 0) {
- return nullptr;
- }
-
- if (ctx->status == TIMER_CONTEXT_NOT_USING) {
- pthread_spin_unlock(&ctx->spin_lock);
- return nullptr;
- }
-
- 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;
}
-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
+void UbrTimerOnFire(void* p) {
+ UbrTimerTask* task = (UbrTimerTask*)p;
- while (1) {
- if (g_timer_module_initialized <= 0) {
- LOG(ERROR) << "The Timer module is not initialized.";
- break;
+ if (task->periodic) {
+ if (!task->stopped.load()) {
+ task->cb(task->arg);
}
-
-#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;
+ // Claim the next schedule's ref before re-reading `stopped' so a
+ // racing delete can neither free the task nor orphan a re-arm.
+ task->ref.fetch_add(1);
+ if (task->stopped.load()) {
+ ReleaseRef(task);
+ } else {
+ uint64_t interval = task->interval_us;
+ if (task->backoff != nullptr) {
+ interval = task->backoff(task->arg, interval);
+ task->interval_us = interval;
}
- 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;
+ bthread_timer_t id = 0;
+ if (bthread_timer_add(
+ &id, butil::microseconds_from_now((int64_t)interval),
+ UbrTimerOnFire, task) == 0) {
+ task->id.store(id);
+ if (task->stopped.load() && bthread_timer_del(id) == 0) {
+ ReleaseRef(task);
}
- 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;
+ } else {
+ LOG(ERROR) << "Fail to re-arm ubring timer";
+ ReleaseRef(task);
}
}
- }
- return nullptr;
-}
-
-void DeleteTimerSafe(uint32_t fd) {
- if (g_timer_fd_ctx_map == nullptr) {
+ ReleaseRef(task);
return;
}
- if (pthread_spin_lock(&g_timer_fd_ctx_map[fd].spin_lock) != 0) {
- return;
+ // One-shot: exit the handle slot first -- after this the wrapper never
+ // touches the storage again, so the callback may release the object
+ // that holds it. Whether the callback runs is decided solely by this
+ // slot competition: every UbrTimerDel that wants the callback
+ // suppressed has to win this exchange first, so owned==true guarantees
+ // no UbrTimerDel is pending. Do not consult `stopped' here: its store
+ // (del thread) and this load (timer thread) are separated by the slot
+ // RMW and seq_cst does not order the store-buffer case -- ownership of
+ // the slot is the single arbiter.
+ UbrTimerId expected = task;
+ const bool owned =
+ __atomic_compare_exchange_n(task->slot, &expected, (UbrTimerId)
nullptr,
+ false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST);
+ if (owned) {
+ task->cb(task->arg);
+ }
+ ReleaseRef(task); // schedule
+ if (owned) {
+ ReleaseRef(task); // owner
}
-
- if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) {
- pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock);
- return;
- }
-
- 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.";
- return;
- }
-
- g_timer_fd_ctx_map[fd].periodical = 0;
+UbrTimerTask* TakeOutTask(UbrTimerId* slot) {
+ return __atomic_exchange_n(slot, (UbrTimerId) nullptr, __ATOMIC_SEQ_CST);
}
-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;
- }
-
-#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;
+RETURN_CODE TimerStartInternal(UbrTimerId* slot, uint64_t delay_us,
+ uint64_t interval_us, void* (*cb)(void*),
+ void* arg, UbrTimerBackoffFn backoff) {
+ if (UNLIKELY(slot == nullptr || cb == nullptr)) {
+ LOG(ERROR) << "Ubr timer start invalid argument, slot=" << slot;
+ return UBRING_ERR;
}
- 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;
+ UbrTimerTask* task = new (std::nothrow) UbrTimerTask();
+ if (UNLIKELY(task == nullptr)) {
+ LOG(ERROR) << "Fail to malloc ubring timer task.";
+ return UBRING_ERR;
}
-
-#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;
+ task->slot = slot;
+ task->id.store(0);
+ task->cb = cb;
+ task->arg = arg;
+ task->backoff = backoff;
+ task->interval_us = interval_us;
+ task->periodic = (interval_us > 0);
+ task->state.store(kStarting);
+ task->stopped.store(false);
+ task->ref.store(3); // owner + schedule + starter
+ task->join_pending.store(false);
+ task->done.store(false);
+
+ // Publish the real task before scheduling so a delete or a DelAndWait
+ // racing the start always has an object to act on or wait for.
+ UbrTimerId expected = nullptr;
+ if (!__atomic_compare_exchange_n(slot, &expected, (UbrTimerId) task, false,
+ __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+ LOG(ERROR) << "Ubr timer start refused, slot already occupied";
+ delete task; // never published
+ return UBRING_ERR;
}
- std::atomic_fetch_add(&g_total_timer_num, 1U);
-
-#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;
+ bthread_timer_t id = 0;
+ if (UNLIKELY(bthread_timer_add(
+ &id, butil::microseconds_from_now((int64_t)delay_us),
+ UbrTimerOnFire, task) != 0)) {
+ LOG(ERROR) << "Fail to add ubring timer";
+ task->state.store(kDead); // wake DelAndWait waiters
+ expected = task;
+ const bool owned =
+ __atomic_compare_exchange_n(slot, &expected, (UbrTimerId) nullptr,
+ false, __ATOMIC_SEQ_CST,
+ __ATOMIC_SEQ_CST);
+ ReleaseRef(task); // schedule, never ran
+ if (owned) {
+ ReleaseRef(task); // owner
}
- CloseTimerFd(timer_fd);
- std::atomic_fetch_sub(&g_total_timer_num, 1U);
- LOG(ERROR) << "Failed to set timer";
- return -1;
+ ReleaseRef(task); // starter
+ return UBRING_ERR;
}
-
- return timer_fd;
+ // A zero-delay task may have fired and re-armed already; keep a newer
+ // id if so.
+ bthread_timer_t expected_id = 0;
+ task->id.compare_exchange_strong(expected_id, id);
+ task->state.store(kScheduled);
+ // No post-add stopped check here: a UbrTimerDel racing the start
+ // returns 1 without consuming the per-task resources, and the armed
+ // timer must fire so that OnFire settles the ownership protocol.
+ ReleaseRef(task); // starter
+ return UBRING_OK;
}
-uint32_t GetActiveTimerNum(void) {
- return std::atomic_load(&g_total_timer_num);
-}
+} // namespace
-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;
- }
+RETURN_CODE UbrTimerStart(UbrTimerId* slot, uint64_t delay_us,
+ uint64_t interval_us, void* (*cb)(void*),
+ void* arg, UbrTimerBackoffFn backoff) {
+ return TimerStartInternal(slot, delay_us, interval_us, cb, arg, backoff);
}
-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;
- }
+int UbrTimerDel(UbrTimerId* slot) {
+ if (slot == nullptr) {
+ return 1;
+ }
+ // Take the ownership of the slot first: after this exchange every
+ // dereference below is safe (the task cannot be freed while we hold
+ // the owner reference the slot used to anchor).
+ UbrTimerTask* task = TakeOutTask(slot);
+ if (task == nullptr) {
+ return 1; // fired and cleared its slot (callback side consumed)
+ // or another del won the exchange (it consumes)
+ }
+ task->stopped.store(true); // meaningful for periodic
only
+ // A start still in flight cannot be cancelled nor dispatched yet; wait
+ // for the starter to settle the fate (kScheduled/kDead). Bounded: the
+ // starter stores the state before taking any lock our caller holds.
+ while (task->state.load() == kStarting) {
+ bthread_usleep(1000);
+ }
+ if (task->state.load() == kDead) {
+ ReleaseRef(task); // owner; schedule/starter are
+ return 1; // settled by the kDead path
+ }
+ bthread_timer_t id = task->id.load();
+ if (id != 0 && bthread_timer_del(id) == 0) {
+ ReleaseRef(task); // schedule: cancelled before
dispatch
+ } // ==1: dispatched, OnFire
(owned==false)
+ // releases it
+ ReleaseRef(task); // owner
+ return 0; // the callback is guaranteed never to run: the caller
+ // consumes the timer/callback reference
}
Review Comment:
`UbrTimerDel` always returns 0 after taking a non-null slot, even when
`bthread_timer_del(id)` reports the timer was already dispatched. For periodic
timers this breaks the contract in timer_mgr.h and can let callers treat the
callback as fully cancelled (freeing `arg`/per-task resources) while an
already-running callback may still access them, leading to use-after-free.
--
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]
