"""Очередь accept: где на самом деле стоит запрос, пока сервер занят.

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

ЧТО ЗДЕСЬ ИЗМЕРЯЕТСЯ, А ЧТО НАБЛЮДАЕТСЯ. Времени почти нет: считаются
соединения и читаются счётчики ядра через `ss`. Единственное время —
подтверждение того, что при переполнении клиент именно ЖДЁТ, а не получает
отказ.

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

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

import errno
import os
import socket
import subprocess
import sys
import time

HOST = "127.0.0.1"
SOMAXCONN = "/proc/sys/net/core/somaxconn"


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


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


def listening_queue(port: int) -> str:
    """Строка `ss` про слушающий сокет: Recv-Q — занято, Send-Q — вместимость."""
    done = subprocess.run(
        ["ss", "-lnt", f"sport = :{port}"], capture_output=True, text=True
    )
    lines = [ln for ln in done.stdout.splitlines() if str(port) in ln]
    if not lines:
        return "not visible in ss"
    parts = lines[0].split()
    return f"Recv-Q={parts[1]} Send-Q={parts[2]}"


def fill(port: int, limit: int = 40, timeout: float = 0.6) -> tuple[int, str]:
    """Подключается, пока подключается. Возвращает счёт и то, чем всё кончилось."""
    kept = []
    for _ in range(limit):
        client = socket.socket()
        client.settimeout(timeout)
        try:
            client.connect((HOST, port))
        except socket.timeout:
            client.close()
            return len(kept), "timeout: the client waits, nobody refused it"
        except OSError as exc:
            client.close()
            name = errno.errorcode.get(exc.errno, str(exc.errno))
            return len(kept), f"{name}: {exc.strerror}"
        kept.append(client)
    for client in kept:
        client.close()
    return len(kept), f"no refusal within {limit} attempts"


def server(backlog: int) -> tuple[socket.socket, int]:
    srv = socket.socket()
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind((HOST, 0))
    srv.listen(backlog)
    return srv, srv.getsockname()[1]


# ------------------------------------------------------------------ 1
def block1() -> None:
    show("1. THE SERVER NEVER CALLS ACCEPT, AND THE CLIENT CONNECTS ANYWAY")

    srv, port = server(backlog=5)
    client = socket.socket()
    client.settimeout(1.0)
    started = time.perf_counter()
    client.connect((HOST, port))
    elapsed = (time.perf_counter() - started) * 1000
    sent = client.send(b"GET / HTTP/1.0\r\n\r\n")

    row("server called accept", "no, not once")
    row("client's connect() returned", f"success in {elapsed:.2f} ms")
    row("bytes the client managed to send", sent)
    row("listening socket in ss", listening_queue(port))
    client.close()
    srv.close()
    print()
    print("  The handshake is done by the kernel, not by the application. From")
    print("  the client's side a queued connection looks exactly like a served")
    print("  one: connected, writable, silent.")


# ------------------------------------------------------------------ 2
def block2() -> None:
    show("2. HOW MANY FIT: BACKLOG 1 AGAINST BACKLOG 8")

    for backlog in (1, 8):
        srv, port = server(backlog=backlog)
        count, ending = fill(port)
        row(f"listen(backlog={backlog}): connections accepted by the kernel", count)
        row("  queue as ss sees it", listening_queue(port))
        row("  what stopped the loop", ending)
        srv.close()
    print()
    print("  Nobody accepted anything: the number is the depth of the queue the")
    print("  kernel keeps for a server that is not asking.")


# ------------------------------------------------------------------ 3
def block3() -> None:
    show("3. WHAT OVERFLOW LOOKS LIKE FROM THE CLIENT")

    srv, port = server(backlog=1)
    count, ending = fill(port)
    row("connections before the loop stopped", count)
    row("how it ended", ending)
    row("tcp_abort_on_overflow", open("/proc/sys/net/ipv4/tcp_abort_on_overflow").read().strip())
    srv.close()
    print()
    print("  With the default setting the kernel does not refuse an overflowing")
    print("  connection: it drops the packet and lets the client retry. So the")
    print("  client reports a timeout, and the server logs nothing at all.")


# ------------------------------------------------------------------ 4
def block4() -> None:
    show("4. THE BACKLOG IN YOUR CODE IS NOT THE QUEUE YOU GET")

    original = open(SOMAXCONN).read().strip()
    row("somaxconn on this machine", original)

    srv, port = server(backlog=100)
    row("listen(backlog=100) with that somaxconn", listening_queue(port))
    srv.close()

    try:
        with open(SOMAXCONN, "w", encoding="utf-8") as fh:
            fh.write("2")
        srv, port = server(backlog=100)
        row("same listen(backlog=100), somaxconn=2", listening_queue(port))
        count, _ = fill(port)
        row("connections the kernel took", count)
        srv.close()
    except OSError as exc:
        row("lowering somaxconn", f"not permitted: {exc}")
    finally:
        try:
            with open(SOMAXCONN, "w", encoding="utf-8") as fh:
                fh.write(original)
        except OSError:
            pass
    row("somaxconn restored", open(SOMAXCONN).read().strip())
    print()
    print("  The value passed to listen() is silently capped by somaxconn. A")
    print("  service tuned in code and never checked against the machine gets")
    print("  the machine's number, not its own.")


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


if __name__ == "__main__":
    main()