diff --git a/src/ruby/end2end/grpc_class_init_client.rb b/src/ruby/end2end/grpc_class_init_client.rb index 5fad1353be1..ee79292119a 100755 --- a/src/ruby/end2end/grpc_class_init_client.rb +++ b/src/ruby/end2end/grpc_class_init_client.rb @@ -42,40 +42,36 @@ def main end end.parse! + test_proc = nil + case grpc_class when 'channel' - thd = Thread.new do + test_proc = proc do GRPC::Core::Channel.new('dummy_host', nil, :this_channel_is_insecure) end - GRPC::Core::Channel.new('dummy_host', nil, :this_channel_is_insecure) - thd.join when 'server' - thd = Thread.new do + test_proc = proc do GRPC::Core::Server.new({}) end - GRPC::Core::Server.new({}) - thd.join when 'channel_credentials' - thd = Thread.new do + test_proc = proc do GRPC::Core::ChannelCredentials.new end - GRPC::Core::ChannelCredentials.new - thd.join when 'call_credentials' - thd = Thread.new do + test_proc = proc do GRPC::Core::CallCredentials.new(proc { |noop| noop }) end - GRPC::Core::CallCredentials.new(proc { |noop| noop }) - thd.join when 'compression_options' - thd = Thread.new do + test_proc = proc do GRPC::Core::CompressionOptions.new end - GRPC::Core::CompressionOptions.new - thd.join else fail "bad --grpc_class=#{grpc_class} param" end + + th = Thread.new { test_proc.call } + test_proc.call + th.join end main diff --git a/src/ruby/ext/grpc/rb_channel.c b/src/ruby/ext/grpc/rb_channel.c index aead45c082e..fb610f548eb 100644 --- a/src/ruby/ext/grpc/rb_channel.c +++ b/src/ruby/ext/grpc/rb_channel.c @@ -89,6 +89,8 @@ typedef struct grpc_rb_channel { static void grpc_rb_channel_try_register_connection_polling( grpc_rb_channel *wrapper); static void grpc_rb_channel_safe_destroy(grpc_rb_channel *wrapper); +static void *wait_until_channel_polling_thread_started_no_gil(void*); +static void wait_until_channel_polling_thread_started_unblocking_func(void*); static grpc_completion_queue *channel_polling_cq; static gpr_mu global_connection_polling_mu; @@ -169,6 +171,8 @@ static VALUE grpc_rb_channel_init(int argc, VALUE *argv, VALUE self) { MEMZERO(&args, grpc_channel_args, 1); grpc_ruby_once_init(); + rb_thread_call_without_gvl(wait_until_channel_polling_thread_started_no_gil, NULL, + wait_until_channel_polling_thread_started_unblocking_func, NULL); /* "3" == 3 mandatory args */ rb_scan_args(argc, argv, "3", &target, &channel_args, &credentials); @@ -440,6 +444,7 @@ static void grpc_rb_channel_try_register_connection_polling( } gpr_mu_lock(&global_connection_polling_mu); + GPR_ASSERT(channel_polling_thread_started || abort_channel_polling); conn_state = grpc_channel_check_connectivity_state(wrapper->wrapped, 0); if (conn_state != wrapper->current_connectivity_state) { wrapper->current_connectivity_state = conn_state; @@ -473,7 +478,7 @@ static void grpc_rb_channel_safe_destroy(grpc_rb_channel *wrapper) { } // Note this loop breaks out with a single call of -// "grpc_rb_event_unblocking_func". +// "run_poll_channels_loop_no_gil". // This assumes that a ruby call the unblocking func // indicates process shutdown. // In the worst case, this stops polling channel connectivity @@ -486,7 +491,7 @@ static void *run_poll_channels_loop_no_gil(void *arg) { gpr_mu_lock(&global_connection_polling_mu); GPR_ASSERT(!channel_polling_thread_started); channel_polling_thread_started = 1; - gpr_cv_signal(&global_connection_polling_cv); + gpr_cv_broadcast(&global_connection_polling_cv); gpr_mu_unlock(&global_connection_polling_mu); for (;;) { @@ -505,10 +510,10 @@ static void *run_poll_channels_loop_no_gil(void *arg) { } // Notify the channel polling loop to cleanup and shutdown. -static void grpc_rb_event_unblocking_func(void *arg) { +static void run_poll_channels_loop_unblocking_func(void *arg) { (void)arg; gpr_mu_lock(&global_connection_polling_mu); - gpr_log(GPR_DEBUG, "GRPC_RUBY: grpc_rb_event_unblocking_func - begin aborting connection polling"); + gpr_log(GPR_DEBUG, "GRPC_RUBY: run_poll_channels_loop_unblocking_func - begin aborting connection polling"); abort_channel_polling = 1; grpc_completion_queue_shutdown(channel_polling_cq); gpr_mu_unlock(&global_connection_polling_mu); @@ -519,12 +524,12 @@ static VALUE run_poll_channels_loop(VALUE arg) { (void)arg; gpr_log(GPR_DEBUG, "GRPC_RUBY: run_poll_channels_loop - create connection polling thread"); rb_thread_call_without_gvl(run_poll_channels_loop_no_gil, NULL, - grpc_rb_event_unblocking_func, NULL); + run_poll_channels_loop_unblocking_func, NULL); return Qnil; } -static void *grpc_rb_wait_until_channel_polling_thread_started(void *arg) { +static void *wait_until_channel_polling_thread_started_no_gil(void *arg) { (void)arg; gpr_log(GPR_DEBUG, "GRPC_RUBY: wait for channel polling thread to start"); gpr_mu_lock(&global_connection_polling_mu); @@ -537,6 +542,14 @@ static void *grpc_rb_wait_until_channel_polling_thread_started(void *arg) { return NULL; } +static void wait_until_channel_polling_thread_started_unblocking_func(void* arg) { + (void)arg; + gpr_mu_lock(&global_connection_polling_mu); + gpr_log(GPR_DEBUG, "GRPC_RUBY: wait_until_channel_polling_thread_started_unblocking_func - begin aborting connection polling"); + abort_channel_polling = 1; + gpr_cv_broadcast(&global_connection_polling_cv); + gpr_mu_unlock(&global_connection_polling_mu); +} /* Temporary fix for * https://github.com/GoogleCloudPlatform/google-cloud-ruby/issues/899. @@ -565,14 +578,9 @@ void grpc_rb_channel_polling_thread_start() { gpr_log(GPR_DEBUG, "GRPC_RUBY: failed to spawn channel polling thread"); gpr_mu_lock(&global_connection_polling_mu); abort_channel_polling = 1; + gpr_cv_broadcast(&global_connection_polling_cv); gpr_mu_unlock(&global_connection_polling_mu); - return; } - - // Drop the gil before sleeping on a gpr_cv so that the background thread - // signaling it can acquire the gil and then start, if it hasn't already. - rb_thread_call_without_gvl(grpc_rb_wait_until_channel_polling_thread_started, NULL, - grpc_rb_event_unblocking_func, NULL); } static void Init_grpc_propagate_masks() {