Sebastian Burckhardt

Alexey Gotsman

Hongseok Yang

Marek Zawirski

Replicated Data Types: Specification, Verification, Optimality

Keywords: D.2.4 [Software Engineering]: Software/Program Verification, F.3.1 [Logics and Meanings of Programs]: Specifying grams Keywords Replication, eventual consistency, weak memory 1. Introduction

Geographically distributed systems often rely on replicated eventually consistent data stores to achieve availability and performance. To resolve conflicting updates at different replicas, researchers and practitioners have proposed specialized consistency protocols, called replicated data types, that implement objects such as registers, counters, sets or lists. Reasoning about replicated data types has however not been on par with comparable work on abstract data types and concurrent data types, lacking specifications, correctness proofs, and optimality results.

To fill in this gap, we propose a framework for specifying replicated data types using relations over events and verifying their implementations using replication-aware simulations. We apply it to 7 existing implementations of 4 data types with nontrivial conflictresolution strategies and optimizations (last-writer-wins register, counter, multi-value register and observed-remove set). We also present a novel technique for obtaining lower bounds on the worstcase space overhead of data type implementations and use it to prove optimality of 4 implementations. Finally, we show how to specify consistency of replicated stores with multiple objects axiomatically, in analogy to prior work on weak memory models. Overall, our work provides foundational reasoning tools to support research on replicated eventually consistent stores.

if clients stop issuing update requests, then the replicas will eventually reach a consistent state.

(

) 1
Eventual consistency is a hot research area, and new replicated stores implementing it appear every year [START_REF]Riak key-value store[END_REF][START_REF] Burckhardt | Eventually consistent transactions[END_REF][START_REF] Conway | Logic and lattices for distributed programming[END_REF][START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF][START_REF] Li | Making geo-replicated systems fast if possible, consistent when necessary[END_REF][START_REF] Lloyd | Don't settle for eventual: scalable causal consistency for wide-area storage with COPS[END_REF][START_REF] Shapiro | Conflictfree replicated data types[END_REF][START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF][START_REF] Zawirski | SwiftCloud: Fault-tolerant georeplication integrated all the way to the client machine[END_REF]. Unfortunately, their semantics is poorly understood: the very term eventual consistency is a catch-all buzzword, and different stores claiming to be eventually consistent actually provide subtly different guarantees. The property [START_REF]Riak key-value store[END_REF], which is a form of quiescent consistency, is too weak to capture these. Although it requires the replicas to converge to the same state eventually, it doesn't say which one it will be. Furthermore, [START_REF]Riak key-value store[END_REF] does not provide any guarantees in realistic scenarios when updates never stop arriving. The difficulty of reasoning about the behavior of eventually consistent stores comes from a multitude of choices to be made in their design, some of which we now explain.

Allowing the replicas to be temporarily inconsistent enables eventually consistent stores to satisfy clients' requests from the local replica immediately, and broadcast the changes to the other replicas only after the fact, when the network connection permits this. However, this means that clients can concurrently issue conflicting operations on the same data item at different replicas; furthermore, if the replicas are out-of-sync, these operations will be applied to its copies in different states. For example, two users sharing an online store account can write two different zip codes into the delivery address; the same users connected to replicas with different views of the shopping cart can also add and concurrently remove the same product. In such situations the store needs to ensure that, after the replicas exchange updates, the changes by different clients will be merged and all conflicts will be resolved in a meaningful way. Furthermore, to ensure eventual consistency [START_REF]Riak key-value store[END_REF], the conflict resolution has to be uniform across replicas, so that, in the end, they converge to the same state.

The protocols achieving this are commonly encapsulated within replicated data types [START_REF]Riak key-value store[END_REF][START_REF] Burckhardt | Cloud types for eventual consistency[END_REF][START_REF] Conway | Logic and lattices for distributed programming[END_REF][START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF][START_REF] Roh | Replicated abstract data types: Building blocks for collaborative applications[END_REF][START_REF] Shapiro | Conflictfree replicated data types[END_REF][START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF] that implement objects, such as registers, counters, sets or lists, with various conflictresolution strategies. The strategies can be as simple as establishing a total order on all operations using timestamps and letting the last writer win, but can also be much more subtle. Thus, a data type can detect the presence of a conflict and let the client deal with it: e.g., the multi-value register used in Amazon's Dynamo key-value store [START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF] would return both conflicting zip codes in the above example. A data type can also resolve the conflict in an applicationspecific way. For example, the observed-remove set [START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF][START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF] processes concurrent operations trying to add and remove the same element so that an add always wins, an outcome that may be appropriate for a shopping cart.

Replicated data type implementations are often nontrivial, since they have to maintain not only client-observable object state, but also metadata needed to detect and resolve conflicts and to handle network failures. This makes reasoning about their behavior challenging. The situation gets only worse if we consider multi-ple replicated objects: in this case, asynchronous propagation of updates between replicas may lead to counterintuitive behaviorsanomalies, in database terminology. The following code illustrates an anomaly happening in real replicated stores [START_REF]Riak key-value store[END_REF][START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF]: Replica r 1 → x.wr(post) i = y.rd // comment ← Replica r 2 y.wr(comment) j = x.rd // empty [START_REF] Adve | Shared memory consistency models: A tutorial[END_REF] We have two clients reading from and writing to register objects x and y at two different replicas; i and j are client-local variables. The first client makes a post by writing to x at replica r1 and then comments on the post by writing to y. After every write, replica r1 might send a message with the update to replica r2. If the messages carrying the writes of post to x and comment to y arrive to replica r2 out of the order they were issued in, the second client can see the comment, but not the post. Different replicated stores may allow such an anomaly or not, and this has to be taken into account when reasoning about them.

In this paper, we propose techniques for reasoning about eventually consistent replicated stores in the following three areas.

1. Specification. We propose a comprehensive framework for specifying the semantics of replicated stores. Its key novel component is replicated data type specifications ( §3), which provide the first way of specifying the semantics of replicated objects with advanced conflict resolution declaratively, like abstract data types [START_REF] Liskov | Programming with abstract data types[END_REF]. We achieve this by defining the result of a data type operation not by a function of states, but of operation contextssets of events affecting the result of the operation, together with some relationships between them. We show that our specifications are sufficiently flexible to handle data types representing a variety of conflict-resolution strategies: last-write-wins register, counter, multi-value register and observed-remove set.

We then specify the semantics of a whole store with multiple objects, possibly of different types, by consistency axioms ( §7), which constrain the way the store processes incoming requests in the style of weak shared-memory models [START_REF] Adve | Shared memory consistency models: A tutorial[END_REF] and thus define the anomalies allowed. As an illustration, we define consistency models used in existing replicated stores, including a weak form of eventual consistency [START_REF]Riak key-value store[END_REF][START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF] and different kinds of causal consistency [START_REF] Li | Making geo-replicated systems fast if possible, consistent when necessary[END_REF][START_REF] Lloyd | Don't settle for eventual: scalable causal consistency for wide-area storage with COPS[END_REF][START_REF] Shapiro | Conflictfree replicated data types[END_REF][START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF]. We find that, when specialized to last-writerwins registers, these specifications are very close to fragments of the C/C++ memory model [START_REF] Batty | Mathematizing C++ concurrency[END_REF]. Thus, our specification framework generalizes axiomatic shared-memory models to replicated stores with nontrivial conflict resolution.

Verification.

We propose a method for proving the correctness of replicated data type implementations with respect to our specifications and apply it to seven existing implementations of the four data types mentioned above, including those with nontrivial optimizations. Reasoning about the implementations is difficult due to the highly concurrent nature of a replicated store, with multiple replicas simultaneously updating their object copies and exchanging messages. We address this challenge by proposing replicationaware simulations ( §5). Like classical simulations from data refinement [START_REF] Hoare | Proof of correctness of data representations[END_REF], these associate a concrete state of an implementation with its abstract description-structures on events, in our case. To combat the complexity of replication, they consider the state of an object at a single replica or a message in transit separately and associate it with abstract descriptions of only those events that led to it. Verifying an implementation then requires only reasoning about an instance of its code running at a single replica.

Here, however, we have to deal with another challenge: code at a single replica can access both the state of an object and a message at the same time, e.g., when updating the former upon receiving the latter. To reason about such code, we often need to rely on certain agreement properties correlating the abstract descriptions of the message and the object state. Establishing these properties re-quires global reasoning. Fortunately, we find that agreement properties needed to prove realistic implementations depend only on basic facts about their messaging behavior and can thus be established once for broad classes of data types. Then a particular implementation within such a class can be verified by reasoning purely locally.

By carefully structuring reasoning in this way, we achieve easy and intuitive proofs of single data type implementations. We then lift these results to stores with multiple objects of different types by showing how consistency axioms can be proved given properties of the transport layer and data type implementations ( §7).

3. Optimality. Replicated data type designers strive to optimize their implementations; knowing that one is optimal can help guide such efforts in the most promising direction. However, proving optimality is challengingly broad as it requires quantifying over all possible implementations satisfying the same specification.

For most data types we studied, the primary optimization target is the size of the metadata needed to resolve conflicts or handle network failures. To establish optimality of metadata size, we present a novel method for proving lower bounds on the worst-case metadata overhead of replicated data types-the proportion of metadata relative to the client-observable content. The main idea is to find a large family of executions of an arbitrary correct implementation such that, given the results of data type operations from a certain fixed point in any of the executions, we can recover the previous execution history. This implies that, across executions, the states at this point are distinct and thus must have some minimal size.

Using our method, we prove that four of the implementations we verified have an optimal worst-case metadata overhead among all implementations satisfying the same specification. Two of these (counter, last-writer-wins register) are well-known; one (optimized observed-remove set [START_REF] Bieniusa | An optimized conflict-free replicated set[END_REF]) is a recently proposed nontrivial optimization; and one (optimized multi-value register) is a small improvement of a known implementation [START_REF] Shapiro | Conflictfree replicated data types[END_REF] that we discovered during a failed attempt to prove optimality of the latter. We summarize all the bounds we proved in Fig. 10.

We hope that the theoretical foundations we develop will help in exploring the design space of replicated data types and replicated eventually consistent stores in a systematic way.

Replicated Data Types

We now describe our formal model for replicated stores and introduce replicated data type implementations, which implement operations on a single object at a replica and the protocol used by replicas to exchange updates to this object. Our formalism follows closely the models used by replicated data type designers [START_REF] Shapiro | Conflictfree replicated data types[END_REF].

A replicated store is organized as a collection of named objects Obj = {x, y, z, . . .}. Each object is hosted at all replicas r, s ∈ ReplicaID. The sets of objects and replicas may be infinite, to model their dynamic creation. Clients interact with the store by performing operations on objects at a specified replica. Each object x ∈ Obj has a type τ = type(x) ∈ Type, whose type signature (Op τ , Valτ ) determines the set of supported operations Op τ (ranged over by o) and the set of their return values Valτ (ranged over by a, b, c, d). We assume that a special value ⊥ ∈ Valτ belongs to all sets Valτ and is used for operations that return no value. For example, we can define a counter data type ctr and an integer register type intreg with operations for reading, incrementing or writing an integer a:

Valctr = Valintreg = Z ∪ {⊥}, Op ctr = {rd, inc} and Op intreg = {rd} ∪ {wr(a) | a ∈ Z}.
We also assume sets Message of messages (ranged over by m) and timestamps Timestamp (ranged over by t). For simplicity, we let timestamps be positive integers: Timestamp = N1. DEFINITION 1. A replicated data type implementation for a data type τ is a tuple Dτ = (Σ, σ0, M, do, send, receive), where σ0 :

do : Op τ × Σ × Timestamp → Σ × Valτ ; send : Σ → Σ × M ; receive : Σ × M → Σ.
We denote a component of Dτ , such as do, by Dτ .do. A tuple Dτ defines the class of implementations of objects with type τ , meant to be instantiated for every such object in the store. Σ is the set of states (ranged over by σ) used to represent the current state of the object, including metadata, at a single replica. The initial state at every replica is given by σ0.

Dτ provides three methods that the rest of the store implementation can call at a given replica; we assume that these methods execute atomically. We visualize store executions resulting from repeated calls to the methods as in Fig. 1(a), by arranging the calls on several vertical timelines corresponding to replicas at which they occur and denoting the delivery of messages by diagonal arrows. In §4, we formalize them as sequences of transitions called concrete executions and define the store semantics by their sets; the intuition given by Fig. 1(a) should suffice for the following discussion.

A client request to perform an operation o ∈ Op τ triggers the call do(o, σ, t) (e.g., event 1 in Fig. 1(a)). This takes the current state σ ∈ Σ of the object at the replica where the request is issued and a timestamp t ∈ Timestamp provided by the rest of the store implementation and produces the updated object state and the return value of the operation. The data type implementation can use the timestamp provided, e.g., to implement the last-writer-wins conflict-resolution strategy mentioned in §1, but is free to ignore it.

Nondeterministically, in moments when the network is able to accept messages, a replica calls send. Given the current state of the object at the replica, send produces a message in M to broadcast to all other replicas (event 2 in Fig. 1(a)); sometimes send also alters the state of the object. Using broadcast rather than point-to-point communication does not limit generality, since we can always tag messages with the intended receiver. Another replica that receives the message generated by send calls receive to merge the enclosed update into its copy of the object state (event 3 in Fig. 1(a)).

We now reproduce three replicated data type implementations due to Shapiro et al. [START_REF] Shapiro | Conflictfree replicated data types[END_REF].They fall into two categories: in opbased implementations, each message carries a description of the latest operations that the sender has performed, and in state-based implementations, a description of all operations it knows about.

Op-based counter (ctr). Fig. 2(a) shows an implementation of the ctr data type. A replica stores a pair a, d , where a is the current value of the counter, and d is the number of increments performed since the last broadcast (we use angle brackets for tuples representing states and messages). The send method returns d and resets it; the receive method adds the content of the message to a. This implementation is correct, as long as each message is delivered exactly once (we show how to prove this in §5). Since inc operations commute, they never conflict: applying them in different orders at different replicas yields the same final state.

State-based counter (ctr).

The implementation in Fig. 2(b) summarizes the currently known history by recording the contri-

Σ = N 0 × N 0 M = N 0 σ 0 = λr. 0, 0 do(rd, a, d , t) = ( a, d , a) do(inc, a, d , t) = ( a + 1, d + 1 , ⊥) send( a, d ) = ( a, 0 , d) receive( a, d , d ′ ) = a + d ′ , d (b) State-based counter (ctr) Σ = ReplicaID × (ReplicaID → N 0 ) σ 0 = λr. r, λs. 0 M = ReplicaID → N 0 do(rd, r, v , t) = ( r, v , {v(s) | s ∈ ReplicaID}) do(inc, r, v , t) = ( r, v[r → v(r) + 1] , ⊥) send( r, v ) = ( r, v , v) receive( r, v , v ′ ) = r, (λs. max{v(s), v ′ (s)}) (c) State-based last-writer-wins register (intreg) Σ = Z × (Timestamp ⊎ {0}) σ 0 = λr. 0, 0 M = Σ do(rd, a, t , t ′ ) = ( a, t , a) do(wr(a ′ ), a, t , t ′ ) = if t < t ′ then ( a ′ , t ′ , ⊥) else ( a, t , ⊥) send( a, t ) = ( a, t , a, t ) receive( a, t , a ′ , t ′ ) = if t < t ′ then a ′ , t ′ else a, t
bution of every replica to the counter value separately (reminiscent of vector clocks [START_REF] Mattern | Virtual time and global states of distributed systems[END_REF]). A replica stores its identifier r and a vector v such that for each replica s the entry v(s) gives the number of increments made by clients at s that have been received by r. A rd operation returns the sum of all entries in the vector. An inc operation increments the entry for the current replica. We denote by v[i → j] the function that has the same value as v everywhere, except for i, where it has the value j. The send method returns the vector, and the receive method takes the maximum of each entry in the vectors v and v ′ given to it. This is correct because an entry for s in either vector reflects a prefix of the sequence of increments done at replica s. Hence, we know that min{v(s), v ′ (s)} increments by s are taken into account both in v(s) and in v ′ (s).

State-based last-writer-wins (LWW) register (intreg). Unlike counters, registers have update operations that are not commutative. To resolve conflicts, the implementation in Fig. 2 uses the last-writer-wins strategy, creating a total order on writes by associating a unique timestamp with each of them. A state contains the current value, returned by rd, and the timestamp at which it was written (initially, we have 0 instead of a timestamp). A wr(a ′ ) compares its timestamp t ′ with the timestamp t of the current value a and sets the value to the one with the highest timestamp. Note that here we have to allow for t ′ < t, since we do not make any assumptions about timestamps apart from uniqueness: e.g., the rest of the store implementation can compute them using physical or Lamport clocks [START_REF] Lamport | Time, clocks, and the ordering of events in a distributed system[END_REF]. We show how to state assumptions about timestamps in §4. The send method just returns the state, and the receive method chooses the winning value by comparing the timestamps in the current state and the message, like wr.

State-based vs. op-based. State-based implementations converge to a consistent state faster than op-based implementations because they are transitively delivering, meaning that they can propagate updates indirectly. For example, when using the counter in Fig. 2(b), in the execution in Fig. 1(a) the read at r3 (event 7) returns 2, even though the message from r1 has not arrived yet, because r3 learns about r1's update via r2. State-based implementations are also resilient against transport failures like message loss, reordering, or duplication. Op-based implementations require the replicated store using them to mask such failures (e.g., using message sequence numbers, retransmission buffers, or reorder buffers).

The potential weakness of state-based implementations is the size of states and messages, which motivates our examination of space optimality in §6. For example, we show that the counter in Fig. 2(b) is optimal, meaning that no counter implementation satisfying the same requirements (transitive delivery and resilience against message loss, reordering, and duplication) can do better.

Specifying Replicated Data Types and Stores

Consider the concrete execution in Fig. 1(a). What are valid return values for the read in event 7? Intuitively, 1 or 2 can be justifiable, but not 100. We now present a framework for specifying the expected outcome declaratively, without referring to implementation details. For example, we give a specification of a replicated counter that is satisfied by both implementations in Fig. 2(a,b).

In presenting the framework, we rely on the intuitive understanding of the way a replicated store executes given in §2. Later we define the store semantics formally ( §4), which lets us state what it means for a store to satisfy our specifications ( §4 and §7).

Abstract Executions and Specification Structure

We define our specifications on abstract executions, which include only user-visible events (corresponding to do calls) and describe the other information about the store processing in an implementation-independent form. Informally, we consider a concrete execution correct if it can be justified by an abstract execution satisfying the specifications that is "similar" to it and, in particular, has the same operations and return values.

Abstract executions are inspired by axiomatic definitions of weak shared-memory models [START_REF] Adve | Shared memory consistency models: A tutorial[END_REF]. In particular, we use their previously proposed reformulation with visibility and arbitration relations [START_REF] Burckhardt | Eventually consistent transactions[END_REF], which are similar to the reads-from and coherence relations from weak shared-memory models. We provide a comparison with shared-memory models in §7 and with [START_REF] Burckhardt | Eventually consistent transactions[END_REF] in §8. DEFINITION 2. An abstract execution is a tuple A = (E, repl, obj, oper, rval, ro, vis, ar), where • E ⊆ Event is a set of events from a countable universe Event;

• each event e ∈ E describes a replica repl(e) ∈ ReplicaID performing an operation oper(e) ∈ Op type(obj(e)) on an object obj(e) ∈ Obj, which returns the value rval(e) ∈ Val type(obj(e)) ; • ro ⊆ E × E is a replica order, which is a union of transitive, irreflexive and total orders on events at each replica;

• vis ⊆ E × E is an acyclic visibility relation such that ∀e, f ∈ E. e vis -→ f =⇒ obj(e) = obj(f ); • ar ⊆ E × E is an arbitration relation, which is a union of
transitive, irreflexive and total orders on events on each object. We also require that ro, vis and ar be well-founded. In the following, we denote components of A and similar structures as in A.repl. We also use (e, f ) ∈ r and e r -→ f interchangeably.

Informally, e vis -→ f means that f is aware of e and thus e's effect can influence f 's return value. In implementation terms, this may be the case if the update performed by e has been delivered to the replica performing f before f is issued. The exact meaning of "delivered", however, depends on how much information messages carry in the implementation. For example, as we explain in §3.2, the return value of a read from a counter is equal to the number of inc operations visible to it. Then, as we formalize in §4, the abstract execution illustrated in Fig. 1 The ar relation represents the ordering information provided by the store, e.g., via timestamps. On the right we show an abstract execution corresponding to a variant of the anomaly (2). The ar edge means that any replica that sees both writes to x should assume that post overwrites empty.

We give a store specification by two components, constraining abstract executions: 1. Replicated data type specifications determine return values of operations in an abstract execution in terms of its vis and ar relations, and thus define conflict-resolution policies for individual objects in the store. The specifications are the key novel component of our framework, and we discuss them next. 2. Consistency axioms constrain vis and ar and thereby disallow anomalies and extend the semantics of individual objects to that of the entire store. We defer their discussion to §7. See Fig. 13 for their flavor; in particular, COCV prohibits the anomaly above. Each of these components can be varied separately, and our specifications will define the semantics of any possible combination. Given a specification of a store, we can determine whether a set of events can be observed by its users by checking if there is an abstract execution with this set of events satisfying the data type specifications and consistency axioms.

Replicated Data Type Specifications

In a sequential setting, the semantics of a data type τ can be specified by a function Sτ : Op + τ → Valτ , which, given a nonempty sequence of operations performed on an object, specifies the return value of the last operation. For a register, read operations return the value of the last preceding write, or zero if there is no prior write. For a counter, read operations return the number of preceding increments. Thus, for any sequence of operations ξ: Sintreg(ξ rd) = a, if wr(0) ξ = ξ1 wr(a) ξ2 and ξ2 does not contain wr operations; Sctr(ξ rd) = (the number of inc operations in ξ); Sintreg(ξ inc) = Sctr(ξ wr(a)) = ⊥.

In a replicated store, the story is more interesting. We specify a data type τ by a function Fτ , generalizing Sτ . Just like Sτ , this determines the return value of an operation based on prior operations performed on the object. However, Fτ takes as a parameter not a sequence, but an operation context, which includes all we need to know about a store execution to determine the return value of a given operation o-the set E of all events that are visible to o, together with the operations performed by the events and visibility and arbitration relations on them. DEFINITION 3. An operation context for a data type τ is a tuple

L = (o, E, oper, vis, ar), where o ∈ Op τ , E is a finite subset of Event, oper : E → Op τ , vis ⊆ E × E is acyclic and ar ⊆ E × E is transitive, irreflexive and total.
We can extract the context of an event e ∈ A.E in an abstract execution A by selecting all events visible to it according to A.vis: Note that Fτ (o, ∅, . . .) returns the value resulting from performing o on the initial state for the data type (e.g., 0 for the LWW-register).

We specify multiple data types used in a replicated store by a partial function F mapping them to data type specifications. DEFINITION 5. An abstract execution A satisfies F, written A |= F, if the return value of every event in A is computed on its context by the specification for the type of the object the event accesses:

∀e ∈ A.E. (A.rval(e) = F(type(A.obj(e)))(ctxt(A, e))).
We specify a whole store by F and a set of consistency axioms ( §7). This lets us determine if its users can observe a given set of events by checking if there is an abstract execution with these events that satisfies F according to the above definition, as well as the axioms.

Note that Fτ is deterministic. This does not mean that so is an outcome of an operation on a store; rather, that all the nondeterminism arising due to its distributed nature is resolved by vis and ar in the context passed to Fτ . These relations are chosen arbitrarily subject to consistency axioms. Due to the determinacy property, two events that perform the same operation and see the same set of events produce the same return values. As we show in §7, this property ensures that our specifications can formalize eventual consistency in the sense of [START_REF]Riak key-value store[END_REF].

We now give four examples of data type specifications, corresponding to the four conflict-resolution strategies mentioned in §1 and §2: (1) operations commute, so no conflicts arise; (2) last writer wins; (3) all conflicting values are returned; and (4) conflicts are resolved in an application-specific way. We start by specifying the data types whose implementations we presented in §2.

Counter (ctr) is defined by

Fctr(inc, E, oper, vis, ar) = ⊥; Fctr(rd, E, oper, vis, ar) = {e ∈ E | oper(e) = inc} . (3)
Thus, according to Def. 5 the executions in Fig. 1(b) and 1(c) satisfy the counter specification: both 1 and 2 are valid return values for the read from x when there are two concurrent increments.

LWW-register (intreg) is defined by

Fintreg(o, E, oper, vis, ar) = Sintreg(E ar o), (4)
where E ar denotes the sequence obtained by ordering the operations performed by the events in E according to ar. Thus, the return value is determined by establishing a total order of the visible operations and applying the regular sequential semantics. For example, by Def. 5 in the example execution from §3.1 the read from x has to return empty; if we had a vis edge from the write of post to the read from x, then the read would have to return post. As we show in §7, weak shared-memory models are obtained by specializing our framework to stores with only LWW-registers.

We can obtain a concurrent semantics Fτ of any data type τ based on its sequential semantics Sτ similarly to (4). For example, Fctr defined above is equivalent to what we obtain using this generic construction. The next two examples go beyond this.

Multi-value register (mvr).

This register [START_REF]Riak key-value store[END_REF][START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF] has the same operations as the LWW-register, but its reads return a set of values: (We write for an expression whose value is irrelevant.) A read returns the values written by currently conflicting writes, defined as those that are not superseded in vis by later writes; ar is not used. For example, a rd would return {2, 3} in the context on the right. We use e : {h 1 = u 1 , h 2 = u 2 } to abbreviate h 1 (e) = u 1 and h 2 (e) = u 2 . We uncurry R ∈ RState where convenient.

Obj τ = {x ∈ Obj | type(x) = τ } RState = X⊆Obj x∈X (ReplicaID → D(type(x)).Σ) TState = MessageID ⇀ τ ∈Type (ReplicaID × Obj τ × D(τ ).M ) Config = RState × TState D(type(x)).do(o, σ, t) = (σ ′ , a) e : {act = do, obj = x, repl = r, oper = o, time = t, rval = a} (R[(x, r) → σ], T ) e -→ D (R[(x, r) → σ ′ ], T ) D(type(x)).send(σ) = (σ ′ , m) mid / ∈ dom(T ) e : {act = send, obj = x, repl = r, msg = mid} (R[(x, r) → σ], T ) e -→ D (R[(x, r) → σ ′ ], T [mid → (r, x, m)]) D(type(x)).receive(σ, m) = σ ′ r = r ′ e : {act = receive, obj = x, repl = r, srepl = r ′ , msg = mid} (R[(x, r) → σ], T [mid → (r ′ , x, m)]) e -→ D (R[(x, r) → σ ′ ], T [mid → (r ′ , x, m)])
4. Observed-remove set (orset). How do we specify a replicated set of integers? The operations of adding and removing different elements commute and thus do not conflict. Conflicts arise from concurrently adding and removing the same element. For example, we need to decide what rd will return as the contents of the set in the context (rd, {e, f }, oper, vis, ar), where oper(e) = add(42) and oper(f ) = remove(42). If we use the generic construction from the LWW-register, the result will depend on the arbitration relation: ∅ if e ar -→ f , and {42} otherwise. An application may require a more consistent behavior, e.g., that an add operation always win against concurrent remove operations. Observed-remove (OR) set [START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF][START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF] achieves this by mandating that remove operations cancel only the add operations that are visible to them:

Forset(rd, E, oper, vis, ar) = {a | ∃e ∈ E. oper(e) = add(a) ∧ ¬∃f ∈ E. oper(f ) = remove(a) ∧ e vis -→ f }, (5)
In the above operation context rd will return ∅ if e vis -→ f , and {42} otherwise. The rationale is that, in the former case, add(42) and remove(42) are not concurrent: the user who issued the remove knew that 42 was in the set and thus meant to remove it. In the latter case, the two operations are concurrent and thus add wins.

As the above examples illustrate, our specifications can describe the semantics of data types and their conflict-resolution policies declaratively, without referring to the internals of their implementations. In this sense the specifications generalize the concept of an abstract data type [START_REF] Liskov | Programming with abstract data types[END_REF] to the replicated setting.

Store Semantics and Data Type Correctness

A data type library D is a partial mapping from types τ to data type implementations D(τ ) from Def. 1. We now define the semantics of a replicated store with a data type library D as a set of its concrete executions, previously introduced informally by Fig. 1(a). We then state what it means for data type implementations of §2 to satisfy their specifications of §3.2 by requiring their concrete executions to be justified by abstract ones. In §7 we generalize this to the correctness of the whole store with multiple object with respect to both data type specifications and consistency axioms.

Semantics. We define the semantics using the relation -→ D : Config × Event × Config in Fig. 3, which describes a single step of the store execution. The relation transforms configurations (R, T ) ∈ Config describing the store state: R gives the object state at each replica, and T the set of messages in transit between them, each identified by a message identifier mid ∈ MessageID. A message is annotated by the origin replica and the object to which it pertains. We allow the store to contain only some objects from Obj and thus allow R to be partial on them. We use a number of functions on events, such as act, obj, etc., to record the information about the corresponding transitions, so that -→ D is implicitly parameterized by them; we give their full list in Def. 6 below.

The first rule in Fig. 3 describes a replica r performing an operation o on an object x using the do method of the corresponding data type implementation. We record the return value using the function rval. To communicate the change to other replicas, we can at any time perform a transition defined by the second rule, which puts a new message m created by a call to send into the set of messages in transit. The third rule describes the delivery of such a message to a replica r other than the origin replica r ′ , which triggers a call to receive. Note that the relation -→ D does not make any assumptions about message delivery: messages can be delivered in any order, multiple times, or not at all. These assumptions can be introduced separately, as we show later in this section. A concrete execution can be thought of as a finite or infinite sequence of transitions:

(R0, T0) e 1 -→ (R1, T1) e 2 -→ . . . en -→ (Rn, Tn) . . . ,
where all events ei are distinct. To ease mapping between concrete and abstract executions in the future, we formalize it as a structure on events, similarly to Def. 2. DEFINITION 6. A concrete execution of a store with a data type library D is a tuple C = (E, eo, pre, post, act, obj, repl, oper, time, rval, msg, srepl).

Here E ⊆ Event, the execution order eo is a well-founded, transitive, irreflexive and total order on E, relating the events according to the order of the transitions they describe, time is injective and pre, post : E → Config form a valid sequence of transitions:

(∀e ∈ E. pre(e) e -→ D post(e)) ∧ (∀e, f ∈ E. e eo -→ f ∧ ¬∃g. e eo -→ g eo -→ f =⇒ post(e) = pre(f )).
We have omitted the types of functions on events, which are easily inferred from Fig. 3: e.g., act : E → {do, send, receive} and time : E ⇀ Timestamp, defined only on e with act(e) = do.

We denote the initial configuration of C by init(C) = C.pre(e0), where e0 is the minimal event in C.eo. If C.E is finite, we denote the final configuration of C by final(C) = C.post(e f ), where e f is the maximal event in C.eo. The semantics D of D is the set of all its concrete executions C that start in a configuration with an empty set of messages and all objects in initial states, i.e.,

∃X ⊆ Obj. init(C) = ((λx ∈ X. D(type(x)). σ0), [ ]),
where [ ] is the everywhere-undefined function.

Transport layer specifications. Data type implementations such as the op-based counter in Fig. 2(a) can rely on some guarantees concerning the delivery of messages ensured by the rest of the store implementation. They may similarly assume certain properties of timestamps other than uniqueness (guaranteed by the injectivity of time in Def. 6). We take such assumptions into account by admitting only a subset of executions from D that satisfy a transport layer specification T , which is a predicate on concrete executions. Thus, we consider a replicated store to be defined by a pair (D, T ) and the set of its executions be D ∩ T .

Even though our definition of T lets it potentially restrict data type implementation internals, the particular instantiations we use only restrict message delivery and timestamps. For technical reasons, we assume that T always satisfies certain closure properties: for every C ∈ T , the projection of C onto events on a given object or a subset of events forming a prefix in the eo order is also in T .

As an example, we define a transport layer specification ensuring that a message is delivered to any single replica at most once, as required by the implementation in Fig. 2(a). Let the delivery relation del(C) ∈ C.E × C.E pair events sending and receiving the same message:

e del(C) ----→ f ⇐⇒ e C.eo ---→ f ∧ C.act(e) = send ∧ C.act(f ) = receive ∧ C.msg(e) = C.msg(f ).
Then the desired condition on concrete executions C is

∀e, f, g ∈ C.E. e del(C) ----→ f ∧ e del(C) ----→ g ∧ C.repl(f ) = C.repl(g) =⇒ f = g. (T-Unique)
Data type implementation correctness. We now state what it means for an implementation Dτ of type τ from Def. 1 to satisfy a specification Fτ from Def. 4. To this end, we consider the behavior of Dτ under the "most general" client and transport layer, performing all possible operations and message deliveries. Formally, let Dτ be the set of executions C ∈ [τ → Dτ ] of a store containing a single object x of a type τ with the implementation Dτ , i.e., init(C) = (R, [ ]) for some R such that dom(R) = {x}.

Then Dτ should satisfy Fτ under a transport specification T if for every concrete execution C ∈ Dτ ∩ T we can find a "similar" abstract execution satisfying Fτ and, in particular, having the same operations and return values. As it happens, all components of the abstract execution except visibility are straightforwardly determined by C; as explained in §3.1, we have some freedom in choosing visibility. We define the choice using a visibility witness As we explained informally in §3.1, the visibility witness depends on how much information the implementation puts into messages. Since state-based implementations, such as the ones in Fig. 2(b,c), are transitively delivering ( §2), for them we use the witness V state (C) = (ro(C) ∪ del(C)) + | do . By the definition of ro(C) and del(C), (ro(C) ∪ del(C)) is acyclic, so V state is welldefined. State-based implementations do not make any assumptions about the transport layer: in this case we write T = T-Any. In contrast, op-based implementations, such as the one in Fig. 2(a), require T = T-Unique. Since such implementations are not transitively delivering, the witness V state is not appropriate for them. We could attempt to define a witness for them by straightforwardly lifting the delivery relation:

V op (C) = ro(C)| do ∪ {(e, f ) | e, f ∈ (C.E)| do ∧ ∃e ′ , f ′ . e ro(C) ---→ e ′ del(C) ----→ f ′ ro(C) ---→ f }.
However, we need to be more careful, since for op-based imple-

mentations e ro(C) ---→ e ′ del(C) ----→ f ′ ro(C)
---→ f does not ensure that the update of e is taken into account by f : if there is another send event e ′′ in between e and e ′ , then e ′′ will capture the update of e and e ′ will not. Hence, we define the witness as:

V op (C) = ro(C)| do ∪ {(e, f ) | e, f ∈ (C.E)| do ∧ ∃e ′ , f ′ . e ro(C) ---→ e ′ del(C) ----→ f ′ ro(C) ---→ f ∧ ¬∃e ′′ . e ro(C) ---→ e ′′ ro(C) ---→ e ′ ∧ C.act(e ′′ ) = send}.
We next present a method for proving data type implementation correctness in the sense of Def. 7. In §7 we lift this to stores with multiple objects and take into account consistency axioms.

Proving Data Type Implementations Correct

The straightforward approach to proving correctness in the sense of Def. 7 would require us to consider global store configurations in executions C, including object states at all replicas and all messages in transit, making the reasoning non-modular and unintuitive. To deal with this challenge, we focus on a single component of a store configuration using replication-aware simulation relations Rr and M, analogous to simulation (aka coupling) relations used in data refinement [START_REF] Hoare | Proof of correctness of data representations[END_REF]. The Rr relation associates the object state at a replica r with an abstract execution that describes only those events that led to this state; M does the same for a message. For example, when proving Dctr in Fig. 2(b) with respect to Fctr in (3), M associates a message carrying a vector v with executions in which each replica s makes v(s) increments. As part of a proof of Dτ , we require checking that the effect of its methods, such as Dτ .do, can be simulated by appropriately transforming related abstract executions while preserving the relations. We define these transformations using abstract methods do ♯ , send ♯ and receive ♯ as illustrated in Fig. 4(a,b). For example, if a replica r executes Dτ .do from a state σ related by Rr to an abstract execution I (we explain the use of I instead of A later), we need to find an I ′ related by Rr to the resulting state σ ′ . We also need to check that the value returned by Dτ .do on σ is equal to that returned by Fτ on I.

These conditions consider the behavior of an implementation method on a single state and/or message and its effect on only the relevant part of the abstract execution. However, by localizing the reasoning in this way, we lose some global information that is actually required to verify realistic implementations. In particular, this occurs when discharging the obligation for receive in Fig. 4(b). Taking a global view, σ and m there are meant to come from the same configuration in a concrete execution C; correspondingly, I and J are meant to be fragments of the same abstract execution abs(C, V). In this context we may know certain agreement properties correlating I and J, e.g., that the union of their visibility relations is itself a well-formed visibility relation and is thus acyclic. Establishing them requires global reasoning about whole executions C and abs(C, V). Fortunately, we find that this can be done knowing only the abstract methods, not the implementation Dτ . Furthermore, these methods state basic facts about the messaging behavior of implementations and are thus common to broad classes of them, such as state-based or op-based. This allows us to establish agreement properties using global reasoning once for a given class of implementations; at this stage we can also benefit from the transport layer specification T and check that the abstract methods construct visibility according to the given witness V. Then a particular implementation within the class can be verified by discharging local obligations, such as those in Fig. 4(a,b), while assuming agreement properties. This yields easy and intuitive proofs.

To summarize, we deal with the challenge posed by a distributed data type implementation by decomposing reasoning about it into

I do ♯ / / I ′ O O Rr σ Dτ .do / / Rr O O σ ′ (I, J) receive ♯ / / I ′ O O Rr (σ, m) Dτ .receive / / Rr ×M O O σ ′ D step(C ′ ,e,D) / / D ′ O O G C e / / G O O C ′
global reasoning done once for a broad class of implementations and local implementation-specific reasoning. We start by presenting the general form of obligations to be discharged for a single implementation within a certain class ( §5.1) and the particular form they take for the class of state-based implementations ( §5.2), together with some examples ( §5.3). We then formulate the obligations to be discharged for a class of implementations ( §5.4), which in particular, establish the agreement properties assumed in the perimplementation obligations. In [12, §B], we give the obligations for op-based implementations, together with a proof of the counter in Fig. 2(a). An impatient reader can move on to §6 after finishing §5.3, and come back to §5.4 later. Since Def. 7 considers only single-object executions, we fix an object x of type τ and consider only concrete and abstract executions over x, whose sets we denote by CEx[x] and AEx[x].

Replication-Aware Simulations

As is typical for simulation-based proofs, we need to use auxiliary state to record information about the computation history. For this reason, actually our simulation relations associate a state or a message with an instrumented execution-a pair (A, info) ∈ IEx of an abstract execution A ∈ AEx[x] and a function info : A.E → AInfo, tagging events with auxiliary information from a set AInfo. As we show below, AInfo can be chosen once for a class of data type implementations: e.g., AInfo = Timestamp for state-based ones ( §5.2). We use I and J to range over instrumented executions and shorten, e.g., I.A.E to I.E. For a partial function h we write h(x)↓ for x ∈ dom(h), and adopt the convention that h(x) = y implies h(x)↓. DEFINITION 8. A replication-aware simulation between Dτ and Fτ with respect to info and abstract methods do ♯ , send ♯ and receive ♯ is a collection of relations {Rr, M | r ∈ ReplicaID} satisfying the conditions in Fig. 5.

Here info and abstract methods are meant to be fixed for a given class of implementations, such as state or op-based. To prove a particular implementation within this class, one needs to find simulation relations satisfying the conditions in Fig. 5. For example, as we show in §5.3, the following relation lets us prove the correctness of the counter in Fig. 2(b) with respect to info and abstract methods appropriate for state-based implementations:

s, v [Rr] I ⇐⇒ (r = s) ∧ (v [M] I); v [M] ((E, repl, obj, oper, rval, ro, vis, ar), info) ⇐⇒ ∀s. v(s) = {e ∈ E | oper(e) = inc ∧ repl(e) = s} . (6)
INIT in Fig. 5 associates the initial state at a replica r with the execution having an empty set of events. DO, SEND and RECEIVE formalize the obligations illustrated in Fig. 4(a,b). Note that do ♯ is parameterized by an event e (required to be fresh in instantiations) and the information about the operation performed.

The abstract methods are partial and the obligations in Fig. 5 assume that their applications are defined. When instantiating receive ♯ for a given class of implementations, we let it be defined only when its arguments satisfy the agreement property for this class, which we establish separately ( §5.4). While doing this, we can also establish some execution invariants, holding of single ex- SEND:

, σ ′ ∈ Σ, m ∈ M , I, I ′ , J ∈ IEx, e ∈ Event, r ∈ ReplicaID, o ∈ Op τ , a ∈ Valτ , t ∈ Timestamp. Rr ⊆ Σ × IEx, r ∈ ReplicaID; M ⊆ M × IEx do ♯ : IEx × Event × ReplicaID × Op τ × Valτ × Timestamp ⇀ IEx send ♯ : IEx ⇀ IEx × IEx; receive ♯ : IEx × IEx ⇀ IEx
(send ♯ (I) = (I ′ , J) ∧ Dτ .send(σ) = (σ ′ , m) ∧ σ [Rr] I) =⇒ (σ ′ [Rr] I ′ ∧ m [M] J) RECEIVE: (receive ♯ (I, J)↓ ∧ σ [Rr] I ∧ m [M] J) =⇒ (Dτ .receive(σ, m) [Rr] receive ♯ (I, J))
ecutions supplied as parameters to do ♯ and send ♯ . We similarly assume them in Fig. 5 via the definedness of these abstract methods.

Instantiation for State-Based Implementations

Fig. 6 defines the domain AInfo and abstract methods appropriate for state-based implementations. In §5.4 we show that the existence of a simulation of Def. 8 with respect to these parameters implies Dτ sat[V state , T-Any] Fτ (Theorems 9 and 10). The do ♯ method adds a fresh event e with the given attributes to I; its timestamp t is recorded in info. In the resulting execution I ′ , the event e is the last one by its replica, observes all events in I and occupies the place in arbitration consistent with t. The send ♯ method just returns I, which formalizes the intuition that, in state-based implementations, send returns a message capturing all the information about the object available at the replica. The receive ♯ method takes the component-wise union I ⊔ J of executions I related to the current state and J related to the message, applied recursively to the components of I.A and J.A. We also assume that I ⊔ J recomputes the arbitration relation in the resulting execution from the timestamps. This is the reason for recording them in info: we would not be able to construct receive ♯ (I, J).ar solely from I.ar and J.ar.

The agreement property agree(I, J) guarantees that I ⊔ J is well-formed (e.g., its visibility relation is acyclic) and that, for each replica, I describes a computation extending J or vice versa. The latter follows from the observation we made when explaining the state-based counter in §2: a message or a state in a state-based implementation reflects a prefix of the sequence of events performed at a given replica. The first conjunct of the execution invariant inv requires arbitration to be consistent with event timestamps; the second conjunct follows from the definition of V state ( §4). When discharging the obligations in Fig. 5 with respect to the parameters in Fig. 6 for a particular implementation, we can rely on the agreement property and the execution invariant.

Examples

We illustrate the use of the instantiation from §5.2 on the statebased counter, LWW-register and OR-set. In [12, §A] we also give proofs of two multi-value register implementations.

Counter: Dctr in Fig. 2(b) and Fctr in (3). Discharging the obligations in Fig. 5 for the simulation ( 6) is easy. The key case is RECEIVE, where the first conjunct of agree in Fig. 6 ensures that min{v(s), v ′ (s)} increments by a replica s are taken into account both in v(s) and in v ′ (s): send

♯ (I) = (I, I), if inv(I) receive ♯ (I, J) = I ⊔ J, if inv(I) ∧ inv(J) ∧ agree(I, J)
This allows establishing receive( r, v , v ′ ) [Rr] (I ⊔ J), thus formalizing the informal justification of correctness we gave in §2.

LWW-register: Dintreg in Fig. 2(c) and Fintreg in (4). We associate a state or a message a, t with any execution that contains a wr(a) event with the timestamp t maximal among all other wr events (as per info). By inv in Fig. 6, this event is maximal in arbitration, which implies that rd returns the correct value; the other obligations are also discharged easily. Formally, ∀r. Rr = M and a, t [M] ((E, repl, obj, oper, rval, ro, vis, ar), info) ⇐⇒

(t = 0 ∧ a = 0 ∧ (¬∃e ∈ E. oper(e) = wr( ))) ∨ (t > 0 ∧ (∃e ∈ E. oper(e) = wr(a) ∧ info(e) = t) ∧ (∀f ∈ E. oper(f ) = wr( ) =⇒ info(f ) ≤ t)).
Optimized OR-set: Dorset in Fig. 7 and Forset in [START_REF] Batty | Mathematizing C++ concurrency[END_REF]. A problem with implementing a replicated set is that we often cannot discard the information about an element from a replica state after it has been removed: if another replica unaware of the removal sends us a snapshot of its state containing this element, the semantics of the set may require our receive to keep the element out of the set. As we prove in §6, for the OR-set keeping track of information about removed elements cannot be fully avoided, which makes its space-efficient implementation very challenging. Here we consider a recently-proposed OR-set implementation [START_REF] Bieniusa | An optimized conflict-free replicated set[END_REF] that, as we show in §6, has an optimal space complexity. It improves on the original implementation [START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF], whose complexity was suboptimal (we have proved the correctness of the latter as well; see [12, §A]).

An additional challenge posed by the OR-set is that, according to Forset, a remove operation may behave differently with respect to different events adding the same element to the set, depending on whether it sees them or not. This causes the implementation to treat internally each add operation as generating a unique instance of the element being added, further increasing the space required. To combat this, the implementation concisely summarizes information about instances. An instance is represented by a unique instance identifier that is generated when a replica performs an add and consists of the replica identifier and the number of adds (of any elements) performed at the replica until then. In a state r, V, w , the vector w determines the identifiers of all instances that the current replica r has ever observed: for any replica s, the replica r has seen w(s) successive identifiers (s, 1), (s, 2), . . . , (s, w(s)) generated at s. To generate a new identifier in do(add(a ′ )), the replica r increments w(r). The connection between the vector w in a state or a message and add events e s,k in corresponding executions is formalized in lines 1-3 of the simulation relation, also shown in Fig. 7. In receive we take the pointwise maximum of the two vectors w and w ′ . Like for the counter, the first conjunct of agree implies that this preserves the clauses in lines 1-3. Optimized OR-set implementation [START_REF] Bieniusa | An optimized conflict-free replicated set[END_REF] and its simulation

Σ = ReplicaID × ((Z × ReplicaID) → N 0 ) × (ReplicaID → N 0 ) σ 0 = λr. r, (λa, s. 0), (λs. 0) M = ((Z × ReplicaID) → N 0 ) × (ReplicaID → N 0 ) do(add(a ′ ), r, V, w , t) = ( r, (λa, s. if a = a ′ ∧ s = r then w(r) + 1 else V (a, s)), w[r → w(r) + 1] , ⊥) do(remove(a ′ ), r, V, w , t) = ( r, (λa, s. if a = a ′ then 0 else V (a, s)), w , ⊥) do(rd, r, V, w , t) = ( r, V, w , {a | ∃s. V (a, s) > 0}) send( r, V, w ) = ( r, V, w , V, w ) receive( r, V, w , V ′ , w ′ ) = r, (λa, s. if (V (a, s) = 0 ∧ w(s) ≥ V ′ (a, s)) ∨ (V ′ (a, s) = 0 ∧ w ′ (s) ≥ V (a, s)) then 0 else max{V (a, s), V ′ (a, s)}), (λs. max{w(s), w ′ (s)}) s, V, w [Rr] I ⇐⇒ (r = s) ∧ ( V, w [M] I) V, w [M] ((E, repl, obj, oper, rval, ro, vis, ar), info) ⇐⇒ 1: ∃ distinct e s,k . ({e s,k | s ∈ ReplicaID ∧ 1 ≤ k ≤ w(s)} = 2: {e ∈ E | oper(e) = add( )}) ∧ 3: (∀s, k, j. (repl(e s,k ) = s) ∧ (e s,j ro -→ e s,k ⇐⇒ j < k)) ∧ 4: (∀a, s. (V (a, s) ≤ w(s)) ∧ (V (a, s) = 0 =⇒ 5:
(oper(e s,V (a,s) ) = add(a)) ∧ 6:

(¬∃k. V (a, s) < k ≤ w(s) ∧ oper(e s,k ) = add(a)) ∧ 7: (¬∃f ∈ E. oper(f ) = remove(a) ∧ e s,V (a,s) vis -→ f ))) ∧ 8: (∀a, s, k. e s,k ∈ E ∧ oper(e s,k ) = add(a) =⇒ 9: (k ≤ V (a, s) ∨ ∃f ∈ E. oper(f ) = remove(a) ∧ e s,k vis -→ f ))
The component w in r, V, w records identifiers of both of those instances that have been removed and those that are still in the set (are active). The component V serves to distinguish the latter. As it happens, we do not need to store all active instances of an element a: for every replica s, it is enough to keep the last active instance identifier generated by an add(a) at this replica. If V (a, s) = 0, this identifier is (s, V (a, s)); if V (a, s) = 0, all instances of a generated at s that the current replica knows about are inactive. The meaning of V is formalized in the simulation: each instance identifier given by V is covered by w (line 4) and, if V (a, s) = 0, then the event e s,V (a,s) performs add(a) (line 5), is the last add(a) by replica s (line 6) and has not been observed by a remove(a) (line 7). Finally, the add(a) events that are not seen by a remove(a) in the execution are either the events e s,V (a,s) or those superseded by them (lines 8-9). This ensures that returning all elements with an active instance in rd matches Forset.

When a replica r performs do(add(a ′ )), we update V (a ′ , r) to correspond to the new instance identifier. Conversely, in do(remove(a ′ )), we clear all entries in V (a ′ ), thereby deactivating all instances of a ′ . However, after this their identifiers are still recorded in w, and so we know that they have been previously removed. This allows us to address the problem with implementing receive we mentioned above: if we receive a message with an active instance (s, V ′ (a, s)) of an element a that is not in the set at our replica (V (a, s) = 0), but previously existed (w(s) ≥ V ′ (a, s)), this means that the instance has been removed and should not be active in the resulting state (the entry for (a, s) should be 0). We also do the same check with the state and the message swapped.

As the above explanation shows, our simulation relations are useful not only for proving correctness of data type implementations, but also for explaining their designs. Discharging obligations in Fig. 5 requires some work for the OR-set; due to space constraints, we defer this to [12, §A].

′ .act(e) = do ∧ C ′ .repl(e) = r step(C ′ , e, D) = D[r → I, C ′ .msg(e) → J], if C ′ .act(e) = send ∧ C ′ .repl(e) = r ∧ C ′ .msg(e) ∈ dom(D) ∧ send ♯ (D(r)) = (I, J) step(C ′ , e, D) = D[r → receive ♯ (D(r), D(C ′ .msg(e)))], if C ′ .act(e) = receive ∧ C ′ .repl(e) = r

Soundness and Establishing Agreement Properties

We present conditions on AInfo and abstract methods ensuring the soundness of replication-aware simulations over them and, in particular, establishing the agreement property and execution invariants assumed via the definedness of abstract operations in Fig. 5. THEOREM 9 (Soundness). Assume AInfo, do ♯ , send ♯ , receive ♯ , V and T that satisfy the conditions in Fig. 9 for some G. If there exists a replication-aware simulation between Dτ and Fτ with respect to these parameters, then Dτ sat[V, T ] Fτ . Conditions in Fig. 9 require global reasoning, but can be discharged once for a class of data types. For example, they hold of the instantiation for state-based implementations from §5.2, as well as one for op-based implementations presented in [12, §B]. THEOREM 10. There exists G such that, for all Dτ , the parameters in Fig. 6 satisfy the conditions in Fig. 9 with respect to this G, V = V state , T = T-Any.

The proofs of Theorems 9 and 10 are given in [12, §B]. To explain the conditions in Fig. 9, here we consider the proof strategy for Theorem 9. To establish Dτ sat[V, T ] Fτ , for any C ∈ Dτ ∩ T we need to show abs(C, V) |= [τ → Fτ ]. We prove this by induction on the length of C. To use the localized conditions in Fig. 5, we require a relation G associating C with a decomposed execution-a partial function D : (ReplicaID ∪ MessageID) ⇀ IEx that gives fragments of abs(C, V) corresponding to replica states and messages in the final configuration of C. We write DEx for the set of all decomposed executions, so that G ⊆ CEx[x] × DEx. The existence of a decomposed execution D such that C [G] D forms the core of our induction hypothesis. G-CTXT in Fig. 9 checks that the abstract methods construct visibility according to V: it requires the context of any event e by a replica r to be the same in D(r) and abs(C, V). Together with DO in Fig. 5, this ensures

abs(C, V) |= [τ → Fτ ].
We write R,T )); see Fig. 4(c). Then the decomposed execution D ′ corresponding to C ′ is given by step(C ′ , e, D), where the function step in Fig. 8 mirrors the effect of the event e from C ′ in D using the abstract methods. G-STEP ensures that it preserves the relation G. Crucially, G-STEP also requires us to establish the definedness of step and thus the corresponding abstract method. This justifies the agreement property and execution invariants encoded by the definedness and allows us to use the conditions in Fig. 5 to complete the induction. We also require G-INIT, which establishes the base case, and G-VIS, which formulates a technical restriction on V. Finally, the conditions in Fig. 9 allow us to use the transport specification T by considering only executions C satisfying it.

C ′ ∼ (C e -→ (R, T )) when C ′ is an extension of C in the following sense: C ′ .E = C.E ⊎ {e}, the other compo- nents of C are those of C ′ restricted to C.E, e is last in C ′ .eo and C ′ .post(e) = (R, T ). For the induction step, assume C [G] D and C ′ ∼ (C e -→ (

Space Bounds and Implementation Optimality

Object states in replicated data type implementations include not only the current client-observable content, but also metadata

C, C ′ ∈ CEx[x] ∩ T , D ∈ DEx, r ∈ ReplicaID, e ∈ Event, (R, T ) ∈ Config. G-CTXT: (C [G] D ∧ e ∈ abs(C, V).E ∧ abs(C, V).repl(e) = r) =⇒ ctxt(D(r).A, e) = ctxt(abs(C, V), e) G-STEP: (C ′ ∼ (C e -→ (R, T )) ∧ (C [G] D)) =⇒ (step(C ′ , e, D)↓ ∧ C ′ [G] step(C ′ , e, D)) G-INIT: (C.E = {e} ∧ C.pre(e) = ( , [ ])) =⇒ (step(C, e, D ∅ )↓ ∧ C [G] step(C, e, D ∅ )), where D ∅ is such that dom(D ∅ ) = ReplicaID ∧ ∀r ∈ ReplicaID. D ∅ (r).E = ∅ G-VIS: (e ∈ abs(C, V).E ∧ (C is a prefix of C ′ under C ′ .eo)) =⇒ ctxt(abs(C, V), e) = ctxt(abs(C ′ , V), e)
needed for conflict resolution or masking network failures. Space taken by this metadata is a major factor determining their efficiency and feasibility. As illustrated by the OR-set in §5.3, this is especially so for state-based implementations, i.e., those that satisfy their data type specifications with respect to the visibility witness V state and the transport layer specification T-Any. We now present a general technique for proving lower bounds on this space overhead, which we use to prove optimality of four state-based implementations (we leave other implementation classes for future work; see §9). As in §5, we only consider executions over a fixed object x of type τ .

Metadata Overhead

To measure space, we need to consider how data are represented. An encoding of a set S is an injective function enc : S → Λ + , where Λ is some suitably chosen fixed finite set of characters (left unspecified). Sometimes, we clarify the domain being encoded using a subscript: e.g., enc N 0 (1). For s ∈ S, we let lenS(s) be the length of encS(s). The length can vary: e.g., for an integer k, len N 0 (k) ∈ Θ(lg k). We use standard encodings (listed in [12, §C]) for return values enc Valτ of the data types τ we consider and assume an arbitrary but fixed encoding of object states encD τ .Σ.

To distinguish metadata from the client-observable content of the object, we assume that each data type has a special rd operation that returns the latter, as is the case in the examples considered so far. For a concrete execution C ∈ Dτ over the object x and a read event e ∈ (C.E)|rd, we define state(e) to be the state of the object accessed at e: state(e) = R(x, C.repl(e)) for (R, ) = C.pre(e).

We now define the metadata overhead as a ratio, by dividing the size of the object state by the size of the observable state. We then quantify the worst-case overhead by taking the maximum of this ratio over all read operations in all executions with given numbers of replicas n and update operations m. To define the latter, we assume that each data type τ specifies a set Upd We consider only executions with m ≥ n, since we are interested in the asymptotic overhead of executions where all replicas are mutated (i.e., perform at least one update operation). Θ(n 2 lg m) † Assuming timestamp encoding is O(lg m), satisfied by Lamport clocks. ‡ An optimization of [START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF] discovered during the optimality proof. DEFINITION • f is an asymptotically tight bound (Dτ ∈ Θ(f (n, m))) if it is both an upper and a lower asymptotic bound. Fig. 10 summarizes our results; as described in §5, we have proved all the implementations correct. Matching lower and upper bounds indicate worst-case optimality of an implementation (note that this is different from optimality in all cases). The derivation of upper bounds relies on standard techniques and is deferred to [12, §C]. We now proceed to the main challenge: how to derive lower bounds that apply to any implementation of τ . We present proofs for ctr and orset; intreg and mvr are covered in [12, §C].

Experiment Families

The goal is to show that for any correct implementation Dτ (i.e., such that Dτ sat[V state , T-Any] Fτ ), the object state must store some minimum amount of information. We achieve this by constructing an experiment family, which is a collection of executions Cα, where α ∈ Q for some index set Q. The proof is the same as that of Lemma 16, except for obligation (i). We therefore give only informal explanations.

The main idea of the experiments defined in the lemma is to force replica 1 to remember element instances even after they have been removed at that replica; cf. our explanation of the challenges of implementing the OR-set from §5.3. The experiments follow a similar pattern to those for ctr, but use different operations. In the common init phase, each replica 2..n performs m-1 n-1 operations adding a designated element 0, which are interleaved with sending messages. In the experiment phase exp(α), one message from each replica r = 2..n, selected by α(r), is delivered to replica 1. At the end of execution, replica 1 removes 0 from the set and performs the read eα. The return value of this read is always the empty set.

To perform the read-back of α(r) for r = 2..n, test(r) delivers all messages by replica r to replica 1 in the order they were sent and, after each such delivery, checks if replica 1 now reports the element 0 as part of the set. From Dorset sat[V state , T-Any] Forset and the definition (5) of Forset, we get that exactly the first α(r) such deliveries will have no effect on the contents of the set: the respective add operations have already been observed by the remove operation that replica 1 performed in the experiment phase. Thus, if 0 appears in the set right after delivering the i-th message of replica r, then α(r) = i -1, and if 0 does not appear by the time the loop is finished, then α(r) = (m -1)/(n -1).

PROOF OF THEOREM 17. Given n0, m0, we pick n = n0 and some m ≥ n0 such that (m -1) is a multiple of (n -1) and m ≥ n 2 . Take the experiment family (Q, n, m, C, e, rb) given by Lemma 18. For any α ∈ Q, Cα.rval(eα) = ∅, which can be encoded with a constant length. Using Lemma 14 and m ≥ n 2 , for some constants K1, K2, K we get:

wcmo(Dorset, n, m) ≥ ⌊lg |Λ| |Q|⌋ / (maxα∈Q len(Cα.rval(eα))) ≥ K1n lg(m/n) ≥ K2n lg √ m ≥ Kn lg m. ⊓ ⊔

Store Correctness and Consistency Axioms

Recall that we define a replicated store by a data type library D and a transport layer specification T ( §4), and we specify its behavior by a function F from types τ ∈ dom(D) to data type specifications and a set of consistency axioms ( §3). The axioms are just constraints over abstract executions, such as those shown in Fig. 13; from now on we denote their sets by X. So far we have concentrated on single data type specifications F(τ ) and their correspondence to implementations D(τ ), as stated by Def. 7. In this section we consider consistency axioms and formulate the notion of correctness of the whole store (D, T ) with respect to its specification (F, X).

Our first goal is to lift the statement of correctness given by Def. 7 to a store (D, T ) with multiple objects of different data types. To this end, we assume a function V mapping each type τ ∈ dom(D) to its visibility witness V. This allows us to construct the visibility relation for a concrete execution C ∈ D ∩ T by applying V(τ ) to its projection onto the events on every object of type τ :

witness(V) = λC. {V(type(x))(C|x) | x ∈ Obj},
where •|x projects to events over x. Then the correctness of every separate data type τ in the store with respect to F(τ ) according to Def. 7 automatically ensures that the behavior of the whole store is consistent with F in the sense of Def. We showed how to discharge (i) in §5. The validity of axioms X required by (ii) most often depends on the transport layer specification T : e.g., to disallow the anomaly (2) from §1, T needs to provide guarantees on how messages pertaining to different objects are delivered. However, data type implementations can also enforce axioms by putting enough information into messages: e.g., implementations correct with respect to V state from §4 ensure that vis is transitive regardless of the behavior of the transport layer. Fortunately, to establish (ii) in practice, we do not need to consider the internals of data type implementations in D-just knowing the visibility witnesses used in the statements of their correctness is enough, as formulated in the following definition. DEFINITION 21. A set W of visibility witnesses and a transport layer specification T validate axioms X, if

∀C, V. (C ∈ T ) ∧ ({V(τ ) | τ ∈ dom(V)} ⊆ W ) =⇒ (abs(C, witness(V)) |= X).
Since visibility witnesses are common to wide classes of data types (e.g., state-or op-based), our proofs of the validity of axioms will not have to be redone if we add new data type implementations to the store from a class already considered. We next present axioms formalizing several variants of eventual consistency used in replicated stores (Fig. 13 and14) and W and T that validate them. We then use this as a basis for discussing connections with weak shared-memory models. Due to space constraints, we defer technical details and proofs to [12, §D].

Basic eventual consistency. EVENTUAL and THINAIR define a weak form of eventual consistency. EVENTUAL ensures that an event cannot be invisible to infinitely many other events on the same object and thus implies (1) from §1: informally, if updates stop, then reads at all replicas will eventually see all updates and will return the same values ( §3.2). However, EVENTUAL is stronger than quiescent consistency: the latter does not provide any guarantees at all for executions with infinitely many updates to the store, whereas our specification implies that the return values are computed according to F(τ ) using increasingly up-to-date view of the store state. We formalize these relationships in [12, §D].

THINAIR prohibits values from appearing "out-of-thinair" [START_REF] Manson | The Java memory model[END_REF], like 42 in Fig. 14(a) (recall that registers are initialized to 0). Cycles in ro ∪ vis that lead to out-of-thin-airs usually arise x, y : intreg i = x.rd j = y.rd y.wr(i)

x.wr(j)

x.rd: 42 y.wr(42) y.rd: 42

x.wr(42) ro ro vis vis (b) Disallowed by POCV:

x : orset x.add(1) i = x.rd j = x.rd x.add(2) x.add [START_REF] Alglave | Software verification for weak memory via program transformation[END_REF] x.add [START_REF]Riak key-value store[END_REF] x.add(2) ro x.rd: {2}

x.add [START_REF] Alglave | Software verification for weak memory via program transformation[END_REF] x.rd: {3} vis vis ro (c) Allowed by COCV and COCA:

x, y : intreg x.wr [START_REF]Riak key-value store[END_REF] y.wr(1) i = y.rd j = x.rd

x.wr [START_REF]Riak key-value store[END_REF] y.rd: 0 y.wr [START_REF]Riak key-value store[END_REF] x.rd: 0 ro ro from effects of speculative computations, which are done by some older replicated stores [START_REF] Terry | Managing update conflicts in Bayou, a weakly connected replicated storage system[END_REF]. THINAIR is validated by {V state , V op } and T-Any, and EVEN-TUAL by {V state , V op } and the following condition on C ensuring that every message is eventually delivered to all other replicas and every operation is followed by a message generation:

(∀e ∈ C.E. ∀r, r ′ . C.act(e) = send ∧ C.repl(e) = r ∧ r = r ′ =⇒ ∃f. C.repl(f ) = r ′ ∧ e del(C) ----→ f ) ∧ (∀e ∈ C.E. C.act(e) = do =⇒ ∃f. act(f ) = send ∧ e roo(C) ----→ f ),
where roo(C) is ro(C) projected to events on the same object.

Causality guarantees. Many replicated stores achieve availability and partition tolerance while providing stronger guarantees, which we formalize by the other axioms in Fig. 13. We call an execution per-object, respectively, cross-object causally consistent, if it is eventually consistent (as per above) and satisfies the axioms POCV and POCA, respectively, COCV and COCA. POCV guarantees that an operation sees all operations connected to it by a causal chain of events on the same object; COCV also considers causal chains via different objects. Thus, POCV disallows the execution in Fig. 14(b), and COCV the one in §3.1, corresponding to (2) from §1. POCA and COCA similarly require arbitration to be consistent with causality. The axioms highlight the principle of formalizing stronger consistency models: including more edges into vis and ar, so that clients have more up-to-date information.

Cross-object causal consistency is implemented by, e.g., COPS [START_REF] Lloyd | Don't settle for eventual: scalable causal consistency for wide-area storage with COPS[END_REF] and Gemini [START_REF] Li | Making geo-replicated systems fast if possible, consistent when necessary[END_REF]. It is weaker than strong consistency, as it allows reading stale data. For example, it allows the execution in Fig. 14(c), where both reads fetch the initial value of the register, despite writes to it by the other replica. It is easy to check that this outcome cannot be produced by any interleaving of the events at the two replicas, and is thus not strongly consistent.

An interesting feature of per-object causal consistency is that state-based data types ensure most of it just by the definition of V state : POCV is validated by {V state } and T-Any. If the witness set is {V state , V op }, then we need T to guarantee the following: informally, if a send event e and another event f are connected by a causal chain of events on the same object, then the message created by e is delivered to C.repl(f ) by the time f is done. POCA is validated by {V state , V op } and the transport layer specification (roo(C) ∪ del(C)) + | do ⊆ ar(C). This states that timestamps of events on every object behave like a Lamport clock [START_REF] Lamport | Time, clocks, and the ordering of events in a distributed system[END_REF]. Conditions for COCV and COCA are similar.

There also exist consistency levels in between basic eventual consistency and per-object causal consistency, defined using socalled session guarantees [START_REF] Terry | Session guarantees for weakly consistent replicated data[END_REF]. We cover them in [12, §D].

Comparison with shared-memory consistency models. Interestingly, the specializations of the consistency levels defined by the axioms in Fig. 13 to the type intreg of LWW-registers are very close to those adopted by the memory model in the 2011 C and C++ standards [START_REF] Batty | Mathematizing C++ concurrency[END_REF]. Thus, POCA and POCV define the semantics of the fragment of C/C++ restricted to so-called relaxed operations; there this semantics is defined using coherence axioms, which are analogous to session guarantees [START_REF] Terry | Session guarantees for weakly consistent replicated data[END_REF]. COCV and COCA are close to the semantics of C/C++ restricted to release-acquire operations. However, C/C++ does not have an analog of EVENTUAL and does not validate THINAIR, since it makes the effects of speculations visible to the programmer [START_REF] Batty | Library abstraction for C/C++ concurrency[END_REF]. We formalize the correspondence to C/C++ in [12, §D]. In the future, this correspondence may open the door to applying technology developed for shared-memory models to eventually consistent systems; promising directions include model checking [START_REF] Alglave | Software verification for weak memory via program transformation[END_REF][START_REF] Burckhardt | Checkfence: checking consistency of concurrent data types on relaxed memory models[END_REF], automatic inference of required consistency levels [START_REF] Liu | Dynamic synthesis for relaxed memory models[END_REF] and compositional reasoning [START_REF] Batty | Library abstraction for C/C++ concurrency[END_REF].

Related Work

For a comprehensive overview of replicated data type research we refer the reader to Shapiro et al. [START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF]. Most papers proposing new data type implementations [START_REF] Bieniusa | An optimized conflict-free replicated set[END_REF][START_REF] Roh | Replicated abstract data types: Building blocks for collaborative applications[END_REF][START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF][START_REF] Shapiro | Conflictfree replicated data types[END_REF] do not provide their formal declarative specifications, save for the expected property (1) of quiescent consistency or first specification attempts for sets [START_REF] Bieniusa | An optimized conflict-free replicated set[END_REF][START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF]. Formalizations of eventual consistency have either expressed quiescent consistency [START_REF] Bosneag | A formal model for eventual consistency semantics[END_REF] or gave low-level operational specifications [START_REF] Fekete | Eventually-serializable data services[END_REF].

An exception is the work of Burckhardt et al. [START_REF] Burckhardt | Cloud types for eventual consistency[END_REF][START_REF] Burckhardt | Eventually consistent transactions[END_REF], who proposed an axiomatic model of causal eventual consistency based on visibility and arbitration relations and an operational model based on revision diagrams. Their store specification does not provide customizable consistency guarantees, and their data type specifications are limited to the sequential S construction from §3.2, which cannot express advanced conflict resolution used by the multi-value register, the OR-set and many other data types [START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF]. More significantly, their operational model does not support general op-or state-based implementations, and is thus not suited for studying the correctness or optimality of these commonly used patterns.

Simulation relations have been applied to verify the correctness of sequential [START_REF] Liskov | Programming with abstract data types[END_REF] and shared-memory concurrent data type implementations [START_REF] Liang | A rely-guarantee-based simulation for verifying concurrent program transformations[END_REF]. We take this approach to the more complex setting of a replicated store, where the simulation needs to take into account multiple object copies and messages and associate them with structures on events, rather than single abstract states. This poses technical challenges not considered by prior work, which we address by our novel notion of replication-aware simulations.

The distributed computing community has established a number of asymptotic lower bounds on the complexity of implementing certain distributed or concurrent abstractions, including one-shot timestamp objects [START_REF] Helmi | The space complexity of long-lived and one-shot timestamp implementations[END_REF] and counting protocols [START_REF] Chen | Optimal gossip-based aggregate computation[END_REF][START_REF] Moran | Concurrent counting[END_REF]. These works have considered either programming models or metrics significantly different from ours. An exception is the work of Charron-Bost [START_REF] Charron-Bost | Concerning the size of logical clocks in distributed systems[END_REF], who proved that the size of vector clocks [START_REF] Mattern | Virtual time and global states of distributed systems[END_REF] is optimal to represent the happens-before relation of a computation (similar to the visibility relation in our model). Specifications of mvr and orset rely on visibility; however, Charron-Bost's result does not directly translate into a lower bound on their implementation complexity, since a specification may not require complete knowledge about the relation and an implementation may represent it in an arbitrary manner, not necessarily using a vector.

Conclusion and Future Work

We have presented a comprehensive theoretical toolkit to advance the study of replicated eventually consistent stores, by proposing methods for (1) specifying the semantics of replicated data types and stores abstractly, (2) verifying implementations of replicated data types, and (3) proving that such implementations have optimal metadata overhead. By proving both correctness and optimality of four nontrivial data type implementations, we have demonstrated that our methods can indeed be productively applied to the kinds of patterns used by practitioners and researchers in this area.

Although our work marks a big step forward, it is only a beginning, and creates plenty of opportunities for future research. We have already made the first steps in extending our specification framework with more features, such as mixtures of consistency levels [START_REF] Li | Making geo-replicated systems fast if possible, consistent when necessary[END_REF] and transactions [START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF][START_REF] Zawirski | SwiftCloud: Fault-tolerant georeplication integrated all the way to the client machine[END_REF]; see [START_REF] Burckhardt | Understanding eventual consistency[END_REF]. In the future we would also like to study more data types, such as lists used for collaborative editing [START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF], and to investigate metadata bounds for data type implementations other than state-based ones, including more detailed overhead metrics capturing optimizations invisible to the worst-case overhead analysis. Even though our execution model for replicated stores follows the one used by replicated data type designers [START_REF] Shapiro | Conflictfree replicated data types[END_REF], there are opportunities for bringing it closer to actual implementations. Thus, we would like to verify the algorithms used by store implementations [START_REF] Lloyd | Don't settle for eventual: scalable causal consistency for wide-area storage with COPS[END_REF][START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF][START_REF] Zawirski | SwiftCloud: Fault-tolerant georeplication integrated all the way to the client machine[END_REF] that our semantics abstracts from. This includes fail-over and session migration protocols, which permit clients to interact with multiple physical replicas, while being provided the illusion of a single virtual replica.

Finally, by bringing together prior work on shared-memory models and data replication, we wish to promote an exchange of ideas and results between the research communities of programming languages and verification on one side and distributed systems on the other.

Figure 1 .vis

 1 Figure 1. Illustrations of a concrete (a) and two abstract executions (b, c) 1: x.inc 2: send 3: receive 4: x.inc 5: send 6: receive 7: x.rd 8: receive to r3

Figure 2 .

 2 Figure 2. Three replicated data type implementations (a) Op-based counter (ctr) Σ = N 0 × N 0 M = N 0 σ 0 = λr. 0, 0

  (b) justifies the op-based implementation in Fig.2(a) reading 1 in the concrete execution in Fig.1(a). The abstract execution in Fig.1(c) justifies the state-based implementation in Fig.2(b) reading 2 due to transitive delivery ( §2). There is no abstract execution that would justify reading 100.

  ctxt(A, e) = (A.oper(e), G, (A.oper)|G, (A.vis)|G, (A.ar)|G),where G = (A.vis) -1 (e) and •|G is the restriction to events in G. Thus, in the abstract execution in Fig.1(b), the operation context of the read from x includes only one increment event; in the execution in Fig.1(c) it includes two. DEFINITION 4. A replicated data type specification for a type τ is a function Fτ that, given an operation context L for τ , specifies a return value Fτ (L) ∈ Valτ .

  Fmvr(rd, E, oper, vis, ar) = {a | ∃e ∈ E. oper(e) = wr(a) ∧ ¬∃f ∈ E. oper(f ) = wr( ) ∧ e vis -→ f }. wr(0) wr(2) vis wr(3) wr(1) vis vis

Figure 3 .

 3 Figure 3. The set of configurations Config and the transition relation -→ D : Config × Event × Config for a data type library D. We use e : {h 1 = u 1 , h 2 = u 2 } to abbreviate h 1 (e) = u 1 and h 2 (e) = u 2 . We uncurry R ∈ RState where convenient.

  V, which maps a concrete execution C ∈ Dτ to an acyclic relation on (C.E)| do defining visibility (here •| do is the restriction to events e with C.act(e) = do). Let e ro(C) ---→ f ⇐⇒ e C.eo ---→ f ∧ C.repl(e) = C.repl(f ); e ar(C) ---→ f ⇐⇒ e, f ∈ (C.E)| do ∧ C.obj(e) = C.obj(f ) ∧ C.time(e) < C.time(f ). Then the abstract execution justifying C ∈ Dτ is defined by abs(C, V) = (C.E| do , E.repl| do , E.obj| do , E.oper| do , E.rval| do , ro(C)| do , V(C), ar(C)). DEFINITION 7. A data type implementation Dτ satisfies a specification Fτ with respect to V and T , written Dτ sat[V, T ] Fτ , if ∀C ∈ Dτ ∩ T . (abs(C, V) |= [τ → Fτ ]), where |= is defined in Def. 5.

Figure 4 .

 4 Figure 4. Diagrams illustrating replication-aware simulations (a) (b) (c)

Figure 5 .

 5 Figure 5. Definition of a replication-aware simulation {Rr, M} between Dτ and Fτ . All free variables in each condition are implicitly universally quantified and have the following types: σ, σ ′ ∈ Σ, m ∈ M , I, I ′ , J ∈ IEx, e ∈ Event, r ∈ ReplicaID, o ∈ Op τ , a ∈ Valτ , t ∈ Timestamp.

INIT:

  Dτ . σ 0 (r) [Rr] I ∅ , where I ∅ .E = ∅ DO: (do ♯ (I, e, r, o, a, t) = I ′ ∧ Dτ .do(o, σ, t) = (σ ′ , a) ∧ σ [Rr] I) =⇒ (σ ′ [Rr] I ′ ∧ a = Fτ (ctxt(I ′ .A, e)))

(

  r, v [Rr] I) ∧ (v ′ [M] J) =⇒ (∀s. min{v(s), v ′ (s)} = |{e ∈ I.E ∩ J.E | I.oper(e) = inc ∧ I.repl(e) = s}|).

Figure 6 .

 6 Figure 6. Instantiation for state-based data type implementations. In do ♯ we omit the straightforward definition of ar ′ in terms of I.info and t. AInfo = Timestamp agree(I, J) ⇐⇒ ∀r. (({e ∈ I.E | I.repl(e) = r}, I.ro) is a prefix of ({e ∈ J.E | J.repl(e) = r}, J.ro) or vice versa) ∧ (I ⊔ J ∈ IEx) inv(I) ⇐⇒ (∀e, f ∈ I.E. (e, f ) ∈ I.ar ⇐⇒ I.info(e) < I.info(f )) ∧ ((I.vis ∪ I.ro) + ⊆ I.vis) do ♯ (I, e, r, o, a, t) = I ′ , if inv(I) ∧ e ∈ I.E ∧ I ′ ∈ IEx where I ′ = ((I.E ∪ {e}, I.repl[e → r], I.obj[e → x], I.oper[e → o], I.rval[e → a], I.ro ∪ {(f, e) | f ∈ I.E ∧ I.repl(f ) = r}, I.vis ∪ {(f, e) | f ∈ I.E}, ar ′ ), I.info[e → t])

Figure 7 .

 7 Figure 7. Optimized OR-set implementation[START_REF] Bieniusa | An optimized conflict-free replicated set[END_REF] and its simulation

Figure 8 .

 8 Figure 8. Function step that mirrors the effect of an event e ∈ C ′ .E from C ′ ∈ CEx[x] in D ∈ DEx, defined when so is the abstract method used step(C ′ , e, D) = D[r → do ♯ (D(r), e, r, C ′ .oper(e), C ′ .rval(e), C ′ .time(e))], if C ′ .act(e) = do ∧ C ′ .repl(e) = r step(C ′ , e, D) = D[r → I, C ′ .msg(e) → J], if C ′ .act(e) = send ∧ C ′ .repl(e) = r ∧ C ′ .msg(e) ∈ dom(D) ∧ send ♯ (D(r)) = (I, J) step(C ′ , e, D) = D[r → receive ♯ (D(r), D(C ′ .msg(e)))], if C ′ .act(e) = receive ∧ C ′ .repl(e) = r

Figure 9 .

 9 Figure 9. Proof obligations for abstract methods. Free variables are implicitly universally quantified and have the following types: C, C ′ ∈ CEx[x] ∩ T , D ∈ DEx, r ∈ ReplicaID, e ∈ Event, (R, T ) ∈ Config.

  τ ⊂ Op τ of update operations; for all examples in this paper Upd τ = Op τ \ {rd}. DEFINITION 11. The maximum metadata overhead of an execution C ∈ Dτ of an implementation Dτ is mmo(Dτ , C) = max lenD τ .Σ(state(e)) len Valτ (C.rval(e)) | e ∈ (C.E)|rd . The worst-case metadata overhead of an implementation Dτ over all executions with n replicas and m updates (2 ≤ n ≤ m) is wcmo(Dτ , n, m) = max{mmo(Dτ , C) | C ∈ Dτ ∧ n = |{C.repl(e) | e ∈ C.E}| ∧ m = |{e ∈ C.E | C.oper(e) ∈ Upd τ }|}.

Figure 10 .Fig. 17 ,

 1017 Figure 10. Summary of bounds on metadata overhead for stated-based implementations, as functions of the number of replicas n and updates m Type Existing implementation Any implementation algorithm ref. overhead overhead ctr Fig. 2(b) [32] Θ(n) Ω(n) orset Fig. 7 [6] Θ(n lg m) Ω(n lg m) Fig. 15, [12, §A] [32] Θ(m lg m) intreg Fig. 2(c) [32] Θ(lg m) † Ω(lg m) mvr Fig. 17, [12, §A] new ‡ Θ(n lg m) Ω(n lg m) Fig. 16, [12, §A] [32]Θ(n 2 lg m)

  5. PROPOSITION 19. (∀τ ∈ dom(D). D(τ ) sat[V(τ ), T ] F(τ )) =⇒ (∀C ∈ D ∩ T . abs(C, witness(V)) |= F). This motivates the following definition of store correctness. Let us write A |= X when the abstract execution A satisfies the axioms X. DEFINITION 20. A store (D, T ) is correct with respect to a specification (F, X), if for some V: (i) ∀τ ∈ dom(D). (D(τ ) sat[V(τ ), T ] F(τ )); and (ii) ∀C ∈ D ∩ T . (abs(C, witness(V)) |= X).

Figure 13 .Figure 14 .

 1314 Figure 13. A selection of consistency axioms over an execution (E, repl, obj, oper, rval, ro, vis, ar) Auxiliary relations sameobj(e, f ) ⇐⇒ obj(e) = obj(f ) Per-object causality (aka happens-before) order: hbo = ((ro ∩ sameobj) ∪ vis) + Causality (aka happens-before) order: hb = (ro ∪ vis) + Axioms EVENTUAL: ∀e ∈ E. ¬(∃ infinitely many f ∈ E. sameobj(e, f ) ∧ ¬(e vis -→ f )) THINAIR: ro ∪ vis is acyclic POCV (Per-Object Causal Visibility): hbo ⊆ vis POCA (Per-Object Causal Arbitration): hbo ⊆ ar COCV (Cross-Object Causal Visibility): (hb ∩ sameobj) ⊆ vis COCA (Cross-Object Causal Arbitration): hb ∪ ar is acyclic Figure 14. Anomalies allowed or disallowed by different axioms (a) Disallowed by THINAIR:x, y : intreg i = x.rd j = y.rd y.wr(i)x.wr(j)

  12. Assume Dτ and a positive function f (n, m). • f is an asymptotic upper bound (Dτ ∈ O(f (n, m))) if sup n,m→∞ (wcmo(Dτ , n, m)/f (n, m)) < ∞, i.e., ∃K > 0. ∀m ≥ n ≥ 2. wcmo(Dτ , n, m) < Kf (n, m);

	• f is an asymptotic lower bound (Dτ ∈ Ω(f (n, m))) if limn,m→∞(wcmo(Dτ , n, m)/f (n, m)) = 0, i.e.,
	∃K > 0. ∀m0 ≥ n0 ≥ 2. ∃n ≥ n0, m ≥ n0. wcmo(Dτ , n, m) > Kf (n, m);

  Each experiment contains a distinguished read event eα. The experiments are designed in such a way that the object states state(eα) must be distinct, which then implies a lower bound lg |Λ| |Q| on the size of their encoding. To prove that they are distinct, we construct black-box tests that execute the methods of Dτ on the states and show that the tests must produce different results for each state(eα) provided Dτ is correct. Formally, the tests induce a read-back function rb that satisfies rb(state(eα)) = α. We encapsulate the core argument in the following lemma. DEFINITION 13. An experiment family for an implementation Dτ is a tuple (Q, n, m, C, e, rb) where Q is a finite set, 2 ≤ n ≤ m, and for each α ∈ Q, Cα ∈ Dτ is an execution with n replicas and m updates, eα ∈ (Cα.E)|rd and rb : Dτ .Σ → Q is a function satisfying rb(state(eα)) = α. LEMMA 14. If (Q, n, m, C, e, rb) is an experiment family, then wcmo(Dτ , n, m) ≥ ⌊lg |Λ| |Q|⌋ / (maxα∈Q len(Cα.rval(eα))).PROOF. Since rb(state(eα)) = α, the states state(eα) are pairwise distinct and so are their encodings enc(state(eα)). Since there are fewer than |Q| strings of length strictly less than ⌊lg |Λ| |Q|⌋, for Figure 12. Example experiment (n = 4 and m = 15) and test for ctr. Gray dashed lines represent the configuration (R init [(x, 1) → Rα(x, 1)], T init ) where the test driver program is applied. Then the operation context in abs(C ′ α , V state ) of the first read in test(r) contains n r=2 α(r) increments, while that of the second read contains (m/(n -1))α(r) more increments. (ii) We have Tα = Tinit because exp(α) does not send any messages. Also, Rα and Rinit [(x, 1) → Rα(x, 1)] can differ only in the states of the replicas 2..n. These cannot influence the run of test(r), since it performs events on replica 1 only. ⊓ ⊔ PROOF OF THEOREM 15. Given n0, m0, we pick n = n0 and some m ≥ n0 such that m is a multiple of (n -1) and m ≥ n 2 . Take the experiment family (Q, n, m, C, e, rb) given by Lemma 16. Then for any α, Cα.rval(eα) is at most the total number of increments m in Cα. Using Lemma 14 and m ≥ n 2 , for some constants K1, K2, K3, K independent from n0, m0 we get: If Dorset sat[V state , T-Any] Forset, then Dorset is Ω(n lg m). LEMMA 18. If Dorset sat[V state , T-Any] Forset, n ≥ 2 and m ≥ n is such that (m -1) is a multiple of (n -1), then the tuple (Q, n, m, C, e, rb) on the right in Fig. 11 is an experiment family.

	wcmo(Dctr, n, m) ≥ ⌊lg |Λ| |Q|⌋/(maxα∈Q len(Cα.rval(eα))) ≥ K1 lg |Λ| ( m n-1 ) n-1 len N 0 (m) ≥ K2 n lg(m/n) lg m ≥ K3 n lg √ m lg m ≥ Kn. ⊓ ⊔
	6.4 Lower Bound for State-Based OR-Set (orset)
	THEOREM 17.
		1	2		3		4
	init	 (2)=3	inc mid 2,3 : send inc inc mid 2,2 : send inc mid 2,1 : send	 (3)=4	inc mid 3,3 : send inc inc mid 3,2 : send inc mid 3,1 : send	 (4)=5	inc mid 4,3 : send inc inc mid 4,2 : send inc mid 4,1 : send
			mid 2,4 : send inc		mid 3,4 : send inc		mid 4,4 : send inc
			mid 2,5 : send		mid 3,5 : send		mid 4,5 : send
	exp()	e  : 12 rd receive receive receive
	test(2)	u 12 rd receive u'14 rd
	where:
	(i) This is due to Dctr sat[V state , T-Any] Fctr, as we explained informally above. Let
		C ′ α = exec(Dctr, (R0, T0), (init; exp(α); test(r))).

Acknowledgements. We thank Hagit Attiya, Anindya Banerjee, Carlos Baquero, Lindsey Kuper and Marc Shapiro for comments that helped improve the paper. Gotsman was supported by the EU FET project ADVENT, and Yang by EPSRC.

. ⊓ ⊔

To apply this lemma to the best effect, we need to find experiment families with |Q| as large as possible and len(C α ′ .rval(e α ′ )) as small as possible. Finding such families is challenging, as there is no systematic way to derive them. We relied on intuitions about "which situations force replicas to store a lot of information" when searching for experiment families.

Driver programs. We define experiment families using driver programs (e.g., see Fig. 11). These are written in imperative pseudocode and use traditional constructs like loops and conditionals. As they execute, they construct concrete executions of the data type library [τ → Dτ ] by means of the following instructions, each of which triggers a uniquely-determined transition from Fig. 3:

do operation o on x at replica r with timestamp t u ← dor o t same, but assign the return value to u sendr(mid ) send a message for x with identifier mid at r receiver(mid ) receive the message mid at replica r Programs explicitly supply timestamps for do and message identifiers for send and receive. We require that they do this correctly, e.g., respect uniqueness of timestamps. When a driver program terminates, it may produce a return value. For a program P , an implementation Dτ , and a configuration (R, T ), we let exec(Dτ , (R, T ), P ) be the concrete execution of the data type library [τ → Dτ ] starting in (R, T ) that results from running P ; we define result(Dτ , (R, T ), P ) as the return value of P in this run. We start by formulating a suitable experiment family. LEMMA 16. If Dctr sat[V state , T-Any] Fctr, n ≥ 2 and m ≥ n is a multiple of (n -1), then tuple (Q, n, m, C, e, rb) as defined in the left column of Fig. 11 is an experiment family.

The idea of the experiments is to force replica 1 to remember one number for each of the other replicas in the system, which then introduces an overhead proportional to n; cf. the implementation in Fig. 2(b). We show one experiment in Fig. 12. All experiments start with a common initialization phase, defined by init, where each of the replicas 2..n performs m/(n -1) increments and sends a message after each increment. All messages remain undelivered until the second phase, defined by exp(α). There replica 1 receives exactly one message from each replica r = 2..n, selected using α(r). An experiment concludes with the read eα on the first replica.

The read-back works by performing separate tests for each of the replicas r = 2..n, defined by test(r). For example, to determine which message was sent by replica 2 during the experiment in Fig. 12, the program test(2): reads the counter value at replica 1, getting 12; delivers the final message by replica 2 to it; and reads the counter value at replica 1 again, getting 14. By observing the difference, the program can determine the message sent during the experiment: α(2) = 5 -(14 -12) = 3.

PROOF OF LEMMA 16. The only nontrivial obligation is to prove rb(state(eα)) = α. Let (Rα, Tα) = final(Cα). Then α(r)

= result(Dctr, (R0, T0), (init; exp(α); test(r)))

= result(Dctr, (Rα, Tα), test(r)) where (R init , T init ) = post(exec(Dτ , (R 0 , T 0 ), init))