chenBright commented on code in PR #3509:
URL: https://github.com/apache/brpc/pull/3509#discussion_r4078591743
##########
src/brpc/ubshm/shm/shm_ubs.cpp:
##########
@@ -377,14 +377,15 @@ RETURN_CODE UbsShmInit(void)
RETURN_CODE UbsShmFini(void)
{
- int ret = ubsmem_finalize();
- if (ret != UBSM_OK) {
- LOG(ERROR) << "Ubs shm finalize fail, ret=" << ret;
+ // Stop the cleanup timer before finalizing the SDK it calls into.
+ if (UNLIKELY(DestroyShmTimer(g_shm_list) != UBRING_OK)) {
Review Comment:
UNLIKELY -> BAIDU_UNLIKELY
##########
src/brpc/ubshm/shm/shm_ubs.cpp:
##########
@@ -418,50 +419,56 @@ void *UbsShmCallback(void* args)
return nullptr;
}
- LOCK_GUARD(shm_list->shm_lock);
- while (shm_list->head != nullptr) {
- SHM shm = shm_list->head->shm;
- if (shm.addr == nullptr) {
- LOG(ERROR) << "Ubs input shm param is invalid, addr is NULL.";
+ // Drain one node per fire and keep the SDK calls outside the lock, so
+ // a slow daemon cannot stall the timer thread for the whole list.
+ SHM shm;
+ {
+ LOCK_GUARD(shm_list->shm_lock);
Review Comment:
LOCK_GUARD -> BAIDU_SCOPED_LOCK
##########
src/brpc/ubshm/timer/timer_mgr.cpp:
##########
@@ -15,454 +15,261 @@
// 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;
Review Comment:
std::atomic -> butil::atomic
##########
src/brpc/ubshm/timer/timer_mgr.cpp:
##########
@@ -15,454 +15,261 @@
// 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,
Review Comment:
Why not use butil::atomic?
##########
src/brpc/ubshm/shm/shm_ubs.cpp:
##########
@@ -418,50 +419,56 @@ void *UbsShmCallback(void* args)
return nullptr;
}
- LOCK_GUARD(shm_list->shm_lock);
- while (shm_list->head != nullptr) {
- SHM shm = shm_list->head->shm;
- if (shm.addr == nullptr) {
- LOG(ERROR) << "Ubs input shm param is invalid, addr is NULL.";
+ // Drain one node per fire and keep the SDK calls outside the lock, so
+ // a slow daemon cannot stall the timer thread for the whole list.
+ SHM shm;
+ {
+ LOCK_GUARD(shm_list->shm_lock);
+ if (shm_list->head == nullptr) {
return nullptr;
}
+ shm = shm_list->head->shm;
+ }
+ if (shm.addr == nullptr) {
+ LOG(ERROR) << "Ubs input shm param is invalid, addr is NULL.";
+ LOCK_GUARD(shm_list->shm_lock);
+ DeleteShmToList(shm_list);
+ return nullptr;
+ }
- int ret = ubsmem_shmem_unmap(shm.addr, shm.len);
- if (ret != UBSM_OK) {
- if (ret == UBSM_ERR_NET) {
- return nullptr;
- }
- LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" <<
shm.len << " failed, ret=" << ret;
- return nullptr;
+ int ret = ubsmem_shmem_unmap(shm.addr, shm.len);
+ if (ret != UBSM_OK) {
+ if (ret == UBSM_ERR_NET) {
+ return nullptr; // retried on the next fire
}
- LOG(INFO) << "Ubs unmap shm=" << shm.name << " length=" << shm.len <<
" success.";
+ LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" << shm.len <<
" failed, ret=" << ret;
+ return nullptr; // node stays at head, retried
+ }
+ LOG(INFO) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << "
success.";
Review Comment:
Is this INFO log necessary?
##########
src/brpc/ubshm/shm/shm_ubs.cpp:
##########
@@ -418,50 +419,56 @@ void *UbsShmCallback(void* args)
return nullptr;
}
- LOCK_GUARD(shm_list->shm_lock);
- while (shm_list->head != nullptr) {
- SHM shm = shm_list->head->shm;
- if (shm.addr == nullptr) {
- LOG(ERROR) << "Ubs input shm param is invalid, addr is NULL.";
+ // Drain one node per fire and keep the SDK calls outside the lock, so
+ // a slow daemon cannot stall the timer thread for the whole list.
+ SHM shm;
+ {
+ LOCK_GUARD(shm_list->shm_lock);
+ if (shm_list->head == nullptr) {
return nullptr;
}
+ shm = shm_list->head->shm;
+ }
+ if (shm.addr == nullptr) {
+ LOG(ERROR) << "Ubs input shm param is invalid, addr is NULL.";
+ LOCK_GUARD(shm_list->shm_lock);
+ DeleteShmToList(shm_list);
+ return nullptr;
+ }
- int ret = ubsmem_shmem_unmap(shm.addr, shm.len);
- if (ret != UBSM_OK) {
- if (ret == UBSM_ERR_NET) {
- return nullptr;
- }
- LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" <<
shm.len << " failed, ret=" << ret;
- return nullptr;
+ int ret = ubsmem_shmem_unmap(shm.addr, shm.len);
+ if (ret != UBSM_OK) {
+ if (ret == UBSM_ERR_NET) {
+ return nullptr; // retried on the next fire
}
- LOG(INFO) << "Ubs unmap shm=" << shm.name << " length=" << shm.len <<
" success.";
+ LOG(ERROR) << "Ubs unmap shm=" << shm.name << " length=" << shm.len <<
" failed, ret=" << ret;
+ return nullptr; // node stays at head, retried
+ }
+ LOG(INFO) << "Ubs unmap shm=" << shm.name << " length=" << shm.len << "
success.";
- ret = ubsmem_shmem_deallocate(shm.name);
- if (ret != UBSM_OK) {
- DeleteShmToList(shm_list);
- LOG(ERROR) << "Ubs delete shm=" << shm.name << " failed, ret=" <<
ret;
- return nullptr;
- }
+ {
+ LOCK_GUARD(shm_list->shm_lock);
DeleteShmToList(shm_list);
- LOG(INFO) << "Ubs free local shm=" << shm.name << " length=" <<
shm.len << " success.";
}
+ ret = ubsmem_shmem_deallocate(shm.name);
+ if (ret != UBSM_OK) {
+ LOG(ERROR) << "Ubs delete shm=" << shm.name << " failed, ret=" << ret;
+ return nullptr;
+ }
+ LOG(INFO) << "Ubs free local shm=" << shm.name << " length=" << shm.len <<
" success.";
Review Comment:
Is this INFO log necessary?
--
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]
