Mathieu Pasquet
email: mathieu.pasquet@ens-rennes.fr

Francisco Maia
email: fmaia@di.uminho.pt

Etienne Rivière

Valerio Schiavoni

Autonomous Multi-dimensional Slicing for Large-Scale Distributed Systems

Keywords: Slicing, Self-organization, Skyline, Gossip-based protocols

des établissements d'enseignement et de recherche français ou étrangers, des laboratoires publics ou privés.

Introduction

Very large-scale heterogeneous distributed systems will be commonplace with the advent of connected objects, the internet-of-things and in general with the massive increase in the number of machines connected through multiple networks. Such very large systems can no longer be operated using the system services and support mechanisms that were designed and implemented for moderate to small scale distributed systems. In particular, system-level primitives and services, that were once supported by centralized entities or by the close collaboration of a few synchronized nodes, may not be adequate or even possible to use in these scenarios. The characteristics of large-scale heterogeneous systems call instead for fully decentralized and autonomous protocols. Such protocols can provide systemwide fundamental system services but do not require centralized components or complex synchronization between nodes. Self-organization is another key requirement. It denotes the ability of a system or service to seamlessly adapt to Work done while interning at Universit de Neuchâtel.

dynamic conditions, such as nodes heterogeneity, dynamic membership, or faults. It allows always ending up in a regular operating state, regardless of the starting conditions or perturbations imposed to the system.

Self-organizing, autonomous distributed protocols and services are typically implemented using the gossip-based paradigm. Gossip-based protocols rely on periodic exchange of information between pairs of peers and appropriate nodelocal decisions. They allow implementing a variety of system services based on overlay networks, such as membership management [START_REF] Jelasity | Gossip-based peer sampling[END_REF], dissemination [START_REF] Eugster | Lightweight probabilistic broadcast[END_REF][START_REF] Matos | efficient, robust epidemic dissemination[END_REF], structure and overlays management [START_REF] Jelasity | Gossip-based fast overlay topology construction[END_REF][START_REF] Voulgaris | VICINITY: A pinch of randomness brings out the structure[END_REF], or key-based routing [START_REF] Maia | Dataflasks: an epidemic dependable key-value substrate[END_REF][START_REF] Montresor | Chord on demand[END_REF].

An important class of services for large-scale heterogeneous distributed systems is resource provisioning. A system might need to support multiple services and applications, each with different requirements in terms of computational capabilities, network requirements, or criticality (i.e., the need for the service to be dependable and available). Resource provisioning allows determining, for each service or application, which resources should be used. In particular, when services are implemented through distributed protocols running on a subset of all the nodes of the system, provisioning can be generalized to the action of splitting the entire set of nodes in subsets, one for each of the service or application.

For instance, in a system that supports indexing and dependable storage atop highly volatile nodes [START_REF] Maia | Dataflasks: an epidemic dependable key-value substrate[END_REF], a small set of nodes -typically the most stable onescan be dedicated to the routing service that implements the indexing layer while the remaining ones handle best effort replication of data items. The selection of the most stable nodes can be based on their uptime. As demonstrated in [START_REF] Bhagwan | Understanding availability[END_REF] based on measurements on a real large peer-to-peer system, uptime is correlated with stability. Nodes with large uptimes tend to stay online for a longer time than nodes with a small uptime, regardless of their other characteristics. Another example is that of a system that supports a BitTorrent-like dissemination and download service. The subset of the most stable node could be used for supporting a self-organizing distributed hash table (DHT) [START_REF] Jelasity | Gossip-based fast overlay topology construction[END_REF][START_REF] Montresor | Chord on demand[END_REF][START_REF] Voulgaris | VICINITY: A pinch of randomness brings out the structure[END_REF], nodes with the largest available bandwidth could be used as helpers for others to download faster [START_REF] Garbacki | 2Fast: Collaborative Downloads in P2P Networks[END_REF], while the remaining ones only act as clients.

Distributed Slicing

The prominent implementation of resource provisioning for very large scale systems is distributed slicing [START_REF] Fernandez | Distributed slicing in dynamic systems[END_REF][START_REF] Gramoli | Slicing distributed systems[END_REF][START_REF] Jelasity | Ordered slicing of very large-scale overlay networks[END_REF][START_REF] Maia | Slead: low-memory, steady distributed systems slicing[END_REF]. It allows splitting a large set of nodes, based on discrete representations of their characteristics, in the form of node-local attributes. Slicing allows forming groups of nodes with increasing attribute values. A slicing schema, known by all nodes, indicate the relative size of each of the slices with respect to the total size of the system. It is important to note that the actual total size of the system does not need to be known by the nodes themselves. As an example, say that the considered characteristic of nodes is their available disk space, as exemplified on the top of Figure 1. Nodes all know the slicing schema S = {50%, 30%, 20%}. A slicing protocol must create three slices of size 50%, 30% and 20% of the total size of the system, e.g., 5,000, 3,000 and 2,000 nodes for a network of 10,000 nodes, and 7, 4 and 3 nodes for the small network of Figure 1. Slices correspond to group of nodes of increasing attribute values. Nodes in the first slice all have attributes that are lower than those of the second and third slices, and similarly between the second and the third slice. Slicing neither requires a centralized view of the system, nor it requires assumptions about the distribution of attribute values. Each node must estimate the position its attribute would have in the sorted list of all attributes, if such a list was materialized. This position is typically obtained by estimating the proportion of nodes in the system that have lower values for the attribute and the proportion of nodes that, instead, have higher values for the attribute. When this information is available, it is then straightforward for a node, based on the slicing schema, to determine the slice it belongs to. For instance, node 10 is able to determine that it belongs to the second slice of the schema knowing that there are 9 nodes with lower values, and 4 nodes with higher values. Several variants of protocols allow to implement this basic principle. We give further details of one of these protocols, sort ranking, in Section 2.

Problem definition

The major drawback of slicing for node provisioning in large-scale network is that it intrinsically only supports a single dimension for attributes. It is possible to form slices based on the available disk space, or the network connectivity, or CPU power, but considering a single of these characteristics at a time only. It is often desirable to be able to provision nodes based on a combination of metrics, in order to form slices that represent classes of lower to better compromises for a set of characteristics. This would allow for instance to provision a largescale network with a set of super-nodes with the best compromise in terms of available bandwidth and processing capacity, for serving as storage nodes or for participating to a backbone indexing overlay, and use nodes with less interesting characteristics for less demanding tasks, such as dissemination [START_REF] Eugster | Lightweight probabilistic broadcast[END_REF] or monitoring [START_REF] Jelasity | Gossip-based aggregation in large dynamic networks[END_REF]. Other examples of compromises involve the node stability as measured by their uptime and processing capacity, in order to form a support overlay composed only of the peers that have both a good capacity, and whose stability ensures good resilience of the structure.

We propose and evaluate in this paper a novel approach to support slicing based on multiple attributes. Our main objective is to form slices that group nodes in classes of quality of the compromises between their different characteristics. This should not require knowing the range of definition for any of the dimensions, which can be practically unbounded. Multi-dimensional slicing should also be independent of the distribution of values for each dimension. Figure 2 illustrates the principle of the considered problem from a high-level perspective. It presents a set of nodes with attributes along two dimensions, the available disk space and the uptime. The latter links to expected stability [START_REF] Bhagwan | Understanding availability[END_REF]. Nodes form three slices under the same schema as for Figure 1 over a single dimension, S = {50%, 30%, 20%}. These slices should represent classes of increasingly better compromises.

We see for instance that node 8 presents both a high uptime and a large amount of available disk space. It is thus expected to be stable, and has enough capacity. On the other hand, node 12 has moderate available disk space but is the most ancient node, so supposedly the stablest. Taking into account this heterogeneity, the first slice will include nodes such as node 8 or 12 that exhibit the best attribute compromises. Alongside, nodes in the second slice shall represent intermediate compromises among the set of available nodes, while nodes with either small storage space for their stability or small stability for their storage space, are left to the first slice.

Our contribution is based on the determination of domination relations between nodes. This notion has been applied to data points in the context of databases for the definition of the SkyLine query [START_REF] Hose | A survey of skyline processing in highly distributed environments[END_REF]. The domination relation between nodes allow comparing compromises between nodes characteristics, and allows nodes to estimate their position in a global domination partial order that would contain all nodes if materialized. This position is used, as for the uni-dimensional slicing, to determine the slice a node belongs to.

The remaining of this paper is organized as follows. In Section 2, we present the principle of a representative slicing algorithm for single-dimension attributes, which serves as the basis for the multi-dimensional slicing presented in Section 3. We present an evaluation of our contribution in Section 4. Related work is surveyed in Section 5. Section 6 concludes the paper.

Background

We present in this Section the principles of gossip-based protocols and the algorithm for single-dimension slicing that we extend for supporting multi-dimensional slicing. This algorithm, which we name sort slicing is based on a self-organizing gossip-based distributed sort and was initially presented in [START_REF] Jelasity | Ordered slicing of very large-scale overlay networks[END_REF].

Gossip-based distributed systems allow implementing autonomous and selforganizing protocols. An early gossip-based protocol was proposed for the reconciliation of database replicas in the 1980s [START_REF] Demers | Epidemic algorithms for replicated database maintenance[END_REF]. Gossip came back into fashion in the 2000s as the complexity of using centralized or rigid management became clear with the increase of systems scales. A typical use of gossip is reliable applicationlevel broadcast [START_REF] Eugster | Lightweight probabilistic broadcast[END_REF]. Another class deal with overlay management [START_REF] Jelasity | Gossip-based fast overlay topology construction[END_REF][START_REF] Jelasity | Gossip-based peer sampling[END_REF][START_REF] Montresor | Chord on demand[END_REF][START_REF] Montresor | Decentralized ranking in large-scale overlay networks[END_REF][START_REF] Voulgaris | VICINITY: A pinch of randomness brings out the structure[END_REF]. Slicing protocols [START_REF] Fernandez | Distributed slicing in dynamic systems[END_REF][START_REF] Gramoli | Slicing distributed systems[END_REF][START_REF] Jelasity | Ordered slicing of very large-scale overlay networks[END_REF][START_REF] Maia | Slead: low-memory, steady distributed systems slicing[END_REF] are a subset of it.

In an overlay management gossip-based protocol, each node maintains a small set of links to other nodes. This small set of nodes is called the node's view. It is often of fixed size. Each node features an active and a passive thread. The active thread periodically initiates a gossip interaction between the node and another node from the system. The period between two consecutive interactions is defined as a cycle. The passive thread is in charge of replying to incoming interaction requests. The interaction typically involves the exchange of information between the two nodes. It results in the two nodes potentially changing their state, or taking some action based on that exchanged information. Gossip-based protocols solely rely on this simple interaction/exchange pattern and do not require further synchronization between nodes. The nature of the information exchanged and the action taken define the nature and goal of a gossip-based protocol.

The partner node for the interaction is either randomly picked from the view, or obtained from a Peer Sampling Service (PSS) [START_REF] Jelasity | Gossip-based peer sampling[END_REF][START_REF] Voulgaris | Inexpensive membership management for unstructured P2P overlays[END_REF] a fundamental service for other gossip-based protocols. It is also itself using a gossip-based implementation. It manages nodes' membership to the system and maintains connectivity. It provides other gossip protocols with a constantly updated stream of random peers drawn from the entire system, forming an overlay that has similar properties to a random graph. It inserts new nodes in the stream of other nodes and ensures that removed nodes do not appear in other nodes' streams after some bounded time. The use of a PSS is mandatory for all gossip-based overlay management protocols to provide convergence and associated self-organizing guarantees [START_REF] Voulgaris | VICINITY: A pinch of randomness brings out the structure[END_REF]: this also applies to gossip-based slicing protocols.

All gossip-based slicing algorithms share the following characteristics. Each node possesses an attribute value a, over a single dimension, that represents one of its characteristics (e.g., the available disk space in Figure 1). The domain of a is unknown and unbounded. It also knows the slicing schema S, a system-wide parameter. Finally, each node's goal is to obtain a position estimation e, that indicates its estimated position in the sorted list of all attribute values, if such a list was materialized. The value of e lies in the bounded domain [0, 1]. Knowing S and e, it is straightforward to determine which slice the node belongs to. For instance, with e = 0.55 and S = {50%, 30%, 20%}, node 10 in Figure 1 can determine it belongs to the second slice.

The gossip-based sort slicing [START_REF] Jelasity | Ordered slicing of very large-scale overlay networks[END_REF] algorithm determines e based on the following method, detailed in Algorithm 1 for a given node p. The view is composed of nodes that had numerically close values of e to the current node, at the last time an interaction took place. Interaction partners are selected randomly from nodes in the union of the view and the current stream of nodes exposed by the PSS. The use of the view is only meant to speed-up the convergence, while the use of the PSS is required for that convergence to happen [START_REF] Voulgaris | VICINITY: A pinch of randomness brings out the structure[END_REF]. Each node is bootstrapped with a random value for e in [0, 1]. The goal of the gossip interactions is to match the ordering of e values on all nodes, to the ordering of a values on all nodes. It implements a distributed swap-sort for this. At each gossip interaction, the two interacting nodes p and q compare the ordering of their values of e and the ordering of their values of a. If the two orders are the same (i.e., if (a p < a q ∧ e p < e q ) ∨ (a p > a q ∧ e p > e q )), no action is taken. Otherwise, there is an order violation that needs to be corrected. Nodes proceed to this correction by simply exchanging their position estimations e p and e q . The authors of [START_REF] Jelasity | Ordered slicing of very large-scale overlay networks[END_REF] show that this simple method yields an exponential reduction of the disorder metric (the sum of the squared distances of position estimation from their final value) with the number of cycles. This translates in a rapid convergence towards the correct slice attribution at each node.

Multi-Dimensional Slicing

We present in this Section our contributions towards autonomous slicing for large-scale distributed systems based on attributes over multiple dimensions. We start by noting that using a utility function does not fit the requirements we have set for multi-dimensional slicing. A utility function maps multiple dimensions onto a single one by applying a linear combination of the values over these dimensions. This requires knowing or estimating the boundaries for each dimension. It also requires assigning weights to metrics (the values over each dimension) that are not comparable. It finally requires making assumptions on the distribution of values over each dimension, as skewed distributions for the value of one attribute would result in the order for other attributes dominating the selection process. We thus dismiss this naive option and consider instead solutions that preserve the multi-dimensional aspect of the workload. First, we report on a negative result on our initial use of space-filling curves. Then, we present our approach using domination relations between multi-dimensional nodes' attributes, the resulting partial order and order violation resolutions, and the resulting multi-dimensional slicing.

Using space filling curves? Our first attempt to provide multi-dimensional slicing was to map attributes of multiple dimensions onto a single dimension using space filling curves (SFC). A SFC is a curve that covers the entire unit square. It can be generalized to a d-dimensional space. Each d-dimensional point in the space is associated with a single point on the curve. The distance on the curve of the representations of two d-dimensional points can be used to estimate the distance between the two original d-dimensional points in the space. SFC are typically defined by a recursive definition, as a special case of fractal construction. There exists many SFCs but we chose to use the Hilbert SFC (illustrated for 2 dimensions on Figure 3). This curve has the property that it tends to preserve ordering for a large number, but not all, possible couple of points mapped on the curve with respect to the ordering of their corresponding vector to the origin in the space. This initial approach proved unsatisfactory for the following two reasons. First, a major issue with SFC is that they require knowing in advance -or estimating-the boundaries of the domain space. This condition is not met by typical workloads for resource provisioning in large-scale systems. Second, while the use of a SFC allows forming a total order between multi-dimensional points, the ordering inconsistencies (e.g., between [2:1], [3:3] and [START_REF] Alvarez | An automated resource provisioning tool for large-scale storage systems[END_REF][START_REF] Alvarez | An automated resource provisioning tool for large-scale storage systems[END_REF] on Figure 3) prevent the algorithm from converging towards stable position estimations. As a result, we discarded the SFC approach and looked into a solution that intrinsically does not feature these two limitations, and does not require mapping multiple dimensions into a single one.

Domination Relations

Our approach for multi-dimensional slicing is based on the intuitive notion of quality of compromises that we presented in our problem definition. It does not require knowing the range of definition of any of the dimensions. It is based on a domination relation . We say that a node n a dominates another node n b when n a is strictly a better compromise than n b according to the value of their attributes over all dimensions. The definition of is as follows, where n i a denotes the value of the attribute of n a for dimension i and where d is the number of dimensions: and in particular for the definition of SkyLine queries [START_REF] Hose | A survey of skyline processing in highly distributed environments[END_REF]. A SkyLine query allows collecting the best-compromise points from a structured multi-dimensional database: the SkyLine points are the points that are not dominated by any other point, under the same definition as above. The SkyLine of Figure 4 would be {n 7 , n 11 , n 2 }. The SkyLine operator has been extended towards the SkyBand operator [START_REF] Vlachou | Skylinebased peer-to-peer top-k query processing[END_REF]. A k-SkyBand is a set of points that are not dominated by more than k points. The 2-SkyBand of Figure 4 would be {n 7 , n 11 , n 2 , n 6 , n 9 , n 10 , n 3 }.

n a n b if ∀ i | 1 ≤ i ≤ d, n i a ≤ n i b . This relation is transitive: n a n b ∧ n b n c ⇒ n a n c and its inverse is : n a n b ⇔ n b n a .
The slicing problem is however different than the k-SkyBand in several ways. First, there is no omniscient view of the data set as in a database. Nodes only know a small amount of other nodes in their view and have no global knowledge of the boundaries of the space, the number of nodes, or the distribution of values for each dimension. Second, the slices are not defined in terms of the number of dominating nodes but in terms of a size relative to the totality of the system. We detail below how we modify the uni-dimensional slicing from Algorithm 1 to support multi-dimensional slicing based on domination relations.

Multi-dimensional Slicing from Domination-based Ordering

The multi-dimensional slicing algorithm is presented by Algorithm 2. It extends the previous Algorithm 1 for slicing over one dimension. An illustration of the algorithm is provided by Figure 5. Description of the illustrated algorithm steps is directly provided below the figure itself. Similarly as before, each node p has access to a PSS and maintains a view. This view of p is split in two parts: the -view and the -view. The former holds the peers that are directly dominated by p and by no other node in -view. The latter has the same semantics for dominating peers under the relation. Each time a gossip exchange takes place, if the domination relation holds between p and the exchange partner q, q is first added to -view (line 10). The -view is then pruned out of nodes that are already dominated by another one from the set (lines 11 to 13). The same principle is applied to dominating nodes and the -view (lines 15 to 19). When an order violation is detected (line 14 or line 20), the position estimation e p and e q are exchanged, in the same way as for Algorithm 1. The slice determination itself is strictly identical as for Algorithm 1. Algorithm 2 can be enhanced in a number of ways. First, instead of keeping the directly dominated nodes and directly dominating nodes in -view andview, we can keep a set of nodes to ensure a faster resolution of "local" order violations ("global" order violations between nodes are expected to be resolved more frequently due to the interactions from the PSS, but close links help in the final steps of convergence as demonstrated in [START_REF] Voulgaris | VICINITY: A pinch of randomness brings out the structure[END_REF]). To this end, node p might keep a set of peers that is not actually the Skyline of the dominated (resp. dominating) nodes but an overset of it, nodes are removed from the -view and -view only when it goes over a target view size (by modifying the conditions on lines 12 and 18). Second, the interaction depicted is single-way: while both p and q participate in the exchange, only p takes advantage of it to fill its views as part of its passive thread (this is known as a push-only operation). It is straightforward for q to operate in the same manner while on its active thread (push-pull). We use these two optimizations in our implementation.

We note that several of the interactions with random nodes chosen using the PSS will lead to no action as the two values will not be comparable in the partial order O . 4 These random interactions are necessary however when the attributes value change over time, as expected in a dynamic provisioning scenario (e.g., the uptime, available disk space, etc. evolve over time). Unlike with uni-dimensional slicing, no total order on the nodes exists. This means that there are a multitude of possible total ordering of the position estimates e p that are compatible with the partial ordering O . As we point out in our evaluation, this also means that the scheme is unsurprisingly subject to the curse of dimensionality [START_REF] Bellman | Curse of dimensionality[END_REF]. When using over 8 or 10 dimensions, the partial order tends to become a very sparse graph as most nodes are not comparable against O anymore. We believe however that the problem of dynamic node provisioning in large-scale systems does not require more than this limit number of dimensions to fulfill its intended role.

Evaluation

We now present the evaluation of our protocol. Our implementation uses Peer-Sim [1], a Java-based cycle-oriented simulation framework that is well adapted to the study of gossip-based protocols. We simulate a network of 10,000 nodes for all the presented experiments. Each node uses an attribute over 2 to 15 dimensions (when not specified, we present results for 2 dimensions). Values for each dimension are random float numbers. Each node has access to an independent PSS. We use an instantiation of the Cyclon [START_REF] Voulgaris | Inexpensive membership management for unstructured P2P overlays[END_REF] PSS in the framework of [START_REF] Jelasity | Gossip-based peer sampling[END_REF]. Each node maintains a -view and a -view of up to 10 entries each.

The reminder of this section is organized as follows. We start by evaluating the protocol correctness. Then, we evaluate the convergence speed with different view configurations. Finally we study the scalability in terms of number of dimensions.

Correctness. We evaluate the correctness of the protocol by its capacity to reduce the number of order violations as defined in Section 3.1. An order violation between two nodes p and q with position estimates e p and e q occurs when ((p q ∧ e p > e q ) ∨ (p q ∧ e p < e q )). We count the number of such violations per node when compared in an omniscient way against all other nodes in the system. Note that we cannot use a disorder metric as in [START_REF] Jelasity | Ordered slicing of very large-scale overlay networks[END_REF] as the partial order between nodes is not based on an Euclidean distance but on domination relationships. Figure 6 depicts the evolution of the average and maximum number of order violations for nodes in the system, as nodes engage in cycles of gossip exchanges (during one cycle, each node initiates one gossip exchange with one partner). As expected from previous work on gossip-based overlay construction protocols [8, 11, 15-17, 20, 21, 24] and similarly to the uni-dimensional slicing in [START_REF] Jelasity | Ordered slicing of very large-scale overlay networks[END_REF], the reduction in the number of violations is exponential. After only 10 cycles, the system converges to less than 0.05% violations per node on average and 3.7% as worst case. This indicates that the node provisioning quality will highly resemble the "ideal" slicing for the given attributes values set. We note that the max converges slowly to 0: this is a result of remaining violations being slowly resolved by using random PSS links in a network where a majority of nodes have converged views. This behavior was pointed out in [START_REF] Voulgaris | VICINITY: A pinch of randomness brings out the structure[END_REF]. It does not pose a significant problem in practice as it has little influence on the slicing decision itself, which is already converged as we see in our last experiment.

Influence of the gossip interaction partner selection. Figure 7 illustrates the influence of the selection of the gossip interaction partner (line 6 in Algorithm 2) on the convergence speed and correctness. We use the same metric as before and present the mean number of violations for the three following cases. The gossip exchanges are initiated by randomly picking a node from the view only (configured by a static bootstrapping using random peers), using nodes obtained from the PSS only, or using a combination of both. As expected, we observe that the convergence requires the use of the PSS. Using only the views bootstrapped with initial random nodes but not constantly updated with new random nodes, the system ends up in a state where nodes with order violations lie in disconnected islands and never get to resolve these violations anymore, preventing the whole system from converging. We also observe that the use of the view for choosing the gossip partner has a slight, but almost negligible impact on the convergence speed. These two observations are in line with the observations made for other gossip-based overlay construction protocols [START_REF] Jelasity | Gossip-based fast overlay topology construction[END_REF][START_REF] Montresor | Chord on demand[END_REF][START_REF] Voulgaris | VICINITY: A pinch of randomness brings out the structure[END_REF].

Scalability in number of dimensions. Our next series of experiments evaluate the impact of the number of dimensions used for nodes' attributes on the stability and observed convergence speed. Figure 8 reports the evolution of the number of order violations when using 2, 4, 6 and 15 dimensions. More dimensions mean less possible comparisons between nodes against . The convergence speed remains good with up to 6 dimensions. As expected, when using a large number of dimensions (in general, starting from 8), an embarrassing majority of couples of nodes are not comparable with . This means that the slicing operation using domination-based order is no longer possible, unless one restricts the operation to a subset of representative dimensions. We believe 6 dimensions is enough characteristics to consider for dynamic node provisioning in large-scale systems.

Finally, we evaluate the impact of the number of dimensions on the stability of the system, in terms of the steadiness of the slice allocation for each node. Figure 9 presents the number of slice changes per cycle (note the logarithmic scale) as the protocol converges. The cost of a slice depends on the supported application. In general, a good practice is to export the slice allocation only when it has been steady for a configurable number of cycles. With 15 dimensions, nodes are basically stuck with the slice that corresponds to their initial random value for the position estimation p, as there are almost no order violations to resolve. Along with the previous observation, this exemplifies the expected effect of the curse of dimensionality [START_REF] Bellman | Curse of dimensionality[END_REF] on multi-dimensional slicing. With 1 to 6 dimensions though, we see that the system quickly converges to stable slice allocations. The protocol starts its execution after the 2nd cycle, where 40% of the nodes decide to change their slide. After about 10 cycles, only 100 slice changes per cycle happen. This concerns a mere 1% of the network. After 30 cycles, there are only 10 such changes (0.1%). In a practical deployment, one can freeze the slice allocation around these times depending on the required precision.

Related Work

Besides the sort slicing protocol that we described in Section 2, the following slicing protocols were proposed. Ranking [START_REF] Fernandez | Distributed slicing in dynamic systems[END_REF] is not based on the exchange of position estimates but on counting mechanisms that estimate how many nodes have lower, resp. higher values for the single-dimension attribute. Sliver [START_REF] Gramoli | Slicing distributed systems[END_REF] improves this protocol by introducing a technique to improve slicing accuracy avoiding mistakenly considering duplicate information. Slead [START_REF] Maia | Slead: low-memory, steady distributed systems slicing[END_REF] is an extension of this idea that make use of Bloom filters to reduce the high memory consumption of these protocols and proposes a hysteresis mechanisms for the slice determination to gain more stability. All these protocols are based on the gossip-based paradigm, for which we covered related work in Section 2.

We review below alternative proposals to slicing for large-scale system resource allocation. Minerva [START_REF] Alvarez | An automated resource provisioning tool for large-scale storage systems[END_REF] contributes tools to efficiently assign host machines to support large-scale storage systems. The assignment of the hosts is centrally executed by an allocator node. Our approach relies on autonomous decisions taken by each participants of the protocol, without any central coordinator. The autonomous resource selection presented in [START_REF] Costa | Autonomous resource selection for decentralized utility computing[END_REF] organizes nodes in a peer-to-peer system over a multi-dimensional space, upon which lookup queries are issued to select nodes matching the specified requirements. Similarly to ours, this protocol has built-in support for multi-dimensional attributes. This system solves a related but slightly different problem than the one addressed in this paper, that is the selection of a fixed number of nodes that satisfy the specified queries, and does not rely on self-organizing techniques but explicit query/response mechanisms.

In [START_REF] Ghodsi | Dominant resource fairness: Fair allocation of multiple resource types[END_REF], a notion of resource dominance is also used. It differs however from the domination between nodes we use in our work. Domination in [START_REF] Ghodsi | Dominant resource fairness: Fair allocation of multiple resource types[END_REF] relates to the prevalence of the usage of one type of resource (CPU, memory, . . . ) over the other types of resources on one node. This information is used by a centralized resource allocator in order to achieve resource allocation fairness for a given workload.

Conclusion

Distributed slicing is a fundamental primitive for resource allocation in large-scale distributed systems. Existing distributed slicing protocols did not offer support to perform this allocation considering nodes with multidimensional attributes. In practical settings, it is often desired to choose a best compromise between various characteristics (i.e. CPU, volatile memory, uptime, etc.). We contribute a gossip-based distributed slicing protocol that fills this need. The proposed protocol exploits domination relations between nodes to define the membership of a node to a given slice. Our evaluation confirms the contributed solution as a sound approach to the problem. We support experimental reproducibility: source code and datasets are released under open-source and available at: https://github.com/mpasquet/multidimslicing_dais2014.

Fig. 2 .

 2 Fig. 2. Principle of the proposed multi-dimensional slicing with two dimensions.

Fig. 3 .

 3 Fig. 3. 2-D Hilbert SFC.

Fig. 4 .

 4 Fig. 4. Domination relations and partial order.

Figure

  Figure 4 (left) gives an example of domination relations between a node n 10 and a set of other nodes, over a 2-dimensional space. Relations that can be determined based on transitivity are not shown. In particular, we have n 10 {n 1 , n 4 , n 5 , n 8 , n 12 } and n 10 {n 7 }. The relations define a partial order O between nodes. The partial order for nodes n 1 , . . . , n 12 is shown on the right side of Figure4: we note that some pairs of nodes have no defined ordering according to and . For instance, n 11 and n 7 have no established or relation. We note that our notion of domination between nodes is similar to the notion of domination between data points that is used in the context of databases,

Fig. 5 .

 5 Fig. 5. Evolution of the view of a node according to Algorithm 2.

Fig. 6 .Fig. 7 .

 67 Fig. 6. Convergence speed.

Fig. 8 .Fig. 9 .

 89 Fig. 8. Convergence vs dimension.

slice 1 (50%) slice 2 (30%) slice 3 (20%)

	node 1 1.4 GB 18 hours		node 2 850 MB 2 hours	node 3 2.4 GB 5.5 hours	node 4 500 MB 1 day	node 5 2.6 GB 4 hours	node 6 900 MB 10 hours	node 7 2.2 GB 1 hour	node 14 2 GB 15 hours	node 12 600 MB
	node 8 1.8 GB 1.2 day		node 9 450 MB 6 hours	node 10 1 GB 20 hours	node 11 200 MB 9 hours	node 12 600 MB 1.5 days	node 13 1.5 GB 8 hours	15 hours 3 GB node 14	1.2 day 1.8 GB node 8	1.5 days
							complete network			node 3	node 5
	available disk space	7	2	5	3	13	6	14	1	10	8		distributed slicing	9 hours 200 MB 2.6 GB 4 hours node 1 1.1 GB 13 hours node 10 10 hours 900 MB 3 GB 5 hours 1 GB 20 hours node 7 2.2 GB 1 hour node 2 850 MB 2 hours node 13 1.5 GB 8 hours node 9 450 MB 6 hours node 11 node 6
					9						4		12	node 4 500 MB
						11								1 day
							uptime (stability)

  . The PSS is Algorithm 1: Simplified Sorting-Slicing Algorithm on node p[START_REF] Jelasity | Ordered slicing of very large-scale overlay networks[END_REF].

	1 variables
	2	ap ← . . .		// local attribute
	3	ep ← rand(0,1)	// local position estimation
	4	S ← . . .	// slice schema, array [1..N ]: N i=1 S[i] = 1
	5 every ∆ time units do		// active thread
	6	select partner q randomly from view ∪ PSS.getView()
	7	ep ← q.receive(ap, ep)		// initiate exchange
	8 function receive(aq, eq)		// passive thread
	9	if (ap < aq ∧ ep > eq) ∨ (ap > aq ∧ ep < eq) then	// order violation?
	10	t ← ep	// save current position estimation
	11	ep ← eq	// exchange current position estimation
	12	return t
	13	else	// no order violation, no exchange
	14	return eq
	15 function getSlice()

16 t ← 0, r ← 1 17 while t < ep do // calculate slice estimation 18 t ← t + S[r], r ← r + 1 19 return r

  In particular, we have n 10 {n 1 , n 4 , n 5 , n 8 , n 12 } and n 10 {n 7 }. The relations define a partial order O between nodes. The partial order for nodes n 1 , . . . , n 12 is shown on the right side of Figure4: we note that some pairs of nodes have no defined ordering according to and . For instance, n 11 and n 7 have no established or relation. We note that our notion of domination between nodes is similar to the notion of domination between data points that is used in the context of databases, Algorithm 2: Multi-dimensional domination-based slicing algorithm on node p (getSlice() is unchanged from Algorithm 1).

	4 (left) gives an exam-ple of domination relations be-tween a node n 10 and a set of other nodes, over a 2-dimensional space. Relations that can be deter-mined based on transitivity are not // in addition to the ones in Algorithm 1 // local multi-dimensional attribute // view, close dominated nodes // view, close dominating nodes // active thread select partner q randomly from -view ∪ -view ∪ PSS.getView() ap = {a 1 p , . . . , a d p } ← . . . -view ← ∅ -view ← ∅ 5 every ∆ time units do 2 3 4 6 7 ep ← q.receive(ap, ep, p) // initiate exchange 8 function receive(aq, eq, q) // passive thread 9 if ap aq then 10 -view ← -view ∪ q // maintain close dominated nodes view shown. 1 variables 11 foreach n ∈ -view do

12 if ∃ n ∈ -view | n = n ∧ a n an then 13 -view ← -view \ n 14 if eq < ep then return exchange(eq) // order violation? 15 else if ap aq then 16 -view ← -view ∪ q // maintain close dominating nodes view 17 foreach n ∈ -view do 18 if ∃ n ∈ -view | n = n ∧ a n an then 19 -view ← -view \ n 20 if eq > ep then return exchange(eq) // order violation? 21 return(eq) // no order violation, no exchange 22 function exchange(eq) 23 t ← ep, ep ← eq, return t

The choice of the partner for the gossip interaction could be based on semantic information piggybacked on the PSS messages. This requires however one more round of interaction to check that the actual known attribute values are up to date (as detailed in[START_REF] Jelasity | Ordered slicing of very large-scale overlay networks[END_REF]). We did not implement this optimization in our prototype.

Acknowledgments. This work received support from the Portuguese Foundation for Science and Technology under grant SFRH/BD/71476/2010.


Abdulrahman Azab
email: abdulrahman.azab@ux.uis.no

Hein Meling

Reggie Davidrajuh
email: reggie.davidrajuh@uis.no

A Fuzzy-Logic Based Coordinated Scheduling Technique for Inter-grid Architectures

des établissements d'enseignement et de recherche français ou étrangers, des laboratoires publics ou privés.

Introduction

Grid computing provides the infrastructure for aggregating different types of resources (e.g. desktops, mainframes, storage servers) for solving intensive problems in different scientific and industrial fields, e.g. DNA analysis, weather forecasting, modelling and simulation of geological phenomenon [START_REF] Foster | The anatomy of the grid: Enabling scalable virtual organizations[END_REF]. Based on the delivered service, grid computing can be classified into computational grids, and data grids [START_REF] Yeo | Utility Computing on Global Grids[END_REF]. The target of computational grid, which is our main focus, is to aggregate many grid compute resources as one powerful unit on which computational intensive applications can run and produce results with low latency. Computational grid model is mainly composed of three components: (i) clients that consume grid resources by submitting computational jobs, (ii) resource brokers who are responsible of allocating submitted jobs to matching workers, and (iii) workers that execute submitted jobs. Due to the rapid growth in the demand for compute resources, as more and more compute intensive applications are being built, there is a need to scale up the grid infrastructure with more workers to fulfil those demands. To scale up, without negatively impacting overall performance and throughput is challenging, as increasing the number of nodes in a single grid could lead to a performance bottleneck since the broker may become overloaded.

Introducing multiple load balanced brokers for the same resource set requires some form of coordination between the brokers [START_REF] Ranjan | Coordinated resource provisioning in federated grids[END_REF]. One possible solution is to establish an interconnection between existing grid domains. This would enable migration of grid jobs between domains to avoid job starvation. The concept of interconnecting grid domains was first introduced by the Condor project in 1992, by means of a technique called flocking [START_REF] Evers | Condor flocking: load sharing between pools of workstations[END_REF]. The initial idea was to use gateway machine in each Condor pool, i.e. a grid domain, to manage the interconnection between domains. Due to the lack of coordination between the broker and the gateway inside domains, this has now been replaced by a client initiated approach [START_REF] For | High Throughput Computing, HTCondor Version 7.9.3 Manual[END_REF], which enables client nodes to submit jobs to external brokers when no matches are found by their local brokers. This however, is not practical when a large number of interconnected domains must be configured, each with hundreds of clients that communicate with many external brokers. Other grid systems (e.g. gLite [START_REF]Programming the Grid with gLite[END_REF], Condor-G [START_REF] Frey | Condor-g: A computation management agent for multi-institutional grids[END_REF], UNICORE [START_REF]Chemomentum -unicore 6 based infrastructure for complex applications in and technology[END_REF], Nimrod-G [START_REF] Buyya | Nimrod/g: An architecture for a resource management and scheduling system in a global computational grid[END_REF]) used the concept of meta-scheduling [START_REF] Schopf | Ten actions when superscheduling[END_REF], known also as super-scheduling. Meta-schedulers work in a layer above traditional brokers so that instead of submitting to their local brokers, users submit their demands to a meta-scheduler which transfers the submissions to a broker on a grid domain which can fulfil the demand. The problem with this approach is the lack of coordination between meta-schedulers. Another approach, broker overlay, is to establish the interconnection through an overlay network between brokers [START_REF] Butt | A self-organizing flock of condors[END_REF][START_REF] Weissman | A federated model for scheduling in widearea systems[END_REF][START_REF] Shan | Job superscheduler architecture and performance in computational grid environments[END_REF][START_REF] Daval-Frerot | Federation of resource traders in objects-oriented distributed systems[END_REF][START_REF] Lai | Tycoon: An implementation of a distributed, market-based resource allocation system[END_REF][START_REF] Ranjan | Coordinated resource provisioning in federated grids[END_REF]. This approach has proven to be scalable [START_REF] Butt | A self-organizing flock of condors[END_REF] but it doesn't achieve load balancing. The reason is that for each external job submission, the local broker sends a query to its neighbouring brokers looking for a match. Any matching outside its neighbourhood will not be detected. Avoiding this problem by constructing a fully-connected logical topology between brokers is not practical, since that will result in increased scheduling overhead as the number of links between brokers increase. This paper presents a coordinated scheduling technique for interconnected grids, which is an addition to our Slick broker overlay architecture [START_REF] Azab | Slick: A coordinated job allocation technique for intergrid architectures[END_REF]. In Slick, brokers maintains resource information about workers in its domain in a reduced information set (RIS) data structure. The RIS of each domain is disseminated to other brokers through a simple gossip based protocol [START_REF] Vahdat | Epidemic routing for partially connected ad hoc networks[END_REF]. Thus, each broker obtains a RIS from all the other domains in the system, and maintains this information in a data structure that we call global information set (GIS). Jobs with no matching workers in their local domains are exported to external domains with matching workers. Selecting the domain to export a job to is performed through a matchmaking process between the job requirements and the resource attributes of each RIS in the local GIS. Previously in Slick [START_REF] Azab | Slick: A coordinated job allocation technique for intergrid architectures[END_REF], we proposed a simple matchmaking technique which finds any match. This paper proposes a fuzzy scheduler that finds the best matching domain for jobs.

The proposed technique is modelled using the PeerSim P2P network simulator [START_REF] Montresor | PeerSim: A scalable P2P simulator[END_REF]. Using the simulation model we have tested the proposed technique by interconnecting simulated Condor pools [START_REF] Litzkow | Condor -a hunter of idle workstations[END_REF] using Slick. Job allocation throughput and load balancing of the proposed technique is tested against existing scheduling techniques. The results show that for a system capacity of 50,000 nodes divided into 512 domains, the proposed technique achieves load balancing and reasonable throughput. In the broker overlay, we used a broker-to-broker neighbourhood degree of k = ln N ln 2 , where N is the total number of brokers. The paper is organised as follows: Section 2 gives the background for the problem, and surveys related work. Section 3 describes the design of the Slick gateway and the Fuzzy scheduling technique. Section 5 presents the simulation model and our evaluation. Section 6 concludes the paper.

Scheduling in Inter-Grid Architectures

The interconnection of grid domains may be implemented in one of three levels:

(1) Client level where the client machine can have access to multiple domains using associated access rights [START_REF] Evers | Condor flocking: load sharing between pools of workstations[END_REF][START_REF] Aiftimiei | Design and implementation of the glite cream job management service[END_REF], [START_REF] Yeo | Utility Computing on Global Grids[END_REF] Worker level where worker nodes can install the task executors of multiple domains and thus becomes available for task submissions from either of those domains [START_REF]NorduGrid: Nordic Testbed for Wide Area Computing and Data Handling[END_REF][START_REF]Programming the Grid with gLite[END_REF], and (3) Broker level where the interconnection is carried out through local resource brokers. Two different methodologies follow the latter approach: (a) Central meta-scheduler, and (b) broker overlay. The role of a central meta-scheduler [START_REF] Schopf | Ten actions when superscheduling[END_REF] is to manage the inter-grid submission requests, allocating each to a broker with matching resource requirements in its domain [START_REF] Frey | Condor-g: A computation management agent for multi-institutional grids[END_REF][START_REF] Buyya | Nimrod/g: An architecture for a resource management and scheduling system in a global computational grid[END_REF]. Scheduling in broker overlay architectures can be further classified into: (i) non-coordinated, and (ii) coordinated [START_REF] Ranjan | Coordinated resource provisioning in federated grids[END_REF]. For non-coordinated scheduling, there is no cooperation between brokers making scheduling decisions for tasks assigned to external domains. One methodology is to have a shared directory service, as in UNICORE [START_REF]Chemomentum -unicore 6 based infrastructure for complex applications in and technology[END_REF], where brokers can retrieve information about workers in external domains for making cross-domain scheduling. Another methodology is implemented in Tycoon [START_REF] Lai | Tycoon: An implementation of a distributed, market-based resource allocation system[END_REF], where each broker declares a public independent auction for its available resources and external brokers can compete to gain access to them. For coordinated scheduling, a broker communicates with other brokers for negotiation about their available resources before taking the inter-grid scheduling decision. This communication is implemented through two main methodologies: (A) Broadcasting, e.g. one-toall, where cross-domain job requests are sent to all external brokers and take decisions based on positive responses [START_REF] Shan | Job superscheduler architecture and performance in computational grid environments[END_REF][START_REF] Weissman | A federated model for scheduling in widearea systems[END_REF][START_REF] Daval-Frerot | Federation of resource traders in objects-oriented distributed systems[END_REF]. This methodology is applicable only for small number of domains, as otherwise the negotiation process would negatively influence the overall throughput. (B) Multi-casting, where instead of negotiating with all external brokers, each broker only negotiate with a select set of neighbouring brokers based on a broker overlay structure [START_REF] Ranjan | Coordinated resource provisioning in federated grids[END_REF][START_REF] Butt | A self-organizing flock of condors[END_REF][START_REF] Weissman | A federated model for scheduling in widearea systems[END_REF][START_REF] Daval-Frerot | Federation of resource traders in objects-oriented distributed systems[END_REF]. This methodology is both scalable and fault-tolerant, but due to the need to disseminate of resource information, its ability to identify available resources depends on the neighbourhood degree [START_REF] Rowstron | Pastry: Scalable, decentralized object location, and routing for large-scale peer-to-peer systems[END_REF]. This means that for a broker to search for some specific resource information, multi-hop communication may be necessary, since a broker overlay network can offer reliable dissemination among available brokers. However, there can be a significant latency involved in such multi-hop communication [START_REF] Stoica | Chord: a scalable peer-to-peer lookup protocol for internet applications[END_REF][START_REF] Rowstron | Pastry: Scalable, decentralized object location, and routing for large-scale peer-to-peer systems[END_REF]. Due to this obstacle, coordinated-scheduling based intergrid systems only rely on a small set of resource specification parameters, e.g. three parameters [START_REF] Ranjan | Coordinated resource provisioning in federated grids[END_REF], for coordination.

In Slick we implement multi-casting based coordinated scheduling, where each broker keeps a replica of the resource information of each external domain, and periodically synchronises local replicas with its neighbours.

Design

Slick performs inter-grid job scheduling over a broker overlay of interconnected grid domains. Each broker in the overlay has a fixed number of neighbours k < N , where N is the total number of brokers. The k value is determined by the overlay topology, that is, k = 2 for the Ring topology, and k = ln N ln 2 for the Hypercube topology. The system architecture is shown in Figure 1. The Slick gateway is designed to be installed on broker nodes as an additional layer on the top of the broker to manage the interaction with external domains. In this paper, we present our implementation of the Slick gateway for Condor3 [START_REF] Litzkow | Condor -a hunter of idle workstations[END_REF] interconnected domains.

Condor

Condor provides a way to harness idle computational cycles for use by computational intensive distributed applications. A Condor pool is composed of: (i) users (clients), which submit jobs for execution, (ii) executors (workers), which runs compute jobs, and (iii) a central manager (broker), which is responsible for collecting job submission requests and allocate jobs to workers. One machine can operate as both client and worker. Condor uses an extensive machine and job description language called ClassAds [START_REF] Raman | Matchmaking: Distributed resource management for high throughput computing[END_REF]. The ClassAd describing a machine's computational power is simply number of slots that it can support, which is equal to the number of CPU cores on the machine. A Condor ClassAd job description can represent: (i) a Simple job, which has one process and needs one slot to run or (ii) a Job cluster, which is a collection of simple jobs and needs n slots to run, where n is the number of jobs in the job cluster. Upon a job submission to the broker, it is placed in a queue. The job waits until the broker finds a worker with resources matching the job requirements. Matchmaking is carried out between the job's ClassAd and each worker's machine ClassAd. The job is allocated to the first matching worker with idle slots. Jobs of one job cluster can be allocated to multiple worker machines based on which matching workers have idle slots.

Domain Resource Description

Condor worker software automatically detects the number of CPU cores and the size of the virtual memory of the local machine, in addition to other machine specifications (e.g. operating system), and builds the machine ClassAd. Each machine ClassAd contains more than 100 attribute-value pairs, which would result in a large amounts of network traffic in the broker overlay, if we want to disseminate the machine ClassAds of the entire domain to other domains. To overcome this problem, we divide machine ClassAds into groups based on similarities in the hardware and software specifications, e.g. all Linux machines with x86 64 architecture, physical memory > 2 GB, supports Java, . . . , are in group 1. The number of machine group is the number of all available machine specifications which is supported by Condor [25], and exist in ≥ 1 worker machines in the system. The domain resource description is stored in the Slick gateway as a RIS, which is composed of: (i) a collection of key-value pairs, where the key is a machine group id and the value is the total number of slots which machines belong to the associated group. (ii) a time-stamp, ts, representing the read time of the RIS. The structure of the RIS is described in [START_REF] Azab | Slick: A coordinated job allocation technique for intergrid architectures[END_REF].

Resource Information Sharing

As described above, in addition to its local RIS, each domain keeps a replica of the RIS of the other domains in the system in the GIS, as shown in Figure 1. The RIS replication between brokers in the overlay is carried out by a data synchronisation manager, DSM. The DSM in each broker implements a simple anti-entropy gossip protocol [START_REF] Vahdat | Epidemic routing for partially connected ad hoc networks[END_REF], to synchronise the local GIS with its neighbours. The gossip protocol is described in [START_REF] Azab | Slick: A coordinated job allocation technique for intergrid architectures[END_REF].

Inter-Grid Scheduling

The roles of the Slick gateway are: (i) Receive jobs submitted by external brokers throughout the overlay and schedule them to the local job queue 4 . (ii) Sched-ule starving local jobs to external brokers. A job j is declared as starving by the starvation detector (see Figure 1), if (CurrentTime -SubmissionTime(j)) > β, where β is a timeout value. Whenever a job is detected to be starving, it is moved to the gateway queue (GQ). The gateway scheduler performs matchmaking between jobs in GQ and each RIS stored in the GIS, and submits each job to a matching broker.

In our previous paper introducing the Slick [START_REF] Azab | Slick: A coordinated job allocation technique for intergrid architectures[END_REF] architecture, we implemented a simple matchmaking technique, any-match. Matchmaking was performed between the job description ClassAd and the RIS replicas of only neighbour domains in the overlay. Upon finding any match D x for job j, implied that domain D x contained one or more workers who's machine group matches the requirements of job j. This would result in job j being allocated to domain D x . This technique works well for scheduling simple jobs. In this section we introduce a new matchmaking technique, best-match, where either a simple job or a job cluster is allocated to the domain with the highest match. The new technique is illustrated with an example of a job cluster in Figure 2. Otherwise, the route with the smallest number of hops between D and D x , r(D x ) selected. The route description is added to each domain in the list. 3. The new match list together with the job description are passed to the fuzzy matchmaker, where an ordered match list is generated based on best-matchfirst. The role of the fuzzy matchmaker is described in Section 4.1. 4. The local gateway scheduler allocates j to the first domain in the list. The other domains in the list are substitutes in case of failure or rejection.

Fuzzy Matchmaker

The fuzzy matchmaking is implemented based on Takagi-Sugeno fuzzy model [START_REF] Takagi | Fuzzy identification of systems and its applications to modeling and control[END_REF], in which output membership functions are either linear or constant. In this implementation, constant output membership functions are used. Here we continue using the example in Figure 2, but with only three domains in the match list for simplicity. The following steps are carried out by the fuzzy matchmaker:

Generation of Input Membership Functions Each domain in the match list is viewed as a fuzzy set. Each fuzzy set has two dynamic input membership functions, related to two crisp inputs: (i) The number of CPU slots required by the job (60 slots in Figure 2), and (ii) Image size which is the maximum amount of memory consumed by the job while running on the worker node (28 MB in Figure 2). The input membership functions for each domain are generated as follows:

1. The fuzzy matchmaker contacts the broker of each domain D x in the list to obtain two values: (i) the current number of idle slots v Dx , and (ii) the current queue size q Dx . 2. The CPU slots membership function is generated by [START_REF] Foster | The anatomy of the grid: Enabling scalable virtual organizations[END_REF], where S Dx is the set of all CPU slots in D x , read from RIS x , and v Dx . Three examples of CPU slots membership functions are shown in Figure 3a, where |S D1 | = 100, and v D1 = 35

Y Dx (x) = 1 x ∈ [0, v Dx ] x-|S Dx | v Dx -|S Dx | x ∈ [v Dx , |S Dx |] (1)
3. The Image size membership function is generated by (2) based on the fact that a job with an image size I must be allocated to a worker with virtual memory capacity ≥ I. The x-axis is divided into intervals (x i-1 , x i ], where i ∈ {1, 2, . . . , 6}, assuming that the image size of any job won't exceed 60 MB. The Image size membership function of domain D x is generated as follows: For each interval (x i-1 , x i ], we generate a set of CPU slots S xi as a subset of the total CPU slots in the domain, and the virtual memory size of each slot's machine ∈ S xi ≥ x i . The result will be a histogram. The membership function is generated using (2c) by connecting the second edge of each bar in the histogram to the next bar's using a straight line, and divide the result by the total number of slots in the domain, |S| in order to keep Y (x) < 1 . Three examples of the Image size membership function are shown in Figure 3b. Fuzzification of Inputs The fuzzification process is to take the inputs, determine the degree to which they belong to each of the appropriate fuzzy sets via input membership functions. Using the two inputs of Figure 2, 60 slots and 28 MB image size, the degree of membership in the three domains, D 15 , D 22 , and D 20 is determined by the intersection as shown in Figure 3a.

S Dx (x i ) = {s| s ∈ S Dx ∧ vm(s) ≥ x i } (2a) y Dx (x) = |S Dx (x i )| ∀x ∈ (x i-1 , x i ] (2b) Y Dx (x) = 1 |S Dx | (y Dx (x i-1 ) -y Dx (x i ))(x -x i ) x i-1 -x i + y Dx (x i ) ∀x ∈ (x i-1 , x i ] (2c)
Applying fuzzy operator Once the inputs have been fuzzified, we know the degree to which each part of the antecedent has been satisfied for each rule.

A separate rule must be created for each domain, D x in the match list of the matchmaking process for job j. The rule for N domains can be formulated as follows:

if CPUSlotsOf (j) is AllocatableInCPUSlots(D i ) and MemoryImageOf (j) is AllocatableInVirtualMemory(D i ) then j is AllocatableIn(D i )
where i = 1, 2, . . . , N and D i is the domain identifier. The and operator used in the rule above is implemented as a minimum function:

V 1 and V 2 = min(V 1 , V 2 ).
Using the example of Figure 3, membership D15 = min(1,0.63) = 0.63.

Applying the Implication Method Before applying the implication method, we set each rule's weight. The rule weight of the fuzzy set associated with each domain D x is calculated as:

RW (D x ) = 1 |r Dx | × (1 + q Dx |S Dx | )
where |r Dx | is the number of hops in the shortest route from the local domain to D x , q Dx is the queue size of D x , and |S Dx | is the total number of slots in D x . Once proper weighting has been assigned to a rule, each of the resulted values from implementing the fuzzy operator in each rule is multiplied with its associated rule weight. The output membership function associated with each fuzzy set will be a single number representing the unique identifier of the associated domain (ID(D x ) i = 1,. . . , N). The outputs of all rules are aggregated into a single fuzzy set whose membership function assigns a weighting for every output value, so that the domain with the highest match will have the highest resulted weight.

Defuzzification The input to the defuzzification process is the aggregate output fuzzy set and the output is a single number. The fuzzy set is the collection of weights resulted from the previous step. In this approach, the maxima aggregate function is implemented. The active membership function, associated with N input fuzzy sets for N domains for matching with a job j, will be:

OUT(j,N) = MAX[w (D 1 ), w (D 2 ), . . . , w (D N )]
Where: w (D i ) is the weight associated with the domain D i , and OUT(j,N) represents the value associated with the ID of the most suitable domain to allocate j. The final output is a List of IDs of the domains in descending according to the associated weights.

We test the efficiency of four inter-grid scheduling techniques: UNICORE service orchestrator [START_REF]Chemomentum -unicore 6 based infrastructure for complex applications in and technology[END_REF], Condor flocking [START_REF] Evers | Condor flocking: load sharing between pools of workstations[END_REF], P2P Condor flocking [START_REF] Butt | A self-organizing flock of condors[END_REF], Slick firstmatch [START_REF] Azab | Slick: A coordinated job allocation technique for intergrid architectures[END_REF], and Slick best-match.

UNICORE service orchestrator is a meta-scheduler that manages job allocation between batch systems and flat-structured Grids, e.g. condor [START_REF] Litzkow | Condor -a hunter of idle workstations[END_REF] and PBS [START_REF] Bode | The portable batch scheduler and the maui scheduler on linux clusters[END_REF], by installing a UNICORE gateway on each domain which is connected to the local broker [START_REF]Chemomentum -unicore 6 based infrastructure for complex applications in and technology[END_REF]. Inter-grid communication and job submission in UNICORE is carried out in a central manner through the service orchestrator.

Condor flocking is a client-to-domain interconnection. In case the local workers are busy, jobs can be submitted to external brokers which are in a manually configured list. No matching is performed before submission. Submission to external brokers is totally based on the order of the list. P2P condor flocking is a resource discovery technique which is built on the top of condor Central-Manager, i.e. local broker, to enable interconnecting condor pools, i.e. domains, in a structured p2p overlay. For cross-domain submissions, Only two attributes are used to decide to which broker to submit two: average CPU utilisation of the workers in that domain, and the broker queue size.

We present the results of simulating a large number of domains using Peer-Sim peer-to-peer simulator [START_REF] Montresor | PeerSim: A scalable P2P simulator[END_REF]. We simulate a system of 50,000 nodes in 512 interconnected domains. The domains are connected through local brokers in a HyperCube logical topology, i.e. in case of N interconnected domains, each broker will have k neighbours in its routing table where k = ln N ln 2 . In case of 512 brokers, each broker will have 9 neighbours. We implement a HyperCube instead of a full Mesh since it uses a smaller neighbourhood degree k and at the same time achieves efficient data synchronisation, described in [START_REF] Azab | Decentralized service allocation in a broker overlay based grid[END_REF]. In case of UNICORE service orchestrator, a logical star topology is used since UNICORE implements central meta-scheduling. Worker machine specifications are of two machine configuration groups:

-group1: 2 CPU slots, 4GB Memory, Windows OS, No java support.

-group2: 4 CPU slots, 8GB Memory, Linux OS, Java support Workers are divided equally between the two groups, 25,000 each, but scattered among the domains. We create a load of total 80,000 synthetic jobs. Job resource requirements are randomly set to be matching either group1 or group2. The execution time of each job is uniformly distributed t ∈ [100, 300] time units. Jobs are divided into 100 sequences. Each sequence is assigned to one broker. Using a uniformly distributed frequency f ∈ [50, 100] time units, a uniformly distributed number of jobs j ∈ [50, 100] is submitted periodically by each sequence. The process continues until all the 80,000 jobs are submitted. The total simulation time of the experiment is set to 2000 time units. The simulation is configured such that each of the following takes places in a one simulation time unit:

-The local scheduler processes one job from the local queue, i.e. either allocating the job in the local domain in case of local matching.

-The gateway scheduler processes one job from the gateway queue, in case of Slick, i.e. allocating the job to the matching domain. The job transmission time is proportional to the route length from the local domain to the matching domain, e.g. if the route length is 3 hops then the job transmission time is 3 time units. -Each Slick broker synchronises GIS with one neighbour broker in the overlay. The GIS synchronisation algorithm is described in [START_REF] Azab | Slick: A coordinated job allocation technique for intergrid architectures[END_REF]. Due to the size reduction of the resource information, each GIS synchronisation takes only one time unit.

We use two benchmarks: Job allocation throughput, and Load balancing.

Job allocation throughput is measured by reading the total number of waiting jobs in the system/time, Figure 4a, and number of job allocation trials Fig- ure 4b. In terms of decreasing the number of waiting jobs, it is clear that both Slick any-match and best-match techniques manage to reach a steady state. The any-match manages to allocate all jobs within 1800 time units, while best-mach does in 1345 time units. This shows the positive influence of applying the new technique on the performance. With other systems, a bottleneck case happens. This can be described that: For the UNICORE case, this is due to the fact that there is only one central meta-scheduler to carryout interconnections, which for cross-domain submissions allocates only one job per time unit. In case of flocking and p2p flocking, the reason of the bottleneck is the difference in worker specifications. for flocking, the brokers in the flocking list may be already saturated and/or don't have matching workers. In p2p flocking, the best broker is chosen based on its queue size and average CPU utilisation of its domain. However, the domain with the least loaded queue and lowest CPU utilisation may not have matching workers in other terms, e.g. operating system. The breakdown in curves after ≈ 800 time units is because we configured all job sequences to complete their submissions by that time. This was done in order to validate the system performance when job allocation is carried out only inter-domain and not intra-domain. The job allocation trials benchmark, Figure 4b, tells how many jobs needed to be reallocated to another domain how many times in order to start running. The value of y(x i ) is the number of jobs which needed to be reallocated x i times. y(0) is the number of jobs which have not been reallocated, i.e. were successfully allocated in their local domains. In case of Slick best-match, the number of re-allocations doesn't exceed 4. +50% of jobs are reallocated once. This is a positive indication that almost all scheduling decisions are accurate. In case of Slick any-match, the number of re-allocations exceeds 300 sometime. This is mainly because a matching domain doesn't necessarily has a number of idle slots matching a particular job in the time of this job's submission. All jobs are managed to find a suitable domains though. Other systems show different behaviours, but not much can be concluded since none of them manage to allocate all jobs. So, the displayed behaviour is not for a small portion of jobs. Load Balancing is measured by calculating for brokers, throughout the simulation: How long did it take to allocate all jobs owned by the domain of each broker, and what is the average waiting time. In Figure 5, it is clear that for Slick any-match technique, the total allocation time never exceeded 1500 time units and the average waiting time is below 800. The output is even better in case of Slick best-match, the total allocation time is below 1000, and the average waiting time is below 700. For UNICORE, none of the domains were able to finish the job allocation before the end of the simulation at 2000 time units. The same case occurs for more than half of the brokers in case of flocking and p2p flocking. The value of 2000 for both total allocation time and average waiting time indicates that this broker's jobs were not totally allocated.

Conclusions

This paper presented a coordinated scheduling technique for interconnected Grid domains. The key feature of the proposed technique is to allocate starving local jobs to the best-match domain, which is carried out by a fuzzy matchmaking. The proposed technique is tested through simulating its implementation on 512 interconnected Condor pools. It was proven that the proposed technique provides accurate cross-scheduling decisions compared to other techniques. Results also show that the proposed technique achieves load-balancing, high throughput, and is stable under high loads. We are planning to develop a failure control model for broker failures and churn. We are also planning to implement Slick on a inter-grid prototype having brokers and workers as virtual machines on Amazon EC2.

Fig. 1 :

 1 Fig. 1: System Architecture

Fig. 2 :

 2 Fig. 2: Matchmaking of a Condor job cluster using the Slick gateway scheduler.

Fig. 3 :

 3 Fig. 3: Fuzzy input membership functions

  Fig.4: system throughput: Overall Job Allocation Performance

Fig. 5 :

 5 Fig. 5: Load balancing: Average waiting time and total allocation time of jobs submitted to each of 512 domains

The new name of the Condor project is 'HTCondor'.

This may result in contention between local jobs and external jobs. We are planning to add a contention manager to avoid this problem in the future.