From 6e437522723ecc7f2bc9691259dcb0c55571cb01 Mon Sep 17 00:00:00 2001 From: Omkar Gaikwad Date: Sun, 9 Aug 2026 14:45:43 +0000 Subject: [PATCH] perf(viewer): share one mtime-keyed cache for the trends CSV and dir listing trends_cache.csv was reparsed on every render by the Status tab, the Charts tab and the run detail view, and the results directory was relisted three times per page load. On the GCS FUSE mount each listing costs one stat per run directory, so both scaled with the number of runs and ran on interactions that needed neither. Add load_trends_df and list_run_dirs, keyed on mtime, and route every caller through them. load_summaries now builds on load_trends_df so the CSV is parsed once rather than twice. The frame is now shared, so the Charts filter chain uses assign() rather than an in-place column write, which would otherwise have welded product_dataset onto the cached object. --- viewer/main.py | 125 +++++++++++++++++++++++++++++++---------------- viewer/trends.py | 27 +++------- 2 files changed, 89 insertions(+), 63 deletions(-) diff --git a/viewer/main.py b/viewer/main.py index f58c750b..1345e28d 100644 --- a/viewer/main.py +++ b/viewer/main.py @@ -133,6 +133,70 @@ def get_results_dir(): return results_dir_candidates[1] # Fallback to default +# (mtime, value). Each is replaced as a whole so a concurrent reader never sees a +# value that disagrees with the mtime it was built from. +_TRENDS_DF_CACHE = (None, None) +_RUN_DIRS_CACHE = (None, []) + + +def _mtime(path): + try: + return os.path.getmtime(path) + except OSError: + return None + + +def load_trends_df(results_dir): + """Parsed trends_cache.csv, reread only when the file changes. + + Shared by every request in this worker, so callers must not mutate it. + """ + global _TRENDS_DF_CACHE + + cache_file = os.path.join(results_dir, "trends_cache.csv") + mtime = _mtime(cache_file) + if mtime is None: + logging.warning(f"Trends cache file not found at {cache_file}") + return None + + cached_mtime, df = _TRENDS_DF_CACHE + if cached_mtime == mtime: + return df + + try: + df = pd.read_csv(cache_file) + except Exception as e: + logging.error(f"Error reading trends cache: {e}") + return None + + logging.info(f"Loaded {len(df)} rows from trends cache.") + _TRENDS_DF_CACHE = (mtime, df) + return df + + +def list_run_dirs(results_dir): + """Run directory names, restatted only when results_dir gains or loses entries. + + Shared by every request in this worker, so callers must not mutate it. + """ + global _RUN_DIRS_CACHE + + mtime = _mtime(results_dir) + if mtime is None: + return [] + + cached_mtime, dirs = _RUN_DIRS_CACHE + if cached_mtime == mtime: + return dirs + + dirs = [ + d for d in os.listdir(results_dir) + if os.path.isdir(os.path.join(results_dir, d)) + ] + _RUN_DIRS_CACHE = (mtime, dirs) + return dirs + + def get_eval_details(results_dir, dir_name): details = { "product": "N/A", @@ -228,14 +292,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_dirs(results_dir) job_id = me.query_params.get("job_id") or me.query_params.get("jobid") if job_id and job_id in directories: @@ -258,14 +315,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_dirs(results_dir) + with me.box( style=me.Style( background="#ffffff", @@ -305,10 +356,9 @@ def on_agent_tab_change(e): # Build summary data from precomputed trends cache data = [] - cache_file = os.path.join(results_dir, "trends_cache.csv") - if os.path.exists(cache_file): + cache_df = load_trends_df(results_dir) + if cache_df is not None: try: - cache_df = pd.read_csv(cache_file) for _, row in cache_df.iterrows(): data.append({ 'AI Score': row['ai_score'] if 'ai_score' in row else None, @@ -326,7 +376,6 @@ def on_agent_tab_change(e): except Exception as e: logging.error(f"Error reading trends cache: {e}") else: - logging.warning(f"Trends cache file not found at {cache_file}") me.text("Trends cache file not found. Please run precompute.") return @@ -545,12 +594,9 @@ def _pct(value): return f"{value:.0f}%" if not pd.isna(value) else "N/A" -def _build_summaries(cache_file): +def _build_summaries(cache_df): import re - cache_df = pd.read_csv(cache_file) - logging.info(f"Loaded {len(cache_df)} rows from trends cache.") - rows = [] for _, row in cache_df.iterrows(): score = row['ai_score'] if 'ai_score' in row else 0.0 @@ -587,21 +633,22 @@ def load_summaries(results_dir): """ global _SUMMARIES_CACHE - cache_file = os.path.join(results_dir, "trends_cache.csv") - try: - mtime = os.path.getmtime(cache_file) - except OSError: - logging.warning(f"Trends cache file not found at {cache_file}") + mtime = _mtime(os.path.join(results_dir, "trends_cache.csv")) + if mtime is None: return [] cached_mtime, rows = _SUMMARIES_CACHE if cached_mtime == mtime: return rows + cache_df = load_trends_df(results_dir) + if cache_df is None: + return [] + try: - rows = _build_summaries(cache_file) + rows = _build_summaries(cache_df) except Exception as e: - logging.error(f"Error reading trends cache: {e}") + logging.error(f"Error building list rows from trends cache: {e}") return [] _SUMMARIES_CACHE = (mtime, rows) @@ -2074,15 +2121,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_dirs(results_dir) + def on_title_click(e: me.ClickEvent): state.selected_directory = "" state.conversation_index = 0 @@ -2308,10 +2348,9 @@ def get_val(cfg_name): me.text("AI Summary", type="headline-5") if not state.ai_summary and state.selected_directory: - trends_cache_file = os.path.join(results_dir, "trends_cache.csv") - if os.path.exists(trends_cache_file): + cache_df = load_trends_df(results_dir) + if cache_df is not None: try: - cache_df = pd.read_csv(trends_cache_file) run_data = cache_df[cache_df['job_id'] == state.selected_directory] if not run_data.empty and 'ai_summary' in run_data.columns: summary = run_data['ai_summary'].values[0] diff --git a/viewer/trends.py b/viewer/trends.py index a8c41706..74e86aba 100644 --- a/viewer/trends.py +++ b/viewer/trends.py @@ -2,7 +2,7 @@ import logging import mesop as me import pandas as pd -from main import State +from main import State, list_run_dirs, load_trends_df def get_results_dir(): # Try to read from environment variable @@ -90,26 +90,12 @@ def trends_component(): me.text(f"Results directory not found at {results_dir}") return - cache_file = os.path.join(results_dir, "trends_cache.csv") - - df = None - - # Try to load from cache - if os.path.exists(cache_file): - try: - df = pd.read_csv(cache_file) - logging.info("Loaded trends data from cache.") - except Exception as e: - logging.error(f"Error reading cache file: {e}") - + df = load_trends_df(results_dir) + # 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_dirs(results_dir) + data = [] for d in directories: @@ -333,7 +319,8 @@ def handler(e: me.ClickEvent): if state.trends_requester_filter: df = df[df['requester'] == state.trends_requester_filter] - df['product_dataset'] = df['product'] + " (" + df['dataset'] + ")" + # assign, not item-set: df may be the shared cached frame from load_trends_df. + df = df.assign(product_dataset=df['product'] + " (" + df['dataset'] + ")") df = df[df['product'].notna() & (df['product'] != 'unknown') & (df['product'].str.strip() != '')] if state.trends_agent_tab == "Gemini":