zchuango commented on code in PR #3509:
URL: https://github.com/apache/brpc/pull/3509#discussion_r4429936772
##########
src/brpc/ubshm/timer/timer_mgr.cpp:
##########
@@ -15,454 +15,378 @@
// 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/atomicops.h"
+#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;
+namespace {
+
+enum UbrTimerState {
+ kStarting = 0, // published, not scheduled
yet
+ kScheduled = 1,
+ kDead = 2 // scheduling failed
+};
+
+// Granularity of the two join waits in UbrTimerDelAndWait. They only run on
+// teardown paths and only for the short window in which a timer is being
+// started or a callback is finishing.
+constexpr uint64_t kTimerPollIntervalUs = 100;
+
+} // 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.
+//
+// A periodic callback that deletes its own timer does NOT take the task out
+// of the slot: it only marks the task stopped, and the firing callback
+// retires the slot right before it returns (see UbrTimerOnFire). Keeping the
+// task anchored for the whole callback is what lets a concurrent
+// UbrTimerDelAndWait still find it and wait for the callback to finish,
+// instead of returning as if no callback were in flight.
+struct UbrTimerTask {
+ butil::atomic<UbrTimerId>* slot;
+ butil::atomic<bthread_timer_t> id;
+ void* (*cb)(void*, uint64_t);
+ void* arg;
+ uint64_t gen; // opaque, passed back to cb
+ UbrTimerBackoffFn backoff;
+ uint64_t interval_us; // timer thread only
+ bool periodic;
+ butil::atomic<int> state; // kStarting/kScheduled/kDead
+ butil::atomic<bool> stopped;
+ butil::atomic<int> ref;
+ butil::atomic<bool> join_pending; // a DelAndWait is waiting
+ butil::atomic<bool> done; // refs hit zero, joiner frees
+ // Set by whoever settles the handle slot: the one-shot wrapper (right
+ // after its slot CAS, before it runs the user callback), a deleter whose
+ // bthread_timer_del cancelled the task before dispatch, or the kDead
+ // path. Once true, no facade code -- including the dispatched wrapper --
+ // will ever touch `*slot` again, so the object storing the slot may be
+ // freed. The scheduled wrapper is kept alive by its own schedule
+ // reference while a deleter waits on this flag.
+ butil::atomic<bool> slot_retired;
+};
+
+namespace {
+
+// The task whose callback is running on this thread, if any. It lets
+// UbrTimerDel recognize a delete issued by the callback's own dispatch and
+// leave the task anchored until the callback returns.
+thread_local UbrTimerTask* t_running_task = nullptr;
+
+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_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;
+void UbrTimerOnFire(void* p) {
+ UbrTimerTask* task = (UbrTimerTask*)p;
+
+ if (task->periodic) {
+ if (!task->stopped.load()) {
+ // Publish the running task so an UbrTimerDel issued by this very
+ // callback leaves the slot anchored for a concurrent waiter.
+ UbrTimerTask* prev_running = t_running_task;
+ t_running_task = task;
+ task->cb(task->arg, task->gen);
+ t_running_task = prev_running;
}
- }
-
- 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;
-}
-
-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
-
- 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;
+ // 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()) {
+ // Retire the slot here, on the callback's own exit: a self-delete
+ // deliberately left the task anchored. A deleter that took the
+ // slot instead has already consumed the owner reference.
+ UbrTimerId anchored = task;
+ if (task->slot->compare_exchange_strong(anchored, nullptr)) {
+ ReleaseRef(task); // owner
}
- 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;
+ ReleaseRef(task); // claimed next schedule
+ } else {
+ uint64_t interval = task->interval_us;
+ if (task->backoff != nullptr) {
+ const uint64_t next = task->backoff(task->arg, interval);
+ if (BAIDU_UNLIKELY(next == 0)) {
+ // Re-arming for "now" would let this one timer monopolize
+ // the process-wide timer thread. A back-off (or a flag it
+ // reads) returning 0 is a bug, so keep the previous
+ // interval instead of honouring it. Guarded here rather
+ // than in each back-off function, because any of them can
+ // make this mistake.
+ LOG_EVERY_SECOND(ERROR) << "Ubr timer back-off returned 0,
keeping interval_us="
+ << interval;
+ } else {
+ interval = next;
}
- continue;
- }
- if (TimerFdCtxValidate((uint32_t)timer_fd) != UBRING_OK) {
- continue;
+ task->interval_us = interval;
}
-
- RETURN_CODE ret = ExecuteCallback(timer_fd);
- if (ret != UBRING_OK) {
- LOG(ERROR) << "Failed execute callback ret=" << ret;
- DeleteTimerInner((uint32_t)timer_fd);
- continue;
+ 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);
+ }
+ } 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 = task->slot->compare_exchange_strong(expected, nullptr);
+ // The slot storage is now retired: whatever the competition above
+ // decided, this wrapper will not dereference `slot' again. A one-shot
+ // UbrTimerDel that won the exchange (or that cancelled a task which had
+ // not been dispatched at all) is waiting for this store before it lets
+ // the caller free the storage, so it must happen before the potentially
+ // long user callback.
+ task->slot_retired.store(true);
+ if (owned) {
+ task->cb(task->arg, task->gen);
+ }
+ 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(butil::atomic<UbrTimerId>* slot) {
+ return slot->exchange(nullptr);
}
-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;
+RETURN_CODE TimerStartInternal(butil::atomic<UbrTimerId>* slot, uint64_t
delay_us,
+ uint64_t interval_us, bool periodic,
+ void* (*cb)(void*, uint64_t),
+ void* arg, uint64_t gen,
+ UbrTimerBackoffFn backoff) {
+ if (BAIDU_UNLIKELY(slot == nullptr || cb == nullptr)) {
+ LOG(ERROR) << "Ubr timer start invalid argument, slot=" << slot;
+ return UBRING_ERR;
}
-
-#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;
+ if (BAIDU_UNLIKELY(periodic && interval_us == 0)) {
+ // Rejecting beats silently arming a one-shot: a caller that asks for a
+ // periodic timer and gets a single fire loses whatever the timer was
+ // watching for the rest of the connection's life.
+ LOG(ERROR) << "Ubr periodic timer start requires a positive interval.";
+ 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 (BAIDU_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->gen = gen;
+ task->backoff = backoff;
+ task->interval_us = interval_us;
+ task->periodic = periodic;
+ task->state.store(kStarting);
+ task->stopped.store(false);
+ task->ref.store(3); // owner + schedule + starter
+ task->join_pending.store(false);
+ task->done.store(false);
+ task->slot_retired.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 (!slot->compare_exchange_strong(expected, task)) {
+ 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 (BAIDU_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 = slot->compare_exchange_strong(expected, nullptr);
+ 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
+
+RETURN_CODE UbrTimerStart(butil::atomic<UbrTimerId>* slot, uint64_t delay_us,
+ void* (*cb)(void*, uint64_t), void* arg, uint64_t
gen) {
+ return TimerStartInternal(slot, delay_us, 0, false, cb, arg, gen, nullptr);
}
-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 UbrTimerStartPeriodic(butil::atomic<UbrTimerId>* slot,
+ uint64_t delay_us, uint64_t interval_us,
+ void* (*cb)(void*, uint64_t), void* arg,
+ uint64_t gen, UbrTimerBackoffFn backoff) {
+ return TimerStartInternal(slot, delay_us, interval_us, true, cb, arg, gen,
+ 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);
- }
+int UbrTimerDel(butil::atomic<UbrTimerId>* slot) {
+ if (slot == nullptr) {
+ return 1;
+ }
+ // A periodic callback deleting its own timer must not pull the task out of
+ // the slot: the anchored task is what lets a concurrent UbrTimerDelAndWait
+ // find the running callback and wait for it. Mark it stopped instead and
+ // let UbrTimerOnFire retire the slot when the callback returns. Deleting a
+ // sibling timer from inside a callback is unaffected: the slots differ, so
+ // the pointer comparison below fails and the normal path runs.
+ if (t_running_task != nullptr && slot->load() == t_running_task) {
+ t_running_task->stopped.store(true);
+ // 0 means "the timer is stopped", not "the handle slot is now free":
+ // the slot is retired by UbrTimerOnFire when this callback returns,
and
+ // a concurrent UbrTimerDelAndWait may still be joining the task.
+ return 0;
+ }
+ // 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) {
+ // The wrapper never ran and never will: this call is the one that
+ // settles the slot storage.
+ task->slot_retired.store(true);
+ 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) {
+ // Cancelled before dispatch: the timer thread will never call the
+ // wrapper, so nothing else can retire the slot storage.
+ task->slot_retired.store(true);
+ ReleaseRef(task); // schedule: cancelled before
dispatch
+ } else if (!task->periodic) {
+ // One-shot already dispatched: the wrapper may be between its dispatch
+ // (which made bthread_timer_del return 1) and its slot CAS, so it
+ // still has to touch the storage before it returns. Wait it out here,
+ // while this caller still owns the task through the owner reference,
+ // so that returning 0 really means "the storage is not used anymore".
+ // The wrapper retires the slot before calling any user code and takes
+ // no lock doing so, hence this wait cannot deadlock, and it is bounded
+ // by the OS scheduling of the timer thread. A periodic task needs no
+ // such wait: its slot storage is the pooled UbrTrx, which the callers
+ // of the non-blocking delete never free (only UbrMgrFini does, after
+ // UbrTimerDelAndWait), and its retire-slot load happens after the user
+ // callback -- which may take the very lock this caller holds.
+ while (!task->slot_retired.load()) {
+ bthread_usleep(kTimerPollIntervalUs);
}
- }
- 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;
- }
+ } // ==1 periodic:
dispatched/running;
+ // OnFire releases the
schedule ref
+ ReleaseRef(task); // owner
+ return 0; // This call won the slot competition. For a one-shot
timer,
+ // the callback will not run and the slot storage is
unused.
+ // For a periodic timer, future rearming is stopped, but an
+ // already dispatched or running callback may still
complete.
}
-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;
+void UbrTimerDelAndWait(butil::atomic<UbrTimerId>* slot) {
+ if (slot == nullptr) {
+ return;
}
- 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;
+ // Called by the callback of that very timer: joining it would wait for
+ // ourselves, so degrade to the non-blocking delete (UbrTimerDel marks the
+ // task stopped and lets UbrTimerOnFire retire the slot on callback exit).
+ // Callers therefore do not need to know whether they run on the timer
+ // thread.
+ if (t_running_task != nullptr && slot->load() == t_running_task) {
+ UbrTimerDel(slot);
+ return;
}
- if (g_timer_fd_ctx_map[fd].cb == nullptr) {
- LOG(ERROR) << "The callback is not set.";
- return UBRING_ERR;
+ UbrTimerTask* task = TakeOutTask(slot);
+ if (task == nullptr) {
+ return;
Review Comment:
Fixed. A joining DelAndWait now publishes a sentinel in the slot: a second
DelAndWait waits for it to clear instead of returning early, UbrTimerDel
reports not-won, and a start behind the join is refused. Added facade tests,
including two concurrent DelAndWait over a blocked dispatch — that one fails on
the previous code.
--
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]
