From b320aa118a69f92712cf6135525084e267612757 Mon Sep 17 00:00:00 2001 From: Prerna Kakkar Date: Mon, 24 Aug 2026 13:36:18 +0000 Subject: [PATCH 1/6] Improve UI performance --- evalbench_service/supervisord_cloudrun.conf | 2 + viewer/main.py | 54 ++++--- viewer/precompute_dataset_quality.py | 6 +- viewer/precompute_lease.py | 157 ++++++++++++++++++++ viewer/precompute_trends.py | 145 +++++++++++++++++- viewer/run_index.py | 77 ++++++++++ viewer/run_precompute.py | 30 ++-- viewer/trends.py | 9 +- viewer/version.txt | 2 +- 9 files changed, 425 insertions(+), 57 deletions(-) create mode 100644 viewer/precompute_lease.py create mode 100644 viewer/run_index.py diff --git a/evalbench_service/supervisord_cloudrun.conf b/evalbench_service/supervisord_cloudrun.conf index 83fae017..33a80acc 100644 --- a/evalbench_service/supervisord_cloudrun.conf +++ b/evalbench_service/supervisord_cloudrun.conf @@ -19,6 +19,8 @@ command=python /evalbench/viewer/run_precompute.py directory=/evalbench/viewer autostart=true autorestart=true + +environment=PRECOMPUTE_WORKERS=12,PRECOMPUTE_INTERVAL=600 stdout_logfile=/dev/stdout stdout_logfile_maxbytes=0 stderr_logfile=/dev/stderr diff --git a/viewer/main.py b/viewer/main.py index 038c8c7c..1bba90c7 100644 --- a/viewer/main.py +++ b/viewer/main.py @@ -7,6 +7,8 @@ import subprocess import precompute_trends import dataset_quality +import run_index +from run_index import list_run_directories from summarizer import summarize_eval_scoring from ai_comparer import compare_evals @@ -229,14 +231,7 @@ def get_color_for_pct(val_str): def on_load(e: me.LoadEvent): state = me.state(State) results_dir = get_results_dir() - directories = [] - if os.path.exists(results_dir): - # List directories only - directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] + directories = list_run_directories(results_dir) job_id = me.query_params.get("job_id") or me.query_params.get("jobid") if job_id and job_id in directories: @@ -259,14 +254,8 @@ def on_load(e: me.LoadEvent): def status_component(): results_dir = get_results_dir() - directories = [] - if os.path.exists(results_dir): - directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] - + directories = list_run_directories(results_dir) + with me.box( style=me.Style( background="#ffffff", @@ -2055,15 +2044,8 @@ def render_app_content(): results_dir = get_results_dir() logging.info(f"render_app_content: selected_directory='{state.selected_directory}', selected_evals='{state.selected_evals}', selected_main_tab='{state.selected_main_tab}'") - directories = [] - if os.path.exists(results_dir): - # List directories only - directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] - + directories = list_run_directories(results_dir) + def on_title_click(e: me.ClickEvent): state.selected_directory = "" state.conversation_index = 0 @@ -2084,7 +2066,11 @@ def on_clear_cache_click(e: me.ClickEvent): os.remove(filters_cache_file) logging.info("Cleared precomputed files. Triggering precompute...") - + + # "Clear cache" should mean every cache, including the + # in-process directory listing. + run_index.invalidate() + import threading threading.Thread(target=precompute_trends.precompute).start() @@ -2289,12 +2275,24 @@ def get_val(cfg_name): me.text("AI Summary", type="headline-5") if not state.ai_summary and state.selected_directory: + # One small file beside the run. The cache only carries + # summaries for the recent window, and reading all of it + # back to find a single run is what made this expensive. + state.ai_summary = precompute_trends.read_ai_summary( + results_dir, state.selected_directory + ) + trends_cache_file = os.path.join(results_dir, "trends_cache.csv") if os.path.exists(trends_cache_file): try: - cache_df = pd.read_csv(trends_cache_file) + wanted = ['job_id', 'ai_score'] + if not state.ai_summary: + wanted.append('ai_summary') + cache_df = pd.read_csv( + trends_cache_file, usecols=lambda c: c in wanted + ) run_data = cache_df[cache_df['job_id'] == state.selected_directory] - if not run_data.empty and 'ai_summary' in run_data.columns: + if not state.ai_summary and not run_data.empty and 'ai_summary' in run_data.columns: summary = run_data['ai_summary'].values[0] if not pd.isna(summary) and summary != "N/A": state.ai_summary = summary diff --git a/viewer/precompute_dataset_quality.py b/viewer/precompute_dataset_quality.py index e623860e..67b62adc 100644 --- a/viewer/precompute_dataset_quality.py +++ b/viewer/precompute_dataset_quality.py @@ -13,6 +13,8 @@ import logging import os +from run_index import list_run_directories + logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') CACHE_FILENAME = "dataset_quality_cache.json" @@ -146,8 +148,8 @@ def precompute(): logging.warning("Could not read %s, rebuilding: %s", processed_dirs_file, e) directories = [ - d for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) and d not in processed + d for d in list_run_directories(results_dir, force=True) + if d not in processed ] logging.info("Dataset quality: %d new directories to scan", len(directories)) diff --git a/viewer/precompute_lease.py b/viewer/precompute_lease.py new file mode 100644 index 00000000..0ae52d14 --- /dev/null +++ b/viewer/precompute_lease.py @@ -0,0 +1,157 @@ +"""Single-writer lease so only one instance precomputes at a time. + +Cloud Run scales the serving container up to maxScale, and supervisord starts a +precompute loop inside *every* instance. Each one then walks the same GCS-backed +results directory and processes the same backlog: five instances were observed +starting identical passes over the same 22k directories. + +That is worse than wasted work. Both caches are rewritten whole, and a GCS +object write replaces the object rather than merging, so when two instances +finish a batch the later write silently discards the earlier one's rows. The +runs stay recorded in processed_dirs.json either way, so anything lost that way +is never reconsidered and never appears in trends again. + +The lease is a small file in the results directory holding a holder id and a +wall-clock timestamp. Acquiring writes it, waits out the window in which a +competing write could still land, then reads it back: because a whole-file +rewrite has exactly one winner and GCS reads are strongly consistent after it, +whoever reads its own id back is the sole holder. The holder renews while it +works so a long pass keeps the lease, and the timestamp lets a survivor take +over if the holder dies -- which matters here, since a large backlog can get the +precompute process SIGKILLed mid-pass. + +Wall-clock time is deliberate: monotonic clocks are not comparable across +instances, and the whole point is comparing timestamps written by other hosts. +""" + +import logging +import os +import socket +import threading +import time +import uuid + +LEASE_FILENAME = "precompute.lease" + +# Long enough to ride out a slow FUSE write, short enough that a killed holder +# does not stall precompute for long. Renewal runs well inside it. +LEASE_TTL_SECONDS = float(os.environ.get("PRECOMPUTE_LEASE_TTL", "120")) +RENEW_INTERVAL_SECONDS = LEASE_TTL_SECONDS / 4 + +# Read-back is only meaningful once any competing write has had time to land. +GUARD_SECONDS = float(os.environ.get("PRECOMPUTE_LEASE_GUARD", "5")) + +_HOLDER_ID = f"{socket.gethostname()}:{os.getpid()}:{uuid.uuid4().hex[:8]}" + + +def _lease_path(results_dir): + return os.path.join(results_dir, LEASE_FILENAME) + + +def _read(results_dir): + """Return (holder, timestamp), or (None, 0.0) if absent or unreadable.""" + try: + with open(_lease_path(results_dir)) as f: + holder, _, timestamp = f.read().partition("\n") + return holder.strip(), float(timestamp.strip()) + except (OSError, ValueError): + # A torn read of a file being rewritten lands here too; treating it as + # "no lease" is safe because the read-back check still has to pass. + return None, 0.0 + + +def _write(results_dir): + with open(_lease_path(results_dir), "w") as f: + f.write(f"{_HOLDER_ID}\n{time.time()}\n") + + +class PrecomputeLease: + """Context manager guarding a precompute pass. Truthy only if held.""" + + def __init__(self, results_dir): + self.results_dir = results_dir + self.acquired = False + self._stop = threading.Event() + self._renewer = None + + def acquire(self): + holder, timestamp = _read(self.results_dir) + age = time.time() - timestamp + + if holder and holder != _HOLDER_ID and age < LEASE_TTL_SECONDS: + logging.info( + "Precompute lease held by %s (renewed %.0fs ago); skipping pass", + holder, age, + ) + return False + if holder and holder != _HOLDER_ID: + logging.warning( + "Taking over precompute lease from %s, stale for %.0fs", holder, age + ) + + try: + _write(self.results_dir) + except OSError: + # Never let a lease problem stop precompute outright: a dashboard + # that silently stops updating is a worse failure than duplicated + # work, which is only what happened before this lease existed. + logging.exception("Could not write precompute lease; running unguarded") + self.acquired = True + return True + + time.sleep(GUARD_SECONDS) + + winner, _ = _read(self.results_dir) + if winner != _HOLDER_ID: + logging.info("Lost precompute lease race to %s; skipping pass", winner) + return False + + self.acquired = True + self._renewer = threading.Thread( + target=self._renew_loop, name="precompute-lease-renew", daemon=True + ) + self._renewer.start() + logging.info("Acquired precompute lease as %s", _HOLDER_ID) + return True + + def _renew_loop(self): + while not self._stop.wait(RENEW_INTERVAL_SECONDS): + holder, _ = _read(self.results_dir) + if holder and holder != _HOLDER_ID: + # Someone judged us stale and took over. Two writers fighting + # over the file is the exact thing this exists to prevent, so + # stand down and re-acquire on the next pass. + logging.warning("Precompute lease taken over by %s; stopping renewal", holder) + return + try: + _write(self.results_dir) + except OSError: + logging.warning("Could not renew precompute lease") + + def release(self): + self._stop.set() + if self._renewer: + self._renewer.join(timeout=RENEW_INTERVAL_SECONDS) + self._renewer = None + if not self.acquired: + return + self.acquired = False + # Only clear the file if it is still ours; a successor's lease must not + # be deleted by the instance it replaced. + holder, _ = _read(self.results_dir) + if holder == _HOLDER_ID: + try: + os.remove(_lease_path(self.results_dir)) + except OSError: + logging.warning("Could not remove precompute lease file") + + def __enter__(self): + self.acquire() + return self + + def __exit__(self, *exc): + self.release() + return False + + def __bool__(self): + return self.acquired diff --git a/viewer/precompute_trends.py b/viewer/precompute_trends.py index f5571060..ba67bce4 100644 --- a/viewer/precompute_trends.py +++ b/viewer/precompute_trends.py @@ -4,6 +4,8 @@ import argparse import pandas as pd +from run_index import list_run_directories + logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') BATCH_SIZE = int(os.environ.get("PRECOMPUTE_BATCH_SIZE", 50)) @@ -14,6 +16,47 @@ # where the summarizer starts seeing 429s, so raising it further buys nothing. MAX_WORKERS = int(os.environ.get("PRECOMPUTE_WORKERS", 50)) +# ai_summary is ~96% of the bytes in trends_cache.csv, and that file is read in +# full on every page render. Only the recent window is kept inline; every +# summary is also written beside its own run, so the list view stays small +# without the older summaries being lost. +AI_SUMMARY_FILENAME = "ai_summary.txt" +AI_SUMMARY_CACHE_DAYS = int(os.environ.get("AI_SUMMARY_CACHE_DAYS", 7)) + + +def ai_summary_path(results_dir, job_id): + return os.path.join(results_dir, job_id, AI_SUMMARY_FILENAME) + + +def write_ai_summary(run_dir, summary): + """Persist a summary beside its run so trimming the cache is not lossy.""" + if not summary or summary == "N/A": + return + try: + with open(os.path.join(run_dir, AI_SUMMARY_FILENAME), "w") as f: + f.write(summary) + except OSError: + logging.warning("Could not write %s for %s", AI_SUMMARY_FILENAME, run_dir) + + +def read_ai_summary(results_dir, job_id): + """Return the stored summary for a run, or "" if it has none.""" + try: + with open(ai_summary_path(results_dir, job_id)) as f: + return f.read() + except OSError: + return "" + + +def _is_recent(run_time, days=AI_SUMMARY_CACHE_DAYS): + """Whether a run is inside the window that keeps its summary inline.""" + ts = pd.to_datetime(run_time, errors='coerce') + if pd.isna(ts): + # An unparseable run_time cannot be aged out safely, so treat it as old + # and let the sidecar serve it. + return False + return ts >= pd.Timestamp.now() - pd.Timedelta(days=days) + def get_results_dir(): # Try to read from environment variable @@ -35,6 +78,21 @@ def get_results_dir(): return results_dir_candidates[1] # Fallback to default +class UnparsableRun(Exception): + """A run whose CSVs are present but structurally unusable. + + Distinct from "not ready yet" (no summary.csv), which returns None so a + later pass retries. A run that is missing a required column looks the same + on every future pass, so retrying it forever burns FUSE reads and floods + the log without ever making progress -- record it as processed instead. + """ + + def __init__(self, job_id, reason): + super().__init__(reason) + self.job_id = job_id + self.reason = reason + + def process_directory(d, results_dir): run_dir = os.path.join(results_dir, d) configs_file = os.path.join(run_dir, "configs.csv") @@ -47,6 +105,8 @@ def process_directory(d, results_dir): try: # Read configs configs_df = pd.read_csv(configs_file) + if 'config' not in configs_df.columns: + raise UnparsableRun(d, "configs.csv has no 'config' column") # Extract requester, product, dataset and generator requester_row = configs_df[configs_df['config'].str.contains('guitar_requester', na=False)] @@ -61,6 +121,8 @@ def process_directory(d, results_dir): # Read summary summary_df = pd.read_csv(summary_file) + if 'metric_name' not in summary_df.columns: + raise UnparsableRun(d, "summary.csv has no 'metric_name' column") # Extract metrics latency_row = summary_df[summary_df['metric_name'] == 'end_to_end_latency'] @@ -118,7 +180,11 @@ def get_metric_pct(row): except Exception as e: logging.warning(f"Error reading {os.path.basename(file_to_read)} for {d}: {e}") - run_time = summary_df['run_time'].values[0] if not summary_df.empty else "unknown" + run_time = ( + summary_df['run_time'].values[0] + if 'run_time' in summary_df.columns and not summary_df.empty + else "unknown" + ) if run_time != "unknown": try: run_time = pd.to_datetime(run_time).strftime('%Y-%m-%d %H:%M:%S') @@ -140,6 +206,10 @@ def get_metric_pct(row): except Exception as e: logging.error(f"Error generating AI summary for {d}: {e}") + # Written for every run regardless of age; the cache keeps only the + # recent window inline, and the detail view falls back to this file. + write_ai_summary(run_dir, ai_summary) + logging.info(f"Successfully processed directory: {d}") return { 'run_time': run_time, @@ -157,11 +227,24 @@ def get_metric_pct(row): 'goal_completion': goal_completion, 'job_id': d, 'ai_score': ai_score, - 'ai_summary': ai_summary + 'ai_summary': ai_summary if _is_recent(run_time) else "", } + except UnparsableRun: + # Must outrun the catch-all below: returning None here would leave the + # run unprocessed and queue it up to fail identically on every pass. + raise + except (pd.errors.ParserError, pd.errors.EmptyDataError, OSError) as e: + # A run caught mid-write and a transient FUSE read both land here, so + # the artifacts may well parse on a later pass. Leave it queued. + logging.warning(f"Deferring {d}, artifacts not readable yet: {e}") + return None except Exception as e: + # Both artifacts exist and parsed, so anything left is a problem with + # their shape and will recur identically on every future pass. Returning + # None here is what jammed the queue: the run is retried forever and + # every pass re-reads it instead of draining the backlog. logging.exception(f"Error reading data from {d}") - return None + raise UnparsableRun(d, f"{type(e).__name__}: {e}") from e def _read_json(path, default): @@ -220,6 +303,47 @@ def _append_rows(cache_file, rows): new_df[header].to_csv(cache_file, mode="a", header=False, index=False) +def _trim_stale_summaries(results_dir, cache_file): + """Blank ai_summary on cache rows that have aged out of the inline window. + + Rows written by an earlier pass were recent when appended and go stale on + their own, so this runs every pass rather than only at write time. Each + summary is written to its sidecar first where one is missing -- without + that backfill this would destroy every summary already in the cache. + """ + if not os.path.exists(cache_file) or os.path.getsize(cache_file) == 0: + return + + try: + df = pd.read_csv(cache_file) + except Exception: + logging.exception("Could not read %s to trim summaries", cache_file) + return + + if not {'ai_summary', 'run_time', 'job_id'} <= set(df.columns): + return + + stale = ( + df['ai_summary'].fillna("").astype(str).str.len().gt(0) + & ~df['run_time'].map(_is_recent) + ) + if not stale.any(): + return + + backfilled = 0 + for row in df.loc[stale, ['job_id', 'ai_summary']].itertuples(index=False): + if not os.path.exists(ai_summary_path(results_dir, row.job_id)): + write_ai_summary(os.path.join(results_dir, row.job_id), row.ai_summary) + backfilled += 1 + + df.loc[stale, 'ai_summary'] = "" + df.to_csv(cache_file, index=False) + logging.info( + "Trimmed %d aged-out summaries from the trends cache (%d sidecars backfilled)", + int(stale.sum()), backfilled, + ) + + def precompute(): results_dir = get_results_dir() logging.info(f"Reading results from {results_dir}") @@ -235,11 +359,7 @@ def precompute(): processed_dirs = set(_read_json(processed_dirs_file, [])) logging.info(f"Loaded {len(processed_dirs)} processed directories from state.") - all_directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] + all_directories = list_run_directories(results_dir, force=True) # Filter for new directories new_directories = [d for d in all_directories if d not in processed_dirs] @@ -249,6 +369,8 @@ def precompute(): if total_new == 0: logging.info("No new directories to process.") + # Rows still age out of the inline window with no new runs at all. + _trim_stale_summaries(results_dir, cache_file) return # Carried forward from the last pass rather than recomputed off the trends @@ -279,6 +401,11 @@ def precompute(): for future, directory in futures.items(): try: res = future.result() + except UnparsableRun as e: + # Will never parse, so record it and stop reconsidering it. + logging.warning(f"Skipping {directory}: {e.reason}") + batch_processed.append(e.job_id) + continue except Exception: # One unreadable run must not cost the rest of the batch. logging.exception(f"Error processing {directory}") @@ -327,6 +454,8 @@ def precompute(): f"appended {len(rows)} rows, {len(processed_dirs)} directories processed." ) + _trim_stale_summaries(results_dir, cache_file) + logging.info(f"Precomputed trends data saved to {cache_file}") logging.info(f"Precomputed filter values saved to {filters_file}") logging.info(f"Saved {len(processed_dirs)} processed directories to state.") diff --git a/viewer/run_index.py b/viewer/run_index.py new file mode 100644 index 00000000..39f14a74 --- /dev/null +++ b/viewer/run_index.py @@ -0,0 +1,77 @@ +"""Cached listing of run directories under the results mount. + +The results directory is a GCS FUSE mount holding tens of thousands of run +directories. The obvious way to enumerate it -- + + [d for d in os.listdir(p) if os.path.isdir(os.path.join(p, d))] + +-- costs one FUSE stat round trip *per entry* on top of the listing itself, +because os.path.isdir cannot reuse anything os.listdir already learned. At +~25k entries that is ~25k round trips, and the Mesop render path ran that walk +three times per page, which pushed /__ui__ past the Cloud Run request timeout. + +os.scandir carries the directory-entry type through from the listing, so +entry.is_dir() is answered without a second trip. The short TTL then collapses +the repeated walks within a single render (and across renders in a session) +down to one. +""" + +import logging +import os +import threading +import time + +CACHE_TTL_SECONDS = float(os.environ.get("RUN_DIRS_CACHE_TTL", "60")) + +_lock = threading.Lock() +_cache = {"path": None, "dirs": [], "at": 0.0} + + +def list_run_directories(results_dir, force=False): + """Return the names of run directories directly under results_dir. + + Results are cached for CACHE_TTL_SECONDS. A missing results_dir yields an + empty list. If a refresh fails but a previous listing for the same path is + still held, the stale listing is returned rather than an empty one -- a + transient FUSE error should not blank out the dashboard. + """ + now = time.monotonic() + + with _lock: + if ( + not force + and _cache["path"] == results_dir + and now - _cache["at"] < CACHE_TTL_SECONDS + ): + return list(_cache["dirs"]) + + try: + dirs = [] + with os.scandir(results_dir) as entries: + for entry in entries: + try: + if entry.is_dir(follow_symlinks=False): + dirs.append(entry.name) + except OSError: + # Entry vanished mid-walk, or FUSE hiccuped on this one + # name. Skipping it beats failing the whole listing. + continue + except FileNotFoundError: + dirs = [] + except OSError: + logging.exception("Failed to list run directories in %s", results_dir) + with _lock: + if _cache["path"] == results_dir: + return list(_cache["dirs"]) + return [] + + with _lock: + _cache.update({"path": results_dir, "dirs": dirs, "at": now}) + + return list(dirs) + + +def invalidate(): + """Drop the cached listing so the next call re-walks the mount.""" + with _lock: + _cache.update({"path": None, "dirs": [], "at": 0.0}) diff --git a/viewer/run_precompute.py b/viewer/run_precompute.py index 91a805f5..64e2ff4c 100644 --- a/viewer/run_precompute.py +++ b/viewer/run_precompute.py @@ -7,22 +7,28 @@ import precompute_trends import precompute_dataset_quality +from precompute_lease import PrecomputeLease def main(): interval = int(os.environ.get("PRECOMPUTE_INTERVAL", 300)) + results_dir = precompute_trends.get_results_dir() while True: - # Dataset quality runs first because it is cheap and always finishes. The - # trends pass makes an LLM call per unprocessed run and gets SIGKILLed - # mid-pass on a large backlog, which the try/except below cannot catch and - # which would otherwise starve every pass after it. - for precompute in ( - precompute_dataset_quality.precompute, - precompute_trends.precompute, - ): - try: - precompute() - except Exception as e: - print(f"Error in {precompute.__module__}: {e}") + # Every Cloud Run instance runs this loop, but they all share one results + # directory, so only the lease holder may write to the caches. + with PrecomputeLease(results_dir) as lease: + if lease: + # Dataset quality runs first because it is cheap and always finishes. + # The trends pass makes an LLM call per unprocessed run and gets + # SIGKILLed mid-pass on a large backlog, which the try/except below + # cannot catch and which would otherwise starve every pass after it. + for precompute in ( + precompute_dataset_quality.precompute, + precompute_trends.precompute, + ): + try: + precompute() + except Exception as e: + print(f"Error in {precompute.__module__}: {e}") time.sleep(interval) if __name__ == "__main__": diff --git a/viewer/trends.py b/viewer/trends.py index 1f1c7436..ab3a3aa2 100644 --- a/viewer/trends.py +++ b/viewer/trends.py @@ -3,6 +3,7 @@ import mesop as me import pandas as pd from main import State +from run_index import list_run_directories def get_results_dir(): # Try to read from environment variable @@ -103,12 +104,8 @@ def trends_component(): # Fallback to computing on the fly if cache is missing or failed if df is None: - directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] - + directories = list_run_directories(results_dir) + data = [] for d in directories: diff --git a/viewer/version.txt b/viewer/version.txt index 3cd76c90..fde302eb 100644 --- a/viewer/version.txt +++ b/viewer/version.txt @@ -1 +1 @@ -797b8c6 +1480f53 From 79db00708bb6d101a40dd2a3f4064a7545d97dee Mon Sep 17 00:00:00 2001 From: Prerna Kakkar Date: Mon, 24 Aug 2026 13:52:34 +0000 Subject: [PATCH 2/6] remove comments --- viewer/main.py | 6 +---- viewer/precompute_lease.py | 49 ++++++++++++------------------------- viewer/precompute_trends.py | 30 ++++++++--------------- viewer/run_index.py | 28 ++++++--------------- viewer/run_precompute.py | 3 +-- 5 files changed, 36 insertions(+), 80 deletions(-) diff --git a/viewer/main.py b/viewer/main.py index 1bba90c7..061b539a 100644 --- a/viewer/main.py +++ b/viewer/main.py @@ -2067,8 +2067,6 @@ def on_clear_cache_click(e: me.ClickEvent): logging.info("Cleared precomputed files. Triggering precompute...") - # "Clear cache" should mean every cache, including the - # in-process directory listing. run_index.invalidate() import threading @@ -2275,9 +2273,7 @@ def get_val(cfg_name): me.text("AI Summary", type="headline-5") if not state.ai_summary and state.selected_directory: - # One small file beside the run. The cache only carries - # summaries for the recent window, and reading all of it - # back to find a single run is what made this expensive. + # The cache only carries summaries for the recent window. state.ai_summary = precompute_trends.read_ai_summary( results_dir, state.selected_directory ) diff --git a/viewer/precompute_lease.py b/viewer/precompute_lease.py index 0ae52d14..6594f011 100644 --- a/viewer/precompute_lease.py +++ b/viewer/precompute_lease.py @@ -1,27 +1,15 @@ """Single-writer lease so only one instance precomputes at a time. -Cloud Run scales the serving container up to maxScale, and supervisord starts a -precompute loop inside *every* instance. Each one then walks the same GCS-backed -results directory and processes the same backlog: five instances were observed -starting identical passes over the same 22k directories. - -That is worse than wasted work. Both caches are rewritten whole, and a GCS -object write replaces the object rather than merging, so when two instances -finish a batch the later write silently discards the earlier one's rows. The -runs stay recorded in processed_dirs.json either way, so anything lost that way -is never reconsidered and never appears in trends again. - -The lease is a small file in the results directory holding a holder id and a -wall-clock timestamp. Acquiring writes it, waits out the window in which a -competing write could still land, then reads it back: because a whole-file -rewrite has exactly one winner and GCS reads are strongly consistent after it, -whoever reads its own id back is the sole holder. The holder renews while it -works so a long pass keeps the lease, and the timestamp lets a survivor take -over if the holder dies -- which matters here, since a large backlog can get the -precompute process SIGKILLed mid-pass. - -Wall-clock time is deliberate: monotonic clocks are not comparable across -instances, and the whole point is comparing timestamps written by other hosts. +Supervisord starts a precompute loop in every Cloud Run instance, but they share +one results directory and the caches are rewritten whole, so a later write +silently discards an earlier one's rows while processed_dirs.json still records +those runs as done -- they are then never reconsidered. + +Acquiring writes a holder id and timestamp, waits out the window in which a +competing write could still land, then reads back: a whole-file rewrite has one +winner and GCS reads are strongly consistent after it, so whoever reads its own +id back is the sole holder. Timestamps are wall-clock because they are compared +across hosts, where monotonic clocks are meaningless. """ import logging @@ -34,7 +22,7 @@ LEASE_FILENAME = "precompute.lease" # Long enough to ride out a slow FUSE write, short enough that a killed holder -# does not stall precompute for long. Renewal runs well inside it. +# does not stall precompute for long. LEASE_TTL_SECONDS = float(os.environ.get("PRECOMPUTE_LEASE_TTL", "120")) RENEW_INTERVAL_SECONDS = LEASE_TTL_SECONDS / 4 @@ -55,8 +43,7 @@ def _read(results_dir): holder, _, timestamp = f.read().partition("\n") return holder.strip(), float(timestamp.strip()) except (OSError, ValueError): - # A torn read of a file being rewritten lands here too; treating it as - # "no lease" is safe because the read-back check still has to pass. + # A torn read lands here too; safe, because read-back must still pass. return None, 0.0 @@ -92,9 +79,8 @@ def acquire(self): try: _write(self.results_dir) except OSError: - # Never let a lease problem stop precompute outright: a dashboard - # that silently stops updating is a worse failure than duplicated - # work, which is only what happened before this lease existed. + # Fail open: a dashboard that silently stops updating is worse than + # the duplicated work this lease exists to prevent. logging.exception("Could not write precompute lease; running unguarded") self.acquired = True return True @@ -118,9 +104,7 @@ def _renew_loop(self): while not self._stop.wait(RENEW_INTERVAL_SECONDS): holder, _ = _read(self.results_dir) if holder and holder != _HOLDER_ID: - # Someone judged us stale and took over. Two writers fighting - # over the file is the exact thing this exists to prevent, so - # stand down and re-acquire on the next pass. + # Stand down rather than fight the successor for the file. logging.warning("Precompute lease taken over by %s; stopping renewal", holder) return try: @@ -136,8 +120,7 @@ def release(self): if not self.acquired: return self.acquired = False - # Only clear the file if it is still ours; a successor's lease must not - # be deleted by the instance it replaced. + # Only clear the file if it is still ours, never a successor's. holder, _ = _read(self.results_dir) if holder == _HOLDER_ID: try: diff --git a/viewer/precompute_trends.py b/viewer/precompute_trends.py index ba67bce4..5d8b1492 100644 --- a/viewer/precompute_trends.py +++ b/viewer/precompute_trends.py @@ -16,10 +16,9 @@ # where the summarizer starts seeing 429s, so raising it further buys nothing. MAX_WORKERS = int(os.environ.get("PRECOMPUTE_WORKERS", 50)) -# ai_summary is ~96% of the bytes in trends_cache.csv, and that file is read in -# full on every page render. Only the recent window is kept inline; every -# summary is also written beside its own run, so the list view stays small -# without the older summaries being lost. +# ai_summary is ~96% of trends_cache.csv, which is read in full on every render. +# Only the recent window stays inline; every summary is also written beside its +# own run so nothing is lost. AI_SUMMARY_FILENAME = "ai_summary.txt" AI_SUMMARY_CACHE_DAYS = int(os.environ.get("AI_SUMMARY_CACHE_DAYS", 7)) @@ -52,8 +51,7 @@ def _is_recent(run_time, days=AI_SUMMARY_CACHE_DAYS): """Whether a run is inside the window that keeps its summary inline.""" ts = pd.to_datetime(run_time, errors='coerce') if pd.isna(ts): - # An unparseable run_time cannot be aged out safely, so treat it as old - # and let the sidecar serve it. + # Cannot be aged out safely, so treat as old and let the sidecar serve it. return False return ts >= pd.Timestamp.now() - pd.Timedelta(days=days) @@ -82,9 +80,8 @@ class UnparsableRun(Exception): """A run whose CSVs are present but structurally unusable. Distinct from "not ready yet" (no summary.csv), which returns None so a - later pass retries. A run that is missing a required column looks the same - on every future pass, so retrying it forever burns FUSE reads and floods - the log without ever making progress -- record it as processed instead. + later pass retries. A missing column looks the same on every pass, so such + a run is recorded as processed rather than retried forever. """ def __init__(self, job_id, reason): @@ -206,8 +203,6 @@ def get_metric_pct(row): except Exception as e: logging.error(f"Error generating AI summary for {d}: {e}") - # Written for every run regardless of age; the cache keeps only the - # recent window inline, and the detail view falls back to this file. write_ai_summary(run_dir, ai_summary) logging.info(f"Successfully processed directory: {d}") @@ -230,19 +225,15 @@ def get_metric_pct(row): 'ai_summary': ai_summary if _is_recent(run_time) else "", } except UnparsableRun: - # Must outrun the catch-all below: returning None here would leave the - # run unprocessed and queue it up to fail identically on every pass. + # Must precede the catch-all: returning None would requeue the run. raise except (pd.errors.ParserError, pd.errors.EmptyDataError, OSError) as e: - # A run caught mid-write and a transient FUSE read both land here, so - # the artifacts may well parse on a later pass. Leave it queued. + # Mid-write or a transient FUSE read; may well parse on a later pass. logging.warning(f"Deferring {d}, artifacts not readable yet: {e}") return None except Exception as e: - # Both artifacts exist and parsed, so anything left is a problem with - # their shape and will recur identically on every future pass. Returning - # None here is what jammed the queue: the run is retried forever and - # every pass re-reads it instead of draining the backlog. + # Both files parsed, so the fault is in their shape and will recur on + # every pass. Requeueing it is what jammed the backlog. logging.exception(f"Error reading data from {d}") raise UnparsableRun(d, f"{type(e).__name__}: {e}") from e @@ -402,7 +393,6 @@ def precompute(): try: res = future.result() except UnparsableRun as e: - # Will never parse, so record it and stop reconsidering it. logging.warning(f"Skipping {directory}: {e.reason}") batch_processed.append(e.job_id) continue diff --git a/viewer/run_index.py b/viewer/run_index.py index 39f14a74..02067662 100644 --- a/viewer/run_index.py +++ b/viewer/run_index.py @@ -1,19 +1,9 @@ """Cached listing of run directories under the results mount. -The results directory is a GCS FUSE mount holding tens of thousands of run -directories. The obvious way to enumerate it -- - - [d for d in os.listdir(p) if os.path.isdir(os.path.join(p, d))] - --- costs one FUSE stat round trip *per entry* on top of the listing itself, -because os.path.isdir cannot reuse anything os.listdir already learned. At -~25k entries that is ~25k round trips, and the Mesop render path ran that walk -three times per page, which pushed /__ui__ past the Cloud Run request timeout. - -os.scandir carries the directory-entry type through from the listing, so -entry.is_dir() is answered without a second trip. The short TTL then collapses -the repeated walks within a single render (and across renders in a session) -down to one. +The mount is GCS FUSE holding tens of thousands of run directories, so +os.listdir + os.path.isdir costs a stat round trip per entry. os.scandir +carries the entry type through from the listing and avoids them; the TTL +collapses the repeated walks the render path makes. """ import logging @@ -30,10 +20,9 @@ def list_run_directories(results_dir, force=False): """Return the names of run directories directly under results_dir. - Results are cached for CACHE_TTL_SECONDS. A missing results_dir yields an - empty list. If a refresh fails but a previous listing for the same path is - still held, the stale listing is returned rather than an empty one -- a - transient FUSE error should not blank out the dashboard. + Cached for CACHE_TTL_SECONDS. On a failed refresh a previous listing for the + same path is returned rather than an empty one, so a transient FUSE error + does not blank the dashboard. """ now = time.monotonic() @@ -53,8 +42,7 @@ def list_run_directories(results_dir, force=False): if entry.is_dir(follow_symlinks=False): dirs.append(entry.name) except OSError: - # Entry vanished mid-walk, or FUSE hiccuped on this one - # name. Skipping it beats failing the whole listing. + # Skipping one bad entry beats failing the whole listing. continue except FileNotFoundError: dirs = [] diff --git a/viewer/run_precompute.py b/viewer/run_precompute.py index 64e2ff4c..a99897f1 100644 --- a/viewer/run_precompute.py +++ b/viewer/run_precompute.py @@ -13,8 +13,7 @@ def main(): interval = int(os.environ.get("PRECOMPUTE_INTERVAL", 300)) results_dir = precompute_trends.get_results_dir() while True: - # Every Cloud Run instance runs this loop, but they all share one results - # directory, so only the lease holder may write to the caches. + # Every instance runs this loop, but only the lease holder may write. with PrecomputeLease(results_dir) as lease: if lease: # Dataset quality runs first because it is cheap and always finishes. From 3105e16d0d374feda36a3c0e2df71dd8d0edb01e Mon Sep 17 00:00:00 2001 From: prernakakkar-google <158031829+prernakakkar-google@users.noreply.github.com> Date: Mon, 24 Aug 2026 19:23:17 +0530 Subject: [PATCH 3/6] Potential fix for pull request finding 'Module is imported with 'import' and 'import from'' Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com> --- viewer/main.py | 1 - 1 file changed, 1 deletion(-) diff --git a/viewer/main.py b/viewer/main.py index 061b539a..c1ee4f2a 100644 --- a/viewer/main.py +++ b/viewer/main.py @@ -7,7 +7,6 @@ import subprocess import precompute_trends import dataset_quality -import run_index from run_index import list_run_directories from summarizer import summarize_eval_scoring from ai_comparer import compare_evals From 0e5684defa48cc502c158803650817ad266d47e7 Mon Sep 17 00:00:00 2001 From: Prerna Kakkar Date: Mon, 24 Aug 2026 13:55:49 +0000 Subject: [PATCH 4/6] fix --- viewer/version.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/viewer/version.txt b/viewer/version.txt index fde302eb..3cd76c90 100644 --- a/viewer/version.txt +++ b/viewer/version.txt @@ -1 +1 @@ -1480f53 +797b8c6 From ffdc270fbdd87aca3ab5a0351fc066e61f21e229 Mon Sep 17 00:00:00 2001 From: Prerna Kakkar Date: Mon, 24 Aug 2026 14:06:07 +0000 Subject: [PATCH 5/6] fix test --- viewer/precompute_trends.py | 73 ++++++++++++++++++++++++++----------- 1 file changed, 52 insertions(+), 21 deletions(-) diff --git a/viewer/precompute_trends.py b/viewer/precompute_trends.py index 5d8b1492..0690e374 100644 --- a/viewer/precompute_trends.py +++ b/viewer/precompute_trends.py @@ -1,4 +1,5 @@ import os +import csv import logging import json import argparse @@ -6,6 +7,8 @@ from run_index import list_run_directories +csv.field_size_limit(10**9) + logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') BATCH_SIZE = int(os.environ.get("PRECOMPUTE_BATCH_SIZE", 50)) @@ -29,12 +32,12 @@ def ai_summary_path(results_dir, job_id): def write_ai_summary(run_dir, summary): """Persist a summary beside its run so trimming the cache is not lossy.""" - if not summary or summary == "N/A": + if not isinstance(summary, str) or not summary or summary == "N/A": return try: with open(os.path.join(run_dir, AI_SUMMARY_FILENAME), "w") as f: f.write(summary) - except OSError: + except Exception: logging.warning("Could not write %s for %s", AI_SUMMARY_FILENAME, run_dir) @@ -294,6 +297,17 @@ def _append_rows(cache_file, rows): new_df[header].to_csv(cache_file, mode="a", header=False, index=False) +class _NothingToTrim(Exception): + """The cache has no usable ai_summary column; leave it untouched.""" + + +def _discard(path): + try: + os.remove(path) + except OSError: + pass + + def _trim_stale_summaries(results_dir, cache_file): """Blank ai_summary on cache rows that have aged out of the inline window. @@ -301,37 +315,54 @@ def _trim_stale_summaries(results_dir, cache_file): their own, so this runs every pass rather than only at write time. Each summary is written to its sidecar first where one is missing -- without that backfill this would destroy every summary already in the cache. + + Streamed a row at a time rather than loaded into a frame: the cache this + exists to shrink is exactly the one too large to hold in memory. The file + is only replaced if something was actually trimmed, so a cache already + inside the window costs one sequential read and no write. """ if not os.path.exists(cache_file) or os.path.getsize(cache_file) == 0: return + required = {'ai_summary', 'run_time', 'job_id'} + temp_file = cache_file + ".trim" + trimmed = 0 + backfilled = 0 + try: - df = pd.read_csv(cache_file) - except Exception: - logging.exception("Could not read %s to trim summaries", cache_file) + with open(cache_file, newline="") as src, open(temp_file, "w", newline="") as dst: + reader = csv.DictReader(src) + if not reader.fieldnames or not required <= set(reader.fieldnames): + raise _NothingToTrim + writer = csv.DictWriter(dst, fieldnames=reader.fieldnames) + writer.writeheader() + for row in reader: + if row.get('ai_summary') and not _is_recent(row.get('run_time')): + job_id = row.get('job_id') or "" + if job_id and not os.path.exists(ai_summary_path(results_dir, job_id)): + write_ai_summary( + os.path.join(results_dir, job_id), row['ai_summary'] + ) + backfilled += 1 + row['ai_summary'] = "" + trimmed += 1 + writer.writerow(row) + except _NothingToTrim: + _discard(temp_file) return - - if not {'ai_summary', 'run_time', 'job_id'} <= set(df.columns): + except Exception: + logging.exception("Could not trim summaries from %s", cache_file) + _discard(temp_file) return - stale = ( - df['ai_summary'].fillna("").astype(str).str.len().gt(0) - & ~df['run_time'].map(_is_recent) - ) - if not stale.any(): + if not trimmed: + _discard(temp_file) return - backfilled = 0 - for row in df.loc[stale, ['job_id', 'ai_summary']].itertuples(index=False): - if not os.path.exists(ai_summary_path(results_dir, row.job_id)): - write_ai_summary(os.path.join(results_dir, row.job_id), row.ai_summary) - backfilled += 1 - - df.loc[stale, 'ai_summary'] = "" - df.to_csv(cache_file, index=False) + os.replace(temp_file, cache_file) logging.info( "Trimmed %d aged-out summaries from the trends cache (%d sidecars backfilled)", - int(stale.sum()), backfilled, + trimmed, backfilled, ) From 880d91bddf55a0e54ae59ea87d19b240290e3b0a Mon Sep 17 00:00:00 2001 From: prernakakkar-google <158031829+prernakakkar-google@users.noreply.github.com> Date: Mon, 24 Aug 2026 20:28:56 +0530 Subject: [PATCH 6/6] Potential fix for pull request finding 'Empty except' Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com> --- viewer/precompute_trends.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/viewer/precompute_trends.py b/viewer/precompute_trends.py index 0690e374..0d175e20 100644 --- a/viewer/precompute_trends.py +++ b/viewer/precompute_trends.py @@ -304,8 +304,8 @@ class _NothingToTrim(Exception): def _discard(path): try: os.remove(path) - except OSError: - pass + except OSError as e: + logging.debug("Best-effort discard failed for %s: %s", path, e) def _trim_stale_summaries(results_dir, cache_file):