Move assignment for Thread, make destructor optional, loop cv waits

3 files changed, 19 insertions, 14 deletions

diff --git a/src/core/lib/iomgr/ev_poll_posix.cc b/src/core/lib/iomgr/ev_poll_posix.cc
index 4fc8ce9ece..6120f9f44b 100644
--- a/src/core/lib/iomgr/ev_poll_posix.cc
+++ b/src/core/lib/iomgr/ev_poll_posix.cc
@@ -31,7 +31,6 @@
 #include <string.h>
 #include <sys/socket.h>
 #include <unistd.h>
-#include <new>
 
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
@@ -259,7 +258,9 @@ typedef struct poll_args {
   grpc_core::Thread poller_thd;
   gpr_cv trigger;
   int trigger_set;
+  bool harvestable;
   gpr_cv harvest;
+  bool joinable;
   gpr_cv join;
   struct pollfd* fds;
   nfds_t nfds;
@@ -1372,6 +1373,8 @@ static poll_args* get_poller_locked(struct pollfd* fds, nfds_t count) {
   gpr_cv_init(&pargs->trigger);
   gpr_cv_init(&pargs->harvest);
   gpr_cv_init(&pargs->join);
+  pargs->harvestable = false;
+  pargs->joinable = false;
   pargs->fds = fds;
   pargs->nfds = count;
   pargs->next = nullptr;
@@ -1380,7 +1383,7 @@ static poll_args* get_poller_locked(struct pollfd* fds, nfds_t count) {
   init_result(pargs);
   cache_poller_locked(pargs);
   gpr_ref(&g_cvfds.pollcount);
-  new (&pargs->poller_thd) grpc_core::Thread("grpc_poller", &run_poll, pargs);
+  pargs->poller_thd = grpc_core::Thread("grpc_poller", &run_poll, pargs);
   pargs->poller_thd.Start();
   return pargs;
 }
@@ -1464,10 +1467,13 @@ static void cache_harvest_locked() {
     if (poll_cache.dead_pollers) {
       poll_cache.dead_pollers->prev = nullptr;
     }
+    args->harvestable = true;
     gpr_cv_signal(&args->harvest);
-    gpr_cv_wait(&args->join, &g_cvfds.mu, gpr_inf_future(GPR_CLOCK_MONOTONIC));
+    while (!args->joinable) {
+      gpr_cv_wait(&args->join, &g_cvfds.mu,
+                  gpr_inf_future(GPR_CLOCK_MONOTONIC));
+    }
     args->poller_thd.Join();
-    args->poller_thd.~Thread();
     gpr_free(args);
   }
 }
@@ -1533,8 +1539,11 @@ static void run_poll(void* args) {
   if (gpr_unref(&g_cvfds.pollcount)) {
     gpr_cv_signal(&g_cvfds.shutdown_cv);
   }
-  gpr_cv_wait(&pargs->harvest, &g_cvfds.mu,
-              gpr_inf_future(GPR_CLOCK_MONOTONIC));
+  while (!pargs->harvestable) {
+    gpr_cv_wait(&pargs->harvest, &g_cvfds.mu,
+                gpr_inf_future(GPR_CLOCK_MONOTONIC));
+  }
+  pargs->joinable = true;
   gpr_cv_signal(&pargs->join);
   gpr_mu_unlock(&g_cvfds.mu);
 }
diff --git a/src/core/lib/iomgr/executor.cc b/src/core/lib/iomgr/executor.cc
index 74e530e898..b017db53f8 100644
--- a/src/core/lib/iomgr/executor.cc
+++ b/src/core/lib/iomgr/executor.cc
@@ -21,7 +21,6 @@
 #include "src/core/lib/iomgr/executor.h"
 
 #include <string.h>
-#include <new>
 
 #include <grpc/support/alloc.h>
 #include <grpc/support/cpu.h>
@@ -102,11 +101,11 @@ void grpc_executor_set_threading(bool threading) {
     for (size_t i = 0; i < g_max_threads; i++) {
       gpr_mu_init(&g_thread_state[i].mu);
       gpr_cv_init(&g_thread_state[i].cv);
-      new (&g_thread_state[i].thd) grpc_core::Thread();
+      g_thread_state[i].thd = grpc_core::Thread();
       g_thread_state[i].elems = GRPC_CLOSURE_LIST_INIT;
     }
 
-    new (&g_thread_state[0].thd)
+    g_thread_state[0].thd =
         grpc_core::Thread("grpc_executor", executor_thread, &g_thread_state[0]);
     g_thread_state[0].thd.Start();
   } else {
@@ -126,7 +125,6 @@ void grpc_executor_set_threading(bool threading) {
     }
     gpr_atm_no_barrier_store(&g_cur_threads, 0);
     for (size_t i = 0; i < g_max_threads; i++) {
-      g_thread_state[i].thd.~Thread();
       gpr_mu_destroy(&g_thread_state[i].mu);
       gpr_cv_destroy(&g_thread_state[i].cv);
       run_closures(g_thread_state[i].elems);
@@ -266,7 +264,7 @@ static void executor_push(grpc_closure* closure, grpc_error* error,
       if (cur_thread_count < g_max_threads) {
         gpr_atm_no_barrier_store(&g_cur_threads, cur_thread_count + 1);
 
-        new (&g_thread_state[cur_thread_count].thd)
+        g_thread_state[cur_thread_count].thd =
             grpc_core::Thread("grpc_executor", executor_thread,
                               &g_thread_state[cur_thread_count]);
         g_thread_state[cur_thread_count].thd.Start();
diff --git a/src/core/lib/iomgr/timer_manager.cc b/src/core/lib/iomgr/timer_manager.cc
index 7efbaa8364..94f288af27 100644
--- a/src/core/lib/iomgr/timer_manager.cc
+++ b/src/core/lib/iomgr/timer_manager.cc
@@ -19,7 +19,6 @@
 #include <grpc/support/port_platform.h>
 
 #include <inttypes.h>
-#include <new>
 
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
@@ -69,7 +68,6 @@ static void gc_completed_threads(void) {
     gpr_mu_unlock(&g_mu);
     while (to_gc != nullptr) {
       to_gc->thd.Join();
-      to_gc->thd.~Thread();
       completed_thread* next = to_gc->next;
       gpr_free(to_gc);
       to_gc = next;
@@ -88,7 +86,7 @@ static void start_timer_thread_and_unlock(void) {
   }
   completed_thread* ct =
       static_cast<completed_thread*>(gpr_malloc(sizeof(*ct)));
-  new (&ct->thd) grpc_core::Thread("grpc_global_timer", timer_thread, ct);
+  ct->thd = grpc_core::Thread("grpc_global_timer", timer_thread, ct);
   ct->thd.Start();
 }