Kill all tracked PIO processes and their children.
(self)
| 126 | logger.debug("Cannot install signal handlers (not in main thread)") |
| 127 | |
| 128 | def _cleanup_all_processes(self) -> None: |
| 129 | """Kill all tracked PIO processes and their children.""" |
| 130 | with self._lock: |
| 131 | if not self._processes: |
| 132 | return |
| 133 | |
| 134 | logger.info( |
| 135 | f"Cleaning up {len(self._processes)} outstanding PIO processes..." |
| 136 | ) |
| 137 | pids_to_cleanup = list(self._processes.keys()) |
| 138 | |
| 139 | for pid in pids_to_cleanup: |
| 140 | self._kill_process_tree(pid) |
| 141 | |
| 142 | with self._lock: |
| 143 | self._processes.clear() |
| 144 | |
| 145 | def _kill_process_tree(self, pid: int) -> int: |
| 146 | """Kill a process and all its children. |
no test coverage detected