Angelo Luiz

Olivier Steffenel

Andrea Schwertner Flauzac

Patricia Pitthan Charão

Benhur Barcelos

Guilherme Weigert Stein

Sergio Cassales

J Nesmachnow

Matías Rey

Manuele Kirsch Cogorno

Pinheiro

L A Steffenel

O Flauzac

A S Charao

P P Barcelos

B Stein

G Cassales

S Nesmachnow

J Rey

M Cogorno

M Kirsch-Pinheiro

C Souveyet

MAPREDUCE CHALLENGES ON PERVASIVE GRIDS

Keywords: MapReduce, Fault-Tolerance, Pervasive Distributed Computing

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

INTRODUCTION

One of the first challenges a user faces when deploying MapReduce is that its most known and popular implementation, AH (2014a), requires a highly structured environment such as a dedicated cluster or a cloud infrastructure to be deployed. Indeed, Hadoop has been designed to be deployed over a dedicated cluster or cloud computing infrastructures such as AEMR (2014). Hadoop relies on a collection of tools (Hadoop Core, HDFS, etc.) developed by different Apache subprojects, which interact through a complicate set of master and slave daemons. As a result, Hadoop installation, although well documented, requires a stable set of computer nodes that shall be known at startup time. The installation procedure also lacks of automatic context adaption, forcing the administrator to manually define the characteristics of each resource, such as the number of cores, their relative speed or the available memory.

Together, these elements prevent a user to quickly launch MapReduce over a set of unused resources (e.g., the enterprise workers' desktops), at least not without a previous effort to prepare and configure the nodes. Indeed, even if Hadoop is now a popular data analysis tool, several companies/organizations do not have a dedicated infrastructure, as sometimes the demand for computing intensive tasks is punctual or executed only JCS Science Publications periodically. In these cases, cloud computing infrastructures represent a popular alternative for usaing MapReduce without a dedicate infrastructure. Unfortunately, several enterprises fear (or are forbidden) to distribute sensible data over the cloud. These externalized infrastructures still suffer from security issues that prevent their application in some cases. We believe that, considering the extreme development of mobile devices and networks inside organizations nowadays, the opportunistic use of existing resources as an internal pervasive grid represents an interesting alternative for those who hesitate to keep a dedicated infrastructure or to rent cloud computing resources. Nevertheless, in order to be fully operational, pervasive grids environments have to first tackle problems related to their dynamic nature where nodes join and leave the network dynamically.

Our project is precisely addressing this point: Proposing scalable techniques to support existing Hadoop applications in the context of loosely coupled networks such as pervasive grids. We consider pervasive grids as a large-scale infrastructure with specific characteristics in terms of volatility, reliability, connectivity, security, etc. According to Parashar and Pierson (2010), pervasive grids represent the extreme generalization of the grid concept, in which the resources are pervasive. For these authors, pervasive grids seamlessly integrate pervasive sensing/actuating instruments and devices together with classical high performance systems. In the general case, pervasive grids rely on volatile resources that may appear and disappear from the grid, according their availability. Indeed, mobile devices should be able to come into the environment in a natural way, as their owner moves (Coronato and De Pietro, 2008) and devices from different natures, from the desktop and laptop PCs until the last generation tablets, should be integrated in seamlessly way. It results an environment characterized by three main requirements:

• The volatility of its components, whose participation on the grid is notably a matter of opportunity and availability • The heterogeneity of these components, whose capabilities may vary on different aspects (platform, OS, memory and storage capacity, network connection, etc.) • The dynamic management of available resources, since the internal status of these devices may vary during their participation into the grid environment

The following scenario can illustrate these requirements: Let us consider that, during the execution of a job, a mobile device becomes available, integrating the pervasive grid. After its integration, the same device may change its network connection, passing from a fixed connection to a wireless one. The same can be observed with the available memory: After starting a job, device's owner may start new applications that modify device memory status. All these changes have an impact on the job performance and consequently on the availability of this device for the pervasive grid.

Pervasive grids environments have to deal with such additional constraints related to the heterogeneity and the volatility of the resources. In such environments, it is essential to adapt the application to the network variable behavior and to coordinate the resources (task scheduling, data placement, etc.). According to Coronato and De Pietro (2008), pervasive grid environments should be able to self-adapt and selfconfigure in order to incoming mobile devices. We strongly believe that context-awareness is needed in order to support such self-adaption. Context-awareness can be defined as the ability of a system to adapt its operations to the current context, aiming at increasing usability and effectiveness by taking environmental context into account [START_REF] Baldauf | A survey on context-aware systems[END_REF]. In order to support environments changes, context-awareness becomes a critical aspect to boost the efficiency of the applications over pervasive grids.

Such dynamic nature of pervasive grids represents an important challenge for executing MapReduce applications over these environments. Context-awareness and nodes volatility become key aspects for successfully executing such applications over pervasive grids.

This work presents the first results of the PER-MARE (2014), whose goal is proposing scalable techniques to support MapReduce in pervasive grids. PER-MARE proposes to fully explore the potential of unused (or underused) resources at enterprises as pervasive grids for MapReduce applications. Our challenge is to adapt MapReduce to these dynamic grids. For this, we focus on the volatility and heterogeneity of the available resources through two complementary approaches: On the one hand, we propose to improve Hadoop with context-awareness and more fault-tolerance concerns; on the other hand, we propose an alternative pervasive grid implementation based on a P2P computing middleware, fully adapted to these dynamic environments. This study demonstrates this vision by its first results, organized in three complementary sections: JCS Science Publications (i) A context-aware scheduler for Hadoop; (ii) a faulttolerant job tracker for Hadoop; and (iii) a pervasive grid implementation of MapReduce.

The remain of this study is structured as follow: Section 2 introduces basic notions of MapReduce and its Hadoop implementations. Section 3 presents related works focusing on context-aware and volatility issues on MapReduce. Section 4 gives an overview of the PER-MARE vision, before presenting our contributions in next sections. Section 5 presents a context-aware scheduler for Hadoop, while on section 6 proposes introducing more fault-tolerance for Hadoop. Section 7 presents a pervasive grid implementation. Section 8 discusses obtained results and remaining opening issues, before concluding.

MAPREDUCE/HADOOP BASICS

MapReduce [START_REF] Dean | Mapreduce: Simplified data processing on large clusters[END_REF]) is a parallel programming paradigm successfully used by large Internet service providers to perform computations on massive amounts of data. This model is currently becoming popular as a solution for rapid implementation of distributed data-intensive applications. The key strength of the MapReduce model is its inherently high degree of potential parallelism that should enable processing of petabytes of data in a couple of hours on large clusters consisting of several thousand nodes.

A MapReduce computation takes a set of input key/value pairs and produces a set of output key/value pairs. The user of the MapReduce paradigm expresses the computation through two functions: MapReduce most known implementation is AH (2014a). This framework takes care of splitting the input data, scheduling the jobs' component tasks monitoring them and re-executing the failed ones. Currently, two Hadoop versions are available, both organized as two superposed entities: A 'MapReduce' engine and a distributed file system, named HDFS. The engine provides the ability to execute map and reduce tasks across the cluster and reports results, while the distributed file system provides a storage scheme that is able to replicate data across nodes for processing.

On the Hadoop 1.x architecture, these entities are organized in a master-worker pattern (Fig. 1), with two different masters (JobTracker and NameNode) and workers (TaskTrackers and DataNodes, respectively).

When a client launches an application on a Hadoop 1.x cluster, the request is initially managed by the JobTracker. The JobTracker collaborates with the NameNode in order to distribute the work tasks as closely as possible to the data on which it will work. Indeed, the NameNode act as the HDFS master, providing metadata services for data distribution and replication. The JobTracker, on its side, coordinates the scheduling of map and reduce tasks into available slots managed by the TaskTrackers. Each TaskTracker executes map and reduce tasks on data from its DataNode, which represents the HDFS slave. When the tasks are complete, the TaskTracker notifies the JobTracker, which identifies when all tasks are complete and eventually notifies the client of job completion.

Due to its centralized architecture, Hadoop 1.x is particularly vulnerable to failures on its master nodes. While failures on the TaskNode and on the NameNode will be supported by Hadoop 1.x, failures on the JobTracker and on the NameNode will compromise job execution. Besides, as illustrated by Fig. 1, Hadoop 1.x is designed considering a single client application at time, which leads to an underexploitation of the resources.

The new Hadoop 2.x version overcomes this 'simgle client' issue, by replacing the initial MapReduce engine by a new one, called YARN. YARN opens Hadoop to the possibility of managing multiple applications and computing models and notably to the deployment of non-MapReduce applications. For this, YARN replaces JobTracker and TaskTracker by a new set of entities that are independent of the application.

On the top of YARN daemons (Fig. 2), we found the ResourceManager, which is in charge of managing the entire cluster and of assigning applications to the underlying compute resources. These resources are controlled by the NodeManagers. When a new job is submitted, the ResourceManager delegates the job supervision to an ApplicationMaster, which executes the tasks in abstract Containers controlled by the NodeManagers. Thus, multiple client applications may execute concurrently, sharing cluster resources. JCS Unfortunately, Hadoop 2.x still adopts a master/worker architecture, making it as vulnerable as Hadoop 1.x to master (ResourceManager and NameNode) failures. This limitation may considerably affect its performance on pervasive grids, since these nodes may leave the grid during the job execution. Actually, neither Hadoop 1.x nor Hadoop 2.x are able to manage nodes volatility required by pervasive grids. Nodes heterogeneity is not observed either. Both Hadoop 1.x and Hadoop 2.x consider nodes with similar characteristics when managing job execution.

Science Publications

RELATED WORKS

This section reviews the main related works on context-awareness on Hadoop and MapReduce implementations on pervasive grids.

Context-Awareness

Because Hadoop performance is tightly dependent on the computing environment but also on the application characteristics, several researchers focused on bringing context-awareness to Hadoop and can be roughly JCS Science Publications classified in three categories: Job schedulers, task schedule optimizers and resource placement facilitators.

In the first category, we find works like [START_REF] Kumar | CASH: Context aware scheduler for Hadoop[END_REF], [START_REF] Tian | A dynamic MapReduce scheduler for heterogeneous workloads[END_REF] or Rasooli and Down (2012). These works assume that most jobs are periodic and demand similar CPU, network and disk usage characteristics. As a consequence, these works propose classification mechanisms that first analyze both jobs and nodes with respect to its CPU or I/O potential, allowing an optimized matching of applications and resources when a job is submitted. Similarly, [START_REF] Isard | Quincy: Fair scheduling for distributed computing clusters[END_REF] propose a generic solution that uses the distribution of resources as context information maps in a capacity-demand graph, calculating the optimum scheduling from a global cost function and with the objective of improving general cluster performance.

While the previous works focus on the improvement of the overall cluster performance through an offline knowledge about the applications and the resources, sometimes this is not enough to ensure a smooth operation. For instance, works like [START_REF] Zaharia | Improving MapReduce performance in heterogeneous environments[END_REF] and [START_REF] Chen | SAMR: A Self-adaptive MapReduce scheduling algorithm in heterogeneous environment[END_REF] focus on improving tasks deployment inside a job as a way to reduce the response time in large clusters, executing many jobs of short duration. Using the environment context information and the job's estimated time to end, these works rely on heuristics to make the connection between elapsed time and a score that represents how much of the job has already been processed. This information is used to generate a threshold, which will determine when a task is slow enough to start a new speculative copy on another possibly faster machine. [START_REF] Chen | SAMR: A Self-adaptive MapReduce scheduling algorithm in heterogeneous environment[END_REF] also uses historical execution traces to improve its predictions.

Finally, works like [START_REF] Xie | Improving MapReduce performance through data placement in heterogeneous Hadoop clusters[END_REF], aims to provide better performance on jobs through better data placement, using mainly the data locality as decisionmaking information. The performance gain is achieved by the data re-balancing in nodes, leaving faster nodes with more data. This lowers the cost of speculative tasks and also of data transfers through the network.

From the analysis of these works, we observe that most of them rely on the categorization of jobs and nodes, which is hard in a dynamic environment like a pervasive grid. Even when runtime parameters such as elapsed time or data placement are considered, they assume a controlled and well-known environment. Because these assumptions are too restrictive, these works fail on responding to the requirements of a pervasive grid environment.

Fault-Tolerance and Volatility

In pervasive grids, nodes have to face both the dynamicity of the resources availability but also the own nodes volatility, where any node can fail/disconnect during the application execution.

To respond to the volatility of nodes and improve fault-tolerance of the execution environment, two main approaches has been considered: (i) To harness Hadoop so that it will be able to support a wider range of failure scenarios; or (ii) to implement the MapReduce paradigm on the top of another middleware support. Hadoop includes basic fault-tolerance techniques, data replication and speculative execution of tasks, to minimize the impact of workers failures, but these techniques are not enough to ensure the operation when running on a true dynamic environment like pervasive grids.

When dealing with Hadoop, one of the elementary problems comes from its master/worker architecture. Hadoop was conceived to tolerate worker's failures through node supervision (heartbeats) and speculative task execution, but a failure at the master level forces the reboot of the entire network. While it is possible to improve fault-tolerance at the file system level by using NameNode replication at the HDFS level or integrating another file system like AC (2014), we observe that, at the job/task level, there is a remaining single point of failure, the JobTracker node. To our knowledge, only the commercial solution MapR (2014) provides faulttolerance at the JobTracker level, but the details about these solutions are not freely available.

Most of the initiatives to improve MapReduce faulttolerance prefer to rely on other middleware environments. Indeed, the wide acceptance of Hadoop somehow hides the fact that MapReduce can be implemented on the top of other computing middleware systems. Due to the simplicity of its processing model (map and reduce phases), data processing can be easily adapted to a given distributed middleware, which can coordinate tasks through different techniques (distributed task schedulers, work-stealing/bag of tasks, etc.). [START_REF] Lin | Moon: Mapreduce on opportunistic environments[END_REF] propose a system called MapReduce On Opportunistic eNvironment (MOON), which extends Hadoop in order to deal with the high unavailability of resources. MOON relies on a hybrid architecture, where a small set of dedicated nodes are used to provide resources with high reliability, in contrast to volatile nodes, which may become inaccessible during computations. One inconvenient of this system is that in spite of its improved fault-tolerance, nodes must be known in advance, i.e., no new node can join the network. JCS Science Publications Also, [START_REF] Tang | Towards MapReduce for desktop grid computing[END_REF] propose a system designed to support MapReduce applications, by exploiting the BitDew middleware [START_REF] Fedak | BitDew: A programmable environment for large-scale data management and distribution[END_REF], which is a programmable environment for automatic and transparent data management on desktop grids (a particular case of pervasive grids aiming at the leverage of unused processing cycles and storage space available within the enterprise). Unfortunately, this study also requires the presence of a stable master node that runs BitDew services and coordinates data/task distribution.

In P2P-MapReduce (Marozzo et al., 2010;2012), the authors present a distributed architecture implemented in JXTA and following the super-peer approach, where the super-peers serve as cache data server, handle jobs submissions and coordinate execution of parallel computation.

P2P-MapReduce supports node disconnection and the integration of new nodes, but is partially dependent on Hadoop as it relies on the Hadoop execution engine to execute the applications. Because this system does not integrate a file system coupled with the task manager, the transmission of data is made together with the tasks to compute. While this study has similar objectives to our own, its architecture is complex and presents several dependencies, which may prevent an easy deployment over pervasive grids. We also believe that performance can be drastically improved by implementing a lightweight execution stack independent of Hadoop and by improving task scheduling through the use of both context and data placement information.

Even if works cited above have improved faulttolerance for MapReduce applications, they still present some limitations when considering pervasive grids and notably the need for small set of stable nodes and a complex architecture (which make difficult their application to heterogeneous resources). The deployment of MapReduce over pervasive grids remains then an open question, since there is no single solution, for the moment, that solves all previously mentioned issues together. We believe that, when considering pervasive grids, where heterogeneity is a major characteristic, data processing/scheduling must be driven by contextual information (resources characteristics, node reliability, network performance, data location) in order to achieve the expected processing performance.

THE PER-MARE VISION

Given the problems presented above, we propose to address the lack of context adaptation of MapReduce applications over pervasive grids all while keeping the compatibility with MapReduce most popular implementation, Hadoop. To meet this global goal, we started the PER-MARE project, which objective is to study the adaptive deployment of MapReduce-based applications over pervasive and desktop grid infrastructures.

Our approach is to improve the behavior of MapReduce applications on pervasive grids using a twofold investigation method. Hence, to better understand the elements that may impact the deployment of MapReduce over pervasive grids, our teams investigate the problem through two different approaches: On the one hand, we try to modify Hadoop in order to implement on it a context-aware scheduling and to improve fault-tolerance, both with the objective to enhance Hadoop against heterogeneity, dynamicity and volatility of the nodes. On the other hand, the second approach relies on the porting of the MapReduce paradigm (and the Hadoop API) over a P2P distributed computing middleware. Because this platform is already adapted to dynamic and volatile networks, it may represent a good alternative for applications implemented with Hodoop's API as shown in Fig. 3.

We believe that this double approach is essential to understand and cover all the facets of the pervasive grid challenges. By comparing these two approaches "side by side" we can propose effective solutions and provide important insights on the adaptability to the heterogeneity of resources and the dynamic nature of the networks. Thus, our vision, illustrated on Erreur ! Source du renvoi introuvable., considers the Hadoop API as common access point for MapReduce applications that will be able to fully exploit resources on pervasive grids through two different implementations, based on a context-aware improved Hadoop implementation, or on a pervasive grid solution.

Next sections describe the first results of our work. Firstly, we improved current Hadoop implementation with a context-aware scheduler, allowing Hadoop to observe real characteristics of the available nodes, instead of static configuration of such nodes. This represents a first step towards a better support of heterogeneous environments. Secondly, we propose a fault-tolerant implementation of Hadoop that is able to replicate its master node in order to prevent crashes due to nodes failures, a necessary step to fully handling nodes volatility on Hadoop. While both solutions extend Hadoop, the third contribution presented in this study considers a completely different implementation, based on a P2P middleware. This latter focuses on the volatility of nodes, allowing new nodes to join or to leave a pervasive grid without stopping job execution. JCS Science Publications

Fig. 3. Overview of PER-MARE vision

CONTEXT-AWARENESS ON HADOOP

As dedicated and cloud computing infrastructures have leveraged the use of MapReduce, it is natural that the most known MapReduce distribution has been tailored to such environments. For instance, most IaaS clouds use sets of virtual machines that share similar characteristics, such as computational power and memory. In such cases, MapReduce does not require specific adaptation to the computational context, as all virtual machines are similar. As a consequence, most users simply rely on MapReduce default configurations such as the number of reduce tasks by machine, the maximum memory, etc. Although this behavior can be modified through property files, there is no mechanism to automatically detect and modify these parameters. When dealing with a heterogeneous environment such as a pervasive grid, MapReduce must be able to automatically tune to the nodes characteristics.

While the computational context is tightly related to the processing power of the resources, it also impacts other aspects such as fault-tolerance and data storage. Indeed, Hadoop allows a certain number of duplicated processes/data in order to circumvent fault situations. If the context on pervasive grid is not considered, tasks may be inefficiently allocated or even disappear if the node volatility is high.

Similarly, HDFS tries to place data for the map and reduce phases as closes as possible to the processes/tasks that will need it, in order to reduce the (slow) access over the network. In a pervasive grid, the placement policy must account also on the volatility and speed of the resources, preventing data losses. While the contextual information required for the adaptation of MapReduce can be obtained from the system properties (CPU and network speed, number of cores, memory size, etc.), the diffusion and analysis of such information must be tightly integrated into the MapReduce framework to boost the platform efficiency. For this reason, context-awareness [START_REF] Preuveneers | Context-Aware Adaptation in an Ecology of Applications[END_REF] and context distribution [START_REF] Kirsch-Pinheiro | Context Grouping Mechanism for Context Distribution in Ubiquitous Environments[END_REF]) are important elements to be considered.

Currently, the Apache Hadoop framework scheduler is mostly designed for homogeneous environments in which nodes characteristics are provided at startup in a static way. This section focuses on improving Hadoop scheduler mechanism in order to make it more contextaware towards resources on the cluster.

Hadoop Schedulers

Hadoop scheduling has evolved along its versions. On Hadoop 1.x, the default scheduler was designed for supporting a batch job submission, organizing jobs in a queue. Similarly, Fair Scheduler (AH, 2013), also proposed on Hadoop 1.x, considers with the same input data size, using a two level scheduling in order to distribute the resources equally: A superior level that allocates queues for each user, using a weighted fair algorithm; and a second level that allocates the resources inside each user queue.

Hadoop 2.x adopts, as a default scheduler, a more sophisticated scheduler, the Capacity Scheduler [START_REF] Ah | Hadoop MapReduce next gen-erationcapacity scheduler[END_REF]. This scheduler considers a shared Hadoop environment across multiple partners. It focuses on guarantees that a minimum share will always be available for each partner. The benefit comes from the fact that different organizations have processing peaks at different times, therefore the organization (partner) using more capacity, will use the idle capacity of the other organizations. This scheduler tracks the resources registered within the ResourceManager and monitors which resources are free and which are being used.

While these schedulers include some basic awareness about the nodes capacity, we observe that this information is often ignored due to a poor startup configuration. Indeed, Hadoop is heavily dependent on XML files provided at startup and ideally every node should provide its own XML files with tailored parameters to express the node capacity. In a large JCS Science Publications heterogeneous cluster, modifying each node configuration can be very time consuming since each node will have a different configuration. Also, the parameters from a XML file are static and any evolution of nodes capacity will not be considered till node reboot.

It is then clear that we must improve the way Hadoop detects and handles context information from the execution environment, by providing updated information about the node capacity.

In order to detect the node capacity, we chose to integrate a context collector into Hadoop, allowing an automatic detection of each node capacity. This information is given to Hadoop scheduler, which can scale the allocation limits in function of the real cluster resource availability. In a first moment, this scaling affects the containers allocation as a function of the available memory and computing cores, impacting therefore on the choice of tasks placement and how speculative task are started. Also, by adapting the capacity to the cluster real resource, no resource would be wasted or left inactive while the scheduler is making tasks wait due to wrong information being received.

Collecting Context Information

To include context information on Hadoop, we integrated a collector module based on standard Java monitoring API (Oracle, 2014), which allows to easily access the real characteristics of a node, with no additional libraries required.

The collector module, illustrated by Fig. 4, allows collecting different context information, such as the number of processors (cores) and the system memory, using a set of interface and abstract/concrete classes that generalize the collecting process. Due to its design, it is easy to integrate new collectors and improve the resources available for the scheduling process, providing data about the CPU load or disk usage, for example.

Context information is described by using a predefined name and a description. Such name corresponds to a given concept identified in the context ontology. This model, inspired from [START_REF] Kirsch-Pinheiro | Representing Context for an Adaptative Awareness Mechanism[END_REF], considers context information as an element (a context element), for which multiple values can be observed. Context ontology allows them to semantically describe such element, while the description gives a human readable definition for it.

This collector module was integrated to the NodeManager daemon, since this entity is in charge of processing tasks and managing node definition. In this first prototype, we only collect node capacity (available memory and number of cores) and this information is then sent to the ResourceManager. As a consequence, the information from the context collector module allowed us to improve the Hadoop scheduler operation without having to modify its implementation. This is especially interesting as further works will be able to compare other schedulers from the literature without having to modify their implementation.

Experiments Description

In order to evaluate our proposal, we compare the container allocation pattern in the original Capacity Scheduler with our context-aware Capacity Scheduler. We perform the analysis of the container allocation pattern when executing the TeraSort algorithm with a 5GB dataset to sort. By requesting enough containers (from nodes expressed capacities) and providing enough data to stress the cluster, we aimed at comparing how the context information influences the containers (tasks) allocation and the overall execution time.

The experiments were performed in a cluster subset of the Grid '5000 http://www.grid5000.fr computing environment. The subset had five nodes, one master and four slaves, each node having the following configuration: 2 CPUs AMD@1.7GHz, 12 cores/CPU and 47GB RAM. All nodes were running an Ubuntu-x64-1204 standard image, with Sun JDK 1.7. The Hadoop distribution was the 2.2.0 YARN version.

For unmodified Capacity Scheduler, we adopt the default Hadoop configuration, which defines, on yarndefault.XML, the memory and CPU properties with default values of 8192 and 8 respectively. In the case of the context-aware scheduler, the same properties are obtained from the context collector, overwriting the default parameters from the XML configuration files, resulting in the parameters from Table 1.

Results and Interpretation

The following charts are consolidated by resources represented by NodeManagers. As stated before, the containers are allocated to a given NodeManager and tasks are executed inside these containers. Please note that each segment represents the tasks that are currently being executed on the node, so the end of a segment indicates the completion of some of the tasks (which can still be present in the next segment).

Figure 5a portraits the execution of the TeraSort with original Capacity Scheduler. It is easy to notice that some containers had to wait for the completion of others in order to start processing their tasks. JCS

Science Publications

Indeed, Hadoop splits the work in 38 Map tasks (numbered 2-39), which are distributed to the nodes according to the known resource capabilities. When the first tasks are completed, new tasks are provided to the nodes, if any available (as illustrated in Fig. 5a, where tasks 32-39 represent the second execution wave).

If no more tasks are available, the nodes may wait for the job completion of be given additional tasks for speculative execution (for example, tasks numbered 40-44). Because speculative tasks depend on the estimated advancement of current tasks, they usually concern the first tasks deployed and do not contribute to accelerate the execution in a homogeneous cluster. Indeed, we observe that the speculative tasks from Fig. 5a do not help improving the execution time as these tasks correspond to tasks from the first wave scheduled on nodes stremi-42 and stremi-44.

Figure 5b portraits the execution of the TeraSort algorithm with context-aware Capacity Scheduler. In this case the overall completion time was reduced, due to the fact that all containers could be started right after the arrival of the request, thanks to the higher resource availability. Without the context information, the scheduler uses the default minimum parameters for the nodes capacities, causing a bad execution performance.

After an analysis and comparison of both charts, we also notice that the default scheduler launches speculative containers 41-43 on node stremi-5 and container 44 on node stremi-42, while the context-aware chart has only the standard containers, which are numbered 2-39. This happens because these extra containers are, in reality, speculative tasks launched because other tasks were taking too long to finish.

To better understand the impact of context information on heterogeneous systems, we also performed a simulation of a heterogeneous cluster. Comparing to the previous experiments, the only difference here is that the nodes are purposely given false capacities when registered to the ResourceManager. Using these false values, a heterogeneous cluster will be simulated with the following capacities: ------------------------------------------------------------------------------------------ Figure 6 portraits the execution of the TeraSort algorithm within the simulated heterogeneous environment, also using context-aware Capacity Scheduler. Compared to the default case, the heterogeneous environment execution shows an improvement, but due to the lower cluster capacity, it is a slightly worse than the context-aware Capacity Scheduler executing on a homogeneous environment.

It is possible to note that the containers started the assignment with the node stremi-33, which is the node with the most capacity in the cluster and also was the first to be added in the node list. As in the other experiments, the scheduler launches containers on a node until its resources are all reserved, then move to the next node on the list.

This experiment shows that it is possible to use this context-aware in a heterogeneous environment, the allocations were adapted to a slightly smaller cluster if compared to the real environment. As a future work, it is possible to set the allocation limits in function not only of total cluster resources but also of each individual node resource capacity. JCS Science Publications

A FAULT-TOLERANT HADOOP IMPLEMENTATION

This section introduces our proposal of designing a fault-tolerant implementation of MapReduce. A review of current approaches for fault-tolerance in Hadoop is presented and our proposal for fault-tolerance in Hadoop based in replicating the JobTracker is described.

Fault-Tolerance on Hadoop

As previously stated, the basic design of Hadoop is still widely directed to cluster and grid computing platforms. In these environments, faults are a second concern as most nodes will operate flawlessly for a long time period. Of course, the failure of a node is still a concern, but some techniques like data replication and speculative execution of tasks may limit the danger on most scenarios.

Because Hadoop was developed to work on a cluster/cloud environment, it has several fault-tolerant mechanisms to circumvent the crash of workers nodes. On Hadoop 1.x this is all orchestrated by the JobTracker, which monitors the status of working nodes while the NameNode coordinates the data replication. HDFS allows the replication of the NameNode (through passive replication), but a failure at the level of the JobTracker forces a job to be restarted.

On Hadoop 2.x, part of the job management responsibility is transferred to the ApplicationMaster, which becomes a task manager. The loss of the ResourceManager does not blocks the execution of a job, only prevents new jobs to be submitted. However, the loss of an ApplicationMaster forces the restart of the job, just like on Hadoop 1.x.

In this section we concentrate on the enhancement of fault-tolerance on Hadoop 1.x (at the JobTracker level), but this solution can also be applied to Hadoop 2.x. We believe that this approach will lead us to improve the reliability of Hadoop, especially in the case of pervasive grids, whose volatility represents a main obstacle to the deployment of Hadoop.

Fault-Tolerance Through Replication of the JobTracker

As stated in the previous section, we want to develop fault-tolerance solutions that enable Hadoop to operate in pervasive environments, which means that we need to ensure the network would not collapse in the event of a JobTracker failure. The replication of the JobTracker (or the ResourceManager, in Hadoop 2.x) is the key, but several strategies can be applied to replicating, monitoring and resuming the JobTracker. In order to ensure high-availability to the JobTracker on a pervasive system, our solution needs to comply with the following properties:

• Fast recovery in the case of a failure • Small impact on the performance • Be able to adapt to the capacity and context of the nodes

The first two properties limit the number of techniques that can be employed. Indeed, a solution that uses an external persistent device would add a nonnegligible overhead to the operation and slow-down the recovery. The third property relates to the heterogeneity of the nodes and connections on a pervasive system: Without context-awareness, we risk to resume the JobTracker on a node without the performance or stability levels required for the role.

For all these reasons, we decided to implement JobTracker replication using AZ (2014). ZooKeeper is one of the tools developed initially inside Hadoop that become a full project as its application was extended to other applications. It provides efficient, reliable and fault-tolerant tools for the coordination of distributed systems. In our case, we use ZooKeeper services to storage snapshots of the JobTracker.

Snapshots are made on a per-attribute basis, where JobTracker attributes are stored in ZooKeeper znodes, as illustrated in Fig. 7. Depending on the importance of the snapshot, some attributes are replicated synchronously, while other attributed are replicated asynchronously (for example, the blacklist is synced asynchronously while the tasks statuses are synced synchronously). By modifying specific parts of the Hadoop code, we were able to insert snapshot triggers in critical events, minimizing the performance impact of the replication.

In addition, the distributed memory of ZooKeeper is used to coordinate the nodes in the case of a JobTracker failure: ZooKeeper keeps a synchronized ordered list of nodes in the system. This list is regularly updated so that volatile nodes are removed from the list and new nodes are inserted at the end of the list, while the node at the top of the list is current JobTracker node. This procedure naturally organizes the nodes by order of stability, but in the future we plan to modify the nod's order to adapt to the context or capacity of the nodes (for example, to avoid giving this role to an old and slow machine). JCS

Science Publications

Fig. 7. Replication with zookeeper

The JobTracker is monitored by regular heartbeats. When detecting a connection failure with the JobTracker, all TaskTracker automatically check for ZooKeeper to confirm the status of the JobTracker and/or get the new JobTracker address. Each TaskTracker checks the first node in the ZooKeeper list of nodes; if the TaskTracker is at the top of the list, it replaces the ancient JobTracker spawning a replica using the last snapshot, but if the TaskTracker is not the first node in the list, it tries to connect with this first node that should be the new JobTracker. If the connection fails, the TaskTracker checks the list of nodes again and repeats the process until it connects or becomes the new JobTracker. All these steps are carried using ZooKeeper services.

Validating the Prototype with Docker-Hadoop

To validate our solutions, however, we need to test different scenarios of node and network faults. Because these experiments require the execution (and reproduction) of well-defined scenarios, we rely on virtual machines (most specifically on container-based virtualization), which allow the researchers to control both as both system images and network interconnections.

In our experiments, we used Docker-Hadoop https://github.com/vierja/docker-hadoop as a testbed to simulate different failure scenarios. Indeed, thanks to Docker-Hadoop dashboard, one can easily switch-off or restart nodes in the environment and reproduce the same scenario at will. Using Docker-Hadoop dashboard allowed us to test different failure scenarios like:

Crash of the JobTracker Node

In this scenario, we kill the JobTracker in order to force a new node to resume the JobTracker role. When a TaskTracker loses connection with the JobTracker, it checks the list of Zookeeper nodes and it tries to connect or becomes the new JobTracker, as described in the previous subsection.

Restart of an Old JobTracker

In this scenario, we investigated the impacts of the return of an old JobTracker node. Two possibilities are analyzed: JCS Science Publications implementation as a way to keep compatibility with the Hadoop API. However, some of these classes were too dependent on inner elements of Hadoop, forcing us to develop our own equivalents, at least for the moment (further works shall reinforce the compatibility with Hadoop API). For instance, we had to substitute the OutputCollector class with our own MultiMap class, while the rest of the application remains compatible with both Hadoop and CloudFIT.

An initial example of MapReduce over CloudFIT was proposed, using the traditional WordCount application. As indicated previously, this first prototype organizes MapReduce in a two rounds execution, but one single difference between this implementation and the one using Hadoop resides on the need to indicate the number of computing tasks, called blocks. Indeed, this behavior is automatized on Hadoop, which tries to guess the required number of Map and Reduce processes. In our prototype, this parameter was defined as to mimic the behavior of Hadoop, i.e., by setting a number of Map tasks to roughly correspond to the number of input files and the number of Reduce tasks to correspond to the number of computing cores available on the CloudFIT network at the time Reduce starts (this number may varies later, due to nodes volatility).

Prototype Evaluation

The experiments were conducted over 16 machines on the Helios cluster from the Grid'5000 network. Each machine is composed by 2 AMD Opteron 275 2.2 GHz CPUs, totalizing 4 cores per node and a Gigabit Ethernet interconnects the nodes.

For the experiments, we evaluate the performance of both CloudFIT and Hadoop solutions when varying the total amount of data and the number/size of input files. For each data size, we measure 3 different input splits: One single file, 1MB splits and 512kB splits. The reason for such approach is to analyze the impact of the input files on the map step from both solutions. For the input data, we chose the Gutenberg Project Science Fiction Bookshelf CD http://www.gutenberg.org/wiki/Gutenberg:The_CD_and _DVD_Project, which contains more than 200 books in text format. The results presented on Fig. 8 represent the median of the performed measures for 16 nodes.

When analyzing the measures, two major scenarios arise: For small data volumes, our prototype largely outperforms Hadoop, while the difference tends to stabilize for large data sets. This is mostly due to CloudFIT lightweight middleware. Even though, the analysis of application and middleware traces shows that the replication pattern used on CloudFIT shall be improved if we want to achieve good performances. Indeed, currently we use a full-replication scheme, so that up to n-1 nodes can fail without losing the job progress. The inconvenient is that this overloads the network with results transfers and also requires an important storage capacity on each node.

Fig. 1 .

 1 Fig. 1. Hadoop 1.x daemons architecture

•Fig. 5 .

 5 Fig. 4. Elements of the context collector module

Fig. 8 .

 8 Fig. 8. Hadoop vs CloudFIT performance comparison

Table 1 .

 1 Configuration parameters used in the experiments OriginalContext-aware -

JCS

Science Publications

JCS

Science Publications

• The returning node was simply disconnected from the network and still thinks it is the JobTracker.

Zookeeper always keeps a reference to the actual JobTracker, so the JobTracker periodically checks that reference to verify whether it is still the JobTracker or not. If it is not, the returning JobTracker kills himself so the JobTracker referenced by Zookeeper is the only JobTracker in the network • The returning node has restarted and has lost all its status, but is still on the top of Zookeeper's list. In this case, the new node is restarted as a TaskTracker, so it will follows the fault-tolerance mechanism described in the previous subsection to become the JobTracker

Heartbeat Tuning

A too lazy heartbeat slows-down the reaction to failures and may lead to some of the situations described in the previous item. An intensive heartbeat may impact negatively on the overall performance.

While Docker provides an environment to create and destroy nodes, the joining of new nodes requires additional procedures. Indeed, Hadoop was designed to work over a cluster where all the resources are known from the beginning. Inserting new nodes require the restart of the job manager, which may represent an important drawback in a dynamic environment in the next section, we present a P2P approach to solve this problem and discuss the challenges it represents.

A P2P IMPLEMENTATION OF MAPREDUCE

Due to its simple task model, MapReduce can be easily implemented in a distributed computing environment. In our project, we rely on the P2P distributed computing middleware CloudFIT [START_REF] Steffenel | Deliverable 2.1: First Steps on the Development of a P2P Middleware for MapReduce[END_REF], implemented over the Pastry [START_REF] Rowstron | Pastry: Scalable, decentralized object location and routing for largescale peer-to-peer systems[END_REF] overlay network. In CloudFIT, the programmer needs to decide how to divide the problem into a finite number of independent tasks and how to compute each individual task. This is the same principle of MapReduce map and reduce steps, which can be considered as a sequences of Finite number of Independent and Irregular Tasks [START_REF] Krajecki | An object oriented environment to manage the parallelism of the FIIT applications[END_REF] problems.

The CloudFIT framework is structured around collaborative nodes connected over a logical oriented ring overlay network. Task status (and partial results) are broadcasted among the nodes, which contributes to the coordination of the computing tasks and form a global view of the calculus.

A node owns the different parameters of the current computations (a list of tasks and associated results). It is able to locally decide which tasks still need to be computed and can carry the work autonomously if no other node can be contacted. If later a node reintegrates a community, it is able to share the results from the tasks it completed and re-synchronize its task's list. For the moment, a simple scheduling mechanism randomly rearranges the list of tasks at each node, which helps the computation of tasks in parallel without requiring additional communication between nodes.

From the strict point of view of FIIT, a MapReduce job can be expressed as a two rounds execution: One handling Map tasks and another handling Reduce tasks. By implementing MapReduce over a P2P platform such as CloudFIT, we can introduce interesting properties on MapReduce that are not always available on Hadoop.

Implementing MapReduce over CloudFIT is quite straightforward and can easily mimic the behavior of Hadoop. Hence, during the Map phase, several tasks are launched according the number of input files, producing a set of (k i , V i ) pairs. The token passing mechanism ensures that all pairs (i.e., the results of each task) are broadcasted to all computing nodes. Therefore, at the end of the Map phase, each node contains a copy of the entire set of (k i , V i ) pairs.

At the end of the first step, a new CloudFIT job is launched, using as input parameter the results from the map phase. The number of tasks during this Reduce phase is calculated based on the number of available nodes. Once a round starts, each node starts a task from the shared task list and broadcasts its results at the end of the task's computation.

Using CloudFIT, MapReduce algorithms are supposed to support nodes failures as well as nodes volatility, allowing nodes to dynamically leave and join the grid. Indeed, as long as a task is not completed, other nodes on the grid may pick it up. In this way, when a node fails or leaves the grid, other nodes may recover tasks originally taken by the crashed node. Inversely, when a node joins the CloudFIT community, it receives a copy of the working data and may pick up available (incomplete) tasks on the shared task list. Thus, CloudFIT should offer a more fault-tolerant behavior than Hadoop, supporting not only nodes disconnections, but also nodes (re-)connection.

Because Hadoop relies on specific classes to handle data, we tried to use the same ones in CloudFIT JCS

Science Publications

To circumvent this bottleneck, we are currently implementing an alternative storage mechanism based on DHTs, like for example the PAST distributed file system [START_REF] Druschel | PAST: A largescale, persistent peer-to-peer storage utility[END_REF]. Using PAST, we allow workers to share the results of the tasks with a configurable replication factor, all while minimizing the data transfer between nodes and the storage requirements. PAST integration is still under work but we expect to integrate this solution and our context acquisition module to make tasks scheduling aware of context elements such as data locality, network and processing capabilities of the nodes.

CONCLUSION

Pervasive grids represent an important step towards the establishment of mobiquitous systems in which concerns high performance computing. While the pervasive computing model has no intention to supersede classical high performance computing, there is a large domain of applications that require more flexible environments, as provided by a pervasive computing model. Indeed, pervasive grids concentrate three main challenges on dynamic environment composed by a multitude of devices: (i) The volatility of its components; (ii) their intrinsic heterogeneity; and (iii) how to manage the dynamic evolution of these resources.

In this study we study these issues when deploying MapReduce applications on top of pervasive grids. We observed that Hadoop, the most known implementation of MapReduce on clusters and cloud infrastructures, fails to respond to the three challenges listed above. We strongly believe that pervasive grids are especially adapted to deploy MapReduce application on enterprises, fully exploring the potential of unused (or underused) resources and therefore reinforcing the enterprises competitiveness.

To reach this goal, we present the basis of the PER-MARE project, which explores a two-fold approach for implementing effective MapReduce support on pervasive grids: First, by improving Hadoop so that it supports a minimum of volatility and context awareness; second, by developing an alternative middleware for MapReduce directly on top of a pervasive grid platform. By proceeding on both fronts, we aim at obtaining better insights on scientific and technical obstacles towards the development of flexible and adaptive MapReduce middleware.

Therefore, in a first moment we have propose to introduce context information on Hadoop schedulers, in order to take in account the heterogeneity and the dynamicity of the nodes. By injecting real-time context information (such as available memory) on Hadoop schedulers, we circumvent the poor dynamicity management of current Hadoop implementations. As the experiments showed encouraging performance speedups, the next steps will include additional context information (CPU load and data location, for example) and the study of alternative scheduling algorithms more tailored to resource variability.

Later, we focused on Hadoop fault-tolerance, studying how to remove one of the last single point of failure in the architecture and therefore allowing a smooth operation in dynamic environments where any node can disconnect or fail. By coupling context-aware scheduling and an improved fault-tolerant architecture, we are able to support the disconnection of any node in the architecture as well as distributing work tasks according to the nodes capabilities.

Even though the previous contributions improve Hadoop operation, Hadoop remains a complex (and heavy) middleware that not always can be deployed on pervasive systems. Therefore, we also presented our efforts to implement MapReduce over a pervasive grid middleware, as a way to embrace and take profit from the volatility of pervasive grids and devices diversity. By proposing a Hadoop-compliant API over a pervasive grid middleware, we offer MapReduce applications a transparent choice between an implementation optimized for data-intensive problems (Hadoop) and one optimized for computing intensive problems over a highly dynamic environments.

From these contributions we pointed out several elements that can be improved in both Hadoop and pervasive grid frameworks. For instance, future works shall continue towards the association of these contributions in order to provide a complete panorama of MapReduce solutions on pervasive grids.

ACKNOWLEDGEMENT

The authors would like to thank their partners in the PER-MARE project and acknowledge the financial support given to this research by the CAPES/MAEE/ANII STIC-AmSud collaboration program (project number 13STIC07). Experiments presented in this study were carried out on Grid'5000 experimental testbed (https://www.grid5000.fr).


Marco Biazzini

Benoit Baudry

"May the Fork Be with You": Novel Metrics to Analyze Collaboration on GitHub

Keywords: D.2.8 [Software Engineering]: Metrics-Process Metrics; K.6.3 [Management of Computing and Information Systems]: Software Management-Software Process Measurement, Management Software process metrics, Distributed Version Control, Git, Github

Multi-repository software projects are becoming more and more popular, thanks to web-based facilities such as GitHub. Code and process metrics generally assume a single repository must be analyzed, in order to measure the characteristics of a codebase. Thus they are not apt to measure how much relevant information is hosted in multiple repositories contributing to the same codebase. Nor can they feature the characteristics of such a distributed development process. We present a set of novel metrics, based on an original classification of commits, conceived to capture some interesting aspects of a multi-repository development process. We also describe an efficient way to build a data structure that allows to compute these metrics on a set of Git repositories. Interesting outcomes, obtained by applying our metrics on a large sample of projects hosted on GitHub, show the usefulness of our contribution.

INTRODUCTION

We witness an impressive growth in the adoption of Decentralized Version Control Systems (from now on DVCS), which are in many cases preferred to centralized ones (CVCS) because of their flexibility for handling concurrent development and distribution of "mergeable" codebases.

The purpose of CVCSs has always been to maintain a single authoritative codebase, while letting each developer have only a single revision of each of its files at a time. DVCSs are primarily meant to let the developers access, maintain and compare various versions of the same codebase, along with their commit histories, in a decentralized fashion. In such a scenario, authoritative repositories (if any) are just conventionally designated as such by the community of developers.

Current software metrics focus either on the analysis of the code (code metrics) [START_REF] Basili | A validation of object-oriented design metrics as quality indicators[END_REF][START_REF] Demeyer | Finding refactorings via change metrics[END_REF][START_REF] Moser | A comparative analysis of the efficiency of change metrics and static code attributes for defect prediction[END_REF] or of design documents [START_REF] Baudry | An object-oriented high-level design-based class cohesion metric[END_REF][START_REF] Genero | Building uml class diagram maintainability prediction models based on early metrics[END_REF], or on characterizing the development process (process metrics) [START_REF] Canfora | A family of experiments to validate metrics for software process models[END_REF][START_REF] Posnett | Dual ecological measures of focus in software development[END_REF][START_REF] Rahman | Ownership, experience and defects: a fine-grained study of authorship[END_REF][START_REF] Rodriguez-Bustos | How distributed version control systems impact open source software projects[END_REF]. They have merits and pitfalls, well studied in the literature. Recent studies focus on the impact of branching and merging on the quality of the product [START_REF] Shihab | The effect of branching strategies on software quality[END_REF]. They all implicitly assume a single repository to be analyzed to measure the characteristics of a single software product. Nowadays, this is no longer a safe assumption.

The recent boost in the adoption of DVCSs has been driven by public web-based aggregators that greatly facilitate the access to DVCS-based repositories and the interaction among different copies of their codebases. Let us take the case of GitHub. Thanks to the "fork-and-contribute" policy of GitHub, the highly non-linear history of Git repositories becomes publicly exposed and easily duplicable in independent but inter-communicating copies at will.

GitHub makes it public what is normally disclosed only among developers sharing branches of their Git repositories. An explicit (and possibly cumbersome) peer-wise synchronization of local repositories is no longer needed. Anyone can contribute to any repository by creating a personal public fork and pushing changes to this fork. Any change may eventually be pulled from any other fork of the same repository, including of course the original one.

The facilities provided by web hubs like GitHub have been shown to have an impact on the characteristics of the development process, both by easing the parallelization of the tasks within a team [START_REF] Dabbish | Social coding in github: transparency and collaboration in an open software repository[END_REF] and by increasing the number of relevant contributions coming from outsiders [START_REF] Thung | Network structure of social coding in github[END_REF]. This fact poses at least two open issues: (i) how to analyze commit histories scattered in multiple repositories and (ii) how to characterize such a distributed development process.

For the purpose of having a consistent chronological history of a project, it may be enough to only consider its mainline. This repository is progressively updated and thus reliably and consistently shows the evolution of the code.

But, since the other forks of a project are independent and publicly available as well, such a choice seems of a too limited scope. It may easily discard the complexity of the state of the software: the complete codebase of a project (or, from a slightly different perspective, the set of all versions of a software) on GitHub is more than what is committed in the mainline.

A legitimate question arises: is the only analysis of the mainline enough to fully grasp how the software is developed by the community of all contributors? To meaningfully answer this question, we need a way to quantify the amount of contributions dispersed in the various forks, in order to understand how much information would one discard by not considering, beyond the mainline repository, the rest of the project-related cosmos out there on GitHub.

In this paper we bring the following contributions:

• A methodology to efficiently aggregate and analyze commit histories of GitHub forks related to the same project.

• A classification of commits, explicitly conceived for the analysis of DVCSs, which characterizes their distributed development process.

• A set of novel metrics to quantify the degree of dispersion of the contributions in a codebase which is distributed on multiple repositories.

We then illustrate the usefulness of our metrics, by reporting outcomes obtained by mining 342 GitHub projects, composed by a total of 3673 forks.

The paper is organized as follows. Section 2 explains the motivations and the challenges of our work; Section 3 presents our contributions; Section 4 describes interesting experimental outcomes obtained by computing our metrics on a large sample of GitHub projects; Section 5 shows how our commit classification and metrics can be used in visual data analysis to enlighten interesting features of multirepository projects. Finally, Section 6 lists the threats to the validity of our outcomes and Section 7 presents our conclusion.

THE IDENTITY OF A CODEBASE ON GITHUB

Using GitHub as a centralized facility produces a previously unseen way of distributing the development process. The mainline repository of a software is no longer the only publicly available codebase.

The status of the various forked repositories of a project is an interesting novelty: these forks are complete codebases, but they do not represent the "official version of the code" -which is in the mainline repository. Thus, hierarchical structures of interconnected public forks, which ultimately ends in an official mainline, are possible.

Public open-access forks, which variably differ from their mainline, greatly increase the chances of software diversification to occur: different variants of the same software are publicly available as distinct codebases, which can be independently modified. Such a phenomenon, that is identifiable as diversification only from the standpoint of a global look at the whole ensemble of project forks, is mostly unintentionally produced in each fork. It starts from pure code redundancy and then evolves towards an emergent diversity. Evolutionary speaking, it is possible for a forked repository to become the mainline of a new "breed". A developer may freely choose which one of many forks is mainline to her, notwithstanding what is currently designated as mainline by the core team of developers in a project.

The challenges to be faced in analyzing a multi-repository project have been well explained, taking the prominent case of Git repositories [START_REF] Bird | The promises and perils of mining git[END_REF]. One of the most serious and unresolved difficulties lies in the fact that, being the commits dispersed in distinct repositories, it is unclear if and to which extent all relevant contributions are expected to be found in the mainline repository. Are the possibly many other forks otherwise worth mining? Even once one had set up his mind for this second option, the way to efficiently aggregate, compare and mine information from a set of forks belonging to the same "family" is still to be investigated.

Whenever a fork is created from an existing repository on GitHub, its commit history is an exact duplicate of that of the original repository. Then, the histories of the repositories may repeatedly diverge and re-converge, following possibly very different evolutions. The divergence of two commit histories may be measured by tracking those commits that are made after the creation of a fork and whose occurrence in the other forks, which derive from the same mainline, presents non-trivial traits (e.g., they are not present in all the forks of a given mainline, but only in some of them). Let us call these interesting commits iCommits, for brevity. By tracking iCommits we can actually measure to which extent the family of forks of a given project contains relevant contents which are not in the mainline of its codebase, or which are shared only among subgroups of its community of developer.

Thanks to our novel commit classification, upon which our original metrics are defined, we are able to answer the following research questions: R.Q. 1 -Are there commits related to a given project that are dispersed in forks other than the project mainline? R.Q. 2 -Are there differences in the collaboration patterns of multi-repository projects which we can track by analyzing their distributed commit history?

METHODOLOGY AND METRICS

We propose in the following a methodology to effectively extract from a set of forks some useful information about their similarity in terms of commit history. We first propose a viable way to create an all-encompassing repository, which gathers the history of all the forks pertaining to the same project. Then we define a classification of commits that allows to quantify the amount of difference among the commit histories of the various forks. Finally, we define a set of metrics based on our commit classification.

One umbrella to rule them all

To get to know which commits belong to each category, we need a practical way to analyze the ensemble of all forks of a given project. We need to know what do they have in common and how each fork differs from the mainline. Since a software repository may have hundreds of forks, each of which may comprise thousands of commits, a clever way of handling the data complexity is needed.

We propose an original approach, consisting of building a single Git repository that includes all GitHub forks of the same project. We call it the umbrella repository. From the operational standpoint, the procedure to build the umbrella repository of a project P is quite straightforward:

1. Create and empty Git repository R; 2. For each fork f in the fork family of P , add f as a git remote to R, naming it with a unique identifier;

3. For each branch b of each fork f added to R, fetch the content of b into R.

By adding all branches of each fork as remotes in the same Git repository, we can let Git work for us in building the common commit history among all forks and optimize the memory needed to store all data coming from different repositories. The umbrella repository contains the official mainline of the project and any other commit published in one of its forks. Identical commits are automatically detected and their presence (or absence) in the various forks is easily traceable. Same considerations hold for information related to branches, authors etc.

By melting together all the forks of the same project we obtain very complex development histories organized in directed acyclic graphs, in which all structural information are preserved and can be matched, compared and mined in a seamless way. In order to ease the task of data organization and metric extraction from Git repositories, we implemented our own toolset, called GitWorks, available on GitHub as well 1 . It is a pure Java application, which works on top of JGit 2 . Thanks to GitWorks, the whole procedure, from the creation of umbrella repositories to the computation of our metrics on all projects, is completely automatized.

Our approach can be useful for different purposes. It can be used to characterize the "official" history of the development of a software with respect to the rest of the contributions, for instance by reporting the differences between the mainline and the various forks. In the following, we use it to "extensionally" characterize the state of the art of a given codebase, across all publicly available forks at a given point in time.

Commit classification

In order to understand if and how much the various forks composing an umbrella repository contain iCommits, we propose to first detect, in each fork, all commits made after the creation of the fork itself. By considering these ones only, we discard all commits that are part of a fork since the very moment of its creation, thus not meaningful to assess developers' activity.

Once we have the set of iCommits, we partition them into the following categories:

unique: iCommits existing in one fork only.

vip: iCommits existing in several (but not all) forks and in the mainline.

u-vip: iCommits existing in the mainline and in one other fork only.

scattered: iCommits existing in several forks, but not in the mainline.

1 See https://github.com/marbiaz/GitWorks. 2 See http://eclipse.org/jgit.

pervasive: iCommits existing in all repositories.

These categories are useful to get a glimpse of the activity in the various forks of a project codebase. unique and scattered commits are interesting in that they are proof of development activity which is independent from the mainline repository. vip and u-vip commits, on the other hand, are evidences of mainline-related activity, which is distributed in subsets of forks. pervasive commits indicate to which extent new contributions are shared among the whole community of contributors.

With the help of some notation, we can formally define our categories as sets of commits related to a given umbrella repository R. Let C be the set of all commits ci belonging to R. Let F be the set of all forks fi composing R. In the next, we assume |F | > 1. Let M be the set of all commits belonging to the history of the mainline in R. Finally, let f Count : C -→ N be a function which, given a commit c ∈ C returns the number of forks in F created before c and whose commit history includes c.

We give the following formal definitions, for a given umbrella repository R.

Def. 1. The set U of unique commits is defined as

U = {ci ∈ C : f Count(ci) = 1}.
Def. 2. The set V of vip commits is defined as

V = {ci ∈ C : ci ∈ M ∧ |F | > f Count(ci) > 2}.
Def. 3. The set W of u-vip commits is defined as

W = {ci ∈ C : ci ∈ M∧|F | > f Count(ci)∧f Count(ci) = 2}.
Def. 4. The set S of scattered commits is defined as

S = {ci ∈ C : ci / ∈ M ∧ f Count(ci) > 1}.
Def. 5. The set P of pervasive commits is defined as

P = {ci ∈ C : f Count(ci) = |F |}.
We are also able to give a more precise definition of iCommits of R: Def. 6. We call iCommits the commits belonging to the union set

I = U ∪ V ∪ W ∪ S ∪ P.
In Section 4, we present some evidence of the occurrence of iCommits on a sample of GitHub projects.

Dispersion metrics

We now define some simple metrics, based on the above given definitions, for a given umbrella repository R. While the * -count metrics are the cardinality of the sets we defined, the * -ratio metrics are the same cardinalities normalized over the total amount of iCommits.

These metrics allow to quantify to which extent the commits of an umbrella repository are scattered among its forks. By computing these metrics, we obtain a set of values that synthetically describe the commit dispersion in a multirepository project.

PRYING UNDER THE UMBRELLAS

According to FLOSSmole [START_REF] Howison | Flossmole: A collaborative repository for floss research data and analyses[END_REF] (Free Libre OpenSource Software) statistics, GitHub had 191765 repositories publicly available at May 2012. In order to obtain a statistically representative sample of GitHub hosted projects, we sort project first repositories (those that are not forks of other repositories) according to the number of watchers. To discard outliers and less significant entries, we decide to cut off the extremals of the range, i.e. projects whose number of watchers is less than 2 or more than 1000. Then we select 1% of the projects in each of three subsets:

• Projects that had from 2 to 9 watchers (total: 30236 ; sampled: 303)

Projects that had from 10 to 99 watchers (total: 3554 ; sampled: 36)

• Projects that had from 100 to 999 watchers (total: 286 ; sampled: 3)

For each sampled project, we clone the selected mainline repository and all the publicly available forks descending from it (direct forks, forks of the forks, etc.). The resulting set of 342 umbrella repositories, each of which has a mainline and all "generations" of its forks, sums up to a total of 3673 Git repositories. This is our GitHub sample. Information about the fork family of each project, the owner and the creation time of each fork, as well as many other metadata, can be retrieved from GitHub via its publicly available rest API3 . The complete list of repositories in our sample is available online 4 . We create a single umbrella repository per project, comprising the mainline and all its descendants. Once we have computed our metrics on the umbrella repositories of all projects in our GitHub sample, we are able to see if and how much our initial intuition is backed up by real data.

To get an overall bird-eye glimpse, we measure the distribution of values for each of our * -ratio metrics, aggregating the data coming from all the various repositories in our GitHub sample. The 342 projects in the sample differ from each other in any quantitative aspect (number of forks, branches, commits, authors, etc.) and the ratios provide values already normalized in the interval [0. Figure 1 shows boxplots for each metric. The boxes extend for the standard interquartile range, while the whiskers cover up to the 95% of the data points. We suppress the outliers, because they are so many, most of all in the upper range of the interval, that they would hinder the legibility of the plots.

Figure 1a shows the aggregates of the metrics over the whole sample. We see that commits shared among mainlines and some of their forks (vip and u-vip) may often represent a remarkable share of the iCommits. Another quite interesting fact: pervasive commits are globally much less present. Their ratio with respect to the total number of iCommits in their umbrella repositories is often close to 0. This fact may be due to two different facts: (i) forks are created but not kept up-to-date with respect to the mainline and the other forks; (ii) forks are created and then no new commit is added to their upstream repository (the one from which they have been forked). Clearly, to find out which case is the occurring one, one must analyze every umbrella repository in detail.

Quite surprisingly, the most represented category in our sample is that of unique commits. This fact, whose entailments would of course require a deeper investigation, shows that the amount of "original" development which stays outside the mainline of a project is often quite large and thus not to be neglected. A similar consideration holds for scattered commits, which, although much less common, may in some case be fairly important (notice the long whisker of the sr boxplot). Intuitively, the uc and sc metrics may be useful to detect emergent diversity in a multi-repository project, since they can point out those forks which are contributing the most to the phenomenon.

As said, we have for all the distributions a large number of outliers. In order to see the variability of the values among the repositories which do have iCommits, we plot in Figure 1b the same dataset excluding the entries equals to 0.

Here we can see the fairly large variety of situations that exist "in the wild". Most of all, it becomes evident that unique commits are extremely common in our sample. Their ur boxplots in the two figures are actually identical, because less than 3% of the umbrella repositories in our sample have no unique commit (thus only outliers, not shown, would differ). Finally, we see that pervasive commits, although generally being a rare specimen, may represent, whenever they occur, a relevant portion of the iCommits of an umbrella repository.

While this is not a rigorous quantitative analysis of the "composition" of the various projects in our sample, it is enough to positively answer our first research question: R.Q. 1 -Are there commits related to a given project that are dispersed in forks other than the project mainline? Answer -As measured by our dispersion metrics, there are relevant amount of information disseminated among various forks of the same project, which cannot be captured by analyzing the mainline repository only.

In Section 5 we see how our classification proves to be useful in getting insights about the characteristics of the families of forks belonging to the same project.

PEACOCK TAILS

The kind of analysis we propose deals with the fact that a software codebase may be scattered among different repositories, which may be only partially synchronized with each others. All existing code and process metrics can be used in order to measure interesting properties of the single forks.

But our dispersion metrics can be used to give some preliminary insights about the composition of a family of forks, which can be useful to guide further analysis towards the more interesting ones. In order to ease the presentation and facilitate the legibility of the aggregates computed on each family of forks, some visualization tool can be used to pictorially represent our outcomes and highlight some features.

In the following we present selected pictographs, which represent the information obtained on our GitHub sample, for some umbrella repositories. The pictures have been drawn with Circos 5 . Given their shape and look, we nickname them "peacock tails". We underline that the graphic representation in itself is not a major concern of ours, but a simple yet very helpful way of presenting the data and spotting out some interesting features of diversely distributed 5 See http://circos.ca Each pictograph represents the mainline of a project with the subset of its forks having one or more iCommits. The largest stripe at the bottom is the mainline repository. Then the forks are sorted, clockwise from the left, according to their creation timestamp. They are shown as stripes, connected to the mainline by elongated commit-links. On the outskirt, centered at each fork stripe, the identifiers of the forks are reported.

The length of a fork stripe (but the mainline) is proportional to the amount of its iCommits, excluding unique commits. The color of each fork stripe and its commit-link is also correlated to its length: from grey and violet for smaller stripes, through blue and green for medium stripes, to orange and then red for larger ones. Thus, while the length of the stripes tells immediately which forks share more iCommits with the mainline, the color of the stripes is useful to quickly see which forks have a similar amount of iCommits.

For each fork (including the mainline) unique commits are plotted as circles centered at their fork stripe. The di- ameter of these circles is thus proportional to the uc value of the fork, though not in the same scale of the length of the fork stripes. Figure 2 graphically explains the peacock tails' characteristics.

Visualizing umbrella repositories as peacock tails allows us to observe different collaboration models. We give few examples in the following. "Seabirds" collaboration model -In the seabird model, the amount of iCommits which link forks and mainline is balanced: several forks are equally involved in the distributed development. The MailCore project in Figure 3 is an example of that model. The peacock visualization highlights the balance via the colors of the stripes and the commit-links: several of them have similar colors, indicating an equivalent amount of iCommits shared with the main fork. "Goose" collaboration model -In the goose model, forks differ more from each other in their activity: some forks are heavily involved, while others very little. The zamboni project in Figure 4 is an example of that model. The peacock visualization highlights the lack of balance: we can recognize four groups of commit-links, by grouping them according to their color, and a fairly large amount of forks with very few iCommits. "Galapagos" Effect -The Galapagos model emphasizes the presence of some forks that have many unique commits. Our intuition is that this fact indicates a "speciation" inside a fork, probably one or several branches that are used to develop alternative solutions that are not shared with the other forks. The pyromcs project in Figure 5 is an example in which the mainline has a very high uc value (the thin orange circle traversing the plot is actually the uc circle centered at the mainline).

An interesting feature highlighted by these pictographs is the relation between the amount of iCommits and the age of the fork. It is a common finding that older forks are the most contributing, but it is not always the case. The uc values, instead, do not show correlations with the age of the forks or their amount of iCommits.

This quick and intuitive look at multi-repository projects can be very helpful to study the composition of the various forks and for a preliminary screening, in order to identify interesting forks that are worth investigating.

We can thus positively answer to our second research question: R.Q. 2 -Are there differences in the collaboration patterns of multi-repository projects which we can track by analyzing their distributed commit history? Answer -The characterization of multi-repository projects based on our dispersion metrics reveals that collaboration patterns may differ significantly among projects. The visual analysis we briefly discuss here can help deciding the initial directions for a deeper investigation.

THREATS TO VALIDITY

The toolset we use to compute our experimental outcomes has been developed by us and manually tested. It may contain unknown bugs which could affect the computation.

Our outcomes are obtained on a sample of GitHub projects which is small, with respect to the amount of existing repositories on GitHub. Thus, reported results may not generalize. To mitigate as much as possible this threat, we selected projects according to a criterion which is, to the best of our knowledge, unrelated to any characterization of collaboration activities in multi-repository projects, as described in Section 4.

CONCLUSION

The widespread adoption of decentralized versioning systems and the advent of web-based aggregators have caused a substantial increment of multi-repository projects. These projects are characterized by the fact that their complete codebase is scattered among distinct and possibly unsynchronized repositories. Existing metrics are not able to feature such a distributed development scenario.

This paper presents novel tools to tackle the analysis of projects, whose codebase is distributed among several forks on GitHub.

We describe a methodology to efficiently aggregate and analyze commit histories of GitHub forks related to the same project. We propose a classification of commits, which characterizes a distributed development process that is typical of DVCSs. We define a set of novel metrics to quantify the degree of dispersion of the overall contributions in a multi-repository project. We finally report aggregate statistics, measured on a sample of thousands of GitHub repositories, which show that our metrics shed some light on novel interesting aspects of the software development process in multi-repository projects.

Analogously to what has been found for branching strategies, the long term goal is to identify distributed development patterns which affect software quality. On a different track, we plan to exploit our metrics in order to devise a measure of emergent software diversity.

M. 1 :

 1 unique-count is defined as uc = |U |. M. 2: unique-ratio is defined as ur = |U |/|I|. M. 3: vip-count is defined as vc = |V|. M. 4: vip-ratio is defined as vr = |V|/|I|. M. 5: u-vip-count is defined as uvc = |W|. M. 6: u-vip-ratio is defined as uvr = |W|/|I|. M. 7: scattered-count is defined as sc = |S|. M. 8: scattered-ratio is defined as sr = |S|/|I|. M. 9: pervasive-count is defined as pc = |P|. M. 10: pervasive-ratio is defined as pr = |P|/|I|.

  .1]. By aggregating values this way, we can have a general idea about the relative importance of each category of commits in the umbrella repositories of our GitHub sample. (a) All repositories (b) Omitting repositories with no contribution

Figure 1 :

 1 Figure 1: Distributions of commits per categories: aggregates on the whole GitHub sample.

Figure 2 :

 2 Figure 2: Peacock tail example with legend.

Figure 3 :

 3 Figure 3: Peacock tail of the MailCore project.

Figure 4 :

 4 Figure 4: Peacock tail of the zamboni project.

See http://developer.github.com/.

See http://people.rennes.inria.fr/Benoit.Baudry/ sampe-github-projects/.

ACKNOWLEDGMENTS

We want to thank Martin Monperrus for his nice advices and useful discussions.

This work is partially supported by the EU FP7-ICT-2011-9 No. 600654 DIVERSIFY project. Figure 5: Peacock tail of the pyrocms project.