From 2a9998bc13d4aae9e0288efa1f7676f706a91073 Mon Sep 17 00:00:00 2001 From: Richard Belleville Date: Mon, 19 Aug 2019 13:19:04 -0700 Subject: [PATCH 01/10] Properly handle exceptions in signal handlers for in-flight outgoing RPCs --- .../grpc/_cython/_cygrpc/channel.pyx.pxi | 27 +++++++--- .../grpcio_tests/tests/unit/_signal_client.py | 38 ++++++++++++-- .../tests/unit/_signal_handling_test.py | 49 +++++++++++++++++++ 3 files changed, 103 insertions(+), 11 deletions(-) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi index ca637094353..5b47d356d6f 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi @@ -146,12 +146,17 @@ cdef _cancel( cdef _next_call_event( _ChannelState channel_state, grpc_completion_queue *c_completion_queue, - on_success, deadline): - tag, event = _latent_event(c_completion_queue, deadline) - with channel_state.condition: - on_success(tag) - channel_state.condition.notify_all() - return event + on_success, on_failure, deadline): + try: + tag, event = _latent_event(c_completion_queue, deadline) + except: + on_failure() + raise + else: + with channel_state.condition: + on_success(tag) + channel_state.condition.notify_all() + return event # TODO(https://github.com/grpc/grpc/issues/14569): This could be a lot simpler. @@ -307,8 +312,14 @@ cdef class SegregatedCall: def on_success(tag): _process_segregated_call_tag( self._channel_state, self._call_state, self._c_completion_queue, tag) + def on_failure(): + self._call_state.due.clear() + grpc_call_unref(self._call_state.c_call) + self._call_state.c_call = NULL + self._channel_state.segregated_call_states.remove(self._call_state) + _destroy_c_completion_queue(self._c_completion_queue) return _next_call_event( - self._channel_state, self._c_completion_queue, on_success, None) + self._channel_state, self._c_completion_queue, on_success, on_failure, None) cdef SegregatedCall _segregated_call( @@ -462,7 +473,7 @@ cdef class Channel: else: queue_deadline = None return _next_call_event(self._state, self._state.c_call_completion_queue, - on_success, queue_deadline) + on_success, None, queue_deadline) def segregated_call( self, int flags, method, host, object deadline, object metadata, diff --git a/src/python/grpcio_tests/tests/unit/_signal_client.py b/src/python/grpcio_tests/tests/unit/_signal_client.py index 65ddd6d858e..a2234623a76 100644 --- a/src/python/grpcio_tests/tests/unit/_signal_client.py +++ b/src/python/grpcio_tests/tests/unit/_signal_client.py @@ -45,6 +45,7 @@ def handle_sigint(unused_signum, unused_frame): if per_process_rpc_future is not None: per_process_rpc_future.cancel() sys.stderr.flush() + # This sys.exit(0) avoids an exception caused by the cancelled RPC. sys.exit(0) @@ -72,13 +73,44 @@ def main_streaming(server_target): assert False, _ASSERTION_MESSAGE +def main_unary_with_exception(server_target): + """Initiate an RPC with wait_for_ready set and no server backing the RPC.""" + channel = grpc.insecure_channel(server_target) + try: + channel.unary_unary(UNARY_UNARY)(_MESSAGE, wait_for_ready=True) + except KeyboardInterrupt: + sys.stderr.write("Running signal handler.\n"); sys.stderr.flush() + + sys.stderr.write("Calling Channel.close()"); sys.stderr.flush() + # This call should not hang. + channel.close() + +def main_streaming_with_exception(server_target): + """Initiate an RPC with wait_for_ready set and no server backing the RPC.""" + channel = grpc.insecure_channel(server_target) + try: + channel.unary_stream(UNARY_STREAM)(_MESSAGE, wait_for_ready=True) + except KeyboardInterrupt: + sys.stderr.write("Running signal handler.\n"); sys.stderr.flush() + + sys.stderr.write("Calling Channel.close()"); sys.stderr.flush() + # This call should not hang. + channel.close() + if __name__ == '__main__': parser = argparse.ArgumentParser(description='Signal test client.') parser.add_argument('server', help='Server target') parser.add_argument( - 'arity', help='RPC arity', choices=('unary', 'streaming')) + 'arity', help='Arity', choices=('unary', 'streaming')) + parser.add_argument( + '--exception', help='Whether the signal throws an exception', + action='store_true') args = parser.parse_args() - if args.arity == 'unary': + if args.arity == 'unary' and not args.exception: main_unary(args.server) - else: + elif args.arity == 'streaming' and not args.exception: main_streaming(args.server) + elif args.arity == 'unary' and args.exception: + main_unary_with_exception(args.server) + else: + main_streaming_with_exception(args.server) diff --git a/src/python/grpcio_tests/tests/unit/_signal_handling_test.py b/src/python/grpcio_tests/tests/unit/_signal_handling_test.py index 8ef156c596d..fbb1280e9d4 100644 --- a/src/python/grpcio_tests/tests/unit/_signal_handling_test.py +++ b/src/python/grpcio_tests/tests/unit/_signal_handling_test.py @@ -13,6 +13,7 @@ # limitations under the License. """Test of responsiveness to signals.""" +import contextlib import logging import os import signal @@ -20,6 +21,7 @@ import subprocess import tempfile import threading import unittest +import socket import sys import grpc @@ -167,6 +169,53 @@ class SignalHandlingTest(unittest.TestCase): client_stdout.read()) +@contextlib.contextmanager +def _get_free_loopback_tcp_port(): + sock = socket.socket(socket.AF_INET6) + sock.bind(('', 0)) + address_tuple = sock.getsockname() + try: + yield "[::1]:%s" % (address_tuple[1]) + finally: + sock.close() + + +# TODO(gnossen): Consider combining classes. +class SignalHandlingTestWithoutServer(unittest.TestCase): + + @unittest.skipIf(os.name == 'nt', 'SIGINT not supported on windows') + def testUnaryHandlerWithException(self): + with _get_free_loopback_tcp_port() as server_target: + with tempfile.TemporaryFile(mode='r') as client_stdout: + with tempfile.TemporaryFile(mode='r') as client_stderr: + client = _start_client(('--exception', server_target, 'unary'), + client_stdout, client_stderr) + # TODO(rbellevi): Figure out a way to determininstically hook + # in here. + import time; time.sleep(1) + client.send_signal(signal.SIGINT) + client.wait() + print(_read_stream(client_stderr)) + self.assertEqual(0, client.returncode) + + @unittest.skipIf(os.name == 'nt', 'SIGINT not supported on windows') + def testStreamingHandlerWithException(self): + with _get_free_loopback_tcp_port() as server_target: + with tempfile.TemporaryFile(mode='r') as client_stdout: + with tempfile.TemporaryFile(mode='r') as client_stderr: + client = _start_client(('--exception', server_target, 'streaming'), + client_stdout, client_stderr) + # TODO(rbellevi): Figure out a way to deterministically hook + # in here. + import time; time.sleep(1) + client.send_signal(signal.SIGINT) + client.wait() + print(_read_stream(client_stderr)) + self.assertEqual(0, client.returncode) + + + + if __name__ == '__main__': logging.basicConfig() unittest.main(verbosity=2) From b4eaccf754cd9fdd53efe41e2de13ba5fc0678c8 Mon Sep 17 00:00:00 2001 From: Richard Belleville Date: Mon, 19 Aug 2019 13:22:58 -0700 Subject: [PATCH 02/10] Make tests deterministic --- .../tests/unit/_signal_handling_test.py | 67 ++++++------------- 1 file changed, 20 insertions(+), 47 deletions(-) diff --git a/src/python/grpcio_tests/tests/unit/_signal_handling_test.py b/src/python/grpcio_tests/tests/unit/_signal_handling_test.py index fbb1280e9d4..837a385be0d 100644 --- a/src/python/grpcio_tests/tests/unit/_signal_handling_test.py +++ b/src/python/grpcio_tests/tests/unit/_signal_handling_test.py @@ -168,54 +168,27 @@ class SignalHandlingTest(unittest.TestCase): self.assertIn(_signal_client.SIGTERM_MESSAGE, client_stdout.read()) - -@contextlib.contextmanager -def _get_free_loopback_tcp_port(): - sock = socket.socket(socket.AF_INET6) - sock.bind(('', 0)) - address_tuple = sock.getsockname() - try: - yield "[::1]:%s" % (address_tuple[1]) - finally: - sock.close() - - -# TODO(gnossen): Consider combining classes. -class SignalHandlingTestWithoutServer(unittest.TestCase): - @unittest.skipIf(os.name == 'nt', 'SIGINT not supported on windows') - def testUnaryHandlerWithException(self): - with _get_free_loopback_tcp_port() as server_target: - with tempfile.TemporaryFile(mode='r') as client_stdout: - with tempfile.TemporaryFile(mode='r') as client_stderr: - client = _start_client(('--exception', server_target, 'unary'), - client_stdout, client_stderr) - # TODO(rbellevi): Figure out a way to determininstically hook - # in here. - import time; time.sleep(1) - client.send_signal(signal.SIGINT) - client.wait() - print(_read_stream(client_stderr)) - self.assertEqual(0, client.returncode) + def testUnaryWithException(self): + server_target = '{}:{}'.format(_HOST, self._port) + with tempfile.TemporaryFile(mode='r') as client_stdout: + with tempfile.TemporaryFile(mode='r') as client_stderr: + client = _start_client(('--exception', server_target, 'unary'), + client_stdout, client_stderr) + self._handler.await_connected_client() + client.send_signal(signal.SIGINT) + client.wait() + self.assertEqual(0, client.returncode) @unittest.skipIf(os.name == 'nt', 'SIGINT not supported on windows') def testStreamingHandlerWithException(self): - with _get_free_loopback_tcp_port() as server_target: - with tempfile.TemporaryFile(mode='r') as client_stdout: - with tempfile.TemporaryFile(mode='r') as client_stderr: - client = _start_client(('--exception', server_target, 'streaming'), - client_stdout, client_stderr) - # TODO(rbellevi): Figure out a way to deterministically hook - # in here. - import time; time.sleep(1) - client.send_signal(signal.SIGINT) - client.wait() - print(_read_stream(client_stderr)) - self.assertEqual(0, client.returncode) - - - - -if __name__ == '__main__': - logging.basicConfig() - unittest.main(verbosity=2) + server_target = '{}:{}'.format(_HOST, self._port) + with tempfile.TemporaryFile(mode='r') as client_stdout: + with tempfile.TemporaryFile(mode='r') as client_stderr: + client = _start_client(('--exception', server_target, 'streaming'), + client_stdout, client_stderr) + self._handler.await_connected_client() + client.send_signal(signal.SIGINT) + client.wait() + print(_read_stream(client_stderr)) + self.assertEqual(0, client.returncode) From 4f04a80a69f48d8971098fd658ae76c0bb686a0b Mon Sep 17 00:00:00 2001 From: Richard Belleville Date: Mon, 19 Aug 2019 13:27:56 -0700 Subject: [PATCH 03/10] Add note about something seemingly suspect. --- src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi index 5b47d356d6f..53abdc5ec15 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi @@ -472,6 +472,9 @@ cdef class Channel: queue_deadline = time.time() + 1.0 else: queue_deadline = None + # NOTE(gnossen): It is acceptable for on_failure to be None here because + # failure conditions can only ever happen on the main thread and this + # method is only ever invoked on the channel spin thread. return _next_call_event(self._state, self._state.c_call_completion_queue, on_success, None, queue_deadline) From ca2fcd647ac2f5603d21c19217b8b065d8f968d7 Mon Sep 17 00:00:00 2001 From: Richard Belleville Date: Mon, 19 Aug 2019 13:36:53 -0700 Subject: [PATCH 04/10] Add docstring --- .../grpc/_cython/_cygrpc/channel.pyx.pxi | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi index 53abdc5ec15..c83ff00fedf 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi @@ -147,10 +147,27 @@ cdef _cancel( cdef _next_call_event( _ChannelState channel_state, grpc_completion_queue *c_completion_queue, on_success, on_failure, deadline): + """Block on the next event out of the completion queue. + + On success, `on_success` will be invoked with the tag taken from the CQ. + In the case of a failure due to an exception raised in a signal handler, + `on_failure` will be invoked with no arguments. Note that this situation + can only occur on the main thread. + + Args: + channel_state: The state for the channel on which the RPC is running. + c_completion_queue: The CQ which will be polled. + on_success: A callable object to be invoked upon successful receipt of a + tag from the CQ. + on_failure: A callable object to be invoked in case a Python exception is + raised from a signal handler during polling. + deadline: The point after which the RPC will time out. + """ try: tag, event = _latent_event(c_completion_queue, deadline) except: - on_failure() + if on_failure is not None: + on_failure() raise else: with channel_state.condition: From 84855a18a9bcb0ea6b83ed1dd23999354bf7fe71 Mon Sep 17 00:00:00 2001 From: Richard Belleville Date: Mon, 19 Aug 2019 13:44:32 -0700 Subject: [PATCH 05/10] Yapf --- .../grpcio_tests/tests/unit/_signal_client.py | 16 +++++++++------- .../tests/unit/_signal_handling_test.py | 7 +++---- 2 files changed, 12 insertions(+), 11 deletions(-) diff --git a/src/python/grpcio_tests/tests/unit/_signal_client.py b/src/python/grpcio_tests/tests/unit/_signal_client.py index a2234623a76..9aa37854a23 100644 --- a/src/python/grpcio_tests/tests/unit/_signal_client.py +++ b/src/python/grpcio_tests/tests/unit/_signal_client.py @@ -79,31 +79,33 @@ def main_unary_with_exception(server_target): try: channel.unary_unary(UNARY_UNARY)(_MESSAGE, wait_for_ready=True) except KeyboardInterrupt: - sys.stderr.write("Running signal handler.\n"); sys.stderr.flush() + sys.stderr.write("Running signal handler.\n") + sys.stderr.flush() - sys.stderr.write("Calling Channel.close()"); sys.stderr.flush() # This call should not hang. channel.close() + def main_streaming_with_exception(server_target): """Initiate an RPC with wait_for_ready set and no server backing the RPC.""" channel = grpc.insecure_channel(server_target) try: channel.unary_stream(UNARY_STREAM)(_MESSAGE, wait_for_ready=True) except KeyboardInterrupt: - sys.stderr.write("Running signal handler.\n"); sys.stderr.flush() + sys.stderr.write("Running signal handler.\n") + sys.stderr.flush() - sys.stderr.write("Calling Channel.close()"); sys.stderr.flush() # This call should not hang. channel.close() + if __name__ == '__main__': parser = argparse.ArgumentParser(description='Signal test client.') parser.add_argument('server', help='Server target') + parser.add_argument('arity', help='Arity', choices=('unary', 'streaming')) parser.add_argument( - 'arity', help='Arity', choices=('unary', 'streaming')) - parser.add_argument( - '--exception', help='Whether the signal throws an exception', + '--exception', + help='Whether the signal throws an exception', action='store_true') args = parser.parse_args() if args.arity == 'unary' and not args.exception: diff --git a/src/python/grpcio_tests/tests/unit/_signal_handling_test.py b/src/python/grpcio_tests/tests/unit/_signal_handling_test.py index 837a385be0d..3c46860fcc5 100644 --- a/src/python/grpcio_tests/tests/unit/_signal_handling_test.py +++ b/src/python/grpcio_tests/tests/unit/_signal_handling_test.py @@ -13,7 +13,6 @@ # limitations under the License. """Test of responsiveness to signals.""" -import contextlib import logging import os import signal @@ -21,7 +20,6 @@ import subprocess import tempfile import threading import unittest -import socket import sys import grpc @@ -185,8 +183,9 @@ class SignalHandlingTest(unittest.TestCase): server_target = '{}:{}'.format(_HOST, self._port) with tempfile.TemporaryFile(mode='r') as client_stdout: with tempfile.TemporaryFile(mode='r') as client_stderr: - client = _start_client(('--exception', server_target, 'streaming'), - client_stdout, client_stderr) + client = _start_client( + ('--exception', server_target, 'streaming'), client_stdout, + client_stderr) self._handler.await_connected_client() client.send_signal(signal.SIGINT) client.wait() From 235b27257c90f1773af8628964bb1c453e478b2b Mon Sep 17 00:00:00 2001 From: Richard Belleville Date: Mon, 19 Aug 2019 14:05:54 -0700 Subject: [PATCH 06/10] Re-add unittest.main. --- src/python/grpcio_tests/tests/unit/_signal_handling_test.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/python/grpcio_tests/tests/unit/_signal_handling_test.py b/src/python/grpcio_tests/tests/unit/_signal_handling_test.py index 3c46860fcc5..6f81e0b2d34 100644 --- a/src/python/grpcio_tests/tests/unit/_signal_handling_test.py +++ b/src/python/grpcio_tests/tests/unit/_signal_handling_test.py @@ -191,3 +191,8 @@ class SignalHandlingTest(unittest.TestCase): client.wait() print(_read_stream(client_stderr)) self.assertEqual(0, client.returncode) + + +if __name__ == '__main__': + logging.basicConfig() + unittest.main(verbosity=2) From 967f55efd633055c1e5419ffd00c51282238d265 Mon Sep 17 00:00:00 2001 From: Richard Belleville Date: Mon, 19 Aug 2019 14:20:01 -0700 Subject: [PATCH 07/10] Add explanatory comment. --- src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi index c83ff00fedf..1799780fce4 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi @@ -165,6 +165,8 @@ cdef _next_call_event( """ try: tag, event = _latent_event(c_completion_queue, deadline) + # NOTE(rbellevi): This broad except enables us to clean up resources before + # propagating any exceptions raised by signal handlers to the application. except: if on_failure is not None: on_failure() From 3d56c83a5f86d7930a99392bc1ee7dc5d2c12f1f Mon Sep 17 00:00:00 2001 From: Richard Belleville Date: Mon, 19 Aug 2019 14:24:45 -0700 Subject: [PATCH 08/10] Correct out-of-date docstrings --- src/python/grpcio_tests/tests/unit/_signal_client.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/python/grpcio_tests/tests/unit/_signal_client.py b/src/python/grpcio_tests/tests/unit/_signal_client.py index 9aa37854a23..97e432d5360 100644 --- a/src/python/grpcio_tests/tests/unit/_signal_client.py +++ b/src/python/grpcio_tests/tests/unit/_signal_client.py @@ -74,7 +74,7 @@ def main_streaming(server_target): def main_unary_with_exception(server_target): - """Initiate an RPC with wait_for_ready set and no server backing the RPC.""" + """Initiate a unary RPC with a signal handler that will raise.""" channel = grpc.insecure_channel(server_target) try: channel.unary_unary(UNARY_UNARY)(_MESSAGE, wait_for_ready=True) @@ -87,7 +87,7 @@ def main_unary_with_exception(server_target): def main_streaming_with_exception(server_target): - """Initiate an RPC with wait_for_ready set and no server backing the RPC.""" + """Initiate a streaming RPC with a signal handler that will raise.""" channel = grpc.insecure_channel(server_target) try: channel.unary_stream(UNARY_STREAM)(_MESSAGE, wait_for_ready=True) From f03ae6d493493b5e2c61822a8c10ee7bbf627bf7 Mon Sep 17 00:00:00 2001 From: Richard Belleville Date: Mon, 19 Aug 2019 16:58:56 -0700 Subject: [PATCH 09/10] Fix streaming test case --- src/python/grpcio_tests/tests/unit/_signal_client.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/python/grpcio_tests/tests/unit/_signal_client.py b/src/python/grpcio_tests/tests/unit/_signal_client.py index 97e432d5360..3e13146d9e2 100644 --- a/src/python/grpcio_tests/tests/unit/_signal_client.py +++ b/src/python/grpcio_tests/tests/unit/_signal_client.py @@ -90,7 +90,8 @@ def main_streaming_with_exception(server_target): """Initiate a streaming RPC with a signal handler that will raise.""" channel = grpc.insecure_channel(server_target) try: - channel.unary_stream(UNARY_STREAM)(_MESSAGE, wait_for_ready=True) + for _ in channel.unary_stream(UNARY_STREAM)(_MESSAGE, wait_for_ready=True): + pass except KeyboardInterrupt: sys.stderr.write("Running signal handler.\n") sys.stderr.flush() From e0d04c9a9e64797d0016a176be4f28dd8305a44a Mon Sep 17 00:00:00 2001 From: Richard Belleville Date: Mon, 19 Aug 2019 18:09:19 -0700 Subject: [PATCH 10/10] Yapf. --- src/python/grpcio_tests/tests/unit/_signal_client.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/python/grpcio_tests/tests/unit/_signal_client.py b/src/python/grpcio_tests/tests/unit/_signal_client.py index 3e13146d9e2..075fe7f7177 100644 --- a/src/python/grpcio_tests/tests/unit/_signal_client.py +++ b/src/python/grpcio_tests/tests/unit/_signal_client.py @@ -90,7 +90,8 @@ def main_streaming_with_exception(server_target): """Initiate a streaming RPC with a signal handler that will raise.""" channel = grpc.insecure_channel(server_target) try: - for _ in channel.unary_stream(UNARY_STREAM)(_MESSAGE, wait_for_ready=True): + for _ in channel.unary_stream(UNARY_STREAM)( + _MESSAGE, wait_for_ready=True): pass except KeyboardInterrupt: sys.stderr.write("Running signal handler.\n")