Suspend execution feature

This commit introduces the suspend execution feature to the nrunner. The
suspend execution was available on the legacy runner, but we didn't move
it to the nrunner. With this feature, it is possible to pause execution
of python based task on process spawner by sending SIGTSTP signal
(ctrl+z). It is helpful for debugging test execution.

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

Author: richtja

avocado/core/nrunner/runner.py

@@ -1,4 +1,5 @@
 import multiprocessing
+import os
 import signal
 import time
 import traceback
@@ -7,6 +8,7 @@
 from avocado.core.nrunner.runnable import RUNNERS_REGISTRY_STANDALONE_EXECUTABLE
 from avocado.core.plugin_interfaces import RunnableRunner
 from avocado.core.utils import messages
+from avocado.utils import process
 
 #: The amount of time (in seconds) between each internal status check
 RUNNER_RUN_CHECK_INTERVAL = 0.01
@@ -51,10 +53,29 @@ class BaseRunner(RunnableRunner):
     #: this runners makes use of.
     CONFIGURATION_USED = []
 
-    @staticmethod
-    def signal_handler(signum, frame):  # pylint: disable=W0613
+    def __init__(self):
+        super().__init__()
+        self.proc = None
+        self.process_stopped = False
+        self.stop_signal = False
+
+    def signal_handler(self, signum, frame):  # pylint: disable=W0613
         if signum == signal.SIGTERM.value:
             raise TestInterrupt("Test interrupted: Timeout reached")
+        elif signum == signal.SIGTSTP.value:
+            self.stop_signal = True
+
+    def pause_process(self):
+        if self.process_stopped:
+            self.process_stopped = False
+            sign = signal.SIGCONT
+        else:
+            self.process_stopped = True
+            sign = signal.SIGSTOP
+        processes = process.get_children_pids(self.proc.pid, recursive=True)
+        processes.append(self.proc.pid)
+        for pid in processes:
+            os.kill(pid, sign)
 
     @staticmethod
     def prepare_status(status_type, additional_info=None):
@@ -76,11 +97,13 @@ def prepare_status(status_type, additional_info=None):
         status.update({"status": status_type, "time": time.monotonic()})
         return status
 
-    @staticmethod
-    def _monitor(queue):
+    def _monitor(self, queue):
         most_recent_status_time = None
         while True:
             time.sleep(RUNNER_RUN_CHECK_INTERVAL)
+            if self.stop_signal:
+                self.stop_signal = False
+                self.pause_process()
             if queue.empty():
                 now = time.monotonic()
                 if (
@@ -118,23 +141,26 @@ def _catch_errors(self, runnable, queue):
             )
 
     def run(self, runnable):
-        # pylint: disable=W0201
+        if hasattr(signal, "SIGTSTP"):
+            signal.signal(signal.SIGTSTP, signal.SIG_IGN)
+            signal.signal(signal.SIGTSTP, self.signal_handler)
         signal.signal(signal.SIGTERM, self.signal_handler)
+        # pylint: disable=W0201
         self.runnable = runnable
         yield messages.StartedMessage.get()
         try:
             queue = multiprocessing.SimpleQueue()
-            process = multiprocessing.Process(
+            self.proc = multiprocessing.Process(
                 target=self._catch_errors, args=(self.runnable, queue)
             )
 
-            process.start()
+            self.proc.start()
 
             for message in self._monitor(queue):
                 yield message
 
         except TestInterrupt:
-            process.terminate()
+            self.proc.terminate()
             for message in self._monitor(queue):
                 yield message
         except Exception as e:

avocado/core/plugin_interfaces.py

@@ -376,6 +376,28 @@ async def terminate_task(self, runtime_task):
         :rtype: bool
         """
 
+    async def stop_task(self, runtime_task):
+        """Stop already spawned task.
+
+        :param runtime_task: wrapper for a Task with additional runtime
+                             information.
+        :type runtime_task: :class:`avocado.core.task.runtime.RuntimeTask`
+        :returns: whether the task has been stopped or not.
+        :rtype: bool
+        """
+        raise NotImplementedError()
+
+    async def resume_task(self, runtime_task):
+        """Resume already stopped task.
+
+        :param runtime_task: wrapper for a Task with additional runtime
+                             information.
+        :type runtime_task: :class:`avocado.core.task.runtime.RuntimeTask`
+        :returns: whether the task has been resumed or not.
+        :rtype: bool
+        """
+        raise NotImplementedError()
+
     @staticmethod
     @abc.abstractmethod
     async def check_task_requirements(runtime_task):

avocado/core/task/runtime.py

@@ -18,6 +18,7 @@ class RuntimeTaskStatus(Enum):
     FAIL_TRIAGE = "FINISHED WITH FAILURE ON TRIAGE"
     FAIL_START = "FINISHED FAILING TO START"
     STARTED = "STARTED"
+    PAUSED = "PAUSED"
 
     @staticmethod
     def finished_statuses():

avocado/core/task/statemachine.py

@@ -5,6 +5,7 @@
 import time
 
 from avocado.core.exceptions import JobFailFast
+from avocado.core.output import LOG_UI
 from avocado.core.task.runtime import RuntimeTaskStatus
 from avocado.core.teststatus import STATUSES_NOT_OK
 from avocado.core.utils import messages
@@ -493,6 +494,31 @@ async def terminate_tasks_interrupted(self):
         terminated = await self._terminate_tasks(task_status)
         await self._send_finished_tasks_message(terminated, "Interrupted by user")
 
+    @staticmethod
+    async def stop_resume_tasks(state_machine, spawner):
+        async with state_machine.lock:
+            try:
+                for runtime_task in state_machine.monitored:
+                    if runtime_task.status == RuntimeTaskStatus.STARTED:
+                        await spawner.stop_task(runtime_task)
+                        runtime_task.status = RuntimeTaskStatus.PAUSED
+                        LOG_UI.warning(
+                            f"{runtime_task.task.identifier}: {runtime_task.status.value}"
+                        )
+                    elif runtime_task.status == RuntimeTaskStatus.PAUSED:
+                        await spawner.resume_task(runtime_task)
+                        runtime_task.status = RuntimeTaskStatus.STARTED
+                        LOG_UI.warning(
+                            f"{runtime_task.task.identifier}: {runtime_task.status.value}"
+                        )
+            except NotImplementedError:
+                LOG.warning(
+                    f"Sending signals to tasks is not implemented for spawner: {spawner}"
+                )
+                LOG_UI.warning(
+                    f"Sending signals to tasks is not implemented for spawner: {spawner}"
+                )
+
     async def run(self):
         """Pushes Tasks forward and makes them do something with their lives."""
         while True:

avocado/plugins/runner_nrunner.py

@@ -21,7 +21,9 @@
 import os
 import platform
 import random
+import signal
 import tempfile
+import threading
 
 from avocado.core.dispatcher import SpawnerDispatcher
 from avocado.core.exceptions import JobError, JobFailFast
@@ -269,6 +271,10 @@ def _abort_if_missing_runners(runnables):
             )
             raise JobError(msg)
 
+    @staticmethod
+    def signal_handler(spawner, state_machine):
+        asyncio.create_task(Worker.stop_resume_tasks(state_machine, spawner))
+
     def run_suite(self, job, test_suite):
         summary = set()
 
@@ -335,6 +341,14 @@ def run_suite(self, job, test_suite):
         ]
         asyncio.ensure_future(self._update_status(job))
         loop = asyncio.get_event_loop()
+        if (
+            hasattr(signal, "SIGTSTP")
+            and threading.current_thread() is threading.main_thread()
+        ):
+            loop.add_signal_handler(
+                signal.SIGTSTP,
+                lambda: self.signal_handler(spawner, self.tsm),
+            )
         try:
             try:
                 loop.run_until_complete(

avocado/plugins/spawners/process.py

@@ -1,5 +1,6 @@
 import asyncio
 import os
+import signal
 import socket
 
 from avocado.core.dependencies.requirements import cache
@@ -106,6 +107,16 @@ async def terminate_task(self, runtime_task):
                 pass
         return returncode is not None
 
+    async def stop_task(self, runtime_task):
+        try:
+            runtime_task.spawner_handle.process.send_signal(signal.SIGTSTP)
+        except ProcessLookupError:
+            return False
+        return
+
+    async def resume_task(self, runtime_task):
+        await self.stop_task(runtime_task)
+
     @staticmethod
     async def check_task_requirements(runtime_task):
         """Check the runtime task requirements needed to be able to run"""

docs/source/guides/contributor/chapters/tips.rst

@@ -33,6 +33,27 @@ During the execution look for::
 
     avocado --show avocado.utils.debug run examples/tests/assets.py
 
+Interrupting test
+-----------------
+
+In case you want to "pause" the running test, you can use SIGTSTP (ctrl+z)
+signal sent to the main avocado process. This signal is forwarded to test
+and it's children processes. To resume testing you repeat the same signal.
+
+.. note::
+    The job and test timeouts are still enabled on stopped processes. This
+    means that after you restart the test can be killed by the timeout if
+    the timeout was reached. You can use run `-p timeout_factor=$int$` to
+    increase the timeouts for your debugging.
+
+.. note::
+    It is supported by the process spawner only.
+
+.. warning::
+    This feature is meant only for debugging purposes and it can
+    cause unreliable behavior especially if the signal is sent during the
+    test initialization. Therefore use it with caution.
+
 Line-profiler
 -------------
 

selftests/check.py

@@ -30,7 +30,7 @@
     "unit": 675,
     "jobs": 11,
     "functional-parallel": 315,
-    "functional-serial": 7,
+    "functional-serial": 9,
     "optional-plugins": 0,
     "optional-plugins-golang": 2,
     "optional-plugins-html": 3,

selftests/functional/serial/basic.py

@@ -0,0 +1,97 @@
+import glob
+import os
+import signal
+import sys
+import time
+import unittest
+
+from avocado.utils import process, script
+from selftests.utils import AVOCADO, TestCaseTmpDir
+
+SLEEP_TEST_PYTHON = """import os
+import time
+
+from avocado import Test
+
+
+class SleepTest(Test):
+
+    def test(self):
+        with open(os.path.join(self.logdir, "sleep.txt"), "w") as f:
+            for _ in range(10):
+                f.write("Sleeping \\n")
+                time.sleep(1)
+"""
+
+SLEEP_TEST_EXEC = """#!/bin/bash
+output_file="$AVOCADO_TEST_LOGDIR/sleep.txt"
+for i in {1..10}; do
+    echo "This is line $i" >> "$output_file"
+    sleep 1
+done
+"""
+
+
+@unittest.skipIf(
+    sys.platform.startswith("darwin"),
+    "The test pause feature is not supported on macOS",
+)
+class RunnerOperationTest(TestCaseTmpDir):
+    def _count_lines(self, file_path):
+        with open(os.path.join(file_path, "sleep.txt"), encoding="utf-8") as file:
+            return sum(1 for _ in file)
+
+    def _check_pause(self, tst):
+        cmd_line = f"{AVOCADO} run --disable-sysinfo --job-results-dir {self.tmpdir.name} -- {tst}"
+        proc = process.SubProcess(cmd_line)
+        proc.start()
+        init = True
+        while init:
+            output = proc.get_stdout()
+            if b"STARTED" in output:
+                init = False
+        time.sleep(2)
+        proc.send_signal(signal.SIGTSTP)
+        time.sleep(1)
+        test_log_dir = glob.glob(
+            os.path.join(self.tmpdir.name, "job-*", "test-results", "*")
+        )[0]
+        lines = self._count_lines(test_log_dir)
+        self.assertNotEqual(
+            lines,
+            10,
+            "The test finished before it was paused",
+        )
+        time.sleep(5)
+        self.assertEqual(
+            lines,
+            self._count_lines(test_log_dir),
+            "The test was not paused",
+        )
+        proc.send_signal(signal.SIGTSTP)
+        proc.wait()
+        full_log_path = os.path.join(self.tmpdir.name, "latest", "full.log")
+        with open(full_log_path, encoding="utf-8") as full_log_file:
+            full_log = full_log_file.read()
+        self.assertIn("PAUSED", full_log)
+        self.assertIn("STARTED", full_log)
+        self.assertEqual(
+            self._count_lines(test_log_dir),
+            10,
+            "The test was not resumed",
+        )
+
+    def test_pause_exec(self):
+        with script.TemporaryScript(
+            "sleep.sh",
+            SLEEP_TEST_EXEC,
+        ) as tst_exec:
+            self._check_pause(tst_exec)
+
+    def test_pause_instrumented(self):
+
+        with script.TemporaryScript(
+            "sleep.py",
+            SLEEP_TEST_PYTHON,
+        ) as tst_python:
+            self._check_pause(tst_python)