Properly handle exceptions in signal handlers for in-flight outgoing RPCs

src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi

@@ -146,8 +146,13 @@ cdef _cancel(
 
 cdef _next_call_event(
     _ChannelState channel_state, grpc_completion_queue *c_completion_queue,
-    on_success, deadline):
+    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()
@@ -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,

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)

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)