郑重声明：原文参见标题，如有侵权，请联系作者，将会撤销发布！ 以下是对本文关键部分的摘抄翻译，详情请参见原文。
ASPLOS 2019
Abstract
最近的研究表明，在机器学习的背景下，去中心化算法比中心化算法能提供更好的性能。这两种方法的主要区别在于它们不同的通信模式，它们都容易在异构环境中性能下降。尽管人们一直致力于支持中心化算法来对抗异构性，但在去中心化算法中很少有人涉及到这个问题。
本文提出了第一个考虑异构的去中心化训练协议Hop。基于我们所发现的去中心化训练的一个独特特性，即迭代间隔，我们提出了一种基于队列的同步机制，该机制可以有效地实现去中心化训练中的备份workers和有界过时。为了应对确定性的减速，我们提出跳过迭代，以便进一步减轻较慢的workers的影响。我们构建了一个基于TensorFlow的Hop原型实现。在CNN和SVM上的实验结果表明，在异构环境下，该方法比标准的去中心化训练有明显的加速效果。
1. Introduction
机器学习（ML）已被证明是从数据中自动提取信息的有效方法。在语音识别[13]、图像分类[27]、自然语言处理[7]等多个领域，机器学习都取得了巨大的成功。获得有用的ML模型需要长时间的大数据集训练。随着模型和数据量的不断增加，分布式训练是目前解决多计算节点加速问题的有效方法。
分布式训练中最常用的算法是随机梯度下降（SGD）。它是一种迭代算法，在每次迭代中计算一小部分数据上的梯度，并相应地修改模型参数，直到收敛。在分布式环境下存在两种方法来执行SGD算法——中心化和去中心化。
在中心化方法中，名为参数服务器（PS）的中心节点负责维护参数，而名为worker的其他机器负责计算。目前可用的协议主要有块同步并行（Bulk Synchronous Parallel, BSP）[8,29,32]、过时同步并行（Stale Synchronous Parallel, SSP）[14,25,30]和异步协同[5,9,23]。在去中心化方法中，每台机器根据指定的通信图维护自己版本的参数和坐标，该通信图可以是密集的，如All-Reduce[2,6,31]，或稀疏的[17,20,21,28]。
分布式训练中的性能瓶颈主要来自两个方面：一是通信热点，尤其是对于PS，所有的工作人员都需要向PS发送/获取更新；二是异构性，这使得快速的worker在同步过程中等待缓慢的worker。第一个问题可以通过去中心化训练得到缓解：由于节点的通信量由图的程度决定，PS不再是瓶颈。因此，去中心化式训练最近显示出优于中心化训练的性能[20]。然而，它缺乏对异质性的支持，尽管在中心化训练中提出了各种方法来解决这个问题[3,14,15,16,23]。此外，与中心化训练相比，去中心化训练从诸如TensorFlow[1]、MXNet[4]和CNTK[24]等ML系统获得的支持要少得多，这使得将这种探索较少但潜在优越的方法应用于现实世界的问题变得困难。
为了弥补这一不足，本文提出了一种异构软件去中心化训练协议Hop。我们从确定迭代间隔这一去中心化训练的独特特征开始，根据worker之间的通信图确定其上界。此属性影响实现的成本和正确性。我们表明，目前的设计之一，NOTIFY-ACK[17]，事实上保守地限制了间隔，但牺牲了性能和实现灵活性。我们提出了一种基于队列的分布式协同同步机制。去中心化训练可以使用基于迭代间隔大小的更新队列来实现，也可以同时使用更新队列和令牌队列来控制迭代间隔。本质上，我们的一般方法将NOTIFY-ACK作为一种特殊情况。
为了减轻异构性的影响，我们提出了在去中心化的环境中备份workers和有界过时性。重要的是，我们证明了基于队列的同步在实现这些概念时是必不可少的，这是有根本原因的。有了备份workers，迭代间隙可以任意变大；因此，受控间隔不再是优化项而是必要项。对于有界过时，基于队列的同步允许在worker之间进行灵活的迭代间隔，这样可以实现多线程的好处。为了应对确定性的减速，我们提出跳过迭代以进一步限制单独的慢速workers的影响。有了以上的关键贡献，Hop显著地超越了分布式训练的最新水平。
我们构建了一个基于TensorFlow的Hop原型实现。我们在CNN和支持向量机上对系统进行了评估。结果表明，在随机和确定性减速的情况下，我们提出的特征、后备workers、有界过时性和跳过迭代，比标准的去中心化训练有显著的加速效果。此外，我们还观察到一个有趣的结果，即精心设计的频谱间隔较小的通信图在异构网络环境下甚至可以表现得更好。
2. Background and motivation
本文考虑的问题是在数据集S上用SGD来最小化损失函数F，每次迭代的更新函数由上式给出，其中ξ是从S中随机抽取的一小批数据，模型参数用x表示。该过程可以并行化，以便在分布式环境中执行[34]。一个著名的并行SGD机制是使用参数服务器（PS）进行训练[19]。
2.1. Distributed Training with Parameter Server
使用PS进行训练需要选择一个或几个中心节点作为PS，负责维护和更新模型参数[19]。其他机器，称为workers，从PS中提取参数，基于随机样本计算梯度并将梯度发送回PS，然后PS将根据接收到的梯度更新参数。
在最基本的设置中，每个迭代结束时workers都会进行同步。在PS收到每个worker的更新并将其应用于参数之前，不允许他们从PS中提取新参数。这样，workers总是使用相同的最新版本的参数。然而，同步设置有两个主要缺点：a）速度快的worker总是要等待速度慢的，这就是所谓的掉队者问题；b）通信瓶颈或热点很容易发生在PS。
目前缓解通信瓶颈的方法是在所有worker中采用ring All-Reduce[12,22]，而不使用PS，通过通信和计算的仔细重叠，消除了PS的通信热点。从逻辑上讲，它实现了All-Reduce，这意味着在迭代结束时，来自一个worker的更新将广播给所有其他worker。虽然ring All-Reduce通过计算隐藏了一些通信延迟，但当来自每个worker的单个更新在环中移动并到达所有其他worker时，实际延迟可能会增加。
2.2. Decentralized Training
最近，理论上首次证明了去中心化算法可以优于中心化算法[20]。这两种算法共享相同的计算复杂度[20]，但去中心化的算法享受更快的通信，因为通信负载分布在网络拓扑图上而不是集中在PS上。
尽管去中心化算法以前也有人研究过，但它的优点却隐藏了很久。先前的工作[10]表明，基于F是凸的假设，随着worker数量的增加，需要更多的迭代才能达到一定的精度。然而，在最近的工作[20]中，没有假设F的凸性，并且表明收敛速度相对于worker的数量呈现渐近线性加速。改进后的结果为去中心化训练的研究提供了动力。
在去中心化训练算法[17,20,28]中，没有中心节点；每个worker都维护自己版本的参数。工作人员基于预定义的网络拓扑彼此通信。在每次迭代中，worker计算梯度，并将其参数发送到其传出邻居，然后通过将其与传入邻居平均来更新其参数。在发送参数之前还是之后，将梯度应用于参数，这仍然是一个可选项。算法如图1所示。在该算法中，在应用梯度之前发送参数，这使得可以并行执行步骤1和2（并行方法）。另一种方法是交换步骤3和4，以便在应用梯度之后发送参数（顺序方法）。我们将此算法称为标准去中心化训练，并将在第3节讨论这两种变体。
2.3. System Heterogeneity
如第2.1节所述，性能下降的一个主要原因是掉队者问题，这是系统异构性的一个例子。一般来说，它涉及随机方面，例如由资源共享和硬件故障引起的减速，以及确定性因素，包括硬件计算能力和网络带宽的差异[15,16,21]。
从根本上讲，由于worker和PS之间或worker本身之间的固定通信模式，PS和ring All-Reduce都减少了缺乏解决异构执行环境的灵活性。对于PS设置，先前的工作已经提出了几种方法来解决这个问题，例如异步更新参数[23]、使用备份workers[3]、允许有界的过时[14]、动态调整学习速率[16]，以及当累积梯度达到显著阈值时发送累积梯度[15]。在ring All-Reduce中，限制性更强的通信模式使得一些技术无法实现，例如备份workers。事实上，执行过程中可能会遇到更多的慢通信链路和/或环中的掉队者。
对于最近才引起人们兴趣的去中心化训练来说，在异构环境中提高性能的努力相对较少。最近的一项工作[21]提出了一种异步方案，其中每个worker用随机选择的邻居来平均参数，而不是所有传入的邻居。然而，正如第5节将详细讨论的那样，它可能会导致死锁，并且只能用于特定类型的通信图。
2.4. Challenges and Motivation
去中心化算法可以优于中心化算法，因为它消除了PS中的通信热点。然而，异质性仍然是一个问题，因为worker仍然需要与它的邻居同步，因此存在掉队效应。可以想象，一个慢的worker或网络链路的影响可以通过连接的节点传播到整个图中。
在下一节中，我们将分析分布式同步的本质，提出基于PS解决方案的异构针对算法，并基于所提出的机制实现分布式训练系统。
3. Distributed Coordination
本节分析计算图和现有协议，即NOTIFY-ACK[17]，用于分布式训练。然后，我们解释并证明了一个重要性质：workers之间的迭代间隔。最后，提出了基于PS的分布式训练中的后备workers和有界过时性，以减轻异构性的影响。我们证明，现有的去中心化训练方案NOTIFY-ACK不能支持它们，因此提出了基于队列的同步（第4节）。
3.1. Notations
我们将分布式训练中workers之间的通信拓扑定义为加权有向图G=(V,E)，其中每个节点表示一个worker。在每个节点 i，Nin(i)={ j | (j, i)∈E } 和Nout(i)={ j | (i, j)∈E }分别表示传入和传出邻居的集合。边e=(i, j)∈E表示worker i 需要在训练期间向worker j 发送更新。边的权重反映了更新对worker j的“影响”程度。每个worker维护所有参数的副本，并且假设本地更新始终可用——在每个节点上都有一个自循环[17,20]，即对于所有i∈E，(i, i)∈E。
如果ui表示worker i 生成的更新，那么worker j 处的聚合更新由上式给出，其中W是G的加权邻接矩阵。先前的工作[17,20]表明，要使去中心化训练表现良好，G必须是连通的，W必须是双随机的，即W中行和列的和必须是一。通常，每个更新都有相同的影响：见上式。从worker i 发送到worker j 的更新由ui→j表示。如果更新是在第k次迭代中生成的，我们用ui→j(k)进一步指明时间戳。
3.2. Computation Graph in Decentralized Training
本节讨论去中心化训练worker的计算图的两种变体和权衡。迭代中的计算包括五个操作。
- Compute：workers使用随机选择的一批数据，并根据其当前模型参数计算梯度。
- Send：workers将其当前参数发送给其传出邻居。此操作是非阻塞的；worker可以发送其更新，而不管其外出邻居的状态如何。我们使用Send(i)来指定在第 i 次迭代中执行的发送操作。
- Recv：worker接收其传入邻居的模型参数。注意，worker不请求参数；相反，它们是由传入的邻居主动发送的。我们使用Recv(i)来指定在第 i 次迭代中发送的接收参数。Recv操作将阻塞，直到完全接收到参数为止。
- Reduce：worker用自己的参数与它收到的参数进行平均。
- Apply：worker在Reduce之前或之后对其当前参数应用梯度。
接下来，我们描述了两个在最近的工作中使用的计算图[17,20,28]，它们与图1中描述的算法一致。
串行方法：如图2(a)所示，在进入新的迭代时，每个worker将计算梯度，将梯度应用到其当前参数，然后将新参数发送到其传出邻居。当它接收到来自同一迭代中发送的所有传入邻居的新参数时，它将执行Reduce并用结果更新本地参数。[17]采用了这种模式。
并行方法：如图2(b)所示，每个worker将在迭代开始时发送其当前参数，同时基于相同的参数集计算梯度。在接收到来自其传入邻居的参数后，它执行Reduce，然后执行Apply，在将梯度应用于Reduce的值之后，生成要更新的局部参数。此模式在[20,28]中使用。
与串行方法相比，并行方法允许并行执行Compute和Reduce。然而，并行性是以不精确的梯度为代价实现的。具体来说，在图2(b)中，梯度是使用Reduce之前的参数计算的，但应用于Reduce之后的参数。这会损害梯度下降的效果。相反，图2(a)中的串行方法确保梯度是用同一组参数生成并应用于同一组参数。因此，并行方法具有更快的迭代速度，但需要更多的迭代来收敛，而串行方法需要更少但更长的迭代来收敛。它反映了执行效率和统计效率之间有趣的权衡[33]。我们在设计中使用并行方法。
3.3. Iteration Gap and the Mixed-Version Problem
去中心化训练的一个重要特征是潜在的大迭代间隔，即在任何给定的时间，在大范围的迭代中都可以找到workers。在中心化设置中，同步计算中不存在这样的间隔，因为workers必须在每次迭代结束时进行同步，从而确保所有的workers总是停留在同一迭代中。对于异步计算，只有在强制执行有界过时性才能保证收敛，这将在最快worker和最慢worker之间的迭代间隔上设置一个固定的上界。
然而，在去中心化设置下，我们发现间隔的大小只受图拓扑的限制，即使在同步训练中也会出现较大的间隔。在深入研究细节之前，我们首先建立一个基本的自然假设。
假设：当且仅当以下所有条件均为真时，worker才能前进到下一个迭代：a）它已完成当前迭代的计算；b）它已将当前迭代的更新发送到其传出邻居；c）它已从其传入邻居接收到当前迭代所需的更新。
上述假设已在先前的工作[17]中采用。注意，假设不会在相邻迭代之间设置全局界限。事实上，一个全局障碍可能会损害性能，因为它引入了不必要的等待：为了进入下一个迭代，一个worker必须等待所有其他worker完成当前迭代，而实际上它只需要等待它的传入邻居。基于这个假设，迭代间隙的一个重要结果如下：
定理1：在上述假设下，在任何给定的时间，worker i 的迭代和worker j 的迭代之间的最大迭代差是从节点 j 到节点 i 的最短路径的长度，即，Iter(i) - Iter(j) ≤ length(Pathj→i)，其中Iter(i)是worker i 的迭代（对于任意i∈V），length(Pathj→i)表示从节点 j 到节点 i 的最短路径长度。
证明：定理的证明是基于一个简单的观察：节点和它的传入邻居之间的最大迭代差为1，即，对任意i∈V，j∈Nin(i)，Iter(i) - Iter(j) ≤ 1。这是因为只有当worker i 接收到worker j 在第Iter(i) - 1次迭代的更新后，它才能够进入Iter(i)。注意，我们不能仅给出从 j 到 i 的有向边来推导Iter(i) - Iter(j)的下界，因为如果worker i 比worker j 慢，Iter(j)可以比Iter(i)大得多。
对于任意两点 i 和 j，考虑到从 j 到 i 的最短路径。从 j 到 i，基于前面的观察，每一次我们经过点v，Iter(v) - Iter(j)的最大可能值加一。由于Iter(j) - Iter(i)=0，并且路径上有length(Pathj→i)个其他节点，我们有Iter(i) - Iter(j) ≤ length(Pathj→i)。
迭代间隔的存在产生了混合版本问题[17]，即，worker可以从其传入邻居同时接收各种迭代的更新。如果网络很小，问题可能并不严重，但是随着网络的大小增长，length(Pathi→j)可以很大，其中j∈Nin(i)。在这种情况下，有大量的ui→j's是由worker j 产生的，但不由worker i 消耗。
虽然本文是第一个提出并证明这个定理的，但我们发现[17]中的一个先前的解确实提供了一种机制来限制worker之间的间隔。具体来说，除非worker已收到先前更新已被使用的确认，否则它将被阻止发送更新。如图2(a)所示，这个名为NOTIFY-ACK的方法需要一个worker在Reduce之后向它的所有传入邻居发送一个ACK消息，宣布它们发送的参数已经被使用。除非接收到ACK，否则传入邻居将不会执行下一次发送。本质上，除了定理1中解释的从发送方到接收方的前向依赖性外，NOTIFY-ACK还强制执行从接收方到发送方的后向依赖性。它导致相邻workers之间的迭代间隙过大，原因如下。
对于j∈Nin(i)，我们已经从定理1知道，Iter(i) - Iter(j) ≤ 1。对于Iter(j) - Iter(i)，由于worker j 需要从worker i 的第(Iter(j) - 1)次迭代接收到一个ACK以保证前进到迭代Iter(j) + 1，它们的迭代差最大为2。
对于任意一组worker(i, j)，Iter(i) - Iter(j)的上界不能仅仅用length(Pathj→i)的函数来表示。这是因为任意在 i 与 j之间的路径，对于位于路径上的任意v∈Nin(u)，我们必须保证 -2 ≤ Iter(u) - Iter(v) ≤ +1 始终成立。换句话说，每一次我们从节点v经过节点u，相比于worker v，worker u 至多超过一次迭代或者至多落后两次迭代。作为结果，Iter(i) - Iter(j)的上界，即，worker i 超过了worker j 多少，是沿着Pathj→i或Pathi→j的最大迭代间隔的最小值，受到 u 与 v 之间的约束：Iter(i) - Iter(j) ≤ min(length(Pathj→i), 2 × length(Pathi→j))——比定理1确定的迭代间隔更具限制性。
尽管NOTIFY-ACK可以确保worker接收更新的序列顺序，但紧密限制的迭代间隔使它成为异构环境中的一个不可取的选择，在这些环境中，worker将以不同的速度前进。一个直观的例子是，一个快速worker可能会在接收到来自传入邻居的所有更新之后等待一个慢速运行的传出邻居的ACK，事实上，它已经准备好进入下一个迭代。为了应对异构性，我们提出了两种机制，备份worker和有界过时性，以适应更大的迭代间隔。在中心化训练[3,14]中存在类似的机制，但它们在去中心化设置中没有被应用。正如我们将要展示的，协议和实现都需要仔细地重新设计。
3.4. Decentralized Training with Backup Workers
减轻异构性影响的一种有效技术是使用备份workers[3]。在中心化训练中，通过允许参数服务器所需的更新数量小于worker数量，可以很容易地实现。假设有n个workers，每个迭代需要的更新数为m(m<n)。从每个PS的角度来看，有效的worker数量是m，因为它在每个迭代中需要m个更新，而剩余的(n - m)个workers是“备份”。这样，在不影响训练速度的情况下，我们可以容忍最多(n - m)个慢workers，以防随机减速甚至意外节点崩溃。
我们自然会将备份workers应用到去中心化训练中，方法是将每个worker所需的更新数量设置为小于其传入邻居的数量。如图3(a)所示，在去中心化的3-worker设置中，每个worker与其他两个worker可以互相传输，并且在每个迭代中只需要一个来自其邻居的更新。在当前状态下，worker A停留在迭代0处，而worker B和C可以分别前进到迭代3和4。如果没有后备worker，worker A会拖累B和C的进度，因为它们都会依靠A的更新来前进。有了备份workers，B和C可以前进到以后的迭代。
然而，这个简单的机制导致了一个根本问题：两个worker之间的迭代间隔可以任意大。从图3可以很容易地看出，实际上B和C可以在任何迭代中，因为它们可能只依赖于彼此的更新。
由于NOTIFY-ACK机制意味着相邻worker的迭代差不超过2，备份worker的好处无法完全实现。例如，worker B和C都可能被困在迭代1中等待来自A的ACK(0)，如果没有A的进度，ACK(0)将不会到达。因此，即使worker A不需要等待worker B和C的更新，worker A仍然可以拖累它们的进度。本质上，是NOTIFY-ACK中的机制阻止了备份workers的实现。
3.5. Decentralized Training with Bounded Staleness
有界过时性[14]是中心化训练的另一种技术，用于容忍缓慢的worker或参数服务器和workers之间的缓慢通信。为了实现这一点，采用了一个异步参数服务器，但对最快的worker迭代和最慢的worker迭代之间的差异施加了一个上限，只要保持过时的界限，worker可以自由地前进到一个新的迭代。
对于去中心化训练，很难实施全局有界过时性，这意味着整个系统中最快和最慢的worker的迭代差不能超过界限。显然，它将通过引入某种全局进展监测机制来确保这种性质，从而违背了去中心化的性质。相反，我们建议以一种局部方式应用有界过时性：一个worker可以进入一个新的迭代，只要它在最多s次迭代之前从它的所有传入邻居接收到更新，其中s是过时性的上界。由于更新是在局部传递的，所以执行过时性绑定是很简单的。我们认为，局部约束是在去中心化的环境中自然地采用有界过时，从而得到有效的实现。
图3(b)显示了在同一个3-worker去中心化设置中使用的过时界限为5的示例。worker A、B和C分别处于第0次、第5次和第3次迭代中。由于随机因素（例如资源共享），worker A暂时减慢。然而，通过使用过时性，worker B和C可以继续前进，直到第5次迭代。有界过时性的好处是，即使某些worker放慢了速度，仍然可以取得一些进展——减缓的影响得到了缓解。
让我们再考虑一下NOTIFY-ACK。不幸的是，它直接对迭代间隔（即2）施加了严格的限制，因此任何大于1的过时性限制都是不可能的。例如，在图3(b)中的情况下，如果使用NOTIFY-ACK作为协议，worker B和C将无法进入迭代1，因为它们仍在等待worker A的ACK(0)。与备份workers类似，NOTIFY-ACK的机制影响局部有界过时性的实现。
到目前为止，讨论的主要收获如下。尽管NOTIFY-ACK指出并遏制了混合版本问题，但它并没有意识到更大的潜在任意迭代间隔。如我们所示，NOTIFY-ACK限制性太强，使得相邻worker之间的间隔非常小，这将1）限制去中心化训练的可能性；2）阻碍后备worker和有界过时性的实施。为了在解决混合版本问题时支持更大的迭代间隔，我们提出了一种基于队列的协调方案。
4. Queue-based synchronization
本节首先介绍更新队列和令牌队列作为去中心化训练协议的构建块。然后，我们将介绍如何使用它们来有效地实现备份workers和有界过时性。
4.1. Update Queue
为了支持混合版本和较大的迭代间隔，我们提出了一种基于队列的协调方案，将接收到的更新存储在FIFO队列中，称为更新队列。worker i处的更新队列由UpdateQ(i)表示。我们进一步定义了以下队列操作：
- q.enqueue(update, iter = None, w_id = None) 将update推送到队列中，其中iter和w_id是标记，用(iter, w_id)表示，q是队列的名称。输入iter表示生成更新的迭代的索引，w_id表示发送方worker的索引。
- q.dequeue(m, iter = None, w_id = None) 从队列中取出标记为(iter, w_id)的前m个条目，并返回包含这些条目的列表。如果队列中没有足够的标记为(iter, w_id)的元素，则此函数将阻塞。如果未指定其中一个标记，则将返回与另一个标记匹配的前m个条目。如果两者都未指定，则返回前m个条目，而不考虑其标记。如果需要，还可以将返回条目的标记也进行返回。
- q.size(iter = None, w_id = None) 返回队列中标记为(iter, w_id)的条目数。如果未指定其中一个标记，则返回与另一个标记匹配的条目数。如果两者都未指定，则返回队列中的条目总数。
基于更新队列，标准的去中心化训练算法如图4所示。要将更新从 i 发送到 j，worker i 直接将参数enqueue到worker j 的更新队列。要接收更新，worker可以在本地将从各种worker和迭代发送的更新进行dequeue。但是，还有一个问题——队列应该有多大才能容纳所有更新？基于定理1，对于任意一个worker i 及其传入邻居 j，worker j 可以在 i 之前length(Pathi→j)次迭代。这意味着UpdateQ(i)必须能够存储来自 j 的length(Pathj→i)+1次不同迭代的更新。当worker的数量很大时，从 j 到 i 的最短路径也可以很大，因此队列的容量也必须很大，这将给系统内存带来相当大的压力。示例如图5所示。
4.2. Token Queue: Controlled Iteration Gaps
为了解决这个问题，我们提出令牌队列作为一种机制来控制相邻workers之间的迭代间隔。请注意，根据标准去中心化训练的性质，每个worker最多只能比其传入邻居提前一次迭代；因此，我们只需要控制worker的速度，与其潜在的较慢的传出邻居相比。
在我们的设计中，每个worker为每个传入邻居维护一个令牌队列。每当一个worker试图进入一个新的迭代，它必须从每一个传出邻居那里获取一个令牌。考虑到本地worker，队列中的令牌数决定传入邻居可以推进的迭代次数。假设我们希望相邻worker之间的迭代间隔不超过预定义的正整数常量max_ig，我们用TokenQ(i→j)表示worker i 的令牌队列存储worker j的令牌，其中 i∈Nout(j)。
- Initialization：在第一次迭代开始时，每个worker都会在其维护的每个令牌队列中放置max_ig个令牌。
- Remove token：当一个worker i 试图进入一个新的迭代时，它必须从它的每个传出邻居中移除一个令牌——对于每个 j∈Nout(i)，从TokenQ(j→i)中移除一个令牌。
- Insert token：当worker i 进入一个新的迭代时，它将在每个本地令牌队列中插入一个令牌——对于每个j∈Nin(i)，将一个令牌插入到TokenQ(i→j)。这使得它所有的传入邻居都能更进一步。
定理2：对于具有令牌队列的标准去中心化训练，Iter(i) - Iter(j)的上界由下式给出：min(length(Pathj→i)), max_ig × length(Pathi→j)).
证明：首先，我们考虑一对相邻的worker(i, j)。对于任何一个worker i 和它的传出邻居 j，我们已经从定理1中知道了Iter(j) - Iter(i) ≤ 1。为了得到Iter(i) - Iter(j)的上界，我们将以推理的方式证明TokenQ(j→i).size() = Iter(j) - Iter(i) + max_ig在所有迭代都成立，其中size()函数返回令牌队列中的令牌数。训练开始时，Iter(i) = Iter(j) = 0，初始化TokenQ(j→i)生成TokenQ(j→i).size() = max_ig。因此，上述等式成立。只有当worker i 或 j 进入新的迭代时，等式中的变量才会改变。如果worker i 前进，它必须从TokenQ(j→i)中移除一个令牌；同时，Iter(i)增加1。因此，等式的两边都减少了1，等式仍然成立。类似地，如果worker j 进入一个新的迭代，它必须在TokenQ(j→i)中插入一个令牌。因此，等式的两边都增加了1，等式仍然成立。因为TokenQ(j→i)是非负的，我们有Iter(j) - Iter(i) + max_ig = TokenQ(j→i).size() ≥ 0，因此Iter(i) - Iter(j) ≤ max_ig。
接下来我们考虑任意对(i, j)。因为图是连通的，所以存在一条从 i 到 j 的路径，反之亦然，如图6(a)所示。这两条路径可以看作是两个基本场景。在图6(b)中，由于相邻workers案例的早期证明，Iter(i) - Iter(j) ≤ max_ig × length(Pathi→j)。在图6(c)中，根据定理1，Iter(i) - Iter(j) ≤ length(Pathj→i)。一般情况下是两种场景的组合，因此我们有Iter(i) - Iter(j) ≤ min(length(Pathj→i), max_ig × length(Pathi→j))。
迭代间隔被较小的间隔限制的直觉是，反之，较大的间隔变得不可行，这是由于没有足够的标记（如果图6(b)中的间隔较大），或者实际上没有足够长的路径（如果图6(c)中的间隔较大）。总的来说，所提出的令牌队列提供了一种灵活的参数化方法来约束迭代间隔。任何令牌队列的容量上限是max_ig · (length(Pathi→j) + 1)。它直接从应用定理2中证明的Iter(i) - Iter(j)的上界到TokenQ(i→j).size() = Iter(i) - Iter(j) + max_ig。
现在我们将令牌队列应用于图5(b)中的示例。我们可以把max_ig设为3。worker A上的令牌队列在第0次迭代开始时包含3个令牌。每当worker B进入一个新的迭代时，它必须从A获得一个令牌。由于A没有进展，B最多可以从A获得3个令牌，这使B能够到达第三个迭代，但不能再继续前进。因此，A一次最多只能处理4个更新，并且防止了图中情况的发生。
使用令牌队列的去中心化训练算法如图7所示。在有界迭代间隔的情况下，无论图的大小或拓扑如何，UpdateQ(i)的所需大小的上界都是(1 + max_ig)|Nin(i)|。尽管令牌队列的使用可能仅对仅基于更新队列的算法提供一个边际改进，但我们稍后将在第4.3节中看到，当使用备份workers来减轻异构性的影响时，限制迭代间隔是绝对必要的。
4.3. Supporting Backup Workers
将后备workers应用于去中心化训练相对直观。如第3.4节所述，worker不需要从每个传入邻居获取更新，而只需要从几个邻居获取更新就可以前进到下一个迭代，即所需的更新数量小于其传入邻居的数量。在图8所示的算法中，当从本地更新队列收集更新时，worker首先通过在dequeue函数的输入中指定数字来确保它有足够的更新数量。然后它检查队列中是否有任何可用的额外更新，因为接收到的更新数可能超过所需的数目。上述过程也可以替换为while循环，该循环将不断从队列中取出可用的更新，直到数量足够为止。
使用较少的更新的一个问题是，稍后到达的未使用的更新可能会累积在更新队列中。随着一次又一次的迭代，它会占用越来越多的空间，不可避免的会导致溢出。我们提出了一个由两部分组成的解决方案：a）定期清除过时的更新；b）在很少通信开销的情况下，通过在Send之前检查接收者的迭代来防止发送不必要的更新。我们将在第6节中详细解释。
备份workers设置的另一个显著特点是迭代间隔是无界的。如我们之前在图3中所示，worker B和C只能根据彼此的更新进展到任意大的迭代，而他们的公共邻居A则保持在迭代0中。因此，限制迭代间隔是正确执行去中心化训练的必要条件。在这种情况下，令牌队列是设计中不可或缺的一部分。
4.4. Supporting Bounded Staleness
如第3.5节所述，在“有界过时”设置中，worker可以进入一个新的迭代，只要它在最多s次迭代之前从所有传入的邻居接收到更新，其中s是过时的上限。然而，在去中心化训练中加入过时的具体方法尚未发现。尤其是，如何处理过时的更新仍然是一个问题。
我们观察到，在以后的迭代中发送的模型参数包含了以前更新所携带的信息，因为以后的更新是建立在以前的更新之上的，而在更新发送的参数中累积了梯度。因此，我们提出尽可能使用最新的可用更新，并丢弃其余的更新。具体地说，当worker从其本地更新队列收集更新时，它将比较标记并从每个传入邻居中选择最新的更新。如果更新在过时范围内，则视为令人满意；否则，将删除更新。如果在当前迭代或先前迭代中接收到的来自传入邻居的更新不在当前过时范围内，则工作进程将阻塞，直到从相应邻居获取更近的更新。当worker从每个传入邻居接收到满意的更新时，它将对新接收到的更新执行Reduce。请注意，要reduce的更新可能来自不同的迭代，因此简单的平均值可能不是聚合它们的最佳方式。我们比较了简单平均法和基于迭代的加权平均法，发现后者的性能稍好一些。对于迭代k中的worker j，我们已经确定的更新公式如下：
对于迭代间隔，在s的过时界限下，对于j∈Nin(i)，我们得到了Iter(i) - Iter(j) ≤ s + 1。这是因为对于worker i 进入Iter(i)+1，它需要worker j 的更新，至少与uj→i(Iter(i) - s)一样新。因此，迭代间隔的上界由由Iter(i) - Iter(j) ≤ (s + 1) · length(Pathj→i)。我们看到，与标准的去中心化设置相比，迭代间隔已经大大增加。因此，我们还使用令牌队列作为限制间隔的方法。算法如图9所示。
5. Deterministic Slowdown
为了减轻异构性的影响，备份workers和有界过时性通过允许一个worker比它的几个甚至所有传入邻居更快地前进来放松同步方案。但是，worker的迭代和其邻居的迭代之间的间隔仍然是有界限的，无论是技术本身（在有界过时情况下）还是令牌队列的使用（在备份workers情况下）。因此，如果一个worker遭受严重的确定性减速，它最终会拖累它的邻居，然后拖累整个节点图。因此，存在一个两难的问题：如果我们不绑定迭代间隔，那么可以针对确定性减速实现更好的性能，但是这样的系统是不现实的，因为worker必须准备从无穷多迭代中接收更新；另一方面，如果我们用像令牌队列之类的机制来约束迭代间隔，然后，每当节点以确定的方式减速时，其他节点只能在等待慢worker之前获得有限的进度，并且它们所能做出的最大进度严格地由表1最后一行所示的约束确定。
有两种可能的解决方案。一种是开发新的算法来消除迭代间隔上的界限，并支持无限大的迭代间隔。事实上，AD-PSGD[21]就是一个例子，在这个例子中，每一个worker通过在每次迭代结束时用随机选择的邻居进行平均来更新自己的参数，而不管邻居的迭代是什么。然而，该算法容易产生死锁，为了防止它的存在，现有的解决方案需要通信图G为有两个部分的[21]，极大地限制了用户对通信拓扑的选择。另一种是开发新的系统机制来识别慢worker，并寻求一种让慢worker进步的方法，以便恢复训练过程。
我们考虑采用一种系统方法，通过检查传出邻居的令牌队列中的令牌数量，让慢worker标识自己。当它在每次迭代结束时从它的传出邻居获取所需的令牌以进入新的迭代时，这可以很方便地完成。对于worker i 及其传出邻居 j，TokenQ(j→i)中的令牌数正是Iter(j) - Iter(i) + max_ig。如果TokenQ(j→i).size()对于所有的 j∈Nout(i)都是大的，我们可以想象，worker i 很可能是一个掉队者。
为了让掉队者取得进展，我们提出跳过迭代，也就是说，一个慢的worker可以跳过几个迭代，允许其他worker前进。根据在第4.2节中提出的令牌队列方案，可以跳过的最大迭代次数是由max_jump = minj∈Nout(i)TokenQ(j→i).size()确定的，因为它进入每一个新迭代都需要一个来自其每一个传出邻居的新令牌。虽然在我们提出的令牌队列方案下允许跳过max_jump次迭代，但我们认为，对于速度慢的worker i 来说，给出的直观上限max_jump - max_ig是荒谬的。
假设worker i 将跳转到迭代k。在我们的设计中，在跳转之前，worker i 将通过执行Recv(k - 1)和一个后续的Reduce来更新其参数，在(k - 1)次迭代中，将当前参数与其传入邻居发送的更新进行平均，即{uj'→i(k - 1): j'∈Nin(i)}。这是为了确保跳转后，worker i 的参数不会显得太过时，以便将来由worker i 发送的更新不会损害训练过程。这样一来，缓慢的worker就不会一直是个掉队者了。请注意，为了确保令牌队列方案的正确性，当执行从迭代k0到迭代k的跳转时，worker将需要从其每个传出邻居获取(k - k0)令牌，并将(k - k0)令牌放入为其传入邻居准备的每个本地令牌队列中。
图10给出了执行跳转的两个示例，一个用于有界过时的情况，另一个用于备份workers。红色的更改表示跳转，绿色的更改表示跳转启用的新进度。(a) Worker B和C因过时界限为4而无法前进。如果没有A的跳转迭代，B和C的速度将不超过A的4次迭代。但是，有了跳转迭代，缓慢的worker A可以快速跳转到迭代9，这样训练就可以在A再次前进之前顺利地恢复4次迭代。(b) Worker B和C被阻塞，因为令牌队列保证了有界的迭代间隔——它们无法从A获取新的令牌，因为令牌队列是空的。如果没有A的跳转迭代，它们的迭代速度将不超过A的5次，但是如果跳转到第10次迭代，它们可以在A取得新进展之前，不间断地训练另外5次迭代。
对于要执行Recv(k - 1)的worker来说，它可能看起来像是一个问题，它必须等待它的传入邻居到达(k - 1)次迭代。然而，我们认为，由于确定性的减速，worker i 很可能也落后于它的传入邻居；即使它的一个传入邻居也很慢，有界过时性或备用workers的机制将确保worker i 可以轻松地进行第k次迭代。此外，虽然跳过迭代意味着缺少来自worker i 的几个迭代更新，但这并不是问题，因为即使发送了更新，它们也会过时，因此会被worker i 的传出邻居丢弃（在备份workers的情况下）/接收到非常小的权重（在有界过时的情况下）。
为了使我们所提出的机制具有灵活的设置，我们的系统允许用户指定一个worker在一次跳跃中可以跳过的最大迭代次数，以及触发跳转的条件，例如，如果一个worker超出了用户指定的迭代次数，那么它可能只跳过迭代。
6. Implementation
我们用TensorFlow实现了我们的系统。具体地说，我们设计的队列是基于TensorFlow中的详尽且灵活的FIFO队列。初始化指定队列条目的数据类型、形状和容量。FIFO队列支持常见的功能，包括enqueue、dequeue、dequeue_many和size。
6.1. Collecting Updates Matching a Tag
要实现第4.1节中定义的队列操作，我们只需要使用标记来增强每个FIFO队列条目，该标记用于匹配特定迭代和/或来自特定邻居的更新。
标记的一个简单实现是使用一个FIFO队列作为每个worker的更新队列，并将标记作为队列条目的一部分。每当一个worker从队列收集更新时，它会从队列中取出所有条目，并保留具有匹配标记的条目。此过程在while循环中进行，直到worker获得所有需要的条目。这种方法的问题是处理不匹配的条目可能会很麻烦。它们不能被丢弃，因为它们可以来自以后的迭代，并且将在将来使用。这可能是因为我们不认为网络会保留消息顺序。我们不能简单地将它们放回队列中，因为随着while循环的继续，它们将一次又一次地退出队列。在根据标记执行部分可用更新的Reduce后，可以将它们存储在本地，但这将使记账复杂化，并消耗大量本地内存——大约是模型大小的max_ig倍。
我们提出了一个解决方案，以防止在几乎零内存开销的情况下对更近的迭代进行出列更新。我们不使用单个队列，而是定义多个队列，每个队列对应一个迭代。队列在迭代中重用的方式类似于旋转寄存器。要选择正确的enqueue或dequeue，worker通过计算模块mod(iter, #queues)来确定队列索引，其中#queues是队列总数。#queues被设置为max_ig + 1，因为根据定理1，一个worker最多可以接收(max_ig + 1)个不同的更近或当前迭代的更新。在标准情况下（第4.1节），worker只能接收较近或当前的更新，并且它始终可以将正确的更新dequeue；在备份workers情况下（第4.3节），worker也可以接收较旧的更新，但较旧的更新将被丢弃。我们的解决方案实质上是将原来的大型单队列划分为多个小型队列，所消耗的总空间基本保持不变。
至于通过w_id标记来区分发送者，也可以通过定义多个队列来实现。但这在我们的系统中是没有必要的，因为我们只在使用有界过时的时候使用w_id标记，这只需要处理所有条目的一次传递：在具有相同w_id标记的条目中，保留最新的一个条目，并丢弃其余条目。
6.2. Handling Late Updates
如第4.3节所述，使用备份workers时，Reduce中未使用的更新可能会累积在队列中。在我们的设计中，过时更新的影响通过以下两种方式得到缓解：a）在以后的迭代中，在dequeue/dequeue_many操作中发现并丢弃过时更新。如第6.1节所述，这已纳入我们的系统。b）在发送更新之前询问接收者的迭代。如果接收者的迭代大于本地worker的迭代，则不要发送更新。此方法创建一个小的通信开销，但在更新过时的时候可以节省更多的开销。更重要的是，它可以有效地减少过时更新的数量，现在过时更新的唯一来源是那些在接收者执行dequeue/dequeue_many操作时正在运行的更新。
我们还考虑了TensorFlow提供的一个更为定制的结构，称为条件累加器，它只接受用正确的local_step发送的更新，这是迭代的另一个概念。如果local_step不正确，则将删除更新。这似乎是一个完美的解决问题的办法，但我们在实验中观察到，这一性质不可能总是得到保证。发送时处于最新状态的更新在接收时可能会过期，条件累加器将错误地接受更新。这与我们遇到的FIFO队列在动态更新时遇到的问题完全相同。
7. Experiments
7.1. Dataset and Models
我们评估两个机器学习任务，即图像分类和网络垃圾邮件检测。对于图像分类，我们在CIFAR-10[18]上训练VGG11[26]网络；对于网络垃圾邮件检测，我们在webspam数据集上[11]训练SVM。
7.2. Experiments Setup
我们使用具有1000Mbit/s以太网连接的CPU集群，在4台机器上运行16个workers：每台机器有4个workers。我们使用http://leon.bottou.org/projects/sgd和AD-PSGD[21]中规定的超参数设置，并做了一些修改：批大小：128；学习率：VGG为0.1，SVM为10；不使用学习率衰减策略；动量：0.9；权重衰减：VGG为10-4，SVM为10-7。对于SVM，我们用对数损失代替hinge损失。
7.3. Results and Analysis
7.3.1. Heterogeneity with Random Slowdown
我们模拟了一个异构环境，在每个迭代中以1/n的概率随机地将每个worker减慢6倍，其中n是workers的数量。我们在三种不同的通信图（如图11所示标记为环、基于环和双环）上进行了有减速和无减速的实验，结果如图12所示。所有的图表都无法避免放缓。此外，我们发现稀疏图受随机减速的影响较小。
7.3.2. Comparison to Parameter Servers
对于去中心化算法，在基于环的拓扑上进行训练（图11）；对于PS，我们采用BSP并使用一台额外的机器作为参数服务器。如图13所示，在异构或同构环境中去中心化训练的收敛速度比同构PS快得多。由于参数服务器算法在异构环境中不可避免地会减慢速度[16]，因此我们不进行此实验。
7.3.3. Effect of BackupWorkers
我们主要针对随机异构性设计备份workers，因为在具有确定性异构性的环境中（例如，当worker运行得慢得多时），由于令牌限制，整个进程仍然会减慢。
我们在两个不同的通讯图，基于环的图和双环图上测试了我们的系统。我们使用一个备份worker（即每个节点可以少接收一个更新），两个图上的结果类似于图14所示：使用备份worker的训练比标准的去中心化算法在挂钟时间上收敛得更快。结合图15所示逐步的损失曲线，我们认为，尽管接收到的更新次数较少会损害每次迭代的进度，但与每次迭代执行时间中获得的加速相比，效果微不足道（图16中显示了高达1.81的加速）。
7.3.4. Effect of Staleness
我们在一个基于环的图上进行了实验，使用了6次随机减速和值为5的过时界限。如17所示，具有过时性的系统可以实现与具有备份workers的系统相似的加速，并且它们都优于标准的去中心化设置。
7.3.5. Effect of Skipping Iterations
实验是在一个有16个workers的基于环的图上进行的，而一个worker被确定地选择为4次减速。我们测试两个设置：一次最多跳跃2个迭代，一次最多跳跃10个迭代。如图19所示，跳过迭代比简单的备份workers设置显示出更好的性能，最多跳过10次迭代可提供最快的收敛速度，其速度比标准去中心化系统快2倍以上。此外，图18显示，在跳过迭代的情况下，掉队者对迭代持续时间的影响可以大大减少——从3.9倍的减慢减少到3.90/3.43 ≈ 1.1倍的减慢，这有助于在挂钟时间上显著提高收敛速度。
7.3.6. Effect of graph topology
我们比较了异构环境中的3个图，其中8个workers在3台机器上分布不均（图21）。基线图是基于环的图，其高光谱间隙为0.6667。我们提出的两个图是受workers的异质分布启发的：一个物理机器内部使用all-reduce图，而机器之间使用环图。它们有更小的光谱间隙，分别为0.2682和0.2688，但是我们的实验表明它们比基于对称环的图表现得更好（图20）。理论上，光谱间隙越大，收敛所需的迭代次数就越少[17, 20]。然而，我们的实验并没有显示出收敛速度关于迭代的显著差异，即使在光谱间隙非常不同的情况下（图20）。此外，由于图的拓扑结构以及系统的异构性，迭代的持续时间可能会有很大的变化，这为设计通信图时应该考虑更多的因素提供了见解。
8. Conclusion
本文提出了一种针对异构的去中心化训练协议Hop。基于我们所发现的去中心化训练的一个独特特性，即迭代间隔，我们提出了一种基于队列的同步机制，该机制可以有效地实现去中心化训练中的备份workers和有界过时性。为了应对确定性的减速，我们提出跳过迭代，以便进一步减轻较慢的workers的影响。我们构建了一个基于TensorFlow的Hop原型实现。在CNN和SVM上的实验结果表明，在异构环境下，该方法比标准的去中心化训练有明显的加速效果。