Near-Linear Scaling Data Parallel Training with Overlapping-Aware Gradient Compression

In default DP training, each worker launches collective operations (e.g., AllReduce) immediately to send their local gradients after the computations of all gradients are finished. The total training time for one iteration in DP becomes:

where $T_{before}$ is the time duration before the backward pass, including data input and forward pass. $T_{comp}$ is the computation time of backward pass. $T_{comm}$ is the communication overhead. In this paper, we focus on homogeneous clusters, so $T_{DP}$ is the same for every worker.

The multi-node training time of optimal DP (i.e., time of linear scaling DP) denoted by $T_{DP-LS}$ is equivalent to single-node training time without communications (i.e., $T_{comm}=0$ in (1)). Then, the speedup of DP (compared with training locally using a single device) can be represented as: $P*T_{DP-LS}\textfractionsolidus T_{DP}$, where $P$ is the number of workers. The speedup of linear scaling is equal to $P$, which is an upper limit.

Let the communication-to-computation ratio ($CCR$) be the ratio of communication overhead and computation time (i.e., $T_{comm}=CCR*T_{comp}$). Since $CCR$ is generally greater than 1 because of the communication bottleneck, the speedup of DP can be expressed as:

Overlapping aims to parallelize computations with communications in DP. Wait-free Back-propagation [21] is the theoretical basis of Overlapping. For more efficient use of the network bandwidth, merging small tensors of one layer’s gradients to a large tensor including several layers’ gradients is introduced to implement better Overlapping [2, 14, 20]. Once a tensor’s gradient is calculated, the communication operation is called on the entire tensor. Equation (3) shows the total training time for one iteration of DP in tensor-based Overlapping:

where b is the number of tensors, $T_{compute}^{i}$ and $T_{comm}^{i}$ are the computation time and communication overhead of the $i_{th}$ tensor. In case of shorter communication in DP, we introduce the concept bubble in (3), where ${bubble}^{i}$ represents the idle time of communication when the computation time of the ${i+1}_{th}$ tensor is longer than the communication time of the $i_{th}$ tensor, as shown in Fig. 1(d). Since the network speed is relatively slow in most cases (i.e., $CCR$$\geq$1), communication operations usually happen back to back, and the ${bubble}^{i}$ will not appear.

In the more common case where computation time is less than communication time, (3) can be simplified to:

where $T_{comm}^{{}^{\prime}}$ represents the time of the part of communication that cannot be overlapped with computation, as Fig. 1(b) shows. Since the communication overhead equals to $CCR*\sum_{i=1}^{n}T_{comp}^{i}$ and Overlapping can parallelize the communication overhead at most equal to the computation time, $T_{comm}^{{}^{\prime}}$ can be approximated as $(CCR-1)*\sum_{i=1}^{n}T_{comp}^{i}$ , which means once $CCR$ is much higher than 1, $T_{comm}^{{}^{\prime}}$ will be large and $T_{ovlp}$ will be much longer than $T_{DP-LS}$. The experiments in Table I further confirmed that the speedups provided by Overlapping (denoted by $S_{ovlp}$) are negatively correlated to $CCR$ and much less than the speedups of linear scaling.

Existent GC schemes often suffer from non-negligible compression overhead. When using GC, the training time for one iteration in DP becomes:

where $T_{compress}$ is the compression overhead (including compression and decompression), and $T_{comm-GC}$ is the communication overhead compressed by GC. In Table II, we observe that the DP training time reductions in some GC schemes are impaired by their high compression overhead. For example, although Top-k reduces communication overhead by about 600ms in each iteration of training VGG-19, it has a significant compression overhead of 1560ms. In contrast, PowerSGD costs less training time by 733ms with respect to communication overhead reduction by about 753ms and only 20ms compression overhead. Therefore, training the same DNN using PowerSGD is several times faster than Top-k.

For additional training time reduction, we apply Overlapping into GC with the training time of one iteration denoted by (6), which is derived from (4) and (5):

where $T_{compress}^{i}$ is the compression overhead of the $i_{th}$ tensor, $T_{comm-GC}^{{}^{\prime}}$ represents the time of the part of compressed communication that cannot be overlapped with computation. As discussed in Section II.B, $T_{comm-GC}^{{}^{\prime}}$ can be approximated as $(CCR-1)*\sum_{i=1}^{n}(T_{comp}^{i}+T_{compress}^{i})$. Since using GC results in much lower $CCR$, $T_{comm-GC}^{{}^{\prime}}$ is much smaller and even near zero. Unfortunately, data dependency in some GC schemes [11, 13, 16, 26] makes (6) not available. We choose 2 GC schemes with no data dependency: Random-k and FP16, to apply GC and Overlapping concurrently. Results in Table III illustrate that reducing the $CCR$ of DP to around 1 by GC achieves near-linear scaling for DP training with Overlapping.

From Section II.B, we know that the high $CCR$ of DP makes Overlapping difficult to achieve linear scaling. Although GC is an effective technique to reduce $CCR$, applying GC into Overlapping raises new challenges: high compression overhead and additional data dependency. Compression overhead generally comes from high time complexity operators in GC schemes, and there is a trade-off between compression overhead and training accuracy when compression ratio is high. However, we observe that in common network with 30-100Gbps bandwidth, the $CCR$s of common DP tasks are not too high, as shown in Table I. Therefore, it is possible for us to redesign a novel GC scheme with lower compression overhead.

Table I also shows that $CCR$ may differ significantly in different DP tasks. To achieve linear scaling, using a constant compression ratio may not be applicable to all DP tasks. $CCR$ depends on many factors, such as network bandwidth, DNN type, communication library, and worker number. It is also a challenge to obtain accurate $CCR$ with cost-tolerant profiling.

In Gradient Compression, the filter represents the gradient selection strategy and often appears in sparsification schemes. Existing fine-grained filters (e.g., Top-k) select gradients according to each iteration’s overall gradient value distribution. However, such strategies often suffer from non-negligible compression overhead, as shown in Table II. Instead, we design a novel coarse-grained gradient filter to decrease time complexity. COVAP considers the implementations of DL frameworks and uses communication unit tensor as the granularity of filters. In practice, our filter discards the complete communication operations of some tensors in each DP iteration.

COVAP’s coarse-grained filter can significantly reduce the time complexity. Tensors for communication often include several consecutive layers of DNN, which implies that the number of communication tensors is much lower than the number of gradients. For example, the default communication tensor size in PyTorch is 25MB. For DNNs of 50 million parameters with a size of about 200MB, PyTorch allocates about 8 communication tensors to contain all gradients (here, we assume the size of each layer is balanced). Since COVAP’s filter only needs to select tensors from those 8 tensors instead of traversing all 50 million gradients, the time complexity is significantly reduced. Another benefit is that from the implementation perspective, using the same granularity of communication operations with DL frameworks will not introduce any additional overhead of rebuilding tensors in each iteration. Empirically, the tensor size is set to 25MB, the same as the default value in Pytorch. Less or greater sizes may result in performance degradation, as presented in [14].

In such coarse-grained filter, the tensor selection strategy is also different from other GC schemes. In each iteration, each worker chooses tensors with same index to communicate with others. Tensors are selected alternately, which means each tensor is only communicated once in every $I$ iterations ($I$ represents the interval). Meanwhile, only one tensor is chosen for communication from every $I$ tensors in each iteration. Tensor $t$ is selected in iteration $num\_steps$ when $(t+num\_steps)\%I=0$. Fig. 2(a) illustrates the above selection strategy. When the interval $I$ is 4, the first tensor is communicated at the first and $5^{th}$ iterations, the second tensor is communicated at the second and $6^{th}$ iterations, and so on. One benefit of such a strategy is that each tensor is communicated and updated by the same interval, which can alleviate the negative effect of staleness [30] on training convergence. Another benefit is that it does not introduce any additional data dependencies mentioned in Section I because each worker can select tensors by themselves according to the current iteration numbers and the interval (The interval is determined in the early stage of training and remains unchanged in subsequent training, we give a detailed discussion in section III.B.), which means no additional communication is required to synchronize the tensor selection results.

The coarse-grained filter has minimized compression overhead. However, for various training tasks in different computing environments, a constant compression ratio is difficult to achieve fully overlapping consistently. In other words, $CCR$s are different in different cases. For example, using different GPU types can significantly influence the $CCR$: replacing the GPU from V100 to A100 will speed up the computation and increase $CCR$. Therefore, to achieve full overlapping in most cases, COVAP adopts different compression ratios for different cases. Note that in this paper, we do not consider the interference from other tasks in the cluster or other performance fluctuations, which means $CCR$ will not change rapidly throughout whole training.

From the analysis in Section II.D, COVAP should decrease communication time by $CCR$ times. Since COVAP uses the same collective communication primitives (i.e., AllReduce) as the baseline of no compression, decreasing communication overhead by $CCR$ times is equivalent to reducing communication data volume by $CCR$ times. Because communication tensors of DL frameworks are generally in fixed sizes [14, 20], COVAP initially assumes each tensor has the same size. Then, reducing data volume by $CCR$ times can be approximated as reducing the number of communicating tensors by $CCR$ times. We can infer from the algorithm in Section III.A that the interval $I$ in our scheme should equal $CCR$.

COVAP measures the $CCR$ in the early stage of the training task and sets its compression ratio according to the result. For implementation details, COVAP measures the $CCR$s of different DNNs based on the PyTorch profiler module. PyTorch profiler’s context manager API can be used to study device kernel activity and visualize the execution trace. Specifically, we use the profiler_cuda module to track cuda events, including communication (i.e., AllReduce and AllGather) and computation (i.e., forward and backward propagation) operators. Since we only need to profile one training iteration to obtain the communication and computation time, COVAP only costs a one-off profiling overhead (less than 5 seconds in our experiments). Compared to hours of training, such profiling overhead is acceptable.

The profiler tool provided by PyTorch is applicable for stand-alone programs. In distributed training, PyTorch creates multi-processes for different workers. Although we can use the original profiler to monitor one of the processes and obtain its exact time of computation events in one iteration, the communication time measured in this way may be inaccurate. That is because there may be a slight time difference in distributed training when different workers start their communications on different nodes, as Fig. 3 shows. In such cases, the communication operators of some workers who finish their computations early will be longer than others since they must wait for other workers to rendezvous in communication. In our experiments, such waiting may cause a 20% communication time measurement error. Therefore, we develop a distributed profiler. The key point is to align the timeline at the end of each communication operator in each training step to eliminate measurement errors caused by such communication waiting latency.

Using that distributed profiler, we can obtain more accurate communication time to compute $CCR$. Then, COVAP sets the interval $I$ of the filter according to that $CCR$. Since $I$ must be an integer but measured $CCR$s may not be integers, we let $I$ equals to $\lceil CCR\rceil$, which implies that COVAP compresses communication by a little more than $CCR$ times to ensure as much communication as possible can be overlapped with computation in DP.

In Section III.B, we assumed that each communication tensor has the same size as the default fixed size of DL frameworks. However, the practical tensor sizes may vary significantly according to severe imbalanced sizes of DNN layers, which could cause communication bottlenecks if the large tensors are selected to communicate in COVAP. In practice, DL frameworks allocate gradients of layers into tensors. The gradient tensor of one layer is used as the minimum unit in allocations, which means each tensor contains integral number of layers and at least one. Even if a layer’s size is much larger than the tensor size of the framework, PyTorch will not split such a large variable into multiple tensors. Although DL frameworks often adopt a much larger default tensor size than the size of most layers in DNNs, there may still exist some much larger layers than the default tensor size in some DNNs. For example, we list some of the layers of VGG-19 in Table IV. It can be seen that the parameter number of Layer FC1 is much larger than other layers, accounting for 71.53% of all parameters. The size of Layer FC1 is 102760448 * 4 Bytes = 401.4 MB. Compared with the default tensor size of 25MB in PyTorch DDP, the tensor of that layer is oversized. We also tested the communication time of all tensors when training VGG-19 across 8 nodes, and the result is shown in Table V. The communication of that large tensor costs about 603.238ms, accounting for 72.67% of total communication time. Such prominent communications of large layers may impair the Overlapping advantage in COVAP due to the imbalance, as shown in Fig. 4(b).

Therefore, COVAP designs a tensor sharding strategy that slices the large tensor into small pieces such that all tensors are balanced in size. Specifically, after building tensors, COVAP counts the elements’ number of all tensors to find the median number of elements in one tensor. Suppose a tensor has multiple times elements than that median number. In that case, COVAP evenly slices it into $\lfloor\frac{numel}{median}\rfloor$ parts, where $numel$ represents the elements’ number of that tensor (i.e., the total number of gradients of all layers in that tensor). Note that if $\lfloor\frac{numel}{median}\rfloor$ is larger than interval $I$, COVAP only slices that tensor into $I$ parts for balance.

Then, when COVAP selects tensors to communicate, such a large tensor is regarded as $\lfloor\frac{numel}{median}\rfloor$ tensors, from which COVAP may choose one tensor to communicate in one iteration. Still using VGG-19 as an example, as shown in Table V, the median element number of tensors is 5590260. Thus, the second and third tensors are sliced into 3 and 19 new tensors (if the interval $I$ is less than 3 or 19, COVAP slices them into $I$ parts instead). Then, the total number of tensors is regarded as 26 when COVAP chooses tensors for communication, which results in a much better balance of tensor sizes. Another example of how COVAP selects tensors when using tensor sharding is shown in Fig. 2(b), and how tensor sharding balances the communication time is shown in Fig. 4(c).

$\mathbf{The}$ $\mathbf{error}$ $\mathbf{feedback}$ $\mathbf{scheduler}$. Although COVAP significantly improves training speed with techniques in Section III.A, III.B and III.C, such compression schemes are theoretically lossy in training accuracy. To alleviate such a problem, COVAP equips error feedback (i.e., local memory accumulation) to ensure convergence and designs a new scheduler to continuously adjust a compensation coefficient of Error feedback in training, which balances the feedback amount of errors.

The idea of error feedback was first introduced in one-bit [19]. Following GC works [3, 7, 11, 13, 16, 24, 26] widely adopted error feedback as a compensation strategy to ensure convergence and preserve accuracy. Several works [23, 24] also conducted theoretical analyses. Generally, error feedback is used as shown in Algorithm 1, where $residuals$ represent the D-values saved in local memory.

In COVAP, we only choose a few tensors to communicate in each iteration. Applying error feedback to COVAP means we save other tensors’ gradients in local memory and add them to their corresponding tensor in the next iteration, just like Algorithm 1. However, in practice, we found that using error feedback in such a way may lead to divergence due to the harmful effect of the staleness of delayed update parameters. To alleviate the staleness effect, we introduce the error feedback scheduler into Algorithm 1 based on the observation in [10] that a large compensation coefficient in early training epochs may harm model accuracy.

The error feedback scheduler multiplies residuals by a compensation coefficient before the residuals are added to current gradients (i.e., line 2 in Algorithm 1). Similar to the learning rate scheduler of variant declining strategies, our scheduler adjusts that coefficient continuously with increasing strategy in training. The coefficient is equal to:

where $init\_value$ represents the initial value of the coefficient, $num\_steps$ represents the number of iterations trained, $ascend\_steps$ represents the interval between ascending, and $ascend\_range$ represents the ascending range. The max value of the coefficient is 1.

$\mathbf{Convergence}$ $\mathbf{analysis}$. From Section III.A, the filter of COVAP is similar to sparsification schemes since COVAP also selects a subset of all gradients in each iteration. Similar to Top-k[3], COVAP can be regarded as a particular case of Random-k [23]. Therefore, we can follow the theoretical proof in [23].

We first give the definition of COVAP’s compression operator:
$\mathbf{Definition}$ $\mathbf{1.}$ $For$ $1\leq l\leq h,$ $where$ $h$ $is$ $the$ $depth$ $of$ $network$ $\mathbf{x},$ $the$ $compression$ $operator$ $COVAP$ $is$ $defined$ $for$ $\mathbf{x}\in\mathbb{R}^{d}$ $as$

$where$ $num\_steps$ $is$ $current$ $number$ $of$ $training$ $iterations$ $and$ $I$ $is$ $the$ $interval$ $set$ $by$ $COVAP.$

Same as Random-k and Top-k, operator $COVAP$ also satisfies the definition of being a k-contraction in [23]:

Then, we utilize the convergence proof process for Top-k and Random-k in the non-convex case presented in the work of [23] to prove the convergence of COVAP.

We conducted our experiments on 8 Alibaba Cloud ECS instances. Each instance contains an Intel Xeon Platinum 8163 CPU and 8 NVIDIA V100 GPUs with 16 GB global memory. We choose a typical bandwidth of 30Gbps in the public cloud. In High-Performance Computing, the bandwidth is usually much higher and reaches 100Gbps. Due to the adaptive compression ratio relying on $CCR$, we believe COVAP can also provide acceleration in such scenarios.

We use four neural networks from different deep learning domains summarized in Table VI for evaluation. For VGG-19 and ResNet-101, we use SGD optimizer with an initial learning rate of 1e-3; for Bert and GPT-2, we use Adam optimizer with initial an learning rate of 5e-5 and 1.5e-4. We compare our scheme COVAP with the state-of-the-art and other popular GC schemes in Table II. For a fair comparison, all schemes are implemented by the DDP communication hook in PyTorch 1.9.0 using NCCL as the communication library.

Most GC implementations are referred to [28]. Note that Ok-topk uses mpi4py as its communication library. We reimplemented it in our environment, which may result in a different performance than reported in [13].

We verify the correctness of the adaptive compression ratio selection strategy introduced in Section III.B on three different DNNs. Fig. 5 shows the performance of COVAP in using different compression ratios. We tested the speedups provided by COVAP in all cases. The speedups were calculated by (1) and (2) such that the upper limit speedup of DP training with 64 GPUs is 64, as dotted lines shown in Fig. 5. Compression ratio 1 represents the baseline without compression (i.e., default PyTorch DDP).

As discussed in Section III.B, COVAP chooses $\lceil CCR\rceil$ as its compression ratio. Using a higher compression ratio than $\lceil CCR\rceil$ is meaningless because COVAP has already reduced most communication overhead through Overlapping. In Fig. 5(a), the speedup of training ResNet-101 increased very slowly as the compression ratio was beyond 3. Similarly, Fig. 5(b) and 5(c) discover the maximum speedups were 51.51 and 54.55 when the compression ratio was 4, which COVAP selected.

We studied the training time and model convergence using real-world applications listed in Table VI. In addition to listing the results of each complete training, we make a further breakdown of the training time of one iteration for better understanding, including compression (e.g., Top-k selection from the gradients), communication (e.g., AllReduce), and computation (i.e., forward and backward pass). The experiments were conducted in an 8-node cluster with a 30Gbps network, each node with 8 NVIDIA V100 GPUs. Note that we use average values for breakdowns since some GC schemes’ training speed varies significantly at different stages of training. For linear scaling, the speedup should be 64 (compared with one GPU), which is also an upper limit of all schemes. Besides, since Random-k diverged in most experiments, we do not give the convergence result of Random-k. Table VII presents the training time and accuracy results of different GC schemes when training 4 DNNs, wherein the speedup is calculated by (2) in Section II.A. Fig. 6 presents the time-to-solution curves of 3 DNNs except for Bert, since we only use the title of each news in THUC-News for text classification and its training time is relatively short. DDPovlp represents the default PyTorch with Overlapping and is the baseline with no gradient compression.

We compared COVAP’s scalability with other GC schemes using 3 DNNs. The experiments were conducted on four clusters of 8, 16, 32 and 64 GPUs. Since we did not need to focus on training accuracy here, we used Cifar-10 as the dataset for ResNet-101 and VGG-19. The result is shown in Fig. 11, where the linear scaling bar represents the upper limit speedup of DP.

Fig. 11(a) shows that COVAP obtained speedups of only 5% less than linear scaling under four different clusters in ResNet-101 training, compared with 30% less of PowerSGD. Besides, AllReduce-based GC schemes showed no degradation as the cluster size increased since AllReduce has better scalability than AllGather. For example, AllGather-based Random-k and EFsignSGD obtained only 3.2x acceleration while the GPU number increased by 8x. COVAP scaled even better than other AllReduce-based schemes because it essentially reduced the number of communication operations in each iteration. Another reason is that COVAP adjusted its compression ratio adaptively according to the change of $CCR$ while other GC schemes kept their compression ratios (or other hyperparameters) unchanged. Specifically, COVAP outperformed other schemes by 1.15x-9.03x on 64 GPUs, compared with 1.04x-3.02x on 8 GPUs.

Fig. 11(b) and Fig. 11(c) show the scalabilities of all schemes in training VGG-19 and Bert. Note that we could not scale Top-k, Random-k, DGC, EFsignSGD, and Ok-topk beyond 16 GPUs in training VGG-19 since AllGather requires more memory than AllReduce when cluster scaled and caused running out of memory. The comparison of Fig. 11(a) and 11(b) disclosed that the 3x longer communication overhead of VGG-19 than ResNet-101 degraded its scalabilities of all schemes, while COVAP preserved the best stability in all cases along with cluster scaling.