Move grpc_shutdown internals to a detached thread

include/grpc/grpc.h

@@ -73,7 +73,9 @@ GRPCAPI void grpc_init(void);
     Before it's called, there should haven been a matching invocation to
     grpc_init().
 
-    No memory is used by grpc after this call returns, nor are any instructions
+    The last call to grpc_shutdown will initiate cleaning up of grpc library
+    internals, which can happen in another thread. Once the clean-up is done,
+    no memory is used by grpc after this call returns, nor are any instructions
     executing within the grpc library.
     Prior to calling, all application owned grpc objects must have been
     destroyed. */

src/core/lib/gprpp/thd.h

@@ -47,6 +47,26 @@ class ThreadInternalsInterface {
 
 class Thread {
  public:
+  class Options {
+   public:
+    Options() : joinable_(true), tracked_(true) {}
+    Options& set_joinable(bool joinable) {
+      joinable_ = joinable;
+      return *this;
+    }
+    Options& set_tracked(bool tracked) {
+      tracked_ = tracked;
+      return *this;
+    }
+    bool joinable() const { return joinable_; }
+    bool tracked() const { return tracked_; }
+
+   private:
+    bool joinable_;
+    // Whether this thread is tracked by grpc internals. Should be true for most
+    // of threads.
+    bool tracked_;
+  };
   /// Default constructor only to allow use in structs that lack constructors
   /// Does not produce a validly-constructed thread; must later
   /// use placement new to construct a real thread. Does not init mu_ and cv_
@@ -57,14 +77,17 @@ class Thread {
   /// with argument \a arg once it is started.
   /// The optional \a success argument indicates whether the thread
   /// is successfully created.
+  /// The optional \a options can be used to set the thread detachable.
   Thread(const char* thd_name, void (*thd_body)(void* arg), void* arg,
-         bool* success = nullptr);
+         bool* success = nullptr, const Options& options = Options());
 
   /// Move constructor for thread. After this is called, the other thread
   /// no longer represents a living thread object
-  Thread(Thread&& other) : state_(other.state_), impl_(other.impl_) {
+  Thread(Thread&& other)
+      : state_(other.state_), impl_(other.impl_), options_(other.options_) {
     other.state_ = MOVED;
     other.impl_ = nullptr;
+    other.options_ = Options();
   }
 
   /// Move assignment operator for thread. After this is called, the other
@@ -79,8 +102,10 @@ class Thread {
       // assert it for the time being.
       state_ = other.state_;
       impl_ = other.impl_;
+      options_ = other.options_;
       other.state_ = MOVED;
       other.impl_ = nullptr;
+      other.options_ = Options();
     }
     return *this;
   }
@@ -95,11 +120,16 @@ class Thread {
       GPR_ASSERT(state_ == ALIVE);
       state_ = STARTED;
       impl_->Start();
+      if (!options_.joinable()) {
+        state_ = DONE;
+        impl_ = nullptr;
+      }
     } else {
       GPR_ASSERT(state_ == FAILED);
     }
-  };
+  }
 
+  // It is only legal to call Join if the Thread is created as joinable.
   void Join() {
     if (impl_ != nullptr) {
       impl_->Join();
@@ -119,12 +149,13 @@ class Thread {
   /// FAKE -- just a dummy placeholder Thread created by the default constructor
   /// ALIVE -- an actual thread of control exists associated with this thread
   /// STARTED -- the thread of control has been started
-  /// DONE -- the thread of control has completed and been joined
+  /// DONE -- the thread of control has completed and been joined/detached
   /// FAILED -- the thread of control never came alive
   /// MOVED -- contents were moved out and we're no longer tracking them
   enum ThreadState { FAKE, ALIVE, STARTED, DONE, FAILED, MOVED };
   ThreadState state_;
   internal::ThreadInternalsInterface* impl_;
+  Options options_;
 };
 
 }  // namespace grpc_core

src/core/lib/gprpp/thd_posix.cc

@@ -44,13 +44,15 @@ struct thd_arg {
   void (*body)(void* arg); /* body of a thread */
   void* arg;               /* argument to a thread */
   const char* name;        /* name of thread. Can be nullptr. */
+  bool joinable;
+  bool tracked;
 };
 
 class ThreadInternalsPosix
     : public grpc_core::internal::ThreadInternalsInterface {
  public:
   ThreadInternalsPosix(const char* thd_name, void (*thd_body)(void* arg),
-                       void* arg, bool* success)
+                       void* arg, bool* success, const Thread::Options& options)
       : started_(false) {
     gpr_mu_init(&mu_);
     gpr_cv_init(&ready_);
@@ -63,11 +65,20 @@ class ThreadInternalsPosix
     info->body = thd_body;
     info->arg = arg;
     info->name = thd_name;
+    info->joinable = options.joinable();
+    info->tracked = options.tracked();
+    if (options.tracked()) {
       grpc_core::Fork::IncThreadCount();
+    }
 
     GPR_ASSERT(pthread_attr_init(&attr) == 0);
+    if (options.joinable()) {
       GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE) ==
                  0);
+    } else {
+      GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED) ==
+                 0);
+    }
 
     *success =
         (pthread_create(&pthread_id_, &attr,
@@ -98,7 +109,12 @@ class ThreadInternalsPosix
                           gpr_mu_unlock(&arg.thread->mu_);
 
                           (*arg.body)(arg.arg);
+                          if (arg.tracked) {
                             grpc_core::Fork::DecThreadCount();
+                          }
+                          if (!arg.joinable) {
+                            grpc_core::Delete(arg.thread);
+                          }
                           return nullptr;
                         },
                         info) == 0);
@@ -108,9 +124,11 @@ class ThreadInternalsPosix
     if (!(*success)) {
       /* don't use gpr_free, as this was allocated using malloc (see above) */
       free(info);
+      if (options.tracked()) {
         grpc_core::Fork::DecThreadCount();
       }
-  };
+    }
+  }
 
   ~ThreadInternalsPosix() override {
     gpr_mu_destroy(&mu_);
@@ -136,10 +154,11 @@ class ThreadInternalsPosix
 }  // namespace
 
 Thread::Thread(const char* thd_name, void (*thd_body)(void* arg), void* arg,
-               bool* success) {
+               bool* success, const Options& options)
+    : options_(options) {
   bool outcome = false;
-  impl_ =
+  impl_ = grpc_core::New<ThreadInternalsPosix>(thd_name, thd_body, arg,
-      grpc_core::New<ThreadInternalsPosix>(thd_name, thd_body, arg, &outcome);
+                                               &outcome, options);
   if (outcome) {
     state_ = ALIVE;
   } else {

src/core/lib/iomgr/fork_posix.cc

@@ -35,6 +35,7 @@
 #include "src/core/lib/iomgr/executor.h"
 #include "src/core/lib/iomgr/timer_manager.h"
 #include "src/core/lib/iomgr/wakeup_fd_posix.h"
+#include "src/core/lib/surface/init.h"
 
 /*
  * NOTE: FORKING IS NOT GENERALLY SUPPORTED, THIS IS ONLY INTENDED TO WORK
@@ -47,11 +48,12 @@ bool registered_handlers = false;
 }  // namespace
 
 void grpc_prefork() {
-  grpc_core::ExecCtx exec_ctx;
+  grpc_maybe_wait_for_async_shutdown();
-  skipped_handler = true;
   if (!grpc_is_initialized()) {
     return;
   }
+  grpc_core::ExecCtx exec_ctx;
+  skipped_handler = true;
   if (!grpc_core::Fork::Enabled()) {
     gpr_log(GPR_ERROR,
             "Fork support not enabled; try running with the "

src/core/lib/surface/init.cc

@@ -61,10 +61,15 @@ extern void grpc_register_built_in_plugins(void);
 static gpr_once g_basic_init = GPR_ONCE_INIT;
 static gpr_mu g_init_mu;
 static int g_initializations;
+static gpr_cv* g_shutting_down_cv;
+static bool g_shutting_down;
 
 static void do_basic_init(void) {
   gpr_log_verbosity_init();
   gpr_mu_init(&g_init_mu);
+  g_shutting_down_cv = static_cast<gpr_cv*>(malloc(sizeof(gpr_cv)));
+  gpr_cv_init(g_shutting_down_cv);
+  g_shutting_down = false;
   grpc_register_built_in_plugins();
   grpc_cq_global_init();
   g_initializations = 0;
@@ -120,6 +125,10 @@ void grpc_init(void) {
 
   gpr_mu_lock(&g_init_mu);
   if (++g_initializations == 1) {
+    if (g_shutting_down) {
+      g_shutting_down = false;
+      gpr_cv_broadcast(g_shutting_down_cv);
+    }
     grpc_core::Fork::GlobalInit();
     grpc_fork_handlers_auto_register();
     gpr_time_init();
@@ -154,16 +163,20 @@ void grpc_init(void) {
   GRPC_API_TRACE("grpc_init(void)", 0, ());
 }
 
-void grpc_shutdown(void) {
+void grpc_shutdown_internal(void* ignored) {
   int i;
-  GRPC_API_TRACE("grpc_shutdown(void)", 0, ());
+  GRPC_API_TRACE("grpc_shutdown_internal", 0, ());
   gpr_mu_lock(&g_init_mu);
-  if (--g_initializations == 0) {
+  // We have released lock from the shutdown thread and it is possible that
+  // another grpc_init has been called, and do nothing if that is the case.
+  if (--g_initializations != 0) {
+    gpr_mu_unlock(&g_init_mu);
+    return;
+  }
   {
     grpc_core::ExecCtx exec_ctx(0);
     {
-        grpc_timer_manager_set_threading(
+      grpc_timer_manager_set_threading(false);  // shutdown timer_manager thread
-            false);  // shutdown timer_manager thread
       grpc_executor_shutdown();
       for (i = g_number_of_plugins; i >= 0; i--) {
         if (g_all_of_the_plugins[i].destroy != nullptr) {
@@ -182,6 +195,23 @@ void grpc_shutdown(void) {
     grpc_core::Fork::GlobalShutdown();
   }
   grpc_core::ExecCtx::GlobalShutdown();
+  g_shutting_down = false;
+  gpr_cv_broadcast(g_shutting_down_cv);
+  gpr_mu_unlock(&g_init_mu);
+}
+
+void grpc_shutdown(void) {
+  GRPC_API_TRACE("grpc_shutdown(void)", 0, ());
+  gpr_mu_lock(&g_init_mu);
+  if (--g_initializations == 0) {
+    g_initializations++;
+    g_shutting_down = true;
+    // spawn a detached thread to do the actual clean up in case we are
+    // currently in an executor thread.
+    grpc_core::Thread cleanup_thread(
+        "grpc_shutdown", grpc_shutdown_internal, nullptr, nullptr,
+        grpc_core::Thread::Options().set_joinable(false).set_tracked(false));
+    cleanup_thread.Start();
   }
   gpr_mu_unlock(&g_init_mu);
 }
@@ -194,3 +224,13 @@ int grpc_is_initialized(void) {
   gpr_mu_unlock(&g_init_mu);
   return r;
 }
+
+void grpc_maybe_wait_for_async_shutdown(void) {
+  gpr_once_init(&g_basic_init, do_basic_init);
+  gpr_mu_lock(&g_init_mu);
+  while (g_shutting_down) {
+    gpr_cv_wait(g_shutting_down_cv, &g_init_mu,
+                gpr_inf_future(GPR_CLOCK_REALTIME));
+  }
+  gpr_mu_unlock(&g_init_mu);
+}

src/core/lib/surface/init.h

@@ -22,5 +22,6 @@
 void grpc_register_security_filters(void);
 void grpc_security_pre_init(void);
 void grpc_security_init(void);
+void grpc_maybe_wait_for_async_shutdown(void);
 
 #endif /* GRPC_CORE_LIB_SURFACE_INIT_H */

test/core/end2end/fuzzers/client_fuzzer.cc

@@ -40,9 +40,8 @@ static void dont_log(gpr_log_func_args* args) {}
 
 extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
   grpc_test_only_set_slice_hash_seed(0);
-  struct grpc_memory_counters counters;
   if (squelch) gpr_set_log_function(dont_log);
-  if (leak_check) grpc_memory_counters_init();
+  grpc_core::testing::LeakDetector leak_detector(leak_check);
   grpc_init();
   {
     grpc_core::ExecCtx exec_ctx;
@@ -160,10 +159,5 @@ extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
     }
   }
   grpc_shutdown();
-  if (leak_check) {
-    counters = grpc_memory_counters_snapshot();
-    grpc_memory_counters_destroy();
-    GPR_ASSERT(counters.total_size_relative == 0);
-  }
   return 0;
 }

test/core/end2end/fuzzers/server_fuzzer.cc

@@ -37,9 +37,8 @@ static void dont_log(gpr_log_func_args* args) {}
 
 extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
   grpc_test_only_set_slice_hash_seed(0);
-  struct grpc_memory_counters counters;
   if (squelch) gpr_set_log_function(dont_log);
-  if (leak_check) grpc_memory_counters_init();
+  grpc_core::testing::LeakDetector leak_detector(leak_check);
   grpc_init();
   {
     grpc_core::ExecCtx exec_ctx;
@@ -136,10 +135,5 @@ extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
     grpc_completion_queue_destroy(cq);
   }
   grpc_shutdown();
-  if (leak_check) {
-    counters = grpc_memory_counters_snapshot();
-    grpc_memory_counters_destroy();
-    GPR_ASSERT(counters.total_size_relative == 0);
-  }
   return 0;
 }

test/core/security/alts_credentials_fuzzer.cc

@@ -66,10 +66,7 @@ extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
     gpr_set_log_function(dont_log);
   }
   gpr_free(grpc_trace_fuzzer);
-  struct grpc_memory_counters counters;
+  grpc_core::testing::LeakDetector leak_detector(leak_check);
-  if (leak_check) {
-    grpc_memory_counters_init();
-  }
   input_stream inp = {data, data + size};
   grpc_init();
   bool is_on_gcp = grpc_alts_is_running_on_gcp();
@@ -111,10 +108,5 @@ extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
     gpr_free(handshaker_service_url);
   }
   grpc_shutdown();
-  if (leak_check) {
-    counters = grpc_memory_counters_snapshot();
-    grpc_memory_counters_destroy();
-    GPR_ASSERT(counters.total_size_relative == 0);
-  }
   return 0;
 }

test/core/slice/percent_encode_fuzzer.cc

@@ -31,9 +31,8 @@ bool squelch = true;
 bool leak_check = true;
 
 static void test(const uint8_t* data, size_t size, const uint8_t* dict) {
-  struct grpc_memory_counters counters;
+  grpc_core::testing::LeakDetector leak_detector(true);
   grpc_init();
-  grpc_memory_counters_init();
   grpc_slice input =
       grpc_slice_from_copied_buffer(reinterpret_cast<const char*>(data), size);
   grpc_slice output = grpc_percent_encode_slice(input, dict);
@@ -49,10 +48,7 @@ static void test(const uint8_t* data, size_t size, const uint8_t* dict) {
   grpc_slice_unref(output);
   grpc_slice_unref(decoded_output);
   grpc_slice_unref(permissive_decoded_output);
-  counters = grpc_memory_counters_snapshot();
-  grpc_memory_counters_destroy();
   grpc_shutdown();
-  GPR_ASSERT(counters.total_size_relative == 0);
 }
 
 extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {

test/core/surface/init_test.cc

@@ -18,6 +18,9 @@
 
 #include <grpc/grpc.h>
 #include <grpc/support/log.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/surface/init.h"
 #include "test/core/util/test_config.h"
 
 static int g_flag;
@@ -30,6 +33,7 @@ static void test(int rounds) {
   for (i = 0; i < rounds; i++) {
     grpc_shutdown();
   }
+  grpc_maybe_wait_for_async_shutdown();
 }
 
 static void test_mixed(void) {
@@ -39,6 +43,7 @@ static void test_mixed(void) {
   grpc_init();
   grpc_shutdown();
   grpc_shutdown();
+  grpc_maybe_wait_for_async_shutdown();
 }
 
 static void plugin_init(void) { g_flag = 1; }
@@ -49,6 +54,7 @@ static void test_plugin() {
   grpc_init();
   GPR_ASSERT(g_flag == 1);
   grpc_shutdown();
+  grpc_maybe_wait_for_async_shutdown();
   GPR_ASSERT(g_flag == 2);
 }
 
@@ -57,6 +63,7 @@ static void test_repeatedly() {
     grpc_init();
     grpc_shutdown();
   }
+  grpc_maybe_wait_for_async_shutdown();
 }
 
 int main(int argc, char** argv) {

test/core/util/BUILD

@@ -31,7 +31,10 @@ grpc_cc_library(
         "memory_counters.h",
         "test_config.h",
     ],
-    deps = ["//:gpr"],
+    deps = [
+        "//:gpr",
+        "//:grpc_common",
+    ],
     data = [
         "lsan_suppressions.txt",
         "tsan_suppressions.txt",

test/core/util/memory_counters.cc

@@ -16,12 +16,17 @@
  *
  */
 
+#include <inttypes.h>
 #include <stdint.h>
 #include <string.h>
 
+#include <grpc/grpc.h>
 #include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
 #include <grpc/support/sync.h>
+#include <grpc/support/time.h>
 
+#include "src/core/lib/surface/init.h"
 #include "test/core/util/memory_counters.h"
 
 static struct grpc_memory_counters g_memory_counters;
@@ -106,3 +111,30 @@ struct grpc_memory_counters grpc_memory_counters_snapshot() {
       NO_BARRIER_LOAD(&g_memory_counters.total_allocs_absolute);
   return counters;
 }
+
+namespace grpc_core {
+namespace testing {
+
+LeakDetector::LeakDetector(bool enable) : enabled_(enable) {
+  if (enabled_) {
+    grpc_memory_counters_init();
+  }
+}
+
+LeakDetector::~LeakDetector() {
+  if (enabled_) {
+    // Wait for grpc_shutdown() to finish its async work.
+    grpc_maybe_wait_for_async_shutdown();
+    struct grpc_memory_counters counters = grpc_memory_counters_snapshot();
+    grpc_memory_counters_snapshot();
+    if (counters.total_size_relative != 0) {
+      gpr_log(GPR_ERROR, "Leaking %" PRIuPTR "bytes",
+              static_cast<uintptr_t>(counters.total_size_relative));
+      GPR_ASSERT(0);
+    }
+    grpc_memory_counters_destroy();
+  }
+}
+
+}  // namespace testing
+}  // namespace grpc_core

test/core/util/memory_counters.h

@@ -32,4 +32,22 @@ void grpc_memory_counters_init();
 void grpc_memory_counters_destroy();
 struct grpc_memory_counters grpc_memory_counters_snapshot();
 
+namespace grpc_core {
+namespace testing {
+
+// At destruction time, it will check there is no memory leak.
+// The object should be created before grpc_init() is called and destroyed after
+// grpc_shutdown() is returned.
+class LeakDetector {
+ public:
+  explicit LeakDetector(bool enable);
+  ~LeakDetector();
+
+ private:
+  const bool enabled_;
+};
+
+}  // namespace testing
+}  // namespace grpc_core
+
 #endif

test/cpp/naming/address_sorting_test.cc

@@ -46,6 +46,7 @@
 #include "src/core/lib/iomgr/iomgr.h"
 #include "src/core/lib/iomgr/resolve_address.h"
 #include "src/core/lib/iomgr/sockaddr_utils.h"
+#include "src/core/lib/surface/init.h"
 #include "test/core/util/port.h"
 #include "test/core/util/test_config.h"
 
@@ -200,7 +201,10 @@ void VerifyLbAddrOutputs(grpc_lb_addresses* lb_addrs,
 class AddressSortingTest : public ::testing::Test {
  protected:
   void SetUp() override { grpc_init(); }
-  void TearDown() override { grpc_shutdown(); }
+  void TearDown() override {
+    grpc_shutdown();
+    grpc_maybe_wait_for_async_shutdown();
+  }
 };
 
 /* Tests for rule 1 */