"""Жизнь и смерть процесса: зомби, сирота и особый статус PID 1.

ЗАЧЕМ ЭТОТ ФАЙЛ. Три вещи из этого урока обычно пересказывают по памяти, и все
три проверяются на одной машине за секунды: что делает ядро с потомком, статус
которого никто не забрал; кому достаётся потомок, чей родитель умер; и почему
`SIGTERM`, посланный первому процессу контейнера, часто не делает ничего.

Последнее — не особенность Docker и не ошибка образа, а правило ядра, записанное
в `pid_namespaces(7)`: первому процессу пространства имён доставляются только те
сигналы, для которых он установил обработчик. Обработчик поставлен — сигнал
работает; не поставлен — сигнал не делает ничего. Здесь это показано обоими
случаями подряд.

ПОЧЕМУ ПОДПИСИ ПО-АНГЛИЙСКИ. Урок существует в двух языках, и запись прогона
цитируется в обоих дословно. Русские подписи означали бы либо два прогона с
расходящимися числами, либо кириллицу в английском материале, которой там быть
не должно. Английские подписи цитируются как есть обеими версиями — так же
устроены записи Go-линии.

ЧТО ЗДЕСЬ ИЗМЕРЯЕТСЯ, А ЧТО ПРОСТО ПЕЧАТАЕТСЯ. Времени тут нет вовсе: вопрос не
«сколько стоит», а «что происходит». Поэтому печатаются состояния из `/proc`,
коды выхода и идентификаторы родителей — то, что можно проверить, не веря на
слово ни автору, ни пересказу.

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

import os
import signal
import subprocess
import sys
import time


def show(title: str) -> None:
    print()
    print(title)
    print("-" * len(title))


def row(label: str, value: object) -> None:
    print(f"  {label:<42} {value}")


def proc_state(pid: int) -> str:
    """Однобуквенное состояние процесса из /proc/<pid>/stat.

    Читается третье поле, но имя команды в скобках может содержать пробелы,
    поэтому строка режется по последней закрывающей скобке, а не по пробелам.
    """
    try:
        with open(f"/proc/{pid}/stat", encoding="utf-8") as fh:
            data = fh.read()
    except FileNotFoundError:
        return "no such process"
    tail = data[data.rfind(")") + 2 :]
    return tail.split()[0]


STATE_WORDS = {
    "R": "R (running)",
    "S": "S (sleeping)",
    "D": "D (uninterruptible sleep)",
    "Z": "Z (zombie)",
    "T": "T (stopped)",
}


def human_state(pid: int) -> str:
    state = proc_state(pid)
    return STATE_WORDS.get(state, state)


def target_child(pid: int) -> int | None:
    """PID потомка процесса — того, кто внутри пространства имён стал первым.

    Список детей ядро отдаёт файлом; из него берётся первый, потому что
    `unshare --fork` порождает ровно одного.
    """
    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


# ------------------------------------------------------------------ 1
def block1() -> None:
    show("1. ZOMBIE: the child is gone, the record is not")

    pid = os.fork()
    if pid == 0:
        os._exit(7)

    time.sleep(0.2)
    row("child finished, status not collected", human_state(pid))
    row("why the kernel keeps the record", "so the parent can read the exit code")

    done, status = os.waitpid(pid, 0)
    row("state after wait", human_state(pid))
    row("exit code the record was held for", os.waitstatus_to_exitcode(status))
    row("conclusion", "a zombie is an uncollected status, not leaked memory")


# ------------------------------------------------------------------ 2
def block2() -> None:
    show("2. ORPHAN: the parent died first")

    # Внук печатает своего родителя ДО и ПОСЛЕ смерти отца: только так видно
    # смену родителя, а не её последствия.
    code = (
        "import os, time, sys\n"
        "print('  grandchild: parent right after start   ', os.getppid(), flush=True)\n"
        "time.sleep(0.6)\n"
        "print('  grandchild: parent after father exits  ', os.getppid(), flush=True)\n"
    )
    parent = (
        "import subprocess, sys, time\n"
        f"subprocess.Popen([sys.executable, '-c', {code!r}])\n"
        "time.sleep(0.2)\n"
    )
    subprocess.run([sys.executable, "-c", parent], check=True)
    time.sleep(0.8)
    row("who the new parent is", "nearest subreaper, or PID 1")
    row("why it matters", "the orphan's status goes to it, not to the lost parent")


# ------------------------------------------------------------------ 3
def block3() -> None:
    show("3. PID 1: one SIGTERM, two outcomes")

    # Программа-подопытная: спит, а по флагу ставит обработчик SIGTERM.
    prog = (
        "import os, signal, sys, time\n"
        "handle = sys.argv[1] == 'handler'\n"
        "if handle:\n"
        "    signal.signal(signal.SIGTERM, lambda *_: sys.exit(143))\n"
        "print(os.getpid(), flush=True)\n"
        "time.sleep(5)\n"
    )

    for mode, label in (("nohandler", "no handler"), ("handler", "with handler")):
        proc = subprocess.Popen(
            ["unshare", "--fork", "--pid", "--mount-proc", sys.executable, "-c", prog, mode],
            stdout=subprocess.PIPE,
            text=True,
        )
        inner_pid = proc.stdout.readline().strip()
        time.sleep(0.3)

        # СИГНАЛ ШЛЁТСЯ НЕ unshare, А ЕГО ПОТОМКУ. Внутри пространства имён у
        # него PID 1, снаружи — обычный номер, и именно снаружи мы его и шлём:
        # так делает и оркестратор. Первый черновик слал сигнал самому
        # `unshare` и получал «жив» в обоих случаях — то есть не проверял
        # ничего, а показывал поведение постороннего процесса.
        target = target_child(proc.pid)
        row(f"PID inside the namespace ({label})", inner_pid)
        if target is None:
            row(f"SIGTERM, {label}", "no child of unshare found")
            proc.kill()
            proc.wait()
            continue

        os.kill(target, signal.SIGTERM)
        deadline = time.time() + 1.5
        outcome = "ALIVE: the signal did nothing"
        while time.time() < deadline:
            state = proc_state(target)
            if state in ("no such process", "Z"):
                outcome = "exited"
                break
            time.sleep(0.05)
        row(f"SIGTERM from outside, {label} (PID {target})", outcome)
        proc.kill()
        proc.wait()

    row("the rule", "no default signal action is carried out for PID 1")
    row("what it means for a container", "an image with no handler ignores SIGTERM")


# ------------------------------------------------------------------ 4
def block4() -> None:
    show("4. EXIT CODES: 128 + signal number")

    cases = [
        ("exited on its own", [sys.executable, "-c", "raise SystemExit(3)"], None),
        ("SIGTERM, default disposition", [sys.executable, "-c", "import time; time.sleep(5)"], signal.SIGTERM),
        ("SIGKILL", [sys.executable, "-c", "import time; time.sleep(5)"], signal.SIGKILL),
    ]
    for label, cmd, sig in cases:
        proc = subprocess.Popen(cmd)
        if sig is not None:
            time.sleep(0.2)
            proc.send_signal(sig)
        code = proc.wait()
        note = ""
        if code < 0:
            note = f"(the shell reports {128 - code})"
        row(label, f"{code} {note}")

    row("why 137 in OOM reports", "128 + 9: killed by SIGKILL")
    row("why 143 during a rollout", "128 + 15: killed by SIGTERM")


def main() -> None:
    print(f"Python {sys.version.split()[0]} · Linux {os.uname().release}")
    block1()
    block2()
    block3()
    block4()


if __name__ == "__main__":
    main()