Completing wakeup story

src/core/iomgr/fd_posix.c

@@ -339,8 +339,8 @@ void grpc_fd_notify_on_write(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
 }
 
 gpr_uint32 grpc_fd_begin_poll(grpc_fd *fd, grpc_pollset *pollset,
-                              gpr_uint32 read_mask, gpr_uint32 write_mask,
-                              grpc_fd_watcher *watcher) {
+                              grpc_pollset_worker *worker, gpr_uint32 read_mask,
+                              gpr_uint32 write_mask, grpc_fd_watcher *watcher) {
   gpr_uint32 mask = 0;
   /* keep track of pollers that have requested our events, in case they change
    */
@@ -351,6 +351,7 @@ gpr_uint32 grpc_fd_begin_poll(grpc_fd *fd, grpc_pollset *pollset,
   if (gpr_atm_no_barrier_load(&fd->shutdown)) {
     watcher->fd = NULL;
     watcher->pollset = NULL;
+    watcher->worker = NULL;
     gpr_mu_unlock(&fd->watcher_mu);
     GRPC_FD_UNREF(fd, "poll");
     return 0;
@@ -369,12 +370,13 @@ gpr_uint32 grpc_fd_begin_poll(grpc_fd *fd, grpc_pollset *pollset,
     mask |= write_mask;
   }
   /* if not polling, remember this watcher in case we need someone to later */
-  if (mask == 0) {
+  if (mask == 0 && worker != NULL) {
     watcher->next = &fd->inactive_watcher_root;
     watcher->prev = watcher->next->prev;
     watcher->next->prev = watcher->prev->next = watcher;
   }
   watcher->pollset = pollset;
+  watcher->worker = worker;
   watcher->fd = fd;
   gpr_mu_unlock(&fd->watcher_mu);
 
@@ -404,7 +406,7 @@ void grpc_fd_end_poll(grpc_exec_ctx *exec_ctx, grpc_fd_watcher *watcher,
     kick = kick || !got_write;
     fd->write_watcher = NULL;
   }
-  if (!was_polling) {
+  if (!was_polling && watcher->worker != NULL) {
     /* remove from inactive list */
     watcher->next->prev = watcher->prev;
     watcher->prev->next = watcher->next;

src/core/iomgr/pollset_multipoller_with_epoll.c

@@ -72,7 +72,7 @@ static void finally_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
   /* We pretend to be polling whilst adding an fd to keep the fd from being
      closed during the add. This may result in a spurious wakeup being assigned
      to this pollset whilst adding, but that should be benign. */
-  GPR_ASSERT(grpc_fd_begin_poll(fd, pollset, 0, 0, &watcher) == 0);
+  GPR_ASSERT(grpc_fd_begin_poll(fd, pollset, NULL, 0, 0, &watcher) == 0);
   if (watcher.fd != NULL) {
     ev.events = (uint32_t)(EPOLLIN | EPOLLOUT | EPOLLET);
     ev.data.ptr = fd;

src/core/iomgr/pollset_multipoller_with_poll_posix.c

@@ -147,8 +147,8 @@ static void multipoll_with_poll_pollset_maybe_work_and_unlock(
   gpr_mu_unlock(&pollset->mu);
 
   for (i = 2; i < pfd_count; i++) {
-    pfds[i].events = (short)grpc_fd_begin_poll(watchers[i].fd, pollset, POLLIN,
-                                               POLLOUT, &watchers[i]);
+    pfds[i].events = (short)grpc_fd_begin_poll(watchers[i].fd, pollset, worker,
+                                               POLLIN, POLLOUT, &watchers[i]);
   }
 
   /* TODO(vpai): Consider first doing a 0 timeout poll here to avoid

src/core/iomgr/pollset_posix.c

@@ -225,8 +225,10 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
   /* pollset->mu already held */
   int added_worker = 0;
   int locked = 1;
+  int queued_work = 0;
   /* this must happen before we (potentially) drop pollset->mu */
   worker->next = worker->prev = NULL;
+  worker->reevaluate_polling_on_wakeup = 0;
   /* TODO(ctiller): pool these */
   grpc_wakeup_fd_init(&worker->wakeup_fd);
   if (!grpc_pollset_has_workers(pollset) &&
@@ -248,29 +250,41 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
     locked = 0;
     goto done;
   }
-  if (!pollset->kicked_without_pollers) {
-    push_front_worker(pollset, worker);
-    added_worker = 1;
-    gpr_tls_set(&g_current_thread_poller, (gpr_intptr)pollset);
-    gpr_tls_set(&g_current_thread_worker, (gpr_intptr)worker);
-    pollset->vtable->maybe_work_and_unlock(exec_ctx, pollset, worker, deadline,
-                                           now);
-    locked = 0;
-    gpr_tls_set(&g_current_thread_poller, 0);
-    gpr_tls_set(&g_current_thread_worker, 0);
-  } else {
-    pollset->kicked_without_pollers = 0;
-  }
-done:
-  if (!locked) {
-    grpc_exec_ctx_flush(exec_ctx);
-    gpr_mu_lock(&pollset->mu);
-    locked = 1;
+  for (;;) {
+    if (!pollset->kicked_without_pollers) {
+      if (!added_worker) {
+        push_front_worker(pollset, worker);
+        added_worker = 1;
+      }
+      gpr_tls_set(&g_current_thread_poller, (gpr_intptr)pollset);
+      gpr_tls_set(&g_current_thread_worker, (gpr_intptr)worker);
+      pollset->vtable->maybe_work_and_unlock(exec_ctx, pollset, worker,
+                                             deadline, now);
+      locked = 0;
+      gpr_tls_set(&g_current_thread_poller, 0);
+      gpr_tls_set(&g_current_thread_worker, 0);
+    } else {
+      pollset->kicked_without_pollers = 0;
+    }
+  done:
+    if (!locked) {
+      queued_work |= grpc_exec_ctx_flush(exec_ctx);
+      gpr_mu_lock(&pollset->mu);
+      locked = 1;
+    }
+    if (worker->reevaluate_polling_on_wakeup) {
+      worker->reevaluate_polling_on_wakeup = 0;
+      if (queued_work) {
+        deadline = gpr_inf_past(GPR_CLOCK_MONOTONIC);
+      }
+      continue;
+    }
+    break;
   }
-  grpc_wakeup_fd_destroy(&worker->wakeup_fd);
   if (added_worker) {
     remove_worker(pollset, worker);
   }
+  grpc_wakeup_fd_destroy(&worker->wakeup_fd);
   if (pollset->shutting_down) {
     if (grpc_pollset_has_workers(pollset)) {
       grpc_pollset_kick(pollset, NULL);
@@ -507,8 +521,8 @@ static void basic_pollset_maybe_work_and_unlock(grpc_exec_ctx *exec_ctx,
     pfd[2].fd = fd->fd;
     pfd[2].revents = 0;
     gpr_mu_unlock(&pollset->mu);
-    pfd[2].events =
-        (short)grpc_fd_begin_poll(fd, pollset, POLLIN, POLLOUT, &fd_watcher);
+    pfd[2].events = (short)grpc_fd_begin_poll(fd, pollset, worker, POLLIN,
+                                              POLLOUT, &fd_watcher);
     if (pfd[2].events != 0) {
       nfds++;
     }