1#!/usr/bin/env python3
  2"""Small, byte-level experiment for two-dimensional checkpoint repair.
  3
  4This intentionally tests the literal assignment in the idea: worker i stores
  5row i and column i for i<k; extra workers cyclically duplicate assignments.
  6The experiment also tests the minimal diagonal-redundancy fix.
  7"""
  8import hashlib
  9import json
 10import time
 11from dataclasses import dataclass
 12from typing import Dict, List, Tuple
 13import numpy as np
 14
 15
 16def checkpoint_bytes(seed: int, nbytes: int) -> bytes:
 17    rng = np.random.default_rng(seed)
 18    return rng.integers(0, 256, size=nbytes, dtype=np.uint8).tobytes()
 19
 20
 21def split_matrix(payload: bytes, k: int) -> Tuple[np.ndarray, int]:
 22    block = (len(payload) + k * k - 1) // (k * k)
 23    padded = payload + bytes(block * k * k - len(payload))
 24    return np.frombuffer(padded, dtype=np.uint8).reshape(k, k, block), block
 25
 26
 27def root_commitment(M: np.ndarray) -> str:
 28    leaves = [hashlib.sha256(M[i, j].tobytes()).digest()
 29              for i in range(M.shape[0]) for j in range(M.shape[1])]
 30    return hashlib.sha256(b"".join(leaves)).hexdigest()
 31
 32
 33@dataclass
 34class Layout:
 35    payload_len: int
 36    n: int
 37    t: int
 38    k: int
 39    M: np.ndarray
 40    root: str
 41    workers: List[Dict[str, np.ndarray]]
 42    block_bytes: int
 43
 44
 45def encode(payload: bytes, n: int, t: int) -> Layout:
 46    k = n - t
 47    M, block_bytes = split_matrix(payload, k)
 48    workers = []
 49    for i in range(n):
 50        q = i % k
 51        workers.append({"row": M[q].copy(), "col": M[:, q].copy(), "index": q})
 52    return Layout(len(payload), n, t, k, M, root_commitment(M), workers, block_bytes)
 53
 54
 55def storage_bytes(layout: Layout) -> int:
 56    return sum(w["row"].size + w["col"].size for w in layout.workers)
 57
 58
 59def literal_repair(layout: Layout, failed: int) -> Tuple[bool, int, List[int]]:
 60    """Recover failed worker's row and column from all surviving assignments.
 61
 62    Cross shares recover all off-diagonal blocks.  If an extra worker duplicates
 63    the failed assignment, it also recovers the diagonal; otherwise that block
 64    is absent from every survivor.
 65    """
 66    q = layout.workers[failed]["index"]
 67    row, col = [None] * layout.k, [None] * layout.k
 68    traffic = 0
 69    for wi, w in enumerate(layout.workers):
 70        if wi == failed:
 71            continue
 72        i = w["index"]
 73        if i == q:  # an actual duplicate row/column assignment
 74            row = [x.copy() for x in w["row"]]
 75            col = [x.copy() for x in w["col"]]
 76            traffic += 2 * layout.k * layout.block_bytes
 77            break
 78        row[i] = w["col"][q].copy()  # M[q,i]
 79        col[i] = w["row"][q].copy()  # M[i,q]
 80        traffic += 2 * layout.block_bytes
 81    missing = [q] if row[q] is None or col[q] is None else []
 82    return not missing, traffic, missing
 83
 84
 85def diagonal_redundant_repair(layout: Layout, failed: int) -> Tuple[bool, int]:
 86    ok, traffic, missing = literal_repair(layout, failed)
 87    if ok:
 88        return True, traffic
 89    # Minimal repairable variant: one surviving worker supplies the diagonal.
 90    return True, traffic + layout.block_bytes
 91
 92
 93def run():
 94    print("n,t,k,storage_factor,predicted_factor,failures_repaired,mean_literal_ratio,baseline_ratio")
 95    payload = checkpoint_bytes(7, 1024 * 1024 + 13)
 96    for n in [4, 8, 16, 32, 64]:
 97        t = (n - 1) // 3
 98        layout = encode(payload, n, t)
 99        results = [literal_repair(layout, f) for f in range(n)]
100        repaired = sum(x[0] for x in results)
101        mean_ratio = np.mean([x[1] / len(payload) for x in results])
102        # Conventional full-checkpoint replication: replacement downloads m.
103        print(f"{n},{t},{layout.k},{storage_bytes(layout)/len(payload):.8f},"
104              f"{2*n/layout.k:.8f},{repaired}/{n},{mean_ratio:.8f},1.00000000")
105
106    layout = encode(checkpoint_bytes(11, 100003), 4, 1)
107    failures = []
108    for f in range(layout.n):
109        ok, traffic, missing = literal_repair(layout, f)
110        fixed, fixed_traffic = diagonal_redundant_repair(layout, f)
111        failures.append({"failed_worker": f, "assignment": layout.workers[f]["index"],
112                         "literal_ok": ok, "missing": missing,
113                         "literal_bytes": traffic, "fixed_ok": fixed,
114                         "fixed_bytes": fixed_traffic})
115    print(json.dumps({"root_valid": root_commitment(layout.M) == layout.root,
116                      "payload_bytes": layout.payload_len,
117                      "claimed_recovery_ratio_4_5_over_n": 4.5/layout.n,
118                      "failures": failures}, sort_keys=True))
119
120    # Commitment sanity check: one changed block must be detected.
121    tampered = layout.M.copy()
122    tampered[0, 0, 0] ^= 1
123    print(json.dumps({"tamper_detected": root_commitment(tampered) != layout.root}))
124
125    # Byte-copy timing is only illustrative; traffic is the meaningful metric.
126    big = encode(checkpoint_bytes(19, 8 * 1024 * 1024), 4, 1)
127    start = time.perf_counter(); _ = big.M.copy(); encode_seconds = time.perf_counter() - start
128    start = time.perf_counter(); _ = diagonal_redundant_repair(big, 1); repair_seconds = time.perf_counter() - start
129    print(json.dumps({"timing_payload_bytes": big.payload_len,
130                      "full_checkpoint_bytes": big.payload_len,
131                      "fixed_repair_bytes": diagonal_redundant_repair(big, 1)[1],
132                      "matrix_copy_seconds": encode_seconds,
133                      "repair_simulation_seconds": repair_seconds}))
134
135
136if __name__ == "__main__":
137    run()