diff options
| -rw-r--r-- | experiments/resnet_crossover_grid.py | 255 |
1 files changed, 255 insertions, 0 deletions
diff --git a/experiments/resnet_crossover_grid.py b/experiments/resnet_crossover_grid.py new file mode 100644 index 0000000..f29e950 --- /dev/null +++ b/experiments/resnet_crossover_grid.py @@ -0,0 +1,255 @@ +#!/usr/bin/env python3 +"""Immutable job registry and driver for the ResNet crossover stages.""" +import argparse +import hashlib +import json +import os +import subprocess +import sys +import time + + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +PROTOCOL = os.path.join(ROOT, "RESNET_CROSSOVER.md") +RESULT_ROOT = os.path.join(ROOT, "results", "resnet_crossover") +METHODS = ( + "bp", "fa", "dfa", "pepita", "ff", "ep", "dualprop", "clean_kp", + "sdil") + + +def sha256(path): + digest = hashlib.sha256() + with open(path, "rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def git_output(*args): + return subprocess.run( + ["git", *args], cwd=ROOT, check=True, capture_output=True, + text=True).stdout.strip() + + +def source_report(): + dirty = git_output( + "status", "--porcelain", "--untracked-files=no") + if dirty: + raise RuntimeError("ResNet crossover requires clean tracked source") + paths = [ + PROTOCOL, + os.path.abspath(__file__), + os.path.join(ROOT, "experiments", "resnet_crossover_native.py"), + os.path.join(ROOT, "experiments", "conv_run.py"), + os.path.join(ROOT, "sdil", "conv_crossover.py"), + os.path.join(ROOT, "sdil", "conv.py"), + os.path.join(ROOT, "sdil", "data.py"), + ] + for path in paths: + tracked = subprocess.run( + ["git", "ls-files", "--error-unmatch", + os.path.relpath(path, ROOT)], + cwd=ROOT, capture_output=True).returncode == 0 + if not tracked: + raise RuntimeError(f"untracked crossover source: {path}") + return { + "git_commit": git_output("rev-parse", "HEAD"), + "protocol_path": PROTOCOL, + "protocol_sha256": sha256(PROTOCOL), + "tracked_files": { + os.path.relpath(path, ROOT): sha256(path) for path in paths}, + "python": sys.executable, + } + + +def rate_tag(rate): + return f"{rate:g}".replace(".", "p") + + +def p1_jobs(): + specifications = [ + ("ep", 0.0003), + ("dualprop", 0.01), + ("ep", 0.001), + ("dualprop", 0.025), + ("ep", 0.003), + ("dualprop", 0.05), + ("pepita", 0.0003), + ("fa", 0.01), + ("pepita", 0.001), + ("dfa", 0.01), + ("pepita", 0.003), + ("fa", 0.03), + ("dfa", 0.03), + ("fa", 0.1), + ("dfa", 0.1), + ("bp", 0.1), + ("ff", 0.03), + ("clean_kp", 0.1), + ("sdil", 0.1), + ] + return [p1_command(method, rate) for method, rate in specifications] + + +def p1_command(method, rate): + name = f"resnet-p1-{method}-d20-lr{rate_tag(rate)}" + output = os.path.join(RESULT_ROOT, "p1", name + ".json") + common = [ + "--device", "cuda", + "--depth", "20", + "--width", "16", + "--seed", "0", + "--loader_seed", "0", + "--split_seed", "2027", + "--batch_size", "128", + "--epochs", "10", + "--train_limit", "0", + "--val_examples", "5000", + "--eval_split", "validation", + "--eval_every", "1", + "--augment_train", "1", + "--lr", str(rate), + "--output_lr", "0.1" if method in ("fa", "dfa") else str(rate), + "--lr_schedule", "constant", + "--momentum", "0.9", + "--weight_decay", "1e-4", + ] + if method in ("pepita", "ff", "ep", "dualprop"): + command = [ + sys.executable, "experiments/resnet_crossover_native.py", + "--method", method, + "--out", output, + "--feedback_seed", "1729", + *common, + ] + if method == "pepita": + command.extend(["--pepita_projection_scale", "0.05"]) + elif method == "ff": + command.extend([ + "--ff_threshold", "2.0", + "--ff_score_from_layer", "1", + ]) + elif method == "ep": + command.extend([ + "--ep_beta", "0.5", + "--ep_dt", "0.5", + "--ep_free_steps", "20", + "--ep_nudge_steps", "4", + ]) + else: + command.extend([ + "--dp_alpha", "0.0", + "--dp_beta", "0.1", + "--dp_inference_passes", "16", + ]) + else: + modes = { + "bp": "bp", + "fa": "hfa", + "dfa": "dfa", + "clean_kp": "kp", + "sdil": "kp_traffic", + } + command = [ + sys.executable, "experiments/conv_run.py", + "--mode", modes[method], + "--out", output, + "--normalization", "batchnorm", + "--a_scale", "1", + "--alignment_probe", "0", + *common, + ] + if method == "dfa": + command.extend(["--vectorizer_mode", "spatial_template"]) + elif method == "sdil": + command.extend([ + "--traffic_rule", "innovation", + "--predictor_mode", "closed_form", + "--neutral_projection", "1", + "--traffic_seed", "5000", + "--traffic_ratio", "4", + "--traffic_calibration_examples", "64", + "--learn_P", "1", + "--eta_P", "0.1", + "--predictor_warmup_steps", "1", + "--predictor_every", "0", + ]) + return { + "stage": "p1", + "method": method, + "architecture": "resnet20", + "rate": rate, + "experiment_name": name, + "output": output, + "timeout_seconds": 12 * 60 * 60, + "command": command, + } + + +def run_job(job, source, dry_run): + manifest_path = job["output"] + ".manifest.json" + if os.path.exists(manifest_path): + print(f"preserving {job['experiment_name']}", flush=True) + return + if os.path.exists(job["output"]): + raise RuntimeError( + f"orphaned output requires audit: {job['output']}") + print("RUN", " ".join(job["command"]), flush=True) + if dry_run: + return + os.makedirs(os.path.dirname(job["output"]), exist_ok=True) + started = time.time() + try: + result = subprocess.run( + job["command"], cwd=ROOT, timeout=job["timeout_seconds"]) + return_code = result.returncode + status = "completed" if return_code == 0 else "nonzero_exit" + except subprocess.TimeoutExpired: + return_code = None + status = "timeout" + output_exists = os.path.isfile(job["output"]) + if status == "completed" and not output_exists: + status = "missing_output" + manifest = { + **job, + "source": source, + "status": status, + "return_code": return_code, + "output_exists": output_exists, + "output_sha256": sha256(job["output"]) if output_exists else None, + "driver_wall_seconds": time.time() - started, + "completed_unix_time": time.time(), + "cuda_visible_devices": os.environ.get("CUDA_VISIBLE_DEVICES"), + } + with open(manifest_path, "w", encoding="utf-8") as handle: + json.dump(manifest, handle, indent=2, sort_keys=True) + handle.write("\n") + print( + f"DONE status={status} {job['experiment_name']}", flush=True) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--stage", choices=("p1",), default="p1") + parser.add_argument("--shard-index", type=int, default=0) + parser.add_argument("--num-shards", type=int, default=1) + parser.add_argument("--method", choices=METHODS) + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args() + if not 0 <= args.shard_index < args.num_shards: + raise ValueError("invalid shard") + source = source_report() + jobs = p1_jobs() + if args.method: + jobs = [job for job in jobs if job["method"] == args.method] + jobs = [ + job for index, job in enumerate(jobs) + if index % args.num_shards == args.shard_index] + if not jobs: + raise ValueError("no jobs selected") + for job in jobs: + run_job(job, source, args.dry_run) + + +if __name__ == "__main__": + main() |