Tomasz Buchert

Lucas Nussbaum

Jens Gustedt

A workflow-inspired, modular and robust approach to experiments in distributed systems

Keywords: experimentation, control of experiments, workflows, large-scale distributed systems, performance analysis

Experimentation in large-scale distributed systems research is very challenging due to the size and complexity of modern systems and applications spanning domains of high performance computing, P2P networks, cloud computing, etc. Some obstacles that each researcher must face are: the difficulty of properly structuring experiments due to their complexity, the inflexibility of existing methodologies and tools and the scalability problems resulting from the size of studied systems.

In this paper, we propose a novel method of representing and executing experiments that solves these problems. To this end, we present an interdisciplinary approach to the control of largescale experiments in distributed systems research that draws its foundations from workflow management and scientific workflows. This workflow-inspired approach distinguishes itself by its representation of experiments, modular architecture and robust error handling. We show how the aforementioned problems are solved by our approach in an exemplary performance study of an HTTP server.

I. INTRODUCTION

A. Context and motivation

Computer science started as a formal, theoretical science, but very rapidly became one of the most important of applied sciences. In parallel to the developments of computer science, computer systems were becoming more and more complex making the formal analysis unfeasible to the point that methods based on purely analytic, mathematical approach are almost non-existing (amounting to less than 4% [START_REF] Wainer | Empirical evaluation in Computer Science research published by ACM[END_REF]). There is, therefore, a trend of turning to scientific method and experimentation, both used routinely with much success in physical and life science research [START_REF] Tichy | Should computer scientists experiment more?[END_REF].

In the domain of distributed systems, the situation is arguably even worse. The objects of study, distributed systems built from loosely related components, with independent sense of passing time and connected using unreliable networks, are intrinsically non-deterministic. Moreover, the modern systems depend on countless software packages, each bringing its own influence on the overall behavior of the system. Largescale distributed systems comprising thousands of independent machines are an extreme case where non-methodological and manual approach simply cannot work. Even a motivated scientist may be unable to produce high-quality results when confronted with such problems.

The difficulty is partially due to how the experiments are conducted. The complexity of existing distributed systems requires that the experiments are well-documented (as software should be), reproducible (so that other researchers can verify and use findings easily) and modular (to encourage reuse and standardize the domain). Moreover, they should be general, independent from any particular platform for example. Finally, they should be robust and address the faulty nature of distributed systems. Sadly, these issues are rarely addressed due to the difficulty and tediousness of such an undertaking.

We need dedicated methods to experimental research in large-scale distributed systems that address issues that the domain struggles with. The advances in the domain of experiment control systems would be welcomed and bring more credibility and quality into the research that is difficult to conduct, despite the existence of many approaches (up to 26% of empirical or engineering publications in computer science have no evaluation at all [START_REF] Wainer | Empirical evaluation in Computer Science research published by ACM[END_REF]).

B. Contribution

In this paper, we describe a novel approach to control experiments, which borrows ideas from experiment control systems, scientific workflows and workflow management systems. This workflow-inspired approach redefines the common activities of experimentation in one consistent, holistic methodology. We illustrate our findings with XPFLOW, a workflow engine tailored to the requirements of distributed systems research.

The rest of this article is structured as follows. First, in Section II, we describe our approach and its main features:

1) a flexible workflow representation of experiments that promotes good practices, code reuse, improves understanding and paves a way to formal analysis, 2) a modular architecture that supports interoperability with low-level tools, instrumentation of experimental workflows and composability of experimental patterns, 3) robust error handling that includes special workflow patterns and checkpointing of workflow execution.

In Section III, we evaluate our approach on a real-life experiment. Then, in Section IV, the existing solutions and methods for control of experiments in distributed systems research are presented and compared with our approach. Finally, Section V concludes the paper and outlines future work. . An example of a mapping between DSL and a workflow. Different language-level constructs (parallel, forall, sequence) result in respective workflow patterns (parallel execution of subworkflows, a parallel loop and sequential execution of activities). The workflow notation is based on Business Process Model and Notation (BPMN) with some extensions (the "ampersand notation" for a parallel loop, for example).

II. WORKFLOW-INSPIRED APPROACH

Our approach is an interdisciplinary merger of three domains: experiment control systems, scientific workflows and workflow management. First, our main motivation is the control of experiments. Second, we share common concepts and goals with scientific workflows, the workflow representation of experiments and provenance in particular. Finally, we borrow the basic model of execution and workflow patterns from the domain of workflow management (a supporting science of Business Process Management).

The approach consists in using a domain-specific language to describe experiment workflows and execute them using a workflow engine tailored to control of experiments (called an experiment control system for short). The workflow engine launches, controls and monitors all aspects of experiment execution. A detailed log is kept which includes timing information, low-level and user-defined debugging information, etc. Additionally, it provides low-level features which cannot be implemented as workflows (e.g., checkpointing and error handling). Precise information about execution of the experiment can be exported, analyzed afterwards and stored as an experiment log.

Our implementation of this workflow-inspired approach, XPFLOW, is written in Ruby which was chosen because of its popularity in configuration management tools (Puppet1 , Chef2 ) and its flexible syntax (for the purpose of implementing a high-level, domain-specific language). XPFLOW is a workflow engine that focuses on the control of experiments in distributed systems research. To achieve our goals a set of different software and low-level tools are used: SSH (to control nodes), TakTuk [START_REF] Claudel | TakTuk, adaptive deployment of remote executions[END_REF] (for a scalable command execution) and some standard Unix programs.

In the remaining part of this section, the three features of our approach are presented: the advantages of workflow representation, the modularity of the approach and the robust workflow execution. These features cover the main activities the experimenter has to cope with, that is: designing an experiment, implementing it and executing it, respectively. Each section starts with a problem that is addressed by that feature and then details of the solution follow.

A. Workflow-based description of experiments

One of main problems with published research is the lack of experimental verification of the presented ideas -substantial amount of research based on "engineering epistemology" does not devote much space to evaluation of presented ideas (only 36% of such publications dedicate more than 20% of space to evaluation [START_REF] Wainer | Empirical evaluation in Computer Science research published by ACM[END_REF]). Even if experiments are presented in a publication, they are often poorly documented using informal and ambiguous language. The lack of a formal description of experiments prevents any formal analysis or verification of correctness. Moreover, the existing practices (use of lowlevel scripts and manual control of experiment execution) make it very difficult to monitor the progress or instrument the execution of experiments.

Our approach to this problem consists in representing experiments using workflows. To achieve that, we provide a domain-specific language to define workflows built from activities. Activities can be implemented imperatively (using a low-level programming language), or declaratively as processes that use various patterns to group activities and other processes into more complex workflows. In other words, the processes orchestrate the execution of activities, including other processes, whereas imperative activities implement final actions. The workflow engine parses the description and builds an internal graph representation of workflows which maintains one-to-one relation with their original description.

1) Domain-specific language representation:

In our approach, workflows are described using a dedicated, domainspecific, declarative language. The building blocks are activities which can be grouped together into more complex workflows using special patterns for: sequential and parallel execution, looping over a set of variables, error handling, etc. (see Fig. 1 for an example). Almost every high-level feature of our approach is provided as a set of activities (including: acquiring resources for experiments, communication with nodes, analysis of data). For handling low-level tasks, the user is free to use traditional, imperative code.

The advantage of representing experiments in the textual form is that it does not diverge from existing practices which consist in describing experiments as a set of text files. Moreover, the human-readable text files can be easily copied, modified and published without any heavyweight or proprietary software.

2) Workflow representation:

The experiment written in the domain-specific language is a workflow which can be represented as a directed graph with annotated nodes. The expressive power is close to that of workflow management systems as it uses standard patterns that were identified in the literature [START_REF] Van Der Aalst | Workflow Patterns[END_REF]. Workflows can be built from user-defined activities or activities from a core library that includes: node management, data collection and statistics, logging, etc.

A workflow has a structure of a graph and is convenient to work with. Such a representation can be stored, modified and analyzed using well-known methods and algorithms. In particular, a static analysis of workflows can be done in order to identify their consistency and correctness (e.g., to ensure that all referenced activities exist). Moreover, workflows can be represented in graphical form which may be easier to understand than traditional ways to structure experiments.

The workflow representation enjoys a variety of monitoring possibilities. Information about execution times of activities is stored and can be analyzed to find bottlenecks, suspicious behavior or causal relations that affected the experiment. The monitoring data can be presented in a graphical form as well, for example as a Gantt chart.

B. Modular and extensible architecture

A common shortcoming of existing methods and tools is difficulty to extend their functionality or reuse them in a different context. Most of the tools support a single testbed and a require a dedicated software stack to function, for example. Similarly, many approaches suffer from limited possibilities to extend them, either by interoperating with external solutions or extending the capabilities of the framework itself. In both cases, the composability of them is important so that the modules can be used together in an arbitrary, flexible way.

In that regard, the workflow representation gives a powerful way to abstract any existing functionality, including one offered by external tools. The activities represent actions in a transparent way and can be grouped together to form more complicated workflows. Moreover, the workflows can be parameters for the execution of other workflows giving a powerful method of composing experiments from well-defined experimental patterns. Thanks to that architecture, most of the features of our approach are provided in a completely modular way.

1) Interoperability with testbeds and external tools:

To illustrate the interoperability of our approach, we show how the control over nodes of the experiment is implemented. In particular we show its independence from an underlying testbed and a way to achieve a scalable method of command execution.

A primary abstraction is that of a node which on the workflow level is treated just as any other variable. To interact with the node, the user uses a set of activities that provide a portable and scalable implementation of common operations. These operations include: command execution, copying files, retrieving files, etc. The method of accessing the nodes and software used to perform these operations is abstracted and transparent to the user.

The first benefit of this architecture is that the testbed used in the experiment can be easily changed by providing an alternative set of activities to perform all necessary actions on the new testbed. Another benefit is that it is equally easy to work with nodes at any level of the infrastructure. It does not matter if a node is accessible directly or via a gateway node of a testbed -from the workflow point of view both situations are exactly the same.

2) Composable experiments:

The experiments are workflows which form a hierarchical structure. There is always a topmost, root experiment that encompasses the process of the experiment. During its execution sub-experiments can be started, which may contain their sub-experiments and so on. Each experiment contains its own log, set of files and properties and remains mostly independent from its parent experiment. An example of such an experiment is presented in Fig. 2.

An experiment can take another experiment as a parameter and execute it, possibly multiple times. For example, a study of scalability may consists in measurement of performance of an application as the number of clients varies. Instead of modeling this experiment as a monolithic loop that exercises more and more nodes, we can model it as two distinct problems: (1) implementing a process that runs a benchmark with a given set of nodes and ( 2) implementing an experiment that executes it with more and more nodes. The high-level experiments can be used in an arbitrary way depending on the needs of a user. This encourages reuse of code and good scientific practices.

C. Workflow execution and failure handling

The ultimate goal of experiment execution is to finish it with a success, even in the presence of failures. Unfortunately, the execution of experiments in large-scale experiments is difficult due to their complexity and size. Large-scale experiments involving hundreds or thousands of nodes reach scalability limits of a platform (e.g., overload network hardware) and fail not only for understandable reasons (e.g., timeouts in communication protocols), but also in completely unexpected ways. Failures cannot be left unnoticed, because they could bias measurements and conclusions. However some failures may be intermittent and therefore not fatal to the experiment execution and its correctness. These characteristics of the domain of large-scale experiments call for proper analysis of how the experiments can fail and what can be done about that.

Error handling in our approach consists of special workflow patterns to handle failures and checkpointing. However, before discussing them, we must start with a discussion about failures during experiment execution. There are two types of failures:

• platform failures -caused by external technical problems, software or hardware bugs, etc., in an idealized environment platform failures would not happen; an example of a platform failure is a dropped HTTP connection due to an overloaded server, • workflow failures -a result of a wrong structure of the experiment or wrong assumptions which, contrary to the platform failures, happens even in an idealized environment; an example is an attempt to copy a file that does not exist.

The workflow execution can result in various outcomes: a success, a failure caused by the platform or a failure caused by the workflow. The platform failures can be further divided . An example of a composable experiment that measures performance while running with more and more "items" (DSL representation on the left, an example how scalability experiment executes :perf experiment with a list of 3 items, and final results of the example on the right). A sub-experiment passed as an argument is executed with more and more items (e.g., clients). Additional options specify an initial size, a step size, etc. The final value is a mapping between number of items and results obtained with them.

into intermittent and fatal ones. In practice, these two types of failures cannot be easily distinguished and standard approach consists in using timeouts and retries to decide if an intermittent failure is actually a fatal one.

The workflow representation offers many different patterns to group workflows, but the primary one is a sequence of workflows to execute in a fixed order. A sequence of workflows finishes with a success if all of them finish with a success. If any of them fails, the result of the sequence is a failure that caused its subworkflow to fail. Other workflow patterns have different semantics, for example a parallel loop fails with a compound failure if any of its parallel iterations fails. The compound failure groups all failures that caused the parallel iterations to fail (as there may be many of them).

Further on, we will say that two executions of workflows are equivalent if they lead to the same effect on the platform. It does not imply exactly the same state of the platform, but two states whose differences have no impact on the execution of the workflow and on the overall result of the experiment. The precise definition depends on the object of study and technical details. For instance, if hard disk performance is studied, any disk access may affect the state and hence experimental results. Therefore, executions that differ by the disk access patterns in general are not equivalent. However, if it is the network that is studied, the state of the hard disk can be arguably ignored.

We will say that a workflow is restartable if it can be executed repeatedly (with a success or a failure) in sequence without causing a workflow failure. In other words the workflow execution cannot break its own assumptions as a result of the execution. The example of an action that is not repeatable is moving of a file -the file will not be present the second time the action is run, breaking its very own assumption necessary for proper execution.

A workflow is idempotent if a sequence of its multiple successful executions is equivalent to a single successful execution. Observe that making a copy of a file is idempotent, but appending to a file is not. A workflow composed of idempotent constituents does not have to be idempotent. For example, the sequence (copy a b; copy c a) is not idempotent if files b and c differ. Similarly, an idempotent workflow is not necessarily composed of idempotent actions -the sequence (copy a b; move b c) is idempotent as a whole even though moving a file is not.

If a workflow is restartable and in presence of intermittent failures will nevertheless eventually succeed in a state equiv-alent to a single successful execution, then we will say that it is eventually successful. If it is also idempotent, then will say it is eventually idempotent.

The aforementioned properties are crucial in a discussion that follows and describes failure handling patterns and checkpointing. We show that with an adequate way of handling failures, one can still properly execute an experiment despite presence of failures.

1) Failure handling patterns: Users do not have to precise what constitutes a failure as any anomalous behavior is treated as such. On the other hand, they can force an error (e.g., as a result of monitoring), signaling the presence of a failure and delegating handling of it to failure handling patterns used in a workflow.

Any workflow can be embedded inside a failure handling pattern that will retry its execution a predefined number of times. The inner workflow should be eventually successful so that it can be restarted multiple times until finally it succeeds. If the limit of retries is reached, the problem is treated as fatal and the workflow finishes with a failure. The primary use for failure handling of workflows is to cope with intermittent failures which are a common problem in large-scale experiments. If, for example, many nodes use a shared resource (e.g., connection bandwidth, broadcast domain, etc.), some requests may fail due to timeouts, operating system limits, software and hardware bugs, etc. A common case is a highly parallel installation of software on many nodes -the bottleneck is the shared access to a single server with packages. Fortunately, the software installation process can be treated as restartable and hence our approach applies.

To sum up, if the experiment is built from eventually successful workflows and proper error handling is used, then the experiment can be eventually finished even in the presence of intermittent failures. Moreover, the results of the execution are equivalent to execution that experienced no failures at all. An example how the pattern is used in practice is shown in Fig. 3.

2) Checkpointing: A state of the experiment workflow can be saved in strategic, predefined places and restored later. A common use of that feature is to restart the execution from already reached checkpoints in a case of a failure, saving time and other resources. The experiments are often implemented iteratively, that is, by repeatedly designing and running the experiment until the researcher is satisfied with it. Thanks to 4. The principal workflow of the experiment consisting of 2 composable experiments and the main experiment (i.e., the HTTP server benchmark). The input of the experiment is a set of nodes provided by a testbed-specific workflow that can be replaced as we did in this paper. One node is designated as a server, the remaining act as clients. The scalability experiment takes a varying number of clients in a sequential loop (denoted with "=") and passes them to the minimal sample experiment. The minimal sample experiment executes the benchmark as many times as necessary to achieve a required statistical precision. checkpointing, the parts that are already finished do not have to be re-executed every time. Moreover, even if a complete experiment fails unexpectedly, the cause of the problem can be found, fixed and the experiment restarted.

Although checkpoints can be used anywhere in the main workflow, the properties of failures suggest that the placement of them is not arbitrary. More precisely, a use of checkpoint assumes that the remaining part of the workflow, executed after the checkpoint, is eventually idempotent.

Thanks to techniques like snapshotting of virtual machines or of the underlying file system, the checkpointing may include a state of the underlying physical platform where the experiment is executed. In that case, restarting a checkpoint consists in restoring the underlying platform before the state of the experiment workflow. The primary advantage is that the workflow following the checkpoint may assume a particular state of the platform and does not have to be even restartable. This functionality is not yet available in our implementation.

D. Applicability and limitations

Our approach is general and should apply to most experiments that evaluate questions regarding distributed systems, especially in grid, cluster and cloud infrastructures. Although most operations managing infrastructure are distributed, the workflow execution itself is centralized making it rather difficult to experiment with sensor networks and similar scenarios. Moreover, the workflow execution is automated and reaction to asynchronous events (in a timely fashion) is not possible (such a feature what greatly complicate checkpointing). As a result, the failures are handled when their effects are encountered, making the experiments that study failures in distributed systems cumbersome, but possible. Finally, the nodes involved in the experiment must provide a POSIX-like environment. These disadvantages are not inherent to our approach and can be addressed by hardening and extending our implementation.

III. EVALUATION

In this section, we will evaluate our approach with its implementation called XPFLOW. To this end, we designed an experiment that covers all relevant features and shows how they are used together. Our experiment is a basic performance analysis of a single server instance of nginx 3 HTTP server (version 1.2.1) with 4 worker threads, while serving a varying number of clients. As a benchmark, we used ApacheBench 4 . A single run of the experiment consists in measuring a throughput of a single HTTP server while serving many clients simultaneously for a timespan of 10 seconds. The result is measured in requests per second that the server was able to successfully deliver. The requested document is a short web page (that easily fits filesystem cache) and therefore we mostly measure the scalability and reliability of a request loop in the server. The operating system used to run all experiments is Debian 64-bit (wheezy) running a Linux kernel (version 3.2).

A. Workflow representation

The workflow representation of the experiment exploits various patterns to build experiments. We used various sequential and parallel loops to iterate over sets of nodes, parallel execution patterns to perform actions in parallel, error handling patterns to harden the execution of the experiment, etc. The workflow representation is verified before execution to check if there are any problems that can be detected before runtime.

Moreover, the workflows can be visualized in a graphical form which is some cases provides an advantage over a plain, textual representation.

The overview of the experiment is presented as a workflow presented in Fig. 4. That part is independent from the underlying testbed and normally is preceded by testbed-specific part that reserves physical nodes and prepares a testbed.

B. Modularity and extensibility

Our experiment is built from reusable, modular components. The exemplary experiment is built from 3 subexperiments with increasing depth of nesting:

• scalability experiment -runs a given experiment with an increasing number of objects (in our case these objects are clients), • minimal-sample experiment -given an experiment adaptively tries to find a minimal number of samples necessary to reach a desired precision, • the benchmark -the main experiment that benchmarks the performance of an HTTP server.

The nested structure of these components is clearly visible in Fig. 4.

The minimal-sample experiment exploits the fact that results are represented as variables passed between activities. In particular, the workflow may have a looping behavior that depends on a result of partially collected results (see Fig. 5 for illustration).

Moreover, we have shown that thanks to our abstractions it is easy to change the underlying testbed. All variations of the experiment share most of the structure, only the part related to deployment and bootstrapping the experiment differs.

1) Lightweight containers-based platform:

This platform was used to rapidly develop the experiment without necessity of using a real, shared testbed. The advantages are numerous: the ideas can be tested rapidly, the "testbed" resources are always available, the connection is much more reliable and fast. This feature can be described as a virtual testbed -a sandbox to test ideas, iterate rapidly and spot early problems.

On a technical level, this feature is implemented using Linux Containers 5 . All instances share a common base image using Copy-On-Write features of Btrfs filesystem, whereas their virtual network interfaces are bridged together and with the host network card. The process of launching containers is implemented as a separate XPFLOW workflow.

The machine used to host containers and develop the experiment was a typical desktop machine running Linux. The results for this "sandbox" testbed are not presented, since they are neither comparable with other testbeds nor can achieve a similar size of an experiment as them.

2) Grid'5000-based platform: This experiment is the most common case that we are interested in. The nodes used in the experiment come from two different clusters. One consists of nodes that are equipped with Intel Xeon X3440 processors, 16 GB RAM, magnetic hard disk and Gigabit Ethernet network cards. The second one consists of machines that only significantly differ in that they have Intel Xeon L5420 processors. We used a total number of 199 nodes during the experiment. This experiment uses a small set of activities to interact with Grid'5000 testbed. The activities are used to submit user reservations, obtain information about associated nodes and install an operating system used during the main experiment. This reusable part amounts to few hundred lines of code, whereas the non-reusable boilerplate added to the original experiment is less than 30 lines.

3) Cloud-like platform: This variation of the experiment uses KVM-virtualized instances of Linux operating system. The physical nodes used to host the virtual machines are the same as those used in the previous experiment and are running Linux as well. The experiment is the most challenging one and the most time-consuming due to a fairly complicated installation process. A complete listing of workflows used to run the deployment is presented in Fig. 6.

The multi-level setup complicates communication with nodes at different levels of the installation, as they may not be reachable directly from the host running XPFLOW. In our case, we use one Grid'5000 node as a gateway to all virtual machines (it is also a gateway that acts as a router for virtual machines). Thanks to our abstractions the whole process is completely transparent (see Fig. 7).

The physical infrastructure that hosts the cloud-like installation is the same as in the Grid'5000-based experiment. Each physical node hosts two VMs per each core available on that node (that is, each core is shared by 2 instances). During the experiment we started 2034 virtual machines (and another VM dedicated as a web server).

Each virtual machine has a one virtual core and 1 GB of RAM available with the exception of the virtual machine that hosts the HTTP server -this node has 4 cores and 4 GB of RAM. Network interfaces of the virtual machines are bridged to the physical Ethernet network. This cloud-like infrastructure is in many respects similar to one of possible configurations offered by OpenStack6 cloud software and arguably serves as a good model to study clouds with many deployed instances. This variation of experimental scenario required to design and implement workflows to deploy the virtualized infrastructure. This (reusable) process amounts to 315 lines of code. Moreover, we had to prepare a KVM image used by the virtual machines. It required an installation of Debian Linux with necessary software. The image can be reused in different experimental scenario.

C. Failure handling

Our method takes a strict, almost paranoid approach to failure handling -by default every single problem will cause the experiment to fail. However, the user can manually define a retry policy for parts of the workflow forcing them to restart automatically if a failure happens during experiment execution, or by restarting the experiment from a checkpoint by manual intervention.

Our approach to error handling allowed to cope with some intermittent failures that happened during the experiments. In particular, the occasional problems with running a benchmark were handled by restarting the part of workflow that failed use :g5k import :http, "../http/common.rb" # import the main exp. IMAGE = "/home/tbuchert/public/wheezy.qcow2" # shadow package installation process :"http.deploy" do |master, slaves| execute master, "service nginx start" end process :install_master do |master| distribute master, "files/master.sh", "/tmp/" #! text="Copy files" execute_many master, "bash /tmp/master.sh" #! text="Configure the node" end process :install_slaves do |slaves| distribute slaves, "files/*.sh", "/tmp/" #! text="Copy files" distribute slaves, $ssh_key, "/tmp/" #! text="Copy SSH key" chain_copy slaves, IMAGE, "/tmp/" #! text="Copy KVM image" execute_many slaves, "bash /tmp/client.sh" #! text="Conf. Three nested levels can be distinguished (XPFLOW engine, Grid'5000 nodes, KVM instances). The red, dashed path describes a connection used to access and control virtual machines. and problems have been mitigated (see use of try workflow pattern in :run nginx workflow in Fig. 6).

The experiment was structured according to the workflow properties described before. For example, the checkpoints were used in proper places to save time and ease the process of debugging. During development of the experiment, they were occasionally used to fix a bug in an experiment that caused it to fail and then restore a state from a previously attained state.

D. Results

The obtained results are presented in Fig. 8. For each data point, the experiment was repeated at most 10 times to obtain a sample mean value that is at most 3% from the real mean value with 95% confidence. The data points that failed to converge are presented with intervals of the same confidence, but of a larger size.

Unsurprisingly, the results obtained on a physical testbed are better than on a virtualized environment. The peak performance reached more than 30000 requests per second for 180 clients and to our knowledge would only improve if more nodes were used. We were however limited by the size of physical clusters available to us.

The results obtained in a virtualized environment are not as good. The performance reached around 10000 requests per second in the range that we experienced with. The lower performance is due to overhead of CPU and network interface virtualization. Moreover, one has to remember that we measure the performance from the perspective of clients -their performance of crafting and sending requests has an effect on the performance as well. We noticed as well that above a number of 900 simultaneous clients some requests were not served successfully, but the overall performance remained the same.

All in all, the nginx HTTP server offers very good performance and scalability while serving a large number of clients. Thanks to a cloud-like infrastructure we were not limited by the size of physical infrastructure and were able to run our experiment with more than 2000 virtualized nodes. The raw results, XPFLOW software and the experimental workflows are available at http://www.loria.fr/ ∼ buchert/xpflow2013.tgz.

IV. RELATED WORK

XPFLOW is a result of an interdisciplinary merger of three domains: it is an experiment control system, it shares common ideas with scientific workflows and borrows from workflow management.

A. Experiment control systems

Experiment control systems have an objective of managing the execution of experiments so that the process is easier and more reliable. Among many ways to differentiate between them, one can look in particular at: the way they treat the problem of describing and structuring experiments, how modular and extensible their architecture is and how the reliability and correctness of experiments are achieved.

1) Experiment description:

The methods of representing experiments range from purely imperative representations to declarative ones based on high-level models and languages. The former approach usually extends an existing programming language with a set of routines to control experiments, effectively turning the experiment description into a program. This representation, although theoretically equivalent to any possible computation, suffers from few problems: it is lowlevel, cannot be thoroughly analyzed and is unsuitable for expressing complex patterns. Therefore most of the tools use a dedicated high-level representation to describe experiments. Although it solves some problems of low-level representation, it does not come without some of its own: one has to learn to use it (at the expense of time), the language may be limited in its expressiveness or be difficult to extend it with existing software or tools.

The former group is represented by Expo [START_REF] Videau | Toward an experiment engine for lightweight grids[END_REF], a tool that extends an imperative programming language (Ruby) with a set of abstractions to control nodes. The experiment is more readable, but has no high-level structure that can be analyzed. Moreover, failures are managed on a level of a programming language, which is difficult and may be inconvenient. Many tools use a low-level programming language to describe highlevel constructs, for example Emulab Workbench [START_REF] Eide | Integrated scientific workflow management for the Emulab network testbed[END_REF] uses it to describe an experimental platform, and OMF [START_REF] Rakotoarivelo | OMF: a control and management framework for networking testbeds[END_REF] to provide an event-based description. Another approach is featured by Splay [START_REF] Leonini | SPLAY: distributed systems evaluation made simple (or how to turn ideas into live systems in a breeze)[END_REF] which offers a description language which is intentionally similar to pseudo-code used commonly in scientific papers in distributed systems research that hides common difficulties of experimentation. Such a description allows to rapidly prototype an experiment and run it on a model of a platform, but is mostly limited to one-time studies. Finally, solutions like Weevil [START_REF] Wang | Automating Experimentation on Distributed Testbeds[END_REF] or Plush [START_REF] Albrecht | Loose synchronization for large-scale networked systems[END_REF] turn to high-level, declarative descriptions of experiments.

Our approach uses a special language to build workflows, but is conceptually independent from any programming language. The high-level workflow representation is easy to understand, expressive, easy to analyze and to work with.

2) Modular architecture: The means of extending a given approach vary greatly and are related to the way the experiments are represented. Approaches based on high-level representations may be constrained by this very fact, whereas low-level tools may offer no convenient way to extend them.

Approaches concerned mostly with one-time studies (like Splay [START_REF] Leonini | SPLAY: distributed systems evaluation made simple (or how to turn ideas into live systems in a breeze)[END_REF]), or tools dedicated to a particular platform (Emulab Workbench [START_REF] Eide | Integrated scientific workflow management for the Emulab network testbed[END_REF] or Plush [START_REF] Albrecht | Loose synchronization for large-scale networked systems[END_REF]) do not tackle this problem directly. Solutions similar to Expo [START_REF] Videau | Toward an experiment engine for lightweight grids[END_REF] offer theoretically unlimited capabilities, but are not modular by design and extending them is difficult. High-level solutions tend to be self-contained and difficult to extend as well, but also offer much more extensive built-in functionality (e.g., Weevil [START_REF] Wang | Automating Experimentation on Distributed Testbeds[END_REF] offers a powerful workload generation functionality). Finally, an example of OMF [START_REF] Rakotoarivelo | OMF: a control and management framework for networking testbeds[END_REF] shows that even a platform-dedicated and relatively high-level approach may be extensible, as is shown by its instrumentation, logging and interoperability capabilities.

Our approach proved to be very flexible as it can be extended in various ways. Not only the functionality can be extended with external software in a modular way, but even the testbed can be replaced if necessary. Moreover, common patterns can be reused as has been shown with composable experiments.

3) Robustness: Due to non-deterministic behavior of distributed systems the failures must be taken into consideration by every approach. The failure handling can be automatic or manual, the former typical to methods that are dedicated to a particular platform and latter to methods constrained by their general applicability. Nevertheless, every solution has some means to cope with problems with scalability and intermittent failures.

Among other features, Plush [START_REF] Albrecht | Loose synchronization for large-scale networked systems[END_REF] can ensure proper connectivity between nodes, Emulab Workbench [START_REF] Eide | Integrated scientific workflow management for the Emulab network testbed[END_REF] and OMF [START_REF] Rakotoarivelo | OMF: a control and management framework for networking testbeds[END_REF] can verify that the platform complies to a specification -a set of features that would be otherwise difficult to provide if not for the tight integration with the platform. For similar reasons, Splay [START_REF] Leonini | SPLAY: distributed systems evaluation made simple (or how to turn ideas into live systems in a breeze)[END_REF] can hide file management and related problems, whereas general solutions like Expo [START_REF] Videau | Toward an experiment engine for lightweight grids[END_REF] turn to mostly manual error handling.

To our knowledge no approach has defined precisely the way the failures are handled and coped with. Although some solutions claim to handle failures and scalability problems, neither the ways to achieve this are explicitly stated nor the effects on results of the experiment discussed. Our workflowinspired approach provides a well-defined framework built around the idea of correct execution and gives a set of properties and methods necessary to achieve this objective.

B. Scientific workflows

Scientific workflows are a broad term encompassing tools aiming at automation of scientific processes. They are mainly used to run complex computations on preexisting data using grids, and are not used to manage experiment execution, especially experiments in distributed systems research. Normally, those systems are provided with GUIs that enable non-expert users to easily construct their workflows as a visual graph. Such graph describes the complete flow of the computation and the dependency between tasks and data generated (i.e., provenance). A prominent feature of scientific workflows is their independence from the computing infrastructure. Thanks to that the failures of the platform are handled in an automatic and invisible way (other types of failures may not be, however). Some examples of scientific workflows are: Kepler [START_REF] Ludäscher | Scientific workflow management and the Kepler system[END_REF], Taverna [START_REF] Hull | Taverna: a tool for building and running workflows of services[END_REF] and Vistrails [START_REF] Callahan | VisTrails: visualization meets data management[END_REF].

The scientific workflows are data-oriented and a distributed system underneath (usually a computational grid) is merely a tool to efficiently process data, not an object of a study. Although our approach represents the experiments as workflows as well, there are incompatible differences. First, scientific workflows are data-oriented whereas in our approach they represent actions performed on the platform. Second, the platform is not hidden from the experimenter, but is a central and explicit element of the workflow execution. Last, the workflows in our approach are based on a different formalism.

C. Workflow management

Traditionally related to workflows as well, is the domain of Business Process Management and Modeling (BPM) [START_REF] Weske | Business Process Management: Concepts, Languages, Architectures[END_REF]. Business processes are found in business organizations and the goal of BPM is to model, understand and improve them using appropriate techniques and tools, most of them computer-aided nowadays. A viewpoint that we adhere to is to think about BPM as a management discipline (so it's neither a technology nor has anything to do with computer science) and workflow management as a computer science discipline supporting it [START_REF] Ko | A computer scientist's introductory guide to business process management (BPM)[END_REF]. In that sense, we are more interested in the latter, but parallels between experimentation and BPM has been observed as well [START_REF] Buchert | Leveraging business workflows in distributed systems research for the orchestration of reproducible and scalable experiments[END_REF]. From the discipline of workflow management we primarily reuse the common workflow patterns and the execution model.

V. CONCLUSION AND FUTURE WORK

In this paper we have shown how research in distributed systems research can profit from the workflow representation of experiments. Our interdisciplinary approach to control of experiments is based on workflow management and scientific workflows. Its advantages include flexible and modular representation that enables building experiments from reusable components and transparent interoperation with testbeds or external software. Our approach has a well defined model of execution based on BPM workflow patterns and robust failure handling. To verify claims about our approach, we designed and conducted an exemplary, large-scale experiment.

In future work, there are a few aspects we would like to concentrate on. First, we would like to push the scalability of our experiments further (say, some tens of thousands of nodes) and address shortcomings of our implementation. For this purpose, we would like to conduct an analysis of XPFlow system itself, including its overhead and performance. Second, various ways to extend our approach should be explored: recording provenance of data in experiments, adaptive methods for execution of experiments, advanced checkpointing, for example. Moreover, we plan to make the user interface more friendly by providing visualization of results and interoperability with other data analysis software. Finally, we want to release XPFLOW so that scientists could profit from our approach and give us a valuable feedback.

1 process :setup_experiment do |clients, server| 2 parallel do 3 forall clients do |node| 4 run

 1234 Figure 1. An example of a mapping between DSL and a workflow. Different language-level constructs (parallel, forall, sequence) result in respective workflow patterns (parallel execution of subworkflows, a parallel loop and sequential execution of activities). The workflow notation is based on Business Process Model and Notation (BPMN) with some extensions (the "ampersand notation" for a parallel loop, for example).

13 endFigure 2

 132 Figure 2. An example of a composable experiment that measures performance while running with more and more "items" (DSL representation on the left, an example how scalability experiment executes :perf experiment with a list of 3 items, and final results of the example on the right). A sub-experiment passed as an argument is executed with more and more items (e.g., clients). Additional options specify an initial size, a step size, etc. The final value is a mapping between number of items and results obtained with them.

1 process :prepare_experiment do |machines| 2 forall machines do |machine| 3 try :retry => 5 4 run :install_software, machine 5 end 6 end 7 endFigure 3 .

 12345673 Figure 3. Error handling during parallel software installation. If there is an intermittent failure during installation of software on each node, the process will be retried 4 times more until it will become an unrecoverable failure. Without the try block any failure what propagate to parent workflows and, if not handled by them, would stop the experiment.

1 process :minimal_sample do |precision| 2 first_samples 13 endFigure 5 .

 12135 Figure 5. A simplified implementation of minimal-sample experiment. First, 3 measures are taken and then the experiment is repeated until a desired precision is reached.

Figure 6 .

 6 Figure 6. A full listing of workflows used to deploy a cloud-like testbed and an autogenerated graphical representation of the deploy cloud workflow (thanks to the annotations inside comments). The blocks defined with process keyword describe workflows, whereas a block defined with activity contains a low-level Ruby code. Various patterns discussed in the paper are presented: failure handling, checkpoints, parallel execution and composable experiments.

Figure 7 .

 7 Figure 7. A hierarchical structure of nodes used in the cloud-like experiment. Three nested levels can be distinguished (XPFLOW engine, Grid'5000 nodes, KVM instances). The red, dashed path describes a connection used to access and control virtual machines.

Figure 8 .

 8 Figure 8. Performance (measured in requests per second) of the nginx HTTP server while serving a varying number of clients (X axis has a logarithmic scale). Results for two different testbeds are shown: one built using two physical clusters and one cloud-like with virtualized nodes. Each data point is presented with its 95% confidence interval.

  FOR EACH k-SUBSET OF n NODES k ∈ {1, . . . , n}

	SCALABILITY EXPERIMENT
	MINIMAL SAMPLE EXPERIMENT
	DRAW INITIAL SAMPLE	ESTIMATE MIN. SAMPLE SIZE (s)
		STORE RESULT FOR k NODES
	RUN THE BENCHMARK s TIMES	RETURN DATA VECTOR
	Figure

= =

http://puppetlabs.com/

http://www.opscode.com/chef/

http://nginx.org/

http://lxc.sourceforge.net/

http://www.openstack.org/

ACKNOWLEDGMENTS

Experiments presented in this paper were carried out using the Grid'5000 experimental testbed, being developed under the INRIA ALADDIN development action with support from CNRS, RENATER and several Universities as well as other funding bodies (see https://www.grid5000.fr, [17]).


Stéphane Martin

Tomasz Buchert

Pierric Willemet

Olivier Richard

Emmanuel Jeanvoine

Lucas Nussbaum

Scalable and Reliable Data Broadcast with Kascade

Keywords: data broadcast, multicast, large scale, faulttolerance

la diffusion de documents scientifiques 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.

I. INTRODUCTION

Over the recent years, many areas of scientific research and industry shifted to a data-driven model, which paves the way to many ground-breaking changes in today's society, science and engineering. However, this important paradigm evolution changes our vision of our computing infrastructure: due to the exponential growth of enormous amounts of information, the storage and the management of data emerged as the new bottleneck for many applications.

Several very different ways to organize data have been designed and used over the years. Traditional storage servers with large RAID array of disks continue to be used, but are often aggregated into storage clusters as part of a distributed file system such as Lustre [START_REF] Schwan | Lustre: Building a file system for 1000-node clusters[END_REF], [START_REF] Braam | The lustre storage architecture[END_REF], or more recently Glus-terFS [START_REF]Glusterfs[END_REF] or Ceph [START_REF] Weil | Ceph: A scalable, high-performance distributed file system[END_REF]. Those file systems provide a POSIX interface, or at least an interface that is very similar to POSIX. Switching from POSIX-compliant semantics to other logical organizations of data enables higher performance, scalability and fault-tolerance, as demonstrated by NoSQL databases such as Apache Cassandra [START_REF]Apache cassandra[END_REF] and MongoDB [START_REF]mongodb[END_REF], or the MapReduce [START_REF] Dean | Mapreduce: simplified data processing on large clusters[END_REF] programming model.

A common feature of those new solutions is that they move away from centralized and expensive storage servers to more commodity hardware. Similarly to the move from large supercomputers to clusters of inexpensive workstations [START_REF] Becker | Beowulf: A parallel workstation for scientific computation[END_REF], recent solutions leverage cheaper and off-the-shelf hardware found in standard machines to achieve higher scalability, addressing the necessary fault-tolerance concerns.

In this paper, we address the case of broadcasting a large amount of data from one storage system to a large number of nodes. This time-critical operation is typically used as the first operation of distributed data analysis, in order to distribute the data being analyzed to each node involved so that it can then be accessed from local storage. A similar operation is also required in other contexts, such as the efficient broadcast of system images in Clouds or HPC clusters -a use case that was our original motivation as part of our work on Kadeploy [START_REF] Jeanvoine | Kadeploy3: Efficient and Scalable Operating System Provisioning[END_REF].

The main advantage of this approach is that it limits the burden on the source storage server -as we will show later, broadcasting to hundreds of nodes can be as easy as sending the data to one node, if done properly. But this approach has two main intrinsic limitations. First, it is only applicable to cases where all nodes require the same data, or at least enough identical data to justify sending everything to all nodes, and letting each node filter the data to store only the pieces required locally. Second, the local storage capacity on each node needs to be large enough to contain all the required data. We will show later that, while those limitations prevent from using this approach in some cases, the remaining cases can benefit from great performance.

In this paper, we present Kascade, a solution for the broadcast of data to a possibly large set of compute nodes. Kascade organizes nodes in a pipeline to achieve high scalability, and includes some fault-tolerance mechanisms to handle the failure of nodes during the transfer.

The remainder of this paper is organized as follows. Section II outlines the challenges presented by data broadcasting (Section II-A), and describes the related works (Section II-B). Kascade is then presented in Section III, and is evaluated in Section IV. Finally, we conclude this paper in Section V.

II. CONTEXT

In this section, we describe the main challenges encountered when broadcasting data, and then present the existing solutions that aim at addressing this problem.

A. Challenges

Data broadcasting needs to overcome several challenges. 1) Local storage performance: First, the local storage on nodes (hard disk drives or SSD drives) is a performance bottleneck: a 7200 RPM SATA drive typically provides a raw write throughput of 100 MB/s, lower than the Gigabit Ethernet bandwidth, and the fastest SSD drives provide from 500 MB/s to 600 MB/s, still much lower than the 10 Gbit/s Ethernet bandwidth. Additionally, when files are written to a file system (and especially small files), performance is even lower. Therefore, one key requirement is that receivers start core switch top-of-the-rack switch nodes Fig. 1. Fat tree network: links between the core switch and the top-of-the-rack switches are fatter (higher available bandwidth) than the ones between the top-of-the-rack switches and the nodes. writing data as soon as possible (that is, as soon as data starts arriving to nodes), rather than waiting for the full data to be received. Moreover, special care should be taken to write data in an optimal way -avoiding disk seeks, for example. Our tests in the context of our work on Kadeploy [START_REF] Jeanvoine | Kadeploy3: Efficient and Scalable Operating System Provisioning[END_REF] showed that writing a whole file system dump with e2image was much faster than writing the equivalent file system content using tar thanks to the use of sequential writes avoiding disk seeks, instead of numerous file operations (open, close, permissions change, etc.).

2) Efficient use of fat tree networks: Most clusters networks are hierarchical, fat tree networks [START_REF] Leiserson | Fat-trees: Universal networks for hardware-efficient supercomputing[END_REF] such as the one pictured in Figure 1. It is very common for such networks that core links (the ones interconnecting top-of-the-rack switches to the main networking equipment) suffer from under-capacity when all the nodes belonging to the same switch communicate with nodes on other switches. As an example, for the experiments described in Section IV, each top-of-the-rack switch is connected to between 30 and 35 nodes using a 1 Gbit Ethernet link, but is only connected via one 10 Gbit link to core switch. One could argue that this is bad infrastructure design. Actually, economical concerns often prevent from purchasing large centralized and non-blocking Ethernet switches. Data broadcast solutions need to be aware of this topology, and preserve core links by taking into account nodes locality.

3) Fault-tolerance: Failures are an important problem in the context of large-scale infrastructures. When dealing with data broadcast, the importance of this problem is even higher, as [START_REF] Schwan | Lustre: Building a file system for 1000-node clusters[END_REF] this operation relies heavily on hard disk drives, which, as mechanical parts, are particularly prone to failures [START_REF] Schroeder | Disk failures in the real world: What does an MTTF of 1, 000, 000 hours mean to you?[END_REF], [START_REF] Pinheiro | Failure trends in a large disk drive population[END_REF]; (2) this operation is often executed early in the execution of a task, possibly after the nodes were turned off or idle, so it is more likely to encounter malfunctioning nodes than an operation executed at the end of tasks. Data broadcast solutions need to be able to properly detect and handle the failure of a node.

B. Related works

Broadcasting has been the subject of a lot of attention both from a theoretical and from a more practical point of view.

From a theoretical point of view, several algorithms based on binomial trees have been designed [START_REF] Johnsson | Optimum broadcasting and personalized communication in hypercubes[END_REF], [START_REF] Kesavan | Optimal multicast with packetization and network interface support[END_REF], which have optimal or near-optimal performance on contention-free networks. However, this is not the case of Ethernet networks, even with a single switch, as the backplane switching capacity is often a bottleneck. Pipelined broadcasts algorithms are more interesting in such networks, but topology-unaware pipelined broadcast performs poorly, as shown in [START_REF] Patarasuk | Techniques for pipelined broadcast on ethernet switched clusters[END_REF], which evaluates pipelined broadcast algorithms in MPI implementations.

High performance network fabrics also provide special operations that can be leveraged to achieve efficient broadcast. In [START_REF] Liu | Fast and scalable MPIlevel broadcast using infiniband's hardware multicast support[END_REF] and [START_REF] Hoefler | A practically constant-time mpi broadcast algorithm for large-scale infiniband clusters with multicast[END_REF], implementations of the MPI_Bcast operation on top of InfiniBand hardware multicast support are proposed.

IP multicast can also be used to achieve reliable broadcast. UDPCast [START_REF]Udpcast[END_REF], used by SystemImager and other disk cloning solutions, is the most popular implementation of this idea. However, the use of multicast has a number of drawbacks:

• Multicast support is usually disabled by default in network switches, requiring special configuration that might not be possible in hosted environments (e.g., cloud computing); • A central allocation of the multicast groups is required to avoid conflicts between applications -or sub-optimal solutions such as randomized multicast addresses or perapplication packet tagging could be used; packets to send). During our experiments, we were unable to get it to work reliably. Also, in that mode the sender is not able to know if the receivers have correctly received the data. In the bidirectional mode, receivers re-request lost packets at the end of the transfer, which can lead to "ACK-implosion" at large scale as pointed in [START_REF] Liu | Fast and scalable MPIlevel broadcast using infiniband's hardware multicast support[END_REF]. P2P-based approaches are also appealing. In [START_REF] Dichev | Mpi vs. bittorrent: switching between large-message broadcast algorithms in the presence of bottleneck links[END_REF], a BitTorrent-based broadcast is compared to a MPI-based implementation. Authors conclude that BitTorrent performs better in heterogeneous networks (networks with bottleneck links). However, in their experiments, BitTorrent only achieves a maximum throughput of about 12 MB/s, which is very disappointing as the bottleneck link in the experiment was a 1 Gbit/s link. Our own experiments with BitTorrent [START_REF] Cohen | Incentives build robustness in bittorrent[END_REF] showed that its verbose protocol and its complex mechanisms (such as tit-fortat) incur a strong performance penalty on high-performance networks.

Finally, Ka [START_REF] Augerat | A scalable file distribution and operating system installation toolkit for clusters[END_REF] (which is an ancestor of Kascade, as it was also developed in the context of the Kadeploy project), Dolly [START_REF] Rauch | Optimizing the distribution of large data sets in theory and practice[END_REF] and Dolly+ [START_REF] Manabe | Disk cloning program 'dolly+'for system management of pc linux cluster[END_REF] are pipelined broadcast solutions that are similar to what we will present in this paper. However, (1) Dolly and Dolly+ were not evaluated at large scale (at most ten nodes); (2) we tried to evaluate Dolly+ by ourselves, but its compilation failed, and its authors were not able to provide a solution; (3) Dolly and Ka do not provide any fault-tolerance mechanism, and whether Dolly+ provides one is unclear: none is mentioned [START_REF] Manabe | Disk cloning program 'dolly+'for system management of pc linux cluster[END_REF], but one is mentioned in a presentation 1 -however, the corresponding code cannot be found in the archive available on Dolly+'s website 2 , so the presentation might describe planned future work at the time.

III. PIPELINED BROADCAST WITH KASCADE

In this section, we present Kascade, a pipelined and fault tolerant file or stream broadcast tool. Kascade leverages standard network technologies (TCP/IP), is written in Ruby, and provides a friendly command-line interface. Figure 2 presents some Kascade use cases.

In the following, we describe how Kascade organizes nodes in a topology-aware pipeline to achieve performance and 1 http://corvus.kek.jp/ ∼ manabe/pcf/dolly/dolly.ppt 2 http://corvus.kek.jp/ ∼ manabe/pcf/dolly/index.htm scalability (III-A), and how it leverages efficient external tools to control destination nodes and initiate the transfer (III-B). Then, we describe Kascade's protocol (III-C) and its support to handle node failures.

A. Topology-aware pipeline

In order to transfer data from the origin node to all the destination nodes, Kascade builds a pipeline: each node receives data from the previous one, and forwards it to the next one.

As described in Section II-A2, the pipeline must be built according to the underlying network topology to avoid saturation of some network links (e.g., inter-switch links).

Kascade assumes that the logical ordering of nodes (the numbers in their host names) matches the underlying physical topology (that is, that nodes 1 to 30 are on the first switch, that nodes 31 to 60 are on the second switch, etc.). Based on that assumption, Kascade sorts the nodes according to their numbers to achieve maximum performance. If needed, it is also possible to specify a custom sort order (if the nodes numbers do not reflect the underlying topology).

As shown in Figure 3, each link is only used once in each direction, avoiding network congestion since network links support nowadays full-duplex communications.

A final connection is performed at the end of the transfer by the last node to the first node in order to send list of failed nodes (if any).

B. Efficient start up and control of destination nodes

One crucial operation to ensure scalability is the start up of Kascade on every node. For that operation, Kascade relies on external tools such as TakTuk [START_REF] Claudel | Taktuk, adaptive deployment of remote executions[END_REF] (by default) or ClusterShell [START_REF] Thiell | Clustershell, a scalable execution framework for parallel tasks[END_REF]. A fallback mode using pure SSH is also available.

To initiate the transfer, Kascade first copies itself and the list of nodes to all the target nodes. This copy is performed using TakTuk or ClusterShell, as they are efficient-enough to copy small files -we will show in Section IV that they are not for larger files.

Then, Kascade starts itself on each receiving node. Taktuk provides two ways to connect to nodes: an adaptive tree, where nodes already reached are used to connect to additional The receiver sends to the previous node an acknowledgment to signal that the report has been sent to the first node Fig. 4. Kascade protocol messages nodes, and a windowed mode where the root node connects to each other node. The adaptive tree distribution is much more efficient [START_REF] Claudel | Taktuk, adaptive deployment of remote executions[END_REF], but is not able to handle failures of nodes in the middle of the tree. Therefore, Kascade uses the windowed mode by default. ClusterShell provides a similar windowed mode, and a treebased mode is planned to be included in the future [START_REF] Thiell | ClusterShell, python library and tools for scalable cluster administration[END_REF].

C. Protocol

TCP/IP provides applications with reliable and ordered data transfer between two hosts. On top of TCP/IP, Kascade builds a pipeline overlay to perform efficient data transfer among a large number of nodes.

Here is a naive version of a fault-tolerant pipelined broadcast protocol. The idea is that a node n i in the pipeline forwards data to its neighbor n i+1 until a failure is detected.

n 1 n 2 n 3
TCP connection TCP connection GET 0 GET 0 REPORT z REPORT z PASSED PASSED Fig. 5. Message sequence chart of communication between three nodes In this case, n i connects to n i+2 and recovers the transfer. This is pretty easy to understand but this does not work.

Actually, there are two issues: 1) the global size of the data is required from the beginning in order to know if the transfer is finished or if the previous node has just crashed. Being able to know at the beginning the amount of data is not an acceptable constraint since we want to support the read from the standard input stream because this is very useful in many situations; 2) there is no way to inform all nodes that the user has requested the end of the transfer. Indeed, in the naive protocol, each node has to wait for the entirety of data. When the connection is closed, they can not differentiate between a failure and a user interruption. Thus, we propose to enhance this naive protocol by adding control instructions. Basically, this consists in defining several messages that correspond to the transitions in the protocol.

The complete list is presented in Figure 4.

Instead of sending raw data as stated in the naive protocol, Kascade splits the stream into chunks. It begins to send a DATA message that contains the size s of the chunk and the chunk itself. Once s bytes are read, the node waits another message that can be either another DATA message (next chunk reception), an END message, or a QUIT message if the user has interrupted the transfer. Splitting the sent data into chunks is also useful to enable the transfer of data without knowing its size beforehand, in the case of streaming the output of another process.

After the end of the transfer the report is sent in a RE-PORT message. The report contains all detected node failures occurred on the previous nodes (considering the pipeline topology). Once the report has been successfully forwarded through the pipeline to the first node (i.e. the sender), the node informs the previous node that the report is sent with PASSED message. After sending a such message, a node can quit.

A message sequence chart of the transfer between three nodes without error is shown in Figure 5.

D. Fault tolerance

1) Fault detection: Kascade uses two mechanisms to detect failures. The first one is the syscall error catching on read() or write() functions. The second one is the use of timers on critical operations. Indeed, a timeout is triggered when the next node stops to read the stream. Such situation can happened in several cases: the next node has crashed, a subsequent node has crashed causing the next node to hang until that crash is detected, the network is congested, etc. To differentiate these cases, Kascade connects to its first alive neighbor and sends a ping message. If the neighbor answers quickly, the sender assumes the receiver is still alive, and the sender waits again that the write finishes. If the node does not respond, it is considered as dead and the sender connects to its next neighbor.

2) Recovery: Upon failure of n i , the sender looks for the next available node n j (in case of multiple adjacent failures n j is not n i+1 ). It is possible that the data sent already by n i has not yet been sent from n j-1 to n j . In this case, n i needs to resend these missing parts to n j . To this end, a Kascade node keeps some data chunks in memory, so it is able to re-send them if needed.

When n i-1 connects to the host n j , n j answers with GET(offset) and n i-1 transmits the required data from this offset. Because n i-1 is probably in advance with respect to the amount of data already sent, it needs to pick data in memory.

If too much data is lost, for example in the case of too many simultaneous node failures, and if the first node reads a file, n j sends a PGET message to the first node in order to get the missing data. If the first node reads from a stream, its buffer probably does not contain relevant data anymore. Thus the transfer fails on n j and on all following nodes in the pipeline. In this case the first node sends a FORGET message, n j and

n 1 n 2 × n 3 TCP connection TCP connection GET 0 GET 0
Kascade is our approach to data distribution and is described in Section III. The version used in our experiments is 0.1.5.

TakTuk is a tool for large-scale remote execution and file distribution using an efficient tree-like topology constructed over the set of nodes (the arity of the tree is configurable). We evaluate two different TakTuk overlays: a tree of arity 1 (i.e., a tree that degrades into a chain, shown as TakTuk/chain) and one that is a tree of arity 2 (TakTuk/tree). The version used is 3.7.4.

UDPCast is a file transfer tool that can send data simultaneously to many destinations on a LAN using IP multicast. It features various methods and enables a fine-grained tuning of parameters. In our evaluation, we use a default mode that uses a feedback to coordinate data transmission. UDPCast features another mode that allows a fully uncoordinated transfer, but we found it difficult to properly tune and very unreliable. The version used is 20120424.

MPI Broadcast is a home-made implementation of data distribution method that uses MPI runtime and MPI primitives (mostly MPI_Bcast) for distribution. The algorithm uses a 1 MB size buffer to send consecutive fragments of a file to participating nodes. The advantage of using MPI is that the implementation is portable to different network technologies, as long as MPI runtime supports them. In fact, in our experiments two ways of execution are considered: with Ethernet (MPI/Eth) and with InfiniBand (MPI/IB). The implementation of MPI used to conduct the experiments is Open MPI (version 1.4.5).

The operating system is Debian/Linux 7 with kernel version 3.2.0. The experiments were run on different clusters of the Grid'5000 infrastructure [START_REF] Cappello | Grid'5000: a large scale, reconfigurable, controlable and monitorable Grid platform[END_REF].

All figures show average, normalized bandwidth as a function of client number (the node initiating the transfer is not included in that number), excepted the last one (Section IV-G) dedicated to the evaluation of fault-tolerance which shows the average normalized bandwidth under several failure conditions. The bandwidth is computed as a size of a file being transmitted divided by the time required to finish transmission. The results are presented with their respective 95 % confidence intervals according to the Student's t-distribution.

Our experiments were conducted using XPFLOW experiment workflow engine [START_REF] Buchert | Leveraging business workflows in distributed systems research for the orchestration of reproducible and scalable experiments[END_REF], [START_REF] Buchert | A workflow-inspired, modular and robust approach to experiments in distributed systems[END_REF] that allows to specify experiments in terms of business workflows, featuring scalability and robustness of execution. The raw results, XPFLOW scripts and XPFLOW distribution are available at http://www.loria.fr/ ∼ buchert/kascade2014.tar.xz.

A. Raw performance and scalability

To measure the raw performance of the methods, the experiment with a varying number of nodes was performed. A single run consisted in distributing a single 2 GB file to all nodes. The source file was stored on a RAM-backed file system and was sent to /dev/null on receiving nodes. Therefore the impact of storage speed is not taken into account, making the network bandwidth and latency the most important factors. The network is Ethernet with 1 Gbit/s links. The nodes are ordered in an optimal way (as was described in Section III-A). The results are presented in Figure 7.

Only Kascade and MPI Broadcast were able to nearly saturate the network links. Moreover, they scale very well with an increasing number of nodes, showing a negligible loss of performance. UDPCast shows a similar behavior for small cases, but its performance degrades rapidly when number of nodes is bigger than 100. It is due to the costly synchronization between the sender and its clients. Both variations of TakTuk perform equally bad showing very low bandwidth utilization (more than one third of the theoretical value).

To sum up, Kascade achieves almost a maximum utilization of a 1 Gbit/s network and scales very well with number of participating nodes.

B. Performance in high-performance networks

The second experiment explores the performance on a cluster of 14 nodes interconnected with 10 Gbit/s Ethernet network. The transmitted file has 5 GB. The results are presented in Figure 8.

In this scenario, none of the evaluated methods is able to saturate the link. Among them, the best one is MPI Broadcast that peaked at approximately 5 Gbit/s (i.e., 50 % of the available bandwidth), but usually stays around 3 Gbit/s. It is followed by UDPCast that is able to reach more than 3 Gbit/s, but usually rests slightly above 2 Gbit/s. Kascade shows more stable behavior with transfer bandwidth slightly above 2 Gbit/s. In contrast, TakTuk-based methods show particularly low performance.

By monitoring CPU usage, it was noticed that it is saturated during transmission with MPI Broadcast and Kascade. The bottleneck is the memory that cannot provide 10 Gbit/s throughput, at least with the current implementations. The problem can be arguably mitigated by avoiding extraneous memory accesses and using multiple threads.

Kascade

TakTuk/chain TakTuk/tree MPI/IB Fig. 9. The performance of the methods using IP over InfiniBand (20 Gbit/s), however MPI/IB is using InfiniBand directly. No method is able to saturate the bandwidth, but among them only Kascade shows scalable behavior.

The second experiment measures the performance using IPover-InfiniBand interconnect (the transmitted file has 5 GB in this case). They are presented in Figure 9.

MPI Broadcast (note that it is used over InfiniBand provided by 2 switches) is very efficient for a small number of nodes, but does not scale very well and with 160 nodes shows a very low performance similar to TakTuk. The reason for that is that with 120 nodes and less, only a single switch is used, but for 160 nodes and more 2 switches must be used, causing the inter-switch link to be saturated. Kascade, although having more modest performance for small number of nodes, is fairly scalable and shows a behavior similar to the experiments with 10 Gbit/s Ethernet network.

To conclude, Kascade does not saturate available bandwidth in two exemplary high-speed networks, but shows much better scalability than other methods. Moreover, inability to saturate the network is likely to disappear with more tuned implementation.

C. Impact of topology and structure of communication on performance

As discussed in Section III-A, some methods rely on a proper ordering of nodes so that they can profit from the underlying topology. The only difference from the experiment in Section IV-A is that the order of nodes is randomized (however the nodes remain in a single L2 network). We also include a plot of Kascade results obtained with optimal ordering of nodes (Kascade/ordered) as was presented in Figure IV-A. The results are presented in Figure 10.

Unsurprisingly, the performance of Kascade deteriorates when the ordering is randomized. This phenomenon, observed also for MPI Broadcast distribution, is due to the fact that the Kascade transmission chain passes switches multiple times and saturates them. This does not pose a problem in our case, since the topology of the network is generally known in practice, especially in systems such as clusters, grids and high-speed networks that are the main target of our work.

D. Impact of hard disk I/O operations on performance

In all previous experiments the hard disk was neither read from nor written to. In practice, however, data is written to permanent storage. Due to disparate bandwidths of computer memory, network and storage, this may have a large effect on the performance of data distribution.

In this experiment, a 2 GB file is distributed just like it was described in Section IV-A. This time, however, the data is written to a hard disk instead of being discarded. The fact of data reaching the disk is not concerned -its presence in the file system cache is enough.

The nodes involved in the experiment belong to the same cluster and are equipped with identical Hitachi Deskstar 7K1000.C (HDS721032CLA362) 320 GB hard disks with 16 MB buffer size and SATA II interface. A simple test using dd revealed that the maximal writing speed is about 83.5 MB/s. The results are presented in Figure 11.

The results show a much lower performance when the file is written to a disk. Among the methods, Kascade has the best performance by being able to write around 45 MB of data per second.

E. Performance in Internet-like, heterogeneous networks

This experiment measures the performance while using a routed, heterogeneous, long-distance network. To this end, we chose 5 geographically distant sites in Grid'5000 and reserved one node on each of them. Moreover, we reserved 2 more nodes on another site so that the first point in each plot represents intra-site distribution (see Figure 12). Each consecutive point, therefore, represents an inter-site link with higher latency and routed traffic. The network between sites features 10 Gbit/s bandwidth, but the latency is substantially higher (an ICMP ping takes about 16 ms between sites, but less than 0.2 ms within one site). As the transmission time is likely to increase, we use a 1 GB file instead. MPI Broadcast, however, showed very low performance and therefore was evaluated with a file of only 100 MB in size. The methods that cannot work with routed traffic are excluded from this experiment. The results are presented in Figure 13.

All methods tend to lose performance when using highlatency and heterogeneous links. Nevertheless, Kascade offers the best overall performance. On the other hand, MPI Broadcast suffers from network and node heterogeneity to the point that it is outperformed by TakTuk.

F. Overhead when transferring small files

To measure the overhead of protocols used by each method when transferring small files, we have run a transmission of a small file (50 MB) in a scenario similar to the one presented in Section IV-A. The obtained results are presented in Figure 14.

The results show that transmission of relatively small files gives a completely different picture than the one presented in

G. Fault tolerance

To evaluate the fault-tolerance capabilities of Kascade described in Section III-D, we use the Distem [START_REF] Sarzyniec | Design and Evaluation of a Virtual Experimental Environment for Distributed Systems[END_REF] emulator. Distem is able to emulate a virtual platform over a regular cluster and is able to inject failures in the platform according to a specification provided by the user. The experimental setup is as follows:

• 20 physical nodes of a single cluster;

• each physical node has a 1 Gbit/s Ethernet network interface; • a Distem virtual platform composed of 100 virtual nodes is launched over the physical nodes (5 virtual nodes are launched on each physical node). In the experiment, the virtual nodes are called n 1 to n 100 . Let us define n [i-j] as the virtual nodes from n i to n j (inclusively), and let n [i,j,k,...] stand for the set {n i , n j , n k , . . .}.

Furthermore we define {t,n i } an event where a failure is applied on n i at the date t (e.g., t seconds after the beginning of the experiment).

The goal of the experiment is to transfer a 5 GB file from n 1 to n [2-100] with Kascade and to measure the average throughput.

Three kinds of scenarios are evaluated: 1) no failure;

2) failures are injected to kill 2 %, 5 %, and 10 % of the virtual nodes simultaneously, 10 seconds after the beginning of the transfer. In the three cases, the following failures have been injected: Figure 15 shows the performance of Kascade when failures are injected. First, in all the cases, the file was transferred correctly. One can notice than the performance without failure is quite far from the maximum network link capacity, 80 MB/s instead of 128 MB/s. This is expected since the node folding and the virtualization technique for running virtual nodes inside Distem induce an overhead. Thus, this value just acts as a reference value. Then we can see that simultaneous failures have a lower cost than sequential failures. This is normal since failure detection can be pipelined in case of simultaneous failures. Indeed, failure detection uses timeouts and every time a timeout is reached, one second is lost. As a consequence, increasing the number of sequential failures has a bigger impact on performance than simultaneous failures.

The goal of this experiment was to ensure that Kascade works despite failures and to get an idea of its the performance. Kascade has a high tuning potential and could be tuned according to the network used in order to reduce timeouts and achieve better performance even in case of sequential failures.

V. CONCLUSIONS AND FUTURE WORK

In this paper, we presented Kascade, a solution for the large scale broadcast of data. Kascade achieves high performance and scales very well -saturating a 1 Gbit/s network even at large scale. Also, a fault-tolerant mechanism enables Kascade to handle node failures during the transfer.

Kascade provides acceptable performance on high performance networks (10 Gbit/s Ethernet, 20 Gbit/s InfiniBand), but it could be further improved, as shown in our extensive performance evaluation. It also performs well in Internet-like settings. Among the evaluated solutions, it is the only one that performs adequately in all situations.

Kascade does not currently defend very well against one specific scenario: the case where the network or disk performance of one specific node is slowing down the whole process. Kascade could be further improved to detect malfunctioning nodes (by measuring their performance during the transfer) and exclude them from the transfer if their performance is lower than a specific threshold.

Kascade is a standalone tool, released under the Ce-CILL v2 Free Software license. It can be found in the addons/kascade directory of the Kadeploy3 source (http: //kadeploy3.gforge.inria.fr/).

Fig. 7 .

 7 Fig. 7.Performance and scalability using up to 200 nodes on 1 Gbit/s Ethernet. Only Kascade and MPI Broadcast fully saturate the available bandwidth and scale very well with number of nodes.

Fig. 8 .

 8 Fig. 8. The performance of the methods using 10 Gbit/s Ethernet network.No method is able to saturate the available bandwidth.

Fig. 10 .

 10 Fig. 10. The performance of the methods with random ordering of nodes. Kascade relies on a proper ordering of nodes and just like other methods shows low performance in this artificial scenario.

Fig. 11 .Fig. 12 .

 1112 Fig. 11. The performance of the methods using 1 Gbit/s Ethernet network if the clients store the file on a hard disk. Kascade shows the best performance among the methods.

Fig. 13 .

 13 Fig. 13. The performance of the methods with multiple sites and routed IP traffic. Kascade offers the best performance among them. See Figure 12 for geographical distribution of sites.

Figure 7 .Fig. 14 .

 714 Fig. 14. Time required to distribute a small file (50 MB). All methods show similar characteristics, with MPI Broadcast outperforming the rest.

  Fig.3. Topology-aware pipeline built by Kascade: node 1 (sending node) connects to node 2, which connects to node 3, etc. Node 10 connects back to node 1, which is used to forward the final report.

	1	2	3	4	5	6	7	8	9	10
	sending
	node			receiving nodes
	GET(o)	Request stream data from offset o
	PGET(o,t)	Request stream between offset o and offset t
		(in bytes)
	FORGET(o) Answer to a GET or PGET request when
		the asked part is not available anymore (case
		of recycled buffer). o indicates the minimal
		offset
	DATA(s)	Answer to a GET or PGET request, followed
		by s bytes of data
	END	Signal the end of stream
	QUIT	Signal the anticipated end of stream (case of
		user interruption for example)
	REPORT(s) After END or QUIT, a report is sent. The
		report's length is s bytes
	PASSED

  Fig. 15. Performance of Kascade in presence of failures.Three situations have been evaluated inside Distem: no failure, simultaneous failures (2%, 5%, and 10% of the nodes), sequential failures (2%, 5%, and 10% of the nodes). {10,n 29 },{20,n 69 } • 5%: {10,n 9 },{14,n 29 },{18,n 49 },{22,n 69 },{26,n 89 } • 10%: {10,n 9 },{12,n 19 },{14,n 29 },{16,n 39 },{18,n 49 },{20,n 59 }, {22,n 69 },{24,n 79 },{26,n 89 },{28,n 99 } Every experiment is repeated 50 times, and results are averaged.

		80
	Throughput (MB/s)	20 40 60
		n o f a i l u r e s i m . f a i l u r e s s i m . f a i l u r e s s i m . f a i l u r e s s e q . f a i l u r e s s e q . f a i l u r e s s e q . f a i l u r e s 0
		2 %	5 %	1 0 %	2 %	5 %	1 0 %
	• 2%: {10,n [29,69] }
	• 5%: {10,n [9,29,49,69,89] }

• 10%: {10,n

[START_REF] Jeanvoine | Kadeploy3: Efficient and Scalable Operating System Provisioning[END_REF][START_REF] Dichev | Mpi vs. bittorrent: switching between large-message broadcast algorithms in the presence of bottleneck links[END_REF][START_REF] Buchert | A workflow-inspired, modular and robust approach to experiments in distributed systems[END_REF]39,49,59

,69,79,89,99] } 3) failures are injected to kill 2 %, 5 %, and 10 % of the virtual nodes in sequence -the failures start 10 seconds after the beginning of the transfer. In the three cases, the following failures have been injected:

• 2%:

ACKNOWLEDGMENTS

Experiments presented in this paper were carried out using the Grid'5000 experimental testbed, being developed under the INRIA ALADDIN development action with support from CNRS, RENATER and several Universities as well as other funding bodies (see https://www.grid5000.fr).

DATA x DATA x DATA x DATA x . . . . . . DATA y DATA y END END DATA x DATA x DATA x DATA x . . . . . . DATA x TCP connection GET a DATA x . . .

DATA y END REPORT z PASSED Fig. 6. Message sequence chart of communication between three nodes when error occurs all nodes after send a QUIT message to their next neighbor and quit. Then n i-1 continues to receive remaining data.

Sending a GET message with an offset on every connection is important because if it is sent only after the first connection and if there is a node crash between n i and n j-1 that crashes before having performed its first connection, that leads to a deadlock. Indeed, n i-1 waits for a GET message whereas n j waits for a first DATA message.

After such failure handling, the protocol continues such as described in previous section without nodes n i to n j-1 . The Figure 6 shows this mechanism with one node failure.

IV. EXPERIMENTAL VALIDATION

In the following section we present an extensive evaluation of our approach to data distribution and its comparison with other existing methods. In particular, we seek to answer the following questions:

• How do the various solutions perform and scale up to large number of nodes? (IV-A) • How does Kascade perform on high-performance networks (10 Gbit/s Ethernet, IP over InfiniBand)? (IV-B) • What is the impact of network topology and communication structure on performance? (IV-C) • What it the impact of I/O performance on the overall performance? (IV-D) • How does Kascade perform on large-scale (Internet-like) setups? (IV-E) • How does Kascade perform on smaller files? (IV-F) • How well does Kascade's fault tolerance mechanism perform? (IV-G) To answer these questions, we evaluated different methods for data distribution in computer networks. They are described in the following paragraphs.