Fix .processes so that it reuses the same objects

Author: wk9874

simvue/executor.py

@@ -22,7 +22,7 @@
 import pathlib
 import time
 import typing
-from simvue.api.objects.alert.fetch import Alert
+from simvue.api.objects.alert.user import UserAlert
 
 if typing.TYPE_CHECKING:
     import simvue
@@ -114,6 +114,7 @@ def __init__(self, simvue_runner: "simvue.Run", keep_logs: bool = True) -> None:
         self._alert_ids: dict[str, str] = {}
         self.command_str: dict[str, str] = {}
         self._processes: dict[str, subprocess.Popen] = {}
+        self._all_processes: list[psutil.Process] = []
 
     def std_out(self, process_id: str) -> str | None:
         if not os.path.exists(out_file := f"{self._runner.name}_{process_id}.out"):
@@ -271,19 +272,48 @@ def processes(self) -> list[psutil.Process]:
         if not self._processes:
             return []
 
-        _all_processes: list[psutil.Process] = []
+        _current_processes: list[psutil.Process] = []
 
         for process in self._processes.values():
             with contextlib.suppress(psutil.NoSuchProcess):
-                _all_processes.append(psutil.Process(process.pid))
+                _current_processes.append(psutil.Process(process.pid))
 
         with contextlib.suppress(psutil.NoSuchProcess, psutil.ZombieProcess):
-            for process in _all_processes:
+            for process in _current_processes:
                 for child in process.children(recursive=True):
-                    if child not in _all_processes:
-                        _all_processes.append(child)
+                    if child not in _current_processes:
+                        _current_processes.append(child)
 
-        return list(set(_all_processes))
+        _current_pids = set([_process.pid for _process in _current_processes])
+        _previous_pids = set([_process.pid for _process in self._all_processes])
+
+        # Find processes which used to exist, which are no longer running
+        _expired_process_pids = _previous_pids - _current_pids
+
+        # Remove these processes from list of all processes
+        self._all_processes = [
+            _process
+            for _process in self._all_processes
+            if _process.pid not in _expired_process_pids
+        ]
+
+        # Find new processes
+        _new_process_pids = _current_pids - _previous_pids
+        _new_processes = [
+            _process
+            for _process in _current_processes
+            if _process.pid in _new_process_pids
+        ]
+
+        # Get CPU usage stats for each of those new processes, so that next time it's measured by the heartbeat the value is accurate
+        if _new_processes:
+            [_process.cpu_percent() for _process in _new_processes]
+            time.sleep(0.1)
+
+        # Add these to the list of all processes
+        self._all_processes += _new_processes
+
+        return self._all_processes
 
     @property
     def success(self) -> int:
@@ -346,7 +376,7 @@ def _update_alerts(self) -> None:
             # We don't want to override the user's setting for the alert status
             # This is so that if a process incorrectly reports its return code,
             # the user can manually set the correct status depending on logs etc.
-            _alert = Alert(identifier=self._alert_ids[proc_id])
+            _alert = UserAlert(identifier=self._alert_ids[proc_id])
             _is_set = _alert.get_status(run_id=self._runner._id)
 
             if process.returncode != 0:

simvue/run.py

@@ -297,16 +297,11 @@ def processes(self) -> list[psutil.Process]:
             return process_list
 
         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 process_list:
-                    process_list.append(child)
+        process_list += self._child_processes
 
         return list(set(process_list))
 
-    def _get_sysinfo(self) -> dict[str, typing.Any]:
+    def _get_sysinfo(self, interval: float | None = None) -> dict[str, typing.Any]:
         """Retrieve system administration
 
         Parameters
@@ -320,7 +315,7 @@ def _get_sysinfo(self) -> dict[str, typing.Any]:
             retrieved system specifications
         """
         processes = self.processes
-        cpu = get_process_cpu(processes, interval=0.1)
+        cpu = get_process_cpu(processes, interval=interval)
         memory = get_process_memory(processes)
         gpu = get_gpu_metrics(processes)
         data: dict[str, typing.Any] = {}
@@ -359,7 +354,9 @@ def _heartbeat(
             last_res_metric_call = time.time()
 
             if self._resources_metrics_interval:
-                self._add_metrics_to_dispatch(self._get_sysinfo(), join_on_fail=False)
+                self._add_metrics_to_dispatch(
+                    self._get_sysinfo(interval=1), join_on_fail=False
+                )
 
             while not heartbeat_trigger.is_set():
                 time.sleep(0.1)
@@ -490,6 +487,9 @@ def _start(self, reconnect: bool = False) -> bool:
             self._pid = os.getpid()
 
         self._parent_process = psutil.Process(self._pid) if self._pid else None
+        self._child_processes = (
+            self._get_child_processes() if self._parent_process else None
+        )
 
         self._shutdown_event = threading.Event()
         self._heartbeat_termination_trigger = threading.Event()
@@ -904,6 +904,16 @@ def kill_all_processes(self) -> None:
             )
         self._executor.kill_all()
 
+    def _get_child_processes(self) -> list[psutil.Process]:
+        _process_list = []
+        # 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 _process_list:
+                    _process_list.append(child)
+
+        return list(set(_process_list))
+
     @property
     def executor(self) -> Executor:
         """Return the executor for this run"""
@@ -959,6 +969,13 @@ def set_pid(self, pid: int) -> None:
         """
         self._pid = pid
         self._parent_process = psutil.Process(self._pid)
+        self._child_processes = self._get_child_processes()
+        # Get CPU usage stats for each of those new processes, so that next time it's measured by the heartbeat the value is accurate
+        [
+            _process.cpu_percent()
+            for _process in self._child_processes + [self._parent_process]
+        ]
+        time.sleep(0.1)
 
     @skip_if_failed("_aborted", "_suppress_errors", False)
     @pydantic.validate_call
@@ -1962,7 +1979,7 @@ def log_alert(
             self._error("Please specify alert to update either by ID or by name.")
             return False
 
-        if self._user_config.run.mode == "offline":
+        if name and self._user_config.run.mode == "offline":
             self._error(
                 "Cannot retrieve alerts based on names in offline mode - please use IDs instead."
             )