Check for any stuck process signals from monitoring threads
(self)
| 1281 | self.monitoring_threads.clear() |
| 1282 | |
| 1283 | def check_for_stuck_processes(self) -> list[StuckProcessSignal]: |
| 1284 | """Check for any stuck process signals from monitoring threads""" |
| 1285 | stuck_processes: list[StuckProcessSignal] = [] |
| 1286 | try: |
| 1287 | while True: |
| 1288 | signal = self.stuck_signals.get_nowait() |
| 1289 | stuck_processes.append(signal) |
| 1290 | except queue.Empty: |
| 1291 | pass |
| 1292 | return stuck_processes |
| 1293 | |
| 1294 | def _monitor_process(self, process: RunningProcess) -> None: |
| 1295 | """Monitor a single process for being stuck (runs in separate thread)""" |