aboutsummaryrefslogtreecommitdiffhomepage
diff options
context:
space:
mode:
authorGravatar Craig Tiller <ctiller@google.com>2017-06-08 23:56:06 +0000
committerGravatar Craig Tiller <ctiller@google.com>2017-06-08 23:56:06 +0000
commit4f07ea8b48d09835152de68402b5a46b2fa67d47 (patch)
tree01e98ac2f133594ce78b6915a9a9af6d3ef8d0f7
parent91b623a10210156b03004325188b9f63b511c08f (diff)
Spread fd events across threads for EPOLLEXCLUSIVE
-rw-r--r--src/core/lib/iomgr/ev_epollex_linux.c97
1 files changed, 58 insertions, 39 deletions
diff --git a/src/core/lib/iomgr/ev_epollex_linux.c b/src/core/lib/iomgr/ev_epollex_linux.c
index a4c6a19d6d..219b8aa7a4 100644
--- a/src/core/lib/iomgr/ev_epollex_linux.c
+++ b/src/core/lib/iomgr/ev_epollex_linux.c
@@ -169,6 +169,8 @@ struct grpc_pollset_worker {
pollable *pollable;
};
+#define MAX_EPOLL_EVENTS 100
+
struct grpc_pollset {
pollable pollable;
pollable *current_pollable;
@@ -176,6 +178,10 @@ struct grpc_pollset {
bool kicked_without_poller;
grpc_closure *shutdown_closure;
grpc_pollset_worker *root_worker;
+
+ int event_cursor;
+ int event_count;
+ struct epoll_event events[MAX_EPOLL_EVENTS];
};
/*******************************************************************************
@@ -438,7 +444,7 @@ static grpc_error *pollable_materialize(pollable *p) {
return err;
}
struct epoll_event ev = {.events = (uint32_t)(EPOLLIN | EPOLLET),
- .data.ptr = &p->wakeup};
+ .data.ptr = (void*)(1 | (intptr_t) &p->wakeup)};
if (epoll_ctl(new_epfd, EPOLL_CTL_ADD, p->wakeup.read_fd, &ev) != 0) {
err = GRPC_OS_ERROR(errno, "epoll_ctl");
close(new_epfd);
@@ -700,6 +706,41 @@ static bool pollset_is_pollable_fd(grpc_pollset *pollset, pollable *p) {
return p != &g_empty_pollable && p != &pollset->pollable;
}
+static grpc_error *pollset_process_events(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, bool drain) {
+ static const char *err_desc = "pollset_process_events";
+ grpc_error *error = GRPC_ERROR_NONE;
+ for (int i = 0; (drain || i < 5) && pollset->event_cursor != pollset->event_count; i++) {
+ int n = pollset->event_cursor++;
+ struct epoll_event *ev = &pollset->events[n];
+ void *data_ptr = ev->data.ptr;
+ if (1 & (intptr_t)data_ptr) {
+ if (GRPC_TRACER_ON(grpc_polling_trace)) {
+ gpr_log(GPR_DEBUG, "PS:%p got pollset_wakeup %p", pollset, data_ptr);
+ }
+ append_error(&error, grpc_wakeup_fd_consume_wakeup((void*)((~(intptr_t)1) & (intptr_t)data_ptr)), err_desc);
+ } else {
+ grpc_fd *fd = (grpc_fd *)data_ptr;
+ 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)) {
+ gpr_log(GPR_DEBUG,
+ "PS:%p got fd %p: cancel=%d read=%d "
+ "write=%d",
+ pollset, fd, cancel, read_ev, write_ev);
+ }
+ if (read_ev || cancel) {
+ fd_become_readable(exec_ctx, fd, pollset);
+ }
+ if (write_ev || cancel) {
+ fd_become_writable(exec_ctx, fd);
+ }
+ }
+ }
+
+ return error;
+}
+
/* pollset_shutdown is guaranteed to be called before pollset_destroy. */
static void pollset_destroy(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) {
pollable_destroy(&pollset->pollable);
@@ -707,16 +748,12 @@ static void pollset_destroy(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) {
UNREF_BY(exec_ctx, (grpc_fd *)pollset->current_pollable, 2,
"pollset_pollable");
}
+ GRPC_LOG_IF_ERROR("pollset_process_events", pollset_process_events(exec_ctx, pollset, true));
}
-#define MAX_EPOLL_EVENTS 100
-
static grpc_error *pollset_epoll(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
pollable *p, gpr_timespec now,
gpr_timespec deadline) {
- struct epoll_event events[MAX_EPOLL_EVENTS];
- static const char *err_desc = "pollset_poll";
-
int timeout = poll_deadline_to_millis_timeout(deadline, now);
if (GRPC_TRACER_ON(grpc_polling_trace)) {
@@ -728,7 +765,7 @@ static grpc_error *pollset_epoll(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
}
int r;
do {
- r = epoll_wait(p->epfd, events, MAX_EPOLL_EVENTS, timeout);
+ r = epoll_wait(p->epfd, pollset->events, MAX_EPOLL_EVENTS, timeout);
} while (r < 0 && errno == EINTR);
if (timeout != 0) {
GRPC_SCHEDULING_END_BLOCKING_REGION;
@@ -740,35 +777,10 @@ static grpc_error *pollset_epoll(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
gpr_log(GPR_DEBUG, "PS:%p poll %p got %d events", pollset, p, r);
}
- grpc_error *error = GRPC_ERROR_NONE;
- for (int i = 0; i < r; i++) {
- void *data_ptr = events[i].data.ptr;
- if (data_ptr == &p->wakeup) {
- if (GRPC_TRACER_ON(grpc_polling_trace)) {
- gpr_log(GPR_DEBUG, "PS:%p poll %p got pollset_wakeup", pollset, p);
- }
- append_error(&error, grpc_wakeup_fd_consume_wakeup(&p->wakeup), err_desc);
- } else {
- grpc_fd *fd = (grpc_fd *)data_ptr;
- bool cancel = (events[i].events & (EPOLLERR | EPOLLHUP)) != 0;
- bool read_ev = (events[i].events & (EPOLLIN | EPOLLPRI)) != 0;
- bool write_ev = (events[i].events & EPOLLOUT) != 0;
- if (GRPC_TRACER_ON(grpc_polling_trace)) {
- gpr_log(GPR_DEBUG,
- "PS:%p poll %p got fd %p: cancel=%d read=%d "
- "write=%d",
- pollset, p, fd, cancel, read_ev, write_ev);
- }
- if (read_ev || cancel) {
- fd_become_readable(exec_ctx, fd, pollset);
- }
- if (write_ev || cancel) {
- fd_become_writable(exec_ctx, fd);
- }
- }
- }
+ pollset->event_cursor = 0;
+ pollset->event_count = r;
- return error;
+ return GRPC_ERROR_NONE;
}
/* Return true if first in list */
@@ -920,10 +932,12 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
gpr_mu_unlock(&worker.pollable->po.mu);
}
gpr_mu_unlock(&pollset->pollable.po.mu);
- append_error(&error, pollset_epoll(exec_ctx, pollset, worker.pollable, now,
- deadline),
- err_desc);
- grpc_exec_ctx_flush(exec_ctx);
+ if (pollset->event_cursor == pollset->event_count) {
+ append_error(&error, pollset_epoll(exec_ctx, pollset, worker.pollable, now,
+ deadline),
+ err_desc);
+ }
+ append_error(&error, pollset_process_events(exec_ctx, pollset, false), err_desc);
gpr_mu_lock(&pollset->pollable.po.mu);
if (worker.pollable != &pollset->pollable) {
gpr_mu_lock(&worker.pollable->po.mu);
@@ -936,6 +950,11 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
if (worker.pollable != &pollset->pollable) {
gpr_mu_unlock(&worker.pollable->po.mu);
}
+ if (grpc_exec_ctx_has_work(exec_ctx)) {
+ gpr_mu_unlock(&pollset->pollable.po.mu);
+ grpc_exec_ctx_flush(exec_ctx);
+ gpr_mu_lock(&pollset->pollable.po.mu);
+ }
return error;
}