Planet-scale leaderless consensus

Vitor Manuel Enes Duarte November 2022 UMinho|2022 Planet-Scale Leaderless Consensus Planet-Scale Leaderless Consensus Vitor Manuel Enes Duarte Universidade do Minho Escola de Engenharia Programa de Doutoramento em Informática (MAP-i) das Universidades do Minho, de Aveiro e do Porto Universidade do Minho universidade de aveiro Universidade do Minho Escola de Engenharia Departamento de Informática Vitor Manuel Enes Duarte Planet-Scale Leaderless Consensus Tese de Doutoramento Programa de Doutoramento em Informática das Universidades do Minho, de Aveiro e do Porto Trabalho realizado sob a orientação de Alexey Gotsman ede Carlos Baquero November 2022 COPYRIGHT AND TERMS OF USE OF THIS WORK BY A THIRD PARTY This is academic work that can be used by third parties as long as internationally accepted rules and good practices regarding copyright and related rights are respected. Accordingly, this work may be used under the license provided below. If the user needs permission to make use of the work under conditions not provided for in the indicated licensing, they should contact the author through the RepositoriUM of Universidade do Minho. License granted to the users of this work Creative Commons Attribution 4.0 International CC BY 4.0 https://creativecommons.org/licenses/by/4.0/deed.en This document was created with the (pdf/Xe/Lua)L A T EX processor and the NOVAthesis template (v6.9.0) [1]. ii Acknowledgements Without the guidance of my advisors, Alexey Gotsman and Carlos Baquero, this thesis would not exist. I didn’t know where we were going most of the times (maybe we never do), but with your help and support somehow we got here. Without me realizing it at the time, this thesis started during my internship at IMDEA. I am immensely thankful for having worked closely with Alexey since then. His patience, invested time and detailed feedback made me a better researcher and this thesis possible. I still remember the day when I entered Carlos’ office to bother him with my interest in CRDTs. Most certainly I would not be here if he hadn’t believe in me back then and since. I am tremendously grateful for having worked with Pierre Sutra and Paulo Sérgio Almeida. You were also my advisors, just not on paper. I learned so much from you and I cannot thank you enough. I especially want to thank Matthieu Perrin, Tuanir França Rezende, Christopher S. Meiklejohn, Junghun Yoo, Peter Van Roy, Annette Bieniusa, João Leitão, and Ali Shoker, with whom I have collaborated during all these years. I am thankful for the numerous anonymous reviewers, as well as Liuba Shrira and Natacha Crooks, all of whom shepherded papers containing results that appear on this thesis. I am also very thankful for Lennart Oldenburg, Antonios Katsarakis, Gonçalo Tomás, Dimitris Vasilas, Manuel Bravo, Borja de Régil, Fedor Ryabinin, Artem Khyzha, Marc Shapiro, Andrey Kuprianov, Adi Seredinschi, and Murat Demirbas for their research advice and comments on papers. I am thankful for everyone at HASLab, IMDEA and Informal Systems for being so welcoming and helpful throughout these years. I am so grateful for all of my labmates. Georges, Ricardo, Rogério, Cláudia, Fábio, Tânia, Neves, João and Marco, thank you for always being there to listen to my ideas and worries. Georges, my friend, I am immensely grateful for having shared this journey with you since day one, for all our trips and discussions about research and life. Friends, I am thankful for your love, help, support, presence and companionship, for our conversations, walks, travels, drinks and laughs. Ana, Rita, Inês, Palma, Rafa, Zé, Cabrita, Tatiana, Catarina, Fialho, Rita, Quim, Andreia, Anthony, Salomé, Marta, Marco, Ricardo, Simone, Elisa, Verónica, Filipa, Filipe, Linhares, AP, Carolina, Isac, Hugo, Ribeiro, Frank, Póvoas, Gil, Sniper, Rúben, Scout, Caxinas, Nélson, Bruno, and Rocky, I am tremendously grateful for your friendship. Above all, a huge thanks to all my family, specially to my mother, father and sister, for always being there for me. A special thanks to my uncle Nuno, for being an inspiration in many aspects of life, to Helena and Rogério, for their trust (and my newly found love for books), and to my godmother, godfather, and cousins. iii This work was partially supported by an FCT – “Fundação para a Ciência e Tecnologia” – PhD Fellowship (PD/BD/142927/2018). iv STATEMENT OF INTEGRITY I hereby declare having conducted this academic work with integrity. I confirm that I have not used plagiarism or any form of undue use of information or falsification of results along the process leading to its elaboration. I further declare that I have fully acknowledged the Code of Ethical Conduct of the Universidade do Minho. v Resumo Planet-Scale Leaderless Consensus As aplicações de web modernas replicam os seus dados à escala planetária e exigem fortes garantias na coerência dos seus dados mais críticos. Essas garantias são geralmente fornecidas por meio de replicação de máquina de estados (RME). Avanços recentes em RME concentraram-se em protocolos sem líder, pois estes melhoram o desempenho e a disponibilidade das soluções tradicionais baseadas em Paxos. Embora os protocolos sem líder se tenham mostrado muito promissores, estes são ainda pouco adequados para sistemas de escala planetária, pois utilizam grandes quóruns, oferecem um desempenho imprevisível e têm mecanismos de recuperação complexos. Nesta tese propomos dois protocolos sem líder, Atlas eTempo, adaptados para sistemas de escala planetária. O Atlas minimiza o tamanho dos seus quóruns fazendo uso da observação de que falhas simultâneas em centros de dados são raras. Também processa uma percentagem elevada de comandos da aplicação em uma única round trip , mesmo quando estes comandos conflituam. O Atlas consegue isto com um mecanismo de recuperação que é significativamente mais simples do que os protocolos sem líder que o precederam. O Tempo baseia-se no Atlas, mas atinge um rendimento superior e oferece um desempenho previsível mesmo em cargas de trabalho com elevado nível de conflitos. Para obter estes benefícios, o Tempo marca cada comando da aplicação com uma timestamp e executa-o somente após esta timestamp se tornar estável, ou seja, quando todos os comandos com uma timestamp menor são conhecidos. Ambos os mecanismos para gerar uma timestamp e detetar quando esta fica estável são totalmente descentralizados, evitando assim a necessidade de um líder. Avaliámos o Atlas eoTempo em ambientes geo-distribuídos reais e simulados e demonstramos que eles superam as alternativas oferecidas pelo estado da arte. Palavras-chave: Consenso distribuído, Geo-replicação, Tolerância a falhas. vi Abstract Planet-Scale Leaderless Consensus 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 performance and availability of traditional Paxos-based solutions. Although leaderless protocols have shown great promise, they are poorly suited to planet-scale systems as they leverage large quorums, offer unpredictable performance and have complex recovery mechanisms. In this thesis we propose two leaderless protocols, Atlas and Tempo, tailored to planet-scale systems. Atlas minimizes the size of its quorums by making use of the 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. Atlas achieves this while having a recovery mechanism that is significantly simpler than that of previous leaderless protocols. Tempo builds upon Atlas, but 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. We evaluate Atlas and Tempo in both real and simulated geo-distributed environments and demonstrate that they outperform state-of-the-art alternatives. Keywords: Consensus, Fault tolerance, Geo-replication. vii 1 Introduction Modern online applications run at multiple sites scattered across the globe: they are now planet-scale . Deploying applications in this way enables high availability and low latency, by allowing clients to access the closest responsive site. A major challenge in developing planet-scale applications is that many of their underlying components, such as coordination kernels [2,3] and critical databases [4], require strong guarantees about the consistency of replicated data. The classical way of maintaining strong consistency in a distributed service is state-machine replication (SMR) [5]. In SMR, a service is defined by a deterministic state machine, and each site maintains its own local replica of the machine. An SMR protocol coordinates the execution of commands at the sites, ensuring that they stay in sync. The resulting system is linearizable [6], i.e., it behaves as if commands are executed sequentially by a single site. Unfortunately, existing SMR protocols are poorly suited to planet-scale systems. Common SMR protocols, such as Paxos [7] and Raft [8], are rooted in cluster computing where a distinguished leader site determines the order in which client commands are executed at the replicas. This is unfair to clients far away from the leader. It impairs scalability, since the leader becomes a bottleneck when the load increases. It also harms availability as, if the leader fails, the system cannot serve requests until a new one is elected. Moreover, adding more sites to the system does not help, but on the contrary, hinders performance, requiring the leader to replicate commands to more sites on the critical path. Recent efforts [9–12] to improve SMR have thus focused on leaderless protocols, which distribute the task of ordering commands among replicas and thus allow a client to contact the closest replica instead of the leader. Compared to centralized solutions, leaderless SMR offers lower average latency, fairer latency distribution with respect to client locations, and higher availability. Moreover, leaderless protocols usually exploit the fact that commands in SMR applications frequently commute [3,4], and for the replicated state machine to be linearizable, it is enough that replicas only agree on the order of non-commuting (aka conflicting) commands [13,14]. This permits processing a command in one round trip from the closest replica using a fast path , e.g., when the command commutes with all commands concurrently submitted for execution. In the presence of concurrent conflicting commands, the protocol may sometimes have to take a slow path , which requires two round trips. 1 CHAPTER 1. INTRODUCTION Leaderless SMR protocols also generalize to the setting of partial replication , where the service state is split into a set of partitions, each stored at a group of replicas. A client command can access multiple partitions, and the SMR protocol ensures that the system is still linearizable, i.e., behaves as if the commands are executed by a single machine storing a complete service state. This approach allows implementing services that are too big to fit onto a single machine. It also enables scalability, since commands accessing disjoint sets of partitions can be executed in parallel. This has been demonstrated by Janus [15] which adapted a leaderless SMR protocol called Egalitarian Paxos (EPaxos) [11] to the setting of partial replication. The resulting protocol provided better performance than classical solutions such as two-phase commit layered over Paxos. Although leaderless protocols show great promise, they have not yet seen industry adoption. In order to be practical for planet-scale systems, we argue that leaderless protocols have to provide the following features: (i) low latency , (ii) simple recovery ,and (iii) predictable performance . In this thesis we propose Atlas and Tempo, two leaderless protocols tailored to planet-scale systems that aim to offer these features. Atlas provides the first two features ( low latency and simple recovery ). Tempo builds on Atlas, and also offers the third one ( predictable performance ). Next we cover each of these features in detail. Low latency We observe that common SMR protocols provide a level of fault-tolerance that is unnecessarily high in a geo-distributed setting. These protocols allow any minority of sites to fail simultaneously: e.g., running a typical protocol over 11 data centers would tolerate 5of them failing. However, natural disasters leading to the loss of a data center are rare, and planned downtime can be handled by reconfiguring the unavailable site out of the system [7,16]. Furthermore, temporary data center outages (e.g., due to connectivity issues) typically have a short duration [17], and, as we confirm experimentally in our evaluation (§5), rarely happen concurrently. For this reason, industry practitioners assume that the number of concurrent site failures in a geo-distributed system is low, e.g., 1or 2[4]. Motivated by this, Atlas allows choosing the maximum number of sites that can fail (5) independently of the overall number of sites (A), and is optimized for small values of the former. Making Atlas offer better latency for larger-scale deployments required two key changes to the baseline scheme of a leaderless SMR protocol. First, the lower latency of the fast path in existing protocols comes with a downside: the fast path must involve a fast quorum of replicas bigger than a majority, which increases latency due to accesses to far-away replicas. For example, in Generalized Paxos [14] the fast quorum consists of at least 2A 3replicas, and in EPaxos of at least 3A 4replicas. To solve this problem, in Atlas the size of the fast quorum is a function of the number of allowed failures 5–namely,bA 2c+5. Smaller values of 5result in smaller fast quorums, thereby decreasing latency. Atlas thus trades off higher fault tolerance for lower latency. Note that violating the assumption the protocol makes about the number of failures may only compromise liveness, but never safety. In particular, if more than 5transient outages occur, due to, e.g., connectivity problems, Atlas will just block until enough sites are reachable1. 1Apart from data centers being down, geo-distributed systems may also exhibit network partitionings, which partition off 2 The second key change introduced by Atlas is that it can take the fast path even when conflicting commands are submitted concurrently, something that is not allowed by existing SMR protocols [11,14]. This permits processing most commands via the fast path when the conflict rate is low-to-moderate, as is typical for SMR applications [3,4]. Moreover, when 5=1,Atlas always takes the fast path and its fast quorum is a plain majority. Simple recovery Failure recovery is the most subtle part of an SMR protocol with a fast path because the protocol needs to recover the decisions reached by the failed replicas while they were short-cutting some of the protocols steps in the fast path. This is only made more difficult with smaller fast quorums, as a failed process leaves information about its computations at fewer replicas. Atlas achieves its performant fast path while having a recovery protocol that is significantly simpler than that of previous leaderless protocols [10,11]. We rigorously prove the correctness of this recovery mechanism. Predictable performance Existing leaderless SMR protocols suffer from drawbacks in the way they order commands. Some protocols [9–11] (including Atlas), maintain explicit dependencies between commands: a replica may execute a command only after all its dependencies get executed. These dependencies may form arbitrary long chains. As a consequence, in theory, the protocols do not guarantee progress even under a synchronous network. In practice, their performance is unpredictable, and in particular, exhibits a high tail latency [9,19]. Other protocols [20,21]needtocontacteveryreplicaonthe critical path of each command. While these protocols guarantee progress under synchrony, they make the system run at the speed of the slowest replica. All of these drawbacks carry over to the setting of partial replication where they are aggravated by the fact that commands span multiple machines. Tempo lifts the above limitations while handling both full and partial replication settings. Tempo guarantees progress under a synchronous network without the need to contact all replicas. It also exhibits low tail latency even in contended workloads, thus ensuring predictable performance . Because it builds on Atlas,Tempo also offers the two features above, low (average) latency and simple recovery –Tempo is the first leaderless protocol to provide all three. Tempo reuses most of Atlas techniques but, instead of maintaining explicit dependencies between commands, it assigns a scalar timestamp to each command and executes commands in the order of these timestamps. To determine when a command can be executed, each replica waits until the command’s timestamp is stable , i.e., all commands with a lower timestamp are known. Ordering commands in this way is used in many protocols [10,21–23]. A key novelty of Tempo is that both timestamping and stability detection are fault-tolerant and fully decentralized, which preserves the key benefits of leaderless SMR. We organize this thesis as follows. Chapter 2puts our work in context, covering the concept of SMR and relevant SMR protocols. Chapter 3and Chapter 4present the Atlas and Tempo protocols, respectively. Chapter 5evaluates the protocols experimentally. Finally, Chapter 6concludes this thesis. several data centers from the rest of the system. Atlas may block for the duration of the partitioning, which is unavoidable due to the CAP theorem [18]. 3 CHAPTER 1. INTRODUCTION List of Publications This thesis is based on the following papers published at EuroSys’20 [24] and at EuroSys’21 [25]: Vitor Enes, Carlos Baquero, Tuanir França Rezende, Alexey Gotsman, Matthieu Perrin, Pierre Sutra. State-Machine Replication for Planet-Scale Systems. European Conference on Computer Systems (EuroSys), Heraklion, Greece, 2020. Vitor Enes, Carlos Baquero, Alexey Gotsman, Pierre Sutra. Efficient Replication via Timestamp Stability. European Conference on Computer Systems (EuroSys), Online, United Kingdom, 2021. 4 2 Background In this chapter we put our work in context. We start by covering the system model in §2.1 and the concept of State-Machine Replication (SMR) in §2.2. This is followed by a detailed explanation of Paxos in §2.3, one of the most prominent SMR protocols. We then briefly cover other SMR protocols such as EPaxos and Caesar in §2.4 and Janus §2.5. Finally, we conclude the chapter with an overall comparison between Atlas,Tempo and the aforementioned protocols (§2.6), summarized in Table 2.1. 2.1 System Model We consider an asynchronous distributed system where processes may fail by crashing but do not behave maliciously. The system consists of Aprocesses I={1,...,A}, of which at most 5may fail. We assume that the set of processes is static. In this thesis we study how we can build a practical system that is geo-replicated among this set of processes. For that, we will leverage the concept of consensus , and so we start with a brief description of the consensus problem. Consensus Solving consensus entails having a set of processes reaching agreement on some value. Consensus is one of the most fundamental problems in distributed systems and has many applications, e.g., deciding whether to commit or abort a transaction, or decide which command to pick for a certain entry of a replicated log (§2.2). A protocol that solves consensus ensures the following properties: Validity. If a process decides on value E,thenEwas proposed by some process. Agreement. If a process decides on value Eand some other process decides on value E0,then E=E0. Liveness. All non-faulty process eventually decide on a value. Having a protocol that ensures these three properties is simple if processes are reliable, but it is quite challenging in the presence of faults. Famously, the FLP theorem [26] states that no protocol can solve consensus in an asynchronous system prone to faults. Hence, to ensure Liveness we assume that the 5 CHAPTER 2. BACKGROUND network is eventually synchronous [27]. This model guarantees that the system becomes synchronous after some unknown time, with message delays between non-failed processes bounded by an unknown constant Xand with process clocks tracking real time. These timing assumptions allow us to implement ⌦, the leader election failure detector [28], which ensures that from some point on, all correct processes nominate the same correct process as the leader. ⌦will be leveraged by the Paxos liveness algorithm presented later in this chapter (§2.3) and by the Tempo liveness algorithm (§4.2.7). 2.2 State-Machine Replication There is an inherent trade-off when building planet-scale replicated systems: if we cannot afford high latencies, then we have to settle for weak consistency; or similarly, if we have to provide strong consistency, then we have to give up low latencies. This trade-off has been famously stated by the CAP theorem [18]and some later variations such as the PACELC theorem [29] and the CAP critique [30] that better capture the latency aspect of the trade-off. Systems providing strong consistency prevent anomalies (e.g., unexpected behavior visible to users) and reduce programming complexity [4,31,32]. In this thesis, we focus on these. The strongest consistency level that replicated systems can offer is linearizability [6]. Informally, a system is linearizable if it appears as if commands submitted by clients are executed by a sequential process in an order consistent with the real-time order , i.e., the order of non-overlapping command invocations. For example, if some command Gby client returns before some command ~by client ⌫is submitted, then Gmust appear before ~in the linearization order. However, if ~is submitted in between Gbeing submitted and returned, then commands Gand ~can be ordered in any way (since their invocations overlap). State-Machine Replication (SMR) is a common way of implementing linearizable replicated services [5, 8,22]. A service is defined by a deterministic state machine accepting a set of commands. Each process maintains a local copy of the state machine and applies commands in an order determined by an SMR protocol. In its most basic form, an SMR protocol decides on a sequence of commands. This sequence is also known as the log . As processes start from the same initial state and execute commands in the order dictated by log, this simple mechanism ensures that processes stay in sync. 2.3 Paxos Paxos [7,33] is one of the most used SMR protocols. Detailed descriptions of the protocol can be found in many recent works [34–38]. In order to aid understanding Atlas (§3)andTempo (§4), which use Paxos as a building block, we also give a review of the protocol. For each entry of the log (§2.2), Paxos ensures that a single command is decided for it, i.e., that all processes agree on which command such entry should have. For this, Paxos uses a single-decree protocol 6 2.3. PAXOS that we present next. This protocol solves the consensus problem (§2.1). In Paxos the system consists of Aprocesses I={1,...,A}, of which at most 5=bA1 2cmay fail. Consensus protocol Algorithm 2.1 specifies the single-decree Paxos protocol at a process 82I. In Paxos there is an unbounded collection of ballot numbers. Ballot numbers are totally ordered and are allocated to processes round-robin. If for example we have 3 processes, we allocate ballots {1,4,7,...}to process 1, ballots {2,5,8,...} to process 2 and ballots {3,6,9,...}to process 3. Paxos ensures that a single command is decided by maintaining the following invariant: Invariant 2.1. Assume that command 2is decided at some ballot 1.Ifcommand20is decided at some ballot 10>1,then20=2. Paxos maintains this invariant with a two-phase protocol. In Paxos phase 1, a process leading with some ballot 1 (i) finds out which commands could have been decided at a ballot lower than 1(if any), and (ii) prevents any command to be decided at a ballot lower than 1. In Paxos phase 2, the process first computes its proposal for the command: either a command that could have been decided at a lower ballot (discovered during phase 1), or, if no command could have been decided, any command. The process then sends this proposal to all processes, and if enough processes accept it, the command is considered as decided. To better match the specification of Atlas and Tempo, Algorithm 2.1 denotes Paxos phase 1 messages by MRec and Paxos phase 2 messages by MConsensus1. Every process stores the ballot number bal it is currently participating in and the last ballot abal in which it accepted a consensus proposal (if any). Initially bal =abal =0.Inordertogetacommand decided, a process first calls the function recover (line 1). In this function, the process picks a ballot number it owns higher than any it participated so far (line 2) and starts Paxos phase 1 by sending an MRec message with this ballot to all processes (line 3). Upon the receipt of such a message, a process accepts the MRec message only if the ballot in the message is greater than its bal (line 5). The process sets its bal to the received ballot. This ensures that, during phase 2, the process will not accept a consensus proposal at a ballot lower than the ballot received. Then, it replies with an MRecAck message containing the command (cmd) and the ballot at which it was previously accepted (abal). Note that abal =0if the process has not yet accepted any consensus proposal for the command. Once a process receives MRecAck messages from a majority quorum (i.e., bA 2c+1processes) (line 8), it starts Paxos phase 2 by sending an MConsensus message containing its proposal for the command. To ensure Invariant 2.1, this proposal may have to be an earlier proposal by another process. Thus, the process first inspects whether a consensus proposal could have been decided by checking if any of the processes replied with a non-zero abal (line 10). If so, the process selects the consensus proposal 1MRec is commonly known by Paxos message 1a, MRecAck by 1b, MConsensus by 2a and MConsensusAck by 2b. 7 CHAPTER 2. BACKGROUND Algorithm 2.1: Single-decree Paxos protocol at process 82I. 1recover() 21 8+A(bbal1 Ac+1) 3send MRec(1)to I 4receive MRec(1)from 9 5pre: bal <1 6bal 1 7send MRecAck(abal,cmd,1)to 9 8receive MRecAck(019,29,1)from 892& 9pre: bal =1^|&|=bA 2c+1 10 if 9:2&.01:<0then 11 let :be such that 01:is maximal 12 send MConsensus(2:,1)to I 13 else 14 send MConsensus(some_cmd(),1)to I 15 receive MConsensus(2,1)from 9 16 pre: bal 1 17 cmd 2;bal 1;abal 1 18 send MConsensusAck(1)to 9 19 receive MConsensusAck(1)from & 20 pre: bal =1^|&|=bA 2c+1 21 decide(cmd) accepted at the highest ballot (line 11). If no consensus proposal could have been accepted before (i.e., if all processes replied with a zero abal), then the process is free to propose any command (e.g., a command submitted by a client) (line 14). A process accepts an MConsensus message only if its bal is not greater than the ballot in the message (line 16). Then it stores the proposal in cmd, sets bal and abal to the ballot in the message, and replies with an MConsensusAck message. Note that abal <0after a process accepts a consensus proposal. As mentioned above, this fact is used by processes executing the MRecAck handler. Once a process gathers MConsensusAck messages by a majority quorum (line 20), it can finally decide this proposal for the command (line 21). This is safe because any majority quorum used for Paxos phase 1 must intersect with any majority quorum used for Paxos phase 2. Thus, after Paxos phase 2, the process is sure that any other process trying to decide a command will find out about this proposal during Paxos phase 1. Skipping Paxos phase 1 By design, line 2only computes ballots greater than A. This allows us to have one special ballot B<A that can skip Paxos phase 1. During Paxos phase 1 with some ballot 1, a process tries to find out about proposals smaller than 1that could have been accepted. Because B<Aand ballots computed in line 2 are greater than A, no proposal smaller Bcan ever be accepted, and thus this step is unnecessary for 8 2.3. PAXOS ballot B. Moreover, when accepting a Paxos phase 1 message with some ballot 1, a process promises to not accept a consensus proposal with a lower ballot. Again, this is not necessary for ballot Bas ballots from line 2are greater than A. For these two reasons, process Bleading with ballot Bcan safely skip Paxos phase 1. In this case, process Bdoes not have to call recover in order to later propose a command – it can simply send an MConsensus message containing its proposal. This mechanism is leveraged by Atlas (§3)andTempo (§4) in their slow paths. Multi-Paxos optimization In Paxos, a command is decided after 4 message delays – 2 message delays for each phase of Paxos. However, practical implementations of Paxos employ the Multi-Paxos [7,33] optimization which allows commands to be decided after only 2 message delays. First, note that Paxos phase 1 can be performed for any entry of the log independently of the command to be proposed for such entry. Thus, in Multi-Paxos, a process performs phase 1 for all entries of the log. If it succeeds, this process is now called the leader – this mechanism is called leader election . When the elected leader receives a command issued by a client, it first selects the next available log entry, and then simply performs Paxos phase 2 by sending an MConsensus message containing the command received. Once it gathers enough MConsensusAck messages, it decides the command as usual, effectively reducing the number of message delays from 4 to 2. Flexible Paxos optimization As observed in Flexible Paxos (FPaxos) [35], Paxos safety relies on the fact that any phase 1 quorum intersects with any phase 2 quorum. Paxos ensures this intersection requirement with majority quorums for both phases, but the requirement can be met with other quorum sizes. For example, if we have A5processes for phase 1 and 5+1processes for phase 2, then any phase 1 quorum must also intersect with any phase 2 quorum. In this scheme, if 5=bA1 2c(as assumed by Paxos), we have again majority quorums for both phases (when Ais odd). However, if we allow 5to be any value such that 15b A1 2c, we can have one of the Paxos phases with a much smaller quorum. For example, with A=7and 5=1, FPaxos phase 1 quorums would be of size A5=6and phase 2 quorums of size 5+1=2. The above observation in FPaxos allows the leader elected in phase 1 to contact fewer processes during phase 2. In planet-scale systems these processes might be located far away from the leader, and thus this strategy can reduce the time the leader takes to complete phase 2. For this reason, the overall protocol latency may be lowered since, in Multi-Paxos, Paxos phase 2 is much more frequent than Paxos phase 1. In Atlas (§3)andTempo (§4) we leverage this observation to reduce the quorum size of slow paths at the expense of larger quorums during recovery. 9 CHAPTER 2. BACKGROUND Algorithm 2.2: Paxos liveness protocol at process 82I. 22 periodically 23 if leader =8^(bal =0_bal _ leader(bal)<8)then 24 recover() 25 receive MConsensus(_,1)or MRec(1)from 9 26 pre: bal >1 27 send MRecNAck(bal)to 9 28 receive MRecNAck(1) 29 pre: leader =8^bal <1 30 bal 1 31 recover() 32 bal_leader(1) 33 return 1A⇤j11 Ak Liveness protocol The Paxos rules we have covered so far ensure that the protocol is always safe, independently of how the whole system behaves (e.g., which processes call function recover, whether messages are lost, duplicated, reordered, and so on). However, if processes are allowed to call recover in an unconstrained way, the protocol is not live, i.e., it may happen that no command is ever decided (even in a synchronous network). Algorithm 2.2 addresses this issue and specifies a liveness protocol for Paxos. Again, we give a detailed description of this mechanism as it is used as a building block by Tempo in its liveness protocol (§4.2.7). This mechanism uses ⌦, the leader election failure detector [28], which ensures that from some point on, all correct processes nominate the same correct process as the leader (§2.1). In Algorithm 2.2, the variable leader denotes the current leader nominated at process 8. We say that leader stabilizes when it stops changing at all correct processes. Algorithm 2.2 refines Algorithm 2.1 by specifying a particular policy for invoking recover at a process 82I. Process 8is allowed to invoke recover at line 23 only if it is the leader according to leader. Furthermore, it only invokes recover at line 23 either if it has not yet participated in consensus (i.e., bal = 0) or, if it did, the consensus was lead by another process (i.e., bal_leader(bal)<8). In particular, process 8does not invoke recover at line 23 if it is the leader of bal (i.e., if bal_leader(bal)=8). This ensures that process 8does disrupt a recovery lead by itself. For a leader to make progress with some MRec(1)message, it is required that bA 2c+1processes (line 8) have their bal <1(line 5). This may not always be the case because, before the variable leader stabilizes, any process can invoke recover at line 23 if it thinks it is the leader. To help the leader select a high enough ballot, and thus ensure it will make progress, we introduce a new message type, MRecNAck. A process sends an MRecNAck(bal)at line 27 when it receives an MConsensus(_,1)or MRec(1) (line 25) with some ballot 1lower than its bal (line 26). When process 8receives an MRecNAck(1)with some ballot number 1higher than its bal, if it is still the leader (line 29), it joins ballot 1(line 30)and 10 3.2. THE ATLAS PROTOCOL the command may return the outcome of 2to the client. Without loss of generality, we assume that each submitted command is unique. The strongest property a replicated service implemented using SMR may satisfy is linearizability (§2). As observed in [13,14], for the replicated service to be linearizable, the SMR protocol does not need to ensure that commands are executed at processes in the exact same order: it is enough to agree on the order of non-commuting commands. We now give the specification of the SMR protocol. We say that commands 2and 3 commute if in every state Bof the state machine: (i) executing 2followed by 3or 3followed by 2in Bleads to the same state; and (ii) 2returns the same response in Bas in the state obtained by executing 3in B, and vice versa. If commands 2and 3do not commute, we say that they conflict 1and denote it by conflict(2,3).Given two commands 2and 3, we write 27!83when conflict(2,3)and 2is executed before 3at some process 82I. We also define the following real-time order :2{3if the command 2returns before the command 3was submitted. Let 7! =(–82I7!8)[{. An SMR protocol ensures the following properties: Validity. If a process executes some command 2,thenitexecutes2at most once and only if 2was submitted before. Ordering. The relation 7! is acyclic. The Ordering property enforces that conflicting commands are executed in a consistent manner across the system. In particular, it prevents two conflicting commands from being executed in contradictory orders by different processes. If the SMR protocol satisfies the above properties, then the replicated service implemented using it is linearizable. In the following sections we present Atlas, which satisfies the above specification. 3.2 The Atlas Protocol To aid understanding, we first illustrate by example the message flow of the Atlas protocol, which corresponds to a common structure of leaderless SMR protocols [11]. We then describe the protocol in detail (§3.2.1-§3.2.5). Figure 3.1 illustrates how Atlas processes two conflicting commands, aand b, with A=5processes and at most 5=2failures. At a given process, a command usually goes through several phases :the initial phase START,thenCOLLECT,COMMIT and EXECUTE (an additional phase RECOVER is used when handling failures). We summarize these phases and allowed phase transitions in Figure 3.2. Command astarts its journey when submit(a)is invoked at process 1. We call process 1the initial coordinator of a. This coordinator is initial because, if it fails or is slow, another process may take over. Command athen enters the COLLECT phase at process 1, whose goal is to compute the set of commands 1Detecting if two commands conflict must be possible without executing them. In practice, this information can often be extracted from the API provided by the replicated service. In cases when such inference is infeasible, it is always safe to consider that a pair of commands conflict. 17 CHAPTER 3. ATLAS: LOW-LATENCY, SIMPLE LEADERLESS CONSENSUS fast quorums 1 2 3 4 5 submit(a) submit(b) recover(a) MCollect MCollect a!b a!b a!b b!a consensus on dep[a] fast path slow path dep[a]={b} dep[a]=ú dep[b]={a} dep[a]=ú dep[a]=ú dep[b]={a} MCommit MCommit execute(a)execute(b) execute(a);execute(b) Figure 3.1: Example of processing two conflicting commands aand bin Atlas with A=5processes and up to 5=2failures. We omit the messages implementing consensus and depict this step abstractly by the consensus box. that are dependencies of a, denoted by dep[a]. These dependencies are later used to determine the order of execution of conflicting commands. To compute dependencies, process 1sends an MCollect message containing command ato a fast quorum of processes, which is at least a majority but may be bigger. In our example the fast quorum picked by 1is {1,2,3,4}. Each process in the fast quorum returns the set of commands conflicting with athat it received before a. In Figure 3.1,!indicates the order in which processes receive commands. For instance, process 4 receives bfirst, whereas 1,2and 3do not receive any command before a. Based on the replies, process 1computes the value of dep[a]as {b}, the union of all conflicting commands reported. If a coordinator of a command is suspected to have failed, another process may try to take over. In Figure 3.1, process 2suspects 1and becomes another coordinator of a, denoted by recover(a). Process 2contacts a majority quorum of processes {2,3,5}and computes its own version of the dependencies of a:dep[a]=ú. Dependencies are used to determine the order in which conflicting commands are executed, and all processes have to execute conflicting commands in the same order. To ensure this, the coordinators of command aneed to reach a consensus on the value of dep[a]. This is implemented using an optimized variant of single-decree Paxos (§2.3), with all Aprocesses acting as acceptors. In our example, this makes the processes agree on dep[a]=ú. The use of consensus represents the slow path of the protocol. If a coordinator can ensure that all the values that can possibly be proposed to consensus are the same, then it can take the fast path of the protocol, avoiding the use of consensus. In Figure 3.1, this is the case for process 4coordinating command b. For a process to take the fast path, we require it to receive a response from every process in the fast quorum, motivating the name of the latter. After consensus or the shortcut via the fast path, a coordinator of a command sends its final dependencies to other processes in an MCommit message. A process stores these dependencies and marks the command as having entered the COMMIT phase. A command can be executed (and thereby transition to the EXECUTE phase) only after all its dependencies are in the COMMIT or EXECUTE phases. Since in our example dep[a]=ú, processes can execute command aright after receiving its final dependencies (ú). This is exploited by processes 1and 2in Figure 3.1.However,asdep[b]={a}, processes must 18 3.2. THE ATLAS PROTOCOL START COLLECT RECOVER COMMIT EXECUTE ww'' // ✏✏''ww ✏✏ Figure 3.2: Command journey through phases in Atlas. delay the execution of buntil ais executed. This is the case for processes 3,4and 5in Figure 3.1.Such an execution mechanism guarantees that the conflicting commands aand bare executed in the same order at all processes. 3.2.1 Commit Protocol Algorithm 3.1 specifies the Atlas commit protocol at process 82Iin the failure-free case. We assume that self-addressed protocol messages are delivered immediately. Start phase A client submits a command 22Cby invoking submit(2)at one of the processes running Atlas, which will serve as the initial command coordinator. When submit(2)is invoked at a process 8(line 1), this coordinator first assigns to command 2a unique identifier using a function next_id. In the following we denote the set of all identifiers by D. We summarize in Table 3.1 the data maintained by each process for a command with identifier id 2D. In particular, the mapping cmd stores the payload of the command, and the mapping phase tracks the progress of the command through phases. For brevity, the name of the phase written in lower case also denotes all the identifiers in that phase, e.g., start ={id 2D|phase[id]=START}. Once the coordinator assigns an identifier to 2, the command starts its COLLECT phase, whose goal is to compute a set of identifiers that are the dependencies of 2. At the end of this phase, the coordinator sends an MCommit(id,2,⇡)message including the computed dependencies ⇡. Before this, it agrees with other possible coordinators on the same final value of ⇡, resulting in the following invariant. Invariant 3.1. For any two messages MCommit(id,2,⇡)and MCommit(id,20,⇡0)sent, 2=20and ⇡=⇡0. Hence, each identifier is associated with a unique command and final set of dependencies. The key property of dependencies is that, for any two distinct conflicting commands, one has to be a dependency of the other. This is stated by the following invariant. 19 CHAPTER 3. ATLAS: LOW-LATENCY, SIMPLE LEADERLESS CONSENSUS Algorithm 3.1: Atlas commit protocol at process 82I. 1submit(2) 2id next_id() 3past conflicts(2) 4& fast_quorum(8) 5send MCollect(id,2,past,&)to & 6receive MCollect(id,2,past,&)from 9 7pre: id 2start 8dep[id] conflicts(2)[past 9cmd[id] 2;quorum[id] & 10 phase[id] COLLECT 11 send MCollectAck(id,dep[id]) to 9 12 receive MCollectAck(id,dep9)from 892& 13 pre: id 2collect ^&=quorum[id] 14 ⇡ –&dep 15 if –&dep =– 5&dep then 16 send MCommit(id,cmd[id],⇡)to I 17 else 18 send MConsensus(id,cmd[id],⇡,8)to I 19 receive MConsensus(id,2,⇡,1)from 9 20 pre: bal[id]1 21 cmd[id] 2;dep[id] ⇡ 22 bal[id] 1;abal[id] 1 23 send MConsensusAck(id,1)to 9 24 receive MConsensusAck(id,1)from & 25 pre: bal[id]=1^|&|=5+1 26 send MCommit(id,cmd[id],dep[id]) to I 27 receive MCommit(id,2,⇡) 28 pre: id 8commit [execute 29 cmd[id] 2;dep[id] ⇡;phase[id] COMMIT 30 conflicts(2) 31 {id 8start |conflict(2,cmd[id])} Invariant 3.2. Assume that messages MCommit(id,2,⇡)and MCommit(id0,20,⇡0)have been sent. If id <id0and conflict(2,20)then either id02⇡or id 2⇡0, or both. This invariant is key to ensure that conflicting commands are executed in the same order at all processes, since we allow processes to execute commands that are not a dependency of each other in any order. We next explain how Atlas ensures the above invariants. Collect phase To compute the dependencies of a command 2, its coordinator first computes the set of commands it knows about that conflict with 2(denoted by past, line 3) using a function conflicts. The coordinator 20 3.2. THE ATLAS PROTOCOL Table 3.1: Atlas variables at a process. cmd[id] noOp 2CCommand phase[id] START Phase dep[id] ú ✓DDependency set quorum[id] ú ✓IFast quorum bal[id] 02NCurrent ballot abal[id] 02NLast accepted ballot then picks a fast quorum &of size bA 2c+5that includes itself (line 4)andsendsanMCollect message with the information it computed to all processes in &. Upon receiving an MCollect message from the coordinator, a process in the fast quorum computes its contribution to 2’s dependencies as the set of commands that conflict with 2, combined with past (line 8). The process stores the computed dependencies, command 2and the fast quorum &in mappings dep,cmd and quorum, respectively, and sets the command’s phase to COLLECT. The process then replies to the coordinator with an MCollectAck message, containing the computed dependencies (line 11). Once the coordinator receives an MCollectAck message from all processes in the fast quorum (line 13), it computes the dependencies for the command as the union of all reported dependencies ⇡=–&dep =–{dep9|92&}(line 14). Since a fast quorum contains at least a majority of processes, the following property implies that this computation maintains Invariant 3.2. Property 3.1. Assume two conflicting commands with identifiers 83 and 830and dependencies ⇡and ⇡0computed as in line 14 over majority quorums. Then either id02⇡or id 2⇡0, or both. Proof. Assume that the property does not hold: there are two conflicting commands with distinct identifiers 83 and 830and dependencies ⇡and ⇡0such that 8308⇡and 83 8⇡0. We know that ⇡was computed over some majority &and ⇡0over some majority &0.Since8308⇡,wehave: (i) the majority &observed 83 before 830. Similarly, since 83 8⇡0: (ii) the majority &0observed 830before 83.However, as majorities &and &0must intersect, we cannot have both (i) and (ii) , which yields a contradiction. ⇤ After computing the command’s dependencies, its coordinator decides to either take the fast path (line 15) or the slow path (line 17). Both fast and slow paths end with the coordinator sending an MCommit message containing the command and its final dependencies. Slow path If the coordinator of a command is suspected to have failed, another process may try to take over its job and compute a different set of dependencies. Hence, before an MCommit message is sent, processes must reach an agreement on its contents to satisfy Invariant 3.1. They can always achieve this by running 21 CHAPTER 3. ATLAS: LOW-LATENCY, SIMPLE LEADERLESS CONSENSUS a consensus protocol – this is the slow path of Atlas. Consensus is implemented using single-decree (Flexible) Paxos [35]. For each identifier we allocate ballot numbers to processes round-robin, with ballot 8reserved for the initial coordinator 8and ballots higher than Afor processes that try to take over. Every process stores for each identifier id the ballot number bal[id]it is currently participating in and the last ballot abal[id]in which it accepted a proposal (if any). Initially, bal[id]=abal[id]=0. When the initial coordinator 8decides to go onto the slow path, it performs an analog of Paxos Phase 2: it sends an MConsensus message with its proposal and ballot 8to a slow quorum that includes itself2. Following Flexible Paxos [35], the size of the slow quorum is only 5+1, rather than a majority like in classical Paxos. This minimizes the additional latency incurred on the slow path in exchange for using larger quorums in recovery (as described below). As usual in Paxos, a process accepts an MConsensus message only if its bal[id]is not greater than the ballot in the message (line 20). Then it stores the proposal, sets bal[id]and abal[id]to the ballot in the message, and replies to the coordinator with MConsensusAck. Once the coordinator gathers 5+1such replies (line 25), it is sure that its proposal will survive the allowed number of failures 5, and it thus broadcasts the proposal in an MCommit message (line 26). Fast path The initial coordinator of a command can avoid consensus when it can ensure that any process performing recovery will propose the same set of dependencies to consensus [40] – this is the fast path of Atlas, in which a command is committed after a single round trip to the closest fast quorum (line 16). In order to take the fast path, previous SMR protocols, such as Generalized Paxos [14] and EPaxos [11], require fast-quorum replies to match exactly. One of the key innovations of Atlas is that it is able to take the fast path even if this is not the case, e.g., when conflicting commands are submitted concurrently. In more detail, the coordinator takes the fast path if every dependency reported by some fast-quorum process is actually reported by at least 5such processes. This is expressed by the condition –&dep = – 5&dep in line 15,where – 5&dep ={id 2D|count(id)5}; count(id)=|{92&|id 2dep9}|. Figure 3.3 contains several examples that illustrate the flexibility of the above fast-path condition. All examples consider A=5processes while tolerating varying numbers of faults 5. The example in Figure 3.3a considers Atlas 5=2. The coordinator of some command, process 1, picks a fast quorum &={1,2,3,4}of size bA 2c+5=4.Itreceivesrepliesdep1={a},dep2={a,b,c},dep3={a,b,d}, dep4={a,c,d}. The coordinator then computes – 2&dep ={a,b,c,d}, i.e., all the dependencies reported at least twice. Since –&dep =– 2&dep, the coordinator takes the fast path. This is not the 2As noted in §2.3, the initial coordinator 8can safely skip Paxos Phase 1: since processes perform recovery with ballots higher than A, no proposal with a ballot lower than 8can ever be accepted. 22 3.2. THE ATLAS PROTOCOL 1 {a} 2 {a,b,c} 3 {a,b,d} 5 4 {a,c,d} (a) 3Atlas 5=2 7 matching replies 1 ú 2 ú 3 ú 5 4 {b} (b) 7Atlas 5=2 7 matching replies 1 {a} 2 {a,b} 3 5 4 {a,c} (c) 3Atlas 5=1 7 matching replies 1 {a} 2 {a} 3 5 4 {a} (d) 3Atlas 5=1 3 matching replies Figure 3.3: Examples in which the fast path is taken 3or not 7, for both Atlas and protocols that require matching replies from fast-quorum processes, such as EPaxos [11]. All examples consider A=5 processes while tolerating 5faults. The coordinator is always process 1, and circles with a solid line represent the processes that are part of the fast quorum. Next to each process we depict the set of dependencies sent to the coordinator (e.g., {a,b}). case for the example in Figure 3.3b where –&dep ={b}<ú=– 2&dep (bis excluded from – 2&dep because count(b)=1). In this case the coordinator has to take the slow path. Back in Figure 3.1 we had the same situation: coordinator 1had to take the slow path because dependency bwas declared solely by process 4. On the other hand, coordinator 4was able to take the fast path, because dependency awas declared by 35processes: 1,2and 3. Notice that in Figure 3.3a, the coordinator takes the fast path even though dependencies reported by processes do not match, a situation which may arise when conflicting commands are submitted concurrently. Furthermore, when 5=1we have {id 2D|count(id)<5}=ú, so that the fast-path condition in line 15 always holds. Hence, Atlas 5=1 always takes the fast path, as is the case in Figures 3.3c and 3.3d. In contrast, EPaxos is able to take the fast path only in Figure 3.3d, since it is the only example in which fast-quorum replies match. 3.2.2 Recovery Protocol The initial coordinator of a command may fail or be slow to respond, in which case Atlas allows a process to take over its role and recover the command and its dependencies. We start by describing the idea of the most subtle part of this mechanism – recovering decisions reached by failed coordinators via the fast path. Let ⇡=–&dep =– 5&dep be some fast-path proposal (line 16). By definition of – 5&dep,each id 2⇡was reported in the MCollectAck message of at least 5fast-quorum processes. It follows that ⇡can be obtained without 51of those processes by taking the union of the dependencies reported by the remaining processes. Moreover, as the initial coordinator is always part of the fast quorum and each process in the quorum combines its dependencies with the ones declared by the coordinator (i.e., past in line 8), the latter is also not necessary to obtain ⇡. Thus, the proposal ⇡can be obtained without 5fast-quorum processes including the initial coordinator (e.g., if the processes fail), by combining 23 CHAPTER 3. ATLAS: LOW-LATENCY, SIMPLE LEADERLESS CONSENSUS Algorithm 3.2: Atlas recovery protocol at process 82I. 32 recover(id) 33 1 8+A(bbal[id]1 Ac+1) 34 send MRec(id,cmd[id],1)to I 35 receive MRec(id,_,_)from 9 36 pre: id 2commit [execute 37 send MCommit(id,cmd[id],dep[id]) to 9 38 receive MRec(id,2,1)from 9 39 pre: bal[id]<1^id 8commit [execute 40 if bal[id]=0^id 2start then 41 dep[id] conflicts(2);cmd[id] 2 42 bal[id] 1 43 phase[id] RECOVER 44 send MRecAck(id,cmd[id],dep[id],quorum[id],abal[id],1)to 9 45 receive MRecAck(id,29,dep9,&0 9,019,1)from 892& 46 pre: bal[id]=1^|&|=A5 47 if 9:2&.01:<0then 48 let :be such that 01:is maximal 49 send MConsensus(id,2:,dep:,1)to I 50 else if 9:2&.&0 :<úthen 51  &\&0 : 52 B initial?(id)2 53 &0 if Bthen &else  54 ⇡ –&0dep 55 send MConsensus(id,2:,⇡,1)to I 56 else send MConsensus(id,noOp,ú,1)to I the dependencies reported by the remaining (bA 2c+5)5=bA 2cprocesses. The following property captures this observation. Property 3.2. Any fast-path proposal can be obtained by taking the union of the dependencies sent in MCollectAck by at least bA 2cfast-quorum processes that are not the initial coordinator. As an example, assume that after the fast path is taken in Figure 3.3a,5=2processes inside the fast quorum fail, one of them being the coordinator, process 1. Independently of which bA 2c=2fastquorum processes survive, the proposal is always recovered by set union: –{2,3}dep =–{2,4}dep = –{3,4}dep ={a,b,c,d}. In the case of Figure 3.3b it is unsafe to take the fast path since the proposal may not be recoverable: the failure of process 4would lead to losing the dependency b, since this dependency was reported exclusively by this process. 24 3.2. THE ATLAS PROTOCOL Recovery in detail The recovery protocol of Atlas at a process 82Iis given in Algorithm 3.2. We use initial(id)to denote a function that extracts from the command identifier id its initial coordinator. A process takes over as the coordinator for some command with identifier id by calling recover(id) (line 32 in Algorithm 3.2). In order to find out if a decision on the dependencies of id has been reached in consensus, the new coordinator first performs an analog of Paxos Phase 1 (§2.3). It picks a ballot number it owns higher than any it participated in so far (line 33) and sends an MRec message with this ballot to all processes. Upon the receipt of such a message, in case id is already committed or executed (line 36), the process notifies the new coordinator with an MCommit message. Otherwise, as is standard in Paxos, the process accepts the MRec message only if the ballot in the message is greater than its bal[id](line 39). In this case, if the process is seeing id for the first time (line 40), it computes its contribution to id’s dependencies as the set of conflicting commands (line 41). Then, the process sets bal[id]to the new ballot and phase[id]to RECOVER. Finally, the process replies with an MRecAck message containing all the information it has regarding id: the corresponding command (cmd), its current set of dependencies (dep), the ballot at which these were previously accepted (abal), and the fast quorum (quorum). Note that quorum[id]=úif the process did not see the initial MCollect message, and abal[id]=0if the process has not yet accepted any consensus proposal. In the MRecAck handler (line 45), the new coordinator computes its proposal given the information provided by processes and sends this proposal in an MConsensus message to all processes. As in Flexible Paxos, the new coordinator waits for A5MRecAck messages. This guarantees that, if a quorum of 5+1processes accepted an MConsensus message with a proposal (which could have thus been sent in an MCommit message), the new coordinator will find out about this proposal. To maintain Invariant 3.1, if any process previously accepted a consensus proposal (line 47), by the standard Paxos rules [7,35], the coordinator selects the proposal accepted at the highest ballot (line 48). If no consensus proposal has been accepted before, the new coordinator checks whether any of the processes that replied has seen the initial MCollect message, by looking for any non-empty fast quorum (line 50). If the fast quorum is known, the new coordinator first computes at line 51 the set of processes that belong both to the recovery quorum &and the fast quorum &0 :. Then, depending on whether the initial coordinator replied or not, there are two possible cases that we describe next. 1) The initial coordinator replies to the new one ( B=tAD4 , line 52). In this case the initial coordinator has not taken the fast path before receiving the MRec message from the new one, as it would have replied with MCommit instead of MRecAck (line 37). It will also not take the fast path in the future, since when processing the MRec message it sets the command phase to RECOVER (line 43), which invalidates the MCollectAck precondition (line 13). Since the initial coordinator never takes the fast path, the new coordinator can choose the command’s dependencies in any way, as long as it maintains Invariant 3.2. By Property 3.1, this is satisfied if the coordinator chooses the set union of the dependencies declared by 25 CHAPTER 3. ATLAS: LOW-LATENCY, SIMPLE LEADERLESS CONSENSUS at least a majority of processes. Hence, the new coordinator takes the union of the dependencies reported by the A5Ab A1 2cb A 2c+1processes in &(line 54). 2) The initial coordinator does not reply to the new one ( B=f0;B4 , line 52). In this case the initial coordinator could have taken the fast path and, if it did, the new coordinator must propose the same dependencies. Given that the recovery quorum &has size A5and the fast quorum &0 :has size bA 2c+5, the set of processes =&\&0 :(line 51) contains at least bA 2cfast-quorum processes (distinct from the initial coordinator, as it did not reply). Furthermore, recall that when a process from  replies to the new coordinator, it sets the command phase to RECOVER (line 43), which invalidates the MCollect precondition (line 7). Hence, if the initial coordinator took the fast path, then each process in must have processed its MCollect before the MRec of the new coordinator, and reported in the latter the dependencies from the former. Then using Property 3.2, the new coordinator recovers the fast-path proposal by taking the union of the dependencies from the processes in (line 54). It can be shown that, even if the initial coordinator did not take the fast path, this computation maintains Invariant 3.2, despite  containing only bA 2cprocesses and Property 3.1 requiring a majority of them. This is for the same reason this number of processes is sufficient in Property 3.2: dependencies declared by the initial coordinator are included into those declared by other fast-quorum processes (line 8). It remains to address the case in which the process performing the recovery observes that no process saw the initial fast quorum, and consequently the submitted command (line 56). For instance, suppose that process 8sends an MCollect(id,2,_,_)only to process 9and then fails. Further, assume that 9receives another MCollect(_,20,_,_)from process :, replies with a dependency set that includes the identifier id of 2, and also fails. Now, process :cannot execute 20without executing 2(since 2is a dependency of 20), and it cannot execute 2because its payload has been lost. We solve this issue similarly to EPaxos: if a process takes over as the new coordinator and cannot find the associated payload, it may replace it by a special noOp command (line 56) that is not executed by the protocol and conflicts with all commands. With this, the final command for some identifier can take two possible values: the one submitted (line 1)ornoOp. It is due to this that we include the command payload in addition to its dependencies into consensus messages associated with a given identifier (e.g., line 18), thus ensuring that a unique payload will be chosen (Invariant 3.1). Due to the possible replacement of a command by anoOp, the protocol actually ensures the following weakening of Invariant 3.2, which is still sufficient to establish its correctness. Invariant 3.20.Assume that messages MCommit(id,2,⇡)and MCommit(id0,20,⇡0)have been sent. If id <id0,conflict(2,20),2<noOp and 20<noOp,theneitherid02⇡or id 2⇡0, or both. 3.2.3 Execution Protocol Algorithm 3.3 describes a background task employed by Atlas that is responsible for executing commands after they are committed. This task runs in an infinite loop trying to execute a batch of commands. We define a batch as the smallest set of committed identifiers (✓commit such that, for each identifier 26 3.3. CORRECTNESS or MRecAck(id0,_,dep0,_,0,_).IfMCollectAck(id,dep)is sent, it must be in response to MCollect(id,3,_,_), and by Invariant 3.6 we have 3=2. Similarly, if MCollectAck(id0,dep0) is sent, it must be in response to MCollect(id0,30,_,_), and by Invariant 3.6 we have 30=20.If MRecAck(id,_,dep,_,0,_)is sent, it must be in response to MRec(id,3,_), and by Invariant 3.6 we have 32{2,noOp}.IfMRecAck(id0,_,dep0,_,0,_)is sent, it must be in response to MRec(id0,30,_), and by Invariant 3.6 we have 302{20,noOp}. Without loss of generality, assume that process 9sends the message about id before the message about id0. We prove that id 2dep0. We have four cases depending on which message (MCollectAck or MRecAck) is sent for each identifier: 1) Process 9 sends MCollectAck(id,dep) and then MCollectAck(id0,dep0) . When handling MCollect(id,2,_,_), process 9stores 2in cmd[id]. By Invariant 3.6,cmd[id]can only change to noOp. When handling MCollect(id0,20,_,_), since cmd[id]2{2,noOp}and noOp conflicts with all commands, we have id 2conflicts(20)in line 8, and thus id 2dep0in MCollectAck(id0,dep0),as required. 2) Process 9 sends MCollectAck(id,dep) and then MRecAck(id0, _ ,dep0, _ ,0, _ ) . When handling MCollect(id,2,_,_), process 9stores 2in cmd[id]. By Invariant 3.6,cmd[id]can only change to noOp. When handling MRec(id0,30,_)with 302{20,noOp}we have two cases depending on phase[id0].Ifid02start, then since cmd[id]2{2,noOp}and noOp conflicts with all commands, we have id 2conflicts(30)in line 41.Ifid08start, then process 9is a member of the original fast quorum for id0and thus included id into dep[id0]when it processed MCollect(id0,20,_,_). Thus, in both cases id 2dep0in MRecAck(id0,_,dep0,_,0,_), as required. 3) Process 9 sends MRecAck(id, _ ,dep, _ ,0, _ ) and then MCollectAck(id0,dep0) . Analogous to the above. 4) Process 9 sends MRecAck(id, _ ,dep, _ ,0, _ ) and then MRecAck(id0, _ ,dep0, _ ,0, _ ) . Analogous to the above. ⇤ Proof of Invariant 3.2 0 . Assume that MCommit(id,2,⇡)and MCommit(id0,20,⇡0)have been sent with id <id0,2<noOp,20<noOp and conflict(2,20). The protocol structure ensures that ⇡=–&dep for &and dep given as parameters of handlers at lines 12 or 45, and the computation of ⇡ occurs at lines 14 or 54. We start by proving that there exists a quorum b &with |b &|b A 2c+1and c dep such that –&dep =–b &c dep, where each process 92b &computes its c dep9in either line 8or line 41 and sends it in either MCollectAck(id,c dep9)or MRecAck(id,_,c dep9,_,0,_). The computation of ⇡occurs either in the transition at line 14 or at line 54. If the computation of ⇡ occurs in the transition at line 14,then&is a fast quorum with size bA 2c+5. In this case, we let b &=& and c dep =dep.Since51, we have |b &|b A 2c+1, as required. If the computation of ⇡occurs at line 54, we have two situations depending on whether initial(id)replies to the new coordinator (line 52). If it does (B=tAD4,line52), then &is a recovery quorum of size A5. In this case, we let b &=&and c dep =dep.Since5b A1 2c, we have |b &|b A 2c+1, as required. If initial(id) 33 CHAPTER 3. ATLAS: LOW-LATENCY, SIMPLE LEADERLESS CONSENSUS does not reply (B=f0;B4, line 52), then &consists of the fast quorum members that are part of the recovery quorum (line 51). Given that fast quorum size is bA 2c+5and the recovery quorum size is A5, in this case &contains at least bA 2c+55=bA 2cfast quorum processes, and thus |&|b A 2c. Since ⇡is computed in the branch where the initial fast quorum is known (line 50), at least one of the fast quorum members in &must have computed its set of dependencies at line 8, including in its dependencies those reported by the original coordinator. In this case, we let b &=&[{initial(id)}, 892&.c dep9=dep9and c depinitial(id)be the set of dependencies sent by initial(id)in its MCollectAck(id,c depinitial(id))message. Since |&|b A 2cand initial(id)8&(as it did not reply), we have |b &|b A 2c+1, as required. Similarly to the above, we can also prove that there exists a quorum b &0with |b &0|b A 2c+1and d dep0 such that –&0dep0=–b &0d dep0, where each process 92b &0computes its d dep0 9in either line 8or line 41 and sends its d dep0 9in either MCollectAck(id0,d dep0 9)or MRecAck(id0,_,d dep0 9,_,0,_). We now prove that 8302⇡or 83 2⇡0. By contradiction, assume that id08⇡and id 8⇡0.Since 8308⇡, we have 892b &.id08c dep9. Similarly, since 83 8⇡0, we have 892b &0.id 8d dep0 9. Given that |b &|b A 2c+1and |b &0|b A 2c+1,b &and b &0must intersect. For this reason, there must exist a process ?2b &\b &0such that 8308c dep?and 83 8d dep0 ?. But this contradicts Invariant 3.14.⇤ We now prove that Atlas ensures Ordering. First we introduce the following two lemmas. Lemma 3.1. The relation –A 8=17!8is asymmetric. Proof. By contradiction, assume that for some processes 8and 9and conflicting commands 2and 20 with identifiers id and id0, we have 27!820and 207!92;then2<noOp and 20<noOp. By Validity we must have 8<9and 2<20. Assume first that 2and 20are executed at process 8in the same batch (. Then by Invariant 3.4 they also have to be executed at process 9in the batch (. Since inside a batch commands are ordered using the fixed order <on their identifiers, 2and 20have to be executed in the same order at the two processes: acontradiction. Assume now that 2and 20are not executed at process 8in the same batch. Then by Invariant 3.4 this also must be the case at process 9. Hence, Invariant 3.3 implies that id08dep[id]at process 8,and id 8dep[id0]at process 9. Then process 8received MCommit(id,2,⇡)with id08⇡, and process 9 received MCommit(id0,20,⇡0)with id 8⇡0, which contradicts Invariant 3.20.⇤ Lemma 3.2. Assume 217! ...7! 2=for =2. Whenever a process 8executes 2=, some process has already executed 21. Proof. We prove the lemma by induction on =. The base case of ==2directly follows from the definition of 7!.Take=>3and assume 217! ... 7! 2=17! 2=. Consider the moment when a process 8executes 2=. We want to show that by this moment some process has already executed 21. 34 3.4. SUMMARY AND RELATED WORK Since 2=17! 2=,either2=1{2=or 2=17!92=for some process 9. Consider first the case when 2=1{2=.Then2=1is executed at some process :before 2=is submitted and, hence, before 2=is executed at process 8. By induction hypothesis, 21is executed at some process before 2=1is executed at process :and, hence, before 2=is executed at process 8, as required. We now consider the case when 2=17!92=for some process 9. Since process 8executes 2=, we must have either 2=17!82=or 2=7!82=1. The latter case would contradict Lemma 3.1, so that 2=17!82=. By induction hypothesis, 21is executed at some process before 2=1is executed at process 8and, hence, before 2=is executed at process 8, as required. ⇤ Proof of Ordering. By contradiction, assume that 217! ... 7! 2==21for =2. Then some process executed 21. Consider the moment when the first process did so. By Lemma 3.2 some process has already executed 21before this, which yields a contradiction. ⇤ 3.4 Summary and Related Work The classical way of implementing SMR is by funnelling all commands through a single leader replica [7, 8,47,48], which can create a bottleneck. A way to mitigate this problem is to distribute the leader responsibilities round-robin among replicas, as done in Mencius [20]. However, this makes the system run at the speed of the slowest replica. Exploiting commutativity to improve the scalability of SMR was first proposed in Generalized Paxos [13] and Generic Broadcast [14]. These protocols still rely on a leader to order concurrent non-commuting commands, which also creates a bottleneck. The closest SMR protocol to Atlas is EPaxos [11], which is also leaderless and exploits commutativity. We compared Atlas with EPaxos in detail in §3.2.4. There have been two follow-up protocols to EPaxos, Alvin [12] and Caesar [10]. Atlas compares to these protocols similarly to EPaxos; in particular, both follow-ups have large fast quorums that depend on the overall number of processes only. Flexible Paxos [35] reduces the size of Paxos Phase 2 quorums to 5+1, a technique we also use on the slow path of Atlas. However, this technique is not directly applicable to computing dependencies via fast path, as required by leaderless SMR. To the best of our knowledge, Atlas is the first protocol to reduce the size of fast quorums to bA 2c+5. An approach to scaling SMR is to shard the state of the application being replicated and add crossshard coordination to preserve consistency [49]. Such approaches build on a non-sharded SMR protocol and are hence orthogonal to our proposal: Atlas can be combined with them to scale SMR even further. Protocols such as M2Paxos [50], WPaxos [51] and DPaxos [52] scale up SMR using a variation of the sharding approach. These protocols exploit access locality by optimizing for workloads where commands do not frequently access objects in multiple locations. 35 CHAPTER 3. ATLAS: LOW-LATENCY, SIMPLE LEADERLESS CONSENSUS There have been recent proposals of SMR protocols that improve scalability using special hardware capabilities, such as low-latency switches or RDMA [53–55]. However, currently these protocols work within a single data center only. Summary This chapter presented Atlas, the first leaderless SMR protocol parameterized with the number of allowed failures. Atlas is designed for planet-scale systems where concurrent site failures are rare. It employs three mechanisms to reduce latency: small quorums, a flexible fast-path condition, and non-fault-tolerant reads. Atlas leverages the same execution mechanism of EPaxos based on explicit dependencies. As highlighted in §2, this mechanism can result in pathological scenarios where the protocol continuously commits commands but can never execute them, even under a synchronous network [11,19]. In practice, this translates into a high tail latency (§5). In the next chapter we introduce Tempo, a leaderless protocol that leverages the ideas of Atlas presented in this chapter, but addresses these liveness and performance issues. 36 4 Tempo: Predictable Leaderless Consensus In this chapter we present Tempo, a leaderless SMR protocol that offers predictable performance . Existing leaderless SMR protocols suffer from drawbacks in the way they order commands. For example, EPaxos and Atlas maintain explicit dependencies between commands: a replica may execute a command only after all its dependencies get executed. These dependencies may form arbitrary long chains. As a consequence, in theory the protocols do not guarantee progress even under a synchronous network (§2). In practice, their performance is unpredictable, and in particular, exhibits a high tail latency [9,19]. These drawbacks carry over to the setting of partial replication where they are aggravated by the fact that commands span multiple machines. To lift these limitations, Tempo redesigns the way commands are typically ordered in leaderless SMR. Tempo assigns a scalar timestamp to each command and executes commands in the order of these timestamps. To determine when a command can be executed, each replica waits until the command’s timestamp is stable , i.e., all commands with a lower timestamp are known. Ordering commands in this way is used in many protocols [10,21–23]. A key novelty of Tempo is that both timestamping and stability detection are fault-tolerant and fully decentralized, which preserves the key benefits of leaderless SMR. This allows Tempo to offer low tail latency even in contended workloads, thus ensuring predictable performance. Tempo builds on Atlas and thus it also offers low (average) latency and simple recovery (§3). In more detail, each Tempo process maintains a local clock from which timestamps are generated. In the case of full replication, to submit a command a client sends it to the closest process, which acts as its coordinator . The coordinator computes a timestamp for the command by forwarding it to a quorum of replicas, each of which makes a timestamp proposal , and taking the maximum of these proposals. If enough replicas in the quorum make the same proposal, then the timestamp is decided immediately ( fast path ). If not, the coordinator does an additional round trip to the replicas to persist the timestamp ( slow path ); this may happen when commands are submitted concurrently. Thus, under favorable conditions, the replica nearest to the client decides the command’s timestamp in a single round trip. To execute a command, a replica then needs to determine when its timestamp is stable, i.e., it knows about all commands with lower timestamps. The replica does this by gathering information about which timestamp ranges have been used up by each replica, so that no more commands will get proposals in 37 CHAPTER 4. TEMPO: PREDICTABLE LEADERLESS CONSENSUS these ranges. This information is piggy-backed on replicas’ messages, which often allows a timestamp of a command to become stable immediately after it is decided. The above protocol easily extends to partial replication: in this case a command’s timestamp is the maximum over the timestamps computed for each of the partitions it accesses. We organize this chapter as follows. We start by covering the system model in §4.1. This is followed by a detailed explanation of the Tempo protocol in §4.2: we cover the single-partition commit protocol in §4.2.1, the execution protocol in §4.2.2, a comparison with Atlas, EPaxos and Caesar in §4.2.3,the multi-partition commit protocol in §4.2.5, the recovery protocol in §4.2.6, the liveness protocol in §4.2.7, and a few optimizations in §4.2.9.In§4.3 we show the correctness of Tempo. We conclude this chapter with a summary and related work in §4.4. 4.1 System Model We consider a general version of State-Machine Replication (SMR) (§3.1) where each machine replicates only a part of the service state – partial SMR (PSMR) [15,56,57]. We assume that the service state is divided into partitions , so that each variable defining the state belongs to a unique partition. Partitions are arbitrarily fine-grained: e.g., just a single state variable. Each command accesses one or more partitions. We assume that a process replicates a single partition, but multiple processes may be co-located at the same machine. Each partition is replicated at Aprocesses, of which at most 5may fail, where 5can be any value such that 15b A1 2c. We write I?for the set of all the processes replicating a partition ?, I2for the set of processes that replicate the partitions accessed by a command 2,andIfor the set of all processes. A PSMR protocol provides a command submit(2), which allows a process 8to submit a command 22Con behalf of a client. For simplicity, we assume that each command is unique and the process submitting it replicates one of the partitions it accesses: 82I2. For each partition ?accessed by 2,the protocol then triggers an upcall execute?(2)at each process storing ?, asking it to apply 2to the local state of partition ?.After2is executed by at least one process in each partition it accesses, the process that submitted the command aggregates the return values of 2from each partition and returns them to the client. PSMR ensures the highest standard of consistency of replicated data – linearizability [6]–which provides an illusion that commands are executed sequentially by a single machine storing a complete service state. To this end, a PSMR protocol has to satisfy the following specification. Given two commands 2and 3, we write 27!83when they access a common partition and 2is executed before 3at some process 82I2\I3. We also define the following real-time order :2{3if the command 2returns before the command 3was submitted. Let 7! =(–82I7!8)[{. A PSMR protocol ensures the following properties: 38 4.2. THE TEMPO PROTOCOL Validity. If a process executes some command 2,thenitexecutes2at most once and only if 2was submitted before. Ordering. The relation 7! is acyclic. Liveness. If a command 2is submitted by a non-faulty process or executed at some process, then it is executed at all non-faulty processes in I2. The Ordering property ensures that commands are executed in a consistent manner throughout the system [58]. For example, it implies that two commands, both accessing the same two partitions, cannot be executed at these partitions in contradictory orders. As usual, to ensure Liveness we assume that the network is eventually synchronous (§2.1). In the following sections we present Tempo, which satisfies the above specification. PSMR is expressive enough to implement a wide spectrum of distributed applications. In particular, it directly allows implementing one-shot transactions, which consist of independent pieces of code (such as stored procedures), each accessing a different partition [15,59,60]. It can also be used to construct general-purpose transactions [15,61]. Note that SMR is a special case of PSMR where each machine replicates the whole service state. In this case, however, the above PSMR specification is not equivalent to the SMR specification introduced in §3.1. The SMR specification in §3.1 is parameterized by a conflict relation which only prevents two commands 2and 3from being executed in contradictory orders by different processes when conflict(2,3). In particular, it allows two commands reading the same partition to be executed in contradictory orders (since read commands do not conflict with each other). On the other hand, the PSMR specification above enforces that any two commands accessing the same partition are executed in the the same order at all the processes replicating such partition, even when the commands do not conflict. We discuss the implications of this assumption at the end of §4.2.3. Lastly, the SMR specification in §3.1 does not include a Liveness property as the PSMR specification above. This is due to the fact that Tempo ensures Liveness, while Atlas does not. 4.2 The Tempo Protocol For simplicity, we first present the protocol in the case when there is only a single partition ?,andcover the general case in §4.2.5. We start with an overview of the single-partition protocol. To ensure the Ordering property of PSMR, Tempo assigns a scalar timestamp to each command. Processes execute commands in the order of these timestamps, thus ensuring that processes execute commands in the same order. To submit a command, a client sends it to a nearby process which acts as the coordinator for the command. The coordinator is in charge of assigning a timestamp to the command and communicating this timestamp to all processes. When a process finds out about the command’s timestamp, we say that the process commits the command. If the coordinator is suspected to have failed, another process takes over its role through a recovery mechanism (§4.2.6). Tempo ensures that, even in 39 CHAPTER 4. TEMPO: PREDICTABLE LEADERLESS CONSENSUS case of failures, processes agree on the timestamp assigned to the command, as stated by the following invariant. Invariant 4.1 (Timestamp agreement).Two processes cannot commit the same command with different timestamps. A coordinator computes a timestamp for a command as follows (§4.2.1). It first forwards the command to a fast quorum of bA 2c+5processes, including the coordinator itself. Each process maintains aClock?variable. When the process receives a command from the coordinator, it increments Clock? and replies to the coordinator with the new Clock?value as a timestamp proposal . The coordinator then takes the highest proposal as the command’s timestamp. If enough processes have made such a proposal, the coordinator considers the timestamp decided and takes the fast path : it just communicates the timestamp to the processes, which commit the command. The protocol ensures that the timestamp can be recovered even if the coordinator fails, thus maintaining Invariant 4.1. Otherwise, the coordinator takes the slow path , where it stores the timestamp at a slow quorum of 5+1processes using a variant of Flexible Paxos [35]. This ensures that the timestamp survives any allowed number of failures. The slow path may have to be taken in cases when commands are submitted concurrently to the same partition (however, recall that partitions may be arbitrarily fine-grained). Since processes execute committed commands in the timestamp order, before executing a command a process must know all the commands that precede it. Invariant 4.2 (Timestamp stability).Consider a command 2committed at 8with timestamp C. Process 8can only execute 2after its timestamp is stable , i.e., every command with a timestamp lower or equal to Cis also committed at 8. To check the stability of a timestamp C(§4.2.2), each process 8tracks timestamp proposals issued by other processes. Once the clocks (Clock?) at any majority of the processes pass C, process 8can be sure that new commands will get higher timestamps: these are computed as the maximal proposal from at least a majority, and any two majorities intersect. Process 8can then use the information gathered about the timestamp proposals from other processes to find out about all the commands that have got a timestamp lower than C. 4.2.1 Commit Protocol Algorithm 4.1 specifies the Tempo single-partition commit protocol at a process 8replicating a partition ?. We assume that self-addressed messages are delivered immediately. A command 22Cis submitted by a client by calling submit(2)at a process 8that replicates a partition accessed by the command (line 1). Process 8then creates a unique identifier id 2Dand a mapping Qfrom a partition accessed by the command to the fast quorum to be used at that partition. Because we consider a single partition for now, 40 4.2. THE TEMPO PROTOCOL START PAYLOAD RECOVER-R PROPOSE RECOVER-P COMMIT EXECUTE   oo//  ++ss ✏✏ Figure 4.1: Command journey through phases in Tempo. in what follows Qcontains only one fast quorum, Q[?]. Finally, process 8sends MSubmit(id,2,Q) to a set of processes I8 2, which in the single-partition case simply denotes {8}. A command goes through several phases at each process: from the initial phase START,toaCOMMIT phase once the command is committed, and an EXECUTE phase once it is executed. We summarize these phases and allowed phase transitions in Figure 4.1. A mapping phase at a process tracks the progress of a command with a given identifier through phases. As before, the name of the phase written in lower case denotes all the commands in that phase, e.g., start ={id 2D|phase[id]=START}. We also define pending as follows: pending =payload [propose [recoverp [recoverr. We summarize in Table 4.1 the data maintained by each process for a command with identifier id,whereOdenotes the set of all partitions. Note that Table 4.1 also contains data that is not specific to any command, such as Clock?. Start phase When a process receives an MSubmit message, it starts serving as the command coordinator (line 5). The coordinator first computes its timestamp proposal for the command as Clock?+1. After computing the proposal, the coordinator sends an MPropose message to the fast quorum Q[?]and an MPayload message to the remaining processes. Since the fast quorum contains the coordinator, the coordinator also sends the MPropose message to itself. As mentioned earlier, self-addressed messages are delivered immediately. Payload phase Upon receiving an MPayload message (line 9), a process simply saves the command payload in a mapping cmd and sets the command’s phase to PAYLOAD. It also saves Qin a mapping quorums. This is necessary for the recovery mechanism to know the fast quorum used for the command (§4.2.6). Propose phase Upon receiving an MPropose message (line 12), a fast-quorum process also saves the command payload and fast quorums, but sets its phase to PROPOSE. Then the process computes its own timestamp proposal 41 CHAPTER 4. TEMPO: PREDICTABLE LEADERLESS CONSENSUS Algorithm 4.1: Tempo commit protocol at process 82I?. 1submit(2) 2pre: 82I2 3id next_id();Q fast_quorums(8,I2) 4send MSubmit(id,2,Q) to I8 2 5receive MSubmit(id,2,Q) 6C Clock?+1 7send MPropose(id,2,Q,C)to Q[?] 8send MPayload(id,2,Q) to I?\Q[?] 9receive MPayload(id,2,Q) 10 pre: id 2start 11 cmd[id] 2;quorums[id] Q;phase[id] PAYLOAD 12 receive MPropose(id,2,Q,C)from 9 13 pre: id 2start 14 cmd[id] 2;quorums[id] Q;phase[id] PROPOSE 15 ts[id] proposal(id,C) 16 send MProposeAck(id,ts[id]) to 9 17 send MBump(id,ts[id]) to I8 2 18 receive MBump(id,C) 19 pre: id 2propose 20 bump(C) 21 receive MProposeAck(id,C9)from 892& 22 pre: id 2propose ^&=quorums[id][?] 23 C max{C9|92&} 24 if count(C)5then send MCommit(id,C)to Icmd[id] 25 else send MConsensus(id,C,8)to I? 26 receive MCommit(id,C9)from 92%s.t. % contains one process from each partition accessed by cmd[id] 27 pre: id 2pending 28 ts[id] max{C9|92%};phase[id] COMMIT;bump(ts[id]) 29 receive MConsensus(id,C,1)from 9 30 pre: bal[id]1 31 ts[id] C;bal[id] 1;abal[id] 1;bump(C) 32 send MConsensusAck(id,1)to 9 33 receive MConsensusAck(id,1)from & 34 pre: bal[id]=1^|&|=5+1 35 send MCommit(id,ts[id]) to Icmd[id] 36 proposal(id,<) 37 C max(<,Clock?+1) 38 Detached Detached [{h8,Di|Clock?+1DC1} 39 Attached[id] {h8,Ci} 40 Clock? C 41 return C 42 bump(C) 43 C max(C,Clock?) 44 Detached Detached [{h8,Di|Clock?+1DC} 45 Clock? C 42 4.2. THE TEMPO PROTOCOL timestamps . . . 3hA,3i 2hB,2i hC,2i 1hA,1i hB,1i hC,1i ABC processes G1G2G3G4 “depends on” //// ~~// // G1G2G3G4 “blocked on” ////// // Figure 4.3: Comparison between timestamp stability (left) and two approaches using explicit dependencies (right). of some process once it receives the promises attached to the three committed commands. Given these promises, timestamp 2is stable at the process. Even though command G4is not committed, timestamp stability ensures that its timestamp must be greater than 2. Thus, commands G1and G2,committedwith timestamp 2, can be safely executed. We now show how two approaches that use explicit dependencies behave in the above example. Dependency-based ordering EPaxos [11], BPaxos [62], Gryff [9]andAtlas (§3) order commands based on their committed dependencies. For example, in EPaxos the above command arrival order results in commands G1,G2and G3 committed with the following dependencies: dep[G1]={G2},dep[G2]={G3},dep[G3]={G1,G4}. These form the dependency graph shown on the top right of Figure 4.3. Since the dependency graph may be cyclic (as in Figure 4.3), commands cannot be simply executed in the order dictated by the graph. Instead, the protocol waits until it forms strongly connected components of the graph and then executes these components one at a time (§3.2.3). The size of such components is a priori unbounded. In fact, as we show next in §4.2.4, there are pathological scenarios where the protocol continuously commits commands but can never execute them, even under a synchronous network [11,19]. It may also significantly delay the execution of committed commands, as illustrated by our example: since command G4has not yet been committed, and the strongly connected component formed by the committed commands G1, G2and G3depends on G4, no command can be executed – unlike in Tempo. As we demonstrate in our experiments (§5), execution delays in such situations lead to high tail latencies. Dependency-based stability Caesar [10] associates each command 2not only with a set of dependencies dep[2], but also with a unique timestamp ts[2]. Commands are executed in timestamp order, and dependencies are used to determine when a timestamp is stable, and thus when the command can be executed. For this, dependencies have to be consistent with timestamps in the following sense: for any two commands 2and 20,ifts[2]<ts[20],then22dep[20]. Then the timestamp of a command can be considered stable when the transitive dependencies of the command are committed. 49 CHAPTER 4. TEMPO: PREDICTABLE LEADERLESS CONSENSUS Caesar determines the predecessors of a command while agreeing on its timestamp. To this end, the coordinator of a command sends the command to a quorum together with a timestamp proposal. The proposal is committed when enough processes vote for it. Assume that in our example Aproposes G1 and G4with timestamps 1and 4, respectively, Bproposes G2with 2,andCproposes G3with 3.WhenB receives command G1with timestamp proposal 1, it has already proposed G2with timestamp 2. If these proposals succeed and are committed, the above invariant is maintained only if G1is a dependency of G2. However, because G2has not yet been committed, its dependencies are unknown and thus Bcannot yet ensure that G1is a dependency of G2. For this reason, Bmust block its response about G1until G2 is committed. Similarly, command G2is blocked at Cwaiting for G3,andG3is blocked at Awaiting for G4. This situation, depicted in the bottom right of Figure 4.3, results in no command being committed – again, unlike in Tempo. In fact, as we show next in §4.2.4, the blocking mechanism of Caesar allows pathological scenarios where commands are never committed at all. Similarly to EPaxos, in practice this leads to high tail latencies (§5). In contrast to Caesar, Tempo computes the predecessors of a command separately from agreeing on its timestamp, via background stability detection. This obviates the need for artificial delays in agreement, allowing Tempo to offer low tail latency (§5). Limitations of timestamp stability Protocols that track explicit dependencies are able to distinguish between read and write commands. In these protocols writes depend on both reads and writes, but reads only have to depend on writes. The latter feature improves the performance in read-dominated workloads. In contrast, Tempo does not distinguish between read and write commands (§4.1), so that its performance is not affected by the ratio of reads in the workload. However, this may result in higher average latencies in certain workloads. We show in §5that this limitation does not prevent Tempo from providing similar throughput as the best-case scenario (i.e., a read-only workload) of protocols such as EPaxos and Janus. We discuss in §6how this limitation can be lifted. 4.2.4 Pathological Scenarios As previously discussed, EPaxos, Caesar and Atlas cannot ensure liveness even under a synchronous network. In this section we present a pathological scenario that demonstrates this. This scenario is a continuation of the example presented in the previous section (Figure 4.3 from §4.2.3) and applies to the three protocols. Example. Let us consider 3 processes: A,B,C. Assume that all the commands in the example are conflicting. The (infinite) example we consider is as follows: •Aproposes G1,G4,G7,... •Bproposes G2,G5,G8,... 50 4.2. THE TEMPO PROTOCOL •Cproposes G3,G6,G9,... Processes receive these commands in the following order: A:G1G4G3G7G6... B:G2G1G5G4G8G7... C:G3G2G6G5G9G8... EPaxos and Atlas.In EPaxos and Atlas, the command arrival order from above results in the following dependency sets being committed: •dep[G1]={x2} •dep[G2]={x3} •dep[G3]={G1,x4} •dep[G4]={G1,G2,x5} •dep[G5]={G2,G3,x6} •dep[G6]={G1,G3,G4,x7} • … With these committed dependencies, no strongly connected component is ever formed. This is because G1depends on G2, which depends on G3, which in turn depends on G4, and so on. As a result, commands are continuously committed but never executed. Caesar. In Caesar, timestamps proposed by processes must be unique. This can be ensured e.g. by assigning timestamps to processes round-robin. For this, assume that in the example above if a process proposes a command GC, then this command is proposed with timestamp C. First, process Aproposes G1.WhencommandG1reaches process B,Bhas already proposed G2. Since G2was proposed with a timestamp higher than the timestamp proposed for G1(i.e., 2>1), the reply about G1is blocked on G2until G2is committed. However, when G2reaches C,Chas already proposed G3, and thus command G2is blocked on G3. This keeps going forever, as depicted in the diagram below (where denotes ”blocked on”). A:G1G4 G3G7 G6... B:G2 G1G5 G4G8 G7... C:G3 G2G6 G5G9 G8... As a result, no command is ever committed. This liveness issue is due to Caesar’s wait condition [10]. 51 CHAPTER 4. TEMPO: PREDICTABLE LEADERLESS CONSENSUS 4.2.5 Multi-Partition Protocol We now explain how the single-partition protocol can be extended to handle commands that access multiple partitions. This is achieved by submitting a multi-partition command at each of the partitions it accesses. Once committed with some timestamp at each of these partitions, the command’s final timestamp is computed as the maximum of the committed timestamps. A command is executed once it is stable at all the partitions it accesses. As previously, commands are executed in the timestamp order. In more detail, when a process 8submits a multi-partition command 2on behalf of a client (line 1), it sends an MSubmit message to a set I8 2. For each partition ?accessed by 2, the set I8 2contains a responsive replica of ?close to 8(e.g., located in the same data center). The processes in I8 2then serve as coordinators of 2in the respective partitions, following the steps in Algorithm 4.1. This algorithm ends with the coordinator in each partition sending an MCommit message to I2, i.e., all processes that replicate a partition accessed by 2(lines 24 and 35; note that I2<I?because 2accesses multiple partitions). Hence, each process in I2receives as many MCommits as the number of partitions accessed by 2.Once this happens (line 26), the process computes the final timestamp of the multi-partition command as the highest of the timestamps committed at each partition (line 28), moves the command to the COMMIT phase and bumps the Clock?to the computed timestamp, generating detached promises. Similarly to the single-partition case, commands are executed using the handler at line 53. This detects command stability using Theorem 4.1, which also holds in the multi-partition case. The handler signals that a command 2is stable at a partition by sending an MStable message (line 57). Once such a message is received from all the partitions3accessed by 2, the command is executed. The exchange of MStable messages follows the approach in [49] and ensures the real-time order constraint in the Ordering property of PSMR (§4.1). Figure 1 in [49] contains an example (that also applies to Tempo) showing why this message exchange is necessary. Example Figure 4.4 shows an example of Tempo 5=1with A=5and 2partitions. Only 3 processes per partition are depicted. Partition Gis replicated at A,Band C, and partition ~at F,Gand H. Processes with the same color (e.g., Band G) are located nearby each other (e.g., in the same machine or data center). Process Aand Fare the coordinators for some command that accesses the two partitions. At partition G,A computes 6as its timestamp proposal and sends it in an MPropose message to the fast quorum {A,B,C} (the downward arrows in Figure 4.4). These processes make the same proposal, and thus the command is committed at partition Gwith timestamp 6. Similarly, at partition ~,Fcomputes 10 as its proposal and sends it to {F,G,H}, all of which propose the same. The command is thus committed at partition ~with timestamp 10. The final timestamp of the command is then computed as max{6,10}=10. 3When a command accesses a single partition, these are messages from a process to itself, and thus, are processed immediately. 52 4.2. THE TEMPO PROTOCOL A 6 6!10 F 10 B 5!6 6!10 G 9!10 C 5!6 6!10 H 9!10 Figure 4.4: Example of Tempo with 2 partitions. Next to each process we show the clock updates upon receiving MPropose messages and, in dashed boxes, the updates upon receiving MCommit or MBump messages (whichever occurs first). Assume that the stable timestamp at Ais 5and at Fis 9when they compute the final timestamp for the command. Once Freceives the attached promises by the majority {F,G,H}, timestamp 10 becomes stable at F. This is not the case at A, as the attached promises by the majority {A,B,C}only make timestamp 6stable. However, processes A,Band Calso generate detached promises up to timestamp 10 when receiving the MCommit messages for the command (line 28). When Areceives these promises, it declares timestamp 10 stable. This occurs after two extra message delays: an MCommit from Aand F to Band C,andthenMPromises from Band Cback to A. Since the command’s timestamp is stable at both Aand F, once these processes exchange MStable messages, the command can finally be executed at each. Faster stability Tempo avoids the above extra delays by generating the detached promises needed for stability earlier than in the MCommit handler. For this, after processing an MPropose message, a process also sends an MBump message containing its proposal to the nearby processes that replicate a partition accessed by the command (line 17). Upon receiving this message (line 18), a process bumps its Clock?to the timestamp in the message, generating detached promises. In Figure 4.4,MBump messages are depicted by horizontal dashed arrows. When Gcomputes its proposal 10, it sends an MBump message containing 10 to process B. Upon reception, Bbumps its Clock?to 10, generating detached promises up to that value. The same happens at Aand C.Oncethe detached promises by the majority {A,B,C}are known at A, the process again declares 10 stable. In this case, Areceives the required detached promises in two message delays earlier than when these promises are generated via MCommit. This strategy often reduces the number of message delays necessary to execute a multi-partition command. However, it is not always sufficient (e.g., imagine that Hproposed 11 instead of 10), and thus, the promises issued in the MCommit handler (line 28) are still necessary for multi-partition commands. 53 CHAPTER 4. TEMPO: PREDICTABLE LEADERLESS CONSENSUS 4.2.6 Recovery Protocol The initial coordinator of a command at some partition ?may fail or be slow to respond, in which case Tempo allows a process to take over its role and recover the command’s timestamp. We now describe the protocol Tempo follows in this case, which is inspired by that of Atlas (§3). We detail the main differences between the two recovery mechanisms at the end of this section. Tempo recovery protocol at a process 82I?is given in Algorithm 4.3. We use initial?(id)to denote a function that extracts from the command identifier id its initial coordinator at partition ?. A process takes over as the coordinator for some command with identifier id by calling function recover(id)at line 63. Only a process with id 2pending can take over as a coordinator (line 64): this ensures that the process knows the command payload and fast quorums. In order to find out if a decision on the timestamp of id has been reached in consensus, the new coordinator first performs an analog of Paxos Phase 1. It picks a ballot number it owns higher than any it participated in so far (line 65)and sends an MRec message with this ballot to all processes. As is standard in Paxos, a process accepts an MRec message only if the ballot in the message is greater than its bal[id](line 68). If bal[id]is still 0(line 69), the process checks the command’s phase to decide if it should compute its timestamp proposal for the command. If id 2payload (line 70), the process has not yet computed a timestamp proposal, and thus it does so at line 71. It also sets the command’s phase to RECOVER-R, which records that the timestamp proposal was computed in the MRec handler. Otherwise, if id 2propose (line 73), the process has already computed a timestamp proposal at line 15. In this case, the process simply sets the command’s phase to RECOVER-P,which records that the timestamp proposal was computed in the MPropose handler. Finally, the process sets bal[id]to the new ballot and replies with an MRecAck message containing the timestamp (ts), the command’s phase (phase) and the ballot at which the timestamp was previously accepted in consensus (abal). Note that abal[id]=0if the process has not yet accepted any consensus proposal. Also note that lines 70 and 73 are exhaustive: these are the only possible phases when id 2pending (line 68)andbal[id]=0(line 69), as the remaining phases RECOVER-P and RECOVER-R (recall that pending =payload [propose [recoverp [recoverr) have non-zero ballots due to line 75. In the MRecAck handler (line 77), the new coordinator computes the command’s timestamp given the information in the MRecAck messages and sends it in an MConsensus message to all processes. As in Flexible Paxos, the new coordinator waits for A5such messages. This guarantees that, if a quorum of 5+1processes accepted an MConsensus message with a timestamp (which could have thus been sent in an MCommit message), the new coordinator will find out about this timestamp. To maintain Invariant 4.1, if any process previously accepted a consensus proposal (line 79), by the standard Paxos rules [7,35], the coordinator selects the proposal accepted at the highest ballot (line 80). If no consensus proposal has been accepted before, the new coordinator first computes at line 83 the set of processes that belong both to the recovery quorum &and the fast quorum quorums[id][?]. Then, depending on whether the initial coordinator replied and in which handler the processes in have 54 4.2. THE TEMPO PROTOCOL Algorithm 4.3: Tempo recovery protocol at process 82I?. 63 recover(id) 64 pre: id 2pending 65 1 8+A(bbal[id]1 Ac+1) 66 send MRec(id,1)to I? 67 receive MRec(id,1)from 9 68 pre: id 2pending ^bal[id]<1 69 if bal[id]=0then 70 if id 2payload then 71 ts[id] proposal(id,0) 72 phase[id] RECOVER-R 73 else if id 2propose then 74 phase[id] RECOVER-P 75 bal[id] 1 76 send MRecAck(id,ts[id],phase[id],abal[id],1)to 9 77 receive MRecAck(id,C9,ph9,ab9,1)from 892& 78 pre: bal[id]=1^|&|=A5 79 if 9:2&·ab:<0then 80 let :be such that ab:is maximal 81 send MConsensus(id,C:,1)to I? 82 else 83  &\quorums[id][?] 84 B initial?(id)2_9:2·ph:=RECOVER-R 85 &0 if Bthen &else  86 C max{C9|92&0} 87 send MConsensus(id,C,1)to I? computed their timestamp proposal, there are two possible cases that we describe next. 1) The initial coordinator replies or some process in  has computed its timestamp proposal in the MRec handler ( B=tAD4 , line 84). In either of these two cases the initial coordinator could not have taken the fast path. If the initial coordinator replies (initial?(id)2), then it has not taken the fast path before receiving the MRec message from the new one, as it would have id 2commit [ execute and the MRec precondition requires id 2pending (line 68). It will also not take the fast path in the future, since when processing the MRec message it sets the command’s phase to RECOVER-P (line 74), which invalidates the MProposeAck precondition (line 22). On the other hand, even if the initial coordinator replies but some fast-quorum process in has computed its timestamp proposal in the MRec handler, the fast path will not be taken either. This is because the command’s phase at such a process is set to RECOVER-R (line 72), which invalidates the MPropose precondition (line 13). Then, since the MProposeAck precondition requires a reply from all fast-quorum processes, the initial coordinator will not take the fast path. Thus, in either case, the initial coordinator never takes the fast path. For this reason, the new coordinator can choose the command’s timestamp in any way, as long as it maintains Property 4.1.Since|&|=A5AbA1 2cb A 2c+1, the new coordinator has the output of proposal 55 CHAPTER 4. TEMPO: PREDICTABLE LEADERLESS CONSENSUS by a majority of processes, and thus it computes the command’s timestamp with max (line 86), respecting Property 4.1. 2) The initial coordinator does not reply and all processes in  have computed their timestamp proposal in the MPropose handler ( B=f0;B4 , line 84). In this case the initial coordinator could have taken the fast path with some timestamp C=max{C9|92quorums[id][?]} and, if it did, the new coordinator must choose that same timestamp C. Given that the recovery quorum &has size A5and the fast quorum quorums[id][?]has size bA 2c+5, the set of processes =&\quorums[id][?]contains at least bA 2cprocesses (distinct from the initial coordinator, as it did not reply). Furthermore, recall that the processes from have the command’s phase set to RECOVER-P (line 74), which invalidates the MPropose precondition (line 13). Hence, if the initial coordinator took the fast path, then each process in must have processed its MPropose before the MRec of the new coordinator, and reported in the latter the timestamp from the former. Then using Property 4.2, the new coordinator recovers Cby selecting the highest timestamp reported in (line 86). 4.2.7 Liveness Protocol Tempo liveness protocol in Algorithm 4.4 builds on the Paxos liveness protocol presented in §2.3.Tempo uses ⌦, the leader election failure detector [28], which ensures that from some point on, all correct processes nominate the same correct process as the leader (§2.1). Tempo runs an instance of ⌦per partition ?. In Algorithm 4.4,leader?denotes the current leader nominated for partition ?at process 8. We say that leader?stabilizes when it stops changing at all correct processes in ?.Tempo also uses I8 2,whichwecallthe partition covering failure detector (§4.2.5). At each process 8and for every command 2,I8 2returns a set of processes such that, for every partition ?accessed by 2,contains one process close to 8that replicates ?. Eventually, I8 2only returns correct processes. In the definition above, the closeness between replicas is measured in terms of latency. Returning close replicas is needed for performance but not necessary for the liveness of the protocol. Both I8 2and ⌦are easily implementable under our assumption of eventual synchrony (§4.1). For every command id with id 2pending (line 88), process 8is allowed to invoke recover(id) at line 91 only if it is the leader of partition ?according to leader?. Furthermore, it only invokes recover(id)at line 91 if it has not yet participated in consensus (i.e., bal[id]=0) or, if it did, the consensus was lead by another process (i.e., bal_leader(bal[id]) <8). In particular, process 8does not invoke recover(id)at line 91 if it is the leader of bal[id](i.e., if bal_leader(bal[id]) =8). This ensures that process 8does disrupt a recovery lead by itself. For a leader to make progress with some MRec(id,1)message, it is required that A5processes (line 78)havetheirbal[id]<1(line 68). This may not always be the case because before the variable leader?stabilizes, any process can invoke recover(id)at line 91 if it thinks it is the leader. To help the leader select a high enough ballot, and thus ensure it will make progress, we introduce a new message type, MRecNAck. A process sends an MRecNAck(id,bal[id]) at line 95 when it receives 56 4.2. THE TEMPO PROTOCOL Algorithm 4.4: Tempo liveness protocol at process 82I?. 88 periodically 89 for id 2pending 90 send MPayload(id,cmd[id],quorums[id]) to Icmd[id] 91 if leader?=8^(bal[id]=0_bal _ leader(bal[id]) <8)then 92 recover(id) 93 receive MConsensus(id,_,1)or MRec(id,1)from 9 94 pre: bal[id]>1 95 send MRecNAck(id,bal[id]) to 9 96 receive MRecNAck(id,1) 97 pre: leader?=8^bal[id]<1 98 bal[id] 1 99 recover(id) 100 receive MCommitRequest(id)from 9 101 pre: id 2commit [execute 102 send MPayload(id,cmd[id],quorums[id]) to 9 103 send MCommit(id,ts[id]) to 9 104 bal_leader(1) 105 return 1A⇤j11 Ak an MConsensus(id,_,1)or MRec(id,1)(line 93) with some ballot number 1lower than its bal[id] (line 94). When process 8receives an MRecNAck(id,1)with some ballot number 1higher than its bal[id], if it is still the leader (line 97), it joins ballot 1(line 98) and invokes recover(id)again. This results in process 8sending a new MRec with some ballot higher than 1lead by itself. As only leader?is allowed to invoke recover(id)at line 91 and line 97, and since leader?eventually stabilizes, this mechanism ensures that eventually the stable leader will start a high enough ballot in which enough processes will participate. Given a command 2with identifier id submitted by a correct process, a correct process in I2can only commit 2if it has id 2pending and receives an MCommit(id,_)from every partition accessed by 2(line 26). Next, we let ?be one such partition, and we explain how every correct process in I2 eventually has id 2pending and receives such an MCommit(id,_)sent by some correct process in I?. We consider two distinct scenarios. In the first scenario, some correct process 82I?already has id 2commit [execute. This scenario is addressed by adding two new lines to the MPromises handler: line 51 and line 52.Since id 2commit [execute at 8, by Property 4.1 there is a majority of processes in each partition @accessed by 2that have called proposal(id,_), and have thus generated a promise attached to id. Moreover, given that at most 5processes can fail, at least one process from such a majority is correct. Let one of these processes be 92I@. Due to line 47, process 9periodically sends an MPromises message that contains a promise attached to id. When a process :2I@receives such a message, if id is neither committed nor executed locally (line 51), :sends an MCommitRequest(id)to I2. In particular, it 57 CHAPTER 4. TEMPO: PREDICTABLE LEADERLESS CONSENSUS sends such message to process 82I?.Asid 2commit [execute at process 8,when8receives the MCommitRequest(id)(line 100), it replies with an MPayload(id,_,_)and an MCommit(id,_).In this way, any process :2I@will eventually have id 2pending and receive an MCommit(id,_)sent by some correct process in I?, as required. In the second scenario, no correct process in I?has id 2commit [execute but some correct process in I2has id 2pending. Due to line 90, such process sends an MPayload message to I2. Thus, every correct process in I?eventually has id 2pending. In particular, this allows leader?to invoke recover(id)and the processes in I?to react to the MRec(id,_)message by leader?.Hence, after picking a high enough ballot using the mechanism described earlier, leader?will eventually send an MCommit(id,_)to I2, as required. Finally, note that calling recover(id)for any command id such that id 2pending (line 88)can disrupt the fast path. This does not affect liveness but may degrade performance. Thus, to attain good performance, any implementation of Tempo should only start calling recover(id)after some reasonable timeout on id. In order to save bandwidth, the sending of MCommitRequest messages can also be delayed in the hope that such information will be received anyway. Atlas recovery vs Tempo recovery As described in §3.2.2, there are situations in Atlas where the payload of some command id is lost and some other command id0depends on it (i.e., id 2dep[id0]). In this case, without the information about id, the execution mechanism of Atlas will simply block, and thus the protocol has to allow id to be committed even though its payload is lost. Since Atlas may have to commit commands with unknown payloads, the protocol introduces a special noOp command. As a result, a command can be committed with the payload initially supplied by the client or noOp when such payload is lost. It is due to this that Atlas includes the command payload into MConsensus messages, thus ensuring that a unique payload will be chosen (Invariant 3.1). Tempo does not require a special noOp command, and thus, MConsensus messages do not have to include the command payload. The noOp command is avoided by Tempo because its execution mechanism can never block (as long as at most 5processes have failed). We prove this formally in §4.3. The second difference between the two recovery mechanisms is in the preconditions of recover(id) and MRec. As we just mentioned, Atlas has to allow any command to be recovered, and thus recover(id)has no precondition. Tempo, however, requires that id 2pending (lines 64 and 68), which ensures that the payload of the command being recovered is known by the process performing recovery. Finally, the two protocols track whether a fast-quorum process has seen the initial message by the coordinator (MCollect in Atlas and MPropose in Tempo) in different ways. In Atlas a fast-quorum process has seen the initial message when the fast quorum is known (line 50 in Algorithm 3.2), and in Tempo when the command phase at the process is RECOVER-P (line 84 in Algorithm 4.3). This is mostly a cosmetic difference and Atlas could have also used phases to track this information. 58 4.3. CORRECTNESS First, assume that id 2commit [execute eventually holds at process 8.ByLemma4.3, there exists a correct process :from @that has id 2dom(Attached). Due to line 47,:continuously sends an MPromises message to all the processes from @, including 9. Note that, since id 2dom(Attached) at :, this MPromises message contains a promise attached to id.Once9receives such a message, since it has id 8commit [execute (line 51), it sends an MCommitRequest(id)to I2(line 52), and in particular to process 8.Sinceid 2commit [execute at 8(line 101), process 8replies with an MPayload(id,_,_)and MCommit(id,_).Once9processes such messages, it has id 8start and received an MCommit(id,_)by process 8, a correct process from partition ?, as required. Now, assume that id 2pending holds forever at process 8. Consider the moment g0when the variable leader?stabilizes. Let process ;be leader?. Assume that id 8commit [execute at processes 8and ;forever (the case where it changes to id 2commit [execute at process 8is covered above; the case for process ;can be shown analogously). Due to line 89, process 8sends an MPayload(id,_,_) message to I2(line 90), in particular to ;and 9.Once;processes this message, it has id 2pending forever (since we have assumed that id 8commit [execute at ;forever). Once 9processes this message, it has id 8start, as required. We now prove that eventually process ;sends an MCommit(id,_)to 9. First, we show by contradiction that the number of MRec(id,_)messages sent by ;is finite. Assume the converse. After g0,duetothecheckleader?=;at line 91 and at line 97, only process ;sends MRec(id,_)messages. Since MRec(id,_)messages by processes other than ;are all sent before g0, their number is finite. For this same reason, the number of MConsensus(id,_,_)messages by processes other than ;are also finite. Thus, each correct process joins only a finite number of ballots that are not owned by ;. It follows that the number of MRec(id,_)messages sent by ;at line 92 because it joined a ballot owned by other processes (i.e., when bal_leader(bal[id]) <;) is finite. (Note that a single MRec(id,_)can be sent here due to bal[id]=0, as process ;sets bal[id]to some non-zero ballot when processing its first MRec(id,_)message). For an MRec(id,_)to be sent at line 99, process ;has to receive an MRecNAck(id,1)with bal[id]<1(line 97). If bal_leader(1)<;, the number of such MRecNAck messages is finite as the number of MRec(id,_)and MConsensus(id,_,_)by processes other than ;are finite. If bal_leader(1)=;,theMRecNAck(id,1)must be in response to an MRec(id,1)or MConsensus(id,_,1)by ;. Note that when ;sends such a message, it sets bal[id]to 1. For this reason, process ;cannot have bal[id]<1, and hence this case is impossible. Thus, there is apointintimeg1g0after which the condition at line 97 does not hold at process ;, and consequently, process ;stops sending new MRec(id,_)messages at line 99, which yields a contradiction. We have established above that id 2pending at process ;forever. We now show that process ; sends at least one MRec(id,_)message. Since id 2pending forever, process ;executes line 91 for this id infinitely many times. If ;does not send at least one MRec(id,_)at this line it is because bal[id]>0 and bal_leader(bal[id]) =;forever. If so, we have two cases to consider depending on the value of bal[id]. In the first case, bal[id]=;at ;forever. In this case, process ;took the slow path by sending to I?an MConsensus message with ballot ;(line 25). We now have two sub-cases. If any process sends an MRecNAck to process ;(line 95), this will make process ;send an MRec(id,_)(line 99), as required. 65 CHAPTER 4. TEMPO: PREDICTABLE LEADERLESS CONSENSUS Otherwise, process ;will eventually gather 5+1MConsensusAck messages and commit the command. As we have established that id 2pending at process ;forever, this sub-case is impossible. In the second case, we eventually have bal[id]>;. Since from some point on, bal_leader(bal[id]) =;at ;, this means that this process sends an MRec(id,bal[id]) (line 75), as required. We have now established that process ;sends a finite non-zero number of MRec(id,_)messages. Let 1be the highest ballot for which process ;sends an MRec(id,1)message. We now prove that ;eventually sends an MCommit(id,_)to I2, in particular to 9. Given that at most 5processes can fail, there are enough correct processes to eventually satisfy the preconditions of MRecAck and MConsensus. First we consider the case where the preconditions of MRecAck and MConsensus eventually hold at process ;. Since the precondition of MRecAck eventually holds, process ;eventually sends an MConsensus(id,_,1). Since the precondition of MConsensus eventually holds, process ;eventually sends an MCommit(id,_)to 9, as required. Consider now the opposite case where the preconditions of MRecAck or MConsensusAck never hold at process ;. Since there are enough correct processes to eventually satisfy these preconditions, the fact that they never hold at process ;implies that there is some correct process 9with bal[id]>1(otherwise 9would eventually reply to process ;). Thus, the precondition of MRecNAck holds at 9(line 94), which causes 9to send an MRecNAck(id,10)with 10=bal[id]>1to ;.When;receives such a message, it sends a new MRec(id,100)with some ballot 100 >10(line 99). It follows that 100 >10>1, which contradicts the fact that 1is the highest ballot sent by ;, and hence this case is impossible. ⇤ Lemma 4.5. Assume that id 8start at some correct process 8from some partition ?accessed by command 2. Then eventually id 2commit [execute at every correct process 92I2. Proof. For 2to be committed at process 9,9has to have id 8start and to receive an MCommit(id,_) from each of the partitions accessed by 2(line 26). We prove that process 9eventually receives such a message from each of these partitions. To this end, fix one such partition @.ByLemma4.4,itis enough to prove that some correct process from partition @eventually has id 8start. By contradiction, assume that all the correct processes from partition @have id 2start forever. We have two scenarios to consider. First, consider the scenario where ?=@. But this contradicts the fact that id 8start at process 8. Now, consider the scenario where ?<@. In this scenario we consider two sub-cases. In the first case, eventually id 2commit [execute at process 8.ByLemma4.3, there is a correct process from each partition accessed by 2, in particular from partition @, that has id 8start, which contradicts our assumption. In the second case, id 2pending at process 8forever (the case where it changes to id 2commit [execute is covered above). Due to line 90, process 8periodically sends an MPayload message to the processes in partition @. Once this message is processed, the correct processes in partition @will have id 2payload. Since partition @contains at least one correct process, this also contradicts our assumption. ⇤ 66 4.3. CORRECTNESS Definition 4.1. We define the set of proposals issued by some process 82I?as LocalPromises8= img(Attached)[Detached. Lemma 4.6. For each process 82I?we have that h8,Ci2LocalPromises8)(8D2{1,...,C}· h8,Di2LocalPromises8). Proof. Follows trivially from Algorithm 4.1 and Algorithm 4.2.⇤ Lemma 4.7. Consider a command 2with an identifier id and the final timestamp C, and assume that id 8 start at some correct process in I2. Then at every correct process in I2, eventually variable Promises contains all the promises up to Cby some set of processes ⇠with |⇠|b A 2c+1. Proof. By Lemma 4.5, eventually id 2commit [execute at all the correct processes in I2. Consider a point in time g0when this happens and fix a process 82I2from a partition ?that has id 2commit.Let ⇠be the set of correct processes from partition ?,MPromises(⇡0 9,0 9)be the MPromises message sent by each process 92⇠in the next invocation of line 47 after g0,andids =–{dom(0 9)|92⇠}. Due to line 28, by Lemma 4.6 we have that (*) 892⇠,D2{1,...,C}·h9,Di2img(0 9)[⇡0 9. Note that, for each id02ids, we have id08start at some correct process (in particular, at the processes 9that sent such id0in their MPromises(⇡0 9,0 9)messages). Thus, by Lemma 4.5, there exists g1g0 at which ids ✓commit [execute at process 8.LetMPromises(⇡1 9,1 9)be the MPromises message sent by each process 92⇠in the next invocation of line 47 after g1. Since at process 8we have that ids ✓commit [execute, once all these MPromises messages are processed by process 8,foreach 92⇠we have that –{1 9[id0]|id02ids}✓Promises and ⇡1 9✓Promises at process 8.Given the definition of ids and since 0 9✓1 9and ⇡0 9✓⇡1 9, we also have that img(0 9)✓Promises and ⇡0 9✓Promises at process 8. From (*), it follows that 892⇠,D2{1,...,C}·h9,Di2Promises at process 8.⇤ Lemma 4.8. Consider a command 2with an identifier id, and assume that id 8start at some correct process in I2. Then every correct process 82I2eventually executes 2. Proof. Consider a command 2with an identifier id, and assume that id 8start at some correct process in I2. By contradiction, assume further that some correct process 82I2never executes 2.Let 20=2,id0=id,andC0be the timestamp assigned to 2. Then by Lemma 4.7, eventually in every invocation of the periodic handler at line 53, we have ⌘C0.ByLemma4.5 and since 8never executes 20, eventually 8has id02commit.Hence,either8never executes another command 21preceding 20 in ids,or8never receives an MStable(id0)message from some correct process 9eventually indicated 67 CHAPTER 4. TEMPO: PREDICTABLE LEADERLESS CONSENSUS by I8 20. The latter case can only be due to the loop in the handler at line 53 being stuck at an earlier command 21preceding 20in ids at 9. Hence, in both cases there exists a command 21with an identifier id1and a final timestamp C1such that hC1,id1i<hC0,id0iand some correct process 812I21never executes 21despite eventually having id12commit. Continuing the above reasoning, we obtain an infinite sequence of commands 20,21,22,...with decreasing timestamp-identifier pairs such that each of these commands is never executed by some correct process. But such a sequence cannot exist because the set of timestamp-identifier pairs is well-founded. This contradiction shows the required. ⇤ Proof of Liveness. Assume that some command 2with identifier id is submitted by a correct process or executed at some process. We now prove that it is eventually executed at all correct processes in I2. By Lemma 4.7 and Lemma 4.8, it is enough to prove that eventually some correct process in I2has id 8start. First, assume that id is submitted by a correct process 8. Due to the precondition at line 2, 82I2.Thenphase[id]is set to PROPOSE at process 8when 8sends the initial MPropose message, and thus id 8start as required. Assume now that id is executed at some (potentially faulty) process. By Lemma 4.3, there exists some correct process in I2with id 8start, as required. ⇤ 4.4 Summary and Related Work Timestamping (aka sequencing) is widely used in distributed systems. In particular, many storage systems orchestrate data access using a fault-tolerant timestamping service [63–67], usually implemented by a leader-based SMR protocol [7,8]. As reported in prior works, the leader is a potential bottleneck and is unfair with respect to client locations [11,62,68,69]. To sidestep these problems, leaderless protocols order commands in a fully decentralized manner. Early protocols in this category, such as Mencius [20], rotated the role of leader among processes. However, this made the system run at the speed of the slowest replica. More recent ones, such as EPaxos [11] and its follow-ups (Atlas (§3), BPaxos [62] and Gryff [9]), order commands by agreeing on a graph of dependencies (§4.2.3). Tempo builds on one of these followups, Atlas (§3), which leverages the observation that correlated failures in geo-distributed systems are rare [4] to reduce the quorum size in leaderless SMR. As we demonstrate next in our evaluation (§5), dependency-based leaderless protocols exhibit high tail latency and suffer from bottlenecks due to their expensive execution mechanism. Timestamping has been used in two previous leaderless SMR protocols. Caesar [10], which we discussed in §4.2.3, suffers from similar problems to EPaxos. Clock-RSM [21] timestamps each newly submitted command with the coordinator’s clock, and then records the association at 5+1processes using consensus. Stability occurs when all the processes indicate that their clocks have passed the command’s timestamp. As a consequence, the protocol cannot transparently mask failures, like Tempo; these have to be handled via reconfiguration. Its performance is also capped by the speed of the slowest replica, similarly to Mencius [20]. 68 4.4. SUMMARY AND RELATED WORK Partial replication is a common way of scaling services that do not fit on a single machine. Some partially replicated systems use a central node to manage access to data, made fault-tolerant via standard SMR techniques [70]. Spanner [4] replaces the central node by a distributed protocol that layers twophase commit on top of Paxos. Granola [23] follows a similar schema using Viewstamped Replication [71]. Other approaches rely on atomic multicast, a primitive ensuring the consistent delivery of messages across arbitrary groups of processes [39,57]. Atomic multicast can be seen as a special case of PSMR as defined in §4.1. Janus [15] generalizes EPaxos to the setting of partial replication. Its authors shows that for a large class of applications that require only one-shot transactions, Janus improves upon prior techniques, including MDCC [72], Tapir [73] and 2PC over Paxos [4]. Our experiments in §5demonstrate that Tempo significantly outperforms Janus due to its use of timestamps instead of explicit dependencies. Unlike Janus, Tempo is also genuine, which translates into better performance. Summary This chapter presented Tempo, the first leaderless SMR protocol to determine the order of command execution solely based on scalar timestamps. Tempo cleanly separates timestamp assignment from detecting timestamp stability, and such mechanisms easily extend to partial replication. As we show next in our evaluation (§5), Tempo’s approach enables the protocol to offer low tail latency and high throughput even under contended workloads, thus addressing the performance issue inherent to the execution mechanism of EPaxos and Atlas. 69 5 Performance Evaluation In this chapter we experimentally evaluate Atlas and Tempo.In§5.1 we assess whether assuming small values of 5is acceptable for geo-distribution. In §5.2 we cover the protocols considered in this evaluation and the framework we developed to implement and evaluate them. In §5.3 we detail our sets of testbeds and benchmarks. Finally, we study the performance of Atlas and Tempo in full replication (§5.4)and partial replication deployments (§5.5). 5.1 Bounds on Failures In a practical deployment of Atlas and Tempo, a critical parameter is the number of concurrent site failures 5the protocol can tolerate. It has been reported that concurrent site failures are rare in geodistributed systems [4]. However, the value of 5should also account for asynchrony periods during which sites cannot communicate due to link failures: if more than 5sites are unreachable in this way, Atlas and Tempo may block for the duration of the outage. We have thus conducted an experiment on Google Cloud Platform (GCP) to check that assuming small values of 5is still appropriate when this is taken into account. Our experiment ran for 3 months (October 2018 – January 2019) among 17 sites, the maximal number of sites available in GCP at the time. During the experiment, sites ping each other every second (in the spirit of [17] but on a much larger scale). A link failure occurs between two sites when one of them does not receive a reply after a (tunable) amount of time. Figure 5.1 reports the number of simultaneous link failures for various timeout thresholds. Note that no actual machine crash occurred during the campaign of measurements. When the timeout threshold is set to 10s, only two events occur, each with a single link failure. Fixing the threshold to either 3s or 5s leads to two events of noticeable length. During the first event, occurring on November 7, the links between the Canadian site and five others are slow for a couple of hours. During the second event, on December 8, the links between Taiwan and seven other sites are slow for around two minutes. From the data collected, we compute the value of 5as the smallest number of sites :such that, 70 5.2. PROTOCOLS AND IMPLEMENTATION 0 1 2 3 4 5 6 7 Oct 07 Oct 19 Oct 30 Nov 11 Nov 23 Dec 04 Dec 16 Dec 27 Jan 08 T/O threshold # link failures 3s 5s 10s Figure 5.1: The number of simultaneous link failures among 17 sites in GCP when varying the timeout threshold. at any point in the experiment, crashing :sites would cover all the slow links. During our experiment, timeouts were reported on the links incident to at most a single site (e.g., the Canadian site on November 7). Thus, we may conclude that 51held during the whole experiment, even with the smallest timeout threshold. In other words, Atlas and Tempo with 51would have been always responsive during this 3-month experiment. In light of these results, we evaluate deployments of Atlas and Tempo in which 5 is set to 1,2or 3. 5.2 Protocols and Implementation We compare Atlas and Tempo with Flexible Paxos (FPaxos) [35], EPaxos [11], Caesar [10]and Janus [15]. All these protocols have been briefly covered in §2. FPaxos is a variant of Paxos that, like Tempo, allows selecting the allowed number of failures 5separately from the replication factor A:it uses quorums of size 5+1during normal operation and quorums of size A5during recovery. EPaxos, Atlas and Caesar are leaderless protocols that track explicit dependencies (§4.2.3). EPaxos and Caesar use fast quorums of size b3A 4cand d3A 4e, respectively. Atlas and Tempo use fast quorums of size bA 2c+5.Atlas also improves the condition EPaxos uses for taking the fast path: e.g., when A=5and 5=1,Atlas always processes commands via the fast path, unlike EPaxos. To avoid clutter, we exclude the results for EPaxos from some of our plots with A=5since its performance is similar to (but never better than) Atlas 5=1. Janus is a leaderless protocol that generalizes EPaxos to the setting of partial replication. This protocol is representative of the state-of-the-art for partial replication, and the authors of Janus have already compared it extensively to prior approaches (including MDCC [72], Tapir [73]and 2PC over Paxos [4]). Janus is based on an unoptimized version of EPaxos whose fast quorums contain all replicas in a given partition. Our implementation of Janus is instead based on Atlas,whichyields quorums of the same size as Tempo and a more permissive fast-path condition. We call this improved version Janus*. 71 CHAPTER 5. PERFORMANCE EVALUATION To improve the fairness of our comparison, all protocols are implemented in the same framework which consists of 35K lines of Rust and contains common functionality necessary to implement and evaluate the protocols. This includes a networking layer, an in-memory key-value store, dstat monitoring, and a set of benchmarks (e.g., YCSB [74]). The source code of the framework is available at github.com/vitorenesduarte/fantoch. The framework provides three execution modes: cloud, cluster and simulator. In the cloud mode, the protocols run in wide area on Amazon EC2. In the cluster mode, the protocols run in a local-area network, with delays injected between the machines to emulate wide-area latencies. Finally, the simulator runs on a single machine and computes the observed client latency in a given wide-area configuration. Our simulator disregards CPU and network bottlenecks: messages between processes are not sent through sockets, and the time it takes to process each message is not accounted for. Thus, the output of the simulator represents the best-case latency for a given scenario. Together with dstat measurements, the simulator allows us to determine if the latencies obtained in the cloud or cluster modes represent the best-case scenario for a given protocol or are the effect of some bottleneck. 5.2.1 Implementation details The framework is built using Tokio1, an asynchronous runtime for Rust with the building blocks needed for writing network applications. Using Tokio, Machines are connected via 16 TCP sockets, each with a 16MB buffer. Sockets are flushed every 5ms or when the buffer is filled, whichever is earlier2. Key-Value Store We have implemented a simple in-memory key-value store (KVS) that supports two operations: read(:)to fetch the content of the KVS record under key :,andwrite(:,E)to update its value to E.SMRcommands can then access one or more keys at a time3. The KVS is used in all our benchmarks (§5.3). Atlas implementation Atlas, EPaxos and Janus rely on sets of dependencies which can grow unboundedly large as the system keeps running. EPaxos [11] solves this issue by compressing a set of dependencies reported by a fastquorum processes as a vector clock with Aentries, where entry 9contains the conflicting command with the highest identifier issued by process 9. Janus [15] suggests instead that fast-quorum processes report only the direct dependencies of each command (i.e., the most recent conflicting commands seen by each fast-quorum process). We follow this second approach in the implementation of Atlas, EPaxos and Janus. For each key in the KVS, we maintain two command identifiers: the identifier of the latest read-only command (that 1https://tokio.rs 2fantoch/src/run/task/server/mod.rs 3fantoch/src/kvs.rs 72 5.3. EXPERIMENTAL SETUP accessed such key), and the identifier of the latest write command (where latest means the command most recently seen by the process). In order to allow multiple worker threads to compute and update conflicts (when handling MCollect and MRec messages), this mapping is maintained in a concurrent hash-map, and each value (i.e. the identifier of a command) is protected by a read-write lock4.Aswe detail next, there are cases in which such values are only read, which explains the use of a read-write lock. When computing the dependencies of a read-only command with identifier id, for each key accessed by the command, we first retrieve the identifier of the latest write, and then set the latest read to be id. When computing the dependencies of a write command with identifier id, for each key accessed by the command, we first retrieve the latest read and latest write, and then set the latest write to be id. With NFR (§3.2.5), the latest read is ignored and only the latest write is retrieved. Overall we have that all commands depend on writes, reads never depend on reads, and writes only depend on reads if NFR is disabled. Tempo implementation We assume that each partition in Tempo contains a single key. When partitions/keys are colocated in the same machine, the message passing in Tempo is optimized and replaced by shared-memory operations (§4.2.8). For that, the implementation of Tempo also makes use of a concurrent hash-map that maps each key to a clock. Instead of protecting this clock with a read-write lock, we use atomics5.When computing the timestamp proposal of a command, the clock associated with each key accessed by the command is bumped by one and its new value is fetched. The timestamp proposal is then the highest new clock value fetched. After this, a new round of bumps occurs, incrementing these clocks up to the timestamp proposal computed. With NFR (§4.2.9), the timestamp proposal of a single-key read command is simply the current clock value associated with such key. 5.3 Experimental Setup Testbeds One of the testbeds uses Amazon EC2 with c5.2xlarge instances (machines with 8 virtual CPUs and 16GB of RAM). Another testbed is a local cluster where we inject wide-area delays similar to those observed in EC2. This cluster contains machines with 6 physical cores and 32GB of RAM connected by a 10GBit network. The experiments in Figure 5.5 and Figure 5.6 are performed in EC2, and most of the remaining experiments are conducted on the local cluster. We have validated some of the experiments performed on the local cluster using Google Cloud Platform with an Erlang implementation of Atlas [24]. 4fantoch_ps/src/protocol/common/graph/deps/keys/locked.rs 5fantoch_ps/src/protocol/common/table/clocks/keys/atomic.rs 73 CHAPTER 5. PERFORMANCE EVALUATION Table 5.1: Ping latency (milliseconds) between Amazon EC2 sites. N. California Singapore Canada S. Paulo Hong Kong N. Virginia Tokyo Stockholm Mumbai Oregon Ireland 141 186 72 183 220 70 204 35 118 127 N. California – 181 78 190 154 59 111 162 231 21 Singapore – 221 338 34 235 77 195 63 162 Canada – 123 202 15 154 114 196 65 S. Paulo – 315 113 267 217 299 180 Hong Kong – 213 51 230 98 144 N. Virginia – 152 105 181 72 Tokyo – 265 136 100 Stockholm – 131 157 Mumbai – 223 Oregon – Experiments span up to 11 EC2 regions, which we call sites : Ireland (eu-west-1), Northern California (us-west-1), Singapore (ap-southeast-1), Canada (ca-central-1), São Paulo (sa-east1). Hong Kong (ap-east-1), Northern Virginia (us-east-1), Tokyo (ap-northeast-1), Stockholm (eu-north-1), Mumbai (ap-south-1)andOregon(us-west-2). Table 5.1 shows the average ping latencies between these sites. When A=3, the first 3sites (eu-west-1,us-west-1 and apsoutheast-1) are used. When A=5, the next 2 sites (ca-central-1 and sa-east-1) are added, and so on, until we reach A=11. Benchmarks We first evaluate full replication deployments (§5.4) initially using a microbenchmark where each command carries a key of 8 bytes and (unless specified otherwise) a payload of 100 bytes. We assume that commands conflict when they carry the same key. In the case of Tempo, each partition contains a single key, and thus, commands conflict when they access the same partition. To measure performance under a conflict rate dof commands, a client chooses key 0with a probability d, and some unique key otherwise. We then evaluate Atlas and Tempo with a geo-replicated key-value store under the YCSB workload [74]. Finally, we evaluate Tempo in partial replication deployments (§5.5) using YCSB+T [75], a transactional version of YCSB. Clients are closed-loop and always deployed in separate machines located in the same regions as servers. 5.4 Full Replication Deployment Fast-path likelihood We first evaluate the benefits of Atlas and Tempo flexible fast-path conditions. To this end, Figure 5.2, Figure 5.3 and Figure 5.4 compare their fast-path ratio with that of EPaxos for different conflict rates and values of 5. We also plot worst-case scenario, i.e., when all conflicting commands take the slow path. Note that we do not consider 5=1as both Atlas and Tempo always take the fast path in this case (i.e., 74 5.4. FULL REPLICATION DEPLOYMENT               FP 7P  7P      )3  )3  (3 Figure 5.8: Average latency with 256 clients per site under a low conflict rate (2%) while increasing the number of sites from 3 to 11. FPaxos performance is closely tied to which sites are considered and which of these sites is chosen as the leader. To better understand this, consider what happens when we move from A=7to A=9by focusing on Table 5.1. With A=7, the closest sites to Ireland, the site where the FPaxos leader is located, are Canada and Northern Virginia (both roughly 70ms away). With A=9, two new sites are added, one of them being Stockholm. Note, however, that this site is only 35ms away from Ireland. For this reason, FPaxos 5=1phase 2, which requires contacting only a single site, is two times faster with A=9than with A=7. As a result, the protocol can offer a average latency lower than the other protocols in this setting. With 5=2, the FPaxos leader has to contact, not one, but two additional processes in FPaxos 5=2phase 2. Although one of these processes is nearby (Stockholm), The distance to the second closest site to Ireland did not change when we moved from A=7to A=9, which is explains why FPaxos 5=2offers a similar latency in both settings. This was not the case when we moved from A=5to A=7, where the improvement in the performance of FPaxos 5=2is quite clear. In this case, the added site responsible for the latency reduction in FPaxos 5=2is Northern Virginia. As reported before in Figure 5.5, FPaxos remains significantly more unfair than the remaining protocols. For example, with A=11, FPaxos 5=1provides 75ms at the leader site, but this latency more than triples at Singapore (241ms), São Paulo (235ms), Hong Kong (275ms) and Tokyo (260ms). On the other hand, leaderless protocols satisfy clients much more uniformly. For example, for the above sites, the latencies offered by Tempo 5=1are respectively 169ms, 193ms, 203ms, 166ms and 149ms. Availability under failures Figure 5.9 depicts an experiment demonstrating that leaderless protocols are inherently more available than a leader-driven protocol. The experiment compares Atlas and FPaxos across 3 Google Cloud Platform (GCP) sites: Taiwan, Finland and South Carolina9. Such configuration tolerates a single site 9Our implementation of the protocols in github.com/vitorenesduarte/fantoch does not include the recovery mechanism of each protocol. For this reason, we report the experiment that was done in the Atlas publication [24] using our Erlang implementation of Atlas. 81 CHAPTER 5. PERFORMANCE EVALUATION       KKSS    )3        )    P    KKSS 6K    P      Figure 5.9: The impact of a failure on the throughput of Atlas and FPaxos (3 sites, 5=1). failure, so FPaxos is the same as Paxos. We do not evaluate the remaining protocols (Tempo, EPaxos and Caesar) as their availability guarantees are similar to those of Atlas in this configuration. Each site hosts 128 clients. Half of the clients issue commands targeting key 0and the other half issue commands targeting a unique key per client. Hence, commands by clients in the first half conflict with each other, while commands by clients in the second half commute with all commands by a different client. After 30s of execution, the SMR service is abruptly halted at the Taiwan site, where the Paxos leader is located. Based on the measurements reported in §5.1, we set the timeout after which a failure is suspected to 10s for both protocols. Upon detecting the failure, the clients located at the failed site (Taiwan) reconnect to the closest alive site, South Carolina. In the case of Paxos, the surviving sites initiate recovery and elect South Carolina as the new leader. In the case of Atlas, the surviving sites recover the commands that were initially coordinated by Taiwan. As shown in Figure 5.9, Paxos blocks during the recovery time. In contrast, Atlas keeps executing commands, albeit at a reduced throughput. The drop in throughput happens largely because the clients issuing commands on key 0(50% of all clients) collect as dependencies some of the commands being recovered (those that also access key 0). The execution of the former commands then blocks until the latter are recovered. In contrast, the clients at non-failed sites issuing commands with per-client keys continue to operate as normal. Since commands by these clients commute with those by other clients, their execution never blocks on the commands being recovered. This means that these clients operate without disruption during the whole experiment. The bottom right plot contains the aggregate throughput of the system. Before failure, Atlas is almost two times faster than Paxos, and operates consistently better during the whole experiment. Note, 82 5.4. FULL REPLICATION DEPLOYMENT             P P  P      ( Figure 5.10: YCSB performance for update-heavy (80% writes), balanced (50% writes), read-heavy (20% writes) and read-only (0% writes) workloads, with A=7sites with NFR optimization is enabled. The latency provided by protocols when the NFR optimization is not enabled is obtained by also considering the light gray bar on top of each colored bar. however, that Paxos has a slightly higher throughput at the leader (Taiwan) before the crash, and at the new leader (South Carolina) after recovery. This is due to the delay between committing and executing commands in Atlas. Key-value store service We now compare Tempo,Atlas and EPaxos when the protocols are applied to a replicated key-value store (KVS) service. When accessing a KVS record stored under key :,aclientexecuteseithercommand read(:)to fetch its content, or write(:,E)to update it to value E. To benchmark the performance of the replicated KVS we use the Yahoo! Cloud Serving Benchmark (YCSB) [74]. We apply four types of workloads, each with a different mix of read/write operations: update-heavy (w=80%), balanced (w=50%), read-heavy (w=20%), and read-only (w=0%), where wrepresents the percentage of write commands. The KVS contains 106records and all workloads select records following a Zipfian distribution with the default YCSB skew (zipf =0.99). In this experiment, Tempo with 52{1,2},Atlas with 52{1,2}and EPaxos are deployed over 7 sites. At each site running the benchmark we execute 256 YCSB client threads. Colored bars represent the latency provided protocols when the NFR optimization (§3.2.5 and §4.2.9) is enabled. As pointed out in §3.2.5 and §4.2.9, this optimization accelerates the execution of read commands and reduces their impact in the protocol stack. With the light gray bar on top of each colored bar we obtain the latency provided by protocols when the NFR optimization is not enabled. In the update-heavy workload without NFR, EPaxos offers 676ms whereas Atlas offers 377ms when 5=1and 643ms when 5=2. Although EPaxos and Atlas 5=2have the same fast quorum size with A=7, due to its more flexible fast-path condition, Atlas is able to provide a slightly better performance than EPaxos: Atlas 5=2takes the fast path for 58% of commands, while EPaxos does so in 43% of cases. Tempo 5=2improves on this and is able to take the fast path for 77% of commands, offering 83 CHAPTER 5. PERFORMANCE EVALUATION                   KJK. K K K 7       Figure 5.11: Maximum throughput with 3 sites per shard under low (zipf =0.5) and moderate contention (zipf =0.7). Three workloads are considered for Janus*: 0% writes (its best-case scenario), 5% writes and 50% writes. 259ms. This is 2.5xfaster than Atlas 5=2and 2.6xfaster than EPaxos. Tempo 5=1offers 192ms, which is 2xfaster than Atlas 5=1and 3.5xfaster than EPaxos. Increasing the percentage of read operations improves the performance of Atlas and EPaxos the protocols because reads do not conflict with other reads. In the read-only workload the performance is simply determined by the quorum size, since the protocols always take the fast path. In this case, both EPaxos and Atlas 5=2offer 172ms, while Atlas 5=1, which has a smaller fast quorum, offers 133ms. As mentioned in §4.2.3,Tempo does not distinguish between reads and writes. For this reason, the performance of the protocol is the same in all four workloads. With the NFR optimization, Tempo 5=1,Tempo 5=2,Atlas 5=1,Atlas 5=2and EPaxos reduce their latency by up to 30%, 48%, 20%, 30% and 27%, respectively. The highest speedup occurs for Tempo as, without NFR, the protocol does not optimize the execution of reads (§4.2.3). In the read-only workload, all the protocols execute commands after a single round trip to the closest majority. In this case, NFR allows Tempo 5=1,Tempo 5=2,Atlas 5=2and EPaxos to match the performance of vanilla Atlas 5=1, while maintaining their higher fault-tolerance level. Overall, Tempo 5=1,Tempo 5=2,Atlas 5=1and Atlas 5=2with NFR are up 3.5x,2.9x,2.2xand 1.5xfaster than vanilla EPaxos, respectively. 5.5 Partial Replication Deployment We now consider a partial replication setting and compare Tempo with Janus* using the YCSB+T benchmark [75]. We define a shard as set of several partitions co-located in the same machine. Each partition contains a single YCSB key. Each shard holds 106keys and is replicated at 3sites (Ireland, N. California and Singapore) emulated in our cluster. Clients submit commands that access two keys picked at random following the YCSB access pattern (a zipfian distribution). In Figure 5.11 we show the maximum throughput for both Tempo and Janus* under low (zipf =0.5) and moderate contention (zipf =0.7). 84 5.5. PARTIAL REPLICATION DEPLOYMENT For Janus*, we consider 3 YCSB workloads that vary the percentage of write commands (denoted by w): read-only (w=0%, YCSB workload C), read-heavy (w=5%, YCSB workload B), and update-heavy (w=50%, YCSB workload A). The read-only workload is a rare workload in SMR deployments. It represents the best-case scenario for Janus*, which we use as a baseline. Since Tempo does not distinguish between reads and writes (§4.2.3), we have a single workload for this protocol. Janus* performance is greatly affected by the ratio of writes and by contention. More writes and higher contention translate into larger dependency sets, which bottleneck execution faster. This is aggravated by the fact that Janus* is non-genuine, and thus requires cross-shard messages to order commands. With zipf =0.5, increasing wfrom 0% to 5% reduces throughput by 25-26%. Increasing wfrom 0% to 50% reduces throughput by 49-56%. When contention increases (zipf =0.7), the above reductions on throughput are larger, reaching 36-60% and 87%-94%, respectively. Tempo provides nearly the same throughput as the best-case scenario for Janus* (w=0%). Moreover, its performance is virtually unaffected by the increased contention. This comes from the parallel and genuine execution brought by the use of timestamp stability (§4.2.5). Overall, Tempo provides 385K ops/s with 2 shards, 606K ops/s with 4 shards, and 784K ops/s with 6 shards (averaged over the two zipf values). Compared to Janus* w=5% and Janus* w=50%, this represents respectively a speedup of 1.2-2.5xand 2-16x. The tail latency issues demonstrated in Figure 5.6 also carry over to partial replication. For example, with 6 shards, zipf =0.7and w=5%, the 99.99th percentile for Janus* reaches 1.3s, while Tempo provides 421ms. We also ran the same set of workloads for the full replication case and the speed up of Tempo with respect to EPaxos and Atlas is similar. 85 6 Conclusions and Future Work In this thesis we have presented Atlas and Tempo, two SMR protocols tailored to planet-scale systems. The two protocols follow a leaderless approach, ordering commands in a fully decentralized manner and thus offering similar quality of service to all clients. These are the first leaderless SMR protocols parameterized with the number of allowed failures 5. Both protocols leverage the fact that concurrent site failures are rare in planet-scale systems, and are thus optimized for small values of 5.Atlas and Tempo employ small fast quorums of size bA 2c+5and offer a flexible fast-path condition that allows a high percentage of operations to be processed within a single round trip. When 5=1, the protocols always take the fast path and the fast quorum is a plain majority. Moreover, the protocols employ an optimization called non-fault-tolerant reads that allows for low-latency linearizable reads with a low impact in the protocol stack. Following EPaxos, Atlas orders commands based on explicit dependencies. In theory, EPaxos-like protocols do not ensure liveness, and in practice they offer high tail latencies. In contrast to previous leaderless protocols, Tempo determines the order of command execution solely based on scalar timestamps, and cleanly separates timestamp assignment from detecting timestamp stability. Moreover, this mechanism easily extends to partial replication. As shown in our evaluation, Tempo’s approach enables the protocol to offer low tail latency and high throughput even under contended workloads. In this thesis we focused on simplifying and improving the performance of leaderless protocols. Below we list some limitations of our work and possible research avenues that may allow leaderless protocols to fulfil their potential and finally be adopted by industry practitioners [76]. Linearizable multi-key reads As mentioned in §4.2.3,Tempo does not currently distinguish between reads and writes (with the exception of the NFR optimization §4.2.9, which only applies to single-key reads). Multi-key reads can be natively supported in Tempo if, instead of maintaining a single clock per key/partition, we maintain two: one for reads, and another for writes [77]. Such strategy would be similar to how we have implemented the dependency tracking mechanism of Atlas that also distinguishes between reads and writes (§5.2.1). 86 Local sequentially consistent reads (LSCR) Many real-world workloads are read-dominated. Common SMR protocols that build a total order (like Paxos [7], Raft [8]andZab[47]) can be adjusted to allow serving reads locally, without incurring a wide-area round trip. The resulting reads are not linearizable, but sequentially consistent [78,79], i.e., they may read stale data. This is acceptable for many applications [80], and thus, local reads are crucial to the scalability of practical systems such as ZooKeeper [2] and etcd [81]. Allowing LSCR in leaderless protocols like EPaxos [11], Atlas or Tempo is challenging because these protocols may deliver commuting write commands in different orders at different replicas. As a consequence, reading locally from a replica may violate sequential consistency. We believe that dependency-based protocols like EPaxos and Atlas cannot be easily modified to support LSCR without great performance costs. These protocols would have to build total order, which would essentially require setting the conflict rate to 100%. This would result in no fast paths and high tail latencies due to the issues in their execution mechanism (§4.2.3). On the other hand, timestamp-based protocols like Tempo can be more easily modified to support LSCR: if a replica wants to serve a read at timestamp C, it is enough to delay the execution until Cbecomes stable at all keys/partitions accessed by the read. For keys/partitions to become stable in a timely fashion in Tempo, replicas should periodically issue detached promises. However, this may result in a fast-path ratio decrease. Tiny fast quorums Tempo can be modified in order to use fast-quorums of size 25. In this case, Property 4.1 and Theorem 4.1 would have to be modified to require 5+1and A5processes, respectively: this would still ensure that the sets of processes used to compute a timestamp and to decide when a timestamp is stable always intersect. Further investigations are required to understand if smaller quorums required for committing a command can compensate for larger quorums required for executing them. 87 Bibliography [1] João M. Lourenço. The NOVAthesis L A T E X Template User’s Manual . NOVA University Lisbon. 2021. url: https://github.com/joaomlourenco/novathesis/raw/master/template. pdf (cit. on p. ii). [2] Patrick Hunt et al. “ZooKeeper: Wait-free Coordination for Internet-scale Systems”. In: USENIX Annual Technical Conference (USENIX ATC) . 2010 (cit. on pp. 1,87). [3] Michael Burrows. “The Chubby Lock Service for Loosely-Coupled Distributed Systems”. In: Symposium on Operating Systems Design and Implementation (OSDI) . 2006 (cit. on pp. 1,3,11,15, 16). [4] James C. Corbett et al. “Spanner: Google’s Globally-Distributed Database”. In: Symposium on Operating Systems Design and Implementation (OSDI) . 2012 (cit. on pp. 1–3,6,11,15,16,28, 68–71). [5] Fred B. Schneider. “Implementing Fault-Tolerant Services Using the State Machine Approach: A Tutorial”. In: ACM Comput. Surv. (1990) (cit. on pp. 1,6). [6] Maurice Herlihy and Jeannette M. Wing. “Linearizability: A Correctness Condition for Concurrent Objects”. In: ACM Trans. Program. Lang. Syst. (1990) (cit. on pp. 1,6,38). [7] Leslie Lamport. “The Part-Time Parliament”. In: ACM Trans. Comput. Syst. (1998) (cit. on pp. 1, 2,6,9,16,25,35,54,68,87). [8] Diego Ongaro and John K. Ousterhout. “In Search of an Understandable Consensus Algorithm”. In: USENIX Annual Technical Conference (USENIX ATC) . 2014 (cit. on pp. 1,6,16,35,68,87). [9] Matthew Burke, Audrey Cheng, and Wyatt Lloyd. “Gryff: Unifying Consensus and Shared Registers”. In: Symposium on Networked Systems Design and Implementation (NSDI) . 2020 (cit. on pp. 1,3, 37,48,49,68). [10] Balaji Arun et al. “Speeding up Consensus by Chasing Fast Decisions”. In: International Conference on Dependable Systems and Networks (DSN) . 2017 (cit. on pp. 1,3,11,12,16,35,37,48,49, 51,68,71,79). 88 BIBLIOGRAPHY [11] Iulian Moraru, David G. Andersen, and Michael Kaminsky. “There Is More Consensus in Egalitarian Parliaments”. In: Symposium on Operating Systems Principles (SOSP) . 2013 (cit. on pp. 1–3,11, 16,17,22,23,28,35,36,48,49,68,71,72,79,87). [12] Alexandru Turcu et al. “Be General and Don’t Give Up Consistency in Geo-Replicated Transactional Systems”. In: International Conference on Principles of Distributed Systems (OPODIS) . 2014 (cit. on pp. 1,35). [13] Fernando Pedone and André Schiper. “Generic Broadcast”. In: International Symposium on Distributed Computing (DISC) . 1999 (cit. on pp. 1,11,17,35). [14] Leslie Lamport. Generalized Consensus and Paxos . Tech. rep. MSR-TR-2005-33. Microsoft Research, 2005 (cit. on pp. 1–3,11,15,17,22,35). [15] Shuai Mu et al. “Consolidating Concurrency Control and Consensus for Commits under Conflicts”. In: Symposium on Operating Systems Design and Implementation (OSDI) . 2016 (cit. on pp. 2,28, 38,39,59,69,71,72). [16] Alexander Shraer et al. “Dynamic Reconfiguration of Primary/Backup Clusters”. In: USENIX Annual Technical Conference (USENIX ATC) .2012(cit.onp.2). [17] Shengyun Liu et al. “XFT: Practical Fault Tolerance beyond Crashes”. In: Symposium on Operating Systems Design and Implementation (OSDI) . 2016 (cit. on pp. 2,16,28,70). [18] Seth Gilbert and Nancy A. Lynch. “Brewer’s Conjecture and the Feasibility of Consistent Available Partition-Tolerant Web Services”. In: SIGACT News (2002) (cit. on pp. 3,6,28). [19] Tuanir França Rezende and Pierre Sutra. “Leaderless State-Machine Replication: Specification, Properties, Limits”. In: International Symposium on Distributed Computing (DISC) . 2020 (cit. on pp. 3,11,36,37,49). [20] Yanhua Mao, Flavio Paiva Junqueira, and Keith Marzullo. “Mencius: Building Efficient Replicated State Machine for WANs”. In: Symposium on Operating Systems Design and Implementation (OSDI) . 2008 (cit. on pp. 3,35,68). [21] Jiaqing Du et al. “Clock-RSM: Low-Latency Inter-datacenter State Machine Replication Using Loosely Synchronized Physical Clocks”. In: International Conference on Dependable Systems and Networks (DSN) . 2014 (cit. on pp. 3,37,68). [22] Leslie Lamport. “Time, Clocks, and the Ordering of Events in a Distributed System”. In: Commun. ACM (1978) (cit. on pp. 3,6,37). [23] James A. Cowling and Barbara Liskov. “Granola: Low-Overhead Distributed Transaction Coordination”. In: USENIX Annual Technical Conference (USENIX ATC) . 2012 (cit. on pp. 3,37,69). [24] Vitor Enes et al. “State-Machine Replication for Planet-Scale Systems”. In: European Conference on Computer Systems (EuroSys) . 2020 (cit. on pp. 4,73,75,79,81). 89 BIBLIOGRAPHY [25] Vitor Enes et al. “Efficient Replication via Timestamp Stability”. In: European Conference on Computer Systems (EuroSys) . 2021 (cit. on p. 4). [26] Michael J. Fischer, Nancy A. Lynch, and Mike Paterson. “Impossibility of Distributed Consensus with One Faulty Process”. In: J. ACM (1985) (cit. on p. 5). [27] Cynthia Dwork, Nancy A. Lynch, and Larry J. Stockmeyer. “Consensus in the Presence of Partial Synchrony”. In: J. ACM (1988) (cit. on p. 6). [28] Tushar Deepak Chandra, Vassos Hadzilacos, and Sam Toueg. “The Weakest Failure Detector for Solving Consensus”. In: J. ACM (1996) (cit. on pp. 6,10,56). [29] Daniel Abadi. “Consistency Tradeoffs in Modern Distributed Database System Design: CAP is Only Part of the Story”. In: Computer (2012) (cit. on p. 6). [30] Martin Kleppmann. “A Critique of the CAP Theorem”. In: arXiv CoRR abs/1509.05393 (2015). url: http://arxiv.org/abs/1509.05393 (cit. on p. 6). [31] Haonan Lu et al. “Existential consistency: measuring and understanding consistency at Facebook”. In: Symposium on Operating Systems Principles (SOSP) . 2015 (cit. on p. 6). [32] Rebecca Taft et al. “CockroachDB: The Resilient Geo-Distributed SQL Database”. In: International Conference on Management of Data (SIGMOD) . 2020 (cit. on p. 6). [33] Leslie Lamport. “Paxos Made Simple”. In: SIGACT News (2001) (cit. on pp. 6,9). [34] Robbert van Renesse and Deniz Altinbuken. “Paxos Made Moderately Complex”. In: ACM Comput. Surv. (2015) (cit. on p. 6). [35] Heidi Howard, Dahlia Malkhi, and Alexander Spiegelman. “Flexible Paxos: Quorum Intersection Revisited”. In: International Conference on Principles of Distributed Systems (OPODIS) . 2016 (cit. on pp. 6,9,16,22,25,28,35,40,45,54,71). [36] Heidi Howard. Distributed consensus revised . Tech. rep. UCAM-CL-TR-935. PhD Thesis. 2019 (cit. on p. 6). [37] Heidi Howard and Richard Mortier. “Paxos vs Raft: have we reached consensus on distributed consensus?” In: Workshop on Principles and Practice of Consistency for Distributed Data (PaPoC@EuroSys) . 2020 (cit. on p. 6). [38] Michael Whittaker et al. “SoK: A Generalized Multi-Leader State Machine Replication Tutorial”. In: JSys (2021) (cit. on p. 6). [39] Rachid Guerraoui and André Schiper. “Genuine Atomic Multicast in Asynchronous Distributed Systems”. In: Theor. Comput. Sci. (2001) (cit. on pp. 12,59,69). [40] Leslie Lamport. “Fast Paxos”. In: Distributed Computing (2006) (cit. on p. 22). [41] Leslie Lamport. “Lower Bounds for Asynchronous Consensus”. In: Distributed Computing (2006) (cit. on p. 28). 90