vibefpmgui.processes

 1"""Ending processes: a QProcess and everything it started."""
 2
 3from __future__ import annotations
 4
 5import os
 6import signal
 7import subprocess
 8
 9from qtpy.QtCore import QProcess
10
11from .console import NO_WINDOW
12
13
14def kill_process_tree(process: QProcess):
15    """Kill a QProcess *and everything it started* (fpm's compilers, the test
16    runner and the test program): killing only fpm would leave them running."""
17    pid = int(process.processId() or 0)
18    if pid > 0:
19        if os.name == "nt":
20            subprocess.run(["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True, check=False, **NO_WINDOW)
21        else:
22            for child in reversed(_descendants(pid)):  # deepest first
23                try:
24                    os.kill(child, signal.SIGKILL)
25                except OSError:
26                    pass  # already gone
27    process.kill()
28
29
30def _descendants(pid: int) -> list[int]:
31    """All processes below ``pid``, parents before children (POSIX)."""
32    try:
33        out = subprocess.run(["ps", "-A", "-o", "pid=", "-o", "ppid="],
34                             capture_output=True, text=True, timeout=5, check=False).stdout
35    except (OSError, subprocess.SubprocessError):
36        return []
37    children: dict[int, list[int]] = {}
38    for line in out.splitlines():
39        parts = line.split()
40        if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit():
41            children.setdefault(int(parts[1]), []).append(int(parts[0]))
42    found, queue = [], [pid]
43    while queue:
44        for child in children.get(queue.pop(0), []):
45            found.append(child)
46            queue.append(child)
47    return found