Rolando Martins
email: rolandomartins@cmu.edu

Rajeev Gandhi
email: rgandhi@ece.cmu.edu

Priya Narasimhan

Soila Pertet
email: spertet@ece.cmu.edu

António Casimiro

Diego Kreutz
email: kreutz@lasige.di.fc.ul.pt

Paulo Veríssimo

Experiences with Fault-Injection in a Byzantine Fault-Tolerant Protocol

Keywords: Byzantine fault-injector, failure diagnosis, cloud-computing, Byzantine fault-tolerance, intrusion-tolerance

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

Introduction

Recent improvements in the performance of Byzantine Fault-Tolerant (BFT) protocols have made such protocols feasible for building fault-and intrusiontolerant systems. Presently, there are multiple implementations of BFT protocols at disposal of system developers to make their own system fault/intrusiontolerant without worrying about having to implement the functionality by themselves. However, as other research projects [START_REF] Clement | Making Byzantine Fault Tolerant Systems Tolerate Byzantine faults[END_REF] have observed, while current stateof-the-art BFT protocol implementations have considerably improved the performance of the fault-free path, they often fail to properly handle all of the corner cases. The end result is that while many BFT implementations can efficiently handle the complexity of Byzantine failures, they often suffer from multiple orders of magnitude reductions in throughput and long periods of unavailability when in the presence of non-independent faults, such as colluding malicious nodes. This often poses a dilemma for system programmers -on the one hand the use of a publicly available implementation of BFT protocol allows programmers to develop fault-and intrusion-tolerant systems without worrying about implementing these complicated aspects themselves while, on the other hand, there is a question about the ability of an implementation to actually handle complex as well as simple failures in an efficient manner.

System developers often need answers to multiple questions about a particular BFT protocol before they are able to select it and be confident that the implementation will actually meet all of their requirements. Questions can vary from performance to robustness and trustworthiness, such as the following examples. What kind of faults does the system tolerates? Does it really tolerates arbitrary faults? Or only more common faults, e.g. crash faults? What is the degradation of system throughput in the presence of faults? Are there thresholds for fault-arrival rates, beyond which the system breaks down? How does a BFT protocol compare with others?

Software fault injection is often used in software testing to quantitatively assess the impact of faults/bugs in the software. We use a similar approach to assess the performance of a BFT protocol and answer questions like the aforementioned ones that developers may have about the behavior of a protocol in the presence of faults. In this paper, we describe Hermes, our fault injection framework created to help BFT protocol developers in the strenuous task of testing the behavior of a BFT implementation under a diverse and broad range variety of faults. To show its usability, we used Hermes to assess the behavior of BFT-SMaRt [START_REF]BFT-SMaRt: High-Performance Byzantine Fault-tolerant State Machine Replication[END_REF], a well-known BFT protocol implementation.

Hermes allows system developers to get insight into the performance of a BFT protocol implementation by allowing them to inject faults and observe the behavior of the system. Hermes's fault injection architecture is flexible and allows protocol independent faults (like crash faults, network faults) as well as protocol dependent faults (like corrupt headers, forged signatures) to be injected into a BFT protocol. Our approach is clearly distinguished from existing ones by the fact that we provide a way to simultaneously inject faults across multiple nodes, allowing different type of faults to be injected in different nodes. Furthermore, by simply selecting the appropriate set of faults, the user can enforce a specific fault model, e.g., if collusion is outside the fault model, then no collusion faults can be used.

We built Hermes using AspectJ [START_REF] Kiczales | Aspect-Oriented Programming[END_REF], for the JAVA runtime, and AspectC++ [START_REF] Spinczyk | AspectC++: an Aspect-Oriented Extension to the C++ Programming Language[END_REF], for the C++ runtime, which allows it to seamlessly weave the fault-injecting infrastructure into the target protocol. The use of Aspect-Oriented Programming (AOP) was to avoid source-code modifications on the target system, especially in context-free faults, i.e., faults that do not modify or access any internal state of the protocol. We decided not to use dynamic weaving because it would in-troduce further complexity into the infrastructure. As such, the injection points are statically weaved and compiled into the target source code.

Hermes does not require a system developer to be familiar with the BFT protocol or its implementation for injecting protocol independent faults. On the other hand, protocol dependent faults, such as payload size corruption, require that the developer performs some modifications to the source code. A minimal amount of adaptation is unavoidable because it depends on the specific details of protocol in use. In our experience with BFT-SMaRt, it took us about two hours to inject protocol dependent faults.

Related Work

There has been considerable work done in developing fault injection systems [START_REF] Chandra | Loki: A State-Driven Fault Injector for Distributed Systems[END_REF][START_REF][END_REF][START_REF] Han | Doctor: An integrateD sOftware fault in-jeCTiOn enviRonment[END_REF][START_REF] Alvarez | Centralized Failure Injection for Distributed, Fault-Tolerant Protocol Testing[END_REF] and analyzing the dependability of fault-tolerant systems [START_REF] Dawson | Testing of Fault-Tolerant and Real-Time Distributed Systems via Protocol Fault Injection[END_REF][START_REF] Looker | Assessing the Dependability of OGSA Middleware by Fault Injection[END_REF][START_REF] Marsden | Failure Analysis of an ORB in Presence of Faults[END_REF]. Loki [START_REF] Chandra | Loki: A State-Driven Fault Injector for Distributed Systems[END_REF] is a fault injector for distributed systems that injects faults based on a partial view of the global system state. Loki allows the user to specify a state machine and a fault injection campaign in which faults are triggered by state changes. DBench Project [START_REF][END_REF] aimed to develop standards for creating dependability benchmarks for computer systems. This joint cooperation characterized benchmarks as representing an agreement that is widely accepted both by the computer industry and/or by the user community. Orchestra [START_REF] Dawson | Testing of Fault-Tolerant and Real-Time Distributed Systems via Protocol Fault Injection[END_REF] is a fault injector that uses an interception approach similar to ours to inject communication faults into any layer in the protocol stack. The fault injection core provides the ability to filter, manipulate, and inject new messages. Ferrari [START_REF] Kanawati | FERRARI: A Flexible Software-Based Fault and Error Injection System[END_REF] (Fault and Error Automatic Real-Time Injection) uses software traps to inject CPU, memory, and bus faults. The Fault Tolerance and Performance Evaluator (Ftape) [START_REF] Tsai | Measuring Fault Tolerance with the FTAPE Fault Injection Tool[END_REF] allows developers to inject faults into user mode registers in CPUs, memory locations, and the disk subsystem. Doctor [START_REF] Han | Doctor: An integrateD sOftware fault in-jeCTiOn enviRonment[END_REF] (Integrated Software Fault Injection Environment) allows developers to inject CPU faults, memory faults, and network communication faults in a system. Xception [START_REF] Carreira | Xception: Software Fault Injection and Monitoring in Processor Functional Units[END_REF] takes advantage of the advanced debugging and performance monitoring features present in many modern processors to inject more realistic faults. Ballista [START_REF] Devale | The Ballista Software Robustness Testing Service[END_REF] is a "black box" software testing tool that uses combinational tests of valid and invalid parameter values for subroutine calls, methods and functions. A good survey of fault injection techniques and tools for testing software dependability is provided in [START_REF] Hsueh | Fault Injection Techniques and Tools[END_REF].

Some of the recent research has looked at the inefficiencies of BFT protocol implementations to handle Byzantine as well as benign faults. In [START_REF] Clement | Making Byzantine Fault Tolerant Systems Tolerate Byzantine faults[END_REF], the authors provide a comparative in-depth analysis of several protocols, namely [START_REF] Castro | Practical Byzantine Fault Tolerance and Proactive Recovery[END_REF][START_REF] Abd-El-Malek | Fault-scalable Byzantine Fault-Tolerant Services[END_REF][START_REF] Kotla | Zyzzyva: Speculative byzantine fault folerance[END_REF][START_REF] Cowling | HQ Replication: A Hybrid Quorum Protocol for Byzantine Fault Tolerance[END_REF], in their pursuit to build Aardvark. Their assessment is based mainly in the use of flooding and packet delay (in both primary and non-primary nodes). Similarly, in Prime [START_REF] Amir | Prime: Byzantine Replication under Attack. Dependable and Secure Computing[END_REF], the authors provide an evaluation of PBFT [START_REF] Castro | Practical Byzantine Fault Tolerance and Proactive Recovery[END_REF], a leader-based Byzantine fault-tolerant replication protocol, but mention that their approach should work well with all BFT protocols that are leader based, such as [START_REF] Kotla | Zyzzyva: Speculative byzantine fault folerance[END_REF][START_REF] Amir | Steward: Scaling Byzantine Fault-Tolerant Replication to Wide Area Networks[END_REF][START_REF] Yin | Separating Agreement From Execution for Byzantine Fault Tolerant Services[END_REF][START_REF] Martin | Fast byzantine consensus[END_REF][START_REF] Amir | Customizable Fault Tolerance forWide-Area Replication[END_REF][START_REF] Li | Beyond One-Third Faulty Replicas in Byzantine Fault Tolerant Systems[END_REF]. The experiments conducted in Prime were based on two attacks. The first involved delaying "pre-prepare" messages, while the second consisted in timeout manipulation, where the system would become stalled until large timeouts occurred.

Fig. 1: Overview of the Hermes's architecture.

Hermes's Overview

Our fault-injection platform, shown in Figure 1, is governed by a fault-injection orchestrator that enables the injection of multiple faults, simultaneous or not, across multiple remote nodes. The actual fault injection is performed by Hermes's runtime, which is incorporated into the BFT replica (through code-weaving). The initial deployment, i.e., the initial view, of both replicas and client is accomplished through the use of the deployment service. However, a BFT protcol can still reconfigure, e.g., adding a new replica, using its usual facilities without having to interact with the deployment service, because the runtime is able to transparently connect to Hermes's orchestrator.

We start by characterizing the types of faults we consider in this work, followed by the description of the orchestrator, runtime and deployment service.

Faults

We consider two types of faults, context-free faults and context-dependent faults. The first relates to faults that can be injected without any context information. For instance, for injecting CPU load it is not necessary to access any information on the protocol. The second, the context-dependent faults need to access data of the protocol being injected in the system. For example, corrupting a specific packet type for a given consensus instance, we need to access the data contained in the protocol header.

Context-Free Faults

• CPU Load -injects a specified amount of CPU load.

• Crash -crashes the runtime and associated BFT replica.

• Sleep -delays processing for a specified amount of time.

• Drop packet -induces a packet drop following a pre-defined policy (onetime or percentage-wise).

Context-Dependent Faults

• Corrupt header -corrupts a packet header with the main goal of breaking low-level protocol buffering, i.e., underflow and overflow. • Corrupt payload -corrupts the payload of a packet with erroneous and random information. • Forge signatures -substitutes part of the signature set with forged signatures, in an effort to convince correct nodes of an erroneous value. • DDOS -causes the malicious nodes to start multicast messages to all correct nodes.

Orchestrator

The orchestrator is the main component of Hermes and its goal is to provide the orchestration between the various runtimes, that are built-in into the replicas and client, and act as a front-end to the developer. In Section 4, we provide an overview of the implementation and an example of its usage. The interactions between the orchestrator and runtime are built on top of three communication primitives: RemoteAction, Action and Notification.

Action

An Action is used by the runtime to verify if the fault that is about to be injected into the replica is enabled and ready, or alternatively, if the fault is disabled. While this operation could be used to retrieve information from the overall execution of the injection protocol, its main purpose is to serve as synchronization barrier (shown in Figure 2) for the injection of simultaneous faults. For example, a fault is injected in the packet send procedure but it could only be run (injected) when all the malicious nodes reach the same fault. This allows us to test colluding among malicious nodes and also target specific test cases in order to explore specific and tricky/uncommon faults.

For achieving this, each active runtime calls an Action when it is about to inject the fault. Because it is a synchronous operation, the runtime waits for the reply from the orchestrator. This reply is only sent by the orchestrator when all the malicious nodes reach the synchronization point, that is, when all the runtimes have called the same Action. After receiving the reply, each runtime proceeds and injects the fault.

RemoteAction

The RemoteAction, shown in Figure 3a, is used by the orchestrator to perform remote procedure calls in the replicas's runtime, and it can be used to manipulate the state of the runtime, or to retrieve some portion of state from the replica.

Notification

The notification mechanism, shown in Figure 3b, provides an asynchronous message passing interface from a runtime to the orchestrator. Its main purpose is to avoid the overhead associated with synchronous operations, i.e. an Action, and is used to inform the orchestrator of the progress of the algorithm.

Runtime

The runtime's main responsibility is to inject faults accordingly to the indications of the orchestrator. The actual fault injection is achieved through the use of Aspect-Oriented Programming (AOP) [START_REF] Kiczales | Aspect-oriented programming[END_REF]. We use aspects as a way to seamlessly introduce fault injection points, as well as all the necessary networking infrastructure needed to interact with the orchestrator. We provide more details in Section 4.

Deployment Service

The deployment service was built to allow remote bootstrap and closure of applications. While it is a general purpose deployment service, its main purpose is to launch replicas and clients to construct the initial view of the system. It should be noted that this does not represent an obstacle for any possible built-in reconfiguration mechanisms, e.g., adding or removing replicas, within the BFT protocol. All the necessary infrastructure to use Hermes is encapsulated inside the runtime, that in turn, is weaved into the replica's (and client) code. Thus, independently of how a replica is deployed, the runtime will perform all the necessary logistics.

Implementation

We have implemented the orchestrator in JAVA, while the runtime has implementations in both JAVA and C++. To avoid the complexity/overhead introduced by JAVA existing serialization mechanisms, we custom built a binary protocol that uses little endian encoding without padding. This also allowed us to easily extend Hermes runtime to C++, and in the future will allow to extend to other programming languages, such as python. The implementations are freely available as open-source projects (under an Apache license version 2) at:

https://github.com/rolandomar/hermes (JAVA runtime and orchestration) https://github.com/rolandomar/hermesCPP (C++ runtime)

Bootstrap Process

It is assumed that each of the hosting nodes has the deployment service running.

The orchestrator upon start-up uses the deployment service to create and bootstrap the target protocol nodes, which in this paper are BFT-SMaRt [START_REF]BFT-SMaRt: High-Performance Byzantine Fault-tolerant State Machine Replication[END_REF] replicas and clients. These nodes were previously weaved, with our runtime and faults, through the use of aspects (shown in Listings 1.3 and 1.4).

After its creation by the deployment service, the runtime within each protocol node bootstraps and connects back to the orchestrator. In turn, the orchestrator creates a barrier for synchronizing the start of all the nodes.

Orchestrator

An overview of the API provided by the orchestrator is shown in the appendix http://www.contrib.andrew.cmu.edu/ ~martinsr/middleware13/apA.pdf. The faultInjection() and simultaneousFaultInjection() are the two most important operations offered by the orchestrator. They allow for a single and simultaneous fault injection, respectively. In order to provide further control over the activation of the faults, on the remote runtimes, we use the Action primitive. When a runtime reaches a fault, it then uses the Action primitive to check if it should proceed with the injection. For example, we use it to only allow the faults to become active after the 500th invocation has taken place (shown in Listing 1.1). For now this value is fixed but can be easily ported as a parameter. Listing 1.2 shows the code associated with Attack 1 presented in our evaluation (Section 5). The attack simulates a simultaneous crash of a set of malicious nodes. It starts with the creation of a CrashFault associated with the specific injection point, given by faultID. In line 5, the orchestrator generates a simultaneous fault, in this case a crash fault. At this point, the orchestrator sends the fault information to the malicious nodes. The test run starts with the creation and bootstrap of a client (line 6-8), identified by "1001", that with perform 1000 invocations, with each invocation incrementing the service counter by 1.

Runtime Code-Weaving

One of our goals was to make our approach as little intrusive as possible. For that purpose, and as previously explained we used AOP, through the use of As-pectJ [START_REF] Kiczales | Aspect-Oriented Programming[END_REF], for the JAVA runtime, and AspectC++ [START_REF] Spinczyk | AspectC++: an Aspect-Oriented Extension to the C++ Programming Language[END_REF] for the C++ runtime. We use two distinct aspects, shown in Listings 1.3 and 1.4.

Runtime Startup Aspect

To seamlessly bootstrap Hermes's runtime into the replica's code, we use aspect HermesStartupAspect depicted in Listing 1.3. The aspect is executed before the actual execution of the main() procedure. It starts by retrieving the runtime identifier from the list of the application arguments (line 5-6). This identifier is then used to instantiate the singleton's instance (line 7-10). The bootstrap of the runtime is followed by the call to the actual application's main() (line 11).

1 @Aspect 2 public class HermesStartupAspect { 3 @Around("execution ( * bftsmart.demo.counter.

CounterServer.main * (..))")

ServersCommunication Aspect

We needed to access the underlying communication infrastructure to inject lowlevel faults, such as payload corruption. For that purpose, we created the aspect shown in Listing 1.4. For the sake of simplicity and space we only shown the code associated with the payload corruption attack.

The initial portion of the aspect, lines 7-13, checks if the fault is active, and if so retrieves the information about the Paxos protocol, namely, the execution identifier and packet type, e.g., weak and strong packet types. If this information could not be retrieved, i.e., the message is not related to core Paxos protocol, but belongs to surrounding infrastructure such as state transfer, we bypass the fault injection and execute the target code (line 11). In line 14, we update the execution identifier within the fault's context. This information is later sent to the orchestrator in the fault validation, that is executed in line [START_REF] Kotla | Zyzzyva: Speculative byzantine fault folerance[END_REF].

Because this is a context-dependent fault, the execution of the fault only triggers the verification of the fault's validity. The actual injection is performed in the sendBytesFailureInjected() procedure. This procedure is a duplicate of the original code with the added fault injection mechanisms. It was not possible to weave code around this procedure, because it was necessary to access the underlying infrastructure, i.e., in the packet formation we needed to corrupt payload but leave the header intact.

Evaluation

Hardware Setup

We used a NUMA (Non-Uniform Memory Access) workstation with dual hexacores, for a total of 12 physical cores and 24 logical threads, with 32GB of RAM and 512GB of RAID-0 storage, comprising 2 SSDs with 256GB each.

For simulating a distributed environment we created 11 virtual machines (VMs), using QEMU/KVM, 10 of which were dedicated to run BFT replicas and 1 for the client. Each VM was allocated with 2GB of RAM and 2 virtual CPUs. The orchestrator ran on the host operating system. Both host the guests used Ubuntu 12.10 LTS as their operating system. Because the virtualized environment already introduces delay and jitter in the network stack (around 1ms latency while measuring with ICMP pings), we only constrained the total amount of bandwidth available (both in-bounding and out-bounding) in each VM to 100Mbit/s, and thus effectively creating a 100Mbit/s network. For the purpose, we used the TC command to manipulate the underlying network stacks, with the following commands: tc qdisc add dev eth0 handle ffff: ingress tc filter add dev eth0 parent ffff: protocol ip prio 50 / u32 match ip src 0.0.0.0/0 police rate 100mbit / burst 100mbit drop flowid :1

Experiments Experimental Setup

In order to assess the resiliency of BFT-SMaRt, we performed 1000 invocations per run and injected the faults midpoint, i.e., in the 500th invocation. For such purpose, we devised the following attacks:

• Attack 1 -Simultaneous crash: simultaneously crash all malicious nodes.

• Attack 2 -Payload forged with MAX INT : all malicious nodes forge their payload size, setting it to MAX INT (2,147,483,647) bytes. • Attack 3 -Delay Prepare messages below detection timeout: all malicious nodes delay propose messages to 90% of the timeout used, e.g., for a timeout of 3s the resulting delay would be of 2.7s. • Attack 4 -Delay Prepare messages above detection timeout: all malicious nodes delay propose messages by 5 times the value of the timeout, e.g., with a timeout of 3s then the delay would be 15s.

Because of space constrains, we only show the results from attacks 1 to 4. The remaining (5 to 9) are available for consultation in the appendix at http://www.contrib.andrew.cmu.edu/ ~martinsr/middleware13/apA.pdf.

For each attack, we tested it against 12 configurations, shown in Table 1, with 1f, 2f and 3f standing for 1, 2 and 3 faults injected, respectively. N is the total number of replicas needed to enforce the 3f +1 requirement, while M is the set of identifiers for the malicious nodes used in a particular configuration. For example, M = {0} stands for the set of malicious nodes only containing node "0", whereas M = {x, y} represents a set with two randomly chosen identifiers. For each of these configurations we ran the attack 16 times, and computed the average and the 95% confidence intervals. Each run (a single test) is comprised of 1000 invocations. The maximum amount of time allowed for each run was 5 minutes. After this time, we considered that the run had failed, even if it was not completely stalled or aborted, but rather progressing very slowly.

Configurations 1f, |N | = 4 2f, |N | = 7 3f, |N | = 10 # M # M # M 0 {0} 2 {0} 6 {0} 1 {x} 3 {x} 7 {x} 4 {0, 1} 8
In our evaluation we collected the following data: Failed Runs (FR) -the number of failed runs, including stalled or aborted runs with a running time higher then 300s. Fault-free Latency (LA) -the invocation latency before the fault was injected (in milliseconds). Faulty Latency (LB) -the invocation latency after the fault was injected (in milliseconds). Total Duration (D) -the total duration of the run. If the run timed out then the total duration is 300s (in seconds). Recovery Time (R) -the invocation latency for the 500th and 501st invocations (in seconds).

Total Faulty Invocations (FI) -the number of successful invocations performed after the fault was injected. We only considered the faulty invocations from runs that produced at least 5 invocations after the 500th invocation (otherwise we considered them stalled without recovery).

The application that we chose to run was the bftsmart.demo.counter, a simple counter built on top of the BFT-SMaRt protocol. On each invocation we increment the counter by 1. After each successful invocation, the client sends a notification to the orchestrator to report the invocation number and latency. When the orchestrator receives the 1000th invocation from the client, it ends the test and calculates the duration of that run.

The experiments were run twice. We first performed the experiments using the default values and without any modification to the source code, whereas in the second time we tuned the timeout values and made modifications to the source code, of which we provide a detailed discussion on Section 5.3. The relevant parameters used in this work are shown in Table 2. Table 2: Parameters considered in the evaluation. For certain cases the SHORT TIMEOUT is used to quickly trigger a reconfiguration, such as a voluntary exit from a group. The TOTAL ORDER TIMEOUT is the main timeout used in the implementation and is used to detect when a request was not processed and subsequently trigger the leader-change protocol. The CONNECTION TIMEOUT controls the timeout associated with the establishment of a new connection to a replica. The last parameter, INVOCATION TIMEOUT, is used by the client, more specifically through the ServiceProxy, to control the timeout associated with each invocation.

Parameters

Results

In this section we will start to discuss the performance using the default parameters and without any modifications to the source code (left sub-table on the results tables 3 to 6). Later, using the knowledge gained throughout the first round of our evaluation, we show how it helped us to track the underlying issues and partially overcome them by tuning some of the system parameters (Table 2) and by applying a modification to the source code (right sub-table ).

Generically, we can see throughout the results that the increase in the number of nodes in the system, from 4 to 10 nodes (1f to 3f ) results in a linear increase on the fault-free invocation latency, from 16ms to 20ms (see column LA in tables 3 to 6). In some cases, the faulty invocation latency drops after a fault has occurred. Because the number of nodes decreases, the latency (and overhead) associated with the protocol also decreases, except in attacks 3 and 4. The recovery time is higher when the leaders are attacked, going upwards to 30s in attack 3. We provide a discussion on the reasons beyond this high recovery time in Section 5.3.

Attack 1

Our goal with attack 1 was to evaluate the impact of simultaneously crashing multiple nodes in the system (for 2f and 3f configurations). The results from the single fault scenario (1f configurations) are presented to provide a baseline comparison. The results from our evaluation of attack 1 are shown in Table 3. In our initial evaluation, using the default values and implementation, we encountered some issues with configurations 4, 8 and 10. These issues seemed related with the change-leader protocol.

After a manual inspection of logs, we found that the problem was a composition of several issues. First, we found that 40s invocation timeout (from within the client's ServiceProxy) was too short, and it should be at least 60s. The stalls that we checked in the results were a direct result from this. When this timeout is triggered, the client aborts its execution and the run ends. This also hid the true values of the recovery time, that can reach roughly 60s (which we concluded after some additional experimentation with larger timeouts).

Overall, we found that the default timeout values (associated with the replicas) were too high for a LAN (Ethernet based network). But that did not account for the low performance that we detected in the protocol after the injections of the faults. After an inspection to the source code, we found a subtle bug in the timeout management that leads to problems in the change-leader protocol. We provide a better explanation to this problem later in Section 5.3.

Attack 1 (Crash Fault)

Default Tuned

C FR LB LA D R FI FR LB LA D R FI (%) (ms) (ms) (s) (s) (#) (%) (ms) (ms) (s) (s) (#) 0 0 16±2 9±0 37±0 20±0 499±0 0 16±2 9±0 23±0 6±0 499±0 1 0 16±2 9±0 19±3 2±3 499±0 0 16±2 8±0 18±1 1±1 499±0 2 0 18±1 10±0 38±0 20±0 499±0 0 18±1 10±0 24±0 6±0 499±0 3 0 17±1 10±0 22±3 3±3 499±0 0 17±1 10±0 18±0 0±0 499±0 4 100 18±1 N/A 300±0 N/A N/A 0 18±1 11±0 30±0 11±0 499±0 5 6 17±1 11±0 38±33 2±3 499±0 0 17±1 11±0 21±1 2±1 499±0 6 0 20±0 12±0 41±0 20±0 499±0 0 20±0 12±0 27±0 6±0 499±0 7 0 20±0 12±0 21±2 1±2 499±0 0 20±0 12±0 20±0 0±0 499±0 8 100 20±0 N/A 300±0 N/A N/A 0 20±0 12±0 32±0 12±0 499±0 9 0 20±0 12±0 26±4 6±4 499±0 0 20±0 12±0 21±1 0±0 499±0 10 100 20±0 N/A 300±0 N/A N/A 0 20±0 14±1 36±0 14±0 499±0 11 13 20±0 13±0 55±45 0±0 499±0 0 20±0 13±0 22±1 1±1 499±0
Table 3: Attack 1 with the default (left) and the tuned (right) configurations.

Attack 2

We designed attack 2 for assessing the impact of overflowing in the protocol. This was achieved by modifying the length field on the PaxosMessage packet. The results from our evaluation are show in Table 4.

Attack 2 (Value/Corruption Fault)

Default Tuned

C FR LB LA D R FI FR LB LA D R FI (%) (ms) (ms) (s) (s) (#) (%) (ms) (ms) (s) (s) (#) 0 0 16±2 8±0 36±0 20±0 499±0 0 16±2 8±0 23±0 6±0 499±0 1 0 16±2 8±0 21±4 5±4 499±0 0 16±2 8±0 17±0 0±0 499±0 2 0 18±1 10±0 38±0 20±0 499±0 0 17±1 10±0 24±0 6±0 499±0 3 0 17±1 10±0 23±4 5±4 499±0 0 17±1 10±0 18±0 0±0 499±0 4 100 18±1 N/A 300±0 N/A N/A 0 18±1 11±0 27±0 9±0 499±0 5 13 17±1 11±0 63±43 11±5 499±0 0 17±1 11±0 20±1 1±1 499±0 6 0 20±0 13±0 40±0 20±0 499±0 0 20±0 13±0 26±0 6±0 499±0 7 0 20±0 13±0 22±2 1±2 499±0 0 20±0 13±0 21±0 0±0 499±0 8 100 20±0 N/A 300±0 N/A N/A 0 20±0 13±0 30±0 9±0 499±0 9 6 20±0 14±0 42±32 4±4 499±0 6 20±0 13±0 39±32 1±1 499±0 10 100 20±0 N/A 300±0 N/A N/A 0 20±0 15±0 34±0 12±0 499±0 11 0 20±0 15±0 31±4 10±4 499±0 0 20±0 15±0 24±1 2±1 499±0
Table 4: Attack 2 with the default (left) and the tuned (right) configurations.

The presence of failures indicates that the protocol is susceptible to the overflow attack (also to underflow and payload corruption, c.f. in appendix under attacks 6 and 7). Furthermore, the failure pattern from attack 2 closely resembles the pattern of attack 1, which indicates that the underlying causes should be the same or similar.

Using this insight, we were able to successfully track down the root of this behavior within the source code. We found that the crash (caused by the overflow or underflow) of the deliver thread in the low-level communication infrastructure is omitted from the overall leader module management. The same applies when we corrupt the payload of messages. The deliver thread detects the mismatch between the payload and the signature but silently ignores it. The protocol is able to recover because the timeout associated with the request is triggered forcing the leader-change sub-protocol to change the leader. It is important to note that BFT-SMaRt is implemented using the principle of decoupling the total ordering of requests from the actual consensus primitive [START_REF] Sousa | From Byzantine Consensus to BFT State Machine Replication: A Latency-Optimal Transformation[END_REF]. The client sends its requests to every replica, not only the leader. This is done to prevent a malicious leader to stall the protocol. When a replica (non-leader) detects that the leader did not propose the request, it forwards the request to the current leader and activates a new timeout. If this fails, then a new regency is activated through the leader-change sub-protocol. Later in Section 5.3, we discuss the impact of this decision on the overall protocol performance.

Possibly it would be feasible to use information from the low-level communication layer to provide further knowledge to the leader module in order to speedup the recovery process when malicious nodes are in the role of the leader. We discuss a possible implementation of this approach later in Section 5.3.

Attack 3 and 4

The use of a leader in BFT protocols creates a potential attack point. This subject was previously explored by Prime [START_REF] Amir | Prime: Byzantine Replication under Attack. Dependable and Secure Computing[END_REF], which shows the impact of the presence of a malicious leader. We devised attacks 3 and 4 for assessing the resilience of BFT-SMaRt to this kind of attack.

In certain cases, a reconfiguration can be triggered with a short timeout, that is about one third of the regular timeout (3s in the default configuration and 2s in our tuned configuration). To avoid detection, we delay conservatively the sending of prepare messages by 90% of the value of this timeout (2.7s for the default configuration and 1.8s for the tuned configuration). The results from attack 3 (Table 5) show that delay of the prepare messages by leader causes a increase of invocation latency to around 2.7s, as expected, causing the runs to fail as they take more than 300s to finish.

Alternatively, in attack 4 (Table 6) we used 5 times the value of the short timeout, resulting in a timeout of 15s, for the default configuration, and 10s for the tuned version. This clearly triggers the change-leader sub-protocol but eventually the protocol itself stalls when about 15 faulty invocations have been processed. Again, the failure pattern also shows some correlation with the failure pattern of attack 1.

Attack 3 (Timing Fault, Delay Below Timeout)

Default Tuned Although we are still in the process of analyzing all the data and logs that we have collected, we found some interesting issues within BFT-SMaRt implementation. Although the BFT-SMaRt is a leader-based BFT protocol we were surprised to verify the impact of simple crash faults on the system. While the implementation was able to sustain flawlessly single crash faults or even multiple random crash faults, it was vulnerable, performance wise, to the simultaneous injection of crash faults within the first elements of the nodes, i.e., the initial leaders. 1) After analyzing the code, we detected a misconfiguration of the client, which normally has a default 40s request timeout. When this timeout is triggered within the client, it aborts the execution of the run. This was the reason behind the stalls we detected while using the default timeout values and original implementation. 2) Because BFT-SMaRt is leader-based, faults in the leader have a high impact on the overall performance (and recovery) of the protocol. After increasing this timeout we were able to measure a recovery time around 60s. The high recovery time that we measured is only partially explained by the high timeout values. An analysis of the source code revealed that the protocol, in the presence of a request timeout, first tries to forward pending requests to the current leader. This was done to accommodate the possibility that the current leader did not propose the request because it might have been dropped by the underlying network infrastructure. However, in the presence of a single malicious leader, this in fact doubles the recovery time. This is because the request is first forwarded to the leader, and the change-leader sub-protocol only runs after this operation has timed out. On top of this, the leader-change protocol goes sequentially through the processes list to find the next leader (from the lowest to the highest identifier). Because we intentionally injected simultaneous faults in the nodes with the lowest identifiers (that are the first nodes to be elected as leaders), this resulted in the protocol having to go through all the malicious nodes. Furthermore, they immediately assume the roles of leaders, introducing further delays. To minimize this situation, we lowered the timeout values (shown in Table 2).

C FR LB LA D R FI FR LB LA D R FI (%) (ms) (ms) (s) (s) (#) (%) (ms) (ms) (s) (s) (#
3) However, this alone did not fully explain the extensive recovery times. A closer inspection of the source code reveled a subtle implementation artifact. The timer used for failure detection (RequestsTimer) for all the pending requests present in the system gets its timeout value doubled each time the changeleader protocol is triggered (for example, when receiving a STOP and STOP DATA messages from other nodes), but it is never reduced, even in the presence of more favorable system conditions, such as the absence of failures or timeouts. Given this, we introduced a modification to the original source code that consisted on only doubling the timeout within the same regency (i.e., the same leader) on the current view, otherwise the value is reset to the default value. Using our tuned parameters and correction to the source code (right sub-table on the results tables) we were able to almost avoid any failed runs, except in attack 3. Sporadically we got a failed run while using attack 2. We are in the process of analyzing the logs to determine the underlying issues associated with those failed runs. 4) Attack 3 was designed in light to previous work on Prime [START_REF] Amir | Prime: Byzantine Replication under Attack. Dependable and Secure Computing[END_REF], and was designed to degrade overall performance of the system by delaying the sending of propose messages. The attack was designed so that the leader was not suspected by the other nodes, by limiting the amount of delay just below to the detection timeout value. By lowering the timeout value we were able to minimize the impact of this type of attack, although it is evident that a more proactive and structural approach has to be taken to solve this issue when using leader-based BFT protocols. While Prime provides a way to minimize this situation, it still does not provide a complete solution, because of the limitations derived from the use of Diffserv [START_REF]An Architecture for Differentiated Services[END_REF], because a flooding attack would compromise the measurements used to adapt the timeouts. We tried to perform a full evaluation of Prime but we discover after the initial tests that the protocol was not completely implemented. Therefore, we were unable to verify their results. To corroborate our findings, we can see in attack 4 that if a sufficiently small timeout is chosen then the attack is contained, with no apparent loss in performance (except for the recovery time associated with the election of the new leaders). 5) For Attack 2 we were able to manually verify the code and found a missing verification in the creation of the packet by the receiving thread (ReceiverThread in the ServerConnection class). Because it does not verify the size of the payload, it allows the attacker to crash the thread either with overflow or underflow attacks. While JAVA provides a managed memory system, some vulnerabilities have been found in the past that explore such cases. 6) Taking the knowledge gathered throughout our evaluation we implemented a second modification to the original protocol. As stated in the discussion of Attack 2, we make use of the low-level network events such as unexpected network errors (e.g., closing of TCP connections) or packet malformation (such as mismatched signatures) to trigger the change-leader sub-protocol. The results from this second modification are shown in the appendix, under section "Second Round of Corrections". We were able to cut roughly in half the recovery time from our first implementation modification (and timeout tuning). From the original implementation and timeout values we were able to provide up to a 10 fold improvement. Nevertheless, it should be noted that these modifications come at a price. Because we are assuming a more synchronous network layer and assuming that any type of fault we get from the network is malicious, we could be inducing false suspicion on nodes leading to unnecessary leader changes. However, this would only affect performance but not correctness.

BFT Limitations and Future Directions

It seems clear that leader-based BFT protocols have their weakest point in the leader and change-leader sub-protocol. The timeout settings also play an important role in the overall performance.

For situations without the presence of malicious node, the introduction of adaptive timeouts like in Adaptare-FD [START_REF] Dixit | Adaptare: Supporting Automatic and Dependable Adaptation in Dynamic Environments[END_REF], could improve seamlessly the overall performance. However to efficiently deal with the presence of malicious nodes, current approaches [START_REF] Amir | Prime: Byzantine Replication under Attack. Dependable and Secure Computing[END_REF] are still not able to deliver degradation-free performance. While making the assumption that the network is totally asynchronous makes a strong case from a standpoint of correctness and safety, we feel that in order to bridge theory and practice, stronger, yet realistic, assumptions must be made about underlying network infrastructure. The use of software-defined networking, for instance by OpenFlow [START_REF] Mckeown | OpenFlow: Enabling Innovation in Campus Networks[END_REF], could allows us to improve on the current state-of-the-art, such as avoiding the issues related with flooding attacks [START_REF] Clement | Making Byzantine Fault Tolerant Systems Tolerate Byzantine faults[END_REF].

Conclusions and Future Work

In this paper we presented a novel fault-injecting framework that enables the assessment of BFT implementations. Furthermore, we demonstrate the importance of providing support for non-independent faults. Using our approach we were able to detect 2 implementation bugs. The first, at the low communication level, allowed overflow and underflow attacks caused by a missing size verification on packet reception, whereas the second bug was related to a high level implementation artifact derived from the ever-increasing timeout values within the leader-change sub-protocol, that in certain cases would effectively stall the protocol for more than 60s in the presence of non-independent faults. Lastly, we proposed a second set of modifications to the source code, where we avoid forwarding messages to a possibly malicious leader prior to a change in regency and enhance it by using low-level networking exceptions/events, such as signature mismatch, to trigger a leader change when in the suspicion of the presence of a malicious leader.

Using our tuned parameters and source code modifications we were able to provide up to a 10 fold improvement over the original implementation and default parameters.

Future Work

We expect to enhance Hermes by providing the necessary infrastructure to support proof forging, by coordinating and distributing all the necessary keys across the malicious nodes. Furthermore, we want to expand the work accomplished in this paper, by introducing a visualization tool that continuously monitors and traces all the nodes present in the protocol with the goal of providing the initial support for debugging.

Fig. 2 :

 2 Fig. 2: Action operation overview.

Fig. 3 :

 3 Fig. 3: RemoteAction (left) and Notification (right) operations.

1 3 case 5 String 6 / 8 } 9 /

 35689 public ActionResult onAction( OrchestrationNodeServerClient client, Action action) 2 switch (action.getSerialID()) { CheckFaultInjectionAction.SERIAL ID: { 4 CheckFaultInjectionAction cfa = (CheckFaultInjectionAction) action; faultID = cfa.getFaultID(); * omitted code * / 7 return new CheckFaultInjectionActionResult( (cfa.getFaultContext().getRun() < 500)); * omitted code * / 10 } Listing 1.1: Orchestration code for fine control over fault injection.

1 6 String 10 if(!ret){ 11 / * handle error * / 12 } 13 }

 610111213 public void attack1(String[] malicious) { 2 String faultID = "2B4FA20ED54E4DA9B6B2A917D1FA723F"; 3 HermesFault fault = new CrashFault(faultID); 4 try { HermesFuture<Boolean> future = simultaneousFaultInjection(malicious,fault); command = HermesConfig. getClientCommandLaunch(); 7 String[] args = new String[]command, "1001", "1", "1000"; 8 launchHermesClient(command, args, 5000); 9 boolean ret = future.get( TIMEOUT, TimeUnit.MILLISECONDS); catch (Exception ex) { 14 / * handle error * / 15 } 16 } Listing 1.2: Orchestrator-side code for Attack 1.

4 public 6 String 10 }

 4610 void advice(ProceedingJoinPoint jp) throws Throwable 5 String[] args = (String[]) jp.getArgs()[0]; catch (Exception e) { / * handle error * /} 11 jp.proceed(); 12 } Listing 1.3: Runtime bootstrap aspect for replicas and client.

  ) 0 100 16±2 2723±0 300±0 5±0 104±0 100 16±2 1822±0 300±0 3±0 157±0 1 13 16±2 86±10 51±46 0±0 449±63 19 16±2 130±10 69±54 0±0 434±65 2 100 18±1 2729±0 300±0 5±0 104±0 100 18±1 1828±0 300±0 3±0 156±0 3 13 17±1 88±10 53±45 0±0 449±64 6 17±1 46±5 35±33 0±0 477±40 4 100 18±1 2730±0 300±0 5±0 104±0 100 18±1 1828±0 300±0 3±0 156±0 5 25 17±1 186±16 88±59 1±1 400±83 38 17±1 296±16 123±66 1±0 370±81 6 100 20±0 2737±0 300±0 5±0 103±0 100 20±0 1835±0 300±0 3±0 154±0 7 6 20±0 49±7 38±33 0±0 474±46 13 20±0 89±8 55±45 0±0 456±55 8 100 20±0 2736±0 300±0 5±0 103±0 100 20±0 1835±0 300±0 3±0 154±0 9 19 20±0 136±13 73±53 1±1 424±75 19 20±0 134±10 72±53 0±0 434±65 10 100 20±0 2737±0 300±0 5±0 103±0 100 20±0 1836±0 300±0 3±0 154±0 11 38 20±0 312±22 126±66 2±1 350±93 50 20±0 444±21 160±68 1±0 327±84

Table 1 :

 1 Configurations used in the attacks evaluation.

				{0, 1}
	5	{x, y}	9	{x, y}
			10 {0, 1, 2}
			11 {x, y, z}

Table 5 :

 5 Attack 3 with the default configuration (left) and the tuned version

	5.3 Lessons Learned	(right)

  Attack 4 (Timing Fault, Delay Above Timeout)

			Default				Tuned
	C	FR LB (%) (ms)	LA (ms)	D (s)	R (s)	FI (#)	FR LB (%) (ms)	LA (ms)	D (s)	R (s)	FI (#)
	0 56 16±2 915±117 252±35 30±0 228±116 0 16±2 8±0 22±0 6±0 499±0
	1 13 16±2 102±28 66±46 9±6 438±77 0 16±2 8±0 17±1 0±0 499±0
	2 19 17±1 353±55 187±38 30±0 408±91 0 17±1 10±0 24±0 6±0 499±0
	3 0 17±1 53±18	45±32 5±5 499±0	0 17±1 10±0 19±1 1±1 499±0
	4 100 18±1 16058±574 300±0 30±0 15±1	0 17±1 11±0 30±0 12±0 499±0
	5 0 17±1 89±24	64±42 7±6 499±0	0 17±1 10±0 20±1 2±1 499±0
	6 38 20±0 357±66 202±39 30±0 336±116 0 20±0 12±0 26±0 6±0 499±0
	7 6 20±0 34±15	38±33 1±3 468±58 0 20±0 12±0 21±1 1±1 499±0
	8 100 20±0 16196±640 300±0 30±0 13±2	0 20±0 13±0 32±0 12±0 499±0
	9 19 20±0 101±31 73±53 5±5 408±92 0 20±0 13±0 23±1 2±1 499±0
	10 100 20±0 15968±521 300±0 30±0 14±2	0 20±0 27±5 40±0 12±0 499±0
	11 25 20±0 154±38 91±59 7±6 379±101 0 20±0 16±0 25±3 3±2 499±0

Table 6 :

 6 Attack 4 with the default configuration (left) and the tuned version (right)

Acknowledgments

We thank Alysson Bessani and the conference reviewers for their feedback. This research was sponsored in part by the project CMUPT/RNQ/0015/2009 (TRONE -Trustworthy and Resilient Operations in a Network Environment), and by Intel via the Intel Science and Technology Center for Cloud Computing (ISTC-CC).


Viswanath Nagarajan

Joel Wolf
email: jlwolf@us.ibm.com

Andrey Balmin
email: andrey@graphsql.com

Kirsten Hildrum
email: hildrum@us.ibm.com

FlowFlex: Malleable Scheduling for Flows of MapReduce Jobs

We introduce FlowFlex, a highly generic and effective scheduler for flows of MapReduce jobs connected by precedence constraints. Such a flow can result, for example, from a single user-level Pig, Hive or Jaql query. Each flow is associated with an arbitrary function describing the cost incurred in completing the flow at a particular time. The overall objective is to minimize either the total cost (minisum) or the maximum cost (minimax) of the flows. Our contributions are both theoretical and practical. Theoretically, we advance the state of the art in malleable parallel scheduling with precedence constraints. We employ resource augmentation analysis to provide bicriteria approximation algorithms for both minisum and minimax objective functions. As corollaries, we obtain approximation algorithms for total weighted completion time (and thus average completion time and average stretch), and for maximum weighted completion time (and thus makespan and maximum stretch). Practically, the average case performance of the FlowFlex scheduler is excellent, significantly better than other approaches. Specifically, we demonstrate via extensive experiments the overall performance of FlowFlex relative to optimal and also relative to other, standard MapReduce scheduling schemes. All told, FlowFlex dramatically extends the capabilities of the earlier Flex scheduler for singleton MapReduce jobs while simultaneously providing a solid theoretical foundation for both.

Introduction

MapReduce [START_REF] Dean | Mapreduce: Simplified Data Processing on Large Clusters[END_REF] is a fundamentally important programming paradigm for processing big data. Accordingly, there has already been considerable work on the design of high quality MapReduce schedulers [START_REF] Zaharia | Job Scheduling for Multi-User MapReduce Clusters[END_REF][START_REF] Zaharia | Delay Scheduling: A Simple Technique for Achieving Locality and Fairness in Cluster Scheduling[END_REF][START_REF] Wolf | FLEX: A Slot Allocation Scheduling Optimizer for MapReduce Workloads[END_REF][START_REF] Agrawal | Scheduling Shared Scans of Large Data Files[END_REF][START_REF] Wolf | On the Optimization of Schedules for MapReduce Workloads in the Presence of Shared Scans[END_REF]. All of the schedulers to date have quite naturally focused on the scheduling of collections of singleton MapReduce jobs. Indeed, single MapReduce jobs were the appropriate atomic unit of work early on. Lately, however, we have witnessed the emergence of more elaborate MapReduce work, and today it is common to see the submission of flows of interconnected MapReduce jobs. Such a MapReduce flow can result, for example, from a single user-level Pig [START_REF] Gates | Building a High-Level Dataflow System on Top of MapReduce: The Pig Experience[END_REF], Hive [START_REF] Thusoo | Hive -a Petabyte Scale Data Warehouse using Hadoop[END_REF] or Jaql [START_REF] Beyer | Jaql: A Scripting Language for Large Scale Semistructured Data Analysis[END_REF] query. Each flow can be represented by a directed acyclic graph (DAG) in which the nodes are singleton Map or Reduce phases and the directed arcs represent precedence. Significantly, flows have become the basic unit of MapReduce work, and it is the completion times of these flows that determines the appropriate measure of goodness, not the completion times of the individual MapReduce jobs.

This paper introduces FlowFlex, a scheduling algorithm for flows of MapReduce jobs. FlowFlex can attempt to optimize an arbitrary metric based on the completion times of the flows. Common examples include makespan, average completion time, average and maximum stretch 3 and metrics involving one or more deadlines. Any given metric will be appropriate for a particular scenario, and the precise algorithmic variant FlowFlex applies will depend on that metric. For example, in a batch environment one might care about makespan, to ensure that the batch window is not elongated. In an interactive environment users would typically care about average or maximum completion time, or about average or maximum stretch. There are also a variety of metrics associated with hard or soft deadlines. To the best of our knowledge scheduling schemes for flows of MapReduce jobs have never been considered previously in the literature.

Our contributions are both theoretical and practical. We advance the theory of malleable parallel scheduling with precedence constraints. Specifically, we employ resource augmentation analysis to provide bicriteria approximation algorithms for both minisum and minimax objective functions. As corollaries, we obtain approximation algorithms for total weighted completion time (and thus average completion time and average stretch), and for maximum weighted completion time (and thus makespan and maximum stretch). We also produce a highly generic and practical MapReduce scheduler for flows of jobs, called FlowFlex, and demonstrate its excellent average case performance experimentally.

The closest previous scheduling work for MapReduce jobs appeared in [START_REF] Wolf | FLEX: A Slot Allocation Scheduling Optimizer for MapReduce Workloads[END_REF]. The Flex scheduler presented there is now incorporated in IBM BigInsights [START_REF][END_REF]. (See also the FlexSight visualization tool [START_REF] De Pauw | Visualizing Jobs with Shared Resources in Distributed Environments[END_REF].) Flex schedules to optimize a variety of metrics as well, but differs from the current work in that it only considers singleton MapReduce jobs, not flows. Architecturally, Flex sits on top of the Fair MapReduce scheduler [START_REF] Zaharia | Job Scheduling for Multi-User MapReduce Clusters[END_REF][START_REF] Zaharia | Delay Scheduling: A Simple Technique for Achieving Locality and Fairness in Cluster Scheduling[END_REF], essentially overriding its decisions while simultaneously making use of its infrastructure. The FlowFlex scheduling problem is clearly a major generalization of the Flex problem. With modest caveats, it can also be said that the FlowFlex algorithms significantly generalize those of Flex. They are also much more theoretically grounded. See also the special purpose schedulers [START_REF] Agrawal | Scheduling Shared Scans of Large Data Files[END_REF] and CircumFlex [START_REF] Wolf | On the Optimization of Schedules for MapReduce Workloads in the Presence of Shared Scans[END_REF], built to amortize shared Map phase scans.

There are fundamental differences between Fair and the three schedulers in the Flex family: Fair makes its decisions based on the current moment in time, and thus considers only the resource (slot) dimension. Fair is indeed fair in the sense of instantaneous progress, but does not directly consider completion time. On the other hand, Flex, CircumFlex and FlowFlex think in two dimensions, both resource and time. And they optimize towards completion time metrics. It is our contention that completion time rather than instantaneous progress determines the true quality of a schedule.

The rest of this paper is organized as follows. Section 2 gives preliminaries and describes the good fit between the theory of malleable scheduling and the MapReduce enviornment. Section 3 introduces the scheduling model and lists our formal theoretical results. The FlowFlex scheduling algorithms are described in Section 4, and the proofs of the performance guarantees are outlined there. Space limitations prevent us from detailing all of these proofs, but interested readers can find them in [START_REF] Balmin | Malleable Scheduling for Flows of MapReduce Jobs[END_REF]. In Section 5 we compare FlowFlex experimentally with Fair and FIFO, both naturally extended in order to handle precedence constraints. We also explain the practical considerations associated with implementing FlowFlex as an epoch-based scheduler. Conclusions appear in Section 6.

Preliminaries

The theory of malleable scheduling fits the reality of the MapReduce environment well. To understand this we give a brief, somewhat historically oriented overview of theoretical parallel scheduling and its relation to MapReduce.

The first parallel scheduling implementations and theoretical results involved what are today called rigid jobs. These jobs run on a fixed number of processors and are presumed to complete their work simultaneously. One can thus think of a job as corresponding to a rectangle whose height corresponds to the number of processors p, whose width corresponds to the execution time t of the job, and whose area s = p • t corresponds to the work performed by the job. Early papers, such as [START_REF] Coffman | Performance Bounds for Level-Oriented Two-Dimensional Packing Algorithms[END_REF], focused on the makespan metric, providing some of the very first approximation algorithms. (These are polynomial time schemes with guaranteed performance bounds.)

Subsequent parallel scheduling research took a variety of directions, again more or less mirroring real scenarios of the time. One such direction involved what has now become known as moldable scheduling: Each job can be run on an arbitrary number of processors, but with an execution time which is a monotone non-increasing function of the number of processors. Thus the height of a job is turned from an input parameter to a decision variable. The first approximation algorithm for moldable scheduling with a makespan metric appeared in [START_REF] Turek | Approximate Algorithms for Scheduling Parallelizable Tasks[END_REF]. Later, [START_REF] Schwiegelshohn | Smart SMART Bounds for Weighted Response Time Scheduling[END_REF] found the first approximation algorithm for both rigid and moldable scheduling problems with a (weighted) average completion time metric.

The notion of malleable scheduling is more general than moldable. Here the number of processors allocated to a job is allowed to vary over time. However, each job must still perform its fixed amount of work. One can consider the most general problem variant in which the rate at which work is done is a function of the number of allocated processors, so that the total work completed at any time is the integral of these rates through that time. However, this problem is enormously difficult, and so the literature to date [START_REF] Drozdowski | Scheduling for Parallel Processing[END_REF][START_REF] Leung | Handbook of Scheduling[END_REF] has focused on the special case where the speedup function is linear through a given maximum number of processors, and constant thereafter. Clearly, malleable schedules can only improve objective function values relative to moldable schedules. On the other hand, malleable scheduling problems are even harder to solve well than moldable scheduling problems. We will concentrate on malleable scheduling problems with linear speedup up to some maximum, with flow precedence constraints and any of several different metrics on the completion times of the flows. See [START_REF] Drozdowski | Scheduling for Parallel Processing[END_REF][START_REF] Leung | Handbook of Scheduling[END_REF] for more details on both moldable and malleable scheduling. The literature on the latter is quite limited, and this paper is a contribution.

Why does MapReduce fit the theory of malleable scheduling with linear speedup and processor maxima so neatly? There are multiple reasons.

1. MapReduce and malleable scheduling are about allocation: There is a natural decoupling of MapReduce scheduling into an Allocation Layer followed by an Assignment Layer. In the Allocation Layer, quantity decisions are made, and that is where any mathematical complexity resides. The Assignment Layer then implements these allocation decisions (to the extent possible, given locality issues [START_REF] Zaharia | Delay Scheduling: A Simple Technique for Achieving Locality and Fairness in Cluster Scheduling[END_REF] and such) in the MapReduce cluster. Fair, Flex, CircumFlex and our new FlowFlex scheduler reside in the Allocation Layer.

The malleable (as well as rigid and moldable) scheduling literature is also about allocation rather than assignment.4 2. MapReduce work exhibit roughly linear speedup, and maximum constraints occur naturally: Both the Map and Reduce phases are composed of many small, independent tasks. Because they are independent they do not need to start simultaneously and can be processed with any degree of parallelism without significant overhead. This, in turn, means that the jobs will have nearly linear speedup: Remember that linear speedup is a statement about the rate at which work is completed. Maximum constraints in either the Map or Reduce phase occur because they happen to be small (and thus have few tasks), or when only a few tasks remain to be allocated. 3. MapReduce fits the malleable model well: Assuming the tasks are many and small, the decisions of the scheduler can be approximated closely. To understand this, consider Figure 1, which depicts the Assignment Layer implementing the decisions of the Allocation Layer. The Allocation Layer output is a hypothetical malleable schedule for three jobs. The Assignment Layer works locally at each node in the cluster. Suppose a task on that node completes, freeing a slot. The Assignment Layer simply determines which job is the most relatively underallocated according to the Allocation Layer schedule. And then, within certain practical constraints, it acts greedily, assigning a new task from that job to the slot. Examining the figure, the tasks are represented as "bricks" in the Assignment Layer. The point is that the large number and small size of the tasks makes the right-hand side a close approximation to the left-hand side. That is, Assignment Layer reality will be an excellent approximation to Allocation Layer theory.

The model is not perfect, of course. For example, if the number of tasks in a MapReduce job is modest, the idealized scenario depicted in Figure 1 will be less than perfect. Furthermore, there is a somewhat delicately defined notion 5 of precedence between Map and Reduce tasks that is not cleanly modeled here. Recall that we model a single MapReduce job as a Map phase followed by a Reduce phase. In reality, some Reduce tasks can begin before all the Map tasks complete. But recall that Flex suffers from precisely the same issues, and has been used successfully in IBM BigInsights [START_REF][END_REF] for several years. Other Models for MapReduce. We briefly mention some other MapReduce models that have been considered in the literature. Moseley et al. [START_REF] Moseley | On Scheduling in Map-Reduce and Flow-Shops[END_REF] consider a "two-stage flexible flow shop" [START_REF] Schuurman | A Polynomial Time Approximation Scheme for the Two-Stage Multiprocessor Flow Shop Problem[END_REF] model, and give approximation and online algorithms for total completion time of independent MapReduce jobs. Berlinska and Drozdowski [START_REF] Berlinska | Scheduling Divisible MapReduce Computations[END_REF] use "divisible load theory" to model a single MapReduce job and its communication details. Theoretical frameworks for MapReduce computation have been proposed in [START_REF] Karloff | A Model of Computation for MapReduce[END_REF][START_REF] Koutris | Parallel evaluation of conjunctive queries[END_REF].

Compared to our setting, these models are at a finer level of granularity, that of individual Map and Reduce tasks. Our model, as described above, decouples the quantity decisions (allocation) from the actual assignment details in the cluster. We focus on obtaining algorithms for the allocation layer, which is abstracted as a precedence constrained malleable scheduling problem.

Formal Model and Results

As discussed, we model the MapReduce application as a parallel scheduling problem. There are P identical processors that correspond to resources (slots) in the MapReduce cluster. Each flow j is described by means of a directed acyclic graph. The nodes in each of these DAGs are jobs, and the directed arcs correspond to precedence relations. We use the standard notation i 1 ≺ i 2 to indicate that job i 1 must be completed before job i 2 can begin. Each job i must perform a fixed amount of work s i (also referred to as the job size), and can be performed on a maximum number δ i ∈ [P ] of processors at any point in time. 6 We consider jobs with linear speedup through their maximum numbers of processors: the rate at which work is done on job i at any time is proportional to the number of processors p ∈ [δ i ] assigned to it. Job i is complete when s i units of work have been performed.

We are interested in malleable schedules. In this setting, a schedule for job i is given by a function

τ i : [0, ∞) → {0, 1, . . . , δ i } where ∞ t=0 τ i (t) dt = s i .
Note that this satisfies both linear speedup and processor maxima. We denote the start time of schedule τ i by S(τ i ) := arg min{t ≥ 0 : τ i (t) > 0}; similarly the completion time is denoted C(τ i ) := arg max{t ≥ 0 : τ i (t) > 0}. A schedule for flow j (consisting of jobs I j ) is given by a set {τ i : i ∈ I j } of schedules for its jobs, where C (τ i1 ) ≤ S (τ i2 ) for all i 1 ≺ i 2 . The completion time of flow j is max i∈Ij C (τ i ), the maximum completion time of its jobs. Our algorithms make use of the following two natural and standard lower bounds on the minimum possible completion time of a single flow j. (See, for example, [START_REF] Drozdowski | Scheduling for Parallel Processing[END_REF].)

• Total load (or squashed area):

1 P i∈Ij s i . • Critical path: maximum of r=1 si r δi r over all chains 7 i 1 ≺ • • • ≺ i in flow j.
Each flow j also specifies an arbitrary non-decreasing cost function w j : R + → R + where w j (t) is the cost incurred when job j is completed at time t. We consider both minisum and minimax objective functions. The minisum (resp. minimax) objective minimizes the sum (resp. maximum) of the cost functions over all flows. In the notation of [START_REF] Drozdowski | Scheduling for Parallel Processing[END_REF][START_REF] Leung | Handbook of Scheduling[END_REF] this scheduling environment is P |var, p i (k) = pi (1) k , δ i , prec| * . 8 We refer to these problems collectively as precedence constrained malleable scheduling with linear speedup. Our highly general cost model can solve all the commonly used scheduling objectives: weighted average completion time, makespan (maximum completion time), average and maximum stretch, and deadline-based metrics associated with number of tardy jobs, service level agreements (SLAs) and so on. The objective functions we consider are either minisum or minimax: Minisum scheduling problems involve the minimization of the (possibly weighted) sum of individual flow metrics, or, equivalently, their (weighted) average. On the other hand, minimax scheduling problems involve the minimization of the maximum of individual metrics, an indication of worst case flow performance. Definition 1. A polynomial time algorithm is said to be an α-approximation if it produces a schedule that has objective value at most α ≥ 1 times optimal.

We would like to provide approximation algorithms for the above malleable scheduling problems. But as shown in [START_REF] Drozdowski | Scheduling Parallel Tasks With Sequential Heads and Tails[END_REF], even under very special precedence constraints (chains of length three) the general minisum and minimax problems admit no finite approximation ratio unless P=NP. Hence we use resource augmentation [START_REF] Kalyanasundaram | Speed is as Powerful as Clairvoyance[END_REF] and focus on bicriteria approximation guarantees. Definition 2. A polynomial time algorithm is said to be an (α, β)-bicriteria approximation if it produces a schedule using β ≥ 1 speed processors that has objective value at most α ≥ 1 times optimal (under unit speed processors).

Our main result is that we can find approximation algorithms in some cases and bicriteria approximation algorithms in all others.

Theorem 1. The precedence constrained malleable scheduling problem with linear speedup admits the following guarantees.

• (2, 3)-bicriteria approximation algorithm for general minisum objectives.

• (1, 2)-bicriteria approximation algorithm for general minimax objectives.

• 6-approximation algorithm for total weighted completion time (including total stretch).

• 2-approximation algorithm for maximum weighted completion time (including makespan and maximum stretch).

The first two results on general minisum and minimax objectives imply the other two as corollaries. The main idea in our algorithms (for both minisum and minimax) is a reduction to strict deadline metrics, for which a simple greedy scheme is shown to achieve a good bicriteria approximation. The reduction from minisum objectives to deadline metrics is based on a minimum cost flow relaxation, and "rounding" the optimal flow solution. The reduction from minimax objectives to deadlines is much simpler and uses a bracket and bisection search.

The FlowFlex Scheduling Algorithm

Our scheduling algorithm has three sequential stages. See Figure 3 for an algorithmic overview. In a little more detail, the stages may be described as follows.

1. First we consider each flow j separately, and convert its (general) precedence constraint into a chain (total order) precedence constraint. We create a pseudo-schedule for each flow that assumes an infinite number of processors, but respects precedence constraints and the bounds δ i on jobs i. Then we partition the pseudo-schedule into a chain of pseudo-jobs, where each pseudo-job k corresponds to an interval in the pseudo-schedule with uniform processor usage. Just like the original jobs, each pseudo-job k specifies a size s k and bound δ k of the maximum number of processors it can be run on. We note that (unlike jobs) the bound δ k of a pseudo-job may be larger than P . An important property here is that the squashed-area and critical-path lower bounds of each chain equal those of its original flow. 2. We now treat each flow as a chain of pseudo-jobs, and obtain a malleable schedule consisting of pseudo-jobs. This stage has two components: a. We first obtain a bicriteria approximation algorithm in the special case of metrics based on strict deadlines, employing a natural greedy scheme. b. We then obtain a bicriteria approximation algorithm for general cost metrics, by reduction to deadline metrics. For minisum cost functions we formulate a minimum cost flow subproblem based on the cost metric, which can be solved efficiently. The solution to this subproblem is then used to derive a deadline for each flow, which we can use in the greedy scheme. For minimax cost metrics we do not need to solve an minimun cost flow problem. We rely instead on a bracket and bisection scheme, each stage of which produces natural deadlines for each chain. We thus solve the greedy scheme multiple times. These performance guarantees are relative to the squashed-area and criticalpath lower bounds of the chains, which, by Stage 1, equal those of the respective original flows. We now have a malleable schedule for the pseudo-jobs satisfying the chain precedence within each flow as well as the bounds δ k . 3. The final stage combines Stages 1 and 2. We transform the malleable schedule of pseudo-jobs into a malleable schedule for the original jobs, while respecting the precedence constraints and bounds δ i . We refer to this as shape shifting. Specifically, we convert the malleable schedule of each pseudo-job k into a malleable schedule for the (portions) of jobs i that comprise it. The full set of these transformations, over all pseudo-jobs k and flows j, produces the ultimate schedule.

1: for j = 1, . . . , m do 2:

Run Stage 1 scheme on flow j, yielding pseudo-schedule for chain of pseudo-jobs. 3: Stage 2 scheme begins. 4: if minsum objective then 5:

Run algorithm in Figure 6. 6: else 7:

Run minimax algorithm in Figure 8. 8: Stage 2 scheme ends. 9: Run Stage 3 shape shifting algorithm using Stages 1 and 2 output.

Stage 2: Scheduling Flows with Chain Precedence Constraints

In this section, we consider the malleable scheduling problem on P parallel processors with chain precedence constraints and general cost functions. Each chain

j ∈ [m] is a sequence k j 1 ≺ k j 2 ≺ • • • k j n(j)
of pseudo-jobs, where each pseudo-job k has a size s(k) and specifies a maximum number δ(k) of processors that it can be run on. We note that the δ(k)s may be larger than P . Each chain j ∈ [m] also specifies a non-decreasing cost function w j : R + → R + where w j (t) is the cost incurred when chain j is completed at time t. The objective is to find a malleable schedule on P identical parallel processors that satisfies precedence constraints and minimizes the total cost.

Malleable schedules for pseudo-jobs (resp. chains of pseudo-jobs) are defined identically to jobs (resp. flows). To reduce notation, we denote a malleable schedule for chain j by a sequence τ j = τ j 1 , . . . , τ j n(j) of schedules for its pseudo-jobs, where τ j r is a malleable schedule for pseudo-job k j r for each r ∈ [n(j)]. Note that chain precedence implies that for each r ∈ {1, . . . , n(j) -1}, the start time of k j r+1 , S(τ j r+1 ) ≥ C(τ j r ), the completion time of k j r . The completion time of this chain is C(τ j ) := C(τ j n(j) ). Unfortunately, as shown in [START_REF] Drozdowski | Scheduling Parallel Tasks With Sequential Heads and Tails[END_REF], this problem does not admit any finite approximation ratio unless P=NP. Given this hardness of approximation, we focus on bicriteria approximation guarantees. We first give a (1, 2)-approximation algorithm when the cost functions are based on strict deadlines. Then we obtain a (2, 3)-approximation algorithm for arbitrary minisum metrics and a (1, 2)approximation algorithm for arbitrary minimax metrics. Importantly, all these performance guarantees are based on the squashed-area and critical-path lower bounds of the chains. Since the Stage 1 transformation (flows to chains) maintains these same lower bounds, the guarantees in Stage 2 are relative to the lower bounds of the original flows. So the objective value incurred in Stage 2 is a good approximation to the optimum of the scheduling instance under the original flows.

Scheduling with Strict Deadlines

We consider the problem of scheduling chains on P parallel processors under a strict deadline metric. That is, each chain j ∈ [m] has a deadline d j and its cost function is:

w j (t) = 0 if t ≤ d j and ∞ otherwise.
We show that a natural greedy algorithm is a good bicriteria approximation.

Theorem 2. There is a (1, 2)-bicriteria approximation algorithm for malleable scheduling with chain precedence constraints and a strict deadline metric.

Proof Sketch. By renumbering chains, we assume that

d 1 ≤ • • • ≤ d m .
The algorithm schedules chains in increasing order of deadlines, and within each chain it schedules pseudo-jobs greedily by allocating the maximum possible number of processors. A formal description appears as Figure 5. The utilization function σ : R + → {0, 1, . . . , P } denotes the number of processors being used by the schedule at each point in time.

1: Initialize utilization function σ : [0, ∞) → {0, 1, . . . , P } to zero. 2: for j = 1, . . . , m do 3: for i = 1, . . . , n(j) do 4:

Set S(τ j i ) ← 0 if i = 1 and S(τ j i ) ← C(τ j i-1 ) otherwise.

5:

Initialize τ j i : [0, ∞) → {0, . . . , P } to zero. 6:

For each time t ≥ S(τ j i ) in increasing order, set

τ j i (t) ← min P -σ(t) , δ(k j i ) , (1)
until the area t≥S(τ j i ) τ j i (t) dt = s(k j i ) the size of pseudo-job k j i .

7:

Set C(τ j i ) ← max{z :

τ j i (z) > 0}. 8:
Update utilization function σ ← σ -τ j i . 9:

Set C(τ j ) ← C(τ j n(j) ). 10:

if C(τ j ) > 2 • dj then 11:

Instance is infeasible.

Fig. 5: Algorithm for Scheduling Chains with Deadline Metric

Notice that this algorithm produces a valid malleable schedule that respects the chain precedence constraints and the maximum processor bounds. To prove the performance guarantee, we show that if there is any solution that meets all deadlines {d } m =1 then the algorithm's schedule satisfies C(τ j ) ≤ 2 • d j for all chains j ∈ [m]. The main idea is to divide the time C(τ j ) taken to complete any chain j into two types of events according to Equation (1), namely times where all P processors are fully utilized (i.e. τ j i (t) = Pσ(t)) and times where a pseudo-job is fully run (i.e. τ j i (t) = δ(k j i )). The first event is bounded by the total area in the earliest j chains and the second by the critical path of chain j, each of which is at most d j . So chain j's completion time C(τ j ) ≤ 2 • d j .

Minisum Scheduling We now consider the problem of scheduling chains on P parallel processors under arbitrary minisum metrics. Recall that there are m chains, each having a non-decreasing cost function w j : R + → R + , where w j (t) is the cost of completing chain j at time t. The goal in the minisum problem is to compute a schedule of minimum total cost. Let opt denote the optimal value of the given minisum scheduling instance. Theorem 3. There is a (2, 3+o(1))-bicriteria approximation algorithm for malleable scheduling with chain precedence constraints under minisum cost metrics.

For each chain j ∈ [m], define

Q j := max    n(j) i=1 s(k j i ) δ(k j i ) , 1 P n(j) i=1 s(k j i )    , (2)
the maximum of the critical path and area lower bounds. Note that the completion time of each chain j (even if it is scheduled in isolation) is at least Q j . So the optimal value opt ≥ m j=1 w j (Q j ). We may assume, without loss of generality, that every schedule for these chains completes by time H := 2m • max j Q j . In order to focus on the main ideas, we assume here that (i) each cost function w j (•) has integer valued breakpoints (i.e. times where the cost changes) and (ii) provide an algorithm with runtime polynomial in H. In the full version, we show that both these assumptions can be removed. Before presenting the algorithm, we recall: Definition 3 (Minimum cost flow). The input is a network given by a directed graph (V, E) with designated source/sink nodes and demand ρ, where each arc e ∈ E has a capacity α e and cost (per unit of flow) of β e . A flow satisfies arc capacities and node conservation (in-flow equals out-flow), and the goal is to find a flow of ρ units from source to sink having minimum cost.

Our algorithm works in two phases. In the first phase, we treat each chain simply as a certain volume of work, and formulate a minimum cost flow subproblem using the cost functions w j s. The solution to this subproblem is used to determine candidate deadlines {d j } m j=1 for the chains. Then in the second phase, we run our algorithm for deadline metrics using {d j } m j=1 to obtain the final solution. The algorithm is described in Figure 6, followed by a high-level proof sketch (the details can be found in the full version).

1: Set volume Vj ← n(j) i=1 s(k j i ) for each chain j ∈ [m]. 2: Define network N on nodes {a1, . . . , am} ∪ {b1, . . . , bH } ∪ {r, r }, where r is the source and r the sink. 3: Define arcs E = E1 ∪ E2 ∪ E3 ∪ E4 of N as follows (see also Figure 7). arc (bt,r ) has cost 0, capacity P , and

E1 := {(r, aj) : j ∈ [m]}, arc (r, aj) has cost 0, capacity Vj, E2 := {(aj, bt) : j ∈ [m], t ∈ [H], t ≥ Qj}, arc (aj, bt) has cost w j (t) V j , capacity ∞, E3 := {(bt, r ) : t ∈ [H]},
E4 = {(bt+1, bt) : t ∈ [H -1]}, arc (bt+1, bt) has cost 0, capacity ∞.
t b H cost w i (t)/V j cap V j cap P
When unspecified,

a j Arcs E 2 Arcs E 1 Arcs E 3
The dashed arcs are E 4 . cost = 0, cap = ∞. Proof Sketch of Theorem 3. In the first phase of our algorithm (Steps 1-4) we treat each chain j ∈ [m] as work of volume V j , which is the total size of pseudojobs in j. The key property of this phase is that the network flow instance on N is a relaxation of the original scheduling instance, i.e. the minimum cost flow f is at most opt. This property relies on the construction of N , where the nodes a j s correspond to chains and b t s correspond to intervals [t -1, t) in time. The arcs E 1 (together with the demand ρ = m j=1 V j ) enforce that V j amount of flow is sent to each a j , i.e. V j work is done on each chain j. The arcs E 2 ensure that flow from a j can only go to nodes b t with t ≥ Q j , i.e. chain j can complete only after Q j . See [START_REF] Balmin | Malleable Scheduling for Flows of MapReduce Jobs[END_REF]. These arcs also model the minisum cost objective. Finally, arcs E 3 and E 4 correspond to using at most P processors at any time.

In the second phase of the algorithm (Steps 5-6) we use the min-cost flow solution f to obtain a feasible malleable schedule for the m chains. The candidate deadlines {d j } m j=1 correspond to times when the chains are "half completed" in the solution f . Since the costs w j (•) are non-decreasing, the cost m j=1 w j (d j ) of completing chains by their deadlines is at most 2 • cost(f ) ≤ 2 • opt. Then, using the definition of network N , we show that for each chain j, its critical path is at most d j and the squashed area of earlier-deadline chains is at most 2 • d j . These two bounds combined with the analysis of the deadline metric algorithm (Theorem 2) imply that each chain j ∈ [m] is completed by time 3 • d j .

In practice we could round the flow based on multiple values, not just the single halfway point described above. This would yield, in turn, multiple deadlines, and the best result could then be employed.

We note that in some cases the bicriteria guarantees can be combined.

Corollary 1. There is a 6-approximation algorithm for minimizing weighted completion time in malleable scheduling with chain precedence constraints, including average stretch.

Proof. This follows directly by observing that if a 3-speed schedule is executed at unit speed then each completion time scales up by a factor of three.

Minimax Scheduling Here we consider the problem of scheduling chains on P parallel processors under minimax metrics. Recall that there are m chains, each having a non-decreasing cost function w j : R + → R + . The goal is to compute a schedule that minimizes the maximum cost of the m chains.

Theorem 4. There is a (1, 2+o(1))-bicriteria approximation algorithm for malleable scheduling with chain precedence constraints under minimax cost metrics.

Recall the definition of Q j s (maximum of critical path and area lower bounds) from [START_REF] Balmin | Malleable Scheduling for Flows of MapReduce Jobs[END_REF]; and H = 2m • max m j=1 Q j an upper bound on the length of any schedule. The algorithm given below is based on a bracket and bisection search that is a common approach to many minimax optimization problems, for example [START_REF] Hochbaum | A Unified Approach to Approximation Algorithms for Bottleneck Problems[END_REF]. It also relies on the algorithm for deadline metrics. See Figure 8.

1: Set lastsuccess ← max m
j=1 wj(H) and lastf ail ← 0. 2: while lastsuccess -lastf ail > 1 do 3:

Set L ← (lastsuccess + lastf ail)/2.

4:

for j = 1, . . . , m do 5:

Compute deadline Dj := arg max{t : wj(t) ≤ L} for each chain j ∈ [m].

6:

Solve this deadline-metric instance using Algorithm 5.

7:

if schedule is feasible with 2-speed then 8:

Set lastsuccess ← L. 9: else 10:

Set lastf ail ← L. 11: Output the schedule corresponding to lastsuccess.

Fig. 8: Algorithm for Scheduling Chains with Minimax Objective

Proof. Let opt denote the optimal minimax value of the given instance. Clearly, 0 ≤ opt ≤ max j w j (H). (We assume that the cost functions w j s are integer valued: this can always be ensured at the loss of a 1 + o(1) factor.)

Observe that for any value L ≥ opt, the deadline instance in Step 5: the optimal schedule itself must meet the deadlines {D j }. Combined with the deadlinemetric algorithm (Theorem 2), it follows that our algorithm's schedule for any value L ≥ opt is feasible using 2-speed. Thus the final lastsuccess value is at most opt, which is also an upper bound on the algorithm's minimax objective.

As in Corollary 1, the bicriteria guarantees can be combined for some metrics.

Corollary 2. There is a 2-approximation algorithm for minimizing maximum completion time in malleable scheduling with chain precedence constraints, including makespan and maximum stretch.

Stage 3: Converting Pseudo-Job Schedule into Valid Schedule

The final stage combines the output of Stages 1 and 2, converting any malleable schedule of pseudo-jobs and chains into a valid schedule of the original jobs and flows. We consider the schedule of each pseudo-job k separately. Using a generalization of McNaughton's Rule [START_REF] Mcnaughton | Scheduling with Deadlines and Loss Functions[END_REF], we will construct a malleable schedule for the (portions of) jobs comprising pseudo-job k. The original precedence constraints are satisfied since the chain constraints are satisfied on pseudo-jobs, and the jobs participating in any single pseudo-job are independent.

Consider any pseudo-job k that corresponds to interval I k in the pseudoschedule (recall Stage 1), during which jobs S ⊆ [n] are executed in parallel for a total of r k = i∈S δ i processors. Consider also any malleable schedule of pseudo-job k, that corresponds to a histogram σ (of processor usage) having area s k = |I k | • r k and maximum number of processors at most r k .

We now describe how to shape shift the pseudo-schedule for S in I k into a valid schedule given by histogram σ. The idea is simple: Decompose the histogram σ into intervals J of constant numbers of processors. For each interval J ∈ J , having height (number of processors) σ(J), we will schedule the work from a time |J|•σ(J) r k sub-interval of I k ; observe that the respective areas in σ and I k are equal. Since J∈J |J|•σ(J) = s k = |I k |•r k , taking such schedules over all J ∈ J gives a full schedule for I k . For a particular interval J, we apply McNaughton's Rule to schedule the work from its I k sub-interval. This rule was extended in [START_REF] Drozdowski | Scheduling Parallel Tasks With Sequential Heads and Tails[END_REF] to cover a scenario more like ours. It has linear complexity. McNaughton's Rule is basically a wrap-around scheme: We order the jobs, and for the first job we fill the area horizontally, one processor at a time, until the total amount of work involving that job has been allotted. Then, starting where we left off, we fill the area needed for the second job, and so on. All appropriate constraints are easily seen to be satisfied.

Simulation Experiments

In this section we describe the performance of our FlowFlex algorithm via a variety of simulation experiments. We consider two competitors, Fair [START_REF] Zaharia | Job Scheduling for Multi-User MapReduce Clusters[END_REF] and FIFO. We will compare the performance of each of these three in terms of the best lower bounds we can find for these NP-hard problems. (There is no real hope of finding the true optimal solutions in a reasonable amount of time, but these lower bounds will at least give pessimistic estimates of the quality of the FlowFlex, Fair and FIFO schedules.) We will consider nearly all combinatorial choices of scheduling metrics, from five basic types. They are based on either completion time, number of tardy jobs, tardiness and SLA step functions. (See Figure 2.) They can be either weighted or non-weighted, and the problem can be to minimize the sum (and hence average) or the maximum over all flows. So, for example, average and weighted average completion time are included for the minisum case. So is average stretch, which is simply completion time weighted by the reciprocal of the amount of work associated with the flow. Similarly, makespan (which is maximum completion time), maximum weighted completion time, and thus maximum stretch is included for the minimax case. Weighted or unweighted numbers of tardy jobs, total tardiness, total SLA costs are included in the minisum case. Maximum tardy job cost, maximum tardiness and maximum SLA cost are included in the minimax case. (A minimax problem involving unit weight tardy jobs would simply be 1 if tardy flows exist, and 0 otherwise, so we omit that metric.) We note that that these experiments are somewhat unfair to both Fair and FIFO, since both are completely agnostic with respect to the metrics we consider. But they do at least make sense, when implemented as ready-list algorithms. (In other words, they simply schedule all ready jobs by either Fair or FIFO rules, repeating as new jobs become ready.) We chose not to compare FlowFlex to Flex, because that algorithm does optimize to a particular metric, and it is not at all obvious how to "prorate" the flow-level metric parameters into a set of per job parameters.

The calculation of the lower bound depends on whether the problem is minisum or minimax. For minisum problems the solution to the minimum cost flow problem provides a bound. For minimax problems the maximum of the critical path objective function values provides a lower bound. But we can also potentially improve this bound based on the solution found via the bracket and bisection algorithm. We perform an additional bisection algorithm between the original lower bound and our solution, since we know that the partial sums of the squashed area bounds must be met by the successive deadlines.

Each simulation experiment was created using the following methodology. The number of flows was chosen from a uniform distribution between 5 and 20. The number of jobs for a given flow was chosen from a uniform distribution between 2 and 20. These jobs were then assumed to be in topological order and the precedence constraint between jobs j 1 and j 2 was chosen based on a probability of 0.5. Then all jobs without successors were assumed to precede the last job in the flow, to ensure connectivity. Sampling form a variety of parameters governed whether the flow itself was "big" in volume, and also whether the jobs in that flow were "tall" and/or "wide" (that is, having maximum height equal to the number of slots). Weights in the case of completion time, number of tardy jobs and tardiness were also chosen from a uniform distribution between 1 and 10. The one exception was for stretch metrics, where the weights are predetermined by the size of the flow.) Similarly, in the case of SLA metrics, the number of steps and the incremental step heights were chosen from similar distributions with a maximum of 5 steps. Single deadlines for the tardy and tardiness cases was chosen so that it was possible to meet the deadline, with a uniform random choice of additional time given. Multiple successive deadlines for the SLA case were chosen similarly. The number of slots was set to 25. 10 Thus each ratio must be at least 1. Ratios close to 1 are by definition very good solutions, but, of course, solutions with poorer ratios may still be close to optimal. Note that FlowFlex performs significantly better than either Fair or FIFO, and often is close to optimal. FIFO performs particularly poorly on average stretch, because the weights can cause great volatility. FlowFlex also does dramatically better than either Fair or FIFO on the tardiness metrics. Similarly, Figure 11 illustrates the comparable minimax experiments, for those 8 metrics which make sense. Here one sees that makespan is fine for all schemes, which is not particularly surprising. But FlowFlex does far better than either Fair or FIFO on all the others, and some of these are very difficult problems. Again, some of the Fair and FIFO ratios can be quite bad. In all 8 sets of experiments, FlowFlex is within 1.26 of "optimal" on average, and generally quite a lot better.

Cluster Experiments

We have prototyped FlowFlex on the IBM Platform Symphony MapReduce framework with IBM's BigInsights product [START_REF][END_REF].

We used a workload based on the standard Hadoop Gridmix2 benchmark. For each experiment we ran 10 flows, each consisting of 2 to 10 Gridmix jobs of random sizes, randomly wired into a dependency graph by the same basic procedure we used for simulation experiments. The experiment driver program submitted a job only when it was ready. That is, all of the jobs it depended upon were completed. We ran two sets of experiments: one where all flows arrived at once and another where flows arrived at random intervals chosen from an exponential distribution. For each type of experiment we ran three different random sets of arrival times and job sizes. A few comments should be mentioned here. First, we have not yet focused on integrating schemes for estimating the amount of work of each job in the various flows. We do know the number of tasks per job, however, and estimate work for unstarted jobs by using a default work prediction per task. For running jobs we continue to refine our work estimates by extrapolating based on data from the completed tasks. All of this can be improved in the future, for example by incorporating the techniques in [START_REF] Popescu | Same Queries, Different Data: Can We Predict Runtime Performance?[END_REF]. Better estimates should improve the quality of our FlowFlex scheduler. The second comment is that we are using a slight variant of FlowFlex for minisum problems. This variant avoids the minimum cost flow problem. It is faster and approximately as effective. The third comment is that the Reduce phase is ready precisely when some fixed fraction of the Map phase tasks preceeding it have finished. FlowFlex simply coalesces all such ready tasks within a single MapReduce job and adjusts the maxima accordingly.

We compared FlowFlex to Fair and FIFO running for submitted jobs. (They were not aware of jobs that were not yet submitted.) Figure 12 reports the relative performance improvement of FlowFlex for average completion time, makespan, and average and maximum stretch for both sets of experiments. Essentially, we are evaluating the four most commonly used scheduling metrics.

Conclusion

In this paper we have introduced FlowFlex, a MapReduce scheduling algorithm of both theoretical and practical interest. Theoretically, we have extended the literature on malleable parallel scheduling with precedence constraints, linear speedup and processor maxima. We have provided a unified three-stage algorithm for any scheduling metric, and given worst-case performance bounds. These include approximation guarantees where possible, and bicriteria approximation guarantees where not. Practically, FlowFlex is the natural and ultimate extension of Flex, a MapReduce scheduler for singleton jobs already in use in IBM's BigInsights. We have evaluated FlowFlex experimentally, showing its excellent average case performance on all metrics. And we have shown the superiority of FlowFlex to natural extensions of both Fair and FIFO.

Fig. 1 :

 1 Fig. 1: MapReduce and Malleable Scheduling. Practically speaking, FlowFlex and its predecessors are epoch-based. In each epoch FlowFlex wakes up, considers the current version of the scheduling problem, produces a hypothetical malleable schedule and outputs the initial allocations to be implemented by the assignment layer. Thus the size data as well as the flows and jobs themselves are each updated for each FlowFlex run, making the scheduler more robust.

  Figure 2 illustrates 4 basic types of cost functions.

Fig. 2 :

 2 Fig. 2: Typical Cost Functions Types.

Fig. 3 :δi 1 .

 31 Fig. 3: High Level Scheme FlowFlex Overview

Fig. 4 :

 4 Fig. 4: FlowFlex Stage 1.

9

  Clearly, the total size of pseudo-jobs h k=1 s k = n i=1 si the total size of jobs. Moreover, there is only one maximal chain of pseudo-jobs, which has critical path h k=1 s k δ k = h k=1 |I k | = T , the original critical path bound.

Figure 4 (

 4 Figure 4(a) illustrates the directed acyclic graph of a particular flow. Figure 4(b) shows the resulting pseudo-schedule. Figures 4(c) and (d) show the decomposition into maximal intervals.

4 :

 4 Compute minimum-cost flow f in N of ρ := m j=1 Vj demand from r to r . 5: Set deadline dj ← arg min t : t s=1 f (aj, bs) ≥ Vj/2 , for all j ∈ [m]. 6: Solve this deadline metric instance using Algorithm 5.

Fig. 6 :

 6 Fig. 6: Algorithm for Scheduling Chains with Minisum Metric

Fig. 7 :

 7 Fig. 7: The Minimum Cost Flow Network.

Figure 9 (

 9 a) shows a Stage 1 pseudo-schedule and highlights the first pseudojob (interval I 1 ). The lowest histogram σ of Figure 9(b) illustrates the corresponding portion for this pseudo-job in the Stage 2 malleable greedy schedule; the constant histogram ranges are also shown. The equal area sub-intervals in I 1 are shown as vertical lines in Figure 9(a). Applying McNaughton's Rule to the first sub-interval of I 1 we get the schedule shown at the bottom-left of Figure 9(b). The scheme then proceeds with subsequent sub-intervals.

Fig. 9 :

 9 Fig. 9: FlowFlex Stage 3: Shape Shifting.

Fig. 10 :

 10 Fig. 10: Minisum Simulation Results: Average and Worst Case.

Fig. 11 :

 11 Fig. 11: Minimax Simulation Results: Average and Worst Case.

Figure 10

 10 Figure10illustrates both average and worst case performance (given 25 simulation experiments each) for 9 minisum metrics. Each column represents the ratio of the FlowFlex, Fair or FIFO algorithm to the best lower bound available.10 Thus each ratio must be at least 1. Ratios close to 1 are by definition very good solutions, but, of course, solutions with poorer ratios may still be close

Fig. 12 :

 12 Fig. 12: Cluster Experiments, Gridmix2-based Live Benchmark. In these cluster experiments, the schedulers are running in something more like their natural environment. Specifically, they are epoch-based: Every epoch (roughly 2 seconds) they examine the newly revised problem instance. Thus the job sizes for FlowFlex change from epoch to epoch. And, of course, flows and jobs arrive and complete. FlowFlex then produces complete (theoretical) schedules for internal consumption, but also, more importantly, initial allocation suggestions. This is then implemented to the extent possible by the Assignment Layer.

Stretch is a fairness metric in which each flow weight is the reciprocal of its size.

In MapReduce, the atomic unit of allocation is called a slot, which can be used for either Map or Reduce tasks. So "processor" in the theoretical literature corresponds to "slot" in a MapReduce context.

This will be clarified in Subsection 5.2.

Throughout the paper, for any integer ≥ 1, we denote by [ ] the set {1, . . . , }.

Chains are a special case of flows in which precedence is sequential.

Here var stands for malleable scheduling, pi(k) = p i (1) k denotes linear speedup, δi is processor maxima, prec stands for precedence, and * is for any objective function.

To deal with lower bounds of 0, which is possible for some metrics, we added 1 to both the numerator and denominator. The effect is typically quite modest.