Merge pull request #416 from simvue-io/hotfix/manual-added-processes

Attach manual processes to executor and executor processes to resource monitoring

Author: james-panayis

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "simvue"
-version = "1.0.0.dev2"
+version = "1.0.0"
 description = "Simulation tracking and monitoring"
 authors = ["Simvue Development Team <[email protected]>"]
 license = "Apache v2"

simvue/executor.py

@@ -11,6 +11,7 @@
 __date__ = "2023-11-15"
 
 import logging
+import contextlib
 import multiprocessing.synchronize
 import sys
 import multiprocessing
@@ -205,6 +206,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"""
@@ -294,15 +313,31 @@ 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
+            parent = psutil.Process(process.pid)
+        elif isinstance(process_id, int):
+            parent = psutil.Process(process_id)
 
         for child in parent.children(recursive=True):
             logger.debug(f"Terminating child process {child.pid}: {child.name()}")
@@ -311,11 +346,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 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

@@ -216,17 +216,21 @@ def time_stamp(self) -> str:
     @property
     def processes(self) -> list[psutil.Process]:
         """Create an array containing a list of processes"""
+
+        process_list = self._executor.processes
+
         if not self._parent_process:
-            return []
+            return process_list
 
-        _all_processes: list[psutil.Process] = [self._parent_process]
+        process_list += [self._parent_process]
 
+        # Attach child processes relating to the process set by set_pid
         with contextlib.suppress(psutil.NoSuchProcess, psutil.ZombieProcess):
             for child in self._parent_process.children(recursive=True):
-                if child not in _all_processes:
-                    _all_processes.append(child)
+                if child not in process_list:
+                    process_list.append(child)
 
-        return list(set(_all_processes))
+        return list(set(process_list))
 
     def _get_sysinfo(self) -> dict[str, typing.Any]:
         """Retrieve system administration
@@ -779,6 +783,14 @@ def kill_process(self, process_id: str) -> None:
 
     def kill_all_processes(self) -> None:
         """Kill all currently running processes."""
+        # Dont kill the manually attached process if it is the current script
+        # but do kill its children. The kill process method of executor by
+        # default refers to its own processes but can also be used on a PID
+        if self._parent_process:
+            self._executor.kill_process(
+                process_id=self._parent_process.pid,
+                kill_children_only=self._parent_process.pid == os.getpid(),
+            )
         self._executor.kill_all()
 
     @property