Coverage for src/lilbee/providers/fleet/proc.py: 100%

1"""Run a short-lived subprocess with a bounded reap. 

2 

3``subprocess.run``'s timeout path waits indefinitely for the killed child to be 

4reaped, so a process wedged in uninterruptible I/O (a GPU driver, a stuck 

5parser) hangs the caller forever. This kills the child's process group and 

6abandons it after a bounded wait if it will not die, so a wedged child costs a 

7bounded wait rather than a permanent hang. 

8""" 

9 

10from __future__ import annotations 

11 

12import contextlib 

13import logging 

14import os 

15import signal 

16import subprocess 

17from typing import Any 

18 

19from lilbee.providers.fleet.child_guard import spawn_bound_child 

20 

21log = logging.getLogger(__name__) 

22 

23 

24def run_bounded( 

25 argv: list[str], 

26 *, 

27 timeout_s: float, 

28 kill_wait_s: float, 

29 env: dict[str, str] | None = None, 

30 merge_stderr: bool = False, 

31 label: str | None = None, 

32 bind_lifetime: bool = False, 

33 cwd: str | os.PathLike[str] | None = None, 

34) -> tuple[str, int]: 

35 """Run *argv*, returning ``(stdout, returncode)``. 

36 

37 The child runs in its own session so its whole group can be killed. With 

38 *merge_stderr* stderr is folded into the returned stdout; otherwise it is 

39 discarded (the caller wants clean stdout, e.g. JSON). The group is SIGKILLed 

40 and awaited for at most *kill_wait_s* on timeout and on any other abort -- a 

41 Ctrl-C reaches this process, not the child's own session -- then the 

42 exception is re-raised, abandoning an unkillable child rather than waiting. 

43 

44 *bind_lifetime* additionally binds the child to this process, for a child 

45 that holds a resource nothing can reclaim by record (the device probe holds 

46 a GPU context and writes no state file). It is off by default because the 

47 Windows binding leaks a job-object handle per spawn by design, which is 

48 right for a handful of engines and wrong for a probe sampled every second. 

49 """ 

50 popen_kwargs: dict[str, Any] = { 

51 "stdout": subprocess.PIPE, 

52 "stderr": subprocess.STDOUT if merge_stderr else subprocess.DEVNULL, 

53 "text": True, 

54 # A probe writes its pipe in its own encoding; a GPU device name that is 

55 # not locale-decodable must not raise out of a sampling call. 

56 "encoding": "utf-8", 

57 "errors": "replace", 

58 "env": env, 

59 "cwd": cwd, 

60 "start_new_session": os.name == "posix", 

61 } 

62 if bind_lifetime: 

63 # Short-lived, so no death pipe: its watcher would outlive the child. 

64 proc = spawn_bound_child(argv, death_pipe=False, **popen_kwargs) 

65 else: 

66 proc = subprocess.Popen( # noqa: S603 - argv is trusted: a resolved binary or a fixed literal 

67 argv, **popen_kwargs 

68 ) 

69 try: 

70 stdout, _ = proc.communicate(timeout=timeout_s) 

71 except BaseException: 

72 _abandon_group(proc, kill_wait_s, label or argv[0]) 

73 raise 

74 return stdout or "", proc.returncode 

75 

76 

77def _abandon_group(proc: subprocess.Popen[str], kill_wait_s: float, label: str) -> None: 

78 """SIGKILL the child's group; log and give up if it cannot be reaped in time.""" 

79 if os.name == "posix": 

80 # start_new_session made the child its own group leader. 

81 with contextlib.suppress(OSError): 

82 os.killpg(proc.pid, signal.SIGKILL) 

83 else: # pragma: no cover - Windows has no process groups to kill 

84 proc.kill() 

85 try: 

86 proc.communicate(timeout=kill_wait_s) 

87 except subprocess.TimeoutExpired: 

88 log.warning( 

89 "%s (pid %d) ignored SIGKILL and was abandoned; it is likely wedged in a driver.", 

90 label, 

91 proc.pid, 

92 )