Spread fd events across threads for EPOLLEXCLUSIVE

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;
 }