Guillaume Aupy

Anne Benoit

Henri Casanova

Yves Robert

Checkpointing Strategies for Scheduling Computational Workflows

Keywords: checkpointing, scheduling, workflow, fault-tolerance, reliability

We study the scheduling of computational workflows on compute resources that experience exponentially distributed failures. When a failure occurs, rollback and recovery is used to resume the execution from the last checkpointed state. The scheduling problem is to minimize the expected execution time by deciding in which order to execute the tasks in the workflow and deciding for each task whether to checkpoint it or not after it completes. We give a polynomialtime optimal algorithm for fork DAGs (Directed Acyclic Graphs) and show that the problem is NP-complete with join DAGs. We also investigate the complexity of the simple case in which no task is checkpointed. Our main result is a polynomial-time algorithm to compute the expected execution time of a workflow, with a given task execution order and specified to-be-checkpointed tasks. Using this algorithm as a basis, we propose several heuristics for solving the scheduling problem. We evaluate these heuristics for representative workflow configurations.

Introduction

Resilience has become a key concern when computing at large scales because enrolling more processors in an application execution leads to more frequent application failures [START_REF] Dongarra | The International Exascale Software Project[END_REF]. (In this work we use the term "processor" in a broad sense to mean a processing elements of the platform on which one can run a portion of a parallel application, e.g., a multi-socket multi-core blade server.) Making each individual processor reliable, for instance via redundant hardware components, is costly. Since costs are highly constrained when designing a parallel platform, one must typically use commercialof-the-shelf (COTS) processors, the reliability of which is driven by the market. Consequently, each processor has a Mean Time Between Failure (MTBF), say µ, that varies from a few years to a century. When enrolling p processors to execute a tightly-coupled parallel application, a failure on any of the processors will cause an application failure. The overall MTBF of this set of processors is µ/p, which can be low (a few hours or less) when p is large. As a result, no matter how reliable the individual processors, there is a value of p above which failures become common rather than exceptional events.

The above considerations have prompted decades of research in the area of fault-tolerant computing. The most well-known approach is checkpoint-rollback-recovery, by which application state is saved to persistent storage at different points, e.g., periodically throughout execution [START_REF] Young | A first order approximation to the optimum checkpoint interval[END_REF][START_REF] Daly | A higher order estimate of the optimum checkpoint interval for restart dumps[END_REF]. When a failure occurs, the application execution can be resumed from the most recently saved such state, or checkpoint. A well-studied question is that of the optimal checkpointing strategy [START_REF] Young | A first order approximation to the optimum checkpoint interval[END_REF][START_REF] Daly | A higher order estimate of the optimum checkpoint interval for restart dumps[END_REF][START_REF] Gelenbe | Performance of rollback recovery systems under intermittent failures[END_REF]. Too infrequent checkpoints lead to wasteful re-computation when a failure occurs, but too frequent checkpoints lead to overhead during failure-free periods of the application execution. Checkpointing can happen in a coordinated or uncoordinated manner, and the advantages and drawbacks of both approaches are well-documented [START_REF] Bouteiller | Coordinated checkpoint versus message log for fault tolerant MPI[END_REF]. Checkpointing can be agnostic to the application, in which case full address space images are saved as checkpoints [START_REF] Chandy | Distributed snapshots : Determining global states of distributed systems[END_REF][START_REF] Elnozahy | A survey of rollback-recovery protocols in message-passing systems[END_REF]. Alternately, checkpointing can be application-aware so that only the application data truly needed to resume execution is saved. This latter approach is more efficient because less data needs to be saved, but requires modifying the implementation of the application [START_REF] Agarwal | Adaptive incremental checkpointing for massively parallel systems[END_REF].

In this work, we study the execution of workflow applications on platforms subject to processor failures. An application is structured as a Directed Acyclic Graph (DAG) in which each vertex represents a tightly-coupled parallel task and each edge represents a data dependency between tasks. This general model is relevant for many scientific workflows [START_REF] Bharathi | Characterization of scientific workflows[END_REF]. The difficulty of scheduling DAGs of parallel tasks, or applications with "mixed parallelism," without considering processor failures, has long been recognized [START_REF] Chakrabarti | Modeling the benefits of mixed data and task parallelism[END_REF]. It comes from the need to not only decide on a traversal of the task graph, as in classical scheduling problems, but also to decide how many processors should be assigned to each task. In addition, complex data redistributions must take place so that output data from one task can serve as input data to another task when the two tasks do not use the same number of processors. It is not clear how to model redistribution costs in practice and thus how to make judicious scheduling and processor allocation decisions [START_REF] Dutot | Scheduling on large scale distributed platforms: from models to implementations[END_REF][START_REF] Suter | Scheduling delta-critical tasks in mixed-parallel applications on a national grid[END_REF]. Because we consider processor failures, which makes the scheduling problem even more difficult, in this work we opt for a simplified scenario in which each task uses all the available processors. In other words, the workflow DAG is linearized and the tasks execute in sequence, using the whole fraction of the platform that is dedicated to the application. This scenario is representative of a large class of compute-intensive scientific applications whose workflow is partitioned into (typically large) tightly-coupled parallel computational kernels. Each parallel task is executed across all available processors, and produces output data that is kept in memory while executing its immediate successors in the DAG. Executing each task on all processors makes it possible to avoid complex data redistributions among tasks that use different numbers of processors [START_REF] Suter | Scheduling delta-critical tasks in mixed-parallel applications on a national grid[END_REF]. While it would be possible to used checkpoint-roll-back recovery within each task, it would require either saving large checkpoints (application-agnostic) or modifying the implementation of the task (application-aware). Given that both approaches have drawbacks, we assume non-modified, and thus non-fault-tolerant, implementations for the tasks. Fault-tolerance must then be achieved by checkpointing the output data generated by each task once it completes. If there is a failure during a task execution, one must recover from the most recently saved checkpoints on all paths from the failed task upward to an entry task of the DAG, re-execute non-checkpointed predecessors of the task if necessary, and then re-execute the task itself. This is repeated until the task is successfully executed and its output possibly checkpointed.

We study the following problem. We are given a DAG of tasks and for each task we know how long it takes to compute its output, how long it takes to checkpoint its output, and how long it takes to recover its checkpointed output. We are given a platform with a known failure rate on which we want to execute the application. In which order should the tasks be executed and which tasks should be checkpointed? We call an answer to this twofold question a schedule. The objective is to find a schedule that minimizes expected application execution time, or expected makespan. We call this problem DAG-ChkptSched.

To the best of our knowledge, DAG-ChkptSched has only been answered for the very specific case in which the DAG is a linear chain [START_REF] Toueg | On the optimum checkpoint selection problem[END_REF]. For general DAGs, the problem is more difficult. In fact, even computing the expected makespan of a given schedule is difficult. This is surprising, because the ordering of the tasks is given by the schedule as well as the location of all checkpoints. But when computing the expected execution time of a task, one has to account for the state of all its predecessors, which depends upon when the last failure has occurred. In this context, we make the following contributions:

• We show that although DAG-ChkptSched can be solved in polynomial time for fork DAGs1 , its associated decision problem is NP-complete for join DAGs2 . This result shows the intrinsic complexity of DAG-ChkptSched, but is largely expected as both the linearization of the DAG and the location of the checkpoints must be determined.

• We study the simple instance of the problem where no task is checkpointed. We show that optimal schedules for tree-shaped DAGs3 are depth-first schedules, but the complexity for general DAGs remains open.

• We provide a polynomial-time algorithm for computing the expected makespan of a schedule. This algorithm gives a fundamental basis for designing and comparing scheduling heuristics for arbitrary DAGs.

• We propose a set of heuristics for solving DAG-ChkptSched for general DAGs and evaluate these heuristics quantitatively. To the best of our knowledge, these heuristics are unique in the literature since previous work lacked an algorithm to estimate the makespan of a schedule (except when the DAG is a linear chain [START_REF] Toueg | On the optimum checkpoint selection problem[END_REF]).

The rest of this paper is organized as follows. Section 2 provides an overview of related work. Section 3 is devoted to formally defining the problem and all model parameters. Section 4 discusses the complexity of several instances of DAG-ChkptSched: fork DAGs (polynomial), join DAGs (NP-hard) and no-checkpoint (open, but polynomial for tree-shaped DAGs). Section 5 provides our key result that DAG-ChkptSched for general DAGs belongs in NP: we give a polynomialtime algorithm to compute the expected makespan of a given schedule. Section 6 presents a set of heuristics for solving the problem with general DAGs. These heuristics are evaluated experimentally in Section 7. Finally, Section 8 summarizes our main findings and discusses directions for future work.

Related work

Resilience to faults is one of the major issues for current and upcoming large-scale parallel platforms. The most common fault-tolerance technique used in high performance computing is checkpointrollback-recovery [START_REF] Chandy | Distributed snapshots : Determining global states of distributed systems[END_REF][START_REF] Elnozahy | A survey of rollback-recovery protocols in message-passing systems[END_REF][START_REF] Bouteiller | Coordinated checkpoint versus message log for fault tolerant MPI[END_REF][START_REF] Young | A first order approximation to the optimum checkpoint interval[END_REF]. A large body of work has studied periodic coordinated checkpointing for a single divisible application. Given the simplicity of the divisible model, a wide range of results are available including first order formulas for the checkpointing period that minimizes execution time [START_REF] Young | A first order approximation to the optimum checkpoint interval[END_REF][START_REF] Daly | A higher order estimate of the optimum checkpoint interval for restart dumps[END_REF] or more accurate formulas for Weibull failure distributions [START_REF] Bouguerra | A flexible checkpoint/restart model in distributed systems[END_REF][START_REF] Ling | A variational calculus approach to optimal checkpoint placement[END_REF][START_REF] Ozaki | Distribution-free checkpoint placement algorithms based on min-max principle[END_REF]. The optimal checkpointing period is known only for exponential failure distributions [START_REF] Bougeret | Checkpointing strategies for parallel jobs[END_REF]. Dynamic programming heuristics for arbitrary distributions have been proposed [START_REF] Toueg | On the optimum checkpoint selection problem[END_REF][START_REF] Bougeret | Checkpointing strategies for parallel jobs[END_REF]. Gelenbe and Derochette [START_REF] Gelenbe | Performance of rollback recovery systems under intermittent failures[END_REF] give a first-order approximation of the optimal period to minimize average response time. They compare it to the period obtained by Young [START_REF] Young | A first order approximation to the optimum checkpoint interval[END_REF] in a model where they do not consider one single long application and a fully-loaded system, but instead multiple small independent applications that arrive in the system following a Poisson process. Finally, Gelenbe and Hernández [START_REF] Gelenbe | Optimum checkpoints with age dependent failures[END_REF] compute the optimal checkpointing period that minimizes computational waste in the case of age-dependent failures: they assume that the failure rate follows a Weibull distribution and that each checkpoint is a renewal point.

Few authors have studied the resilience problem with workflows when checkpointing can only take place at the end of each task. Bouguerra et al. [START_REF] Bouguerra | Complexity Analysis of Checkpoint Scheduling with Variable Costs[END_REF] have studied a restricted version of DAG-ChkptSched when the workflow is a linear chain (with a single processor). They propose a greedy heuristic to minimize the total execution time in case of arbitrary failures. As already mentioned, Toueg and Babaoglu [START_REF] Toueg | On the optimum checkpoint selection problem[END_REF] have computed the optimal execution time for a linear chain of tasks using a dynamic programming algorithm to decide which tasks to checkpoint.

Our work is not restricted to linear chains and, as seen in upcoming sections, removing this restriction makes the problem fundamentally more difficult. In fact, even when a schedule is given (hence both a linearization of the DAG and a list of tasks to checkpoint), it is difficult to determine which tasks to re-execute and from which tasks to recover after one or more failures have occurred during application execution.

Framework

We consider a (subset of a) parallel platform with p processors, where each processor is a processing element that is subject to its own individual failures. When a failure occurs a processor experiences a downtime before it can be used again. In a production system, this downtime corresponds to replacing the processor by a logical spare. Like most works in the literature, we simply assume that a downtime lasts D seconds, where D is a constant. We assume that failures are i.i.d. (independent and identically distributed) across the processors and that the failure inter-arrival time at each processor is exponentially distributed with Mean Time Between Failures (MTBF) µ proc = 1/λ proc .

On the set of processors we want to execute a task-parallel application that is structured as a DAG G = (V, E), where V is a set of vertices and E a set of edges. Each vertex is a tightly coupled data-parallel task that is executed on all p available processors. Consequently, in all that follows, we can view the set of processors as a single macro-processor that experiences exponentially distributed failures with parameter λ = pλ proc , i.e., with MTBF µ = µ proc /p. Each edge corresponds to a data dependencies between two tasks. Since no two tasks run simultaneously, the sequence of executed tasks corresponds to one of the (many) linearizations of the DAG, i.e., task sequences that respect data dependencies. The DAG has n vertices, and the task corresponding to the i-th vertex is denoted by T i . A failure-free execution of task T i on the p processors takes w i seconds (the task's computational weight). This execution produces an output that can be checkpointed in c i seconds, and can be recovered from a checkpoint in r i seconds. If task T i executes successfully, then its successor tasks in the DAG can begin execution immediately since T i 's output data is available in memory (distributed over the p processors). If the output of a task is saved to a checkpoint, we say that the task is checkpointed. Informally, when a failure strikes during the execution of a task, all data that is stored in the processor memories is lost. Thus the output data of all tasks that have not been checkpointed are lost. Of course to recompute the failed task, we only need to recompute those of its ancestors that have not been checkpointed. But we will also need to recompute all non-checkpointed tasks so as to proceed with the execution of their successors. It is thus necessary to remember when the last failure occured, explaining why the algorithm in Section 5, which computes the expected makespan, is so intricate.

More precisely, if a failure strikes during the execution of T i , then T i must be re-executed. This re-execution requires that the input data to T i be available in memory. For each reverse path in the DAG from T i back to an entry task, one must find the most recently executed checkpointed task. One must then recover from that checkpoint, and re-execute all the tasks that were executed after that checkpointed task, i.e., all tasks whose output was lost and that are ancestors of T i along the reverse path. It may be that on such a path from T i to an entry task no checkpointed task is found, in which case one must begin by re-executing the entry task. An example DAG is shown in Figure 1, for which tasks whose output is checkpointed are shadowed (T 3 and T 4 ). Consider the following linearization of the DAG:

T 0 T 3 T 1 T 2 T 4 T 5 T 6 T 7 .
Let us assume that the first and only failure occurs during the execution of T 5 . To re-execute T 5 , one needs to recover the checkpointed output of T 3 . To execute T 6 , one then needs to recover the checkpointed output of T 4 and use the output of T 5 that is now available in memory. This sequence of recoveries and re-executions must be re-attempted until T 6 executes successfully. Finally, the output of T 2 was lost due to the failure, and no task is checkpointed on the reverse path from T 7 to T 1 . One must therefore re-execute T 1 , T 2 , and then finally T 7 . This example is for a single failure occurrence and yet is not straightforward, hinting at the complexity of the problem in the general case. As seen in the example, the DAG can have multiple entry tasks. The entry tasks (sometimes also referred to as sources), when restarted, do not have to recover any output from predecessors. In practice, each entry task would read the application's input data from disk, the overhead of which is included in the task's weight. The DAG can also have multiple exit tasks (sometimes also referred to as sinks). As soon as an exit task completes, it is removed from the DAG as well as any of its ancestors that have no remaining exit tasks as descendants. In practice, each exit task would write the application's output data to disk, and here again this overhead is included in the exit task's weight.

Executing the DAG in a fault-tolerant manner boils down to re-executing all the work that has been lost due to a failure, restarting from the most recent checkpoints if found and re-executing entry tasks otherwise. We enforce that the most recent checkpoint be used when recovering from a failure. It would be conceivable to ignore the checkpoints and, for instance, always re-execute the path completely from each entry task. This is only useful when the w i values are small and the r i values are large. Such situations are of dubious practical interest. It makes little sense to checkpoint a task if the time to recover the checkpoint is known to be longer than the time to re-execute that task. If this were the case, then the task could be fused with some of its predecessors for instance. So, in this work, when recovering from a failure, we enforce the use of the most recent checkpoints whenever possible.

Formally, let E[t(w; c; r)] denote the expected time to execute a task that would take w seconds in a fault-free execution and c seconds to checkpoint the output of this computation, with a recovery time of r seconds if a failure occurs during computation or checkpointing. If failures are exponentially distributed with mean 1/λ, and the processor downtime is D, it is shown in [START_REF] Bougeret | Checkpointing strategies for parallel jobs[END_REF][START_REF] Robert | On the complexity of scheduling checkpoints for computational workflows[END_REF] that:

E[t(w; c; r)] = e λr 1 λ + D e λ(w+c) -1 . (1)
We make extensive use of this notation and this result in this work. It is crucial to note that the above formula is valid even if failures occur during checkpointing or recovery. Many works in the literature assume that checkpointing and recovery are failure-free, an assumption that is not realistic for large numbers of processors. We define a schedule as a linearization of the DAG in which, for each task, it is specified whether the task's output should be checkpointed. The objective is to find the schedule that has the minimum expected makespan. Note that if λ = 0, i.e., if there are no failures, then one should do no checkpointing and all the linearizations of the DAG are equivalent. However, in the presence of failures, there is the usual trade-off between spending too much time checkpointing or spending too much time recovering and re-executing.

Complexity

In this section, we present several theoretical results to assess the complexity of DAG-ChkptSched. First, in Section 4.1, we introduce a polynomial-time algorithm for fork DAGs. Then we establish the NP-completeness of the decision problem associated to DAG-ChkptSched for join DAGs in Section 4.2 and we exhibit particular cases that can be solved in polynomial time. We conclude this section by discussing the problem instance where no task is checkpointed. While the complexity of this (apparently simple) instance remains open for general DAGs, we show that for tree-shaped DAGs any depth-first schedule is optimal (Section 4.3).

Fork DAGs

Theorem 1. DAG-ChkptSched can be solved in linear time for a fork DAG.

Proof. We consider a fork DAG with an entry task T entry and n exit tasks T 1 , . . . , T n . If T entry is checkpointed, then when T i fails we recover the checkpoint and try again. If T entry is not checkpointed, then we re-execute T entry but without re-executing the T i tasks that have already completed. The question is to decide whether or not T entry should be checkpointed, and to decide for the ordering of the n exit tasks.

We renumber the tasks so that task T i is the i th task executed in the linearization. Let X i be the random variable that corresponds to the execution time between the end of the first successful execution of task T i-1 and the end of the first successful execution of task T i . Let X 0 be the random variable that corresponds to the execution time of T entry followed by a checkpoint. Note that the case where T entry is not checkpointed is equivalent to considering c entry = 0, r entry = w entry . The expected execution time of the DAG is E[

n i=0 X i ].
By definition, E[X 0 ] = E[t (w entry ; c entry ; 0)]. Furthermore, it is straightforward to see that at the beginning of interval X i , the output of T entry is still available in memory (meaning that there has been no fault between the end of the last recovery -or execution-of T entry and the beginning of

T i ). As a result E[X i ] = E[t (w i ; 0; r entry )].
The execution time of the schedule does not depend on the linearization of the tasks. This is because, with the assumption that failures are exponentially distributed, the set of tasks that follow the checkpoint can be executed in any order.

In

Join DAGs

Consider a join DAG with a single exit task T exit and n entry tasks T 1 , . . . , T n . We denote by I Ckpt , resp. I NCkpt , the subset of {T 1 , . . . , T n } composed of the tasks that are checkpointed, resp. not checkpointed.

We first introduce Lemmas 1 and 2 which help characterize the structure of the optimal solution:

Lemma 1. In an optimal schedule, the tasks in I Ckpt are executed before the tasks in I NCkpt . When a failure occurs, the recoveries from the previously executed tasks in I Ckpt are executed after the last task from I Ckpt .

Lemma 2. Given the two sets I Ckpt and I NCkpt , we can compute the optimal expected makespan, which is achieved by scheduling the tasks in I Ckpt in non-increasing values of g(i), where g(i) = e -λ(wi+ci+ri) + e -λri -e -λ(wi+ci) .

Proof. The order in which the tasks from I NCkpt and recoveries are executed does not matter. This is because all these must be executed consecutively without failures, followed by T exit . The probability of a correct execution simply depends on the sum of the corresponding w i and c i , not their order.

Let us now consider the expected execution time for a given schedule order σ of the tasks from I Ckpt (meaning that in the schedule, T σ(1) is scheduled before T σ(2) , . . . , scheduled before

T σ(|I Ckpt |) ).
The time to execute a task T i ∈ I Ckpt and its checkpoint is independent of the rest of the computation and is equal to

E[t(w i , 0; c i )] = 1 λ + D e λ(wi+ci) -1 .
The expected time to execute the NCkpt tasks, the recoveries, and T exit depends on when the last failure occurred. This is because the number of recoveries to perform will differ depending on when that failure occurred. Let us call W NCkpt = i∈I NCkpt w i + w exit ; this is a constant amount of work that needs to be done regardless of when the last fault occurred:

• If the last fault occurred during the computation of the NCkpt tasks, recoveries or exit task, then all recoveries should be done and the expected time to execute the NCkpt tasks, the recoveries, and T exit is:

t 0 = 1 λ + D e λ W NCkpt + i∈I Ckpt ri -1 .
• If the last fault occurred during the computation of the k th checkpointed task (event E k ), then we first execute only the k -1 first recoveries. With probability p

(σ) k = e -λ(W NCkpt + k-1 i=1 r σ(i))
there is no subsequent failure, otherwise there is a failure and all recoveries should be reexecuted. The expected time is thus:

t (σ) k = p (σ) k W NCkpt + k-1 i=1 r σ(i) + 1 -p (σ) k × E[t lost (W NCkpt + k-1 i=1 r σ(i) )] + D + t 0 = 1 -p (σ) k 1 λ + D + t 0 ,
because E[t lost (w)] = 1/λ -w/(e λw -1). Finally, we have seen in Lemma 1 that a schedule proceeds in two distinct phases: first we execute the tasks in I Ckpt (with known execution time), then we execute the tasks in I NCkpt , necessary recoveries and exit task. At the end of the first phase depending on when the last fault occurred, we are in either one of the events E 1 , . . . , E |I Ckpt | . Precisely, with probability q (σ) i we are in the event E i , where

q (σ) 1 = e -λ |I Ckpt | j=2 (w σ(j) +c σ(j) ) , q (σ) i =1 = 1 -e -λ(w σ(i) +c σ(i) ) e -λ |I Ckpt | j=i+1 (w σ(j) +c σ(j) ) .
Finally, the expected execution of the second phase is

|I Ckpt | i=1 q (σ) i t (σ)
i , and the total expected execution time is:

t σ = i∈I Ckpt 1 λ + D e λ(wi+ci) + |I Ckpt | i=1 q (σ) i 1 -p (σ) i 1 λ + D + t 0 t σ = 1 λ + D   i∈I Ckpt e λ(wi+ci) + e λ W NCkpt + i∈I Ckpt ri |I Ckpt | i=1 q (σ) i 1 -p (σ) i   . (2)
Now that we have computed the expected execution time for a given order, let us focus on finding the optimal order: for a given schedule σ of the tasks in I Ckpt , let us compare its execution time to the same schedule where T σ(i) and T σ(i+1) are permuted (φ such that φ(i) = σ(i+1), φ(i + 1) = σ(i) and φ(j) = σ(j) for all other j). One can notice that for j = i, i + 1, then q

(σ) j = q (φ) j and t (σ) j = t (φ) j . Therefore: t σ -t φ 1 λ + D + t 0 = q (σ) i 1 -p (σ) i -q (φ) i 1 -p (φ) i + q (σ) i+1 1 -p (σ) i+1 -q (φ) i+1 1 -p (φ) i+1
.

Let us first consider the case where i = 1. For convenience we define

R = i-1 j=1 r σ(j) and W = |I Ckpt | j=i+2 (w σ(j) + c σ(j) ) (W = 0 when i + 1 = |I Ckpt |). Then: t σ -t φ 1 λ + D + t 0 = 1 -e -λ(w σ(i) +c σ(i) ) e -λ(W +w σ(i+1) +c σ(i+1) ) 1 -e -λ(W NCkpt +R) -1 -e -λ(w σ(i+1) +c σ(i+1) ) e -λ(W +w σ(i) +c σ(i)) 1 -e -λ(W NCkpt +R) + 1 -e -λ(w σ(i+1) +c σ(i+1) ) e -λW 1 -e -λ(W NCkpt +R+r σ(i)) -1 -e -λ(w σ(i) +c σ(i) ) e -λW 1 -e -λ(W NCkpt +R+r σ(i+1)) , e λW (t σ -t φ ) 1 λ + D + t 0 = e -λ(w σ(i+1) +c σ(i+1) ) -e -λ(w σ(i) +c σ(i) ) 1 -e -λ(W NCkpt +R) + e -λr σ(i+1) -e -λr σ(i) e -λ(W NCkpt +R) -e -λ(w σ(i+1) +c σ(i+1) ) -e -λ(w σ(i) +c σ(i) )
+ e -λ(r σ(i+1) +w σ(i+1) +c σ(i+1)) -e -λ(r σ(i) +w σ(i) +c σ(i) ) e -λ(W NCkpt +R)

= e -λ(W NCkpt +R) (g(σ(i + 1)) -g(σ(i))) ,

where g : i → e -λ(wi+ci+ri) + e -λri -e -λ(wi+ci) . In the case when i = 1, similarly we obtain:

t σ -t φ 1 λ + D + t 0 = e -λ(W +w σ(i+1) +c σ(i+1) ) 1 -e -λ(W NCkpt +R) -e -λ(W +w σ(i) +c σ(i) ) 1 -e -λ(W NCkpt +R) + 1 -e -λ(w σ(i+1) +c σ(i+1) ) e -λW 1 -e -λ(W NCkpt +R+r σ(i)) -1 -e -λ(w σ(i) +c σ(i) ) e -λW 1 -e -λ(W NCkpt +R+r σ(i+1) ) , e λW (t σ -t φ ) 1 λ + D + t 0 = e -λ(W NCkpt +R) (g(σ(i + 1)) -g(σ(i))) .
We conclude that in the optimal schedule, the set of tasks in I Ckpt should be sorted by non-increasing g values.

The first consequence of Lemmas 1 and 2 is that given the two sets I Ckpt and I NCkpt , we can construct the optimal solution in polynomial time. Furthermore, from Lemma 2 and Equation (2) we have: Corollary 1. When r i = 0 for all i, task ordering does not matter. The optimal expected execution time is then:

1 λ +D   i∈I Ckpt e λ(w i +c i ) -1 + e λ( i∈I NCkpt w i +w exit ) -1   . (3)
Theorem 2. DAG-ChkptSched for join DAGs is NP-complete.

Proof. Consider the associated decision problem: given a join DAG, λ, and a bound on the expected execution time, can we find the sets I Ckpt and I NCkpt , and the order in which the tasks are executed such that the bound on the expected execution time is respected? The problem is clearly in NP: we have shown in Lemma 2 that given the sets I Ckpt and I NCkpt we can compute the expected execution time in polynomial time (via an analytical formula).

To establish the NP-completeness, we use a reduction from SUBSET-SUM [START_REF] Garey | Computers and Intractability, a Guide to the Theory of NP-Completeness[END_REF]. Let I 1 be an instance of SUBSET-SUM: given n strictly positive integers w 1 , . . . , w n , and a positive integer X, does there exist a subset I of {1, . . . , n} such that i∈I w i = X? Let S = n i=1 w i . We build the following instance I 2 of our problem. We have a join DAG with n entry tasks T 1 , . . . , T n and an exit task T exit with w exit = 0 and, for all tasks

T i , i = 1..n,      w i = w i , c i = (X -w i ) + 1 λ log λw i + e -λX , r i = 0 .
We assume that λ ≥ 1 mini wi , so that for all i, c i > 0. Finally, we assume that the bound on the expected execution time is:

t min = λe λX (S -X) + e λX -1.
Let us show that I 2 has a solution (I Ckpt , I NCkpt ) if and only if we can find a set of tasks I NCkpt such that i∈I NCkpt w i = X. We will thus have shown that I 2 has a solution if and only if I 1 has one, the set I NCkpt from I 2 being the set I from I 1 if such a set exists.

Let us call W = i∈I NCkpt w i . We have seen in Equation (3) that the expected execution time is:

E[T ] = i∈I Ckpt e λ(wi+ci) -1 + e λ( i∈I NCkpt wi+wexit) -1 = i∈I Ckpt λe λX w i + e λW -1 = λe λX (S -W ) + e λW -1 .
We can differentiate E[T ] with respect to W : E[t(W )] = -λe λX +λe λW . This function is increasing, and equal to 0 when W = X. Therefore E[T ] is minimum when W = X, and its value is exactly t min . We conclude that (I Ckpt , I NCkpt ) is a solution to I 2 if and only if i∈I NCkpt w i = X, which concludes the proof.

Without checkpoints

In this section, we investigate the simple instance of the problem where no task is checkpointed. Given a DAG, the problem then reduces to finding the optimal linearization. We start with a few examples to show that this problem is not as simple as it appears at first sight. Consider the trivial DAG in Figure 2. There are three tasks T 1 , T 2 , T 3 and a unique dependency edge from T 1 to T 3 . Once T 3 has been successfully completed, we can remove tasks T 1 and T 3 from the DAG, hence the linearization (T 1 , T 3 , T 2 ) is optimal while (T 1 , T 2 , T 3 ) is not. Of course the linearization (T 2 , T 1 , T 3 ) is optimal too. This small example gives us some intuition to the design of efficient schedules:

1. Choose an output task T (a task without any successor in the DAG), execute all predecessors of T in sequence, and then T itself. 2. Repeat the above step until T is successfully completed. 3. Prune the DAG by removing T , and possibly other tasks that have become output tasks. 4. Repeat. Consider now the bipartite DAG in Figure 3. There are 3 input tasks, and 3 output tasks. It is not clear which output task to select and execute first, because each output task has a predecessor set (of size two) which is intersecting that of another output task. The problem of finding the best linearization seems of a combinatorial nature. We have not been able to determine its complexity but we conjecture that this problem is NP-complete, even for bipartite DAGs.

Although the complexity of the problem remains open, we expect depth-first linearizations to be more efficient than other traversals such as breadth-first ones. To further support this intuitive claim, we show that for tree-shaped DAGs4 any depth-first traversal of the DAG leads to an optimal linearization. Note that we assess the efficiency of depth-first traversals for general DAGs empirically in Section 7.

Theorem 3. When no task is checkpointed, a linearization for DAG-ChkptSched on a tree-shaped DAG is optimal if and only if it is a depth-first traversal.

Algorithm 1 Depth-First Traversal on a tree 1: procedure Pre-order(T )

2:

Execute the root element (or current element)

3:

for x child of T do 4:

Let Tree(x) be the subtree of T rooted in x 5:

Traverse Tree(x) by recursively calling Pre-order.

6:

end for 7: end procedure Proof. We begin by giving a characterization of a depth-first traversal (DFT) on a tree. The linearization of a tree by a schedule σ is a DFT if any subtree that started being executed is finished before the algorithm starts the execution of a node not contained in this subtree. Obviously, because of data dependencies, any DFT is a pre-order traversal.

Let G = (V, E) be a tree of n tasks, rooted in T 1 , and let σ be a traversal of G (without any checkpoint). Let E σ be the expected execution time of the traversal σ. If T i is executed in the j = σ -1 (i) th position, then we define X i as the random variable that corresponds to the execution time between the end of the first successful execution of task T σ(j-1) (the predecessor of T i in the schedule σ) and the end of the first successful execution of T i . Intuitively, X i is the time necessary to execute T i in the schedule.

We have

E σ = E[ n i=1 X σ(i) ]
. Indeed, the time elapsed between the beginning of the execution and the end of the first execution of the first task is X σ(1) , then the time elapsed between the end of the first execution of the (i -1) th task and the time between the end of the first execution of the i th task is X σ(i) . Overall the X σ(i) s do not overlap and cover the whole execution. Hence we have:

E σ = E[ n i=1 X i ] = n i=1 E[X i ]
by linearity of the expectation. For any task T i , we give a lower bound to E[X i ]: it is minimized when all necessary data is stored in memory, and when the amount of work to execute is only w i if there are no failure, and j∈ancestors(i) w j if a failure occurred. Hence,

E[X i ] ≥ E[t   w i ; 0; j∈ancestors(i) w j   ] .
Finally, because we are considering an execution without any checkpoint, then at the end of a successful execution of a given task T i , necessarily all its ancestors are alive (with output stored in memory) and do not have to be recomputed were we execute one of their successors. Let us now prove the result.

Assume the linearization strategy σ is a DFT. In this case, we show that for all i

E[X i ] = E[t   w i ; 0; j∈ancestors(i) w j   ].
Note first that in any linearization strategy, σ(1) = 1: the root of the tree is executed first because of precedence constraints, and E[X 1 ] = E[t (w 1 ; 0; 0)]. Consider T i0 a task with i 0 = 1. Let j 0 be its rank in the execution: σ(j 0 ) = i 0 . Then consider i 1 , the direct predecessor of i 0 in the tree DAG, and let j 1 be its rank in the execution: σ(j 1 ) = i 1 . By definition of the precedence graph, necessarily, j 1 < j 0 : the first execution of T i1 is necessarily anterior to the first execution of T i0 .

By definition of a DFT, any task executed between j 1 and j 0 belongs to the subtree rooted in T i1 , Tree(T i1 ). Then in particular, T σ(i0-1) is a successor of T i1 , and at the end of its execution, all the tasks on the path from T 1 to T i1 are alive (with output stored in memory). Then, E[X i0 ] = E[t w i0 ; 0; j∈ancestors(i0) w j ].

As a result, the execution time of the schedule is:

E σ = n i=1 E[t   w i ; 0; j∈ancestors(i) w j   ] ,
which matches the lower bound for each task, hence showing the optimality of the schedule.

Assume the linearization strategy is not a DFT. In this case, according to the characterization of a DFT, there exists a subtree Tree(T i0 ), not reduced to a single node T i0 , such that:

• the root of the subtree was executed at position j 0 = σ -1 (i 0 ),

• there exist j out < j in such that T σ(j out ) / ∈ Tree(T i0 ) and T σ(j in ) ∈ Tree(T i0 ). Intuitively, the task T σ(j out ) is executed in the middle of the tree rooted in T i0 . Let T i1 the first successor of T i0 executed in the schedule after T σ(j out ) , and let j 1 = σ -1 (i 1 ). Clearly, j out < j 1 ≤ j in and T σ(j1-1) / ∈ Tree(i 0 ). Hence with probability at least p 1 = 1-e -λw σ(j 1 -1) , there is a failure during X σ(j1-1) (the amount of work to do during X σ(j1-1) is at least w σ(j1-1) ), in which case at the beginning of X i1 , there is at least w i0 + w i1 units of work to do (T i0 would not have been recomputed during X σ(j1-1) ). Note that this is independent of anything that happened in the past and only depends on the schedule, potentially p 1 is greater than this value, and the amount of work that needs to be recomputed is also greater than this value. This shows that

E[X i1 ] ≥ (1 -p 1 )E[t   w i1 ; 0; j∈ancestors(i1) w j   ] + p 1 E[t   w i0 + w i1 ; 0; j∈ancestors(i1) w j   ] > E[t   w i1 ; 0; j∈ancestors(i1) w j   ] .
Finally, we obtain that:

E σ > n i=1 E[t   w i ; 0; j∈ancestors(i) w j   ] ,
and the schedule is not optimal (any DFT schedule has a lower expected execution time).

Evaluating a schedule for a general DAG

In this section, we consider a general DAG and a given schedule that specifies both a linearization of the DAG and which tasks are checkpointed. Our main contribution is to provide a polynomial-time algorithm to compute the expected execution time of the DAG with this schedule. For simplicity, we renumber the tasks so that task T i is the i th task executed in the linearization of the DAG.

Theorem 4. Given a DAG, and a schedule for this DAG, it is possible to compute the expected execution time in polynomial time.

Proof. Let X i be the random variable that corresponds to the execution time between the end of the first successful execution of task T i-1 and the end of the first successful execution of task T i . The expected execution time of the DAG is E[ n i=1 X i ]. Let F (X i ) be the event "There was a fault during X i ." Let Z i k be the event "There was a fault during X k and no fault during X k+1 to X i-1 , given that T i-1 was successfully executed." We have:

Z i k = i-1 j=k+1 F (X j ) F (X k ) (4)
(for the limit cases,

Z i i-1 = F (X i-1 ) and Z i 0 = i-1 j=1 F (X j ))
. The set of events Z i k for 0 ≤ k ≤ i -1 partitions the set of possibilities for X i . Hence we can write

E[X i ] = i-1 k=0 P(Z i k )E[X i |Z i k ]. (5)
We now need to show how to compute the

P(Z i k ) and E[X i |Z i k ]. Definition 1 (T ↓k i ).
Given a schedule, let j < k ≤ i, then we say that T j ∈ T ↓k i , if for all k ≤ l < i, T j / ∈ T ↓k l , and (i) either T j is a direct predecessor of T i , (ii) or there exists T l ∈ T ↓k i such that T l not checkpointed and T j is a direct predecessor of T l .

Less formally, the set T ↓k i corresponds to all the predecessors of T i (in the DAG), whose output is lost if the event Z i k occurs and needed for the computation of T i . For instance, it is not lost if it has been recomputed for another task executed after the last fault (that occurred during the computation of T k ) but still before T i . Furthermore, it is not needed if for all paths between T j Assume that a single fault occurs, and that it occurs during the execution of T 5 . Then when the execution of T 6 begins, the output of T 1 and T 5 are already in memory, due to the re-execution of T 5 , and do not need to be recomputed. However the output of T 4 needs to be recovered. Hence T 4 ∈ T ↓5 6 . However, because it was checkpointed, we do not need to recover T 2 and T 3 . Then, when computing T 7 , the output of T 3 is needed, hence T 3 ∈ T ↓5 7 . To compute T 3 , we further need T 2 . Hence T 2 ∈ T ↓5 7 .

and T i , there is a task whose output is not lost. If T j ∈ T ↓k i was not checkpointed, then we need to execute its work w j again, otherwise we need to execute the recovery r j . We give an illustrative example in Figure 4 Computing all sets T ↓k i is the key to evaluating the schedule makespan. Let W i k be the sum of the w j such that (i) T j is a non-checkpointed task and (ii) T j ∈ T ↓k i . Similarly, let R i k be the sum of the r j such that (i) T j is a checkpointed task and (ii) T j ∈ T ↓k i . We now show the following three properties:

A. ∀k, 0 ≤ k < i -1, P(Z i k ) = e -λ i-1 j=k+1 (W j k +R j k +wj +δj cj ) • P(Z k+1 k ),
where δ j is 0 if T j is not checkpointed, 1 otherwise.

B. ∀i ≥ 1, P(Z

i i-1 ) = 1 - i-2 k=0 P(Z i k ). C. ∀k, 0 ≤ k < i, E[X i |Z i k ] = E[t W i k +R i k +w i ; δ i c i ; W i i +R i i -W i k +R i k ]
, where δ i is 0 if T i is not checkpointed, 1 otherwise.

Computing P(Z i k ) for 0 ≤ k < i -1.
Let Y i k be the event "There is no fault during X k+1 to X i-1 given that there was a fault during X k ." We have:

Y i k = { i-1 j=k+1 F (X j )|F (X k )}.
Then by definition, P(Z

i k ) = P(Y i k )•P(F (X k )|T i-1 is successfully executed). Then we derive P(Y i k ) = e -λ i-1 j=k+1 (W j k +R j k +wj +δj cj )
. This is because we need to execute i-1 j=k+1 W j k + R j k + w j + δ j c j consecutive units of work without fault by definition of the W i k and R i k . Also, P(F (X k )|T i-1 is successfully executed) = P(F (X k )), as the probability of a fault during X k is independent of the execution of T i-1 since i -1 > k. Finally, one can see that P(F (X k )) = P(Z k+1 k ) by definition of Z k+1 k . Computing P(Z i i-1 ) for i ≥ 1. We have seen that the Z i k for 0 ≤ k ≤ i -1 partition the set of possibilities. Hence, by definition, i-1 k=0 P(Z i k ) = 1. We derive the value of P(Z i i-1 ) from the i -2 other values.

Computing E[X i |Z i k ] for 0 ≤ k < i. To compute E[X i |Z i k ]
, it suffices to see that we need to execute a work of W i k + R i k + w i with a checkpoint δ i c i . Then, if there is a fault, the recovery cost is W i i + R i i for a work of w i , which is identical to having a recovery cost of

W i i + R i i -W i k + R i k for a work of W i k + R i k + w i .
Hence, using the notation of Equation ( 1), we obtain that:

E[X i |Z i k ] = E[t W i k + R i k + w i ; δ i c i ; W i i + R i i -W i k + R i k ]
To conclude the proof, we need to show that we can compute the W i k and R i k values. Lemma 3. FindWikRik (Algorithm 2) computes W i k and R i k in polynomial time for all i ≥ k.

Proof. We consider the following invariant H i k for FindWikRik: (H i k ): At the end of the iteration i of the "for" loop (line 4), for all (j, i ) such that j < k

≤ i < i + 1, then • if T j ∈ T ↓k i , then -tab k .(i ).(j) ∈ {1, 2} (1 if T j is not checkpointed, 2 otherwise), -for i > i , tab k .(i ).(j) = 0 (0 means T j / ∈ T ↓k i because T j ∈ T ↓k i ), • else,
if there exists l < i , and

T j ∈ T ↓k l , then tab k .(i ).(j) = 0, -else tab k .(i ).(j) = -1.
For all (j, l) such that l > i > j, and T j ∈ T ↓k l , then tab k .(l).(j) = -1. To establish the invariant, we first introduce the following definition:

Definition 2 (path of T j in T ↓k i ). Let T j ∈ T ↓k i , then T j = T p0 , T p1 , . . . , T p l = T i is a path of T j in T ↓k i of length l, if (i) l = 1, or (ii) T p1 ∈ T ↓k i , T p1 is not checkpointed and T p1 , . . . , T p l = T i is a path of T p1 in T ↓k i of length l -1.
We define the distance l

(i,k) j of T j in T ↓k i
as the minimal length of a path of T j in T ↓k i . Here are some preliminary remarks before starting the proof:

• Once a value of tab k is set, it is never modified by Traverse (the switch on line 19).

• If tab k .(i ).(j) ∈ {1, 2}, then for all i > i , tab k .(i ).(j) = 0. Indeed, tab k .(i ).(j) is only set to 1 or 2 in the switch line 19, and when it is the first step of this switch (line 25) is to set tab k .(i ).(j) to 0 for all i > i .

• The only calls Traverse (j, i, k, tab k ) are for j = i or T j ∈ T ↓k i and T j not checkpointed. Hence for T j ∈Pred(T j ), either T j ∈ T ↓k i or ∃l < i, T j ∈ T ↓k l . This shows that for all (j, l) such that l > i > j, and T j ∈ T ↓k l , then tab k .(l).(j) = -1 since we will never visit such a node during iteration i of the "for" loop.

We are now ready to prove the invariant by induction. Let us show that H i k holds for i ≥ k. Let us show H k k . At the beginning of the "for" iteration (line 4), for i = k, tab k .(k).(j) = -1. We show that H k k holds for all tasks in T ↓k k (the case for tasks not in T ↓k k is trivial), and do this by induction on their distance (as defined in Definition 2) in T ↓k k . First, we verify that for all predecessors T j of T k whose distance is 1 in T ↓k k , the call Traverse (k, k, k, tab k ) checks whether T j ∈ T ↓k k (answer, yes) and has not been studied (the switch on line 19). If it is the case, then it assigns 1 or 2 to tab k .(k).(j), and then calls Traverse (j, k, k, tab k ) if and only if T j is not checkpointed. Then there is a call Traverse (j, k, k, tab k ) for all not-checkpointed elements of T ↓k k whose distance is 1 in T ↓k k . Let us now assume H k k holds for all T j ∈ T ↓k k such that l (i,k) j = l. Let us show the result for all

T j ∈ T ↓k k such that l (i,k) j = l + 1. Let T j , T p1 , . . . , T p l = T k path of T j in T ↓k k of length l + 1.
Then when T p1 was studied, by hypothesis because it is not checkpointed, there was a call Traverse (p 1 , k, k, tab k ). Because T j is a direct predecessor of T p1 , then either its value in tab k was already set to 1 or 2 through another path or it was set to -1 and this call has set it up to 1 or 2. By induction we obtain

H k k . Assuming ∀k ≤ i < i, H i k , let us show H i k . First note that H i-1 k
gives us (i) if there exists l < i , and T j ∈ T ↓k l , then tab k .(i ).(j) = 0, and (ii) ∀j, T j ∈ T ↓k i , then at the beginning of iteration i, tab k .(i).(j) = -1. Furthermore, with the preliminary remark, to show H i k , we simply need to show that for all j < k,

• if T j ∈ T ↓k i , then tab k .(i).(j) ∈ {1, 2} (1 if T j is not checkpointed, 2 otherwise),
• else, if for all l < i, T j / ∈ T ↓k l , then tab k .(i).(j) = -1.

The proof can be done by induction and is similar to H k k . The first call Traverse (i, i, k, tab k ) makes sure that this is true for all predecessors T j of T i whose distance is 1 in T ↓k i (the only reason why a predecessor T j of T i would not be in

T ↓k i is if ∃l < i, T j ∈ T ↓k
l , and in that case by induction hypothesis, tab k .(i).(j) = 0). Then there is a call to Traverse only for the predecessor tasks T j ∈ T ↓k i that are not checkpointed. Finally, H n k gives the correctness of Algorithm 2, whose complexity is O(n 3 ).

Altogether, Algorithm 2 is invoked for each task, and the complexity of the whole evaluation method is O(n 4 ).

Because we can compute the expected makespan of a schedule, a schedule of a DAG is a sufficient certificate to verify whether the expected makespan is below a certain threshold. Hence we have derived the following result: Corollary 2. The decision problem associated to DAG-ChkptSched is in NP for general DAGs (and is NP-complete by Theorem 2).

Heuristics for general DAGs

In this section, we develop polynomial-time heuristics in the case of general DAGs. A heuristic that computes a schedule for a given instance of DAG-ChkptSched must answer two questions: (i) how should the DAG be linearized? and (ii) which tasks should be checkpointed? To answer the first question, we consider three possible linearization strategies: Depth First (DF), Breadth First (BF), and Random First (RF). For DF and BF, we prioritize the tasks by decreasing outweight (i.e., the sum of the weights of the task's successors). The rationale is that tasks that have "heavy" subtrees should be executed first.

To answer the second question, we propose four checkpointing strategies. The first and second strategies are baseline comparators, and correspond to either never checkpointing (CkptNvr) or always checkpointing (CkptAlws). For both these strategies, we only consider the DF linearization. A DF linearization makes sense when no checkpoints are taken because one should make progress toward exit tasks aggressively rather than pursuing multiple exit tasks simultaneously (which is risky in the presence of failures). The choice of the DAG linearization is inconsequential when all tasks are checkpointed.

The third and fourth strategies fix the total number of checkpoints taken throughout application execution, say N , and checkpoint N tasks based on some criteria. Then they do an exhaustive search for the N value, N = 1, . . . , n -1 (recall that n is the number of tasks), that achieves the lowest expected makespan, which is computed in polynomial time as explained in Section 5.

In the third strategy, tasks are sorted by decreasing w i (checkpoint first the tasks whose computations are the longest), by increasing c i (checkpoint first the tasks whose checkpointing overheads are the shortest), or by decreasing d i , the sum of the weights of the successors (checkpoint first the tasks whose successors are more likely to fail). The top N tasks taken in these orders are checkpointed. We name the three versions of this strategy CkptW, CkptC, CkptD. end for 39:

tab k =Traverse (i, i, k, tab k ) 6: for j = 1 . . . k -1 do 7: switch tab k .(i).(j) do 8: case 1 9: W k .(i) ← W k .(i) + wj
Return tab k 40: end procedure

The fourth strategy, CkptPer, relies on the idea of periodic checkpointing [START_REF] Young | A first order approximation to the optimum checkpoint interval[END_REF][START_REF] Daly | A higher order estimate of the optimum checkpoint interval for restart dumps[END_REF]. Given a linearization of the DAG, consider a failure-free execution. If W is the sum of the w i values over all tasks, CkptPer checkpoints the task that completes the earliest after time x × W/N for x = 1, . . . , N -1. While periodic checkpointing is a typical approach for data-parallel computation, it does not account for the structure of the DAG.

Heuristic names are concatenations of the name of the linearization strategy and of the checkpointing strategy (e.g., RF-CkptC). Combining the three linearization strategies (DF, BF, RF) and the checkpointing strategies, we have a total of 14 heuristics.

Unfortunately, there are no heuristics in the literature to which we can compare the above heuristics. This is because no method to evaluate the expected makespan of a schedule was available before this work, thus precluding the design (and the straightforward evaluation) of reasonable heuristics.

Experimental evaluation

In this section, we present experimental results that quantify the performance of the heuristics in Section 6. The source-code (implemented in OCaml) and all input and output data are publicly available at [START_REF] Aupy | Source code and data[END_REF].

Experimental methodology

To evaluate the heuristics with representative DAGs, we use the Pegasus Workflow Generator (PWG) [START_REF] Bharathi | Characterization of scientific workflows[END_REF][START_REF] Pegasus | Pegasus workflow generator[END_REF]. PWG uses the information gathered from actual executions of scientific workflows as well as domain-specific knowledge of these workflows to generate representative and realistic synthetic workflows. We consider four different workflows generated by PWG (information on the corresponding scientific applications is available in [START_REF] Pegasus | Pegasus workflow generator[END_REF][START_REF] Juve | Characterizing and profiling scientific workflows[END_REF]):

• Montage: The NASA/IPAC Montage application stitches together multiple input images to create custom mosaics of the sky. The average weight of a Montage task is 10s. Structurally, Montage is a three-level graph [START_REF] Deelman | Pegasus: A framework for mapping complex scientific workflows onto distributed systems[END_REF]. The first level (reprojection of input image) consists of a bipartite directed graph. The second level (background rectification) is a bottleneck that consists in a join followed by a fork. Then the third level (co addition to form the final mosaic) is simply a join. • Ligo: LIGO's Inspiral Analysis workflow is used to generate and analyze gravitational waveforms from data collected during the coalescing of compact binary systems. The average weight of a Ligo task is 220s. Structurally, Ligo can be seen as a succession of Fork-Joins super tasks, that contain themselves either fork-join graphs or bipartite graphs (see the LIGO IHOPE workflow on [START_REF] Pegasus | Pegasus workflow generator[END_REF]). • CyberShake: The CyberShake workflow is used by the Southern California Earthquake Center to characterize regional earthquake hazards. The average weight of a CyberShake task is 25s. Structurally, CyberShake is harder to categorize than the other graphs. It's first part is less structured, but end with many parellel linear chains (see the CyberShake workflow on [START_REF] Pegasus | Pegasus workflow generator[END_REF]). • Genome: The epigenomics workflow created by the USC Epigenome Center and the Pegasus team automates various operations in genome sequence processing. The average weight of a Genome task depends on the number of tasks and is greater than 1000s.

Structurally, Genome starts with many parallel fork-join graphs, whose exit tasks are then both joined into an new exit task and generate new fork graphs (see the Epigenomics workflow on [START_REF] Pegasus | Pegasus workflow generator[END_REF]). In all experiments, c i = r i (checkpoint and recovery costs are identical for a task) and D = 0 (downtime is zero seconds). We focus mainly on the particular case where c i = 0.1w i , and for a MTBF of 10 3 s (except for Genome where the average weight of each task is significantly longer than for other DAGs, in which case we consider a MTBF of 10 4 s). We vary the number of tasks in each workflow from 50 to 700. Unless stated otherwise, the figures show the number of tasks on the horizontal axis and the ratio of the expected execution time (T ) over the execution time of a failure-free, checkpoint-free execution (T inf ) on the vertical axis (lower values are better). The expected execution time T is computed using the method described in Section 5.

Results

We find that our results strongly depend on the structure of the DAG, meaning that the relative performance of the heuristics vary between each workflow type. Consequently, we do not show results aggregated over all workflows. The goal of our experiments is to determine for each workflow (i) which DAG linearization strategy is best, and (ii) which checkpointing strategy is best, hoping to identify strategies that work well across different workflows. Linearization strategies -Figure 5 shows results for the CyberShake, Ligo, and Genome workflows for two checkpointing strategies, CkptW and CkptC, and for all three linearization strategies. CkptW and CkptC are the best checkpointing strategies in our results (see the discussion of the results in Figure 6 hereafter). Figure 5 does not show results for the Montage workflow. For this workload, the choice of the linearization strategy has almost no impact on the results (at most a 1% relative difference). Overall, the DF linearization is almost always the best. This makes sense as this strategy stipulates that if some work that depends on the most recently completed work can be done, then it should be done. Otherwise, by following a different branch of the workflow, one risks losing that recent work and having to do it again (or recover it). The only case where DF is not the best linearization approach is for the Montage DAG and the CkptPer heuristic (see Figure 6a). We have no explanation but since CkptPer is the worst checkpointing strategy for that workflow, this result is not particularly relevant. Finally, it is interesting to see in Figure 5b that, for the Ligo workflow, RF performs better than BF. This is because RF sometimes can be close to a DF strategy. Checkpointing strategies -Figure 6 shows results for all four workflows. For each checkpointing strategy, we only show results for the linearization strategy that leads to the best results (the line symbols indicate which linearization strategy is used). First, we note that our checkpointing heuristics always perform better than the two baseline comparators, CkptNvr and CkptAlws. Second, an interesting (but expected) result is that CkptPer does not behave well, and sometimes even worse than CkptNvr or CkptAlws. The CkptPer approach was specifically designed in the literature for divisible applications. As such, it does not account for the structure of the DAG. This causes it to make poor checkpointing choices. For instance, consider the example workload in Figure 1 with the linearization T 0 , T 3 , T 1 , T 2 , etc. It makes sense to checkpoint T 3 before executing T 1 , which is an entry task. But CkptPer may checkpoint T 1 instead because w 0 + w 3 + w 1 happens to correspond to the chosen checkpointing period. The main result from Figure 6 is that two checkpointing strategies outperform the other strategies: CkptW (for Montage, Ligo and Genome) and CkptC (for CyberShake). These two heuristics behave very differently because we have c i = 0.1w i . CkptW checkpoints the tasks by decreasing weight (hence by decreasing checkpointing time since it is proportional to the weight of the tasks), while CkptC checkpoints the tasks by increasing checkpointing time (hence increasing weight). The good performance of both heuristics in different scenarios can be explained intuitively. After finishing a long/large task, it is useful to checkpoint it as quickly as possible in case a failure occurs soon (which is what CkptW does). Conversely, checkpointing a short/small task (which may be the successor of a long task) is also useful because its checkpointing time is low (which is what CkptC does).

Examining the structure of the DAGs helps understanding the results. Indeed, Montage, Ligo and Genome are all very parallel task graphs composed of successive fork and joins. Intuitively, while this is not necessarily the optimal strategy, it makes sense to checkpoint the large tasks first when considering a join: indeed, when executed, the entry tasks of a join are put on hold while the remaining are executed, which may increase the likeliness that a failure occurs. Hence checkpointing the large tasks first may be a good strategy. On the contrary, Montage is the only DAG that does not consist in successive fork-joins, but instead comprises linear chains. When executing a linear chain, it can make sense to save only the smallest output data. In particular, when the checkpointing time depends on the size of the task, if a big task is followed by a smaller task, then it might be cheaper to execute and checkpoint the smallest task than to checkpoint the bigger task. This could explain the fact that CkptC performs well for Montage. In summary, a good strategy should be workflow-dependent and take into account the structural shape of the DAG (fork joins, linear chains).

To verify the impact of the checkpointing overhead model on the above results, Figure 7 presents similar results but with a lower c i = 0.01w i . A noticeable and expected effect of the lowered checkpointing costs is that CkptAlws performs slightly better. Otherwise, these results confirm the same trends and conclusions as that with the results shown in Figure 6.

Constant checkpoint overhead -To better assess the impact of checkpointing costs, we discuss results with a constant checkpoint cost, i.e., checkpointing costs that are independent of task weights. Expectedly, when CkptW performs better with proportional checkpoint costs it also perform better with constant checkpoint costs. This is because the ratio of the amount of computation that risks being lost over the checkpointing time will be even more beneficial to large tasks. However, for workflows where CkptC performs better, the question of the impact of constant checkpointing costs is interesting. Figure 8 shows results for CyberShake that allow a comparison of CkptW and CkptC when the checkpointing cost is constant (for c i = 10s and c i = 5s ). This plot can be compared to Figure 5a where the checkpoint is proportional to the computation. We can see that when the checkpointing cost is constant, CkptW tends to behave as well as CkptC on CyberShake workflows.

We also evaluate the impact of the checkpointing strategy with a constant checkpointing cost (c i = 5s) in Figure 9 for all four workflow types. Again, the same conclusions hold as observed previously, although CkptAlws becomes more costly for the Montage and CyberShake workflows.

Impact of λ -Finally, we wish to assess the impact of the MTBF on our conclusions. Figure 10 shows results for n = 200 nodes vs. λ. We observe the same general trends as reported above, i.e., DF-CkptW is the winning strategy in most cases regardless of the value of λ.

Summary -We have compared our heuristics over a range of experimental scenarios. In general, DF-CkptW leads to the best results, which in practice would translate to shorter makespans. DF-CkptC performs well in some cases. These performance differences depend on the structure of the DAG (CkptW performs better for fork-joins while CkptC performs better for linear chains). These differences can be discovered empirically, as done in this section, or perhaps by analyzing the underlying shape of the DAG.

Overall, the heuristics that rely on the computation of the expected makespan given in Section 5 lead to significantly better results than the baseline CkptAlws and CkptNvr approaches. A significant, if expected finding, is that taking into account the structure of the DAG is important, as highlighted by the poor results of the CkptPer heuristic.

Conclusion

In this work, we have studied the problem of scheduling computational workflows on a failure-prone platform. We have used a framework where applications are scheduled on the full platform where processors are subject to i.i.d. exponentially distributed failures. Checkpoint-rollback-recovery is used to tolerate failures. Our main contribution over previous work [START_REF] Toueg | On the optimum checkpoint selection problem[END_REF][START_REF] Bouguerra | Complexity Analysis of Checkpoint Scheduling with Variable Costs[END_REF] is that we consider general Directed Acyclic Graphs instead of linear chains. Our theoretical results include polynomial-time algorithms for fork DAGs and for some join DAGs (when the checkpoint and recovery costs are constant) and the intractability of the problem for join DAGs in general. We have also discussed the complexity of the simple problem instance where no task is checkpointed.

Our main theoretical result is a polynomial-time algorithm to evaluate the expected makespan of a schedule for general DAGs. This is a key result as it makes it possible to design heuristics for general DAGs, i.e., heuristics that can construct a schedule with a known objective. Without this result, the only way to attempt to find a good schedule would be to run numerous and likely prohibitively time-consuming Monte-Carlo experiments with a fault generator (either in simulation or on a real platform).

We have proposed several heuristics and have evaluated them for four representative scientific workflow configurations. Overall, we find that DAGs should be traversed depth-first (DF) and that checkpointing should be done by prioritizing tasks based on weight (CkptW) or checkpointing cost (CkptC). The two resulting heuristics, DF-CkptW and DF-CkptC perform differently on different workflows depending on the DAG structure. We found that a periodic checkpointing approach, although widely used for divisible applications, is not effective, precisely because it does not account for the structure of the DAG. A future direction for this work is to consider non-blocking checkpointing operations, i.e., a processor can compute a task, perhaps at a reduced speed, while checkpointing a previously executed task. Overlapping of computation and checkpointing can improve performance, but changes the problem. In particular, it would be interesting to see how our theoretical results are impacted when considering non-blocking checkpointing. A broader future direction would be to remove the assumption that the DAG is linearized, i.e., that each task executes on the entire platform. The scheduling problem then becomes much more complex since one must decide how many processors are allocated to each task, and possibly account for data redistribution costs.

Figure 1 :

 1 Figure 1: Example DAG. Tasks whose output is checkpointed (T 3 and T 4 ) are shadowed.

Figure 2 :Figure 3 :

 23 Figure 2: First example with a single dependency edge.

Figure 4 :

 4 Figure 4: Consider the DAG in Figure 1. For clarity tasks are renumbered according to their execution order in a given linearization for the DAG. The output of T 1 and T 4 are checkpointed.Assume that a single fault occurs, and that it occurs during the execution of T 5 . Then when the execution of T 6 begins, the output of T 1 and T 5 are already in memory, due to the re-execution of T 5 , and do not need to be recomputed. However the output of T 4 needs to be recovered. Hence T 4 ∈ T ↓56 . However, because it was checkpointed, we do not need to recover T 2 and T 3 . Then, when computing T 7 , the output of T 3 is needed, hence T 3 ∈ T ↓5 7 . To compute T 3 , we further need T 2 . Hence T 2 ∈ T ↓5 7 .

Algorithm 2 FindWikRik 1 : 2 : 3 : 4 :

 21234 procedure FindWikRik(k) tab k : n × n array initialized with -1 W k , R k : n arrays initialized with 0 for i = k . . . n do 5:

17 : 1 Tj ∈ T ↓k i , not yet studied 25 :if

 17125 k .(i) ← R k .(i) + rj 12: procedure Traverse(l, i, k, tab k ) 18:for Tj ∈ Pred(T l ) do 19: switch tab k .(i).(j) do for r = i + 1 . . . n do 26: tab k .(r).(j) ← 0 Tj ∈ T ↓k i =⇒Tj / Tj is checkpointed then 30: tab k .(i).(j) ← 2 31: else 32: tab k .(i).(j) ← 1 33:tab k =Traverse (j, i, k, tab k )

  CyberShake: λ = 0.001, c i = 0.1w i . Ligo: λ = 0.001, c i = 0.1w i . Genome: λ = 0.0001, c i = 0.1w i .

Figure 5 :

 5 Figure 5: Impact of the linearization strategy.

  (a) Montage: λ = 0.001, c i = 0.1w i .(b) Ligo: λ = 0.001, c i = 0.1w i . CyberShake: λ = 0.001, c i = 0.1w i .

  Genome: λ = 0.0001, c i = 0.1w i .

Figure 6 :

 6 Figure 6: Impact of the checkpointing strategy. For each checkpointing strategy, we plot the best linearization strategy.

  Montage: λ = 0.001, c i = 0.01w i . Ligo: λ = 0.001, c i = 0.01w i . CyberShake: λ = 0.001, c i = 0.01w i . Genome: λ = 0.0001, c i = 0.01w i .

Figure 7 :

 7 Figure 7: Impact of the checkpointing strategy when c i = 0.01w i . For each checkpointing strategy, we plot the best linearization strategy.

  CyberShake: λ = 0.001, ∀i, c i = 10s. CyberShake: λ = 0.001, ∀i, c i = 5s.

Figure 8 :

 8 Figure 8: Impact of the linearization strategy for a constant checkpoint.

  CyberShake: λ = 0.001, c i = 5s.

  Genome: λ = 0.0001, c i = 5s.

Figure 9 :

 9 Figure9: Impact of the checkpointing strategy when c i = 5s. For each checkpointing strategy, we plot the best linearization strategy.

  Montage: 200 nodes, c i = 0.1w i . Ligo: 200 nodes, c i = 0.1w i . CyberShake: 200 nodes, c i = 0.1w i . Genome: 200 nodes, c i = 0.1w i .

Figure 10 :

 10 Figure 10: Impact of the checkpointing strategy with different values of λ. For each checkpointing strategy, we plot the best linearization strategy.

A fork DAG with n + 1 tasks has an entry task Tentry, n exit tasks T 1 , . . . , Tn, and n edges from Tentry to each T i .

A join DAG with n + 1 tasks has n entry tasks T 1 , . . . , Tn, an exit task T exit , and n edges from each T i to T exit .

A tree-shaped DAG has an entry task (the root), and all other tasks have in-degree 1 (single parent).

We deal with out-trees here. For in-trees, the problem is trivial since all linearizations are optimal: either all tasks (including the root task) are successfully executed, or the schedule must restart from scratch.

Acknowledgments This work was supported in part by the French Research Agency (ANR) through the Rescue project. Yves Robert is with Institut Universitaire de France.