Jean-Claude Charr
email: jean-claude.charr@univ-fcomte.fr

Raphaël Couturier
email: raphael.couturier@univ-fcomte.fr

David Laiymani
email: david.laiymani@univ-fcomte.fr

J.-C Charr

A decentralized and fault tolerant convergence detection algorithm for asynchronous iterative algorithms

Keywords: Decentralized global convergence detection mechanism, Peer-to-Peer environment, Distributed clusters, Fault tolerance

This article presents an algorithm that performs a decentralized detection of the global convergence of parallel asynchronous iterative applications. This algorithm is fault tolerant. It runs a decentralized saving procedure which enables this algorithm, after a node's crash, to replace the dead node by a new one which will continue the computing task from the last check point. Combined with the advantages of the asynchronous iteration model, this method allows us to compute very large scale problems using highly volatile parallel architectures like Peer-to-Peer and distributed clusters architectures. We also present the implementation of this algorithm in the JaceP2P platform which is dedicated to designing and executing parallel asynchronous iterative applications in volatile environments. Numerous experiments show the robustness and the efficiency of our algorithm.

and to understand their behaviors and consequences. There are two classes of methods for solving numerical problems: direct or iterative methods. Direct methods give the exact solution of a problem after a finite number of operations (we can cite for instance: the LU or the Cholesky method). On the other hand, iterative methods compute many times the same block of operations until obtaining a good approximation of the solution (here we can cite the Jacobi or the conjugate gradient method). We then say that the method has converged to the solution of the problem. For some types of problems, iterative methods are more efficient than direct ones and they could be the only way to solve some problems. Most of the time, natural phenomena are represented by large and complex numerical problems that cannot be solved using a single regular computing unit. Therefore, the resolution methods must be parallelized in order to be simultaneously executed on many computing units. Iterative methods are easier to parallelize than direct ones and when dealing with very large problems; they are preferred over direct methods (especially if they give a good approximation of the solution in a small number of iterations).

These properties have led to the large expansion of parallel iterative algorithms. One of the major problems of parallelizing an iterative algorithm is how to detect its convergence. This mechanism is necessary to stop the iterative process in these methods. In sequential iterative algorithms, this consists in ensuring that the solution vector is stable and does not vary between two successive iterations. In practice, a residual value is computed (for example, R = max i (|x k+1 i x k i |) where x k i denotes the value of the component i of array x at iteration k) and the system converges when R is smaller than a requested threshold.

The most common solutions for detecting the convergence of parallel iterative algorithms are both centralized and synchronous. For example, if a node detects that its subsystem has locally converged, it informs a server which is responsible for detecting if all the application's subsystems have converged. If so, the server declares that the application has globally converged. This centralized solution requires that all the nodes communicate with the server when they locally converge which can lead to a bottleneck if a large number of nodes locally converge at the same time. Moreover, this centralized solution is only compatible with parallel synchronous iterative algorithms. The parallel iterative algorithms can be classified in three main classes depending on how iterations and communications are managed. In the Synchronous Iterations-Synchronous Communications (SISC) model data are synchronously exchanged at the end of each iteration. All the computing units must begin the same iteration at the same time and important idle times are generated while waiting for other computing units to finish computing their iterations. The Synchronous Iterations-Asynchronous Communications (SIAC) model can be compared to the previous one except that data required on neighbors are sent asynchronously and without stopping current computations. This technique allows to partially overlap communications by computations, but unfortunately the overlapping is only partial and important idle times remain. Figure 1 illustrates two computing units executing a parallel iterative application implemented in the SIAC model. The grey blocks in Fig. 1 represent the computation phases, the white spaces represent the idle times and the arrows the asynchronous communications. It is clear that in a grid computing context, where the computational nodes are numerous, heterogeneous and widely distributed, the Fig. 1 Two processors computing in the Synchronous Iterations-Asynchronous Communications (SIAC) model Fig. 2 Two processors computing in the Asynchronous Iterations-Asynchronous Communications (AIAC) model: the iterations and the communications are not synchronized idle times generated by synchronizations are very penalizing. One way to overcome this problem is to use the Asynchronous Iterations-Asynchronous Communications (AIAC) model which is also known as the asynchronous iteration model [START_REF] Bahi | Asynchronism for iterative algorithms in a global computing environment[END_REF]. Using this model, local computations do not need to wait to receive data from their neighbors. The computing units can perform their iterations using the data present at that time. Figure 2 shows that the idle times are totally suppressed between iterations. With this algorithmic model, the number of iterations required before the convergence is usually greater than for the two former classes. But as detailed in [START_REF] Bahi | Performance comparison of parallel programming environments for implementing AIAC algorithms[END_REF][START_REF] Bertsekas | Parallel and distributed computation: numerical methods[END_REF][START_REF] Vuillemin | Calcul itératif asynchrone sur infrastructure pair-à-pair : la plate-forme JaceP2P[END_REF], the overall execution time is smaller if the numerical problem is being solved over heterogeneous and distant clusters. Since each computing unit can execute the next iteration without waiting for fresh data from neighbors, this model tolerates the loss of dependencies messages and moreover two nodes solving the same problem can execute different iterations at the same time. In Fig. 2, the first processor finishes the seventh iteration while the second processor is still executing the sixth iteration. These properties make this model unaffected by the crashes of computing nodes: if a node crashes, all the rest of the computing nodes may normally continue their tasks. The neighbors of the dead node do not have to wait for the reception of the dependencies messages from the dead node, and they can compute the next iterations using the last received dependency messages.

The centralized convergence detection mechanism, described above, requires a global synchronization at each recovery of the global state, which would indirectly synchronize the iterative process itself and would drastically reduce the ratio of asynchronism and its benefits. Moreover, in the asynchronous iteration model, the computing units do not necessarily receive new dependencies at each iteration. So, a node can compute many iterations without receiving any new data which could lead it to a false local convergence detection. Therefore, the centralized convergence detection mechanism cannot be used with the asynchronous iteration mode. Nevertheless, this problem has been studied for a long time and the first global convergence detection method for asynchronous parallel algorithms was introduced in [START_REF] Bertsekas | Convergence rate and termination of asynchronous iterative algorithms[END_REF]. Then a new method that modifies the iterative process and is based on very strict assumptions was proposed by Savari and Bertsekas in [START_REF] Savari | Finite termination of asynchronous iterative algorithms[END_REF]. Recently, a completely decentralized algorithm that does not modify the iterative process was introduced by Bahi et al. in [START_REF] Bahi | Parallel iterative algorithms: from to grid computing[END_REF][START_REF] Bahi | A decentralized convergence detection algorithm for asynchronous parallel iterative algorithms[END_REF]. This algorithm, we called Decentralized Convergence Detection (DCD), is based on the "Leader election" algorithm [START_REF] El-Ruby | Leader election in distributed computing systems[END_REF][START_REF] Antonoiu | A self-stabilizing leader election algorithm for tree graphs[END_REF] and it is well suited to asynchronous computing. However, this algorithm is not fault tolerant.

This article has two objectives. First of all, it presents a fault tolerant decentralized algorithm which detects the global convergence of a parallel iterative asynchronous application. It is an evolution of the DCD algorithm in which a decentralized automatic saving mechanism was added. This allows a new node to replace a dead one by retrieving its last backup and continuing the calculation from that checkpoint without disturbing the other nodes or interrupting the application. Secondly, it illustrates the implementation of this algorithm into the JaceP2P platform [START_REF] Bahi | JaceP2P: an environment for asynchronous computations on Peer-to-Peer networks[END_REF]. This platform offers a design and execution environment specifically dedicated to AIAC algorithms. Nevertheless, in its current version, JaceP2P is only fault tolerant in the computing phase. Integrating our algorithm into JaceP2P allows us to present a numerical parallel computing platform completely fault tolerant in both phases: the computing phase and the global convergence detection phase. The experiments we conducted on the Grid'5000 platform with more than 700 cores, show that the computing and convergence detection processes in JaceP2P are now adequate for volatile environments and for a very small acceptable cost.

The rest of this paper is organized as follows. In Sect. 2, we present some related works, and in particular we explain the DCD algorithm. We also describe the different schemes used to make a parallel platform fault tolerant. Section 3 describes the old version of JaceP2P with the centralized global convergence detection mechanism. In Sect. 4, we introduce our distributed and fault tolerant algorithm and we present its implementation into JaceP2P. The results of the experiments are exposed in Sect. 5. They show the low influence of these modifications on the platform's performances and reveal the benefits of this decentralization. Finally, we end this paper with a conclusion and some research perspectives.

Related work

Global convergence detection has been widely studied (for example, in [START_REF] Dijkstra | Derivation of a termination detection algorithm for distributed computation[END_REF][START_REF] Francez | Distributed termination[END_REF]), however, most of the proposed algorithms are centralized and inadequate for large scale computing over distributed architectures. In general, the objective of the global convergence mechanism is to find out as soon as possible if all the computing nodes have locally converged, while assuming that all nodes are regularly updated. In a synchronous model, this task can be easily accomplished because at each iteration all the nodes are synchronized and receive fresh data from their neighbors. On the other hand, in an asynchronous iteration model, this operation is not trivial.

The DCD algorithm

This decentralized algorithm detects the global convergence of parallel iterative asynchronous applications. It is articulated around two phases:

• Local convergence detection. When the residue, locally evaluated, is smaller than the precision requested by the user, we say that the subsystem computed on that node has converged. However, the residue does not always decrease uniformly. Sometimes it oscillates around the threshold which can lead to a false detection of the global convergence. To our knowledge, there is no method that guarantees the local convergence of a node. However, it is possible to reduce the occurrence of false local convergence detections by using the pseudo-period concept. A pseudoperiod can be defined as the smallest amount of time where a node receives at least one dependency from each of its neighbors. Therefore, a node converges locally if the residue stays under the threshold during one or many pseudo-periods. • Global convergence detection. This phase is based on the Leader election algorithm which chooses a node dynamically so that it can execute a specified task (here, it is the detection of the global convergence). This method consists in transforming the graph of computing nodes into a tree. When a node converges locally, it checks the number of neighbors that did not converge (noted as s):

-If s > 1, the node continues its iterative process.

-If s = 1, it means that the node is a leaf in the tree or that all its neighbors but one have converged (that implies that its neighbors subtrees have also converged). In this case, the node sends a convergence message to its neighbor that did not converge yet. When the neighbor receives this message, it decrements by one its number of neighbors that did not converge. -If s = 0, it means that all the neighbors of the node (and all their subtrees) have converged. Therefore, it is the last node to converge locally and it is elected as Leader in order to detect the global convergence of the system.

After electing a Leader, there are four steps to perform in order to be sure that all the nodes are still in a local convergence state:

1. The Leader broadcasts a verification message to all its neighbors which also transmit the message to all their neighbors. In this way, the verification message is propagated to all the computing nodes and signals the beginning of the verification phase. 2. After receiving the verification message, each node begins a new pseudo-period in which it waits for new data from all its neighbors and it computes an iteration using these new dependencies. If, between the sending of the convergence message to the end of the pseudo-period, the residue does not ever cross over the threshold, the response to the verification phase is positive, otherwise it is negative. This phase allows the system to verify that the state of the nodes has evolved. 3. If a node develops a negative response, it directly sends it to the neighbor that sent him the verification message which propagates it to the Leader. On the other hand, if the response is positive, the node has to wait until its neighbors send him positive responses. Then it propagates the response to the Leader.

4. If the Leader receives or computes a negative response, it directly broadcasts a negative verdict message to all its neighbors which also propagates the verdict to all the nodes. When a node receives a negative verdict, it starts the whole global convergence mechanism all over again. On the other hand, if the Leader develops and receives from all its neighbors positive responses, it broadcasts a positive verdict to all the computing nodes. This positive verdict means that the system has globally converged. When a node receives a positive verdict, it ends its iterative process.

Fault tolerance

Fault tolerance is not a new research area, however, the emergence of large scale computing architectures makes it a vital subject to be developed. In fact, when using these types of architectures (distributed clusters or Peer-to-Peer architectures), the probability that a node fails is very high. If a node dies, the user may have to restart the application which may need dozens of computing hours using hundreds of nodes. One of the main advantages of AIAC algorithms is their tolerance to the loss of dependencies messages. This property makes it easier for these algorithms to develop fault tolerance policies for the computing processes than for other types of iterative algorithms. In this context, it seems very interesting to make the global convergence detection process also fault tolerant. This will provide an iterative algorithmic model well suited to computing environments with mainly volatile nodes.

Many schemes have been used to develop platforms that tolerate crashes. Each one has its advantages and drawbacks and choosing the right method highly depends on the characteristics of the targeted computing environment. Here are some common concepts:

• Checkpointing [START_REF] Plank | Libckpt: transparent checkpointing under UNIX[END_REF]: This is the most common scheme. It is based on saving for each node the essential data required to continue an application after a crash. This procedure is executed on precise checkpoints which are usually specified by the application's programmer. Nowadays, a good fault tolerant platform automatically detects the checkpoints without any intervention from the user; we say it provides transparent fault tolerance services. Checkpointing can be divided into two groups:

1. Coordinated checkpointing [START_REF] Cao | On coordinated checkpointing in distributed systems[END_REF]: When using this method, at each checkpoint, all the computing nodes in the system must be synchronized with a barrier call and a snapshot of all the system is taken and saved. If the system is synchronous and a node crashes, all the other nodes are blocked. Once the replacement node retrieves the last backup, all the nodes must also rollback to the last checkpoint and continue their tasks from that checkpoint. The coordinated checkpointing method cannot be used in an asynchronous model because at each checkpoint, it synchronizes the whole system which eliminates all the advantages of the asynchronous model. 2. Uncoordinated checkpointing: Using this method, after a checkpoint, each node saves its data without any synchronization with the others. This allows the nodes to have backups at various times. This method suits the asynchronous models well.

Most of the fault tolerant platforms that use the checkpointing concept save their backups on a server. This requires having at least one reliable station. This centralization may cause a bottleneck. To reduce the effect of this problem, stations with high computing capacities and large bandwidth are used as backup servers. This method is not well adapted to distributed clusters and Peer-to-Peer environments where the nodes are volatile and have limited capacities. In such environments, the distributed redundant checkpointing method is used. In this method, the backups are saved on the computing nodes: each node saves its backups on its neighbors, so the number of backups per node is equal to the number of saving neighbors (which is specified by the user). This method does not make the platform fully tolerant to all types of crashes. Indeed, if a node and all its saving neighbors die at the same time, the backups for this node are lost and the platform cannot replace the dead node which will terminate the application. To make the platform more resistant to crashes, the user can increase the number of saving nodes, but, this could reduce the performances of the platform. Finally, the backups saved either on a server or on computing nodes can be stored in the memory or in files (XML files, binaries, . . . ) on hard disks. If the size of the backups is too big, it is preferable to store them in files to prevent overloading the memory. However, this method slows down the platform because it has to access the hard disk every time a node demands a backup from its neighbors. As an example, the Open MPI [START_REF] Hursey | The design and implementation of checkpoint/restart process fault tolerance for open MPI[END_REF] platform saves its backups in files on a local or distributed file system. • Process Replication [START_REF] Elnozahy | Replicated distributed process in Manetho[END_REF]: Using this scheme, each computing process is executed on many nodes. A master is chosen to represent each computing process and the others are considered as slaves. All the replications of a computing process execute the same process at the same time, but only the master communicates with the other masters. After each communication, it sends the information to its slaves which enables them to continue their tasks. If a master crashes, it is replaced by a slave. The dead node cannot be replaced by a new one because there is no backup to retrieve. So, after many crashes a computing process may be terminated if all its replications are dead. This method is not very efficient because if an application requires m machines, the platform has to reserve n * m machines where n is the number of replicas per computing process. This method has been implemented in the P2P-MPI platform [START_REF] Genaud | A Peer-to-Peer framework for robust execution of message passing parallel programs on grids[END_REF]. • Message logging [START_REF] Alvisi | Message logging: pessimistic, optimistic, and causal[END_REF]: using this scheme, each node saves the data that it receives on a log. When a node crashes, it is replaced by a new one which retrieves the log and executes from the beginning all the operations using the data saved in the log. This method saves the logs on a reliable server and requires to restart the computing process of the dead node from the beginning. If the application is large, the log may become full. For this reason, most of the time message logging is used in conjunction with the uncoordinated checkpointing method: Each node saves in a local log the messages it sent to its neighbors. When the log becomes too big, the node's data are saved on a reliable server. If a node crashes, the replacement node has to retrieve the backup from the backup server and the logs from its neighbors, then it executes all the operations that happened after the backup using the data stored on the logs. If a neighbor's log does not contain the required data, the neighbor must retrieve its backup and rollback [START_REF] Elnozahy | Manetho: transparent rollback-recovery with low overhead, limited rollback, and fast output commit[END_REF] to that state then it has to execute again all the operations executed after the backup. So, one crash may cause a domino effect and oblige many nodes to retrieve their old backups and rollback to that state. This method guarantees that the system will tolerate all the crashes that affect the computing nodes but it takes a lot of time to replace a dead node. This method is not adapted to volatile environments because it requires a reliable backup server which is impossible to have in such environments. This scheme has been implemented in the MPICH-V2 platform [START_REF] Bouteiller | MPICH-V2: a fault tolerant MPI for volatile nodes based on pessimistic sender based message logging[END_REF].

Since we are interested in executing asynchronous parallel iterative algorithms on Peer-to-Peer and distributed clusters architectures, we chose the uncoordinated checkpointing mechanism (on neighboring nodes) to make the decentralized convergence detection algorithm fault tolerant. It is the best method in these conditions. Nevertheless, it does not guarantee a 100% resistance but it drastically increases it: if an application uses 100 nodes with a 5% probability that a node crashes in a 100 seconds time interval, the probability that the system fails after 100 s is 5 * 100 = 500%. Now, if we applied the uncoordinated checkpointing on neighboring nodes and assumed that replacing a dead node takes 5 s and each node has five saving neighbors, the probability that the system fails after 100 s is (5 * 5/100) 6 = 0.000244% (equivalent to the probability that a node and its five saving neighbors die in the same five seconds time interval). So, the system resists to most of the crashes. In the rest of this article, we present a fault tolerant adaptation of the DCD algorithm and its implementation into the JaceP2P platform.

JaceP2P

The JaceP2P platform allows the development and the execution of AIAC algorithms that are implemented in Java (with/without dependencies between computing nodes) over volatile architectures. To tackle the specificities of AIAC algorithms, JaceP2P provides many essential mechanisms for these algorithms. In particular, it has an asynchronous communication mechanism and offers different threads for computing and communicating. The execution environment provided by JaceP2P allows computing nodes to connect and disconnect from the platform at any moment, in particular while executing a task. To resist to the volatility of the computing units, JaceP2P saves the state of each task at regular checkpoints.

Saving checkpoints

As mentioned before, the decentralized uncoordinated checkpointing is the best scheme for AIAC fault tolerant algorithms. It consists in saving automatically the state of a task at specific checkpoints on neighboring nodes. There are two drawbacks to this method. First of all, the process of checkpointing assigned to each node slows down the computing process because it runs in parallel with the computing process and consumes some of the node's computing power. Secondly, there is still a small risk that a backup could be lost if all backup neighbors die at the same time. It is possible to increase the robustness of the platform by increasing the number of backups for each node, but this will slow down the computing process even further. A compromise between security and performance must be found. In JaceP2P, the computing data are saved every n iterations with n defined by the user as a parameter. If a node dies after saving its state, the platform retrieves the last backup and reruns this task on a new node. The fundamental properties of AIAC algorithms make it possible for this node to continue the task from the last checkpoint while other computing units continue their tasks without having any concern for the dead node.

Architecture

There are three types of nodes in JaceP2P (see Fig. 3):

1. Super-node. All the super-nodes form a network that stores in a distributed manner (in registers) the identifiers (name and IP address) of each daemon connected to the platform. 2. Spawner. The main role of a spawner is to launch the application. It needs to know the URL address of the server that contains the application's .class files, the number of nodes needed to solve the problem and all the eventual parameters of the application. All this information must be provided by the user. Once it has this information, the spawner contacts a super-node to reserve the nodes needed for this application (see next paragraph), and stores their identifiers in its register. This register is broadcasted to all the reserved nodes, so they can communicate with each others. When the application starts, the spawner becomes responsible for detecting the death of any computing node. If a daemon dies, the spawner contacts once again the super-node and reserves a new daemon which has to replace the dead one. The spawner modifies its register by adding the identifier of the new daemon and deleting that of the dead one. The new register is broadcasted to all the computing nodes (so they can communicate with the newly added node).

The spawner also detects the global convergence of the parallel iterative application: when a daemon detects that its iterative task has converged, it notifies the spawner. If the spawner receives local convergence messages from all the daemons, it broadcasts a global convergence message to them. This message stops the computing processes and frees the daemons.

3. Daemons. The daemons execute the computing tasks forming the application. When a daemon is launched, it connects to a super-node to signal its availability. When it is reserved by a spawner, it retrieves from it a task and begins its execution. In order to be able to continue the task of a dead node, each daemon regularly saves its data on its neighbors. In practice, every n iterations each daemon sends its backup to a different neighbor according to the round-robin strategy. If a node is selected to replace a dead one, it contacts all its neighbors, retrieves from them the last backup and continues the computing task. When a daemon detects the local convergence of its iterative task, it signals it to the spawner. When a daemon receives a global convergence message, it begins computing the next step if there is one, otherwise it terminates its computing process and reconnects to a super-node.

A decentralized and fault tolerant mechanism for detecting the global convergence

The drawbacks of centralized global convergence detection mechanisms

In JaceP2P, the global convergence is centralized and is detected by the spawner. This centralization limits the scalability of the platform (the spawner may be overloaded when detecting the convergence of a problem solved by hundreds of nodes) and the global convergence currently implemented in JaceP2P does not necessarily reflect the state of all the nodes at the time of the detection. Indeed, in the AIAC algorithmic model, a node that executes many iterations without receiving new data from its neighbors, may locally converge (i.e., its residue is lower than the threshold) for a moment and signal it to the spawner, then diverge again. If the other daemons have also converged, the spawner will detect a false global convergence. Finally, in its current version, JaceP2P is vulnerable to crashes while detecting the global convergence.

Here are some examples:

• If we assume that a daemon receives a global convergence message from the spawner and crashes just before saving it on its neighbors. The daemon replacing the dead one will retrieve an old backup and will not know that the parallel application has converged. If the application is composed of many steps, the node will continue to compute the old step indefinitely while the other nodes compute the next step. In this case, the application is blocked. • If the spawner broadcasts a global convergence message to the daemons, and because of the network congestion, a node does not receive the message; this node will never know that the system has converged and that it must compute the next step. Thus, it will continue to compute the same step indefinitely and the application will be blocked forever.

To fix up these problems, we have implemented a fault tolerant version of the DCD algorithm. We used an hypercube topology in the implementation to reduce the broadcast time of the convergence messages. If n is the number of computing nodes, the maximum number of nodes that a message has to cross is equal to x with x equal to the smallest integer that verifies n < 2 x+1 . In our implementation of the DCD algorithm, each node discovers its convergence neighbors using Algorithm 1.

Algorithm 1 Discovering convergence neighbors for an hypercube tree

Critical procedure and backups

To be able to transform the DCD algorithm into a fault tolerant algorithm, we have defined a critical procedure as follows: "each sequence of instructions that affects two neighbors and blocks one of them if that node crashes before saving its state". Thus, in the DCD algorithm, if a node n1 locally converges and all its neighbors but one (n2) have also converged, n1 has to send a convergence message to n2. This message signals to n2 that n1 has converged. n2 decrements by one the number of its neighbors that did not converge yet and returns an acknowledge message to n1 which means that n2 has well received the convergence message. Once n1 receives the acknowledge message, it waits for the verification phase and stops sending convergence messages. Such a sequence of instructions is a critical procedure because it affects two neighbors (n1 and n2), modifies their states and if one of the two nodes dies during this procedure, the algorithm will be blocked indefinitely. In fact, if n2 dies or disconnects from the platform, the spawner detects that this node has not been sending any heartbeat message for a while, so it considers that it is dead. Then the spawner tries to replace this dead node: It contacts a super-node and requests an available node. The reserved node n3 replaces the dead one and to be able to continue the task, retrieves the last backup. If n2 died after returning the acknowledge message to n1, all the modifications, made after receiving the convergence message, are lost and n3 have retrieved an old backup with a false number of neighbors that have not converged yet. n3 will wait indefinitely for a convergence message from n1. To solve this problem, n2 must save its data before sending the acknowledge message to n1.

On the other hand, if n1 receives the acknowledge message and dies just before saving, the daemon replacing the dead node will retrieve an old backup. Thereafter, it detects again the local convergence of its task and sends a new convergence message to n2. The recipient notices that it has already received this message, so it ignores it but returns an acknowledge message to n1 so it can execute the next phase.

Identifying a critical procedure while detecting the global convergence is very difficult. The easiest approach is to save data after the execution of each instruction concerning the global convergence detection. This solution is not practical given its execution cost. Thus, it is necessary to determine the critical sequences of instructions and to minimize their numbers (see Sect. 4.4, for more information on the different critical procedures in the DCD algorithm). Now, two types of backups are implemented into JaceP2P:

• Saving computing data. It is executed every n iterations and once the node is sure of the local convergence of its subsystem after the verification phase. This backup contains the solution and dependencies vectors. Using this backup, a node can continue the task from that solution vector. • Saving convergence data. It is executed during a critical procedure. It contains all the values necessary to detect the global convergence and maintains the system's coherence.

Distinguishing between these two types reduces the bad influence of checkpointing on the computing process. Actually, the size of the convergence backup is very small compared to the size of the computing backup. So, saving the convergence data again and again will not have a great impact on the performance of the computing process.

Acknowledge messages

The critical procedure concept, described in the previous paragraph, shows the importance of the acknowledge messages. It allows the sender to be sure that the recipient has received the convergence message. The DCD algorithm is composed of many phases. If a crash prevents a daemon from receiving a message that informs it that it has to pass to the next phase, the daemon and its subtrees nodes will be stuck in this phase while the others pass onto the next one. Therefore, we have to make sure that all the notifications concerning the global convergence detection are well received by their recipients. To accomplish that, we propose two solutions. First, the recipient treats the message and saves the results of that treatment on its neighbors. Then it returns the acknowledge message to the sender. Second, the recipient just saves the information sent in the message and returns the acknowledge message to the sender then handles the saved information afterward. Each approach has its advantages and drawbacks:

• Using the first solution, while the recipient treats the convergence message and saves the results on its neighbors, the sender remains blocked. However, the recipient has to save just once after processing the message's data. • Using the second solution, the recipient saves the message's data and quickly returns the acknowledge message to the sender. However, after processing the message's data, it saves the results again on its neighbors.

To be able to choose between those two methods, we must evaluate the execution time cost for these two solutions. Unfortunately, it is very hard to achieve this test because it depends on many parameters like the network's latency, the processing speed of nodes and the number of components solved on each node. In practice, we can use the first method for the messages that do not require a lot of treatment (like local convergence messages) and the second method for the broadcasting messages because they could block the sender for a long time (while the message is propagated to all the subtrees nodes).

Overview of the fault tolerant algorithm

As described before, the decentralized convergence detection algorithm is decomposed into many phases, in this section, we will show how each phase has been modified in order to make the whole algorithm fault tolerant:

• Before local convergence: each node computes its iterative task and at the end of an iteration it asynchronously exchanges dependencies vectors with its neighbors then evaluates its residue in order to detect the local convergence. Throughout this phase, only data backups are saved on neighbors because in the asynchronous iteration model the loss of dependencies messages is tolerated and does not affect the convergence of the parallel iterative application (i.e., the solution of the studied problem).This phase corresponds to lines 5 to 9 in Algorithm 2. • After detecting the local convergence: this is the first critical procedure. When a node detects that its residue is smaller than the requested precision, it computes a pseudo-period. It waits to receive new dependencies vectors from all its neighbors, then it computes a new iteration using these new data. If the residue evaluated after this iteration is still smaller than the threshold, the node declares its local convergence. This phase corresponds to lines 12 to 16 in Algorithm 3. As mentioned before in the DCD algorithm, the locally converged node tests the number of its unconverged neighbors (nbNotConv): if nbNotConv = 1, it executes the sequence of instructions illustrated in Fig. 4 and from lines 26 to 33 in Algorithm 3.

If nbNotConv = 0 the node declares itself Leader and propagates the beginning of the verification phase on all its neighbors as in Fig. 5 and from lines 17 to 24 in Algorithm 3. If a node involved in this procedure fails, the node recovery mechanism described in Sect. 4.2 is applied. • Verification phase: when a node receives a verification message it executes the sequence of instructions illustrated in Fig. 6 and in the ReceiveVerification procedure in Algorithm 4 where it begins the verification phase. During the verification phase, each node waits for new dependencies tagged with the verification tag. Afterward, it computes a new iteration using those new data. If during all these operations the residue is still under the threshold, the node elaborates a positive response, otherwise it elaborates a negative response.

Algorithm 2 Main()

1: for each step do 2:
Initialize the global convergence variables Broadcast a verification message / Broadcast the verdict 3:

Wait for an acknowledge message from each neighbor NbNeighboursNotConv ← NbNeighboursNotConv -1

Save convergence data on its neighbors 4:

Return an acknowledgment message to the sender 5: end if procedure: ReceiveVerification() Save convergence data on its neighbors 7:

Return an acknowledge message to the sender Store the response in the response vector 3:

Save convergence data on its neighbors 4:

Return an acknowledge message to the sender Save convergence data on its neighbors 9:

Return an acknowledge message to the sender 10: end if

• Responding to the verification phase: if a node elaborates or receives a negative message, it directly sends it to the sender of the verification message as explained in Algorithm 3 from lines 72 to 79. On the other hand, if it receives from all its neighbors (except from the sender of the verification message) positive responses and elaborates a positive response, it positively responds to the sender of the verification message. This case is presented in Algorithm 3 from lines 81 to 87 and sending a response is illustrated in Fig. 4. If the response is positive, the sender saves its computing and convergence data on its neighbors. This phase is also considered as a critical procedure. As above, if a node involved in this procedure fails, the node recovery mechanism described in Sect. 4.2 is applied. • Declaring a verdict: when the Leader receives or elaborates a negative response, it broadcasts a negative verdict to all its neighbors as in Fig. 5 and in Algorithm 3

Fig. 4 The instructions executed when sending a convergence or a response message Fig. 5 The instructions executed when broadcasting a verification or a verdict message from lines 53 to 60. Then it restarts the convergence detection algorithm all over again. When the Leader elaborates and receives from all its neighbors positive responses, it sets its state to global convergence and broadcasts a positive verdict to all its neighbors, as illustrated in Fig. 5 and in Algorithm 3 from lines 61 to 68. This is also a critical procedure and the node recovery mechanism described in Sect. 4.2 may be applied, if a node involved in this procedure fails. • Receiving a verdict: when a node receives a verdict, it executes the sequence of instructions illustrated in Fig. 6 and in the ReceiveVerdict procedure in Algorithm 4.

If the verdict is negative, it reinitializes its convergence variables and restarts the convergence detection algorithm for the same step. On the other hand, if the verdict is positive, it sets its state to global convergence and begins computing the next step or terminates the application. Here is the fourth critical procedure. If a node involved in it fails, the node recovery mechanism described in Sect. 4.2 is applied.

For more details on the algorithm, the reader can refer to the algorithms at the end of this paper.

Mathematical description

In order to evaluate the robustness of our approach, we have implemented on JaceP2P a numerical iterative application that solves a three dimensional advection-diffusion equation. This equation represents mathematically the transport processes of pollutants, salinity, and so on, combined with their biochemical interactions. It follows an initial boundary value problem for a nonlinear system of PDEs, in which nonlinearity only comes from the biochemical interspecies reactions.

A system of 3D advection-diffusion-reaction equations has the following form:

∂c ∂t + A(c, a) = D(c, d) + R(c, t), (1)
where c denotes the vector of unknown species concentrations, of length m, and the two vectors,

A(c, a) = [J(c)] × a T , (2)
D(c, d) = [J(c)] × d × ∇ T , (3)
respectively, define the advection and diffusion processes (J(c) denotes the Jacobian of c with respect to (x, y, z)). The local fluid velocities u, v, and w of the field a = (u, v, w) and the diffusion coefficients matrix d are supposed to be known in advance.

A simulation of pollution evolution in shallow seas is obtained if a is provided by a hydro-dynamical model. The chemical species dynamic transport is defined by both advection and diffusion processes, whereas the term R includes interspecies chemical reactions and emissions or absorption from sources. Following a common approach, R(c, t) can be expressed using production and loss terms, denoted respectively by P and L: R(c, t) = P (c, t) -L(c, t).

(4)

Both terms can be further refined:

P (c, t) = P I (c, t) + P S (t), L(c, t) = (L I (c) + L S (t)) × c.
While the terms P and L indexed by I denote the contributions (emission and absorption) from chemical interspecies reaction, the contributions from sources are indexed by S.

As we consider a test problem in three spatial dimensions, reduced to two chemical species, the system described by (1) becomes a system of two PDEs:

⎛ ⎝ ∂c 1 ∂t ∂c 2 ∂t ⎞ ⎠ + ∇c 1 × a ∇c 2 × a = ∇ • ((∇c 1 ) × d) ∇ • ((∇c 2 ) × d) R 1 (c, t) R 2 (c, t) , (5)
where matrix d satisfies

d = ⎛ ⎜ ⎝ (x) 0 0 0 (y) 0 0 0 (z) ⎞ ⎟ ⎠ , (6)
and the vector function R(c, t) is defined according to:

• production terms P I and P S P I (c, t) = q 4 (t)c 2

q 1 c 1 c 3 , P S (t) = 2q 3 (t)c 3 0 , (7)
• loss terms L I and

L S L I (c) = 0 q 2 c 1 q 2 c 2 0 , L S (t) = q 1 c 3 0 0 q 4 (t) . (8)
Clearly, the coupling of the two PDEs is induced by the reaction term R.

As far as transport is concerned, in this work, vertical advection (dimension z) is neglected, and the velocity field vector a is supposed to have a constant value. Hence, a is given by

a = (u, v, w) = (-V , -V , 0), (9)
with V = 10 -3 . Regarding diffusion coefficients, horizontally they are positive constants, whereas the vertical ones vary. Thus, we have a matrix d of the form

d = ⎛ ⎜ ⎝ K h 0 0 0 K h 0 0 0 K v (z) ⎞ ⎟ ⎠ , ( 10
)
where K h = 4.0 × 10 -6 and K v (z) = 10 -8 × exp( z 5 ).

For the reaction term (apart from the two unknown concentrations of the contaminants, i.e., c 1 and c 2 ), the different quantities in ( 5)-( 8), are chosen as follows:

• c 3 = 3.7 × 10 16 , q 1 = 1.63 × 10 -16 and q 2 = 4.66 × 10 -16 , • q 3 (t), q 4 (t) are chosen according to (j = 3, 4),

q j (t) = exp[ -a j sin(ωt) ] if sin(ωt) > 0, q j (t) = 0 i fs i n (ωt) ≤ 0, (11)
using the following parameters: ω = π/43200, a 3 = 22.62 and a 4 = 7.601.

For more information concerning the mathematical description of the problem, the reader can refer, for example, to [23]. On the other hand, for more information concerning the implementation of the problem, we can cite [START_REF] Byrne | User documentation for PVODE an ODE solver for parallel computers[END_REF].

To solve this problem we used the Multisplitting-Newton method [START_REF] Bahi | Asynchronous multisplitting methods for nonlinear fixed point problems[END_REF]: the principle is to split the initial domain into several subdomains in order to assign one of them to each computing unit involved in the parallel computation. At each iteration, each subdomain is solved sequentially using the standard Newton method. All the nodes send their boundaries values asynchronously to their neighbors at the end of an iteration. Then the fault tolerant DCD algorithm is executed in order to detect if the parallel iterative application has globally converged.

Experiments' conditions and results

The primary tests were conducted on a small local cluster containing just 20 nodes in order to be sure of the stability of the platform. After this phase and in order to prove that our modifications increased the scalability of JaceP2P, we have run many experiments on the Grid'5000 platform [25] using hundreds of nodes. As shown in Fig. 7, Grid'5000 is a public national experimental Grid, geographically distributed on nine sites in France. It is composed of many heterogeneous clusters. The connection's bandwidth between two sites is 10 Gb/s and between two nodes on the same site is about 1 Gb/s. In our experiments, the node's crashes were produced by a shell script that kills n nodes every m seconds.

Experiment n • 1: one site

In this experiment, 252 biprocessor machines (so, in total about 500 processors) located on a single site (e.g., Orsay) were used to solve a system containing 405.224.000 components which simulates a 90 seconds time interval. The nodes used are equipped with 2 AMD Opteron 246 2.0 GHz processors or with 2 AMD Opteron 250 2.4 GHz processors. Initially, we considered evaluating the cost of our modifications on the performance of JaceP2P. However, we quickly noticed that it is almost impossible to accurately measure this cost on this platform because our experiments test an heterogeneous application (the computing load is different between two distinct computing nodes) using heterogeneous nodes and executing an asynchronous application. Furthermore, the platform uses a decentralized mechanism to detect the global convergence of the system and for two successive executions, JaceP2P may order the same node to solve two distinct parts of the system. These characteristics make it almost impossible to obtain similar execution times when running the same application twice with the same initial conditions. For this reason, we chose to test JaceP2P while changing the number of crashes during a simulation. In Table 1, we present the results of these experiments. We can notice that the platform has resisted to the huge amount of crashes. New daemons have replaced the dead nodes and thanks to the decentralized backups, they were able to continue their tasks from the last backup. JaceP2P seems to be more scalable than before (the old version of JaceP2P with a centralized global convergence detection was not able to use a huge number of nodes like in this experiment because the spawner was quickly overloaded with convergence messages) and the overprice due to crashes is quite acceptable.

Experiment n • 2: three distant sites

In order to simulate a Peer-to-Peer environment (huge number of nodes, heterogeneous and volatile nodes, heterogeneous networks . . . ), we have executed the same application over 3 distant clusters:

• Site 1 (Nancy), where each machine is equipped with 2 double cores 1.6 GHz Intel Xeon 5110. • Site 2 (Sophia), where each machine is equipped with 2 processors AMD Opteron 246 2.0 GHz. • Site 3 (Orsay), described in the first experiment.

Therefore, 252 stations, with more than 700 cores, are used to solve a system containing 405.224.000 components which simulates a 90 second time interval. Table 2 presents the results of the experiments.

We noticed that while computing on distant clusters, the platform suffers more from crashes than when computing on a single cluster. Indeed, the communications between nodes on distant clusters are slower than between nodes on a single cluster which implies that the nodes need more time to retrieve backups from their neighbors and to broadcast convergence messages. As the previous one, this experiment proves that our modifications to the JaceP2P platform have made it resistant to crashes without significantly reducing its performance.

Conclusions and perspectives

In this article, we have presented a decentralized global convergence detection algorithm that works in a volatile environment. To our knowledge, this algorithm is the first of its kind. It is an extension of the Bahi et al. algorithm based on uncoordinated checkpointing and acknowledge messages. This new version of the algorithm has been implemented into the JaceP2P platform. The experiments conducted on the Grid'5000 platform showed that JaceP2P is now more stable and more scalable than before and that the backups do not significantly reduce the performance of the platform. In this way, this algorithm can be used in a grid computing environment where crashes are very frequent.

Our current and future works concern the decentralization of the detection of dead nodes (now this task is assigned to the spawner) and the attribution of a specific register to each node based on its needs. At the moment, all the nodes have the same register containing all the nodes addresses. We would like each node to have a register that only contains the addresses of its computing, saving, and convergence detection neighbors. These modifications will increase the scalability of the platform and reduce the bottlenecks and network congestions. Finally, we would like to duplicate the spawner in order to make it fault tolerant.

Fig. 3

 3 Fig.[START_REF] Bertsekas | Parallel and distributed computation: numerical methods[END_REF] The JaceP2P's architecture with 4 daemons. D1 and D2 are connected to the spawner which stores their respective addresses A1 et A2

1 :

 1 if action = SendVerification / action = SendVerdict then {To broadcast the verification and verdict messages} 2:

4 : 4

 44 Algorithm Reception methods procedure: ReceiveConvergence() 1: if reloading = false and action=nothing then 2:

Fig. 6

 6 Fig.[START_REF] Savari | Finite termination of asynchronous iterative algorithms[END_REF] The instructions executed when receiving a verification or a verdict message

Fig. 7

 7 Fig.7The sites of Grid'5000 in France

Table 1

 1 Execution time with 3 random crashes every n seconds

	n	∞	90	60	50	25
	Execution time	522 s 873 s 1003 s 1135 s 1738 s
	Total number of crashes 0	30	51	68	205

Table 2

 2 Execution time while randomly killing a node every n seconds on each site

	n	∞	90	60	50
	Execution time	565	1438	2008	2050 s
	Total number of crashes	0	48	100	122

if received an acknowledge message from each neighbor then Wait for an acknowledge message from each neighbor


Jean-Claude Charr
email: jean-claude.charr@univ-fcomte.fr

Raphaël Couturier
email: raphael.couturier@univ-fcomte.fr

David Laiymani
email: david.laiymani@univ-fcomte.fr

JACEP2P-V2: a Fully Decentralized and Fault Tolerant Environment for Executing Parallel Iterative Asynchronous Applications on Volatile Distributed Architectures

Keywords: Decentralized global Convergence, Peer-to-Peer architectures, Distributed clusters, Parallel iterative asynchronous algorithms

published or not. The documents may come   L'archive ouverte pluridisciplinaire

Introduction

The simulation of natural and nuclear reactions (like climate change or nuclear fusion) requires solving very large and complex numerical problems and necessitates computing very large data in order to obtain precise and reliable results. These problems cannot be solved using a single computing unit because most of the time, the numerical problems are so huge that a single computing unit does not have enough memory nor computing power to store the application and to solve it. These problems could only be solved using simultaneously many computing resources. With the development of new reliable network equipments and the emergence of cheap and fast desktops, scientists are able nowadays to create distributed architectures using only these simple low cost devices. Most of the time, these distributed architectures tends to replace equally powerful but more expensive supercomputers. In numerical computing we can distinguish three kinds of distributed architectures:

1. Local clusters are composed of similar workstations connected via a local network with low latency and large bandwidth. 2. Distributed clusters are composed of many distant clusters with heterogeneous computing units that are connected via heterogeneous networks with high latency and large bandwidth. 3. Global computing architectures are mainly composed of public unused heterogeneous workstations connected to Internet. These architectures offer free and unlimited computing power but suffer from the volatility of the nodes and from the slowness of communications.

Most of the time, the local cluster architecture does not have enough computing power to solve very large numerical problems. Therefore, in this paper, we are only interested in distributed clusters and global computing architectures. Using one of these parallel architectures, developers have to parallelize the method that solves the numerical problem in order to execute a subsystem of the problem on each computing unit. However, to use these architectures, the developer needs to manage carefully the exchange of data between the different computing units, especially when using high latency networks with heterogeneous and volatile nodes.

There are two classes of methods to solve numerical problems:

-Direct methods give the exact solution of a numerical problem after executing a finite number of operations. However, they are not really suited to distributed clusters and global computing architectures because they require several synchronizations. -Iterative methods iterate many times the same block of instructions until obtaining a good approximation of the solution (e.g., Jacobi or Conjugate Gradient algorithms [START_REF] Saad | Iterative Methods for Sparse Linear Systems[END_REF]). An iterative method converges when the "residual vector" (there is many methods to evaluate the value of the residue, for example Residue = max i (|x k+1 i x k i |) where x k i denotes the value of the component i at iteration k) is inferior to the precision ( ) requested by the user. Iterative methods are well adapted for very large problems.

Since we would like to solve very large numerical problems, in the rest of this paper we only focus on iterative methods. Now, from a parallel point of view, there are two models of parallel iterative algorithms:

-The synchronous iteration model. Using this model, as shown in figure 1, after each iteration (represented by a filled rectangle in the figure), a node sends its dependencies to its neighbors and waits for the reception of all the dependency messages from all its neighbors. Then all the nodes must synchronize to test if the system has globally converged. This results in large periods of idle time (represented by white spaces between the rectangles). These synchronizations can drastically penalize the overall performances in the case of large scale heterogeneous platforms. Moreover, if a dependency message is lost, the receiver will wait forever for that message and the application will be blocked. In the same way, if a computing node is dead, all the rest of the nodes will be blocked until the dead node is replaced. In conclusion, this model is not well suited for large scale volatile computing environments.

-The asynchronous iteration model [START_REF] Bahi | Parallel Iterative Algorithms: from sequential to grid computing[END_REF]. Using this model, as shown in figure 2, after each iteration, a computing node sends its dependencies to its neighbors and begins the next iteration using the last received dependency data. The node does not have to wait for the reception of the dependencies messages from its neighbors, consequently, there is no idle time anymore (no white spaces between iterations). The sending and the receiving mechanisms are asynchronous and the computing nodes tolerate the loss of data messages. Even if a node dies, the rest of the nodes can continue the computation process using the last dependency message sent by the dead node.

In conclusion, the asynchronous iteration model is well adapted for volatile environments like peer-to-peer architectures or distributed clusters.

To tackle the specificities of the asynchronous iteration model on distributed clusters or global computing architectures, we have developed JACEP2P-V2, a new and improved version of JACEP2P [START_REF] Bahi | JACEP2P: an environment for asynchronous computations on peer-to-peer networks[END_REF]. JACEP2P-V2 is a fully decentralized and fault tolerant platform dedicated to designing and executing parallel iterative asynchronous algorithms on volatile architectures. The aim of this paper is to present the design and the features of this new platform. The rest of this paper is organized as follows: in the next section we present some existing platforms related to our work. These platforms are briefly described and the differences between our work and these platforms are emphasized. In the third section, we present JACEP2P's architecture and its limits. Then, we introduce JACEP2P-V2 and we describe in details its mechanisms and functionalities. In the fourth section, we present the experiments conducted on the Grid5000[4] testbed using JACEP2P-V2 to solve a numerical problem. Finally, we end this paper with a conclusion and some perspectives.

Related Work

Recently, many middlewares for distributed clusters and global computing platforms have been developed. However, most of them are not well adapted for large numerical computing. Here are some examples:

-Seti@home [5]: The amazing success that this platform has achieved, helped the creation of generalized environments like Xtrem Web [START_REF] Cappello | Computing on large-scale distributed systems: XtremWeb architecture, programming models, security, tests and convergence with grid[END_REF] and Boinc [7]. They are independent of the application and fault tolerant. The user creates a parallel application and executes it using the "workers". However, in these platforms, the clients cannot communicate with each others. So they cannot execute a parallel computing application with dependencies between nodes.

-JXTA [8]:

It is an open-source project, composed of a set of peer-to-peer protocols that allows any connected device (cell phone to PDA, PC to server) on the network to communicate and collaborate. However, JXTA is a low level platform and offers a lot of general functionalities that are not well adapted for executing complex computing applications. -ProActive [START_REF]ProActive[END_REF]: It is an Open Source Java library for parallel, distributed, and multi-threaded computing. Although this environment provides direct communications between nodes using the RMI technology, when two nodes communicate, they must be synchronized (even if the concept of future objects exists). Moreover ProActive uses a global checkpointing mechanism [START_REF] Cao | On coordinated checkpointing in distributed systems[END_REF] that requires synchronizing all the nodes in case of failures. In consequence, the asynchronous iteration model cannot simply be used on this platform. -JACE [START_REF] Bahi | Jace: a java environment for distributed asynchronous iterative computations[END_REF]: "Java Asynchronous Computation Environment" is a multithre-aded Java based library designed to build asynchronous iterative algorithms and execute them in a Grid environment. In JACE, two nodes exchange data (synchronously or asynchronously) using either Sockets, RMI or NIO (New Input/Output). However, this platform is not fault tolerant, so it cannot be used in large scale volatile environments.

JACEP2P-V2

JACEP2P

JACEP2P is a distributed platform implemented using the Java programming language and dedicated to developing and executing parallel iterative asynchronous applications. JACEP2P executes parallel iterative asynchronous applications with dependencies between computing nodes. On the other hand, JA-CEP2P is fault tolerant which allows it to execute parallel applications over volatile environments and even for stable environments like local clusters and grids, it offers a safer and crash free platform.

JACEP2P's architecture. Figure 3, presents the architecture of JACEP2P and the various components that form the platform:

-The first entity is the "super-node" (represented by a big circle in figure 3). Each super-node stores in its register the identifiers (IP address) of all the computing nodes that are connected to it and are not executing an application. The super-node regularly receives heartbeat messages (represented by doted lines in figure 3) from the computing nodes connected to it. If the super-node does not receive a heartbeat message from a computing node included in its register for a given period of time, it declares that this computing node is dead and deletes its identifier from the register. -The second entity is the "spawner" (represented by a square in figure 3).

When a user wants to execute a parallel application, he or she launches a spawner with the required parameters which contacts a super-node to reserve the required computing nodes. The super-node reserves the demanded daemons (see next paragraph) which are removed from the super-node's register and returns to the spawner a register containing the identifiers of the reserved computing nodes. When the spawner receives the register, it creates a task for each computing node and starts the execution of the tasks on the respective daemons. The spawner has to send its register to all the computing nodes in order for them to be able to communicate with each others. Moreover, the spawner is responsible for detecting the disconnection of a computing node. Indeed, when the computing nodes are reserved by the spawner, they start sending their heartbeat messages to the spawner. If the spawner detects that a computing node has not sent to it a heartbeat message for a while, it declares that this computing node is dead. Then, it fetches a new one from the super-node in order to replace the dead one. The spawner initializes the new daemon, which retrieves the last backup of the dead node and continues the computing task from that checkpoint. Finally, the spawner is also responsible for detecting the global convergence of the parallel iterative application. When a subsystem converges locally, the computing node executing it sends a convergence message to the spawner. If the spawner receives a convergence message from all the computing nodes, it declares that the parallel iterative application has globally converged. -The third entity is the "daemon" or the computing node (represented in figure 3 by a hashed small circle if it is free and by a white small circle if it is executing an application). Once launched, it connects to a super-node and waits for a task to execute. During the execution of the parallel application, the daemons can communicate with each others and they regularly save their state on their neighbors. At the end of a task, the daemons reconnect to the super-node.

To be able to execute asynchronous iterative applications, JACEP2P has an asynchronous messaging mechanism (for more details interested readers can refer to [START_REF] Bahi | JACEP2P: an environment for asynchronous computations on peer-to-peer networks[END_REF]) and to resist to daemons' failures, it implements a distributed backup mechanism called the uncoordinated distributed transparent checkpointing [START_REF] Plank | Libckpt: Transparent checkpointing under UNIX[END_REF]. This method allows daemons to save their data on neighboring daemons without any user intervention. The asynchronous nature of the application allows two daemons to execute two different iterations, thus each daemon saves its status without synchronizing with other daemons. This decentralized procedure allows the platform to be very scalable, with no weak points and does not require a secure and stable station for backups. Moreover, if a daemon dies, the other computing nodes continue their tasks and they are not affected by this failure.

JACEP2P's limitations. In [START_REF] Charr | A decentralized convergence detection algorithm for asynchronous iterative algorithms in volatile environments[END_REF], the experiments' results proved that the first version of JACEP2P performs very well and presents a relatively small overhead. Nevertheless, this version has some important limits:

-JACEP2P is not fully fault tolerant. Indeed, in this version, spawners' crashes are not tolerated. Moreover, while executing the global convergence process, the platform does not resist well to the disconnection of daemons. -JACEP2P has a centralized failure detection. The spawner receives heartbeat messages from all the daemons and detects if a daemon is dead. If the application is being executed by a large number of daemons, the spawner will be overloaded with heartbeat messages. This will delay the detection of a dead daemon and could even lead to a false crash detection. Moreover, if many daemons die successively and there is only one spawner to handle the dead daemons, then the spawner will take a lot of time to replace them. This may reduce the performance of the platform. -JACEP2P has a centralized global convergence detection mechanism which is not well adapted for executing asynchronous parallel iterative algorithms on volatile architectures. The daemons executing such applications do not receive dependencies messages from their neighbors at each iteration. This may lead to a false local convergence and thus result to false global convergence detection. Furthermore, the spawner could be overloaded by convergence messages, if many daemons converge locally at the same time. -JACEP2P has many centralized mechanisms like launching the application, detecting the global convergence and detecting the dead nodes. These centralizations limit the scalability of JACEP2P. -In JACEP2P, each daemon receives the whole register which contains the identifiers of all the daemons executing the application. If a daemon crashes and is replaced by a new one, the spawner has to notify the modifications to all the daemons in order to update their registers. This could overload the spawner and increase the congestion of messages in the network.

To remedy these problems we present JACEP2P-V2 a fully decentralized and fault tolerant platform. In the next subsection, we will describe in details the functionalities and characteristics of the new platform.

JACEP2P-V2's Architecture

Figure 4 shows the architecture of JACEP2P-V2 where we notice that there are two spawners handling the execution of a single application and each group of entities (spawners, daemons and super-nodes) forms a circular network. JACEP2P-V2 has similar entities as JACEP2P but with different functionalities:

-Super-nodes. They form a circular network now and store in an equally distributed manner the identifiers of all the computing nodes that are connected to the platform and that are not executing any application. Each super-node has a status table containing the number of connected computing nodes to each super-node and all the super-nodes share a "token" that is passed successively from a super-node to the next one. Once a super-node has the token, it computes the average number of computing nodes connected to a super-node (avg) using the status table. If avg is lower than the number of computing nodes connected to it, then it sends the identifiers of the extra computing nodes to the super-nodes that have the number of computing nodes connected to them less than avg. If the number of computing nodes connected to it has changed, it broadcasts the information to all the super-nodes in the platform. Finally, it passes the token to the next super node. This distribution reduces the overload of the super-nodes. -Spawners. When a user wants to execute a parallel application that requires N computing nodes, he or she launches a spawner. The spawner contacts a super-node to reserve the N computing nodes plus some extra nodes in order to transform them into spawners. When the spawner receives the register from the super-node, it transforms the extra daemons into spawners and stores the identifiers of the rest of the daemons in its own register. Once the extra nodes are transformed into spawners, they form a circular network and they receive the register containing the identifiers of the computing nodes. Then each spawner becomes responsible for a subgroup of computing nodes, starts the tasks on the computing nodes under its command and sends a specified register to them. So each computing node receives a specified register that only contains the identifiers of the daemons it interacts with and that depends on the application being executed. These specified registers reduce the number of messages sent by the spawners to update the register of the daemons after a daemon crashes because usually a small number of daemons is affected by this crash.

-Daemons. Once they begin executing an application they form a circular network which is only used in the failure detection mechanism. Each daemon can communicate directly with the daemons whose identifiers it has in its register.

JACEP2P-V2 Functionalities and Characteristics

After describing quickly the modifications made on the architecture of JACEP2P-V2, in this section we present in details the different new functionalities and characteristics implemented in JACEP2P-V2:

-Completely decentralized. JACEP2P-V2 is completely decentralized. In fact, all the tasks are divided between the entities of the same type. For example, a daemon can be connected to any super-node and the group of super nodes shares equally the control of the free daemons connected to the super-node network. The spawners are also decentralized: once a spawner is launched to execute an application, it quickly duplicates itself into several spawners (depending on the number of daemons required to execute the parallel application) by transforming some daemons into spawners. Each spawner becomes responsible for starting the application on a subgroup of daemons and handling the needs of that subgroup. For the computing nodes, each one executes a part of the application and the sum of their work gives the solution of the global problem. This distribution of tasks, allows JACEP2P-V2 to solve very large problems and thus to become very scalable with theoretically no limiting conditions. -Completely fault tolerant. We have implemented many mechanisms to make the three entities that form the core of JACEP2P-V2 fault tolerant. An important concept available for the three entities is the decentralized crash detection mechanism. It enables the neighbors of a node to detect if it is dead or alive. Each group of entities forms a circular network. This organization is needed to apply the decentralized crash detection mechanism we have implemented. Each entity has a "heartbeat thread" that signals regularly to the next node in the circular network that the sender is still alive and another thread, the "scan thread" , that tests at each iteration if the previous node in the circular network has recently sent a heartbeat message. If for a given period of time the node does not receive a heartbeat message from the previous one, the scan thread detects that the previous node is probably dead. Depending on the type of the dead node, the disconnection is handled.

In fact, each entity has a restoring mechanism which is also dependent on the saving mechanism used for each type of entity.

• For daemons we use the distributed backup mechanism described before and we have implemented two types of backup in JACEP2P-V2.

The first backup contains all the information concerning the state of a node (convergence data) and its computing process (solution vector). This backup is saved each N iterations (N given by the user and usually depends of the length of an iteration) on a different neighbor using the "round-robin" strategy. On the other hand, the second backup only contains the status data. This backup has a smaller size and it is saved when the status of a daemon has changed, especially when it concerns the global convergence detection mechanism. This backup is saved on all the backup neighbors simultaneously. Once a daemon detects that the previous daemon is dead, the daemon signals it to the spawner responsible for it. The spawner contacts a super-node and acquires a new daemon. The new daemon replaces the dead one and retrieves the last status backup and the last data backup. Once it has the backups, it continues the task from that last checkpoint. During all this operation, all the other daemons continue their tasks normally. • For spawners, we use the duplication mechanism. The spawner is duplicated into many spawners. All the spawners have all the information concerning all the daemons and each one manages only a subgroup of daemons. If a spawner dies, the next spawner detects it (using the same scheme described before). Then, that spawner contacts a super-node, gets a new daemon and transforms it into a spawner. Once it becomes a spawner, it receives the register containing the identifiers of all the daemons executing the application, it identifies its subgroup of daemons, it informs them that it is the new spawner and it is reintegrated into the circular spawner network. • For super-nodes, there is no saving mechanism, they do not contain very valuable information. When a dead super-node is detected by the next super-node, it is rejected from the circular super-node network. All the daemons that were connected to the dead super-node will reconnect onto another super-node.

-Multi-threaded. JACEP2P-V2 is multi-threaded. The computing process is never blocked by the exchange of data messages between daemons. Each functionality (communicating, detecting crashes, saving and computing) has its own thread. -Decentralized global convergence detection algorithm: JACEP2P-V2 has implemented this algorithm to detect efficiently the global convergence of the asynchronous iterative parallel algorithms executed on the platform. It consists of two phases: the detection phase and the verification phase. This algorithm is presented in details in these papers [START_REF] Bahi | Parallel Iterative Algorithms: from sequential to grid computing[END_REF][START_REF] Charr | A decentralized convergence detection algorithm for asynchronous iterative algorithms in volatile environments[END_REF].

-Acknowledge mechanism: JACEP2P-V2 tolerates the loss of data messages when it executes parallel asynchronous iterative algorithms. However, it has to ensure the right reception of the convergence messages by the receivers in order to ensure a coherent system. Therefore, we had to implement an acknowledge mechanism dedicated to the convergence messages. When a computing node sends a convergence message to another one, the receiver handles the received message (this usually changes the state of the receiver) then it saves its new state on its backup neighbors and next it returns an acknowledge message to the sender. Once the sender receives the acknowledge message, it knows that the receiver has received the convergence message and then it saves its state on its backup neighbors so it does not send the convergence message again. On the other hand, if the sender does not receive the acknowledge message, it knows that the message did not reach its destination and that it has to send the convergence message all over again.

Experiments

In order to evaluate the benefits of the improvements that have been implemented in JACEP2P-V2, we have conducted two sets of experiments on Grid'5000 French national grid. The same experiments were realized using JA-CEP2P and JACEP2P-V2 in order to compare both platforms. Both sets of experiments were realized on two different architectures. During these tests, both platforms had to execute a parallel iterative application that solves a three dimensional advection-diffusion equations system. This system represents mathematically the transport processes of pollutants, salinity, and so on, combined with their bio-chemical interactions.

Mathematical Description

A system of 3D advection-diffusion-reaction equations has the following form: and the diffusion coefficients matrix d are supposed to be known in advance. A simulation of pollution evolution in shallow seas is obtained if a is provided by a hydro-dynamical model. The chemical species dynamic transport is defined by both advection and diffusion processes, whereas the term R includes interspecies chemical reactions and emissions or absorption from sources. For more details, readers can refer to [START_REF] Bahi | Synchronous and asynchronous solution of a 3D transport model in a grid computing environment[END_REF].

∂c ∂t + A (c, a) = D (c, d) + R (c, t) ( 1

First Experiment: Local Cluster

In this experiment, we compare JACEP2P to JACEP2P-V2 while executing the same application on a single site. This application solves a system containing 405.224.000 components and that simulates a 90 seconds time interval. 252 biprocessors computing units, located in Orsay, were used to run this application. The computing nodes were equipped with 2 AMD Opteron 246 2.0GHz or 250 2.4GHz processors. To prove that the two platforms are fault tolerant, we used a shell script that randomly kills three computing nodes each n seconds. The results for this set of experiments are presented in table1. It shows the execution times taken by JACEP2P and JACEP2P-V2 to solve the problem with various frequencies of nodes crashes. It is obvious that JACEP2P-V2 outperforms JACEP2P in each category. We also notice that JACEP2P-V2 is less affected than JACEP2P by the disconnection of computing nodes. Indeed, when the computing nodes disconnect frequently, JACEP2P suffers a lot because of the centralized nature of some of its components. On the other hand, with the JACEP2P-V2's decentralized dead nodes detection, the dead nodes are detected faster by their neighbors and thus they are replaced quickly by new ones to continue their tasks. These mechanisms reduce the influence of the crashes on the performance of JACEP2P-V2 platform.

2nd Experiment: Distributed Clusters

In this second set of experiments, we aimed at simulating a global computing architecture which has the following characteristics: large number of heterogeneous computing units, high latency communications and volatile nodes. So, we used the same number of computing nodes but this time we have chosen them from three distant sites in order to have heterogeneous computing nodes. Moreover, the latency between two nodes from distinct sites is superior to the one between two nodes located on the same site, thus the latency of the communications is also heterogeneous. The computing nodes were selected from the following sites: Nancy where each station is equipped with 2 double cores 1.6 GHz Intel Xeon 5110, Sophia where each station is equipped with 2 processors AMD Opteron 246 2.0GHz and Orsay which is described in the first experiment. We executed the same application as in the first experiment using JACEP2P and JACEP2P-V2. We have also simulated the volatility of the computing nodes by using the same perturbator script. However in this experiment, the script killed one daemon on each site each n seconds. The results for this set of experiments are presented in As in the previous experiment, JACEP2P-V2 outperforms JACEP2P, in particular when the environment is highly volatile. Moreover, the crashes overhead is totally acceptable in JACEP2P-V2. These experiments prove that the modifications implemented in JACEP2P improve its performance on volatile architectures that suffer from high latency between computing nodes.

Conclusion and Perspectives

In this paper we have presented the new version of JACEP2P, called JACEP2P-V2. This parallel platform is dedicated for designing and executing parallel asynchronous iterative applications in volatile environments. This new version is fully fault tolerant which makes it able to resist the failure of any node in the platform, especially the ones executing an application. We have also implemented a decentralized mechanism for detecting dead nodes in order to replace them. We also conducted two sets of experiments using two different architectures. In all these tests JACEP2P-V2 outperformed JACEP2P.

In the near future, we want to test JACEP2P-V2 using more computing units. We also would like to test it on a real global computing architecture, using unused public computing units connected via Internet. Finally, we want to implement many types of iterative asynchronous applications on JACEP2P-V2 to show the benefits of this platform and its general utility.

Fig. 1 . 1 Processor 2 TimeFig. 2 .

 1122 Fig. 1. Two processors using the synchronous iteration model

Fig. 3 .

 3 Fig. 3. JACEP2P's architecture and the different components

Fig. 4 .

 4 Fig. 4. JACEP2P-V2's architecture and different components

  ) where c denotes the vector of unknown species concentrations, of length m, and the two vectors A(c, a) = [J(c)] * a T and D(c, d) = [J(c)] * d * ∇ T respectively define the advection and diffusion processes (J (c) denotes the Jacobian of c with respect to (x, y, z)). The local fluid velocities u, v and w of the field a = (u, v, w)

Table 1 .

 1 Execution time with 3 random crashes every n seconds

	n ∞ 90	60	30
	Execution time for JACEP2P 522s 873s 1003s 1611s
	Total number of crashes for JACEP2P	0 30	51 159
	Execution time for JACEP2P-V2 495s 565s 595s 744s
	Total number of crashes for JACEP2P-V2	0 18	28	68

Table 2 .

 2 Execution time with one crash every n seconds at each site

	n ∞	90	60	50
	Execution time for JACEP2P 565s 1438s 2008s 2050s
	Total number of crashes JACEP2P	0	48 100 122
	Execution time for JACEP2P-V2 581s 624s 632s 663s
	Total number of crashes JACEP2P-V2	0	19	30	38

table 2 .

 2