Dynamic end-to-end reliable causal delivery middleware for geo-replicated services

Dynamic End-to-End Reliable Causal Delivery Middleware for Geo-Replicated Services 2023 Georges Younes UMinho | Departamento de Informática Georges Younes Dynamic End-to-End Reliable Causal Delivery Middleware for Geo-Replicated Services Programa de Doutoramento em Informática das Universidades do Minho, de Aveiro e do Porto June 2023 Universidade do Minho Escola de Engenharia Departamento de Informática Georges Younes Dynamic End-to-End Reliable Causal Delivery Middleware for Geo-Replicated Services 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 Carlos Baquero ede Ali Shoker June 2023 DIREITOS DE AUTOR E CONDIÇÕES DE UTILIZAÇÃO DO TRABALHO POR TERCEIROS Este é um trabalho académico que pode ser utilizado por terceiros desde que respeitadas as regras e boas práticas internacionalmente aceites, no que concerne aos direitos de autor e direitos conexos. Assim, o presente trabalho pode ser utilizado nos termos previstos na licença abaixo indicada. Caso o utilizador necessite de permissão para poder fazer um uso do trabalho em condições não previstas no licenciamento indicado, deverá contactar o autor, através do RepositoriUM da Universidade do Minho. Licença concedida aos utilizadores deste trabalho Creative Commons Atribuição 4.0 Internacional CC BY 4.0 ?iiTb,ff+`2iBp2+QKKQMbXQ`;fHB+2Mb2bf#vf9Xyf/22/XTi This document was created with the (pdf/Xe/Lua)L A T EX processor and the NOVAthesis template (v6.9.0) [1]. ii Acknowledgements I would like to express my gratitude to my advisors Carlos Baquero and Ali Shoker, for supporting my work with great advise, patience and encouragement. Also a special thanks to Professor Paulo Sérgio Almeida for his constant support and guidance and to Vitor Enes for being a great colleague and friend, both heavily involved in this work. To all of them, thank you for the continuous support and encouragement throughout this work and for the counseling provided. I learned a lot. Without their guidance and dedication this thesis would not have been possible. Thanks to my all of my colleagues and friends at the HASLab laboratory, for the great work environment created. Finally, a special thanks to my friends and my family, for all the love and support given, and for always having confidence in me. iii 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. , (Place) (Date) (Georges Younes) iv Braga 2 of June 2023 “You cannot teach a man anything; you can only help him discover it in himself.” (Galileo) v Resumo Causalidade Dinâmica Extremo a Extremo em Replicação Geográfica O crescimento da dependência de serviços baseados na Internet, durante as últimas duas décadas, causou um aumento na adoção de sistemas geo-replicados. O desenho deste sistemas é enquadrado à luz do teorema CAP . Neste contexto os modelos de coerência relaxada – Eventual Consistency –permitem reduzir o tempo de resposta para com os utilizadores finais e, assim, aumentar a disponibilidade dos sistemas e obter interações mais rápidas. O advento de novas técnicas de convergência como Conflictfree Replicated Data Types , amplamente adotados na indústria de geo-replicação como seja no Facebook, PayPal, Microsoft, SoundCloud, entre outros, permitiu também um maior enquadramento formal destas técnicas. Em particular, o modelo de coerência causal, provou ser o modelo mais forte para sistemas sempre disponíveis. Assim torna-se relevante revisitar as técnicas de comunicação causal em grupo, e associado middleware de transmissão, pois sendo que muitos destes sistemas foram originalmente construídas à perto de três décadas, precisam de ser adequados ao contexto actual de utilização. Esta tese principia pela análise de novas abstrações para a garantia de propriedades end-to-end no registo e entrega causal. Prossegue com a observação de anomalias e ineficiências resultantes de implementações multi-threaded de entrega causal, e com a identificação de uma primeira abordagem para garantir causalidade entres os dois extremos do sistema. Após a identificação de problemas na escalabilidade nas implementações que se baseiam em vectores versão ou relógios lógicos, é proposta uma nova solução baseada na manipulação de grafos de dependências e numa eficiente manutenção e simplificação dos mesmos, recorrendo à observação de propriedades de estabilidade. É também proposta uma nova API aos utilizadores do middleware de comunicação. A avaliação das soluções propostas foi feita com base num sistema programado em Erlang e foi feita a sua avaliação de desempenho e aplicação a quatro casos de estudo. Palavras-chave: Coerência causal, Entrega causal, Replicação geográfica vi Abstract Dynamic End-to-End Reliable Causal Delivery Middleware for Geo-Replicated Services The reliance on Internet-based services during the past two decades caused a leap in geo-replicated systems, as a means to target clients across the globe, in the light of the CAP theorem . Therefore, relaxed consistency models got a lot of attention to reduce the response time to end users, and thus boost the availability of the systems at the cost of delayed – Eventual Consistency –. Together with the advent of new convergence techniques like Conflict-free Replicated Data Types —widely adopted in the georeplication industry like Facebook, PayPal, Microsoft, SoundCloud, etc., this lead to the reliance on more useful tradeoff consistency models like the causal consistency model, proven to be strongest model for available systems. Intuitively, this suggested another visit to revise the causality techniques, broadcast middleware, and abstractions, originally built three decades earlier for a different set of digital services, i.e., applications, capabilities, and usage. The research in this thesis analyzes the end to end workflow of causality-based services, leading to the identification of new problems and shortcoming in state of the art causality techniques and abstractions, and proposing novel corresponding ones. First, this work discovers that, given that many applications are today multi-threaded, handling causality while overlooking this fact will lead into semantic pitfalls in some classes of applications. A corresponding technique is proposed in this thesis to apply end-to-end time-stamping at the application level instead of the causal middleware. Second, this thesis points out a scalability problem in state of the art causal broadcast middlewares that rely on vector clocks for timestamping. This thesis proposes the first graph-based abstraction for timestamping which is proven to be one order of magnitude more scalable and efficient than its state of the art counterpart. Third, this work identifies existing redundancy in the time-stamping methods used in both causal middleware and application logic, and thus proposes a slightly modified, but effective, API that reduces the bandwidth metadata overhead by half. The API includes the notion of causal stability that makes garbage collection fast and easy. Fourth, this thesis introduces the first technique for dynamic causality middleware, crucial in elastic services, leading to guaranteed causal delivery under dynamic membership. These contributions are then implemented in a comprehensive well-engineered codebase in Erlang . To demonstrate its usefulness and feasibility, this work has been applied to four practical use-cases and projects during the course of this thesis. vii 24 Evolution of the causal DAG (with causal stability) ................... 73 25 Evolution of a new node joining process ........................ 83 26 Evolution of an existing node leaving process ...................... 85 27 Full mesh topology with a slow link between 0and 1.................. 95 28 Experiment 1: transmission and memory, without stability. ............... 101 29 Experiment 1: Memory and Non-causal Stability latency, with stability. ......... 102 30 Experiment 2: transmission and memory metadata, without stability. .......... 103 31 Experiment 3: transmission and memory, without stability. ............... 104 32 Experiment 2 and 3: Median memory metadata, with stability. ............. 104 33 Experiment 2 and 3: non-causal stability latency, with stability. ............. 105 34 Experiment 2: non-causal delivery latency. ....................... 106 35 Experiment 3: non-causal delivery latency. ....................... 106 36 Experiment 2: non-causal delivery latency on slow link. ................. 107 37 Experiment 3: non-causal delivery latency on slow link. ................. 108 38 ASPAS architecture. .................................. 109 39 TQHQ; based Add-Wins Set with Clone functionality ................... 111 40 Lasp runtime system. ................................. 112 41 Lasp pure op-based CRDTs. .............................. 112 42 Concrete Resettable Counter Implementation ..................... 113 43 The general architecture of our multi-master proposed solution. ............ 114 44 High-level architecture of Minidote. .......................... 116 xiv List of Algorithms 1 CBCAST algorithm using Vector clocks on process 82I............... 24 2 Vector-based tagged causal delivery algorithm without causal stability at client process for node 82I..................................... 53 3 Vector-based tagged causal delivery algorithm without causal stability at middleware process for node 82I................................. 54 4 Graph-based tagged causal delivery algorithm without causal stability at client process for node 82I..................................... 58 5 Graph-based tagged causal delivery algorithm without causal stability at middleware process for node 82I................................. 59 6 Vector-based tagged causal delivery algorithm with causal stability at client process for node 82I..................................... 66 7 Vector-based tagged causal delivery algorithm with causal stability at middleware process for node 82I................................... 67 8 Graph-based tagged causal delivery algorithm with causal stability at client process for node 82I..................................... 69 9 Graph-based tagged causal delivery algorithm with causal stability at middleware process for node 82I................................... 70 10 Graph-based tagged causal delivery algorithm with causal stability and phantom messages at client process for node 82I........................... 75 11 Graph-based tagged causal delivery algorithm with causal stability and phantom messages at middleware process for node 82I........................ 76 12 Dynamic tcb client process for node 82I...................... 80 13 Dynamic tcb middleware process for node 82I.................. 81 14 Causal check algorithm ............................... 89 xv xvi R Introduction From social network platforms to entertainment services, storage and backup, collaborative tools and many more, online services are continuously increasing in scale and in demand. More and more users access these services around the world expecting a reliable service and a good, fast user experience. For system designers, this means ensuring a correct, uninterrupted operation, with fast response time. This technically translates to high availability which implies tolerance to both network and replica failures, as well as low latency of user requests. A common approach to address the above challenges is to use geo-replication, where full or partial replicas are located in different regions to provide fast access to nearby users. On one hand, this allows the possibility of switching to other replicas when a certain replica crashes. On the other hand, distributed services provide a minimal degree of performance such as being highly-responsive to client requests. Reaching this quality of service can become a challenge, especially over large geographical areas and under large numbers of requests. Geo-replication allows distributing the data and placing it in the proximity of clients, often using relaxed consistency models. This decreases the access delays, resulting from the transmission of requests and responses between servers and clients. In addition to this, replication reduces the load of client requests by allowing different replicas to share that load and divide the work among them. Data consistency is the main challenge that has to be faced when dealing with geo-replication. Ensuring that all replicas have a consistent state is expected by users, but it comes at the price of delaying the system and serving the users. Users also expect high-responsiveness and thus cannot always accept the high latency of ensuring data consistency. To solve this, distributed systems offer a wide range of consistency models and tradeoffs between consistency and availability to choose from. Strong consistency models prioritize consistency over availability, while weak consistency models offer less guarantees for consistency and offer higher availability. Causal consistency [2] is the strongest achievable consistency that does not sacrifice availability under network partitions [3]. It is an attractive model for distributed systems and data stores as it guarantees data consistency that reflects causality between events. 1 CHAPTER 1. INTRODUCTION This thesis identifies three main problems for ensuring correct, efficient, and dynamic causal delivery. We summarize these problems in the following section with hints on the proposed solutions. We then present the main contributions in the following Section 1.2. Both, problems and contributions, are then presented in detail in the different chapters of this thesis according to the organization described in Section 1.3. 1.1 Problem statement and objectives Tracking causality and guaranteeing that causal dependencies are satisfied has taken a large interest in distributed systems and still does. It has also led to many contributions by distributed systems researchers and engineers such as causal delivery middleware, causality tracking data structures and mechanisms, garbage collection for causality metadata and so on. End-to-end potential causality. Traditional causal delivery middleware [4] provide a delivery order in each process that is consistent with causality. This means that messages are guaranteed to be delivered at every process in an order respecting (consistent with) the causality between the messages. As for concurrent messages that are not causally related, the order of delivery is arbitrary and dictated only by their arrival at different processes. While such traditional causal delivery middleware guarantee causality, they do not provide the application with knowledge about concurrency between events. Given two messages <1and <2delivered in that sequence to some process, no information is provided to the application whether <1causally preceded <2or if they were originated independently (concurrently) of each other. In this thesis, we show that providing such knowledge to the application is necessary for a class of applications that require knowledge of concurrency in order to apply arbitration rules [5]. Ad hoc arbitration of concurrent messages does not influence causality but that might lead to anomalies and break constraints set by the application. Causal stability. Causal stability concerns knowledge about the end-to-end happens-before regarding future deliveries at each client process, namely whether concurrent messages (to a given message) can still be delivered. The concept itself is not new, and it has been used many times, but hidden in applications, without being properly recognized as such. This has lead to successive ad-hoc re-implementations, sometimes overly complex. As an example, causal stability is hidden inside the difficult to understand implementation of Replicated Growable Arrays (RGAs) [6], while not being recognized as a building block. We propose this concept is important enough to be recognized, advertised, and provided by the middleware as a building block, to avoid “reinventing the wheel”. For this we follow the causal stability concept, coined as such in a paper that defines pure operation-based CRDTs [7]. Dynamicity and node churn. Traditionally when nodes in group membership leave or join the group, the system blocks until the remaining nodes are aware of the new change in the membership and ready 2 1.2. MAIN CONTRIBUTIONS AND RESULTS to resume the distributed computation. This may not be problematic in some scenarios were node churn is rare. However, in highly dynamic scenarios where nodes join and leave regularly and often, this could have a high cost on the responsiveness and availability of the service. 1.2 Main contributions and results Over the course of this thesis, we have made several contributions to the state-of-the-art of distributed data stores to address the aforementioned problems and challenges. Our main four contributions can be summarized by the following. 1.2.1 Tagged Causal Delivery (Broadcast) In particular, we define a novel definition of end-to-end happens-before that focuses on application (client) level visibility for tracking causal dependencies. This definition was motivated by the lack of an end-to-end happens-before definition that has lead to incorrect behaviour and anomalies in a class of applications as we explain in Chapter 3. We also show the pitfalls of trying to achieve end-to-end happens-before that distributed systems practitioners could fall in. And finally, we develop a Tagged Causal Delivery middleware that characterizes end-to-end happens-before, showing the correct architecture, API and algorithm for both a VV-based version and our graph-based TCB version. 1.2.2 Graph-based Tagged Causal Delivery middleware A lot of implementation of traditional causal middleware use version vectors, delivery queues and other data structures that are not best suited for the partial order of causality. There have been works that improved the efficiency vector clocks to reduce the metadata sent on the network, and other works that try to optimise the causal delivery mechanisms. However, we argue that a causal delivery middleware should be viewed as a standalone service where all its components work together to provide an efficient and scalable service in a holistic fashion. We then present the correct API, architecture and algorithm of such middleware. We implement also, in addition to our graph-based TCB, an optimized end-to-end mechanism, E2E VV-based in order to have a fair evaluation that compares the performance of both. In the evaluation, we show how our TCB performs better than the optimized VV-based algorithm in terms of transmission of meta data, size of meta data in memory and non-causal delivery and stability latency. 1.2.3 Causal Stability Our work throughout the thesis focuses around the importance on causal stability in the Tagged Causal Delivery middleware. We show how using causal stability allows designing and implementing correct causal middleware for applications like CRDTs, where knowledge about concurrency between events is available to the application and the causality order characterizes the end-to-end happens before. Moreover, 3 CHAPTER 1. INTRODUCTION we show how causal stability allows the safe garbage collection of causality meta data. We also implement the causal stability mechanism for the VV-based version, which is completely different than that of the graph-based version, in order to compare the performance of both algorithms when the causal stability mechanism is running. Again, we notice that our TCB performs and scales better than the VV-based algorithm for causal stability. Some other smaller contributions were made throughout the thesis, such as phantom messages, a feature to keep causal stability working even when the majority of peers are passive. We explain this in detail in Section 5.4. We also design and implement our own causality checker tool as shown in Chapter 7. This causal checker serves as a verification tool for causal delivery and stability. 1.2.4 Dynamic Tagged Causal Delivery middleware In this thesis, we introduce, to our knowledge, the first dynamic causal middleware were new nodes can join and existing nodes can leave safely, without blocking the system. We also extend the TCD (or TCB) middleware, through causal stability, to provide a version of the algorithm suited for unreliable and dynamic environments. Our causal checker also verifies the correctness of causal delivery and stability when different nodes join and leave concurrently. Publications. Part of the work described in this thesis was published or is under submission in several peer-reviewed conferences and journals: • ASPAS: As Secure as Possible Available Systems. Published at IFIP DAIS’21. Best Paper Award at IFIP DisCoTec 2021. • Integration Challenges of Pure Operation-based CRDTs in Redis. Published at EuroSys PAPOC workshop, 2016. • Compact Resettable Counter through Causal Stability. Published at EuroSys PAPOC workshop, 2017. • Pure Operation-Based Replicated Data Types. Journal under re-submission. • The Pitfalls in Achieving Tagged Causal Stability. Presented at EuroSys PAPOC workshop, 2018. • Tagged Causal Delivery: Efficient End-to-End Causal Middleware for Replicated Systems at Scale. Under submission. • Graph-based Causal Middleware for Dynamic and Unreliable Networks. To be submitted. 4 1.3. OUTLINE 1.3 Outline The rest of this thesis is organized as follows: •Chapter 2gives a brief presentation on the required background and an overview of the state of the art literature relevant to the rest of the thesis; •Chapter 3presents Tagged Causal Delivery that characterizes end-to-end happens-before, the problems and pitfalls of achieving it incorrectly and the correct architecture and implementation of TCB; •Chapter 4presents our graph-based TCB middleware, its architecture, API and algorithm, with a detailed explanation of the dependency dots and causal DAG; •Chapter 5presents and motivates Causal Stability and how to implement it in both vector-based and graph-based algorithms; •Chapter 6presents the Dynamic version of our TCB algorithm; •Chapter 7presents our own causal checker that verifies a correct delivery and stability respecting the causality of events; •Chapter 8presents the empirical evaluation comparing our TCB to a vector-based causal delivery middleware; •Chapter 9presents use cases where our contributions were implemented in existing software and data stores; •Chapter 10 is the conclusion of the thesis with some remarks about future work directions. 5 k State of The Art The work in this thesis is related to three main areas, namely, geo-replication, consistency models, and causality tracking and reordering. Consequently, we present in this chapter the state of the art in the three areas in order to provide an adequate background to understad the work described in the following chapters. I start with geo-replication, its importance nowadays, considering the high number of users and the large amount of data used in distributed services, and the challenges that geo-replication introduces for system designers and engineers. I discuss the properties of such systems and their trade-offs when they arise. I then narrow the focus to consistency in replicated storage systems, visiting notable theorems like CAP, PACELEC, CAC and TV. I then visit the different consistency models that range on a spectrum from Strong consistency to weak consistency. I focus mostly on causal consistency being the strongest consistency model for higher availability, scalability and fault-tolerance. Nest, I explore the different solutions used for causality tracking and ordering and show the different data versioning and timestamping techniques that have been used to guarantee causal consistency. Finally, I touch upon different mechanisms, solutions and architectures used in building weakly consistency and causally consistent key-value stores. I conlcude this chapter with a state of the art analysis and discussion, showing some of the motivations of my work. 2.1 Geo-replication Although geo-replication (in Figure 1) solves many distributed systems problems, it generates new ones that have to be dealt with. Data consistency, or more precisely maintaining data/state consistency among replicas, is the burden brought by geo-replication. The challenge here becomes designing systems that guarantee high availability and low latency without sacrificing data consistency which was proven to be a hopeless endeavor due to the need of some form of synchronization [8–10]. What differentiates a system design from another are the tradeoffs and properties of such designs. Although consistency and availability are usually the most desired ones, other properties such as latency, throughput, and partition-tolerance, could not be overlooked due to their importance, that often comes from the semantics of the higher level applications. Some of these properties are achievable together while others are inherently antagonistic. 6 2.2. (GEO-)REPLICATION PROPERTIES AND TRADE-OFFS Figure 1: A Geo-replicated system. The system properties are defined to satisfy its design goals, and whenever conflicts between these design goals are encountered, an array of trade-offs arises. 2.2 (Geo-)replication properties and trade-offs A geo-replicated system is a variant of distributed system whose properties can broadly follow the classical Safety properties and Liveness properties [11]. Safety properties address the correctness of an algorithm which informally translates to ”nothing bad will ever happen” while liveness properties translates to ”something good will eventually happen”. The categorization of distributed system properties as safety or liveness properties is helpful as the trade-offs between its properties are in some sense safety-liveness trade-offs. This is intuitively correct as the stricter (safer) the property the more complex the underlying protocols need to be and this negatively influences the performance (liveness) of the system. In the following, I start with presenting the FLP impossibility result [12], which specifies that either the liveness or the safety of consensus of replicas must be sacrificed if those replicas are prone to failure in an asynchronous setting. I then narrow down the context from general setting State-Machine Replication (SMR) towards trade-offs in storage systems where operations are restricted to read and write operations. In this vein, I start with the CAP theorem [9,10] and then the PACELC, considered as its extension [8]. I then visit the CAC and TV (throughput vs visibility trade-offs) theorems which focus on causal consistency, as a more specific form of consistency. 7 CHAPTER 2. STATE OF THE ART P1• ,1(G)0//P1• ,1(G)0// P2• '2(G)0• ,2(G)1//P2• ,2(G)1// P3• '3(G)1• '3(G)0//P3• '3(G)1• '3(G)0// P4• '4(G)0• '4(G)1//P4• '4(G)0• '4(G)1// Figure 5: Example of Causal consistency concurrent writes. A causally-consistent store does not require concurrent writes to be globally ordered and there for causal order has not been violated. 2.3.1.4 Eventual Consistency Eventual Consistency [22] is a data-centric consistency model. It is what most databases offer as the weakest consistency level. It states that if writes to the database stop, eventually (i.e., after some unspecified time) the replicated state will be consistent, returning the same value to every process. Implicitly, there is the assumption that data aims to be stored, ruling out vacuous implementations which returned the same value forever. This consistency model can be thought simply as convergence guarantee, without any time guarantee. 2.3.2 Client-centric consistency models In this section, I shed light on three well-known client-centric consistency models that provide application session guarantees: Monotonic Reads, Monotonic Writes, and Read Your Writes. 2.3.2.1 Monotonic Reads Monotonic reads is a client-centric consistency model also called session guarantees. It ensures that if a process reads the value of a data item G, any successive read operation on Gby that process will always return that same value or a more recent one. Monotonic reads only applies to reads invoked by the same process and does not apply to operations performed by different processes. Monotonic reads can be totally available: even during a network partition, all nodes can make progress. We see in Figure 6an example of a process %with two different local copies !1and !2.Theread operations ('(G1),'(G2)) are carried out by the same process %.The,((G8)operations are a set of write operations on Gand not necessarily by %only that return G8. Process %first performs a read operation on Gat !1, returning the value of G1(at that time). This value results from the write operations in ,((G1)performed at !1. Later, %performs a read operation on Gat !2, shown as '(G2). To guarantee 14 2.3. CONSISTENCY MODELS L1• ,((G1)//• '(G1)//L1• ,((G1)//• '(G1)// L2• ,((G1;G2)• '(G2)//L2• ,((G2)• '(G2)// Figure 6: Example of Monotonic Reads L1• ,(G1)//L1• ,(G1)// L2• ,((G1)• ,(G2)//L2• ,(G2)// Figure 7: Example of Monotonic Writes monotonic-read consistency, all operations in ,((G1)should have been propagated to !2before the second read operation takes place. In the left diagram, we see that ,((G1; G2)took place meaning all operation that happened at !1before '(G1)have been propagated at !2before '(G2)and therefore the reads are monotonic. In the right diagram„ no guarantees are given that the set ,((G2)also contains all operations contained in,((G1). Thus, monotonic-reads consistency is not guaranteed. 2.3.2.2 Monotonic Writes Monotonic writes is a client-centric consistency model also called session guarantees. It ensures that if a process performs two write operations F1,thenF2on an item G, then all processes will observe F1 before F2. Monotonic writes only applies to writes invoked by the same process and does not apply to operations performed by different processes. Monotonic writes can be totally available: even during a network partition, all nodes can make progress. We see herein Figure 7an example of a process %with two different local copies !1and !2.The write operations performed by a single process %at two different local copies of the same data store. Process %performs a write operation on Gat local copy !1, presented as the operation,(G1). Later, % performs another write operation on G, but this time at !2, shown as,(G2). To ensure monotonic-write consistency, the previous write operation at !1must have been propagated to !2. In the left diagram, operation,(G1)at !2takes place before,(G2)and therefore writes are monotonic. In the right diagram, monotonic-write consistency is not guaranteed as the propagation of ,(G1)to copy !2is missing. 2.3.2.3 Read Your Writes Read your writes, also known as read my writes, is a client-centric consistency model also called session guarantees. It requires that if the effect of a write operation by a process on data item Gwill always be seen by a successive read operation on Gby the same process. Note that read your writes does not apply to operations performed by different processes. There is no guarantee, for instance, that if a process % 15 CHAPTER 2. STATE OF THE ART L1• ,(G1)//L1• ,(G1)// L2• ,((G1;G2)• '(G2)//L2• ,((G2)• '(G2)// Figure 8: Example of Read Your Writes writes a value successfully, that a process &will subsequently observe that write. Read your writes is sticky available: if a network partition occurs, every node can make progress, so long as clients never change which server they talk to. In Figure 8, Process %performed a write operation ,(G1)and later a read operation at a different local copy. Read-your-writes consistency guarantees that the effects of the write operation can be seen by the succeeding read operation. In the left diagram, this is expressed by ,((G1; G2), which states that ,(G1)is part of ,((G2). However, in the right diagram, ,(G1)has been left out of ,((G2), meaning that the effects of the previous write operation by process %have not been propagated to !2. 2.4 Causality tracking and broadcast Being at the core of this thesis, causality tracking techniques and broadcast are reviewed in this section. I address broadcast/multicast algorithms that are used to guarantee casual delivery. I also list key compaction techniques that were used to reduce the metadata needed for guaranteeing causality. Prior to that, I start by introducing key abstractions for causality tracking techniques. In particular, I start with physical clocks, explaining why we cannot rely on them in geo-replicated systems. Then, I introduce the notion of logical clocks and causality tracking that was introduced by Lamport [23], establishing the foundation for the subsequent mechanisms and theory [23–28]. I conclude this section with a small discussion on single-object and multi-object logical clocks. 2.4.1 Causality tracking primitives 2.4.1.1 Physical Clocks The ability of timestamping events is crucial in a system to makes sense of the ordering of events. In most applications, some constraints on the order of events must be respected, and thus being able to timestamp those events is important in order to maintain these constraints. A simple way to address this is to rely on physical clocks as they are already present in computing machines and can be used to provide timestamps systems events. When we have a single machine, creating a timeline of events is easy to achieve, even with multiple processes or threads. The reason for this is the fact that all these events will be timestamped by the same physical clock providing a single source and a global view of time. 16 2.4. CAUSALITY TRACKING AND BROADCAST P1•01• ## 02// P2•11• 12// Figure 9: Diagram showing a distributed execution using Lamport’s Happens-before However, in the case of a distributed system, we cannot rely on physical clocks as a solution. This is because different machines have different physical clocks that are prone to diverge, and consequently there will be no guarantees that the timestamps provided by these clocks will give a correct ordering and consistent data versions at all replicas. In other terms, there is no global clock to provide a global view of time anymore. To mitigate the skew between these clocks it is required that these clocks periodically use some synchronization mechanism such as NTP, Coordinated Universal Time UTC, etc. However, this synchronization does not guarantee perfect synchronization as the quality depends on factors such as network load, or the NTP server could be offline, etc. Another issue is that even perfect physical timestamps would provide a total order of events and thus would not capture concurrency between events which could be needed for some applications. Some systems like Cassandra [29] workaround this by tagging each version of an object with a physical timestamp and use a last-writer-wins (LWW) policy as a tie-breaker when clocks are equal. 2.4.1.2 Happens-Before relation Lamport introduced the happens-before relation to capture time between events logically, without the reliance on physical clocks. The happens-before relation characterizes the notion of causality or more accurately potential causality. It translates to the ability or possibility of one event to have influenced another event in its future. Formally, the happens-before relation can be described as follows. Given two events 4and 40, we say that 4happens-before 40denoted by 4!40if one of the following conditions holds true: •4and 40are two events on the same process ?8where 4occurs before 40; •4is the send event of a message <from process ?8and 40the receive event of <by process ?9; •There exists and event 400 such that 4!400 and 400 !40. Not all events are causally related by the happens-before relation. Such events are called concurrent events and are denoted by 4k40which is equivalent to 4940and 4094. In Figure 9, we see two processes %1and %2. We notice that 01!02and 11!12since 01 occurred before 02on %1and 11before 12on %2. This is captured by the first condition. Also 02!12 as 02is the send event of a message from %1and 12its relative receive event on %2captured by the second condition. Finally, 01!12captured by the transitivity in condition 3. Events 01and 11are concurrent or 01k11. 17 CHAPTER 2. STATE OF THE ART P1• 01((⇠1=1)• )(<=2 ## 02((⇠1=2)// P2• 11((⇠2=1)• 12((⇠2=3)// Figure 10: Diagram showing a distributed execution using Lamport clocks 2.4.1.3 Lamport Clocks Lamport clocks [27] are scalar logical clocks that increase monotonically and used to enforce causal ordering. Each process ?8maintains a scalar clock (⇠8which is basically a non-negative monotonically increasing integer counter that starts with (⇠8:=0. The algorithm used to maintain this clocks is presented as follows. Considering an event 4at ?8and (⇠8set to 0: •If event 4is an internal event (excluding send and receive events), increment the value of (⇠8by 1, (⇠8:=(⇠8+1; •If event 4is a send event, increment the value of (⇠8by 1, (⇠8:=(⇠8+1and send the new )(<:=(⇠8along with the message <; •If event 4is a receive event of message <with )(<, set (⇠8:=<0G((⇠8,)(<)+1. As scalar clocks are integers, the order on timestamps is the total order relation (<) among integers. For each process ?8,ifevent4occurs before event 40then we have )(4<)(0 4because the value of (⇠8 is strictly increasing. If 4was the send event of message <and 40the receive event of the corresponding message, then also we have )(4<)(0 4because based on the third rule )(0 4is greater than )(4at least by 1. Finally, knowing that <is transitive we get that if 4!40then )(4<)(0 4.However,the converse is not true, i.e., )(4<)(0 4does not necessarily mean that 4!40. The problem here is that the happens-before relation is a partial order while the scalar timestamps are totally ordered. This does not allow the distinction of concurrent events. In Figure 10, we see two processes %1and %2. We notice that at %1,01!02and therefore )(01=1<2=)(02. Similar observations at %2where11!12and therefore)(11=1<3=)(12. Also 02!12and therefore )(02=2<3=<0G (1,2)+1=)(12. However, even though )(11=1<2=)(02it does not mean that 11!02because in fact 11k02. 2.4.1.4 Causal Histories Causal Histories [28] are a simple intuitive way to capture causality. Every event has a unique identifier, and the causal history of an event the set of identifiers of events in its causal past. Every process in the system assign a unique identifier for every event, e.g. process/node name and a local increasing counter. Locally, a process %assigns for a new event the identifier ?8+1where 8was the last value of the local 18 2.4. CAUSALITY TRACKING AND BROADCAST P1• {a1}• ## {01,a2}// P2• {b1}• {01,02,11,b2} Figure 11: Diagram showing a distributed execution using Causal Histories increasing counter (starting from 0 initially). When a process %sends a message, a new id is assigned for the send event and the causal history at %until that message is sent along with it. Upon receiving the message at process &, a new id is assigned for the receive event and the causal history at &is merged with the casual history in the message. The crucial point is that identifiers have to be globally unique to correctly represent causality. The partial order of causality can be precisely tracked by comparing these sets under set inclusion. In this case, an event 4happens-before 40(4!40) is equivalent to the Causal History of 4being included in the Causal History of 40(4 0 4). In the same logic, 4k40if and only if 4 0 4and 0 4 4. Marking the last local event added to the history (marked in bold) allows for a simpler test: 40(4!40) if and only if 420 4. Causal Histories correctly track causality relations, but have a major drawback: they grow linearly with the number of writes. In Figure 11, we see two processes %1and %2. We notice that the second event at %1is assigned the id 02and has as causal history 01,a2. The causal history 01,a2is sent with message to be received at %2.Onreceiving,%2assigns the id 12with causal history 01,02,11,b2=01,02[11[12.Wecan clearly notice that 01!12since 01201,02,11,12=12. 2.4.1.5 Vector Clocks While causal histories can capture causality, they have a drawback: their size grows linearly with the number of events and therefore are not very compact. Vector Clocks [30,31] address this problem by offering a more compact representation of causal histories. As we observed earlier, if a causal history H includes an event 4occurring at process ?then it also includes all events occurring at ?and that precede 4. Thus it suffices to store the latest event from every process and consequently reduce the size of these causal histories to the number of processes. For instance, the causal history =01,02,11,12can be represented as 0!2,1!2or simply the vector [2,2]. In causal histories, 4!40iff 4 0 4, which informally means that every event in 4is also in 0 4and there is at least one event in 0 4which is not in 4. In terms of vector clocks, this translates to checking if every entry +⇠4[8]+⇠0 4[8]and there exists at least one entry 9<8for which +⇠4[8]<+⇠0 4[8],or more compactly: 4!40iff +⇠4<+⇠0 4. Furthermore, Lamport clocks or scalar clocks that 4!40means )(4<)(0 4but the converse is not true. This does not allow us to capture concurrency because timestamps are totally ordered. Vector Clocks do not face this problem as the order is a partial order and we can distinguish causally related events from concurrent ones. And by that we get the following: 19 CHAPTER 2. STATE OF THE ART P1• 01[1,0]• ## 02[2,0]// P2• 11[0,1]• 12[2,2]// Figure 12: Diagram showing a distributed execution using vector clocks •+⇠4<+⇠0 4iff 4!40 •4k40iff +⇠4⇥+⇠40^+⇠40⇥+⇠4 The Vector clock mechanism works as follows. Each process ?8maintains a local vector clock +⇠8, avectorof=scalars, where each entry is initially set to +⇠8[:]:=0. Considering an event 4at ?8: •If event 4is an internal event (excluding send and receive events), increment the value of +⇠8[8] by 1, +⇠8[8]:=+⇠8[8]+1; •If event 4is a send event, increment the value of +⇠8[8]by 1, +⇠8[8]:=+⇠8[8]+1and send the new )(<:=+⇠8along with the message <; •If event 4is a receive event of message <with )(<, set every entry in +⇠8to pair-wise maximum of +⇠8and )(<as in +⇠8:=<0G (+⇠8[:],)(<[:]),8:|1<:<=,andthenincrementthe local entry by one +⇠8[8]:=+⇠8[8]+1. In Figure 12, we see two processes %1and %2. We notice that at %1,01!02and also )(01= [1,0]<[2,0]=)(02. Similar observations at %2where 11!12and also )(11=[0,1]<[2,2]= )(12. Also 02!12and also )(02=[2,0]<[2,2]=)(12. Now when we try to compare 11and 02, we notice that )(11)(02and )(02)(11which means that 11k02. 2.4.1.6 Version Vectors Parker et al. [26] introduced versions vectors as a causality tracking mechanism that share the same structure as vector clocks, whereas the update differs. The reason is that the main purpose of using version vectors is to detect inconsistencies among replicas. Basically each replica has a version vector that tracks the last updates known from other replicas as well as from itself. Version Vectors do not need to keep track of every event in the distributed computation, but exclusively updates that alter the replica state. Two main operations are relevant in this case, the update of replicas state and the synchronization between two replicas leading to a convergent state. Therefore the mechanism of maintaining version vectors is different than the ones of vector clocks. Considering a system of n replicas, version vectors are updates as follows: •When a replica A8is updated, it increments the local entry of its++ 8[8]by 1, ++ 8[8]:=++ 8[8]+1; 20 2.4. CAUSALITY TRACKING AND BROADCAST R1• 01[1,0,0]  [1,0,0]// R2• 11[0,1,0]• 12[1,2,0]  [1,2,0]// R3 [0,0,0] [0,0,0] [1,2,0]// Figure 13: Diagram showing a distributed execution using version vectors •When replica A8synchronizes with a replica A9, both versions vectors are updated such as ++ 8= ++9:=<0G(++ 8[:],++9[:]),8:|1<:<= In Figure 13, we see three replicas '1,'2and '3.The•represents replica updates that we want to causally track whereas is used to denote events that are not considered for causality tracking as they do not alter the state of a replica. Replica '1suffers an update 01and concurrently to that '2suffers another update 11. When a message with the update 01from '1reaches '2, two concurrent updates are detected. A new update 12merging the two updates 01and 11is created at '2leading to a new state. When the state of '2is propagated to '3, no concurrent updates are detected and therefore no new version is created and ++'2=++'3=[1,2,0]. 2.4.1.7 Matrix Clocks Matrix clocks are also a type of logical clocks that give the local node a global view of what every node in the system knows about other nodes. They are seen as extensions of vector clocks, a vector of vector clocks. When a message <from process 8is timestamped with a vector clock, it provides information on how many messages where seen by 8from every node prior to send message<. When the same happens using a Matrix clock as a timestamp, the information encoded provides the knowledge that process 8has of what every node :has seen from others until the message <was sent. Formally, time is represented by =⇥=matrices of non-negative integers, where =is the number of nodes or processes in the distributed computation. A process ?8maintains "⇠8011 .. 01= .. .. .. 0=1.. 0==  •"⇠8[8,8]denotes the local logical clock of ?8and tracks the progress of the computation at process ?8 •"⇠8[8,9]denotes the latest knowledge that process?8has about the local logical clock, "⇠9[9,9], of process ?9 •"⇠8[9,:]represents the knowledge that process ?8has about the latest knowledge that ?9has about the local logical clock, "⇠:[:,:],of?: •The entire matrix "⇠8denotes ?8’s local view of the global logical time. 21 CHAPTER 2. STATE OF THE ART To update ts clocks, process ?8uses the following rules: •If event 4is an internal event (excluding send and receive events), increment the value of "⇠8[8,8] by 1,"⇠8[8,8]:="⇠8[8,8]+1; •If event 4is a send event from ?8to ?9, increment the values of "⇠8[8,8]and "⇠8[8,9]by 1, "⇠8[8,8]:="⇠8[8,8]+1and "⇠8[8,9]:="⇠8[8,9]+1and send the new )(<:="⇠8along with the message <; •If event4is a receive event of message<from?9on?8with)(<, update "⇠8as follows: 8:,;|1 :,;=,"⇠8[8,:]:=Kt("⇠8[8,:],)(<[9,:]) and "⇠8[:,;]:=Kt("⇠8[:,;],)(<[:,;]) and finally "⇠8[8,8]:="⇠8[8,8]+1. 2.4.2 Causal Broadcast Causal broadcast is a main group communication delivery technique used in causality middlewares. It makes use of the aforementioned causality primitives to provide a causal ordering of events between a group of processes, following some algorithms. The point of a causal broadcast [24] algorithm is to provide rules on how to use timing primitives, mainly logical clocks, to correctly order events, following the happens-before relation, also known as potential causality, among processes in a group communication environment. In group communication, we consider a number of processes forming a group, where these processes communicate with each other through the sending and receiving of messages. A group could be a closed , where a process should be part of the group to be able to send a message to the group, or an open where this constraint does not need to be satisfied. Another thing to consider in a group is if the group is static , where the membership of that group is fixed, or if it is dynamic , where new processes can join and existing ones can leave. In this section, I will consider closed, static groups for simplicity, and to be able to focus on the causal ordering aspect of the communication. To focus on the ordering, I also assume that the communication is reliable, meaning: • No message is delivered more than once • No message is delivered unless it was broadcast •If ?8and ?9are correct, then every message broadcast by ?8is eventually delivered by ?9 • If one correct process delivers a message <, every correct process eventually delivers < Remark. Message delivery is different from message receiving. We use the term receive to note that a message is received at a certain process via the communication layer, while delivering a message is 22 2.4. CAUSALITY TRACKING AND BROADCAST meant to describe that an already received message can be delivered to the application layer to be used or executed. This differentiation is important when discussing the ordering of events. 2.4.2.1 Causal ordering specifications In a broadcast event, a process ?8sends a message <to all other processes in the system. An event 4=120BC(<)causally comes ahead another event 4=120BC (<0)if at least one of following condition is true: •4and 40have been produced by the same process ?8and 120BC(<)happens before 120BC(<0) •4and 40have been produced by different processes ?8and ?9respectively and the delivery of message <precedes 40at ?9 •9<00|120BC(<)!120BC(<00)^120BC(<00)!120BC(<0) The safety guarantee states that: Given two broadcast messages <and <0such that 120BC(<)! 120BC(<0)then each process have to deliver <before <0(34;8E4A(<)!34;8E4A(<0)). The liveness guarantee states that: eventually every message will be delivered. Consequently, if 2 messages <and <0are such that 120BC(<)k120BC (<0)then <and <0can be delivered in different order on the other processes. 2.4.2.2 Causal Broadcast (CBCAST) algorithm Algorithm 1presents an example of a causal broadcast (CBCAST) implementation. The algorithm assumes the following: • Each process step takes a finite time to occur • Message transfer delays are unpredictable but finite • Communication channels are reliable •Computation is failure free 23 CHAPTER 2. STATE OF THE ART partitions, and that combines both causal consistency and strong convergence. Maintaining causal relationships among events is an expected behavior for most applications and reflects a behavior aligned with our perception of reality. In addition, as we are interested in data stores, data convergence is an important property as users expect that the replicas converge to equivalent states despite if messages are concurrent or not and no matter what happens under the hood. We also visited the different mechanisms and protocols used in the literature to track and guarantee a causal order of events and messages across replicas. But all these properties, trade-offs, algorithms and protocols are as useful as the systems they help design and the range of applications they serve. In this section, we address some techniques used in designing and optimizing causally consistent data-stores and review the most known systems that advanced the literature and made an impact in the complex world of distributed systems. 2.5.1 Causally Consistency Data Stores We will briefly describe some of the most well known causally consistent systems in the literature. 2.5.1.1 COPS In [13] Lloyd et al. present COPS and its variant COPS-GT as causally consistent geo-replicated data stores. COPS stands for Clusters of Order Preserving Systems and uses causal+ consistency to preserve a causal order and a convergent state across its datacenters (DC). COPS-GT also provides get transactions (GT) to the client. The implementation of COPS a small number of clusters, where each is contained in a datacenter. Each datacenter holds a copy of the whole data partitioning the keyset into N linearizable partitions. Each datacenter serves requests locally and then replicates to other DCs asynchronously in the background. The inter-DC replication is causally consistent. Each update is assigned a Lamport clock as a unique version number, and a list of explicit causal dependencies is used as its causal history. Clients are in charge of maintaining their causal history, which is called a context. When a client reads a new object or a newer value of an existing object, they add the object version to its context. When a client wants to update an object, they look at its context which holds a list of dependencies that would consist of the causal history of the new update. To reduce the overhead of causal metadata, COPS offers optimizations by sending less metadata using a transitive reduction on the causal dependencies and sending the “nearest dependencies”. After that, the client sends a message containing the object’s key, the new value and the nearest dependencies to the local server. The server has all the dependencies as the client keeps a session with it and so the server can immediately apply the update, tag it, and return the new timestamp to the client which replaces the value of its context with it. The replication protocol for remote updates follows similar steps. However, as some dependencies might not have been delivered at the remote server, every new update received is blocked at the remote server until all its causal dependencies are delivered. And only after that check the update becomes visible. For concurrent updates, COPS uses a Last Writer Wins (LWW) strategy which is a simple approach but one that is prone to data loss and errors. Also note that client can have different contexts for the same server connection. And as dependencies are tracked per 30 2.5. WEAKLY-CONSISTENT DATA STORES client context, independent operations can have less common dependencies which as a result reduces the time for remote updates to be visible (visibility latency). Moreover, COPS offers a garbage collection mechanism to reduce the local causal metadata that becomes obsolete. 2.5.1.2 Eiger Eiger [60] is a scalable, geo-replicated storage system which could be seen as a successor of COPS. It supports causal+ consistency but uses a column family data model (popularized by Cassandra [17]) instead of a key-value model as most systems. It supports both read-only and write-only transactions and uses operations instead of data versions as dependencies. Write transactions are achieved with a modified two-phase commit protocol that functions differently depending if it is running on the DC that accepted the operation initially or in a remote DC. 2.5.1.3 Orbe Orbe [61] offers causal-consistency by using data versions as dependencies like COPS, but instead of Lamport clocks it uses version vectors that are encoded in a sparse matrix. It has an extension to the protocol that allows for read transactions that added physical timestamps to the dependency matrix. It does not support partial replication or concurrent writes with a multi-value API. Client sessions are also bound to the DC they are running. It also suffers from the same problem as COPS and Eiger regarding data replication, since it needs to apply updates in causal order, thus potentially blocking replication until dependencies arrive. Similarly to COPS, Orbe [61] ensure a causal consistency by using data versions as dependencies. However, instead of using Lamport clocks, it uses a protocol based on dependency matrices (DM). Each datacenter is divided into N partitions, and fully replicated at M different datacenters. Clients only contacts servers in the same datacenters (sticky availability) and are in charge of maintaining causal metadata to track dependencies in the session. A client stores the nearest dependencies of its session in a #⇥"matrix. To read an object the client send a get request with the key of the object. The server looks up the object and return the latest value, an update timestamp and the source replica index. When the client receives the reply, if the entry pointed by the replica index has a smaller value than the returned timestamp, it overwrites that position with the newer value. To update an object, a client sends the key, the new value and its dependency matrix to the local server responsible for the partition in question. Before tagging the update clock time, the server has to increment its local clock, guaranteeing that the assigned value is unique. After storing both the objects value and metadata in an atomic, non-blocking manner, the server replies with the updates clock and its index. Similarly to COPS, the client resets its causal history (dependency matrix) to default values, only saving the returned clock at the given index. Note that the matrix sent by the client in the update request is used for inter-DC replication. For inter-DC replication, each partition in a datacenter communicates updates with its replicas in remote datacenters. The partition forward the updates in their creation order along with the saved metadata comprising the key, value, creation clock, dependency matrix and replica index. Each partition maintains 31 CHAPTER 2. STATE OF THE ART their own version vectors with one entry per replica. To check if remote updates can be safely applied, the receiving server compares the dependency matrix of this updates with their version vector. To be able to safely apply the update, the server must do the following checks: For the same partition, the server checks the corresponding row in the matrix, which has a vector format. If the server’s version vector is larger or equal than the vector (row) of the matrix, the server can proceed to the next step. For other partitions, every time there is a value different from the default one in the partition column, it means that the operation depends on other elements that the current datacenter may not store. So, the server contacts the respective partition, making an explicit dependency check. If all checks pass, the update can safely be made visible and the corresponding version vector’s entry is updated. 2.5.1.4 GentleRain GentleRain [62] is a causally consistent geo-replicated data store that could be considered as the successor to the Orbe system and by the same authors. The main difference with Orbe is that it uses scalar timestamps derived from loosely synchronized physical clocks and only keeps a single scalar to track dependencies and enforce casual consistency. This also means that the timestamps have a fixed size that is not affected by the number of objects or replicas. This reduces the storage and communication overhead but also increases the update visibility latency. Each datacenter is divided into N partitions, and fully replicated at M different datacenters. The data store is a multiversion one, so older versions of an object are kept when the object is updated, until they can be deleted by the garbage collecting mechanism. Clients maintain two timestamps each: a dependency time (DT) which holds the highest update timestamp for all object accessed by the client session, and a global stable time (GST) that will be used to decide the visibility of remote updates. Servers maintain three timestamps each: a version vector (VV), a local stable time (LST) and a global stable time (GST). The VV is a vector of M timestamps, one entry per datacenter. The LST of a server is the smallest update timestamp (minimum entry value in its VV). The GST of a server is the lower bound on minimum LST of all partitions in the same datacenter. To read an object, a client sends a get request to the local server including the key of the object to read and its GST. The server updates its GST if it was smaller than the client’s. The server fetched the latest version of the object for that key if it was created at the local datacenter, otherwise its update timestamp has to be no greater than the server’s GST. The server returns the fetched value, the update timestamp and the server’s GST. The client updates its GST if the server’s one was larger and sets its DT to the returned update timestamp if the latter was larger. To update an object, a client sends a put request to the local server including the object’s key and new value and its DT. The server must guarantee that the update has a larger timestamp than the client dependency time which may result in blocking the operation until its physical clock time is larger. This new update timestamp replaces the current datacenter’s entry in the server’s VV. The server returns the update timestamp to the client to update its DT. For inter-DC replication, the receiving server updates the remote datacenter’s entry of its VV with the update’s timestamp. The update is visible only when its timestamp is lower than the server’s GST. The GST calculation happens periodically inside every 32 2.5. WEAKLY-CONSISTENT DATA STORES DC and is computed by as the minimum of all partitions’ LSTs. As this computation happens periodically and it would be costly to partitions broadcast their LSTs, GentleRain uses a tree structure to disseminate the LSTs (from leaves to root) and then re-disseminate the GST back (from root to leaves) more efficiently. GentleRain uses GST to decide when certain objects are safe to be visible and to garbage collect metadata. Therefore, clock skew resulting from the use of physical timestamps should be kept to a minimum since it affects GST. 2.5.1.5 Cure Cure [63] is a distributed causally consistent key-value store. Although this is not the focus of this chapter or thesis in general, Cure was the first system to allow causally consistent transactions for both object reads and updates. Prior systems used to allow either one or the other. Each datacenter is divided into N non-overlapping partitions, and fully replicated at D different datacenters. The causal dependencies are maintains through the use of vector timestamps with one entry per datacenter and derived from loosely synchronized physical clocks. Each partition maintains two vector clocks. The first (PVC) tracks remote updates received from the replicated partitions at remote DCs. The second maintains the latest globally stable snapshot (GSS) known to the partition. PVCs hold the values of commit timestamps derived from physical clocks. They are updated when a local or remote update is received. The GSS is calculated by the partitions at a datacenter by exchanging their PVCs and computing the lower bound minimum. When a client wants to start a transaction, it must first contact a coordinator of local datacenter to obtain a transaction identifier. The coordinator in a local datacenter is any partition participating in the transaction and it is responsible for committing the client transaction. This transaction identifier is to guarantee the client future reads during the transaction only include objects having version with a lower timestamp that the transaction identifier. As for the client writes, they are buffered locally until the commit phase. During the commit phase, the client sends the buffered writes to the coordinator. The coordinator then contacts all the source replicas of objects included in the transaction. Each of the contacted replicas send their physical clocks back to the coordinator which chooses the maximum as the commit timestamp and propagates it back to inform the replicas. The transaction effects will be visible t the client since the dependencies are locally satisfied. After a successful commit, the updates are propagated to remote replicas asynchronously. Inter-DC replication is a pair-wise process. The receiving server updates the sender’s DC entry in its PVC to the update timestamp. Partitions in the same DC exchange timestamps to calculate a GSS. Remote updates with timestamps lower or equal than the GSS can be made visible. 2.5.1.6 ChainReaction ChainReaction [64] is a geo-distributed key-value datastore that, as the name hints, uses (a variant of) the chain replication [65] technique to provide causal+ consistency using minimal metadata. A datacenter consists of data servers and client proxies. Data servers serve read and write operations for a number of data items. Client proxies receive the requests from end users or client applications and forward them to 33 CHAPTER 2. STATE OF THE ART the appropriate data server. Data servers are organized in a DHT ring fashion and use consistent hashing to partition data items. Also every data item is replicated on R consecutive servers, forming a replication chain. The head of chain receives write requests, the updates are propagated down until reaching the tail. If the tail has received an update then it is safe to assume that every server in the chain has received it too. To track causal history, clients maintain a table, where every entry is per object accessed by that client. Each of these entries contain the object’s key, its version’s timestamp and the chain index which is an index to the node in the chain that last processed and replied to a request of this object. When a client sends a read request, any node between the head of the chain and the node at the chain index can server the request without waiting for a remote update. When a client sends an update request, it sends the object’s key, new value and a compression of the metadata of all objects accessed since the last update. The client proxy forwards the request to the head of the chain. The head of the chain increments its replica’s entry in the timestamp and assigns it to the new object’s version. The update is then propagated down the chain. The update is k-stable when it is replicated on k nodes in the chain and then return to the client the object version and the index of the last receiving node in the chain. The update continues to be propagated down the chain until reaching the tail. When it does, the update is DC-Write-Stable for that replica. Updates are delayed until every object it depends on is DC-Write-Stable, preventing clients from reading inconsistent versions. Moreover, DC-Write-Stable updates need not to be kept in the client’s table. An update is globally stable when it is DC-Write-Stable on all chain replicas in remote DCs. For inter-DC replication on the update timestamp is needed and remote-proxies are the responsible entities of exchanging theses updates across DCs. When a remote update arrives, the receiving remote-proxy compares the updates timestamp to its own. The latter makes sure that the entry value of the sending DC is one greater than its own and equal or higher values for all other entries in its timestamp vector. This means that the dependencies are stable locally and the update can be applied. Otherwise, the update is delayed until the conditions are met. 2.5.1.7 Saturn Saturn [66] is not a data store itself but was designed as a metadata service for existing geo-replicated systems. Te main idea behind Saturn is to efficiently provide causal consistency to systems that do not guarantee it by design. In Saturn, there is a separation between handling the updates themselves and their corresponding metadata. Saturn makes sure that, for every update, the metadata is propagated to the different datacenters so that delivery happens in order that respects causality. It uses physical scalar timestamps to track causality and a tree structure and FIFO channels to propagate them across datacenters. The metadata used to tag updates is called a label. Labels are generated by gears associated to every storage server. A label sink is a component that collects all labels in a DC and is responsible of propagating them to other DCs in an order respecting causality. Remote proxies are responsible of applying remote updates in a causal order. Before being able to send requests, a client has to attach to a datacenter and provide the latest label it has observed. As a result its causal past is visible to the 34 2.5. WEAKLY-CONSISTENT DATA STORES datacenter and thus can safely interact without breaking causality. To read an object, the client sends a request with its key. The gear receiving the request, the value of the object and the label associated with it and returns them to the client. If the returned label is more recent, the client updates its label. To update an object, the client sends the key, new value and its label. The gear receiving the request, creates a new label and stores it along with the new value in the server’s persistent storage. Both update data and metadata are sent to the label sink for inter-DC replication and the new label is returned to the client to replace the old label. The propagation of data and metadata to remote DCs happen separately. Labels are sent following a tree topology through serializers while data is sent in any order. Remote updates can only become visible when their labels arrive. As the the labels sent are scalar, concurrent operations cannot be differentiated from causally related ones in the serialization process. The choice of the serialization then can impact the latency experienced by clients and remote updates visibility latencies when labels (false dependencies) are delivered prematurely. The label dissemination tree and serializers are optimized to minimize such latencies by choosing the best serializations and sometimes delaying label arrivals. 2.5.1.8 Occult Occult [67], which stands for Observable Causal Consistency Using Lossy Timestamps, is causally consistent geo-replicated data store that aims to solve the problem of slowdown cascades. Each replica is located in a datacenter with full copy of the data, sharded on different physical hosts. The system uses asynchronous mater-slave replication instead of multi-master replication. Writes happen on master shards and are then replicated asynchronously but in order to the slave shards in remote DCs. Clients read from their local replica and write to the master shard. Causal consistency is enforced through the use of shardstamps and causal timestamps. Every shard has a shardstamp representing the number of writes it has accepted. A causal timestamp is a vector of shardstamps having one entry per shard. Causal timestamps are used to encode the last state of the data store observed by a client and also to capture the causal dependencies of writes. A client usually reads from their local replica but can contact any replica for load balancing purposes. A client sends a read request with the object’s key. The receiving server replies with the object’s value and its causal timestamp and the shard’s shardstamp. The client compares the received shardstamp to the shard’s entry in its causal timestamp. If the client’s entry is at least equal to the shardstamp, then the version is consistent. If the replica contacted by the client was the master, the check will succeed. The check may fail however if the replica was a slave due to replication delays. In the case of a failed check, the client can retry again or contact the master. When the check succeeds, the client updates its causal timestamp to reflect the dependencies in the returned timestamp. For an update, a client sends a request to the master replica including the object’s key, new value, and the client’s timestamp. The master uses the client’s timestamp to derive the update timestamp by assigning it shardstamp in the replica’s entry and returns the shardstamp to the client. The client uses the returned shardstamp to update its causal timestamp. Before returning to the client, the master starts the replica processes by sending the updates to the slaves with their associated timestamps and the master’s shardstamps by their 35 CHAPTER 2. STATE OF THE ART replicated order. The slaves apply the updates and update their shardstamps to the master’s shardstamp. As the number of partitions tends to be very large, Occult uses three kinds of optimizations to compress the timestamp size: structural compression, temporal compression and isolating datacenters. This lead to compact timestamps but also increases the amount of false dependencies leading to higher visibility latency for updates. 2.5.1.9 Kronos Kronos [68] is a centralized service that uses a dependency graph to encode relationships between events and provide a time ordering for distributed applications. Despite it not being a causally consistent data store, we mention it here for its use of a graph data structure to track dependencies between events. The nodes in the graph represent events and the edges define the causal relationships between them. An identifier of an event is called a reference. A client contacts Kronos to get a reference to an event. After gathering a set of references, a client can contact the service with the set as input to get a sequence of those references. This sequence represents an order of the events that does not violate causality. As concurrent operations can be applied in different order without affecting causality, Kronos can return different results for different clients even for the same set of references. A client can create an event and receive its reference. This adds a node with that reference to graph. The client can then use the references of created events to assign them an order. There are two ways to order events: must and prefer. A must relationship encodes a causal ordering between two events. A prefer ordering is used to assign relationships between concurrent events. In an assign call, all must conditions are applied or the operation is aborted. For prefer events, the operation is not aborted in case of consistencies, but the ordering is discarded. As Kronos is not a geo-replicated solution, there are no remotes updates to be propagated. This leads to a simple design. However, one of the weaknesses of Kronos is that if a client does not receive a certain reference that depends on another already sent to server, the resulting ordering will create an inconsistency at the client state. The client must have all the references to get a global ordering. 36 j Tagged Causal Broadcast 3.1 Introduction In an ideal world, distributed services are able to guarantee strong consistency for data, high availability for reads and writes, and manage to stay operational under failures and partitions. Unfortunately, this was proven to be impossible in what is known as the CAP theorem [9,10]. As there is no “one-sizefits-all” solution, a range of consistency models exist, providing different guaranties and making different trade-offs, each suitable to some classes of applications and services. Causal consistency [69] is the strongest consistency model achievable providing availability under network partitions [21,70]. Under causal consistency, systems can aim for high availability while providing a rich set of session guaranties [71]. Traditional causal delivery middleware [4] provides a delivery order in each process that is consistent with causality, i.e., delivering messages at each process in some order which does not contradict causality. However, it does not provide client applications with knowledge about concurrency under the partial order of causality. Given two messages <1and <2delivered in that sequence to some process, no information is provided to the application whether <1causally preceded <2or if they were originated independently of each other. In this thesis, we argue that providing such knowledge to the application is a mandatory feature for several classes of applications that require knowledge of concurrency in order to apply arbitration rules [5]. An example is an application that given two concurrent bids will arbitrate to consider only the higher bid. Since current implementations of causal delivery middleware lack this feature, they are only suitable for a more limited class of applications where this knowledge is not needed, and thus cannot be used as a general abstraction. We also show that trying to overcome this limitation, using current middleware, often leads to solutions that are inefficient in terms of metadata size and/or delivery delays due to false dependencies. We make a case for a Tagged Causal Delivery (TCD) middleware, which delivers messages together with causality tags (e.g. logical clocks) that characterize the end-to-end happens-before, defined according to client-visible events in the client process order, and ignoring purely internal middleware events (such 37 CHAPTER 3. TAGGED CAUSAL BROADCAST as receiving a message and queuing it for later deliver). We contrast this with previous definitions which can leak internal middleware events, such as the one in [4]. We present both traditional and modern application examples that require the end-to-end happensbefore. First, we revisit the well known critique of CATOCS (Causally and Totally Ordered Communication Support) paper [72] and show that in their stock trading example, if the ordering is exposed to the application, it would be possible to overcome the anomaly. This means that their diagnose that neither causal nor total multicast would work is valid only for current middleware, but not for TCD. As for modern applications targeting high availability, we consider the case of Conflict-free Replicated Data Types (CRDTs) [73]. We argue that to efficiently implement operation-based CRDTs causal delivery is necessary, but not sufficient, as the partial order of operation invocations is needed for the data-type semantics [7]. Together with the ability to compare causality tags in events, TCD provides information about causal stability . We clarify this concept and its importance at the API level, contrast it with classic message stability [4] (which is a purely implementation concern), and argue that it has been reimplemented in several applications (unknowingly). We then focus on the implementation aspects. Achieving TCD should be easy by extending current causal delivery middleware. One could think that it would be a matter of simply exposing to the client API the causality tags (e.g., vector-clocks) that already exist in causal delivery middleware implementations. Surprisingly, we came across several pitfalls in doing so, which we describe in Section 3.5,aswellashowto overcome them. These pitfalls originate from the models of possible interactions between the middleware and the client process, together with the current design goal of simply not contradicting causality, which would lead to an incorrect characterization of causality if internal tags were naively exposed. Finally, we address explicit causality [14], and the tagging by the client application of each individual message with its predecessors. We argue that, although more powerful, explicit causality should (and can always) be left as a last resort, making TCD together with the more “static” choice of delivery groups as the way to achieve cross-object causal consistency among selected groups of objects. 3.2 Classical Causal Delivery Causality ordering is about ensuring that effects are observable only after their causes. In a group communication context, causal delivery guarantees that a message is delivered only after all causally preceding messages. For instance, consider the following scenario where a group of friends communicate using some kind of distributed chat service: - Ross sends to the group: “I won’t be able to join, I have to give a late lecture” - Rachel: “Oh, too bad!” - Then Ross replies: “Lecture canceled, joining soon!” 38 3.2. CLASSICAL CAUSAL DELIVERY - Phoebe: “Great news!!” If causality in not preserved (only some weaker delivery semantics such as FIFO), Joey, who is reading the conversation, could see the following: - Ross sends to the group: “I won’t be able to join, I have to give a late lecture” - Phoebe: “Great news!!” - Then Ross replies: “Lecture canceled, joining soon!” - Rachel: “Oh, too bad!” Joey then might feel confused on why Rachel and Phoebe are not happy with Ross being able to join their gathering. This, however, would not have happened if the chat service used a causal delivery service. Causal delivery was introduced by Birman [24]andSchiper[34], and made popular by the ISIS toolkit. It was further developed in [4], where Birman specifies a potential causality relation (!), similarly to the happens-before relation introduced by Lamport [23], but allowing for each process ?a partial-order ( ? !), as the transitive closure of: 1. if 9?·4? !40,then4!40; 2. 8<·b2M/(<)!`2+2Bp2(<) It distinguishes the receipt of a message (`2+2Bp2) from its subsequent delivery (/2HBp2`) when causal dependencies are satisfied, and makes ? !, which intends to reflect the dependence of events at process ?, respect `2+2Bp2?(<)? !/2HBp2`?(<). A causal delivery middleware ensures delivery order, at each process, for causally related messages, by making them respect potential causality. For any processes ?,@,A, if two sends are related by potential causality, i.e., b2M/@(<1)!b2M/A(<2)the causal delivery service guarantees that /2HBp2`?(<1)? !/2HBp2`?(<2)at every process ?where they are both delivered. Two immediate remarks can be made about this classic specification: • it defines a partial order over all system events (a “system happens-before”), including internal events from the middleware (such as `2+2Bp2), which are not visible to the client; • it defines delivery guaranties provided to the client using this relation, therefore leaking internal events to the specification. 39 CHAPTER 3. TAGGED CAUSAL BROADCAST (a) Internal tag (b) End-to-end tag Figure 17: Difference between internal tagging and end-to-end tagging. one more process. The key insight is that timestamp generation must be made by middleware functions which operate on metadata stored inside some (opaque) middleware object given to the client and stored at the client process. Then: Figure 17a shows an example of how messages are tagged in traditional causal broadcast middleware. The middleware is responsible of tagging the messages upon their arrival. We call it internal tagging as the tagging is dictated by the internal communication layer. In this figure, we notice that in the case where 3 (respectively 3⌫) which is the broadcast and reception of message <B6at ⌫(respectively <B6⌫at ) happens-before 2⌫(respectively 2) which is the internal tagging of message <B6⌫(respectively <B6), then <B6(respectively <B6⌫) will be tagged as happens-before <B6⌫(respectively <B6). While this does not break the causal order provided it has two flaws: • correctness: From the application perspective, <B6and <B6⌫are concurrent. This concurrency is not preserved and therefore and end-to-end causal order is not provided. This is important for a class of application where there is priority/order between concurrent messages. • efficiency: As both messages are in fact concurrent, one could be delivered independently of the delivery of the other. However, as they were tagged as one happens-before the other, this overordering could lead to an extra delay in the delivery mechanism. The general problem stems form trying to solve an application-level problem (semantics, end-to-end) at the communication-level. This renders traditional causal broadcast middleware not suitable for a class of application that, for a correct behavior as expected by the user, requires the preservation of this end-to-end happens-before semantics. Conserving the end-to-end semantics is important for a class of applications where there is priority/order between concurrent messages, and therefore not conserving this order leads to breaking the semantics of the application. In fact two “ingredients” are necessary to be provided by a causal broadcast middleware to fill this gap. The first is providing a way to tag every message at the application (client) 46 3.8. EXPLICIT CAUSALITY VERSUS EXPLICIT GROUPING based on what was seen (delivered) by that application. The second is delivering all messages with that tag and providing a way to compare those tags as causally related or concurrent. We should note that exposing current middleware tags towards achieving tagged causal delivery does not work in general, as they do not characterize the end-to-end happens-before. The general reason for that is the unspecified behavior in terms of the relation between client-visible events (send and deliver) and all other internal middleware events. This should not be surprising, after some thought, as current middleware was designed with the only constraint of delivering in some order consistent with end-to-end happens-before. A simple solution is to allow adapting any middleware using internal tags that describe a partial-order (e.g., vector-clocks) between client-events. The key insight is that timestamp generation must be made by middleware functions which operate on metadata stored inside some (opaque) middleware object given to the client and stored at the client process. In Figure 17b, the end-to-end happens-before order is preserved. The reason is that messages are being tagged at the application level (steps 1,1⌫) independently of the order of their arrival at respective nodes. This follows the next mechanism: • Each time a send is invoked by the client, a timestamp should be immediately generated as part of the (atomic from the client perspective) send event, using the middleware object at the client. If the message is then added to a queue, it has already been tagged. • Each time a deliver event happens at the client process, the timestamp metadata at the middleware object should be immediately updated, before the client action (e.g., a callback) runs. This means that if the client action invokes a send, it will be considered in the causal future (i.e., as resulting from) that delivered message, but not in the future of some ready-to-be-delivered messages that are still in some middleware queue. 3.8 Explicit Causality versus Explicit Grouping In [72], the authors focus primarily on the limitations of CATOCS and mention that a more suited solution that does not violate the end-to-end argument would be a state-level solution. Although the mentioned solution is not covered in detail it seems to be one that provides an “explicit causality ordering” in contrast to the “potential causality” order provided by traditional middleware and by our TCD proposal. Explicit causality [14] is an interesting approach that provides application-specified causal dependencies. Instead of considering all delivered messages at a process as dependencies of the next message to be sent by that process, the application specifies the explicit dependencies (“actual causes”) of the current message. The advantage is a considerable reduction in the causal dependencies that are enforced by the system at the cost of requiring the application to provide accurate information on the dependencies to enforce. In practice this requirement can be hard to achieve, placing a burden on application programmers. A 47 CHAPTER 3. TAGGED CAUSAL BROADCAST benign example, where this is achievable, is to consider in a discussion forum with multiple threads of topic conversations (e.g., Reddit1) only the causal dependencies inside each given topic, as opposed to the whole forum. The limitation is that possible causal connections between topics can be lost, in contrast to the far more encompassing potential-causality that would track them. A middle-ground between potential causality (assuming a total order per process) versus explicitly listing actual causal dependencies, can be obtained by a partitioning into groups of related objects and using TCD to enforce end-to-end happens-before within each group. In the discussion forum example, each topic could be contained in its own group or packed together with other topics in a shared group according to some desired application policy. Explicit grouping is not as versatile, but it is simpler and enough to cover many cases; explicit causality can be left as a last-resort, when everything else fails. 1?iiTb,ffrrrX`2//BiX+QK 48 9 Middleware 4.1 Introduction In chapter 3, we made a case for Tagged causal delivery which provides an end-to-end happens-before and preserves concurrency information as part of the causal metadata. This chapter is the complementary of the previous one and addresses the implementation of a new class of middleware that provide Tagged causal delivery and stability. Our aim was to develop a generic middleware that could replace current implementations that provide the classical causal delivery, which did not address the problems mentioned in chapter 3, and therefore be more suitable for a wider range of applications. In our development of the middleware, different contributions were made and that we present incrementally in this chapter in order to reduce the complexity of the content and offer an easier outline that maximizes the reader’s understanding. In section 4.2 we present the architecture of the system, the internal processes it comprises, what every process does, and how they communicate with each other. We then describe the API of both the system in general so the reader can get the gist of what it is supposed to do. In section 4.3 we start with the most basic implementation of a causal delivery middleware that offers tagged causal delivery using state of the art techniques and data structures such as version vectors and delivery queues. Then, in section 4.4 we start explaining the causal directed acyclic graph (DAG) and dependency dots, one of the main contributions that aims to improve the performance of our algorithm, explain how it works compared to the previous presented techniques in section 2.4, and present a better, simple implementation which uses the dependency dots and the causal DAG in section 4.5. 4.2 API and Architecture In this section, we present a middleware service providing end-to-end happens-before delivery, an efficient tagging and causal delivery protocol. We start by presenting the communication pattern between client and middleware processes in subsection 4.2.1 and then the API of our system in subsection 4.2.2. 49 CHAPTER 4. MIDDLEWARE 4.2.1 Architecture tcbcast queue delivery / stability queue TACS Middleware process Causal Delivery Causal Stability Storage End-to-End Tagging Mngnt TACS Client process tcbcast queue delivery / stability queue TACS Middleware process Causal Delivery Causal Stability Storage End-to-End Tagging Mngnt TACS Client process tcbcast queue delivery / stability queue TACS Middleware process Causal Delivery Causal Stability Storage End-to-End Tagging Mngnt TACS Client process Figure 18: Tagged causal delivery architecture In Figure 18, we present the architecture of our system, the processes that run on every node (replica) in our system in order for the tagged causal delivery algorithm to function correctly as well as the communication pattern between the different processes, both intraand inter-nodes. In every node, runs a client process and a middleware process. The client is a client library that the application using the tagged causal delivery service uses to communicate with the middleware process. The middleware process is the one responsible with most of the work required in ensuring a tagged causal delivery and stability. All requests to query or mutate the state of data objects arrive from the application at the client process. The query requests are trivial: query requests are not broadcast, and do not mutate the state at the client or middleware processes. Therefore, we will only focus on update requests issued by a user and received at the client process. Upon arrival, the client process prepares the request (or message) before being broadcast by the middleware process to every node in the group. The prepare, basically tags the message by a globally-unique identifier and the local context, then enqueues (calls 2M[m2m22< 8(3>C8,2CGC8,<B6)) the prepared message in a FIFO queue, called i+#+bi queue as shown in Figure 18. Every update message has a globally-unique identifier. The local context at the client process is a set of globally-unique Ids that represent the last messages delivered at the client process. The local context represents the causal dependencies of a new update and it is the task of the client process to keep this local context and 50 4.2. API AND ARCHITECTURE manage the end-to-end tagging. The middleware process, on the other hand, handles the receiving, delivery and stability of updates both from its client process as well as from its peer nodes. There are multiple mechanisms involved in the management of correct and reliable transmission of messages, causal delivery as well as causal stability. We keep those for later sections. The middleware process dequeues the local messages from the tcbcast queue, updates the local data structures as needed (that we discuss later) and then broadcasts the message to all other peers. The middleware also receives messages from peers through `2+2Bp29,8(3>C9,2CGC9,<B6), updates the local data structures as needed, and checks the delivery/stability of received messages. When a message is ready to be delivered (state is .Go), it is enqueued in another FIFO queue called delivery/stability queue as shown in Figure 18. Similarly to delivery, when a message is locally marked as causally stable (state is .Go and bstr is {0}#), the stable messages are enqueued in the same queue used for delivery. When the client process dequeues messages from the delivery/stability queue, through /2[m2m2<2 8(), it checks (by pattern matching) if it should deliver a new message or stabilize an already delivered message. In the former case, the i+/2HBp2`8() callback is invoked, in the latter i+bi#H28. Upon delivering a message, the client process updates its local context to reflect the changes in causal dependencies. 4.2.2 API To correctly characterize and end-to-end happens-before, a Tagged Causal Delivery middleware must provide two main functionalities: • a message being sent must be tagged reflecting the causal knowledge according to all delivery events at the application (client process). • a message must be delivered, by the middleware to the application, with its tag that carries causality information, making it comparable (using end-to-end happens-before) to other messages. Therefore, TRCB provides in its API, the following callbacks: •i+#+bi: (tagged causal broadcast) broadcast a message to every process in the group membership. The client process will be responsible for tagging the message to be sent with the causal predecessors known (delivered) by the client process. •i+/2HBp2`: (tagged causal delivery) delivers a message to the application with the tag. For instance, in the case of pure operation-based AW-Set (0,i+/2HBp2` would add the operation (033,4, and respective timestamp C) to the state (0. It uses the existence of comparable timestamps for operations to update the state, based on the semantics of AW-Set. •i+bi#H2: (tagged causal stability) We postpone explaining this callback to the next chapter. 51 CHAPTER 4. MIDDLEWARE 4.3 Vector-Clock-based algorithm for tagged causal delivery A vector clock is a mechanism to determine the causal relation between events. In a group with N peers, each of these has a version vector with N entries. Initially every position in the vector starts at 0and increments by 1each time an event happens in the peer corresponding to that entry. The events that trigger an update of a node’s version vector can either be the broadcast or delivery of a message. When sending, the peer increments its entry in the local version vector before copying it to the message. Upon receiving a message, the peer compares the message’s version vector with its own. A message can only be delivered if all the events that caused it have also been delivered. For example, peer i can only deliver a message m from j with version vector +<if: +<[9]=+8[9]+1^+<[:]+8[:]|8:<9 When this condition is not met, the message is queued in a data structure called message queue, until it can be delivered later on when this condition is met. The components for this version include vector clocks are used here as timestamps to characterize causality between events, a queue that stores the messages that cannot be delivered yet, as well as the previous condition presented above to check causal delivery of a message. We extend current causal middleware to provide tagged causal delivery and we show that in the next subsections 4.3.1 for the client process and 4.3.2 for the middleware process . 4.3.1 Client process The algorithm for the client process can be seen in Algorithm 2.The Client keeps locally as its state a version vector +8, with length equal to the group size (number of participants or nodes). Initially every position of this vector starts at 0and is incremented upon a send or delivery. An application uses the client library (process) to broadcast an update or any kind of message by calling tcbcast .Forpeer i the +8[8]entry of its version vector is incremented. The tcbdeliver callback is called and the payload along with the process’ updated version vector are enqueued using 2M[m2m22< 8(the superscript 2< denotes the direction from client to middleware) to tcbcast and sent to the Middleware process. /2[m2m2<2 8is a another FIFO queue storing messages that are tagged by the middleware process as ready to be delivered by the client process. The sender’s entry in +8is incremented before calling the tcdeliver callback. 52 4.3. VECTOR-CLOCK-BASED ALGORITHM FOR TAGGED CAUSAL DELIVERY G:P_Ah>J k, Vector-based tagged causal delivery algorithm without causal stability at client process for node 82I Rstate: k+8:I!Nf /2HBp2`2/ p2`bBQM p2+iQ` f jproc BMBi8() 9+8:={97! 0|92I} 8proc i+#+bi8(<B6) e+8:=+8[8]+1 di+/2HBp2`8(8,+8,<B6) 32M[m2m22< 8(msg,+8,<B6) Non /2[m2m2<2 8(9,+<,<B6) Ry +8[9]:=+<[9] RR i+/2HBp2`8(9,+<,<B6) Rk fun i+/2HBp2`8(9,+<,<B6) f /2HBp2`v +HH#+F f 4.3.2 Middleware process The algorithm for the Middleware process is shown in Algorithm 3.Eachpeer i has a version vector '8for received messages, a version vector +8for delivered messages and a delivery queue ⇡&8.The version vectors '8and +8have, respectively, the id of the last received and delivered message. Initially, their entries are initialized to 0 and delivery queue ⇡&8is set to empty (). /2[m2m22< 8dequeues the local updates from the client process, updates +8to reflect the delivered message and broadcasts a message containing the tag msg to denote its type, the replica id 8, its version vector +as sent by the client process, and its payload <B6. When a message from j is received at i the '8[9]entry is incremented and when a message is delivered the+8[9]entry is incremented. If a received message cannot be immediately delivered then it is enqueued to ⇡&8to be later delivered after all its causal predecessors have also been delivered. On the other hand, when a received message is delivered, ⇡&8is traversed until there are no more messages that can be delivered at the moment. This is because upon a message delivery, other messages that were queued could be ready to be delivered due to one of their causal predecessors being received and delivered. Removing an element from the middle of a vector can hinder performance since all the elements after it have to be shifted to the left. Instead of this, while looping through the queue to check delivery, if a message cannot be delivered then it is individually shifted left to the right place, otherwise it is enqueued to tcdeliver .Whentheendof⇡&8is reached and at least one delivery was made then it is again traversed from the beginning because other messages might be deliverable. If at the end no deliveries were made then ⇡&8is truncated, since the first messages on this queue were considered received but not deliverable, and they were shifted to the left while traversing, leaving unused positions in the end of the vector. 53 CHAPTER 4. MIDDLEWARE G:P_Ah>J j, Vector-based tagged causal delivery algorithm without causal stability at middleware process for node 82I Rstate: k+8:I!Nf /2HBp2`2/ p2`bBQM p2+iQ` f j'8:I!Nf `2+2Bp2/ p2`bBQM p2+iQ` f 9⇡&8:f /2HBp2`v [m2m2 f 8on BMBi8() e+8:={97! 0|92I} d'8:={97! 0|92I} 3⇡&8:=; Non /2[m2m22< 8(hmsg,+,<i) Ry +8[8]:=+8[8]+1 RR #`Q/+bi8(hmsg,8,+,<i) Rk on `2+2Bp29 8(h9,+<,<i) Rj if '8[9]<+<[9]then R9 '8[9]:='8[9]+1 R8 if +<[9]=+8[9]+1^+<[:]+8[:]|8:<9then Re 2M[m2m2<2 8(hdlv,9,+<,<i) Rd /2HBp2`8() R3 else RN ⇡&8:=⇡&8+(9,+<,<) ky proc /2HBp2`8() kR 5A>< :=0 kk C> :=0 kj while true do k9 if 5A>< ;4=(⇡&8)then k8 if C> >=5A><then ke 1A40: kd CAD=20C4(⇡&,C>) k3 if ;4=(⇡&8)=0then kN 1A40: jy 5A>< :=0 jR C> :=0 jk else jj (9,E,E<)=⇡&[5A><] j9 if E<[9]=+8[9]+1^+<[:]+8[:]|8:<9then j8 +8[9]=+8[9]+1 je 2M[m2m2<2 8(hdlv,9,+<,<i) jd else j3 ⇡&[5A><]:=⇡&[C>] jN C> :=C> +1 9y 5A>< :=5A><+1 4.4 Causal DAG In the section we revisit the use of data structures and algorithms used in the classical causal delivery implementations and propose different ones that we consider more suitable and efficient for representing causality. In the following we explain why we moved to using “dependency dots” instead of vector clocks and then propose a causality Directed Acyclic Graph to hold the causal metadata as well as replace the traditional message queue. A1 C1 C2 A2 B2 SLT B1 Xn Xn Xn RCV DLV A2 B1 C2 A2 B1 C0 A0 B2 C0 A0 B1 C0 B1 110 101 101 101 101 111 depends on 111 bitstring Figure 19: Graphical representation of causal delivery using a delivery queue and a causal DAG 54 4.4. CAUSAL DAG 4.4.1 Reducing the causal information overhead In the initial implementation, we used vector clocks to track causal information between operations. This information is coupled with each operation to allow the delivery of the operations in an order respecting causality of events. Vector clocks have been used widely to track causality and achieve causal delivery in a group communication environment. However, as their size grows linearly with the number of participants in the group, they become less efficient in large scale settings. To support large scale group communication services, reducing the causal information overhead imposed by the use of vector clocks is a problem to be solved. We tackled this problem by implementing “dependency dots” (see Figure 19). Those “dependency dots” are similar to some approaches used to compact the size of vector clocks [4,45,46]. We also applied the transitive reduction on those dependency dots that allowed reducing the size even more. While similar ideas exist in the literature, we present the implementation of the protocol using this alternative and how the transitive reduction can be applied, which is not tackled previously. The main idea here is intuitive. A vector clock has one slot per peer representing the number of updates seen (delivered) from the peer. This is a very compact information to represent causality. However in large scale settings two things happen: the number of slots needed increases which makes the vector clocks large in size, and most often only some of the nodes (at least at any point in time) are the ones making updates. What this means is that a whole vector clock for every update when only a few slots need to be updated seems like a lot of redundant information that could be trimmed down to compact the size of the vectors and reduce the causality information overhead. A dot is represented as § BI⇥Nand is used as a unique identifier for a message. When a message <1represented by the dot 3causally happens-before another message <2represented by the dot 30it means that 30causally depends on 3and therefore we call 3a causal dependency of 30.Adotcanhave more than one dependency dot, a set of dependency dots to characterize causality. At most the size of this set of dependency dots is equal to the number of participants, thus making the size of our “dependency dots” equal to the size of a vector clock only in the worst case scenario. But in general it small as a node only needs to send a number of dots equal to the number of updates delivered since its last send. Obviously, many updates could be delivered at a node between two consecutive sends, however, these updates are not necessarily concurrent and therefore there exists some dependency between them. This information can be reduced to only keep the most recent dots in the set of dependencies and it’s called a transitive reduction on the dependency graph. For example, looking at Figure 19, we can see that⇠2could have in its set of dependencies {1,⌫1,2,⇠1}. By applying a transitive reduction on this subgraph, we only need to keep {2,⇠1}because ⇠2directly depends on only 2and ⇠1but transitively on 1and ⌫1. So having 2in ⇠1’s dependency dots is enough to mean 1,⌫1,and2. Now, the size is of our dependency dots that a node need to send in a message is reduced even more to be equal to the number of concurrent updates delivered at that node since its last sent message. 55 8 Tagged causal delivery and Causal Stability In this section, we introduce causal stability, explain its definition and how it differs from other stability related concept in distributed systems. We then explain how the implementation of causal stability works in both the vector-based and graph-based versions. Then we show a new version of the tagged causal delivery algorithms extended to provide causal stability, also in both vector-based and graph-based versions. 5.1 Causal Stability Stability is one of the most overloaded terms in distributed systems. Examples are stable storage ,for durable storage which survives process failure in a crash-recover model, or self stabilization [80]toreferto reaching some global predicate within a finite number of steps. For causal delivery middleware, Birman [4] defined message stability to mean that it is known that a message has been received at all intended nodes (and : -stability to mean that it has been received by :nodes). This concept is relevant for implementation purposes, namely for garbage collecting messages in a fault-tolerant system, as it allows a message to be discarded when it been delivered locally and it has become stable. Adifferentconceptis causal stability , named as such in a paper that defines pure operation-based CRDTs [7], which make use of causal delivery. Causal stability concerns knowledge about the end-to-end happens-before regarding future deliveries at each client process. Definition 1 (Causal Stability). A causal timestamp C , and corresponding message, is causally stable at node 8 when all messages subsequently delivered at 8 will have causal timestamp C0>C , according to the end-to-end happens-before. This implies that no message with a timestamp C0concurrent with Ccan be delivered at 8once Cis causally stable at 8. For a message to be causally stable at some node, not only is it needed to have been received everywhere, but also that it has been delivered everywhere, and that no further concurrent messages may be delivered at that node. Therefore, causal stability is a stronger notion, implying classic message stability. We notice that, depending on how knowledge is piggybacked, middleware can infer that some message has been received everywhere, therefore being stable, even if itself or other causally 62 5.1. CAUSAL STABILITY concurrent messages have not yet been delivered. This notion differs from classic message stability in two fundamental ways: • it is a per-node property (a message may be causally stable at node 8but not at node 9), while a message is stable if (it is known that) it was received everywhere; • it is about message deliveries, therefore, a client-visible concept, relevant for the client API, while message stability is essentially an implementation aspect about internal middleware events. So, even if there are some similarities, causal stability is a different concept, and we consider it important to use a different name, to avoid confusion with classic message stability. In fact, we have already observed this confusion many times, namely in discussions about implementing causal stability. The concept itself is not new, and it has been used many times, but hidden in applications, without being properly recognized (because it has not been clearly identified). This has lead to successive ad-hoc reimplementations, sometimes overly complex. As an example, causal stability is hidden inside the difficult to understand implementation of Replicated Growable Arrays (RGAs) [6], while not being recognized as a building block. We argue that this concept is important enough to be recognized and provided by the middleware. The TCD middleware can offer causal stability information through extending its API with a i+bi#H28(g) event, which informs the client application that message with timestamp gis now causally stable locally. Node 8can check this by verifying if a message with timestamp C>ghas already been delivered at 8 from every other node 9in the set of nodes I. More formally, if /2HBp2`8() represents the set of message timestamps that have been delivered at node 8and b`+(C)denotes the node that sent the message corresponding to C: i+bi#H28(g)if 892I\{8}·9C2/2HBp2`8() · b`+(C)=9^g<C. One possible implementation of causal stability would be similar to the RGA tombstone deletion algorithm [6]. Essentially, each node only needs to keep the last causality timestamp delivered from each origin. 5.1.1 Causal Stability in the vector clock-based algorithm for TCD Each node 8keeps a map !8(from node identifiers to vector clocks) with the last vector clock, delivered locally, from each other node. This allows defining a function HQr for the greatest lower bound on messages issued at 9and delivered to all nodes: HQr(!,9)⌘KBM({!(:)(9)|:2I}). This allows knowing at node 8that, e.g., if HQr(!8,9)=4, then all nodes have delivered at least message number 4 from node 9. Using this function we can now define a causal stability oracle. A timestamp gis 63 CHAPTER 5. TAGGED CAUSAL DELIVERY AND CAUSAL STABILITY causally stable at node 8if g(b`+(g)) HQr(!8,b`+(g)). This ensures at 8that message gwas already delivered at all nodes and that each node issued a message after delivering g, that is already delivered at 8. Since any messages concurrent to gfrom any node :must have been issued prior to gbeing delivery at :then, due to causal delivery, they must have also been delivered at node 8. 5.1.2 Causal Stability in the graph-based algorithm for TC In the graph-based implementation there is no need for keeping an extra matrix or other data structure for stability. In fact, the bit string used for checking causal delivery of a dot can be used for checking causal stability once the dot is marked as DLV. Every bit position in the bit string represents one of the peers in the membership. When the bit string reflects that every peer have delivered this dot it can become stable at that node. Using bit strings is efficient as it requires minimal space in memory, and the bitwise operations are not costly in time complexity. Whenever a message from node 9is delivered at 8, we know that it has also been delivered at 9as well as all its predecessors. Therefore, all the bit position of node 9can be updated in the bit strings of that dot and every predecessor, immediate and transitive. Marking a dot a stable by the middleware happens in an order respecting causal order, meaning that older dots get stabilized before their successors and so on. Whenever a dot is marked as stable, the client is notified to invoke the i+bi#H2 callback and then the client informs the middleware that this dot has been stabilized and it is safe to be removed from the graph. D Dot removed from the graph SLT RCV DLV STB stability D D Dot received local VV updated D Dot updated Add Dot to graph as RCV Dot exists delivery Figure 22: Stages of a dot in causal DAG (with causal stability) In Figure 22, we tackle again the different stages of a dot in our causal DAG but now with causal stability. As we explained previously the different stages, we now add one more stage for causal stability 64 5.1. CAUSAL STABILITY that we call stb. Moreover, we mentioned in the previous version without stability a dot does not stay in the graph with stage dlv. Here however, a dot stays in the graph after being delivered and the bitstring is used to compute when a dot is causally stable. Basically, whenever a dot is delivered, the algorithm updates the botstring of all its predecessors by setting to 0 the bit corresponding to the actor/node from which the delivered dot was sent originally. For instance, when a dot ⇡5which represents the 5th message sent from node ⇡is delivered all its predecessors of ⇡5in the causal DAG at the receiving node update their bitstring values by setting the bit corresponding to node ⇡to 0. When the bitstring value of delivered dot (stage dlv) is 0, it can be stabilized and its stage becomes stb. After a dot is stb it is ready to be stabilized by the client process. When the client process dequeues the stabilize message of a certain dot, the middleware can safely delete the stable dot from the causal DAG. A1 C1 C2 A2 B2 SLT B1 Xn Xn Xn Xn RCV DLV STB A2 B1 C2 A2 B1 C0 A0 B2 C0 A0 B1 C0 B1 A2 B1 C2 A0 B2 C0 A2 B1 C2 Min A0 B1 C0 = ][ 010 110 100 011 010 depends on 111 delivery/stability bitstring Figure 23: Graphical representation of causal delivery using a delivery queue and a causal DAG (with causal stability) Figure 22 is also similar to Figure 20 from the previous chapter. Here we add to the previous explanation, the part related to causal stability. When using version vectors to represent and track causal dependency among messages, causal stability can be calculated by caculating the greatest lower bound of the version vectors in the stabilty matrix seen in Figure 22. The stability matrix holds at each entry/row the last delivered version vector of the node corresponding to that row. In the figure, the first row corresponds to node , the second row to node ⌫and the third to node ⇠. The greatest lower bound allows to compute a stable version vector (SVV). The calculation of the SVV and causal stability in general is local to the node and not global, meaning that a message can be stable at but not necessarily stable at ⌫. At each node, when the SVV is computed, we know that it is less or equal to every last version vector delivered from every node. This means that any message that will be delivered after the SVV is computed at that node will be greater than the SVV. A message can be stabilized at a node if its version vector is less than the SVV at that node. As for the causal DAG-based version, the bitstring functionality changes whenever a dot is delivered to be used for stability. In the previous version where we did not use causal stability, the bitstrinng was only used to check if a dot is ready to be delivered when the bitrsting is equal 65 CHAPTER 5. TAGGED CAUSAL DELIVERY AND CAUSAL STABILITY to 0. Now that causal stability is used, the bitstring is used to check delivery when the stage is rcv and once delivered used similarly to check causal stability when the stage of the dot is dlv. 5.2 Causal Stability for VV-based Algorithm 5.2.1 Client process The algorithm for the client process can be seen in Algorithm 6. The algorithm is very similar to the previous version without stability presented earlier in Algorithm 2with some incremental changes for causal stability. Therefore, we will only address the changes without repeating the previous explanation. We notice that /2[m2m2<2 8is not only for dequeuing messages to be delivered. Instead it expects a type based on which it can either result in delivering a message stabilizing a message. If it is a delivery, the sender’s entry on +8is incremented before invoking the tcdeliver callback. Otherwise, if it is a stable message the tcstable callback is invoked. G:P_Ah>J e, Vector-based tagged causal delivery algorithm with causal stability at client process for node 82I Rstate: k+8:I!Nf /2HBp2`2/ p2`bBQM p2+iQ` f jproc BMBi8() 9+8:={97! 0|92I} 8proc i+#+bi8(<B6) e+8:=+8[8]+1 di+/2HBp2`8(8,+8,<B6) 32M[m2m22< 8(msg,+8,<B6) Non /2[m2m2<2 8(C~?4,8,+<,<B6) Ry if C~?4 =dlv RR +8[9]:=+<[9] Rk i+/2HBp2`8(9,+<,<B6) Rj else if C~?4 =stb R9 i+bi#H28(9,+<,<B6) R8 fun i+/2HBp2`8(9,+<,<B6) f /2HBp2`v +HH#+F f Re fun i+bi#H28(9,+<,<B6) f bi#BHBxiBQM +HH#+F f 5.2.2 Middleware process The algorithm for the Middleware process is shown in Algorithm 7. Also, this algorithm shares parts in common with its previous version without causal stability, presented in Algorithm 3. In this subsection, 66 5.2. CAUSAL STABILITY FOR VV-BASED ALGORITHM we will only explain the new notations, data structures, functions and procedures that have not been explained previously. G:P_Ah>J d, Vector-based tagged causal delivery algorithm with causal stability at middleware process for node 82I Rstate: k+8:I!Nf /2HBp2`2/ oo f j'8:I!Nf `2+2Bp2/ oo f 9⇡&8:f /2HBp2`v [m2m2 f 8"8:I!+8f bi#BHBiv Ki`Bt f e(+8:I!Nf bi#H2 oo f d("0?8:Dõ!Mf KT Q7 bi#H2 /Qib f 32CA8:Nf +QmMi2` f Non BMBi8() Ry +8:={97! 0|92I} RR '8:={97! 0|92I} Rk ⇡&8:=; Rj "8:={97! +8|92I} R9 (+8:={97! 0|92I} R8 ("0?8:=; Re "8=8:={97! 0|92I} Rd 2CA8:=0 R3 on /2[m2m22< 8(hmsg,+,<i) RN +8[8]:=+8[8]+1 ky mT/i2bi#BHBiv8(8,+,<) kR #`Q/+bi8(hmsg,8,+,<i) kk on `2+2Bp28(h9,+<,<i) kj if '8[9]<+<[9]then k9 '8[9]:='8[9]+1 k8 if +<[9]=+8[9]+1 ke ^(+<[:]+8[:]|8:<9)then kd 2M[m2m2< 8(hdlv,9,+<,<i) k3 mT/i2bi#BHBiv8(9,+<,<) kN /2HBp2`8() jy else jR ⇡&8:=⇡&8+(9,+<,<) jk proc mT/i2bi#BHBiv8(9,+<,<) jj "8[8]:=+8 j9 if 8<9then j8 "8[9]:=+< je 2CA8:=⇠CA8+1 jd ("0?8[(9,+<[9])] :=(+<,<,2CA8) j3 if 92"8=8then jN #4F(+ =+H+mHi2ao(9) 9y if #4F(+ <(+8then 9R (C01;4⇡>CB =#4F(+ (+8 9k bi#BHBx28((C01;4⇡>CB) 9j (+8:=#4F(+ 99 proc /2HBp2`8() 98 5A>< :=0 9e C> :=0 9d while true do 93 if 5A>< ;4=(⇡&8)then 9N if C> >=5A><then 8y 1A40: 8R CAD=20C4(⇡&,C>) 8k if ;4=(⇡&8)=0then 8j 1A40: 89 5A>< :=0 88 C> :=0 8e else 8d (9,E,E<)=⇡&[5A><] 83 if E<[9]=+8[9]+1^+<[:]+8[:]|8:<9then 8N +8[9]=+8[9]+1 ey 2M[m2m2< 8(hdlv,9,+<,<i) eR mT/i2bi#BHBiv8(9,+<,<) ek else ej ⇡&[5A><]:=⇡&[C>] e9 C> :=C> +1 e8 5A>< :=5A><+1 ee proc bi#BHBx28((⇡) ed !={("0?8[3]|32(⇡} e3 (=B>AC(!,|G,~|G.2<~.2) eN for (E,<,2)2(do dy 2M[m2m2<2 8(hstb,E,<i) dR ("0?8:=(⇡ C("0?8 dk fn +H+mHi2ao(9) dj =4F(+ :=(+8 d9 for 2>; =0..=do d8 if "8=8[2>;]=9then de <8= :="8[0][2>;] dd <8=A>F :=0 d3 for A>F =1..=do dN if "[A>F][2>;]<<8= then 3y <8= :="8[A>F][2>;] 3R <8=A>F :=A>F 3k =4F(+ [2>;]:=<8= 3j "8=8[2>;]:=<8=A>F 39 return =4F(+ For determining causal stability, the middleware has a stability matrix "8, a stable version vector (+8, a map of stable dots ("0?8,avector"8=8and a counter2CA8. For a group with N peers, each middleware 67 CHAPTER 5. TAGGED CAUSAL DELIVERY AND CAUSAL STABILITY has a #⇥#matrix "8, where row j is the version vector of the last delivered message from node j .(+8 is a vector with N entries, where each position is the minimum of the same column in the matrix "8. With the minimum of each column it is possible to know what is stable (from the perspective of i that has "8) by comparing the message’s version vector with these minimums. The "8=8is vector where each entry is the row with minimum of the corresponding column of "8=8. Delivered but not yet stable message are saved in the ("0?8.The2CA8field is an integer that is used to order stable messages before enqueuing them to the Client . When a message <9is delivered, row "8[9]is updated with its version vector, and it is also added to the map ("0?8. Calculating the (+8vector every time "8is updated can become costly, specially when dealing with large groups. To overcome this problem the "8=8vector was created, by checking if the sender’s id is in it. If it is not, then the minimums of the columns are the same and (+8has not changed. Otherwise, it means that the row where the minimum was has changed, and so the could the minimum have also changed. Therefore, for each entry k on "8=8where "8=8[:]is the same as the sender’s id, the column k of "8is traversed to check for the minimum and a row with that minimum. Then every entry in the current (+8with the sender’s id is updated and if the new vector is not the same as the previous, it means that there are new stable messages. The entries that are different when comparing these stable vectors are the new stable messages. For example, for a group with 4 peers, if the previous and new (+8vector were, respectively, [15,9,10,12]and[16,9,10,13], then the message with id 16 from peer 0and the message with id 13 from peer 3became stable. By having the rows with the minimums instead of calculating them at each update reduces overhead and time on calculations. 68 5.3. CAUSAL STABILITY FOR GRAPH-BASED TCB ALGORITHM 5.3 Causal Stability for graph-based TCB Algorithm 5.3.1 Client Process G:P_Ah>J 3, Graph-based tagged causal delivery algorithm with causal stability at client process for node 82I Rstate: k382D j2CGC82C 9proc BMBi8() 838:=(8,0) e2CGC8:=; dproc i+#+bi8(<B6) 3i+/2HBp2`8(38,2CGC8,<B6) N2M[m2m22< 8(msg,38,2CGC8,<B6) Ry 38:=(8,38.1+1) RR 2CGC8:=3 Rk on /2[m2m2<2 8(C~?4,3,2CGC,<B6) Rj if C~?4 =dlv R9 2CGC8:={3}[2CGC8\2CGC R8 i+/2HBp2`8(3,2CGC,<B6) Re else if C~?4 =stb Rd i+bi#H28(3,2CGC,<B6) R3 2M[m2m22< 8(stb,3) RN fun i+/2HBp2`8(3,2CGC,<B6) f /2HBp2`v +HH#+F f ky fun i+bi#H28(3,2CGC,<B6) f bi#BHBxiBQM +HH#+F f The algorithm for the client process can be seen in Algorithm 8. The algorithm is very similar to the previous version without stability presented earlier in Algorithm 4with some incremental changes for causal stability. Therefore, we will only address the changes without repeating the previous explanation. /2[m2m2<2 8(C~?4,3>C,2CGC,<B6)checks the arrival of any messages at the delivery/stability queue. Whenever this queue has a message ready, if the message is a delivery message then the callback i+/2HBp2`8(3>C,2CGC,<B6)is invoked to deliver the message to the application. Also the context C8 is updated by removing any dot in C8that already exists in C9of the message to be delivered, and adding {3>C9}. This is what makes the context at each client process hold only the immediate predecessor without any transitive predecessors. In the case where the message dequeued is a stability message, i+bi#H28(3>C,2CGC,<B6)callback is invoked and the 3>C is stabilized at the application. Here the context is not updated. Finally, a message is enqueued back to the middleware process signaling that the dot has been stabilized and it can be deleted form the graph safely. 69 CHAPTER 5. TAGGED CAUSAL DELIVERY AND CAUSAL STABILITY 5.3.2 Middleware Process G:P_Ah>J N, Graph-based tagged causal delivery algorithm with causal stability at middleware process for node 82I Rinputs: k#Bi :I!Nf #Bi KbF ;Bp2M MQ/2 B/ f jstate: 9⌧8:Dõ!Mf K2bb;2 ;`T? f 8+8:I!Nf /2HBp2`2/ p2`bBQM p2+iQ` f eon BMBi8() d⌧8:=; 3+8:={97! 0|92I} Non /2[m2m2< 8(hmsg,3,%,<i) Ry #`Q/+bi8(hmsg,3,%,<i) RR +8[3.0]:=3.1 Rk %0={?2%|¬bi#H28(?)} Rj for ?in %0do R9 ⌧8[?].bm++ :=⌧8[?].bm++ [{3} R8 ⌧8[3]:={bi;2 :dlv,#Bib :⇠#Bi(8), Re T`2/ :%0,bm++ :;,Kb; :<} Rd mT/i2bi#BHBiv8(⇠#Bi(8),3) R3 on `2+2Bp28(hmsg,3,%,<i) RN if +8[3.0]<3.1^¬(⌧8[3].bi;2 =rcv)then ky %0={?2%|¬bi#H28(?)} kR for ?in %0do kk if ?8/QK(⌧8)then kj ⌧8[?]:={bi;2 :slt,bm++ :;} k9 ⌧8[?].bm++ :=⌧8[?].bm++ [{3} k8 1=Õ{#Bi(?.0)|?2%0^⌧8[?]<dlv} ke (=if 32/QK(⌧8)then ⌧8[3].bm++ kd else ; k3 ⌧8[3]:={bi;2 :rcv,#Bib :1 kN T`2/ :%0,bm++ :(,Kb; :<} jy if 1=0then jR /2HBp2`8(3) jk on /2[m2m28(hstb,3i) jj /2H2i2bi#H28(3) j9 fn bi#H28(3) j8 3.1+8[3.0]^(38/QK(⌧8)_⌧8[3].bi;2 =stb) je proc /2HBp2`8(3) jd 2M[m2m2< 8(hdlv,3,⌧8[3].T`2/,⌧8[3].Kb;i) j3 (9,=)=3 jN +8[9]:== 9y ⌧8[3]:=⌧8[3]{bi;2 :dlv, 9R #Bib :⇠(#Bi(8)|#Bi(9))} 9k mT/i2bi#BHBiv8(⇠#Bi(9),3) 9j for Bin ⌧8[3].bm++ do 99 ⌧8[B].#Bib :=⌧8[B].#Bib &⇠#Bi(9) 98 if ⌧8[B].#Bib =0then 9e /2HBp2`8(B) 9d proc mT/i2bi#BHBiv8(1,3) 93 for ?in ⌧8[3].T`2/ do 9N if ⌧8[?].bi;2 <stb then 8y 10=⌧8[?].#Bib &1 8R if 10<⌧8[?].#Bib then 8k ⌧8[?].#Bib :=10 8j if 10=0then 89 bi#BHBx28(?) 88 else 8e mT/i2bi#BHBiv8(1,?) 8d proc bi#BHBx28(3) 83 %=⌧8[3].T`2/ 8N for ?in %do ey if ⌧8[?].bi;2 <stb then eR bi#BHBx28(?) ek ⌧8[3].bi;2 :=stb ej 2M[m2m2< 8(hstb,3,%,⌧8[3].Kb;i) e9 proc /2H2i2bi#H28(3) e8 for Bin ⌧8[3].bm++ do ee ⌧8[B].T`2/ :=⌧8[B].T`2/ \{3} ed ⌧8:=3C⌧8 The algorithm for the Middleware process is shown in Algorithm 9. Also, this algorithm shares parts in common with its previous version without causal stability, presented in Algorithm 5. In this subsection, we will only explain the new notations, data structures, functions and procedures that have not been explained previously. As mentioned before, the causal relations between messages are maintained through a graph, specifically a Directed Acyclic Graph or DAG . The middleware has as state a version vector +8and a DAG ⌧8. The number of entries on +8is the same as the number of peers in the group and each of them represent the cntr field from the dot of the last delivered message from the peer corresponding to that 70 5.3. CAUSAL STABILITY FOR GRAPH-BASED TCB ALGORITHM position. Initially, +8starts with 0in all positions and an empty ⌧8.Anodein⌧8represents a message and has five parameters: stage , bits , pred , succ and msg . The stage is an enum of the node’s current state and can either be as a slot SLT ,received RCV , delivered DLV or stable STB .A SLT node is a message that has not yet been received but it is a predecessor of another received message that has added it to the graph to serve as a placeholder. A node marked as RCV means that the message has been received but not delivered; if it is marked as DLV then it has been delivered but it is not stable and only it is considered as such when it has stage STB . Every node has a bits field that is a string of bits with the same size as the number of peers in the group. Depending on the node’s stage these bits are used either to determine if a message can be delivered or considered stable. If a bit is 1while the node has stage RCV it means that a causal predecessor from the peer in that position has not been delivered. When a predecessor is delivered, the bit position of the sender in their successors are set to 0. Only when all the bits of a node are 0can the message be delivered. Otherwise, when a message is marked as DLV this string is used to determine causal stability. After delivery, the bit field starts with all elements as 1(except the local and sender peers positions) and the delivery of a message successor to the node sets a bit to 0in the respective position of the sender peer. The bits of peer i and the sender peer are 0after delivery because the node’s message already is in the past of the their next message. Besides these, a node also has a pred and succ field. These are, respectively, the dots of the message’s causal predecessors and successors. Furthermore, the msg is the message that a peer sent. When the Client process calls tcbcast , a message m is enqueued to tcbcast , along with a dot d and a list of predecessor P as a context. The Middleware then calls the 34@D4D4< 8callback and dequeues from tcbcast . First the message is sent to be broadcast with the broadcast callback. Then the peer’s entry on +8is updated with the cntr of the message’s dot. After this the message’s predecessors that are not stable are calculated. As shown in the stable function of the algorithm, a message is stable if it has been delivered (its sender’s entry on +8is equal or less than the cntr from its dot) and either its stage on the graph is STB or it has already been removed from the graph. The message’s dot is then added to the succ field of its predecessors that are not stable. After that the message with stage DLV , with the i bit as 0and all the others as 1is added to ⌧8, calling in the end D?30C4BC018;8C~8. If the Middleware receives a message from a peer in the group A4248E48is called. A message can only be considered as received if the cntr field on its dot is greater than the sender’s entry on +8and if its stage on the DAG is not received. With this only messages that are not in ⌧8or have a placeholder node with stage SLT can be added to the graph. Then this function calculates from the predecessors P of the message the ones that are not stable. The received message’s dot is added to its predecessors’ succ field and if a predecessor has not yet been received then a node with stage SLT is added to the graph as a placeholder. A bit string with entry 0as a delivered predecessor (or not a predecessor at all) from that peer or 1as a predecessor that has not been delivered is created. In the algorithm, (is the list of the successors’ dots of the received message. In case the message’s node previously had stage SLT then the list remains the same as the succ field, 71 CHAPTER 6. DYNAMIC MEMBERSHIP membership will eventually change and all the existing nodes need to know about the new membership. In addition to that, the joining node needs to have a correct and up-to-date state before starting to send and receive messages for the existing peers. One way to do that is to block the system until the new node joins, receives the the updates needed to have a correct state and inform every existing node of the new node. This might not seem costly if joins are rare. However, in dynamic systems, many nodes might be joining or leaving continuously which makes blocking the system very costly. Causal Stability allows new nodes to join without blocking the existing nodes from continuing their distributed computation or sending and receiving requests. Every new node that wants to join the group send a request to an existing node in the group membership along with its unique ID. The contacted node handles this request and is responsible of making sure that the new node becomes a member of the group safely. Upon receiving the request, the contacted node assigns a dot (of his own) to the join request, its context containing the causal dependencies and adds it to its local causal DAG as delivered. It also updates its membership locally to include the new node. Next, it collects the application data needed as well as causality related meta data that causally precedes the join request and sends it back to the new joining node to initialize its state. The new node initializes its state and spawns a middleware process. Until this point the new node cannot safely send or broadcast any messages to the group, and it only receives messages from the contacted node. Now the contacted node broadcasts the join request as any normal operation to every node in the membership. It also keeps forwarding for the new node, all the previous messages it has received and that were concurrent to the join request. It keeps forwarding so the new node can receive all the messages and stay up-to-date. When the other nodes deliver the join request, they update their membership to include the new node which will starts receiving the broadcasts. Until this point, the new node only receives the messages but cannot send or broadcast yet. When the join request is stable at the contacted node, the latter knows that every node in the group membership has delivered the join request and therefore all the new broadcasts will also be received by the new node. Thus, there is no need to forward messages anymore. Only then the new node is included in the membership at all nodes, and receives every message. It also can safely start to send or broadcast messages yet. Causal stability allows new nodes to safely join the group without blocking the system or breaking causal consistency. 6.1.2 Causal Stability and Leave Requests The logic is similar when an existing node decides to leave the group membership, with a few changes. When a node of the group decided to leave the group, it broadcasts to all membership a leave request. The leave request is also added to the local DAG at every node as normal message. Every node that delivers the leave request stops sending messages to the leaving node. The leave request is also added and delivered locally at the leaving node. The leaving node cannot yet leave safely. When the leave request is causally stable at the leaving node, the latter knows that all the other nodes have delivered the leave request and can safely exit the group membership. As for the other nodes, each of them can only delete 78 6.2. ALGORITHM the meta data related to the to the leaving node and from their membership when the leave request is causally stable for them locally. The reason is that until the leave request is causally stable locally, a node can still receive messages concurrent to the leave request, and that require the meta data including the leaving node. Causal stability allows existing nodes to safely leave the group without blocking the system or breaking causal consistency. 6.2 Algorithm In this section, we present the dynamic version of our Tagged Causal Delivery middleware in Algorithm 12 for the client process, and Algorithm 13 for the middleware process. We start by extending the latest algorithms presented previously in Algorithms 10 and 11. 6.2.1 System Startup At the client process, we add a boolean called B4=3BC0CDB8.IfB4=3BC0CDB8is set to 1, the node can broadcast messages i+#+bi8(). Otherwise, it can only receive messages as it is temporarily the case for new joining nodes. In fact, the system can be bootstrapped with a number of nodes in the same group membership. Those nodes are initialized using BMBi8(()where ((=active). They can send and receive messages to every node in the group and have a middleware process that is responsible of managing causal meta data such as the causal DAG and ensuring Tagged Causal Delivery. As for new joining nodes, they are initialized with ((=inactive)and can only spawn a middleware process (line 42 in Algorithm 12) and start broadcasting messages (line 34 in Algorithm 12) when the join request is stable at the contacted node. At the middleware process, 8, a list of joining nodes is added to the state. When contacted by a new new joining node, the contacted node 8stores the former’s ID in 8. It is used by contacted nodes to keep temporarily forwarding messages to joining nodes (lines 25, 54-55, 103-104 in Algorithm 13). Also, BMBi8(⌫,⌧,+,)takes 4 arguments. This is needed for when new joining nodes spawn a middleware process and need to initialize it with meta data transfered by the contacted node (line 19 in Algorithm 13 and 40, 42 in Algorithm 12). 79 CHAPTER 6. DYNAMIC MEMBERSHIP G:P_Ah>J Rk, Dynamic tcb client process for node 82I Rstate: k382D j2CGC82C 9B4=3BC0CDB82{0,1} 8proc BMBi8(() e38:=(8,0) d2CGC8:=; 3B4=3BC0CDB8=0 Nif ((=active)then Ry C={{87! 1},;,{97! 0|92I},[]} RR bTrM(Middleware,C)B4=3BC0CDB8=1 Rk proc i+#+bi8(<B6) Rj if B4=3BC0CDB8then R9 i+/2HBp2`8(38,2CGC8,<B6) R8 2M[m2m22< 8(msg,38,2CGC8,<B6) Re 2CGC8:=38 Rd 38:=(8,38.1+1) R3 proc bi#S?MiQK8() RN 2M[m2m22< 8(phantom,38,2CGC8) ky proc H2p28() kR 2M[m2m22< 8(leave,38,2CGC8) kk 2CGC8:=38 kj 38:=(8,38.1+1) k9 on /2[m2m2<2 8(deliver,3,C~?4,2CGC,<B6) k8 2CGC8:={3}[2CGC8\2CGC ke i+/2HBp2`8(3,2CGC,<B6) kd on /2[m2m2<2 8(stable,3,C~?4,2CGC,<B6) k3 i+bi#H28(3,2CGC) kN 2M[m2m22< 8(stable,3) jy if <B6 =8then jR if C~?4 =leave then jk 2tBi8() jj else if C~?4 =join then j9 B4=3BC0CDB8=1 Uj8V on /2[m2m2<2 8(join,D83) UjeV ⇡0C0 :=64C⇡0C0() UjdV 2M[m2m22< 8(join,38,2CGC8,D83,⇡0C0) Uj3V 2CGC8:=38 UjNV 38:=(8,38.1+1) U9yV on `2+2Bp28(joinInit,⌫,⌧,+,,⇡) U9RV B4C⇡0C0(⇡) U9kV bTrM(Middleware,{⌫,⌧,+,}) U9jV fn i+/2HBp2`8(3,2CGC,<B6) f /2HBp2`v +HH#+F f U99V fn i+bi#H28(3,2CGC) f bi#BHBxiBQM +HH#+F f 80 6.2. ALGORITHM G:P_Ah>J Rj, Dynamic tcb middleware process for node 82I URV state: UkV #Bi8:Iõ!N, bit mask given node id UjV ⌧8:Dõ!M, message graph U9V +8:Iõ!N, delivered version vector U8V 8:I⇤, list of ids UeV on BMBi8(⌫,⌧,+,) UdV #Bi8:=⌫ U3V ⌧8:=⌧ UNV +8:=+ URyV 8:= URRV on /2[m2m22< 8(hmsg,3,%,<i) URkV HQ+H8(hmsg,3,%,<i) URjV on /2[m2m22< 8(hstable,3i) UR9V /2H2i2bi#H28(3) UR8V on /2[m2m22< 8(hjoin,3,%,9,⇡0C0i) UReV HQ+H8(hjoin,3,%,9i) URdV ++ =;2M2`i2pp8(3) UR3V {?0BC,2>=2}=7BHi2`8(3) URNV B4=389(joinInit,#Bi8,?0BC,++,[],⇡0C0) UkyV for :2/QK(#Bi8),:<9do UkRV B4=38: (h9>8=,3,%,9i) UkkV for {30in 2>=2|⌧8[30].bi;2 <slt}do UkjV <=⌧8[30].Kb; Uk9V B4=389(h⌧8[30].ivT2,30,⌧8[30].T`2/,<i) Uk8V 8:=8+[9] UkeV on /2[m2m22< 8(hleave,3,%i) UkdV HQ+H8(hmsg,3,%,3.0i) Uk3V proc HQ+H8(hC~?4,3,%,<i) UkNV if C~?4 <join then UjyV #`Q/+bi8(hC~?4,3,%,<i) UjRV +8[3.0]:=3.1 UjkV %0={?2%|¬bi#H28(?)} UjjV for ?in %0do Uj9V ⌧8[?].bm++ :=⌧8[?].bm++ [{3} Uj8V ⌧8[3]:={bi;2 :dlv,#Bib :⇠0,ivT2 :C~?4 UjeV T`2/ :%0,bm++ :;,Kb; :<} UjdV 1=⇠#Bi8(8) Uj3V if C~?4 =join then UjNV mT/i2K2K#2`b?BT8(C~?4,9) U9yV 1=⇠(#Bi8(8)|#Bi8(<)) U9RV mT/i2bi#BHBiv8(1,3) U9kV on `2+2Bp28(hC~?4,3,%,<i) U9jV if +8[3.0]<3.1^¬(⌧8[3].bi;2 =rcv)then U99V %0={?2%|¬bi#H28(?)} U98V for ?in %0do U9eV if ?8/QK(⌧8)then U9dV ⌧8[?]:={bi;2 :slt,bm++ :;} U93V ⌧8[?].bm++ :=⌧8[?].bm++ [{3} U9NV 1=Õ{#Bi8(?.0)|?2%0^⌧8[?]<dlv} U8yV (=if 32/QK(⌧8)then ⌧8[3].bm++ U8RV else ; U8kV ⌧8[3]:={bi;2 :rcv,#Bib :1,ivT2 :C~?4 U8jV T`2/ :%0,bm++ :(,Kb; :<} U89V for 928do U88V b2M/89(hC~?4,3,%,<i) U8eV if 1=0then U8dV /2HBp2`8(3) U83V +?2+Fbi#BHBiv8(3,%0) U8NV on `2+2Bp28(hphantom,3,%i) UeyV +?2+Fbi#BHBiv8(3,%) UeRV on `2+2Bp28(hjoin,9i) UekV if 98/QK(#Bi8)then UejV 2M[m2m2<2 8(hjoin,9i) Ue9V fn bi#H28(3) Ue8V 3.1+8[3.0]^(38/QK(⌧8)_⌧8[3].bi;2 =stb) UeeV proc /2HBp2`8(3) UedV C~?4 =⌧8[3].ivT2 Ue3V 2M[m2m2<2 8(hdeliver,3,C~?4,⌧8[3].Kb;i) UeNV (9,=)=3 UdyV +8[9]:== UdRV ⌧8[3]:=⌧8[3]{bi;2 :dlv, UdkV #Bib :⇠#Bi8(8)} UdjV 1=⇠#Bi8(9) Ud9V if C~?4 =join then Ud8V <=⌧8[3].Kb; UdeV mT/i2K2K#2`b?BT8(C~?4,<) UddV 1=⇠(#Bi8(9)|#Bi8(<)) Ud3V mT/i2bi#BHBiv8(1,3) UdNV for Bin ⌧8[3].bm++ do U3yV ⌧8[B].#Bib :=⌧8[B].#Bib &⇠#Bi8(9) U3RV if ⌧8[B].#Bib =0then U3kV /2HBp2`8(B) 81 CHAPTER 6. DYNAMIC MEMBERSHIP U3jV proc +?2+Fbi#BHBiv8(3,%) U39V if 3.1=1++8[3.0]then U38V for ?2%do U3eV mT/i2bi#BHBiv8(⇠#Bi8(3.0),?) U3dV proc mT/i2bi#BHBiv8(1,3) U33V if ⌧8[3].bi;2 =dlv then U3NV 10=⌧8[3].#Bib &1 UNyV if 10<⌧8[3].#Bib then UNRV ⌧8[3].#Bib :=10 UNkV if 10=0then UNjV bi#BHBx28(3) UN9V else UN8V for ?in ⌧8[3].T`2/ do UNeV mT/i2bi#BHBiv8(1,?) UNdV proc bi#BHBx28(3) UN3V for ?in ⌧8[3].T`2/ do UNNV if ⌧8[?].bi;2 <stb then URyyV bi#BHBx28(?) URyRV ⌧8[3].bi;2 :=stb URykV C~?4 =⌧8[3].ivT2 URyjV if C~?4 =join ^3.0=8then URy9V 8:=iBH(8) URy8V 2M[m2m2<2 8(hstable,3,C~?4,⌧8[3].Kb;i) URyeV proc /2H2i2bi#H28(3) URydV for Bin ⌧8[3].bm++ do URy3V ⌧8[B].T`2/ :=⌧8[B].T`2/ \{3} URyNV ⌧8:=3C⌧8 URRyV if ⌧8[3].ivT2 =leave ^3.0<8then URRRV mT/i2K2K#2`b?BT8(;40E4,3.0) URRkV proc mT/i2K2K#2`b?BT8(C~?4,9) URRjV =if C~?4 =join then URR9V bQ`i(/QK(#Bi8)[{9}) URR8V else if C~?4 =leave then URReV bQ`i(/QK(#Bi8)\{9}) URRdV #Bi8:={:!2TQb(:,)|:2} URR3V for 32⌧8|⌧8[3].bi;2 2{rcv,dlv}do URRNV 5=if ⌧8[3].bi;2 =rcv then 0else 1 URkyV ⌧8[3].#Bib :=b?B7i8(C~?4,⌧8[3].#Bib,2TQb(9,),5) URkRV fn b?B7i8(C~?4,1,?,5) URkkV if C~?4 =join then URkjV ((((1?)⌧1)|5)⌧(?)) | (1&(2?1)) URk9V else if C~?4 =leave then URk8V ((1(?+1)) ⌧?)|(1&(2?1)) 82 6.2. ALGORITHM 6.2.2 Joining Nodes Client Middleware Client Middleware Client Middleware New Node join request Client Middleware Client Middleware Client Middleware New Node 1 2 •prepare initial data transfer" •add join to causal DAG" •locally update membership transfer initial data 3 send join message spawn middleware process 4A 4B 5A peers add join to causal DAG forward concurrent messages to new node 5B 5B Client Middleware Client Middleware Client Middleware Client forward messages concurrent and in the future of the join request to new node 6A Middleware •join request delivered at peer" •peer updates membership" •peer starts sending to new node 6B Client Middleware Client Middleware Client Middleware Client Middleware the joining node receives from every peer that delivers the join request 7 •join request is stable at contacted node" •forwarding messages stops 8 •join request is stable at new joining node" •new node is part of the group membership 9 Figure 25: Evolution of a new node joining process To join the group, a new node sends a join request and its unique ID to an existing node. The contacted node receives the join request (line 61 in Algorithm 13). It checks if the new node is part of the membership and if not enqueues the request to its client process (line 62, 63). The client process (line 35) receiving the join request, prepares the application data/state to be transfered (line 36), enqueues a new request to the middleware process with a dot, a context and the application state (line 37), and finally updates its local dot and context (lines 38-39). The middleware process (line 15) dequeue the request. It starts by adding the the join request to its local causal DAG (line 16). In HQ+H8the dot representing the join request is added locally as delivered at the contacted node (lines 28-41) as any other message. Notice, however, that for join requests, the contacted node updates its local membership to include the new node (lines 38-40). At line 18, the contacted node filters all causality meta data in the causal DAG in two parts: ?0BC that includes all dots 83 CHAPTER 6. DYNAMIC MEMBERSHIP causally preceding the join request, and 2>=2 that includes the dots concurrent to it. Now the contacted node can transfer the initial state to joining node (line 19). It sends the join request to the membership (lines 20-21), and forwards the concurrent messages to the joining node (lines 22-24). When the middleware process at other nodes (excluding the contacted node) deliver the join request, they also update their group membership to include the new node (lines 74-77). The new node now can start receiving messages from them too. When the join request is causally stable at the contacted node 8(line 103), the latter can remove the new node from 8(line 104), and stops forwarding messages received from peers to the new node (as it used to do in lines 54-55). The initial state transfered to the new node was the same as the contacted node. Then all messages received by the contacted node were also forwarded to the joining node. This means that whenever a message is delivered or becomes stable at the contacted node, it also becomes stable at the joining node. When the join request becomes stable at the joining node (same time as at the contacted node), the middleware process informs the client process. The client process checks that the join request (line 33) is causally stable locally (line 30), and sets its B4=3BC0CDB8to 1 (line 34). Now the new node is part of the new group membership, and can safely send and receive messages to the group. 84 6.2. ALGORITHM 6.2.3 Leaving Nodes Client Middleware Client Middleware Client Middleware Client Middleware broadcast leave request to group 2 •prepare leave request" •add leave request to causal DAG 1 Client Middleware Client Middleware Client Middleware Client Middleware peers add received leave request to causal DAG peers add received leave request to causal DAG 3A 3B 3C peers deliver leave request 4A 4B 4C peers deliver leave request Client Middleware Client Middleware Client Middleware Client Middleware 5A 5B 5C leave request is locally stable at peer leave request is locally stable at peer leave request is stable at leaving node 5D Client Middleware Client Middleware Client Middleware Client Middleware 6A 6B 6C" leaving node removed from membership leaving node removed from membership node leaves group 6D Figure 26: Evolution of an existing node leaving process To leave the group, an existing node prepares a leave request, tagged with a dot and context and enqueues it to the middleware process (lines 20-23 in Algorithm 12). When the leave request is dequeued at the middleware process (lines 26-27), the latter creates a dot for it in its causal DAG as delivered (lines 28-41). The leave request is also broadcast to the group (lines 29-30). At every node, the leave request is added to the local causal DAG as any other message, waiting to be delivered, stable and them removed. For each of these nodes, when the leave request is stabilized and deleted from the causal DAG, the middleware process updates it membership by removing meta data related to the leaving node (lines 110-111). 85 CHAPTER 6. DYNAMIC MEMBERSHIP As for the leaving node, it can leave the group when the leave request is locally stable (lines 30-32 in Algorithm 12). The leaving node does not have to wait more than that to leave. It does not need to receive any messages form its peers or send any messages later than the leave request. The other peers do not expect to receive any messages from a leaving node. However, they cannot remove meta data related to the leaving node, as it could affect the causal stability computation related to messages concurrent to the leave request. This is the reason the other nodes have to wait for the leave request to become causally stable before the membership update. 6.2.4 Updating Group Membership In this section we explain how updating the meta data such the causal DAG and the bitmask #Bi8,toreflect the changes in group membership. We focus on the procedure mT/i2K2K#2`b?BT8() (lines 112-120) and the function b?B7i8() (lines 121-125) in Algorithm 13. mT/i2K2K#2`b?BT8() takes two arguments, C~?4 and 9.C~?4 has for value join for updating the membership by adding a new node, or leave for removing an existing node from the groups membership. 9is the globally unique identifier (GUID) assigned to the corresponding node. We use the IP address and port to generate a globally unique ID, but any mechanism that generates GUIDs can be used instead. Those IDs are also comparable. The bitmask #Bi8, as explained earlier in previous chapters, is map from GUIDs to an integer (a power of two). The GUIDs are sorted in lexicographic order. The value associated to a GUID in the bitmask is 2?>B,where?>B is the position of the GUID in lexicographic order. For instance, three IDs,⌫and⇠will have the values 20=1,21=2,and22=4respectively. Notice that as the values are powers of two integers, they represent unique bits in a bitstring (001,010,100). This is very useful for representing causal delivery or causal stability for every dot. The bitstring is a very memory efficient representation to mark the set of nodes that have causally delivered or stabilized a dot. In our example 1or its equivalent bitstring (001) represent the node ,2or 010 represent ⌫,6or 110 represent and ⌫, and so on. Every dot in the causal DAG stores a bitstring, which is used to represent/compute causal delivery or causal stability. Therefore, when the group membership changes, the bitmask #Bi8needs to be updated as well as every bitstring in the causal DAG. For adding new nodes, the GUIDs represented by /QK(#Bi8)need to be sorted again including the GUID 9of the joining node (lines 113-114). Similarly for removing nodes, /QK(#Bi8)is sorted excluding the existing GUID 9of the leaving node (lines 115-116). Then, every bitstring in the causal DAG is shifted accordingly (lines 118-120). b?B7i8() is responsible of updating every bitstring by shifting the bits and adding a new bit for the joining node without affecting the values of the existing bits (lines 122-123), or to remove the bit corresponding to the leaving node (lines 124-125). 86 d Causality Checker Due to the complexity of the algorithms so far, we developed our own verification tool that we called Causality Checker. This Causality Checker is a standalone service that verifies that the different versions of our Tagged Causal Delivery Algorithm, ensures causal delivery and causal stability. The Causality Checker also gives insight to the developer to what might be the problem when the verification fails by printing the failed step and the state of the corresponding nodes. It makes it easier to for developers and implementors to add changes and updates to their algorithms, test them, debug them and finally verify their correctness. The tool works by reading the serialized events logged by the middleware in test mode. The serialized events are a sequence of Send (broadcast), Deliver and Stable events from every peer in the group membership. The order of events in a sequence follow the same order they were logged in by the peers in runtime. Every sequence corresponds for the serialized events at one peer. Every entry in a sequence is a dot and its meta data (3,E0;).ThemetadataE0; stores the type of event (3.BC064)whichtakesas value (⇢#⇡,⇡!+ or () ⌫ for send, deliver and stable events respectively. The causality checker verifies if these dot sequences respect the following rules: • A Send with a dot only appears in the sender’s sequence • A Deliver with a dot appears in every peer sequence except the sender’s • A Stable with a dot appears in both the sender and the receiving peer sequences The logged sequences are passed to the causality checker which implements a backtracking algorithm designed to verify causal order by looping through them. The causal history of each dot is built while iterating over these sequences. Version vectors are used to represent causality. Each dot is associated with a version vector that represents its causal dependencies. A version vector is associated for every peer to represent causal delivery. While looping through the sequences, the peer version vectors are updated and the dot version vectors are computed accordingly. A causal delivery check is a comparison between the version vectors of a peer and the dot being delivered in the sequence. A causal stability check is a comparison between the version vector associated with the dot being stabilized and the peer’s causal 87 CHAPTER 8. EVALUATION To use a Poisson distribution, an average time between events, which in this case are broadcasts, is needed. According to [82], to determine these intervals, a random value U between 0and 1from a uniform distribution has to be generated. This number is a value from the y-axis and to locate the corresponding time value on the x-axis, the inverse of the exponential function is used: )=;=* _ The _is the average rate of events, and because the average time interval is passed through the configuration file, it has to be transformed into the equivalent rate. Since the period is the reciprocal of the frequency, _can be calculated for an average interval A by using _=1 . Using a Poisson distribution can sometimes return samples much higher than the desired average. To avoid these, the calculated value from this distribution is truncated at a maximum of 4times the average from the configuration file. 8.1.2 Network latency The reason for simulating network latency is to add more control over the experiments executed when simulating the delay a message has when travelling over a network. The network latency can either be a fixed value or a fixed value plus another component calculated through a Weibull distribution. The network latency is simulated by adding to the sender’s clock a value in microseconds calculated through the network latency parameters. If the configuration file only has the baseline, then the network latency is a fixed value for all the links in the group. However, to specify a different fixed latency between the peers, a path to a matrix can be set through the configuration file. This is an #⇥#matrix, where N is the group size, the row is the sender and the column is the receiver. For example, the latency of the link from peer 3to peer 2is on row 3and column 2of this matrix. In this file the matrix is represented as N rows, each with N integers separated by commas. At least one of these previous parameters is mandatory. To use a baseline value plus another from a Weibull distribution, both scale and shape parameters have to be specified. A Weibull distribution is a continuous probability distribution and has two parameters: scale and shape. This is also used as a model to characterise end-to-end network delay measurements [83]. This baseline, as the previous one, is passed through the same fields. The network latency calculation using the Weibull distribution was based from [83–85]. To achieve three standard deviations from the baseline, using B⌘0?4 =2and a B20;4 =0.15 had to be used. The mean of a Weibull distribution [86] can be calculated by using the gamma function ,as_⇥(1+1 :), where _and k are, respectively, the scale and shape parameters. The latency is calculated with this distribution as follows: =⌫⇥(1+,(_,:)) 94 8.1. CONFIGURING THE EXPERIMENTS A and B are the calculated latency and baseline value, whereas,(_,:)is a sample from the Weibull distribution. Using the previous expression to calculate the mean of a Weibull distribution, for a scale _=0.15 and shape :=2, this is ⇡0.1329. By using the preceding expression we have: =1.1329 ⇥⌫ With this we can calculate the needed baseline value for a given average latency. For example, if we wanted an average network latency A of 5seconds, the baseline value B to use together with the Weibull distribution to achieve it would be 4.4135. However, sometimes the calculated delay might be very large which in turn could affect the results. Therefore, the calculated value is truncated at a maximum given by ⌫⇥(1+3⇥_). 8.1.3 Simulating slow links Regarding the network latency between peers, the experiments can be divided into symmetric and asymmetric network topologies. In the first, every link between the peers has the same latency. For the asymmetric scenario there’s a link with a higher network latency that further delays the arrival of messages. For the experiments the slow link is always between peer 0and 1. This slow link scenario aims to test the middleware’s robustness to transient network partitions. However, for the experiments a fixed value is used for the delay in the slow link, as to avoid having large fluctuations in the calculated latency and have more control over the buffered messages by the peers between runs. Network partitions occur when groups of nodes have difficulties communicating with other nodes outside of their group. This happens either because of higher latencies or unavailable peers. These partitions affect information availability because messages take longer to arrive at their destination, and other messages that depend on them cannot be delivered and have to be buffered until then. This causes a buildup of received messages waiting to be delivered at the middleware. Imagine a group with 5peers on a full mesh topology, where everyone has a network latency of 1second between them, except the link between 0and 1that has a latency of 20 seconds as show on Figure 27. Peer 0 Peer 1 Peer 2Peer 3 Peer 4 1 s 1 s 1 s 1 s 1 s 1 s 1 s 20 s 1 s 1 s Figure 27: Full mesh topology with a slow link between 0and 1. 95 CHAPTER 8. EVALUATION Messages are broadcast from the peers that have as dependencies messages from other peers. Since they are part of the slow link, messages between 0and 1take a longer time to arrive at the their destination. At peer 0or 1, messages with a dependency from 1or 0have to wait because it takes longer for that dependency to arrive and be delivered. 8.2 Deploying environment This section details the on deployment of the experiments. First, it explains what type of architecture was used for the deployment in order to simulate a network with multiple peers interacting with each other. After this, an introduction to Docker, some of its advantages and how it fits on the experiments will be given. Then the last part of deploying the experiments with Kubernetes will also be explained. 8.2.1 Architecture For the deployment architecture, an Erlang program was written so that the middleware is used for broadcasting messages. Through configurations parameters this program spawns multiples nodes and each of these is either a master or a worker. Each deployment only has one master node but there can be multiple workers. A worker node locally spawns peers that broadcast messages with peers spawned by other worker nodes (depending on the configuration). The master is the first node to be deployed by Kubernetes and after it the workers start. After it finishes setting up, it waits for all the workers to connect to it. The purpose of the master is to receive information from the workers, namely port numbers and IP addresses of the peers they spawn and, after the run has ended, their logged information. Upon receiving all the necessary information to establish connections between the peers in the nodes, the master processes these addresses and sends them over to the workers, so they can have a view of the network layout. Furthermore, the master also assigns a globally unique id to each worker node and to the peers they spawn. With this information the peers spawned by the workers can start their middlewares, connect to each other and begin sending and delivering messages. In this scenario, the worker nodes run a local logging service where their peers log the events to. The difference from before is that when they finish broadcasting, the causal check cannot be executed locally, since only a subset of all logs is available. To overcome this, when the peers stop executing, their worker node sends the logged information to the master. Only after this will the master have a complete view of the logging events at each peer and the post-processor is then executed, checking also the delivery and stability order of messages. From this results the JSON file with the information collected during the experiments from the peers at every node. 96 8.3. COMPARING EXPERIMENTS 8.2.2 Docker Docker [87] is a tool that allows packaging applications in containers , including their dependencies, by using OS-level virtualization. These containers run applications isolated from the host system they’re on and are designed to make it easy and fast to provide a consistent experience as software is replicated and moved between environments. Containers are similar to virtual machines, but instead of creating a complete virtual operating system, they only replicate the components configured to them which leads to an important boost on performance. A Docker image can be shared by different operating systems and the software inside of it can also be executed, as long as the host has Docker installed. For the deployment of the experiences, a Docker image was created from a container that has Erlang installed and the software for spawning the peers. Since images can easily be shared, the master and worker nodes both result from them. The master and worker nodes execute this image and because an Erlang environment was configured in the container, there is no need for them to have it installed. 8.2.3 Kubernetes The other component for deploying the experiments is Kubernetes [88], which is a tool for automating software deployment. Kubernetes provides a framework for running code on clusters and also manage its execution. From running a Docker image on Kubernetes, a pod is created. Kubernetes was used to deploy the broadcast experiment image on a cluster hosted on Emulab [89] and the pods will be the master and worker nodes. The Emulab nodes where Kubernetes deploys the images runs on different machines and these nodes have a delay between them of around 1millisecond. Because the clocks are not fully synchronized, the latencies used during the experiments were high enough so that a 1millisecond difference would not affect the results. For every experiment the same type of node with the same hardware was used. The deployments ran on machines with 8CPU cores. Because there were limitations on the number of available nodes, each pod was limited to only using 4cores. Furthermore, each pod spawned a single local peer and by using this configuration, by having 64 Emulab nodes it was possible to run experiments with groups of up to 128 peers. 8.3 Comparing experiments Non-causal latency is used to compare the results of the experiments. Any causal delivery mechanism has an algorithm induced message delay. For example, this happens when a message is received by a process before all its predecessors have also been delivered. So that the algorithm can maintain causal delivery order of messages, this delay is necessary and is known as causal latency [90]. Furthermore, 97 CHAPTER 8. EVALUATION this latency is not the algorithm’s fault because outside factors can delay the receiving of a message, e.g., network latency. Some algorithms will delay delivery without knowledge that there is a causally prior message to deliver that has not yet been received. Rather the algorithm delays delivery based on lack of knowledge that such message does not exist. That is, the delay on the delivery of messages even though all their causal predecessors have already been delivered. This is known as non-causal latency [90]. The optimum causal delivery algorithm would have no non-causal latency. During the experiments, when a received message can be immediately delivered it might trigger other deliveries. The time between a receive and the triggered delivery of a message is the non-causal latency of this message. Therefore, for both middleware implementations, when this occurs, the elapsed time between the receive of the last causal predecessor and the delivery of its successors is measured and sent to the log service. The lower this non-causal latency is the lower the delay induced in delivering messages due to the algorithm. The time for the messages to become causally stable is another metric to compare both approaches. Like for non-causal latency, the time, after the receive that made it possible, that it takes for a message to be deemed stable is another useful measurement. This stability time for the messages at every peer is also sent to the logging service. 8.3.1 Memory Metadata The memory allocated during the experiments is calculated by counting the words required for the algorithm being used by the middleware (either graph based and version vectors). A word is a fixed-sized piece of data and also serves as a unit. A word is 8 Bytes for 64-bit Erlang, and this is what we used for measuring the memory metadata. Since both middleware implementations require different data structures, the method for counting their allocated words differ. The purpose of this metric is to have an idea and compare the memory required by the two algorithms. Further below are the methods used for counting the words in each middleware version. It is noteworthy to mention that for this metric the size of the message’s payload was ignored and only considered the necessary words for tracking causality. 8.3.1.1 Graph based For the graph based implementation each message is tagged with a pair ( id , ctr )called dot ,where id is the sender’s globally unique identifier and ctr is a monotonically increasing counter. Besides this, each message also has a list context of dots that represent its immediate causal predecessors. For the graph based algorithm, it was considered that the id and ctr each represent one word. Therefore, each dot is two words. In this approach, a directed acyclic graph is used for tracking the causal dependency between messages. Each message is a node in this graph, all outgoing edges of this node are successors of that message and every incoming edges are predecessors. In Erlang, we used a map to represent the graph, 98 8.3. COMPARING EXPERIMENTS where each entry is a dot, mapped to a data structure representing the causality information. Namely, a set of dots for the predecessors, a set of dots for the successors, a bitstring for causal stability and an enum for the stage to mark if it is received, delivered, etc. The number of words in this middleware implementations is the result of the following sum for every node in the graph: •2words for the dot •2words for each predecessor •2words for each successor •1word for the stage •1word for the bitstring 8.3.1.2 Version vectors In the version vector implementation of the middleware, each message has a vector with same length as the number of peers in the network. Each position is considered to be one word. Therefore, a version vector of length N requires N words. Just like in the previous graph based approach, the size for the message’s payload is ignored. Each middleware has two version vectors to track, respectively, the received and delivered messages: R and V . Therefore, if N is the number of peers in the network, these version vectors require 2#words in total. Furthermore, there is the matrix that tracks causal stability. In this matrix, row8is the version vector of the last message sent by peer8. In this case, the number of words for this matrix is #2. This implementation approach also has a queue DQ for received messages that could not be delivered because of causal dependencies. Each element in this queue is a message that has a payload, version vector and the sender’s globally unique identifier id . Since the size of the payload is ignored, only the version vector and id are taken into account. As already mentioned, a version vector requires N words, considering a system of N peers. On the other hand, the id is only an integer. Therefore, it is considered that this id is one word. The last data structure in this implementation were counting words is required to measure the size is the stability map SMap . This object maps a pair of ids (the sender’s and message’s) to a stable message. The pair of ids used here is similar to the dot from the graph based approach. Therefore, it is considered that it requires two words. The stable message has the message’s payload and version vector. Since the payload is ignored, only the N words from the version vector are considered. For the SMap with K elements and N peers, the word number is (2+#). In summary, for the version vector middleware implementation, the number of words, for a group with N peers and K stable messages, is the sum of the following calculations: 99 CHAPTER 8. EVALUATION • N words from the received messages version vector • N words from the delivered messages version vector •#2words from the stability version matrix • (2+#)words from the stable messages map 8.4 Broadcast Experiments This section focuses on comparing results of the experiments using a graph based and version vector approach to deliver messages in a causal order and determine causal stability. Between runs either the number of peers in the group or send interval of peers are updated in order to further demonstrate the behaviour of the middleware implementation. Each subsection details the configurations used while running the experiment. 8.4.1 Experiment1: Classical end-to-end vector-based VV vs our end-to-end graph-based TCB For this experiment the peers are in a full mesh topology where all the links have the same network latency. The purpose of this is to see how the middleware implementations behave when we modify send intervals and gradually increase the group size. The effect on performance due to calculating causal stability is also measured by doing runs with and without having to calculate it. The following experiment was done with the same send interval for every peer and the interval between broadcasts is calculated following a Poisson distribution, and each peer sends 100 messages. The send intervals chosen for the peers in this experiment were, in milliseconds: 10,100 and 1000. Moreover, the network latency is calculated via the Weibull distribution, as to make the peers more desynchronized from each other. Runs were executed for groups with 4,8,16,32,64 and 128 peers. The network latency chosen for the links in the runs of this experiments was 10 milliseconds. In this experiment we compare a version vector based approach where the application uses an additional version vector to be able to provide an end-to-end happens-before against our E2E TCB approach. We call it classical E2E VV. In this version, we use a classical VV-based causal delivery algorithm using version vectors and a delivery queue for causal delivery. Plus, the application uses an additional version vector to provide an end-to-end happens-before.; This is what we called previously a duplication of effort. Our aim in this experiment is to show the overhead in terms of transmission on the network and the amount of causality metadata stored in memory between both approaches. In Figure 28a, we can clearly see that the number of causal predecessors considered as dependencies in the graph approach is smaller than the size of version vectors needed in the classical E2E VV-based approach. At 128 nodes, the number of predecessors needed is at worst 80 for the E2E TCB approach 100 8.4. BROADCAST EXPERIMENTS ●●● ● ● ● 4 8 16 32 64 128 0 50 100 150 200 250 Network Metadata (in entries) Number of Nodes ●E2E TCB SI:10ms E2E TCB SI:100ms E2E TCB SI:1000ms Classical E2E VV SI:10ms Classical E2E VV SI:100ms Classical E2E VV SI:1000ms (a) Median transmission entries. ●●●●●● 4 8 16 32 64 128 0 100000 200000 300000 400000 500000 Memory Metadata (Bytes) Number of Nodes ●E2E TCB SI:10ms E2E TCB SI:100ms E2E TCB SI:1000ms Classical E2E VV SI:10ms Classical E2E VV SI:100ms Classical E2E VV SI:1000ms (b) Median memory metadata. Figure 28: Experiment 1: transmission and memory, without stability. to provide end-to-end happens-before. However, in the classical E2E VV-based approach, we need twice the size of nodes which is 256 entries. The size of those vectors also affects the size of metadata needed to be stored locally to ensure causal delivery (and stability as we show later). In Figure 28b, we only show the causality metadata stored in memory when stability is not being calculated. Meaning, less data structures are used to store causality metadata. Those data structures will use the dependencies in case of the E2E TCB approach, and the version vectors for the classical E2E VV approach. Therefore, the size of metadata stored in memory will be affected as we see in Figure 28b. In the case of the vector-based approach the data stored can get to more than 500KB for 128 nodes, where it is negligible for the E2E TCB approach. When stability is calculated, we notice in Figure 29a that the memory size grows a lot for the classical E2E VV approach and reaches more than 40MB for 128 peers. This is due to the larger size of version vectors for this experiment, the additional data structures used for stability such as the stability matrix and the stable map. The size of metadata in memory for the E2E TCB approach also increases compared to the experiment without stability in Figure 28b, but remains significantly smaller than the classical E2E VV approach. In Figure 29b we can see that the E2E TCB approach takes significantly less time (2-3 milliseconds for 128 peers) for non causal stability compared to the classical E2E VV approach (above 10 milliseconds for 128 peers). 8.4.2 Experiment2: Classical vector-based VV vs graph-based For this experiment the peers are in a full mesh topology where all the links have the same network latency. The purpose of this is to see how the middleware implementations behave when we modify send interval 101 CHAPTER 8. EVALUATION ●●●●●● 4 8 16 32 64 128 0e+00 1e+07 2e+07 3e+07 4e+07 Memory Metadata (Bytes) Number of Nodes ●E2E TCB SI:10ms E2E TCB SI:100ms E2E TCB SI:1000ms Classical E2E VV SI:10ms Classical E2E VV SI:100ms Classical E2E VV SI:1000ms (a) Median memory metadata. ●●● ● ● ● 4 8 16 32 64 128 0 1000 2000 3000 4000 5000 6000 7000 8000 9000 10000 11000 12000 13000 Non−causal Stability Latency (microseconds) Number of Nodes ●E2E TCB SI:10ms E2E TCB SI:100ms E2E TCB SI:1000ms Classical E2E VV SI:10ms Classical E2E VV SI:100ms Classical E2E VV SI:1000ms (b) Median Non-causal Stability latency (in microseconds). Figure 29: Experiment 1: Memory and Non-causal Stability latency, with stability. and gradually increase the group size. The effect on performance due to calculating causal stability is also measured by doing runs with and without having to calculate it. The following experiment was done with the same send interval for every peer and the interval between broadcasts is calculated following a Poisson distribution, and each peer sends 100 messages. The send intervals chosen for the links in the runs of this experiments were, in milliseconds: 10,100 and 1000. Moreover, the network latency is calculated via the Weibull distribution, as to make the peers more desynchronized from each other. Runs were executed for groups with 4,8,16,32,64 and 128 peers. The network latency chosen for the links in the runs of this experiments is 10 milliseconds. In this experiment, we compare our E2E TCB approach against a correct implementation of the end-toend vector-based approach. This way we can show the performance between an efficient implementation of the VV-based causal middleware against our graph-based implementation in terms transmission overhead on the network and memory size for causality metadata. In Figure 30a, we see similar results to the previous experiment. However, the size for the optimized E2E VV-based approach here is half of the one seen previously in the classical E2E. The E2E TCB approach provides an end-to-end happens-before with less causal dependencies transmitted on the network. At 128 nodes, the number of predecessors needed is at worst 80 for E2E TCB to provide end-to-end happensbefore. However, in Optimized E2E VV, we need twice the size of nodes which is 128 entries. As the size of version vectors is reduced to half, compared to the previous experiment, we see that reflected in the memory size. Figure 30b. In the case of the Optimzed E2E VV approach the data stored can get to more than 250KB for 128 nodes, where it is negligible for the E2E TCB approach. 102 8.4. BROADCAST EXPERIMENTS ●● ● ● ● ● 4 8 16 32 64 128 0 10 20 30 40 50 60 70 80 90 100 110 120 130 Network Metadata (in entries) Number of Nodes ●E2E TCB SI:10ms E2E TCB SI:100ms E2E TCB SI:1000ms Optimized E2E VV SI:10ms Optimized E2E VV SI:100ms Optimized E2E VV SI:1000ms (a) Median transmission entries, without stability. ●●●●●● 4 8 16 32 64 128 0 50000 100000 150000 200000 250000 Memory Metadata (Bytes) Number of Nodes ●E2E TCB SI:10ms E2E TCB SI:100ms E2E TCB SI:1000ms Optimized E2E VV SI:10ms Optimized E2E VV SI:100ms Optimized E2E VV SI:1000ms (b) Median memory metadata, without stability. Figure 30: Experiment 2: transmission and memory metadata, without stability. 8.4.3 Experiment3: Constant amount of work for vector-based VV vs graph-based Another experiment performed over a symmetric topology was adjusting the send interval to the group size so that the number of messages sent by all peers, per second, was constant. The motivation behind this is to see what would happen if a service’s workload became more and more distributed by adding peers and dividing the work between them. Having 64 peers with a send interval of 500 milliseconds generates (on average) the same amount of messages as 128 broadcasting every second. The network latency with a Weibull distribution was set to 31 milliseconds for all group sizes. For groups with 4,8,16,32,64 and 128 peers, the send intervals were, respectively, 31,63,125,250,500 and 1000 milliseconds. The same results from the previous experiment are shown in this experiment as well. Our graph-based algorithm scales better in terms of causality metadata transmitted on the network as shown in Figure 31a and in terms of causality metadata stored in memory as in Figure 31b. 8.4.4 Causal Stability In this section, we revisit experiments 2 and 3 from earlier, but we focus here on causal stability. In the previous experiments shown earlier, causal stability was not calculated. Here, we show how calculating causal stability affects the size of metadata stored in memory. We also show the non-causal stability time required by each of the two algorithm. We first start with showing for each of the two previous experiments how the metadata size in memory is affected when the algorithm are running with causal stability on. For Experiment 2, we see in Figure 32a that the memory size is half the one in Figure 28b (from 103 CHAPTER 9. USE CASES 9.1.1 Data Types for backup and recovery The possibility to guarantee Byzantine Fault Tolerant reads in the system without forfeiting system’s availability is based on the ability to provide certificates for operations by the BFT cluster. For that to work correctly we provide a mechanism that allows the system to rebuild “safe” states i.e. free of operations initiated by byzantine nodes. We present below two versions of such mechanism: a naive one using “undo” and “redo”, and a more elegant, generic, and safe alternative that we adopt, using the “clone” functionality. 9.1.1.1 Undo and Redo functionality This mechanism is made possible through the use of Undo and Redo functionality for the Data Types in use. For every instance of a data type, the received operations that mutate the state of that instance are stored in a partially-ordered log, TQHQ;. This allows updating the state of the data types at the application servers without delaying it with the byzantine fault tolerance protocol running in the BFT cluster. The BFT cluster protocol, running in parallel to the updates on the application server, provides information on which operations are safe and which ones were initiated by a byzantine node. The information provided is through the version vectors that act as a timestamp for each operation. In addition to the TQHQ; logging all the operations of an instance, each data type supports an undo ( redo respectively) functionality, that given the TQHQ;, an unsafe (safe respectively) timestamp and the state of the instance, undoes (redoes respectively) the effect of that operation from the TQHQ;. The undo and redo functionality allows to rebuild the state of a data type, from the original TQHQ;, allowing to filter unsafe operations and redoing the safe ones. However, this functionality is data type specific and requires extra care in specifying the “undo” for every new data type. We opt for a more elegant and generic approach using the “clone” functionality, that we explain next. 9.1.1.2 Data Types with clone functionality A more efficient way to provide byzantine fault tolerant state convergence would be by a using the clone functionality. This functionality is equivalent to the undo/redo functionality, but it provides a simpler, more generic and elegant approach, independently of the data type used. In this version, the state of each data type is specified as a TQHQ; mapping timestamps (version vectors) to the relative operation (as in Pure op-based CRDTs [7]). This means that the TQHQ; used “outside” the data type in undo/redo version is not needed anymore. The clone functionality receives two timestamps and the old safe state as input: a lower bound Cas the last safe version vector (bottom ?by default), and the upper bound C0as the new safe version vector calculated by the BFT cluster and B, the previously calculated safe state of the data type (by default the state is the initial empty state the data type). Instead of redoing all the operations as previously, the clone functionality “clones” the state of the current data type by copying all operations having timestamps C00 such that CC00 C0(as per the Lamport’s happens-before relation). As all 110 9.2. LASP data types in this versions have a TQHQ; as their state, the clone functionality is generic for all of them. However, we illustrate this with an example of the Set data type. ⌃=)õ!$f 0 8={} 2772+i(>,C,B)=B[{(C,>)} (with >=either [//,E]or [`Kp,E]) +HQM2(C,C0,B,B0)=B0[{(C00,>)2B|CC00 C0} 2pH(`/,B)={E|(C,[//,E]) 2B^ ö(C0,[`Kp,E]) 2B·C<C0} Figure 39: TQHQ; based Add-Wins Set with Clone functionality Add-Wins Set Figure 39 presents the specs of the Add-Wins Set with the clone functionality. The state of the AWSet is a TQHQ; mapping timestamps C2)to operations >2$={[//,E],[`Kp,E]}.2772+i updates the state by adding the timestamped operation (C,>)to the TQHQ;. Finally, +HQM2 creates a new state (TQHQ;) by copying all timestamped operations between Cand C0and merges it with the previously calculated safe state B0. As the application server guarantees causal consistency, all replicas have seen the same operations between the previously calculated lower bound and the upper bound version vector provided by the BFT cluster in the same causal order. It is trivial to note that cloning the state, through copying operations having timestamps between the two provided bounds, preserves the convergence and thus provides a byzantine consistent state among all replicas. Remark on the Counter Data Type In this approach, the state TQHQ;-based Counter increases with the number of operations (BM+,/2+). This is an overhead in terms of space when compared to an integer state. We do not address this here, but compaction techniques for a TQHQ;-based Counter has been proposed in [92] to solve this problem. 9.2 Lasp Another contribution done during my Ph.D. was participating in the Google Summer of Code’16 program where I saw an interesting opportunity to improve upon existing software and contribute to the open source community. I submitted a proposal to implement the operation-based CRDTs framework in Lasp [93], a programming model for building correct distributed applications. The proposal was accepted and the work was rewarded with a certificate from Google. The code is now public and available on GitHub (Section 9.6). I also had the opportunity to present the work in a poster session at INESC TEC, Porto for “HASLAB OPEN DAY 2016”. 111 CHAPTER 9. USE CASES Lasp Runtime System Figure 40: Lasp runtime system. Operation-based replication Pure Op-based architecture Figure 41: Lasp pure op-based CRDTs. 112 9.3. ANTIDOTEDB ⌃=N⇥P())f0=(0,{}) T`2T`2(>,(=,B)) =>(with >either BM+ or `2b2i) 2772+i(BM+,C,(=,B)) =(=,B[{C}) 2772+i(`2b2i,C,(=,B)) =(0,B\{C02B|C0<C}) bi#BHBx2(C,(=,B)) =(=+1,B\{C}) 2pH(pHm2,(=,B)) ==+|B| Figure 42: Concrete Resettable Counter Implementation I improved the design of the reliable causal broadcast middleware implemented for Lasp. This middleware is an essential component of the op-based platform and, for that reason, improving it is a main contribution that would lead to optimizing the op-based platform. The current implementation of this middleware presents a more scalable and dynamic version. 9.3 AntidoteDB As part of the Syncfree project [94], a FP7 EU project on large-scale replicated computation without synchronization. One of the major results of this project has been the geo-replicated highly available data store Antidote [95]. A central component of this data store is a library for conflict-free replicated data types. As a contribution to this project, I was responsible of replacing the initial implementation of CRDTs with a new optimized version. I improved the design of existing data types (add-wins set, remove-wins set, pn-counter) and adapted the design of further op-based types such as maps and flags for Antidote. I have also been contributing to Antidote under the EU project Lightkone [96] by implementing a mechanism that allows compressing operations before sending to other replicas. The code is available on GitHub (Section 9.6) in Erlang with unit and property-based tests. Moreover, while working on the map data type in Antidote, I tackled the problem of embedding counters inside maps that led to a paper entitled “Compact Resettable Counter through Causal Stability” [92] that was submitted to PAPOC/EUROSYS’17. In that paper, I presented the problem, proposed a new design for a counter that solves it and showed how causal stability could be used to garbage collect meta-data and lead to more compact state. 9.4 Redis with relaxed consistency The popularity of Redis [76] stems from its speed, rich semantics, and stability. It was primarily designed and generally used in a single server deployment model. A single Redis instance is a remote centralized solution, and not a distributed system. However, Redis provides distributed solutions known as Redis 113 CHAPTER 9. USE CASES Figure 43: The general architecture of our multi-master proposed solution. Sentinel and Redis Cluster which allow using multiple instances, asynchronous master-slave replication, and automatic fail over process. We contributed in implementing pure op-based CRDTs to build a multi-master replication feature in Redis [76], the famous in-memory cache system. We choose Redis because: (1) it helps us demonstrate how to integrate pure op-based CRDTs in an existing popular system; (2) we contribute with the community in building a crucial multi-master feature that is currently missing in Redis. Our experience reveals that implementing pure op-based CRDTs for many data types is straightforward, whereas integrating them in an existing system is somehow challenging if the aim is to preserve the legacy API and code-base intact. The code is available on GitHub (Section 9.6)inC. For the implementation of the multi-master replication feature using pure op-based CRDTs, we integrated three layers into Redis Server’s code. We present below the three layers, shown in Figure 43, with a brief description of each layer. Request handler The handler layer is the intermediate layer between Redis Client API and the TRCB and CRDT layers. Every client request is redirected by the original Redis Client API to this layer. Then, this layer prepares a client object containing the operation and arguments (if any), serializes it and ships it to the TRCB layer for dissemination. Tagged Reliable Causal Broadcast The second layer (TRCB) is used to broadcast client requests (operations and arguments) to all nodes in the cluster. This layer tags the client objects received from the handler with a timestamp needed to guarantee causal delivery at each node. Also, the TRCB is implemented in a way to guarantee exactly-once delivery of each client object to each node in the cluster. CRDT layer The CRDT layer is where we implement the pure op-based CRDTs and their related structures, such as the POLog (Partially-Ordered Log), a map where the key is a timestamp and the value is the 114 9.5. MINIDOTE operation and arguments, following the designs and specifications in [7]. In addition to that, we perform a two-phase POLog compaction to make CRDTs even more efficient. The first phase removes obsolete and redundant information from the POLog in a way that does not affect the result of the queries, keeping only relevant of operations. The second one is by using causal stability information from the middleware to discard the timestamps of causally stable operations and move them from the POLog to the Redis data types. Challenges We had to figure out solutions for the challenges we faced. The challenges are in terms of architecture design, preserving original API with minor changes, reusability of Redis code, configurability of the system, choice of messaging pattern and implementation of causal stability. A more detailed version of this work is published in [97]. 9.5 Minidote Minidote [98] is a lightweight, replicated key-CRDT store. The anticipated use cases for Minidote are those where the managed data fits well into the memory of each node in the cluster. The data is automatically replicated on each node, while concurrent updates are resolved using CRDTs. Minidote provides causal consistency with atomic batch-reads and batch-updates. Compared to Antidote, Minidote includes the following simplifications, which makes it more lightweight and therefore allow it to run well on less powerful devices: 1. There is no sharding of data onto multiple machines within a data center. Each replication site only consists of a single machine. 2. There is (currently) no support for interactive transactions. Figure 44 sketches the high-level architecture of Minidote . The components drawn with dashed lines are reusable components that are shared with Antidote. My contribution was in replacing the inter-dc replication service used in Antidote with a different causal broadcast service provided by Camus. Camus is a CA usal MU lticast S ervice, that provides different backends to guarantee a reliable dissemination and delivery respecting causal order at all replicas using the service. The back-end used in Minidote is an implementation of tagged casual broadcast (TCB) protocol. In short, TCB guarantees that messages will be delivered respecting the end-to-end happens-before relation as seen by the application. Moreover, it uses smaller size timestamps than vector clocks to encode the causality between messages, a causal graph to store the dependencies between messages and an efficient algorithm for causal delivery and stability. The code is available on GitHub (Section 9.6) in Erlang with unit and property-based tests. The minidote_server process, at every Minidote instance, stores locally in its state a dot, i.e. a pair of (=>343,2>D=C4A), which serves as a unique identifier for a message to be broadcast, and a context, 115 CHAPTER 9. USE CASES Figure 44: High-level architecture of Minidote. i.e. a set of dots, which stores the ids/dots of the most recently delivered messages. The context serves as the causal dependency of the next message to be broadcast. To explain the integration of TCB in Minidote, we explain below the API of Camus: • tcbcast(Msg, State): broadcasts a message to every process in the group membership. The client process (here minidote_server) will be responsible for tagging the message to be sent with a new dot and the causal context. • tcdeliver(Msg, Timestamp, State): delivers a message to the application with the tag. The client process or application (here: minidote_server) also implements message delivery. For instance, in the case of an operation-based AW-Set (0,C234;8E4A would add the operation (033,4,and respective timestamp C0) to the state (0. • tcstable(Msg, Timestamp, State): stabilizes a delivered message. The minidote_server process also needs to handle messages coming from the TCB service in the format camus, Opaque. Upon receiving such a message, the function camus:handle(Opaque, State) is called, where State comprises the local dot and context. The TCB process will update the state and return it along with the type of message (34;8E4A,BC01;4), and the payload. The minidote_server process updates the state and based on the type of message calls C234;8E4A or C2BC01;4. Minidote does not fully benefit from all the features of the TCB protocol. The reason is that Minidote uses Antidote’s classical operation-based CRDTs as data types which use random unique tokens as a unique identifier and do not provide meta data information on the provenance of the operations. This 116 9.6. SOFTWARE, LIBRARIES AND ARTIFACTS means that currently Minidote does its own causal tagging of operations (using random tokens) to be used for comparing operations, and the dot and context tagging provided by TCB to provide causal delivery order. An improved implementation should to use the dot and context as the only tag to provide endto-end causal delivery, without a need of duplication of effort (tagging) and extra metadata. With such modifications to the CRDTs, Minidote could benefit from an efficient garbage collection of CRDT metadata that is provided by causal stability and that is not possible with the current CRDT library. Finally, the current version works in the scope of static systems and therefore does not allow nodes to join and leave. As the anticipated use cases for Minidote are targeting deployments closer to the edge, nodes are likely to join and leave more frequently as with Antidote’s data center deployment. We therefore plan to extend the current version to cover the scope of dynamic systems by extending the API of Minidote to deal with joining, leaving, state-transfer and using the dynamic version of TCB. 9.6 Software, libraries and artifacts •?iiTb,ff;Bi?m#X+QKf;vQmM2bf+Kmb, Erlang implementation of different causal delivery and stability middleware inluding: Classical VV-based, Classical E2E VV-based, Optimised E2E VVbased, E2E TCB algorithm, Dynamic E2E TCB. •?iiTb,ff;Bi?m#X+QKf;vQmM2bf+Kmbn2tT, Erlang implementation for deployment, orchestration and evaluation of Camus using docker and kubernetes. •?iiTb,ff;Bi?m#X+QKf;vQmM2bfKBMB/Qi2, Erlang implementation of Miniodte, a lightweight CRDT data store using TCB as causal delivery middleware. •?iiTb,ff;Bi?m#X+QKfMiB/Qi2."fMiB/Qi2n+`/i, Erlang implementation of Op-based Conflict-free Replicated Data Types (CRDTs) in Erlang. •?iiTb,ff;Bi?m#X+QKf;vQmM2bf_*", Erlang implementation of reliable causal broadcast. •?iiTb,ff;Bi?m#X+QKfHbT@HM;fivT2b, Prototype implementation of Conflict-free Replicated Data Types (CRDTs) in Erlang. •?iiTb,ff;Bi?m#X+QKfHbT@HM;fBb?BFr, Erlang implementation of tagged reliable causal broadcast over partisan/hyparview. •?iiTb,ff;Bi?m#X+QKf?bH#f1*fi`22fKbi2`fiQTB+Rf`2/Bb, PoC: Implementation of multimaster replication in Redis). 117 Ry Conclusions and Future Perspectives The work in this thesis leveraged the importance of giving another thought to causal consistency techniques designed for systems of an old era, back three decades. My work revisited the end to end causal delivery stack from the application to the causal middleware API, including its internal logic. This lead to novel class Dynamic End-to-End Reliable Causal Delivery Middleware for Geo-Replicated Services, tailored for modern causality systems. My research started by analyzing the causality requirements in modern applications that are complex and multi-threaded. I explained how distributed systems practitioners could try to achieve end-to-end happens-before incorrectly by identifying the pitfalls in trying to achieve that. This work affirms the importance of going to fundamentals while designing complex systems before engineering them. To this end, I introduced Tagged Causal Delivery (Broadcast) that ensure an end-to-end happens-before. Tagged Causal Delivery redefines the happens-before relation to reflect application level dependencies and provide knowledge about the causal relations and concurrency between operations to the application. On the other hand, studying the interaction between applications and the underlying causal broadcast middlewares was helpful to inspire new functionalities that makes consistency at the application semantics correct and efficient. For this, I introduced a new causal broadcast architecture and API that grants the application layer access to causality information for timestamping as well as for garbage collection, using causal stability. I have shown that these new middleware functionalities reduce the bandwidth and memory overhead significantly without compromising latency. Nevertheless, systems in which causal stability is not required may not take advantage of this optimization, while paying a little cost in causal delivery. Indeed, the delivery latency overhead in my work is negligible overhead (less than millisecond) over classical systems, while improving the causal stability one order of magnitude. One more aspect addressed in the work was the internal abstractions used in causal middlwares. I have shown that the classical causal broadcast algorithms, based on version clocks, are not efficient in memory and stability as the number of replicas increase. This lead to the introduction of dependency dots and a causal DAG as data structures to represent the causal dependencies and metadata. I designed and implemented an efficient graph-based Tagged Causal Delivery middleware. In the experiments, I showed how the Tagged Causal Delivery middleware scales better that an optimized E2E VV-based algorithm in 118 10.1. FUTURE WORK terms of transmission of meta data on the network, size of metadata in memory and the non-causal stability latency. I also showed that the algorithm is more resilient in case of unreliable networks, where delays and partitions occur. Dynamic membership was another focus in my thesis, inspired by the advent of elastic systems, enabled through virtualization techniques. I introduced the first causal delivery middleware that tolerates dynamic membership while maintaining the causal consistency guarantees. I showed how the new graphbased TCB stability allows for the system operation in a high churn dynamic environment and in a nonblocking fashion. During the course of my thesis, I applied these concepts to many use cases presented in the last chapter. The work was part of the EU H2020 projects, LightKone and Syncfree, as well as other national projects. Part of the work was also funded through the Google Summer Code grant. All these works obviously consumed significant time of my thesis time; however, this paid off in two main ways (in addition to the funding support). The first is digging in the engineering of the middleware which helped in realizing the theoretical pifalls as well as resulting in a well structured and reliable TCB codebase in Erlang .The second benefit was increasing the impact of my work through other works, like Lasp and Antidote, that have seen high adoption in the industry. 10.1 Future Work One direction would be extending our current middleware. More work can be done on batching operations that would improve the performance in terms of non-causal latency and stability. Also, work on causal stability required the delivery of operations to the entire system. It would be interesting to study the usefulness and technique of partial stability in a quorum-based fashion, e.g., K-stability, where the system is causality stable considering quorum of K replicas out of the total membership. Making our causal middleware more adaptive by allowing it to dynamically pick the optimal send rates to be used based on the network delays. An important part to work on is to improve the mechanism used in the TCB middleware to compare operations at the client; the mechanism used now is still slower than comparing two version vectors. Finally, we would like to work on finding solutions for node migration, persistence and mostly extend the middleware to allow partial replication. Another direction would be extending and improving the existing pure operation-based CRDTs design and implementation. This novel CRDTs library and the current middleware could be used together to develop a highly-available and scalable data store that could offer interesting trade-offs. Moreover, it would be also interesting to work on new CRDTs like sequences that could be used for collaborative editing, without the need to synchronise often. 119 BIBLIOGRAPHY [59] M.H. Kalantar and K.P. Birman. “Causally ordered multicast: the conservative approach”. In: Proceedings. 19th IEEE International Conference on Distributed Computing Systems (Cat. No.99CB37003) . 1999, pp. 36–44. doi: RyXRRyNfA*.*aXRNNNXdde8y9 (cit. on p. 29). [60] Wyatt Lloyd et al. “Stronger Semantics for Low-Latency Geo-Replicated Storage”. In: Proceedings of the 10th USENIX Conference on Networked Systems Design and Implementation . nsdi’13. Lombard, IL: USENIX Association, 2013, pp. 313–328 (cit. on p. 31). [61] Jiaqing Du et al. “Orbe: Scalable Causal Consistency Using Dependency Matrices and Physical Clocks”. In: Proceedings of the 4th Annual Symposium on Cloud Computing . SOCC ’13. Santa Clara, California: Association for Computing Machinery, 2013. isbn: 9781450324281. doi: RyXR R98fk8kjeReXk8kjek3. url: ?iiTb,ff/QBXQ`;fRyXRR98fk8kjeReXk8kjek3 (cit. on p. 31). [62] Jiaqing Du et al. “GentleRain: Cheap and Scalable Causal Consistency with Physical Clocks”. In: Proceedings of the ACM Symposium on Cloud Computing . SOCC ’14. Seattle, WA, USA: Association for Computing Machinery, 2014, pp. 1–13. isbn: 9781450332521. doi: RyXRR98fkedyNdNXk edyN3j. url: ?iiTb,ff/QBXQ`;fRyXRR98fkedyNdNXkedyN3j (cit. on p. 32). [63] Deepthi Devaki Akkoorath et al. “Cure: Strong Semantics Meets High Availability and Low Latency”. In: 2016 IEEE 36th International Conference on Distributed Computing Systems (ICDCS) . 2016, pp. 405–414. doi: RyXRRyNfA*.*aXkyReXN3 (cit. on p. 33). [64] Sérgio Almeida, João Leitão, and Luı�s Rodrigues. “ChainReaction: A Causal+ Consistent Datastore Based on Chain Replication”. In: Proceedings of the 8th ACM European Conference on Computer Systems . EuroSys ’13. Prague, Czech Republic: Association for Computing Machinery, 2013, pp. 85–98. isbn: 9781450319942. doi: RyXRR98fk9e8j8RXk9e8jeR. url: ?iiTb, ff/QBXQ`;fRyXRR98fk9e8j8RXk9e8jeR (cit. on p. 33). [65] Robbert van Renesse and Fred B. Schneider. “Chain Replication for Supporting High Throughput and Availability”. In: Proceedings of the 6th Conference on Symposium on Operating Systems Design & Implementation - Volume 6 . OSDI’04. San Francisco, CA: USENIX Association, 2004, p. 7 (cit. on p. 33). [66] Manuel Bravo, Luı�s Rodrigues, and Peter Van Roy. “Saturn: A Distributed Metadata Service for Causal Consistency”. In: Proceedings of the Twelfth European Conference on Computer Systems . EuroSys ’17. Belgrade, Serbia: ACM, 2017, pp. 111–126. isbn: 978-1-4503-4938-3. doi: RyXRR9 8fjye9RdeXjye9kRy. url: ?iiT,ff/QBX+KXQ`;fRyXRR98fjye9RdeXjye9kRy (cit. on p. 34). [67] Syed Akbar Mehdi et al. “I Can’t Believe It’s Not Causal! Scalable Causal Consistency with No Slowdown Cascades”. In: Proceedings of the 14th USENIX Conference on Networked Systems Design and Implementation . NSDI’17. Boston, MA, USA: USENIX Association, 2017, pp. 453– 468. isbn: 9781931971379 (cit. on p. 35). 126 BIBLIOGRAPHY [68] Robert Escriva et al. “Kronos: The Design and Implementation of an Event Ordering Service”. In: Proceedings of the Ninth European Conference on Computer Systems . EuroSys ’14. Amsterdam, The Netherlands: Association for Computing Machinery, 2014. isbn: 9781450327046. doi: RyXR R98fk8NkdN3Xk8Nk3kk. url: ?iiTb,ff/QBXQ`;fRyXRR98fk8NkdN3Xk8Nk3kk (cit. on p. 36). [69] Mustaque Ahamad et al. “Causal memory: Definitions, implementation and programming”. In: IEEE Transactions on Parallel and Distributed Systems 1 (1990), pp. 6–16 (cit. on pp. 37,40). [70] Prince Mahajan, Lorenzo Alvisi, Mike Dahlin, et al. “Consistency, availability, and convergence”. In: University of Texas at Austin Tech Report 11 (2011) (cit. on p. 37). [71] Douglas B. Terry et al. “Session Guarantees for Weakly Consistent Replicated Data”. In: Proceedings of the Third International Conference on Parallel and Distributed Information Systems . PDIS ’94. Washington, DC, USA: IEEE Computer Society, 1994, pp. 140–149. isbn: 0-8186-6400-2. url: ?iiT,ff/HX+KXQ`;f+BiiBQMX+7K\B/4e98dNkXee3jyk (cit. on p. 37). [72] David R Cheriton and Dale Skeen. Understanding the limitations of causally and totally ordered communication . Vol. 27. 5. ACM, 1994 (cit. on pp. 38,40,47). [73] Marc Shapiro et al. “Conflict-free Replicated Data Types”. In: Proceedings of the 13th International Conference on Stabilization, Safety, and Security of Distributed Systems . SSS’11. Grenoble, France: Springer-Verlag, 2011, pp. 386–400. isbn: 978-3-642-24549-7. url: ?iiT,ff/HX+KXQ`;f +BiiBQMX+7K\B/4ky8yeRjXky8ye9k (cit. on pp. 38,41,42). [74] J. H. Saltzer, D. P. Reed, and D. D. Clark. “End-to-end Arguments in System Design”. In: ACM Trans. Comput. Syst. 2.4 (Nov. 1984), pp. 277–288. issn: 0734-2071. doi: RyXRR98fj8d9yR Xj8d9yk. url: ?iiT,ff/QBX+KXQ`;fRyXRR98fj8d9yRXj8d9yk (cit. on p. 41). [75] Microsoft Azure. Multi-master at global scale with Azure Cosmos DB . 2018. url: ?iiTb,ff/Q+bX KB+`QbQ7iX+QKf2M@mbfxm`2f+QbKQb@/#fKmHiB@`2;BQM@r`Bi2`b (cit. on p. 42). [76] Redis Labs. Under the Hood: Redis CRDTs . 2017? url: ?iiTb,ff`2/BbH#bX+QKf/Q+bf +iBp2@+iBp2@r?Bi2TT2`f (cit. on pp. 42,113,114). [77] Basho. Riak KV Concepts: Data Types . url: ?iiT,ff/Q+bX#b?QX+QKf`BFfFpfkXkXj fH2`Mf+QM+2Tibf+`/ibf (cit. on p. 42). [78] Atom Editor. Code together in real time with Teletype for Atom . 2017. url: ?iiT,ff#HQ;XiQKX BQfkyRdfRRfR8f+Q/2@iQ;2i?2`@BM@`2H@iBK2@rBi?@i2H2ivT2@7Q`@iQKX?iKH (cit. on p. 42). [79] Protocol Labs. Decentralized Real-Time Collaborative Documents - Conflict-free editing in the browser using js-ipfs and CRDTs . 2017. url: ?iiTb,ffBT7bXBQf#HQ;fjy@Db@BT7b@+`/ibXK/ (cit. on p. 42). 127 BIBLIOGRAPHY [80] Edsger W. Dijkstra. “Self-stabilizing Systems in Spite of Distributed Control”. In: Commun. ACM 17.11 (Nov. 1974), pp. 643–644. issn: 0001-0782. doi: Ry X RR98 f jeRRdN X jeRkyk. url: ?iiT,ff/QBX+KXQ`;fRyXRR98fjeRRdNXjeRkyk (cit. on p. 62). [81] The Poisson Distribution and Poisson Process Explained . https://towardsdatascience.com/thepoisson-distribution-and-poisson-process-explained-4e2cb17d459 (cit. on p. 93). [82] Donald E. Knuth. The Art of Computer Programming, Volume 1 (3rd Ed.): Fundamental Algorithms . USA: Addison Wesley Longman Publishing Co., Inc., 1997. isbn: 0201896834 (cit. on p. 94). [83] Jose A. Hernandez Gutierrez and Iain Phillips. Weibull mixture model to characterise end-to-end Internet delay at coarse time-scales . Jan. 2006. url: ?iiTb,ff?/HX?M/H2XM2ifkRj9fkRkR (cit. on p. 94). [84] T. Holleczek, V. Venus, and S. Naegele-Jackson. “Statistical Analysis of IP Delay Measurements as a Basis for Network Alert Systems”. In: 2009 IEEE International Conference on Communications (2009). doi: RyXRRyNfB++XkyyNX8RNN93d (cit. on p. 94). [85] Paulo Jesus, Carlos Baquero, and Paulo Sérgio Almeida. “Flow updating: Fault-tolerant aggregation for dynamic networks”. In: Journal of Parallel and Distributed Computing 78 (2015), pp. 53–64. doi: RyXRyRefDXDT/+XkyR8XykXyyj (cit. on p. 94). [86] Weibull Distribution . https://en.wikipedia.org/wiki/Weibull_distribution (cit. on p. 94). [87] Dirk Merkel. “Docker: lightweight linux containers for consistent development and deployment”. In: Linux journal 2014.239 (2014), p. 2 (cit. on p. 97). [88] Kubernetes Manual .TH2b2Tmii?2m`H. [Online; accessed 04-Dec-2017]. 2017 (cit. on p. 97). [89] Brian White et al. “An Integrated Experimental Environment for Distributed Systems and Networks”. In: Boston, MA, Dec. 2002, pp. 255–270 (cit. on p. 97). [90] Paul A. S. Ward. “Algorithms for Causal Message Ordering in Distributed Systems”. In: 2007 (cit. on pp. 97,98). [91] Houssam Yactine, Ali Shoker, and Georges Younes. “ASPAS: As Secure as Possible Available Systems”. In: Distributed Applications and Interoperable Systems . Ed. by Miguel Matos and Fabíola Greve. Cham: Springer International Publishing, 2021, pp. 57–73. isbn: 978-3-030-78198-9 (cit. on p. 109). [92] Georges Younes, Paulo Sérgio Almeida, and Carlos Baquero. “Compact Resettable Counters through Causal Stability”. In: Proceedings of the 3rd International Workshop on Principles and Practice of Consistency for Distributed Data . PaPoC ’17. Belgrade, Serbia: Association for Computing Machinery, 2017. isbn: 9781450349338. doi: RyXRR98fjye933NXjye93Nk. url: ?iiTb,ff/QBX Q`;fRyXRR98fjye933NXjye93Nk (cit. on pp. 111,113). 128 BIBLIOGRAPHY [93] Christopher Meiklejohn and Peter Van Roy. “Lasp: A Language for Distributed, Coordination-Free Programming”. In: Proceedings of the 17th International Symposium on Principles and Practice of Declarative Programming . PPDP ’15. Siena, Italy: Association for Computing Machinery, 2015, pp. 184–195. isbn: 9781450335164. doi: RyXRR98fkdNy99NXkdNy8k8. url: ?iiTb,ff/QBX Q`;fRyXRR98fkdNy99NXkdNy8k8 (cit. on p. 111). [94] Large-scale computation without synchronisation. ?iiTb , f f bvM+7`22 X HBTe X 7` (cit. on p. 113). [95] Antidotedb. ?iiT,ffbvM+7`22X;Bi?m#XBQfMiB/Qi2f (cit. on p. 113). [96] Ali Shoker et al. “Lightkone: Towards general purpose computations on the edge”. In: White Paper published on http://www. lightkone. eu 40 (2016) (cit. on p. 113). [97] Georges Younes et al. “Integration Challenges of Pure Operation-Based CRDTs in Redis”. In: First Workshop on Programming Models and Languages for Distributed Computing . PMLDC ’16. Rome, Italy: Association for Computing Machinery, 2016. isbn: 9781450347754. doi: RyXRR98fkN8d jRNXkN8djd8. url: ?iiTb,ff/QBXQ`;fRyXRR98fkN8djRNXkN8djd8 (cit. on p. 115). [98] Minidote .?iiTb,ff;Bi?m#X+QKfGB;?iEQM2fJBMB/Qi2 (cit. on p. 115). 1bi2/Q+mK2MiQ7QB;2`/Q+QKQT`Q+2bb/Q`UT/7fs2fGmVG  h 1s2QKQ/2HQLPoi?2bBbUpeXNXyV(R)X (R) CQ½QJXGQm`2MÏQXh?2LPoi?2bBbG  h 1sh2KTHi2lb2`ǶbJMmHXLPolMBp2`bBivGBb#QMXkykRXl_G,?iiTb,ff;Bi?m#X+QKfDQQKHQm`2M+QfMQpi?2bBbf`rfKbi2`fi2KTHi2XT/7U+BiXQMTXRkNVX 129