1 files changed, 106 insertions, 160 deletions

diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index d2f615271e..83ff1f26a5 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -54,31 +54,8 @@
 #include <grpc/support/tls.h>
 #include <grpc/support/useful.h>
 
-static grpc_pollset g_backup_pollset;
-static int g_shutdown_backup_poller;
-static gpr_event g_backup_poller_done;
-static gpr_event g_backup_pollset_shutdown_done;
-
 GPR_TLS_DECL(g_current_thread_poller);
 
-static void backup_poller(void *p) {
-  gpr_timespec delta = gpr_time_from_millis(100);
-  gpr_timespec last_poll = gpr_now();
-
-  gpr_mu_lock(&g_backup_pollset.mu);
-  while (g_shutdown_backup_poller == 0) {
-    gpr_timespec next_poll = gpr_time_add(last_poll, delta);
-    grpc_pollset_work(&g_backup_pollset, gpr_time_add(gpr_now(), gpr_time_from_seconds(1)));
-    gpr_mu_unlock(&g_backup_pollset.mu);
-    gpr_sleep_until(next_poll);
-    gpr_mu_lock(&g_backup_pollset.mu);
-    last_poll = next_poll;
-  }
-  gpr_mu_unlock(&g_backup_pollset.mu);
-
-  gpr_event_set(&g_backup_poller_done, (void *)1);
-}
-
 void grpc_pollset_kick(grpc_pollset *p) {
   if (gpr_tls_get(&g_current_thread_poller) != (gpr_intptr)p && p->counter) {
     p->vtable->kick(p);
@@ -99,44 +76,14 @@ static void kick_using_pollset_kick(grpc_pollset *p) {
 
 /* global state management */
 
-grpc_pollset *grpc_backup_pollset(void) { return &g_backup_pollset; }
-
 void grpc_pollset_global_init(void) {
-  gpr_thd_id id;
-
   gpr_tls_init(&g_current_thread_poller);
 
   /* Initialize kick fd state */
   grpc_pollset_kick_global_init();
-
-  /* initialize the backup pollset */
-  grpc_pollset_init(&g_backup_pollset);
-
-  /* start the backup poller thread */
-  g_shutdown_backup_poller = 0;
-  gpr_event_init(&g_backup_poller_done);
-  gpr_event_init(&g_backup_pollset_shutdown_done);
-  gpr_thd_new(&id, backup_poller, NULL, NULL);
-}
-
-static void on_backup_pollset_shutdown_done(void *arg) {
-  gpr_event_set(&g_backup_pollset_shutdown_done, (void *)1);
 }
 
 void grpc_pollset_global_shutdown(void) {
-  /* terminate the backup poller thread */
-  gpr_mu_lock(&g_backup_pollset.mu);
-  g_shutdown_backup_poller = 1;
-  gpr_mu_unlock(&g_backup_pollset.mu);
-  gpr_event_wait(&g_backup_poller_done, gpr_inf_future);
-
-  grpc_pollset_shutdown(&g_backup_pollset, on_backup_pollset_shutdown_done,
-                        NULL);
-  gpr_event_wait(&g_backup_pollset_shutdown_done, gpr_inf_future);
-
-  /* destroy the backup pollset */
-  grpc_pollset_destroy(&g_backup_pollset);
-
   /* destroy the kick pipes */
   grpc_pollset_kick_global_destroy();
 
@@ -145,32 +92,34 @@ void grpc_pollset_global_shutdown(void) {
 
 /* main interface */
 
-static void become_empty_pollset(grpc_pollset *pollset);
-static void become_unary_pollset(grpc_pollset *pollset, grpc_fd *fd);
+static void become_basic_pollset(grpc_pollset *pollset, grpc_fd *fd_or_null);
 
 void grpc_pollset_init(grpc_pollset *pollset) {
   gpr_mu_init(&pollset->mu);
-  gpr_cv_init(&pollset->cv);
   grpc_pollset_kick_init(&pollset->kick_state);
   pollset->in_flight_cbs = 0;
   pollset->shutting_down = 0;
-  become_empty_pollset(pollset);
+  pollset->called_shutdown = 0;
+  become_basic_pollset(pollset, NULL);
 }
 
 void grpc_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) {
   gpr_mu_lock(&pollset->mu);
   pollset->vtable->add_fd(pollset, fd);
-  gpr_cv_broadcast(&pollset->cv);
   gpr_mu_unlock(&pollset->mu);
 }
 
 void grpc_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) {
   gpr_mu_lock(&pollset->mu);
   pollset->vtable->del_fd(pollset, fd);
-  gpr_cv_broadcast(&pollset->cv);
   gpr_mu_unlock(&pollset->mu);
 }
 
+static void finish_shutdown(grpc_pollset *pollset) {
+  pollset->vtable->finish_shutdown(pollset);
+  pollset->shutdown_done_cb(pollset->shutdown_done_arg);
+}
+
 int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) {
   /* pollset->mu already held */
   gpr_timespec now = gpr_now();
@@ -193,9 +142,10 @@ int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) {
   if (pollset->shutting_down) {
     if (pollset->counter > 0) {
       grpc_pollset_kick(pollset);
-    } else if (pollset->in_flight_cbs == 0) {
+    } else if (!pollset->called_shutdown && pollset->in_flight_cbs == 0) {
+      pollset->called_shutdown = 1;
       gpr_mu_unlock(&pollset->mu);
-      pollset->shutdown_done_cb(pollset->shutdown_done_arg);
+      finish_shutdown(pollset);
       /* Continuing to access pollset here is safe -- it is the caller's
        * responsibility to not destroy when it has outstanding calls to
        * grpc_pollset_work.
@@ -209,21 +159,24 @@ int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) {
 void grpc_pollset_shutdown(grpc_pollset *pollset,
                            void (*shutdown_done)(void *arg),
                            void *shutdown_done_arg) {
-  int in_flight_cbs;
-  int counter;
+  int call_shutdown = 0;
   gpr_mu_lock(&pollset->mu);
+  GPR_ASSERT(!pollset->shutting_down);
   pollset->shutting_down = 1;
-  in_flight_cbs = pollset->in_flight_cbs;
-  counter = pollset->counter;
+  if (!pollset->called_shutdown && pollset->in_flight_cbs == 0 &&
+      pollset->counter == 0) {
+    pollset->called_shutdown = 1;
+    call_shutdown = 1;
+  }
   pollset->shutdown_done_cb = shutdown_done;
   pollset->shutdown_done_arg = shutdown_done_arg;
-  if (counter > 0) {
+  if (pollset->counter > 0) {
     grpc_pollset_kick(pollset);
   }
   gpr_mu_unlock(&pollset->mu);
 
-  if (in_flight_cbs == 0 && counter == 0) {
-    shutdown_done(shutdown_done_arg);
+  if (call_shutdown) {
+    finish_shutdown(pollset);
   }
 }
 
@@ -233,41 +186,13 @@ void grpc_pollset_destroy(grpc_pollset *pollset) {
   pollset->vtable->destroy(pollset);
   grpc_pollset_kick_destroy(&pollset->kick_state);
   gpr_mu_destroy(&pollset->mu);
-  gpr_cv_destroy(&pollset->cv);
 }
 
 /*
- * empty_pollset - a vtable that provides polling for NO file descriptors
+ * basic_pollset - a vtable that provides polling for zero or one file
+ *                 descriptor via poll()
  */
 
-static void empty_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) {
-  become_unary_pollset(pollset, fd);
-}
-
-static void empty_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) {}
-
-static int empty_pollset_maybe_work(grpc_pollset *pollset,
-                                    gpr_timespec deadline, gpr_timespec now,
-                                    int allow_synchronous_callback) {
-  return 0;
-}
-
-static void empty_pollset_destroy(grpc_pollset *pollset) {}
-
-static const grpc_pollset_vtable empty_pollset = {
-    empty_pollset_add_fd, empty_pollset_del_fd, empty_pollset_maybe_work,
-    kick_using_pollset_kick, empty_pollset_destroy};
-
-static void become_empty_pollset(grpc_pollset *pollset) {
-  pollset->vtable = &empty_pollset;
-}
-
-/*
- * unary_poll_pollset - a vtable that provides polling for one file descriptor
- *                      via poll()
- */
-
-
 typedef struct grpc_unary_promote_args {
   const grpc_pollset_vtable *original_vtable;
   grpc_pollset *pollset;
@@ -275,7 +200,7 @@ typedef struct grpc_unary_promote_args {
   grpc_iomgr_closure promotion_closure;
 } grpc_unary_promote_args;
 
-static void unary_poll_do_promote(void *args, int success) {
+static void basic_do_promote(void *args, int success) {
   grpc_unary_promote_args *up_args = args;
   const grpc_pollset_vtable *original_vtable = up_args->original_vtable;
   grpc_pollset *pollset = up_args->pollset;
@@ -293,7 +218,7 @@ static void unary_poll_do_promote(void *args, int success) {
 
   gpr_mu_lock(&pollset->mu);
   /* First we need to ensure that nobody is polling concurrently */
-  while (pollset->counter != 0) {
+  if (pollset->counter != 0) {
     grpc_pollset_kick(pollset);
     grpc_iomgr_add_callback(&up_args->promotion_closure);
     gpr_mu_unlock(&pollset->mu);
@@ -321,33 +246,33 @@ static void unary_poll_do_promote(void *args, int success) {
     fds[0] = pollset->data.ptr;
     fds[1] = fd;
 
-    if (!grpc_fd_is_orphaned(fds[0])) {
+    if (fds[0] && !grpc_fd_is_orphaned(fds[0])) {
       grpc_platform_become_multipoller(pollset, fds, GPR_ARRAY_SIZE(fds));
-      grpc_fd_unref(fds[0]);
+      GRPC_FD_UNREF(fds[0], "basicpoll");
     } else {
       /* old fd is orphaned and we haven't cleaned it up until now, so remain a
        * unary poller */
       /* Note that it is possible that fds[1] is also orphaned at this point.
        * That's okay, we'll correct it at the next add or poll. */
-      grpc_fd_unref(fds[0]);
+      if (fds[0]) GRPC_FD_UNREF(fds[0], "basicpoll");
       pollset->data.ptr = fd;
-      grpc_fd_ref(fd);
+      GRPC_FD_REF(fd, "basicpoll");
     }
   }
 
-  gpr_cv_broadcast(&pollset->cv);
   gpr_mu_unlock(&pollset->mu);
 
   if (do_shutdown_cb) {
     pollset->shutdown_done_cb(pollset->shutdown_done_arg);
   }
 
-  /* Matching ref in unary_poll_pollset_add_fd */
-  grpc_fd_unref(fd);
+  /* Matching ref in basic_pollset_add_fd */
+  GRPC_FD_UNREF(fd, "basicpoll_add");
 }
 
-static void unary_poll_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) {
+static void basic_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) {
   grpc_unary_promote_args *up_args;
+  GPR_ASSERT(fd);
   if (fd == pollset->data.ptr) return;
 
   if (!pollset->counter) {
@@ -358,92 +283,108 @@ static void unary_poll_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) {
     fds[0] = pollset->data.ptr;
     fds[1] = fd;
 
-    if (!grpc_fd_is_orphaned(fds[0])) {
+    if (fds[0] == NULL) {
+      pollset->data.ptr = fd;
+      GRPC_FD_REF(fd, "basicpoll");
+    } else if (!grpc_fd_is_orphaned(fds[0])) {
       grpc_platform_become_multipoller(pollset, fds, GPR_ARRAY_SIZE(fds));
-      grpc_fd_unref(fds[0]);
+      GRPC_FD_UNREF(fds[0], "basicpoll");
     } else {
       /* old fd is orphaned and we haven't cleaned it up until now, so remain a
        * unary poller */
-      grpc_fd_unref(fds[0]);
+      GRPC_FD_UNREF(fds[0], "basicpoll");
       pollset->data.ptr = fd;
-      grpc_fd_ref(fd);
+      GRPC_FD_REF(fd, "basicpoll");
     }
     return;
   }
 
   /* Now we need to promote. This needs to happen when we're not polling. Since
    * this may be called from poll, the wait needs to happen asynchronously. */
-  grpc_fd_ref(fd);
+  GRPC_FD_REF(fd, "basicpoll_add");
   pollset->in_flight_cbs++;
   up_args = gpr_malloc(sizeof(*up_args));
   up_args->pollset = pollset;
   up_args->fd = fd;
   up_args->original_vtable = pollset->vtable;
-  up_args->promotion_closure.cb = unary_poll_do_promote;
+  up_args->promotion_closure.cb = basic_do_promote;
   up_args->promotion_closure.cb_arg = up_args;
   grpc_iomgr_add_callback(&up_args->promotion_closure);
 
   grpc_pollset_kick(pollset);
 }
 
-static void unary_poll_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) {
+static void basic_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) {
+  GPR_ASSERT(fd);
   if (fd == pollset->data.ptr) {
-    grpc_fd_unref(pollset->data.ptr);
-    become_empty_pollset(pollset);
+    GRPC_FD_UNREF(pollset->data.ptr, "basicpoll");
+    pollset->data.ptr = NULL;
   }
 }
 
-static int unary_poll_pollset_maybe_work(grpc_pollset *pollset,
-                                         gpr_timespec deadline,
-                                         gpr_timespec now,
-                                         int allow_synchronous_callback) {
+static int basic_pollset_maybe_work(grpc_pollset *pollset,
+                                    gpr_timespec deadline, gpr_timespec now,
+                                    int allow_synchronous_callback) {
   struct pollfd pfd[2];
   grpc_fd *fd;
   grpc_fd_watcher fd_watcher;
+  grpc_kick_fd_info *kfd;
   int timeout;
   int r;
+  int nfds;
 
-  if (pollset->counter) {
-    return 0;
-  }
   if (pollset->in_flight_cbs) {
     /* Give do_promote priority so we don't starve it out */
-    return 0;
+    gpr_mu_unlock(&pollset->mu);
+    gpr_mu_lock(&pollset->mu);
+    return 1;
   }
   fd = pollset->data.ptr;
-  if (grpc_fd_is_orphaned(fd)) {
-    grpc_fd_unref(fd);
-    become_empty_pollset(pollset);
-    return 0;
+  if (fd && grpc_fd_is_orphaned(fd)) {
+    GRPC_FD_UNREF(fd, "basicpoll");
+    fd = pollset->data.ptr = NULL;
   }
   if (gpr_time_cmp(deadline, gpr_inf_future) == 0) {
     timeout = -1;
   } else {
-    timeout = gpr_time_to_millis(gpr_time_sub(deadline, now));
-    if (timeout <= 0) {
+    timeout = gpr_time_to_millis(
+        gpr_time_add(gpr_time_sub(deadline, now), gpr_time_from_micros(500)));
+    if (timeout < 0) {
       return 1;
     }
   }
-  pfd[0].fd = grpc_pollset_kick_pre_poll(&pollset->kick_state);
-  if (pfd[0].fd < 0) {
+  kfd = grpc_pollset_kick_pre_poll(&pollset->kick_state);
+  if (kfd == NULL) {
     /* Already kicked */
     return 1;
   }
+  pfd[0].fd = GRPC_POLLSET_KICK_GET_FD(kfd);
   pfd[0].events = POLLIN;
   pfd[0].revents = 0;
-  pfd[1].fd = fd->fd;
-  pfd[1].revents = 0;
-  pollset->counter = 1;
-  gpr_mu_unlock(&pollset->mu);
-
-  pfd[1].events = grpc_fd_begin_poll(fd, pollset, POLLIN, POLLOUT, &fd_watcher);
+  nfds = 1;
+  pollset->counter++;
+  if (fd) {
+    pfd[1].fd = fd->fd;
+    pfd[1].revents = 0;
+    gpr_mu_unlock(&pollset->mu);
+    pfd[1].events =
+        grpc_fd_begin_poll(fd, pollset, POLLIN, POLLOUT, &fd_watcher);
+    if (pfd[1].events != 0) {
+      nfds++;
+    }
+  } else {
+    gpr_mu_unlock(&pollset->mu);
+  }
 
   /* poll fd count (argument 2) is shortened by one if we have no events
      to poll on - such that it only includes the kicker */
-  r = poll(pfd, GPR_ARRAY_SIZE(pfd) - (pfd[1].events == 0), timeout);
+  r = poll(pfd, nfds, timeout);
   GRPC_TIMER_MARK(GRPC_PTAG_POLL_FINISHED, r);
 
-  grpc_fd_end_poll(&fd_watcher, pfd[1].revents & POLLIN, pfd[1].revents & POLLOUT);
+  if (fd) {
+    grpc_fd_end_poll(&fd_watcher, pfd[1].revents & POLLIN,
+                     pfd[1].revents & POLLOUT);
+  }
 
   if (r < 0) {
     if (errno != EINTR) {
@@ -453,39 +394,44 @@ static int unary_poll_pollset_maybe_work(grpc_pollset *pollset,
     /* do nothing */
   } else {
     if (pfd[0].revents & POLLIN) {
-      grpc_pollset_kick_consume(&pollset->kick_state);
-    }
-    if (pfd[1].revents & (POLLIN | POLLHUP | POLLERR)) {
-      grpc_fd_become_readable(fd, allow_synchronous_callback);
+      grpc_pollset_kick_consume(&pollset->kick_state, kfd);
     }
-    if (pfd[1].revents & (POLLOUT | POLLHUP | POLLERR)) {
-      grpc_fd_become_writable(fd, allow_synchronous_callback);
+    if (nfds > 1) {
+      if (pfd[1].revents & (POLLIN | POLLHUP | POLLERR)) {
+        grpc_fd_become_readable(fd, allow_synchronous_callback);
+      }
+      if (pfd[1].revents & (POLLOUT | POLLHUP | POLLERR)) {
+        grpc_fd_become_writable(fd, allow_synchronous_callback);
+      }
     }
   }
 
-  grpc_pollset_kick_post_poll(&pollset->kick_state);
+  grpc_pollset_kick_post_poll(&pollset->kick_state, kfd);
 
   gpr_mu_lock(&pollset->mu);
-  pollset->counter = 0;
-  gpr_cv_broadcast(&pollset->cv);
+  pollset->counter--;
   return 1;
 }
 
-static void unary_poll_pollset_destroy(grpc_pollset *pollset) {
+static void basic_pollset_destroy(grpc_pollset *pollset) {
   GPR_ASSERT(pollset->counter == 0);
-  grpc_fd_unref(pollset->data.ptr);
+  if (pollset->data.ptr != NULL) {
+    GRPC_FD_UNREF(pollset->data.ptr, "basicpoll");
+    pollset->data.ptr = NULL;
+  }
 }
 
-static const grpc_pollset_vtable unary_poll_pollset = {
-    unary_poll_pollset_add_fd, unary_poll_pollset_del_fd,
-    unary_poll_pollset_maybe_work, kick_using_pollset_kick,
-    unary_poll_pollset_destroy};
+static const grpc_pollset_vtable basic_pollset = {
+    basic_pollset_add_fd,    basic_pollset_del_fd,  basic_pollset_maybe_work,
+    kick_using_pollset_kick, basic_pollset_destroy, basic_pollset_destroy};
 
-static void become_unary_pollset(grpc_pollset *pollset, grpc_fd *fd) {
-  pollset->vtable = &unary_poll_pollset;
+static void become_basic_pollset(grpc_pollset *pollset, grpc_fd *fd_or_null) {
+  pollset->vtable = &basic_pollset;
   pollset->counter = 0;
-  pollset->data.ptr = fd;
-  grpc_fd_ref(fd);
+  pollset->data.ptr = fd_or_null;
+  if (fd_or_null) {
+    GRPC_FD_REF(fd_or_null, "basicpoll");
+  }
 }
 
 #endif /* GPR_POSIX_POLLSET */