parallel-programming/common/benchmark.py

# -*- coding: utf-8 -*-
"""
common.benchmark — утилиты для замеров времени и построения таблиц/графиков.

Используется во всех лабораторных (1–7), чтобы отчёты были единообразными.

Пример:

    from common.benchmark import time_call, make_table, save_speedup_plot

    elapsed, result = time_call(lambda: task["kernel"](chunk), repeats=1)
    rows = [("последовательно", T1), ("потоки, p=4", T2)]
    print(make_table(rows))
"""
from __future__ import annotations

import os
import statistics
import time
from typing import Callable, Dict, List, Sequence, Tuple


def time_call(fn: Callable, args: tuple = (), repeats: int = 3) -> Tuple[float, any]:
    """Замер времени выполнения fn(*args) с повторами.

    Возвращает (лучшее_время_сек, результат). Лучшее (минимальное) время —
    стандартная практика: случайные помехи только удлиняют замер.
    """
    best = float("inf")
    result = None
    for _ in range(repeats):
        t0 = time.perf_counter()
        result = fn(*args)
        best = min(best, time.perf_counter() - t0)
    return best, result


def speedup(T1: float, Tp: float) -> float:
    """Ускорение S(p) = T(1) / T(p)."""
    if Tp <= 0:
        return 0.0
    return T1 / Tp


def efficiency(speedup_val: float, p: int) -> float:
    """Эффективность E(p) = S(p) / p (доля, а не проценты)."""
    if p <= 0:
        return 0.0
    return speedup_val / p


def amdahl(speedup_bound: float, p: int) -> float:
    """Теоретическое ускорение по закону Амдала.

    speedup_bound — доля последовательной части (0..1), т.е. доля кода,
    которая не распараллеливается.
    """
    s = max(1e-12, speedup_bound)
    return 1.0 / (s + (1.0 - s) / p)


def make_table(rows: Sequence[Tuple[str, float]], T1: float = None,
               title: str = "Замеры") -> str:
    """Текстовая таблица: подход | T, c | S(p) | E(p).

    rows — пары (название, время в секундах). T1 — время последовательного
    запуска (первая строка, если не задано отдельно).
    """
    if T1 is None:
        T1 = rows[0][1]
    lines = [
        f"{title}",
        f"{'подход':<32} {'T, c':>10} {'S(p)':>8} {'E(p)':>8}",
        "-" * 62,
    ]
    for name, t in rows:
        s = speedup(T1, t)
        p = _p_from_name(name)
        e = efficiency(s, p)
        lines.append(f"{name:<32} {t:>10.3f} {s:>8.2f} {e:>8.2f}")
    return "\n".join(lines)


def _p_from_name(name: str) -> int:
    """Пытается вытащить число воркеров из названия ('потоки, p=4' -> 4)."""
    if "p=" in name:
        try:
            return int(name.split("p=")[1].split()[0].strip("),"))
        except ValueError:
            pass
    return 1


def save_speedup_plot(results: Dict[str, List[Tuple[int, float]]], title: str,
                      out_path: str, show: bool = False):
    """График ускорения S(p) для нескольких подходов.

    results = {"потоки": [(1, 1.0), (2, 1.9), (4, 3.8)],
               "процессы": [(1, 1.0), (2, 1.95), (4, 3.7)]}

    Для каждого подхода S(p) считается относительно его же T(1) — так
    кривые корректно сравниваются между собой.
    """
    import matplotlib
    if not show:
        matplotlib.use("Agg")
    import matplotlib.pyplot as plt

    plt.figure(figsize=(8, 5))
    for name, pts in results.items():
        pts = sorted(pts)
        base = next((t for p, t in pts if p == 1), None)
        if base is None:
            continue
        xs = [p for p, _ in pts]
        ys = [base / t for _, t in pts]
        plt.plot(xs, ys, marker="o", label=name)
    plt.plot(xs, xs, "--", color="gray", label="идеал S(p)=p")
    plt.xlabel("число воркеров p")
    plt.ylabel("ускорение S(p) = T(1)/T(p)")
    plt.title(title)
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.tight_layout()
    os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True)
    plt.savefig(out_path, dpi=150)
    if show:
        plt.show()
    plt.close()
    print(f"график сохранён: {out_path}")


def save_bar_plot(results: Dict[str, float], title: str, out_path: str,
                  ylabel: str = "время, c", show: bool = False):
    """Столбчатая диаграмма: подход -> значение (время или ускорение)."""
    import matplotlib
    if not show:
        matplotlib.use("Agg")
    import matplotlib.pyplot as plt

    names = list(results.keys())
    vals = [results[k] for k in names]
    plt.figure(figsize=(9, 5))
    plt.bar(names, vals, color="steelblue")
    plt.ylabel(ylabel)
    plt.title(title)
    plt.xticks(rotation=20, ha="right")
    plt.grid(True, axis="y", alpha=0.3)
    plt.tight_layout()
    os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True)
    plt.savefig(out_path, dpi=150)
    if show:
        plt.show()
    plt.close()
    print(f"график сохранён: {out_path}")


def median_time(fn: Callable, args: tuple = (), repeats: int = 5) -> Tuple[float, any]:
    """Аналог time_call, но по медиане — устойчив к выбросам."""
    times = []
    result = None
    for _ in range(repeats):
        t0 = time.perf_counter()
        result = fn(*args)
        times.append(time.perf_counter() - t0)
    return statistics.median(times), result