Merge pull request #302 from ctiller/pollset

Fix some TSAN reported errors

src/core/iomgr/pollset_multipoller_with_poll_posix.c

@@ -147,8 +147,6 @@ static int multipoll_with_poll_pollset_maybe_work(
       grpc_fd_unref(h->fds[i]);
     } else {
       h->fds[nf++] = h->fds[i];
-      h->pfds[np].events =
-          grpc_fd_begin_poll(h->fds[i], pollset, POLLIN, POLLOUT);
       h->selfds[np] = h->fds[i];
       h->pfds[np].fd = h->fds[i]->fd;
       h->pfds[np].revents = 0;
@@ -168,6 +166,11 @@ static int multipoll_with_poll_pollset_maybe_work(
   pollset->counter = 1;
   gpr_mu_unlock(&pollset->mu);
 
+  for (i = 1; i < np; i++) {
+    h->pfds[i].events =
+        grpc_fd_begin_poll(h->selfds[i], pollset, POLLIN, POLLOUT);
+  }
+
   r = poll(h->pfds, h->pfd_count, timeout);
   if (r < 0) {
     if (errno != EINTR) {

src/core/iomgr/pollset_posix.c

@@ -75,11 +75,14 @@ static void backup_poller(void *p) {
 }
 
 void grpc_pollset_kick(grpc_pollset *p) {
-  if (!p->counter) return;
-  grpc_pollset_kick_kick(&p->kick_state);
+  if (p->counter) {
+    grpc_pollset_kick_kick(&p->kick_state);
+  }
 }
 
-void grpc_pollset_force_kick(grpc_pollset *p) { grpc_pollset_kick(p); }
+void grpc_pollset_force_kick(grpc_pollset *p) {
+  grpc_pollset_kick_kick(&p->kick_state);
+}
 
 /* global state management */
 
@@ -244,11 +247,12 @@ static int unary_poll_pollset_maybe_work(grpc_pollset *pollset,
   pfd[0].events = POLLIN;
   pfd[0].revents = 0;
   pfd[1].fd = fd->fd;
-  pfd[1].events = grpc_fd_begin_poll(fd, pollset, POLLIN, POLLOUT);
   pfd[1].revents = 0;
   pollset->counter = 1;
   gpr_mu_unlock(&pollset->mu);
 
+  pfd[1].events = grpc_fd_begin_poll(fd, pollset, POLLIN, POLLOUT);
+
   r = poll(pfd, GPR_ARRAY_SIZE(pfd), timeout);
   if (r < 0) {
     if (errno != EINTR) {
@@ -269,9 +273,9 @@ static int unary_poll_pollset_maybe_work(grpc_pollset *pollset,
   }
 
   grpc_pollset_kick_post_poll(&pollset->kick_state);
+  grpc_fd_end_poll(fd, pollset);
 
   gpr_mu_lock(&pollset->mu);
-  grpc_fd_end_poll(fd, pollset);
   pollset->counter = 0;
   gpr_cv_broadcast(&pollset->cv);
   return 1;

src/core/iomgr/pollset_posix.h

@@ -78,7 +78,11 @@ void grpc_pollset_add_fd(grpc_pollset *pollset, struct grpc_fd *fd);
    poll after an fd is orphaned) */
 void grpc_pollset_del_fd(grpc_pollset *pollset, struct grpc_fd *fd);
 
-/* Force any current pollers to break polling */
+/* Force any current pollers to break polling: it's the callers responsibility
+   to ensure that the pollset indeed needs to be kicked - no verification that
+   the pollset is actually performing polling work is done. At worst this will
+   result in spurious wakeups if performed at the wrong moment.
+   Does not touch pollset->mu. */
 void grpc_pollset_force_kick(grpc_pollset *pollset);
 /* Returns the fd to listen on for kicks */
 int grpc_kick_read_fd(grpc_pollset *p);