Kill all BG processes which are still running.
(self)
| 259 | print("All background processes were killed.") |
| 260 | |
| 261 | def kill_bg_processes(self): |
| 262 | """Kill all BG processes which are still running.""" |
| 263 | if not self.bg_processes: |
| 264 | return |
| 265 | for p in self.bg_processes: |
| 266 | if p.poll() is None: |
| 267 | try: |
| 268 | p.send_signal(signal.SIGINT) |
| 269 | except: |
| 270 | pass |
| 271 | time.sleep(0.1) |
| 272 | self._gc_bg_processes() |
| 273 | if not self.bg_processes: |
| 274 | return |
| 275 | for p in self.bg_processes: |
| 276 | if p.poll() is None: |
| 277 | try: |
| 278 | p.terminate() |
| 279 | except: |
| 280 | pass |
| 281 | time.sleep(0.1) |
| 282 | self._gc_bg_processes() |
| 283 | if not self.bg_processes: |
| 284 | return |
| 285 | for p in self.bg_processes: |
| 286 | if p.poll() is None: |
| 287 | try: |
| 288 | p.kill() |
| 289 | except: |
| 290 | pass |
| 291 | self._gc_bg_processes() |
| 292 | |
| 293 | def _gc_bg_processes(self): |
| 294 | self.bg_processes = [p for p in self.bg_processes if p.poll() is None] |
no test coverage detected