From 39ac83a49ea73f619edbfd7ebde47f12d67a18f3 Mon Sep 17 00:00:00 2001 From: Tommy Chen Date: Wed, 26 Dec 2018 11:17:48 -0800 Subject: ruby-sigint ready to be merged! --- examples/ruby/greeter_server.rb | 5 +- examples/ruby/route_guide/route_guide_server.rb | 5 +- src/ruby/end2end/graceful_sig_handling_client.rb | 61 ++++++++++++++++ src/ruby/end2end/graceful_sig_handling_driver.rb | 83 ++++++++++++++++++++++ src/ruby/end2end/graceful_sig_stop_client.rb | 78 ++++++++++++++++++++ src/ruby/end2end/graceful_sig_stop_driver.rb | 62 ++++++++++++++++ src/ruby/lib/grpc/generic/rpc_server.rb | 61 ++++++++++++++++ .../helper_scripts/run_ruby_end2end_tests.sh | 2 + 8 files changed, 355 insertions(+), 2 deletions(-) create mode 100755 src/ruby/end2end/graceful_sig_handling_client.rb create mode 100755 src/ruby/end2end/graceful_sig_handling_driver.rb create mode 100755 src/ruby/end2end/graceful_sig_stop_client.rb create mode 100755 src/ruby/end2end/graceful_sig_stop_driver.rb diff --git a/examples/ruby/greeter_server.rb b/examples/ruby/greeter_server.rb index dca61714b8..5290429742 100755 --- a/examples/ruby/greeter_server.rb +++ b/examples/ruby/greeter_server.rb @@ -39,7 +39,10 @@ def main s = GRPC::RpcServer.new s.add_http2_port('0.0.0.0:50051', :this_port_is_insecure) s.handle(GreeterServer) - s.run_till_terminated + # Runs the server with SIGHUP, SIGINT and SIGQUIT signal handlers to + # gracefully shutdown. + # User could also choose to run server via call to run_till_terminated + s.run_till_terminated_or_interrupted([1, 'int', 'SIGQUIT']) end main diff --git a/examples/ruby/route_guide/route_guide_server.rb b/examples/ruby/route_guide/route_guide_server.rb index 5eb268b533..ffcebd8418 100755 --- a/examples/ruby/route_guide/route_guide_server.rb +++ b/examples/ruby/route_guide/route_guide_server.rb @@ -172,7 +172,10 @@ def main s.add_http2_port(port, :this_port_is_insecure) GRPC.logger.info("... running insecurely on #{port}") s.handle(ServerImpl.new(feature_db)) - s.run_till_terminated + # Runs the server with SIGHUP, SIGINT and SIGQUIT signal handlers to + # gracefully shutdown. + # User could also choose to run server via call to run_till_terminated + s.run_till_terminated_or_interrupted([1, 'int', 'SIGQUIT']) end main diff --git a/src/ruby/end2end/graceful_sig_handling_client.rb b/src/ruby/end2end/graceful_sig_handling_client.rb new file mode 100755 index 0000000000..14a67a62cc --- /dev/null +++ b/src/ruby/end2end/graceful_sig_handling_client.rb @@ -0,0 +1,61 @@ +#!/usr/bin/env ruby + +# Copyright 2015 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +require_relative './end2end_common' + +# Test client. Sends RPC's as normal but process also has signal handlers +class SigHandlingClientController < ClientControl::ClientController::Service + def initialize(stub) + @stub = stub + end + + def do_echo_rpc(req, _) + response = @stub.echo(Echo::EchoRequest.new(request: req.request)) + fail 'bad response' unless response.response == req.request + ClientControl::Void.new + end +end + +def main + client_control_port = '' + server_port = '' + OptionParser.new do |opts| + opts.on('--client_control_port=P', String) do |p| + client_control_port = p + end + opts.on('--server_port=P', String) do |p| + server_port = p + end + end.parse! + + # Allow a few seconds to be safe. + srv = new_rpc_server_for_testing + srv.add_http2_port("0.0.0.0:#{client_control_port}", + :this_port_is_insecure) + stub = Echo::EchoServer::Stub.new("localhost:#{server_port}", + :this_channel_is_insecure) + control_service = SigHandlingClientController.new(stub) + srv.handle(control_service) + server_thread = Thread.new do + srv.run_till_terminated_or_interrupted(['int']) + end + srv.wait_till_running + # send a first RPC to notify the parent process that we've started + stub.echo(Echo::EchoRequest.new(request: 'client/child started')) + server_thread.join +end + +main diff --git a/src/ruby/end2end/graceful_sig_handling_driver.rb b/src/ruby/end2end/graceful_sig_handling_driver.rb new file mode 100755 index 0000000000..e12ae28485 --- /dev/null +++ b/src/ruby/end2end/graceful_sig_handling_driver.rb @@ -0,0 +1,83 @@ +#!/usr/bin/env ruby + +# Copyright 2016 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# smoke test for a grpc-using app that receives and +# handles process-ending signals + +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 + client_started_mu = Mutex.new + 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) + + client_started_mu.synchronize do + client_started_cv.wait(client_started_mu) until client_started + end + + 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' + exit(1) + end + else + STDERR.puts 'Client did not close gracefully' + exit(1) + end + + STDERR.puts 'Client ended gracefully' + + # no need to call cleanup, client should already be dead + server_runner.stop +end + +main diff --git a/src/ruby/end2end/graceful_sig_stop_client.rb b/src/ruby/end2end/graceful_sig_stop_client.rb new file mode 100755 index 0000000000..b672dc3f2a --- /dev/null +++ b/src/ruby/end2end/graceful_sig_stop_client.rb @@ -0,0 +1,78 @@ +#!/usr/bin/env ruby + +# Copyright 2015 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +require_relative './end2end_common' + +# Test client. Sends RPC's as normal but process also has signal handlers +class SigHandlingClientController < ClientControl::ClientController::Service + def initialize(srv, stub) + @srv = srv + @stub = stub + end + + def do_echo_rpc(req, _) + response = @stub.echo(Echo::EchoRequest.new(request: req.request)) + fail 'bad response' unless response.response == req.request + ClientControl::Void.new + end + + def shutdown(_, _) + # Spawn a new thread because RpcServer#stop is + # synchronous and blocks until either this RPC has finished, + # or the server's "poll_period" seconds have passed. + @shutdown_thread = Thread.new do + @srv.stop + end + ClientControl::Void.new + end + + def join_shutdown_thread + @shutdown_thread.join + end +end + +def main + client_control_port = '' + server_port = '' + OptionParser.new do |opts| + opts.on('--client_control_port=P', String) do |p| + client_control_port = p + end + opts.on('--server_port=P', String) do |p| + server_port = p + end + end.parse! + + # The "shutdown" RPC should end very quickly. + # Allow a few seconds to be safe. + srv = new_rpc_server_for_testing(poll_period: 3) + srv.add_http2_port("0.0.0.0:#{client_control_port}", + :this_port_is_insecure) + stub = Echo::EchoServer::Stub.new("localhost:#{server_port}", + :this_channel_is_insecure) + control_service = SigHandlingClientController.new(srv, stub) + srv.handle(control_service) + server_thread = Thread.new do + srv.run_till_terminated_or_interrupted(['int']) + end + srv.wait_till_running + # send a first RPC to notify the parent process that we've started + stub.echo(Echo::EchoRequest.new(request: 'client/child started')) + server_thread.join + control_service.join_shutdown_thread +end + +main diff --git a/src/ruby/end2end/graceful_sig_stop_driver.rb b/src/ruby/end2end/graceful_sig_stop_driver.rb new file mode 100755 index 0000000000..7a132403eb --- /dev/null +++ b/src/ruby/end2end/graceful_sig_stop_driver.rb @@ -0,0 +1,62 @@ +#!/usr/bin/env ruby + +# Copyright 2016 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# smoke test for a grpc-using app that receives and +# handles process-ending signals + +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 + client_started_mu = Mutex.new + 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) + + client_started_mu.synchronize do + client_started_cv.wait(client_started_mu) until client_started + end + + cleanup(control_stub, client_pid, server_runner) +end + +main diff --git a/src/ruby/lib/grpc/generic/rpc_server.rb b/src/ruby/lib/grpc/generic/rpc_server.rb index 3b5a0ce27f..f0f73dc56e 100644 --- a/src/ruby/lib/grpc/generic/rpc_server.rb +++ b/src/ruby/lib/grpc/generic/rpc_server.rb @@ -240,6 +240,13 @@ module GRPC # the call has no impact if the server is already stopped, otherwise # server's current call loop is it's last. def stop + # if called via run_till_terminated_or_interrupted, + # signal stop_server_thread and dont do anything + if @stop_server.nil? == false && @stop_server == false + @stop_server = true + @stop_server_cv.broadcast + return + end @run_mutex.synchronize do fail 'Cannot stop before starting' if @running_state == :not_started return if @running_state != :running @@ -354,6 +361,60 @@ module GRPC alias_method :run_till_terminated, :run + # runs the server with signal handlers + # @param signals + # List of String, Integer or both representing signals that the user + # would like to send to the server for graceful shutdown + # @param wait_interval (optional) + # Integer seconds that user would like stop_server_thread to poll + # stop_server + def run_till_terminated_or_interrupted(signals, wait_interval = 60) + @stop_server = false + @stop_server_mu = Mutex.new + @stop_server_cv = ConditionVariable.new + + @stop_server_thread = Thread.new do + loop do + break if @stop_server + @stop_server_mu.synchronize do + @stop_server_cv.wait(@stop_server_mu, wait_interval) + end + end + + # stop is surrounded by mutex, should handle multiple calls to stop + # correctly + stop + end + + valid_signals = Signal.list + + # register signal handlers + signals.each do |sig| + # input validation + if sig.class == String + sig.upcase! + if sig.start_with?('SIG') + # cut out the SIG prefix to see if valid signal + sig = sig[3..-1] + end + end + + # register signal traps for all valid signals + if valid_signals.value?(sig) || valid_signals.key?(sig) + Signal.trap(sig) do + @stop_server = true + @stop_server_cv.broadcast + end + else + fail "#{sig} not a valid signal" + end + end + + run + + @stop_server_thread.join + end + # Sends RESOURCE_EXHAUSTED if there are too many unprocessed jobs def available?(an_rpc) return an_rpc if @pool.ready_for_work? diff --git a/tools/run_tests/helper_scripts/run_ruby_end2end_tests.sh b/tools/run_tests/helper_scripts/run_ruby_end2end_tests.sh index 7ff877e830..1c48ed20ba 100755 --- a/tools/run_tests/helper_scripts/run_ruby_end2end_tests.sh +++ b/tools/run_tests/helper_scripts/run_ruby_end2end_tests.sh @@ -30,4 +30,6 @@ time ruby src/ruby/end2end/multiple_killed_watching_threads_driver.rb || EXIT_CO time ruby src/ruby/end2end/load_grpc_with_gc_stress_driver.rb || EXIT_CODE=1 time ruby src/ruby/end2end/client_memory_usage_driver.rb || EXIT_CODE=1 time ruby src/ruby/end2end/package_with_underscore_checker.rb || EXIT_CODE=1 +time ruby src/ruby/end2end/graceful_sig_handling_driver.rb || EXIT_CODE=1 +time ruby src/ruby/end2end/graceful_sig_stop_driver.rb || EXIT_CODE=1 exit $EXIT_CODE -- cgit v1.2.3