From 4f0c38500b065c4fcce45e6b2398bfe2e9054f6f Mon Sep 17 00:00:00 2001 From: jze9 Date: Wed, 8 Jul 2026 04:56:32 +0500 Subject: [PATCH] =?UTF-8?q?=D0=9C=D0=BE=D1=80=D0=B4=D0=B0=20O(1)=20+=20GPU?= =?UTF-8?q?=20=D0=B1=D0=B5=D0=B7=20=D1=83=D0=B4=D1=83=D1=88=D0=B5=D0=BD?= =?UTF-8?q?=D0=B8=D1=8F=20=D0=BE=D0=B4=D0=BD=D0=B8=D0=BC=20=D0=BF=D0=BE?= =?UTF-8?q?=D1=82=D0=BE=D0=BA=D0=BE=D0=BC=20Python?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Дашборд: /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 --- src/gausse/gpu/batch_sweep.py | 47 +++++++++++++++- src/gausse/optim/evolutionary.py | 42 ++++++++------ src/gausse/storage/database.py | 96 +++++++++++++++++++++++++++++++- src/gausse/storage/schema.py | 15 +++++ src/gausse/web/stats.py | 53 ++++++++++-------- 5 files changed, 210 insertions(+), 43 deletions(-) diff --git a/src/gausse/gpu/batch_sweep.py b/src/gausse/gpu/batch_sweep.py index 26ae632..119dde0 100644 --- a/src/gausse/gpu/batch_sweep.py +++ b/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, diff --git a/src/gausse/optim/evolutionary.py b/src/gausse/optim/evolutionary.py index 07f5e70..df9f47c 100644 --- a/src/gausse/optim/evolutionary.py +++ b/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( - 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: + # пул нужен в ОБОИХ режимах: на 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), + ) 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: diff --git a/src/gausse/storage/database.py b/src/gausse/storage/database.py index ac6b74a..fcdaff7 100644 --- a/src/gausse/storage/database.py +++ b/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() diff --git a/src/gausse/storage/schema.py b/src/gausse/storage/schema.py index a8d76a7..817cc84 100644 --- a/src/gausse/storage/schema.py +++ b/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)", ] diff --git a/src/gausse/web/stats.py b/src/gausse/web/stats.py index 727ec65..54fe2f9 100644 --- a/src/gausse/web/stats.py +++ b/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 = [] - 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}) + reasons = [ + {"reason": reason, "count": cnt} + for reason, cnt in conn.execute( + "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):