Hadrien Croubois

Eddy Caron

Communication-aware task placement for workflow scheduling on DaaS-based Cloud

Keywords: workflow, Cloud, scheduling, clustering Part 1

Cloud platforms have emerged as a leading solution for computation. In the meantime, large computations have shifted from big parallel tasks to workflows of smaller tasks with data dependencies between them. Task placement is a major issue on Cloud platforms, especially considering the impact of data exchanges on cost and makespan. In this paper, we investigate the consequences of network contention regarding the use of existing scheduling policies on DaaS-based platforms (DaaS for Data as a Service). We show here that the legacy algorithms use inefficient network models. We then modify those algorithms using a new model inspired by DaaS-based Cloud platforms. Thus, we manage to statically pack tasks so that a batch scheduler could deploy many real-time submitted workflows on a dynamic Cloud platform. Simulations of Fork-Join workflows deployment using SimGrid show that our algorithm reduces computation time as well as deployment costs.

Introduction

Nowadays, computing platforms are used to execute increasingly complex operations composed of multiple interdependent tasks. Examples of such workflows are given in [START_REF] Abrishami | Deadline-constrained workflow scheduling algorithms for Infrastructure as a Service Clouds[END_REF]. These can be submitted in real time by multiple users. Meanwhile, computing platforms have moved from institution-owned clusters to externalized Clouds. This new computing paradigm calls for new mechanisms to efficiently schedule tasks and provision resources.

Previous scheduling mechanisms were designed to statically schedule workflows on fixed computing clusters, and focused on communication cost. On Cloudbased platforms, recent approaches now focus on other issues such as Virtual Machine (VM) provisioning, resource sharing among users and failure handling.

As Cloud instances can migrate and the topology of the network is usually unknown, it is difficult to estimate communication cost and to tweak task placement accordingly. A usual solution is to deal with data locality and to migrate the environment.

Another argument against the use of task clustering algorithms is that it requires knowledge of all the jobs. In a dynamic context, the whole schedule would have to be recomputed for every new job submitted to the platform, leading to poor platform scalability. On the other end of the spectrum, batch scheduling algorithms are scalable in the context of many users submitting many independent jobs in real time.

Still, many Cloud solutions rely on Data as a Service (DaaS) tools for communication. Among such tools are Amazon S3, Dropbox, NFS and others. On an otherwise distributed infrastructure, DaaS are centralized elements that somehow constrain the level of parallelism. In fact, the capacity of a node to send and receive files from the rest of the network is limited by the bandwidth available between this node and the DaaS. Even with a distributed structured DaaS, availability of a specific piece of data to all nodes is bounded by the capacity of the nodes to access the data on different instances of the DaaS and consequently by the DaaS inner synchronization mechanisms. The whole DaaS can therefore be seen as a single entity potentially distributed among multiple machines.

To build an autonomous workflow manager for the Cloud, we focused on task packing. Our idea was to analyze each job workflow and to determine tasks packing for optimal placement without any prior knowledge of the platform current deployment. The resulting information could be useful to a batch scheduler downstream. Furthermore, if this workflow analysis could be done independently of the status of the platform, then it would not be necessary to recompute it with each subsequent change in the platform deployment, therefore ensuring good scalability.

In this paper, we focus on static clustering algorithms for workflows with data dependencies. We start by building workflow and platform models (Section 3). Later, we show how legacy algorithms behave on cloud platform (Section 4), and propose an evolution based on a Cloud-inspired network topology (Section 5). We then see how this new clustering method can help to pack tasks for efficient workflow deployment in the Cloud (Section 6).

Part 2

Related work

Since the dawn of parallel and distributed systems, the scheduling issue has been considered in many context and with many different objectives, each relevant to some platform specificity. In this section we will give an overview of the existing approaches.

2.1

Clustering algorithms

The goal of clustering algorithms is to pack tasks into clusters prior to any execution. Given a global vision of the platform and the tasks to execute, a clustering is a mapping of tasks to the nodes of the platform, with all tasks being executed on the same node constituting a cluster. With this approach, expensive computation is needed to achieve a very efficient clustering, yet any change on the platform or the tasks to execute would lead to the whole clustering being recomputed.

In this category we find algorithm dealing with both homogeneous platforms [START_REF] Wu | A programming aid for message-passing systems[END_REF][START_REF] Yang | List scheduling with and without communication delays[END_REF][START_REF]Dsc: Scheduling parallel tasks on an unbounded number of processors[END_REF][START_REF] Kwok | A static scheduling algorithm using dynamic critical path for assigning parallel algorithms onto multiprocessors[END_REF] and heterogeneous platforms [START_REF] Topcuoglu | Performance-effective and low-complexity task scheduling for heterogeneous computing[END_REF].

2.2

Batch scheduling for Cloud

The issue of the Cloud batch is the scheduling of many independent tasks and services, submitted dynamically, on heterogeneous platforms. One of the required key features is to have the addition of new tasks to the schedule be a very simple operation, thus ensuring the scalability of the system. Other elements, such as tasks priorities or energy consumption, can be used to make the schedule fit ones' objectives. On Cloud platforms, batch schedulers also adjust the platform size in order to meet all the deadlines while limiting the deployment cost [START_REF] Yang | An algorithm in swindew-c for scheduling transaction-intensive cost-constrained cloud workflows[END_REF].

While some tools handle graph deployment, the few that try to optimize placement based on data dependencies have poor network modeling [START_REF] Varalakshmi | An Optimal Workflow Based Scheduling and Resource Allocation in Cloud[END_REF].

2.3

Workflow deployment on Cloud

It is still an open problem to efficiently consider the constraints imposed by the real-time submission of workflows while handling the dynamism of Cloud platforms. All previous contributions focused on specific issues while leaving other aspects unconsidered.

A survey showed that most contribution did not consider the impact of data transfers [START_REF] Wieczorek | Towards a general model of the multi-criteria workflow scheduling on the grid[END_REF]. Moreover, when this issue was considered, it was on simplified DAGs which failed to accurately model the whole extent of real applications.

We have identified two recent contributions which depict the current state of the art.

Mao et al. [START_REF] Mao | Scaling and scheduling to maximize application performance within budget constraints in cloud workflows[END_REF] dealt with a single workflow in which the tasks have different resources requirements. Their algorithm not only packs tasks but also determines which type of nodes to deploy. However, this result does not account for multiple workflows. It also does not handle contexts in which communications go through a DaaS. We might be interested in investigating the underlying communication model in respect of the targeted platforms' behavior.

Malawski et al. [START_REF] Malawski | Cost-and deadline-constrained provisioning for scientific workflow ensembles in iaas clouds[END_REF] handled multiple dynamically submitted workflows composed of single-threaded tasks. In this work, the Cloud platform used for execution is dynamic but has an upper bound. Therefore, some workflows -with a lesser priority -can be dropped if the platform cannot be extended as required. Similarly to Mao et al., the communication model used in this paper does not match our observations of DaaS-based platforms.

Part 3

Workflow and platform modeling for communication-aware scheduling

Scheduling algorithms rely on specific definitions of resources and of the way tasks are executed on these resources. Before discussing the scheduling strategies themselves, we have to define the different objects we will be dealing with.

3.1

Generic network topology for DaaS-based Cloud platforms

It is commonly accepted that characterizing platform topology is a real issue on the Cloud. VM can migrate and resources such as routers can be shared with other users. Still, in an effort to build a scheduling policy both task and data transfer-oriented, we need to understand and predict the behavior of data transfers on this platform. As discussed in the introduction, communicating through a DaaS induces a centralization of the communications. Our analysis is that this centralization will lead to contention at the links between a node and the DaaS. While the DaaS is a critical element, commonly located near the center of the network, the computing stations can potentially be very far from it. This led us to a model where a node ability to place and retrieve data from the DaaS is constrained by the bandwidth between the node itself and the code of the network. This topology is illustrated in Fig. 1. This model is generic enough for it to be independent of the actual Cloud deployment and of VM migrations, as long as the nodes bandwidth is not overestimated.

3.2

Data-centric representation of workflows

Workflows are usually represented as weighted graphs of tasks, with the weights on the nodes representing the computational cost of the tasks (in flops -floating point operations -) and the weights on the edges representing the cost of the communications (in bits of data transferred). However, as we moved to DaaSbased Cloud platforms, our representation of workflows needs to evolve accordingly.

While previous representations focused on the amount of data to be transferred between tasks, a more relevant approach would be to focus on data objects.

If we consider a fork-join distribution pattern, there is a major difference between sending n different pieces of data to n different agents and sending a single piece of data to the same n different agents. While in the second case we have a single upload from the initial task to the DaaS and n parallel downloads from the DaaS, in the first case we have a single task (the initial task) uploading the n different pieces of data to the DaaS at the same time, which would cause contention between the initial task and the core of the network where the DaaS is located.

Therefore, the first contribution of this paper is an extended "data-centric" representation of workflows which includes details about the different pieces of data produced and consumed by the tasks. Our representation (Table 1) is an acyclic-oriented bipartite graph, with nodes from one side representing weighted tasks and nodes from the other representing pieces of data weighted by their size. Edges have no weight and only represent dependencies between pieces of data and their producers/consumers. Fig. 2 shows how two very different communication patterns have the same legacy representation. According to our network model, we expect bouts of network congestion when a single task uploads or downloads multiple files. Thus, we can expect some contention for the final task downloads in both cases (Fig. 2b and Fig. 2c). Yet, only the second case (Fig. 2c) should undergo upload contention for the initial task.

Part 4

Legacy communication-aware scheduling algorithms behavior on current platforms

Historical static scheduling algorithms such as DCP [START_REF] Kwok | A static scheduling algorithm using dynamic critical path for assigning parallel algorithms onto multiprocessors[END_REF] were designed to take into account the impact of communication latency on workflows deployment. In this section we will see how they perform on DaaS-based platforms and analyze why they do not stand up to the task.

Contention on DaaS-based Cloud platforms

A common distribution pattern in workflows is the fork-join mechanism. In such context, an initial task is to send pieces of data to n independent tasks. Afterwards, those n tasks are sending back the results to a final task. We assume here that the n pieces sent by the initial tasks are different pieces of data of the same size.

When scheduling such a DAG using DCP, the algorithm takes action to achieve a high level of parallelism and thus places most tasks on distinct nodes. However, when executing the resulting task placement on a Cloud infrastructure, we see that communication between the first task and the DaaS (upload of n files to the DaaS) and between the DaaS and the final task (download of n files from the DaaS) drastically suffers from contention. Fig. 3 shows that while DCP plans for all transfers to be simultaneous, the network congestion drastically reduces the performances of both upload to the DaaS and download from the DaaS. This results in transfer times about n times slower then predicted by DCP.

This gap between the communication model underlying DCP and the reality of current distributed platforms, in addition to the difficulty of handling the dy- Experiments were carried out using Grid'5000 testbed. Used nodes were on the Sagittaire cluster and the DaaS was a 10G chunk reserved on storage5k.

namic submission of workflows by users, explains why such scheduling algorithms are not efficient for tasks placement on Cloud platforms. While other static scheduling algorithms have improved on DCP in many aspects, communication modelization is not among them. Algorithms like HEFT or CPOP have the benefit of handling heterogeneous platforms. However their design is based on the assumption of a limited number of ressources and thus do not match the cloud paradigm where new large instances are always available. Focusing on DCP helps us study the communication aspect of things without getting in the trouble of dealing with specificities that would not fit the cloud paradigm.

c(u, v) = 0 if C(u) = C(v) ω(u → v) otherwise (1a) T L(v) =    0 if v has no predecessor max u∈pred(v) (T L(u) + ω(u) + c(u, v), avail T L (C, v)) (1b) BL(u) =    ω(u) if u has no successor ω(u) + max v∈succ(u) (c(u, v) + BL(v), avail BL (C, u)) (1c)
The critical path computation used in DCP static scheduling algorithm is as shown in Eq. 1 (with u and v tasks in a DAG of tasks -workflow -, ω a weighting function on the tasks and data and C a linear clustering of the DAG)

From those values we can compute other metrics:

• Dynamic Critical Path Length (Makespan):

DCP L = max t∈T (T L(t) + ω(t)) = max t∈T (BL(t))
• Absolute Earliest Start Time:

AEST (t) = T L(t)
• Absolute Latest Start Time:

ALST (t) = DCP L -BL(t)
• Path Length (equal to DCP L for critical tasks):

P L(t) = T L(t) + BL(t)
We see that these formulas do not consider the possible impact simultaneous transfers could have on one another. In fact they disregard any form of contention. At first it looks like it considers a complete clique network where any pair of nodes can exchange data without being affected by the rest of the network, but it is in fact even stronger than that as any number of transfers between the same two nodes can take place at the same time without them having to share the bandwidth.

While such a topology could have made sense when dealing with small clusters, the gap with new distributed platforms is tremendous and explains the incapacity of DCP to efficiently predict communications. This leads to poor performances of the resulting task placements.

Part 5

DCP evolution for DaaS-based cloud infrastructures

In the previous section, we discussed the behavior of DCP and its inadequacy to efficiently schedule workflows on DaaS-based Cloud platforms. In this section we will see how the models discussed in section 3 can be used to improve DCP ability to schedule workflows on modern platforms.

A communication model for Cloud infrastructures

Using the unmodified structure of the DCP algorithm (see Eq. 1), our objective is to use the platform and workflow models developed in Section 3 to modify the way communications affect workflow deployment.

In DCP equations (see Eq. 1), communications are modeled by the c(u, v) formula. This is to be modified in order to match our communication model. The modification described in Eq. 2 involves the computation of the worst case latency between tasks depending on their placement. If tasks u and v are placed on the same node, the communication cost between them is null (Eq. 2a). On the other hand, if u and v are placed on different nodes, we have to consider the upload time of all data produced by u and the download time of all data required by v. The worst case being when the tasks produced by u and required by v are the last to be uploaded by u and the first to be downloaded by v (Fig. 4).

c(u, v) = 0, if C(u) = C(v) c(u, v) = d∈data u∈d.src v / ∈d.dst islocal C (d)=0 ω(d) (2a)
In Eq. 2, the first sum (Eq. 2a) corresponds to the upload by task u of all data not required by v. The second line (Eq. 2b) corresponds to the interlaced upload by task u and download by task v of all data produced by u and consumed by v. Finally, the last sum (Eq. 2c) corresponds to the download by task v of all required data produced by tasks other than u. Those are also visible in Fig. 4.

In the third sum of Eq. 2 (Eq. 2c) it is not necessary to consider the download of data produced on the same node. Similarly, in the first sum (Eq. 2a) we do not consider the upload of data which are consumed locally (producer and consumers all on the same node). The islocal predicate is computed as following.

islocal C (d, u) = 0, if C(d.src) = C(u) 1, otherwise (3a) islocal C (d) = v∈d.dst islocal C (d, v) (3b)
This evaluation of the communication-induced dependencies between two tasks corresponds to a worstcase scenario. It is most likely that a specific ordering of the communications could give us better results but, as always in static scheduling, we put ourselves in the worst-case scenario.

5.2

Modified top-and bottom-level computations

While DCP takes computation resources into account, we also need to consider node-based networking resources. If two tasks, with short computation time but with large amount of data to upload, are placed on the same node, the second task might be over before the data produced by the first task have been sent to the DaaS. The second task would therefore try to send data using an already busy uplink. Keeping track of the uplink and downlink availability is paramount when scheduling multiple tasks on the same node. The modified formulas for critical path computation, accounting for each node network availability, are described in Eq. 4 and graphically shown in Fig. 5.

Similarly to the way DCP deals with node availability, CPU (equivalent of node availability), uplink and downlink availability are updated during the linear scheduling of tasks. Each time a task is placed, the availability values of the concerned node, which are used to determine deployment timings, are updated in anticipation of the next task to be placed on this node. Initially, all availability values (which can be seen as time constraints on the task deployment) are initialized to 0. .

cup(u) = d∈data u∈d.src islocal C (d)=0 ω(d) (4a) c down (v) = d∈data v∈d.dst islocal C (d,v)=0 ω(d) (4b) c total (u, v) = d∈data u∈d.src v / ∈d.dst islocal C (d)=0
T L(v) = max u∈pred(v) (T L(u) + ω(u) + c total (u, v), avail up T L (C, u) + c total (u, v), avail down T L (C, v) + c down (v), avail cpu T L (C, v)) (4d) BL(u) = max v∈succ(u) (c total (u, v) + BL(v), c total (u, v) + avail down BL (C, v), cup(u) + avail up BL (C, u), avail cpu BL (C, u)) + ω(u) (4e)
As previously mentioned, we retained the structure of the DCP algorithm (Algorithm 1), to which we added our tailor-made formulas to compute the critical path. This gave us a generic task placement scheme which can deal with any DAG and which takes potential network congestion into account.

Part 6

Results and discussions

In the previous sections, we described a generic model for DaaS-based platforms as well as a variant of DCP that fits this model. In order to validate the relevance of this clustering algorithm to deploy DAGs on the Cloud, we are now going to compare it against a fully distributed scheme (all tasks are deployed on their own node, with no clustering) as well as against DCP. This validation will also take into consideration our DAG description model. We will consider two forkjoin cases which have the same description according to the legacy representation (see Fig. 2) but for which the new representation can help make more adequate decisions. True values were obtained on a simulated Cloud platform using SimGrid which has been shown to efficiently model concurrency and link sharing in large-scale networks [START_REF] Casanova | Versatile, scalable, and accurate simulation of distributed applications and platforms[END_REF].

6.1

Comparison of clustering heuristics Fig. 6 shows both predicted and experimental Gantt charts obtained using different placement policies on different platforms. DCP leads to a very high level of communication parallelism to efficiently use many nodes and achieve a low makespan. However, using such a clustering on a simulated Cloud platform showed that congestion limits the actual data parallelism, leading to a much longer makespan.

We also see that results obtained using our model are very close to those simulated by SimGrid. This shows that our model makes clustering decisions based on a realistic approximation, leading to good results. The lower level of parallelism given by our algorithm not only reduces communication-induced latency but also limits the number of nodes to deploy. Further results also show that the different data distribution patterns in single-data and multiple-data fork-join DAGs lead to relevant clustering decisions.

6.2

Economical outcomes

While all deployments of a DAG, regardless of clustering, correspond to the same tasks being executed and therefore to the same amount of core-hours used, changing the clustering can affect the deployment cost. It is important to note that the main objective of static clustering algorithms such as DCP is to reduce the global makespan, assuming unlimited resources. In a Cloud context, this translates to an assignment of tasks to nodes, with a potentially very large number of nodes deployed. In a context where nodes are billed proportionally to the number of core-hours used, we could hope that, as the same amount of computation is achieved, the cost would roughly be the same, with any difference being caused by constraints on the reservation time (e.g. at least one hour).

However, these assumptions do not take into account the time during which a node is retrieving data prior to running a task or sending data after having run a task. During this time, we have to pay for the node even though we do not use its computing potential. Avoiding network congestion and ensuring smooth data transfers is a way of limiting this waste of CPU time. Table 2 shows the number of nodes used, the makespan, and the total deployment cost for our two models of fork-join DAGs using different scheduling algorithms. It is clearly visible that, in addition to reducing the makespan of the DAG, using the right clustering algorithm can help reduce deployment costs.

6.3

Computation to communication ratio

For our experiments, we used synthetic fork-join workflows where the computation time of each task was equivalent to the transfer time of one piece of data from/to the DaaS using the full bandwidth of the node. This ratio between computation and transfer times highlights congestion issues. In this case, single-node clustering achieves good results, as parallelism rapidly causes network congestion. With other ratios, we observe similar behaviors, with the DaaS-aware DCP being the fastest of all, only beaten on the cost aspect by the very slow single-node approach.

6.4

Future work: cluster consolidation While the absence of dependencies between some tasks prevents both DCP and the DaaS-aware DCP from merging them, doing so could have a positive effect on the cost while maintaining the lowest possible makespan. This step, called clustering consolidation [START_REF] Mao | Scaling and scheduling to maximize application performance within budget constraints in cloud workflows[END_REF] is out of the scope of this paper. However, further work should explore how to adapt existing algorithms to our model.

Part 7

Conclusion

In this paper, we showed that the network topology is a key factor in predicting communication patterns and should therefore be considered by clustering algorithms. By designing a generic network model, we managed to improve the results of static scheduling in the context of DaaS-based Cloud platforms. In fact, the resulting clusters are both more efficient in terms of makespan (primary objective) and in terms of deployment cost compared to previous non-networkaware clustering algorithms.

We expect to use those results as the first component of an autonomous workflow manager. The next step is to integrate the computed task packing into a Cloud batch scheduler. With both components, we plan to contribute to or to build a tool that would efficiently deploy workflows on Cloud platforms and where the generic platform model could easily be reconfigured to adapt to changes in the platform paradigm. organizations (see https://www.grid5000.fr). Platform simulations were carried out using SimGrid [START_REF] Casanova | Versatile, scalable, and accurate simulation of distributed applications and platforms[END_REF].

Figure 1 :

 1 Figure 1: A generic model of DaaS-based network topology.

Figure 2 :Figure 3 :

 23 Figure 2: Representation of a fork-join DAG with n = 5 independent jobs.

Algorithm 1

 1 DCP static scheduling algorithm C ← empty clustering (one node per task) compute BL and T L for each task using C while ∃ unmarked dependency between tasks do (u, v) ← edge with the largest path length (most critical). Resolve ties by edge size (select largest).C ← C.mergeClusters(u, v) compute BL and T L for each task usingC if DCP L(BL , T L ) ≤ DCP L(BL, T L) then (C, T L, BL) ← (C , T L , BL ) end if mark (u, v)Dissecting the network model underlying the critical path computation DCP relies on the computation of the critical path and on the zeroing of critical dependencies (see Algorithm 1). This critical path is determined through the computation of the Bottom Level (BL) and Top Level (T L) values for each task. Those values are used to highlight the constraints from the DAG input to the specified task and from this task to the DAG output.In fact, those BL and T L computations account for a worst-case scenario with respect to communication latency. This implies that the computation of those values is based on an implicit communication model.

Figure 4 :

 4 Figure 4: Preview of the communications between two tasks for a data-based workflow on a DaaS-based platform (see Eq. 2).

Figure 5 :

 5 Figure 5: Preview of the critical path computation taking node constraints into account in DaaS-based platforms (see Eq. 4).

Figure 6 :

 6 Figure 6: Comparison of the different clustering policies (Gantt charts and their associated makespan) for multi-data fork-join DAG (n = 16). For each sub-figure, the X axis represents time (arbitrary units). Grey rows represent cores. Boxes in those rows are, from top to bottom, downlink utilization, tasks execution and uplink utilization.

Table 1 :

 1 Workflow and Clustering notations.

	Notation	Space	Description
	T		Set of all tasks
	ω(t)	T → R	Computational cost of t
	D		Set of all pieces of data
	d.src	T	Producer of data d
	d.dst	P(T )	Consumers of data d
	ω(d)	D → R	Communication cost of d
	C	T → VMs	Clustering
	4.1

Table 2 :

 2 Cost and makespan details of different clustering policies for single-data and multi-data fork-join DAG.

	DAG	Algorithm	#Nodes	Makespan (t)	Cost (core×t)
	Single Data Fork-join, as showcased in Fig. 2b, with n = 16	One task per node Single node DCP DaaS-aware DCP	18 1 14 2	22.024 18.000 18.024 13.012	67.204 18.000 56.168 20.012
	Multiple Data Fork-join, as showcased in Fig. 2c, with n = 16	One task per node Single node DCP DaaS-aware DCP	18 1 14 5	37.024 18.000 33.803 14.000	82.204 18.000 70.156 26.048

Acknowledgments

Experiments presented in this paper were carried out using the Grid'5000 testbed, supported by a scientific interest group hosted by Inria and including CNRS, RENATER and several Universities as well as other


Yves Caniou
email: yves.caniou@ens-lyon.fr

Hadrien Croubois
email: hadrien.croubois@ens-lyon.fr

Gaël Le Mahec
email: gael.le.mahec@u-picardie.fr

Standardized multi-protocol data management for grid and cloud GridRPC frameworks

published or not. The documents may come

Standardized multi-protocol data management for grid and cloud GridRPC frameworks

Yves Caniou, Hadrien Croubois, Gaël Le

Introduction

Many applications use RPC-like mechanisms to distribute computations over nodes of clusters and supercomputers composing some distributed systems like a grid, a cloud, or both (now referred as sky computing). Combined with connections to huge databases, they more or less transparently provide scientists with the possibility to focus on their core thematic, giving them more time to deal with data analysis, without dealing with the underlying complexity of all the different mechanisms involved into job and data management. More lately applications even directly couple analysis, graphical representations and such, making platform management only a part of their project, whose actions are generally available through some web site. And surprisingly, when considering a new area, a new platform, new independent pieces of software are often developed instead of using previous work, software or standardized APIs.

The Open Grid Forum standard defining the GridRPC paradigm, namely Remote Procedure Call over the Grid, has been published in 2007, benefiting from 10 years of experience by their respective authors. Simple and easy to use, it has been completed with a standardized data extension only recently. This extension to the native API proposes to expert users to easily handle remote data and to optimize distributed applications with prefetch, migration or replication of possibly distant data using multiple asynchronous transfers together with remote procedure calls on available distributed computing resources.

Based on preliminary experiments [START_REF] Caniou | Transparent Collaboration of GridRPC Middleware using the OGF Standardized GridRPC Data Management API[END_REF][START_REF] Camillo | Improvement of the data management in GridTLSE, a sparse linear algebra expert system[END_REF], applications also benefit from multiadministration sites resources managed by multi-middleware (inherent to interoperability provided with the implementation of the API data extension) and target not only traditional Grids but any distributed platform possibly composed of resources from the Cloud [START_REF] Muresan | Scheduling and deployment of large-scale applications on Cloud platforms[END_REF].

In an attempt to simplify and develop interoperability, and to unify previous works, we propose here a library managing both GridRPC and GridRPC Data Management APIs. We present an overview of the project architecture, designed with a very modular prospect, relying on middleware and data manager modules but also bringing inner data manager capabilities and transfer protocols. Having in mind not to go too much into details, we highlight here some of its features, such as the asynchronous requests management and the transfer management, which involves mapping and scheduling aspects: there is interesting potentiality for optimization at the data operation level, with scheduling to reduce the completion time of a data operation when several sources and several destinations are provided but not necessarily interconnected; and at the workflow/dataflow level to reduce any [sub part of an] application graph. At the moment, the library provides modules for the grid middleware Diet and Ninf , and data manager modules for projects and protocols like Dagda, iRods, webdav (used for webbased repositories like dropbox, owncloud), ftp and rsync.

The rest of the paper is organized as follows: next section explains the motivations behind the GridRPC DM API and some related work. Section 3 presents the global design of the implementation, the different issues that the API leads to and their solution. Section 4 presents some validation experiments and after explaining some future work directions, we conclude in Section 6.

State of the Art

The GridRPC Data Management API, Summary

The GridRPC DM API [START_REF] Caniou | Data management API within the GridRPC[END_REF] introduces the concept of data handle and with it, several GridRPC data types to provide standardized information, for example lists of input and output URIs to give the locations of respectively source and destination [remote] data, with the according protocols to access it at the considered location). It also defines mode managements for a client to characterize the persistence of the data in the system, etc. All actions (initializing, transferring, waiting for completion of asynchronous transfers, etc. ) are provided with only 12 functions.

This standard answers at the API level to issues related to feasibility of the computation by decoupling the data from its locations and from protocols to access it; to performance using different sources and protocols to access a remote data, providing explicit data management with the possibility to prefetch and to migrate data, as well as the possibility to rely on some smart middleware to transparently handle data management; and to extensibility by providing containers of data. It also solves portability, making GridRPC codes portable from one middleware to another.

Related Work

Similar works can address some data management issues in the GridRPC but only separately and without integration into remote procedure call: one can store data on a distributed file system like GlusterFS4 or GFarm [START_REF] Tatebe | Gfarm grid file system[END_REF] to deal with automatic replication; OmniRPC introduced omniStorage [START_REF] Nakajima | Performance evaluation of data management layer by data sharing patterns for GridRPC applications[END_REF] as a Data Management layer relying on several Data Managers such as GFarm and Bittorrent. It aims to provide data sharing patterns (worker to worker, broadcast and all-exchange) to optimize communications between a set of resources, but needs knowledge on the topology and middleware deployment to be useful; Diet also introduced its own data managers (DTM and Dagda [START_REF] Del-Fabbro | DTM: a service for managing data persistency and data replication in network-enabled server environments[END_REF][START_REF] Desprez | DAGDA: Data Arrangement for the Grid and Distributed Applications[END_REF]), which focus on both user explicit data management and persistence of data across the resources, with transparent migrations and replications.

At a higher level, Stork [START_REF] Mclaren | Coordination of access to large-scale datasets in distributed environments[END_REF] is a batch scheduler specialized in data placement and data movement. If the transfer protocol specified in the job description file fails for some reason, Stork can automatically switch to any alternative protocol available between the same source and the destination hosts and complete the transfer; Galaxy5 is a web interface written in python allowing on-line design of task workflows. Galaxy focuses mainly on bioinformatics but could be used for all type of applications relying on workflow execution. By default Galaxy is configured to execute application on its host server but can use the OGF DRMAA API to distribute computations on remote servers. Data can only be transferred as files. On the contrary of classical RPC, there is no simple way to upload data directly on the application memory address space. Moreover, the GridRPC API modularity allows to combine simplicity of such data management systems and tunability by choosing where and when data are transfered.

By using standardized GridRPC code with our implementation and its corresponding modules, it should be possible to benefit at a upper layer from previous works, gaining in portability and interoperability with middleware and data managers, which in turn provides access to a potentially larger set of resources and architectures.

Implementation: architecture and features

We present in this section the system underlying our implementation of the GridRPC and GridRPC Data Management standards. We highlight the features of the library, its data management capabilities as well as scheduling possibilities between and for each data operation, i.e., the set of all transfers requested between the URIs provided as sources and destinations for the same data. The library is developed in C++ and C, using internally boost, and cmake to build the project. It is freely available from a sourceforge repository: http: //sourceforge.net/projects/gridrpcdm/. The proposed implementation can be viewed as a meta-implementation) of the APIs (see Figure 1) since it provides the two GridRPC APIs, adding some seamless mechanisms for performance (scheduling etc.) in a middleware and protocol "agnostic" manner. The library does not interact directly with the middleware nor the data storage servers. It proposes a fully interoperable API for any middleware and protocol/data manager with only very few specific developments of simple modules. The module developers do not have to take care about which data transfer protocol is available, like the data manager module developpers do not have to care about which middleware is used to call remote procedures. To do so, different interfaces are provided by the library:

Modularity of the solution

-The client application interface: the external client API. Clients can use the API directly without any knowledge about the underlying GridRPC middleware. However, by adding a prefix to the service name, users can force the library to use a specific middleware (e.g., "DIET:matmul" and "Ninf:matmul" select respectively Diet and Ninf-G for the "matmul" service). -The Services interface: it is a subset of the client API with some additional utility functions facilitating servers conversions from standard GridRPC servers to GridRPC Data Management servers.

-The Modules interface: the library defines a set of functions that should be exposed by the module to extend the library capacities.

Integrating a new middleware requires to fulfill a set of 10 main functions and one optional. Most of them are just type conversions functions from the existing middleware data-type to the new GridRPC data-types. The most complex function of a middleware module is grpc remote transfer() which initiates a transfer from a remote host to another remote host. A default implementation is included in the library relying on a middleware service call: the module developpers have just to implement this simple service using the library transfer capabilities on the server side.

Remarks:

-To avoid "name conflicts" between existing GridRPC implementations and the new definitions of the library, definitions in the library headers files are automatically prefixed when needed, allowing an easy reuse of the existing functions without name-clashing at the compilation step. -Note on asynchronous calls: they are internally managed by the library from synchronous calls to middleware. However, middleware functions must be reentrant for a safe asynchronous use.

At the moment, the modules for the Diet and Ninf GridRPC middleware are available.

Integrating a new data manager module requires to provide 4 functions: 2 initialization functions corresponding to input and output data, which can most of the time be left empty; and 2 transfer functions to get and put a data. They are generally wrappers of existing transfer protocol libraries (e.g., libcurl for http and ftp).

At the moment, the library implements the data manager modules for rsync and scp, using the shell commands; 2) iRODS, using the shell command (the library is only available for Java and PHP); webdav, to access Owncloud and Dropbox servers. It uses the neon library and; curl, to access http and ftp via the curl library.

Module initialization

The library global initialization process reads the global configuration file to determine which module should be dynamically loaded at execution time, where to find it, and some parameters available for each module in its own separate section. The initialization function of each module is then processed sequentially, passing the arguments of the module specific configuration, and potentially reading more parameters in the deployed module-specific configuration file.

Asynchronicity management

We call a request the inner action managed in the library: they correspond to API calls for remote procedure, API calls for a transfer or a group of transfers involving a unique data. For example when one source and several destinations are provided as input and ouput URIs, several transfers are involved in group to provide the unique API transfer call. All requests are managed the same way by the request controller: this entity registers each of them during their initialization, and with the help of threads and semaphores it limits their number and immediately knows the idendity of a request that completes without active wait. Some additional dependency information is also recorded with each request, and thus a hierarchy of requests (the link being the temporal dependency) can be built. It is used to express the concept of a group of requests reported above in the transfer example, but it is also a powerful way to handle waits for one or a group of asynchronous remote procedure calls as well.

Requests are managed with a priority system, which has been instantiated in the current implementation with a queue managed with a FIFO algorithm and a limitation on the number of parallel threads executed at a given moment: There is not much more that can be done at the moment: since there is no dependency information between data transfers operated at the API level, one cannot try any optimization between requests that do not belong to the same group because it could generate inconsistency in data or failure. However coupled with a system that handles workflow/dataflow, some meta-scheduling over available GridRPC middleware and data managers may be performed.

Data manager capabilities

The library does not only operate with underlying data transfer projects. It must provide the data persistence as defined in the API, integrate the possibility to communicate in-memory data (which possibly avoids at least one copy to disk), and make the junction between different locations where the data is available, and the protocols with which one can access them. The latter induces possible hidden (automatic and mandatory) copies and scheduling for the data to be transferred to all requested destinations.

Data persistence

The GridRPC Data Management API defines numerous persistence modes: the data can be volatile, i.e., there is no special requirement on its management and this can be considered as the default mode; it can be strictly volatile, meaning that the library has to provide means to remove the data from the platform after a computation (thus some protocols and data managers cannot be used); when defined as sticky, the data or a copy must be kept on the location where the client requests are executed; if unique sticky, no replication nor migration can be performed; finally, the client can also request the library to transparently manage prefetch, replication and migration of data. Then, by also handling procedure calls the library can perform some scheduling in order to reduce some metrics. At the moment persistent data are managed through Dagda .

The memory protocol Each data is referenced by a given set of specific URIs, providing the transfer protocol or the underlying data manager to use (for example http or dagda). But when trying to get performance, on linear algebra computation for example, there is a need to keep data in memory and avoid file transfers. The GridRPC Data Management API foresaw this kind of use and introduced the memory protocol. In addition to this protocol management, our implementation lets a client (or the library itself) use URIs with query and fragment. This leads to possible evolution for improvements (see after) and to manage more data managers (like P2P middleware that initiate torrents with specific files).

Implementing the memory protocol means that the library has to use GridRPC middleware inner data manager which can hopefully communicate between its own components to achieve such a need. But when a data is in memory and has to be transferred either on another GridRPC middleware components or on a storage server, it has to be written to a file and then be manipulated (transferred and possibly handled remotely) to be in the requested status. This part uses (de)serialization functions, defined by the GridRPC Data Management API, partly relying on tools provided by the boost library. But that maybe shows an unclear part of the API: the protocol to use in that specific case to manipulate the file is not precised. In our current implementation, the protocol is static and is read from the middleware configuration file at initialization time.

But if going a bit further than the API, we can use the query part of the URI. Indeed considering a data available in memory, the API does not provide a mean to know which protocol(s) can be used to send or to receive it since the URI would be similar to memory://graal.ens-lyon.fr/matrixA. In ongoing work, our library is going to explore what can be done with specifying protocols within the URI query part, e.g., ?protocol=rsync?protocol=webdav.

Scheduling for implicit and explicit data transfers Data transfers are operated 1) when data participate to a remote procedure call. They are in that case implicit or automatic, and; 2) can be explicitly requested by a client with a call to grpc_data_transfer().

Implicit and automatic transfers: When a remote procedure call is performed, meta-data are serialized and transferred to the distant service. They contain sets of URIs which may lead to additional transfers before and after the service execution (Fig 2).

-If one of the input URI refers to a memory or file data on the client, the data must be available to the service before its execution so that it can remotely access it. -If one of the ouput URI refers to a memory or file data on the client, the service must have made it available and the client must get it.

Explicit data transfers: Several transfers are operated by grpc data transfer(), i.e., a call to an explicit transfer operation: the data should be present in all locations set in the input URIs list, and must be present in all locations set in the output URIs list. This can be treated with a sequential set of transfers from one given source to each destination for example, but that would be inefficient.

In addition, it is not mandatory that all participants are directly interconnected (either by network or by protocol) and transfers may have to be scheduled to make the whole operations possible (destinations of completed transfers being considered as potential sources). The library also makes possible to delegate transfers on all GridRPC servers that offers some library specific service. Hence transfers can be distributed over nodes to reduce the bandwidth impact, and/or to try to reduce the transfer operation completion date for example.

In order to build a schedule in our implementation (made by the dispatcher, Fig 3), we list the nodes that can participate to a transfer operation: To our benefit, since the library contains middleware modules, it can also rely on underlying GridRPC middleware to potentially add relay servers to [remotely] distribute the transfer load or a part of it. To discover those middleware nodes, the library provides an echo service, that must be deployed, i.e., registered in the GridRPC server capabilities (at the moment, only middleware nodes with the memory protocol available are considered. If the service is not deployed, the node is simply not considered as a possible relay). The above algorithm loops until all transfers to ouput locations are done. In case of failure due to an unresponsive input middleware, the middleware is not considered anymore in the next scheduling/mapping cycle.

L-S: The transfer is initiated locally. M-S: The transfer is processed through a call to the remote transfer service. S-L: The transfer is initiated locally. S-M: The transfer is performed through a call to the distant transfer service. L-L: The transfer is initiated and performed locally. L-M: The library makes the local data available via the GridRPC middleware inner data manager whose remote counter part will download afterwards. M-L: The remote middleware is being asked to make the data available, this data is then downloaded by the GridRPC middleware. M-M: The source middleware is asked to make the data available so that the destination middleware can download it when needed. S-S: The library first tries to invoke a remote service on the destination server to initiate the transfer. If the transfer fails, data is downloaded on the library client, then transferred to the destination server. If there is no available protocol to proceed to such transfers, the call fails returning an error code.

4 Experimental results

Multi-protocol and dispatcher scheduling/mapping validation

Table 1 lists the experiment deployment. We used 3 computing resources, 2 in Japan and 1 in France, on which we deployed iRODS and ssh servers, and Diet components: a client, a dietAgent (the registry), and a server (matrix addition) written with the GridRPC APIs requirements together with our library. Two matrices are defined with a list of input URIs depending on the running test, described hereafter.

Machine (location) Services Data (protocol) Arcterix (JFLI -Japon) dietAgent, client, sshd matA (ssh) yume (JFLI -Japon) service '+', iRODS, sshd matA, matB (ssh) graal ( Éns-Lyon -France) sshd matB (ssh) Table 1

. Resources involved in Experiment 1

There are four tests, built with the scenario of getting the two matrices through ssh, performing the addition, and uploading the result to an iRODS server (here locally):

-Remote/Remote: the client does not include the URIs concerning the host yume in the input list used for the remote call.

are present on the server side (the addition was chosen for the operation since it requires less time than a transfer, which leads to show the wanted behavior. Besides, it also makes sense conceptually since it's a commutative operation).

Figure 5 shows the activity of the service and its duration on the y-axis, related to the progression over the number of matrices downloaded for each limit on the x-axis. The same evolution by group of cardinal equal to the possible limit of both the waiting time (non-active wait) and transfer time highlights that the number of simultaneous transfers operated by the library is indeed configurable (for the moment the information is static in the configuration file. We intend to look if it makes sens to have it self-tuned by the library, depending on the dynamicity of both the network and computing performance). It also shows that every computations occur when enough matrices are finished to be downloaded. As a side effect, it also confirms the observations made in [START_REF] Mclaren | Coordination of access to large-scale datasets in distributed environments[END_REF]: there's a real need to limit the number of possible parallel transfers. We can indeed observe on this small example that the overall completion time of the addition of the 16 matrices is a bit reduced when the limit is fixed to 2 for our small testbed.

Future works

Future works are heading towards different directions. If the library is already usable and implements most of the API, more performance can be obtained with more efficient scheduling: at the request controller (Section 3.2), and at the dispatcher level (Section 3.3); and more development: for example including a middleware module for ssh would add more scheduling possibilities; the protocol memory leads to already complex data management mechanisms, yet to be continued in addition to a file protocol that would help avoiding useless data copies, making the use of the library even more scalable. Modules for dCache and GridFTP would possibly make transfers faster, but further control would have to be done on the bandwidth consumption; a data manager module for Amazon S36 would give further access to cloud storage resources leading for a need to also take into account some financial criteria in the above scheduling process, and possible migration of data when possible (e.g., when the data is requested as GRPC PERSISTENT).

Conclusion

With the GridRPC Data Management standard completing previous works on GridRPC, both at the API and software level, feasibility of computations and performance is at reach with immediate portability and interoperability between GridRPC middleware. To ease its spread, while giving access to GridRPC middleware and to existing data managers, we provide an implementation of both APIs relying on a very modular architecture. Fulfilling the standard requirements, the library also implements the data management modes as well as a memory protocol to avoid useless copy to disk. We showed that an efficient system to handle waiting mechanisms is in place and that we operate some mapping/scheduling when several transfers are involved in the same data management. We conducted some experiments and obtained results validating the expected behaviors. From now on we will focus on more theoretical work to improve the yet non-trivial mapping/scheduling of transfers involved for a given data, and we are considering to plug a workflow/dataflow analyzing tool to schedule transfers of different data with remote procedure calls altogether. Further developments will also occur, giving more adaptability and choices to the end-user while bringing new issues concerning scheduling possibilities, for example with Cloud storage resources.

Fig. 1 .

 1 Fig. 1. A very modular implementation

Fig. 2 .Fig. 3 .

 23 Fig. 2. Automatic transfers during a GridRPC call

http://www.gluster.org/

http://galaxyproject.org/

http://aws.amazon.com/

Acknowledgment: This work is partially founded by the ÉNS Lyon. The authors want to thank Hidemoto Nakada for the Ninf middleware module.

This simple experiment aims to show both 1) the seamless multi-protocol management of the library, as well as 2) the possibility for the dispatcher, described page 9, to perform a schedule: due to its priority matching combined with its Round-Robin algorithm, the library uses the local data first if available. Figure 4 clearly shows this behavior, the blue region showing the time spent during each transfer when it occurs.

Asynchronous transfers management and bounded number of simultaneous transfers

For this experiment, we designed the following scenario: a remote procedure call is performed to add a given number of matrices which are available remotely through ssh. Matrices are downloaded and added as soon as the operation is possible, i.e., at first when two of them are finished to be downloaded, then every time a new one has been downloaded and the previous computation has finished. We performed 4 tests, corresponding to the number of simultaneous transfers that it is possible to make at a given instant, resp. 1 unique transfer, 2, 4 and 8 transfers maximum at a given time. The number of matrices is fixed to 16, and one matrix is 32MB (i.e., 2000 × 2000 of 64bits integers). We designed this experiment to validate the request controller behavior, i.e., the implementation of the possibility to limit the number of simultaneous transfers occurring in a transfer operation, and the possibility to use the waiting functions, here with grpc_data_transfer() and the GRPC_WAIT_ANY parameter, making the server able to perform an operation as soon as enough matrices