Merge pull request #23625 from apolcyn/remove_sleeps

Fix parent/child process synchronization in two ruby tests

src/ruby/end2end/call_credentials_returning_bad_metadata_doesnt_kill_background_thread_driver.rb

@@ -49,15 +49,6 @@ def create_server_creds
     true) # force client auth
 end
 
-# Useful to update a value within a do block
-class MutableValue
-  attr_accessor :value
-
-  def initialize(value)
-    @value = value
-  end
-end
-
 def run_rpc_expect_unavailable(stub)
   exception = nil
   begin

src/ruby/end2end/call_credentials_timeout_driver.rb

@@ -49,15 +49,6 @@ def create_server_creds
     true) # force client auth
 end
 
-# Useful to update a value within a do block
-class MutableValue
-  attr_accessor :value
-
-  def initialize(value)
-    @value = value
-  end
-end
-
 # rubocop:disable Metrics/AbcSize
 # rubocop:disable Metrics/MethodLength
 def main

src/ruby/end2end/channel_closing_driver.rb

@@ -31,8 +31,17 @@ def main
   sleep 3
 
   begin
-    Timeout.timeout(10) do
+    Timeout.timeout(20) do
-      control_stub.shutdown(ClientControl::Void.new)
+      loop do
+        begin
+          control_stub.shutdown(ClientControl::Void.new)
+          break
+        rescue GRPC::BadStatus => e
+          STDERR.puts "control_stub.shutdown RPC received error:|#{e}|. " \
+          "This could mean that that child process e.g. isn't running yet, " \
+          "so we'll retry the RPC"
+        end
+      end
       Process.wait(client_pid)
     end
   rescue Timeout::Error

src/ruby/end2end/end2end_common.rb

@@ -33,12 +33,47 @@ require_relative '../spec/support/helpers'
 
 include GRPC::Spec::Helpers
 
+# Useful to update a value within a do block
+class MutableValue
+  attr_accessor :value
+
+  def initialize(value)
+    @value = value
+  end
+end
+
 # GreeterServer is simple server that implements the Helloworld Greeter server.
+# This service also has a mechanism to wait for a timeout until the first
+# RPC has been received, which is useful for synchronizing between parent
+# and child processes.
 class EchoServerImpl < Echo::EchoServer::Service
+  def initialize
+    @first_rpc_received_mu = Mutex.new
+    @first_rpc_received_cv = ConditionVariable.new
+    @first_rpc_received = MutableValue.new(false)
+  end
+
   # say_hello implements the SayHello rpc method.
   def echo(echo_req, _)
+    @first_rpc_received_mu.synchronize do
+      @first_rpc_received.value = true
+      @first_rpc_received_cv.broadcast
+    end
     Echo::EchoReply.new(response: echo_req.request)
   end
+
+  def wait_for_first_rpc_received(timeout_seconds)
+    Timeout.timeout(timeout_seconds) do
+      @first_rpc_received_mu.synchronize do
+        until @first_rpc_received.value
+          @first_rpc_received_cv.wait(@first_rpc_received_mu)
+        end
+      end
+    end
+  rescue => e
+    fail "Received error:|#{e}| while waiting for #{timeout_seconds} " \
+         'seconds to receive the first RPC'
+  end
 end
 
 # ServerRunner starts an "echo server" that test clients can make calls to

src/ruby/end2end/graceful_sig_handling_driver.rb

@@ -19,51 +19,23 @@
 
 require_relative './end2end_common'
 
-# A service that calls back it's received_rpc_callback
-# upon receiving an RPC. Used for synchronization/waiting
-# for child process to start.
-class ClientStartedService < Echo::EchoServer::Service
-  def initialize(received_rpc_callback)
-    @received_rpc_callback = received_rpc_callback
-  end
-
-  def echo(echo_req, _)
-    @received_rpc_callback.call unless @received_rpc_callback.nil?
-    @received_rpc_callback = nil
-    Echo::EchoReply.new(response: echo_req.request)
-  end
-end
-
 def main
   STDERR.puts 'start server'
-  client_started = false
+  echo_service = EchoServerImpl.new
-  client_started_mu = Mutex.new
+  server_runner = ServerRunner.new(echo_service)
-  client_started_cv = ConditionVariable.new
-  received_rpc_callback = proc do
-    client_started_mu.synchronize do
-      client_started = true
-      client_started_cv.signal
-    end
-  end
-
-  client_started_service = ClientStartedService.new(received_rpc_callback)
-  server_runner = ServerRunner.new(client_started_service)
   server_port = server_runner.run
   STDERR.puts 'start client'
   control_stub, client_pid = start_client('graceful_sig_handling_client.rb', server_port)
-
+  # use receipt of one RPC to indicate that the child process is
-  client_started_mu.synchronize do
+  # ready
-    client_started_cv.wait(client_started_mu) until client_started
+  echo_service.wait_for_first_rpc_received(20)
-  end
+  # now get the client to send an RPC
-
   control_stub.do_echo_rpc(
     ClientControl::DoEchoRpcRequest.new(request: 'hello'))
-
   STDERR.puts 'killing client'
   Process.kill('SIGINT', client_pid)
   Process.wait(client_pid)
   client_exit_status = $CHILD_STATUS
-
   if client_exit_status.exited?
     if client_exit_status.exitstatus != 0
       STDERR.puts 'Client did not close gracefully'
@@ -75,7 +47,6 @@ def main
   end
 
   STDERR.puts 'Client ended gracefully'
-
   # no need to call cleanup, client should already be dead
   server_runner.stop
 end

src/ruby/end2end/graceful_sig_stop_driver.rb

@@ -19,43 +19,16 @@
 
 require_relative './end2end_common'
 
-# A service that calls back it's received_rpc_callback
-# upon receiving an RPC. Used for synchronization/waiting
-# for child process to start.
-class ClientStartedService < Echo::EchoServer::Service
-  def initialize(received_rpc_callback)
-    @received_rpc_callback = received_rpc_callback
-  end
-
-  def echo(echo_req, _)
-    @received_rpc_callback.call unless @received_rpc_callback.nil?
-    @received_rpc_callback = nil
-    Echo::EchoReply.new(response: echo_req.request)
-  end
-end
-
 def main
   STDERR.puts 'start server'
-  client_started = false
+  echo_service = EchoServerImpl.new
-  client_started_mu = Mutex.new
+  server_runner = ServerRunner.new(echo_service)
-  client_started_cv = ConditionVariable.new
-  received_rpc_callback = proc do
-    client_started_mu.synchronize do
-      client_started = true
-      client_started_cv.signal
-    end
-  end
-
-  client_started_service = ClientStartedService.new(received_rpc_callback)
-  server_runner = ServerRunner.new(client_started_service)
   server_port = server_runner.run
   STDERR.puts 'start client'
   control_stub, client_pid = start_client('./graceful_sig_stop_client.rb', server_port)
-
+  # use receipt of one RPC to indicate that the child process is
-  client_started_mu.synchronize do
+  # ready
-    client_started_cv.wait(client_started_mu) until client_started
+  echo_service.wait_for_first_rpc_received(20)
-  end
-
   cleanup(control_stub, client_pid, server_runner)
 end
 

src/ruby/end2end/sig_handling_driver.rb

@@ -19,43 +19,16 @@
 
 require_relative './end2end_common'
 
-# A service that calls back it's received_rpc_callback
-# upon receiving an RPC. Used for synchronization/waiting
-# for child process to start.
-class ClientStartedService < Echo::EchoServer::Service
-  def initialize(received_rpc_callback)
-    @received_rpc_callback = received_rpc_callback
-  end
-
-  def echo(echo_req, _)
-    @received_rpc_callback.call unless @received_rpc_callback.nil?
-    @received_rpc_callback = nil
-    Echo::EchoReply.new(response: echo_req.request)
-  end
-end
-
 def main
   STDERR.puts 'start server'
-  client_started = false
+  echo_service = EchoServerImpl.new
-  client_started_mu = Mutex.new
+  server_runner = ServerRunner.new(echo_service)
-  client_started_cv = ConditionVariable.new
-  received_rpc_callback = proc do
-    client_started_mu.synchronize do
-      client_started = true
-      client_started_cv.signal
-    end
-  end
-
-  client_started_service = ClientStartedService.new(received_rpc_callback)
-  server_runner = ServerRunner.new(client_started_service)
   server_port = server_runner.run
   STDERR.puts 'start client'
   control_stub, client_pid = start_client('sig_handling_client.rb', server_port)
-
+  # use receipt of one RPC to indicate that the child process is
-  client_started_mu.synchronize do
+  # ready
-    client_started_cv.wait(client_started_mu) until client_started
+  echo_service.wait_for_first_rpc_received(20)
-  end
-
   count = 0
   while count < 5
     control_stub.do_echo_rpc(
@@ -64,7 +37,6 @@ def main
     Process.kill('SIGINT', client_pid)
     count += 1
   end
-
   cleanup(control_stub, client_pid, server_runner)
 end
 

src/ruby/end2end/sig_int_during_channel_watch_client.rb

@@ -34,6 +34,16 @@ def main
 
   trap('SIGINT') { exit 0 }
   STDERR.puts 'sig_int_during_channel_watch_client.rb: SIGINT trap has been set'
+  # First, notify the parent process that we're ready for a SIGINT by sending
+  # an RPC
+  begin
+    stub = Echo::EchoServer::Stub.new(
+      "localhost:#{server_port}", :this_channel_is_insecure)
+    stub.echo(ClientControl::DoEchoRpcRequest.new)
+  rescue => e
+    fail "received error:|#{e}| while sending an RPC to the parent process " \
+         'to indicate that the SIGINT trap has been set'
+  end
 
   thd = Thread.new do
     child_thread_channel = GRPC::Core::Channel.new("localhost:#{server_port}",

src/ruby/end2end/sig_int_during_channel_watch_driver.rb

@@ -21,16 +21,19 @@ require_relative './end2end_common'
 
 def main
   STDERR.puts 'start server'
-  server_runner = ServerRunner.new(EchoServerImpl)
+  echo_service = EchoServerImpl.new
+  server_runner = ServerRunner.new(echo_service)
   server_port = server_runner.run
   STDERR.puts 'start client'
   _, client_pid = start_client('sig_int_during_channel_watch_client.rb',
                                server_port)
+  # use receipt of one RPC to indicate that the child process is
+  # ready for a SIGINT
+  echo_service.wait_for_first_rpc_received(20)
   # give time for the client to get into the middle
   # of a channel state watch call
   sleep 1
   Process.kill('SIGINT', client_pid)
-
   begin
     Timeout.timeout(10) do
       Process.wait(client_pid)
@@ -43,12 +46,10 @@ def main
     raise 'Timed out waiting for client process. It likely hangs when a ' \
       'SIGINT is sent while there is an active connectivity_state call'
   end
-
   client_exit_code = $CHILD_STATUS
   if client_exit_code != 0
     fail "sig_int_during_channel_watch_client failed: #{client_exit_code}"
   end
-
   server_runner.stop
 end