PS架构

有一个参数服务器(CPU/GPU均可), 负责广播数据, 梯度聚合

环同步算法 Ring All Reduce

GPU间构成一个单向的环结构

1. Scatter-reduce
   遍历[GPU设备数量 - 1]轮, 每轮将GPU的某项数据给下一GPU, 同时接受上一GPU传来的数据, 然后下一轮发送上一轮接受的数据
2. All Gather
   广播接受到的数据, 同样广播[GPU设备数量 - 1]轮

通信实现方式

• 机器内通信

  • 共享内存
  • PCIe
  • NVLink

• 机器间通信

  • TCP/IP网络
  • RDMA网络

通信硬件

• PCIe
  走PCIe插槽进行通信
• NVLink
  高速, 8通道, 差分, 可用于GPU与GPU间通信, 也可用于支持NVLink的CPU与GPU通信
• RDMA Remote Direct Memory Access
  无需cpu干预, 用户态传输数据, 每个应用程序都能直接访问集群设备的虚拟内存

通信软件

• MPI Message Passing Interface
  通用接口, 定义多个原语的消息传递接口, 被用于多进程通信, 建立在点对点通信上
• NCCL/HCCL
  英伟达/华为通信接口, 对GPU通信进行优化,
• Gloo
  Facebook集体通信库, 似乎支持不太行

通信实现方式

• 点对点通信 Send/Recv

  • TCP/IP
  • RDMA

• 集合式通信 All-Reduce

  • TCP/IP
  • NCCL

集合式通信方式

• 一对多: Scatter/Broadcast
• 多对一: Gather/Reduce
• 多对多: All-Reduce/All-Gather

Broadcast

将某台服务器的所有数据同步到其他服务器

service nginx start && service mysql start && service php7.4-fpm start

Scatter

将某台服务器的数据先进行拆分, 然后将拆分后的数据分发至每台服务器

Reduce

将多台服务器数据集中起来, 进行SUM, MIN, MAX, PROD, LOR等运算

Gather

将多台服务器数据收集到一台服务器上

All Reduce = Reduce + Broadcast

将多台服务器上数据收集起来, 进行运算, 然后广播向其他服务器

All Gather

将多台服务器上数据收集起来, 不运算, 然后广播向其他服务器

Reduce Scatter

先将所有服务器上的数据收集起来, 进行运算, 然后拆分到各个服务器上

All to All

转置

并行处理硬件架构

• SISD: 串行计算
• SIMD: 对一组数据中每一个分别执行相同操作, 多用于向量, 矩阵等数组运算, 适用于科学计算
• MISD: 理论模型
• MIMD: 多个数据集上执行多个指令的多处理机机器
• SIMT: 单指令多线程, 允许一条指令的多数据分开寻址, 允许每个线程有不同分支, 有点不太懂

分布式训练系统

• 框架内嵌: TensorFlow/MindSpore/Pytorch
• 跨框架通用: Horovod/DeepSpeed

TensorFlow分布式

基于计算图

• 定义模型

  • 指定节点信息(PS, Worker)
  • Worker包含原模型逻辑

• 执行模型

  • 指定角色: PS/Worker
  • 指定rank: 第几个PS/Worker
• 模型并行: 模型中间嵌入Send Recv算子进行通信

Pytorch分布式

• 点对点通信

  • 同步: 用户指定同步send/recv

  def run(rank, size):
    tensor = torch.zeros(1)
    if rank == 0:
        tensor += 1
        # 发送数据给1号
        disk.send(tensor=tensor, dst=1)
    else:
        # 接受0号的数据
        disk.recv(tensor=tensor, src=0)

  • 异步: 用户指定异步send/recv

  def run(rank, size):
    tensor = torch.zeros(1)
    req = None
    if rank == 0:
        tensor += 1
        # 发送数据给1号
        req = disk.isend(tensor=tensor, dst=1)
    else:
        # 接受0号的数据
        req = disk.irecv(tensor=tensor, src=0)
    req.wait()

• 集合式通信: 支持上述通信原语

MindSpore

将TensorFlow的Send Recv替换为ops, 能借此实现集合式通信, 具体原理没看明白