defer grpc_init and background threads until first grpc object init

8 files changed, 40 insertions, 17 deletions

diff --git a/src/ruby/ext/grpc/rb_call_credentials.c b/src/ruby/ext/grpc/rb_call_credentials.c
index 280f21c973..fac2cbb04d 100644
--- a/src/ruby/ext/grpc/rb_call_credentials.c
+++ b/src/ruby/ext/grpc/rb_call_credentials.c
@@ -221,6 +221,8 @@ static VALUE grpc_rb_call_credentials_init(VALUE self, VALUE proc) {
   grpc_call_credentials *creds = NULL;
   grpc_metadata_credentials_plugin plugin;
 
+  grpc_ruby_once_init();
+
   TypedData_Get_Struct(self, grpc_rb_call_credentials,
                        &grpc_rb_call_credentials_data_type, wrapper);
 
@@ -281,8 +283,6 @@ void Init_grpc_call_credentials() {
                    grpc_rb_call_credentials_compose, -1);
 
   id_callback = rb_intern("__callback");
-
-  grpc_rb_event_queue_thread_start();
 }
 
 /* Gets the wrapped grpc_call_credentials from the ruby wrapper */
diff --git a/src/ruby/ext/grpc/rb_channel.c b/src/ruby/ext/grpc/rb_channel.c
index 1c20c8813f..6d12ff9ebd 100644
--- a/src/ruby/ext/grpc/rb_channel.c
+++ b/src/ruby/ext/grpc/rb_channel.c
@@ -166,6 +166,8 @@ static VALUE grpc_rb_channel_init(int argc, VALUE *argv, VALUE self) {
   grpc_channel_args args;
   MEMZERO(&args, grpc_channel_args, 1);
 
+  grpc_ruby_once_init();
+
   /* "3" == 3 mandatory args */
   rb_scan_args(argc, argv, "3", &target, &channel_args, &credentials);
 
@@ -521,10 +523,12 @@ static VALUE run_poll_channels_loop(VALUE arg) {
  * calls - so that c-core can reconnect if needed, when there aren't any RPC's.
  * TODO(apolcyn) remove this when core handles new RPCs on dead connections.
  */
-static void start_poll_channels_loop() {
-  channel_polling_cq = grpc_completion_queue_create(NULL);
+void grpc_rb_channel_polling_thread_start() {
+  GPR_ASSERT(!abort_channel_polling);
+  GPR_ASSERT(channel_polling_cq == NULL);
+
   gpr_mu_init(&global_connection_polling_mu);
-  abort_channel_polling = 0;
+  channel_polling_cq = grpc_completion_queue_create(NULL);
   rb_thread_create(run_poll_channels_loop, NULL);
 }
 
@@ -597,7 +601,6 @@ void Init_grpc_channel() {
   id_insecure_channel = rb_intern("this_channel_is_insecure");
   Init_grpc_propagate_masks();
   Init_grpc_connectivity_states();
-  start_poll_channels_loop();
 }
 
 /* Gets the wrapped channel from the ruby wrapper */
diff --git a/src/ruby/ext/grpc/rb_channel.h b/src/ruby/ext/grpc/rb_channel.h
index 77e1f6acbc..fdceb79bca 100644
--- a/src/ruby/ext/grpc/rb_channel.h
+++ b/src/ruby/ext/grpc/rb_channel.h
@@ -41,6 +41,8 @@
 /* Initializes the Channel class. */
 void Init_grpc_channel();
 
+void grpc_rb_channel_polling_thread_start();
+
 /* Gets the wrapped channel from the ruby wrapper */
 grpc_channel* grpc_rb_get_wrapped_channel(VALUE v);
 
diff --git a/src/ruby/ext/grpc/rb_channel_credentials.c b/src/ruby/ext/grpc/rb_channel_credentials.c
index 5b7aa3417e..d334c09148 100644
--- a/src/ruby/ext/grpc/rb_channel_credentials.c
+++ b/src/ruby/ext/grpc/rb_channel_credentials.c
@@ -156,6 +156,9 @@ static VALUE grpc_rb_channel_credentials_init(int argc, VALUE *argv, VALUE self)
   grpc_ssl_pem_key_cert_pair key_cert_pair;
   const char *pem_root_certs_cstr = NULL;
   MEMZERO(&key_cert_pair, grpc_ssl_pem_key_cert_pair, 1);
+
+  grpc_ruby_once_init();
+
   /* "03" == no mandatory arg, 3 optional */
   rb_scan_args(argc, argv, "03", &pem_root_certs, &pem_private_key,
                &pem_cert_chain);
diff --git a/src/ruby/ext/grpc/rb_compression_options.c b/src/ruby/ext/grpc/rb_compression_options.c
index 6b2467ee46..b23e82b0db 100644
--- a/src/ruby/ext/grpc/rb_compression_options.c
+++ b/src/ruby/ext/grpc/rb_compression_options.c
@@ -100,8 +100,11 @@ static rb_data_type_t grpc_rb_compression_options_data_type = {
    Allocate the wrapped grpc compression options and
    initialize it here too. */
 static VALUE grpc_rb_compression_options_alloc(VALUE cls) {
-  grpc_rb_compression_options *wrapper =
-      gpr_malloc(sizeof(grpc_rb_compression_options));
+  grpc_rb_compression_options *wrapper = NULL;
+
+  grpc_ruby_once_init();
+
+  wrapper = gpr_malloc(sizeof(grpc_rb_compression_options));
   wrapper->wrapped = NULL;
   wrapper->wrapped = gpr_malloc(sizeof(grpc_compression_options));
   grpc_compression_options_init(wrapper->wrapped);
diff --git a/src/ruby/ext/grpc/rb_grpc.c b/src/ruby/ext/grpc/rb_grpc.c
index 17cd165a91..584b5dbc63 100644
--- a/src/ruby/ext/grpc/rb_grpc.c
+++ b/src/ruby/ext/grpc/rb_grpc.c
@@ -50,6 +50,8 @@
 #include "rb_server.h"
 #include "rb_server_credentials.h"
 #include "rb_compression_options.h"
+#include "rb_event_thread.h"
+#include "rb_channel.h"
 
 static VALUE grpc_rb_cTimeVal = Qnil;
 
@@ -291,17 +293,14 @@ VALUE sym_metadata = Qundef;
 
 static gpr_once g_once_init = GPR_ONCE_INIT;
 
-static void grpc_ruby_once_init() {
+static void grpc_ruby_once_init_internal() {
   grpc_init();
+  grpc_rb_event_queue_thread_start();
+  grpc_rb_channel_polling_thread_start();
   atexit(grpc_rb_shutdown);
 }
 
-void Init_grpc_c() {
-  if (!grpc_rb_load_core()) {
-    rb_raise(rb_eLoadError, "Couldn't find or load gRPC's dynamic C core");
-    return;
-  }
-
+void grpc_ruby_once_init() {
   /* ruby_vm_at_exit doesn't seem to be working. It would crash once every
    * blue moon, and some users are getting it repeatedly. See the discussions
    *  - https://github.com/grpc/grpc/pull/5337
@@ -312,7 +311,14 @@ void Init_grpc_c() {
    * then loaded again by another VM within the same process, we need to
    * schedule our initialization and destruction only once.
    */
-  gpr_once_init(&g_once_init, grpc_ruby_once_init);
+  gpr_once_init(&g_once_init, grpc_ruby_once_init_internal);
+}
+
+void Init_grpc_c() {
+  if (!grpc_rb_load_core()) {
+    rb_raise(rb_eLoadError, "Couldn't find or load gRPC's dynamic C core");
+    return;
+  }
 
   grpc_rb_mGRPC = rb_define_module("GRPC");
   grpc_rb_mGrpcCore = rb_define_module_under(grpc_rb_mGRPC, "Core");
diff --git a/src/ruby/ext/grpc/rb_grpc.h b/src/ruby/ext/grpc/rb_grpc.h
index 6ea6cbd0b6..4bf11e75df 100644
--- a/src/ruby/ext/grpc/rb_grpc.h
+++ b/src/ruby/ext/grpc/rb_grpc.h
@@ -82,4 +82,6 @@ VALUE grpc_rb_cannot_init_copy(VALUE copy, VALUE self);
 /* grpc_rb_time_timeval creates a gpr_timespec from a ruby time object. */
 gpr_timespec grpc_rb_time_timeval(VALUE time, int interval);
 
+void grpc_ruby_once_init();
+
 #endif /* GRPC_RB_H_ */
diff --git a/src/ruby/ext/grpc/rb_server.c b/src/ruby/ext/grpc/rb_server.c
index 7b2f5774aa..de79a5121f 100644
--- a/src/ruby/ext/grpc/rb_server.c
+++ b/src/ruby/ext/grpc/rb_server.c
@@ -131,11 +131,15 @@ static VALUE grpc_rb_server_alloc(VALUE cls) {
 
   Initializes server instances. */
 static VALUE grpc_rb_server_init(VALUE self, VALUE channel_args) {
-  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
+  grpc_completion_queue *cq = NULL;
   grpc_rb_server *wrapper = NULL;
   grpc_server *srv = NULL;
   grpc_channel_args args;
   MEMZERO(&args, grpc_channel_args, 1);
+
+  grpc_ruby_once_init();
+
+  cq = grpc_completion_queue_create(NULL);
   TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type,
                        wrapper);
   grpc_rb_hash_convert_to_channel_args(channel_args, &args);