session: implement forcibly stop

Change the way we complete session by introducing forcibly stop. The
way that kirk works after this patch is the following:

- if user sends SIGINT or CTRL+C, we simply wait for the running
  tests to complete and finally stop the session

- if user sends SIGINT or CTRL+C again, we stop any execution on the SUT
  and we terminate the testing suite, flagging the killed test as TSKIP

Signed-off-by: Andrea Cervesato <[email protected]>
Closes: #78

Author: acerv

libkirk/ltp.py

@@ -10,6 +10,7 @@
 import logging
 import os
 import re
+import signal
 from typing import (
     Any,
     Dict,

libkirk/main.py

@@ -460,20 +460,12 @@ async def session_run() -> None:
         exit_code = RC_ERROR
     finally:
         try:
-            # at this point loop has been closed, so we can collect all
-            # tasks and cancel them
-            loop.run_until_complete(
-                # pyrefly: ignore[bad-argument-type]
-                asyncio.gather(
-                    *[
-                        session.stop(),
-                        libkirk.events.stop(),
-                    ]
-                )
-            )
-            libkirk.cancel_tasks(loop)
+            loop.run_until_complete(session.stop())
         except KeyboardInterrupt:
-            pass
+            loop.run_until_complete(session.stop())
+
+        libkirk.cancel_tasks(loop)
+        loop.run_until_complete(libkirk.events.stop())
 
     parser.exit(exit_code)
 

libkirk/scheduler.py

@@ -9,6 +9,7 @@
 import asyncio
 import logging
 import os
+import signal
 import sys
 import time
 from typing import (
@@ -147,7 +148,7 @@ def __init__(
         self._timeout = 0.0 if timeout < 0.0 else timeout
         self._max_workers = 1 if max_workers < 1 else max_workers
         self._results = []
-        self._stop = False
+        self._stop_cnt = 0
         self._stopped = False
         self._running_tests_sem = asyncio.Semaphore(1)
         self._schedule_lock = asyncio.Lock()
@@ -203,7 +204,12 @@ def stopped(self) -> bool:
 
     async def stop(self) -> None:
         self._logger.info("Stopping tests execution")
-        self._stop = True
+        self._stop_cnt += 1
+
+        if self._stop_cnt > 1:
+            # by stopping SUT first, we cause scheduler to complete
+            # current test immediatelly without waiting
+            await self._sut.stop()
 
         try:
             # we enter in the semaphore queue in order to get highest
@@ -217,7 +223,7 @@ async def stop(self) -> None:
             async with self._schedule_lock:
                 pass
         finally:
-            self._stop = False
+            self._stop_cnt = 0
             self._stopped = True
 
         self._logger.info("All tests have been completed")
@@ -227,7 +233,7 @@ async def _run_test(self, test: Test) -> None:
         Run a single test and populate the results array.
         """
         async with self._running_tests_sem:
-            if self._stop:
+            if self._stop_cnt > 0:
                 self._logger.info("Test '%s' has been stopped", test.name)
                 return None
 
@@ -284,6 +290,19 @@ async def _run_test(self, test: Test) -> None:
                 except asyncio.TimeoutError:
                     status = self.KERNEL_TIMEOUT
 
+            # raise kernel errors at the end so we can collect test results
+            if status == self.KERNEL_TAINTED:
+                await libkirk.events.fire("kernel_tainted", tainted_msg)
+                raise KernelTaintedError()
+
+            if status == self.KERNEL_PANIC:
+                await libkirk.events.fire("kernel_panic")
+                raise KernelPanicError()
+
+            if status == self.KERNEL_TIMEOUT:
+                await libkirk.events.fire("sut_not_responding")
+                raise KernelTimeoutError()
+
             # create test results and save it
             if status not in [self.STATUS_OK, self.KERNEL_TAINTED]:
                 test_data = {
@@ -294,6 +313,12 @@ async def _run_test(self, test: Test) -> None:
                     "exec_time": exec_time,
                 }
 
+            # we won't consider killed tests, unless they have been
+            # killed by kirk by forcibly stop
+            if test_data["returncode"] == -signal.SIGKILL:
+                self._logger.info("Test killed: %s", test.name)
+                return
+
             results = await self._framework.read_result(
                 test,
                 test_data["stdout"],
@@ -304,19 +329,6 @@ async def _run_test(self, test: Test) -> None:
             self._logger.debug("results=%s", results)
             self._results.append(results)
 
-            # raise kernel errors at the end so we can collect test results
-            if status == self.KERNEL_TAINTED:
-                await libkirk.events.fire("kernel_tainted", tainted_msg)
-                raise KernelTaintedError()
-
-            if status == self.KERNEL_PANIC:
-                await libkirk.events.fire("kernel_panic")
-                raise KernelPanicError()
-
-            if status == self.KERNEL_TIMEOUT:
-                await libkirk.events.fire("sut_not_responding")
-                raise KernelTimeoutError()
-
             await libkirk.events.fire("test_completed", results)
             await self._write_kmsg(test, results)
 
@@ -380,11 +392,10 @@ async def schedule(self, jobs: List[Any]) -> None:
                 exc_name = err.__class__.__name__
                 self._logger.info("%s caught during tests execution", exc_name)
 
-                raise_exc = not self._stop
                 async with self._running_tests_sem:
                     pass
 
-                if raise_exc:
+                if self._stop_cnt == 0:
                     self._logger.info("Propagating %s exception", exc_name)
                     raise err
 

libkirk/session.py

@@ -90,7 +90,8 @@ def __init__(
         self._sut = sut
         self._exec_timeout = exec_timeout
         self._force_parallel = force_parallel
-        self._stop = False
+        self._stop_cnt = 0
+        self._stop_lock = asyncio.Lock()
         self._exec_lock = asyncio.Lock()
         self._run_lock = asyncio.Lock()
         self._results = []
@@ -356,7 +357,7 @@ async def _exec_command(self, command: str) -> None:
             except asyncio.TimeoutError:
                 exc = KirkException(f"Command timeout: {repr(command)}")
             except KirkException as err:
-                if not self._stop:
+                if self._stop_cnt == 0:
                     exc = err
 
             if exc:
@@ -369,13 +370,16 @@ async def _inner_stop(self) -> None:
         if self._scheduler:
             await self._scheduler.stop()
 
-        await self._stop_sut()
+        if self._stop_cnt < 1:
+            await self._stop_sut()
 
     async def stop(self) -> None:
         """
         Stop the current session.
         """
-        self._stop = True
+        async with self._stop_lock:
+            self._stop_cnt += 1
+
         try:
             await self._inner_stop()
 
@@ -385,8 +389,12 @@ async def stop(self) -> None:
             async with self._exec_lock:
                 pass
         finally:
-            await libkirk.events.fire("session_stopped")
-            self._stop = False
+            if self._stop_cnt == 1:
+                await libkirk.events.fire("session_stopped")
+            elif self._stop_cnt > 1:
+                await libkirk.events.fire("session_killed")
+
+            self._stop_cnt = 0
 
     async def _schedule_once(self, suites_obj: List[Suite]) -> None:
         """
@@ -403,7 +411,7 @@ async def _schedule_infinite(self, suites_obj: List[Suite]) -> None:
         suites_list.extend(suites_obj)
 
         count = 1
-        while not self._stop:
+        while self._stop_cnt == 0:
             await self._schedule_once(suites_obj)
             if self._scheduler.stopped:
                 break
@@ -516,7 +524,7 @@ async def run(
 
                     await self._run_scheduler(suites_obj, runtime)
             except KirkException as err:
-                if not self._stop:
+                if self._stop_cnt == 0:
                     self._logger.exception(err)
                     await libkirk.events.fire("session_error", str(err))
                     raise err

libkirk/ui.py

@@ -46,6 +46,7 @@ def __init__(self, no_colors: bool = False) -> None:
         libkirk.events.register("session_restore", self.session_restore)
         libkirk.events.register("session_started", self.session_started)
         libkirk.events.register("session_stopped", self.session_stopped)
+        libkirk.events.register("session_killed", self.session_killed)
         libkirk.events.register("sut_start", self.sut_start)
         libkirk.events.register("sut_stop", self.sut_stop)
         libkirk.events.register("sut_restart", self.sut_restart)
@@ -128,6 +129,9 @@ async def session_started(self, tmpdir: str) -> None:
     async def session_stopped(self) -> None:
         await self._print("Session stopped")
 
+    async def session_killed(self) -> None:
+        await self._print("Session killed")
+
     async def sut_start(self, sut: str) -> None:
         await self._print(f"Connecting to SUT: {sut}")