Rank、Process Group 与集合通信

先定义参与者与范围

rank 是一个 process group 内的整数参与者编号，范围 $0\ldots p-1$；$p$ 是 world size。一个进程可以属于多个 groups，并在不同 group 中拥有不同 local rank。rank 不必等同物理 GPU id、主机号或全局进程号；生产系统必须明确“哪个 group 的 rank”。

Collective 要求所有参与 rank 以相同顺序进入兼容操作。若 rank 0 调 All-Reduce、rank 1 同时调 All-Gather，通常会死锁或报错。异步 API 返回 work handle 只表示已提交；第一次消费结果前仍要建立正确同步。

三个核心语义的 shape 与所有权

设 $p$ 个 rank，各自有 local tensor $X_r\in\mathbb{R}^{n}$：

All-Gather：分片 → 每 rank 完整拼接

输出在每个 rank 上 replicated。它不是求和；拼接维与各 rank 顺序必须明确。

Reduce-Scatter：每 rank 大输入 → 归约后的不同分片

每个 rank 输入可看作 $p$ 个 chunks：$X_r=[X_r^{(0)};\ldots;X_r^{(p-1)}]$。输出：

每个 rank 只拥有不同的 reduced shard。它同时 reduce 与 scatter，不是“先在一个 root 求和再手工切片”的 API 语义承诺。

All-Reduce：每 rank 同 shape 输入 → 每 rank 相同归约结果

输出 shape 不变且 replicated。现有 All-Reduce 页面进一步解释 Ring 的一种实现；逻辑 collective 也可以用 Tree 或层次化算法完成。

本机两进程 Gloo 实验

实验问题：在没有多张 GPU 的机器上，能否用 CPU/Gloo 精确验证三种 collective 的 shape、数值和 rank 所有权？

import os,platform,socket
import torch
import torch.distributed as dist
import torch.multiprocessing as mp

def free_port():
    with socket.socket() as sock:
        sock.bind(("127.0.0.1",0)); return sock.getsockname()[1]

def worker(rank,world_size,port):
    os.environ["MASTER_ADDR"]="127.0.0.1"; os.environ["MASTER_PORT"]=str(port)
    dist.init_process_group("gloo",rank=rank,world_size=world_size)
    try:
        local=torch.tensor([rank*2.+1.,rank*2.+2.])
        gathered=[torch.empty_like(local) for _ in range(world_size)]
        dist.all_gather(gathered,local)
        gathered_tensor=torch.cat(gathered)
        torch.testing.assert_close(gathered_tensor,torch.tensor([1.,2.,3.,4.]))

        # concatenated input 沿 dim 0 切成 world_size 份，再逐位置求和。
        rs_input=torch.tensor([1.,10.,2.,20.])*(rank+1)
        rs_output=torch.empty(2)
        dist.reduce_scatter_single(rs_output,rs_input,op=dist.ReduceOp.SUM)
        expected=torch.tensor([3.,30.]) if rank==0 else torch.tensor([6.,60.])
        torch.testing.assert_close(rs_output,expected)

        all_reduced=local.clone()
        dist.all_reduce(all_reduced,op=dist.ReduceOp.SUM)
        torch.testing.assert_close(all_reduced,torch.tensor([4.,6.]))
        print(f"[rank {rank}] local={local.tolist()} all_gather={gathered_tensor.tolist()} "
              f"reduce_scatter={rs_output.tolist()} all_reduce={all_reduced.tolist()}",flush=True)
    finally:
        dist.destroy_process_group()

def main():
    print(f"python={platform.python_version()} torch={torch.__version__} backend=gloo world_size=2")
    mp.spawn(worker,args=(2,free_port()),nprocs=2,join=True)

if __name__=="__main__": main()

完整文件位于 examples/collectives_gloo.py，并用 python -W error 实际执行。rank 0/1 的 All-Gather 都得到 [1,2,3,4]；Reduce-Scatter 分别得到 [3,30] 与 [6,60]；All-Reduce 都得到 [4,6]。

Gloo/CPU 验证的是语义，不是 NCCL/CUDA 性能。NCCL 的带宽、拓扑、stream ordering、异步错误和多机网络无法由这台单 GPU 机器验证；不得把本地进程时间写成多卡性能结论。

成本模型与算法边界

通信成本常用 $\alpha$-$\beta$ 模型：固定每轮延迟 $\alpha$ 加每字节成本 $\beta$。但轮数和字节数取决于 Ring、Tree、hierarchical、NVLink/NVSwitch/PCIe/网络拓扑及消息大小。应用声明 All-Reduce 语义，不一定声明 Ring。

分析 collective 时至少标明：

• payload bytes 与 dtype；
• 每 rank 输入/输出 shape；
• 结果是 replicated、sharded 还是 partial；
• 同步点与是否可与独立计算重叠；
• group 跨不跨节点、高速互联域；
• 实际 backend/算法证据，而不是只看 API 名称。

GPU 互联拓扑把这些 logical bytes 映射到 PCIe/NVLink/NIC paths，并用 byte-hop toy model说明相同 group size 的 rank placement也可能产生不同跨域负载。

与张量并行的关系

• 张量并行的列并行产生 output-feature shards；只有下游需要完整输出时才 All-Gather。该页用同一组随机权重证明 shard+concat 等于 dense。
• 行并行产生同 shape partial sums，需要 All-Reduce，或在 sequence-parallel layout 下用 Reduce-Scatter 把 reduced 结果保持为分片。
• All-Reduce页面用 Ring 展开具体数据移动；本页作为统一语义总览，防止把 Ring 当成 All-Reduce 的定义。
• MoE Router 与 Expert Parallel需要 All-to-All 把 ragged token routes 从 source rank 送到 expert owner，再把结果送回；它不是拼接或求和语义。
• 并行模式总览进一步说明 Sequence/Context/Pipeline Parallel 在哪些布局边界需要 gather、softmax partial merge 或点对点 stage transfer。

参考资料

• PyTorch Distributed communication package
• NVIDIA NCCL — Collective Operations