A combined priority scheduling method for distributed machine learning

Suppose that a distributed machine learning job to be scheduled in a cluster queue is denoted as \(J = \{ J_{1} ,...,J_{i} ,...,J_{n} \}\), where the \({\text{i}}\) th job \(J_{i}\) contains one or more tasks \(\left\{ {t_{i1} ,t_{i2} , \ldots ,t_{ij} , \ldots ,t_{im} } \right\}\) and task \(t_{ij}\) is the \(j\) th task of the \({\text{i}}\) th job \(J_{i}\). Job \(J_{i}\) cannot run normally until sufficient resources are successfully created for all tasks \(\left\{ {t_{i1} ,t_{i2} , \ldots ,t_{ij} , \ldots ,t_{im} } \right\}\) of this job. These tasks can be executed on one or more pods as a pod group \(P_{i} = \left\{ {p_{i1} ,p_{i2} , \ldots ,p_{ij} , \ldots ,p_{im} } \right\}\) in a Kubernetes cluster. In this paper, task \(t_{ij}\) runs on pod \(p_{ij}\). A Kubernetes cloud platform receives a queue of distributed machine learning jobs from different users, which require different resources determined by multiple impact factors including the user priority, job priority, job urgency, job parallelism, and affinity and anti-affinity.

If a Kubernetes cloud platform receives a queue of distributed machine learning jobs \(J = \left\langle {J_{1} ,J_{i} ,J_{n} } \right\rangle\), these jobs come from multiple users with different priorities. Job \(J_{i}\) comes from the \(i\) th user, and the user priority is \(u_{i} (J_{i} )\). The following formula defines the user priority \(U(J)\) for job queue \(J\).

The user priority \(u_{i} \left( {J_{i} } \right)\) has a relation \(u_{i} \left( {J_{i} } \right) \in N\), where \(N^{*}\) is a positive integer.

Different jobs running on a Kubernetes cloud platform have different job priorities, as described in the following.where \(B\left( J \right)\) is a priority of job queue \(J\). Job priority \(b_{i} (J_{i} )\) is related to \(b_{i} (J_{i} ) \in M^{*}\), where \(M^{*}\) is a positive integer.

User jobs have different degrees of urgency, which determine the longest wait times of the jobs. Suppose that the maximum wait time \(L(J)\) of the job queue is defined as follows:

There exists a relation \(L_{i} (J_{i} ) \in Q^{*}\), where \(Q^{*}\) is a positive integer within a certain range.

Distributed machine learning \(J_{i}\) contains multiple tasks that are executed using a parameter server architecture. This architecture involves a parameter server and a worker node. The worker nodes conduct training tasks and push the trained model parameters to the parameter nodes. The parameter servers store and update the model parameters. The worker nodes pull the updated model parameters and continue with the subsequent iterative training. When the distributed machine learning model involves massive parameters, model training requires a large number of work nodes to conduct tasks in parallel. The task parallelism \(M(J)\) of the job queue \(J\) can be denoted as the number of worker nodes for each job \(J_{i}\).

There exists a relation \(m_{i} (J_{i} ) \in X^{*}\), where \(X^{*}\) is a bounded positive integer.

Numerous data interactions occur between the worker and parameter nodes when conducting job \(J_{i}\) using the TensorFlow parameter server architecture. The placement of the parameter server and worker nodes on the same server reduces the transmission time loss between them and improves the job execution efficiency. When possible, the same type of node should not be placed on the same server. Affinity and anti-affinity relationships exist among parameter server or worker nodes. The affinity and anti-affinity among the parameter server and work nodes were therefore set as the two factors affecting the resource scheduling. Suppose that the worker nodes running job \(J_{i}\) are represented by the following expression \(w_{i1} ,w_{i2} , \ldots ,w_{ij} , \ldots ,w_{im}\). The parameter nodes are denoted as variables \(s_{i1} ,s_{i2} , \ldots ,s_{ij} , \ldots ,s_{il}\). The storage unit \(B_{i} = \{ b_{i1} ,b_{i2} , \ldots ,b_{ij} , \ldots ,b_{ir} \}\) is set to store tasks with an affinity relationship, such as a set of tasks with an affinity relation in storage unit \(b_{ij}\).

The involved factors of the combined priority scheduling (CPS) method is shown in Fig. 1. To improve the execution efficiency of different jobs and maximize the resource utilization, this method considers the user priority, job priority, job execution parallelism, maximum wait time of the job, and affinity and anti-affinity relations among the parameter server and worker nodes in the parameter server architecture. The priority of each job is influenced by the user priority, job priority, urgency, and task parallelism. Each job is composed of multiple tasks. The task priority is mapped to the priority of resource allocation with anti-affinity between the parameter server nodes and between the parameter server and work nodes.

Suppose that a job queue of distributed machine learning is denoted as \(J = \{ J_{1} , \ldots ,J_{i} , \ldots ,J_{n} \}\), where the number of jobs is larger than 1, that is, \(n \ge 1\). Each job \(J_{i}\) contains \(h\) tasks \(\{ t_{i1} ,t_{i2} , \ldots ,t_{ij} , \ldots ,t_{ih} \}\), where the number of tasks is greater than 1, that is, \(h \ge 1\). Assume that job \(J_{i}\) has user priority \(u_{i} (J_{i} )\), job priority \(b_{i} (J_{i} )\), maximum job wait time \(L_{i} (J_{i} )\), and task parallelism \(m_{i} (J_{i} )\). The specific process combining these priorities is shown in Fig. 2.

The relations \(hp \in N^{*}\), \(np \in N^{*}\), and \(lp \in N^{*}\) exist in which high job priority \({\text{hp}}\), medium job priority \({\text{np}}\), and low priority \({\text{lp}}\) are expressed by different values from large to small, such as \({\text{hp}}\; = \,10\), \({\text{np}}\; = \;5\), and \({\text{lp}}\; = \;1\).

The symbol “⌈⌉” is an upward integral function. The value of this variable is limited to within the range of \(1 \le L_{i} \left( {J_{i} } \right) \le 60\). The maximum wait time is 60 min, and the minimum wait time is 1 min. The time-constraint priorities are defined separately as 1 and 100 for time frames of more than 60 and 1 min, respectively. The value of the executed priority \(l_{i} (J_{i} )\) is limited to the range [1,100].

There exists a relation \(e_{i} (J_{i} ) \in N^{*}\), where symbol \(N^{*}\) is a positive integer. The set \(E(J) = \{ e_{1} (J_{1} ), \ldots ,e_{i} (J_{i} ), \ldots ,e_{n} (J_{n} )\}\) is the urgency of all jobs.

The job urgency \(e_{i} (J_{i} )\) is normalized as shown in Eq. (9), where \(M(J)_{\max }\) indicates the maximum value of job urgency \(E(J)\) for all jobs, \(E(J)_{\min }\) indicates the minimum value of job urgency \(E(J)\) for all jobs, and \(e_{i} (J_{i} )_{n}\) indicates the normalized value of job urgency \(e_{i} (J_{i} )\).

If the priorities of the parameter server nodes \(p_{i1} ,p_{i2}\) are higher than those of the worker nodes \(p_{i3} \ldots p_{im}\), the resources of the parameter server nodes \(p_{i1} ,p_{i2}\) are preferentially allocated to the worker nodes \(p_{i3} \ldots p_{im}\). Affinity and anti-affinity relationships are not mandatory. If a server has insufficient resources for the parameter server nodes, other servers that may violate the relations can allocate resources to these parameter server nodes.

The algorithm used by the CPS method (i.e., the CPS algorithm) is designed based on the Volcano scheduling framework, which provides basic scheduling modules and algorithms through plugins and can combine various scheduling algorithms. In this study, the algorithm design of the CPS method was completed, and to implement the algorithm, a plugin module called mypriority.go was developed. The algorithm design includes two main parts: inter- and intra-group priorities. Different priorities were defined for some YAML files. The mypriority.go plugin integrated into the Volcano scheduling framework calls these priorities to implement the inter- and intra-group priorities. The main steps of intra-group priority (i.e., job priority in a job queue) are as follows:

The time-constraint priority \(l_{i} (J_{i} )\) can be calculated using Eq. (7).

The inter-group priority, that is, the priority among tasks from a job, considers the affinity between parameter server pods and worker pods, and the anti-affinity among parameter server pods. The main steps are as follows:

A Kubernetes cloud environment is constructed using a virtual machine (VM) cluster for the experiment. The VM cluster has 10 VM nodes, including 1 master node and 9 computing nodes, as shown in Fig. 4. These nodes install Kubernetes components such as kubeadm, kubelet, kubectl, and docker. In the experiment, different numbers of computing nodes were used to validate the different scenarios. The node configurations are listed in Table 1.

The public MNIST dataset was used in the experiment. The TensorFlow and MindSpore frameworks were applied to train the model using the MNIST dataset. The experiment verified the priority scheduling between multiple jobs and the priority scheduling of tasks within a job. To analyze the effectiveness and applicability of the CPS algorithm under different impact factors, various experimental scenarios were established, as listed in Tables 2, 3 and 4. Table 2 presents the experimental scenarios with different impact factors for multiple jobs. In Scenario S1, multiple jobs have different job priorities but the same user priority, maximum wait time, and task parallelism. Scenario S2 indicates that multiple jobs have different maximum wait times but the same job priority, task parallelism, and user priorities. Scenarios S3 and S4 were similar to scenarios S1 and S2, respectively.

Table 3 presents the experimental setup for intra-group priority scheduling, including seven groups. Groups 1–4 use the same number of virtual machine nodes, training framework, and workers to verify the priority effectiveness of the CPS algorithm under different scenarios. Group 5 verifies whether the priority effectiveness of the CPS algorithm is affected by the training framework compared to group 3. Groups 6 and 7 verify whether the priority effectiveness is affected by the number of VM nodes.

Table 4 presents the experimental setup used for inter-group priority scheduling, including six of the groups. Each group reflects the state of the job execution. For example, group 1 uses the TensorFlow training framework, two parameter server pods, and six worker pods to execute multiple tasks. The Kubernetes cloud environment operates on six nodes: one master node and five computing nodes. Experiments 1 and 2 were conducted to verify whether the priority effectiveness of the CPS algorithm is affected by the number of VM nodes on the Kubernetes cloud platform. The group 3 experiment verified whether the priority effectiveness of the CPS algorithm is affected by the training framework compared with the group 2 experiment. Similarly, the group 4 experiment was conducted to verify whether the priority effectiveness of the CPS algorithm is affected by different numbers of worker pods compared to the group 2 experiment. The built-in priority and CPS algorithm plugins in the Volcano scheduler were separately used to conduct resource scheduling experiments. Groups 2, 5, and 6 were used to verify whether the priority effectiveness of the CPS algorithm is affected by the pod configuration.

The job parameters of distributed machine learning are listed in Table 5. By combining these parameters into different scenarios, the experiment results validate the idea that the final job priority is influenced by the four impact factors.

Because a job runs on a pod group in the Volcano scheduler, the priority among jobs is mapped to the scheduling order of the pod groups. The initial job priority is set to 1 for a low level, 50 for a normal level, and 100 for a high level. Under sufficient resources, the built-in priority scheduling and CPS algorithm plugins of the Volcano scheduler are implemented for five jobs. The scheduling order of the pod groups created on the resource nodes is obtained using the system log and timestamp.

The group 1 experiment conducted five distributed machine learning jobs using the same MNIST dataset under the TensorFlow framework and a Kubernetes cluster with three nodes. These five jobs were configured as the combined conditions of rows 1, 4, 5, and 7 in Table 5, that is, the corresponding scenario S1 in Table 2 with different job priorities and other identical parameters. Table 6 lists the scheduling order of the pod groups for these five jobs using the built-in priority and combined priority algorithms, for which each algorithm is implemented five times.

It can be seen from Table 6 that pod groups 3 (podg3) and 4 (podg4) were preferentially created owing to their higher job priority compared to other jobs. The pod group podg5 is created last because it has the lowest job priority. The results show that both the built-in and CPS algorithms can be prioritized according to their job priorities.

The results of group 2 were similar to those of group 1. Five jobs were set as the combined conditions for rows 1, 3, 3, 5, and 8 in Table 5. The maximum wait time for the jobs differed, and the other parameters were the same, corresponding to scenario S2 in Table 2. To ensure sufficient resources, the built-in priority and combined priority algorithms were tested five times. Table 7 lists the scheduling orders for the pod groups.

The maximum wait time increases from pod groups podg1 to podg5. The built-in priority algorithm of the Volcano scheduler cannot allocate resources in advance for jobs with a long wait time, whereas the CPS algorithm can preferentially allocate resources to jobs with the shortest wait time. Pod group podg1 preferentially allocates resources because it has the shortest wait time, followed by pod groups podg2, podg3, podg4, and podg5, the latter of which has the longest wait time.

The five jobs in group 3 were configured using the combined conditions of rows 1, 3, 6, and 7 in Table 5. These jobs have a different task parallelism and identical parameters, which correspond to scenario S3 in Table 2. The task parallelism of the five jobs decreased from job1 to job5. Thus, the pod group for job1 will be allocated resources first, followed by pod groups podg2, podg3, podg4, and podg5 in the CPS algorithm. The experiment results show that the CPS algorithm starts the pod groups from large to small according to the task parallelism, whereas the built-in priority scheduling order is random and unaffected by the task parallelism, as shown in Table 8.

The five jobs in the group 4 experiment were set as the combined conditions of rows 2, 3, 6, and 7 in Table 5 with different user priorities and other identical parameters, corresponding to scenario S4 in Table 2. The scheduling order of the pod groups under an experimental procedure similar to those of groups 1 and 2 is shown in Table 9. The scheduling order of the built-in priority algorithm is random and unaffected by the user priority, whereas the scheduling order of the CPS algorithm is arranged from large to small according to the user priority.

For group 5, the MNIST dataset was used to conduct five jobs of distributed machine learning under the MindSpore framework, which were configured using the combined conditions in rows 1, 3, 5, and 8 in Table 5. That is, the maximum wait times of the five jobs were different, and the other parameters were identical. It can be seen from the experiment results in Table 10 that the scheduling order of the pod groups for the five jobs is random, whereas pod group podg1 with the shortest wait time is preferentially scheduled, followed by pod groups podg2 and podg5, with the longest wait time created last. These results are similar to the scheduling of the TensorFlow framework, which indicates that the scheduling priority is unaffected by the distributed machine learning framework.

To conduct the experiments in the TensorFlow framework, groups 6 and 7 used 6 and 10 VM nodes, respectively. Except for the training framework, the five jobs were configured under the same conditions as in group 5. The experiment results are presented in Tables 11 and 12. The experiment results are similar to those for the cluster with three VM nodes, indicating that the built-in priority and CPS algorithms are unaffected by the cluster size.

In conclusion, the CPS algorithm can effectively solve the priority scheduling problem among multiple jobs by preferentially allocating resources to jobs with shorter wait times, large task parallelism, higher user priority, and higher job priority. The training framework of distributed machine learning and the cluster size cannot affect the effectiveness of a priority scheduling algorithm.

Six groups of experiments were conducted using the MNIST dataset, and each group was executed 20 times. The experiment results are listed in Table 13.

The inter-group priority scheduling success rate of the CPS algorithm exceeded 80%. The two task priorities were compared before the pods were created. The task information was obtained before comparing the priorities. If a task belongs to the “ps” type, this parameter sever pod will be unconditionally scheduled. Otherwise, it enters a normal-priority comparison program. Owing to the randomness of the task priority comparison, the parameter sever pod cannot be scheduled first every time. The CPS algorithm first increases the scheduling probability of the parameter server pods by determining the pod type created for a task. Second, this algorithm places worker nodes in the corresponding storage unit according to their affinity with the parameter server nodes when it is placed in a storage unit, which can prevent many worker nodes from being placed into a storage unit owing to a lack of affinity or anti-affinity relations between worker nodes. This ensures that the placement of pods is more reasonable in the cluster server and further improves the efficiency of job completion by optimizing the distance of the information interaction between the parameter server and worker nodes. The average job implementation time of the CPS algorithm is less than that of the built-in algorithm. The job implementation time of the CPS algorithm is more than 6% shorter than that of the built-in algorithm. It is unaffected by the cluster size or pod configuration. The job completion time of the group 5 experiment with 2ps6worker pods and all pods with a 1CPU1GB memory configuration is nearly double that of groups 1 and 2. The job completion time of the group 6 experiment with 2ps6worker pods and all pods with the 4CPU4GB memory configuration was greatly reduced to only 36.3 s. To further analyze the effectiveness of parameter server priority scheduling and job execution in detail, the experiment results of the randomly selected group 2 experiment are shown in Fig. 5. The scheduling order of the parameter server pods in the CPS algorithm clearly precedes the built-in priority algorithm in the Volcano scheduler. In 20 of the experiments, the CPS algorithm achieved the later scheduling order of the parameter server pods only three times and the same order as the built-in priority algorithm five times. In the ten experiments that recorded the job completion time shown in Fig. 6, the job completion time of the CPS algorithm was lower than that of the built-in priority algorithm 9 times. The remaining time gap was extremely small, i.e., 224 s for the CPS algorithm and 222 s for the built-in priority algorithm. The job completion time is reduced by setting the affinity or anti-affinity relations between the parameter server and worker pods in the CPS algorithm.

Group 5 conducted distributed machine learning on 2ps6worker pods under the TensorFlow framework, where each pod was allocated one CPU core and 1 GB of memory, and there were 10 VM nodes in the Kubernetes cloud environment. Group 6 uses a pod with four CPU cores and 4 GB of memory. The experiment results are presented in Figs. 7 and 8.

The left sides of Figs. 7 and 8 show that the scheduling order of most of the parameter sever pods of the CPS algorithm was lower than that of the built-in priority algorithm. The job completion time of the CPS algorithm was shorter than that of the built-in priority algorithm in most of the experiments. These two situations indicate that the CPS algorithm is more effective than the built-in priority algorithm for priority scheduling. This phenomenon is more evident when a task requires a smaller pod resource. For example, when the tasks request one CPU core and a 1-GB memory pod resource, the job completion time increases when applying the CPS or built-in priority algorithm, and their difference is large, with a maximum value of 32 s, as shown in Fig. 7. When the tasks requested four CPU cores and a 4-GB memory pod resource, the job completion time reduced the performance of the CPS or built-in priority algorithm, and their difference was small, with an average value of 1.5 s, as shown in Fig. 8.

To verify the effectiveness of the CPS algorithm in avoiding resource deadlock and waste, an experiment was conducted on five distributed machine learning jobs using the parameters listed in Table 14. Here, job5 was configured with 2ps6worker pods, and job1, job2, job3, and job4 were configured with 2ps4worker pods. Each pod requested two CPU cores and 2 GB of memory. The number of VMs for the Kubernetes cloud environment was set to three. The experiment results are shown in Figs. 9 and 10.

The scheduling of pod groups under the built-in priority algorithm in the Volcano scheduler was podg1, podg2, podg3, podg4, and podg5. Pod group podg5 with a higher user priority cannot be created because of insufficient resources, which can result in a longer job completion time and resource usage. However, job5 had a higher combined priority owing to its high user priority, high task parallelism, and shortest wait time. Therefore, podg5 preferentially allocates resources to conduct job5. Pod group podg1 does not allocate resources to conduct job1 owing to its low combined priority. This experiment verifies that the CPS algorithm works together with the group scheduling strategy for distributed machine learning jobs.