From 0e868b670e05e123a523ea18986b3e1423e11726 Mon Sep 17 00:00:00 2001 From: YurenHao0426 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