Scheduling Workflow Applications on the Heterogeneous Cloud Resources

As cloud computing model recently become promising and enables users to obtain their required services, many users desirous to run their workflow applications on it. Scheduling workflow is one of the most important challenges in the cloud. For optimal use of the capabilities of the distributed system, an efficient scheduling algorithm is needed. Addressing the problem of scheduling workflow applications onto Cloud environment is the main contribution of this paper. Heterogeneity of resource types is one of the most important issues which significantly affect workflow scheduling in Cloud environment. On the other hand, a workflow application is usually consisting of different tasks with the need for different resource types to complete which we call it heterogeneity in workflow. The main idea in this paper is to match the heterogeneity in workflow application to the heterogeneity in Cloud environment. To obtain this objective a new scheduling algorithm is introduced, which is based upon the idea of detecting the set of tasks that could run concurrently and distribute them into different sub-workflows and then allocate each sub-workflow in resource cluster instead of allocating individual tasks. This can reduce inter-task communication cost and thus improve workflow execution performance. First we perform global scheduling and then conduct local scheduling. On the Global-scheduling to achieve high parallelism the received DAG partition into multiple sub-workflows that is realized by WPRC algorithm. On the Local-scheduling, sub-workflows were generated at the global level are dispatched to selected resource clusters. We used the simulation to evaluate the performance of the proposed algorithm in comparison with three well-known approaches. The results show that the proposed algorithm outperforms other approaches in different QoS related terms. *Author for correspondence Indian Journal of Science and Technology, Vol 8(12), 57984, June 2015 ISSN (Print) : 0974-6846 ISSN (Online) : 0974-5645 Scheduling Workflow Applications on the Heterogeneous Cloud Resources R. Bagheri1 and M. Jahanshahi2* 1Department of Computer Engineering, Gazvin Branch, Islamic Azad University, Tehran, Iran 2Department of Computer Engineering, Central Tehran Branch, Islamic Azad University, Tehran, Iran; mjahanshahi@iauctb.ac.ir


Introduction
Cloud computing is the service-focused that delivers hardware infrastructure and software application as services with low cost and high quality 1,2 . These technology advances have led to the possibility of using geographically distributed to solve large-scale problems in science, engineering, and commerce. Currently, cloud computing services are categorized into three classes: Infrastructure as a Service (IaaS), Platform as a Service (PaaS) and Software as a Service (SaaS). These services are available in a pay-per-use on demand model 3,4

. Some researchers
Keywords: Cloud Computing, Cluster resources, Scheduling Algorithm, Workflow consider the benefits of using cloud computing for executing scientific workflows [5][6][7][8] . Several features that are distinct cloud computing from other computing environments consist of: (1) The type and number of compute resources assigned to a workflow are determined by service requests. (2) Compute resources in Cloud are exposed as services that provide a standardized interface for services to access over the network 9 . Workflows constitute a common model for describing a wide range of scientific applications in distributed systems. Usually, a workflow can be represented by a Directed Acyclic Graph (DAG) in which each computational task is represented by a node, and each data or control dependency between tasks is represented by a directed edge between the corresponding nodes. Workflow scheduling is the way of choosing a suitable resource for each task. With Growing up and complexity of workflow, the need of workflow's scheduling be felt more than ever and has become one of the most important challenges in the cloud. The workflow scheduler has to schedule and allocate each task according to the dependency of the workflow's tasks.
As task scheduling is a well-known NP-complete problem 10 . Many heuristic methods have been proposed for distributed system. Most of them try to minimize the total completion time of all tasks (make span) and cost of the workflow [11][12][13][14] . Zeng et al. proposed a budget-conscious scheduler to minimize many-task workflow execution time within a certain budget 15 . In 16 , Abrishami et al. designed a QoS-based workflow scheduling algorithm based on Partial Critical Paths (PCP) in SaaS clouds to minimize the cost of workflow execution within a user defined deadline. Workflow scheduling algorithms are classified into four classes: 1) list-based, 2) duplication-based, 3) clustering-based, and 4) level-based. Most previous workflows scheduling research are based on a list-based heuristic approach. List-based scheduling is a class of scheduling heuristics in which tasks are assigned with priorities and placed in a list ordered in decreasing magnitude of priority. An important issue in DAG scheduling is how to rank the nodes. The rank of a node is used as its priority in the scheduling and then allocates each individual task onto processors. Many list-based heuristics proposed in the literature 14,17,18 . In this paper, we propose an algorithm according to the clustering-based heuristic approach. Clustering is another efficient way to reduce a communication delay in DAGs by grouping heavily communicating tasks to the same labeled clusters 19 . In general, clustering algorithms have two phases: the task clustering phase that partitions the original task graph into clusters for allocation instead of allocating individual tasks. This can reduce inter-task communication cost and thus improve workflow execution performance and post-clustering phases which can refine the clusters produced in the previous phase and get the final taskto-resource map. The rest of the paper is organized as follow: Section 2, describes the scheduling architecture and model used by algorithm. Section 3, the proposed scheduling algorithms are explained. Section 4, evaluates simulation results. Section 5, concludes.

Workflow Scheduling Model
The proposed workflow scheduling model consists of a Cloud application model and a cloud resource model. A Cloud application is modeled by a Directed Acyclic Graph (DAG), G = (V, E, q, w), in which V = {t i |i =1, 2, . . . , m} be the finite set of tasks t i and E be the set of directed arcs of the form e(t i , t j ), An edge e (t i ,t j ) represents the communication from t i to t j where t i is called a prior of t j , and t j is a successor of t i and q i represents the computational cost of task t i . The weight w of e (t i ,t j ) represents the communication cost from task t i to t j .
In cloud resource model, resources have been clustered that formed with multiple VMs. Resource clusters are connected by a WAN, and within a cluster, computational nodes are connected by high speed LAN. Usually, the communication cost by WAN is much higher than LAN. We name the cost of communication between two resources in different cluster as external communication cost similarly; we name the cost of communication within a cluster, internal communication cost. The internal data transfer in cloud similar to the most recently publish is negligible. Each resource cluster represented by C i (1≤i≤m), where i is the unique identification number of the cluster and m is the number of clusters.
When a workflow receives, it will break into subgraphs, according to knowledge about available resources, by executing the Workflow Partition Resource Clusters (WPRC) algorithm and then scheduler will run the local level scheduling and map tasks in subgraph to local computational node. A computational resource is denoted as R i,j where i is the resource cluster id to which this resource belongs and j is the resource id within its cluster. For a resource cluster C i , the number of resources in C i is represented by n i . Let p i,j is the expected performance of R i,j so P i is total computation power of C i , which is the sum of the computational power of all resources in C i : The average performance of all available Computational resources in C i is given by: P is the average computational power of all resource clusters that compute by: We assumed that communication cost between resource cluster C i and C j is represented by Com Cost_C i,j and the average cross-cluster communication cost is defined as: We assumed that the computation power of resource effect on the time required completing a task and the time to finish a data transfer is commensurate with the communication cost of the link.

WPRC: Scheduling Algorithm
High parallelism means to dispatch more tasks simultaneously to different resources. To achieve this, the main task graph needs to be partitioned into subgraphs and each subgraph has to be assigned to a resource cluster. The main parameter must be determined in partitioning of graph, is the number of partitions should be made (N). To determine N, CTC parameter is used. CTC is the ratio of communication-to-computation. A high CTC value means a task graph is computation intensive. Formally, CTC is defined as: Where q − is the average processing requirement of all tasks. As the CTC increases, high parallelism is preferred because more computational power is required. WPRC determine the number of graph partitions to be created, N, according to different workflow patterns and communication costs: Here β is the accelerating factor and ∆p is the standard deviation on the computational power of different resource clusters. It is clear that N is always no greater than the number of available resource clusters.
With the number of subgraphs, to be created, WPRC is to specify how tasks in the main graph should be assigned. To achieve high parallelism and avoid inessential external communication, the size of a subgraph assigned to a resource cluster should be as large as possible under a certain threshold value. The weights of edges connecting different subgraphs should be as small as possible to minimize communication cost. According to purpose, we need to detect the set of tasks that could run concurrently and distribute them into different subgraphs and then specify the maximum number of nodes that could run concurrently when assigned to the same resource cluster that call Maximum Concurrent Node (MCN). To get MCN, two parameters are defined: For a node t i in a DAG, its Earliest Start Time (EST) is defined as follows: Where pred (t i ) is the set of immediate predecessors of t i and Execution Time (ET) of t j is defined as: The Earliest Finish Time (EFT) of t i is defined as follows: Now, we can specify which nodes could run concurrently. We call t i and t j parallel peers, if following equation is satisfied to one of them: if the EST( t i ) after EST (t j ) and before the EFT (t j ) or the EFT (t i ) after EST (t j )and before the EFT (t j ). By checking parallel peers of every node, we can find the largest set of concurrent nodes in task graph G, whose size is the value of MCN. The size of a partition is also related with the computational power of resource clusters. We assume that the set of resource cluster services that are offered include: If the number of resources in cluster i that offer service j is n i,sj ,the average number of all available resources in cluster i for service j is given by: Sum of Computing power of all resources in the cluster i that offer service j is Computation power (p sj ), the average Computation power of all available resources in cluster i for service j is given by: ET t j Where, d i,j is the standard deviation of the performance fluctuation of P i,j in vari ous time slots, and W i is sum of weights of all resources in cluster i for various services that can be obtain by the following equation: And ΔS i is the standard deviation of the computational capacity of resources in r i (so a larger d i,j means the performance of p i,j is more unstable and a larger ΔS i implies that r i is more heterogeneous). After resource clusters are ranked, N out of m of them, having the N highest ranks is selected for the current job. We assume these clusters are r 1 , r N . Then the initial threshold value T i ' of r i is defined as: The above equation, the parallel computation power of each cluster is also taken into account. Then the parallel threshold value T i of each subgraph to be created is given by: The pseudo-code for the WPRC is given in Algorithm1. WPRC is described as follows. When a scheduler receives a job, it first traverses the job's DAG G to compute its CTC, the number of partitions to create N, and the level of each task node. Then, the scheduler selects N resource clusters whose ranks are the highest N out of m, according to its knowledge. A graph partition iteration checks every remaining nodes in G to determine whether the node can be put into a subgraph G'.

Algorithm 1 WPRC
Input: A task DAG G (V, E) and available Cloud resource clusters C 1 …C m . Output: A subgraph of G and assigned to resource cluster C i .

Compute CTC, N and EFT and EST of each node in G,
and cluster ranks R and threshold values T; 2. Mark all nodes unassigned; Resource failures to be statistically independent and follow a constant failure rate Fr j for each resource j. We consider the reliability of an activity i as the Probability of successful completion on a resource j, modeled using an exponential distribution 20,21 : Total reliability of all resources in the cluster i that offer service j is given by: Where, T is the average of tasks computation. The average reliability of all available resource resources in cluster i for different service is given by: Then for each resource cluster, a weight (W) can be obtained for each type of service in the resource cluster by the following equation: After we get the weight of all services in the all resource clusters, we then create the clusters matrix: In which W ij is weight of the resource cluster i for the service j.
To be adaptive to the dynamic and heterogeneous nature of the computational cloud, WPRC introduces two parameters to describe the related properties of a resource cluster, namely the Cluster Rank (R i ) and Parallel Threshold (T i ). Thus 3. Select the resource cluster C i with the highest threshold value Ti; 4. Find the largest edge e (a, b) in which a, b have not been checked; 5. IF Br (G' + a) ≤ Ti and Br (G' + b) ≤ Ti { 6 . Add a and b to G' and mark a and b as checked;  7. } 8. G = G -G'; 9. Put (G', C i ) in the output set.
WPRC itself does not give the task-to-resource map, but on the global level only partitions original workflow to different subgraphs and then dispatches subgraphs to different resource clusters. So, on the local level, another scheduling algorithm is needed to get the final schedule of the received subgraphs. For scheduling a subgraph in special resource cluster need to rank all nodes and then assign node to resource according to its priority. Usually, the priority of a task node can be obtained by finding the maximum distance from this node to the starting node. Distance means the sum of computational and communication costs along a certain path. To estimate the completion time of nodes, we use the average performance value of resource cluster. The rank or priority of a node is defined as: For the tasks to resources mapping, initially priority of each task is computed. Then at each step, a task that has the highest priority in the ready queue RQ is selected. The task is placed in a ready queue RQ if all its predecessors have been implemented and the middle results are provided. Once the task node is selected, the function Select Processor (ti) for choice suitable source is called. Mapping algorithm adopts a forward-looking approach to decision-making based on only the current state of resources and task. Function Select Processor (t i ) compute the execution time of task t i for all the resources that provide the required service t i . It can be obtained from the following equation: After the task execution time were computed from all sources, then the resource, where is minimize, is selected. After each assignment, the rank of tasks will be update. The pseudo-code for the Mapping is given in Algorithm 2.

Algorithm 2 Mapping
Input: A task graph G and a set of resources R 1 … R n Output: A mapping of tasks to resources

Simulation Model
We evaluate the performance of WPRC using CloudSim 22 , which has been widely adopted for the modeling and evaluation of cloud-based solutions. In the experiments, three resource clusters are used. Each cluster consists of different resources number connected by a LAN. The resource clusters are connected by a WAN. In each resource cluster, the resources in it have different computing power and delivered different type of services. In terms of input task graphs, We used random graph generation with the ability of generating a variety of task graphs according to different configuration parameters, such as average number of task nodes of each graph, average outgoing and incoming degrees for each node in a graph, and computational and communication cost for each type of task nodes and edges. The count of nodes in each workflow graph set between 25 and 100 nodes. Each graph has a single entry and a single exit node. We used the random rank t max rank t q p e i,j inter ET t = q /P i i ( ) (24) graph generator discussed in 17 . This random graph generator requires following input parameters: • V: The number of task in the DAG.
• Out degree: The ratio of maximum out edges of a node to total nodes of the DAG. • Communication to Computation Ratio (CCR). It is the ratio of the average communication cost to the average computation cost. • Β: The computational heterogeneity factor of resources.
• α: The depth parameter of the DAG. This parameter indicates the depth of a DAG by using the uniform distribution with the mean value equal to The values for the input parameters are shown in Table 1. The following four metrics are used to evaluate the performance of the proposed algorithm:

Simulation Results
To evaluate the performance of the proposed algorithm we compared it with the HEFT 17 , the QRS 23 and the PFAS 24 which are well-known methods for DAG scheduling. The performance metric used in the simulation is the average make span. To test our proposed algorithm, the following parameters are considered in the experiment: 1) The average number of task nodes in a graph v 2) The ratio of the average degree of a task node to the total number of tasks in a graph (Edge density in a graph) 3) CCR. Figure 1 present the average make span of our approach over the others approach with respect to different number of task. Figure 2 present the average makespan of our approach over the others approach with respect to different degree. Figure 3 present the average make span of our approach over the others approach with respect to different CCR.
The result of simulation indicates that the average make span of workflow's graph with our approach algorithm is better than other algorithms.

Further Analysis
Because our proposed algorithm is based on clustering method and it is base on concurrent tasks, we define tasks Graph Concurrency degree (GC a ) that can be obtained from the following equation:    Where MCN is the maximum number of tasks in a workflow's graph that can be executed in parallel and NT is the Number of tasks in a workflow. Simulation results in Figure 4 show that whatever amount GC becomes larger (ratio the number of concurrent tasks to the total tasks) Difference in performance of the proposed algorithm with other algorithms will be more and improvement of the proposed algorithm compared to other algorithms will be more.

Conclusion
We propose a new Workflow Partition algorithm WPRC for a workflow scheduling in the cloud. Scheduling is in two phases, on the global level, WPRC clusters workflow to achieve high parallelism and on the local level, Sub workflows was generated at the global level are then dispatched to selected resource clusters. It not only considers the heterogeneity and dynamism of cloud resources, but also uses an adaptive strategy according to different workflow patterns and resource topologies. When it partitions a task graph, WPRC try to minimizing the cost of workflow execution and the make span. Future work includes improving the current strategy, and uses other parameters for clustering the workflow.