Remove support for detached threads. All threads must be joined.

1 files changed, 34 insertions, 8 deletions

diff --git a/src/core/lib/iomgr/ev_poll_posix.cc b/src/core/lib/iomgr/ev_poll_posix.cc
index e630ddf8e0..404665c11f 100644
--- a/src/core/lib/iomgr/ev_poll_posix.cc
+++ b/src/core/lib/iomgr/ev_poll_posix.cc
@@ -253,6 +253,7 @@ typedef struct poll_result {
 } poll_result;
 
 typedef struct poll_args {
+  gpr_thd_id poller_thd;
   gpr_cv trigger;
   int trigger_set;
   struct pollfd* fds;
@@ -262,13 +263,19 @@ typedef struct poll_args {
   struct poll_args* prev;
 } poll_args;
 
+typedef struct poller_dead {
+  gpr_thd_id poller_thd;
+  struct poller_dead* next;
+} poller_dead;
+
 // This is a 2-tiered cache, we mantain a hash table
 // of active poll calls, so we can wait on the result
-// of that call.  We also maintain a freelist of inactive
-// poll threads.
+// of that call.  We also maintain freelists of inactive
+// poll args and of dead poller threads.
 typedef struct poll_hash_table {
   poll_args* free_pollers;
   poll_args** active_pollers;
+  poll_args* dead_pollers;
   unsigned int size;
   unsigned int count;
 } poll_hash_table;
@@ -1299,6 +1306,7 @@ static void pollset_set_del_fd(grpc_pollset_set* pollset_set, grpc_fd* fd) {
 
 static void run_poll(void* args);
 static void cache_poller_locked(poll_args* args);
+static void cache_harvest_locked();
 
 static void cache_insert_locked(poll_args* args) {
   uint32_t key = gpr_murmur_hash3(args->fds, args->nfds * sizeof(struct pollfd),
@@ -1368,11 +1376,8 @@ static poll_args* get_poller_locked(struct pollfd* fds, nfds_t count) {
   pargs->trigger_set = 0;
   init_result(pargs);
   cache_poller_locked(pargs);
-  gpr_thd_id t_id;
-  gpr_thd_options opt = gpr_thd_options_default();
   gpr_ref(&g_cvfds.pollcount);
-  gpr_thd_options_set_detached(&opt);
-  GPR_ASSERT(gpr_thd_new(&t_id, "grpc_poller", &run_poll, pargs, &opt));
+  GPR_ASSERT(gpr_thd_new(&pargs->poller_thd, "grpc_poller", &run_poll, pargs));
   return pargs;
 }
 
@@ -1437,7 +1442,22 @@ static void cache_destroy_locked(poll_args* args) {
     poll_cache.free_pollers = args->next;
   }
 
-  gpr_free(args);
+  // Now move this args to the dead poller list for later join
+  if (poll_cache.dead_pollers != nullptr) {
+    poll_cache.dead_pollers->prev = args;
+  }
+  args->prev = nullptr;
+  args->next = poll_cache.dead_pollers;
+  poll_cache.dead_pollers = args;
+}
+
+static void cache_harvest_locked() {
+  while (poll_cache.dead_pollers) {
+    poll_args* args = poll_cache.dead_pollers;
+    poll_cache.dead_pollers = poll_cache.dead_pollers->next;
+    gpr_thd_join(args->poller_thd);
+    gpr_free(args);
+  }
 }
 
 static void decref_poll_result(poll_result* res) {
@@ -1469,6 +1489,7 @@ static void run_poll(void* args) {
     poll_result* result = pargs->result;
     int retval = g_cvfds.poll(result->fds, result->nfds, CV_POLL_PERIOD_MS);
     gpr_mu_lock(&g_cvfds.mu);
+    cache_harvest_locked();
     if (retval != 0) {
       result->completed = 1;
       result->retval = retval;
@@ -1488,6 +1509,7 @@ static void run_poll(void* args) {
       deadline = gpr_time_add(deadline, thread_grace);
       pargs->trigger_set = 0;
       gpr_cv_wait(&pargs->trigger, &g_cvfds.mu, deadline);
+      cache_harvest_locked();
       if (!pargs->trigger_set) {
         cache_destroy_locked(pargs);
         break;
@@ -1496,7 +1518,6 @@ static void run_poll(void* args) {
     gpr_mu_unlock(&g_cvfds.mu);
   }
 
-  // We still have the lock here
   if (gpr_unref(&g_cvfds.pollcount)) {
     gpr_cv_signal(&g_cvfds.shutdown_cv);
   }
@@ -1512,6 +1533,7 @@ static int cvfd_poll(struct pollfd* fds, nfds_t nfds, int timeout) {
   nfds_t nsockfds = 0;
   poll_result* result = nullptr;
   gpr_mu_lock(&g_cvfds.mu);
+  cache_harvest_locked();
   pollcv = static_cast<grpc_cv_node*>(gpr_malloc(sizeof(grpc_cv_node)));
   pollcv->next = nullptr;
   gpr_cv pollcv_cv;
@@ -1575,12 +1597,14 @@ static int cvfd_poll(struct pollfd* fds, nfds_t nfds, int timeout) {
     pargs->trigger_set = 1;
     gpr_cv_signal(&pargs->trigger);
     gpr_cv_wait(&pollcv_cv, &g_cvfds.mu, deadline);
+    cache_harvest_locked();
     res = result->retval;
     errno = result->err;
     result->watchcount--;
     remove_cvn(&result->watchers, pollcv);
   } else if (!skip_poll) {
     gpr_cv_wait(&pollcv_cv, &g_cvfds.mu, deadline);
+    cache_harvest_locked();
   }
 
   idx = 0;
@@ -1637,6 +1661,7 @@ static void global_cv_fd_table_init() {
   for (unsigned int i = 0; i < poll_cache.size; i++) {
     poll_cache.active_pollers[i] = nullptr;
   }
+  poll_cache.dead_pollers = nullptr;
 
   gpr_mu_unlock(&g_cvfds.mu);
 }
@@ -1655,6 +1680,7 @@ static void global_cv_fd_table_shutdown() {
   grpc_poll_function = g_cvfds.poll;
   gpr_free(g_cvfds.cvfds);
 
+  cache_harvest_locked();
   gpr_free(poll_cache.active_pollers);
 
   gpr_mu_unlock(&g_cvfds.mu);