scripts/_scratch/_batch_sweep.py

"""batch size 扫描：把我们这台 iGPU 的真实吞吐曲线量出来。

为什么要单独量
--------------
原版配方是 `batch=1`，我们现在也是 `batch=1`。但那是原版拿 2×A100 40GB 跑出来的选择 ——
我们这边实测峰值显存只用了 **29.23 GiB / 99.7 GiB**，有 3.4 倍余量。

关键问题不是"能不能开大"，是**开大到底换不换来吞吐**：
  * agent A 已经测过 `grad_accum=8` 耗时是 `accum=2` 的 8 倍 —— **梯度累积不省算力**，
    它只是把 N 次前反向攒成 1 个 optimizer step；
  * 而**真正的 batch**（一次前向里塞 B 条）能把每步的固定开销（kernel launch / 调度）
    摊薄，这才是可能提速的地方。

这个脚本不改任何训练代码，直接调 `Stage1Aligner.build_batch` 量：

    B ∈ {1,2,4,8,16,32}   s/step · s/样本 · 峰值显存 · 有效 token 吞吐

**注意 pad 的代价**：B>1 时 RNA-FM 和 tokenizer 都要 pad 到 batch 内最长，
所以"样本数/秒"的收益会小于 B 的倍数 —— 这正是要量出来的东西。

跑法（**必须独占 GPU**）：
    python scripts/_batch_sweep.py --data data/processed/rnaqa_rnacentral_all.jsonl \
        --batches 1,2,4,8,16 --steps 4 --out reports/batch_sweep.json
"""

from __future__ import annotations

import argparse
import json
import statistics
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))

import rnaqwen  # noqa: E402,F401

import torch  # noqa: E402

from rnaqwen.models.alignment import Stage1Aligner, Stage1Config  # noqa: E402
from rnaqwen.paths import default_encoder, describe_encoder  # noqa: E402


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--data", type=Path, default=ROOT / "data/processed/rnaqa_rnacentral_all.jsonl")
    ap.add_argument("--llm", default=str(ROOT / "models/Qwen3-8B"))
    ap.add_argument("--encoder", default=str(ROOT / default_encoder()),
                    help="不填就走 E 轨默认（ERNIE-RNA）")
    ap.add_argument("--batches", default="1,2,4,8,16")
    ap.add_argument("--steps", type=int, default=4, help="每个 batch 测几步（取中位数）")
    ap.add_argument("--nt-band", default=None,
                    help="限定序列长度区间 lo,hi（**控住 pad 这个混杂变量**；不填则自动取中位数 ±15%%）")
    ap.add_argument("--warmup-steps", type=int, default=2, help="先跑几步不算")
    ap.add_argument("--grad-checkpointing", action="store_true", default=True)
    ap.add_argument("--no-grad-checkpointing", dest="grad_checkpointing", action="store_false")
    ap.add_argument("--out", type=Path, default=ROOT / "reports/batch_sweep.json")
    args = ap.parse_args()

    max_b = max(int(x) for x in args.batches.split(","))
    rows = [json.loads(l) for l in args.data.read_text(encoding="utf-8").splitlines() if l.strip()]
    all_samples = [{"sequence": r["sequence"], "target": r.get("annotation") or ""}
                   for r in rows if (r.get("annotation") or "").strip()]

    # ⚠️ **长度分带是必须的**。第一版扫描没做这件事，量出来的结论是错的：
    #     B=1 抽到一条 94 nt 的 miRNA、B=2 抽到 1022 nt 的 lncRNA，
    #     于是"batch=2 比 batch=1 慢 4 倍" —— 那是 **pad 的代价，不是 batch 的代价**。
    # 语料是双峰的（22 nt 的成熟 miRNA + 1022 nt 的 lncRNA），不控长度就没法比。
    if args.nt_band:
        lo, hi = (int(x) for x in args.nt_band.split(","))
        samples = [s for s in all_samples if lo <= len(s["sequence"]) <= hi]
    else:
        lens = sorted(len(s["sequence"]) for s in all_samples)
        med = lens[len(lens) // 2]
        lo, hi = int(med * 0.85), int(med * 1.15)
        samples = [s for s in all_samples if lo <= len(s["sequence"]) <= hi]
    if len(samples) < max_b * 4:
        # 带太窄就放宽
        lo2, hi2 = int(lo * 0.6), int(hi * 1.4)
        samples = [s for s in all_samples if lo2 <= len(s["sequence"]) <= hi2]
    print(f"[sweep] 长度分带 [{lo}, {hi}] → {len(samples)} 条可用样本（控住 pad 这个混杂变量）")
    print(f"[sweep] 全部样本 {len(all_samples)} 条；batch ∈ {args.batches}；每档 {args.steps} 步")

    device = torch.device("cuda")
    cfg = Stage1Config(llm_path=args.llm, encoder_path=args.encoder,
                       grad_checkpointing=args.grad_checkpointing)
    t0 = time.time()
    model = Stage1Aligner(cfg)
    # ⚠️ 必须**逐个子模块**搬，不能 `Stage1Aligner(cfg).to(device)`。
    # `nn.Module.to()` 走的是内部 `_apply()`，不调用子模块自己的 `.to()` 覆写；
    # 而且实测**父模块 `.to()` 这条路径会触发一次 C++ 层 abort**
    # （无 Python traceback，栈落在 SDPA 的 math fallback 上）。
    # `train_stage1` 用的就是逐个子模块的写法 —— 这里必须跟它一致，否则量出来的不是训练的真实路径。
    model.encoder.to(device)
    model.llm.to(device)
    model.projection.to(device)
    model.projection.train()
    print(f"[sweep] 模型就绪 {time.time()-t0:.1f}s；{model.projection.describe()}")
    print(f"[sweep] device 校验: encoder={next(model.encoder.parameters()).device} "
          f"llm={next(model.llm.parameters()).device} proj={next(model.projection.parameters()).device}")

    opt = torch.optim.AdamW(model.projection.parameters(), lr=1e-6, weight_decay=0.05)

    report: dict = {"batches": {}, "data": str(args.data), "n_samples": len(samples),
                    "device": torch.cuda.get_device_name(0),
                    "mem_total_GiB": round(torch.cuda.get_device_properties(0).total_memory / 1024**3, 2),
                    "grad_checkpointing": args.grad_checkpointing}
    print(f"\n{'batch':>6s} {'s/step(中位)':>13s} {'s/样本':>9s} {'样本/s':>9s} "
          f"{'相对加速':>9s} {'峰值显存 GiB':>12s} {'批内最大 RNA token':>16s}")
    print("-" * 92)

    base_per_sample = None
    cursor = 0
    for B in [int(x) for x in args.batches.split(",")]:
        # 取一段连续样本；循环使用
        batch_samples = [samples[(cursor + i) % len(samples)] for i in range(B)]
        cursor += B
        torch.cuda.empty_cache()
        torch.cuda.reset_peak_memory_stats()

        # warmup
        for _ in range(args.warmup_steps):
            loss = model.forward_loss([s["sequence"] for s in batch_samples],
                                      [s["target"] for s in batch_samples], device)
            loss.backward()
            opt.step()
            opt.zero_grad(set_to_none=True)

        torch.cuda.synchronize()
        times = []
        for _ in range(args.steps):
            t1 = time.time()
            loss = model.forward_loss([s["sequence"] for s in batch_samples],
                                      [s["target"] for s in batch_samples], device)
            loss.backward()
            opt.step()
            opt.zero_grad(set_to_none=True)
            torch.cuda.synchronize()
            times.append(time.time() - t1)

        med = statistics.median(times)
        per_sample = med / B
        peak = torch.cuda.max_memory_allocated() / 1024**3
        nt = max(len(s["sequence"]) for s in batch_samples)
        if base_per_sample is None:
            base_per_sample = per_sample
        speedup = base_per_sample / per_sample

        report["batches"][str(B)] = {
            "sec_per_step_median": round(med, 4),
            "sec_per_step_mean": round(statistics.mean(times), 4),
            "sec_per_sample": round(per_sample, 4),
            "samples_per_sec": round(1.0 / per_sample, 3),
            "speedup_vs_batch1": round(speedup, 3),
            "peak_mem_GiB": round(peak, 2),
            "max_rna_nt_in_batch": nt,
            "steps": args.steps,
        }
        print(f"{B:6d} {med:13.3f} {per_sample:9.3f} {1.0/per_sample:9.2f} "
              f"{speedup:9.2f}x {peak:12.2f} {nt:16d}")

    print()
    print("怎么读：")
    print("  `相对加速` 就是 batch 换来的真实吞吐收益（已经扣掉 pad 的代价）")
    print("  `峰值显存` 离 99.7 GiB 还有多少余量，决定还能不能往上开")
    print("  ⚠️ 开大 batch = 训练步数变少 ⟹ **LR 与 warmup 必须跟着调**，否则等于换了个优化问题")

    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8")
    print(f"\n[sweep] 报告 -> {args.out}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())