"""Проверка самого стенда: чем на деле был канал между ДЦ — блоки 11-14.

ОТКУДА ВЗЯЛСЯ ЭТОТ СКРИПТ. Разбор снаружи задал вопрос уровнем ниже всех
прежних. Прежние спрашивали «что мы заключили из замера»; этот спрашивает «что
замер на самом деле моделировал». Претензия была конкретная: задержка в
`link.py` ставилась на каждую порцию, которую вернул `recv()`, а TCP границ
сообщений не хранит — значит, измерялось не «Redis плюс фиксированная задержка
распространения», а «Redis плюс задержка, умноженная на то, на сколько порций
разбился поток».

ПОЧЕМУ НЕ `tc netem`, КАК ПРОСИЛ РАЗБОР. Ядро песочницы собрано без
`CONFIG_NET_SCH_NETEM`; `tc qdisc add dev veth0 root netem delay 20ms` отвечает
`Error: Specified qdisc kind is unknown`. Права тут ни при чём — дисциплины
очереди в ядре просто нет, и достать её неоткуда. Сверить с настоящим
эмулятором пакетов в этой среде нельзя, поэтому сверяемся с тем, что доступно:
измеряем сам прибор.

ЧТО ЗДЕСЬ ПРОВЕРЯЕТСЯ.

 11. Сколько стоит стенд, если задержку выкрутить в ноль. Это отделяет цену
     ретранслятора, петлевого TCP и планировщика потоков Python от цены
     расстояния. Здесь же выясняется происхождение слагаемого в 42-44 мс,
     которое блок 10 нашёл, но объяснить не смог.
 12. Сколько РАЗ задержка применяется к одной операции. Счётчик порций в
     ретрансляторе и есть ответ; контрольное окно без операций отделяет
     служебный трафик репликации от нашего.
 13. Сериализующий ретранслятор против конвейерного на одних и тех же
     операциях. Разница между ними — цена ошибки, на которую указал разбор.
 14. Калибровка: какую задержку канал даёт на простом «запрос — ответ».

ЗАПУСК: python3 bench/stretched-cache/relaycheck.py
Нужен redis-server в PATH (или в REDIS_SERVER_BIN).
"""

from __future__ import annotations

import os
import socket
import statistics
import subprocess
import sys
import tempfile
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from link import DelayedLink, SerializingLink  # noqa: E402

import redis  # noqa: E402

SERVER = os.environ.get("REDIS_SERVER_BIN", "redis-server")
DELAYS_MS = (5.0, 20.0, 40.0)
WRITES = 20
PAD = 44


def free_port() -> int:
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return port


def start_redis(port: int, dirname: str) -> subprocess.Popen:
    proc = subprocess.Popen([
        SERVER, "--port", str(port), "--dir", dirname,
        "--save", "", "--appendonly", "no",
        "--logfile", os.path.join(dirname, "redis.log"),
    ])
    deadline = time.time() + 15
    while time.time() < deadline:
        try:
            c = redis.Redis(port=port, socket_timeout=0.5)
            if c.ping():
                c.close()
                return proc
        except Exception:
            time.sleep(0.05)
    raise RuntimeError(f"redis на порту {port} не поднялся")


def wait_sync(client: redis.Redis, timeout: float = 20.0) -> None:
    deadline = time.time() + timeout
    while time.time() < deadline:
        if client.info("replication").get("master_link_status") == "up":
            return
        time.sleep(0.05)
    raise RuntimeError("реплика не синхронизировалась")


def median_ms(f, n: int) -> float:
    samples = []
    for i in range(n):
        t0 = time.perf_counter()
        f(i)
        samples.append((time.perf_counter() - t0) * 1000.0)
    return statistics.median(samples)


def row(label: str, value, pad: int = PAD) -> None:
    print(f"  {label:<{pad}}{value}")


def note(text: str) -> None:
    print()
    for line in text.strip("\n").splitlines():
        print("  " + line.strip() if line.strip() else "")
    print()


def head(number: int, title: str) -> None:
    print()
    print(f"{number}. {title}")
    print("-" * 66)


class Pair:
    """Мастер и реплика; реплику можно переподключать через разные каналы."""

    def __init__(self) -> None:
        self.dir_a = tempfile.mkdtemp(prefix="rc-a-")
        self.dir_b = tempfile.mkdtemp(prefix="rc-b-")
        self.port_a, self.port_b = free_port(), free_port()
        self.proc_a = start_redis(self.port_a, self.dir_a)
        self.proc_b = start_redis(self.port_b, self.dir_b)
        self.a = redis.Redis(port=self.port_a, decode_responses=True)
        self.b = redis.Redis(port=self.port_b, decode_responses=True)
        self.link = None

    def attach(self, link_cls=None, delay_ms: float = 0.0, nodelay: bool = True):
        """Подключить реплику к мастеру напрямую или через канал."""
        self.detach()
        if link_cls is None:
            self.b.replicaof("127.0.0.1", self.port_a)
        else:
            self.link = link_cls(self.port_a, delay_ms=delay_ms, nodelay=nodelay)
            self.b.replicaof("127.0.0.1", self.link.port)
        wait_sync(self.b)
        time.sleep(0.6)
        # Прогрев: первая запись после синхронизации несёт лишнее.
        self.a.set("warm", "1")
        self.a.wait(1, 5000)
        return self.link

    def detach(self) -> None:
        try:
            self.b.replicaof("no", "one")
        except Exception:
            pass
        if self.link is not None:
            self.link.cut()
            self.link = None
        time.sleep(0.2)

    def plain_and_wait(self, tag: str) -> tuple[float, float]:
        plain = median_ms(lambda i: self.a.set(f"{tag}:p:{i}", i), WRITES)

        def with_wait(i: int) -> None:
            self.a.set(f"{tag}:w:{i}", i)
            self.a.wait(1, 5000)

        return plain, median_ms(with_wait, WRITES)

    def close(self) -> None:
        self.detach()
        for proc in (self.proc_a, self.proc_b):
            proc.terminate()
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()


def block11(pair: Pair) -> None:
    head(11, "ЧТО СТОИТ САМ СТЕНД, ЕСЛИ ЗАДЕРЖКУ ВЫКРУТИТЬ В НОЛЬ")

    pair.attach(None)
    direct_plain, direct_wait = pair.plain_and_wait("d")
    row("без канала вовсе: обычная запись, мс", f"{direct_plain:.2f}")
    row("без канала вовсе: запись с WAIT 1, мс", f"{direct_wait:.2f}")
    print()

    # Ровно одна разница между этими двумя опытами — флаг TCP_NODELAY на двух
    # соединениях ретранслятора. Задержка в обоих нулевая.
    pair.attach(SerializingLink, 0.0, nodelay=False)
    old_plain, old_wait = pair.plain_and_wait("o")
    print("  канал 0 мс, БЕЗ TCP_NODELAY (стенд, на котором снят блок 10):")
    row("    обычная запись, мс", f"{old_plain:.2f}")
    row("    запись с WAIT 1, мс", f"{old_wait:.2f}")
    print()

    pair.attach(SerializingLink, 0.0, nodelay=True)
    new_plain, new_wait = pair.plain_and_wait("n")
    print("  канал 0 мс, С TCP_NODELAY (исправленный стенд):")
    row("    обычная запись, мс", f"{new_plain:.2f}")
    row("    запись с WAIT 1, мс", f"{new_wait:.2f}")
    print()

    row("цена прежнего стенда на одной записи, мс", f"{old_wait - new_wait:+.2f}")

    note(
        f"""
        Вот и ответ на вопрос, который блок 10 оставил открытым, — и ответ
        неприятный. Там нашлось слагаемое в 42-44 мс сверх круга, одинаковое
        на всех трёх расстояниях, и было честно сказано, что замер его
        происхождения не устанавливает. Устанавливает этот: слагаемое целиком
        принадлежит ИЗМЕРИТЕЛЬНОМУ ПРИБОРУ.

        Три строки читаются подряд. Без канала вовсе `SET` + `WAIT 1` стоит
        {direct_wait:.2f} мс. Через канал с НУЛЕВОЙ задержкой, но без
        TCP_NODELAY — {old_wait:.2f} мс. Тот же канал, та же нулевая задержка,
        флаг включён — {new_wait:.2f} мс. Расстояния в опыте нет ни в одной из
        трёх строк.

        Механизм известный. Redis ставит TCP_NODELAY на своих соединениях сам;
        ретранслятор создаёт ДВА НОВЫХ соединения, и на них флага не было.
        Дальше работает пара «алгоритм Нейгла на отправителе — отложенное
        подтверждение на получателе»: мелкая порция придерживается до
        подтверждения предыдущей, а подтверждение придерживается до сорока
        миллисекунд. Отсюда и ровно сорок с небольшим, и их независимость от
        расстояния, и то, что они появлялись только там, где по каналу шло
        несколько мелких сообщений подряд.

        ЧТО ЭТО ЗНАЧИТ ДЛЯ СТАТЬИ. Числа блока 10 (51,99 / 84,05 / 124,01 мс)
        измеряли Redis плюс расстояние плюс залипание в сорок миллисекунд на
        собственном стенде. Вывод «с расстоянием цена растёт на один круг, а
        не на два» устоял — он держится на разностях, а слагаемое было
        постоянным, — но сами числа и «слагаемое неизвестной природы» надо
        заменить. Ниже они пересняты.
        """
    )


def block12(pair: Pair) -> None:
    head(12, "СКОЛЬКО РАЗ ЗАДЕРЖКА ПРИМЕНЯЕТСЯ К ОДНОЙ ОПЕРАЦИИ")

    delay = 20.0
    for name, cls in (("сериализующий", SerializingLink), ("конвейерный", DelayedLink)):
        link = pair.attach(cls, delay)

        # Контрольное окно: репликация переговаривается и без нас (REPLCONF ACK
        # раз в секунду), и без контроля наш трафик от фонового не отличить.
        link.reset_stats()
        time.sleep(1.0)
        idle = (link.stats.chunks_down, link.stats.chunks_up)

        link.reset_stats()
        t0 = time.perf_counter()
        for i in range(WRITES):
            pair.a.set(f"c:{name}:{i}", i)
            pair.a.wait(1, 5000)
        elapsed = (time.perf_counter() - t0) * 1000.0
        busy = (link.stats.chunks_down, link.stats.chunks_up)

        print(f"  канал {name}, {delay:.0f} мс в одну сторону, {WRITES} раз SET+WAIT:")
        row("    порций мастер → реплика вхолостую", idle[0])
        row("    порций реплика → мастер вхолостую", idle[1])
        row("    порций мастер → реплика с нагрузкой", busy[0])
        row("    порций реплика → мастер с нагрузкой", busy[1])
        total = busy[0] + busy[1] - idle[0] - idle[1]
        row("    порций сверх фона, всего", total)
        row("    то же на одну операцию", f"{total / WRITES:.2f}")
        row("    задержка применена, мс на операцию", f"{total / WRITES * delay:.1f}")
        row("    измерено на операцию, мс", f"{elapsed / WRITES:.2f}")
        print()

    note(
        """
        Вот в чём была претензия разбора, выраженная числом. Задержка ставится
        не на операцию, а на ПОРЦИЮ, и порций на одну операцию три: сама
        запись, запрос подтверждения и подтверждение. Число это не
        гарантировано: TCP границ сообщений не хранит, и соседние сообщения
        могут слиться в одну порцию или разойтись по двум.

        Три порции на операцию — но на критическом пути их две, а не три:
        запись и запрос подтверждения идут в одну сторону и уходят вместе,
        подтверждение возвращается. Отсюда и круг у конвейерного канала.

        Разница между каналами ровно в том, складываются ли задержки. У
        сериализующего поток останавливается на время каждого полёта, поэтому
        запись и запрос подтверждения, ушедшие бы вместе, идут по очереди — и
        операция стоит на один перелёт дороже. Насколько именно, печатает
        следующий блок.
        """
    )


def block13(pair: Pair) -> None:
    head(13, "СЕРИАЛИЗУЮЩИЙ КАНАЛ ПРОТИВ КОНВЕЙЕРНОГО НА ТРЁХ РАССТОЯНИЯХ")

    print("  в одну сторону    круг   сериализующий   конвейерный   разница")
    rows = []
    for one_way in DELAYS_MS:
        pair.attach(SerializingLink, one_way)
        _, ser = pair.plain_and_wait(f"S{one_way:.0f}")
        pair.attach(DelayedLink, one_way)
        _, pipe = pair.plain_and_wait(f"P{one_way:.0f}")
        rt = 2 * one_way
        rows.append((one_way, ser, pipe))
        print(f"  {one_way:>10.0f} мс {rt:>7.0f} мс {ser:>13.2f} мс "
              f"{pipe:>10.2f} мс {ser - pipe:>+8.2f}")

    print()
    print("  сверх круга (цена минус круг):")
    for one_way, ser, pipe in rows:
        rt = 2 * one_way
        row(f"    при {one_way:>4.0f} мс в одну сторону",
            f"сериализующий {ser - rt:>7.2f} · конвейерный {pipe - rt:>7.2f}")

    note(
        """
        Это и есть цена ошибки, на которую указал разбор, — в миллисекундах, а
        не в рассуждении. Столбец разницы и есть ответ: сериализующий канал
        стоит ровно на ОДИН перелёт в одну сторону дороже на каждом
        расстоянии. Не на случайную величину и не на долю процента — на ту
        самую задержку, которую он применяет к порции, потому что ровно одна
        порция из трёх из-за него идёт по очереди вместо того, чтобы уйти
        вместе с соседней.

        Пока залипание в сорок миллисекунд из блока 11 сидело в обоих
        каналах, этой разницы не было видно вовсе: она тонула в артефакте,
        который был вчетверо больше её самой. Два дефекта стенда маскировали
        друг друга — и это, пожалуй, главный урок всего разбора.

        Числа надо публиковать по конвейерному каналу: у него цена `WAIT` —
        это круг плюс около полутора миллисекунд, то есть ровно то, чем она и
        должна быть по устройству команды.

        Чего этот блок НЕ доказывает: что конвейерный ретранслятор равен
        `netem`. Задержка здесь по-прежнему применяется к порциям потока, а не
        к пакетам; нет ни очередей, ни потерь, ни джиттера, ни ограничения
        полосы. На нагрузке, где порций много, расхождение с настоящей сетью
        вырастет, и проверить его в этой среде нечем.
        """
    )


def block14(pair: Pair) -> None:
    head(14, "КАЛИБРОВКА: КАКУЮ ЗАДЕРЖКУ КАНАЛ ДАЁТ НА «ЗАПРОС — ОТВЕТ»")

    direct = median_ms(lambda i: pair.a.get("warm"), 40)
    row("GET напрямую, мс", f"{direct:.2f}")
    print()
    print("  через канал, GET мимо ретранслятора к тому же мастеру:")
    print("  в одну сторону    круг   сериализующий   конвейерный")
    for one_way in DELAYS_MS:
        out = []
        for cls in (SerializingLink, DelayedLink):
            link = cls(pair.port_a, delay_ms=one_way)
            try:
                c = redis.Redis(port=link.port, decode_responses=True, socket_timeout=10)
                c.get("warm")  # первое обращение несёт установку соединения
                out.append(median_ms(lambda i: c.get("warm"), 20))
                c.close()
            finally:
                link.cut()
        print(f"  {one_way:>10.0f} мс {2 * one_way:>7.0f} мс "
              f"{out[0]:>13.2f} мс {out[1]:>10.2f} мс")

    note(
        """
        На простом «запрос — ответ» оба канала дают одно и то же, и дают ровно
        круг плюс около миллисекунды: одна порция туда, одна обратно,
        складывать нечего. Это граница применимости стенда, названная замером,
        а не оговоркой.

        Читать её так. Там, где за операцию по каналу проходит одна порция в
        каждую сторону, ретранслятор — хорошее приближение расстояния, и числа
        блоков 7 и 1 можно брать как есть. Там, где по каналу идёт несколько
        сообщений подряд — репликация, конвейер команд, `WAIT`, — прибор
        начинает добавлять своё, и добавлял: сорок миллисекунд залипания и
        лишний перелёт на сериализации. Оба слагаемых теперь измерены и
        убраны, но третьего рода погрешность — отсутствие очередей, потерь,
        джиттера и ограничения полосы — остаётся, и снять её в этой среде
        нечем.

        Отсюда правило для чтения всей статьи: качественные выводы (знак,
        кратность, порядок событий, какая половина кластера ответила) держатся;
        абсолютные миллисекунды на потоковых операциях — приближение, чья
        погрешность известна по знаку, но не по величине.
        """
    )


def main() -> None:
    pair = Pair()
    try:
        version = pair.a.info("server")["redis_version"]
        print(f"Redis {version} · один хост, два процесса")
        print("проверяется не Redis, а измерительный стенд: ретранслятор,")
        print("который во всех остальных замерах изображает канал между ДЦ")
        block11(pair)
        block12(pair)
        block13(pair)
        block14(pair)
    finally:
        pair.close()


if __name__ == "__main__":
    main()