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 <[email protected]>

Author: richtja

avocado/core/nrunner/runner.py

@@ -1,5 +1,6 @@
 import abc
 import multiprocessing
+import os
 import signal
 import time
 import traceback
@@ -8,6 +9,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
@@ -99,16 +101,37 @@ class PythonBaseRunner(BaseRunner, abc.ABC):
     Base class for Python runners
     """
 
-    @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
 
-    @staticmethod
-    def _monitor(queue):
+    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)
+
+    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 (
@@ -126,23 +149,26 @@ def _monitor(queue):
                     break
 
     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._run, 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
@@ -109,6 +110,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 on on 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": 661,
     "jobs": 11,
     "functional-parallel": 314,
-    "functional-serial": 7,
+    "functional-serial": 8,
     "optional-plugins": 0,
     "optional-plugins-golang": 2,
     "optional-plugins-html": 3,

selftests/functional/serial/basic.py

@@ -0,0 +1,53 @@
+import os
+import re
+import signal
+import time
+
+from avocado.utils import process, script
+from selftests.utils import AVOCADO, TestCaseTmpDir
+
+SLEEP_TEST = """import time
+
+from avocado import Test
+
+
+class SleepTest(Test):
+
+    def test(self):
+        self.log.debug("Sleeping starts: %s", time.time())
+        time.sleep(5)
+        self.log.debug("Sleeping ends: %s", time.time())
+"""
+
+
+class RunnerOperationTest(TestCaseTmpDir):
+    def test_pause(self):
+        with script.TemporaryScript(
+            "sleep.py",
+            SLEEP_TEST,
+        ) as 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(1)
+            proc.send_signal(signal.SIGTSTP)
+            time.sleep(10)
+            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("SleepTest.test: PAUSED", full_log)
+            self.assertIn("SleepTest.test: STARTED", full_log)
+            self.assertIn("Sleeping starts:", full_log)
+            self.assertIn("Sleeping ends:", full_log)
+            regex_start = re.search("Sleeping starts: ([0-9]*)", full_log)
+            regex_end = re.search("Sleeping ends: ([0-9]*)", full_log)
+            start_time = int(regex_start.group(1))
+            end_time = int(regex_end.group(1))
+            self.assertGreaterEqual(end_time - start_time, 10)