Runners refactoring

This is a refactoring of runners. It moves most of the code related to
running and monitoring processes into base class. This change removes
duplicities across the runners, and it will help with implementation of
common features like signal handling. Also it makes the future
development of new runners easier, because developers can focus on
actual runner and don't have to deal with things like process
monitoring, logging and exception handling

Signed-off-by: Jan Richter <jarichte@redhat.com>

Author: richtja

avocado/core/nrunner/runner.py

@@ -1,7 +1,12 @@
+import multiprocessing
+import signal
 import time
+import traceback
 
+from avocado.core.exceptions import TestInterrupt
 from avocado.core.nrunner.runnable import RUNNERS_REGISTRY_STANDALONE_EXECUTABLE
 from avocado.core.plugin_interfaces import RunnableRunner
+from avocado.core.utils import messages
 
 #: The amount of time (in seconds) between each internal status check
 RUNNER_RUN_CHECK_INTERVAL = 0.01
@@ -46,6 +51,11 @@ class BaseRunner(RunnableRunner):
     #: this runners makes use of.
     CONFIGURATION_USED = []
 
+    @staticmethod
+    def signal_handler(signum, frame):  # pylint: disable=W0613
+        if signum == signal.SIGTERM.value:
+            raise TestInterrupt("Test interrupted: Timeout reached")
+
     @staticmethod
     def prepare_status(status_type, additional_info=None):
         """Prepare a status dict with some basic information.
@@ -66,23 +76,80 @@ def prepare_status(status_type, additional_info=None):
         status.update({"status": status_type, "time": time.monotonic()})
         return status
 
-    def running_loop(self, condition):
-        """Produces timely running messages until end condition is found.
+    @staticmethod
+    def _monitor(queue):
+        most_recent_status_time = None
+        while True:
+            time.sleep(RUNNER_RUN_CHECK_INTERVAL)
+            if queue.empty():
+                now = time.monotonic()
+                if (
+                    most_recent_status_time is None
+                    or now >= most_recent_status_time + RUNNER_RUN_STATUS_INTERVAL
+                ):
+                    most_recent_status_time = now
+                    yield messages.RunningMessage.get()
+                continue
+            else:
+                message = queue.get()
+                yield message
+                if message.get("status") == "finished":
+                    break
+
+    def _catch_errors(self, runnable, queue):
+        """Wrapper around runners methods for catching and logging failures."""
+        try:
+            messages.start_logging(runnable.config, queue)
+            signal.signal(signal.SIGTERM, self.signal_handler)
+            for message in self._run(runnable):
+                queue.put(message)
+        except TestInterrupt as e:
+            queue.put(messages.StderrMessage.get(str(e)))
+            queue.put(messages.FinishedMessage.get("interrupted", fail_reason=str(e)))
+        except Exception as e:
+            queue.put(messages.StderrMessage.get(traceback.format_exc()))
+            queue.put(
+                messages.FinishedMessage.get(
+                    "error",
+                    fail_reason=str(e),
+                    fail_class=e.__class__.__name__,
+                    traceback=traceback.format_exc(),
+                )
+            )
+
+    def run(self, runnable):
+        # pylint: disable=W0201
+        signal.signal(signal.SIGTERM, self.signal_handler)
+        self.runnable = runnable
+        yield messages.StartedMessage.get()
+        try:
+            queue = multiprocessing.SimpleQueue()
+            process = multiprocessing.Process(
+                target=self._catch_errors, args=(self.runnable, queue)
+            )
+
+            process.start()
+
+            for message in self._monitor(queue):
+                yield message
+
+        except TestInterrupt:
+            process.terminate()
+            for message in self._monitor(queue):
+                yield message
+        except Exception as e:
+            yield messages.StderrMessage.get(traceback.format_exc())
+            yield messages.FinishedMessage.get(
+                "error",
+                fail_reason=str(e),
+                fail_class=e.__class__.__name__,
+                traceback=traceback.format_exc(),
+            )
+
+    def _run(self, runnable):
+        """
+        Run the Runnable
 
-        :param condition: a callable that will be evaluated as a
-                          condition for continuing the loop
+        :param runnable: the runnable object
+        :type runnable: :class:`Runnable`
         """
-        most_current_execution_state_time = None
-        while not condition():
-            now = time.monotonic()
-            if most_current_execution_state_time is not None:
-                next_execution_state_mark = (
-                    most_current_execution_state_time + RUNNER_RUN_STATUS_INTERVAL
-                )
-            if (
-                most_current_execution_state_time is None
-                or now > next_execution_state_mark
-            ):
-                most_current_execution_state_time = now
-                yield self.prepare_status("running")
-            time.sleep(RUNNER_RUN_CHECK_INTERVAL)

avocado/plugins/runners/asset.py

@@ -1,9 +1,8 @@
 import sys
-import time
-from multiprocessing import Process, SimpleQueue, set_start_method
+from multiprocessing import set_start_method
 
 from avocado.core.nrunner.app import BaseRunnerApp
-from avocado.core.nrunner.runner import RUNNER_RUN_STATUS_INTERVAL, BaseRunner
+from avocado.core.nrunner.runner import BaseRunner
 from avocado.core.settings import settings
 from avocado.utils import data_structures
 from avocado.utils.asset import Asset
@@ -36,7 +35,7 @@ class AssetRunner(BaseRunner):
     CONFIGURATION_USED = ["datadir.paths.cache_dirs"]
 
     @staticmethod
-    def _fetch_asset(name, asset_hash, algorithm, locations, cache_dirs, expire, queue):
+    def _fetch_asset(name, asset_hash, algorithm, locations, cache_dirs, expire):
 
         asset_manager = Asset(
             name, asset_hash, algorithm, locations, cache_dirs, expire
@@ -52,50 +51,25 @@ def _fetch_asset(name, asset_hash, algorithm, locations, cache_dirs, expire, que
             result = "error"
             stderr = str(exc)
 
-        output = {"result": result, "stdout": stdout, "stderr": stderr}
-        queue.put(output)
+        return {"result": result, "stdout": stdout, "stderr": stderr}
 
-    def run(self, runnable):
-        # pylint: disable=W0201
-        self.runnable = runnable
-        yield self.prepare_status("started")
-
-        name = self.runnable.kwargs.get("name")
+    def _run(self, runnable):
+        name = runnable.kwargs.get("name")
         # if name was passed correctly, run the Avocado Asset utility
         if name is not None:
-            asset_hash = self.runnable.kwargs.get("asset_hash")
-            algorithm = self.runnable.kwargs.get("algorithm")
-            locations = self.runnable.kwargs.get("locations")
-            expire = self.runnable.kwargs.get("expire")
+            asset_hash = runnable.kwargs.get("asset_hash")
+            algorithm = runnable.kwargs.get("algorithm")
+            locations = runnable.kwargs.get("locations")
+            expire = runnable.kwargs.get("expire")
             if expire is not None:
                 expire = data_structures.time_to_seconds(str(expire))
 
-            cache_dirs = self.runnable.config.get("datadir.paths.cache_dirs")
+            cache_dirs = runnable.config.get("datadir.paths.cache_dirs")
             if cache_dirs is None:
                 cache_dirs = settings.as_dict().get("datadir.paths.cache_dirs")
-
-            # let's spawn it to another process to be able to update the
-            # status messages and avoid the Asset to lock this process
-            queue = SimpleQueue()
-            process = Process(
-                target=self._fetch_asset,
-                args=(
-                    name,
-                    asset_hash,
-                    algorithm,
-                    locations,
-                    cache_dirs,
-                    expire,
-                    queue,
-                ),
+            output = self._fetch_asset(
+                name, asset_hash, algorithm, locations, cache_dirs, expire
             )
-            process.start()
-
-            while queue.empty():
-                time.sleep(RUNNER_RUN_STATUS_INTERVAL)
-                yield self.prepare_status("running")
-
-            output = queue.get()
             result = output["result"]
             stdout = output["stdout"]
             stderr = output["stderr"]
@@ -104,7 +78,6 @@ def run(self, runnable):
             result = "error"
             stdout = ""
             stderr = 'At least name should be passed as kwargs using name="uri".'
-
         yield self.prepare_status("running", {"type": "stdout", "log": stdout.encode()})
         yield self.prepare_status("running", {"type": "stderr", "log": stderr.encode()})
         yield self.prepare_status("finished", {"result": result})