Add script to download skypile data

scripts/download_skypile.py

@@ -0,0 +1,163 @@
+"""
+Streaming 下载 SkyPile-150B 子采样，转成 minimind PretrainDataset 兼容格式（{"text": ...}）。
+
+特性：
+- Streaming：不全下，边读边写，磁盘只占目标量
+- Token 计数：精确控制下载量（按 budget 而非样本数停）
+- 断点续传：进程被杀也能从最后写到的行接着下
+- 进度日志：每 N 个样本打印进度
+- 简易过滤：太短 / 重复字符多的样本跳过
+
+Usage:
+    # 默认下 25B tokens（1B 模型 Chinchilla 最优）
+    python download_skypile.py
+
+    # 改下载量
+    python download_skypile.py --target_tokens 10_000_000_000
+
+    # 改输出路径
+    python download_skypile.py --output ../dataset/skypile_25b.jsonl
+
+需要 fwdproxy（HuggingFace 在 allowlist 但 OD 无直接 DNS）：
+    export https_proxy=http://fwdproxy:8080
+    export http_proxy=http://fwdproxy:8080
+"""
+import os
+import sys
+
+__package__ = "scripts"
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+import argparse
+import json
+import time
+import warnings
+from datasets import load_dataset
+from transformers import AutoTokenizer
+
+warnings.filterwarnings('ignore')
+
+
+def count_tokens_fast(text: str) -> int:
+    """快速估算 token 数：中文按字符 ≈ token，英文按词 × 1.3。
+    用于在线计数（精确 tokenize 太慢）。"""
+    n_chars = len(text)
+    # 中文 token ≈ 1 char；英文 1 token ≈ 4 char
+    # 简单估算：取个折中
+    return int(n_chars / 1.8)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Stream-download SkyPile-150B subset")
+    parser.add_argument('--repo', default='Skywork/SkyPile-150B', type=str)
+    parser.add_argument('--split', default='train', type=str)
+    parser.add_argument('--output', default='../dataset/skypile_25b.jsonl', type=str)
+    parser.add_argument('--target_tokens', type=int, default=25_000_000_000,
+                        help='目标 token 数（estimate），到达后停止')
+    parser.add_argument('--min_length', type=int, default=200,
+                        help='样本最小字符数，太短跳过')
+    parser.add_argument('--log_every', type=int, default=5000)
+    parser.add_argument('--tokenizer_path', type=str, default='../model',
+                        help='用 minimind tokenizer 精确计算 token 数（每 N 个样本核对一次）')
+    parser.add_argument('--exact_count_every', type=int, default=50000,
+                        help='每 N 个样本精确 tokenize 一次校准估算误差')
+    args = parser.parse_args()
+
+    output_path = os.path.abspath(args.output)
+    os.makedirs(os.path.dirname(output_path), exist_ok=True)
+
+    # 断点续传：count 已有文件的行数
+    n_written, est_tokens_written = 0, 0
+    if os.path.exists(output_path):
+        print(f'[resume] found existing {output_path}, counting...')
+        with open(output_path) as f:
+            for line in f:
+                n_written += 1
+                try:
+                    est_tokens_written += count_tokens_fast(json.loads(line)['text'])
+                except Exception:
+                    pass
+        print(f'[resume] {n_written:,} samples already written, ~{est_tokens_written/1e9:.2f}B tokens')
+        if est_tokens_written >= args.target_tokens:
+            print('[resume] target already reached, exit')
+            return
+
+    # 校准 tokenizer（可选；下载完后跑 eval_perplexity 时会用到）
+    tokenizer = None
+    if args.tokenizer_path:
+        try:
+            tokenizer = AutoTokenizer.from_pretrained(args.tokenizer_path)
+            print(f'[tokenizer] loaded for calibration: vocab={tokenizer.vocab_size}')
+        except Exception as e:
+            print(f'[tokenizer] failed to load ({e}), will skip calibration')
+
+    print(f'[start] streaming {args.repo} split={args.split} → {output_path}')
+    print(f'[target] {args.target_tokens/1e9:.1f}B tokens (estimated)')
+
+    ds = load_dataset(args.repo, split=args.split, streaming=True)
+
+    skipped_short = 0
+    n_seen = 0
+    t0 = time.time()
+    calibration_ratio = 1.0  # actual_tokens / estimated_tokens，每次校准更新
+
+    with open(output_path, 'a') as fout:
+        for sample in ds:
+            n_seen += 1
+            # 跳过已写过的（断点续传）
+            if n_seen <= n_written:
+                continue
+
+            text = sample.get('text', '')
+            if len(text) < args.min_length:
+                skipped_short += 1
+                continue
+
+            est_tok = count_tokens_fast(text)
+            actual_est = int(est_tok * calibration_ratio)
+
+            # 写入 minimind 格式
+            fout.write(json.dumps({'text': text}, ensure_ascii=False) + '\n')
+            n_written += 1
+            est_tokens_written += actual_est
+
+            # 进度日志
+            if n_written % args.log_every == 0:
+                elapsed = time.time() - t0
+                rate = (n_written / elapsed) if elapsed > 0 else 0
+                progress = est_tokens_written / args.target_tokens * 100
+                eta_h = (args.target_tokens - est_tokens_written) / max(est_tokens_written / elapsed, 1) / 3600
+                print(f'[{n_written:>9,} samples] {est_tokens_written/1e9:>5.2f}B tok '
+                      f'({progress:>5.1f}%) | {rate:.0f} samples/s | ETA {eta_h:.1f}h | '
+                      f'skipped_short={skipped_short:,}')
+                fout.flush()
+
+            # 周期性精确 tokenize 几条校准估算
+            if tokenizer and n_written % args.exact_count_every == 0:
+                actual = sum(len(tokenizer(t, add_special_tokens=False).input_ids)
+                             for t in [text])
+                estimate = est_tok
+                if estimate > 0:
+                    new_ratio = actual / estimate
+                    calibration_ratio = 0.9 * calibration_ratio + 0.1 * new_ratio
+                    print(f'[calibration] est_ratio updated → {calibration_ratio:.3f} '
+                          f'(this sample: est={estimate}, actual={actual})')
+
+            if est_tokens_written >= args.target_tokens:
+                print(f'\n[done] target reached: {est_tokens_written/1e9:.2f}B est tokens')
+                break
+
+    elapsed = time.time() - t0
+    print(f'\n{"=" * 60}')
+    print(f'Total: {n_written:,} samples, ~{est_tokens_written/1e9:.2f}B est tokens')
+    print(f'Elapsed: {elapsed/60:.1f} min ({elapsed/3600:.1f}h)')
+    print(f'Skipped (too short): {skipped_short:,}')
+    print(f'Output: {output_path} ({os.path.getsize(output_path)/1e9:.2f} GB)')
+    print(f'{"=" * 60}')
+    if tokenizer:
+        print(f'⚠️  估算 token 数有 ~{abs(1-calibration_ratio)*100:.1f}% 误差。')
+        print(f'   精确数：跑 eval_perplexity.py 时会有 actual count。')
+
+
+if __name__ == "__main__":
+    main()

trainer/train_full_sft.py

@@ -16,17 +16,23 @@ from torch.nn.parallel import DistributedDataParallel
 from torch.utils.data import DataLoader, DistributedSampler
 from model.model_minimind import MiniMindConfig
 from dataset.lm_dataset import SFTDataset
-from trainer.trainer_utils import get_lr, Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, init_model, SkipBatchSampler
+from trainer.trainer_utils import (
+    get_lr, Logger, is_main_process, lm_checkpoint, init_distributed_mode,
+    setup_seed, init_model, SkipBatchSampler,
+    detect_gpu_peak_tflops_bf16, compute_model_flops_per_token,
+)
 
 warnings.filterwarnings('ignore')
 
 
 def train_epoch(epoch, loader, iters, start_step=0, wandb=None):
     start_time = time.time()
+    last_log_time, last_log_step = start_time, start_step
     last_step = start_step
     for step, (input_ids, labels) in enumerate(loader, start=start_step + 1):
-        input_ids = input_ids.to(args.device)
-        labels = labels.to(args.device)
+        # non_blocking 仅在 --enable_optimizations=1 + pin_memory=True 时有效
+        input_ids = input_ids.to(args.device, non_blocking=args.enable_optimizations == 1)
+        labels = labels.to(args.device, non_blocking=args.enable_optimizations == 1)
         last_step = step
         lr = get_lr(epoch * iters + step, args.epochs * iters, args.learning_rate)
         for param_group in optimizer.param_groups:
@@ -49,14 +55,21 @@ def train_epoch(epoch, loader, iters, start_step=0, wandb=None):
             optimizer.zero_grad(set_to_none=True)
 
         if step % args.log_interval == 0 or step == iters:
-            spend_time = time.time() - start_time
+            now = time.time()
+            spend_time = now - start_time
+            window_dt = max(now - last_log_time, 1e-6)
+            window_steps = step - last_log_step
+            window_its = window_steps / window_dt
+            tokens_window = window_steps * args.batch_size * args.max_seq_len * world_size
+            mfu = (flops_per_token * tokens_window / window_dt) / (gpu_peak_tflops * 1e12) * 100
+            last_log_time, last_log_step = now, step
             current_loss = loss.item() * args.accumulation_steps
             current_aux_loss = res.aux_loss.item() if res.aux_loss is not None else 0.0
             current_logits_loss = current_loss - current_aux_loss
             current_lr = optimizer.param_groups[-1]['lr']
             eta_min = spend_time / max(step - start_step, 1) * (iters - step) // 60
-            Logger(f'Epoch:[{epoch + 1}/{args.epochs}]({step}/{iters}), loss: {current_loss:.4f}, logits_loss: {current_logits_loss:.4f}, aux_loss: {current_aux_loss:.4f}, lr: {current_lr:.8f}, epoch_time: {eta_min:.1f}min')
-            if wandb: wandb.log({"loss": current_loss, "logits_loss": current_logits_loss, "aux_loss": current_aux_loss, "learning_rate": current_lr, "epoch_time": eta_min})
+            Logger(f'[{time.strftime("%H:%M:%S")}] Epoch:[{epoch + 1}/{args.epochs}]({step}/{iters}), it/s: {window_its:.2f}, MFU: {mfu:.1f}%, loss: {current_loss:.4f}, logits_loss: {current_logits_loss:.4f}, aux_loss: {current_aux_loss:.4f}, lr: {current_lr:.8f}, epoch_time: {eta_min:.1f}min')
+            if wandb: wandb.log({"loss": current_loss, "logits_loss": current_logits_loss, "aux_loss": current_aux_loss, "learning_rate": current_lr, "epoch_time": eta_min, "it/s": window_its, "MFU": mfu})
 
         if (step % args.save_interval == 0 or step == iters) and is_main_process():
             model.eval()
@@ -104,7 +117,13 @@ if __name__ == "__main__":
     parser.add_argument('--from_resume', default=0, type=int, choices=[0, 1], help="是否自动检测&续训（0=否，1=是）")
     parser.add_argument("--use_wandb", action="store_true", help="是否使用wandb")
     parser.add_argument("--wandb_project", type=str, default="MiniMind-Full-SFT", help="wandb项目名")
-    parser.add_argument("--use_compile", default=0, type=int, choices=[0, 1], help="是否使用torch.compile加速（0=否，1=是）")
+    parser.add_argument("--use_compile", default=1, type=int, choices=[0, 1],
+                        help="是否使用 torch.compile 加速（0=否，1=是，A100 上推荐 1）")
+    parser.add_argument("--enable_optimizations", default=1, type=int, choices=[0, 1],
+                        help="开关：A/B 对比优化效果。1=启用 fused AdamW + non_blocking + persistent_workers（默认）；"
+                             "0=baseline（关掉这些，仅保留 log 测量）。compile 由 --use_compile 独立控制。")
+    parser.add_argument("--gpu_peak_tflops", type=float, default=0,
+                        help="GPU bf16 理论峰值 TFLOPS（用于 MFU 计算）；默认 0 表示按 GPU 名自动检测")
     args = parser.parse_args()
 
     # ========== 1. 初始化环境和随机种子 ==========
@@ -136,8 +155,20 @@ if __name__ == "__main__":
     train_ds = SFTDataset(args.data_path, tokenizer, max_length=args.max_seq_len)
     train_sampler = DistributedSampler(train_ds) if dist.is_initialized() else None
     scaler = torch.cuda.amp.GradScaler(enabled=(args.dtype == 'float16'))
-    optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate)
-    
+    # fused=True 把所有 param 的 Adam 更新融成一个 CUDA kernel
+    optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate,
+                            fused=(args.enable_optimizations == 1))
+
+    # ========== 5b. MFU 测量准备（始终启用，无 perf 影响）==========
+    flops_per_token = compute_model_flops_per_token(model, lm_config)
+    gpu_peak_tflops = detect_gpu_peak_tflops_bf16(override=args.gpu_peak_tflops)
+    world_size = dist.get_world_size() if dist.is_initialized() else 1
+    Logger(f'[MFU] params_flops/token={flops_per_token/1e9:.3f} GFLOPS, '
+           f'GPU peak={gpu_peak_tflops:.0f} TFLOPS bf16, world_size={world_size}')
+    Logger(f'[OPT] enable_optimizations={args.enable_optimizations} '
+           f'(non_blocking={args.enable_optimizations}, fused_adam={args.enable_optimizations}, '
+           f'persistent_workers={args.enable_optimizations}), use_compile={args.use_compile}')
+
     # ========== 6. 从ckp恢复状态 ==========
     start_epoch, start_step = 0, 0
     if ckp_data:
@@ -149,8 +180,10 @@ if __name__ == "__main__":
     
     # ========== 7. 编译和分布式包装 ==========
     if args.use_compile == 1:
-        model = torch.compile(model)
-        Logger('torch.compile enabled')
+        # enable_optimizations=1 时用 max-autotune（找更优 kernel），0 时用 default mode
+        compile_mode = "max-autotune" if args.enable_optimizations == 1 else "default"
+        model = torch.compile(model, mode=compile_mode)
+        Logger(f'torch.compile enabled (mode={compile_mode})')
     if dist.is_initialized():
         model = DistributedDataParallel(model, device_ids=[local_rank])
     
@@ -160,7 +193,9 @@ if __name__ == "__main__":
         setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
         skip = start_step if (epoch == start_epoch and start_step > 0) else 0
         batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip)
-        loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=True)
+        loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers,
+                            pin_memory=True,
+                            persistent_workers=(args.enable_optimizations == 1 and args.num_workers > 0))
         if skip > 0: 
             Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step，从step {start_step + 1}开始')
             train_epoch(epoch, loader, len(loader) + skip, start_step, wandb)