Register a PIO subprocess for tracking. Args: proc: subprocess.Popen instance to track
(self, proc: subprocess.Popen[Any])
| 40 | self._signal_handlers_installed = False |
| 41 | |
| 42 | def register(self, proc: subprocess.Popen[Any]) -> None: |
| 43 | """Register a PIO subprocess for tracking. |
| 44 | |
| 45 | Args: |
| 46 | proc: subprocess.Popen instance to track |
| 47 | """ |
| 48 | pid = proc.pid |
| 49 | if pid is None: |
| 50 | return |
| 51 | |
| 52 | with self._lock: |
| 53 | self._processes[pid] = proc |
| 54 | self._ensure_cleanup_handlers() |
| 55 | logger.debug(f"Registered PIO process: pid={pid}") |
| 56 | |
| 57 | def unregister(self, proc: subprocess.Popen[Any]) -> None: |
| 58 | """Unregister a PIO subprocess (called when process completes normally). |
no test coverage detected