1"""Toy numerical verification for phase-blind checkpoint scheduling.
  2
  3This deliberately tests the mathematical mechanism, not distributed PyTorch
  4I/O.  Run with: python3 phase_blind.py
  5"""
  6import json
  7from pathlib import Path
  8import numpy as np
  9
 10
 11def anonymous_pair_map(delta, T=1.0, d=0.1):
 12    """Return map for identical periodic jobs on an anonymous resource."""
 13    if not (0 < d < T):
 14        raise ValueError("require 0 < d < T")
 15    return np.asarray(delta, dtype=float).copy()
 16
 17
 18def priority_pair_map(delta, alpha=0.25):
 19    """Toy identity/age-priority comparator with deterministic contraction."""
 20    return np.asarray(delta, dtype=float) * (1.0 - alpha)
 21
 22
 23def fleet_map(gaps, f):
 24    """Linearized ordered-gap map from the anonymous-throughput formula."""
 25    gaps = np.asarray(gaps, dtype=float)
 26    n = gaps.size + 1
 27    out = np.empty_like(gaps)
 28    for j in range(1, n):
 29        out[j - 1] = ((n - j) * f(j)) / (j * f(n - j)) * gaps[j - 1]
 30    return out
 31
 32
 33def verify_pair_map():
 34    rows = []
 35    for T, d in [(0.5, 0.01), (1.0, 0.1), (2.0, 1.9)]:
 36        deltas = np.linspace(1e-4, T / 2 - 1e-4, 100)
 37        nxt = anonymous_pair_map(deltas, T, d)
 38        slope, intercept = np.polyfit(deltas, nxt, 1)
 39        rows.append({"T": T, "d": d, "slope": float(slope),
 40                     "intercept": float(intercept),
 41                     "max_identity_error": float(np.max(abs(nxt - deltas)))})
 42    deltas = np.linspace(1e-4, .49, 100)
 43    priority_slope = np.polyfit(deltas, priority_pair_map(deltas), 1)[0]
 44    return {"sweep": rows, "priority_slope": float(priority_slope)}
 45
 46
 47def verify_fleet():
 48    rows = []
 49    functions = [("constant", lambda k: 1.0),
 50                 ("inverse_n", lambda k: 1.0 / k),
 51                 ("sqrt_n", lambda k: 1.0 / np.sqrt(k))]
 52    for n in range(2, 11):
 53        for name, fn in functions:
 54            base = np.linspace(.01, .01 * n, n - 1)
 55            lambdas = fleet_map(base, fn) / base
 56            rows.append({"N": n, "f": name, "det": float(np.prod(lambdas)),
 57                         "reciprocal_error": float(max(
 58                             abs(lambdas[j] * lambdas[-j - 1] - 1)
 59                             for j in range(n - 1))),
 60                         "max_lambda": float(max(lambdas))})
 61    return rows
 62
 63
 64def hitting_time(m, sigma, trials=3000, max_steps=200000, seed=0):
 65    """First passage of a gap random walk; each endpoint has noise sigma."""
 66    rng = np.random.default_rng(seed)
 67    x = np.full(trials, float(m))
 68    alive = np.ones(trials, dtype=bool)
 69    hit = np.full(trials, max_steps, dtype=np.int64)
 70    for step in range(1, max_steps + 1):
 71        idx = np.flatnonzero(alive)
 72        if not len(idx):
 73            break
 74        x[idx] += rng.normal(0.0, np.sqrt(2.0) * sigma, size=len(idx))
 75        crossed = x[idx] <= 0
 76        if np.any(crossed):
 77            hit[idx[crossed]] = step
 78            alive[idx[crossed]] = False
 79    return hit
 80
 81
 82def verify_noise_scaling():
 83    m = 0.20
 84    sigmas = np.array([.004, .006, .009, .013, .019])
 85    medians = np.array([np.median(hitting_time(m, s, seed=100 + i))
 86                        for i, s in enumerate(sigmas)])
 87    slope = np.polyfit(np.log(sigmas), np.log(medians), 1)[0]
 88    ratios = medians * sigmas ** 2 / m ** 2
 89    # A direct fourfold prediction test, interpolated from the nearest pair.
 90    ratio_006_012 = medians[1] / np.median(hitting_time(m, .012, seed=222))
 91    return {"m": m, "sigmas": sigmas.tolist(), "median_steps": medians.tolist(),
 92            "loglog_exponent": float(slope), "C_median": ratios.tolist(),
 93            "C_median_mean": float(np.mean(ratios)),
 94            "median_ratio_sigma_.006_to_.012": float(ratio_006_012)}
 95
 96
 97def summarize(result):
 98    pair = result["pair"]["sweep"]
 99    fleet = result["fleet"]
100    noise = result["noise"]
101    max_det_err = max(abs(x["det"] - 1) for x in fleet)
102    max_recip_err = max(x["reciprocal_error"] for x in fleet)
103    max_pair_err = max(x["max_identity_error"] for x in pair)
104    # Predictions are mechanism checks, not claims about production storage.
105    result["prediction_checks"] = {
106        "P1_pair_return_slope_predicted_1_observed": [x["slope"] for x in pair],
107        "P1_max_absolute_identity_error": max_pair_err,
108        "P2_determinant_predicted_1_max_abs_error": max_det_err,
109        "P2_reciprocal_eigenvalue_max_error": max_recip_err,
110        "P3_noise_exponent_predicted_-2_observed": noise["loglog_exponent"],
111        "P3_sigma_doubling_lifetime_ratio_predicted_4_observed":
112            noise["median_ratio_sigma_.006_to_.012"],
113        "all_mechanism_checks_pass": bool(
114            max_pair_err < 1e-12 and max_det_err < 1e-12 and
115            max_recip_err < 1e-12 and abs(noise["loglog_exponent"] + 2) < .15 and
116            2.5 < noise["median_ratio_sigma_.006_to_.012"] < 6.0)
117    }
118
119
120def main():
121    result = {"pair": verify_pair_map(), "fleet": verify_fleet(),
122              "noise": verify_noise_scaling()}
123    d = .30
124    anon, priority = d, d
125    for _ in range(20):
126        anon = float(anonymous_pair_map(anon))
127        priority = float(priority_pair_map(priority))
128    result["comparison_20_cycles"] = {"initial_gap": d,
129        "anonymous_gap": anon, "priority_gap": priority}
130    summarize(result)
131    Path("results.json").write_text(json.dumps(result, indent=2))
132    print(json.dumps(result, indent=2))
133
134
135if __name__ == "__main__":
136    main()