"""Таймаут против дедлайна: что происходит с цепочкой сервисов, когда клиент ушёл.

ЗАЧЕМ ЭТОТ ФАЙЛ. «Поставьте таймаут» — совет, который звучит на каждом разборе
инцидента и почти никогда не разбирается до механизма. А механизм здесь
неочевидный: таймаут ограничивает ОЖИДАНИЕ вызывающего, но ничего не говорит
исполнителю. Поэтому цепочка из трёх сервисов с таймаутом в секунду на каждом
шаге может работать три секунды, а работа продолжается и после того, как
клиент ушёл.

ЧТО ЗДЕСЬ ИЗМЕРЯЕТСЯ. Времена — на loopback, поэтому сетевой задержки в них
почти нет: всё, что видно, — это выдержки, которые скрипт задаёт сам. Главное
здесь не миллисекунды, а два счётчика: сколько работы выполнено ПОСЛЕ ухода
клиента без дедлайна и с ним.

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

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

import os
import socket
import sys
import threading
import time

HOST = "127.0.0.1"
WORK_MS = 400  # сколько «работает» каждый сервис
CLIENT_TIMEOUT = 0.5
HOP_TIMEOUT = 1.0  # таймаут, который каждый сервис ставит следующему


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


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


class Service(threading.Thread):
    """Сервис на loopback.

    Протокол одной строкой: `<оставшийся бюджет в мс>`. Если бюджета меньше,
    чем нужно на работу, сервис отвечает `deadline` немедленно — это и есть
    разница между таймаутом и дедлайном. Бюджет −1 означает «никто не сказал».
    """

    def __init__(self, work_ms: int, downstream_port: int | None = None) -> None:
        super().__init__(daemon=True)
        self.work_ms = work_ms
        self.downstream_port = downstream_port
        self.sock = socket.socket()
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind((HOST, 0))
        self.sock.listen(16)
        self.port = self.sock.getsockname()[1]
        self.completed = 0  # сколько раз работа доведена до конца
        self.refused = 0  # сколько раз отказано по дедлайну — обеими проверками
        self.refused_start = 0  # из них: отказ, не начиная работу
        self.called = 0  # сколько раз сервис вообще позвали
        self.stop = threading.Event()

    def run(self) -> None:
        while not self.stop.is_set():
            try:
                conn, _ = self.sock.accept()
            except OSError:
                return
            threading.Thread(target=self.handle, args=(conn,), daemon=True).start()

    def handle(self, conn: socket.socket) -> None:
        with conn:
            self.called += 1
            entered = time.perf_counter()
            data = conn.recv(64).decode().strip()
            budget = int(data) if data else -1

            def left() -> int:
                """Сколько бюджета осталось. −1 значит «бюджета никто не назвал»."""
                if budget < 0:
                    return -1
                return budget - int((time.perf_counter() - entered) * 1000)

            # ПЕРВАЯ ПРОВЕРКА ДЕДЛАЙНА: браться ли за работу вообще.
            if budget >= 0 and left() < self.work_ms:
                self.refused += 1
                self.refused_start += 1
                conn.sendall(b"deadline\n")
                return

            time.sleep(self.work_ms / 1000)
            self.completed += 1

            if self.downstream_port is not None:
                # ВТОРАЯ ПРОВЕРКА: звать ли следующего. Своей работой бюджет
                # уже потрачен, и решение принимается по остатку.
                inner = socket.socket()
                inner.settimeout(HOP_TIMEOUT)
                inner.connect((HOST, self.downstream_port))
                inner.sendall(f"{left()}\n".encode())
                answer = inner.recv(64)
                inner.close()
                if answer.strip() == b"deadline":
                    self.refused += 1
                    try:
                        conn.sendall(b"deadline\n")
                    except OSError:
                        pass
                    return

            try:
                conn.sendall(b"done\n")
            except OSError:
                # Клиент ушёл, ответ некуда девать — но работа уже сделана.
                pass


def call(port: int, budget_ms: int, timeout: float) -> str:
    client = socket.socket()
    client.settimeout(timeout)
    try:
        client.connect((HOST, port))
        client.sendall(f"{budget_ms}\n".encode())
        answer = client.recv(64).decode().strip()
        return answer or "empty"
    except socket.timeout:
        return "client timeout"
    finally:
        client.close()


# ------------------------------------------------------------------ 1
def block1() -> None:
    show("1. THE CLIENT GAVE UP; THE WORK DID NOT")

    tail = Service(work_ms=WORK_MS)
    middle = Service(work_ms=WORK_MS, downstream_port=tail.port)
    front = Service(work_ms=WORK_MS, downstream_port=middle.port)
    for svc in (tail, middle, front):
        svc.start()
    time.sleep(0.1)

    started = time.perf_counter()
    answer = call(front.port, budget_ms=-1, timeout=CLIENT_TIMEOUT)
    waited = (time.perf_counter() - started) * 1000
    row("client timeout, ms", int(CLIENT_TIMEOUT * 1000))
    row("services in the chain", 3)
    row("work each service does, ms", WORK_MS)
    row("what the client got", answer)
    row("how long the client waited, ms", f"{waited:.0f}")

    time.sleep(1.6)
    row("services that did the work anyway", tail.completed + middle.completed + front.completed)
    row("services that refused before starting", tail.refused_start + middle.refused_start + front.refused_start)
    row("services never called at all", sum(1 for s in (tail, middle, front) if s.called == 0))
    print()
    print("  A timeout bounds the caller's waiting. It says nothing to the")
    print("  callee, which keeps holding a connection, a worker and a database")
    print("  session for work whose result nobody will read.")


# ------------------------------------------------------------------ 2
def block2() -> None:
    show("2. THE SAME CHAIN WITH A DEADLINE PASSED ALONG")

    tail = Service(work_ms=WORK_MS)
    middle = Service(work_ms=WORK_MS, downstream_port=tail.port)
    front = Service(work_ms=WORK_MS, downstream_port=middle.port)
    for svc in (tail, middle, front):
        svc.start()
    time.sleep(0.1)

    started = time.perf_counter()
    answer = call(front.port, budget_ms=int(CLIENT_TIMEOUT * 1000), timeout=CLIENT_TIMEOUT)
    waited = (time.perf_counter() - started) * 1000
    row("budget the client announced, ms", int(CLIENT_TIMEOUT * 1000))
    row("what the client got", answer)
    row("how long the client waited, ms", f"{waited:.0f}")

    time.sleep(1.6)
    row("services that did the work anyway", tail.completed + middle.completed + front.completed)
    row("services that refused before starting", tail.refused_start + middle.refused_start + front.refused_start)
    row("services never called at all", sum(1 for s in (tail, middle, front) if s.called == 0))
    print()
    print("  Same services, same work, same 500 ms. The difference is that the")
    print("  budget travelled with the request, so the second service could see")
    print("  there was no point in starting.")


# ------------------------------------------------------------------ 3
def block3() -> None:
    show("3. WHY PER-HOP TIMEOUTS DO NOT ADD UP TO THE CLIENT'S")

    tail = Service(work_ms=WORK_MS)
    middle = Service(work_ms=WORK_MS, downstream_port=tail.port)
    front = Service(work_ms=WORK_MS, downstream_port=middle.port)
    for svc in (tail, middle, front):
        svc.start()
    time.sleep(0.1)

    started = time.perf_counter()
    answer = call(front.port, budget_ms=-1, timeout=10.0)
    waited = (time.perf_counter() - started) * 1000
    row("timeout each service sets for the next one, ms", int(HOP_TIMEOUT * 1000))
    row("services in the chain", 3)
    row("what the client got", answer)
    row("how long the whole chain took, ms", f"{waited:.0f}")
    row("longest any single hop waited, ms", f"under {int(HOP_TIMEOUT * 1000)}")
    print()
    print("  Every hop restarts its own clock. Three hops with a one-second")
    print("  timeout each promise the client one second and can take three.")
    print("  A deadline does not have this property: it is one point in time,")
    print("  and it does not restart.")


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


if __name__ == "__main__":
    main()