Proc. Structures in Concurrency Theory (STRICT), Berlin 1995, Workshops in Computing Series, Springer-Verlag (1995) 249–263.

Keeping Track of the Latest Gossip in Message-Passing Systems

Madhavan Mukund^{∗} K. Narayan Kumar^{†} Milind Sohoni^{‡}

Abstract

Consider a distributed system in which processes exchange information by passing messages. The gossip problem is the following: Whenever a process q receives a message from another process p, q must be able to decide which of p and q has more recent information about r, for every other process r in the system. With this data, q is in a position to update its knowledge about the global state of the system.

We propose a solution where each message between processes carries information about the current state of knowledge of the sender. This information is uniformly bounded if we make reasonable assumptions about the number of undelivered messages present at any time in the system. This means that the overhead of maintaining the latest gossip is a constant, independent of the length of the underlying computation.

Introduction

We tackle a natural problem from distributed computing, involving time-stamps. Let P be a set of computing agents or processes which exchange information by passing messages. The gossip problem is the following: Whenever a process q receives a message from another process p, q must be able to decide which of p and q has more recent information about r, for every other process r.

By keeping track of the latest gossip about other agents, each process can consistently update its knowledge about the global state of the system whenever it receives some new information from another process. Computing global information about the system from local information is, of course, a central issue in distributed computing. So, a solution to the gossip problem would be useful in a wide variety of applications involving distributed systems.

The gossip problem has been investigated in [5, 9] for systems where processes synchronize periodically and exchange information. We extend the solution proposed in [5, 9] to a general message-passing model, where processes communicate by sending messages on point-to-point channels. We assume that communication is guaranteed—all messages in the system are eventually delivered. However, we permit indefinite delays in transit. Also, messages need not be received in the order in which they were sent.

In our solution to the gossip problem, processes exchange only a bounded amount of gossip information with each message they send. At the heart of our solution is a protocol for time-stamping messages in the system using a finite set of labels.

∗School of Mathematics, SPIC Science Foundation, 92 G.N. Chetty Road, T. Nagar, Madras 600 017, India. E-mail: madhavan@ssf.ernet.in

†Computer Science Group, Tata Institute of Fundamental Research, Homi Bhabha Road, Colaba, Bombay 400 005, India. E-mail: kumar@tcs.tifr.res.in

‡Department of Computer Science and Engineering, Indian Institute of Technology, Bombay 400 076, India, E-mail: sohoni@cse.iitb.ernet.in

Time-stamping is a well-established technique for ordering events in a distributed setting [6, 7]. Time-stamping protocols which use only a bounded set of labels to tag events have attracted a fair amount of attention in recent years. Protocols have been exhibited for systems in which processes communicate via a shared memory [3, 4], as also for systems where processes synchronize periodically and exchange information [1, 2, 9]. However, no such protocol seems to exist for message-passing systems.

It is not difficult to see that in the completely general message-passing model described above, bounded time-stamping is not possible. In order to make the problem tractable, we have to restrict the model by placing a bound on the number of unacknowledged messages that can be present in the system at any time. We believe that this restriction is a natural one and that all “reasonable” distributed algorithms for message-passing systems would actually conform to the paradigm we propose.

An important feature of our solution to the gossip problem is that it does not introduce any additional messages—it just adds some additional data to each message of the underlying computation. This additional data is guaranteed to be uniformly bounded. So, given any distributed algorithm which conforms to the restricted model we work with, we can enhance the algorithm to also keep track of the latest gossip with only a constant overhead in message complexity.

The paper is organized as follows. In the next section we introduce our model of computation and formulate the gossip problem in terms of a natural partial order on the events in the system. Section 2 describes ideals, which capture the notion of a partial view of a distributed computation. Sections 3 and 4 describe a protocol to solve the gossip problem for fifo computations, where messages are delivered in the order they were sent. Each process maintains what we call primary information about the computation, using potentially unbounded labels to distinguish messages in the system. In Section 5 we show how to convert this protocol to one which uses bounded time-stamps, thereby establishing a solution to the gossip problem in the fifo case. In the final section, we give a quick sketch of how to extend this solution to the case where messages are not necessarily delivered in the order in which they were sent.

1 The Model

Let P = {p_{1}, p_{2}, …, p_{N}} be a set of processes which communicate with each other through messages. We assume that messages are never lost—that is, the communication medium is reliable. However, there may be an arbitrary delay between the sending of a message and its receipt. Further, messages need not be received in the order in which they were sent.

We assume that communication is point-to-point. So, each message is addressed to a specific process and is not seen by any of the other processes in the system. Thus, each transmission of a message from a process p to a process q consists of two distinct actions; the action s_{p→q} corresponds to the sending of the message from p to q and the action r_{q←p} corresponds to its receipt by q.

We can regard a computation of the system as a word over the alphabet C = C_{S} ∪ C_{R} where C_{S} = {s_{p→q} | p, q ∈ P} is the set of send actions and C_{R} = {r_{p←q} | p, q ∈ P} is the set of receive actions. Each action in C is “executed” by a single process. So, we can also partition C across processes— for each process p, C_{p} = {s_{p→q} | q ∈ P} ∪ {r_{p←q} | q ∈ P} is the set of p-actions that p participates in directly.

We shall regard a word u ∈ C^{∗} of length m as a function u : [1..m] → C, where [1..m] denotes the set {1, 2, …, m} if m ≥ 1 and is ∅ if m = 0. For u ∈ C^{∗} and c ∈ C, #_{c}(u) denotes the number of occurrences of c in u. We can extend this to subsets X ⊆ C: #_{X}(u) = ∑_{c∈X} #_{c}(u).

Not every word corresponds to a valid computation—for instance, we must insist that messages are received only after they are sent. In addition, since messages need not be received in the order they were sent, to completely specify a computation we need to match each receive event to the corresponding send event. With this in mind, we define computations as follows:

Computations A computation over C is a pair (u, ϕ) where u : [1..m] → C is a word and ϕ : [1..m] → [1..m] is a partial function such that:

(i) The domain of ϕ, dom(ϕ) is {i | u(i) ∈ C_{R}}.

(ii) ϕ is injective over dom(ϕ)—for each i, j ∈ dom(ϕ), i ≠ j ⇒ ϕ(i) ≠ ϕ(j).

(iii) For each i ∈ dom(ϕ), ϕ(i) < i.

(iv) If u(i) = r_{q←p} then u(ϕ(i)) = s_{p→q}.

If ϕ(i) = j, then u(i) is a receive action whose corresponding send action is u(j). By condition (ii), we may also refer to i unambiguously as ϕ^{−1}(j).

Example: Let P = {p, q} and let u be the string s_{p→q}s_{p→q}s_{p→q}r_{q←p}r_{q←p} and ϕ be the function where ϕ(4) = 3 and ϕ(5) = 1. So, in this computation, the message sent from p to q at u(1) is overtaken by the message sent at u(3). Moreover, the message sent at u(2) has not reached q at all.

Events and causality The word u imposes a total, temporal order on the actions observed during a computation (u, ϕ). However, in order to analyze the flow of information between processes, we need a more accurate description of the cause and effect relationship between the different actions in u.

Let (u, ϕ) be a computation, where u : [1..m] → C. We associate with (u, ϕ) a set of events E_{u} = {(i, u(i)) | i ∈ [1..m]}.

Let e = (i, u(i)) be an event in E_{u}. When there is no ambiguity, we shall use e to denote both i and u(i). For instance, e ∈ C_{p} denotes that u(i) ∈ C_{p}— in other words, e is a p-event. Similarly, if we say f = ϕ(e) we mean that f = (j, u(j)) is an event such that ϕ(i) = j. We shall also use E_{u} and u interchangeably in expressions such as #_{c}(E_{u}), which denotes #_{c}(u).

As we mentioned earlier, u imposes a total, temporal order on the events in E_{u}. Let e, f ∈ E_{u}. Then e < f provided e = (i, u(i)), f = (j, u(j)) and i < j. As usual e ≤ f if e < f or e = f.

Messages introduce causality across processes. For each pair (p, q) ∈ P × P such that p ≠ q, define ◁_{pq} to be the ordering

e ◁_{pq} f =^{△} e ∈ C_{p}, f ∈ C_{q} and ϕ(f) = e.

[figure: 1; s_{p→q}; 2; s_{p→r}; 3; s_{p→r}; 4; r_{r←p}; 5; s_{r→q}; 6; r_{q←r}; 7; s_{p→q}; 8; r_{r←p}; 9; r_{q←p}; 10; s_{r→q}]

Figure 1: An example

In addition, each process p orders the events it participates in. Define ◁_{pp} to be the strict ordering

e ◁_{pp} f =^{△} e < f, e ∈ C_{p}, f ∈ C_{p} and for all e < g < f, g ∉ C_{p}.

The set of all p-events in E_{u} is totally ordered by ◁_{pp}^{∗}, the reflexive, transitive closure of ◁_{pp}.

Define e ⊏ f if for some p, q ∈ P (where p and q need not be distinct), e ◁_{pq} f and e ⊑ f if e = f or e ⊏ f. Let ⊑^{∗} denote the transitive closure of ⊑. If e ⊑^{∗} f then we say that e is below f. The partial order ⊑^{∗} records the information we require about causality and independence between events in E_{u}.

Let e ∈ E_{u} be a p-event. The set of events below e is e↓= {f | f ⊑^{∗} e}. These represent the only actions which are known to p when e occurs.

Latest information Consider a computation (u, ϕ) and its associated set of events E_{u}. The ⊑^{∗}-maximum p-event in E_{u} is denoted max_{p}(E_{u})—this is the last event in E_{u} in which p has taken part. This quantity is well defined whenever #_{C_{p}}(E_{u}) > 0, since all p-events in E_{u} are totally ordered by ⊑^{∗}. (Recall that C_{p} ⊆ C is the set of p-actions, so #_{C_{p}}(E_{u}) denotes, by convention, the number of p-actions mentioned in the string u.)

Let p, q ∈ P. If #_{C_{p}}(E_{u}) > 0, the latest information p has about q in E_{u} corresponds to the ⊑^{∗}-maximum q-event in the set max_{p}(E_{u})↓, provided the set of q-events below max_{p}(E_{u}) is not empty. We denote this event by latest_{p←q}(E_{u}). (If there are no q-events in max_{p}(E_{u})↓, then latest_{p←q}(E_{u}) is undefined.)

Example: Let P = {p, q, r}. Consider the computation (u, ϕ), where u = s_{p→q}s_{p→r}s_{p→r}r_{r←p}s_{r→q}r_{q←r}s_{p→q}r_{r←p}r_{q←p}s_{r→q}, ϕ(4) = 2, ϕ(6) = 5, ϕ(8) = 3 and ϕ(9) = 1. Figure 1 is a picture of (E_{u}, ⊑^{∗}). The arrows in the figure correspond to the basic relations ◁_{pq}, which generate ⊑^{∗}.

In this computation, max_{q}(E_{u}) = (9, r_{q←p}). Though at max_{q}(E_{u}), process q hears from process p, latest_{q←p}(E_{u}) does not correpond to ϕ(max_{q}(E_{u})). Instead, latest_{q←p}(E_{u}) = (2, s_{p→r})—process q hears this information indirectly, via process r.

The gossip problem

Let p, q and r be processes such that latest_{p←r}(E_{u}) and latest_{q←r}(E_{u}) are both defined. Since both of these are r-events, they must be ordered by ◁_{rr}^{∗}. So, the latest information that p and q have about r will always be comparable.

The gossip problem is the following.

Whenever a process q receives a message from another process p, q must be able to decide which of p and q has heard more recently from r, for every other process r in the system.

One way to resolve this problem is as follows. As the computation progresses, each action is assigned a label by the process involved in that action. These labels allow processes to refer to events in an unambiguous manner. Each process then maintains the labels corresponding to its latest information. These labels are passed on with each communication in such a way that the process receiving the message can consistently update its own latest information.

The labels which are assigned to events during a computation are essentially time-stamps. A trivial solution to the time-stamping problem is for each process to maintain a local counter and assign strictly increasing counter values to the actions it executes. Then, two processes p and q can compare their latest information about r by checking which of their “latest” r labels is larger.

This scheme has the following drawback: As the computation progresses, the time-stamps assigned to events grow without bound. As a result, processes need to send longer and longer messages to transfer the labels corresponding to their latest information.

We seek a solution to the gossip problem where message lengths are bounded. This will ensure that the overhead of maintaining gossip information remains a constant, regardless of the length of the underlying computation.

To achieve this, we need to devise a scheme for labelling events using a bounded set of time-stamps. This means that, eventually, the same time-stamp will be assigned to more than one event. We need to ensure that time-stamps are reused in such a way that the update of latest information is not affected.

In principle, this should be possible. Let E_{u} be the events corresponding to the computation (u, ϕ) and let the number of processes in the system be N. Regardless of the number of events in E_{u}, at most N^{2} of them are relevant for solving the gossip problem—we only need to be able to compare the labels of events of the form latest_{p←q}(E_{u}) for each pair p, q ∈ P. So, in effect, at most N^{2} of the events in E_{u} constitute “current” gossip. Moreover, once an event becomes “obsolete” its time-stamp can be safely reused—an “obsolete” event can never become “current” at a later stage in the computation.

However, in the completely general model we have considered so far, it is impossible to achieve our goal. Since messages can be delayed indefinitely, a process p may send unboundedly many messages to q without knowing whether any or all of them have reached. Until p receives some confirmation from q that a particular message has reached, that message’s time-stamp cannot be reused. Thus, p will potentially need to use an unbounded number of time-stamps to label its messages to q. (Notice that this problem arises even if messages are delivered in the order in which they were sent—the main source of difficulty is the fact that there is no bound on the delay in delivering a particular message.)

B-bounded computations To overcome this problem, we need to restrict the class of computations we permit. Intuitively, we must bound the number of unacknowledged messages between any pair of processes. More formally, we say that (u, ϕ) is a B-bounded computation provided the following holds:

For each event e = (i, s_{p→q}) in E_{u}, #_{s_{p→q}}(e↓) − #_{r_{q←p}}(e↓) ≤ B.

In other words, p can send a fresh message to q only if, as far as it knows, the number of messages which it has already sent to q and which are yet to be delivered is less than B. Notice that p need not get direct acknowledgments from q. For instance, q could tell r that it has received a particular message m from p and r, in turn, could pass on this information to p.

Even for B-bounded computations, it is not immediate that the gossip problem has a solution. Suppose process p sends process q a message m. Though p is guaranteed to receive an acknowledgment for this message by the time it sends its next B messages to q, it cannot naïvely reuse m’s time-stamp once m is acknowledged. In between, q may have passed on the information in m to another process r, in which case the message m would still constitute “current” gossip for r. So, p has to have some means of recording which of its time-stamps are “in use” in the system at any given time.

Fifo computations Though we eventually present a solution to the gossip problem for all B-bounded computations, it will be convenient to first look at the problem in a restricted setting, where messages are delivered in the order in which they were sent. Formally, we say that a computation (u, ϕ) is fifo if the following holds: For every pair of processes p and q and for any two receive events e and f of the form r_{p←q}, e ⊏^{+} f implies ϕ(e) ⊏^{+} ϕ(f), where ⊏^{+} denotes the transitive closure of the relation ⊏.

Though messages between processes may not overtake each other in a fifo computation, gossip information along one route may overtake gossip information on another route. For instance, suppose p sends a message m to q and then sends a message m′ to r, following which r sends a message m′′ to q which reaches q before m does. Then, by the time m reaches q, the information that m contains about the state of p is obsolete. So, even in a fifo computation, each process has to have a way of resolving whether the message it has just received has already been superseded by earlier, indirect information.

For the next four sections, we assume that every computation we deal with is B-bounded and fifo.

2 Ideals

Let us fix a computation (u, ϕ), where u : [1..m] → C, and the corresponding set of events E_{u}, which we shall denote as just E from now on, for convenience.

The main source of difficulty in solving the gossip problem is the fact that the processes in P need to compute global information about the computation (u, ϕ) while each process only has access to a local, “partial” view of u. Although partial views of (u, ϕ) correspond to subsets of E, not every subset of E arises from such a partial view. Those subsets of E which do correspond to partial views of (u, ϕ) are called ideals.

Ideals A set of events I ⊆ E is called an order ideal if I is closed with respect to ⊑^{∗}—i.e., e ∈ I and f ⊑^{∗} e implies f ∈ I as well. We shall always refer to order ideals as just ideals.

The requirement that an ideal be closed with respect to ⊑^{∗} guarantees that the observation it represents is “consistent”—whenever an event e has been observed, so have all the events in the computation which necessarily precede e. Clearly the entire set E is an ideal, as is e↓ for any e ∈ E. It is easy to see that if I and J are ideals, so are I ∪ J and I ∩ J.

Example: In Figure 1, the set I = {(1, s_{p→q}), (2, s_{p→r}), (4, r_{r←p}), (5, s_{r→q}), (6, r_{q←r})} is an ideal. However, the set J = {(1, s_{p→q}), (2, s_{p→r}), (3, s_{p→r}), (5, s_{r→q}), (6, r_{q←r})} is not an ideal, since (4, r_{r←p}) ⊑^{∗} (5, s_{r→q}) but (4, r_{r←p}) ∉ J.

We need to generalize the notion of max_{p}(E), the maximum p-event in E, to all ideals I ⊆ E.

p-views For an ideal I, the ⊑^{∗}-maximum p-event in I is denoted max_{p}(I), provided #_{C_{p}}(I) > 0. The p-view of I is the set I_{p} = max_{p}(I)↓. So, I_{p} is the set of all events in I which p can “see”. (By convention, if max_{p}(I) is undefined—i.e., if there is no p-event in I—the p-view I_{p} is the empty set.)

Example: In Figure 1, consider the ideal I = {(1, s_{p→q}), (2, s_{p→r}), (4, r_{r←p}), (5, s_{r→q}), (6, r_{q←r})} Then I_{p}, the p-view of I is {(1, s_{p→q}), (2, s_{p→r})} whereas I_{q}, the q-view of I, is the entire ideal I.

3 Primary information

For processes p, q ∈ P, we have already defined latest_{p←q}(E), the latest information that p has about q after (u, ϕ). We can extend this definition to arbitrary ideals.

Let I ⊆ E be an ideal and p, q ∈ P. Then latest_{p←q}(I) denotes the ⊑^{∗}- maximum q-event in I_{p}, provided #_{C_{q}}(I_{p}) > 0. So, latest_{p←q}(I) is the latest q-event in I that p knows about. (As usual, if there is no q-event in I_{p}, the quantity latest_{p←q}(I) is undefined.)

It is clear that latest_{p←q}(I) will always correspond to a send action from C_{q}. However latest_{p←q}(I) need not be of the form s_{q→p}; the latest information that p has about q in I may have been obtained indirectly.

To maintain and update the latest information of processes, we need to expand the set of events that each process keeps track of. This expanded set will be called primary information. The primary information of a process contains not only its latest information about every other process but also information about acknowledged and unacknowledged messages in the system.

Message acknowledgments Let I ⊆ E be an ideal and e ∈ I an event of the form s_{p→q}. Then, e is said to have been acknowledged in I if ϕ^{−1}(e) exists and, moreover, belongs to I_{p}. Otherwise, e is said to be unacknowledged in I.

Notice that it is not enough for a message to have been received in I to deem it to be acknowledged. We demand that the event corresponding to the receipt of the message be “visible” to the sending process.

For an ideal I and a pair of processes p, q, let received_{q←p}(I) be the most recent message from p which q has received in I. In other words, received_{q←p} is the ⊑^{∗}-maximum s_{p→q} event e in I such that ϕ^{−1}(e) belongs to I. Since (u, ϕ) is a fifo computation, all messages corresponding to s_{p→q} events below e must have also been received by q.

Let unack_{p→q}(I) be the set of unacknowledged s_{p→q} events in I. By definition, this set consists of all s_{p→q} events in I which lie above received_{q←p}(I_{p}). Since the underlying computation (u, ϕ) is B-bounded, unack_{p→q}(I) never contains more than B events.

It will be convenient to define the notion of primary information with respect to ideals rather than processes.

Primary information Let I ⊆ E be an ideal. The primary information of I, primary(I), consists of the following events in I:

• The set latest(I) = {max_{p}(I) | p ∈ P}.

• The collection of sets unack(I) = {unack_{p→q}(I) | p, q ∈ P}.

• The set received(I) = {received_{q←p}(I) | p, q ∈ P}.

Observe that we can treat unack(I) to be just a set of events, though we have defined it as a collection of sets. For each event e = (i, u(i)) in unack(I), the action u(i) immediately determines the particular set unack_{p→q}(I) to which e belongs—e ∈ unack_{p→q}(I) iff u(i) = s_{p→q}. Similarly, we can regard latest(I) and received(I) as just sets of events. For each event e = (i, u(i)) in latest(I), e = max_{p}(I) iff u(i) ∈ C_{p}. In the same way, for each event (i, u(i)) in received(I), e = received_{q←p}(I) iff u(i) = s_{p→q}.

On the other hand, primary(I) is an indexed set of events. If e belongs to primary(I) we also need to record which of the three components e belongs to. However, often we shall abuse notation and treat primary(I) as just a set of events. It is clear that primary(I) contains only a bounded number of events.

Let I ⊆ E be an ideal and p a process such that I_{p} ≠ ∅. Then, primary(I_{p}) denotes the primary information of p in I—i.e., p’s primary information is just the primary information of the p-view of I. Clearly, the “latest information” of p after I is contained in its primary information—for every process q, latest_{p←q}(I) is just max_{q}(I_{p}).

To compare and update primary information, processes also need to remember how their primary events are ordered by ⊑^{∗}.

Primary graph Let I ⊆ E. The primary graph of I, primary-graph(I) is the directed graph (V, E) where:

• V = primary(I) (where primary(I)) is represented as an indexed set of events.)

• For v_{1}, v_{2} ∈ V, let e_{1} and e_{2} be the corresponding events from I. Then, (v_{1}, v_{2}) ∈ E iff e_{1} ⊑^{∗} e_{2}.

As with primary information, the primary graph of a process p in I is just the graph primary-graph(I_{p}).

For the moment we shall ignore the issue of assigning bounded time-stamps to events and assume that events are assigned unambiguous labels by some mechanism. For instance, as we mentioned earlier, each process could maintain a local counter and assign an increasing sequence of unique time-stamps to the events that it participates in.

Our first goal is to exhibit a procedure by which processes update their primary graphs without relying on the temporal order of events. Processes will only utilise the information about causality recorded in the primary graphs. All comparisons and updates of primary information will based purely on equality of event labels. This feature will allow us to extend the algorithm smoothly to the case where processes reuse labels.

4 Comparing primary information

Let E be the events corresponding to a computation (u, ϕ). Then, each ideal I ⊆ E corresponds to a possible partial computation of (u, ϕ). Let us assume that at the end of any partial computation I, each process maintains the information primary-graph(I_{p}).

In general, for an ideal I ⊆ E, each pair of processes p and q will have incomparable information about I. The events known to both p and q lie in the ideal I_{p} ∩ I_{q}. Events lying “above” the intersection are known to only one of the processes.

Suppose that q receives a message from p during the computation. Then, we have an event e_{p} ∈ E of the form s_{p→q} and a corresponding event e_{q} ∈ E of the form r_{q←p} such that ϕ(e_{q}) = e_{p}.

Let e′_{q} = max_{q}(e_{q}↓ −{e_{q}})—i.e., e′_{q} is the maximum q-event strictly below e_{q}. So, e′_{q}↓ represents the state of q’s knowledge before receiving this message from p. Let I be the ideal e_{p}↓ ∪ e′_{q}↓.

For the moment, let us assume that max_{p}(e′_{q}↓) ⊏^{+} e_{p}—i.e., the message sent at e_{p} has “new” information for q about the state of p. Then, e_{p}↓= I_{p} and e′_{q}↓= I_{q}. The message sent by p at e_{p} contains the primary graph primary-graph(I_{p}). On the other hand, before receiving this message, q’s primary graph corresponds to primary-graph(I_{q}).

Our first observation is that if q knows both primary-graph(I_{p}) and primary-graph(I_{q}), it can determine which events in the two primary graphs lie within I_{p} ∩ I_{q} and which lie outside this intersection.

Lemma 1 Let I ⊆ E be an ideal and p, q a pair of processes. Then, for each maximal event e in I_{p} ∩ I_{q}, either e ∈ latest(I_{p}) ∩ unack(I_{q}) or e ∈ unack(I_{p}) ∩ latest(I_{q}).

Proof Let e be a maximal event in I_{p} ∩ I_{q}. Suppose it is an r-event, for some r ∈ P. The event e must have ⊏-successors in both I_{p} and I_{q}. However, observe that any event f in E can have at most two immediate ⊏-successors—one “internal” successor within the process and, if f is a send event, one “external” successor corresponding to the matching receive event.

Thus, the maximal event e will have a ◁_{rr} successor e_{r} and a ◁_{rs} successor e_{s}, corresponding to some s ∈ P. Assume that e_{r} ∈ I_{q} − I_{p} and e_{s} ∈ I_{p} − I_{q}. Since the r-successor of e is outside I_{p}, e = max_{r}(I_{p}). So e belongs to latest(I_{p}). On the other hand, e is an unacknowledged s_{r→s} event in I_{q}. So, e ∈ unack_{r→s}(I_{q}), which is part of unack(I_{q}).

Symmetrically, if e_{r} ∈ I_{p} − I_{q} and e_{s} ∈ I_{q} − I_{p}, we find that e belongs to unack(I_{p}) ∩ latest(I_{q}). □

Thus, when q receives p’s primary graph, q can collect together in a set M all the events that lie in latest(I_{p}) ∩ unack(I_{q}) and unack(I_{p}) ∩ latest(I_{q}). By the preceding lemma, the events in M subsume the maximal events in the intersection I_{p} ∩ I_{q}. (It is easy to see that those events in M which are not actually maximal still lie within the intersection.)

The process q can use M to check whether a primary event e ∈ primary(I_{p})∪ primary(I_{q}) lies inside or outside the intersection—e lies inside the intersection iff it lies below one of the elements in M. These comparisons can be made using the edge information in the graphs primary-graph(I_{p}) and primary-graph(I_{q}).

Now, it is easy for q to compare the events in latest(I_{p}) with those in latest(I_{q}) to determine which of p and q have more recent information about every other process r.

Lemma 2 Let I ⊆ E be an ideal and p, q a pair of processes. Let e = max_{r}(I_{p}) and f = max_{r}(I_{q}) such that e ≠ f. Then, e ⊏^{+} f iff f ∈ I_{q} − I_{p}.

Proof (⇒) Suppose that e ⊏^{+} f. If f belongs to I_{p} then e ≠ max_{r}(I_{p}), which is a contradiction. On the other hand, f clearly belongs to I_{q}, so f ∈ I_{q} − I_{p}.

(⇐) Suppose that f ∈ I_{q} − I_{p}. If f ⊏^{+} e, then f ∈ I_{p} since I_{p} is an ideal. This is a contradiction. So it is not the case that f ⊏^{+} e. Since f ≠ e and all r-events are totally ordered by ⊑^{∗}, we must have e ⊏^{+} f. □

Once q has compared all events of the form max_{r}(I_{p}) and max_{r}(I_{q}), it can easily update its sets unack_{r→s}(I_{q}), where r ≠ q. The process which has better information about r also has better information about unacknowledged events of the form s_{r→s} in I. In other words, q inherits the sets unack_{r→s}(I_{p}) for every process r such that max_{r}(I_{p}) is more recent than max_{r}(I_{q}). On the other hand, if max_{r}(I_{p}) is older than max_{r}(I_{q}), then q ignores p’s sets unack_{r→s}(I_{p}) since it already has better information about these events.

At this stage, using the data in primary-graph(I_{p}) and primary-graph(I_{q}), q has updated all of primary(e_{q}↓) except for the sets {unack_{q→r}(e_{q}↓)}_{r∈P} and received(e_{q}↓). (Recall that e_{q} was the event where q received p’s message sent at e_{p}.)

We first describe how to construct received(e_{q}↓). Clearly received_{q←p}(e_{q}↓) = e_{p}. To fill in the rest of received(e_{q} ↓), we need the following observation, which we state without proof.

Proposition 3 For every pair of processes (r, s) such that (r, s) ≠ (p, q),

received_{s←r}(e_{q}↓) = \{ \begin{matrix}received_{s←r}(I_{p}) & if max_{s}(I_{q}) ⊏^{+} max_{s}(I_{p}) \\ received_{s←r}(I_{q}) & otherwise \end{matrix}

So, once q has compared the events in latest(I_{p}) and latest(I_{q}), it can decide which of the events in received(e_{q}↓) are inherited from received(I_{p}) and which are retained from received(I_{q}).

Process q can now use the information in received(e_{q}↓) to update the sets {unack_{q→r}(I_{q})}_{r∈P} by purging acknowledged events from these lists. Formally, for every process r, unack_{q→r}(e_{q}↓) consists of all the events in unack_{q→r}(I_{q}) which lie above received_{r←q}(e_{q} ↓). This update can be made with the edge information in primary-graph(I_{q}).

Having constructed the sets latest(e_{q}↓), unack(e_{q}↓) and received(e_{q}↓), we need to extend this set to the graph primary-graph(e_{q}↓).

Let f_{1}, f_{2} ∈ primary(e_{q}↓). If both f_{1} and f_{2} came from primary(I_{p}), then we add an edge from f_{1} to f_{2} in primary-graph(e_{q}↓) iff a corresponding edge existed in primary-graph(I_{p}). A symmetric situation applies if both f_{1} and f_{2} were contributed by primary(I_{q}).

So, the only interesting case is when f_{1} and f_{2} originally came from different processes. Without loss of generality, suppose that f_{1} came from primary(I_{p}) and f_{2} from primary(I_{q}). Then, from the method which we used to compare events, we know that f_{1} must have been in I_{p} − I_{q} and f_{2} must have been in I_{q} − I_{p}. So, it is clear that f_{1} and f_{2} are unordered in E and there is therefore no edge between them in primary-graph(e_{q}↓).

So far, we have shown how to construct primary-graph(e_{q}↓) assuming that max_{p}(e′_{q} ↓) ⊏^{+} e_{p}. If q already has better information about p—i.e., e_{p} ⊑^{∗} max_{p}(e′_{q}↓)—it is easy to see that e_{p} must in fact lie strictly below max_{p}(e′_{q}↓), since q’s knowledge of e_{p} must have come from a message sent by p after e_{p} to some other process r. In this case, q can no longer use Lemma 1 to compute the maximal elements in the intersection e_{p}↓ ∩e′_{q}↓ since the p-view of the ideal I = e_{p}↓ ∪e′_{q}↓ is not e_{p}↓ but max_{p}(e′_{q}↓)↓.

However, it is easy to see that if e_{p} ⊏^{+} max_{p}(e′_{q}↓) then e_{p}↓⊆ e′_{q}↓. So, for every other process r, max_{r}(e_{p}↓) ⊑^{∗} max_{r}(e′_{q}↓). It then follows that the only update that q has to make to primary-graph(e′_{q}↓) is to set received_{q←p}(e_{q}↓) to e_{p}. The rest of primary-graph(e_{q}↓) is inherited from primary-graph(e′_{q}↓).

All we need is a means for q to detect whether e_{p} is a “new” message. Based on this, q can decide whether to ignore the information in primary-graph(e_{p}↓). This is not very difficult because of the following observation.

Proposition 4 Let e_{p} be a s_{p→q} event such that ϕ^{−1}(e_{p}) = e_{q} and let e′_{q} = max_{q}(e_{q}↓ −{e_{q}}). Then e_{p} ⊑^{∗} max_{p}(e′_{q}↓) iff e_{p} ∈ unack_{p→q}(e′_{q}↓).

In other words, to check if the message is old, all q has to do is to see if e_{p} is already in its set unack(e′_{q}↓). This leads to the following general statement which summarizes the results of this section.

Lemma 5 Let e_{p} be a s_{p→q} event in E such that ϕ^{−1}(e_{p}) = e_{q}. Let e′_{q} = max_{q}(e_{q}↓ −{e_{q}}). Then, q can construct primary-graph(e_{q}↓) from the graphs primary-graph(e_{p}↓) and primary-graph(e′_{q}↓).

5 Bounded time-stamps

To make the protocol described in the previous section effective, we have to bound the amount of information recorded in the primary graph of each process by limiting the size of the labels used to identify events.

Notice that the procedure for updating primary graphs only checks the labels of events which actually lie in the primary graphs of the sending and receiving processes. Call an event e “current” in I if e belongs to primary(I_{p}) for some process p.

Let N be the number of processes in the system. Since the underlying computation is B-bounded, we know that there are at most N + (B+1)N^{2} distinct events in primary(I_{p}) for process p—there are at most N events in latest(I_{p}), at most B events in each of the sets unack_{q→r}(I_{p}) and at most N^{2} events overall in received(I_{p}). So, at any given time, the number of events across the system which are current is bounded by N(N + (B+1)N^{2}).

From the way primary information is defined, it is clear that all current events are actually send events. Each send event s_{p→q} begins by being current— the moment the message is sent, the event is added to the list of unacknowledged messages from p to q. Eventually, q acknowledges this message and p gets rid of it from its unacknowledged list. Meanwhile, as the computation progresses, this event may get added to the primary information of other processes. However, gradually it recedes into the past, until it drops out of the primary information of all processes. At this time, the label assigned to this event can be reused— the old event with the same label can never become current again.

Processes can keep track of which events in the system are current by maintaining one additional level of data, called secondary information.

Secondary information Let I be an ideal. The secondary information of I is the collection of (indexed) sets primary(e↓) for each event e in primary(I). This collection of sets is denoted secondary(I).

The following lemma says that the only p-events which can be current in the system are those which occur in p’s secondary information.

Lemma 6 Let I ⊆ E be an ideal and e a p-event which belongs to primary(I_{q}) for some process q. Then, e ∈ secondary(I_{p}).

A proof of this lemma can be found in the full paper [8]. We will use the preceding result in the following form.

Corollary 7 Let e be a p-event such that e ∉ secondary(I_{p}). Then e ∉ primary(I_{q}) for any q ∈ P.

Our update procedure does not rely on the temporal order of events. So long as all processes which refer to the same label in their primary information are actually talking about the same event, reusing labels should cause no confusion. Therefore, if p knows that no p-event labelled ℓ is currently part of the primary information of any process in the system, it can safely use ℓ to time-stamp the next message which it sends.

Secondary information can be updated in a straightforward manner when we update primary information—if q inherits an event e from p’s primary information, it also inherits the secondary information primary(e↓) associated with e. Notice that it suffices to maintain secondary information as an indexed set—we do not need to maintain secondary graphs as we do primary graphs.

This at once gives us a protocol which solves the gossip problem for Bbounded fifo computations.

The gossip protocol

Let L be a finite set of labels of such that |L| > (N^{2} +(B+1)N^{3}). All processes use the set L to time-stamp messages. Each process p maintains its primary graph primary-graph_{p} = (V_{p}, E_{p}), where V_{p} consists of the (indexed) sets of labels latest_{p}, unack_{p} and received_{p}.

A typical element of latest_{p} is a pair of the form (ℓ, q)—this will mean that the maximum q-event known to p is time-stamped ℓ. Similarly, elements of unack_{p} and received_{p} are triples. An entry (ℓ, q, r) in unack_{p} signifies that, as far as p knows, the s_{q→r} event labelled ℓ has not been acknowledged. In the same vein, a typical entry (ℓ, q, r) in received_{p} denotes that, as far as p knows, the most recent message from q to r which has actually been delivered is time-stamped ℓ.

Finally, the process p maintains its secondary information secondary_{p} as an indexed set of labels. If ē is a tuple from latest_{p} ∪ unack_{p} ∪ received_{p}, then an event in latest(ē↓) will be represented as (ℓ′, r, ē), indicating that the maximum r-event in ē↓ is time-stamped ℓ′. In a similar manner, an entry (ℓ′, r, s, ē) in unack(ē ↓) signifies that there is a s_{r→s} event time-stamped ℓ′ which is unacknowledged within ē↓. And, finally, an entry (ℓ′, r, s, ē) in received(ē↓) means that the most recent message from r to s which has been delivered within e↓ is timestamped ℓ′.

Initially, for each p, latest_{p}, unack_{p}, received_{p} and secondary_{p} are empty.

Sending a message When p sends a message to q it does the following:

• Choose a label ℓ from L which does not appear as the first component of any tuple in latest_{p} ∪ unack_{p} ∪ received_{p} ∪ secondary_{p}.

• Remove the old event (ℓ′, p) from latest_{p}, if it exists. Also remove all associated events from secondary_{p}—i.e., tuples of the form (ℓ′′, p′, ℓ′, p) and (ℓ′′, p′, p′′, ℓ′, p).

• Add (ℓ, p, q) to unack_{p} and (ℓ, p) to latest_{p}. Add an edge in E_{p} from each tuple in latest_{p} ∪ unack_{p} ∪ received_{p} to the new tuples (ℓ, p, q) ∈ unack_{p} and (ℓ, p) ∈ latest_{p}.

• For each pair (ℓ′, p′) in latest_{p}, add (ℓ′, p′, ℓ, p) to secondary_{p}. Similarly, for each triple (ℓ′, p′, p′′) in unack_{p} ∪ received_{p}, add (ℓ′, p′, p′′, ℓ, p) to secondary_{p}.

• Send primary-graph_{p} and secondary_{p} to q.

Receiving a message On receiving a message from p, q does the following:

• Extract the label ℓ of the new message.

• Replace the current triple (ℓ′, p, q) in received_{q}, if it exists, by (ℓ, p, q).

• If (ℓ, p, q) does not already belong to unack_{q} then update primary-graph_{q} and secondary_{q} by comparing primary-graph_{p} and secondary_{p} in the message with primary-graph_{q} and secondary_{q} currently maintained by q.

So, on sending a message, p chooses a label ℓ which is not currently in use in the system and uses ℓ to time-stamp the message. It then replaces the latest p label in latest_{p} by ℓ and also adds ℓ to the list of unacknowledged s_{p→q} events. Finally, it places the new event at the “top” of its primary graph and sets the secondary information with respect to the new event to be its overall primary information. It then sends its new data structures primary-graph_{p} and secondary_{p} to q.

When q receives the message labelled ℓ, it first records in received_{q} that the most recently received message from p is labelled ℓ. (Extracting the label of the message can be done by looking, for instance, for the E_{p}^{∗}-maximal event in primary-graph_{p}). It then checks whether the message is new. If so, it updates its primary graph and secondary information following the results in Lemmas 1 and 2 and Proposition 3.

Message complexity

Lemma 8 Let N be the number of processes in the system. For each process p, the information in primary-graph_{p} and secondary_{p} requires at most O(N^{4}logN) bits to write down.

Proof We know that |L| is O(N^{3}). So each label in L can be written down using O(log N) bits. Similarly, each process name can be written down using O(log N) bits. So each tuple in the sets latest_{p}, unack_{p}, received_{p} and secondary_{p} requires only O(log N) bits to write down.

There are O(N^{2}) entries in primary_{p}. So, the edge relation E_{p} of primarygraph_{p} can be represented in terms of an adjacency matrix, using O(N^{4}) bits. So, all of primary-graph_{p} can be written down in O(N^{4}) bits.

The real bottleneck turns out to be secondary_{p}. There are O(N^{2}) elements in primary_{p}. For each of these elements ē, we have to maintain primary(ē↓), which requires O(N^{2} log N) bits. So, overall secondary_{p} requires O(N^{4} log N) bits to write down.

□

Putting together all the results we have proved so far, we can state the following theorem

Theorem 9 The protocol we have described solves the gossip problem for fifo, B-bounded computations with only a bounded amount of additional information being added to each message of the underlying computation.

6 Non-fifo computations

We briefly describe how to extend our protocol to deal with B-bounded computations which are not fifo. We do not go into all the formal details here due to a lack of space. Details are available in the full paper [8].

The essential difference is in the procedure for collecting acknowledgements. In a non-fifo computation, messages may be received out of order. So, we cannot infer from the fact that e = received_{q←p}(I) that all messages sent by p to q before e have actually reached q. In general, we have to explicitly acknowledge each message label to the sending process. This can be done by recording for each pair of processes p, q, the set received-set_{q←p}(I) consisting of all events e of the form s_{p→q} such that ϕ^{−1}(e) belongs to I.

So, we extend the information maintained by each process p to include the sets received-set_{r←q}(I_{p}) for each pair of processes q, r. These sets can be updated using an extension of Proposition 3. The problem is, of course, that these sets grow without bound as the computation proceeds.

However, it can be shown that once e is added to received-set_{q←p}(I_{q}) by q, it will become known to p (i.e., it will appear in received-set_{q←p}(I′_{p}) for I′ ⊇ I) within a bounded number of communications in the system. This follows from the assumption that the underlying computation is B-bounded. Once e appears in received-set_{q←p}(I′_{p}), it has been acknowledged and can be removed from q’s list of received messages.

As a result, it turns out that each process p needs to retain only the B-most recent events in received-set_{r←q}(I_{p}), where “most recent” refers to the order in which the messages were received by r and not the order in which they were sent by q. So, acknowledgments can still be propogated using a bounded amount of information, even when messages arrive out of order and our protocol can be extended to solve the gossip problem for arbitrary B-bounded computations.

References

[1] R. Cori, Y. Metivier: Approximations of a trace, asynchronous automata and the ordering of events in a distributed system, Proc. ICALP ’88, LNCS 317 (1988) 147–161.

[2] R. Cori, Y. Metivier, W. Zielonka: Asynchronous mappings and asynchronous cellular automata, Inform. and Comput., 106 (1993) 159–202.

[3] D. Dolev, N. Shavit: Bounded concurrent time-stamps are constructible, Proc. ACM STOC (1989) 454–466.

[4] A. Israeli, M. Li: Bounded time-stamps, Proc. 28th IEEE FOCS (1987) 371– 382.

[5] R. Krishnan, S. Venkatesh: Optimizing the gossip automaton, Report TCS-94-3, School of Mathematics, SPIC Science Foundation, Madras, India (1994).

[6] L. Lamport: Time, clocks and the ordering of events in a distributed system, Comm. ACM 17(8) (1978) 558–565.

[7] L. Lamport, N. Lynch: Distributed Computing: Models and Methods, in: J. van Leeuwen (ed.), Handbook of Theoretical Computer Science: Volume B, North-Holland, Amsterdam (1990) 1157–1200.

[8] M. Mukund, K. Narayan Kumar, M. Sohoni: Keeping track of the latest gossip in message-passing systems, Report TCS-95-3, School of Mathematics, SPIC Science Foundation, Madras, India (1995).

[9] M. Mukund, M. Sohoni: Keeping track of the latest gossip: Bounded timestamps suffice, Proc. FST&TCS ’93, LNCS 761 (1993) 388–399.