diff options
Diffstat (limited to 'ep_run')
| -rw-r--r-- | ep_run/casc_eq_train.py | 127 |
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 |