Do not attach process function interrupt handlers for daemon runners

Recently, signal handlers were added to catch interrupt signal to kill
active processes that were `run` in a local interpreter. Since process
functions create their own runner, separate handlers had to be added
there. However, these handlers were than also attached when the
functions are run in a daemon runner. Shutting the daemon down would
then trigger these signal handlers and the process functions would be
killed.

aiida/engine/daemon/runner.py

@@ -30,7 +30,9 @@ def start_daemon():
     configure_logging(daemon=True, daemon_log_file=daemon_client.daemon_log_file)
 
     try:
-        runner = get_manager().create_daemon_runner()
+        manager = get_manager()
+        runner = manager.create_daemon_runner()
+        manager.set_runner(runner)
     except Exception as exception:
         LOGGER.exception('daemon runner failed to start')
         raise

aiida/engine/processes/functions.py

@@ -119,14 +119,15 @@ def run_get_node(*args, **kwargs):
             Run the FunctionProcess with the supplied inputs in a local runner.
 
             The function will have to create a new runner for the FunctionProcess instead of using the global runner,
-            because otherwise if this workfunction were to call another one from within its scope, that would use
+            because otherwise if this process function were to call another one from within its scope, that would use
             the same runner and it would be blocking the event loop from continuing.
 
             :param args: input arguments to construct the FunctionProcess
             :param kwargs: input keyword arguments to construct the FunctionProcess
             :return: tuple of the outputs of the process and the process node pk
             :rtype: (dict, int)
             """
+            current_runner = get_manager().get_runner()
             runner = get_manager().create_runner(with_persistence=False)
             inputs = process_class.create_inputs(*args, **kwargs)
 
@@ -140,18 +141,22 @@ def run_get_node(*args, **kwargs):
 
             process = process_class(inputs=inputs, runner=runner)
 
-            def kill_process(_num, _frame):
-                """Send the kill signal to the process in the current scope."""
-                LOGGER.critical('runner received interrupt, killing process %s', process.pid)
-                process.kill(msg='Process was killed because the runner received an interrupt')
+            # Only add handlers for interrupt signal to kill the current process if we are not in a daemon runner
+            # Without this check, running process functions in a daemon worker would be killed if the daemon is shutdown
+            if not current_runner.is_daemon_runner:
 
-            signal.signal(signal.SIGINT, kill_process)
-            signal.signal(signal.SIGTERM, kill_process)
+                def kill_process(_num, _frame):
+                    """Send the kill signal to the process in the current scope."""
+                    LOGGER.critical('runner received interrupt, killing process %s', process.pid)
+                    process.kill(msg='Process was killed because the runner received an interrupt')
 
-            result = process.execute()
+                signal.signal(signal.SIGINT, kill_process)
+                signal.signal(signal.SIGTERM, kill_process)
 
-            # Close the runner properly
-            runner.close()
+            try:
+                result = process.execute()
+            finally:
+                runner.close()
 
             store_provenance = inputs.get('metadata', {}).get('store_provenance', True)
             if not store_provenance:

aiida/engine/runners.py

@@ -116,6 +116,15 @@ def job_manager(self):
     def controller(self):
         return self._controller
 
+    @property
+    def is_daemon_runner(self):
+        """Return whether the runner is a daemon runner, which means it submits processes over RabbitMQ.
+
+        :return: True if the runner is a daemon runner
+        :rtype: bool
+        """
+        return self._rmq_submit
+
     def is_closed(self):
         return self._closed