Removed threads from executor and add child process termination

Author: kzscisoft

simvue/executor.py

@@ -15,6 +15,7 @@
 import sys
 import multiprocessing
 import os
+import psutil
 import subprocess
 import pathlib
 import time
@@ -30,13 +31,8 @@ def _execute_process(
     proc_id: str,
     command: typing.List[str],
     runner_name: str,
-    exit_status_dict: typing.Dict[str, int],
-    std_err: typing.Dict[str, str],
-    std_out: typing.Dict[str, str],
-    run_on_exit: typing.Optional[typing.Callable[[int, int, str], None]],
-    trigger: typing.Optional[multiprocessing.synchronize.Event],
     environment: typing.Optional[typing.Dict[str, str]],
-) -> None:
+) -> subprocess.Popen:
     with open(f"{runner_name}_{proc_id}.err", "w") as err:
         with open(f"{runner_name}_{proc_id}.out", "w") as out:
             _result = subprocess.Popen(
@@ -47,24 +43,7 @@ def _execute_process(
                 env=environment,
             )
 
-    _status_code = _result.wait()
-    with open(f"{runner_name}_{proc_id}.err") as err:
-        std_err[proc_id] = err.read()
-
-    with open(f"{runner_name}_{proc_id}.out") as out:
-        std_out[proc_id] = out.read()
-
-    exit_status_dict[proc_id] = _status_code
-
-    if run_on_exit:
-        run_on_exit(
-            status_code=exit_status_dict[proc_id],
-            std_out=std_out[proc_id],
-            std_err=std_err[proc_id],
-        )
-
-    if trigger:
-        trigger.set()
+    return _result
 
 
 class Executor:
@@ -88,13 +67,14 @@ def __init__(self, simvue_runner: "simvue.Run", keep_logs: bool = True) -> None:
         """
         self._runner = simvue_runner
         self._keep_logs = keep_logs
-        self._manager = multiprocessing.Manager()
-        self._exit_codes = self._manager.dict()
-        self._std_err = self._manager.dict()
-        self._std_out = self._manager.dict()
+        self._completion_callbacks = {}
+        self._completion_triggers = {}
+        self._exit_codes = {}
+        self._std_err = {}
+        self._std_out = {}
         self._alert_ids: dict[str, str] = {}
-        self._command_str: typing.Dict[str, str] = {}
-        self._processes: typing.Dict[str, multiprocessing.Process] = {}
+        self._command_str: dict[str, str] = {}
+        self._processes: dict[str, subprocess.Popen] = {}
 
     def add_process(
         self,
@@ -207,26 +187,16 @@ def callback_function(status_code: int, std_out: str, std_err: str) -> None:
         _command += _pos_args
 
         self._command_str[identifier] = " ".join(_command)
+        self._completion_callbacks[identifier] = completion_callback
+        self._completion_triggers[identifier] = completion_trigger
 
-        self._processes[identifier] = multiprocessing.Process(
-            target=_execute_process,
-            args=(
-                identifier,
-                _command,
-                self._runner.name,
-                self._exit_codes,
-                self._std_err,
-                self._std_out,
-                completion_callback,
-                completion_trigger,
-                env,
-            ),
+        self._processes[identifier] = _execute_process(
+            identifier, _command, self._runner.name, env
         )
+
         self._alert_ids[identifier] = self._runner.create_alert(
             name=f"{identifier}_exit_status", source="user"
         )
-        logger.debug(f"Executing process: {' '.join(_command)}")
-        self._processes[identifier].start()
 
     @property
     def success(self) -> int:
@@ -324,12 +294,22 @@ def kill_process(self, process_id: str) -> None:
                 f"Failed to terminate process '{process_id}', no such identifier."
             )
             return
-        _process.kill()
+
+        _parent = psutil.Process(_process.pid)
+
+        for child in _parent.children(recursive=True):
+            logger.debug(f"Terminating child process {child.pid}: {child.name()}")
+            child.kill()
+
+        logger.debug(f"Terminating child process {_process.pid}: {_process.args}")
+        _process.terminate()
+
+        self._execute_callback(process_id)
 
     def kill_all(self) -> None:
         """Kill all running processes"""
-        for process in self._processes.values():
-            process.kill()
+        for process in self._processes.keys():
+            self.kill_process(process)
 
     def _clear_cache_files(self) -> None:
         """Clear local log files if required"""
@@ -338,11 +318,28 @@ def _clear_cache_files(self) -> None:
                 os.remove(f"{self._runner.name}_{proc_id}.err")
                 os.remove(f"{self._runner.name}_{proc_id}.out")
 
+    def _execute_callback(self, identifier: str) -> None:
+        with open(f"{self._runner.name}_{identifier}.err") as err:
+            std_err = err.read()
+
+        with open(f"{self._runner.name}_{identifier}.out") as out:
+            std_out = out.read()
+
+        if self._completion_callbacks[identifier]:
+            self._completion_callbacks[identifier](
+                status_code=self._processes[identifier].returncode,
+                std_out=std_out,
+                std_err=std_err,
+            )
+        if self._completion_triggers[identifier]:
+            self._completion_triggers[identifier].set()
+
     def wait_for_completion(self) -> None:
         """Wait for all processes to finish then perform tidy up and upload"""
-        for process in self._processes.values():
-            if process.is_alive():
-                process.join()
+        for identifier, process in self._processes.items():
+            process.wait()
+            self._execute_callback(identifier)
+
         self._update_alerts()
         self._save_output()