Function terminate_process_tree

scripts/run-test-wave.py:228–314

(active: ActiveSuite, kill_grace: int)

def terminate_process_tree(active: ActiveSuite, kill_grace: int) -> None:
    process = active.process
    leader_exited = process.poll() is not None
    if os.name == "nt":
        if leader_exited:
            # The leader can exit on its own between the timeout decision and
            # this call. Refusing outright made the harness itself lose a race:
            # a natural exit at the wrong moment failed the whole wave, which is
            # how a deliberately-hanging fixture suite reddened a release run.
            # taskkill /T cannot walk a tree from a dead PID, so prove cleanup
            # the only way still available -- nothing is parented to it.
            blocker = windows_tree_cleanup_blocker(process.pid)
            if blocker is not None:
                raise RuntimeError(
                    f"suite {active.name!r} leader exited and tree cleanup "
                    f"could not be proven: {blocker}"
                )
            return
        try:
            completed = subprocess.run(
                [
                    "taskkill.exe",
                    "/PID",
                    str(process.pid),
                    "/T",
                    "/F",
                ],
                check=False,
                stdin=subprocess.DEVNULL,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                timeout=kill_grace,
            )
        except (OSError, subprocess.TimeoutExpired):
            completed = None
        if completed is None or completed.returncode != 0:
            if process.poll() is None:
                process.kill()
                try:
                    process.wait(timeout=kill_grace)
                except subprocess.TimeoutExpired:
                    pass
            raise RuntimeError(
                f"suite {active.name!r} taskkill could not prove process-tree cleanup"
            )
        try:
            process.wait(timeout=kill_grace)
        except subprocess.TimeoutExpired as exc:
            process.kill()
            raise RuntimeError(
                f"suite {active.name!r} process tree resisted forced termination"
            ) from exc
        return

    def group_active() -> bool:
        try:
            os.killpg(process.pid, 0)
            return True