"""Практика к уроку про сигналы: пять ответов и одна доля.

ЗАЧЕМ ОТДЕЛЬНЫЙ ФАЙЛ. Задачи урока проверяются сборкой (scripts/validate-practice.mjs):
показанный читателю код обязан построчно быть в скрипте, верный вариант —
дословно встречаться в записи прогона, а заявленное число — печататься самой
программой, а не считаться редакцией в уме. Поэтому здесь ровно то, что видит
читатель, и ничего больше.

ПОЧЕМУ КАЖДАЯ ИЗ ПЯТИ СТРОК ИЗМЕРЕНА, А НЕ ВПИСАНА. Первый черновик печатал
единицу третьей строкой прямо литералом: эксперимент с сиротой запускался,
его вывод отбрасывался, а «1» бралась из головы автора. Строка была бы верной
и совершенно бесполезной — она ничего не проверяла. Теперь номер родителя
читается из вывода внука, и если ядро когда-нибудь передаст сироту не первому
процессу, задача сломается, как и должна.

ЧТО ЗДЕСЬ ИЗМЕРЯЕТСЯ ОТДЕЛЬНО. Вторая половина — цена: сколько потомков
останутся зомби, если не звать `wait`, и сколько времени занимает остановка
процесса, который обрабатывает `SIGTERM`, против того, который его
игнорирует. Оркестратор ждёт вторую сторону весь льготный срок и только потом
посылает `SIGKILL`; здесь срок сокращён до двух секунд, чтобы прогон был
быстрым.

ЗАПУСК: python3 bench/signals/practice.py
Вывод: runs/practice.txt
"""

import os
import signal
import subprocess
import sys
import time

GRACE = 2.0
CHILDREN = 1000

SLEEPER = "import signal, sys, time\ntime.sleep(30)\n"
HANDLED = (
    "import signal, sys, time\n"
    "signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))\n"
    "time.sleep(30)\n"
)


def child_of(pid: int) -> int | None:
    """Потомок процесса — тот, кто внутри пространства имён получил PID 1."""
    try:
        with open(f"/proc/{pid}/task/{pid}/children", encoding="utf-8") as fh:
            kids = fh.read().split()
    except FileNotFoundError:
        return None
    return int(kids[0]) if kids else None


def state_of(pid: int) -> str:
    try:
        with open(f"/proc/{pid}/stat", encoding="utf-8") as fh:
            data = fh.read()
    except FileNotFoundError:
        return "no such process"
    return data[data.rfind(")") + 2 :].split()[0]


def orphan_ppid() -> int:
    """Номер родителя процесса, чей родитель уже умер.

    Внук печатает `os.getppid()` ПОСЛЕ смерти отца; его вывод и есть ответ.
    Читается он из канала, а не подставляется автором.
    """
    parent = (
        "import subprocess, sys\n"
        "subprocess.Popen([sys.executable, '-c', "
        "'import os, time; time.sleep(0.5); print(os.getppid())'])\n"
    )
    done = subprocess.run(
        [sys.executable, "-c", parent], capture_output=True, text=True, check=True
    )
    return int(done.stdout.split()[-1])


def pid1_state_after_sigterm(program: str) -> str:
    """Состояние первого процесса пространства имён через секунду после SIGTERM.

    Ответ задачи — однобуквенное состояние из /proc, а не русское слово:
    урок существует в двух языках и цитирует эту запись дословно обоими.
    """
    proc = subprocess.Popen(
        ["unshare", "--fork", "--pid", "--mount-proc", sys.executable, "-c", program],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    time.sleep(0.3)
    target = child_of(proc.pid)
    if target is None:
        proc.kill()
        proc.wait()
        return "no child of unshare"
    os.kill(target, signal.SIGTERM)
    time.sleep(1.0)
    state = state_of(target)
    proc.kill()
    proc.wait()
    return state


def stop_time(program: str) -> float:
    """Сколько занимает остановка процесса-PID-1 по правилам оркестратора.

    Оркестратор посылает SIGTERM, ждёт льготный срок и добивает SIGKILL.
    Возвращается время от SIGTERM до фактического конца процесса.
    """
    # stderr гасится намеренно: когда родителя-`unshare` добивают, он пишет
    # «sigprocmask unblock failed» — жалобу собственной уборки, к измеряемому
    # отношения не имеющую. В записи прогона она была бы шумом, который
    # читателю нечем объяснить.
    proc = subprocess.Popen(
        ["unshare", "--fork", "--pid", "--mount-proc", sys.executable, "-c", program],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    time.sleep(0.3)
    target = child_of(proc.pid)
    if target is None:
        proc.kill()
        proc.wait()
        return float("nan")

    started = time.perf_counter()
    os.kill(target, signal.SIGTERM)
    deadline = started + GRACE
    while time.perf_counter() < deadline:
        if state_of(target) in ("no such process", "Z"):
            break
        time.sleep(0.005)
    else:
        os.kill(target, signal.SIGKILL)
        while state_of(target) not in ("no such process", "Z"):
            time.sleep(0.005)
    elapsed = time.perf_counter() - started

    proc.kill()
    proc.wait()
    return elapsed


def main() -> None:
    print(f"Python {sys.version.split()[0]} · Linux {os.uname().release}")
    print(f"orchestrator grace period in this run: {GRACE} s")
    print()

    # --- Часть 1: пять ответов о поведении ------------------------------
    pid = os.fork()
    if pid == 0:
        os._exit(7)
    time.sleep(0.2)
    print(state_of(pid))
    print(os.waitstatus_to_exitcode(os.waitpid(pid, 0)[1]))
    print(orphan_ppid())
    print(pid1_state_after_sigterm(SLEEPER))
    killed = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(5)"])
    time.sleep(0.2)
    killed.send_signal(signal.SIGKILL)
    print(128 - killed.wait())

    # --- Часть 2: сколько потомков останутся зомби -----------------------
    #
    # ПОЧЕМУ ЗДЕСЬ ДОЛЯ, А НЕ ВРЕМЯ. Первый черновик печатал кратность «во
    # сколько раз дольше останавливается процесс без обработчика» — и она
    # прыгала между прогонами от двухсот до шестисот, потому что делится
    # льготный срок на пять миллисекунд. Число, у которого в знаменателе
    # планировщик, не годится в задачу с одним верным ответом. Доля зомби
    # такой зависимости не имеет: незабранным остаётся каждый потомок, и это
    # ровно то, что урок и утверждает.
    print()
    children = []
    for _ in range(CHILDREN):
        kid = os.fork()
        if kid == 0:
            os._exit(0)
        children.append(kid)
    time.sleep(0.5)
    zombies = sum(1 for kid in children if state_of(kid) == "Z")
    print(f"children created                     {len(children)}")
    print(f"of them in state Z                   {zombies}")
    print(f"share of children in state Z, %      {100 * zombies / len(children):.1f}")
    for kid in children:
        os.waitpid(kid, 0)
    print(f"left after wait                      {sum(1 for kid in children if state_of(kid) == 'Z')}")

    # Цена остановки — отдельно и как наблюдение, а не как задача: делитель
    # здесь миллисекундный, и кратность между прогонами гуляет.
    print()
    with_handler = stop_time(HANDLED)
    without_handler = stop_time(SLEEPER)
    print(f"stop time, SIGTERM handled           {with_handler * 1000:.0f} ms")
    print(f"stop time, no handler                {without_handler * 1000:.0f} ms")
    print(f"stop time, no handler, s             {without_handler:.1f}")


if __name__ == "__main__":
    main()