path: root/external/dualprop_patches/0015-experiment-audit-P2-execution-and-resource-use.patch

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