don't hold the gil while waiting for bg thread to start

pull/10670/head
Alexander Polcyn 8 years ago
parent d1143abaa8
commit deeed7f12b
  1. 26
      src/ruby/end2end/grpc_class_init_client.rb
  2. 32
      src/ruby/ext/grpc/rb_channel.c

@ -42,40 +42,36 @@ def main
end end
end.parse! end.parse!
test_proc = nil
case grpc_class case grpc_class
when 'channel' when 'channel'
thd = Thread.new do test_proc = proc do
GRPC::Core::Channel.new('dummy_host', nil, :this_channel_is_insecure) GRPC::Core::Channel.new('dummy_host', nil, :this_channel_is_insecure)
end end
GRPC::Core::Channel.new('dummy_host', nil, :this_channel_is_insecure)
thd.join
when 'server' when 'server'
thd = Thread.new do test_proc = proc do
GRPC::Core::Server.new({}) GRPC::Core::Server.new({})
end end
GRPC::Core::Server.new({})
thd.join
when 'channel_credentials' when 'channel_credentials'
thd = Thread.new do test_proc = proc do
GRPC::Core::ChannelCredentials.new GRPC::Core::ChannelCredentials.new
end end
GRPC::Core::ChannelCredentials.new
thd.join
when 'call_credentials' when 'call_credentials'
thd = Thread.new do test_proc = proc do
GRPC::Core::CallCredentials.new(proc { |noop| noop }) GRPC::Core::CallCredentials.new(proc { |noop| noop })
end end
GRPC::Core::CallCredentials.new(proc { |noop| noop })
thd.join
when 'compression_options' when 'compression_options'
thd = Thread.new do test_proc = proc do
GRPC::Core::CompressionOptions.new GRPC::Core::CompressionOptions.new
end end
GRPC::Core::CompressionOptions.new
thd.join
else else
fail "bad --grpc_class=#{grpc_class} param" fail "bad --grpc_class=#{grpc_class} param"
end end
th = Thread.new { test_proc.call }
test_proc.call
th.join
end end
main main

@ -89,6 +89,8 @@ typedef struct grpc_rb_channel {
static void grpc_rb_channel_try_register_connection_polling( static void grpc_rb_channel_try_register_connection_polling(
grpc_rb_channel *wrapper); grpc_rb_channel *wrapper);
static void grpc_rb_channel_safe_destroy(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 grpc_completion_queue *channel_polling_cq;
static gpr_mu global_connection_polling_mu; 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); MEMZERO(&args, grpc_channel_args, 1);
grpc_ruby_once_init(); 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 */ /* "3" == 3 mandatory args */
rb_scan_args(argc, argv, "3", &target, &channel_args, &credentials); 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_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); conn_state = grpc_channel_check_connectivity_state(wrapper->wrapped, 0);
if (conn_state != wrapper->current_connectivity_state) { if (conn_state != wrapper->current_connectivity_state) {
wrapper->current_connectivity_state = conn_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 // 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 // This assumes that a ruby call the unblocking func
// indicates process shutdown. // indicates process shutdown.
// In the worst case, this stops polling channel connectivity // 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_mu_lock(&global_connection_polling_mu);
GPR_ASSERT(!channel_polling_thread_started); GPR_ASSERT(!channel_polling_thread_started);
channel_polling_thread_started = 1; 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); gpr_mu_unlock(&global_connection_polling_mu);
for (;;) { for (;;) {
@ -505,10 +510,10 @@ static void *run_poll_channels_loop_no_gil(void *arg) {
} }
// Notify the channel polling loop to cleanup and shutdown. // 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; (void)arg;
gpr_mu_lock(&global_connection_polling_mu); 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; abort_channel_polling = 1;
grpc_completion_queue_shutdown(channel_polling_cq); grpc_completion_queue_shutdown(channel_polling_cq);
gpr_mu_unlock(&global_connection_polling_mu); gpr_mu_unlock(&global_connection_polling_mu);
@ -519,12 +524,12 @@ static VALUE run_poll_channels_loop(VALUE arg) {
(void)arg; (void)arg;
gpr_log(GPR_DEBUG, "GRPC_RUBY: run_poll_channels_loop - create connection polling thread"); 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, 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; 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; (void)arg;
gpr_log(GPR_DEBUG, "GRPC_RUBY: wait for channel polling thread to start"); gpr_log(GPR_DEBUG, "GRPC_RUBY: wait for channel polling thread to start");
gpr_mu_lock(&global_connection_polling_mu); 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; 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 /* Temporary fix for
* https://github.com/GoogleCloudPlatform/google-cloud-ruby/issues/899. * 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_log(GPR_DEBUG, "GRPC_RUBY: failed to spawn channel polling thread");
gpr_mu_lock(&global_connection_polling_mu); gpr_mu_lock(&global_connection_polling_mu);
abort_channel_polling = 1; abort_channel_polling = 1;
gpr_cv_broadcast(&global_connection_polling_cv);
gpr_mu_unlock(&global_connection_polling_mu); 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() { static void Init_grpc_propagate_masks() {

Loading…
Cancel
Save