A fast implementation of Parallel Snapshot Isolation

A FAST IMPLEMENTATION OF PARALLEL SNAPSHOT ISOLATION UNA IMPLEMENTACIÓN RÁPIDA DE PARALLEL SNAPSHOT ISOLATION Borja Arnau de Régil Basáñez Trabajo de Fin de Grado del Grado en Ingeniería Informática Facultad de Informática, Universidad Complutense de Madrid Junio 2020 Director: Maria Victoria López López Co-director: Alexey Gotsman Contents Acknowledgements iv Abstract v Resumen vi 1. Introduction 1 1.1. Motivation..................................... 1 1.2. Goals........................................ 2 1.3. WorkPlan..................................... 3 1.4. DocumentStructure ............................... 3 1.5. Sources and Repositories . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 4 1.6. RelatedProgramCourses ............................ 4 2. Preliminaries 5 2.1. Notation...................................... 5 2.1.1. Objects and Replication . . . . . . . . . . . . . . . . . . . . . . . . . 5 2.1.2. Transactions................................ 5 2.1.3. Histories.................................. 6 2.2. ConsistencyModels................................ 6 2.2.1. ReadCommitted(RC).......................... 7 2.2.2. Serialisability (SER) . . . . . . . . . . . . . . . . . . . . . . . . . . . 8 2.2.3. Snapshot Isolation (SI) . . . . . . . . . . . . . . . . . . . . . . . . . . 9 2.2.4. Parallel Snapshot Isolation (PSI) . . . . . . . . . . . . . . . . . . . . 10 2.2.5. Non-Monotonic Snapshot Isolation (NMSI) . . . . . . . . . . . . . . . 11 2.2.6. Anomaly Comparison . . . . . . . . . . . . . . . . . . . . . . . . . . . 11 3. The fastPSI protocol 12 3.1. ConsistencyGuarantees ............................. 12 3.2. Overview and System Model . . . . . . . . . . . . . . . . . . . . . . . . . . . 13 3.3. Serverdatastructures .............................. 14 3.4. ProtocolDescription ............................... 16 3.4.1. Transaction Execution . . . . . . . . . . . . . . . . . . . . . . . . . . 16 3.4.2. Transaction Termination . . . . . . . . . . . . . . . . . . . . . . . . . 20 3.5. Consistency Tradeoffs and Read Aborts . . . . . . . . . . . . . . . . . . . . . 23 ii 4. Implementation and Evaluation 26 4.1. Implementation.................................. 26 4.2. Evaluation..................................... 27 4.2.1. Performance & Scalability Limits . . . . . . . . . . . . . . . . . . . . 28 4.2.2. AbortRatio................................ 31 5. Related Work 35 6. Conclusions and Future Work 37 6.1. Conclusions .................................... 37 6.2. FutureWork.................................... 37 A. Serialisable and Read Committed Protocols 39 A.1.Serialisability ................................... 39 A.2.ReadCommitted ................................. 44 Bibliography 50 iii Acknowledgements To my advisors Maria Victoria López López at Universidad Complutense de Madrid, and Alexey Gotsman and Manuel Bravo at the IMDEA Software Institute, for their guidance and support, and for giving me the opportunity to work along them. I also thank Christopher Meiklejohn, who allowed me to work with him, and gave me the opportunity to discover the IMDEA Software Institute. To my colleagues at IMDEA, thank you for giving me advice, and for offering a helping hand. Finally, to my girlfriend Paula, and my family, for their love, patience and support. iv Abstract Most distributed database systems offer weak consistency models in order to avoid the performance penalty of coordinating replicas. Ideally, distributed databases would offer strong consistency models, like serialisability, since they make it easy to verify application invariants, and free programmers from worrying about concurrency. However, implementing and scaling systems with strong consistency is difficult, since it usually requires global communication. Weak models, while easier to scale, impose on the programmers the need to reason about possible anomalies, and the need to implement conflict resolution mechanisms in application code. Recently proposed consistency models, like Parallel Snapshot Isolation (PSI) and NonMonotonic Snapshot Isolation (NMSI), represent the strongest models that still allow to build scalable systems without global communication. They allow comparable performance to previous, weaker models, as well as similar abort rates. However, both models still provide weaker guarantees than serialisability, and may prove difficult to use in applications. This work shows an approach to bridge the gap between PSI, NMSI and strong consistency models like serialisability. It introduces and implements fastPSI, a consistency protocol that allows the user to selectively enforce serialisability for certain executions, while retaining the scalability properties of weaker consistency models like PSI and NMSI. In addition, it features a comprehensive evaluation of fastPSI in comparison with other consistency protocols, both weak and strong, showing that fastPSI offers better performance than serialisability, while retaining the scalability of weaker protocols. Keywords Consistency models, Transactions, Parallel Snapshot Isolation, Non-Monotonic Snapshot Isolation, Concurrency control. v Resumen La mayoría de las bases de datos distribuidas ofrecen modelos de consistencia débil, con la finalidad de evitar la penalización de rendimiento que supone la coordinación de las distintas réplicas. Idealmente, las bases de datos distribuidas ofrecerían modelos de consistencia fuerte, como serialisability, ya que facilitan la verificación de los invariantes de las aplicaciones, y permiten que los programadores no deban preocuparse sobre posibles problemas de concurrencia. Sin embargo, implementar sistemas escalables que con modelos de consistencia fuerte no es fácil, pues requieren el uso de comunicación global. Sin embargo, aunque los modelos de consistencia más débiles permiten sistemas más escalables, imponen en los programadores la necesidad de razonar sobre posibles anomalías, así como implementar mecanismos de resolución de conflictos en el código de las aplicaciones. Dos modelos de consistencia propuestos recientemente, Parallel Snapshot Isolation (PSI) y Non-Monotonic Snapshot Isolation (NMSI), representan los modelos más fuertes que permiten implementaciones escalables sin necesidad de comunicación global. Permiten, a su vez, implementar sistemas con rendimientos similares a aquellos con modelos más débiles, a la vez que mantienen tasas de cancelación de transacciones similares. Aun así, ambos modelos no logran ofrecer las mismas garantías que serialisability, por lo que pueden ser difíciles de usar desde el punto de vista de las aplicaciones. Este trabajo presenta una propuesta que busca acortar la distancia entre modelos como PSI y NMSI y modelos fuertes como serialisability. Con esa finalidad, este trabajo presenta fastPSI, un protocolo de consistencia que permite al usuario ejecutar de manera selectiva transacciones serializables, reteniendo a su vez las propiedades de escalabilidad propias de modelos de consistencia débiles como PSI o NMSI. Además, este trabajo cuenta con una evaluación exhaustiva de fastPSI, comparándolo con otros protocolos de consistencia, tanto fuertes como débiles. Se muestra así que fastPSI logra un rendimiento mayor que serialisability sin por ello renunciar a la escalabilidad de protocolos más débiles. Palabras clave Modelos de Consistencia, Transacciones, Parallel Snapshot Isolation, Non-Monotonic Snapshot Isolation, Control de Concurrencia. vi Chapter 1 Introduction 1.1. Motivation Modern cloud applications are characterised by being globally available, and users expect to use the services provided by these applications with low latency, and in a reliable manner. To satisfy these requirements, programmers usually resort to distributed databases and storage, that allow to partition application data and place it geographically close to the users that need it. For example, a social media site would place data related to European users on data centers located in the same region, and the same for users in the United States. Under this design, however, those users requesting data from another region would suffer from high latency, as requests travel across different geographical regions. To this end, these data partitions are also replicated across different geographical regions, to ensure both low latency for all kinds of user requests, and fault tolerance, which allows applications to ensure a smooth operation even if an entire region goes offline. These approaches, however, add significant complexity to the design and implementation of applications and the underlying databases. The presence of multiple replicas raises the question of how to keep them consistent, that is, reflecting an unified version of the data they contain. Traditional mechanisms to deal with database consistency prove harder to implement in efficient ways in distributed databases. For example, transactions should satisfy a set of desirable properties, commonly known as ACID: Atomicity, Consistency, Isolation, and Durability. These properties allow programmers to reason about concurrency as a set of isolated, atomic operations, but in geo-distributed scenarios it requires the coordination of multiple replicas in order to apply their updates. The usual approach to bridge these problems involves relaxing the consistency guarantees that databases offer programmers [40]. Indeed, the CAP Theorem [16,23] proves it is impossible to build applications that continue operating in the presence of network partitions without sacrificing consistency guarantees. However, the degree to which these guarantees can be relaxed offers a trade-off: on the one hand, weak consistency allows to build scalable applications without loss of availability, but proves difficult to reason about given that it allows non-serialisable behaviours called anomalies, and forces programmers to deal with 1 inconsistent data at the application level; on the other hand, strengthening consistency guarantees can reduce performance and hurt application availability, while being much easier to reason about. Until recently, systems that offered weak consistency guarantees did not provide transactions (e.g. Dynamo [20]). In the recent years, however, a large number of transactional consistency models have been proposed for large-scale databases [6,10,32,39]. Given the proliferation of different consistency models, it can be hard to choose which one is appropriate for a particular application, as it requires the programmer to think about the possible anomalies that can arise during an execution and about how they can interfere with application logic. Ideally, one would want to run all applications under strong consistency models, like serialisability [15], as programmers only need to check that application invariants hold as if transactions executed one after the other, without worrying about concurrency. Unfortunately, guaranteeing a serialisable execution in distributed databases is not possible without requiring global communication, which increases latency and limits availability [23]. This leaves the programmers with the responsibility of choosing an adequate consistency model for their applications. However, programmers often lack techniques to ensure that a given consistency model is safe to use for a particular application. One way to address this problem is to rely on the notion of application robustness [14,22]: an application is robust against a particular consistency model if it behaves in the same way whether using a database providing this model or serialisability. When an application is robust, the programmer can take advantage of the scalability properties of a weak consistency model without paying the price of anomalous behaviours. Previous work has focused on static analysis of applications [29,34], which let programmers know which parts of their applications are susceptible to anomalies. In these cases, programmers can selectively run transactions under serialisability: Fekete et al. [21,22] propose several techniques that allow transactions executing under snapshot isolation (SI) [13] to exhibit serialisable behaviours, effectively making them equivalent to transactions running under serialisability. Most recently, Bernardi and Gotsman [14] proposed a way to check the robustness of parallel snapshot isolation (PSI)1[39], which relaxes the consistency guarantees of snapshot isolation to allow more efficient implementations for distributed databases. PSI is also the strongest model that is weaker than SI [17], thus it is an obvious candidate to investigate its impact on the correctness of applications. Although there are several implementations that guarantee PSI [6,33,39], none of them were implemented with the focus on exploring the relation between PSI and application robustness. 1.2. Goals The goal of this work is to help programmers bridge the gap between weak and strong consistency protocols, without sacrificing application correctness. To that end, fastPSI is proposed, an implementation of Parallel Snapshot Isolation that allows to selectively enforce serialisability for transactions through careful grouping of database objects into entity 1Also known as non-monotonic snapshot isolation [6]. This is discussed in §2.2.5. 2 groups [11]: transactions accessing objects in the same group execute as if they were running under a system guaranteeing SI (instead of the weaker PSI). Following the techniques proposed by Fekete et al. [22], these transactions can be further constrained so that they execute as if running under serialisability. As such, the contributions of this work are: A hybrid consistency protocol that allows mixing Snapshot Isolation with Parallel Snapshot Isolation, by relying on entity groups. This protocol allows programmers to combine the scalability of Parallel Snapshot Isolation with the familiarity and intuitiveness of well-known consistency models like Snapshot Isolation and serialisability. A comprehensive evaluation of the proposed protocol, and a comparison against alternative implementations of both weak and strong consistency models. An exposition of the drawbacks and trade-offs of the protocol, and a discussion of how their impact can be minimised. 1.3. Work Plan The work carried out for this project was divided in the following phases: Explore previous contributions. It’s necessary to get familiarised with existing terminology, as well as previous work and implementations. Delimit project scope. After having the necessary knowledge to carry out the work, the limits and specific contributions of the work are established. In addition, the hypotheses that the final implementation should validate are proposed. Implementation phase. After setting clear objectives and milestones, the bulk of the implementation is done. Throughout this phase, testing and validation of the software is done, both with unit tests and with model checking techniques. Benchmark and validation. With the implementation complete, representative benchmarks are designed, as well as scenarios to validate the performance of the implementation. The results are used validate previous hypotheses, and also framed in the context of the existing literature. 1.4. Document Structure The rest of this document is structured as follows. Chapter 2provides an overview of previous work and the state of the art with regards to relevant consistency models, as well as basic notions that will be used throughout this document. Chapter 3introduces fastPSI, 3 2.2.4. Parallel Snapshot Isolation (PSI) Parallel Snapshot Isolation (PSI), proposed by Sovran et al. [39], is a consistency model aimed at solving the scalability limits of classical Snapshot Isolation in geo-replicated systems. While concurrent conflicting transactions are not allowed, PSI allows non-conflicting transactions to exhibit a relative commit order that varies between replicas. This means that PSI can propagate transactions to replicas in causal order, sidestepping another scalability limit of SI. However, allowing different commit orders for non-conflicting transactions at different replicas (or sites) makes PSI susceptible to the Long Fork anomaly [39]. Consider the history depicted in Figure 2.3. If transactions T4and T5execute in different replicas, they are allowed to observe different commit orders for T2and T3. r4(y1) r2(x1).w2(x2).c2 r3(y1).w3(y3).c3 r5(y3).c5 r4(x2).c4 r5(x1) T2 T3 T4 T5 w1(x1).w1(y1).c1 T1 Figure 2.3:Example of a history showing the Long Fork anomaly. Transaction T4observes T1T2T3while T5observes T1T3T2. Definition 2.6 (Long Fork).ALong Fork occurs whenever transactions are able to observe different commit orders of previous non-conflicting update transactions. Like SI, PSI exhibits base freshness [8]: a transaction Tis limited to read only versions of objects written by transactions that committed before Tstarted. In the geo-replicated scenarios that PSI is meant to address, this limitation leads to a high number of stale data reads. Consider the example of two sites s1and s2separated by a high latency link: if a transaction starts in s1and subsequently tries to read an object located at s2, it might be the case that the version it is forced to read due to base freshness has already been overwritten by other transactions. Saeida Ardekani et al. [38; Theorem 4] prove that base freshness requires replicas that do not replicate data accessed by a transaction Tto coordinate in order to commit T, which results in lower system performance and limits scalability. Indeed, the original implementation advanced by Sovran et al. communicates with all the replicas in the system [39]. 10 2.2.5. Non-Monotonic Snapshot Isolation (NMSI) The Non-Monotonic Snapshot Isolation (NMSI) consistency model also alleviates the total commit order scalability problem in classical Snapshot Isolation. NMSI was proposed by Saeida Ardekani et al. [6] as an improvement over the previously described Parallel Snapshot Isolation model, by exhibiting forward freshness: a transaction Tis allowed to read versions written by transactions that committed after Tstarted, as long as those reads form a causally consistent snapshot. A transaction Tobserves a causally consistent snapshot if all the versions read by Tare written by its direct causal dependencies, i.e. if a transaction Tiperforms ri(xj)such that it depends on Tj, then there’s no such wk(xk)such that Tj/Tk/Ti. In spite of this, the set of possible anomalies produced by both Parallel Snapshot Isolation and Non-Monotonic Snapshot Isolation are the same, as can be seen in Figure 2.4. In addition, both models can be proved to be equivalent, as shown by A. Cerone (2016, personal communication with the author), with the only difference being the choice of the concurrency control algorithm. 2.2.6. Anomaly Comparison Figure 2.4 summarises the consistency models reviewed, together with the anomalies that they allow. Anomalies Consistency Models SER SI PSI NMSI RC Dirty Write x x x x x Dirty Read x x x x x Non-Repeatable Read x x x x X Lost Update x x x x X Write Skew x X X X X Long Fork x x X X X Figure 2.4:Anomaly Comparison of Consistency Models (x:disallowed, X:allowed). Adapted from Saeida Ardekani et al. [6]. 11 Chapter 3 The fastPSI protocol This chapter describes fastPSI, a transactional protocol that implements Parallel Snapshot Isolation and allows stronger consistency guarantees for transactions accessing objects inside entity groups. The chapter begins with an overview of what entity groups are, and by explaining the consistency guarantees of fastPSI for transactions executing both within and across different groups. It follows with a summary of how the different participants of the protocol interact with each other, and with a description of the different data structures involved. Next, it shows how the protocol is structured by going over the execution of a transaction. The chapter concludes with a discussion of the possible drawbacks of the design. 3.1. Consistency Guarantees The fastPSI protocol considers a system in which objects are aggregated in entity groups [11]. An entity group σis defined as a proper partition of objects Obj. This means that two properties hold: (i) ∀σ, σ0=⇒σ∩σ0=∅, and (ii) ∀x∈Obj.∃σ. x ∈σ. The first property says that the sets of objects covered by different entity groups are disjoint, while the second property states that any object that exists in the system is part of an entity group. The goal of fastPSI is as follows: transactions that only access objects inside a single entity group should satisfy Snapshot Isolation (SI), while transactions that access objects across entity groups should satisfy Parallel Snapshot Isolation (PSI). Intuitively, if one has a single entity group that encompasses every object, any execution of fastPSI is equivalent to an execution under Snapshot Isolation, thereby precluding the Long Fork anomaly. Conversely, if one has an entity group per object in the system, then any execution of fastPSI is equivalent to an execution under Parallel Snapshot Isolation. The decision of which objects to place into different entity groups is left to the programmer, who should take application requirements into account. Recall from Section 2.2.3 that transactions executing under SI are only partially ordered, in contrast with serialisability, where they are totally ordered. However, the requirement for 12 transactions to take monotonic start and commit timestamps induces a total commit order for transactions, even for those that are not in conflict with each other. In the presence of different entity groups, this requirement would require transactions to communicate with every group in order to determine a monotonic timestamp. In fastPSI, this requirement is relaxed so that transactions have multiple, independent timestamps, one per entity group. In order to guarantee Parallel Snapshot Isolation across entity groups, the protocol incorporates the notion of forward freshness [6], which allows a transaction Tto read versions of objects written by transactions that committed after Tstarts. The fastPSI protocol accomplishes this by making a transaction Tfix its start timestamp at a particular entity group only when Treads an object from that group. This also allows a transaction to acquire start timestamps only at the groups it reads from, thus avoiding coordination with other groups. The fastPSI protocol leverages both approaches to offer its consistency guarantees: a transaction is able to read from versions written by latter transactions as long as those reads form a causally consistent snapshot. When a transaction Tperforms its first read operation in an entity group, it fixes a snapshot that includes all the transactions that committed before T’s read occurred. As Tperforms further read operations on other entity groups, the snapshots that Tfixes are restricted to versions written by transactions that are not causally dependent on the transactions that Talready included in its previous snapshots. 3.2. Overview and System Model The protocol consists of three components: client processes that provide the system interface for managing transactions, server processes that handle the individual operations of transactions, and entity groups—managed by a server—which store individual data objects. Given that entity groups properly divide the range of objects into disjoint partitions, for the remainder of this chapter the term partition is used as a shorthand for entity group. Both client and server processes are considered reliable and connected by reliable channels1. Processes communicate with each other using an asynchronous message-passing system. Server processes are denoted as a set S={s1, . . . , sN}, and clients as C={c1, . . . , cM}. Data objects are denoted by a set Obj, split into Npartitions, each stored by a server process. In addition, partition(x)represents the index of the partition the object xbelongs to, such that it is managed by server spartition(x). For simplicity, it is assumed that server processes only manage a single partition. Clients provide the transactional interface of the protocol through the start,read, write and commit operations. Transactions are interactive, i.e., when a transaction starts, the client does not know which operations it will perform in advance. In fastPSI, the start and write operations are local to the client. Clients issue read operations to servers, which return the values and metadata associated with the objects the client requested. Clients 1Fault-tolerance concerns are orthogonal to the problem addressed, although several approaches are discussed in Chapter 5. 13 handle write operations locally by storing the updates in a buffer, called the write-set of a transaction. At commit time, the client acts a coordinator of a two-phase commit protocol (2PC) [15], issuing prepare and decide operations to all the participating servers. The written values buffered locally are transmitted to the servers at prepare time, together with the accumulated metadata for the objects that the client read. The decision to commit or abort a transaction is taken based on this metadata. Servers handle three operations issued by clients: read,prepare and decide. When a server handles read operations, it forwards the request to the partition responsible for the object being requested. When receiving a prepare request for a certain transaction, the server checks for conflicts with other transactions waiting to be decided, which are stored in a commit queue. If the transaction contained in the client request does not conflict, it is added to the queue, and the server replies to the client with a commit vote. If, on the other hand, the transaction is found to be conflicting, an abort vote is sent instead. Partitions are responsible for fulfilling read requests on behalf of servers, and for maintaining causally consistent snapshots on behalf of the clients. Partitions store multiple versions of an object in accordance with a multi-version concurrency control protocol. When executing a read request for an object x, a partition finds and returns the most recent causally consistent version of x, along with some metadata of the chosen version. Each partition stores multiple versions of an object represented by a tuple hval, vidi, where val is the value of a given version, and vid is a logical identifier for the transaction that committed this version. This logical identifier is represented using version vectors [35]. Such a vector consists of Nentries, one for each partition, storing a non-negative integer. Each entry vid[i]in the vector can also be represented by a pair (si, k), where kis the actual value of the i-th entry of the vector. These pairs are also called dots [12]. Version vectors are compared according to the following relation, showing when one vector covers more dots than another: V1vV2⇐⇒ ∀i. V1[i]≤V2[i]. In addition, there exists a join operation on vectors, taking their entry-wise maximum, which will be denoted by max from now on. The set of all version vectors is denoted by VerVector, and the vector with all entries set to 0by ~ 0. 3.3. Server data structures Each server simaintains five main data structures, summarised in Figure 3.1. This section now follows with a more detailed explanation of each of them. LastPrep is a counter of the number of update transactions that initiated their commit phase at a given server. When a transaction commits at a server si, it gets assigned the value of the counter as a sequence number k, which induces the commit order of transactions at si. A transaction computes its commit vector Vcfrom these sequence numbers, such that the vector represents the set of dots {(si, k)|k≤Vc[i]}which identify the writes by previous transactions whose sequence number at siis no higher than Vc[i]. VersionLog is a mapping of objects to a list of versions, called the database. As noted before, each version is a tuple hval,Vcommisuch that val is a value and Vcomm is the 14 Data Structures at a server si LastPrep Integer The number of update transactions that tried to commit at the server. VersionLog Map[Object, Set[hValue val,VerVector Vcommi]] Database: a mapping from objects to lists of pairs of a value and the commit vector of the transaction that wrote it. The lists are ordered by the i-th component of the commit vectors. CommitLog Sequence[hTx T, VerVector Vaggri]Log of update transactions Tcommitted at the server, ordered by Vaggr[i]. Here Vaggr is the aggregate vector of T: the join of the commit vectors of all transactions up to Tin CommitLog. Vtotal VerVector The join of the commit vectors of all transactions in CommitLog. CommitQueue Sequence[hTx,pending,WriteSeti ∪ hTx,decided,WriteSet,VerVectori] Queue containing information about update transactions trying to commit at the server. Figure 3.1:Data structures used by servers in the protocol. The orders of entries in CommitLog,VersionLog and CommitQueue are consistent with the commit order of the associated transactions. Components of various tuples are selected using the names given in the figure. commit vector of the transaction that wrote val. The VersionLog at siis ordered by the i-th component of the commit vector of each version, which follows the commit order of transactions at the server. The most recent entry in the list for the object xis denoted by VersionLog[x].last. CommitLog is an ordered list that maintains a tuple hT, Vaggrifor each update transaction that committed at a server, such that Tis the identifier of a committed transaction and Vaggr is an aggregate vector. The aggregate vector represents the join of the commit vectors of all the transactions up to Tin CommitLog. Entries in the log at siare totally ordered according to the i-th entry of their aggregate vectors, Vaggr[i], which also follows the commit order of transactions at the server. These aggregate vectors are stored for efficiency, and are used to compute the snapshot of a transaction at the server. The aggregate vector of the last committed transaction is stored in Vtotal. Initially, the CommitLog contains a single placeholder entry h_,~ 0i. Finally, CommitQueue is an ordered queue of transactions trying to commit updates at the server. The queue has entries of two types. An entry hT,pending,WSimeans that Tis successfully prepared to commit at si, but the final decision on it is not yet known; WS is the write-set of the transaction, containing object-value pairs. An entry hT,decided,WS, V i in the queue means that Thas been decided to commit with a commit vector V, but its writes have not yet been added to VersionLog. The order of transactions in CommitQueue follows the commit order at the server. 15 3.4. Protocol Description The protocol is now described in detail by following the execution of a transaction. This section begins by describing how a transaction is initialised, along with the mechanism to build a causally consistent snapshot. Later, it shows the mechanism to validate and commit a transaction. 3.4.1. Transaction Execution The fastPSI protocol uses optimistic concurrency control, that is, the execution of a transaction is speculative. Clients read objects from servers and buffer writes locally. At the end of the execution, the decision whether to commit or abort a transaction is taken based on the existence of conflicts with concurrently executing transactions. Clients executing a transaction maintain a transaction context including several pieces of data, summarised in Figure 3.2 and explained below. In the following, the steps taken by both clients and servers to execute Trefer to the algorithms in Figures 3.3 and 3.4. Context for a transaction Tat a client ci T.WS WriteSet Write-set of T. T.HasRead Vector[Bool]Mapping showing whether Thas read a given partition. T.Vsnap VerVector Snapshot vector: determines snapshots fixed at partitions T has read from and possible causal dependencies at all other partitions. T.Vdep VerVector Dependency vector, representing all causal dependencies developed by Tduring its execution. Figure 3.2:Data structures used in the transaction context, kept by the clients in the protocol. In the table, WriteSet =Set[hObject,Valuei] When a client starts a transaction T, it first initialises its context (line 1). When a transaction Twrites a value vto an object x(line 3), the client buffers this write in T’s write-set,T.WS, while discarding any previously written value of x. 1function start() 2return new Tx(WS =∅,HasRead =~ ⊥,Vsnap =~ 0,Vdep =~ 0); 3function write(T, x, v) 4T.WS ←(T.WS \ {hx, _i})∪ {hx, vi}; Figure 3.3:Initialisation of a transaction and update of an object xat client ci. When the transaction Tissues a read operation on an object x(line 5), the client first checks T.WS (line 6): if Thas already written to x, the value stored in the write-set is 16 returned. Otherwise, and assuming that j=partition(x), the client sends a READREQUEST message to the server sjto fetch the value of the object (line 9). When the transaction Treads an object from a partition jfor the first time, the server sjfixes a snapshot of versions from which it will serve all future reads by T. This snapshot is defined by an integer k: it will include the versions written by all the transactions that committed at the server with a sequence number up to k. The client keeps this information in the transaction context, by storing kin the j-th entry of a snapshot vector T.Vsnap, and by marking the current partition as read in T.HasRead, a boolean mapping its j-th entry to >if Tread an object from sj, and ⊥otherwise. If T.HasRead[j] = >, then the vector T.Vsnap is equal to the join of the commit vectors of all transactions committed at sjwith a sequence number no higher than Vsnap[j]. 5function read(T, x) 6if hx, vi ∈ T.WS then 7return v; 8j←partition(x); 9send READREQUEST(x, T.Vsnap, T.HasRead)to sj; 10 wait receive READRETURN(m)from sj; 11 if m=abort then 12 throw abort; 13 else if m=hv, Vdep,Vaggrithen 14 T.HasRead[j]← >; 15 T.Vdep ←max(T.Vdep,Vdep); 16 T.Vsnap ←max(T.Vsnap,Vaggr); 17 return v; 18 when received READREQUEST(x, Vsnap,HasRead)from cj 19 if HasRead[i]then 20 V←Vsnap; 21 else 22 wait until Vtotal[i]≥Vsnap[i]; 23 r←max{r∈CommitLog | ∀j. HasRead[j] =⇒(r.Vaggr[j]≤Vsnap[j])}; 24 if r.Vaggr[i]<Vsnap[i]then 25 send READRETURN(abort)to cj; 26 return; 27 V←r.Vaggr; 28 ver = max{ver ∈VersionLog |ver.Vcomm[i]≤V[i]}; 29 send READRETURN(ver.val,ver.Vcomm, V )to cj; Figure 3.4:Local and remote read of object x. Thus, the entries in the snapshot vector for partitions that Thas not yet read from 17 T1 T2 ij k (a) T3 T4 T1 T2 ij k (b) T3 T4 Figure 3.5:Illustrations of the snapshot computation. Vertical lines depict the commit order at the corresponding partitions (top to bottom) and horizontal lines the cut-offs of various snapshots. Arrows between partitions depict causal dependencies. delimit all the possible causal dependencies Tmay develop at these partitions. Both T.Vsnap and T.HasRead are supplied by the client when issuing a read operation on object x, by using them as parameters to the READREQUEST message sent to a server si. When the server receives this message (line 18), it first checks, using the HasRead mapping, if the transaction has read from it before (line 19). In this case, the snapshot is determined by Vsnap[i], and the server returns the latest version ver of the object xwritten by a transaction in the snapshot, i.e., with a sequence number no higher than Vsnap[i](line 28). This version is determined by examining the i-th entry of the commit vectors in VersionLog. The server then replies to the client with a READRETURN message containing the value of the chosen version and its associated version vector, as well as the unmodified snapshot vector provided by the client: since the server used a previously fixed snapshot, no updates to the vector are required. In the case when the client reads from the server sifor the first time (line 21), it is necessary to fix the snapshot for the transaction Tat this server. Choosing a suitable snapshot is complicated by the fact that transactions are allowed to be interactive—that is, it is not know in advance which objects a transaction will read in the future. The snapshot is hence fixed in such a way that any later read from this snapshot will be causally consistent with any other read from the snapshots that Thas already fixed, as specified by HasRead and Vsnap. To ensure this, the selected snapshot has to satisfy two requirements, depicted in Figure 3.5. On the one hand, the snapshot cannot be too fresh. For example, suppose a transaction T1that wrote to partition jis excluded from the snapshot chosen by Tat j. Then, the snapshot chosen by Tat partition icannot contain T1, nor any other transaction that causally depends on it, like T2(Figure 3.5a). If the snapshot chosen by Tincluded T2, it would be able to read some of T2’s writes at i, thereby forcing Tto read the writes by T2’s causal dependencies, including T1; but Tcannot see these writes, because it excluded 18 them from the snapshot at partition j. Thus, when building the snapshot, the server needs to take into account the snapshots taken by Tat the partitions it already read; the server selects the longest prefix of CommitLog transactions, such that their writes—and the ones by their causal dependencies—are included in T’s previous snapshots. This prefix is denoted by r(line 23), and is computed using the Vaggr vector included in each of the entries of CommitLog, summarising the causal dependencies of all transactions up to a given record in the log. Thus, the snapshot at siincludes all transactions with sequence numbers up to r.Vaggr[i]. On the other hand, the snapshot selected cannot be too stale. Continuing with the previous example, if a transaction T3is included in the snapshot taken by Tat some partition k, then the snapshot of Tat partition ihas to include the writes by T3and its causal dependencies, e.g., the transaction T4in Figure 3.5a. The snapshot vector T.Vsnap summarises the updates of the transactions (and of its causal dependencies) included in the snapshots fixed by T. Thus, after determining the appropriate snapshot in line 23, the server sichecks that this snapshot covers transactions with sequence numbers at iup to Vsnap[i] (line 24). To maximise the chances of passing this check, before allowing a transaction Tto proceed with a read, the server siwaits until the writes from the prefix up to Vsnap[i]have been incorporated into its state (line 22). It may be the case that it’s impossible to satisfy both of the above requirements when selecting a snapshot; e.g., in the situation illustrated in figure 3.5b. Assuming that the transaction Thas fixed a valid snapshot at jand k, it is impossible to build a consistent snapshot at partition i; given that Tincluded T3at partition k, it is forced to read T4’s writes at i. At the same time, because it excluded T1from the snapshot at j,Tcan’t read the writes by T2at i. In this case, without the second requirement, the server siwould build a snapshot excluding both T2and T4, violating T’s causal dependency on T3. In this case the server sends to the client a READRETURN message with a special value abort (line 25), which will cause the client to abort the transaction (line 12). Once the server fixes a new snapshot, it selects the most recent version of the object x, defined by e.Vaggr[i](line 28), to return to the client. The server replies with a READRETURN message, carrying a triple of the value of the object, its associated version vector, and the aggregate vector for e.Vaggr, summarising the causal dependencies of all the transactions in the snapshot. When the client receives the message (line 13), it first sets the j-th entry of T.HasRead to >, to indicate that Thas read an object at partition j, and joins the returned aggregate vector to T.Vsnap. The client also joins the commit vector associated with the version read to a dependency vector T.Vdep, which represents all causal dependencies developed by Tduring its execution. This ensures that, upon reading a version of object x, Twill causally depend on the transaction T0that wrote that version of x, along with the causal dependencies of T0. Consider the example depicted in Figure 3.6a, which shows a complete execution of the protocol. Client c1issues a pair of read requests to servers s1and s2. In turn, these servers reply with the value of the object requested, along with its version vector and the new aggregate vector for the transaction, denoted by Vdep and Vaggr at the bottom. Since this is the first read request issued on behalf of this transaction, the server s1replies with its 19 Chapter 4 Implementation and Evaluation This chapter offers an evaluation that attempts to explore the overhead of fastPSI’s strong consistency model compared to the weak consistency of Read Committed. The evaluation also shows how fastPSI is able to outperform a protocol implementing the stronger serialisability consistency model. Finally, it evaluates the scalability of fastPSI as more servers and partitions are added to the system, and discusses some of the limitations of the protocol. 4.1. Implementation The implementation of fastPSI consists of a client-side library [1] and a server [3], the latter being written as a plug-in transactional protocol for Antidote [5], a reference platform for evaluating consistency protocols. Both the client library and the server are written in the Erlang programming language, with a total of 6K lines of code. The Antidote platform provides a key-value database, supports both in-memory and disk-based storage, and implements full replication. For simplicity, the implementation of fastPSI only supports inmemory storage, and lacks a replication mechanism. The client-side library communicates with the server using Google’s Protocol Buffers [2]. To enhance network efficiency, client messages are transmitted in periodic batches to the servers. To validate the results of the evaluation, two alternative protocols are also implemented, satisfying the Read Committed and serialisability consistency models, called naiveRC and naiveSER, respectively. Both are built on top of the original fastPSI implementation and are as efficient as possible. The pseudocode for both implementations can be found in Appendix A. As the implementation of fastPSI doesn’t target replicated scenarios, this document refrains from comparing against previous implementations of Parallel Snapshot Isolation. Since the protocols and implementations as described in the literature are influenced by the choice of replication mechanisms, a comprehensive evaluation of fastPSI against other implementations of PSI is deferred to future work, which could explore incorporating either partial or full replication to fastPSI. 26 Given that fastPSI requires the use of multiple versions per object, it becomes necessary to prevent an unbounded growth of the number of versions in the VersionLog database, and in the number of entries in the per-partition CommitLog. To that end, a simple garbage collection mechanism in the implementation ensures a fixed number of versions, and regularly prunes the oldest versions from the state of a partition. 4.2. Evaluation This section evaluates the performance of fastPSI using several workloads inspired by the Yahoo! Cloud Serving Benchmark (YCSB) [18], modified to generate transactional workloads [6,9]. The implementation of Read Committed is used as a baseline for comparison, in order to show the maximum possible performance. Figure 4.1 describes the workloads used. All experiments are run on a cluster consisting of machines running Debian 4.19.67-2 (Stretch) with 3.80 GHz to 4.70 GHz Intel Xeon processors with six cores, 32 GB of RAM, and one gigabit network port. The cluster is partitioned in up to three different sites, with four server machines and four client machines at each site. Thus, there is no shared memory between clients and servers, as if clients were acting as proxies in the same data centre as servers. Since all the machines are located in the same local network, the tc Linux command is used to artificially add latency between sites. In all benchmarks, the system is loaded with one million random keys and 256-byte values prior to receiving any operations from the clients. Partitions are distributed uniformly across servers, such that a server might be responsible for multiple partitions. Keys are mapped to partitions using consistent hashing [30], with clients being aware of the distribution of keys to partitions and server machines. Thus, clients can directly address the correct server and partition for a specific key. Each client machine spawns multiple concurrent threads that execute transactions and communicate with servers in a closed-loop fashion. When transactions read more than one object, clients perform those operations serially. For the experiments that involve more than one site, the latency between sites is of 10 ms. Key Selection Distribution Operations Read-Only Tran. Update Tran. B Uniform 4 Reads 3 Reads, 1 Update C Uniform 2 Reads 1 Read, 1 Update D Uniform 3 Reads 3 Reads, 1 Update E Uniform 3 Reads 3 Reads, 3 Updates Figure 4.1:Transactional YCSB Workload Types. 27 ●●● ● ●● ● ● ●● ● 90% Read−only Transactions 80% Read−only Transactions 70% Read−only Transactions 0 250 500 750 1,000 1,250 1,500 1,750 2,000 0 250 500 750 1,000 1,250 1,500 1,750 2,000 0 250 500 750 1,000 1,250 1,500 1,750 2,000 0 5 10 15 20 25 0 5 10 15 20 25 0 5 10 15 20 25 Throughput (Ktps) Termination Latency of Updt. txn (ms) Workload C on 3 sites ● naiveSER fastPSI naiveRC Figure 4.2:Comparison of throughput and termination latency of update transactions. 4.2.1. Performance & Scalability Limits Performance. The first experiment serves to investigate the overall performance profile of the implementations. The throughput and latency of the different protocols is measured and compared as the number of update transactions increases, while keeping the number of sites constant. For this experiment, the number of concurrent client threads varies such that the resources of the CPU never saturate. The aim is to explore the overall overhead of fastPSI in comparison with naiveRC, as well as the performance benefits it offers in comparison with naiveSER. Figure 4.2 shows the results of using Workload C and three sites, with 64 partitions uniformly distributed across sites. It measures the termination latency of update transactions (i.e., the amount of time spent on the validation of a transaction) as the ratio of read-only to update transaction ranges from 90%/10% to 70%/30% (left to right). Since the latency is measured in the client, it only reflects the amount of time spent on the prepare phase of the commit validation, as the client does not need to wait until the changes of a transaction are committed to the partition state. Transactions as executed by naiveRC need minimal synchronisation during its commit phase, and no synchronisation at all during read operations. This is reflected in its high performance, with the validation of update transactions as the only bottleneck in the system. Thus, as the proportion of update transactions increases, the impact on overall throughput is pronounced, dropping by as much as 20%. For both naiveSER and fastPSI, read operations from a transaction Tmust wait until the causal dependencies of Tcommit at a particular partition, bounded in the worst case by the maximum latency across sites. In addition, read operations accessing the same partition suffer from having to synchronise while fixing a snapshot, as the implementation of CommitLog is not thread-safe. These two shortcomings explain the overall low throughput in comparison with naiveRC. Nevertheless, both implementations exhibit stable performance as the proportion of update transactions increases. By comparing fastPSI with naiveSER, one can observe that the latter implementation is limited by its need to validate every transaction, in comparison with fastPSI, which only validates update transactions. In addition, the weaker consistency model offered by fastPSI 28 allows it to outperform naiveSER in all cases by approximately 150%, while showing similar latencies. Scalability. The next experiment explores the overall scalability of fastPSI, by examining how the maximum performance of each protocol changes as the number of machines in the system is increased. Workload B is used, with a fixed ratio of 10% update transactions, and the number of sites is varied from one to three, while keeping the number of partitions fixed to 64. Figure 4.3 shows the overall performance of the protocols. ●●● 10 25 50 100 250 500 750 1,000 1,250 1 Sites 2 Sites 3 Sites Throughput (Ktps) (log) ● naiveSER fastPSI naiveRC Workload B, 90% read−only transactions Figure 4.3:Maximum Throughput of Consistency Models. As before, the performance of naiveRC increases almost in a linear fashion as more servers are added, as explained by its minimum need for synchronisation. Although the scalability of fastPSI is limited by the fixed number of partitions, it benefits moderately from increasing the number of machines: as the overall number of partitions per machine decreases, servers free resources, and can thus fulfil more client requests. This is reflected in its performance at three sites being 1.52 times its base throughput at a single site. In contrast, the overall performance of naiveSER stays almost constant as the number of sites is increased, reflecting its need to validate every transaction, which requires greater levels of synchronisation. As the number of sites increases, so does the difference between naiveSER and fastPSI. Overall, fastPSI manages to outperform naiveSER by a factor of 2.88 with a single site, and by a factor of 3.52 at three sites. Parameter Range Default Sites 1–3 3 Update Tran. Proportion 10%–30% 10% Figure 4.4:Parameter space used in the comparison workload. 29 ● naiveSER fastPSI naiveRC ● ●● 0.03 0.10 0.30 1.00 1 Sites 2 Sites 3 Sites Number of Sites (a) ●●● 0.03 0.10 0.30 1.00 10 20 30 Update Transactions (%) (b) Throughput (normalized) Figure 4.5:Parameter space exploration to reflect the performance comparison of the protocols. Each experiment varies one parameter while keeping the other fixed at its default value (represented by the grey vertical line). Throughput is shown normalised compared to naiveRC. Overall comparison. The last two experiments have shown how the performance and scalability of the implementations is determined by the proportion of update transactions and the number of sites. To better visualise the relationship between workload choice and performance, the next experiment explores the parameter space described in Figure 4.4 when using Workload B. As in the previous experiment, the number of partitions is kept constant as the number of sites increases. The results are shown in Figure 4.5, with the throughput depicted normalised compared to the performance of naiveRC. As shown in Figure 4.5a, fastPSI and naiveSER have different scalability properties. Although both implementations suffer in comparison with naiveRC, fastPSI exhibits much better scalability in comparison with naiveSER. For naiveSER, the need to validate every transaction imposes a performance penalty that increases as more sites are added, and consequently increases the overall latency of the commit phase for every transaction. In contrast, the impact on fastPSI is less severe, as the increased latency only affects the read operations, since the proportion of update transactions is low. Figure 4.5b shows the performance comparison as the proportion of update transactions increases. While the difference in throughput between naiveRC and fastPSI stays constant, the overall performance of fastPSI is hindered by the need to compute a causally compatible snapshot. Nevertheless, the fact that the relative performance stays constant in comparison with naiveRC shows that the impact of the validation process does not grow with the number of update transactions. The impact of update transactions on naiveSER is less pronounced, and its performance difference with fastPSI gets smaller as the proportion of updates increases. This is explained by the choice of workload: read-only transactions perform four reads, while update transactions perform three. Since naiveSER also validates read-only transactions, as the proportion of updates grows, the average number of partitions that participate in the voting phase shrinks from four to three, which decreases the runtime cost of validation. 30 10% 20% 30% 40% 50% 0.00 0.05 0.10 0.15 0.20 0.25 0.30 0.35 0.40 naiveSER fastPSI Abort ratio Workload D, overall abort rate 10% 20% 30% 40% 50% 0.00 0.05 0.10 0.15 0.20 0.25 0.30 0.35 0.40 naiveSER fastPSI Abort ratio Workload E, overall abort rate Figure 4.6:Overall transaction abort ratio for different consistency models and workloads. 4.2.2. Abort Ratio This section focuses on another advantage of the relaxed consistency model of fastPSI in comparison with serialisability, namely, the ratio of aborted transactions. One of the characteristics of fastPSI is that the snapshots of transactions exhibit forward freshness: a transaction Tis able to read object versions written by transactions that committed after Tstarted. In contrast, a transaction Tunder serialisability or classical Snapshot Isolation can only read versions written by transactions that commit before T’s start time. This limitation leads to a high number of aborted transactions due to stale reads in high latency settings. As such, this section aims to explore the advantages of forward freshness in fastPSI in comparison with naiveSER, and how different workloads affect it. Figure 4.6 shows how the abort ratio of transactions varies—under different workloads—with the number of update transactions. The results were obtained with two sites. The graph on the left shows the advantages of the forward freshness of transactions, with the abort ratio of fastPSI being 30 percentage points better than naiveSER, on average. However, as the number of updates increases, so does the number of conflicting transactions. This is reflected in an increase of aborted transactions, from less than one percent to around five percent. The graph on the right, however, shows different results: as the number of update transactions grows, so does the overall abort ratio of fastPSI, from 7% to 25%. Overall, in the worst case the abort ratio of fastPSI is only 10 percentage points better than naiveSER. The difference between workloads is explained in the two graphs depicted in Figure 4.7, which explores the reasons why transactions abort. In fastPSI, a transaction might abort for two reasons: by updating an object that is overwritten by a concurrent transaction, or by failing to form a causally consistent snapshot, as detailed in 3.5. By virtue of having the same implementation of causally consistent snapshots, naiveSER inherits the same reasons, and adds a third: a transaction is forced to abort if it reads a version of an object that is later overwritten. For both protocols, a transaction might abort at two points during its execution: read aborts occur when a transaction fails to fix a causally consistent snapshot, and otherwise the transaction aborts during validation, i.e., during the termination of the 31 ● ●●●● 0.005 0.010 0.025 0.050 0.100 1.000 10 20 30 40 50 Update Transactions (%) Abort ratio (log) ●naiveSER fastPSI Workload D, aborts during validation ● ●●●● 0.005 0.010 0.025 0.050 0.100 1.000 10 20 30 40 50 Update Transactions (%) Abort ratio (log) ●naiveSER fastPSI Workload E, aborts during validation Figure 4.7:Ratio of aborts that happen during the validation phase, for different consistency models and workloads. transaction. As seen on the left graph in Figure 4.7, in Workload D all aborted transactions in fastPSI do so during the validation phase, while for naiveSER, only a small percentage of aborted transactions (between 0.5% and 2.5%) do so during termination. In contrast, fastPSI exhibits a very different behaviour with Workload E, as shown in the graph on the right. With this workload, transactions abort for the same reason in both protocols: they are unable to build causally consistent snapshots. In both cases, the amount of transactions that abort while attempting to fix a snapshot grows as the number of update transactions increases. Recall that in Workload D, update transactions update one single key, while in Workload E, they update three. This small difference in the number of updated keys explains the difference in aborted transactions for fastPSI. As described in 3.5, a transaction Twill be unable to fix a causally consistent snapshot when a) different partitions commit nonconflicting transactions in different order, and b) Tfixes a snapshot in one of those partitions before the updates of all the non-conflicting transactions become visible. Since update transactions in Workload D only update a single key, update transactions will only commit at a single partition, and therefore, no transaction can observe a different commit order. This explains why there are no aborted transactions due to inconsistent snapshots when executing this workload for fastPSI. In contrast, update transactions in Workload E update three keys, and therefore commit at three different partitions,1such that the probability of partitions committing transactions in different order grows. In addition, since transactions read three keys, the probability of observing different commit orders also grows. Thus, as the number of update transactions increases, so does the probability of transactions attempting to fix inconsistent snapshots. In naiveSER transactions also commit at the partitions they read from, meaning that update transactions in Workload D commit at three partitions. 1Since transactions choose keys following an uniform distribution, most keys will be managed by distinct partitions. 32 0.000 0.025 0.050 0.075 0.100 0.125 0.150 0.175 0.200 0.225 0.250 0.275 0.300 234 Read Keys (a) Abort ratio 2 writes / 64 partitions Read keys effect on abort ratio (Two keys written) 0.0 0.1 0.2 0.3 1234 Written Keys (b) Abort ratio 4 reads / 64 partitions Written keys effect on abort ratio 0.0 0.1 0.2 0.3 0.4 0.5 0.6 0.7 0.8 0.9 1.0 8 16 32 64 128 256 Partitions (c) Abort ratio 4 reads / 4 writes Partitions number effect on abort ratio Figure 4.8:Results from exploring the abort ratio of fastPSI across different workload and deployment scenarios. This explains why its abort ratio is similar in both workloads. Since small changes in workload choice are able to affect the overall abort ratio of fastPSI in a significant manner, the evaluation is concluded with an exploration of the parameters that have the highest impact on the number of aborted transactions. As explained previously, one of the reasons why transactions observe inconsistent snapshots is because partitions commit transactions in different orders. Thus, by changing the number of keys updated by a transaction, one can modify the number of partitions involved, which affects the chances of different commit orders. Another reason for inconsistent snapshots is that transactions are able to observe the transactions that commit in different order, which leads to change the number of keys read by the transactions. If transactions read from a small amount of partitions, the probability that they observe different commit orders will also be small. Finally, the most important factor is the number of partitions, or rather, the amount of keys per partition. When a few partitions manage all the keys, most transactions will commit at the same partitions. Therefore, the probability of having different commit orders grows, as well as the probability that a given transaction observes those orders. Conversely, when the number of partitions is large, or partitions manage only a small amount of keys, the probability of different commit orders shrinks, as does the probability of them being observed by transactions. Figure 4.8 shows the results of modifying each of the parameters mentioned, noting that, for simplicity, a single site is used while modifying the number of partitions, instead of changing the total amount of keys in the database. The percentage of update transactions is set to 50%, to show the worst possible abort ratio. The effect of changing the number of keys read by a transaction, shown in Figure 4.8a, is small. Although the overall abort ratio never grows larger than 10%, modifying the number of read keys results, at best, in an improvement of 7.5 percentage points. Blind updates are not allowed, and as such the minimum number of read keys is two, since transactions need to update at least two keys in two different partitions to introduce inconsistent snapshots in the system. In contrast, changing the number of keys updated by a transaction produces a bigger impact, as shown 33 ● ● ●● ● ● 0 5,000 10,000 15,000 20,000 25,000 30,000 8 16 32 64 128 256 Partitions Throughput (tps) 4 reads / 4 writes Partitions number effect on throughput Figure 4.9:Performance degradation of fastPSI as the number of partitions increases, with a fixed number of servers. in Figure 4.8b, with an overall change of 23 percentage points in the number of aborted transactions. When transactions update a single key, all transactions abort during validation. At this point, the system shows the best possible abort ratio for 64 partitions, at 5%. Finally, modifying the number of partitions yields the biggest change in the number of aborted transactions, as shown in Figure 4.8c. With 8 partitions, the overall abort ratio is of almost 70%, which serves as an extreme example of the importance of this parameter. As the number of partitions grows, the system reaches an abort ratio of 27%. By increasing the number of partitions to 256, the result is an abort ratio of 10%. With an overall difference of 60 percentage points in the proportion of aborted transactions, this shows that the number of partitions is the biggest influence in the abort ratio of transactions in fastPSI. It is important to note, however, that increasing the number of partitions also has a big impact on the performance of the system, as shown in Figure 4.9. During the experiment, the maximum throughput is reached at 32 partitions across 4 machines. With a small number of partitions, the increased contention causes the throughput to drop. With a big number of partitions, each server machine is also responsible for a big number of partitions. As a result, the system overloads. Thus, the number of available machines constraints the number of partitions. 34 Chapter 5 Related Work This chapter gives an overview of the previous work in the settings of transactional protocols, consistency, and application robustness. It also highlights the main differences between the contributions of this work and previous approaches. Application Robustness. The notion of robustness as applied to databases was first investigated by Fekete et al. [22], proposing a way to analyse if applications were robust against Snapshot Isolation (SI) [13]. The work of Fekete et al. has resulted in the proliferation of static analysis tools for detecting the presence of anomalies in applications [29], as well as several run-time techniques for ensuring serialisable transactions [37]. More recently, Bernardi and Gotsman [14] proposed several robustness criteria for a variety of consistency models, including Parallel Snapshot Isolation (PSI) [39]. The work on fastPSI builds on the robustness criteria of Parallel Snapshot Isolation and of Snapshot Isolation to build a hybrid protocol that allows programmers to selectively strengthen consistency guarantees for individual transactions. Entity Groups. The concept of entity was introduced by Helland [28] to refer to a selfcontained data object, with application-defined boundaries, and uniquely identified by an entity key. In addition, Helland argued for the entity to be the largest scope of transactional serialisability: transactions can only guarantee atomicity for objects held within the same entity, and are prevented from modifying objects across entities. Later systems, such as Megastore [11], introduced the concept of entity groups as disjoint aggregations of individual entities. Such systems allowed transactions to access distinct entities, and offered strong consistency for transactions accessing entities within a group, while offering almost no consistency guarantees for transactions that accessed different groups. Such systems thus broadened the scope of transactional serialisability to encompass entire entity groups. In contrast, fastPSI provides PSI for all transactions, even those that access objects in multiple entity groups. At the same time, it strengthens the consistency guarantees further for transactions accessing only individual entity groups, by providing SI. 35 5function read(T, x) 6if hx, vi ∈ T.WS then 7return v; 8j←partition(x); 9send READREQUEST(x, T.Vsnap, T.HasRead)to sj; 10 wait receive READRETURN(m)from sj; 11 if m=abort then 12 throw abort; 13 else if m=hv, Vdep,Vaggrithen 14 T.HasRead[j]← >; 15 T.RS ←(T.RS \ {hx, _i})∪ {hx, Vdep[j]i}; 16 T.Vdep ←max(T.Vdep,Vdep); 17 T.Vsnap ←max(T.Vsnap,Vaggr); 18 return v; 19 when received READREQUEST(x, Vsnap,HasRead)from cj 20 if HasRead[i]then 21 V←Vsnap; 22 else 23 wait until Vtotal[i]≥Vsnap[i]; 24 r←max{r∈CommitLog | ∀j. HasRead[j] =⇒(r.Vaggr[j]≤Vsnap[j])}; 25 if r.Vaggr[i]<Vsnap[i]then 26 send READRETURN(abort)to cj; 27 return; 28 V←r.Vaggr; 29 ver = max{ver ∈VersionLog |ver.Vcomm[i]≤V[i]}; 30 send READRETURN(ver.val,ver.Vcomm, V )to cj; Figure A.3:Serialisable local and remote read of object x 42 31 function commit(T) 32 forall sj∈partitions(T.RS ∪T.WS)do 33 send PREPARE(T, T.RS, T.WS,Vdep)to sj; 34 Vcomm ←T.Vdep; 35 decision ←commit; 36 forall sj∈partitions(T.RS ∪T.WS)do 37 wait receive VOTE(m)from sj; 38 if m=hT, abortithen 39 decision ←abort; 40 break; 41 else if m=hT, commit, kithen 42 Vcomm[j]←k; 43 forall sj∈partitions(T.RS ∪T.WS)do 44 send DECIDE(T,Vcomm,decision)to sj; 45 return decision; 46 when received PREPARE(T, RS,WS,Vdep)from cj 47 if (∃T0.(hT0,pending,RS0,WS0i ∈ CommitQueue ∨ hT0,decided,_,_,_i ∈ CommitQueue) ∧(WS0∩RS 6=∅ ∧ RS0∩WS 6=∅) ∨(∃x. hx, vsni ∈ RS ∧(VersionLog[x].last.Vcomm[i]> vsn)) then 48 send VOTE(t, abort)to cj; 49 return; 50 LastPrep ←LastPrep + 1; 51 CommitQueue.put(T,pending,RS,WS); 52 send VOTE(T,commit,LastPrep)to cj; 53 when received DECIDE(T, Vcomm,decision)from cj 54 if decision =commit then 55 CommitQueue.update(hT,decided,_,_,Vcommi); 56 else 57 CommitQueue.remove(T); 58 upon hT, decided,_,WS,Vcommi=CommitQueue.head() 59 forall {hx, vi|hx, vi ∈ WS ∧partition(x) = i}do 60 VersionLog.add(hx, v, Vcommi); 61 Vtotal ←max(Vtotal,Vcomm); 62 CommitLog.add(T, Vtotal); 63 CommitQueue.remove(T); Figure A.4:Serialisable termination protocol. 43 A.2. Read Committed Read Committed (RC) is the weakest consistency model that satisfies the isolation property required by ACID transactions. It forbids concurrent transactions from observing any data that has not been committed, but it does not place any restriction on the ordering of transactions, and does not preclude write-write conflicts. Thus, transactions may be ordered in any way. Figure A.5 shows a summary of the data structures involved in the protocol. Variables at a server si Name Domain Description CommitQueue Sequence[hTx,State,WriteSeti] where State ={pending,decided} Queue containing information about update transactions trying to commit at the server. Database Set[hObject,Valuei]Set representing the key-value store as a mapping from objects to values. Context for a transaction Tat a client ci T.WS WriteSet Write-set of T. Figure A.5:List of variables used in the Read Committed protocol, where WriteSet = Set[hObject,Valuei]. Since transactions only need to observe the last committed version of an object, it is sufficient to store only one version. Thus, the VersionLog mapping can be substituted with aDatabase that simply maps an object to its latest version. In addition, transactions don’t need to observe a consistent snapshot of the state of a partition, and therefore all data structures related to computing a snapshot can be removed. This is reflected in the execution of a transaction, as can be seen in Figure A.7. A server siexecuting a remote read on behalf of a transaction Tsimply fetches the currently available value of the requested object, and returns it to the client (line 13). A protocol satisfying Read Committed still needs to offer atomic visibility. To do so, the implementation uses two-phase commit to guarantee that a transaction commits at every partition (line 14). Servers that participate during the commit phase always vote commit (line 30), since RC does not preclude write-write conflicts. After a successful commit phase, all servers incorporate the updates of the transaction to its partition state (line 38). 1function start() 2return new Tx(WS =∅); 3function write(T, x, v) 4T.WS ←(T.WS \ {hx, _i})∪ {hx, vi}; Figure A.6:Initialisation of a transaction and update of an object xat client ciunder Read Committed. 44 5function read(T, x) 6if hx, vi ∈ T.WS then 7return v; 8j←partition(x); 9send READREQUEST(x)to sj; 10 wait receive READRETURN(v)from sj; 11 return v; 12 when received READREQUEST(x)from cj 13 send READRETURN(Databasei.get(x)) to cj; 14 function commit(T) 15 if t.ws =∅then 16 return commit; 17 forall sj∈partitions(T.WS)do 18 send PREPARE(T)to sj; 19 decision ←commit; 20 forall sj∈partitions(T.WS)do 21 wait receive VOTE(m)from sj; 22 if m=hT, abortithen 23 decision ←abort; 24 break; 25 forall sj∈partitions(T.WS)do 26 send DECIDE(T,decision)to sj; 27 return decision; 28 when received PREPARE(T)from cj 29 CommitQueue.put(T, pending,WS); 30 send VOTE(T, commit)to cj; 31 when received DECIDE(T, decision)from cj 32 if decision =commit then 33 CommitQueue.update(hT, decided,_i); 34 else 35 CommitQueue.remove(T); 36 upon hT, decided,WSi=CommitQueue.head() 37 forall {hx, vi|hx, vi ∈ WS ∧partition(x) = i}do 38 Databasei.apply(x, v); 39 CommitQueue.remove(T); Figure A.7:Read Committed execution protocol. 45 Bibliography [1] fastPSI client-side library. URL https://github.com/ergl/pvc/tree/v0.8.0. [2] Protocol Buffers. URL https://github.com/protocolbuffers/protobuf. [3] fastPSI Server. URL https://github.com/ergl/antidote/tree/pvc. [4] Atul Adya. Weak Consistency: A Generalized Theory and Optimistic Implementations for Distributed Transactions. Ph.D., MIT, Cambridge, MA, USA, March 1999. [5] Deepthi Devaki Akkoorath, Alejandro Z. Tomsic, Manuel Bravo, Zhongmiao Li, Tyler Crain, Annette Bieniusa, Nuno Preguica, and Marc Shapiro. Cure: Strong Semantics Meets High Availability and Low Latency. In Proceedings of the 36th International Conference on Distributed Computing Systems (ICDCS 2016), 2016. [6] M. S. Ardekani, P. Sutra, and M. Shapiro. Non-monotonic snapshot isolation: Scalable and strong consistency for geo-replicated transactional systems. In 2013 IEEE 32nd International Symposium on Reliable Distributed Systems, pages 163–172, Sep. 2013. doi: 10.1109/SRDS.2013.25. [7] Masoud Saeida Ardekani. Ensuring Consistency in Partially Replicated Data Stores. Ph.d., UPMC, Paris, France, September 2014. [8] Masoud Saeida Ardekani, Marek Zawirski, Pierre Sutra, and Marc Shapiro. The space complexity of transactional interactive reads. In Proceedings of the 1st International Workshop on Hot Topics in Cloud Data Processing, HotCDP ’12, New York, NY, USA, 2012. Association for Computing Machinery. ISBN 9781450311625. doi: 10.1145/ 2169090.2169094. URL https://doi.org/10.1145/2169090.2169094. [9] Masoud Saeida Ardekani, Pierre Sutra, and Marc Shapiro. G-DUR: A middleware for assembling, analyzing, and improving transactional protocols. In Proceedings of the 15th International Middleware Conference, Middleware ’14, page 13–24, New York, NY, USA, 2014. Association for Computing Machinery. ISBN 9781450327855. doi: 10.1145/2663165.2663336. URL https://doi.org/10.1145/2663165.2663336. [10] Peter Bailis, Alan Fekete, Joseph M. Hellerstein, Ali Ghodsi, and Ion Stoica. Scalable atomic visibility with RAMP transactions. In Proceedings of the 2014 ACM SIGMOD International Conference on Management of Data, SIGMOD ’14, page 27–38, New York, NY, USA, 2014. Association for Computing Machinery. ISBN 9781450323765. doi: 10.1145/2588555.2588562. URL https://doi.org/10.1145/2588555.2588562. 46 [11] Jason Baker, James C. Corbett, JJ Furman, Andrey Khorlin, James Larson, JeanMichel Leon, Yawei Li, Alexander Lloyd, and Vadim Yushprakh. Megastore: Providing Scalable, Highly Available Storage for Interactive Services. In Proceedings of the Conference on Innovative Data system Research (CIDR), pages 223–234, 2011. URL http://www.cidrdb.org/cidr2011/Papers/CIDR11_Paper32.pdf. [12] Carlos Baquero and Nuno M. Preguiça. Why logical clocks are easy. Commun. ACM, 59(4):43–47, 2016. [13] Hal Berenson, Phil Bernstein, Jim Gray, Jim Melton, Elizabeth O’Neil, and Patrick O’Neil. A Critique of ANSI SQL Isolation Levels. In Proceedings of the 1995 ACM SIGMOD international conference on Management of data - SIGMOD ’95, 1995. [14] Giovanni Bernardi and Alexey Gotsman. Robustness against Consistency Models with Atomic Visibility. In Josée Desharnais and Radha Jagadeesan, editors, 27th International Conference on Concurrency Theory (CONCUR 2016), volume 59 of Leibniz International Proceedings in Informatics (LIPIcs), pages 7:1–7:15, Dagstuhl, Germany, 2016. Schloss Dagstuhl–Leibniz-Zentrum fuer Informatik. ISBN 978-3-95977-0170. doi: 10.4230/LIPIcs.CONCUR.2016.7. URL http://drops.dagstuhl.de/opus/ volltexte/2016/6165. [15] Philip A. Bernstein, Vassos Hadzilacos, and Nathan Goodman. Concurrency Control and Recovery in Database Systems. Addison-Wesley, 1987. [16] Eric A Brewer. Towards Robust Distributed Systems (keynote). In 19th ACM Symposium on Principles of Distributed Computing (PODC), July 2000. [17] Andrea Cerone, Giovanni Bernardi, and Alexey Gotsman. A framework for transactional consistency models with atomic visibility. In 26th International Conference on Concurrency Theory, CONCUR 2015, Madrid, Spain, September 1-4, 2015. [18] Brian F. Cooper, Adam Silberstein, Erwin Tam, Raghu Ramakrishnan, and Russell Sears. Benchmarking cloud serving systems with ycsb. In Proceedings of the 1st ACM Symposium on Cloud Computing, SoCC ’10, New York, NY, USA, 2010. [19] James C. Corbett, Jeffrey Dean, Michael Epstein, Andrew Fikes, Christopher Frost, J. J. Furman, Sanjay Ghemawat, Andrey Gubarev, Christopher Heiser, Peter Hochschild, and et al. Spanner: Google’s globally distributed database. ACM Trans. Comput. Syst., 31(3), August 2013. ISSN 0734-2071. doi: 10.1145/2491245. URL https://doi.org/10.1145/2491245. [20] Giuseppe DeCandia, Deniz Hastorun, Madan Jampani, Gunavardhan Kakulapati, Avinash Lakshman, Alex Pilchin, Swaminathan Sivasubramanian, Peter Vosshall, and Werner Vogels. Dynamo: amazon’s highly available key-value store. SIGOPS Oper. Syst. Rev., 41(6):205, October 2007. ISSN 01635980. doi: 10.1145/1323293.1294281. 47 [21] Alan Fekete. Allocating Isolation Levels to Transactions. In Proceedings of the TwentyFourth ACM SIGMOD-SIGACT-SIGART Symposium on Principles of Database Systems, PODS ’05, page 206–215, New York, NY, USA, 2005. Association for Computing Machinery. ISBN 1595930620. doi: 10.1145/1065167.1065193. URL https: //doi.org/10.1145/1065167.1065193. [22] Alan Fekete, Dimitrios Liarokapis, Elizabeth O’Neil, Patrick O’Neil, and Dennis Shasha. Making Snapshot Isolation Serializable. ACM Trans. Database Syst., 30 (2):492–528, June 2005. ISSN 0362-5915. doi: 10.1145/1071610.1071615. URL https://doi.org/10.1145/1071610.1071615. [23] Seth Gilbert and Nancy Lynch. Brewer’s conjecture and the feasibility of consistent, available, partition-tolerant web services. SIGACT News, 33(2):51–59, June 2002. ISSN 0163-5700. doi: 10.1145/564585.564601. URL https://doi.org/10.1145/564585. 564601. [24] Alexey Gotsman, Hongseok Yang, Carla Ferreira, Mahsa Najafzadeh, and Marc Shapiro. ’Cause I’m Strong Enough: Reasoning about Consistency Choices in Distributed Systems. In Proceedings of the 43rd Annual ACM SIGPLAN-SIGACT Symposium on Principles of Programming Languages, POPL ’16, page 371–384, New York, NY, USA, 2016. Association for Computing Machinery. ISBN 9781450335492. doi: 10.1145/2837614.2837625. URL https://doi.org/10.1145/2837614.2837625. [25] Jim Gray and Leslie Lamport. Consensus on Transaction Commit. ACM Transactions on Database Systems, 31(1):133–160, March 2006. ISSN 0362-5915. doi: 10.1145/ 1132863.1132867. URL https://doi.org/10.1145/1132863.1132867. [26] Rachid Guerraoui and André Schiper. Genuine atomic multicast in asynchronous distributed systems. Theoretical Computer Science (Elsevier), 254:297–316, 2001. URL http://infoscience.epfl.ch/record/49965. [27] R. C. Hansdah and Lalit M. Patnaik. Update Serializability in Locking. In Proceedings of the International Conference on Database Theory, ICDT ’86, page 171–185, Berlin, Heidelberg, 1986. Springer-Verlag. ISBN 3540171878. [28] Pat Helland. Life beyond Distributed Transactions: an Apostate’s Opinion. In CIDR 2007, Third Biennial Conference on Innovative Data Systems Research, Asilomar, CA, USA, January 7-10, 2007, Online Proceedings, pages 132–141. www.cidrdb.org, 2007. URL http://cidrdb.org/cidr2007/papers/cidr07p15.pdf. [29] Sudhir Jorwekar, Alan Fekete, Krithi Ramamritham, and S. Sudarshan. Automating the detection of snapshot isolation anomalies. In Proceedings of the 33rd International Conference on Very Large Data Bases, VLDB ’07, page 1263–1274. VLDB Endowment, 2007. ISBN 9781595936493. 48 [30] David Karger, Eric Lehman, Tom Leighton, Rina Panigrahy, Matthew Levine, and Daniel Lewin. Consistent hashing and random trees: Distributed caching protocols for relieving hot spots on the World Wide Web. In Proceedings of the Twenty-Ninth Annual ACM Symposium on Theory of Computing, STOC ’97, page 654–663, New York, NY, USA, 1997. Association for Computing Machinery. ISBN 0897918886. doi: 10.1145/258533.258660. URL https://doi.org/10.1145/258533.258660. [31] Leslie Lamport. The part-time parliament. ACM Trans. Comput. Syst., 16(2):133–169, May 1998. ISSN 0734-2071. doi: 10.1145/279227.279229. URL https://doi.org/10. 1145/279227.279229. [32] Wyatt Lloyd, Michael J. Freedman, Michael Kaminsky, and David G. Andersen. Don’t Settle for Eventual: Scalable Causal Consistency for Wide-Area Storage with COPS. In Proceedings of the Twenty-Third ACM Symposium on Operating Systems Principles, SOSP ’11, page 401–416, New York, NY, USA, 2011. Association for Computing Machinery. ISBN 9781450309776. doi: 10.1145/2043556.2043593. URL https://doi.org/10.1145/2043556.2043593. [33] Henrique Moniz, João Leitão, Ricardo J. Dias, Johannes Gehrke, Nuno Preguiça, and Rodrigo Rodrigues. Blotter: Low Latency Transactions for Geo-Replicated Storage. In Proceedings of the 26th International Conference on World Wide Web, WWW ’17, page 263–272, Republic and Canton of Geneva, CHE, 2017. International World Wide Web Conferences Steering Committee. ISBN 9781450349130. doi: 10.1145/3038912.3052603. URL https://doi.org/10.1145/3038912.3052603. [34] Mahsa Najafzadeh, Alexey Gotsman, Hongseok Yang, Carla Ferreira, and Marc Shapiro. The CISE Tool: Proving Weakly-Consistent Applications Correct. In Proceedings of the 2nd Workshop on the Principles and Practice of Consistency for Distributed Data, PaPoC ’16, New York, NY, USA, 2016. Association for Computing Machinery. ISBN 9781450342964. doi: 10.1145/2911151.2911160. URL https: //doi.org/10.1145/2911151.2911160. [35] D. S. Parker, G. J. Popek, G. Rudisin, A. Stoughton, B. J. Walker, E. Walton, J. M. Chow, D. Edwards, S. Kiser, and C. Kline. Detection of mutual inconsistency in distributed systems. IEEE Trans. Softw. Eng., 9(3):240–247, May 1983. [36] Sebastiano Peluso, Pedro Ruivo, Paolo Romano, Francesco Quaglia, and Luis Rodrigues. GMU: Genuine Multiversion Update-Serializable Partial Data Replication. IEEE Transactions on Parallel and Distributed Systems, 27(10):2911–2925, October 2016. ISSN 1045-9219. doi: 10.1109/TPDS.2015.2510998. URL https://doi.org/10. 1109/TPDS.2015.2510998. [37] Dan R. K. Ports and Kevin Grittner. Serializable Snapshot Isolation in PostgreSQL. Proc. VLDB Endow., 5(12):1850–1861, August 2012. ISSN 2150-8097. doi: 10.14778/ 2367502.2367523. URL https://doi.org/10.14778/2367502.2367523. 49 [38] Masoud Saeida Ardekani, Pierre Sutra, Marc Shapiro, and Nuno Preguiça. On the scalability of snapshot isolation. In Felix Wolf, Bernd Mohr, and Dieter an Mey, editors, Euro-Par 2013 Parallel Processing, pages 369–381, Berlin, Heidelberg, 2013. Springer Berlin Heidelberg. ISBN 978-3-642-40047-6. [39] Yair Sovran, Russell Power, Marcos K. Aguilera, and Jinyang Li. Transactional storage for geo-replicated systems. In Proceedings of the Twenty-Third ACM Symposium on Operating Systems Principles - SOSP ’11, 2011. [40] Werner Vogels. Eventually consistent. ACM Queue, 6(6):14–19, October 2008. 50