Re-apply lost changes

Author: kzscisoft

simvue/executor.py

@@ -17,6 +17,7 @@
 import os
 import psutil
 import subprocess
+import contextlib
 import pathlib
 import time
 import typing
@@ -229,6 +230,24 @@ def callback_function(status_code: int, std_out: str, std_err: str) -> None:
             name=f"{identifier}_exit_status", source="user"
         )
 
+    @property
+    def processes(self) -> list[psutil.Process]:
+        """Create an array containing a list of processes"""
+        if not self._processes:
+            return []
+
+        _all_processes: list[psutil.Process] = [
+            psutil.Process(process.pid) for process in self._processes.values()
+        ]
+
+        with contextlib.suppress(psutil.NoSuchProcess, psutil.ZombieProcess):
+            for process in _all_processes:
+                for child in process.children(recursive=True):
+                    if child not in _all_processes:
+                        _all_processes.append(child)
+
+        return list(set(_all_processes))
+
     @property
     def success(self) -> int:
         """Return whether all attached processes completed successfully"""
@@ -320,15 +339,37 @@ def _save_output(self) -> None:
                     f"{self._runner.name}_{proc_id}.out", category="output"
                 )
 
-    def kill_process(self, process_id: str) -> None:
-        """Kill a running process by ID"""
-        if not (process := self._processes.get(process_id)):
-            logger.error(
-                f"Failed to terminate process '{process_id}', no such identifier."
-            )
-            return
+    def kill_process(
+        self, process_id: typing.Union[int, str], kill_children_only: bool = False
+    ) -> None:
+        """Kill a running process by ID
 
-        parent = psutil.Process(process.pid)
+        If argument is a string this is a process handled by the client,
+        else it is a PID of a external monitored process
+
+        Parameters
+        ----------
+        process_id : typing.Union[int, str]
+            either the identifier for a client created process or the PID
+            of an external process
+        kill_children_only : bool, optional
+            if process_id is an integer, whether to kill only its children
+        """
+        if isinstance(process_id, str):
+            if not (process := self._processes.get(process_id)):
+                logger.error(
+                    f"Failed to terminate process '{process_id}', no such identifier."
+                )
+                return
+            try:
+                parent = psutil.Process(process.pid)
+            except psutil.NoSuchProcess:
+                return
+        elif isinstance(process_id, int):
+            try:
+                parent = psutil.Process(process_id)
+            except psutil.NoSuchProcess:
+                return
 
         for child in parent.children(recursive=True):
             logger.debug(f"Terminating child process {child.pid}: {child.name()}")
@@ -337,17 +378,13 @@ def kill_process(self, process_id: str) -> None:
         for child in parent.children(recursive=True):
             child.wait()
 
-        logger.debug(f"Terminating child process {process.pid}: {process.args}")
-        process.kill()
-        process.wait()
-
-        if trigger := self._completion_triggers.get(process_id):
-            trigger.set()
-
-        if trigger_process := self._completion_processes.get(process_id):
-            trigger_process.join()
+        if not kill_children_only:
+            logger.debug(f"Terminating process {process.pid}: {process.args}")
+            process.kill()
+            process.wait()
 
-        self._execute_callback(process_id)
+        if isinstance(process_id, str):
+            self._execute_callback(process_id)
 
     def kill_all(self) -> None:
         """Kill all running processes"""

simvue/run.py

@@ -298,6 +298,7 @@ def _heartbeat(
                         and self._abort_on_alert
                         and self._simvue.get_abort_status()
                     ):
+                        logger.debug("Received abort request from server")
                         self._alert_raised_trigger.set()
                         self.kill_all_processes()
                         if self._dispatcher and self._shutdown_event:

tests/refactor/test_run_class.py

@@ -487,23 +487,6 @@ def testing_exit(status: int) -> None:
         run.kill_all_processes()
         raise AssertionError("Run was not terminated")
 
-@pytest.mark.run
-def test_abort_on_alert_raise(create_plain_run: typing.Tuple[sv_run.Run, dict], mocker: pytest_mock.MockerFixture) -> None:
-    def testing_exit(status: int) -> None:
-        raise SystemExit(status)
-    mocker.patch("os._exit", testing_exit)
-    run, _ = create_plain_run
-    run.config(resources_metrics_interval=1)
-    run._heartbeat_interval = 1
-    run._testing = True
-    alert_id = run.create_alert("abort_test", source="user", trigger_abort=True)
-    run.add_process(identifier="forever_long", executable="bash", c="sleep 10")
-    time.sleep(2)
-    run.log_alert(alert_id, "critical")
-    time.sleep(4)
-    if not run._status == "terminated":
-        run.kill_all_processes()
-        raise AssertionError("Run was not terminated")
 
 @pytest.mark.run
 def test_abort_on_alert_python(create_plain_run: typing.Tuple[sv_run.Run, dict], mocker: pytest_mock.MockerFixture) -> None:
@@ -530,6 +513,25 @@ def testing_exit(status: int) -> None:
     assert run._status == "terminated"
 
 
+@pytest.mark.run
+def test_abort_on_alert_raise(create_plain_run: typing.Tuple[sv_run.Run, dict], mocker: pytest_mock.MockerFixture) -> None:
+    def testing_exit(status: int) -> None:
+        raise SystemExit(status)
+    mocker.patch("os._exit", testing_exit)
+    run, _ = create_plain_run
+    run.config(resources_metrics_interval=1)
+    run._heartbeat_interval = 1
+    run._testing = True
+    alert_id = run.create_alert("abort_test", source="user", trigger_abort=True)
+    run.add_process(identifier="forever_long", executable="bash", c="sleep 10")
+    time.sleep(2)
+    run.log_alert(alert_id, "critical")
+    time.sleep(4)
+    if not run._status == "terminated":
+        run.kill_all_processes()
+        raise AssertionError("Run was not terminated")
+
+
 @pytest.mark.run
 def test_kill_all_processes(create_plain_run: typing.Tuple[sv_run.Run, dict]) -> None:
     run, _ = create_plain_run