分布式编译与训练——多设备编排的编译器支持 / Compiler Support for Distributed Training and Multi-Device Orchestration
📅 创建时间：2026-06-03 🏷️ 标签：#分布式 #FSDP #PipelineParallelism #TensorParallelism #CollectiveOps #AllReduce #SPMD #梯度同步 📚 前置知识：[[08-operator-fusion]]（算子融合） [[12-scheduling]]（调度） 📚 相关知识：[[../training-infra/02-distributed-training]]（../training-infra/02-distributed-training） [[22-xla-internals]]（XLA 内部）
┌──────────────────────────────────────────────────────────────────────────────┐
│ 📌 场景：FSDP 训练 GPU 利用率周期性下降 │
├──────────────────────────────────────────────────────────────────────────────┤
│ 你在用 PyTorch FSDP 训练一个 70B 模型： │
│ ```python │
│ model = FSDP(model, auto_wrap_policy=..., mixed_precision=BFloat16) │
│ optimizer = ShardedDDPOptimizer(optimizer) │
│ ``` │
│ │
│ 训练过程中，GPU 利用率周期性下降——每隔几个 step， │
│ 利用率就掉到 10%。你怀疑是梯度同步（AllReduce）导致的 │
│ 通信-计算 overlap 不好。但你不知道编译器能做什么来优化这个过程。 │
└──────────────────────────────────────────────────────────────────────────────┘第1节：分布式训练的三个维度
1.1 Data Parallel（数据并行）
核心思想：每个设备都有完整模型的副本，不同设备处理不同的数据 batch。
python
# 朴素数据并行示意
class DataParallel:
"""
最简单的数据并行实现
"""
def __init__(self, model, devices):
self.models = [model.to(d) for d in devices] # 每个设备一个模型副本
def train_step(self, batch):
# 分割数据
batches = batch.split(len(devices))
# 并行前向
outputs = []
for model, sub_batch, device in zip(self.models, batches, devices):
output = model(sub_batch.to(device))
outputs.append(output)
# 汇总 loss
losses = [out['loss'] for out in outputs]
total_loss = sum(losses) / len(losses)
# 并行反向
for model, sub_batch, output in zip(self.models, batches, outputs):
total_loss.backward() # 反向传播
# 问题：每个设备独立更新自己的模型副本
# 如果不同步，模型会逐渐不一致！
for model in self.models:
model.step()
return total_loss
# 真正的 DataParallel 需要梯度同步
# 所有设备计算完梯度后，需要对梯度做平均（AllReduce）梯度同步的 AllReduce 操作：
python
import torch.distributed as dist
# AllReduce：所有设备贡献自己的梯度，然后得到平均后的梯度
def distributed_all_reduce(tensor):
"""
AllReduce 操作：
- 所有设备调用这个函数
- 每个设备得到结果：tensor = sum(all_tensors) / num_devices
"""
# 创建一个与所有其他设备的连接
tensor = tensor.clone() # AllReduce 会修改 tensor
dist.all_reduce(tensor, op=dist.ReduceOp.SUM)
tensor.div_(dist.get_world_size()) # 求平均
return tensor
# 朴素实现的问题：
# 1. 同步等待：所有设备必须等到最慢的设备完成才能继续
# 2. 通信开销大：每次梯度同步都要等
# 3. GPU 利用率低：计算和通信不能 overlap1.2 Tensor Parallel（张量并行）
核心思想：把模型参数（权重矩阵）按维度切分到不同设备。
python
# Tensor Parallel 示意：Column Parallel Linear
class ColumnParallelLinear(torch.nn.Module):
"""
列切分：权重矩阵按输出维度切分
输入 X: [batch, seq_len, hidden_dim]
权重 W: [hidden_dim, out_dim / n_devices]
输出 Y: [batch, seq_len, out_dim / n_devices]
"""
def __init__(self, in_features, out_features, world_size):
super().__init__()
self.world_size = world_size
# 每个设备只持有部分权重
self.weight = torch.nn.Parameter(
torch.randn(in_features, out_features // world_size)
)
def forward(self, x):
# 所有设备获得完整的输入（broadcast）
x = x.clone() # 避免修改原数据
# 本地计算：每个设备只计算自己负责的输出列
local_output = x @ self.weight
# AllReduce：收集所有设备的输出，拼接成完整结果
outputs = [torch.zeros_like(local_output) for _ in range(self.world_size)]
dist.all_gather(outputs, local_output)
output = torch.cat(outputs, dim=-1)
return output
# Megatron-LM 的 Tensor Parallelism
# 核心是 3D/2D 并行中的张量切分策略Row Parallel Linear：
python
class RowParallelLinear(torch.nn.Module):
"""
行切分：权重矩阵按输入维度切分
输入 X: [batch, seq_len, hidden_dim / n_devices]
权重 W: [hidden_dim / n_devices, out_features]
输出 Y: [batch, seq_len, out_features]
"""
def __init__(self, in_features, out_features, world_size):
super().__init__()
self.world_size = world_size
# 每个设备持有部分权重（按输入维度切分）
self.weight = torch.nn.Parameter(
torch.randn(in_features // world_size, out_features)
)
def forward(self, x):
# x 已经按列切分了（由上一层的 Column Parallel 提供）
# 本地计算
local_output = x @ self.weight
# AllReduce：求和（因为输出需要完整）
dist.all_reduce(local_output)
return local_output1.3 Pipeline Parallel（流水线并行）
核心思想：把模型按层切分到不同设备。
python
# Pipeline Parallel 示意
class PipelineParallel(torch.nn.Module):
"""
流水线并行：把模型按层分配到不同设备
"""
def __init__(self, layers_per_device, num_devices):
super().__init__()
# 假设模型有 80 层，4 个设备，每个设备 20 层
self.devices = [f"cuda:{i}" for i in range(num_devices)]
# 按层分配
self.stages = []
for i in range(num_devices):
stage = torch.nn.Sequential(*layers_per_device[i])
self.stages.append(stage.to(self.devices[i]))
def forward(self, x):
# Micro-batch 处理
# Stage 0 (GPU 0) -> Stage 1 (GPU 1) -> ... -> Stage N (GPU N)
x = x.to(self.devices[0])
for i, stage in enumerate(self.stages):
x = stage(x)
if i < len(self.stages) - 1:
# 异步发送到下一个设备
x = x.to(self.devices[i + 1], non_blocking=True)
return x流水线调度的两种模式：
F-then-B (Flush at every bubble):
┌─────────┬─────────┬─────────┬─────────┐
│ Forward │ Forward │ Forward │ Forward │ <- GPU 0
│ F1 │ F2 │ F3 │ F4 │
├─────────┼─────────┼─────────┼─────────┤
│ - │ Forward │ Forward │ Forward │ <- GPU 1
│ - │ F1 │ F2 │ F3 │
├─────────┼─────────┼─────────┼─────────┤
│ - │ - │ Forward │ Forward │ <- GPU 2
│ - │ - │ F1 │ F2 │
├─────────┼─────────┼─────────┼─────────┤
│ - │ - │ - │ Forward │ <- GPU 3
│ - │ - │ - │ F1 │
└─────────┴─────────┴─────────┴─────────┘
时间 ─────────────────────────────────►
1F1B (One-Forward-One-Backward):
┌─────────┬─────────┬─────────┬─────────┐
│ Forward │ Backwd │ Forward │ Backwd │ <- GPU 0
│ F1 │ B1 │ F2 │ B2 │
├─────────┼─────────┼─────────┼─────────┤
│ - │ Forward │ Backwd │ Forward │ <- GPU 1
│ - │ F1 │ B1 │ F2 │
├─────────┼─────────┼─────────┼─────────┤
│ - │ - │ Forward │ Backwd │ <- GPU 2
│ - │ - │ F1 │ B1 │
├─────────┼─────────┼─────────┼─────────┤
│ - │ - │ - │ Forward │ <- GPU 3
│ - │ - │ - │ F1 │
└─────────┴─────────┴─────────┴─────────┘
1F1B 的 bubble 更小，但需要更大的显存来容纳更多 micro-batches1.4 三种并行策略对比
| 维度 | Data Parallel | Tensor Parallel | Pipeline Parallel |
|---|---|---|---|
| 切分方式 | 数据切分 | 参数切分 | 层切分 |
| 模型副本 | 每个设备完整 | 每个设备部分 | 每个设备部分 |
| 通信模式 | AllReduce (梯度) | AllReduce/AllGather (激活) | P2P (激活) |
| 通信量 | O(梯度大小) | O(激活大小 × 设备数) | O(激活大小) |
| 显存需求 | 高（全模型） | 中（部分参数） | 低（单阶段） |
| 扩展性 | 线性（数据） | 有限（通信瓶颈） | 线性（层数） |
| 编程复杂度 | 低 | 高 | 中 |
| 最适合 | 数据密集场景 | 超大单层（如 Large Embedding） | 超深模型 |
第2节：FSDP 的编译支持
2.1 FSDP 原理
FSDP (Fully Sharded Data Parallel)：将模型参数分片到所有设备，每个设备只持有 1/N 的参数。
python
import torch
import torch.distributed as dist
from torch.distributed.fsdp import FullyShardedDataParallel as FSDP
from torch.distributed.fsdp import ShardingStrategy, MixedPrecision
from torch.distributed.fsdp.wrap import transformer_auto_wrap_policy
# ============================================================
# FSDP 基础配置
# ============================================================
# 混合精度配置
mixed_precision_policy = MixedPrecision(
param_dtype=torch.bfloat16, # 参数精度
reduce_dtype=torch.float32, # AllReduce 用高精度（避免精度损失）
buffer_dtype=torch.bfloat16, # Buffer 精度
)
# 自动 wrap 策略：每个 transformer 层作为一个 FSDP 单元
def auto_wrap_policy(module, recurse, non_wrappable):
wrapped = set()
if recurse:
return True
if is_trainable_layer(module):
wrapped.add(module)
return wrapped
# 包装模型
model = FSDP(
model,
sharding_strategy=ShardingStrategy.FULL_SHARD, # 分片 + 梯度 + Optimizer state
# 或者: ShardingStrategy.SHARD_GRAD_OP (只分片梯度和优化器状态)
# 或者: ShardingStrategy.NO_SHARD (只分片参数)
auto_wrap_policy=transformer_auto_wrap_policy,
mixed_precision=mixed_precision_policy,
device_id=torch.cuda.current_device(),
)
# 优化器也需要分片
# 使用 ShardedDDPOptimizer 或 FSDP 包装
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4)
optimizer = FSDP.optim_state_dict_to_load(
optimizer, model, optimizer.state_dict()
)2.2 FSDP 的通信-计算 Overlap
python
# ============================================================
# FSDP 编译优化：通信计算Overlap
# ============================================================
# 关键参数
fsdp_config = {
"sharding_strategy": ShardingStrategy.FULL_SHARD,
# Forward 时的分片策略
"forward_prefetch": True, # 提前获取下一个 shard
"backward_prefetch": BackwardPrefetch.PRE_GATHER, # 提前聚合梯度
# 算子融合
"use_orig_params": True, # 使用原始参数（更好的融合机会）
# CPU offload（内存不够时）
"cpu_offload": None, # or CPUOffload(offload_params=True)
}
model = FSDP(model, **fsdp_config)
# ============================================================
# 梯度检查点 + FSDP
# ============================================================
# 梯度检查点：在反向传播时重新计算前向（节省显存）
# FSDP 下需要特殊处理
from torch.distributed.algorithms._checkpoint.checkpoint_wrapper import (
checkpoint_wrapper,
CheckpointImpl,
apply_activation_checkpointing,
)
# 方法 1：包装整个模型
model = checkpoint_wrapper(
model,
checkpoint_impl=CheckpointImpl.NO_REENTRANT,
)
# 方法 2：为特定层应用检查点
apply_activation_checkpointing(
model,
checkpoint_wrapper_fn=checkpoint_wrapper,
auto_wrap_policy=..., # 与 FSDP 的 wrap policy 一致
)
# ============================================================
# 梯度累积 + FSDP
# ============================================================
# FSDP 下做梯度累积需要注意
# 只有在梯度同步完成后才能更新参数
model = FSDP(model)
for batch in data_loader:
# 累积足够的 micro-batches
micro_batch_count = 0
while micro_batch_count < grad_accum_steps:
# 分片输入
sharded_input = batch[micro_batch_count]
# 前向（不同步梯度）
output = model(sharded_input)
loss = loss_fn(output, target) / grad_accum_steps
loss.backward() # FSDP 会自动聚合梯度
micro_batch_count += 1
# 优化器步骤（此时所有 micro-batches 的梯度已同步）
optimizer.step()
optimizer.zero_grad()2.3 FSDP 编译配置详解
python
# torch.compile 与 FSDP 的结合
import torch
# FSDP 需要特殊处理才能与 torch.compile 一起用
# PyTorch 2.1+ 支持
# 1. 编译 FSDP 模型
compiled_model = torch.compile(model, mode="reduce-overhead")
# 2. 或者分别编译 FSDP 包装内的模型
# （需要在 FSDP 之前编译）
# ============================================================
# FSDP 性能调优参数
# ============================================================
fsdp_params = {
# 通信优化
"limit_all_gathers": True, # 限制同时进行的 gather 数量
"overlap_with_ backward": True, # 与反向传播 overlap
"use_border_checkpointing": True, # 边界检查点（减少通信）
# 内存优化
"move_grads_to_cpu": False, # 梯度放 CPU（省显存但慢）
"move_params_to_cpu": False, # 参数放 CPU
# 调试
"use_full_prec_state_dict": True, # 保存/加载用全精度
}
# ============================================================
# 常见 FSDP 问题排查
# ============================================================
# 问题 1：显存不够
# 解决：减小 batch size、使用梯度检查点、增加设备
# 问题 2：GPU 利用率低
# 解决：检查通信瓶颈、增加 forward_prefetch
# 问题 3：梯度同步慢
# 解决：检查网络带宽、使用更高效的 AllReduce 实现（NCCL）第3节：Tensor Parallelism 编译
3.1 AllReduce 在计算图中的位置
python
# Tensor Parallel 下 AllReduce 的位置
class TensorParallelLinear(torch.nn.Module):
"""
Tensor Parallel Linear 层编译决策
"""
def __init__(self, tp_size):
self.tp_size = tp_size
# ... 初始化代码 ...
def forward(self, x):
# 计算图分析：
#
# Column Parallel:
# x -> [local_matmul] -> [local_output] -> [AllReduce] -> output
# (独立) (通信)
#
# Row Parallel:
# x -> [AllReduce] -> [local_matmul] -> output
# (通信) (独立)
# 编译器的优化机会：
# 1. 如果 AllReduce 前后有其他算子，可以融合
# 2. 如果多个连续的 AllReduce，可以合并
# 3. 通信可以与计算 overlap
return output3.2 Column/Row Parallel 的编译决策
python
# 编译优化：融合连续的小 AllReduce
class CollectiveFusion:
"""
Collective 融合：将多个小的通信操作合并为一个大的
"""
def should_fuse_collectives(self, graph):
"""
判断是否应该融合 collective 操作
"""
collectives = self.find_collectives(graph)
for i in range(len(collectives) - 1):
c1, c2 = collectives[i], collectives[i + 1]
# 如果两个 collective 之间没有依赖，可以融合
if self.no_dependency_between(c1, c2):
# 但也要考虑内存开销
if self.memory_budget_ok(c1, c2):
return True
return False
def fuse_collectives(self, c1, c2):
"""
融合两个 collective
"""
# 方案 1：直接合并数据
# 先发 c1，再发 c2
# 方案 2：打包成更大的消息
# 一次性发送合并后的数据
# 方案 3：延迟合并
# 等待更多 collective，然后一起发送第4节：Pipeline Parallelism 编译
4.1 微批次调度
python
# Pipeline Parallel 调度器
class PipelineScheduler:
"""
编译器生成的流水线调度
"""
def schedule_1f1b(self, num_microbatches, num_stages):
"""
1F1B 调度：每个设备交替执行前向和反向
"""
schedule = []
# 预热阶段：只有前向
for m in range(num_stages):
schedule.append(("forward", m))
# 稳定阶段：交替前向和反向
for m in range(num_microbatches - num_stages):
schedule.append(("forward", m + num_stages))
schedule.append(("backward", m))
# 冷却阶段：只有反向
for m in range(num_stages):
schedule.append(("backward", m + num_microbatches - num_stages))
return schedule
def compute_pipeline_bubble(self, num_stages, num_microbatches):
"""
计算流水线空闲时间（bubble）
"""
# 1F1B 的 bubble 时间
bubble_time = (num_stages - 1) * 2
# F-then-B 的 bubble 时间
bubble_time_ftb = (num_stages - 1)
efficiency = num_microbatches / (num_microbatches + bubble_time)
return efficiency
def auto_select_schedule(self, model, num_microbatches):
"""
编译器自动选择最优调度
"""
# 考虑因素：
# 1. 显存限制（1F1B 需要更多显存）
# 2. 通信延迟（更深的流水线延迟更敏感）
# 3. 计算不对称性（不同 stage 计算量不同）
# 简单启发式：
if self.get_available_memory() < self.estimate_1f1b_memory():
return "F-then-B"
else:
return "1F1B"4.2 编译器流水线优化
python
# 流水线并行编译优化
class PipelineOptimizer:
"""
编译器层面的流水线优化
"""
def balance_stages(self, model, num_stages):
"""
平衡各阶段的计算量
"""
stage_boundaries = []
# 计算每层的 FLOPs
layer_flops = []
for layer in model.layers:
flops = self.estimate_flops(layer)
layer_flops.append(flops)
# 动态规划找最优切分点
target_flops = sum(layer_flops) / num_stages
current_flops = 0
for i, flops in enumerate(layer_flops):
current_flops += flops
if current_flops >= target_flops:
stage_boundaries.append(i)
current_flops = 0
return stage_boundaries
def optimize_schedule_for_heterogeneous(self, stages):
"""
针对异构设备的调度优化
"""
# 不同设备可能有不同的计算能力
# 调度应该让慢设备承担更少的计算
device_capabilities = self.get_device_capabilities()
# 重新分配 stage
rebalanced_stages = self.rebalance_stages(
stages, device_capabilities
)
return rebalanced_stages第5节：Collective Operations 编译优化
5.1 AllReduce 算法
python
# AllReduce 的不同实现算法
# ============================================================
# Ring AllReduce
# ============================================================
# 时间复杂度: O(N)，适合大消息
#
# 4 个 GPU 的 Ring AllReduce:
# Step 1: 每个 GPU 发一块给下一个 GPU（接收一块）
# Step 2: 继续传递，直到所有 GPU 都有完整数据
#
# 通信量: 2 * (N-1) / N * message_size
def ring_allreduce(rank, world_size, tensor):
"""
Ring AllReduce 实现
"""
# 分为两个阶段：
# 1. Scatter Reduce：每个 GPU 获得最终结果的一部分
# 2. AllGather：每个 GPU 获得完整结果
block_size = tensor.numel() // world_size
# Scatter Reduce 阶段
for i in range(world_size - 1):
send_idx = (rank - i - 1) % world_size
recv_idx = (rank - i) % world_size
send_block = tensor[send_idx * block_size:(send_idx + 1) * block_size]
recv_block = tensor[recv_idx * block_size:(recv_idx + 1) * block_size]
# 异步发送和接收
send_req = dist.isend(send_block, dst=(rank + 1) % world_size)
recv_req = dist.irecv(recv_block, src=(rank - 1) % world_size)
recv_req.wait()
recv_block += send_block # 累加
send_req.wait()
# AllGather 阶段
for i in range(world_size - 1):
send_idx = (rank - i + 1) % world_size
recv_idx = (rank - i) % world_size
send_block = tensor[send_idx * block_size:(send_idx + 1) * block_size]
recv_block = tensor[recv_idx * block_size:(recv_idx + 1) * block_size]
# 发送和接收
send_req = dist.isend(send_block, dst=(rank + 1) % world_size)
recv_req = dist.irecv(recv_block, src=(rank - 1) % world_size)
recv_req.wait()
send_req.wait()
# ============================================================
# Tree AllReduce
# ============================================================
# 时间复杂度: O(log N)，适合小消息
# 使用二叉树结构，根节点收集和分发数据5.2 NCCL 通信原语
python
import torch.distributed as dist
# ============================================================
# 常用 NCCL Collective 操作
# ============================================================
# AllReduce：所有设备求和并广播
dist.all_reduce(tensor, op=dist.ReduceOp.SUM)
# AllGather：收集所有设备的数据
output = [torch.zeros_like(tensor) for _ in range(world_size)]
dist.all_gather(output, tensor)
# ReduceScatter：求和后分发到各设备
output = torch.zeros_like(tensor)
dist.reduce_scatter(output, input_list)
# Broadcast：广播数据到所有设备
dist.broadcast(tensor, src=0)
# Gather：收集到一个设备
if rank == 0:
output = [torch.zeros_like(tensor) for _ in range(world_size)]
else:
output = None
dist.gather(tensor, output, dst=0)
# ============================================================
# 异步 Collective 操作
# ============================================================
# 异步发送
req = dist.isend(tensor, dst=dest_rank)
# ... 做其他计算 ...
req.wait()
# 异步接收
req = dist.irecv(tensor, src=src_rank)
# ... 做其他计算 ...
req.wait()
# 使用 barrier 同步
dist.barrier()5.3 Collective 融合
python
# Collective 融合：将多个小操作合并为一个大操作
class CollectiveFusionOptimizer:
"""
编译器优化：识别可以融合的 Collective 操作
"""
def find_fusable_collectives(self, graph):
"""
查找图中可以融合的 collective
"""
fusable = []
for node in graph.nodes:
if node.is_collective():
# 检查是否可以与下一个 collective 融合
next_node = node.next
if self.can_fuse(node, next_node):
# 检查融合后的内存开销是否可接受
fused_size = self.estimate_fused_size(node, next_node)
if fused_size < memory_budget:
fusable.append((node, next_node))
return fusable
def fuse(self, c1, c2):
"""
融合两个 collective
"""
# 方案 1：打包到单个消息
# 先合并数据，再发送
# 方案 2：流水线化
# 第一个 collective 完成后立即开始第二个
# 方案 3：延迟执行
# 等待更多 collective 到达后再一起发送第6节：分布式训练中的图优化
6.1 梯度累积下的图简化
python
# 梯度累积时的编译优化
class GradientAccumulationOptimizer:
"""
梯度累积下的图简化
"""
def optimize_for_gradient_accumulation(self, model, accum_steps):
"""
优化梯度累积的计算图
"""
# 1. 识别可以跨 step 共享的计算
# 例如：LayerNorm 的统计量（训练时不能用，但 eval 可以）
# 2. 优化梯度同步时机
# 不需要每 step 都同步，可以累积多个 step
# 3. 融合 optimizer 更新
# 将多个小梯度更新合并
pass
def reduce_gradient_sync_frequency(self, model, sync_every_n_steps):
"""
减少梯度同步频率
"""
# Local SGD：多步后同步一次
# 可以减少通信开销
local_steps = 0
for step in training_loop:
output = model(input)
loss.backward()
local_steps += 1
if local_steps >= sync_every_n_steps:
# AllReduce 梯度
dist.all_reduce(grad)
optimizer.step()
optimizer.zero_grad()
local_steps = 06.2 重计算策略
python
# 重计算（Recomputation）优化
class RecomputationOptimizer:
"""
编译器决定哪些算子需要重计算
"""
def select_recomputation_candidates(self, model, memory_budget):
"""
选择需要重计算的算子
"""
candidates = []
for node in model.graph.nodes:
# 显存节省 = 输出的显存 - 重计算的开销
memory_saved = node.output_size
# 重计算成本 = 前向 + 反向的成本
recompute_cost = self.estimate_forward_cost(node) + \
self.estimate_backward_cost(node)
# 只重计算那些显存节省大于成本的算子
if memory_saved > recompute_cost:
candidates.append((node, memory_saved - recompute_cost))
# 按收益排序，选择在显存预算内的
candidates.sort(key=lambda x: -x[1])
selected = []
total_saved = 0
for node, benefit in candidates:
if total_saved + node.output_size <= memory_budget:
selected.append(node)
total_saved += node.output_size
return selected升华
┌──────────────────────────────────────────────────────────────────────────────┐
│ 分布式编译实践原则 │
├──────────────────────────────────────────────────────────────────────────────┤
│ 1. 通信即计算：把通信当作计算图的一部分，编译器可以做全局优化 │
│ 2. 重叠是关键：让通信和计算尽可能 overlap，提高 GPU 利用率 │
│ 3. 分片要平衡：各设备的计算量和通信量要尽量均衡 │
│ 4. 内存换通信：重计算可以节省显存，但会增加计算量 │
│ 5. 调度决定性能：好的调度可以减少 bubble，提高流水线效率 │
└──────────────────────────────────────────────────────────────────────────────┘"AI 可查 vs 必须理解"清单
必须理解（不理解就等于不会）：
- 🔴 Data Parallel / Tensor Parallel / Pipeline Parallel 的核心思想
- 🔴 AllReduce 的作用和 Ring/Tree 算法的 trade-off
- 🔴 FSDP 的分片策略和通信模式
- 🔴 流水线并行的 bubble 问题和 1F1B vs F-then-B 调度
- 🔴 通信-计算 overlap 的原理
AI 可查（知道去哪查就行）：
- ✅ FSDP API 的具体参数和用法
- ✅ NCCL 的具体 collective API
- ✅ Megatron-LM 的 tensor parallel 实现细节
- ✅ 特定硬件的网络带宽和延迟
- ✅ 各框架的分布式训练 API
学习状态：🟡 开始学习