Skip to content

perf(viewer): share one mtime-keyed cache for the trends CSV and dir listing #555

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
omkargaikwad23 wants to merge 1 commit into from
Open

perf(viewer): share one mtime-keyed cache for the trends CSV and dir listing #555

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
125 changes: 82 additions & 43 deletions viewer/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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:
Expand All @@ -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",
Expand Down Expand Up @@ -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,
Expand All @@ -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

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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]
Expand Down
27 changes: 7 additions & 20 deletions viewer/trends.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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":
Expand Down
Loading