aboutsummaryrefslogtreecommitdiffhomepage
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/ruby/ext/grpc/rb_channel.c33
-rw-r--r--src/ruby/spec/channel_connection_spec.rb19
2 files changed, 24 insertions, 28 deletions
diff --git a/src/ruby/ext/grpc/rb_channel.c b/src/ruby/ext/grpc/rb_channel.c
index 8cd489345d..d143c54d21 100644
--- a/src/ruby/ext/grpc/rb_channel.c
+++ b/src/ruby/ext/grpc/rb_channel.c
@@ -89,7 +89,7 @@ static void grpc_rb_channel_try_register_connection_polling(
static void grpc_rb_channel_safe_destroy(grpc_rb_channel *wrapper);
static grpc_completion_queue *channel_polling_cq;
-static gpr_mu channel_polling_mu;
+static gpr_mu global_connection_polling_mu;
static int abort_channel_polling = 0;
/* Destroys Channel instances. */
@@ -188,13 +188,13 @@ static VALUE grpc_rb_channel_init(int argc, VALUE *argv, VALUE self) {
gpr_mu_lock(&wrapper->channel_mu);
wrapper->current_connectivity_state = grpc_channel_check_connectivity_state(wrapper->wrapped, 0);
- gpr_cv_signal(&wrapper->channel_cv);
- gpr_mu_unlock(&wrapper->channel_mu);
-
- gpr_mu_lock(&wrapper->channel_mu);
wrapper->safe_to_destroy = 0;
wrapper->request_safe_destroy = 0;
+
+ gpr_cv_signal(&wrapper->channel_cv);
gpr_mu_unlock(&wrapper->channel_mu);
+
+
grpc_rb_channel_try_register_connection_polling(wrapper);
if (args.args != NULL) {
@@ -277,7 +277,6 @@ static VALUE grpc_rb_channel_watch_connectivity_state(VALUE self,
}
if (wrapper->safe_to_destroy) {
gpr_mu_unlock(&wrapper->channel_mu);
- gpr_log(GPR_DEBUG, "GRPC_RUBY_RB_CHANNEL: attempt to watch_connectivity_state on a non-state-polled channel");
return Qfalse;
}
gpr_cv_wait(&wrapper->channel_cv, &wrapper->channel_mu, grpc_rb_time_timeval(deadline, /* absolute time */ 0));
@@ -394,7 +393,7 @@ static VALUE grpc_rb_channel_get_target(VALUE self) {
// destroy.
// Not safe to call while a channel's connection state is polled.
static void grpc_rb_channel_try_register_connection_polling(
- grpc_rb_channel *wrapper) {
+ grpc_rb_channel *wrapper) {
grpc_connectivity_state conn_state;
gpr_timespec sleep_time = gpr_time_add(
gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(20, GPR_TIMESPAN));
@@ -408,9 +407,8 @@ static void grpc_rb_channel_try_register_connection_polling(
gpr_mu_unlock(&wrapper->channel_mu);
return;
}
- gpr_mu_lock(&channel_polling_mu);
+ gpr_mu_lock(&global_connection_polling_mu);
- gpr_mu_lock(&wrapper->channel_mu);
conn_state = grpc_channel_check_connectivity_state(wrapper->wrapped, 0);
if (conn_state != wrapper->current_connectivity_state) {
wrapper->current_connectivity_state = conn_state;
@@ -424,8 +422,7 @@ static void grpc_rb_channel_try_register_connection_polling(
wrapper->safe_to_destroy = 1;
gpr_cv_signal(&wrapper->channel_cv);
}
- gpr_mu_unlock(&wrapper->channel_mu);
- gpr_mu_unlock(&channel_polling_mu);
+ gpr_mu_unlock(&global_connection_polling_mu);
gpr_mu_unlock(&wrapper->channel_mu);
}
@@ -454,7 +451,6 @@ static void grpc_rb_channel_safe_destroy(grpc_rb_channel *wrapper) {
// early and falls back to current behavior.
static void *run_poll_channels_loop_no_gil(void *arg) {
grpc_event event;
- grpc_rb_channel *wrapper;
(void)arg;
for (;;) {
event = grpc_completion_queue_next(
@@ -463,27 +459,28 @@ static void *run_poll_channels_loop_no_gil(void *arg) {
break;
}
if (event.type == GRPC_OP_COMPLETE) {
- wrapper = (grpc_rb_channel *)event.tag;
-
- grpc_rb_channel_try_register_connection_polling(wrapper);
+ grpc_rb_channel_try_register_connection_polling((grpc_rb_channel *)event.tag);
}
}
grpc_completion_queue_destroy(channel_polling_cq);
+ gpr_log(GPR_DEBUG, "GRPC_RUBY: run_poll_channels_loop_no_gil - exit connection polling loop");
return NULL;
}
// Notify the channel polling loop to cleanup and shutdown.
static void grpc_rb_event_unblocking_func(void *arg) {
(void)arg;
- gpr_mu_lock(&channel_polling_mu);
+ gpr_mu_lock(&global_connection_polling_mu);
+ gpr_log(GPR_DEBUG, "GRPC_RUBY: grpc_rb_event_unblocking_func - begin aborting connection polling");
abort_channel_polling = 1;
grpc_completion_queue_shutdown(channel_polling_cq);
- gpr_mu_unlock(&channel_polling_mu);
+ gpr_mu_unlock(&global_connection_polling_mu);
}
// Poll channel connectivity states in background thread without the GIL.
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);
return Qnil;
@@ -501,7 +498,7 @@ static VALUE run_poll_channels_loop(VALUE arg) {
*/
static void start_poll_channels_loop() {
channel_polling_cq = grpc_completion_queue_create(NULL);
- gpr_mu_init(&channel_polling_mu);
+ gpr_mu_init(&global_connection_polling_mu);
abort_channel_polling = 0;
rb_thread_create(run_poll_channels_loop, NULL);
}
diff --git a/src/ruby/spec/channel_connection_spec.rb b/src/ruby/spec/channel_connection_spec.rb
index d8e10f7b76..b344052a21 100644
--- a/src/ruby/spec/channel_connection_spec.rb
+++ b/src/ruby/spec/channel_connection_spec.rb
@@ -63,7 +63,7 @@ EchoStub = EchoService.rpc_stub_class
def start_server(port = 0)
@srv = GRPC::RpcServer.new
- server_port = @srv.add_http2_port("0.0.0.0:#{port}", :this_port_is_insecure)
+ server_port = @srv.add_http2_port("localhost:#{port}", :this_port_is_insecure)
@srv.handle(EchoService)
@server_thd = Thread.new { @srv.run }
@srv.wait_till_running
@@ -84,24 +84,25 @@ describe 'channel connection behavior' do
req = EchoMsg.new
expect(stub.an_rpc(req)).to be_a(EchoMsg)
stop_server
- expect { stub.an_rpc(req) }.to raise_error(GRPC::Unavailable)
+ sleep 1
# TODO(apolcyn) grabbing the same port might fail, is this stable enough?
start_server(port)
expect(stub.an_rpc(req)).to be_a(EchoMsg)
stop_server
end
- it 'observably connects and reconnects to transient server when using the channel state API', trial: true do
+ it 'observably connects and reconnects to transient server' \
+ 'when using the channel state API' do
port = start_server
- ch = GRPC::Core::Channel.new("localhost:#{port}", {}, :this_channel_is_insecure)
+ ch = GRPC::Core::Channel.new("localhost:#{port}", {},
+ :this_channel_is_insecure)
expect(ch.connectivity_state).to be(GRPC::Core::ConnectivityStates::IDLE)
state = ch.connectivity_state(true)
count = 0
- while count < 20 and state != GRPC::Core::ConnectivityStates::READY do
- STDERR.puts "first round of waiting for state to become READY"
+ while count < 20 && state != GRPC::Core::ConnectivityStates::READY
ch.watch_connectivity_state(state, Time.now + 60)
state = ch.connectivity_state(true)
count += 1
@@ -114,8 +115,7 @@ describe 'channel connection behavior' do
state = ch.connectivity_state
count = 0
- while count < 20 and state == GRPC::Core::ConnectivityStates::READY do
- STDERR.puts "server shut down. waiting for state to not be READY"
+ while count < 20 && state == GRPC::Core::ConnectivityStates::READY
ch.watch_connectivity_state(state, Time.now + 60)
state = ch.connectivity_state
count += 1
@@ -128,8 +128,7 @@ describe 'channel connection behavior' do
state = ch.connectivity_state(true)
count = 0
- while count < 20 and state != GRPC::Core::ConnectivityStates::READY do
- STDERR.puts "second round of waiting for state to become READY"
+ while count < 20 && state != GRPC::Core::ConnectivityStates::READY
ch.watch_connectivity_state(state, Time.now + 60)
state = ch.connectivity_state(true)
count += 1
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-30 06:47:34 +0000