#!/usr/bin/env python3 """Run fold on SV-COMP benchmarks and score the results. The runner has two entry points: 1. Batch mode for local evaluation across one or more SV-COMP categories. 2. Single-task mode for BenchExec via ``--benchexec-task ``. The current integration is intentionally conservative: - ``valid-deref`` maps to ``null-deref``. - ``no-overflow`` maps to ``int-overflow``. - ``unreach-call`` and ``no-deadcode`` map to ``dead`` as a partial proxy. - ``valid-free`` and ``valid-memtrack`` are reported as unsupported because the current fold tree does not yet ship a use-after-free / double-free detector. This means MemSafety scoring is partial today: only the ``valid-deref`` portion of the benchmark contributes to the score. """ from __future__ import annotations import argparse import concurrent.futures import datetime as dt import json import os import shlex import subprocess import sys import tempfile import time from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent RESULTS_ROOT = REPO_ROOT / "bench" / "results" DEFAULT_SVCOMP_DIR = REPO_ROOT / "bench" / "corpora" / "svcomp" DEFAULT_FOLD = REPO_ROOT / "build" / "release" / "src" / "cli" / "fold" SVCOMP_REPO = "https://gitlab.com/sosy-lab/benchmarking/sv-benchmarks.git" DEFAULT_SVCOMP_REF = "svcomp25" FOLD_SUPPORTED_DETECTORS = {"null-deref", "int-overflow", "dead"} CATEGORY_CONFIG = { "MemSafety-Heap": { "set_file": Path("c/MemSafety-Heap.set"), "description": "MemSafety tasks; only valid-deref is scored today.", }, "NoOverflows-BitVectors": { "set_file": Path("c/NoOverflows-BitVectors.set"), "description": "Signed-overflow tasks mapped to fold int-overflow.", }, "ReachSafety-Loops": { "set_file": Path("c/ReachSafety-Loops.set"), "description": "Loop reachability tasks mapped to fold dead-branch as a partial proxy.", }, } PROPERTY_MAP = { "valid-deref": { "detectors": ["null-deref"], "svcomp_false": "false(valid-deref)", "notes": "mapped to fold null-deref", }, "no-overflow": { "detectors": ["int-overflow"], "svcomp_false": "false(no-overflow)", "notes": "mapped to fold int-overflow", }, "unreach-call": { "detectors": ["dead"], "svcomp_false": "false(unreach-call)", "notes": "partial proxy via fold dead-branch", }, "no-deadcode": { "detectors": ["dead"], "svcomp_false": "false(no-deadcode)", "notes": "partial proxy via fold dead-branch", }, "valid-free": { "detectors": [], "svcomp_false": "false(valid-free)", "notes": "unsupported: no use-after-free detector yet", }, "valid-memtrack": { "detectors": [], "svcomp_false": "false(valid-memtrack)", "notes": "unsupported: no use-after-free/double-free detector yet", }, } def now_iso() -> str: return dt.datetime.now(dt.UTC).strftime("%Y-%m-%dT%H-%M-%SZ") def repo_dir_for(svcomp_dir: Path) -> Path: return svcomp_dir / "sv-benchmarks" def print_fetch_instructions(svcomp_dir: Path, svcomp_ref: str) -> None: repo_dir = repo_dir_for(svcomp_dir) print("SV-COMP benchmarks are not available locally.", file=sys.stderr) print("", file=sys.stderr) print("To fetch the C benchmark subset:", file=sys.stderr) print( f" python3 bench/run_svcomp.py --fetch --svcomp-dir {shlex.quote(str(svcomp_dir))} --svcomp-ref {shlex.quote(svcomp_ref)}", file=sys.stderr, ) print("", file=sys.stderr) print("Equivalent manual commands:", file=sys.stderr) print( f" git clone --depth 1 --branch {shlex.quote(svcomp_ref)} --filter=blob:none --sparse {SVCOMP_REPO} {shlex.quote(str(repo_dir))}", file=sys.stderr, ) print( f" git -C {shlex.quote(str(repo_dir))} sparse-checkout set c properties", file=sys.stderr, ) def ensure_svcomp_repo(svcomp_dir: Path, svcomp_ref: str, fetch: bool) -> Path | None: repo_dir = repo_dir_for(svcomp_dir) if repo_dir.is_dir(): return repo_dir if not fetch: return None svcomp_dir.mkdir(parents=True, exist_ok=True) clone_cmd = [ "git", "clone", "--depth", "1", "--branch", svcomp_ref, "--filter=blob:none", "--sparse", SVCOMP_REPO, str(repo_dir), ] sparse_cmd = [ "git", "-C", str(repo_dir), "sparse-checkout", "set", "c", "properties", ] subprocess.run(clone_cmd, check=True) subprocess.run(sparse_cmd, check=True) return repo_dir def parse_bool(text: str) -> bool | None: low = text.strip().strip("'\"").lower() if low == "true": return True if low == "false": return False return None def load_task_yaml(task_path: Path) -> dict: try: import yaml # type: ignore data = yaml.safe_load(task_path.read_text()) return data if isinstance(data, dict) else {} except Exception: pass lines = task_path.read_text().splitlines() doc: dict[str, object] = {"properties": [], "options": {}} properties: list[dict[str, object]] = [] options: dict[str, object] = {} input_files: list[str] = [] current_prop: dict[str, object] | None = None in_properties = False in_options = False in_input_list = False for raw_line in lines: line = raw_line.split("#", 1)[0].rstrip() if not line.strip(): continue stripped = line.strip() if stripped.startswith("format_version:"): continue if stripped.startswith("input_files:"): in_properties = False in_options = False value = stripped.split(":", 1)[1].strip() in_input_list = value == "" if value: input_files.append(value.strip("'\"")) continue if stripped == "properties:": in_properties = True in_options = False in_input_list = False continue if stripped == "options:": in_properties = False in_options = True in_input_list = False continue if in_input_list and stripped.startswith("-"): input_files.append(stripped[1:].strip().strip("'\"")) continue if in_properties and stripped.startswith("- "): if current_prop: properties.append(current_prop) current_prop = {} tail = stripped[2:].strip() if tail.startswith("property_file:"): current_prop["property_file"] = tail.split(":", 1)[1].strip().strip("'\"") continue if in_properties and current_prop is not None and ":" in stripped: key, value = [part.strip() for part in stripped.split(":", 1)] bool_value = parse_bool(value) current_prop[key] = bool_value if bool_value is not None else value.strip("'\"") continue if in_options and ":" in stripped: key, value = [part.strip() for part in stripped.split(":", 1)] bool_value = parse_bool(value) options[key] = bool_value if bool_value is not None else value.strip("'\"") if current_prop: properties.append(current_prop) doc["properties"] = properties doc["options"] = options if not input_files: sibling = task_path.with_suffix("") for suffix in (".c", ".i"): candidate = sibling.with_suffix(suffix) if candidate.exists(): input_files = [candidate.name] break doc["input_files"] = input_files[0] if len(input_files) == 1 else input_files return doc def as_list(value: object) -> list[str]: if value is None: return [] if isinstance(value, list): return [str(v) for v in value] return [str(value)] def property_key_from_path(property_file: str) -> str: name = Path(property_file).name if name.endswith(".prp"): name = name[:-4] return name def detector_union(properties: list[dict]) -> list[str]: detectors: set[str] = set() for prop in properties: detectors.update(prop.get("detectors", [])) return sorted(detectors & FOLD_SUPPORTED_DETECTORS) def load_task_record(task_path: Path) -> dict: raw = load_task_yaml(task_path) input_files = [task_path.parent / rel for rel in as_list(raw.get("input_files"))] normalized_properties = [] for prop in raw.get("properties", []): if not isinstance(prop, dict): continue property_file = str(prop.get("property_file", "")) key = property_key_from_path(property_file) mapping = PROPERTY_MAP.get( key, { "detectors": [], "svcomp_false": f"false({key})", "notes": "unsupported property", }, ) normalized_properties.append( { "property_key": key, "property_file": property_file, "expected_verdict": prop.get("expected_verdict"), "detectors": list(mapping["detectors"]), "svcomp_false": mapping["svcomp_false"], "supported": bool(mapping["detectors"]), "notes": mapping["notes"], } ) options = raw.get("options", {}) if isinstance(raw.get("options"), dict) else {} return { "task": str(task_path), "task_path": task_path, "input_files": [path.resolve() for path in input_files if path.exists()], "properties": normalized_properties, "data_model": str(options.get("data_model", "ILP32")), "language": str(options.get("language", "C")), } def parse_set_file(set_path: Path) -> list[Path]: includes: list[str] = [] excludes: list[str] = [] for raw_line in set_path.read_text().splitlines(): line = raw_line.strip() if not line or line.startswith("#"): continue if line.startswith("-"): excludes.append(line[1:].strip()) else: includes.append(line) matched: list[Path] = [] seen: set[Path] = set() for pattern in includes: for candidate in sorted(set_path.parent.glob(pattern)): resolved = candidate.resolve() if resolved.suffix == ".yml" and resolved not in seen: seen.add(resolved) matched.append(resolved) if excludes: excluded: set[Path] = set() for pattern in excludes: excluded.update(candidate.resolve() for candidate in set_path.parent.glob(pattern)) matched = [path for path in matched if path not in excluded] return matched def category_tasks(repo_dir: Path, category: str) -> list[dict]: config = CATEGORY_CONFIG[category] set_path = repo_dir / config["set_file"] if not set_path.is_file(): raise FileNotFoundError(f"missing SV-COMP set file: {set_path}") return [load_task_record(task_path) for task_path in parse_set_file(set_path)] def compile_flags_for(task: dict, source_path: Path) -> list[str]: flags = ["clang", "-std=gnu11", "-DSVCOMP=1", "-D__attribute__(x)="] data_model = str(task.get("data_model", "ILP32")).upper() if data_model == "LP64": flags.extend(["-D__LP64__=1", "-D__WORDSIZE=64"]) else: flags.extend(["-U__LP64__", "-D__ILP32__=1", "-D__WORDSIZE=32"]) flags.extend(["-c", str(source_path)]) return flags def write_compile_db(tasks: list[dict], db_path: Path) -> list[dict]: entries: list[dict] = [] seen: set[Path] = set() for task in tasks: for source in task["input_files"]: if source in seen: continue seen.add(source) entries.append( { "directory": str(source.parent), "command": shlex.join(compile_flags_for(task, source)), "file": str(source), } ) db_path.parent.mkdir(parents=True, exist_ok=True) db_path.write_text(json.dumps(entries, indent=2) + "\n") return entries def run_fold( fold_bin: Path, compile_db: Path, detectors: list[str], timeout_s: int, source_paths: list[Path] | None = None, processes: int = 1, ) -> dict: with tempfile.TemporaryDirectory(prefix="fold-svcomp-") as temp_dir: sarif_path = Path(temp_dir) / "fold.sarif" command = [ str(fold_bin), "scan", "--compile-db", str(compile_db), "--format", "sarif", "--output", str(sarif_path), "--min-body-stmts", "1", "--processes", str(processes), ] for detector in detectors: command.extend(["--detectors", detector]) for source_path in source_paths or []: command.append(str(source_path)) started = time.monotonic() try: proc = subprocess.run( command, capture_output=True, text=True, timeout=timeout_s, ) timed_out = False except subprocess.TimeoutExpired as exc: proc = exc timed_out = True elapsed = time.monotonic() - started findings: list[dict] = [] if sarif_path.exists(): doc = json.loads(sarif_path.read_text()) for run in doc.get("runs", []): for result in run.get("results", []): kind = str(result.get("ruleId", "fold.unknown")).removeprefix("fold.") for loc in result.get("locations", []): physical = loc.get("physicalLocation", {}) artifact = physical.get("artifactLocation", {}) region = physical.get("region", {}) findings.append( { "file": str(Path(artifact.get("uri", "")).resolve()), "line": int(region.get("startLine", 0) or 0), "kind": kind, "id": result.get("properties", {}).get("id", ""), "message": result.get("message", {}).get("text", ""), } ) stdout = getattr(proc, "stdout", "") or "" stderr = getattr(proc, "stderr", "") or "" return { "command": command, "elapsed_s": elapsed, "exit_code": getattr(proc, "returncode", 124 if timed_out else 1), "timed_out": timed_out, "stdout": stdout, "stderr": stderr, "findings": findings, } def findings_by_file_and_kind(findings: list[dict]) -> dict[tuple[str, str], list[dict]]: grouped: dict[tuple[str, str], list[dict]] = {} for finding in findings: key = (str(Path(finding["file"]).resolve()), str(finding["kind"])) grouped.setdefault(key, []).append(finding) return grouped def run_task_scan(task: dict, fold_bin: Path, compile_db: Path, timeout_s: int) -> dict: detectors = detector_union(task["properties"]) if not detectors: return { "task": task["task"], "detectors": [], "elapsed_s": 0.0, "exit_code": 0, "timed_out": False, "stdout": "", "stderr": "", "findings": [], "command": [], } fold_run = run_fold( fold_bin, compile_db, detectors, timeout_s, source_paths=task["input_files"], processes=1, ) return {"task": task["task"], "detectors": detectors, **fold_run} def evaluate_task(task: dict, grouped_findings: dict[tuple[str, str], list[dict]]) -> dict: task_files = {str(path.resolve()) for path in task["input_files"]} supported = [prop for prop in task["properties"] if prop["supported"]] unsupported = [prop for prop in task["properties"] if not prop["supported"]] if not supported: return { "task": task["task"], "status": "unsupported", "svcomp_result": "unknown", "matched_property": None, "supported_properties": [], "unsupported_properties": [prop["property_key"] for prop in unsupported], "detectors": [], "findings": [], } matching_findings: list[dict] = [] for prop in supported: prop_hits: list[dict] = [] for detector in prop["detectors"]: for file_path in task_files: prop_hits.extend(grouped_findings.get((file_path, detector), [])) if prop_hits: matching_findings.extend(prop_hits) if prop["expected_verdict"] is False: return { "task": task["task"], "status": "correct", "svcomp_result": prop["svcomp_false"], "matched_property": prop["property_key"], "supported_properties": [p["property_key"] for p in supported], "unsupported_properties": [p["property_key"] for p in unsupported], "detectors": prop["detectors"], "findings": sorted( { (finding["file"], finding["line"], finding["kind"]): finding for finding in prop_hits }.values(), key=lambda item: (item["file"], item["line"], item["kind"]), ), } for prop in supported: if prop["expected_verdict"] is True: prop_hits: list[dict] = [] for detector in prop["detectors"]: for file_path in task_files: prop_hits.extend(grouped_findings.get((file_path, detector), [])) if prop_hits: return { "task": task["task"], "status": "incorrect", "svcomp_result": prop["svcomp_false"], "matched_property": prop["property_key"], "supported_properties": [p["property_key"] for p in supported], "unsupported_properties": [p["property_key"] for p in unsupported], "detectors": prop["detectors"], "findings": sorted( { (finding["file"], finding["line"], finding["kind"]): finding for finding in prop_hits }.values(), key=lambda item: (item["file"], item["line"], item["kind"]), ), } return { "task": task["task"], "status": "unknown", "svcomp_result": "unknown", "matched_property": None, "supported_properties": [p["property_key"] for p in supported], "unsupported_properties": [p["property_key"] for p in unsupported], "detectors": detector_union(supported), "findings": sorted( { (finding["file"], finding["line"], finding["kind"]): finding for finding in matching_findings }.values(), key=lambda item: (item["file"], item["line"], item["kind"]), ), } def score_category(category: str, tasks: list[dict], task_runs: list[dict]) -> dict: run_lookup = {run["task"]: run for run in task_runs} task_results = [] for task in tasks: task_run = run_lookup.get(task["task"], {"findings": [], "elapsed_s": 0.0, "exit_code": 0, "timed_out": False, "stderr": "", "detectors": []}) grouped = findings_by_file_and_kind(task_run.get("findings", [])) scored = evaluate_task(task, grouped) scored["fold"] = { "elapsed_s": task_run.get("elapsed_s", 0.0), "exit_code": task_run.get("exit_code", 0), "timed_out": task_run.get("timed_out", False), "detectors": task_run.get("detectors", []), "finding_count": len(task_run.get("findings", [])), "stderr_tail": (task_run.get("stderr", "") or "")[-1000:], } task_results.append(scored) scored = [result for result in task_results if result["status"] != "unsupported"] counts = {status: 0 for status in ("correct", "incorrect", "unknown", "unsupported")} for result in task_results: counts[result["status"]] += 1 total_scored = len(scored) score_pct = (counts["correct"] / total_scored * 100.0) if total_scored else 0.0 detectors = sorted({det for task in tasks for det in detector_union(task["properties"])}) unsupported_properties = sorted( { prop["property_key"] for task in tasks for prop in task["properties"] if not prop["supported"] } ) return { "category": category, "description": CATEGORY_CONFIG[category]["description"], "tasks_total": len(tasks), "tasks_scored": total_scored, "correct": counts["correct"], "incorrect": counts["incorrect"], "unknown": counts["unknown"], "unsupported": counts["unsupported"], "score_pct": score_pct, "detectors": detectors, "unsupported_properties": unsupported_properties, "fold": { "elapsed_s": sum(run.get("elapsed_s", 0.0) for run in task_runs), "timed_out_tasks": sorted(run["task"] for run in task_runs if run.get("timed_out")), "finding_count": sum(len(run.get("findings", [])) for run in task_runs), }, "task_results": task_results, } def render_table(results: list[dict], fold_version: str) -> str: title = f"SV-COMP Results for fold {fold_version}" rows = [ title, "=" * len(title), "", f"{'Category':24} {'Tasks':>6} {'Correct':>8} {'Incorrect':>10} {'Unknown':>8} {'Score':>8}", ] total_tasks = 0 total_correct = 0 total_incorrect = 0 total_unknown = 0 total_scored = 0 for result in results: total_tasks += result["tasks_scored"] total_correct += result["correct"] total_incorrect += result["incorrect"] total_unknown += result["unknown"] total_scored += result["tasks_scored"] rows.append( f"{result['category']:24} {result['tasks_scored']:6d} {result['correct']:8d} {result['incorrect']:10d} {result['unknown']:8d} {result['score_pct']:7.1f}%" ) rows.append("-" * 70) total_score = (total_correct / total_scored * 100.0) if total_scored else 0.0 rows.append( f"{'Total':24} {total_tasks:6d} {total_correct:8d} {total_incorrect:10d} {total_unknown:8d} {total_score:7.1f}%" ) return "\n".join(rows) def read_fold_version(fold_bin: Path) -> str: try: proc = subprocess.run([str(fold_bin), "--version"], capture_output=True, text=True, check=False) except OSError: return "unknown" text = (proc.stdout or proc.stderr).strip() return text or "unknown" def check_mode(svcomp_dir: Path, svcomp_ref: str) -> int: repo_dir = repo_dir_for(svcomp_dir) if repo_dir.is_dir(): print(f"SV-COMP benchmarks available: {repo_dir}") return 0 print_fetch_instructions(svcomp_dir, svcomp_ref) return 2 def run_single_task(task_path: Path, fold_bin: Path, timeout_s: int) -> int: task = load_task_record(task_path) detectors = detector_union(task["properties"]) if not detectors: print("SVCOMP_RESULT=unknown") return 0 with tempfile.TemporaryDirectory(prefix="fold-svcomp-task-") as temp_dir: db_path = Path(temp_dir) / "compile_commands.json" write_compile_db([task], db_path) fold_run = run_fold( fold_bin, db_path, detectors, timeout_s, source_paths=task["input_files"], processes=1, ) result = evaluate_task(task, findings_by_file_and_kind(fold_run["findings"])) print(f"SVCOMP_RESULT={result['svcomp_result']}") return 0 def run_batch(args: argparse.Namespace) -> int: svcomp_dir = Path(args.svcomp_dir).resolve() repo_dir = ensure_svcomp_repo(svcomp_dir, args.svcomp_ref, args.fetch) if repo_dir is None: print_fetch_instructions(svcomp_dir, args.svcomp_ref) return 2 fold_bin = Path(args.fold).resolve() if not fold_bin.is_file(): print(f"fold binary not found: {fold_bin}", file=sys.stderr) return 2 stamp = now_iso() out_root = Path(args.out_root).resolve() / f"{stamp}_svcomp" out_root.mkdir(parents=True, exist_ok=True) all_results: list[dict] = [] manifest = { "generated_at": dt.datetime.now(dt.UTC).isoformat(), "svcomp_repo": str(repo_dir), "svcomp_ref": args.svcomp_ref, "fold": str(fold_bin), "fold_version": read_fold_version(fold_bin), "categories": args.categories, "timeout_s": args.timeout, "jobs": args.jobs, } for category in args.categories: if category not in CATEGORY_CONFIG: print(f"unsupported SV-COMP category: {category}", file=sys.stderr) return 2 tasks = category_tasks(repo_dir, category) category_out = out_root / category category_out.mkdir(parents=True, exist_ok=True) compile_db = category_out / "compile_commands.json" write_compile_db(tasks, compile_db) task_runs: list[dict] = [] with concurrent.futures.ThreadPoolExecutor(max_workers=args.jobs) as pool: futures = [ pool.submit(run_task_scan, task, fold_bin, compile_db, args.timeout) for task in tasks ] for future in concurrent.futures.as_completed(futures): task_runs.append(future.result()) task_runs.sort(key=lambda item: item["task"]) score = score_category(category, tasks, task_runs) (category_out / "fold.json").write_text( json.dumps( { "tool": "fold", "category": category, "task_runs": task_runs, }, indent=2, sort_keys=True, ) + "\n" ) (category_out / "score.json").write_text(json.dumps(score, indent=2, sort_keys=True) + "\n") all_results.append(score) summary = render_table(all_results, manifest["fold_version"]) (out_root / "meta.json").write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n") (out_root / "summary.md").write_text(summary + "\n") print(summary) print(f"results written to {out_root}") return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--svcomp-dir", default=str(DEFAULT_SVCOMP_DIR)) parser.add_argument("--svcomp-ref", default=DEFAULT_SVCOMP_REF) parser.add_argument( "--categories", nargs="+", default=["MemSafety-Heap", "NoOverflows-BitVectors", "ReachSafety-Loops"], ) parser.add_argument("--fold", default=str(DEFAULT_FOLD)) parser.add_argument("--jobs", type=int, default=max(1, (os.cpu_count() or 4) // 2)) parser.add_argument("--timeout", type=int, default=900, help="Per-task timeout in seconds") parser.add_argument("--out-root", default=str(RESULTS_ROOT)) parser.add_argument("--fetch", action="store_true", help="Download the sparse SV-COMP checkout if missing") parser.add_argument("--check", action="store_true", help="Check whether sv-benchmarks is available locally") parser.add_argument("--benchexec-task", help="Run a single SV-COMP .yml task and print an SV-COMP verdict") return parser def main() -> int: args = build_parser().parse_args() if args.check: return check_mode(Path(args.svcomp_dir).resolve(), args.svcomp_ref) if args.benchexec_task: return run_single_task(Path(args.benchexec_task).resolve(), Path(args.fold).resolve(), args.timeout) return run_batch(args) if __name__ == "__main__": raise SystemExit(main())