Add portable 135M BP sweep handoff

Match the active EP configuration and support four-GPU BP candidates from collaborators' data locations.

Co-authored-by: Cursor <cursoragent@cursor.com>

1 files changed, 38 insertions, 10 deletions

diff --git a/ep_run/casc_bp_train.py b/ep_run/casc_bp_train.py
index 0515ec1..39a69d2 100644
--- a/ep_run/casc_bp_train.py
+++ b/ep_run/casc_bp_train.py
@@ -1,8 +1,9 @@
 """BP-train a small cascade-form standard transformer (L distinct blocks), saving ckpts
 every --save_every for the A0.2 on-trajectory gradient gate (cascade_probe.py --ckpt).
 Plain LLM training — this is also the BP twin for the C-tier money runs."""
-import argparse, math, pickle, time, json
+import argparse, math, os, pickle, time, json
 import numpy as np, torch, torch.nn as nn, torch.nn.functional as F
+import torch.distributed as dist
 from pathlib import Path
 
 ap = argparse.ArgumentParser()
@@ -30,17 +31,28 @@ ap.add_argument('--zloss', type=float, default=0.0)      # z-loss coefficient; 0
 ap.add_argument('--qup_bits', type=int, default=0)        # STAGE-0 mirror: naked resident-cell writes
 ap.add_argument('--qcomp_bits', type=int, default=0)      # STAGE-0 mirror: compute on DAC grid, fp32 master
 ap.add_argument('--data', default='tinystories_bpe')     # dataset dir under ep_run/data
+ap.add_argument('--ddp_backend', default='nccl', choices=['nccl', 'gloo'])
 args = ap.parse_args()
 if args.olmo2 and args.tok_init <= 0: args.tok_init = 0.02
-torch.manual_seed(args.seed)
+
+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
+torch.manual_seed(args.seed)  # identical initialization on every rank
+DGEN = torch.Generator().manual_seed(args.seed * 7919 + RANK * 104729 + 11)
 dev = 'cuda' if torch.cuda.is_available() else 'cpu'
 
-DD = Path('/home/yurenh2/ept/ep_run/data') / args.data
+DATA_ROOT = Path(os.environ.get('EPT_DATA_ROOT', Path(__file__).resolve().parent / 'data'))
+DD = DATA_ROOT / 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)
@@ -148,7 +160,13 @@ if args.resume:
         with torch.no_grad(): W_out.copy_(_ck['wout'].to(dev))
     if _ck.get('lnf') is not None and not isinstance(ln_f, nn.Identity): ln_f.load_state_dict(_ck['lnf'])
     start_step = int(_ck.get('step', 0))
-    print(f'[resume] loaded {args.resume} at step {start_step}', flush=True)
+    if RANK == 0: print(f'[resume] loaded {args.resume} at step {start_step}', flush=True)
+if DDP:
+    with torch.no_grad():
+        for p in params:
+            dist.broadcast(p.data, 0)
+    if RANK == 0:
+        print(f'[ddp] world={WORLD} backend={args.ddp_backend}; effective batch {args.B}x{WORLD}={args.B * WORLD}', flush=True)
 if args.opt == 'muon':
     from muon import build_hybrid
     opt, sched = build_hybrid(blocks, params, args.lr, args.muon_lr, args.warmup,
@@ -187,7 +205,7 @@ def evaluate(nb=6):
 wb = None
 if args.wandb == 'auto':
     args.wandb = 'ept-fineweb-72m' if 'fineweb' in args.data else 'ept-tinystories-42m'
-if args.wandb:
+if args.wandb and RANK == 0:
     try:
         import wandb as _w
         wb = _w.init(entity='eqprop-llm-training', project=args.wandb, name=args.wandb_run or args.tag, id=args.wandb_run or args.tag,
@@ -196,7 +214,8 @@ if args.wandb:
         print(f'[wandb] disabled ({e})', flush=True)
 
 n = sum(p.numel() for p in params)
-print(f'[{args.tag}] cascade-BP L{args.L} C{args.C} H{args.H} T{args.T} | {n/1e6:.2f}M params | {dev}', flush=True)
+if RANK == 0:
+    print(f'[{args.tag}] cascade-BP L{args.L} C{args.C} H{args.H} T{args.T} | {n/1e6:.2f}M params | {dev}', flush=True)
 best, t0 = 1e9, time.time()
 outdir = Path('runs'); outdir.mkdir(exist_ok=True)
 for _ in range(start_step): sched.step()   # advance LR schedule to the resumed step
@@ -216,6 +235,12 @@ for step in range(start_step, args.steps + 1):
         if args.zloss > 0:
             loss = loss + args.zloss * (torch.logsumexp(logits.float(), -1) ** 2).mean()
     opt.zero_grad(set_to_none=True); loss.backward()
+    if DDP:
+        for p in params:
+            if p.grad is None:
+                p.grad = torch.zeros_like(p)
+            dist.all_reduce(p.grad, op=dist.ReduceOp.SUM)
+            p.grad.div_(WORLD)
     if args.qcomp_bits > 0:
         with torch.no_grad():
             for p, q in zip(params, QSAVE): p.copy_(q)
@@ -231,19 +256,22 @@ for step in range(start_step, args.steps + 1):
                 q = p / g_
                 fl = q.floor()
                 p.copy_((fl + (torch.rand_like(p) < (q - fl)).float()) * g_)
-    if step % args.log == 0:
+    if step % args.log == 0 and RANK == 0:
         val = evaluate(); best = min(best, val)
         print(f'step {step:5d}/{args.steps} | train {loss.item():.4f} val {val:.4f} (best {best:.4f}) '
               f'| {step/max(time.time()-t0,1e-9):.2f} it/s', flush=True)
         if wb is not None:
             try: wb.log({'train_ce': loss.item(), 'val_ce': val, 'best': best}, step=step)
             except Exception: pass
-    if step % args.save_every == 0 or step == args.steps:
+    if (step % args.save_every == 0 or step == args.steps) and RANK == 0:
         torch.save({'tok': tok.state_dict(), 'pos': pos.state_dict(), 'blocks': blocks.state_dict(),
                     'wout': (W_out.detach().cpu() if args.olmo2 else None),
                     'lnf': (ln_f.state_dict() if not isinstance(ln_f, nn.Identity) else None),
                     'step': step, 'val': best, 'config': vars(args)}, outdir / f'{args.tag}_s{step}.pt')
-print(f'[{args.tag}] DONE best val CE {best:.4f} (random ln({vocab})={math.log(vocab):.3f})', flush=True)
+if RANK == 0:
+    print(f'[{args.tag}] DONE best val CE {best:.4f} (random ln({vocab})={math.log(vocab):.3f})', flush=True)
 if wb is not None:
     try: wb.summary['best_val_ce'] = best; wb.finish()
     except Exception: pass
+if DDP:
+    dist.destroy_process_group()