Gabriel Antoniu

Antonio Congiusta
email: acongiusta@deis.unical.it

Sébastien Monnet
email: sebastien.monnet@irisa.fr

Domenico Talia
email: talia@deis.unical.it

Paolo Trunfio
email: trunfio@deis.unical.it

PEER-TO-PEER METADATA MANAGEMENT FOR KNOWLEDGE DISCOVERY APPLICATIONS IN GRIDS *

Computational Grids are powerful platforms gathering computational power and storage space from thousands of geographically distributed resources. The applications running on such platforms need to efficiently and reliably access the various and heterogeneous distributed resources they offer. This can be achieved by using metadata information describing all available resources. It is therefore crucial to provide efficient metadata management architectures and frameworks. In this paper we describe the design of a Grid metadata management service. We focus on a particular use case: the Knowledge Grid architecture which provides high-level Grid services for distributed knowledge discovery applications. Taking advantage of an existing Grid data-sharing service, namely JuxMem, the proposed solution lies at the border between peer-to-peer systems and Web services.

Introduction

Computational Grids are powerful platforms gathering computational power and storage space from thousands of resources geographically distributed in several sites. These platforms are large-scale, heterogeneous, geographically distributed and dynamic architectures. Furthermore they contain many types of resources such as software tools, data sources, specific hardware, etc. These resources are spread over the whole platform. Therefore, it is crucial to provide a mean for the applications running on Grids to localize and access the available resources in such large-scale, heterogeneous, dynamic, distributed environment.

Each Grid resource can be described by a metadata item (eg., an XML document). Such a metadata document may contain the 1) the description of a particular resource, 2) its localization and 3) information on the resource usage (eg., command line options of a software tool, format of a data source, protocol used to access a particular node, etc.). Thus, given a resource metadata, it is possible to access the resource. All the metadata items, describing the whole set of resources available in a given Grid have to be managed in an efficient and reliable way especially in large-scale Grids.

In this paper we propose a software architecture of a scalable Grid metadata management service. We focus on a particular use case: metadata management for the Knowledge Grid [START_REF] Cannataro | The knowledge grid[END_REF]. The Knowledge Grid is a service-oriented software distributed framework that aims to offer high-level Grid services for knowledge discovery applications running on computational Grids. The Knowledge Grid services are built on top of existing, low-level Grid services such as GRAM [START_REF] Czajkowski | Resource Management Architecture for Metacomputing Systems[END_REF], GridFTP [START_REF] Allcock | Secure, Efficient Data Transport and Replica Management for High-Performance Data-Intensive Computing[END_REF] or MDS [START_REF] Czajkowski | Grid Information Services for Distributed Resource Sharing[END_REF].

Within the Knowledge Grid architecture, metadata provides information about how an object (either a data source or an algorithm) can be accessed. It consists of information on its actual location and on its format (for a data source) or its usage (for an algorithm).

As metadata is actually stored as pieces of data (eg., XML files), they may be treated as such. We take advantage of the good properties exhibited by an already existing Grid data-sharing service, JuxMem [START_REF] Antoniu | Future Generation Grids, chapter GDS: An Architecture Proposal for a Grid Data-Sharing Service[END_REF][START_REF] Antoniu | How to bring together fault tolerance and data consistency to enable grid data sharing[END_REF], to store and retrieve metadata. We then build a distributed and replicated hierarchical index of available metadata.

In the next section we briefly present the architecture of the Knowledge Grid and focus on its metadata management needs. Section 3 presents the JuxMem Grid data-sharing service that we use to reliably store and retrieve both resource metadata and the distributed replicated index. Section 4 describes our architecture for a metadata management Grid service tailored for the Knowledge Grid and based on JuxMem. Finally, Section 5 presents ongoing work and concludes this paper.

2.

The Knowledge Grid

Knowledge discovery in Grids

Nowadays, big companies have to deal with daily generated large amounts of data. They need tools to both store this information and retrieve knowledge from it. Computational Grids [START_REF] Kesselman | The Grid: Blueprint for a New Computing Infrastructure[END_REF] offering high computational power and large storage resources can be used to store and process large amounts of data. Furthermore their geographically distributed nature fits well with the companies architecture. Indeed companies data sources and computational power may be spread all over the world.

However, performing knowledge discovery over such a distributed and often heterogeneous architecture, using data sources and data mining algorithms spread over thousands of nodes is not a trivial task. Building and running a distributed knowledge discovery application on a Grid requires high-level services. Data sources to be mined have to be located, furthermore their format has to be discovered somehow (they could be relational databases, text files, etc.). As well, data mining algorithms and software tools have to be localized and their usage has to be known. Then the computations (data mining algorithms running over data sources) have to be scheduled over available Grid nodes. A knowledge discovery application can be complex, consisting in numerous sequential or parallel data mining algorithms working on identical or different data sources. Some data mining algorithm may be run with the data produced by another data mining algorithm, leading to task dependencies, etc.

The Knowledge Grid provides high-level services and a user-friendly interface VEGA [START_REF] Congiusta | VEGA: A Visual Environment for Developing Complex Grid Applications[END_REF] that allows a user to easily describe a distributed knowledge discovery application, it then takes care of locating the resources (data sources, algorithms, computational nodes), scheduling tasks, and executing the application. Within the Knowledge Grid, the application designer only has to describe an abstract execution plan, with VEGA, he can even do it graphically. An abstract execution plan defines at high level the algorithms to be executed and the data sources to be mined. The Knowledge Grid services (called K-Grid services for short thereafter) are responsible to locate the resources and services and instantiate the execution plan which becomes an instantiated execution plan like the one presented in Figure 1.

An instantiated execution plan contains a set of tasks -with assigned Grid resources-to be done (data transfers and computations). It is executed by the K-Grid services and it may be refined as resources may become available or unavailable in a Grid.

2.2

The Knowledge Grid architecture

The K-Grid services are organized in a two-layer software architecture: 1) the High-level K-Grid layer and 2) the Core-level K-Grid layer. In its current implementation, the different services composing the Knowledge Grid are Grid services interacting by using the WSRF [START_REF] Congiusta | Distributed data mining services leveraging WSRF[END_REF] standard. The organization of the K-Grid services is described by Figure 2. The High-level K-Grid layer includes services to compose, validate and execute distributed knowledge discovery applications. The main services of the High-level K-Grid services are:

The Data Access Service (DAS), responsible for data sources and mining results publication and search.

The Tools and Algorithms Access Service (TAAS), responsible for data mining and visualization tools and algorithms publication and search.

The Execution Plan Management Service (EPMS), allowing to describe a distributed knowledge discovery application by building an execution graph with constraints on resources. It generates an abstract execution plan (resources are not know yet).

The Results Presentation Service (RPS), offering services for knowledge discovery results presentation;

The services exhibited by the Core K-Grid layer are:

The Knowledge Discovery Service (KDS), responsible for metadata management. Every resource (nodes, algorithms and tools, data sources and mining results) of the Knowledge Grid is described by a metadata item. In the Knowledge Grid, resource metadata is a XML document stored in a Knowledge Metadata Repository (KMR).

The Resource Allocation and Execution Management Service (RAEMS), responsible to instantiate an abstract execution plan. It uses the KDS service to find resources satisfying the constraints imposed by the abstract execution plan. It is also responsible for the application execution management.

DAS TAAS EPMS RPS

RAEMS

Core-level K-Grid layer

KDS

KBR KMR KEPR

Tools and algorithms

Hight-level K-Grid layer Figure 2. The Knowledge Grid software architecture.

Current KDS design and limitations

The Knowledge Directory Service (KDS) is responsible for handling metadata describing Knowledge Grid resources. A sample metadata is presented by Figure 3. Such resources include hosts, data repositories, tools and algorithms used to extract, analyze, and manipulate data, execution plans, and knowledge models obtained as result of mining processes.

The metadata information is represented by XML documents stored in a component called Knowledge Metadata Repository (KMR). The functionalities <DataMiningSoftware name="AutoClass"> <Description> <KindOfData>flat file</KindOfData> <KindOfKnowledge>clusters</KindOfKnowledge> <KindOfTecnique>statistics</KindOfTecnique> <DrivingMethod>autonomous knowledge miner</DrivingMethod> </Description> <Usage> ... <Syntax> <Arg description="executable" type="required" value="/usr/autoclass/autoclass"> <Arg description="make a classification" type="alternative" value="-search"> <Arg description="a .db2 file" type="required"/> <Arg description="a .hd2 file" type="required"/> <Arg description="a .model file" type="required"/> <Arg description="a .s-params file" type="required"/> </Arg> <Arg description="create a report" type="alternative" value="-reports"> <Arg description="a .results-bin file" type="required"/> .. of the KDS are mostly used by DAS and TAAS services while publishing and searching for datasets and tools to be used in a KDD application. DAS and TAAS services always interact with a local instance of the KDS, which in turn may invoke one or more other remote KDS instances.

The KDS exports three main operations:

-publishResource, used to publish metadata related to a given resource into the KMR;

-searchResource, for locating resources that match some given search criteria;

-retrieveMetadata, invoked to retrieve metadata associated to a given resource identified by a provided KDS URL.

It should be noted that when a publishResource is performed, only an interaction between a DAS/TAAS service and the local KDS is needed, because each KMR instance stores metadata about resources available on the same Grid node on which the KMR itself is hosted. On the contrary, when a searchResource is invoked, the related query is first dispatched from the DAS/TAAS to the co-located KDS service, which then answers by checking the local KMR, and in turn forwards the same query to remote KDSs with the aim of finding more matches.

The retrieveMetadata receives a KDS URL returned by a previous invocation of the searchResource operation, and uses it to contact the remote KDS on which the resource is available to retrieve the associated metadata document.

It appears clear, thus, that the searchResource is the most complex activity performed by the KDS, because it involves interactions and coordination with remote instances of the same service. On the other hand, it should be mentioned that the architecture of the Knowledge Grid does not prescribe any particular mode of interaction and/or protocol between the different KDS instances.

The current implementation, for instance, is adopting to such purpose one of the simplest strategies: the query forwarding is performed by contacting concurrently all of the known remote KDS instances (avoiding loops).

In this paper we propose a new KDS design based on a shared distributed index handled by a Grid data-sharing service and a peer-to-peer technique. This is useful for reducing the number of remote KDS instances contacted when forwarding a search query.

Resources metadata like the one presented by Figure 3 should be stored in a persistent and fault tolerant storage. Furthermore, they may be shared by multiple applications, and sometimes updated. Therefore, it is necessary to maintain the consistency between the different copies that may exist in the Grid. Thus we use a data-sharing service, JuxMem, which offers transparent access to persistent mutable data, to store the XML files corresponding to pieces of metadata.

3.

JuxMem: a Grid data-sharing service

In this section we present the JuxMem Grid data-sharing service used in the design of the metadata management Grid service.

A hierarchical architecture

From the metadata management Grid service perspective, JuxMem is a service providing transparent access to persistent, mutable, shared data. When allocating memory, a client has to specify in how many sites1 the data should be replicated, and on how many nodes in each site. This results into the instantiation of a set of data replicas, associated to a group of peers called data group. Usually each node runs one single peer. The allocation primitive returns a global data-ID, which can be used by the other nodes to identify existing data. To obtain read and/or write access to a data block, the clients only need to use this data-ID.

The data group is hierarchically organized, as illustrated on Figure 4: the Global Data Group (GDG) gathers all provider nodes holding a replica of the same piece of data. These nodes can be distributed in different sites, thereby increasing the data availability if faults occur. The GDG is divided into Local Data Groups (LDG), which correspond to data copies located in a same site.

In order to access a piece of data, a client has to be attached to a specific LDG (to "map" the data). Then, when the client performs the read/write and synchronization operations, the consistency protocol layer manages data synchronization and data transmission between clients, LDGs and GDG, within the strict respect of the consistency model.

JuxMem software architecture

The JuxMem Grid service is composed of a set of layers presented in Figure 5. The lower layer Juk is the JuxMem kernel. It relies on JXTA [START_REF] Traversat | Project JXTA 2.0 Super-Peer Virtual Network[END_REF] to offer to the uppers layers publish/subscribe operations, efficient communication and storage facilities. Every node involved or using JuxMem is therefore managed in a peer-to-peer way using JXTA. JXTA is is a set of protocols allowing nodes (Grid nodes, PDA, etc.) to communicate and collaborate in a P2P manner. The implementations of these protocols provide the ability to obtain efficient communications on Grids [START_REF] Antoniu | Enabling the P2P JXTA Platform for High-Performance Networking Grid Infrastructures[END_REF].

Above Juk, a fault-tolerance layer is responsible for hierarchical data replication. It offers the concept of Self-Organizing Group (SOG), a SOG is a replication group that is able to adapt itself in case of dynamic changes (by creating new replicas or removing old ones), this provides the ability to keep fault tolerance guarantees even in presence of failures.

The upper layer is responsible for data consistency management, it serves data access requests, manages locks and maintain pending requests lists.

A multi-protocol architecture. The layers presented above are built as interchangeable software modules. Therefore, it is possible for each data item stored by the Grid data-sharing service to specify a particular consistency protocol or a particular SOG implementation. The JuxMem service can not be used in its current design to manage metadata, but additional features must be provided. Data stored in JuxMem is accessible (localizable) only by using its associated data-ID. Metadata items have to be localizable using only names and attributes. The following section presents our approach to build a metadata management service using the JuxMem data-sharing service.

A Grid metadata management service

From the Knowledge Grid viewpoint, the Grid metadata management service consists of a particular design and implementation for the KDS service and the KMR repository. The service presented below serves requests from the TAAS and the DAS High-level K-Grid services but also from the RAEMS Core K-Grid service (see Figure 2).

Our approach relies on the use of the JuxMem Grid data-sharing service prototype presented in the previous section. Metadata items are stored within the Grid data-sharing service.

Metadata storage and retrieval

Requirements. Resources metadata should remain available in the Grid. Therefore they should be stored in a fault tolerant and persistent manner. This may provide the ability to access metadata information in spite of failures and disconnections. Furthermore, some metadata should be updatable. In the Knowledge Grid use case, a piece of metadata may describe the result of a knowledge discovery task, this result may be refined later which leads to metadata modifications. If a resource location is changing, it should also be reflected by updating the associated metadata. Finally, metadata has to be localizable by providing name, attributes and constraints upon the described resource.

Storing metadata in a Grid data-sharing service. To achieve high availability of metadata despite failures we store them in the JuxMem Grid datasharing service. Each metadata item describing a resource is though replicated and associated to one unique ID as described in Section 3. This ID can then be used to retrieve metadata information stored in the Grid data-sharing service. Availability and consistency (eg., in case of concurrent updates) is then also managed by the Grid data-sharing service. The metadata items that will not be updated (e.g. describing a large data source that will not be updated and will not be moved) can take advantage of JuxMem multi-protocol feature by using a very simple and efficient consistency protocol without synchronization operations. Thus, JuxMem is used as a fault-tolerant, distributed and shared KMR (see Section 2) and JuxMem's data-IDs are used as KDS URLs.

Locality. Metadata information is strongly linked with the resource it describes. Therefore if the resource becomes unavailable, its corresponding metadata information would become useless (it can also become misleading). Therefore, regarding JuxMem hierarchical architecture, metadata information should be stored within the site containing the resource it describes (i.e. over one unique JuxMem LDG). If all the nodes of the site fail (due to a power failure in a computer room for instance) the resources metadata of this site become unavailable but it is also the case of the described resources. Thus, we choose to store metadata information in the described resources' site using only one JuxMem LDG per metadata item. However notice that LDG are reliable self-organizing groups, ie. the failure of a node does not lead to the loss of metadata items.

Fault-tolerant distributed indexes

While looking for metadata information using the searchResource operation, applications2 can provide information like a name (eg., a data source name "clientdata1" or an algorithm name "J48") or a set of attributes and constraints as the one in the "Description" section of the metadata presented in Figure 3. An accurate description of the kind of requests the KDS service should be able to serve is given in [START_REF] Mastroianni | Metadata for managing grid resources in data mining applications[END_REF].

Therefore it is necessary to have a mean to find a metadata identifier (which then permit to retrieve the metadata information itself) using names and attributes that represent the resource described by the metadata.

Distributed indexes.

Usual approaches rely on the use of a centralized indexing system. It can be either a relational database like MySQL [18] or a LDAP [START_REF] Hodges | Lightweight Directory Access Protocol (v3): Technical Specification[END_REF] server (used in previous Knowledge Grid implementations). We use distributed indexes: in each site composing the Grid, we maintain a site index of the published resources metadata within this site. This site index contains tuples consisting of the resource name, attributes (as a byte vector) and the resource metadata identifier (its JuxMem data-ID).

Fault tolerance. There again we rely on the JuxMem data-sharing service: the site indexes are data item that can be stored in JuxMem. Therefore they are automatically replicated for fault tolerance. Notice that a site index only contains information of its own site, furthermore it does not contain the whole metadata information but only metadata item names and some relevant attributes. Thus, a site index size remains limited. Index sharing. The WSRF KDS instances serving publishResource and searchResource requests are clients of the JuxMem service. In each site it is possible to have multiple KDS services having mapped the site's index as illustrated in Figure 6. The KDS are responsible for parsing the index, finding the metadata identifier, fetching the metadata (using the identifier) and sending back the retrieved metadata to the requester (either DAS, TAAS or RAEMS). These tasks are achieved by interacting with the JuxMem service. It is important to notice that the site index is a data item stored by JuxMem and mapped by the multiple KDS: the consistency of the shared index is ensured by the grid data-sharing service while new publications occur.

The shared site indexes allows KDS instances to retrieve locally (on their node) KDS URLs of metadata describing resources spread over their site nodes. The KDS then releases the lock upon the index. To allow the other KDS to continue serving search requests while an update occurs, we use a particular consistency protocol allowing read operations concurrent to a write operation. Such a protocol is available in JuxMem, it is described and evaluated in details in [START_REF] Gabriel Antoniu | Extending the entry consistency model to enable efficient visualization for code-coupling grid applications[END_REF]. Note that the publication of a site resource affects only the index stored in this site and used by the local KDSs, furthermore resources metadata are also stored on intra-site JuxMem providers (in LDGs). Therefore a publish operation does not imply inter-site communications.

4.3.2

Metadata search. When K-Grid services need to search a particular resource metadata, they request the KDS running on the same node or a randomly chosen remote KDS within their own site 3 . The KDS receiving such a request search in its mapped (in its local memory) site index. If the resource is found, it gets the corresponding XML file using the data-ID stored in the site index. In this case the resource is available within the same site. If a corresponding resource can not be found in the site index, the KDS forwards it to one randomly chosen KDS in each other site involved in the Grid using JXTA peer-to-peer communication layers as illustrated by Figure 7. To achieve this, a partial list of KDS instances is stored and maintained within the JuxMem Grid 3 A round robin policy could also be used.

data-sharing service. This list is replicated hierarchically in the whole platform using the GDG/LDG hierarchy presented in Section 3. At initialization, a KDS maps its site index and the KDS list, it can then add itself into this list. If a KDS does not answer to a request (either publish or search), it is removed from this list. The KDS list is not expected to be frequently updated as Grid nodes are assumed more stable than peers in peerto-peer systems.

KDS

Technical concerns

From a technical view point our solution implies integrating the JuxMem and the Knowledge Grid research prototypes. JuxMem entities are managed in a peer-to-peer manner, using Sun Microsystems JXTA protocols, while the entities involved in the Knowledge Grid service use the WSRF Grid standard.

The junction between the two different sets of protocols is done by the new KDS implementation: KDS instances are part of both the JuxMem platform, as clients of the JuxMem Grid service, and they serve WSRF requests from the Knowledge Grid services (publishResource and searchResource).

The KDSs are also responsible to parse the distributed index. The distributed nature of the index implies a cooperation between the KDS instances distributed in different sites all over the Grid. This cooperation is made is a peer-2-peer manner, taking advantage of the Grid data-sharing service to store, manage and share a neighbor list (the KDS list).

Conclusion

Metadata data management in large scale, heterogeneous, geographically distributed and dynamic architectures such as computational Grids is an important problem. Providing an efficient and reliable metadata management service allows applications to easily access heterogeneous resources spread over thousands of nodes.

The solution we presented in this paper takes advantage of already existing work in Grids. By integrating the JuxMem Grid data-sharing service in the design of a metadata management service for the Knowledge Grid, XML metadata files are stored on a fault tolerant and consistent support, and are kept close to the resources they describe. The proposed two-level index hierarchy allows the applications to get resources located in their own site if they exist or in remote ones otherwise, enhancing locality.

The integration of the two research prototypes is in progress, and we plan to evaluate this solution on a real Grid platform such as Grid'5000 [START_REF]Grid'5000 Project[END_REF][START_REF] Cappello | Grid'5000: A Large Scale, Reconfigurable, Controlable and Monitorable Grid Platform[END_REF]. The format of the distributed index should then be further investigated, using binaries trees for instance. The peer-to-peer cooperation between KDS instances should also be enhanced, for instance by selecting several KDSs for inter-site cooperation.

Figure 1 .

 1 Figure1. A sample instantiated execution plan (from[START_REF] Mastroianni | Metadata for managing grid resources in data mining applications[END_REF]).

Figure 3 .

 3 Figure 3. An extract from an XML metadata sample for the AutoClass software (presented in[START_REF] Mastroianni | Metadata for managing grid resources in data mining applications[END_REF]).

Figure 4 .

 4 Figure 4. JuxMem: a hierarchical architecture.

Figure 5 .

 5 Figure 5. JuxMem layered software architecture.

  Metadata publications done by the DAS and the TAAS are made through the publishResource operation provided by KDS. When a KDS receives such a request:1 It stores the corresponding XML file within JuxMem, 2 locks the index to ensure no concurrent publish occurs, 3 updates it, adding the new resource metadata index entry (attributes and JuxMem data-ID of the XML file).

Figure 6 .

 6 Figure[START_REF] Antoniu | Enabling the P2P JXTA Platform for High-Performance Networking Grid Infrastructures[END_REF]. In each site KDS interact with the JuxMem service to access 1) the local index, 2) the KDS list, and 3) the metadata information itself.

Figure 7 .

 7 Figure 7. Among sites, KDS cooperate using the hierarchically replicated KDS list.

on minos.cs.icar.cnr.it"/> <Input

  href="minos.cs.icar.cnr.it/data/car-imports_db2.xml" title="car-imports.db2

	<ExecutionPlan type="instantiated">
	<Task label="task1">
	<Program href="minos.cs.icar.cnr.it/software/DB2Extractor.xml"
	title="DB2Extractor

on minos.cs.icar.cnr.it"/> <Output

	href="minos.cs.icar.cnr.it/data/imports-85c_db2.xml"
	title="imports-85c.db2

on minos.cs.icar.cnr.it"/>

	</Task>
	<Task label="check1">
	<ResourceCheck method="soft"/>
	</Task>
	<Task label="task2">
	<Program href="minos.cs.icar.cnr.it/software/GridFTP.xml"
	title="GridFTP

on minos.cs.icar.cnr.it"/> <Input

	href="minos.cs.icar.cnr.it/data/imports-85c_db2.xml"
	title="imports-85c.db2 on minos.cs.icar.cnr.it"/>
	<Output href="abstract_host1/data/imports-85c_db2.xml"
	title="imports-85c.db2 on abstract_host1"/>
	</Task>
	...
	<Task label="task6">
	<Program href="abstract_host1/software/autoclass3-3-3.xml"
	title="autoclass on abstract_host1"/>
	<Input href="abstract_host1/data/imports-85c_db2.xml"
	title="imports-85c.db2 on abstract_host1"/>
	<Output href="abstract_host1/data/classes.xml"
	title="classes on abstract_host1"/>
	</Task>
	...
	<TaskLink ep:from="task1" ep:to="check1"/>
	<TaskLink ep:from="check1" ep:to="task2"/>
	<TaskLink ep:from="task2" ep:to="task3"/>
	...
	<TaskLink ep:from="task5" ep:to="task6"/>
	...
	<ResourceInstantiation abstractResource="abstract_host1">
	<candidateResource>icarus.cs.icar.cnr.it</candidateResource>
	<candidateResource>telesio.cs.icar.cnr.it</candidateResource>
	</ResourceInstantiation>
	</ExecutionPlan>

A site is a set of clustered nodes, it can be a physical cluster within a cluster federation, or close from a latency viewpoint.

In our current use case the applications are the DAS, TAAS and RAEMS K-Grid services.

* This work was carried out for the CoreGRID IST project n˚004265, funded by the European Commission.


Marco Aldinucci
email: aldinuc@di.unipi.it

Marco Danelutto

Gabriel Antoniu
email: gabriel.antoniu@irisa.fr

Mathieu Jan
email: mathieu.jan@irisa.fr

FAULT-TOLERANT DATA SHARING FOR HIGH-LEVEL GRID PROGRAMMING: A HIERARCHICAL STORAGE ARCHITECTURE

Keywords: Grid, shared memory, High-level programming, memory hierarchy, P2P

Enabling high-level programming models on grids is today a major challenge. A way to achieve this goal relies on the use of environments able to transparently and automatically provide adequate support for low-level, grid-specific issues (fault-tolerance, scalability, etc.). This paper discusses the above approach when applied to grid data management. As a case study, we propose a 2-tier software architecture that supports transparent, fault-tolerant, grid-level data sharing in the ASSIST programming environment (University of Pisa), based on the JuxMem grid data sharing service (INRIA Rennes).

Introduction

Grid computing has emerged as an attempt to provide users with the illusion of an infinitely powerful, easy-to-use computer, which can solve very complex problems. This very appealing illusion is to be provided [START_REF]The JXTA (juxtapose) project[END_REF] by relying on the aggregated power of standard (so, inexpensive), geographically distributed resources owned by multiple organizations; (2) by hiding as much as possible the complexity of the distributed infrastructure to users. However, the current status in most software grid infrastructures available today is rather far away from this vision. When designing programs able to run on such large-scale platforms, programmers often need to explicitly take into account resource heterogeneity, as well as the unreliability of the distributed infrastructure. In this context, the grid community converges towards a consensus about the need for a high-level programming model, whereas most of the grid-specific efforts are moved away from programmers to grid tools and run-time systems. This direction is currently pursued by several research initiatives and programming environments, such as ASSIST [START_REF] Vanneschi | The programming model of ASSIST, an environment for parallel and distributed portable applications[END_REF], eSkel [START_REF] Cole | Bringing skeletons out of the closet: A pragmatic manifesto for skeletal parallel programming[END_REF], GrADS [START_REF] Vadhiyar | Self adaptability in grid computing[END_REF], ProActive [START_REF] Baude | On hierarchical, parallel and distributed components for grid programming[END_REF], Ibis [START_REF] Van Nieuwpoort | Ibis: a flexible and efficient Java-based grid programming environment[END_REF], Higher Order Components [START_REF] Dünnweber | HOC-SA: A grid service architecture for higher-order components[END_REF], etc., along the lines of CoreGRID's "invisible grid" approach to next generation grid programming models.

In this work, we explore the applicability of the above approach to data management for high-level grid programming. We consider three main aspects that need to be automatically handled by the data storage infrastructure: data access transparency, data persistence and storage fault-tolerance.

Transparent access to data across the grid. One of the major goals of the grid concept is to provide an easy access the underlying resources, in a transparent way. The user should not need to be aware of the localization of the resources allocated to the application submitted. When applied to the management of the data used and produced by applications, this principle means that the grid infrastructure should automatically handle data storage and data transfer among clients, computing servers and storage servers as needed. However, most projects currently still rely on the explicit data access model, where clients have to move data to computing servers. In this context, grid-enabled file transfer tools have been proposed, such as GridFTP [START_REF] Allcock | Secure, Efficient Data Transport and Replica Management for High-Performance Data-Intensive Computing[END_REF], DiskRouter [START_REF] Kola | Diskrouter: A Flexible Infrastructure for High Performance Large Scale Data Transfers[END_REF], etc. In order to achieve a real virtualization of the management of large-scale distributed data, a step forward has been made by proposing a transparent data access model, as a part of the concept of grid data-sharing service, illustrated by the JuxMem software experimental platform [START_REF] Antoniu | JuxMem: An adaptive supportive platform for data sharing on the grid[END_REF].

Persistent storage of data on the grid. Since grid applications typically handle large masses of data, data transfer among sites can be costly, in terms of both latency and bandwidth. Therefore, the data-sharing service has to provide persistent data storage. Data produced by one computation can be made available to some other computation through direct access via globally shared data identifiers, by avoiding repeated transfers of large volumes of data between the different components of the grid.

Fault-tolerant storage of grid data. Data storage on the grid must cope with events such as storage resources joining and leaving, or unexpectedly failing. Replication techniques and failure detection mechanisms are thus necessary, in order to enhance data availability despite disconnections and failures. Such mechanisms need to be coupled with checkpoint/restart techniques for application components with data replication mechanisms. This way, in reaction to faults, checkpointed application components could be migrated to available resources and restarted, as long as the application status and the application data remain available thanks to fault-tolerant storage.

While the above properties are desirable in general for grid data sharing, in this paper we restrict our discussion to the more specific case of applications based on the ASSIST [START_REF] Aldinucci | ASSIST as a research framework for high-performance grid programming environments[END_REF] programming environment, developed at the University of Pisa. ASSIST provides a programming model that enables transparent data sharing among distributed execution entities. However, its dedicated storage component (called ad-HOC [START_REF] Aldinucci | Accelerating apache farms through ad-HOC distributed scalable object repository[END_REF]) does not enforce data persistence and is not tolerant to node failures. Such features are exhibited by the JuxMem [START_REF] Antoniu | JuxMem: An adaptive supportive platform for data sharing on the grid[END_REF] grid data-sharing service developed at INRIA Rennes, which enables transparent access to grid-scale storage, based on P2P techniques. The goal of this paper is to study how ASSIST and JuxMem could be integrated, in order to better support a high-level grid programming approach.

2.

Analysis: using JuxMem to enable grid-level, fault-tolerant storage in ASSIST

This section describes ASSIST's main features, then briefly introduces the data sharing facilities available with ad-HOC and JuxMem (which exhibit complementary features); it finally proposes an integrated architecture which takes advantage of this complementarity.

Data sharing in ASSIST

The ASSIST programming model. ASSIST applications are described by means of a coordination language, which can express arbitrary graphs of software modules, interconnected by typed streams of data. Modules can be either sequential or parallel. A sequential module wraps a sequential function. A parallel module (parmod) can be used to describe the parallel execution of a number of sequential activities that run as Virtual Processes (VPs) on items coming from input streams. Each stream realizes a one-way, one-tomany asynchronous channel between sequential or parallel modules. VPs can be organized according to programmer-defined topologies (array, etc.). The ASSIST compiler implements VPs according to the underlying middleware. It selects an appropriate granularity by mapping a subset of a parmod's VPs to the basic execution entity, e.g. a POSIX process/thread, a Globus Grid Service, or a Web Service. The full support for the forthcoming CoreGRID Grid Component Model (GCM) is currently under investigation [START_REF]Proposals for a Grid Component Model[END_REF][START_REF] Aldinucci | Self-configuring and self-optimizing grid components in the GCM model and their ASSIST implementation[END_REF].

Data sharing in ASSIST. The ASSIST programming model enables data sharing both within VPs of the same module and between different modules by means of two different methods, respectively:

Attributes, which are global, typed variables of the module. Attributes are owned either by the module itself or by VPs; in particular arrays ownership can be partitioned among VPs according to user-defined rules. References, which are global pointers in a shared memory space that is logically external to all application modules, thus can be accessed across different modules. ASSIST provides the programmer with an API to allocate, read, and write data structures in the shared space. Note that, in both cases, ASSIST does not implement a built-in DSM to enable memory sharing; it rather provides hooks to attach ASSIST run-time support to an external DSM. This allows to test different solutions for data sharing, since the DSM can be replaced with almost no impact on the core code of the compiler and of its run-time support. Currently, ASSIST comes with a DSM built on top of a set of cooperating ad-HOC memory servers [START_REF] Aldinucci | Accelerating apache farms through ad-HOC distributed scalable object repository[END_REF].

These variables adhere to the transparent data access model: programmers should neither care about data placement nor data splitting and distribution. These tasks are transparently and efficiently carried out by ad-HOC. However, in order to enforce fault-tolerance (and irrespectively of the checkpointing technique), ASSIST run-time needs to rely on permanent and robust data storage.

Existing building blocks: ad-HOC and JuxMem

Cluster-level sharing: ad-HOC. ad-HOC (Adaptive Distributed Herd of Object Caches), is a distributed object repository [START_REF] Aldinucci | Accelerating apache farms through ad-HOC distributed scalable object repository[END_REF] developed at University of Pisa. It provides the programming environment designer with building blocks to set up client-server and service-oriented infrastructures for data storage management. Clients may access data in servers through different protocols, which are implemented on client-side within a proxy library. A set of cooperating ad-HOC servers implements a permanent storage facility, i.e. a repository for arbitrary length, contiguous segments of data, namely objects since each data chuck is wrapped and stored and in an object homed in one of the servers [START_REF] Aldinucci | Accelerating apache farms through ad-HOC distributed scalable object repository[END_REF]. Objects can be grouped in ordered collections of objects, which can be spread across different servers. Both objects and their collections are identified by keys with fixed length. In particular, the key of a collection specifies to which spread-group the collection belongs. Such a group specifies how adjacent objects in the collection are mapped across the set of servers. Both classes of shared objects described in the previous section (attributes and references) are implemented in ASSIST via ad-HOC object collections and can thus be stored in a distributed way. The ad-HOC API enables to get/put/remove an object, and to create/destroy a key for a collection of objects. Each ad-HOC manages an object storage area for server home objects and a write-back cache for objects with a remote home. Basic ad-HOC operations do not natively ensure data coherence of cached objects. Nevertheless, server operations can be extended via the special operation execute that enables application proxies to put a serialized C++ object1 in a server and invoke its run method to issue a consistent access (e.g. lock/unlock, coherent put, etc).

The extremely simple ad-HOC API is mainly aimed to implementation efficiency, which relies on non-blocking I/O. On each port, an ad-HOC can efficiently serve many clients, each of which supports thousands of concurrent connections while squeezing a close to ideal bandwidth even for fine-grained data accesses [START_REF] Aldinucci | Accelerating apache farms through ad-HOC distributed scalable object repository[END_REF]. A set of ad-HOCs can efficiently cooperate across multi-tier networks and clusters with private address ranges, even when they are protected by firewalls. However, ad-HOC does not implement any form of fault-tolerance or automatic data replication: in the case a server permanently leaves the community implementing the data storage, some of the stored data will be lost. This problem is particularly severe if data structures are spread across all servers, since losing a single server may induce the corruption of the whole dataset.

Grid-level sharing: JuxMem. The JuxMem [START_REF] Antoniu | JuxMem: An adaptive supportive platform for data sharing on the grid[END_REF] grid data sharing service developed at INRIA Rennes transparently manages data localization and persistence in dynamic, large-scale, distributed environments. The data sharing service concept is based on a hybrid approach inspired by Distributed Shared Memory (DSM) systems (for transparent and consistent data sharing) and peerto-peer (P2P) systems (for scalability and volatility tolerance). The data sharing service can also transparently replicate data to ensure its persistence in case of failures. The consistency of the various replicas is then implemented through adequate fault-tolerant consistency protocols [START_REF] Antoniu | How to bring together fault tolerance and data consistency to enable grid data sharing[END_REF]. This approach will serve as a basis to the architecture proposed in this paper.

The JuxMem API provides the users with classical functions for allocating (juxmem malloc) and mapping/unmapping memory blocks (juxmem mmap, etc.) When allocating a memory block, the client can choose to replicate data to enhance fault-tolerance, and then has to specify: (1) on how many clusters the data should be replicated; (2) on how many providers in each cluster the data should be replicated; (3) the consistency protocol that should be used to manage this data. This results into the instantiation of a set of data replicas (associated to a group of nodes), called data group. The allocation operation returns a global data ID. This ID can be used by other nodes in order to identify existing data. It is JuxMem's responsibility to localize the data and perform the necessary data transfers based on this ID. This is how JuxMem provides a transparent access to data. To obtain read and/or write access on a data, a process that uses JuxMem should acquire the lock associated to the data through either juxmem acquire or juxmem acquire read. This permits to apply consistency guarantees according to the consistency protocol specified by the user at the allocation time of the data. Note that juxmem acquire read allows multiple readers to simultaneously access the same data.

The general architecture of JuxMem mirrors a federation of distributed clusters and is therefore hierarchical. The goal is to accurately map the physical network topology, in order to efficiently use the underlying high performance networks available on grid infrastructures. Consequently, the architecture of JuxMem relies on node sets to express the hierarchical nature of the targeted testbed. They are called cluster groups and correspond to physical clusters. These groups are included in a wider group, the juxmem group, which gathers all the nodes running the data-sharing service. Any cluster group consists of provider nodes which supply memory for data storage. Any node may use the service to allocate, read or write data as clients, in a peer-to-peer approach. This architecture has been implemented using the JXTA [START_REF]The JXTA (juxtapose) project[END_REF] generic P2P platform.

3.

Proposal: an integrated 2-tier architecture

Memory hierarchy is an effective way to enhance data management efficiency while respecting data access transparency. We propose a 2-tier memory hierarchy which trades off storage performance vs. robustness: it enables ASSIST modules to rely on a fault-tolerant grid storage while still exploiting low-latency/high-bandwidth data accesses, which are typical of a fast, clusterlevel DSM. The big picture is sketched in Fig. 1: the two levels of hierarchy are called grid tier and cluster tier. The hierarchy is connected to an additional tier where computation is figured out (at the bottom on the figure). This third level (represented at the bottom) also includes data storage that is local to platforms2 . The top two tiers of the memory hierarchy are specified as follows:

The grid tier enables data sharing across multiple parmods running on different clusters. It is implemented via JuxMem. Accesses to data stored in JuxMem are assumed to be less frequent, coarse-grained, with less performance constraints.

The cluster tier enables data sharing among multiple parallel codes (VPs, parmods) within a single cluster. It is implemented via an ad-HOC-based storage component. Accesses to data stored in ad-HOC are assumed to be frequent, fine-grained, with high-performance constraints. In the following, we define why and when the data stored at the cluster level should be copied/stored at the grid level and vice-versa. The answers come directly from the analysis of hierarchy goals.

Sharing across multiple clusters. A natural interaction between the two levels of the hierarchy will take place whenever data produced and stored on a cluster (at the ad-HOC level) has to be made available on another cluster (multisite deployment). At the programming level this may be triggered by passing a reference across the clusters via streams. Notice that this functionality can also be implemented at ad-HOC level by "horizontally" connecting ad-HOC storage components of different clusters. However, the hierarchical solution has some advantages: (1) data can be shared also among different applications thanks to persistence of grid level; (2) ASSIST does not need any more to handle the coallocation of resources (possibly via job schedulers) for all parmods distributed on different clusters, since JuxMem can serve as temporary storage; (3) data shared at this level can be considered safe w.r.t. node and networks faults (as this aspect is handled by the JuxMem-based grid-level).

The first point allows the user to use command scripts implementing a functionality similar to UNIX pipes in grid distributed fashion, while sharing data among the different processes through JuxMem. The second point addresses a very hot topic in grid middleware design: multi-site co-allocation. Let us suppose as an example that modules mapped on cluster A (see Fig. 1) start well before modules on cluster B. Data coming on stream from parmod 2 can be transiently buffered on grid storage up to the moment parmod 4 starts. This scenario can be extended to all cases in which data dependencies among clusters can be described by a Direct Acyclic Graph (co-allocation is not logically needed). The third point is discussed in details in the next paragraph.

Fault tolerant data storage/checkpointing. Since the ad-HOC provides efficient access latency, it can be used as a cache for intra-cluster data sharing. Since it is not fault-tolerant, the ad-HOC can periodically save application data to JuxMem. This feature could also be used for application checkpointing in general, as the computation status can be saved together with the data. Checkpointing is driven by ASSIST run-time support, which stores checkpoint information on the cluster tier, and then triggers a flush of "dirty" objects to the grid tier. Basically the cluster tier is used as a write-back cache. In the same way, this information could also be used to migrate parmods from one cluster to another. In such a case, ad-HOC data can be saved to JuxMem on the initial cluster and then read from JuxMem after migration on the second cluster.

Locality. The use of a memory hierarchy adequately supports clustered locality. Grids are generally structured as federations of clusters, where each cluster is pretty homogeneous and exhibits high-bandwidth/low-latency connectivity due to the spatial proximity and to the reduced security constraints, but also to the use of high-performance System Area Networks, such as Myrinet or Infiniband. High-level parallel languages and their run-time environments aim at enhancing clustered locality (1) by providing programmers with language paradigms (constructs, skeletons, design patters, etc.) leading to known and regular interaction patterns among concurrent activities; (2) by statically and/or dynamically mapping/enforcing clustered locality of a groups of activities demanding frequent interactions. A few examples are: the iterated halo-swap paradigm and the block data distribution [START_REF] Cole | Bringing skeletons out of the closet: A pragmatic manifesto for skeletal parallel programming[END_REF], the Divide&Conquer paradigm supported by dynamical load balance based on hierarchical work-stealing [START_REF] Van Nieuwpoort | Efficient load balancing for widearea divide-and-conquer applications[END_REF]. Component technology and related design methodologies further enforce clustered locality by encouraging application designer to aggregate related activities within a (possibly compound) component. As result, most of the shared memory accesses have a limited scope in terms of accessing entities, and those entities are preferably mapped and deployed on the same cluster, thereby improving the local-to-remote access ratio.

Design and Implementation

A prototypal implementation has recently been completed and is currently being tested. It will mainly be used to analyze the hierarchy's efficiency and limitations in order to refine the design phase. The current implementation relies on a set of external mediator processes that are started with ad-HOC servers by the ASSIST launcher. The mediator can take into account remote commands (through a TCP port), which trigger data copying from the cluster tier to the grid tier and vice-versa. Available commands include copying a set of ad-HOC objects (which can be spread across the nodes of the distributed ad-HOC server) to new or to existing JuxMem memory areas, and vice-versa. In both cases, the data exchanged between the two tiers flow through the mediator, which in one way reads a series of data objects from ad-HOC, collapses them in a single data chunk, and writes it into JuxMem; in the other case, it fetches a data chunk from JuxMem, splits and spreads it to ad-HOC servers of the cluster. Such an interaction is represented in Fig. 2, where J processes act as JuxMem storage providers and M processes (representing the mediators) act as JuxMem clients. Mediators and providers can be hosted on the same node, on different nodes in the same cluster or on different clusters.

These tasks leverage existing features of the ASSIST compiler. Since the compiler can produce dynamically adaptable code, it already identifies within parmod so-called reconf-safe points and instruments it with the required agreement protocols for coordinated checkpointing. In these points, the compiler can ensure a coherent and known state of application processes and the shared memory. This information can be dumped, as it can serve to restart the whole application or a proper subset of it from last reconf-safe point. This technique has been successfully used to enable dynamic migration of VPs and parmods [START_REF]Proposals for a Grid Component Model[END_REF].

Fault-tolerant storage. The proposed architecture can transparently be used to provide fault-tolerant storage for ASSIST by using the primary-backup approach, where the cluster tier behaves as a primary replica and the grid tier as backup. In particular, attributes can be safely stored in reconf-safe points. In these points ASSIST run-time triggers an update of backup copies via the mediator process. A similar technique is used for references. However since these are managed directly by programmers in the user code, the triggering is achieved by extending the API with a safe write operation (alternatively, the write operation can be transparently turned into a safe write operation).

Checkpoint/rollback. A fault-tolerant storage is necessary but not sufficient to make ASSIST fault-tolerant. However, a parmod can easily be made faulttolerant by triggering a checkpoint of its data and status into JuxMem. Starting from one of these checkpoints, a whole parmod can be rolled back by using already existing primitives for dynamic adaptation (see [START_REF]Proposals for a Grid Component Model[END_REF] for details).

Ensuring fault-tolerance of a complete ASSIST application is slightly more complex. This requires to equally taking into account the global status of streams established among parmods. Classic techniques based on message logging could than be used, while relying on the proposed memory hierarchy as a stable storage. This point is under investigation.

Conclusion and long-term aims

This paper addresses the problem of how to support of high-level grid programming by means of a software infrastructure that automatically and transparently handles low-level, grid-specific issues, such as multi-site resource distribution, fault-tolerance, etc. It proposes a hierarchical grid data storage architecture whose goal is to provide the ASSIST grid programming environment with grid-scale, fault-tolerant data-sharing facilities, as provided by the JuxMem grid data-sharing service. This work is a specific integration effort between existing researches carried out by two partners within the Institute on Programming Model (WP3) of the CoreGRID NoE. We mainly focus on architecture design, however a prototypal implementation has recently been completed and preliminary experiments are under way (the first experimental results should be available by the workshop date). The work in progress concerns the assessment of the performance of memory accesses between the two tiers of the hierarchy.

Long-term aims. Many of the approaches to data management on grid aim to optimize the access to very large, quite localized, mostly read-only scientific data [START_REF] Laure | Performance engineering in data Grids[END_REF]. However, in many real applications, data is inherently distributed across application components. We believe that a mature programming model for the grid should provide designers with an abstract view of data, and a high-level API to access it. Due to hierarchical nature of grid platform, any implementation of data management system will benefit from the clear understanding of qualitative and quantitative aspects governing a distributed memory hierarchy on the grid. These aspects may be used to enhance and automatize the distributed management of data in development tools for the grid. The proposed architecture appears flexible enough to investigate the benefits and overheads of hierarchical data management, and the extent to which data can be transparently moved (horizontally and vertically) across tiers of the hierarchy.

Figure 1 .

 1 Figure 1. Target architecture illustrating the interaction of ad-HOC and JuxMem.

2 A

 2 ASSIST process M mediator (JuxMem client)H ad-HOC process J JuxMem process

Figure 2 .

 2 Figure2. Implementation of proposed architecture using ad-HOC and JuxMem (the application is sketched in Fig.1). Dashed lines represent synchronizations inducing mediator actions.

Table 1 .

 1 Properties for cluster-level and grid-level sharing.

		Cluster sharing (ad-HOC)	Grid sharing (JuxMem)
	Throughput	High	Medium/High
	Latency	Low	High
	Fault-tolerance	No	Yes
	Protocols for data consistency	No	Yes
	ad-HOC and JuxMem: a quick comparison at hand. Due to different
	design goals, ad-HOC distributed storage server and of the JuxMem data-
	sharing service exhibit different features (as summarized in Table 1).
	ad-HOC provides transparent and permanent access to distributed data. It
	enables the distribution and the parallel access to collections of objects (e.g.
	arrays). It is robust w.r.t. node additions and removal, which should be driven
	by a proper node agreement protocol to guarantee data safety. It does not
	checkpoint/replicate data, thus is not fault-tolerant. Data accesses exhibit low

Object data is serialized at run-time and send to the server, while code must be provided (in advance) to the server as dynamically linkable library.

Register file level in classic memory hierarchies.

Acknowledgments. This work has been partially supported by the FP6 Network of Excellence CoreGRID funded by the European Commission (Contract IST-2002-004265), and by the Region of Brittany (France).


Gabriel Antoniu
email: gabriel.antoniu@irisa.fr

Luc Bougé

Sébastien Lacour

Making a DSM Consistency Protocol Hierarchy-Aware: An Efficient Synchronization Scheme

We consider the design of DSM consistency protocols for hierarchical architectures. Such architectures typically consist of a constellation of loosely-interconnected clusters, each cluster consisting of a set of tightly-interconnected nodes running multithreaded programs. We claim that high performance can only be reached by taking into account this interconnection hierarchy at the very core of the protocol design. Previous work has focused on improving locality in data management by caching remote data within clusters. In contrast, our idea is to improve locality in the synchronization management. We demonstrate the feasibility through an experimental implementation of this idea in a home-based protocol for Release Consistency, and we provide a preliminary evaluation of the expectable performance gain.

Introduction

Many recent high-performance computing platforms have been built by assembling together a large number of commercial off-the-shelf PCs. Such architectures are usually made of a constellation of loosely-connected clusters, each of them being made of a set of tightly-connected nodes. To exploit these architectures efficiently, many modern applications use multithreaded programming techniques to overlap communication delays with computation smoothly. Thus, the grand picture is a hierarchical interconnection structure with (at least) three levels of latency: 1) inter-cluster communication, through low-cost, mediumlatency Local Area Networks (FastEthernet, etc.); 2) internode communication through specific low-latency System Area Networks (SCI, Myrinet, etc.); and 3) inter-thread communication, through direct memory-level interaction. The ratio of latency between each level typically ranges from 10 to 100.

Designing middleware for such large, hierarchical configurations is a major scientific and technical challenge, as most existing solutions have been designed within a completely different context: a small to moderate number of nodes, e.g., a few dozens; a flat, hierarchy-unaware interconnection, where the communication latency may be considered as uniform across the partners. The common observation is that such approaches do not scale well in general, and that the hierarchical nature of the configuration must be taken into account at the very early design steps. An extensive re-thinking of the design is unavoidable.

Significant work has already been carried out to adapt MPI implementations on such large-scale, hierarchical architectures [START_REF] Kielmann | MPI's reduction operations in clustered wide area systems[END_REF][START_REF] Kielmann | MagPIe: MPI's collective communication operations for clustered wide area systems[END_REF][START_REF] Karonis | Exploiting hierarchy in parallel computer networks to optimize collective operation performance[END_REF], but very little study has been devoted to improve the performance of Distributed Shared Memory systems in such hierarchical configurations. A number of efforts have been dedicated to the design of efficient DSM systems for (flat) clusters of SMP nodes, like Cashmere-2L [START_REF] Stets | Cashmere-2L: Software coherent shared memory on a clustered remote-write network[END_REF] and HLRC-SMP [START_REF] Samanta | Home-based SVM protocols for SMP clusters: Design and performance[END_REF], which exhibit a two-level hierarchy: 1) message-passing communication at the higher, inter-node level; and 2) physically shared memory at the lower, intra-node level. In contrast, we are interested here in constellations of clusters of nodes, should these nodes be mono-or multi-processors. Message-passing is used for constellation-level and cluster-level communications, whereas node-level communication between threads relies on virtual-memory sharing.

When used "blindly" on such architectures, traditional consistency protocols intended for flat configurations usually fail to deliver good performance. Even the widelyaccepted Multiple-Writer DSM protocols designed for relaxed consistency models, are affected. Indeed, these protocols heavily rely on data transfers to and from nodes sharing a page, to inform each other about the modifications. Should a page be shared among different clusters, the performance of the whole system is limited by the high-latency inter-cluster links. An early work in this domain was for instance the Dosmos system [START_REF] Brunie | New propositions to improve the efficiency and scalability of DSM systems[END_REF]. One of the most advanced work we are aware of on this topic is the Clustered-LRC [START_REF] Arantes | An effective logical cache for a clustered LRC-based DSM system[END_REF] DSM system. That system extends a protocol for Lazy Release Consistency (LRC) used by TreadMarks [START_REF] Amza | TreadMarks: Shared memory computing on networks of workstations[END_REF], by taking into account the hierarchical interconnection in the management of replicated data. The main idea is to introduce cluster-based proxies, whose role is to cache page modifications at the level of each cluster. Successive accesses to the same page by nodes located within the same cluster can re-use the locally cached diffs, in contrast to the original TreadMarks DSM system.

We claim that this data management-directed approach is not the only way to obtain high performance from DSM systems on large, hierarchical architectures running multithreaded programs. Indeed, another major source of inefficiency in this context is synchronization, since DSM protocols for relaxed consistency models heavily rely on systemwide locks. Any access to shared data involves acquiring a lock, so that the latency of this operation is a crucial performance factor. Therefore, our purpose is to explore alternative, hierarchy-aware implementations for locks, where the acquisition time can take advantage of locality. Then we show how consistency protocols can benefit from this optimized implementation.

In this paper, we focus on Release Consistency protocols. We consider a flat protocol based on an eager, homebased approach, and we show that it is possible to make it hierarchy-aware while preserving the original semantics of the consistency model. The price to pay is a more elaborate management of fairness at the inter-node and inter-cluster levels. Though this paper focuses on a specific protocol, we claim that the approach is generic and can be applied to other similar protocols for relaxed consistency models.

The paper is organized as follows. In Section 2, we describe the original flat protocol and we show how it can be made hierarchy-aware, by introducing the notion of partial lock release. Section 3 illustrates some noteworthy implementation details related to this technique and Section 4 reports preliminary performance results obtained on the DSM-PM 2 generic implementation platform [START_REF] Antoniu | DSM-PM2: A portable implementation platform for multithreaded DSM consistency protocols[END_REF] for DSM protocols. Unfortunately, the experiments could only be run on a PC cluster far too small with respect to modern largescale architectures. Nevertheless, we can already demonstrate significant performance improvements on synthetic benchmarks.

A Hierarchy-Aware Protocol for Release Consistency 2.1. Hierarchy Awareness

The performance criterion we consider is latency: while bandwidth increases with technology, the latency of remote requests will always remain a function of switch delays and Node Thread Constellation Cluster Figure 1. Hierarchy of interconnections between threads, nodes and clusters: the thicker the line, the higher the latency.

distance, becoming the limiting factor of a network connection [START_REF] Keleher | Consistency maintenance in large-scale systems[END_REF]. Another reason why we focus on latency rather than bandwidth is that we are dealing with small messages of a few bytes (invalidation messages, acknowledgements) and relatively small messages of 4 kB or 8 kB (a page). We assume that we have only one process per node for the sake of simplicity, and we consider a 3-level hierarchy defining two gaps in the communication performance (see Figure 1):

• several threads running on a node, sharing the same address space within a single process;

• several nodes inside the same cluster communicating through a low-latency network such as SCI [START_REF]IEEE Standard for Scalable Coherent Interface (SCI)[END_REF];

• several clusters in a constellation communicating through a higher-latency network such as FastEthernet.

Our experimental platform is made of PCs (PentiumII at 450 MHz) connected over a fully-switched FastEthernet network, and equipped with relatively old SCI cards (type D310). On this platform, we observe an inter-cluster FastEthernet/TCP latency of 100 µs and an intra-cluster SCI/SISCI latency of 8 µs. Thus, the ratio of latency is around 12.

To cope with such a gap in the latencies, we try to minimize the number of messages sent over the high-latency links, and to make synchronization and consistency operations as local as possible.

A flat Home-Based Protocol

Our goal is to illustrate how a consistency protocol can benefit from hierarchy-aware synchronization. For the sake of simplicity, we use an eager variant of the HLRC [START_REF] Iftode | Home-based Shared Virtual Memory[END_REF][START_REF] Zhou | Performance evaluation of two home-based lazy release consistency protocols for shared memory virtual memory systems[END_REF] (Home-Based Lazy Release Consistency) protocol as a starting point. HLRC allows multiple writers (i.e., concurrent threads running on different nodes) to modify different parts of a page simultaneously. These modifications are made within critical sections. Each page is statically attached to a particular node (called home node), which is in charge of maintaining an up-to-date version of the page. When a thread has finished its write accesses to a page, it exits its critical section by releasing a lock. At that moment, the thread computes its modifications to the page (diffs) and sends them to the home node, which applies them immediately. The other possible copies of the modified page are invalidated by sending invalidation messages to the nodes holding the copies. Later, on a page fault following such an invalidation, the faulting nodes fetch the whole page from the home node.

In the original HLRC protocol, the replicated pages are lazily invalidated at the acquire operation following the critical section in which a page was modified. In contrast, we consider an eager variant of HLRC, in which invalidation messages are eagerly (i.e., immediately) sent out to the nodes holding copies of the modified pages. This variant is simpler, since it avoids the need for timestamps for page version handling. A multithreaded version of this eager variant of HLRC, called HBRC [START_REF] Antoniu | Implementing multithreaded protocols for release consistency on top of the generic DSM-PM2 platform[END_REF] (Home-Based Release Consistency), has been previously designed and implemented on top of the DSM-PM 2 generic implementation platform [START_REF] Antoniu | DSM-PM2: A portable implementation platform for multithreaded DSM consistency protocols[END_REF] for DSM protocols. We used this protocol as a starting point and derived a new, hierarchy-aware protocol, which we implemented and evaluated using the DSM-PM 2 platform.

The efficiency of the HBRC protocol is limited by two key factors. First, we note that in "traditional" implementations of distributed locks, the scheduling policy used to handle lock requests does not take into account locality [START_REF]Algorithms for Mutual Exclusion[END_REF][START_REF] Wu | An efficient distributed tokenbased mutual exclusion algorithm with central coordinator[END_REF]. Our experience shows that using locality-based lock acquisition scheduling can significantly improve the overall efficiency. The second key factor is related to the management of the invalidation acknowledgements following a lock release. In the HBRC protocol, upon receipt of diffs during a release operation, the home node sends out invalidation messages to all the nodes holding copies of the modified pages. In order to ensure memory consistency, the thread which releases the lock must not grant it to another thread before receiving all invalidation acknowledgements. This operation is system-wide, since acknowledgements may come from local nodes of the same cluster and from remote nodes of distant clusters. Thus, the delay for such an operation is limited by the latency of the intercluster communications.

The main contribution of this paper is to show how a hierarchy-aware approach can be used to tackle these two limitations.

High-latency inter-cluster communications

Hierarchy-Aware Synchronization

Let us consider Node 1 in Cluster A which has acquired a lock and is currently executing in critical section (see Figure 2). Now, Node 2 in Cluster B and then Node 0 in Cluster A want to acquire the same lock. A hierarchyunaware implementation of the distributed mutual exclusion may grant the lock to the first requester (remote Node 2), so we may incur two high-latency communications to exchange the lock between both clusters. In our hierarchyaware implementation of distributed mutual exclusion, the lock will be granted first to Node 0 in the local cluster A, even if the request from Cluster B arrived first. Thus, we incur one low-latency communication for the lock to travel within Cluster A plus one high-latency communication for the lock to go from Cluster A to Cluster B. We re-order the lock acquisitions at the cluster level to trade a high-latency message for a low-latency one.

We reproduce the same priority mechanism between the threads of a node as between the nodes of a cluster. Thus, we re-order the lock acquisitions at the node level to minimize the number of communications between nodes.

Partially Releasing Locks Within Clusters

In the flat version of HBRC, while releasing a lock, Node 1 must wait for all the page invalidation acknowl-edgements prior to granting the lock to another node. Yet, it is likely that the acknowledgements coming from the local cluster will arrive before those coming from remote clusters through a higher-latency network. So, in our hierarchical protocol, Node 1 will grant the lock to Node 0 (see Figure 2) in the local cluster A after receiving the acknowledgements from the local cluster, without waiting for those coming from remote clusters: we call that Partial Release.

Then, when Node 0 enters the critical section, its memory accesses are ensured to be consistent because the pages modified during the previous critical section have been invalidated. We are certain that the DSM pages on Node 0 have actually been invalidated, since Node 1 received the acknowledgements from all the nodes in its local cluster before granting the lock.

Thus, the lock can travel from node to node within the same cluster several times, without requiring receipt of the invalidation acknowledgements from remote clusters, without wasting time waiting for remote acknowledgements. As and when invalidation acknowledgements are received from remote clusters, the lock gets fully released on the nodes which successively acquired it in Cluster A, and a notification is propagated along that chain of nodes. The lock will not leave Cluster A until after being fully released by all the nodes of the chain.

Avoiding Useless Modification Propagations

In the flat version of HBRC, while releasing a lock, a thread systematically sends to the home nodes the modifications made on the DSM pages in the latest critical section. In contrast, in our hierarchical protocol, when a thread releases a lock, it does not send the modifications if another thread on the same node is granted the lock immediately.

That scheme is correct from the perspective of memory consistency because two threads on a node share the same address space. The modifications made by any number of threads which acquired the same lock successively on the same node will be sent all at once when the lock is granted to another node.

In Section 2.3, we saw that a thread on a node prefers granting a lock to another thread on the same node rather than to another node. That priority mechanism makes it more frequent for a thread to grant a lock to another thread on the same node. Therefore, it exhibits more situations where it is useless to send the modifications to the home nodes.

Handling the Lack of Fairness

As we saw in Section 2.3, our hierarchy-aware implementation of the distributed mutual exclusion prefers granting a lock to local threads and to local nodes: when two threads are in competition with each other to acquire a lock, the thread currently holding the lock will prefer granting it to another thread on the same node rather than to another node; similarly, when two nodes are in competition with each other to acquire a lock, the node currently holding the lock will prefer granting it to a node in the same cluster rather than to a node in a remote cluster.

That re-ordering of lock acquisitions may lead to unfairness. That lack of fairness may result in situations of starvation. For instance, a node in Cluster B may want to acquire a lock while some nodes in Cluster A also want to acquire the same lock indefinitely: if the lock is initially in Cluster A, then it will never travel to Cluster B because of the mechanism of priority.

To overcome that shortcoming, bounds have been set and can be tuned to limit the number of consecutive acquisitions of a lock by threads on the same node while other nodes have requested the lock; similarly, bounds have also been set and can be tuned to limit the number of consecutive acquisitions of a lock inside the same cluster while nodes in other clusters have requested the lock. Note that if those bounds are set to 1, then we have a flat implementation of distributed mutual exclusion, with no priority granted to local threads or local nodes; when the bounds are set to infinity, then fairness is not enforced any longer.

Thus, we trade fairness for performance in terms of execution time, without sacrificing correctness or leading to deadlocks.

Noteworthy Implementation Details

Conceptually, a list of requesting threads is attached to each lock. That list is ordered in function of the order in which the lock requests were received. Two counters are also attached to the lock: thread_privilege and node_privilege, initially set to zero.

When a thread wants to release a lock to exit a critical section, it searches the list of requesting threads for another thread on the same node. If such a requester is found, then the modifications made on the DSM pages are not sent to the home nodes (Section 2.5), and the lock is immediately granted to the local thread. If the new owner of the lock was not the first requester in the list, then the counter thread_ privilege is incremented by 1, meaning that this thread was selected to the detriment of the first requester of the list. When the counter thread_privilege reaches a limit max_tp (which can be tuned by the user application for more fairness), the priority to the local thread is not applied: the lock will be granted to a thread on another node if any, and the counter thread_privilege will be reset to zero.

In case the list does not contain a requesting thread on the same node, or if the counter thread_privilege reached its limit, then the modifications made on the DSM pages are sent to the home nodes. After receiving the invalidation acknowledgements from the nodes in the local cluster (Section 2.4), the thread exiting the critical section searches the list of requesters attached to the lock for a thread in the local cluster. If such a requesting thread is found, then it is granted the lock. If the new owner of the lock was not the first requester in the list, then the counter node_privilege is incremented by 1, meaning that this thread was selected to the detriment of the first requester of the list. When the counter node_privilege reaches a limit max_np (which can be tuned by the user application for more fairness), the priority to the local node is not applied: the lock will be granted to a thread in another cluster, and the counter node_privilege will be reset to zero.

Preliminary Performance Evaluation

This section presents the experiments we carried out to quantify the performance gains due to hierarchy-aware synchronization and to the hierarchical consistency protocol. Those experiments were run on the platform described in Section 2.1: the PCs we used were PentiumII's at 450 MHz under Linux 2.2.18, connected over a fullyswitched FastEthernet network (for the inter-cluster links) and equipped with relatively old SCI cards (type D310) for the intra-cluster links. On this particular platform, we observed an inter-cluster FastEthernet/TCP latency of 100 µs and an intra-cluster SCI/SISCI latency of 8 µs. Thus, the ratio of latency was around 12.

Our experiments have been conducted using the DSM-PM 2 experimental implementation platform for multithreaded DSM consistency protocols. This user-level platform provides basic building blocks, allowing for an easy design, implementation and evaluation of a large variety of multithreaded consistency protocols within a unified framework. It relies on the PM 2 (Parallel Multithreaded Machine, [START_REF] Namyst | PM2: an environment for a portable design and an efficient execution of irregular parallel applications[END_REF]), a runtime system for distributed, multithreaded applications. PM 2 provides a POSIX-like programming interface for thread creation, manipulation and synchronization in user space, on cluster architectures. PM 2 is available on most UNIX-like operating systems, including Linux and Solaris. For network portability, PM 2 uses a communication library called Madeleine [START_REF] Aumage | Madeleine II: A portable and efficient communication library for high-performance cluster computing[END_REF], which has been ported on top of a large number of communication interfaces: highperformance interfaces, like SISCI/SCI and VIA, but also more traditional interfaces, like TCP and MPI. DSM-PM 2 inherits this portability, since all its communication routines rely on Madeleine. Communication operations provided by Madeleine are guaranteed to be reliable (no message loss). We used the version 3 of Madeleine, which copes with hierarchical, network-heterogeneous clusters. This feature allowed us to perform our experimental evaluation without any network-specific code at the protocol level.

Hierarchy-Aware Synchronization

The table below demonstrates the performance gain obtained from our hierarchy-aware implementation of distributed mutual exclusion using a synthetic application. That program runs on four nodes connected over our SCI network in a single cluster: each node runs four threads. Each thread executes an empty critical section 10,000 times using a unique lock: executing an empty critical section consists in acquiring the lock and releasing it immediately. Varying the parameter max_tp which limits the number of consecutive acquisitions of a lock by the threads of a node (Section 3), we quantify the effect of the priority given to local threads for granting locks. We measured the time T flat it takes to execute our synthetic application using a flat version of our implementation of mutual exclusion, with no priority given to local threads. Then we measured the time T hierarchy it takes to execute the same program using our hierarchy-aware implementation of mutual exclusion for different values of max_tp.

The time ratios T flat T hierarchy given in the table show the speedups due to hierarchy-awareness in granting locks. As expected, the greater max_tp, the greater the speedup, because more threads can acquire the lock in a row on the same node, so less messages are exchanged between the nodes to transmit the lock.

Partially Releasing Locks Within Clusters

Figure 3 reports the performance gain obtained from the partial release mechanism using a synthetic application. The program runs on a varying number of clusters: each cluster has two nodes connected over SCI, and the clusters are interconnected through a fully-switched FastEthernet network as described in Section 2.1. There is just one thread per node executing 10,000 critical sections consisting in acquiring a unique lock, incrementing a unique shared integer and releasing the lock.

We measured the time it takes to execute a single critical section in our synthetic application without the partial release, i.e., waiting for all the invalidation acknowledgements before releasing a lock. Then we measured the time it takes to execute a single critical section in the same program using our hierarchical protocol with the partial release. Both measurements were performed without any limit on the number of consecutive acquisitions of a lock inside a cluster (max_np = ∞). different numbers of 2-node clusters. The hierarchical protocol performs 3 times as fast as the flat protocol with 5 clusters, and 4 times as fast with 6 clusters: the more we have clusters, the more we have high-latency links and the more our hierarchical protocol can take advantage of the partial release. Indeed, as the number of clusters grows, the flat protocol will be more likely to wait for remote acknowledgements, while our hierarchical protocol does not need to wait for the remote acknowledgements.

Avoiding Useless Modification Propagations

The table below demonstrates the performance gain obtained from not sending modifications to the home node when a thread exiting a critical section grants the lock immediately to another thread on the same node. We use the same configuration as in Section 4.1: four nodes connected over SCI in a single cluster, each of which running four threads. Each thread executes 10,000 critical sections consisting in acquiring a unique lock, incrementing a unique shared integer and releasing the lock. We measured the time T flat it takes to execute our synthetic application, with a flat protocol which systematically sends the modifications to the home node at the release operation. Then we measured the time T hierarchy it takes to execute the same program using our hierarchical protocol, i.e., without sending page modifications when a thread exiting a critical section grants the lock immediately to another thread on the same node. The table gives the time ratios T flat T hierarchy for different values of max_tp: the higher the priority to local threads, the greater performance gain our protocol achieves by not sending the modifications systematically, but the less fairness is enforced. Once again, our hierarchical protocol exchanges less messages between the nodes and waits less often for invalidation acknowledgements.

Conclusion

Our objective is to obtain high performance from DSM systems on large, hierarchical architectures, typically constellations of loosely-connected clusters of tightlyconnected nodes running multithreaded programs. We claim that this goal cannot be reached without considering this hierarchical architecture at the very core of the design. Some work has already been carried out regarding the management of data, more precisely caching the diffs at cluster-level in a Lazy Release Consistency protocol [START_REF] Arantes | An effective logical cache for a clustered LRC-based DSM system[END_REF]. In this paper, we have explored an alternative, complementary approach: managing synchronization in a hierarchy-aware manner, by taking into account the lower communication latency between partners located closer within the hierarchy. Our contribution is twofold. First, we propose a hierarchyaware approach to distributed synchronization. Second, we introduce the concept of Partial Release for locks, which allows consistency protocols to efficiently exploit the earlier delivery of acknowledgements issued by closer partners.

We believe that this concept of partial release is general enough to be applied to other synchronization objects than locks, such as semaphores or monitors. However, it cannot be used in conjunction with barriers: by definition, no thread can exit a barrier before all the participating threads all over the system have synchronized with each other. We also suggest that the partial release concept could be applied to any Eager Release Consistency protocol, where the thread exiting a critical section must wait for some kind of acknowledgements before actually releasing the lock.

Our preliminary experiments with micro-benchmarks demonstrate significant improvements in terms of performance. Unfortunately, the experiments could only be run on a platform far too small with respect to modern, largescale architectures. Also, we still need to carry out further tests using realistic applications such as Splash-2 programs [START_REF] Woo | The SPLASH-2 programs: Characterization and methodological considerations[END_REF]. We anticipate that lock-intensive applications will especially take advantage of our specific partial release implementation. Among the Splash-2 applications, Ocean and Cholesky look particularly favorable.

We anticipate that the greater the ratio between inter-and intra-cluster latencies, the better performance gain our solution will yield. For instance, with recent hardware, the ratio of latency between Local-Area and System-Area communications can exceed 20; between Wide-Area and Local-Area communications, the ratio of latency can be as high as 500.

Our work focused on synchronization locality, as op-posed to the work done by Arantes et al. [START_REF] Arantes | An effective logical cache for a clustered LRC-based DSM system[END_REF], which concentrates on data locality. It would be interesting to consider merging these two approaches to add up their respective performance improvements.

Figure 2 .

 2 Figure 2. Reordering of lock acquisitions and partial release.

Figure 3 Figure 3 .

 33 Figure3reports the performance gain obtained from the partial release mechanism using a synthetic application. The program runs on a varying number of clusters: each cluster has two nodes connected over SCI, and the clusters are interconnected through a fully-switched FastEthernet network as described in Section 2.1. There is just one thread per node executing 10,000 critical sections consisting in acquiring a unique lock, incrementing a unique shared integer and releasing the lock.We measured the time it takes to execute a single critical section in our synthetic application without the partial release, i.e., waiting for all the invalidation acknowledgements before releasing a lock. Then we measured the time it takes to execute a single critical section in the same program using our hierarchical protocol with the partial release. Both measurements were performed without any limit on the number of consecutive acquisitions of a lock inside a cluster (max_np = ∞). Figure3shows the timings for