Merge pull request #1577 from ctiller/we-dont-need-no-backup

Remove backup poller from GRPC core

1 files changed, 48 insertions, 16 deletions

diff --git a/src/core/iomgr/fd_posix.c b/src/core/iomgr/fd_posix.c
index 28ed7708f7..347d8793c8 100644
--- a/src/core/iomgr/fd_posix.c
+++ b/src/core/iomgr/fd_posix.c
@@ -109,16 +109,40 @@ static void destroy(grpc_fd *fd) {
   gpr_free(fd);
 }
 
+#ifdef GRPC_FD_REF_COUNT_DEBUG
+#define REF_BY(fd, n, reason) ref_by(fd, n, reason, __FILE__, __LINE__)
+#define UNREF_BY(fd, n, reason) unref_by(fd, n, reason, __FILE__, __LINE__)
+static void ref_by(grpc_fd *fd, int n, const char *reason, const char *file,
+                   int line) {
+  gpr_log(GPR_DEBUG, "FD %d %p  ref %d %d -> %d [%s; %s:%d]", fd->fd, fd, n,
+          gpr_atm_no_barrier_load(&fd->refst),
+          gpr_atm_no_barrier_load(&fd->refst) + n, reason, file, line);
+#else
+#define REF_BY(fd, n, reason) ref_by(fd, n)
+#define UNREF_BY(fd, n, reason) unref_by(fd, n)
 static void ref_by(grpc_fd *fd, int n) {
+#endif
   GPR_ASSERT(gpr_atm_no_barrier_fetch_add(&fd->refst, n) > 0);
 }
 
+#ifdef GRPC_FD_REF_COUNT_DEBUG
+static void unref_by(grpc_fd *fd, int n, const char *reason, const char *file,
+                     int line) {
+  gpr_atm old;
+  gpr_log(GPR_DEBUG, "FD %d %p unref %d %d -> %d [%s; %s:%d]", fd->fd, fd, n,
+          gpr_atm_no_barrier_load(&fd->refst),
+          gpr_atm_no_barrier_load(&fd->refst) - n, reason, file, line);
+#else
 static void unref_by(grpc_fd *fd, int n) {
-  gpr_atm old = gpr_atm_full_fetch_add(&fd->refst, -n);
+  gpr_atm old;
+#endif
+  old = gpr_atm_full_fetch_add(&fd->refst, -n);
   if (old == n) {
-    grpc_iomgr_add_callback(&fd->on_done_closure);
-    freelist_fd(fd);
+    if (fd->on_done_closure) {
+      grpc_iomgr_add_callback(fd->on_done_closure);
+    }
     grpc_iomgr_unregister_object(&fd->iomgr_object);
+    freelist_fd(fd);
   } else {
     GPR_ASSERT(old > n);
   }
@@ -135,12 +159,9 @@ void grpc_fd_global_shutdown(void) {
   gpr_mu_destroy(&fd_freelist_mu);
 }
 
-static void do_nothing(void *ignored, int success) {}
-
 grpc_fd *grpc_fd_create(int fd, const char *name) {
   grpc_fd *r = alloc_fd(fd);
   grpc_iomgr_register_object(&r->iomgr_object, name);
-  grpc_pollset_add_fd(grpc_backup_pollset(), r);
   return r;
 }
 
@@ -178,24 +199,35 @@ static void wake_all_watchers_locked(grpc_fd *fd) {
   }
 }
 
-void grpc_fd_orphan(grpc_fd *fd, grpc_iomgr_cb_func on_done, void *user_data) {
-  grpc_iomgr_closure_init(&fd->on_done_closure, on_done ? on_done : do_nothing,
-                          user_data);
+void grpc_fd_orphan(grpc_fd *fd, grpc_iomgr_closure *on_done,
+                    const char *reason) {
+  fd->on_done_closure = on_done;
   shutdown(fd->fd, SHUT_RDWR);
-  ref_by(fd, 1); /* remove active status, but keep referenced */
+  REF_BY(fd, 1, reason); /* remove active status, but keep referenced */
   gpr_mu_lock(&fd->watcher_mu);
   wake_all_watchers_locked(fd);
   gpr_mu_unlock(&fd->watcher_mu);
-  unref_by(fd, 2); /* drop the reference */
+  UNREF_BY(fd, 2, reason); /* drop the reference */
 }
 
 /* increment refcount by two to avoid changing the orphan bit */
+#ifdef GRPC_FD_REF_COUNT_DEBUG
+void grpc_fd_ref(grpc_fd *fd, const char *reason, const char *file, int line) {
+  ref_by(fd, 2, reason, file, line);
+}
+
+void grpc_fd_unref(grpc_fd *fd, const char *reason, const char *file,
+                   int line) {
+  unref_by(fd, 2, reason, file, line);
+}
+#else
 void grpc_fd_ref(grpc_fd *fd) { ref_by(fd, 2); }
 
 void grpc_fd_unref(grpc_fd *fd) { unref_by(fd, 2); }
+#endif
 
 static void process_callback(grpc_iomgr_closure *closure, int success,
-                          int allow_synchronous_callback) {
+                             int allow_synchronous_callback) {
   if (allow_synchronous_callback) {
     closure->cb(closure->cb_arg, success);
   } else {
@@ -235,7 +267,7 @@ static void notify_on(grpc_fd *fd, gpr_atm *st, grpc_iomgr_closure *closure,
       GPR_ASSERT(gpr_atm_no_barrier_load(st) == READY);
       gpr_atm_rel_store(st, NOT_READY);
       process_callback(closure, !gpr_atm_acq_load(&fd->shutdown),
-                    allow_synchronous_callback);
+                       allow_synchronous_callback);
       return;
     default: /* WAITING */
       /* upcallptr was set to a different closure.  This is an error! */
@@ -279,7 +311,7 @@ static void set_ready(grpc_fd *fd, gpr_atm *st,
   /* only one set_ready can be active at once (but there may be a racing
      notify_on) */
   int success;
-  grpc_iomgr_closure* closure;
+  grpc_iomgr_closure *closure;
   size_t ncb = 0;
 
   gpr_mu_lock(&fd->set_state_mu);
@@ -319,7 +351,7 @@ gpr_uint32 grpc_fd_begin_poll(grpc_fd *fd, grpc_pollset *pollset,
   gpr_uint32 mask = 0;
   /* keep track of pollers that have requested our events, in case they change
    */
-  grpc_fd_ref(fd);
+  GRPC_FD_REF(fd, "poll");
 
   gpr_mu_lock(&fd->watcher_mu);
   /* if there is nobody polling for read, but we need to, then start doing so */
@@ -374,7 +406,7 @@ void grpc_fd_end_poll(grpc_fd_watcher *watcher, int got_read, int got_write) {
   }
   gpr_mu_unlock(&fd->watcher_mu);
 
-  grpc_fd_unref(fd);
+  GRPC_FD_UNREF(fd, "poll");
 }
 
 void grpc_fd_become_readable(grpc_fd *fd, int allow_synchronous_callback) {