diff options
Diffstat (limited to 'src')
18 files changed, 651 insertions, 60 deletions
diff --git a/src/python/grpcio/grpc/__init__.py b/src/python/grpcio/grpc/__init__.py index daf869b156..70d7618e05 100644 --- a/src/python/grpcio/grpc/__init__.py +++ b/src/python/grpcio/grpc/__init__.py @@ -23,6 +23,11 @@ from grpc._cython import cygrpc as _cygrpc logging.getLogger(__name__).addHandler(logging.NullHandler()) +try: + from ._grpcio_metadata import __version__ +except ImportError: + __version__ = "dev0" + ############################## Future Interface ############################### diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pxd.pxi index e0e068e452..01b8237484 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pxd.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pxd.pxi @@ -28,19 +28,22 @@ cdef tuple _wrap_grpc_arg(grpc_arg arg) cdef grpc_arg _unwrap_grpc_arg(tuple wrapped_arg) -cdef class _ArgumentProcessor: +cdef class _ChannelArg: cdef grpc_arg c_argument cdef void c(self, argument, grpc_arg_pointer_vtable *vtable, references) except * -cdef class _ArgumentsProcessor: +cdef class _ChannelArgs: cdef readonly tuple _arguments - cdef list _argument_processors + cdef list _channel_args cdef readonly list _references cdef grpc_channel_args _c_arguments - cdef grpc_channel_args *c(self, grpc_arg_pointer_vtable *vtable) except * - cdef un_c(self) + cdef void _c(self, grpc_arg_pointer_vtable *vtable) except * + cdef grpc_channel_args *c_args(self) except * + + @staticmethod + cdef _ChannelArgs from_args(object arguments, grpc_arg_pointer_vtable * vtable) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pyx.pxi index b7a4277ff6..bf12871015 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/arguments.pyx.pxi @@ -50,7 +50,7 @@ cdef grpc_arg _unwrap_grpc_arg(tuple wrapped_arg): return wrapped.arg -cdef class _ArgumentProcessor: +cdef class _ChannelArg: cdef void c(self, argument, grpc_arg_pointer_vtable *vtable, references) except *: key, value = argument @@ -82,27 +82,34 @@ cdef class _ArgumentProcessor: 'Expected int, bytes, or behavior, got {}'.format(type(value))) -cdef class _ArgumentsProcessor: +cdef class _ChannelArgs: def __cinit__(self, arguments): self._arguments = () if arguments is None else tuple(arguments) - self._argument_processors = [] + self._channel_args = [] self._references = [] + self._c_arguments.arguments = NULL - cdef grpc_channel_args *c(self, grpc_arg_pointer_vtable *vtable) except *: + cdef void _c(self, grpc_arg_pointer_vtable *vtable) except *: self._c_arguments.arguments_length = len(self._arguments) - if self._c_arguments.arguments_length == 0: - return NULL - else: + if self._c_arguments.arguments_length != 0: self._c_arguments.arguments = <grpc_arg *>gpr_malloc( self._c_arguments.arguments_length * sizeof(grpc_arg)) for index, argument in enumerate(self._arguments): - argument_processor = _ArgumentProcessor() - argument_processor.c(argument, vtable, self._references) - self._c_arguments.arguments[index] = argument_processor.c_argument - self._argument_processors.append(argument_processor) - return &self._c_arguments - - cdef un_c(self): - if self._arguments: + channel_arg = _ChannelArg() + channel_arg.c(argument, vtable, self._references) + self._c_arguments.arguments[index] = channel_arg.c_argument + self._channel_args.append(channel_arg) + + cdef grpc_channel_args *c_args(self) except *: + return &self._c_arguments + + def __dealloc__(self): + if self._c_arguments.arguments != NULL: gpr_free(self._c_arguments.arguments) + + @staticmethod + cdef _ChannelArgs from_args(object arguments, grpc_arg_pointer_vtable * vtable): + cdef _ChannelArgs channel_args = _ChannelArgs(arguments) + channel_args._c(vtable) + return channel_args diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi index 135d224095..70d4abb730 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi @@ -423,16 +423,15 @@ cdef class Channel: self._vtable.copy = &_copy_pointer self._vtable.destroy = &_destroy_pointer self._vtable.cmp = &_compare_pointer - cdef _ArgumentsProcessor arguments_processor = _ArgumentsProcessor( - arguments) - cdef grpc_channel_args *c_arguments = arguments_processor.c(&self._vtable) + cdef _ChannelArgs channel_args = _ChannelArgs.from_args( + arguments, &self._vtable) if channel_credentials is None: self._state.c_channel = grpc_insecure_channel_create( - <char *>target, c_arguments, NULL) + <char *>target, channel_args.c_args(), NULL) else: c_channel_credentials = channel_credentials.c() self._state.c_channel = grpc_secure_channel_create( - c_channel_credentials, <char *>target, c_arguments, NULL) + c_channel_credentials, <char *>target, channel_args.c_args(), NULL) grpc_channel_credentials_release(c_channel_credentials) self._state.c_call_completion_queue = ( grpc_completion_queue_create_for_next(NULL)) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/server.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/server.pxd.pxi index 52cfccb677..4a6fbe0f96 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/server.pxd.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/server.pxd.pxi @@ -16,7 +16,6 @@ cdef class Server: cdef grpc_arg_pointer_vtable _vtable - cdef readonly _ArgumentsProcessor _arguments_processor cdef grpc_server *c_server cdef bint is_started # start has been called cdef bint is_shutting_down # shutdown has been called diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi index e89e02b171..d72648a35d 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi @@ -29,11 +29,9 @@ cdef class Server: self._vtable.copy = &_copy_pointer self._vtable.destroy = &_destroy_pointer self._vtable.cmp = &_compare_pointer - cdef _ArgumentsProcessor arguments_processor = _ArgumentsProcessor( - arguments) - cdef grpc_channel_args *c_arguments = arguments_processor.c(&self._vtable) - self.c_server = grpc_server_create(c_arguments, NULL) - arguments_processor.un_c() + cdef _ChannelArgs channel_args = _ChannelArgs.from_args( + arguments, &self._vtable) + self.c_server = grpc_server_create(channel_args.c_args(), NULL) self.references.append(arguments) self.is_started = False self.is_shutting_down = False diff --git a/src/python/grpcio_health_checking/grpc_health/v1/health.py b/src/python/grpcio_health_checking/grpc_health/v1/health.py index 0583659428..0a5bbb5504 100644 --- a/src/python/grpcio_health_checking/grpc_health/v1/health.py +++ b/src/python/grpcio_health_checking/grpc_health/v1/health.py @@ -23,15 +23,61 @@ from grpc_health.v1 import health_pb2_grpc as _health_pb2_grpc SERVICE_NAME = _health_pb2.DESCRIPTOR.services_by_name['Health'].full_name +class _Watcher(): + + def __init__(self): + self._condition = threading.Condition() + self._responses = list() + self._open = True + + def __iter__(self): + return self + + def _next(self): + with self._condition: + while not self._responses and self._open: + self._condition.wait() + if self._responses: + return self._responses.pop(0) + else: + raise StopIteration() + + def next(self): + return self._next() + + def __next__(self): + return self._next() + + def add(self, response): + with self._condition: + self._responses.append(response) + self._condition.notify() + + def close(self): + with self._condition: + self._open = False + self._condition.notify() + + class HealthServicer(_health_pb2_grpc.HealthServicer): """Servicer handling RPCs for service statuses.""" def __init__(self): - self._server_status_lock = threading.Lock() + self._lock = threading.RLock() self._server_status = {} + self._watchers = {} + + def _on_close_callback(self, watcher, service): + + def callback(): + with self._lock: + self._watchers[service].remove(watcher) + watcher.close() + + return callback def Check(self, request, context): - with self._server_status_lock: + with self._lock: status = self._server_status.get(request.service) if status is None: context.set_code(grpc.StatusCode.NOT_FOUND) @@ -39,14 +85,30 @@ class HealthServicer(_health_pb2_grpc.HealthServicer): else: return _health_pb2.HealthCheckResponse(status=status) + def Watch(self, request, context): + service = request.service + with self._lock: + status = self._server_status.get(service) + if status is None: + status = _health_pb2.HealthCheckResponse.SERVICE_UNKNOWN # pylint: disable=no-member + watcher = _Watcher() + watcher.add(_health_pb2.HealthCheckResponse(status=status)) + if service not in self._watchers: + self._watchers[service] = set() + self._watchers[service].add(watcher) + context.add_callback(self._on_close_callback(watcher, service)) + return watcher + def set(self, service, status): """Sets the status of a service. - Args: - service: string, the name of the service. - NOTE, '' must be set. - status: HealthCheckResponse.status enum value indicating - the status of the service - """ - with self._server_status_lock: + Args: + service: string, the name of the service. NOTE, '' must be set. + status: HealthCheckResponse.status enum value indicating the status of + the service + """ + with self._lock: self._server_status[service] = status + if service in self._watchers: + for watcher in self._watchers[service]: + watcher.add(_health_pb2.HealthCheckResponse(status=status)) diff --git a/src/python/grpcio_tests/commands.py b/src/python/grpcio_tests/commands.py index d5327711d3..582ce898de 100644 --- a/src/python/grpcio_tests/commands.py +++ b/src/python/grpcio_tests/commands.py @@ -141,6 +141,7 @@ class TestGevent(setuptools.Command): 'unit._exit_test.ExitTest.test_in_flight_partial_unary_stream_call', 'unit._exit_test.ExitTest.test_in_flight_partial_stream_unary_call', 'unit._exit_test.ExitTest.test_in_flight_partial_stream_stream_call', + 'health_check._health_servicer_test.HealthServicerTest.test_cancelled_watch_removed_from_watch_list', # TODO(https://github.com/grpc/grpc/issues/17330) enable these three tests 'channelz._channelz_servicer_test.ChannelzServicerTest.test_many_subchannels', 'channelz._channelz_servicer_test.ChannelzServicerTest.test_many_subchannels_and_sockets', diff --git a/src/python/grpcio_tests/tests/health_check/BUILD.bazel b/src/python/grpcio_tests/tests/health_check/BUILD.bazel index 19e1e1b2e1..77bc61aa30 100644 --- a/src/python/grpcio_tests/tests/health_check/BUILD.bazel +++ b/src/python/grpcio_tests/tests/health_check/BUILD.bazel @@ -9,6 +9,7 @@ py_test( "//src/python/grpcio/grpc:grpcio", "//src/python/grpcio_health_checking/grpc_health/v1:grpc_health", "//src/python/grpcio_tests/tests/unit:test_common", + "//src/python/grpcio_tests/tests/unit/framework/common:common", ], imports = ["../../",], ) diff --git a/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py b/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py index c1d9436c2f..35794987bc 100644 --- a/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py +++ b/src/python/grpcio_tests/tests/health_check/_health_servicer_test.py @@ -13,6 +13,8 @@ # limitations under the License. """Tests of grpc_health.v1.health.""" +import threading +import time import unittest import grpc @@ -21,22 +23,36 @@ from grpc_health.v1 import health_pb2 from grpc_health.v1 import health_pb2_grpc from tests.unit import test_common +from tests.unit.framework.common import test_constants + +from six.moves import queue + +_SERVING_SERVICE = 'grpc.test.TestServiceServing' +_UNKNOWN_SERVICE = 'grpc.test.TestServiceUnknown' +_NOT_SERVING_SERVICE = 'grpc.test.TestServiceNotServing' +_WATCH_SERVICE = 'grpc.test.WatchService' + + +def _consume_responses(response_iterator, response_queue): + for response in response_iterator: + response_queue.put(response) class HealthServicerTest(unittest.TestCase): def setUp(self): - servicer = health.HealthServicer() - servicer.set('', health_pb2.HealthCheckResponse.SERVING) - servicer.set('grpc.test.TestServiceServing', - health_pb2.HealthCheckResponse.SERVING) - servicer.set('grpc.test.TestServiceUnknown', - health_pb2.HealthCheckResponse.UNKNOWN) - servicer.set('grpc.test.TestServiceNotServing', - health_pb2.HealthCheckResponse.NOT_SERVING) + self._servicer = health.HealthServicer() + self._servicer.set('', health_pb2.HealthCheckResponse.SERVING) + self._servicer.set(_SERVING_SERVICE, + health_pb2.HealthCheckResponse.SERVING) + self._servicer.set(_UNKNOWN_SERVICE, + health_pb2.HealthCheckResponse.UNKNOWN) + self._servicer.set(_NOT_SERVING_SERVICE, + health_pb2.HealthCheckResponse.NOT_SERVING) self._server = test_common.test_server() port = self._server.add_insecure_port('[::]:0') - health_pb2_grpc.add_HealthServicer_to_server(servicer, self._server) + health_pb2_grpc.add_HealthServicer_to_server(self._servicer, + self._server) self._server.start() self._channel = grpc.insecure_channel('localhost:%d' % port) @@ -46,37 +62,160 @@ class HealthServicerTest(unittest.TestCase): self._server.stop(None) self._channel.close() - def test_empty_service(self): + def test_check_empty_service(self): request = health_pb2.HealthCheckRequest() resp = self._stub.Check(request) self.assertEqual(health_pb2.HealthCheckResponse.SERVING, resp.status) - def test_serving_service(self): - request = health_pb2.HealthCheckRequest( - service='grpc.test.TestServiceServing') + def test_check_serving_service(self): + request = health_pb2.HealthCheckRequest(service=_SERVING_SERVICE) resp = self._stub.Check(request) self.assertEqual(health_pb2.HealthCheckResponse.SERVING, resp.status) - def test_unknown_serivce(self): - request = health_pb2.HealthCheckRequest( - service='grpc.test.TestServiceUnknown') + def test_check_unknown_serivce(self): + request = health_pb2.HealthCheckRequest(service=_UNKNOWN_SERVICE) resp = self._stub.Check(request) self.assertEqual(health_pb2.HealthCheckResponse.UNKNOWN, resp.status) - def test_not_serving_service(self): - request = health_pb2.HealthCheckRequest( - service='grpc.test.TestServiceNotServing') + def test_check_not_serving_service(self): + request = health_pb2.HealthCheckRequest(service=_NOT_SERVING_SERVICE) resp = self._stub.Check(request) self.assertEqual(health_pb2.HealthCheckResponse.NOT_SERVING, resp.status) - def test_not_found_service(self): + def test_check_not_found_service(self): request = health_pb2.HealthCheckRequest(service='not-found') with self.assertRaises(grpc.RpcError) as context: resp = self._stub.Check(request) self.assertEqual(grpc.StatusCode.NOT_FOUND, context.exception.code()) + def test_watch_empty_service(self): + request = health_pb2.HealthCheckRequest(service='') + response_queue = queue.Queue() + rendezvous = self._stub.Watch(request) + thread = threading.Thread( + target=_consume_responses, args=(rendezvous, response_queue)) + thread.start() + + response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT) + self.assertEqual(health_pb2.HealthCheckResponse.SERVING, + response.status) + + rendezvous.cancel() + thread.join() + self.assertTrue(response_queue.empty()) + + def test_watch_new_service(self): + request = health_pb2.HealthCheckRequest(service=_WATCH_SERVICE) + response_queue = queue.Queue() + rendezvous = self._stub.Watch(request) + thread = threading.Thread( + target=_consume_responses, args=(rendezvous, response_queue)) + thread.start() + + response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT) + self.assertEqual(health_pb2.HealthCheckResponse.SERVICE_UNKNOWN, + response.status) + + self._servicer.set(_WATCH_SERVICE, + health_pb2.HealthCheckResponse.SERVING) + response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT) + self.assertEqual(health_pb2.HealthCheckResponse.SERVING, + response.status) + + self._servicer.set(_WATCH_SERVICE, + health_pb2.HealthCheckResponse.NOT_SERVING) + response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT) + self.assertEqual(health_pb2.HealthCheckResponse.NOT_SERVING, + response.status) + + rendezvous.cancel() + thread.join() + self.assertTrue(response_queue.empty()) + + def test_watch_service_isolation(self): + request = health_pb2.HealthCheckRequest(service=_WATCH_SERVICE) + response_queue = queue.Queue() + rendezvous = self._stub.Watch(request) + thread = threading.Thread( + target=_consume_responses, args=(rendezvous, response_queue)) + thread.start() + + response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT) + self.assertEqual(health_pb2.HealthCheckResponse.SERVICE_UNKNOWN, + response.status) + + self._servicer.set('some-other-service', + health_pb2.HealthCheckResponse.SERVING) + with self.assertRaises(queue.Empty): + response_queue.get(timeout=test_constants.SHORT_TIMEOUT) + + rendezvous.cancel() + thread.join() + self.assertTrue(response_queue.empty()) + + def test_two_watchers(self): + request = health_pb2.HealthCheckRequest(service=_WATCH_SERVICE) + response_queue1 = queue.Queue() + response_queue2 = queue.Queue() + rendezvous1 = self._stub.Watch(request) + rendezvous2 = self._stub.Watch(request) + thread1 = threading.Thread( + target=_consume_responses, args=(rendezvous1, response_queue1)) + thread2 = threading.Thread( + target=_consume_responses, args=(rendezvous2, response_queue2)) + thread1.start() + thread2.start() + + response1 = response_queue1.get(timeout=test_constants.SHORT_TIMEOUT) + response2 = response_queue2.get(timeout=test_constants.SHORT_TIMEOUT) + self.assertEqual(health_pb2.HealthCheckResponse.SERVICE_UNKNOWN, + response1.status) + self.assertEqual(health_pb2.HealthCheckResponse.SERVICE_UNKNOWN, + response2.status) + + self._servicer.set(_WATCH_SERVICE, + health_pb2.HealthCheckResponse.SERVING) + response1 = response_queue1.get(timeout=test_constants.SHORT_TIMEOUT) + response2 = response_queue2.get(timeout=test_constants.SHORT_TIMEOUT) + self.assertEqual(health_pb2.HealthCheckResponse.SERVING, + response1.status) + self.assertEqual(health_pb2.HealthCheckResponse.SERVING, + response2.status) + + rendezvous1.cancel() + rendezvous2.cancel() + thread1.join() + thread2.join() + self.assertTrue(response_queue1.empty()) + self.assertTrue(response_queue2.empty()) + + def test_cancelled_watch_removed_from_watch_list(self): + request = health_pb2.HealthCheckRequest(service=_WATCH_SERVICE) + response_queue = queue.Queue() + rendezvous = self._stub.Watch(request) + thread = threading.Thread( + target=_consume_responses, args=(rendezvous, response_queue)) + thread.start() + + response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT) + self.assertEqual(health_pb2.HealthCheckResponse.SERVICE_UNKNOWN, + response.status) + + rendezvous.cancel() + self._servicer.set(_WATCH_SERVICE, + health_pb2.HealthCheckResponse.SERVING) + thread.join() + + # Wait, if necessary, for serving thread to process client cancellation + timeout = time.time() + test_constants.SHORT_TIMEOUT + while time.time() < timeout and self._servicer._watchers[_WATCH_SERVICE]: + time.sleep(1) + self.assertFalse(self._servicer._watchers[_WATCH_SERVICE], + 'watch set should be empty') + self.assertTrue(response_queue.empty()) + def test_health_service_name(self): self.assertEqual(health.SERVICE_NAME, 'grpc.health.v1.Health') diff --git a/src/python/grpcio_tests/tests/tests.json b/src/python/grpcio_tests/tests/tests.json index f202a3f932..de4c2c1fdd 100644 --- a/src/python/grpcio_tests/tests/tests.json +++ b/src/python/grpcio_tests/tests/tests.json @@ -64,6 +64,7 @@ "unit._server_ssl_cert_config_test.ServerSSLCertReloadTestWithoutClientAuth", "unit._server_test.ServerTest", "unit._session_cache_test.SSLSessionCacheTest", + "unit._version_test.VersionTest", "unit.beta._beta_features_test.BetaFeaturesTest", "unit.beta._beta_features_test.ContextManagementAndLifecycleTest", "unit.beta._connectivity_channel_test.ConnectivityStatesTest", diff --git a/src/python/grpcio_tests/tests/unit/BUILD.bazel b/src/python/grpcio_tests/tests/unit/BUILD.bazel index 1b462ec67a..a9bcd9f304 100644 --- a/src/python/grpcio_tests/tests/unit/BUILD.bazel +++ b/src/python/grpcio_tests/tests/unit/BUILD.bazel @@ -7,6 +7,7 @@ GRPCIO_TESTS_UNIT = [ "_api_test.py", "_auth_context_test.py", "_auth_test.py", + "_version_test.py", "_channel_args_test.py", "_channel_close_test.py", "_channel_connectivity_test.py", diff --git a/src/python/grpcio_tests/tests/unit/_version_test.py b/src/python/grpcio_tests/tests/unit/_version_test.py new file mode 100644 index 0000000000..3d37b319e5 --- /dev/null +++ b/src/python/grpcio_tests/tests/unit/_version_test.py @@ -0,0 +1,30 @@ +# Copyright 2018 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. +"""Test for grpc.__version__""" + +import unittest +import grpc +import logging +from grpc import _grpcio_metadata + + +class VersionTest(unittest.TestCase): + + def test_get_version(self): + self.assertEqual(grpc.__version__, _grpcio_metadata.__version__) + + +if __name__ == '__main__': + logging.basicConfig() + unittest.main(verbosity=2) 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? |