"""TIME_WAIT и эфемерные порты: почему клиент упирается раньше сервера.

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

Проверить это словами нельзя: нужно увидеть, на чьей стороне возникает
`TIME_WAIT`, сколько он живёт и во что упирается счёт.

ЧТО ЗДЕСЬ ИЗМЕРЯЕТСЯ:
  1. На чьей стороне оказывается TIME_WAIT — того, кто закрыл первым.
  2. Сколько он длится и подчиняется ли настройке.
  3. Во что упирается клиент: диапазон портов исчерпывается точно.
  4. Потолок новых соединений в секунду — арифметика над двумя замерами.
  5. Что меняет переиспользование одного соединения.

ПОЧЕМУ СЕРВЕР СВОЙ И НА LOOPBACK. Предмет замера — состояние сокета и номера
портов, а не сеть. На loopback оба конца видны одновременно, и по выводу `ss`
можно сказать, на какой стороне возник `TIME_WAIT`: у той стороны его локальный
порт — порт сервера, у другой он стоит в колонке собеседника.

ЗАПУСК: sudo python3 bench/timewait/ports.py
Вывод: runs/ports.txt

Прогон идёт около полутора минут: блок 2 честно дожидается, пока состояние
исчезнет само. Блоку 3 нужны права записи в /proc/sys/net/ipv4/ip_local_port_range;
без них он печатает `not permitted`, а исходный диапазон возвращается в finally.
"""

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

PORT_RANGE = "/proc/sys/net/ipv4/ip_local_port_range"


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


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


def time_wait_sides(port: int) -> tuple[int, int]:
    """Сколько сокетов в TIME_WAIT со стороны сервера и со стороны клиента.

    У сокета сервера локальный адрес кончается на порт сервера, у сокета
    клиента этим портом кончается адрес собеседника. Так одна команда
    отвечает на вопрос «кто закрыл первым».
    """
    out = subprocess.run(
        ["ss", "-tan", "state", "time-wait"], capture_output=True, text=True
    ).stdout
    server_side = client_side = 0
    for line in out.split("\n")[1:]:
        fields = line.split()
        if len(fields) < 4:
            continue
        if fields[2].endswith(f":{port}"):
            server_side += 1
        if fields[3].endswith(f":{port}"):
            client_side += 1
    return server_side, client_side


def in_time_wait(local_port: int) -> bool:
    """Стоит ли в TIME_WAIT сокет с этим локальным портом."""
    out = subprocess.run(
        ["ss", "-tan", "state", "time-wait"], capture_output=True, text=True
    ).stdout
    for line in out.split("\n")[1:]:
        fields = line.split()
        if len(fields) >= 3 and fields[2].endswith(f":{local_port}"):
            return True
    return False


class Echo:
    """Сервер, который отвечает и закрывает соединение первым или вторым."""

    def __init__(self, closes_first: bool, keep_alive: bool = False) -> None:
        self.closes_first = closes_first
        self.keep_alive = keep_alive
        self.sock = socket.socket()
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(("127.0.0.1", 0))
        self.sock.listen(512)
        self.port = self.sock.getsockname()[1]
        threading.Thread(target=self._loop, daemon=True).start()

    def _loop(self) -> None:
        while True:
            try:
                conn, _ = self.sock.accept()
            except OSError:
                return
            threading.Thread(target=self._serve, args=(conn,), daemon=True).start()

    def _serve(self, conn: socket.socket) -> None:
        try:
            if self.keep_alive:
                # Соединение живёт, пока клиент шлёт запросы: это и есть
                # предмет пятого блока.
                while True:
                    data = conn.recv(16)
                    if not data:
                        return
                    conn.sendall(b"ok")
            conn.recv(16)
            conn.sendall(b"ok")
            if not self.closes_first:
                # Ждём, пока закроет клиент: тогда первым будет он.
                conn.recv(16)
        except OSError:
            pass
        finally:
            conn.close()

    def close(self) -> None:
        self.sock.close()


def exchange(port: int, wait_for_close: bool) -> int:
    """Одно соединение, один обмен. Возвращает занятый локальный порт."""
    sock = socket.create_connection(("127.0.0.1", port), timeout=5)
    local = sock.getsockname()[1]
    sock.sendall(b"x")
    sock.recv(16)
    if wait_for_close:
        # Сервер закрывает первым: дожидаемся его FIN, потом закрываемся сами.
        try:
            sock.recv(16)
        except OSError:
            pass
    sock.close()
    return local


# ------------------------------------------------------------------ 1
def block1() -> int:
    show("1. TIME_WAIT BELONGS TO WHOEVER CLOSED FIRST")

    for closes_first, label in ((False, "the client closes first"), (True, "the server closes first")):
        echo = Echo(closes_first)
        for _ in range(20):
            exchange(echo.port, wait_for_close=closes_first)
        time.sleep(0.4)
        server_side, client_side = time_wait_sides(echo.port)
        echo.close()
        row(label, f"server side {server_side}, client side {client_side}")

    print()
    print("  Twenty identical exchanges each time. The only difference is which")
    print("  end called close first, and the sockets land entirely on that end.")
    print("  So TIME_WAIT on a server is not a client problem to fix: it is a")
    print("  statement about who is ending the connections.")
    return 20


# ------------------------------------------------------------------ 2
def block2() -> float:
    show("2. HOW LONG IT LASTS")

    echo = Echo(closes_first=False)
    local = exchange(echo.port, wait_for_close=False)
    started = time.perf_counter()
    row("in TIME_WAIT right after close", in_time_wait(local))

    elapsed = -1.0
    while time.perf_counter() - started < 150:
        time.sleep(0.5)
        if not in_time_wait(local):
            elapsed = time.perf_counter() - started
            break
    echo.close()

    row("seconds until it disappeared", f"{elapsed:.1f}")
    row("tcp_fin_timeout on this machine", read_sysctl("tcp_fin_timeout"))
    row("tcp_max_tw_buckets on this machine", read_sysctl("tcp_max_tw_buckets"))
    row("tcp_tw_reuse on this machine", read_sysctl("tcp_tw_reuse"))
    print()
    print("  The first two numbers are close, and that closeness is a trap:")
    print("  tcp_fin_timeout is about waiting for a FIN in another state, not")
    print("  about this one. The other two settings are the ones people reach")
    print("  for, and neither shortens the wait: one caps how many such sockets")
    print("  may exist at once, the other lets an outgoing connection reuse a")
    print("  port early. No setting on this machine changes the duration.")
    return elapsed


def read_sysctl(name: str) -> str:
    try:
        with open(f"/proc/sys/net/ipv4/{name}") as handle:
            return handle.read().strip()
    except OSError:
        return "not readable"


# ------------------------------------------------------------------ 3
def block3() -> tuple[int, str]:
    show("3. WHAT THE CLIENT RUNS OUT OF IS PORT NUMBERS")

    original = None
    try:
        with open(PORT_RANGE) as handle:
            original = handle.read().strip()
    except OSError:
        pass

    echo = Echo(closes_first=False)
    made = 0
    stopped = "not permitted"
    if original is not None:
        try:
            with open(PORT_RANGE, "w") as handle:
                handle.write("50000 50049\n")
            row("port range for this block", "50000 50049, that is 50 numbers")
            for _ in range(400):
                try:
                    exchange(echo.port, wait_for_close=False)
                    made += 1
                except OSError as exc:
                    stopped = errno.errorcode.get(exc.errno, str(exc.errno))
                    break
            else:
                stopped = "nothing: 400 exchanges fitted"
        except OSError as exc:
            stopped = errno.errorcode.get(exc.errno, type(exc).__name__)
        finally:
            with open(PORT_RANGE, "w") as handle:
                handle.write(original + "\n")

    row("connections made", made)
    row("what stopped the loop", stopped)
    echo.close()
    print()
    print("  Fifty numbers, fifty connections, and the fifty-first has nowhere")
    print("  to come from: every previous port is still in TIME_WAIT. The error")
    print("  is not about memory or file descriptors - it is the kernel saying")
    print("  there is no local address left to use.")
    return made, stopped


# ------------------------------------------------------------------ 4
def block4(seconds: float) -> None:
    show("4. THE CEILING IS ARITHMETIC OVER THE TWO NUMBERS ABOVE")

    try:
        low, high = (int(x) for x in open(PORT_RANGE).read().split())
        span = high - low + 1
    except OSError:
        span = 0
    row("ip_local_port_range on this machine", f"{low} {high}")
    row("port numbers in it", span)
    row("seconds a port stays in TIME_WAIT", f"{seconds:.1f}")
    if seconds > 0:
        row("new connections per second, ceiling", f"{span / seconds:.1f}")
    print()
    print("  This is not a measurement but division of one measured number by")
    print("  another, and it holds only for connections to a single address and")
    print("  port: the kernel keeps the four-tuple, so a second destination")
    print("  gets its own set of ports. It is still the right first estimate,")
    print("  because a client usually hammers one service.")


# ------------------------------------------------------------------ 5
def block5() -> None:
    show("5. A REUSED CONNECTION SPENDS ONE PORT, WHATEVER THE LOAD")

    echo = Echo(closes_first=False, keep_alive=True)
    sock = socket.create_connection(("127.0.0.1", echo.port), timeout=5)
    local = sock.getsockname()[1]
    ports = {local}
    done = 0
    for _ in range(400):
        sock.sendall(b"x")
        sock.recv(16)
        done += 1
    sock.close()
    time.sleep(0.4)
    left = 1 if in_time_wait(local) else 0
    echo.close()

    row("requests sent", done)
    row("local ports spent", len(ports))
    row("TIME_WAIT sockets left behind", left)
    print()
    print("  Four hundred requests, one port, one socket in TIME_WAIT at the")
    print("  end. The previous block ran out of ports at fifty. Nothing about")
    print("  the kernel changed between them - only whether the connection was")
    print("  kept.")


def main() -> None:
    print(f"Python {sys.version.split()[0]} · Linux {os.uname().release} · loopback")
    print("this run takes about a minute and a half: block 2 waits it out")

    block1()
    seconds = block2()
    block3()
    block4(seconds)
    block5()


if __name__ == "__main__":
    main()