4. 分布式工作流程
本章把前面章节的概念串成一条线：从用户提交一个分布式训练 job，到它在集群中实际运行、通信、容错、保存 checkpoint 的完整链路。
4.1 一条分布式 AI 训练 job 的全链路
4.2 阶段 1：Job 提交与资源调度
- 用户通过
kubectl apply -f training-job.yaml或torchrun --nproc_per_node=8 train.py提交 job； - 调度器根据 GPU、显存、网络拓扑、优先级决定把 job 放在哪些节点（Kubernetes 生态由 Volcano / Kueue / KubeRay 提供 gang 调度与排队）；
- 调度结果写入控制面（K8s apiserver / Ray GCS / Slurm controller）；
- 每个 worker 被拉起，获得唯一的 rank、world size、master address。
这一步本质上是把“一个逻辑任务”映射到“多个物理进程”。
4.3 阶段 2：成员发现与初始化
分布式训练需要所有 worker 知道彼此地址，以便建立通信组：
- PyTorch DDP/FSDP：由 rank 0 作为 coordinator，其他 worker 通过
MASTER_ADDR/MASTER_PORT连接； - Ray Train：由 Ray GCS 维护 actor 列表，每个 worker 通过 GCS 发现彼此；
- Kubernetes + MPI Operator：由 MPI operator 生成 hostfile；
- Horovod：由 driver 维护 worker 列表。
成员发现必须满足：
- 一致性：所有 worker 对“谁是成员”达成一致；
- 活性：新 worker 加入、旧 worker 退出能被及时感知；
- 容错性：coordinator 故障时不能导致整个 job 崩溃。
4.4 阶段 3：建立通信组
成员发现完成后，worker 之间需要建立集合通信组：
- NCCL / Gloo / MPI：提供 all-reduce、all-gather、broadcast、reduce-scatter 等原语；
- ring / tree / hierarchical all-reduce：根据网络拓扑选择最优梯度聚合算法；
- barrier：在关键节点同步，确保所有 worker 进度一致。
4.5 阶段 4：Forward / Backward / All-Reduce 循环
一次训练 step 的分布式执行：
- 每个 worker 读取自己的 mini-batch（数据并行）；
- 各自执行 forward 计算 loss；
- 各自执行 backward 计算梯度；
- 通过 all-reduce 把梯度同步到所有 worker；
- 每个 worker 用相同的梯度更新本地模型副本。
因为 all-reduce 后所有 worker 拥有相同梯度，所以模型副本保持一致。
4.6 阶段 5：Barrier 与同步
在以下场景需要 barrier：
- 每个 epoch 开始时重新分片数据；
- checkpoint 保存前确保所有 worker 处于同一步；
- 动态调整 batch size 或学习率时。
barrier 的实现依赖底层通信库或分布式协调服务：
torch.distributed.barrier()；- Ray 的
tune.report()+ 同步调度器； - Kubernetes 的 Init Container 或 MPI 的
MPI_Barrier()。
4.7 阶段 6：Checkpoint 协同保存
分布式 checkpoint 比单机复杂得多：
- 每个 worker 保存自己负责的 shard（模型并行）或副本（数据并行）；
- 需要保存“元数据”记录每个 shard 属于哪个 rank；
- 写入过程必须原子：要么全部成功，要么整体不可见。
现代实现普遍用 manifest + 原子重命名（rename） 代替传统两阶段提交：先把所有 shard 写完，最后原子更新一个 manifest 文件指向新版本；恢复时读取最新 manifest 即可。这避免了 2PC 的协调者单点与锁开销。
2024 年后的主流做法已经演进成“三级 checkpoint 体系”，用空间换恢复时间：
| 层级 | 介质 | 用途 |
|---|---|---|
| L1 内存检查点 | 冗余 GPU / CPU 显存 | 秒级恢复（Google Gemini，SOSP'24） |
| L2 本地 NVMe | 节点本地 SSD | 节点级故障恢复 |
| L3 对象存储 | 跨区域持久存储 | 机架 / 集群级灾难恢复 |
配合 PyTorch Distributed Checkpoint（DCP）的异步保存，训练主线程不阻塞，后台线程把状态逐级搬到 CPU DRAM、本地盘和远端存储。
常见做法对比：
- 同步 checkpoint：所有 worker 同时保存，简单但会造成停顿；
- 异步 checkpoint：后台线程保存，训练继续，配合 manifest 版本号保证恢复点正确；
- 增量 checkpoint：只保存变化的部分，减少 IO。
4.8 阶段 7：故障检测与回滚
训练过程中可能出现：
- 某个 worker OOM 或 NCCL 超时；
- 某个节点掉卡；
- 网络分区导致部分 worker 失联。
处理流程：
- 检测：心跳、watchdog、NCCL timeout；
- 定位：通过日志、监控、分布式追踪确定故障范围；
- 决策：重启 job、弹性伸缩、或从 checkpoint 恢复；
- 恢复：重新成员发现、重新建立通信组、加载 checkpoint。
现代训练框架越来越强调弹性训练：
- PyTorch Elastic Training：允许 worker 动态加入/退出；
- Kubernetes + Volcano / Kueue：gang 调度保证要么全部拉起要么不启动，失败后快速补位；
- Ray Train：自动容错与 lineage 重建。
在万卡级集群中，故障不再是“异常”而是“常态”：Meta 在 Llama 3 405B 训练（16,384 张 H100，单个 run 54 天）中共发生 466 次任务中断，约 78% 与硬件有关（The Llama 3 Herd of Models，2024）。设计目标随之从“尽量不出错”变成“出错后分钟级恢复（MTTR）”：
- 快速检测：NCCL watchdog 超时上报 + 硬件遥测（GPU Xid 错误、ECC、温度）预测性隔离；
- 局部回滚：只重放受影响的流水线 stage，而不是整个任务（如 Oobleck 的流水线模板，SOSP'23）；
- Straggler 缓解：慢节点检测后动态摘除，避免 all-reduce 被单个慢节点拖住；
- 抢占处理：共享集群中 spot/抢占式实例回收时的状态保护与热迁移。
另一条路线是异步 / WAN 训练：DeepMind 的 DiLoCo 让每个 worker 用本地优化器独立跑数百步、周期性同步伪梯度，把通信频率降低约百倍；Prime Intellect 的 OpenDiLoCo 用 DHT（libp2p，无中心节点）做成员发现，在横跨两大洲、多家云、32 张 H100、worker 间带宽不足 1 Gbps 的环境下训练 1.1B 模型，保持了 90-95% 的算力利用率。它放弃了“所有梯度每步强一致”，换取跨地域、异构、易失资源的可用性——这是 CAP 权衡在训练通信上的直接体现，详见生产实践章。
4.9 推理服务的分布式工作流
推理服务的工作流与训练不同：
- 模型加载到多个 GPU / 多个副本；
- 请求通过负载均衡器路由到某个副本；
- 副本内部可能使用张量并行或流水线并行；
- 输出返回给用户，同时更新 KV Cache（如果共享）。
关键分布式问题：
- 多副本之间是否需要共享 KV Cache？
- 如何保持模型版本一致（A/B 测试、金丝雀）？
- 某个副本故障时如何快速摘除而不影响整体可用性？
4.10 一句话总结
一次分布式 AI job 的完整流程，本质上是“提交 → 发现 → 通信 → 计算 → 同步 → 持久化 → 容错”的循环；每个阶段都依赖分布式系统的共识、复制、分区与超时机制。