"""Что стоит между `write` и диском: page cache, `fsync` и цена долговечности.

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

ЧТО ЗДЕСЬ ИЗМЕРЯЕТСЯ, А ЧТО НАБЛЮДАЕТСЯ. Времена — измеряются и от машины
зависят сильно: на этом диске числа одни, на вашем другие. Наблюдение о том,
что данные переживают убийство процесса, от машины не зависит вовсе: это
свойство того, кому принадлежит кеш.

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

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

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

RECORD = b"x" * 4096
ROUNDS = 200
WORK = os.path.join(tempfile.gettempdir(), "de_durability")


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


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


def timed_writes(path: str, mode: str, rounds: int = ROUNDS) -> float:
    """Среднее время одной записи в миллисекундах при заданной строгости."""
    fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC)
    samples = []
    try:
        for _ in range(rounds):
            started = time.perf_counter()
            os.write(fd, RECORD)
            if mode == "fsync":
                os.fsync(fd)
            elif mode == "fdatasync":
                os.fdatasync(fd)
            samples.append((time.perf_counter() - started) * 1000)
    finally:
        os.close(fd)
    return statistics.median(samples)


# ------------------------------------------------------------------ 1
def block1() -> None:
    show("1. THE PRICE OF DURABILITY, PER WRITE")

    os.makedirs(WORK, exist_ok=True)
    path = os.path.join(WORK, "records.bin")

    # Три режима меряются ЧЕРЕДУЯСЬ и по пять кругов: иначе первый режим
    # получает холодный файл, а последний — прогретый, и разница между
    # fsync и fdatasync тонет в этом перекосе.
    rounds = {"none": [], "fdatasync": [], "fsync": []}
    for _ in range(5):
        for mode in rounds:
            rounds[mode].append(timed_writes(path, mode))

    plain = statistics.median(rounds["none"])
    fdata = statistics.median(rounds["fdatasync"])
    full = statistics.median(rounds["fsync"])

    row("write only (4 KiB), median", f"{plain * 1000:.1f} us")
    row("write + fdatasync, median", f"{fdata * 1000:.1f} us")
    row("write + fsync, median", f"{full * 1000:.1f} us")
    row("fsync over plain write", f"{full / plain:.0f}x")

    # САМОПРОВЕРКА. Разницу между fsync и fdatasync объявлять можно только
    # если она больше разброса между кругами. На этой машине она обычно
    # меньше — и тогда честный ответ «не различимы здесь», а не число.
    spread = max(
        max(rounds["fsync"]) - min(rounds["fsync"]),
        max(rounds["fdatasync"]) - min(rounds["fdatasync"]),
    )
    gap = abs(full - fdata)
    row("gap between fsync and fdatasync", f"{gap * 1000:.1f} us")
    row("run-to-run spread of the same mode", f"{spread * 1000:.1f} us")
    row("difference resolvable on this machine", "yes" if gap > spread else "no")
    print()
    print("  A plain write does not reach the disk. It reaches the page cache,")
    print("  which is the kernel's memory, and returns.")


# ------------------------------------------------------------------ 2
def block2() -> None:
    show("2. WHOSE MEMORY THE PAGE CACHE IS")

    # Потомок пишет БЕЗ сброса и умирает от SIGKILL — самой грубой смертью,
    # какая бывает. Если бы кеш принадлежал процессу, данные исчезли бы
    # вместе с ним.
    path = os.path.join(WORK, "killed.bin")
    program = (
        "import os, sys, time\n"
        "fd = os.open(sys.argv[1], os.O_CREAT | os.O_WRONLY | os.O_TRUNC)\n"
        "os.write(fd, b'survived the kill')\n"
        "time.sleep(30)\n"
    )
    proc = subprocess.Popen([sys.executable, "-c", program, path])
    time.sleep(0.7)
    proc.kill()
    proc.wait()

    with open(path, "rb") as fh:
        content = fh.read()
    row("child wrote without fsync, then SIGKILL", f"exit {proc.returncode}")
    row("file content after the kill", content.decode())
    print()
    print("  A process crash loses nothing that reached the page cache: the")
    print("  cache belongs to the kernel. What loses it is the machine going")
    print("  down before the kernel writes the pages out.")


# ------------------------------------------------------------------ 3
def block3() -> None:
    show("3. ONE FSYNC FOR MANY RECORDS, OR ONE EACH")

    path = os.path.join(WORK, "batch.bin")
    batch = 100

    fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC)
    started = time.perf_counter()
    for _ in range(batch):
        os.write(fd, RECORD)
        os.fsync(fd)
    each = time.perf_counter() - started
    os.close(fd)

    fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC)
    started = time.perf_counter()
    for _ in range(batch):
        os.write(fd, RECORD)
    os.fsync(fd)
    once = time.perf_counter() - started
    os.close(fd)

    row(f"{batch} records, fsync after each", f"{each * 1000:.1f} ms")
    row(f"{batch} records, one fsync at the end", f"{once * 1000:.1f} ms")
    row("ratio", f"{each / once:.0f}x")
    print()
    print("  Same bytes, same disk, same durability at the end of the batch.")
    print("  What differs is how many times the promise was demanded.")


# ------------------------------------------------------------------ 4
def block4() -> None:
    show("4. WHAT FSYNC IS FOR: THE FILE, NOT THE DIRECTORY")

    # Создание файла — изменение КАТАЛОГА. fsync файла о нём не говорит
    # ничего; чтобы имя пережило падение машины, синхронизируют каталог.
    path = os.path.join(WORK, "fresh.bin")
    if os.path.exists(path):
        os.unlink(path)
    fd = os.open(path, os.O_CREAT | os.O_WRONLY)
    os.write(fd, b"data")
    os.fsync(fd)
    os.close(fd)

    dir_fd = os.open(WORK, os.O_RDONLY)
    started = time.perf_counter()
    os.fsync(dir_fd)
    dir_cost = (time.perf_counter() - started) * 1000
    os.close(dir_fd)

    row("file synced", "yes")
    row("directory sync cost", f"{dir_cost:.2f} ms")
    print()
    print("  Syncing the file makes its CONTENT durable. The fact that the")
    print("  file exists under that name lives in the directory, and that is")
    print("  a separate object with a separate sync.")


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


if __name__ == "__main__":
    main()