Start a monitoring thread for the given process
(self, process: RunningProcess)
| 1257 | self.shutdown_event = threading.Event() |
| 1258 | |
| 1259 | def start_monitoring(self, process: RunningProcess) -> None: |
| 1260 | """Start a monitoring thread for the given process""" |
| 1261 | if process in self.monitoring_threads: |
| 1262 | return # Already monitoring this process |
| 1263 | |
| 1264 | monitor_thread = threading.Thread( |
| 1265 | target=self._monitor_process, |
| 1266 | args=(process,), |
| 1267 | name=f"StuckMonitor-{_extract_test_name(process.command)}", |
| 1268 | daemon=True, |
| 1269 | ) |
| 1270 | self.monitoring_threads[process] = monitor_thread |
| 1271 | monitor_thread.start() |
| 1272 | |
| 1273 | def stop_monitoring(self, process: RunningProcess) -> None: |
| 1274 | """Stop monitoring a specific process""" |
no test coverage detected