In a distributed deep learning system based on mobile service computing, there may be differences in performance between computing nodes, which may be affected by the external environment, resulting in training interruption or low convergence speed. Therefore, in this study, we propose a strategy to dynamically adjust the number of tasks between computing nodes for a scenario in which node performance changes lead to the slow training of distributed deep learning systems. Based on this, we designed a distributed parallel computing strategy called weight-based load balancing (WLBS). The WLBS divides and adjusts the task volume of computing nodes through initial division and partial adjustment; thus, computing nodes with lagging performance can exert their maximum computing power under the premise of reducing the lag impact on the entire system training. Experiments show that in distributed deep learning model training, the WLBS can effectively reduce the model training time while ensuring the accuracy of the model.
Research Article
Dynamic Distribution Strategy of Distributed Tasks Based on Limited Synchronous Parallel
https://doi.org/10.21203/rs.3.rs-1768291/v1
This work is licensed under a CC BY 4.0 License
Version 1
posted
You are reading this latest preprint version
Distributed System
Deep Learning
Parameter Server
Load Balancing
With the rapid growth of data types and scales in human society and the advent of the era of big data, the use of machine learning algorithms to process large-scale data has become the core scenario of many modern application services [1, 2]. However, for data from terabytes to petabytes and onwards to zettabytes and for training complex machine learning models at large scales, it often takes days or weeks to execute using a single compute node [3]. The computing power of a single computing node can no longer satisfy the timeliness and accuracy requirements of complex machine learning model training [4]. Therefore, distributed machine-learning systems have gradually become a research hotspot for solving large-scale machine-learning problems in the era of big data [5]. To adapt to the complexity of big data, improve the training accuracy of distributed machine learning, and reduce computing time in model training, industry and academia have developed a variety of distributed machine learning systems [6, 7].
In essence, the process of machine learning training involves abstracting a problem into a mathematical model, and it is necessary to design an index and learning algorithm to evaluate the model. Through repeated iterative calculations, the computer can calculate the parameters of the mathematical model in an optimal situation to obtain a mathematical model to solve the problem [8, 9]. In the process of solving the model, all datasets were trained once through the neural network, which is defined as an epoch. To use data resources more efficiently, it is often necessary to propagate the complete dataset in the same neural network multiple times, that is multiple epochs are required. Because not all data sets are passed through the neural network at one time, but the dataset is divided into multiple batches, the calculation size of one pass through the neural network is defined as the batch size. Distributed machine learning is applied in the following three situations: the training data size is too large, model size is too large, and computational load is too large[10].
Datasets are large-scale; for a situation where the training data has exceeded the tolerance of a single node, training is performed in a data-parallel manner [11]. The data-parallel method first divides the overall dataset into multiple subsets and then distributes them to the computing nodes participating in the training in the distributed cluster. Finally, the computing nodes perform iterative training on the allocated subsets, aggregate the training model results at the end of each round of training, and consider the aggregated model parameters as the model parameters of the next iterative training.
For a situation in which the model scale exceeds the computing scale of a single node, model parallelism [12] is used for training. A common method is to divide the mathematical model into several sub-models, where some of the model parameters correspond to a sub-model, and then assign the sub-models to computing nodes for training.
Excessive number of calculations: This situation is caused by the first two. Data parallelism or model parallelism can be implemented by multithreading or parallel computing [13] based on shared or virtual memory.
The distributed machine learning framework mainly includes a data domain model partition module, stand-alone optimization module, communication module, and data and model aggregation module. Figure 1.1 shows the distributed machine learning system framework [10].The current mainstream structure of distributed machine learning comprises parameter servers and computing nodes. Compared to traditional distributed computing, more attention has been paid to the synchronization of parameters and gradients. Communication is the biggest bottleneck in the training of distributed machine-learning systems [14-16]. Many scholars have proposed various solutions for distributed machine learning communication mechanisms. The classic mainstream synchronization strategies include bulk synchronous parallel (BSP) [17, 18], asynchronous parallel (AP) [19-21], stable synchronous parallel (SSP) [22], dynamic synchronous parallel (DSP) [23], and limited synchronous parallel (LSP) [24].
The above strategies have their advantages in different scenarios, but they cannot dynamically adapt to scenarios in which the performance of computing nodes in the cluster changes. Therefore, in this study, we propose a resource allocation method for dynamic cluster computer points, which can be divided into initial and partial adjustments.
The initial division was mainly used before model training. According to the basic performance indicators (such as the number of CPUs, number of cores, and memory size) of the computing nodes in the cluster, the load of each computing node was calculated, and the computing nodes entered the next stage according to the data volume of the initial division in one round of iterative training.
Part of the adjustment is mainly used in scenarios in which the performance of the computing nodes is severely degraded during the model training process and cannot be recovered in a short time between the computing nodes with the best and worst performance. It underwent multiple partial adjustments in iterations until each node satisfied the optimal amount of training data.
The process of partial adjustment is divided into large adjustments and small adjustments:
(1) Major adjustment: The number of node training samples is adjusted until the training time for each computing node to complete all the samples tends to be the same.
(2) Small adjustment: Adjust the batch size of each round of calculation until the time for each round of calculation parameters of each computing node tends to be the same.
Through the investigation and research on the status quo of distributed machine learning systems globally, in this study, we analyze the advantages and disadvantages of distributed parallel computing models, including the BSP, SSP, and LSP models. The main contributions of this study are as follows.
- To address the shortcomings of the aforementioned static strategies, a task volume was proposed, the dynamic allocation strategy.
- This strategy makes each computing node close to the optimal workload ratio through initial division and partial adjustment and can fully use its computing resources, thereby speeding up the training speed of the model.
- The feasibility of this strategy is verified by theoretical analysis.
The remainder of this paper is organized as follows. The design motivation of the model is described in Section 2, and the proposed model is described in detail in Section 3. In Section 4, we discuss the experimental setup and follow-up benchmark results, and conclude our work in Section 5.
Distributed machine learning systems have become the mainstream solution to the most challenging problems to adapt to the complexity of big data and reduce the application computing time [25, 26]. Distributed machine learning has achieved considerable research results worldwide. Academia and industry have implemented multiple distributed machine-learning systems based on multiple parallel strategies and parameter communication consistency models. The following is a brief analysis of parallel strategies and parameter communication consistency models in distributed machine learning.
|
FRAME |
TensorFlow |
Torch |
Caffe |
Parallel |
CNTK |
|---|---|---|---|---|---|
|
SOURCE |
|
|
Ph.D.Jia |
Baidu |
Microsoft |
|
SUP- LAN |
C++, Python |
Python |
C++, Python |
C++, Python |
C++ |
|
HARDWARE |
CPU, GPU |
||||
|
DISTRIBUTE |
Support |
Support |
Not Support |
Support |
Support |
|
P-STRATEGY |
Model and Data Parallel |
||||
The BSP [17, 18] introduces a global synchronization barrier (GSB). In each round of training, all computing nodes must enter GSB before starting the next round of iterative training. Because each round of computation needs to wait for all nodes to enter the GSB, the computation speed of each round depends on the computing node with the slowest training speed. Therefore, the main problem with the overall synchronous parallel strategy was the lag problem. The lag problem refers to the phenomenon in which computing nodes with poor performance slow down the overall running speed owing to differences in the performance of the cluster nodes. In a real production environment, the scale of computing nodes and the time it takes for computing nodes to complete iterative training have dynamic variability, which makes the lag problem more serious and causes the training performance of the BSP model to drop significantly.
To solve the problems existing in the BSP model, some scholars have proposed an asynchronous parallel (AP)[19–21] strategy for distributed machine learning. The advantage of the AP strategy is that the computing node can use local model parameters to execute the next iteration before receiving the global model parameters. There are lag problems owing to unbalanced cluster loads, which significantly reduce the time cost of model training. However, its shortcomings are also evident. Because the training of each computing node is not globally synchronized, the consistency of the model parameters cannot be guaranteed; therefore, the model is likely to fall into a local optimum and eventually fail to converge.
The SSP [22] strategy introduces a delay threshold. During the training process, when the difference between the number of iterations of the fastest computing node and the number of iterations of the slowest computing node exceeded the delay threshold, the fast computing node entered the GSB and waited for all nodes to complete the training before entering the next round of iterative training. SSP allows each computing node to use local model parameters during iterative training and strictly controls the number of times that the nodes use local model parameters for iterative training. The SSP strategy combines synchronization and asynchrony, which significantly reduces the lag time of computing nodes on the premise of ensuring model convergence. However, the ability of the SSP strategy to balance the cluster load is fixed, and it cannot adapt well to dynamic changes in the cluster node performance in a real production environment, resulting in the inability to guarantee the accuracy of the model.
Based on the SSP strategy, the DSP [23] strategy introduces a weak threshold. As shown in Fig. 5, when a weak threshold is met, GSB should be entered. The DSP strategy relaxes the conditions for entering global synchronization. The purpose is to prevent a situation when the performance difference is not large; that is, it is difficult for the difference in the number of iterations between fast and slow nodes. Additionally, the strategy can dynamically adjust the delay threshold according to the conditions of entering the GSB. If the barrier is entered by satisfying the weak threshold condition many times, the value of the delay threshold should be reduced and the delay threshold should be increased. However, the fault-tolerance properties of the iterative-convergent algorithm are limited, that is, when the performance difference is large, the DSP model cannot completely solve the lag problem.
Finally, in the traditional parallel communication strategy, BSP is the representative of the synchronous communication strategy, which ensures that the nodes maintain strong consistency of the model during the training process, but the training speed of each round will be affected by the nodes whose performance lags and depends on the slowest compute node. The AP is a representative asynchronous communication strategy. The computing nodes of the cluster using this strategy do not need to wait for computing nodes with lagging performance during the training process. Therefore, the impact of the lagging nodes on the training time of the entire cluster was less than that of the BSP strategy. However, because no synchronization barrier is set, the consistency of the model during training cannot be guaranteed; therefore, the model is likely to fall into a local optimum and eventually fail to converge. The SSP strategy combines synchronization and asynchrony. This is representative of a strategy that combines synchronization and asynchrony. The computing nodes of the cluster adopting this strategy enter the synchronization barrier after performing limited asynchronous calculations to ensure that the model does not fall into local optimality, the premise of increasing the model training speed. However, this asynchronous fault tolerance of SSP is limited and cannot adapt well to dynamic changes in cluster node performance in real production environments. Moreover, none of the above strategies can adapt to dynamic changes in the cluster environment, such as network anomalies, hardware failures, and other problems that may cause training interruptions and do not have good robustness. Section 3 introduces a strategy for solving these problems.
In a traditional distributed machine learning training system, the number of computing tasks of each computing node is statically allocated, that is, divided by equal amounts before node training. Aiming at the bottleneck of static allocation of LSP computing tasks, in this study, we propose a weight-based load-balancing strategy (WLBS). The core idea of the WLBS strategy is to calculate the ratio of the optimal computing task volume to the total computing task volume of each computing node in the cluster according to the performance of the computing nodes. Thereafter, the computing tasks are divided according to the ratio, and finally, the effect of node load balancing in the cluster is achieved. In the WLBS strategy, the number of computing tasks of each computing node can be adjusted in the later stage according to the computing performance; thus, computing nodes with high computing performance can train more data while reducing the amount of training data for nodes with lagging performance to achieve the load of the entire system balanced. This section discusses the three aspects of the WLBS strategy design, algorithm implementation, and related theoretical proof.
3.1 Limited Synchronous Parallel Strategy
Conversely, the LSP strategy alleviates the synchronization lag problem of the BSP strategy; however, it ensures the accuracy of model training and improves the training efficiency of distributed machine learning models. This section discusses the existing problems in the static scheduling process of the LSP strategy. LSP is based on the synchronization barrier, and proposes the concept of the limited synchronization barrier (LSB) and a limited threshold l, that is, when the current l computing nodes complete one iteration, the l computing nodes perform limited synchronization. The LSP strategy reduces the asynchronous defect to a certain extent, but it may still write the old local model to the parameter server. Another disadvantage is that the LSP strategy cannot adapt well to the dynamic changes of cluster node performance in the real environment, resulting in its inability to ensure accuracy.
The following combines the analysis of the limited synchronous parallel strategy based on static scheduling, and then provides a formal description. Suppose there are k computing nodes in a distributed cluster, \({u}_{k,t}\) is the model update increment of the k-th node in the t-round synchronization, and the state of the k-th node in the t-round synchronization is recorded as \({\tilde{\omega }}_{k,t}\), that is, the state of the initial model after t-round synchronization. l is the difference between the maximum number of iterations between the fast and slow nodes, that is, the limited threshold, where l∈[1,K], and the synchronization process that satisfies this condition is called limited synchronization. l is the difference in the maximum number of iterations between fast and slow nodes, that is, the limited threshold, where l ∈ [1, k]. The synchronization process satisfying this condition is called limited synchronization. When the number of limited synchronizations reaches m, all nodes enter the global synchronization process; that is, the number of limited synchronizations between two adjacent global synchronizations shall not be greater than m. The LSP model can then be described as:
$${\tilde{\omega }}_{k,t}={\omega }_{0}+\left[\sum _{{t}^{\text{'}}}^{t-m-1}\sum _{{k}^{\text{'}}}^{l}{u}_{{k}^{\text{'}},{t}^{\text{'}}}\right]+\left[\sum _{({k}^{\text{'}},{t}^{\text{'}})\in {u}_{k,t}}{u}_{{k}^{\text{'}},{t}^{\text{'}}}\right]$$
where \({\tilde{\omega }}_{k,t}\) represents the current model parameters,\({u}_{k,t}\) represents the updated subset of all computing nodes during the [t-m, t + m-1] iteration, and \({\tilde{\omega }}_{k,t}\) represents the current model parameters. \(\left[\sum _{{t}^{{\prime }}}^{t-m-1}\sum _{{k}^{{\prime }}}^{l}{u}_{{k}^{{\prime }},{t}^{{\prime }}}\right]\) represents the global update set of the previous local iteration, and \(\left[\sum _{({k}^{{\prime }},{t}^{{\prime }})\in {u}_{k,t}}{u}_{{k}^{{\prime }},{t}^{{\prime }}}\right]\) represents the update of this local iteration.
The limited synchronous parallel strategy has the following characteristics. The proportion of time occupied by the training of computing nodes increases, which avoids frequent synchronization waiting to a certain extent. This supports the composition of computing nodes with performance differences over time. This difference disappears with the recovery of computing node performance within a period of time; c. The delay threshold can be changed, but it cannot fundamentally solve the "delay" problem caused by stale model parameters for training.
In distributed systems, problems, such as network anomalies and hardware failures are frequently encountered. These problems cause differences in the training speeds of the computing nodes. To simplify the description, the differences in the training speeds of these computing nodes are called performance differences. The distributed machine learning system based on the limited synchronous parallel strategy can be applied to scenarios in which the computing nodes have performance differences over time. In a limited time, through the limited threshold l, one node with a faster training speed can perform limited synchronization first. After several limited synchronizations, the global synchronization threshold m can force all computing nodes with faster computing to stop training and wait for computing nodes with a lagging performance to perform global parameter synchronization. In an actual distributed cluster scenario, the time for which the performance difference between computing nodes exists is unpredictable. Moreover, as the scale of the distributed cluster increases, the possibility of a performance failure of one of the computing nodes is higher, and the recovery time of such a failure cannot be determined. In the limited synchronization parallel strategy, it is assumed that the faulty node in the cluster takes a long time to recover performance. Thereafter, when the number of limited synchronizations of the entire system reaches the threshold m, the global synchronization will still be affected by the faulty node, resulting in a distributed machine learning model that depends on the slowest compute node. Under more extreme conditions, the faulty node is down, and the iterative training is stopped. When the global synchronization conditions are met, the entire distributed training will stop running, which seriously affects the efficiency and results of distributed training and wastes a lot of computing resources.
As shown in Fig. 6, assume that in a statically allocated cluster with a total of five computing nodes participating in training, the global synchronization threshold m = 3. When the limited number of synchronizations in the entire system reaches 3, the global synchronization of parameters is forced. The statically allocated cluster means that in a distributed cluster, the number of computing tasks of each computing node is pre-allocated, that is, the amount of training data of each computing node is the same. Assuming an extreme situation, before entering the GSB, computing node 5 goes down and stops computing tasks; then, other computing nodes will be stuck in the synchronization waiting process until the performance of computing node 5 recovers.
3.2 Strategy Design
The LSP model for the static allocation of computing tasks cannot solve the long-standing performance difference problem in distributed systems. In this section, the core idea of the weight-based load balancing strategy WLBS is to determine the optimal amount of training data for each computing node and to recalculate and adjust within a certain period to finally achieve the effect of load balancing. The WLBS strategy recalculates and adjusts the optimal training task ratio of each node in a certain period, and adopts the LSP communication strategy for training in a certain period. Therefore, there are certain restrictions on the cluster environment conditions, and it is necessary to assume that the computing nodes in the cluster remain relatively stable during training in this period.
The key to the WLBS strategy is to determine the optimal amount of training data for each computing node and how to set the recalculation-adjustment period. Under the condition that the task training volume in a certain period is maintained unchanged, the training data volume of some computing nodes is dynamically adjusted according to the average time-consuming iterative training of each computing node in the previous cycle, until the optimal training data volume ratio is obtained. The dynamic allocation process was divided into initial and partial adjustments.
The initial division was mainly used before model training. According to the basic performance indicators (such as the number of CPUs, number of cores, and memory size) of the computing nodes in the cluster, the performance indicators of all computing nodes were aggregated into a performance indicator ratio array, and then the array was used in the initial division of the training data. To ensure that the total training data are certain, the load of each computing node is calculated according to the performance index ratio array, and then the computing node enters the next round of iterative training according to the initial divided data volume. Partial adjustment is mainly used in scenarios in which the performance of computing nodes is severely degraded during model training and cannot be recovered in a short time. Part or all of the training data of the node with the highest lagging performance is transferred to the computing node with the best performance. Tuning is performed only between the best-performing compute node and the worst-performing compute node in the previous cycle (the time between two partial tunings is defined as the partial tuning cycle). It underwent multiple partial adjustments in iterations until each node satisfied the optimal amount of training data. The partial adjustment process is divided into large and small adjustments.
(1) Major adjustment: The number of training samples of a node is adjusted, and the corresponding training time is one epoch. After multiple major adjustments, the training time for each computing node to complete all the samples tends to be the same.
(2) Small adjustment: The batch size is adjusted for each round of calculation. The corresponding training time was the time taken for one batch. After several small adjustments, the time required for each round of parameter calculation for each computing node tends to be the same.
After several partial adjustments, because the training time for each computing node to complete all the samples and the time for calculating the parameters in each round tend to be the same, the training time of each computing node tends to be stable. Because the training time for completing a batch between each computing node tends to be the same, the problem of being stuck in synchronization waiting owing to changes in the computing performance of the computing nodes is solved. When the computing node satisfies the condition that the time difference of any computing node to complete a batch is less than or equal to \({\theta }_{b}\), the small adjustment process of partial adjustment will be stopped, and when the computing node satisfies the requirement that the time-consuming difference of any computing node to complete an epoch is less than or equal to \({\theta }_{e}\), the large adjustment process of part of the adjustment will be stopped. The conditional expression is as follows:
$$\left|{bTime}_{i}-{bTime}_{j}\right|\le {\theta }_{b},{\theta }_{e}, i\ne j,1\le i,j\le N$$
Here, \(b{Time}_{i}\) represents the training time of the ith batch (single iteration) or epoch, and N represents the number of computing nodes. \({\theta }_{b}\) represents the time precision coefficient of batch (single iteration) training and \({\theta }_{e}\) represents the time precision coefficient of epoch training.
Considering the LSP communication model as an example, it is assumed that there are five computing nodes in the distributed system participating in task training. As shown in Fig. 2.2, before partial adjustment, the computing performance of nodes 1 and 3 is better, thus resulting in limited synchronization waiting. The computing performance of computing nodes 2 and 5 is relatively lagging, which may be caused by an unstable network, preemption of computing resources by other applications, hardware failure, and others. By reducing the computing task volume of node 5 and increasing the computing task volume of node 1, the difference in time-consuming calculation iterations between the two is gradually reduced. Simultaneously, computing node 2 may not perform training owing to downtime and other reasons. After partial adjustment, its computing tasks are distributed to other nodes such that the robustness of the distributed system is not affected. In addition, after several partial adjustments, except for some faulty nodes, the training speeds of other computing nodes are similar, thereby improving the global consistency of computing to a certain extent; that is, the size of the limited threshold is close to that of the computing nodes in the distributed system. Difference between the total number of nodes that cannot continue to participate in task training owing to failure.
3.3 Algorithm Implementation
For the strategy design in Section 3.2, this section elaborates on the WLBS strategy from the perspective of algorithm implementation. The WLBS strategy is designed based on the parameter-server architecture, and the nodes are divided into computing and parameter server nodes. The communication between the computing node and the parameter server node is realized using PYTORCH's distributed communication GLOO back-end framework primitives, the point-to-point TENSOR object communication is realized through the IRECV and ISEND primitives, and the parameter server is used to realize the multi-node or non-TENSOR object communication through the GATHER primitive point-to-point communication, and there is no communication between computing nodes.
First, the pseudo code of the WLBS strategy calculation node is given:
The implementation steps of the above strategy are as follows:
(1) Load the training data of the node.
(2) If the current epoch number is less than or equal to the predetermined epoch number, the epoch training is entered; otherwise, if the epoch number is greater than the predetermined epoch number, the node training ends.
(3) Use the GATHER primitive of PYTORCH's distributed communication GLOO back-end framework to determine whether the current computing task volume signal needs to be updated; in such a case, update the current node workload; otherwise, proceed to step (4).
(4) Obtain and record the current timestamp. If the BATCH number is a multiple of K, communicate with the parameter server and obtain the semaphore whether K needs to be updated through the GATHER primitive. The amount of training data for one iteration of the computing node is equal to the product of K and BATCH_SIZE; therefore, adjusting the size of K is equivalent to adjusting the amount of training data for one iteration. In such a case, receive it from the parameter server node; otherwise, proceed to step (5).
(5) Calculate the local gradient using the back-propagation algorithm and send it to the parameter server through the ISEND primitive.
(6) Receive the new gradient parameters from the parameter server node through the IRECV primitive as model parameters for the next iterative calculation.
(7) Determine whether the iteration is the last in the epoch, send its value to the parameter server, and then skip to step (2) for repeated training.
The pseudocode for the WLBS policy parameter server is given:
The implementation steps of the above strategy are as follows:
(1) Initialize the initial allocation of variables, the large adjustment and small adjustment semaphores of partial adjustment to TRUE, and the semaphore that stops partial adjustment to FALSE.
(2) If the current epoch number is less than or equal to the predetermined epoch number, the epoch training is entered; otherwise, if the epoch number is greater than the predetermined epoch number, the node training ends.
(3) The GATHER primitive is used to distribute the updated calculation task amount signal to all the nodes. If the update node calculation task amount signal is TRUE, it is judged again whether it is the initial training. In such a case, the initial division is adopted; otherwise, partial adjustment is used, and the initial division is used. The result of the partial adjustment is sent to all the nodes through the GATHER primitive.
(4) The GATHER primitive is used to distribute an iteratively adjusted training data volume signal to all the nodes. If the signal is true, the partial division is performed and the result is distributed to the designated computing node.
(5) Collect the gradient of the nodes in the limited list, the time information of this round of iterations, and whether the current iteration is the last iteration signal of this epoch.
(6) The collected gradients are aggregated into a global gradient and distributed to nodes in a limited list.
(7) If the current iteration is the last iteration of the epoch, enter the next epoch, skip to step (3), and repeat the training; otherwise, continue iterative training, skip to step (4), and repeat the training.
3.4 Theoretical Proof
In the distributed machine-learning training process, the training dataset was distributed on each computing node. This section analyzes the feasibility of the dynamic optimization strategy for computing tasks by combining it with the parameter update method of the node-training process. The machine-learning training process involves solving the optimal model parameters through repeated iterations, which is:
$${\omega }^{*}=\text{arg}\underset{\omega \in {R}^{P}}{\text{min}}\sum _{i=1}^{d}f\left(\omega ;{x}_{i},{y}_{i}\right)+ \tau R\left(\omega \right)$$
1
In the training that does not use the dynamic strategy of computing task volume, the training dataset D is evenly distributed among K computing nodes, denoted as \({D}_{1}，{D}_{2}，{D}_{3}，\dots ，{D}_{k}\)., t, s, s, umed that each computing node must perform n iterations in one epoch, and each iteration is random. Select a small batch sample set \({S}_{t}\subset \left\{1,\dots ,n\right\}\), then the gradient calculation expression for each time is
$$\nabla {f}_{{s}_{t}}\left({\omega }_{t}\right)=\frac{1}{\left|{S}_{t}\right|}\sum _{i\in {S}_{t}}\nabla {f}_{i}\left({\omega }_{t}\right)$$
2
The single node parameter update method is:
$${\omega }_{t+1}={\omega }_{t}-{\alpha }_{t}\nabla {f}_{{s}_{t}}\left({\omega }_{t}\right)$$
3
1)The parameter update method of the limited synchronous parallel model is:
$${\tilde{\omega }}_{k,t}={\omega }_{0}+\left[\sum _{{t}^{\text{'}}}^{t-m-1}\sum _{{k}^{\text{'}}}^{l}{u}_{{k}^{\text{'}},{t}^{\text{'}}}\right]+\left[\sum _{({k}^{\text{'}},{t}^{\text{'}})\in {u}_{k,t}}{u}_{{k}^{\text{'}},{t}^{\text{'}}}\right]$$
4
where \({\tilde{\omega }}_{k,t}\) represents the current model parameters, \({u}_{k,t}\) represents the updated subset of all computing nodes during the [t-m, t + m-1] iteration, and \({\tilde{\omega }}_{k,t}\) represents the current model parameters. \(\left[\sum _{{t}^{\text{'}}}^{t-m-1}\sum _{{k}^{\text{'}}}^{l}{u}_{{k}^{\text{'}},{t}^{\text{'}}}\right]\)represents the global update set of the previous local iteration, and \(\left[\sum _{({k}^{\text{'}},{t}^{\text{'}})\in {u}_{k,t}}{u}_{{k}^{\text{'}},{t}^{\text{'}}}\right]\) represents the update of this local iteration.
For a convex function \(f\left(\omega \right)=\frac{1}{T}\sum _{t=1}^{T}{f}_{t}(\omega )\), assuming that its component \({f}_{t}\left(\omega \right)\) is also a convex function, we determine the model parameter \({\omega }^{*}\) under its optimal model. Suppose there are K computing nodes in the distributed cluster, limited threshold \(l\in [1，K]\), and global synchroniz, tion threshold m > 1. Let \({u}_{t}≔-{\alpha }_{t}\nabla {f}_{t}\left({\tilde{\omega }}_{t}\right)\), where \({\alpha }_{t}=\frac{\sigma }{\sqrt{t}}\), \(\sigma =\frac{F}{L\sqrt{2(m+1)l}}\), and L and F are constants. Assuming that \({f}_{t}\) satisfies the Lipschitz continuity condition, and the parameters satisfy the \(‖\omega -{\omega }^{*}‖\le C,\forall \omega ,{\omega }^{*}\) condition, we obtain from [24]:
$$R\left(\omega \right)≔ \frac{1}{T}\sum _{t=1}^{T}{f}_{t}\left({\tilde{\omega }}_{t}\right)-{f}_{t}\left({\omega }^{*}\right)\le 4FLK\sqrt{2\left(m+1\right)lT}$$
5
In the case where the training data of the computing nodes are different, for a distributed cluster of K nodes, the number of samples that each node iteratively processes is \({size}_{k}\), and the total number of samples for one round of iterative training is \(\text{S}\text{I}\text{Z}\text{E} = {\sum }_{k=1}^{K}{size}_{k}\). Assuming that all samples are trained on one computing node, the learning rate is \(\alpha\), and the parameter update method is
$${\omega }_{i+1}={\omega }_{i}-\frac{\alpha }{SIZE}\sum _{j=1}^{M}\nabla {f}_{j}$$
6
The samples were distributed to K computing nodes and the parameter update method was
7
Here, \(\widehat{\nabla }{f}_{k}\) represents the cumulative value of the gradient of the loss function of the kth computing node, and the value of \(\frac{1}{K\times {size}_{k}}\) will change with the number of samples iteratively processed by each node, which can be dynamically updated according to the performance of the computing node. This expression can be obtained by reducing the influence of the gradient of the loss function of the computing nodes with higher performance and reducing the loss caused by the local model update on the overall training.
3) It can be observed from (1) and (2) that the limited synchronous parallel model is convergent, and the dynamic adjustment of the calculation amount will not change the convergence property of the model; therefore, the distributed machine learning training using the WLBS strategy is convergent. Therefore, it was theoretically proven that the WLBS strategy is feasible.
This section conducts an empirical study on the effectiveness of the WLBS strategy and experiments from the following three aspects.
1) Verify various communication strategies involved in related work, such as the overall synchronous parallel strategy, delayed synchronous parallel strategy, limited synchronous parallel strategy, and analysis of experimental data to verify the feasibility of theoretical derivation.
2) Combined with the application scenario of computing node performance fluctuation, design experiments and compare the experimental results with the experimental results of related work to verify the feasibility of the strategy proposed in this study to solve problems in related work.
3) Verify the scalability and robustness of the distributed machine learning system using the WLSB strategy.
4.1 Experimental environment
This section introduces the hardware and software environments, training dataset, and model.
4.1.1 Hardware Environment
In this experiment, four Linux server clusters in the laboratory were used as the hardware platform, and a deep-training model code module using the WLSB strategy was deployed on the cluster. In the cluster, one is used as a parameter server and the other three are used as computing nodes. The CPU of the parameter server is an Intel(R) Xeon(R) CPU E5-2620 v4 @ 2.10 GHz, with a total of 12 physical cores, memory of 32 GB, and disk capacity of 304 GB; the CPU of the computing node is an Intel(R) Xeon(R) CPU E5-2620 v4 @ 2.10 GHz, 32 GB memory, and 564 GB disk capacity. The computing node network in the cluster adopts Gigabit Ethernet. Table 3.1 shows the hardware configuration information of the servers in the cluster.
|
TYPE |
NUMs |
CONFIG |
CONTENT |
|---|---|---|---|
|
CPU |
Intel(R) Xeon(R) CPU E5-2620 v4 @ 2.10 GHz |
||
|
PARAMETERSERVER |
1 |
MEMORY |
32 GB |
|
HARD DRIVE |
304 GB |
||
|
CPU |
Intel(R) Xeon(R) CPU E5-2620 v4 @ 2.10 GHz |
||
|
NODE |
3 |
MEMORY |
32 GB |
|
HARD DRIVE |
564 GB |
4.1.2 Software Environment
The server used in this experiment was the CentOS Linux release 7.9.2009 (Core) operating system, which relies on several open-source software implementations. Table 3.2 shows the software that this experiment depends on, as well as the specific version information and function introduction.
|
NAME |
VERSION |
FUNCTION |
|---|---|---|
|
Python |
3.6.13 |
Cross-platform programming language, widely used in artificial intelligence, machine learning, and deep learning. |
|
Pip |
21.3.1 |
Standard librarian in Python |
|
TORCHVISION |
0.11.2 |
A library of handy tools for image manipulation independent of PYTORCH. |
|
NUMPY |
1.19.5 |
An extension library of the Python language that supports a large number of dimensional array and matrix operations, and also provides a large number of mathematical function libraries for array operations. |
|
PYTORCH |
1.10.1 + cpu |
Python-first deep learning framework. |
|
CONDA |
4.11.0 |
Opensource package management system and environment management system running on Windows, macOS and Linux. |
4.1.3 Training Dataset
Because the CIFAR-10 and CIFAR100 datasets are complex enough to train most of the capabilities of PYTORCH's large-scale models, and small enough to train quickly, this is ideal for trying out new ideas and experimenting with new techniques. Therefore, in this experiment, CIFAR-10 and CIFAR-100 were used as training datasets for the computing nodes. The dataset consisted of 60,000 32 × 32 color images in 10 categories, and each category contained 6,000 images. The specific images and file structures are shown in Figs. 7 and 8, respectively. There were 50,000 training and 10,000 test images. Because this experiment uses Python as the script training execution language, the dataset is the Python/MATLAB version, and the archive contains data_batch_1, data_batch_2,..., data_batch_5, and test_batch files.
Each batch file loaded this way contains an element dictionary of data and labels:
The data are a Numpy 2D array of size 10, 000 × 3072 and type uint8. Each row of the array stores 32×32 color image information; that is, each row stores 32×32×3 = 3072 digital information. The first 1024 data points in each row of the array represent the red channel values, the next 1024 data points represent the green channel values, and the last 1024 data points represent the blue channel values. The images are stored in row-major order; therefore, the first 32 values of the array represent the red channel values of the pixels in the first row of the image.
Labels are a list of numbers in the range of 0–9, with 10000 labels in a batch file. The index i of the label ranges from 1 to 10000, and the label at the ith position represents the category of the ith image in the data.
The dataset also contains a file with an object named batches.meta, which contains a list of size 10 named label_names, giving the above label numbers meaningful names. For example, label_names[0] == "airplane,” label_names[1] == "car,” etc.
Contrary to the CIFAR-10 dataset, there were 100 categories in the CIFAR-100 dataset, each of which contained 500 training data and 100 testing data.
4.1.4 Training The Model
The training model in this experiment used the visual geometry group (VGG) convolutional neural network model, and the model structure is shown in Fig. 9. This model is widely used in image recognition, face recognition, and other imaging fields. The VGG fully connected layer has three layers, which can be divided into VGG11–VGG19, according to the total number of convolutional layers and fully connected layers. This experiment used the VGG11 model, which contains eight convolutional layers and three fully connected layers. Before the VGG model, the convolutional neural network CNN rarely had a model with more than 10 layers of structure. The VGG model contributed to the training model and added CNN network layers. The VGG model also has its limitations and cannot increase the network in an unlimited manner. However, it cannot be denied that the VGG model has achieved good results on the ImageNet competition dataset.
4.2 Model Validity and Feasibility Analysis
First, because the selection of different machine learning models will have different effects on the experiment, it is difficult to have a fixed index value to measure the advantages and disadvantages of the distributed system. This experiment counts the use of BSP, SSP, and LSP strategies. The training time and accuracy of the VGG11 convolutional neural network model are used as a comparison reference for the WLBS strategy to more intuitively analyze the effectiveness and feasibility of the WLBS strategy. Moreover, to ensure the validity and feasibility of the model, it is necessary to ensure that the strategy proposed in this study is effective in the scenario of similar computing node performance; therefore, this experiment will statistically simulate the experimental results running in a scenario with similar computing node performance. Finally, because the node scale of the cluster in the production environment is large, this experiment simulated the operation of multiple computing nodes in the cluster through a multi-process mode to verify the scalability of the model.
4.3 Strategy Comparison Analysis
By analyzing the comparison of the BSP, SSP, LSP, and LSP strategies under the WLSB strategy in the same heterogeneous environment, with the increase in the iterative calculation super steps, the decline of the machine learning loss function, and the change in the training time, it is further explained. Such problems occur in heterogeneous node environments.
4.3.1 Similarity of Node Performance in Distributed Systems
This subsection first simulates the iterative training of computing nodes under a single thread and uses different model strategies. When the SSP delay threshold is set to 3 and the limited threshold l of the LSP model is set to 3, the experimental results of the iterative super steps, loss function value, and training time are shown in Figure 10–13:
As can be observed from Figs. 10 and 11, under the condition of similar cluster performance, after the same number of iteration steps, the magnitude of the loss function decline is relatively close because the loss function can reflect the accuracy to a certain extent. That is, under the same conditions, the accuracy rates of each computing node in a situation with a similar cluster performance are similar. It can be observed from Figs. 12 and 13 that the training times of the SSP, LSP, and WLSB strategies are significantly shorter than those of the BSP under the same number of super steps, and the training time of the WLSB strategy is close to that of the SSP. Therefore, to a certain extent, it can be reflected that the WLSB strategy is effective in the scenario where the performance of computing nodes is similar.
4.3.2 Differences In Node Performance In Distributed Systems
As in the previous subsection, this subsection simulates the iterative training of computing nodes under a single thread using different model strategies and sets up a slow node in the experiment to simulate the performance differences in distributed systems. The slow node will train 0.3–0.4 s slower than other nodes in each superstep. In addition, when the SSP delay threshold is set to 3 and the limited threshold l of the LSP model is set to 3, the experimental results of the iterative super steps, loss function value, and training time are shown in Fig. 14–17:
As can be observed from Figs. 14 and 15, in the case of differences in cluster performance, after the same number of iteration steps, the loss function of the WLSB strategy is slightly lower than that of other strategies because the loss function can reflect to a certain extent. The accuracy rate, that is, under the same conditions, the WLSB accuracy rate is better when the cluster performance is different. It can be observed from Figs. 16 and 17 that the training times of the SSP, LSP, and WLSB strategies are significantly shorter than those of the BSP under the same number of super steps, and the training time of the WLSB strategy is slightly lower than that of the LSP strategy. Therefore, the WLSB strategy is effective in situations where there are differences in cluster performance.
Based on the limited synchronous parallel model of static computing task allocation, in this study, we propose a dynamic optimization strategy for computing task volume, WLBS, which can accelerate the overall training speed of distributed machine learning training in heterogeneous environments. The WLBS strategy design is divided into initial and partial adjustments. The initial division can initially divide the computing resources according to the hardware device information of the distributed cluster nodes, and the partial adjustment adjusts the distributed ML computing tasks when the machine learning algorithm is running. These are divided into large and small adjustments. A large adjustment refers to the adjustment of the number of computing tasks in an epoch during training, and a small adjustment refers to one iteration in training, that is, the interval between each communication between the computing node and the parameter server. The number of computing tasks in each cycle is adjusted. According to the experimental results, it can be observed that conversely, WLBS has fast training speed, strong scalability, and portability. On the contrary, dividing data according to the performance of computing nodes can further improve model accuracy, achieve load balancing, and reduce redundant training.
- Zhou Q, Guo S, Lu H, Li L, Guo M, Sun Y, et al. Falcon: Addressing Stragglers in Heterogeneous Parameter Server Via Multiple Parallelism. IEEE Transactions on Computers. 2021;70(1):139-55. doi: 10.1109/tc.2020.2974461.
- Zhang Z, Wang C. SaPus: Self-Adaptive Parameter Update Strategy for DNN Training on Multi-GPU Clusters. IEEE Transactions on Parallel and Distributed Systems. 2022;33(7):1569-80. doi: 10.1109/tpds.2021.3118609.
- Wang S, Pi A, Zhou X. Elastic Parameter Server: Accelerating ML Training With Scalable Resource Scheduling. IEEE Transactions on Parallel and Distributed Systems. 2022;33(5):1128-43. doi: 10.1109/tpds.2021.3104242.
- Deng S, Zhao H, Xiang Z, Zhang C, Jiang R, Li Y, et al. Dependent Function Embedding for Distributed Serverless Edge Computing. IEEE Transactions on Parallel and Distributed Systems. 2022;33(10):2346-57. doi: 10.1109/tpds.2021.3137380.
- Cheng G, Guo D, Luo L, Xia J, Gu S. LOFS: A Lightweight Online File Storage Strategy for Effective Data Deduplication at Network Edge. IEEE Transactions on Parallel and Distributed Systems. 2022;33(10):2263-76. doi: 10.1109/tpds.2021.3133098.
- Xu H, Ho C-Y, Abdelmoniem AM, Dutta A, Bergou EH, Karatsenidis K, et al. GRACE: A Compressed Communication Framework for Distributed Machine Learning. 2021 IEEE 41st International Conference on Distributed Computing Systems (ICDCS)2021. p. 561-72.
- Wang S, Pi A, Zhou X, Wang J, Xu C-Z. Overlapping Communication With Computation in Parameter Server for Scalable DL Training. IEEE Transactions on Parallel and Distributed Systems. 2021;32(9):2144-59. doi: 10.1109/tpds.2021.3062721.
- Provatas N, Konstantinou I, Koziris N. Is Systematic Data Sharding able to Stabilize Asynchronous Parameter Server Training? 2021 IEEE International Conference on Big Data (Big Data)2021. p. 1092-101.
- Duriakova E, Tragos E, Lawlor A, Smyth B, Hurley N. Boosting the Training Time of Weakly Coordinated Distributed Machine Learning. 2021 IEEE International Conference on Big Data (Big Data)2021. p. 1023-9.
- yan. Lt. Distribute Machine Learning Theories，Algorithms，and Systems. Beijing: Ji xie gong ye chu ban she; 2018.
- Gupta V, Choudhary D, Tang P, Wei X, Wang X, Huang Y, et al. Training recommender systems at scale: Communication-efficient model and data parallelism. Proceedings of the 27th ACM SIGKDD Conference on Knowledge Discovery & Data Mining2021. p. 2928-36.
- Moreno-Alvarez S, Haut JM, Paoletti ME, Rico-Gallego JAJN. Heterogeneous model parallelism for deep neural networks. 2021;441:1-12.
- Wu H, Sun C, Li Y, Kuang YJIJoIPT. Simulation on dynamic load balancing of distributed parallel computing network system. 2021;14(3):139-46.
- Meng XR, Bradley J, Yavuz B, Sparks E, Venkataraman S, Liu D, et al. MLlib: Machine Learning in Apache Spark. J Mach Learn Res. 2016;17:7.
- Zheng SX, Meng Q, Wang TF, Chen W, Yu NH, Ma ZM, et al. Asynchronous Stochastic Gradient Descent with Delay Compensation. 34th International Conference on Machine Learning. Sydney, AUSTRALIA: Jmlr-Journal Machine Learning Research; 2017.
- Meng Q, Chen W, Wang Y, Ma ZM, Liu TY. Convergence analysis of distributed stochastic gradient descent with shuffling. Neurocomputing. 2019;337:46-57. doi: 10.1016/j.neucom.2019.01.037.
- Gava F, Loulergue F. A static analysis for Bulk Synchronous Parallel ML to avoid parallel nesting. Futur Gener Comp Syst. 2005;21(5):665-71. doi: 10.1016/j.future.2004.05.006.
- Ferrarotti F, Gonzalez S, Schewe KD. BSP abstract state machines capture bulk synchronous parallel computations. Sci Comput Program. 2019;184:12. doi: 10.1016/j.scico.2019.102319.
- Liu J, Wright SJ, Re C, Bittorf V, Sridhar S. An Asynchronous Parallel Stochastic Coordinate Descent Algorithm. J Mach Learn Res. 2015;16:285-322.
- Leblond R, Pedregosa F, Lacoste-Julien S. Improved Asynchronous Parallel Optimization Analysis for Stochastic Incremental Methods. J Mach Learn Res. 2018;19:68.
- Cannelli L, Facchinei F, Kungurtsev V, Scutari G. Asynchronous parallel algorithms for nonconvex optimization. Math Program. 2020;184(1-2):121-54. doi: 10.1007/s10107-019-01408-w.
- Shi H, Zhao Y, Zhang BF, Yoshigoe K, Vasilakos AV, Assoc Comp M. A Free Stale Synchronous Parallel Strategy for Distributed Machine Learning. International Conference on Big Data Engineering (BDE). Hong Kong, HONG KONG: Assoc Computing Machinery; 2019. p. 17-23.
- Zhao X, An AJ, Liu JF, Chen BX, Soc IC. Dynamic Stale Synchronous Parallel Distributed Training for Deep Learning. 39th IEEE International Conference on Distributed Computing Systems (ICDCS). Richardson, TX: Ieee Computer Soc; 2019. p. 1507-17.
- Yang R, Zhang JL, Wan J, Zhou L, Shen J, Zhang YC, et al. Parameter Communication Consistency Model for Large-Scale Security Monitoring Based on Mobile Computing. IEEE Access. 2019;7:171884-97. doi: 10.1109/access.2019.2956632.
- Liu C, Zhang P, Tang B, Shen H, Lai Z, Lo E, et al. Towards Self-Tuning Parameter Servers. 2020 IEEE International Conference on Big Data (Big Data)2020. p. 310-9.
- Abd Elrahman A, El Helw M, Elshawi R, Sakr S. D-SmartML: A Distributed Automated Machine Learning Framework. 2020 IEEE 40th International Conference on Distributed Computing Systems (ICDCS)2020. p. 1215-8.
No competing interests reported.