Peter Bailis

João Leitão

Allen Clement

Vivien Quéma

Tyler Crain

Pierpaolo Ardekani

Lisong Cincilla

Florian Guo

Alejandro David

Valter Tomsic

Balegas

Web applications rely on replicated databases to place data close to their users, and to tolerate failures. Many replication systems opt for eventual consistency, which offers excellent responsiveness and availability, but may expose applications to the complexity of concurrency and failures.

To alleviate this problem, recent replication systems have begun to strengthen their interface with additional guarantees, such as causal consistency, which protects the application from ordering anomalies, and Replicated Data Types (RDTs), which encapsulate concurrent updates via high level object interface and ensure convergent semantics. However, dependable algorithms for these abstractions come at a high cost in metadata size. This thesis studies three related topics: (i) design of RDT implementations with minimized metadata; (ii) design of consistency algorithms with minimized metadata; and (iii) the limits of the design space.

Our first contribution is a study of metadata complexity of RDTs. RDTs need metadata to provide rich semantics under concurrency and failures. Several existing implementations incur high overhead in storage space, in the number of updates, or polynomial in the number of replicas. Minimizing metadata is nontrivial without impacting their semantics or fault-tolerance.

We design optimized set and register RDTs with metadata overhead reduced to the number of replicas. We also demonstrate metadata lower bounds for six RDTs, thereby proving optimality of four implementations. As a result, RDT designers and users can better navigate the design space.

Our second contribution is the design of SwiftCloud, a replicated causally-consistent RDT object database for client-side applications, e.g., mobile or in-browser apps. We devise algorithms to support high numbers of client-side partial replicas backed by the cloud, in a fault-tolerant manner, with small and bounded metadata. We demonstrate how to support available and consistent reads and updates, at the expense of some slight data staleness; i.e., our approach trades freshness for scalability (small and bounded metadata, parallelism), and availability (ability to switch between data centers on failure). Our experiments with thousands of client replicas confirm the design goals were achieved at a low staleness cost.

iii Acknowledgement This work would not be possible without many inspiring and friendly people that I had a chance to meet and work with.

I am thankful to my advisor, Marc Shapiro, for his insights and constant

Chapter 1

Introduction

High availability and responsiveness are essential for many interactive web applications. Such applications often share mutable data between users in geographically distributed locations.

To ensure availability and responsiveness, they rely on geo-replicated databases or even on client-side data replicas. Geo-replication provides users with access to a local replica in a nearby data center, whereas client-side replication provides access to a local replica at a user's device. This local data access is neither affected by the cost of network latency between replicas nor by failures (e.g., replica disconnection) [START_REF] Burckhardt | Bringing TouchDevelop to the cloud. Inside Microsoft Research Blog[END_REF][START_REF] Cairns | Build collaborative apps with Google Drive Realtime API. Google Apps Developers Blog[END_REF][START_REF] Corbett | Spanner: Google's globally-distributed database[END_REF][START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF]. When, furthermore, performance and availability of update operations are crucial for an application, then these updates need to be also performed locally, without coordination with remote replicas. This entails that concurrent updates must be accepted at different replicas and replicated asynchronously.

Unfortunately, asynchronous replication is at odds with strong consistency models, such as linearizability or serializability [START_REF] Bernstein | Concurrency Control and Recovery in Database Systems[END_REF][START_REF] Herlihy | Linearizability: a correcteness condition for concurrent objects[END_REF]. Strong consistency offers applications a single common view of the distributed database, but requires to execute operations in a global total order at all replicas (synchronous replication). The incompatibility of availability and fault-tolerance with strong consistency, known as the CAP theorem, forces a choice of weaker consistency models, i.e., eventual consistency [START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF][START_REF] Gilbert | Brewer's conjecture and the feasibility of consistent, available, partition-tolerant web services[END_REF][START_REF] Petersen | Flexible update propagation for weakly consistent replication[END_REF].

Under eventual consistency, replicas are allowed to diverge transiently due to concurrent updates, but they are expected to eventually converge to a common state that incorporates all updates [START_REF] Petersen | Flexible update propagation for weakly consistent replication[END_REF][START_REF] Vogels | Eventually consistent[END_REF]. Intermediate states expose applications to consistency anomalies. This is undesirable for users, and poses a challenge to the implementation of the database and the application [START_REF] Corbett | Spanner: Google's globally-distributed database[END_REF]. Difficulties include detection of concurrent conflicting updates, and their convergent resolution, or protection from asynchronous delivery of updates and from failures.

Two complementary abstractions were proposed to alleviate the issues of eventually consistent replication. These abstractions encapsulate the complexity of asynchronous replication and failures behind an interface with a well-defined behavior. A dependable implementation of such an interface guarantees the behavior regardless of the underlying infrastructure (mis)behavior.

First, Replicated Data Types (RDTs) expose data in the database as typed objects with high-CHAPTER 1. INTRODUCTION level methods. RDTs rely on type and method semantics for convergence [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF][START_REF] Shapiro | Conflict-free replicated data types[END_REF]. With RDTs, the database consists of objects such as counter, set, or register, that offer read and update methods, such as increment for a counter, or add and remove for a set. A data type encapsulates convergent replication, and resolves concurrent updates according to the type's defined semantics.

Second, causal consistency offers partial ordering of updates across objects [START_REF] Ahamad | Causal memory[END_REF]66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF]. Informally, under causal consistency, every application process observes a monotonically nondecreasing set of updates, which includes its own updates, in an order that respects the causality between operations. Applications are protected from causality violations, e.g., where read could observe update b but not update a that b is based upon. Transactional causal consistency further simplifies programming with atomicity. It guarantees that all the reads of a transaction come from a same database snapshot, and either all its updates are visible, or none is [START_REF] Bailis | Scalable atomic visibility with RAMP transactions[END_REF]66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF].

These abstractions facilitate programming against a replicated database, but their dependable implementation comes at the cost of storage and network metadata. Metadata is required, for instance, to relate asynchronously replicated updates, i.e., to determine which one caused another, or to compare replica states after a transient failure. Some metadata cost is inherent to concurrency, and to uncertainty of the status of an unresponsive replica, which could be unavailable either permanently or transiently (and could possibly perform concurrent updates).

This thesis studies the design of dependable RDT and causal consistency algorithms with minimized metadata, and the limits and the trade-offs of their design space.

Contributions

This dissertation makes two main contributions in the area of dependable algorithms for eventual consistency: a study of space optimality of RDTs, including impossibility results and optimized implementations, and the design of a causally-consistent RDT database for client-side applications.

We discuss both contributions in more detail in the remainder of this section.

Optimality of Replicated Data Types

In the first part of the thesis, we consider the problem of minimizing metadata incurred by RDT implementations. In addition to client-observable data, RDTs use metadata to ensure correctness in the presence of concurrent updates and failures. The metadata manifests either in the implementation of RDTs or in a middleware for updates dissemination. A category of state-based RDT implementations, i.e., objects that communicate by exchanging their complete state [START_REF] Shapiro | Conflict-free replicated data types[END_REF], uses metadata directly in the RDT state. Such state-based implementations are used, for example, by server-side geo-replicated object databases [START_REF]Data Types in Riak[END_REF]. Some state-based RDTs incur substantial metadata overhead that impacts storage and bandwidth cost, or even the viability of an implementation. On the other hand, the design of RDT implementations that reduce the size of metadata is challenging, as it creates a tension between correctness w.r.t. type semantics, efficiency, and fault-tolerance.

We formulate and study the metadata complexity problem for state-based RDTs. We define a metadata overhead metric, and we perform a worst-case analysis in order to express asymptotic metadata complexity as a function of the number of replicas or of updates. An analysis of six existing data type implementations w.r.t. this metric indicates that most incur substantial metadata overhead, higher than required, linear in the number of updates, or square in the number of replicas. We also observe that the concurrent semantics of a data type, i.e., its behavior under concurrent updates, has a critical impact on the exact extent of the metadata complexity.

The two main contributions of this part of the thesis are positive results, i.e., optimized data type implementations, and impossibility results, i.e., lower bound proofs on the relative metadata complexity for some data type. It is natural to search for positive results first. However, it is a laborious process that requires developing nontrivial solutions, potentially incorrect ones. A lower bound sets the limits of possible optimizations and can show that an implementation is asymptotically optimal. Our lower bound proofs use a common structure that we apply to each data type using semantics-specific argument. This marks the end of the design process, saves designers from seeking the unachievable, and may lead them towards new design assumptions.

Together, both kinds of results offer a comprehensive view of the metadata optimization problem. Table 1.1 summarizes our positive and impossibility results for all studied data types. From left to right, the table lists prior implementations and their relative complexity, expressed as metadata overhead, as well as the complexity of our optimizations (if any), and our lower bounds on complexity of any implementation. We underline asymptotically optimal implementations. We show that the existing state-based implementation of counter data type is optimal. Counter requires vector of integers to handle concurrent increments and duplicated or reordered messages.

Set data type interface offers a range of different semantics choices w.r.t. how to treat concurrent operations on the same element to converge. For instance, a priority can be given to some type of operation (add or remove) or operations can be arbitrated by timestamps (last-writerwins [START_REF] Johnson | The maintenance of duplicate databases[END_REF]). These alternatives are uneven in terms of the metadata cost. Prior implementations all incur overhead in the order of the number of updates, as they keep track of the logical time of remove operations. However, this is not necessary for all set variants. Our optimized implementation of add-wins set reduces this cost to the optimum, i.e., the order of the number of replicas, by adapting a variant of version vectors to efficiently summarize observed information [START_REF] Parker | Detection of mutual inconsistency in distributed systems[END_REF]. The lower bound on remove-wins set shows that such an optimization cannot be applied on this semantics. It is an open problem whether it is possible for last-writer-wins set.

Register type also comes in different variants. Similarly, we show they are uneven in metadata complexity. A last-writer-wins register uses timestamps to arbitrate concurrent assignments, whereas multi-value register identifies values of all conflicting writes and presents them to the application. The existing implementation of last-writer-wins register has negligible overhead and is the optimal one. On the contrary, we find that the existing implementation of multi-value register has substantial overhead, square in the number of replicas, due to inefficient treatment of concurrent writes of the same value. Our optimization alleviates the square component, and reaches the asymptotically optimal overhead, using nontrivial merge rules for version vectors.

Causally-Consistent Object Database for Client-Side Applications

In the second part of the thesis, we study the problem of providing object database with extended, causal consistency guarantees across RDT objects, at the client-side.

Client-side applications, such as in-browser and mobile apps, are poorly supported by the current technology for sharing mutable data over the wide-area. App developers resort to implementing their own ad-hoc application-level cache and buffers, in order to avoid slow, costly and sometimes unavailable round-trips to a data center, but they cannot solve system issues such as fault tolerance, or consistency/session guarantees [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF][START_REF] Terry | Session guarantees for weakly consistent replicated data[END_REF]. Recent client-side systems ensure only some of the desired properties, i.e., either make only limited consistency guarantees (at the granularity of a single object or of a small database only), do not tolerate failures, and/or do not scale to large numbers of client devices [START_REF] Belaramani | PRACTI replication[END_REF][START_REF] Burckhardt | Bringing TouchDevelop to the cloud. Inside Microsoft Research Blog[END_REF][START_REF] Cairns | Build collaborative apps with Google Drive Realtime API. Google Apps Developers Blog[END_REF][START_REF] Chun | Mobius: Unified messaging and data serving for mobile apps[END_REF][START_REF] Mahajan | Depot: Cloud storage with minimal trust[END_REF]. Standard algorithms for geo-replication [START_REF] Almeida | ChainReaction: a causal+ consistent datastore based on Chain Replication[END_REF][START_REF] Bailis | Bolt-on causal consistency[END_REF]46,66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF] are not easily adaptable, because they were not designed to support high numbers of client replicas located outside of the server-side infrastructure.

Our thesis is that the system should be ensuring correct and scalable database access to clientside applications, addressing the (somewhat conflicting) requirements of consistency, availability, and convergence [START_REF] Mahajan | Consistency, availability, and convergence[END_REF], at least as well as server-side systems. Under these requirements, the strongest consistency model is transactional causal consistency with RDT objects [66,[START_REF] Mahajan | Consistency, availability, and convergence[END_REF].

Supporting thousands or millions of client-side replicas, under causal consistency with RDTs, challenges standard assumptions. To track causality precisely, per client, would create unacceptably fat metadata; but the more compact server-side metadata management approach has fault-tolerance issues. Additionally, full replication at high numbers of resource-poor devices 1.1. CONTRIBUTIONS would be unacceptable [START_REF] Belaramani | PRACTI replication[END_REF]; but partial replication of data and metadata could cause anomalous message delivery or unavailability. Furthermore, it is not possible to assume, like many previous systems [START_REF] Almeida | ChainReaction: a causal+ consistent datastore based on Chain Replication[END_REF]46,66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF], that the application is located inside the data center (DC), or has a sticky session to a single DC, to solve fault tolerance or consistency problems [START_REF] Bailis | Highly Available Transactions: Virtues and limitations[END_REF][START_REF] Terry | Session guarantees for weakly consistent replicated data[END_REF].

In the second part of the thesis, we address these challenges. We present the algorithms, design, and evaluation of SwiftCloud, the first distributed object database designed for a high number of replicas. It efficiently ensures consistent, available, and convergent access to client nodes, tolerating failures. To achieve this, SwiftCloud uses a flexible client-server topology, and decouples reads from writes. The client writes fast into the local cache, and reads in the past (also fast) data that is consistent, but occasionally stale. Our approach includes two major techniques:

1. Cloud-backed support for partial replicas. To simplify consistent partial replication at the client side and at the scale of client-side devices, we leverage the DC-side full replicas to provide a consistent view of the database to the client. The client merges this view with his own updates to achieve causal consistency. In some failure situations, a client may connect to a DC that happens to be inconsistent with its previous DC. Because the client does not have a full replica, it cannot fix the issue on its own. We leverage "reading in the past" to avoid this situation in the common case, and provide control over the inherent trade-off between staleness and unavailability: namely, a client observes a remote update only if it is stored in some number K ∏ 1 of DCs [START_REF] Mahajan | Depot: Cloud storage with minimal trust[END_REF]. The higher the value of K is, the more likely that an update is in both DCs, but the higher is the staleness.

2. Protocols with decoupled, bounded metadata. Our design funnels all communication through DCs. Thanks to this, and to "reading in the past," SwiftCloud can use metadata that decouples two aspects [START_REF] Ladin | Lazy replication: Exploiting the semantics of distributed services[END_REF]: it tracks causality to enforce consistency, using small vectors assigned in the background by DCs, and uniquely identifies updates to protect from duplicates, using client-assigned scalar timestamps. This ensures that the metadata remains small and bounded. Furthermore, a DC can prune its log independently of clients, replacing it with a summary of delivered updates.

We implement SwiftCloud and demonstrate experimentally that our design reaches its objective, at a modest staleness cost. We evaluate SwiftCloud in Amazon EC2, against a port of WaltSocial [START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF] and against YCSB [START_REF] Cooper | Benchmarking cloud serving systems with YCSB[END_REF]. When data is cached, response time is two orders of magnitude lower than for server-based protocols with similar availability guarantees. With three DCs (servers), the system can accommodate thousands of client replicas. Metadata size does not depend on the number of clients, the number of failures, or the size of the database, and increases only slightly with the number of DCs: on average, 15 bytes of metadata per update, with 3 DCs, compared to kilobytes for previous algorithms with similar safety guarantees. Throughput is comparable to server-side replication for low locality workloads, and improved for high locality CHAPTER 1. INTRODUCTION ones. When a DC fails, its clients switch to a new DC in under 1000 ms, and remain consistent.

Under evaluated configurations, 2-stability causes fewer than 1% stale reads.

Organization

The thesis is divided into three parts. The first part contains this introduction and Chapter 2, which introduces the common background of our work: RDT model with examples.

The second part focuses on the problem of RDT metadata complexity. In Chapter 3 we formulate the problem, we evaluate existing implementations w.r.t. a common metadata metric, we explore opportunities for improvement, and we propose improved implementations of add-wins set and multi-value register RDTs. In Chapter 4, we formally demonstrate a lower bound for metadata overhead of six data types, thereby proving optimality of four implementations. We discuss the scope of our results and compare them to related work in Chapter 5.

The third part of the thesis presents the design of SwiftCloud object database. We overview and formulate the client-side replication problem in Chapter 6. In Chapter 7, we present the SwiftCloud approach, and demonstrate an implementation of this approach using small and safe metadata. Chapter 8 presents our experimental evaluation. We discuss related work in Chapter 9, where we categorize existing approaches and compare them to SwiftCloud.

Chapter 10 concludes the thesis.

Authorship and Published Results

Part of the presented material appeared in earlier publications with our co-authors.

Although we were involved in the formulation of the RDT model [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF][START_REF] Shapiro | Convergent and commutative replicated data types[END_REF][START_REF] Shapiro | Conflict-free replicated data types[END_REF][START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF], presented in Chapter 2, it is not our main contribution, and it is not the focus of this thesis.

The semantics and optimizations of set RDTs were co-authored with Annette Bieniusa, Nuno Preguiça, Marc Shapiro, Carlos Baquero, Sérgio Duarte, and Valter Balegas, and published as a brief announcement at DISC'12 [START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF] and a technical report [START_REF] Bieniusa | An optimized conflict-free replicated set[END_REF]. Some of the lower bounds and large part of the theory behind it are a result of an internship with Sebastian Burckhardt at Microsoft Research in Redmond. Both the optimized register RDT and four lower bounds were published at POPL'14 [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF], co-authored with Sebastian Burckhardt, Alexey Gotsman and Hongseok Yang.

SwiftCloud was designed with Nuno Preguiça, Annette Bieniusa, Sérgio Duarte, Valter Balegas, and Marc Shapiro. The first four contributed to its implementation, which drew from an earlier prototype developed by Valter Balegas. This work is under submission, and partially covered by an earlier technical report [START_REF] Zawirski | SwiftCloud: Fault-tolerant geo-replication integrated all the way to the client machine[END_REF]. Some of the ideas related to the staleness vs. metadata trade-off exploited in SwiftCloud were seeded in a short paper at EuroSys'12 HotCDP workshop [START_REF] Saeida Ardekani | The space complexity of transactional interactive reads[END_REF], co-authored with Masoud Saeida Ardekani, Pierre Sutra and Marc Shapiro.

Chapter 2

Replicated Data Types

Any problem in computer science can be solved with another level of indirection.

David Wheeler CHAPTER 2. REPLICATED DATA TYPES

This chapter presents the background on the concept of Replicated Data Type (RDT), previously formalized in similar ways by Baquero and Moura [START_REF] Baquero | Using structural characteristics for autonomous operation[END_REF], Roh et al. [START_REF] Roh | Replicated Abstract Data Types: Building blocks for collaborative applications[END_REF], and Shapiro et al. [START_REF] Shapiro | Conflict-free replicated data types[END_REF]. 1 An RDT is a type of replicated object accessed via highly available methods, i.e., methods that provide immediate response, with well-defined concurrent semantics [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF][START_REF] Shapiro | Conflict-free replicated data types[END_REF]. RDT objects encapsulate the complexity of convergent replication over distributed, unreliable infrastructure behind a relatively simple interface. RDTs include basic types such as counters, sets, and registers.

In Section 2.1, we recall the motivation for RDTs. We outline what problems they address, and what is specific to the RDT approach. In particular, we highlight what makes RDTs a popular abstraction for managing highly available replicated data [START_REF]Data Types in Riak[END_REF][START_REF] Bonér | Server-managed CRDTs based on Akka[END_REF][START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF].

In Section 2.2, we introduce a formal model of a data type and its implementation. We illustrate the model with examples from different categories of implementations, in order to demonstrate the spectrum of RDT implementations, semantics, and some of the design challenges. Not all RDT implementations provide useful behavior. We examine and specify what can characterize the intended behavior in Section 2.3: from a primary convergence requirement, to a precise specification of concurrent semantics of a data type. In Section 2.4, we define an execution model that describes object implementation and allows us to formally relate an implementation with a specification by a correctness condition. The correctness definition is parameterized, which leads us to formal categorization of implementations.

The material in this chapter is a prerequisite to Part II of the thesis, and is recommended reading before Part III. We highlight the relevant problems as they appear, in particular the challenges related to metadata.

The formalism and presentation used throughout this chapter follows closely a more general theory of Burckhardt et al. [START_REF] Burckhardt | Understanding eventual consistency[END_REF][START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF], with some simplifications in both notation and model.

Motivation

High availability and responsiveness are important requirements for many interactive web applications. These applications often share mutable data between users in geographically distributed locations, i.e., data can be accessed and modified by users in different locations. To address the requirements, applications often rely on geo-replicated databases that replicate data across data centers spread across the world, or even on client-side data replicas at user devices [START_REF] Cairns | Build collaborative apps with Google Drive Realtime API. Google Apps Developers Blog[END_REF][START_REF] Corbett | Spanner: Google's globally-distributed database[END_REF]66,[START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF]. Replication systems provide users with fast access to local replica. This local access is unaffected by high network latency between replicas or by failures [START_REF] Corbett | Spanner: Google's globally-distributed database[END_REF][START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF]66,[START_REF] Shapiro | Conflict-free replicated data types[END_REF][START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF]. 2 1 We use the term Replicated Data Types after a common model of Burckhardt et al. [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF]. The same or specialized concept appears in the literature under a variety of names: Conflict-free/Convergent/Commutative Replicated Data Types (CRDT) [START_REF] Shapiro | Conflict-free replicated data types[END_REF][START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF], Replicated Abstract Data Types (RADT) [START_REF] Roh | Replicated Abstract Data Types: Building blocks for collaborative applications[END_REF], or Cloud Types [START_REF] Burckhardt | Cloud types for eventual consistency[END_REF]. We highlight the differences between these models where relevant. 2 Inter-continental network round-trip times reach hundreds of milliseconds, which exceeds the threshold of perception by human users and negatively impacts their experience [START_REF] Jay | Modeling the effects of delayed haptic and visual feedback in a collaborative virtual environment[END_REF][START_REF] Schurman | Performance related changes and their user impact[END_REF].

MOTIVATION

Failures are prevalent in large-scale systems, and range from individual server failures, variety of data center failures, to network failures, including partitions [START_REF] Bailis | The network is reliable: An informal survey of real-world communications failures[END_REF][START_REF] Kansal | Using dark fiber to displace diesel generators[END_REF]. When client-side replication is involved, extended replica disconnections are granted [START_REF] Petersen | Flexible update propagation for weakly consistent replication[END_REF].

When high performance and availability of update operations are important to the application, they need to be performed at a local replica, without coordination with remote replicas [START_REF] Bailis | Coordination avoidance in database systems[END_REF]. This entails that updates must be accepted concurrently at different replicas and that they must be replicated asynchronously, which causes replicas to diverge and to execute updates in different orders. Unfortunately, this is inherently incompatible with strong consistency models, such as linearizability [START_REF] Herlihy | Linearizability: a correcteness condition for concurrent objects[END_REF] or serializability [START_REF] Bernstein | Concurrency Control and Recovery in Database Systems[END_REF], which rely on a global total order of updates (synchronous replication). This incompatibility, known as the CAP theorem [START_REF] Gilbert | Brewer's conjecture and the feasibility of consistent, available, partition-tolerant web services[END_REF], forces asynchronous replication to offer weaker consistency models only; namely, variants of eventual consistency [START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF][START_REF] Petersen | Flexible update propagation for weakly consistent replication[END_REF].

Under eventual consistency, replicas are allowed to transiently diverge under the expectation of eventual convergence towards a common value that incorporates all updates.

Let us examine more closely the differences between strongly and eventually consistent data replication. We discuss some challenges of implementing and using the latter that motivate our interest in the RDT abstraction.

Any replication protocol needs to address conflicts, which we discuss next. Consider some replicated data item. For example, assume that the item represents a set of elements, such as the members of a group in a social network application, with the initial value {a, b}. In a replicated system, two clients connecting to different replicas might concurrently overwrite it with values {a} and {a, b, c}. Concurrent writes to the same item are conflicting. There are four main approaches to address conflicts. One is to accept only some updates to prevent conflicts (strong consistency).

The other three optimistically accept all concurrent updates and resolve conflicts eventually (eventual consistency). We review these approaches and illustrate them with the help of our {a, b} example in Figure 2.1.

A. Conflict avoidance. Before accepting an update, the database executes a synchronous protocol to agree on a common order of updates execution on all replicas. Such a protocol consistently rejects all but one of the conflicting updates. This forces other writers to abort and retry rejected operations [START_REF] Corbett | Spanner: Google's globally-distributed database[END_REF]. As this solution is strongly consistent, it is easy to program against, because replication is transparent to the application. However, it is not highly available [START_REF] Gilbert | Brewer's conjecture and the feasibility of consistent, available, partition-tolerant web services[END_REF], so it is out of the scope of this work.

B. Conflict arbitration using timestamps. The database accepts all updates and replicates them, but of two concurrent updates, one dominates the other, the "last" one according to timestamps of write operations. This heuristic is known as last-writer-wins (LWW) [START_REF] Johnson | The maintenance of duplicate databases[END_REF]. Arbitrary updates may be lost, i.e., overwritten without being observed, even if they were reported as accepted to the client [START_REF] Kingsbury | Call me maybe: Cassandra[END_REF]. In our example in Figure 2.1B, assuming that timestamp of {a} is larger than that of {a, b, c}, the outcome of write({a, b, c}) is lost. This is not a satisfying solution.

C. Conflict detection and application-level resolution.

To avoid lost updates, the database detects concurrent writes after they were accepted, and presents the set of concurrentlywritten values to the application. It is up to the application to resolve the conflict [START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF][START_REF] Kistler | Disconnected operation in the Coda file system[END_REF][START_REF] Popek | Replication in Ficus distributed file systems[END_REF][START_REF] Terry | Managing update conflicts in Bayou, a weakly connected replicated storage system[END_REF]. Figure 2.1C illustrates this approach. Unfortunately, resolving conflicts is difficult ad-hoc at the application layer, and may not converge. For example, it is not obvious how to combine values {a} and {a, b, c} from our example without violating intentions of the clients that wrote them. It may require the application to add more metadata to the values it stores (e.g., to record the fact that b was removed and c was added), which may be costly and complex, especially in the presence of replica failures [START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF]. As a last resort, the application may present the conflict to the user and let the user decide or confirm its choice. In either case, however, the conflict resolution may need to be run concurrently with other updates and across different application processes or users, which may generate even more conflicts, and prevent or delay database convergence [START_REF] Baquero | Scaling up reconciliation in eventual consistency[END_REF]. 3 For example, one user may resolve the conflict to {a, c}, while another resolves it to {a, b, c}, creating a new conflict. This approach can quickly become unstable and complex at the application layer. D. Automatic database-level conflict resolution. As an alternative, the database itself may automatically resolve conflicting updates. This is different from the previous approach, because it assumes the database has enough knowledge to embed convergent conflict resolution in the replication protocol. For example, as illustrated in Figure 2.1D, if the database is aware that the read of {a, b} followed by the write of {a} is meant to remove element b from a set, and the other pair is meant to add c, it can integrate both updates. This can be also more efficient than the application-level solution, since the metadata can be managed by the database, i.e., lower in the abstraction stack. The challenge is then to design a database protocol that will tolerate concurrency and failures, still converging to a sensible value that integrates all updates. For some updates the right integration is simple (e.g., add and remove of different elements), for others it is harder (e.g., operations on the same element).

A Replicated Data Type (RDT) abstraction can express all three optimistic conflict resolution approaches (B)-(D), and in particular, addresses the challenges of automated database-level conflict resolution [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF][START_REF] Shapiro | Conflict-free replicated data types[END_REF]. The insight is to use typing to indicate semantics of a data item (naming the role of object) and to mutate it via high-level object interface (naming the role of operations), rather than low-level read-write operations. The database implementation, and in particular a data type implementation class, can leverage the data type semantics to automate and to optimize conflict resolution. The data type semantics can also serve as a contract describing the interface to an application developer, thus separating concerns and responsibilities.

With RDTs, an application organizes its shared state as a collection of typed objects. For instance counter objects have increment, decrement and read methods, whereas set objects have add(a), remove(a) and read methods. It is an object implementation duty to handle the burden of replication, and to resolve concurrent updates in a sensible manner, thereby releasing application programmers from this duty. Moreover, some categories of RDT implementations also address fault-tolerance of convergent replication, and tolerate messages reordering, message loss, disconnected replicas etc. An RDT operates at the limited scope of an object. It does not support conflict-resolution involving complete database [START_REF] Bjørner | Models and software model checking of a distributed file replication system[END_REF][START_REF] Petersen | Flexible update propagation for weakly consistent replication[END_REF], and does not address the other problems of eventual consistency at a larger scale, across different objects. Specifically, it does not address the somewhat orthogonal problem of update ordering, nor of atomicity guarantees. Examples of such desirable mechanisms might include cross-object causal consistency and its transactional variants [START_REF] Ahamad | Causal memory[END_REF][START_REF] Belaramani | PRACTI replication[END_REF]66]. We do not consider such ordering and atomicity guarantees for simplicity here, although the model we adopt can be extended to express them [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF]. We return to intra-object ordering in Section 2.4.3, and to inter-object ordering in Part III.

System Model and Object Implementation

In this section, we formalize an implementation model of Replicated Data Type. We use the model of Burckhardt et al. [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF] that can express existing implementations [START_REF] Baquero | Using structural characteristics for autonomous operation[END_REF][START_REF] Burckhardt | Cloud types for eventual consistency[END_REF][START_REF] Roh | Replicated Abstract Data Types: Building blocks for collaborative applications[END_REF][START_REF] Shapiro | Conflict-free replicated data types[END_REF].

Figure 2.2 gives an informal overview of the system components involved in our model. We consider a replicated database that hosts one or more named objects on a set of replicas. We assume that the database implements each object independently. In other words, a database consists of a single object, without loss of generality. 4 The type signature of an object defines its interface. An object implementation defines its behavior, including its replication protocol.

An application has multiple processes that run client sessions. Each session issues a sequence of operations to a replica by calling methods; for simplicity, we assume that every session matches exactly one logical replica. A database replica delegates method calls to the corresponding object implementation. It responds by performing a local computation step. Local processing ensures that all operations are highly available and responsive. Replicas communi-cate in the background via a message passing network layer to exchange updates. An object implementation includes functions that produce and apply messages.

Object Implementation Model

The type ø of an object comprises a type signature tuple (Op An object is instantiated from an implementation class of some type, to be defined shortly.

An object is hosted on a set of database replicas ReplicaID (ranged over by r). We assume that the database can provide the object implementation with unique timestamps from a domain Timestamp (ranged over t), totally ordered by <. Scalar clocks, such as Lamport's logical clock [START_REF] Lamport | Time, clocks, and the ordering of events in a distributed system[END_REF],

or real-time clock concatenated with a unique replica ID, can serve this purpose. For simplicity, we often assume integer timestamps.

Definition 2.1 (Replicated Data Type Implementation

). An implementation of replicated data type ø is a tuple D ø = (ß, M, initialize,do,send,deliver), where:

• Set ß (ranged over by ae) is the domain of replica states.

• Set M (ranged over by m) is the domain of messages exchanged between replicas.

• Function initialize : ReplicaID ! ß defines the initial state at every replica.

• Function do : ß £ Op ø £ Timestamp ! ß £ Val ø defines methods.

• Functions send : ß ! ß £ M and deliver : ß £ M ! ß define update replication primitives.

We call D ø an implementation class of type ø. We refer to the components of a tuple D ø using dot notation, e.g., D ø .ß for its states.

We now discuss the role of implementation components and the intuition behind an execution model, to be formalized later. The database instantiates the object at every replica r 2 ReplicaID by initialize(r). From that point on, it manages the object at each replica using the three implementation functions explained next. This produces a tuple (ae 0 , m) = send(ae), where ae 0 is the new state of the object (send may have a side-effect), and m is a message that the replica sends, or broadcasts, to all other replicas. 5 When a replica receives a message m, it can apply deliver(ae, m) on the object in state ae, which produces the new state ae 0 that incorporates message m.

Replication

Data type implementations can be categorized by how they replicate updates, and what requirements they impose on the network layer. In our prior work, Shapiro et al. [START_REF] Shapiro | Conflict-free replicated data types[END_REF] propose two main implementation categories: operation-based (op-based) and state-based. In opbased implementation, each message carries information about the latest updates that have been performed at the sender replica. Such message usually must not be lost, and sometimes requires further ordering or no-duplication guarantees. In contrast, in state-based implementation, each message carries information about all updates that are known to the sender; such messages can be duplicated, lost or reordered. We will formalize these differences after we familiarize the reader with their intuition. Both categories have the same expressive power, in the sense that every state-based object can be also expressed as an op-based object and vice versa [START_REF] Shapiro | Conflict-free replicated data types[END_REF].

One of the main challenges are the two expectations behind an object implementation:

that the replicas converge towards a state that integrates all updates, and that both final and intermediate results provide meaningful semantics. This is complicated, because the convergence process driven by asynchronous replication is inherently concurrent itself, and with continuous stream of new updates. Moreover, the network layer may drop, reorder or duplicate messages. To achieve convergence and reasonable semantics, an implementation must often encode additional metadata in its state when it performs updates (with do function), so that the replication protocol (send and deliver functions) have enough information to interpret (reconcile) concurrent operations, or to handle network failures.

We formulated sufficient convergence conditions specific to each of the two main implementation categories in a prior work [START_REF] Shapiro | Conflict-free replicated data types[END_REF]. For an op-based implementation to be convergent, it suffices that all concurrently generated messages commute, i.e., they are insensitive to the order of Algorithm 2.1 Template for data type implementation. ae √ ≠ definition of the state integrating message m AE delivery. For a state-based implementation to be convergent, it suffices that the domain of states ß form a (partially ordered) join semilattice, such that the deliver function is a join operator for that lattice, and update methods can only cause the state to advance in the partial order of the lattice. This typically requires that the object state represents some summary of its history, which allows any two states to be integrated. We will illustrate these conditions alongside the following example implementations, and return formally to the problem when we define a category-agnostic convergence and semantics specification.

Examples

We illustrate the model with simple example implementations from both op-based and state-based categories. They help us to highlight some of the design alternatives and challenges, as well as primitive mechanisms used by implementations, such as clocks. More complex examples will appear later. The selected examples are a small part of our catalog of implementations [START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF].

Template. Algorithm 2.1 serves as a template for presentation of all data type implementations. a √ a + d . add to the local value 2) with three replicas. Every event is labelled with the execution step, the return value of the function performing that step (above line; except for unlabelled initialize events), and replica state after the step if it changed (below line). We indicate messages exchanged between replicas with arrows, thus we omit repeating their content in deliver. Timestamps in do are specified only if used.

r 1 inc (1, 1) (0, 0) send:1 (1, 0) r 2 (0, 0) r 3 (0, 0) inc (1, 1) deliver (1, 0) deliver (2, 1) deliver (2, 0) send:1 (2, 0) deliver (2, 0) read:1 read:2

Counter Implementations

Op-based. Algorithm 2.2 presents an op-based implementation of a counter type (Ctr). The state of a replica contains the current counter value, noted a, and an integer buffer that records the number of local increments since the last message was sent, noted b. Each increment operation, inc, simply increments both the current value and the buffer. When a replica sends a message, the implementation flushes the buffer, i.e., the local buffer becomes the content of the message and the buffer is zeroed. A replica that delivers a message increments its local counter by the value provided in the message without modifying its local buffer.

Figure 2.3 illustrates both an example execution of the op-based counter, and our presentation format of executions. After all replicas are initialized with the initialize function, clients at replica r 1 and r 3 perform one inc each, which is implemented by the database replica applying the do function. The read operation that follows at replica r 1 yields value 1, which includes the outcome of the earlier local increment. The message with update from r 1 is generated using send, and reaches r 2 and r 3 before r 3 flushes his buffer. The message is incorporated using deliver, which vv √ ∏s. max{vv(s), vv m (s)} . compute entry-wise maximum of vectors adds its content to the counter value a, but does not touch his buffer b. Eventually, r 3 sends a message, flushing his buffer b, which reaches all remaining replicas. Replicas converge towards value a = 2. At the end, a read at replica r 1 yields return value 2.

r 1 inc (r 1 , [1 0 0]) (r 1 , [0 0 0]) r 2 (r 2 , [0 0 0]) r 3 (r 3 , [0 0 0]) inc (r 3 , [0 0 1]) read:1 read:2 send:[1 0 0] deliver (r 2 , [1 0 0]) deliver (r 3 , [1 0 1]) deliver (r 1 , [1 0 1]) deliver (r 2 , [1 0 1]) send:[1 0 0] deliver (r 3 , [1 0 1]) send:[0 0 1] read:2
Intuitively, the op-based counter converge if every replica eventually applies the send function on every non-zero buffer and the network layer delivers every message to every replica exactly once [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF]; we will formalize these conditions later. The state converges, because the addition operation employed in deliver is commutative, i.e., different increments can be delivered at replicas in different order and produce the same effect. Furthermore, the value that the counter replicas converge to is the total number of increments inc issued at all replicas [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF].

State-based.

Let us now consider a state-based implementation of counter, illustrated in Algorithm 2.3. Recall that in state-based implementation a message can be lost, reordered or duplicated, and should contain the complete set of operations known by the sender. A scalar integer is insufficient in this case. To see why, consider two replicas that concurrently increment their counter; if the current value of a counter was a scalar, the receiver's replica would have no means of distinguishing whether the received increment is already (partially) known locally or if it is a fresh one. We will demonstrate this impossibility formally, in Part II of the thesis.

Instead, every replica of a state-based counter stores information about all increments CHAPTER 2. REPLICATED DATA TYPES it knows about. These can be represented efficiently as a vector, formally a map from replica identity to the number of increments, noted vv : ReplicaID ! N 0 , such that each replica increments its own entry only. A value of a counter is simply the sum of all entries in the vector vv. The inc operation increments the replica's own entry in the vector (the identity of this entry is recorded as r during the initialization). To replicate, the state-based counter sends its complete vector. The receiving replica computes entry-wise maximum with its own vector. Baquero and Moura [START_REF] Baquero | Using structural characteristics for autonomous operation[END_REF], and our later work [START_REF] Shapiro | Conflict-free replicated data types[END_REF], show that the maximum operator on a vector used by deliver, together with the way the vector is incremented, guarantee convergence of the counter state towards a maximum vector. Moreover, the maximum operator is resilient to message duplication, loss, and out-of-order delivery. This is thanks to the fact that the domain of states (vectors) is a semilattice under the partial order defined as:

[0 0 0] [1 0 0] [0 1 0] [0 0 1] [2 0 0] [0 2 0] [ 0 0 2] [1 1 0] [1 1 1] [0 1 1] ... ... ... [1 0 1]
vv v vv 0 () dom(vv) µ dom(vv 0 ) ^8r 2 dom(vv 0 ).vv(r) ∑ vv 0 (r).
(2.1) (We note by < a strict variant of this relation.)

Each inc update advances the vector in the v order, whereas the entry-wise maximum operator used by deliver is in fact the least upper bound of the semilattice:

(2.2) vv t vv 0 = ∏r. max{vv(r), vv 0 (r)}, i.e., it generates the minimum vector that dominates both input vectors. We illustrate a fragment of the lattice on Figure 2.5. Throughout an execution of the implementation, replica states advance (a, t) √ (a m , t m ) . overwrite the existing value if the existing timestamp is dominated in this lattice (indicated with dotted paths on the figure), adding each time more information about the operations performed. Replicas traverse the lattice in parallel, while merging states with least upper bound ensures that they converge.

Compared to the op-based implementation, the state of a replica has no designated outgoing buffer, but the whole state is transferred instead. Therefore, state-based implementation transitively delivers updates, i.e., a replica may serve as an active relay point (as r 2 does for r 3 , in our example). The implementation, however, is more complex.

It is possible to extend this solution to build a counter with an additional decrement method [START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF], by using a separate vector for decrements. Ensuring strong invariants over a counter, e.g., ensuring that the value remains positive, is more difficult and requires stronger consistency model in the general case.

Register Implementations

State-based LWW register. Another RDT example is an object that resolves concurrent updates by arbitration, using approach (B) from Section 2.1. Algorithm 2.4 presents a state-based register (LWWReg). Without loss of generality, we consider a register that stores integer values.

The register stores both a value, and a unique timestamp generated when the value was written; the latter is not visible to the read operation. Concurrent assignments to the register are resolved using the last-writer-wins (LWW) policy applied at all replicas: the write with a higher timestamp wins and remains visible [START_REF] Johnson | The maintenance of duplicate databases[END_REF]. LWW guarantees that all replicas eventually select the same value, since the order of timestamp is interpreted in the same way everywhere.

The implementation may use any kind of totally ordered timestamps. However, ideally, every write should be provided with a timestamp higher than that of the current value, so it takes effect. A popular implementation is a logical clock, or Lamport clock [START_REF] Lamport | Time, clocks, and the ordering of events in a distributed system[END_REF].

A √ {(a, vv) 2 A [ A m |6 9(a 0 , vv 0 ) 2 A [ A m : vv < vv 0 }
. keep non-overwritten entries is a pair (k, r) 2 N £ ReplicaID, where k is a natural number, and r is a replica ID. The order of timestamps is defined on natural numbers first, and replica IDs serve as tie-breakers, i.e., (k, r) < (k 0 , r 0 ) () k < k 0 _ (k = k 0 ^r < r 0 ). To generate a timestamp, replica r i computes the highest number k it has observed in all timestamps so far, noted k max , and assigns the generated timestamp value (k max + 1, r i ). The new timestamp is fresh and dominates the observed ones.

r 1 write(13) (r 1 , {(13, [1 0 0])}) (r 1 , ∅) r 2 (r 2 , ∅) r 3 (r 3 , ∅) write(39) (r 3 , {(39, [0 0 1])}) send:{(13, [1 0 0])} read:13 read:{13,39} write(26) (r 2 , {(26, [1 1 1])}) deliver (r 1 , {(26, [1 1 1])}) deliver (r 2 , {(13, [1 0 0])}) deliver (r 2 , {(13, [1 0 0]), ( 39
, [0 0 1]}) send:{(39, [0 0 1])} deliver (r 3 , {(26, [1 1 1])}) send:{(26, [1 1 1])}
The implementation of LWW is simple, but as we illustrated in Figure 2.1B, LWW implies that arbitrary assignments are entirely lost, despite being previously accepted.

State-based multi-value register.

A popular alternative to LWW register is a multi-value register (MVReg), implementing conflict-detection approach (C), which does not lose updates. The multi-value register stores all non-overwritten values, and lets the application decide how to resolve them. Thus, if there are concurrent updates, it returns a set of values, Val MVReg = P(Z), rather than a single one.

Practical state-based implementations of multi-value register encode information about overwritten values compactly, using a variation of version vectors by Parker et al. [START_REF] Parker | Detection of mutual inconsistency in distributed systems[END_REF]. A version vector has the same structure as the vector we described in the counter implementation: it is a map from replica ID to the number of observed updates. The usage of a version vector is 2.2. SYSTEM MODEL AND OBJECT IMPLEMENTATION more complex, and so is its semantics. In counter, a vector is associated directly with a state and describes the number of increment events observed from each replica. In multi-value register, a vector is associated with a value written in a certain state and represents the number of operations from each replica that value overwrites. Equivalently, it represents the knowledge of the replica that wrote the value, in terms of the number of observed writes from each replica. For instance, a vector vv = {r 1 7 ! 3, r 2 7 ! 5} associated to a value indicates that the value overwrites the first three writes from replica r 1 , and the first five from replica r 2 .

The intended behavior of multi-value register is to overwrite a value only if the writer of the new value has observed (or could have observed) the overwritten value. To this end, the implementation interprets the partial order v that vectors induce (Equation 2.1) as potential causality between operations [START_REF] Lamport | Time, clocks, and the ordering of events in a distributed system[END_REF][START_REF] Schwarz | Detecting causal relationships in distributed computations: In search of the holy grail[END_REF]. If vectors of two values are ordered by vv 1 v vv 2 , then they are causally related: the operation with vv 1 has been visible at the replica that issued the operation with vv 2 , and could have caused or impacted the latter. Absence of causality between operations indicates logical concurrency. Algorithm 2.5 presents a multi-value register implementation based on a standard version vector algorithm [START_REF]Riak distributed database[END_REF][START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF][START_REF] Kistler | Disconnected operation in the Coda file system[END_REF][START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF]. The state of a replica contains local replica ID r, used to assign vectors, and a set of entries A, pairs of values and their vectors. The entries represent the set of non-overwritten values; formally, the entries represent an antichain of the causality relation induced by v. Thus, read simply returns all values of the entries. write operation overwrites the set of observed values. To this end, replica needs to produce a vector that dominates them in causality. The replica applies the least upper bound operator t on all observed vectors (Equation 2.2), and increments the replica's own entry in the vector, to strictly dominate observed vectors, according to <. The generated vector is assigned to the written value and replaces the existing set of entries.

The replication protocol sends all entries. The receiving replica integrates local entries with the received ones. It computes the new antichain, without losing any updates, as a union of local and received entries with eliminated overwritten entries, according to their version vectors. and r 3 perform concurrent writes of values 13 and 39, respectively. Replica r 2 receives their updates, and identifies that these are writes that do not overwrite one another, since their vectors are incomparable according to v. Therefore, the read at r 2 returns {13, 39}. The client at r 2 subsequently overwrites the two values with value 26; the vector it assigns dominates known vectors. Therefore, when it propagates his updates, his entry for value 26 overwrites existing entries at r 1 and r 3 .

Although vectors are a powerful concept that prevents data loss, they occupy space, in the order of the number of replicas. As we shall see in Part II, for some types this cost is unavoidable.

Specification of Intended Behavior

The primary expectation behind any replicated object is convergence. Category-specific convergence conditions can be successfully applied to implementations [START_REF] Shapiro | Conflict-free replicated data types[END_REF], including the examples from the previous section. However, convergence alone is not sufficient. Towards what value the object converges, and what are the intermediate states and their values, is also important. For instance, an increment-only counter that provably converges towards 0, or occasionally returns value 0 before it converges, does not provide the intended behavior.

A precise specification of type semantics can act as a contract between an application and a data type implementation. There can be multiple implementations of the same type, including different optimizations or models (e.g., Algorithm 2.2 vs. Algorithm 2.3). It would be natural to express that these implementations behave in the same way. Implementation is often not an effective way of expressing semantics, since it requires considering low-level details of implementation and execution (consider, for example, Algorithm 2.5, or an execution from Figure 2.4).

In this section, we define and illustrate a declarative RDT specification model, after Burckhardt et al. [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF]. This form of specification offers a unified and concise representation of RDT semantics, that embodies the convergence condition.

Specification Model

To introduce the specification model we rely on the reader's intuitive understanding of the execution model from the previous section. At a high level, our goal is to define a minimal abstract description of each concrete execution that would express the behavior of an implementation, and that would justify its correctness. We give an overview of the components involved in the semantic description model and the definition of correctness condition in Figure 2 operations with additional relations. A data type specification is defined on abstract execution structure, and indirectly identifies a subset of abstract executions that satisfy the specification.

Our ultimate goal is to define a correctness condition for a data type implementation, that relates each concrete execution with a witness abstract execution that satisfies the specification.

The client-observable behavior of an execution can be characterized by a trace that carries information on operations performed by each client session, the order in which they were issued by a client session, and their return values. The intended behavior concerns return values in a trace. However, a trace alone is insufficient to specify the intended behavior, because it is missing information about relation of operations performed in different client sessions (at different replicas). For example, it is impossible to determine if a counter implementation behaves correctly if the operations in the trace do not specify what increments were delivered to a replica of the client that performed the operation. A trace must be extended with additional information.

Sequential specification. In a strongly-consistent setting, objects are, roughly speaking, guaranteed to execute operations in the same sequence at all replicas. Therefore, the intended behavior of type ø can be expressed as a sequential specification function S ø :

Op + ø ! Val ø , which
specifies the return value of the last operation in a provided sequence [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF]. A trace extended with a sequence of operations from the execution can be evaluated against a sequential specification.

For example, the sequential specification of a counter can be defined as [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF]:

S Ctr (ª inc) = ?;
S Ctr (ª read) = a where a is the number of inc operations in ª,

where ª 2 Op +
Ctr is any sequence of operations. Similarly, the semantics of a sequential register is:

S LWWReg (ª write) = ?; S LWWReg (ª read) = a iff the last operation in ª| write is write(a), (2.3)
where ª 2 Op + LWWReg , and ª| write is a restriction of ª to write operations.

A client trace, extended with the sequence order, is correct if it satisfies the above specification.

Intuitively, a sequence with write(1) operation, followed by read that returns 1, is correct w.r.t.

S LWWReg , but a sequence with write(2), followed by a read returning 1 is not.

Concurrent specification.

In the case of RDTs, the order of operations may not be the same across replicas. Thus, a specification have more elaborate structure. Burckhardt et al. [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF] identify two complementary abstract orders that can impact the return value of an RDT operation: and metadata were involved, what path that update followed through replicas until it was delivered, whether there were duplicated messages delivered, lost messages, etc.)

2. An order in which concurrent operations are arbitrated. Arbitration abstracts away the implementation of timestamps.

Formally, an abstract execution describes the client-observable effects of a concrete execution (trace) with additional visibility and arbitration information, as follows.

Definition 2.2 (Abstract Execution

). An abstract execution of a data type ø is a tuple A = (E, repl,op,rval,ro,vis,ar), where:

• E µ Event is a set of events from a countable universe Event.

• Each event e 2 E represents a replica repl(e) 2 ReplicaID performing an operation op(e) 2 Op ø that returned value rval(e) 2 Val ø .

• ro µ E £ E is a replica order, which is a union of strict total prefix-finite orders on events at each replica.

• vis µ E £ E is an acyclic prefix-finite visibility relation.

• ar µ E £ E is a strict total prefix-finite arbitration relation.

We require that ro [ vis is acyclic, as we do not consider speculative systems [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF], and assume that local updates are immediately visible, i.e., ro µ vis. Pairs of events that are not ordered by visibility are said concurrent. Visibility is often a partial order; we assume this in our examples.

SPECIFICATION OF INTENDED BEHAVIOR

We give two examples of abstract executions in Figure 2.8, which characterize concrete executions of counter and multi-value register implementations from the previous section. 6Specification of a data type is defined on each event in an abstract execution and its context, i.e., the set of events visible to that event. Other desirable system properties that we do not discuss here, such as cross-object consistency or session guarantees, can be also formulated as a condition on abstract executions [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF].

Every type specification has an implementation (provided it does not return abstract events E, invisible to the implementation). Intuitively, a naive state-based implementation of a specification stores a complete graph of visibility and arbitration, and implements read as F ø .

Examples

We illustrate the specification model with a few examples from Bieniusa et al. [START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF], Burckhardt et al. [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF], and a new one, covering the implementations from Section 2.2. These are primitive data types that could be composed to build more complex objects [START_REF] Gotsman | Composite replicated data types[END_REF]; we do not study composition in this work.

Counter Specification

The specification of a counter (Ctr) is given by:

F Ctr (read, E, op,vis,ar) = |{e 2 E | op(e) = inc}|; (2.5)
F Ctr (inc, E, op,vis,ar) = ?, i.e., read should return the number of all visible increment operations. For example, an abstract execution from Figure 2.8A satisfies specification F Ctr for every read since the number of visible inc operations matches the return value. Specification F Ctr characterizes both Algorithm 2.2 and Algorithm 2.3 [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF], as we formalize later. However, although it is a reasonable and popular counter semantics [START_REF]Data Types in Riak[END_REF][START_REF] Sérgio | Scalable eventually consistent counters over unreliable networks[END_REF][START_REF] Shapiro | Conflict-free replicated data types[END_REF], this is not the only possible one. For example, one could imagine an alternative, where multiple concurrent increments count as one.

Counter specification does not make use of the arbitration relation. Indeed, none of its implementations (Algorithm 2.2 and Algorithm 2.3) uses timestamps.

Register Specifications

Last-writer-wins register. The specification of LWW register (LWWReg), characterizing Algorithm 2.4 [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF], is defined by:

F LWWReg (o, E, op,vis,ar) = S LWWReg (E ar o), (2.6)
where E ar is a sequence of events in E ordered by arbitration. Thus, the specification is identical to the sequential specification from Equation 2.3, applied to a context of operation linearly ordered by arbitration. It does not make use of the visibility relation between events in a context.

Relying on sequential specification is appealing in its simplicity (it could also express the counter specification F Ctr ), but in the general case this specification pattern cannot express all semantics alternatives that do not lose arbitrary updates, as the following examples demonstrate.

Multi-value register. The specification of a multi-value register (MVReg), characterizing Algorithm 2.5 [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF], is given by: (Hereafter, we omit specification of methods that do not return any value.)

F MVReg (read, E,
Here, visibility acts as a causality relation. The read operation returns values of all the writes that are not dominated in visibility by writes of other values. 7Specification of data types that use visibility structure, like multi-value register, is difficult to interpret if the visibility is not a partial (transitive) order [START_REF] Mukund | Optimized OR-sets without ordering constraints[END_REF]. Therefore, database implementations often enforce transitivity of visibility for these objects, as we will discuss shortly.

Value-wins. In between LWW and multi-value registers, we propose a new value-wins register (VWReg), which resolves concurrent assignments using a predefined order on values. Assuming a totally ordered domain of values, such as integers, the signature of object is the same as the LWW register, (Op VWReg , Val VWReg ) = (Op LWWReg , Val LWWReg ). We define the specification as follows:

F VWReg (read, E, op,vis,ar) = max{a |9e 2 E : op(e) = write(a) (2.8) ^6 9e 0 2 E : e 0 = write(b) ^e vis °°! e 0 }.
Value-wins register can express data-driven conflict resolution. A specific case of a value-wins register is a false-wins flag and a true-wins flag [START_REF]Data Types in Riak[END_REF]. We shall see that some set types rely on a similar approach.

If the order on the domain of values is partial, a similar extension to the multi-value register is feasible. This could reduce the number of conflicts to resolve by the application (user).

Set Specifications

The set is an example of a primitive data type for which a concurrent specification is nontrivial.

Many alternatives are possible. As we shall see in Part II, the implementations are also complex.

Without loss of generality, we consider a set of integers, where element can be added or In our prior work, Bieniusa et al. [START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF] recommend that any concurrent semantics should respect the sequential behavior as much as possible, in the sense that, for executions where the sequential specification applied to all linear extensions of visibility produce the same return value, that same value should be returned in the concurrent case. Some operations of a set are independent, such as operations on different elements, or idempotent, such as adding (removing) the same element twice, or removing it twice. For these independent or idempotent operations, a concurrent semantics could reduce to the sequential one. Concurrent add and remove on the same element are problematic. Indeed, they have no obvious sequential counterpart -according to sequential specification, the outcome depends on the order. The following three set specifications are popular convergent heuristics that treat this case in different ways, and otherwise rely on the common design principle of Bieniusa et al. [START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF].

The LWW set is specified as follows [START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF][START_REF] Johnson | The maintenance of duplicate databases[END_REF]:

F LWWSet (read, E, op,vis,ar) = S Set (E ar read), (2.9)
where LWWSet is a type with the same signature as Set, S Set is a standard sequential specification of set, and E ar is defined as before. Concurrent operations on the same element are resolved according to timestamp arbitration. The weakness of this approach is that it loses some arbitrary concurrent updates on a same element, similarly to LWW register.

An alternative is to give priority to one of the operation types. A popular choice is the add-wins set semantics (also known as observed-remove) defined as [START_REF]Data Types in Riak[END_REF][START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF][START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF][START_REF] Shapiro | Conflict-free replicated data types[END_REF]:

(2.10) F AWSet (read, E, op,vis,ar) = {a | 9e 2 E : op(e) = add(a)^6 9e 0 2 E : op(e 0 ) = rem(a)

^e vis °°! e 0 },
where AWSet has the same signature as Set. In add-wins, whenever there are concurrent operations on the same element a, add(a) operation wins and "cancels" the effects of every visible or concurrent rem(a) operation. Note that add-wins set specification could be also viewed as a composition of presence flags (VWReg), one for each element, where a positive value wins.

The opposite approach is known as remove-wins set [START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF]:

(2.11) Element a is in the remove-wins set if all rem(a) operations are covered in visibility by some visible add(a) operation.

F
Note that the set of covering add operations does not need to be the same for all removes.

Otherwise, the set would have an anomaly that violates the principle of Bieniusa et al. [START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF]:

consider two clients that concurrently perform operations rem(a);add(a) for the same element a;

although initially both clients observe element a in the set, after they synchronize, a disappears.

Both F AWSet and F RWSet definitions make nontrivial use of the visibility relation, since their policy is applied only to concurrent operations. In the case of operations ordered by visibility, the sequential semantics applies.

An example operation context in Figure 2.9 illustrates the three discussed set specifications applied to read operation. The LWW set returns {13}, because add(13) Other alternatives include the counting set of Aslan et al. [START_REF] Aslan | C-Set: a Commutative Replicated Data Type for Semantic Stores[END_REF] and of Sovran et al. [START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF].

Here, an element is present if the difference between the number of visible add and remove operations is positive. This approach has anomalous semantics, as it violates the principle of Bieniusa et al. [START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF], similarly to the example that motivated remove-wins definition. Finally, an alternative is to set a conflict flag in case of concurrent add(a) and rem(a) [START_REF] Bieniusa | Brief announcement: Semantics of eventually consistent replicated sets[END_REF] (cf. approach (C) from Section 2.1).

F AWSet (L, read) = {13, 26} F RWSet (L, read) = ; F LWWSet (L, read) = {13}

Execution Model and Correctness

In this section, we formalize a concrete execution model (Section 2.4.1), and relate concrete executions and abstract executions with an implementation correctness definition (Section 2.4.2), after Burckhardt et al. [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF]. The correctness condition is parameterized with a pair of conditions, which leads us to a formal definition of some implementation categories (Section 2.4.3). The reader interested only in Part III of the thesis may proceed directly to Section 2.4.3.

Execution Model

We model a concrete execution of an implementation D ø as a sequence of transitions over configurations, according to the definitions of Figure 2.10. A configuration is a pair (R, N) 2

Config, which describes the state of a database: R maps each replica to an object state, and N maps all messages that were sent over network and can be delivered, identified by their message ID, mid 2 MessageID. The initial empty configuration (R 0 , N 0 ) starts with the initial state of an object and no sent messages. A transition relation °!D ø : Config £ Event £ Config describes an atomic execution step. An execution can be visualized as a finite or infinite sequence of transitions:

(R 0 , N 0 ) e 1 °!D ø (R 1 , N 1 ) e 2 °!D ø ... e k °!D ø (R k , N k ) ...,
where every transition

e i
°!D ø is explicitly labelled by a unique event e i and implicitly parameterized by functions that record information about event e i , such as action(e i ), replica(e i ), etc.

There are three types of configuration transitions, corresponding to the three rules of Fig- Configurations:

RState = ReplicaID ! D ø .ß R 0 = D ø .initialize NState = MessageID * D ø .M N 0 = [] Config = RState £ NState (R 0 , N 0 ) 2 Config
Transitions over configurations: some fresh identifier mid. In the third rule, replica r delivers message N(mid) from the map of sent messages N, which mutates its state R(r) according to the implementation function deliver.

D ø .do(o, ae, t) = (ae 0 , v) action(e) = do replica(e) = r op(e) = o timestamp(e) = t rval(e) = v (R[r 7 ! ae], N) e °!D ø (R[r 7 ! ae 0 ], N) D ø .send(ae) = (ae 0 , m) action(e) = send replica(e) = r mid › dom(N) msg(e) = mid (R[r 7 ! ae], N) e °!D ø (R[r 7 ! ae 0 ], N[mid 7 ! m]) D ø .deliver(ae, m) = ae 0 action(e) = deliver replica(e) = r msg(e) = mid (R[r 7 ! ae], N[mid 7 ! m]) e °!D ø (R[r 7 ! ae 0 ], N[mid 7 ! m])
Note that the transitions put no restriction whatsoever on when the first two rules can be applied; the third rule requires only that every delivered message must have been previously sent. 8 Otherwise, there is no restriction on message delivery: messages can be lost, duplicated, and reordered. We introduce these optional restrictions separately in Section 2.4.2.

We now define concrete execution, similarly to the abstract one.

Definition 2.6 (Concrete Execution).

A concrete execution of a database with an implementation D ø of a type ø is a tuple C = (E, eo, pre, post, action, replica, op, timestamp, rval, msg), where:

• E µ Event is a countable subset of events.

• eo µ E £ E is a strict total prefix-finite execution order of transition events.

• pre,post : E ! Config define, resp., predecessor and successor configurations, s.t.:

8e 2 E : pre(e) e °!D ø post(e), and 8e, f 2 E : e eo °! f ^(6 9e eo °! g eo °! f ) =) post(e) = pre( f ).
• The remaining are partial labelling functions: action tells a type of event; timestamp, op and rval are defined for do events, where timestamp is injective to guarantee uniqueness; msg : E ! MessageID is defined for send and deliver events.

We denote the initial configuration of C by init(C) = C.pre(e 0 ); if C is finite, we note the final configuration by final(C) = C.post(e k ), where e k is the last event.

Implementation Correctness

The semantics of an implementation can be characterized by all of its executions.

Definition 2.7 (Implementation Semantics). The semantics of an implementation D ø , noted ÇD ø É, is the set of all its concrete executions that start in the empty configuration (R 0 , M 0 ). Some implementations (or categories of implementations) may not provide a meaningful behavior under all executions, but only when certain network layer conditions are met. For example, the op-based counter from Algorithm 2.2 requires operations to be delivered exactly once, rather than an arbitrary number of times. Similarly, the LWW register may require a condition on the order of supplied timestamps. We express such restrictions as a network layer specification, noted T, a set of allowed concrete executions defined by a condition on concrete executions. 9 Therefore, when considering the correctness of an implementation, we will reason about ÇD ø É \ T, i.e., the semantics of an implementation D ø under network specification T.

In order to state a correctness condition, we would like to relate each concrete execution with a correct witness abstract execution, i.e., to find a correct representation of a concrete execution in the specification domain. Both types of executions, given by Definitions 2.2 and 2.6, share a similar structure, and most components of a witness execution can be directly extracted from the concrete execution. Thus, we define a witness execution for concrete execution C 2 ÇD ø É \ T as:

(2.12)

abs(C,V) = (C.E| do , E.replica| do , E.op| do , E.rval| do , ro(C)| do , V(C), ar(C))
, where components are restricted to client-observable do events of C, and replica order ro(C) and arbitration order ar(C) are extracted from C, using information about replicas and timestamps:

e ro(C) °°°! f () e C.eo °°°! f ^C.replica(e) = C.replica( f ) e ar(C) °°°! f () e, f 2 C.E| do ^C.timestamp(e) < C.timestamp( f )
The remaining witness component V(C) offers some freedom in the witness selection: visibility order. A concrete execution specifies the delivery of messages and operation return values, but not which operations must be visible after message delivery. This parameter varies across implementation categories, and we model it as a visibility witness function. A visibility witness takes a concrete execution C, and produces the visibility order V(C) of its witness abstract execution.

We will give some examples of universal witness functions shortly. For a concrete application of visibility witness, compare an execution in Figure 2.4 with the visibility in Figure 2.8A.

This leads us to the final definition of implementation correctness.

Definition 2.8 (Implementation Correctness

). A data type implementation D ø satisfies a type specification F ø w.r.t. network specification T and visibility witness V, noted

D ø sat[V,T] F ø , if
the witness abstract execution of every concrete execution satisfies the specification:

8C 2 ÇD ø É \ T : abs(C,V) |= F ø . CHAPTER 2. REPLICATED DATA TYPES
All example implementations from Section 2.2.3 satisfy their specifications w.r.t. network specifications and visibility witnesses of their respective categories, listed next. We are not considering formal verification of correctness in this thesis, addressed by other work [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF].

Implementation Categories

We now define and discuss a number of network specification and visibility witness choices that together define implementation categories.

To specify them, we define the delivery order del(C) of a concrete execution C as follows:

e del(C) °°°°! f () e C.eo °°°! f ^C.action(e) = send ^C.action( f ) = deliver ^C.msg(e) = C.msg( f ).

Network Specifications

Network specification T defines what message delivery patterns are allowed. We reproduce a number of popular network specifications [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF], discuss them here, and formally define them in Appendix A. They correspond to well-known definitions from distributed system models [START_REF] Cachin | Introduction to Reliable and Secure Distributed Programming[END_REF].

If an implementation tolerates lost, duplicated, and reordered delivery, it can operate under any network specification T any , such that ÇD ø É \ T any = ÇD ø É. State-based implementations, such as Algorithm 2.3, do tolerate any network specification.

The at-most-once delivery specification T ∑1 requires that a message is not delivered twice to the same replica, and never delivered to the sender's replica, where it is already known.

At-least-once delivery T ∏1 requires that every sent message must be eventually delivered at every remote replica. The conjunction of the two is reliable delivery T 1 , or exactly-once delivery. For example, the op-based counter from Algorithm 2.2 requires reliable delivery.

Some op-based implementations require additional ordering guarantees for message delivery.

An important condition is causal delivery T c , where delivery of a message requires that all messages known to the sender were delivered at the receiver [START_REF] Cachin | Introduction to Reliable and Secure Distributed Programming[END_REF]. In Appendix A.2, we give the example of an optimized op-based multi-value register that requires reliable causal delivery.

Stronger delivery assumptions can make the object implementations simpler and more spaceefficient. However, implementing the underlying network delivery with stronger semantics can be more complex and costly [START_REF] Cachin | Introduction to Reliable and Secure Distributed Programming[END_REF].

The behavior of some arbitration-based objects relies on the properties of timestamps. Causal timestamp specification T ct requires that timestamp provided by a replica are greater than any timestamp it has previously observed. This is the case, for instance, with Lamport clocks.

Note that correctness w.r.t. Definition 2.8 does not in itself ensure the liveness of convergence, i.e., eventual visibility (Equation 2.4). This may require additional liveness restrictions, such as that every update is eventually followed by send event, i.e., eventual flush specification T f . Moreover, even implementations that tolerate any network specification may require additional fairness of network channels to ensure convergence, so that not all messages are lost [START_REF] Cachin | Introduction to Reliable and Secure Distributed Programming[END_REF].

EXECUTION MODEL AND CORRECTNESS

Our system model does not treat replica failures explicitly. Transient and permanent replica failures are indistinguishable from lost messages. We discuss the consequences where relevant.

Visibility Witnesses

A visibility witness V specifies the visibility of events as affected by message delivery.

Implementations with a transitively delivering witness, e.g., state-based implementations, transmit all updates known to the sender replica to the receiver replica. Formally:

V state (C) = (ro(C) [ del(C)) + |
do . Conversely, implementations with a selectively delivering witness [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF], e.g., op-based implementations, transmit only the latest updates of the sender:

V op (C) =ro(C) [ {(e, f ) | e, f , 2 C.E| do ^9e 0 , f 0 : e ro(C) °°°! e 0 del(C) °°°°! f 0 ro(C) °°°! f ^¬9e 00 : e ro(C)
°°°! e 00 ro(C)

°°°! e 0 ^C.action(e 00 ) = send}

Other possible witnesses include: protocols with intermediate approaches (i.e., neither opbased nor state-based) [START_REF] Almeida | Efficient state-based CRDTs by delta-mutation[END_REF], protocols that require additional communication round-trips to transmit updates [START_REF] Sérgio | Scalable eventually consistent counters over unreliable networks[END_REF], or protocols with topology-restrictions [START_REF] Burckhardt | Cloud types: Robust abstractions for replicated shared state[END_REF].

An implementation with a witnesses that transmits updates earlier and transitively may speed up convergence, especially during failures. However, it may also saturate the network with larger messages.

Visibility witness and network specifications can together enforce desirable updates ordering guarantees at an object level, such as causal consistency [START_REF] Ahamad | Causal memory[END_REF][START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF]. We make use of it in Part III.

Main Categories

We conclude with a comparison of the two main implementation categories, op-based and statebased, presented in Table 2.1. Although it is not an exhaustive list, other categories tend to be some variation or combination of the two. We will discuss some of them in Chapter 5.

The main characteristics of an implementation category are its network specification and visibility witness. Op-based implementations impose strong requirements on the network layer specification; details depend on type and implementation, but in the common case, they rely on reliable causal delivery. In contrast, state-based objects have almost no network requirements.

An op-based implementation transmits only the recent local updates, whereas a state-based transmits all updates transitively.

The presence of different implementation categories raises a natural question: which one is the best choice? There is no universal answer. We review this question from different angles, in the rows of

State metadata size lower higher

Multi-versioning integration external / easy (Part III) internal / more difficult Cross-object guarantees integration external / easy (Part III) internal / more difficult Bandwidth / buffer optimizations log reduction [START_REF] Burckhardt | Cloud types: Robust abstractions for replicated shared state[END_REF][START_REF] Pereira | Semantically reliable multicast: Definition, implementation, and performance evaluation[END_REF] deltas [START_REF] Almeida | Efficient state-based CRDTs by delta-mutation[END_REF][START_REF] Deftu | A scalable conflict-free replicated set data type[END_REF] Topology-specific optimizations possible [START_REF] Burckhardt | Cloud types: Robust abstractions for replicated shared state[END_REF] possible [START_REF] Sérgio | Scalable eventually consistent counters over unreliable networks[END_REF] Example implementations SwiftCloud (Part III), Riak DT [1] Gemini [START_REF] Li | Making geo-replicated systems fast as possible, consistent when necessary[END_REF], Walter [START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF] Table 2.1: Comparison of op-based and state-based implementation categories. them permit type-specific metadata optimizations. This motivates our metadata optimality study in Part II for state-based implementations.

Low metadata size of the op-based implementations does not indicate their cost is always lower in absolute terms, since the network layer specification that they require also has a cost.

The stronger the network specification, the smaller the op-based object metadata, but the higher is the cost and the complexity of the update delivery protocol implementation. In contrast, the delivery protocol for state-based implementations does not impose any significant cost, and in particular, does not require to store a log of messages to send. On the other hand, the delivery protocol is easily shared between op-based objects.

Another angle of comparison is support for cross-object consistency protocols, and their multi-versioning mechanisms [START_REF] Bailis | Scalable atomic visibility with RAMP transactions[END_REF]46,66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF]. These can be easily integrated externally with opbased implementations, using a standard log-based implementation of the database. In contrast, for state-based implementations this is more complex and less modular (it requires custom implementation for every type).

These factors motivate our choice of op-based category for the SwiftCloud system with crossobject consistency and versioning, described in Part III. Conversely, state-based implementations are used by object stores implementing independent objects without cross-object consistency.

Bandwidth and buffer utilization optimizations are possible for variants of both categories.

Update delivery protocols for op-based implementations can use semantics of messages (updates)

to reduce the log of updates to propagate [START_REF] Burckhardt | Cloud types: Robust abstractions for replicated shared state[END_REF][START_REF] Kistler | Disconnected operation in the Coda file system[END_REF][START_REF] Pereira | Semantically reliable multicast: Definition, implementation, and performance evaluation[END_REF][START_REF] Nuno Preguiça | Data management support for asynchronous groupware[END_REF]. Similarly, recent variants of statebased implementations send only fragments of their states [START_REF] Almeida | Efficient state-based CRDTs by delta-mutation[END_REF][START_REF] Deftu | A scalable conflict-free replicated set data type[END_REF], as we discuss in Chapter 5.

Topology-specific optimizations that rely on type semantics also exist for both categories [START_REF] Sérgio | Scalable eventually consistent counters over unreliable networks[END_REF][START_REF] Burckhardt | Cloud types: Robust abstractions for replicated shared state[END_REF]. This chapter concerns the problem of minimizing the metadata cost incurred by RDT implementations. Metadata is information required by implementations to handle concurrent updates and failures. For some implementations the space complexity of metadata is substantial, especially for state-based ones. Designing a space-optimized implementation is nontrivial -an incorrect design may violate correctness under complex concurrency or failure scenarios. This chapter studies this problem.

Part II

Optimality of Replicated Data Types

In Section 3.1, we define a metric for metadata overhead, which supports an asymptotic worstcase analysis. In Section 3.2, we review some existing state-based implementations, and study the opportunities for improvements. In particular, we look at a few difficult instances of the problem for variants of set, register and counter types. For many of them existing implementations incur an overhead in the order of the number updates, or in (the square of) the number the replicas in the system. We present the design of two implementations that improve over the existing ones, and minimize the metadata overhead down to the number of replicas. We also report on unsuccessful optimization attempts.

Problem Statement

Object states in replicated data types include not only the client-observable content, but also metadata needed for conflict resolution and for masking network failure. Similarly, messages include metadata. The space taken by this metadata is a major factor of storage and bandwidth cost, or even of feasibility. As illustrated earlier by both counter (Algorithm 2.3) and by register (Algorithm 2.5) implementations, metadata is particularly large for state-based implementations.

Such implementations are used, for example, by object stores without cross-object consistency guarantees, such as Riak 2.0 [START_REF]Data Types in Riak[END_REF]. This motivates our study of state metadata complexity for state-based implementations in this part of the thesis. We do not directly consider message size; the size of a state sets a bound on the size of a message sending that state. 1 To measure and compare space requirements of different implementations of a data type, we need a common metric. To this end, we consider how data are represented. An encoding of a set S is an injective function enc : S ! § + (i.e., encoding is decodable), where § is some suitably chosen finite set of characters that is fixed and common for all data representation; for example, binary encoding ( § = {0, 1}) or byte encoding ( § = {0, 1,...,255}). Sometimes, we clarify the domain being encoded using a subscript: e.g., enc N (1) indicates the encoding of value 1 from the domain of integers. For s 2 S, we define len S (s) as the length of enc S (s), i.e., the number of characters. The length can vary within the encoded domain: e.g., the length of encoding of an integer k is proportional to its value, i.e., len N 0 (k) 2 £(lg k). We use standard encodings schemes for primitive values (such as integers, sets, tuples, etc.) used in object return values enc Val ø of 1 We discuss some orthogonal work on message size optimizations in Chapter 5.

PROBLEM STATEMENT

each data type ø we consider; the standard encodings are explicitly specified in Appendix B.1. For sake of generality, we allow arbitrary but fixed encoding of implementation states, enc D ø .ß .

The state of an object contains both client-observable value and metadata, and both can vary within and across different executions. As the size of the value is a lower bound on the size of the state, our analysis focuses on metadata, or how much overhead does the metadata causes. We call metadata the part of the state that is not returned in a read query. Formally, for a concrete execution C 2 ÇD D ø É and a read event e 2 C.E| read , we define state(e) to be the state of the object accessed at e: state(e) = R(C.replica(e)) for (R, _) = C.pre(e). The metadata overhead is the ratio of the size of the object state over the size of the read value.

In this work, we focus on the worst-case analysis, which is relevant, for example, under high concurrency or during network partitions. It does not necessarily represent average or long-term behavior. We observe that for all the data types that we analyze the overhead is related to the number of update operations and/or the number of replicas involved in an execution. We quantify the worst-case overhead by taking the maximum of the overhead ratio over all read operations in all executions with a given number of replicas n and update operations m.

Definition 3.1 (Maximum and Worst-Case Metadata Overhead). The maximum metadata overhead of an execution

C 2 ÇD D ø É of an implementation D ø , noted mmo(D ø , C), is: mmo(D ø , C) = max Ω len D ø .ß (state(e)) len Val ø (C.rval(e))
| e 2 C.E| read ae .

The worst-case metadata overhead of an implementation D ø over all executions with n replicas and m updates (2 ∑ n ∑ m), noted wcmo(D ø , n, m), is:

wcmo(D ø , n, m) = max{mmo(D ø , C) | C 2 ÇD ø É ^n = |{C.replica(e) | e 2 C.E}| ^m = |{e 2 C.E | C.op(e) 2 Op ø \ {read}}|}.
The definition of mmo takes into account only states prior to read queries, which may appear limited. However, wcmo definition quantifies over all executions, which includes any number of read queries. Similarly, we consider only executions with m ∏ n, since we are interested in the asymptotic overhead of executions where all replicas can be mutated (i.e., perform at least one update operation); we ignore read-only replicas for brevity.

The definition of the worst-case metadata overhead allows us to express the asymptotic complexity of implementations, a classical tool in worst-case analysis. • f is an asymptotic upper bound, noted

D ø 2 b O( f (n, m)), if sup n,m!1 wcmo(D ø , n, m) f (n, m) < 1, i.e., 9K > 0 : 8m ∏ n ∏ 2 : wcmo(D ø , n, m) < K f (n, m). • f is an asymptotic lower bound, noted D ø 2 b ≠( f (n, m)), if lim n,m!1 wcmo(D ø , n, m) f (n, m) 6 = 0, i.e., 9K > 0 : 8m 0 ∏ n 0 ∏ 2 : 9n ∏ n 0 , m ∏ n 0 : wcmo(D ø , n, m) > K f (n, m).
• f is an asymptotically tight bound, noted D ø 2 b £( f (n, m)), if it is both an upper and a lower asymptotic bound.

Note that we use the d hat symbol to distinguish metadata overhead bounds from function bounds.

An asymptotic bound characterizes metadata overhead of an implementation concisely, and serves to derive a notion of asymptotic implementation optimality.

O( f (n, m)) if it is correct, i.e., D ø sat[V state , T any ] F ø ,
and its upper bound is a lower bound for all correct implementations:

8D 0 ø : D 0 ø sat[V state , T any ] F ø =) D 0 ø 2 b ≠( f (n, m))
The goal of this part of the thesis is the design and/or identification of asymptotically optimal state-based implementations.

Optimizing Implementations

In this section, we analyze the metadata overhead of some existing data type implementations, and explore opportunities for improved, optimized designs.

Successful Optimizations

We begin with two successful nontrivial metadata optimizations using a variant of version vectors.

They reduce the overhead to the order of the number of replicas in the system.

Add-Wins Set

The set is one of the most fundamental data types. It can be used directly and as a basis for types such as maps or graphs [START_REF] Brown | Riak DT map: A composable, convergent replicated dictionary[END_REF][START_REF] Gotsman | Composite replicated data types[END_REF][START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF]. Among the alternatives discussed in Section 2.3.2.3, the add-wins set semantics F AWSet (Equation 2.10) is a popular one [START_REF]Data Types in Riak[END_REF][START_REF] Deftu | A scalable conflict-free replicated set data type[END_REF][START_REF] Mukund | Optimized OR-sets without ordering constraints[END_REF].

Existing implementation. Algorithm 3.1 presents the basic state-based implementation of add-wins set by Shapiro et al. [START_REF] Shapiro | Conflict-free replicated data types[END_REF], translated into our notation.

To implement add-wins, the idea is to distinguish different invocations of add(a) by attaching, to each added element, a hidden unique token, or timestamp, t. We use a custom timestamp t = (r, k), where r is the replica ID and k is a natural number assigned by that replica. The concrete implementation stores a set A of active element instance triples (a, r, k), one for each 1:

ß = ReplicaID £ P(Z £ ReplicaID £ N 0 ) £ P(ReplicaID £ N 0 ) 2: M = P(Z £ ReplicaID £ N 0 ) £ P(ReplicaID £ N 0 ) 3: initialize(r i ) : (r, A, T) 4: let r = r i . replica ID 5: let A = ;
. active element instances: triples (element a, replica r 0 , counter k)

6: let T = ;
. tombstones: timestamps of removed instances: pairs (replica r 0 , counter k)

7: do(read, t o ) : V 8: let V = {a | 9r 0 , k : (a,
let k = max{k 0 | (_, r, k 0 ) 2 A _ (r, k 0 ) 2 T _ k 0 = 0} + 1 . next timestamp for new instance 11: A √ A [ {(a,
let D = {(a, r 0 , k) | 9r 0 , k : (a, r 0 , k) 2 A}
. identify all instances of element a 14:

A √ A \ D . remove them from active instances 15:

T √ T [ {(r 0 , k) | (a, r 0 , k) 2 D}
. add their timestamps to tombstones 16: send()

: (A m , T m ) 17: let A m = A 18: let T m = T 19: deliver((A m , T m )) 20: A √ (A [ A m ) \ {(a, r 0 , k) | (r 0 , k) 2 T [ T m }
. merge all instances, except tombstones 21:

T √ T [ T m
. merge sets of tombstones add(a), rather than just elements. An element a is removed by removing every instance (a, r, k)

of that element from the active set and recording its timestamp (r, k) in a tombstone set T.

Timestamps in A and tombstones in T are therefore maintained as disjoint sets. An element is in the set, i.e., is included in return value of read, if there is an instance of it in the active set; tombstones are not visible to read, and they are used only during replication as a trace of removal. A tombstoned element, can be always added again by creating a new instance (a, r 0 , k 0 ) with a fresh timestamp (r 0 , k 0 ), different from the old one, (r 0 , k 0 ) 6 = (r, k). If the same element a is both added and removed concurrently, the remove concerns only observed instances and not the concurrently-added unique instance. Therefore the add wins, by adding a new instance. The replication protocol transmits information about active and tombstone instances. The receiver removes active instances that are present in the received tombstones set. Burckhardt et al. [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF] prove that this implementation indeed satisfies the specification F AWSet . This approach is correct, but incurs high metadata overhead, in the order of number of updates m in an execution. A formal proof is in Appendix B.2. Intuitively, every add requires a new instance (even sequential or concurrent adds of a same element), which remains as a timestamp in the tombstone set after being removed. The factor lg m is the cost of storing the identity of each instance.

∅ send deliver {(a,r 1 ,1),(b,r 1 ,2),(a,r 2 ,1)} ∅ add(b) {(a,r 1 ,1),(b,r 1 ,2)} ∅ rem(a) {(b,r 1 ,2)} {(a,r 1 ,1)} send rem(b) ∅ {(a,r 1 ,1),(b,r 1 ,2)} deliver {(a,r 2 ,1)} {(a,r 1 ,1),(b,r 1 ,2)} deliver {(a,r 2 ,1)} {(a,
r 1 r 2 r 3 A=∅ vv=[0 0 0] A=∅ vv=[0 0 0] A=∅ vv=[0 0 0] add(a) {(a,r 1 ,1)} [1 0 0] send add(a) {(a,r 2 ,1)} [0 1 0] deliver {(a,r 1 ,1),(b,r 1 ,2),(a,r 2 ,1)} [2 1 0] send deliver {(a,r 1 ,1),(b,r 1 ,2),(a,r 2 ,1)} [2 1 0] add(b) {(a,r 1 ,1),(b,r 1 ,2)} [2 0 0] rem(a) {(b,r 1 ,2)} [2 0 0] send rem(b) ∅ [2 0 0] deliver {(a,r 2 ,1)} [2 1 0] deliver {(a,r 2 ,1)} [2 1 0] read:{a} read:{a} deliver {(a,r 2 ,1), (c,r 3 ,1)} [2 1 1] read:{a,c} add(c) {(a,r 1 ,1),(b,r 1 ,2),(a,r 2 ,1),(c,r 3 ,1 } [2 1 1] send GC not needed (B) Execution of the optimized implementation (Algorithm 3.2).
Algorithm 3.2 Optimized state-based implementation of add-wins set (AWSet).

1: ß = ReplicaID £ (ReplicaID ! N 0 ) £ P(Z £ ReplicaID £ N 0 ) 2: M = (ReplicaID ! N 0 ) £ P(Z £ ReplicaID £ N 0 ) 3: initialize(r i ) : (r,
let D = {(a, r 0 , k) | 9r 0 , k : (a, r 0 , k) 2 A}
. identify all instances of element a

let A 0 = A \ {(a, r 0 , k) 6 2 A m | k 2 N ^k ∑ vv m (r 0 )} . keep local instances not rem. remotely 21: let A 00 = A m \ {(a, r 0 , k) 6 2 A | k 2 N ^k ∑ vv(r 0 )} .

keep remote instances not removed locally 22:

A √ A 0 [ A 00

23:

vv √ vv t vv m . update vector to entry-wise maximum (cf. Equation 2.2)

Discarding instances is not simple. Unfortunately, getting rid of instances is not easy in a system with failures. They protect from duplicated and reordered messages, and disconnected replicas. A background garbage collection (GC) protocol could collect unnecessary instances [START_REF] Golding | Weak-consistency group communication and membership[END_REF][START_REF] Gene | Efficient solutions to the replicated log and dictionary problems[END_REF] (we will discuss it in Chapter 5). However, this incurs overhead too, and requires update stability for progress, i.e., every replica must acknowledge receiving an update. Replica can be exempted from acknowledgment only if it permanently crashed, but such an exemption requires perfect failure detector [START_REF] Cachin | Introduction to Reliable and Secure Distributed Programming[END_REF], thus network or replica failure can prevent GC for extended duration.

On the other hand, it is unsafe to GC an instance before it is known to be replicated everywhere. Optimization 1: Efficient summary of removed elements without tombstones. Our solution is similar to the metadata of a replicated file system of Bjørner [START_REF] Bjørner | Models and software model checking of a distributed file replication system[END_REF] (see Chapter 5).

An execution from

It is based on the observation that an implementation does not need to store an explicit set of tombstones to represent information about removals. If each replica assigns timestamps contiguously, i.e., (r i , 1), (r i , 2),..., and the protocol ensures their in-order delivery (as in the case of state-based transitive delivery), it suffices to recall the number of adds from each replica. Thus, a version vector can compactly summarize knowledge about the timestamps that the replica has observed. Formally, let function TS(vv) define the timestamps represented by a vector vv:

(3.1)

TS(vv) = {(r, k) | r 2 dom(vv) ^k 2 N ^k ∑ vv(r)}.
The concrete state in the optimized implementation is a triple (r, vv, A), where r is the replica ID, vv is a vector, and A is a set of active instances. There is no tombstone set. The timestamps represented by vv include both active instances A and timestamps of the missing tombstones.

Although the tombstones set is not part of state, it is possible to logically reconstruct it from a concrete state (r, vv, A) of the optimized implementation. We can write:

TS(vv) = {(r, k) | 9a : (a, r, k) 2 A} ] T(vv, A)
, where T(vv, A) is the logical tombstone set. Therefore: The optimized algorithm needs to use this new data structure. The add method now associates an instance to the replica's own entry in the vector, which is incremented. The rem method is unchanged compared to the naive implementation, except for the absence of the tombstone set.

(3.2) T(vv, A) = TS(vv) \ {(r, k) | 9a : (a, r, k) 2 A}.
The most complex optimization concerns the deliver replication logic. Version vector is used to check for timestamp inclusion. One way to reason about correctness of the optimization is to relate it with the naive implementation. The deliver algorithm is equivalent to the naive implementation. Consider any state (r, A, T) of the naive implementation (resp., an equivalent optimized state (r, vv, A)) and a message (A m , T m ) (resp., an equivalent optimized message (vv m , A m )). The delivery of the message in Line 20 of the naive implementation is equivalent to Line 22 of the optimized one:

A (i) √ (A [ A m ) \ {(a, r 0 , k) | 9(r 0 , k) 2 T [ T m } (3.3) = (A \ {(a, r 0 , k) | 9(r 0 , k) 2 T [ T m }) [ (A m \ {(a, r 0 , k) | 9(r 0 , k) 2 T [ T m }) (ii) = (A \ {(a, r 0 , k) | 9(r 0 , k) 2 T m }) [ (A m \ {(a, r 0 , k) | 9(r 0 , k) 2 T}) (iii) = (A \ {(a, r 0 , k) 6 2 A m | (r 0 , k) 2 TS(vv m )}) [ (A m \ {(a, r 0 , k) 6 2 A | (r 0 , k) 2 TS(vv)}) (iv) = (A \ {(a, r 0 , k) 6 2 A m | k 2 N ^k ∑ vv m (r 0 )}) [ (A m \ {(a, r 0 , k) 6 2 A | k 2 N ^k ∑ vv(r 0 )}),
where (i) is the assignment from Line 20 of the naive implementation; step (ii) is due to disjointness of A and T (also, A m and T m ), (iii) is an application of Equation 3.2, and (iv) comes from application of Equation 3.1. A similar argument holds for vectors. Therefore, we can "coalesce" them: rather than storing all instances of add(a), it suffices to record a vector of the latest timestamps for element a generated by each replica, or equivalently, a set of the latest timestamps for a from each replica.

The optimized add(a) method does exactly that in Line 10 of Algorithm 3.2. When a new instance of element a is generated, the set D of prior instances of a generated by this replica is discarded; D is an empty set or a singleton. This guarantees that in any state A the number of different instances of a same element a is at most in the number of replicas n.

such that D § AWSet sat[V state , T any ] F AWSet . The complexity of D § AWSet is b £(n lg m).
A proof is in Appendix B.2. Intuitively, the vector vv has n entries (or less, with some additional minor optimizations that we omit), and every element a has at most n different instances in the set A. The lg m factor is the size of a vector entry. In Chapter 4 we will show that this implementation is optimal, i.e., further state reduction for the worst-case is impossible without affecting fault-tolerance or semantics.

Burckhardt et al. [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF] formally prove the optimization correct w.r.t. F AWSet specification.

Mukund et al. [START_REF] Mukund | Optimized OR-sets without ordering constraints[END_REF] offers an alternative proof via bisimulation between the naive and the optimized implementation, along the lines of Equation 3.3 that shows part of the obligations.

Applicability. The optimization for add-wins set is relatively general and could be applied to other similar types, for instance, an add-wins map object [START_REF]Data Types in Riak[END_REF]. The optimized set can be also viewed as an optimized implementation of a collection of value-wins registers F VWReg (Equation 2.8)

restricted to boolean values. Every register corresponds to a unique element; if an element is present in the set, then the corresponding register has the winning value.

Multi-Value Register

The multi-value register is a basic primitive for conflict detection and application-level conflict resolution. Its semantics F MVReg are specified in Equation 2.7.

Existing implementation. We first analyze the standard implementation from Algorithm 2.5 (Section 2.2.3), proposed in similar forms by different authors [START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF][START_REF] Kistler | Disconnected operation in the Coda file system[END_REF][START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF].

Theorem 3.3. Let D MVReg be the multi-value register implementation defined in Algorithm 2.5,

such that D MVReg sat[V state , T any ] F MVReg . The complexity of D MVReg is b £(n 2 lg m).
A proof is in Appendix B.2. The overhead may appear surprisingly high, since each value stored by the implementation carries a single version vector, which could suggest an overhead of n lg m. However, in the most costly case, the same value could be written concurrently by up

let A 0 = {(a, F {vv | (a, vv 0 ) 2 A [ A M }) | (a, _) 2 A [ A M } . compute max. vector per element 5:
A √ {(a, vv) 2 A 0 | 8(a 0 , vv 0 ) 2 A 0 : vv 6 < vv 0 } . keep elem. not dominated by any other elem.

to n replicas. In this case, the actual observable value is a singleton, whereas the state stores n version vectors. 3We illustrate this problem in Figure 3.4A. Three replicas concurrently write value 0, assigning it three different version vectors. A fourth replica r 4 delivers their writes, and needs to store all three vectors. This causes n 2 lg m overhead, as demonstrated by a read operation on replica r 4 .

The overhead reduces only with subsequent writes.

This overhead may be problematic in practice, as we expect states made of concurrent writes of the same value, which do not require conflict resolution, to last longer than transient states with unresolved conflicting writes. Moreover, this happens in the desirable case of concurrent convergent conflict resolution.

Incorrect optimization.

In an optimization attempt, we explored the idea of merging the vectors of identical elements in order to store at most one vector per unique value. Algorithm 3.3 applies this idea in the deliver function.

It merges two states by first merging entries for identical elements. The vector of a merged entry is the least upper bound of all vectors for the element (Equation 2.2). This step eliminates multiple entries per element, and brings the overhead down to b £(n lg m). In the second step of deliver, the implementation attempts to eliminate dominated entries, in the same way as the original implementation, by removing an entry if it is dominated by another one. As it turns out, these optimizations are incorrect. This demonstrates the difficulty of correct optimization. where i 2 {1, 2, 3}, therefore 0 should not appear in the return value at replica r 4 , Correct optimization. Algorithm 3.4 presents a correct optimized design of multi-value register implementation. This design uses a more complex logic to eliminate merged entries.

r 1 ∅ r 2 ∅ r 3 ∅ r 4 ∅ read:{1,2,3} write(0) {(0, [1 0 0])} send write(0) {(0, [0 1 0])} write(0) {(0, [0 0 1])} send send deliver {(0, [0 0 1])} deliver {(0, [0 0 1]), (0, [0 1 0])} deliver {(0, [0 0 1]), (0, [0 1 0]), (0, [1 0 0])} read:{0} write(1) {(1, [2 0 0])} send write(2) {(2, [0 2 0])} write(3) {(3, [0 0 2])} send send deliver {(3, [0 0 2]), (0, [0 1 0]), (0, [1 0 0])} deliver {(3, [0 0 2]), (2, [0 2 0]), (0, [1 0 0])} deliver {(3, [0 0 2]), (2, [0 2 0]), (1, [2 0 0])} (A) Execution of the basic implementation (Algorithm 2.5). r 1 ∅ r 2 ∅ r 3 ∅ r 4 ∅ read:{0,1,2,3} write(0) {(0, [1 0 0])} send write(0) {(0, [0 1 0])} write(0) {(0, [0 0 1])} send send deliver {(0, [0 0 1])} deliver {(0, [0 1 1])} deliver {(0, [1 1 1])} read:{0} write(1) {(1, [2 0 0])} send write(2) {(2, [0 2 0])} write(3) {(3, [0 0 2])} send send deliver {(3, [0 0 2]), (0, [1 1 1])} deliver {(3, [0 0 2]), (2, [0 2 0]), (0, [1 1 1])} deliver {(3, [0 0 2]), (2, [0 2 0]), (1, [2 0 0]), (0, [1 1 1])}
(B) Execution of the incorrectly optimized implementation (Algorithm 3.3).

r 1 ∅ r 2 ∅ r 3 ∅ r 4 ∅ read:{1,2,3} write(0) {(0, [1 0 0])} send write(0) {(0, [0 1 0])} write(0) {(0, [0 0 1])} send send deliver {(0, [0 0 1])} deliver {(0, [0 1 1])} deliver {(0, [1 1 1])} read:{0} write(1) {(1, [2 0 0])} send write(2) {(2, [0 2 0])} write(3) {(3, [0 0 2])} send send deliver {(3, [0 0 2]), (0, [1 1 1])} deliver {(3, [0 0 2]), (2, [0 2 0]), (0, [1 1 1])} deliver {(3, [0 0 2]), (2, [0 2 0]), (1, [2 0 0])} (C) Execution of the correctly optimized implementation (Algorithm 3.4).
let A 0 = {(a, F {vv | (a, vv 0 ) 2 A [ A M }) | (a, _) 2 A [ A M } . compute max. vector per element 5: A √ {(a, vv) 2 A 0 | vv 6 v F {vv 0 | 9a 0 6 = a : (a 0 , vv 0 ) 2 A 0 }}} . keep elem. not domin.

by all others

Similarly to the previous implementation, deliver first merges entries with identical elements, using the least upper bound of all vectors for the element as a merged vector. The second phase, of eliminating dominated entries, is different. It removes an entry if the least upper bound of vectors of entries for all other elements dominates the vector of that entry, i.e., it expresses the case where an entry was merged, but different writes overwrite the original parts of the merged entry. This rule subsumes the incorrect rule which considered only a direct domination by a single entry. This logic allows to discard merged entries correctly, as proved by Burckhardt et al. [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF].

An execution from Figure 3.4C illustrates how the optimized implementation corrects the problematic counterexample. This time, when replica r 4 delivers concurrent write(1), write(2), and write(3) operations that overwrite write(0), it discards the entry for value 0 correctly. Indeed, the merged vector it stores for value 0, namely [r A proof is in Appendix B.2. This implementation is guaranteed to store one entry per element, yielding the overhead of a single vector. In Chapter 4, we will show that this implementation is asymptotically optimal; in Chapter 5 we will also discuss a recent improvement of Almeida et al.

1 7 ! 1, r 2 7 ! 1, r 3 7 ! 1], is
[6], with the same asymptotic complexity, but improved average case.

Prior Implementations and Unsuccessful Optimizations

A number of other important data types exist, for which we did not manage to find a correct optimization. Some of them provide a similar behavior to add-wins set and multi-value register, and could act as their replacement. We report on their complexity here, which motivates some of the questions regarding optimality that we address in Chapter 4.

Last-Writer-Wins Register

In the presence of nonnegligible overhead of multi-value register implementation, it is natural to investigate the complexity of arbitration-based last-writer-wins register. LWW register is specified as F LWWReg in Equation 2.6. We presented the implementation in Section 2.2.3 (Algorithm 2.4).

Algorithm 3.5 State-based implementation of remove-wins set (RWSet).

1: ß = P(Z £ Timestamp) £ P(Z £ Timestamp) M = P(Z £ Timestamp) £ P(Z £ Timestamp) 2: initialize(r i ) : (A, T)

3: let T = ;
. remove instances: pairs (element a, timestamp t)

4: let A = ;
. add instances covering remove instances: pairs (element a, timestamp t)

let A m = A 18: let T m = T 19: deliver((A m , T m )) 20: T √ (T [ T m ) \ (A [ A m )
. union remove instances excluding those covered by adds A proof is in Appendix B.2. LWW register incurs a negligible logarithmic overhead (the cost of a timestamp). In Chapter 4, we will show that this marginal cost is unavoidable. Compared to the multi-value register, the overhead is much lower for the LWW register, particularly when the number of replicas is high or there is a replica churn, i.e., replicas disappear and join frequently.

Remove-Wins Set

Given that implementations of the add-wins set have noticeable overhead, it is natural to consider other semantics of sets. Algorithm 3.5 is an implementation of remove-wins set semantics F RWSet (Equation 2.10) by Bieniusa et al. [START_REF] Bieniusa | An optimized conflict-free replicated set: Additional material[END_REF].

The implementation is somewhat similar to the naive add-wins set, as it keeps track of every add and rem operations in two sets: a set of add instances A with pairs (a, t), where a is an element and t is a timestamp, and a set of remove instances T with the same type of pairs. The two sets are disjoint. The read method returns an element a if there is an add(a) instance and no rem(a) instance. The rem method simply creates a new remove instance. The add operation turns any remove instances of the element into add instances, or, if there is none, creates a new add instance. The replication protocol is similar to the naive add-wins set.

OPTIMIZING IMPLEMENTATIONS

To our surprise, the overhead of the remove-wins implementation is even higher than that of the naive add-wins implementation. Recall that the naive add-wins implementation stored only timestamps of removed elements, not their value. The implementation of remove-wins semantics must store also the value of removed elements, because removes must dominate concurrent adds unaware of their timestamps. Since removed elements are of variable, unbounded size, there is no upper bound w.r.t. m or n. 4 Formally, we can give only an underestimated lower bound.

Theorem 3.6. Let D RWSet be the remove-wins set implementation defined in Algorithm 3.5,

assuming that D RWSet sat[V state , T any ] F RWSet . The complexity of D RWSet is b ≠(m lg m).
A proof is in Appendix B.2. Intuitively, the implementation stores at least a timestamp for every remove operation, which incurs the worst-case overhead in the order of the number of operations.

In Chapter 4, we prove that the remove-wins semantics requires to maintain precise information about every rem operation, which makes it impossible to efficiently summarize information about removed instances, similar to the add-wins set. 5 A practical implementation can only resort to a background garbage collection protocol [START_REF] Golding | Weak-consistency group communication and membership[END_REF][START_REF] Johnson | The maintenance of duplicate databases[END_REF], which raises other issues.

Last-Writer-Wins Set

Another alternative set semantics is the last-writer-wins (LWW) F LWWSet , specified in Equation 2.9.

Algorithm 3.6 presents the implementation of LWW set of Bieniusa et al. [START_REF] Bieniusa | An optimized conflict-free replicated set: Additional material[END_REF], improved over Shapiro et al. [START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF]. The implementation maintains the set of latest operations, one for each element updated by either add or rem. The latest operation is selected according to the timestamp.

The type of operation is represented as a presence flag: 1 for add and 0 for remove. An element a is included in read value if the operation with the latest timestamp has a positive flag value.

Although the implementation is simple, it also incurs an unbounded metadata overhead, as it stores removed elements. Therefore, we claim only an underestimated lower bound.

Theorem 3.7. Let D LWWSet be the LWW set implementation defined in Algorithm 3.6, assuming

that D LWWSet sat[V state , T any ] F LWWSet . The complexity of D LWWSet is b ≠(m lg m).
A proof is in Appendix B.2. Intuitively, the implementation stores timestamps for removed elements, which incurs the overhead in the order of the number of removes.

It is an open problem whether storing timestamps per updated element is necessary for LWW set semantics. In Chapter

S √ {(a, t, v) 2 S [ S m | 8(a, t 0 , v 0 ) 2 S [ S m : t 0 6 = t =) t 0 < t} .
keep the latest flag per elem.

Note that LWW set can be viewed as a collection of LWW boolean registers of presence flags, one for each element, similarly to the add-wins set considered earlier as a collection of presencewins registers. Unlike the add-wins set, where some metadata can be shared in a collection, an implementation of the LWW set is no cheaper than implementation of independent registers.

Counter

The counter implementation in Algorithm 2.3 (Section 2.2.3) uses vectors to keep track of the number of increments per replica. We can now formally state its complexity.

Theorem 3.8. Let D § Ctr be the counter implementation defined in Algorithm 2.3, such that

D § Ctr sat[V state , T any ] F Ctr . The complexity of D § Ctr is b £(n).
A proof is in Appendix B.2. Intuitively, the size of a vector is n times bigger than the size of a scalar counter value. In Chapter 4 we will prove that this implementation is optimal.

Summary

In this chapter, we introduced the metadata space complexity problem and investigated optimization opportunities for six data types. We found that many implementations incur high metadata overhead, linear in the number of updates, or polynomial in the number of replicas. We presented two optimizations, for the add-wins set and the multi-value register. We also found that the complexity of the best known implementations highly depends on the data type semantics.

In the previous chapter, we focused on positive results, i.e., implementations with reduced metadata complexity. The design of optimized implementation is a laborious process that requires developing non-trivial solutions. There is a limit to this process, beyond which optimizations within an implementation category are impossible. In this chapter, we introduce general lower bound proofs on metadata overhead, which identify the limits, i.e., we present impossibility results. Lower bounds show that some of the presented implementations, and in particular our optimized designs, are optimal.

In Section 4.1, we present a technique for proving a lower bound on metadata overhead that applies to any state-based implementation of a certain type. In Section 4.2, we apply it to prove optimality of four state-based implementations, and simply lower bounds for two other data types. In Section 4.3, we summarize both positive and impossibility results from both chapters.

Together, they offer a comprehensive view of the metadata overhead problem.

Although we present optimization and impossibility results separately, as an intellectual process, they influenced one another. Specifically, designing optimized implementations helped us with lower bound proofs. Conversely, proofs directed us towards some optimizations. Generally, they help to understand the limiting factors in the semantics and in the implementation category.

Proof Technique

Our previous proofs of lower and upper bounds for a single concrete implementation rely on relatively standard techniques. (Which is why we deferred all prior proofs to the appendix).

However, proving a lower bound that applies to any implementation is challenging.

Experiment Family

The goal is to show that for any correct state-based implementation D ø (that is, such that We encapsulate this core argument in the following definition and lemma. .

D ø sat[V state , T any ] F ø ),
The lemma indicates that some culprit read in the experiment family is responsible for the worst-case metadata overhead related to the size of the family and of the returned value. To apply Lemma 4.1 to the best effect, we need to find experiment families with |Q| as large as possible and len(C AE .rval(e AE )) as small as possible. Finding such families for a specific data type is challenging, as we have no systematic way to derive them. When searching for experiment families for a given specification, we rely on intuitions about which situations force replicas to store a lot of information.

Driver Programs

To simplify the presentation, we avoid specifying an experiment family directly as a sequence of concrete execution steps, by defining it indirectly using a driver program that generate these execution steps (e.g., see Table 4.1). A driver program is written in imperative pseudocode with three types of instructions, each of which triggers a uniquely-determined configuration transition:

• v √ do r o t Do operation o at replica r with timestamp t, assign the return value to v.

• send r (mid) Send a message with identifier mid at replica r.

• deliver r (mid) Deliver the message with identifier mid at replica r.

We structure code into procedures, and define a program as a sequence of procedure calls.

When a driver program terminates, it may produce return value from the last procedure.

For a program P, an implementation D ø , and a configuration (R, N), we let exec(D ø , (R, N), P) be the concrete execution of the data type implementation D ø starting in (R, N) that results from running P; we define result(D ø , (R, N), P) as the return value of P in this run.

Programs explicitly supply timestamps for the do instruction and message identifiers for send and deliver instructions. We require that they do this correctly, i.e., that they respect the uniqueness of timestamps, message IDs, and deliver only previously-sent messages. This is the case for all our driver programs. 1 For most of our programs, timestamps (arbitration) are irrelevant, as they concern visibility-based data types, but we include them for completeness and make use of them for arbitration-based data types.

Lower Bounds

In this section we apply the above proof technique to obtain lower bounds for six data types.

Counter

We begin with one of the simplest examples: the counter data type (Ctr). We start by formulating a suitable experiment family. send r (mid r,i )

6: procedure exp(AE)

7:
for all r 2 [2.

.n] do 8:

deliver 1 (mid r,AE(r) )

9:

do 1 read (n+2)m B e AE 10:
B (culprit read)

11: procedure test(r)

12:

v √ do 1 read (n+3)m 13:

deliver 1 (mid r, m n°1 ) 14:

v 0 √ do 1 read (n+4)m 15:
return m n°1 °(v 0 °v)

Definition of exp. execution AE 2 Q C AE = exec(D ø , (R 0 , N 0 ), init; exp(AE))
where (R 0 , N 0 ) = (D ø .initialize, ;) 2) Informally, 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.

Definition of read-back function readback(ae) = ∏r : [2..n].result(D ø , (R init [1 7 ! ae], N init ), test(r)) readback : D ø .ß ! Q where (R init , N init ) = post(exec(D ø , (R 0 , N 0 ), init))
α(2) = 1 α(3) = 3 configuration (R init [1→R α ( 1
∏K 1 lg | §| ( m n°1 ) n°1 len N (m) ∏ K 2 n lg(m/n) lg m ∏ K 3 n lg p m lg m ∏ K n.
Our experiment family illustrates a specific case where the overhead may reach level b £(n). It is not exhaustive, in the sense that many other executions with n replicas and m updates may reach the same level of overhead. Other scenarios may be more complex, e.g., involve replicas that were disconnected for a long time and communicate back, or transitive delivery of updates.

Add-Wins Set

Theorem 4.2.

If D AWSet sat[V state , T any ] F AWSet , then the complexity of D AWSet is b ≠(n lg m). Lemma 4.3. If D AWSet sat[V state , T any ] F AWSet , n ∏ 2 and m ∏ n is such that (m °1
) is a multiple of (n °1), then the tuple (Q, n, m, C, e, readback) in Table 4.2 is an experiment family.

The main idea of the experiments defined in this lemma is to force replica 1 to remember the number of observed element instances even after they have been removed at that replica; cf. the optimized add-wins set implementation from Section 3.2.1.1. The experiments follow a similar pattern to those for Ctr, but use different operations. In the 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(AE), one message from each replica r = 2..n, selected by AE(r), is delivered to replica 1. At the end of execution, replica 1 removes 0 from the set and performs the culprit read e AE . The return value of this read is always the empty set, by F AWSet .

To perform the read-back of AE(r) for replica 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 D AWSet sat[V state , T any ] F AWSet and the definition of 2.10, we get that exactly the first AE(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 AE(r) = i °1. If 0 does not appear by the time the loop is finished, then AE(r) = (m °1)/(n °1).

F AWSet from Equation
For example, consider an experiment and test(2) in send r (mid r,i )

6: procedure exp(AE)

7:
for all r 2 [2.

.n] do 8:

deliver 1 (mid r,AE(r) )

9:

do 1 rem(0) (n+2)m 10:

do 1 read (n+3)m B e AE 11:
B (culprit read)

C AE = exec(D ø , (R 0 , N 0 ), init; exp(AE))
where (R 0 , N 0 ) = (D ø .initialize, ;) 2) Therefore, the second send was the first that made element 0 appear, which indicates how many add(0) operations from replica 2 were sent to replica 1 during the experiment:

Definition of read-back function readback(ae) = ∏r : [2..n].result(D ø , (R init [1 7 ! ae], N init ), test(r)) readback : D ø .ß ! Q where (R init , N init ) = post(exec(D ø , (R 0 , N 0 ), init))
α(2) = 1 α(3) = 3 configuration (R init [1→R α (1)], N init ) add (
AE(2) = 2 °1 = 1.
Proof of Lemma 4.3. The only nontrivial obligation is to prove that readback(state(e AE ))

= AE. Let (R AE , N AE ) = final(C AE ). Then AE(r) (i) = result(D AWSet , (R 0 , N 0 ), (init; exp(AE); test(r))) = result(D AWSet , (R AE , N AE ), test(r)) (ii) = result(D AWSet , (R init [1 7 ! R AE (1)], N init ), test(r)) = readback(R AE (1))(r) = readback(state(e AE ))(r),
where:

(i) This is due to We will now analyze the visibility relation between all add(0) and rem(0) events. None of the add(0) observed the rem(0) in its operation context in abs(C 0 AE , V state ). The operation context of rem(0) includes the first AE(r 0 ) operations add(0) from every replica r 0 , including replica r, i.e., it covers events (b) and part of events (a). Therefore, read into v i observes max(0, i °AE(r)) operations add(0) from r that were not visible to rem(0). By specification F AWSet the return value of the read in this case is:

D AWSet sat[V state , T any ] F AWSet ,
v i = ( {0} if i °AE(r) > 0
; otherwise i.e., {0} appears in the return value when read observes add(0) concurrent to rem(0).

Therefore:

AE(r) = min{i | v i+1 = {0} _ i = m °1 n °1 }. (ii) We have N init = N AE because exp(AE) does not send any messages. Besides, R init [1 7 ! R AE (1)]
and R AE can differ only in the states of the replicas 2..n. These cannot influence the run of test(r), since it performs execution steps on replica 1 only.

A proof of Theorem 4.2 is similar to the proof for the counter. Thus, we defer it to Appendix B.3.

Since the remaining proofs also follow a similar structure, hereafter we focus on the intuition and their distinctive aspects, and refer the reader interested in the details to the appendix. 4.3 is an experiment family.

Remove-Wins Set

The main idea of the experiments is to force replica 1 to remember a bit of information about every element removed from the set; cf. the implementation of remove-wins set in Algorithm 3.5, which remembers even more information in tombstone instances. The structure of the experiment is different from previous examples. Since the overhead bound is independent of the number of replicas, three replicas are sufficient for the experiment; the remaining replicas perform only "dummy" read operations to fulfill the wcmo definition w.r.t. the number of replicas n. In the init phase, replica 2 performs m/4 remove operations for elements 1..m/4, interleaved with sending messages; concurrently, replica 3 also removes all these elements, sends a message, adds all of them again, and sends another message, performing m/2 operations in total. Removes from replica 3 will hide and uncover elements when desired.

In the experiment, replica 1 delivers a subset of the messages with removes sent by replica 2 in the order that they were sent, and adds elements i = 1..m/4. The subset of messages to deliver is controlled by index AE which contains a set of elements. If an element a is in AE, then replica 1 delivers a message containing rem(a) operation from replica 2 before replica 1 adds a to the set; otherwise it adds a to the set without being aware of the concurrent remove. Eventually, replica 1 delivers a message from replica 3 that removes all elements, and performs the culprit read. The read returns an empty set, by F RWSet , since removes from replica 3 dominate all adds.

The read-back function retrieves the whole index AE at once. First, it delivers the last messages from replicas 2 and 3 to replica 1, which includes all removes at replica 2 and all adds at replica 3, dominating previous removes from replica 3. Then it performs a read that returns the set of elements corresponding directly to the index AE. To understand how it works, consider the witness abstract execution for this execution (e.g., Figure 4.3B). The read observes the complete set of add and remove operations from all replicas. The visibility relation is partially determined by the experiment index AE. All removes from replica 3 are covered by adds at replica 3, whereas a subset of removes from replica 2 is covered by adds from replica 1, according to the experiment index AE; remaining removes are concurrent to adds. From F RWSet , those elements for which there is a removes without covering add must not appear in the read return value.

For example, consider an experiment and test in Figure 4.4. In the experiment phase, we observe that add(1) and add(3) at replica 1 observe, respectively, rem(1) and rem(3) from replica 2, whereas add( 2) is concurrent to rem(2) at replica 2. Therefore, when the read-back delivers the last message from replica 3 to replica 1, the read returns value {1, 3}.

C AE = exec(D ø , (R 0 , N 0 ), init; exp(AE))
where (R 0 , N 0 ) = (D ø .initialize, ;) 4.4 is an experiment family.

Definition of read-back function readback(ae) = result(D ø , (R init [1 7 ! ae], N init ), test) readback : D ø .ß ! Q where (R init , N init ) = post(exec(D ø , (R 0 , N 0 ), init))

Last-Writer-Wins Set

The idea of the experiments is to force replica 1 to remember the latest timestamp it observed from each replica, for some elements, which then introduces an overhead proportional to the n lg m. The existing implementation in Algorithm 3.6 introduces higher overhead.

In the init phase of experiments, each replica 2..n performs m 2(n°1) operations, interchangeably adding and removing the same element equal to replica ID, and sending a message after each operation. We use monotonically growing timestamps for operations at each replica, which induces their arbitration order consistent with the order of operations at each replica. All messages send in the init phase of the program are undelivered until the second phase, defined by exp(AE). In the experiment phase exp(AE), one message from each replica r = 2..n is delivered to replica 1, selected by 2AE(r). This message includes the first AE(r) add operations and the first AE(r) remove operations at replica r. The experiment concludes with the culprit read e AE . The read returns an empty set, by F LWWSet , since among the visible operations, removes are the latest in arbitration for each element.

To perform the read-back of AE(r) for replica r = 2..n, test(r) delivers all odd messages by replica r to replica 1 in the order they were sent. After delivery of each message, the test checks if replica 1 now reports the element r as part of the set. Each odd message includes one add operation that is not covered by subsequent remove at replica 1. From D LWWSet sat[V state , T any ] F LWWSet and the definition of F LWWSet , we get that exactly the first AE(r) such deliveries will have no effect on the contents of the set: they deliver add(r) operations that have been already observed by the rem(r) operation (delivered at replica 1 in the experiment phase) that dominates them in the arbitration order. Thus, if element r appears in the set right after delivering the i-th message of replica r, then AE(r) = i °1. If r does not appear by the time the loop is finished, then AE(r) = m/(2n °2).

For example, consider an experiment and test(2) in Figure 4.5. The read-back delivers the first message from replica 2 at replica 1, and reads value ;, since all visible adds are covered in arbitration by visible remove; next, it delivers the second odd (third overall) message from replica 2 and reads again, which returns value {2} this time, because the second add(2) from replica 2 is the latest operation in arbitration among all the visible operations on element 2. Therefore, the second send was the first that made element 2 appear, which indicates how many rem(2) operations from replica 2 was delivered to replica 1 during the experiment:

AE(2) = 2 °1 = 1.
Formal proofs of Lemma 4.5 and Theorem 3.7 are in Appendix B.3.

Multi-Value Register

C AE = exec(D ø , (R 0 , N 0 ), init; exp(AE))
where (R 0 , N 0 ) = (D ø .initialize, ;) 2) add( 2) rem( 2) 4.5 is an experiment family.

Definition of read-back function readback(ae) = ∏r : [2..n].result(D ø , (R init [1 7 ! ae], N init ), test(r)) readback : D ø .ß ! Q where (R init , N init ) = post(exec(D ø , (R 0 , N 0 ), init))
α(2) = 1 α(3) = 2 configuration (R init [1→R α (1)], N init ) rem(3)
r 1 : r 2 :
The experiments are similar to the add-wins set experiments. The idea is to force replica 1 to remember a number of write operations from each replica dominated by the current write, which then introduces an overhead proportional to n lg m; cf. the optimized implementation in Algorithm 3.4. All experiments start with the init phase, where each of the replicas 2..n performs (m °1)/(n °1) writes write(0) and sends a message after each write. All messages remain undelivered until the second phase, defined by exp(AE). There replica 1 delivers exactly one message from each replica r = 2..n, selected using AE(r), and subsequently overwrites the register with write( 1). An experiment concludes with the culprit read e AE on the first replica that returns the result of the last write, {1}.

The read-back works by performing separate tests for each replica r = 2..n, defined by test(r).

To determine which message was sent by replica r during the experiment, the program test(r) enforces (re)delivery of all messages sent by replica r to replica 1, in the order they were sent, and performs a read after each delivery. By inspecting the return value of each such read, the program identifies the first message that causes read to return both values 0 and 1. By F MVReg , this corresponds to observing write(0) performed at r that was not visible at the time of write(1),

i.e., that is concurrent to write(1); the index of this message corresponds to AE(r) + 1.

For example, consider an experiment and test(2) in Figure 4.6. The read-back delivers the first message from replica 2 at replica 1, and reads value {1}, since all visible write(0) are covered by write(1); next, it delivers the second message from replica 2 and reads again, which returns value {0, 1} this time, because the second write(0) from replica 2, concurrent to write(1), became visible.

Therefore, the second send was the first that made value 1 appear, which indicates how many write(0) operations from replica 2 were sent to replica 1 during the experiment:

AE(2) = 2 °1 = 1.
Formal proofs of Lemma 4.6 and Theorem 4.5 are in Appendix B.3.

Last-Writer-Wins Register

Theorem 4.6.

If D LWWReg sat[V state , T any ] F LWWReg , then the complexity of D LWWReg is b ≠(lg m).
A proof is Appendix B.3. It is a specific case of Theorem 4.4, and of minor practical importance.

Summary

Table 4.6 summarizes our positive and impossibility results for all studied data types. From left to right, the table lists prior implementations and their complexity, as well as our optimizations (if any) and their complexity, and lower bounds on complexity of any implementation. We underline every implementation optimal in the sense of Definition 3.3, i.e., with an upper bound that matches a general lower bound.

C AE = exec(D ø , (R 0 , N 0 ), init; exp(AE))
where (R 0 , N 0 ) = (D ø .initialize, ;)

Definition of read-back function readback(ae) = ∏r : [2..n].result(D ø , (R init [1 7 ! ae], N init ), test(r)) readback : D ø .ß ! Q where (R init , N init ) = post(exec(D ø , (R 0 , N 0 ), init))
Table 4.5: Experiment family (Q, n, m, C, e, readback) used in the lower bound proof for multi-value register (MVReg). For the counter data type (Ctr), the existing state-based implementation is optimal.

The set data type offers a range of different semantics choices, (AWSet, RWSet,LWWSet) which are uneven in terms of the metadata space complexity. Prior implementations all incur overhead in the order of number of updates, at least (recall that for remove-wins and LWW sets there is no upper bound, as actual elements are stored, and their size is variable). However, this is not necessary for all implementations of all variants. Our improved design of add-wins set implementation reduces this cost to the order of number of replicas, which we prove is the optimal achievable cost. Furthermore, this is asymptotically the cheapest known implementation of all three sets. This optimization is impossible for the remove-wins set, as our lower bound shows:

the order of number of updates or worse is necessary. The existing implementation of the LWW set incurs similar overhead, but it is an open problem whether this is necessary. We managed neither to improve it, nor to prove it is impossible, although we conjecture the latter.

The register type also comes in different variants (MVReg, LWWReg), which too, are uneven in the metadata complexity. The existing implementation of the LWW register has negligible overhead and is the optimal one. On the contrary, the existing implementation of the multivalue register has a substantial overhead, in the square of the number of replicas, due to inefficient treatment of concurrent writes of the same value. Our optimizations alleviate the square component, and reach the asymptotically optimal metadata overhead.

Chapter 5

Related Work and Discussion

We know no more than you. In this chapter, we discuss the scope of these results, in reference to other related work. Although a formal model of RDTs is a recent development [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF][START_REF] Shapiro | Conflict-free replicated data types[END_REF], there is already a sizeable body of work on RDTs, and some older algorithms use similar techniques.

Specifically, in Section 5.1 we discuss the completeness of our results in terms of data type coverage. In Section 5.2, we discuss known and potential optimizations to state-based implementations that are not covered by our metadata metrics. In Section 5.3, we discuss our results for state-based category compared to other approaches proposed in the literature. Finally, in Section 5.4, we compare our complexity proofs to similar ones in the field of distributed computing.

Other Data Types

Our metadata analysis cannot be exhaustive in terms of data types, although we cover many important primitive types, from which other data types can be derived by composition [START_REF] Brown | Riak DT map: A composable, convergent replicated dictionary[END_REF][START_REF] Gotsman | Composite replicated data types[END_REF].

An important data type absent in our analysis is the list. A list object is a dynamically-sized totally ordered list of elements, where an element can be inserted at or removed from a designated position [START_REF] Ellis | Concurrency control in groupware systems[END_REF][START_REF] Nédelec | LSEQ: An adaptive structure for sequences in distributed collaborative editing[END_REF][START_REF] Oster | Data consistency for P2P collaborative editing[END_REF][START_REF] Preguiça | A commutative replicated data type for cooperative editing[END_REF][START_REF] Roh | Replicated Abstract Data Types: Building blocks for collaborative applications[END_REF][START_REF] Weiss | Logoot: a scalable optimistic replication algorithm for collaborative editing on P2P networks[END_REF]. One of the most important usages of the list type is a model of a shared document in collaborative editing applications [START_REF] Ellis | Concurrency control in groupware systems[END_REF][START_REF] Oster | Data consistency for P2P collaborative editing[END_REF][START_REF] Preguiça | A commutative replicated data type for cooperative editing[END_REF][START_REF] Weiss | Logoot: a scalable optimistic replication algorithm for collaborative editing on P2P networks[END_REF].

Implementing a list is far from trivial. There is a long history of incorrect designs [START_REF] Oster | Proving correctness of transformation functions in collaborative editing systems[END_REF], and specifications are implementation-driven. Moreover, all known implementations suffer from high bounds on metadata overhead, in the order of the number of updates. Every implementation stores some trace of removed elements, in order to position concurrently inserted elements correctly. This manifests in metadata either as tombstones [START_REF] Oster | Tombstone Transformation Functions for ensuring consistency in collaborative editing systems[END_REF][START_REF] Oster | Data consistency for P2P collaborative editing[END_REF][START_REF] Roh | Replicated Abstract Data Types: Building blocks for collaborative applications[END_REF], or as position identifiers of unbounded size [START_REF] Nédelec | LSEQ: An adaptive structure for sequences in distributed collaborative editing[END_REF][START_REF] Preguiça | A commutative replicated data type for cooperative editing[END_REF][START_REF] Weiss | Logoot: a scalable optimistic replication algorithm for collaborative editing on P2P networks[END_REF]. We conjecture this is fundamental to the list semantics; the proof is a major direction of future work.

State-Based Optimizations Beyond Our Metric

Our definitions of metadata metrics (Definition 3.2) and optimality (Definition 3.3) are oriented towards worst-case asymptotic analysis and generality; so are our optimizations and lower bounds. However, there are multiple sensible ways of formulating the concepts of complexity and optimality, some of which can express optimizations that our results do not cover. In this section, we motivate our choice for the definitions, and discuss some of their properties. We identify some optimizations from the literature, and potential optimizations, not covered by our formulation, and we evaluate their significance.

STATE-BASED OPTIMIZATIONS BEYOND OUR METRIC

Background Compaction of Stable Metadata

Our definition of maximum metadata overhead (Definition 3.1) concerns the worst situation in an execution. This worst case may involve relatively short-lived peaks in metadata and only a subset of replicas. This perspective is important as it models what could happen under high concurrency or failure scenarios. For instance, the case where a set of replicas is partitioned away, or a replica becomes unresponsive. It shows the capacity that the system should be planned for.

Nevertheless, there exist RDT implementations that can compact, or specifically garbage collect (GC), metadata that is not useful anymore. In particular, stable updates [START_REF] Johnson | The maintenance of duplicate databases[END_REF][START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF] can often be collected. An update is stable if it is known to be replicated at all replicas. For example, in a set implementation, a stable tombstone can be discarded. When all replicas have received a given tombstone instance, then it is guaranteed that no further message will include the tombstone, except for a possibly delayed message. If furthermore, the implementation protects itself from delayed messages, a replica can discard the stable tombstone safely.

Johnson and Thomas [START_REF] Johnson | The maintenance of duplicate databases[END_REF] use this protocol to discard timestamps of removed entries in a LWW map, whereas Wuu and Bernstein [START_REF] Gene | Efficient solutions to the replicated log and dictionary problems[END_REF] use it for a map where elements are guaranteed to be added no more than once. A list implementation by Roh et al. [START_REF] Roh | Replicated Abstract Data Types: Building blocks for collaborative applications[END_REF] also uses stability to collect removed list elements.

A stability protocol computes stable updates that can be compacted [START_REF] Golding | Weak-consistency group communication and membership[END_REF][START_REF] Johnson | The maintenance of duplicate databases[END_REF][START_REF] Gene | Efficient solutions to the replicated log and dictionary problems[END_REF]. The stability protocol maintains information about the set of updates that each replica has received.

Such information is typically encoded as a one-dimensional vector or a two-dimensional matrix.

Stability-based metadata compaction has its own drawbacks. The data structures of the stability protocol incur an overhead on their own [START_REF] Johnson | The maintenance of duplicate databases[END_REF][START_REF] Gene | Efficient solutions to the replicated log and dictionary problems[END_REF]. It takes time until stability is detected by the protocol. Finally, stability protocol is not live in the presence of failures, since it requires to communicate with every replica. Our model assumes a static or grow-only set of replicas and does not explicitly consider their failures. However, even under a model that explicitly considers dynamic set of replicas and their failures, it requires a perfect failure detector to safely eliminate an unresponsive replica from the set of correct replicas that must acknowledge an update. 1 In contrast, with our metadata design from Section 3.2.1, an implementation discards information right away, independent of network conditions, and does not require additional GC. Nevertheless, for some data types (e.g., the remove-wins set or the list), our optimizations are not applicable and the GC approach is the only way to decrease metadata size.

Finer-Grained Optimizations

Our definition of implementation optimality (Definition 3.3) is relatively coarse-grained, in the sense that the optimal implementation is not required to incur minimum overhead in every execution. Instead, it suffices that it performs well enough in all cases, incurring an overhead that is no higher than any other implementation in the worst-case. In practice, some optimizations for cases other than worst-case are also desirable, albeit they tend to be less fundamental.

Many minor finer-grained optimizations can be considered. For example, a multi-value register does not need to store a complete vector until every replica performs at least one write.

An important exception is the recent work of Almeida et al. [START_REF] Almeida | Scalable and accurate causality tracking for eventually consistent stores[END_REF]. They improve over our optimized multi-value register implementation (Algorithm 3.4) in the case when different values are written concurrently. Their implementation uses a new dotted version vector set metadata, which consists of a single vector extended with a set of at most n scalar timestamps. The storage requirements of their register is equivalent to a single vector, regardless of the number of concurrent values stored in the state. In contrast, our optimized implementation stores one vector per value stored in the state. Both implementations are asymptotically optimal, i.e., their overhead is bounded by b £(n lg m), but the one of Almeida et al. [START_REF] Almeida | Scalable and accurate causality tracking for eventually consistent stores[END_REF] performs better for common executions. In particular, it incurs a lower size of absolute state.

As a first approach, we chose to not use a fine-grained optimality definition, suitable for such optimizations. Indeed, this could tie the definition too closely to a particular implementation or a data type. The metric would need to be type-or implementation-specific, increasing complexity and decreasing readability. Moreover, it could make optimality proofs more implementationspecific, i.e., difficult and tedious.

It is an open question whether an implementation-agnostic metric that captures such optimizations exists. With our coarse-grained definition, lower bound proofs are not specific to any particular implementation (cf. Chapter 4), and the metric is general and concise.

Custom Timestamps

Our model and bounds require that the implementations of an arbitration-based type, such as LWW register or LWW set, use externally-provided timestamps to arbitrate updates, rather than any internal mechanism. This requirement stems from the fixed definition of witness abstract execution (Equation 2.12), which is part of the implementation correctness condition. Thanks to external timestamps all implementations can support any type of timestamp and cross-object arbitration consistency properties [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF] (e.g., a consistent choice of the "last" writer among different LWW registers). However, such a lack of flexibility in timestamp choice can appear as a potential implementation restriction. Although we are not aware of any space optimizations based on the use of implementation-assigned timestamps for arbitration, our lower bound proofs are limited to implementations that do respect external timestamps. Note that all of our lower bound proofs provide only causal timestamps, so the implementations do not actually need to handle particularly adversary timestamps.

OTHER IMPLEMENTATION CATEGORIES

Other Implementation Categories

A number of recent implementations or implementation categories extend the concept of the statebased category. Extensions related to metadata optimization include message size optimizations and topology restrictions.

State-Based Implementations With Smaller Messages

Pure state-based implementations have the drawback that they transfer the complete state in each message [START_REF] Almeida | Efficient state-based CRDTs by delta-mutation[END_REF]. Recent extensions address this problem, at a modest expense in state size.

Almeida et al. [START_REF] Almeida | Efficient state-based CRDTs by delta-mutation[END_REF] propose a new implementation category: state-based with deltas. The core idea is that each replica maintains not only a complete durable object state (as in state-based), but also a smaller buffer of recent updates, called deltas, which does not need to be durable.

The expectation is that, in the common case, the system transfers only deltas, and resorts to complete state transfer only as a fallback (e.g., when the buffer has been discarded). 2 In contrast to op-based implementation, the buffer of deltas is not just a log of independent updates, but a compressed representation of updates using their semantics. A custom but type-independent protocol is used to disseminate deltas; optionally, to maintain causality. The challenge for this category of implementations is to design deltas significantly smaller than the complete state.

Almeida et al. present two examples of efficient adaptations of existing state-based imple-

mentations to deltas. For example, the delta-based counter buffers, as a delta, only the modified entries of the increments vector (cf. Algorithm 2.3); in the common case, only a small number of vector entries are transferred rather than full vector. In the absence of causal delivery, the optimized add-wins set (Algorithm 3.2) adaptation is more challenging. Recall that the optimization uses a vector to represent a set of element instances with contiguous timestamps, but delta instances might be non-contiguous. A more general representation is needed, e.g., concise version vectors [START_REF] Malkhi | Concise version vectors in WinFS[END_REF], or their interval-based representation [START_REF] Bjørner | Models and software model checking of a distributed file replication system[END_REF], which support timestamp gaps in the set.

The deltas category optimizes message size, which is a problem mostly orthogonal to the state optimization problem that we considered. In examples provided by Almeida et al. [START_REF] Almeida | Efficient state-based CRDTs by delta-mutation[END_REF], the reduced message size comes at cost of a slight increase in local storage size: a new non-durable delta buffer and bookkeeping metadata on successful deliveries are needed. These examples do not contradict our complexity bounds, which hold for the complete state and fall-back state transfer.

Deftu and Griebsch [START_REF] Deftu | A scalable conflict-free replicated set data type[END_REF] propose a somewhat similar extension, tailored specifically to the basic add-wins set (Algorithm 3.1). Their set implementation stores all element instances until discarded by a GC protocol. The GC protocol is not specified formally. Storing instances serves a purpose of naive delta implementation (a replica can request from another replica all new in-stances since a certain logical time), but increases space requirements. 3 To accommodate sets that do not fit on a single machine and/or to improve synchronization efficiency, the implementation of Deftu and Griebsch supports object sharding, which we do not consider.

Our original optimized add-wins set implementation [START_REF] Bieniusa | An optimized conflict-free replicated set[END_REF] (not presented in this thesis) com- (i) replacing version vectors with an interval-based representation [START_REF] Bjørner | Models and software model checking of a distributed file replication system[END_REF]; and by (ii) arbitrarily extending specification F AWSet to handle visibility that is not transitive. The efficiency of their implementation depends on k: the higher k, the higher metadata overhead. With k = 1, it reduces to our optimization. The practical significance of the optimization is unknown, since we are unaware of any k-causal delivery protocol.

Similarly to deltas, these designs are also orthogonal to our work, as they concern message size.

Optimizations Based on Topology Restrictions and Delayed Visibility

Our results assume a full network communication topology, where every pair of replicas communicates directly. An important group of optimizations make use of a restricted topology to reduce the metadata overhead; in particular, of variants of the client-server topology.

For example, the metadata overhead of the state-based counter is proportional to the number of vector entries, i.e., the number of replicas. This can be too much in a large or dynamic system.

To address this issue, Almeida and Baquero [START_REF] Sérgio | Scalable eventually consistent counters over unreliable networks[END_REF] propose counters optimized for a tiered topology, where replicas in the lowest tier cannot communicate directly with one another, but only via small set of replicas in a higher tier. In the simplest case, i.e., two tiers, it reduces to client and server replicas. The goal is to use a smaller number of vector entries in the average case, in the order of the number of servers, by eventually representing all client increments in these limited server entries.

The state of client and server replica is different in Almeida's counter. They share a vector of entries for server replicas. Additionally, a client replica maintains a single entry for his own increments that are not included in server entries so far. The increments from this entry are handed off to the server tier replicas by means of a fault-tolerant non-blocking two-phase protocol.

When a hand-off instance succeeds, the increments are exposed through the server tier to all replicas, without "polluting" their vectors with the client entry, and do not leave any trace on 5.3. OTHER IMPLEMENTATION CATEGORIES server replicas. However, until a hand-off instance finishes, i.e., between the first and the second round of the protocol, the server replica needs to maintain a dedicated slot for the protocol instance, to tolerate potential message duplicates from the client. In the worst-case, if the client fails after the first round, an unused slot remains opened in the server tier forever.

For such a protocol, it is useful to define metadata overhead separately for client and server replicas. The worst case for a server replica is identical to the normal state-based counter, since every client may fail after the first round of the protocol. Under favorable conditions, the common case is better. The worst case for a client replica is significantly improved, since it requires only one vector entry per server replica. Thus, servers can amortize the metadata cost for clients.

Our general lower bound proofs do not cover this protocol, due to different visibility witness.

This protocol could be characterized by a visibility witness V 6 = V state , which exposes updates only after an extra round of messages. We conjecture that the lower bound for server replicas for this category of protocols is the same as in the state-based implementation; the proof for such flexibly delayed visibility witnesses is future work.

Topology restrictions are also successfully applied to variants of op-based implementations. This is the case for our SwiftCloud system from Part III, and for a protocol of Burckhardt et al. [START_REF] Burckhardt | Cloud types: Robust abstractions for replicated shared state[END_REF], which assume multiple servers or a single server, respectively. The protocol of Burckhardt additionally allows to apply semantics-specific reduction on the log of stored or transferred operations. 4 In a previous work [START_REF] Zawirski | Asynchronous rebalancing of a replicated tree[END_REF], we also show a protocol for tree-based list implementation, where the tree can be compacted (rebalanced) at a small set of servers, independently from concurrent operations at client replicas, to optimize message size.

Replicated File Systems

A number of replicated file systems, such as Ficus [START_REF] Popek | Replication in Ficus distributed file systems[END_REF], Coda [START_REF] Kistler | Disconnected operation in the Coda file system[END_REF], or DFS-R [START_REF] Bjørner | Models and software model checking of a distributed file replication system[END_REF], rely on eventually consistent replication. Some of them use the state-based replication model, and share similarities with state-based RDT implementations.

Bjørner [START_REF] Bjørner | Models and software model checking of a distributed file replication system[END_REF] describes DFS-R, a replicated file system that ships with Microsoft products.

DFS-R can be viewed as a complex RDT object. Its state is a map from unique file identifier to file record (including file name, data etc.), and a version vector. When a file is created, it is assigned a unique id. A file can be also updated, deleted, and moved. DFS-R uses the LWW approach to resolve concurrent updates to the same file identifier, and more complex heuristics at the global file system level. When a file is deleted and created again, it uses a fresh identifier.

DFS-R implementation shares commonalities with the two add-wins set implementations.

In fact, we can see part of an DFS-R as an implementation of a token set, which supports two updates: creation of a fresh token (file identifier) and removal of an existing token. Token set is almost equivalent to the set of element instances in the add-wins implementations.

In this context, it is interesting how DFS-R treats remove operations and tombstones. Bjørner describes two approaches to removal: a standard tombstone-based and a tombstone-free. The latter relies on the knowledge in a version vector, maintained along a map of file identifiers to file records (tokens). By using timestamps as tokens, version vector can serve as a summary of all observed tokens. Our first optimization to the optimized add-wins set in Section 3.2.1.1 can be viewed as an adaptation of the tombstone-free implementation of a token set in Bjørner's file system, to build an implementation of a high level set RDT.

Lower Bound Proofs in Distributed Computing

The distributed computing community has studied the complexity of implementing different distributed or concurrent abstractions, and has developed techniques to demonstrate asymptotic lower bounds on their complexity. It is natural to compare our lower bound proofs to these works.

According to the survey of Fich and Ruppert [START_REF] Fich | Hundreds of impossibility results for distributed computing[END_REF], most of these works consider either system models or metrics significantly different from ours. Results for message passing models are usually concerned with time and message complexity required to solve a certain problem, or size of messages involved. Results for shared memory models typically evaluate the number of registers or other objects required to solve a problem. Nevertheless, the core counting argument of the lower bound proofs is often similar: the goal is to demonstrate that there is a (large) set of states that are distinguishable, because if they are not, the correctness would be affected. The structure of our experiment family, in particular the states during culprit reads in Definition 4.1, takes a similar approach.

The most relevant is the work of Charron-Bost [START_REF] Charron-Bost | Concerning the size of logical clocks in distributed systems[END_REF], who proves 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. Happens-before relation is a counterpart of the visibility relation at the level of concrete executions. Specifications of multivalue register (MVReg), sets (AWSet and RWSet), and other data types, rely crucially on visibility.

However, Charron-Bost's result does not translate into a lower bound on their implementation complexity. A specification may not require complete knowledge about the relation for all pair of events, and an implementation may not even need to store information about all events. It is often sufficient to store a fragment of this knowledge, using a semantics-optimized representation.

For example, the optimized add-wins set implementation (Algorithm 3.2) does not require to store a complete vector clock about every rem event it observed. Instead, it stores only a version vector that summarizes all known operations, and a vector for each visible element (corresponding to add event).

Part III

Causally-Consistent Object Database

for Client-Side Applications CHAPTER 6. PROBLEM OVERVIEW

In this chapter we introduce a client-side replication problem that we study and that we address in the remaining part of the thesis. The problem concerns client-side applications, such as in-browser and mobile apps, which are poorly supported by the current technology for sharing mutable data over the wide-area. App developers resort to implementing their own ad-hoc application-level cache and buffer [START_REF] Kroeger | Gmail for mobile HTML5 series: Cache pattern for offline HTML5 web applications[END_REF][START_REF] Shoemaker | Build an HTML5 offline application with Application Cache, Web Storage and ASP.NET MVC[END_REF], in order to avoid slow, costly and sometimes unavailable round-trips to a data center, but they are not in the position to solve system issues such as fault tolerance, or consistency/session guarantees [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF][START_REF] Terry | Session guarantees for weakly consistent replicated data[END_REF]. Recent client-side systems ensure only some of the desired properties, i.e., either make only limited consistency guarantees (at a granularity of a single object or a small database only), do not tolerate failures, and/or do not scale to large numbers of client devices [START_REF] Belaramani | PRACTI replication[END_REF][START_REF] Burckhardt | Bringing TouchDevelop to the cloud. Inside Microsoft Research Blog[END_REF][START_REF] Cairns | Build collaborative apps with Google Drive Realtime API. Google Apps Developers Blog[END_REF][START_REF] Chun | Mobius: Unified messaging and data serving for mobile apps[END_REF][START_REF] Mahajan | Depot: Cloud storage with minimal trust[END_REF]. Standard algorithms for geo-replication [START_REF] Almeida | ChainReaction: a causal+ consistent datastore based on Chain Replication[END_REF][START_REF] Bailis | Bolt-on causal consistency[END_REF]46,66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF] were not designed to support high numbers of client replicas outside of the server-side infrastructure.

Our thesis is that it is possible to build a scalable fault-tolerant system that takes the responsibility of providing client-side applications a local database access that is available, (causally) consistent, and convergent, i.e., that addresses the client-side replication problem.

Before we demonstrate such a system, we define the problem more precisely and highlight the challenges. In Section 6.1, we outline the system model and basic requirements. In Section 6.2, we motivate and define causal consistency model for RDT objects. In Section 6.3, we summarize the requirements with the client API. In Section 6.4, we show why the problem is nontrivial; first, by discussing prior metadata designs, and why it is difficult to apply them in the context of client-side replication; second, why partial client-side replication complicates the problem.

Hereafter, we depart from the rigorous formalism of Part II, since our focus is not anymore on the RDT model and RDT implementation metadata itself, but on the update delivery layer. We focus on log-exchange protocols supporting both op-based RDTs, with strong message delivery requirements, and consistency across objects. We believe that a less rigorous description is adequate in the context of research and system engineering challenges of log-exchange protocols.

System Model and Basic Requirements

We consider support for a variety of client-side applications sharing a database of RDT objects that the client can read and update. We aim to scale to thousands of clients, spanning the whole internet, and to a database of arbitrary size. peer-to-peer way. A DC may fail (due to disaster, power outage, WAN partition or misconfiguration, etc. [START_REF] Bailis | The network is reliable: An informal survey of real-world communications failures[END_REF][START_REF] Kansal | Using dark fiber to displace diesel generators[END_REF]) and recover with its persistent memory intact.

Clients do not communicate directly, but only via DCs. 2 Normally, a client connects to a single DC; in case of failure or roaming, to zero or more. A client may fail and recover (e.g., disconnection during a flight) or permanently (e.g., destroyed phone) without prior warning. We consider only non-byzantine failures.

Client-side apps often require high availability and responsiveness for good user experience, i.e., to be able to read and update data quickly and at all times. This can be achieved by replicating RDT objects locally, and by synchronizing updates in the background. However, a client has limited resources; therefore, it hosts a cache that contains only the small subset of the database of current interest to the local app. It should not have to receive messages relative to objects that it does not currently replicate [START_REF] Belaramani | PRACTI replication[END_REF][START_REF] Schiper | P-Store: Genuine partial replication in wide area networks[END_REF]. For similar reasons, control messages and piggy-backed metadata should have small and bounded size.

Since a client replica is only partial, there cannot be a guarantee of complete availability of all operations. The best we can expect is partial availability, whereby an operation returns without remote communication if the requested object is cached; and after retrieving the object from a remote node (DC) otherwise. If the object is not there and the network is down, the operation may be unavailable, i.e., it either blocks or returns an error.

Consistency with Convergence

Application programmers wish to observe a single, consistent view of the database. With availability and convergence requirements, consistency options are limited [START_REF] Gilbert | Brewer's conjecture and the feasibility of consistent, available, partition-tolerant web services[END_REF][START_REF] Mahajan | Consistency, availability, and convergence[END_REF]. In Section 2.1, we outlined why strongly consistent (i.e., linearizable) objects are unachievable in an available system, which is why we resort to RDTs. Similarly, strong consistency across objects, e.g., serializability [START_REF] Bernstein | Concurrency Control and Recovery in Database Systems[END_REF], is unachievable too. Nevertheless, some inter-object guarantees are achievable.

Causal Consistency

We consider support for the strongest known available and convergent consistency model: causal consistency with RDT objects [START_REF] Ahamad | Causal memory[END_REF][START_REF] Mahajan | Consistency, availability, and convergence[END_REF]. An execution H is causally consistent if every read operation in H observes all the updates that causally precede the read, applied in some linear extension of causal order.

Informally, under causal consistency, every process observes a monotonically non-decreasing set of updates, including its own updates, in an order that respects the causality between operations. 3 The following well-known example illustrates it [66]. In a social network, Bob sets permissions to disallow his boss Alice from viewing his photos. Some time later, Bob posts a questionable photo of himself. Without causal consistency, Alice may view the bad photo, delivered before the new permissions. Under causal consistency, the change of permission is guaranteed to be delivered before the post, and Alice cannot view the photo.

More generally, if an application process reads object x, and later reads object y, and if the state of x causally-depends on some update u to y, then the state of y that it reads will include update u. When the application requests y, we say there is a causal gap if the local replica has not yet delivered u. The system must detect such a gap, and wait until u is delivered before returning y, or avoid a gap in the first place. Otherwise, reads with a causal gap expose both application programmers and users to ordering anomalies [66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF].

We consider a transactional variant of causal consistency to better support multi-object operations. A transaction is a sequence of operations issued by a single client session. All the reads of a causal transaction come from a same database snapshot, and either all its updates are visible as a group, or none is [START_REF] Bailis | Scalable atomic visibility with RAMP transactions[END_REF]66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF]. Causal transactions facilitate, for instance, maintenance of secondary indices and symmetric relations [START_REF] Bailis | Scalable atomic visibility with RAMP transactions[END_REF], and object composition [START_REF] Gotsman | Composite replicated data types[END_REF].

Convergence with Replicated Data Types

Recall from Section 2.3 that convergence consists of safety and liveness components:

1. Eventual visibility / at-least-once delivery (liveness): an update that is delivered (i.e., is visible by the app) at some node, is eventually delivered to all (interested) connected nodes; for timely convergence, we require that it is delivered after a finite number of message exchanges; Causal consistency alone is not enough to guarantee confluence, as two replicas might deliver concurrent updates in different orders. Therefore, we rely on RDT objects for order-insensitive confluence. Specifically, we consider the op-based implementation category. Recall that op-based objects can use a common update delivery protocol to enforce consistency, and to share metadata, across objects. The metadata cost lies primarily in the delivery protocol. The delivery protocol can be topology-optimized and can address the problem once and for all. In contrast, state-based objects do not easily support cross-object consistency, and require type-specific optimizations.

Implementations of op-based RDTs require adequate support from the system, especially in terms of network layer guarantees (cf. discussion in Section 2.4.3). Many existing causal consistency protocols offer only LWW register support [START_REF] Almeida | ChainReaction: a causal+ consistent datastore based on Chain Replication[END_REF][START_REF] Belaramani | PRACTI replication[END_REF]46,66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF][START_REF] Mahajan | Depot: Cloud storage with minimal trust[END_REF], which is simpler than general purpose support of higher-level types such as sets or counters. For instance, as high-level RDT updates are often not idempotent (consider for instance inc in counter), safety also demands at-most-once delivery. Op-based implementations, especially optimized ones, often require causal delivery of object updates. Since an RDT object's value may be defined not just by the last update (as in the case of LWW register), but also depend on earlier updates, causal dependencies may span several different object updates, and need to be encoded and respected.

Application Programming Interface

To summarize, our client API resembles both modern object stores, such as Riak 2.0 or Redis, and prototype causal transactional stores, such as COPS or Eiger [START_REF]Data Types in Riak[END_REF]66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF][START_REF]Redis. Redis is an open source, BSD licensed, advanced key-value store[END_REF]]: 4• begin_transaction() Opens a causal transaction (at most one at a time).

• read(object_id) : v

Performs read on object object_id with return value v.

• update(object_id,method) Performs an update operation method on object_id.

• commit_transaction()

Terminates the open transaction.

Object id embeds an object type. A read can be configured to create a non-existent object. The implementation additionally supports fine control over caching, and a non-blocking API.

Challenge

Although each of discussed requirements may seem familiar or simple in isolation, the combination with scalability to high numbers of nodes and database size is a novel challenge.

Metadata Design

System metadata serves to identify individual updates, and sets of updates, to ensure correct delivery. Metadata is piggy-backed on update messages, increasing the cost of communication.

One common metadata design assigns each update a timestamp as soon as it is generated on some originating node. This is flexible and supports tracking causality at a fine grain. However, the metadata data structures tend to grow "fat." For instance, dependency lists [66] grow with the number of updates (cf. Du et al. [46], Lloyd et al. [START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF]Section 3.3]), whereas version vectors [START_REF] Belaramani | PRACTI replication[END_REF][START_REF] Mahajan | Depot: Cloud storage with minimal trust[END_REF] grow with the number of clients (indeed, our experiments show that their size becomes unreasonable). We call this the Client-Assigned, Safe but Fat approach.

An alternative delegates timestamping to a small number of DC servers [START_REF] Almeida | ChainReaction: a causal+ consistent datastore based on Chain Replication[END_REF]46,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF]. This enables the use of small vectors, at the cost of losing some parallelism. However, this is not fault tolerant if the client does not reside in a DC failure domain. For instance, it may violate at-most-once delivery. Consider a client transmitting update u to be timestamped by DC1. If it does not receive an acknowledgement, it retries, say with DC2 (fail-over). This may result in u receiving two distinct timestamps, and being delivered twice. Duplicate delivery violates safety for many op-based RDTs, or otherwise increases their space and implementation complexity [START_REF] Sérgio | Scalable eventually consistent counters over unreliable networks[END_REF][START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF][START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF]. We call this the Server-Assigned, Lean but Unsafe approach.

Clearly, neither "fat" nor "unsafe" is satisfactory.

Causal Consistency with Partial Replication is Hard

Since a partial replica receives only a subset of the updates, and hence of metadata, it could miss some causal dependencies [START_REF] Belaramani | PRACTI replication[END_REF]. Consider the following example: Alice posts a photo on her wall (update e). Bob sees the photo and mentions it in a message to Charles (update f ), who in turn mentions it to David (update g). When David looks at Alice's wall, he expects to observe update e and view the photo. However, if David's machine does not cache Charles' inbox, it cannot observe the causal chain e ! f ! g and might incorrectly deliver g without e. Metadata design should protect from such causal gaps, caused by transitive dependency over absent objects.

Failures complicate the picture even more. Suppose David sees Alice's photo, and posts a comment to Alice's wall (update h). Now a failure occurs, and David's machine fails over to a new DC. Unfortunately, the new DC has not yet received Bob's update f , on which comment h causally depends. Therefore, it cannot deliver the comment, i.e., fulfill convergence, without violating causal consistency. David cannot read new objects from the DC for the same reason. 5Finally, a DC logs an individual update for only a limited amount of time, but clients may be unavailable for unlimited periods. Suppose that David's comment h is accepted by the DC, but David's machine disconnects before receiving the acknowledgement. Much later, after h has been executed and purged away, David's machine comes back, only to retry h. This could violate at-most-once delivery; some previous systems avoid this with fat version vectors [START_REF] Belaramani | PRACTI replication[END_REF][START_REF] Mahajan | Depot: Cloud storage with minimal trust[END_REF] or with log compaction protocol that relies on client availability [START_REF] Ladin | Lazy replication: Exploiting the semantics of distributed services[END_REF].

7.1. DESIGN

x.add [START_REF]Riak distributed database[END_REF] x.add(3)

V 1 V 2 y.add(2) y.add(1) U 1 (A) Initial configuration.
x.add [START_REF]Riak distributed database[END_REF] x.add(3)

V 1 V 2
y.add(2) y.add [START_REF]Riak distributed database[END_REF] x.add(4)

U C read x {1,3} (
B) Continuation from 7.1A to risky configuration.

x.add(1)

x.add(3)

V 1 V 2 y.add (2) y.add(1)
x.add(4)

read x {1} U C
(C) Read-in-the-past: continuation from 7.1A to conservative configuration. should be avoided, because, otherwise, subsequent accesses might violate causality. On the other hand, waiting for the gap to be filled would increase latency and decrease availability. To side-step this conundrum, we adopt the approach of "reading in the past" [START_REF] Ahamad | Causal memory[END_REF]66]. Thus, a DC exposes a gapless but possibly delayed state, noted V .

To illustrate, consider the example of Figure 7.1A. Objects x and y are of type set. DC 1 is

in state U 1 that includes version V 1 µ U 1 , and DC 2 in a later state V 2 . Versions V 1 with value [x 7 ! {1}, y 7 ! {1}] and V 2 with value [x 7 ! {1, 3}, y 7 ! {1, 2}] are both gapless. However, version U 1 ,
with value [x 7 ! {1, 3}, y 7 ! {1}] has a gap, missing update y.add (2). When a client requests to read x at DC 1 in state U 1 , the DC could return the most recent version, x : {1, 3}. However, if the application later requests y, to return a safe value of y requires to wait for the missing update from DC 2 . By "reading in the past" instead, the same replica exposes the older but gapless version V 1 , reading x : {1}. Then, the second read will be satisfied immediately with y : {1}. Once the missing update is received from DC 2 , DC 1 may advance from version V 1 to V 2 .

A gapless algorithm maintains a causally-consistent, monotonically non-decreasing progression of replica states [START_REF] Ahamad | Causal memory[END_REF]. Given an update u, let us note u.deps its set of causal predecessors, called its dependency set. If a full replica, in some consistent state V , receives u, and its dependencies are satisfied, i.e., u.deps µ V , then it applies u. The new state is V 0 = V © {u}, where we note by © a log merge operator that filters out duplicates caused by failures, further discussed in Section 7.2.1. State V 0 is consistent, and monotonicity is respected, since V µ V 0 .

If the dependencies are not met, the replica buffers u until the causal gap is filled.

Causal Consistency at Partial Client Replicas

As a client replica contains only part of the database and its metadata, this complicates consistency [START_REF] Belaramani | PRACTI replication[END_REF]. E.g., encoding and respecting causal dependencies is more difficult. To avoid the complexity of the protocol and to minimize metadata size, we leverage the DC's full replicas to manage large part of gapless versions for the clients.

Given some interest set of objects the client is interested in, its initial state consists of the projection of a DC state onto the interest set. This is a causally-consistent state, as shown in the Therefore, a hypothetical full version V DC © U C would be causally consistent. Its projection is equivalent to the client state:

(V DC ©U C )| O = V DC | O ©U C | O = V C .
Liveness. This approach ensures partial availability. If a version is in the cache, it is guaranteed causally consistent, although possibly slightly stale. If it misses in the cache, the DC returns a consistent version immediately. Furthermore, the client app can write fast, because it does not wait to commit updates, but the client replica transfers them to its DC in the background.

Convergence is ensured, because (i) the client's base version is maintained up to date by the DC, in the background, and internal log is propagated to the DC; and (ii) RDTs are confluent.

Failing Over: The Issue with Transitive Causal Dependency

The approach described so far assumes that a client connects to a single DC. In fact, a client can switch to a new DC at any time, in particular in response to a failure. Although each DC's state is consistent, an update that is delivered to one is not necessarily delivered in the other (because geo-replication is asynchronous, to ensure availability and performance at the DC level [START_REF] Bailis | Coordination avoidance in database systems[END_REF]), which may create a causal gap in the client.

To illustrate the problem, return to the example of Figure 7.1A. Consider two DCs: DC 1 is in (consistent) state V 1 , and DC 2 in (consistent) state V 2 ; DC 1 does not include two recent updates of V 2 . Client C, connected to DC 2 , replicates object x only; its state is V 2 | {x} . Suppose that the client reads the set x : {1, 3}, and performs update u : x.add(4), transitioning to the configuration shown in Figure 7.1B.

In this configuration, if the client now fails over to DC 1 , and the two DCs cannot communicate, the system is not live:

(1) Reads are not available: DC 1 cannot satisfy a request for y, since the version read by the client is newer than the DC 1 version,

V 2 6 µ V 1 .
(2) Updates cannot be delivered (divergence): DC 1 cannot deliver u, due to a missing dependency:

u.deps 6 µ V 1 .
Therefore, DC 1 must reject the client to avoid creating the gap in state V 1 ©U C .1

Conservative Read: Possibly Stale, But Safe

To avoid such gaps that cannot be satisfied, we use an approach similar to Mahajan et al. [START_REF] Mahajan | Depot: Cloud storage with minimal trust[END_REF],

and depend on updates that are likely to be present in the fail-over DC, called K-stable updates.

A version V is K-stable if every one of its updates is replicated in at least K DCs, i.e.,

|{i 2 DC | V µ V i }| ∏ K,
where K ∏ 1 is a threshold configured w.r.t. expected failure model, and DC is a set of all data centers. To this effect, in our system every DC maintains a consistent K-stable version V K i µ V i , which contains the updates for which DC i has received acknowledgements from at least K °1 distinct other DCs.

A client's base version must be

K-stable, i.e., V C = V K i | O © U C | O ,
to support failover. In this way, the client depends, either on external updates that are likely to be found in any DC (V K i ), or internal ones, which the client can always transfer to the new DC (U C ).

To illustrate, let us return to Figure 7.1A, and consider the conservative progression to configuration in Figure 7.1C, assuming K = 2. The client's read of x returns the 2-stable version {1}, avoiding the dangerous transitive dependency via an update on y. If DC 2 is unavailable, the client can fail over to DC 1 , reading y and propagating its update remain both live.

Correctness with K-stability. By the same arguments as in Section 7.1.2, a DC version V K i is causally consistent and monotonically non-decreasing, and hence the client's version as well.

Note that a client observes his internal updates immediately, even if not K-stable. Our approach is flexible w.r.t. which consistent version the DC offers to the client as a base version.

Parameter K can be adjusted dynamically. Decreasing it has immediate effect without impacting correctness. Increasing K has effect only for future updates, to preserve montonicity.

Discussion

The source of the problem with the basic algorithm from Section 7.1.2 is an indirect causal dependency on an update that the two replicas do not both know about (y.add(2) in our example).

As this is an inherent issue, we conjecture a general impossibility result, stating that genuine partial replication, causal consistency, partial availability and timely at-least-once delivery (convergence) are incompatible. Accordingly, some requirements must be relaxed.

Note that in many previous systems, this impossibility translates to a trade-off between consistency and availability on the one hand, and response time and throughput on the other [START_REF] Corbett | Spanner: Google's globally-distributed database[END_REF]66,[START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF]. By "reading in the past," we displace this to a trade-off between freshness and availability, controlled by adjusting K. A higher K increases availability, but updates take longer to be delivered;2 in the limit, K = N ensures complete availability, but no client can deliver a new update when some DC is unavailable. A lower K improves freshness, but increases the probability that a client will not be able to fail over, and that it will block until its original DC recovers. In the limit, K = 1 is identical to the basic protocol from Section 7.1.2, and is similar to previous blocking session-guarantee protocols [START_REF] Terry | Session guarantees for weakly consistent replicated data[END_REF]. K = 2 is a good compromise for deployments with three or more DCs that covers common scenarios of a DC failure or disconnection [START_REF] Corbett | Spanner: Google's globally-distributed database[END_REF][START_REF] Kansal | Using dark fiber to displace diesel generators[END_REF]. Our evaluation with K = 2 shows that it incurs a negligible staleness.

Network partitions.

Client failover between DCs is safe and generally live, except in the unlikely case when the original set of K DCs were partitioned away from both other DCs and the client, shortly after they delivered a version to the client. In this case, the client blocks. To side-step this unavoidable possibility, we provide an unsafe API to read inconsistent data.

When a set of fewer than K DCs is partitioned from other DCs, the clients that connect to them are available and safe, but do not deliver their mutual updates until the partition heals.

To improve liveness in this scenario, SwiftCloud supports two heuristics: (i) a partitioned DC announces its "isolated" status, automatically recommending clients to use another DC, and (ii) clients who cannot reach another DC that would satisfy their dependencies can use the isolated DCs with K temporarily lowered, risking unavailability if another DC fails.

Precision vs. missing dependencies. The probability of a client blocked due to an unsatisfied transitive causal dependency depends on many factors, such as workload-and deployment- This may cause a spurious missing dependency, when a DC rejects a client because it misses some update that is not an actual dependence. Finer-grained dependency representation, such as in causality graphs [66], or resorting to application-provided explicit dependencies under a weaker variant of causal consistency [START_REF] Bailis | Bolt-on causal consistency[END_REF], avoid some spurious dependencies at the expense of fatter metadata. However, the missing dependency issue remains under any dependency precision. Thus, our approach is to fundamentally minimize the chances of any missing dependency, both genuine and spurious.

Implementation

We now describe a metadata and concrete protocols implementing the abstract design.

Timestamps, Vectors and Log Merge

The SwiftCloud approach requires metadata: (1) to uniquely identify an update; (2) to encode its causal dependencies; (3) to identify and compare versions; (4) and to identify all the updates of a transaction. We now describe a type of metadata, which fulfills the requirements and has a low cost. It combines the strengths of the two approaches outlined in Section 6.4.1, and is both lean and safe Recall that a logical timestamp is a pair (i, k) 2 (DC [ C) £ N, where i identifies the node that assigned the timestamp (either a DC or a client) and k is a sequence number. Similarly to the solution of Ladin et al. [START_REF] Ladin | Lazy replication: Exploiting the semantics of distributed services[END_REF], our metadata assigned to some update u combines both:

(i) a single client-assigned timestamp u.t C that uniquely identifies the update, and (ii) a set of zero or more DC-assigned timestamps u.T DC . Before being delivered to a DC, the update has no DC timestamp; it has one thereafter; it may have more than one in case of delivery to multiple DCs (on failover). The updates in a transaction all have the same timestamp(s), to ensure all-or-nothing delivery [START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF]. Our approach provides the flexibility to refer to an update via any of its timestamps, which is handy during failover.

We represent a version or a dependency as a version vector [START_REF] Parker | Detection of mutual inconsistency in distributed systems[END_REF], introduced in Section 2.2.3. Recall that a version vector is a partial map from node id to integer, e.g., vv = [DC 1 7 ! 1, DC 2 7 ! 2], which we interpret as a set of timestamps. For example, when vv is used as a dependency for some update u, it means that u causally depends on a set of timestamps TS(vv) = {(DC 1 , 1), (DC 2 , 1), (DC 2 , 2)} (TS is defined in Equation 3.1). In SwiftCloud protocols, every vector has at most one client entry, and multiple DC entries; thus, its size is bounded by the number of DCs, limiting network overhead. In contrast to a dependence graph [66], a vector compactly represents transitive dependencies and can be evaluated locally by any node.

A version decoding function V of vector vv on a state U selects every update in state U that matches the vector (it is defined for states U that cover all timestamps of vv):

V(vv,U) = {u 2 U | (u.T DC [ {u.t C }) \ TS(vv) 6 = ;}
For the purpose of the decoding function V, a given update can be referred equivalently through any of its timestamps, which we leverage for failover. Moreover, V is stable with growing state U, which is handy to identify a version in a (large) state that undergoes concurrent log appends.

The log merge operator U 1 © U 2 , which eliminates duplicates, is defined using client timestamps. Two updates

u 1 2 U 1 , u 2 2 U 2 are identical if u 1 .t C = u 2 .t C . The merge operator merges their DC timestamps into u 2 U 1 ©U 2 , such that u.T DC = u 1 .T DC [ u 2 .T DC .

Protocols

We are now in the position to describe the concrete protocols of SwiftCloud, by following the lifetime of an update, and with reference to the names in Figure 6.1.

State

A DC replica maintains its state U DC in durable storage. The state respects causality and atomicity for each individual object, but due to internal concurrency, this may not be true across objects. Therefore, the DC also has a vector vv DC that identifies a safe, monotonically nondecreasing causal version in the local state, which we note V DC = V(vv DC ,U DC ). Initially, U DC is an empty log of updates, and vv DC is zeroed.

A client replica stores the commit log of its own updates U C , and the projection of the base version from the DC, restricted to its interest set O, V DC | O , as described previously in Section 7.1.2.

It also stores a copy of vector vv DC that describes the base version.

Client-Side Execution

When the application starts a transaction ø at client C, the client replica initializes it with an empty buffer of updates ø.U = ; and a snapshot vector of the current base version ø.vvDeps = vv DC ; the base version can be updated by the notification protocol concurrently with the transaction execution. A read in transaction ø is answered from the version identified by the snapshot vector, merged with recent internal updates,

ø.V = V(ø.vvDeps,V DC | O ) ©U C | O © ø.U. If
the requested object is not in the client's interest set, x 6 2 O, the clients extends its interest set, and returns the value once the DC updates the base version projection.

When the application issues internal update u, it is appended to the transaction buffer ø.U √ ø.U © {u}, and included in any later read. To simplify the notation, and without loss of generality, we assume hereafter that a transaction performs at most one update. 3 The transaction

Transfer Protocol: Client to Data Center

The transfer protocol transmits committed updates from a client to its current DC, in the background. It repeatedly picks the first unacknowledged committed update u from the log. If any of u's internal dependencies has recently been assigned a DC timestamp, it merges this timestamp into the dependency vector, u.vvDeps. Then, the client sends a copy of u to its current DC. The client expects to receive an acknowledgement from the DC, containing the timestamp(s)

T that the DC assigned to update u. If so, the client records the DC timestamp(s) in the original update record u.T DC √ T. T is a singleton set, unless failover was involved.

The client may now iterate with the next update in the log.

A transfer request may fail for three reasons: A. Timeout: the DC is suspected unavailable; the client connects to another DC (failover) and repeats the protocol.

B. The DC reports a missing internal dependency, i.e., it has not received some update of the client, as a result of a previous failover. The client recovers by marking as unacknowledged all internal updates starting from the oldest missing dependency, and restarting the transfer protocol from that point.

C. The DC reports a missing external dependency; this is also an effect of failover. In this case, the client tries yet another DC. The approach from Section 7.1.3.1 avoids repeated failures.

Upon receiving update u, the DC verifies if its dependencies are satisfied, i.e., if TS(u.vvDeps) µ

TS(vv DC ). (If this check fails, it reports an error to the client, indicating either case (B) or (C)).

If the DC has not received this update previously, as determined by the client-assigned timestamp, i.e., 8u 0 2 U DC : u 0 .t C 6 = u.t C , the DC does the following: (1) assign it a DC timestamp u.T DC √ {(DC, vv DC (DC) + 1)}, (2) store it in its durable state U DC © {u}, (3) make the update visible in the DC version V DC , by incorporating its timestamp(s) into vv DC . This last step makes u available to the geo-replication and notification protocols, described hereafter. If the update has been received before, the DC only looks up its previously-assigned DC timestamps, u.T DC . In either case, the DC acknowledges the transfer to the client with the DC timestamp(s). Note that some of these steps can be parallelized between transfer requests received from different client replicas, e.g., using batching for timestamp assignment.

Geo-replication Protocol: Data Center to Data Center

The geo-replication protocol relies on a uniform reliable broadcast across DCs [START_REF] Cachin | Introduction to Reliable and Secure Distributed Programming[END_REF]. An update enters the geo-replication protocol when a DC accepts a fresh update during the transfer protocol.

The accepting DC broadcasts it to all other DCs. The broadcast implementation stores an update in a replication log until every DC receives it. A DC that receives a broadcast message containing u does the following: (1) if the dependencies of u are not met, i.e., if TS(u.vvDeps) 6 µ TS(vv DC ), buffer it until they are; and (2) incorporate u into durable state U DC © {u} (if u is not fresh, the duplicate-resilient log merge © safely unions all timestamps), and incorporate its timestamp(s) into the DC version vector vv DC . This last step makes it available to the notification protocol. The K-stable version V K DC , and a corresponding vector vv K DC , are maintained similarly.

Notification Protocol: Data Center to Client

A DC maintains a best-effort notification session, over a FIFO channel, to each of its connected clients. The soft state of a session at a DC includes a copy of the client's interest set O and the last known base version vector used by the client, vv 0 DC . Both of them can be recovered from the client when necessary, and the client provides both to initiate a session. The DC accepts a new session only if its own state is consistent with the base version of the client, i.e., if TS(vv 0 DC ) µ TS(vv DC ).

Otherwise, the DC would cause a causal gap with the client's state; in this case, the client is redirected to another DC (the solution from Section 7.1.3.1 avoids repeated rejections).

The DC sends over each channel a causal stream of update notifications. 4 Notifications are batched according to either time or to rate [START_REF] Belaramani | PRACTI replication[END_REF]. A notification packet consists of a new base version vector vv DC , and a log of all the updates U ± to the objects of the interest set, between the client's previous base vector vv 0 DC and the new one. Formally:

(7.1)

U ± = {u 2 U DC | O | u.T DC \ (TS(vv DC ) \ TS(vv 0 DC )) 6 = ;}.
The client applies the newly-received updates to its local state, described by the old base version, When the client detects a broken channel, it reinitiates the session, possibly on a new DC.

V DC | O √ V DC | O ©U ± ,
The interest set can change dynamically. When an object is evicted from the cache, the notifications are lazily unsubscribed to save resources. When it is extended with object x, the DC responds with the current version of x, which includes all updates to x up to the base version vector. To avoid races, a notification includes a hash of the interest set, which the client checks.

Object Checkpoints and Log Pruning

The log-based representation of database state contributes to substantial storage and, to smaller extent, network costs. This is a reminiscent of the similar problem at the object level of statebased RDT implementations from Part II, which manifests here in the log-exchange protocol for op-based implementations.

To avoid unbounded growth of the log, a background pruning protocol replaces the prefix of the log by a checkpoint. In the common case, a checkpoint is more compact than the corresponding log of updates. For instance, a log containing one thousand increments to an op-based counter object from Algorithm 2.2, including system timestamps, can be replaced by a checkpoint containing just the number 1000, and system metadata, a version vector.

Log Pruning in the Data Center

The log at a DC provides (i) unique timestamp identification of each update, which serves to filter out duplicates by © operator, to ensure reliable causal delivery, as explained earlier, and

(ii) the capability to compute different versions, for application processes reading at different causal times. Conversely, for the protocol described so far, an update u is expendable once all of its duplicates have been filtered out, and once u has been delivered to all interested application processes. However, evaluating expendability precisely would require access to the client replica.

In practice, we need to prune aggressively, but still avoid the above issues, as we explain next.

In order to reduce the risk of pruning a version not yet delivered to an interested application, we prune only a delayed version vv ¢ DC , where ¢ is a real-time delay [66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF]. If this heuristic fails, the consequences are not fatal: an ongoing client transaction may need to restart and repeat prior reads or return inconsistent data if desired, but the committed updates are never aborted.

To avoid duplicates, we extend our DC local metadata as follows. DC i maintains an at-mostonce guard G i , which records the sequence number of each client's last pruned update G i : C ! N.

Whenever the DC receives a transfer request or a geo-replication message for update u with timestamp (C, k) = u.t C and cannot find it in its log, it checks the at-most-once guard entry G i (C) whether u is contained in the checkpoint. If the update was already pruned away (G i (C) ∏ k), it is recognized as a pruned duplicate. In this case, the DC neither assigns the update a new timestamp (as it normally does in geo-replication protocol), nor stores it in the durable statate, nor delegates the actual update to the notification protocol (as in transfer and geo-replication protocols), but it updates its version vector to include all existing timestamps of the update; in transfer reply, the DC overapproximates u's timestamps by vector vv DC , since the information about the exact set of update's DC timestamps is discarded.

The notification protocol also uses checkpoints. On a client cache miss, the DC does not send updates log for an object, but the object state that consists of the most recent checkpoint of the object and the client's guard entry, so that the client can merge it with his updates safely.

Note that a guard is local to and shared by all objects at a DC. It is never fully transmitted.

EXPERIMENTAL SETUP

with single-object reads and writes. We use two of the standard YCSB workloads: update-heavy Workload A, and read-dominated Workload B. The object access popularity can be set to either uniform or Zipfian. YCSB does not rely on transactional semantics or high-level RDTs.

SocialApp is a social network application modelled closely after WaltSocial [START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF]. 1 It employs high-level data types such as sets, for friends and posts, LWW register for profile information, counter for counting profile visits, and inter-object references (implemented at the application level). Many SocialApp objects grow in size over time (e.g., sets of posts). 2 SocialApp accesses multiple objects in a causal transaction to ensure that operations such as reading a wall page and profile information behave consistently. Percentage in parentheses next to the operation type in Table 8.1 indicate the frequency of each operation in the workload. The SocialApp workload is read-dominated, but the ostensibly read-only operation of visiting a wall can actually increment the wall visit counter when statistics are enabled, in metadata experiments. The user popularity distribution is uniform.

We use a 50,000-user database for both applications, except for a smaller 10,000-user database for metadata experiments, to increase the stability of measurements.

In order to model the locality behavior of a client, which is a condition to benefit from any kind of client-side replication, both YCSB and SocialApp workloads are augmented with a facility to control access locality, mimicking social network access patterns [START_REF] Benevenuto | Characterizing user behavior in online social networks[END_REF]. Within a client session, a workload generator draws uniformly from a pool of session-specific objects with either 40% (low locality) or 80% (high locality) probability. For SocialApp, the pool contains objects of user's friends. Objects not drawn from this local pool are drawn from the global (uniform or Zipfian) distribution described above. The size of the local pool is smaller than the size of cache.

Experimental Setup

We run three DCs in geographically distributed Amazon EC2 availability zones (Europe, Virginia, and Oregon), and a pool of distributed clients, with the following Round-Trip Times (RTTs): B. Safe But Fat metadata mode with cache, but with client-assigned metadata only, modeled after PRACTI, or Depot without cryptography [START_REF] Belaramani | PRACTI replication[END_REF][START_REF] Mahajan | Depot: Cloud storage with minimal trust[END_REF]; C. Server-side replication mode without client caches. In this mode, a read incurs one RTT to a DC, whereas an update incurs two RTTs, modelling the cost of a synchronous writes to a quorum of servers to ensure fault-tolerance comparable to SwiftCloud.

Experimental Results

Response Time and Throughput

We run several experiments to evaluate SwiftCloud's client-side caching, with reference to the caching locality potential and geo-replication on server-side only. For each workload we evaluate the system stimulated with different rates of aggregated incoming transactions, until it becomes saturated. We use a number of clients that is throughput-optimized for each workload and protocol mode. We report aggregated statistics for all clients.

Response time. Figure 8.1 shows response times for YCSB, comparing server-side (left side) with client-side replication (right side), under low (top) and high locality (bottom), when the system is not overloaded. Recall that in server-side replication, a read incurs a RTT to the DC, whereas a fault-tolerant update incurs 2 RTTs. We expect SwiftCloud to provide much faster response, at least for cached data. Indeed, the figure shows that a significant fraction of operations respond immediately in SwiftCloud mode, and this fraction tracks the locality of the workload (marked "locality potential" on the figure), within a ±7.5 percentage-point margin. 3 The remaining operations require one round-trip to the DC, indicated as 1 RTT.

These results demonstrate that the consistency guarantees and the rich programming interface of SwiftCloud do not affect responsiveness of cached read and updates.

Our measurements for SocialApp show the same message, we do not report them here. An important lesson from a preliminary evaluation of SocialApp (not plotted) was that transactions involving reads on more than one object may incur multiple RTTs, compared to 1 RTT for server-side replication. We optimized our implementation to use predeclared objects and parallel reads to bring the response time down to 1 RTT. However, some transactions involving several predefined objects, yet producing small return value, could benefit from optional server-side execution support.

Maximum throughput.

In the next study, we saturate the system to determine its maximum aggregated throughput. Figure 8 As expected, the lower (cached) percentile consistently outperforms the server-side baseline, for all workloads and transaction rates. A separate analysis, not reported in detail here, reveals that a saturated DC slows down its rate of notifications, increasing staleness, but this does not impact response time, as desired. In contrast, the higher percentile follows the trend of server-side replication response time, increasing remote access time with DC load.

Varying the target notification rate (not plotted) between 500 ms and 1000 ms, reveals the same trend: response time is not affected by the increased staleness. At a longer refresh rate, notification batches are less frequent but larger. This increases throughput for the update-heavy YCSB A (up to tens of percent points), but has no effect on the throughput of read-heavy YCSB B. We expect the impact of refresh rate to be amplified for workloads with very low rate of updates.

EXPERIMENTAL RESULTS

•

• • • • • • • • • • • • • • • • • / limit • • • • • • • • • • • • • • • • • • / limit • • • • • • • • • • • • • • • • • • • • YCSB A,

Scalability

Next, we measure how well SwiftCloud scales with increasing numbers of DC and of client replicas. Of course, performance is expected to increase with more DCs, but most importantly, the size of metadata should be small, should increase only marginally with the number of DCs, and should not depend on the number of clients. Our results support these expectations.

In this experiment, we run SwiftCloud with a variable number of client (500-2500) and DC (1-3) replicas. We report only on the uniform object distribution, because under the Zipfian distribution different numbers of clients skew the load differently, making any comparison meaningless. To control staleness, we run SwiftCloud with two notification rates: 1 s and 10 s. The capacity of a single DC in our hardware configuration peaks at 2,000 active client replicas for YCSB, and 2,500 for SocialApp. Beyond that, the DC drops excessive number of packets.

As to be expected, additional DC replicas increase the system capacity for operations that can be performed at only one replica such as read operations or sending notification messages.

Whereas a single SwiftCloud DC supports at most 2,000 clients, with three DCs SwiftCloud supports at least 2,500 clients for all workloads. Unfortunately, as we ran out of resources for client machines at this point, we cannot report an upper bound.

For some fixed number of DCs, adding client replicas increases the aggregated system throughput, until a point of approximately 300-500 clients per DC, where the cost of maintaining client replicas up to date saturates the DCs, and further clients do not absorb enough reads to overcome these costs. Note that the lower refresh rate can reduce the load at a DC by 5 to 15%.

Metadata. In the same experiment, Figure 8.5 presents the distribution of metadata size notification messages. (Notifications are the most common and the most costly network messages.)

We plot the size of metadata (in bytes) on the Y axis, varying the number of clients along the X axis. Left to right, the same workloads as in the previous figure. Thin lines are for one DC, thick lines for three DCs. A solid line represents SwiftCloud "Lean and Safe" metadata, and dotted lines the classical "Safe but Fat" approach. Note that our Safe-but-Fat implementation includes the optimization of sending vector deltas rather than the full vector [START_REF] Mahajan | Depot: Cloud storage with minimal trust[END_REF]. Vertical bars represent standard error across clients. As notifications are batched, we normalize metadata size to a message carrying exactly 10 updates, corresponding to under approx. 1 KB of data.

The experiment confirms that the SwiftCloud metadata is small and constant, at 100-150 bytes/notification (10-15 bytes per update); data plus metadata together fit inside a single standard network packet. It is independent both from the number of client replicas and from the workload; and from the number of objects in the database, as an additional experiment (not plotted) validates. Increasing the number of DC replicas from one to three causes a negligible increase in metadata size, of under 10 bytes. We attribute some variability across clients (bars) to the data encoding and inaccuracies of measurements, including the normalization process.

In contrast, the classical Safe but Fat metadata grows linearly with the number of clients and exhibits higher variability. Its size reaches approx. 1 KB for 1,000 clients in all workloads, and 10 KB for 2,500 clients. Clearly, metadata being up to 10£ larger than the actual data represents a substantial overhead.

EXPERIMENTAL RESULTS

YCSB -all objects

SocialApp -stats counters

• • • • • • • • • • • • • • • • • • • • 1K 10K 100K 1M

Tolerating Client Churn

We now turn to fault tolerance. In the next experiment, we evaluate SwiftCloud under client churn, by periodically disconnecting client replicas and replacing them with a new set of clients.

At any point in time, there are 500 active clients and a variable number of disconnected clients, up to 5000. Figure 8.6 illustrates the storage occupation of a DC for representative workloads, which is also a proxy for the size of object checkpoints transferred. We compare SwiftCloud's log compaction to a protocol without at-most-once delivery guarantees (Lean But Unsafe). SwiftCloud storage size is approximately constant. This is achievable safely thanks to the at-most-once guard table per DC. Although the size of the guard (bottom curve) grows with the number of clients, it requires orders of less storage than the actual database itself.

A protocol without at-most-once delivery guarantees can use Lean-but-Unsafe metadata, without SwiftCloud's at-most-once guard. However this requires higher complexity in each RDT implementation, to protect itself from duplicates (cf. state-based implementations from Part II). This increases the size of objects, impacting both storage and network costs. As is visible in the figure, the cost depends on the object type: none for YCSB's LWW map, which is naturally idempotent, vs. linear in the number of clients for SocialApp's Counter objects.

We conclude that the cost of maintaining SwiftCloud's shared at-most-once guard is negligible, and easily amortized by the possible savings and the stable behavior it provides.

Tolerating Data Center Failures

The next experiment studies the behavior of SwiftCloud when a DC disconnects. The scatterplot in Figure 8 Recall that a standard server-side replication system with similar fault-tolerance incurs high response time (cf. Section 8.3.1, or Corbett et al. [START_REF] Corbett | Spanner: Google's globally-distributed database[END_REF]) and does not ensure at-most-once delivery.

Staleness Cost

The price to pay for our read-in-the-past approach is an increase in staleness. Our next experiment illustrates the impact of K-stability for evaluated workloads. A read incurs staleness overhead if a version more recent (but not K-stable) than the one it returns exists at the current DC of a client that performed the read. A transaction is stale if any of its reads is stale. In the experiments so far, we observed a negligible number of stale reads. The reason is that the window of vulnerability (the time it takes for an update to become K-stable) is small: approximately the RTT to the closest DC. For this experiment, we artificially increase the probability of staleness by various means: using a smaller database, setting cache size to zero, and transferring to the farthest DC, with a RTT of around 170 ms. We run the SocialApp benchmark with 1000 clients in Europe connected to the Ireland DC and replicated in the Oregon DC. Figure 8.8 shows that stale reads and stale transactions remain under 1% and 2.5% respectively. This number decreases as we increment the size of the database. This shows that even under high contention, accessing a slightly stale snapshot has very little impact on the data read by transactions.

Chapter 9

Related Work

If somebody wants a sheep, that is a proof that one exists.

The Little Prince by Antoine de Saint-Exupéry In this chapter, we categorize and present the work related to SwiftCloud, and compare to it.

We discuss relevant theory, in Section 9.1, and relevant systems and algorithms, in Section 9.2.

Consistency Models for High Availability

Causal consistency. Ahamad et al. [4] propose the original causal consistency model for shared memory, with highly available read and writes. Their definition concerns only updates ordering, and does not embody the confluence requirement, i.e., replicas may not converge in the presence of concurrent updates (although it is rarely the case for practical implementations).

More recently, Lloyd et al. [66] and Mahajan et al. [START_REF] Mahajan | Consistency, availability, and convergence[END_REF] independently strengthen the original definition with the confluence requirement. This model is sometimes called causal+ consistency [START_REF] Almeida | ChainReaction: a causal+ consistent datastore based on Chain Replication[END_REF]66]. Burckhardt et al. [START_REF] Burckhardt | Understanding eventual consistency[END_REF] provide a formal and abstract treatment of these definitions, by explicitly modelling confluence with RDTs; it is a generalization of the model we use in Chapter 2 to the multi-object setting.

Trade-offs. Mahajan et al. [START_REF] Mahajan | Consistency, availability, and convergence[END_REF] prove that no stronger consistency model than causal consistency is available and convergent, in full replication system. They do not consider partial replication. We conjecture, in Section 7.1.3.2, that these properties are not simultaneously achievable under partial replication, and show how to weaken one of them. Bailis et al. [START_REF] Bailis | Highly Available Transactions: Virtues and limitations[END_REF] also study liveness and limits of a variety of weak consistency models, including causal consistency and weak transactional models. They give an argument for a similar impossibility of highly available causal consistency for a client switching server replicas, but do not take into account the capabilities of a client replica, as we do in our solution.

Transactions. Lloyd et al. propose two extensions to causal consistency that facilitate multiobject programming: read-only and write-only transactions [66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF] (Burckhardt et al. [START_REF] Burckhardt | Eventually consistent transactions[END_REF] also propose a similar model). A read-only transaction reads from a consistent snapshot [66]. A writeonly transaction, proposed independently of our work, ensures atomic visibility of all updates in a transaction [START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF]. The transactions of Lloyd are non-interactive, i.e., the application needs to predefine the set of accessed objects. SwiftCloud supports interactive read-update transactions.

Atomic visibility transactions can be considered independently from causal consistency. Bailis et al. [START_REF] Bailis | Scalable atomic visibility with RAMP transactions[END_REF] argue that atomic transactions without causality incur lower cost and are sufficiently strong. SwiftCloud, similarly to non-interactive transactional systems like Eiger [START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF], supports transactions that are both atomic and causal, interactively.

Relevant Systems

In this section, we overview relevant replication systems and algorithms that support consistent, available and convergent data access, at different scales, scopes, and to different extent.

RELEVANT SYSTEMS

For directly comparable causally-consistent partial replication systems, we evaluate their metadata in Table 9.1. The columns indicate: (i) the extent of support for partial replication;

(ii) which nodes assign timestamps; (iii) the worst-case size of causality metadata; (iv) the scope of validity of metadata representing a database version (is it valid only in a local replica or anywhere); (v) whether it ensures at-most-once delivery; (vi) whether it supports general RDTs.

Replicated Databases for Client-Side Applications

Replication systems for client-side applications have been previously proposed in the literature and some systems have been deployed in production. Some of them use client-side replicas.

Systems that Support Inter-Object Consistency

The most relevant systems are the ones that provide causal consistency support for a large database with many objects. [START_REF] Belaramani | PRACTI replication[END_REF] is a seminal work on causal consistency under partial replication. PRACTI uses Safe-but-Fat client-assigned metadata and an ingenious log-exchange protocol that supports an arbitrary communication topology. It implements flexible replication options including lazy checkpoint transfer, object prefetching etc., absent in SwiftCloud implementation. While the generality of PRACTI has advantages, it is not viable for a large-scale client-side replication deployment: (i) its fat metadata approach (version vectors sized as the number of clients) is prohibitively expensive (see Figure 8.5), and (ii) any replica can make another one unavailable, because of the indirect dependence issue discussed in Section 7.1.3.2. Our cloud-backed support of client-side replication addresses these issues at the cost of lower flexibility in communication topology. We are considering support for a limited form of peer-to-peer communication that would not cause these issues, e.g., between devices of a same user or a local group of collaborators. Depot. Our high availability techniques are similar to those of Depot [START_REF] Mahajan | Depot: Cloud storage with minimal trust[END_REF]. Depot is a causallyconsistent system, which provides client-side applications a reliable storage on top of untrusted cloud replicas. In the normal mode of operation, Depot replicates only metadata at the clients. To tolerate Byzantine cloud behavior, Depot additionally uses cryptographic metadata signatures, in order to detect misbehavior, as well as fat metadata, in order to support direct client-to-client communication. Furthermore, it either exposes updates signed by K different servers or forces clients to receive all transitive causal dependencies of their reads. This conservative approach is at odds with genuine partial replication at the client side. Although Depot does not replicate all data to the client, it does require that every client processes the metadata of every update, and puts the burden of computing a K-stable version on the client. In the case of extensive failures and DC partitions, it additionally floods all updates to the client, not only the fat metadata. In contrast, SwiftCloud relies on DCs to compute K-stable and consistent versions with lean metadata. In 9.1: Qualitative comparison of metadata used by causally consistent systems with some support for partial replication. "Timestamp assignment" indicates which nodes assign timestamps. "Causality metadata" indicates the type and maximum cardinality of entries in causality metadata, given as the expected order of magnitude (e.g., we expect hundreds to thousands servers). "Version metadata" indicates if a metadata to represent a consistent version is valid only in a local replica (DC), or globally; for the latter, if it is fault-tolerant or not. "∑ 1 delivery" indicates at-most-once delivery support. 9.2. RELEVANT SYSTEMS the event of an extensive failure that cannot be handled by K-stability, SwiftCloud provides the flexibility to decrease K dynamically or to weaken consistency. Depot tolerates Byzantine clients by refusing misbehavior, or by representing late-detected misbehavior as concurrent updates.

PRACTI. PRACTI

Our current implementation does not address Byzantine failures.

Both PRACTI and Depot use Safe-but-Fat metadata, as indicated in Table 9.1. They support only LWW registers, but their rich metadata could conceivably accommodate other RDTs too.

Systems that Support Intra-Object Consistency Only

Some other systems ensure consistency only for individual objects, or for a small fully-replicated database, accessed from client-side applications.

Lazy Replication. Lazy Replication technique of Ladin et al. [START_REF] Ladin | Lazy replication: Exploiting the semantics of distributed services[END_REF] includes causal consistency protocols for client-side application accessing server replicas, which share many similarities with SwiftCloud. In Lazy Replication, every server node hosts a full replica of the database.

There is no client-side replica; server replicas execute operations on behalf of the clients. A client operation issues query or update requests to one or more server replicas; update requests are asynchronously replicated across servers. To implement this scheme, Ladin et al. propose safe and lean metadata similar to the metadata we employed in SwiftCloud. It involves client-assigned timestamps to ensure at-most-once execution of client operations and server-assigned (DCassigned) timestamps to encode operation dependencies. To achieve high availability, similarly to SwiftCloud, (i) an update can reach multiple servers that can assign it multiple timestamps safely; and moreover (ii) in a later work, Ladin et al. [START_REF] Ladin | Providing high availability using lazy replication[END_REF] also suggest reading only stable updates, but that forces Lazy Replication clients to execute them synchronously.

Our work differs in a number of aspects. Lazy Replication considers only global operations, at the scope of the entire database, with possibly complex conflict resolution. Instead, we structure the database into smaller RDT objects, making them the elementary replication and confluence unit, and provide causal transactions across RDT objects to implement global operations, including reads from a snapshot and atomic updates. Primarily, this allows SwiftCloud to offer partial client replicas that host only a subset of objects. We demonstrate how to leverage such client replicas to offer causal consistency with increased availability and responsiveness on cached objects (in contrast, Lazy Replication performs all operations on server replicas), and how to combine local updates with K-stable updates into a consistent view (without resorting to slow and unavailable synchronous updates of Lazy Replication). Moreover, our database structure translates into more structured implementation and provides natural parallelism potential, enabling multi-versioning and sharding techniques (cf. related work in Section 9.2.2).

Ensuring at-most-once execution incurs some cost, both in Lazy Replication and in SwiftCloud.

Lazy Replication uses client-driven log compaction protocol that relies on replicas availability and loosely-synchronized clocks for progress. In contrast, SwiftCloud relies on at-most-once guard table and prunes the log aggressively. Our solution is optimized for the worst-case, whereas Lazy Replication is optimized for a favorable case where client replicas do not disappear without warning. The technique of Lazy Replication could be also incorporated into SwiftCloud. However, it comes with a trade-off, caused by clock skew and network delay, between a fraction of rejected transfer requests and log pruning period.

In addition to causal consistency, Lazy Replication offer other consistency modes for operations.

In particular, it allows the application to explicitly specify causal dependencies for each operation, using application-specific knowledge. This solutions adds complexity and results in a weaker form of causal consistency, but can potentially increase performance and availability; additionally, it can serve as a tool to implement dependencies across different server-side services [START_REF] Ladin | Providing high availability using lazy replication[END_REF]. In addition to causal consistency, Ladin et al. also describe a protocol extension that supports strongly consistent, server-ordered or externally-ordered operations.

Bayou. Bayou [START_REF] Petersen | Flexible update propagation for weakly consistent replication[END_REF][START_REF] Terry | Managing update conflicts in Bayou, a weakly connected replicated storage system[END_REF] is an influential weakly-consistent replicated database system for mobile computing environments with weak connectivity. A Bayou node replicates the full database, and communicates in peer-to-peer manner with other nodes. Bayou also predates the RDT concept, and delegates conflict detection and resolution logic to the application. The application defines a write to comprise three parts: an update, a dependency check, which specifies the conditions in which the update can be executed, and a merge procedure. A replicated write is applied locally if the dependency check succeeds. Otherwise, the merge procedure modifies the original update.

A write operation can be submitted to any Bayou node, but a single primary node assigns a stable reference execution order followed thereafter by all replicas; this is can be viewed as the stable prefix of an arbitration order in the RDT specification model (Section 2.3) [START_REF] Burckhardt | Cloud types for eventual consistency[END_REF]. Each Bayou server maintains two data versions: committed and tentative. The committed version reflects the execution of write operations that have been assigned a fixed position in the reference execution order. The tentative version extends the committed version with non-stable writes, for which the final position in the execution order is not known.

The main advantage of Bayou is high flexibility. It can express consistency stronger than causal, provided the application is willing to wait until its write is committed, i.e., at the expense of responsiveness and availability. However, Bayou does not tolerate all failures, its programming model is not modular, and the system does not scale with database size (due to full replication) or with system size (as it uses fat version vectors).

TouchDevelop. TouchDevelop [START_REF] Burckhardt | Bringing TouchDevelop to the cloud. Inside Microsoft Research Blog[END_REF] is a recent mobile application framework, somewhat similar to Bayou, but optimized for a star communication topology. A primary data replica resides in the cloud, and all mobile devices synchronize with that replica. TouchDevelop provides a programming model based on a set of predefined Cloud Types [START_REF] Burckhardt | Cloud types for eventual consistency[END_REF], which is a category of RDTs similar to op-based. It provides facilities to compose and to couple objects, using a few predefined patterns.

From a consistency perspective, the database can be considered a single large object. Just like 9.2. RELEVANT SYSTEMS in Bayou, to implement stronger consistency, an application can enforce and await synchronous synchronization with the cloud, in order to determine the final position of recent operations in the execution order (arbitration order).

A recent work of Burckhardt et al. [START_REF] Burckhardt | Cloud types: Robust abstractions for replicated shared state[END_REF] presents TouchDevelop protocols. Although independent from SwiftCloud, both designs have major similarities. Likewise to SwiftCloud, TouchDevelop also relies on a hybrid of client-assigned and cloud-assigned update timestamps, and on a shared at-most-once guard on the cloud replica. However, it assumes only a single cloud replica (DC). Therefore, TouchDevelop (i) assigns at most one cloud timestamp per update, meaning that a version is identified by a scalar timestamp rather than a vector; (ii) can use the presence of a central replica to apply an optimized log reduction scheme that saves bandwidth and storage; and

(iii) does not tolerate cloud replica failures (or assumes a synchronous fault-tolerant protocol).

Similarly to Bayou, TouchDevelop does not support partial replication, nor a large database.

Database of independent objects.

A number of other systems provide partial database replication without cross-object guarantees, i.e., they provide several independent small objects.

The Google Drive Realtime API [START_REF] Cairns | Build collaborative apps with Google Drive Realtime API. Google Apps Developers Blog[END_REF] is a framework for programming responsive web applications with replicated objects backed by the cloud. It provides op-based RDTs with a fixed structure, based on Operational Transformation functions to resolve concurrent updates [START_REF] Ellis | Concurrency control in groupware systems[END_REF]. Mobius [START_REF] Chun | Mobius: Unified messaging and data serving for mobile apps[END_REF] provides both data replication and messaging facilities for mobile applications backed by the cloud, with a range of cloud-controlled caching and freshness policies. Cimbiosys [START_REF] Venugopalan Ramasubramanian | Cimbiosys: A platform for content-based partial replication[END_REF] supports partial replication with content-based filtering, supporting various topologies, such as a hierarchy of personal devices and a cloud.

Compared to SwiftCloud, the implementations of these system can be simpler or more flexible, at the expense of consistency (programming) model limited to the reduced scope of an object.

Session Guarantees

Session guarantees are desirable client-centric consistency properties for systems where a client may use several servers [START_REF] Terry | Session guarantees for weakly consistent replicated data[END_REF]. They include monotonic reads, read-your-writes, and others. In our context, their conjunction is equivalent to causal consistency [START_REF] Brzezi Ński | From session causality to causal consistency[END_REF][START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF]. The original session guarantees algorithm of Terry et al. [START_REF] Terry | Session guarantees for weakly consistent replicated data[END_REF] is not available in the presence of faults, which corresponds to K = 1 parameter in SwiftCloud. Brzezi ński et al. [START_REF] Brzezi Ński | Kresilient session guarantees synchronization protocol for mobile ad-hoc networks[END_REF] describe a fault-tolerant protocol that also relies on K-stability. However, their protocol assumes that updates are synchronous and must be atomically accepted by K servers. In contrast, updates in SwiftCloud are asynchronous, and the transfer protocol tolerates partial failures.

Geo-replicated Databases for Server-Side Applications

A number of server-side geo-replicated systems support variants of causal consistency across DCs. We first present unique aspects of these approaches, and then evaluate them in the context 9.2. RELEVANT SYSTEMS assigning update timestamps contiguously at each shard replica, multiple dependencies on a shard replica can be encoded with a single matrix entry, reducing the size of transmitted metadata. Orbe uses dependencies only to ensure monotonic causal progression of states, and implements read-only transactions using another mechanism. To compute a snapshot, it relies on a global real-time clock. When it reads from a snapshot at a shard server, it may need to wait out some uncertainty period to ensure consistency, risking unavailability.

ChainReaction. ChainReaction [START_REF] Almeida | ChainReaction: a causal+ consistent datastore based on Chain Replication[END_REF] takes a different approach to sharding and replication.

ChainReaction uses dedicated replication protocols within a DC, based on a variant of chain replication [START_REF] Van Renesse | Chain replication for supporting high throughput and availability[END_REF]. 1 The protocol is optimized for reads; it is slower for updates, as it accepts an update only after all of its dependencies are fully replicated in the local DC. For geo-replication, ChainReaction relies on a geo-replication service that is shared between all shards. This service enforces coarse-grained causal dependencies independent of shards, using version vectors, similar to the DC-assigned metadata in SwiftCloud. In order to avoid the bottleneck of sequential processing at the shared geo-replication service, ChainReaction additionally uses bloom filters to encode finer-grained dependencies, which enables more processing parallelism. Similarly, ChainReaction relies on a shared operation sequencer within a DC to implement read-only transactions. The sequencer logically serializes all updates and read transactions in a DC.

Causal dependencies and stability. All these algorithms make use of full stability condition to spare unnecessary causal dependencies on stable updates. Further, Du et al. [START_REF] Du | Closing the performance gap between causal consistency and eventual consistency[END_REF] propose to expose only n-stable updates and client's own updates (as in SwiftCloud), to reduce dependency messages and increase throughput.

Walter. Walter [START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF] is a partial geo-replication system, i.e., a system without a DC sharding support, but with arbitrary partial replication at a global scale. Walter supports causal consistency with a single predefined RDT counting-set type (cf. Section 2.3.2.3) and an additional support for strongly consistent register type, mixing weakly and strongly consistent transactions. 2 Walter provides general purpose read-update transactions, using a metadata based on timestamp and version vectors, the same as DC-assigned metadata of SwiftCloud.

Comparison and Applicability to Client-Side Replication

Geo-replication systems offer causal consistency only to a client tied to a specific DC replica in its failure domain. None of the existing designs is directly applicable to client-side replication, both due to metadata design, and other issues, summarized in Table 9.1 and below.

Geo-replication protocols do not tolerate client or DC failures. All existing designs suffer from the transitive dependency issue from Section 7.1.3.2, violating consistency or availability for a remote client. They mostly rely on Lean-but-Unsafe metadata (Eiger, Orbe, ChainReaction, and Walter), which does not offer at-most-once delivery for operations on client-side replicas and assumes that data is updated by overwriting, or otherwise rely on the overly costly Safebut-Fat metadata (COPS-GT and Bolt-on). This is a consequence of designing for LWW register support only, which makes implementing other RDTs more complex and costly (see, for example, Another type of metadata used by these systems is a version representation, which identifies a snapshot to read, and can also serve to compare states when replicas synchronize. Many geo-replication systems use a version representation valid only in a local DC. This makes implementation of failover between DCs difficult. Orbe is one exception, as it uses a globally valid real-time clock, but its read protocol is prone to DC outages. Walter and SwiftCloud represent a version with a globally valid version vector, the same that they use for causality metadata.

The exact size of causality metadata depends on runtime factors, whereas the worst-case size is primarily impacted by the location of timestamp assignment. When fewer nodes assign metadata, it tends to be smaller, but this may limit throughput scalability with database size. For instance, SwiftCloud, Walter and ChainReaction offer the smallest guaranteed size of causality metadata, but do not provide sharding, or rely on a centralized component to support sharding.

SwiftCloud support for sharding is limited compared to the most scalable decentralized serverside designs. The presented design does not support sharding; our earlier prototype [START_REF] Zawirski | SwiftCloud: Fault-tolerant geo-replication integrated all the way to the client machine[END_REF] (not evaluated in this thesis) offered sharding with a centralized component, similar to ChainReaction.

Reconciling client-side replication with a more decentralized sharding support, without growing metadata size (like some decentralized sharding protocols, e.g., Eiger or Orbe), is future work.

Chapter 10

Conclusion

This chapter concludes the thesis with a summary of findings from both parts of the thesis, and of their limitations, and identifies areas for future work.

Summary

In this thesis, we studied the design of dependable and efficient algorithms for RDTs and causal consistency, and explored the trade-offs between dependability and metadata optimizations.

Our two main contributions are: a comprehensive study of metadata space optimality for RDT implementations, including new optimized designs and impossibility results, and the design of a scalable causally-consistent RDT database for client-side applications.

Metadata optimization for state-based RDT implementations. In the first part of the thesis, we considered the problem of minimizing metadata incurred by state-based RDT implementations, i.e., objects communicating by exchanging their complete state. They can be found, for instance, in server-side geo-replicated object databases without cross-object consistency guarantees. The design of optimized implementations, to reduce the size of metadata, is nontrivial, as it creates a tension between correctness w.r.t. type semantics, efficiency, and fault-tolerance.

We formalized worst-case metadata overhead analysis and applied it to six existing data type implementations. We found that most of them incur substantial overhead, linear in the number of updates, or polynomial in the number of replicas. Moreover, we found that the concurrent semantics of a data type has a critical impact on the extent of the overhead.

Our first major contribution are optimized designs of add-wins set and multi-value register RDTs. Existing implementations store a trace of removed elements (tombstones) or of concurrently assigned values. We proposed safe optimizations based on algorithms that efficiently aggregate removed information using a variant of version vectors [START_REF] Parker | Detection of mutual inconsistency in distributed systems[END_REF]. These optimizations reduce the metadata overhead by orders of magnitude, down to the number of replicas.

Part IV

Appendix

=) 9 f 2 C.E : C.replica( f ) = r 0 ^e del(C) °°°°! f ) Reliable delivery T 1 T ∏1 \ T ∑1 Causal delivery T c 8e, f 2 C.E : C.action(e) = send ^e (ro(C)[del(C)) + °°°°°°°°°°°! f =) e ro(C) °°°! f _ 9g 2 C.E : e del(C) °°°°! g ro(C) § °°°°! f Rel. causal delivery T 1c T 1 \ T c
Causal timestamps

T ct (ro(C) [ del(C)) + µ ar(C) Eventual flush T f 8e 2 C.E : C.action(e) = do =) 9 f 2 C.E : C.action( f ) = send ^e ro(C) °°°! f Table A.1:
Catalog of popular network specifications (variation of Burckhardt et al. [START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF]). . buffer of the latest written value let b R = ;

.

: (w m , R m ) let (w m , R m ) = (b w , b R ) . flush the buffer (b w , b R ) = (;, ;) deliver((w m , R m )) W √ (W \ R m ) [ w m
. keep only overwritten values; add new values Similar optimizations are possible for other data types, such as add-wins set. We adopted them in our RDT implementations for the SwiftCloud system from Part III. In addition to the presented optimizations of state size, further optimizations of message size are also feasible [START_REF] Baquero | Making operation-based CRDTs operation-based[END_REF]. For some constant K 4 , return value V = C.rval(e) of read consumes no less than:

len P(Z) (V ) ∏ K 4 (1 + X a2V len(a)).
From the definition of read, we observe V = U(V ). This leads to the overhead limit that holds for some constant K: For some constant K 4 , return value V = C.rval(e) of read consumes no less than:

len P(Z) (V ) ∏ K 4 (1 + X a2V len(a)).
From the definition of read, we observe V = U(V ). This leads to the overhead limit that holds for some constant K:

C.2.1 Présentation du problème

Nous considérons le support pour une variété d'applications côté clients partageant une base de données des objets de RDTs, que le client peut lire ou modifier. Nous visons à s'adapter aux milliers de clients couvrant l'ensemble de l'internet, ainsi qu'à une base de données de taille arbitraire. x.add [START_REF]Riak distributed database[END_REF] x.add(3)

V 1 V 2
y.add(2) y.add [START_REF]Riak distributed database[END_REF] x.add(4) x.add [START_REF]Riak distributed database[END_REF] x.add(3)

U C read x {1,3} (
V 1 V 2
y.add(2) y.add [START_REF]Riak distributed database[END_REF] x.add(4) Un algorithme sans lacune maintient un état de réplique causalement cohérent avec progression monotone non décroissante [START_REF] Ahamad | Causal memory[END_REF]. Donné la mise à jour u, nous notons u.deps son ensemble de prédécesseurs causales, dit son ensemble de dépendances. Si une réplique complète étant dans un état cohérant V , reçoit u, et si ses dépendances sont satisfaites, soit u.deps µ V , elle applique u. Le nouveau état est V 0 = V © {u}, où nous notons © un opérateur de fusion de log, qui filtre tous les doubles survenus pendant la transmission. L'état V 0 est cohérent, ainsi que son monotonie est satisfaite, car V µ V 0 .

Si les dépendances ne sont pas remplies, la réplique met u dans la mémoire du tampon jusqu'au remplissement de la lacune.

C.2.2.2 Cohèrence causale dans les répliques client partielles

La cohérence se complique, quand la réplique client possède seulement une partie de la base de données et de ses métadonnées [START_REF] Belaramani | PRACTI replication[END_REF]. Pour éviter la complexité du protocole et pour mimiser la

C.2.3 La mise en oeuvre et l'évaluation

Nous implémentons le SwiftCloud et démontrons expérimentalement que notre modèle atteint ses objectifs avec un coût modeste de manque de fraîcheur. Nous évaluons le SwiftCloud dans Amazon EC2, contre un port de WaltSocial [START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF] et contre YCSB [START_REF] Cooper | Benchmarking cloud serving systems with YCSB[END_REF]

  Conflict avoidance approach with synchronous protocol offering strong consistency. read:{a,b} write({a,b,c}) read:{a,b,c} (B) Conflict arbitration approach with asynchronous protocol causing lost updates. read:{a,b} write({a,b,c}) protocol: conflict detection + app-level resolution ACCEPTED ACCEPTED {a} || {a,b,c} {a} || {a,b,c} read:{a} read:{a,b,c} (C) Conflict detection approach with asynchronous protocol relying on application-level conflict resolution. read:{a,b,c} (D) Automatic conflict resolution approach with asynchronous protocol embedding conflict resolution.

Figure 2 . 1 :

 21 Figure 2.1: Comparison of different approaches to conflicts on an example execution with two clients issuing conflicting updates on two different replicas, r 1 and r 2 . Boxes indicate operations issued by clients; arrows indicate operation invocation and response.

Figure 2 . 2 :

 22 Figure 2.2: Components of the system model.

Algorithm 2 . 2 3 : 5 :v = a 6 : 7 :a √ a + 1 . record increment in the current value 8 :b √ b + 1 . record increment in the buffer 9 :

 2235671819 Op-based implementation of counter (Ctr). 1: ß = N 0 £ N 0 M = N 0 2: initialize(r i ) : (a, b) let (a, b) = (0, 0) . a: current value; b: buffer of increments 4: do(read, t o ) : v let do(inc, t o ) send() : d

Figure 2 . 3 :

 23 Figure 2.3: Time-space diagram of an execution of the op-based counter (Algorithm 2.2) with three replicas. Every event is labelled with the execution step, the return value of the function

Figure 2 . 4 :

 24 Figure 2.4: An example execution of the state-based counter (Algorithm 2.3) with three replicas. For brevity, we note vector [r 1 7 ! a, r 2 7 ! b, r 3 7 ! c] as [a b c].

Figure 2 . 5 :

 25 Figure 2.5: Hasse diagram of a fragment of the join-semilattice of states (vectors) of the statebased counter implementation. Solid arrows between states indicate a transitive reduction of < order; dotted blue paths indicate replicate state transitions during execution from Figure 2.4. We omit replica's own ID in a state, which is irrelevant for the order, and show only vectors.

Figure 2 .

 2 Figure 2.4 illustrates an execution of the state-based counter. Note that the first increment at replica r 1 reaches replica r 3 both indirectly from replica r 2 , and later, directly in a delayed message from r 1 . Nevertheless, all replicas converge towards a vector [r 1 7 ! 1, r 2 7 ! 1, r 3 7 ! 0], which corresponds to counter value 2.

Figure 2 . 6 :

 26 Figure 2.6: An example execution of the basic state-based multi-value register (Algorithm 2.5).

Figure 2 .

 2 Figure 2.6 illustrates an example execution of the implementation. Clients at replicas r 1

CHAPTER 2 .Figure 2 . 7 :

 227 Figure 2.7: Components of implementation and specification models, and their relation through correctness definition. Squares (resp. empty squares) indicate concrete (resp. abstract) executions.

  op,vis,ar) = {a |9e 2 E : op(e) = write(a)^6 9e 0 2 E : e 0 = write(b) ^e vis °°! e 0 }. (2.7)

removed,

  Op Set = {read} [ {add(a) | a 2 Z} [ {rem(a) | a 2 Z}, and read returns Val Set = P(Z) [ {?}.

CHAPTER 2 .

 2 REPLICATED DATA TYPES

  RWSet (read, E, op,vis,ar) = {a | E| add(a) 6 = ; ^8e 2 E| rem(a) : 9e 0 2 E| add(a) : e vis °°! e 0 }, where RWSet has the same signature as Set, and E| o is the restriction of events E to operations o.

  [START_REF] Briquemont | Optimising client-side geo-replication with partially replicated data structures[END_REF]. The add-wins set returns {13, 26}, because both rem(a) operations are concurrent in visibility with respective add(a) operations. Conversely, the remove-wins returns an empty set.

Figure 2 . 9 :

 29 Figure 2.9: Set semantics alternatives illustrated on an operation context of read, with different return value specifications. We show only transitive reduction of partial visibility order.

ure 2 . 10 .

 210 In the first rule, replica r performs an operation o with timestamp t that transforms state R(r) and yields a return value v, as computed by the implementation function do. We record this information using functions replica, action, op, timestamp, and rval, respectively. In the second rule, replica r sends a message m created by the implementation function send, which transforms the state R(r), and puts a message into the map of sent messages N(mid) = m, using CHAPTER 2. REPLICATED DATA TYPES

Figure 2 . 10 :

 210 Figure 2.10: Definitions of the set of configurations and the transition relation for a data type implementation D ø . (R, N) 2 Config is a configuration, and (R 0 , N 0 ) is the initial configuration.

Chapter 3 Metadata

 3 Space Complexity ProblemEvery town has a story. Tombstone has a legend.Tombstone by George P. Cosmatos Contents 3.1 Problem Statement . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 40 3.2 Optimizing Implementations . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 42 3.2.1 Successful Optimizations . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 42 3.2.1.1 Add-Wins Set . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 42 3.2.1.2 Multi-Value Register . . . . . . . . . . . . . . . . . . . . . . . . . . . 48 3.2.2 Prior Implementations and Unsuccessful Optimizations . . . . . . . . . . . 51 3.2.2.1 Last-Writer-Wins Register . . . . . . . . . . . . . . . . . . . . . . . 51 3.2.2.2 Remove-Wins Set . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 52 3.2.2.3 Last-Writer-Wins Set . . . . . . . . . . . . . . . . . . . . . . . . . . 53 3.2.2.4 Counter . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 54 3.3 Summary . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 54 CHAPTER 3. METADATA SPACE COMPLEXITY PROBLEM

Definition 3 . 2 (

 32 Asymptotic Metadata Overhead). Consider an implementation D ø and a positive function f (n, m).

Definition 3 . 3 (

 33 Optimality). Implementation D ø is an asymptotically optimal state-based implementation of type ø with an upper bound complexity D ø 2 b

Algorithm 3 . 1

 31 Naive state-based implementation of add-wins set (AWSet).

Figure 3 .

 3 Figure 3.1A illustrates an example execution. Initially the state is empty. Replica r 1 adds some integer elements a and b, and later removes them by placing their timestamps in the tombstone set. Concurrent add of element a at replica r 2 generates a new instance. Therefore,when replicas r 1 and r 2 synchronize, the instance of a from replica r 2 "wins" over the remove and makes a visible. Replica r 3 delivers a message from r 2 including all add updates, but no rem updates, before it becomes disconnected (as indicated by a squiggly line). During disconnection, it

  r 1 ,1),(b,r 1 ,2)} read:{a} read:{a} deliver {(a,r 2 ,1), (c,r 3 ,1)} {(a,r 1 ,1),(b,r 1 ,2)} read:{a,c} add(c) {(a,r 1 ,1),(b,r 1 ,2),(a,r 2 ,1),(c,r 3 ,1 } ∅ send unsafe GC period add(a) {(a,r 1 ,1)} ∅ add(a) {(a,r 2 ,1)} ∅ A=∅ T=∅ A=∅ T=∅ A=∅ T=∅ (A) Execution of the naive implementation (Algorithm 3.1).

  read:{a,c} (C) Witness abstract execution for the two executions. Arrows show visibility, arbitration is irrelevant.

Figure 3 . 1 :

 31 Figure 3.1: Concrete executions of two implementations of add-wins set (AWSet), and their witness abstract execution. Replica IDs in an object state, and message content for send (the copy of state) are ommited. The period of interrupted connectivity is indicated with a squiggly line.

Theorem 3 . 1 .

 31 Let D AWSet be the naive add-wins set implementation defined in Algorithm 3.1, such that D AWSet sat[V state , T any ] F AWSet . The complexity of D AWSet is b £(m lg m).

15 :A

 15 √ A \ D . remove them from active instances 16: send() : (vv m , A m ) 17: let vv m = vv 18: let A m = A 19: deliver((vv m , A m )) 20:

Figure 3 .

 3 1A, described earlier, illustrates this problem. While replica r 3 is disconnected (or failed), replicas r 2 and r 3 do not know if r 3 received tombstones for elements a and b. If they were to remove tombstone for a and b, when r 3 connects again to r 2 , r 2 would be unable to distinguish a fresh instance of element c and previously removed instances of a and b. Optimized implementation. Algorithm 3.2 presents a new optimized design of add-wins set implementation. It addresses the metadata complexity problem by summarizing: (1) removed and (2) re-added element instances. We explain our design by comparison to the naive one. We describe optimization (1) first, assuming D = ; in Line 10, which disables optimization (2).

Figure 3 . 2 :

 32 Figure 3.2: Semantics of a version vector in the optimized add-wins set. The illustration refers to the last state of replica r 2 , (r 2 , vv, A) in Figure 3.2. Horizontally-aligned boxes illustrate the instances (timestamps) represented by vector entry for that replica, whereas dashed rectangles represent the sets A and T(vv, A).

Figure 3 .

 3 Figure 3.2 illustrates this relation between version vector vv, set A, and the computed tombstone set T(vv, A). It is based on the example state of replica r 2 from Figure 3.1B.

FFigure 3 . 3 :

 33 Figure 3.3: Operation context of an add-wins set illustrating opportunity for coalescing adds. The set of operations that are irrelevant for semantics of read is indicated with a blue dashed shape.

Optimization 2 :

 2 Coalescing adds. Another optimization coalesces multiple active instances of a same element. We observe that, among multiple add operations on the same element, only the latest ones are relevant in the add-wins semantics. Consider for example an operation context from Figure 3.3. The latest add(a) at replica r 1 and r 2 , and the latest add(b) at r 1 are sufficient to determine the correct return value of read. Old adds become irrelevant when new adds arrive.

Figure 3 .

 3 Figure 3.1B illustrates an execution of the optimized implementation, with the same client and network behavior as Figure 3.1A. Note that replicas r 1 and r 2 are now unaffected by unavailability of replica r 3 , and discard removed element instances immediately. When r 3 communicates back, the version vector of r 2 prevents removed instances of a and b from reappearing.

Figure 3 .

 3 Figure 3.4B illustrates the counterexample. When three concurrent writes of value 0 reach replica r 4 , they are merged into a single vector, as explained above. However, each of replicas r 1 , r 2 , and r 3 , subsequently overwrite value 1 with new values, respectively, 1, 2, and 3, being unaware of each other's updates. When these three writes reach r 4 , both earlier value 0, and the new values remain in the state, since no vector dominates another. The read operation returns all four values {0, 1, 2, 3}, which is incorrect. As illustrated in the correct abstract execution in Figure3.4D, all three events of write(0) operation were overwritten in visibility by write(i)

  Witness abstract execution for correct executions. Arrows show visibility, arbitration is irrelevant.

Figure 3 . 4 :

 34 Figure 3.4: Example execution of three different implementations of multi-value register (MVReg), and a corresponding abstract execution. Note the problematic concurrent writes of the same value. The figure omit the replica ID. The message content for send is always a copy of the state.

Algorithm 3 . 4

 34 Optimized state-based implementation of multi-value integer register (MVReg).1: ß = ReplicaID £ P(Z £ (ReplicaID 7 ! N 0 )) M = P(Z £ (ReplicaID 7 ! N 0 )) 2: . . .. all other methods are identical to Algorithm 2.53: deliver(A m ) 4:

Theorem 3 . 4 .

 34 dominated by the least upper bound of vectors for values 1, 2, and 3. Therefore, read on replica r 4 yields {1, 2, 3} correctly, as in the witness abstract execution (Figure 3.4D). Let D § MVReg be the multi-value register implementation defined in Algorithm 3.4, such that D § MVReg sat[V state , T any ] F MVReg . The complexity of D § MVReg is b £(n lg m).

21 :ATheorem 3 . 5 .

 2135 √ A [ A m . union sets of add instances Let D § LWWReg be the last-writer-wins register implementation defined in Algorithm 2.4, such that D § LWWReg sat[V state , T any ] F LWWReg . The complexity of D § LWWReg is b £(lg m).

Figure 4 . 1 :

 41 Figure 4.1: Structure of experiment family and implications of Lemma 4.1 on the size of object state encoding. For simplicity, the index set Q is a set of positive integers.

Definition 4 . 1 ( 2 CLemma 4 . 1 .

 41241 Experiment Family). An experiment family for an implementation D ø is a tuple (Q, n, m, C, e, readback) where Q is a finite index set, 2 ∑ n ∑ m, and for each experiment index AE 2 Q, C AE 2 ÇD ø É is an experiment execution with n replicas and m updates, e AE AE .E| read is a culprit read in that execution and readback : D ø .ß ! Q is a readback function satisfying readback(state(e AE )) = AE. If (Q, n, m, C, e, readback) is an experiment family for an implementation D ø , then the worst-case metadata overhead of D ø is at least: wcmo(D ø , n, m) ∏ blg | §| |Q|c / (max AE2Q len(C AE .rval(e AE ))). Proof. Since readback(state(e AE )) = AE, the states state(e AE ) are all distinct and so are their encodings enc(state(e AE )), by injectivity. Since there are fewer than |Q| strings of length strictly less than blg | §| |Q|c, for some experiment AE 2 Q we have len(enc(state(e AE ))) ∏ blg | §| |Q|c. Then wcmo(D ø , n, m) ∏ mmo(D ø , C AE ) ∏ len(state(e AE )) len(C AE .rval(e AE )) ∏ blg | §| |Q|c max AE 0 2Q len(C AE 0 .rval(e AE 0 ))

Figure 4 .

 4 Figure 4.1 illustrates the structure of an experiment family, and shows the consequences of Lemma 4.1 on the state space complexity.

CHAPTER 4 .

 4 LOWER BOUNDS ON COMPLEXITY AND IMPLEMENTATION OPTIMALITY

Theorem 4 . 1 .

 41 If D Ctr sat[V state , T any ] F Ctr , then the complexity of D Ctr is b ≠(n).

4. 2 .

 2 LOWER BOUNDS Conditions on #replicas & #updates m ∏ n ∏ 2 ^m mod (n °1) = 0 Index set Q = ([2..n] ! [1..

  Witness abstract execution abs(C 0 AE , V state ). Arrows indicate visibility. Arbitration is not shown.

Figure 4 . 2 :Lemma 4 . 2 .

 4242 Figure 4.2: Example experiment (n = 3 and m = 6) and test for counter (Ctr). Dashed lines and colors indicate phases of the experiment; blue dotted shape represents the configuration where the test driver program is applied to read-back AE(2); underlined read indicates the culprit read.

Proof of Theorem 4 . 1 .

 41 To fulfill the definition of lower bound b ≠ (Definition 3.2), for any given n 0 , m 0 , we pick n = n 0 and some m ∏ n 0 such that m is a multiple of (n °1) and m ∏ n 2 . Take the experiment family (Q, n, m, C, e, readback) given by Lemma 4.2. Then for any AE, C AE .rval(e AE ) is at most the total number of increments m in C AE , by F Ctr . Using Lemma 4.1 and m ∏ n 2 , for some constants K 1 , K 2 , K 3 , K independent from n 0 , m 0 we get: wcmo(D Ctr , n, m) ∏blg | §| |Q|c/(max AE2Q len(C AE .rval(e AE )))

Figure 4 . 3 .

 43 The read-back delivers the first message from replica 2 at replica 1, and reads value ;, since all visible adds are covered CHAPTER 4. LOWER BOUNDS ON COMPLEXITY AND IMPLEMENTATION OPTIMALITY Conditions on #replicas & #updates m ∏ n ∏ 2 ^(m °1) mod (n °1) = 0 Index set Q = ([2..n] ! [1..

  Witness abstract execution abs(C 0 AE , V state ). Arrows indicate visibility. Arbitration is not shown.

Figure 4 . 3 :

 43 Figure 4.3: Example experiment (n = 3 and m = 7) and test for add-wins set (AWSet). Dashed lines and colors indicate phases of the experiment; blue dotted shape represents the configuration where the test driver program is applied to read-back AE(2); underlined read indicates the culprit read.

  as we explained informally above. Let C 0 AE = exec(D AWSet , (R 0 , N 0 ), (init; exp(AE); test(r))). be an extension of experiment C AE with the steps of test program (e.g., Figure 4.3A), and abs(C 0 AE , V state ) be the witness abstract execution of this extension (e.g., Figure 4.3B). Let v i denote value of a read into v in the i-th iteration of the loop in the program test(r) (Line 15). By F AWSet the value of v i is determined by the set of visible add and rem operations in the operation context, and their relation. The operation context contains: (a) first max(AE(r), i) operations add(0) from replica r, and (b) first AE(r 0 ) operations add(0) from every replica r 0 6 = r, and (c) and a single rem(0) made on replica 1.

Theorem 4 . 3 .Lemma 4 . 4 .

 4344 If D RWSet sat[V state , T any ] F RWSet , then the complexity of D RWSet is b ≠(m). If D RWSet sat[V state , T any ] F RWSet , m ∏ n ∏ 3and m is a multiple of 4, then the tuple (Q, n, m, C, e, readback) in Table

Figure 4 . 4 :

 44 Figure 4.4: Example experiment (n = 3 and m = 12) and test for remove-wins set (RWSet). Dashed lines and colors indicate phases of the experiment; blue dotted shape represents the configuration where the test driver program is applied to read-back AE; underlined read indicates the culprit read.

Theorem 4 . 4 .Lemma 4 . 5 .

 4445 If D LWWSet sat[V state , T any ] F LWWSet , then the complexity of D LWWSet is b ≠(n lg m). If D LWWSet sat[V state , T any ] F LWWSet , n ∏ 2and m is a multiple of 2n °2, then the tuple (Q, n, m, C, e, readback) in Table

Theorem 4 . 5 .

 45 If D MVReg sat[V state , T any ] F MVReg , then the complexity of D MVReg is b ≠(n lg m).

  Witness abstract execution abs(C 0 AE , V state ). Solid arrows indicate visibility. Horizontal position indicates arbitration order.

Figure 4 . 5 :

 45 Figure 4.5: Example experiment (n = 3 and m = 8) and test for last-writer-wins set (LWWSet). Dashed lines and colors indicate phases of the experiment; blue dotted shape represents the configuration where the test driver program is applied to read-back AE(2); underlined read indicates the culprit read.

Lemma 4 . 6 .

 46 If D MVReg sat[V state , T any ] F MVReg , n ∏ 2 and m ∏ n is such that (m °1) is a multiple of (n °1), then the tuple (Q, n, m, C, e, readback) in Table

= 3 configuration 3 (

 33 (R init [1→R α (1)], N init ) A) Concrete execution C 0 AE . write(1) read:{1} (e α ) write(0) write(0) write(0) read:{1} (v 1 ) read:{0,1} (v 2 )write(0) write(0) write(0) Witness abstract execution abs(C 0 AE , V state ). Arrows indicate visibility. Arbitration is not shown.

Figure 4 . 6 :

 46 Figure 4.6: Example experiment (n = 3 and m = 7) and test for multi-value register (MVReg). Dashed lines and colors indicate phases of the experiment; blue dotted shape represents the configuration where the test driver program is applied to read-back AE(2); underlined read indicates the culprit read.

Figure 6 .Figure 6 . 1 :

 661 Figure 6.1 illustrates our system model. A cloud infrastructure connects a small set (say, tens) of geo-replicated data centers, and a large set (thousands) of clients. A DC has abundant computational, storage and network resources. Similarly to Sovran et al.[START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF], we abstract a DC as a powerful sequential process that hosts a full replica of the database.1 DCs communicate in a

Definition 6 . 1 (

 61 Causal Order and Consistency). Let an execution H be a set of sequences (one per application process session) of object operation invocations with their return values. Operations e and f are potentially causally-related in H, noted e ! f and called causal order, if:A. An application process session invoked f after it invoked e; or B. Read f observed update e on the same object; or C. There exists an operation g 2 H such that e ! g ! f .

6. 3 .

 3 APPLICATION PROGRAMMING INTERFACE 2. Confluence (safety): two nodes that delivered the same set of updates read the same value.

Figure 7 . 1 :

 71 Figure 7.1: Example evolution of configurations for two DCs, and a client. x and y are sets; box = update; arrow = causal dependence (an optional text indicates the source of dependency); dashed box = named database version/state.

  previous section. Client state can change either because of an update generated by the client itself, called an internal update, or because of one received from a DC, called external. An internal update obviously maintains causal consistency. If an external update arrives, without gaps, from the same DC as the previous one, it also maintains causal consistency. More formally, consider some recent DC state, which we will call the base version of the client, noted V DC . The interest set of client C is noted O µ x,y,.... The client state, noted V C , is restricted to these objects. It consists of two parts. One is the projection of base version V DC onto its interest set, noted V DC | O . The other is the log of internal updates, noted U C . The client state is their merge V C = V DC | O ©U C | O . On cache miss, the client adds the missing object to its interest set, and fetches the object from base version V DC , thereby extending the projection. Safety. Base version V DC is a monotonically non-decreasing causal version (it might be slightly behind the actual current state of the DC due to propagation delays). By induction, internal updates can causally depend only on internal updates, or on updates taken from the base version.

7. 2 .

 2 IMPLEMENTATIONspecific ones. Representation of dependencies also contributes. SwiftCloud uses coarse-grained representation of dependencies, at the granularity of the complete base version used by the client.

7. 2 .

 2 IMPLEMENTATION commits locally at the client; it never fails. If the transaction made update u 2 ø.U, the client replica commits it locally as follows: (1) assign it client timestamp u.t C = (C, k), where k counts the number of committed transactions at the client; (2) assign it a dependency vector initialized with the transaction snapshot vector u.vvDeps = ø.vvDeps; (3) append it to the commit log of local updates on stable storage U C √ U C © {u}. This terminates the transaction, which becomes irrevocable; the client can start a new one, which will observe the committed updates.

  and assumes the new vector vv DC . If any of received updates is a duplicate w.r.t. to the old version or to a local update, the log merge operator handles it safely. Note that transaction atomicity is preserved by Equation7.1, since all updates of a transaction share a common timestamp.

Figure 8 . 1 :

 81 Figure 8.1: Response time for YCSB operations (workload A, Zipfian object popularity) under different system and workload locality configurations, aggregated for all client sessions.

Figure 8 . 2 :

 82 Figure 8.2: Maximum system throughput for different workloads and protocols. Percentage over bars indicates throughput increase/decrease for SwiftCloud compared to server-side replication.

Figure 8 . 3 :

 83 Figure 8.3: Throughput vs. response time for different system configurations and YCSB variants.

  38%, or decrease by up to 11% with SwiftCloud.Throughput vs. response time. Our next experiment studies how response times vary with server load and with the staleness settings. The results show that, as expected, cached objects respond immediately and are always available, but the responsiveness of cache misses depends on server load.For this study, Figure8.3 plots throughput vs. response time, for YCSB A (left side) and B (right side), both for the Zipfian (top) and uniform (bottom) distributions. Each point represents the aggregated throughput and latency for a given transaction incoming rate, which we increase until reaching the saturation point. The curves report two percentiles of response time: the lower (70 th percentile) line represents the response time for requests that hit in the cache (the session locality level is 80%), whereas the higher (95 th percentile) line represents misses, i.e., requests served by a DC.

Figure 8 . 5 :

 85 Figure 8.5: Size of metadata in notification message for a variable number of replicas, mean and standard error. Normalized to a notification of 10 updates.

Throughput. Figure 8 .

 8 4 shows the maximum system throughput on the Y axis, increasing the number of replicas along the X axis. The thin lines are for a single DC, the bold ones for three DCs. Solid lines represent the fast notification rate, dashed lines the slow one. The figure shows, left to right, YCSB Workload A, YCSB Workload B, and SocialApp.

Figure 8 . 6 :

 86 Figure 8.6: Storage occupation at a single DC in reaction to client churn for Lean-and-Safe SwiftCloud and Lean-but-Unsafe alternative.

Figure 8 . 7 :Figure 8 . 8 :

 8788 Figure 8.7: Response time for a client that hands over between DCs during a 30 s failure of a DC.

Contents 9 . 1

 91 Consistency Models for High Availability . . . . . . . . . . . . . . . . . . . . . . . . . 110 9.2 Relevant Systems . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 110 9.2.1 Replicated Databases for Client-Side Applications . . . . . . . . . . . . . . . 111 9.2.1.1 Systems that Support Inter-Object Consistency . . . . . . . . . . . 111 9.2.1.2 Systems that Support Intra-Object Consistency Only . . . . . . . 113 9.2.1.3 Session Guarantees . . . . . . . . . . . . . . . . . . . . . . . . . . . 115 9.2.2 Geo-replicated Databases for Server-Side Applications . . . . . . . . . . . . 115 9.2.2.1 Approaches . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 116 9.2.2.2 Comparison and Applicability to Client-Side Replication . . . . . 117

  entry † not live during DC failures ‡ only for server-side replicas Table

Figure 8 .

 8 Figure 8.6).

  f , g 2 C.E : e del(C) °°°°! f ^e del(C) °°°°! g ^C.replica( f ) = C.replica(g) =) f = g) (8e, f 2 C.E : e del(C) °°°°! f =) C.replica(e) 6 = C.replica( f ))At-least-once deliveryT ∏1 (8e2C.E : 8r, r 0 : C.action(e) = send ^C.replica(e) = r 6 = r 0

Algorithm A. 1

 1 Optimized op-based multi-value integer register implementation (MVReg).ß = P(Z £ Timestamp) 3 M = P(Z £ Timestamp) 2 initialize(r i ) : (W, b w , b R ) let W = ;. set of non-overwritten values with their unique IDs (timestamps) let b w = ;

B. 2 . 1 :. read e § 6 :all i 2 [ 2 ..n] do 7 : 1 m/ 2

 21622712 METADATA OVERHEAD OF SPECIFIC IMPLEMENTATIONSProof. Consider the following driver program (introduced in Section 4.1) defined for any positive n and m such that m mod 2 = 0: procedure inflate(n, m)for do i read m+1+i. dummy read on other replicas to fulfill wcmo definition Given any number of replicas n 0 and any number of updates m 0 ∏ n 0 , we pick n = n 0 and m = 2m 0 to demonstrate overhead on execution:C = exec(D AWSet , (D AWSet .initialize, ;), inflate(n, m)).Clearly, the execution has n replicas and m updates. The execution is well-defined since the driver program uses message identifiers and timestamps correctly. Let (r, k, A, T) = state(e § ) be the state of replica 1 at the time of read e § , where A is a set of active instances and T is a set of timestamps of removed instances. By definitions of add and rem in D AWSet , counter k = m, the set of active instances is empty for this execution, A = ;, whereas the set of removed instances contains tombstones for all removed elements, |T| = m/2. Thus, for some constants K 1 and K 2 , state (r, k, A, T) needs to occupy at least:len(r, k, A, T, ) ∏ K X i=1 (lg n + lg m + len Z (i)) ∏ K 2 m lg m.By definition of read, the return value of read e § is an empty set, C.rval(e § ) = ;, which can be encoded with a constant length string. Therefore, for some constant K the overhead at the time of read e § fulfills the definition of b ≠:wcmo(D AWSet , n, m) ∏ mmo(D AWSet , C) ∏ len(r, k, A, T) len(C.rval(e § ))∏ K m lg m. Theorem B.3 (Originally, Theorem 3.1). Let D AWSet be the naive add-wins set implementation defined in Algorithm 3.1, such that D AWSet sat[V state , T any ] F AWSet (by [34]). The complexity of D AWSet is b £(m lg m). Proof. By Theorem B.1 and Theorem B.2. Theorem B.4. Let D § AWSet be the optimized add-wins set implementation defined in Algorithm 3.2, such that D § AWSet sat[V state , T any ] F AWSet (by [34]). The complexity of D § AWSet is b O(n lg m).

B. 2 .

 2 METADATA OVERHEAD OF SPECIFIC IMPLEMENTATIONS vector vv (a mapping from a replica ID to an integer). By definitions of write operation and deliver function, the two mutators of entries in A, 8(a, vv) 2 A, r 0 2 ReplicaID.vv(r 0 ) ∑ m. Also, by definitions of write and deliver (alternatively, by specification F MVReg applied to visibility witness), the number of concurrent entries in the state cannot exceed the number of sources of concurrency:|A| ∑ n. Let U(A) = {a | 9(a, vv) 2 A}.Then, for some constants K 1 , K 2 and K 3 , state (r, A) can be encoded with a string no bigger than:len(r, A) ∑ K 1 (lg n + X (a,vv)2A (len(a) + n lg m)) ∑ K 2 (lg n + n X a2U(A)(len(a) + n lg m))∑ K 3 n 2 lg m(1 + X a2U(A)len(a)).

Theorem B. 7 .if r 6 = 1 then 12 :

 712 len Z (a) = K n 2 lg m, which satisfies the definition of b O. Let D MVReg be the baisc multi-value register implementation defined in Algorithm 2.5, such that D MVReg sat[V state , T any ] F MVReg . The complexity of D MVReg is b ≠(n 2 lg m). Proof. Consider the following driver program (introduced in Section 4.1) defined for any positive n and m such that m°n n ∏ 1 and (m °n) mod n = 0: 1: procedure inflate(n, m) send r (mid r,1 ) 11: deliver 1 (mid r,1 ) B.2. METADATA OVERHEAD OF SPECIFIC IMPLEMENTATIONS and deliver function, the two mutators of entries in A, 8(a, vv) 2 A, r 0 2 ReplicaID.vv(r 0 ) ∑ m. Let U(A) = {a | 9(a, vv) 2 A}. The definition of deliver ensures that every a is unique in A, i.e., |U(A)| = |A|. Then, for some constants K 1 , K 2 and K 3 , state (r, A) can be encoded with a string no bigger than: len(r, A) ∑ K 1 (lg n + X (a,vv)2A (len(a) + n lg m)) ∑ K 2 (lg n + X a2U(A) (len(a) + n lg m)) ∑ K 3 n lg m(1 + X a2U(A) len(a)).

1 :. read e § 5 :C. 2

 152 len Z (a) = K n lg m, which satisfies the definition of b O. Theorem B.10 (Originally, Theorem 3.4). Let D § MVReg be the multi-value register implementation defined in Algorithm 3.4, such that D § MVReg sat[V state , T any ] F MVReg . The complexity of D § MVReg is b £(n lg m). Proof. By Theorem B.9 and Theorem 4.5. Theorem B.11 (Originally, Theorem 3.6). Let D RWSet be the basic add-wins set implementation defined in Algorithm 3.5, assuming that D RWSet sat[V state , T any ] F RWSet . The complexity of D RWSet is b ≠(m lg m). Proof. Consider the following driver program (introduced in Section 4.1) defined for any positive n and m: procedure inflate(n, m) for all i 2 [2..n] do 6: do i read m+i . dummy read on other replicas to fulfill wcmo definition B.3. LOWER BOUND PROOFS (b) rem(a) operation performed at replica 2, either followed by, or concurrent with, add(a) at replica 1, depending whether a 2 AE or not. If rem(a) and add(a) in (b) are sequential, then a 2 v according to F RWSet , since all rem(a) in the context are covered by some add(a). Otherwise, a 6 2 v, since there is a rem(a) that is not covered by any add(a). Therefore, by lifting the argument to all elements, v = AE. (ii) We have N init = N AE because exp(AE) does not send any messages. Besides, R init [1 7 ! R AE (1)] and R AE can differ only in the states of the replicas 2..n. These cannot influence the run of test(r), since it performs execution steps on replica 1 only. Theorem B.18 (Originally, Theorem 4.3). If D RWSet sat[V state , T any ] F RWSet , then the complexity of D RWSet is b ≠(m).Proof. Given n 0 , m 0 , we pick n = max{n 0 , 3} and some m ∏ n such that m is a multiple of 4. Take the experiment family (Q, n, m, C, e, readback) given by Lemma 4.4. For any AE 2 Q, C AE .rval(e AE ) = ; by F RWSet applied to the witness, which can be encoded with a constant length. Using Lemma 4.1 and m ∏ n 2 , for some constant K we get: wcmo(D RWSet , n, m) ∏ blg | §| |Q|c / (max AE2Q len(C AE .rval(e AE ))) ∏ K m. B.3.3 Last-Writer-Wins Set Lemma B.2 (Originally, Lemma 4.5). If D LWWSet sat[V state , T any ] F LWWSet , n ∏ 2 and m ∏ n is such that m is a multiple of 2n °2, then the tuple (Q, n, m, C, e, readback) in Table 4.4 is an experiment family. Proof. Some parts of the proof are straightforward, such as checking that n and m match the number of replicas/updates, and that the driver programs use message identifiers and timestamps correctly. The only nontrivial obligation is to prove readback(state(e AE)) = AE. Let (R AE , N AE ) = final(C AE ). Then AE(r) (i) = result(D LWWSet , (R 0 , N 0 ), (init; exp(AE); test(r))) = result(D LWWSet , (R AE , N AE ), test(r)) (ii) = result(D LWWSet , (R init [1 7 ! R AE (1)], N init ), test(r)) = readback(R AE (1))(r) = readback(state(e AE ))(r),where:le traitement des livraisons asynchrones des mises à jour et des échecs.Deux abstractions complémentaires ont été proposées pour atténuer les problèmes de la réplication dans le cas de la cohérence à terme. Ces abstractions prennent en compte la complexité de la réplication asynchrone et les échecs derrière une interface fiable, à savoir une interface avec un comportement bien défini, qui est garanti quel que soit le comportement de l'infrastructure sous-jacente. Tout d'abord, les Types de Données Répliquées (RDTs) exposent les données de la base de données comme des objets classifiés possédant des méthodes de haut niveau. Elle se basent sur le type et la sémantique de la méthode pour assurer la convergence vers un état raisonnable [34, 92]. La base de données comprenant des RDTs contient des objets de types tels que le compteur, l'ensemble où le registre, possédant des méthodes de l'écriture et de mise à jour, tel que l'incrémentation pour le compteur ou l'ajout et la suppression pour l'ensemble. Un type de données encapsule la réplication et s'oocupe de la résolution des mises à jour simultanées en se basant sur la sémantique définie. Ainsi, les RDTs fournissent un fiable bloc de construction. Deuxièmement, la cohérence transactionnelle causale offre un ordonnancement partiel des mises à jour et des garanties de visibilité indépendantes des frontières de l'objet [4, 66, 67]. Officieusement, sous cohérence causale, tous les processus de l'application observent un ensemble non-décroissant de mises à jour, qui comprend ses propres mises à jour, dans l'ordre respectant les causalités entre les opérations. Les applications sont protégées des violations de causalité, par exemple d'une situation où la méthode d'écriture pourrait observer la mise à jour de b, mais ne pourrait pas mettre à jour a sur laquelle b se baserait. Ces abstractions fiables facilitent la programmation de la base de données cohérente à terme, mais entrennent un cout de stockage et un surcoût des métadonnées dans le réseau causé par l'information de la causalité et l'historique des opérations. Cette thèse étudie la conception des fiables RDTs et des fiables algorithmes de cohérence causale utilisant des métadonnées minimisées, ainsi que les limites et les compromis de leur espace de conception. Contributions. Cette thèse apporte deux contributions dans le domaine de la fiabilité des algorithmes de la cohérence à terme: une étude de l'espace optimale des implémentations des RDTs, y compris des mises en oeuvre optimisées et des résultats d'impossibilité, ainsi que la conception des bases de données causalement cohérentes incluant les RDTs pour les applications côté client. Nous présentons un aperçu de ces résultats dans le reste de ce résumé. C.2. UNE BASE DE DONNÉES CAUSALEMENT COHÉRENTE POUR LES APPLICATIONS COTÉ CLIENT vecteurs de versions. De plus, cela est (asymptotiquement) la moins cher des réalisations connues pour toutes sortes d'ensembles. La borne inférieure pour l'ensemble suppression-gagnante montre, que telle optimalisation n'est pas possible dans le cas d'une sémantique suppression-gagnante. La question si cela est possible pour un ensemble dernier-écrivain-gagnant reste ouverte. Le type de registre possède aussi de nombreuses variantes, qui eux aussi sont inégaux en terme de coût des métadonnées encouru. Une variante du registre de type dernier-écrivaingagnant utilise le horodatage pour arbitrer les attributions simultanées, tandis que le registre multi-valeur identifie les valeurs de tous les conflits d'écriture pour les présenter à l'application. La réalisation existante du registre dernier-écrivain-gagnant est caractérisée par un surcoût négligeable et est considérée comme optimale. Au contraire, la réalisation actuelle du registre à valeurs multiples possède un surcoût important en terme de carré de nombre de répliques due à un traitement inefficace des écritures simultanées de la même valeur. Nos optimisations atténuent la composante carré, et atteignent la surcharge de métadonnées asymptotiquement optimal en utilisant une règles de fusion pour les vecteurs de versions. Une base de données causalement cohérente pour les applications coté client Dans la deuxième partie de la thèse, nous étudions le problème de la fourniture des garanties étendues de la cohérence (causale), à travers les frontières de l'objet et au-delà de l'infrastructure côté serveur, pour les applications côté client. Nous considérons le problème de la réplication côté client. La technologie actuelle est peu adaptée pour supporter le partage de données envers une large zone dans des applications côté client, comme par exemple dans le navigateur ou les applications mobiles. Les développeurs d'applications font recours à la mémoire cache et aux tampons, afin d'éviter des lents, coûteux et parfois indisponible allers-retours au centre de données. Cependant, ils ne sont pas capables de résoudre les problèmes du système tels que la tolérance aux pannes, ou les garanties de cohérence / session [34, 96]. Les systèmes côté client assurent seulement quelques-unes des propriétés souhaitées. Notamment, ils assurent des garanties de cohérence limitées (seulement à la granularité d'un objet unique ou celle d'une petite base de données), ne tolèrent pas les défaillance et/ou ne peuvent pas intégrer un grand nombre de périphérique clients. Notre thèse est que le système doit être responsable d'assurer un accès correct et évolutif à la base de données aux applications côté client. Il doit répondre aux exigences (parfois contradictoires) de la cohérence, de la disponibilité et de la convergence [68] au moins aussi bien que les systèmes de géo-réplication récents. Dans ces conditions, le modèle de cohérence le plus fort est une variante de la cohérence causale pour les objets de RDTs. La possibilité d'offrir des milliers ou mêmes des millions de répliques côté client avec la cohérence causale conteste les hypothèses standard. Pour tracer avec précision et par client la causalité dans et entre les objets il faut utiliser un nombre inacceptable de métadonnées. De l'autre côté, un management compact des métadonnées du côté serveur possède une tolérance aux pannes plus faible. De plus, une réplication complète dans un grand nombre de dispositifs pauvres en ressources serait inacceptable [18]. Mais aussi une réplication partielle des données et des métadonnées pourrait être la source d'indisponibilité ou des anomalies de livraison de messages. Par ailleurs, il n'est pas possible de supposer, comme dans nombreux systèmes antérieurs, que la tolérance aux pannes et la cohérence sont résolues, seulement parce que l'application est localisé à l'intérier d'un centre de données (DC), ou parce qu'elle possède une session collante avec un seule DC [13, 96]. Dans cette partie de la thèse, nous abordons les défis mentionnés. Nous présentons les algorithmes, la conception et l'évaluation du SwiftCloud -la première base de données d'objets distribuée conçue pour un grand nombre de répliques. Elle assure d'une façon efficace un accès constant, disponible, et convergent aux noeuds clients, supporte les défaillances et consume peut de métadonnées. Pour atteindre ce but, le SwiftCloud utilise une topologie de client-serveur flexible, et découple l'écriture et la lecture. Le client écrit rapidement dans le cache local, et lit dans le passé (aussi rapidement) les données qui sont compatibles, mais parfois altérées.

Figure C. 1

 1 Figure C.1 présente le modèle de notre système. Une infrastructure "cloud" relie un petit ensemble (par exemple, une dizaine) de centres de données géo-réplicatifs, ainsi qu'un grand nombre (des milliers) de clients. Un DC dispose d'abondantes ressources de calcul, de stockage et de réseau. De même que Sovran et al. [95], nous faisons abstractions d'un DC comme un processus séquentiel hébergeant des répliques complètes de la base de données. 1 Le DC

C. 2 .

 2 UNE BASE DE DONNÉES CAUSALEMENT COHÉRENTE POUR LES APPLICATIONS COTÉ CLIENT

  B) Continuation depuis C.2A vers un état à risque.

  Lecture dans le passé: continuation depuis C.2A vers un état prudent.

Figure C. 2 :

 2 Figure C.2: L'exemple d'évolution d'états de deux DCs et un client. x et y sont les objets; boîte = mise à jour; flèche = dépendance causale (un texte facultatif indique la source de la dépendance); boîte pointillée = versions de bases de données nommées /les états.

C. 2 . 2 . 4

 224 taille des métadonnées requis, nous nous s'appuyons sur les répliques complètes des DCs pour gérer des versions sans lacune pour le client. Donné un ensemble d'intérêt des objets intéressants le client, son état initiale consiste d'une projection d'un état du DC envers l'ensemble. Celui-ci est un état causalement consistent comme démontré dans les sections précédentes. L'état client peut change à cause d'une mise à jour générée par le client lui même, notée une mise à jour interne, ou à cause d'une mise à jour reçue du DC, notée externe. Evidement, une mise à jour interne maintient la cohérence causale. Si une mise à jour externe arrive, sans lacune, du même DC que la précédente, elle maintient aussi la cohérence causale. Plus formellement, nous considérons un état récent du DC, que nous nommons l'état de base du client, noté V DC . L'ensemble d'intérêt du client C est noté O µ x, y,.... L'état du client, noté V C est limité à ces objets. Il se compose de deux parties. La première est la projection de la version de base V DC sur l'ensemble d'intérêt, noté V DC | O . L'autre est le log des mises à jour interne, noté U C . L'état client est leur fusion V C = V DC | O © U C | O . Dans le cas de manque de mémoire cache, le client ajoute les objets manquant à l'ensemble d'intérêt et récupère les objets à partir de la version de base V DC , étendant ainsi la projection. La version de base V DC est une version monotone, non-décroissante et causale (elle peut être un peu retardée en comparaison avec la version actuelle du DC en raison des délais de propagation). Par induction, les mises à jour internes peuvent dépendre causalement uniquement des mises à jour internes ou des mises à jour prises de la version de base. Ainsi, une version hypothétique complète V DC © U C serrait causalement cohérente. Sa projection est équivalente au état client : (V DC ©U C )| O = V DC | O ©U C | O = V C . Cette approche assure la disponibilité partielle. Si une version se trouve dans la mémoire cache, elle a la garantie d'être causalement cohérente, bien qu'elle peut être une peu viciée. Si la version ne se trouve pas dans la mémoire cache, le DC retourne immédiatement une version Protocoles avec les métadonnées découplées et délimitées Grace à la communication canalisante entre DCs et la "lecture dans le passé," notre implémentation de l'approche SwiftCloud peut utiliser un nouveau modèle de métadonnées qui découple deux aspect. Il trace la causalité dans le but de renforcer la cohérence, en utilisant des petits vecteurs attribués dans la tâche de fond par le DCs, et il identifie de manière unique les mises à jour à protéger des doublures causé par la retransmission, 5 en utilisant un horodatage scalaire assigné par le client. En conséquence, une mise à jour possède deux types d'estampilles : une estampille assignée par un simple client et une estampille assignée par un ou plus DCs. Avant d'être délivrées dans un DC, les mises à jour ne possèdent pas d'estampille ; elle posséderont une par la suite ; une mise à jour peut avoir plus qu'une estampille dans le cas d'une livraison aux plusieurs DCs (pendant le basculement, Section 7.1.3.1). Une fois l'estampille d'un DC assignée, une mise à jour peut être efficacement référencée en utilisant les vecteurs de causalité. Un vecteur de causalité associe l'identifiant du DC à un nombre naturel k, en indiquant les dépendances des k mises à jour transmises à ce DC. La taille des métadonnées causales est petite et délimitée, étant donné qu'elles possèdent une entré pour chaque DC. De plus, grace à l'extention décrite dans cette thèse, un DC peut élaguer ces logs indépendamment de la disponibilité client, en assurant la sûreté en stockant un sommaire local des mises à jour délivrées.

Table 1 .

 1 1: Summary of metadata overhead results for different data types. Underlined implementations are optimal; n and m are the numbers of, respectively, replicas and updates; b ≠, b O, b £ are lower, upper and tight bounds on metadata, respectively.

  Val ø ), which defines the client interface as a set of methods Op ø (ranged over by o) and a set of return values Val ø (ranged over by v). For simplicity, all methods share the same return value domain, with a special value ? 2 Val ø for methods that do not return any value. Without loss of generality, we assume that every type has a single side-effect free query method read 2 Op ø , which returns the value of an object. The remaining methods Op ø \ {read} are called update methods and are assumed to have side effects.

ø , For example, a counter type, noted Ctr, has a signature (Op Ctr , Val Ctr ). Methods Op Ctr = {inc, read}, respectively, increment and read the counter value. read returns a natural number, Val Ctr = N [ {?}. Another example is an integer register type, noted LWWReg, with methods Op LWWReg = {write(a) | a 2 Z} [ {read}. The latter returns integer values, Val Ctr = Z [ {?}. Note that we model arguments of a method as part of a method name for simplicity.

  The timestamp can be used by the implementation to identify an update or to arbitrate updates. The outcome is a tuple (ae 0 , v) = do(ae, o, t), where ae 0 2 ß becomes the new object state at the replica invoking the operation, and the operation return value v 2 Val ø is presented to the client. Note that for queries ae 0 = ae. Since method invocation is a local step, the client is never blocked by an external network or replica failure.Replication functions are applied by a database replica non-deterministically, when the network receives a message or permits a transmission, and the replica decides to perform a replication step. To send a message, a replica applies send(ae) with the local state of the object ae.

The client can perform a method o 2 Op ø on a selected replica at any time. Method invocation causes the replica to perform do(ae, o, t) on the current object state ae 2 ß, with timestamp t 2 CHAPTER 2. REPLICATED DATA TYPES Timestamp.

Lamport clock Algorithm 2.5

  Basic state-based implementation of multi-value integer register (MVReg).

	1: ß = ReplicaID £ P(Z £ (ReplicaID 7 ! N 0 )) M = P(Z £ (ReplicaID 7 ! N 0 )) 2: initialize(r i ) : (r, A)
	3: 4: 5: do(read, t o ) : V let r = r i let A = ; . non-overwritten entries: set of pairs (a, vv) with values and version vectors . replica ID
	6: 7: do(write(a), t o ) let V = {a | 9vv : (a, vv) 2 A} 8: let vv = F {vv 0 | (_, vv 0 ) 2 A} 9: let vv 0 = vv[r 7 ! (vv(r) + 1)] 10: A √ {(a, vv 0 )} 11: send() : A m	. all stored values are the latest concurrent writes . compute entry-wise maximum of known vectors . increment own entry to dominate other vectors . replace the current value with the new entry
	12: 13: deliver(A m ) let A m = A
	14:

  Event is a finite subset of Event; op : E ! Op ø defines type of operations in E; vis µ E £ E is an acyclic visibility relation; ar µ E £ E is a strict total arbitration relation. We note Context ø the domain of contexts for type ø that obey this structure. For abstract execution A, the context of event e 2 A.E can be extracted with context function: context(A, e) = (A.op(e), E e , A.op| E e , A.vis| E e , A.ar| E e ), where E e = vis °1(e) are the events in E visible to e, and R| E e indicates restriction of relation R to the elements of E e . For example, the context of the underlined blue read in Figure 2.8A is indicated with a blue dashed line. (Specification). A specification for a data type ø is a partial function F ø : Context ø * Val ø that given an operation context for type ø specifies a return value.

	Definition 2.3 (Operation Context). An operation context of operation o 2 Op ø for a data type ø is a tuple L = (o, E, op,vis,ar), where: E µ Definition 2.4 By applying the specification to every event in an abstract execution, we obtain a correctness
	condition for abstract executions.
	Definition 2.5 (Correct Abstract Execution). An abstract execution A of type ø satisfies a
	specification F ø , noted A |= F ø , if the return value of every operation in A is the one computed by the specification applied to the context of that operation: 8e 2 A.E : A.rval(e) = F ø (context(A, e)).
	Specification defines the expected value for a given context in a deterministic way. The non-
	determinism of a distributed execution is encapsulated in client invocations, and visibility and
	arbitration relations of an abstract execution.
	Note that Definition 2.5 embeds the safety aspect of the convergence requirement, which we
	call confluence. Confluence requires that if two operations observe the same set of operations,
	they must return the same value. Deterministic specification embodies this condition, and covers
	implementation category-specific conditions, such as semilattice of states or commutativity of
	operations. For eventual convergence, it remains to require the liveness aspect of convergence,
	which we call eventual visibility of operations. Formally, we state it as a restriction on an
	abstract execution A, that requires that operations are not infinitely invisible [34]:
	(2.4)	8e 2 A.E :6 9 infinitely many f 2 A.E s.t. f 6 vis °°! e,
	where e

vis °°! f indicates that e and f are ordered by vis relation.

CHAPTER 2. REPLICATED DATA TYPES

Table 2

 2

	Op-based	State-based

.1.

State and message metadata size is typically low for op-based implementations, and much higher for state-based ones (compare, for instance, Algorithm 2.2 with Algorithm 2.3). Both of

  r 0 , k) 2 A} . return values of active instances

9: do(add(a), t o ) 10:

  r, k)} . create and add a new instance of a

12: do(rem(a), t o ) 13:

  vv, A)

	4: 5: 6: 7: do(read, t o ) : V let r = r i let vv = ∏s.0 let A = ; 8: let V = {a | 9r 0 , k : (a, r 0 , k) 2 A} . active element instances: triples (element a, replica r 0 , counter k 0 ) . replica ID . vector summarizing timestamps known to this replica . return values of active instances 9: do(add(a), t o )
	10: 11: 12:	. collect all instances of a created by local replica . add new instance of a and replace local instances if any let D = {(a, r, k) | 9k : (a, r, k) 2 A} A √ A [ {(a, r, vv(r) + 1)} \ D

vv √ vv[r 7 ! vv(r) + 1]

. increment own entry in the vector 13: do(rem(a), t o )

14:

  Incorrect state-based implementation of multi-value integer register (MVReg).

		3.2. OPTIMIZING IMPLEMENTATIONS
	Algorithm 3.3 2: . . . 3: deliver(A m )	. all other methods are identical to Algorithm 2.5

1: ß = ReplicaID £ P(Z £ (ReplicaID 7 ! N 0 )) M = P(Z £ (ReplicaID 7 ! N 0 )) 4:

  ^(6 9t 0 : (a, t 0 ) 2 T)} . elements with add instance but no rem ins.A √ A [ {(a, t o )} . add a new add instance for a if there is no remove instance to cover

	7: do(add(a), t o )
	8: 9: 10:	let D = {(a, t) 2 T} if D = ; then	. identify existing rem instances for element a
	11:	else
	12:

5: do(read, t o ) : V 6: let V = {a | (9t : (a, t) 2 A) A √ A [ D . otherwise, turn remove instances into add instances. . . 13: T √ T \ D . . . . and discard remove instances 14: do(rem(a), t o ) 15: T √ T [ {(a, t o )} . add a new remove instance for element a 16: send() : (A m , T m ) 17:

  [START_REF]Data Types in Riak[END_REF] we answer it only partially. It appears difficult to apply optimizations inspired by the add-wins, since different removed elements may have different timestamps. If this overhead is indeed unavoidable, one could only resort to background garbage collection[START_REF] Johnson | The maintenance of duplicate databases[END_REF]. State-based implementation of last-writer-wins set (LWWSet). S} . return values of elements that are present (presence flag = 1)

	CHAPTER 3. METADATA SPACE COMPLEXITY PROBLEM
	Algorithm 3.6 7: do(add(a), t o )
	8: 9: 10: 11: do(rem(a), t o ) if t o > t then let D = {(a, t, v) 2 A} A √ (A \ D) [ {(a, t o , 1}	. sanity check for new timestamp . replace the latest operation on a (if any). . . . . . . with a positive presence flag
	12: 13:	if t o > t then	. sanity check for new timestamp

1: ß = P(Z £ Timestamp £ {0, 1}) 2: M = P(Z £ Timestamp £ {0, 1}) 3: initialize(r i ) : S 4: let S = ;

. set of latest operations: triples (element a, timestamp t, presence-flag v) 5: do(read, t o ) : V 6: let V = {a | 9t : (a, t, 1) 2 let D = {(a, t, v) 2 A} . replace the latest operation on a (if any). . . 14: A √ (A \ D) [ {(a, t o , 0} . . . . with a negative presence flag 15: send() : S m 16: let S m = S 17: deliver(S m ) 18:

  AE ). The experiments are designed in such a way that the object states must be distinct across experiments, which then implies a lower bound lg | §| |Q| on the size of their encoding (i.e., the size required to encode an element of a set of cardinality |Q|). prove that the culprit states 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

	|Q| executions	readback(state(e 1 ))= 1 by F τ test phase To 4.1. PROOF TECHNIQUE e 1 :read state(e 1 ) m updates n replicas r 1 r 2 r n … e α :read state(e α ) r 1 r 2 r n … … e |Q| :read state(e |Q| ) r 1 r 2 r n … … C 1 C α C |Q| |Q| strings execution phase experiment family (Definition 4.1) state size (Lemma 4.1)

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 AE of the same size, where AE 2 Q is an experiment index from some index set Q, that identifies an individual experiment. Each experiment execution in the collection ends with a distinguished culprit read event e AE 2 C AE .E, where the overhead argument applies. Culprit reads execute in the culprit object states, noted state(e AE ) provided D ø is correct. In other words, the tests demonstrate that these states are distinguishable by future steps of an execution, after the read. Formally, the tests induce a read-back function that satisfies readback(state(e AE )) = AE, relying on the data type specification. readback(state(e |Q| ))=|Q| by F τ len(state(e α )) ≥ lg |Q| readback(state(e α ))= α by F τ

Table 4 .

 4

			init	exp(α)	test(
	1
	2	inc	send mid 2,2	inc send mid 2,3
	3	mid 3,1 inc send	mid 3,2 inc send	inc

1: Experiment family (Q, n, m, C, e, readback) used in the lower bound proof for counter (Ctr).

Table 4 . 2

 42

	1
	2
	3

: Experiment family (Q, n, m, C, e, readback) used in the lower bound proof for add-wins set (AWSet).

Table 4 .

 4

					init					exp(α)	test
	1					del	add(1)	add(2)	del	add(3) del	e α read:∅	del	del	(v) read:{1,3}
	2	rem(1) send mid 2,1	rem(2) send mid 2,2	rem(3) send mid 2,3
	3	mid 3,1 send add(1) rem(3) rem(1) rem(2)	add(2) add(3)	mid 3,2 send
			configuration (R init [1→R α (1)], N init )
						(A) Concrete execution C 0 AE .
		r 1 :				add(1) add(2) add(3)	read:∅ (e α )	read:{1,3} (v)
		r 2 :	rem(1) rem(2) rem(3)
		r 3 :	rem(1) rem(2) rem(3)	add(1) add(2) add(3)
		(B) Witness abstract execution abs(C 0 AE , V

3: Experiment family (Q, n, m, C, e, readback) used in the lower bound proof for removewins set (RWSet). state ). Arrows indicate visibility. Arbitration is not shown.

Table 4 .

 4

	1
	2	add(2) 8
	3

4: Experiment family (Q, n, m, C, e, readback) used in the lower bound proof for last-writerwins set (LWWSet).

Table 4 .

 4 6: Summary of metadata overhead results for different data types. Underlined imple-

mentations are optimal; n and m, are, respectively, the number of replicas and updates in an execution; b ≠, b O, b £ are lower, upper and tight bounds on metadata, respectively.

  In previous chapters we introduced the metadata complexity problem, presented positive and impossibility results for six data types, and related them to known state-based implementations.

. . But maybe it is worth investigating the unknown, if only because the very feeling of not knowing is a painful one. Krzysztof Kieślowski Contents 5.1 Other Data Types . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 72 5.2 State-Based Optimizations Beyond Our Metric . . . . . . . . . . . . . . . . . . . . . 72 5.2.1 Background Compaction of Stable Metadata . . . . . . . . . . . . . . . . . . 73 5.2.2 Finer-Grained Optimizations . . . . . . . . . . . . . . . . . . . . . . . . . . . . 73 5.2.3 Custom Timestamps . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 74 5.3 Other Implementation Categories . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 75 5.3.1 State-Based Implementations With Smaller Messages . . . . . . . . . . . . 75 5.3.2 Optimizations Based on Topology Restrictions and Delayed Visibility . . . 76 5.3.3 Replicated File Systems . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 77 5.4 Lower Bound Proofs in Distributed Computing . . . . . . . . . . . . . . . . . . . . . 78

  bined both state-and op-based replication, allowing updates to be delivered either as individual operations, under causal delivery, or as complete state, under any delivery condition. Mukund et al. [72] discuss a follow-up on this idea, and extends it to support a new, weaker delivery network specification for op-based messages, k-causal delivery. Under k-causal delivery, a message can be delivered if the receiver has not missed more than k °1 causal dependencies from the sender. Mukund et al. generalize our solution from causal delivery to k-causal delivery, by

  .2 compares SwiftCloud with server-side replication for all workloads. Client-side replication is a mixed blessing: it lets client replicas absorb read requests that would otherwise reach the DC, but also puts extra load of maintaining client replicas on DCs. A detailed analysis reveals the sources of this error margin. The default Zipfian object access distribution of YCSB increases the fraction of local accesses due to added "global" locality (up to 82 % local accesses for target 80 % of workload session locality). On the other hand, low locality workload decreases amount of local accesses, due to magnified imperfections of the LRU cache eviction algorithm (down to 34 % local accesses for target 40 %).

		YCSB A (50% updates)	YCSB B (5% updates)
	response time [ms]	1 10 100 1000 1 10 100 1000					Zipfian distrib. uniform distrib.
		1250	2500	5000	5000	10000	20000
				throughput [txn/s]
		server replicas only, 70th percentile of response time (exp. local)
		client replicas, 70th percentile of response time (exp. local)
		server replicas only, 95th percentile of respone time (remote)
		client replicas, 95th percentile of respone time (remote)
	SwiftCloud's client-side replication consistently improves throughput for high-locality workloads,
	by 7% up to 128%. It is especially beneficial to read-heavy workloads. In contrast, low-locality
	workloads show no clear trend; depending on the workload, throughput either increases by up to

3

  Figure 8.4: Maximum system throughput for a variable number of client and DC replicas.

	max. throughput [txn/s]	0 5000 10000 15000 20000 25000		uniform	YCSB B, uniform	• • • •	SocialApp refresh rate 1s refresh rate 10s 3 DC replicas 1 DC replica
			500	1500	2500 500	1500	2500 500	1500	2500
						#client replicas
	notification metadata [B]	10 100 1K 10K 100K	YCSB A, uniform max. notification data 500 1500 2500 500 YCSB B, uniform max. notification data 1500 2500 500 SocialApp w/stats max. notification data 1500 2500
						#client replicas
				1 DC replica		SwiftCloud metadata
				3 DC replicas		Safe But Fat metadata (Depot*)

  . Quand la mémoire cache est utilisée, le temps de réponse est d'un ordre de grandeur deux fois plus faible que pour les protocoles basés sur le serveur avec des garanties de disponibilité similaires. Avec trois serveurs de DC, le système peut accueillir des milliers de répliques client. La taille des métadonnées ne dépend pas du nombre des clients, du nombre des panne ou de la taille de la base de données. Elle augmente légèrement seulement avec le nombre des DCs: en moyenne, 15 octets de métadonnées par mise à jour, avec 3 DCs, comparé aux kilooctets des algorithmes précédents aves les mêmes garanties de sûreté. Le débit est comparable à la réplication côté serveur pour une faible localité de charge, et amélioré pour une haute localité. Quand un DC tombe en panne, ses clients basculent vers un nouveau DC dans les 1000 ms, ainsi que restent cohérents. Sous conditions normales, la version 2-stable est la cause de moins d'1% de lectures viciées.

An alternative would be to coordinate conflict resolution. However, this results in an unavailable protocol, similar to approach (A).

This assumption matches many models from the literature[START_REF] Baquero | Using structural characteristics for autonomous operation[END_REF][START_REF] Roh | Replicated Abstract Data Types: Building blocks for collaborative applications[END_REF][START_REF] Shapiro | Conflict-free replicated data types[END_REF][START_REF] Gene | Efficient solutions to the replicated log and dictionary problems[END_REF], and recent industrial implementations of RDTs[START_REF]Data Types in Riak[END_REF][START_REF] Bonér | Server-managed CRDTs based on Akka[END_REF]. It also allows us to express, in Part III, a system with independent object implementations yet offering cross-object consistency.

2.2. SYSTEM MODEL AND OBJECT IMPLEMENTATION

A broadcast message produced by send may reach more than one replica, but the implementation can emulate a unicast primitive on top of it, by encoding information about the intended recipient inside a message (at the expense of unnecessary message complexity). None of our results is affected by this simplification.

In the general case more than one abstract execution may characterize a single concrete execution.

2.3. SPECIFICATION OF INTENDED BEHAVIOR

Alternatively, an extended specification could also return a multiset of values to indicate to the application when the same value was written concurrently[START_REF] Parker | Detection of mutual inconsistency in distributed systems[END_REF][START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF].

This translates into no creation property of network links, in a distributed systems parlance[START_REF] Cachin | Introduction to Reliable and Secure Distributed Programming[END_REF].

2.4. EXECUTION MODEL AND CORRECTNESS

More precisely, we consider only prefix-closed specifications that restrict delivery events or timestamps only.

Note that this is different from how the multi-value register (Section 2.2.3) uses vectors, comparing vectors assigned to events to check if events are concurrent.

Even for a variant of multi-value register that returns a multi-set in this case[START_REF] Shapiro | A comprehensive study of Convergent and Commutative Replicated Data Types[END_REF], the overhead would be n 2 .

If all values stored in a set have a constant size, bound exists; however, metadata overhead metric would become an absolute metadata metric in this case, and the m factor would be bounded by the (constant) cardinality of the domain of values.

Minor optimizations are possible. As they do not change the worst-case behavior, we omit them in Algorithm 3.5.

Moreover, all executions generated by our driver programs respect causal timestamps T ct . Therefore, the results hold also under a stronger network specification.

Similarly, there is no other safe known way of removing an entry of a replica from the version vector in our optimizations from Section 3.2.1.

In terms our formal model from Chapter 2, this means that there are two kinds of send and deliver events, and a new visibility witness, V 6 = V state , treats each kind differently.

The implementation of Almeida et al.[START_REF] Almeida | Efficient state-based CRDTs by delta-mutation[END_REF], described earlier, demonstrates that storing indidividual instances is not necessary to generate deltas.

Somewhat similar optimization techniques have been exploited in the context of log-based protocols without topology restrictions: Conway et al.[START_REF] Conway | Edelweiss: Automatic storage reclamation for distributed programming[END_REF] propose automated synthesis of optimized general purpose log-based distributed applications, whereas Baquero et al.[START_REF] Baquero | Making operation-based CRDTs operation-based[END_REF] propose manually optimized op-based RDT implementations.

We refer to prior work for the somewhat orthogonal issues of ensuring parallelism and fault-tolerance within a DC, using sharding and replication, respectively[START_REF] Almeida | ChainReaction: a causal+ consistent datastore based on Chain Replication[END_REF] 46, 66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF]. We discuss them in Chapter 9.

Although auxiliary client-to-client communication is viable in some environments, we do not consider it here.

This subsumes the well-known session guarantees, such as read-your-writes or monotonic reads[START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF][START_REF] Terry | Session guarantees for weakly consistent replicated data[END_REF].

Unlike COPS or Eiger, we consider interactive transactions, i.e., accessed objects do not need to be predefined.

Note that David can still perform update. However, the system as a whole does not converge.

The DC may accept to store client's updates to improve durability, but it cannot deliver them or offer notifications.

The increased number of concurrent updates that this causes is not a big problem, thanks to RDTs.

This is easily extended to multiple updates, by assigning the same timestamp to all the updates of the same transaction, ensuring the all-or-nothing property of causal transactions[START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF].

Alternatively, the client can ask for invalidations instead, trading responsiveness for lower bandwidth utilization and higher DC throughput[START_REF] Belaramani | PRACTI replication[END_REF][START_REF] Mahajan | Depot: Cloud storage with minimal trust[END_REF].

SocialApp implements WaltSocial's user registration operation partially, since user uniqueness with custom user names requires additional support for strong consistency.

In practice, this could be an issue, however, a recent work of Briquemont et al.[START_REF] Briquemont | Optimising client-side geo-replication with partially replicated data structures[END_REF] demonstrates how to implement object sharding in a SwiftCloud-like system.

Other work treat replication in a DC as an orthogonal aspect, and assume linearizable object in a DC[46, 66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF].

More precisely, Walter offers a Parallel/Non-Monotonic variant of Snapshot Isolation[START_REF] Saeida | Non-Monotonic Snapshot Isolation: scalable and strong consistency for geo-replicated transactional systems[END_REF][START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF].

Server-side execution can be also beneficial for transactions that involve a lot of data to produce a small result.

If the multi-value register would return a multi-set of values, it would also require a maximum of lg n extra overhead.

Les traveaux antérieurs abordent les questions un peu orthogonales du parallélisme et de la tolérance aux pannes dans un DC[START_REF] Almeida | ChainReaction: a causal+ consistent datastore based on Chain Replication[END_REF] 46, 66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF].

Cela englobe aussi les garanties de sessions[START_REF] Burckhardt | Replicated data types: Specification, verification, optimality[END_REF].

Du point de vue de David, l'écriture reste disponible. Cependant, le système dans son ensemble ne converge pas.

L'augmentation du nombre des mises à jour concurrentes, que cela implique, n'est pas un problème grâce aux RDTs.

Causé par les pannes réseaux, le DC, les pannes du client, le basculement sur erreur.

the optimal implementation in Algorithm 2.3. We show one experiment execution in Figure 4.2, together with the visibility witness for that execution. 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 unique experiment phase, defined by exp(AE). There, replica 1 delivers exactly one message from each replica r = 2..n, selected using AE(r). An experiment concludes with the culprit read e AE on replica 1.

The role of read-back is to determine the experiment index provided the culprit state of replica 1 in some experiment. Read-back works by performing separate tests for each of the replicas r = 2..n, defined by test(r), i.e., it reads-back the index piece by piece. For example, to determine which message was sent by replica 2 during the experiment in Figure 4.2, the program test(2): reads the counter value at replica 1, getting 4; delivers the final message by replica 2 to it; and reads the counter value at replica 1 again, getting 6. By observing the difference, the program can determine which message from replica 2 was sent during the experiment:

Note that all messages delivered during the read-back were sent during the initialization phase, which is common to all executions.

Proof of Lemma 4.2. The only nontrivial obligation is to prove readback(state(e AE )) = AE. In the following steps, we first demonstrate that applying program test in the final configuration of an experiment produces the experiment index, and later that the test program does not rely on any other information than the final state of replica 1.

Let the final configuration after the experiment phase be (R AE , N AE ) = final(C AE ). Then (ii) We have N AE = N init because exp(AE) does not send any messages. Also, R AE and R init [1 7 ! R AE [START_REF]Riak distributed database[END_REF]] can differ only in the states of the replicas 2..n. These cannot influence the run of test(r), since it performs execution steps on replica 1 only.

Chapter 6

Problem Overview

What's normal anyways?

Forrest Gump by Robert Zemeckis The SwiftCloud Approach I don't have to wait to realize the good old days.

Ziggy Marley CHAPTER 7. THE SWIFTCLOUD APPROACH

In this chapter, we present the design, algorithms, and evaluation of SwiftCloud, a distributed RDT object database that addresses the challenges of client-side replication. It efficiently ensures causally consistent, available, and convergent access to high number of client replicas, tolerating failures. To achieve this, SwiftCloud uses a flexible client-server protocol. The client writes fast into the local cache, and reads in the past (also fast) data that is consistent, but occasionally stale. In Section 7.1, we present the design of SwiftCloud, based on the cloud-backed support for partial replicas. To avoid the complexity of consistent partial replication at the client side and at the scale of client-side devices, we leverage the DC's full replicas to provide a consistent view of the database to the client. The client merges this view with his own updates. To tolerate DC failures, we explore a trade-off between liveness and data freshness, and serve clients a slightly delayed, but consistent and sufficiently replicated version. In Section 7.2, we show how to realize the abstract design with concrete protocols using a form of lean and safe metadata. Thanks to funnelling communication through DCs and to "reading in the past," we can use a metadata design that decouples two aspects: causality tracking using small vectors assigned in the background by DCs, and unique identification of updates to protect from duplicates, using client-assigned scalar timestamps. This ensures that the size of metadata is small and bounded, and allows DC to prune its log independently of clients' availability.

Design

We first describe an abstract design that addresses the client-side replication challenges, starting from the failure-free case, and then, how we support DC failure. Our design demonstrates how to apply principles of causally consistent algorithms for full replication systems to build a cloud-based support for partial client replicas.

Causal Consistency at Full Data Center Replicas

Ensuring causal consistency at full geo-replicated DC replicas is a well-known problem [START_REF] Ahamad | Causal memory[END_REF]46,66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF]. Our design is primarily based on a log-based approach that stores updates in a log and transfers them incrementally, combined with checkpointing optimizations [START_REF] Belaramani | PRACTI replication[END_REF][START_REF] Petersen | Flexible update propagation for weakly consistent replication[END_REF], where the system occasionally stores and transmits the state of an object, called checkpoint, in place of the complete log. We focus on the log-based angle, and discuss checkpoints only where relevant.

A database version is made of any subset of updates, noted U, ordered by causality. A version maps object identifiers to object state, by applying the relevant subsequence of the updates log. The value of an object in a version is exposed via the read API. Our log-based implementation stores the state in log chunks, one per object, ordered in a linear extension of their causal order.

We say that a version U has a causal gap, or is inconsistent if it is not causally-closed, i.e., if 9u, u 0 : u ! u 0 ^u 6 2 U ^u0 2 U. As we illustrate shortly, reading from an inconsistent version CHAPTER 7. THE SWIFTCLOUD APPROACH

Pruning the Client's Log

Managing the log at a client is comparatively simpler. A client logs his own updates U C , which may include updates to object that is currently out of his interest set -this enables the client to read its own updates, and to propagate them lazily to a DC when connected and convenient. An update u can be discarded as soon as it appears in K-stable base version vector vv K DC , i.e., when the client becomes dependent on the presence of u at a DC.

Chapter 8

Experimental Evaluation

If you torture the data enough, nature will always confess. YCSB [START_REF] Cooper | Benchmarking cloud serving systems with YCSB[END_REF] SocialApp à la WaltSocial [START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF] Type

Ronald H. Coase

Prototype and Applications

SwiftCloud, and the benchmark applications are implemented in Java. SwiftCloud uses Berke-leyDB for durable storage (turned off in the present experiments), a custom RPC implementation, and Kryo for data marshalling [2]. A client cache has a fixed size and uses an LRU eviction policy; more elaborate approaches, such as object prefetching [START_REF] Belaramani | PRACTI replication[END_REF], are feasible. Applications can use predefined RDT types, as well as define their own op-based implementations.

Along the lines of previous studies of causally consistent systems [START_REF] Almeida | ChainReaction: a causal+ consistent datastore based on Chain Replication[END_REF][START_REF] Bailis | Bolt-on causal consistency[END_REF][START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF][START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF], we use two different benchmarks, YCSB and SocialApp, summarized in Table 8.1.

Yahoo! Cloud Serving Benchmark (YCSB) [START_REF] Cooper | Benchmarking cloud serving systems with YCSB[END_REF] serves as a kind of micro-benchmark, with simple requirements, measuring baseline costs and specific system properties in isolation. It has a simple key-field-value object model, implemented as a LWW map RDT (a generalization of LWW set), using a default payload size of ten fields of 100 bytes each. YCSB issues transactions of applicability to client-side applications, comparing them to SwiftCloud.

Approaches

Several recent geo-replication systems scale-out by sharding [START_REF] Almeida | ChainReaction: a causal+ consistent datastore based on Chain Replication[END_REF][START_REF] Bailis | Bolt-on causal consistency[END_REF]46,66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF], which is a specific case of partial replication, where full replica is split into disjoint shards. Each shard can be replicated by a different server within a DC to increase overall throughput. However, shards require additional metadata and communication to ensure consistency.

COPS-GT.

COPS-GT, a variant of COPS [66] Later publications show that using a complete dependency graph is costly [START_REF] Almeida | ChainReaction: a causal+ consistent datastore based on Chain Replication[END_REF][START_REF] Bailis | Bolt-on causal consistency[END_REF]46,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF].

Bolt-on. Bailis et al. [START_REF] Bailis | Bolt-on causal consistency[END_REF] reduce the number of dependencies by forcing application to provide (fewer) explicit causal dependencies, like in a variant of Lazy Replication. They build a weaker variant of causal consistency with a safety layer on top of a scalable eventually consistent sharded database, e.g., Cassandra [START_REF] Lakshman | Cassandra, a decentralized structured storage system[END_REF]. A background process computes causal cut of the database, avoiding races at the underlying database.

Eiger. Eiger [START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF] We presented the design of SwiftCloud, the system that offers client-side apps a causallyconsistent, available and convergent local access to partial database replica, i.e., an interface comparable to server-side systems. SwiftCloud relies on op-based RDT implementations, i.e., objects that transfer individual operations via log-exchange protocol. The main challenge was the design of a log-exchange protocol that ensures causal consistency and supports RDTs.

Experimental evaluation demonstrates that our fault-tolerant cloud-backed support for partial client replicas is able to provide immediate response on reads and updates on local objects, and to maintain consistency and throughput of a server-side replication system. Our protocols handle failures nearly transparently, at the only cost of approx. 1 % of stale reads.

SwiftCloud leverages metadata that separates causality tracking from update identification [START_REF] Ladin | Lazy replication: Exploiting the semantics of distributed services[END_REF]. Our experiments demonstrated that with this metadata, a configuration with 1-3 DCs (servers) scales up safely to thousands of concurrent clients with metadata size of 15 bytes per update, independent of the number of clients (both available and unavailable ones).

SwiftCloud's design leverages a common principle that helps to achieve several goals: client buffering and controlled staleness can absorb the cost of scalability, availability, and consistency.

Limitations and Perspectives

Several aspects remain open for improvements and investigation.

Open metadata optimization problems. Some questions regarding the metadata overhead remained open. We would like to take into account a wider range of implementation categories in our metadata complexity analysis and optimizations. We also wish to improve lower bounds,

LIMITATIONS AND PERSPECTIVES

or implementations, for costly data types that use arbitration to resolve concurrency, such as the LWW set (Algorithm 3.6) or the list (Section 5.1). For example, several existing list implementations suffer from high metadata size [START_REF] Ellis | Concurrency control in groupware systems[END_REF][START_REF] Nédelec | LSEQ: An adaptive structure for sequences in distributed collaborative editing[END_REF][START_REF] Oster | Data consistency for P2P collaborative editing[END_REF][START_REF] Preguiça | A commutative replicated data type for cooperative editing[END_REF][START_REF] Roh | Replicated Abstract Data Types: Building blocks for collaborative applications[END_REF][START_REF] Weiss | Logoot: a scalable optimistic replication algorithm for collaborative editing on P2P networks[END_REF], in the order of the number of updates. We conjecture this is necessary in a peer-to-peer communication topology, and avoidable for restricted topologies. To this end, we study variants of the list specification.

Optimized composite objects, object shards. We witness growing importance of object composition techniques [START_REF] Brown | Riak DT map: A composable, convergent replicated dictionary[END_REF][START_REF] Gotsman | Composite replicated data types[END_REF]. Composition at the specification level can translate directly into composition of object implementations [START_REF] Gotsman | Composite replicated data types[END_REF]. However, we observe that composition permits optimizations based on composed semantic and on metadata sharing between composed objects (cf. our optimized add-wins set vs. a naive register-based implementation); object shards are a specific case of composite object implementations that could leverage similar metadata sharing optimizations. We are looking into ways of using faimiliar optimization patterns to facilitate, or automate, synthesis of efficient composite type implementations and shard implementations.

Scalable Data Center implementation in SwiftCloud. Our DC implementation of Swift-Cloud has a centralized architecture, with limited throughput scalability. We wish to extend it with decentralized sharding support [46,66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF], without overly increasing the size of metadata.

We believe this is possible to achieve by, once again, trading data freshness for performance, i.e., by managing a slightly stale consistent version at a high throughput, with small metadata [START_REF] Du | Closing the performance gap between causal consistency and eventual consistency[END_REF].

Support for untrusted clients. Database clients cannot be trusted. An adequate protection, both for data and metadata, is missing in SwiftCloud. We are considering an adaptation of Depot's support for Byzantine clients, to detect forged metadata at the DC perimeter, and to translate any Byzantine behavior into concurrency [START_REF] Mahajan | Depot: Cloud storage with minimal trust[END_REF]. Moreover, standard access control algorithms and new privacy mechanisms are required. For the latter, partial RDTs [START_REF] Briquemont | Optimising client-side geo-replication with partially replicated data structures[END_REF] or lenses [START_REF] Hofmann | Symmetric lenses[END_REF] could address the problem at the object level. They are also useful independent of security concerns, as they help to save client bandwidth and to integrate server-and client-side application logic [START_REF] Briquemont | Optimising client-side geo-replication with partially replicated data structures[END_REF][START_REF] Veeraraghavan | Fidelity-aware replication for mobile devices[END_REF].

Beyond causal transactions. Causal transactions with

RDTs is a powerful model, but some rare operations require strongly consistent objects. For instance, we observed that our SocialApp port would benefit from strongly consistent user registration. Similarly, conflict handling (even strong) at the granularity of an object is sometimes insufficient, e.g., to enforce cascading deletes.

It currently requires ad-hoc solutions in the application logic. Prior work demonstrates that combining strong and weak consistency is possible on shared data [START_REF] Li | Making geo-replicated systems fast as possible, consistent when necessary[END_REF][START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF]. We are looking into ways of integrating server-side strong consistency support, akin to the models of Burckhardt et al. [START_REF] Burckhardt | Cloud types for eventual consistency[END_REF], Ladin et al. [START_REF] Ladin | Lazy replication: Exploiting the semantics of distributed services[END_REF], Red-Blue [START_REF] Li | Making geo-replicated systems fast as possible, consistent when necessary[END_REF], or Walter [START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF]. 1 Determining adequate programming models and tools, to address such challenges of eventual consistency, is an open problem [START_REF] Alvaro | Consistency without borders[END_REF].

Appendix A Additional Material on Replicated

Data Types

A.1 Formal Network Layer Specifications

A.2 Optimized Op-Based Implementations

We present some optimized op-based implementations here to better illustrate mechanisms used by the op-based category, compared to the state-based category, and their cost.

Multi-value register. Algorithm

B.2 Metadata Overhead of Specific Implementations

Tight bound for an implementation is a conjunction of an upper and lower bound. In order to demonstrate upper bounds of a specific implementation, we need to make a general argument about every execution. On the other hand, to demonstrate an implementation of a data type is suboptimal, we need to show a lower bound higher than the upper bound of the optimal implementation, using a specific counter-example execution for a given n and m The following proofs follow this scheme. We do not prove implementation-specific lower bounds for optimal implementations, since they are covered by a general lower bound theorems from Section 4.2.

Theorem B.1. Let D AWSet be the naive add-wins set implementation defined in Algorithm 3.1,

Proof. Consider any execution C 2 ÇD AWSet É with n replicas, m ∏ n updates, and any read event e in this execution (assuming there is at least one).

Recall that the state of a set replica at the time of read is a tuple (r, k, A, T) = state(e), where r is a replica ID, k is a counter of locally assigned timestamps, A is a set of active instances (a, r 0 , k 0 ) 2 S with element a and timestamp (r 0 , k 0 ), and T is a set of timestamps of removed elements. We first establish some properties of state. By definition of add, k ∑ m for any state and timestamp in this execution. By inspection of add, rem, and deliver, the only mutators of sets A and T, an invariant (a, r 0 , k 0 ) 2 A =) (r 0 , k 0 ))

Table B.1: Standard encoding schemes for components of replica state, return values, and their cost. We denote concatenation of strings by • symbol.

new timestamps and remaining functions only transfer them between components A and T,

Then, for some constants K 1 , K 2 , K 3 and K 4 , the state can be encoded with at most:

len Z (a))

len Z (a))

len Z (a)).

For some constant K 5 , return value V = C.rval(e) of read consumes no less than:

.

By the definition of read, the set of returned values V matches the active elements, i.e., V = U(A). Thus, for some constant K the overhead is: AWSet É with n replicas, m ∏ n updates, and any read event e in this execution (assuming there is at least one).

Recall that the state of a set replica at the time of read is a tuple (r, vv, A) = state(e), where r is a replica ID, vv is a version vector (a map from replica ID to integers), and A is a set of active instances. Each instance (a, r 0 , k 0 ) 2 A is made of an element a, a replica ID r 0 , and an integer k 0 , where (r 0 , k 0 ) is a timestamp. We first establish some properties of state. By definition of add, send and deliver, the only mutators of vector vv, vector entries are bounded by number of updates, i.e., vv(r 0 ) ∑ m in any state for any replica r 0 . Similarly, by definition of mutators of the set of instances A, 8(a, r 0 , k 0 ) 2 A : k 0 ∑ m. Finally, thanks to the definition of operation add, there is at most n different instances of any element a in A (the optimization noted as coalescing adds). Let

Then, for some constants K 1 and K 2 , component A of the state can be encoded with a string no bigger than:

Therefore, for some constants K 3 and K 4 , a complete state occupies at most:

len Z (a)).

For some constant K 5 , return value V = C.rval(e) of read consumes no less than:

From the definition of read, we observe that V = U(A). Thus, for some constant K, the overhead is bounded by: Proof. Consider any execution C 2 ÇD MVReg É with n replicas, m ∏ n updates, and any read event e in this execution (assuming there is at least one).

Recall that the state of a register replica used in the read is a tuple (r, A) = state(e), where r is a replica ID and A is a set of active entries; each entry (a, vv) 2 A is made of value a and a version Clearly, the execution has n replicas and m updates. The execution is well-defined since the driver program uses message identifiers and timestamps correctly. Let (r, A) = state(e § ) be the state of replica 1 at the time of read e § , where r is a replica ID and A is a set of entries (a, vv)

with a value a and a version vector vv. By definition of implementation and program execution,

A contains n entries (1, vv 1 ), (1, vv 2 ),...,(1, vv n ) with concurrent vectors, such that each vector vv i has values: 8r 0 . vv i (r 0 ) ∏ m°n n . Therefore, for some constants K 1 , K 2 , K 3 , K 4 , K 5 , K 6 encoded state (r, A) string has a length bounded by:

By definition of read, the return value V of read e § is a singleton set, C.rval(e § ) = {1}, which can be encoded with a constant length string. 1 Therefore, for some constant K the overhead at the time of read e § fulfills the definition of b ≠: Recall that the state of a register replica used in the read is a tuple (r, A) = state(e), where r is a replica ID and A is a set of active entries; each entry (a, vv) 2 A is made of value a and a version vector vv (a mapping from a replica ID to an integer). By definitions of write operation Given any number of replicas n 0 and any number of updates m 0 ∏ n 0 , we pick n = n 0 and m = m 0 to demonstrate overhead on execution:

Clearly, the execution has n replicas and m updates. The execution is well-defined since the driver program uses message identifiers and timestamps correctly. Let (A, T) = state(e § ) be the state of replica 1 at the time of read e § , where A is a set of add instances and T is a set of removed instances. By definitions of add and rem in D RWSet , the set of add instances is empty for this execution, A = ;, whereas the set of remove instances contains all removed elements with timestamps of remove operations, |T| = m. We assume Lamport clock [START_REF] Lamport | Time, clocks, and the ordering of events in a distributed system[END_REF] as a timestamp t, the best known space-efficient timestamp implementation, made of a pair t = (k, r), where k is a natural number, r is a replica ID, and k ∑ m. Thus, for some constants K 1 and K 2 , state (A, T)

needs to occupy at least: Recall that the state of a counter replica used in the read is a tuple (r, v) = state(e), where r is a replica ID and v is a map (vector) from replica ID to the number of increments made at that replica.

By definitions of inc and deliver, the two mutators of component v, 8s 2 ReplicaID. v(s) ∑ m 0 .

Therefore, for some constants K 1 and K 2 , state (r, v) can be encoded as a string bounded by:

By F Ctr , the value of a counter must be m 0 , hence the return value v = C.rval(e) encoding is bounded by: len N (v) ∏ K 3 (lg(m 0 + 1)) for some constant K 3 . Finally, for some constant K, metadata overhead is bounded by: LWWReg É with n replicas, m ∏ n updates, and any read event e in this execution (assuming there is at least one).

Recall that the state of a register replica used in the read is a tuple (a, t) = state(e), where a is a register value and t is a timestamp of the value. We assume Lamport clock [START_REF] Lamport | Time, clocks, and the ordering of events in a distributed system[END_REF] as a timestamp t, where timestamp is a pair t = (k, r) with a natural number k ∑ m, and with replica ID r. For some constant K 1 , state (a, t) encoding is bounded by: len(a, t) ∑ K 1 (len Z (a) + lg m + lg n)

Note that return value v = C.rval(e) of read matches the register value stored in the state (cf.

the definition of read for D § LWWReg ), v = a. Therefore, for some constants K 2 , K (picked independently of a, n and m) we obtain the following bound on metadata overhead: len(a, t len(v)

B.3 Lower Bound Proofs

This section contains proofs of theorems and lemmas from Section 4.2. = result(D RWSet , (R 0 , N 0 ), (init; exp(AE); test(r)))

B.3.1 Add-Wins

= result(D RWSet , (R AE , N AE ), test(r))

where:

(i) This is due to D RWSet sat[V state , T any ] F RWSet , as we explained informally before. Let By F LWWSet , v must include element r if among visible updates concerning r, add(r) is the latest in the arbitration order ar(C 0 AE ). Therefore, we will now consider the arbitration relation between visible add(r) and rem(r) events. If an even number of operations form r is visible (2AE(r)), then the latest one must be rem(r). If an odd number of operations form r is visible (2i + 1), then the latest one must be add(r), By specification F LWWSet the return value of the read is in this case:

; otherwise i.e., {0} appears in the return value when read observes add(r) that was not observed in the culrpit read. Therefore: Proof. Given n 0 , m 0 , we pick n = n 0 and some m ∏ n 0 such that m is a multiple of 2(n °1) and m ∏ n 2 . Take the experiment family (Q, n, m, C, e, readback) given by Lemma 4.5. For any AE 2 Q, C AE .rval(e AE ) = ; by F LWWSet applied to the witness abstract execution; empty set can be encoded with a constant length. Using Lemma 4.1 and m ∏ n 2 , for some constants K 1 , K 2 , K we get:

B.3.4 Multi-Value

where:

(i) This is due to Then the operation context in abs(C 0 AE , V state ) of the read into v i in test(r) contains:

(a) first max(AE(r), i) operations write(0) from replica r, and (b) first AE(r 0 ) operations write(0) from every replica r 0 6 = r, and (c) and a single write(1) made on replica 1.

We will now analyze the visibility relation between writes write(0) and write [START_REF]Riak distributed database[END_REF]. None of writes write(0) observed write(1) in its operation context in abs(C 0 AE , V state ). The operation context of write(1) includes first AE(r 0 ) operations write(0) from every replica r 0 , including replica r. That means, read into u i observes max(0, i °AE(r)) operations write(0) from r that were not visible to write [START_REF]Riak distributed database[END_REF]. By specification F MVReg the return value of the read is in this case:

i.e., 0 appears in the return value whenever there is write(0) concurrent to write(1) visible.

Therefore: for all i 2 [1..m)] do 14:

deliver 1 (mid r,i )

15:

where (R 0 , N 0 ) = (D ø .initialize, ;) Proof. Given n 0 , m 0 , we pick n = n 0 and some m ∏ n 0 such that (m °1) is a multiple of (n °1) and m ∏ n 2 . Take the experiment family (Q, n, m, C, e, readback) given by Lemma 4.6. Then for any AE, C AE .rval(e AE ) = {1} by F MVReg applied to the witness abstract execution; singleton set can be encoded with a constant length string. Using Lemma 4.1 and m ∏ n 2 , for some constants K 1 , K 2 , K 3 , K independent from n 0 , m 0 we get: as defined in Table B.2 is an experiment family.

The idea of the experiments is to force replica 1 to remember the number of writes that the write of the current value dominates, which then introduces an overhead proportional to the lg m, the minimum cost of storing a timestamp; cf. the implementation in Algorithm 2.4. Only two replicas are involved the experiment, since the overhead is not related to the number of replicas.

Other replicas perform only dummy reads to ensure that execution uses at least n replicas, and fulfills wcmo definition.

All experiments start with a common initialization phase, defined by init, where replica 2 performs m write operations, writing interchangeably values 0 and 1, starting with 1, and sending a message after each write. For each write operation we define a timestamp such that the execution of driver program consists of the arbitration order consistent with the order of operations on replica 2. All messages send in the init phase of the program are undelivered until the second phase, defined by exp(AE). There replica 1 receives exactly one message from replica 2, selected using AE. An experiment concludes with the read e AE on the first replica.

The read-back works by performing a test defined by test program. To determine which message was delivered by replica 1 during the experiment, it first performs a reference read on the first replica, and then forces (re)delivery of all messages sent by replica 2 to replica 1, in the order they were sent, performing a read after each message delivered. By comparing return value of each such read with the reference read, the program identifies the first message that makes the read include an outcome of write operation dominating in arbitration order all write operations observed in the reference read; the index of this message corresponds to AE + 1.

Proof of Lemma B.4. Some parts of the proof are straightforward, such as checking that n and m match the number of replicas/updates, and that the driver programs use message identifiers and timestamps correctly. The only nontrivial obligation is to prove readback(state(e AE ))

where:

(i) This is due to D LWWReg sat[V state , T any ] F LWWReg , as we explained informally above. Let

Then the operation context in abs(C 0 AE , V state ) of the reference read into v in test contains AE first write operations. By F LWWReg , v must reflect value of the visible write that is the latest in the arbitration order ar(C 0 AE ). By assigned timestamps the arbitration order corresponds to the order of write operations on replica 2. Hence, v = AE mod 2.

Let v 0

i denote value of a read into v 0 in the i-th iteration of the loop in the program test. The operation context of read into v 0 i in abs(C 0 AE , V state ) contains max(i, AE) first write operations performed at replica 2. Again, by specification F LWWReg the value must be v

By comparing values of v and all v 0 i , we obtain:

and R AE can differ only in the states of the replicas 2..n. These cannot influence the run of test, since it performs events on replica 1 only.

Proof of Theorem 4.

Appendix C

Résumé de la thèse

La haute disponibilité, ainsi que la réactivité sont les qualités essentielles des applications web interactifs partageant les données modifiables. Pour assurer ces exigences, les applications comptent sur les bases de données géo-réplicatives répliquant les données dans des endroits géographiquement dispersés à travers le monde. La géo-réplication permet aux utilisateurs d'accéder à la réplique locale des données située dans le centre de donnée le plus proche ou sur leur propre station. De plus, l'accès aux données locales n'est pas entravé par le coût de la latence du réseau connectant les répliques, ni par l'impact des défaillances de l'infrastructure [START_REF] Corbett | Spanner: Google's globally-distributed database[END_REF][START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF][START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF].

Lorsque les performances et la disponibilité des opérations de mise à jour sont cruciales pour le fonctionnement de l'application, les opérations de mises à jour doivent être également réalisées localement, sans la coordination avec les répliques distantes. Cela implique une acceptation des mises à jours simultanées se reproduisant d'une façon asynchrone. Malheureusement, cela est en conflit avec les modèles de forte cohérence de données, notamment avec l'atomicité ou la sérialisation [START_REF] Bernstein | Concurrency Control and Recovery in Database Systems[END_REF][START_REF] Herlihy | Linearizability: a correcteness condition for concurrent objects[END_REF]. La forte cohérence offre aux applications une vue unique de la base de données distribuée, mais nécessite une exécution des opérations suivant un ordre global sur toutes les répliques à l'aide de la réplication synchrone. L'incompatibilité entre la disponibilité, la tolérance aux pannes et la forte cohérence, connue sous le nom du théorème CAP, force la réplication asynchrone à se recourir aux plus faibles modèles de cohérence à terme [START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF][START_REF] Gilbert | Brewer's conjecture and the feasibility of consistent, available, partition-tolerant web services[END_REF][START_REF] Petersen | Flexible update propagation for weakly consistent replication[END_REF].

Dans le cas de la cohérence à terme, les répliques sont autorisées à s'écarter de façon transitoire, par exemple, au cours des mises à jour simultanées ou pendant une panne de réseau.

Cependant, les répliques doivent, à terme, converger vers un état commun intégrant toutes les mises à jour [START_REF] Petersen | Flexible update propagation for weakly consistent replication[END_REF][START_REF] Vogels | Eventually consistent[END_REF]. Les états intermittents exposent l'application au risque des anomalies de cohérence, qui sont non seulement embarrassantes pour les utilisateurs, mais aussi complique la réalisation de l'application, ainsi que de la base de données [START_REF] Corbett | Spanner: Google's globally-distributed database[END_REF]. Les défis comprennent surtout la détection des mises à jours simultanés étant en conflit, leur résolution convergente, ainsi que

C.2.2 L'approche SwiftCloud

Nous proposons un modèle soulevant les défis mentionnés. Premièrement, nous le décrivons dans un cas sans pannes. Deuxièmement, nous traitons le cas de la panne du DC. Finalement nous introduisons une nouvelles forme de métadonnées utilisée dans notre solution.

C.2.2.1 Cohèrence causale dans les répliques complètes des Centre de Données

Le problème de la garantie de la cohérence dans les DCs complétement répliqués est bien connu [START_REF] Ahamad | Causal memory[END_REF]46,66,[START_REF] Lloyd | Stronger semantics for low-latency geo-replicated storage[END_REF]. Nous se concentrons sur les protocoles basées sur les logs, transmettant les opérations d'une façon incrémentale [START_REF] Belaramani | PRACTI replication[END_REF][START_REF] Petersen | Flexible update propagation for weakly consistent replication[END_REF].

Une version de base de données, notée U, est un sous-ensemble des mises à jour ordonné selon la causalité. Une version associe les identifiants des objets à leurs valeurs (exposé par les méthodes read) en appliquant une subséquence pertinente de log au état initial de chaque objet. Remarquons que dans beaucoup de systèmes précédents, cette impossibilité impliqué un compromis entre, d'une part, la cohérence et la disponibilité et performance de l'autre [START_REF] Corbett | Spanner: Google's globally-distributed database[END_REF]66,[START_REF] Sovran | Transactional storage for geo-replicated systems[END_REF].

Nous disons qu'une version

En "lisant le passé," nous déplaçons cela vers un compromis entre la fraîcheur et la disponibilité, contrôlées par l'ajustement de K. Un K plus haut augmente la disponibilité, mais les mises à jour prennent plus de temps pour être livrées; 4 Dans une certaine limite, K = N assure une complète disponibilité, mais aucun client ne peut délivrer une mises à jour si un DC n'est pas disponible.

Un K inférieur améliore la fraîcheur, mais augmente la probabilité que le client ne pourra pas basculer sur erreur et qu'il attendra la récupération du DC. Dans une certaine limite, K = 1 est identique que le protocole de base de Section C.2.2.2, ainsi que similaire au protocoles précédents bloquant des garanties session [START_REF] Terry | Session guarantees for weakly consistent replicated data[END_REF]. K = 2 constitue un bon compromis pour le déploiement de trois ou plus DCs couvrant les scénarios commun incluant la panne du DC ou la déconnexion [START_REF] Corbett | Spanner: Google's globally-distributed database[END_REF][START_REF] Kansal | Using dark fiber to displace diesel generators[END_REF]. Notre évaluation avec K = 2 montre qu'il encourt un manque de fraîcheur négligeable.