COMPUTER AND JOB SCHEDULING METHOD

A processor acquires and stores a new-job, acquires information regarding an execution state of existing-jobs run on the compute nodes for each group of compute nodes that have a short communication distance, when the new-job is deployed in the compute nodes that belong to the group, based on the acquired information regarding the execution state, obtains, for each group, a probability in which the existing-jobs or a part of the new-job is deployed in the compute nodes that belong to a group different from a deployment destination group in which the new-job is deployed, determines a group in which the new-job is deployed, based on the obtained probability and a usage amount of the compute nodes for each group by the existing-jobs, and acquires the stored new-job, and deploy the new-job in the compute node, based on the determination of the group in which the new-job is to be deployed.

CROSS-REFERENCE TO RELATED APPLICATION

This application is based upon and claims the benefit of priority of the prior Japanese Patent Application No. 2021-42968, filed on Mar. 16, 2021, the entire contents of which are incorporated herein by reference.

FIELD

The embodiments discussed herein are related to a computer and a job scheduling method.

BACKGROUND

Neural network learning has been greatly advanced due to a strong and efficient parallel processing capability of a graphics processing unit (GPU). Moreover, in order to handle a larger learning model using more learning parameters, a GPU cluster including compute nodes, each of which has a plurality of GPUs, is generally used. Here, the compute node is a device that executes a job and generally indicates a server or the like. The GPU cluster executes a plurality of jobs in parallel, and the plurality of jobs shares the same compute node in many cases. A state where the plurality of jobs shares the same compute node may be referred to as co-located. For example, among large GPU clusters, there is a system that includes 1000 or more compute nodes each of which mounts four or eight GPUs and includes a shared file system, a cloud storage, an InfiniBand network that couples these, or the like.

Most of a workflow when the neural network is learned by each compute node of such a GPU cluster is offload-executed calculated by the GPU. In other words, for example, it can be said that most of resources consumed in the learning of the neural network are in the GPU side.

A processing flow for learning a neural network is as follows. First, initialization is performed, and a data library is loaded. Next, a gradient is calculated by a GPU. Next, a learning model is updated, based on the calculated gradient. Then, until an evaluation index exceeds a target value or an execution elapsed time reaches an upper limit, the calculation of the gradient and the update of the learning model are repeated.

In recent years, the size of a leaning task has been increased due to complexity of the learning model or the like, and the number of cases where distributed training using the plurality of GPUs (compute nodes) is performed has been increased. In a case where one process is allocated to each GPU and distributed training is performed, the processes (compute nodes) may communicate among them for reasons such as gradient sharing. Here, a case will be described where distributed training is performed by synchronous data parallelism. For example, forward calculation and backward calculation are performed by a GPU mounted on a specific compute node using specific data, and a GPU mounted on another compute node performs forward calculation and backward calculation using another piece of data. Then, the GPU of each compute node shares calculation results with each other and updates the learning model. However, at this time, communication occurs after the processes are synchronized.

In such distributed training, there is a case where auto-scaling such as scale-in or scale-out of a process is performed after job execution is started. The scale-in is a technique that excludes some processes from among a plurality of processes for executing jobs. Furthermore, the scale-out is a technique that adds a new process to processes for executing jobs. In order to efficiently and accurately perform the neural network, the auto-scaling techniques have been actively studied also in the academic world.

For example, as an application example of scale-in, a method called straggler mitigation exists. The straggler mitigation is a method for excluding a process executed by a GPU from a job in a case where a straggler occurs in a GPU of which processing delays than other GPUs. Straggler occurs for various reasons such as a high temperature of a GPU or a shared job. In a case where a straggler occurs in a job, a waiting time before synchronization increases due to the straggler. However, by excluding the straggler process executed by the GPU from the job, the waiting time can be shortened.

On the other hand, scale-out is performed so as to shorten a time required for learning the neural network by increasing the processes for executing the job in a case where learning of the neural network takes more time than expected or the like. In addition, scale-out is performed also in a case where the straggler recovers from slowdown and the process executed by the GPU returns to the job after the straggler is excluded by scale-in.

Here, in a case where distributed training is performed, a job is arranged in a compute node at the time start of job execution, scale-in, and scale-out. As this technique for arranging the job in the compute node, there is a technique for waiting for job execution until an ideal compute node group becomes available. Furthermore, there is a technique for dynamically migrating a process being executed by a remote node according to operation characteristics.

In addition, there is a technique for determining a compute node where a job is assigned, based on communication characteristics of the job and communication characteristics of interconnect between the compute nodes. Moreover, as a technique regarding auto-scale, there is a technique for moving data, including one held by a server previous to a server to be added, to the server to be added among the data of jobs that are continuously executed by each server, at the time of adding the server.

Japanese Laid-open Patent Publication No. 2016-099972, Japanese Laid-open Patent Publication No. 2011-175573, and Japanese Laid-open Patent Publication No. 2012-038053 are disclosed as related art.

SUMMARY

According to an aspect of the embodiments, a computer including a plurality of compute nodes that executes a job and is communicable with each other, the computer includes a memory, and a processor coupled to the memory and configured to acquire a new job and store the new job in the memory, acquire information regarding an execution state of existing jobs run on the compute nodes for each group of the compute nodes that have a short communication distance, when the new job is deployed in the compute nodes that belong to the group, based on the acquired information regarding the execution state, obtain, for each group, a probability in which the existing jobs or a part of the new job is deployed in the compute nodes that belong to a group different from a deployment destination group in which the new job is deployed, determine a group in which the new job is deployed, based on the obtained probability and a usage amount of the compute nodes for each group by the existing jobs, and acquire the stored new job, and deploy the new job in the compute nodes, based on the determination of the group in which the new job is to be deployed.

DESCRIPTION OF EMBODIMENTS

When a job is assigned to a compute node, there is a case where some processes of the job are assigned to compute nodes that are in a remote positional relationship. For example, in a case where a compute node assigned near a compute node on which a process of a specific job executes is occupied by another job, it is difficult to collectively assign all the processes of the specific job to an adjacent node. In that case, the number of hops for connecting between the compute nodes increases, and there is a possibility that communication cost increases. Furthermore, in a case where communication is performed via a plurality of switches, collision with communication of another job occurs. This causes the performance degradation due to the increased load on the switch, and there is a possibility that the communication cost further increases.

Therefore, it is considered to realize ideal assignment using a technique for waiting for job execution until an ideal compute node group becomes available. However, because unnecessary waiting time occurs and this delays entire processing, this technique is not realistic. Furthermore, with the technique for dynamically migrating a process being executed in a remote node according to operation characteristics, the cost of migration is high. Therefore, it is difficult to obtain an effect of reducing communication cost. Furthermore, with the technique for determining the compute node where the job is assigned, based on the communication characteristics of the job and the communication characteristics of the interconnect between the compute nodes, a communication path between the compute nodes is not considered, and it is difficult to reduce the communication cost. Moreover, with the technique for moving the data including one held by the previous server to the server to be added among the data of the jobs continuously executed, an effect is considered for equalizing loads and reducing the number of communications. However, there is a possibility that a communication path is lengthened, and it is difficult to reduce the communication cost.

Hereinafter, embodiments of the technology for reducing communication cost will be described in detail with reference to the drawings. Note that a computer and a job scheduling method disclosed in the present application are not limited to the following embodiments.

First Embodiment

FIG. 1is a configuration diagram illustrating an example of a cluster system. As illustrated inFIG. 1, a cluster system100according to the present embodiment is a computer including racks21and22. Here, inFIG. 1, the two racks21and22are illustrated. However, the number of racks included in the cluster system100may be equal to or more than three and is not particularly limited. Hereinafter, in a case where each rack including the racks21and22included in the cluster system100is not distinguished, the racks are referred to as a “rack20”.

Each of the racks21and22mounts a plurality of compute nodes10. Furthermore, switches31respectively connected at multiple stages are arranged in the racks21and22. The compute nodes10mounted on the rack21can communicate with each other using the switches31arranged in the rack21. Similarly, the compute nodes10mounted on the rack22can communicate with each other using the switches31arranged in the rack22.

Moreover, a network in the rack21and a network in the rack22are connected with switches32that connect the racks21and22. Then, the compute node10arranged in the rack21and the compute node10arranged in the rack22can communicate with each other via the switch32that connects the racks21and22. Here, even in a case where the cluster system100includes three or more racks20, the switches32for connecting the racks20are arranged, and the compute node10arranged in each rack20can communicate to a compute node10arranged in any rack20.

In a case of the configuration illustrated inFIG. 1, the five switches31and32relay transmission of a signal from the compute node10arranged in the rack21to the compute node10arranged in the rack22. In other words, for example, the number of hops between the compute node10arranged in the rack21and the compute node10arranged in the rack22is six. Here, as the number of hops between the compute nodes10for communicating with each other increases, the communication cost increases. In other words, for example, the cost of communication within the rack21or22can be suppressed to be lower than that of the communication between the racks21and22.

FIG. 2is a hardware configuration diagram of a compute node. As illustrated inFIG. 2, the compute node10includes central processing units (CPU)11A and11B, memories12A and12B, network interfaces13A and13B, an auxiliary storage device14, peripheral component interconnect expresses (PCIe)15A and15B, and GPUs16A to16D.

The CPUs11A and11B are connected to be communicable with each other. Furthermore, the CPU11A is connected to the memory12A, the network interface13A, and the PCIe15A. The CPU11A communicates with each of the GPUs16A to16D via the PCIe15A. Furthermore, the CPU11B is connected to the memory12B, the network interface13B, the auxiliary storage device14, and the PCIe15B. The CPU11B communicates with each of the GPUs16A to16D via the PCIe15B. The CPUs11A and11B control and manage operations of each unit of the compute node10. Furthermore, the CPUs11A and11B manage communication with another compute node10.

The memories12A and12B are primary storage devices. For the memories12A and12B, for example, a double data rate (DDR)4synchronous dynamic random-access memory (SDRAM) can be used.

The network interfaces13A and13B are, for example, InfiniBand host bus adapters (HBA). The network interfaces13A and13B are connected to an external switch31(refer toFIG. 1) and relay the communication with another compute node10.

The auxiliary storage device14is, for example, a non-volatile storage medium connected using the non-volatile memory express (NVMe).

The GPUs16A to16D are connected to be communicable with each other. The GPUs16A to16D receive an input of a job from the CPU11A or11B and execute the input job. The GPUs16A to16D execute, for example, a job for training a neural network. The GPUs16A to16D read data and a library and calculate a gradient using a learning model. Then, the GPUs16A to16D update the learning model using the calculated gradient. The GPUs16A to16D train the learning model by repeating to calculate the gradient and update the learning model. In the present embodiment, the GPUs16A to16D perform distributed training. Therefore, when the learning model is updated, for example, the GPU16A communicates with GPUs16B to16D on a compute node10, mounting the GPU16A, that performs distributed training in cooperation or GPUs16A to16D mounted on another compute node10and shares the gradients or the like.

Here, as described above, as the number of hops between the compute nodes10for communicating with each other increases, the communication cost increases. Therefore, the cluster system100according to the present embodiment arranges jobs so as to reduce the number of hops in the communication occurred by the execution of the job.

FIG. 3is a block diagram regarding a job deployment function of a cluster system. Job deployment by a cluster system100according to the present embodiment will be described below with reference toFIG. 3.

The cluster system100is connected to an external device such as client devices200and201via a network. The client devices200and201are terminal devices operated by a user who requests the cluster system100to execute a job to train a neural network. Here, inFIG. 3, as an example, the two client devices200and201are illustrated. However, in many cases, a large number of terminal devices actually exist.

For example, the user requests job deployment to the cluster system100using the client device200. At this time, the client device200transmits information used to designate a job to be executed and data used to execute the job to the cluster system100.

The cluster system100holds a large number of racks20including the racks21and22. Each rack20mounts a plurality of compute nodes10. As illustrated inFIG. 1, the compute nodes10are connected to be individually communicable with each other using the plurality of switches31and32.

Moreover, the cluster system100includes a cluster management unit101. The cluster management unit101determines a compute node10to be a job arrangement destination and deploys the job to the assigned compute nodes10. In order to implement functions of the cluster management unit101, one compute node10may be allocated, or a part of resources of the compute node10that executes the job may be used. The cluster management unit101includes a queue111, a job deployment unit112, a job deployment destination determination unit113, and a job management unit114.

A new job designated in response to an execution request from the client device200is input to the queue111, and the queue111stores the input job. For example, multiple jobs are arranged in order in the queue111.

The job management unit114confirms an existing job being executed in the cluster system100. Specifically, for example, the job management unit114transmits a management command to each compute node10and acquires information regarding a job execution status. Then, the job management unit114transmits, via a network, each execution status of each job to, for example, the client device200or201that has requested to execute each existing job that is being executed. Furthermore, the job management unit114outputs the execution status of each job to the job deployment destination determination unit113. This job execution status includes information used to acquire the execution state of the existing job such as information representing a resource amount in use for each rack20or information representing a possibility of a fluctuation in the resource amount in use.

Furthermore, the job management unit114monitors collective communication performed in the existing job. Then, the job management unit114determines to execute scale-out or scale-in of each job according to the number of processes participating in communication. Then, the job management unit114notifies the job deployment destination determination unit113of an instruction to execute scale-in or scale-out of the job on the basis of the determination.

FIG. 4is a diagram for explaining auto-scale processing. InFIG. 4, the vertical axis indicates a type of a process that executes processing, and the horizontal axis indicates a type of processing to be executed with time. Here, a case will be described where n processes P0 to P (n−1) exist as processes of a job. As the types of the processing to be executed, F represents forward processing, B represents backward processing, and U represents learning model update processing referred to as update. In this case, communication between processes is performed in the update processing.

For example, the job management unit114monitors collective communication in a period301and confirms that the processes P0 to P (m−1) participate in the collective communication. Then, in a case where it is determined that the number of processes participating in the execution of the job is large with respect to a load of the job to be executed, the job management unit114determines to perform scale-in on the job. As a result, the processes participating in the execution of the job are reduced to processes P0 to P (k−1).

Next, the job management unit114monitors collective communication in a period302and confirms that the processes P0 to P (k−1) participate in the collective communication here. Then, in a case where it is determined that the number of processes participating in the execution of the job is small with respect to the load of the job to be executed, the job management unit114determines to perform scale-out on the job. As a result, the processes participating in the execution of the job are increased to processes P0 to P (n−1) and execute the job as indicated in a period303.

The job deployment destination determination unit113receives an input of the execution status of each job from the job management unit114. Next, the job deployment destination determination unit113calculates a score indicating a probability that some processes of a job to be input are deployed in another rack20with respect to the rack20to be a deployment destination candidate using the information, regarding the execution state of each job being executed, acquired from the execution status.

For example, the job deployment destination determination unit113calculates the score using the following formula (1).

Here, ki is a coefficient with respect to an element xi and is a value preset according to how much the element xi is emphasized. The larger ki is, the more important the element xi is. Furthermore, the xi is information representing a job execution state. The information representing the job execution state is information used to obtain an amount of resources in use for each rack20, information representing a possibility of a fluctuation in the amount of the resources in use, or the like.

More specifically, for example, as xi, information representing a job processing load or a free space can be used. For example, as xi, the number of executions of the process in each rack20can be used. Furthermore, as xi, the number of processes that is reduced by scale-in can be used. From this information, it can be determined that there is a high possibility of scale-out performed in order to return the reduced processes. Furthermore, as xi, a remaining time to an upper limit value of a job execution time can be used. From this information, it can be determined that there is a high possibility that the number of processes is increased in a case of immediately after job start, and it can be determined that there is a high possibility that the number of processes is decreased in a case where the job is close to end. Furthermore, as xi, characteristics of a user or characteristics of a process obtained from operation information in the past can be used. For example, it can be determined that there is a high possibility that the same user executes the same binary a plurality of times for hyperparameter tuning or the like. Furthermore, a probability that the number of processes fluctuates can be obtained, based on the past results.

FIG. 5is a diagram for explaining a method for calculating a rack empty state. Here, a case will be described where racks21to23exist. InFIG. 5, the entire resources of the respective racks21to23are the same, and all the resources are normalized to one. A left-side part of each of the racks21to23as facing the paper surface indicates an actual resource, and a right-side part as facing the paper surface indicates a state according to a probability that the existing job or a new job to be input is deployed in the compute node10of another rack20. Furthermore, here, a job of which a requested resource amount is p is inserted as a new job. The requested resource amount is a value indicating how many compute nodes10are required by the job as resources to be used.

The score calculated for each of the racks21to23by the job deployment destination determination unit113is represented by a right-side portion of each of the racks21to23on a paper surface inFIG. 5. The score is an index indicating a probability that a part of processes included in an existing job or a new job to be deployed is deployed in another rack20. In other words, for example, it can be said that the rack20having the higher score has a higher possibility that a part of the job deployed in the rack20is deployed in another rack20.

Next, the job deployment destination determination unit113obtains a resource amount in use in each rack20from the job execution status. For example, the resource amount in use obtained by the job deployment destination determination unit113is represented by a left-side portion of each of the racks21to23on the paper surface inFIG. 5.

Moreover, the job deployment destination determination unit113acquires the information of the new job designated by the execution request from the client device200from the queue111and acquires a requested resource amount that is a resource amount used to deploy the new job. Then, for the determination regarding the rack20where the new job is deployed, the job deployment destination determination unit113uses a bin packing problem for packing the new job in any one of the racks20using the calculated score, the resource amount in use of each rack20, and the requested resource amount. By solving this bin packing problem, the job deployment destination determination unit113determines the rack20in which the designated job is deployed.

In the present embodiment, the job deployment destination determination unit113solves the bin packing problem using a Best-Fit method that is a method for loading an item in a box with the smallest free space among boxes in which items may be loaded, as an algorithm for solving the bin packing problem. For example, the job deployment destination determination unit113calculates fi that is an arrangement possibility index of each rack20using the following formula (2).

Here, i is a number sequentially allocated to each rack20from one. Hereinafter, an i-th rack20is referred to as a rack i. The reference ri indicates a resource amount used by the rack i. Furthermore, the reference ρ indicates a requested resource amount of a newly deployed job. Furthermore, the reference λ indicates a coefficient for giving a score importance. Furthermore, the reference si indicates a score of the rack i.

The job deployment destination determination unit113determines that the rack i having a smaller value fi is a preferable rack20as a deployment destination of the new job. In other words, for example, the job deployment destination determination unit113selects the rack20where the job is deployed in consideration of that the rack20can use the resource amount up to the upper limit as possible and the rack20has a low probability that some jobs are deployed in another rack20. However, in a case where the total of the resource amount in use and the requested resource amount of the job to be deployed exceeds the resource amount of the entire rack20, the job deployment destination determination unit113does not deploy the job in the rack20.

Then, the job deployment destination determination unit113determines the rack20with the smallest fi as the rack20in which the new job is deployed, using the calculated score and the resource amount in use in each rack20. Thereafter, the job deployment destination determination unit113notifies the job deployment unit112of information regarding the compute nodes10in the rack20that execute the new job to be deployed.

FIG. 6is a diagram illustrating an example of new job deployment. An example of the new job deployment will be described with reference toFIG. 6. Here, a case will be described where a requested resource amount of a new job is 0.25 and a coefficient representing a score importance is one. For example, the job deployment destination determination unit113calculates that the score is 0.4 with a resource amount in use in the rack21of 0.8, the score is 0.6 with a resource amount in use in the rack22of 0.55, and the score is 0.75 with a resource amount in use in the rack23of 0.4. InFIG. 6, rAand sArespectively represent the resource amount in use in the rack21and the score, rBand sBrespectively represent the resource amount in use in the rack22and the score, and rCand sCrespectively represent the resource amount in use in the rack23and the score.

In this case, the job deployment destination determination unit113determines that it is not possible to deploy the new job in the rack21because the sum of the resource amount in use in the rack21and the requested resource amount of the new job exceeds the resource amount of the entire rack21. Then, because an arrangement possibility index of the rack22is smaller than an arrangement possibility index of the rack23, the job deployment destination determination unit113determines to deploy the new job in the rack22.

Furthermore, the job deployment destination determination unit113receives a scale-in execution instruction from the job management unit114. Then, the job deployment destination determination unit113determines a compute node10to be a scale-in target from among the compute nodes10executing the process of the job that performs scale-in. In particular, in a case where the jobs to be the scale-in targets operate in the compute nodes10in the different racks20, the job deployment destination determination unit113determines the compute node10to be the scale-in target so that all the processes of the job are executed by the compute nodes10in the single rack20. Then, the job deployment destination determination unit113notifies the job deployment unit112of information regarding the compute node10determined to be the scale-in target.

Furthermore, the job deployment destination determination unit113receives a scale-out execution instruction from the job management unit114. Then, the job deployment destination determination unit113determines which compute nodes10will be assigned for a job to be a scale-out target. In a case where the compute nodes10mounted on the rack20in which the job to be the scale-out target operates are not sufficient, the job deployment destination determination unit113assigns a compute node10of another rack20as a scale-out target. Then, the job deployment destination determination unit113notifies the job deployment unit112information regarding the compute node10determined as the scale-out target.

The job deployment unit112acquires a new job designated by the execution request from the client device200from the queue111. Moreover, the job deployment unit112receives a notification of a compute node10, which is a job deployment destination, mounted on the rack20that is determined as the job deployment destination from the job deployment destination determination unit113. Then, the job deployment unit112deploys the new job acquired from the queue111in the compute node10designated as the deployment destination.

Furthermore, the job deployment unit112receives a notification of the compute node10to be the scale-in target from the job deployment destination determination unit113. Then, the job deployment unit112stops the process of the job that operates on the compute node10designated as the scale-in target.

Furthermore, the job deployment unit112receives a notification of the compute node10to be the scale-out target from the job deployment destination determination unit113. Then, the job deployment unit112makes the compute node10designated as the scale-out target start to execute the process of the scale-out target job.

FIG. 7is a diagram illustrating an outline of processing for determining a deployment destination compute node. For example, as illustrated inFIG. 7, a case will be described where there are the racks21and22on which eight compute nodes10are mounted. Here, a case will be described where a new job sets resource amounts of two compute nodes10as a requested resource amount.

In the rack21, six compute nodes10in a range211are secured by an existing job. Then, the existing job uses four compute nodes10in a range212, and two compute nodes10in a range213are excluded from job execution by scale-in.

On the other hand, in the rack22, four compute nodes10in a range221are secured by an existing job. Then, the existing job uses three compute nodes10in a range222, and one compute node10in the range223is excluded from job execution by scale-in. We assume that the existing job executed by the three compute nodes10in the range222will be completed shortly.

In this state, a free resource amount of each of the racks21and22is equal to or more than the requested resource amount, and a new job can be deployed. However, in the rack21, where two compute nodes10are excluded by scale-in, there is a high possibility that these compute nodes10return to execution of the existing job. In this case, a free resource amount is largely decreased. On the other hand, in the rack22, one compute node10is excluded by scale-in. Even if the compute node10returns to execution of the existing job, the decrease in the free resource amount is small. Moreover, in the rack22, the execution of the existing job will be immediately completed, and this generates a free space.

In this case, when the new job is deployed in the compute node10of the rack21, a resource amount of the entire rack21can be used up to the upper limit. However, in a case where the existing job or the new job performs scale-out, a possibility that a part of the job is deployed in another rack20is high, and a score is calculated to be high. Therefore, in a case ofFIG. 7, in the cluster system100, the new job is deployed in the compute nodes10mounted on the rack22.

FIG. 8is a flowchart of job deployment processing by a cluster system according to the first embodiment. Next, a flow of job deployment processing by the cluster system1according to the first embodiment will be described with reference toFIG. 8.

The queue111receives a job deployment request from the client device200and stores the job deployment request (operation S1).

The job management unit114confirms a job being executed by the compute node10mounted on each rack20in the cluster system1(operation S2).

The job management unit114notifies the job deployment destination determination unit113of an execution status of each job. The job deployment destination determination unit113calculates a score for each rack20using information obtained from the execution status of each job acquired from the job deployment destination determination unit113in the formula (1) (operation S3).

Next, the job deployment destination determination unit113calculates a resource amount in use in each rack20from the execution status of each job. Furthermore, the job deployment destination determination unit113acquires information regarding a new job from the queue111and obtains a requested resource amount. Then, the job deployment destination determination unit113determines a rack20to be a job deployment destination and a compute node10to be a deployment destination among compute nodes10mounted on the rack20using the score, the resource amount in use, and the requested resource amount (operation S4).

Next, the job deployment destination determination unit113notifies the job deployment unit112of information regarding the compute node10determined as the deployment destination. The job deployment destination determination unit113deploys a job in the compute node10designated by the job deployment destination determination unit113as a deployment destination (operation S5).

As described above, the cluster system according to the present embodiment determines a rack where a new job is deployed, based on resource usage statuses of the respective racks and a possibility that a part of the new job is deployed in another rack and deploys the new job in a compute node mounted on the rack. As a result, a probability of securing a compute node group having a short communication distance is improved at the time of input of a new job. Furthermore, in a case where the job being executed performs scale-out, a probability that an additional process is deployed at a position with a short communication distance to the secured node group is improved. Therefore, the compute node group that executes the job has a positional relationship with a short communication distance, and the number of communication hops can be reduced. Therefore, it is possible to reduce the communication cost.

Furthermore, in the above description, the rack is used as a group of the compute nodes10having a positional relationship in which the compute nodes10have a short communication distance. However, it is possible to use another section in a system as long as a communication distance between the belonging compute nodes10is short. For example, in the system in which a plurality of compute nodes10is connected to each other, the compute nodes10can be classified into groups in which communication between belonging compute nodes10is within a predetermined number of hops at the maximum, and the group can be treated similarly to the rack in the first embodiment. As a result, a probability of securing a compute node group having a short communication distance is improved at the time of inputting a new job into the system.

In the first embodiment, the Best-Fit is used as the algorithm for solving the bin packing problem. However, by solving the bin packing problem using another algorithm other than this, a rack20to be a job deployment destination can be determined.

For example, the algorithm for solving the bin packing problem includes a First-Fit method for selecting a box with the smallest subscript among boxes in which items can be loaded and a Worst-Fit method for selecting a box with the maximum free space among boxes in which items can be loaded. In a case where the Worst-Fit is used, processing is as follows.

In this case, the job deployment destination determination unit113calculates fi that is an arrangement possibility index of each rack20using the following formula (3). Here, ri+ρ represents a resource usage amount of each rack20after the new job is added.

Then, the job deployment destination determination unit113determines a rack20with the smallest fi as a new job deployment destination. In this case, broadly speaking, the job deployment destination determination unit113preferentially selects the rack20with the smallest ri+ρ. Note that, depending on the value of λsi, the job deployment destination determination unit113may select a rack20other than the rack20with the smallest ri+ρ.

As described above, even if the algorithm other than the Best-Fit is used as the algorithm for solving the bin packing problem, the deployment destination rack can be determined so that the compute node group that executes the job has a close positional relationship. Therefore, the communication cost can be reduced.

Second Embodiment

Next, a second embodiment will be described. A cluster system according to the present embodiment is illustrated in the block diagram inFIG. 3. A cluster system1according to the present embodiment determines a job deployment destination in a case where a plurality of jobs shares the same compute node10. In the following description, description of a function of each unit similar to that in the first embodiment will be omitted.

A job deployment destination determination unit113acquires a resource requirement condition of each job for operating in the cluster system1from an execution status of each job acquired from a job management unit114. The resource requirement condition is information indicating which one of a node-occupying job that occupies the compute node10where the job is deployed and does not allow the compute node10to be shared with another job or a node-sharing job that allows to share the compute node10where the job is deployed with another job.

Then, the job deployment destination determination unit113sets a free resource to zero for the compute node10where the node-occupying job is deployed. Furthermore, for the compute node10where the node-sharing job is deployed, the job deployment destination determination unit113calculates a free space of the compute node10by subtracting a resource amount used for the job.

FIG. 9is a diagram for explaining free space calculation according to the second embodiment. Here, an example of a state will be described where compute nodes #1 to #4 are mounted on a rack20as compute nodes10and jobs J1 to J3 are deployed therein. A requested resource amount of the job J1 in a case where a total resource amount of each compute node10is set to one is 0.75. Moreover, the job J1 is a node-occupying job. Furthermore, a requested resource amount of the job J2 is 1.5. Moreover, the job J2 is a node-sharing job. Furthermore, a requested resource amount of the job J3 is 0.25. Moreover, the job J3 is a node-sharing job.

In this case, although the job J1 is deployed in the compute node #1, and a resource of 0.25 is actually left, the job J1 is a node-occupying job. Therefore, the job deployment destination determination unit113sets a free space of the compute node #1 to zero. Furthermore, the job J2 is deployed in the compute nodes #2 and #3, and all the resource amount of the compute node #2 is used by the job J2. Therefore, the job deployment destination determination unit113sets a free space of the compute node #2 to zero. On the other hand, the job J3 is also deployed in the compute node #3, and the job deployment destination determination unit113sets 0.25 that is a value obtained by subtracting a remaining requested resource amount of the job J2 and a requested resource amount of the job J3 from the entire resource amount as a free space of the compute node #3. Then, because both of the jobs J2 and J3 are node-sharing jobs, the job deployment destination determination unit113sets the free space of the compute node #3 to 0.25. Furthermore, regarding a compute node10such as a compute node #4 where a job is not deployed, the job deployment destination determination unit113sets a free space to one.

Next, the job deployment destination determination unit113calculates a value obtained by subtracting a resource amount in use from the entire resource amount of each rack20using the free space of each compute node10. Then, the job deployment destination determination unit113calculates fi that is an arrangement possibility index of each rack20using the formula (2).

Then, the job deployment destination determination unit113determines a rack20with the smallest fi as a deployment destination rack20. Next, the job deployment destination determination unit113treats the plurality of compute nodes10mounted on the deployment destination rack20as a bin packing problem for packing a new job in each compute node10having the calculated free space and determines a compute node10that is a new job deployment destination.

As described above, the cluster system according to the present embodiment calculates a free space of each compute node, based on a resource requirement condition of each job and determines a deployment destination rack and compute nodes using the calculated free space. As a result, even in a case where the jobs share the node, a probability that a compute node group having a close relationship is secured at the time of input of a new job is improved, and the number of communication hops can be reduced. Therefore, communication cost can be reduced.