diff options
| -rw-r--r-- | RESNET_CROSSOVER.md | 17 | ||||
| -rw-r--r-- | experiments/analyze_resnet_crossover_r2.py | 430 | ||||
| -rwxr-xr-x | experiments/finalize_accept.sh | 4 | ||||
| -rw-r--r-- | experiments/resnet_crossover_native.py | 11 | ||||
| -rw-r--r-- | experiments/resnet_crossover_r2.py | 445 | ||||
| -rw-r--r-- | experiments/resnet_crossover_r2_smoke.py | 118 |
6 files changed, 1025 insertions, 0 deletions
diff --git a/RESNET_CROSSOVER.md b/RESNET_CROSSOVER.md index 126c051..ee1b28e 100644 --- a/RESNET_CROSSOVER.md +++ b/RESNET_CROSSOVER.md @@ -110,6 +110,23 @@ and its manifest records the resolved physical UUID. Formal launch occurs from a detached worktree so unrelated main-branch commits cannot mix revisions inside a shard. +The R2 executable contract is +`experiments/resnet_crossover_r2.py`. It refuses any selector that is not the +complete 19-record P1 pass, requires the selector itself to be committed, +binds its hash and selected rates into a new clean-source launch lock, and +materializes all 27 method--depth cells before sharding. The two shards +interleave EP, Dual Propagation, and Forward--Forward rather than assigning all +expensive state methods to one GPU. +`experiments/analyze_resnet_crossover_r2.py` requires every manifest, retains +timeout/nonfinite outcomes, rejects test access and +source/split/architecture drift, and checks method-specific presentation, +relaxation, local-VJP, and logical-query counts. The registry and schedule can +be checked before P1 finishes with: + +```bash +python experiments/resnet_crossover_r2_smoke.py +``` + ## R3 untouched confirmation No R3 test endpoint opens until all 27 R2 records and the complete analyzer diff --git a/experiments/analyze_resnet_crossover_r2.py b/experiments/analyze_resnet_crossover_r2.py new file mode 100644 index 0000000..3c2bf46 --- /dev/null +++ b/experiments/analyze_resnet_crossover_r2.py @@ -0,0 +1,430 @@ +#!/usr/bin/env python3 +"""Audit the complete failure-retaining 27-cell ResNet R2 panel.""" +import argparse +import hashlib +import json +import math +import os +import sys + + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, ROOT) +from experiments.resnet_crossover_r2 import ( + DEPTHS, + METHODS, + DEFAULT_SELECTOR, + r2_jobs, + registry_sha256, + selector_report, +) + + +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 read_json(path): + with open(path, encoding="utf-8") as handle: + return json.load(handle) + + +def finite(value): + return ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and math.isfinite(float(value)) + ) + + +def expected_mode(method): + return { + "bp": "bp", + "fa": "hfa", + "dfa": "dfa", + "clean_kp": "kp", + "sdil": "kp_traffic", + }[method] + + +def history_metrics(record, job): + method = job["method"] + expected_epochs = job["epochs"] + losses = [] + validation_accuracies = [] + history_examples = 0 + if method == "ff": + layers = record.get("layers", []) + layer_indices = [row.get("layer") for row in layers] + assert layer_indices == list(range(len(layers))) + for layer in layers: + epochs = layer.get("epochs", []) + assert [row.get("epoch") for row in epochs] == list( + range(1, len(epochs) + 1) + ) + assert len(epochs) <= expected_epochs + losses.extend(row.get("loss") for row in epochs) + history_examples += sum( + int(row.get("examples", 0)) for row in epochs + ) + complete = ( + len(layers) == job["depth"] + and all( + len(layer.get("epochs", [])) == expected_epochs + for layer in layers + ) + ) + completed_epochs = sum( + len(layer.get("epochs", [])) for layer in layers + ) + else: + epochs = record.get("epochs", []) + assert [row.get("epoch") for row in epochs] == list( + range(1, len(epochs) + 1) + ) + assert len(epochs) <= expected_epochs + complete = len(epochs) == expected_epochs + completed_epochs = len(epochs) + if method in ("pepita", "ep", "dualprop"): + losses = [row.get("train_loss") for row in epochs] + history_examples = sum( + int(row.get("train_examples", 0)) for row in epochs + ) + validation_accuracies = [ + row["validation"]["accuracy"] + for row in epochs + if row.get("validation") is not None + ] + else: + losses = [row.get("train_loss") for row in epochs] + history_examples = sum( + int(row.get("train_examples", 0)) for row in epochs + ) + validation_accuracies = [ + row["eval_accuracy"] + for row in epochs + if row.get("eval_accuracy") is not None + ] + final = record.get("final", {}) + final_accuracy = final.get("accuracy") + final_loss = final.get("loss") + finite_accuracies = [ + float(value) + for value in validation_accuracies + [final_accuracy] + if finite(value) + ] + best_accuracy = max(finite_accuracies) if finite_accuracies else None + losses_finite = bool(losses) and all(finite(value) for value in losses) + final_finite = ( + final.get("finite") is True + and finite(final_accuracy) + and (final_loss is None or finite(final_loss)) + ) + first_nonfinite = record.get("first_nonfinite_step") + return { + "complete_trajectory": complete, + "epochs_completed": completed_epochs, + "history_training_examples": history_examples, + "all_training_losses_finite": losses_finite, + "final_validation_accuracy": + float(final_accuracy) if finite(final_accuracy) else None, + "best_validation_accuracy": + float(best_accuracy) if best_accuracy is not None else None, + "final_validation_loss": + float(final_loss) if finite(final_loss) else None, + "first_nonfinite_step": first_nonfinite, + "finite": ( + complete + and losses_finite + and final_finite + and first_nonfinite is None + ), + } + + +def work_metrics(record, job, history): + method = job["method"] + if method in ("pepita", "ff", "ep", "dualprop"): + work = record["work"] + ordinary = int(work["ordinary_training_examples"]) + validation = int(work["ordinary_validation_examples"]) + presentations = int(work["training_example_presentations"]) + relaxation = int(work["relaxation_example_passes"]) + candidate = int(work["candidate_label_evaluation_presentations"]) + queries = int(work["logical_task_loss_queries"]) + local_vjp = int(work["local_vjp_example_evaluations"]) + local_backward = int( + work["local_target_backward_example_evaluations"] + ) + assert ordinary == history["history_training_examples"] + if method == "ff": + assert presentations == 2 * ordinary + assert candidate == 10 * validation + assert relaxation == 0 + assert local_backward == 2 * ordinary + assert local_vjp == 0 + elif method == "pepita": + assert presentations == 2 * ordinary + assert relaxation == candidate == local_vjp == local_backward == 0 + elif method == "ep": + assert presentations == ordinary + assert relaxation == 24 * ordinary + 20 * validation + assert local_vjp == relaxation + assert candidate == local_backward == 0 + else: + assert presentations == ordinary + assert relaxation == 16 * ordinary + assert local_vjp == relaxation + assert candidate == local_backward == 0 + total_work = { + "forward_macs_per_example": work["forward_macs_per_example"], + "feedforward_example_passes": work["feedforward_example_passes"], + "relaxation_example_passes": relaxation, + "local_vjp_example_evaluations": local_vjp, + "candidate_label_evaluation_presentations": candidate, + } + else: + work = record["work"] + counters = record["counters"] + ordinary = int(counters["ordinary_examples"]) + validation = ( + int(record["split"]["validation_examples"]) + * int(record["evaluation_protocol"]["validation_evaluations"]) + ) + presentations = ordinary + queries = int(work["logical_batch_loss_queries"]) + total_work = { + "forward_macs_per_example": work["forward_macs_per_example"], + "total_macs_estimate": work["total_macs_estimate"], + "elementwise_operations_estimate": + work["elementwise_operations_estimate"], + "total_forward_equivalent_examples": + work["total_forward_equivalent_examples"], + } + assert ordinary == history["history_training_examples"] + assert queries == 0 + return { + "ordinary_training_examples": ordinary, + "ordinary_validation_examples": validation, + "training_example_presentations": presentations, + "logical_task_loss_queries": queries, + "work": total_work, + } + + +def audit_job(job, expected_source, expected_selector): + manifest_path = job["output"] + ".manifest.json" + if not os.path.isfile(manifest_path): + raise AssertionError(f"missing R2 manifest: {job['experiment_name']}") + manifest = read_json(manifest_path) + for key in ( + "stage", + "method", + "architecture", + "depth", + "rate", + "epochs", + "lr_schedule", + "experiment_name", + "output", + "timeout_seconds", + "command", + ): + assert manifest[key] == job[key], ( + f"{job['experiment_name']}: manifest drift in {key}" + ) + assert manifest["source"] == expected_source + assert manifest["selector"] == expected_selector + hardware = manifest["hardware_lock"] + assert hardware["physical_gpu_index"] in (5, 7) + assert hardware["physical_gpu_uuid"] + common = { + "cell_id": f"resnet{job['depth']}::{job['method']}", + "method": job["method"], + "depth": job["depth"], + "rate": job["rate"], + "status": manifest["status"], + "manifest": os.path.relpath(manifest_path, ROOT), + "driver_wall_seconds": float(manifest["driver_wall_seconds"]), + "physical_gpu_index": hardware["physical_gpu_index"], + "physical_gpu_uuid": hardware["physical_gpu_uuid"], + "output_sha256": manifest["output_sha256"], + } + if manifest["status"] != "completed": + assert manifest["status"] in { + "timeout", "nonzero_exit", "missing_output" + } + assert ( + manifest["output_sha256"] is None + if not manifest["output_exists"] + else manifest["output_sha256"] == sha256(job["output"]) + ) + return { + **common, + "finite": False, + "complete_trajectory": False, + "epochs_completed": 0, + "best_validation_accuracy": None, + "final_validation_accuracy": None, + "final_validation_loss": None, + "first_nonfinite_step": None, + "ordinary_training_examples": None, + "ordinary_validation_examples": None, + "training_example_presentations": None, + "logical_task_loss_queries": None, + "forward_parameter_count": None, + "peak_memory_allocated_bytes": None, + "total_wall_seconds": None, + "work": None, + } + assert manifest["output_exists"] is True + assert manifest["output_sha256"] == sha256(job["output"]) + record = read_json(job["output"]) + provenance = record["provenance"] + assert provenance["git_commit"] == expected_source["git_commit"] + assert provenance["git_tracked_dirty"] is False + args = record["args"] + for key, expected in { + "depth": job["depth"], + "width": 16, + "seed": 0, + "loader_seed": 0, + "split_seed": 2027, + "batch_size": 128, + "epochs": job["epochs"], + "train_limit": 0, + "val_examples": 5000, + "eval_split": "validation", + "eval_every": 1, + "augment_train": 1, + "lr": job["rate"], + "lr_schedule": job["lr_schedule"], + "momentum": 0.9, + "weight_decay": 1e-4, + }.items(): + assert args[key] == expected, ( + f"{job['experiment_name']}: argument drift in {key}" + ) + if job["method"] in ("pepita", "ff", "ep", "dualprop"): + assert args["method"] == job["method"] + else: + assert args["mode"] == expected_mode(job["method"]) + split = record["split"] + assert split["train_examples"] == 45_000 + assert split["validation_examples"] == 5_000 + assert split["split_seed"] == 2027 + evaluation = record["evaluation_protocol"] + assert evaluation["test_evaluations"] == 0 + assert evaluation["test_used_for_selection"] is False + architecture = record["architecture"] + assert architecture["depth"] == job["depth"] + assert architecture["base_width"] == 16 + history = history_metrics(record, job) + work = work_metrics(record, job, history) + if history["complete_trajectory"]: + expected_ordinary = 45_000 * job["epochs"] + if job["method"] == "ff": + expected_ordinary *= job["depth"] + assert work["ordinary_training_examples"] == expected_ordinary + hardware_record = record["hardware"] + assert hardware_record["cuda_visible_devices"] in ("5", "7") + assert hardware_record["cuda_device_name"] == "NVIDIA GeForce GTX 1080" + parameter_count = architecture.get("forward_parameter_count") + if parameter_count is None: + parameter_count = architecture["forward_parameters"] + return { + **common, + **history, + **work, + "forward_parameter_count": int(parameter_count), + "peak_memory_allocated_bytes": int( + hardware_record["peak_memory_allocated_bytes"] + ), + "total_wall_seconds": float( + record.get("total_wall_seconds") + if "total_wall_seconds" in record + else record["timing"]["total_timed_wall_s"] + ), + } + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--selector", default=DEFAULT_SELECTOR) + parser.add_argument( + "--out", default="results/resnet_crossover/r2_audit.json" + ) + args = parser.parse_args() + selector = selector_report(args.selector) + jobs = r2_jobs(selector["selected_rates"]) + launch_path = os.path.join( + ROOT, "results", "resnet_crossover", "r2_launch.json" + ) + assert os.path.isfile(launch_path), "missing ResNet R2 launch lock" + launch = read_json(launch_path) + assert launch["stage"] == "r2" + assert launch["selector"] == selector + assert launch["registry_sha256"] == registry_sha256(jobs) + assert launch["num_jobs"] == 27 + assert launch["allowed_physical_gpus"] == [5, 7] + source = launch["source"] + records = [audit_job(job, source, selector) for job in jobs] + assert len(records) == 27 + assert len({record["cell_id"] for record in records}) == 27 + assert { + (record["method"], record["depth"]) for record in records + } == {(method, depth) for method in METHODS for depth in DEPTHS} + for depth in DEPTHS: + counts = { + record["forward_parameter_count"] + for record in records + if record["depth"] == depth + and record["forward_parameter_count"] is not None + } + assert len(counts) <= 1, f"forward parameter mismatch at depth {depth}" + failures = [ + record["cell_id"] for record in records if not record["finite"] + ] + report = { + "audit_status": "passed", + "stage": "resnet_crossover_r2", + "complete_grid": True, + "failure_retaining": True, + "num_expected_cells": 27, + "num_audited_cells": len(records), + "num_finite_cells": len(records) - len(failures), + "failed_or_incomplete_cells": failures, + "test_policy": "none", + "source": source, + "selector": selector, + "launch_lock": { + "path": os.path.relpath(launch_path, ROOT), + "sha256": sha256(launch_path), + "registry_sha256": launch["registry_sha256"], + }, + "records": records, + } + os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True) + with open(args.out, "w", encoding="utf-8") as handle: + json.dump(report, handle, indent=2, sort_keys=True) + handle.write("\n") + print( + json.dumps( + { + "audit_status": report["audit_status"], + "num_audited_cells": len(records), + "num_finite_cells": report["num_finite_cells"], + "failed_or_incomplete_cells": failures, + }, + indent=2, + sort_keys=True, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/experiments/finalize_accept.sh b/experiments/finalize_accept.sh index ea311a0..9a511b9 100755 --- a/experiments/finalize_accept.sh +++ b/experiments/finalize_accept.sh @@ -23,6 +23,8 @@ experiments/finalize_claims.sh experiments/bci_v2_calibrated_smoke.py /home/yurenh2/miniconda3/envs/ep_pascal/bin/python3 \ experiments/oral_a_dynamic_scaling_smoke.py +/home/yurenh2/miniconda3/envs/ep_pascal/bin/python3 \ + experiments/resnet_crossover_r2_smoke.py /home/yurenh2/miniconda3/envs/ep_pascal/bin/python3 -m py_compile \ experiments/bci_td_run.py experiments/analyze_bci_td_development.py \ experiments/bci_td_confirmation.py \ @@ -38,6 +40,8 @@ experiments/finalize_claims.sh experiments/analyze_oral_a_dynamic_scaling.py \ experiments/oral_a_dynamic_scaling_v2.py \ experiments/analyze_oral_a_dynamic_scaling_v2.py \ + experiments/resnet_crossover_r2.py \ + experiments/analyze_resnet_crossover_r2.py \ experiments/plot_resnet_confirmation.py \ experiments/plot_oral_a_scaling.py \ experiments/plot_bci_v2_confirmation.py \ diff --git a/experiments/resnet_crossover_native.py b/experiments/resnet_crossover_native.py index e346678..0c2ba97 100644 --- a/experiments/resnet_crossover_native.py +++ b/experiments/resnet_crossover_native.py @@ -196,6 +196,13 @@ def work_report(net, args, ordinary_examples, validation_examples, if args.method == "ep": relaxation_examples += ( args.ep_free_steps * validation_examples) + local_vjp_examples = ( + relaxation_examples + if args.method in ("ep", "dualprop") else 0 + ) + local_target_backward_examples = ( + 2 * ordinary_examples if args.method == "ff" else 0 + ) return { "forward_parameter_count": net.n_forward_parameters, "forward_macs_per_example": net.forward_macs_per_example, @@ -206,6 +213,10 @@ def work_report(net, args, ordinary_examples, validation_examples, "feedforward_example_passes": feedforward_examples, "relaxation_example_passes": relaxation_examples, "candidate_label_evaluation_presentations": candidate_examples, + "logical_task_loss_queries": 0, + "local_vjp_example_evaluations": local_vjp_examples, + "local_target_backward_example_evaluations": + local_target_backward_examples, "completed_global_epochs": completed_epochs, } diff --git a/experiments/resnet_crossover_r2.py b/experiments/resnet_crossover_r2.py new file mode 100644 index 0000000..a9d5dfe --- /dev/null +++ b/experiments/resnet_crossover_r2.py @@ -0,0 +1,445 @@ +#!/usr/bin/env python3 +"""Immutable driver for the complete 27-cell ResNet R2 crossover.""" +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") +DEFAULT_SELECTOR = os.path.join(RESULT_ROOT, "p1_selector.json") +METHODS = ( + "bp", "fa", "dfa", "pepita", "ff", "ep", "dualprop", + "clean_kp", "sdil", +) +DEPTHS = (20, 32, 56) +ORDER = ( + ("ep", 56), + ("dualprop", 56), + ("ff", 56), + ("ep", 32), + ("dualprop", 32), + ("ff", 32), + ("ep", 20), + ("dualprop", 20), + ("ff", 20), + ("pepita", 56), + ("sdil", 56), + ("clean_kp", 56), + ("fa", 56), + ("dfa", 56), + ("bp", 56), + ("pepita", 32), + ("sdil", 32), + ("clean_kp", 32), + ("fa", 32), + ("dfa", 32), + ("bp", 32), + ("pepita", 20), + ("sdil", 20), + ("clean_kp", 20), + ("fa", 20), + ("dfa", 20), + ("bp", 20), +) + + +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 read_json(path): + with open(path, encoding="utf-8") as handle: + return json.load(handle) + + +def rate_tag(rate): + return f"{rate:g}".replace(".", "p") + + +def selector_report(path): + selector = read_json(path) + if not ( + selector.get("gate") == "pass" + and selector.get("stage") == "resnet_crossover_p1" + and selector.get("num_expected_records") == 19 + and selector.get("num_audited_records") == 19 + and selector.get("missing_experiments") == [] + and selector.get("test_policy") == "none" + and set(selector.get("selected", {})) == set(METHODS) + ): + raise RuntimeError("ResNet R2 requires the complete passed P1 selector") + selected = {} + for method in METHODS: + row = selector["selected"][method] + rate = float(row["rate"]) + if not rate > 0: + raise RuntimeError(f"invalid selected rate for {method}") + selected[method] = rate + return { + "path": os.path.relpath(os.path.abspath(path), ROOT), + "sha256": sha256(path), + "p1_source": selector["source"], + "selected_rates": selected, + } + + +def source_report(selector_path): + dirty = git_output("status", "--porcelain", "--untracked-files=no") + if dirty: + raise RuntimeError("ResNet R2 requires clean tracked source") + paths = [ + PROTOCOL, + os.path.abspath(__file__), + os.path.join(ROOT, "experiments", "analyze_resnet_crossover_r2.py"), + 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"), + os.path.abspath(selector_path), + ] + for path in paths: + relative = os.path.relpath(path, ROOT) + tracked = subprocess.run( + ["git", "ls-files", "--error-unmatch", relative], + cwd=ROOT, + capture_output=True, + ).returncode == 0 + if not tracked: + raise RuntimeError(f"untracked ResNet R2 source: {relative}") + 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 schedule(method): + if method == "pepita": + return { + "epochs": 100, + "lr_schedule": "pepita", + "lr_milestones": "60,90", + } + if method == "ff": + return { + "epochs": 40, + "lr_schedule": "constant", + "lr_milestones": "100,150", + } + if method == "ep": + return { + "epochs": 100, + "lr_schedule": "constant", + "lr_milestones": "100,150", + } + return { + "epochs": 200, + "lr_schedule": "step", + "lr_milestones": "100,150", + } + + +def r2_command(method, depth, rate): + name = ( + f"resnet-r2-{method}-d{depth}-lr{rate_tag(rate)}" + ) + output = os.path.join(RESULT_ROOT, "r2", name + ".json") + method_schedule = schedule(method) + output_rate = 0.1 if method in ("fa", "dfa") else rate + common = [ + "--device", "cuda", + "--depth", str(depth), + "--width", "16", + "--seed", "0", + "--loader_seed", "0", + "--split_seed", "2027", + "--batch_size", "128", + "--epochs", str(method_schedule["epochs"]), + "--train_limit", "0", + "--val_examples", "5000", + "--eval_split", "validation", + "--eval_every", "1", + "--augment_train", "1", + "--lr", str(rate), + "--output_lr", str(output_rate), + "--lr_schedule", method_schedule["lr_schedule"], + "--lr_milestones", method_schedule["lr_milestones"], + "--lr_gamma", "0.1", + "--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" if method == "bp" else "32", + *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": "r2", + "method": method, + "architecture": f"resnet{depth}", + "depth": depth, + "rate": rate, + "epochs": method_schedule["epochs"], + "lr_schedule": method_schedule["lr_schedule"], + "experiment_name": name, + "output": output, + "timeout_seconds": 48 * 60 * 60, + "command": command, + } + + +def r2_jobs(selected_rates): + if set(selected_rates) != set(METHODS): + raise ValueError("R2 requires one selected rate for every method") + jobs = [ + r2_command(method, depth, float(selected_rates[method])) + for method, depth in ORDER + ] + identifiers = [job["experiment_name"] for job in jobs] + cells = {(job["method"], job["depth"]) for job in jobs} + if ( + len(jobs) != 27 + or len(set(identifiers)) != 27 + or cells != {(method, depth) for method in METHODS for depth in DEPTHS} + ): + raise AssertionError("ResNet R2 registry must contain 27 unique cells") + return jobs + + +def registry_sha256(jobs): + encoded = json.dumps( + jobs, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def ensure_launch(source, selector, jobs): + path = os.path.join(RESULT_ROOT, "r2_launch.json") + expected = { + "stage": "r2", + "source": source, + "selector": selector, + "registry_sha256": registry_sha256(jobs), + "num_jobs": len(jobs), + "allowed_physical_gpus": [5, 7], + } + if os.path.isfile(path): + if read_json(path) != expected: + raise RuntimeError("ResNet R2 launch lock drift") + return path + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as handle: + json.dump(expected, handle, indent=2, sort_keys=True) + handle.write("\n") + return path + + +def assert_source_unchanged(source): + if git_output("status", "--porcelain", "--untracked-files=no"): + raise RuntimeError("tracked source changed after ResNet R2 launch") + if git_output("rev-parse", "HEAD") != source["git_commit"]: + raise RuntimeError("source commit changed after ResNet R2 launch") + for relative, expected_hash in source["tracked_files"].items(): + if sha256(os.path.join(ROOT, relative)) != expected_hash: + raise RuntimeError(f"ResNet R2 source drift: {relative}") + + +def physical_gpu_report(dry_run=False): + visible = os.environ.get("CUDA_VISIBLE_DEVICES") + if dry_run: + return { + "cuda_visible_devices": visible, + "physical_gpu_index": None, + "physical_gpu_uuid": None, + } + if visible not in {"5", "7"}: + raise RuntimeError("ResNet R2 requires physical GPU 5 or 7") + query = subprocess.run( + [ + "nvidia-smi", + "--query-gpu=index,uuid,name", + "--format=csv,noheader,nounits", + ], + check=True, + capture_output=True, + text=True, + ).stdout.splitlines() + rows = {} + for line in query: + index, uuid, name = [value.strip() for value in line.split(",", 2)] + rows[index] = {"uuid": uuid, "name": name} + if visible not in rows: + raise RuntimeError(f"physical GPU {visible} not found") + return { + "cuda_visible_devices": visible, + "physical_gpu_index": int(visible), + "physical_gpu_uuid": rows[visible]["uuid"], + "physical_gpu_name": rows[visible]["name"], + } + + +def run_job(job, source, selector, gpu, dry_run): + if not dry_run: + assert_source_unchanged(source) + if sha256(os.path.join(ROOT, selector["path"])) != selector["sha256"]: + raise RuntimeError("ResNet R2 selector changed after launch") + 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 R2 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, + "selector": selector, + "hardware_lock": gpu, + "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("--selector", default=DEFAULT_SELECTOR) + 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("--depth", type=int, choices=DEPTHS) + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args() + if not 0 <= args.shard_index < args.num_shards: + raise ValueError("invalid ResNet R2 shard") + selector = selector_report(args.selector) + source = source_report(args.selector) + gpu = physical_gpu_report(args.dry_run) + complete_jobs = r2_jobs(selector["selected_rates"]) + if not args.dry_run: + launch = ensure_launch(source, selector, complete_jobs) + print(f"R2 launch lock: {launch}", flush=True) + jobs = complete_jobs + if args.method: + jobs = [job for job in jobs if job["method"] == args.method] + if args.depth is not None: + jobs = [job for job in jobs if job["depth"] == args.depth] + jobs = [ + job + for index, job in enumerate(jobs) + if index % args.num_shards == args.shard_index + ] + if not jobs: + raise ValueError("no ResNet R2 jobs selected") + for job in jobs: + run_job(job, source, selector, gpu, args.dry_run) + + +if __name__ == "__main__": + main() diff --git a/experiments/resnet_crossover_r2_smoke.py b/experiments/resnet_crossover_r2_smoke.py new file mode 100644 index 0000000..d98c638 --- /dev/null +++ b/experiments/resnet_crossover_r2_smoke.py @@ -0,0 +1,118 @@ +#!/usr/bin/env python3 +"""Deterministic registry audit for the complete ResNet R2 crossover.""" +import os +import sys +from types import SimpleNamespace + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from experiments.resnet_crossover_r2 import DEPTHS, METHODS, r2_jobs +from experiments.resnet_crossover_native import work_report + + +RATES = { + "bp": 0.1, + "fa": 0.03, + "dfa": 0.03, + "pepita": 0.001, + "ff": 0.03, + "ep": 0.001, + "dualprop": 0.025, + "clean_kp": 0.1, + "sdil": 0.1, +} + + +def value(command, flag): + index = command.index(flag) + return command[index + 1] + + +def main(): + jobs = r2_jobs(RATES) + assert len(jobs) == 27 + assert {(job["method"], job["depth"]) for job in jobs} == { + (method, depth) for method in METHODS for depth in DEPTHS + } + assert len({job["experiment_name"] for job in jobs}) == 27 + assert len({job["output"] for job in jobs}) == 27 + for job in jobs: + method = job["method"] + command = job["command"] + expected_epochs = ( + 100 if method in ("pepita", "ep") + else 40 if method == "ff" + else 200 + ) + expected_schedule = ( + "pepita" if method == "pepita" + else "constant" if method in ("ff", "ep") + else "step" + ) + assert job["epochs"] == expected_epochs + assert job["lr_schedule"] == expected_schedule + assert value(command, "--depth") == str(job["depth"]) + assert value(command, "--epochs") == str(expected_epochs) + assert value(command, "--lr_schedule") == expected_schedule + assert value(command, "--eval_split") == "validation" + assert value(command, "--val_examples") == "5000" + assert value(command, "--eval_every") == "1" + assert value(command, "--lr") == str(RATES[method]) + assert "--test" not in command + assert job["timeout_seconds"] == 48 * 60 * 60 + if method in ("fa", "dfa"): + assert value(command, "--output_lr") == "0.1" + else: + assert value(command, "--output_lr") == str(RATES[method]) + if method == "pepita": + assert value(command, "--lr_milestones") == "60,90" + elif method not in ("ff", "ep"): + assert value(command, "--lr_milestones") == "100,150" + if method == "sdil": + assert value(command, "--traffic_ratio") == "4" + assert value(command, "--neutral_projection") == "1" + assert value(command, "--alignment_probe") == "32" + if method == "ep": + assert value(command, "--ep_free_steps") == "20" + assert value(command, "--ep_nudge_steps") == "4" + if method == "dualprop": + assert value(command, "--dp_inference_passes") == "16" + shard0 = jobs[0::2] + shard1 = jobs[1::2] + assert len(shard0) == 14 and len(shard1) == 13 + assert any(job["method"] == "ep" for job in shard0) + assert any(job["method"] == "ep" for job in shard1) + assert any(job["method"] == "dualprop" for job in shard0) + assert any(job["method"] == "dualprop" for job in shard1) + assert any(job["method"] == "ff" for job in shard0) + assert any(job["method"] == "ff" for job in shard1) + dummy_net = SimpleNamespace( + n_hidden=19, + n_forward_parameters=123, + forward_macs_per_example=456, + ) + for method in ("pepita", "ff", "ep", "dualprop"): + args = SimpleNamespace( + method=method, + ep_free_steps=20, + ep_nudge_steps=4, + dp_inference_passes=16, + ) + work = work_report(dummy_net, args, 100, 20, 1) + assert work["logical_task_loss_queries"] == 0 + if method == "ff": + assert work["local_target_backward_example_evaluations"] == 200 + assert work["candidate_label_evaluation_presentations"] == 200 + elif method == "ep": + assert work["relaxation_example_passes"] == 2800 + assert work["local_vjp_example_evaluations"] == 2800 + elif method == "dualprop": + assert work["relaxation_example_passes"] == 1600 + assert work["local_vjp_example_evaluations"] == 1600 + else: + assert work["training_example_presentations"] == 200 + assert work["local_vjp_example_evaluations"] == 0 + print("ResNet R2 registry: 27/27 cells; schedules and shards exact") + + +if __name__ == "__main__": + main() |