# An Efficient Fault Tolerant Workflow Scheduling Approach using Replication
Heuristics and Checkpointing in the Cloud

Amrith Rajagopal Setlur<EMAIL_ADDRESS>S. Jaya Nirmala<EMAIL_ADDRESS>Har Simrat Singh<EMAIL_ADDRESS>Sudhanshu Khoriya
<EMAIL_ADDRESS>1National Institute of Technology  Tiruchirappalli
- 620 015  India

###### Abstract

Scientific workflows have been predominantly used for complex and large scale
data analysis and scientific computation/automation and the need for robust
workflow scheduling techniques has grown considerably. But, most of the
existing workflow scheduling algorithms do not provide the required
reliability and robustness. In this paper, a new fault tolerant workflow
scheduling algorithm that learns replication heuristics in an unsupervised
manner has been proposed. Furthermore, the use of light weight synchronized
checkpointing enables efficient resubmission of failed tasks and ensures
workflow completion even in precarious environments. The proposed technique
improves upon metrics like Resource Wastage and Resource Usage in comparison
to the Replicate-All algorithm, while maintaining an acceptable increase in
Makespan as compared to the vanilla Heterogeneous Earliest Finish Time (HEFT).

###### keywords:

Workflow , Scheduling , Replication , Heuristics , Resubmission ,
Checkpointing , Clustering , HEFT

††journal: Journal of Parallel and Distributed Computing

##  1 Introduction

Scientific workflows are described as a ”useful paradigm to describe, manage,
and share complex scientific analysis” [1]. A workflow is a formal way to
express a calculation. The workflow involves multiple tasks of different sizes
and characteristics, with control and data dependencies between them. They
also capture the various parameters of the task such as their input, output
etc.[2]. Workflows have emerged as comprehensive tools for managing complex
computations and managing storage requirements. They are used in a lot of
applications like neuroscience, high-energy physics and genetics.

Many authors have studied the advantages of using the cloud environment for
executing scientific workflows [3, 4, 5] and have claimed that the cloud
environment enables workflow execution with low cost and that the
virtualization overhead due to the cloud would be very minimal (in the range
of 1% to 10%).

Efficient scheduling of scientific workflows helps in reducing the makespan or
execution time, in meeting deadlines and in minimizing the cost. As the
problem of scheduling tasks simultaneously on multiple processors with a start
and an end time is NP-complete, researchers have relied on heuristic and meta-
heuristic optimization techniques to schedule them.

Failures in scientific workflows increase the makespan and waste a lot of
workforce and time. The different types of failures that can occur during an
execution are task failures, VM failures and workflow-level failures. Fault
tolerance for scientific workflows can be provided either at the task-level or
workflow-level [6]. Task-level scheduling techniques involve retry,
checkpointing and the use of alternate resources for the same task. On the
other hand, workflow level scheduling involves the usage of alternate tasks,
redundancy, user-defined exception handling and rescue workflows. In this
paper, a new fault tolerant workflow scheduling approach called Checkpointing
and Replication based on Clustering Heuristics (CRCH) is proposed. It uses
replication, resubmission, checkpointing and provides fault-tolerance in an
efficient manner.

In the scheduling step, the workflow tasks are replicated and then scheduled.
The multiple copies prevent the task from failing and increase the probability
of its successful completion. If one copy fails, one of its replicas is
scheduled and executed [7, 8]. Task resubmission [9, 7] is also widely used
for fault tolerance in the workflow scheduling. It takes place during the
execution phase. In task resubmission, the failed task is resubmitted either
to the same or different resource. The resource usage and wastage in task
resubmission is less as compared to replication but the execution time may be
more. Replication generates identical copies of a task. Hence they have same
dependencies, and thus sufficient parallel systems can afford to execute them
in parallel saving execution cost. Checkpointing [9] [8] is one of the time
efficient fault tolerant methods. In synchronous checkpointing, the states of
the tasks or processes are saved promptly at regular intervals. Whenever a
Virtual Machine (VM) fails, the process starts from the previously saved
state. Thus, this method is gainful over methods which reschedule a duplicate
of the task.

One of the key contributions of this paper is an unsupervised way of learning
replication counts for tasks. In comparison to other replication heuristics
[7], this approach is much quicker and robust, as it doesn’t involve exploring
every possible solution (HEFT schedules with varying sets of replicas) in a
combinatorial optimization problem. Along with this, a checkpointing mechanism
that stimulates dynamic resubmission of tasks on the most optimum resource has
been proposed. In an elaborate analysis of well established metrics like
Resource Usage, Resource Wastage and Total Execution Time it has been shown
that the algorithm proposed performs better than the existing state-of-the-art
workflow scheduling techniques even in highly faulty environments.

The outline of the paper is as follows: Section 2 discusses some of the novel
and significant progress made in the field of fault tolerant workflow
scheduling. Section 3 and Section 4 discuss in depth the algorithms we
propose, followed by a performance benchmarking against state-of-the-art
methods. The concluding remarks along with future research is presented in
Section 5.

##  2 Related Work

Yu and Buyya [10] give a brief overview of the various fault tolerant workflow
scheduling techniques. Fault tolerance to workflow applications are provided
either at the task or workflow level. Replication of tasks or data,
Resubmission, Checkpointing and Alternate Resource are widely used techniques
at the task-level, whereas Alternate Task, User-defined Exception Handling and
Rescue Workflow are widely used at the workflow level. Poola et al. [9] give a
comprehensive survey of the fault tolerant techniques employed in various
Workflow Management Systems (WFMS). They also present a detailed taxonomy of
the different techniques employed for fault tolerance in distributed
environments. Also, the paper discusses a variety of metrics used for
quantifying fault tolerance. [11] discusses the fault-tolerant techniques
employed in various grid WFMS. The survey reveals that resubmission techniques
are most widely used for providing fault tolerance in workflows followed by
replication and checkpointing.

Poola et al. [12] use the concept of slack time to generate robust schedules
for scientific workflows to enable them to withstand failures in the cloud
environment. They use a common set of parameters to model the stochasticity of
all VMs. They also assume that there exists no resource contention, which can
be a strong assumption in highly faulty environments. In this proposed
methodology, resource failure parameters are sampled from various
distributions, thus making the system more robust. Plankensteiner et al. [7]
estimate the replication count of a task from its Resubmission Impact (RI)
heuristic. Their approach creates multiple workflows, each with a particular
task duplicated by a constant value. The replication count is estimated from a
normalized score assigned to each task, based on how much they impact the
execution time (had they been replicated). On the other hand, the approach in
this paper infers replication count using an unsupervised machine learning
algorithm that gives more accurate estimates and saves the time involved in
re-computing HEFT schedules.

Zhang et al. [8] integrate the vanilla HEFT [13]/ Duplication Scheduling
Heuristic (DSH)[14] schedules with the over-provisioning algorithm proposed by
Kandaswamy et al.[15]. To meet the constraints on the overall workflow DAG
(Directed Acyclic Graph) success probability, which diminishes exponentially
with the addition of tasks to the workflow, the entire DAG is over-provisioned
on a distinct set of resources. This leads to increased Resource Usage. The
over-provisioning algorithm proposed by [15] finds the solution for a
combinatorial optimization problem, which meets both performance and
reliability constraints for a task. Although the assumption of independent
binomial distributions for resource failures seems reasonable, the computation
of expected execution time of a task on a resource cannot be agnostic to the
state of the current workflow execution.

Chen and Deelman [16] introduce horizontal/vertical task clustering based on
the workflow structure. Having defined Gamma/Weibull distributions, for task
runtime, overhead time and job (collection of tasks) runtime, they use Maximum
Likelihood Estimates (MLE) for the distribution parameters, along with
corresponding conjugate priors for dynamic estimation. For the parameter
estimation to converge, large datasets of task/job runtimes are required. But
the clustering technique proposed in this paper relies on grouping dense task
embeddings. These embeddings are based on task/task-neighborhood structural
characteristics (like edges, DAG order, etc.)

Zhang et al. [8] use the technique to find the smallest subset of resources to
replicate the tasks such that they satisfy their performance and reliability
constraints. If the smallest subset of resources could not be found, the
success probability for all the resource combinations are calculated, and the
tasks are replicated on the resource set with highest success probability. The
method proposed by [7] does not use checkpointing, but resubmits a task when
all of its replicas have failed. Resubmission of the whole task significantly
increases the execution time of the task, which in turn increases the
workflow’s makespan. But the fault tolerant approach proposed in this paper
employs replication heuristics and light-weight checkpoint/ restart techniques
at the task level. The replication heuristic employed calculates the number of
replications needed for each of the tasks in the workflow, and thereby reduce
the resource waste and execution cost. Light-weight checkpointing enables the
system to have minimal stable storage, and the transfer of intermediate data
more manageable, and hence reduces execution time.

Other popular checkpoint/restart algorithms include the Scalable
Checkpoint/Restart (SCR) [17] and the FTI [18] algorithms. SCR is an
asynchronous multi-level checkpointing algorithm that employs a combination of
frequent inexpensive checkpoints and resilient time consuming backups to the
Parallel File System (PFS). The checkpoint overhead of the vanilla SCR
algorithm has been compared with CRCH in the latter half of the paper.

##  3 Proposed Methodology

Figure 1: CRCH Algorithmic Design Table 1: Notation Index Notation | Explanation
---|---
$EST_{t}$ | Estimated Start Time of task $t$
$EFT_{t}$ | Estimated Finish Time of task $t$ as decided by Algorithm-2.
$AST_{t}$ | Actual Start Time of task $t$
$AFT_{t}$ | Actual Finish Time of task $t$
$TET$ | Total Execution Time of the Workflow
$w_{t}$ | Average execution time of task $t$
$e(t,t^{\prime},d)$ | Average time to transfer $d$ units of data from $t$ to $t^{\prime}$
$timeOnVm(t,r)$ | Time taken for task $t$ to execute on resource $r$
$dataTransfer(r,r^{\prime})$ | Data transfer rate between $r$ and $r^{\prime}$
$taskList$ | List of tasks in workflow
$vmList$ | List of resources available
$dependenciesList$ | {(t,t’,d) $\mid$ $t^{\prime}$ is a parent of $t$ that sends $d$ units of data}
$Pa_{t}$ | All the parents of task $t$
$failures_{t}$ | Number of times task $t$ has failed to complete execution
$repCount_{t}$ | Replica count for task $t$
$features_{t}$ | Set of features for task $t$
PCA | Principal Component Analysis
COV | Coverage of Variance
$\alpha_{t}$ | Number of completed checkpoints for task $t$
$replicas_{t}$ | Set of replicas generated for the given task $t$
$isBusy(v)$ | Returns true if the VM $v$ is busy executing a backlog of tasks

HEFT is a suitable base algorithm for scheduling the tasks of a workflow.
Wieczorek et al. [19] analyze and evaluate the performance of HEFT, Genetic
Algorithms (GA) and simple ”myopic” for scheduling scientific workflows. Their
results show that the full-graph scheduling technique with HEFT algorithm
performs best when compared to other strategies. Hence, it is decided to use
HEFT to determine the initial schedule of tasks and their corresponding
replicas. Furthermore, the performance of CRCH in faulty environments has been
benchmarked against that of HEFT. The basic HEFT algorithm does not involve
any fault tolerance. If a resource fails, any task executing on that resource
also fails and therefore the workflow itself fails. One approach to providing
fault tolerance is to generate replicas of the tasks. Even if the task fails,
the replicas can generate the intermediate results. [8] mentions a
ReplicateAll algorithm that replicates each task by a constant factor. This
redundancy improves the probability of completion of the workflow, although it
increases the Total Execution Time (TET) or Makespan by a huge margin. [9, 7]
use resubmission wherein a task that could not complete execution due to a
failed resource is resubmitted on another. This method suffers from the loss
of computation involved in resubmitting a task that has almost completed. [9,
8] use checkpointing wherein the states of the tasks are saved at regular
intervals so that upon resubmission they can start execution from the
previously saved state. Table 1 elucidates the various terms/abbreviations
referenced throughout the paper.

###  3.1 CRCH Algorithm

Figure 1 highlights the data flow path, processing involved in the proposed
method, CRCH. It consists of three modules namely Clustering, Replication and
Checkpointing. The Clustering phase facilitates the computation of replication
counts based on the properties of tasks. In the Replication phase, the tasks
are replicated by applying the replication heuristics and the standard HEFT
algorithm decides the overall schedule. In the final Checkpointing and
resubmission phase, a light weight strategy is used, wherein only pointers to
the saved state are stored.

The average execution time of the task $w_{t}$ and the average transfer time
$e(t,t^{\prime},d)$ are represented using the Equations (1) and (2)
respectively.

$w_{t}=\dfrac{\sum_{r\,\epsilon\,vmList}{}timeOnVm(t,r)}{vmList.size()}$ (1)

$e(t,t^{\prime},d)=\dfrac{\sum_{\forall\,r\,,r^{\prime}\,\in\,vmList}{}d/dataTransfer(r,r^{\prime})}{|{\\{(r,r^{\prime})\,\,\forall\,r\,,r^{\prime}\,\in\,vmList}\\}|}$
(2)

Further, each task is represented in an $n$-dimensional space as a point. Let
task $t_{i}$ be denoted with feature vector
$F_{i}=[x_{1i},\,x_{2i},\,...,\,x_{ni}]$, that is, each task has $n$ features
associated with it. These features can be nominal (priority of a task) or
numeric (average execution time). The axes of the $n$-dimensional space denote
the various features. Some of the possible features are :

  1. 1.

Average execution time of a task : $w_{t}$

  2. 2.

Average time to transfer data from the parents of the task :

$e(t)=max_{(t,t^{\prime},d)\,\in\,dependenciesList}\,(e(t,t^{\prime},d))$

  3. 3.

Priority of task

  4. 4.

Number of parents:

$\|\\{t^{\prime}\,|\,\,\exists\,e(t,t^{\prime},d)\in dependenciesList\\}\|$

  5. 5.

Number of children:

$\|\\{t^{\prime}\,|\,\,\exists\,e(t^{\prime},t,d)\in dependenciesList\\}\|$

####  3.1.1 Clustering Module

The replication count for each task can be determined using multiple Machine
Learning techniques ranging from Supervised Classification like Logistic
Regression, Max Entropy Models, etc. to Unsupervised Classification like
Clustering, LSA, PLSA [20, 21]. In general, this can be treated as a multi-
class classification problem, where the inputs are task representations and
the target values are one-hot vectors $y_{i}$ ($y_{i_{j}}$=1 if replication
count for the input task $X_{i}$ is $j$). When substantial labeled training
data is present, a Multilayered Perceptron (MLP) works reasonably well. For
each target class $j$, there exists a weight vector $W_{j}$. $P_{j}(t_{i})$
represented by Equation (3) denotes the probability of task $t_{i}$ to have
replication count $j$, where $F_{i}$ is the feature representation for task
$i$.

$P_{j}(t_{i})=\dfrac{exp(F_{i}\cdot W_{j})}{\sum_{k}exp(F_{i}\cdot W_{k})}$
(3)

Equation (3) is a standard softmax formulation, and the MLP can be trained
with a cross entropy loss (Equation (4)) and an optimizer like Stochastic
Gradient Descent, Batch Gradient Descent, or for faster convergence RMSprop or
Adam is used [20].

$Loss=\frac{1}{N}\sum_{i}{}(\,-\sum_{j}{}s_{ij}\cdot log(p_{ij}))$ (4)

where $S_{i}=[s_{i1},\,s_{i2},.....,\,s_{iK}]$ is a one hot encoding vector
denoting the class to which the $i^{th}$ observation belongs and
$P_{i}=[p_{i1},\,p_{i2},....,\,p_{ik}]$ is a vector denoting the probability
distribution over $K$ classes predicted by the logistic classifier for the
$i^{th}$ observation.

To generalize well, this method requires a training set of large size with
already existing replication counts for a task in multiple workflows [20].
Since such a set of observations is not readily available and needs to be
compiled over a period of time, a move towards an unsupervised classification
scheme is adopted. One such method is Principal Component Analysis (PCA) [21]
where the feature vectors are projected onto less than $K=|F|$ component
vectors, so that the coverage of variance in the data is greater than a
predefined threshold. PCA can be attributed to recognize correlated features,
which help in representing the points in a lower dimensional space. These
principal components are orthogonal unit vectors. With the addition of each
such vector, the portion of variance covered with respect to the original
variance in the data is improved. This improvised representation not only
facilitates faster clustering but also prevents over-fitting to a certain
extent [21]. Since the features used are of different scales like numeric
regarding the average execution time of a task or the number of
parents/children of a task to ordinal in case of priority, the data needs to
be standardized before the application of PCA. PCA predominantly uses
correlation/ covariance matrix of the set of features. The analysis has been
done on a training set of 100 points, each representing a task in a ten
dimensional space, with a covariance threshold of 0.8. Intuitively, it can be
seen that the number of parents is positively correlated with the average
transfer time of data from parents. PCA would help remove such co-dependence.
Before performing PCA, the data is mean subtracted and standardized
(whitened). Since a dataset like task-features can be enormous regarding the
basic characteristics of a task (execution time, priority) and its basis
expansion would be even higher, a method like PCA would help identify coherent
features like criticality against the number of dependents of the task. Steps
2-9 in Algorithm 1 focus on determining the set of principal components ($K$
of them) and step 10 deals with the projection of the original points onto the
new basis.

CRCH involves hierarchical clustering on the projected points [21, 20]. The
distance between two clusters, characterized using the average Euclidean
distance between every pair of points belonging to the two clusters is given
by Equation (5). This can be referred to as the affinity between the two
clusters [22]. In each iteration, clusters are merged to form superclusters,
based on their affinity to neighbors. The agglomerative clustering strategy
used has been derived from the triplet loss method which has been popular in
clustering high dimensional deep neural embeddings [22].

Triplet loss validates that the supercluster under consideration $C_{m}$,
would be merged with one of the neighbors $C_{n}$ not only based on the
closeness of the two superclusters but also on the additional condition that
any other $C_{k}\neq C_{n}$ ($C_{k}$, $C_{n}\in\eta(C_{m},R)$, where
$\eta(C_{m},R)$ is the set of $R$ closest superclusters to $C_{m}$) is much
further away from $C_{m}$. The pair of clusters that minimize Equation (6) is
merged into a supercluster for the subsequent time step. At any given time
step, let $D_{ij}$ represent the distance between two superclusters, $C_{i}$
and $C_{j}$. Then,

$D_{ij}=\frac{\sum_{p_{i}\in C_{i}}\sum_{p_{j}\in
C_{j}}||p_{i}-p_{j}||_{2}}{|C_{i}|\dot{|}C_{j}|}$ (5)

$loss(C_{i},C_{j})=D_{ij}+\frac{\lambda}{R-1}\sum_{k\in\eta(C_{i},R)}(D_{ij}-D_{ik})$
(6)

The time step at which the clustering would converge is decided by the
dendrogram generated by the superclusters [21, 20]. At each level the branches
in the dendrogram reduce, and there exists a point at which the minimum inter-
cluster distance exceeds a certain threshold, the number of branches in the
dendrogram at this time step is indicative of the final number of
superclusters [21, 20]. Once steps 11-17 are done building the superclusters,
steps 18-19 deal with the assignment of replication counts, which can be
decided by the size and other summary statistics of the superclusters like
average execution time, average priority, etc.

Figure 2 shows the state of the agglomerative clustering procedure at a time
step $t$. Each blob denotes the latent representation of a task, while the
dotted lines represent the inter-cluster distances for tasks $C1$ and $C2$
(distances not impacting the loss have not been shown). Blobs in the same
color belong to the same supercluster. $R$ and $B$ denote the figurative
cluster centers for the superclusters in $red$ and $blue$. In a traditional
clustering approach, $C1$ would end up getting merged with its closest
neighbor. In the presence of triplet loss, $C2$, which is not as close as $C1$
is to its closest neighbor, but is considerably far from the remaining $R-1$
neighbors forms the green supercluster. This prevents the agglomeration from
collapsing into superclusters of sizes that lie on either end of the spectrum.
For example, in traditional approaches, it is not uncommon to notice
agglomerative clustering from converging into either very huge clusters or
very tiny ones (size$<=$3). Figure 3 shows the progress of supercluster
formation across iterations along with the assignment of replication counts
based on the final sizes of the superclusters.

Figure 2: Action taken at a given time step by triplet loss (left) and
agglomerative clustering (right).
(to be viewed in color) Figure 3: Clustering progress across iterations. (to
be viewed in color)

Many scientific workflows exhibit properties wherein most of the tasks can be
segmented into a few large clusters. Each segment encompasses tasks having
similar features, and they end up having low replication counts. The outliers
based on high priority or high average execution time belong to clusters of
much smaller sizes and hence are assigned a higher replication count. Such an
assignment policy may lead to cases wherein a low priority task taking less
time to execute (on average) gets allotted more replicas than needed as it
ends up being an outlier. Simple rule ensembles, based on sufficient
statistics of the feature values in the supercluster can be learned to avoid
this.

1:procedure replicationCount(taskList, threshold, K)

2: Let COV = 0.0 $\triangleright$ Coverage of Variance

3: Let principalComponents = {}

4: Let dataSet = $\\{\textit{t.features}\,|\,t\epsilon\textit{taskList}\\}$

5: do

6: Let PC = nextPrincipalComponent(dataSet) $\triangleright$ Orthogonal unit
vector

7: COV = COV \+ variance(PC) $\triangleright$ Additional variance accounted

8: principalComponents = principalComponents $\bigcup$ PC

9: while $\textit{COV}<\textit{threshold}$

10: dataSet = dataSet $\times$ principalComponents $\triangleright$ Project
points in lower dimensional space

11: Let clusters = dataSet

12: do

13: Let newClusters = hierarchicalClustering(clusters) $\triangleright$
triplet loss

14: Let oldClusters = {C $|$ C $\in$ clusters $\wedge$ $\exists$ C’ $\in$
newClusters $\wedge$ C $\in$ C’}

15: clusters = (clusters \- oldClusters) $\bigcup$ newClusters

16: while $|\textit{clusters}|>\textit{K}$

17: Sort clusters based on the size of each cluster in descending order of
size (or other statistics)

18: for $C_{i}$ $\in$ clusters do

19: $repCount_{t}$ = i $\forall$ t $\in$ $C_{i}$ $\triangleright$ Assign
replication counts

Algorithm 1 Replication Count Algorithm

####  3.1.2 Replication Module

The tasks part of the initial workflow graph are the original tasks, and their
duplicates are referred to as replicas. Once Algorithm 1 has ascertained the
replication count for each task, Algorithm 2 defines the HEFT schedule for
each original task and its replicas.

The original tasks are sorted in descending order of their estimated start
times (calculated based on the inherent dependencies). Once an initial task is
scheduled on a resource, the algorithm checks whether all of its original
sibling tasks have also been scheduled. If so, the replicas of this initial
task are mapped to resources based on which the resources would be allocated
an execution interval with the minimum EST.

The following section discusses the influence of task checkpoints, on
resubmissions (due to resource failures) at runtime. Here, it is assumed that
there exists a subset of VMs which are entirely reliable (non-failing). Other
non-reliable resources, may not be available when a task is scheduled to
execute or may go down during the execution of a task. Resource failures cause
delays in task execution and this results in AFT (Actual Finishing Time) being
greater than EFT (Estimated Finishing Time). Such delays, when propagated
along the critical path of the workflow have a direct impact on TET. Some
proportions of the execution delays for tasks that are not on the critical
path tends to get subsumed under the slack time that occurs between two
noncritical tasks. It may also be possible that a resource is busy executing a
backlog of tasks (in HEFT order), right when a task scheduled to run on that
resource has met its dependencies, and as a result ends up waiting for that
resource.

Algorithm 2 HEFT with Over-Provisioning

1:procedure HEFT(taskList, dependenciesList)

2: for $t\in\textit{taskList}$ do

3: Calculate $taskRank_{t}$ = B-Level of the task in critical path calculation

4: sort $taskList$ based on $taskRank_{t}$ $\forall\,t\,\epsilon\,taskList$ in
descending order

5: for $t\in\textit{taskList}$ do

6: Schedule $t$ on $v$ with minimum $EST_{t}$

7: if $(t,t^{\prime},d)\in\textit{dependenciesList}$ then

8: if
$t^{\prime\prime}\,is\,scheduled\,\,\forall\,(t^{\prime\prime},t^{\prime},d)\in\textit{dependenciesList}$
then

9: Schedule replicas of $t^{\prime}$ on VMs with minimum ESTs

####  3.1.3 Checkpointing Module

Algorithm 3 defines the execution of a workflow at runtime concerning
checkpointing and resubmission. The environment is modeled to have a set of
resources which fail depending on their respective Mean Time To Repair (MTTR)
and Mean Time Between Failures (MTBF). The size and duration of these failures
is dependent on the type of environment (Stable, Normal or Unstable). In the
case of a perfect environment with no failures and assuming no checkpoints are
involved, the TET ($TET_{perfect}$) is determined by the task schedule, as
described in section 3.1.2. At any given point, let $\alpha_{i}$ be the number
of checkpoints that have been completed for $t_{i}$. If task $t_{i}$ is
scheduled to start on $v_{j}$ at $EST_{i}$ and has $runtime_{ij}$ then,
$TET_{perfect}$ is represented using Equation 7.

$TET_{perfect}=\max_{t_{i}\,on\,v_{j}}(EST_{i}+\,runtime_{ij})\,\forall\,\,i\in
taskList$ (7)

A schedule S is a set of tasks with $t_{i}$ scheduled on a VM $v_{i}$ from
$EST_{i}$ to $EFT_{i}$. Only when $v_{i}$ belongs to a set of Failing VMs
($FVM$), $t_{i}$ may or may not be rescheduled on a nonfailing VM, that is, VM
$\notin$ FVM. The two cases to consider here are whether $v_{i}$ fails during
the execution of $t_{i}$ (Case 1) or $v_{i}$ is down at $EST_{i}$ (Case 2).
$\alpha_{i}$ is the number of checkpoints that have been completed for
$t_{i}$. These checkpoints are global and are synchronized, that is, at each
checkpoint, the working memory is dumped to a nonvolatile stable storage
associated with each VM. When a task $t_{i}$ completes its execution on
$v_{i}$, the result of its execution is stored in the stable storage
associated with $v_{i}$. The pointer to the location on stable storage is
stored in a global memory. This pointer can be referenced using a hash value
of the task id for quick access. Along with the pointer, if $v_{i}$ $\in$
$FVM$, the results are also stored in the global memory. If $children_{i}$ is
the set of tasks that use the results generated from $t_{i}$, then the data
can be fetched by $child_{k}$ $\in$ $children_{i}$ using pointers obtained
from dereferencing the global memory. In case the VM on which the parent was
executed is down, the data can be fetched directly from the global memory.
Each VM $v$ $\in$ $FVM$ has a set of time intervals denoted by $L_{v}$ which
indicate the intervals in time when the resource would be unavailable to
execute a task that has been scheduled on it (using $MTTR_{v}$, $MTBF_{v}$
sampled from log-normal and Weibull distributions respectively).

A task may fail to comply with the HEFT schedule for one of the following
reasons.

  * 1.

Resource fails during the execution of a task.

  * 2.

Task is scheduled to start on a VM that is currently down.

  * 3.

Task is scheduled to start on a VM that is executing a backlog of tasks.

Algorithm 3 CheckpointHEFT

1:while $\exists$ t such that $t$ is not scheduled do

2: schedule t on VM v using Algorithm 1, Algorithm 2

3: if isBusy($v$) then

4: if $\displaystyle EFT_{t}=max_{t^{\prime}\in replicas_{t}}EFT_{t^{\prime}}$
then $\triangleright$ $t$ is the last unscheduled replica

5: Wait for $v$ to be online

6: else

7: Terminate $t$

8: $failures_{t}\leftarrow failures_{t}+1$

9: if $v\in FVM$ then

10: if t has started execution on v then

11: $(X,Y)\leftarrow\operatorname*{argmin}_{(x,y)\in L_{v}\,\wedge\,x\geq
AST_{t}}(x-AST_{t})$

12: if $AFT_{t}\leq X$ then $\triangleright$ The task finishes before X

13: Continue

14: $failures_{t}\leftarrow failures_{t}+1$

15: if $failures_{t}$ = $repCount_{t}$ then

16: $minEST\leftarrow
minimum\;EST\;for\;t\;on\;a\;VM\;v_{new}\;st\;v_{new}\;is\;nonfailing$

17: $vmWithMinEST\leftarrow v_{new}$

18: $executionSaved\leftarrow t.\alpha\times\lambda$

19: $overhead\leftarrow executionSaved$

20: if ($minEST$ \+ $overhead$) $<$ Y then

21: $Schedule(t,vmWithMinEST)$

22: else

23: $Schedule(adjustExecutionTime(t),v)$ $\triangleright$ Reduced Execution
Time

24: else$\triangleright$ $t$ is scheduled on a resource that is currently down

25: $failures_{t}\leftarrow failures_{t}+1$

26: if $failures_{t}$ = $repCount_{t}$ then

27: $(X,Y)\leftarrow\operatorname*{argmin}_{(x,y)\in L_{v}\,\wedge\,x\leq
AST_{t}}(AST_{t}-x)$

28: $minEST\leftarrow
minimum\;EST\;for\;t\;on\;a\;VM\;v_{new}\;st\;v_{new}\;is\;nonfailing$

29: $vmWithMinEST\leftarrow v_{new}$

30: if $minEST$ $<$ Y then

31: $Schedule(t,vmWithMinEST)$

32: else

33: $Schedule(t,v)$

Light-weight checkpointing reduces the load on the global memory and thus
reduces the chance of a possible bottleneck that can result from a large
number of memory accesses.

Algorithm 3 takes as input the HEFT schedule defined by Algorithm 1 and
Algorithm 2. There is a possibility that a task $t_{i}$ scheduled on a failing
VM $v_{i}$ at $EST_{i}$ cannot begin execution because of a backlog of tasks
that are currently executing on $v_{i}$. If $t_{i}$ is the last replica of a
given task which has not executed successfully, it needs to wait until a VM is
available, else it can be terminated and deemed as a failure (Steps 3-8).
Tasks that have their dependencies satisfied and yet can’t be executed on
their scheduled resource due to the backlog of tasks being executed on them
can be deemed as failures. In unstable environments where a decent proportion
of tasks aren’t able to find their scheduled resource free, even when their
dependencies are fulfilled, it has been found that it is better not to
consider them as failures, as it would lead to unnecessary resubmissions on a
set of few non failing VMs, thus forming a bottleneck. This would simply
increase resubmissions and would not exploit the usage of replicas. Step 9
checks if a task is scheduled on an unreliable VM, and if so, whether the task
has begun execution or not. Step 11 defines variables X,Y which represent the
points in time during which the resource in question is down. If the task
completes before the next time interval when the resource is down, execution
proceeds normally (Steps 12-13). Steps 16-17 identify the nonfailing VM with
the minimum EST whereas steps 18-19 calculate the proportion of the task
completed (in terms of execution time), which would constitute the overhead
involved in scheduling the task on a different resource. If the task is to be
scheduled on the same VM it can resume from the latest checkpoint, if not it
pays the overhead of re-executing already completed checkpoints. Steps 20-23
describe this comparison. Steps 14-15 and 25-26 ensure that a task gets
resubmitted only if and only if all its preceding replicas including itself
have failed execution.

The case when the resource is down at $EST_{i}$ for $t_{i}$ is dealt with by
comparing the minimum possible EST at a non-failing VM with the point in time
when the currently down resource would be available at the latest (Steps
26-32).

Since failure count of an original workflow task is a task level attribute,
multiple processes running on different VMs may contest to update the count
simultaneously. This justifies the need for a cache coherence strategy like
MESI to avoid inconsistencies.

###  3.2 Dynamic Checkpoint Interval

The checkpoint interval $\lambda$ is a global parameter which defines the
interval between two consecutive checkpoints. Each checkpoint has an overhead
$\gamma$ associated with it. To improve the TET, $\lambda$ can hold an optimal
value depending on the environment of execution. In a highly stable
environment failures are rare. Hence TET would reduce with a higher $\lambda$.
Hence higher $\lambda$ would mean less number of total checkpoints and reduced
total checkpoint overhead. In the case of an unstable environment with a large
number of failures, TET may improve with a smaller $\lambda$, as the data lost
from having checkpoints further apart would be the delimiting factor instead
of the overhead of checkpoints.

Checkpoint interval analysis is heavily dependent on the critical path. A
critical path of a DAG is a path from an entry node to an exit node, whose
length is the maximum [23]. To compute the critical path of the workflow after
the addition of replicas and for determining the schedule using Algorithm 2, a
simple backtracking approach is sufficient. Let $V_{max}$ be the VM on which
the task $t_{max}=\operatorname*{argmax}_{t}{EFT_{t}}$ is executed. Since
$t_{max}$ is scheduled on a VM with minimum possible $EST_{t_{max}}$, there
exists at least one such task $t_{max-1}$, where
$EFT_{t_{max-1}}=EST_{t_{max}}$. By induction, the critical path can be
backtracked.

This section highlights a proof which suggests a varying checkpoint interval
for an optimal TET with the same level of fault tolerance. It assumes that
there is no shift in the critical path with the involvement of resubmissions.

######  Lemma 3.1.

The checkpoint interval $\lambda$ introduced in the CRCH algorithm is
environment dependent and can be modified to best suit the TET of the workflow
in a given environment.

######  Assumption 1.

The waiting time of a task is distributed normally, and the probability
density function depends on the ancestors of the task, environmental
parameters. The range of values observed can only be explained by a normal
distribution which spans $[\mu-3\sigma,\mu+3\sigma]$. The set of observations
for waiting times of tasks taken over a large range of inputs closely mimicked
a normal distribution, indicated by a QQ (quantile-quantile) plot [24]. Upon
reasonable & realistic variations in $\lambda$, while maintaining the same
environmental conditions it was found that the shift in the distribution
parameters is negligible.

######  Assumption 2.

The probability of a task to fail on the scheduled resource is independent of
$\lambda$.

######  Assumption 3.

The resubmission of ancestor tasks does not influence the probability of a
task to fail on the resource it is meant to execute on.

We go on to prove Lemma 3.1 based on the aforementioned assumptions. Notations
used in the proof are as follows :

  * $\lambda$ : checkpoint interval (time interval between two successive globally synchronized checkpoints )

  * $\gamma$ : overhead at each checkpoint

  * $TET_{CRCH}$ : expected time to execute CRCH algorithm (equation 11)

  * $CO$ : checkpoint overhead (equation 10)

  * $TET_{CRCH/CO}$ : expected time to execute CRCH excluding $CO$ (equation 9)

  * $TET_{C_{i}}$ : expected time taken for task $t_{i}$ to execute in CRCH (eqution 8)

  * $CP$ : critical path - a set of tasks lying on the critical path

  * $WT_{i}$ : expected waiting time for task $t_{i}$

  * $RO_{i}$ : expected resubmission time for task $t_{i}$

  * $TET_{H_{i}}$ : time taken for task $t_{i}$ to execute in HEFT

###### Proof.

According to the definitions mentioned above:

$TET_{C_{i}}=TET_{H_{i}}+WT_{i}+RO_{i}\\\ $ (8) $TET_{CRCH/CO}=\sum_{i\in
CP}TET_{Ci}\\\ $ (9) $CO=TET_{CRCH/CO}\times\dfrac{\gamma}{\lambda}\\\ $ (10)
$TET_{CRCH}=TET_{CRCH/CO}+CO$ (11)

If the task $t_{i}$ is the last of its replicas, it may not follow the
schedule given by Algorithm 2 due to:

  1. 1.

All the parents in $Pa_{i}$ haven’t completed execution that is, there exists
one task $t_{p}$ $\in$ $Pa_{i}$ st. $\forall$$t_{j}$ $\in$ $replicas(t_{p})$,
$t_{j}$ hasn’t completed execution.

  2. 2.

The resource on which the task has been scheduled is executing some backlog of
tasks.

  3. 3.

The resource is unavailable, as it has failed.

$WT_{i}$ (equation 12) depends on $A(i)$, that is, the ancestor graph for
$t_{i}$ (which includes the VMs on which the tasks have been scheduled & their
corresponding MTBFs, MTTRs). $N_{w}(\mu_{w},\sigma_{w})$, is a Gaussian
distribution which determines the expected amount of time a task has to wait
on its immediate parents and the sufficient statistics are functions of
$A(i)$. (Assumption 1) Thus:

$WT_{i}\sim N_{w}(\mu_{w}(A(i)),\sigma_{w}(A(i)))$ (12)

$RO_{i}$ is the overhead involved in resubmission. A task $t_{i}$ is
resubmitted only if it is the last replica and all the other replicas have
failed, that is,

$\displaystyle EFT_{i}=\max_{r\in replicas(i)}EFT_{r}$ (13)

The following calculations for the probability of failure of a task and the
corresponding expected resubmission overhead are only done for tasks compliant
with equation (13).

The probability that a task $t_{i}$ fails, defined as $P_{t_{i}}$, depends on
the probability that the task has been scheduled on a failing VM $v_{i}$ and
the probability for an intersection of the execution time with the interval
$L_{i}$ where

$\displaystyle L_{i}=\\{(X,Y):Y-X=mttr_{i}\text{ and $v_{i}$ fails during
$(X,Y)$}\\}$ (14)

Since the set of failing VMs is chosen from uniform distribution:

$P_{v_{i}}=\dfrac{|FVM|}{|V|}$ (15)

$P_{v_{i}}$ is the probability of a VM $v_{i}$ to fail

$FVM$ is the set of failing VMs

Now to evaluate the probability of interval overlap, $MTBF_{i}$ and $MTTR_{i}$
are needed which are determined from Weibull distributions $W_{MTBF}$ and
$W_{MTTR}$. Thus:

$P(overlap)=g(EST_{i},EFT_{i},W_{MTBF},W_{MTTR})\\\ $ (16)
$Using\,\eqref{6}\eqref{7}:\,\,P_{t_{i}}=P(overlap)\cdot\dfrac{|FVM|}{|V|}$
(17)

As the executions of all replicas are mutually independent, with $NF_{i}$
being the number of failing replicas of $t_{i}$ :

$P(NF_{i}=R_{i})=\prod_{\small r\in replicas(i)}P_{t_{r}}$ (18)

$NF_{i}$ is the number of failing replicas of task $t_{i}$

$R_{i}$ is the total number of replicas of $t_{i}$

If $E(RO)$ is the expected overhead in resubmissions then by using (18):

$RO_{i}=P(NF_{i}=R_{i})\cdot E(RO)$ (19)

Let $new$ be the resource on which $t_{i}$ is resubmitted. Hence
$P(new=v_{i})$ is the probability that the task is resubmitted on the same
resource $v_{i}$ on which it was originally submitted. $new$ follows a
multinomial distribution conditional on the schedule decided by 2 and any
other workflow or resource parameters (including $\lambda$). Overhead due to
submitting on the same resource:

$Overhead_{same}=E(minEST_{same})+PF_{i}-\bigg{\lfloor}\dfrac{PF_{i}}{\lambda}\bigg{\rfloor}\cdot\lambda$
(20)

where

$PF_{i}$ is the point in time where the execution of $t_{i}$ was stopped on
$v_{i}$ due to failure with respect to $EST_{i}$.

$E(minEST_{same})$ is the expected value for the difference in minimum
Estimated Start Time when the task is rescheduled on the same resource and
$PF_{i}$ (point of failure).

$E(minEST_{diff})$ is the expected value for the difference in minimum
Estimated Start Time when the task is rescheduled on a different non-failing
resource and $PF_{i}$ (point of failure).

Therefore, the probability of the task $t_{i}$ being resubmitted on a
different resource is $(1-P(new=v_{i})).$ Overhead due to submission on a
different resource:

$Overhead_{diff}=E(minEST_{diff})+TET_{Hi}$ (21)

Using (18), (19), (20), (21):

$\begin{split}RO_{i}=P_{t_{i}}^{R_{i}}\cdot[P(new=v_{i})\cdot
Overhead_{same}+(1-P(new=v_{i}))\cdot Overhead_{different}]\end{split}$ (22)

That is,

$\displaystyle\begin{split}RO_{i}=P_{t_{i}}^{R_{i}}\cdot\bigg{[}P(new=v_{i})\cdot\big{\\{}E(minEST_{same})+PF_{i}-\bigg{\lfloor}\dfrac{PF_{i}}{\lambda}\bigg{\rfloor}\cdot\lambda\big{\\}}&\\\
+(1-P(new=v_{i}))\cdot\big{\\{}E(minEST_{diff})+TET_{Hi}\big{\\}}\bigg{]}\end{split}$
(23)

Using (8), (9), (10), (11), (23)

$TET_{CRCH/CO}=\sum_{i\in CP}\bigg{(}TET_{Hi}+\mu_{w}(A(i))\\\
+P_{t_{i}}^{R_{i}}\cdot\bigg{[}P(new=v_{i})\cdot\big{\\{}E(minEST_{same})+PF_{i}-\bigg{\lfloor}\dfrac{PF_{i}}{\lambda}\bigg{\rfloor}\cdot\lambda\big{\\}}\\\
+(1-P(new=v_{i}))\cdot\big{\\{}E(minEST_{diff})+TET_{Hi}\big{\\}}\bigg{]}\bigg{)}$
(24)

From (10), (11)

$TET_{CRCH}=TET_{CRCH/CO}\cdot\bigg{[}1+\dfrac{\gamma}{\lambda}\bigg{]}$ (25)

It can be seen that $TET_{CRCH}$ is a product of two separate terms involving
$\lambda$. Let $Term1\leftarrow TET_{CRCH/CO}$ and
$Term2\leftarrow\bigg{[}1+\dfrac{\gamma}{\lambda}\bigg{]}.$ For $Term2$, it is
clear that as $\lambda$ increases the value of $Term2$ decreases. In $Term1$
the expressions involving $\lambda$ are
$PF_{i}-\bigg{\lfloor}\dfrac{PF_{i}}{\lambda}\bigg{\rfloor}\cdot\lambda$,
$P(new=v_{i})$, $E(minEST_{diff})$, $E(minEST_{same})$ where $PF_{i}$ is
independent of $\lambda$ (Assumptions 2 and 3). The first expression among
these is a non differentiable function in $\lambda$, although it is piece-wise
continuous. The average value of the function over a contiguous range of
values for $\lambda$ (where the function is continuous and $\lambda$ is chosen
randomly from a uniform distribution) would increase as $\lambda$ increases.

In stable environments, $P_{t_{i}}^{R_{i}}\ll 1.$ Therefore $Term1$ is
insignificant and $\lambda$ can be increased. This means a greater checkpoint
interval is affordable in a stable environment. On the contrary, in an
unstable environment $P_{t_{i}}^{R_{i}}$ is considerable and cannot be
ignored. $P(new=v_{i})$ decreases as $\lambda$ increases since the overhead
involved in re-scheduling on a different resource would be comparable to the
loss in execution time units incurred from distant checkpoints. In general,
with low resource availability there are many instances of resubmissions,
which leads to an increase in $E(minEST_{diff})$ and $E(minEST_{same})$. Thus
a lower value of $\lambda$ would be needed to reduce the overhead caused due
to checkpointing.

For cases where $\lambda\ll 1$, $WT_{i}$ would reduce as the resubmission
would not account for much loss in execution time units. In $unstable$
environments, the contributing factor to the waiting time of a task would be
the unavailability of resources to schedule ancestor tasks. Hence the
observations on waiting times for tasks indicated a negligible impact on
changes in $\lambda$. For $stable$ environments the number of resubmissions
are low. Hence the waiting time of a task is not significantly influenced by
changes in the parameter $\lambda$. (Assumption 1)

∎

##  4 Performance Analysis

###  4.1 Experimental Setup

To evaluate the performance of the proposed approach, the WorkflowSim
simulator is used. The cloud environment is modeled with one data center and
20 Condor VMs. It is assumed that there are at least four reliable VMs at any
point in time. Each VM is assumed to be fully connected to every other VM
using two-way dedicated lines. This assumption is based on the fact that if
they are connected using a single interconnect bus then wait time to access
and gain control over the bus also needs to be considered and becomes a
substantial factor in TET as compared to factors of significance such as
transmission time, checkpoint overhead etc.

Four different workflows: Montage, Cybershake, Inspiral and SIPHT are used in
the simulation, with sizes ranging from 100 to 700, in multiples of 100. The
workflows are given as input in the form of a DAX file. The dependencies
between the tasks in the workflow, runtime of the tasks on multiple VMs, the
rate of file transmission between VMs and the size of the files to be
transmitted between the tasks are read from the DAX file. Each DAX file is
executed ten times and the average of the results in those executions is
considered.

In some cases, the same DAX file shows high run times on a few executions and
low on the others. This is attributed to the fact that if a critical task gets
scheduled on a failing VM, the runtime increases by a huge margin. A task is a
critical task if that lies on the critical path and has zero free and total
float [20]. An increase in the runtime of this task would also increase the
final execution time linearly.

The runtimes of the tasks are different on each of the Condor VMs. Each task
has a set of defined parent and children tasks and the data to be transferred
between tasks is specific and predefined. Thus three different matrices are
used as input.

  1. 1.

$(Task\times Task)$: Data to be transferred.

  2. 2.

$(Task\times VM)$: Run time of a task on a given VM.

  3. 3.

$(VM\times VM)$: Transmission time between two VMs.

A light-weight checkpointing [25] model is proposed. Checkpoints of a task
only contain the program state and pointers to a data store (not the input
data itself). Each VM has a non-volatile stable storage. Each globally
synchronized checkpoint dumps the working memory of the VM to its non-volatile
storage. It involves the instruction pointer, register state, current stack
etc. There also exists a global storage which holds references to the outputs
generated by the parents of a task. The parents’ output is stored on the
resource, on which the parent was executed. Hence the local checkpoint need
only save the light-weight program state. The references pointers to the
locations where an output of a parent task is stored can be retrieved from the
global reliable storage. The data required for execution is fetched from these
references as and when required. Instead of fetching all the data from its
references before the execution of a task, wastage is reduced in the
transmission of unwanted data by fetching only those pages which are required
at any given point during execution. Since dedicated lines exist between the
VMs, the time to fetch data from disk belonging to another VM is comparable to
fetching data from its own disk.

The resource failures for a simulated environment consider Mean Time Between
Failures (MTBF), the size of the failure in terms of the number of VMs
affected, and the Mean Time to Repair (MTTR). Three different failure models
Unstable, Normal & Stable have been modeled by drawing values from the
following distributions. For each environment, the hyper-parameters of the
distribution vary. The MTBF is modeled using a Weibull distribution with shape
parameters ranging from 11.5 to 12.5 [7]. The size of the failure is also
modeled using a Weibull distribution with shape parameters ranging from 1.5 to
2.4 [7]. The set of failing VMs and MTTR are modeled using uniform and log-
normal distributions respectively [7]. The shape parameters used for MTTR are
10 and 5. MTBF values decrease as one moves from stable to unstable
environment as failures are more frequent. The MTTR values (in minutes) are
roughly 6/3/1 for unstable/normal/stable environments.

###  4.2 Experimental Results

The performance of CRCH is compared with HEFT and another heuristic approach
called ReplicateAll which was proposed in [11]. ReplicateAll uses a simple
approach for task replication wherein each task in the workflow is replicated
repCount number of times. repCount is assumed to be three for ReplicateAll,
and hence it is referred to as ReplicateAll(3) in this work. Both are run on
multiple workflows with the same failure models as that of CRCH. A majority of
the executions for HEFT could not complete due to the absence of fault
tolerance. Average Total Execution Time (TET), Resource Usage, Resource
wastage and the Standard Length Ratio (SLR) are the metrics used for
evaluating the performance of CRCH.

A resource is considered good if a majority of the tasks have a very low
execution time on it as compared to other VMs. If the VM chosen to fail is
good, lots of tasks are likely to be scheduled on it (especially in a sparse
workflow) and hence more resubmissions. Otherwise, the number of tasks
originally scheduled on that resource is itself low, hence lesser number of
resubmissions and a comparatively smaller increase in TET. The goodness of a
failing VM and consequentially the number of resubmissions are a decisive
factor in all the metrics mentioned and thus explain a majority of the
anomalies.

The results for Montage workflow have been presented but similar patterns were
observed for SIPHT, LIGO and CyberShake. TET of CRCH is greater than HEFT in
all the environments. This is expected as CRCH involves execution of
replicated tasks and at the time of resource failures, CRCH waits for the
failed resource to be back online or schedules the task on a less optimal but
available resource. This increase is more profound in the normal environment,
as the rate of resource failures is higher than that of the stable
environment. The TET of ReplicateAll(3) in all three environments (not shown)
is much higher than its counterparts, as all the tasks of the workflow have to
be executed four times. Albeit, the replicas can be scheduled in parallel, [8]
suggests a task schedule where replicas for a task are scheduled after its
children. Moreover, a large number of tasks on the same level (as seen in many
scientific workflows) [26] would generate more replicas than that can be
handled by existing VMs. These factors lead to a manifold increase in TET for
ReplicateAll(3). Figure 4 presents the comparison of HEFT and CRCH algorithms
under stable and normal environments. HEFT fails to execute in the unstable
environment because of a higher probability of resource failure and lack of
fault tolerance. TET in stable is, in general, lower as compared to the normal
environment as there exist very few cases of resubmission, whereas replication
amounts to the same time in both the environments. In general, as CRCH
considers resubmission on the same resource, TET values are considerably lower
than what can be expected, if the resubmissions were always on a different
resource.

Figure 4: Total Execution Time

Figures 5 and 6 analyze the impact of the parameters involved in the
Clustering Algorithm on the average TET of a workflow (at optimum values of
checkpointing parameters). Prior to hierarchical clustering Algorithm 1
transforms the task features to a lower dimensional space using PCA. The
stopping criterion for PCA is determined by a threshold on Coverage of
Variance (COV). We hypothesize that task features aren’t completely
independent and are prone to noise. Thus using all the features to estimate
task affinities would lead to suboptimal cluster assignments. This is
confirmed by an increase in average TET as we increase the threshold on COV.
Similarly using lower number of features would lead to information loss. We
observe that a threshold of $0.3-0.4$ leads to cluster assignments with more
accurate estimates of the replication counts. The other parameter concerning
clustering is the max. replication count or the number of final super
clusters. An increase in this hyper-parameter leads to an expected increase in
TET across all environments. It is interesting to note that the average TET
doesn’t increase dramatically when one moves away from the optimum max.
replication count as the new cluster that is formed has a relatively small
size. Having zero or very few replicas also leads to an increase in TET which
can be attributed to the resubmission overhead.

Figure 5: Clustering Overhead [Coverage of Variance] Figure 6: Clustering
Overhead [Max. Replication Count]

Figure 7(a) compares the performance of the Checkpoint/Restart algorithm
(Algorithm 3) against the SCR Algorithm when the former involves no replicas
for any task. In an unstable environment CRCH’s checkpointing marginally
outperforms SCR while the checkpointing overhead is comparable in the normal
and stable environments. In the case of SCR, due to multiple node failures in
an unstable environment the workflow needs to rely on the infrequent stable
checkpoints which save the intermediate states on a Parallel File System.
Restarting tasks from these points leads to excessive resource wastage and the
workflow takes longer to complete. On the other hand, CRCH can afford to have
more frequent light-weight checkpoints (smaller $\lambda$). In most workflows,
a child task requires the data computed by its parent to restart from a saved
program state. Since this data is always dumped to a stable storage upon the
completion of a task and can also be accessed via a pointer saved on a stable
global storage, the recovery is less expensive. For a constant $\lambda$
across all environments SCR outperforms CRCH under stable conditions,
primarily due to overlapping computations and data transfers. But since
$\lambda$ has been proposed to be a function of the failure parameters (MTTR,
MTBF) an increased $\lambda$ in a stable environment leads to a TET that is
comparable with SCR. Figure 7(b) shows that the average TET is quite sensitive
to $\lambda$ in a stable environment when no task has any replicas. A smaller
$\lambda$ leads to frequent checkpoints while a larger value leads to
increased re-computations.

(a) Comparison with SCR

(b) Impact of Checkpoint Interval

Figure 7: Checkpoint Overhead

Resource Usage is the sum of the actual processor time in minutes used for
executing all the tasks of the workflow [8]. Figure 8 shows the average
resource usage of the CRCH, HEFT and ReplicateAll(3) algorithms as a fraction
of TET. HEFT schedules the task of the workflow in the most optimum manner
[13] and thus has a minimum resource usage. HEFT cannot finish the execution
of workflows in the unstable environment, as it has no fault tolerance. CRCH,
on the other hand takes into account resource failures and uses replication
and resubmission of tasks, and thus the resource usage increases. In case of
the stable environment, the resource usage of CRCH is higher than HEFT by 16%.
This small increase is attributed only to the replication of tasks, as
resubmission is occasionally done due to the low resource failure rate.
However, in the case of a normal environment, the increase is at 17%, which is
much more prominent as along with replication, resubmission is also
considered. Resubmission of the tasks increases as the failure rate of the
resources increases. All the tasks are replicated three times in
ReplicateAll(3). That leads to higher resource usage. In the stable, normal
and unstable environments the average resource usage increase compared to CRCH
is 41%, 30% and 17% respectively. This is attributed to the fact that CRCH
uses Replication Heuristics in the form of hierarchical clustering. A decline
in the percentage increase in resource usage for ReplicateAll(3) over CRCH is
observed across the environments. This observation is due to the fact that
when a large number of resources are failing most of the replicated tasks do
not complete execution. Hence the futile usage of resources to re-execute
completed tasks reduces. Instead the resources are idle. The increase in usage
is highly dependent on the workflow types.

Figure 8: Average Resource Usage

Two major types of resource wastage are considered in CRCH:

  1. 1.

When a task is running on a resource and the resource fails, wastage occurs
when the task has executed beyond a certain checkpoint and needs to be
executed on a different resource. The processing power wasted in executing a
task beyond the last checkpoint is the effective wastage.

  2. 2.

When a task is replicated and the first replica executes successfully, any
subsequent replica executed on any of the resources are now deemed as waste.

In case of HEFT where there is no resubmission, there is a possibility of a
workflow failure - all tasks executed as part of the workflow are considered
as waste. The wastage in HEFT is calculated over ten executions, where all
failed workflow executions contribute to the waste and the wastage is averaged
over the executions. There is no wastage in an execution that is completed. No
replication heuristic is considered in ReplicateAll(3) for deciding the number
of replicas of tasks and so a lot of unnecessary replicas are created.
Resource wastage in ReplicateAll(3) is the time that is spent for executing
the unnecessary replicas of tasks, after their first replicas have already
executed successfully. The resource wastage that occurs due to resubmission is
eliminated, as there is no resubmission of tasks in ReplicateAll(3). The
average resource wastage of the HEFT, CRCH and ReplicateAll(3) as a fraction
of TET is compared in Figure 4. On average, CRCH gives a 22% and 46% reduction
in the resource waste in the normal and stable environments over HEFT
respectively. The improvement in resource wastage over HEFT in the normal
environment is lesser than the stable environment, because more number of
resubmissions are needed in the normal environment. The resource waste is
higher in ReplicateAll(3) than CRCH by 70%, 47% and 29% for the stable, normal
and unstable environments respectively. Overall CRCH performs better than its
counterparts.

Figure 9: Average Resource Wastage Figure 10: Average SLR

SLR is defined as the ratio between the execution time and the B-level of the
first entry task of the workflow [8]. In the case of CRCH and ReplicateAll(3),
SLR is calculated as the ratio between the execution time and the B-level of
the first task on the critical path after introducing the replicas. The
B-level of a node $n_{i}$ is the length of the longest path from node $n_{i}$
to an exit node bound by the length of the critical path [27]. An optimal
algorithm tries to choose the best scheduling plan and thus lowers the SLR
value. HEFT has no consideration for task or resource failures and hence has
low SLR value. CRCH assumes resource failures, and therefore the optimal
resource for a task may not be available at the ready time of the task. The
task might need to be scheduled on a second-best resource or wait for the
failed resource to come online, and thus the SLR values are higher than HEFT.
In the stable environment, as shown in Figure 10, the SLR values of CRCH are
comparable with that of HEFT and is only slightly higher by 5%. The slight
improvement in SLR for CRCH can be intuitively attributed to the reason that
SLR is dependent on the critical path identified. Since every task has some
number of replicas, a shorter critical path may be found from a replica of the
task instead of the original one. In the case of failure of an optimal
resource in CRCH, resubmissions on the same resource improve the SLR value. In
a normal environment more resources are expected to fail, hence the increase
in SLR value for CRCH. Both the execution time (numerator of SLR) and the
critical path found (denominator of SLR) increase and hence the increase of
10% is observed. In ReplicateAll(3), there is a slight increase in SLR, but
not as substantial as both the numerator and denominator of the SLR ratio
increase. SLR is also affected by resubmissions and there is no resubmission
in ReplicateAll(3).

Metrics like Average Resource Usage (Figure 11) and Average Resource Wastage
(Figure 12) show specific trends across workflow types which resonate with the
structural differences, different data and computational characteristics among
them. As CyberShake is much more CPU intensive than Montage [5], a 35%
increase in Resource Usage is seen under the normal environment with the CRCH
Algorithm. This increase is as high as 129% for LIGO (used in the physics
field with the aim of detecting gravitational waves) which is characterized by
having CPU intensive tasks that consume large memory [5]. Due to the fixed
number of replicas, the Resource Usage is higher for ReplicateAll(3), as
compared to CRCH under all environments and workflow types. It is interesting
to note that while comparing each workflow’s Resource Usage to that of Montage
with the ReplicateAll(3) algorithm, the percentage increase is lower compared
to CRCH. This is because the Resource Usage attributed to futile replication
overshadows the Resource Usage attributed to inherent workflow
characteristics. For example, there is a mere 47% increase when LIGO is
compared to Montage. An increase of 111%, 129%, and 144% respectively is
observed while comparing LIGO to Montage in the three different environments
under CRCH. This rise in percentage is once again caused by the reduction in
the failure rates of replicas as the environment stability improves. Identical
patterns are observed while comparing Montage with CyberShake, and CyberShake
with SIPHT.

Corollary to the reasons mentioned for Resource Usage, similar trends are
observed in the Average Resource Wastage. A 109% increase is seen while
comparing Montage with LIGO under the normal environment using the CRCH
Algorithm. This figure is close to the 106% increase seen in case of
ReplicateAll(3) as the number of unused replicas is comparable in most of the
random draws. Montage has a high I/O and low CPU utilization [5] and hence has
the least Resource Wastage irrespective of the algorithm or the environment.
The reverse is true for SIPHT [5] and the increase in wastage is nominal. For
a heavily CPU intensive workflow like LIGO [5] the percentage increase is as
high as 211% while comparing Montage with LIGO under the stable environment
using the CRCH algorithm. This percentage increase drops to 110% in the
unstable environment owing to the drop in MTBF (or increase in failure rate).
In other words, the unstable environment justifies the creation of more
replicas and checkpoints.

Figure 11: Resource Usage Across Workflow Types Figure 12: Resource Wastage
Across Workflow Types

##  5 Conclusion and Future Work

It has been shown that modeling replication heuristics using unsupervised
statistical techniques on readily available unlabelled data, improves
efficiency in faulty environments. In other words, CRCH can capture the
correlations between similar tasks and their corresponding replication counts.
On the other hand, it fails to incorporate the probability distributions over
resource failure parameters like MTTR and MTBF, in deciding the magnitude of
task replication. This would mean that corresponding tasks in identical
workflows would end up having a similar number of replications, irrespective
of the environment of execution.

The CRCH analysis goes to show that using checkpoints to store light-weight
pointers to saved states not only improves memory access time but also makes
the system robust. Distributed non-volatile storage complements the use of
checkpoints. The analysis reflects an improvement in resource wastage.
Resources on an average spend less time recomputing intermediate results.
Since the results are also stored at a non-volatile memory location, improving
checkpoint retrieval through increased network capacity is trivial.

It has been proven that varying checkpoint parameters with failure parameters,
can lead to better Average Total Execution Time. Albeit, the resource failure
distributions for different resources are assumed to be mutually independent
over time. This may not be the case when resource failures are modeled at a
resource group level. In such situations, the probability of failure of a
resource is dependent on the current state of its neighborhood resources and
the resource failure probability distributions are merely conditionally
independent, given the state of a few other resources [12].

The improvements in Total Execution Time, Resource Usage and Resource Wastage,
over other HEFT based algorithms across normal, stable and unstable
environments has been shown empirically. Further work can be done by
establishing theoretical bounds on the extent of improvements in the
aforementioned metrics.

The failure models considered assume the availability of a few reliable
resources, which are used for task resubmissions, ensuring that a task is
resubmitted at most once. Thus helping in estimating the expected resubmission
overhead. Although in practical circumstances, reliable resources may not
always be available.

This paper can be extended to deal with real-world workflow management systems
like Pegasus [26]. An elaborate set of training samples for replication counts
can further improve the machine learning aspect. The DAX files for tasks with
multiple direct and derived features can also improve the heuristic accuracy.

In estimating the increase in Total Execution Time, the model takes into
account only the failures of tasks that lie on the critical path as they have
a direct impact on the increase in TET. The DAX inputs with multiple/dynamic
critical paths or cloud environments with highly varying MTBF/MTTR
distributions for individual VMs requires further analysis.

## References

## References

  * [1] Taverna, Why use workflows?, https://taverna.incubator.apache.org/introduction/why-use-workflows, online; Accessed 2017-06-22 (2014).
  * [2] S. Callagan, Overview of scientific workflows, https://pegasus.isi.edu/2017/03/15/pegasus-featured-in-blue-waters-webinar-series/, online; Accessed 2017-06-22 (2017).
  * [3] C. Hoffa, G. Mehta, T. Freeman, E. Deelman, K. Keahey, B. Berriman, J. Good, On the use of cloud computing for scientific workflows, in: eScience, 2008. eScience’08. IEEE Fourth International Conference on, IEEE, 2008, pp. 640–645.
  * [4] E. Deelman, Grids and clouds: Making workflow applications work in heterogeneous distributed environments, in: International Journal of High Performance Computing Applications, Vol. 24, SAGE Publications, 2010, pp. 284–298.
  * [5] G. M. Juve, Resource management for scientific workflows, Ph.D. thesis, Citeseer (2012).
  * [6] S. Hwang, C. Kesselman, Grid workflow: A flexible failure handling framework for the grid, in: High Performance Distributed Computing, 2003. Proceedings. 12th IEEE International Symposium on, IEEE, 2003, pp. 126–137.
  * [7] K. Plankensteiner, R. Prodan, T. Fahringer, A new fault tolerance heuristic for scientific workflows in highly distributed environments based on resubmission impact, in: e-Science, 2009. e-Science’09. Fifth IEEE International Conference on, IEEE, 2009, pp. 313–320.
  * [8] Y. Zhang, A. Mandal, C. Koelbel, K. Cooper, Combined fault tolerance and scheduling techniques for workflow applications on computational grids, in: Cluster Computing and the Grid, 2009. CCGRID’09. 9th IEEE/ACM International Symposium on, IEEE, 2009, pp. 244–251.
  * [9] D. Poola, M. A. Salehi, K. Ramamohanarao, R. Buyya, A taxonomy and survey of fault-tolerant workflow management systems in cloud and distributed computing environments, in: I. Mistrik, R. Bahsoon, N. Ali, M. Heisel, B. Maxim (Eds.), Software Architecture for Big Data and the Cloud, Morgan Kaufmann, Morgan Kaufmann, 2017, Ch. 15, pp. 285–315.
  * [10] J. Yu, R. Buyya, A taxonomy of workflow management systems for grid computing, in: Journal of Grid Computing, Vol. 3, Springer, 2005, pp. 171–200.
  * [11] K. Plankensteiner, R. Prodan, T. Fahringer, A. Kertesz, P. K. Kacsuk, Fault-tolerant behavior in state-of-the-art grid workflow management systems, in: 2009 Fifth IEEE International Conference on e-Science, CoreGRID, 2007, pp. 313–320.
  * [12] D. Poola, S. K. Garg, R. Buyya, Y. Yang, K. Ramamohanarao, Robust scheduling of scientific workflows with deadline and budget constraints in clouds, in: Advanced Information Networking and Applications (AINA), 2014 IEEE 28th International Conference on, IEEE, 2014, pp. 858–865.
  * [13] H. Topcuoglu, S. Hariri, M.-y. Wu, Performance-effective and low-complexity task scheduling for heterogeneous computing, in: IEEE transactions on parallel and distributed systems, Vol. 13, IEEE, 2002, pp. 260–274.
  * [14] B. Kruatrachue, T. Lewis, Grain size determination for parallel processing, in: IEEE software, Vol. 5, IEEE, 1988, pp. 23–32.
  * [15] G. Kandaswamy, A. Mandal, D. A. Reed, Fault tolerance and recovery of scientific workflows on computational grids, in: Cluster Computing and the Grid, 2008. CCGRID’08. 8th IEEE International Symposium on, IEEE, 2008, pp. 777–782.
  * [16] W. Chen, E. Deelman, Fault tolerant clustering in scientific workflows, in: Services (SERVICES), 2012 IEEE Eighth World Congress on, IEEE, 2012, pp. 9–16.
  * [17] A. Moody, G. Bronevetsky, K. Mohror, B. R. d. Supinski, Design, modeling, and evaluation of a scalable multi-level checkpointing system, in: Proceedings of the 2010 ACM/IEEE International Conference for High Performance Computing, Networking, Storage and Analysis, SC ’10, IEEE Computer Society, Washington, DC, USA, 2010, pp. 1–11.  doi:10.1109/SC.2010.18.
URL https://doi.org/10.1109/SC.2010.18

  * [18] L. Bautista-Gomez, S. Tsuboi, D. Komatitsch, F. Cappello, N. Maruyama, S. Matsuoka, Fti: High performance fault tolerance interface for hybrid systems, in: SC ’11: Proceedings of 2011 International Conference for High Performance Computing, Networking, Storage and Analysis, 2011, pp. 1–12.
  * [19] M. Wieczorek, R. Prodan, T. Fahringer, Scheduling of scientific workflows in the askalon grid environment, in: ACM SIGMOD Record, Vol. 34, ACM, 2005, pp. 56–62.
  * [20] G. James, D. Witten, T. Hastie, R. Tibshirani, An Introduction to Statistical Learning, Springer, 2013.
  * [21] I. Jolliffe, Principal Component Analysis Second Edition, springer, 2002.
  * [22] J. Yang, D. Parikh, D. Batra, Joint unsupervised learning of deep representations and image clusters, in: Proceedings of the IEEE Conference on Computer Vision and Pattern Recognition, 2016, pp. 5147–5156.
  * [23] Y.-K. Kwok, I. Ahmad, Benchmarking and comparison of the task graph scheduling algorithms, in: Journal of Parallel and Distributed Computing, Vol. 59, Elsevier, 1999, pp. 381–422.
  * [24] M. Natrella, Nist/sematech e-handbook of statistical methods, http://www.itl.nist.gov/div898/handbook/ (2010).
  * [25] R. Duan, R. Prodan, T. Fahringer, Dee: A distributed fault tolerant workflow enactment engine for grid computing, in: International Conference on High Performance Computing and Communications, Springer, 2005, pp. 704–716.
  * [26] E. Deelman, K. Vahi, G. Juve, M. Rynge, S. Callaghan, P. J. Maechling, R. Mayani, W. Chen, R. F. da Silva, M. Livny, et al., Pegasus, a workflow management system for science automation, in: Future Generation Computer Systems, Vol. 46, Elsevier, 2015, pp. 17–35.
  * [27] I. A. Yu-Kwong Kwok, Static scheduling algorithms for allocating directed task graphs to multiprocessors, in: ACM Computing Surveys (CSUR), ACM, 1999, pp. 406–471.