Manual DDP for cascade-EP: guard-synced (drift/kretry via global max, gn on reduced grads), sigma broadcast, per-rank data RNG, None-preserving flat all-reduce, bitwise param sync-check, grad-equivalence test PASSED (cos 0.999999999 vs single-GPU big batch)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014FAPDWQ49M5Ye3NpTndTpn

1 files changed, 114 insertions, 13 deletions

diff --git a/ep_run/casc_eq_train.py b/ep_run/casc_eq_train.py
index 3f34daa..05273df 100644
--- a/ep_run/casc_eq_train.py
+++ b/ep_run/casc_eq_train.py
@@ -47,6 +47,9 @@ ap.add_argument('--dtop_every', type=int, default=1)    # 1 = exact (DEFAULT, BP
 ap.add_argument('--gate_every', type=int, default=200)  # in-training cos(EP,BP) telemetry; <=0 = fully BP-free (no bp_gate at all)
 ap.add_argument('--gate_govern', action='store_true')   # let gate cos adjust K/bscale (default: observe-only => training control is BP-free)
 ap.add_argument('--data', default='tinystories_bpe')     # dataset dir under ep_run/data (train.bin/val.bin/meta.pkl)
+ap.add_argument('--sync_check', type=int, default=500)   # DDP: verify bitwise param sync every N steps (0=off)
+ap.add_argument('--ddp_backend', default='nccl', choices=['nccl', 'gloo'])  # gloo = correctness tests on shared GPUs
+ap.add_argument('--ddp_grad_test', action='store_true')  # one-step grad equivalence test vs single-GPU big batch, then exit
 args = ap.parse_args()
 if args.olmo2:
     args.untie = True
@@ -54,12 +57,56 @@ if args.olmo2:
 torch.manual_seed(args.seed)
 dev = 'cuda' if torch.cuda.is_available() else 'cpu'
 
+# ---- DDP (manual: autograd.grad path, guard-synced; torchrun --standalone --nproc_per_node=N) ----
+import os
+import torch.distributed as dist
+DDP = int(os.environ.get('WORLD_SIZE', '1')) > 1
+if DDP:
+    dist.init_process_group(args.ddp_backend)
+    RANK, WORLD = dist.get_rank(), dist.get_world_size()
+    torch.cuda.set_device(int(os.environ['LOCAL_RANK']) % max(torch.cuda.device_count(), 1))
+else:
+    RANK, WORLD = 0, 1
+DGEN = torch.Generator().manual_seed(args.seed * 7919 + RANK * 104729 + 11)  # per-rank DATA stream ONLY
+                                                                             # (init/bsign RNGs stay rank-identical)
+def ddp_avg(gs, params):
+    """average a grad list across ranks; preserves the None pattern (identical graphs => identical
+    pattern) so optimizer skip-semantics match single-GPU exactly."""
+    if not DDP: return gs
+    none_mask = [g is None for g in gs]
+    filled = [g if g is not None else torch.zeros_like(p) for p, g in zip(params, gs)]
+    flat = torch.cat([g.reshape(-1) for g in filled])
+    if args.ddp_backend == 'gloo':
+        cf = flat.cpu(); dist.all_reduce(cf, op=dist.ReduceOp.SUM); flat = cf.to(flat.device)
+    else:
+        dist.all_reduce(flat, op=dist.ReduceOp.SUM)
+    flat /= WORLD
+    out, o = [], 0
+    for p in params:
+        n = p.numel(); out.append(flat[o:o + n].view_as(p)); o += n
+    return [None if m else g for m, g in zip(none_mask, out)]
+
+def ddp_max_scalar(v):
+    """global max of a python float (guard decisions must be identical on every rank)."""
+    if not DDP: return v
+    t = torch.tensor([v if math.isfinite(v) else float('inf')], device=dev if dev == 'cuda' else 'cpu')
+    if args.ddp_backend == 'gloo': t = t.cpu()
+    dist.all_reduce(t, op=dist.ReduceOp.MAX)
+    return float(t[0])
+
+def ddp_bcast_scalar(v):
+    if not DDP: return v
+    t = torch.tensor([v], device=dev if dev == 'cuda' else 'cpu')
+    if args.ddp_backend == 'gloo': t = t.cpu()
+    dist.broadcast(t, 0)
+    return float(t[0])
+
 DD = Path('/home/yurenh2/ept/ep_run/data') / args.data
 vocab = pickle.load(open(DD / 'meta.pkl', 'rb'))['vocab_size']
 
 def get_batch(split):
     data = np.memmap(DD / ('train.bin' if split == 'train' else 'val.bin'), dtype=np.uint16, mode='r')
-    ix = torch.randint(len(data) - args.T - 1, (args.B,))
+    ix = torch.randint(len(data) - args.T - 1, (args.B,), generator=DGEN)
     x = torch.stack([torch.from_numpy(data[i:i + args.T].astype(np.int64)) for i in ix])
     y = torch.stack([torch.from_numpy(data[i + 1:i + 1 + args.T].astype(np.int64)) for i in ix])
     return x.to(dev), y.to(dev)
@@ -293,7 +340,7 @@ def ep_step(x, y):
     global SIG0
     if GOV['K'] is None: GOV['K'] = args.K
     if GOV.get('step', 0) % args.sig_every == 0 or GOV.get('sig', 0) == 0:
-        GOV['sig'] = tok_sigma()
+        GOV['sig'] = ddp_bcast_scalar(tok_sigma())   # all ranks run it (keeps global-RNG lockstep); rank0's value wins
     GOV['step'] = GOV.get('step', 0) + 1
     sig = GOV['sig']
     if SIG0 is None: SIG0 = args.sig0 if args.sig0 > 0 else sig
@@ -310,9 +357,10 @@ def ep_step(x, y):
     with torch.no_grad():
         drift = sum(float((a - b).norm()) for a, b in zip(zp, zs_free)) / max(
             sum(float(b.norm()) for b in zs_free), 1e-9)
-    if (not math.isfinite(drift)) or (drift > 0.5 and not args.noguard):
+    gdrift = ddp_max_scalar(drift)   # guard DECISIONS on the global worst -> identical on every rank
+    if (not math.isfinite(gdrift)) or (gdrift > 0.5 and not args.noguard):
         ok_retry = False
-        if args.kretry > 0 and math.isfinite(drift) and not args.noguard:
+        if args.kretry > 0 and math.isfinite(gdrift) and not args.noguard:
             GOV['skr'] = GOV.get('skr', 0) + 1   # marginal batch: retry once with deeper relaxation
             z0, zs, ins, outs = free_states_graphed(x)
             zs_free = [z.clone() for z in zs]
@@ -320,16 +368,18 @@ def ep_step(x, y):
             with torch.no_grad():
                 drift = sum(float((a - b).norm()) for a, b in zip(zp, zs_free)) / max(
                     sum(float(b.norm()) for b in zs_free), 1e-9)
-            ok_retry = math.isfinite(drift) and drift <= 0.5
+            gdrift = ddp_max_scalar(drift)
+            ok_retry = math.isfinite(gdrift) and gdrift <= 0.5
         if not ok_retry:
             GOV['skd'] = GOV.get('skd', 0) + 1   # drift-guard reject (relaxation non-convergence)
             for p in all_params: p.grad = None
             return free_ce, beta_t, GOV['K'], False
-    GOV['drift'] = drift
+    GOV['drift'] = gdrift
     E = 0.0
     for z, o in zip(zp, last_outs): E = E + 0.5 * ((z.detach().float() - o.float()) ** 2).sum()   # fp32 accumulation (bf16-safe; no-op in fp32)
     obj = E / (NBT * beta_t) + obj_loss(readout(zp[-1].detach()).reshape(-1, vocab), y.reshape(-1))
     gs = torch.autograd.grad(obj, all_params, allow_unused=True)
+    gs = ddp_avg(gs, all_params)   # global-batch gradient; gn/gema/guard below see identical values on all ranks
     gn = 0.0
     for g in gs:
         if g is not None: gn += float((g ** 2).sum())
@@ -350,7 +400,7 @@ def bp_gate(x, y):
     z = emb(x)
     for b in blocks: z = b(z, mask)
     ce = obj_loss(readout(z).reshape(-1, vocab), y.reshape(-1))
-    return list(torch.autograd.grad(ce, all_params, allow_unused=True))
+    return ddp_avg(list(torch.autograd.grad(ce, all_params, allow_unused=True)), all_params)
 
 @torch.no_grad()
 def evaluate(nb=6):
@@ -362,8 +412,17 @@ def evaluate(nb=6):
         tot += F.cross_entropy(readout(z).reshape(-1, vocab), y.reshape(-1)).item()
     return tot / nb
 
+if DDP:   # belt & suspenders on top of identical init seeds: rank0's params are law
+    with torch.no_grad():
+        for p in all_params:
+            if args.ddp_backend == 'gloo':
+                t = p.data.cpu(); dist.broadcast(t, 0); p.data.copy_(t)
+            else:
+                dist.broadcast(p.data, 0)
+    if RANK == 0: print(f'[ddp] world={WORLD} backend={args.ddp_backend} params broadcast; eff batch {args.B}x{WORLD}={args.B*WORLD}', flush=True)
+
 wb = None
-if args.wandb:
+if args.wandb and RANK == 0:
     try:
         import wandb as _w
         wb = _w.init(project=args.wandb, name=args.wandb_run or args.tag, id=args.wandb_run or args.tag,
@@ -372,8 +431,39 @@ if args.wandb:
         print(f'[wandb] disabled ({e})', flush=True)
 
 n = sum(p.numel() for p in all_params)
-print(f'[{args.tag}] cascade-EP(EQUILIBRIUM/fb) L{args.L} C{args.C} T{args.T} beta={args.beta} '
-      f'K={args.K} geta={args.geta} | {n/1e6:.2f}M | {dev}', flush=True)
+if RANK == 0:
+    print(f'[{args.tag}] cascade-EP(EQUILIBRIUM/fb) L{args.L} C{args.C} T{args.T} beta={args.beta} '
+          f'K={args.K} geta={args.geta} | {n/1e6:.2f}M | {dev}', flush=True)
+
+if args.ddp_grad_test:
+    # one-step equivalence: DDP(WORLD ranks x B) averaged grad must equal single-GPU grad on the
+    # SAME WORLD*B batch (exact algebra: per-sample-independent relaxation + mean-linear readout).
+    # Protocol: run WORLD=1 with --B (W*B) first, then torchrun WORLD=N with --B B; both seed 4242.
+    _g = torch.Generator().manual_seed(4242)
+    _data = np.memmap(DD / 'train.bin', dtype=np.uint16, mode='r')
+    _full = torch.randint(len(_data) - args.T - 1, (WORLD * args.B,), generator=_g)
+    _ix = _full[RANK * args.B:(RANK + 1) * args.B]
+    _x = torch.stack([torch.from_numpy(_data[i:i + args.T].astype(np.int64)) for i in _ix]).to(dev)
+    _y = torch.stack([torch.from_numpy(_data[i + 1:i + 1 + args.T].astype(np.int64)) for i in _ix]).to(dev)
+    _ce, _bt, _r, _ok = ep_step(_x, _y)
+    assert _ok, 'grad test: ep_step guarded'
+    _flat = torch.cat([(p.grad if p.grad is not None else torch.zeros_like(p)).reshape(-1).double().cpu()
+                       for p in all_params])
+    if RANK == 0:
+        _f = Path('runs') / f'ddp_grad_w{WORLD}.pt'
+        torch.save({'flat': _flat, 'beta': _bt, 'W': WORLD, 'B': args.B}, _f)
+        print(f'[gradtest] W={WORLD} B/rank={args.B} beta_t={_bt:.3e} ce={_ce:.4f} saved {_f}', flush=True)
+        _ref = Path('runs') / 'ddp_grad_w1.pt'
+        if WORLD > 1 and _ref.exists():
+            _r1 = torch.load(_ref, weights_only=False)
+            assert _r1['B'] == WORLD * args.B, f"ref B={_r1['B']} != {WORLD*args.B}"
+            _rf = _r1['flat']
+            _cos = float((_flat @ _rf) / (_flat.norm() * _rf.norm()))
+            _rel = float((_flat - _rf).norm() / _rf.norm())
+            print(f'[gradtest] VERDICT cos={_cos:.9f} relerr={_rel:.2e} (DDP avg vs single-GPU big-batch)', flush=True)
+    import sys
+    sys.exit(0)
+
 best, t0 = 1e9, time.time()
 skips = 0
 for _ in range(start_step): sched.step()   # advance LR schedule to the resumed step
@@ -396,7 +486,16 @@ for step in range(start_step, args.steps + 1):
                 GOV['K'] -= 1; GOV['bscale'] = min(GOV['bscale'] * 1.05, 1.0)
     torch.nn.utils.clip_grad_norm_(all_params, 1.0)
     opt.step(); sched.step(); opt.zero_grad(set_to_none=True)
-    if step % args.log == 0:
+    if DDP and args.sync_check > 0 and step % args.sync_check == 0 and step > 0:
+        with torch.no_grad():
+            h = torch.stack([torch.stack((p.double().sum(), (p.double() ** 2).sum())) for p in all_params]).sum(0)
+        hc = h.cpu() if args.ddp_backend == 'gloo' else h
+        hs = [torch.zeros_like(hc) for _ in range(WORLD)]
+        dist.all_gather(hs, hc)
+        if any(bool((x != hs[0]).any()) for x in hs[1:]):
+            print(f'[ddp] PARAM DESYNC step {step} rank {RANK}: {[x.tolist() for x in hs]}', flush=True)
+            raise RuntimeError('DDP param desync — aborting rather than training garbage')
+    if step % args.log == 0 and RANK == 0:
         val = evaluate(); best = min(best, val)
         gtag = '' if math.isnan(gcos) else f' cos={gcos:.4f}'
         print(f'step {step:5d}/{args.steps} | train {ce:.4f} val {val:.4f} (best {best:.4f}) '
@@ -406,12 +505,14 @@ for step in range(start_step, args.steps + 1):
             try: wb.log({'train_ce': ce, 'val_ce': val, 'best': best, 'beta_t': beta_t,
                          'rounds': rounds, 'skips': skips, 'gate_cos': (None if math.isnan(gcos) else gcos)}, step=step)
             except Exception: pass
-    if step % args.save_every == 0 and step > 0:
+    if step % args.save_every == 0 and step > 0 and RANK == 0:
         torch.save({'tok': tok.state_dict(), 'pos': pos.state_dict(), 'blocks': blocks.state_dict(),
                     'wout': (W_out.detach().cpu() if args.untie else None),
                     'lnf': (ln_f.state_dict() if not isinstance(ln_f, nn.Identity) else None),
                     'step': step, 'val': best, 'config': vars(args)}, Path('runs') / f'{args.tag}_s{step}.pt')
-print(f'[{args.tag}] DONE best val CE {best:.4f} (BP twin 2.9746; zil-diagnostic 3.3236)', flush=True)
+if RANK == 0:
+    print(f'[{args.tag}] DONE best val CE {best:.4f}', flush=True)
+if DDP: dist.destroy_process_group()
 if wb is not None:
     try: wb.summary['best_val_ce'] = best; wb.finish()
     except Exception: pass