David G Luenberger

THE CONJUGATE RESIDUAL METHOD FOR CONSTRAINED MINIMIZATION PROBLEMS

Introduction. The method of conjugate gradients [ l]

-{5], a particular version of the general class of conjugate direction methods, was originally developed for solving a linear system of equations (1.1)

Ax = b,
where A is an .n x n symmetric, positive defini te matrix, x is an unknown n-vector and bis a fixed n-vector. The conjugate gradient method can be viewed as a descent procedure for the problem : Minimize

( 1.2) (x , Ax) -2(x , b);
and through this viewpoint the method can be generalized so as to apply to the solution of general unconstrained minimization problems. For these general problems conjugate gradient methods are among the most effective first order descent procedures. Minimization problems subject to constraints cannot be treated directly by the conjugate gradient methods although they have often been employed indirectly through the artifice of introducing penalty functions or of suitably projecting the descent directions onto the constraint surface. The major motivation for the research presented in this paper is to develop a conjugate direction procedure that is capable of handling constraints directly. This quickly leads one to consider equations of the form (1.1) with the matrix A symmetric and nonsingular but not necessarily positive definite. For, consider the simple problem: Minimize

(1.3) (u, Qu)-2(c, u) subject to Bu= d,
where Q is an n x n symmetric, positive definite matrix, B is an m x n matrix of rank m < n. The n-vector u is unknown and c and d are fixed nand m-vectors respectively. By introducing an m-dimensional Lagrange multiplier this problem is found to be equivalent to that of solving the linear vector equation

(1.4)
Denoting the (n + m) x (n + m) matrix in (1.4) by A we see that A is symmetric, nonsingular, but not positive definite. Of course, (1.1) can be modified to the equivalent form (1.5)

A 2 x =Ab,
where now A 2 is positive defini te, but this stratagem requires roughly twice the computational effort of methods which could be applied directly if A itself were positive defini te [START_REF] Beckman | The solution of linear equations by the conjugale gradient method, Mathematical Methods for Digital Computers[END_REF].

In this paper we discuss methods requiring essentially the same computational effort as those employed if A is positive definite. An alternative approach to this problem, which salvages the standard conjugate gradient algorithm, is considered in [START_REF]Hyperbolic pairs in the method of conjugale gradients[END_REF].

The method of conjugate residuals.

DEFINITION. Given an n x n matrix A, a pair of n-vectors x, y is said to be

A-orthogonal if (x, Ay) = O.
We now develop an iterative technique for solving

(2.1) Ax = b
which has the same important characteristics as the method of conjugate gradients but which in addition is applicable in the case where A is symmetric but not necessarily positive definite. The procedure is a descent procedure in that a sequence of approximate solutions where ak is chosen to minimize E. The novel feature of the method given below is of course the method for selecting the direction vectors Pk. ln the procedure the direction vectors are chosen to be an A 2 -orthogonalized version of the residuals rk = b -Axk. This is similar to the method of conjugate gradients but now since the gradient of the underlying functional E at xk is -2Arb and not -2rk, we refer to the scheme as conjugate residuals.

Here is the fundamental iteration scheme.

Conjugate residual scheme. Select x 1 arbitrarily. Set p 1 = r 1 = b -Ax 1 , and repeat the following steps omitting (2.3a, b) on the first step.

Ifak_ 1 =F 0, (2.3a)
If ak_ 1 = 0, (2.3b)

Pk = Ark -'YkPk-1 -OkPk -2, (Ark, A 2 Pk -d 'Yk = (p A2 )' k-t'
Pk-1

(2.3c)

(2.3d)

We prove below that the process defined by (2.3) converges to the unique solution of (2.1) in n steps or less. Furthermore, the residuals r 1 , r 2 , • • • form an A-orthogonal sequence and the direction vectors p 1 , p 2 , • • • form an A 2 -orthogonal sequence.

If the ma tri x A is positive defini te, then for each k, ak > 0 and hence (2.3b) is never used. In this case the procedure can be regarded as one of the family of general conjugate gradient methods discussed by Hestenes [START_REF] Hestenes | The conjugale gradient method for so/ving /inear systems[END_REF]. However, even in the positive definite case this method has not been explicitly pointed out and in terms of total computation it is as efficient as ali standard procedures.

DEFINITION. Corresponding to the symmetric matrix A, a vector x is said to be singular if (x, Ax) = O.

If Ais not positive defini te, ak may be positive, negative or zero. The condition ak = 0 corresponds, as shown below, to singularity of the residual rk.

THEO REM 1. The following relations hold for the method of conjugate residuals: Proof The proof is by induction. The vectors r 1 , p 1 = r 1 and r 2 satisfy these relations since:

(a) (pi, A 2 p) = 0, i # j, (b) (ri, Ap) = 0, i > j + 1, (c) (d) (e
(a) If r 1 is singular,

r 2 = r 1 = p 1 and (r 1 ,Ard =O. (b) If r 1 is not singular, (r 1 , Ar 1 ) = (p 1 ,Ar 2 ) = (r 1 ,Ard-a 1 (p 1 ,A 2 p 1 )
= 0 by (2.3c). Now assume that these relations are true for the first k -1 steps, i.e., through rb Pk -l , ak-l, f3k -l. There are now two distinct cases of interest: ak-l =/= 0 and ak_ 1 = O. We must show that rk+ 1 and Pk can be added to these relations in either case.

Case 1. ak -l =1= O. To prove (e) we have immediately

(rb Apk) = (rb Ark) -f3k(rb APk-d = (rb Ark)
by the induction hypothesis (b). Also,

(rb Ark) = (rb Ark _ 1 ) -ak _ 1 (rkl A 2 pk _ 1 )
whence, by (d) , which gives the formula for f3k•

To prove (a) we write

(pbA 2 pJ = (rk,A 2 pi)-f3k(Pk-t,A 2 Pi).
For i < k --1 the first term on the right vanishes because of the induction hypotheses {c) and (b) and the second term vanishes by (a). For i = k-1 the two terms cancel.

To prove (b) we write

(rk+ 1 , Api) = (rk, Api) -ak(Apb Api).
For i < k the right side vanishes by (b) and (a). For i = k the two terms on the right cancel.

To prove (c) we have obviously, for j > 1,

j-1 (2.4) rj = P1 -L aiAPi i = 1
and hence settingj = k and using (2.3a), we have k-2

Pk = Pt -L aiAPi -ak -1APk -1 -f3kPk -1• i= 1
Th us, in view of the induction hypothesis (c) and ak_ 1 '# 0 we obtain Apk -l

E [p 1 , P2, • • • , Pk]• To prove (d) we write (rk+ 1 , ArJ = (rk , Ari) -ak(Apb ArJ
For i < k the first term on the right vanishes by the induction hypothesis. The second term also vanishes for i < k by the second part of (c) and (a). For i = k the right side becomes, by (e),

(rb Ark) (rb Ark) -( 2 ) (A pb Ark). Pk, A Pk

But this vanishes because (pk, A 2 pk) = (rk -f3kPk -1, A 2 pk) = (rb A 2 pk).

To prove (g) suppose Pk = 8. Then

rk = f3kPk-1 •
But by (e) and (b) this implies {3k = 0 and hence rk = 8.

Case 2. ak _ 1 = O. To prove (f) we note that by (2.3b),

(2.5)

(rb A pk) = (rk, A 2 rk) -'Yk(rb A pk -d -bk(rb Apk _ 2 ).
The second two terms on the right are zero by the hypothesis (b), and hence if rk =1= 8 we have ak =/: O. This proves that there cannet be generated two consecutive singular points in the procedure.

To prove (a) we write To prove (c) we note that since Pk -1 = 0 we have Pk-1 = rk-t and since

(pk, A 2 pJ = (Ark , A 2 Pi)-Yk(pk -1, A 2 pi)-bk(Pk -2, A 2 PJ For i = k -1, i = k -2
ak _ 1 = 0 we have rk = rk _ 1 . Thus, by (2.3b),

Pk = Apk -1 -YkPk-t -bkPk-2
which proves (c).

To prove (d) we write

(rk + 1 , ArJ = (rb Ari) -cxk(Apb ArJ
For i < k, this vanishes by the same argument as Case 1. For i = k we note that rk = rk-t = Pk-1 and hence the first term vanishes by the induction hypothesis on (d) and the second term vanishes by (a). Part (g) follows immediately from (2.5). This completes the proof of Theo rem 1.

We now state the fundamental minimization property of the method of conjugate residuals. Proof From part (g) of Theorem 1 we see that the sequence Pt , p 2 , • • • does not termina te un til a solution is obtained. Furthermore, by the A 2 -orthogonality of the p;'s this sequence is linearly independent. The result then follows from Theorem 2.

To summarize the results derived in this section, the method of conjugate residuals progresses in a fashion entirely analogous to that of conjugate gradients by genera ting A 2 -orthogonal direction vectors and minimizing Il A x -b 11 2 over increasing subspaces. If at sorne stage the residual rk should be singular, then the next direction vector Pk is equal to rk and the functional E cannet be reduced by moving in that direction since xk is already at the minimum along that line. The next direction Pk + 1 is then generated by a special formula and this new direction will not be singular. Thus singular directions occur only as isolated members of the sequence Pt, p 2 , • • • , Pn • The process converges within n steps.

3. Computational requirements. An important feature of the method of conjugate residuals, which sets it apart from other schemes for handling the nonpositive definite case, is that only a single multiplication of a vector by a matrix is required at each step. To see this, suppose that Pk -2 , Apk_ 2 , Pk-1 , Apk -1 , rk have been stored. Th en we have the following.

Case 1. ak _ 1 :::1= O.

Step la. Calculate Ark. This requires one matrix multiplication.

Step lb. Calculate {Jk> Pk from (2.3a).

Step le. Calculate Apk from Apk = Ark -[JkAPk-1 , and then o:b x k+ 1 from (2.3c).

Step ld. Calculate rk + 1 from rk + 1 = rk -akAPk• Case 2. ak-l = O. In this case, rk_ 1 is singular and it follows from part (e) of Theorem 1 that Pk-1 = 0 and hence Pk -l = rk_ 1 . Also since ak-1 = 0 we have rk = rk -l. Thus Ark = Apk -l is on hand.

Step 2a. Calculate A 2 rk = A(APk -d. This requires one multiplication of a vector by a matrix.

Step 2b. Calculate Yb bk and Pk from (2.3b).

Step 2c. Calcula te Apk from Apk = A 2 rk -yk APk -1 -bkAPk -2 and then o:b xk + 1 from (2.3c).

Step 2d. Calculate rk+ 1 = rk -akAPk.

Note that the storage requirements are really not as large as indicated since

Pk -2 , Apk _ 2 are only needed in the singular case and then Pk-l = rk _ 1 = rk, so extra storage is available.

A triangulation method.

In this section we show that the nonsingular part of the scheme of conjugate residuals can be implemented by performing only a sequence of line searches (i.e., one-dimensional minimization problems) and residual evaluations, without explicitly computing inner products. Although this procedure is of little value for solving a linear equation, its obvious generalization to nonlinear equations is of substantial importance.

The method is more or less a straightforward extension of the method of parallel tangents (PART AN) which implements the method of conjugate gradients when A is positive defini te [START_REF] Shah | Sorne a/gorithmsfor minimizing afunction of severa/ variables[END_REF], [START_REF] Wolfe | Methods of nonlinear programming[END_REF]. The short derivation of the method given here is of sorne independent interest in view of the lengthy standard derivations. Our derivation exploits the results of Theorem 1.

Triangulation procedure. Given an arbitrary x 1 let p 1 = r 1 and determine x 2 as the point minimizing E(x ) = II A x -bll 2 along the line x 1 + ap 1 , -oo <a < oo .

After that the process is defined by induction. Having determined

x 1 , x 2 , • • • , xk> put
where flk is determined to minimize E(yk ). Then let where wk is determined to minimize E(xk+ t).

We shall show that the triangulation procedure described above is equivalent to the method of conjugate residuals in the nonsingular case. If during the process a residual rk should become singular, the process breaks down and recovery must be made in sorne other fashion.

THEOREM 4. The method of triangulation, in the nonsingular case, is equivalent to the method of conjugate residuals.

Proof The proof is by induction. The statement is obviously true for the first step. Suppose it is true for x 1 , x 2 , • • • , xk. For simplicity and without loss of generality we assume that x 1 = 8. According to the method of conjugate residuals the point xk+ 1 is chosen so as to minimize the functional E over the subspace

[p 1 , p 2 , • • • , Pk -1 , rk].
Furthermore, by the orthogonality relations of the method this is equivalent to minimizing E over the subspace determined by Pk -l and rk.

This in turn is equivalent to fin ding a point xk + 1 in that plane such that the residual there, rk+ 1 = b -Axk+ 1 , is A-orthogonal to both rk and Pk-t.

Consider the points xk -1 and Yk• The residuals at these points, rk-l, b -Ayk, are bothA-orthogonal to rb the first by the induction hypothesis and part (d) of Theorem 1, the second because Yk is the minimizing point of E in the direction rk.

Therefore, by linearity, the residual at any point along the line through xk-l and Yk is A-orthogonal to rk.

1t is therefore only necessary to find a point on this line where the residual is A-orthogonal to Pk -l. This clearly can be accomplished by minimizing E along the line.

Application to constrained minimization problems and nonlinear equations.

Consider the nonlinear equation (5.1)

F(x) = 8 ,
where Fis a mapping from En into En and 8 is the zero vector in En. We restrict attention to the case where the matrix offirst partial derivatives F '(x) is symmetric. As a primary motivating example consider the constrained minimization problem: Minimize

(5.2) f(u) subject to H(u) = 8,
where fis a nonlinear functional on Em and H is a nonlinear mapping from Em to EP, where p < m. If u 0 is the solution of (5.2) and the rank of H'(u 0 ) is equal to p, then, as is weil known, there is a 1 0 E EP such that (u 0 , 1 0 ) is a solution of the simultaneous equations

(5.3) f'(u) + l'H'(u) = 8,
H(u) = 8.

The system (5.3) is of the type (5.1) with n = m + p.

The procedures developed here also have application to minimization problems involving inequality constraints, but this topic is deferred to another paper. Solution of(5.1) is, of course, equivalent to minimizing the form tf;( x) = Il F(x) ll 2 and indeed this is the approach taken here. Standard methods of minimization, however, require the direction vectors to be selected on the basis of the gradient. In the case of problem (5.2) this in turn would require knowledge of second derivatives off and H. By employing the philosophy of the conjugate residual method, the descent directions are based on the residuals F(x ).

We shall briefiy discuss three ways in which the conjugate residual scheme can be generalized to nonlinear problems. These are identical in spirit to the standard methods for adopting the conjugate gradient procedure to nonlinear problems and hence we shall not pursue the details.

A. If one is willing to assume the availability of F'(x ) , then the local approximation A "' F'(x ) can be used in the formulas (2.3). The conjugate gradient analogue of this procedure has been developed by Daniel [START_REF] Daniel | The conjugale gradient method for linear and nonlinear operator equations[END_REF] , [START_REF]Convergence of the conjugale gradient method with computationally convenient modifications[END_REF] .

B. Multiplication of a vector by the matrix A can be approximated by the difference of two residuals by the formula

F(xk + y) -F(xk) ~ Ay
which is exact in the linear case (the step size can of course be adjusted for best results). Sin ce, in the linear case, the conjugate residual scheme can be implemented with a single A multiplication, soin the nonlinear case it can be approximated by taking a single extra step. We give here the procedure corresponding to Case 1 of § 3, the other case being a direct analogue. Assume xk, Pk -l, qk -l, rk are at hand, where Pk+ 1 is the actual direction vector used to get xk , q k _ 1 is an artificial vector used to approximate Apk-l , and rk = F(xd. Then:

Calculate an approximation sk to Ark by A more refined procedure would be to calculate Pk from (5.4a) and (5.4b) and then search along the direction Pk for a point minimizing Il F(x)ll 2 initia ting the search at the point predicted by (5.4c) . For details on such line searching techniques and the analogue of this procedure for conjugate gradients, see [START_REF] Fletcher | Functional minimization by conjugale gradients[END_REF]. C. The method of triangulation can be employed directly since it simply entails evaluation of the residual and line searching. There is a slight difficulty if a singular pointis encountered. In this case a good procedure is probably to take a gradient step and then start again. As in B above the gradient can be approximated by F(xk + rk) -F(x k).

There are a few general remarks that apply to the application of any of the above schemes when applied to nonlinear equations. The first consideration is the necessity for occasionally restarting the process. The methods of conjugate gradients and conjugate residuals that we have discussed are not self-correcting. Specifically, if the initial direction vector should for sorne reason be in error, the usual conjugate properties of the subsequently generated direction vectors will not hold and the process may not termina te in a fini te number of steps. This implies, in particular, that even if a nonlinear equation is exactly linear in a region containing the solution, any of the nonlinear versions of conjugate residuals (or conjugate gradients) may not converge in a fini te number of steps. For this reason, it is good practice to restart these procedures every n steps orso. In view of this comment, the ploy of taking a gradient step and restarting when encountering a singular point does not seem unattractive.

The second comment relates to the detection of singular points. In a nonlinear problem, a singular direction (or a nearly singular direction) is characterized by a hait or near stoppage of the descent process. In other words, if a singular pointis approached, the values of the objective functional II F(x )ll 2 will converge to sorne positive value. However, since the desired value of this objective functional is known to be zero, such a singularity is easily identified and the singularity procedure or a gradient step can be employed to move away and continue the descent.

) a;- 1 #

 1 0 implies (rb Ap;) = (r;, Ar;), {3; = -(ri,Ari)/(r; _ 1 ,Ar;-d, (f) a;-1 = 0 and ri # 0 implies ai# 0, (g) ri =/= 8 implies Pi =/= 8 (8 denotes zero vector).

  the terms on the right cancel. For i < k -2 we have by the induction hypothesis on (c) that A 2 pi E [p 1 , p 2 , • • • , Pk -1 ] and hence by (b) the first term vanishes. The ether two terms vanish by (a).The proof of (b) is the same as for Case 1.Now if ak-t = 0, then we have cxk-2 =/: 0 and also (rk-t, Apk_ 1 ) =O. Since (rk-1,APk -1) = (rk -1,Ark-d-Pk-1 (rk-1 ,APk-2 ), it follows, using (b), that (rk_ 1 , Ark _ 1 ) = 0, and by (e) , we have Pk -1 = O. (Equivalently, ak _ 1 = 0 if and only if rk _ 1 is singular.)

THEOREM 2 .

 2 For each k the point xk+ 1 minimizes thefunctional E(x) = IIAx-bll 2 over the linear variety x 1 + [pt, P2, • • • , Pk]. Proof This follows immediately from the projection theorem and the fact that rk+t is A-orthogonal to the subspace [p 1 ,p 2 , •• •, Pk]. THEOREM 3. The method of conjugate residuals converges in n steps or less to the unique solution of Ax = b.


Devaki Deepthi

Alejandro Akkoorath

Manuel Tomsic

Zhongmiao Bravo

Tyler Li

Annette Crain

Nuno Bieniusa

Marc Preguiça

Shapiro

Deepthi Devaki Akkoorath

Alejandro Z Tomsic

Manuel Bravo

Zhongmiao Li

Tyler Crain

Annette Bieniusa

Nuno Preguiça

Marc Shapiro

Cure: Strong semantics meets high availability and low latency

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

I. INTRODUCTION

Internet-scale applications are typically layered above a high-performance distributed database engine running in a data centre (DC). A recent trend is to use geo-replication across several DCs to avoid wide-area network latency and to tolerate downtime. This scenario poses big challenges to the distributed database. Since network failures (called partitions) are unavoidable, according to the CAP theorem [? ] the database design must sacrifice either strong consistency or availability. Traditional databases are "CP"; they provide consistency and a high-level SQL interface, but lose availability. NoSQL-style databases are "AP", highly available, which brings significant performance benefits. However, AP-databases expose application developers to inconsistency anomalies, and most provide only low-level key-value interface.

To alleviate this problem, recent work has focused on enhancing AP designs with stronger semantics [? ? ? ]. This paper presents Cure, our contribution in this direction. While providing availability and performance, Cure supports: (i) causal+ consistency, ensuring that if one update happens before another, they will be observed in the same order, and that replicas converge to the same state under concurrent conflicting updates, (ii) support for high-level replicated data types (CRDTs) such as counters, sets, tables and sequences, with intuitive semantics and guaranteed convergence even in the presence of concurrent conflicting updates and partial failures, and (iii) transactions, ensuring that multiple keys (objects) are both read and written consistently, in an interactive manner.

Causal+ consistency (CC+) [? ? ] represents a sweet spot in the availability-consistency tradeoff. It is the strongest model compatible with availability [? ] for individual operations. Since it ensures that the causal ordering of operations is respected, it is easier to reason about for programmers and users. Consider, for instance, a user who posts a new photo to her social network profile, then comments on the photo on her wall. Without causal consistency, a user might observe the comment but not be able to see the photo, which requires extra programming effort to avoid the anomaly at the application level.

CC+ requires that replicas converge to the same state under concurrent conflicting updates. For guaranteeing convergence, many existing causal+ consistent systems adopt the last-writer-wins rule [? ? ? ? ? ], where the update that occurs "last" overwrites the previous ones. We rely on CRDTs, developer-friendly high-level data types that guarantee convergence and have rich semantics [? ]. Operations on CRDTs are not only register-like assignments, but methods corresponding to a CRDT object's type. For example, a set supports add(item) and remove(item) operations. The implementation of a CRDT set guarantees that no matter the order in which a replica receives two conflicting add and remove operations, the state of the set will converge at different replicas without the need for synchronization or application conflict handling. For instance, the Bet365 developers report that using Set CRDTs changed their life, freeing them from low-level detail and from having to compensate for concurrency anomalies [? ].

Performing multiple operations in a transaction enables the application to maintain relations between multiple objects or keys. Highly Available Transactions (HATs) eschew traditional strong isolation properties, which require synchronisation, in favour of availability and low latency [? ? ]. Existing CC+ HAT implementations provide either reading from a snapshot [? ? ? ? ? ] or atomicity of updates [? ? ]; we introduce Transactional Causal Consistency (TCC), where all transactions provide both.

Taken together, the above features provide clear and strong semantics to developers. In fact, the combination of the three equip Cure with the strongest semantics ever provided by an always-available data store. The absence of any of these features can be compensated by ad-hoc mechanisms at the application level, but this is tricky and error prone [? ].

The contributions of this paper are the following:

• a novel programming model providing causally consistent interactive transactions with high-level, confluent data types ( §II-A); • a high-performance protocol, supporting this programming model for geo-replicated datastores ( §III); • a comprehensive evaluation, comparing our approach to state-of-the-art data stores ( §V).

II. CURE OVERVIEW

A. Transactional Programming Model

A body of research has extended the causal+ consistency (CC+) model [? ] by adding multi-key operations. There are two major efforts in this direction: read-only transactions [? ? ? ? ? ] that provide clients with a consistent view of multiple keys, and write-only transactions [? ] that permit clients to perform atomic multi-key write operations.

Unfortunately, strictly separating these two multi-key operations limits the flexibility of the programming model. For instance, it is a common practice to read an object before updating it. By separating these operations, concurrent updates over the same object may interleave. Consequently, programmers would have to deal with these anomalies at the application level, adding complexity.

We introduce a novel Transactional Causal Consistency (TCC) model, that explores a new spot along the consistency spectrum; the strongest semantics a system can achieve under high-availability and low-latency. Cure's TCC extends CC+ functionality by adding interactive transactions and CRDT support to ensure replica convergence.

Interactive transactions allow programmers to combine read and write operations flexibly within the same transaction, ensuring the following properties:

• Transactions read from a causally consistent snapshot, which represents a view of the data store that includes the effects of all transactions that causally precede it [? ? ]. For any pair of objects a and b, with a i , the version i of a, and b j , the version j of b, belong to the same causally consistent snapshot if there does not exist a k , another version of a such that (i) a k is created after a i , and (ii) a k is causally before b j . • A transaction updating multiple objects respects atomicity, i.e., all updates occur and are made visible simultaneously, or none does. Applying updates in this fashion creates a new causally consistent snapshot of the store.

CRDTs are high-level data types that can be replicated and modified concurrently while guaranteeing that replicas converge. CRDT abstractions such as counters, sets, maps/ tables and sequences provide a richer interface for programmers than the usual key-value interface with last-writer-wins (LWW) conflict resolution, which is the default policy used by previous works [? ? ? ? ? ]. Comparison to other models. Table I compares TCC with the consistency models implemented by other storage systems.

Serializability (SER) is the strongest isolation level defined by the SQL standard, providing a model equivalent to a sequential database.

Snapshot Isolation (SI) [? ] is the isolation level provided by popular commercial databases such as SQL Server and Oracle. Under SI, a transaction reads from a snapshot and updates are applied respecting total order. Concurrent transactions conflict only if they update the same data items. Two concurrent transactions, t 1 and t 2 , are not serializable when t 1 reads an object written by t 2 and t 2 reads an object written by t 1 . This anomaly is called write skew or short fork (as the state of the database inside a transaction is like a fork from the database state).

Parallel Snapshot Isolation (PSI) [? ] and Non-Monotonic Snapshot Isolation (NMSI) [? ] are weaker forms of SI that allow the long fork anomaly, where it is possible that two concurrent transactions t 1 and t 2 commit, writing to different data items, and then two other transactions that start subsequently, where one sees the effects of t 1 but not t 2 , and the other sees the effects of t 2 but not t 1 . Still, PSI and NMSI avoid write/write conflicts by aborting transactions that attempt to update the same items concurrently.

Our consistency model, TCC, also allows short and long fork. However it adds a powerful new feature: convergent forks, where concurrent transactions can modify the same data items, with concurrent updates being merged using CRDTs 1 .

TCC-S is the model provided by Eiger [? ], which combines CC+ with read-only and write-only transactions (COPS [? ], GentleRain [? ], Orbe [? ] and ChainReaction [? ] provide a similar model but only support read-only transactions). Unlike TCC, TCC-S does not support interactive transactions that include reads and writes, but only readonly or write-only transactions. Additionally, convergence is achieved using last-writer-wins rules.

Finally, eventual consistency (EC), used in systems such as Dynamo [? ] and the commercial Riak database [? ], includes no support for transactions.

B. Design

We assume a geo-replicated key-value store that handles a large number of objects. The full set of objects is replicated Stronger ← - ----------------- across different data centres (DC) to provide availability and low latency. Each of the data centres has N partitions, where each partition stores a non overlapping subset of the keyspace. We assume that every data centre employs the same partitioning scheme.

The challenge in the design of Cure consists of providing TCC under such a setting without compromising availability, while still achieving high scalability. To meet this goal, our protocol adopts three key design decisions. First, our protocol relies on the time-stamps of events to encode the causal dependencies which allow partitions to take decisions locally. This avoids using explicit dependency check messages, which substantially penalize performance [? ]. Second, the protocol does not rely on centrally assigned time-stamps [? ], as used in other systems. Third, the protocol decouples the problem of propagating updates among replicas from the problem of making these updates visible. This allows partitions to propagate updates pairwise without requiring coordination with other partitions. A lightweight protocol involving all partitions runs asynchronously to establish the set of transactions that are up to date, i.e. all causal dependencies are known, in a given data centre.

C. Programming interface

Cure's interface offers the following operations:

• TxId ← START_TRANSACTION(CausalClock)
initiates a transaction that causally depends on all updates issued before CausalClock. It returns a transaction handle that is used when issuing reads and updates. When an application does not specify the CausalClock, the most recent version available in the data centre is used. TxId) returns the list of values that correspond to the state of the objects stored under the Keys in the version given in the transaction's snapshot. discards the updates and aborts the transaction.

• Values ← READ_OBJECTS(Keys,
Under this programming model, a user issues transactions where each operation is defined by the CRDT specification.

III. PROTOCOL DESCRIPTION

Cure keeps multiple versions of each object in order to serve requests from causally consistent snapshots. Each version stores its value along with the metadata that encodes its causal dependencies. Old versions are periodically garbage collected by the system.

Our protocol assumes that each partition is equipped with a physical clock. Clocks are loosely synchronized by a time synchronization protocol such as NTP [? ]. Each clock generates monotonically increasing timestamps. The correctness of the protocol does not depend on the synchronization precision, though clock skew between servers can impact performance.

Cure annotates updates with the transaction commit time; a vector clock with an entry per DC. Commit times produce a partial order that respects causal consistency. The protocol uses these commit times to make transactions visible in accordance with causality. Transactions originating at the local DC are immediately visible to clients when they commit, as their causal dependencies are automatically satisfied. In contrast, transactions arriving from remote DCs depend on the globally stable snapshot (GSS), which represents a view of the data store that is known to be available at every partition of the local DC; they are only made visible when the GSS advances past their commit timestamp. This ensures that all causally related transactions, which have a smaller commit timestamp, are already visible locally.

A. Notation and definitions

Table II introduces the notation followed in this section. We assume a total number of D DCs. A partition m at DC d, denoted by p m d , keeps the following state: II).

• pvc m d ,

B. Transaction Execution

Algorithms 1 and 2 show the pseudocode of the protocol for executing transaction T at DC d followed by the transaction coordinator (T C T ) and the involved partitions, respectively. Start transaction. The transaction coordinator T C T starts by ensuring that it assigns T a snapshot no older than the last one seen by the client, represented by cvc (Alg. 1, line 3). This is necessary to ensure that clients observe monotonically increasing causally consistent views of the data store.

To define the causally consistent snapshot T will access, it sets the vector clock svc T to include all remote transactions that are stable in the local DC plus all locally committed transactions. The former is achieved by setting the vector to the value of GSS m d (Alg. 1, line 4), while the latter is achieved by setting the entry for the local DC in svc T to the maximum of either the physical clock of the server or the client's previously observed timestamp (Alg. 1, line 5). The snapshot must include the updates of all transactions that have a commit vector clock smaller than or equal to svc T . This guarantees that the snapshot is causally consistent, since it includes the dependencies of all transactions. Update objects. The client provides a list of key-update pairs, which T C T buffers in a per-partition write set (ws T [m]) to be sent, when committing T , to each updated partition at DC d. Read objects. The client provides a list of keys that she wants to read. T C T forwards, for each of the requested keys, a read request to the local replica that stores it (retrieved by the call to the PARTITION function). Upon receiving such request, and in order to respect the snapshot defined by svc T , i.e., that the snapshot includes all updates with commit time smaller than svc T , p for all Key, Operation ∈ U pdates do 10:

p i d ← PARTITION(Key) 11: if p i d / ∈ UpdatedPartitionsT then 12: UpdatedPartitionsT ← UpdatedPartitionsT ∪ {p i d } 13:
wsT

[i] ← wsT [i] ∪ {
GSS m d [k] ← min i=1...N P M C m d [i][k]
commit timestamp. Next, it stores its write set in its log. T C T computes the transaction's commit timestamp as the maximum of all proposed prepare timestamps (Alg. 1, line 31), and generates ct T , the commit vector clock of T , by applying this commit time, at position d, to svc T . The coordinator then sends a commit message to all involved partitions that includes the transaction's commit vector clock. When a partition receives the commit message, it removes T from prepT x m d , stores the ct T in its log, and adds T and its commit timestamp to committedT x m d for propagating its updates to the other DCs.

Choosing the maximum proposed timestamp as the commit timestamp of a transaction is important for correctness [? ]. The read protocol waits for prepared transactions expected to be included in a snapshot (Alg. 2, line 2). If T C T were to choose a ct smaller than the prepare timestamp of a participant partition, a transaction reading from the partition with svc smaller than the prepare timestamp but greater than this ct would not be delayed to include the committing transaction. Therefore, it would read an incorrect version.

C. Replication to remote DCs

Each partition periodically synchronizes with sibling partitions in other DCs. When there are no new updates to send, a heartbeat is sent. On receiving a heartbeat (Alg. 2, line 36), a replica advances pvc m d [k] stating that it has received all updates from DC k until the received timestamp. When there are updates to send, a replica sends all committed updates with timestamp smaller than any prepared but not yet committed transaction2 (Alg. 2, lines 27-29). On receiving an update replication message from DC k, a replica inserts the received updates in its log and advances pvc m d [k], setting it to the update's commit timestamp ct T [k].

Our algorithm decouples propagating updates among replicas from making these updates visible. An update received from a remote replica is only made visible after it is known that all updates from the same transaction (and their dependencies) have already been received. To this end, partitions in each DC exchange their pvc d vectors in the background (Alg. 2, line 39), and each partition m computes its GSS m d as the aggregate minimum of known pvc d (Alg. 2, line 43).

IV. DISCUSSION

A. Why vector clocks?

Cure uses a stabilization protocol for making updates visible while respecting causal consistency. In the work that introduced the GentleRain protocol, Du et al. [? ] showed that this approach improves throughput over using explicit dependency check messages, which also incurs large dependency metadata. The protocol underlying GentleRain tracks causal dependencies using a single scalar timestamp. Although this provides a compact representation of dependencies, it penalises update visibility latency at remote sites, and limits progress in the presence of network partitions between DCs. Cure's global stabilization mechanism relies on a vector clock sized by the number of DCs in the system. In the following paragraphs, we discuss these problems and how the use of vector clocks reduces their unwanted effects. Remote update visibility latency. We define the visibility latency of an update operation as the amount of time that passes between the moment it is committed at its local DC, and the time at the receiving partition at a remote DC when the partition makes that update visible to be safely read without violating causality. The use of a single scalar penalizes GentleRain because in order to make a remote update visible locally, the partition must wait until it receives a heartbeat with a time greater than that of the update from all other servers at all other DCs. In other words, in GentleRain, update visibility latency at a DC is dependent on the latency to the furthest DC.

By using a vector clock to timestamp events, Cure is able to make a remote update from some DC d with a commit vector clock ct T visible when the servers in DC k have received all updates up to ct T [d] from replicas at DC d, i.e.,

GSS p k [d] ≥ ct T [d].
Therefore, similarly to implementations relying on dependency check messages, Cure incurs update visibility latency dependent on the latency to the DC where the update originated. The cost of this improvement is an increased meta-data size over the use of a single scalar, which we find reasonable for geo-replicated data stores, normally deployed at a small number of DCs. Progress in the presence of failures. In the presence of a DC failure or network partition between DCs, relying on a single scalar timestamp means remote updates can no longer become visible. using a single scalar timestamp translates into stopping to make updates from any remote DC visible. In this situation, the state that a DC is able to observe from remote DCs remains frozen until the system recovers from the failure, while local updates continue to be made visible. In contrast, by using a vector clock, updates in Cure from healthy DCs are made visible continously even under network partitions, and only updates from the failed or disconnected DC remain frozen until the system recovers.

B. Session Guarantees

Cure ensures that transactions in a client session see (i) the effects of previously committed transactions by the same client, and (ii) monotonically-increasing snapshots of the data store. When a client finishes a read-only transaction, its snapshot vector clock is returned. Similarly, when a client successfully commits an update transaction, its commit vector clock is returned. The client vector clock cvc must accordingly be updated to the value returned by the last transaction completed. When a client starts a new transaction, it sends cvc with its request. If cvc is greater than the GSS at the server receiving the request, it is blocked until GSS proceeds past cvc. Otherwise, it starts immediately.

C. Efficient GSS computation

Under Cure, partitions within the same DC periodically exchange their pvc to compute their GSS. To do this efficiently, Cure builds a tree over all servers in a DC and computes an aggregate minimum using the tree [? ].

When compared to a simple broadcast approach, this reduces the number of messages exchanged in the network, while computing and distributing GSS in a reasonable amount of time. This is important for remote update visibility latency as updates from remote transactions are only made visible after the updated GSS exceeds commit vector clock.

D. Garbage Collection

In Cure, partitions periodically exchange the oldest snapshot vector clock of its active transactions and compute the aggregate minimum. Once a partition computes this minimum, it can safely remove from its log the versions of its stored objects that are no longer required, i.e., that no transaction will request in the future.

E. CRDT support

Cure relies on operation-based CRDTs [? ]. Their implementation requires adequate support from the system, as an object's value is defined not just by the last update, but also by the state it is applied on. This requires causal consistency to ensure that updates are neither lost nor delivered out-oforder. Hence, Cure encodes with each update a vector clock that represents the state to which it applies at a remote DC. As high-level updates are often not idempotent, safety also demands that updates are applied exactly once. Consider, for instance, an increment operation over a counter being sent from its originating DC to a remote one. If the network replicated this message, applying the increment twice may leave the system in an inconsistent state. To avoid such scenarios, the updates are assigned unique identifiers and the log filters the duplicates when they are inserted.

F. Correctness

We provide an informal proof that Cure implements TCC by showing that the snapshot read by a transaction is causally consistent and respects the atomicity of committed transactions.

Proposition 1. If an update u 1 depends on an update u 2 , then u 2 .ct < u 1 .ct.

An update u 1 depends on u 2 if the transaction of u 2 reads from a snapshot that contains u 1 . From Alg. 2 line 7, the proposed timestamps are always greater than its snapshot time (in DC d, the entry d of its snapshot vector clock). Since the commit timestamp is generated as the maximum of proposed timestamps, the commit time of a transaction is always greater than its snapshot time. Then, by Alg 1 lines 32-33, the commit vector clock of an update is always greater than its snapshot vector clock. Proposition 2. A partition vector clock pvc m d = t implies that p m d has received all updates with commit vector clock ≤ t.

First, we show that the proposition is valid for remote updates. We prove this by contradiction. Assume there is a remote update u from DC j such that u.ct < t, and p m d has not received u. By Alg. 2 lines 33-34, the partition would have received an update u 1 such that u 1 .ct[j] = t[j]. Because the updates are sent in the order of their timestamps, the partition cannot receive another update u 1 before u if u 1 .ct[j] > u.ct[j]. Hence u.ct[j] > t[j], implying u.ct < t, leading to the contradiction. Now we show that there are no pending local updates with commit vector clock ≤ t. When updating pvc[d], the partition finds the minimum prepared time stamps of the transactions in the prepared phase. Since the physical clock is monotonic and the commit time is calculated as the maximum of all prepared times, it is guaranteed that all future transactions will receive a commit time which is greater than or equal to this minimum prepared time stamp. So, when the pvc[d] is set to the minimum prepared time minus 1 (Alg. 2, line 20), the partition has already received all updates for the snapshot pvc. Proposition 3. Reads return values from a causally consistent snapshot.

Proposition 1 and 2 and the delaying of reads until pvc includes the snapshot time (Alg 2 line 2), together guarantees that by including all updates in a partition which have a commit vector clock ≤ snapshot vector clock, the read returns values from a causally consistent snapshot. This is true even when reading from multiple partitions because it reads from the same snapshot. Proposition 4. Reading from a snapshot respects atomicity.

Atomicity is not violated even though updates (local and remote) are made visible independently by each partition. All updates from a transaction belong to the same snapshot because they receive the same commit vector clock. A read is delayed until the same snapshot is available in all (accessed) partitions, thus reading all or no updates from a transaction.

Cure implements TCC, as every transaction reads from a causally consistent snapshot (Proposition 3) that includes all effects (Proposition 4) of its caually dependent transactions.

V. EVALUATION

A. Setup

We build Cure on top of Antidote [? ], an open-source reference platform that we have created for fairly evaluating distributed consistency protocols. The platform is built using the Erlang/OTP programming language, a functional language designed for concurrency and distribution. To partition the set of keys across distributed physical servers we use riak-core [? ], an open source distribution platform using a ring-like distributed hash table (DHT), partitioning keys using consistent hashing. Key-value pairs are stored in an in-memory hash table with updates being persisted to an on disk operation log using Erlang's disk-log module.

In addition to Cure, we have implemented eventual consistency, Eiger and GentleRain protocols for comparison. Our implementation of Eventual consistency is a singleversioned key-value store, supporting LWW registers, where the ordering of concurrent updates is determined by local physical clocks, and state-based CRDTs [? ]. Eiger supports causal consistency using LWW registers and tracks one-hop nearest dependencies, requiring explicit dependency checks. GentleRain supports causal consistency using a global stable time mechanism, requiring all-to-all communication for updates from external DCs to become visible (local updates are visible immediately). In addition to LWW registers, Cure supports operation-based CRDTs. All three causally consistent protocols provide (static) read-only and atomic update transactions. Cure additionally supports interactive read and update transactions.

Objects in Cure, Eiger, and GentleRain are multiversioned. For each key, a linked-list of recent updates and snapshots is stored in memory. Old versions are garbage collected when necessary. An update operation appends a new version of the object to the in-memory list and asynchronously writes a record to the operation log. If a client requests a version of an object that is no longer available in memory then it is retrieved from the operation log on disk.

Hardware. All experiments are run on the Grid5000

[? ] experimental platform using dedicated servers. Each server consists of 2 CPUs Intel Xeon E5-2630 v3, with 8 cores/CPU, 126GB RAM, and 2 558GB hard drives. Nodes are connected through shared 10Gbps switches. The measured latency within the cluster was aproximatelly 0.15 ms. Before running each experiment, clocks were synchronized through an NTP [? ] server running within the cluster.

Experiments are run using a variable number of DCs, each comprised by a variable number of servers, located within a Grid 5000 cluster at Rennes. Nodes within the same DC communicate using the distributed message passing framework of Erlang/OTP running over TCP. Connections across separate DCs use ZeroMQ [? ] sockets running TCP, with each node connecting to all other nodes to avoid any centralization bottlenecks. To simulate the DCs being geodistributed, we added a 50ms delay to all messages sent over ZeroMQ. Lost messages are detected at the application level and resent.

Workload generation. The data set used in the experiments includes 100k key-value pairs per partition with each pair being replicated at all 3 DCs. Tests are performed with LWW registers and CRDT sets.

A custom version of Basho Bench [? ] is used to generate workloads with clients repeatably running single operation transactions of either a read or an update using a power law distribution over all keys. The ratio of reads and updates is varied depending on the benchmark. For Cure, Eiger, and We found this ratio to sufficiently load the system without overstressing it for any workload. Each client server uses 40 threads to send requests at full load. Each instance of the benchmark is run for two minutes with the first minute being used as a warm up period. Google's Protocol Buffer interface is used to serialize messages between Basho Bench clients and Antidote servers.

B. Cure's scalability

To evaluate the scalability of Cure we run a DC configuration and vary the number of servers from 5 to 25. Then, we run 2 and 3 DCs configuration with 25 servers each (50 and 75 Cure servers in total). In both cases the read/update ratio is varied from a 99 percent read workload to a 50 percent update workload. Objects are LWW registers of 1 kbyte size values.

As Figure 1 shows, throughput increases 4.8 times when going from 5 to 25 nodes within a single DC under all workloads. Furthermore, on the configurations of 2 and 3 DCs consisting of 25 servers each, we observe a 1.8x and a 2.8x respective increase for 99 percent reads, and 1.8x and 2.6x for 50 percent writes when compared to a single DC with the same number of servers.

The observed scalability is expected due to the decentralized design of Cure. Still, numbers do not show a perfect linear progression due to the cost of replicating updates across DCs and because the background stable time calculation becomes more expensive as the number of servers increases per DC.

For this experiment, the median latency for reads was 0.7 ms for all workloads, and grew between 1 and 2ms for writes when increasing the update rate. Writes are more expensive than reads as they require updating in memory data structures and writing to disk. Additionally, given that updates are replicated 3 times, they create a larger load on the system than reads.

Impact of waiting. In order to evaluate the impact of clock skew in performance, we implemented an unsafe version of Cure that avoids waiting for a snapshot to be ready at a partition receiving a read request (Alg. 2, line 2). The No clock wait bar shows the throughput obtained by this protocol when run at a single DC consisting of 25 servers, which displays up to 1.25x increase when compared to the correct implementation, corresponding to the read dominant workload.

C. Comparison to other systems

To evaluate the performance of Cure when compared to other protocols we run a 3 DC benchmark with 25 servers per DC, varying the update to read ratio. We compare all systems using LWW registers, and Cure to eventual consistency using CRDT sets. Figure 2 shows our results. LWW registers. We compare all systems using LWW registers of 1 kbytes values each (Figure 2a). Unsurprisingly, eventual consistency performs better than all other protocols, outperforming Cure by approximately 30 percent across all workloads. Under eventual consistency, reads and updates are cheaper as they are single versioned and do not require causal dependency processing.

In the 99 percent read workload, causally consistent systems perform similarly, with Cure slightly outperforming the rest. This happens because at this ratio, the amount of dependency checks performed by Eiger is small. As soon as the update rate is increased to 10 percent, the cost of explicitly checking dependencies increases dramatically and the throughput of Eiger is negatively affected. This trend continues and remains throughout the higher update rate workloads. Cure and GentleRain perform similarly under all workloads. We expect a small additional overhead for GentleRain, which normally needs to retrieve slightly older versions of objects than Cure because of its larger remote update visibility latency, thus incurring extra processing of lists of object versions. CRDT sets. We compare Cure to eventual consistency systems using CRDT sets (Figure 2b). Being causally consistent, Cure supports operation based CRDTs, where objects transfer updates among replicas. On the contrary, eventual consistency can only support state-based CRDTs, where replicas synchronise by transferring object state. For this experiment, we use "small" and "big" sets that grow up to 10 and 100 elements of 100 bytes each (1 and 10 kbytes in total), respectively. Once sets reach this size, the workload balances the amount of add and remove operations to keep their average size constant.

For both sizes of sets, we observe a similar behavior. For 99 percent reads, eventual consistency outperforms Cure by a similar amount as with LWW registers. This is expected given that under this read-dominant workload, as updates are rare, systems behave similarly to the previous experiment, and therefore the same conclusions hold. For 90 percent reads, this difference becomes smaller. Finally, with higher update rates, Cure overtakes eventual consistency's performance. This is due to eventual consistency having to process large CRDT state transferred by each update operation at remote DCs (1 and 10 kbytes for small and big sets respectively). Under Cure , replicas transfer light operations (100 bytes when performing an add operation). Update visibility latency. To calculate the stable time, each node within a DC broadcasts its vector clock to other nodes within the DC at a frequency of 10ms. Additionally, heartbeats between DCs are sent at a rate of 10ms in the absence of updates.

For all experiments, we measured the remote visibility latency observed by DC 1 for updates coming from DCs 2 and 3. Under Cure we observed an average remote update visibility latency of between 80 and 90 ms for updates originating at DCs 2 and 3. Under GentleRain, we observed a visibility latency of 90 ms for both DCs 2 and 3. Moreover, under the update-intensive workloads, we observed frequent short lived peaks of around 150 ms visibility latency for one or both of the DCs due to the cost of processing external updates. Under such conditions, Cure only observed that delay for updates from the affected DC while under GentleRain, the visibility latency of both DCs was penalized under load. The use of a single scalar penalizes GentleRain, which is able to make updates visible at the rate of the slowest DC ( §IV-A). By using a vector clock, Cure is able to make updates coming from different DCs visible independently.

VI. RELATED WORK

A large amount of research has explored the consistency vs. availability tradeoff involved in building distributed data stores. As a result, there are a number of systems providing different semantics to application developers. On one extreme of the spectrum, strongly-consistent systems [? ? ? ] offer well-defined semantics through a simpleto-reason-about transactional interface. Unfortunately, due to the required intensive communication among parties, these solutions penalise latency in the general case and availability in the presence of failures and network partitions. On the other side of the spectrum lies eventual consistency, which includes Dynamo [? ], Voldemort [? ], Riak [? ] and some configurations of Cassandra [? ]. These systems offer excellent scalability, availability and low-latency at the cost of providing a model to programmers that is hard to reason about. They lack clear semantics and programming mechanisms (as transactions) that simplify application development. Cure takes an intermediate position in this tradeoff by embracing transactional causal+ consistency semantics.

Recently, a number of causally-consistent, partitioned and geo-replicated data stores were proposed [? ? ? ? ? ]. These solutions offer a variety of limited but interesting transactional interfaces that aim at easing the development of applications. COPS [? ] introduced the concept of causallyconsistent read-only transactions, which other solutions, such as ChainReaction [? ], Orbe [? ] and GentleRain [? ], adopted. Eiger [? ], extended this transactional interface with causally-consistent write-only transactions. Cure provides programmers with stronger semantics, i.e., general transactions and support for confluent data types (CRDTs).

In order to decide when remote updates can be made visible, COPS, Eiger, ChainReaction and Orbe use mechanisms that rely on piggybacking causal dependency information with updates and exchanging explicit dependency check messages among partitions at remote data centres. Even when they employ various optimisations to reduce the size of dependencies and the number of messages, in the worst case their metadata grows linearly with the number of partitions [? ]. GentleRain avoids such expensive checks. Instead, it uses a global stabilisation algorithm for making updates visible at remote data centres. This algorithm achieves throughput close to eventually consistent systems, at the cost of increased remote update visibility latency. Cure follows this design choice and achieves similar throughput while providing stronger semantics. Furthermore, by using a vector clock sized with the number of data centers, our protocol is able to reduce remote update visibility latency and is more resilient to network partitions and data centre failures.

Finally, SwiftCloud [? ] addresses the challenge of providing causally consistent guarantess and fault-tolerance for client-side applications. Although the semantics provided by SwiftCloud are similar to ours, this work is orthogonal to Cure, since our focus is on making server-side causally consistent systems with rich semantics highly scalable, a problem that is not tackled by SwiftCloud.

VII. CONCLUSION

We have introduced Cure, a distributed storage system presenting the strongest semantics achievable while remaining highly available. Cure provides a novel programming model: causal+ consistency and CRDT support through an interactive transactional interface.

We have presented a highly-scalable protocol implementation over a partitioned geo-replicated setting. We have evaluated Cure showing that it presents scalability compatible with eventual consistency with both the number of servers per DC and the total number of DCs in the system, while offering stronger semantics. Our results also show that, when comparing Cure to existing causally-consistent systems that provide similar but weaker semantics under different workloads, it presents higher performance while achieving better update visibility latency and tolerance to full DC and network failures.

d

  • ok ← UPDATE_OBJECTS(Updates, TxId) declares a list of Updates for a transaction. • CommitTime ← COMMIT(TxId) commits the transaction under transaction handler TxId Matrix of received pvc i d at p m d T Transaction T CT Transaction Coordinator of T svcT Snapshot vector clock of T ctT Commit vector clock of T wsT [m] Write set of T for partition m Table II: Notation used in the protocol description. and makes the updates visible. • ok ← ABORT(TxId)

  a vector clock of size D, where position k indicates that p m d has received updates up to pvc m d [k] from p m k ; • GSS m d , a vector clock of size D that denotes the latest globally stable consistent snapshot known by p m d , i.e., the snapshot that p m d knows to be available at all partitions of its DC. In order to advance GSS m d , partitions of the same DC periodically exchange their pvc. Each p m d computes its GSS m d as the aggregate minimum of known pvc d ; Clients connect to Cure servers to issue transactions. A server receiving a client request to start transaction T acts as its transaction coordinator (T C T ). T C T keeps the necessary state for executing T (See Table

2 : 5 : 8 :

 258 m d might need to wait for its pvc m d [d] to Algorithm 1 Transaction coordinator at server m of DC d 1: function START_TRANSACTION(cvc) for k = 1 . . . D, k = d do 3: wait until cvc[k] ≤ GSS m d [k] 4: svcT ← GSS m d svcT [d] ← MAX(cvc[d], Clock m d ) function UPDATE_OBJECTS(T , U pdates) 9:

Commit. 6 : 8 : 9 : 13 : 26 : return 27 : 30 :

 68913262730 When receiving a commit request from a client, T C T starts a 2PC protocol to atomically commit the updates of transaction T at local DC d. In the first phase, T C T sends a prepare message including ws T [m] to each of the updated partitions (Alg. 1, lines[START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF][START_REF] Sumbaly | Serving large-scale batch computed data with project voldemort[END_REF][START_REF] Zawirski | Write fast, read in the past: Causal consistency for client-side applications[END_REF]. Upon receiving such message, each partition takes the current value of its physical clock (Alg. 2, line 8) and proposes it as the transaction's Algorithm 2 Protocol executed by partition p m d 1: function READ_KEY(svcT , Key) 2: wait until svcT [d] ≤ pvc m d [d] 3: V al ← SNAPSHOT(Key, svcT , Log m d ) 4: send V al to T CT 5: function PREPARE(T , wsT [m], svcT ) 7: wait until svcT [d] ≤ Clock m d P repT ime ← Clock m d Log m d ← Log m d ∪ { wsT [m], P repT ime, svcT } 10: prepT x m d ← prepT x m d ∪ { T, P repT ime } 11: send T , P repT ime to T CT 12: function COMMIT(T , ctT ) if committedT x m d = ∅ then 24: for k = 1 . . . D, k = d do 25: send HEARTBEAT(pvc m d [d], d) to p m k for all T, ctT ∈ committedT x m d | ctT < pvc m d [d] do 28: for k = 1 . . . D, k = d do 29: send REPLICATE_TX(wsT [p], ctT , svcT , d) to p m k committedT x m d ← committedT x m d \ { T, ctT } 31: 32: function REPLICATE_TX(wsT [p], ctT , svcT , k) 33: Log m d ← Log m d ∪ { wsT [p], ctT , svcT } 34: pvc m d [k] ← ctT [k] 35: 36: function HEARTBEAT(T imeStamp, k) for i = 1 . . . N, i = m do 41: send UPDATE_GSS(m, pvc m d ) to p i d 42: 43: function UPDATE_GSS(i, pvc) 44: P M C m d [i] ← pvc 45: for k = 1 . . . D, k = d do 46:

Figure 2 :

 2 Figure 1: Scalability of Cure

  → Weaker

	Properties	SER SI	PSI NMSI	Cure	TCC-S	EC
	Transactions	yes yes yes	yes	read-only write-only	no
	Short fork	x
	Long fork	x	x
	Convergent fork	x	x	x
	Convergence	-	-	-	CRDT	LWW	LWW
	Table I: Properties of consistency models (x: disallowed).

  Key, Operation } Alg. 2, line 2). Once this is satisfied, p m d returns the latest version of the object with commit time no newer than svc T , which is retrieved by calling the SNAPSHOT function (Alg. 2, line 3). Once T C T receives this reply, it applies the update operations on the same object (if any) issued by T during previous UPDATE_OBJECTS operations (Alg. 1, line 21), generating a new version of the object. Note that this is a consequence of providing more developer-friendly data types than just basic registers. T C T caches this result until all objects in the operation are read. This process is repeated for every requested key. Once it finishes, T C T returns all read values to the client.

	14:	return ok
	15:
	16: function READ_OBJECTS(T , Keys)
	17:	for all Key ∈ Keys do
	18:	p i d ← partition(Key)
	23:	return V alues
	24:
	25: function COMMIT(T )
	26:	if UpdatedPartitionsT = ∅ then
	27:	return svcT
	28:	for all p i d ∈ UpdatedPartitionsT do
	29:
		d
	31:	CommitTime ←MAX(all prepare times)
	32:	ctT ← svcT
	33:	ctT [d] ← CommitTime
	34: 35:	for all p i d ∈ UpdatedPartitionsT do send COMMIT(T , ctT ) to p i d
	36:	return ctT
	catch up (

19:

V al ← send READ_KEY(svcT , Key) to p i d 20: for all Key, Operation ∈ wsT [i] do 21: V al ← APPLY_OPERATION(V al, Operation) 22: V alues ← V alues ∪ {V al} send PREPARE(T , wsT [i], svcT ) to p i d 30:

wait until receiving (T , P repT ime) from p i

Walter [? ], an implementation of PSI, also supports a similar consistency model restricted to counters and sets.

A transaction being prepared with a given prepare timestamp can commit before a concurrent transaction with a lower one when they update different partitions.

ACKNOWLEDGMENTS

We would like to thank our shepherd, David Shue, the anonymous reviewers, Christopher Meiklejohn, Michał Jabczyński, Marek Zawirski and Sérgio Duarte for their helpful comments on a previous version of this work. This research is supported in part by European FP7 project 609 551 SyncFree (2013-2016). Nuno Preguiça is partially supported by FCT-MCTES NOVA LINCS project (UID/CEC/04516/2013). Zhongmiao Li and Manuel Bravo are partially supported by the Erasmus Mundus Doctorate Programme under Grant Agreement No. 2012-0030.