| """Run pinned upstream projects and gate compatible CLI candidates. |
| |
| This is an evaluation bridge, not a learned proposer. Reference build success |
| must never be counted as LightNest generating or improving either project. |
| Only trusted source trees and locally authorized candidate executables should |
| be supplied. Process resource limits are not a security sandbox. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import gzip |
| import hashlib |
| import io |
| import json |
| import os |
| from pathlib import Path |
| import platform |
| import random |
| import re |
| import resource |
| import shutil |
| import signal |
| import statistics |
| import subprocess |
| import tarfile |
| import time |
| import zlib |
|
|
|
|
| PROJECTS = { |
| "xxhash": { |
| "version": "0.8.3", "directory": "xxHash-0.8.3", |
| "url": "https://codeload.github.com/Cyan4973/xxHash/tar.gz/refs/tags/v0.8.3", |
| "archive_sha256": "aae608dfe8213dfd05d909a57718ef82f30722c392344583d3f39050c7f29a80", |
| "upstream_test_target": "make -j2 check", |
| }, |
| "zlib": { |
| "version": "1.3.2", "directory": "zlib-1.3.2", |
| "url": "https://zlib.net/fossils/zlib-1.3.2.tar.gz", |
| "archive_sha256": "bb329a0a2cd0274d05519d61c667c062e06990d72e125ee2dfa8de64f0119d16", |
| "upstream_test_target": "make -j2 test", |
| }, |
| } |
|
|
|
|
| def sha(path: Path) -> str: |
| digest = hashlib.sha256() |
| with path.open("rb") as stream: |
| while chunk := stream.read(1 << 20): |
| digest.update(chunk) |
| return digest.hexdigest() |
|
|
|
|
| def save(path: Path, value) -> None: |
| path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n") |
|
|
|
|
| def source_manifest(path: Path) -> dict: |
| files = [] |
| for item in sorted(path.rglob("*")): |
| if item.is_symlink(): |
| raise ValueError(f"source symlinks are not supported: {item}") |
| if item.is_file(): |
| files.append({"path": str(item.relative_to(path)), "bytes": item.stat().st_size, |
| "sha256": sha(item)}) |
| encoded = json.dumps(files, sort_keys=True, separators=(",", ":")).encode() |
| return {"files": files, "file_count": len(files), "bytes": sum(f["bytes"] for f in files), |
| "tree_sha256": hashlib.sha256(encoded).hexdigest()} |
|
|
|
|
| def verify_source_archive(archive: Path, source: Path, expected_hash: str) -> None: |
| if sha(archive) != expected_hash: |
| raise ValueError(f"pinned archive hash mismatch: {archive}") |
| expected = {} |
| with tarfile.open(archive) as contents: |
| for entry in contents.getmembers(): |
| if not entry.isfile(): |
| continue |
| name = Path(entry.name).relative_to(source.name) |
| if ".." in name.parts or name.is_absolute(): |
| raise ValueError("invalid archive member") |
| with contents.extractfile(entry) as stream: |
| expected[str(name)] = hashlib.sha256(stream.read()).hexdigest() |
| actual = {str(p.relative_to(source)): sha(p) for p in source.rglob("*") if p.is_file()} |
| if actual != expected: |
| raise ValueError(f"reference tree differs from pinned source archive: {source}") |
|
|
|
|
| def child_limits() -> None: |
| resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) |
| resource.setrlimit(resource.RLIMIT_FSIZE, (64 << 20, 64 << 20)) |
| resource.setrlimit(resource.RLIMIT_AS, (1 << 30, 1 << 30)) |
| resource.setrlimit(resource.RLIMIT_CPU, (120, 120)) |
|
|
|
|
| class Runner: |
| def __init__(self, output: Path): |
| self.output = output |
| self.logs = output / "logs" |
| self.logs.mkdir() |
| self.commands = [] |
| self.measure = output / "measure-process" |
| subprocess.run(["gcc", "-std=c11", "-O2", "-Wall", "-Wextra", |
| str(Path(__file__).with_name("measure_process.c")), "-o", str(self.measure)], |
| check=True, timeout=30) |
|
|
| def run(self, argv, cwd: Path, stdin: Path | None = None, |
| timeout: float = 90, required: bool = True) -> dict: |
| index = len(self.commands) |
| base = self.logs / f"{index:04d}" |
| stdout = base.with_suffix(".stdout") |
| stderr = base.with_suffix(".stderr") |
| metrics = base.with_suffix(".metrics") |
| argv = [str(a) for a in argv] |
| command = [str(self.measure), str(metrics), *argv] |
| begin = time.perf_counter() |
| with (stdin.open("rb") if stdin else open(os.devnull, "rb")) as inp, \ |
| stdout.open("wb") as out, stderr.open("wb") as err: |
| process = subprocess.Popen(command, cwd=cwd, stdin=inp, stdout=out, stderr=err, |
| start_new_session=True, preexec_fn=child_limits) |
| timed_out = False |
| try: |
| status = process.wait(timeout=timeout) |
| except subprocess.TimeoutExpired: |
| timed_out = True |
| os.killpg(process.pid, signal.SIGKILL) |
| status = process.wait() |
| finally: |
| |
| |
| try: |
| os.killpg(process.pid, signal.SIGKILL) |
| except ProcessLookupError: |
| pass |
| elapsed = time.perf_counter() - begin |
| stats = metrics.read_text().strip().splitlines() if metrics.exists() else [] |
| numbers = stats[-1].split() if stats else [] |
| measured = {} |
| if len(numbers) == 6: |
| try: |
| measured = {"user_cpu_s": float(numbers[0]), "system_cpu_s": float(numbers[1]), |
| "peak_rss_kib": int(numbers[2]), "wall_s": float(numbers[3]), |
| "exited_normally": numbers[4] == "1", "term_signal": int(numbers[5])} |
| except ValueError: |
| pass |
| result = {"argv": argv, "cwd": str(cwd), "stdin": str(stdin) if stdin else None, |
| "returncode": status, "timed_out": timed_out, "controller_wall_s": elapsed, |
| "wall_s": elapsed, |
| "stdout": str(stdout.relative_to(self.output)), |
| "stderr": str(stderr.relative_to(self.output)), "metrics_valid": bool(measured), **measured} |
| self.commands.append(result) |
| save(self.output / "commands.json", self.commands) |
| if required and status != 0: |
| raise RuntimeError(f"command failed ({status}); see {stderr}") |
| return result |
|
|
| def data(self, result: dict) -> bytes: |
| return (self.output / result["stdout"]).read_bytes() |
|
|
|
|
| def make_corpus(output: Path) -> list[dict]: |
| directory = output / "corpus" |
| directory.mkdir() |
| records = [] |
| rng = random.Random(202610170220) |
| |
| for size in (0, 1, 31, 4096, 1 << 20, 16 << 20): |
| for kind in ("random", "repeated"): |
| name = f"{kind}-{size}.bin" |
| path = directory / name |
| if kind == "random": |
| with path.open("wb") as stream: |
| remaining = size |
| while remaining: |
| block = rng.randbytes(min(remaining, 1 << 20)) |
| stream.write(block) |
| remaining -= len(block) |
| else: |
| block = (b"LightNest project compatibility 0123456789\n" * 26000)[:1 << 20] |
| with path.open("wb") as stream: |
| remaining = size |
| while remaining: |
| part = block[:remaining] |
| stream.write(part) |
| remaining -= len(part) |
| records.append({"name": name, "bytes": size, "kind": kind, "sha256": sha(path)}) |
| save(output / "corpus.json", records) |
| return records |
|
|
|
|
| def digest_output(data: bytes) -> str: |
| |
| text = data.decode("ascii").strip() |
| match = re.search(r"(?:=\s*|^XXH3_|^)([0-9a-fA-F]{8,32})(?:\s|$)", text) |
| if not match: |
| raise ValueError(f"unrecognized xxhsum digest: {text[:160]!r}") |
| return match.group(1).lower() |
|
|
|
|
| def evaluate_xxhash(runner: Runner, binary: Path, corpus: list[dict], expected=None) -> dict: |
| cases = [] |
| digests = {} |
| for item in corpus: |
| source = runner.output / "corpus" / item["name"] |
| for mode in ("-H0", "-H3"): |
| key = item["name"] + mode |
| result = runner.run([binary, mode], binary.parent, stdin=source, required=False) |
| digest = None |
| if result["returncode"] == 0: |
| try: |
| digest = digest_output(runner.data(result)) |
| except (UnicodeError, ValueError): |
| pass |
| ok = digest is not None and (expected is None or digest == expected[key]) |
| digests[key] = digest |
| cases.append({"case": key, "ok": ok, "digest": digest, "run": result}) |
| return {"passed": sum(c["ok"] for c in cases), "total": len(cases), "cases": cases, |
| "digests": digests} |
|
|
|
|
| def evaluate_zlib(runner: Runner, binary: Path, corpus: list[dict]) -> dict: |
| cases = [] |
| compressed_sizes = {} |
| for item in corpus: |
| source = runner.output / "corpus" / item["name"] |
| raw = source.read_bytes() |
| compressed = runner.run([binary, "-c"], binary.parent, stdin=source, required=False) |
| encoded = runner.data(compressed) |
| try: |
| with gzip.GzipFile(fileobj=io.BytesIO(encoded)) as stream: |
| decoded = stream.read(len(raw) + 1) |
| ok = compressed["returncode"] == 0 and decoded == raw |
| except (OSError, EOFError, ValueError, zlib.error): |
| ok = False |
| cases.append({"case": item["name"] + ":compress", "ok": ok, "run": compressed}) |
| compressed_sizes[item["name"]] = len(encoded) |
| |
| |
| reference_gzip = source.with_suffix(".reference.gz") |
| if not reference_gzip.exists(): |
| reference_gzip.write_bytes(gzip.compress(raw, mtime=0)) |
| expanded = runner.run([binary, "-d", "-c"], binary.parent, stdin=reference_gzip, required=False) |
| decode_ok = expanded["returncode"] == 0 and runner.data(expanded) == raw |
| cases.append({"case": item["name"] + ":decompress", "ok": decode_ok, "run": expanded}) |
| if item["bytes"] >= 4096: |
| broken = source.with_suffix(".corrupt.gz") |
| corrupted = bytearray(reference_gzip.read_bytes()) |
| corrupted[-8] ^= 1 |
| broken.write_bytes(corrupted) |
| bad = runner.run([binary, "-d", "-c"], binary.parent, stdin=broken, required=False) |
| cases.append({"case": item["name"] + ":reject_bad_crc", |
| "ok": decode_ok and bad["returncode"] != 0 and bad.get("exited_normally", False) |
| and not bad["timed_out"], "run": bad}) |
| return {"passed": sum(c["ok"] for c in cases), "total": len(cases), "cases": cases, |
| "compressed_sizes": compressed_sizes, |
| "independent_gzip_runtime": __import__("zlib").ZLIB_RUNTIME_VERSION} |
|
|
|
|
| def performance(runner: Runner, binary: Path, project: str, corpus: list[dict]) -> list[dict]: |
| measurements = [] |
| order = [(item, rep) for item in corpus if item["bytes"] >= 1 << 20 for rep in range(5)] |
| random.Random(742191).shuffle(order) |
| samples = {} |
| for item, rep in order: |
| source = runner.output / "corpus" / item["name"] |
| args = [binary, "-H3"] if project == "xxhash" else [binary, "-c"] |
| result = runner.run(args, binary.parent, stdin=source, required=False) |
| samples.setdefault(item["name"], []).append({"repeat": rep, **result}) |
| for item in corpus: |
| if item["name"] not in samples: |
| continue |
| runs = samples[item["name"]] |
| valid = all(r["returncode"] == 0 and r["metrics_valid"] and r.get("exited_normally", False) |
| and not r["timed_out"] for r in runs) |
| walls = [r["wall_s"] for r in runs] |
| median = statistics.median(walls) if valid else None |
| measurements.append({"case": item["name"], "bytes": item["bytes"], "runs": runs, |
| "all_runs_ok": valid, "metrics_valid": valid, |
| "median_wall_s": median, "min_wall_s": min(walls) if valid else None, |
| "max_wall_s": max(walls) if valid else None, |
| "median_mib_s_including_process_io": item["bytes"] / (1 << 20) / median if valid else None, |
| "peak_rss_kib": max(r["peak_rss_kib"] for r in runs) if valid else None}) |
| return measurements |
|
|
|
|
| def main(argv=None) -> int: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--sources", type=Path, required=True) |
| parser.add_argument("--archives", type=Path, help="defaults to sources/../downloads") |
| parser.add_argument("--output", type=Path, required=True) |
| parser.add_argument("--xxhash-candidate", type=Path) |
| parser.add_argument("--zlib-candidate", type=Path) |
| args = parser.parse_args(argv) |
| output = args.output.resolve() |
| output.mkdir(parents=True, exist_ok=False) |
| runner = Runner(output) |
| protocol = {"projects": PROJECTS, "seed": 202610170220, "max_input_bytes": 16 << 20, |
| "benchmark_repeats": 5, |
| "evaluation_role": "public development/reference alignment, not hidden final testing", |
| "candidate_gate": "all upstream baseline tests and all compatibility cases pass", |
| "performance_claim": "exploratory measurements only; no automatic superiority claim", |
| "learner_status": "current RV32 patch policy cannot propose a compatible native project candidate", |
| "resource_limits": {"address_space_bytes": 1 << 30, "output_bytes_per_command": 64 << 20, |
| "cpu_seconds_per_process": 120, "wall_seconds_per_command": 90}, |
| "environment": {"python": platform.python_version(), "platform": platform.platform(), |
| "cpu_count": os.cpu_count(), |
| "compiler": subprocess.check_output(["gcc", "--version"], text=True).splitlines()[0]}} |
| save(output / "protocol.json", protocol) |
| corpus = make_corpus(output) |
| summary = {"protocol": protocol, "projects": {}, "lightnest_project_improvement": False} |
| for project, spec in PROJECTS.items(): |
| original = args.sources.resolve() / spec["directory"] |
| archives = args.archives or args.sources.parent / "downloads" |
| archive = archives / ("xxHash-v0.8.3.tar.gz" if project == "xxhash" else "zlib-1.3.2.tar.gz") |
| verify_source_archive(archive, original, spec["archive_sha256"]) |
| manifest = source_manifest(original) |
| save(output / f"{project}-source-manifest.json", manifest) |
| build = output / "build" / spec["directory"] |
| shutil.copytree(original, build) |
| if project == "xxhash": |
| upstream = runner.run(["make", "-j2", "check"], build) |
| binary = build / "xxhsum" |
| else: |
| runner.run(["./configure"], build) |
| upstream = runner.run(["make", "-j2", "test"], build) |
| binary = build / "minigzip" |
| reference = (evaluate_xxhash(runner, binary, corpus) if project == "xxhash" |
| else evaluate_zlib(runner, binary, corpus)) |
| result = {"source_tree_sha256": manifest["tree_sha256"], "source_bytes": manifest["bytes"], |
| "source_files": manifest["file_count"], "upstream_test": upstream, |
| "reference_binary_sha256": sha(binary), "reference_binary_bytes": binary.stat().st_size, |
| "reference": reference, "performance": performance(runner, binary, project, corpus)} |
| candidate = getattr(args, project + "_candidate") |
| if candidate: |
| candidate = candidate.resolve(strict=True) |
| candidate_result = (evaluate_xxhash(runner, candidate, corpus, reference["digests"]) |
| if project == "xxhash" else evaluate_zlib(runner, candidate, corpus)) |
| candidate_result["sha256"] = sha(candidate) |
| candidate_result["alignment_cases_pass"] = candidate_result["passed"] == candidate_result["total"] |
| candidate_result["upstream_candidate_suite"] = "not supplied; a CLI-only check cannot establish full-library compatibility" |
| candidate_result["promoted"] = False |
| result["candidate"] = candidate_result |
| else: |
| result["candidate"] = {"status": "not_proposed", "promoted": False, |
| "reason": "current learner supports RV32 windows, not these native x86-64 project interfaces"} |
| summary["projects"][project] = result |
| save(output / "report.json", summary) |
| print(json.dumps({name: {"upstream_test_ok": data["upstream_test"]["returncode"] == 0, |
| "compatibility": f"{data['reference']['passed']}/{data['reference']['total']}", |
| "source_files": data["source_files"], "source_bytes": data["source_bytes"], |
| "candidate": data["candidate"]} |
| for name, data in summary["projects"].items()}, indent=2)) |
| return 0 if all(d["reference"]["passed"] == d["reference"]["total"] |
| and d["candidate"].get("alignment_cases_pass", True) |
| for d in summary["projects"].values()) else 1 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
