Stanza: Layer Separation for Distributed Training in Deep Learning
Xiaorui Wu Hong Xu Bo Li Yongqiang Xiong










 


: Layer Separation for Distributed Training in Deep Learning

    
Xiaorui Wu    Hong Xu

City University of Hong Kong





Bo Li

HKUST

Yongqiang Xiong

Microsoft Research






    
=====================================================================================================================






The parameter server architecture is prevalently used for distributed deep
learning. Each worker machine in a such system trains the complete model,
which leads to a large amount of network data transfer between workers and
servers. We empirically observe that the data transfer has a major impact on
training time.

We present a new distributed training system called Stanza to tackle this
problem. Stanza exploits the fact that in many models such as convolution
neural networks, most data exchange is attributed to the fully connected
layers, while most computation is carried out in convolutional layers. Thus,
we propose layer separation in distributed training: most nodes of the cluster
train only the convolutional layers, while the rest train the fully connected
layers. Gradients and parameters of the fully connected layers no longer need
to be exchanged across the entire cluster, thereby substantially reducing the
data transfer volume. We implement Stanza on PyTorch and evaluate its
performance on Azure and EC2. Results show that Stanza accelerates training
significantly over current parameter server systems: on EC2 instances with
Tesla V100 GPU and 10Gb bandwidth for example, Stanza is 1.34x–13.9x faster
for common deep learning models.











§ INTRODUCTION


Deep learning (DL) has recently achieved prominent success in many
problem domains, particularly image classification 
<cit.> and speech recognition <cit.>. As DL models are made more complicated and data are procured from more sources at a faster pace, distributed training becomes increasingly common in large organizations.
Most distributed DL systems follow the parameter server architecture first proposed in <cit.> and later refined in <cit.>. 
A parameter server system has two types of machines: workers and servers. 
Workers perform training on local data shards they are assigned to, and send the gradients of the model parameters to the corresponding servers. 
Servers update the parameters based on aggregated gradients from all workers, and push the latest parameters back to them for the next iteration.

In current parameter server systems workers train the complete DL model, and
exchange the entire set of parameters with the servers in each iteration. This
is termed the data parallelism paradigm. Given the sheer size of DL
models with hundreds of millions of parameters
<cit.>,
this approach produces an substantial amount of network data transfer. For
example VGG-16 <cit.>, Inception-V3
<cit.>, and ResNet <cit.> have over
1100MB, 210MB, and 480MB data, respectively, that each worker has to exchange
with servers in each iteration. Even with 40G or 100G interconnects the
parameter exchange has a non-negligible impact on training time, especially
when modern GPUs with tens of TFLOPs computation capacity are deployed
<cit.>.


Most DL models, in particular convolutional neural networks (CNNs), are
composed of two basic building blocks: convolutional (CONV) layers and fully
connected (FC) layers. 
More interestingly, we find that CONV and FC layers exhibit distinct
characteristics (<ref>). CONV layers are essentially small filters
that extract features through convolution over the training data. 
Thus they usually have a very small number of parameters (<10% of the total),
but need a lot of computation (>90%). On the contrary, FC layers have full
connectivity between layers in order to perform high-level reasoning. 
They generally use most of the parameters in the model, but only require a small
amount of computation. 

Motivated by these observations, we propose to decouple the training of
CONV and FC layers instead of bundling them as in current parameter server
architecture (<ref>). 
The idea is simple: We use most machines in the cluster as convolutional
workers, or CONV workers, whose job is to train just the CONV layers that
are computationally demanding. A few remaining machines then act as FC
workers that train the FC layers and update their parameters.
The FC layer gradient and parameter exchange, which accounts for most of the
communication cost as discussed, now only happens among a few FC workers,
which reduces the amount of data transfer and training time considerably.  

To demonstrate the feasibility of our idea, we design and implement a new
distributed DL system called (<ref>). 's design addresses
two new challenges introduced by the separate training of models.

Communication Strategy.

In order for training (with stochastic gradient descent
<cit.>) to work, CONV workers now
need to send activations of the last CONV layer to FC workers to complete the
forward pass. Similarly FC workers need to push the gradients of the last CONV
layer to CONV workers for backpropagation. Further, without centralized
servers, CONV (resp. FC) workers need to exchange among themselves CONV (resp.
FC) layer gradients.

adopts hybrid communication strategies to minimize the cost of these
communication patterns (<ref>). For CONV-FC communication,
since the number of activations of the last CONV layer is very small compared
to the complete model, simply uses many-to-one and one-to-many
communication here. For the more expensive gradient exchange among CONV
(resp. FC) workers themselves, adopts efficient algorithms for the allreduce operation in the MPI literature <cit.> to
fully parallelize the duplex communication among nodes.
also overlaps the CONV worker and FC worker gradient exchange to further reduce the training time. 

Node Assignment.
The second design challenge is how to determine the node assignment in .
That is, how many nodes in the cluster should be used as CONV/FC workers. We
take a principled approach here (<ref>). First we develop
a performance model to characterize 's training throughput for a given
node assignment scheme based on the hybrid communication strategies. We then
cast the node assignment problem as an optimization program that aims to
maximize the
training performance given the total number of nodes, which can be solved
offline efficiently.

We implement on PyTorch <cit.>
(<ref>) and evaluate it with small-scale deployments on
Azure and AWS EC2 (<ref>). Our evaluation uses modern GPUs such as
Nvidia Tesla V100 and widely used CNNs in image classification including
AlexNet <cit.>, VGG-16 and VGG-19
<cit.>, Inception-V3 <cit.>, and
ResNet-152 <cit.>. We show that delivers significant
speedup in training and saves up to 10x data transfer compared to parameter
server systems. With 10 CONV workers and 1 FC worker for example,
provides 13.9x and 8.2x speedups for AlexNet and VGG-16, respectively,
using a 10G network. Even with 40G or 100G network bandwidth, our numerical
simulation in <ref> shows that still provides salient
benefits: it achieves 1.55x and 1.72x speedups for AlexNet and VGG-16,
respectively, with 100G bandwidth.




§ BACKGROUND 



We start by introducing background on deep learning and the parameter server
architecture.



 §.§ Deep Learning









Deep learning (DL) is a class of machine learning algorithms that uses a large
number of connected layers of different processing functions to handle complex
tasks. The layered structure of DL models composes a large artificial neural
network which resembles the biological structure of the human brain
<cit.>. The objective of DL is to find a model that minimizes
the difference between the inference result from the model and the ground
truth, which is usually represented by a loss function. This is effectively an
optimization problem and thus many optimization algorithms are used to
iteratively train the DL models.

Particularly, stochastic gradient descent (SGD) is widely used in DL
<cit.>. The
process consists of two phases with mini-batch SGD: forward pass and
backpropagation. In the forward pass of the t-th iteration, a batch of input
data D_t is fed to the model, and a loss value l(x, w_t) is computed as
the result for each sample x. Here w_t represents model parameters. Then
in backpropagation, the parameters are revised according to the loss so that
the model “learns” about the correct answers. Mathematically, the parameter
update rule is:

    w_t+1⟵ w_t - η/n∑_x ∈ D_t∇ l(x, w_t),

where η is the learning rate and n the batch size. The loss value
is calculated at the output layer of the neural network, and gradients are
generated from the output layer all the way back using the chain rule, so that
parameters at each layer can be updated.




 §.§ Parameter Server Systems




DL Training is usually done in a distributed setting with a cluster of
machines in order to cope with the increasingly large datasets and complex
models. The parameter server architecture has emerged as the de facto
solution for distributed training since its inception
<cit.>, and is implemented in
almost all DL frameworks <cit.>.



The parameter server (PS) system is a centralized design as shown in
Figure <ref>. There are two types of nodes, servers and workers. A server node maintains a partition of the globally shared model
parameters. Server nodes communicate with each other to replicate and/or to
migrate parameters for reliability and scaling.
A worker node performs training using data parallelism <cit.>, and
is responsible for one partition or shard of the training dataset. Each worker
goes through a mini-batch (or simply batch) of its local data shard in
parallel to compute gradients as discussed in <ref>, and push them to
the corresponding servers. Servers then aggregate the gradients from all
workers to perform a global update of the model, and disseminate the new
parameters to each worker. This completes one iteration of training. When
workers go through all the samples in their shards in multiple iterations, they
complete one epoch of training. Servers and workers are usually distinct
machines for better fault-tolerance and performance. 



The parameter server architecture provides implementation of basic engineering components of DL systems such as consistency models, fault-tolerance, and scalability, thereby enabling users to focus on application-specific code <cit.> cite more.
There is a scheduler node that assigns training tasks and data to workers and monitors their progress. 
If workers are added or removed, or upon worker failures, it reschedules unfinished tasks. There are two situations for the fault-tolerance when the worker is failure: (1) If the failure worker can recovery locally, so the worker restart training process locally; (2) If the failure worker cannot recovery locally or there is a new worker joining the training, the worker should pull the entir model prelica from the parameter server and get a partition of the training data from file system, then start the training process.
this needs a little more detail on say fault-tolerance












 §.§ Sizable Data Exchange in Training



Most DL models contain a large number of parameters in order to capture the
complex features of the input data and their impact on the prediction results.
Further, each worker needs to push the gradient of every parameter, and pull
every updated parameter from the servers in each iteration. As a result,
distributed training in a PS system entails a substantial amount of data
exchange between workers and servers. 

Table <ref> shows the number of parameters for common DL models
in image classification, and the corresponding data exchange volume per
iteration in a PS system with 4 workers and 1 server. 
All workers send gradients to and receive parameters from the server.
With tens or
hundreds of millions of parameters, the PS server has to transfer about 0.8GB
to 4.4GB data over the network. The bulky data transfer incurs non-negligible
overhead to training time performance. The impact can be considerable
especially as GPUs are now prevalently used to train large DL models. Owing to
the high memory bandwidth and massive number of cores, GPU accelerates the
matrix computation in DL tremendously, leaving the system more susceptible to
other overheads such as the network data transfer.







To demonstrate the problem, we implement a PS system on PyTorch
<cit.> and use an EC2 GPU cluster with the same setup above
(4 workers and 1 server). We use p3.2xlarge instances each with a
state-of-the-art Nvidia Tesla V100 datacenter GPU <cit.> and 10Gbps
bandwidth, and train the DL models listed in Table <ref>. More
details of the implementation and testbed setup can be found in the evaluation
section <ref>.
We measure the per iteration training time by inserting timestamps
before and after each iteration in the Python code. Note operations in PyTorch
are executed immediately when the Python statements are invoked due to its
dynamic graph design <cit.>; we do not add any extra
barriers. The results averaged over 100 iterations are shown in
Table <ref>. We also show the estimated communication time to
complete the data transfer at the single server, which is calculated simply by
dividing the data size by the 10G bandwidth. 








We measure the total training time and computation time in each iteration, and summarize the average results over 100 iterations in Table <ref>.
Execution is greedy w.r.t. dynamical graph design of PyTorch <cit.>. Operations are executed immediately when the Python statements are invoked. We put many time markers on some start points of processes, such like data loading, forwarding, synchronization and so on. We can get the time cost directly and accuracy.  
The computation time includes time to load images to the GPU and to compute gradients on each worker. 
The rest of training time is then spent on communication which is a period from the first gradients arriving at the server to last parameter sending to the worker. 


Observe from Table <ref> that the estimated communication time
takes ∼70%–80% of the measured training time for all models.

Note that actual impact of communication time may be less severe due to
various optimization techniques such as overlapping communication with
computation. Nonetheless, the results reflect that data transfer has a
critical impact on distributed training with GPUs in PS systems. Prior work
<cit.> has reported similar observations and the problem
has attracted increasing attention recently in both DL
<cit.> and systems <cit.>
communities.

The impact of data transfer certainly can be mitigated with higher network
bandwidth, which would require additional time and investment for the
infrastructure overhaul. On the other hand the problem may also aggravate with
the rapidly improving GPU or special-purpose hardware (ASIC, FPGA) that
slashes computation time. We therefore ask, is the massive data transfer
unavoidable in distributed training? Can we take a more fundamental approach
to minimize the data transfer and the training time in turn without affecting
the convergence or accuracy of the model?


Observe from Table <ref> that for all DL models, single worker needs to exchange hundreds of megabytes with the parameter server. For some models such as VGG-16 and VGG-19, the size of data even beyonds one thousand megabytes.
The results demonstrates empirically that large data exchange in the PS systems has the critical influence on the training time, owing to the massive update traffic generated by GPUs. The large data exchange indirectly proves that the cluster for distributed deep learning also need high capacity network.




Observe from Table <ref> that for all DL models, communication attributes to >80% of total training time per iteration. For some models such as AlexNet, VGG-16, and VGG-19, communication takes more than 90% of training time. 
The results demonstrates empirically that communication is the performance bottleneck in PS systems, owing to the massive update traffic generated by GPUs. 
In fact some prior work <cit.> has reported similar observations and the problem has attracted increasing attention recently in both DL <cit.> and systems <cit.> communities.

Note that some ML frameworks such as MXNet <cit.> designate
each node as both a worker and a server <cit.>. Each node is
responsible for 1/N of the model parameters. This can mitigate the bandwidth
bottleneck at the servers. However, the amount of parameters each node needs
to send and receive is still massive (4 (N-1)/N parameters), which does not
fundamentally overcome the issue. It is estimated that for such a system, the
network bandwidth has to be at least 26Gbps in order for it not to become the
bottleneck when training AlexNet on Titan X GPU <cit.>, which is slower
than the state-of-the-art GPUs now. Shi et al. <cit.> also report that
MXNet has relatively high communication overhead when scaling to multiple
nodes even with 56Gb Infiniband.



network communication is the performance bottleneck in distributed DL systems with GPU. 
The communication bottleneck is likely to aggravate as the capability of GPU or special-purpose hardware (ASIC, FPGA) is improving more rapidly than the network bandwidth evolution.
For example Tesla V100 which is used in our experiments before is released in 2017 for production DL workloads. 
It offers 14 teraFLOPs single precision <cit.> which is 1.5x faster than its predecessor P100 released a year earlier. 
The server interconnects and routers on the other hand are still running at 40G or 100G at best, which has been available on the market for many years. 
Efficient solutions are thus required to tame the amount of parameter exchange in current PS systems, without sacrificing convergence or model accuracy.







§ SEPARATING THE LAYERS




To answer our quest, we examine the characteristics of DL models, and propose
to separately train the layers in order to substantially reduce the data
transfer in distributed training. We focus on convolutional neural networks
(CNNs) which are arguably the most widely used class of DL models.




 §.§ Layers are Remarkably Different





CNNs have been successfully applied in image recognition
<cit.>,
video analysis <cit.>, natural language processing
<cit.>, drug discovery <cit.>, etc. A CNN
typically consists of two core building blocks, convolutional (CONV) layers
and fully connected (FC) layers
<cit.>.
A CONV layer has a set of learnable filters each of which is spatially small
(e.g. 5×5 along width and height). During the forward pass of training,
each filter is convolved across the input data to extract certain features at
some spatial position of the input. The neural network learns filters that
activate when they detect certain features (e.g. pointy ears, curled tails),
which are then passed on to the successive layers. Thus CONV layers are
usually positioned right after the input layer. After CONV layers, the
high-level reasoning is performed by the FC layers. Neurons in an FC layer
have connections to all outputs in the previous layer. FC layers employ
classifiers such as softmax to classify the output.

As a result of their functionality distinctions, CONV and FC layers exhibit
remarkably different characteristics. CONV layers have a small number of
parameters due to the small filters, but carry out a large amount of
convolution computations. On the contrary, FC layers usually have a very large
number of parameters due to its fully connected nature, but only require a
small amount of simple calculations. As shown in
Table <ref> for example, for AlexNet and VGG-16, CONV
layers incur most of the computations (>90%), while FC layers account for
most of the parameters (∼90%).[We count the number of
parameters which require gradients in the update. PyTorch provides API calls
to get this statistic. We cite the computation flops from <cit.>.]






 §.§ Separating CONV and FC Layers 


The distinction between CONV and FC layers presents salient opportunities for
us to optimize the communication cost of PS systems.

Specifically, it is now clear that much of the data transfer between workers
and servers in a PS system is for FC layer parameters and gradients. 
Since CONV layers require most computation, we can assign most of the machines
to train just the CONV layers, and the rest to train just the FC layers, as
shown in Figure <ref>. With the separation of the layers, in the
forward pass the CONV workers send the output of the last CONV layer, i.e. the
activations, to the FC workers. The FC workers in turn send gradients of
the last CONV layer back to CONV workers to continue backpropagation. FC layer
gradients are now irrelevant to CONV workers, and no long need to be pushed
over the network to each worker. As a result, a significant part of the
communication in the traditional PS system can be eliminated to accelerate
training. This is our key idea.



Training Process with Layer Separation.
Figure <ref> illustrates the training process with more detail.
Suppose there is one FC worker now since most computation concentrates on
CONV layers. In each iteration t, each CONV worker processes a sample up to
the last CONV layer, and sends the activations from the last CONV layer to the
FC worker. The FC worker collects all activations a^lc_t for a batch of
samples, and continues the forward pass to compute the loss values. It then
starts backpropagation and computes gradients for all FC layers and the last
CONV layer. It pushes gradients of the last CONV layer ∇ l(x, w^lc_t)
to the CONV workers, and updates FC layer parameters. CONV workers then
continue the backpropagation to compute their gradients in parallel, exchange
them among all CONV workers to get global information ∑_x ∈ D_t∇
l(x, w^conv_t) from all samples of the batch D_t, and perform CONV layer
parameter update accordingly. This completes one iteration of training. The
process is effectively equivalent to SGD in PS systems as explained in
<ref>.[Gradients from all CONV (resp. FC) workers are
disseminated to each CONV (resp. FC) worker.]



Note that a (max) pooling layer may follow the last CONV layer to reduce the number of activations. As discussed pooling layer does not have any parameters. Thus in this case we separate the model by the last pooling layer and include it in CONV workers to reduce communication between FC and CONV workers. This does not change the training process.

Benefits Overview.
Now compared to PS, we eliminate the FC layer parameter exchange at the expense of: (1) sending activations and gradients of the last CONV layer, and (2) exchanging CONV layer gradients among CONV workers.
First, comparing with the number of parameters of the entire model, the number
of activations and gradients of the last CONV layer is small.
For example, the number of activations from the last pooling layer in AlexNet is just 9216 for each sample <cit.>. With a typical batch size of 128, a CONV worker sends a total of ∼1.2M activations compared to 61.1M parameters for the entire model in each iteration.
Second, the number of CONV layer parameters (gradients as well) is much smaller than that of large FC layers as discussed in <ref>, and we can optimize the exchange with parallel communication to further reduce the time (<ref>).
Thus, training time can be significantly improved by separating the CONV and FC layers. 

Essentially, our idea of separating the training of CONV and FC layers, and using activations to replace gradients, can be regarded as a mixed use of both data parallelism and model parallelism in the DL literature <cit.>. 
Data parallelism—where each node trains the complete model with different data
partitions—is prevalent in current DL systems due to implementation
simplicity.
We also employ data parallelism among CONV workers which train the same CONV layers independently with different data partitions. 
At the same time, training of FC layers is done on a separate group of workers which is an example of model parallelism. 
Amid the discussion about the two paradigms in the community <cit.>, our hybrid approach represents a promising alternative to the common conception of using one in lieu of the other exclusively.





§ DESIGN



1. how to divide the layers? just conv/FC
2. how many workers for FC? just 1
3. how do the conv workers comm.? 


To demonstrate the feasibility of our idea, we design a new distributed DL
system called . With the separate training of CONV and FC layers, needs to address two new challenges in system design: First, how to design the
communication strategies between CONV and FC workers, and amongst CONV workers
themselves in the backpropagation phase? Particularly, the CONV workers now
need to synchronize parameters among themselves, and it is critical that minimizes such overhead. Second, how to assign the nodes to perform CONV or FC
training, in order to ensure that the system performance is optimized in terms
of training throughput?








 §.§ Hybrid Communication



utilizes hybrid communication strategies to address the first design challenge we just presented. 

For simplicity, we start the discussion assuming only one FC worker in the system. 
We discuss the case of multiple FC workers in <ref>. 
With one FC worker, communication in happens (1) between the CONV workers and the FC worker, and (2) among CONV workers themselves.



  §.§.§ CONV-FC Communication




Activations of CONV workers are different since they process different input data. 
Thus gradients are also distinct for each CONV worker and has to be sent to the corresponding CONV worker in order for backpropagation to work. 
It is thus not possible to combine the activations or gradients to reduce the transmission volume. 
The number of activations and gradients of the last CONV layer is very small (a few MBs with a batch size of 128) and does not consume much bandwidth, as explained in <ref>. 
Therefore, just uses simple many-to-one and one-to-many transmissions for CONV-FC communication as shown in Figure <ref>. This is also easy to implement.







  §.§.§ CONV Worker Communication



CONV worker communication is more demanding since most nodes in the cluster are CONV workers. 
The number of gradients of all CONV layers is also larger than activations: AlexNet for example has 2.3M CONV layer parameters compared to 1.2M activations with a batch size of 128. 
Moreover, performing the gradient exchange in an all-to-all manner generates a large volume of data transfer that scales quadratically with the number of workers, and is clearly not efficient.
Note that the objective of CONV worker communication is to disseminate the
global average gradients among all CONV workers, which is essentially an allreduce operation in MPI of parallel computing.
Therefore we adopt an efficient implementation of allreduce called the
recursive doubling algorithm  <cit.> to schedule the communication.[Other more efficient implementations <cit.> cannot be adapted here because of non-commutative operations on tensors in PyTorch (operation cannot be commutative because this could introduce different numerical errors on different workers).] 
In recursive doubling, neighboring workers exchange data in the first round. Then workers that are two nodes away from each other exchange the reduced data from the first round. Generally workers exchange cumulative data with nodes 2i-1 spots away in round i, as illustrated in Figure <ref>.
In this case it requires log_2 8, i.e. 3 rounds to complete allreduce
with 8 CONV workers.


If the number of CONV workers n_c is not a power of two, we first randomly pick n_c - 2^⌊log_2 n_c⌋ “surplus” nodes and have them send their gradients to another n_c - 2^⌊log_2 n_c⌋ nodes randomly in parallel.
Then recursive doubling can be applied on the 2^⌊log_2 n_c⌋ nodes without the surplus ones.
Finally the result is sent to surplus nodes in one round in parallel by n_c - 2^⌊log_2 n_c⌋ random nodes. 
In other words it takes ⌊log_2 n_c⌋ + 2 rounds to finish the allreduce.




  §.§.§ FC Worker Communication



Finally in the case of multiple FC workers, each is responsible for an equal
number of the CONV workers. Many-to-one and one-to-many communication
strategies are then used between the corresponding FC and CONV workers.
Each FC worker maintains a complete set of FC layers.
Among the FC workers, relies on the same recursive doubling algorithm
described above to exchange gradients and parameters efficiently. Further, FC
worker communication is overlapped with CONV worker communication to reduce
training time, since FC workers can start the parameter exchange after sending
gradients to CONV workers.




 §.§ Node Assignment




We now proceed to discuss how allocates the number of FC and CONV workers.  

There are many tradeoffs involved in node assignment. For example, the number of CONV workers determines the total batch size: more CONV workers enables more images to be trained in parallel at each iteration. At the same time, more activations have to be sent to the FC workers who also need to send back more gradients as well, and the communication among CONV workers also takes longer to complete. 

Thus we rely on a principled approach to tackle this challenge:  
First we develop a performance model to characterize the key tradeoffs and their impact on 's training throughput, given a particular node assignment scheme. 
Then the optimal node assignment is obtained by solving the throughput maximization problem given the total number of nodes. 




  §.§.§ A Performance Model



We consider a homogeneous setting where each node has identical hardware resources.
The key notations in our model are summarized in Table <ref>. 
Lower case notations denote unknowns and upper case notations are known constants.
For tractability, our model only captures the ideal case performance without considering various overheads in for example node synchronization and parallelization, which are rather difficult to model explicitly. 





We first characterize the computation time per iteration in t^c_s.
There are two components: (1) CONV computation time T_c, including time to compute activations in the forward pass, and time to compute CONV layer gradients in backpropagation. This term only depends on the per-node batch size and a node's GPU resources, and is constant with respect to node assignment since CONV workers work in parallel.
(2) FC computation time, including the time to finish the forward pass based on activations, and then generate gradients for the FC layers. 
Denote the FC computation time to handle activations from one CONV worker on one FC worker as T_f. 
It is also constant since T_f only depends on the per-node batch size and a node's GPU resources.
With n_c CONV workers, the FC computation time grows to n_c T_f due to the increase in activations, and with n_f FC workers it becomes n_c/n_f T_f due to parallel processing.
Taken everything together, we have:

    t^c_s = T_c + n_c/n_f T_f.


Now consider communication time in each iteration, which includes (1) CONV-FC communication time, (2) FC worker communication time, and (3) CONV worker communication time. 
Notice that (2) and (3) overlap in time as explained in <ref>.

Thus we do not consider FC worker communication here.

CONV workers send n_cAK activations of the last CONV layer to the FC workers, which in turns send back n_cAK gradients. Thus, CONV-FC communication takes 2n_cAK/n_fB.
CONV worker communications takes log_2 n_c rounds if n_c is a power of 2, and ⌊ log_2 n_c ⌋ +2 rounds if otherwise as explained in <ref>.
Each round takes P_c/B. 
Therefore, training time per iteration in , including communication and computation, can be written as:

    t_s =
    	
    	t_s^c + 2n_cAK/n_f B + P_c log_2 n_c/B,    if  n_c  is a power of 2,
    
    	t_s^c + 2n_cAK/n_f B + P_c (⌊ log_2 n_c ⌋ +2)/B,    otherwise.


The performance of a distributed DL system is characterized by training throughput, i.e. how many samples can be processed in unit time. 
We know that the total batch size is n_c K in , i.e. with n_c CONV workers it processes n_c K samples in one iteration. 
Hence, its training throughput can be expressed as:

    throughput_s(n_c, n_f) = n_c K/t_s.






  §.§.§ Node Assignment



Observe from the performance model that a larger n_c increases the total batch size, but also increases the computation time t_s and communication time. 
Node assignment then strives to maximize the throughput of the cluster given the total number of nodes N.
Mathematically, the optimal node assignment solves the following program:

    n_c,n_fmaximize       throughput_s(n_c,n_f) 
       subject to       N = n_c + n_f, 
              n_c > 0, n_f > 0.

The optimization program can be solved offline by an exhaustive search over all
possible values of n_c, as the total number of nodes is at most hundreds or
thousands in practice and the computational cost is very small. More discussion
on searching for the optimal deployment policy is in <ref>.
Note that for a given CNN model, the constants T_c and T_f depend on the node's GPU resources and can be profiled offline fairly accurately. 


 yet GPU memory also decides the batch size in the FC workers and CONV workers. For the CONV workers, We can still use the single-node batchsize, the GPU memory needed is less without FC layers. In the opposite, the GPU memory needed should be considered carefully. It depends on the model, the batchsize, the ratio of n_c and n_f. We use the GRAMneed to compute the GPU RAM needed in the FC workers.
In the , we can still use the single-node batchsize in CONV workers as what we did in <ref>, but the batchsize in FC workers depends on the N and n, which means the GRAM needed in FC workers is also variational. Figure <ref> gives the composition of the GPU memory needed in FC workers when the optimizer is momentum SGD. The memory in dashline rectangle depends on the models. While the memory in solidline rectangle varies from the input. The memory in squashed rectangle is the overhead brought by the framework. We realize the GRAMneed function to compute the GPU ram.


The node assignment should maximize the throughput of the cluster of . We assume the size of GPU RAM is GPUram. Thus,





 §.§ How to Train

Because adopts the hybrid data parallelism and model parallelism, the training process is a little different from the normal. As the figure <ref> shows, all CONV workers send the activation from the last layer to the FC worker, the FC worker stacks all the activation to a large batch. At this time, the FC worker should remember the source of the activation, which will be used in the backpropagation process. Then, the FC worker takes the large batch size input and computes the gradient for its model and the last layer of CONV worker. According to the source of the activation, the FC worker sends the gradient to CONV workers. The activation and the gradient link the two computation graphs in the FC worker and CONV workers as a complete graph, so trains the model as training process in a single node. What's more, doesn't need additional hyper-parameters.






§ IMPLEMENTATION



We implement based on the popular DL framework PyTorch <cit.> and its distributed communication package with TCP as the backend.
Our prototype has two main components: a controller that maintains the model and cluster configuration, and a communication library that can be used in PyTorch programs to handle parameter communication for both PS and . 
The software architecture is similar to prior implementation of distributed DL training <cit.>. 

 



 §.§ Controller



The client program in PyTorch first instantiates our prototype by creating a controller within its process, and passing information about the entire CNN model and hyperparameters (e.g. batch size) to it. 
The program also specifies which architecture is to be used for distributed training, PS or . 

For PS, the controller obtains the list of server and worker nodes, partitions the parameters equally across the servers by hashing, sends the mapping to each worker, and partitions the training data equally into data shards. 
It also sends the entire CNN model to workers to prepare for training.

For , the controller collects information about the throughput maximization program (<ref>) in <ref> such as available bandwidth, single worker CONV/FC computation times (T_c and T_f), etc., and solves the problem to compute the number of FC and CONV workers needed. 
It sends all FC layers and the last CONV layer to the FC workers, and all CONV layers to CONV workers. 
In case of multiple FC workers, the CONV workers are equally partitioned into n_f groups and the mapping is maintained and synchronized by each CONV worker.



 §.§ Communication Library



The communication library can be plugged into the training program. 
It provides APIs to support parameter communication with and PS as shown in Table <ref>. 
The push and pull methods are used by workers in PS to send gradients to servers, and acquire parameters from servers, respectively, based on the parameter mapping information. The pull method is called immediately after push, and blocks until it receives all parameter updates. The push method is nonblocking. 
The push and pull are implemented using PyTorch's send and receive primitives, respectively, with the widely used bulk synchronous parallel (BSP) model <cit.>.

For , the CONV workers invoke the push_activation method, which collects only activations from the last CONV layer for a batch of samples, then uses PyTorch's send primitive to perform many-to-one communication to the corresponding FC workers.  
They then immediately call the blocking pull_fc method to obtain gradients of the last CONV layer from the corresponding FC workers.
Finally, the pull_grad method is called by each CONV worker to perform gradient exchange with the recursive doubling algorithm discussed in <ref>. It uses PyTorch's all_reduce primitive. It blocks until the operation is finished for all CONV workers.
It is also called by each FC worker to exchange gradients when there are more than one FC worker.
The APIs for also apply BSP for model consistency. 





 §.§ Fault-Tolerance



For fault-tolerance, we use checkpointing throughout the system. 
Each node regularly creates checkpoints for current parameters and training state. 
In , there are very few FC workers. Thus a FC worker will create additional checkpoints for its FC parameters in a randomly chosen CONV worker. The redundant checkpoint is sent when both CONV and FC workers are training the model on GPUs, creating little overhead on training time.


 
Pipelining is also applied whenever possible to optimize training performance.
For example, in FC workers update the FC parameters after finishing sending the gradients to CONV workers. 
This overlaps the FC worker computation with CONV worker computation, thereby hiding the latency cost. 


 




 §.§ Others

Data serving. Network has already been the bottleneck of the distributed DL system, we don't want to bring more pressure in this architecture. We use the private file system instead of network file system (NFS), which needs users to move datasets to workers in advance. We use two different dataload methods for different disk I/O: (1) the dataload provided by the PyTroch, which needs high I/O speed. Even leading the time of data preparing as the bottleneck of training, it is not friendly to HDD; (2) Dataflow a package from Tensorpack <cit.>, the third-part package to efficiently prepare the data from low I/O speed disk, which is more suitable for HDD.  

Model Consistency. can realize pure model consistency. adopts the bulk synchronous parallel (BSP) strategy. In the communication between FC workers and CONV workers, the communication method abides the order to get the activations from each CONV workers, then start the rest computing processes. And the Sync function wraps the all_reduce method in the PyTorch, witch guarantees parameters in all CONV workers consistent. BSP can bring better convergence speed and performance. The pure model consistency can also bring the better reliable fault tolerance.








§ EVALUATION



We conduct testbed experiments on GPU clusters to assess the performance of the . 
We first verify its effectiveness with small datasets, by showing that achieves the same training accuracy as PS using 2x to 3.8x less time. 
Then we demonstrate the performance benefit of in production settings with large datasets and models, by showing that it achieves 1.34x to 13.9x speedup over PS, and the gain is more salient as the cluster grows.




 §.§ Setup


Testbed in Public Clouds. 
To understand 's performance in a realistic environment, we deploy our PyTorch-based prototype on GPU clusters from two cloud providers, Azure and EC2. 
(1) We use standard_NC6 VMs from Azure as our first cluster. 
Each node is equipped with 6-core vCPUs, 56GB RAM, and a Nvidia Tesla K80 GPU (half of a physical card). Nodes are interconnected with high bandwidth, which we find to be ∼2.6Gbps.
(2) For the EC2 cluster we use p3.2xlarge instances.
Each node has 8-core vCPUs, 61GB RAM, and a Nvidia Tesla V100 GPU with 16GB RAM. The network bandwidth across nodes is 10Gbps. 
All nodes run Ubuntu 16.04, Nvidia driver version 384.13, CUDA 9.0, cuDNN 7.0, and PyTorch 0.3.1. 

and PS always use the same number of nodes as CONV workers and workers, respectively, for fairness. 
uses 1 FC worker in all settings here. This does not imply that node
assignment in <ref> is not useful. Instead, one FC worker
is indeed optimal for throughput maximization when the cluster is smaller than
10 nodes in our testbeds.[We find that GPU memory of a single FC worker
is enough for
10 CONV workers.] For larger clusters we need to use multiple FC workers as
will be shown in <ref>. PS also uses 1 server for fairness
unless otherwise stated. Thus the total batch size is identical for the two
systems, and we focus on training time per iteration as the  performance
metric here.

Datasets. We use two image classification datasets that are widely used in prior work <cit.>. 
(1) CIFAR-10 <cit.>: It is a small dataset consisting of 60K
32×32 color images in 10 classes, with 50K training images and 10K
validation images. 
(2) ImageNet-12 <cit.>: This is a large dataset used in the annual
ImageNet Large Scale Visual Recognition Challenge (ILSVRC), with 1.28 million
training images and 50K validation images in 1K classes. 


 


Models and Hyperparameters.
We evaluate across many common CNNs for image classification. 

(1) AlexNet <cit.>: This is the model that won the ImageNet Challenge 2012 and subsequently spurred much attention on deep learning. AlexNet has only 8 layers, the first 5 are CONV layers, and the last 3 FC layers. 
(2) VGG-16 and VGG-19 <cit.>: Proposed in 2014, VGG models are deeper than AlexNet to extract more features from images. 
VGG-16 has 13 CONV layers and 3 FC layers, and VGG-19 has 3 more CONV layers. 
(3) Inception-V3 <cit.>: This is also a winner of the ImageNet Challenge in 2016. 
It uses the Inception module, a mini-model that replaces large convolutions to reduce the computations and parameters. 
(4) ResNet-152 <cit.>: This is a very deep CNN with 152 layers. This model adopts residual blocks which can ease the training of deep networks.

We list the model configurations in Table <ref>.
We use the momentum SGD <cit.> as the optimization algorithm in both and PS. 
Small weight decay and dropouts are used to get better performance. 






 §.§ Baseline Performance 



We first evaluate the baseline performance of using the small CIFAR-10 dataset and VGG-16. 

We modify the last softmax layer in the original VGG-16 in order to use CIFAR-10 with less classes, which reduces the number of parameters to 33.6M as shown in Table <ref>. 
Since the dataset and model are small, the experiments are done on the Azure cluster with less bandwidth. 

The same hyperparameters in Table <ref> are used.
Consistent with <cit.> we decrease the learning rate by 0.1 every 30 epochs. 
We train the model for 40 epochs.

Recall that as in <ref>, one epoch is one pass of the full training set and consists of many iterations during which each worker processes one batch of samples. 



Table <ref> shows the performance.  
effectively achieves the same (slightly better actually) accuracy of ∼90% compared to PS in all cases after 40 epochs. 
Further, brings significant savings in training time (at least 2x speedup), and the savings are more salient when the cluster grows. 
With 8 workers, obtains about 3.8x speedup compared to PS. 

We also note that as the cluster grows the total training time for PS barely improves. 
The main reason is that network bandwidth is the bottleneck here, and though adding workers reduces the number of iterations per epoch, the amount of gradient and parameter transfer to/from the server also increases proportionally in PS. The total amount of gradient and parameter exchange per iteration stays the same as a result.
We measure the time spent on computation in PS over 40 epochs, which is 0.47h and 0.25h for 4 and 8 workers, respectively. The numbers are reasonable, and they imply that the communication time is 3.29h and 3.27h with 4 and 8 workers, respectively.[
The results are in line with our settings in Table <ref>. For instance with 8 workers in PS, one epoch has 50000/8/128=48 iterations (the remainder samples are dropped). With 2.6Gbps bandwidth we can calculate the communication time to be 33.3M*4*2*8*48*40*8/1024/1024/1024/2.6/3600=3.25h.] 


Quick recap:
The experiments here verify 's effectiveness: It improves training time by up to 3.8x compared to PS without trading off training accuracy, in settings with limited bandwidth. 





 §.§ Performance for Large CNNs 




We now use large CNNs to evaluate with the ImageNet-12 dataset. 
The experiments are done on the EC2 cluster with Tesla V100 GPUs and 10Gb bandwidth, which represent typical production settings for training DL models.
One node is used as the server in PS and FC worker in , and up to 10 nodes are used as workers in PS and CONV workers in .

The same hyperparameters in the previous section are used here, and the learning rate is linearly scaled with the cluster size. 
Training lasts for 30 epochs and both PS and achieve same accuracy, and we omit the accuracy results for brevity.

We focus on three performance metrics now: 
(1) Speedup: This is the ratio between training times of PS and . 
Larger speedup indicates better training time improvements. 
(2) FC-Layer Data Transfer (FC-Data): This is the amount of data transfer needed (in MB) to update the FC layer parameters in each iteration. 
Since we use 1 FC worker, FC-Data for Stanza effectively includes the number of activations and corresponding gradients only. 
(3) Total Data Transfer (Total-Data): Total-Data is defined as the total amount of data transfer (in MB) across all nodes in the system in an epoch. 
We use different time granularities for Total-Data and FC-Data to provide more insights into the performance analysis.
Note that with the ImageNet-12 dataset an epoch has many iterations now. For example from Table <ref>, an epoch has 1251 iterations when we train AlexNet with 8 workers and a per-worker batch size of 128. 

Figure <ref> depicts the performance results with a varying number of (CONV) workers. 
We make two interesting observations here.
First consistently shows salient speedups from 1.34x to 13.9x over PS as shown in Figure <ref>. 
Among all models, AlexNet has the best speedup, and Inception-V3 and ResNet-152 the smallest. 
This is because Inception-V3 and ResNet-152 have proportionally more parameters in the CONV layers than other models, so the benefit of eliminating FC layer parameter exchange is not as substantial. 


Second, the speedup in general increases as the cluster size grows.
The main reason is that network is the bottleneck for PS even with 10Gb bandwidth, and adding workers to accelerate computation only slightly reduces the total training time. On the contrary, cuts the communication time significantly by replacing gradients with activations, and thus the benefit of parallelism from adding workers prevails. 


Figure <ref> and <ref> corroborate the above analysis. 
We only show the results for AlexNet and VGG-16 for brevity.
Figure <ref> shows that PS requires ∼100x larger data transfer than for FC layers, which is the main reason for 's training time improvement. 
FC-Data increases with more workers because in one iteration, the amount of FC layer data transfer depends on the total batch size and scales with the number of workers.
Figure <ref> shows the Total-Data comparison. 
Observe that PS's total data transfer per epoch is over 4x larger than 's for all cluster sizes (the figure is in log-scale). In other words, PS needs to perform data transfer that is at least 4 times larger than . 
Total-Data of PS does not change with number of workers because although FC-Data per iteration increases with more workers, the number of iterations decreases proportionally. Essentially Total-Data in PS only depends on the total number of samples.
Total-Data of increases slightly with more CONV workers because the CONV worker communication takes more rounds to complete based on the analysis in <ref>. 


Quick recap:
We demonstrate that in production settings with 10G bandwidth and state-of-the-art data center GPU, provides significant improvements in training time over PS, which increases as the cluster size grows. 




 §.§ PS with Two Servers

Now since in PS all the gradient and parameter exchange bottlenecked at the server, adding more servers can certainly alleviate the bandwidth bottleneck and improve training time. 
The cost of doing so is also high though, since servers do not contribute to the computation. 

Here we conduct experiments using two servers in PS and compare to with just one FC worker. 
The number of (CONV) workers for both systems is still the same. 
In other words PS uses one more node than to perform gradient and parameter exchange. 

Table <ref> shows that with 10 CONV workers, still achieves favorable speedups for all CNNs (up to 9.8x). 
Figure <ref> shows more speedup results for AlexNet and VGG-16 with varying number of (CONV) workers. 
Evidently the additional server improves PS's training time. 
is able to provide 3.8x to 10x speedup for AlexNet, and 3.2x to 5.8x speedup for VGG-16.






Quick recap:
We demonstrate that in production settings with 10G bandwidth and state-of-the-art data center GPU, provides significant improvements in training time over PS, which increases as the cluster size grows. It even outperforms PS with additional servers. 






§ NUMERICAL RESULTS





Generally, some research institutes and industrial institutes deploy their missions on cloud providers, which is easy to scale the cluster of training. We survey the top-2 cloud providers in the world, AWS and Azure. The table <ref> shows bandwidth resources of the common VMs in deep learning. Because of the distributed training, we show the VMs just with one GPU card. We find the bandwidth of p3.2xlarge is up to 10Gbps which is the burst bandwidth. In our evaluation part, we adopt the NC6 and p3.xlarge, which presents the typical VMs in the distributed training.



Our evaluation so far covers small and medium clusters with typical GPU and bandwidth settings in public clouds. 
In reality, companies may use larger clusters for training with huge data. 
Also some clusters may deploy faster networks (40GbE, 100GbE, or RDMA) <cit.>.  
Yet it is difficult for us to access these hardware resources especially in a large scale. 

Therefore in this section, we numerically evaluate in large-scale settings based on the empirical data from testbed experiments in <ref> and the performance model in <ref>. 
We first develop a performance model for PS systems, similar to the one for , to estimate the training time and throughput. 
Then we verify the fidelity of the two models with experiments in public clouds. 
Finally, we estimate the throughput of both systems in large-scale settings using the performance models. 




 §.§ Performance Models and Verification



The performance model for is explained in detail in <ref>. 
Here we develop a similar model for PS. 
Denote the computation time to process a batch of samples on one worker, including the forward pass and backpropagation, as T_ps.
With n_s servers, each is responsible for an equal share of P/n_s parameters. 
The communication time in each iteration composes of (1) each server receiving P/n_s gradients from each worker, which takes n_w P/n_sB time, and then (2) each server sending P/n_s parameters to each worker, which takes n_w P/n_sB as well. Thus, training time per iteration is

    t_ps = 2n_wP/n_sB + T_ps.

Training throughput is then

    throughput_ps(n_w, n_s) = n_w K/t_ps.



We measure T_ps in PS and the single worker computation times T_c and T_f for in our EC2 GPU cluster, and feed them to the performance models to estimate the per-iteration training time. 
The model estimation is compared against the empirical data to verify their fidelity.
We use AlexNet on ImageNet-12 with the same setup in <ref> here. 
Figure <ref> shows the results with varying number of (CONV) workers. 
Observe that our models are fairly accurate. 
Comparing to experimental results, our model in (<ref>) estimates the per-iteration training time of with ∼5% errors, and Equation (<ref>) estimates PS training time with ∼20% errors. 
The errors may be attributed to an array of factors our models do not capture, such as fluctuating network bandwidth, memory copy time between CPU and GPU, etc.
More importantly, they are acceptable for our purpose here because our model underestimates PS's training time, and thus serves as a conservative lower bound for the speedup comparison against . 









 §.§ Numerical Simulation



We now proceed to estimate the throughput performance of and PS in large-scale clusters.
We use AlexNet, VGG-16, and VGG-19 on ImageNet-12, and obtain empirical data from our EC2 experiments for the model constants T_ps, T_c and T_f. 
They reflect the computation performance on ImageNet-12 with Tesla V100 GPU, the fastest data center GPU presently.   
We assume each node has a single GPU. 
We vary the network bandwidth from 25G to 40G and 100G.

and PS use the same number of nodes in total, which varies from 5 to 80. 
applies the node assignment algorithm in <ref> to
determine the optimal number of FC workers in each setting, which can be
larger than one especially for large clusters.[For CONV workers, we
adopt the batchsizes used in the original papers of the models. Because a
large batchsize may cause GPU memory overflow, we also consider the capacity
of GPU memory in node assignment.] Similarly PS also uses the model in
<ref> to obtain the throughput-maximizing node assignment in
each case. For this reason the two systems may have different total batch
sizes, and thus we focus on throughput instead of training time as the
performance metric.


The number of FC workers are 1, 1, 2, 4, 4 separately. 
The ratio of workers and servers in PS is 4:1, i.e. from 1 to 16.


Figure <ref> shows the numerical results.
We make several observations.
First, still outperforms PS significantly in a large cluster with fast network. With 80 nodes and 100G bandwidth for AlexNet, the throughput improvement is ∼1.4x.
Second, as the network bandwidth increase, communication becomes less of a problem, both systems have the performance improvement. At 25G bandwidth, training AlexNet, VGG-16, and VGG-19 can be improved by ∼2.29x, ∼2.35x, and ∼2.33x, respectively using with 40 nodes.
  

Figure <ref> shows the numerical results. 
We make several observations.
First, still outperforms PS significantly in a large cluster with fast network. With 80 nodes and 40G bandwidth for example, the throughput improvement is ∼1.79x for AlexNet, ∼1.97x for VGG-16, and ∼1.87x for VGG-19.
Second, as the network bandwidth increases, communication becomes less of a problem, and 's improvement decays. 
At 25G bandwidth, training AlexNet, VGG-16, and VGG-19 can be improved by ∼2.29x, ∼2.35x, and ∼2.33x, respectively using with 40 nodes.
When the bandwidth is 100G, 's benefit decreases to ∼1.55x, ∼1.72x, and ∼1.84x, respectively for the three CNNs. 

Third, enjoys better scalability compared to PS. 
With 100G bandwidth, delivers 2x throughput improvement for AlexNet (resp. VGG-19) when the number of nodes increases from 40 to 80; PS only achieves 1.85x (resp. 1.73x) improvement in the same scenario. 

Lastly, a related observation is that PS's scalability largely depends on the network bandwidth.
Without enough bandwidth PS scales poorly due to again the communication bottleneck. 
For example with 25G bandwidth, PS has 1.73x, 1.81x, and 1.70x throughput benefits from 40 nodes to 80 for AlexNet, VGG-16, and VGG-19, respectively.
With 100G bandwidth, the gains are better at 1.85x, 1.91x, and 1.73x, respectively. on the other hand consistently delivers 2x throughput scalability even with 25G bandwidth for all three CNNs.


Quick recap:
Our numerical study shows that achieves ∼2x better throughput over PS in large-scale clusters with up to 100G bandwidth. also has better scalability due to its ability to remove the communication bottleneck.




§ DISCUSSION




We discuss several concerns one may have about .


GPU memory.
Our node assignment optimization program can be extended to consider GPU memory constraints. 
GPU memory limits the amount of data that can be processed in a batch and affects performance.
When data is too large, they have to be loaded to the GPU in multiple passes with additional latency. It is not related to GPU memory.check this

Therefore for our node assignment problem, when the number of nodes is large, one may need to ensure that not too sure about the problem


Model decomposition. 
We separate the DL model at the boundary of CONV and FC layers, which is intuitive and easy to implement. 
Other strategies of model decomposition <cit.> can be considered to further re-distribute computation. 
For example multiple decompositions can be realized by using additional machine groups to train the specific model partitions. 
The performance model may also be modified correspondingly for optimal node assignment. 


 

We use the GRAMneed to presents the GPU memory needed for the FC worker. The requirement is that GRAMneed is smaller than GPU memory(GPUram).


We have introduced a solution to alleviate the communication bottleneck based on the features of layers and CNNs. The experiment shows that is intuitive but efficient.
 However, we find that large deep learning models can be divided into more components. In big data system, there are a lot of schedulers <cit.> to make the system more efficient. Their scheduling strategies are based of resource requirement of different tasks. What's more, Monotasks <cit.> even schedules the task just need one type resource. The cost metrics of component can be treated as the resource requirement in a sense. That means we can also treat the compents as tasks to schedule them. And Couplet is surely simple but efficient, but dividing the models into componets maybe achieve more efficiency. Resources provided by the cluster are not always regular for training the whole models. So, training the models in compoents looks like more flexible and resource friendly. In the feature, we will analysis how to build the components resonable and schedule to train the compoents more efficiently in large-scale distributed DL. We think components scheduling is a promise point to improve the training efficieny when training large neural netowrk in the cluster.  



Beyond Parameter Server.
Recently, to cope with the scalability issue of PS systems, new 
communication strategies for distributed training have been proposed and
deployed in some
cases.

Uber proposes Horovod <cit.> which uses ring-reduce
communication. Baidu's internal system uses optimized allreduce communication 
<cit.>. This is orthogonal to our approach of
reducing data transfer by layer separation, because both systems still rely on
data parallelism with each node going through the complete model. For example
we can use Horovod on the CONV workers to accelerate their communication.
also applies to these non-PS systems to further improve training time,
though we leave it as future work the implementation and evaluation of such an
extension.

also adopts the allreduce communication strategy for CONV workers communication. Both ring-reduce and allreduce are from the Message Passing Interface (MPI).
There are two reasons can explain why doesn't completely use allreduce among all nodes.
First, the challenge for the MPI is it can not work well in the asynchronous environment, for example, both Horovod and Baidu's system are running in synchronous environment. In the contrast, it is easy to convert from synchronous environment to asynchronous environment. just needs one more node to be a parameter server which only asynchronously communicates with any Conv node. We give the details in section <ref>.
Second, the nature of reduce algorithm is to reduce the data through merging the data with same index among different nodes, while the activation among different nodes cannot be merged, so the characteristic determines that cannot use allreduce between FC worker and CONV worker.


Beyond 10G Network Bandwidth.
High bandwidth networks at 40G or 100G are deployed in some private data
centers to speed up distributed training. However, 10Gbps network is the
mainstream in public clouds and most private DL clusters. Clearly 's gain
is less substantial with higher bandwidth since the impact of data transfer is
smaller. Yet as we have numerically shown in <ref>, still
provides over 55% gain over the current PS systems with 100G bandwidth. As
GPUs are improving at a rapid pace, we believe is instrumental for many
deployment scenarios.



Beyond CNNs.
We have focused on CNNs in this paper. The idea of layer separation works for
many other DL models, where nodes exchange only the activations of the
boundary layer instead of the full set of model parameters.





For example, recurrent neural networks (RNNs)
<cit.> have recently received much attention
with many applications
<cit.>. Generally, an
RNN model consists of cells each with the same multilayer perceptron (MLP)
models, which have multiple FC layers without CONV layers. To show that our
idea is also beneficial for RNNs, we consider the MLP in the RNN cells without
dependency. We conduct a simple experiment, where an MLP is separated after
the first 2 hidden layers, and we use just one node to train the last hidden
layer and the output layer while the rest of nodes to train the first two
hidden layers. As Table <ref> shows, layer separation achieves 3.1x to
4.2x speedups over PS under the same resource and hyperparameter settings
without sacrificing the accuracy. We plan to investigate in future work how to
deal with the dependency between RNN cells in order to fully extend to
RNN and other DL models.


The complex RNN models also have a huge number of parameters, which also slows the training process of distributed RNN training. Generally, there are multiple FC layers in RNN cell. However, some RNNs cooperate with convolution layers for a better performance, which means that RNNs contains FC layers and CONV layers, such like DeepSpeeck<cit.>. Because we have presented the model with FC and CONV layers before, so we only discuss the RNN only contains FC layers, such like RNN models in <cit.>. Multiple FC layers are also called multilayer perceptron(MLP) which is important in RNNs and occupies most parameters in RNNs. Because can reduce the communication footprint in distributed MLP training, can also be applied on distributed RNNs training.
First, we simulate the MLP structure without internal dependency of RNNs. We separate the MLP at the middle position between tow layers, then use to train the MLP with 100 epochs in different deployment settings. We use the CIFAR-10 as the dataset. In table<ref>, comparing with the PS, can achieve four times speed up in the same settings while achieving the similar accuracy. It shows that has the potential to speed up the training process of distributed RNNs training.
There is a strong internal dependency in the RNNs. The following cell, a unit in the RNNs, needs the middle output from the previous cell. In other words, the MLP in the following cell needs the specified activations from the MLP in last cell. This is the reason why we cannot directly apply in training RNNs. Solving the dependency constrains and speeding the RNNs training will be a future work for us and . 






response 6.1. Cost for Searching Deployment Policy
Training a complex model is expensive, it can spend weeks or days time. For example, Google released Bert<cit.>, a language representation model, which takes weeks to train the model on the TPU cluster. So, it is urgent to speed the training period. Comparing with the long training period, it only takes little time to pre-tuning the deployment policy for the training in . There are two part cost. First, the may need to sample the training time for the separated model. This cost depends on how to separate the model, which is flexible, but it is not too long. In our experiment, we find that the ration of parameters to activations is the index. The index is larger, the more communication cost is saving. The second part is to compute the optimal deployment, which is a result from equation <ref> and stable. The pre-tuning phase is short, which doesn't have on really impact on the whole training cost.











§ RELATED WORK




We survey related work in this section.

Parameter Server Architecture.
Parameter server has attracted much attention in both academia and industry
since Distbelief <cit.>. Li et al. <cit.> propose
a general PS system design that supports flexible model consistency, elastic
scalability, and fault tolerance. Its variants and extensions have since been
widely deployed. Examples include popular DL frameworks such as Tensorflow
<cit.> and MXNet <cit.>.

Several approaches have been proposed to deal with the communication problem
in distributed training. We discuss some important ones other than those mentioned in
<ref> now.

Communication Strategies.

Xie et al. <cit.> propose sufficient factor broadcasting that transmits
vectors to fully re-construct the parameter matrix. The amount of data
transfer thus scales linearly instead of quadratically with the dimensions of
the parameter matrix. Poseidon <cit.> exploits layered structures of DL
models to overlap communication and computation and hide the communication
cost. Both strategies can be applied in to further reduce the
communication bottleneck.

Another approach is to use decentralized SGD algorithm <cit.>
instead of centralized SGD. Nodes can be arranged in a ring topology. Each
node only communicates with its neighbors instead of the entire cluster to
exchange gradients and parameters at each iteration. This approach certainly
speeds up the training time per iteration. Yet it demands multiple iterations
to populate the parameters across the entire cluster, and the convergence
speed may degrade in the end in practice.

Synchronization.
Current PS systems generally use Bulk Synchronous Parallel (BSP)
<cit.> to ensure model consistency among workers. BSP
tends to prolong training time when stragglers are present. Asynchronous
Parallel (ASP) and Stale Synchronous Parallel (SSP)
<cit.> alleviates the impact of
stragglers by removing the barriers in BSP. Yet they
deteriorate the convergence speed and model performance due to the staled
parameter information <cit.>. How to achieve a good
tradeoff between convergence and synchronization is still an open problem.
adopts BSP in order to achieve fast convergence and better performance. 
It can also be readily extended to use ASP or SSP.

Gradient Compression.
Another interesting approach is to compress the gradients and parameters. 

Gradient sparsification exploits the fact that the parameter matrix is sparse
and some weights are small. Aji et al. <cit.> propose to
truncate the insignificant gradients and only transmit the larger ones. Hsieh
et al. <cit.> use a similar method that only sends important
gradients to servers in a WAN setting. Gradient quantization reduces the
number of bits to represent gradients and parameters <cit.>. CNTK
<cit.> reduces the size of gradients via 1-bit quantization,
which performs fairly well in speech recognition <cit.>. TernGrad
<cit.> uses 2-bit quantization and is shown to have little
accuracy loss.
is orthogonal to these efforts, and can use them to further alleviate the
communication cost.




§ CONCLUSION AND FUTURE WORK


We have presented the design and implementation of , a
communication-efficient DL system for distributed training.
utilizes the unique characteristics of CNNs to separate the training of
CONV and FC layers. FC layer parameter exchange is limited to among a few FC
workers as a result, and the bulky data transfer across the network in
conventional parameter server systems is largely removed. Testbed experiments
in Azure and EC2 show that improves training time substantially over
parameter server systems with latest datacenter GPUs and 10G bandwidth.
Numerical studies also indicate that still provides modest speedup even
with 100G bandwidth in large-scale clusters.

In this work we decouple the DL model at the boundary of CONV and FC layers,
which is intuitive and easy to implement. As future work, other strategies of
model decomposition <cit.> can be considered to further
re-distribute computation. For example multiple decompositions can be realized
by using additional machine groups to train the specific model partitions. The
performance model may also be modified correspondingly for optimal node
assignment. Other than layer based model decomposition, it is also possible to
use more fine-grained model decomposition. For example it has been shown that
scheduling the different operations of DL training to different computing
devices can also improve training speed, though finding the optimal device
placement itself is computationally expensive <cit.>.
Extending to non-PS systems and general DL models is also a promising
direction of future work as detailed in <ref>.






Furthermore, we use BSP to ensure model consistency among workers in . 
However, the distributed training has the small probability to face the straggler which will dwarf the training speed. As a future work, we want to implement ASP in the , which can reduce the impact brought by the straggler. What's more, ASP can further reduce communication pressure in .
It is convenient to convert from BSP to ASP. We should add one more node in . This node is responsible for asynchronously updating the parameters in CONV workers. In ASP pattern, there is no inter communication among CONV workers. Each worker should send the activation to the FC worker and update parameters with the new added node.
 






abbrv



