4005-730 Distributed Systems
Lecture Notes -- Module 4. Distributed Algorithms

Theory of Distributed Algorithms

• Assumption 1: The processes in the distributed system communicate using messages
  • "Messages" in the abstract sense
  • These may be TCP messages, UDP datagrams, remote method invocations, etc.

• Assumption 2: The distributed system topology is known
  • Each process knows the processes to which it will send messages
  • Each process knows the processes from which it will receive messages
  • Service discovery, if any, has already happened

• Assumption 3A: Message passing is synchronous
  • There is a known upper bound on the time required to send a message
    • Including the computation time needed to generate the message
    • And the communication time needed to transfer the message across the network
  • This lets the processes organize the message passing into rounds
  • If a process has not received a message by the end of a round, the process can conclude the message will never be received

  - OR -

• Assumption 3B: Message passing is asynchronous
  • There is no upper bound on the time required to send a message
  • Message passing cannot be organized into rounds
  • Designing distributed algorithms for asynchronous message passing is much harder than for synchronous message passing

Logical Clocks

• Sometimes a distributed system needs to know about the ordering of events throughout the system
  • For example, in a distributed (replicated) database system, database updates must be done in the same order in all replicas

• One way to impose an order on events is timestamps
  • Each process attaches its current clock value to each event the process generates
  • All processes handle events in timestamp order
  • However, we can't assume the clocks are identical in all processes
  • Real-time clock synchronization protocols exist, but are complicated and only guarantee agreement to some nonzero tolerance

• Another way to impose an order on events is a logical clock
  • Invented by Leslie Lamport in 1978

• There are N processes p₁, p₂, . . . p_N

• Each process experiences a sequence of events e_i
  • Send a message: e_i = send(m)
  • Receive a message: e_i = receive(m)

• There is an event causality relation: e_i → e_j , "e_i caused e_j"
  • If e_i and e_j happen in the same process and e_i happens first, then e_i → e_j
  • If e_i = send(m) and e_j = receive(m) (send and receive of the same message), then e_i → e_j
  • The causality relation is transitive: If e_i → e_j and e_j → e_k , then e_i → e_k
  • The causality relation defines an ordering (partial ordering) among events
  • If neither e_i → e_j nor e_j → e_i is true, we say that e_i and e_j are concurrent events

• Each process has a logical clock LC, an integer

• The logical clock is used to associate a logical timestamp with each event in the process: LC(e_i)

• We want to use the events' logical timestamps to reflect the order of the events; specifically:
  • If e_i → e_j , then LC(e_i) < LC(e_j)

• Here's how to do it. In each process:
  • Initialize LC = 0
  • To send a message, e_i = send(m):
    • Increment logical clock: LC = LC + 1
    • Use that as the logical timestamp of the send event: LC(e_i) = LC
    • Also use that as the logical timestamp of the message: TS(m) = LC
    • Send m and TS(m) to recipient
  • When a message is received, e_i = receive(m):
    • Fast-forward logical clock to message timestamp if necessary: LC = max (LC, TS(m))
    • Increment logical clock: LC = LC + 1
    • Use that as the logical timestamp of the receive event: LC(e_i) = LC

• Example

  
  Sape Mullender, editor, Distributed Systems, 2nd Edition (ACM Press, 1993), page 67.

• Totally ordered broadcast (or multicast)
  • Example: Updates in a replicated distributed database
  • Processes are simultaneously sending messages
  • Messages are to be delivered in the same order in all processes
  • Assumes no failures
  • Assumes FIFO communication channels
    • Messages sent from p_i to p_j are received at p_j in the same order as p_i sent them

• Implementing totally ordered broadcast with a logical clock
  • Each process maintains a queue of received but not-yet-delivered messages
  • Queue is maintained in ascending order of message timestamps TS(m)
    • Break ties in the message timestamps using sending process number
  • When a process has an outgoing message, it broadcasts the message
    • "Broadcast" means a copy of the message gets sent to all processes, including the sending process itself
    • Each copy of the message has the same message timestamp
  • When a process receives an incoming message, it puts the message in the queue, then it broadcasts an acknowledgment
  • An incoming message is delivered when:
    • The message is at the head of the queue
    • Acknowledgments for the message have been received from all processes
    • The process removes the message from the queue, then delivers the message

Vector Clocks

• Logical clocks cannot be used to infer causality
  • It is true that if e_i → e_j , then LC(e_i) < LC(e_j)
  • It is not true that if LC(e_i) < LC(e_j), then e_i → e_j

• A vector clock is an extension of a logical clock that can be used to infer causality

• Each process has a vector clock VC[1..N], a vector of integers, indexed by process number

• The vector clock is used to associate a vector timestamp with each event in the process: VC(e_i)[1..N]

• We want to use the events' vector timestamps to infer the causality of the events; specifically:
  • e_i → e_j  if and only if VC(e_i) < VC(e_j)

• The "less than" relation between vector clocks is defined as:
  • VC(e_i) < VC(e_j) if and only if
  • VC(e_i)[k] <= VC(e_j)[k] for all k = 1, 2, . . . N; and
  • VC(e_i)[k] != VC(e_j)[k] for some k = 1, 2, . . . N

• Here's how to do it. In each process:
  • Initialize VC[k] = 0 for all k = 1, 2, . . . N
  • To send a message from process k, e_i = send(m):
    • Increment vector clock element for current process: VC[k] = VC[k] + 1
    • Use that as the vector timestamp of the send event: VC(e_i) = VC
    • Also use that as the vector timestamp of the message: TS(m) = VC
    • Send m and TS(m) to recipient
  • When a message is received in process k, e_i = receive(m):
    • Fast-forward vector clock to message timestamp if necessary:
      VC[k] = max (VC[k], TS(m)[k]) for all k = 1, 2, . . . N
    • Increment vector clock element for current process: VC[k] = VC[k] + 1
    • Use that as the vector timestamp of the receive event: VC(e_i) = VC

• The elements of an event's vector timestamp VC(e_i) in process k have these meanings:
  • VC(e_i)[j] = The number of events in process j that causally precede event e_i in process k, for j != k
  • VC(e_i)[k] = The number of events that have occurred in process k up to and including event e_i

• Example

  
  Sape Mullender, editor, Distributed Systems, 2nd Edition (ACM Press, 1993), page 72.

• Causally ordered broadcast (or multicast)
  • Example: A newsgroup
  • Processes are simultaneously sending messages
  • If message A causally precedes message B, then A must be delivered before B in all processes
  • Assumes no failures
  • Assumes FIFO communication channels
    • Messages sent from p_i to p_j are received at p_j in the same order as p_i sent them

• Implementing causally ordered broadcast with a vector clock
  • Each process maintains a list of received but not-yet-delivered messages, including their vector timestamps
  • Each process maintains a vector of counters D[1..N], initially all 0
  • When a process has an outgoing message, it broadcasts the message
    • "Broadcast" means a copy of the message gets sent to all processes, including the sending process itself
    • Each copy of the message has the same vector timestamp
  • When a process receives an incoming message, it puts the message in the list
  • An incoming message m received from process j is delivered when:
    • m is the next message expected from process j:
      TS(m)[j] = D[j] + 1; and
    • All messages that causally precede m have already been delivered:
      TS(m)[k] ≤ D[k] for all k ≠ j
  • When m is delivered, remove it from the list and set D[j] = TS(m)[j]

Global Snapshot Algorithms

• Global state of a distributed system includes:
  • Local state of each process p_i
    • Internal state of each process's computation
  • Channel state of each channel c_ij
    • The messages that have been sent from p_i to p_j that have not yet been received by p_j

• A global snapshot is the global state of the system at a certain instant

• Only an omniscient observer outside the system can get a true look at the global state

• We would like an algorithm that runs within the system itself and that captures a global snapshot in the following sense:
  • Consider the true global state S₁ when the algorithm starts and the true global state S₂ when the algorithm finishes
  • There are typically many possible sequences of states the system could traverse going from S₁ to S₂
  • The global snapshot algorithm will capture one of the states from among all these possible sequences of states
  • This is not necessarily a state that the system did go through (as seen by the omniscient external observer)
  • But it is a state the system could have gone through during the course of the algorithm

• Chandy and Lamport invented such a global snapshot algorithm in 1985

• There is a special process p₀
  • p₀ initiates the snapshot
  • p₀ collects information from the other processes p₁ through p_N to build the snapshot

• The processes send special "take snapshot" messages to each other
  • These are different from the actual application messages

• Each process has a set of incoming channels
  • Each channel comes from a process that is sending messages to this process in the distributed application

• Each process has a set of outgoing channels
  • Each channel goes to a process that is receiving messages from this process in the distributed application

• During the global snapshot algorithm, each process j keeps a list c_ij of messages that process j receives from process i

• The global snapshot algorithm
  • p₀ sends a "take snapshot" message to every process p₀ through p_N
  • The first time p_j receives a "take snapshot" message, p_j does the following:
    • Records its own local state
    • Sets c_ij to an empty list for all incoming channels i
    • Sends a "take snapshot" message on all outgoing channels
  • Whenever p_j receives a regular application message after the first "take snapshot" message, p_j does the following:
    • Records the application message in the list c_ij for the incoming channel i on which the application message arrived
  • Whenever p_j receives another "take snapshot" message after the first "take snapshot" message, p_j does the following:
    • Stops recording application messages for the incoming channel i on which the "take snapshot" message arrived
  • When p_j has received a "take snapshot" message from all incoming channels, p_j does the following:
    • Sends its local state and all message lists c_ij to p₀

• Example

  
  Sape Mullender, editor, Distributed Systems, 2nd Edition (ACM Press, 1993), page 81.

• Applications of the global snapshot algorithm
  • Distributed termination detection
  • Deadlock detection

Leader Election Algorithms

• Assumptions
  • Every process has a unique identifier (process number)
  • Every process knows the identifiers of all the other processes
  • Processes may be up or down
  • One process is the leader
    • The leader carries out some special role in the distributed system
  • If the leader goes down, the other processes must elect one of their number to be the new leader
  • Synchronous message passing

• The bully algorithm, invented by Garcia-Molina in 1982
  • When a process notices the leader is down, the process sends an "election" message to all higher-numbered processes
  • When a process receives an "election" message from a lower-numbered process, the higher-numbered process sends an "OK" message back to the lower-numbered process, then sends an "election" message to all higher-numbered processes
  • When a process receives an "OK" message, the process does nothing further and waits for a "leader" message
  • If a process receives no response to an "election" message, that process becomes the new leader
  • The new leader sends a "leader" message to all the processes, informing them who the leader (coordinator) is
  • The highest-numbered process always becomes the leader, like a bully
  • When a process starts up, it initiates an election, and becomes the leader (or not) if it is the highest-numbered process (or not)

• Example

  
  Andrew Tanenbaum and Maarten van Steen, Distributed Systems: Principles and Paradigms (Prentice Hall, 2002), page 264.

• A ring algorithm
  • The processes are arranged in a ring in ascending order of process number
  • When a process notices the leader is down, the process sends an "election" message to the next process in the ring
    • The "election" message contains a list of all process numbers that have seen the "election" message so far
    • Initially, the list contains the process number that initiated the election
  • The process waits to receive an acknowledgment from the recipient
  • If there is no acknowledgment, the process sends the "election" message to the second next process in the ring, and so on until it gets an acknowledgment
  • When a process receives an "election" message, the process adds itself to the list in the message, and passes the message on
  • Eventually the process that started the election receives an "election" message that has gone all the way around the ring
    • The list in the message contains all the processes that are still up
    • The highest-numbered process will be the leader
    • The process changes the message to a "leader" message and passes it on around the ring
  • When a process receives a "leader" message, the process records who the new leader is and passes the message on
  • Eventually the process that started the election receives a "leader" message that has gone all the way around the ring
    • The process does not forward the message, and the election is over

• Example

  
  Andrew Tanenbaum and Maarten van Steen, Distributed Systems: Principles and Paradigms (Prentice Hall, 2002), page 265.

An Impossibility Result

• E. Akkoyunlu, K. Ekanadham, and R. Huber. Some constraints and tradeoffs in the design of network communications. In 5th ACM Symposium on Operating Systems Principles, pages 67-74, 1975.

• The Two Armies Problem
  • Two hills with a valley between
  • One half of the Red Army, 5000 soldiers, is on the west hill
  • The other half of the Red Army, 5000 soldiers, is on the east hill
  • The entire Blue Army, 8000 soldiers, is in the valley
  • If both halves of the Red Army attack the Blue Army, the Red Army will win
  • If only one half of the Red Army attacks the Blue Army, the Blue Army will win
  • The two Red Army generals need to coordinate their attacks to guarantee victory
  • But the only way they can coordinate is to send messengers across the valley, through the thick of the Blue Army
  • There is no guarantee that any messenger will be successful -- the Blue Army might capture the messenger
  • Can the two Red Army generals agree to attack simultaneously?

• An example of a consensus problem
  • All processes need to make a decision between two possible outcomes, usually represented as 0 and 1
  • Processes or messages can fail arbitrarily
  • The non-failed processes must eventually all decide 0 or all decide 1

• Another example: Distributed database transactions
  • Either there are no failures during the transaction, so all processes apply the database updates . . .
  • Or there are failures during the transaction, so none of the processes apply the database updates

• It is impossible to reach consensus assuming arbitrary failures
  • Red General A sends a message to Red General B, "Attack!"
  • B receives the message, but B knows that A will not attack unless A knows that B received the original message (because the original message might have failed), so B sends a message to A, "Attack!"
  • A receives the message, but A knows that B will not attack unless B knows that A knows that B received the original message, so A sends a message to B, "Attack!"
  • B receives the message, but B knows that A will not attack unless A knows that B knows that A knows that B received the original message, so B sends a message to A, "Attack!"
  • This can go on forever without A or B ever reaching a decision
  • Thus, it is impossible to reach consensus assuming arbitrary failures

• Solutions to the consensus problem are possible, but they must make stricter assumptions than arbitrary failures

Distributed Algorithms vs. Client-Server

• The need for distributed algorithms mostly vanishes in a client-server architecture
  • Logical clocks, vector clocks: The server processes all events in the desired order
  • Global snapshots: The server holds all the state of the system
  • Leader election: The server is always the leader
  • Consensus: There is no other process with which the server needs to reach agreement

• This is one reason why client-server systems are not truly distributed systems

Distributed Systems 4005-730-01 Spring Quarter 2013

Course Page

Alan Kaminsky • Department of Computer Science • Rochester Institute of Technology • 4486 + 2220 = 6706

Home Page

Copyright © 2013 Alan Kaminsky. All rights reserved. Last updated 28-Mar-2013. Please send comments to ark­@­cs.rit.edu.