Handle signals within the asyncio loop. (#476)

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

Author: hidmic

launch/launch/launch_service.py

@@ -18,6 +18,7 @@
 import collections.abc
 import contextlib
 import logging
+import platform
 import signal
 import threading
 import traceback
@@ -42,10 +43,7 @@
 from .launch_description import LaunchDescription
 from .launch_description_entity import LaunchDescriptionEntity
 from .some_actions_type import SomeActionsType
-from .utilities import install_signal_handlers
-from .utilities import on_sigint
-from .utilities import on_sigquit
-from .utilities import on_sigterm
+from .utilities import AsyncSafeSignalManager
 from .utilities import visit_all_entities_and_collect_futures
 
 
@@ -62,11 +60,6 @@ def __init__(
         """
         Create a LaunchService.
 
-        If called outside of the main-thread before the function
-        :func:`launch.utilities.install_signal_handlers()` has been called,
-        a ValueError can be raised, as setting signal handlers cannot be done
-        outside of the main-thread.
-
         :param: argv stored in the context for access by the entities, None results in []
         :param: noninteractive if True (not default), this service will assume it has
             no terminal associated e.g. it is being executed from a non interactive script
@@ -80,10 +73,6 @@ def __init__(
         # Setup logging
         self.__logger = launch.logging.get_logger('launch')
 
-        # Install signal handlers if not already installed, will raise if not
-        # in main-thread, call manually in main-thread to avoid this.
-        install_signal_handlers()
-
         # Setup context and register a built-in event handler for bootstrapping.
         self.__context = LaunchContext(argv=self.__argv, noninteractive=noninteractive)
         self.__context.register_event_handler(OnIncludeLaunchDescription())
@@ -197,12 +186,7 @@ def _prepare_run_loop(self):
             # Setup custom signal handlers for SIGINT, SIGTERM and maybe SIGQUIT.
             sigint_received = False
 
-            def _on_sigint(signum, frame, prev_handler):
-                # Ignore additional signals until we finish processing this one.
-                current_handler = signal.signal(signal.SIGINT, signal.SIG_IGN)
-                if current_handler is signal.SIG_IGN:
-                    # This function has been called re-entrantly.
-                    return
+            def _on_sigint(signum):
                 nonlocal sigint_received
                 base_msg = 'user interrupted with ctrl-c (SIGINT)'
                 if not sigint_received:
@@ -214,57 +198,25 @@ def _on_sigint(signum, frame, prev_handler):
                     sigint_received = True
                 else:
                     self.__logger.warning('{} again, ignoring...'.format(base_msg))
-                if callable(prev_handler):
-                    try:
-                        # Run pre-existing signal handler.
-                        prev_handler(signum, frame)
-                    except KeyboardInterrupt:
-                        # Ignore exception.
-                        pass
-                # Restore current signal handler (not necessarily this one).
-                signal.signal(signal.SIGINT, current_handler)
-
-            on_sigint(_on_sigint)
-
-            def _on_sigterm(signum, frame, prev_handler):
-                # Ignore additional signals until we finish processing this one.
-                current_handler = signal.signal(signal.SIGTERM, signal.SIG_IGN)
-                if current_handler is signal.SIG_IGN:
-                    # This function has been called re-entrantly.
-                    return
+
+            def _on_sigterm(signum):
+                signame = signal.Signals(signum).name
+                self.__logger.error(
+                    'user interrupted with ctrl-\\ ({}), terminating...'.format(signame))
                 # TODO(wjwwood): try to terminate running subprocesses before exiting.
-                self.__logger.error('using SIGTERM or SIGQUIT can result in orphaned processes')
+                self.__logger.error('using {} can result in orphaned processes'.format(signame))
                 self.__logger.error('make sure no processes launched are still running')
                 this_loop.call_soon(this_task.cancel)
-                if callable(prev_handler):
-                    # Run pre-existing signal handler.
-                    prev_handler(signum, frame)
-                # Restore current signal handler (not necessarily this one).
-                signal.signal(signal.SIGTERM, current_handler)
-
-            on_sigterm(_on_sigterm)
-
-            def _on_sigquit(signum, frame, prev_handler):
-                # Ignore additional signals until we finish processing this one.
-                current_handler = signal.signal(signal.SIGQUIT, signal.SIG_IGN)
-                if current_handler is signal.SIG_IGN:
-                    # This function has been called re-entrantly.
-                    return
-                self.__logger.error('user interrupted with ctrl-\\ (SIGQUIT), terminating...')
-                _on_sigterm(signum, frame, prev_handler)
-                # Restore current signal handler (not necessarily this one).
-                signal.signal(signal.SIGQUIT, current_handler)
-
-            on_sigquit(_on_sigquit)
-
-            # Yield asyncio loop and current task.
-            yield self.__loop_from_run_thread, this_task
-        finally:
-            # Unset the signal handlers while not running.
-            on_sigint(None)
-            on_sigterm(None)
-            on_sigquit(None)
 
+            with AsyncSafeSignalManager(this_loop) as manager:
+                # Setup signal handlers
+                manager.handle(signal.SIGINT, _on_sigint)
+                manager.handle(signal.SIGTERM, _on_sigterm)
+                if platform.system() != 'Windows':
+                    manager.handle(signal.SIGQUIT, _on_sigterm)
+                # Yield asyncio loop and current task.
+                yield this_loop, this_task
+        finally:
             # No matter what happens, unset the loop.
             with self.__loop_from_run_thread_lock:
                 self.__context._set_asyncio_loop(None)
@@ -309,9 +261,6 @@ async def run_async(self, *, shutdown_when_idle=True) -> int:
         This should only ever be run from the main thread and not concurrently with other
         asynchronous runs.
 
-        Note that custom signal handlers are set, and KeyboardInterrupt is caught and ignored
-        around the original signal handler. After the run ends, this behavior is undone.
-
         :param: shutdown_when_idle if True (default), the service will shutdown when idle.
         """
         # Make sure this has not been called from any thread but the main thread.
@@ -397,14 +346,20 @@ def run(self, *, shutdown_when_idle=True) -> int:
         This should only ever be run from the main thread and not concurrently with
         asynchronous runs (see `run_async()` documentation).
 
+        Note that KeyboardInterrupt is caught and ignored, as signals are handled separately.
+        After the run ends, this behavior is undone.
+
         :param: shutdown_when_idle if True (default), the service will shutdown when idle
         """
         loop = osrf_pycommon.process_utils.get_loop()
         run_async_task = loop.create_task(self.run_async(
             shutdown_when_idle=shutdown_when_idle
         ))
-        loop.run_until_complete(run_async_task)
-        return run_async_task.result()
+        while True:
+            try:
+                return loop.run_until_complete(run_async_task)
+            except KeyboardInterrupt:
+                continue
 
     def __on_shutdown(self, event: Event, context: LaunchContext) -> Optional[SomeActionsType]:
         self.__shutting_down = True

launch/launch/utilities/__init__.py

@@ -19,10 +19,7 @@
 from .ensure_argument_type_impl import ensure_argument_type
 from .normalize_to_list_of_substitutions_impl import normalize_to_list_of_substitutions
 from .perform_substitutions_impl import perform_substitutions
-from .signal_management import install_signal_handlers
-from .signal_management import on_sigint
-from .signal_management import on_sigquit
-from .signal_management import on_sigterm
+from .signal_management import AsyncSafeSignalManager
 from .visit_all_entities_and_collect_futures_impl import visit_all_entities_and_collect_futures
 
 __all__ = [
@@ -32,10 +29,7 @@
     'create_future',
     'ensure_argument_type',
     'perform_substitutions',
-    'install_signal_handlers',
-    'on_sigint',
-    'on_sigquit',
-    'on_sigterm',
+    'AsyncSafeSignalManager',
     'normalize_to_list_of_substitutions',
     'visit_all_entities_and_collect_futures',
 ]