An obstruction-free implementation guarantees progress to every operation that is given enough time to take steps in isolation. But, as we show in this paper, the mere presence of concurrent operations alone does not have to prevent progress; only incomplete conflicting (non-commuting) operations may do so. This progress condition, that we call conflict-freedom, is a natural generalization of obstruction-freedom that promises efficient implementations for objects exhibiting semantic commutativity. We show that, as with obstruction-freedom, every sequential object has a read-write conflict-free linearizable implementation. Our conflict-free universal construction is based on a novel generalization of the instrumental commit-adopt object, interesting in its own right.

Parallel programs require software support to coordinate access to shared data. For this purpose, modern programming languages provide strongly-consistent shared objects. To account for their many usages, these objects offer a large API. However, in practice, each program calls only a small subset of the interface. Leveraging such an observation, we propose to tailor a shared object for a specific usage. We call this principle adjusted objects. Adjusted objects already exist in the wild. This paper provides their first systematic study. We explain how everyday programmers already adjust common shared objects (such as queues, maps, and counters) for better performance. We present the formal foundations of adjusted objects using a new tool to characterize scalability, the indistinguishability graph. Leveraging this study, we introduce a library named DEGO to inject adjusted objects in a Java program. In micro-benchmarks, objects from the DEGO library improve the performance of standard JDK shared objects by up to two orders of magnitude. We also evaluate DEGO with a Retwis-like benchmark modeled after a social network application. On a modern server-class machine, DEGO boosts by up to 1.7x the performance of the benchmark.

Classical state-machine replication protocols, such as Paxos, rely on a distinguished leader process to order commands. Unfortunately, this approach makes the leader a single point of failure and increases the latency for clients that are not co-located with it. As a response to these drawbacks, Egalitarian Paxos [Iulian Moraru et al., 2013] introduced an alternative, leaderless approach, that allows replicas to order commands collaboratively. Not relying on a single leader allows the protocol to maintain non-zero throughput with up to f crashes of any processes out of a total of n = 2f+1. The protocol furthermore allows any process to execute a command c fast, in 2 message delays, provided no more than e = ⌈(f+1)/2⌉ other processes fail, and all concurrently submitted commands commute with c; the latter condition is often satisfied in practical systems. Egalitarian Paxos has served as a foundation for many other replication protocols. But unfortunately, the protocol is very complex, ambiguously specified and suffers from nontrivial bugs. In this paper, we present EPaxos* - a simpler and correct variant of Egalitarian Paxos. Our key technical contribution is a simpler failure-recovery algorithm, which we have rigorously proved correct. Our protocol also generalizes Egalitarian Paxos to cover the whole spectrum of failure thresholds f and e such that n ≥ max{2e+f-1, 2f+1} - the number of processes that we show to be optimal.

A key question in distributed computing is determining the synchronization power of a shared object. This power is commonly measured using consensus, a distributed problem in which processes agree on a common input value. An object has consensus number n when n is the largest number for which we may solve consensus with copies of this object and registers. The seminal work of Herlihy and Ruppert [14] provides an exact characterization of the consensus number for deterministic one-shot objects (that can be accessed by each process at most once). This paper extends that study to deterministic two-shot objects (that can be accessed by each process at most twice) in a two-process system. We introduce three disjoint classes of two-shot objects: The first class is similar to one-shot objects in the sense that the first operation call gives enough information to solve consensus. Objects in the second class do not provide any useful information after the first call to one of the two processes. The last class contains objects for which calling the object twice is always necessary. In this class, the second operation to call is chosen adaptively, which may lead to using different operations in different schedules. For instance, the second operation used in a solo run might differ from the one called when processes interleave. We show that these three classes provide an exact characterization of the two-shot deterministic objects able to solve two-process consensus. To establish this, we first prove that any two-shot deterministic object solving two-process consensus must belong to one of the three classes. Then, we present matching consensus algorithms, one for each class.

A seminal result by Lamport shows that at least max{2e + f + 1, 2f + 1} processes are required to implement partially synchronous consensus that tolerates f process failures and can furthermore decide in two message delays under f failures. This lower bound is matched by the classical Fast Paxos protocol. However, more recent practical protocols, such as Egalitarian Paxos, provide two-step decisions with fewer processes, seemingly contradicting the lower bound. We show that this discrepancy arises because the classical bound requires two-step decisions under a wide range of scenarios, not all of which are relevant in practice. We propose a more pragmatic condition for which we establish tight bounds on the number of processes required. Interestingly, these bounds depend on whether consensus is implemented as an atomic object or a decision task. For consensus as an object, max{2e + f - 1, 2f + 1} processes are necessary and sufficient for two-step decisions, while for a task the tight bound is max{2e + f, 2f + 1}.

Atomic broadcast is a group communication primitive to order messages across a set of distributed processes. Atomic multicast is its natural generalization where each message m is addressed to dst (m), a subset of the processes called its destination group. A solution to atomic multicast is genuine when a process takes steps only if a message is addressed to it. Genuine solutions are the ones used in practice because they have better performance.

Atomic broadcast is a group communication primitive to order messages across a set of distributed processes. Atomic multicast is its natural generalization where each message $m$ is addressed to $dst(m)$, a subset of the processes called its destination group. A solution to atomic multicast is genuine when a process takes steps only if a message is addressed to it. Genuine solutions are the ones used in practice because they have better performance. Let $G$ be all the destination groups and $F$ be the cyclic families in it, that is the subsets of $G$ whose intersection graph is hamiltonian. This paper establishes that the weakest failure detector to solve genuine atomic multicast is $μ=(\wedge_{g,h \in G}~Σ_{g \cap h}) \wedge (\wedge_{g \in G}~Ω_g) \wedge γ$, where (i) $Σ_P$ and $Ω_P$ are the quorum and leader failure detectors restricted to the processes in $P$, and (ii) $γ$ is a new failure detector that informs the processes in a cyclic family $f \in F$ when $f$ is faulty. We also study two classical variations of atomic multicast. The first variation requires that message delivery follows the real-time order. In this case, $μ$ must be strengthened with $1^{g \cap h}$, the indicator failure detector that informs each process in $g \cup h$ when $g \cap h$ is faulty. The second variation requires a message to be delivered when the destination group runs in isolation. We prove that its weakest failure detector is at least $μ\wedge (\wedge_{g, h \in G}~Ω_{g \cap h})$. This value is attained when $F=\varnothing$.

Serverless computing greatly simplifies the use of cloud resources. In particular, Function-as-a-Service (FaaS) platforms enable programmers to develop applications as individual functions that can run and scale independently. Unfortunately, applications that require fine-grained support for mutable state and synchronization, such as machine learning (ML) and scientific computing, are notoriously hard to build with this new paradigm. In this work, we aim at bridging this gap. We present Crucial , a system to program highly-parallel stateful serverless applications. Crucial retains the simplicity of serverless computing. It is built upon the key insight that FaaS resembles to concurrent programming at the scale of a datacenter. Accordingly, a distributed shared memory layer is the natural answer to the needs for fine-grained state management and synchronization. Crucial allows to port effortlessly a multi-threaded code base to serverless, where it can benefit from the scalability and pay-per-use model of FaaS platforms. We validate Crucial with the help of micro-benchmarks and by considering various stateful applications. Beyond classical parallel tasks (e.g., a Monte Carlo simulation), these applications include representative ML algorithms such as k -means and logistic regression. Our evaluation shows that Crucial obtains superior or comparable performance to Apache Spark at similar cost (18%–40% faster). We also use Crucial to port (part of) a state-of-the-art multi-threaded ML library to serverless. The ported application is up to 30% faster than with a dedicated high-end server. Finally, we attest that Crucial can rival in performance with a single-machine, multi-threaded implementation of a complex coordination problem. Overall, Crucial delivers all these benefits with less than 6% of changes in the code bases of the evaluated applications.

Modern web applications replicate their data across the globe and require strong consistency guarantees for their most critical data. These guarantees are usually provided via state-machine replication (SMR). Recent advances in SMR have focused on leaderless protocols, which improve the availability and performance of traditional Paxos-based solutions. We propose Tempo - a leaderless SMR protocol that, in comparison to prior solutions, achieves superior throughput and offers predictable performance even in contended workloads. To achieve these benefits, Tempo timestamps each application command and executes it only after the timestamp becomes stable, i.e., all commands with a lower timestamp are known. Both the timestamping and stability detection mechanisms are fully decentralized, thus obviating the need for a leader replica. Our protocol furthermore generalizes to partial replication settings, enabling scalability in highly parallel workloads. We evaluate the protocol in both real and simulated geo-distributed environments and demonstrate that it outperforms state-of-the-art alternatives.

Edge applications, such as gaming, cooperative engineering, or in-the-field information sharing, enjoy immediate response, autonomy and availability by distributing and replicating data at the edge. However, application developers and users demand the highest possible consistency guarantees, and specific support for group collaboration. To address this challenge, Colony guarantees Transactional Causal Plus Consistency (TCC+) globally, strengthened to Snapshot Isolation within edge groups. To help with scalability, fault tolerance and security, its logical communication topology is forest-like, with replicated roots in the core cloud, but with the flexibility to migrate a node or a group. Despite this hybrid approach, applications enjoy the same semantics everywhere in the topology. Our experiments show that local caching and peer groups improve throughput and response time significantly, performance is not affected in offline mode, and that migration is seamless.

This paper presents J-NVM, a framework to access efficiently Non-Volatile Main Memory (NVMM) in Java. J-NVM offers a fully-fledged interface to persist plain Java objects using failure-atomic blocks. This interface relies internally on proxy objects that intermediate direct off-heap access to NVMM. The framework also provides a library of highly-optimized persistent data types that resist reboots and power failures. We evaluate J-NVM by implementing a persistent backend for the Infinispan data store. Our experimental results, obtained with a TPC-B like benchmark and YCSB, show that J-NVM is consistently faster than other approaches at accessing NVMM in Java.

Online applications now routinely replicate their data at multiple sites around the world. In this paper we present Atlas, the first state-machine replication protocol tailored for such planet-scale systems. Atlas does not rely on a distinguished leader, so clients enjoy the same quality of service independently of their geographical locations. Furthermore, client-perceived latency improves as we add sites closer to clients. To achieve this, Atlas minimizes the size of its quorums using an observation that concurrent data center failures are rare. It also processes a high percentage of accesses in a single round trip, even when these conflict. We experimentally demonstrate that Atlas consistently outperforms state-of-the-art protocols in planet-scale scenarios. In particular, Atlas is up to two times faster than Flexible Paxos with identical failure assumptions, and more than doubles the performance of Egalitarian Paxos in the YCSB benchmark.

Modern Internet services commonly replicate critical data across several geographical locations using state-machine replication (SMR). Due to their reliance on a leader replica, classical SMR protocols offer limited scalability and availability in this setting. To solve this problem, recent protocols follow instead a leaderless approach, in which each replica is able to make progress using a quorum of its peers. In this paper, we study this new emerging class of SMR protocols and states some of their limits. We first propose a framework that captures the essence of leaderless state-machine replication (Leaderless SMR). Then, we introduce a set of desirable properties for these protocols: (R)eliability, (O)ptimal (L)atency and (L)oad Balancing. We show that protocols matching all of the ROLL properties are subject to a trade-off between performance and reliability. We also establish a lower bound on the message delay to execute a command in protocols optimal for the ROLL properties. This lower bound explains the persistent chaining effect observed in experimental results.

Serverless computing is an emerging paradigm that greatly simplifies the usage of cloud resources and suits well to many tasks. Most notably, Function-as-a-Service (FaaS) enables programmers to develop cloud applications as individual functions that can run and scale independently. Yet, due to the disaggregation of storage and compute resources in FaaS, applications that require fine-grained support for mutable state and synchronization, such as machine learning and scientific computing, are hard to build.

Generalized Paxos (GPaxos) is a recent solution to Generalized Consensus, a distributed problem to which several key agreement problems reduce. We envision that GPaxos may unify within a single and novel Agreement-as-a-Service infrastructure multiple distributed protocols. To date this potential is however not fully unleashed, due to the steep learning curve of the protocol and the high complexity of its implementation. Moreover, before GPaxos reaches a real world usage, several computationally expensive operations have to be optimized and simplified. This paper aims at closing this gap between theory and practice. To this end, we first provide a concise tour of Generalized Paxos, hardly found elsewhere. Then, we assess the versatility of the Generalized Consensus problem by presenting a variation of GPaxos that solves the lease coordination problem. Our last contribution consists in three optimizations that apply to the critical phases of the algorithm: (i) a method to quickly start a new round, (ii) a novel approach to execute a checkpoint, and (iii) a data structure that speeds-up the detection of an agreement.

In a Cloud environment, the ability to share and persist objects simplifies the design of applications. Storing objects in a NoSQL database ensures their availability and provides scalability to applications. When Object-NoSQL Mapping is performed at the client side, objects that are accessed by several clients are repeatedly converted between their in-memory and serialized representations. This negatively impacts performance and increases replication costs. In this paper, we describe the design of CRESON, a system supporting callable objects over NoSQL, in which application objects are mapped and instantiated directly on the storage nodes. CRESON supports composition by reference and ensures strong consistency. Objects are replicated and maintained coherent using State Machine Replication. The implementation of CRESON leverages the support of a listenable key-value store (LKVS), a novel NoSQL storage abstraction that we introduce in this paper. We discuss the performance and complexity of CRESON with the example of the portage of a personal cloud storage service, initially developed using an object-relational mapping over a sharded PostgreSQL database. Our results show that CRESON offers a simpler programming experience both in terms of learning time and lines of code, while performing better on average and being more scalable.

As the wealth of information available on the web keeps growing, being able to harvest massive amounts of data has become a major challenge. Web crawlers are the core components to retrieve such vast collections of publicly available data. The key limiting factor of any crawler architecture is however its large infrastructure cost. To reduce this cost, and in particular the high upfront investments, we present in this paper a geo-distributed crawler solution, UniCrawl. UniCrawl orchestrates several geographically distributed sites. Each site operates an independent crawler and relies on well-established techniques for fetching and parsing the content of the web. UniCrawl splits the crawled domain space across the sites and federates their storage and computing resources, while minimizing thee inter-site communication cost. To assess our design choices, we evaluate UniCrawl in a controlled environment using the ClueWeb12 dataset, and in the wild when deployed over several remote locations. We conducted several experiments over 3 sites spread across Germany. When compared to a centralized architecture with a crawler simply stretched over several locations, UniCrawl shows a performance improvement of 93.6% in terms of network bandwidth consumption, and a speedup factor of 1.75.

The k-set agreement problem is a generalization of the consensus problem. Namely, assuming that each process proposes a value, every non-faulty process should decide one of the proposed values, and no more than k different values should be decided. This is a hard problem in the sense that we cannot solve it in an asynchronous system, as soon as k or more processes may crash. One way to sidestep this impossibility result consists in weakening the termination property, requiring that a process must decide a value only if it executes alone during a long enough period of time. This is the well-known obstruction-freedom progress condition. Consider a system of n anonymous asynchronous processes that communicate through atomic read/write registers, and such that any number of them may crash. In this paper, we address and solve the challenging open problem of designing an obstruction-free k-set agreement algorithm using only (n-k+1) atomic registers. From a shared memory cost point of view, our algorithm is the best algorithm known so far, thereby establishing a new upper bound on the number of registers needed to solve the problem, and in comparison to the previous upper bound, its gain is (n-k) registers. We then extend this algorithm into a space-optimal solution for the repeated version of k-set agreement, and an x-obstruction-free solution that employs 0(n-k+x) atomic registers (with 1 <= x <= k < n).

A large family of distributed transactional protocols have a common structure, called Deferred Update Replication (DUR). DUR provides dependability by replicating data, and performance by not re-executing transactions but only applying their updates. Protocols of the DUR family differ only in behaviors of few generic functions. Based on this insight, we offer a generic DUR middleware, called G-DUR, along with a library of finely-optimized plug-in implementations of the required behaviors. This paper presents the middleware, the plugins, and an extensive experimental evaluation in a geo-replicated environment. Our empirical study shows that:(i) G-DUR allows developers to implement various transactional protocols under 600 lines of code; (ii) It provides a fair, apples-to-apples comparison between transactional protocols; (iii) By replacing plugs-ins, developers can use G-DUR to understand bottlenecks in their protocols; (iv) This in turn enables the improvement of existing protocols; and (v) Given a protocol, G-DUR helps evaluate the cost of ensuring various degrees of dependability.

The ability to access and query data stored in multiple versions is an important asset for many applications, such as Web graph analysis, collaborative editing platforms, data forensics, or correlation mining. The storage and retrieval of versioned data requires a specific API and support from the storage layer. The choice of the data structures used to maintain versioned data has a fundamental impact on the performance of insertions and queries. The appropriate data structure also depends on the nature of the versioned data and the nature of the access patterns. In this paper we study the design and implementation space for providing versioning support on top of a distributed key-value store (KVS). We define an API for versioned data access supporting multiple writers and show that a plain KVS does not offer the necessary synchronization power for implementing this API. We leverage the support for listeners at the KVS level and propose a general construction for implementing arbitrary types of data structures for storing and querying versioned data. We explore the design space of versioned data storage ranging from a flat data structure to a distributed sharded index. The resulting system, ALEPH, is implemented on top of an industrial-grade open-source KVS, Infinispan. Our evaluation, based on real-world Wikipedia access logs, studies the performance of each versioning mechanisms in terms of load balancing, latency and storage overhead in the context of different access scenarios.

Cloud computing infrastructures leverage fault-tolerant and geographically distributed services in order to meet the requirements of modern applications. Each service deals with a large number of clients that compete for the resources it offers. When the load increases, the service needs to scale. In this paper, we investigate a scalability solution which consists in partitioning the service state. We formulate specific conditions under which a service is partitionable. Then, we present a general algorithm to build a dependable and consistent partitioned service. To assess the practicability of our approach, we implement and evaluate the ZooFence coordination service. ZooFence orchestrates several instances of ZooKeeper and presents the exact same API and semantics to its clients. It automatically splits the coordination service state among ZooKeeper instances while being transparent to the application. By reducing the convoy effect on operations and leveraging the workload locality, our approach allows proposing a coordination service with a greater scalability than with a single ZooKeeper instance. The evaluation of ZooFence assesses this claim for two benchmarks, a synthetic service of concurrent queues and the BookKeeper distributed logging engine.

Modern cloud systems are geo-replicated to improve application latency and availability. Transactional consistency is essential for application developers; however, the corresponding concurrency control and commitment protocols are costly in a geo-replicated setting. To minimize this cost, we identify the following essential scalability properties: (i) only replicas updated by a transaction T make steps to execute T; (ii) a read-only transaction never waits for concurrent transactions and always commits; (iii) a transaction may read object versions committed after it started; and (iv) two transactions synchronize with each other only if their writes conflict. We present Non-Monotonic Snapshot Isolation (NMSI), the first strong consistency criterion to allow implementations with all four properties. We also present a practical implementation of NMSI called Jessy, which we compare experimentally against a number of well-known criteria. Our measurements show that the latency and throughput of NMSI are comparable to the weakest criterion, read-committed, and between two to fourteen times faster than well-known strong consistencies.

Consensus (agreeing on a sequence of commands) is central to the operation and performance of distributed systems. A well-known solution to consensus is Fast Paxos. In a recent paper, Lamport enhances Fast Paxos by lever aging the commutativity of concurrent commands. The new primitive, called Generalized Paxos, reduces the collision rate, and thus the latency of Fast Paxos. However if a collision occurs, Generalized Paxos needs four communication steps to recover, which is slower than Fast Paxos. This paper presents FGGC, a novel consensus algorithm that reduces recovery delay when a collision occurs to one. FGGC tolerates f <; n/2 replicas crashes, and during failure-free runs, processes learn commands in two steps if all commands commute, and three steps otherwise; this is optimal. Moreover, as long as no fault occurs, FGGC needs only f + 1 replicas to progress.

Partial replication is a way to increase the scalability of replicated systems: updates only need to be applied to a subset of the system's sites, thus allowing replicas to handle independent parts of the workload in parallel. In this paper, we propose P-Store, a partially replicated key-value store for wide area networks. In P-Store, each transaction T optimistically executes on one or more sites and is then certified to guarantee serializability of the execution. The certification protocol is genuine, it only involves sites that replicate data items read or written by T, and incorporates a mechanism to minimize a convoy effect. P-Store makes a thrifty use of an atomic multicast service to guarantee correctness: no messages need to be multicast during T's execution and a single message is multicast to certify T. In case T is global, that is, T's execution is distributed at different geographical locations, an extra vote phase is required. Our approach may offer better scalability than previously proposed solutions that either require multiple atomic multicast messages to execute T or are non-genuine. Experimental evaluations reveal that the convoy effect plays an important role even when one percent of the transactions are global. We also compare the scalability of our approach to a fully replicated solution when the proportion of global transactions and the number of sites vary.

We study atomic multicast, a fundamental abstraction for building fault-tolerant systems. We suppose a system composed of data centers, or groups, that host many processes connected through high-end local links; a few groups exist, interconnected through high-latency communication links. A recent paper showed that no multicast protocol can deliver messages addressed to multiple groups in one inter-group delay and be genuine, i.e., to deliver a message m, only the addressees of m are involved in the protocol. We propose a non-genuine multicast protocol that may deliver messages addressed to multiple groups in one inter-group delay. Experimental comparisons against a latency-optimal genuine protocol show that the non-genuine protocol offers better performance in almost all considered scenarios. We also identify a convoy effect in multicast algorithms that may delay the delivery of local messages, i.e., messages addressed to a single group, by as much as the latency of global messages, i.e., messages addressed to multiple groups, and propose techniques to minimize this effect. To complete our study, we evaluate a latency-optimal protocol that tolerates disasters, i.e., group crashes.

Wikis, a popular tool for sharing knowledge, are basically collaborative editing systems. However, existing Wiki systems offer limited support for co-operative authoring, and they do not scale well, because they are based on a centralised architecture. This paper compares the well-known centralised MediaWiki system with several peer-to-peer approaches to editing of wiki pages: an operational transformation approach (MOT2), a commutativity-oriented approach (WOOTO) and a conflict resolution approach (ACF). We evaluate and compare them, according to a number of qualitative and quantitative metrics.