CFDManager/docs/theory/2d_solver/bench/parity.py

#!/usr/bin/env python3
"""Сверка GPU-пути с эталонным CPU-путём.

Зачем. На настоящем сервере GPU-путь идёт через драйвер NVIDIA. В WSL такого драйвера
для Linux нет, и единственный доступный Vulkan — dzn (Dozen), транслирующий вызовы в
D3D12. Он честно сообщает о себе `conformanceVersion = 0.0.0.0`, то есть набор тестов
соответствия не проходил. Арифметика f32 в обоих API описана одним и тем же IEEE 754,
но точность трансцендентных функций (exp, log, sqrt — а они в энтропийном равновесии
на каждом узле) стандартами задаётся с допуском в несколько ULP, и допуски эти разные.
Плюс энтропийный стабилизатор сравнивает знаменатель с относительным порогом GREL=1e-8:
достаточно мелкого расхождения, чтобы решение «вырожден или нет» поменялось.

Поэтому пригодность dzn нельзя принимать на веру — её надо мерить. Скрипт гоняет два
коротких эталона на CPU (f64) и на GPU и сверяет числа из summary.json.

Прогоны подобраны намеренно НЕ хаотические: вихрь Тейлора–Грина затухает гладко и имеет
точное решение, обтекание цилиндра при Re=20 стационарно. В таких течениях расхождение
f32 и f64 не нарастает, поэтому любое заметное различие означает проблему драйвера, а не
разрядности. На развитой дорожке (Re >= 100) сверять бессмысленно: там разойдётся любая
пара запусков с разной арифметикой.

Порог взят с запасом от факта: на нативном Vulkan разница Cd между CPU-f64 и GPU-f32
измерена и составила 0.007%.

  python3 parity.py                 # сверка
  python3 parity.py --tol 2e-3      # свой порог
  python3 parity.py --keep out/     # оставить сводки для разбора

Код возврата 1, если хоть одно поле вышло за порог или прогон развалился.
"""

import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile

HERE = os.path.dirname(os.path.abspath(__file__))
BIN = os.environ.get("KBC2D_BIN") or os.path.join(
    HERE, "..", "target", "release", "kbc2d" + (".exe" if os.name == "nt" else ""))

# Поля сводки: "fields" сверяются с порогом, "report" только печатаются.
#
#  * energy_end, enstrophy_end — интегралы поля, ловят расхождение самого счёта;
#  * cd — интегральная сила, накапливает ошибку по всей границе тела;
#  * gamma_mean и degenerate_frac — поведение энтропийного стабилизатора, самый чуткий
#    индикатор расхождений в exp/log;
#  * rho_mean — сохранение массы, обязано совпадать почти побитово.
#
# Пороги откалиброваны по замеру на НАТИВНОМ драйвере Vulkan (Intel Iris Xe, Windows):
# там же, где f64 и f32 сравниваются в отсутствие какой-либо трансляции, отклонения
# составили cd 1.7e-5, rho_mean 1.5e-6, gamma_mean 6.5e-4, energy_end 3.4e-3 (вихрь
# доведён до затухания). Пороги ниже взяты с запасом к этим числам, но заметно ниже
# того, что дал бы по-настоящему сломанный драйвер.
#
# Прогоны намеренно НЕ доводятся до глубокого затухания: когда энергия падает до 1e-7,
# f32 упирается в собственный шум, и сверять становится нечего — это свойство
# разрядности, а не драйвера.
CASES = [
    {
        "name": "Тейлор-Грин 128x128, 500 шагов",
        "args": ["--case", "taylor-green", "--nx", "128", "--ny", "128",
                 "--refine", "1", "--steps", "500", "--case-every", "100"],
        # taylor_green_error намеренно НЕ в списке сверяемых, только в отчётных: это
        # разность почти равных величин, и её относительное отклонение f32 от f64
        # достигает десятков процентов на любом, в том числе нативном, драйвере.
        # Замерено на нативном Vulkan: 0.0116 (f64) против 0.0218 (f32). Сверять по
        # ней бессмысленно, а видеть её полезно.
        "fields": {"energy_end": 1.0, "enstrophy_end": 1.0, "gamma_mean": 1.0},
        "report": ["taylor_green_error"],
    },
    {
        "name": "цилиндр Re=20, стационар, 8000 шагов",
        "args": ["--shape", "cylinder", "--size", "16", "--nx", "320", "--ny", "192",
                 "--re", "20", "--refine", "1", "--steps", "8000"],
        "fields": {"cd": 1.0, "rho_mean": 0.05, "gamma_mean": 2.0,
                   "degenerate_frac": 5.0},
        "report": ["strouhal"],
    },
]


def run(backend, args, summary_path):
    """Один прогон. Возвращает разобранную сводку либо строку с ошибкой."""
    cmd = [BIN] + args + ["--backend", backend, "--verbose", "quiet",
                          "--report-every", "0", "--summary", summary_path]
    proc = subprocess.run(cmd, capture_output=True, text=True, errors="replace")
    if proc.returncode != 0:
        tail = (proc.stderr or proc.stdout or "").strip().splitlines()
        return None, "\n".join(tail[-6:]) or f"код возврата {proc.returncode}"
    try:
        with open(summary_path, encoding="utf-8") as f:
            return json.load(f), None
    except Exception as e:
        return None, f"сводка не читается: {e}"


def deviation(ref, got):
    """Относительное отклонение; для околонулевых величин — абсолютное."""
    if abs(ref) > 1e-12:
        return abs(got - ref) / abs(ref)
    return abs(got - ref)


def main():
    ap = argparse.ArgumentParser(description="Сверка GPU-пути с CPU-эталоном")
    ap.add_argument("--tol", type=float, default=1e-3,
                    help="базовый относительный порог (по умолчанию 1e-3)")
    ap.add_argument("--keep", metavar="DIR", help="куда положить сводки прогонов")
    args = ap.parse_args()

    if not os.path.exists(BIN):
        print(f"не найден бинарь решателя: {BIN}")
        return 1

    workdir = args.keep or tempfile.mkdtemp(prefix="kbc2d-parity-")
    os.makedirs(workdir, exist_ok=True)
    failures = []
    speeds = []

    for case in CASES:
        print(f"\n=== {case['name']} ===")
        summaries = {}
        for backend in ("cpu", "gpu"):
            path = os.path.join(workdir, f"{backend}.json")
            data, err = run(backend, case["args"], path)
            if err:
                print(f"  {backend}: ПРОГОН НЕ УДАЛСЯ\n    {err}")
                failures.append(f"{case['name']}: {backend} не отработал")
                break
            if data.get("blew_up"):
                print(f"  {backend}: счёт РАЗВАЛИЛСЯ")
                failures.append(f"{case['name']}: {backend} развалился")
                break
            summaries[backend] = data
        if len(summaries) != 2:
            continue

        cpu, gpu = summaries["cpu"], summaries["gpu"]
        m_cpu, m_gpu = cpu.get("mlups", 0.0), gpu.get("mlups", 0.0)
        speeds.append((case["name"], m_cpu, m_gpu))
        ratio = f", ускорение x{m_gpu / m_cpu:.1f}" if m_cpu > 0 else ""
        print(f"  скорость: CPU {m_cpu:.1f} MLUPS, GPU {m_gpu:.1f} MLUPS{ratio}")

        print(f"  {'поле':<22}{'CPU (f64)':>16}{'GPU':>16}{'откл.':>12}  порог")
        for field in case.get("report", []):
            if field in cpu and field in gpu:
                ref, got = float(cpu[field]), float(gpu[field])
                print(f"  {field:<22}{ref:>16.6g}{got:>16.6g}{deviation(ref, got):>11.2e}"
                      f"  (только к сведению)")
        for field, mult in case["fields"].items():
            if field not in cpu or field not in gpu:
                print(f"  {field:<22}{'нет в сводке':>16}")
                continue
            ref, got = float(cpu[field]), float(gpu[field])
            dev = deviation(ref, got)
            tol = args.tol * mult
            mark = "" if dev <= tol else "   <-- ВЫШЕ ПОРОГА"
            print(f"  {field:<22}{ref:>16.6g}{got:>16.6g}{dev:>11.2e}  {tol:.1e}{mark}")
            if dev > tol:
                failures.append(f"{case['name']}: {field} отклонилось на {dev:.2e} "
                                f"при пороге {tol:.1e}")

    print()
    if speeds:
        print("Скорость по прогонам:")
        for name, m_cpu, m_gpu in speeds:
            print(f"  {name}: CPU {m_cpu:.1f} MLUPS, GPU {m_gpu:.1f} MLUPS")
        print()

    if failures:
        print("СВЕРКА НЕ ПРОЙДЕНА:")
        for f in failures:
            print(f"  * {f}")
        print("\nGPU-путь на этом драйвере доверия не заслуживает — кампанию на нём "
              "запускать нельзя.")
    else:
        print("Сверка пройдена: GPU считает то же, что CPU в двойной точности.")

    if not args.keep:
        shutil.rmtree(workdir, ignore_errors=True)
    return 1 if failures else 0


if __name__ == "__main__":
    sys.exit(main())