diff --git a/.gitignore b/.gitignore index 6b744f9..fadbd5b 100644 --- a/.gitignore +++ b/.gitignore @@ -13,6 +13,7 @@ # Local research and experiment state. /.knowledge/ /experiments/ +/artifacts/ # Python caches. __pycache__/ diff --git a/benchmarks/cnc/README.md b/benchmarks/cnc/README.md index 203983a..e220adf 100644 --- a/benchmarks/cnc/README.md +++ b/benchmarks/cnc/README.md @@ -1,44 +1,76 @@ -# Auditable Cube-and-Conquer measurements +# Factoring Cube-and-Conquer -## Current CnC comparison +This directory contains only the operations needed for factoring runs: -The primary comparison uses online stopping rules on both sides: +1. generate paired SAT/UNSAT `n×n` factoring instances in CircuitSAT and CNF; +2. solve a CNF directly with Kissat; +3. cube a CNF with `march_cu`, then solve the cubes in parallel with Kissat; +4. use the Rust solver either to export a complete frontier or to stream cubes + directly into a bounded parallel Kissat pool; +5. solve a frozen frontier in Python for reproducible paper measurements. -- Region cubing reads structured CircuitSAT and stops at classical CC - difficulty `D^2(D+I)/N > threshold`. Calibrate the threshold by frontier - count only with `calibrate_cc_difficulty.py`. -- `march_cu` reads a globally encoded CNF and uses its upstream default dynamic - cutoff unchanged. Do not pass `-d`, `-n`, `-e`, or `-f` in the primary arm. -- Both frontiers are conquered by the same solver and resource policy. Report - preprocessing, cubing, and conquer work/span separately and end to end. +## Generate instances -Machine-specific Slurm wrappers are intentionally not versioned. Build the -frontiers with `cnc_cuber`, `calibrate_cc_difficulty.py`, and upstream -`march_cu`, then run both arms through `conquer_parallel.py`. Rejected -static/product cutoff workflows are not maintained. +```sh +PYTHONPATH=. python3 -m benchmarks.cnc.factoring \ + --width 26 --width 28 --count 10 \ + --out-dir artifacts/factoring +``` + +Each manifest row points to `instance.circuitsat.json` and `instance.cnf` with +matching SAT/UNSAT metadata and hashes. + +## Direct Kissat + +```sh +PYTHONPATH=. python3 -m benchmarks.cnc.solve INSTANCE.cnf \ + --kissat cnc-tools/bin/kissat \ + --timeout-s 600 --out-dir artifacts/direct +``` + +## march_cu then parallel Kissat + +```sh +PYTHONPATH=. python3 -m benchmarks.cnc.cubing march INSTANCE.cnf \ + --march-cu cnc-tools/bin/march_cu \ + --kissat cnc-tools/bin/kissat --workers 32 \ + --out-dir artifacts/march +``` + +Pass `--remaining-vars N` to override `march_cu`'s dynamic cutoff. -Each measurement bundle is a directory containing a hash-linked `bundle.json`, -the input DIMACS file, a frontier JSONL, a monotonic event JSONL, per-cube raw -result JSONL, and a SAT witness when applicable. The verifier treats those raw -records as authoritative and independently reconstructs: +## Rust solver: export all cubes + +```sh +cargo build --release --bin cnc_cuber +target/release/cnc_cuber INSTANCE.circuitsat.json \ + --cc-threshold 65536 -o artifacts/project/frontier.icnf +``` -- complete, non-overlapping frontier coverage; -- every cube's solved, cancelled, timed-out, or never-started lifecycle; -- cubing wall/CPU time, conquer CPU work and scheduled makespan; -- orchestration and end-to-end wall time; -- maximum worker concurrency and the aggregate verdict; -- input, tool, and executable provenance plus SAT witness validity. +This mode finishes the whole cubing traversal and never starts Kissat. + +## Rust solver: streaming Cube-and-Conquer + +```sh +target/release/cnc_cuber INSTANCE.circuitsat.json \ + --cc-threshold 65536 \ + --solve-cnf INSTANCE.cnf \ + --kissat cnc-tools/bin/kissat --workers 32 +``` -Run the positive fixture and the generated negative controls from the -repository root: +The Rust cuber submits each open leaf immediately to a bounded worker pool. +Exit codes follow SAT conventions: `10` for SAT and `20` for UNSAT. -```bash -python3 benchmarks/cnc/verify_measurements.py \ - --bundle tests/fixtures/cnc/measurement-valid +## Analyze a frozen frontier -python3 -m unittest tests/test_cnc_measurements.py +```sh +PYTHONPATH=. python3 -m benchmarks.cnc.cubing frontier \ + INSTANCE.cnf artifacts/project/frontier.icnf \ + --kissat cnc-tools/bin/kissat --workers 32 \ + --out-dir artifacts/project-analysis ``` -The exhaustive frontier representation is intended for small audit fixtures. -Large production runs should add a branching-tree certificate before using -this format beyond a tractable number of frontier variables. +The Python path never invokes the project cuber. It consumes a complete, +frozen frontier and records per-cube timing, decisions, conflicts, and aggregate +statistics. March and project frontiers can therefore use the same analysis +backend. diff --git a/benchmarks/cnc/calibrate_cc_difficulty.py b/benchmarks/cnc/calibrate_cc_difficulty.py deleted file mode 100644 index a4a44e9..0000000 --- a/benchmarks/cnc/calibrate_cc_difficulty.py +++ /dev/null @@ -1,214 +0,0 @@ -#!/usr/bin/env python3 -"""Calibrate the classical online CC difficulty cutoff by emitted task count.""" - -from __future__ import annotations - -import argparse -import hashlib -import json -import math -import re -import resource -import subprocess -import time -from pathlib import Path -from typing import Any - - -_STATS = re.compile(r"status=OK cubes=(\d+).*cutoff=CcDifficulty\((\d+)\)") - - -class CalibrationError(ValueError): - pass - - -def sha256_file(path: Path) -> str: - digest = hashlib.sha256() - with path.open("rb") as stream: - while chunk := stream.read(1024 * 1024): - digest.update(chunk) - return digest.hexdigest() - - -def task_lines(path: Path) -> int: - return sum( - line.startswith("a ") or line == "a 0\n" - for line in path.read_text(encoding="utf-8").splitlines(keepends=True) - ) - - -def run_cuber( - cuber: Path, - instance: Path, - threshold: int, - cubes: Path, - log: Path, - max_rows: int, - trace: Path | None = None, -) -> dict[str, int | float]: - command = [ - str(cuber.resolve()), - str(instance.resolve()), - "--cc-threshold", - str(threshold), - "-o", - str(cubes), - "--max-rows", - str(max_rows), - ] - if trace: - command.extend(("--trace", str(trace))) - before = resource.getrusage(resource.RUSAGE_CHILDREN) - started = time.monotonic() - process = subprocess.run(command, capture_output=True, text=True, check=False) - elapsed = time.monotonic() - started - after = resource.getrusage(resource.RUSAGE_CHILDREN) - log.write_text(process.stderr, encoding="utf-8") - match = _STATS.search(process.stderr) - if process.returncode or not match or int(match.group(2)) != threshold: - raise CalibrationError( - f"cuber threshold={threshold} failed: {process.stderr[-500:]}" - ) - tasks = int(match.group(1)) - if task_lines(cubes) != tasks: - raise CalibrationError("reported and emitted task counts differ") - return { - "threshold": threshold, - "tasks": tasks, - "elapsed_s": elapsed, - "user_s": after.ru_utime - before.ru_utime, - "system_s": after.ru_stime - before.ru_stime, - } - - -def choose( - rows: list[dict[str, int | float]], target: int, minimum: int, maximum: int -) -> dict[str, int | float]: - inside = [row for row in rows if minimum <= int(row["tasks"]) <= maximum] - pool = inside or rows - if not pool: - raise CalibrationError("empty cutoff response") - return min( - pool, - key=lambda row: ( - abs(math.log2(max(1, int(row["tasks"])) / target)), - abs(int(row["tasks"]) - target), - int(row["threshold"]), - ), - ) - - -def calibrate( - instance: Path, - cuber: Path, - out_dir: Path, - target: int, - minimum: int, - maximum: int, - initial: int, - maximum_threshold: int, - max_rows: int, -) -> dict[str, Any]: - if target <= 0 or minimum <= 0 or maximum < minimum or initial <= 0: - raise CalibrationError("invalid task range or threshold") - out_dir.mkdir(parents=True, exist_ok=True) - candidates = out_dir / "candidates" - candidates.mkdir(exist_ok=True) - observed: dict[int, dict[str, int | float]] = {} - - def probe(threshold: int) -> dict[str, int | float]: - if threshold not in observed: - observed[threshold] = run_cuber( - cuber, - instance, - threshold, - candidates / f"threshold-{threshold}.icnf", - candidates / f"threshold-{threshold}.log", - max_rows, - ) - return observed[threshold] - - lower, upper = 0, initial - probe(lower) - while int(probe(upper)["tasks"]) < target and upper < maximum_threshold: - lower, upper = upper, min(upper * 2, maximum_threshold) - if int(probe(upper)["tasks"]) < target: - raise CalibrationError("maximum threshold did not reach target task count") - while upper - lower > 1: - middle = (lower + upper) // 2 - if int(probe(middle)["tasks"]) >= target: - upper = middle - else: - lower = middle - - response = sorted(observed.values(), key=lambda row: int(row["threshold"])) - selected = choose(response, target, minimum, maximum) - threshold = int(selected["threshold"]) - final = run_cuber( - cuber, - instance, - threshold, - out_dir / "frontier.icnf", - out_dir / "final.log", - max_rows, - out_dir / "nodes.jsonl", - ) - candidate = candidates / f"threshold-{threshold}.icnf" - if sha256_file(candidate) != sha256_file(out_dir / "frontier.icnf"): - raise CalibrationError("traced rerun changed frontier bytes") - record = { - "schema_version": 1, - "method": "classical-cc-difficulty-task-count-calibration", - "formula": "D^2*(D+I)/N > threshold", - "instance": str(instance), - "instance_sha256": sha256_file(instance), - "cuber_sha256": sha256_file(cuber), - "target_tasks": target, - "accepted_task_range": [minimum, maximum], - "selected_threshold": threshold, - "tasks": int(final["tasks"]), - "cubing_elapsed_s": float(final["elapsed_s"]), - "cubing_cpu_s": float(final["user_s"]) + float(final["system_s"]), - "frontier_sha256": sha256_file(out_dir / "frontier.icnf"), - "trace_sha256": sha256_file(out_dir / "nodes.jsonl"), - "within_target_range": minimum <= int(final["tasks"]) <= maximum, - "response": response, - } - (out_dir / "selection.json").write_text( - json.dumps(record, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) - return record - - -def main() -> int: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("instance", type=Path) - parser.add_argument("--cuber", type=Path, required=True) - parser.add_argument("--out-dir", type=Path, required=True) - parser.add_argument("--target", type=int, default=512) - parser.add_argument("--min-tasks", type=int, default=384) - parser.add_argument("--max-tasks", type=int, default=640) - parser.add_argument("--initial-threshold", type=int, default=1024) - parser.add_argument("--max-threshold", type=int, default=1 << 60) - parser.add_argument("--max-rows", type=int, default=512) - args = parser.parse_args() - try: - result = calibrate( - args.instance, - args.cuber, - args.out_dir, - args.target, - args.min_tasks, - args.max_tasks, - args.initial_threshold, - args.max_threshold, - args.max_rows, - ) - except (CalibrationError, OSError) as exc: - parser.error(str(exc)) - print(json.dumps(result, indent=2, sort_keys=True)) - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/benchmarks/cnc/conquer_parallel.py b/benchmarks/cnc/conquer_parallel.py index 68c68da..1c55add 100644 --- a/benchmarks/cnc/conquer_parallel.py +++ b/benchmarks/cnc/conquer_parallel.py @@ -3,8 +3,8 @@ from __future__ import annotations -import argparse import concurrent.futures +import hashlib import json import math import os @@ -20,7 +20,8 @@ _CNF_HEADER = re.compile(rb"^p cnf (\d+) (\d+)\s*$", re.MULTILINE) _STAT = re.compile(r"^c\s+(decisions|conflicts):\s+(\d+)", re.MULTILINE) -_BASE_HEADER: bytes +_BASE_VARIABLES: int +_BASE_CLAUSES: int _BASE_BODY: bytes _KISSAT: str _TIMEOUT_S: float @@ -36,7 +37,7 @@ def parse_cnf(data: bytes) -> tuple[int, int, bytes]: return variables, clauses, body -def read_cubes(path: Path) -> Iterator[list[int]]: +def read_cubes(path: Path, variables: int | None = None) -> Iterator[list[int]]: with path.open(encoding="utf-8") as stream: for lineno, line in enumerate(stream, 1): fields = line.split() @@ -47,6 +48,12 @@ def read_cubes(path: Path) -> Iterator[list[int]]: literals = [int(field) for field in fields[1:-1]] if any(literal == 0 for literal in literals): raise ValueError(f"{path}:{lineno}: embedded zero literal") + if variables is not None and any(abs(literal) > variables for literal in literals): + raise ValueError(f"{path}:{lineno}: literal exceeds CNF variable range") + if len(set(literals)) != len(literals): + raise ValueError(f"{path}:{lineno}: duplicate literal") + if any(-literal in literals for literal in literals): + raise ValueError(f"{path}:{lineno}: contradictory literals") yield literals @@ -62,25 +69,17 @@ def child_cpu_seconds(before: resource.struct_rusage) -> tuple[float, float]: return after.ru_utime - before.ru_utime, after.ru_stime - before.ru_stime -def percentile(values: list[float], quantile: float) -> float: - if not values: - return math.nan - ordered = sorted(values) - index = min(len(ordered) - 1, max(0, round(quantile * (len(ordered) - 1)))) - return ordered[index] - - -def distribution(values: list[float]) -> dict[str, float]: +def distribution(values: list[float]) -> dict[str, float | None]: if not values: return { "total": 0.0, - "mean": math.nan, - "cv": math.nan, - "p50": math.nan, - "p95": math.nan, - "p99": math.nan, - "p99_over_p95": math.nan, - "max": math.nan, + "mean": None, + "cv": None, + "p50": None, + "p95": None, + "p99": None, + "p99_over_p95": None, + "max": None, } ordered = sorted(values) @@ -99,11 +98,11 @@ def ordered_percentile(quantile: float) -> float: return { "total": total, "mean": mean, - "cv": math.sqrt(variance) / mean if mean else math.nan, + "cv": math.sqrt(variance) / mean if mean else None, "p50": ordered_percentile(0.50), "p95": p95, "p99": p99, - "p99_over_p95": p99 / p95 if p95 > 0 else math.nan, + "p99_over_p95": p99 / p95 if p95 > 0 else None, "max": ordered[-1], } @@ -124,46 +123,57 @@ def _configure_worker( timeout_s: float, tmpdir: str | None, ) -> None: - global _BASE_HEADER, _BASE_BODY, _KISSAT, _TIMEOUT_S, _TMPDIR - _BASE_HEADER = f"p cnf {variables} {clauses}".encode() + global _BASE_VARIABLES, _BASE_CLAUSES, _BASE_BODY, _KISSAT, _TIMEOUT_S, _TMPDIR + _BASE_VARIABLES = variables + _BASE_CLAUSES = clauses _BASE_BODY = body _KISSAT = kissat _TIMEOUT_S = timeout_s _TMPDIR = tmpdir -def _solve_cube(task: tuple[str, int, list[int]]) -> dict[str, Any]: - arm, index, cube = task - header_fields = _BASE_HEADER.split() - clause_count = int(header_fields[3]) + len(cube) - header = b" ".join((*header_fields[:3], str(clause_count).encode())) + b"\n" - units = b"".join(f"{literal} 0\n".encode() for literal in cube) - payload = header + _BASE_BODY + units +def _solve_cube(task: tuple[str, int, list[int], int]) -> dict[str, Any]: + arm, index, cube, released_ns = task started_ns = time.monotonic_ns() + common = { + "schema_version": 1, + "arm": arm, + "cube_index": index, + "cube_literals": len(cube), + "cube_sha256": hashlib.sha256( + (" ".join(map(str, cube)) + " 0\n").encode() + ).hexdigest(), + "released_monotonic_ns": released_ns, + "started_monotonic_ns": started_ns, + "worker_pid": os.getpid(), + } temporary = tempfile.NamedTemporaryFile( prefix=f"cube-{arm}-{index}-", suffix=".cnf", dir=_TMPDIR, delete=False ) try: with temporary: - temporary.write(payload) + temporary.write( + f"p cnf {_BASE_VARIABLES} {_BASE_CLAUSES + len(cube)}\n".encode() + ) + temporary.write(_BASE_BODY) + for literal in cube: + temporary.write(f"{literal} 0\n".encode()) usage_before = resource.getrusage(resource.RUSAGE_CHILDREN) try: process = subprocess.run( [_KISSAT, "--statistics", "--relaxed", temporary.name], capture_output=True, text=True, - timeout=_TIMEOUT_S, + timeout=None if _TIMEOUT_S == 0 else _TIMEOUT_S, check=False, ) elapsed_s = (time.monotonic_ns() - started_ns) / 1e9 + finished_ns = time.monotonic_ns() user_s, system_s = child_cpu_seconds(usage_before) decisions, conflicts = parse_stats(process.stdout) result = {10: "sat", 20: "unsat"}.get(process.returncode, "error") return { - "schema_version": 1, - "arm": arm, - "cube_index": index, - "cube_literals": len(cube), + **common, "result": result, "returncode": process.returncode, "elapsed_s": elapsed_s, @@ -172,23 +182,23 @@ def _solve_cube(task: tuple[str, int, list[int]]) -> dict[str, Any]: "decisions": decisions, "conflicts": conflicts, "censored": False, + "finished_monotonic_ns": finished_ns, "stderr_tail": process.stderr[-500:] if result == "error" else "", } except subprocess.TimeoutExpired: + finished_ns = time.monotonic_ns() user_s, system_s = child_cpu_seconds(usage_before) return { - "schema_version": 1, - "arm": arm, - "cube_index": index, - "cube_literals": len(cube), + **common, "result": "timeout", "returncode": None, - "elapsed_s": (time.monotonic_ns() - started_ns) / 1e9, + "elapsed_s": (finished_ns - started_ns) / 1e9, "user_s": user_s, "system_s": system_s, "decisions": None, "conflicts": None, "censored": True, + "finished_monotonic_ns": finished_ns, "stderr_tail": "", } finally: @@ -208,25 +218,53 @@ def summarize( decisions: list[float], conflicts: list[float], workers: int, + replay_workers: list[int], wall_s: float, + measured_makespan_s: float, + not_started: int = 0, ) -> dict[str, Any]: time_stats = distribution(durations) cpu_stats = distribution(cpu_durations) decision_stats = distribution(decisions) conflict_stats = distribution(conflicts) + result = ( + "sat" + if sat + else "error" + if errors + else "timeout" + if timeouts + else "unsat" + if unsat == cubes + else "incomplete" + ) + complete = result == "sat" or (result == "unsat" and completed == cubes) + lpt_wall = {str(count): lpt_makespan(durations, count) for count in replay_workers} + lpt_cpu = { + str(count): lpt_makespan(cpu_durations, count) for count in replay_workers + } return { "cubes": cubes, + "terminal_records": completed + timeouts + errors, "completed": completed, "timeouts": timeouts, "errors": errors, + "not_started": not_started, "sat": sat, "unsat": unsat, + "result": result, + "complete": complete, + "censored": bool(timeouts), "total_solver_s": time_stats["total"], "total_cpu_s": cpu_stats["total"], "cpu_time": cpu_stats, "observed_parallel_wall_s": wall_s, + "measured_makespan_s": measured_makespan_s, "lpt_makespan_s": lpt_makespan(durations, workers), "cpu_lpt_makespan_s": lpt_makespan(cpu_durations, workers), + "lpt_makespan_by_workers_s": lpt_wall, + "cpu_lpt_makespan_by_workers_s": lpt_cpu, + "lpt_is_lower_bound": bool(timeouts or errors or not_started), "p50_s": time_stats["p50"], "p95_s": time_stats["p95"], "p99_s": time_stats["p99"], @@ -249,6 +287,7 @@ def run_arm( cubes: Iterator[list[int]], total_cubes: int, workers: int, + replay_workers: list[int], output: Path, worker_args: tuple[Any, ...], ) -> dict[str, Any]: @@ -259,6 +298,8 @@ def run_arm( counts = {"completed": 0, "timeouts": 0, "errors": 0, "sat": 0, "unsat": 0} progress_every = 10_000 if total_cubes > 10_000 else 200 started = time.monotonic() + earliest_release_ns: int | None = None + latest_collection_ns: int | None = None with output.open("w", encoding="utf-8", buffering=1) as stream: with concurrent.futures.ProcessPoolExecutor( max_workers=workers, @@ -273,10 +314,12 @@ def submit_one() -> bool: index, cube = next(indexed_cubes) except StopIteration: return False - pending.add(pool.submit(_solve_cube, (arm, index, cube))) + pending.add( + pool.submit(_solve_cube, (arm, index, cube, time.monotonic_ns())) + ) return True - for _ in range(workers * 4): + for _ in range(workers): if not submit_one(): break @@ -288,9 +331,27 @@ def submit_one() -> bool: for future in done: pending.remove(future) row = future.result() - row["completed_monotonic_ns"] = time.monotonic_ns() + row["collected_monotonic_ns"] = time.monotonic_ns() stream.write(json.dumps(row, sort_keys=True) + "\n") + row_released = int(row["released_monotonic_ns"]) + row_collected = int(row["collected_monotonic_ns"]) + earliest_release_ns = ( + row_released + if earliest_release_ns is None + else min(earliest_release_ns, row_released) + ) + latest_collection_ns = ( + row_collected + if latest_collection_ns is None + else max(latest_collection_ns, row_collected) + ) done_count += 1 + durations.append(float(row["elapsed_s"])) + cpu_durations.append(float(row["user_s"]) + float(row["system_s"])) + if row["decisions"] is not None: + decisions.append(float(row["decisions"])) + if row["conflicts"] is not None: + conflicts.append(float(row["conflicts"])) if row["censored"]: counts["timeouts"] += 1 elif row["result"] == "error": @@ -298,14 +359,11 @@ def submit_one() -> bool: else: counts["completed"] += 1 counts[row["result"]] += 1 - durations.append(float(row["elapsed_s"])) - cpu_durations.append(float(row["user_s"]) + float(row["system_s"])) - decisions.append(float(row["decisions"])) - conflicts.append(float(row["conflicts"])) - submit_one() + if not counts["sat"]: + submit_one() if done_count % progress_every == 0 or done_count == total_cubes: print(f"{arm}: {done_count}/{total_cubes}", flush=True) - if done_count != total_cubes: + if done_count != total_cubes and not counts["sat"]: raise RuntimeError(f"{arm}: expected {total_cubes} cubes, completed {done_count}") return summarize( cubes=total_cubes, @@ -314,65 +372,13 @@ def submit_one() -> bool: decisions=decisions, conflicts=conflicts, workers=workers, + replay_workers=replay_workers, wall_s=time.monotonic() - started, + measured_makespan_s=( + 0.0 + if earliest_release_ns is None or latest_collection_ns is None + else (latest_collection_ns - earliest_release_ns) / 1e9 + ), + not_started=total_cubes - done_count, **counts, ) - - -def main() -> None: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("cnf", type=Path) - parser.add_argument("--arm", action="append", required=True, metavar="NAME=CUBES") - parser.add_argument("--kissat", type=Path, required=True) - parser.add_argument("--workers", type=int, required=True) - parser.add_argument("--timeout-s", type=float, default=600.0) - parser.add_argument("--out-dir", type=Path, required=True) - parser.add_argument("--tmp-dir", type=Path) - args = parser.parse_args() - if args.workers < 1 or args.timeout_s <= 0: - parser.error("workers and timeout-s must be positive") - - variables, clauses, body = parse_cnf(args.cnf.read_bytes()) - args.out_dir.mkdir(parents=True, exist_ok=True) - if args.tmp_dir: - args.tmp_dir.mkdir(parents=True, exist_ok=True) - worker_args = ( - variables, - clauses, - body, - str(args.kissat.resolve()), - args.timeout_s, - str(args.tmp_dir.resolve()) if args.tmp_dir else None, - ) - summaries: dict[str, Any] = {} - for spec in args.arm: - try: - arm, cube_path = spec.split("=", 1) - except ValueError as error: - raise SystemExit(f"invalid --arm {spec!r}; expected NAME=PATH") from error - cube_file = Path(cube_path) - total_cubes = sum(1 for _ in read_cubes(cube_file)) - summaries[arm] = run_arm( - arm, - read_cubes(cube_file), - total_cubes, - args.workers, - args.out_dir / f"{arm}.jsonl", - worker_args, - ) - bundle = { - "schema_version": 1, - "workers": args.workers, - "timeout_s": args.timeout_s, - "cnf": str(args.cnf), - "kissat": str(args.kissat), - "arms": summaries, - } - (args.out_dir / "summary.json").write_text( - json.dumps(bundle, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) - print(json.dumps(bundle, indent=2, sort_keys=True)) - - -if __name__ == "__main__": - main() diff --git a/benchmarks/cnc/cubing.py b/benchmarks/cnc/cubing.py new file mode 100644 index 0000000..83d156b --- /dev/null +++ b/benchmarks/cnc/cubing.py @@ -0,0 +1,173 @@ +#!/usr/bin/env python3 +"""Run March or analyze a frozen cube frontier with parallel Kissat.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any + +from benchmarks.cnc.conquer_parallel import read_cubes +from benchmarks.cnc.solve import conquer_frontier, run_process +from benchmarks.pipeline.circuit import sha256_file, write_json + + +def run_cuber( + command: list[str], + frontier: Path, + *, + timeout_s: float, + out_dir: Path, +) -> dict[str, Any]: + frontier.parent.mkdir(parents=True, exist_ok=True) + process = run_process( + command, + timeout_s=timeout_s, + out_dir=out_dir, + label="cubing", + ) + if process["timed_out"]: + raise TimeoutError("cuber timed out") + if process["returncode"] != 0: + raise RuntimeError(f"cuber exited with status {process['returncode']}") + if not frontier.is_file(): + raise RuntimeError("cuber reported success without producing a frontier") + cubes = sum(1 for _ in read_cubes(frontier)) + record = { + "command": command, + "returncode": process["returncode"], + "wall_s": process["wall_s"], + "cubes": cubes, + "frontier": str(frontier.resolve()), + "frontier_sha256": sha256_file(frontier), + "frontier_bytes": frontier.stat().st_size, + } + write_json(out_dir / "cubing.json", record) + return record + + +def march_then_conquer( + cnf: Path, + march_cu: Path, + kissat: Path, + *, + workers: int, + cube_timeout_s: float, + cubing_timeout_s: float, + out_dir: Path, + remaining_vars: int | None = None, + tmp_dir: Path | None = None, +) -> dict[str, Any]: + frontier = out_dir / "frontier.icnf" + command = [str(march_cu), str(cnf)] + if remaining_vars is not None: + command.extend(["-n", str(remaining_vars)]) + command.extend(["-o", str(frontier)]) + cubing = run_cuber( + command, + frontier, + timeout_s=cubing_timeout_s, + out_dir=out_dir, + ) + conquer = conquer_frontier( + cnf, + frontier, + kissat, + workers=workers, + timeout_s=cube_timeout_s, + out_dir=out_dir / "conquer", + tmp_dir=tmp_dir, + total_cubes=cubing["cubes"], + ) + record = {"schema_version": 1, "mode": "march-cu", "cubing": cubing, "conquer": conquer} + write_json(out_dir / "summary.json", record) + return record + + +def frozen_then_conquer( + cnf: Path, + frontier: Path, + kissat: Path, + *, + workers: int, + cube_timeout_s: float, + out_dir: Path, + tmp_dir: Path | None = None, +) -> dict[str, Any]: + cubes = sum(1 for _ in read_cubes(frontier)) + conquer = conquer_frontier( + cnf, + frontier, + kissat, + workers=workers, + timeout_s=cube_timeout_s, + out_dir=out_dir / "conquer", + tmp_dir=tmp_dir, + total_cubes=cubes, + ) + record = { + "schema_version": 1, + "mode": "frozen-frontier", + "cnf": str(cnf.resolve()), + "cnf_sha256": sha256_file(cnf), + "frontier": str(frontier.resolve()), + "frontier_sha256": sha256_file(frontier), + "cubes": cubes, + "conquer": conquer, + } + write_json(out_dir / "summary.json", record) + return record + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="mode", required=True) + march = subparsers.add_parser("march") + march.add_argument("cnf", type=Path) + march.add_argument("--march-cu", type=Path, required=True) + march.add_argument("--remaining-vars", type=int) + frozen = subparsers.add_parser("frontier") + frozen.add_argument("cnf", type=Path) + frozen.add_argument("frontier", type=Path) + for command in (march, frozen): + command.add_argument("--kissat", type=Path, required=True) + command.add_argument("--workers", type=int, required=True) + command.add_argument("--cube-timeout-s", type=float, default=600.0) + command.add_argument("--out-dir", type=Path, required=True) + command.add_argument("--tmp-dir", type=Path) + march.add_argument("--cubing-timeout-s", type=float, default=3600.0) + args = parser.parse_args() + if ( + args.workers < 1 + or args.cube_timeout_s < 0 + or getattr(args, "cubing_timeout_s", 0) < 0 + ): + parser.error("workers must be positive and timeouts non-negative") + if args.mode == "march": + record = march_then_conquer( + args.cnf, + args.march_cu, + args.kissat, + workers=args.workers, + cube_timeout_s=args.cube_timeout_s, + cubing_timeout_s=args.cubing_timeout_s, + out_dir=args.out_dir, + remaining_vars=args.remaining_vars, + tmp_dir=args.tmp_dir, + ) + else: + record = frozen_then_conquer( + args.cnf, + args.frontier, + args.kissat, + workers=args.workers, + cube_timeout_s=args.cube_timeout_s, + out_dir=args.out_dir, + tmp_dir=args.tmp_dir, + ) + print(json.dumps(record, indent=2, sort_keys=True, allow_nan=False)) + + +if __name__ == "__main__": + main() diff --git a/benchmarks/cnc/factoring.py b/benchmarks/cnc/factoring.py new file mode 100644 index 0000000..d4386c7 --- /dev/null +++ b/benchmarks/cnc/factoring.py @@ -0,0 +1,287 @@ +#!/usr/bin/env python3 +"""Generate n×n SAT/UNSAT factoring instances as CircuitSAT and DIMACS CNF.""" + +from __future__ import annotations + +import argparse +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Literal + +from benchmarks.pipeline.circuit import ( + pin_port_values, + sha256_file, + write_json, + write_jsonl, +) +from benchmarks.pipeline.cnf import encode_validated_circuit +from benchmarks.pipeline.multipliers import ( + DETERMINISTIC_MILLER_RABIN_LIMIT, + generate_multiplier, + is_prime, + records as factoring_records, +) + +Outcome = Literal["sat", "unsat"] +SAT_SEED_BASE = 20260709 +UNSAT_MATCH_CANDIDATES = 64 + + +@dataclass(frozen=True, slots=True) +class FactoringTarget: + instance_id: str + factor_bits: int + expected_outcome: Outcome + target: int + target_bits: int + seed: int + sequence_index: int + paired_sat_id: str | None + + +def sat_targets( + width: int, + count: int, + *, + seed_base: int = SAT_SEED_BASE, + instance_prefix: str = "factoring", +) -> tuple[list[FactoringTarget], list[dict]]: + if not 2 <= width <= 39 or count < 1: + raise ValueError("factor width must be between 2 and 39; count must be positive") + if width <= 12: + prime_count = sum( + is_prime(candidate) + for candidate in range(1 << (width - 1), 1 << width) + ) + distinct_products = prime_count * (prime_count + 1) // 2 + if count > distinct_products: + raise ValueError( + f"width {width} has only {distinct_products} distinct " + "balanced semiprime products" + ) + targets = [] + oracles = [] + for public, oracle in factoring_records([width], count, seed_base): + target = public["target"] + index = public["sequence_index"] + instance_id = f"{instance_prefix}-n{width}-sat-{index:02d}" + targets.append( + FactoringTarget( + instance_id=instance_id, + factor_bits=width, + expected_outcome="sat", + target=target, + target_bits=target.bit_length(), + seed=public["seed"], + sequence_index=index, + paired_sat_id=None, + ) + ) + oracles.append( + { + "instance_id": instance_id, + "expected_outcome": "sat", + "left_factor": oracle["left_factor"], + "right_factor": oracle["right_factor"], + } + ) + return targets, oracles + + +def unsat_targets( + width: int, + sat: list[FactoringTarget], + *, + instance_prefix: str = "factoring", +) -> tuple[list[FactoringTarget], list[dict]]: + factor_max = (1 << width) - 1 + reachable_max = factor_max * factor_max + targets = [] + oracles = [] + seen: set[int] = set() + for index, paired in enumerate(sat): + candidates = [] + delta = 2 + lower = 1 << (paired.target_bits - 1) + upper = min(1 << paired.target_bits, reachable_max + 1) + max_delta = max(paired.target - lower, upper - 1 - paired.target) + while len(candidates) < UNSAT_MATCH_CANDIDATES and delta <= max_delta: + for target in (paired.target - delta, paired.target + delta): + if ( + lower <= target < upper + and target > factor_max + and target not in seen + and target < DETERMINISTIC_MILLER_RABIN_LIMIT + and is_prime(target) + ): + candidates.append(target) + if len(candidates) == UNSAT_MATCH_CANDIDATES: + break + delta += 2 + if not candidates: + raise ValueError( + f"{paired.instance_id}: no distinct prime UNSAT target in range" + ) + target = min( + candidates, + key=lambda value: ( + (value ^ paired.target).bit_count(), + abs(value - paired.target), + value, + ), + ) + seen.add(target) + instance_id = f"{instance_prefix}-n{width}-unsat-{index:02d}" + targets.append( + FactoringTarget( + instance_id=instance_id, + factor_bits=width, + expected_outcome="unsat", + target=target, + target_bits=target.bit_length(), + seed=paired.seed, + sequence_index=index, + paired_sat_id=paired.instance_id, + ) + ) + oracles.append( + { + "instance_id": instance_id, + "expected_outcome": "unsat", + "target_is_prime": True, + "target_exceeds_max_factor": True, + "target_within_multiplier_range": True, + "paired_sat_id": paired.instance_id, + } + ) + return targets, oracles + + +def validate_targets(targets: list[FactoringTarget], oracles: list[dict]) -> None: + by_id = {target.instance_id: target for target in targets} + oracle_ids = [oracle.get("instance_id") for oracle in oracles] + if len(by_id) != len(targets) or len(oracle_ids) != len(set(oracle_ids)): + raise ValueError("duplicate factoring target or oracle id") + if set(oracle_ids) != set(by_id): + raise ValueError("factoring targets and oracles are not one-to-one") + for oracle in oracles: + target = by_id[oracle["instance_id"]] + if target.target_bits != target.target.bit_length(): + raise ValueError(f"{target.instance_id}: stale target bit length") + if target.expected_outcome == "sat": + left = int(oracle["left_factor"]) + right = int(oracle["right_factor"]) + if ( + target.paired_sat_id is not None + or left.bit_length() != target.factor_bits + or right.bit_length() != target.factor_bits + or left * right != target.target + ): + raise ValueError(f"{target.instance_id}: invalid SAT oracle") + elif target.expected_outcome == "unsat": + factor_max = (1 << target.factor_bits) - 1 + if ( + target.paired_sat_id not in by_id + or target.target <= factor_max + or target.target > factor_max * factor_max + or target.target >= DETERMINISTIC_MILLER_RABIN_LIMIT + or not is_prime(target.target) + ): + raise ValueError(f"{target.instance_id}: invalid UNSAT oracle") + else: + raise ValueError(f"{target.instance_id}: invalid expected outcome") + + +def materialize( + widths: list[int], + count: int, + out_dir: Path, + *, + seed_base: int = SAT_SEED_BASE, + instance_prefix: str = "factoring", +) -> list[dict]: + if not widths or len(widths) != len(set(widths)): + raise ValueError("factor widths must be non-empty and unique") + all_targets: list[FactoringTarget] = [] + all_oracles: list[dict] = [] + for width in widths: + sat, sat_oracles = sat_targets( + width, + count, + seed_base=seed_base, + instance_prefix=instance_prefix, + ) + unsat, unsat_oracles = unsat_targets( + width, + sat, + instance_prefix=instance_prefix, + ) + all_targets.extend((*sat, *unsat)) + all_oracles.extend((*sat_oracles, *unsat_oracles)) + validate_targets(all_targets, all_oracles) + + raw_by_width = { + width: generate_multiplier(width, "array-ripple") for width in widths + } + manifest = [] + for target in all_targets: + instance_dir = ( + out_dir + / f"n{target.factor_bits}" + / target.expected_outcome + / f"{target.sequence_index:02d}" + ) + circuit_path = instance_dir / "instance.circuitsat.json" + cnf_path = instance_dir / "instance.cnf" + metadata_path = instance_dir / "instance.meta.json" + circuit = pin_port_values( + raw_by_width[target.factor_bits], {"product": target.target} + ) + circuit.setdefault("metadata", {})["factoring"] = { + **asdict(target), + } + write_json(circuit_path, circuit) + encode_validated_circuit(circuit).write_dimacs(cnf_path) + metadata = { + **asdict(target), + "circuit": str(circuit_path.relative_to(out_dir)), + "circuit_sha256": sha256_file(circuit_path), + "cnf": str(cnf_path.relative_to(out_dir)), + "cnf_sha256": sha256_file(cnf_path), + } + write_json(metadata_path, metadata) + manifest.append( + { + **metadata, + "metadata": str(metadata_path.relative_to(out_dir)), + } + ) + write_jsonl(out_dir / "manifest.jsonl", manifest) + write_jsonl(out_dir / "oracles.jsonl", all_oracles) + return manifest + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--width", type=int, action="append", required=True) + parser.add_argument("--count", type=int, default=10) + parser.add_argument("--seed-base", type=int, default=SAT_SEED_BASE) + parser.add_argument("--instance-prefix", default="factoring") + parser.add_argument("--out-dir", type=Path, required=True) + args = parser.parse_args() + widths = sorted(args.width) + try: + manifest = materialize( + widths, + args.count, + args.out_dir, + seed_base=args.seed_base, + instance_prefix=args.instance_prefix, + ) + except ValueError as error: + parser.error(str(error)) + print(f"wrote {len(manifest)} CircuitSAT/CNF instance pairs to {args.out_dir}") + + +if __name__ == "__main__": + main() diff --git a/benchmarks/cnc/solve.py b/benchmarks/cnc/solve.py new file mode 100644 index 0000000..1ab3a3f --- /dev/null +++ b/benchmarks/cnc/solve.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 +"""Solve factoring CNFs directly with Kissat or conquer a cube frontier.""" + +from __future__ import annotations + +import argparse +import json +import resource +import subprocess +import time +from pathlib import Path +from typing import Any + +from benchmarks.cnc.conquer_parallel import ( + parse_cnf, + parse_stats, + read_cubes, + run_arm, +) +from benchmarks.pipeline.circuit import write_json + + +def _text(value: str | bytes | None) -> str: + if value is None: + return "" + return value.decode(errors="replace") if isinstance(value, bytes) else value + + +def run_process( + command: list[str], + *, + timeout_s: float, + out_dir: Path, + label: str, +) -> dict[str, Any]: + out_dir.mkdir(parents=True, exist_ok=True) + before = resource.getrusage(resource.RUSAGE_CHILDREN) + started = time.monotonic() + try: + process = subprocess.run( + command, + capture_output=True, + text=True, + timeout=None if timeout_s == 0 else timeout_s, + check=False, + ) + stdout = process.stdout + stderr = process.stderr + returncode = process.returncode + timed_out = False + except subprocess.TimeoutExpired as error: + stdout = _text(error.stdout) + stderr = _text(error.stderr) + returncode = None + timed_out = True + after = resource.getrusage(resource.RUSAGE_CHILDREN) + (out_dir / f"{label}.stdout").write_text(stdout, encoding="utf-8") + (out_dir / f"{label}.stderr").write_text(stderr, encoding="utf-8") + return { + "command": command, + "stdout": stdout, + "stderr": stderr, + "returncode": returncode, + "timed_out": timed_out, + "wall_s": time.monotonic() - started, + "user_s": after.ru_utime - before.ru_utime, + "system_s": after.ru_stime - before.ru_stime, + } + + +def run_kissat( + cnf: Path, + kissat: Path, + *, + timeout_s: float, + out_dir: Path, +) -> dict[str, Any]: + process = run_process( + [str(kissat), "--statistics", "--relaxed", str(cnf)], + timeout_s=timeout_s, + out_dir=out_dir, + label="kissat", + ) + decisions, conflicts = parse_stats(process["stdout"]) + result = ( + "timeout" + if process["timed_out"] + else {10: "sat", 20: "unsat"}.get(process["returncode"], "error") + ) + record = { + "schema_version": 1, + "mode": "direct-kissat", + "result": result, + "returncode": process["returncode"], + "timed_out": process["timed_out"], + "wall_s": process["wall_s"], + "user_s": process["user_s"], + "system_s": process["system_s"], + "decisions": decisions, + "conflicts": conflicts, + "cnf": str(cnf.resolve()), + "kissat": str(kissat.resolve()), + } + write_json(out_dir / "summary.json", record) + return record + + +def conquer_frontier( + cnf: Path, + frontier: Path, + kissat: Path, + *, + workers: int, + timeout_s: float, + out_dir: Path, + tmp_dir: Path | None = None, + total_cubes: int | None = None, +) -> dict[str, Any]: + variables, clauses, body = parse_cnf(cnf.read_bytes()) + out_dir.mkdir(parents=True, exist_ok=True) + if tmp_dir: + tmp_dir.mkdir(parents=True, exist_ok=True) + if total_cubes is None: + total_cubes = sum(1 for _ in read_cubes(frontier, variables)) + summary = run_arm( + "factoring", + read_cubes(frontier, variables), + total_cubes, + workers, + [workers], + out_dir / "cubes.jsonl", + ( + variables, + clauses, + body, + str(kissat.resolve()), + timeout_s, + str(tmp_dir.resolve()) if tmp_dir else None, + ), + ) + record = { + "schema_version": 1, + "mode": "parallel-conquer", + "cnf": str(cnf.resolve()), + "frontier": str(frontier.resolve()), + "kissat": str(kissat.resolve()), + "workers": workers, + "per_cube_timeout_s": timeout_s, + **summary, + } + write_json(out_dir / "summary.json", record) + return record + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("cnf", type=Path) + parser.add_argument("--kissat", type=Path, required=True) + parser.add_argument("--timeout-s", type=float, default=600.0) + parser.add_argument("--out-dir", type=Path, required=True) + args = parser.parse_args() + if args.timeout_s < 0: + parser.error("timeout must be non-negative") + record = run_kissat( + args.cnf, + args.kissat, + timeout_s=args.timeout_s, + out_dir=args.out_dir, + ) + print(json.dumps(record, indent=2, sort_keys=True, allow_nan=False)) + + +if __name__ == "__main__": + main() diff --git a/benchmarks/cnc/verify_measurements.py b/benchmarks/cnc/verify_measurements.py deleted file mode 100644 index 47932db..0000000 --- a/benchmarks/cnc/verify_measurements.py +++ /dev/null @@ -1,460 +0,0 @@ -#!/usr/bin/env python3 -"""Verify a self-contained Cube-and-Conquer measurement evidence bundle.""" - -from __future__ import annotations - -import argparse -import hashlib -import itertools -import json -import re -from collections import defaultdict -from pathlib import Path - - -class BundleError(ValueError): - """The bundle is incomplete, inconsistent, or unauditable.""" - - -def load_json(path: Path) -> dict: - try: - value = json.loads(path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError) as exc: - raise BundleError(f"cannot read {path}: {exc}") from exc - if not isinstance(value, dict): - raise BundleError(f"{path}: expected a JSON object") - return value - - -def read_jsonl(path: Path) -> list[dict]: - records = [] - for line_number, line in enumerate( - path.read_text(encoding="utf-8").splitlines(), 1 - ): - if not line.strip(): - continue - value = json.loads(line) - if not isinstance(value, dict): - raise BundleError(f"{path}:{line_number}: expected a JSON object") - records.append(value) - return records - - -def bundle_path(root: Path, relative: object) -> Path: - if not isinstance(relative, str) or not relative: - raise BundleError("bundle artifact path must be a non-empty string") - root = root.resolve() - path = (root / relative).resolve() - if path != root and root not in path.parents: - raise BundleError(f"bundle artifact escapes its root: {relative}") - return path - - -def sha256_file(path: Path) -> str: - digest = hashlib.sha256() - with path.open("rb") as stream: - while chunk := stream.read(1024 * 1024): - digest.update(chunk) - return digest.hexdigest() - - -def checked_artifact(root: Path, spec: object, label: str) -> Path: - if not isinstance(spec, dict): - raise BundleError(f"manifest.{label} must be an object") - path = bundle_path(root, spec.get("path")) - expected = spec.get("sha256") - if not isinstance(expected, str) or not re.fullmatch(r"[0-9a-f]{64}", expected): - raise BundleError(f"manifest.{label}.sha256 must be a lowercase SHA-256") - if sha256_file(path) != expected: - raise BundleError(f"{label} SHA-256 mismatch") - return path - - -def parse_dimacs(path: Path) -> tuple[int, list[list[int]]]: - variables = None - clauses: list[list[int]] = [] - pending: list[int] = [] - for line in path.read_text(encoding="utf-8").splitlines(): - stripped = line.strip() - if not stripped or stripped.startswith("c"): - continue - if stripped.startswith("p "): - fields = stripped.split() - if len(fields) != 4 or fields[:2] != ["p", "cnf"]: - raise BundleError(f"{path}: malformed DIMACS header") - variables = int(fields[2]) - declared_clauses = int(fields[3]) - continue - for token in stripped.split(): - literal = int(token) - if literal == 0: - clauses.append(pending) - pending = [] - else: - pending.append(literal) - if variables is None or pending or len(clauses) != declared_clauses: - raise BundleError(f"{path}: malformed DIMACS body") - if any(abs(literal) > variables for clause in clauses for literal in clause): - raise BundleError(f"{path}: literal exceeds declared variable count") - return variables, clauses - - -def parse_cube(record: dict, variables: list[int]) -> tuple[str, dict[int, bool]]: - cube_id = record.get("cube_id") - literals = record.get("literals") - if not isinstance(cube_id, str) or not cube_id or not isinstance(literals, list): - raise BundleError("every frontier record needs cube_id and literals") - assignment: dict[int, bool] = {} - for literal in literals: - if not isinstance(literal, int) or literal == 0 or abs(literal) not in variables: - raise BundleError(f"cube {cube_id}: invalid literal {literal!r}") - variable = abs(literal) - value = literal > 0 - if variable in assignment and assignment[variable] != value: - raise BundleError(f"cube {cube_id}: contradictory literal for {variable}") - assignment[variable] = value - return cube_id, assignment - - -def verify_frontier(records: list[dict], variables: list[int]) -> tuple[int, int]: - cubes: list[tuple[str, dict[int, bool]]] = [] - ids: set[str] = set() - for record in records: - cube_id, cube = parse_cube(record, variables) - if cube_id in ids: - raise BundleError(f"duplicate cube id {cube_id!r}") - ids.add(cube_id) - cubes.append((cube_id, cube)) - total = 1 << len(variables) - covered = 0 - for bits in itertools.product((False, True), repeat=len(variables)): - assignment = dict(zip(variables, bits, strict=True)) - owners = [ - cube_id - for cube_id, cube in cubes - if all(assignment[var] == value for var, value in cube.items()) - ] - label = "".join("1" if bit else "0" for bit in bits) - if not owners: - raise BundleError(f"frontier: assignment {label} is uncovered") - if len(owners) > 1: - raise BundleError( - f"frontier: assignment {label} is covered by {len(owners)} cubes" - ) - covered += 1 - return covered, total - - -def solver_verdict(raw_output: object) -> str | None: - if not isinstance(raw_output, str): - raise BundleError("raw_solver_output must be a string") - statuses = { - line.strip().upper() - for line in raw_output.splitlines() - if line.strip().upper().startswith("S ") - } - mapping = { - "S SATISFIABLE": "sat", - "S UNSATISFIABLE": "unsat", - "S UNKNOWN": "unknown", - } - parsed = {mapping[status] for status in statuses if status in mapping} - if len(parsed) > 1: - raise BundleError("raw solver output contains contradictory verdicts") - return next(iter(parsed), None) - - -def number(value: float) -> str: - return str(int(value)) if value.is_integer() else str(value) - - -def verify_events( - events: list[dict], results: list[dict], cube_ids: set[str], worker_count: int -) -> dict[str, float]: - if [event.get("seq") for event in events] != list(range(len(events))): - raise BundleError("events must have contiguous sequence numbers from zero") - times = [event.get("monotonic_seconds") for event in events] - if not all(isinstance(value, (int, float)) for value in times): - raise BundleError("every event needs a numeric monotonic_seconds") - if any(left > right for left, right in zip(times, times[1:])): - raise BundleError("event timestamps are not monotonic") - - by_kind: dict[str, list[dict]] = defaultdict(list) - for event in events: - kind = event.get("event") - if not isinstance(kind, str): - raise BundleError("every event needs an event name") - by_kind[kind].append(event) - for marker in ("run_started", "cubing_started", "cubing_finished", "run_finished"): - if len(by_kind[marker]) != 1: - raise BundleError(f"event log needs exactly one {marker}") - - result_by_id: dict[str, dict] = {} - for record in results: - cube_id = record.get("cube_id") - termination = record.get("termination") - verdict = record.get("verdict") - cpu_seconds = record.get("cpu_seconds") - if not isinstance(cube_id, str) or cube_id in result_by_id: - raise BundleError(f"duplicate or malformed result cube id {cube_id!r}") - if termination not in {"solved", "cancelled", "timed_out", "never_started"}: - raise BundleError(f"cube {cube_id}: invalid termination state") - if verdict not in {"sat", "unsat", "unknown", None}: - raise BundleError(f"cube {cube_id}: invalid verdict") - if not isinstance(cpu_seconds, (int, float)) or cpu_seconds < 0: - raise BundleError(f"cube {cube_id}: invalid cpu_seconds") - parsed = solver_verdict(record.get("raw_solver_output")) - if parsed != verdict: - raise BundleError(f"cube {cube_id}: raw output disagrees with verdict") - if termination == "solved" and verdict not in {"sat", "unsat"}: - raise BundleError(f"cube {cube_id}: solved cube needs SAT or UNSAT verdict") - if termination != "solved" and verdict not in {"unknown", None}: - raise BundleError(f"cube {cube_id}: non-solved cube has definitive verdict") - result_by_id[cube_id] = record - if set(result_by_id) != cube_ids: - missing = sorted(cube_ids - set(result_by_id)) - extra = sorted(set(result_by_id) - cube_ids) - raise BundleError(f"result coverage mismatch: missing={missing}, extra={extra}") - - started: dict[str, dict] = {} - terminal: dict[str, dict] = {} - intervals = [] - for event in events: - if event["event"] not in {"cube_started", "cube_terminal"}: - continue - cube_id = event.get("cube_id") - worker = event.get("worker") - if cube_id not in cube_ids or not isinstance(worker, int) or not 0 <= worker < worker_count: - raise BundleError("cube event has invalid cube_id or worker") - target = started if event["event"] == "cube_started" else terminal - if cube_id in target: - raise BundleError(f"cube {cube_id}: duplicate {event['event']}") - target[cube_id] = event - - for cube_id, result in result_by_id.items(): - termination = result["termination"] - if termination == "never_started": - if cube_id in started or cube_id in terminal: - raise BundleError(f"cube {cube_id}: never-started cube has events") - continue - if cube_id not in started or cube_id not in terminal: - raise BundleError(f"cube {cube_id}: missing lifecycle event") - begin = started[cube_id] - end = terminal[cube_id] - if begin["worker"] != end["worker"] or end.get("termination") != termination: - raise BundleError(f"cube {cube_id}: inconsistent terminal event") - if begin["monotonic_seconds"] >= end["monotonic_seconds"]: - raise BundleError(f"cube {cube_id}: terminal event must follow start") - intervals.append( - ( - begin["monotonic_seconds"], - end["monotonic_seconds"], - cube_id, - begin["worker"], - ) - ) - - by_worker: dict[int, list[tuple[float, float, str]]] = defaultdict(list) - for begin, end, cube_id, worker in intervals: - by_worker[worker].append((begin, end, cube_id)) - for worker, work in by_worker.items(): - previous_end = None - for begin, end, cube_id in sorted(work): - if previous_end is not None and begin < previous_end: - raise BundleError( - f"worker {worker} runs overlapping cube {cube_id}" - ) - previous_end = end - - points = [] - for begin, end, _, _ in intervals: - points.extend(((begin, 1), (end, -1))) - active = maximum = 0 - for _, delta in sorted(points, key=lambda item: (item[0], item[1])): - active += delta - if active < 0: - raise BundleError("worker interval accounting became negative") - maximum = max(maximum, active) - if maximum > worker_count: - raise BundleError( - f"observed concurrency {maximum} exceeds worker limit {worker_count}" - ) - - run_start = float(by_kind["run_started"][0]["monotonic_seconds"]) - run_end = float(by_kind["run_finished"][0]["monotonic_seconds"]) - cube_start = float(by_kind["cubing_started"][0]["monotonic_seconds"]) - cube_end = float(by_kind["cubing_finished"][0]["monotonic_seconds"]) - cpu_start = by_kind["cubing_started"][0].get("process_cpu_seconds") - cpu_end = by_kind["cubing_finished"][0].get("process_cpu_seconds") - if not isinstance(cpu_start, (int, float)) or not isinstance(cpu_end, (int, float)): - raise BundleError("cubing events need process_cpu_seconds counters") - if not run_start <= cube_start <= cube_end <= run_end or cpu_end < cpu_start: - raise BundleError("run or cubing counters are not properly nested") - conquer_start = min((begin for begin, _, _, _ in intervals), default=cube_end) - conquer_end = max((end for _, end, _, _ in intervals), default=cube_end) - if conquer_start < cube_end or conquer_end > run_end: - raise BundleError("conquer intervals fall outside the scheduled run phase") - metrics = { - "cubing_wall": cube_end - cube_start, - "cubing_cpu": float(cpu_end - cpu_start), - "conquer_cpu": float(sum(record["cpu_seconds"] for record in results)), - "conquer_makespan": float(conquer_end - conquer_start), - "end_to_end_wall": run_end - run_start, - } - metrics["orchestration"] = ( - metrics["end_to_end_wall"] - - metrics["cubing_wall"] - - metrics["conquer_makespan"] - ) - if any(value < 0 for value in metrics.values()): - raise BundleError("derived accounting contains a negative duration") - metrics["maximum_concurrency"] = float(maximum) - return metrics - - -def verify_model( - clauses: list[list[int]], model: list[int], variables: int -) -> dict[int, bool]: - assignment: dict[int, bool] = {} - for literal in model: - if not isinstance(literal, int) or literal == 0 or abs(literal) > variables: - raise BundleError(f"witness contains invalid literal {literal!r}") - variable = abs(literal) - value = literal > 0 - if variable in assignment and assignment[variable] != value: - raise BundleError(f"witness contradicts variable {variable}") - assignment[variable] = value - missing = set(range(1, variables + 1)) - assignment.keys() - if missing: - raise BundleError(f"witness omits variables: {sorted(missing)}") - if not all( - any(assignment[abs(literal)] == (literal > 0) for literal in clause) - for clause in clauses - ): - raise BundleError("returned model does not satisfy the input") - return assignment - - -def verify(bundle: Path) -> list[str]: - manifest = load_json(bundle / "bundle.json") - if manifest.get("format_version") != 1: - raise BundleError("unsupported bundle format_version") - worker_count = manifest.get("worker_count") - if not isinstance(worker_count, int) or worker_count < 1: - raise BundleError("worker_count must be positive") - for key in ("scheduler", "termination_policy"): - if not isinstance(manifest.get(key), str) or not manifest[key]: - raise BundleError(f"manifest needs {key}") - provenance = manifest.get("provenance") - if not isinstance(provenance, dict): - raise BundleError("manifest needs provenance") - for tool in ("cuber", "conquer"): - record = provenance.get(tool) - if not isinstance(record, dict) or not all( - isinstance(record.get(key), str) and record[key] - for key in ("id", "version", "path", "executable_sha256") - ): - raise BundleError(f"provenance needs complete {tool} identity") - if not re.fullmatch(r"[0-9a-f]{64}", record["executable_sha256"]): - raise BundleError(f"{tool} executable_sha256 is malformed") - executable = bundle_path(bundle, record["path"]) - if sha256_file(executable) != record["executable_sha256"]: - raise BundleError(f"{tool} executable SHA-256 mismatch") - - input_path = checked_artifact(bundle, manifest.get("input"), "input") - frontier_path = checked_artifact(bundle, manifest.get("frontier"), "frontier") - events_path = checked_artifact(bundle, manifest.get("events"), "events") - results_path = checked_artifact(bundle, manifest.get("results"), "results") - variables, clauses = parse_dimacs(input_path) - frontier_spec = manifest["frontier"] - frontier_variables = frontier_spec.get("variables") - if frontier_spec.get("mode") != "exhaustive" or not isinstance( - frontier_variables, list - ): - raise BundleError("frontier must declare exhaustive mode and variables") - if frontier_variables != list(range(1, variables + 1)): - raise BundleError("exhaustive frontier variables must match the input") - frontier = read_jsonl(frontier_path) - covered, total = verify_frontier(frontier, frontier_variables) - results = read_jsonl(results_path) - events = read_jsonl(events_path) - cube_ids = {record["cube_id"] for record in frontier} - metrics = verify_events(events, results, cube_ids, worker_count) - - declared = manifest.get("accounting") - if not isinstance(declared, dict): - raise BundleError("manifest needs declared accounting") - for key in ( - "cubing_wall", - "cubing_cpu", - "conquer_cpu", - "conquer_makespan", - "orchestration", - "end_to_end_wall", - ): - if declared.get(key) != metrics[key]: - raise BundleError( - f"accounting mismatch for {key}: declared={declared.get(key)!r}, " - f"derived={metrics[key]!r}" - ) - - verdicts = {record["verdict"] for record in results} - if "sat" in verdicts: - aggregate = "sat" - elif all(record["termination"] == "solved" for record in results) and verdicts == {"unsat"}: - aggregate = "unsat" - else: - aggregate = "unknown" - if manifest.get("aggregate_verdict") != aggregate: - raise BundleError("aggregate verdict disagrees with cube records") - verdict_message = "all cube outcomes justify the aggregate verdict" - if aggregate == "sat": - witness_path = checked_artifact(bundle, manifest.get("witness"), "witness") - witness = load_json(witness_path).get("model") - if not isinstance(witness, list): - raise BundleError("witness needs a model list") - model = verify_model(clauses, witness, variables) - model_owner = next( - ( - cube_id - for cube_id, cube in ( - parse_cube(record, frontier_variables) for record in frontier - ) - if all(model[variable] == value for variable, value in cube.items()) - ), - None, - ) - sat_cubes = { - record["cube_id"] for record in results if record["verdict"] == "sat" - } - if model_owner not in sat_cubes: - raise BundleError("returned model does not belong to a SAT cube") - verdict_message = "returned model satisfies the input" - - return [ - f"PASS frontier: {covered}/{total} assignments are covered exactly once", - f"PASS verdict: {verdict_message}", - "PASS accounting: " - f"cubing_wall={number(metrics['cubing_wall'])} " - f"conquer_cpu={number(metrics['conquer_cpu'])} " - f"conquer_makespan={number(metrics['conquer_makespan'])} " - f"end_to_end_wall={number(metrics['end_to_end_wall'])}", - f"PASS workers: observed concurrency does not exceed {worker_count}", - ] - - -def main() -> int: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--bundle", type=Path, required=True) - args = parser.parse_args() - try: - messages = verify(args.bundle) - except (BundleError, OSError, json.JSONDecodeError, ValueError) as exc: - print(f"FAIL {exc}") - return 1 - print("\n".join(messages)) - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/benchmarks/pipeline/circuit.py b/benchmarks/pipeline/circuit.py index d8d26e8..f8310f2 100644 --- a/benchmarks/pipeline/circuit.py +++ b/benchmarks/pipeline/circuit.py @@ -5,6 +5,8 @@ import copy import hashlib import json +import os +import tempfile from collections import defaultdict, deque from collections.abc import Iterable from pathlib import Path @@ -89,6 +91,21 @@ def write_json(path: Path, value: Any) -> None: ) +def atomic_write_json(path: Path, value: Any) -> None: + """Atomically replace *path* with a JSON document via a unique sibling file.""" + path.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp( + dir=path.parent, prefix=f".{path.name}.", suffix=".tmp" + ) + os.close(descriptor) + temporary = Path(temporary_name) + try: + write_json(temporary, value) + os.replace(temporary, path) + finally: + temporary.unlink(missing_ok=True) + + def write_jsonl(path: Path, values: Iterable[Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") as stream: diff --git a/benchmarks/pipeline/multipliers.py b/benchmarks/pipeline/multipliers.py index 87c248a..c8c4927 100644 --- a/benchmarks/pipeline/multipliers.py +++ b/benchmarks/pipeline/multipliers.py @@ -16,6 +16,9 @@ var, ) +DETERMINISTIC_MILLER_RABIN_LIMIT = 318_665_857_834_031_151_167_461 + + def is_prime(value: int) -> bool: if value < 2: return False @@ -28,8 +31,16 @@ def is_prime(value: int) -> bool: while odd % 2 == 0: odd //= 2 power += 1 - # Deterministic for unsigned 64-bit integers; still a strong fixed-base test above it. - for base in (2, 325, 9375, 28178, 450775, 9780504, 1795265022): + if value < (1 << 64): + bases = (2, 325, 9375, 28178, 450775, 9780504, 1795265022) + else: + # Deterministic below DETERMINISTIC_MILLER_RABIN_LIMIT. This covers + # every product of two factors up to 39 bits, including the 34-bit + # scale-extension corpus. Above the bound it remains a strong + # fixed-base probable-prime test, so scientific corpus generators + # must reject values outside the certified range. + bases = small + for base in bases: if base % value == 0: continue witness = pow(base, odd, value) diff --git a/cnc-tools/Makefile b/cnc-tools/Makefile index 8bf447d..f1016d2 100644 --- a/cnc-tools/Makefile +++ b/cnc-tools/Makefile @@ -5,12 +5,10 @@ # make clean remove temporary build trees (keeps bin/ + proofix/) # make distclean remove bin/ and proofix/ too (full reset) # -# Idempotent: a tool already present in bin/ is left untouched (delete it, or -# run `make distclean`, to force a rebuild). Each tool is built from upstream -# source into a scratch dir, the binary copied into bin/, and the source tree -# removed — so only the ~2 MB of binaries (self-contained: libc++/libSystem -# only) and the small proofix/ Python clone remain. The whole directory is -# git-ignored via .git/info/exclude. +# Pinned tools are built into revision-qualified paths and the convenience +# names are refreshed from those exact binaries on every make invocation. A +# revision change therefore creates a new prerequisite instead of silently +# accepting a stale bin/kissat or bin/march_cu. Source trees remain temporary. BIN := $(CURDIR)/bin BUILD := $(CURDIR)/.build @@ -20,7 +18,14 @@ KISSAT_REPO := https://github.com/arminbiere/kissat.git CNC_REPO := https://github.com/marijnheule/CnC.git PROOFIX_REPO := https://github.com/zaxioms0/proofix.git -.PHONY: all check clean distclean +# Frozen source revisions for the issue #51 hard-regime table. Update these +# only by revising the experiment contract before any held-out run. +KISSAT_REV := 8af8e56f174b778aef3aa45af9f739b2a5f492c2 +CNC_REV := 705b60c6491ef2b61988b3ce6ac674be1b90571d +KISSAT_PINNED := $(BIN)/kissat-$(KISSAT_REV) +CNC_PINNED := $(BIN)/march_cu-$(CNC_REV) + +.PHONY: all check clean distclean FORCE bin/kissat bin/march_cu all: $(BIN)/cadical $(BIN)/kissat $(BIN)/march_cu proofix @rm -rf $(BUILD) @@ -28,6 +33,12 @@ all: $(BIN)/cadical $(BIN)/kissat $(BIN)/march_cu proofix @echo "cnc-tools ready. Put the binaries on PATH:" @echo " export PATH=$(BIN):\$$PATH" +# Relative aliases make explicit targets behave like the default build even +# though the canonical recipes use absolute paths. +bin/kissat: $(BIN)/kissat + +bin/march_cu: $(BIN)/march_cu + $(BIN): mkdir -p $(BIN) @@ -40,17 +51,25 @@ $(BIN)/cadical: | $(BIN) rm -rf $(BUILD)/cadical # kissat — the fixed conquer solver for every cuber (fairness). -$(BIN)/kissat: | $(BIN) +$(BIN)/kissat: $(KISSAT_PINNED) FORCE | $(BIN) + @cmp -s $< $@ || cp $< $@ + +$(KISSAT_PINNED): | $(BIN) rm -rf $(BUILD)/kissat - git clone --depth 1 $(KISSAT_REPO) $(BUILD)/kissat + git clone $(KISSAT_REPO) $(BUILD)/kissat + cd $(BUILD)/kissat && git checkout --detach $(KISSAT_REV) cd $(BUILD)/kissat && ./configure && $(MAKE) cp $(BUILD)/kissat/build/kissat $@ rm -rf $(BUILD)/kissat # march_cu — the lookahead cuber baseline (bundled in marijnheule/CnC). -$(BIN)/march_cu: | $(BIN) +$(BIN)/march_cu: $(CNC_PINNED) FORCE | $(BIN) + @cmp -s $< $@ || cp $< $@ + +$(CNC_PINNED): | $(BIN) rm -rf $(BUILD)/CnC - git clone --depth 1 $(CNC_REPO) $(BUILD)/CnC + git clone $(CNC_REPO) $(BUILD)/CnC + cd $(BUILD)/CnC && git checkout --detach $(CNC_REV) cd $(BUILD)/CnC/march_cu && $(MAKE) cp $(BUILD)/CnC/march_cu/march_cu $@ rm -rf $(BUILD)/CnC @@ -61,8 +80,8 @@ proofix: check: @printf "cadical "; $(BIN)/cadical --version - @printf "kissat "; $(BIN)/kissat --version - @printf "march_cu "; $(BIN)/march_cu 2>&1 | head -1 + @printf "kissat "; $(KISSAT_PINNED) --version + @printf "march_cu "; $(CNC_PINNED) 2>&1 | head -1 @test -f proofix/proofix.py && echo "proofix present (proofix/proofix.py)" || echo "proofix MISSING" clean: diff --git a/cnc-tools/README.md b/cnc-tools/README.md index 04f55c8..89f0c4e 100644 --- a/cnc-tools/README.md +++ b/cnc-tools/README.md @@ -8,11 +8,15 @@ once it is on `PATH`: export PATH=cnc-tools/bin:$PATH ``` +`make` also keeps revision-qualified experiment binaries such as +`kissat-8af8e56f...` and `march_cu-705b60c...`. Hard-regime toolchain locks must +name those qualified paths; the bare names are convenience copies only. + | Tool | Role | Version | Source/build policy | |---|---|---|---| | `cadical` | conquer solver + DRAT emitter | record `--version` and executable hash | build with the pinned Makefile target | -| `kissat` | fixed conquer solver | record `--version` and executable hash | build with the pinned Makefile target | -| `march_cu` | external lookahead cuber | record executable hash | build upstream source with the Makefile target | +| `kissat` | fixed conquer solver | `8af8e56f174b778aef3aa45af9f739b2a5f492c2`; also record `--version` and executable hash | build with the pinned Makefile target | +| `march_cu` | external lookahead cuber | `705b60c6491ef2b61988b3ce6ac674be1b90571d`; also record executable hash | build upstream source with the Makefile target | | Proofix | optional proof-prefix cuber | SAT 2025 | pinned clone in `proofix/` | ## Primary cube generation diff --git a/src/bin/cnc_cuber.rs b/src/bin/cnc_cuber.rs index 94caab7..0ec18b6 100644 --- a/src/bin/cnc_cuber.rs +++ b/src/bin/cnc_cuber.rs @@ -1,4 +1,4 @@ -//! Generate Cube-and-Conquer assumptions with the current Rust region cuber. +//! Export a complete cube frontier or stream it into parallel Kissat workers. //! //! The primary stopping rule is the classical online Cube-and-Conquer //! difficulty cutoff (`--cc-threshold`). A march-compatible remaining-variable @@ -11,9 +11,10 @@ use std::path::{Path, PathBuf}; use boolean_inference::adapter::BranchSolver; use boolean_inference::circuit::network_from_circuit_sat; +use boolean_inference::conquer::{ConquerResult, StreamingConquer}; use boolean_inference::cube::{ generate_cubes_with_cutoff, generate_cubes_with_cutoff_trace, CubeCutoff, CubeNodeKind, - CubeNodeTrace, + CubeNodeTrace, CubeRefutationReason, }; use boolean_inference::dimacs::network_from_dimacs; use boolean_inference::measure::Measure; @@ -22,13 +23,45 @@ use boolean_inference::problem::TnProblem; use boolean_inference::selector::Selector; use optimal_branching_core::GreedyMerge; -const USAGE: &str = "usage: cnc_cuber (-n | --cc-threshold ) -o \ - [--max-rows ] [--trace ]"; +const USAGE: &str = + "usage: cnc_cuber (-n | --cc-threshold ) \ + (-o | --solve-cnf --kissat --workers ) \ + [--selector ] [--max-rows ] [--trace ]"; +const SOLVED: &str = "streaming-conquer-found-sat"; + +#[derive(Clone, Copy, Debug)] +enum SelectorKind { + Region, + StructureBlind, +} + +impl SelectorKind { + fn parse(value: &str) -> Result { + match value { + "region" => Ok(Self::Region), + "structure-blind" => Ok(Self::StructureBlind), + _ => Err(format!( + "invalid --selector value: {value}; expected region or structure-blind" + )), + } + } + + fn label(self) -> &'static str { + match self { + Self::Region => "region", + Self::StructureBlind => "structure-blind", + } + } +} struct Args { input: PathBuf, - output: PathBuf, + output: Option, + solve_cnf: Option, + kissat: Option, + workers: Option, cutoff: CubeCutoff, + selector: SelectorKind, max_rows: usize, trace: Option, } @@ -49,9 +82,13 @@ fn parse_args() -> Result { let raw: Vec = std::env::args().skip(1).collect(); let mut input = None; let mut output = None; + let mut solve_cnf = None; + let mut kissat = None; + let mut workers = None; let mut cutoff_vars = None; let mut cc_threshold = None; let mut max_rows = 512usize; + let mut selector = SelectorKind::Region; let mut trace = None; let mut i = 0usize; @@ -77,7 +114,22 @@ fn parse_args() -> Result { ); } "-o" => output = Some(take_value(&raw, &mut i, "-o")?), + "--solve-cnf" => solve_cnf = Some(take_value(&raw, &mut i, "--solve-cnf")?), + "--kissat" => kissat = Some(take_value(&raw, &mut i, "--kissat")?), + "--workers" => { + let value = take_value(&raw, &mut i, "--workers")?; + let count = value + .parse::() + .map_err(|_| format!("invalid --workers value: {value}"))?; + if count == 0 { + return Err("--workers must be greater than zero".into()); + } + workers = Some(count); + } "--trace" => trace = Some(take_value(&raw, &mut i, "--trace")?), + "--selector" => { + selector = SelectorKind::parse(&take_value(&raw, &mut i, "--selector")?)?; + } "--max-rows" => { let value = take_value(&raw, &mut i, "--max-rows")?; max_rows = value @@ -104,10 +156,19 @@ fn parse_args() -> Result { return Err("-n and --cc-threshold are mutually exclusive".to_string()) } }; + match (&output, &solve_cnf, &kissat, workers) { + (Some(_), None, None, None) => {} + (None, Some(_), Some(_), Some(_)) => {} + _ => return Err("select either -o, or all of --solve-cnf/--kissat/--workers".to_string()), + } Ok(Command::Run(Args { input: PathBuf::from(input.ok_or_else(|| "missing input instance".to_string())?), - output: PathBuf::from(output.ok_or_else(|| "missing -o output".to_string())?), + output: output.map(PathBuf::from), + solve_cnf: solve_cnf.map(PathBuf::from), + kissat: kissat.map(PathBuf::from), + workers, cutoff, + selector, max_rows, trace: trace.map(PathBuf::from), })) @@ -146,6 +207,14 @@ fn node_kind(kind: CubeNodeKind) -> &'static str { } } +fn refutation_reason(reason: CubeRefutationReason) -> &'static str { + match reason { + CubeRefutationReason::RootPropagation => "root-propagation-contradiction", + CubeRefutationReason::SelectorNoFeasibleConfig => "selector-no-feasible-config", + CubeRefutationReason::BranchPropagation => "branch-propagation-contradiction", + } +} + fn write_trace_node( writer: &mut dyn Write, node: CubeNodeTrace, @@ -180,6 +249,7 @@ fn write_trace_node( "child_index": node.child_index, "depth": node.depth, "kind": node_kind(node.kind), + "refutation_reason": node.refutation_reason.map(refutation_reason), "literals": literals, "sigma_dec": node.sigma_dec, "sigma_all": node.sigma_all, @@ -194,8 +264,12 @@ fn write_trace_node( .map_err(|error| format!("write trace: {error}")) } -fn run(args: Args) -> Result<(), String> { - if args.trace.as_deref() == Some(args.output.as_path()) { +fn run(args: Args) -> Result { + if args + .trace + .as_ref() + .is_some_and(|trace| args.output.as_ref() == Some(trace)) + { return Err("--trace must differ from the cube output path".into()); } let network = load_network(&args.input)?; @@ -213,7 +287,16 @@ fn run(args: Args) -> Result<(), String> { return Err("constraint network has an incomplete variable map".into()); } - let mut writer = output_writer(&args.output)?; + let mut writer: Box = match &args.output { + Some(output) => output_writer(output)?, + None => Box::new(io::sink()), + }; + let mut conquer = match (&args.solve_cnf, &args.kissat, args.workers) { + (Some(cnf), Some(kissat), Some(workers)) => { + Some(StreamingConquer::start(cnf, kissat, workers).map_err(|error| error.to_string())?) + } + _ => None, + }; let mut trace_writer = match &args.trace { Some(path) => { Some(BufWriter::new(File::create(path).map_err(|error| { @@ -225,12 +308,41 @@ fn run(args: Args) -> Result<(), String> { let mut problem = match TnProblem::from_network(network) { Ok(problem) => problem, Err(_) => { + if let Some(trace_writer) = trace_writer.as_mut() { + write_trace_node( + trace_writer, + CubeNodeTrace { + node_id: 0, + parent_id: None, + child_index: None, + depth: 0, + kind: CubeNodeKind::Refuted, + refutation_reason: Some(CubeRefutationReason::RootPropagation), + decisions: Vec::new(), + sigma_dec: 0, + sigma_all: 0, + freevars: nvars, + variables: Vec::new(), + clauses: Vec::new(), + }, + &new_to_orig, + )?; + trace_writer + .flush() + .map_err(|error| format!("flush trace: {error}"))?; + } writer.flush().map_err(|e| format!("flush output: {e}"))?; eprintln!( "status=UNSAT_AT_ROOT cubes=0 refuted=1 sat_leaves=0 cutoff={:?}", args.cutoff ); - return Ok(()); + if let Some(conquer) = conquer.take() { + let summary = conquer.finish(true).map_err(|error| error.to_string())?; + debug_assert_eq!(summary.result, ConquerResult::Unsat); + println!("s UNSATISFIABLE"); + return Ok(20); + } + return Ok(0); } }; let root_unfixed = problem.count_unfixed(); @@ -238,15 +350,26 @@ fn run(args: Args) -> Result<(), String> { let mut emitted = 0usize; let mut min_remaining = usize::MAX; let mut max_remaining = 0usize; - let selector = Selector::MostOccurrence { - max_rows: args.max_rows, + let selector = match args.selector { + SelectorKind::Region => Selector::MostOccurrence { + max_rows: args.max_rows, + }, + SelectorKind::StructureBlind => Selector::BinaryOccurrence, }; let solver = BranchSolver::Greedy(GreedyMerge); let mut emit = |cube: boolean_inference::cube::Cube| { - if cube.refuted || cube.sat { + if cube.refuted { return Ok(()); } + let leaf_sat = cube.sat; + if leaf_sat { + if let Some(conquer) = conquer.as_ref() { + conquer.mark_sat(); + return Err(SOLVED.to_string()); + } + } + let remaining = nvars - cube.sigma_all; let stopped = match args.cutoff { CubeCutoff::RemainingVars(n) => remaining < n.get(), @@ -255,31 +378,44 @@ fn run(args: Args) -> Result<(), String> { > threshold * (nvars as u128) } }; - if !stopped { + if !leaf_sat && !stopped { return Err(format!( "internal cutoff error: emitted cube does not satisfy {:?}", args.cutoff )); } - writer - .write_all(b"a") - .map_err(|e| format!("write output: {e}"))?; + let mut literals = Vec::with_capacity(cube.decisions.len()); for &(compressed, value) in &cube.decisions { let literal = (new_to_orig[compressed] + 1) as i64; let literal = if value { literal } else { -literal }; - write!(writer, " {literal}").map_err(|e| format!("write output: {e}"))?; + literals.push(literal); + } + if let Some(conquer) = conquer.as_ref() { + if !conquer + .submit(literals) + .map_err(|error| error.to_string())? + { + return Err(SOLVED.to_string()); + } + } else { + writer + .write_all(b"a") + .map_err(|e| format!("write output: {e}"))?; + for literal in literals { + write!(writer, " {literal}").map_err(|e| format!("write output: {e}"))?; + } + writer + .write_all(b" 0\n") + .map_err(|e| format!("write output: {e}"))?; } - writer - .write_all(b" 0\n") - .map_err(|e| format!("write output: {e}"))?; emitted += 1; min_remaining = min_remaining.min(remaining); max_remaining = max_remaining.max(remaining); Ok(()) }; - let stats = match trace_writer.as_mut() { + let generated = match trace_writer.as_mut() { Some(trace_writer) => generate_cubes_with_cutoff_trace( &mut problem, selector, @@ -297,7 +433,18 @@ fn run(args: Args) -> Result<(), String> { args.cutoff, &mut emit, ), - }?; + }; + let stopped_on_sat = matches!(&generated, Err(error) if error == SOLVED); + let stats = match generated { + Ok(stats) => Some(stats), + Err(error) if error == SOLVED => None, + Err(error) => { + if let Some(conquer) = conquer.take() { + let _ = conquer.finish(false); + } + return Err(error); + } + }; writer.flush().map_err(|e| format!("flush output: {e}"))?; if let Some(trace_writer) = trace_writer.as_mut() { trace_writer @@ -310,37 +457,74 @@ fn run(args: Args) -> Result<(), String> { } else { format!("{min_remaining}..={max_remaining}") }; - eprintln!( - "status=OK cubes={} refuted={} sat_leaves={} visited={} cutoff={:?} \ - root_unfixed={} remaining_range={} max_rows={}", - stats.cubes, - stats.refuted, - stats.sat_leaves, - stats.visited, - args.cutoff, - root_unfixed, - remaining_range, - args.max_rows - ); - if emitted != stats.cubes { - return Err(format!( - "internal accounting error: wrote {emitted} cubes, expected {}", - stats.cubes - )); + if let Some(stats) = stats { + eprintln!( + "status=OK cubes={} refuted={} sat_leaves={} visited={} cutoff={:?} \ + root_unfixed={} remaining_range={} selector={} max_rows={}", + stats.cubes, + stats.refuted, + stats.sat_leaves, + stats.visited, + args.cutoff, + root_unfixed, + remaining_range, + args.selector.label(), + args.max_rows + ); + let expected = stats.cubes + stats.sat_leaves; + if emitted != expected { + return Err(format!( + "internal accounting error: wrote {emitted} cubes, expected {}", + expected + )); + } + } else { + eprintln!( + "status=SAT_EARLY cubes_submitted={} cutoff={:?} selector={}", + emitted, + args.cutoff, + args.selector.label() + ); } - Ok(()) + if let Some(conquer) = conquer.take() { + let summary = conquer + .finish(!stopped_on_sat) + .map_err(|error| error.to_string())?; + eprintln!( + "streaming submitted={} sat={} unsat={} errors={}", + summary.submitted, summary.sat, summary.unsat, summary.errors + ); + return match summary.result { + ConquerResult::Sat => { + if let Some(witness) = summary.witness { + print!("{witness}"); + } else { + println!("s SATISFIABLE"); + } + Ok(10) + } + ConquerResult::Unsat => { + println!("s UNSATISFIABLE"); + Ok(20) + } + ConquerResult::Incomplete => Err("streaming conquer was incomplete".into()), + }; + } + Ok(0) } fn main() { match parse_args() { Ok(Command::Help) => println!("{USAGE}"), - Ok(Command::Run(args)) => { - if let Err(message) = run(args) { + Ok(Command::Run(args)) => match run(args) { + Ok(0) => {} + Ok(code) => std::process::exit(code), + Err(message) => { eprintln!("error: {message}"); eprintln!("{USAGE}"); std::process::exit(2); } - } + }, Err(message) => { eprintln!("error: {message}"); eprintln!("{USAGE}"); diff --git a/src/conquer.rs b/src/conquer.rs new file mode 100644 index 0000000..09d2eda --- /dev/null +++ b/src/conquer.rs @@ -0,0 +1,313 @@ +//! Bounded streaming conquer pool for Cube-and-Conquer frontiers. + +use std::fs; +use std::io::{BufRead, BufReader, Write}; +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{mpsc, Arc, Mutex}; +use std::thread::{self, JoinHandle}; + +#[derive(Debug, thiserror::Error)] +pub enum ConquerError { + #[error("read CNF {path}: {source}")] + ReadCnf { + path: PathBuf, + source: std::io::Error, + }, + #[error("CNF has no valid 'p cnf ' header")] + MissingHeader, + #[error("start Kissat: {0}")] + StartKissat(std::io::Error), + #[error("write CNF to Kissat: {0}")] + WriteKissat(std::io::Error), + #[error("read Kissat output: {0}")] + ReadKissat(std::io::Error), + #[error("streaming conquer worker disconnected")] + Disconnected, + #[error("streaming conquer worker panicked")] + WorkerPanicked, +} + +#[derive(Clone)] +struct CnfTemplate { + variables: usize, + clauses: usize, + body: Arc>, +} + +impl CnfTemplate { + fn read(path: &Path) -> Result { + let text = fs::read_to_string(path).map_err(|source| ConquerError::ReadCnf { + path: path.to_owned(), + source, + })?; + let mut header = None; + let mut body = Vec::with_capacity(text.len()); + for line in text.split_inclusive('\n') { + let fields: Vec<_> = line.split_whitespace().collect(); + if fields.first() == Some(&"p") { + if fields.len() != 4 || fields[1] != "cnf" { + return Err(ConquerError::MissingHeader); + } + let variables = fields[2].parse().map_err(|_| ConquerError::MissingHeader)?; + let clauses = fields[3].parse().map_err(|_| ConquerError::MissingHeader)?; + if header.replace((variables, clauses)).is_some() { + return Err(ConquerError::MissingHeader); + } + } else { + body.extend_from_slice(line.as_bytes()); + } + } + let (variables, clauses) = header.ok_or(ConquerError::MissingHeader)?; + Ok(Self { + variables, + clauses, + body: Arc::new(body), + }) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ConquerResult { + Sat, + Unsat, + Incomplete, +} + +#[derive(Debug)] +pub struct ConquerSummary { + pub result: ConquerResult, + pub submitted: usize, + pub sat: usize, + pub unsat: usize, + pub errors: usize, + pub witness: Option, +} + +struct Shared { + stopped: AtomicBool, + submitted: AtomicUsize, + sat: AtomicUsize, + unsat: AtomicUsize, + errors: AtomicUsize, + witness: Mutex>, +} + +impl Shared { + fn new() -> Self { + Self { + stopped: AtomicBool::new(false), + submitted: AtomicUsize::new(0), + sat: AtomicUsize::new(0), + unsat: AtomicUsize::new(0), + errors: AtomicUsize::new(0), + witness: Mutex::new(None), + } + } +} + +/// A fixed-size Kissat pool fed directly by the cuber's leaf callback. +pub struct StreamingConquer { + sender: Option>>, + shared: Arc, + workers: Vec>, +} + +impl StreamingConquer { + pub fn start(cnf: &Path, kissat: &Path, workers: usize) -> Result { + assert!(workers > 0, "workers must be positive"); + let template = CnfTemplate::read(cnf)?; + let shared = Arc::new(Shared::new()); + let (sender, receiver) = mpsc::sync_channel(workers); + let receiver = Arc::new(Mutex::new(receiver)); + let mut handles = Vec::with_capacity(workers); + for _ in 0..workers { + let receiver = Arc::clone(&receiver); + let shared = Arc::clone(&shared); + let template = template.clone(); + let kissat = kissat.to_owned(); + handles.push(thread::spawn(move || { + worker_loop(receiver, shared, template, kissat) + })); + } + Ok(Self { + sender: Some(sender), + shared, + workers: handles, + }) + } + + /// Submit one open cube. Returns `false` once another cube has proved SAT. + pub fn submit(&self, cube: Vec) -> Result { + if self.shared.stopped.load(Ordering::Acquire) { + return Ok(false); + } + let sent = self + .sender + .as_ref() + .ok_or(ConquerError::Disconnected)? + .send(cube); + if sent.is_err() { + return if self.shared.stopped.load(Ordering::Acquire) { + Ok(false) + } else { + Err(ConquerError::Disconnected) + }; + } + self.shared.submitted.fetch_add(1, Ordering::Relaxed); + Ok(!self.shared.stopped.load(Ordering::Acquire)) + } + + /// Record a satisfying leaf found by the cuber itself. + pub fn mark_sat(&self) { + self.shared.sat.fetch_add(1, Ordering::Relaxed); + self.shared.stopped.store(true, Ordering::Release); + } + + pub fn finish(mut self, cubing_complete: bool) -> Result { + self.sender.take(); + for worker in self.workers.drain(..) { + worker.join().map_err(|_| ConquerError::WorkerPanicked)?; + } + let submitted = self.shared.submitted.load(Ordering::Relaxed); + let sat = self.shared.sat.load(Ordering::Relaxed); + let unsat = self.shared.unsat.load(Ordering::Relaxed); + let errors = self.shared.errors.load(Ordering::Relaxed); + let result = if sat > 0 { + ConquerResult::Sat + } else if cubing_complete && errors == 0 && unsat == submitted { + ConquerResult::Unsat + } else { + ConquerResult::Incomplete + }; + let witness = self.shared.witness.lock().expect("witness lock").take(); + Ok(ConquerSummary { + result, + submitted, + sat, + unsat, + errors, + witness, + }) + } +} + +fn worker_loop( + receiver: Arc>>>, + shared: Arc, + template: CnfTemplate, + kissat: PathBuf, +) { + loop { + if shared.stopped.load(Ordering::Acquire) { + break; + } + let cube = match receiver.lock().expect("cube receiver lock").recv() { + Ok(cube) => cube, + Err(_) => break, + }; + if shared.stopped.load(Ordering::Acquire) { + break; + } + match solve_cube(&template, &kissat, &shared, &cube) { + Ok(CubeResult::Sat(output)) => { + shared.sat.fetch_add(1, Ordering::Relaxed); + *shared.witness.lock().expect("witness lock") = Some(output); + shared.stopped.store(true, Ordering::Release); + } + Ok(CubeResult::Unsat) => { + shared.unsat.fetch_add(1, Ordering::Relaxed); + } + Ok(CubeResult::Cancelled) => {} + Ok(CubeResult::Unknown) | Err(_) => { + shared.errors.fetch_add(1, Ordering::Relaxed); + } + } + } +} + +enum CubeResult { + Sat(String), + Unsat, + Cancelled, + Unknown, +} + +fn solve_cube( + template: &CnfTemplate, + kissat: &Path, + shared: &Shared, + cube: &[i64], +) -> Result { + if cube + .iter() + .any(|literal| literal.unsigned_abs() as usize > template.variables) + { + return Ok(CubeResult::Unknown); + } + let mut child = Command::new(kissat) + .arg("--relaxed") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .map_err(ConquerError::StartKissat)?; + let stdout = child.stdout.take().expect("piped Kissat stdout"); + let reader = thread::spawn(move || read_solution(BufReader::new(stdout))); + let write_result = (|| { + let mut stdin = child.stdin.take().expect("piped Kissat stdin"); + writeln!( + stdin, + "p cnf {} {}", + template.variables, + template.clauses + cube.len() + ) + .and_then(|_| stdin.write_all(&template.body)) + .map_err(ConquerError::WriteKissat)?; + if !template.body.ends_with(b"\n") { + stdin.write_all(b"\n").map_err(ConquerError::WriteKissat)?; + } + for literal in cube { + writeln!(stdin, "{literal} 0").map_err(ConquerError::WriteKissat)?; + } + Ok(()) + })(); + if let Err(error) = write_result { + let _ = child.kill(); + let _ = child.wait(); + let _ = reader.join(); + return Err(error); + } + + let status = loop { + if shared.stopped.load(Ordering::Acquire) { + let _ = child.kill(); + let _ = child.wait(); + reader.join().expect("Kissat output reader panicked")?; + return Ok(CubeResult::Cancelled); + } + if let Some(status) = child.try_wait().map_err(ConquerError::StartKissat)? { + break status; + } + thread::sleep(std::time::Duration::from_millis(5)); + }; + let solution = reader.join().expect("Kissat output reader panicked")?; + match status.code() { + Some(10) => Ok(CubeResult::Sat(solution)), + Some(20) => Ok(CubeResult::Unsat), + _ => Ok(CubeResult::Unknown), + } +} + +fn read_solution(reader: impl BufRead) -> Result { + let mut solution = String::new(); + for line in reader.lines() { + let line = line.map_err(ConquerError::ReadKissat)?; + if line.starts_with("s ") || line.starts_with("v ") { + solution.push_str(&line); + solution.push('\n'); + } + } + Ok(solution) +} diff --git a/src/cube.rs b/src/cube.rs index 4745a7d..be8bf41 100644 --- a/src/cube.rs +++ b/src/cube.rs @@ -61,6 +61,14 @@ pub enum CubeNodeKind { Sat, } +/// Why a traced leaf is known to be closed without a conquer cube. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum CubeRefutationReason { + RootPropagation, + SelectorNoFeasibleConfig, + BranchPropagation, +} + /// One branching clause in the bit encoding over `CubeNodeTrace::variables`. #[derive(Clone, Debug, PartialEq, Eq)] pub struct TraceClause { @@ -78,6 +86,7 @@ pub struct CubeNodeTrace { pub child_index: Option, pub depth: usize, pub kind: CubeNodeKind, + pub refutation_reason: Option, pub decisions: Vec<(usize, bool)>, pub sigma_dec: usize, pub sigma_all: usize, @@ -287,6 +296,7 @@ where child_index: None, depth: 0, kind: CubeNodeKind::Refuted, + refutation_reason: Some(CubeRefutationReason::RootPropagation), decisions: Vec::new(), sigma_dec: 0, sigma_all: 0, @@ -384,6 +394,7 @@ where child_index, depth, kind: CubeNodeKind::Cutoff, + refutation_reason: None, decisions: decisions.clone(), sigma_dec, sigma_all, @@ -417,6 +428,7 @@ where child_index, depth, kind: CubeNodeKind::Sat, + refutation_reason: None, decisions: decisions.clone(), sigma_dec, sigma_all, @@ -459,6 +471,7 @@ where child_index, depth, kind: CubeNodeKind::Refuted, + refutation_reason: Some(CubeRefutationReason::SelectorNoFeasibleConfig), decisions: decisions.clone(), sigma_dec, sigma_all, @@ -496,6 +509,7 @@ where child_index, depth, kind: CubeNodeKind::Branch, + refutation_reason: None, decisions: decisions.clone(), sigma_dec, sigma_all, @@ -538,6 +552,7 @@ where child_index: Some(branch_index), depth: depth + 1, kind: CubeNodeKind::Refuted, + refutation_reason: Some(CubeRefutationReason::BranchPropagation), decisions: decisions.clone(), sigma_dec: decisions.len(), sigma_all: doms.len() - branch_freevars, diff --git a/src/lib.rs b/src/lib.rs index e41c58c..981c463 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2,6 +2,7 @@ pub mod adapter; pub mod api; pub mod canonicalize; pub mod circuit; +pub mod conquer; pub mod contract; pub mod ct; pub mod cube; diff --git a/tests/cnc_cuber_trace.rs b/tests/cnc_cuber_trace.rs index 9964c26..fdf2829 100644 --- a/tests/cnc_cuber_trace.rs +++ b/tests/cnc_cuber_trace.rs @@ -65,6 +65,9 @@ fn trace_flag_preserves_cubes_and_writes_original_variable_ids() { assert!(!records.is_empty()); assert_eq!(records[0]["node_id"], 0); assert!(records[0]["parent_id"].is_null()); + assert!(records + .iter() + .all(|record| { record["kind"] == "refuted" || record["refutation_reason"].is_null() })); assert_eq!( records .iter() @@ -83,3 +86,82 @@ fn trace_flag_preserves_cubes_and_writes_original_variable_ids() { fs::remove_dir_all(dir).expect("remove temp directory"); } + +#[test] +fn structure_blind_selector_is_auditable_binary_control() { + let dir = temp_dir(); + fs::create_dir_all(&dir).expect("create temp directory"); + let input = dir.join("input.cnf"); + let cubes = dir.join("blind.cubes"); + let trace = dir.join("blind.jsonl"); + fs::write(&input, "p cnf 3 4\n1 2 0\n-1 -2 0\n2 3 0\n-2 -3 0\n").expect("write CNF"); + + let run = Command::new(env!("CARGO_BIN_EXE_cnc_cuber")) + .args([ + input.as_os_str(), + "-n".as_ref(), + "3".as_ref(), + "-o".as_ref(), + cubes.as_os_str(), + "--selector".as_ref(), + "structure-blind".as_ref(), + "--trace".as_ref(), + trace.as_os_str(), + ]) + .output() + .expect("run structure-blind cuber"); + assert!(run.status.success(), "{:?}", run); + let stderr = String::from_utf8(run.stderr).expect("UTF-8 stderr"); + assert!(stderr.contains("selector=structure-blind"), "{stderr}"); + + let records: Vec = fs::read_to_string(&trace) + .unwrap() + .lines() + .map(|line| serde_json::from_str(line).expect("valid trace JSON")) + .collect(); + for branch in records.iter().filter(|record| record["kind"] == "branch") { + assert_eq!(branch["rule_variables"].as_array().unwrap().len(), 1); + let clauses = branch["rule_clauses"].as_array().unwrap(); + assert_eq!(clauses.len(), 2); + assert_eq!(clauses[0]["mask"], 1); + assert_eq!(clauses[0]["value"], 0); + assert_eq!(clauses[1]["mask"], 1); + assert_eq!(clauses[1]["value"], 1); + } + + fs::remove_dir_all(dir).expect("remove temp directory"); +} + +#[test] +fn root_refutation_trace_records_a_semantic_closure_reason() { + let dir = temp_dir(); + fs::create_dir_all(&dir).expect("create temp directory"); + let input = dir.join("root-unsat.cnf"); + let cubes = dir.join("root-unsat.cubes"); + let trace = dir.join("root-unsat.jsonl"); + fs::write(&input, "p cnf 1 2\n1 0\n-1 0\n").expect("write CNF"); + + let run = Command::new(env!("CARGO_BIN_EXE_cnc_cuber")) + .args([ + input.as_os_str(), + "-n".as_ref(), + "1".as_ref(), + "-o".as_ref(), + cubes.as_os_str(), + "--trace".as_ref(), + trace.as_os_str(), + ]) + .output() + .expect("run root-UNSAT cuber"); + assert!(run.status.success(), "{:?}", run); + assert!(fs::read_to_string(&cubes).unwrap().is_empty()); + let record: serde_json::Value = + serde_json::from_str(fs::read_to_string(&trace).unwrap().trim()).unwrap(); + assert_eq!(record["kind"], "refuted"); + assert_eq!( + record["refutation_reason"], + "root-propagation-contradiction" + ); + + fs::remove_dir_all(dir).expect("remove temp directory"); +} diff --git a/tests/cnc_streaming.rs b/tests/cnc_streaming.rs new file mode 100644 index 0000000..6c81b9d --- /dev/null +++ b/tests/cnc_streaming.rs @@ -0,0 +1,151 @@ +#![cfg(unix)] + +use std::fs; +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use boolean_inference::conquer::{ConquerResult, StreamingConquer}; + +fn temp_dir() -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock after epoch") + .as_nanos(); + std::env::temp_dir().join(format!( + "boolean-inference-streaming-{}-{nonce}", + std::process::id() + )) +} + +fn fake_kissat(path: &Path, result: &str, code: i32) { + fs::write( + path, + format!( + "#!/bin/sh\n\ + [ \"$#\" -eq 1 ] && [ \"$1\" = --relaxed ] || exit 3\n\ + input=$(cat)\n\ + case \"$input\" in *\"p cnf 2 \"*) ;; *) exit 4 ;; esac\n\ + echo 'c output that streaming mode must discard'\n\ + echo 's {result}'\n\ + exit {code}\n" + ), + ) + .expect("write fake Kissat"); + let mut permissions = fs::metadata(path).unwrap().permissions(); + permissions.set_mode(0o755); + fs::set_permissions(path, permissions).unwrap(); +} + +fn run_streaming(dir: &Path, kissat: &Path) -> std::process::Output { + let cnf = dir.join("input.cnf"); + fs::write(&cnf, "p cnf 2 2\n1 2 0\n-1 -2 0\n").expect("write CNF"); + Command::new(env!("CARGO_BIN_EXE_cnc_cuber")) + .args([ + cnf.as_os_str(), + "-n".as_ref(), + "2".as_ref(), + "--solve-cnf".as_ref(), + cnf.as_os_str(), + "--kissat".as_ref(), + kissat.as_os_str(), + "--workers".as_ref(), + "2".as_ref(), + ]) + .output() + .expect("run streaming solver") +} + +#[test] +fn streaming_mode_reports_unsat_after_all_open_cubes_close() { + let dir = temp_dir(); + fs::create_dir_all(&dir).unwrap(); + let kissat = dir.join("kissat-unsat"); + fake_kissat(&kissat, "UNSATISFIABLE", 20); + + let output = run_streaming(&dir, &kissat); + + assert_eq!(output.status.code(), Some(20), "{output:?}"); + assert!(String::from_utf8_lossy(&output.stdout).contains("s UNSATISFIABLE")); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("submitted=1"), "{stderr}"); + assert!(stderr.contains("unsat=1"), "{stderr}"); + fs::remove_dir_all(dir).unwrap(); +} + +#[test] +fn streaming_mode_stops_after_a_sat_cube() { + let dir = temp_dir(); + fs::create_dir_all(&dir).unwrap(); + let kissat = dir.join("kissat-sat"); + fake_kissat(&kissat, "SATISFIABLE", 10); + + let output = run_streaming(&dir, &kissat); + + assert_eq!(output.status.code(), Some(10), "{output:?}"); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!(stdout.contains("s SATISFIABLE")); + assert!(!stdout.contains("output that streaming mode must discard")); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("sat=1"), "{stderr}"); + fs::remove_dir_all(dir).unwrap(); +} + +#[test] +fn streaming_mode_kills_an_inflight_solver_after_sat() { + let dir = temp_dir(); + fs::create_dir_all(&dir).unwrap(); + let cnf = dir.join("input.cnf"); + fs::write(&cnf, "p cnf 2 0\n").unwrap(); + let started = dir.join("slow-started"); + let kissat = dir.join("kissat-race"); + fs::write( + &kissat, + format!( + "#!/bin/sh\n\ + input=$(cat)\n\ + case \"$input\" in\n\ + *'-1 0'*) while [ ! -e '{}' ]; do sleep 0.01; done; echo 's SATISFIABLE'; exit 10 ;;\n\ + *) touch '{}'; exec sleep 30 ;;\n\ + esac\n", + started.display(), + started.display(), + ), + ) + .unwrap(); + let mut permissions = fs::metadata(&kissat).unwrap().permissions(); + permissions.set_mode(0o755); + fs::set_permissions(&kissat, permissions).unwrap(); + + let conquer = StreamingConquer::start(&cnf, &kissat, 2).unwrap(); + assert!(conquer.submit(vec![1]).unwrap()); + assert!(conquer.submit(vec![-1]).unwrap()); + let before = Instant::now(); + let summary = conquer.finish(true).unwrap(); + + assert_eq!(summary.result, ConquerResult::Sat); + assert_eq!(summary.sat, 1); + assert!(started.exists(), "slow Kissat never started"); + assert!( + before.elapsed() < Duration::from_secs(10), + "in-flight Kissat was not killed: {:?}", + before.elapsed() + ); + fs::remove_dir_all(dir).unwrap(); +} + +#[test] +fn streaming_mode_never_claims_unsat_after_a_worker_error() { + let dir = temp_dir(); + fs::create_dir_all(&dir).unwrap(); + let kissat = dir.join("kissat-error"); + fake_kissat(&kissat, "UNKNOWN", 1); + + let output = run_streaming(&dir, &kissat); + + assert_eq!(output.status.code(), Some(2), "{output:?}"); + assert!(!String::from_utf8_lossy(&output.stdout).contains("UNSATISFIABLE")); + assert!(String::from_utf8_lossy(&output.stderr).contains("incomplete")); + fs::remove_dir_all(dir).unwrap(); +} diff --git a/tests/test_calibrate_cc_difficulty.py b/tests/test_calibrate_cc_difficulty.py deleted file mode 100644 index 40c20cf..0000000 --- a/tests/test_calibrate_cc_difficulty.py +++ /dev/null @@ -1,36 +0,0 @@ -import unittest - -from benchmarks.cnc.calibrate_cc_difficulty import CalibrationError, choose - - -class CalibrateCcDifficultyTests(unittest.TestCase): - def test_choose_prefers_closest_count_in_accepted_range(self): - rows = [ - {"threshold": 100, "tasks": 300}, - {"threshold": 200, "tasks": 480}, - {"threshold": 300, "tasks": 540}, - {"threshold": 400, "tasks": 900}, - ] - self.assertEqual(choose(rows, 512, 384, 640)["threshold"], 300) - - def test_choose_falls_back_when_range_is_skipped(self): - rows = [ - {"threshold": 100, "tasks": 100}, - {"threshold": 200, "tasks": 1000}, - ] - self.assertEqual(choose(rows, 512, 384, 640)["threshold"], 200) - - def test_tie_breaks_toward_lower_threshold(self): - rows = [ - {"threshold": 200, "tasks": 512}, - {"threshold": 100, "tasks": 512}, - ] - self.assertEqual(choose(rows, 512, 1, 2000)["threshold"], 100) - - def test_empty_response_fails_closed(self): - with self.assertRaises(CalibrationError): - choose([], 512, 384, 640) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_cnc_measurements.py b/tests/test_cnc_measurements.py deleted file mode 100644 index d097138..0000000 --- a/tests/test_cnc_measurements.py +++ /dev/null @@ -1,100 +0,0 @@ -import hashlib -import json -import shutil -import subprocess -import tempfile -import unittest -from pathlib import Path - - -ROOT = Path(__file__).resolve().parents[1] -VERIFIER = ROOT / "benchmarks" / "cnc" / "verify_measurements.py" -FIXTURES = ROOT / "tests" / "fixtures" / "cnc" - - -class CncMeasurementsTest(unittest.TestCase): - def run_bundle(self, path: Path) -> subprocess.CompletedProcess[str]: - return subprocess.run( - ["python3", str(VERIFIER), "--bundle", str(path)], - cwd=ROOT, - text=True, - capture_output=True, - check=False, - ) - - def run_fixture(self, name: str) -> subprocess.CompletedProcess[str]: - return self.run_bundle(FIXTURES / name) - - def test_valid_bundle_recomputes_issue_41_metrics(self): - result = self.run_fixture("measurement-valid") - self.assertEqual(result.returncode, 0, result.stdout + result.stderr) - self.assertEqual( - result.stdout.splitlines(), - [ - "PASS frontier: 8/8 assignments are covered exactly once", - "PASS verdict: returned model satisfies the input", - "PASS accounting: cubing_wall=2 conquer_cpu=9 " - "conquer_makespan=5 end_to_end_wall=7", - "PASS workers: observed concurrency does not exceed 2", - ], - ) - - def test_missing_cube_fails_with_uncovered_assignment(self): - with tempfile.TemporaryDirectory() as directory: - bundle = Path(directory) / "bundle" - shutil.copytree(FIXTURES / "measurement-valid", bundle) - frontier_path = bundle / "frontier.jsonl" - frontier = "\n".join(frontier_path.read_text().splitlines()[1:]) + "\n" - frontier_path.write_text(frontier, encoding="utf-8") - manifest_path = bundle / "bundle.json" - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) - manifest["frontier"]["sha256"] = hashlib.sha256( - frontier.encode("utf-8") - ).hexdigest() - manifest_path.write_text(json.dumps(manifest), encoding="utf-8") - result = self.run_bundle(bundle) - self.assertNotEqual(result.returncode, 0) - self.assertIn("FAIL frontier: assignment 000 is uncovered", result.stdout) - - def test_accounting_is_recomputed_instead_of_trusted(self): - with tempfile.TemporaryDirectory() as directory: - bundle = Path(directory) / "bundle" - shutil.copytree(FIXTURES / "measurement-valid", bundle) - manifest_path = bundle / "bundle.json" - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) - manifest["accounting"]["conquer_cpu"] = 8.0 - manifest_path.write_text(json.dumps(manifest), encoding="utf-8") - result = self.run_bundle(bundle) - self.assertNotEqual(result.returncode, 0) - self.assertIn("accounting mismatch for conquer_cpu", result.stdout) - - def test_worker_identity_cannot_hide_overlapping_work(self): - with tempfile.TemporaryDirectory() as directory: - bundle = Path(directory) / "bundle" - shutil.copytree(FIXTURES / "measurement-valid", bundle) - events_path = bundle / "events.jsonl" - events = [ - json.loads(line) - for line in events_path.read_text(encoding="utf-8").splitlines() - ] - for event in events: - if event.get("cube_id") == "010": - event["worker"] = 1 - encoded = "".join( - json.dumps(event, sort_keys=True, separators=(",", ":")) + "\n" - for event in events - ) - events_path.write_text(encoded, encoding="utf-8") - manifest_path = bundle / "bundle.json" - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) - manifest["events"]["sha256"] = hashlib.sha256( - encoded.encode("utf-8") - ).hexdigest() - manifest_path.write_text(json.dumps(manifest), encoding="utf-8") - result = self.run_bundle(bundle) - self.assertNotEqual(result.returncode, 0) - self.assertIn("worker 1 runs overlapping cube 010", result.stdout) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_conquer_parallel.py b/tests/test_conquer_parallel.py index 685f5ac..5d43e9b 100644 --- a/tests/test_conquer_parallel.py +++ b/tests/test_conquer_parallel.py @@ -1,64 +1,104 @@ import concurrent.futures -import json -import sys + +import pytest from benchmarks.cnc import conquer_parallel +from benchmarks.cnc.solve import conquer_frontier -def test_parallel_conquer_streams_cubes_and_reports_conflict_distribution( - tmp_path, monkeypatch -): +def test_parallel_conquer_reports_unsat_distribution(tmp_path, monkeypatch): cnf = tmp_path / "base.cnf" - cnf.write_text("p cnf 2 1\n1 2 0\n") + cnf.write_text("p cnf 2 1\n1 2 0\n", encoding="utf-8") cubes = tmp_path / "cubes.icnf" - cubes.write_text("".join(f"a {literal} 0\n" for literal in [1, -1, 2, -2, 1])) - fake_kissat = tmp_path / "kissat" - fake_kissat.write_text( + cubes.write_text("".join(f"a {literal} 0\n" for literal in [1, -1, 2, -2])) + kissat = tmp_path / "kissat" + kissat.write_text( "#!/bin/sh\n" - "case \" $* \" in *\" --statistics \"*) ;; *) exit 2 ;; esac\n" "echo 'c decisions: 7'\n" "echo 'c conflicts: 3'\n" - "exit 20\n" + "exit 20\n", + encoding="utf-8", ) - fake_kissat.chmod(0o755) - out_dir = tmp_path / "out" - temp_dir = tmp_path / "tmp" - + kissat.chmod(0o755) monkeypatch.setattr( concurrent.futures, "ProcessPoolExecutor", concurrent.futures.ThreadPoolExecutor, ) + + result = conquer_frontier( + cnf, + cubes, + kissat, + workers=2, + timeout_s=5, + out_dir=tmp_path / "out", + tmp_dir=tmp_path / "tmp", + ) + + assert result["result"] == "unsat" + assert result["complete"] is True + assert result["completed"] == 4 + assert result["total_decisions"] == 28 + assert result["total_conflicts"] == 12 + assert result["not_started"] == 0 + + +def test_parallel_conquer_stops_submitting_after_sat(tmp_path, monkeypatch): + cnf = tmp_path / "base.cnf" + cnf.write_text("p cnf 1 0\n", encoding="utf-8") + cubes = tmp_path / "cubes.icnf" + cubes.write_text("a 1 0\n" + "a -1 0\n" * 9, encoding="utf-8") + kissat = tmp_path / "kissat" + kissat.write_text("#!/bin/sh\nexit 10\n", encoding="utf-8") + kissat.chmod(0o755) monkeypatch.setattr( - sys, - "argv", - [ - "conquer_parallel.py", - str(cnf), - "--arm", - f"test={cubes}", - "--kissat", - str(fake_kissat), - "--workers", - "2", - "--timeout-s", - "5", - "--tmp-dir", - str(temp_dir), - "--out-dir", - str(out_dir), - ], + concurrent.futures, + "ProcessPoolExecutor", + concurrent.futures.ThreadPoolExecutor, ) - conquer_parallel.main() - summary = json.loads((out_dir / "summary.json").read_text()) - result = summary["arms"]["test"] - assert result["cubes"] == 5 - assert result["completed"] == 5 - assert result["errors"] == 0 - assert result["timeouts"] == 0 - assert result["total_decisions"] == 35 - assert result["total_conflicts"] == 15 - assert result["conflicts_p50"] == 3 - assert result["conflicts_p99_over_p95"] == 1 - assert len((out_dir / "test.jsonl").read_text().splitlines()) == 5 + result = conquer_frontier( + cnf, + cubes, + kissat, + workers=1, + timeout_s=5, + out_dir=tmp_path / "out", + ) + + assert result["result"] == "sat" + assert result["complete"] is True + assert result["completed"] == 1 + assert result["not_started"] == 9 + + +def test_frontier_literals_are_checked_against_cnf_variables(tmp_path): + frontier = tmp_path / "bad.icnf" + frontier.write_text("a 3 0\n", encoding="utf-8") + with pytest.raises(ValueError, match="variable range"): + list(conquer_parallel.read_cubes(frontier, variables=2)) + + +def test_censored_cubes_remain_in_work_accounting(): + result = conquer_parallel.summarize( + cubes=2, + completed=1, + timeouts=1, + errors=0, + sat=0, + unsat=1, + durations=[1.0, 5.0], + cpu_durations=[0.5, 4.5], + decisions=[7.0], + conflicts=[3.0], + workers=2, + replay_workers=[2], + wall_s=5.2, + measured_makespan_s=5.0, + ) + assert result["result"] == "timeout" + assert result["complete"] is False + assert result["censored"] is True + assert result["total_solver_s"] == 6.0 + assert result["lpt_is_lower_bound"] is True diff --git a/tests/test_cubing.py b/tests/test_cubing.py new file mode 100644 index 0000000..d30ff59 --- /dev/null +++ b/tests/test_cubing.py @@ -0,0 +1,58 @@ +from pathlib import Path + +from benchmarks.cnc import cubing + + +def successful_cuber(command): + frontier = Path(command[command.index("-o") + 1]) + frontier.parent.mkdir(parents=True, exist_ok=True) + frontier.write_text("a 1 0\na -1 0\n", encoding="utf-8") + + +def fake_run_process(command, **kwargs): + successful_cuber(command) + return {"returncode": 0, "timed_out": False, "wall_s": 0.1} + + +def fake_conquer(*args, **kwargs): + return {"result": "unsat", "complete": True, "cubes": 2} + + +def test_march_pipeline_uses_cnf_then_shared_conquer(tmp_path, monkeypatch): + monkeypatch.setattr(cubing, "run_process", fake_run_process) + monkeypatch.setattr(cubing, "conquer_frontier", fake_conquer) + + record = cubing.march_then_conquer( + tmp_path / "instance.cnf", + tmp_path / "march_cu", + tmp_path / "kissat", + workers=4, + cube_timeout_s=5, + cubing_timeout_s=5, + remaining_vars=20, + out_dir=tmp_path / "run", + ) + + assert record["mode"] == "march-cu" + assert record["cubing"]["cubes"] == 2 + assert record["cubing"]["command"][1].endswith("instance.cnf") + assert record["cubing"]["command"][-4:-2] == ["-n", "20"] + + +def test_frozen_frontier_goes_directly_to_shared_conquer(tmp_path, monkeypatch): + monkeypatch.setattr(cubing, "conquer_frontier", fake_conquer) + (tmp_path / "instance.cnf").write_text("p cnf 1 0\n", encoding="utf-8") + (tmp_path / "frontier.icnf").write_text("a 1 0\na -1 0\n", encoding="utf-8") + + record = cubing.frozen_then_conquer( + tmp_path / "instance.cnf", + tmp_path / "frontier.icnf", + tmp_path / "kissat", + workers=4, + cube_timeout_s=5, + out_dir=tmp_path / "run", + ) + + assert record["mode"] == "frozen-frontier" + assert record["cubes"] == 2 + assert record["frontier"].endswith("frontier.icnf") diff --git a/tests/test_factoring_corpus.py b/tests/test_factoring_corpus.py new file mode 100644 index 0000000..9f0675a --- /dev/null +++ b/tests/test_factoring_corpus.py @@ -0,0 +1,45 @@ +import json + +import pytest + +from benchmarks.cnc.factoring import materialize, sat_targets, unsat_targets +from benchmarks.pipeline.circuit import load_json, read_jsonl, sha256_file +from benchmarks.pipeline.multipliers import is_prime + + +def test_sat_and_unsat_targets_are_deterministic_and_paired(): + first_sat, _ = sat_targets(12, 3) + second_sat, _ = sat_targets(12, 3) + unsat, oracles = unsat_targets(12, first_sat) + + assert first_sat == second_sat + assert all(target.paired_sat_id == sat.instance_id for target, sat in zip(unsat, first_sat)) + assert all(is_prime(target.target) for target in unsat) + assert all(oracle["target_exceeds_max_factor"] for oracle in oracles) + + +def test_tiny_width_rejects_more_distinct_instances_than_exist(): + with pytest.raises(ValueError, match="only 3 distinct"): + sat_targets(4, 4) + + +def test_materialize_writes_matching_circuit_and_cnf(tmp_path): + manifest = materialize([4], 1, tmp_path) + + assert len(manifest) == 2 + assert {row["expected_outcome"] for row in manifest} == {"sat", "unsat"} + for row in manifest: + circuit_path = tmp_path / row["circuit"] + cnf_path = tmp_path / row["cnf"] + circuit = load_json(circuit_path) + assert circuit["metadata"]["factoring"]["target"] == row["target"] + assert circuit["metadata"]["pinned_outputs"]["product"] == row["target"] + assert cnf_path.read_text(encoding="utf-8").startswith("c var ") + assert "\np cnf " in cnf_path.read_text(encoding="utf-8") + assert sha256_file(circuit_path) == row["circuit_sha256"] + assert sha256_file(cnf_path) == row["cnf_sha256"] + + assert read_jsonl(tmp_path / "manifest.jsonl") == manifest + oracles = read_jsonl(tmp_path / "oracles.jsonl") + assert len(oracles) == 2 + assert json.loads((tmp_path / manifest[0]["metadata"]).read_text())["target"] == manifest[0]["target"] diff --git a/tests/test_solve.py b/tests/test_solve.py new file mode 100644 index 0000000..8c68ba8 --- /dev/null +++ b/tests/test_solve.py @@ -0,0 +1,24 @@ +import json + +from benchmarks.cnc.solve import run_kissat + + +def test_run_kissat_records_direct_result(tmp_path): + cnf = tmp_path / "instance.cnf" + cnf.write_text("p cnf 1 1\n1 0\n", encoding="utf-8") + kissat = tmp_path / "kissat" + kissat.write_text( + "#!/bin/sh\n" + "echo 'c decisions: 7'\n" + "echo 'c conflicts: 3'\n" + "exit 10\n", + encoding="utf-8", + ) + kissat.chmod(0o755) + + record = run_kissat(cnf, kissat, timeout_s=5, out_dir=tmp_path / "run") + + assert record["result"] == "sat" + assert record["decisions"] == 7 + assert record["conflicts"] == 3 + assert json.loads((tmp_path / "run/summary.json").read_text()) == record