# A Selective Mirrored Task Based Fault Tolerance Mechanism for Big Data Application Using Cloud

Authors: Hao Wu; Qinggeng Jin; Chenghua Zhang; He Guo
Journal: Wireless Communications and Mobile Computing
 (2019)

---

## Abstract

With the wide deployment of cloud computing in big data processing and the growing scale of big data application, managing reliability of resources becomes a critical issue. Unfortunately, due to the highly intricate directed-acyclic-graph (DAG) based application and the flexible usage of processors (virtual machines) in cloud platform, the existing fault tolerant approaches are inefficient to strike a balance between the parallelism and the topology of the DAG-based application while using the processors, which causes a longer makespan for an application and consumes more processor time (computation cost). To address these issues, this paper presents a novel fault tolerant framework named Fault Tolerance Algorithm using Selective Mirrored Tasks Method (FAUSIT) for the fault tolerance of running a big data application on cloud. First, we provide comprehensive theoretical analyses on how to improve the performance of fault tolerance for running a single task on a processor. Second, considering the balance between the parallelism and the topology of an application, we present a selective mirrored task method. Finally, by employing the selective mirrored task method, the FAUSIT is designed to improve the fault tolerance for DAG based application and incorporates two important objects: minimizing the makespan and the computation cost. Our solution approach is evaluated through rigorous performance evaluation study using real-word workflows, and the results show that the proposed FAUSIT approach outperforms existing algorithms in terms of makespan and computation cost.


---

## Body

## 1. Introduction

Recent years have witnessed that the big data analysis grows dramatically, and the related applications have been used everywhere in both academia [1] and industry [2]. There is no denying that the developmental cloud platform technologies played a key role in this process; the plenty of processers in cloud make sure that the scholars can handle the significant large-scale big data processing [3–6]. However, due to voltage fluctuation, cosmic rays, thermal changes, or variability in manufacturing, the chip level soft errors and the physical flaws are inevitable for a processer even when the probability of that is extremely low [7, 8]. Furthermore, the abundant use of processers by a big data application induces that the probability cannot be ignored.Big data and big data analysis have been proposed for describing data sets as analytical technologies in large-scale complex programs, which need to be analyzed with advanced analytical methods [9, 10]. No matter whether the big data applications are developed for commercial purposes or scientific researches, most of these applications require significant amount of computing resources, such as market structure analysis, customer trade analysis, environmental research, and astrophysics data processing. Motivated by the reasonable price, rapid elasticity, and shifting responsibility of maintenance, backups, and management to cloud providers, more and more big data applications have been deployed to clouds, such as EC2 [11], Google Cloud [12], and Microsoft Azure [13].The clouds provide unlimited computing resources (from the user’s point of view) including CPU resources and GPU resources. The on-demand recourses facilitate users to choose apposite processors (with CPU or GPU resources) for executing their big data applications efficiently [14]. However, according to [15], there are many factors such as voltage fluctuation, cosmic rays, thermal changes, or variability in manufacturing, which cause the processors (both CPU and GPU) to be more vulnerable. Indeed, the probability of fault rate is really low. But, as already noted, plenty of processors participate in computing the big data application, and the computing time of each processor may be very long. These potential factors will lead to an exponential growth for the fault rate in a cycle of running a big data application.The failures caused by processors are disastrous for a big data application which is deployed on the processors; i.e., once a fault occurs on any processor, the application will have to be executed all over again, and that will waste lots of monetary cost and time cost. Thus, improving the robustness (or reducing the fault rate) for running a big data application has attracted many scholars’ attention; many studies have been exploring this problem. According to the literature, these studies are classified into two main categories: resolving the problem in hardware or software level.From the hardware level’s perspective, improving the mean time between failures (MTBF) [16] of the processors is the key to reduce the fault rate for running a big data application. As everyone knows, it is impossible to eradicate the failures in a processor. Apart from lifting tape-out technology, the [15] proposed an adaptive low-overhead fault tolerance mechanism for many-core processor, which treats fault tolerance as a device that can be configured and used by application when high reliability is needed. Although [15] improves the MTBF, but the risk of occurring failures remains. Therefore, the other scholars seek solutions in software level.From the software level’s perspective, the developed check-point technology [17] makes sure that a big data application can be completed under any size of MTBF. Thus, many check-point strategies are proposed to resolve this problem such as [18, 19]. These strategies only pay a little extra cost, but they can complete the applications under any size of MTBF. As a result, almost all of cloud platforms provide check-point interface for users. However, these strategies did not consider the date dependencies between the processors, which make it inappropriate to big data application running on cloud.In order to handle the DAG based applications, the copy task based method (also known as primary backup based method) is proposed to resolve the problem. In [20], Qin and Jiang proposed an algorithm eFRD to enable a systems fault tolerance and maximize its reliability. On the basis of eFRD, Zhu et al. [21] developed an approach for task allocation and message transmission to ensure faults can be tolerated during the workflow execution. But the task based methods will make the makespan have a long delay due to the fact that the backup tasks are not starting with the original tasks.To the best of our knowledge, for the big data application, there are no proper check-point strategies which can handle the data dependencies among the processors simultaneously. In this paper, we propose a novel check-point strategy for big data applications, which considers the effect of the data communications. The proposed strategy adopts high level failure model to resolve the problem, which makes it closer to practice. Meanwhile, the subsequent effect caused by data dependencies after a failure is also considered in our strategy.The main contributions of this paper are as follows:(i) A selective mirrored task method is proposed for the fault tolerance of the key subtasks in a DAG based application.(ii) On the basis of the selective mirrored task method, a novel check-point framework is proposed to resolve the fault tolerance for a big data application running on cloud; the framework is named Fault Tolerance Algorithm using Selective Mirrored Tasks Method (FAUSIT).(iii) A thorough performance analysis is conducted for FAUSIT through experiments on randomly generated test big data application as well as real-world application traces.The rest of this paper is structured as follows. The related work is summarized in Section2. Section 3 introduces the models of the big data application, the cloud platform, and the MTBF and then formally defines the problem the paper is addressing. Section 4 presents the novel check-point strategy for the big data application running on cloud. Section 5 conducts extensive experiments to evaluate the performance of our algorithm. Section 6 concludes the paper with summary and future directions.


## 2. Related Work

Over the last two decades, owing to the increasing scale of the big data applications [22], the fault tolerance for big data application is becoming more and more crucial. Considerable research has been explored by scholars. In this section, we summarize the research in terms of theoretic methods and heuristic methods.A number of theoretic methods have been explored by scholars. For a task running on a processor, Young [21] proposed a first order failure model and figured out an approximation to the optimum check-point interval which is φopt=2δM, where δ is the time to write a check-point file, M is the mean time between failures for the processor, and φopt is the optimum computation interval between writing check-point files. However, the model in Young [21] will never have more than a single failure in any given computation interval. This assumption goes against some practical situations; for instance, there may be more than one failure occurring in a computation interval.Due to the downside of the model in Yang [21], Daly [22] proposed a higher order failure model to estimate the optimum check-point interval. The model of Daly [22] assumes that there may be more than one failure occurring in a computation interval, which is closer to the realistic situation. The optimum computation interval figured out by Daly [22] is(1)φopt=2δM1+13δ2M1/2+19δ2M-δifδ<2M,Mifδ≥2M.(2)φopt=2δM-δifδ<12M,Mifδ≥12M.which is a good rule of thumb for most practical systems.However, the models in both Yang [22] and Daly [22] are aimed at one task running on a processor, which are not applicable to DAG based application running on cloud for the following reasons. First, there are many subtasks running on different processors; the completion time of each subtask may have influence on the successive subtasks. Second, the importance of each subtask in a DAG (a DAG based application) is different; for instance, the subtasks on the critical path of the DAG are more important than the others. Therefore, some scholars proposed heuristic methods aiming at DAG based application running on cloud.Aiming to resolve the fault tolerance for DAG based application running on cloud, Zhu [23] and Qin [24] proposed copy task based methods. In general, the basic idea of copy task based methods is running an identical copy of each subtasks on different processors, the subtasks and their copies can be mutually excluded in time. However, these approaches assume that tasks are independent of one other, which cannot meet the needs of real-time systems where tasks have precedence constraints. In [24], for given two tasks, the authors defined the necessary conditions for their backup copies to safely overlap in time with each other and proposed a new overlapping scheme named eFRD (efficient fault-tolerant reliability-driven algorithm), which can tolerate processors failures in a heterogeneous system with fully connected network.In Zhu [23], on the basis of Qin [24], the authors established a real-time application fault-tolerant model that extends the traditional copy task based model by incorporating the cloud characteristics. Based on this model, the authors developed approaches for subtask allocation and message transmission to ensure faults can be tolerated during the application execution and proposed a dynamic fault tolerant scheduling algorithm, named FASTER (fault tolerant scheduling algorithm for real-time scientific workflow). The experiment results show that the FASTER is better than eFRD [24].Unfortunately, the disadvantage of copy task based methods including Zhu [23] and Qin [24] is very conspicuous. First, the copy of each subtask may consume more resources on the cloud, which makes them uneconomical. Second, the copy of each subtask will be executed only when the original subtask failed; this process will waste a lot of time, and it will be even worse due to the DAG based application; i.e., the deadline of the application will be not guaranteed in most of cases if the deadline is near to the critical path.Thus, in this paper, we will combine the advantage of theoretic methods and heuristic methods to propose a novel fault tolerance algorithm for a big data application running on cloud.


## 3. Models and Formulation

In this section, we introduce the models of big data application, cloud platform, and the failure model then formally define the problem this paper is addressing. To improve the readability, we sum up the main notations used throughout this paper in Table1.Table 1
Cost optimization factors.
Symbols Definitions  T=GV,E a scientific application GV,E DAG of the tasks inT CP the critical path ofGV,E τi thei-th subtask in V wi the weight ofi-th tasks in V V the set ofτi in T ei,j data dependence fromτi to τj E the set ofei,j δ the time to make a check-point file φopt the optimal check-point interval TW(φ) the practical completion time for the task R the time to read a check-point file X the time before a failure in an interval Tc the computation time for a task S a schedule which maps the tasks to processors wt(τi) the weight ofτi pred(τi) the predecessor subtasks set ofτi in GV,E succ(τi) the successors subtasks set ofτi in GV,E succP(τi) the next subtaskτi on the same processor ft(τi) the finish time ofτi on the S estS(τi) the earliest start time ofτi on the S lftS(τi) the latest finish time ofτi on the S slackT(τi) the slack time ofτi on the S slackT¯(τi) the indirect slack time ofτi on the S α(τi) denotes the quantitative importance ofτi ᾱ The threshold ofα KeyT The set of key subtasks Ms(T) The makespan of the applicationT


### 3.1. Big Data Application Model

The model of a big data applicationT is denoted by GV,E, where GV,E represents a DAG. Besides, we use Tc to denote the execution time of the critical path in GV,E. Each node ni∈V represents a subtask τi(1≤i≤v) of T, and v is the total number of subtasks in T. W is the set of weights, in which each wire presents the execution time of a subtask ni on a VM. E is the set of edges in GV,E, and an edge (ni,nj) represents the dependence between ni and nj; i.e., a task can only start after all its predecessors have been completed.Figure1 shows an example of DAG for a big data workflow, consisting of twelve tasks from τ1to τ12. The DAG vertices related to tasks in T are represented by circles, while the directed edges denote the data dependencies among the tasks.Figure 1
A DAG based application.


### 3.2. Cloud Platform Model

A cloud platformC is modeled as a set of processors {P1,P2,…,PN} and N is the total number of processors on the cloud. We use N to denote the number of processors rented by users for executing an application. Actually, the N is much greater than the mount which the user need. In general, to reduce the cost (monetary cost or time cost), the users apply proper schedule arithmetic to deploy their big data applications on cloud. But most of schedule algorithms did not consider the failure in each processor, which may consume extra cost (monetary cost and time cost).


### 3.3. The Check-Point Model

The check-point technology has been used on cloud for years, which makes application complete successfully in the shortest amount of time. In general, the check-point model is defined as bellow: ideally, the time to complete a task on a processor is denoted byTc; we use φ to denote the check-point interval. After each computation interval (φ), the processor makes a backup for the current status of the system, and the time consumed by this process is represented by δ. If a failure occurs, the time consumed to read the latest backup and restart the computation is denoted by R. Finally, we use TW(φ) to denote the practical completion time for the task running on a processor while the check-point interval is φ.Referring to Figure2, the ideal completion time for the task is Tc=5φ. Actually, there is a failure occurring after X time in the third interval, and it takes the processor R time to restart the third interval. At last, the practical time consumed by the tasks is TW(φ)=5φ+R+X+4δ.Figure 2
An example of a failure.


### 3.4. The Failure Model

For a given MTBF (mean time between failures) which is denoted byM, according to [21], the life distribution model for mechanical and electrical equipment is described by an exponential model. Thus, the probability density function is(3)ft=1Me-t/M.Then, the probability of a failures occurring before time Δt for a processor is represented by a cumulative distribution function(4)Pt≤Δt=∫Δt01Me-t/Mdt=1-e-Δt/M.Obviously, the probability of successfully completing for a time Δt without a failure is(5)Pt>Δt=1-Pt≤Δt=e-Δt/M.We use Tc to denote the computation time for a subtask running on a processor; the φ denotes the compute interval between two check-points. Then, the average number of attempts (represented by No.a) needed to complete Tc is(6)No.a=Tc/φPt>Δt=TceΔt/Mφ.Therefore, the total number of failures during Δt is the number of attempts minus the number of successes.(7)nΔt=TceΔt/Mφ-Tcφ=TcφeΔt/M-1.Notice that this assumes that we will never have more than a single failure in any given computation interval. Obviously, this assumption is relaxed in a real-life scenario. Thus, in [22], the scholar presented a multiple failures model.(8)nφ=Tc-δ+δTc/φM-Eφ+δ+RPφ-ER+φ+δ1-Pφ,where(9)Eφ+δ=M+φ+δ1-eφ+δ/MER+φ+δ=M+R+φ+δ1-eφ+δ/MPφ=e-R+φ+δ/MThe derivation process of Formula (8) is detailed in [22]; we will not repeat the process in this paper. In this paper, we will take Formula (8) as the failure model in our framework for two reasons; first, this could only provide the MTBF (M, mean time between failures) determined by statistics [19, 25] and, second, this model is closer to reality than the other model in [21].


### 3.5. Definitions

In order to make readers have a better understanding of this algorithm, we make some definitions first. For a given big data applicationT=GV,E and a schedule S, we define the following terms.


#### 3.5.1. Schedule (S)

A schedule S is a map from the subtasks inGV,E to the processors on the cloud; meanwhile, the start time and the finish time of each subtask have been figured out. In general, the S is determined by a static schedule algorithm, such as HEFT [26] and MSMD [27].


#### 3.5.2. Weight (wt(τi))

Thewt(τi) is the weight (execution time) of τi running on a processor.


#### 3.5.3. Predecessors (pred(τi))

For a subtaskτi, its predecessors subtask set is defined as below:(10)predτi=tauj∣τj∈V∧τi,τj∈E.


#### 3.5.4. Successors (succ(τi))

For a subtaskτi, its successors subtask set is defined as below:(11)predτi=tauj∣τj∈V∧τj,τi∈E.


#### 3.5.5. Successor on the Processor (SuccP(τi))

For a given scheduleS, the SuccP(τi) of τi is the set of the next subtask deployed on the same processor.


#### 3.5.6. Finish Time (ft(τi))

Theft(τi) denotes the finish time of τi on the schedule S.


#### 3.5.7. Earliest Start Time on the Schedule (estS(τi))

For a given scheduleS, the start time of τi is the estS(τi). It should be noted that the estS(τi) is different with the traditional earliest start time in DAG.


#### 3.5.8. Latest Finish Time on the Schedule (lftS(τi))

For a given scheduleS, the lftS(τi) of τi is defined below:(12)lftSτi=ftτi,ifsuccτiandSuccPτiare empty,minτj∈succτi∨SuccPτi⁡estSτi,otherwise.Same with estS(τi), the lftS(τi) is different with the traditional earliest start time in DAG.


#### 3.5.9. Slack Time (slackT(τi))

For a given schedule S, theslackT(τi) of τi is defined as follows:(13)slackTτi=lftSτi-estSτi-weightτi.


#### 3.5.10. Indirect Slack Time (slackT¯(τi))

For a given schedule S, theslackT¯(τi) of τi is defined as follows:(14)slackT¯τi=slackTτi,ifslackTτi≠0,min⁡wtτiwfτi+wtτjslackT¯τj,else ifsuccτi∨SuccPτi=∅,0,otherwise.The slackT¯(τi) denotes that the slack time of a subtask can be shared by its predecessors.


#### 3.5.11. Importance (α(τi))

Theα(τi) denotes the quantitative importance of the subtask τi in the schedule S. The α(τi) is calculated by(15)ατi=slackT¯τiwtτi.


#### 3.5.12. Threshold of Importance (ᾱ)

The thresholdᾱ denotes whether a subtask is a key subtask; i.e., if α(τi)<ᾱ, then the subtask τi is a key subtask.


#### 3.5.13. The Set of Key Subtask (KeyT)

The signalKeyT denotes the set of key subtask for a given schedule S.


### 3.6. Problem Formalization

The ultimate objective of this work is to provide a high-performance fault tolerance mechanism and make sure that the proposed fault tolerance mechanism will consume less computation cost and makespan. The computation cost represents the processor time consumed by all the subtasks inT; thus, the object of minimizing computation cost is defined by(16)Minimize∑τi∈TTWτi.The makespan can be defined as the overall time to execute the whole workflow by considering the finish time of the last successfully completed task. For an application T, this object is denoted by(17)MinimizeMsT.


## 3.1. Big Data Application Model

The model of a big data applicationT is denoted by GV,E, where GV,E represents a DAG. Besides, we use Tc to denote the execution time of the critical path in GV,E. Each node ni∈V represents a subtask τi(1≤i≤v) of T, and v is the total number of subtasks in T. W is the set of weights, in which each wire presents the execution time of a subtask ni on a VM. E is the set of edges in GV,E, and an edge (ni,nj) represents the dependence between ni and nj; i.e., a task can only start after all its predecessors have been completed.Figure1 shows an example of DAG for a big data workflow, consisting of twelve tasks from τ1to τ12. The DAG vertices related to tasks in T are represented by circles, while the directed edges denote the data dependencies among the tasks.Figure 1
A DAG based application.


## 3.2. Cloud Platform Model

A cloud platformC is modeled as a set of processors {P1,P2,…,PN} and N is the total number of processors on the cloud. We use N to denote the number of processors rented by users for executing an application. Actually, the N is much greater than the mount which the user need. In general, to reduce the cost (monetary cost or time cost), the users apply proper schedule arithmetic to deploy their big data applications on cloud. But most of schedule algorithms did not consider the failure in each processor, which may consume extra cost (monetary cost and time cost).


## 3.3. The Check-Point Model

The check-point technology has been used on cloud for years, which makes application complete successfully in the shortest amount of time. In general, the check-point model is defined as bellow: ideally, the time to complete a task on a processor is denoted byTc; we use φ to denote the check-point interval. After each computation interval (φ), the processor makes a backup for the current status of the system, and the time consumed by this process is represented by δ. If a failure occurs, the time consumed to read the latest backup and restart the computation is denoted by R. Finally, we use TW(φ) to denote the practical completion time for the task running on a processor while the check-point interval is φ.Referring to Figure2, the ideal completion time for the task is Tc=5φ. Actually, there is a failure occurring after X time in the third interval, and it takes the processor R time to restart the third interval. At last, the practical time consumed by the tasks is TW(φ)=5φ+R+X+4δ.Figure 2
An example of a failure.


## 3.4. The Failure Model

For a given MTBF (mean time between failures) which is denoted byM, according to [21], the life distribution model for mechanical and electrical equipment is described by an exponential model. Thus, the probability density function is(3)ft=1Me-t/M.Then, the probability of a failures occurring before time Δt for a processor is represented by a cumulative distribution function(4)Pt≤Δt=∫Δt01Me-t/Mdt=1-e-Δt/M.Obviously, the probability of successfully completing for a time Δt without a failure is(5)Pt>Δt=1-Pt≤Δt=e-Δt/M.We use Tc to denote the computation time for a subtask running on a processor; the φ denotes the compute interval between two check-points. Then, the average number of attempts (represented by No.a) needed to complete Tc is(6)No.a=Tc/φPt>Δt=TceΔt/Mφ.Therefore, the total number of failures during Δt is the number of attempts minus the number of successes.(7)nΔt=TceΔt/Mφ-Tcφ=TcφeΔt/M-1.Notice that this assumes that we will never have more than a single failure in any given computation interval. Obviously, this assumption is relaxed in a real-life scenario. Thus, in [22], the scholar presented a multiple failures model.(8)nφ=Tc-δ+δTc/φM-Eφ+δ+RPφ-ER+φ+δ1-Pφ,where(9)Eφ+δ=M+φ+δ1-eφ+δ/MER+φ+δ=M+R+φ+δ1-eφ+δ/MPφ=e-R+φ+δ/MThe derivation process of Formula (8) is detailed in [22]; we will not repeat the process in this paper. In this paper, we will take Formula (8) as the failure model in our framework for two reasons; first, this could only provide the MTBF (M, mean time between failures) determined by statistics [19, 25] and, second, this model is closer to reality than the other model in [21].


## 3.5. Definitions

In order to make readers have a better understanding of this algorithm, we make some definitions first. For a given big data applicationT=GV,E and a schedule S, we define the following terms.


### 3.5.1. Schedule (S)

A schedule S is a map from the subtasks inGV,E to the processors on the cloud; meanwhile, the start time and the finish time of each subtask have been figured out. In general, the S is determined by a static schedule algorithm, such as HEFT [26] and MSMD [27].


### 3.5.2. Weight (wt(τi))

Thewt(τi) is the weight (execution time) of τi running on a processor.


### 3.5.3. Predecessors (pred(τi))

For a subtaskτi, its predecessors subtask set is defined as below:(10)predτi=tauj∣τj∈V∧τi,τj∈E.


### 3.5.4. Successors (succ(τi))

For a subtaskτi, its successors subtask set is defined as below:(11)predτi=tauj∣τj∈V∧τj,τi∈E.


### 3.5.5. Successor on the Processor (SuccP(τi))

For a given scheduleS, the SuccP(τi) of τi is the set of the next subtask deployed on the same processor.


### 3.5.6. Finish Time (ft(τi))

Theft(τi) denotes the finish time of τi on the schedule S.


### 3.5.7. Earliest Start Time on the Schedule (estS(τi))

For a given scheduleS, the start time of τi is the estS(τi). It should be noted that the estS(τi) is different with the traditional earliest start time in DAG.


### 3.5.8. Latest Finish Time on the Schedule (lftS(τi))

For a given scheduleS, the lftS(τi) of τi is defined below:(12)lftSτi=ftτi,ifsuccτiandSuccPτiare empty,minτj∈succτi∨SuccPτi⁡estSτi,otherwise.Same with estS(τi), the lftS(τi) is different with the traditional earliest start time in DAG.


### 3.5.9. Slack Time (slackT(τi))

For a given schedule S, theslackT(τi) of τi is defined as follows:(13)slackTτi=lftSτi-estSτi-weightτi.


### 3.5.10. Indirect Slack Time (slackT¯(τi))

For a given schedule S, theslackT¯(τi) of τi is defined as follows:(14)slackT¯τi=slackTτi,ifslackTτi≠0,min⁡wtτiwfτi+wtτjslackT¯τj,else ifsuccτi∨SuccPτi=∅,0,otherwise.The slackT¯(τi) denotes that the slack time of a subtask can be shared by its predecessors.


### 3.5.11. Importance (α(τi))

Theα(τi) denotes the quantitative importance of the subtask τi in the schedule S. The α(τi) is calculated by(15)ατi=slackT¯τiwtτi.


### 3.5.12. Threshold of Importance (ᾱ)

The thresholdᾱ denotes whether a subtask is a key subtask; i.e., if α(τi)<ᾱ, then the subtask τi is a key subtask.


### 3.5.13. The Set of Key Subtask (KeyT)

The signalKeyT denotes the set of key subtask for a given schedule S.


## 3.5.1. Schedule (S)

A schedule S is a map from the subtasks inGV,E to the processors on the cloud; meanwhile, the start time and the finish time of each subtask have been figured out. In general, the S is determined by a static schedule algorithm, such as HEFT [26] and MSMD [27].


## 3.5.2. Weight (wt(τi))

Thewt(τi) is the weight (execution time) of τi running on a processor.


## 3.5.3. Predecessors (pred(τi))

For a subtaskτi, its predecessors subtask set is defined as below:(10)predτi=tauj∣τj∈V∧τi,τj∈E.


## 3.5.4. Successors (succ(τi))

For a subtaskτi, its successors subtask set is defined as below:(11)predτi=tauj∣τj∈V∧τj,τi∈E.


## 3.5.5. Successor on the Processor (SuccP(τi))

For a given scheduleS, the SuccP(τi) of τi is the set of the next subtask deployed on the same processor.


## 3.5.6. Finish Time (ft(τi))

Theft(τi) denotes the finish time of τi on the schedule S.


## 3.5.7. Earliest Start Time on the Schedule (estS(τi))

For a given scheduleS, the start time of τi is the estS(τi). It should be noted that the estS(τi) is different with the traditional earliest start time in DAG.


## 3.5.8. Latest Finish Time on the Schedule (lftS(τi))

For a given scheduleS, the lftS(τi) of τi is defined below:(12)lftSτi=ftτi,ifsuccτiandSuccPτiare empty,minτj∈succτi∨SuccPτi⁡estSτi,otherwise.Same with estS(τi), the lftS(τi) is different with the traditional earliest start time in DAG.


## 3.5.9. Slack Time (slackT(τi))

For a given schedule S, theslackT(τi) of τi is defined as follows:(13)slackTτi=lftSτi-estSτi-weightτi.


## 3.5.10. Indirect Slack Time (slackT¯(τi))

For a given schedule S, theslackT¯(τi) of τi is defined as follows:(14)slackT¯τi=slackTτi,ifslackTτi≠0,min⁡wtτiwfτi+wtτjslackT¯τj,else ifsuccτi∨SuccPτi=∅,0,otherwise.The slackT¯(τi) denotes that the slack time of a subtask can be shared by its predecessors.


## 3.5.11. Importance (α(τi))

Theα(τi) denotes the quantitative importance of the subtask τi in the schedule S. The α(τi) is calculated by(15)ατi=slackT¯τiwtτi.


## 3.5.12. Threshold of Importance (ᾱ)

The thresholdᾱ denotes whether a subtask is a key subtask; i.e., if α(τi)<ᾱ, then the subtask τi is a key subtask.


## 3.5.13. The Set of Key Subtask (KeyT)

The signalKeyT denotes the set of key subtask for a given schedule S.


## 3.6. Problem Formalization

The ultimate objective of this work is to provide a high-performance fault tolerance mechanism and make sure that the proposed fault tolerance mechanism will consume less computation cost and makespan. The computation cost represents the processor time consumed by all the subtasks inT; thus, the object of minimizing computation cost is defined by(16)Minimize∑τi∈TTWτi.The makespan can be defined as the overall time to execute the whole workflow by considering the finish time of the last successfully completed task. For an application T, this object is denoted by(17)MinimizeMsT.


## 4. The Fault Tolerance Algorithm

In this section, we first discuss the basic idea of our algorithm for the fault tolerance of running big data application on cloud. Then, on the basis of the idea, we will propose the fault tolerance algorithm using Selective Mirrored Tasks Method.


### 4.1. The Basic Idea

As show in Section2, the theoretic methods which are devoted to find the optimal φopt are not applicable to the DAG based application, even if the φopt they have determined is very accurate for one task running on a processor. Besides, the heuristic methods based on the copy task will waste a lot of extra resource, and the completion time of the application may be delayed by much more time.To find a better solution, we will integrate the advantages of theoretic methods and heuristic methods to propose a high-performance and economical algorithm for big data application. The check-point method with an optimal computation intervalφopt is a dominant and economical method for one task running on a processor; thus, the check-point mechanism is the major means in our approach. Furthermore, owing to the parallelism and the dependencies in a DAG based application, the importance of each subtask is different; i.e., the subtasks on the critical path are more important than the others. The fault tolerance performance of these subtasks which adopt check-point method is insufficient, because the completion time of an application depends to a great extent on the completion time of these subtasks. Therefore, for the important subtasks, we will improve the fault tolerance performance of an application by introducing the task copy based method. In the task copy based methods [24], the original task and the copy do not start at the same time; to reduce the completion time, the original task and the copy will start at the same time.In summary, the basic idea is as follows. First, identify the important subtasks in the DAG based application, which are named as key subtasks in the rest of this article. Then, apply the task copy based methods to the key subtasks; meanwhile, all the subtasks will employ the check-point technology to improve the fault tolerance performance, but the key subtasks and the normal subtasks will use different optimal computation intervalφopt, the details of which will be described in the following sections.It should be noted that our fault tolerance algorithm will not schedule the subtasks on the processors; we just provide a fault tolerance mechanism based on the existing static scheduler algorithm (such as HEFT and MSMD) to make sure that the application can be completed with the minimum of time.


### 4.2. Fault Tolerance Algorithm Using Selective Mirrored Tasks Method (FAUSIT)

Based on the basic idea above, we propose the FAUSIT to improve the fault tolerance for executing a large-scale big data application; the pseudocode of the FAUSIT is listed in Algorithm1.Algorithm 1:Fault Tolerance Algorithm using Selective Mirrored Tasks Method (FAUSIT).
Input: a schedule S of an application T Output: a fault tolerance operation F ( 1) DetermineKeyTasks(S,T) →KeyT ( 2) DeployKeyTasks(KeyT) →F ( 3) returnFAs shown in Algorithm1, the input of FAUSIT is a map from subtasks to processors which is determined by a static scheduler algorithm and the output is fault tolerance operation determined by FAUSIT. The functionDetermineKeyTasks() in Line (1) is to find the key subtasks according to the schedule S and the application T. Then, the functionDeployKeyTasks() in Line (2) deploys the mirrored subtasks and determine the proper φopt for the subtasks.In the following content in this subsection, we will expound the two functions in FAUSIT.


#### 4.2.1. Function ofDetermineKeyTasks()

The function ofDetermineKeyTasks() is to determine the key subtasks in the DAG based application. In order to make readers have a better understanding of this function, we need to expound the key subtask and the indirect slack time clearly.Definition 1.
Key subtask: in a scheduleS, the finish time of a subtask has influence on the start time of its successors; if the influence exceeds a threshold, we define the subtask is a key subtask.
The existence of the key subtasks is very meaningful to our FAUSIT algorithm. For a given scheduleS, in the ideal case, each subtask as well as the application will be finished in accordance with the S. In practice, a subtask may fail when it has executed for a certain time; then, the processor will load the latest check-point files for continuation. At this point, the delay produced by the failure subtask may affect the successors. For the subtasks which have sufficient slack time, the start time of the successors is free from the failed subtask. On the contrary, if the failed subtask has little slack time, it will affect the start time of the successors undoubtedly. Given all that, we need to deal with the key subtasks which has little slack time.Definition 2.
Indirect slack time: for two subtasksτi and τj, τj is the successor of τi, if τj has slack time (defined in Section 3.5.9), the slack time can be shared by τi, and the shared slack time is indirect slack time for τi.
The indirect slack time is a useful parameter in our FAUSIT algorithm, the existence of which will make the FAUSIT save a lot of time (makespan of the application) and cost. For a given scheduleS, a subtask may have sufficient slack time which can be shared by predecessors. Thus, the predecessors may have enough slack time to deal with failures; then, the completion time of the predecessors and the subtask will not delay the makespan of the application. Indeed, the indirect slack time is the key parameter to determine whether a subtask is a key subtask. Moreover, the indirect slack time reduces the count of the key subtasks in a big data application, which will save a lot of cost, because the key subtask will apply mirrored task method.The pseudocode for functionDetermineKeyTasks() is shown in Algorithm 2. The input ofDetermineKeyTasks() is a schedule S of an application T and the threshold ᾱ; the output is the set of key subtasks. First, in Line (1), the ft(), estS() and the lftS() of the subtasks are determined according to Sections 3.5.6 and 3.5.7 and Formula (12). Then, the slackT() of each subtask is figured out by Formula (13) in Line (2). Line (3) determines the slackT¯() of the subtasks by recursion. Finally, the set of key subtasks is determined according to the threshold α.Algorithm 2:DetermineKeyTasks().
Input: a schedule S of an application T, ᾱ. Output: the key subtask set KeyT. ( 1) Determine the ft(), estS() and the lftS() of the sub- tasks inT according to S. ( 2) Determine the slackT() of each subtask. ( 3) Determine the slackT¯() of the subtasks by recursion. ( 4) Determine the set of key subtasks according to α →KeyT ( 5) returnKeyT.Table2 shows the process ofDetermineKeyTasks(). When the ᾱ=0.15, Figure 3 shows the key subtasks which shall adopt the mirrored task method to improve the performance of fault tolerance.Table 2
An example ofDetermineKeyTasks().
Subtasks Weight l f t ( ) e s t ( ) S t ( ) S t ′ ( ) α ( )  τ 1 50 0 50 0 18.4 0.37 τ 2 75 0 75 0 0 0 τ 3 45 50 130 35 35 0.78 τ 4 55 75 130 0 0 0 τ 5 36 130 166 0 0 0 τ 6 60 166 226 0 6.8 0.11 τ 7 45 166 211 0 21.2 0.47 τ 8 65 226 291 0 14.1 0.22 τ 9 40 211 291 40 40 1 τ 10 200 166 366 0 0 0 τ 11 50 291 366 25 25 0.5 τ 12 75 366 441 0 0 0Figure 3
An example ofᾱ=0.15.It should be noted that the thresholdᾱ is given by the users, which is related to the makespan of the application; i.e., the higher ᾱ leads to more key subtasks. Then, the makespan of the application is shorter. On the contrary, the smaller ᾱ will lead to a longer makespan.


#### 4.2.2. Function ofDeployKeyTasks()

The function ofDeployKeyTasks() is to deal with the key subtasks, which minimizes the makespan of the application to the least extent. The main operation ofDeployKeyTasks() is using mirrored task method; in order to make readers have a better understanding of this function, we need to expound the mirrored task method first.The mirrored task method is to deploy a copy of a key subtask on another processor; the original subtask is denoted byτiP and the copy of the subtask is denoted by τiC. The τiP and the τiC start at the same time, and the check-point interval of them is 2φopt (the φopt is determined by [22]). The distinction between the τiP and the τiC is that the first check-point interval of τiP is φopt; meanwhile, the first check-point interval of τiC is 2φopt. Obviously, once a failure occurs in one of the processors, the interlaced check-point interval of the two same subtasks makes sure that the time delayed by dealing with the failure is W (the time to read a check-point file).Figure4 shows an example of the mirrored task method. The Figure 4(a) displays the ideal situation of the mirrored task method; i.e., there are no failures happen in both Pk and Pl, and the finish time is 4φ+2δ. In Figure 4(b), there is only one failure happening on Pl in the second check-point interval φ2. First, the processor Pl reads the latest check-point file named δk1. Then, with time goes by, the processor Pl will immediately load the latest check-point file δk3 when it generates. Thus, the finish time is 4ϕ + 2δ + W. Figure 4(c) illustrates the worst case; both Pk and Pl have a failure in the same interval φ2, the two processors will have to load the latest check-point file δk1. Thus, the finish time is 4φ+2δ+W+X.Figure 4
An example of mirrored tasks method.
(a)
The ideal situation (b)
One failure occurs (c)
Two failures occur simultaneouslyObviously, the mirrored task method is far better than the traditional copy task based method, since the copy task will start only when the original task failed, and it will waste a lot of time. Moreover, the mirrored task method is also better than the method in [22], since the probability of the worst case is far less than the probability of the occurrence for one failure in a processor.The pseudocode for functionDeployKeyTasks() is displayed in Algorithm 3. The loop in Line (1) makes sure that all the key subtasks can be deploy on the processors. The applied processors should be used first when deploying the key subtasks; the loop in Line (2) illustrates this constraint.Algorithm 3: DeployKeyTasks().
Input: the key subtask set yT. Output:a fault tolerance operation F ( 1) forτiB in KeyTdo ( 2) forPj in the processors which have been applied do ( 3) ifτiB has no overlapping with other tasks on Pjthen ( 4) Deploy the τiB on Pj ( 5) End for in Line (2). ( 6) end if ( 7) end for ( 8) ifτiB has not be deployed on the processor. then ( 9) Apply a new processor. ( 10) Deploy τiB on the new processor. ( 11) Put the new processor in the set of processors. ( 12) end if ( 13) end for ( 14) Deploy the check-point interval to the processors. ( 15) Save the operations in F. ( 16) returnFLine(3) makes sure that the key subtasks τiB have no overlapping with other tasks on Pj. Lines (4)-(5) deploy τiB on Pj. If all the applied processors have no idle interval for τiB (Line (8)), we will have to apply a new processor and deploy τiB on it; then, we put the new processor into the set of applied processors (Lines (9)-(11)). At last, we deploy the check-point interval (φopt or 2φopt) to the processors (Line (14)) and save these operations in T (Line (15)).It should be noted that the overlapping in Line(3) is not just the computation time of the τiB; we also consider the delayed time which may be caused by failures. In order to avoid the overlap caused by the delayed τiB, we use the 1.3wi as the execution time of τiB to determine whether an overlap happen, since the 1.3wi is much greater than a delayed τiB.


### 4.3. The Feasibility Study of FAUSIT

The operations to the processors in our FAUSIT are complex, such as the different check-point interval and the mirrored subtasks; the readers may doubt the feasibility of FAUSIT. Thus, we will explain the feasibility study of FAUSIT in this subsection.According to [15], Google Cloud provides thegcloud suit for users to operate the processors (virtual machine) remotely. The operations ofgcloud include (but not limited to) applying processors, releasing processors, check-point, and loading check-point files. These operations ingcloud can make user implement the FAUSIT easily.


## 4.1. The Basic Idea

As show in Section2, the theoretic methods which are devoted to find the optimal φopt are not applicable to the DAG based application, even if the φopt they have determined is very accurate for one task running on a processor. Besides, the heuristic methods based on the copy task will waste a lot of extra resource, and the completion time of the application may be delayed by much more time.To find a better solution, we will integrate the advantages of theoretic methods and heuristic methods to propose a high-performance and economical algorithm for big data application. The check-point method with an optimal computation intervalφopt is a dominant and economical method for one task running on a processor; thus, the check-point mechanism is the major means in our approach. Furthermore, owing to the parallelism and the dependencies in a DAG based application, the importance of each subtask is different; i.e., the subtasks on the critical path are more important than the others. The fault tolerance performance of these subtasks which adopt check-point method is insufficient, because the completion time of an application depends to a great extent on the completion time of these subtasks. Therefore, for the important subtasks, we will improve the fault tolerance performance of an application by introducing the task copy based method. In the task copy based methods [24], the original task and the copy do not start at the same time; to reduce the completion time, the original task and the copy will start at the same time.In summary, the basic idea is as follows. First, identify the important subtasks in the DAG based application, which are named as key subtasks in the rest of this article. Then, apply the task copy based methods to the key subtasks; meanwhile, all the subtasks will employ the check-point technology to improve the fault tolerance performance, but the key subtasks and the normal subtasks will use different optimal computation intervalφopt, the details of which will be described in the following sections.It should be noted that our fault tolerance algorithm will not schedule the subtasks on the processors; we just provide a fault tolerance mechanism based on the existing static scheduler algorithm (such as HEFT and MSMD) to make sure that the application can be completed with the minimum of time.


## 4.2. Fault Tolerance Algorithm Using Selective Mirrored Tasks Method (FAUSIT)

Based on the basic idea above, we propose the FAUSIT to improve the fault tolerance for executing a large-scale big data application; the pseudocode of the FAUSIT is listed in Algorithm1.Algorithm 1:Fault Tolerance Algorithm using Selective Mirrored Tasks Method (FAUSIT).
Input: a schedule S of an application T Output: a fault tolerance operation F ( 1) DetermineKeyTasks(S,T) →KeyT ( 2) DeployKeyTasks(KeyT) →F ( 3) returnFAs shown in Algorithm1, the input of FAUSIT is a map from subtasks to processors which is determined by a static scheduler algorithm and the output is fault tolerance operation determined by FAUSIT. The functionDetermineKeyTasks() in Line (1) is to find the key subtasks according to the schedule S and the application T. Then, the functionDeployKeyTasks() in Line (2) deploys the mirrored subtasks and determine the proper φopt for the subtasks.In the following content in this subsection, we will expound the two functions in FAUSIT.


### 4.2.1. Function ofDetermineKeyTasks()

The function ofDetermineKeyTasks() is to determine the key subtasks in the DAG based application. In order to make readers have a better understanding of this function, we need to expound the key subtask and the indirect slack time clearly.Definition 1.
Key subtask: in a scheduleS, the finish time of a subtask has influence on the start time of its successors; if the influence exceeds a threshold, we define the subtask is a key subtask.
The existence of the key subtasks is very meaningful to our FAUSIT algorithm. For a given scheduleS, in the ideal case, each subtask as well as the application will be finished in accordance with the S. In practice, a subtask may fail when it has executed for a certain time; then, the processor will load the latest check-point files for continuation. At this point, the delay produced by the failure subtask may affect the successors. For the subtasks which have sufficient slack time, the start time of the successors is free from the failed subtask. On the contrary, if the failed subtask has little slack time, it will affect the start time of the successors undoubtedly. Given all that, we need to deal with the key subtasks which has little slack time.Definition 2.
Indirect slack time: for two subtasksτi and τj, τj is the successor of τi, if τj has slack time (defined in Section 3.5.9), the slack time can be shared by τi, and the shared slack time is indirect slack time for τi.
The indirect slack time is a useful parameter in our FAUSIT algorithm, the existence of which will make the FAUSIT save a lot of time (makespan of the application) and cost. For a given scheduleS, a subtask may have sufficient slack time which can be shared by predecessors. Thus, the predecessors may have enough slack time to deal with failures; then, the completion time of the predecessors and the subtask will not delay the makespan of the application. Indeed, the indirect slack time is the key parameter to determine whether a subtask is a key subtask. Moreover, the indirect slack time reduces the count of the key subtasks in a big data application, which will save a lot of cost, because the key subtask will apply mirrored task method.The pseudocode for functionDetermineKeyTasks() is shown in Algorithm 2. The input ofDetermineKeyTasks() is a schedule S of an application T and the threshold ᾱ; the output is the set of key subtasks. First, in Line (1), the ft(), estS() and the lftS() of the subtasks are determined according to Sections 3.5.6 and 3.5.7 and Formula (12). Then, the slackT() of each subtask is figured out by Formula (13) in Line (2). Line (3) determines the slackT¯() of the subtasks by recursion. Finally, the set of key subtasks is determined according to the threshold α.Algorithm 2:DetermineKeyTasks().
Input: a schedule S of an application T, ᾱ. Output: the key subtask set KeyT. ( 1) Determine the ft(), estS() and the lftS() of the sub- tasks inT according to S. ( 2) Determine the slackT() of each subtask. ( 3) Determine the slackT¯() of the subtasks by recursion. ( 4) Determine the set of key subtasks according to α →KeyT ( 5) returnKeyT.Table2 shows the process ofDetermineKeyTasks(). When the ᾱ=0.15, Figure 3 shows the key subtasks which shall adopt the mirrored task method to improve the performance of fault tolerance.Table 2
An example ofDetermineKeyTasks().
Subtasks Weight l f t ( ) e s t ( ) S t ( ) S t ′ ( ) α ( )  τ 1 50 0 50 0 18.4 0.37 τ 2 75 0 75 0 0 0 τ 3 45 50 130 35 35 0.78 τ 4 55 75 130 0 0 0 τ 5 36 130 166 0 0 0 τ 6 60 166 226 0 6.8 0.11 τ 7 45 166 211 0 21.2 0.47 τ 8 65 226 291 0 14.1 0.22 τ 9 40 211 291 40 40 1 τ 10 200 166 366 0 0 0 τ 11 50 291 366 25 25 0.5 τ 12 75 366 441 0 0 0Figure 3
An example ofᾱ=0.15.It should be noted that the thresholdᾱ is given by the users, which is related to the makespan of the application; i.e., the higher ᾱ leads to more key subtasks. Then, the makespan of the application is shorter. On the contrary, the smaller ᾱ will lead to a longer makespan.


### 4.2.2. Function ofDeployKeyTasks()

The function ofDeployKeyTasks() is to deal with the key subtasks, which minimizes the makespan of the application to the least extent. The main operation ofDeployKeyTasks() is using mirrored task method; in order to make readers have a better understanding of this function, we need to expound the mirrored task method first.The mirrored task method is to deploy a copy of a key subtask on another processor; the original subtask is denoted byτiP and the copy of the subtask is denoted by τiC. The τiP and the τiC start at the same time, and the check-point interval of them is 2φopt (the φopt is determined by [22]). The distinction between the τiP and the τiC is that the first check-point interval of τiP is φopt; meanwhile, the first check-point interval of τiC is 2φopt. Obviously, once a failure occurs in one of the processors, the interlaced check-point interval of the two same subtasks makes sure that the time delayed by dealing with the failure is W (the time to read a check-point file).Figure4 shows an example of the mirrored task method. The Figure 4(a) displays the ideal situation of the mirrored task method; i.e., there are no failures happen in both Pk and Pl, and the finish time is 4φ+2δ. In Figure 4(b), there is only one failure happening on Pl in the second check-point interval φ2. First, the processor Pl reads the latest check-point file named δk1. Then, with time goes by, the processor Pl will immediately load the latest check-point file δk3 when it generates. Thus, the finish time is 4ϕ + 2δ + W. Figure 4(c) illustrates the worst case; both Pk and Pl have a failure in the same interval φ2, the two processors will have to load the latest check-point file δk1. Thus, the finish time is 4φ+2δ+W+X.Figure 4
An example of mirrored tasks method.
(a)
The ideal situation (b)
One failure occurs (c)
Two failures occur simultaneouslyObviously, the mirrored task method is far better than the traditional copy task based method, since the copy task will start only when the original task failed, and it will waste a lot of time. Moreover, the mirrored task method is also better than the method in [22], since the probability of the worst case is far less than the probability of the occurrence for one failure in a processor.The pseudocode for functionDeployKeyTasks() is displayed in Algorithm 3. The loop in Line (1) makes sure that all the key subtasks can be deploy on the processors. The applied processors should be used first when deploying the key subtasks; the loop in Line (2) illustrates this constraint.Algorithm 3: DeployKeyTasks().
Input: the key subtask set yT. Output:a fault tolerance operation F ( 1) forτiB in KeyTdo ( 2) forPj in the processors which have been applied do ( 3) ifτiB has no overlapping with other tasks on Pjthen ( 4) Deploy the τiB on Pj ( 5) End for in Line (2). ( 6) end if ( 7) end for ( 8) ifτiB has not be deployed on the processor. then ( 9) Apply a new processor. ( 10) Deploy τiB on the new processor. ( 11) Put the new processor in the set of processors. ( 12) end if ( 13) end for ( 14) Deploy the check-point interval to the processors. ( 15) Save the operations in F. ( 16) returnFLine(3) makes sure that the key subtasks τiB have no overlapping with other tasks on Pj. Lines (4)-(5) deploy τiB on Pj. If all the applied processors have no idle interval for τiB (Line (8)), we will have to apply a new processor and deploy τiB on it; then, we put the new processor into the set of applied processors (Lines (9)-(11)). At last, we deploy the check-point interval (φopt or 2φopt) to the processors (Line (14)) and save these operations in T (Line (15)).It should be noted that the overlapping in Line(3) is not just the computation time of the τiB; we also consider the delayed time which may be caused by failures. In order to avoid the overlap caused by the delayed τiB, we use the 1.3wi as the execution time of τiB to determine whether an overlap happen, since the 1.3wi is much greater than a delayed τiB.


## 4.2.1. Function ofDetermineKeyTasks()

The function ofDetermineKeyTasks() is to determine the key subtasks in the DAG based application. In order to make readers have a better understanding of this function, we need to expound the key subtask and the indirect slack time clearly.Definition 1.
Key subtask: in a scheduleS, the finish time of a subtask has influence on the start time of its successors; if the influence exceeds a threshold, we define the subtask is a key subtask.
The existence of the key subtasks is very meaningful to our FAUSIT algorithm. For a given scheduleS, in the ideal case, each subtask as well as the application will be finished in accordance with the S. In practice, a subtask may fail when it has executed for a certain time; then, the processor will load the latest check-point files for continuation. At this point, the delay produced by the failure subtask may affect the successors. For the subtasks which have sufficient slack time, the start time of the successors is free from the failed subtask. On the contrary, if the failed subtask has little slack time, it will affect the start time of the successors undoubtedly. Given all that, we need to deal with the key subtasks which has little slack time.Definition 2.
Indirect slack time: for two subtasksτi and τj, τj is the successor of τi, if τj has slack time (defined in Section 3.5.9), the slack time can be shared by τi, and the shared slack time is indirect slack time for τi.
The indirect slack time is a useful parameter in our FAUSIT algorithm, the existence of which will make the FAUSIT save a lot of time (makespan of the application) and cost. For a given scheduleS, a subtask may have sufficient slack time which can be shared by predecessors. Thus, the predecessors may have enough slack time to deal with failures; then, the completion time of the predecessors and the subtask will not delay the makespan of the application. Indeed, the indirect slack time is the key parameter to determine whether a subtask is a key subtask. Moreover, the indirect slack time reduces the count of the key subtasks in a big data application, which will save a lot of cost, because the key subtask will apply mirrored task method.The pseudocode for functionDetermineKeyTasks() is shown in Algorithm 2. The input ofDetermineKeyTasks() is a schedule S of an application T and the threshold ᾱ; the output is the set of key subtasks. First, in Line (1), the ft(), estS() and the lftS() of the subtasks are determined according to Sections 3.5.6 and 3.5.7 and Formula (12). Then, the slackT() of each subtask is figured out by Formula (13) in Line (2). Line (3) determines the slackT¯() of the subtasks by recursion. Finally, the set of key subtasks is determined according to the threshold α.Algorithm 2:DetermineKeyTasks().
Input: a schedule S of an application T, ᾱ. Output: the key subtask set KeyT. ( 1) Determine the ft(), estS() and the lftS() of the sub- tasks inT according to S. ( 2) Determine the slackT() of each subtask. ( 3) Determine the slackT¯() of the subtasks by recursion. ( 4) Determine the set of key subtasks according to α →KeyT ( 5) returnKeyT.Table2 shows the process ofDetermineKeyTasks(). When the ᾱ=0.15, Figure 3 shows the key subtasks which shall adopt the mirrored task method to improve the performance of fault tolerance.Table 2
An example ofDetermineKeyTasks().
Subtasks Weight l f t ( ) e s t ( ) S t ( ) S t ′ ( ) α ( )  τ 1 50 0 50 0 18.4 0.37 τ 2 75 0 75 0 0 0 τ 3 45 50 130 35 35 0.78 τ 4 55 75 130 0 0 0 τ 5 36 130 166 0 0 0 τ 6 60 166 226 0 6.8 0.11 τ 7 45 166 211 0 21.2 0.47 τ 8 65 226 291 0 14.1 0.22 τ 9 40 211 291 40 40 1 τ 10 200 166 366 0 0 0 τ 11 50 291 366 25 25 0.5 τ 12 75 366 441 0 0 0Figure 3
An example ofᾱ=0.15.It should be noted that the thresholdᾱ is given by the users, which is related to the makespan of the application; i.e., the higher ᾱ leads to more key subtasks. Then, the makespan of the application is shorter. On the contrary, the smaller ᾱ will lead to a longer makespan.


## 4.2.2. Function ofDeployKeyTasks()

The function ofDeployKeyTasks() is to deal with the key subtasks, which minimizes the makespan of the application to the least extent. The main operation ofDeployKeyTasks() is using mirrored task method; in order to make readers have a better understanding of this function, we need to expound the mirrored task method first.The mirrored task method is to deploy a copy of a key subtask on another processor; the original subtask is denoted byτiP and the copy of the subtask is denoted by τiC. The τiP and the τiC start at the same time, and the check-point interval of them is 2φopt (the φopt is determined by [22]). The distinction between the τiP and the τiC is that the first check-point interval of τiP is φopt; meanwhile, the first check-point interval of τiC is 2φopt. Obviously, once a failure occurs in one of the processors, the interlaced check-point interval of the two same subtasks makes sure that the time delayed by dealing with the failure is W (the time to read a check-point file).Figure4 shows an example of the mirrored task method. The Figure 4(a) displays the ideal situation of the mirrored task method; i.e., there are no failures happen in both Pk and Pl, and the finish time is 4φ+2δ. In Figure 4(b), there is only one failure happening on Pl in the second check-point interval φ2. First, the processor Pl reads the latest check-point file named δk1. Then, with time goes by, the processor Pl will immediately load the latest check-point file δk3 when it generates. Thus, the finish time is 4ϕ + 2δ + W. Figure 4(c) illustrates the worst case; both Pk and Pl have a failure in the same interval φ2, the two processors will have to load the latest check-point file δk1. Thus, the finish time is 4φ+2δ+W+X.Figure 4
An example of mirrored tasks method.
(a)
The ideal situation (b)
One failure occurs (c)
Two failures occur simultaneouslyObviously, the mirrored task method is far better than the traditional copy task based method, since the copy task will start only when the original task failed, and it will waste a lot of time. Moreover, the mirrored task method is also better than the method in [22], since the probability of the worst case is far less than the probability of the occurrence for one failure in a processor.The pseudocode for functionDeployKeyTasks() is displayed in Algorithm 3. The loop in Line (1) makes sure that all the key subtasks can be deploy on the processors. The applied processors should be used first when deploying the key subtasks; the loop in Line (2) illustrates this constraint.Algorithm 3: DeployKeyTasks().
Input: the key subtask set yT. Output:a fault tolerance operation F ( 1) forτiB in KeyTdo ( 2) forPj in the processors which have been applied do ( 3) ifτiB has no overlapping with other tasks on Pjthen ( 4) Deploy the τiB on Pj ( 5) End for in Line (2). ( 6) end if ( 7) end for ( 8) ifτiB has not be deployed on the processor. then ( 9) Apply a new processor. ( 10) Deploy τiB on the new processor. ( 11) Put the new processor in the set of processors. ( 12) end if ( 13) end for ( 14) Deploy the check-point interval to the processors. ( 15) Save the operations in F. ( 16) returnFLine(3) makes sure that the key subtasks τiB have no overlapping with other tasks on Pj. Lines (4)-(5) deploy τiB on Pj. If all the applied processors have no idle interval for τiB (Line (8)), we will have to apply a new processor and deploy τiB on it; then, we put the new processor into the set of applied processors (Lines (9)-(11)). At last, we deploy the check-point interval (φopt or 2φopt) to the processors (Line (14)) and save these operations in T (Line (15)).It should be noted that the overlapping in Line(3) is not just the computation time of the τiB; we also consider the delayed time which may be caused by failures. In order to avoid the overlap caused by the delayed τiB, we use the 1.3wi as the execution time of τiB to determine whether an overlap happen, since the 1.3wi is much greater than a delayed τiB.


## 4.3. The Feasibility Study of FAUSIT

The operations to the processors in our FAUSIT are complex, such as the different check-point interval and the mirrored subtasks; the readers may doubt the feasibility of FAUSIT. Thus, we will explain the feasibility study of FAUSIT in this subsection.According to [15], Google Cloud provides thegcloud suit for users to operate the processors (virtual machine) remotely. The operations ofgcloud include (but not limited to) applying processors, releasing processors, check-point, and loading check-point files. These operations ingcloud can make user implement the FAUSIT easily.


## 5. Empirical Evaluations

The purpose of the experiments is to evaluate the performance of the developed FAUSIT algorithm. We evaluate the fault tolerant of FAUSIT by comparing it with two other algorithms published in the literature. They are FASTER algorithm [23] and the method in [22]. The main differences of these algorithms to FAUSIT and their uses are briefly described below.(i)
FASTER: a novel copy task based fault tolerant scheduling algorithm. On the basis of copy task based method, the FASTER adjust the resources dynamically.(ii)
The method in [22]: it is a theoretic method which provides an optimal check-point interval φopt.


### 5.1. Experiment Settings

The DAG based big data applications we used for the evaluation are obtained from the DAG based applications benchmark provided by Pegasus WorkflowGenerator [28]. We use the largest application from the benchmark, i.e., Epigenomics, since the bigger application is more sensitive to failures. The detailed characteristics of the benchmark applications can be found in [29, 30]. In our experiments, the number of subtasks in an application is ranging from 100 to 1000. Since the benchmark does not assign deadlines for each application, we need to specify the deadlines; we assign a deadline for the applications: it is 1.0CP. Table 3 gives the characteristics of these applications including the count of tasks and edges, average task execution time and the deadlines.Table 3
The Characteristics of the Big Data Application.
Workflows Task Count Range Edge Count Avg. Run Time of Task (Sec) Deadlines Range Epigenomics 100 322 3856.51 1.0CP 200 644 … … 1000 3228Because FAUSIT does not schedule the subtasks on the processors, we hire a high-performance schedule algorithm (MSMD) to determine the map from subtasks to the processors. MSMD is a novel static schedule algorithm to reduce the cost and the makespan for an application. On the basis of MSMD, we use FAUSIT to improve the fault tolerant of an application.


### 5.2. Evaluation Criteria

The main objectives of the FAUSIT are to find the optimal fault tolerant mechanism for a big data application running on cloud. The first criterion to evaluate the performance of a fault tolerant mechanism is how much time is delayed to complete the application, i.e., the makespan to finish the application. We introduce the concept ofmakespan delay rate to indicate how much extra time consumed by the fault tolerant mechanism. It is defined as follows:(18)Mak.Del.Rate=The practical makespanThe ideal makespan.where the ideal makespan represents the completion time for running an application without fault tolerant mechanism and any failures, and the practical makespan is the completion time consumed on practical system which has a fault tolerant mechanism and the probability of failures.The second goal of the FARSIT is to minimize the extra cost consumed by the fault tolerant mechanism. Undoubtedly, to improve the performance of fault tolerant, any fault tolerant mechanisms will have to consume extra computation time for an application. Thus, the extra computation time is a key criterion to evaluate the performance of fault tolerant mechanism. Therefore, we define extra cost rate for the evaluation:(19)Ex.Cost.Rate=The practical processors timeThe ideal processors time.where the practical processors time denotes the time consumed by the processors to running an application for a practical system with failures, The ideal processors time is the sum of the wi in G(V,E) for an application.


### 5.3. The Optimalα for FAUSIT

Before comparing with the other algorithms, we need to determine the optimalᾱ for FAUSIT first. Due to the optimal ᾱ is an experimental parameter, we have to figure it out by experiments.We test theᾱ by the Epigenomics application with 1000 subtasks for 10 times and make the Td=1.0Tc; the results are shown in Figures 5(a) and 5(b). In Figure 5(a), theMak.Del.Rate becomes lower alone with the increased; i.e., a larger ᾱ will lead to a shorter makespan for an application. On the contrary, Figure 5(b) shows that the larger ᾱ will lead to more cost for an application.Figure 5
(a) TheMak.Del.Rate and (b) theEx.Cost.Rate of ᾱ.Through a comprehensive analysis of Figures5(a) and 5(b), we make the ᾱ=0.033 which can equalize theMak.Del.Rate and theEx.Cost.Rate and make sure that both theMak.Del.Rate and theEx.Cost.Rate are smaller than other situations.


### 5.4. Comparison with the Other Solutions

Since the proposed FAUSIT algorithm is a heuristic algorithm, the most straightforward way to evaluate its performance is to compare with the optimal solution when possible. We randomly generate 10 different applications for each scale of Epigenomics shown in Table3, and we make the Td=1.0CP.What should be paid attention to is that although the FASTER is a copy tasks based method, but it is developed for multiple DAG based big data applications. As a result, it cannot compare the performance of it with the other two methods directly. In order to make a comprehensive comparison, on the basis of FASTER, we modified the FASTER to make it handle a single DAG based application, and this new FASTER is denoted by FASTER∗.The average ofMak.Del.Rate andEx.Cost.Rate are displayed in Figure 6. In Figure 6(a), our FAUSIT has the minimumMak.Del.Rate for any size of application. Meanwhile, the FASTER∗ has higherMak.Del.Rate than our FAUSIT, and the Daly’s method has the highestMak.Del.Rate. The result in Figure 6(a) shows that our FAUSIT has the best performance for minimizing the makespan of an application among these methods.Figure 6
The result ofMak.Del.Rate andEx.Cost.Rate.
(a)
Mak.Del.Rate (b)
Ex.Cost.RateIn Figure6(b), due to the fact that the Daly’s method only adopts the check-point mechanism, this makes it has the minimumEx.Cost.Rate. Meanwhile, the FASTER∗ consumes the maximum cost, since theEx.Cost.Rate of it is very large. Interestingly, along with the increase of the count of subtasks in an application, theEx.Cost.Rate of our FAUSIT is becoming smaller, which indicates that our FAUSIT has the ability to handle much bigger scale of application.In conclusion, compared with the FASTER, our FAUSIT outperforms FASTER on bothMak.Del.Rate andEx.Cost.Rate. Compared with the Daly’s method, our FAUSIT outperforms it much more onMak.Del.Rate and only consume 9% extra cost, which still makes our FAUSIT have competitiveness in practical system. Besides, the ᾱ in our FAUSIT can satisfy the requirements from different users; i.e., if the users need a shorter makespan for an application, they can turn the ᾱ up. On the contrary, if the users care about the cost, they can turn the ᾱ down. Thus, the ᾱ make the FAUSIT have strong usability.


## 5.1. Experiment Settings

The DAG based big data applications we used for the evaluation are obtained from the DAG based applications benchmark provided by Pegasus WorkflowGenerator [28]. We use the largest application from the benchmark, i.e., Epigenomics, since the bigger application is more sensitive to failures. The detailed characteristics of the benchmark applications can be found in [29, 30]. In our experiments, the number of subtasks in an application is ranging from 100 to 1000. Since the benchmark does not assign deadlines for each application, we need to specify the deadlines; we assign a deadline for the applications: it is 1.0CP. Table 3 gives the characteristics of these applications including the count of tasks and edges, average task execution time and the deadlines.Table 3
The Characteristics of the Big Data Application.
Workflows Task Count Range Edge Count Avg. Run Time of Task (Sec) Deadlines Range Epigenomics 100 322 3856.51 1.0CP 200 644 … … 1000 3228Because FAUSIT does not schedule the subtasks on the processors, we hire a high-performance schedule algorithm (MSMD) to determine the map from subtasks to the processors. MSMD is a novel static schedule algorithm to reduce the cost and the makespan for an application. On the basis of MSMD, we use FAUSIT to improve the fault tolerant of an application.


## 5.2. Evaluation Criteria

The main objectives of the FAUSIT are to find the optimal fault tolerant mechanism for a big data application running on cloud. The first criterion to evaluate the performance of a fault tolerant mechanism is how much time is delayed to complete the application, i.e., the makespan to finish the application. We introduce the concept ofmakespan delay rate to indicate how much extra time consumed by the fault tolerant mechanism. It is defined as follows:(18)Mak.Del.Rate=The practical makespanThe ideal makespan.where the ideal makespan represents the completion time for running an application without fault tolerant mechanism and any failures, and the practical makespan is the completion time consumed on practical system which has a fault tolerant mechanism and the probability of failures.The second goal of the FARSIT is to minimize the extra cost consumed by the fault tolerant mechanism. Undoubtedly, to improve the performance of fault tolerant, any fault tolerant mechanisms will have to consume extra computation time for an application. Thus, the extra computation time is a key criterion to evaluate the performance of fault tolerant mechanism. Therefore, we define extra cost rate for the evaluation:(19)Ex.Cost.Rate=The practical processors timeThe ideal processors time.where the practical processors time denotes the time consumed by the processors to running an application for a practical system with failures, The ideal processors time is the sum of the wi in G(V,E) for an application.


## 5.3. The Optimalα for FAUSIT

Before comparing with the other algorithms, we need to determine the optimalᾱ for FAUSIT first. Due to the optimal ᾱ is an experimental parameter, we have to figure it out by experiments.We test theᾱ by the Epigenomics application with 1000 subtasks for 10 times and make the Td=1.0Tc; the results are shown in Figures 5(a) and 5(b). In Figure 5(a), theMak.Del.Rate becomes lower alone with the increased; i.e., a larger ᾱ will lead to a shorter makespan for an application. On the contrary, Figure 5(b) shows that the larger ᾱ will lead to more cost for an application.Figure 5
(a) TheMak.Del.Rate and (b) theEx.Cost.Rate of ᾱ.Through a comprehensive analysis of Figures5(a) and 5(b), we make the ᾱ=0.033 which can equalize theMak.Del.Rate and theEx.Cost.Rate and make sure that both theMak.Del.Rate and theEx.Cost.Rate are smaller than other situations.


## 5.4. Comparison with the Other Solutions

Since the proposed FAUSIT algorithm is a heuristic algorithm, the most straightforward way to evaluate its performance is to compare with the optimal solution when possible. We randomly generate 10 different applications for each scale of Epigenomics shown in Table3, and we make the Td=1.0CP.What should be paid attention to is that although the FASTER is a copy tasks based method, but it is developed for multiple DAG based big data applications. As a result, it cannot compare the performance of it with the other two methods directly. In order to make a comprehensive comparison, on the basis of FASTER, we modified the FASTER to make it handle a single DAG based application, and this new FASTER is denoted by FASTER∗.The average ofMak.Del.Rate andEx.Cost.Rate are displayed in Figure 6. In Figure 6(a), our FAUSIT has the minimumMak.Del.Rate for any size of application. Meanwhile, the FASTER∗ has higherMak.Del.Rate than our FAUSIT, and the Daly’s method has the highestMak.Del.Rate. The result in Figure 6(a) shows that our FAUSIT has the best performance for minimizing the makespan of an application among these methods.Figure 6
The result ofMak.Del.Rate andEx.Cost.Rate.
(a)
Mak.Del.Rate (b)
Ex.Cost.RateIn Figure6(b), due to the fact that the Daly’s method only adopts the check-point mechanism, this makes it has the minimumEx.Cost.Rate. Meanwhile, the FASTER∗ consumes the maximum cost, since theEx.Cost.Rate of it is very large. Interestingly, along with the increase of the count of subtasks in an application, theEx.Cost.Rate of our FAUSIT is becoming smaller, which indicates that our FAUSIT has the ability to handle much bigger scale of application.In conclusion, compared with the FASTER, our FAUSIT outperforms FASTER on bothMak.Del.Rate andEx.Cost.Rate. Compared with the Daly’s method, our FAUSIT outperforms it much more onMak.Del.Rate and only consume 9% extra cost, which still makes our FAUSIT have competitiveness in practical system. Besides, the ᾱ in our FAUSIT can satisfy the requirements from different users; i.e., if the users need a shorter makespan for an application, they can turn the ᾱ up. On the contrary, if the users care about the cost, they can turn the ᾱ down. Thus, the ᾱ make the FAUSIT have strong usability.


## 6. Conclusion

This paper investigates the problem of improving the fault tolerant for a big data application running on cloud. We first analyze the characteristics of running a task on a processor. Then, we present a new approach called selective mirrored task method to deal with the imbalance between the parallelism and the topology of the DAG based application running on multiple processors. Based on the selective mirrored task method, we proposed an algorithm named FAUSIT to improve the fault tolerant for a big data application running on cloud; meanwhile, the makespan and the computation cost is minimized. To evaluate the effectiveness of FAUSIT, we conduct extensive simulation experiments in the context of randomly generated workflows which are real-world application traces. Experimental results show the superiorities of FAUSIT compared with other related algorithms, such as FASTER and Daly’s method. In our future work, due to the superiorities of selective mirrored task method, we will try to apply it to other big data applications processing scenarios, such as improving the fault tolerant of multiple applications in the respect of could providers. Furthermore, we will also investigate the effectiveness of the selective mirrored task method in parallel real-time applications running on cloud.



---

*Source: 4807502-2019-02-26.xml*