diff options
Diffstat (limited to 'external/dualprop_patches/0015-experiment-audit-P2-execution-and-resource-use.patch')
| -rw-r--r-- | external/dualprop_patches/0015-experiment-audit-P2-execution-and-resource-use.patch | 341 |
1 files changed, 341 insertions, 0 deletions
diff --git a/external/dualprop_patches/0015-experiment-audit-P2-execution-and-resource-use.patch b/external/dualprop_patches/0015-experiment-audit-P2-execution-and-resource-use.patch new file mode 100644 index 0000000..8a019cd --- /dev/null +++ b/external/dualprop_patches/0015-experiment-audit-P2-execution-and-resource-use.patch @@ -0,0 +1,341 @@ +From 0e868b670e05e123a523ea18986b3e1423e11726 Mon Sep 17 00:00:00 2001 +From: YurenHao0426 <Blackhao0426@gmail.com> +Date: Mon, 27 Jul 2026 13:50:34 -0500 +Subject: [PATCH 15/19] experiment: audit P2 execution and resource use + +--- + crossover_grid.py | 289 +++++++++++++++++++++++++++++++++++++++++++++- + 1 file changed, 288 insertions(+), 1 deletion(-) + +diff --git a/crossover_grid.py b/crossover_grid.py +index 846a2f1..cbd9b98 100644 +--- a/crossover_grid.py ++++ b/crossover_grid.py +@@ -5,6 +5,7 @@ import glob + import hashlib + import json + import os ++import signal + import subprocess + import sys + import time +@@ -37,6 +38,30 @@ def sha256(path): + return digest.hexdigest() + + ++def sha256_text(value): ++ return hashlib.sha256(value.encode("utf-8")).hexdigest() ++ ++ ++def hash_tree(path): ++ digest = hashlib.sha256() ++ files = [] ++ for directory, _, names in os.walk(path): ++ for name in sorted(names): ++ files.append(os.path.join(directory, name)) ++ for filename in sorted(files): ++ relative = os.path.relpath(filename, path) ++ digest.update(relative.encode("utf-8") + b"\0") ++ with open(filename, "rb") as handle: ++ for chunk in iter(lambda: handle.read(1024 * 1024), b""): ++ digest.update(chunk) ++ return { ++ "path": os.path.realpath(path), ++ "sha256": digest.hexdigest(), ++ "num_files": len(files), ++ "bytes": sum(os.path.getsize(filename) for filename in files), ++ } ++ ++ + def source_report(): + commit = output("git", "rev-parse", "HEAD") + dirty = output( +@@ -52,14 +77,21 @@ def source_report(): + text=True).stdout.strip() + if main_dirty: + raise RuntimeError("crossover runs require clean tracked main source") +- return { ++ pip_freeze = output(sys.executable, "-m", "pip", "freeze") ++ report = { + "author_crossover_commit": commit, + "main_commit": main_commit, + "protocol_path": PROTOCOL, + "protocol_sha256": sha256(PROTOCOL), + "python": sys.executable, + "cuda_visible_devices": os.environ.get("CUDA_VISIBLE_DEVICES"), ++ "pip_freeze_sha256": sha256_text(pip_freeze), ++ "python_version": sys.version, + } ++ dataset = os.path.join(ROOT, "data", "cifar10", "3.0.2") ++ if os.path.isdir(dataset): ++ report["cifar10_tfds_tree"] = hash_tree(dataset) ++ return report + + + def rate_tag(rate): +@@ -345,7 +377,262 @@ def completed_histogram(experiment_name): + return paths[0] + + ++def physical_gpu_report(): ++ index = os.environ.get("CUDA_VISIBLE_DEVICES") ++ if index not in ("5", "7"): ++ raise RuntimeError( ++ "formal P2 is authorized only on physical GPUs 5 and 7") ++ query = output( ++ "nvidia-smi", ++ "--query-gpu=index,uuid,name,memory.total", ++ "--format=csv,noheader,nounits") ++ rows = [row.split(", ", 3) for row in query.splitlines()] ++ matches = [row for row in rows if row[0] == index] ++ if len(matches) != 1: ++ raise RuntimeError(f"cannot resolve physical GPU {index}") ++ physical_index, uuid, name, memory_mib = matches[0] ++ return { ++ "physical_index": int(physical_index), ++ "uuid": uuid, ++ "name": name, ++ "total_memory_mib": int(memory_mib), ++ } ++ ++ ++def descendant_pids(root_pid): ++ listing = subprocess.run( ++ ["ps", "-eo", "pid=,ppid="], check=True, capture_output=True, ++ text=True).stdout ++ children = {} ++ for row in listing.splitlines(): ++ pid, parent = (int(value) for value in row.split()) ++ children.setdefault(parent, []).append(pid) ++ descendants = {root_pid} ++ frontier = [root_pid] ++ while frontier: ++ parent = frontier.pop() ++ for child in children.get(parent, ()): ++ if child not in descendants: ++ descendants.add(child) ++ frontier.append(child) ++ return descendants ++ ++ ++def process_gpu_memory_mib(root_pid, gpu_uuid): ++ query = subprocess.run( ++ [ ++ "nvidia-smi", ++ "--query-compute-apps=pid,gpu_uuid,used_memory", ++ "--format=csv,noheader,nounits", ++ ], ++ check=True, capture_output=True, text=True).stdout ++ descendants = descendant_pids(root_pid) ++ total = 0 ++ for row in query.splitlines(): ++ pid, uuid, memory = row.split(", ") ++ if uuid == gpu_uuid and int(pid) in descendants: ++ total += int(memory) ++ return total ++ ++ ++def architecture_work(architecture): ++ specifications = { ++ "minicnn": ( ++ [64, 64], [True, True], [10]), ++ "vgglike": ( ++ [128, 256, 512, 512], [True, True, True, True], [10]), ++ "vgg16": ( ++ [64, 64, 128, 128, 256, 256, 256, 512, 512, 512, 512, ++ 512, 512], ++ [False, True, False, True, False, False, True, False, ++ False, True, False, False, True], ++ [4096, 4096, 10]), ++ } ++ features, pooling, dense = specifications[architecture] ++ height = width = 32 ++ channels = 3 ++ parameters = 0 ++ forward_macs = 0 ++ for output_channels, pool in zip(features, pooling): ++ forward_macs += ( ++ height * width * channels * output_channels * 3 * 3) ++ parameters += 3 * 3 * channels * output_channels + output_channels ++ channels = output_channels ++ if pool: ++ height //= 2 ++ width //= 2 ++ inputs = height * width * channels ++ for outputs in dense: ++ forward_macs += inputs * outputs ++ parameters += inputs * outputs + outputs ++ inputs = outputs ++ return { ++ "forward_parameter_count": parameters, ++ "forward_affine_macs_per_example": forward_macs, ++ "num_trainable_layers": len(features) + len(dense), ++ } ++ ++ ++def p2_work_report(job): ++ epochs = job["expected_epochs"] ++ method = job["method"] ++ architecture = architecture_work(job["architecture"]) ++ train_examples = 45000 * epochs ++ validation_examples = 5000 if method == "ff" else 5000 * epochs ++ layers = architecture["num_trainable_layers"] ++ presentations = train_examples ++ if method == "pepita": ++ presentations *= 2 ++ elif method == "ff": ++ presentations *= 2 * layers ++ feedforward_train_multipliers = { ++ "bp": 2, ++ "fa": 2, ++ "dfa": 2, ++ "pepita": 3, ++ "ff": 2 * layers, ++ "ep": 1, ++ "dualprop": 1, ++ "clean_kp": 2, ++ "sdil": 2, ++ } ++ relaxation_passes = 0 ++ if method == "ep": ++ relaxation_passes = train_examples * (20 + 4) ++ relaxation_passes += validation_examples * 20 ++ elif method == "dualprop": ++ relaxation_passes = train_examples * 16 ++ candidate_evaluations = 10 * validation_examples if method == "ff" else 0 ++ feedforward_passes = ( ++ train_examples * feedforward_train_multipliers[method] ++ + (0 if method == "ff" else validation_examples) ++ + candidate_evaluations) ++ return { ++ **architecture, ++ "ordinary_training_examples": train_examples, ++ "ordinary_validation_examples": validation_examples, ++ "training_example_presentations": presentations, ++ "feedforward_example_passes": feedforward_passes, ++ "relaxation_example_passes": relaxation_passes, ++ "candidate_label_evaluation_presentations": candidate_evaluations, ++ "task_loss_query_batches": ( ++ 0 if method == "ff" else 450 * epochs), ++ "neutral_predictor_observations": 64 if method == "sdil" else 0, ++ "counting_note": ( ++ "Explicit example/pass counts; affine correlation work is " ++ "audited separately and is not inferred from wall time."), ++ } ++ ++ ++def record_path(experiment_name): ++ paths = glob.glob(os.path.join( ++ ROOT, "runs", experiment_name, "*", "crossover_manifest.json")) ++ if len(paths) > 1: ++ raise RuntimeError( ++ f"expected at most one record for {experiment_name}") ++ return paths[0] if paths else None ++ ++ ++def result_directory(experiment_name, create_if_missing=False): ++ root = os.path.join(ROOT, "runs", experiment_name) ++ paths = [ ++ path for path in glob.glob( ++ os.path.join(root, "*")) ++ if os.path.isdir(path) ++ ] ++ if not paths and create_if_missing: ++ directory = os.path.join( ++ root, "driver_" + time.strftime("%Y_%m_%d_%H_%M_%S")) ++ os.makedirs(directory) ++ paths = [directory] ++ if len(paths) != 1: ++ raise RuntimeError( ++ f"expected one run directory for {experiment_name}, " ++ f"found {len(paths)}") ++ return paths[0] ++ ++ ++def terminate_process_group(process): ++ try: ++ os.killpg(process.pid, signal.SIGTERM) ++ process.wait(timeout=10) ++ except subprocess.TimeoutExpired: ++ os.killpg(process.pid, signal.SIGKILL) ++ process.wait() ++ ++ ++def run_p2_job(job, source, dry_run): ++ existing = record_path(job["experiment_name"]) ++ if existing is not None: ++ print(f"preserving {job['experiment_name']} -> {existing}", flush=True) ++ return ++ run_root = os.path.join(ROOT, "runs", job["experiment_name"]) ++ if os.path.isdir(run_root): ++ raise RuntimeError( ++ f"orphaned P2 directory requires audit: {run_root}") ++ print("RUN", " ".join(job["command"]), flush=True) ++ if dry_run: ++ return ++ gpu = physical_gpu_report() ++ started = time.time() ++ process = subprocess.Popen( ++ job["command"], cwd=ROOT, start_new_session=True) ++ peak_memory_mib = 0 ++ timed_out = False ++ while process.poll() is None: ++ elapsed = time.time() - started ++ if elapsed > job["timeout_seconds"]: ++ timed_out = True ++ terminate_process_group(process) ++ break ++ try: ++ peak_memory_mib = max( ++ peak_memory_mib, ++ process_gpu_memory_mib(process.pid, gpu["uuid"])) ++ except (OSError, subprocess.SubprocessError, ValueError): ++ pass ++ time.sleep(1) ++ return_code = process.returncode ++ elapsed = time.time() - started ++ histogram = completed_histogram(job["experiment_name"]) ++ if timed_out: ++ status = "timeout" ++ elif return_code != 0: ++ status = "nonzero_exit" ++ elif histogram is None: ++ status = "missing_histogram" ++ else: ++ status = "completed" ++ directory = result_directory( ++ job["experiment_name"], create_if_missing=True) ++ manifest = { ++ **job, ++ "source": source, ++ "status": status, ++ "return_code": return_code, ++ "histogram": histogram, ++ "histogram_sha256": sha256(histogram) if histogram else None, ++ "driver_wall_seconds": elapsed, ++ "completed_unix_time": time.time(), ++ "hardware": { ++ **gpu, ++ "peak_process_gpu_memory_mib": peak_memory_mib, ++ "memory_sampling_period_seconds": 1, ++ }, ++ "work": p2_work_report(job), ++ } ++ manifest_path = os.path.join(directory, "crossover_manifest.json") ++ with open(manifest_path, "w") as handle: ++ json.dump(manifest, handle, indent=2, sort_keys=True) ++ handle.write("\n") ++ print( ++ f"DONE status={status} {job['experiment_name']} -> {manifest_path}", ++ flush=True) ++ ++ + def run_job(job, source, dry_run): ++ if job["stage"] == "p2": ++ return run_p2_job(job, source, dry_run) + result = completed_histogram(job["experiment_name"]) + if result is not None: + print(f"preserving {job['experiment_name']} -> {result}", flush=True) +-- +2.54.0 + |