Морда O(1) + GPU без удушения одним потоком Python

Дашборд: /api/overview занимал 5.2с на 630к строк (сканы всей таблицы на
каждый рефреш при автообновлении 3с — морда хронически отставала, и с
ростом базы становилось бы хуже). Теперь:
- счётчики/гистограмма/причины/режимы — инкрементальные таблицы stats_*,
  обновляются В ТРАНЗАКЦИИ вставки (insert_runs), чтение O(1);
  ensure_stats() мигрирует старую базу одним пересчётом
- топы — частичные индексы (efficiency/exit_velocity DESC WHERE feasible=1)
- хвост лога — чтение последних 64КБ с конца, а не всего файла

GPU: сборка записей (decode+build_detail+JSON+стоимость) вынесена в пул
процессов (gpu_record_worker), GPU больше не ждёт один поток Python между
батчами.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/gausse/gpu/batch_sweep.py

@@ -156,10 +156,16 @@ def _simulate_states(xp, states: list) -> None:
             st.cur_x = p["coil_center"] + float(exit_x[k])
 
 
-def evaluate_genomes_gpu(xp, genomes: list, db: ComponentDatabase, bounds: SearchBounds) -> list:
+def evaluate_genomes_gpu(
+    xp, genomes: list, db: ComponentDatabase, bounds: SearchBounds, build_details: bool = True
+) -> list:
     """Оценка списка геномов ОДНИМ батчем (для эволюции): та же физика и та же
     формула фитнеса, что в objective.evaluate (КПД либо -1+доля пройденных
     ступеней), но разряды всех геномов интегрируются вместе на GPU/numpy.
+
+    build_details=False — не строить detail/стоимость (фитнесу они не нужны):
+    их параллельно собирает пул процессов (`gpu_record_worker`), иначе один
+    поток Python душит GPU (тот простаивал на ~2/3 времени поколения).
     """
     from gausse.optim.objective import EvaluationResult, compute_cost_rub
 
@@ -175,12 +181,12 @@ def evaluate_genomes_gpu(xp, genomes: list, db: ComponentDatabase, bounds: Searc
 
     results = []
     for st in states:
-        cost = compute_cost_rub(st.config, db)
+        cost = compute_cost_rub(st.config, db) if build_details else None
         ok = st.alive and st.energy_in_j > 0
         detail = build_detail(
             st.config, _FEASIBLE if ok else None, db, st.initial_x_m, st.initial_v_mps,
             st.genome.tube_inner_d_m, st.genome.tube_wall_m,
-        )
+        ) if build_details else None
         if ok:
             eff = st.kinetic_delta_j / st.energy_in_j
             results.append(EvaluationResult(
@@ -204,6 +210,41 @@ def evaluate_genomes_gpu(xp, genomes: list, db: ComponentDatabase, bounds: Searc
     return results
 
 
+def gpu_record_worker(payload) -> RunRecord:
+    """Собирает RunRecord (decode + build_detail + стоимость + JSON) в процессе
+    пула — параллельно с GPU-оценкой следующих батчей. Тяжёлая часть записи
+    (детализация + сериализация) не должна душить GPU одним потоком Python.
+
+    Работает в воркере ProcessPoolExecutor с worker_context.init_worker
+    (база компонентов уже загружена в процессе).
+    """
+    from gausse.optim import worker_context
+    from gausse.optim.objective import MODEL_VERSION, compute_cost_rub
+
+    genome, outcome, search_mode = payload
+    db, bounds = worker_context.db, worker_context.bounds
+    config, ix, iv = decode(genome, db, bounds)
+    detail = build_detail(
+        config, _FEASIBLE if outcome["feasible"] else None, db, ix, iv,
+        genome.tube_inner_d_m, genome.tube_wall_m,
+    )
+    return RunRecord(
+        run_id=str(uuid.uuid4()),
+        timestamp=datetime.now(timezone.utc).isoformat(),
+        search_mode=search_mode,
+        genome_json=json.dumps(genome_to_dict(genome)),
+        decoded_summary_json=json.dumps(detail, ensure_ascii=False),
+        feasible=outcome["feasible"],
+        model_version=MODEL_VERSION,
+        infeasible_reason=outcome["reason"],
+        failed_stage_index=outcome["failed_stage_index"],
+        efficiency=outcome["efficiency"],
+        exit_velocity_mps=outcome["exit_velocity_mps"],
+        cost_rub=compute_cost_rub(config, db),
+        energy_breakdown_json=json.dumps(outcome["energy_breakdown"]) if outcome["energy_breakdown"] else None,
+    )
+
+
 def run_gpu_sweep(
     db_path: Path,
     n_runs: int,

src/gausse/optim/evolutionary.py

@@ -7,7 +7,6 @@
 компонентов.
 """
 
-import contextlib
 import copy
 import multiprocessing
 import random
@@ -141,28 +140,39 @@ def run_evolution(
     n_evaluated = 0
 
     try:
-        executor_cm = (
-            ProcessPoolExecutor(
+        # пул нужен в ОБОИХ режимах: на CPU он оценивает геномы, в GPU-режиме
+        # параллельно собирает записи (detail+JSON) — иначе один поток Python
+        # душит GPU (карта простаивала ~2/3 времени поколения)
+        with ProcessPoolExecutor(
             max_workers=n_workers,
             mp_context=ctx,
             initializer=worker_context.init_worker,
             initargs=(data_dir, bounds),
-            )
-            if gpu_xp is None
-            else contextlib.nullcontext()
-        )
-        with executor_cm as executor:
+        ) as executor:
             for generation in range(n_generations):
                 if gpu_xp is not None:
-                    from gausse.gpu.batch_sweep import evaluate_genomes_gpu
+                    from gausse.gpu.batch_sweep import evaluate_genomes_gpu, gpu_record_worker
 
-                    pairs = list(zip(population, evaluate_genomes_gpu(gpu_xp, population, db, bounds)))
+                    results = evaluate_genomes_gpu(gpu_xp, population, db, bounds, build_details=False)
+                    pairs = list(zip(population, results))
+                    payload = [
+                        (g, {
+                            "feasible": r.feasible,
+                            "efficiency": r.efficiency,
+                            "exit_velocity_mps": r.exit_velocity_mps,
+                            "reason": r.reason,
+                            "failed_stage_index": r.failed_stage_index,
+                            "energy_breakdown": r.energy_breakdown,
+                        }, mode)
+                        for g, r in pairs
+                    ]
+                    records = executor.map(gpu_record_worker, payload, chunksize=64)
                 else:
                     pairs = list(executor.map(_evaluate_genome, population))
+                    records = (build_run_record(g, r, db, search_mode=mode) for g, r in pairs)
                 n_evaluated += len(pairs)
 
-                for genome, result in pairs:
-                    record: RunRecord = build_run_record(genome, result, db, search_mode=mode)
+                for record, (genome, result) in zip(records, pairs):
                     queue.put(record)
                     logger.update(result.feasible, result.efficiency)
                     if best_pair is None or result.fitness > best_pair[1].fitness:

src/gausse/storage/database.py

@@ -12,7 +12,7 @@ import sys
 from dataclasses import asdict, fields
 from pathlib import Path
 
-from gausse.storage.schema import CREATE_INDEXES_SQL, CREATE_RUNS_TABLE_SQL, RunRecord
+from gausse.storage.schema import CREATE_INDEXES_SQL, CREATE_RUNS_TABLE_SQL, CREATE_STATS_SQL, RunRecord
 
 _COLUMNS = [f.name for f in fields(RunRecord)]
 
@@ -25,10 +25,101 @@ def open_connection(db_path: Path) -> sqlite3.Connection:
     conn.execute(CREATE_RUNS_TABLE_SQL)
     for statement in CREATE_INDEXES_SQL:
         conn.execute(statement)
+    for statement in CREATE_STATS_SQL:
+        conn.execute(statement)
     conn.commit()
     return conn
 
 
+def short_reason(reason: str | None) -> str | None:
+    """Причина отказа без деталей конкретного прогона (для группировки)."""
+    return reason.split(":", 1)[0] if reason else reason
+
+
+def _efficiency_bucket(efficiency: float | None) -> int | None:
+    """Корзина гистограммы КПД 0..9 (та же логика, что раньше в SQL-агрегате)."""
+    if efficiency is None:
+        return None
+    b = min(int(efficiency * 10), 9)
+    return b if 0 <= b <= 9 else None
+
+
+def _apply_stats(conn: sqlite3.Connection, records: list[RunRecord]) -> None:
+    """Инкремент агрегатов дашборда — вызывается внутри транзакции вставки."""
+    n_feasible = sum(1 for r in records if r.feasible)
+    conn.execute(
+        "INSERT INTO stats_counters(key, value) VALUES('total', ?) "
+        "ON CONFLICT(key) DO UPDATE SET value = value + excluded.value",
+        (len(records),),
+    )
+    conn.execute(
+        "INSERT INTO stats_counters(key, value) VALUES('feasible', ?) "
+        "ON CONFLICT(key) DO UPDATE SET value = value + excluded.value",
+        (n_feasible,),
+    )
+    hist: dict[int, int] = {}
+    reasons: dict[str, int] = {}
+    modes: dict[str, int] = {}
+    for r in records:
+        modes[r.search_mode] = modes.get(r.search_mode, 0) + 1
+        if r.feasible:
+            b = _efficiency_bucket(r.efficiency)
+            if b is not None:
+                hist[b] = hist.get(b, 0) + 1
+        elif r.infeasible_reason:
+            key = short_reason(r.infeasible_reason)
+            reasons[key] = reasons.get(key, 0) + 1
+    conn.executemany(
+        "INSERT INTO stats_hist(bucket, count) VALUES(?, ?) "
+        "ON CONFLICT(bucket) DO UPDATE SET count = count + excluded.count",
+        list(hist.items()),
+    )
+    conn.executemany(
+        "INSERT INTO stats_reasons(reason, count) VALUES(?, ?) "
+        "ON CONFLICT(reason) DO UPDATE SET count = count + excluded.count",
+        list(reasons.items()),
+    )
+    conn.executemany(
+        "INSERT INTO stats_modes(mode, count) VALUES(?, ?) "
+        "ON CONFLICT(mode) DO UPDATE SET count = count + excluded.count",
+        list(modes.items()),
+    )
+
+
+def rebuild_stats(conn: sqlite3.Connection) -> None:
+    """Пересобирает агрегаты из таблицы runs (миграция старой базы; один раз)."""
+    conn.execute("DELETE FROM stats_counters")
+    conn.execute("DELETE FROM stats_hist")
+    conn.execute("DELETE FROM stats_reasons")
+    conn.execute("DELETE FROM stats_modes")
+    total = conn.execute("SELECT COUNT(*) FROM runs").fetchone()[0]
+    feasible = conn.execute("SELECT COUNT(*) FROM runs WHERE feasible=1").fetchone()[0]
+    conn.execute("INSERT INTO stats_counters VALUES('total', ?)", (total,))
+    conn.execute("INSERT INTO stats_counters VALUES('feasible', ?)", (feasible,))
+    conn.execute(
+        "INSERT INTO stats_hist SELECT b, COUNT(*) FROM "
+        "(SELECT MIN(CAST(efficiency*10 AS INT), 9) AS b FROM runs WHERE feasible=1 AND efficiency IS NOT NULL) "
+        "WHERE b BETWEEN 0 AND 9 GROUP BY b"
+    )
+    for reason, cnt in conn.execute(
+        "SELECT infeasible_reason, COUNT(*) FROM runs WHERE feasible=0 AND infeasible_reason IS NOT NULL "
+        "GROUP BY infeasible_reason"
+    ).fetchall():
+        conn.execute(
+            "INSERT INTO stats_reasons VALUES(?, ?) ON CONFLICT(reason) DO UPDATE SET count = count + excluded.count",
+            (short_reason(reason), cnt),
+        )
+    conn.execute("INSERT INTO stats_modes SELECT search_mode, COUNT(*) FROM runs GROUP BY search_mode")
+    conn.commit()
+
+
+def ensure_stats(conn: sqlite3.Connection) -> None:
+    """Если агрегаты пусты, а прогоны есть — база со старой схемой: пересобрать."""
+    has_stats = conn.execute("SELECT COUNT(*) FROM stats_counters").fetchone()[0] > 0
+    if not has_stats and conn.execute("SELECT EXISTS(SELECT 1 FROM runs)").fetchone()[0]:
+        rebuild_stats(conn)
+
+
 def _record_to_params(record: RunRecord) -> dict:
     row = asdict(record)
     row["feasible"] = int(row["feasible"])
@@ -54,6 +145,9 @@ def insert_runs(conn: sqlite3.Connection, records: list[RunRecord]) -> None:
         f"INSERT INTO runs ({columns}) VALUES ({placeholders})",
         [_record_to_params(r) for r in records],
     )
+    # агрегаты дашборда — в той же транзакции: либо и строки, и счётчики,
+    # либо ничего (иначе цифры на морде разъедутся с таблицей)
+    _apply_stats(conn, records)
     conn.commit()
 
 

src/gausse/storage/schema.py

@@ -31,6 +31,21 @@ CREATE_INDEXES_SQL = [
     "CREATE INDEX IF NOT EXISTS idx_runs_efficiency ON runs(efficiency)",
     "CREATE INDEX IF NOT EXISTS idx_runs_velocity ON runs(exit_velocity_mps)",
     "CREATE INDEX IF NOT EXISTS idx_runs_search_mode ON runs(search_mode)",
+    # частичные индексы под топы дашборда: ORDER BY ... DESC при feasible=1
+    # без них top-15 на сотнях тысяч строк занимал ~1.2с на каждый рефреш
+    "CREATE INDEX IF NOT EXISTS idx_runs_feas_eff ON runs(efficiency DESC) WHERE feasible=1",
+    "CREATE INDEX IF NOT EXISTS idx_runs_feas_vel ON runs(exit_velocity_mps DESC) WHERE feasible=1",
+]
+
+# Инкрементальные агрегаты для дашборда: счётчики/гистограмма/причины/режимы
+# обновляются В ТОЙ ЖЕ транзакции, что и вставка прогонов (insert_runs), и
+# читаются за O(1) — иначе каждый рефреш морды сканировал ВСЮ таблицу runs
+# (5.2с на 630к строк, а база растёт годами).
+CREATE_STATS_SQL = [
+    "CREATE TABLE IF NOT EXISTS stats_counters (key TEXT PRIMARY KEY, value INTEGER NOT NULL)",
+    "CREATE TABLE IF NOT EXISTS stats_hist (bucket INTEGER PRIMARY KEY, count INTEGER NOT NULL)",
+    "CREATE TABLE IF NOT EXISTS stats_reasons (reason TEXT PRIMARY KEY, count INTEGER NOT NULL)",
+    "CREATE TABLE IF NOT EXISTS stats_modes (mode TEXT PRIMARY KEY, count INTEGER NOT NULL)",
 ]
 
 

src/gausse/web/stats.py

@@ -4,13 +4,22 @@ import json
 import time
 from pathlib import Path
 
-from gausse.storage.database import fetch_run_by_id, fetch_runs, open_connection
+from gausse.storage.database import ensure_stats, fetch_run_by_id, fetch_runs, open_connection
 
 
 def _tail_log(log_path: Path, n_lines: int = 40) -> list[str]:
+    """Последние строки лога чтением С КОНЦА файла (лог растёт годами —
+    перечитывать его целиком на каждый рефреш нельзя)."""
     if not log_path.exists():
         return []
-    lines = log_path.read_text(encoding="utf-8", errors="replace").splitlines()
+    with open(log_path, "rb") as f:
+        f.seek(0, 2)
+        size = f.tell()
+        f.seek(max(0, size - 64 * 1024))
+        chunk = f.read().decode("utf-8", errors="replace")
+    lines = chunk.splitlines()
+    if size > 64 * 1024 and lines:
+        lines = lines[1:]  # первая строка чанка может быть обрезана
     return lines[-n_lines:]
 
 
@@ -32,34 +41,32 @@ def _gpu_stat(db_path: Path) -> dict | None:
 
 
 def overview(db_path: Path, log_path: Path | None = None, top_n: int = 15) -> dict:
-    """Сводка для дашборда: счётчики, гистограмма КПД, топ конфигураций, хвост лога."""
+    """Сводка для дашборда: счётчики, гистограмма КПД, топ конфигураций, хвост лога.
+
+    Счётчики/гистограмма/причины/режимы читаются из инкрементальных
+    агрегатов (stats_*, ведутся в транзакции вставки) за O(1) — раньше
+    каждый рефреш сканировал всю таблицу runs (5.2с на 630к строк).
+    """
     conn = open_connection(db_path)
     try:
-        total = conn.execute("SELECT COUNT(*) FROM runs").fetchone()[0]
-        feasible = conn.execute("SELECT COUNT(*) FROM runs WHERE feasible=1").fetchone()[0]
+        ensure_stats(conn)  # миграция базы со старой схемой (один раз)
+        counters = dict(conn.execute("SELECT key, value FROM stats_counters"))
+        total = counters.get("total", 0)
+        feasible = counters.get("feasible", 0)
         infeasible = total - feasible
 
-        # распределение по типу датчика ступени 0 (реализуемо / всего)
-        by_mode = {}
-        for mode, cnt in conn.execute("SELECT search_mode, COUNT(*) FROM runs GROUP BY search_mode"):
-            by_mode[mode] = cnt
+        by_mode = dict(conn.execute("SELECT mode, count FROM stats_modes"))
 
-        # причины отказа (топ)
-        reasons = []
+        reasons = [
+            {"reason": reason, "count": cnt}
             for reason, cnt in conn.execute(
-            "SELECT infeasible_reason, COUNT(*) c FROM runs WHERE feasible=0 AND infeasible_reason IS NOT NULL "
-            "GROUP BY infeasible_reason ORDER BY c DESC LIMIT 10"
-        ):
-            short = reason.split(":", 1)[0] if reason else reason
-            reasons.append({"reason": short, "count": cnt})
+                "SELECT reason, count FROM stats_reasons ORDER BY count DESC LIMIT 10"
+            )
+        ]
 
-        # гистограмма КПД — агрегируем в SQL (не тянем все строки в питон)
         buckets = [0] * 10  # 0-10%,...,90-100%
-        for bucket, cnt in conn.execute(
-            "SELECT MIN(CAST(efficiency*10 AS INT), 9) AS b, COUNT(*) FROM runs "
-            "WHERE feasible=1 AND efficiency IS NOT NULL GROUP BY b"
-        ):
-            if bucket is not None and 0 <= bucket <= 9:
+        for bucket, cnt in conn.execute("SELECT bucket, count FROM stats_hist"):
+            if 0 <= bucket <= 9:
                 buckets[int(bucket)] = cnt
 
         def _summarize(rows):