MTP: An Atomic Multicast

Transport Protocol

Alan O. Freier

Keith Marzullo*

TR 90-1141

July 1990

Department of Computer Science

Cornell University

Ithaca, NY 14853-7501

*Keith Marzullo is supported in part by the Defense Advanced Research Projects

Agency (DoD) under NASA Ames grant number NAG 2-593, Contract N00140-87-C8904. The views, opinions, and findings contained in this report are those of the

authors and should not be construed as an official Department of Defense position,

policy, or decision.

MTP: An Atomic Multicast Transport Protocol

Alan O. Freier

Networking and Communications

Apple Computer, Inc.

Cupertino, CA 95014

freier_apple.com

Keith MarzuUo "

Cornell University

Computer Science Dept.

Ithaca, NY 14853

marzullo@cs.cornell.edu

July 30, 1990

Abstract

This paper describes MTP: a reliable transport protocol that utilizes the multicast strategy of applicable lower layer network architectures. In addition to transporting data reliably and efficiently, MTP

provides the client synchronization necessary for agreement on the receipt of data and the joining of the group of communicants.

Keywords: reliable transport, multicast, broadcast, atomic broadcast,agreement.

1 Introduction

A multicast transport is a virtual circuit connection among a set of communicating peer-level processes. As such, any multicast transport protocol has

to satisfy somewhat conflicting goals. Being a transport protocol, it should

supply quick and reliable delivery of large amounts of client data among the

communicants. Yet, being a multicast protocol, it should be able to supply

the ordering and agreement on the delivery of the data that is necessary

for writing decentralized applications. Agreement on order and delivery can

"Keith Marzullo is supported in part by the Defense Advanced Research Projects

Agency (DoD) under NASA Ames grant number NAG 2-593, Contract N00140-87-C8904. The views, opinions, and findings contained in this report are those of the *uthors

and should not be construed as an official Department of Defense pc_sition, policy, or

decision.
MTP: An Atomic Multicast Transport Protocol

take time, thereby slowing the delivery of the data. Hence, most multicast protocols concentrate on a smaller set of goals; for example, [CP88]

and [CW89] concentrate on fast delivery while [KTHB90] concentrates on

the fast ordered delivery of relatively small messages.

MTP, the transport described in this paper, attempts to satisfy both of

these goals. MTP is a full-duplex, flow-controlled, reliable multicast protocol

in which the data is sequenced into (perhaps long) messages. Messages are

sent within a process group called a web, where each message has a single

sender and is received by all members of the web. The members of the web

agree on the order of receipt of all messages and can agree on the delivery

of the message even in the face of partitions I.

MTP can be thought of as two protocols: a transport layer running underneath an ordering and agreement layer. The transport layer is a negative

acknowledgement (or NAK) based protocol exploiting the high probability

of successful message delivery that the local area networks of today provide [CLZ87]. Additionally, this transport utilizes the underlying data link

and physical layer's capability to do multicast addressing. The ordering and

agreement protocol uses a sequencer site [CM87,KTHBg0] called the master

that grants serialized tokens to producers.

The rest of this paper proceeds as follows. In Section 2, the class of

applications for which MTP is meant is contrasted with those applications

other atomic broadcast protocols support. The protocol is presented in

Section 3. Suggestions for values of the protocol's parameters are derived in

Section 4, and a discussion of MTP is given in Section 5.

2 Applications

MTP is designed to support applications that consist of a large number of

processes, where the processes send large messages and where the application must be fault-tolerant (we consider crash failures of processes and

communication link failures that can lead to partitioning). Examples of

such applications include multimedia teleconferencing systems, multiscreen

educational systems, and stock brokerage systems. In making this assumption, we intentionally exclude certain classes of applications that have been

considered elsewhere; in particular, those structured as client-server systems

with highly available services (e.g. [Sch86,MS88]).

tA partition is the separation of a network of processes into two or more disjoint sets

that cannot communicate with each other.
MTP: An Atomic Multicast Transport Protocol

O_e issue that MTP must address is the efficient handling of network

partitioning. An argument can be made that transient partitioning is a

very common failure in the kind of applications we are considering [Cri90].

Timeouts are used to detect both crash failures and communication failures.

If a machine uses a timeout period that is too short, then it will appear to

the machine that the network has temporarily partitioned. For CSMA/CD

type data links, there is no upper bound on message delay (communication

and operating system software can also increase the variance of this delay),

so such transient partitions will be unavoidable. The application designer

must balance the cost of recovery from partitioning against the penalty of

using excessively long timeouts. Additionally, packets can be dropped due

to temporary congestion at both routers and workstations, again creating

transientpartitions.

Our approach totoleratingpartitionsisto choose one processin the web

to be a distinguishedprocesscalledthe master. Sincean MTP web contains

such a distinguishedprocess,partitionscan be treatedin the same way as

crashor timing failures.Ifthe master processP0 cannot communicate with

a member processPI, then P0 assumes that Pl has failed.Ifpl has instead

partitionedaway from Po,Pl willknow that P0 considersPl to have failed

and behave accordingly.The vulnerabilityof a web to the failureof the

master isa matter of concern, however. Ifthe applicationis to be longlived,care must be taken in choosingthe machine that runs the master. In

Section5.2,we discusssome techniquesformaking a master more robust.

Most other atomic broadcast algorithmsare structuredina very decentralizedmanner so the failureof any (usuallysize-bounded)subset of the

processeswillnot cause the applicationto fail.Being fault-tolerantin this

manner isvery important for implementing highly-availableservices,but

itmeans that the complex issueof toleratingpartitionsin a decentralized

manner must be addressed [DGMS85] 2.

A more detaileddescriptionon the issuesand uses of reliablebroadcast

protocolscan be found in [JB89].

_A notableexampleofanatomicbroadcastprotocolthatdoesnothavea decentralized

structureisdescribedin[KTIIB90],althoushaspresented,thisprotocolcannottolerate

partitionin 8.
MTP: An Atomic Multicast Transport Protocol

3 Protocol

Section 3.1 describes the overall structure of an MTP web. In Section 3.2, the

ordering and agreement protocol is described assuming an abstract transport protocol. In Section 3.3, the transport protocol is described, and in

Section 3.4 the ordering and agreement protocol is extended to support the

establishment of a web and the joining of a member.

3.1 Web Structure

An MTP web consists of a master process and a set of member processes.

Member processes may join and leave the web, but the master process cannot, as the web is both instantiated and terminated by the master. A]J

data is reliably multicast: that is, every process agrees on the order that

a given message will be processed, and the transport guarantees that any

given message is either accepted by all non-failed processes or not accepted

by any non-failed processes.

There are four transport service access points (TSAPs) associated with

a given web:

1. Multicast transport addresses: These are the addresses to which all

messages targeted for the entire web axe transmitted. Each consists

of a multicast network service access point (NSAP) catenated with a

unique transport connection identifier.

2. Master's transport address: This is the TSAP for the master process.

This address is the destination of messages for the master process,

such as requesting a token or leaving the web. This address is also the

source of any message sent by the master process.

3. Join transport address: This is the NSAP for the service 3 catenated

with the predefined join transport connection identifier. This address

is the destination of all requests to join the web.

4. Member transport addresses: These axe the addresses of all the processes that are currently members of the web. Each consists of the

member process NSAP catenated with a unique transport connection

identifier. The source of any packet transmitted by a process, regardless of the packet's destination, is a member of this set.

_Determining this multicut NSAP for a given instantiation is not a function performed

by MTP.
3ITP: An Atomic Multicast Transport Protocol

3.2 Sequencing Messages

The agreement and ordering layer of MTP ensures that aJ.[ processes agree on

which messages are accepted and in what order they are accepted. Let pi be

a member process and Mi be the sequence of messages that Pi has delivered

to its client. The agreement layer ensures the following two properties:

AB-1 The sequence of messages that processes have delivered to the clients

do not diverge; that is, for all processes Pi and pj, Mi is a prefix of Mj

or fv/j is a prefix of Mi.

AB-2 There exists a connected _ubset of the nonfaulty (i.e. noncrashed)

processes that make progress.

Figures 1, 2 and 3 shows pseudocode for the MTP agreement and ordering protocol. In these figures, the primitive send p x sends the message

x to process p without blocking, the primitive receive p x is a CSP-like

guard [Hoa78] that receives a matching message from process p and stores

it into z, and multlcast P x multicasts the message z to the processes in

the set P without blocking. The predicate failed(s) represents a timeout; it

will become true at some point after the processor that was issued the token

for message number s has crashed or remained partitioned away from the

master.

To send a message m to a web, a member process first requests one of a

set of t tokens from the master of the web. This token contains:

• the message number to be assignedto m,

• the multicast transport addresses as discussed in Section 3.1,

• the status of the last t messages. Such messages can be accepted,

rejected, or pending. Furthermore, the earliest of these t messages

must either be accepted or be rejected.

The master sets the status of the last t messages using the following rule.

Let m is one of these last t messages:

• if the master has seen the message m, then the status is accepted 4;

4The master has seen message m when it has received a data packet of message m

containing an end o] message indicator; see Section 3.3.
MTP: An .4 tomic ,_fultica, st Transport Protocol 6

process Master

begin

members: integer set := { ... };

status (1.. t): Status := undefined ..... undefined;

t: constant integer; the number of tokens

next: integer := 1;

s: integer;

m: Data;

last (1_ t): Status;

do receive Sender(i: 1..n) ["token_request"] and status(t - 1) # pending --

begin

status(l_ t) := pending, status(1., t -- 1);

next := next + 1;

send Sender(i) ["token_grant", next, status, members]

end

receive Receiver(i: 1..n)["data", s, m, last] --*

if next -- s _< t and status(next - s) = pending

then status(next - s) := accepted;

failure(s) and next - s _< t

and status(next - s) = pending --* status(next - s) = rejected

O

od

end

Figure 1: Agreement Protocol for Web Master

• if the master has not seen the message m but the sender of m is

still operational and connected to the master (as determined by the

master), then the status is pending;

• otherwise, the status is rejected.

An abbreviated proof of this protocol is presented in the Appendix. Informally, the specification is met because the behavior of the web is defined

by the behavior of the master. In particular, a member process accepts a

message m only ff the master accepts m, and all messages are accepted in

the order of their message sequence numbers; thus, AB-1 is met. We define the connected subset of correct processes referred to in AB-2 as those

processes S that remain connected to the master. The master will accept

messages sent by processes in S and possibly reject other messages, and the
MTP: An Atomic Multicast Transport Protocol

process Sender(i:i..n)

begin

last(i..t):Status;

members: integer set;

s: integer;

m: Data;

do receive producer(i) [m] -*

begin

send Master ["token_ request"];

receive Master ["token_grant", s, last, members];

multicast {Master} U Receiver(members) ["data", s, last, m]

end

od

end

Figure 2: Agreement Protocol for Web Producer

members of 5" will in turn accept and reject these same messages as other

messages are sent 5.

Having obtained a token, a multicasts message m with the token for message m included in the header of the data packets that carry m. Processes

learn the status of earlier messages by seeing such packets, and can accept

and reject messages accordingly. This protocol can tolerate up to a sequence

of t failures; if there are t + 1 failures, then the master could send tokens

to these processes which could then fail before any nonfaulty process sees

any data sent with these t + 1 tokens. The headers of these tokens carry

information about the status of earlier messages, and since no other process

received any data seat with the earliest token, the status of some message

will never be propagated to the members of the web.

3.3 Sending Messages

The transport multicast layer of MTP is implemented using the multicast

capability provided by the network layer (which in turn is provided by the

data link and physical layers). For the purposes of this paper, we assume

SAs written, a message can be acknowledged (and hence delivered) only when another

message is sent. However, the master can send empty packeta, defined in Section 3.3, in

order to expedite the delivery of a message when subsequent messages axe slow in being

generated.
MTP: An Atomic Multicast Transport Protocol 8

process Receiver(h 1.. n)

begin

data (1..): Data :-- empty .. ;

status (1..): Status := pending .. ;

nextln, nextOut: integer := 1, 1;

last (I.. t): Status;

s: integer;

m: Data;

do receive Receiver(j: 1..n) ["data", s, last, m] --

begin

k: integer :-- 2;

data(s - 1):'- m;

do k _ t -_ status(s - k) := last(k); k := k + [ od;

nextln := max s, nextln

end

_] receive consumer(i) and status(nextOut) = accepted

if data(nextOut) _ empty--.

send consumer(i) [data(nextOut)]; nextOut := nextOut + 1

0 data(nextOut) = empty .-.* rejoin

fi

_] status(nextOut) -- rejected-.* nextOut := nextOut -'F 1

0 status(nextOut) = pendin 9 and (nextln - nextOut) > t --, rejoin

od

end

Figure 3: Agreement Protocol for Web Consumer

that a multicast to all of the processes in a web can be accomplished by

performing multicasts to a small number of transport service access points

(TSAPs)--no more than can be included in the data portion of an MTP

packet. Network facilities similar to those described in [DC90] support this

facility, but are not necessary for MTP to operate.

The transport layer treats a message as an uninterpreted sequence of

bytes terminated by an end of message marker. The transport layer fragments a message into a sequence of packets. Each packet carries a sequence

number of the form (re,p) where m is the message number and p is the packet

number in this message, starting at zero. For example, if message 5 were

broken into 3 packets, then the packets would be sequenced as (5,0), (5, 1)
MTP: An Atomic Multicast Transport Protocol

and _5,2) (of which the last would carry an end of message marker), and

the next packet would be sequenced as (6, 0).

There are three parameters that control the flow of data in the transport

layer. They are:

• heartbeat: A base unit of time, in milliseconds.

• window: The maximum number of data packets a producer can send

during any heartbeat.

• retention: The maximum number of heartbeats a producer must

buffer packets for possible retransmissions.

Data is transmitted in a burst of packets such that no more than the

current window of data packets will be sent during a single heartbeat. Every

packet transmitted (including control packets) always contains the latest

heartbeat, window and retention information along with the statuses of the

previous t messages and the next message sequence number. If full packets

are not available 6, empty packets will be transmit ted instead (defined below).

The only data packets that will be transmitted containing less than the

maximum capacity will be those that mark a client state transition.

A empty packet is a control packet that is multicast into the web at regular intervals whenever the producer owning a token cannot transmit client

data. Empty packets are sent to maintain synchronization and to advertise

the maximum sequence number of the producer. Empty packets provide the

opportunity for consuming processes to detect and request retransmission

of missed data as well as identifying the owner of a transmit token.

If a producer receives a NAK from a consumer requesting the retransmission of one or more packets, those packets will be multicast to the entire web

or to a selected subset of the multicast TSAPs. All retransmitted packets

will contain the original client information and sequence number. However,

the retransmitted packets will contain updated parameter information (the

heartbeat, window and retention). As no more than than a full window of

data messages can be sent during one heartbeat, retransmitted packets have

priority over new packets during the next heartbeat.

The producer is obligated to retransmit a packet upon request for at least

retention heartbeats after its original transmission (even after the message

has been completely sent). If the producer receives a NAK from a consumer

SThe resource being flow controlled is a packet carrying client data. Consequently, full

packets provide the greatest efficiency.
MTP: An Atomic Multicast Transport Protocol 10

process requesting the retransmission of a packet that is no longer available.

the producer sends a nak deny to the source of the request. If the consumer

cannot recover from the loss of this packet, then the consumer rejoins the

web to resynchronize.

Figure 4 shows a space-time diagram of a process transmitting into a

web assuming no NAKs, and Figure 5 illustrates data transmission and

NAK processing.

3.4 Consistency and Joining the Web

A process pt may become unrecoverably inconsistent with the master of the

web for several reasons. The most likely reason is that ps has partitioned

away long enough from the master so that Pi missed learning the status of a

message. A less likely scenario is that some process Pi transmits a message

that is received by the master but not by Pi, and p_ crashes before pi can ask

for retransmission of the missed packets. In any case, when a process finds

itself inconsistent with the master, it can resynchronize itself by rejoining

the web.

As described in Section 3.1, the master of a web constructs the master

transport address by catenating the NSAP with a locally generated unique

transport connection identifier. A process that wishes to join or rejoin the

web will send a join request message to the join transport address, and the

master will answer with a join response carrying a source of the master

transport address. Note that a rejoining process can determine whether the

web is the same session with which it became inconsistent by comparing

the previous and new transport connection identifiers it obtained in the join

confirm messages.

In general, a process that repeatedly receives no join confirm cannot

elect itself the master. Another process may follow the same reasoning in

another partition, and then if the partition were to end, there would be two

inconsistent webs with undesirable properties; for example, a third joining

process would nondeterministically join one of the two existing webs. Any

"merging" of such inconsistent webs would have to be done outside of MTP,

as the semantics of such a merge would depend on the application. A better

method for master selection would be for a process to know a priori if it

were the master or not. Doing so would both guarantee that there exists

only one active web with a given NSAP and would allow the master to be

located on a machine that is known to be reliably available.

Having joined a web, a process p must be informed which message it
MTP: An Atomic Multicast Transport Protocol 11

should first accept. If p does not need to be given any state in order to

process the next message, then the master can immediately reply to the join

request message with a join confirm message containing the sequence number

s of the next token the master will hand out. Then, the joining process p

need only start receiving messages with sequence numbers greater than or

equal to s. However, for some applications p would need to be initialized

with the state of the web after all message before s have been accepted or

rejected. In this case, having received a join request, the master will stop

granting token requests and will delay sending a join confirm message to p

until all message before s have been accepted or rejected. Then, the master

can respond with the join confirm, p's state can be initialized (either by

having the master send the state or through a protocol outside of MTP),

and the master can resume granting tokens.

Figure 6 shows a space-time diagram illustrating the sequence of messages during a join with a transfer of state from the master.

4 Parameter Values

The values of heartbeat, window and retention can be adjusted by the transport to reflect the capability of the members, the type of application being

supported and the network topology. In general, the producers will try to

drive these numbers towards a higher performance level, and the consumers

will try to drive these numbers towards a higher reliability level. By doing

so, both are trying to optimize the quality of service.

Producers can try to improve the performance by reducing the heartbeat

interval and by increasing the window size. This will have the effect of

increasing the resources committed to the transport at any time. To level

the resource commitment, the producer may also reduce the retention. In

the worst case, a producer must commit enough storage to hold window size

x retention maximum-size packets for heartbeat x retention milliseconds.

Consumers must rely on their clients to consume the data occupying the

resources of the transport. The consumer transport implementation must

monitor the level of committed resources in order to ensure that resources

are not overcommitted. Since MTP is a NAK-based protocol, the consumer

is required to inform a producer if a change in parameters is required. A

consumer must be capable of committing at least t times the memory committed by a producer.

For more reliable operation, a consumer would try to extend the heart-
MTP: An Atomic Multicast Transport Protocol 12

beat interval and increase retention. This has the effect of increasing the

resources needed to support the transport. To counteract this, the consumer

could reduce the window.

In order to make these parameters more concrete, consider MTP running

on a collection of 1-MIP workstations with local industry-standard disks,

communicating over a IEEE 802.3 local area network. The heartbeat is

approximately the transport time constant. Assuming that the transport

can be modeled as a closed loop function, reaction to feedback into the

transport should settle out in three time constants. In a transport that is

constrained to a single network, the dominant cause of processing delay will

most likely be the page fault resolution time. The time to service a page

fault is overwhelmingly the disk access time, and for the current industrystandard disks, around 40 milliseconds is the average worst-case access time.

In the worst case, this time could double in order to reclaim a dirty page.

Allowing for additional overhead and scheduling delays, two times the worst

case page fault resolution time should be a suitable minimum transport time

constant, which is 160 milliseconds.

The window is the number of packets that can be consumed during

one heartbeat. For IEEE 802.3 local area networks, the transmit time per

packet is 1.2 milliseconds for a full packet of 1500 bytes. The processing

time on a 1-MIP machine running Unix should be around 5 milliseconds for

a full packet (where 2.5 - 3 ms of this is incurred by the operating system).

Assuming that the data for the packet originated from a disk backing store

and that disk service overhead is comparable to network service overhead,

the resulting overhead is 11.2 milliseconds per packet, corresponding to a

bandwidth of 1 Mbit/sec. During a heartbeat of 160 milliseconds 14 packets

can be sent, so the maximum window would be approximately 14 packets

per heartbeat.

At worst, each producer could consume 10 percent of the available network bandwidth, so MTP will not be limited by the network bandwidth.

Each producer consumes about 80 percent of the consumer's processing

time, so having more than one producer outstanding could saturate a consumer, ttowever, to a point, having multiple tokens allows some producers

to acquire a token shortly before it is required (presumably overlapping the

transmission of an earlier message) without locking out another producer.

Additionally, increasing t decreases the average message delivery time (until

thrashing becomes a problem). Since the peak resource requirement scales

linearly with t, a reasonable value of t would probably be two or three.

Redudng retention may introduce instability because a consumer will
MTP: .4a Atomic Multicast Transport Protocol 13

have less opportunity to react to missing data. Data can be missed for

a variety of reasons. If constrained to the local net, the data lost due to

corruption should be around one packet in 50,000 r. Four orders of magnitude

more packets are lost at receiving stations, including packet switch touters,

than over physical links. The losses are usually the result of congestion and

resource starvation at lower layers due to the processing of (nearly) back to

back packets. One can only require that a receiving station be capable of

receiving some number of back to back packets successfully, and that number

must be at least greater than the window size. The probability of success

can be made as high as needed by providing the receiver the opportunity to

observe the data multiple times.

At worst, the receiving station detects packet loss using timers. Such

timers might have a granularity of more than two orders of magnitude

greater than the maximum packet transmit time. As such, the worst case

is much worse than detecting data loss due to gaps in sequence numbers.

When the loss is detected, the response (a NAK) is transmitted and should

be received at the producing process in less than two heartbeats after the

data it references was transmitted. Again, it is the detection time that dominates, not the transmission of the NAK. NAKs are also subject to loss, but

the probability of delivery can be made close to one by retransmitting. In

order to be able to respond to a second NAK, the minimum retention is

three.

The resources committed to a transport using the above assumptions are

buffers sufficient for 126 packets of 1500 bytes each, and each buffer will be

committed for at least 480 milliseconds.

The parameters would be very different for a web that spans an internetwork of several LANs, and could be adjusted to accommodate the properties

of the network. For example, if a producer is separated from a set of consumers by a router and the router drops a packet due to congestion then all

of the consumers will simultaneously send NAKs, further aggravating the

congestion. To avoid this burst of NAKs, the master could have previously

set the web's retention to .f + 3 for some positive value of f. Each NAKing

consumer would then dally for some number of heartbeats between 0 and

f before NAKing a missed packet. Not only would this dallying reduce the

number of simultaneous NAKs by a factor of f, but most processes would

probably receive the retransmission without sending a NAK.

7Telephone links (between touters, for example) are capable of exhibiting similar corruption rates.
MTP: An Atomic Multicast Transport Protocol 14

5 Discussion

5.1 Number of Tokens

In Section 4, it was argued that a reasonable number of tokens would be

around two or three. It isn't clear what the number of tokens should be

when a web spans a larger collection of networks. On one hand, having more

tokens allows more processors to pre-allocate tokens, thereby overlapping the

longer round-trip message time with (hopefully) other processing. On the

other hand, the maximum number of buffers increases with the number of

tokens, and processors distant from the master are more likely to partition

away from the master, thereby increasing the number of failures.

One can allow the master to find a balance by varying the number of

tokens. This is done by logically splitting t into the two values tmaz, which is

the maximum number of tokens that can be outstanding and is the number of

message statuses carried in a header, and tcur, which is the current maximum

number of tokens that can be outstanding and need be known only by the

master. The number of failures that can be tolerated is determined by

tm,,z (see the discussion in the Appendix). The master could then vary tcur

between 1 and tmoz depending on the web performance.

5.2 Resiliency Against Failure of the Master

The main vulnerability of MTP is that the failure of the master can cause

the web to fail. For some applications (e.g., a stock brokerage system),

such a failure could be intolerable. In this case, it would become desirable

to replicate the web master. Replicating the master for high tolerance to

processor failure can be done without changing MTP, but having a replicated

master would be noticed by the members as an increase in the response time

to a token request (and less importantly, to a join request).

All the master replicas plo,P2o,...I_o would reside on an unpartitionable

network (for example, a single local area network), guaranteeing that if a

member p_ is connected with p01 and member/_ is connected with p02, then

pl is connected with p_ and p2 is connected with p01. The web's master

TSAP would be a multicast address for these replicated masters.

The masters would choose one amongst themselves to be the coordinator,

with the rest being cohorts [BJ87]. Any replica receiving a request would

atomically broadcasts the request to all the master replicas before the coordinator would respond. Similarly, when the coordinator decides that a
MTP: An Atomic Multicast Transport Protocol 15

message becomes accepted, the coordinator would first atomically broadcasts this fact to all the master replicas s. If the coordinator were then to

fail, one cohort would become the new coordinator. This new coordinator

would reject all messages that it considered pending and start responding to

master requests.

5.3 Web Membership

One issue we have not discussed in this paper is how a process can determine

the current membership of a web. Knowing this information can be very

useful; for example, if all the processes agree on the current web membership,

then each can agree a priori on how work should be partitioned amongst

themselves. The group membership problem is essentially that of having the

web members agree on when a process joins the web and when a process

leaves the web (either by failing, by partitioning away, or under its own

volition) [Cri88,Ricg0]. The difficulty with the group membership problem

is that it really cannot be "solved"; since a process can fail without notifying

any other process, a member of a web cannot be sure whether or not another

process is currently a member. The best that can be done is to have the

web members agree on the membership of the web, and accept the fact that

there may be members that have crashed, and that there may be processes

that, due to the asynchronism in the system, have been excluded from the

web even though they have not crashed or partitioned away 9.

Group membership protocols operate by having processes monitor each

other. If a process id decides that another process p has failed, then p_ uses

some reliable broadcast protocol to disseminate this information to the other

web members [Ricg0]. A common method of detecting whether a process

p has failed or not is to use low-level "alive" messages: other processes

periodically expect such messages from p (perhaps as the result of periodic

SAs stated in this paper, the only time a member learns the status of a message is

when it receives a token or a data message from another member. So, if the coordinator

were to notify the cohorts before granting a token, then the cohorts would be consistent.

However, in the actual protocol the master may send periodic empty packets to expedite

the delivery of messages. If this empty packet s_/vertises a new status, then the coordinator

must inform the cohorts

_Web members must be careful in the deductions they ms_e from the purported group

membership. For example, even if a process p was a member of a web through the delivery of some message m, other web members cannot assume that p actually processed

any message ordered before m unless p specifically acknowledged this fact. To do otherwise would be assuming a solution exists to the coorc]ina|ed attack problem, which is

unsolvable [Gra79].
MTP: An Atomic Multicast Transport Protocol 16

requests), and assume that p has failed if such messages cease to arrive.

Once all web members agree that p has failed (even if it has not), the new

web membership is defined.

Since MTP is a NAK-based protocol, there is no defined low-level "alive"

protocol. A web membership protocol, however can be implemented on top

of MTP as part of the application protocol. Each web member maintains

a set that contains the current web membership. When a process p joins a

web, p multicasts this fact to the web, and all web members (including p)

add p to their membership set when they receive this message. Similarly, if

a process p/decides, for any reason, that another process p has failed, then

p_ multicasts this fact to the web. If p_ is still a member of the web when

this message is delivered, then each process (including f) removes p from

its membership set when it receive this message.

Such membership information is of interest to the master. As discussed

in Section 3.1, the master includes a list of multicast TSAPs in a token

grant message. This list of TSAPs covers the membership of the web as

known by the master, which as currently presented may not be the same

as the membership set described above. The solution in MTP is to allow

a producer and receiver to execute with the master. These processes can

exchange membership changes each observes-the master seeing token losses

and the receiver seeing member-observed failures. By doing so, the master

can remove a multicast TSAP from its list when all processes reached via

that TSAP have left the web, and the producer can multicast the removal

of a member process when that member loses a token.

5.4 Conclusions

MTP is a multicast transport that supports the strong conditions of agreement on delivery, agreement on order and agreement on web membership.

An implementation of MTP is currently under way.

Acknowledgements Susie Armstrong is one of the architects of MTP,

and her careful readings of this paper have greatly increased its clarity.

Fred Schneider aided in formulating the proofs in the appendix, and Aleta

Ricciaxdi helped clarify the issued discussed in the group membership section. Additionally, the authors would like to thank Ken Birmaa, Navin

Budhiraja, Tushar Chandra and Patrick Stephenson for their comments on

the specifications presented here and on early drafts of this paper.
MTP: An Atomic Multicast Transport Protocol 17

Appendix: Specification and proof

This appendix presents a specification and a proof of the ordering and agreement protocol. In interest of brevity, the proof is somewhat informal and

incomplete; in particular, several simple lemmas are stated and used without

proof.

Let p0 be the master process and pl through Pn be the member processes.

The sequence of messages that Pi has delivered to its client is denoted as Mi,

and we write Mi _ Mj to mean that Mi is a prefix of Mj or M i is a prefix

of Mi. Similarly, we will denote by Ai the messages that Pi has marked as

accepted and R, the messages that p, has marked as rejected. Both Ro and

A0 are defined, but as there is no client of the master, M0 is not defined.

The sequence number of a message sent with the statement multicast ...

["data", s, last, m] is s - 1, which we will denote as m.seq lo. We will write

ml < m2 as shorthand for ml.seq < m2.seq A ml E Ao A m2 E .40.

The subset of processes that are not faulty are denoted as C. The state

predicate conn(pi,pj) is true when Pi and p) are connected, the state predicate send(m,pi) is true when Pi sends message m, the state predicates

produce(i) and consume(i) are true when the client on Pi requests a message

to be sent and requests data respectively, and the state function S is a subset

of the processes Pl, P_, • • .P,.

The specification consists of two properties. The first is a safety property,

which specifies that "bad" states do not occur, while the second is a liveness

property, which specifies that "good" states will eventually occur.

AB-I The sequence of messages delivered to the clients do not diverge:

o (vp.pj: ~ M,)

AB-2 There exists a connected subset S of the correct processes C that

make progress:

L°Formslly, uy reference to m is actually a reference to m.meq. The values of R_ and

A_ for i :> 0 are state functions who6e values are defined by the array Producer.status

and Producer.data: if there exists a state in which process pi hu status[k] = accepted and

data[k] _ empty, then in that state m: rn.seq = k: m E At, and if there exists a state

in which process pl has status[k] = rejected, then in that state m: m.seq = k: m E R_.

We can then define m E Mi aa m E At A nextOut > m.Jeq. Similarly, the values of Ro

and ,40 are defined by the array Muter.status; if in some state status[k] -- accepted, then

henceforth m: m.meq = next - k: m E Ao, and if in some state status[k] = rejected then

henceforth m: m.seq = next - k: m 6 Ro.
MTP: An Atomic Multicast Transport Protocol 18

[] (Vm,pi,pj: Pi,Pj E C: send(m,pi) ^ [] (Pi,Pj E S)

Ms)

Our assumptions are:

• All Mi, Ri, and Ai are initially empty;

• the master never fails: Po q C;

• conn(pi,pj) is an equivalence relation (i.e., it is symmetric and transitive);

• unbounded fairness is followed in the selection of enabled guards i.e.,

a guard that remains true willeventuallybe selected;

• clientson correctprocessorsalways continue to send messages and

consume messages:

t3 Vpi: Pi E C: O produce(i) ^ 0 consume(i)

Additionally, we will assume without proof that the protocol satisfies the

following three lemmas:

1. The delivery of a message is monotonic:

L1:12 (Vrn,pi: (m E Mi) _ 12 (m E Mi))

2. A process cannot both accept and reject the same message:

r,2:13 (¥pi: (me Ai) =_ (m f[ Ri))

3. Clients receive messages in message sequence number order:

L3:12 (Vrnl,pi: ral E Mi =_

(Vm2: ra2.seq < ml.seq: rn2 E Mi v rn2 E Ri))
MTP: An Atomic Multicast Transport Protocol 19

Showing Safety One can show a program satisfies a safety property E by

finding a property I such that the initial conditions Init imply I, I implies

I, and I implies E. For I, we will use the conjunct of the two predicates

Ii and I2:

11: Vm,pi: (me Ri) _ (m e Ro)

I2: Vml, m2,pi: (rnl < rn2 A ml g M,) ==_m 2 ¢ Mi

Initially, all M; are empty, making the antecedents of I1 and /2 both

false; thus, Init =_ I. To show Ii and I2 implies AB-1, note that together

they state that for all pi, Mi is a prefix of M0. Since Mi is a prefix of M0

and M i is a prefix of Mo, at least one of (Mi, Mj) is a prefix of the other,

meaning Mi _ M i.

We now prove/i =_ O/1. By Zl,/1 can become false only if a member

Pi rejects a message m before P0 rejects m. For Pi to reject m, it received a

data message from a member process pj containing values of s and last such

that last(s - m.seq) = rejected. To send such a data message, Pi must have

received from po a token grant message containing the same values of s and

last. By the definition of R0, m E R0. Thus, I1 =_ n I1.

We now prove/2 =_ [] /2. /2 can become false only if the expression

rnl, m2,p,: (m: < m2Aml _ Mi)Am2 E Mi becomes true for some messages

rnl and ms and member process Pi; that is, pi delivers a message m2 to its

client but has not (yet) delivered ml to its client, where ml and rn2 have

both been accepted by P0 and ml.seq < rn2.seq. By £3, we know that

ml E Mi V ml E R,, and since by assumption m: _[ Mi, we know that

ml E Ri. By/1, we know that ml E R0, but since ml < m2 we know that

rnl E A0. This is a contradiction (it violates L2), so/'2 :_ []/2.

Showing Liveness To show liveness,we willfirstassume that for the

master O(t > next) which implies[](status(t- 1) = nu/l).We will then

show the effectswhen t isassumed to have a more reasonablevalue.

Property AB-2 isexpressedin terms of the setof processes,9;we will

definethissetas p_ G S - conn(po,p,').Rewriting,we get

/3: [](Vm,pi,Pi: Pi,Pi 6 C: send(m,pi)h

[](conn(po, pi) ^ conn(po,pj)) _ 0 m e Mj)

To show I3, we will need the following five liveness properties, of which

/4, h and Is imply I3:
MTP: An Atomic MulticastTransportProtocol 2O

/4: [] (Vm, pi: pi E C: send(re,p,) A 0 (conn(po,pi)) =_ 0 m E Ao)

15: [] (Vm,pj: Pi E C: m E Ao ^ [] (conn(po,pj)) =¢, <> m E Aj)

I6: [] (Vm,p3: pj E C: m • Ro ^ 0 (conn(po,pj)) _ (> m • R3)

/r: [] (Vm: O (m • .40 V m • Do))

Is: [] (Vm,p/: pj • C: ra E Aj ^ rn (conn(po,pj)) _ 0 m E Mj)

For brevity, only an informal proof for Is will be shown. If there are

no pj that satisfy I5, then the lemma is vacuously true, so we will assume

that there is at least one such pj, say pk. By assumption, the producer

on p_ will eventually request a message to be sent, and by finite progress

pk will eventually send a message to Po requesting a token. By fairness and

connectivity, Po will eventually select the guard (status(t- 1) = null). By the

definition of Ao, rn E Ao _ status(next - m.seq) = accepted, which is passed

back to Pk in the token grant message (again by fairness and connectivity).

By finite progress, Pt, will send a message containing the value of last,

and since connectivity is an equivalence, any pj is connected to pk, and will

therefore receive this message. Then, by finite progress pj will eventually

set m • Aj, and the lemma holds.

The effect of letting t to be smaller than the maximum sequence number

is that a nonfaulty process pj that is connected to p0 may not satisfy AB-2;

in particular, Is and/6 may not hold. A sequence of f > t token requests by

processes that appear to fall after having been granted a token will generate

a sequence of f rejected messages. However, when pj receives message m, it

only sets the status for messages m_: mCseq >m.seq - t, so there will be some

message whose status will remain pending. Eventually, nextln - nextOut

will be greater than t and nextOut will point to the pending message, forcing

pj to rejoin. Thus, the algorithm is live only if there are no sequences of

rejected messages with a length of f > t.

References

[BJ87] Ken Birman and Thomas Joseph. Exploitingvirtualsynchrony

in distributedsystems. In Proceedin#sof the Eleventh S_imposium on Operating S_lstem Principles, pages 123-138. ACM

SIGOPS, 1987.
MTP: .4n Atomic Multicast Transport Protocol 21

[CLZ87]

[cx[871

[cP88]

[CriSS]

[Cri90]

[CW891

[DCJ0I

[DGMS851

[Gra79]

[Hoa78]

[JB891

[KTHB90]

D. Clark, M. Lambert, and L. Zhang. NETBLT: A high throughput transport protocol. In Proceedings of ACMSIGCOMM '87

Workshop, pages 353-359, 1987.

J. Chang and M. Maxemchuck. Atomic broadcast. A CM Transactions on Computer Systems, 2(3):251-273, August 1987.

J. Crowcroft and K. Paliwoda. A multicast transport protocol.

In Proceedings of SIGCOMM '88, pages 247-256. ACM, August

1988.

Flaviu Cristian. Reaching agreement on processor group membership in synchronous distributed systems. In Proceedings o/

the 18th International Conference on Fault-Tolerant Computing.

IEEE TCOS, 1988.

Flaviu Cristian. Understanding fault-tolerant distributed systems. Communications of the ACM, 33(8), August 1990.

David Cheriton and Carey Williamson. VMTP as the transport

layer for high-performance distributed systems. IEEE Communications Magazine, pages 37-44, June 1989.

Stephen E. Deering and David R. Cheriton. Multicast routing

in datagram internetworks and extended LANs. ACM Transactions on Computer Systems, 8(2):85-110, May 1990.

S. B. Davidson, H. Garcia-Molina, and D. Skeen. Consistency in

partitioned networks. A CM Computing Surveys, 17(3):341-370,

September 1985.

J. N. Gray. Notes on Database Operating Systems. SpringerVerlag, Munich, 1979.

C. A. R. Hoare. Communicating sequential processes. Communicatiar_ of the A CM, 21(8):666-677, August 1978.

Thomas Joseph and Kenneth Birman. Reliable Broadcast Prot_cols, pages 294-318. ACM Press, New York, 1989.

M. Franz Kaashoek, Andrew S. Tanenbaum, Susan Flynn Hummel, and Henri E. Bal. An eflident reliable broadcast protocol.

Operating Systems Review, 23(4):5-19, October 1990.
MTP: An Atomic Multicast Transport Protocol O0

[ tsss]

[mc9O]

[Sch86]

Keith Marzullo and Frank Schmuck. Supplying high availability with a standard network file system. In Proceedings of the

Eighth International Conference on Distributed Computing Systems, pages 447-455. IEEE Computer Society, June 1988.

AJeta Pdcciardi. A formalism for fault-tolerant applications in

asynchronous systems. In Fourth SIGOPS European Workshop,

September 1990.

Fred B. Schneider. The state machine approach: A tutorial.

Technical Report TR 86-600, Cornell University, Dept. of Computer Science, Upson Hall, Ithaca, NY 14853, December 1986.
MTP: .-in Atomic Multicast Transport Protocol 23

T

h

1

window w = 3

retention r = 2

heartbeat h

data(n)

data(n + 1)

data(n + w - 1)

data(n + w)

data(n + w + 1)

data(n + 2w - I)

empty(n + 2w): n..n + w - 1

can be released

data(n + 2w) with eom:

n + 2..n + w + 2 can be released

Figure 4: Normal Data Transmission
MTP: An Atomic Multicast Transport Protocol 24

window w = 3

retention r - 2

heartbeat h

data(.)

data(n + 1)

data(n + w - 1)

nak(n')

retrans(n')

data(n + w)

data(n + w + 1)

nak(n')

data(n + 2w - 1): n..n + w - 1

can be released

nak deny(n')

data(n + 2w) with eom:

n + 2..n + w + 2 can be released

Figure 5: NAKs and Retransmission
MTP: An Atomic Multicast Transport Protocol

25

token req

token resp[s]

message[s]

token req

token resp[s+ 1]

message[s+ i]

member master

joinreq

join resp[s+ 1]

transfer state

joiner

Figure 6: Joining and State Transfer