feat: add --timing flag for phase-level performance breakdown (#2928)

* feat: add --timing flag for phase-level performance breakdown

Add PhaseTimer utility for instrumenting Slither phases:
- New --timing CLI flag to enable timing collection
- Instruments parse_contracts, analyze_contracts, convert_to_slithir,
  compute_dependency, compute_storage_layout, and detector runs
- Outputs sorted timing report showing total/count/avg per phase

Co-Authored-By: Claude Opus 4.5 <[email protected]>

* feat: add timing infrastructure and fix benchmark compatibility

Add phase-level timing support:
- New --timing flag for slither CLI to capture execution phases
- Timing data exported via --json-types timing
- Benchmark script updated to display phase breakdowns

Replace aave-v3-core with safe-smart-account in benchmarks:
- aave-v3-core requires Node 14/16/18, incompatible with Node 22+
- safe-smart-account uses modern Node.js (22.14.0 per .nvmrc)

Co-Authored-By: Claude Opus 4.5 <[email protected]>

---------

Co-authored-by: Claude Opus 4.5 <[email protected]>
This commit is contained in:
Dan Guido
2026-01-19 17:14:48 -05:00
committed by GitHub
co-authored by Claude Opus 4.5
parent f05f5478e8
commit 85c6711ec7
7 changed files with 501 additions and 41 deletions
+180 -35
View File
@@ -24,6 +24,9 @@ Usage:
# Compare against baseline and post to PR
python scripts/benchmark_perf.py --slither-path /path/to/worktree \\
--baseline /tmp/baseline.json --pr 123
# Include phase-level timing breakdown
python scripts/benchmark_perf.py --no-branch-compare --timing
"""
import argparse
@@ -32,9 +35,11 @@ import json
import os
import subprocess
import sys
import tempfile
import time
from contextlib import contextmanager
from pathlib import Path
from typing import NamedTuple
BENCH_DIR = Path("/tmp/slither-bench")
LOCK_FILE = Path("/tmp/slither-bench.lock")
@@ -53,13 +58,13 @@ PROJECTS = {
"url": "https://github.com/compound-finance/comet",
"setup": ["yarn install"],
},
"aave-v3-core": {
"url": "https://github.com/aave/aave-v3-core",
"setup": ["npm install"],
"safe-smart-account": {
"url": "https://github.com/safe-global/safe-smart-account",
"setup": ["npm install", "npx hardhat compile"],
},
"lido-dao": {
"url": "https://github.com/lidofinance/lido-dao",
"setup": ["yarn install"],
"setup": ["yarn install", "forge build --build-info"],
},
"optimism": {
"url": "https://github.com/ethereum-optimism/optimism",
@@ -123,20 +128,59 @@ def setup_projects(project_names: list[str]) -> None:
subprocess.run(cmd, shell=True, cwd=work_dir, check=True)
def run_slither(project_path: str, slither_cwd: Path | None = None) -> float:
"""Run slither and return elapsed time.
class BenchmarkResult(NamedTuple):
"""Result from a single slither run."""
elapsed: float
timing: dict[str, dict] | None = None
def run_slither(
project_path: str,
slither_cwd: Path | None = None,
capture_timing: bool = False,
) -> BenchmarkResult:
"""Run slither and return elapsed time and optional timing data.
Args:
project_path: Path to the project to analyze.
slither_cwd: Directory to run slither from (for worktree support).
capture_timing: If True, capture phase-level timing data.
Returns:
BenchmarkResult with elapsed time and optional timing breakdown.
"""
start = time.perf_counter()
subprocess.run(
["uv", "run", "slither", project_path],
capture_output=True,
cwd=slither_cwd,
)
return time.perf_counter() - start
timing_data = None
cmd = ["uv", "run", "slither", project_path]
if capture_timing:
# Use temp file to capture JSON with timing data
# Create path but don't keep the file (Slither won't overwrite existing files)
with tempfile.NamedTemporaryFile(suffix=".json", delete=True) as f:
json_path = f.name
# File is deleted when context exits, so Slither can create it fresh
cmd.extend(["--json", json_path, "--json-types", "timing", "--timing"])
start = time.perf_counter()
subprocess.run(cmd, capture_output=True, cwd=slither_cwd)
elapsed = time.perf_counter() - start
# Extract timing data from JSON
try:
with open(json_path) as f:
data = json.load(f)
timing_data = data.get("results", {}).get("timing")
except (json.JSONDecodeError, FileNotFoundError):
pass
finally:
Path(json_path).unlink(missing_ok=True)
else:
start = time.perf_counter()
subprocess.run(cmd, capture_output=True, cwd=slither_cwd)
elapsed = time.perf_counter() - start
return BenchmarkResult(elapsed=elapsed, timing=timing_data)
def get_project_path(name: str) -> str:
@@ -148,17 +192,29 @@ def get_project_path(name: str) -> str:
return str(base)
class ProjectResult(NamedTuple):
"""Result for a single project benchmark."""
elapsed: float
timing: dict[str, dict] | None = None
def benchmark_current(
project_names: list[str],
quiet: bool = False,
slither_cwd: Path | None = None,
) -> dict[str, float]:
capture_timing: bool = False,
) -> dict[str, ProjectResult]:
"""Benchmark current checkout against specified projects.
Args:
project_names: List of project names to benchmark.
quiet: Suppress output if True.
slither_cwd: Directory to run slither from (for worktree support).
capture_timing: If True, capture phase-level timing data.
Returns:
Dictionary mapping project name to ProjectResult.
"""
results = {}
for name in project_names:
@@ -168,11 +224,13 @@ def benchmark_current(
print(f" {name}: SKIPPED (not found, run --setup first)")
continue
times = [run_slither(path, slither_cwd) for _ in range(ITERATIONS)]
avg = sum(times) / len(times)
results[name] = round(avg, 1)
runs = [run_slither(path, slither_cwd, capture_timing) for _ in range(ITERATIONS)]
avg_elapsed = sum(r.elapsed for r in runs) / len(runs)
# Use timing from last run (they should be similar)
timing = runs[-1].timing if capture_timing else None
results[name] = ProjectResult(elapsed=round(avg_elapsed, 1), timing=timing)
if not quiet:
print(f" {name}: {avg:.1f}s")
print(f" {name}: {avg_elapsed:.1f}s")
return results
@@ -193,14 +251,39 @@ def benchmark_branches(project_names: list[str], branches: list[str], quiet: boo
return results
def summarize_timing(timing: dict[str, dict]) -> dict[str, float]:
"""Summarize timing data into core phases and detector totals.
Args:
timing: Raw timing data from slither.
Returns:
Dictionary with summarized timing (core phases + detector_total).
"""
core_phases = {}
detector_total = 0.0
for phase, data in timing.items():
total_sec = data.get("total_sec", 0)
if phase.startswith("detector:"):
detector_total += total_sec
else:
core_phases[phase] = total_sec
return {**core_phases, "detectors_total": round(detector_total, 3)}
def format_results_markdown(
results: dict[str, float], baseline: dict[str, float] | None = None
results: dict[str, ProjectResult],
baseline: dict | None = None,
show_timing: bool = False,
) -> str:
"""Format benchmark results as markdown table.
Args:
results: Benchmark results {project_name: time_seconds}.
results: Benchmark results {project_name: ProjectResult}.
baseline: Optional baseline results for comparison.
show_timing: If True, include phase timing breakdown.
Returns:
Markdown-formatted table string.
@@ -211,17 +294,34 @@ def format_results_markdown(
"| Project | Time | vs Baseline |",
"|---------|------|-------------|",
]
for proj, time_s in results.items():
for proj, result in results.items():
time_s = result.elapsed
if baseline and proj in baseline:
base_time = baseline[proj]
base_data = baseline[proj]
base_time = (
base_data.get("elapsed", base_data) if isinstance(base_data, dict) else base_data
)
delta = ((base_time - time_s) / base_time) * 100 if base_time else 0
lines.append(f"| {proj} | {time_s}s | {delta:+.1f}% |")
else:
lines.append(f"| {proj} | {time_s}s | - |")
if show_timing:
lines.append("")
lines.append("### Phase Breakdown")
lines.append("")
for proj, result in results.items():
if result.timing:
summary = summarize_timing(result.timing)
lines.append(f"**{proj}:**")
for phase, secs in sorted(summary.items(), key=lambda x: -x[1]):
lines.append(f"- {phase}: {secs:.2f}s")
lines.append("")
return "\n".join(lines)
def load_baseline(baseline_path: Path) -> dict[str, float]:
def load_baseline(baseline_path: Path) -> dict:
"""Load baseline results from JSON file.
Args:
@@ -235,16 +335,20 @@ def load_baseline(baseline_path: Path) -> dict[str, float]:
def print_pr_comment_command(
pr_number: int, results: dict[str, float], baseline: dict[str, float] | None = None
pr_number: int,
results: dict[str, ProjectResult],
baseline: dict | None = None,
show_timing: bool = False,
) -> None:
"""Print the gh command to post results as PR comment (no auto-posting).
Args:
pr_number: GitHub PR number.
results: Benchmark results {project_name: time_seconds}.
results: Benchmark results {project_name: ProjectResult}.
baseline: Optional baseline results for comparison.
show_timing: If True, include phase timing in output.
"""
body = format_results_markdown(results, baseline)
body = format_results_markdown(results, baseline, show_timing)
# Print markdown preview
print("=== PR Comment Preview ===")
print(body)
@@ -262,16 +366,43 @@ def print_comparison(results: dict) -> None:
for branch, data in results.items():
if branch == "master":
print(f"{branch}: {data}")
formatted = {k: f"{v.elapsed}s" for k, v in data.items()}
print(f"{branch}: {formatted}")
else:
changes = {}
for proj, t in data.items():
base = baseline.get(proj, t)
pct = ((base - t) / base) * 100 if base else 0
changes[proj] = f"{t}s ({pct:+.1f}%)"
for proj, result in data.items():
base_result = baseline.get(proj)
base_time = base_result.elapsed if base_result else result.elapsed
pct = ((base_time - result.elapsed) / base_time) * 100 if base_time else 0
changes[proj] = f"{result.elapsed}s ({pct:+.1f}%)"
print(f"{branch}: {changes}")
def print_timing_breakdown(results: dict[str, ProjectResult]) -> None:
"""Print detailed timing breakdown for each project."""
print("\n=== Phase Timing Breakdown ===")
for proj, result in results.items():
if not result.timing:
continue
summary = summarize_timing(result.timing)
total = sum(summary.values())
print(f"\n{proj} ({result.elapsed}s total):")
for phase, secs in sorted(summary.items(), key=lambda x: -x[1]):
pct = (secs / total * 100) if total else 0
print(f" {phase}: {secs:.2f}s ({pct:.0f}%)")
def results_to_json(results: dict[str, ProjectResult]) -> dict:
"""Convert ProjectResult dict to JSON-serializable format."""
return {
proj: {
"elapsed": result.elapsed,
"timing": summarize_timing(result.timing) if result.timing else None,
}
for proj, result in results.items()
}
def main():
parser = argparse.ArgumentParser(
description="Benchmark Slither performance against smart contract projects",
@@ -324,6 +455,11 @@ def main():
action="store_true",
help="Skip lock acquisition (for isolated environments)",
)
parser.add_argument(
"--timing",
action="store_true",
help="Capture and display phase-level timing breakdown",
)
args = parser.parse_args()
@@ -373,7 +509,12 @@ def main():
if args.no_branch_compare or args.slither_path:
if not args.json:
print("=== Benchmark Results ===")
return benchmark_current(project_names, quiet=args.json, slither_cwd=slither_cwd)
return benchmark_current(
project_names,
quiet=args.json,
slither_cwd=slither_cwd,
capture_timing=args.timing,
)
return benchmark_branches(project_names, branches, quiet=args.json)
if args.no_lock:
@@ -384,15 +525,19 @@ def main():
# Print PR comment command if requested (no auto-posting)
if args.pr:
print_pr_comment_command(args.pr, results, baseline)
print_pr_comment_command(args.pr, results, baseline, show_timing=args.timing)
return # Don't print JSON when using --pr (output already includes preview)
# Output results
if args.json:
print(json.dumps(results, indent=2))
json_output = results_to_json(results)
print(json.dumps(json_output, indent=2))
elif not args.no_branch_compare and not args.slither_path:
print_comparison(results)
print("\n" + json.dumps(results, indent=2))
json_output = {branch: results_to_json(data) for branch, data in results.items()}
print("\n" + json.dumps(json_output, indent=2))
elif args.timing:
print_timing_breakdown(results)
if __name__ == "__main__":
+28
View File
@@ -649,6 +649,13 @@ def parse_args(
default=False,
)
parser.add_argument(
"--timing",
help="Print phase-level timing breakdown",
action="store_true",
default=False,
)
# Disable the throw/catch on partial analyses
parser.add_argument(
"--disallow-partial", help=argparse.SUPPRESS, action="store_true", default=False
@@ -777,6 +784,14 @@ def main_impl(
cp = cProfile.Profile()
cp.enable()
# Enable phase timing if requested
if args.timing:
from slither.utils.timing import PhaseTimer
timer = PhaseTimer.get()
timer.reset() # Clear accumulated data from previous runs
timer.enabled = True
# Set colorization option
set_colorization_enabled(False if args.disable_color else sys.stdout.isatty())
@@ -894,6 +909,12 @@ def main_impl(
_, printers = get_detectors_and_printers()
json_results["list-printers"] = output_printers_json(printers)
# Add timing data to JSON if requested
if "timing" in args.json_types and args.timing:
from slither.utils.timing import PhaseTimer
json_results["timing"] = PhaseTimer.get().report()
# Output our results to markdown if we wish to compile a checklist.
if args.checklist:
output_results_to_markdown(
@@ -945,6 +966,13 @@ def main_impl(
stats = pstats.Stats(cp).sort_stats("cumtime")
stats.print_stats()
if args.timing:
from slither.utils.timing import PhaseTimer
# Skip text output when JSON goes to stdout to avoid corruption
if not outputting_json_stdout:
print("\n" + PhaseTimer.get().report_text())
fail_on = FailOnLevel(args.fail_on)
if fail_on == FailOnLevel.HIGH:
fail_on_detection = any(result["impact"] == "High" for result in results_detectors)
+15 -3
View File
@@ -197,9 +197,14 @@ class Slither(SlitherCore):
self._init_parsing_and_analyses(kwargs.get("skip_analyze", False))
def _init_parsing_and_analyses(self, skip_analyze: bool) -> None:
from slither.utils.timing import PhaseTimer
timer = PhaseTimer.get()
for parser in self._parsers:
try:
parser.parse_contracts()
with timer.phase("parse_contracts"):
parser.parse_contracts()
except Exception as e:
if self.no_fail:
continue
@@ -209,7 +214,8 @@ class Slither(SlitherCore):
if not skip_analyze:
for parser in self._parsers:
try:
parser.analyze_contracts()
with timer.phase("analyze_contracts"):
parser.analyze_contracts()
except Exception as e:
if self.no_fail:
continue
@@ -282,9 +288,15 @@ class Slither(SlitherCore):
"""
:return: List of registered detectors results.
"""
from slither.utils.timing import PhaseTimer
timer = PhaseTimer.get()
self.load_previous_results()
results = [d.detect() for d in self._detectors]
results = []
for d in self._detectors:
with timer.phase(f"detector:{d.ARGUMENT}"):
results.append(d.detect())
self.write_results_to_hide()
return results
@@ -582,12 +582,23 @@ class SlitherCompilationUnitSolc(CallerContextExpression):
self._parsed = True
def analyze_contracts(self) -> None:
from slither.utils.timing import PhaseTimer
if not self._parsed:
raise SlitherException("Parse the contract before running analyses")
self._convert_to_slithir()
timer = PhaseTimer.get()
with timer.phase("convert_to_slithir"):
self._convert_to_slithir()
if not self._compilation_unit.core.skip_data_dependency:
compute_dependency(self._compilation_unit)
self._compilation_unit.compute_storage_layout()
with timer.phase("compute_dependency"):
compute_dependency(self._compilation_unit)
with timer.phase("compute_storage_layout"):
self._compilation_unit.compute_storage_layout()
self._analyzed = True
def _analyze_all_enums(self, contracts_to_be_analyzed: list[ContractSolc]) -> None:
+1
View File
@@ -25,6 +25,7 @@ JSON_OUTPUT_TYPES = [
"printers",
"list-detectors",
"list-printers",
"timing",
]
+70
View File
@@ -0,0 +1,70 @@
"""Fine-grained timing instrumentation for Slither phases."""
import json
import threading
import time
from collections import defaultdict
from collections.abc import Generator
from contextlib import contextmanager
class PhaseTimer:
"""Singleton for collecting timing data across Slither phases."""
_instance: "PhaseTimer | None" = None
_lock: threading.Lock = threading.Lock()
def __init__(self) -> None:
self.timings: dict[str, list[float]] = defaultdict(list)
self.enabled = False
@classmethod
def get(cls) -> "PhaseTimer":
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = PhaseTimer()
return cls._instance
@contextmanager
def phase(self, name: str) -> Generator[None, None, None]:
"""Context manager to time a named phase."""
if not self.enabled:
yield
return
start = time.perf_counter()
try:
yield
finally:
self.timings[name].append(time.perf_counter() - start)
def report(self) -> dict[str, dict]:
"""Return timing statistics sorted by total time."""
return {
name: {
"count": len(times),
"total_sec": round(sum(times), 3),
"avg_sec": round(sum(times) / len(times), 3) if times else 0,
}
for name, times in sorted(self.timings.items(), key=lambda x: sum(x[1]), reverse=True)
}
def report_json(self) -> str:
return json.dumps(self.report(), indent=2)
def report_text(self) -> str:
"""Return human-readable timing report."""
lines = ["Phase Timing Report:"]
for name, stats in self.report().items():
total = stats["total_sec"]
count = stats["count"]
avg = stats["avg_sec"]
if count == 1:
lines.append(f" {name}: {total:.3f}s")
else:
lines.append(f" {name}: {total:.3f}s ({count} calls, avg {avg:.3f}s)")
return "\n".join(lines)
def reset(self) -> None:
"""Clear all timing data."""
self.timings.clear()
+193
View File
@@ -0,0 +1,193 @@
"""Unit tests for slither.utils.timing module."""
import json
import threading
import time
import pytest
from slither.utils.timing import PhaseTimer
@pytest.fixture(autouse=True)
def reset_singleton():
"""Reset the PhaseTimer singleton before and after each test."""
PhaseTimer._instance = None
yield
PhaseTimer._instance = None
def test_phase_timer_disabled_by_default():
"""Timer should be disabled by default."""
timer = PhaseTimer.get()
assert timer.enabled is False
def test_phase_timer_no_collection_when_disabled():
"""Timer should not collect timing data when disabled."""
timer = PhaseTimer.get()
assert timer.enabled is False
with timer.phase("test_phase"):
time.sleep(0.01)
assert len(timer.timings) == 0
def test_phase_timer_collects_when_enabled():
"""Timer should collect timing data when enabled."""
timer = PhaseTimer.get()
timer.enabled = True
with timer.phase("test_phase"):
time.sleep(0.01)
assert "test_phase" in timer.timings
assert len(timer.timings["test_phase"]) == 1
assert timer.timings["test_phase"][0] >= 0.01
def test_phase_timer_multiple_calls():
"""Timer should accumulate multiple calls to the same phase."""
timer = PhaseTimer.get()
timer.enabled = True
for _ in range(3):
with timer.phase("repeated_phase"):
time.sleep(0.001)
assert "repeated_phase" in timer.timings
assert len(timer.timings["repeated_phase"]) == 3
def test_phase_timer_report_structure():
"""Report should have correct structure with count, total_sec, and avg_sec."""
timer = PhaseTimer.get()
timer.enabled = True
with timer.phase("phase_a"):
time.sleep(0.01)
with timer.phase("phase_b"):
time.sleep(0.02)
report = timer.report()
assert "phase_a" in report
assert "phase_b" in report
for phase_name in ["phase_a", "phase_b"]:
assert "count" in report[phase_name]
assert "total_sec" in report[phase_name]
assert "avg_sec" in report[phase_name]
assert report[phase_name]["count"] == 1
def test_phase_timer_report_text_format():
"""Report text should be human-readable with proper formatting."""
timer = PhaseTimer.get()
timer.enabled = True
with timer.phase("my_phase"):
time.sleep(0.01)
text = timer.report_text()
assert "Phase Timing Report:" in text
assert "my_phase" in text
assert "s" in text # Time suffix
def test_phase_timer_reset():
"""Reset should clear all timing data."""
timer = PhaseTimer.get()
timer.enabled = True
with timer.phase("phase_to_clear"):
time.sleep(0.001)
assert len(timer.timings) == 1
timer.reset()
assert len(timer.timings) == 0
def test_phase_timer_singleton():
"""PhaseTimer.get() should always return the same instance."""
timer1 = PhaseTimer.get()
timer2 = PhaseTimer.get()
assert timer1 is timer2
def test_phase_timer_singleton_thread_safe():
"""Singleton should be thread-safe under concurrent access."""
instances = []
errors = []
def get_instance():
try:
instance = PhaseTimer.get()
instances.append(instance)
except Exception as e:
errors.append(e)
threads = [threading.Thread(target=get_instance) for _ in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
assert len(errors) == 0
assert len(instances) == 10
# All instances should be the same object
assert all(inst is instances[0] for inst in instances)
def test_phase_timer_report_json():
"""report_json() should return valid JSON string."""
timer = PhaseTimer.get()
timer.enabled = True
with timer.phase("json_test"):
time.sleep(0.001)
json_str = timer.report_json()
parsed = json.loads(json_str)
assert "json_test" in parsed
assert parsed["json_test"]["count"] == 1
def test_phase_timer_report_sorted_by_total_time():
"""Report should be sorted by total time in descending order."""
timer = PhaseTimer.get()
timer.enabled = True
with timer.phase("short"):
time.sleep(0.001)
with timer.phase("long"):
time.sleep(0.02)
report = timer.report()
phases = list(report.keys())
# "long" should come before "short" since it has more total time
assert phases.index("long") < phases.index("short")
def test_phase_timer_multiple_phases_text():
"""Report text should show call count and average for multiple calls."""
timer = PhaseTimer.get()
timer.enabled = True
for _ in range(3):
with timer.phase("multi_call"):
time.sleep(0.001)
text = timer.report_text()
assert "3 calls" in text
assert "avg" in text