International audience

Packet processing on Linux can be slow due to its complex network stack. To solve this problem, there are two main solutions: eXpress Data Path (XDP) and Data Plane Development Kit (DPDK). XDP and the AF _XDP socket offer full interoperability with the legacy system and is being adopted by major internet players like Open vSwitch or Facebook. While the performance evaluation of AF _XDP against the legacy protocol stack in the kernel or against DPDK has been studied in the literature, the impact of the multiple socket parameters and the system configuration on its latency has been left aside. To address this, we conduct an experimental study to understand the XDP/AF _XDP ecosystem and detect microseconds delays to better architect future latency-sensitive applications. Since the performance of AF _XDP depends on multiple parameters found in different layers, finding the configuration minimizing its latency is a challenging task. We rely on a classification algorithm to group the performance results, allowing us to easily identify parameters with the biggest impact on performance at different loads. Last, but not least, we show that some configurations can significantly decrease the benefits of AF _XDP, leading to undesirable behaviors, while other configurations are able to reduce such round trip delays to an impressive value of 6.5 μS in the best case, including the tracing overhead. In summary, AF _XDP is a promising solution, and careful selection of both application and socket parameters can significantly improve performance.

The importance of Big Data is nowadays established, both in industry and research fields, especially stream processing for its capability to analyze continuous data streams and provide statistics in real-time. Several data stream processing (DSP) platforms exist like the Storm, Flink, Spark Streaming and Heron Apache projects, or industrial products such as Google MillWheel. Usually, each platform is tested and analyzed using either specifically crafted benchmarks or realistic applications. Unfortunately, these applications are only briefly described and their source code is generally not available. Hence, making quick evaluations often involves rewriting complete applications on different platforms. The lack of a generic prototype application also makes it difficult for a developer to quickly evaluate the impact of some design choices. To address these issues, we present NAMB (Not only A Micro-Benchmark), a generic application prototype generator for DSP platforms. Given a high-level description of a stream processing application and its workload, NAMB automatically generates the code for different platforms. It features a flexible architecture which makes it easy to support new platforms. We demonstrate the benefits of our proposal to quickly generate application prototypes as well as benchmarks used in published papers. Overall, our approach provides easily replicable, comparable and customizable prototypes for data stream platforms. Moreover, NAMB provides similar performance in terms of latency and throughput to existing benchmarks, while only requiring a simple high-level description.

The relevance of Data Stream Processing (DSP) is nowadays established, thanks to its capability to analyze continuous streams and provide statistics in real-time. A considerable amount of work has been dedicated to improve performance and features of DSP platforms. Thus, benchmark application are necessary for comparison and evaluation. Unfortunately, in literature, these applications are often briefly described, the source is not available, they are too context-specific or don't provide enough flexibility. That makes it difficult for a developer to quickly evaluate the impact of some design choices. To address these issues, we introduce a high-level description model of stream applications. Based on fundamental DSP characteristics, this description allow an easy and flexible definition of benchmark topologies. With this model we aim to provide easily replicable, comparable and customizable benchmarks for DSP. We then use a framework prototype that translates the high-level description into platform-specific code simulating the application workload.

The widespread use of social networks and applications such as IoT networks generates a continuous stream of data that companies and researchers want to process, ideally in real-time. Data stream processing systems (DSP) enable such continuous data analysis by implementing the set of operations to be performed on the stream as directed acyclic graph (DAG) of tasks. While these DSP systems embed mechanisms to ensure fault tolerance and message reliability, only few studies focus on the impact of these mechanisms on the performance of applications at runtime. In this paper, we demonstrate the impact of the message reliability mechanism on the performance of the application. We use an experimental approach, using the Storm middleware, to study an acknowledgment-based framework. We compare the two standard schedulers available in Storm with applications of various degrees of parallelism, over single and multi cluster scenarios. We show that the acking layer may create an unforeseen bottleneck due to the acking tasks placement; a problem which, to the best of our knowledge, has been overlooked in the scientific and technical literature. We propose two strategies for improving the acking tasks placement and demonstrate their benefit in terms of throughput and latency.

Presents corrections to the paper, “K nearest neighbour joins for big data on MapReduce: A theoretical and experimental analysis,” (Song, G., et al), IEEE Trans. Knowl. Data Eng., vol. 28, no. 9, pp. 2376–2392, Sep. 2016.

Summary Distributed systems for big data management very often face the problem of load imbalance among nodes. To address this issue, there exist almost as many load balancing strategies as there are different systems. When designing a scalable distributed system geared towards handling large amounts of information, it is often not so easy to anticipate which kind of strategy will be the most efficient to maintain adequate performance regarding response time, scalability, and reliability at any time. Based on this observation, we describe a generic API to implement and experiment any strategy independently from the rest of the code, prior to a definitive choice for instance. We then show how existing load balancing strategies used by famous systems could be implemented with this API. We also present how this work has helped us implement load balancing on our distributed system and modify the behavior of our strategy in a few lines of code. This led us to easily perform various experiments to determine the most efficient scheme for our system. This paper is an extension to our work presented at Workshop on Parallel and Distributed Computing for Big Data Applications (WPBA) 2014. We detail here more experiments and extend the use of the API to a broad class of big data storage systems. Copyright © 2015 John Wiley & Sons, Ltd.

Given a point$p$and a set of points$S$, the kNN operation finds the$k$closest points to in$S$. It is a computational intensive task with a large range of applications such as knowledge discovery or data mining. However, as the volume and the dimension of data increase, only distributed approaches can perform such costly operation in a reasonable time. Recent works have focused on implementing efficient solutions using the MapReduce programming model because it is suitable for distributed large scale data processing. Although these works provide different solutions to the same problem, each one has particular constraints and properties. In this paper, we compare the different existing approaches for computing kNN on MapReduce, first theoretically, and then by performing an extensive experimental evaluation. To be able to compare solutions, we identify three generic steps for kNN computation on MapReduce: data pre-processing, data partitioning, and computation. We then analyze each step from load balancing, accuracy, and complexity aspects. Experiments in this paper use a variety of datasets, and analyze the impact of data volume, data dimension, and the value of k from many perspectives like time and space complexity, and accuracy. The experimental part brings new advantages and shortcomings that are discussed for each algorithm. To the best of our knowledge, this is the first paper that compares kNN computing methods on MapReduce both theoretically and experimentally with the same setting. Overall, this paper can be used as a guide to tackle kNN-based practical problems in the context of big data.

Given a point p and a set of points S, the kNN operation finds the k closest points to p in S. It is a computational intensive task with a large range of applications such as knowledge discovery or data mining. However, as the volume and the dimension of data increase, only distributed approaches can perform such costly operation in a reasonable time. Recent works have focused on implementing efficient solutions using the MapReduce programming model because it is suitable for large scale data processing. Also, it can easily be executed in a distributed environment. Although these works provide different solutions to the same problem, each one has particular constraints and properties. There is no readily available comparison to help users choose the one most appropriate for their needs. This is the problem we address in this work. Firstly, we show that all kNN implementations go through a common workflow, which we use as a basis for classification. Secondly, we describe precisely the different techniques published so far. And lastly, we provide a set of objective criteria that can be used to make informed decisions.

Storing highly skewed data in a distributed system has become a very frequent issue, in particular with the emergence of semantic Web and Big Data. This often leads to biased data dissemination among nodes. Addressing load imbalance is necessary, especially to minimize response time and avoid workload being handled by only one or few nodes. Our contribution aims at dynamically managing load imbalance by allowing multiple hash functions on different peers, while maintaining consistency of the overlay. Our experiments, on highly skewed data sets from the semantic web, show we can distribute data on at least 300 times more peers than when not using any load balancing strategy.

MapReduce is a programming model which allows the processing of vast amounts of data in parallel, on a large number of machines. It is particularly well suited to static or slow changing set of data since the execution time of a job is usually high. However, in practice data-centers collect data at fast rates which makes it very difficult to maintain up-to-date results. To address this challenge, we propose in this paper a generic mechanism for dealing with dynamic data in MapReduce frameworks. Long-standing MapReduce jobs, called continuous Jobs, are automatically re-executed to process new incoming data at a minimum cost. We present a simple and clean API which integrates nicely with the standard MapReduce model. Furthermore, we describe cHadoop, an implementation of our approach based on Hadoop which does not require modifications to the source code of the original framework. Thus, cHadoop can quickly be ported to any new version of Hadoop. We evaluate our proposal with two standard MapReduce applications (Word Count and Word Count-N-Count), and one real world application (RDF Query) on real datasets. Our evaluations on clusters ranging from 5 to 40 nodes demonstrate the benefit of our approach in terms of execution time and ease of use.

There are several programming paradigms that help programmers write efficient and verifiable code for distributed environments. These solutions, however, often lack proper support for local parallelism. In this article we try to improve existing solutions for providing a distributed, highly parallel framework that is easy to program. We propose an extension to the active object programming model which optimizes the local performance of applications by harnessing the full computing power of multi-core CPUs. The need for explicit locking mechanisms is reduced by the addition of meta-information to the methods in the source code. This paper describes this language-independent meta-information, and the way we intend to use it for parallelizing execution inside an active object.

With the increasing demand and benefits of cloud computing infrastructure, real time computing can be performed on cloud infrastructure. A real time system can take advantage of intensive computing capabilities and scalable virtualized environment of cloud computing to execute real time tasks. In most of the real time cloud applications, processing is done on remote cloud computing nodes. So there are more chances of errors, due to the undetermined latency and loose control over computing node. On the other side, most of the real time systems are also safety critical and should be highly reliable. So there is an increased requirement for fault tolerance to achieve reliability for the real time computing on cloud infrastructure. In this paper, a fault tolerance model for real time cloud computing is proposed. In the proposed model, the system tolerates the faults and makes the decision on the basis of reliability of the processing nodes, i.e. virtual machines. The reliability of the virtual machines is adaptive, which changes after every computing cycle. If a virtual machine manages to produce a correct result within the time limit, its reliability increases. And if it fails to produce the result within time or correct result, its reliability decreases. A metric model is given for the reliability assessment. In the model, decrease in reliability is more than increase. If the node continues to fail, it is removed, and a new node is added. There is also a minimum reliability level. If any processing node does not achieve that level, the systems will perform backward recovery or safety measures. The proposed technique is based on the execution of design diverse variants on multiple virtual machines, and assigning reliability to the results produced by variants. The virtual machine instances can be of same type or of different types. The system provides both the forward and backward recovery mechanism, but main focus is on forward recovery. The main essence of the proposed technique is the adaptive behavior of the reliability weights assigned to each processing node and adding and removing of nodes on the basis of reliability.

Resource discovery is a challenging issue in unstructured peer-to-peer networks. Blind search approaches, including flooding and random walks, are the two typical algorithms used in such systems. Blind flooding is not scalable because of its high communication cost. On the other hand, the performance of random walks approaches largely depends on the random choice of walks. Some informed mechanisms use additional information, usually obtained from previous queries, for routing. Such approaches can reduce the traffic overhead but they limit the query coverage. Furthermore, they usually rely on complex protocols to maintain information at each peer. In this paper, we propose two schemes which can be used to improve the search performance in unstructured peer-to-peer networks. The first one is a simple caching mechanism based on resource descriptions. Peers that offer resources send periodic advertisement messages. These messages are stored into a cache and are used for routing requests. The second scheme is a dynamic Time-To-Live (TTL) enabling messages to break their horizon. Instead of decreasing the query TTL by 1 at each hop, it is decreased by a value v such as 0 <; v <; 1. Our aim is not only to redirect queries towards the right direction but also to stimulate them in order to reliably discover rare resources. We then propose a Dynamic Resource Discovery Protocol (DRDP) which uses the two previously described mechanisms. Through extensive simulations, we show that our approach achieves a high success rate while incurring a low search traffic.

Distributed computing environments have evolved from in-house clusters to Grids and now Cloud platforms. We, as others, provide HPC benchmarks results over Amazon EC2 that show a lower performance of Cloud resources compared to private resources., So, it is not yet clear how much of impact Clouds will have in high performance computing (HPC). But hybrid Grid/Cloud computing may offer opportunities to increase overall applications performance, while benefiting from in-house computational resources extending them by Cloud ones only whenever needed. In this paper, we advocate the usage of Proactive, a well established middleware in the grid community, for mixed Grid/Cloud computing, extended with features to address Grid/Cloud issues with little or, no effort for application developers. We also introduce a framework, developed in the context of the Disco Grid project, based upon the Proactive middleware to couple HPC domain-decomposition SPMD applications in heterogeneous multi-domain environments. Performance results, coupling Grid and Cloud resources for the execution of such, kind of highly communicating and processing intensive applications, have shown an overhead of about 15%, which is a non-negligible value, but lower enough to consider using such environments to achieve a better cost-performance trade-off than using exclusively Cloud resources.

Grids are distributed systems aiming at the aggregation of geographically distributed resources for high performance computing. Heterogeneity and volatility are the main characteristics of resources in a grid environment. This aspect makes the resource discovery in a grid a crucial problem: given the resource requirements of an application, a resource discovery mechanism returns the set of resources matched by the description. By considering a grid as a peer-to-peer network, it is possible to use decentralized algorithms to locate resources. We propose in this paper a new mechanism for peer-to-peer resource discovery in a grid environment. Our system is based on push-pull strategy, i.e, each peer may ask for resources or advertise them if they are available. Also, each peer maintains a local cache of messages and uses this information for routing. We compare our proposal with other search methods through simulation and we show that it provides higher success rate with lower overhead.

About ten years after the Java Grande effort, this paper aims at providing a snapshot of the comparison of Fortran MPI, with Java performance for parallel computing, using the ProActive library. We first analyze some performance details about ProActive behaviors, and then compare its global performance from the MPI library. This comparative is based on the five kernels of the NAS parallel benchmarks. From those experiments we identify benchmarks where parallel Java performs as well as Fortran MPI, and lack of performance on others, together with clues for improvement.

Previous experiments with high-performance Java were initially disappointing. After several years of optimization, this paper investigates the current suitability of such object-oriented middle-ware for High-Performance and Grid programming. Using a middleware o®ering high level abstractions (ProActive), we have replaced the standard Java RMI layer with the optimized Ibis RMI interface. Ibis is a grid programming environment featuring e±cient communications. Using a 3D electromagnetic application (an object-oriented time domain ¯nite volume solver for 3D Maxwell equations) we have ¯rst conducted benchmarks on single clusters, including comparisons with the same application in Fortran MPI. Finally, Grid experiments have been conducted simultaneously on up to 5 di®erent clusters. Overall, the paper reports extremely promising results. For instance, a speed up of 12 on 16 machines (vs. 13.8 for Fortran), a speedup of 100 on 150 machines on a Grid. 1

Increasing complexity of distributed applications and commodity of resources through grids are making the tasks of deploying those applications harder. There is a clear need for standard tools allowing versatile deployment and analysis of distributed applications. We present here a solution for the deployment and monitoring of applications written using ProActive, an experimental Java-based library for concurrent, distributed and mobile computing. We describe the use of XML-based descriptor for the deployment part of a distributed application and the use of IC2D (Interactive Control and Debugging of Distribution), for the monitoring and steering of the running application. Those ideas, concepts, and experiments are a contribution towards the construction of integrated environments for component-based grid programming.

The Internet has allowed the creation of huge amounts of data located on many sites. Performing complex operations on some data requires that the data be transferred first to the machine on which the operations are to be executed, which may require a non-negligible amount of bandwidth and may seriously limit performance if it is the bottleneck. However, instead of moving the data to the code, it is possible to move the code to the data, and perform all the operations locally. This simple idea has led to a new paradigm called code-mobility: a mobile object --- sometimes called an agent --- is given a list of destinations and a series of operations to perform on each one of them. The agent will visit all of the destinations, perform the requested operations and possibly pass the result on to another object. Any mobility mechanism must first provide a way to migrate code from one host to another. It must also ensure that any communication following a migration will not be impaired by it, namely that two objects should still be able to communicate even if one of them has migrated. Such a mechanism is referred to as a location mechanism since it often relies on the knowledge of the location of the objects to ensure communications. Two location mechanisms are widely used: the first one uses a centralized server whereas the second one relies on special objects called forwarders.This paper evaluates and compares the performance of an existing implementation of these approaches in terms of cost of communication in presence of migration. Based on a Markov chain analysis, we will construct and solve two mathematical models, one for each mechanism and will use them to evaluate the cost of location. For the purpose of validation, we have developed for each mechanism a benchmark that uses ProActive [2], a Java library that provides all the necessary primitives for code mobility. Experiments conducted on a LAN and on a MAN have validated both models and have shown that the location server always performs better than the forwarders. Using our analytical models we will nevertheless identify situations where the opposite conclusion holds. However, under most operational conditions location servers will perform better than forwarders.