from agi_web import python_ui as st import json import os import threading import time from pathlib import Path import sys # Ensure project root is in path for imports sys.path.insert(0, str(Path(__file__).resolve().parent)) import benchmark import free_threading_core as core # Lock to serialize benchmark requests if "benchmark_lock" not in st.session_state: st.session_state.benchmark_lock = threading.Lock() if "results" not in st.session_state: st.session_state.results = None if "is_running" not in st.session_state: st.session_state.is_running = False if "last_params" not in st.session_state: st.session_state.last_params = None st.title("Free-threading lab") st.markdown(""" Compare CPU scaling of Mandelbrot rendering using the AGILAB worker pool engine on a free-threaded Python build. **Scope:** Local CPU scaling validation. This proves the included AGILAB pool engine behavior, not full AGILAB stack compatibility on free-threaded Python. **Modes:** 1. **GIL-on threads**: Forced thread pool, GIL enabled (control). 2. **GIL-off threads**: Auto backend (threads on free-threaded), GIL disabled. 3. **GIL-on processes**: Forced process pool, GIL enabled. All modes use the *same* free-threaded interpreter build. """) # Sidebar controls with st.sidebar: st.header("Controls") workload = st.selectbox("Workload", ["Small", "Medium"], index=0) if workload == "Small": width, height, iterations = 96, 64, 100 else: width, height, iterations = 192, 128, 160 # Determine max workers eff_cpus = benchmark.effective_cpus()["effective_cpus"] max_workers = min(eff_cpus, 8) if max_workers < 2: st.warning("Scaling unavailable: Only 1 effective CPU detected.") workers = 1 else: workers = st.slider("Workers (N)", min_value=2, max_value=max_workers, value=min(4, max_workers)) repeats = st.slider("Repeats", min_value=1, max_value=3, value=2) run_button = st.button("Run analysis", use_container_width=True) # Display deterministic preview on first load if st.session_state.results is None: st.subheader("Deterministic Preview") # Generate a small preview image preview_w, preview_h, preview_iter = 96, 64, 50 pixels = core.reference_image(preview_w, preview_h, preview_iter) # Convert to grayscale image data import numpy as np img_array = np.array(pixels).reshape(preview_h, preview_w) # Normalize to 0-255 img_array = (img_array - img_array.min()) / (img_array.max() - img_array.min() + 1e-9) * 255 img_array = img_array.astype(np.uint8) st.image(img_array, caption="Mandelbrot Preview (Small)", use_container_width=True) st.info("Click 'Run analysis' to start the benchmark.") # Handle Run analysis if run_button: if st.session_state.is_running: st.warning("Benchmark already running. Please wait.") else: st.session_state.is_running = True st.session_state.last_params = { "width": width, "height": height, "iterations": iterations, "workers": workers, "repeats": repeats } status_container = st.status("Running benchmark...") progress_bar = st.progress(0) try: # Acquire lock to serialize requests acquired = st.session_state.benchmark_lock.acquire(blocking=False) if not acquired: status_container.error("Another benchmark is currently running. Please try again later.") st.session_state.is_running = False else: try: with status_container: st.write("Starting benchmark...") st.write(f"Params: {width}x{height}, iter={iterations}, workers={workers}, repeats={repeats}") # Run benchmark result = benchmark.run_benchmark( width=width, height=height, iterations=iterations, workers=workers, repeats=repeats ) st.session_state.results = result st.session_state.is_running = False status_container.update(label="Benchmark completed!", state="complete") progress_bar.progress(100) finally: st.session_state.benchmark_lock.release() except Exception as e: st.session_state.is_running = False status_container.error(f"Benchmark failed: {str(e)}") import traceback st.code(traceback.format_exc()) # Display results if st.session_state.results: results = st.session_state.results # Check if params changed current_params = { "width": width, "height": height, "iterations": iterations, "workers": workers, "repeats": repeats } if st.session_state.last_params != current_params: st.warning("⚠️ Controls changed. These results are from a previous run.") st.header("Results") # Metrics col1, col2, col3 = st.columns(3) with col1: st.metric("Python Build", results["python_build"].split()[0]) with col2: st.metric("Effective CPUs", results["hardware"]["effective_cpus"]) with col3: st.metric("Same Work Verified", "Yes" if results["same_work_verified"] else "No") # Summary Table st.subheader("Summary") import pandas as pd df_summary = pd.DataFrame(results["summary"]) st.dataframe(df_summary, use_container_width=True) # Chart st.subheader("Timing Comparison") import plotly.express as px df_chart = df_summary.copy() fig = px.bar(df_chart, x="mode", y="wall_seconds", color="role", barmode="group", title="Median Wall Time by Mode and Role") st.plotly_chart(fig, use_container_width=True) # Worker Timeline (Expandable) with st.expander("Recorded Worker Timeline (Sample)"): st.markdown("*Note: This shows recorded task activity from the last run, not live monitoring.*") # Show a sample of records from the first parallel run parallel_runs = [r for r in results["runs"] if r["role"] == "parallel"] if parallel_runs: sample_run = parallel_runs[0] records = sample_run["records"] # Create a simple timeline visualization import plotly.graph_objects as go fig_timeline = go.Figure() # Group by thread/pid identities = {} for rec in records: key = f"{rec['pid']}:{rec['thread_id']}" if key not in identities: identities[key] = [] identities[key].append(rec) for key, recs in identities.items(): fig_timeline.add_trace(go.Scatter( x=[r["start"] for r in recs] + [r["end"] for r in recs], y=[key] * len(recs) + [key] * len(recs), mode="lines", name=key )) fig_timeline.update_layout( title="Worker Activity Timeline", xaxis_title="Time (s)", yaxis_title="Worker (PID:ThreadID)" ) st.plotly_chart(fig_timeline, use_container_width=True) # Hardware/Methodology with st.expander("Hardware & Methodology"): st.json(results["hardware"]) st.markdown(""" **Methodology:** - Mandelbrot escape counts computed in pure Python. - Image tiled into 2-4 row chunks, interleaved for load balancing. - AGILAB pool engine used for dispatch. - GIL state verified before/after execution. - Digests verified identical across all modes. - Medians used for speedup calculations. """) # Download JSON st.download_button( label="Download JSON Evidence", data=json.dumps(results, indent=2), file_name="free_threading_results.json", mime="application/json" )