CSE_lecture27:Distributed Training
Distributed Computing frameworks: MapReduce, computation graph & Distributed training
MapReduce也有其局限性：
- 单个MapReduce难以实现排序等应用，因此使用多个MapReduce并串联在一起
- chaining multiple MapReduce tasks并不是好的解决方法，因为需要拆分，同时容错难度上升，且在任务之前存在反复写入GFS的开销
from MapReduce to DAG(computation graph)
计算图相当于将任务和数据抽象成节点，构成有向无环图，此时Reducer知道哪些节点需要它的output，因此只需要放在内存即可，不用访问GFS，减小开销
不存在依赖关系的节点可以并行计算
DAG的容错更加复杂，需要级联的恢复。对于小型计算任务，优先使用re-execution；对于大型计算任务，优先使用checkpoint
DAG看似比MapReduce更通用，但不同应用的并行度不一样，怎么生成高并行度的DAG也不一样
how to express AI program to distributed framework? dataflow graph
复杂的计算可以表征成计算图，好处在于：
- 方便计算导数，自动求导并链式反向传播
- 兼容性好，根据计算图实现对应硬件的算子即可
- 将优化与算法工作解耦
case study: distributed training
计算图是顺序执行的，难以在不同节点上并行执行。但对于模型训练的模式相对确定，利用这些模式来做并行处理
(sync) data parallel
$w^1 = w^0 - \frac{\alpha}{B}\sum_{i = 1}^{B}\frac{\partial J(w^0)}{\partial w}$，因此计算梯度时可以并行计算各个数据，累加在一起即可
不需要每次load W，直接使用locality即可；也不需要job manager，因为每台机器做的事情一样(SPMD)，即程序一样，只是数据不同
设计的目标为high GPU utilization，衡量的指标为MFU(model FLOPS utilization)，即GPU的利用率，影响因素有：
- 能否在forward和backward时用满GPU，取决于算子
- 最小化空转时间（数据加载时间），可以在数据加载的同时并行进行forward+backward的计算
- 数据同步时间的优化，即sum的时间。根据amdahl’s law，串行部分的比例越高，即使并行机器增加也很难提升性能
关键步骤为allreduce：在进行累加之后，还要广播结果，让每台机器更新参数
尝试1: parameter server(PS)，中心化的服务器聚合数据，由它reduce后广播出去。该尝试的瓶颈为网络传输，假设N为参数量，P为processor个数，则需要的带宽为 $O(P*N)$，而这很难达到
尝试1.5: co-located & sharded PS，即利用processors之间的流量，此时每台机器只做一个partition的数据，传输量降低到 $O(N * (P - 1) / P)$，完成之后再广播出去
但问题在于fan-in，即多台机器同时向一台机器发送消息，导致网卡缓存溢出，出现频繁的切换，发起拥塞控制，使网络性能变差
尝试2: de-centralized approach for allreduce，即一个接一个发送数据，fan-in性能很好，但通讯轮数达到 $O(P * P)$，因此将后续的通信提前，构成一个环
此时数据传输还是 $O(P * N)$，因此使用partition，每个节点只做自己的partition，但还有一些问题：GPU需要始终维持与其他所有的节点通信，需要频繁在host查询连接信息；节点间需要同步。因此期望每台机器只和相邻节点通信
尝试3: ring allreduce，将每个节点的partition组装，只进行相邻节点通信；最后再做一轮广播，让每个节点拥有所有partition，此时传输量为 $2 * (P - 1) * N / P$，fan-in为O(1)
不同partition的allreduce顺序不一致，因此reduce需要具有交换律
ring allreduce的通信轮数高，故瓶颈在于latency而不是bandwidth
使用tree可以进一步降低通信轮数，而fan-in变为2，使用两棵tree来避免机器的空转
model parallelism
参数量的增长远快于GPU显存的扩大，因此需要将模型切分到不同的GPU上
一种切分是根据layer切分。stage包含多个layer，分发给多个GPU运行。为了防止GPU空转，需要将stage拆分成多个小的batch(micro-batching)，从而构成parallelism，减小bubble
问题在于怎么设置p和m的值，bubble的数量是p-1，因此浪费的时间为：$(p-1)*(t_f+t_b)$，占比为 $\frac{p-1}{m}$，故需要增加m或减小p
增加m会导致计算效率变低，这是因为随着batch size增加，locality利用越来越好，使执行时间增长越来越慢，故batch size越小效益越低。但增大batch size也会影响训练的准确度，且让内存压力增加
减小p也不现实，因为单个GPU放不下海量的参数
tensor parallelism对单个layer作进一步切分，但需要网络通信将矩阵合并在一起，传输量为 $(P-1)\times \frac{di \times B}{P} \sim = di \times B$
model parallelism在micro batch之间存在通信开销
tensor parallelism的开销为：$Forward \sim = 2 \times di \times B \ Backward \sim = 4 \times di \times B$，而layer之间的pipeline parallelism的开销为：$Forward \sim = di \times B \ Backward \sim = di \times B$，因此前者的网络速度需要远快于后者。因此现代GPU内部的带宽会是跨机器之间的数倍，在同一台机器之内使用TP，跨机器使用pipeline parallelism
为了应对不同机器的速度不同的问题，进行异步训练，当一部分机器计算出来时就进行allreduce，剩下的等到下一次iteration进行，会牺牲正确性