|
|
|
@ -57,10 +57,8 @@ |
|
|
|
|
#define MAX_EPOLL_EVENTS 100 |
|
|
|
|
#define MAX_EPOLL_EVENTS_HANDLED_EACH_POLL_CALL 5 |
|
|
|
|
|
|
|
|
|
#ifndef NDEBUG |
|
|
|
|
grpc_tracer_flag grpc_trace_pollable_refcount = |
|
|
|
|
GRPC_TRACER_INITIALIZER(false, "pollable_refcount"); |
|
|
|
|
#endif |
|
|
|
|
grpc_core::DebugOnlyTraceFlag grpc_trace_pollable_refcount(false, |
|
|
|
|
"pollable_refcount"); |
|
|
|
|
|
|
|
|
|
/*******************************************************************************
|
|
|
|
|
* pollable Declarations |
|
|
|
@ -261,7 +259,7 @@ static gpr_mu fd_freelist_mu; |
|
|
|
|
unref_by(ec, fd, n, reason, __FILE__, __LINE__) |
|
|
|
|
static void ref_by(grpc_fd *fd, int n, const char *reason, const char *file, |
|
|
|
|
int line) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_trace_fd_refcount)) { |
|
|
|
|
if (grpc_trace_fd_refcount.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"FD %d %p ref %d %" PRIdPTR " -> %" PRIdPTR " [%s; %s:%d]", |
|
|
|
|
fd->fd, fd, n, gpr_atm_no_barrier_load(&fd->refst), |
|
|
|
@ -295,7 +293,7 @@ static void fd_destroy(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { |
|
|
|
|
#ifndef NDEBUG |
|
|
|
|
static void unref_by(grpc_exec_ctx *exec_ctx, grpc_fd *fd, int n, |
|
|
|
|
const char *reason, const char *file, int line) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_trace_fd_refcount)) { |
|
|
|
|
if (grpc_trace_fd_refcount.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"FD %d %p unref %d %" PRIdPTR " -> %" PRIdPTR " [%s; %s:%d]", |
|
|
|
|
fd->fd, fd, n, gpr_atm_no_barrier_load(&fd->refst), |
|
|
|
@ -357,7 +355,7 @@ static grpc_fd *fd_create(int fd, const char *name) { |
|
|
|
|
gpr_asprintf(&fd_name, "%s fd=%d", name, fd); |
|
|
|
|
grpc_iomgr_register_object(&new_fd->iomgr_object, fd_name); |
|
|
|
|
#ifndef NDEBUG |
|
|
|
|
if (GRPC_TRACER_ON(grpc_trace_fd_refcount)) { |
|
|
|
|
if (grpc_trace_fd_refcount.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "FD %d %p create %s", fd, new_fd, fd_name); |
|
|
|
|
} |
|
|
|
|
#endif |
|
|
|
@ -481,7 +479,7 @@ static grpc_error *pollable_create(pollable_type type, pollable **p) { |
|
|
|
|
static pollable *pollable_ref(pollable *p) { |
|
|
|
|
#else |
|
|
|
|
static pollable *pollable_ref(pollable *p, int line, const char *reason) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_trace_pollable_refcount)) { |
|
|
|
|
if (grpc_trace_pollable_refcount.enabled()) { |
|
|
|
|
int r = (int)gpr_atm_no_barrier_load(&p->refs.count); |
|
|
|
|
gpr_log(__FILE__, line, GPR_LOG_SEVERITY_DEBUG, |
|
|
|
|
"POLLABLE:%p ref %d->%d %s", p, r, r + 1, reason); |
|
|
|
@ -496,7 +494,7 @@ static void pollable_unref(pollable *p) { |
|
|
|
|
#else |
|
|
|
|
static void pollable_unref(pollable *p, int line, const char *reason) { |
|
|
|
|
if (p == NULL) return; |
|
|
|
|
if (GRPC_TRACER_ON(grpc_trace_pollable_refcount)) { |
|
|
|
|
if (grpc_trace_pollable_refcount.enabled()) { |
|
|
|
|
int r = (int)gpr_atm_no_barrier_load(&p->refs.count); |
|
|
|
|
gpr_log(__FILE__, line, GPR_LOG_SEVERITY_DEBUG, |
|
|
|
|
"POLLABLE:%p unref %d->%d %s", p, r, r - 1, reason); |
|
|
|
@ -514,7 +512,7 @@ static grpc_error *pollable_add_fd(pollable *p, grpc_fd *fd) { |
|
|
|
|
static const char *err_desc = "pollable_add_fd"; |
|
|
|
|
const int epfd = p->epfd; |
|
|
|
|
|
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "add fd %p (%d) to pollable %p", fd, fd->fd, p); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -556,7 +554,7 @@ static void pollset_global_shutdown(void) { |
|
|
|
|
/* pollset->mu must be held while calling this function */ |
|
|
|
|
static void pollset_maybe_finish_shutdown(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_pollset *pollset) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"PS:%p (pollable:%p) maybe_finish_shutdown sc=%p (target:!NULL) " |
|
|
|
|
"rw=%p (target:NULL) cpsc=%d (target:0)", |
|
|
|
@ -580,14 +578,14 @@ static grpc_error *kick_one_worker(grpc_exec_ctx *exec_ctx, |
|
|
|
|
GRPC_STATS_INC_POLLSET_KICK(exec_ctx); |
|
|
|
|
GPR_ASSERT(specific_worker != NULL); |
|
|
|
|
if (specific_worker->kicked) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PS:%p kicked_specific_but_already_kicked", p); |
|
|
|
|
} |
|
|
|
|
GRPC_STATS_INC_POLLSET_KICKED_AGAIN(exec_ctx); |
|
|
|
|
return GRPC_ERROR_NONE; |
|
|
|
|
} |
|
|
|
|
if (gpr_tls_get(&g_current_thread_worker) == (intptr_t)specific_worker) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PS:%p kicked_specific_but_awake", p); |
|
|
|
|
} |
|
|
|
|
GRPC_STATS_INC_POLLSET_KICK_OWN_THREAD(exec_ctx); |
|
|
|
@ -596,7 +594,7 @@ static grpc_error *kick_one_worker(grpc_exec_ctx *exec_ctx, |
|
|
|
|
} |
|
|
|
|
if (specific_worker == p->root_worker) { |
|
|
|
|
GRPC_STATS_INC_POLLSET_KICK_WAKEUP_FD(exec_ctx); |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PS:%p kicked_specific_via_wakeup_fd", p); |
|
|
|
|
} |
|
|
|
|
specific_worker->kicked = true; |
|
|
|
@ -605,7 +603,7 @@ static grpc_error *kick_one_worker(grpc_exec_ctx *exec_ctx, |
|
|
|
|
} |
|
|
|
|
if (specific_worker->initialized_cv) { |
|
|
|
|
GRPC_STATS_INC_POLLSET_KICK_WAKEUP_CV(exec_ctx); |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PS:%p kicked_specific_via_cv", p); |
|
|
|
|
} |
|
|
|
|
specific_worker->kicked = true; |
|
|
|
@ -619,7 +617,7 @@ static grpc_error *kick_one_worker(grpc_exec_ctx *exec_ctx, |
|
|
|
|
|
|
|
|
|
static grpc_error *pollset_kick(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, |
|
|
|
|
grpc_pollset_worker *specific_worker) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"PS:%p kick %p tls_pollset=%p tls_worker=%p pollset.root_worker=%p", |
|
|
|
|
pollset, specific_worker, |
|
|
|
@ -630,7 +628,7 @@ static grpc_error *pollset_kick(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, |
|
|
|
|
if (specific_worker == NULL) { |
|
|
|
|
if (gpr_tls_get(&g_current_thread_pollset) != (intptr_t)pollset) { |
|
|
|
|
if (pollset->root_worker == NULL) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PS:%p kicked_any_without_poller", pollset); |
|
|
|
|
} |
|
|
|
|
GRPC_STATS_INC_POLLSET_KICKED_WITHOUT_POLLER(exec_ctx); |
|
|
|
@ -656,7 +654,7 @@ static grpc_error *pollset_kick(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, |
|
|
|
|
exec_ctx, pollset->root_worker->links[PWLINK_POLLSET].next); |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PS:%p kicked_any_but_awake", pollset); |
|
|
|
|
} |
|
|
|
|
GRPC_STATS_INC_POLLSET_KICK_OWN_THREAD(exec_ctx); |
|
|
|
@ -763,7 +761,7 @@ static grpc_error *pollable_process_events(grpc_exec_ctx *exec_ctx, |
|
|
|
|
struct epoll_event *ev = &pollable_obj->events[n]; |
|
|
|
|
void *data_ptr = ev->data.ptr; |
|
|
|
|
if (1 & (intptr_t)data_ptr) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PS:%p got pollset_wakeup %p", pollset, data_ptr); |
|
|
|
|
} |
|
|
|
|
append_error(&error, |
|
|
|
@ -775,7 +773,7 @@ static grpc_error *pollable_process_events(grpc_exec_ctx *exec_ctx, |
|
|
|
|
bool cancel = (ev->events & (EPOLLERR | EPOLLHUP)) != 0; |
|
|
|
|
bool read_ev = (ev->events & (EPOLLIN | EPOLLPRI)) != 0; |
|
|
|
|
bool write_ev = (ev->events & EPOLLOUT) != 0; |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"PS:%p got fd %p: cancel=%d read=%d " |
|
|
|
|
"write=%d", |
|
|
|
@ -803,7 +801,7 @@ static grpc_error *pollable_epoll(grpc_exec_ctx *exec_ctx, pollable *p, |
|
|
|
|
grpc_millis deadline) { |
|
|
|
|
int timeout = poll_deadline_to_millis_timeout(exec_ctx, deadline); |
|
|
|
|
|
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
char *desc = pollable_desc(p); |
|
|
|
|
gpr_log(GPR_DEBUG, "POLLABLE:%p[%s] poll for %dms", p, desc, timeout); |
|
|
|
|
gpr_free(desc); |
|
|
|
@ -823,7 +821,7 @@ static grpc_error *pollable_epoll(grpc_exec_ctx *exec_ctx, pollable *p, |
|
|
|
|
|
|
|
|
|
if (r < 0) return GRPC_OS_ERROR(errno, "epoll_wait"); |
|
|
|
|
|
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "POLLABLE:%p got %d events", p, r); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -891,7 +889,7 @@ static bool begin_worker(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, |
|
|
|
|
worker->initialized_cv = true; |
|
|
|
|
gpr_cv_init(&worker->cv); |
|
|
|
|
gpr_mu_unlock(&pollset->mu); |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace) && |
|
|
|
|
if (grpc_polling_trace.enabled() && |
|
|
|
|
worker->pollable_obj->root_worker != worker) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PS:%p wait %p w=%p for %dms", pollset, |
|
|
|
|
worker->pollable_obj, worker, |
|
|
|
@ -900,18 +898,18 @@ static bool begin_worker(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, |
|
|
|
|
while (do_poll && worker->pollable_obj->root_worker != worker) { |
|
|
|
|
if (gpr_cv_wait(&worker->cv, &worker->pollable_obj->mu, |
|
|
|
|
grpc_millis_to_timespec(deadline, GPR_CLOCK_REALTIME))) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PS:%p timeout_wait %p w=%p", pollset, |
|
|
|
|
worker->pollable_obj, worker); |
|
|
|
|
} |
|
|
|
|
do_poll = false; |
|
|
|
|
} else if (worker->kicked) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PS:%p wakeup %p w=%p", pollset, |
|
|
|
|
worker->pollable_obj, worker); |
|
|
|
|
} |
|
|
|
|
do_poll = false; |
|
|
|
|
} else if (GRPC_TRACER_ON(grpc_polling_trace) && |
|
|
|
|
} else if (grpc_polling_trace.enabled() && |
|
|
|
|
worker->pollable_obj->root_worker != worker) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PS:%p spurious_wakeup %p w=%p", pollset, |
|
|
|
|
worker->pollable_obj, worker); |
|
|
|
@ -982,7 +980,7 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, |
|
|
|
|
#ifndef NDEBUG |
|
|
|
|
WORKER_PTR->originator = gettid(); |
|
|
|
|
#endif |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PS:%p work hdl=%p worker=%p now=%" PRIdPTR |
|
|
|
|
" deadline=%" PRIdPTR " kwp=%d pollable=%p", |
|
|
|
|
pollset, worker_hdl, WORKER_PTR, grpc_exec_ctx_now(exec_ctx), |
|
|
|
@ -1023,7 +1021,7 @@ static grpc_error *pollset_transition_pollable_from_empty_to_fd_locked( |
|
|
|
|
grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_fd *fd) { |
|
|
|
|
static const char *err_desc = "pollset_transition_pollable_from_empty_to_fd"; |
|
|
|
|
grpc_error *error = GRPC_ERROR_NONE; |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"PS:%p add fd %p (%d); transition pollable from empty to fd", |
|
|
|
|
pollset, fd, fd->fd); |
|
|
|
@ -1039,7 +1037,7 @@ static grpc_error *pollset_transition_pollable_from_fd_to_multi_locked( |
|
|
|
|
grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_fd *and_add_fd) { |
|
|
|
|
static const char *err_desc = "pollset_transition_pollable_from_fd_to_multi"; |
|
|
|
|
grpc_error *error = GRPC_ERROR_NONE; |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log( |
|
|
|
|
GPR_DEBUG, |
|
|
|
|
"PS:%p add fd %p (%d); transition pollable from fd %p to multipoller", |
|
|
|
@ -1189,7 +1187,7 @@ static void pollset_set_unref(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss) { |
|
|
|
|
|
|
|
|
|
static void pollset_set_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss, |
|
|
|
|
grpc_fd *fd) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PSS:%p: add fd %p (%d)", pss, fd, fd->fd); |
|
|
|
|
} |
|
|
|
|
grpc_error *error = GRPC_ERROR_NONE; |
|
|
|
@ -1213,7 +1211,7 @@ static void pollset_set_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss, |
|
|
|
|
|
|
|
|
|
static void pollset_set_del_fd(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss, |
|
|
|
|
grpc_fd *fd) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PSS:%p: del fd %p", pss, fd); |
|
|
|
|
} |
|
|
|
|
pss = pss_lock_adam(pss); |
|
|
|
@ -1234,7 +1232,7 @@ static void pollset_set_del_fd(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss, |
|
|
|
|
|
|
|
|
|
static void pollset_set_del_pollset(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_pollset_set *pss, grpc_pollset *ps) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PSS:%p: del pollset %p", pss, ps); |
|
|
|
|
} |
|
|
|
|
pss = pss_lock_adam(pss); |
|
|
|
@ -1285,7 +1283,7 @@ static grpc_error *add_fds_to_pollsets(grpc_exec_ctx *exec_ctx, grpc_fd **fds, |
|
|
|
|
|
|
|
|
|
static void pollset_set_add_pollset(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_pollset_set *pss, grpc_pollset *ps) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PSS:%p: add pollset %p", pss, ps); |
|
|
|
|
} |
|
|
|
|
grpc_error *error = GRPC_ERROR_NONE; |
|
|
|
@ -1322,7 +1320,7 @@ static void pollset_set_add_pollset(grpc_exec_ctx *exec_ctx, |
|
|
|
|
static void pollset_set_add_pollset_set(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_pollset_set *a, |
|
|
|
|
grpc_pollset_set *b) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PSS: merge (%p, %p)", a, b); |
|
|
|
|
} |
|
|
|
|
grpc_error *error = GRPC_ERROR_NONE; |
|
|
|
@ -1356,7 +1354,7 @@ static void pollset_set_add_pollset_set(grpc_exec_ctx *exec_ctx, |
|
|
|
|
if (b_size > a_size) { |
|
|
|
|
GPR_SWAP(grpc_pollset_set *, a, b); |
|
|
|
|
} |
|
|
|
|
if (GRPC_TRACER_ON(grpc_polling_trace)) { |
|
|
|
|
if (grpc_polling_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_DEBUG, "PSS: parent %p to %p", b, a); |
|
|
|
|
} |
|
|
|
|
gpr_ref(&a->refs); |
|
|
|
@ -1453,10 +1451,6 @@ const grpc_event_engine_vtable *grpc_init_epollex_linux( |
|
|
|
|
return NULL; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#ifndef NDEBUG |
|
|
|
|
grpc_register_tracer(&grpc_trace_pollable_refcount); |
|
|
|
|
#endif |
|
|
|
|
|
|
|
|
|
fd_global_init(); |
|
|
|
|
|
|
|
|
|
if (!GRPC_LOG_IF_ERROR("pollset_global_init", pollset_global_init())) { |
|
|
|
|