Damien Imbs

Achour Mostéfaoui

Matthieu Perrin

Michel Raynal

Set-Constrained Delivery Broadcast: a Communication Abstraction for Read/Write Implementable Distributed Objects

Keywords: Abstraction, Asynchronous system, Communication abstraction, Communication pattern, Conflict-free replicated data type, Design simplicity, Distributed task, Linearizability, Messagepassing system, Process crash, Read/write atomic register, Sequential consistency, Snapshot object

This paper introduces a new communication abstraction, called Set-Constrained Delivery Broadcast (SCD-broadcast), whose aim is to provide its users with an appropriate abstraction level when they have to implement objects or distributed tasks in an asynchronous message-passing system prone to process crash failures. This abstraction allows each process to broadcast messages and deliver a sequence of sets of messages in such a way that, if a process delivers a set of messages including a message m and later delivers a set of messages including a message m , no process delivers first a set of messages including m and later a set of message including m.

After having presented an algorithm implementing SCD-broadcast, the paper investigates its programming power and its computability limits. On the "power" side it presents SCD-broadcast-based algorithms, which are both simple and efficient, building objects (such as snapshot and conflict-free replicated data types), and distributed tasks. On the "computability limits" side it shows that SCDbroadcast and read/write registers are computationally equivalent.

Introduction

Programming abstractions. Informatics is a science of abstractions, and a main difficulty consists in providing users with a "desired level of abstraction and generality -one that is broad enough to encompass interesting new situations, yet specific enough to address the crucial issues" as expressed in [START_REF] Fischer | Appraising two decades of distributed computing theory research[END_REF]. When considering sequential computing, functional programming and object-oriented programming are well-know examples of what means "desired level of abstraction and generality".

In the context of asynchronous distributed systems where the computing entities (processes) communicate -at the basic level-by sending and receiving messages through an underlying communication network, and where some of them can experience failures, a main issue consists in finding appropriate communication-oriented abstractions, where the meaning of the term "appropriate" is related to the problems we intend to solve. Solving a problem at the send/receive abstraction level is similar to the writing of a program in a low-level programming language. Programmers must be provided with abstractions that allow them to concentrate on the problem they solve and not on the specific features of the underlying system. This is not new. Since a long time, high level programming languages have proved the benefit of this approach. From a synchronization point of view, this approach is the one promoted in software transactional memory [START_REF] Shavit | Software transactional memory[END_REF], whose aims is to allow programmers to focus on the synchronization needed to solve their problems and not on the way this synchronization must be implemented (see the textbooks [START_REF] Herlihy | The Art of Multiprocessor Programming[END_REF][START_REF] Raynal | Concurrent programming: algorithms, principles and foundations[END_REF]).

If we consider specific coordination/cooperation problems, "matchings" between problems and specific communication abstractions are known. One of the most famous examples concerns the consensus problem whose solution rests on the total order broadcast abstraction (also called atomic broadcast). Another "matching" example is the causal message delivery broadcast abstraction [START_REF] Birman | Reliable communication in the presence of failures[END_REF][START_REF] Raynal | The causal ordering abstraction and a simple way to implement it[END_REF], which allows for a very simple implementation of a causal read/write memory [START_REF] Ahamad | Causal memory: definitions, implementation and programming[END_REF] 1 .

Aim of the paper. The aim of this paper is to introduce and investigate a high level communication abstraction which allows for simple and efficient implementations of concurrent objects and distributed tasks, in the context of asynchronous message-passing systems prone to process crash failures. The concurrent objects in which we are interested are defined by a sequential specification [START_REF] Herlihy | Linearizability: a correctness condition for concurrent objects[END_REF] (e.g., a queue). Differently, a task extends to the distributed context the notion of a function [START_REF] Biran | A combinatorial characterization of the distributed tasks which are solvable in the presence of one faulty processor[END_REF][START_REF] Moran | Extended impossibility results for asynchronous complete networks[END_REF]. It is defined by a mapping from a set of input vectors to a set of output vectors, whose sizes are the number of processes. An input vector I defines the input value I[i] of each process p i , and, similarly, an output vector O defines the output O[j] of each process p j . Agreement problems such as consensus and k-set agreement are distributed tasks. What makes the implementation of a task difficult is the fact that each process knows only its input, and, due to net effect of asynchrony and process failures, no process can distinguish if another process is very slow or crashed. The difficulty results in an impossibility for consensus [START_REF] Fischer | Impossibility of distributed consensus with one faulty process[END_REF], even in a system in which at most one process may crash.

A new broadcast abstraction. The Set-Constrained Delivery broadcast (in short SCD-broadcast) communication abstraction proposed in the paper allows a process to broadcast messages, and to deliver sets of messages (instead of a single message) in such a way that, if a process p i delivers a message set mset containing a message m, and later delivers a message set mset containing a message m , then no process p j can deliver first a set containing m and later another set containing m. Let us notice that p j is not prevented from delivering m and m in the same set. Moreover, SCD-broadcast imposes no constraint on the order in which a process must process the messages it receives in a given message set.

After having defined SCD-broadcast, the paper presents an implementation of it in asynchronous systems where a minority of processes may crash. This assumption is actually a necessary and sufficient condition to cope with the net effect of asynchrony and process failures (see below). The SCD-broadcast of an application message generates O(n 2 ) implementation messages, and assuming an upper bound ∆ on message transfer delays and zero processing time, its time complexity is upper bounded by 2∆ time units (by "time complexity" we mean the time elapsed between the SCD-broadcast of an application message m and the latest time at which m is locally delivered at the application layer by all the processes that have not crashed).

Implementing objects and tasks. Then, the paper addresses two fundamental issues of SCD-broadcast: its abstraction power and its computability limits. As far as its abstraction power is concerned, i.e., its ability and easiness to implement atomic (linearizable) or sequentially consistent concurrent objects [START_REF] Herlihy | Linearizability: a correctness condition for concurrent objects[END_REF][START_REF] Lamport | On interprocess communication, Part I: basic formalism[END_REF] and read/write solvable distributed tasks, the paper presents, on the one side, two algorithms implementing atomic objects (namely a snapshot object [START_REF] Afek | Atomic snapshots of shared memory[END_REF][START_REF] Anderson | Multi-writer composite registers[END_REF], and a distributed increasing/decreasing counter), and, on the other side, an algorithm solving the lattice agreement task [START_REF] Attiya | Atomic snapshots using lattice agreement[END_REF][START_REF] Faleiro | Generalized lattice agreement[END_REF].

The two concurrent objects (snapshot and counter) have been chosen because they are encountered in many applications, and are also good representative of the class of objects identified in [START_REF] Aspnes | Wait-free data structures in the asynchronous PRAM model[END_REF]. The objects of this class are characterized by the fact that each pair op1 and op2 of their operations either commute (i.e., in any state, executing op1 before op2 leads to the same state as executing op2 before op1, as it is the case for a counter), or any of op1 and op2 can overwrite the other one (e.g., executing op1 before op2 leads to the same state as executing op2 alone). Our implementation of a counter can be adapted for all objects with commutative operations, and our implementation of the snapshot object illustrates how overwriting operations can be obtained directly from the SCD-broadcast abstraction. Concerning these objects, it is also shown that a slight change in the algorithms allows us to obtain implementations (with a smaller cost) in which the consistency condition is weakened from linearizability to sequential consistency [START_REF] Lamport | How to make a multiprocessor computer that correctly executes multiprocess programs[END_REF].

In the case of read/write solvable tasks, SCD-broadcast shows how the concurrency inherent (but hidden) in a task definition can be easily mastered and solved.

A distributed software engineering dimension. All the algorithms presented in the paper are based on the same communication pattern. As far as objects are concerned, the way this communication pattern is used brings to light two genericity dimensions of the algorithms implementing them. One is on the variety of objects that, despite their individual features (e.g., snapshot vs counter), have very similar SCD-broadcast-based implementations (actually, they all have the same communication pattern-based structure). The other one is on the consistency condition they have to satisfy (linearizability vs sequential consistency). On programming languages for distributed computing. Differently from sequential computing for which there are plenty of high level languages (each with its idiosyncrasies), there is no specific language for distributed computing. Instead, addressing distributed settings is done by the enrichment of sequential computing languages with high level communication abstractions. When considering asynchronous systems with process crash failures, total order broadcast is one of them. SCD-broadcast is a candidate to be one of them, when one has to implement read/write solvable objects and distributed tasks.

The computability limits of SCD-broadcast. The paper also investigates the computability power of the SCD-broadcast abstraction, namely it shows that SCD-broadcast and atomic read/write registers (or equivalently snapshot objects) have the same computability power in asynchronous systems prone to process crash failures. Everything that can be implemented with atomic read/write registers can be implemented with SCD-broadcast, and vice versa.

As read/write registers (or snapshot objects) can be implemented in asynchronous message-passing systems where only a minority of processes may crash [START_REF] Attiya | Sharing memory robustly in message passing systems[END_REF], it follows that the proposed algorithm implementing SCD-broadcast is resilience-optimal in these systems. From a theoretical point of view, this means that the consensus number of SCD-broadcast is 1 (the weakest possible).

Roadmap. The paper is composed of 10 sections. Section 2 defines the SCD-broadcast abstraction and the associated communication pattern used in all the algorithms presented in the paper. Section 4 presents a resilience-optimal algorithm implementing SCD-broadcast in asynchronous message-passing systems prone to process crash failures, while Section 5 adopts a distributed software engineering point of view and presents a communication pattern associated with SCD-broadcast. Then, Sections 6-8 present SCD-broadcast-based algorithms for concurrent objects and tasks. Section 9 focuses on the computability limits of SCD-broadcast. Finally, Section 10 concludes the paper.

Remark. It is worth noticing that the a self-stabilizing version of the SCD algorithm presented in this article is described in [START_REF] Lundström | Self-stabilizing set-constrained delivery broadcast[END_REF], and an extension of it where up to t < n/4 processes may commit Byzantine failures is described in [START_REF] Auvolat | Byzantine-tolerant set-constrained delivery broadcast[END_REF]. It is still an open problem to find an SCD algorithm coping with up to t < n/3 Byzantine processes.

The SCD-broadcast Communication Abstraction

Process model. The computing model is composed of a set of n asynchronous sequential processes, denoted p 1 , ..., p n . "Asynchronous" means that each process proceeds at its own speed, which can be arbitrary and always remains unknown to the other processes.

A process may halt prematurely (crash failure), but it executes its local algorithm correctly until it crashes (if it ever does). The model parameter t denotes the maximal number of processes that may crash in a run r. A process that crashes in a run is said to be faulty in r. Otherwise, it is non-faulty.

Definition of SCD-broadcast. The set-constrained delivery broadcast abstraction (SCD-broadcast) provides the processes with one operation scd_broadcast(m) and one event scd_deliver(mset). The operation takes a message m to broadcast as input parameter. When the event is triggered at a process p i , it dispenses a non-empty set of messages mset to p i . Using a classical terminology, when a process invokes the operation scd_broadcast(m), we say that it "scd-broadcasts a message m". Similarly, when scd_deliver(mset) is triggered at a process, we say that it "scd-delivers the set of messages mset". By a slight abuse of language, when we are interested in a message m, we say that a process "scd-delivers the message m" when actually it scd-delivers a message set mset containing m.

SCD-broadcast is defined by the following set of properties, where we assume -without loss of generality-that all the messages that are scd-broadcast are different.

• Validity. If a process scd-delivers a set containing a message m, then m was scd-broadcast by a process.

• Integrity. A message is scd-delivered at most once by each process.

• MS-Ordering. Let p i be a process that scd-delivers first a message set mset i and later a message set mset i . For any pair of messages m ∈ mset i and m ∈ mset i , no process p j scd-delivers first a message set mset j containing m and later a message set mset j containing m.

• Termination-1. If a non-faulty process scd-broadcasts a message m, it terminates its scd-broadcast invocation and scd-delivers a message set containing m.

• Termination-2. If a process scd-delivers a message m, every non-faulty process scd-delivers a message set containing m.

Termination-1 and Termination-2 are classical liveness properties (found for example in Uniform Reliable Broadcast [START_REF] Attiya | Distributed computing: fundamentals, simulations and advanced topics[END_REF][START_REF] Raynal | Fault-tolerant message-passing distributed systems: an algorithmic approach[END_REF]). The other ones are safety properties. Validity and Integrity are classical communication-related properties. The first states that there is neither message creation nor message corruption, while the second states that there is no message duplication.

The MS-Ordering property is new, and characterizes SCD-broadcast. It states that the contents of the sets of messages scd-delivered at any two processes are not totally independent: the sequence of sets scd-delivered at a process p i and the sequence of sets scd-delivered at a process p j must be mutually consistent in the sense that a process p i cannot scd-deliver first m ∈ mset i and later m ∈ mset i = mset i , while another process p j scd-delivers first m ∈ mset j and later m ∈ mset j = mset j . Let us nevertheless observe that if p i scd-delivers first m ∈ mset i and later m ∈ mset i , p j may scd-deliver m and m in the same set of messages.

Let us remark that, if the MS-Ordering property is suppressed and messages are scd-delivered one at a time, SCD-broadcast boils down to the well-known Uniform Reliable Broadcast abstraction [START_REF] Chandra | Unreliable failure detectors for reliable distributed systems[END_REF][START_REF] Raynal | Fault-tolerant message-passing distributed systems: an algorithmic approach[END_REF].

An example. Let m 1 , m 2 , m 3 , m 4 , m 5 , m 6 , m 7 and m 8 be messages that have been scd-broadcast by different processes. Processes deliver sets of messages and do not deliver more than one set at once; whereas there is no particular order within each set. The following scd-deliveries of message sets by p 1 , p 2 and p 3 respect the definition of SCD-broadcast:

• at p 1 : {m 1 , m 2 }, {m 3 , m 4 , m 5 }, {m 6 }, {m 7 , m 8 }. • at p 2 : {m 1 }, {m 2 , m 3 }, {m 4 , m 5 , m 6 }, {m 7 }, {m 8 }. • at p 3 : {m 1 , m 2 , m 3 }, {m 4 , m 5 , m 6 }, {m 7 }, {m 8 }.
Differently, due to the scd-deliveries of the sets including m 2 and m 3 , the following scd-deliveries by p 1 and p 2 do not satisfy the MS-broadcast property:

• at p 1 : {m 1 , m 2 }, {m 3 , m 4 , m 5 }, ... • at p 2 : {m 1 , m 3 }, {m 2 }, ...
A containment property. Let mset i denote the -th message set scd-delivered by p i . Hence, at some time, p i scd-delivered the sequence of message sets mset 1 i , . . . , mset x i . Let MSet x i = mset 1 i ∪ . . . ∪ mset x i . The following Containment property follows directly from the MS-Ordering and Termination-2 properties: ∀ i, j, x, y:

(MSet x i ⊆ MSet y j ) ∨ (MSet y j ⊆ MSet x i ).
Partial order on messages created by the message sets. The MS-Ordering and Integrity properties establish a partial order on the set of all the messages, defined as follows. Let → i be the local message delivery order at process p i defined as follows: m → i m if p i scd-delivers the message set containing m before the message set containing m . As no message is scd-delivered twice, it is easy to see that → i is a partial order (locally know by p i ). The containment property implies that there is a total order (which remains unknown to the processes) on the whole set of messages, that complies with the partial order →= ∪ 1≤i≤n → i . This is where SCD-broadcast can be seen as a weakening of total order broadcast.

Underlying communication network

Send/receive asynchronous network. Each pair of processes communicate through two uni-directional channels on which they send and receive messages. Hence, the communication network is a complete network: any process p i can directly send a message to any process p j (including itself). A process p i invokes the operation "send TYPE(m) to p j " to send to p j the message m, whose type is TYPE. The operation "receive TYPE() from p j " allows p i to receive from p j a message whose type is TYPE.

Each channel is reliable (no loss, corruption, nor creation of messages), not necessarily FIFO, and asynchronous (while the transit time of each message is finite, there is no upper bound on message transit times) Let us notice that, due to process and message asynchrony, no process can know if another process crashed or is only very slow.

Uniform FIFO-broadcast abstraction. To simplify the presentation, and without loss of generality, we consider that the system is equipped with a FIFO-broadcast abstraction. Such an abstraction can be built on top of the previous basic system model without enriching it with additional assumptions (see e.g. [START_REF] Raynal | Fault-tolerant message-passing distributed systems: an algorithmic approach[END_REF]). It is defined by an operation fifo_broadcast() and an event fifo_deliver(), which satisfy the properties of Uniform Reliable Broadcast (Validity, Integrity, Termination-1, and Termination-2, with the same definitions as in SCD-broadcast), plus the following message ordering property.

• FIFO-Order. For any pair of processes p i and p j , if p i fifo-delivers first a message m and later a message m , both from p j , no process fifo-delivers m before m.

An Implementation of SCD-broadcast

This section shows that the SCD-broadcast communication abstraction is not an oracle-like object (oracles allow us to extend our understanding of computing, but cannot be implemented). It describes an implementation of SCD-broadcast in an asynchronous send/receive message-passing system in which any minority of processes may crash. This system model is denoted CAMP n,t [t < n/2] (where CAMP n,t stands for "Crash Asynchronous Message-Passing" and t < n/2 is its restriction on failures). As t < n/2 is the weakest assumption on process failures that allows a read/write register to be built on top of an asynchronous message-passing system [5]2 , and SCD-broadcast and read/write registers are computationally equivalent (as shown in Sections 6 and 9), the proposed implementation is optimal from a resilience point of view.

Algorithm

This section describes Algorithm 1, which implements SCD-broadcast in CAMP n,t [t < n/2]. From a terminology point of view, an application message is a message that has been scd-broadcast by a process, while a protocol message is an implementation message generated by the algorithm.

Local metadata quadruplets. For each application message m, each process stores a quadruplet qdplt = qdplt.msg, qdplt.sd, qdplt.sn, qdplt.cl whose fields have the following meaning.

• qdplt.msg contains an application message m,

• qdplt.sd contains the id of the sender of qdplt.msg,

• qdplt.sn contains the local date (sequence number) associated with m by its sender. Hence, the pair qdplt.sd, qdplt.sn is the identity of the application message m, denoted ts(qdplt).

• qdplt.cl is an array of size n, initialized to [+∞, . . . , +∞]. Then, qdplt.cl[x] will contain the sequence number associated with m by p x when it broadcast the message FORWARD_MSG(msg.m, -, -, -, -). This last field is crucial in the scd-delivery by the process p i of a message set containing m.

Local variables at a process p i . Each process p i manages the following local variables.

• buffer i : buffer (initially empty) where the quadruplets containing messages that have been fifodelivered but not yet scd-delivered in a message set are stored.

• to_deliver i : set of quadruplets containing messages to be scd-delivered.

• sn i : local logical clock which takes the values 0, 1, 2, ..., that measure the local progress of p i . Each application message scd-broadcast by p i is identified by a pair i, sn , where sn is the current value of sn i .

• clock i [1.
.n]: array of clock values; clock i [j] is the greatest lock value x such that the application message m identified x, j has been scd-delivered by p i .

Protocol message. The algorithm uses a single type of protocol message denoted FORWARD_MSG(m, sd, sn, g, sn g ). Such a message is made up of five fields: the first field is an associated application message m, the second and third form a pair sd, sn that is the identity of the application message and the fourth and fifth form a pair g, sn g that describes the local progress (as captured by sn g ) of the forwarder process p g when it fifo-broadcast this protocol message to the other processes by invoking fifo_broadcast FORWARD_MSG(m, sd, sn sd , g, sn (line 11).

Operation scd_broadcast(). When p i invokes the operation scd_broadcast(m), where m is an application message, it executes the internal operation forward(m, i, sn i , i, sn i ), which initializes the algorithm, and waits until it has no more message from itself pending in buffer i , which means it has scd-delivered a set containing m (lines 19 and 20).

Uniform fifo-broadcast of a message FORWARD_MSG. When a process p i fifo-delivers a protocol message FORWARD_MSG(m, sd, sn sd , g, sn g ), it first invokes the internal operation forward(m, sd, sn sd , g, sn g ).

In addition to other statements, the first fifo-delivery of such a message by a process p i entails its participation in the uniform reliable fifo-broadcast of this message (lines 5 and 11). In addition to the invocation of forward(), the fifo-delivery of FORWARD_MSG() also invokes try_deliver(), which strives to scd-deliver a message set (lines 4).

The core of the algorithm. Expressed with the relations → i , 1 ≤ i ≤ n, introduced in Section 2, the main issue of the algorithm is to ensure that, if there are two message m and m and a process p i such that m → i m , then there is no p j such that m → j m. To this end, a process p i is allowed to scd-deliver a message m before a message m only if it knows that a majority of processes p j have fifo-delivered a protocol message FORWARD_MSG(m, -, -, -, -) before a protocol message FORWARD_MSG(m , -, -, -, -); p i knows it either (i) because it fifo-delivered from p j a message FORWARD_MSG(m, -, -, -, -) but not yet a message FORWARD_MSG(m , -, -, -, -), or (ii) because it fifo-delivered from p j both FORWARD_MSG(m, -, -, -, snm) and FORWARD_MSG(m , -, -, -, snm ) and the sending date smn is smaller than the sending date snm . The MS-Ordering property follows then from the impossibility that a majority of processes "sees m before m ", while another majority "sees m before m".

Internal operation forward(). This operation can be seen as an enrichment (with the fields g and sn g ) of the reliable fifo-broadcast implemented by the protocol messages FORWARD_MSG(m, sd, sn sd , -, -).

Considering such a message FORWARD_MSG(m, sd, sn sd , g, sn g ), m was scd-broadcast by p sd at its local time sn sd , and relayed by the forwarding process p g at its local time sn g . If sn sd ≤ clock i [sd], p i has already scd-delivered a message set containing m (see lines 18 and 20). If sn sd > clock i [sd], there are two cases defined by the predicate of line 6.

• No quadruplet qdplt in buffer i is such that qdplt.msg = m. In this case, p i creates a quadruplet associated with m, and adds it to buffer i (lines 8-10). Then, p i participates in the fifo-broadcast of m identified by sd, sn sd . (line 11) and records its local progress by increasing sn i (line 12). • There is a quadruplet qdplt in buffer i associated with m, i.e., qdplt = m, -, -, -∈ buffer i . In this case, p i assigns sn g to qdplt.cl[g] (line 7), thereby indicating that m was known and forwarded

operation scd_broadcast(m) is (1) forward(m, i, sni, i, sni); (2) wait( qdplt ∈ bufferi : qdplt.sd = i).
when the message FORWARD_MSG(m, sd, sn sd , g, sng ) is fifo-delivered do % from pg (3) forward(m, sd, sn sd , g, sng ); (4) try_deliver().

procedure forward(m, sd, sn sd , g, sng ) is

(5) if (sn sd > clocki[sd]) (6) then if (∃ qdplt ∈ bufferi : qdplt.sd = sd ∧ qdplt.sn = sn sd ) (7) then qdplt.cl[g] ← sng (8) else threshold[1..n] ← [∞, . . . , ∞]; threshold[g] ← sng ; (9) let qdplt ← m, sd, sn sd , threshold[1.
.n] ; [START_REF] Auvolat | Byzantine-tolerant set-constrained delivery broadcast[END_REF] bufferi ← bufferi ∪ {qdplt}; [START_REF] Biran | A combinatorial characterization of the distributed tasks which are solvable in the presence of one faulty processor[END_REF] fifo_broadcast FORWARD_MSG(m, sd, sn sd , i, sni);

(12) sni ← sni + 1 (13) end if (14) end if. procedure try_deliver() is (15) let to_deliveri ← {qdplt ∈ bufferi : |{f : qdplt.cl[f ] < ∞}| > n 2 }; (16) while (∃ qdplt ∈ to_deliveri , ∃ qdplt ∈ bufferi \ to_deliveri : |{f : qdplt.cl[f ] < qdplt .cl[f ]}| ≤ n 2 ) do to_deliveri ← to_deliveri \ {qdplt} end while; (17) if (to_deliveri = ∅) (18) then for each qdplt ∈ to_deliveri do clocki[qdplt.sd] ← max(clocki[qdplt.sd], qdplt.sn) end for; (19) bufferi ← bufferi \ to_deliveri ; (20) ms ← {m : ∃ qdplt ∈ to_deliveri : qdplt.msg = m}; scd_deliver(ms) (21) end if. Algorithm 1: An implementation of SCD-broadcast in CAMP n,t [t < n/2] (code for p i )
by p g at its local time sn g .

Internal operation try_deliver(). When a process p i executes try_deliver(), it first computes the set to_deliver i of the quadruplets qdplt containing application messages m which have been seen by a majority of processes (line 15). From p i 's point of view, a message has been seen by a process p g if qdplt.cl[g] has been set to a finite value (line 7).

As indicated in a previous paragraph, if a majority of processes received first a message FOR-WARD_MSG carrying m and later another message FORWARD_MSG carrying m, it might be that some process p j scd-delivered a set containing m before scd-delivering a set containing m. Therefore, p i must avoid scd-delivering a set containing m before scd-delivering a set containing m . This is done at line 16, where p i withdraws the quadruplet qdplt corresponding to m if it cannot deliver m yet (i.e. the corresponding qdplt is not in to_deliver i ) or it does not have the proof that the situation cannot happen, i.e. no majority of processes saw the message corresponding to qdplt before the message corresponding to qdplt (this is captured by the predicate |{g

: qdplt.cl[g] < qdplt .cl[g]}| ≤ n 2 )
. If to_deliver i is not empty after it has been purged (lines [START_REF] Delporte-Gallet | Linear space bootstrap communication schemes[END_REF][START_REF]How hard is it to take a snapshot?[END_REF], p i computes a message set to scddeliver. This set mset contains all the application messages in the quadruplets of to_deliver i (line 20). These quadruplets are withdrawn from buffer i (line 18). Moreover, before this scd-delivery, p i needs to updates clock i [x] for all the entries such that x = qdplt.sd where qdplt ∈ to_deliver i (line 18). This update is needed to ensure that the future uses of the predicate of line 17 are correct.

Cost and proof of correctness

Lemma 1 (Validity) If a process scd-delivers a message set containing m, some process cd-broadcast m. Proof If a process p i scd-delivers a set containing a message m, it previously added into buffer i a quadruplet qdplt such that qdplt.msg = m (line 10), for which it follows that it fifo-delivered a protocol message FORWARD_MSG(m, -, -, -, -). Due to the fifo-validity property, it follows that a process generated the fifo-broadcast of this message, which originated from an invocation of scd_broadcast(m).

2 Lemma 1

Lemma 2 (Integrity) No process scd-delivers the same message twice.

Proof Let us observe that, due to the wait statement at line 2, and the increase of sn i at line 15 between two successive scd-broadcast by a process p i , no two application messages can have the same identity i, sn . It follows that there is a single quadruplet m, i, sn,that can be added to buffer i , and this is done only once (line 10). Finally, let us observe that this quadruplet is suppressed from buffer i , just before m is scd-delivered (line [START_REF] Fischer | Impossibility of distributed consensus with one faulty process[END_REF][START_REF] Fischer | Appraising two decades of distributed computing theory research[END_REF], which concludes the proof of the lemma.

2 , so the previous case applies. After p j broadcasts its message FORWARD_MSG(m, sd, sn sd , j, sn j ) on line 11, there is a qdplt ∈ buffer j with ts(qdplt) = sd, sn sd , until it is removed on line 16 and clock j [sd] ≥ sn sd . Therefore, one of the conditions at lines 5 and 6 will stay false for the stamp ts(qdplt) and p j will never execute line 11 with the same stamp sd, sn sd later.

2 Lemma 3

Lemma 4 (MS-Ordering) Let p i be a process that scd-delivers a set ms i containing a message m and later scd-delivers a set ms i containing a message m . No process p j scd-delivers first a set ms j containing m and later a message set ms j containing m.

Proof Let us suppose there are two messages m and m and two processes p i and p j such that p i scddelivers a set ms i containing m and later scd-delivers a set ms i containing m and p j scd-delivers a set ms j containing m and later scd-delivers a set ms j containing m. When m is delivered by p i , there is an element qdplt ∈ buffer i such that qdplt.msg = m and because of line 15, p i has received a message FORWARD_MSG(m, -, -, -, -) from more than n 2 processes.

• If there is no element qdplt ∈ buffer i such that qdplt .msg = m , since m has not been delivered by p i yet, p i has not received a message FORWARD_MSG(m , -, -, -, -) from any process (lines 10 and 19). Hence, because the communication channels are FIFO, more than n 2 processes have sent a message FORWARD_MSG(m, -, -, -, -) before sending a message FORWARD_MSG(m , -, -, -, -).

• Otherwise, qdplt / ∈ to_deliver i after line 16. As the communication channels are FIFO, more than half of the processes have sent a message FORWARD_MSG(m, -, -, -, -) before a message FORWARD_MSG(m , -, -, -, -).

Using the same reasoning, it follows that when m is delivered by p j , more than n 2 processes have sent a message FORWARD_MSG(m , -, -, -, -) before sending a message FORWARD_MSG(m, -, -, -, -). There exists a process p k in the intersection of the two majorities, that has (a) sent FORWARD_MSG(m, -, -, -, -) before sending FORWARD_MSG(m , -, -, -, -) and (b) sent FORWARD_MSG(m , -, -, -, -) before sending a message FORWARD_MSG(m, -, -, -, -). However, it follows from Lemma 3 that p k can send a single message FORWARD_MSG(m , -, -, -, -) and a single message FORWARD_MSG(m, -, -, -, -), which leads to a contradiction.

2 Lemma 4 Proof Let p i be a non-faulty process. For any pair of messages qdplt and qdplt ever inserted in buffer i , let ts = ts(qdplt) and ts = ts(qdplt ). Let → i be the dependency relation defined as follows:

p i p g • • • scd_broadcast(m k ) • sng(k) scd_broadcast(m k1 ) • sng(k1) scd_broadcast(m k2 ) • sng(k2) FORWARD_MSG(m, sd, sn sd , -, -) • FORWARD_MSG(m l+1 , sd l+1 , sn sd l+1 , -, -) ••• m i m k1 i m k2 i
ts → i ts def = |{j : qdplt .cl[j] < qdplt.cl[j]}| ≤ n 2 (i.
e. the dependency does not exist if p i knows that a majority of processes have seen the first update -due to qdplt -before the second -due to qdplt). Let → i denote the transitive closure of → i .

Let us suppose (by contradiction) that the timestamp sd, sn sd associated with the message m (carried by the protocol message FORWARD_MSG(m, sd, sn sd , i, sn i ) fifo-broadcast by p i ), has an infinity of predecessors according to → i . As the number of processes is finite, an infinity of these predecessors have been generated by the same process, let us say p g . Let g, sn g (k) k∈N be the infinite sequence of the timestamps associated with the invocations of the scd_broadcast() issued by p g . The situation is depicted by Figure 1.

As p i is non-faulty, p g eventually receives a message FORWARD_MSG(m, sd, sn sd , i, sn i ), which means p g broadcast an infinity of messages FORWARD_MSG(m(k), g, sn g (k), g, sn g (k)) after having broadcast the message FORWARD_MSG(m, sd, sn sd , g, sn g ). Let g, sn g (k1) and g, sn g (k2) be the timestamps associated with the next two messages scd-broadcast by p g , with sn g (k1) < sn g (k2). By hypothesis, we have g, sn g (k2) → i sd, sn sd . Moreover, all processes received their first message FORWARD_MSG(m, sd, sn sd , -, -) before their first message FORWARD_MSG(m(k), g, sn g (k), -, -), so sd, sn sd → i g, sn g (k1) . Let us express the path g, sn g (k2) → i g, sn g (k1) : g, sn g (k2) = sd (1), sn (1)

→ i sd (2), sn (2) → i • • • → i sd(m), sn (m) = g, sn g (k1) .
In the time interval starting when p g sent the message FORWARD_MSG(m(k1), g, sn g (k1), g, sn g (k1)) and finishing when it sent the message FORWARD_MSG(m(k2), g, sn g (k2), g, sn g (k2)), the waiting condition of line 2 became true, so p g scd-delivered a set containing the message m(k1), and according to Lemma 1, no set containing the message m(k2). Therefore, there is an index l such that process p g delivered sets containing messages associated with a timestamp sd (l), sn (l) for all l > l but not for l = l. Because the channels are FIFO and thanks to lines 15 and 16, it means that a majority of processes have sent a message FORWARD_MSG(-, sd (l + 1), sn (l + 1), -, -) before a message FORWARD_MSG(-, sd (l), sn (l), -, -), which contradicts the fact that sd (l), sn (l) → i sd (l + 1), sn (l + 1) .

Let us suppose a non-faulty process p i has fifo-broadcast a message FORWARD_MSG(m, sd, sn sd , i, sn i ) (line 10). It inserted a quadruplet qdplt with timestamp sd, sn sd on line 9 and by what precedes, sd, sn sd has a finite number of predecessors sd 1 , sn 1 , . . . , sd l , sn l according to → i . As p i is nonfaulty, according to Lemma 3, it eventually receives a message FORWARD_MSG(-, sd k , sn k , -, -) for all 1 ≤ k ≤ l and from all non-faulty processes, which are in majority.

Let pred be the set of all quadruplets qdplt such that qdplt .sd, qdplt .sn → i sd, sn sd . Let us consider the moment when p i receives the last message FORWARD_MSG(-, sd k , sn k , g, sn g ) sent by a correct process p g . For all qdplt ∈ pred , either qdplt .msg has already been delivered or qdplt is inserted to_deliver i on line 15. Moreover, no qdplt ∈ pred will be removed from to_deliver i , on line 16, as the removal condition is the same as the definition of → i . In particular for qdplt = qdplt, either m has already been scd-delivered or m is present in to_deliver i on line 17 and will be scddelivered on line 20.

2 Lemma 5

Lemma 6 (Termination-1) If a non-faulty process scd-broadcasts a message m, it scd-delivers a message set containing m.

Proof If a non-faulty process p i scd-broadcasts a message m, it executes the procedure forward(m, i, sn i , i, sn i ) (Lines ( 22) and ( 23)). Similarly to Lemma 3, as no message FORWARD_MSG(m, i, sn i , g, sn g ) was previously broadcast, (sn i > clock i [i]) and there is no qdplt ∈ buffer i such that (qdplt.sd, qdplt.sn) = (i, sn i ). Therefore, p i fifo-broadcasts the message FORWARD_MSG(m, sd, sn sd , i, sn i ) at line 11).

Then, due to Lemma 5, it scd-delivers a message set containing m.

2 Lemma 6

Lemma 7 (Termination-2) If a process scd-delivers a message m, every non-faulty process scd-delivers a message set containing m.

Proof Let p i be a process p i that scd-delivers a message m. At line 20, there is a quadruplet qdplt ∈ to_deliver i such that qdplt.msg = m. At line 15, qdplt ∈ buffer i , and qdplt was inserted in buffer i at line 10, just before p i fifo-broadcast the message FORWARD_MSG(m, sd, sn sd , i, sn i ). By Lemma 3, every non-faulty process p j sends a message FORWARD_MSG(m, sd, sn sd , j, sn j ), so by Lemma 5, p j scd-delivers a message set containing m. The O(n 2 ) message complexity comes from the fact that, due to the predicates of line 5 and 6, each application message m is forwarded at most once by each process (line 11). The 2∆ follows from the same argument.

2 T heorem 1

An SCD-broadcast-based Communication Pattern

All the algorithms implementing concurrent objects and tasks, which are presented in this paper, are based on the same communication pattern described by Algorithm 2. This pattern involves each process, either as a client (when it invokes an operation), or as a server (when it scd-delivers a message set). When a process p i invokes an operation op(), it executes once the lines 1-3 for a task, and 0, 1, or 2 times for an operation on a concurrent object. In this last case, this number of times depends on the consistency condition which is implemented (linearizability [START_REF] Herlihy | Linearizability: a correctness condition for concurrent objects[END_REF] or sequential consistency [START_REF] Lamport | How to make a multiprocessor computer that correctly executes multiprocess programs[END_REF]).

operation op() is

According to the object/task that is implemented, and its consistency condition (if it is an object, linearizability vs seq. consistency), execute 0, 1, or 2 times the lines 1-3 where the message type TYPE is either a pure synchronization message SYNC or an object/task-dependent message MSG; (1) donei ← false;

(2) scd_broadcast TYPE(a, b, ..., i); a, b, ... are data, and i is the id of the invoking process; a message SYNC carries only the id of its sender; (3) wait(donei); (4) According to the states of the local variables, compute a result r; return(r).

when the message set { MSG(..., j1), . . . , MSG(..., jx)), SYNC(jx+1), . . . , SYNC(jy) } is scd-delivered do (5) for each message m = MSG(..., j) do statements specific to the object/task that is implemented end for;

(6) if ∃ : j = i then donei ← true end if.
Algorithm 2: Communication pattern (Code for p i ) All the messages sent by a process p i are used to synchronize its local data representation of the object, or its local view of the current state of the task. This synchronization is realized by the Boolean done i and the parameter i carried by every message (lines 1, 3, and 6): p i is blocked until the message it scd-broadcast just before is scd-delivered. The values carried by a message MSG are related to the object/task that is implemented, and may require local computation.

It appears that the combination of this communication pattern and the properties of SCD-broadcast provides us with a single simple framework that allows for correct implementations of a specific family of concurrent objects and tasks.

The next three sections describe algorithms implementing a snapshot object, a counter object, and the lattice agreement task, respectively. All these algorithms consider the system model CAMP n,t [∅] enriched with SCD-broadcast (denoted CAMP n,t [SCD-broadcast]), and use the previous communication pattern. [START_REF] Attiya | Atomic snapshots using lattice agreement[END_REF] The Power of SCD-broadcast: Snapshot Object .m] as if it was executed instantaneously. Hence, in any execution of a snapshot object, its operations write() and snapshot() are linearizable.

The underlying atomic registers can be Single-Reader (SR) or Multi-Reader (MR) and Single-Writer (SR) or Multi-Writer (MW). We consider only SWMR and MWMR registers. If the registers are SWMR the snapshot is called SWMR snapshot (and we have then m = n). Moreover, we always have r = i, when p i invokes write(r, -). If the registers are MWMR, the snapshot object is called MWMR.

Implementations based on read/write registers. Implementations of both SWMR and MWMR snapshot objects on top of read/write atomic registers have been proposed (e.g., [START_REF] Afek | Atomic snapshots of shared memory[END_REF][START_REF] Anderson | Multi-writer composite registers[END_REF][START_REF] Imbs | Help when needed, but no more: efficient read/write partial snapshot[END_REF][START_REF] Inoue | Linear time snapshots using multi-writer multi-reader registers[END_REF]). The "hardness" to build snapshot objects in read/write systems and associated lower bounds are presented in the survey [START_REF]How hard is it to take a snapshot?[END_REF]. The best algorithm known ( [START_REF] Attiya | Atomic snapshots in O(n log n) operations[END_REF]) to implement an SWMR snapshot requires O(n log n) read/write on the base SWMR registers for both write() and snapshot(). As far as MWMR snapshot objects are concerned, there are implementations where each operation has an O(n) cost 3 .

As far as the construction of an SWMR (or MWMR) snapshot object in crash-prone asynchronous message-passing systems where t < n/2 is concerned, it is possible to stack two constructions: first an algorithm implementing SWMR (or MWMR) atomic read/write registers (e.g., [START_REF] Attiya | Sharing memory robustly in message passing systems[END_REF])), and, on top of it, an algorithm implementing an SWMR (or MWMR) snapshot object. This stacking approach provides objects whose operation cost is O(n 2 log n) messages for SWMR snapshot, and O(n 2 ) messages for MWMR snapshot.

An algorithm for atomic MWMR snapshot in CAMP n,t [SCD-broadcast]

Local representation of REG at a process p i . At each process p i , REG[1..m] is represented by three local variables reg i [1..m] (data part), plus tsa i [1..m] and done i (control part).

• done i is a Boolean variable.

• reg i [1.
.m] contains the current value of REG[1..m], as known by p i . Algorithm 3: Construction of an MWMR snapshot object CAMP n,t [SCD-broadcast] (code for p i ) Algorithm 3: snapshot operation. This algorithm consists of one instance of the communication pattern introduced in Section 5 (line 1), followed by the return of a copy of the local value of reg i [1..m] (line 2). The message SYNC(i), which is scd-broadcast is a pure synchronization message, whose aim is to entail the refreshment of the value of reg i [1..m] (lines 5-11) which occurs before the setting of done i to true (line 12). Algorithm 3: write operation. (Lines 3-4) When a process p i wants to assign a value v to REG[r], it invokes REG.write(r, v). This operation is made up of two instances of the communication pattern. The first one (line 3) is a re-synchronization, as in the snapshot operation, whose side effect is here to provide p i with an up-to-date value of tsa i [r].date. In the second instance of the communication pattern, p i associates the timestamp tsa i [r].date + 1, i with v, and scd-broadcasts the data/control message WRITE(r, v, tsa i [r].date + 1, i ). In addition to informing the other processes on its write of REG[r], this message WRITE() acts as a re-synchronization message, exactly as a message SYNC(i). When this synchronization terminates (i.e., when the Boolean done i is set to true), p i returns from the write operation.

• tsa i [1..m]
Algorithm 3: scd-delivery of a set of messages. When process p i scd-delivers a message set, namely, { WRITE(r j 1 , v j 1 , date j 1 , j 1 ), . . . , WRITE(r jx , v jx , date jx , j x ), SYNC(j x+1 ), . . . , SYNC(j y ) } it first looks if there are messages WRITE(). If it is the case, for each register REG[r] for which there are messages WRITE(r, -, -) (line 5), p i computes the maximal timestamp carried by these messages (line 6), and updates accordingly its local representation of REG[r] (lines 7-10). Finally, if p i is the sender of one of these messages (WRITE() or SYNC()), done i is set to true, which terminates p i 's resynchronization (line 12).

Remark: Linearizability imposes that, when two snapshot operations are not concurrent, the second operation returns a value at least as recent as the first one. In [START_REF] Attiya | Sharing memory robustly in message passing systems[END_REF], this is done by forcing the readers to mimic the algorithm executed by the writers, which is usually summarized by the adage "readers must write". This step is not required in Algorithm 3 because the SYNC() message sent during a snapshot operation by p i (Line 1) serves two purposes: 1) it gathers all WRITE(-, -, -) messages sent by write operations that have already terminated when p i started its snapshot operation, and 2) it orders, by transitivity, all SYNC() messages sent by subsequent snapshot operations with all WRITE(-, -, -) messages received by p i before the end of its snapshot operation.

Proof of Algorithm 3

As they are implicitly used in the proofs that follow, let us recall the properties of the SCD-broadcast abstraction. The non-faulty processes scd-deliver the same messages (exactly one each), and each of them was scd-broadcast. As a faulty process behaves correctly until it crashes, it scd-delivers a subset of the messages scd-delivered by the non-faulty processes.

Without loss of generality, we assume that there is an initial write operation issued by a non-faulty process. Moreover, if a process crashes in a snapshot operation, its snapshot is not considered. If a process crashes in a write operation, its write is considered only if the message WRITE() it sent at line 4 is scd-delivered to at least one non-faulty process (and by the Termination-2 property, to all non-faulty processes). Let us notice that a message SYNC() scd-broadcast by a process p i does not modify the local variables of the other processes.

Lemma 8 If a non-faulty process invokes an operation, it returns from its invocation.

Proof Let p i be a non-faulty process that invokes a read or write operation. By the Termination-1 property of SCD-broadcast, it eventually receives a message set containing the message SYNC() or WRITE() it sends at line 2, 3 or 4. As all the statements associated with the scd-delivery of a message set (lines 5-12) terminate, it follows that the synchronization Boolean done i is eventually set to true. Consequently, p i returns from the invocation of its operation.

2 Lemma 8

Extension of the relation < ts . The relation < ts is extended to a partial order on arrays of timestamps, denoted ≤ tsa , defined as follows:

tsa1[1..m] ≤ tsa tsa2[1..m] def = ∀r : (tsa1[r] = tsa2[r] ∨ tsa1[r] < ts tsa2[r]). Moreover, tsa1[1..m] < tsa tsa2[1..m] def = (tsa1[1..m] ≤ tsa tsa2[1..m]) ∧ (tsa1[1..m] = tsa2[1..m]).
Definition. Let TSA i be the set of the array values taken by tsa i [1..m] at line 12 (end of the processing of a message set by process p i ). Let TSA = ∪ 1≤i≤n TSA i .

Lemma 9

The order ≤ tsa is total on TSA.

Proof Let us first observe that, for any i, all values in TSA i are totally ordered (this comes from ts i [1..m] whose entries can only increase, lines 7 and 10). Hence, let tsa1[1..m] be an array value of TSA i , and tsa2[1..m] an array value of TSA j , where i = j.

Let us assume, by contradiction, that ¬(tsa1 ≤ tsa tsa2) and ¬(tsa2 ≤ tsa tsa1). As ¬(tsa1 ≤ tsa tsa2), there is a registers r such that tsa2[r] < tsa1[r]. According to lines 7 and 9, there is a message WRITE(r, -, tsa1[r]) received by p i when tsa i = tsa1 and not received by p j when tsa j = tsa2 (because tsa2[r] < tsa1[r]). Similarly, there is a message WRITE(r , -, tsa2[r ]) received by p j when tsa j = tsa2 and not received by p i when tsa i = tsa1. This situation contradicts the MS-Ordering property, from which we conclude that either tsa1 ≤ tsa tsa2 or tsa2 ≤ tsa tsa1.

2 Lemma 9

Definitions. Let us associate a timestamp ts(write(r, v)) with each write operation as follows. Let p i be the invoking process; ts(write(r, v)) is the timestamp of v as defined by p i at line 4, i.e., tsa i [r].date + 1, i . Let op1 and op2 be any two operations. The relation ≺ on the whole set of operations is defined as follows: op1 ≺ op2 if op1 terminated before op2 started. It is easy to see that ≺ is a real-time-compliant partial order on all the operations.

Lemma 10

No two write operations on the same register write1(r, v) and write2(r, w) have the same timestamp, and (write1(r, v) ≺ write2(r, w)) ⇒ (ts(write1) < ts ts(write2)).

Proof Let date1, i and date2, j be the timestamp of write1(r, v) and write2(r, w), respectively. If i = j, write1(r, v) and write2(r, w) have been produced by different processes, and their timestamp differ at least in their process identity.

So, let us consider that the operations have been issued by the same process p i , with write1(r, v) first. As write1(r, v) precedes write2(r, w), p i first (line 4) invoked scd_broadcast WRITE(r, v, date1, i ), and later WRITE(r, w, date2, i ). It follows that these SCD-broadcast invocations are separated by a local reset of the Boolean done i at line 4. Moreover, before the reset of done i due to the scd-delivery of the message {. . . ,WRITE(r, v, date1, i ), . . .}, we have tsa i [r].date i ≥ date1 (lines 6-10). Hence, we have tsa i [r].date ≥ date1 before the reset of done i (line 12). Then, due to the "+1" at line 4, WRITE(r, w, date2, i ) is such that date2 > date1, which concludes the proof of the first part of the lemma.

Let us now consider that write1(r, v) ≺ write2(r, w). If write1(r, v) and write2(r, w) have been produced by the same process we have date1 < date2 from the previous reasoning. So let us assume that they have been produced by different processes p i and p j . Before terminating write1(r, v) (when the Boolean done i is set true at line 12), p i received a message set ms1 i containing the message WRITE(r, v, date1, i ). When p j executes write2(r, w), it first invokes scd_broadcast SYNC(j) at line 3. Because write1(r, v) terminated before write2(r, w) started, this message SYNC(j) cannot belong to ms1 i .

Due to Integrity and Termination-2 of SCD-broadcast, p j eventually scd-delivers exactly one message set ms1 j containing WRITE(r, v, date1, i ). Moreover, it also scd-delivers exactly one message set ms2 j containing its own message SYNC(j). On the the other side, p i scd-delivers exactly one message set ms2 i containing the message SYNC(j). It follows from the MS-Ordering property that, if ms2 j = ms1 j , p j cannot scd-deliver ms2 j before ms1 j . Then, whatever the case (ms1 j = ms2 j or ms1 j is scd-delivered at p j before ms2 j ), it follows from the fact that the messages WRITE() are processed (lines 5-11) before the messages SYNC(j) (line 12), that we have tsa j [r] ≥ date1, i when done j is set to true. It then follows from line 4 that date2 > date1, which concludes the proof of the lemma.

2 Lemma 10

Associating timestamp arrays with operations. Let us associate a timestamp array tsa(op) [1..m] with each operation op() as follows.

• Case op() = snapshot(). Let p i be the invoking process; tsa(op) is the value of tsa i [1..m] when p i returns from the snapshot operation (line 2).

• Case op() = write(r, v). Let min tsa ({A}), where A is a set of array values, denote the smallest array value of A according to < tsa . Let tsa(op

) def = min tsa ({tsa[1.
.m] ∈ TSA such that ts(op) ≤ ts tsa[r]}). Hence, tsa(op) is the first tsa[1..m] of TSA, that reports the operation op() = write(r, v).

Lemma 11 Let op and op be two distinct operations such that op ≺ op . We have tsa(op) ≤ tsa tsa(op ). Moreover, if op is a write operation, we have tsa(op) < tsa tsa(op ).

Proof Let p i and p j be the processes that performed op and op , respectively. Let SYNC j be the SYNC(j) message sent by p j (at line 1 or 3) during the execution of op . Let term_tsa i be the value of tsa i [1..m] when op terminates (line 2 or 4), and sync_tsa j the value of tsa j [1..m] when done j becomes true for the first time after p j sent SYNC j (line 1 or 3). Let us notice that term_tsa i and sync_tsa j are elements of the set TSA.

According to lines 7 and 10, for all r, tsa i [r] is the largest timestamp carried by a message WRITE(r, v, -) received by p i in a message set before op terminates. Let m be a message such that there is a set sm scd-delivered by p i before it terminated op. As p j sent SYNC j after p i terminated, p i did not receive any set containing SYNC j before it terminated op. By the properties Termination-2 and MS-Ordering, p j received message m in the same set as SYNC j or in a message set sm received before the set containing SYNC j . Therefore, we have term_tsa i ≤ tsa sync_tsa j .

If op is a snapshot operation, then tsa(op) = term_tsa i . Otherwise, op() = write(r, v). As p i has to wait until it processes a set of messages including its WRITE() message (and executes line 12), we have ts(op) < ts term_tsa i [r]. Finally, due to the fact that term_tsa i ∈ TSA and Lemma 9, we have tsa(op) ≤ tsa term_tsa i .

If op is a snapshot operation, then sync_tsa j = tsa(op ) (line 2). Otherwise, op() = write(r, v) and thanks to the +1 in line 4, sync_tsa j [r] is strictly smaller than tsa(op )[r] which, due to Lemma 9, implies sync_tsa j < tsa tsa(op ).

It follows that, in all cases, we have tsa(op) ≤ tsa term_tsa i ≤ tsa sync_tsa j ≤ tsa tsa(op ) and if op is a write operation, we have tsa(op) ≤ tsa term_tsa i ≤ tsa sync_tsa j < tsa tsa(op ), which concludes the proof of the lemma.

2 Lemma 11

The previous lemmas allow the operations to be linearized (i.e., totally ordered in an order compliant with both the sequential specification of a register, and their real-time occurrence order) according to a total order extension of the reflexive and transitive closure of the → lin relation defined thereafter. Definition 1 Let op, op be two operations. We define the → lin relation by op → lin op if one of the following properties holds:

• op ≺ op ,
• tsa(op) < tsa tsa(op ),

• tsa(op) = tsa(op ), op is a write operation and op is a snapshot operation,

• tsa(op) = tsa(op ), op and op are two write operations on the same register and ts(op) < ts ts(op ), Lemma 12 The snapshot object built by Algorithm 3 is linearizable.

Proof We recall the definition of the → lin relation: op → lin op if one of the following properties holds:

• op ≺ op ,
• tsa(op) < tsa tsa(op ),

• tsa(op) = tsa(op ), op is a write operation and op is a snapshot operation,

• tsa(op) = tsa(op ), op and op are two write operations on the same register and ts(op) < ts ts(op ),

We define the → lin relation as the reflexive and transitive closure of the → lin relation.

Let us prove that the → lin relation is a partial order on all operations. Transitivity and reflexivity are given by construction. Let us prove antisymmetry. Suppose there are op 0 , op 1 , ..., op m such that op 0 = op m and op i → lin op i+1 for all i < m. By Lemma 11, for all i < m, we have tsa(op i ) ≤ tsa tsa(op i+1 ), and tsa(op m ) = tsa(op 0 ), so the timestamp array of all operations are the same. Moreover, if op i is a snapshot operation, then op i ≺ op (i+1)%m is the only possible case (% stands for "modulo") , and by Lemma 11 again, op (i+1)%m is a snapshot operation. Therefore, only two cases are possible.

• Let us suppose that all the op i are snapshot operations and for all i, op i ≺ op (i+1)%m . As ≺ is a partial order relation, it is antisymmetric, so all the op i are the same operation.

• Otherwise, all the op i are write operations. By Lemma 11, for all op i ≺ op (i+1)%m . The operations op i and op i+1%m are ordered by the fourth point, so they are write operations on the same register and ts(op i ) < ts ts(op i+1%m ). By antisymmetry of the < ts relation, all the op i have the same timestamp, so by Lemma 10, they are the same operation, which proves antisymmetry.

Let ≤ lin be a total order extension of → lin . Relation ≤ lin is real-time compliant because → lin contains ≺.

Let us consider a snapshot operation op and a register r such that tsa(op)[r] = date1, i . According to line 4, it is associated to the value v that is returned by read1() for r, and comes from a WRITE(r, v, date1, i ) message sent by a write operation op r = write(r, v). By definition of tsa(op r ), we have tsa(op r ) ≤ tsa tsa(op) (Lemma 11), and therefore op r ≤ lin op. Moreover, for any different write operation op r on r, by Lemma 10, ts(op r ) = ts(op r ). If ts(op r ) < ts ts(op r ), then op r ≤ lin op r . Otherwise we have tsa(op) < tsa tsa(op r ), and (due to the first item of the definition of → lin ) we have op ≤ lin op r . In both cases, the value written by op r is the last value written on r before op, according to ≤ lin .

2 Lemma 12

Time and Message costs. An invocation of snapshot() involves one invocation of scd_broadcast(), while an invocation of write() involves two. As scd_broadcast() costs O(n 2 ) protocol messages and 2∆ time units, snapshot() cost the same, and write() costs the double.

Theorem 2 Algorithm 3 builds an MWMR atomic snapshot object in the model CAMP n,t [SCD-broadcast].

The operation snapshot costs one SCD-broaddast, the write() operation costs two.

Linearizable register

Linearizable snapshot object

Sequentially consistent snapshot object [START_REF] Attiya | Sharing memory robustly in message passing systems[END_REF] [32]

[5] + [START_REF] Attiya | Atomic snapshots in O(n log n) operations[END_REF] [15] This paper [START_REF] Perrin | On composition and implementation of sequential consistency[END_REF] This paper The next corollary follows from (i)Theorem 1, (ii) Theorem 2 and (iii) the fact that the constraint (t < n/2) is an upper bound on the number of faulty processes to build a read/write register (or snapshot object) [START_REF] Attiya | Sharing memory robustly in message passing systems[END_REF].

Read / Snapshot # msgs O(n) O(n) O(n 2 log n) O(n 3 ) O(n 2 ) 0 0 latency 4∆ 2 -4∆ O(n log n∆) O(n∆) 2∆ 0 -4∆ 0 Write # msgs O(n) O(n 2 ) O(n 2 log n) O(n) O(n 2 ) O(n 2 ) O(n 2 ) latency 2∆ 2∆ O(n log n∆) O(n∆) 4∆ 0 2∆
Corollary 1 Algorithm 1 is resiliency optimal.

Comparison with other algorithms. Interestingly, Algorithm 3 is more efficient (from both time and message point of views) than the stacking of a read/write snapshot algorithm running on top of a message-passing emulation of a read/write atomic memory (as presented on Figure 2, such a stacking would costs O(n 2 log n) messages and O(n∆) time units, see Section 6.1).

Sequentially consistent snapshot object. When considering Algorithm 3, let us suppress line 1 and line 3 (i.e., the messages SYNC are suppressed). The resulting algorithm no more implements a linearizable snapshot object. However, it still implements a snapshot object that is sequentially consistent [START_REF] Lamport | How to make a multiprocessor computer that correctly executes multiprocess programs[END_REF]. This means that the order in which each process sees the operations performed on the snapshot object is consistent with a same total order similarly to linearizability except that this order may not respect real time. This results from the suppression of the real-time compliance due to the messages SYNC. The operation snapshot() is purely local, hence its cost is 0. The cost of the operation write() is one SCD-broadcast, i.e., 2∆ time units and O(n 2 ) protocol messages. The proof of this algorithm is left to the reader. [START_REF] Attiya | Atomic snapshots in O(n log n) operations[END_REF] The Power of SCD-broadcast: Counter Object Definition. Let a counter be an object which can be manipulated by three parameterless operations denoted increase(), decrease(), and read(). Let C be a counter. From a sequential specification point of view C.increase() adds 1 to C, C.decrease() subtracts 1 from C, C.read() returns the value of C. As indicated in the Introduction, due to its commutative operations, this object is a good representative of a class of CRDT objects (conflict-free replicated data type as defined in [START_REF] Shapiro | Conflict-free replicated data types[END_REF]).

An algorithm satisfying linearizability. Algorithm 4 implements an atomic counter C. Each process p i manages a local variable counter i which stores it s loval view of the counter C. The text of the algorithm is self-explanatory. operation increase() is (1) donei ← false; scd_broadcast PLUS(i); wait(donei);

(2) return(). operation decrease() is the same as increase() where PLUS(i) is replaced by MINUS(i).

operation read() is (3) donei ← false; scd_broadcast SYNC(i); wait(donei); (4) return(counteri).

when the message set { PLUS(j1), . . . ,MINUS(jx), . . . , SYNC(jy), . . . } is scd-delivered do (5) let p = number of messages PLUS() in the message set; (6) let m = number of messages MINUS() in the message set; [START_REF] Attiya | Atomic snapshots in O(n log n) operations[END_REF]

counteri ← counteri + p -m; (8) if ∃ : j = i then donei ← true end if.
Algorithm 4: Construction of an atomic counter in CAMP n,t [SCD-broadcast] (code for p i )

The operation read() is similar to the operation snapshot() of the snapshot object. Differently from the write() operation on a snapshot object (which requires a synchronization message SYNC() and a data/synchronization message WRITE()), the update operations increase() and decrease() require only one data/synchronization message PLUS() or MINUS(). This is the gain obtained from the fact that, from a process p i point of view, the operations increase() and decrease() which appear between two consecutive of its read() invocations are commutative.

Lemma 13 If a non-faulty process invokes an operation, it returns from its invocation.

Proof Let p i be a non-faulty process that invokes increase(), decrease() or read(). By the Termination-1 property of SCD-broadcast, it eventually receives a message set containing the message PLUS(), MINUS() or SYNC() it sends at line 1 or 3. As all the statements associated with the scd-delivery of a message set (lines 5-8) terminate, it follows that the synchronization Boolean done i is eventually set to true. Consequently, p i returns from the invocation of its operation.

2 Lemma 13

Definition 2 Let op i be an operation performed by p i . We define past(op i ) as a set of messages by: • If op i is an increase() or decrease() operation, and m i is the message sent during its execution at line 1, then past(op i ) = {m : m → m i }. • If op i is a read() operation, then past(op i ) is the union of all sets of messages scd _delivered by p i before it executed line 4.

We define the → lin relation by op → lin op if one of the following conditions hold:

• past(op) past(op );

• past(op) = past(op ), op is an increase() or a decrease() operation and op is a read() operation.

Lemma 14

The counter object built by Algorithm 4 is linearizable.

Proof Let us prove that → lin is a strict partial order relation. Let us suppose op → lin op → lin op .

If op is a read() operation, we have past(op) ⊆ past(op ) past(op ). If op is an increase() or a decrease() operation, we have past(op) past(op ) ⊆ past(op ). In both cases, we have past(op) past(op ), which proves transitivity as well as antisymmetry and irreflexivity since it is impossible to have past(op) past(op).

Let us prove that → lin is real-time compliant. Let op i and op j be two operations performed by processes p i and p j respectively, and let m i and m j be the message sent during the execution of op i and op j respectively, on line 1 or 3. Suppose that op i ≺ op j (op i terminated before op j started). When p i returns from op i , by the waiting condition of line 1 or 3, it has received m i , but p j has not yet sent m j . Therefore, m i → i m j , and consequently m j / ∈ past(op i ). By the waiting condition during the execution of op j (line 1 or 3), we have m j ∈ past(op j ). By the Containment property of SCD-broadcast, we therefore have past(op i ) past(op j ), so op i → lin op j . Let ≤ lin be a total order extension of → lin . It is real-time compliant because → lin contains ≺.

Let us now consider the value returned by a read() operation op. Let p be the number of PLUS() messages in past(op) and let m be the number of MINUS() messages in past(op). According to line 1, op returns the value of counter i that is modified only at line 7 and contains the value p -m, by commutativity of additions and subtractions. Moreover, due to the definition of → lin , all pairs composed of a read() and an increase() or decrease() operations are ordered by → lin , and consequently, op has the same increase() and decrease() predecessors according to both → lin and to ≤ lin . Therefore, the value returned by op is the number of times increase() has been called, minus the number of times increase() has been called, before op according to ≤ lin , which concludes the lemma.

2 Lemma 14

Theorem 3 Algorithm 4 implements an atomic counter.

Proof Follows from Lemmas 13 and 14.

2 T heorem 3 operation increase() is (1) lsci ← lsci + 1;
(2) scd_broadcast PLUS(i);

(3) return().

operation decrease() is the same as increase() where PLUS(i) is replaced by MINUS(i).

operation read() is (4) wait(lsci = 0); An algorithm satisfying sequential consistency. The previous algorithm can be easily modified to obtain a sequentially consistent counter. To this end, a technique similar to the one introduced in [START_REF] Attiya | Sequential consistency versus linearizability[END_REF] can be used to allow the operations increase() and decrease() to have a fast implementation. "Fast" means here that these operations are purely local: they do not require the invoking process to wait in the algorithm implementing them. Differently, the operation read() issued by a process p i cannot be fast, namely, all the previous increase() and decrease() operations issued by p i must be applied to its local copy of the counter for its invocation of read() to terminate (this is the rule known under the name "read your writes"). Algorithm 5 is the resulting algorithm. In addition to counter i , each process manages a synchronization counter lsc i initialized to 0, which counts the number of increase() and decrease() executed by p i and not yet locally applied to counter i . Only when lsc i is equal to 0, p i is allowed to read counter i .

The cost of an operation increase() and decrease() is 0 time units plus the O(n 2 ) protocol messages of the underlying SCD-broadcast. The time cost of the operation read() by a process p i depends on the Remark 1. SCD-broadcast can be built on top of read/write registers (see below Theorem 5). It follows that the combination of Algorithm 6 and Algorithm 7 provides us with a pure read/write algorithm solving the lattice agreement task. As far as we know, this is the first algorithm solving lattice agreement, based only on read/write registers.

Remark 2. Similarly to the algorithms implementing snapshot objects and counters satisfying sequential consistency (instead of linearizability), Algorithm 6 uses no message SYNC().

Let us also notice the following. Objects are specified by "witness" correct executions, which are defined by sequential specifications. According to the time notion associated with these sequences we have two consistency conditions: linearizability (the same "physical" time for all the objects) or sequential consistency (a logical time is associated with each object, independently from the other objects). Differently, as distributed tasks are defined by relations from input vectors to output vectors (i.e., without referring to specific execution patterns or a time notion), the notion of a consistency condition (such as linearizability or sequential consistency) is meaningless for tasks.

The Computability Limit of SCD-broadcast

This section presents an algorithm building SCD-broadcast on top of SWMR snapshot objects. (Such snapshot objects can be easily obtained from MWMR snapshot objects [START_REF] Delporte-Gallet | Linear space bootstrap communication schemes[END_REF].) Hence, it follows from (a) this algorithm, (b) Algorithm 1, and (c) the impossibility proof to build an atomic register on top of asynchronous message-passing systems where t ≥ n/2 process may crash [START_REF] Attiya | Sharing memory robustly in message passing systems[END_REF], that SCD-broadcast cannot be implemented in CAMP n,t [t ≥ n/2], and snapshot objects and SCD-broadcast are computationally equivalent.

9.1 From snapshot to SCD-broadcast Shared objects. The shared memory is composed of two SWMR snapshot objects. Let denote the empty sequence.

• SENT [1.

.n]: snapshot object (initialized to [∅, . . . , ∅]), such that SENT [i] contains the messages scd-broadcast by p i . • SETS _SEQ [1..n]: snapshot object (initialized to [ , . . . , ]), such that SETS _SEQ[i] contains the sequence of the sets of messages scd-delivered by p i .

The notation ⊕ is used for the concatenation of a message set at the end of a sequence of message sets.

Local objects Each process p i manages the following local objects.

• sent i : local copy of the snapshot object SENT .

• sets_seq i : local copy of the snapshot object SETS _SEQ.

• to_deliver i : auxiliary variable whose aim is to contain the next message set that p i has to scddeliver.

The function members(set_seq) returns the set of all the messages contained in set_seq.

Description of Algorithm 7. When a process p i invokes scd_broadcast(m), it adds m to sent i [i] and SENT [i] to inform all the processes on the scd-broadcast of m. It then invokes the internal procedure progress() from which it exits once it has a set containing m (line 1).

A background task T ensures that all messages will be scd-delivered (line 2). This task invokes repeatedly the internal procedure progress(). As, locally, both the application process and the underlying task T can invoke progress(), which accesses the local variables of p i , those variables are protected by a The procedure progress() first invokes the internal procedure catchup(), whose aim is to allow p i to scd-deliver sets of messages which have been scd-broadcast and not yet locally scd-delivered.

To this end, catchup() works as follows (lines 12-17). Process p i first obtains a snapshot of SETS _SEQ, and saves it in sets_seq i (line 12). This allows p i to know which message sets have been scd-delivered by all the processes; p i then enters a "while" loop to scd-deliver as many message sets as possible according to what was scd-delivered by the other processes. For each process p j that has scd-delivered a message set set containing messages not yet scd-delivered by p i (predicate of line 13), p i builds a set to_deliver i containing the messages in set that it has not yet scd-delivered (line 14), and locally scddelivers it (line 16). This local scd-delivery needs to update accordingly both sets_seq i [i] (local update) and SETS _SEQ[i] (global update).

When it returns from catchup(), p i strives to scd-deliver messages not yet scd-delivered by the other processes. To this end, it first obtains a snapshot of SENT , which it stores in sent i (line 5). If there are messages that can be scd-delivered (computation of to_deliver i at line 6, and predicate at line 7), p i scd-delivers them and updates sets_seq i [i] and SETS _SEQ[i] (lines 7-9) accordingly.

Proof of Algorithm 7

Lemma 15 If a process scd-delivers a set containing a message m, some process invoked scd_broadcast(m).

Proof The proof follows directly from the text of the algorithm, which copies messages from SENT to SETS _SEQ, without creating new messages.

2 Lemma 15

Lemma 16 No process scd-delivers the same message twice.

Proof Let us first observe that, due to lines 7 and 15, all messages that are scd-delivered at a process p i have been added to sets_seq i [i]. The proof then follows directly from (a) this observation, (b) the fact

• p i scd-delivers the message m at line 16. Due to the predicate used at line 13 to build a set of messages to scd-deliver, this means that there is a process p j that has previously scd-delivered a set of messages containing m . Moreover, let us observe that the first time the message m is copied from SENT to some SETS _SEQ[x] occurs at line 8. As m was written in SENT before m , the corresponding process p x cannot see m and not m. It follows from the previous item that p x has scd-delivered m in the same message set (as the one including m ), or in a previous message set. It then follows from the predicate of line 13 that p i cannot scd-deliver m before m.

To summarize, the scd-deliveries of message sets in the procedure catchup()

cannot

Conclusion

This paper has introduced a new communication abstraction, suited to asynchronous message-passing systems where computing entities (processes) may crash. Denoted SCD-broadcast, it allows processes to broadcast messages and deliver sets of messages (instead of delivering each message one after the other). More precisely, if a process p i delivers a set of messages containing a message m, and later delivers a set of messages containing a message m , no process p j can deliver a set of messages containing m before a set of messages containing m. Moreover, there is no local constraint imposed on the processing order of the messages belonging to a same message set. SCD-broadcast has the following noteworthy features:

• It can be implemented in asynchronous message passing systems where any minority of processes may crash. Its costs are upper bounded by twice the network latency (from a time point of view) and O(n 2 ) (from a message point of view).

• Its computability power is the same as the one of atomic read/write register (anything that can be implemented in asynchronous read/write systems can be implemented with SCD-broadcast).

• It promotes a communication pattern which is simple to use, when one has to implement concurrent objects defined by a sequential specification or distributed tasks.

• When interested in the implementation of a concurrent object O, a simple weakening of the SCDbroadcast-based atomic implementation of O provides us with an SCD-broadcast-based implementation satisfying sequential consistency (moreover, the sequentially consistent implementation is more efficient than the atomic one).

On programming languages for distributed computing. Differently from sequential computing for which there are plenty of high level languages (each with its idiosyncrasies), there is no specific language for distributed computing. Instead, addressing distributed settings is done by the enrichment of sequential computing languages with high level communication abstractions. When considering asynchronous systems with process crash failures, total order broadcast is one of them. SCD-broadcast is a candidate to be one of them, when one has to implement read/write solvable objects and distributed tasks.

Remark. It is worth noticing that the a self-stabilizing version of the SCD algorithm presented in this article is described in [START_REF] Lundström | Self-stabilizing set-constrained delivery broadcast[END_REF], and an extension of it where up to t < n/4 processes may commit Byzantine failures is described in [START_REF] Auvolat | Byzantine-tolerant set-constrained delivery broadcast[END_REF]. It is still an open problem to find an SCD algorithm coping with up to t < n/3 Byzantine processes.

Figure 1 :Lemma 5

 15 Figure 1: Message pattern introduced in Lemma 5

2 7 Theorem 1

 271 Lemma Algorithm 1 implements the SCD-broadcast communication abstraction in CAMP n,t [t < n/2]. Moreover, each invocation of the operation scd_broadcast() requires O(n 2 ) protocol messages. If there is an upper bound ∆ on messages transfer delays (and local computation times are equal to zero), each SCD-broadcast takes at most 2∆ time units. Proof The proof follows from Lemma 1 (Validity), Lemma 2 (Integrity), Lemma 4 (MS-Ordering), Lemma 6 (Termination-1), and Lemma 7 (Termination-2).

6. 1

 1 Snapshot objectDefinition. The snapshot object was introduced in[START_REF] Afek | Atomic snapshots of shared memory[END_REF][START_REF] Anderson | Multi-writer composite registers[END_REF]. A snapshot object is an array REG[1..m] of atomic read/write registers which provides the processes with two operations, denoted write(r, -) and snapshot(). The invocation of write(r, v), where 1 ≤ r ≤ m, by a process p i assigns atomically v to REG[r]. The invocation of snapshot() returns the value of REG[1.

Figure 2 :

 2 Figure 2: Comparison of various algorithms (best complexities are highlighted)

Algorithm 7 :

 7 An implementation of SCD-broadcast on top of snapshot objects (code for p i ) local fair mutual exclusion algorithm providing the operations enter_mutex() and exit_mutex() (lines 3 and 11).

  Lemma 3 If p i fifo-broadcasts FORWARD_MSG(m, sd, sn sd , i, sn i ) (i.e., executes line 11), each nonfaulty process p j executes once fifo_broadcast FORWARD_MSG(m, sd, sn sd , j, sn j ).

	Lemma 2

Proof Let p j be a correct process. First, we prove that p j broadcasts a message FORWARD_MSG(m, sd, sn sd , j, sn j ). As p i is non-faulty, p j will eventually receive the message sent by p i . At that time, if sn sd > clock j [sd], after the condition on line 6 and whatever its result, buffer i contains a quadruplet qdplt with qdplt.sd = sd and qdplt.sn = sn sd . That qdplt was inserted at line 10 (possibly after the reception of a different message), just before p j sent a message FORWARD_MSG(m, sd, sn sd , j, sn j ) at line 11. Otherwise, clock j [sd] was incremented on line 18, when validating some qdplt added to buffer j after p j received a (first) message FORWARD_MSG(qdplt .msg, sd, sn sd , g, clock g [sd]) from p g . Because the messages FORWARD_MSG() are fifo-broadcast (hence they are delivered in their sending order), p sd sent message FORWARD_MSG(qdplt.msg, sd, sn sd , sd, sn sd ) before FORWARD_MSG(qdplt .msg, sd, clock j [sd], sd, clock j [sd]), and all other processes only forward messages, p j received FORWARD_MSG(qdplt.msg, sd, sn sd , -, -) from p g before the message FORWARD_MSG(qdplt .msg, sd, clock j [sd], -, -). At that time, sn sd > clock j

[sd]

  donei ← false; scd_broadcast WRITE(r, v, tsai[r].date + 1, i ); wait(donei).when the message set { WRITE(rj 1 , vj 1 , datej 1 , j1 ), . . . , WRITE(rj x , vj x , datej x , jx ), SYNC(jx+1), . . . , SYNC(jy) } is scd-delivered do (5) for each r such that WRITE(r, -, -) ∈ scd-delivered message set do if ∃ : j = i then donei ← true end if.

		def = (h1 <
	h2) ∨ ((h1 = h2) ∧ (i1 < i2)).
	operation snapshot() is
	(1) donei ← false; scd_broadcast SYNC(i); wait(donei);
	(2) return(reg i [1..m]).
	operation write(r, v) is
	(3) donei ← false; scd_broadcast SYNC(i); wait(donei);
	(4) (6)	let date, writer be the greatest timestamp in the messages WRITE(r, -, -);
	(7)	if (tsai[r] <ts date, writer )
	(8)	then let v the value in WRITE(r, -, date, writer );
	(9)	regi[r] ← v; tsai[r] ← date, writer
	(10)	end if;
	(11) end for;
	(12)

is an array of timestamps associated with the values stored in reg i [1..m]. A timestamp is a pair made of a local clock value and a process identity. Its initial value is 0, -. The fields associated with tsa i [r] are denoted tsa i [r].date, tsa i [r].proc . Timestamp-based order relation. We consider the classical lexicographical total order relation on timestamps, denoted < ts . Let ts1 = h1, i1 and ts2 = h2, i2 . We have ts1 < ts ts2

  operation scd_broadcast(m) is (1) senti[i] ← senti[i] ∪ {m}; SENT .write(senti[i]); progress().(2) background task T is repeat forever progress() end repeat. senti ← SENT .snapshot();(6) to_deliveri ← (∪ 1≤j≤n senti[j]) \ members(sets_seq i [i]); (7) if (to_deliveri = ∅) (8) then sets_seq i [i] ← sets_seq i [i] ⊕ to_deliveri ; SETS _SEQ.write(sets_seq i [i]); sets_seq i ← SETS _SEQ.snapshot(); (13) while (∃j, set : set is the first set in sets_seq i [j] : set ⊆ members(sets_seq i [i]) do (14) to_deliveri ← set \ members(sets_seq i [i]); (15) sets_seq i [i] ← sets_seq i [i] ⊕ to_deliveri ; SETS _SEQ.write(sets_seq i [i]);

	procedure progress() is
	(3) enter_mutex();
	(4) catchup();
	(5) (9)	scd_deliver(to_deliveri )
	(10) end if;
	(11) exit_mutex().
	procedure catchup() is
	(12) (16)	scd_deliver(to_deliveri )
	(17) end while.

  violate the MS-Ordering property, which is established at lines 6-10. 2 Lemma 20 Theorem 5 Algorithm 7 implements SCD-Broadcast in the classical wait-free read/write model CARW n,t [t < n]. Proof The proof follows from Lemma 15 (Validity), Lemma 16 (Integrity), Lemmas 17 and 18 (Termination-1), Lemma 19 (Termination-2), and Lemma 20 (MS-Ordering). 2 T heorem 5

More generally, in terms of consensus numbers[START_REF] Herlihy | Wait-free synchronization[END_REF], as soon as we can build atomic registers on top of an asynchronous message-passing system[START_REF] Attiya | Sharing memory robustly in message passing systems[END_REF], the consensus number of total order broadcast is +∞ and the one SCD-broadcast is 1. A communication abstraction, called k-BO-Broadcast, has been introduced in[START_REF] Imbs | Which broadcast abstraction captures k-set agreement?[END_REF], where it is shown that its computabilty power is exactly the one of k-set agreement (a natural generalization of consensus that allows up to k different values to be decided; consensus is 1-set agreement). In a system of n processes, SCD-broadcast and n-BO-Broadcast are equivalent. Unfortunately, except for the extreme cases k = 1 and k = n, it is not know how to implement k-BO-Broadcast.

From the point of view of the maximal number of process crashes that can be tolerated, assuming failures are independent.

Snapshot objects built in read/write models enriched with operations such as Compare&Swap, or LL/SC, have also been considered, e.g.,[START_REF] Imbs | Help when needed, but no more: efficient read/write partial snapshot[END_REF][START_REF] Jayanti | An optimal multiwriter snapshot algorithm[END_REF]. Here we are interested in pure read/write models.

Let us notice that it is possible that a process scd-delivers them in two different message sets, while another process scd-delivers them in the same set (which does not contradicts the lemma).

Acknowledgments

The authors want to thank the referees for their constructive comments that helped improve the content and the presentation of the paper.

This work was partially supported by the French ANR project DESCARTES (16-CE40-0023-03) devoted to layered and modular structures in distributed computing, the French "Étoile Montante en Pays De La Loire" regional project BROCCOLI devoted to the computability aspects of broadcast abstractions, and the French ANR project ByBLoS (ANR-20-CE25-0002-01) devoted the modular design of building blocks for large-scale trustless multi-users applications.

value of lsc i . It is 0 when p i has no "pending" counter operations.

Remark As in [START_REF] Attiya | Sequential consistency versus linearizability[END_REF], using the same technique, it is possible to design a sequentially consistent counter in which the operation read() is fast, while the operations increase() and decrease() are not. [START_REF] Attiya | Sequential consistency versus linearizability[END_REF] The Power of SCD-broadcast: Lattice Agreement Task Definition. Let S be a partially ordered set, and ≤ its partial order relation. Given S ⊆ S, an upper bound of S is an element x of S such that ∀ y ∈ S : y ≤ x. The least upper bound of S is an upper bound z of S such that, for all upper bounds y of S , z ≤ y. S is called a semilattice if all its finite subsets have a least upper bound. Let lub(S ) denotes the least upper bound of S .

Let us assume that each process p i has an input value in i that is an element of a semilattice S. The lattice agreement task was introduced in [START_REF] Attiya | Atomic snapshots using lattice agreement[END_REF] and generalized in [START_REF] Faleiro | Generalized lattice agreement[END_REF]. It provides each process with an operation denoted propose(), such that a process p i invokes propose(in i ) (we say that p i proposes in i ); this operation returns an element z ∈ S (we say that it decides z). The task is defined by the following properties, where it is assumed that each non-faulty process invokes propose().

• Validity. If process p i decides out i , we have in i ≤ out i ≤ lub({in 1 , . . . , in n }).

• Containment. If p i decides out i and p j decides out j , we have out i ≤ out j or out j ≤ out i .

• Termination. If a non-faulty process proposes a value, it decides a value.

Algorithm. Algorithm 6 implements the lattice agreement task. It is a very simple algorithm, which uses one instance of the communication pattern introduced in Section 5. The text of the algorithm is self-explanatory.

operation propose(ini) is (1) donei ← false; scd_broadcast MSG(i, ini); wait(donei);

(2) return(lub(reci)).

when the message set { MSG(j1, vj 1 ), . . . , MSG(jx, vj x )} is scd-delivered do

Algorithm 6: Solving Lattice Agreement in CAMP n,t [SCD-broadcast] (code for p i ) Theorem 4 Algorithm 6 solves lattice agreement.

Proof The Termination property follows from the Termination-1 property of SCD-broadcast (if a nonfaulty process SCD-broadcasts a message m, it SCD-delivers a message set containing m). The Validity property follows from the definition of the lub() operation, and the fact that, when a process p i executes line 2, rec i contains in i (it executed before lines 3-4 when it received a message set containing the message MSG(i, in i ) it SCD-broadcast at line 1).

As far as the Containment property is concerned we have the following. Let us assume, by contradiction, that there are two processes p i ans p j such that we have neither out i ≤ out j nor out j ≤ out i .This means that there is a value v ∈ out i \ out j , and a value v ∈ out j \ out i . Let ms i and ms i be the message sets (scd-delivered by p i ) which contained v and v respectively. As v ∈ out i and v / ∈ out i , we have ms i = ms i , and ms i was scd-delivered before ms i .

Defining similarly ms j (containing v ) and ms j (containing v), we have ms j = ms j , and ms j was scd-delivered before ms j . It follows that m → i m and m → j m, from which it follows that → = ∪ 1≤x≤n → x is not a partial order. A contradiction with the SCD-broadcast definition. 2 T heorem 4 that (due to the local mutual exclusion at each process) sets_seq i [i] is updated consistently, and (c) lines 6 and 14, which state that a message already scd-delivered (i.e., a message belonging to sets_seq i [i]) cannot be added to to_deliver i .

2 Lemma 16

Lemma 17 Any invocation of scd_broadcast() by a non-faulty process p i terminates.

Proof The proof consists in showing that the internal procedure progress() terminates. As the mutex algorithm is assumed to be fair, process p i cannot block forever at line 3. Hence, p i invokes the internal procedure catchup(). It then issues first a snapshot invocation on SETS _SEQ and stores the value it obtains the value of sets_seq i . There is consequently a finite number of message sets in sets_seq i . Hence, the "while" of lines 13-17 can be executed only a finite number of times, and it follows that any invocation of catchup() by a non-faulty process terminates. The same reasoning (replacing SETS _SEQ by SENT ) shows that process p i cannot block forever when it executes the lines 5-10 of the procedure progress().

2 Lemma 17

Lemma 18 If a non-faulty process scd-broadcasts a message m, it scd-delivers a message set containing m.

Proof Let p i be a non-faulty process that scd-broadcasts a message m. As it is non-faulty, p i adds m to SENT [i] and then invokes progress() (line 1). As m ∈ SENT , it is eventually added to to_deliver i

if not yet scd-delivered (line 6), and scd-delivered at line 9, which concludes the proof of the lemma.

Lemma 19 If a non-faulty process scd-delivers a message m, every non-faulty process scd-delivers a message set containing m.

Proof Let us assume that a process scd-delivers a message set containing a message m. It follows that the process that invoked scd_broadcast(m) added m to SENT (otherwise no process could scd-deliver m). Let p i be a correct process. It invokes progress() infinitely often (line 2). Hence, there is a first execution of progress() such that sent i contains m (line 5). If then follows from line 6 that m will be added to to_deliver i (if not yet scd-delivered). If follows that p i will scd-deliver a set of messages containing m at line 9.

2 Lemma 19

Lemma 20 Let p i be a process that scd-delivers a set ms i containing a message m and later scddelivers a set ms i containing a message m . No process p j scd-delivers first a set ms j containing m and later a set ms j containing m.

Proof Let us consider two messages m and m . Due to total order property on the operations on the snapshot object SENT , it is possible to order the write operations of m and m into SENT . Without loss of generality, let us assume that m is added to SENT before m . We show that no process scd-delivers m before m. 4 Let us consider a process p i that scd-delivers the message m . There are two cases.

• p i scd-delivers the message m at line 9. Hence, p i obtained m from the snapshot object SENT (lines 5-6). As m was written in SENT before m , we conclude that SENT contains m. It then follows from line 6 that, if p i has not scd-delivered m before (i.e., m is not in sets_seq i [i]), then p i scd-delivers it in the same set as m .