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) -> tuple[str, int]: 

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

35 

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

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

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

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

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

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

42 

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

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

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

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

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

48 """ 

49 popen_kwargs: dict[str, Any] = { 

50 "stdout": subprocess.PIPE, 

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

52 "text": True, 

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

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

55 "encoding": "utf-8", 

56 "errors": "replace", 

57 "env": env, 

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

59 } 

60 if bind_lifetime: 

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

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

63 else: 

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

65 argv, **popen_kwargs 

66 ) 

67 try: 

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

69 except BaseException: 

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

71 raise 

72 return stdout or "", proc.returncode 

73 

74 

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

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

77 if os.name == "posix": 

78 # start_new_session made the child its own group leader. 

79 with contextlib.suppress(OSError): 

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

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

82 proc.kill() 

83 try: 

84 proc.communicate(timeout=kill_wait_s) 

85 except subprocess.TimeoutExpired: 

86 log.warning( 

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

88 label, 

89 proc.pid, 

90 )