Asynchronous Latency and Fast Atomic Snapshot
Abstract
This paper introduces a novel, fast atomic-snapshot protocol for asynchronous message-passing systems. In the process of defining what “fast” means exactly, we spot a few interesting issues that arise when conventional time metrics are applied to long-lived asynchronous algorithms. We reveal some gaps in latency claims made in earlier work on snapshot algorithms, which hamper their comparative time-complexity analysis. We then come up with a new unifying time-complexity metric that captures the latency of an operation in an asynchronous, long-lived implementation. This allows us to formally grasp latency improvements of our atomic-snapshot algorithm with respect to the state-of-the-art protocols: optimal latency in fault-free runs without contention, short constant latency in fault-free runs with contention, the worst-case latency proportional to the number of active concurrent failures, and constant amortized latency.
Keywords and phrases:
Asynchronous systems, time complexity, atomic snapshot, crash faultsFunding:
João Paulo Bezerra: Mazars, TrustShare Innovation Chair.Copyright and License:
2012 ACM Subject Classification:
Theory of computation Distributed algorithmsEditor:
Dariusz R. KowalskiSeries and Publisher:
Leibniz International Proceedings in Informatics, Schloss Dagstuhl – Leibniz-Zentrum für Informatik
1 Introduction
The distributed snapshot abstraction [14, 26] allows us to determine a consistent view of the global system state. Originally proposed in the asynchronous fault-free message-passing context, it was later cast to shared-memory models [3] as a vector of shared variables, exporting an update operation that writes to one of them and a snapshot operation that returns the current vector state. Atomic snapshot can be implemented from conventional read-write registers in a wait-free manner, i.e., tolerating unpredictable delays or failures of any number of processes. By applying the reduction from shared memory to message-passing [6], one can get an asynchronous distributed atomic-snapshot implementation that tolerates up to a minority of faulty processes. The atomic-snapshot object (ASO) is, in a strong sense, equivalent to lattice agreement (LA) [8, 17]111Lattice agreement can be seen as a weak version of consensus, where decided values form totally ordered joins of proposed values in a join semi-lattice.: one can implement the other with no time overhead. A long line of results improve time and space complexities of ASO and LA algorithms in shared-memory [5, 4, 7] and message-passing [17, 21, 19, 16, 18] models.
In this paper, we focus on the latency of operations in message-passing ASO implementations. We propose an LA (and, thus, ASO) algorithm that is faster than (or matches) state-of-the-art solutions in all execution scenarios: with or without failures and with or without contention. The comparative analysis of our algorithm with respect to the existing work appeared to be challenging: as we show, earlier work considered diverging metrics and execution scenarios, and sometimes used over-simplified reasoning. We observed that conventional metrics [13, 6, 2] are not always suitable for long-lived asynchronous algorithms. Besides, prior latency analyses of ASO and LA algorithms [17, 21, 19, 16, 18] used different ways to measure time, which complicated the comparison. We therefore propose a unifying time-complexity analysis of prior asynchronous ASO and LA algorithms with respect to a new metric, which we take as a contribution on its own.
|
|
Worst-case |
|
|||||||
| Faleiro et al. [17] | yes | |||||||||
| Imbs et al. [21] | no | |||||||||
| Garg et al. [18] | yes | |||||||||
| Garg et al. [18] + Zheng et al. [27] | no | |||||||||
| Delporte et al. [16] | no | |||||||||
| This paper | yes |
Lamport [25] proposed to measure time in asynchronous systems as the length of the longest chain of causally related messages, the metric used to to determine the best-case latency of consensus [25] and Crusader Agreement [1]. However, as we show in this paper, the metric may produce counter-intuitive results for protocols involving all-to-all communication. For instance, in the failure-free case, -process reliable broadcast [12] exhibits a causal chain of hops, even though, intuitively one expects it to terminate in one.
Building upon the classical approach by Canetti and Rabin [13], Abraham et al. [2] recently proposed an elegant metric to grasp the good-case latency of broadcast protocols. We observe, however, that the metric does not really apply to executions of long-lived abstractions, which may contain holes – periods of inactivity when no protocol messages are in transit. Moreover, we get diverging results when applying [2] and [13] to operation latency, i.e., the time between invocation and response events of a given operation.
We therefore extend the round-based approach to long-lived abstractions (such as ASO and LA) and establish a framework to measure the time between arbitrary events, subsequently showing that the results align with those from earlier classical metrics [6, 13].
To summarize, our main contribution is a novel LA (and, thus, ASO) protocol that is generally faster than prior solutions, i.e., it exhibits shorter latency of its operations in various scenarios. In our complexity analysis, we compared our protocol to the original long-lived LA algorithm by Faleiro et al. [17]222We consider the ASO protocol built atop the lattice agreement protocol proposed in [17]., the first direct message-passing ASO implementation by Delporte et al. [16], the ASO algorithm based on the set-constraint broadcast by Imbs et al. [21], and the ASO algorithms by Garg et al. based on generic construction of ASO from one-shot LA with constant latency in fault-free runs [18] or worst-case latency by Zheng et al. [27] (where is the number of processes).
As shown in Table 1, in a fault-free run, the latency of an operation of our protocol is the optimal two rounds if there is no contention and eight rounds in the presence of contention (four rounds if we ignore the “buffering” period when a value is submitted but not yet proposed), regardless of the number of contending operations. Moreover, the worst-case latency of our algorithm is proportional to the number of active failures , i.e., the number of faulty processes whose messages are received within the operation’s interval, therefore the amortized latency (averaged over a large number of operations in a long-lived execution) converges to the fault-free constant.
Our protocol can be seen as a novel combination of techniques employed separately in prior work. These include the use of generalized (long-lived) lattice agreement as a basis for ASO [23], the helping mechanism where all the learned lattice values are shared [23], relaying of messages to all replicas instead of quorum-based rounds [21, 18, 22, 15], and buffering proposed values until previous proposals get committed [17]. Similar to earlier proposals [17], our algorithm involves (all-to-all) communication, which is compensated by its constant (amortized) latency. An interesting open question is whether one can reduce the communication cost in good runs, while maintaining constant amortized latency.
The paper is organized as follows. In Section 2, we present our model assumptions, and in Section 3, we state the problem of atomic snapshot and relate it to generalized lattice agreement. In Section 4, we present our protocol and analyze its correctness. In Section 5, we discuss several gaps in the complexity analyses of earlier work. In Section 6, we present a comparative analysis of time metrics. Certain proofs and a detailed discussion of time complexity of earlier protocols are delegated to the appendix.
2 System Model
Processes and Channels.
We consider a system of processes (or nodes). Processes communicate by exchanging messages with a sender , a receiver , and a message content data.
A process is an automaton modeled as a tuple , where is a set of inputs (messages and application calls) it can receive, is a set of outputs (messages and application responses), is a (potentially infinite) set of possible internal states, is an initial state and is a transition function mapping a set of inputs and a state to a set of outputs and a new state. Each process is assigned an algorithm which defines , a distributed algorithm is an array .
Events and Configurations.
Application calls and responses are tuples and with a process identifier, a request, and a reply respectively.
An event is a tuple where is a set of received messages and/or application calls, is the set of nodes producing the event and is a set of messages sent and/or application responses. We denote as the set of messages received in the event, conversely, is the set of messages sent. A message hop is a pair in which receives at least one message that was sent in .
Messages in transit are stored in the message buffer.333We assume that every message in the message buffer is unique. A configuration is an -array with the buffer’s state and the local state of each node (). Let denote the initial configuration in which every is an initial state and the buffer is empty.
Executions.
An execution (or run) is an alternating sequence of configurations and events, where for each and :
-
1.
;
-
2.
consists of messages and application outputs that the nodes in produce, given their algorithms, their states in and their inputs in ; the nodes in carry their states from to , accordingly;
-
3.
for the nodes , .
Each triple is called a step. In this paper, we consider algorithms defined by deterministic automata, and we assume a default initial state. Thus, we sometimes skip configurations and simply write .
In an infinite execution, a process is correct if it takes part in infinitely many steps, and faulty otherwise. We only consider infinite executions in which , where is the number of faulty processes and is the total number of processes. Moreover, in an infinite execution, messages exchanged among correct processes are eventually received, i.e., if there is an event from a correct process sending a message to another correct process, then there is succeeding such that .
We also assume that the communication channels neither alter nor create messages. Finally, we assume that the channels are FIFO: messages from a given source to a given destination arrive in the order they were sent. A FIFO channel can be implemented by attaching sequence numbers to messages, without extra communication or time overhead.
3 Lattice Agreement and Atomic Snapshot
3.1 Lattice Agreement
A join semi-lattice is defined as a tuple , where is a partial order on a set , such that for any pair of values and in , there exists a unique least upper bound ( is called the join operator). Also, and in are said to be comparable if .
The (generalized) Lattice Agreement abstraction LA [17] defined over can be accessed by every node with operation , (we say that the node proposes ) which triggers the reply event (we say that the node learns ). Each node may invoke Propose any number of times but does so sequentially, that is, it initiates a new operation only after the previous one has returned.444Following [23], without loss of generality, we slightly modified the conventional LA interface [8, 17] by introducing the explicit Propose operation that combine proposing and learning the values, the properties of the abstraction are adjusted accordingly. The abstraction must satisfy:
Definition 1 (Lattice Agreement (LA)).
-
Validity. Any value learned by a node is the join of some set of proposed values that includes its last proposal.
-
Stability. The values learned by any node increase monotonically, with respect to .
-
Consistency. All values learned are comparable, with respect to .
-
Liveness. If a correct node proposes , it eventually learns a value .
3.2 Atomic Snapshot Object (ASO)
An atomic snapshot object (ASO) stores a vector of values and exports two operations: and . The operation writes the value in and returns OK, and returns the entire vector . An ASO implementation guarantees that every operation invoked by a correct process eventually completes. It also ensures that each of its operations appears to take effect in a single instance of time within its interval, i.e., it is linearizable [20].
We say that an ASO is single writer SW (resp. multi writer MW) if for each of its registers , only a single process can call (resp. every process can call ). In this paper, we focus mostly on SWMR atomic snapshot objects. In Table 1 we give results only for SWMR. A MWMR ASO can be devised from SWMR by adding an additional “read” phase when updating values (see Section 3.3 for more details).
Next, we show that ASO can be implemented on top of LA with no additional overhead.
3.3 From LA to ASO
To implement a SWMR ASO on top of LA, we build upon the transformation algorithm introduced in [23]. We consider a partially ordered set of -vectors (recall that is the size of the ASO vector and is the number of nodes), defined as follows.
A vector position is defined as a tuple , where is an element of a value set equipped with a total order , and is the number of write operations on position . A total order on is defined in the natural way: for any two tuples . For each process , the vector position stores the number of snapshot operations executed by .
The lattice of -position vectors is then the composition . The partial order on is then naturally defined as the compositions of . The composed join operator is the composition of operators, one for each position in the -position vectors. The construction implies a join semi-lattice [23].
In Algorithm 1, we show how to implement an SWMR atomic snapshot on top of LA defined over the semi-lattice . For simplicity, we assume that , i.e., the size of the array is the total number of nodes, and that each node has a dedicated register where it can write. Elements of are then -vectors.
When a node calls i,v, it increments its local writing sequence number and proposes a -vector with in position and initial values in all other positions to the LA object. The vector learned from this proposal is ignored. When the node calls , it increments its local reading sequence number proposes a -vector with in position and initial values in all other positions to the LA object. The values in the first positions of the returned vector is then returned as the snapshot outcome.
Algorithm 1 can be extended to implement a MWMR ASO: to update a position in the array, a node first takes a snapshot to get the current state, gets up-to-date sequence number in position and proposes its value with a higher sequence number. With this modification, the update operation takes two LA operations instead of one. We refer the reader to [23] for further details.
Theorem 2.
Algorithm 1 implements ASO.
Proof.
See Appendix A.
4 LA Protocol
In Algorithm 2, we describe our protocol for solving LA. To guarantee amortized constant complexity, the protocol relies on two basic mechanisms, employed separately in earlier work [17, 23]. First, when a node receives a request (e.g., a value from the application), it first adds the request to a buffer () and then relays it before starting a proposal. This ensures that “idle” nodes also help in committing the request. Second, the node relays every learned value so that nodes that are “stuck” can adopt values from other nodes.
4.1 Overview
The protocol is based on helping: every node tries to commit every proposed value it is aware of. As long as the node has active proposals that are not yet committed, it buffers newly arriving proposals in the local variable . Intuitively, in the worst case, an operation has to wait until one of the concurrently invoked operations complete. Once this happens, the currently buffered value is put in the local dictionary and shared with the other nodes (lines 31 and 32) via a PROPOSE message. In turn, the other nodes relay the message to each other (line 38). The dictionary maps a value to the number of times it is “supported” by the nodes (using PROPOSE messages). Once a value in the dictionary assembles a quorum of of messages, i.e., (line 39), the value is added to the variable. Once every value currently stored in is in (line 41), the operation completes with as the learned value. As the final element of the helping mechanism, each process broadcasts every value it learns (lines 45 and 51), ensuring that processes that might otherwise remain “stuck” can complete their current proposal.
In summary, the algorithm relies on four main ideas: 1) buffering incoming requests when already proposing, 2) sharing every received proposal so all processes are quickly aware of active ones, 3) initiating a new proposal only after all currently seen proposals have been validated, and 4) broadcasting learned values to help other processes make progress.
Message Complexity.
The protocol is comprised of three all-to-all communication phases: processes send and relay requests at lines 23 and 27, proposals at lines 32 and 38, and accepted values at lines 45 and 51. The total number of messages is therefore . However, a value in a PROPOSE message can include up to distinct requests, and a value in a ACCEPT message may have arbitrary size. Therefore, in the extended version [11], we present a refined protocol description in which processes exchange messages per individual request. This efficiency is achieved by relaying only the differences between current and previously received proposals and the learned values in phases and , thus eliminating redundant messages with the same requests.
4.2 Correctness
Validity and Stability are immediate. We now proceed with Consistency and Liveness.
Lemma 3.
If nodes and learn, resp., values and , then and are comparable.
Proof.
Suppose that . Then there must exist and such that and .
Let (resp. ) be the quorum used to include at line 39. Since , there is a common node that sent to and to , but since channels are FIFO, either received or received from before learning a value, therefore adding the value to . Suppose it was that received before , from the condition of line 41, could not have learned if .
Lemma 4.
If a correct node sets , eventually learns a value with .
Proof.
A node sends a PROPOSE message to every other node whenever it adds a new value to (line 38). If is correct, it will receive at least PROPOSE messages for every value in , adding the value to . Therefore, the condition in line 41 is never satisfied from some point on only if keeps adding a new value to before all the current ones are validated.
Since each node proposes only one value at a time (until it learns a value, lines 28, 44, 50), for to indefinitely add new values to , there must be at least one other node that keeps learning values and proposing new ones. Without loss of generality, let be one such node. Since faulty nodes eventually crash and stop taking steps, must be correct. Every time learns a new value it sends to , and because channels are FIFO, receives the ACCEPT message before the new value proposed by . Eventually (because sent its proposal to ), one of the received values contains and the condition on line 46 is satisfied, then learns .
Lemma 5.
If a correct node calls , it eventually sets , .
Proof.
Let a correct node call , then includes in (line 22). If is not currently proposing, that is, the current value of is , then it meets the condition in line 28 and immediately sets . Otherwise, by Lemma 4, it eventually learns a value and sets in lines 44 and 50, thus meeting the condition in line 28 and setting .
Theorem 6.
Algorithm 2 implements Generalized Lattice Agreement.
4.3 Time Metric
We now define the latency metric we are going to use in evaluating time complexity. Our metric is inspired by the metric proposed by Abraham et al. [2] (which in turn rephrases the original metric by Canetti and Rabin [13]). The distinguishing feature of our approach is that it also applies to long-lived executions and executions with holes (illustrated in Figure 1).555In Section 6, we show that the three metrics are equivalent in “hole-free” executions.
Algorithm 3 describes the iterative method that assigns rounds to events in an execution. We give an informal description of the metric below.
Definition 8 (Iterative Round Assignment - Informal).
Algorithm 3 assigns round to the initial event, and defines the end of round as the last event that receives a message sent in round . In addition, if there are no more messages to be received (or in transit), the event inherits the round number of its immediate predecessor.
Definition 9 (IRA - Arbitrary Events).
To measure the latency between two events and , we assign rounds according to Algorithm 3, starting from , with all events up to and including receiving round . The latency between and is then given by the round assigned to .
We say that an application request (or simply request, when there is no ambiguity) completes once the receiving node learns a value which includes the request. For a specific node , we are interested in measuring the latency between the event in which receives a value from the application software, and an event , in which learns a value with .
4.4 Time Complexity of Algorithm 2
We define latency as the number of rounds spanning between the moment a correct process receives an application call and the moment it returns from the operation. In evaluating the latency of our protocol, we consider two types of executions: (1) the fault-free case, when all processes are correct, and (2) the worst-case, when only a majority of processes are correct.
A snapshot operation op precedes another operation if the response event of op happens before the call event for . Two operations are said to be concurrent if none precedes the other. For ASO protocols, we analyze latency in fault-free runs of an operation op in two distinct scenarios: (a) without contention, i.e., when no other operation overlaps in time with op, and (b) with contention, i.e., when there might be an arbitrary number of concurrent operations.
Garg et al. [19] use the notion of amortized time complexity, i.e., the average operation latency taken over a large number of operations in an execution. In some protocols, including ours, the latency of an operation is only affected by the number of faulty processes whose messages are received during the operation’s interval (we call these processes active-faulty). Intuitively, faulty processes take a finite amount of steps, so in these protocols a failure can only affect a finite number of operations. In this paper, we also distinguish ASO protocols with constant time complexity. In Appendix B, we establish the optimality of our protocol under no-contention and prove the following results:
Theorem 10.
An operation completes in at most rounds in fault-free runs w/o contention.
Consider an execution of our algorithm, and let () be its set of faulty processes.
Theorem 11.
An operation op takes at most rounds to complete if, during its interval, no correct node receives a message from a faulty one.
We say that there are active faulty nodes during an operation op if, in between the call and return events for op, a message is received from a total of distinct faulty nodes.
Theorem 12.
An operation op takes rounds to complete, where is the number of active faulty nodes during op.
5 Measuring Latency of ASO Protocols
We conclude the paper with an overview of time complexity of earlier LA and ASO protocols [17, 16, 21, 18, 19]. We highlight certain gaps in their latency analyses and discuss the ways to fix them. Formalities and proofs can be found in the full version [11].
The First Message-Passing LA Protocol.
Faleiro et al. [17] came up with the first LA implementation for asynchronous message-passing systems. They use the metric of [6] to measure latency and conclude that it takes rounds to output from a lattice agreement operation in their protocol.
We show the somewhat surprising result that this protocol has constant latency of rounds in fault-free runs. The upper bound holds as long as no message from faulty processes is received during the interval of the operation, implying that their LA protocol has constant amortized time complexity. We conjecture that the protocol has worst-case latency, where is the number of actual failures in the execution.
The First Direct ASO Implementation.
Delporte et al. [16] is the first paper to directly implement ASO in message passing systems, instead of using an atomic register implementation [6] and the shared-memory snapshot construction [3].
In fault-free runs without contention, the latency of their protocol is only rounds. In fault-free runs with contention, we support the claim of a bound of rounds from [19].
ASO with SCD-Broadcast.
Imbs et al. [21] introduce the abstraction of Set Constrained Delivery Broadcast (), and show that it allows for implementing LA and ASO with no complexity overhead. In their complexity analysis, they assume bounded message delays and show that the latency of their ASO algorithm in faulty-free and contention-free runs is rounds. We show that an operation of their resulting ASO algorithm can take rounds in fault-free runs with contention. We conjecture that this bound is tight, and so the time complexity of their ASO protocol is .
A Generic ASO Algorithm.
Garg et al. [18, 19] give a generic construction for atomic snapshot which uses any one-shot LA protocol as a building block (with constant latency overhead). The protocol thus inherits the asymptotic complexity of the underlying LA algorithm. They also provide a protocol for one-shot LA with rounds latency in fault-free runs (using [13]’s metric). Their protocol requires 2 rounds of communication plus two lattice agreement invocations in the good case w/o contention and three lattice invocations with contention, making it at least 6 and 8 message delays, respectively.
For the worst-case latency analysis, they assume an additional requirement over communication channels: if a process executes , sending to a correct process, then is eventually received (even if the sender is faulty). Using this assumption, they show a worst-case latency of for their LA protocol.
In this paper, we assume a weaker channel that only guarantees delivery of messages among correct processes. We show that under this model, the LA protocol of [18] has an execution that takes rounds. We conjecture the upper bound of their protocol to be , and also that when using the stronger assumption, both our (Section 4) and [21]’s protocol have worst-case latency.
The generic ASO construction may also be combined with the one-shot LA protocol presented in [27], which has worst-case latency of , providing an object whose update and snapshot operations take in both fault-free fault-prone executions. For the sake of completeness, we also provide the time complexity analysis for the one-shot LA protocols from [17] and [21] in the full version [11].
6 Comparative Analysis of Time Measurement Metrics
In this section, we recall metrics used in the literature [6, 13, 2, 24] for measuring time in asynchronous systems. We exhibit executions where the metrics by Attiya et al. [6] and Canetti and Rabin [13] yield arbitrary results due to the presence of holes – “periods of silence” during which no messages are in transit – which are common in long-lived protocols. We show that in a subset of executions without holes, which we refer to as covered executions, these metrics align with the one proposed by Abraham et al. [2]. This is not surprising, as these metrics were designed for distributed tasks, which assume finite hole-free executions. We also recall Lamport’s longest causal chain metric [24] and show that it is not suitable for comparing the ASO protocols we consider here.
Next, we show that the metric from [2] diverges from [6] and [13] when naïvely applied to measure time between arbitrary events. We then show that, after employing our refined method from Section 4.3 (Definition 9), they match when measuring rounds between arbitrary events in covered executions. Finally, we show that both our metric and that of [2] yield equivalent results in cases where [2] is applicable.
Altogether, we establish that our metric generalizes [2] and aligns with classical metrics [6, 13] when applied to distributed tasks. A summary of the comparative analysis is presented in Table 2. Proofs are provided in the extended version [11].
|
|
|
|
||||||||
| CR [13] | Yes | - | - | No | |||||||
| Round [6, 9] | Yes | Yes | Yes | No | |||||||
| NTR [2] | No | Yes | Yes | No | |||||||
| LCC [25] | No | No | No | Yes | |||||||
| IRA | No | Yes | Yes | Yes |
6.1 Definitions
Timed Executions.
We assume a global clock, not accessible to the nodes. A timed event is a pair in which is a non-negative real number, we also say that is a time assignment of . A timed execution is an alternating sequence where , where events are equipped with monotonically increasing times :
-
1.
whenever ;
-
2.
as .666We require this property to avoid the case where a never-terminating execution has a finite time duration.
A time assignment of is a timed execution in which every event in is matched with a timed event in and the sequences of configurations in and are the same. Notice that an execution allows for infinitely many time assignments.
Let be a message sent in and received in , the delay of is then defined as . For a finite timed execution , we define , (we use and when there is no ambiguity) and .
In the subsequent discussion, given an execution , let denote the set of all timed executions based on .
Time Metrics.
It is conventional to measure the execution time by the number of communication rounds, typically calculated using the “longest message delay.” These metrics can be applied to both executions and timed executions. The first metric we consider is defined in Definition 14 [6]. When applied to timed executions, this metric assumes a known upper bound on message delays, which can be normalized to one time unit without loss of generality. To apply this metric to an execution, we consider the maximum duration of all possible timed executions that adhere to the upper-bound communication constraint.
Definition 14 (Round metric).
Given a timed execution , in which the maximum message delay is bounded by one unit of time, takes rounds.
By extension, an execution takes rounds.
In the metric proposed by Attiya and Welch [9, 10], the time assignments are scaled so that the maximum message delay is always , thus, the metric produces the same results for executions as Definition 14. A more general metric introduced by Canetti and Rabin [13] captures the time complexity of any finite execution. Let be a timed execution, and let be the maximum message delay in it. Then takes CR rounds.
Definition 15 (CR metric).
A finite execution takes rounds, where is the maximum message delay of each corresponding timed execution.
Example 16.
Figure 2 shows an execution with four events, where we assign a delay of to the message exchanges and , and a delay of () to . By making arbitrarily small, the number of rounds in this execution converges to in the CR metric. The same result is obtained in the Round metric by setting .
Recently, Abraham et al. [2] proposed an approach that can be applied to executions without relying on time assignments. We call this metric non-timed rounds (NTR):
Definition 17 (NTR metric).
Given an execution , each event in is assigned a round number as follows:
-
The first event is assigned round . We also write ;
-
For any , let be the last event where a message of round is delivered. All events after until (and including) event are in round .
The number of rounds in is the round assigned to its last event.
Example 18.
Lamport [25] proposed a metric for latency based on the causal chain of messages. The Longest Causal Chain (LCC) was used to show best-case latency of protocols such as consensus [25] and Crusader Agreement [1].
Definition 19 (Longest Causal Chain).
Let be an event in and the set of messages received by , then is assigned round , where is the maximum round of an event originating a message in . If , then . The number of rounds in an execution becomes the highest round assigned to one of its events.
This metric, however, diverges from CR and NTR.
Example 20 (Reliable Broadcast).
In the reliable broadcast primitive [12], a dedicated source broadcasts a message and, if the source is correct, then all correct nodes should deliver the message. Furthermore, if a correct process delivers a message, then every correct process eventually delivers it. The following protocol satisfies this property:
-
When the source invokes broadcast(), it delivers and sends it to everyone;
-
When a process receives for the first time, it delivers and sends it to everyone.
In Figure 3, we depict an execution of this protocol with four processes: , , and . Here, is the source and broadcasts , the message is received by which then sends to everyone. Process receives from before receiving it from , and finally, receives from in the last event. This execution has LCC rounds, while having round according to CR and NTR.
Example 20 shows that the LCC metric diverges from the others in cases where a fast exchange of messages happens in the interval of one slow message. This is the case for several ASO protocols in the literature (including ours) which heavily rely on relaying values to speed up the validation phase, making the metric unsuitable for our use case. On the other hand, CR and NTR provide equivalent results in covered executions, described next.777The Round and CR metrics also provide equivalent results in covered executions [11].
6.2 Covered Executions and Holes
Consider an execution illustrated in Figure 4(a) where no process receives a message from another process, i.e., events may add messages to the buffer but no event removes a message from it. is not defined in any time assignment .
Now consider an execution in which:
-
A message is sent in and received in ;
-
A message is sent in and received in ;
-
No message from is received in .
In this example, illustrated in Figure 4(b) with events, exists for any time assignment of , but we can still assign an arbitrary time difference to and without affecting , which results in the number of CR rounds to be unbounded.
The two executions in the examples above have events whose time difference is unrelated to message delays. By consequence, the duration of these executions can grow irrespective of any bound imposed by message exchanges. Similarly, in Figure 4(b), since there is no message being received in from , there is no round assignment defined when using NTR to , and .
We then restrict the analysis of these metrics to executions that are covered. Formally:
Definition 21 (Covered Execution).
A hole in an execution is a pair in which no event in receives a message from , in other words, there are no message hops among the two sequence of events. An execution is covered iff it has no holes.
Abraham et al. [2] introduce NTR as an equivalent to CR, however, no formal proof is provided. The next result corroborates this claim in covered executions. Later in Example 24, we show that using NTR naively to measure time between events may not match CR.
Theorem 22.
A finite covered execution has CR rounds iff it has NTR rounds.
6.3 Time Between Arbitrary Events
In long-lived executions (such as those of atomic snapshot algorithms) we are interested in measuring time between two events, for instance, between an application call and response. Definition 15 can easily be adapted to measure the number of rounds between two events as follows:
Definition 23 (Generalized CR metric).
Let be an execution, let denote the set of all timed executions based on , and - the maximum message delay in . Let and () be events in , and and time assignments in for them respectively. Then we say that in between and there are: CR rounds.
An appealing way of defining time between two events and using a non-timed metric is to assign rounds according to NTR, and then take the difference of rounds assigned to and . As illustrated in Example 24, this definition can diverge from generalized CR.
Example 24.
Consider the execution shown in Figure 5. We can assign times to , and such that the two message hops have delay of . Now consider the number of rounds between and , since we can assign a time for that is arbitrarily close to ’s assignment, there are CR rounds between and . However, the round assignments using NTR to and are and respectively, so simply taking the difference between them leads to a value that diverges from CR.
We then give the following definition, using the approach described in Section 4.3:
Definition 25 (Generalized NTR).
Given an execution , let and () be events in . The number of rounds between and is given by the round assigned to according to the following:
-
All events up to (and including) are assigned round . We also write ;
-
For any , let be the last event where a message of round is delivered. All events after until (and including) event are in round .
6.4 Relating IRA to NTR
Theorem 27.
Let be a finite covered execution and suppose that all events of are assigned rounds according to IRA after all iterations of the algorithm. It holds that:
-
1.
Round is composed only of (the initial event).
-
2.
The final event of round is the last event to receive a message from round .
Corollary 28.
IRA and NTR assign the same rounds to events in covered executions.
References
- [1] Ittai Abraham, Naama Ben-David, Gilad Stern, and Sravya Yandamuri. On the round complexity of asynchronous crusader agreement. Cryptology ePrint Archive, 2023.
- [2] Ittai Abraham, Kartik Nayak, Ling Ren, and Zhuolun Xiang. Good-case latency of byzantine broadcast: a complete categorization. CoRR, abs/2102.07240, 2021. arXiv:2102.07240.
- [3] Yehuda Afek, Hagit Attiya, Danny Dolev, Eli Gafni, Michael Merritt, and Nir Shavit. Atomic snapshots of shared memory. J. ACM, 40(4):873–890, 1993. doi:10.1145/153724.153741.
- [4] James Aspnes, Hagit Attiya, Keren Censor-Hillel, and Faith Ellen. Limited-use atomic snapshots with polylogarithmic step complexity. J. ACM, 62(1):3:1–3:22, 2015. doi:10.1145/2732263.
- [5] James Aspnes and Keren Censor-Hillel. Atomic snapshots in o(log3 n) steps using randomized helping. In Yehuda Afek, editor, Distributed Computing - 27th International Symposium, DISC 2013, Jerusalem, Israel, October 14-18, 2013. Proceedings, volume 8205 of Lecture Notes in Computer Science, pages 254–268. Springer, 2013. doi:10.1007/978-3-642-41527-2_18.
- [6] Hagit Attiya, Amotz Bar-Noy, and Danny Dolev. Sharing memory robustly in message-passing systems. J. ACM, 42(1):124–142, January 1995. doi:10.1145/200836.200869.
- [7] Hagit Attiya, Faith Ellen, and Panagiota Fatourou. The complexity of updating snapshot objects. J. Parallel Distributed Comput., 71(12):1570–1577, 2011. doi:10.1016/J.JPDC.2011.08.002.
- [8] Hagit Attiya, Maurice Herlihy, and Ophir Rachman. Atomic snapshots using lattice agreement. Distributed Comput., 8(3):121–132, 1995. doi:10.1007/BF02242714.
- [9] Hagit Attiya and Jennifer Welch. Distributed computing: fundamentals, simulations, and advanced topics, volume 19. John Wiley & Sons, 2004.
- [10] Hagit Attiya and Jennifer L Welch. Multi-valued connected consensus: A new perspective on crusader agreement and adopt-commit. In 27th International Conference on Principles of Distributed Systems, 2024.
- [11] João Paulo Bezerra, Luciano Freitas, and Petr Kuznetsov. Asynchronous latency and fast atomic snapshot. arXiv preprint arXiv:2408.02562, 2024.
- [12] Christian Cachin, Rachid Guerraoui, and Luís Rodrigues. Introduction to reliable and secure distributed programming. Springer Science & Business Media, 2011.
- [13] Ran Canetti and Tal Rabin. Fast asynchronous byzantine agreement with optimal resilience. In Proceedings of the twenty-fifth annual ACM symposium on Theory of computing, pages 42–51, 1993. doi:10.1145/167088.167105.
- [14] K. Mani Chandy and Leslie Lamport. Distributed snapshots: Determining global states of distributed systems. ACM Trans. Comput. Syst., 3(1):63–75, 1985. doi:10.1145/214451.214456.
- [15] George Danezis, Lefteris Kokoris-Kogias, Alberto Sonnino, and Alexander Spiegelman. Narwhal and tusk: a dag-based mempool and efficient BFT consensus. In EuroSys, pages 34–50. ACM, 2022. doi:10.1145/3492321.3519594.
- [16] Carole Delporte-Gallet, Hugues Fauconnier, Sergio Rajsbaum, and Michel Raynal. Implementing snapshot objects on top of crash-prone asynchronous message-passing systems. IEEE Transactions on Parallel and Distributed Systems, 29(9):2033–2045, 2018. doi:10.1109/TPDS.2018.2809551.
- [17] Jose M. Faleiro, Sriram Rajamani, Kaushik Rajan, G. Ramalingam, and Kapil Vaswani. Generalized lattice agreement. In Proceedings of the 2012 ACM Symposium on Principles of Distributed Computing, PODC ’12, pages 125–134, New York, NY, USA, 2012. Association for Computing Machinery. doi:10.1145/2332432.2332458.
- [18] Vijay Garg, Saptaparni Kumar, Lewis Tseng, and Xiong Zheng. Amortized constant round atomic snapshot in message-passing systems. arXiv preprint arXiv:2008.11837, 2020. arXiv:2008.11837.
- [19] Vijay K. Garg, Saptaparni Kumar, Lewis Tseng, and Xiong Zheng. Fault-tolerant snapshot objects in message passing systems. In 2022 IEEE International Parallel and Distributed Processing Symposium (IPDPS), pages 1129–1139, 2022. doi:10.1109/IPDPS53621.2022.00113.
- [20] Maurice Herlihy and Jeannette M. Wing. Linearizability: A correctness condition for concurrent objects. ACM Trans. Program. Lang. Syst., 12(3):463–492, 1990. doi:10.1145/78969.78972.
- [21] Damien Imbs, Achour Mostéfaoui, Matthieu Perrin, and Michel Raynal. Set-constrained delivery broadcast: Definition, abstraction power, and computability limits. In Proceedings of the 19th International Conference on Distributed Computing and Networking, pages 1–10, 2018. doi:10.1145/3154273.3154296.
- [22] Idit Keidar, Eleftherios Kokoris-Kogias, Oded Naor, and Alexander Spiegelman. All you need is DAG. In PODC, pages 165–175. ACM, 2021. doi:10.1145/3465084.3467905.
- [23] Petr Kuznetsov, Thibault Rieutord, and Sara Tucci-Piergiovanni. Reconfigurable Lattice Agreement and Applications. In Pascal Felber, Roy Friedman, Seth Gilbert, and Avery Miller, editors, 23rd International Conference on Principles of Distributed Systems (OPODIS 2019), volume 153 of Leibniz International Proceedings in Informatics (LIPIcs), pages 31:1–31:17, Dagstuhl, Germany, 2020. Schloss Dagstuhl – Leibniz-Zentrum für Informatik. doi:10.4230/LIPIcs.OPODIS.2019.31.
- [24] Leslie Lamport. Time, clocks, and the ordering of events in a distributed system. Communications, 1978.
- [25] Leslie Lamport. Lower bounds for asynchronous consensus. Distributed Computing, 19:104–125, 2006. doi:10.1007/S00446-006-0155-X.
- [26] Friedemann Mattern. Efficient algorithms for distributed snapshots and global virtual time approximation. J. Parallel Distributed Comput., 18(4):423–434, 1993. doi:10.1006/JPDC.1993.1075.
- [27] Xiong Zheng, Vijay K. Garg, and John Kaippallimalil. Linearizable Replicated State Machines With Lattice Agreement. In Pascal Felber, Roy Friedman, Seth Gilbert, and Avery Miller, editors, 23rd International Conference on Principles of Distributed Systems (OPODIS 2019), volume 153 of Leibniz International Proceedings in Informatics (LIPIcs), pages 29:1–29:16, Dagstuhl, Germany, 2020. Schloss Dagstuhl – Leibniz-Zentrum für Informatik. doi:10.4230/LIPIcs.OPODIS.2019.29.
Appendix A Linearizability of Algorithm 1
The history of an execution is the subsequence of consisting of invocations and responses of ASO operations (update and snapshot). A history is sequential if each of its invocations is followed by a matching response. An execution is linearizable if, to each of its operation (update or snapshot, except, possibly, for incomplete ones), we can assign an indivisible point within its interval (called a linearization point), so that the operations put in the order of its linearizaton points constitute a legal sequential history of ASO (called a linearization), i.e., every snapshot operation returns a vector where every position contains the last value written to it (using an update operation), or the initial value if there are no such prior updates. Equivalently, a linearizable execution with history should have a linearization , a legal sequential history that (1) no node can locally distinguish a completion of and and (2) respect the real-time order of , i.e., if operation completes before operation in , then cannot precede in .
We show that every execution of Algorithm 1 is linearizable. Consider an execution of Algorithm 1, let be its history. Every operation (snapshot or update) is associated with a unique sequence number and performs a Propose operation on the LA object. If there is an operation that returns in position , by Validity of LA, there is an operation executed by node with sequence number that started before the completed and invoked a . In this case, we say that the update operation is successful. Notice that by Validity of LA, the update must have invoked with a vector containing in position .
Now we order complete snapshot operations and complete successful update operations in the order of the values returned by their operations (by Consistency of LA, these values are totally ordered. As each of these returns a value containing its unique sequence number (Stability of LA) , this order respects the real-time order of . A successful update operation performed by node with in position that has no complete is placed right before the first snapshot whose returns this value. By construction, the resulting sequential history is legal and locally indistinguishable from a completion of .
Finally, Liveness implies that every operation invoked by a correct process eventually completes.
Appendix B Time Complexity of Algorithm 2
We establish the optimality of our protocol under no-contention. A protocol implementing LA tolerates faults if it satisfies all the properties of Definition 1 in every execution with at most faulty processes.
Theorem 29.
Let be a distributed protocol that implements LA and tolerates at least one faulty process. Then, there exists a fault-free run of in which an LA operation requires at least two rounds of communication to complete without contention.
Proof.
Consider an operation op initiated by node , with call event and response event . Suppose op completes in at most one round in fault-free, contention-free executions.
We first show that there exists an execution such that:
-
is the only process to take a step in ,
-
no message sent by in is received by any other process before .
If multiple processes perform steps in the same event , we can conceptually “split” into a sequence of events , where each process takes the step in its own dedicated event. Since their steps are independent, these split events are indistinguishable from the original from each process’s perspective. This reasoning also applies to .
Now, assume for the sake of contradiction that in every fault-free, contention-free execution containing both and , there exists some process that receives a message –sent by in the interval –before occurs.
Let denote the event where receives . We define rounds from ’s perspective:
-
all events up to are in round ,
-
round ends at the last event that receives a message originating in round .
If is sent after , then we can construct so that all messages from round are received before . This ensures that occurs after , meaning is in round 2. Since occurs after , it too is assigned round 2–contradicting our assumption that op completes in one round.
If instead is sent in , we can again construct the execution so that all round messages are received before or at the same time as , making . Since occurs after , it is again assigned to round 2–a contradiction.
These contradictions hold regardless of whether op is concurrent with any other operation. Hence, such an execution must exist. Now consider an extension of where all messages sent by after (and including) are indefinitely delayed, while messages from other nodes are not.
Suppose a node invokes a new operation after , making non-concurrent with op. Since protocol tolerates at least one faulty process, and appears to have crashed in , node must eventually complete without any process receiving any messages from .
Let and be the value proposed and the value learned by in op, and let and be the corresponding values for in . By Validity, we know , and by Consistency, we know , hence .
However, since no process receives a message from since , no one could have known about , contradicting the requirement that must contain .
Finally, after completes, we can allow all delayed messages from to be received, making all processes correct in the final execution . This completes the proof.
Theorem 30.
An operation completes in at most rounds in fault-free runs w/o contention.
Proof.
Consider a contention-free request with call event and return event invoked by a node . There are no call events for other nodes between and , but some messages from previous proposals may still be in transit.
Suppose is the value to be proposed for the application call. If is not proposing (has ) when it receives , then it directly sends to everyone. Let be the last event in which a process receives from , then every process also sends by at most . Now take as the final event in which a process receives in the execution, and as the corresponding sending event. It must be that happens between and (potentially including) . Also, because the channels are FIFO, every previous proposal must have been validated before , and will learn a value containing by at most . Let be assigned round , then happens at most in round . As a consequence, is assigned either or , thus can be assigned at most round . Then, by the end of round , already has validated.
Now suppose that is proposing when it receives , so it still has a value in that is not validated, w.l.o.g. assume that is the only one. This value must be from a call that already finished, and the corresponding node sent containing before . Consider two pairs of events: and . In the first pair, is the event where was first sent, and is the last event in which is received from . In the second, is the usual application call event and is the last event in which is received from . There are two cases to consider: 1) happens before and 2) happens before .
If it is the first case, then at the moment happens, every node was already able to propose (since there was no other value to be learned). Take the last event in which a (or a value containing ) is received, and as the corresponding sending event, it follows that validates by at most and can learn a value containing it. Let be assigned round , and can be assigned at most round , and since receives a message from , it can be assigned at most round . If it is the second case, then all nodes received and put in before . Every node proposes by at most (since they can adopt and stop any current proposal). Let be the last event in which a process receives a proposal for and it’s corresponding sending event, similarly to the above cases, happens between and . Now, let and be assigned round . can be assigned at most round ( happens before or at ) and at most , which concludes the proof.
Lemma 31.
Consider an event in which a correct node sends and the first event in which a correct node learns a value including . If no correct node receives a message from a faulty one between these two events, then there are at most rounds between them.
Proof.
A message sent by a correct node is received by every correct node in the execution, and since correct nodes do not receive messages from faulty ones in the interval we are analyzing, we can consider only events originated from correct nodes. Therefore, we only refer to correct nodes in the following.
Let be the node sending , be the corresponding event and the last event a node receives from . Because also sends , by every node received the request and must be proposing. Any value learned after contains since all nodes have in .
Now, at the configuration just after applying , let be the set in which satisfies: there exists a (correct) node where is in but is not yet validated. Consider a value that is the last whose is received by any node, where is the event in which is last received and the corresponding sending event. It follows that some node learns a value containing by at most .
Next, take the first event in which a node sent , and the event in which the last from is received. Note that happens at most at and at most at . Let be assigned round , then (and thus ) can be assigned at most round , (and thus ) at most and lastly, can be assigned at most round . Therefore, there are at most rounds between a propose and the first learn event for .
Theorem 32.
An operation op takes at most rounds to complete if, during its interval, no correct node receives a message from a faulty one.
Proof.
Let be the value received from the application call for op, be the event in which node proposes (or a value containing ) and the event in which a value including is learned for the first time. From Lemma 31, there are at most rounds between and . Since the node that learns sends to everyone, receives and adopts it in one extra round. We conclude that in at most rounds every correct node can learn .
If is already proposing a value when it receives a call for , it sends to everyone and put it in , so it is proposed next. Let be the event in which initiated its previous proposal to , and consider the worst case where the application call with happens just after . From to the event in which learns its previous proposal (and thus starts proposing ), there are at most rounds, and from to the learning event of there are also at most rounds. Therefore, the operation completes in at most rounds.
Theorem 33.
An operation op takes rounds to complete, where is the number of active faulty nodes during op.
Proof.
We show that an operation op in Algorithm 2 takes rounds to complete, where is the number of active faulty nodes during op.
Messages from and to faulty nodes may not arrive, however, a message sent by (and to) a faulty node at round is received at most by round . Moreover, since channels are FIFO, when a node receives a message from another node , must also have received all previous messages sent to , irrespective of them being correct or faulty.
If a correct node receives (even from a faulty node) in round , every correct node will have added to by the end of round , and will have validated by the end of round . Also, faulty nodes wait for its current proposal to finish before starting a new one, in which case they send an ACCEPT message for the last learned value before sending the new proposal.
We say that a node introduces a new value during the operation if it is the first node to send a for in the interval of the operation. A node can introduce a new value coming from an internal source, i.e., the value was buffered and proposed when the node had already finished its previous proposal, or from an external source, i.e., after receiving a proposal originated from another node before the operation started.
Let be the value received from the application call for op and (as well as all previous events) be assigned round . If there are no active faulty nodes, a correct node learns a value containing by at most round (by Lemma 31, here, we include the time can remain buffered). Also by the end of round , every correct node has sent a PROPOSE message for and has validated by the end of round (including buffering time, a correct node proposes in round at the latest). By that point, all correct nodes are waiting for their proposals to complete and, therefore, cannot introduce a value from an internal source. In order to delay a correct node from leaning a value containing by round , every correct node should receive a new value in a PROPOSE message before, which is added to but is not validated. Using a simple inductive argument, new proposals originated from faulty nodes are necessary to delay a correct node from learning a value from round to round .
Suppose that there is an execution where it takes rounds for node to complete an operation. But there are only active faulty nodes, which means that at least extra proposals were introduced by active faulty nodes.
Let be an active faulty node that introduced more than one of the values that delayed the operation (assuming w.l.o.g. that there are exactly new proposals). Let and be the first and the second values introduced by respectively. If was received by from an internal source, should have finished its previous proposal (and learned a value containing ) before proposing . But because was one of the values that delayed the operation, and since channels are FIFO, needs to add to before validating (at least a majority of correct nodes sent a PROPOSE for before sending a PROPOSE for ). then learns a value containing and sends ACCEPT with that value to everyone. The ACCEPT message is received by correct processes before , and they would be able to adopt it.
So must have received from an external source at most by round , which means it issued proposals for that can be received by at most round . We can also conclude that at least values were introduced by active faulty nodes from external sources. Now let be the th such value used to delay correct nodes from learning . The earliest round can delay is , which means that by round all correct nodes already sent a propose for , but by the end of round no correct node has done it (otherwise would have been validated in round by every correct process). Take the first active faulty node from which a correct node received . Since the earliest this message is received is in round , the earliest it could be sent is in round , so first received from another distinct active faulty node, , which sent it in round the earliest. But was introduced from an external source and it needs to be received by a faulty node at round . Following the chain above, for the node to receive it in round , there would be necessary a chain of active nodes, although there are only .
Therefore, an operation takes less than rounds to complete.
