Python 子进程超时后仍在写文件？在 Linux 用进程组停止工作进程

python3 --version
mkdir process-group-demo
cd process-group-demo

"""Run a trusted Linux command; stop its process group on timeout. Python 3.9+."""
import argparse
import math
import os
from pathlib import Path
import signal
import subprocess
import sys
import time


def signal_group(pgid, signum):
    try:
        os.killpg(pgid, signum)
    except ProcessLookupError:
        pass  # The group has already disappeared.


def run_group(argv, log_path, *, timeout=60.0, grace=1.0):
    if isinstance(argv, (str, bytes)) or not argv:
        raise ValueError("argv must be a nonempty argument list")
    if not math.isfinite(timeout) or timeout <= 0:
        raise ValueError("timeout must be finite and positive")
    if not math.isfinite(grace) or grace < 0:
        raise ValueError("grace must be finite and nonnegative")
    with Path(log_path).open("xb") as log:
        process = subprocess.Popen(
            argv, stdin=subprocess.DEVNULL,
            stdout=log, stderr=subprocess.STDOUT,
            shell=False, start_new_session=True,
        )
        try:
            code = process.wait(timeout=timeout)
        except subprocess.TimeoutExpired:
            # Do not poll/wait/reap the leader before the final group signal.
            try:
                signal_group(process.pid, signal.SIGTERM)
                time.sleep(grace)
            finally:
                signal_group(process.pid, signal.SIGKILL)
                process.wait()  # Reap our direct child after group cleanup.
            raise TimeoutError("process group timeout")
        return code if code >= 0 else 128 - code


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("--timeout", type=float, default=60.0)
    parser.add_argument("--grace", type=float, default=1.0)
    parser.add_argument("log", type=Path)
    parser.add_argument("command", nargs=argparse.REMAINDER)
    args = parser.parse_args()
    command = args.command
    if command[:1] == ["--"]:
        command = command[1:]
    if not command:
        parser.error("provide a command after the log path and --")
    try:
        code = run_group(command, args.log, timeout=args.timeout, grace=args.grace)
    except TimeoutError:
        print(f"TIMEOUT: log={args.log}", file=sys.stderr)
        return 124
    except (OSError, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2
    if code:
        print(f"FAILED: exit={code}; log={args.log}", file=sys.stderr)
    else:
        print(f"OK: log={args.log}")
    return code


if __name__ == "__main__":
    raise SystemExit(main())

python3 run_group.py --help
python3 run_group.py --timeout 5 ok.log -- python3 -c 'print("processed 3 records")'
cat ok.log

OK: log=ok.log
processed 3 records

python3 run_group.py --timeout 5 fail.log -- python3 -c 'import sys; print("bad input", file=sys.stderr); sys.exit(7)'
printf 'exit=%s\n' "$?"
cat fail.log

FAILED: exit=7; log=fail.log
exit=7
bad input

"""A direct child plus a worker that ignores SIGTERM; Linux demo only."""
from pathlib import Path
import subprocess
import sys
import time

worker_code = """
from pathlib import Path
import os
import signal
import time
signal.signal(signal.SIGTERM, signal.SIG_IGN)
Path('worker.pid').write_text(str(os.getpid()), encoding='ascii')
time.sleep(6)
Path('late.txt').write_text('worker survived', encoding='utf-8')
"""
worker = subprocess.Popen([sys.executable, "-c", worker_code])
# Print only after the worker installed its handler and recorded its PID.
deadline = time.monotonic() + 3
while not Path("worker.pid").exists():
    if time.monotonic() > deadline:
        raise RuntimeError("worker did not become ready")
    time.sleep(0.01)
print(f"READY: worker={worker.pid}", flush=True)
worker.wait()

python3 run_group.py --timeout 1 --grace 0.5 tree.log -- python3 tree_job.py
printf 'exit=%s\n' "$?"
cat tree.log

python3 - <<'PY_CHECK'
from pathlib import Path
import time

pid = int(Path("worker.pid").read_text(encoding="ascii"))
def still_running(pid):
    try:
        status = Path(f"/proc/{pid}/status").read_text()
    except FileNotFoundError:
        return False
    state = next(line for line in status.splitlines() if line.startswith("State:"))
    return state.split()[1] not in {"Z", "X"}

deadline = time.monotonic() + 2
while still_running(pid) and time.monotonic() < deadline:
    time.sleep(0.05)
assert not still_running(pid), "worker is still running"
time.sleep(6)
assert not Path("late.txt").exists(), "worker wrote after timeout"
print("PASS: worker stopped; no late output")
PY_CHECK

python3 - <<'PY_CHECK'
from pathlib import Path
import subprocess
import sys

before = Path("ok.log").read_bytes()
result = subprocess.run(
    [sys.executable, "run_group.py", "ok.log", "--",
     sys.executable, "-c", "print('replacement')"],
    capture_output=True, text=True,
)
assert result.returncode == 2, result.stderr
assert Path("ok.log").read_bytes() == before
print("PASS: existing log preserved")
PY_CHECK