Observable atomic consistency for CvRDTs
Xin Zhao Philipp Haller








Observable atomic consistency for CvRDTs (Extended Version)


X. Zhao and P. Haller 


Xin Zhao and Philipp Haller

KTH Royal Institute of Technology, Stockholm, Sweden

{xizhao, phaller}@kth.se









Observable atomic consistency for CvRDTs
    Xin Zhao Philipp Haller
    December 30, 2023
========================================




The development of distributed systems requires developers to balance the need for consistency, availability, and partition tolerance. Conflict-free replicated data types (CRDTs) are widely used in eventually consistent systems to reduce concurrency control. However, due to the lack of consistent non-monotonic operations, the usage of CRDTs can be difficult.

In this paper, we propose a new consistency protocol, the observable atomic consistency protocol (OACP). OACP enables a principled relaxation of strong consistency to improve performance in specific scenarios. OACP combines the advantages of mergeable data types, specifically, convergent replicated data types, and reliable total order broadcast to provide on-demand strong consistency. By providing observable atomic consistency, OACP avoids the anomalies of related protocols.

We provide a distributed, cluster-enabled implementation of OACP based on Akka, a widely-used actor-based middleware. Our experimental evaluation shows that OACP can reduce the coordination between replicas compared to other protocols providing atomic consistency in several benchmarks. Our results also suggest that OACP gains availability through mergeable data types and provides acceptable latency for achieving strong consistency.






§ INTRODUCTION

Conflict-free replicated data types (CRDTs) <cit.> are widely used in industrial distributed systems such as Riak <cit.> and Cassandra. They are objects which can be updated concurrently without consensus and eventually converge to the same state if all updates are executed by all replicas eventually. Thus, they can provide high availability and scalability for replicated shared data. However, the main challenges are the fact that they only provide eventual consistency and they necessarily provide only a restricted set of operations. In particular, they do not support consistent non-monotonic operations; for example, read operations may return outdated values before the replicas have converged. This makes the usage of CRDTs difficult.

There are two principal approaches of CRDTs, one is operation-based CRDTs and the other is state-based CRDTs. Operation-based CRDTs are also called CmRDTs, and they send only the update operations which are commutative. State-based CRDTs are also called CvRDTs, and they propagate the whole state to other replicas which can be merged by a commutative function. There are many specifications for CRDT counters, sets, and registers.

In this paper, we focus on CvRDTs and we address these challenges by extending CvRDTs with on-demand strong consistency. We devise a novel observable atomic consistency protocol (OACP) which avoids some anomalies of other recently proposed consistency models.



  §.§.§ Contributions.

This paper makes the following contributions:


  * We introduce the observable atomic consistency (OAC) model which
  enables a novel extension of CvRDTs with non-monotonic
  operations. Thus, we lift a major limitation of CvRDTs, and
  significantly simplify programming with CvRDTs. We also provide a
  precise and formal definition of the OAC model.


  * We prove that systems providing observable atomic consistency
  are state convergent. The paper summarizes our definitions and
  results, while our companion technical report <cit.>
  contains the complete proofs.
  


  * We provide the observable atomic consistency protocol (OACP)
  which guarantees observable atomic consistency.


  * We provide a distributed implementation of OACP which is enabled
  to run on clusters. Our implementation is based on the widely-used
  Akka <cit.> actor-based middleware. The system is available
  open-source on GitHub.[See
    <https://github.com/CynthiaZ92/OACP>]


  * We provide an experimental evaluation of OACP, including latency, throughput and coordination. Our evaluation shows that OACP benefits more when there are more commutative operations and can reduce the number of exchanged protocol messages compared to the baseline protocol. Using the case study of a Twitter-like microblogging service, we experimentally evaluate optimization of OACP, which we call O^2ACP.


The rest of the paper is organized as follows. Section <ref> illustrates how application programmers use OACP. In Section <ref> we formalize the observable atomic consistency (OAC) model, and prove that systems providing OAC are state convergent. Section <ref> explains the observable atomic consistency protocol (OACP), and compares it to the Global Sequence Protocol <cit.> (GSP). In Section <ref> we present a performance evaluation of an actor-based implementation of OACP using microbenchmarks as well as a Twitter-like application. Section <ref> discusses related work, and Section <ref> concludes.



§ OVERVIEW

We provide a usage overview of the OACP system from the perspective of a programmer to provide a user-friendly interface towards distributed application development. First, we introduce the system structure, and then we demonstrate how to use the provided API through an example.



  
System structure.
Figure <ref> shows the system structure. There are three layers from bottom to top: storage, distributed protocol, and application. We first focus on the top layer and discuss how applications interface with the protocol.



The operations submitted by the application are divided into two categories: CvRDT operations (CvOps) and totally-ordered operations (TOps). CvOps are commutative. TOps are supported by reliable total order broadcast (RTOB) <cit.>, so that their ordering is preserved across the entire system.

The submission of a TOp causes all replicas to atomically (a) synchronize their convergent states, and (b) lock their convergent states. A replica with locked convergent state buffers CvOps until the original TOp has been committed. Moreover, at the point when a TOp is executed, all replicas are guaranteed to have consistent convergent states. Thus, submitting a TOp ensures the consistency of all replicas, including their convergent states.



  
User API.
The user-facing API consists of the following three parts.


    
  * trait 
    
  * CvOp (CvValue, CvUpdate)
    
  * TOp (TUpdate)

The  trait[A trait in Scala can be thought of as similar to a Java interface enabling a safe form of multiple inheritances.] exposes the implementation of CvRDT to developers which allows them to define their own CvRDT types and operations which are provided by the trait such as initialisation, add, remove and merge.

The last two methods provide access to use the OACP, allowing developers to decide whether the operation should be operated as CvRDTs or ordered messages. CvOps are used when one wants to gain benefits from high availability since they are directly sent to the closest available server. While TOps are good options if the developer wants to make all the replicas reach the same state for some essential operations. These two message types give developers more flexibility when they want to achieve certain consistency as well as the performance of the implementation. The following is an example showing how to use the interface.



  
Resettable counter.

The grow-only counter (GCounter) is one of the most basic counters which is widely used, e.g., in real-time analytics or in distributed gaming. It is a CvRDT which only supports increment and merge operations. However, when a system employs a GCounter to achieve eventual consistency, often a special “reset” operation is needed for resetting the counter to its default initial state. In other words, we need an “observed-reset” <cit.> operation for GCounter to go back to the bottom state, which means when “reset” is invoked, all effects of the GCounter observed in different replicas should be equivalently reset. However, the standard GCounter cannot solve this problem; this limitation is also well-known in the popular Riak DT implementation.

Thus, we need an implementation of a resettable counter to make sure that all the replicas are reset at the same time. A straightforward solution is to define “reset” as a totally-ordered operation, leveraging the property of TOps.

Figure <ref> shows a GCounter definition in Scala. We extend the  trait to have an instance of GCounter which supports operations such as  and . Each replica in the cluster is assigned an ID; this enables each  instance to increment locally. When merging the states of two GCounters, we take the maximum counter of each index. The  method is used to express the partial order relationship between different s.



Then the message handlers need to be defined on the client side so that OACP system can recognize the behavior. We extend the client actor by OACP protocol to connect application layer with OACP protocol. The developer only needs to define which message to send using the following “Akka-style” message handler. When the CounterClient receives a certain message, it behaves according to the user's definition.

[numbers=left, numberstyle=, backgroundcolor = ]
class CounterClient extends Protocol[GCounter] 
  val CounterClientBehavior: Receive = 
    case Incr  => self forward CvOp("incr")
    case Reset => self forward TOp("Reset")
    ...
  
  override def receive = CounterClientBehavior.orElse(super.receive)
  ...


For example, case  handles a CvOp which is forwarded to the protocol layer.



§ OBSERVABLE ATOMIC CONSISTENCY


We now formalize the consistency model of OACP.


Given a set of operations U = C ∪ T where C ∩ T = ∅, a CvT order is a partial order O=(U,≺) with the following restrictions:

  
  * ∀ u, v ∈ T such that u ≠ v. u ≺ v  v ≺ u
  
  * ∀ p ∈ C, u ∈ T. p ≺ u  u ≺ p
  
  



According to the transitivity of the partial order, we could derive that ∀ l, m, n ∈ U such that l ≺ m, m ≺ n. l ≺ n.



  Given a set of operations U = C ∪ T where C ∩ T =
  ∅, a Cv-set C_i is a set of C
  operations with the restriction that:

  
    
  * ∀ p, q ∈ C_i ⇒ p ⊀ q  q ⊀ p
    
    
  * ∀ p ∈ C ∖ C_i.∃ q ∈ C_i such that p ≺ q  q ≺ p
    
  





In Figure <ref>, the partial order is labeled with black dash arrow, while we could derive line 1 and 2 (red dash arrow) from the transitivity of ≺. There are two different Cv-sets in this situation while the CvRDT updates inside have no partial order relationship.

In CvT order, only the operations in one same Cv-set could happen concurrently. Now we consider the operations on different sites. Suppose each site i executes a linear extension compatible with the CvT order. Then, the replicated system with n sites provides local atomic consistency, which is defined as follows.


  A replicated system provides local atomic consistency (LAC) if each site i applies operations according to a linear extension of the CvT order.


The three replicas in Figure <ref> could have different linear extensions of the CvT order. However, despite possible re-orderings, LAC guarantees state convergence.






  A LAC system is state convergent if all linear extensions of the underlying CvT order O reach the same state S.



  Given a CvT order, if all operations in each Cv-set are commutative, then any LAC system is state convergent.



In order to give a complete proof for theorem <ref>, we first introduce the following lemmas and their proofs.

Given a legal serialization O_i = (U, <_i) of CvT order O=(U, ≺), if ∃ u, v ∈ U such that u <_i v and u ⊀ v, then ∃ ! C_i∈ U such that u ∈ C_i∧ v ∈ C_i.


Let's consider two operations in the following cases and find which situation satifies the requriment.

Case 1: if u ∈ T and v ∈ T, u <_i v then according to the definition, u ≺ v. 

Case 2: if u ∈ T and v ∈ C, u <_i v, there must ∃ C_i such that v ∈ C_i, since u ≺ (∀ c|c ∈ C_i, u ≺ v.

Case 3: if u ∈ C and v ∈ T, u <_i v, using the same argument from Case 2, we have u ≺ v.

Case 4: if u ∈ C and v ∈ C, u ∈ C_i∧ v ∈ C_j∧ C_i∩ C_j = ∅, u <_i v so that C_i≺ C_j, then we have u ≺ v.

Case 5: if u ∈ C_i and v ∈ C_i, u ∈ C_i∧ v ∈ C_i, we know from the definition that u ⊀ v.

Consider the above five cases, only case 5 can satisfy u <_i v ∧ u⊀ v, so the lemma is correct.



Assume O_i = (U,<_i) and O_j = (U,<_j) are both legal serializations of CvT order O = (U,≺) that are identical except for two adjacent operations a and b such that a <_i b and b <_j a and that all operations c ∈ C are commutive inside each single Cv-set. Then S(O_i) = S(O_j).



Let P and S be the greatest common prefix and suffix of O_i and O_j, then S_ab=S(P)+a+b, S_ba=S(P)+b+a.

From lemma <ref>, we know that only the CvRDT operations inside the same Cv-set can be the two adjacent operations a and b. The operation in one Cv-set commute, so S_ab = S_ba. And we could also derive that S_ab + S(Q) = S_ba + S(Q). Since S_ab + S(Q) = S_(O_i) and S_ba + S(Q) = S_(O_j) then we have S(O_i) = S(O_j)


Now we come back to the proof of Theorem 1.1.

Considering when:

(1) O_i = O_j. Then we have S(O_i) = S(O_j).

(2) O_i ≠ O_j, then there exists at least one pair of adjacent operations such as u and v that u <_i v and v <_j u. From lemma <ref> we know if we swap the sequence of u and v in O_i, we could get O_i+1 = O_i. If O_i+1≠ O_j, then we continue to find adjacent operation pairs in O_i+1 until we finally construct a sequence of operations which is equal to O_j.

Then we proved that any LAC system is state convergent.



  A replicated system provides observable atomic consistency (OAC) if
  it provides local atomic consistency and for all p ∈ C, u ∈
  T. (a) p ≺_PO u in program order (of some client) implies
  that p ≺ u in the CvT order, and (b) u ≺_PO p in
  program order (of some client) implies that u ≺ p in the CvT
  order.


The constraints for OAC guarantee that the system state is consistent with the order of operations on each client side. Since OAC is a particular case for LAC, the state convergence also holds for OAC, and we have the following corollary:


  Given a CvT order, if all operations in each Cv-set are commutative, then any OAC system is state convergent.




  
Discussion.
Observable atomic consistency (OAC) is a novel consistency model. It guarantees that invocations of totally-ordered operations (TOps) establish consistency across all replicas. Between two subsequent TOps there can be invocations of convergent operations (CvOps) which can be re-ordered as long as they happen between the two original TOps. We now compare our consistency model with previous notions.



  
*Comparison to RedBlue Consistency.

A closely related consistency model is RedBlue consistency <cit.>. Red operations are the ones which need to be totally ordered while the blue ones can commute globally. Two operations can commute means that the order of the operations does not affect the final result. In order to adapt an existing system to a RedBlue system, shadow operations with commutativity property need to be created and then adjust the application to use these shadow operations. The shadow operations can “introduce some surprising anomalies to a user experience” <cit.>. In other words, the final system state might not take all messages into account that were received by the different replicas. Thus, RedBlue consistency can only maintain a local view of the system. In OAC, the CvRDT updates can only commute inside a specific scope, namely, between two totally-ordered operations. This restricts the flexibility of CvRDT updates, but at the same time, it provides a consistent view of the state. Importantly, the final system state always matches the state resulting from a linear extension of the original partial order of the operations. An example comparing OAC and RedBlue consistency is included in our companion technical report <cit.>.


Here is an example for comparison. There are Alice in the EU and Bob in the US, performing some operations on the same account. Suppose the sequence of bank operations arriving at any of the replicas is as follows:
Alice(deposit(20)) → Bob(accrueinterest()) → Bob(withdraw(60)) → Bob(deposit(10)) → Alice(withdraw(70)).

RedBlue consistency requires the definition of withdrawAck' and withdrawFail' as shadow operations for an original “withdraw” operation. Then, withdrawAck' is marked as a red operation (i.e., totally ordered).

In RedBlue consistency, when Alice wants to execute withdraw(70) (blue frame in Figure <ref>(A)), there is an immediate response withdrawFail'() since it is a blue operation. In the local view of Alice, the deposit(10) from Bob is not visible yet. However, in the assumed arriving sequence of operations, deposit(10) from Bob was already received by one of the replicas; thus, in principle the withdraw could have succeeded.



Compared with RedBlue consistency, in OAC system, the evolution of the system state looks as shown in Figure <ref>(B). we only need to define which operations should be CvOps and which should be TOps. Meanwhile, when Alice wants to execute withdraw(70), since it is a TOp, the system will make all the replicas reach the same state to make sure Alice has a consistent view of the system (red frame in Figure <ref>(B)).



§ OBSERVABLE ATOMIC CONSISTENCY PROTOCOL (OACP)

After the definition of OAC in Section <ref>, now we denote the semantics of the protocol using a so-called data model together with the pseudocode.



 §.§ Data model

In this part, we describe the data models that are used in the protocol. Consider all the clients and servers as a set of objects. In general, we use notation Op to define the operations that are performed on objects. Op^* represents a sequence of operations.

Operation types are classified according to different names. Thus, the function Eval: Op^* × Op → Val presents the result when applying the sequence of operations on one object.

We define abstract type CvUpdate and TUpdate stand for CvRDT updates and totally-ordered updates, Value for result from reading the log, Read for an ordered read operation. Then we could give a data model using the function OACPVal: Read ×{CvUpdate, TUpdate}^* → Value. The protocol will be explained in a client-server topology.

In the following sections, we use Log to record all the sequence of operations and states. The unit of Log is called Entry, which contains the object information, number of next log index, CvRDT state cState as well as totally-ordered updates tUpdate. We use the denotation ++ to represent the combination of two sequences.



§ OBSERVABLE ATOMIC CONSISTENCY PROTOCOL (OACP)


Following the definition of observable atomic consistency <ref>, we now introduce a protocol that enforces this notion of consistency. We present the observable atomic consistency protocol (OACP) in four steps: first, we describe the so-called data model, similar to prior related work <cit.>; second, we describe the client-side protocol; third, we describe the server-side protocol; finally, we discuss differences to the most closely related previous protocol.



  
Data model.
Operation types are classified according to different names. Thus, the function 𝐸𝑣𝑎𝑙: Op^* × Op →𝑉𝑎𝑙 presents the result when applying the sequence of operations on one object. We define abstract type 𝐶𝑣𝑈𝑝𝑑𝑎𝑡𝑒 and 𝑇𝑈𝑝𝑑𝑎𝑡𝑒 stand for CvRDT updates and totally-ordered updates, Value for the result from reading the log, 𝑅𝑒𝑎𝑑 for an ordered read operation. Then we could give a data model using the function 𝑂𝐴𝐶𝑃𝑉𝑎𝑙: 𝑅𝑒𝑎𝑑×{𝐶𝑣𝑈𝑝𝑑𝑎𝑡𝑒, 𝑇𝑈𝑝𝑑𝑎𝑡𝑒}^* →𝑉𝑎𝑙𝑢𝑒. The protocol will be explained in a client-server topology.

In the following, we use the type 𝐿𝑜𝑔 to represent a log that records the (totally-ordered) sequence of all operations and states. Whenever a totally-ordered operation is committed, an entry is created and added to the log. Each log entry is of type 𝐸𝑛𝑡𝑟𝑦 which is defined as follows:


class Entry  origin: Client, nextLogIndex: N, cState: CvRDT, op: TOp 

Each entry contains a reference to the client that submitted the TOp (); the index of the next log entry (); the state of the underlying CvRDT (); and the committed totally-ordered operation (). We use the notation ++ to represent the concatenation of two sequences.



  
OACP client protocol.
 We now show the basic version of the OACP client-side protocol in Listing <ref>. When CvOp invokes, the client will send the update to a random available server; while TOp happens, the client will submit the update to the current leader according to the chosen consensus protocol (in our case is Raft <cit.>). The abstraction greatly simplifies the distinction between these two kinds of operations on the client side, so that it provides more flexibility for the extension on top of the client actor.

[numbers=left, numberstyle=, backgroundcolor = , caption= Client-side OACP., label = list:egsp-client]
role OACP_Client 
  var result: Promise[Value];
  var response: Promise[Value];

  // client interface
  CvOp(u: CvUpdate) 
    response := new Promise[Value];
    CRDT_submit(u); // send to random replica
    return response;
  

  TOp(msg: TUpdate | Read): Promise[Value] 
    result := new Promise[Value];
    RTOB_submit(msg); // send to leader replica
    return result;
  

  // network interface
  onReceive(log: Log) 
    result.complete(OACPVal(updates(log)));
  

  onReceive(response) 
    response.complete()
  

  function updates(l: Log): (TUpdate*, cState) = 
    return (l[0].tUpdate ++ ... ++ l[l.length-1].tUpdate, l.cState);
  


An important assumption of the protocol is that the client never invokes an operation before the promise of a previous invocation has been completed, especially when a TOp happens directly after a CvOp, the client needs an acknowledge of the previous message from the server side to invoke the TOp. In this case, the program order on the client side will also be preserved on the server side.



  
OACP server protocol.
We now move on to the server side of the protocol. We assume that the application has continuous access to the network. This mirrors the practical usage of many existing applications, such as chat and Twitter-like micro-blogging.

In Listing <ref>, 𝐶𝑅𝐷𝑇.𝑚𝑒𝑟𝑔𝑒 is an abstract merge operation for any type of CRDT; 𝐿𝑜𝑔.𝑛𝑒𝑥𝑡𝐼𝑛𝑑𝑒𝑥 is the next store index for 𝐸𝑛𝑡𝑟𝑦 according to Raft protocol; 𝑐𝑆𝑡𝑎𝑡𝑒𝐶𝑜𝑙𝑙𝑒𝑐𝑡() is a message for getting all the cState from other servers; 𝐵𝑟𝑜𝑎𝑑𝑐𝑎𝑠𝑡() is a message which will be sent to all the other servers in the cluster; 𝑅𝑇𝑂𝐵_𝑠𝑢𝑏𝑚𝑖𝑡 wraps an 𝐸𝑛𝑡𝑟𝑦 structure which contains the client, the number of current log index, the current monotonic state and the update from the client, add the 𝐸𝑛𝑡𝑟𝑦 in the log and broadcast to all the other servers to keep consensus on each replica.

When the server receives the CvOp, it will merge the current CvRDT state and broadcast the change to all the other servers; when TOp is received, the leader server will collect the current states from all the replicas and make an RTOB so that each replica keeps the same log. RTOB in OACP system is implement by Raft consensus protocol, and each server will do commit when they get confirmation from the leader server. When 𝑇𝑈𝑝𝑑𝑎𝑡𝑒 and 𝑅𝑒𝑎𝑑 handled by non-leader server, it will forward the update to the current leader in the cluster. Moreover we define 𝑅𝑒𝑎𝑑 as TOp to guarantee for getting the newest result directly from the leader.

In the protocol description, we set up a special flag “frozen”. When a TOp is taking place, the flag will turn to true for each replica. For coming operations after that, the system will stash all the updates without changing the current state to keep OAC for all replicas.

[numbers=left, numberstyle=, backgroundcolor = , caption = Server-side OACP (unoptimized)., label = list:egsp-server]
role OACP_Server 
  var currentState: CRDT;
  var log: Log;
  var currentLeader: Server;
  var frozen: Boolean;
  var result: Promise[Value];

  onReceive(u: CvUpdate) 
    if frozen then  buffer.stash(u); 
    else 
      currentState = CRDT.merge(currentState, u);
      Broadcast(currentState);
      client.reply(); //acknowledge to client
    
  

  onReceive(msg: TUpdate | Read) 
    if currentRol.isLeader        frozen then  stash(msg); 
    else if (currentRole.isLeader) 
      frozen = true;
      numStateMsgReceived = 0;
      result = new Promise[Value];
      result.onSuccess  v => client.reply(v);
      Broadcast(GetState);
    
    else forward(currentLeader, msg);
  

  onReceive(msg: GetState) 
    frozen = true;
    reply(StateIs(currentState));
  

  onReceive(msg: StateIs) 
    if currentRole.isLeader then 
      numStateMsgReceived += 1;
      currentState = CRDT.merge(currentState, msg.cState);
      if numStateMsgReceived == numReplicas-1 then 
        RTOB(new Entry 
            origin = msg.sender,
            number = Log.nextIndex(log),
            cState = currentState,
            toUpdate = msg );
        Broadcast(Melt);
        result.complete(log);
      
      if timeout then  // fault handler
        RTOB(new Entry 
            origin = msg.sender,
            number = Log.nextIndex(log),
            cState = Log.cState,
            toUpdate = Recovery );
        Broadcast(Melt);
        result.complete(failure);
        
      
    
  

  onReceive(msg: Melt) 
    frozen = false;
    buffer.unstash() or discard();
  



When leader receives acknowledges from majority number of servers, a “Melt” message will be sent around to set the “frozen” flag to false again. If there are n servers in the cluster, then there will be 2(n-1) messages adding to the whole protocol. Thus we come up with the optimized version (see List <ref>) to turn the flag to false each time when a server makes a commit in the consensus protocol.

[numbers=left, numberstyle=, backgroundcolor = , caption = Server-side OACP (optimized), label = list:egsp-server-op]
  onCommit(e: Entry) 
    frozen = false;
    buffer.unstash() or discard();
  




  
Optimisation considering the sequence of operations
Then we go deeper to the inside of OACP. It’s the combination of two types of systems, one is the representative of on-demand strong consistency and one is on behalf of a popular way to achieve strong eventual consistency. Taken the low latency from CRDT together with the consensus from Raft, OACP takes the advantage of both systems. When the two systems need to interface, we need to think about how the system behave.

Now let’s discuss about the how the system behaviour change when two different types of operations happen in certain sequence. There’re four situations which shows in Table <ref>.


We could illustrate the above chart as follows: (1) CvOp will use CvRDT property to operate so that the order of the operation can be disordered, the system is behaving like the CvRDT system. (2) When two TOps occurs, then the system will behave according to raft protocol. (3) When a TOp happens after the CvOp, first the system need to stop receiving any coming CvOps ("Frozen" state) and then using elected leader to decide the total order for the current state. (4) When a CvOp wants to take effect after the TOp, then it should make the system "Melt" first, and then behave like what described in (1). The interesting aspect we observe when we make this classification is that in situation (3) and (4), we need to deal with the shift state ("Frozen" ↔ "Melt") of the system. It can be optimised by introducing another parameter called "auto-melt". The property of "auto-melt" is that once defined, the system will decide whether to melt the system automatically. Now le'ts have a look at two examples about how the argument works in different selections.

In the shopping cart example, when we checkout at certain point, the user could still be able to do add or remove operations to another shopping cart. These following MonOps will not be considered for this specific "checkout", but after the checkout, the updates should be able to take effect on the system. At this point, we need to use auto-melt so that the CvOps can take effect after certain TOps. In another example such as a simple bidding system, we could define all the "Bid" operations as CvOps and the final "WinnerCheck" is the TOp. When the "WinnerCheck" is called, the system will freeze as usual but the following "Bid" will be simply dropped. Another bid need to be started manually by calling a special "Melt" message to the system.




  
*Fault Model. During the process of RTOB, the failure of the system will be handled by Raft protocol where n nodes crash failure will be tolerated in a 2n+1 nodes cluster. During the state gathering phase (See Line 38, 47), if the receiving time of leader exceeds the setup timeout duration, then failure recovery strategy is to retrieve the cState in the last log entry and synchronize all the replicas. The OACP protocol provides a neat solution to avoid staleness read in CvRDTs.




  
*Comparison to GSP.
GSP is an operational model for replicated shared data. Figure <ref> shows the abstraction of GSP protocol. It supports update and read operations from the client. Update operations are stored both in the local pending buffer so that when a read happens, it will perform “read your own writes” directly from the local storage. That property makes offline read possible so that even when the network is broken, the application can work properly and the following updates will be stashed in the buffer and resent until the network is recovered again.





GSP also relies on RTOB to send back a totally ordered sequence of updates to the locally known buffer. The existence of known and pending buffer provides the possibility for updating these buffers fully asynchronously. In order to achieve high throughput, it also provides the batching option so that it does not require RTOB for every operation.

Compared with OACP, GSP not only has only totally-ordered operations but also performs locally read operation. When the relative order of operations of different clients is observable, it will provide some difficult condition. Given the example in paper <cit.>:


  wr(A, 2)     . 

  .     wr(B, 1) 

  .     wr(A, 1) 

  .     rd(A) → 2 

  rd(B) → 0     . 



Above is the key-value store shared data model which initially stores 0 for each address. In GSP protocol, such interleaving will be possible since rd(B) can get 0 before the local storage get the update from wr(B, 1).

Local read from GSP has a strange effect that it can not provide the strong consistency when processing locally read. While in OACP, this confusion will not happen since we always process read as a TOp and it will always get the updated state from the server. So only the following is possible, and the observable strong consistency is always preserved:


  wr(A, 2)     . 

  .     wr(B, 1) 

  .     wr(A, 1) 

  .     rd(A) → 1 

  rd(B) → 1     . 





§ PERFORMANCE EVALUATION


We evaluate the performance of OACP from the perspective of latency, throughput and coordination. We use a microbenchmark as well as a Twitter-like application inspired by Twissandra <cit.>.




  
Experimental set-up.
Our OACP implementation is based on the cluster extension of the widely-used Akka <cit.> actor-based middleware. The cluster environment is configured using three seed nodes; each seed node runs an actor that detects changes in cluster membership (i.e., nodes joining or leaving the cluster). This enables freely adding and removing cluster nodes.



The experiments on coordination are performed on a 1.6 GHz Intel Core i5 processor with 8 GB 1600 MHz DDR3 memory running macOS 10.13. The experiments on latency and throughput are performed on Amazon EC2 cluster which includes three T2 micro instances (1 vCPU, 1 Gb memory, and EBS-only storage. ) running in Ohio, London, and Syndey. Figure <ref> shows the average round-trip latency between each pair of sites. Our implementation is based on Scala 2.11.8, Akka 2.4.12, AspectJ 1.8.10, and JDK 1.8.0_162 (build 25.162-b01). The OACP implementation is available open-source on GitHub.[See <https://github.com/CynthiaZ92/OACP>]








  
Latency.
In OACP, any CvOp gets an immediate response once the request arrives to any of the servers in the cluster. In contrast, the TOp runs consensus protocol underneath to achieve consistency. In order to understand the effect of CvOps and TOps, we measure the latency for CvOps and TOps on Amazon EC2 in three different regions: Ohio, London and Sydney and we plot the CDFs of observed latencies in Figure <ref>. The cluster in our experiments is consists of three nodes which locate in different regions. A leader node needs to be elected to keep consensus. We put the client node in London and measure different conditions when the leader node locates in different regions. In general, CvOps get quick responses as we can see that the maximum latency is 60 ms, and 90% of the response latency are within 40 ms. The latency time of TOps depends on the location of the leader node.  When the leader node locates in Sydney, the maximum latency is 600 ms. And when the leader node locates in the other two regions, the maximum latency is around 350ms.



  
Throughput.
Now we focus on the throughput of OACP. We generate benchmarks with different proportion of CvOps and TOps. While we increase the number of concurrent requests to the same consistent log in the cluster, we measure the duration for processing all of the requests. The throughput is then the number of requests divided by the duration. The results in Figure <ref> show that increasing the ratio of TOps decreases the throughput. TOps require the leader node to force all the other replicas to reach consensus on the same log. Thus there will be a request queue on the server side. Any of the server nodes can process CvOps, and the commutative property of CvOps allows them to be processed in a random order. The throughput of the mix workloads is located between the pure workloads which give the programmer a range of choices.



  
Coordination.
We consider this aspect because previous work has shown that reducing the coordination within the protocol can improve the throughput of user operations dramatically <cit.>. In order to evaluate the performance independent of specific hardware and cluster configurations, our experiments count the number of exchanged messages. The message counting logic is added via automatic instrumentation of the executed JVM bytecode using AspectJ <cit.>.





  §.§.§ Microbenchmark.

We start the evaluation with a simple shopping cart benchmark to see the advantages and weakness of the OACP protocol. We define the “add” and the “remove” operation as CvOps in OACP. The “checkout” operation in OACP is defined as a TOp (to ensure consistency upon checkout). Then we generate sequences of n operations where n ∈ (0, 1000]; each operation can be either “add”, “remove”, or “checkout”.

We use Akka's messaging interface to define an AspectJ pointcut (Figure <ref>, lines 1–2). This allows us to detect every message sent between actors (Figure <ref>, lines 4–5). In order to count only messages generated by each protocol, we filter out internal Akka messages (Figure <ref>, line 9). Furthermore, we filter out heartbeat messages, since all protocols are assumed to exchange the same number of heartbeat messages, e.g., to implement RTOB.

