In a message passing system, processors communicate by sending messages over communication channels, where each channel provides a bidirectional connection between two specific processors. We call the pattern of connections described by the channels, the topology of the system. This topology is represented by an undirected graph, where each node represents a processor, and an edge is present between two nodes if and only if there is a channel between the two processors represented by the nodes. The collection of channels is also called the network. An algorithm for such a message passing system with a specific topology consists of a local program for each processor in the system. This local program provides the ability to the processor to perform local computations, to send and receive messages from each of its neighbours in the given topology.

Each processor in the system is modeled as a possibly infinite state machine. A configuration is a vector where each is the state of a processor . Activities that can take place in the system are modeled as events (or actions) that describe indivisible system operations. Examples of events include local computation events and delivery events where a processor receives a message. The behaviour of the system over time is modeled as an execution, a (finite or infinite) sequence of configurations () alternating with events (): . Executions must satisfy a variety of conditions that are used to represent the correctness properties, depending on the system being modeled. These conditions can be classified as either safety or liveness conditions. A safety condition for a system is a condition that must hold in every finite prefix of any execution of the system. Informally it states that nothing bad has happened yet. A liveness condition is a condition that must hold a certain (possibly infinite) number of times. Informally it states that eventually something good must happen. An important liveness condition is fairness, which requires that an (infinite) execution contains infinitely many actions by a processor, unless after some configuration no actions are enabled at that processor.

We say that a system is asynchronous if there is no fixed upper bound on how long it takes for a message to be delivered or how much time elapses between consecutive steps of a processor. An obvious example of such an asynchronous system is the Internet. In an implementation of a distributed system there are often upper bounds on message delays and processor step times. But since these upper bounds are often very large and can change over time, it is often desirable to develop an algorithm that is independent of any timing parameters, that is, an asynchronous algorithm.

In the asynchronous model we say that an execution is admissible if each processor has an infinite number of computation events, and every message sent is eventually delivered. The first of these requirements models the fact that processors do not fail. (It does not mean that a processor's local program contains an infinite loop. An algorithm can still terminate by having a transition function not change a processors state after a certain point.)

We assume that each processor's set of states includes a subset of terminated states. Once a processor enters such a state it remains in it. The algorithm has terminated if all processors are in terminated states and no messages are in transit.

The message complexity of an algorithm in the asynchronous model is the maximum over all admissible executions of the algorithm, of the total number of (point-to-point) messages sent.

A timed execution is an execution that has a nonnegative real number associated with each event, the time at which the event occurs. To measure the time complexity of an asynchronous algorithm we first assume that the maximum message delay in any execution is one unit of time. Hence the time complexity is the maximum time until termination among all timed admissible executions in which every message delay is at most one. Intuitively this can be viewed as taking any execution of the algorithm and normalising it in such a way that the longest message delay becomes one unit of time.

In the synchronous model processors execute in lock-step. The execution is partitioned into rounds so that every processor can send a message to each neighbour, the messages are delivered, and every processor computes based on the messages just received. This model is very convenient for designing algorithms. Algorithms designed in this model can in many cases be automatically simulated to work in other, more realistic timing models.

In the synchronous model we say that an execution is admissible if it is infinite. From the round structure it follows then that every processor takes an infinite number of computation steps and that every message sent is eventually delivered. Hence in a synchronous system with no failures, once a (deterministic) algorithm has been fixed, the only relevant aspect determining an execution that can change is the initial configuration. On the other hand in an asynchronous system, there can be many different executions of the same algorithm, even with the same initial configuration and no failures, since here the interleaving of processor steps, and the message delays, are not fixed.

The notion of terminated states and the termination of the algorithm is defined in the same way as in the asynchronous model.

The message complexity of an algorithm in the synchronous model is the maximum over all admissible executions of the algorithm, of the total number of messages sent.

To measure time in a synchronous system we simply count the number of rounds until termination. Hence the time complexity of an algorithm in the synchronous model is the maximum number of rounds in any admissible execution of the algorithm until the algorithm has terminated.

We start with a simple algorithm Spanning-Tree-Broadcast for the (single message) broadcast problem, assuming that a spanning tree of the network graph with nodes (processors) is already given. Later, we will remove this assumption. A processor wishes to send a message to all other processors. The spanning tree rooted at is maintained in a distributed fashion: Each processor has a distinguished channel that leads to its parent in the tree as well as a set of channels that lead to its children in the tree. The root sends the message on all channels leading to its children. When a processor receives the message on a channel from its parent, it sends on all channels leading to its children.

Spanning-Tree-Broadcast

       Initially  is in transit from  to all its children in the spanning tree.
       Code for :
  1    upon receiving no message: // first computation event by  
  2       TERMINATE 
        
       Code for , , :
  3    upon receiving  from parent: 
  4       SEND  to all children 
  5       TERMINATE 

The algorithm Spanning-Tree-Broadcast is correct whether the system is synchronous or asynchronous. Moreover, the message and time complexities are the same in both models.

Using simple inductive arguments we will first prove a lemma that shows that by the end of round , the message reaches all processors at distance (or less) from in the spanning tree.

Lemma 13.1 In every admissible execution of the broadcast algorithm in the synchronous model, every processor at distance from in the spanning tree receives the message in round .

Proof. We proceed by induction on the distance of a processor from . First let . It follows from the algorithm that each child of receives the message in round 1.

Assume that each processor at distance received the message in round . We need to show that each processor at distance receives the message in round . Let be the parent of in the spanning tree. Since is at distance from , by the induction hypothesis, received in round . By the algorithm, will hence receive in round .

By Lemma 13.1 the time complexity of the broadcast algorithm is , where is the depth of the spanning tree. Now since is at most (when the spanning tree is a chain) we have:

Theorem 13.2 There is a synchronous broadcast algorithm for processors with message complexity and time complexity , when a rooted spanning tree with depth is known in advance.

We now move to an asynchronous system and apply a similar analysis.

Lemma 13.3 In every admissible execution of the broadcast algorithm in the asynchronous model, every processor at distance from in the spanning tree receives the message by time .

We proceed by induction on the distance of a processor from . First let . It follows from the algorithm that is initially in transit to each processor at distance from . By the definition of time complexity for the asynchronous model, receives by time 1.

Assume that each processor at distance received the message at time . We need to show that each processor at distance receives the message by time . Let be the parent of in the spanning tree. Since is at distance from , by the induction hypothesis, sends to when it receives at time . By the algorithm, will hence receive by time .

We immediately obtain:

Theorem 13.4 There is an asynchronous broadcast algorithm for processors with message complexity and time complexity , when a rooted spanning tree with depth is known in advance.

The asynchronous algorithm called Flood, discussed next, constructs a spanning tree rooted at a designated processor . The algorithm is similar to the Depth First Search (DFS) algorithm. However, unlike DFS where there is just one processor with “global knowledge” about the graph, in the Flood algorithm, each processor has “local knowledge” about the graph, processors coordinate their work by exchanging messages, and processors and messages may get delayed arbitrarily. This makes the design and analysis of Flood algorithm challenging, because we need to show that the algorithm indeed constructs a spanning tree despite conspiratorial selection of these delays.

       Code for any processor , 
  1  INITIALISATION 
  2    parent  NIL 
  3    children  
  4    other  
        
  5  PROCESS MESSAGE <adopt> that has arrived on link  
  6    IF parent NIL 
  7       THEN parent  
  8          SEND <approved> to link  
  9          SEND <adopt> to all links in neighbours  
 10       ELSE SEND <rejected> to link  
        
 11  PROCESS MESSAGE <approved> that has arrived on link  
 12    children  children  
 13    IFchildren other neighbours {parent} 
 14       THEN TERMINATE 
        
 15  PROCESS MESSAGE <rejected> that has arrived on link  
 16    other other  
 17    IFchildren other neighbours {parent} 
 18       THEN TERMINATE 
       Extra code for the designated processor 
 19  IFparent NIL 
 20    THEN parent NONE 
 21       SEND <adopt> to all links in neighbours 

The leader election problem is to elect exactly leader among a set of processors. Formally each processor has a local variable leader initially equal to NIL. An algorithm is said to solve the leader election problem if it satisfies the following conditions:

Bully elects a leader among asynchronous processors . Identifiers of processors are used by the algorithm in a crucial way. Briefly speaking, each processor tries to become the leader, the processor that has the largest identifier among all processors blocks the attempts of other processors, declares itself to be the leader, and forces others to declare themselves not to be leaders.

Let us begin with a simpler version of the algorithm to exemplify some of the ideas of the algorithm. Suppose that each processor sends a message around the ring containing the identifier of the processor. Any processor passes on such message only if the identifier that the message carries is strictly larger than the identifier of the processor. Thus the message sent by the processor that has the largest identifier among the processors of the ring, will always be passed on, and so it will eventually travel around the ring and come back to the processor that initially sent it. The processor can detect that such message has come back, because no other processor sends a message with this identifier (identifiers are distinct). We observe that, no other message will make it all around the ring, because the processor with the largest identifier will not pass it on. We could say that the processor with the largest identifier “swallows” these messages that carry smaller identifiers. Then the processor becomes the leader and sends a special message around the ring forcing all others to decide not to be leaders. The algorithm has message complexity, because each processor induces at most messages, and the leader induces extra messages; and one can assign identifiers to processors and delay processors and messages in such a way that the messages sent by a constant fraction of processors are passed on around the ring for a constant fraction of hops. The algorithm can be improved so as to reduce message complexity to , and such improved algorithm will be presented in the remainder of the section.

The key idea of the Bully algorithm is to make sure that not too many messages travel far, which will ensure message complexity. Specifically, the activity of any processor is divided into phases. At the beginning of a phase, a processor sends “probe” messages in both directions: CW and CCW. These messages carry the identifier of the sender and a certain “time-to-live” value that limits the number of hops that each message can make. The probe message may be passed on by a processor provided that the identifier carried by the message is larger than the identifier of the processor. When the message reaches the limit, and has not been swallowed, then it is “bounced back”. Hence when the initial sender receives two bounced back messages, each from each direction, then the processor is certain that there is no processor with larger identifier up until the limit in CW nor CCW directions, because otherwise such processor would swallow a probe message. Only then does the processor enter the next phase through sending probe messages again, this time with the time-to-live value increased by a factor, in an attempt to find if there is no processor with a larger identifier in twice as large neighbourhood. As a result, a probe message that the processor sends will make many hops only when there is no processor with larger identifier in a large neighbourhood of the processor. Therefore, fewer and fewer processors send messages that can travel longer and longer distances. Consequently, as we will soon argue in detail, message complexity of the algorithm is .

We detail the Bully algorithm. Each processor has five local variables. The variable id stores the unique identifier of the processor. The variable leader stores TRUE when the processor decides to be the leader, and FALSE when it decides not to be the leader. The remaining three variables are used for bookkeeping: asleep determines if the processor has ever sent a <probe,id,0,0> message that carries the identifier id of the processor. Any processor may send <probe,id,phase, > message in both directions (CW and CCW) for different values of phase. Each time a message is sent, a <reply,id,phase> message may be sent back to the processor. The variables and are used to remember whether the replies have already been processed the processor.

The code of each processor is composed of five segments. The first segment (lines 1–5) initialises the local variables of the processor. The second segment (lines 6–8) can only be executed when the local variable asleep is TRUE. The remaining three segments (lines 9–17, 1–26, and 27–31) describe the actions that the processor takes when it processes each of the three types of messages: <probe,ids,phase,ttl>, <reply,ids,phase> and <terminate> respectively. The messages carry parameters , phase and that are natural numbers.

We now describe how the algorithm works. Recall that we assume that the local variables of each processor have been initialised before time 0 of the global clock. Each processor eventually sends a <probe,id,0,0> message carrying the identifier id of the processor. At that time we say that the processor enters phase number zero. In general, when a processor sends a message <probe,id,phase, >, we say that the processor enters phase number phase. Message <probe,id,0,0> is never sent again because FALSE is assigned to asleep in line 7. It may happen that by the time this message is sent, some other messages have already been processed by the processor.

When a processor processes message <probe,ids,phase,ttl> that has arrived on link CW (the link leading in the clock-wise direction), then the actions depend on the relationship between the parameter and the identifier id of the processor. If is smaller than id, then the processor does nothing else (the processor swallows the message). If is equal to id and processor has not yet decided, then, as we shall see, the probe message that the processor sent has circulated around the entire ring. Then the processor sends a <terminate> message, decides to be the leader, and terminates (the processor may still process messages after termination). If is larger than id, then actions of the processor depend on the value of the parameter (time-to-live). When the value is strictly larger than zero, then the processor passes on the probe message with decreased by one. If, however, the value of is already zero, then the processor sends back (in the CW direction) a reply message. Symmetric actions are executed when the <</sl>probe,ids,phase,ttl</sl>> message has arrived on link CCW, in the sense that the directions of sending messages are respectively reversed – see the code for details.

Bully

       Code for any processor , 
  1  INITIALISATION 
  2    asleep TRUE 
  3    CWreplied FALSE 
  4    CCWreplied FALSE 
  5    leader NIL 
        
  6  IF asleep 
  7    THEN asleep FALSE 
  8       SEND <probe,id,0,0> to links CW and CCW 
        
  9  PROCESS MESSAGE <probe,ids,phase,ttl> that has arrived 
          on link CW (resp. CCW)
 10    IF id ids and leader NIL 
 11       THEN SEND <terminate> to link CCW 
 12          leader TRUE 
 13          TERMINATE 
 14    IF ids id and ttl  
 15       THEN SEND <probe,ids,phase,ttl > 
             to link CCW (resp. CW)
 16    IF ids id and ttl  
 17       THEN SEND <reply,ids,phase> to link CW (resp. CCW) 
        
 18  PROCESS MESSAGE <reply,ids,phase> that has arrived on link CW (resp. CCW) 
 19    IF id ids 
 20       THEN SEND <reply,ids,phase> to link CCW (resp. CW) 
 21       ELSE CWreplied TRUE (resp. CCWreplied) 
 22    IF CWreplied and CCWreplied 
 23       THEN CWreplied FALSE 
 24          CCWreplied FALSE 
 25          SEND <probe,id,phase+1, > 
                to links CW and CCW
        
 26  PROCESS MESSAGE <terminate> that has arrived on link CW 
 27    IF leader NIL 
 28       THEN SEND <terminate> to link CCW 
 29          leader FALSE 
 30          TERMINATE 

When a processor processes message <reply,ids,phase> that has arrived on link CW, then the processor first checks if ids is different from the identifier id of the processor. If so, the processor merely passes on the message. However, if , then the processor records the fact that a reply has been received from direction CW, by assigning TRUE to CWreplied. Next the processor checks if both CWreplied and CCWreplied variables are true. If so, the processor has received replies from both directions. Then the processor assigns false to both variables. Next the processor sends a probe message. This message carries the identifier id of the processor, the next phase number , and an increased time-to-live parameter . Symmetric actions are executed when <reply,ids,phase> has arrived on link CCW.

The last type of message that a processor can process is <terminate>. The processor checks if it has already decided to be or not to be the leader. When no decision has been made so far, the processor passes on the <terminate> message and decides not to be the leader. This message eventually reaches a processor that has already decided, and then the message is no longer passed on.

We begin the analysis by showing that the algorithm Bully solves the leader election problem.

Theorem 13.8 Bully solves the leader election problem on any ring with asynchronous processors.

Proof. We need to show that the two conditions listed at the beginning of the section are satisfied. The key idea that simplifies the argument is to focus on one processor. Consider the processor with maximum id among all processors in the ring. This processor eventually executes lines 6–8. Then the processor sends <probe,id,0,0> messages in CW and CCW directions. Note that whenever the processor sends <probe,id,phase, > messages, each such message is always passed on by other processors, until the ttl parameter of the message drops down to zero, or the message travels around the entire ring and arrives at . If the message never arrives at , then a processor eventually receives the probe message with ttl equal to zero, and the processor sends a response back to . Then, eventually receives messages <reply,id,phase> from each directions, and enters phase number by sending probe messages <probe,id,phase+1, > in both directions. These messages carry a larger time-to-live value compared to the value from the previous phase number phase. Since the ring is finite, eventually ttl becomes so large that processor receives a probe message that carries the identifier of . Note that will eventually receive two such messages. The first time when processes such message, the processor sends a <terminate> message and terminates as the leader. The second time when processes such message, lines 11–13 are not executed, because variable leader is no longer NIL. Note that no other processor can execute lines 11–13, because a probe message originated at cannot travel around the entire ring, since is on the way, and would swallow the message; and since identifiers are distinct, no other processor sends a probe message that carries the identifier of processor . Thus no processor other than can assign TRUE to its leader variable. Any processor other than will receive the <terminate> message, assign FALSE to its leader variable, and pass on the message. Finally, the <terminate> message will arrive at , and will not pass it anymore. The argument presented thus far ensures that eventually exactly one processor assigns TRUE to its leader variable, all other processors assign FALSE to their leader variables, and once a processor has assigned a value to its leader variable, the variable remains unchanged.

Our next task is to give an upper bound on the number of messages sent by the algorithm. The subsequent lemma shows that the number of processors that can enter a phase decays exponentially as the phase number increases.

Lemma 13.9 Given a ring of size , the number of processors that enter phase number is at most .

Proof. There are exactly processors that enter phase number , because each processor eventually sends <probe,id,0,0> message. The bound stated in the lemma says that the number of processors that enter phase 0 is at most , so the bound evidently holds for . Let us consider any of the remaining cases i.e., let us assume that . Suppose that a processor enters phase number , and so by definition it sends message <probe,id,i, >. In order for a processor to send such message, each of the two probe messages <probe,id,i-1, > that the processor sent in the previous phase in both directions must have made hops always arriving at a processor with strictly lower identifier than the identifier of (because otherwise, if a probe message arrives at a processor with strictly larger or the same identifier, than the message is swallowed, and so a reply message is not generated, and consequently cannot enter phase number ). As a result, if a processor enters phase number , then there is no other processor hops away in both directions that can ever enter the phase. Suppose that there are processors that enter phase . We can associate with each such processor , the consecutive processors that follow in the CW direction. This association assigns distinct processors to each of the processors. So there must be at least distinct processor in the ring. Hence , and so we can weaken this bound by dropping , and conclude that , as desired.

Theorem 13.10 The algorithm Bully has message complexity, where is the size of the ring.

Note that any processor in phase , sends messages that are intended to travel away and back in each direction (CW and CCW). This contributes at most messages per processor that enters phase number . The contribution may be smaller than if a probe message gets swallowed on the way away from the processor. Lemma 13.9 provides an upper bound on the number of processors that enter phase number . What is the highest phase that a processor can ever enter? The number of processors that can be in phase is at most . So when , then there can be no processor that ever enters phase . Thus no processor can enter any phase beyond phase number , because . Finally, a single processor sends one termination message that travels around the ring once. So for the total number of messages sent by the algorithm we get the

upper bound.

Burns furthermore showed that the asynchronous leader election algorithm is asymptotically optimal: Any uniform algorithm solving the leader election problem in an asynchronous ring must send the number of messages at least proportional to .

Theorem 13.11 Any uniform algorithm for electing a leader in an asynchronous ring sends messages.

The proof, for any algorithm, is based on constructing certain executions of the algorithm on rings of size . Then two rings of size are pasted together in such a way that the constructed executions on the smaller rings are combined, and additional messages are received. This construction strategy yields the desired logarithmic multiplicative overhead.

Exercises

13.3-1 Show that the simplified Bully algorithm has message complexity, by appropriately assigning identifiers to processors on a ring of size , and by determining how to delay processors and messages.

13.3-2 Show that the algorithm Bully has message complexity.

We consider a system in which each processor has a special state component , called the input and , called the output (also called the decision). The variable initially holds a value from some well ordered set of possible inputs and is undefined. Once an assignment to has been made, it is irreversible. Any solution to the consensus problem must guarantee:

Note that in the case of crash failures this validity condition is equivalent to requiring that every nonfaulty decision value is the input of some processor. Once a processor crashes it is of no interest to the algorithm, and no requirements are put on its decision.

We begin by presenting a simple algorithm for consensus in a synchronous message passing system with crash failures.

Since the system is synchronous, an execution of the system consists of a series of rounds. Each round consists of the delivery of all messages, followed by one computation event for every processor. The set of faulty processors can be different in different executions, that is, it is not known in advance. Let be a subset of at most processors, the faulty processors. Each round contains exactly one computation event for the processors not in and at most one computation event for every processor in . Moreover, if a processor in does not have a computation event in some round, it does not have such an event in any further round. In the last round in which a faulty processor has a computation event, an arbitrary subset of its outgoing messages are delivered.

Consensus-with-Crash-Failures

       Code for processor , .       
       Initially 
       round , 
  1  SEND {  has not already sent } to all processors 
  2  RECEIVE  from , ,  
  3   
  4  IF   
  5    THEN  

In the previous algorithm, which is based on an algorithm by Dolev and Strong, each processor maintains a set of the values it knows to exist in the system. Initially, the set contains only its own input. In later rounds the processor updates its set by joining it with the sets received from other processors. It then broadcasts any new additions to the set of all processors. This continues for rounds, where is the maximum number of processors that can fail. At this point, the processor decides on the smallest value in its set of values.

To prove the correctness of this algorithm we first notice that the algorithm requires exactly rounds. This implies termination. Moreover the validity condition is clearly satisfied since the decision value is the input of some processor. It remains to show that the agreement condition holds. We prove the following lemma:

Lemma 13.12 In every execution at the end of round , , for every two nonfaulty processors and .

Proof. We prove the claim by showing that if at the end of round then at the end of round .

Let be the first round in which is added to for any nonfaulty processor . If is initially in let . If then, in round sends to each , causing to add to , if not already present.

Otherwise, suppose and let be a nonfaulty processor that receives for the first time in round . Then there must be a chain of processors that transfers the value to . Hence sends to in round one etc. until sends to in round . But then is a chain of processors. Hence at least one of them, say must be nonfaulty. Hence adds to its set in round , contradicting the minimality of .

This lemma together with the before mentioned observations hence implies the following theorem.

Theorem 13.13 The previous consensus algorithm solves the consensus problem in the presence of crash failures in a message passing system in rounds.

The following theorem was first proved by Fischer and Lynch for Byzantine failures. Dolev and Strong later extended it to crash failures. The theorem shows that the previous algorithm, assuming the given model, is optimal.

Theorem 13.14 There is no algorithm which solves the consensus problem in less than rounds in the presence of crash failures, if .

What if failures are not benign? That is can the consensus problem be solved in the presence of Byzantine failures? And if so, how?

In a computation step of a faulty processor in the Byzantine model, the new state of the processor and the message sent are completely unconstrained. As in the reliable case, every processor takes a computation step in every round and every message sent is delivered in that round. Hence a faulty processor can behave arbitrarily and even maliciously. For example, it could send different messages to different processors. It can even appear that the faulty processors coordinate with each other. A faulty processor can also mimic the behaviour of a crashed processor by failing to send any messages from some point on.

In this case, the definition of the consensus problem is the same as in the message passing model with crash failures. The validity condition in this model, however, is not equivalent with requiring that every nonfaulty decision value is the input of some processor. Like in the crash case, no conditions are put on the output of faulty processors.

Pease, Shostak and Lamport first proved the following theorem.

Theorem 13.15 In a system with processors and Byzantine processors, there is no algorithm which solves the consensus problem if .

The following algorithm uses messages of constant size, takes rounds, and assumes that . It was presented by Berman and Garay.

This consensus algorithm for Byzantine failures contains phases, each taking two rounds. Each processor has a preferred decision for each phase, initially its input value. At the first round of each phase, processors send their preferences to each other. Let be the majority value in the set of values received by processor at the end of the first round of phase . If no majority exists, a default value is used. In the second round of the phase processor , called the king of the phase, sends its majority value to all processors. If receives more than copies of (in the first round of the phase) then it sets its preference for the next phase to be ; otherwise it sets its preference to the phase kings preference, received in the second round of the phase. After phases, the processor decides on its preference. Each processor maintains a local array pref with entries.

We prove correctness using the following lemmas. Termination is immediate. We next note the persistence of agreement:

Lemma 13.16 If all nonfaulty processors prefer at the beginning of phase , then they all prefer at the end of phase , for all , .

Proof. Since all nonfaulty processors prefer at the beginning of phase , they all receive at least copies of (including their own) in the first round of phase . Since , , implying that all nonfaulty processors will prefer at the end of phase .

Consensus-with-Byzantine-failures

       Code for processor , .
       Initially , for any 
       round , 
  1  SEND  to all processors 
  2  RECEIVE  from  and assign to , for all ,  
  3  let maj be the majority value of ( if none) 
  4  let mult be the multiplicity of maj 
       round ,  
  5  IF  
  6    THEN SEND  to all processors 
  7  RECEIVE  from  ( if none) 
  8  IF  
  9    THEN  
 10    ELSE  
 11  IF  
 12    THEN  

This implies the validity condition: If they all start with the same input they will continue to prefer and finally decide on in phase . Agreement is achieved by the king breaking ties. Since each phase has a different king and there are phases, at least one round has a nonfaulty king.

Lemma 13.17 Let be a phase whose king is nonfaulty. Then all nonfaulty processors finish phase with the same preference.

Proof. Suppose all nonfaulty processors use the majority value received from the king for their preference. Since the king is nonfaulty, it sends the same message and hence all the nonfaulty preferences are the same.

Suppose a nonfaulty processor uses its own majority value for its preference. Thus receives more than messages for in the first round of phase . Hence every processor, including receives more than messages for in the first round of phase and sets its majority value to . Hence every nonfaulty processor has for its preference.

Hence at phase all processors have the same preference and by Lemma 13.16 they will decide on the same value at the end of the algorithm. Hence the algorithm has the agreement property and solves consensus.

Theorem 13.18 There exists an algorithm for processors which solves the consensus problem in the presence of Byzantine failures within rounds using constant size messages, if .

As shown before, the consensus problem can be solved in synchronous systems in the presence of both crash (benign) and Byzantine (severe) failures. What about asynchronous systems? Under the assumption that the communication system is completely reliable, and the only possible failures are caused by unreliable processors, it can be shown that if the system is completely asynchronous then there is no consensus algorithm even in the presence of only a single processor failure. The result holds even if the processors only fail by crashing. The impossibility proof relies heavily on the system being asynchronous. This result was first shown in a breakthrough paper by Fischer, Lynch and Paterson. It is one of the most influential results in distributed computing.

The impossibility holds for both shared memory systems if only read/write registers are used, and for message passing systems. The proof first shows it for shared memory systems. The result for message passing systems can then be obtained through simulation.

Theorem 13.19 There is no consensus algorithm for a read/write asynchronous shared memory system that can tolerate even a single crash failure.

And through simulation the following assertion can be shown.

Theorem 13.20 There is no algorithm for solving the consensus problem in an asynchronous message passing system with processors, one of which may fail by crashing.

Note that these results do not mean that consensus can never be solved in asynchronous systems. Rather the results mean that there are no algorithms that guarantee termination, agreement, and validity, in all executions. It is reasonable to assume that agreement and validity are essential, that is, if a consensus algorithm terminates, then agreement and validity are guaranteed. In fact there are efficient and useful algorithms for the consensus problem that are not guaranteed to terminate in all executions. In practice this is often sufficient because the special conditions that cause non-termination may be quite rare. Additionally, since in many real systems one can make some timing assumption, it may not be necessary to provide a solution for asynchronous consensus.

Exercises

13.4-1 Prove the correctness of algorithm Consensus-Crash.

13.4-2 Prove the correctness of the consensus algorithm in the presence of Byzantine failures.

13.4-3 Prove Theorem 13.20.

The design of distributed algorithms is easier when processors have access to (Newtonian) global clock, because then each event that occurs in the distributed system can be labeled with the reading of the clock, processors agree on the ordering of any events, and this consensus can be used by algorithms to make decisions. However, construction of a global clock is difficult. There exist algorithms that approximate the ideal global clock by periodically synchronising drifting local hardware clocks. However, it is possible to totally order events without using hardware clocks. This idea is called the logical clock.

Recall that an execution is an interleaving of instructions of the programs. Each instruction can be either a computational step of a processor, or sending a message, or receiving a message. Any instruction is performed at a distinct point of global time. However, the reading of the global clock is not available to processors. Our goal is to assign values of the logical clock to each instruction, so that these values appear to be readings of the global clock. That is, it possible to postpone or advance the instants when instructions are executed in such a way, that each instruction that has been assigned a value of the logical clock, is executed exactly at the instant of the global clock, and that the resulting execution is a valid one, in the sense that it can actually occur when the algorithm is run with the modified delays.

The Logical-Clock algorithm assigns logical time to each instruction. Each processor has a local variable called counter. This variable is initially zero and it gets incremented every time processor executes an instruction. Specifically, when a processor executes any instruction other than sending or receiving a message, the variable counter gets incremented by one. When a processor sends a message, it increments the variable by one, and attaches the resulting value to the message. When a processor receives a message, then the processor retrieves the value attached to the message, then calculates the maximum of the value and the current value of counter, increments the maximum by one, and assigns the result to the counter variable. Note that every time instruction is executed, the value of counter is incremented by at least one, and so it grows as processor keeps on executing instructions. The value of logical time assigned to instruction is defined as the pair , where counter is the value of the variable counter right after the instruction has been executed, and id is the identifier of the processor. The values of logical time form a total order, where pairs are compared lexicographically. This logical time is also called Lamport time. We define to be a quotient , which is an equivalent way to represent the pair.

Remark 13.21 For any execution, logical time satisfies three conditions:

(i) if an instruction is performed by a processor before an instruction is performed by the same processor, then the logical time of is strictly smaller than that of ,

(ii) any two distinct instructions of any two processors get assigned different logical times,

(iii) if instruction sends a message and instruction receives this message, then the logical time of is strictly smaller than that of .

Our goal now is to argue that logical clock provides to processors the illusion of global clock. Intuitively, the reason why such an illusion can be created is that we can take any execution of a deterministic algorithm, compute the logical time of each instruction , and run the execution again delaying or speeding up processors and messages in such a way that each instruction is executed at the instant of the global clock. Thus, without access to a hardware clock or other external measurements not captured in our model, the processors cannot distinguish the reading of logical clock from the reading of a real global clock. Formally, the reason why the re-timed sequence is a valid execution that is indistinguishable from the original execution, is summarised in the subsequent corollary that follows directly from Remark 13.21.

Corollary 13.22 For any execution , let be the assignment of logical time to instructions, and let be the sequence of instructions ordered by their logical time in . Then for each processor, the subsequence of instructions executed by the processor in is the same as the subsequence in . Moreover, each message is received in after it is sent in .

In a system execution, an instruction can affect another instruction by altering the state of the computation in which the second instruction executes. We say that one instruction can causally affect (or influence) another, if the information that one instruction produces can be passed on to the other instruction. Recall that in our model of distributed system, each instruction is executed at a distinct instant of global time, but processors do not have access to the reading of the global clock. Let us illustrate causality. If two instructions are executed by the same processor, then we could say that the instruction executed earlier can causally affect the instruction executed later, because it is possible that the result of executing the former instruction was used when the later instruction was executed. We stress the word possible, because in fact the later instruction may not use any information produced by the former. However, when defining causality, we simplify the problem of capturing how processors influence other processors, and focus on what is possible. If two instructions and are executed by two different processors, then we could say that instruction can causally affect instruction , when the processor that executes sends a message when or after executing , and the message is delivered before or during the execution of at the other processor. It may also be the case that influence is passed on through intermediate processors or multiple instructions executed by processors, before reaching the second processor.

We will formally define the intuition that one instruction can causally affect another in terms of a relation called happens before, and that relates pairs of instructions. The relation is defined for a given execution, i.e., we fix a sequence of instructions executed by the algorithm and instances of global clock when the instructions were executed, and define which pairs of instructions are related by the happens before relation. The relation is introduced in two steps. If instructions and are executed by the same processor, then we say that happens before if and only if is executed before . When and are executed by two different processors, then we say that happens before if and only if there is a chain of instructions and messages

for , such that is either equal to or is executed after by the same processor that executes ; is either equal to or is executed before by the same processor that executes ; is executed before by the same processor, ; and sends a message that is received by , . Note that no instruction happens before itself. We write when happens before . We omit the reference to the execution for which the relation is defined, because it will be clear from the context which execution we mean. We say that two instructions and are concurrent when neither nor . The question stands how processors can determine if one instruction happens before another in a given execution according to our definition. This question can be answered through a generalisation of the Logical-Clock algorithm presented earlier. This generalisation is called vector clocks.

The Vector-Clocks algorithm allows processors to relate instructions, and this relation is exactly the happens before relation. Each processor maintains a vector of integers. The -th coordinate of the vector is denoted by . The vector is initialised to the zero vector . A vector is modified each time processor executes an instruction, in a way similar to the way counter was modified in the Logical-Clock algorithm. Specifically, when a processor executes any instruction other than sending or receiving a message, the coordinate gets incremented by one, and other coordinates remain intact. When a processor sends a message, it increments by one, and attaches the resulting vector to the message. When a processor receives a message, then the processor retrieves the vector attached to the message, calculates coordinate-wise maximum of the current vector and the vector , except for coordinate that gets incremented by one, and assigns the result to the variable .

        
       FOR ALL 
          

We label each instruction executed by processor with the value of the vector right after the instruction has been executed. The label is denoted by and is called vector timestamp of instruction . Intuitively, represents the knowledge of processor about how many instructions each processor has executed at the moment when has executed instruction . This knowledge may be obsolete.

Vector timestamps can be used to order instructions that have been executed. Specifically, given two instructions and , and their vector timestamps and , we write that when the vector is majorised by the vector i.e., for all , the coordinate is at most the corresponding coordinate . We write when but .

The next theorem explains that the Vector-Clocks algorithm indeed implements the happens before relation, because we can decide if two instructions happen or not before each other, just by comparing the vector timestamps of the instructions.

Theorem 13.23 For any execution and any two instructions and , if and only if .

Proof. We first show the forward implication. Suppose that . Hence and are two different instructions. If the two instructions are executed on the same processor, then must be executed before . Only finite number of instructions have been executed by the time has been executed. The Vector-Clock algorithm increases a coordinate by one as it calculates vector timestamps of instructions from until inclusive, and no coordinate is ever decreased. Thus . If and were executed on different processors, then by the definition of happens before relation, there must be a finite chain of instructions and messages leading from to . But then by the Vector-Clock algorithm, the value of a coordinate of vector timestamp gets increased at each move, as we move along the chain, and so again .

Now we show the reverse implication. Suppose that it is not the case that . We consider a few subcases always concluding that it is not that case that . First, it could be the case that and are the same instruction. But then obviously vector clocks assigned to and are the same, and so it cannot be the case that . Let us, therefore, assume that and are different instructions. If they are executed by the same processor, then cannot be executed before , and so is executed after . Thus, by monotonicity of vector timestamps, , and so it is not the case that . The final subcase is when and are executed by two distinct processors and . Let us focus on the component of vector clock of processor right after was executed. Let its value be . Recall that other processors can only increase the value of their components by adopting the value sent by other processors. Hence, in order for the value of component of processor to be or more at the moment is executed, there must be a chain of instructions and messages that passes a value at least , originating at processor . This chain starts at or at an instruction executed by subsequent to . But the existence of such chain would imply that happens before , which we assumed was not the case. So the component of vector clock is strictly smaller than the component of vector clock . Thus it cannot be the case that .

This theorem tells us that we can decide if two distinct instructions and are concurrent, by checking that it is not the case that nor is it the case that .

The happens before relation can be used to compute a global state of distributed system, such that this state is in some sense consistent. Shortly, we will formally define the notion of consistency. Each processor executes instructions. A cut is defined as a vector of non-negative integers. Intuitively, the vector denotes the states of processors. Formally, denotes the number of instructions that processor has executed. Not all cuts correspond to collections of states of distributed processors that could be considered natural or consistent. For example, if a processor has received a message from and we record the state of in the cut by making appropriately large, but make so small that the cut contains the state of the sender before the moment when the message was sent, then we could say that such cut is not natural—there are instructions recorded in the cut that are causally affected by instructions that are not recorded in the cut. Such cuts we consider not consistent and so undesirable. Formally, a cut is inconsistent when there are processors and such that the instruction number of processor is causally affected by an instruction subsequent to instruction number of processor . So in an inconsistent cut there is a message that “crosses” the cut in a backward direction. Any cut that is not inconsistent is called a consistent cut.

The Consistent-Cut algorithm uses vector timestamps to find a consistent cut. We assume that each processor is given the same cut as an input. Then processors must determine a consistent cut that is majorised by . Each processor has an infinite table of vectors. Processor executes instructions, and stores vector timestamps in consecutive entries of the table. Specifically, entry of the table is the vector timestamp of the -th instruction executed by the processor; we define to be the zero vector. Processor begins calculating a cut right after the moment when the processor has executed instruction number . The processor determines the largest number that is at most , such that the vector is majorised by . The vector that processors collectively find turns out to be a consistent cut.

Theorem 13.24 For any cut , the cut computed by the Consistent-Cut algorithm is a consistent cut majorised by .

Proof. First observe that there is no need to consider entries of further than . Each of these entries is not majorised by , because the -th coordinate of any of these vectors is strictly larger than . So we can indeed focus on searching among the first entries of . Let be the largest entry such that the vector is majorised by the vector . We know that such vector exists, because is a zero vector, and such vector is majorised by any cut .

We argue that is a consistent cut by way of contradiction. Suppose that the vector is an inconsistent cut. Then, by definition, there are processors and such that there is an instruction of processor subsequent to instruction number , such that happens before instruction number of processor . Recall that is the furthest entry of majorised by . So entry is not majorised by , and since all subsequent entries, including the one for instruction , can have only larger coordinates, the entries are not majorised by either. But, happens before instruction number , so entry can only have larger coordinates than respective coordinates of the entry corresponding to , and so cannot be majorised by either. This contradicts the assumption that is majorised by . Therefore, must be a consistent cut.

There is a trivial algorithm for finding a consistent cut. The algorithm picks . However, the Consistent-Cut algorithm is better in the sense that the consistent cut found is maximal. That this is indeed true, is left as an exercise.

There is an alternative way to find a consistent cut. The Consistent Cut algorithm requires that we attach vector timestamps to messages and remember vector timestamps for all instructions executed so far by the algorithm which consistent cut we want to compute. This may be too costly. The algorithm called Distributed-Snapshot avoids this cost. In the algorithm, a processor initiates the calculation of consistent cut by flooding the network with a special message that acts like a sword that cuts the execution of algorithm consistently. In order to prove that the cut is indeed consistent, we require that messages are received by the recipient in the order they were sent by the sender. Such ordering can be implemented using sequence number.

In the Distributed-Snapshot algorithm, each processor has a variable called counter that counts the number of instructions of algorithm executed by the processor so far. In addition the processor has a variable that will store the -th coordinate of the cut. This variable is initialised to . Since the variables counter only count the instructions of algorithm , the instructions of Distributed-Snapshot algorithm do not affect the counter variables. In some sense the snapshot algorithm runs in the “background”. Suppose that there is exactly one processor that can decide to take a snapshot of the distributed system. Upon deciding, the processor “floods” the network with a special message <Snapshot>. Specifically, the processor sends the message to all its neighbours and assigns counter to . Whenever a processor receives the message and the variable is still , then the processor sends <Snapshot> message to all its neighbours and assigns current to . The sending of <Snapshot> messages and assignment are done by the processor without executing any instruction of (we can think of Distributed-Snapshot algorithm as an “interrupt”). The algorithm calculates a consistent cut.

Theorem 13.25 Let for any processors and , the messages sent from to be received in the order they are sent. The Distributed-Snapshot algorithm eventually finds a consistent cut . The algorithm sends messages, where is the number of edges in the graph.

Proof. The fact that each variable is eventually different from follows from our model, because we assumed that instructions are eventually executed and messages are eventually received, so the <Snapshot> messages will eventually reach all nodes.

Suppose that is not a consistent cut. Then there is a processor such that instruction number or later sends a message < > other than <Snapshot>, and the message is received on or before a processor executes instruction number . So the message < > must have been sent after the message <Snapshot> was sent from to . But messages are received in the order they are sent, so processes <Snapshot> before it processes < >. But then message < > arrives after snapshot was taken at . This is a desired contradiction.

Exercises

13.5-1 Show that logical time preserves the happens before () relation. That is, show that if for events and it is the case that , then , where is the logical time of an event.

13.5-2 Show that any vector clock that captures concurrency between processors must have at least coordinates.

13.5-3 Show that the vector calculated by the algorithm Consistent-Cut is in fact a maximal consistent cut majorised by . That is that there is no that majorises and is different from , such that is majorised by .

In the broadcast problem, a selected processor , called a source or a sender, has the message , which must be delivered to all processors in the system (including the source). The interface of the broadcast service is specified as follows:

In above definitions qos denotes the quality of service provided by the system. We consider two kinds of quality service:

The basic model of a message-passing distributed system normally does not guarantee any ordering or reliability of messaging operations. In the basic model we only assume that each pair of processors is connected by a link, and message delivery is independent on each link — the order of received messages may not be related to the order of the sent messages, and messages may be lost in the case of crashes of senders or receivers.

We present some of the most useful requirements for ordering and reliability of broadcast services. The main question we address is how to implement a stronger service on top of the weaker service, starting with the basic system model.

We now describe implementations of algorithms for various broadcast services.

       Code for any processor , 
<01>INITIALISATION</01>
<02>    for every </02>
        
 11  IF  occurs 
 12    THEN  
 13       ENABLE  
        
 21  IF  occurs 
 22    THEN ADD triple  to pending 
 23        
 24    IF  
 25       THEN  
 26          ENABLE  
        
 31  IF  occurs 
 32    THEN  
        
 41  IF 
 42     is the pending triple with the smallest  and 
           for every 
 43  THEN ENABLE  
 44    REMOVE triple  from pending 

       Code for any processor , 
<01>INITIALISATION</01>
<02>    for every </02>
<03>   pending list is empty</03>
        
 11  IF  occurs 
 12    THEN  
 13       ENABLE  
        
 21  IF  occurs 
 22    THEN ADD triple  to pending 
        
 31  IF  is the pending triple, and 
 32    , and 
 33     for every  
 34    THEN ENABLE  
 35       REMOVE triple  from pending 
 36        

Multicast services are similar to the broadcast services, except each multicast message is destined for a specified subset of all processors.In the multicast service we provide two types of events, where qos denotes a quality of service required:

Note that the event mc-recv is similar to bc-recv.

As in case of a broadcast service, we would like to provide useful ordering and reliable properties of the multicast services. We can adapt ordering requirements from the broadcast services. Basic Multicast does not require any ordering properties. Single-Source FIFO requires that if one processor multicasts messages (possibly to different destination sets), then the messages received in each processors (if any) must be received in the same order as sent by the source. Definition of Causal Order remains the same. Instead of Total Order, which is difficult to achieve since destination sets may be different, we define another ordering property:

The reliability conditions for multicast are somewhat different from the conditions for reliable broadcast.

One way of implementing ordered and reliable multicast services is to use the corresponding broadcast services (for Sub-Total Order the corresponding broadcast requirement is Total Order). More precisely, if event occurs processor enables event . When an event occurs, processor enables event if , otherwise it ignores this event. The proof that such method provides required multicast quality of service is left as an exercise.

The classic problem of collecting rumors, or gossip, is defined as follows:

At the beginning, each processor has its distinct piece of information, called a rumor, the goal is to make every processor know all the rumors.

However in the model with processor crashes we need to re-define the gossip problem to respect crash failures of processors. Both Integrity and No-Duplicates properties are the same as in the reliable broadcast service, the only difference (which follows from the specification of the gossip problem) is in Liveness requirements:

The efficiency of gossip algorithms is measured in terms of time and message complexity. Time complexity measures number of (synchronous) steps from the beginning to the termination. Message complexity measures the total number of point-to-point messages sent (more precisely, if a processor sends a message to three other processors in one synchronous step, it contributes three to the message complexity).

The following simple algorithm completes gossip in just one synchronous step: each processor broadcasts its rumor to all processors. The algorithm is correct, because each message received contains a rumor, and a message not received means the failure of its sender. A drawback of such a solution is that a quadratic number of messages could be sent, which is quite inefficient.

We would like to perform gossip not only quickly, but also with fewer point-to-point messages. There is a natural trade-off between time and communication. Note that in the system without processor crashes such a trade-off may be achieved, e.g., sending messages over the (almost) complete binary tree, and then time complexity is , while the message complexity is . Hence by slightly increasing time complexity we may achieve almost linear improvement in message complexity. However, if the underlying communication network is prone to failures of components, then irregular failure patterns disturb a flow of information and make gossiping last longer. The question we address in this section is what is the best trade-off between time and message complexity in the model with processor crashes?

In this part we describe the family of gossip algorithms, among which we can find some efficient ones. They are all based on the same generic code, and their efficiency depends on the quality of two data structures put in the generic algorithm. Our goal is to prove that we may find some of those data structures that obtained algorithm is always correct, and efficient if the number of crashes in the execution is at most , where is a parameter.

We start with description of these structures: communication graph and communication schedules.

Generic algorithm.

We start with specifying a goal that gossiping algorithms need to achieve. We say that processor has heard about processor if either knows the original input rumor of or knows that has already failed. We may reformulate correctness of a gossiping algorithm in terms of hearing about other processors: algorithm is correct if Integrity and No-Duplicates properties are satisfied and if each processor has hard about any other processor by the termination of the algorithm.

The code of a gossiping algorithm includes objects that depend on the number of processors in the system, and also on the bound on the number of failures which are “efficiently tolerated” (if the number of failures is at most then message complexity of design algorithm is small). The additional parameter is a termination threshold which influences time complexity of the specific implementation of the generic gossip scheme. Our goal is to construct the generic gossip algorithm which is correct for any additional parameters and any communication graph and set of schedules, while efficient for some values and structures and .

Each processor starts gossiping as a collector. Collectors seek actively information about rumors of the other processors, by sending direct inquiries to some of them. A collector becomes a disseminator after it has heard about all the processors. Processors with this status disseminate their knowledge by sending local views to selected other processors.

Local views. Each processor starts with knowing only its ID and its input information . To store incoming data, processor maintains the following arrays:

, and ,

each of size . All these arrays are initialised to store the value nil. For an array of processor , we denote its th entry by - intuitively this entry contains some information about processor . The array Rumor is used to store all the rumors that a processor knows. At the start, processor sets to its own input . Each time processor learns some , it immediately sets to this value. The array Active is used to store a set of all the processors that the owner of the array knows as crashed. Once processor learns that some processor has failed, it immediately sets to failed. Notice that processor has heard about processor , if one among the values and is not equal to NIL.

The purpose of using the array Pending is to facilitate dissemination. Each time processor learns that some other processor is fully informed, that is, it is either a disseminator itself or has been notified by a disseminator, then it marks this information in . Processor uses the array to send dissemination messages in a systematic way, by scanning to find those processors that possibly still have not heard about some processor.

The following is a useful terminology about the current contents of the arrays Active and Pending. Processor is said to be active according to , if has not yet received any information implying that crashed, which is the same as having nil in . Processor is said to need to be notified by if it is active according to and is equal to nil.

Phases. An execution of a gossiping algorithm starts with the processors initialising all the local objects. Processor initialises its list with nil at all the locations, except for the th one, which is set equal to . The remaining part of execution is structured as a loop, in which phases are iterated. Each phase consists of three parts: receiving messages, local computation, and multicasting messages. Phases are of two kinds: regular phase and ending phase. During regular phases processor: receives messages, updates local knowledge, checks its status, sends its knowledge to neighbours in communication graphs as well as inquiries about rumors and replies about its own rumor. During ending phases processor: receives messages, sends inquiries to all processors from which it has not heard yet, and replies about its own rumor. The regular phases are performed times; the number is a termination threshold. After this, the ending phase is performed four times. This defines a generic gossiping algorithm.

Generic-Gossip

       Code for any processor , 
<01>INITIALISATION</01>
<02>   processor  becomes a collector</02>
<03>   initialisation of arrays ,  and </03>
        
 11  REPEAT  times 
 12    PERFORM regular phase 
        
 20  REPEAT  times 
 21    PERFORM ending phase 

Now we describe communication and kinds of messages used in regular and ending phases.

Graph and range messages used during regular phases. A processor may send a message to its neighbour in the graph , provided that it is is still active according to . Such a message is called a graph one. Sending these messages only is not sufficient to complete gossiping, because the communication graph may become disconnected as a result of node crashes. Hence other messages are also sent, to cover all the processors in a systematic way. In this kind of communication processor considers the processors as ordered by its local permutation , that is, in the order . Some of additional messages sent in this process are called range ones.

During regular phase processors send the following kind of range messages: inquiring, reply and notifying messages. A collector sends an inquiring message to the first processor about which has not heard yet. Each recipient of such a message sends back a range message that is called a reply one.

Disseminators send range messages also to subsets of processors. Such messages are called notifying ones. The target processor selected by disseminator is the first one that still needs to be notified by . Notifying messages need not to be replied to: a sender already knows the rumors of all the processors, that are active according to it, and the purpose of the message is to disseminate this knowledge.

Regular-Phase

       Code for any processor , 
<01>RECEIVE messages</01>
        
 11  PERFORM local computation 
 12    UPDATE the local arrays 
 13    IF  is a collector, that has already heard about all the processors 
 14       THEN  becomes a disseminator 
 15    COMPUTE set of destination processors: FOR each processor  
 16       IF  is active according to  and  is a neighbour of  in graph  
 17          THEN add  to destination set for a graph message 
 18       IF     is a collector and  is the first processor 
                   about which  has not heard yet
 19          THEN send an inquiring message to  
 20       IF  is a disseminator and  is the first processor 
                   that needs to be notified by 
 21          THEN send a notifying message to  
 22       IF  is a collector, from which an inquiring message was received 
                   in the receiving step of this phase
 23          THEN send a reply message to  
        
 30  SEND graph/inquiring/notifying/reply messages to corresponding destination sets 

Last-resort messages used during ending phases. Messages sent during the ending phases are called last-resort ones. These messages are categorised into inquiring, replying, and notifying, similarly as the corresponding range ones, which is because they serve a similar purpose. Collectors that have not heard about some processors yet send direct inquiries to all of these processors simultaneously. Such messages are called inquiring ones. They are replied to by the non-faulty recipients in the next step, by way of sending reply messages. This phase converts all the collectors into disseminators. In the next phase, each disseminator sends a message to all the processors that need to be notified by it. Such messages are called notifying ones.

The number of graph messages, sent by a processor at a step of the regular phase, is at most as large as the maximum node degree in the communication graph. The number of range messages, sent by a processor in a step of the regular phase, is at most as large as the number of inquiries received plus a constant - hence the global number of point-to-point range messages sent by all processors during regular phases may be accounted as a constant times the number of inquiries sent (which is one per processor per phase). In contrast to that, there is no a priori upper bound on the number of messages sent during the ending phase. By choosing the termination threshold to be large enough, one may control how many rumors still needs to be collected during the ending phases.

Updating local view. A message sent by a processor carries its current local knowledge. More precisely, a message sent by processor brings the following: the ID , the arrays , , and , and a label to notify the recipient about the character of the message. A label is selected from the following: graph_message, inquiry_from_collector, notification_from_disseminator, this_is_a_reply, their meaning is self-explanatory. A processor scans a newly received message from some processor to learn about rumors, failures, and the current status of other processors. It copies each rumor from the received copy of into , unless it is already there. It sets to failed, if this value is at . It sets to done, if this value is at . It sets to done, if is a disseminator and the received message is a range one. If is itself a disseminator, then it sets to done immediately after sending a range message to . If a processor expects a message to come from processor , for instance a graph one from a neighbour in the communication graph, or a reply one, and the message does not arrive, then knows that processor has failed, and it immediately sets to failed.

Ending-Phase

       Code for any processor , 
<01>RECEIVE messages</01>
        
 11  PERFORM local computation 
 12    UPDATE the local arrays 
 13    IF  is a collector, that has already heard about all the processors 
 14       THEN  becomes a disseminator 
 15    COMPUTE set of destination processors: FOR each processor  
 16       IF  is a collector and it has not heard about  yet 
 17          THEN send an inquiring message to  
 18       IF  is a disseminator and  needs to be notified by  
 19          THEN send a notifying message to  
        
 20       IF an inquiring message was received from  
                in the receiving step of this phase
 21          THEN send a reply message to  
        
 30  SEND inquiring/notifying/reply messages to corresponding destination sets 

Correctness. Ending phases guarantee correctness, as is stated in the next fact.

Lemma 13.29 Generic-Gossip is correct for every communication graph and set of schedules .

Proof. Integrity and No-Duplicates properties follow directly from the code and the multicast service in synchronous message-passing system. It remains to prove that each processor has heard about all processors. Consider the step just before the first ending phases. If a processor has not heard about some other processor yet, then it sends a last-resort message to in the first ending phase. It is replied to in the second ending phase, unless processor has crashed already. In any case, in the third ending phase, processor either learns the input rumor of or it gets to know that has failed. The fourth ending phase provides an opportunity to receive notifying messages, by all the processors that such messages were sent to by .

The choice of communication graph , set of schedules and termination threshold influences however time and message complexities of the specific implementation of Generic Gossip Algorithm. First consider the case when is a communication graph satisfying property from Definition 13.27, contains random permutations, and for sufficiently large positive constant . Using Theorem 13.28 we get the following result.

Theorem 13.30 For every and , for some constant , there is a graph such that the implementation of the generic gossip scheme with as a communication graph and a set of random permutations completes gossip in expected time and with expected message complexity , if the number of crashes is at most .

Consider a small modification of Generic Gossip scheme: during regular phase every processor sends an inquiring message to the first (instead of one) processors according to permutation , where is a maximum degree of used communication graph . Note that it does not influence the asymptotic message complexity, since besides inquiring messages in every regular phase each processor sends graph messages.

Theorem 13.31 For every there are parameters and and there is a graph such that the implementation of the modified Generic Gossip scheme with as a communication graph and a set of random permutations completes gossip in expected time and with expected message complexity , for any number of crashes.

Since in the above theorem set is selected prior the computation, we obtain the following existential deterministic result.

Theorem 13.32 For every there are parameters and and there are graph and set of schedules such that the implementation of the modified Generic Gossip scheme with as a communication graph and schedules completes gossip in time and with message complexity , for any number of crashes.

Exercises

13.7-1 Design executions showing that there is no relation between Causal Order and Total Order and between Single-Source FIFO and Total Order broadcast services. For simplicity consider two processors and two messages sent.

13.7-2 Does broadcast service satisfying Single-Source FIFO and Causal Order requirements satisfy a Total Order property? Does broadcast service satisfying Single-Source FIFO and Total Order requirements satisfy a Causal Order property? If yes provide a proof, if not show a counterexample.

13.7-3 Show that using reliable Basic Broadcast instead of Basic Broadcast in the implementation of Single-Source FIFO service, then we obtain reliable Single-Source FIFO broadcast.

13.7-4 Prove that the Ordered Broadcast algorithm implements Causal Order service on a top of Single-Source FIFO one.

13.7-5 What is the total number of point-to-point messages sent in the algorithm Ordered-Broadcast in case of broadcasts?

13.7-6 Estimate the total number of point-to-point messages sent during the execution of Reliable-Causally-Ordered-Broadcast, if it performs broadcast and there are processor crashes during the execution.

13.7-7 Show an execution of the algorithm Reliable-Causally-Ordered-Broadcast which violates Total Order requirement.

13.7-8 Write a code of the implementation of reliable Sub-Total Order multicast service.

13.7-9 Show that the described method of implementing multicast services on the top of corresponding broadcast services is correct.

13.7-10 Show that the random graph - in which each node selects independently at random edges from itself to other processors - satisfies property from Definition 13.27 and has degree with probability at least .

13.7-11 Leader election problem is as follows: all non-faulty processors must elect one non-faulty processor in the same synchronous step. Show that leader election can not be solved faster than gossip problem in synchronous message-passing system with processors crashes.

The shared memory is modeled in terms of a collection of shared variables, commonly referred to as registers. We assume the system contains processors, , and registers . Each processor is modeled as a state machine. Each register has a type, which specifies:

Each register can have an initial value.

For example, an integer valued read/write register can take on all integer values and has operations read(R,v) and write(R,v). The read operation returns the value of the last preceding write, leaving unchanged. The write(R,v) operation has an integer parameter , returns no value and changes 's value to . A configuration is a vector , where is a state of and is a value of register . The events are computation steps at the processors where the following happens atomically (indivisibly):

A finite sequence of configurations and events that begins with an initial configuration is called an execution. In the asynchronous shared memory system, an infinite execution is admissible if it has an infinite number of computation steps.

In this problem a group of processors need to access a shared resource that cannot be used simultaneously by more than a single processor. The solution needs to have the following two properties. (1) Mutual exclusion: Each processor needs to execute a code segment called a critical section so that at any given time at most one processor is executing it (i.e., is in the critical section). (2) Deadlock freedom: If one or more processors attempt to enter the critical section, then one of them eventually succeeds as long as no processor stays in the critical section forever. These two properties do not provide any individual guarantees to any processor. A stronger property is (3) No lockout: A processor that wishes to enter the critical section eventually succeeds as long as no processor stays in the critical section forever. Original solutions to this problem relied on special synchronisation support such as semaphores and monitors. We will present some of the distributed solutions using only ordinary shared variables.

We assume the program of a processor is partitioned into the following sections:

A processor cycles through these sections in the order: remainder, entry, critical and exit. A processor that wants to enter the critical section first executes the entry section. After that, if successful, it enters the critical section. The processor releases the critical section by executing the exit section and returning to the remainder section. We assume that a processor may transition any number of times from the remainder to the entry section. Moreover, variables, both shared and local, accessed in the entry and exit section are not accessed in the critical and remainder section. Finally, no processor stays in the critical section forever. An algorithm for a shared memory system solves the mutual exclusion problem with no deadlock (or no lockout) if the following hold:

In the context of mutual exclusion, an execution is admissible if for every processor , either takes an infinite number of steps or ends in the remainder section. Moreover, no processor is ever stuck in the exit section (unobstructed exit condition).

A single bit suffices to guarantee mutual exclusion with no deadlock if a powerful test&set register is used. A test&set variable is a binary variable which supports two atomic operations, test&set and reset, defined as follows:

       test&set(: memory address) returns binary value:
          
          
          return ()
       reset(: memory address):
          

The test&set operation atomically reads and updates the variable. The reset operation is merely a write. There is a simple mutual exclusion algorithm with no deadlock, which uses one test&set register.

Mutual exclusion using one test&set register

       Initially  equals 
        
        :
  1  wait until  
        
        :
  2   
        

Assume that the initial value of is . In the entry section, processor repeatedly tests until it returns . The last such test will assign to , causing any following test by other processors to return , prohibiting any other processor from entering the critical section. In the exit section resets to ; another processor waiting in the entry section can now enter the critical section.

Theorem 13.33 The algorithm using one test&set register provides mutual exclusion without deadlock.

If a powerful primitive such as test&set is not available, then mutual exclusion must be implemented using only read/write operations.

       Code for processor , .
       Initially  and
        FALSE, for 
        
        :
  1  TRUE 
  2   
  3  FALSE 
  4  FOR  TO  DO 
  5    WAIT UNTIL FALSE 
  6    WAIT UNTIL  or   
        :
  7   
        

       procedure Node(: integer; side: )
  1   
  2  WAIT UNTIL ( or ) 
  3   
  4  IF  
  5    THEN IF () 
  6          THEN goto line 1 
  7          ELSE WAIT UNTIL  
  8  IF  
  9    THEN  
 10    ELSE Node() 
 11     
 12     
       end procedure

In all mutual exclusion algorithms presented so far, the number of steps taken by processors before entering the critical section depends on , the number of processors even in the absence of contention (where multiple processors attempt to concurrently enter the critical section), when a single processor is the only processor in the entry section. In most real systems however, the expected contention is usually much smaller than .

A mutual exclusion algorithm is said to be fast if a processor enters the critical section within a constant number of steps when it is the only processor trying to enter the critical section. Note that a fast algorithm requires the use of multi-writer, multi-reader shared variables. If only single writer variables are used, a processor would have to read at least variables.

Such a fast mutual exclusion algorithm is presented by Lamport.

Fast-Mutual-Exclusion

       Code for processor , . Initially Fast-Lock and Slow-Lock are , and          is false for all , 
       
        :
  1  TRUE 
  2   
  3  IF  
  4    THEN FALSE 
  5    WAIT UNTIL  
  6    goto 1 
  7   
  8  IF  
  9    THEN FALSE 
 10       for all , WAIT UNTIL FALSE 
 11       IF  
 12          THEN WAIT UNTIL  
 13       goto 1 
        
        :
 14   
 15  FALSE 
        

Lamport's algorithm is based on the correct combination of two mechanisms, one for allowing fast entry when no contention is detected, and the other for providing deadlock freedom in the case of contention. Two variables, Fast-Lock and Slow-Lock are used for controlling access when there is no contention. In addition, each processor has a boolean variable whose value is true if is interested in entering the critical section and false otherwise. A processor can enter the critical section by either finding - in this case it enters the critical section on the fast path - or by finding in which case it enters the critical section along the slow path.

Consider the case where no processor is in the critical section or in the entry section. In this case, Slow-Lock is and all Want entries are . Once now enters the entry section, it sets to and Fast-Lock to . Then it checks Slow-Lock which is . then it checks Fast-Lock again and since no other processor is in the entry section it reads and enters the critical section along the fast path with three writes and two reads.

If then waits until all Want flags are reset. After some processor executes the for loop in line , the value of Slow-Lock remains unchanged until some processor leaving the critical section resets it. Hence at most one processor may find and this processor enters the critical section along the slow path. Note that the Lamport's Fast Mutual Exclusion algorithm does not guarantee lockout freedom.

Theorem 13.39 Algorithm Fast-Mutual-Exclusion guarantees mutual exclusion without deadlock.

Exercises

13.8-1 An algorithm solves the 2-mutual exclusion problem if at any time at most two processors are in the critical section. Present an algorithm for solving the 2-mutual exclusion problem using test & set registers.

13.8-2 Prove that bakery algorithm satisfies the mutual exclusion property.

13.8-3 Prove that bakery algorithm provides no lockout.

13.8-4 Isolate a bounded mutual exclusion algorithm with no lockout for two processors from the tournament tree algorithm. Show that your algorithm has the mutual exclusion property. Show that it has the no lockout property.

13.8-5 Prove that algorithm Fast-Mutual-Exclusion has the mutual exclusion property.

13.8-6 Prove that algorithm Fast-Mutual-Exclusion has the no deadlock property.

13.8-7 Show that algorithm Fast-Mutual-Exclusion does not satisfy the no lockout property, i.e. construct an execution in which a processor is locked out of the critical section.

13.8-8 Construct an execution of algorithm Fast-Mutual-Exclusion in which two processors are in the entry section and both read at least variables before entering the critical section.

13-1 Number of messages of the algorithm Flood

Prove that the algorithm Flood sends messages in any execution, given a graph with vertices and edges. What is the exact number of messages as a function of the number of vertices and edges in the graph?

13-2 Leader election in a ring

Assume that messages can only be sent in CW direction, and design an asynchronous algorithm for leader election on a ring that has message complexity.

Hint. Let processors work in phases. Each processor begins in the active mode with a value equal to the identifier of the processor, and under certain conditions can enter the relay mode, where it just relays messages. An active processor waits for messages from two active processors, and then inspects the values sent by the processors, and decides whether to become the leader, remain active and adopt one of the values, or start relaying. Determine how the decisions should be made so as to ensure that if there are three or more active processors, then at least one will remain active; and no matter what values active processors have in a phase, at most half of them will still be active in the next phase.

13-3 Validity condition in asynchronous systems

Show that the validity condition is equivalent to requiring that every nonfaulty processor decision be the input of some processor.

13-4 Single source consensus

An alternative version of the consensus problem requires that the input value of one distinguished processor (the general) be distributed to all the other processors (the lieutenants). This problem is also called single source consensus problem. The conditions that need to be satisfied are:

So if the general is faulty, then the nonfaulty processors need not decide on the general's input, but they must still agree with each other. Consider the synchronous message passing system with Byzantine faults. Show how to transform a solution to the consensus problem (in Subsection 13.4.5) into a solution to the general's problem and vice versa. What are the message and round overheads of your transformation?

13-5 Bank transactions

Imagine that there are banks that are interconnected. Each bank starts with an amount of money . Banks do not remember the initial amount of money. Banks keep on transferring money among themselves by sending messages of type <10> that represent the value of a transfer. At some point of time a bank decides to find the total amount of money in the system. Design an algorithm for calculating that does not stop monetary transactions.

The definition of the distributed systems presented in the chapter are derived from the book by Attiya and Welch [24]. The model of distributed computation, for message passing systems without failures, was proposed by Attiya, Dwork, Lynch and Stockmeyer [23].

Modeling the processors in the distributed systems in terms of automata follows the paper of Lynch and Fisher [229].

The concept of the execution sequences is based on the papers of Fischer, Gries, Lamport and Owicki [229], [261], [262].

The definition of the asynchronous systems reflects the presentation in the papers of Awerbuch [25], and Peterson and Fischer [270].

The algorithm Spanning-Tree-Broadcast is presented after the paper due to Segall [297].

The leader election algorithm Bully was proposed by Hector Garcia-Molina in 1982 [127]. The asymptotic optimality of this algorithm was proved by Burns [51].

The two generals problem is presented as in the book of Gray [144].

The consensus problem was first studied by Lamport, Pease, and Shostak [214], [268]. They proved that the Byzantine consensus problem is unsolvable if [268].

One of the basic results in the theory of asynchronous systems is that the consensus problem is not solvable even if we have reliable communication systems, and one single faulty processor which fails by crashing. This result was first shown in a breakthrough paper by Fischer, Lynch and Paterson [108].

The algorithm Consensus-with-Crash-Failures is based on the paper of Dolev and Strong [90].

Berman and Garay [40] proposed an algorithm for the solution of the Byzantine consensus problem for the case . Their algorithm needs rounds.

The bakery algorithm [212] for mutual exclusion using only shared read/write registers to solve mutual exclusion is due to Lamport [212]. This algorithm requires arbitrary large values. This requirement is removed by Peterson and Fischer [270]. After this Burns and Lynch proved that any deadlock-free mutual exclusion algorithm using only shared read/write registers must use at least shared variables, regardless of their size [52].

The algorithm Fast-Mutual-Exclusion is presented by Lamport [213]. The source of the problems 13-3, 13-4, 13-5 is the book of Attiya and Welch [24].

Important textbooks on distributed algorithms include the monumental volume by Nancy Lynch [228] published in 1997, the book published by Gerard Tel [320] in 2000, and the book by Attiya and Welch [24]. Also of interest is the monograph by Claudia Leopold [221] published in 2001, and the book by Nicola Santoro [296], which appeared in 2006.

A recent book on the distributed systems is due to A. D. Kshemkalyani and M. [206].

Finally, several important open problems in distributed computing can be found in a recent paper of Aspnes et al. [21].

The section summaries the main features of the widely used discrete event simulation tools, OPNET and COMNET, and the supporting network analysers, Network Associates' Sniffer and OPNET's Application Characterisation Environment.

OPtimized Network Engineering Tools (OPNET) is a comprehensive simulation system capable of modelling communication networks and distributed systems with detailed protocol modelling and performance analysis. OPNET consists of a number of tools that fall into three categories corresponding to the three main phases of modelling and simulation projects: model specification, data collection and simulation, and analysis.

During model specification the network modeller develops a representation of the network system under study. OPNET implements the concept of model reuse, i.e., models are based on embedded models developed earlier and stored in model libraries. The model is specified at various levels of details using specification editors. These editors categorise the required modelling information corresponding to the hierarchical structure of an actual network system. The highest level editor, the Project Editor develops network models consisting of network topology, subnets, links, and node models specified in the Node Editor. The Node Editor describes nodes' internal architecture, functional elements and data flow between them. Node models in turn, consist of modules with process models specified by the Process Editor. The lowest level of the network hierarchy, the process models, describes the module's behaviour in terms of protocols, algorithms, and applications using finite state machines and a high-level language.

There are several other editors to define various data models referenced by process- or node-level models, e.g., packet formats and control information between processes. Additional editors create, edit, and view probability density functions (PDFs) to control certain events, such as the interarrival time of sending or receiving packets, etc. The model-specification editors provide a graphical interface for the user to manipulate objects representing the models and the corresponding processes. Each editor can specify objects and operations corresponding to the model's abstraction level. Therefore, the Project Editor specifies nodes and link objects of a network, the Node Editor specifies processors, queues, transmitters, and receivers in the network nodes, and the Process Editor specifies the states and transitions in the processes. Figure 14.12 depicts the abstraction level of each editor:

Figure 14.12.  The three modelling abstraction levels specified by the Project, Node, and Process editors.

OPNET can produce many types of output during simulation depending on how the modeller defined the types of output. In most cases, modellers use the built in types of data: output vectors, output scalars, and animation:

Typically, much of the data collected during simulations is stored in output scalar and output vector files. In order to analyse these data OPNET provides the Analysis Tool which is a collection of graphing and numerical processing functions. The Analysis Tool presents data in the form of graphs or traces. Each trace consists of a list of abscissa and ordinate pairs. Traces are held and displayed in analysis panels. The Analysis Tool supports a variety of methods for processing simulation output data and computing new traces. Calculations, such as histograms, PDF, CDF, and confidence intervals are included. Analysis Tool also supports the use of mathematical filters to process vector or trace data. Mathematical filters are defined as hierarchical block diagrams based on a predefined set of calculus, statistical, and arithmetic operators. The example diagrams below (Figures 14.13 and 14.14) shows graphs generated by the Analysis Tool:

Figure 14.13.  Example for graphical representation of scalar data (upper graph) and vector data (lower graph).

Figure 14.14.  Figure 14.14 shows four graphs represented by the Analysis Tool.

Figure 14.14 Analysis Tool Showing Four Graphs.

COMNET is another popular discrete-event simulation system. We will discuss it briefly and demonstrate its features in Section 14.9.

There is an increasing interest in predicting, measuring, modelling, and diagnosing application performance across the application lifecycle from development through deployment to production. Characterising the application's performance is extremely important in critical application areas, like in eCommerce. In the increasingly competitive eCommerce, the application's performance is critical, especially where the competition is just “one click” away. Application performance affects revenue. When an application performs poorly it is always the network that is blamed rather than the application. These performance problems may result from several areas including application design or slow database servers. Using tools, like ACE and Network Associates' Sniffer, network modellers can develop methodologies to identify the source of application slowdowns and resolve their causes. After analysing the applications, modellers can make recommendations for performance optimisation. The result is faster applications and better response times. The Application Characterisation Environment (ACE) is a tool for visualising, analysing, and troubleshooting network applications. Network managers and application developers can use ACE to

The performance of an application is determined by network attributes that are affected by the various components of a communication network. The following list contains some example for these attributes and the related network elements:

Analysis of an application requires two phases:

After creating the application task, we can perform the following operations over the captured traffic traces:

Without going into specific details we illustrate some of the features above through a simple three-tier application. We want to determine the reason or reasons of the slow response time from a Client that remotely accesses an Application Server (App Server) to retrieve information from a Database Server (DB Server). The connection is over an ADSL line between the client and the Internet, and a 100Mbps Ethernet connection between the App Server and the DB Server. We want to identify the cause of the slow response time and recommend solutions. We deployed capture agents at the network segments between the client and the App Server and between the servers. The agents captured traffic traces simultaneously during a transaction between the client and the App Server and the App Server and the DB Server respectively. Then, the traces were merged and synchronised to obtain the best possible analysis of delays at each tier and in the network.

After importing the trace into ACE, we can analyse the transaction in the Data Exchange Chart, which depicts the flow of application messages among tiers over time.

Figure 14.15.  Data Exchange Chart.

The Data Exchange Chart shows packets of various sizes being transmitted between the Client and the servers. The overall transaction response time is approximately 6 seconds. When the “Show Dependencies” checkbox is checked, the white dependency lines indicate large processing delays on the Application Server and Client tiers. For further analysis, we generate the “Summary of Delays” window showing how the total response time of the application is divided into four general categories: Application delay, Propagation delay, Transmission delay and Protocol/Congestion delay. Based on this chart we can see the relation between application and network related delays during the transaction between the client and the servers. The chart clearly shows that the application delay far outweighs the Propagation, Transmission, and Protocol/Congestion delays slowing down the transaction.

Figure 14.16.  Summary of Delays.

The “Diagnosis” function (Figure 14.17) provides a more granular analysis of possible bottlenecks by analysing factors that often cause performance problems in networked applications. Values over a specified threshold are marked as bottlenecks or potential bottlenecks.

Figure 14.17.  Diagnosis window.

The diagnosis of the transaction confirms that the primary bottleneck is due to Processing Delay on the Application Server. The processing delay is due to the file I/O, CPU processing, or memory access. It also reveals another bottleneck: the chattiness of the application that leads us to the next step. We investigate the application behaviour in terms of application turns that can be obtained from the transaction statistics. An application turn is a change in direction of the application-message flow.

The statistics of the transaction (Figure 14.18) disclose that the number of application turns is high, i.e., the data sent by the transaction at a time is small. This may cause significant application and network delays. Additionally, a significant portion of application processing time can be spent processing the many requests and responses. The Diagnosis window indicates a “Chattiness” bottleneck without a “Network Cost of Chattiness” bottleneck, which means the following:

Figure 14.18.  Statistics window.

The recommendation is that the application should send fewer, larger application messages. This will utilise network and tier resources more efficiently. For example, a database application should avoid sending a set of records one record at a time.

Would the response time decrease significantly if we added more bandwidth to the link between the client and the APP Server (Figure 14.19)? Answering this question is important because adding more bandwidth is expensive. Using the prediction feature we can answer the question. In the following chart we selected the bandwidth from 128K to 10Mbps. The chart shows that beyond approximately 827 Kbps there is no significant improvement in response time, i.e., for this application the recommended highest bandwidth is no more than 827Kbps, which can be provided by a higher speed DSL line.

Figure 14.19.  Impact of adding more bandwidth on the response time.

After the analysis of the application's performance, we can immediately create the starting baseline model from the captured traffic traces for further simulation studies as illustrated in Figure 14.20.

Figure 14.20.  Baseline model for further simulation studies.

Another popular network analyser is Network Associates' Sniffer. (Network Associates has recently renamed it to Netasyst.) It is a powerful network visualisation tool consisting of a set of functions to:

For further details we refer the reader to the vendors' documentations on http://www.nai.com.

Each transaction between a client and a server consists of active periods followed by inactive periods. Transactions consist of groups of packets sent in each direction. Each group of packets is called a burst. The burstiness of the traffic can be characterised by the following time parameters:

The M/Pareto traffic model and the Hurst parameter.

Recent results have proven that the M/Pareto model is appropriate for modelling long-range dependent traffic flow characterised by long bursts. Originally, the model was introduced and applied in the analysis of ATM buffer levels. The M/Pareto model was also used to predict the queueing performance of Ethernet, VBR video, and IP packet streams in a single server queue. We apply the M/Pareto model not just for a single queue, but also for predicting the performance of an interconnected system of links, switches and routers affecting the individual network elements' performance.

The M/Pareto model is a Poisson process of overlapping bursts with arrival rate . A burst generates packets with arrival rate . Each burst, from the time of its interval, will continue for a Pareto-distributed time period. The use of Pareto distribution results in generating extremely long bursts that characterise long-range dependent traffic.

The probability that a Pareto-distributed random variable exceeds threshold is:

The mean of , the mean duration of a burst and its variance is infinite. Assuming a time interval, the mean number of packets in the time interval is:

where

The M/Pareto model is asymptotically self-similar and it is shown that for the Hurst parameter the following equation holds:

We implemented the Hurst parameter and a modified version of the M/Pareto model in the discrete event simulation system COMNET. By using discrete event simulation methodology, we can get realistic results in measuring network parameters, such as utilisation of links and the queueing performance of switches and routers. Our method can model and measure the harmful consequences of aggregated bursty traffic and predict its impact on the overall network's performance.

Traffic measurements.

In order to build the baseline model, we collected traffic traces in a large corporate network by the Concord Network Health network analyser system. We took measurements from various broadband and narrow band links including 45Mbps ATM, 56Kbps, and 128 Kbps frame relay connections. The Concord Network Health system can measure the traffic in certain time intervals at network nodes, such as routers and switches. We set the time intervals to 6000 seconds and measured the number of bytes and packets sent and received per second, packet latency, dropped packets, discard eligible packets, etc. Concord Network Health cannot measure the number of packets in a burst and the duration of the bursts as it is assumed in the M/Pareto model above. Due to this limitation of our measuring tool, we slightly modify our traffic model according to the data available. We took snapshots of the traffic in every five minutes from a narrow band frame relay connection between a remote client workstation and a server at the corporate headquarters as traffic destination in the following format:

Figure 14.23.  Traffic traces.

The mean number of bytes, the message delay from the client to server, the input buffer level at the client's local router, the number of blocked packets, the mean utilisations of the 56Kbps frame relay, the DS-3 segment of the ATM network, and the 100Mbps Ethernet link at the destination are summarised in Figure 14.24.

Figure 14.24.  Measured network parameters.

COMNET represents a transaction by a message source, a destination, the size of the message, communication devices, and links along the path. The rate at which messages are sent is specified by an interarrival time distribution, the time between two consecutive packets. The Poisson distribution in the M/Pareto model generates bursts or messages with arrival rate , the number of arrivals, which are likely to occur in a certain time interval. In simulation, this information is expressed by the time interval between successive arrivals . For this purpose, we use the Exponential distribution. Using the Exponential distribution for interarrival time will result in an arrival pattern characterised by the Poisson distribution. In COMNET, we implemented the interarrival time with the function Exp(). The interarrival time in the model is set to one second matching the sampling time interval set in Concord Network Health and corresponding to an arrival rate /sec.

In the M/Pareto model, each burst continues for a Pareto-distributed time period. The Concord Network Health cannot measure the duration of a burst; hence, we assume that a burst is characterised by the number of bytes in a message sent or received in a second. Since the ATM cell rate algorithm ensures that equal length messages are processed in equal time, then longer messages require longer processing time. So we can say that the distribution of the duration of bursts is the same as the distribution of the length of bursts. Hence, we can modify the M/Pareto model by substituting the Pareto-distributed duration of bursts with the Pareto-distributed length of bursts. We derive of the Pareto distribution not from the mean duration of bursts, but from the mean length of bursts.

The Pareto distributed length of bursts is defined in COMNET by two parameters- the location and the shape. The location parameter corresponds to the , the shape parameter corresponds to the parameter of the M/Pareto model in (1) and can be calculated from the relation (4) as

The Pareto distribution can have infinite mean and variance. If the shape parameter is greater than 2, both the mean and variance are finite. If the shape parameter is greater than 1, but less than or equal to 2, the mean is finite, but then the variance is infinite. If the shape parameter is less than or equal to 1, both the mean and variance are infinite.

From the mean of the Pareto distribution we get:

The relations (5) and (6) allow us to model bursty traffic based on real traffic traces by performing the following steps:

• a. Collect traffic traces using the Concord Network Health network analyser.

• b. Compute the Hurst parameter by making use of the Benoit package with the traffic trace as input.

• c. Use the Exponential and Pareto distributions in the COMNET modelling tool with the parameters calculated above to specify the distribution of the interarrival time and length of messages.

• d. Generate traffic according to the modified M/Pareto model and measure network performance parameters.

The traffic generated according to the steps above is bursty with parameter H calculated from real network traffic.

We validate our baseline model by comparing various model parameters of a 56Kbps frame relay and a 6Mbps ATM connection with the same parameters of a real network as the Concord Network Health network analyser traced it. For simplicity, we use only the “Bytes Total/sec” column of the trace, i.e., the total number of bytes in the “Bytes Total/sec” column is sent in one direction only from the client to the server. The Hurst parameter of the real traffic trace is calculated by the Benoit package. The topology is as follows:

Figure 14.25.  Part of the real network topology where the measurements were taken.

The “Message sources” icon is a subnetwork that represents a site with a token ring network, a local router, and a client sending messages to the server in the “Destination” subnetwork:

Figure 14.26.  “Message Source” remote client.

The interarrival time and the length of messages are defined by the Exponential and Pareto functions Exp (1) and Par (208.42, 1.9) respectively. The Pareto distribution's location (208.42) and shape (1.9) are calculated from formulas (5) and (6) by substituting the mean length of bursts (440 bytes from Table 2.) and .

Figure 14.27.  Interarrival time and length of messages sent by the remote client.

The corresponding heavy-tailed Pareto probability distribution and cumulative distribution functions are illustrated in Figure 14.28 (The represents the number of bytes):

Figure 14.28.  The Pareto probability distribution for mean 440 bytes and Hurst parameter .

The “Frame Relay” icon represents a frame relay cloud with 56K committed information rate (CIR). The “Conc” router connects the frame relay network to a 6Mbps ATM network with variable rate control (VBR) as shown in Figures 14.29 and 14.30:

Figure 14.29.  The internal links of the 6Mbps ATM network with variable rate control (VBR).

Figure 14.30.  Parameters of the 6Mbps ATM connection.

The “Destination” icon denotes a subnetwork with server :

Figure 14.31.  The “Destination” subnetwork.

The results of the model show almost identical average for the utilisation of the frame relay link () and the utilisation of the real measurements (3.1%):

Figure 14.32.  utilisation of the frame relay link in the baseline model.

The message delay in the model is also very close to the measured delay between the client and the server (78 msec):

Figure 14.33.  Baseline message delay between the remote client and the server.

The input buffer level of the remote client's router in the model is almost identical with the measured buffer level of the corresponding router:

Figure 14.34.  Input buffer level of remote router.

Similarly, the utilisations of the model's DS-3 link segment of the ATM network and the Ethernet link in the destination network closely match with the measurements of the real network:

Figure 14.35.  Baseline utilisations of the DS-3 link and Ethernet link in the destination.

It can also be shown from the model's traffic trace that for the model generated messages the Hurst parameter , i.e., the model generates almost the same bursty traffic as the real network. Furthermore, the number of dropped packets in the model was zero similarly to the number of dropped packets in the real measurements. Therefore, we start from a model that closely represents the real network.

In order to illustrate our method, we developed a COMNET simulation model to measure the consequences of bursty traffic on network links, message delays, routers' input buffers, and the number of dropped packets due to the aggregated traffic of large number of users. The model implements the Hurst parameter as it has been described in Section 3. We repeated the simulation for 6000 sec, 16000 sec and 18000 sec to allow infrequent events to occur a reasonable number of times. We found that the results are very similar in each simulation.

Topology of bursty traffic sources.

The “Message Source” subnetworks transmit messages as in the baseline model above, but with different burstiness: , and with fixed size. Initially, we simulate four subnetworks and four users per subnetwork each sending the same volume of data (mean 440 bytes per second) as in the validating model above:

Figure 14.36.  Network topology of bursty traffic sources with various Hurst parameters.

Link utilisation and message delay.

First, we are going to measure and illustrate the extremely high peaks in frame relay link utilisation and message delay. The model traffic is generated with message sizes determined by various Hurst parameters and fixed size messages for comparison. The COMNET modelling tool has a trace option to capture its own model generated traffic. It has been verified that for the model-generated traffic flows with various Hurst parameters the Benoit package computed similar Hurst parameters for the captured traces.

The following table shows the simulated average and peak link utilisation of the different cases. The utilisation is expressed in the [0, 1] scale not in percentages:

Figure 14.37.  Simulated average and peak link utilisation.

The enclosed charts in Appendix A clearly demonstrate that even though the average link utilisation is almost identical, the frequency and the size of the peaks increase with the burstiness, causing cell drops in routers and switches. We received the following results for response time measurements:

Figure 14.38.  Response time and burstiness.

The charts in the Appendix A graphically illustrate the relation between response times and various Hurst parameters.

Input buffer level for large number of users.

We also measured the number of cells dropped at a router's input buffer in the ATM network due to surge of bursty cells. We simulated the aggregated traffic of approximately 600 users each sending the same number of bytes in a second as in the measured real network. The number of blocked packets is summarised in the following table:

Figure 14.39.  Relation between the number of cells dropped and burstiness.

Theis chapter presented a discrete event simulation methodology to measure various network performance parameters while transmitting bursty traffic. It has been proved in recent studies that combining bursty data streams will also produce bursty combined data flow. The studies imply that the methods and models used in traditional network design require modifications. We categorise our modelling methodology as a structural model contrary to a black box model. Structural models focus on the environment in which the models' data was collected; i.e., the complex hierarchies of network components that make up today's communications systems. Although black box models are useful in other contexts, they are not easy to use in designing, managing and controlling today's networks. We implemented a well-known model, the M/Pareto model within the discrete event simulation package COMNET that allows the analysis of the negative impact of self-similar traffic on not just one single queue, but on the overall performance of various interrelated network components as well. Using real network traces, we built and validated a model by which we could measure and graphically illustrate the impact of bursty traffic on link utilisation, message delays, and buffer performance of Frame Relay and ATM networks. We illustrated that increasing burstiness results in extremely high link utilisation, response time, and dropped packets, and measured the various performance parameters by simulation.

The choice of the package emphasises the need for integrated tools that could be useful not just for theoreticians, but also for network engineers and designers. Our paper intends to narrow the gap between existing, well-known theoretical results and their applicability in everyday, practical network analysis and modelling. It is highly desirable that appropriate traffic models should be accessible from measuring, monitoring, and controlling tools. Our model can help network designers and engineers, the ultimate users of traffic modelling, understand the dynamic nature of network traffic and assist them in their everyday practice.