Effective backup of data used by multiple nodes executing parallel processing

When barrier synchronization is executed between multiple nodes that include a node and execute processing for a same job, the node transmits first data indicating an intermediate result of processing of the node to another node with which processing of the node is to be synchronized first among the multiple nodes, and receives second data indicating an intermediate result of processing of the other node from the other node. The node stores the first data of the node in a first memory region of a memory provided for the node, and store the second data of the other node in a second memory region of the memory.

CROSS-REFERENCE TO RELATED APPLICATION

This application is based upon and claims the benefit of priority of the prior Japanese Patent Application No. 2018-198773, filed on Oct. 22, 2018, the entire contents of which are incorporated herein by reference.

FIELD

The embodiment relates to effective backup of data used by multiple nodes executing parallel processing.

BACKGROUND

A job, which is executed in a system with multiple data processing devices (nodes) coupled to each other via a network or is executed in a system used in the high performance computing (HPC) field, has a characteristic in which large-scale data is processed by many nodes for a long time period. When the processing is to be stopped due to a hardware error or the like, intermediate processing results are not guaranteed. Thus, an operation may be executed so that intermediate results are periodically backed up.

As a related technique, for example, there is a technique for using previously captured data to establish a state corresponding to a new checkpoint when a map of data written to a basic memory element is captured by a remote checkpoint memory element during a normal process and the new checkpoint is requested. In addition, there is a technique for acquiring job management information including a node number conversion table indicating an association relationship between a logical node number and a physical node number, and process management information including a logical node number, and generating a restart file for restarting a job.

An example of related art is Japanese National Publication of International Patent Application No. 2000-501216. Another example of related art is Japanese Laid-open Patent Publication No. 2011-186606.

SUMMARY

According to an aspect of the embodiments, a node includes a memory including a first memory region and a second memory region. When barrier synchronization is executed between multiple nodes that include the node and execute processing for a same job, transmit first data indicating an intermediate result of the processing of the node to another node with which processing of the node is to be synchronized first among the multiple nodes, and receive second data indicating an intermediate result of processing of the other node from the other node. The node stores the first data of the node in the first memory region of the memory, and store the second data of the other node in the second memory region of the memory.

DESCRIPTION OF EMBODIMENTS

The related techniques have a problem that the length of time it takes to back up intermediate processing results of nodes during parallel execution of a job by the multiple nodes increases.

It is desirable to reduce the length of time taken to execute backup.

Hereinafter, an embodiment of a parallel processing device disclosed herein, a parallel computation execution program disclosed herein, and a backup method disclosed herein is described with reference to the drawings.

Embodiment

First, a configuration of a processing system100according to the embodiment is described below.

FIG. 1is an explanatory diagram illustrating an example of the configuration of the processing system100. Referring toFIG. 1, the processing system100includes nodes N0to Nn (n is a natural number of 1 or more), a redundant node RN (two redundant nodes in the example illustrated inFIG. 1), a managing server101, a storage server102, and a login server103. In the processing system100, the nodes N0to Nn, the redundant node RN, the managing server101, the storage server102, and the login server103are coupled to and able to communicate with each other via a network110. The network110is, for example, a local area network (LAN), a wide area network (WAN), the Internet, or the like.

Each of the nodes N0to Nn is a computer for executing a computation process and is an example of a parallel processing device. The nodes N0to Nn execute computation processes while communicating with each other via, for example, an interconnect. The nodes N0to Nn are, for example, servers. Each of the nodes N0to Nn may be enabled by a virtual machine activated in a physical server.

In the following description, an arbitrary node among the nodes N0to Nn is referred to as “node Ni (i=0, i, . . . , n) in some cases.

The redundant node RN takes over and executes a computation process of a node Ni when the computation process of the node Ni is stopped due to a hardware failure or the like. The managing server101is a computer that controls the execution of a job and handles an error when the error occurs during the execution of the job.

The storage server102is a computer that has external storage120and stores data and a program that are used to execute the job. For example, the storage server102loads the data and the program into each of the nodes Ni in order to start the execution of the job. When the execution of the job is terminated, the storage server102saves a result of executing the job.

The login server103is a computer that is used by a user of the processing system100after login by the user. The login server103generates and compiles the program. The compiled program and initial data are stored in the storage server102from the login server103via, for example, the interconnect.

In the example illustrated inFIG. 1, the managing server101, the storage server102, and the login server103are different computers, but are not limited to this. For example, the managing server101, the storage server102, and the login server103may be enabled by a single server or a single node Ni. The nodes Ni and the servers101to103may communicate with each other via a different network from the network between the nodes so that the communication does not affect the computation processes of the nodes Ni.

When a process is to be stopped due to a hardware error or the like during the execution of the job by multiple nodes, a result of the process may not be guaranteed. In this case, when a node in which the hardware error or the like has occurred is removed and the job is executed again from the beginning, all resources (the number of nodes×an execution time) used until the occurrence of the error may become pointless.

When the job is to be executed to process a larger amount of data by a larger number of nodes for a longer time period, a loss that occurs when the processing is stopped is larger. A general-purpose device causes intermediate results of nodes to be periodically stored in external storage. When a hardware error occurs, the general-purpose device restarts a job from an intermediate state of the job to reduce an amount of a process to be re-executed (so-called checkpoint restart function).

However, when the checkpoint restart function is applied to the HPC field, intermediate results are written from many nodes (for example, several tens of thousands of nodes) to storage located at a single location or storage located at some locations in the case where striping is used. Thus, access is concentrated to the storage, it takes a longer time to write (back up) intermediate results to the storage than a processing time of the job, and system performance may be reduced.

Thus, the embodiment describes a backup method of executing mutual backup between nodes forming a pair for every execution of barrier synchronization to inhibit access from being concentrated to the storage when periodical backup is executed and reduce the length of time it takes to execute the backup. An example of a process by the processing system100is described with reference toFIG. 2.

FIG. 2is an explanatory diagram illustrating an example of the backup method according to the embodiment.FIG. 2illustrates the nodes N0to N3and the redundant node RN. This example assumes that the nodes N0to N3execute the same job in parallel (n=3, total 4 nodes). Each of the nodes Ni executes a computation process while executing a synchronization process by a barrier synchronization mechanism. InFIG. 2, circles indicate states of the nodes N0to N3.

Barrier synchronization is one of synchronization methods of synchronizing progress statuses of processes (threads or processes) executed in parallel. A barrier synchronization point at which synchronization is established is set for the barrier synchronization. For example, when a process of executing the barrier synchronization reaches the barrier synchronization point, the process is temporarily stopped. When all processes executed in parallel reach the barrier synchronization point, the stopped processes are restarted. Thus, the multiple processes that are executed in parallel are synchronized with each other.

In an example of the embodiment, the barrier synchronization may be implemented using the butterfly barrier. The butterfly barrier is a method of dividing a synchronization process into multiple stages and causing each of processes to transmit and receive signals with another process among the processes in each of the stages to synchronize the processes with each other. Barrier synchronization implemented using the butterfly barrier has been disclosed in, for example, Japanese Laid-open Patent Publication No. 2010-122848.

In the case where the barrier synchronization is executed between the nodes N0to N3, the nodes N0and N1are synchronized with each other first and the nodes N2and N3are synchronized with each other first (in a barrier stage1). Thus, the nodes N0and N1are a pair to be synchronized first. Similarly, the nodes N2and N3are a pair to be synchronized first.

In this case, the node N0transmits information of the concerned node N0to the node N1and receives information of the node N1from the node N1. The information of each of the nodes is, for example, checkpoint data including an intermediate result of the node. Then, the node N0stores the transmitted information of the concerned node N0in a first memory region of a memory device201of the concerned node N0and stores the received information of the node N1in a second memory region of the memory device201.

In addition, the node N1transmits the information of the concerned node N1to the node N0and receives the information of the node N0from the node N0. Then, the node N1stores the transmitted information of the concerned node N1in a first memory region of a memory device202of the concerned node N1and stores the received information of the node N0in a second memory region of the memory device202.

Similarly, the node N2transmits information of the concerned node N2to the node N3and receives information of the node N3from the node N3. Then, the node N2stores the transmitted information of the concerned node N2in a first memory region of a memory device203of the concerned node N2and stores the received information of the node N3in a second memory region of the memory device203.

In addition, the node N3transmits the information of the concerned node N3to the node N2and receives the information of the node N2from the node N2. Then, the node N3stores the transmitted information of the concerned node N3in a first memory region of a memory device204of the concerned node N3and stores the received information of the node N2in a second memory region of the memory device204.

Next, the nodes N0and N2are synchronized with each other and the nodes N1and N3are synchronized with each other (barrier stage2). By executing this, the nodes N0to N3are synchronized with each other. In a stage (for example, the barrier stage2) other than the barrier stage1, the checkpoint data of the nodes is not transmitted and received between the nodes.

As described above, when the barrier synchronization is executed, checkpoint data of each of the pairs of nodes N0to N3is held between nodes that form the pair and are to be synchronized with each other in the barrier stage1. Thus, it may be possible to inhibit access from being concentrated to the storage in the periodical backup during the execution of the job and reduce the length of time it takes to back up the checkpoint data (intermediate results) of the nodes N0to N3.

When a hardware error or the like occurs, the job may be restarted from an intermediate state of the job using the checkpoint data mutually held between the nodes. For example, when a hardware error occurs in the node N1, the job is restarted from an intermediate state of the job by migrating the information (checkpoint data) of the node N1, stored in the memory device201in the node N0, from the node N0to the redundant node RN.

According to the processing system100, even when the program is to be stopped due to a hardware failure or the like, the program is not executed again from the beginning and is executed (restarted) again from an intermediate state of the program by implementing a function of saving the intermediate state (checkpoint) and causing the redundant node prepared in advance to take over a process of the node in which the hardware failure has occurred.

Example of Hardware Configuration of Node Ni

FIG. 3is a block diagram illustrating an example of a hardware configuration of a node Ni. InFIG. 3, the node Ni includes a central processing unit (CPU)301, a main memory302, and a disk303. The CPU301includes a main interface304, an input and output (IO) interface305, a barrier device306, and an internode interface307. The constituent units304to307are coupled to each other via a bus300.

The CPU301controls the entire node Ni. The CPU301may include a plurality of cores. The main memory302includes, for example, a read only memory (ROM), a random-access memory (RAM), and a flash ROM. The CPU301accesses the main memory302via the memory interface304. For example, various programs are stored in the flash ROM and the ROM, and the RAM is used as a work area of the CPU301. The programs stored in the main memory302are loaded into the CPU301and cause the CPU301to execute a coded process.

The disk303is a memory device for storing various information. For example, the disk303may be a hard disk drive (HDD), a solid state drive (SSD), or the like. The CPU301accesses the disk303via the IO interface305.

The disk303has a small storage region310. The small storage region310includes a concerned node memory region311and another node memory region312. The small storage region310has, for example, a memory capacity that is up to twice as large as the memory capacity of the main memory302. The concerned node memory region311corresponds to any of the first memory regions described with reference toFIG. 2. The other node memory region312corresponds to any of the second memory regions described with reference toFIG. 2. The disk303is an example of any of the memory devices201to204illustrated inFIG. 2.

The barrier device306executes the barrier synchronization. For example, the barrier device306transmits information of a state of the concerned node Ni to another node via the interconnect, receives information of a state of the other node from the other node, and detects how much an entire computation process is progressed.

The internode interface307is coupled to the network110via a communication line, and further coupled to another computer via the network110. In addition, the internode interface307functions as an interface between the network110and the inside of the concerned node Ni, and controls input and output of data from and to the other computer.

The barrier device306may be installed in a unit different from the CPU301or may be enabled by software. The small storage region310may be included in the main memory302, a memory card (not illustrated), or the like. Each of the redundant node RN, the managing server101, the storage server102, and the login server103, which are illustrated inFIG. 1, may be enabled by a hardware configuration that is the same as or similar to the hardware configuration of the node Ni. Each of the managing server101, the storage server102, and the login server103, however, may not include the barrier device306.

Example of Functional Configuration of Node Ni

FIG. 4is a block diagram illustrating an example of a functional configuration of the node Ni. InFIG. 4, the node Ni includes a communication control unit401, a storage unit402, a detecting unit403, and a restart control unit404. The communication control unit401, the storage unit402, the detecting unit403, and the restart control unit404are functions serving as a control unit and are enabled by, for example, causing the CPU301to execute a program stored in a memory device, such as the main memory302illustrated inFIG. 3, the disk303illustrated inFIG. 3, or the like. Processing results of the functional units401to404are stored, for example, in a memory device, such as the main memory302or the disk303.

When the barrier synchronization is executed between the nodes N0to Nn for executing the same job, the communication control unit401transmits information of the concerned node Ni to another node, which is among the nodes N0to Nn and is to be synchronized with the concerned node Ni first, and receives information of the other node from the other node. In this case, the information of the nodes Ni is intermediate results of the nodes Ni or is checkpoint data including results of computation processes executed by the nodes Ni before the execution of the barrier synchronization.

The other node is among the nodes N0to Nn and is a node Nj (j≠i, j=0, 1, . . . , n) among nodes included in a group to be synchronized with the concerned node Ni first. The group to be synchronized with the concerned node Ni first is to be synchronized in the first stage (barrier stage1) of the barrier synchronization. The number of nodes included in the group to be synchronized with the concerned node Ni first is 2 or more and may be an even number or an odd number.

An example of operations of nodes Ni in the case where the number of nodes that are among the nodes N0to Nn and are included in the group to be synchronized with the concerned node first is “2” is described later with reference toFIG. 5. An example of operations of nodes Ni in the case where the number of nodes that are among the nodes N0to Nn and are included in the group to be synchronized with the concerned node first is “3” is described later with reference toFIG. 7.

In the following description, another node that is among the nodes N0to Nn and is to be synchronized with the concerned node first is referred to as “other node forming a pair with the concerned node in some cases. Information of the other node to be synchronized with the concerned node first is stored, for example, in a node ID table (not illustrated) in advance. The node ID table is enabled, for example, by a memory device, such as the main memory302or the disk303.

For example, when a process, executed by the concerned node, of the job reaches the barrier synchronization point during the execution of the job, the communication control unit401synchronizes the concerned node with the other node forming the pair with the concerned node. For example, the communication control unit401uses the barrier device306illustrated inFIG. 3to notify the states of the nodes between the nodes, thereby synchronizing the concerned node with the other node forming the pair with the concerned node.

Then, when the synchronization with the other node forming the pair with the concerned node is completed, the communication control unit401reads data from the main memory302and transmits the read data as checkpoint data to the other node. In addition, after the synchronization with the other node forming the pair with the concerned node is completed, the communication control unit401receives checkpoint data from the other node.

The checkpoint data of the other node is read from the main memory302of the other node. The data to be read may be all data within the main memory302or may be data stored in the main memory302and excluding data stored in a system region of the main memory302.

When processes, executed by nodes Ni, of the job reach the barrier synchronization point, the processes, executed by the nodes Ni, of the job are stopped. When the barrier synchronization is established between the nodes N0to Nn, the stopped processes of the job are restarted in the nodes Ni.

The storage unit402causes the information, transmitted by the communication control unit401, of the concerned node to be stored in the first memory region within the memory device included in the concerned node. The storage unit402causes the information, received by the communication control unit401, of the other node to be stored in the second memory region within the memory device included in the concerned node.

For example, the storage unit402causes the transmitted checkpoint data (data within the main memory302included in the concerned node) of the concerned node to be stored in the concerned node memory region311of the small storage region310within the disk303illustrated inFIG. 3. In addition, the storage unit402causes the received checkpoint data (data within the main memory302included in the other node) of the other node to be stored in the other node memory region312of the small storage region310within the disk303.

Thus, when the barrier synchronization is executed between the nodes N0to Nn executing the same job, data (checkpoint data) within the main memory302of each of the nodes to be synchronized with each other in the first stage (barrier stage1) of the barrier synchronization is backed up in the main memory302of another node among the nodes.

The job that is executed by the nodes N0to Nn is divided and executed, for example, so that an execution time of a process by each of the nodes Ni after the division is longer than the length of time it takes to transmit checkpoint data (data of the main memories302) of each of the nodes Ni to another node among the nodes Ni.

The detecting unit403detects that a failure has occurred in another node that is among the nodes N0to Nn and is to be synchronized with the concerned node first. For example, the detecting unit403determines that a failure has occurred in a node among the nodes N0to Nn, in response to the reception of an ON ERROR cancellation trap from the managing server101.

The ON ERROR cancellation trap is information notifying the existence of the node in which the failure (unrecoverable error) has occurred. The ON ERROR cancellation trap includes information (for example, a node number) of the node in which the failure has occurred, and information (for example, physical node information) of the redundant node RN that takes over a computation process of the node in which the failure has occurred.

Then, the detecting unit403determines whether the node that is identified by the ON ERROR cancellation trap and in which the failure has occurred is the other node forming the pair with the concerned node. When the node in which the failure has occurred is the other node forming the pair with the concerned node, the detecting unit403detects that the failure has occurred in the other node.

When the detecting unit403detects that the failure has occurred in the other node that is among the nodes N0to Nn and is to be synchronized with the concerned node first, the communication control unit401transmits the information, stored in the second memory region within the memory device of the concerned device, of the other node to the redundant node RN that takes over the process of the other node.

For example, when a failure has occurred in the other node forming the pair with the concerned node, the communication control unit401reads the checkpoint data of the other node from the other node memory region312of the small storage region310included in the disk303. Then, the communication control unit401transmits the read checkpoint data of the other node to the redundant node RN identified from the ON ERROR cancellation trap. In this case, the communication control unit401uses the information (for example, physical node information) of the redundant node RN to update the node ID table (not illustrated).

In addition, the program of the job is loaded into the redundant node RN from the storage server102under control by the managing server101. As a result, the redundant node RN is able to take over the process of the other node in which the failure has occurred, and the redundant node RN is set in a node group for executing the job in parallel.

When a failure has occurred in the other node forming the pair with the concerned node, the communication control unit401may read the checkpoint data of the concerned node from the concerned node memory region311of the small storage region310. Then, the communication control unit401may transmit the read checkpoint data of the concerned node and the checkpoint data of the other node to the redundant node RN. Thus, the redundant node RN may hold the checkpoint data of the nodes Ni.

The restart control unit404restarts executing the job based on the information, stored in the first memory region within the memory device of the concerned node, of the concerned node. For example, the restart control unit404recognizes the redundant node RN for taking over the process of the other node in which the failure has occurred. The other node in which the failure has occurred is switched to the redundant node RN, and the node group for executing the same job in parallel is updated to a node group including the redundant node RN and the nodes, which are among the nodes N0to Nn and exclude the node in which the failure has occurred.

Then, the restart control unit404restores a restart point of the job based on the checkpoint data, stored in the concerned node memory region311of the small storage region310within the disk303, of the concerned node. Then, after the restoration of restart points is completed in all the nodes, the restart control unit404restarts executing the job by executing a process (routine) again during which an error has occurred.

In the aforementioned description, the communication control unit401uses the barrier device306to notify the states of the nodes between the nodes, thereby synchronizing the concerned node with the other node forming the pair with the concerned node. The communication control unit401, however, is not limited to this. For example, the communication control unit401may transmit the information of the concerned node to the other node at the time of the synchronization of the concerned node with the other node. The communication control unit401may determine that the synchronization with the other node has been completed, in response to the transmission of the information of the concerned node to the other node and the reception of the information of the other node from the other node.

For example, when the process, executed by the concerned node, of the job reaches the barrier synchronization point during the execution of the job, the communication control unit401reads data from the main memory302and transmits the read data as checkpoint data to the other node. Then, the communication control unit401may determine that the synchronization with the other node has been completed, in response to the transmission of the checkpoint data of the concerned node to the other node and the reception of the checkpoint data of the other node from the other node.

By executing this, the concerned node is synchronized with the other node forming the pair with the concerned node without notifying the states of the nodes between the nodes. The concerned node may determine the completion of the first stage (barrier stage1) of the barrier synchronization in response to the completion of the mutual holding of the checkpoint data between the nodes and reduce communication related to the confirmation of the synchronization between the nodes. The communication control unit401may determine that the synchronization with the other node has been completed when the checkpoint data of the concerned node is stored in the concerned node memory region311and the checkpoint data of the other node is stored in the other node memory region312.

In addition, the communication control unit401may transmit the information of the concerned node to the other node to be synchronized with the concerned node first and receive the information of the other node from the other node for every execution of the barrier synchronization. By executing this, checkpoint data may be mutually held between nodes to be synchronized with each other in the first stage of the barrier synchronization for every execution of the barrier synchronization during the execution of the job by the nodes N0to Nn.

In addition, the communication control unit401may transmit the information of the concerned node to the other node and receive the information of the other node from the other node for every execution of the barrier synchronization after a predetermined time period T elapses after the start of the execution of the job. Before the predetermined time period T elapses after the start of the execution of the job, even at the time of the execution of the barrier synchronization, checkpoint data of the nodes is not mutually held between the nodes. The predetermined time period T may be set to an arbitrary time period.

Since an advantage of backup executed within a certain time period after the start of the execution of the job is small, the mutual holding of checkpoint data between the nodes may be delayed. An example of operations in the case where checkpoint data is mutually held between nodes after the predetermined time period T elapses after the start of the execution of the job is described later with reference toFIG. 6.

Example of Operations of Node Ni

Next, an example of operations of a node Ni in the case where checkpoint data is mutually held in nodes included in a group to be synchronized first when the barrier synchronization is executed is described below. The following example is described below. In the example, the nodes included in the group to be synchronized first are the “nodes N0and N1”, and checkpoint data is mutually held between the two nodes.

FIG. 5is a first explanatory diagram illustrating an example of operations of the nodes Ni.FIG. 5illustrates the nodes N0and N1that are to be synchronized with each other first when the barrier synchronization is executed.FIG. 5omits illustration of the barrier devices306of the nodes N0and N1. The example of the operations of the nodes N0and N1in the case where the checkpoint data is mutually held between the nodes N0and N1is described below.

When a process, executed by the node N0, of the job reaches the barrier synchronization point during the execution of the job, the node N0is synchronized with the node N1forming a pair with the concerned node N0. When the synchronization of the node N0with the node N1is completed, the node N0reads data from the main memory302of the node N0and transmits the read data as checkpoint data of the concerned node N0to the node N1(refer to an arrow501illustrated inFIG. 5).

Similarly, when a process, executed by the node N1, of the job reaches the barrier synchronization point during the execution of the job, the node N1is synchronized with the node N0forming the pair with the concerned node Ni. When the synchronization of the node Ni with the node N0is completed, the node N1reads data from the main memory302of the node N1and transmits the read data as checkpoint data of the concerned node Ni to the node N0(refer to an arrow502illustrated inFIG. 5).

Then, the node N0stores the transmitted checkpoint data (data within the main memory302of the node N0) of the concerned node N0to be stored in the concerned node memory region311within the disk303of the node N0(refer to an arrow503illustrated inFIG. 5). In addition, the node N0stores the received checkpoint data (data within the main memory302of the node N1) of the node N1to be stored in the other node memory region312within the disk303of the node N0(refer to the arrow502illustrated inFIG. 5).

Similarly, the node N1stores the transmitted checkpoint data (data within the main memory302of the node N1) of the concerned node N1in the concerned node memory region311within the disk303of the node N1(refer to an arrow504illustrated inFIG. 5). In addition, the node N1stores the received checkpoint data (data within the main memory302of the node N0) of the node N0to be stored in the other node memory region312within the disk303of the node Ni (refer to the arrow501illustrated inFIG. 5).

The length of time it takes to mutually hold the checkpoint data between the nodes N0and N1is described below. The following case assumes that each of the main memories302of the nodes N0and N1has a memory capacity of “64 GB” and that a bandwidth between the CPU301and the disk303in each of the nodes N0and N1is “100 MB/sec”. In addition, it is assumed that a main memory bandwidth and a network bandwidth are larger than the bandwidth between the CPU301and the disk303in each of the nodes N0and N1.

In this case, the length of time it takes to mutually hold the checkpoint data between the nodes N0and N1is approximately 22 minutes (=64 GB×2/100 MB/sec/60=21.33 (minutes)+latency (α). It is, therefore, desirable that a time interval between the transmission of the checkpoint data of the node N0to the node N1and the transmission of the checkpoint data of the node Ni to the node N0be equal to or longer than (22+a) minutes.

The example assumes that the nodes N0and N1transmit all the data within the main memories302to each other. However, only data (arrays) to be used for the processes of the nodes N0and N1may be extracted and stored in the small storage regions310of the nodes N1and NO. Thus, the length of time it takes to mutually hold the checkpoint data between the nodes N0and N1may be reduced.

Case where Checkpoint Data is Mutually Held after Predetermined Time Period T Elapses

Next, an example of operations in the case where checkpoint data is mutually held between nodes after the predetermined time period T elapses after the start of the execution of the job is described below.

FIG. 6is a second explanatory diagram illustrating an example of operations of a node Ni.6-1and6-2illustrated inFIG. 6indicate the length of time it takes to execute data processing by the node Ni and the length of time it takes to execute each saving process by the node Ni. The data processing corresponds to a computation process, executed by the node Ni, of the job. Each of the saving processes corresponds to a process, to be executed by the node Ni, of mutually holding checkpoint data between the node Ni and another node.

6-1indicates an example of operations in the case where the checkpoint data is mutually held between the nodes for every execution of the barrier synchronization during the execution of the job. For example, every time data processing (i), (ii), and (iii) is completed, saving processes (a), (b), and (c) is executed.

6-2indicates an example of operations in the case where the checkpoint data is mutually held between the nodes for every execution of the barrier synchronization after the predetermined time period T elapses after the start of the execution of the job. In this case, the predetermined time period T elapses during the execution of the data processing (ii). In this case, when the execution of the data processing (i) is completed, a saving process is not executed. Every time the execution of the data processing (ii) and (iii) is completed, the saving processes (a) and (b) is executed.

For example, it is assumed that it takes 24 hours to complete an entire process, a saving process is executed at time intervals of 1 hour (or the barrier synchronization is executed at time intervals of 1 hour), and it takes 10 minutes to completely execute each saving process, as indicated by6-1. In this case, the length of time it takes to complete the job is “27.8 hours=24×60+23×10=1670 minutes”.

In addition, it is assumed that the predetermined time period T is “10 hours” and that the checkpoint data is mutually held between the nodes for every execution of the barrier synchronization after 10 hours elapse after the start of the execution of the job, as indicated by6-2. In this case, the length of time it takes to complete the job is “25.8 hours that is nearly equal to 24×60+11×10=1550 minutes”.

Before the predetermined time period T elapses after the start of the execution of the job, even at the execution of the barrier synchronization, the checkpoint data of the nodes is not mutually held between the nodes. Thus, the length of time it takes to complete the execution of the job may be reduced.

Example of Operations in Case Where Number of Nodes Included in Each Group to be Synchronized First is “3”

An example of operations of the nodes Ni in the case where the number of nodes that are among the nodes N0to Nn and are included in each group to be synchronized first is “3” is described below.

FIG. 7is a third explanatory diagram illustrating the example of the operations of the nodes Ni.FIG. 7illustrates the nodes N0to N8. This example assumes that the nodes N0to N8execute the same job in parallel (n=8). It is assumed that groups to be synchronized first are a group of the “nodes N0, N1, and N2”, a group of the “nodes N3, N4, and N5”, and a group of the “nodes N6, N7, and N8”. InFIG. 7, circles indicate the states of the nodes N0to N8.

In the case where the barrier synchronization is executed between the nodes N0to N8, the nodes N0, N1, and N2are synchronized with each other, the nodes N3, N4, and N5are synchronized with each other, and the nodes N6, N7, and N8are synchronized with each other (in the barrier stage1). An example of operations for mutually holding checkpoint data between the nodes NO, N1, and N2included in the group is described below.

For example, the node N0transmits the checkpoint data of the concerned node N0to the node N1and receives checkpoint data of the node N2from the node N2in order to synchronize the nodes N0, N1, and N2with each other. Then, the node N0stores the transmitted checkpoint data of the concerned node N0in the concerned node memory region311of the node N0and stores the received checkpoint data of the node N2in the other node memory region312of the node N0.

In addition, for example, the node N1transmits the checkpoint data of the concerned node N1to the node N2and receives the checkpoint data of the node N0from the node N0in order to synchronize the nodes N0, N1, and N2with each other. Then, the node N1stores the transmitted checkpoint data of the concerned node N1in the concerned node memory region311of the node N1and stores the received checkpoint data of the node N0in the other node memory region312of the node Ni.

In addition, for example, the node N2transmits the checkpoint data of the concerned node N2to the node N0and receives the checkpoint data of the node N1from the node N1in order to synchronize the nodes N0, N1, and N2with each other. Then, the node N2stores the transmitted checkpoint data of the concerned node N2in the concerned node memory region311of the node N2and stores the received checkpoint data of the node N1in the other node memory region312of the node N2.

In this manner, even when the number of nodes included in a group to be synchronized first is “3”, checkpoint data of the nodes may be mutually held between the nodes of the group. The configuration of the processing system100may be expanded to a configuration in which the number of nodes included in a group that is to be synchronized first when the barrier synchronization is executed is a number other than 2. Furthermore, the configuration may be expanded to a configuration in which the number of all the nodes is not an even number. The length of time it takes to synchronize 3 nodes with each other may increase, compared with a configuration in which the number of nodes included in a group to be synchronized first is “2”.

Another Example of Small Storage Region310

Next, another example of the small storage region310within the disk303included in each node Ni is described. An example in which the number of nodes that are among the nodes N0to Nn and included in a group to be synchronized first is “3” is described below.

FIG. 8is a block diagram illustrating another example of the small storage region310. InFIG. 8, the small storage region310includes a concerned node memory region311, another node memory region312-1, and another node memory region312-2. The small storage region310has the memory regions for storing checkpoint data of other nodes included in the group.

In this case, when the barrier synchronization is executed, the communication control unit401of the node Ni transmits information of the concerned node Ni to the other nodes included in the group and receives information of the other nodes from the other nodes. Then, the storage unit402causes the transmitted information of the concerned node Ni to be stored in the first memory region within the memory device of the concerned node Ni. In addition, the storage unit402causes the received information of the other nodes to be stored in the second memory regions included in the memory device of the concerned node Ni and corresponding to the other nodes.

Specific processing details of the node N0included in the group of the nodes N0, N1, and N2are described as an example. For example, when the process, executed by the node N0, of the job reaches the barrier synchronization point, the node N0reads data from the main memory302of the N0and transmits the read data as checkpoint data to the nodes N1and N2. In addition, after the process, executed by the node Ni, of the job and a process, executed by the node N2, of the job reach the barrier synchronization point, the node N0receives the checkpoint data of the nodes N1and N2from the nodes N1and N2.

Then, the node N0stores the transmitted checkpoint data (data within the main memory302of the concerned node N0) of the concerned node N0to be stored in the concerned node memory region311of the small storage region310of the concerned node N0. In addition, the node N0stores the received checkpoint data (data within the main memory302of the node N1) of the node N1to be stored in the other node memory region312-1of the small storage region310of the concerned node N0. In addition, the node N0stores the received checkpoint data (data within the main memory302of the node N2) of the node N2to be stored in the other node memory region312-2of the small storage region310of the concerned node N0.

Thus, the node N0holds the checkpoint data of the other nodes (nodes N1and N2) included in the group to be synchronized first. For example, even when hardware failures simultaneously occur in the nodes N1and N2included in the group, two redundant nodes RN are prepared and the job may be executed again based on the information stored in the small storage region310of the node N0.

The node N0may determine the completion of the first stage (barrier stage1) of the barrier synchronization when the checkpoint data of the nodes is mutually held between the nodes included in the group. Thus, the first stage (barrier stage1) of the barrier synchronization may be completed by executing data transfer once (the data transfer is executed twice in the example illustrated inFIG. 7).

Examples of Various Process Procedures of Processing System100

Next, examples of various process procedures of the processing system100are described. First, a procedure for a job execution request process by the login server103is described.

FIG. 9is a flowchart illustrating an example of the procedure for the job execution request process by the login server103. In the flowchart illustrated inFIG. 9, first, the login server103generates the program of the job in response to an input operation by a user (in step S901). Then, the login server103compiles the generated program (in step S902).

After that, the login server103causes the compiled program to be stored in the external storage120of the storage server102(in step S903). Then, the login server103transmits a request to execute the job to the managing server101(in step S904) and terminates the job execution request process of the flowchart. Thus, the execution of the job is requested to the managing server101.

Next, a procedure for a loading process by the storage server102.

FIG. 10is a flowchart illustrating an example of the procedure for the loading process by the storage server102. In the flowchart illustrated inFIG. 10, first, the storage server102determines whether the storage server102has received a request to execute loading to all the nodes N0to Nn from the managing server101(in step S1001).

The storage server102waits to receive the load request (No in step S1001). When the storage server102has received the load request (Yes in step S1001), the storage server102transmits, to all the nodes N0to Nn, the program and the data that are already stored in the external storage120and are used to execute the job (in step S1002).

After that, the storage server102determines whether the storage server102has received a result of executing the job from a node Ni (in step S1003). When the storage server102has not received the result of executing the job (No in step S1003), the storage server102determines whether the storage server102has received a request to execute loading to the redundant node RN from the managing server101(in step S1004).

When the storage server102has not received the request to execute the loading to the redundant node RN (No in step S1004), the storage server102causes the loading process to return to step S1003. When the storage server102has received the request to execute the loading to the redundant node RN (Yes in step S1004), the storage server102transmits, to the redundant node RN, the program that is already stored in the external storage120and is used to execute the job (in step S1005). After that, the storage server102causes the loading process to return to step S1003.

When the storage server102has received the result of executing the job from the node Ni (Yes in step S1003), the storage server102stores the received result of executing the job in the external storage120(in step S1006). Then, the storage server102terminates the loading process.

Thus, the storage server102may transmit, in response to a request from the managing server101, the program and the data to the nodes N0to Nn for executing the job in parallel and transmit the program to the redundant node RN for taking over a process of a node in which a hardware failure or the like has occurred.

Next, a procedure for an execution control process by the managing server101is described.

FIG. 11is a flowchart illustrating an example of the procedure for the execution control process by the managing server101. In the flowchart illustrated inFIG. 11, first, the managing server101determines whether the managing server101has received a request to execute the job from the login server103(in step S1101). The managing server101waits to receive the request to execute the job (No in step S1101).

When the managing server101has received the request to execute the job (Yes in step S1101), the managing server101transmits a request to execute loading to all the nodes N0to Nn to the storage server102(in step S1102). After that, the managing server101determines whether the execution of the job has been terminated (in step S1103).

When the execution of the job has not been terminated (No in step S1103), the managing server101determines whether the managing server101has received a fatal error from any node Ni among the nodes N0to Nn (in step S1104). The fatal error is a signal indicating that a hardware failure or the like has occurred.

When the managing server101has not received the fatal error (No in step S1104), the managing server101causes the execution control process to return to step S1103. When the managing server101has received the fatal error (Yes in step S1104), the managing server101prepares the redundant node RN (in step S1105).

After that, the managing server101transmits a request to execute loading to the prepared redundant node RN to the storage server102(in step S1106). Then, the managing server101transmits the ON ERROR cancellation trap to all nodes (in step S1107) and causes the execution control process to return to step S1103.

The ON ERROR cancellation trap notifies all the nodes that the redundant node RN is newly used. In this case, all the nodes are among the nodes N0to Nn and include the prepared redundant node RN and exclude the node Ni that has transmitted the fatal error.

When the execution of the job has been terminated (Yes in step S1103), the managing server101terminates the execution control process of the flowchart. By executing this, the execution of the job is controlled. For example, a failed node in which a hardware failure or the like has occurred during the execution of the job is removed, and the redundant node RN takes over a process of the failed node.

Next, a procedure for a parallel computation execution process by a node Ni is described. The following case is described below as an example. In the case, the number of nodes included in a group that is to be synchronized first when the barrier synchronization is executed during the execution of the job is “2”.

FIGS. 12 to 14are flowcharts illustrating the procedure for the parallel computation execution process by the node Ni. In the flowchart illustrated inFIG. 12, first, the node Ni receives, from the storage server102, the program and the data that are used to execute the job (in step S1201). Then, the node Ni uses the received program and the received data to start executing the job (in step S1202).

Next, the node Ni determines whether the node Ni has received the ON ERROR cancellation trap from the managing server101(in step S1203). When the node Ni has not received the ON ERROR cancellation trap (No in step S1203), the node Ni determines whether a process, executed by the node Ni, of the job has reached the barrier synchronization point (in step S1204).

When the process, executed by the node Ni, of the job has not reached the barrier synchronization point (No in step S1204), the node Ni causes the parallel computation execution process to return to step S1203. When the process, executed by the node Ni, of the job has reached the barrier synchronization point (Yes in step S1204), the node Ni waits to be synchronized with another node to be synchronized with the node Ni first (in step S1205).

Then, the node Ni determines whether the synchronization between the node Ni and the other node to be synchronized with the node Ni first has been completed (in step S1206). When the synchronization between the two nodes has not been completed (No in step S1206), the node Ni causes the parallel computation execution process to return to step S1205. When the synchronization between the two nodes has been completed (Yes in step S1206), the node Ni causes the parallel computation execution process to proceed to step S1301illustrated inFIG. 13.

In the flowchart illustrated inFIG. 13, the node Ni determines whether all computation processes of the job have been terminated (in step S1301). When one or more of all the computation processes of the job have not been terminated (No in step S1301), the node Ni transmits checkpoint data of the concerned node Ni to the other node completely synchronized with the concerned node Ni and receives checkpoint data from the other node (in step S1302). The checkpoint data of the concerned node Ni is read from the main memory302of the node Ni.

Then, the node Ni stores the transmitted checkpoint data of the concerned node Ni in the concerned node memory region311of the node Ni and stores the received checkpoint data of the other node in the other node memory region312of the node Ni (in step S1303). Then, the node Ni waits to be synchronized with all the other nodes (in step S1304).

Then, the node Ni determines whether the synchronization between the node Ni and all the other nodes has been completed (in step S1305). When the synchronization between the node Ni and all the other nodes has not been completed (No in step S1305), the node Ni causes the parallel computation execution process to return to step S1304. When the synchronization between the node Ni and all the other nodes has been completed (Yes in step S1305), the node Ni causes the parallel computation execution process to return to step S1203illustrated inFIG. 12.

When the synchronization between the node Ni and another node forming a pair with the node Ni is established, the barrier synchronization transitions to the phase in which checkpoint data of the two nodes is to be mutually held between the two nodes. After the checkpoint data is completely mutually held between the two nodes, the node Ni waits for the completion of the barrier synchronization of the node Ni with all the other nodes. After the synchronization of the node Ni with all the other nodes is completed, the node Ni may execute a next computation process.

When all the computation processes are terminated (Yes in step S1301), the node Ni waits to be synchronized with all the other nodes (in step S1306). Then, the node Ni determines whether the synchronization between the node Ni and all the other nodes has been completed (in step S1307).

When the synchronization between the node Ni and all the other nodes has not been completed (No in step S1307), the node Ni causes the parallel computation execution process to return to step S1306and waits for the completion of the synchronization between the node Ni and all the other nodes. When the synchronization between the node Ni and all the other nodes has been completed (Yes in step S1307), the node Ni transmits a result of executing the job to the storage server102(in step S1308). After that, the node Ni terminates the parallel computation execution process.

When the node Ni has received the ON ERROR cancellation trap (Yes in step S1203illustrated inFIG. 12), the node Ni causes the parallel computation execution process to proceed to step S1401illustrated inFIG. 14. When a failure, such as a hardware failure, occurs in the concerned node Ni, the node Ni transmits the fatal error to the managing server101. However, when the node Ni is not able to communicate due to a failure, another node forming a pair with the node Ni detects an abnormality of the node Ni using a method, such as existence monitoring, and transmits the fatal error to the managing server101.

In the flowchart illustrated inFIG. 14, first, the node Ni determines whether the concerned node Ni is the redundant node RN to be set on behalf of a failed node (in step S1401). When the node Ni is the redundant node RN (Yes in step S1401), the node Ni receives, from the storage server102, the program to be used to execute the job (in step S1402).

Then, the node Ni initializes a message passing interface (MPI) to form a node group (nodes N0to Nn) in which the failed node is replaced with the concerned node Ni and that execute the job (in step S1403). After that, the node Ni receives checkpoint data from the node that forms the pair with the node Ni and is to be synchronized with the node Ni first when the barrier synchronization is executed (in step S1404). The received checkpoint data is stored in the concerned node memory region311of the node Ni.

Then, the node Ni waits to be synchronized with all the other nodes (in step S1405). Then, the node Ni determines whether the synchronization between the node Ni and all the other nodes has been completed (in step S1406). The node Ni waits for the completion of the synchronization between the node Ni and all the other nodes (No in step S1406).

When the synchronization between the node Ni and all the other nodes has been completed (Yes in step S1406), the node Ni uses the checkpoint data stored in the concerned node memory region311of the node Ni to restart executing the job (in step S1407). Then, the node Ni causes the parallel computation execution process to return to step S1203illustrated inFIG. 12.

When the node Ni is not the redundant node RN (No in step S1401), the node Ni reads the checkpoint data from the concerned node memory region311of the small storage region310of the node Ni (in step S1408). After that, the node Ni updates the node ID table based on information of the redundant node RN (in step S1409).

After that, the node Ni transmits the read checkpoint data to the other node that is to be synchronized with the node Ni first when the barrier synchronization is executed (in step S1410). Then, the node Ni causes the parallel computation execution process to proceed to step S1405. When the node Ni does not form a pair with a failed node and is not the redundant node RN (No in step S1401), the node Ni may cause the parallel computation execution process to proceed to step S1405.

By executing this, checkpoint data may be mutually backed up between nodes to be synchronized with each other in the first stage of the barrier synchronization for every execution of the barrier synchronization during the execution of the job. When a node forming a pair with the node Ni fails, the node Ni may transmit the checkpoint data to the redundant node RN set in the group on behalf of the failed node and restart executing the job from an intermediate state of the job.

When the node Ni waits to be synchronized in steps S1205, S1206, S1304, S1305, S1306, and S1307and receives the ON ERROR cancellation trap, the node Ni causes the parallel computation execution process to return to step S1203and continuously executes the parallel computation execution process, although not illustrated inFIGS. 12 and 13. This is executed to inhibit the node Ni from being hung up in a state in which the node Ni waits to be synchronized.

Example of Processing System100

Next, an example of the processing system100is described using a specific example of a target problem to be computed by many nodes.

FIG. 15is an explanatory diagram describing the specific example of the target problem. An example in which a matrix vector illustrated inFIG. 15is computed by nodes N0to Np is described below. The following items (1) to (5), however, are assumed.

(2) The nodes N0to Np have a number “A=M×N” of elements.

(3) The multiple nodes N0to Np process each row (including a number M of elements) of the matrix vector, cause one node to collect all results, and cause the results to be stored in the storage server102.

(4) The nodes N0to Np execute the barrier synchronization for each row of the matrix vector. For example, even when a hardware failure occurs, the barrier synchronization is executed again for only one row.

(5) Data of A and x is loaded in all the nodes N0to Np in advance.

For example, the node N0executes computation on a[0:n][0] to a[0:n][2], the node N1executes computation on a[0:n][3] to a[0:n][5], and the node Np executes computation on a[0:n][m−2] to a[0:n][m].

An example of the program to be loaded into each of the nodes N0to Np is described below.

FIGS. 16A, 16B, 16C, 16D, 16E, 16F, 16G, 16H, and 16Iare explanatory diagrams illustrating the example of the program.FIGS. 16A to 16Iillustrate a program1600obtained by simplifying the program to be loaded into each of the nodes N0to Np.FIGS. 16A to 16Iillustrate a portion of the program1600. The following items <1> to <8> are assumed.

<1> Physical node information of the redundant node RN and a virtual node number of a failed node are transmitted in response to the ON ERROR cancellation trap from the managing server101.

<2> When a hardware failure occurs, the program loaded in the redundant node RN based on a request from the managing server101includes information of the redundant node RN. For example, when MPI communication is used, and a process of initializing an MPI is executed in the redundant node RN, a node group including the redundant node RN with which the failed node has been replaced is formed.

<3> copyarea is of the int type and is secured as a region.

<4> A function (On_error) of an error handler exists and is a function of enabling interruption and executing processing even during the time when a node waits for the barrier synchronization.

<5> A function (PairBarrier) of recognizing a termination status between two nodes forming a pair exists.

<6> A function (MutualCp) of executing mutual holding between two nodes forming a pair exists.

<7> A function (IntermediateCP) of copying a concerned node memory region exists.

<8> A function (PhysicalNode) of acquiring physical node information of a concerned node exists.

FIGS. 17A and 17Bare explanatory diagrams illustrating an example of memory maps.FIGS. 17A and 17Billustrate memory maps of nodes Ni and N(i+1). In the example illustrated inFIGS. 17A and 17B, a small storage region exists in a range from 0x80000000 in each of the memory maps.

It is assumed that a system region (OS region) and a program region exist in a range from 0x00000000 to 0x3FFFFFFF in each of the memory maps and that an area to be used for computation exists in a range from 0x40000000 to 0x7FFFFFFF for the number CA of elements in each of the memory maps. The area has a (2) table region, a (3) data region y, a (4) data region A, and a (5) data region x in each of the memory maps. A computation result is stored in the (3) data region y.

The small storage region has a data backup region in a range from 0x80000000 to 0xBFFFFFFF for the concerned node and a backup region in a range from 0xC0000000 to 0xFFFFFFFF for the node forming the pair with the concerned node. The number of elements of each of the backup regions is CA.

The (2) table region has a table region for a barrier point number and virtual-to-actual node conversion. The barrier point number indicates a row on which matrix computation has been executed last. Since the (2) table region has a table for the virtual-to-actual node conversion, the redundant node RN is used without a change in the program, and the (2) table region does not affect the re-execution of the program.

A portion1601illustrated inFIGS. 16E and 16Fis a main routine of the matrix computation. Matrix elements of each row are divided into elements including elements of ist and ien and are computed. The computation of rows to be processed is progressed using line. Final results are obtained using an MPI_Allreduce function. When a hardware failure does not occur, a while loop is repeatedly executed to execute the computation until the number of elements of y is equal to N.

A portion1602illustrated inFIGS. 16G, 16H, and 16Iis a routine of an On_error process. When a hardware failure occurs in a certain node, the certain node or another node forming a pair with the certain node detects an abnormality using a method, such as existence monitoring, and notifies the managing server101that the hardware failure has occurred. When the managing server101detects the hardware failure, the managing server101prepares the redundant node RN and issues the ON ERROR cancellation trap to a node group (combination indicated by MPI_COMM_WORLD in the program) of nodes that operate in parallel.

Most of the nodes that have received the ON ERROR cancellation trap wait for the barrier synchronization and are in a loop state, but execute the On_error function in an interrupt process. The On_error function causes the redundant node RN and each of nodes (normal nodes), in which a hardware failure does not occur, to operate in different manners.

The program is loaded into the redundant node RN from the storage server102, and the redundant node RN executes the initialization process to use an MPI function, receives data of copyarea_pair from a node forming a pair with the redundant node RN. It is assumed that the program that includes the information of the redundant node RN is loaded into the redundant node RN from the managing server101, the MPI is initialized in a state in which a failed node is replaced with the redundant node RN at the time of the execution of MPI_init, and copyarea_pair is received via the MPI communication (the assumption <2>). After that, the redundant node RN enters the same routine of waiting for the barrier synchronization as the other node, and transitions to a state of waiting for the barrier synchronization.

Each of the normal nodes copies an area of copyarea_my of the concerned node to basearea, and replaces, based on trap information (ON ERROR cancellation trap), a physical node number of a virtual node in which a hardware failure has occurred with a physical node number of the redundant node RN. When a hardware failure occurs in a node forming a pair with a normal node, the normal node transmits copyarea_pair stored in small storage to the redundant node RN and transitions to a state of waiting for the barrier synchronization after the transmission of copyarea_pair.

The nodes transition out of barrier synchronization states in order from a node that recognizes that all the nodes have entered a state of waiting for the barrier synchronization, and the nodes execute the program again from the top of a while statement. Since line indicating a row to be processed is not updated at the time of the execution of MPI_Barrier within the while statement, a value included in basearea is used for line without a change in the value.

As described above, a node Ni according to the embodiment transmits checkpoint data of the concerned node Ni to another node, which is among the nodes N0to Nn and to be synchronized with the concerned node Ni first, and receives checkpoint node of the other node from the other node when the barrier synchronization is executed between the nodes N0to Nn for executing the same job. Then, the node Ni stores the transmitted checkpoint data of the concerned node Ni in the concerned node memory region311within the disk303of the concerned node Ni and stores the received checkpoint data of the other node in the other node memory region312within the disk303. The checkpoint data of the node Ni is information stored in the main memory302of the node Ni. The checkpoint data of the other node is information stored in the main memory302of the other node.

At the time of the execution of the barrier synchronization between the nodes N0to Nn executing the same job, data (checkpoint data) within the main memories302of nodes to be synchronized with each other in the first stage (barrier stage1) of the barrier synchronization may be backed up between the nodes.

In addition, when the barrier synchronization is executed, a node Ni transmits checkpoint data of the concerned node Ni to one or more other nodes, which are among the nodes N0to Nn and are included in a group to be synchronized with the concerned node Ni first, and receives checkpoint data of the one or more other nodes included in the group.

Thus, even when the number of nodes included in the group to be synchronized first in the first stage of the barrier synchronization is “3”, checkpoint data of the nodes may be mutually held between the nodes included in the group. Since each pair of nodes is formed so that checkpoint data of each of nodes Ni included in a group is held in another node included in the group, it may be possible to inhibit an increase in a memory capacity to be used to back up checkpoint data of each of the nodes Ni.

In addition, when the barrier synchronization is executed, a node Ni transmits checkpoint data of the concerned node Ni to each of other nodes, which are among the nodes N0to Nn and are included in a group to be synchronized with the concerned node Ni first, and receives checkpoint data of the other nodes included in the group. Then, the node Ni stores the transmitted checkpoint data of the concerned node Ni in the concerned node memory region311within the disk303of the concerned node Ni and stores the received checkpoint data of the other nodes in the other node memory regions312(for example, the other node memory regions312-1and312-2illustrated inFIG. 8) within the disk303.

Thus, the node Ni holds the checkpoint data of the other nodes (nodes N1and N2) included in the group to be synchronized in the first stage of the barrier synchronization. Thus, even when hardware failures simultaneously occur in multiple nodes included in a group, the job may be executed again from an intermediate state of the job based on the information stored in the small storage region310of a normal node included in the group.

In addition, a node Ni may transmit checkpoint data of the concerned node Ni to another node at the time of the synchronization of the node Ni with the other node or when a process, executed by the node Ni, of the job reaches the barrier synchronization point. Then, the node Ni may determine that the synchronization with the other node has been completed in response to the transmission of checkpoint data of the concerned node Ni to the other node and the reception of checkpoint data of the other node from the other node.

Thus, the completion of the first stage of the barrier synchronization may be determined when the checkpoint data is mutually held between the nodes. Thus, the nodes may not notify the states of the nodes to each other when the synchronization is executed between the nodes, and it may be possible to reduce communication to be executed to confirm the synchronization between the nodes.

In addition, a node Ni may transmit checkpoint data of the concerned node Ni to another node and receive checkpoint data of the other node from the other node for every execution of the barrier synchronization after the predetermined time period T elapses after the start of the execution of the job.

Since an advantage of backup within a certain time period after the start of the execution of the job is small, the mutual holding of checkpoint data between the nodes may be delayed.

In addition, when a failure occurs in the other node, the node Ni may transmit the checkpoint data, stored in the other node memory region312within the disk303of the node Ni, of the other node to the redundant node RN for taking over a process of the other node. Then, the node Ni may restart executing the job based on the checkpoint data, stored in the concerned node memory region311within the disk303of the concerned node Ni, of the concerned node Ni.

Thus, even when a failure occurs in the other node to be synchronized with the concerned node in the first stage of the barrier synchronization, a process (routine during which an error has occurred) may be executed again using checkpoint data at the time of successful execution of the barrier synchronization between all the nodes, and the job may be restarted. Thus, even when a node is not available due to a hardware failure or the like, a process may be returned to a state when or before a periodical saving process has been previously executed. For example, it is assumed that a job that executes an entire process for a time period of 24 hours is divided into sub-jobs, each of which executes a process for a time period of 1 hour. A penalty when the backup method according to the embodiment is not used is a time period for which the job is executed before the occurrence of a hardware failure. The penalty when the backup method according to the embodiment is not used is a time period of up to 24 hours. On the other hand, a penalty when the backup method according to the embodiment is used is a time period of “(up to) 1 hour+23×the length of time it takes to mutually hold checkpoint data+α (the length of time it takes to prepare the redundant node RN by the managing server101or the like)”.

For example, it is assumed that it takes 24 hours to complete the entire process, a saving process is executed at time intervals of 1 hour (or the barrier synchronization is executed at time intervals of 1 hour), and it takes 10 minutes to completely execute each saving process. In this case, the length of time it takes to complete the job is “27.8 hours=24×60+23×10=1670 minutes”.

In addition, it is assumed that a hardware failure occurs after 10 hours elapse after the start of the execution of the job once in three times. In this case, in the processing system100, the length of time it takes to execute 6 programs (jobs) is “169.0 hours (=1670×6+60×(6/3)=10,140 minutes)”.

In an existing processing system, the length of time it takes to execute the 6 programs (jobs) is “187.0 hours (=1670×6+600×(6/3)=11,220 minutes)” (the length of time, however, is calculated using the penalty as 10 hours). In this case, the length of time to execute the 6 programs (jobs) in the processing system100is shorter than that in the existing processing system and has an advantage.

In addition, the length of time it takes to mutually hold checkpoint data in the embodiment is compared with that in a related method for a system including one storage device under the following assumptions (1) to (5).

(1) The amount of data mutually held per node is 8 MB.

(2) The number of nodes is 100.

(3) A communication bandwidth is larger than an IO bandwidth, and transmission and reception may be simultaneously processed.

(4) An IO bandwidth of the storage device used in the related method is 1 GB/sec by using a striping technique.

(5) The IO interface according to the embodiment is implemented in each of the nodes and is configured using a disk of 100 MB/sec, and an IO bandwidth according to the embodiment is 100 MB/sec.

In the related method, data of 8 MB for each of 100 nodes is stored in the storage device with a bandwidth of 1 GB/sec, and thus it takes 0.8 seconds (=8 MB×100/1 GB/sec). In the embodiment, data is mutually held between nodes forming a pair, and thus a data amount is 8 MB even when the 100 nodes are used. It takes 0.08 seconds (=8 MB×1/100 MB/sec) and the length of time it takes to execute the backup is 1/10 of that in the related method. In the embodiment, the storage of the system may not have a wide bandwidth, and thus the cost of the storage may be reduced.

Based on the above description, the processing system100according to the embodiment and the nodes Ni may inhibit access from being concentrated to the storage in the periodical backup during the execution of the job, reduce a load of the entire system, and reduce the length of time it takes to execute the backup.

The backup method described in the embodiment may be enabled by causing a computer, such as a personal computer or a workstation, to execute a program prepared in advance. The parallel computation execution program is recorded in a computer-readable recording medium, read by the computer from the recording medium, and executed by the computer. The computer-readable recording medium is a hard disk, a flexible disk, a compact disc-read only memory (CD-ROM), a magneto-optical (MO) disk, a digital versatile disc (DVD), a Universal Serial Bus (USB) memory, or the like. In addition, the parallel computation execution program may be distributed via a network, such as the Internet.

In addition, the parallel processing device (node Ni) described in the embodiment may be realized by an application specific IC, such as a standard cell or a structured application specific integrated circuit (ASIC), or a programmable logic device (PLD), such as a field-programmable gate array (FPGA).