David Navalho

Sérgio Duarte

Nuno Preguiça

Marc Shapiro

Incremental Stream Processing using Computational Conflict-free Replicated Data Types

Keywords: D.1.3 [Concurrent Programming]: Distributed programming, Parallel programming; E.1 [Data Structures]: Distributed data structures Stream Processing, Distributed Processing, Incremental Computation, Real Time

Information has become a key commodity for most service providers. Analyzing streams of data efficiently, in real time, has become increasingly more important for supporting new products and applications.

This paper outlines a novel abstraction for performing incremental stream processing based on Computational Conflict-free Replicated Data Types. C-CRDTs are replicated objects that can be updated concurrently without coordination to perform a computation and still converge to a consistent state that reflects all contributions.

Results obtained with a preliminary prototype show that C-CRDTs have the potential to match and improve computational throughput when compared with a state of the art stream processing system.

Introduction

Information has become a key commodity for all major Internet players providing core services, such as search, social networking or e-commerce. Analyzing personal and business data is now common practice; doing so efficiently, in real time, is increasingly more important for supporting new products and applications.

Data processing solutions for cloud computing infrastructures, such as Map-Reduce [START_REF] Dean | Mapreduce: simplified data processing on large clusters[END_REF] and DryadLINQ [START_REF] Yu | Dryadlinq: a system for generalpurpose distributed data-parallel computing using a high-level language[END_REF], have been designed for batch processing of bulk data. Effective for their primary goal, they proved inadequate for real-time Permission to make digital or hard copies of all or part of this work for personal or classroom use is granted without fee provided that copies are not made or distributed for profit or commercial advantage and that copies bear this notice and the full citation on the first page. To copy otherwise, to republish, to post on servers or to redistribute to lists, requires prior specific permission and/or a fee. CloudDP '13, April 14, 2013, Prague, Czech Republic. Copyright c 2013 ACM 978-1-4503-2075-7. . . $15.00 processing. Latency can be improved by extending batch processing systems, by exposing partial computations and early results, as in (e.g. Map-Reduce Online [START_REF] Condie | Mapreduce online[END_REF]) or pursuing incremental models [START_REF] Bhatotia | Incoop: Mapreduce for incremental computations[END_REF]. However, pitfalls remain that stem from operating over persisted bulk data and the penalty inherent to secondary storage.

In this paper we outline the foundations of Titan, a system for real-time, incremental data processing that we are currently building. Titan aims to tightly integrate computations with a decentralized storage system, striving for both performance and resilience, to ensure deterministic computations.We expect to achieve the low latency benefits of soft state based computations, with the fault tolerance guarantees of a backing store.

Titan is built on top of the Conflict-free Replicated Data Type (CRDT) abstraction [START_REF] Shapiro | Conflict-free replicated data types[END_REF]. CRDTs allow multiple replicas of the same data (structure) to be modified without coordination, guaranteeing that replicas can be merged later without need for any conflict resolution policy. This allows Titan to aggressively cache data in memory in multiple nodes for achieving high throughput, while resorting to secondary storage as a fallback for long term persistence.

Performing computations over soft state is one of the advantages of stream processing systems (e.g., Yahoo's S4 [START_REF] Neumeyer | S4: Distributed stream computing platform[END_REF]), but resilience to faults becomes a major concern, as they impact consistency and determinism of ongoing computations. The use of CRDTs may help with providing deterministic and consistent fault tolerance.

Unlike existing alternatives, such as stream processing, that favor the execution of arbitrary application code, we want to capture much of the processing logic as a set of known operations over specialized Computational CRDTs, with particular semantics and invariants, such as min/max/average/median registers, accumulators, top-N sets, sorted sets/maps, and so on. Keeping state also allows the system to decrease the amount of propagated information.

Preliminary results obtained in a single example show that Titan has an higher throughput when compared with state of the art stream processing systems.

The remainder of this paper is organized as follows. Section 2 discusses related work. Section 3 introduces compu-tational CRDTs, while Section 4 present Titan's programming model. Section 5 overviews the system design. Section 6 presents a preliminary evaluation of the system and Section 7 concludes the paper with some final remarks.

Related work

Our work has been influenced by research from several areas, which we briefly survey.

MapReduce [START_REF] Dean | Mapreduce: simplified data processing on large clusters[END_REF] is actively used by researchers and companies alike, in order to provide computation of massive amounts of data, with a simple programming model. In recent years, it became clear this model was insufficient for the increasing needs of social applications, improved user interactivity, or simply response time.

Many improvements to MapReduce have been proposed as a response, but one or more of the key problems usually remain: computations have to start and stop with a set amount of data, changes require recomputing everything again [START_REF] Zaharia | Resilient distributed datasets: a fault-tolerant abstraction for in-memory cluster computing[END_REF]; jobs are based on batches with a certain size, and even for incremental computation, one small change on a chunk of data requires recomputing the whole chunk and whatever chunks are associated with it [START_REF] Bhatotia | Incoop: Mapreduce for incremental computations[END_REF]; in all cases, computations are intended as, at most, a single pipeline of transformations over a dataset [START_REF] Bhatotia | Incoop: Mapreduce for incremental computations[END_REF][START_REF] Condie | Mapreduce online[END_REF][START_REF] Dean | Mapreduce: simplified data processing on large clusters[END_REF][START_REF] Zaharia | Resilient distributed datasets: a fault-tolerant abstraction for in-memory cluster computing[END_REF] and do not allow reuse of partial computations by multiple data flows; linking several computations requires work; adding more resources or simply new data or computations after being launched is impractical or impossible; and usually it requires some amount of complexity from the programmer to offer incremental computations [START_REF] Bhatotia | Incoop: Mapreduce for incremental computations[END_REF].

More recently, some stream-based systems have been introduced in order to tackle the processing of alwaysincoming data. Systems such as Yahoo's S4 [START_REF] Neumeyer | S4: Distributed stream computing platform[END_REF], Twitter's Storm [START_REF]Storm[END_REF] and StreamMine [START_REF] Brito | Scalable and Low-Latency Data Processing with StreamMapReduce[END_REF] present good approaches for stream-based computing at the large scale, although they each lack in important areas in order to provide scalable systems for true real-time computations.

None of the systems allow adding more nodes or resources after launch, which is a problem, especially when it is expected that participatory sensing systems are able to scale on demand (i.e., elastic). Additionally, we identify that some computations require faster response times, while others can work more similarly to a MapReduce computation, taking their time and less resources. S4 provides mechanisms for periodic messaging, instead of immediate. Storm doesn't take such an approach into account, and adding that functionality can only be done partially, and with effort.

Regarding failure recovery, none of the approaches are very good. Storm guarantees that messages are delivered at least once, but the user has to take that into account on his code -this can be very hard. In the case of S4, the solutions are to either keep going and assume data was lost, or fail the forward Nodes gracefully, and recompute every partial computation again (similar to Incoop [START_REF] Bhatotia | Incoop: Mapreduce for incremental computations[END_REF]). StreamMine [START_REF] Brito | Scalable and Low-Latency Data Processing with StreamMapReduce[END_REF] presents a checkpoint-based approach for fault-tolerance, although it is unclear how it affects computation latency.

All of the approaches are stuck with the logical notion of a single computational flow, even if they allow multiple applications running in parallel on the system. We think there is a lot of repeated information and computation that may be interesting to reuse for other workflows.

Various solutions for handling concurrent updates have been previously used in storage systems [START_REF] Saito | Optimistic replication[END_REF]. With Titan, we intend to explore methods of using commutativity for parallel computation. We will take advantage of the novel concept of Conflict-free Replicated Data Types(CRDTs) [START_REF] Shapiro | Conflict-free replicated data types[END_REF]. CRDTs can guarantee that replicas converge without additional concurrency control, as long as all replicas receive all updates (either by receiving the operations or synchronizing state). Section 3 expands on this concept.

One of S4's intended objectives is to be science friendly: easy to program. We agree with this statement. We have recently taken an interest on social-scale computations (e.g, Participatory Sensing). There has been increasing research into what can be obtained and processed from sensor data, such as building location-based systems in the absence of GPS [START_REF] Bahl | Radar: An in-building rfbased user location and tracking system[END_REF], road conservation and traffic monitoring [START_REF] Eriksson | The Pothole Patrol: Using a Mobile Sensor Network for Road Surface Monitoring[END_REF][START_REF] Hull | CarTel: A Distributed Mobile Sensor Computing System[END_REF][START_REF] Mohan | Nericell: Rich Monitoring of Road and Traffic Conditions using Mobile Smartphones[END_REF], to extracting and interpreting information from user generated input, such as [START_REF] Cha | The world of connections and information flow in twitter[END_REF]. Not only that, but there has been increased interest in crossing information from multiple sources, in order to provide richer visualizations and information, that would otherwise not be possible [START_REF] Cranshaw | The livehoods project: Utilizing social media to understand the dynamics of a city[END_REF].

In order to provide a framework that allows individuals from different backgrounds to create and test topologies at will, building applications that manage to feed off of each other, and where adding just an additional computational step is easy, some more work and engineering is necessary to eventually achieve this.

Computational CRDTs

Conflict-free replicated data types (CRDTs) [START_REF] Shapiro | Conflict-free replicated data types[END_REF] are data types that allow replicas to be modified without coordination, while guaranteeing that upon synchronization all replicas converge to the same state that considers all concurrently executed updates. There are two flavours of CRDTs, with different sufficient conditions for ensuring final consistency. In an operation-based model, all concurrent operations must commute. In a state-based model, the successive states must form a partial order, updates must always advance in the partial order, and merging states must compute a least-upperbound. A large number of CRDTs have been designed in the past [START_REF] Shapiro | Convergent and commutative replicated data types[END_REF], including sets, maps, sequences, graphs, etc.

Computational CRDTs (C-CRDTs) are CRDTs that perform the result of a computation. Figure 1 presents a simple example: the state-based implementation of an object that keeps the result of computing a sequence of add and subtracts to the initial state of 0. The sum CRDT has a payload with two vectors, one for elements with positive value and

let v = i P [i] -i N [i] 11: compare (X, Y) : boolean b 12: let b = (∀i ∈ [0, n -1] : X.P [i] ≤ Y.P [i] ∧ ∀i ∈ [0, n -1] : X.N [i] ≤ Y.N [i]) 13: merge (X, Y ) : payload Z 14: let ∀i ∈ [0, n -1] : Z.P [i] = max(X.P [i], Y.P [i]) 15: let ∀i ∈ [0, n -1] : Z.N [i] = max(X.N [i], Y.N [i]) 16: merge delta (X, Y ) : payload Z 17: let ∀i ∈ [0, n -1] : Z.P [i] = X.P [i] + Y.P [i] 18: let ∀i ∈ [0, n -1] : Z.N [i] = X.N [i] + Y.N [i] Figure 1. Sum CRDT (state-based)
the other one for elements with negative value. Each replica can independently modify its entry in the vectors, as add and subtract operations are executed. The result of the computation is the difference between the sum of add entries and the sum of subtract entries. On merge, the resulting state will include the maximum for each entry in both vectors.

This guarantees that the state of all replicas will evolve to the same state. The state of each replica includes the computed result of a sequence of updates, thus efficiently propagating the contributions of a a potentially large number of operations in a single synchronization operation.

Although the previous example is very simple, it is clear that the same approach could be used for a large number of other computations -e.g., a counter would include only the vector with positive values; for computing the average, the sum of elements and the number of elements would be kept; for computing top-N, the subset of top-N elements is kept; etc. For example, C-CRDTs can be created by leveraging previous work on the definition of decentralized algorithms for computing aggregation functions [2, 7, 8, 10, 13].

Hollow replicas

In the context of C-CRDTs, for efficiency purposes, it is often interesting to instantiate pristine replicas of C-CRDTs lacking remote state -we call these replicas hollow replicas. For example, while counting the number of words in a text stream, the application code that processes the stream wants to increase the sum of words whenever a new word appears. In this case, it is only necessary to accumulate the incremental part of the sum computation, without need of having access to the current value.

To this end, we use hollow replicas -a replica with no initial state. In this example, we would instantiate a hollow replica of the sum CRDT. Thus, a hollow replica accumulates and later transports the contribution of the stream processed by the application to the overall result. Merging a hollow replica with a full replica is done by a new merge function defined in C-CRDTs: the merge delta function. The merge delta for the sum C-CRDT defined in Figure1 consists in adding the value of each entry in the vectors. For a map, it would consist in: for new keys, adding them to the full replica; for existing keys, merging the sum CRDTs.

The state of a hollow replica can be synchronized with a full replica, thus being promoted to a full replica and maintaining the complete state of the object. In this case, it is still possible to accumulate and propagate the delta of the state to the other replicas (in Figure 1 we omit this code for simplicity). While hollow replicas perform a task similar to combiners in Map-reduce systems [START_REF] Dean | Mapreduce: simplified data processing on large clusters[END_REF], keeping the full state of the replica can reduce the information that needs to be further propagated -e.g. in a top-N object, it is not necessary to propagate the local top-N elements that will not feature the global top-N.

In the next section we discuss how Titan combines C-CRDTs to express computations.

Programming model

The Titan programming model involves two main levels of algorithmic abstraction. One concerns the assembly of a pipeline of C-CRDTs that transforms the input data into the desired result, in as many intermediate steps as needed. The other pertains to selecting appropriate partitioning strategies for those C-CRDTs to lead to an efficient and balanced use of the available computational resources and potentiate the desired level of performance.

Titan is geared towards continuous, incremental processing, where intermediate results can be exposed to feed other computations or external consumer applications. Sharing common parts of computational pipeline and partial results among applications is transparent to Titan applications. To that end, Titan allows a new computation to reference the output of any pipeline stage of other instantiated computations as one of its inputs.

Chaining C-CRDTs to produce a complex computation in Titan is, first and foremost, centered on exploiting the specific computational semantics and invariants that are embodied in each C-CRDT type. For instance, a set besides storing values, also filters duplicates; whereas, a sorted set will extend that contract with an ordering invariant. As such, typical Titan application code will not explicitly perform operations such as testing for duplicates or sort data; likewise, if a computation step consists in a sum of numerical values, that result will be achieved via the inclusion of a sum C-CRDT in the computation pipeline, rather than explicitly resorting to the addition operator of the programming language.

In general, complex computations will be decomposed into a series of processing steps, where a partial result will feed the next stage of a processing pipeline, incrementally. The glue that connects the pipeline stages together and carries data from beginning to the end can be abstracted, as it is transparently provided at the system level. However, in most cases, the application is expected to provide the code that takes the output of a C-CRDT of a given type and adapts it to the input type of the next computation stage, such as splitting a sentence into its words before they can be individually added to the hollow replica of a set that is computing the unique words of a text stream in the next pipeline stage.

Decentralized and parallel processing is vital for speeding up Titan computations. The distributed nature of Titan computations arises lenghtwise in the processing pipelines, but also at each pipeline processing stage via partitioning of C-CRDTs. For C-CRDT types that support partitioning, typically containers such as sets and maps, partitioning is achieved by pairing each C-CRDT instance with a map function that projects its domain (e.g., the elements of a set) into one of several logical partitions. Via partitioning it is possible, for instance, to exploit spatial or temporal coherence to group and process logically related data in the the same node, when each logical partition is mapped to the actual physical partitions supported by the available computing nodes. Note that, unlike logical partitioning, mapping logical to physical partitions is automatic by default.

One important aspect that is reflected in the overall algorithmic logic of Titan computations is that C-CRDT partitioning embodies the map-reduce mechanism that is also found in other stream processing architectures. Partitioning implies a reduce step that in C-CRDT nomenclature is captured in the merge operation specific to each C-CRDT type. The C-CRDT merge operation is fundamental for combining the partial computations performed in the hollow replicas induced by C-CRDT partitioning and, in general, for pushing forth incremental updates through the computation pipeline. As such, when reasoning about the chained effects of the semantics of several C-CRDTs, the particular behaviour of merge operations involved merits close attention. The general rule is that the merge operation preserves the C-CRDT invariants, if needed via some built-in compensation logic. However, for more complex C-CRDTs, such as containers of C-CRDTs, merging may have far reaching side-effects as part of the invariant preservation, possibly producing cascading merge of other C-CRDTs. For instance, when instances of a C-CRDT map are merged, any colliding keys will cause their respective values to be merged. If those values are themselves C-CRDTs then another merge operation will be involved, and so forth. This behaviour can be exploited to perform complex computations, especially when matched with judicious partitioning of the computation pipeline C-CRDTs. To further illustrate the programming model, we next describe a couple of simple computation examples.

Examples

Our two examples consist, respectively, in counting the total of words arriving in a Twitter stream and computing the top N most popular words, as determined by the number of occurrences. For this purpose, consider each individual tweet as message payload comprising a piece of text, made of words, accompanied by some metadata that includes an unique identifier. The computation is incremental and in realtime, meaning the values computed represent the results obtained so far. They will be continuously updated until the source stream runs out. A diagram representing the two computations is presented in Figure 2.

Counting Words Computing the total number of words in the stream begins with storing incoming tweets in a set (TweetSet). Any duplicates will be filtered out by the set semantics. To allow the computation to run in parallel, the TweetSet will be partitioned into a K fixed number of logical partitions. The partition function in this case simply maps the tweet identifier present in the metadata to a numerical value between 0 and K. Using hashing and modular arithmetic, we can expect to distribute tweets uniformly across the K partitions. The total number of words seen so far is produced in the next stage of the computation, which consists of a C-CRDT counter (TotalWords). As such, each of the TweetSet's K partitions will compute (independently) a subtotal from the tweets it stores; that subtotal will be accumulated in a local hollow replica of TotalWords. As the computation progresses TotalWords' hollow replicas are periodically merged with the primary, allowing the TotalWords C-CRDT to converge to the global total count.

TopN Words

The most frequent words in a Twitter stream are found by counting the number of occurrences of each word. That is accomplished with a map, whose keys are words and its values C-CRDT counters (MapOfWordOccurrences). Just as before, this C-CRDT will be fed, independently, by the K partitions of TweetSet C-CRDT to populate local hollow replicas of MapOfWordOccurrences. Periodic merging of these hollow replicas will update the counters in the primary, taking advantage of the semantics of the map merge operation, which merges values of colliding keys. The next step in the computation is sorting the words and retain-ing the top N with highest frequencies. This is done by the topN CRDT (TopNWords) -a sorted (value) map of a maximum fixed size, storing words as its keys and counters for values. However, computing a top requires sorting -an expensive operation best performed in parallel. To achieve that, MapOfWordOccurrences needs to be partitioned. Since updating or merging topN C-CRDT replicas (hollow or otherwise) can discard excess elements, it mandates that the partitioning scheme must not scatter words across different partitions, otherwise the global top-N computed from the partials produced at each MapOfWordOccurrences partition would not be correct. Partitioning MapOfWordOccurrences based on its keys (i.e., words) avoids the problem. As tweets enter the pipeline, they produce a cascading effect that eventually gets to the TopNWords hollow replicas and, consequently via periodic merge, update the primary.

Titan Overview

In this section, we briefly present an overview of the current Titan design. Titan is designed to run in a set of nodes in a single cluster. The nodes are organised in a one-hop DHT, which allows any pair of nodes to communicate directly. Each physical node manages and stores and provides execution for a collection of C-CRDTs or C-CRDT partitions. The C-CRDT storage is versioned, maintaining for each C-CRDT, a sequence of versions. C-CRDT are partitioned according to the partition scheme defined by the programmer, upon instantiation. Logical C-CRDT partitions are mapped automatically to their physical counterparts, by hashing logical partition keys to DHT keys and selecting a node accordingly. Physical partitions are created or loaded on demand, as data arrives and hollow replicas perform their first merge with the primary C-CRDT.

An application installs a computation pipeline by creating and naming the desired C-CRDTs and providing the necessary gluing code. Typically, there will also be some client code that receives the data stream from an external source and starts the computation pipeline. As a means to consume computation results, client applications can request to have C-CRDT update notifications to be streamed to them, or perform an explicit read. In the example of Figure 2, this application would consist in consuming the Twitter stream and adding the new tweets to the TweetSet object, itself exposed as an hollow replica.

At system level, the programmer supplied code that glues C-CRDTs together and populates hollow replicas are used in triggers, associated with each C-CRDT instance. As in Percolator [START_REF] Peng | Large-scale incremental processing using distributed transactions and notifications[END_REF], triggers are the basis for incremental processing. They can run according to multiple policies, depending on how the C-CRDT was instantiated: whenever new data exists; whenever some condition on the data is true; or periodically. Each C-CRDT can have multiple triggers associated if it is involved in multiple computation pipelines. In the example of Figure 2, the TweetSet partitions have an un-derlying trigger that runs whenever a new tweet is available, which feeds the C-CRDTs that follow in the computational pipeline, by populating their respective hollow replicas.

Evaluation

In this section, we present a few experimental results with our preliminary and incomplete Titan prototype. We deployed a similar setup using Storm [START_REF]Storm[END_REF] to have some measure of comparability. We chose Storm instead of S4 [START_REF] Neumeyer | S4: Distributed stream computing platform[END_REF] because it seems to have a very active user community, with much more frequent fixes and updates.

In the following sections, we will describe our example application and it's deployment strategy on both Titan and Storm. We then describe our testing environment and finally present our initial results using both systems.

Example Application: Word Count Our test application consists in counting the total words found in a Twitter stream, as outlined in Section 4.1. The Storm implementation followed its programming model. We used an initial set of bolts (Logical Nodes) for receiving and splitting Tweet-Messages; followed by several Spouts (Logical Nodes) to gather these words and count them. Finally, one last Spout gathers the word counts into a total word count. To provide a fair comparison, we made an effort to optimize the Storm application, by implementing bulk processing and finding the best deployment topology in our testbed.

Testing and Configurations

We ran the experiments presented in this section in a cluster of 8 nodes, each one equipped with a Quad-Core AMD Opteron 2376 at 2.3Ghz, 4x512KB cache L2, and 8GB of RAM. The operating system is Debian 5.0.8, running Linux 2.6.26-2-amd64 kernel, and the nodes are interconnected via a private gigabit ethernet. The installed Java Platform was version 6 (OpenJDK, IcedTea6 1.8.10, 6b18-1.8.10-0 lenny2). For serving Twitter data, we had several clients preload Tweets into memory on three of the cluster computers. The remaining five were used to gather and process the data according to the configurations explained next. We ran several Storm topologies, with varying configurations. For the purpose of this article, we present the best configuration we found for running the Storm topology, with 4 initial nodes (spouts) for splitting the incoming tweets into words, 4 nodes (bolts) counting word occurrences, and a final node (bolt) that summed everything. Under this setup, Storm was gathering 100 Tweets per read from the clients.

Regarding the Titan experiment, we present two different runs: one similar to Storm, as described before, using the 5 machines for running Titan computations, and a second configuration using a single machine for Titan computations.

The results from these experiments are found in Figure 3. Despite the preliminary nature of our results and the simplicity of the example, the outlook is promising. Parallel Titan's throughput is both higher and more consistent than Storm's, reaching its maximum performance much earlier. Parallel Titan finishes first and compares well to the single-machine execution, taking advantage of decentralized processing.

Conclusions

This paper outlined Titan, a system aiming for real-time, incremental data processing, whose main abstraction, introduced in this paper, is the Computational CRDT.

C-CRDTs perform elementary computations and maintain specific invariants. As any other CRDT, they can be replicated in any number of nodes and modified without coordination, while being guaranteed that replicas can be merged automatically with all updates being considered.

Titan can be seen as a decentralized storage and execution environment for C-CRDTs, supporting computations defined as a graph of C-CRDT nodes interfaced through small pieces of application code. Results obtained with our preliminary prototype are encouraging, showing an improvement in throughput when compared to Storm.

Titan is still under active development. Our future work will focus on dynamic and multi-partitioning of C-CRDTs, resilience to faults and improving persistent storage integration in the system.

1 : 7 :

 17 payload number[n] P , number[n] N --One entry per replica 2: initial [0, 0, . . . , 0], [0, 0, . . . , 0] 3: update add (N) 4: let g = myID() -g: source replica 5: P [g] := P [g] + N 6: update subtract (N) let g = myID() 8: N [g] := N [g] + N 9: query value () : number v 10:

Figure 2 .

 2 Figure 2. Example of two computations sharing part of the processing pipeline

Figure 3 .

 3 Figure 3. Counting Words experimental results

Acknowledgments

We sincerely thank our anonymous reviewers for their insightful comments and suggestions. This research has been supported by CITI grant PEst-OE/EEI/UI0527/2011 and FCT/MEC projects PTDC/EIA-EIA/108963/2008 and PTDC/EEI-SCR/1837/2012.


Pierpaolo Cincilla
email: pierpaolo.cincilla@lip6.fr

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

Marc Shapiro

Gargamel: boosting DBMS performance by parallelising write transactions

Keywords: Cloud computing, distributed DBMSes, scheduling alghorithms

Parallel transactions in distributed DBs incur high overhead for concurrency control and aborts. We propose an alternative approach by pre-serializing possibly conflicting transactions, and parallelizing non-conflicting update transactions to different replicas. Our system provides strong transactional guarantees. In effect, Gargamel partitions the database dynamically according to the update workload. Each database replica runs sequentially, at full bandwidth; mutual synchronisation between replicas remains minimal. Our simulations show that Gargamel improves both response time and load by an order of magnitude when contention is high (highly loaded system with bounded resources), and that otherwise slow-down is negligible.

I. INTRODUCTION

Databases often scale poorly in distributed configurations, due to the cost of concurrency control and to resource contention. The alternative of centralizing writes works well only for read-intensive workloads, whereas weakening transactional properties is problematic for application developers.

Our approach is to classify transactions according to their predicted conflict relations at a front-end to the replicated database. Non-conflicting transactions execute in parallel at separate replicas, ensuring high throughput; both read-only and update transactions are parallelised. Transactions that may conflict are submitted sequentially, ensuring that they never abort, thus optimising resource utilisation. This approach guarantees Parallel Snapshot Isolation (PSI), which is similar to SI, but allows the system to replicate transactions asynchronously [START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF]. It does not require (costly and slow) global synchronisation. Our system, Gargamel, operates as a front-end to an underlying database, obviating the cost of lock conflicts and aborts. Our approach also improves locality: effectively, Gargamel partitions the database dynamically according to the transaction mix. All this results in better thoughput, response times, and use of resources: our simulations show considerable performance improvement in highly-contended workloads, with negligible loss otherwise.

Our current classifier is based on a static analysis of the transaction text (stored procedures). This approach is realistic, since the business logic of many applications (e.g., This research is supported in part by ANR project ConcoRDanT (ANR-10-BLAN 0208) and by ANR project Prose (ANR-09-VERS-007-02).

e-commerce site OLTP applications) is encapsulated into a small fixed set of parameterized transaction types, and since ad-hoc access to the database is rare [START_REF] Stonebraker | The end of an architectural era (it's time for a complete rewrite)[END_REF].

The static analysis is complete, i.e., there are no false negatives: if a conflict exists it will be predicted. However, false positives are allowed: a conflict may be predicted where none occurs at run time.

Our contributions are the following:

• We show how to parallelize non-conflicting transactions by augmenting a DBMS with a transaction classifier front end, and we detail the corresponding scheduling algorithm. Each replica runs sequentially, with no resource contention. • We propose a simple prototype classifier, based on a static analysis of stored procedures. • We demonstrate the effectiveness of our approach with extensive simulation results, varying a number of parameters, and comparing against a simple Round-Robin scheduler, the state-of-the-art Tashkent+ [START_REF] Elnikety | Tashkent+: memory-aware load balancing and update filtering in replicated databases[END_REF] and against a centralised-writes system. • We conclude from the simulations that: (i) At high load, compared to competing systems, Gargamel improves response time and throughput of update transactions by 25% and 75% respectively, in the TPC-C benchmark. At low load, Gargamel provides no benefit, but overhead is negligible. (ii) The Gargamel approach requires far fewer resources, substantially reducing monetary cost. The paper proceeds as follows. Section II overviews the Gargamel approach. The justification for its correctness is discussed in Section III. We describe our simulation model in Section IV. The experimental simulation results are detailed in Section V. We compare with related work in Section VI. Finally, we conclude and consider future work in Section VII.

II. GARGAMEL ARCHITECTURE

A. Overview

The aim of Gargamel is to dynamically partition the workload into disjoint queues, such that no transaction in some queue conflicts with one in any other. Two queues can execute in paralle on separate replicas. To avoid aborts, transactions within a queue execute sequentially. The conflict definition is determined by the isolation level. Our current classifier implements PSI, i.e., transactions conflict iff their write-sets intersect. The classifier implementation for the TPC-C benchmark is based on a static analysis of the workload.

The Gargamel load balancer is interposed between clients and database replicas (called workers hereafter), as illustrated in Figure 1.

When it receives a new transaction, Gargamel checks for conflicts against already-scheduled transactions, and assigns it to a queue accordingly; more details later.

B. Scheduling Algorithm

As transactions enter and exit the system, conflict relations appear and disappear in complex ways. To keep track of this, Gargamel maintains generalized queues called chains. Conflicting transactions execute sequentially at a single worker; non-conflicting transactions are assigned to parallel queues, which execute, without mutual synchronisation, at different workers.

For some incoming transaction t, if t is classified as conflicting with t ′ , then t is queued after t ′ . If t conflicts with transactions t ′ and t ′′ that are in two distinct queues, then the transaction is scheduled in a chain merging the two queues. Similarly, if two transactions t ′ and t ′′ both conflict with transaction t, but t ′ and t ′′ do not conflict with each other, then they will be put into two distinct queues, both after t. We call this case a split.

Figure 2 presents some examples. Clients submit transactions t 1 , t 2 , etc. Initially, the system is empty: t 1 does not conflict with any transaction; therefore the scheduler allocates its first chain, containing only t 1 . When the scheduler receives t 2 , it compares it with the only other transaction, t 1 . If they conflict, the scheduler will append t 2 at the end of the queue containing t 1 ; otherwise, the scheduler assigns An initially empty system receives t 1 , t 2 , etc.: (a) t 1 and t 2 do not conflict. (b) t 1 and t 2 do not conflict, and t 3 conflicts with t 1 but not with t 2 . (c) t 3 conflicts with both t 1 and t 2 . Consider the latter case: transaction t 3 arrives; the scheduler compares it to t 1 and t 2 . If t 3 conflicts with neither of them, it is placed into a new queue. If t 3 conflicts with a single one, it is queued after it (Figure 2(b)). If it conflicts with both, the two queues merge (Figure 2(c)). If transactions t 4 and t 5 both conflict with t 3 but not with each other, they will be on parallel queues but both after t 3 (the queue splits (Figure 2(d))). Repeating the algorithm, Gargamel computes chains that extend, merge or split according to conflicts between pairs of transactions (Figure 2(e)).

The number of classifications is, in the worst case, equal to the number of transactions queued and not yet committed plus the last transactions that update items affected by the incoming transaction.

Once a worker has finished a transaction it propagates the write-set to all other workers using causal order broadcast [START_REF] Guerraoui | Introduction to Reliable Distributed Programming[END_REF]. The write-set propagation is done in the background. Workers wait for propagation of write-sets of conflicting transactions to proceed before starting execution. For example to execute a transaction after a merge, the worker waits to receive the write-sets of transactions queued before the merging point.

III. CORRECTNESS

Gargamel ensures the Parallel Snapshot Isolation (PSI) property [START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF] thanks to the following features:

(i) Gargamel ensures that conflicting transactions do not execute concurrently. This implies a fortiori that a transaction commits only if its updates do not conflict with a concurrent transaction. (ii) Workers propagate their updates using causal order broadcast. When a replica receives the write-set of some transaction, it is guaranteed to have already received the write-sets of preceding transactions. (iii) Each replica's database engine ensures SI locally.

Causal order propagation of updates ensures that a Transaction durations are taken from a Gaussian distribution with the indicated mean and variance. The numerical parameters of TPC-C are taken from the referenced measurements [START_REF] Hewlett-Packard | TPC benchmark C: full disclosure report for HP ProLiant BL685c G7[END_REF].

Table I SIMULATION PARAMETERS

transaction t can't commit after t ′ at any site if t ′ has observed the updates of t. (iv) A worker does not start an update transaction until it has received the updates of preceding transactions.

The conjunction of these properties ensures that concurrent transactions commute.

To explain property (iv) in more detail, note that, since communication between the scheduler and replicas is asynchronous, a replica might receive an update transaction from the scheduler before it is ready. The replica delays the execution of the new transaction until it has received the updates from its predecessors. Read only transactions are never delayed, they can read stale values.

The scheduler maintains the identifier of the last transaction that updated any given data item. The classifier maintains a list of data items affected by any given transaction type. Therefore it is easy to extract the identifier of the latest transaction that conflict with any new transaction.

When the scheduler sends a new transaction to some replica, it piggy-backs the list of recent conflicting transactions executed at a different replica (this may occur, for instance, when chains merge). When it receives this, the replica checks that the corresponding updates have been received; if not, the new transaction is delayed.

IV. SIMULATION MODEL

To evaluate the Gargamel approach, we have implemented a discrete event simulator. In this section we present the simulation model and describe the benchmark we use. The experiments and results are discussed in Section V.

A simulation has two phases. In the first phase, clients submit transactions to the scheduler at the specified incoming rate. During the second phase, no more transactions are submitted. It lasts until every transaction has finished executing. The first phase lasts approximately 28 s.

Unless specified otherwise, all simulations use the parameters presented in Table I.

A. TPC-C simulation

TPC-C [START_REF] Tpc | TPC benchmark C Standard Specification Revision 5[END_REF] is a standard benchmark for OLTP transactions. The numeric simulation parameters of Table I are taken from actual TPC-C measurements [START_REF] Hewlett-Packard | TPC benchmark C: full disclosure report for HP ProLiant BL685c G7[END_REF]. We simulate TPC-C with ten warehouses.

It is composed of five kinds of transactions: New-Order (N O), Payment (P ), Order-Status (OS), Delivery (D), and Stock-Level (SL). 92% of the workload consists of update read-write transactions; the remaining 8% are read-only (OS and SL).

A N O(w, d, I) transaction adds an order to a warehouse. Its parameters are a warehouse w, a district d, and a list of items I. Each item i(w ′ , d ′ ) ∈ I has two parameters: its warehouse ID w ′ and the item ID d ′ . An I list contains between 5 and 15 elements. N O transactions occur with high frequency and relatively costly.

The parameters of a Payment transaction P (w, c, cw , cd ) are a warehouse ID w, a customer c, a customer warehouse cw , and a customer district cd . The customer c is selected 60% of the time by name, and 40% of time by unique identifier. Homonyms are possible in the former case. P and NO transactions dominate the workload.

The single parameter of a Delivery transaction D w is the warehouse ID w.

Our classifier is based on a very simple static analysis: two transactions are considered to conflict: (i) if they update the same column of a table, and (ii) unless it is clear from the analysis that they never update the same row. In the case of TPC-C, conflicts may happen between pairs of the same transaction type (N O and N O, P and P , D and D) and between Payment and Delivery transactions. Table II shows which transactions conflict according to their parameters.

Because homonyms cannot be checked statically, transaction classification admits false positives between two Payment transactions and between a Payment and a Delivery transaction. If the customer is identified by name, the classifier conservatively assumes that a conflict is possible. Around 0.04% of classifications return a false positive, i.e. they predict a conflict that will not appear at run-time.

Delivery transaction pairs do not cause false positives because the customer selection here is based on the lowest ID (representing the oldest N O), which is unique for a given snapshot.

B. Round Robin, Centralised-Writes and Tashkent+ simulation

Our simulations compare Gargamel with a simple Round Robin scheduler, and with the state-of-the-art Tashkent+ [START_REF] Elnikety | Tashkent+: memory-aware load balancing and update filtering in replicated databases[END_REF].

Transaction pairs Conflict condition N O(w 1 , d 1 , I 1 ) × N O(w 2 , d 2 , I 2 ) (w 1 = w 2 ∧ d 1 = d 2 )∨ I 1 ∩ I 2 = ∅ P (w 1 , c 1 , cw 1 , cd 1 ) × P (w 2 , c 2 , cw 2 , cd 2 ) (w 1 = w 2 ) ∨ ((cw 1 = cw 2 ∧ cd 1 = cd 2 ) ∧ (c 1 = c 2 )) D(w 1 ) × D(w 2 ) w 1 = w 2 D(w 1 ) × P (w 2 , c 2 , cw 2 , cd 2 ) w 1 = cw 2
The subscripts represent two concurrent transactions. Please refer to Section IV-A for an explanation of variable names.

Table II CONFLICTS OF TPC-C UNDER SNAPSHOT ISOLATION

We also compare Gargamel with a centralised-writes system when possible.

Round-Robin aims to maximise throughput by running as many transactions in parallel as possible. It works as follows.

Transactions are assigned to each worker in equal portions and in circular order. Because concurrent transactions proceed in parallel, Round-Robin suffers from a lot of abortrestarts, i.e., wasted work.

A Centralised-Write system runs all read-only transactions concurrently, but serialises all update transactions at a single worker in order to avoid wasted work. It can be considered as an idealized version of Ganymed [START_REF] Plattner | Ganymed: Scalable replication for transactional web applications[END_REF]. Centralised-Write was simulated quite simply by classifying all update transactions as mutually-conflicting; thus all update transactions are put into a single queue, executed by a single worker. Our simulations show that Centralised-Write is overly conservative on standard benchmarks.

Like Round-Robin, Tashkent+ aims to maximise throughput, but optimises the assignment of transactions to workers by ensuring that the working set of the transactions sent to a worker fits into the worker's main memory. Workers are assembled in groups executing the same set of transaction types. To balance the load, Tashkent+ monitors each worker's CPU and disk usage, and rearranges groups, by moving workers from the least loaded group to the most loaded group.

Tashkent+ estimates the working set of an incoming transaction by examining the database execution plan. Our simulated Tashkent+ extracts the execution plan from TPCC-UVA [START_REF] Llanos | TPCC-UVA: An open-source TPC-C implementation for global performance measurement of computer systems[END_REF], an open source TPC-C implementation.

Our simulator implements the Tashkent+ group allocation/re-allocation algorithm as described in the literature [START_REF] Elnikety | Tashkent+: memory-aware load balancing and update filtering in replicated databases[END_REF]. Since CPU and disk usage are not significant in this simulation, we estimate the load by the ratio of busy to available workers. Replica allocation and re-allocation are implemented in such a way that the balance remains optimal all the time. Our simulations are favorable to Tashkent+ because we assume that re-allocation has no cost.

As the literature shows that Tashkent+ improves performance by reducing disk access, our simulation takes this into account by reducing the duration of every transaction by 10% under Tashkent+. Nevertheless, our simulations hereafter show that Tashkent+ suffers from lost work under TPC-C. All systems perform transactions at a single replica and then diffuse the write-set to other replicas.

V. EXPERIMENTAL RESULTS

We measure transaction throughput, response time, and amount of resources consumed, comparing Gargamel, Tash-kent+ and Round-Robin. When relevant, we compare also with Centralised-Write. This set of experiments is based on TPC-C, varying the transaction incoming rate.

A. Performance

When the incoming rate is low, parallelism is low, therefore conflicts are not an issue. In such a situation, all schedulers are basically equivalent. For instance, Gargamel will schedule each incoming transaction in a new chain and execute it immediately. Figure 3 confirms that at 10 trans/s (transactions per second) Gargamel, Tashkent+ and Round-Robin have indistinguishable throughput. Vertical lines show, for each system, when the first and the last client stop submit transactions.

Things get more interesting at high incoming rates. Figure 4 compares the throughput of the three systems at 200 trans/s. Figure 5 shows maximum throughput, varying the incoming rate. The two figures show that Gargamel exhibit a significant improvement compared to the other systems during the first phase (while clients submit transactions). In the second phase (when no more transactions are submitted), parallelism decreases and the throughput of all three systems decreases consequently.

Figure 5 shows that the improvement of Gargamel over Tashkent+ and Round-Robin grows with the incoming rate. Tashkent+ and Round-Robin both flatten out at ≃ 150 trans/ s. Gargamel's maximum throughput is twice that of Tash-kent+ and Round-Robin.

At 250 trans/s, Gargamel saturates the available workers, and, at constant number of workers, any increase in incoming rate does not provide any improvement.

We estimate the response time by measuring the "penalty ratio," i.e., response time (time elapsed between transaction submission and transaction commitment) divided by transaction duration. The lower the scheduling delays, the lower the penalty ratio suffered by clients. Figures 6 and7 show the penalty ratio and its CDF, comparing Gargamel, Tashkent+ and Round-Robin. Figure 6 shows the average penalty ratio of ten runs, Figure 7 shows the CDF of one run. Gargamel's penalty is approximately 20% lower than Tashkent+ and Round-Robin. 51% of the transactions in Gargamel suffer a penalty of 4 or less, whereas this is the case of only 10% of transactions (approximately) in the competing systems.

The speedup is estimated by dividing the sequential execution time by the total execution time. The speed-up improvement is low (Figure 8): while most transactions execute with little delay (as shown by the penalty ratio CDF) the longest-waiting transaction is delayed almost identically in all three systems. This is due to the fact that conflicting transactions must be serialised anyway; the TPC-C workload is dominated by the New Order and Payment transactions, and two New Order transactions can conflict, as well as two Payment transactions. The serialization of conflicting New Order and Payment transactions dominate the speedup. The speed-up of all three systems flatten at an incoming rate of around 50 transactions per second. For higher incoming rates even if the throughput is higher (see Figure 5) the time to execute serially conflicting New Order and Payment transactions remain the same.

B. Resource Utilisation, Bounded Workers

Our next experiments examine resource utilisation and queue size in our systems.

The bottom part of Figure 9 shows the number of busy workers as the simulation advances in time. The top part shows the number of queued transactions (note that for readability the scale of the ordinate differs between the bottom and the top).

For the default number of workers (100), at the default incoming rate (150 trans/s), Gargamel always finds an available worker; queue size is close to zero. This means that the number of parallelizable transactions remains lower than the number of workers, at all times.

As Gargamel, Centralised-Write always finds an available worker, but it paralelise mutch less than other systems, then its execution time is mutch longher.

In contrast, the policy of both Tashkent+ and Round-Robin is to execute as many transactions as possible in parallel, as soon as they arrive. However, since many of those transactions conflict, there are many aborts, and they do not make progress. They quickly saturate the number of workers, incoming transactions are delayed, and queue size grows rapidly.

In the second phase, after transactions stop arriving (incoming rate goes to zero), Gargamel frees most of the workers. Indeed, at this point, all the read-only and nonconflicting transactions finished executing; Gargamel only needs a few workers for the remaining chains of conflicting transactions.

In contrast, Tashkent+ and Round-Robin continue to saturate the workers by attempting to parallelise conflicting transactions. At some point during the second phase, Tash-kent+ re-assigns groups and continues to empty the queue of waiting transactions more slowly. This is because, at this point, all the read-only and non-conflicting transactions have terminated. The remaining transactions conflict with higher probability.

We have also simulated a rate of 300 trans/s (Figure 10). Even for Gargamel the load is too high for the default number of workers, and Gargamel builds a (very small) queue during the first phase.

C. Resource Utilisation, Unbounded Workers

We now consider a system where the number of workers is unbounded, e.g., an elastic cloud computing environment. In this case, both Tashkent+ and Round-Robin mobilise a much higher amount of resources than Gargamel. Figure 11 shows that, at the end of the first phase, Tashkent+ needs 1500 concurrent workers, whereas Gargamel needs less than 100.

This can be directly translated into monetary cost. Considering that Amazon EC2 advertises a CPU cost of approximately 1 euro/hour [START_REF] Inc | Amazon Elastic Compute Cloud (Amazon EC2)[END_REF], Figure 12 plots the cost of the three systems, varying the incoming rate, with both the

VI. RELATED WORK

Database replication potentially improves performance and availability, by allowing several transactions to proceed in parallel at different replicas. This works well for readonly transactions, but remains challenging in the presence of updates. Concurrency control is an expensive mechanism; it is also wasteful to execute conflicting transactions concurrently, since one of them must abort and restart. This well-known issue prevents DBMSes from making effective use of modern low-cost concurrent architectures such as multicores, clusters, grids and clouds. For instance, Salomie et al. report that the performance of both PostgreSQL and MySQL degrade on multicore architectures [START_REF] Salomie | Database engines on multicores, why parallelize when you can distribute?[END_REF]. As we have shown in this paper, this is also the case of the state-of-theart Tashkent+ system [START_REF] Elnikety | Tashkent+: memory-aware load balancing and update filtering in replicated databases[END_REF]: as it aims to make full use of all available parallelism, it suffers high abort rates on standard benchmarks.

One approach to circumvent this problem is to parallelise only read-only transactions, and to centralise all update transactions at a single replica. This is the case of both Ganymed [START_REF] Plattner | Ganymed: Scalable replication for transactional web applications[END_REF] (in a cluster) and Multimed [START_REF] Salomie | Database engines on multicores, why parallelize when you can distribute?[END_REF] (on a multicore machine). However, this conservative approach results in poor response times, as it ignores that non-conflicting transactions can be parallelised, even if they perform updates. Our simulations, hereafter, confirm this intuition. In contrast, Gargamel is capable of parallelising at least some update transactions. Therefore, it does not need a master replica, which constitutes a scalability bottleneck.

An alternative is to give up on strong transactional consistency [START_REF] Vogels | Eventually consistent[END_REF]. Thus, systems such as Facebook, Zynga and Twitter make heavy use of memcached, a key-value store with weak consistency guarantees [START_REF] Petrovic | Using memcached for data distribution in industrial environment[END_REF]. This approach is promising for some particular classes of application, but is bug-prone and difficult to get right for application developers, and therefore not considered in this paper.

H-Store [START_REF] Stonebraker | The end of an architectural era (it's time for a complete rewrite)[END_REF] is a DBMS designed and optimized for OLTP applications. It requires the complete workload to be specified in advance as statically defined stored procedures. This advance knowledge allows H-Store to partition and to parallelise the load between different single-threaded replicas operating in a share-nothing environment. Under this assumption, H-Store improves the performance by orders of magnitude compared with other commercial databases. Gargamel also parallelises the load between single-threaded replicas, but does so above an unmodified DBMS. Furthermore, Gargamel requires only approximate advance knowledge, encapsulated in the conflict classifier, and does not require the whole workload to be specified in advance.

The system of Pacitti et al. [START_REF] Pacitti | Preventive multimaster replication in a cluster of autonomous databases[END_REF] is a lazy multi-master replicated database system. It enforces a total order of transactions, by using reliable multicast and a timestampbased message ordering mechanism. Gargamel, in contrast, can execute non-conflicting update transactions in parallel; in addition, it minimises resource utilisation. Furthermore, since Gargamel ensures proper transaction ordering in the background, it scales well to a WAN environment, in contrast to the a priori total order approach of Pacitti et al.

Sarr et al. [START_REF] Sarr | DTR: Distributed transaction routing in a large scale network[END_REF] introduce a solution for transaction routing in a grid. Their system, like Gargamel, is conflict-aware. However, they check for conflicts only in order to propagate updates among replicas in a consistent way; they do not serialise conflicting transactions, as Gargamel does.

A conflict-aware scheduling system is proposed by Amza et al. [START_REF] Amza | Distributed versioning: Consistent replication for scaling back-end databases of dynamic content web sites[END_REF], [START_REF]Conflict-aware scheduling for dynamic content applications[END_REF]. Their system ensures 1-copy-SI by executing all update transactions in all replicas in a total order. Gargamel parallelizes non-conflicting write transactions and transmits the write-sets off the critical path. Moreover Gargamel executes a given transaction only once, at a single replica, which ensures that replicas do not diverge in the presence of non-determinism.

Middle-R [START_REF] Martinez | MIDDLE-R: Consistent database replication at the middleware level[END_REF], similarly to Gargamel, serialises conflicting transactions at one replica classifying them before execution. Middle-R associates each (update) transaction to one or more conflict class. Each conflict class has a master copy. Transactions are executed at the master copy of the conflict they belong. After execution the write-set is diffused to other replicas. The key difference between Gargamel and Middle-R is that Middle-R statically associates replicas to transaction class and does not avoid aborts. At the opposite Gargamel partitions the database dynamically according to the transaction mix and never aborts transactions.

VII. CONCLUSION AND FUTURE WORK

We described Gargamel, a middleware that maximizes parallelism in distributed databases while limiting the amount of wasted work. Instead of parallelising conflicting transactions, like previous systems, Gargamel uses a transaction classifier to pre-serialise possibly conflicting transactions. This improves the average response time (lowers the penalty ratio) and dramatically decreases resource utilisation with respect to competing approaches.

Since Gargamel runs at the load balancer, it does not impose any changes to existing DBMS engines.

We show, by simulation, that under high incoming rates and with a good classifier, Gargamel performs considerably better than both Tashkent+ and Round-Robin: it provides higher throughput, lower response time, and consumes fewer resources.

To provide PSI we consider only write-write conflicts. We plan to extend Gargamel to provide higher isolation levels (e.g. one-copy serializability). We expect that a higher isolation level will decrease even more the resource utilisation, with respect to certification-based approaches.

In future work, we plan to implement a distributed version of Gargamel to support the co-existence of multiple concurrent classifying load balancers, one for each site. Load balancers synchronisation can be done optimistically to maximize performance. In the distributed version of Gargamel, we plan to study a scheduling policy that favors performance over resource utilisation, executing the prefix of split chains at all sites in parallel. If a site is less loaded or has faster computers than the others, it will send its write-set early, allowing the slower site(s) to catch up.

Figure 1 .

 1 Figure 1. System architecture

  (d) Merge and Split. (e) General chains.

Figure 2 .

 2 Figure 2. Example scheduling scenario.

t 2

 2 to a new queue (Figure2(a)).

Figure 3 .

 3 Figure 3. Throughput (TPC-C, 10 trans/s).

Figure 4 .

 4 Figure 4. Throughput (TPC-C, 200 trans/s).

Figure 5 .

 5 Figure 5. Maximal throughput (TPC-C).

Figure 6 .

 6 Figure 6. Penalty ratio (TPC-C).

Figure 7 .

 7 Figure 7. Cumulative distribution function of penalty ratio (TPC-C).

Figure 8 .Figure 9 .Figure 10 .Figure 11 .

 891011 Figure 8. Speedup for TPC-C

Figure 12 .

 12 Figure 12. Cost comparison