feat(sim): add CAM performance sweep test infrastructure

- hw/sim/tests/test_cam_perf.py: new cocotb perf test with bounded wait helpers
- scripts/sweep_cam_perf.py: sweep data model, matrix, and make command builders  
- tests/test_sweep_cam_perf.py: unit tests for sweep helpers
- tests/conftest.py: pytest path configuration for scripts package
- hw/sim/Makefile: deterministic noise params override for perf test compatibility
This commit is contained in:
2026-05-15 18:43:25 +08:00
parent 24b8750f0f
commit 2e0e36eea5
5 changed files with 1396 additions and 0 deletions

21
tests/conftest.py Normal file
View File

@@ -0,0 +1,21 @@
"""pytest configuration — ensure ``scripts/`` is importable via package syntax.
The ``scripts`` directory is a PEP420 implicit namespace package (no
``__init__.py``). Because ``uv run pytest`` does **not** add the project root
to ``sys.path``, ``from scripts.sweep_cam_perf import ...`` would otherwise
fail with ``ModuleNotFoundError``.
This conftest inserts the repository root at the front of ``sys.path`` so that
``scripts`` is discoverable as a top-level package.
"""
from __future__ import annotations
import sys
from pathlib import Path
_repo_root = Path(__file__).resolve().parent.parent
_repo_root_str = str(_repo_root)
if _repo_root_str not in sys.path:
sys.path.insert(0, _repo_root_str)

View File

@@ -0,0 +1,295 @@
#!/usr/bin/env python3
"""Fast unit tests for scripts/sweep_cam_perf.py (sweep helpers)."""
from __future__ import annotations
import subprocess
from pathlib import Path
from scripts.sweep_cam_perf import (
CSV_FIELDNAMES,
PerfRun,
SweepConfig,
_run_subprocess,
build_make_command,
default_sweep_configs,
logical_sweeps_for_config,
parse_perf_result,
plot_figures,
run_all,
)
# ---------------------------------------------------------------------------
# parse_perf_result
# ---------------------------------------------------------------------------
def test_parse_perf_result_extracts_metrics():
log = (
"PERF_RESULT latency_cycles=148 total_cycles=158 "
"accepted_queries=1 completed_queries=1 "
"queries_per_cycle=0.006329 status=pass"
)
result = parse_perf_result(log)
assert result == {
"latency_cycles": "148",
"total_cycles": "158",
"accepted_queries": "1",
"completed_queries": "1",
"queries_per_cycle": "0.006329",
"status": "pass",
}
def test_parse_perf_result_returns_empty_when_missing():
log = "some random simulation output without the marker"
assert parse_perf_result(log) == {}
# ---------------------------------------------------------------------------
# default_sweep_configs
# ---------------------------------------------------------------------------
def test_default_sweep_configs_are_paper_supporting_dense_matrix():
configs = default_sweep_configs()
assert len(configs) == 19
tuples = {(c.num_rows, c.hash_bits, c.lanes) for c in configs}
# NUM_ROWS sweep (lanes fixed at 16)
for nr in (64, 128, 192, 256, 384, 512, 768, 1024):
assert (nr, 512, 16) in tuples, f"Missing NUM_ROWS config {(nr, 512, 16)}"
# HASH_BITS sweep (num_rows fixed at 512, lanes fixed at 16)
for hb in (64, 128, 192, 256, 320, 384, 448, 512):
assert (512, hb, 16) in tuples, f"Missing HASH_BITS config {(512, hb, 16)}"
# LANES sweep (num_rows and hash_bits fixed at 512)
for la in (2, 4, 8, 16, 32):
assert (512, 512, la) in tuples, f"Missing LANES config {(512, 512, la)}"
assert len(tuples) == 19
# All configs must satisfy num_rows % lanes == 0
for c in configs:
assert c.num_rows % c.lanes == 0, (
f"num_rows={c.num_rows} not divisible by lanes={c.lanes} for {c}"
)
# ---------------------------------------------------------------------------
# logical_sweeps_for_config
# ---------------------------------------------------------------------------
def test_logical_sweeps_baseline_belongs_to_all_three_groups():
"""The baseline (512,512,16) config belongs to all three sweep groups,
even though its CSV sweep_name is 'num_rows' (first-occurrence dedup)."""
configs = default_sweep_configs()
# Locate the shared baseline config.
baseline = next(c for c in configs if (c.num_rows, c.hash_bits, c.lanes) == (512, 512, 16))
groups = logical_sweeps_for_config(baseline)
assert set(groups) == {"num_rows", "hash_bits", "lanes"}, (
f"Baseline should belong to all three groups, got {groups}"
)
assert len(configs) == 19, "Sweep config count must remain 19"
# ---------------------------------------------------------------------------
# build_make_command
# ---------------------------------------------------------------------------
def test_build_make_command_uses_perf_test_module():
cfg = SweepConfig("smoke", 128, 128, 4)
cmd = build_make_command(cfg)
assert cmd == [
"make",
"-C",
"hw/sim",
"NUM_ROWS=128",
"HASH_BITS=128",
"LANES=4",
"COCOTB_TEST_MODULES=tests.test_cam_perf",
]
# ---------------------------------------------------------------------------
# CSV_FIELDNAMES
# ---------------------------------------------------------------------------
def test_csv_fieldnames_match_spec():
assert CSV_FIELDNAMES == [
"sweep_name",
"num_rows",
"hash_bits",
"lanes",
"latency_cycles",
"total_cycles",
"accepted_queries",
"completed_queries",
"queries_per_cycle",
"sim_time_sec",
"status",
"log_path",
]
# ---------------------------------------------------------------------------
# _run_subprocess — timeout preserves partial output
# ---------------------------------------------------------------------------
def test_run_subprocess_timeout_preserves_partial_output(monkeypatch):
"""Timeout error string includes partial stdout/stderr from the process."""
def mock_run(*args, **kwargs):
raise subprocess.TimeoutExpired(
cmd=["mock"],
timeout=1,
output="partial stdout data",
stderr="partial stderr data",
)
monkeypatch.setattr(subprocess, "run", mock_run)
result, duration, error = _run_subprocess(
["mock"], cwd=Path("/tmp"), env={}, timeout=1,
)
assert result is None
assert duration >= 0
assert "timed out after 1s" in error
assert "[partial stdout before timeout]" in error
assert "partial stdout data" in error
assert "[partial stderr before timeout]" in error
assert "partial stderr data" in error
def test_run_subprocess_timeout_without_partial_output(monkeypatch):
"""Timeout error is clean when the subprocess produced no output."""
def mock_run(*args, **kwargs):
raise subprocess.TimeoutExpired(cmd=["mock"], timeout=5)
monkeypatch.setattr(subprocess, "run", mock_run)
result, duration, error = _run_subprocess(
["mock"], cwd=Path("/tmp"), env={}, timeout=5,
)
assert result is None
assert "timed out after 5s" in error
assert "[partial stdout" not in error
# ---------------------------------------------------------------------------
# run_all — output_root anchoring
# ---------------------------------------------------------------------------
def _make_mock_run_one_config(status="pass"):
"""Return a ``run_one_config`` stand-in that never invokes a subprocess."""
def mock(config, *, output_dir, repo_root, timeout=1800):
return PerfRun(
config=config,
status=status,
returncode=0 if status == "pass" else 1,
duration_sec=0.0,
log_path=Path("logs/test.log"),
metrics={"status": status},
)
return mock
def test_run_all_anchors_relative_output_root_to_repo_root(tmp_path, monkeypatch):
"""A relative ``output_root`` is resolved under ``repo_root``."""
monkeypatch.setattr(
"scripts.sweep_cam_perf.run_one_config",
_make_mock_run_one_config("pass"),
)
configs = [SweepConfig("test", 512, 512, 16)]
exit_code = run_all(
configs,
output_root="outputs/cam_perf",
date="2026-06-01",
repo_root=tmp_path,
)
expected_csv = tmp_path / "outputs" / "cam_perf" / "2026-06-01" / "sweep.csv"
assert expected_csv.exists(), f"sweep.csv not found at {expected_csv}"
assert exit_code == 0
def test_run_all_uses_absolute_output_root_unmodified(tmp_path, monkeypatch):
"""An absolute ``output_root`` is used as-is, not anchored under repo_root."""
monkeypatch.setattr(
"scripts.sweep_cam_perf.run_one_config",
_make_mock_run_one_config("pass"),
)
configs = [SweepConfig("test", 512, 512, 16)]
abs_output = tmp_path / "custom_dir"
repo_root = Path("/nonexistent/repo")
exit_code = run_all(
configs,
output_root=abs_output,
date="2026-06-01",
repo_root=repo_root,
)
expected_csv = abs_output / "2026-06-01" / "sweep.csv"
assert expected_csv.exists(), f"sweep.csv not found at {expected_csv}"
assert exit_code == 0
# ---------------------------------------------------------------------------
# plot_figures — required PNG outputs
# ---------------------------------------------------------------------------
def test_plot_figures_creates_required_pngs(tmp_path):
"""plot_figures creates four PNGs under figures/ when enough pass runs exist
across all three logical sweep groups. No exception with Agg backend."""
configs = [
SweepConfig("num_rows", 128, 512, 16),
SweepConfig("num_rows", 256, 512, 16),
SweepConfig("num_rows", 512, 512, 16), # baseline: all three groups
SweepConfig("hash_bits", 512, 128, 16),
SweepConfig("hash_bits", 512, 256, 16),
SweepConfig("lanes", 512, 512, 4),
SweepConfig("lanes", 512, 512, 8),
]
runs = [
PerfRun(
config=cfg,
status="pass",
returncode=0,
duration_sec=1.0,
log_path=Path("logs/test.log"),
metrics={
"latency_cycles": str(50 + cfg.num_rows % 100),
"total_cycles": str(100 + cfg.num_rows % 100),
"accepted_queries": "10",
"completed_queries": "10",
"queries_per_cycle": "0.05",
"status": "pass",
},
)
for cfg in configs
]
plot_figures(tmp_path, runs)
fig_dir = tmp_path / "figures"
assert fig_dir.is_dir()
assert (fig_dir / "cam_num_rows_vs_latency.png").exists()
assert (fig_dir / "cam_hash_bits_vs_latency.png").exists()
assert (fig_dir / "cam_lanes_vs_throughput.png").exists()
assert (fig_dir / "cam_perf_multipanel.png").exists()