Merge branch 'master' into fd_rw_atm_closure

9 files changed, 70 insertions, 110 deletions

diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c
index 5c77eda006..31ef433229 100644
--- a/src/core/lib/iomgr/ev_epoll_linux.c
+++ b/src/core/lib/iomgr/ev_epoll_linux.c
@@ -1968,13 +1968,12 @@ static grpc_pollset_set *pollset_set_create(void) {
   return pss;
 }
 
-static void pollset_set_destroy(grpc_pollset_set *pss) {
+static void pollset_set_destroy(grpc_exec_ctx *exec_ctx,
+                                grpc_pollset_set *pss) {
   gpr_mu_destroy(&pss->po.mu);
 
   if (pss->po.pi != NULL) {
-    grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-    PI_UNREF(&exec_ctx, pss->po.pi, "pss_destroy");
-    grpc_exec_ctx_finish(&exec_ctx);
+    PI_UNREF(exec_ctx, pss->po.pi, "pss_destroy");
   }
 
   gpr_free(pss);
diff --git a/src/core/lib/iomgr/ev_poll_posix.c b/src/core/lib/iomgr/ev_poll_posix.c
index 21eb62753e..c03fadaebb 100644
--- a/src/core/lib/iomgr/ev_poll_posix.c
+++ b/src/core/lib/iomgr/ev_poll_posix.c
@@ -149,7 +149,7 @@ static void fd_end_poll(grpc_exec_ctx *exec_ctx, grpc_fd_watcher *rec,
 static bool fd_is_orphaned(grpc_fd *fd);
 
 /* Reference counting for fds */
-/*#define GRPC_FD_REF_COUNT_DEBUG*/
+//#define GRPC_FD_REF_COUNT_DEBUG
 #ifdef GRPC_FD_REF_COUNT_DEBUG
 static void fd_ref(grpc_fd *fd, const char *reason, const char *file, int line);
 static void fd_unref(grpc_fd *fd, const char *reason, const char *file,
@@ -191,6 +191,7 @@ struct grpc_pollset {
   int kicked_without_pollers;
   grpc_closure *shutdown_done;
   grpc_closure_list idle_jobs;
+  int pollset_set_count;
   /* all polled fds */
   size_t fd_count;
   size_t fd_capacity;
@@ -228,7 +229,7 @@ static grpc_error *pollset_kick_ext(grpc_pollset *p,
 
 /* Return 1 if the pollset has active threads in pollset_work (pollset must
  * be locked) */
-static int pollset_has_workers(grpc_pollset *pollset);
+static bool pollset_has_workers(grpc_pollset *pollset);
 
 /*******************************************************************************
  * pollset_set definitions
@@ -282,8 +283,8 @@ cv_fd_table g_cvfds;
 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);
+          (int)gpr_atm_no_barrier_load(&fd->refst),
+          (int)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)
@@ -297,8 +298,8 @@ 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);
+          (int)gpr_atm_no_barrier_load(&fd->refst),
+          (int)gpr_atm_no_barrier_load(&fd->refst) - n, reason, file, line);
 #else
 static void unref_by(grpc_fd *fd, int n) {
   gpr_atm old;
@@ -658,10 +659,18 @@ static void remove_worker(grpc_pollset *p, grpc_pollset_worker *worker) {
   worker->next->prev = worker->prev;
 }
 
-static int pollset_has_workers(grpc_pollset *p) {
+static bool pollset_has_workers(grpc_pollset *p) {
   return p->root_worker.next != &p->root_worker;
 }
 
+static bool pollset_in_pollset_sets(grpc_pollset *p) {
+  return p->pollset_set_count;
+}
+
+static bool pollset_has_observers(grpc_pollset *p) {
+  return pollset_has_workers(p) || pollset_in_pollset_sets(p);
+}
+
 static grpc_pollset_worker *pop_front_worker(grpc_pollset *p) {
   if (pollset_has_workers(p)) {
     grpc_pollset_worker *w = p->root_worker.next;
@@ -800,6 +809,7 @@ static void pollset_init(grpc_pollset *pollset, gpr_mu **mu) {
   pollset->fd_count = 0;
   pollset->fd_capacity = 0;
   pollset->fds = NULL;
+  pollset->pollset_set_count = 0;
 }
 
 static void pollset_destroy(grpc_pollset *pollset) {
@@ -1061,7 +1071,7 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
   if (pollset->shutting_down) {
     if (pollset_has_workers(pollset)) {
       pollset_kick(pollset, NULL);
-    } else if (!pollset->called_shutdown) {
+    } else if (!pollset->called_shutdown && !pollset_has_observers(pollset)) {
       pollset->called_shutdown = 1;
       gpr_mu_unlock(&pollset->mu);
       finish_shutdown(exec_ctx, pollset);
@@ -1093,7 +1103,7 @@ static void pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
   if (!pollset_has_workers(pollset)) {
     grpc_closure_list_sched(exec_ctx, &pollset->idle_jobs);
   }
-  if (!pollset->called_shutdown && !pollset_has_workers(pollset)) {
+  if (!pollset->called_shutdown && !pollset_has_observers(pollset)) {
     pollset->called_shutdown = 1;
     finish_shutdown(exec_ctx, pollset);
   }
@@ -1127,12 +1137,27 @@ static grpc_pollset_set *pollset_set_create(void) {
   return pollset_set;
 }
 
-static void pollset_set_destroy(grpc_pollset_set *pollset_set) {
+static void pollset_set_destroy(grpc_exec_ctx *exec_ctx,
+                                grpc_pollset_set *pollset_set) {
   size_t i;
   gpr_mu_destroy(&pollset_set->mu);
   for (i = 0; i < pollset_set->fd_count; i++) {
     GRPC_FD_UNREF(pollset_set->fds[i], "pollset_set");
   }
+  for (i = 0; i < pollset_set->pollset_count; i++) {
+    grpc_pollset *pollset = pollset_set->pollsets[i];
+    gpr_mu_lock(&pollset->mu);
+    pollset->pollset_set_count--;
+    /* check shutdown */
+    if (pollset->shutting_down && !pollset->called_shutdown &&
+        !pollset_has_observers(pollset)) {
+      pollset->called_shutdown = 1;
+      gpr_mu_unlock(&pollset->mu);
+      finish_shutdown(exec_ctx, pollset);
+    } else {
+      gpr_mu_unlock(&pollset->mu);
+    }
+  }
   gpr_free(pollset_set->pollsets);
   gpr_free(pollset_set->pollset_sets);
   gpr_free(pollset_set->fds);
@@ -1143,6 +1168,9 @@ static void pollset_set_add_pollset(grpc_exec_ctx *exec_ctx,
                                     grpc_pollset_set *pollset_set,
                                     grpc_pollset *pollset) {
   size_t i, j;
+  gpr_mu_lock(&pollset->mu);
+  pollset->pollset_set_count++;
+  gpr_mu_unlock(&pollset->mu);
   gpr_mu_lock(&pollset_set->mu);
   if (pollset_set->pollset_count == pollset_set->pollset_capacity) {
     pollset_set->pollset_capacity =
@@ -1178,6 +1206,17 @@ static void pollset_set_del_pollset(grpc_exec_ctx *exec_ctx,
     }
   }
   gpr_mu_unlock(&pollset_set->mu);
+  gpr_mu_lock(&pollset->mu);
+  pollset->pollset_set_count--;
+  /* check shutdown */
+  if (pollset->shutting_down && !pollset->called_shutdown &&
+      !pollset_has_observers(pollset)) {
+    pollset->called_shutdown = 1;
+    gpr_mu_unlock(&pollset->mu);
+    finish_shutdown(exec_ctx, pollset);
+  } else {
+    gpr_mu_unlock(&pollset->mu);
+  }
 }
 
 static void pollset_set_add_pollset_set(grpc_exec_ctx *exec_ctx,
diff --git a/src/core/lib/iomgr/ev_posix.c b/src/core/lib/iomgr/ev_posix.c
index 95b1d99d75..b5be5504b9 100644
--- a/src/core/lib/iomgr/ev_posix.c
+++ b/src/core/lib/iomgr/ev_posix.c
@@ -215,8 +215,9 @@ grpc_pollset_set *grpc_pollset_set_create(void) {
   return g_event_engine->pollset_set_create();
 }
 
-void grpc_pollset_set_destroy(grpc_pollset_set *pollset_set) {
-  g_event_engine->pollset_set_destroy(pollset_set);
+void grpc_pollset_set_destroy(grpc_exec_ctx *exec_ctx,
+                              grpc_pollset_set *pollset_set) {
+  g_event_engine->pollset_set_destroy(exec_ctx, pollset_set);
 }
 
 void grpc_pollset_set_add_pollset(grpc_exec_ctx *exec_ctx,
diff --git a/src/core/lib/iomgr/ev_posix.h b/src/core/lib/iomgr/ev_posix.h
index 1aea7d61f3..1a9e5c115a 100644
--- a/src/core/lib/iomgr/ev_posix.h
+++ b/src/core/lib/iomgr/ev_posix.h
@@ -74,7 +74,8 @@ typedef struct grpc_event_engine_vtable {
                          struct grpc_fd *fd);
 
   grpc_pollset_set *(*pollset_set_create)(void);
-  void (*pollset_set_destroy)(grpc_pollset_set *pollset_set);
+  void (*pollset_set_destroy)(grpc_exec_ctx *exec_ctx,
+                              grpc_pollset_set *pollset_set);
   void (*pollset_set_add_pollset)(grpc_exec_ctx *exec_ctx,
                                   grpc_pollset_set *pollset_set,
                                   grpc_pollset *pollset);
diff --git a/src/core/lib/iomgr/network_status_tracker.c b/src/core/lib/iomgr/network_status_tracker.c
index 1601a39002..4104bf927a 100644
--- a/src/core/lib/iomgr/network_status_tracker.c
+++ b/src/core/lib/iomgr/network_status_tracker.c
@@ -31,95 +31,18 @@
  *
  */
 
-#include <grpc/support/alloc.h>
-#include <grpc/support/log.h>
 #include "src/core/lib/iomgr/endpoint.h"
 
-typedef struct endpoint_ll_node {
-  grpc_endpoint *ep;
-  struct endpoint_ll_node *next;
-} endpoint_ll_node;
-
-static endpoint_ll_node *head = NULL;
-static gpr_mu g_endpoint_mutex;
-
-void grpc_network_status_shutdown(void) {
-  if (head != NULL) {
-    gpr_log(GPR_ERROR,
-            "Memory leaked as not all network endpoints were shut down");
-  }
-  gpr_mu_destroy(&g_endpoint_mutex);
-}
+void grpc_network_status_shutdown(void) {}
 
 void grpc_network_status_init(void) {
-  gpr_mu_init(&g_endpoint_mutex);
   // TODO(makarandd): Install callback with OS to monitor network status.
 }
 
-void grpc_destroy_network_status_monitor() {
-  for (endpoint_ll_node *curr = head; curr != NULL;) {
-    endpoint_ll_node *next = curr->next;
-    gpr_free(curr);
-    curr = next;
-  }
-  gpr_mu_destroy(&g_endpoint_mutex);
-}
-
-void grpc_network_status_register_endpoint(grpc_endpoint *ep) {
-  gpr_mu_lock(&g_endpoint_mutex);
-  if (head == NULL) {
-    head = (endpoint_ll_node *)gpr_malloc(sizeof(endpoint_ll_node));
-    head->ep = ep;
-    head->next = NULL;
-  } else {
-    endpoint_ll_node *prev_head = head;
-    head = (endpoint_ll_node *)gpr_malloc(sizeof(endpoint_ll_node));
-    head->ep = ep;
-    head->next = prev_head;
-  }
-  gpr_mu_unlock(&g_endpoint_mutex);
-}
+void grpc_destroy_network_status_monitor() {}
 
-void grpc_network_status_unregister_endpoint(grpc_endpoint *ep) {
-  gpr_mu_lock(&g_endpoint_mutex);
-  GPR_ASSERT(head);
-  bool found = false;
-  endpoint_ll_node *prev = head;
-  // if we're unregistering the head, just move head to the next
-  if (ep == head->ep) {
-    head = head->next;
-    gpr_free(prev);
-    found = true;
-  } else {
-    for (endpoint_ll_node *curr = head->next; curr != NULL; curr = curr->next) {
-      if (ep == curr->ep) {
-        prev->next = curr->next;
-        gpr_free(curr);
-        found = true;
-        break;
-      }
-      prev = curr;
-    }
-  }
-  gpr_mu_unlock(&g_endpoint_mutex);
-  GPR_ASSERT(found);
-}
+void grpc_network_status_register_endpoint(grpc_endpoint *ep) { (void)ep; }
 
-// Walk the linked-list from head and execute shutdown. It is possible that
-// other threads might be in the process of shutdown as well, but that has
-// no side effect since endpoint shutdown is idempotent.
-void grpc_network_status_shutdown_all_endpoints() {
-  gpr_mu_lock(&g_endpoint_mutex);
-  if (head == NULL) {
-    gpr_mu_unlock(&g_endpoint_mutex);
-    return;
-  }
-  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+void grpc_network_status_unregister_endpoint(grpc_endpoint *ep) { (void)ep; }
 
-  for (endpoint_ll_node *curr = head; curr != NULL; curr = curr->next) {
-    curr->ep->vtable->shutdown(&exec_ctx, curr->ep,
-                               GRPC_ERROR_CREATE("Network unavailable"));
-  }
-  gpr_mu_unlock(&g_endpoint_mutex);
-  grpc_exec_ctx_finish(&exec_ctx);
-}
+void grpc_network_status_shutdown_all_endpoints() {}
diff --git a/src/core/lib/iomgr/pollset_set.h b/src/core/lib/iomgr/pollset_set.h
index 34bb728c41..d11801d63b 100644
--- a/src/core/lib/iomgr/pollset_set.h
+++ b/src/core/lib/iomgr/pollset_set.h
@@ -44,7 +44,8 @@
 typedef struct grpc_pollset_set grpc_pollset_set;
 
 grpc_pollset_set *grpc_pollset_set_create(void);
-void grpc_pollset_set_destroy(grpc_pollset_set *pollset_set);
+void grpc_pollset_set_destroy(grpc_exec_ctx *exec_ctx,
+                              grpc_pollset_set *pollset_set);
 void grpc_pollset_set_add_pollset(grpc_exec_ctx *exec_ctx,
                                   grpc_pollset_set *pollset_set,
                                   grpc_pollset *pollset);
diff --git a/src/core/lib/iomgr/pollset_set_uv.c b/src/core/lib/iomgr/pollset_set_uv.c
index e5ef8b29e0..836cfee4ef 100644
--- a/src/core/lib/iomgr/pollset_set_uv.c
+++ b/src/core/lib/iomgr/pollset_set_uv.c
@@ -41,7 +41,8 @@ grpc_pollset_set* grpc_pollset_set_create(void) {
   return (grpc_pollset_set*)((intptr_t)0xdeafbeef);
 }
 
-void grpc_pollset_set_destroy(grpc_pollset_set* pollset_set) {}
+void grpc_pollset_set_destroy(grpc_exec_ctx* exec_ctx,
+                              grpc_pollset_set* pollset_set) {}
 
 void grpc_pollset_set_add_pollset(grpc_exec_ctx* exec_ctx,
                                   grpc_pollset_set* pollset_set,
diff --git a/src/core/lib/iomgr/pollset_set_windows.c b/src/core/lib/iomgr/pollset_set_windows.c
index 645650db9b..ae18c8a3ce 100644
--- a/src/core/lib/iomgr/pollset_set_windows.c
+++ b/src/core/lib/iomgr/pollset_set_windows.c
@@ -42,7 +42,8 @@ grpc_pollset_set* grpc_pollset_set_create(void) {
   return (grpc_pollset_set*)((intptr_t)0xdeafbeef);
 }
 
-void grpc_pollset_set_destroy(grpc_pollset_set* pollset_set) {}
+void grpc_pollset_set_destroy(grpc_exec_ctx* exec_ctx,
+                              grpc_pollset_set* pollset_set) {}
 
 void grpc_pollset_set_add_pollset(grpc_exec_ctx* exec_ctx,
                                   grpc_pollset_set* pollset_set,
diff --git a/src/core/lib/iomgr/timer_generic.c b/src/core/lib/iomgr/timer_generic.c
index 40c8351472..8a5617e7c1 100644
--- a/src/core/lib/iomgr/timer_generic.c
+++ b/src/core/lib/iomgr/timer_generic.c
@@ -121,12 +121,6 @@ void grpc_timer_list_shutdown(grpc_exec_ctx *exec_ctx) {
   g_initialized = false;
 }
 
-/* This is a cheap, but good enough, pointer hash for sharding the tasks: */
-static size_t shard_idx(const grpc_timer *info) {
-  size_t x = (size_t)info;
-  return ((x >> 4) ^ (x >> 9) ^ (x >> 14)) & (NUM_SHARDS - 1);
-}
-
 static double ts_to_dbl(gpr_timespec ts) {
   return (double)ts.tv_sec + 1e-9 * ts.tv_nsec;
 }
@@ -181,7 +175,7 @@ void grpc_timer_init(grpc_exec_ctx *exec_ctx, grpc_timer *timer,
                      gpr_timespec deadline, grpc_closure *closure,
                      gpr_timespec now) {
   int is_first_timer = 0;
-  shard_type *shard = &g_shards[shard_idx(timer)];
+  shard_type *shard = &g_shards[GPR_HASH_POINTER(timer, NUM_SHARDS)];
   GPR_ASSERT(deadline.clock_type == g_clock_type);
   GPR_ASSERT(now.clock_type == g_clock_type);
   timer->closure = closure;
@@ -247,7 +241,7 @@ void grpc_timer_cancel(grpc_exec_ctx *exec_ctx, grpc_timer *timer) {
     return;
   }
 
-  shard_type *shard = &g_shards[shard_idx(timer)];
+  shard_type *shard = &g_shards[GPR_HASH_POINTER(timer, NUM_SHARDS)];
   gpr_mu_lock(&shard->mu);
   if (!timer->triggered) {
     grpc_closure_sched(exec_ctx, timer->closure, GRPC_ERROR_CANCELLED);