From bb5cb6fa83e353799109ca709af762d8e57a994b Mon Sep 17 00:00:00 2001 From: Peter Lord Date: Fri, 10 Jul 2026 11:15:22 -0700 Subject: [PATCH] Parallelize the snapshot rebuild across cores (opt-in, validated) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The snapshot walk reads every run blob (~750k files) and is CPU-bound in the pure-Python accumulation, so it takes ~80 min single-threaded. Split _build_cache_data into _accumulate (per-run-chunk) + _finalize, so the walk can run as a map-reduce: each core accumulates a chunk in its own process, the raw accumulators are merged, and the result is finalized once. Every accumulator is a plain sum (or min/max for the three records), so the merge is lossless. OFF by default (ENTITY_STATS_REBUILD_WORKERS=1). Set it to N>1 to use N cores, with an automatic fall back to serial on any error. Uses the "spawn" start method, not fork: this runs in the leader's refresher thread, and forking a multithreaded process can deadlock the child and hang the refresher. Safety: the refactored serial path is byte-identical to the original (verified), and backend/scripts/validate_parallel_rebuild.py runs the full walk both ways and proves they match — run it against the real run data before enabling parallel. --- backend/app/services/charts_stats.py | 44 ++ backend/app/services/community_stats.py | 61 +++ backend/app/services/encounter_stats.py | 19 + backend/app/services/run_entity_stats.py | 401 ++++++++++++++----- backend/scripts/validate_parallel_rebuild.py | 89 ++++ 5 files changed, 513 insertions(+), 101 deletions(-) create mode 100644 backend/scripts/validate_parallel_rebuild.py diff --git a/backend/app/services/charts_stats.py b/backend/app/services/charts_stats.py index b6b1ec5dc..588ff5087 100644 --- a/backend/app/services/charts_stats.py +++ b/backend/app/services/charts_stats.py @@ -641,6 +641,50 @@ def new_accumulator() -> dict[str, Any]: return {b: _new_acc_one() for b in _BLOB_BRACKETS} +# Merging two accumulators (from parallel run-chunk walks) adds per key. Most +# charts cells are `key -> [numbers]` (element-wise add); enc_hist and death_room +# are `key -> count` (scalar add). +_CHART_LIST_FIELDS = ( + "hp_floor", + "enc", + "traj", + "elites", + "smiths", + "events", + "entity_week", + "week_totals", + "copies", + "ench", +) +_CHART_COUNTER_FIELDS = ("enc_hist", "death_room") + + +def _merge_list_dict(dst: dict, src: dict) -> None: + """dst[key] += src[key] element-wise (both are lists of the same length).""" + for k, v in src.items(): + cur = dst.get(k) + if cur is None: + dst[k] = list(v) + else: + for i, x in enumerate(v): + cur[i] += x + + +def merge(dst: dict, src: dict) -> None: + """Fold accumulator `src` into `dst` (both from new_accumulator()).""" + for bracket, s in src.items(): + d = dst.get(bracket) + if d is None: + dst[bracket] = s + continue + for field in _CHART_LIST_FIELDS: + _merge_list_dict(d[field], s[field]) + for field in _CHART_COUNTER_FIELDS: + df = d[field] + for k, v in s[field].items(): + df[k] = df.get(k, 0) + v + + def _bump2(d: dict, key: Any, win: bool) -> None: cell = d.setdefault(key, [0, 0]) cell[0] += 1 diff --git a/backend/app/services/community_stats.py b/backend/app/services/community_stats.py index 4003324c2..87be43893 100644 --- a/backend/app/services/community_stats.py +++ b/backend/app/services/community_stats.py @@ -112,6 +112,67 @@ def new_accumulator() -> dict[str, Any]: return {b: _new_acc_one() for b in _BLOB_BRACKETS} +# Field groups for merging two accumulators (from parallel run-chunk walks). +# `_INT_FIELDS` add; `_LIST_DICT_FIELDS` add element-wise per key; `events` is a +# nested event -> {option -> count}; the counter dicts add per key; the three +# records keep the best (min run_time / max run_time / max deck_size). +_COMMUNITY_INT_FIELDS = ("total_runs", "total_wins", "reward_screens", "reward_skips") +_COMMUNITY_LIST_DICT_FIELDS = ("by_ascension", "by_character", "map_danger", "rest") +_COMMUNITY_COUNTER_FIELDS = ( + "deaths_encounter", + "deaths_event", + "ancient", + "removed", + "stolen", +) + + +def merge(dst: dict, src: dict) -> None: + """Fold accumulator `src` into `dst` (both from new_accumulator()). Called in + run-chunk order, so records tie-break to the earlier chunk (matching the + serial walk's first-seen-wins on an exact tie).""" + for bracket, s in src.items(): + d = dst.get(bracket) + if d is None: + dst[bracket] = s + continue + for f in _COMMUNITY_INT_FIELDS: + d[f] += s[f] + for f in _COMMUNITY_LIST_DICT_FIELDS: + df = d[f] + for k, v in s[f].items(): + cur = df.get(k) + if cur is None: + df[k] = list(v) + else: + for i, x in enumerate(v): + cur[i] += x + for f in _COMMUNITY_COUNTER_FIELDS: + df = d[f] + for k, v in s[f].items(): + df[k] = df.get(k, 0) + v + # events: event_id -> {option_id -> count} + de = d["events"] + for eid, opts in s["events"].items(): + cur = de.setdefault(eid, {}) + for opt, n in opts.items(): + cur[opt] = cur.get(opt, 0) + n + # Records: replace only when src is STRICTLY better, so the earlier chunk + # (already in dst) wins ties, matching the serial walk. + if s["fastest_win"] is not None and ( + d["fastest_win"] is None or s["fastest_win"][0] < d["fastest_win"][0] + ): + d["fastest_win"] = s["fastest_win"] + if s["longest_run"] is not None and ( + d["longest_run"] is None or s["longest_run"][0] > d["longest_run"][0] + ): + d["longest_run"] = s["longest_run"] + if s["biggest_deck"] is not None and ( + d["biggest_deck"] is None or s["biggest_deck"][0] > d["biggest_deck"][0] + ): + d["biggest_deck"] = s["biggest_deck"] + + def _bump(d: dict, key: Any, n: int = 1) -> None: d[key] = d.get(key, 0) + n diff --git a/backend/app/services/encounter_stats.py b/backend/app/services/encounter_stats.py index e77a4b14d..b5256aa36 100644 --- a/backend/app/services/encounter_stats.py +++ b/backend/app/services/encounter_stats.py @@ -57,6 +57,25 @@ def new_accumulator() -> dict[str, Any]: return {b: _new_acc_one() for b in _BLOB_BRACKETS} +def merge(dst: dict, src: dict) -> None: + """Fold accumulator `src` into `dst` (both from new_accumulator()). The only + field is `cells`: key -> [total, fatal, total_damage, total_turns], a plain + element-wise add — so parallel run-chunk walks combine losslessly.""" + for bracket, s in src.items(): + d = dst.get(bracket) + if d is None: + dst[bracket] = s + continue + dcells = d["cells"] + for k, v in s["cells"].items(): + cur = dcells.get(k) + if cur is None: + dcells[k] = list(v) + else: + for i, x in enumerate(v): + cur[i] += x + + def accumulate( acc: dict[str, Any], blob: dict, diff --git a/backend/app/services/run_entity_stats.py b/backend/app/services/run_entity_stats.py index 96fa9688c..50619d32c 100644 --- a/backend/app/services/run_entity_stats.py +++ b/backend/app/services/run_entity_stats.py @@ -828,23 +828,12 @@ def _finalize_bracket(acc: dict[str, Any]) -> tuple[dict, dict]: return cache, baselines -def _build_cache_data() -> tuple[dict, dict, dict, dict]: - """Walk every run JSON + DB row and return (cache, totals, - type_baselines, bracket_meta) WITHOUT mutating module globals. The heavy - 100k-file walk lives here so the leader refresher can compute + persist a - snapshot, and the in-process fallback can reuse the same logic. +def _accumulate(rows, official_chars, wr_map): + """Accumulate a chunk of run rows into the raw (pre-finalize) accumulators. - The all-runs aggregate lives in the top-level entity fields (what - scores/stats read). Each run ALSO feeds the brackets it matches - (solo/2p/3p/4p/a10/daily/custom); those land nested under each entity's - ``brackets`` key, and `bracket_meta` carries their baselines + totals. + Pure over its inputs (no module-global mutation) so N chunks can run in + parallel and have their raw accumulators merged. """ - # Phase timing so a stalled rebuild is diagnosable from the logs (the walk is - # otherwise silent until it completes). Which phase the last log names tells - # us where a rebuild that never finishes is stuck. - _t0 = time.time() - logger.info("snapshot rebuild: starting (loading run rows from the DB)") - new_cache: dict[tuple[str, str], dict[str, Any]] = {} new_totals = {"total_runs": 0, "total_wins": 0} @@ -867,86 +856,6 @@ def _build_cache_data() -> tuple[dict, dict, dict, dict]: k: _new_bracket_acc() for k in _BRACKET_KEYS } - # Source of truth depends on which DB the app is using. Either way we end - # up with rows carrying win/character/submission + the bracket fields - # (player_count / ascension / game_mode). - if _USING_MONGO: - from .runs_db_mongo import _get_collection - - coll = _get_collection() - rows = list( - coll.find( - {}, - { - "_id": 1, - "character": 1, - "win": 1, - "submitted_at": 1, - "player_count": 1, - "ascension": 1, - "game_mode": 1, - "killed_by": 1, - # Owner key for the win-rate skill brackets (wr30/50/75). - "username": 1, - }, - ) - ) - # Normalise to a common dict-like shape so the loop below - # doesn't have to branch. - rows = [ - { - "run_hash": d["_id"], - "character": d.get("character") or "", - "win": bool(d.get("win")), - "submitted_at": d.get("submitted_at"), - "player_count": d.get("player_count") or 1, - "ascension": d.get("ascension") or 0, - "game_mode": d.get("game_mode") or "standard", - "killed_by": d.get("killed_by"), - "username": d.get("username") or "", - } - for d in rows - ] - else: - with get_conn() as conn: - rows = conn.execute( - "SELECT run_hash, character, win, submitted_at, " - "player_count, ascension, game_mode, killed_by, username FROM runs" - ).fetchall() - rows = [dict(r) for r in rows] - - logger.info( - "snapshot rebuild: loaded %d run rows in %.1fs, now walking blob files", - len(rows), - time.time() - _t0, - ) - - official_chars = _official_character_ids() - - # Per-username overall win rate (fraction, 5-run floor) for the skill-bracket - # brackets. Computed from the loaded rows (username + win) so it's DB-agnostic - # (works on the SQLite fallback too) yet matches get_user_winrates exactly, - # the same definition the /api/runs/list winrate filter uses: every run - # counted, abandoned = loss, with a 5-run floor. - try: - from .runs_db_mongo import WINRATE_MIN_RUNS - except Exception: - WINRATE_MIN_RUNS = 5 - _wr_counts: dict[str, list[int]] = {} - for row in rows: - uname = (row.get("username") or "").lower() - if not uname: - continue - c = _wr_counts.setdefault(uname, [0, 0]) - c[0] += 1 - if row.get("win"): - c[1] += 1 - wr_map: dict[str, float] = { - u: (w / t) - for u, (t, w) in _wr_counts.items() - if t >= WINRATE_MIN_RUNS and t > 0 - } - for row in rows: # Official runs only: the game's ascensions are A0-A10. A11+ are modded # and a negative/placeholder like A-1 is bad data; both must be skipped @@ -1234,12 +1143,35 @@ def _build_cache_data() -> tuple[dict, dict, dict, dict]: skipped_ids, ) - logger.info( - "snapshot rebuild: blob walk done in %.1fs (%d entities), finalizing " - "(Codex Elo fits + bracket serialization)", - time.time() - _t0, - len(new_cache), - ) + return { + "new_cache": new_cache, + "new_totals": new_totals, + "pick_counts": pick_counts, + "pair_wins": pair_wins, + "upgrade_pair_wins": upgrade_pair_wins, + "bracket_accs": bracket_accs, + "community_acc": community_acc, + "charts_acc": charts_acc, + "encounter_acc": encounter_acc, + } + + +def _finalize( + new_cache, + new_totals, + pick_counts, + pair_wins, + upgrade_pair_wins, + bracket_accs, + community_acc, + charts_acc, + encounter_acc, +) -> tuple[dict, dict, dict, dict]: + """Fold the merged raw accumulators into the finalized snapshot payload. + + Runs once, after accumulation (serial or parallel+merged). + """ + from . import charts_stats, community_stats, encounter_stats # Fold the card-reward pick stats + fitted Codex Elo into the cache. # Cards that were offered but never decked still get an entry (picks=0 @@ -1334,6 +1266,273 @@ def _build_cache_data() -> tuple[dict, dict, dict, dict]: return new_cache, new_totals, new_type_baselines, bracket_meta +# ── Parallel rebuild (map-reduce over run chunks) ──────────────────────────── +# The walk is CPU-bound in the pure-Python accumulation, so it parallelizes +# across cores: split the rows into chunks, accumulate each in its own process +# (reusing _accumulate verbatim), merge the raw accumulators, finalize once. +# Every accumulator is a plain sum (or min/max for the three records), so the +# merge is lossless. OFF by default; ENTITY_STATS_REBUILD_WORKERS=N (N>1) enables +# it, with an automatic fall back to serial on any error. +_PARALLEL_MIN_RUNS = 5000 + + +def _rebuild_workers() -> int: + try: + n = int(os.environ.get("ENTITY_STATS_REBUILD_WORKERS", "1")) + except ValueError: + n = 1 + return max(1, min(n, os.cpu_count() or 1)) + + +def _merge_cache_entry(dst: dict, src: dict) -> None: + """Merge one entity aggregate `src` into `dst` (pre-finalize shape: picks/ + wins plus optional by_character / base / upg / act_picks / act_wins / + last-submission).""" + dst["picks"] += src["picks"] + dst["wins"] += src["wins"] + sbc = src.get("by_character") + if sbc: + dbc = dst.setdefault("by_character", {}) + for ch, cs in sbc.items(): + dc = dbc.setdefault(ch, {"picks": 0, "wins": 0}) + dc["picks"] += cs["picks"] + dc["wins"] += cs["wins"] + for k in ("base", "upg"): + sk = src.get(k) + if sk: + dk = dst.setdefault(k, {"picks": 0, "wins": 0}) + dk["picks"] += sk["picks"] + dk["wins"] += sk["wins"] + # Per-act relic pickup arrays (relics only): element-wise add. + for k in ("act_picks", "act_wins"): + sk = src.get(k) + if sk: + dk = dst.get(k) + if dk is None: + dst[k] = list(sk) + else: + for i, x in enumerate(sk): + dk[i] += x + # last submission: keep the later; the earlier chunk (already in dst) wins a + # tie, matching the serial walk's first-seen-on-tie (chunks merge in order). + sla = src.get("last_submitted_at") + if sla and (not dst.get("last_submitted_at") or sla > dst["last_submitted_at"]): + dst["last_submitted_at"] = sla + dst["last_run_hash"] = src.get("last_run_hash") + + +def _merge_pick_counts(dst: dict, src: dict) -> None: + for cid, pc in src.items(): + dp = dst.get(cid) + if dp is None: + dst[cid] = pc + continue + dp["offered"] += pc["offered"] + dp["picked"] += pc["picked"] + for i, x in enumerate(pc["off_act"]): + dp["off_act"][i] += x + for i, x in enumerate(pc["pick_act"]): + dp["pick_act"][i] += x + + +def _merge_counter(dst: dict, src: dict) -> None: + for k, v in src.items(): + dst[k] = dst.get(k, 0) + v + + +def _merge_raw(dst: dict, src: dict) -> None: + """Merge one chunk's raw accumulators `src` into `dst` in place.""" + from . import charts_stats, community_stats, encounter_stats + + for ent, se in src["new_cache"].items(): + de = dst["new_cache"].get(ent) + if de is None: + dst["new_cache"][ent] = se + else: + _merge_cache_entry(de, se) + dst["new_totals"]["total_runs"] += src["new_totals"]["total_runs"] + dst["new_totals"]["total_wins"] += src["new_totals"]["total_wins"] + _merge_pick_counts(dst["pick_counts"], src["pick_counts"]) + _merge_counter(dst["pair_wins"], src["pair_wins"]) + _merge_counter(dst["upgrade_pair_wins"], src["upgrade_pair_wins"]) + for ck, sacc in src["bracket_accs"].items(): + dacc = dst["bracket_accs"][ck] + for ent, se in sacc["cache"].items(): + de = dacc["cache"].get(ent) + if de is None: + dacc["cache"][ent] = se + else: + _merge_cache_entry(de, se) + _merge_pick_counts(dacc["pick_counts"], sacc["pick_counts"]) + _merge_counter(dacc["pair_wins"], sacc["pair_wins"]) + dacc["totals"]["total_runs"] += sacc["totals"]["total_runs"] + dacc["totals"]["total_wins"] += sacc["totals"]["total_wins"] + community_stats.merge(dst["community_acc"], src["community_acc"]) + charts_stats.merge(dst["charts_acc"], src["charts_acc"]) + encounter_stats.merge(dst["encounter_acc"], src["encounter_acc"]) + + +def _accumulate_worker(args): + """Top-level (picklable) entry point for a parallel chunk.""" + rows, official_chars, wr_map = args + return _accumulate(rows, official_chars, wr_map) + + +def _accumulate_parallel(rows, official_chars, wr_map, workers): + """Map-reduce _accumulate across processes. Contiguous chunks keep record + tie-breaks aligned with the serial order.""" + import concurrent.futures + import math + import multiprocessing + + n = len(rows) + size = math.ceil(n / workers) + chunks = [rows[i : i + size] for i in range(0, n, size)] + # Use "spawn": this runs inside the leader's refresher thread, and forking a + # multithreaded process can deadlock the child (a lock held by another thread + # at fork stays locked) — which would hang the refresher and stop all + # rebuilds. Spawn re-imports cleanly instead. Workers need only the run files + # + the args, so the ~1s startup is negligible against the walk. + try: + ctx = multiprocessing.get_context("spawn") + except ValueError: + ctx = None + with concurrent.futures.ProcessPoolExecutor( + max_workers=len(chunks), mp_context=ctx + ) as ex: + partials = list( + ex.map( + _accumulate_worker, + [(c, official_chars, wr_map) for c in chunks], + ) + ) + raw = partials[0] + for p in partials[1:]: + _merge_raw(raw, p) + return raw + + +def _build_cache_data() -> tuple[dict, dict, dict, dict]: + """Walk every run JSON + DB row and return (cache, totals, + type_baselines, bracket_meta) WITHOUT mutating module globals. The heavy + 100k-file walk lives here so the leader refresher can compute + persist a + snapshot, and the in-process fallback can reuse the same logic. + + The all-runs aggregate lives in the top-level entity fields (what + scores/stats read). Each run ALSO feeds the brackets it matches + (solo/2p/3p/4p/a10/daily/custom); those land nested under each entity's + ``brackets`` key, and `bracket_meta` carries their baselines + totals. + """ + # Phase timing so a stalled rebuild is diagnosable from the logs (the walk is + # otherwise silent until it completes). Which phase the last log names tells + # us where a rebuild that never finishes is stuck. + _t0 = time.time() + logger.info("snapshot rebuild: starting (loading run rows from the DB)") + + # Source of truth depends on which DB the app is using. Either way we end + # up with rows carrying win/character/submission + the bracket fields + # (player_count / ascension / game_mode). + if _USING_MONGO: + from .runs_db_mongo import _get_collection + + coll = _get_collection() + rows = list( + coll.find( + {}, + { + "_id": 1, + "character": 1, + "win": 1, + "submitted_at": 1, + "player_count": 1, + "ascension": 1, + "game_mode": 1, + "killed_by": 1, + # Owner key for the win-rate skill brackets (wr30/50/75). + "username": 1, + }, + ) + ) + # Normalise to a common dict-like shape so the loop below + # doesn't have to branch. + rows = [ + { + "run_hash": d["_id"], + "character": d.get("character") or "", + "win": bool(d.get("win")), + "submitted_at": d.get("submitted_at"), + "player_count": d.get("player_count") or 1, + "ascension": d.get("ascension") or 0, + "game_mode": d.get("game_mode") or "standard", + "killed_by": d.get("killed_by"), + "username": d.get("username") or "", + } + for d in rows + ] + else: + with get_conn() as conn: + rows = conn.execute( + "SELECT run_hash, character, win, submitted_at, " + "player_count, ascension, game_mode, killed_by, username FROM runs" + ).fetchall() + rows = [dict(r) for r in rows] + + logger.info( + "snapshot rebuild: loaded %d run rows in %.1fs, now walking blob files", + len(rows), + time.time() - _t0, + ) + + official_chars = _official_character_ids() + + # Per-username overall win rate (fraction, 5-run floor) for the skill-bracket + # brackets. Computed from the loaded rows (username + win) so it's DB-agnostic + # (works on the SQLite fallback too) yet matches get_user_winrates exactly, + # the same definition the /api/runs/list winrate filter uses: every run + # counted, abandoned = loss, with a 5-run floor. + try: + from .runs_db_mongo import WINRATE_MIN_RUNS + except Exception: + WINRATE_MIN_RUNS = 5 + _wr_counts: dict[str, list[int]] = {} + for row in rows: + uname = (row.get("username") or "").lower() + if not uname: + continue + c = _wr_counts.setdefault(uname, [0, 0]) + c[0] += 1 + if row.get("win"): + c[1] += 1 + wr_map: dict[str, float] = { + u: (w / t) + for u, (t, w) in _wr_counts.items() + if t >= WINRATE_MIN_RUNS and t > 0 + } + + workers = _rebuild_workers() + if workers > 1 and len(rows) >= _PARALLEL_MIN_RUNS: + logger.info( + "snapshot rebuild: accumulating in parallel across %d workers", workers + ) + try: + raw = _accumulate_parallel(rows, official_chars, wr_map, workers) + except Exception: + logger.warning( + "parallel rebuild failed; falling back to serial", exc_info=True + ) + raw = _accumulate(rows, official_chars, wr_map) + else: + raw = _accumulate(rows, official_chars, wr_map) + logger.info( + "snapshot rebuild: blob walk done in %.1fs (%d entities), finalizing " + "(Codex Elo fits + bracket serialization)", + time.time() - _t0, + len(raw["new_cache"]), + ) + + return _finalize(**raw) + + def _apply_cache( cache: dict, totals: dict, baselines: dict, bracket_meta: dict | None = None ) -> None: diff --git a/backend/scripts/validate_parallel_rebuild.py b/backend/scripts/validate_parallel_rebuild.py new file mode 100644 index 000000000..eacdb8c77 --- /dev/null +++ b/backend/scripts/validate_parallel_rebuild.py @@ -0,0 +1,89 @@ +"""Prove the parallel snapshot rebuild is identical to the serial one. + +Run inside the backend container against the real run data BEFORE enabling +parallel rebuilds: + + python -m scripts.validate_parallel_rebuild [workers] + +It runs the full walk twice in one process — serial, then parallel — and deep- +compares the results (ints exact, floats within 1e-9). Because both finalize in +the same process, hash-seed-dependent set ordering matches, so the compare is +exact. Exit 0 + "IDENTICAL" means the parallel path is safe to turn on +(ENTITY_STATS_REBUILD_WORKERS=); a MISMATCH means do NOT enable it. + +The `if __name__ == "__main__"` guard is required: the parallel path uses the +"spawn" start method, which re-imports this module in each worker. +""" + +import os +import sys +import time + + +def _diff(x, y, path=""): + if type(x) is not type(y): + return [f"{path}: type {type(x).__name__} vs {type(y).__name__}"] + if isinstance(x, dict): + out = [f"{path}: keys ({len(x)} vs {len(y)})"] if set(x) != set(y) else [] + for k in x: + if k in y: + out += _diff(x[k], y[k], f"{path}.{k}") + return out + if isinstance(x, (list, tuple)): + if len(x) != len(y): + return [f"{path}: len {len(x)} vs {len(y)}"] + # A list whose elements are themselves lists/tuples is a serialized + # dict/set (e.g. charts cells), so it's order-independent — the parallel + # merge builds it in a different order than the serial run, same data. + # Sort both before comparing so only genuine value differences surface. + # A list of scalars (e.g. per-act [n0, n1, n2]) is positional: compare + # in place. + if x and all(isinstance(e, (list, tuple)) for e in x): + xs = sorted(x, key=repr) + ys = sorted(y, key=repr) + else: + xs, ys = x, y + out = [] + for i, (u, v) in enumerate(zip(xs, ys)): + out += _diff(u, v, f"{path}[{i}]") + return out + if isinstance(x, float): + return [] if abs(x - y) < 1e-9 else [f"{path}: {x} vs {y}"] + return [] if x == y else [f"{path}: {x!r} vs {y!r}"] + + +def main() -> int: + workers = int(sys.argv[1]) if len(sys.argv) > 1 else (os.cpu_count() or 4) + import app.services.run_entity_stats as r + + r._PARALLEL_MIN_RUNS = 1 # force the parallel path regardless of run count + + print("running serial rebuild...", flush=True) + os.environ["ENTITY_STATS_REBUILD_WORKERS"] = "1" + t0 = time.time() + ser = r._build_cache_data() + t_ser = time.time() - t0 + print(f" serial: {len(ser[0])} entities in {t_ser:.1f}s", flush=True) + + print(f"running parallel rebuild ({workers} workers)...", flush=True) + os.environ["ENTITY_STATS_REBUILD_WORKERS"] = str(workers) + t0 = time.time() + par = r._build_cache_data() + t_par = time.time() - t0 + print(f" parallel: {len(par[0])} entities in {t_par:.1f}s", flush=True) + + d = _diff(ser, par) + print(f"\ndifferences: {len(d)}") + for line in d[:30]: + print(" ", line) + if d: + print("\nRESULT: MISMATCH — DO NOT enable parallel rebuilds") + return 1 + speedup = t_ser / t_par if t_par else 0 + print(f"\nRESULT: IDENTICAL — parallel is safe ({speedup:.1f}x faster). Enable") + print(f"with ENTITY_STATS_REBUILD_WORKERS={workers}.") + return 0 + + +if __name__ == "__main__": + sys.exit(main())