Distributed Systems

Books

Distributed Computing -- H. Attiya and J. Welch -- McGraw-Hill, 1998, ch. 1
Concurrent Systems -- J. Bacon -- Addison-Wesley 1994, ch. 1
Distributed Systems -- Coulouris et al. -- A-W 1993, ch. 1
Distributed Systems -- Mullender (ed.) -- A-W 1993, ch. 1 and 2
Distributed Algorithms -- N. Lynch -- Morgan Kaufman 1996, ch. 1

Introduction

A distributed system is a system composed of several computers that are connected in some way and that together provide some service or achieve some goal, e.g.

communication, e.g. cellphones, email
process control, e.g. in aircraft
information processing, e.g. financial transactions
scientific computing, e.g. weather forecasting

The hardware of a distributed system may for instance be a VLSI chip, a shared-memory multiprocessor machine, a LAN of workstations or PCs, or a global network.

In this course we study some ideas, concepts and techniques that are useful in understading, designing and implementing distributed systems.

1 A Mathematical Model

We begin by describing a mathematical model that can be used to express many distributed systems.

Consider a finite connected graph (P, L). We call the nodes processes and the edges links. A link makes possible communication of data from a set D between the processes that it joins.

Example: Suppose the graph has 12 processes P₀,... ,P₁₁ connected in a ring, i.e. P = {P₀,... ,P₁₁} and L = {(P_i, P_i+1) | 0 £ i < 12} where arithmetic is mod 12. Let D = N È {done}.

[Diagram goes here]

The set M of messages is P × D × P.

Example: The message (P₃, done, P₄) is the datum done set from P₃ to P₄.

The state of a process at a given moment is determined by the control state it is in, the data it is storing, and the arrived messages that have been delivered to it but not examined by it.

Example: Suppose each P_i has an associated name n_i Î N. Let the set S of states be Q × N × F(M) where Q = {init, active, leader, follower} and F(M) is the set of all finite multisets¹ of messages. A state of P₂ might be (active, 20, {(P₁, 0, P₂), (P₁, 10, P₂)}).

A configuration of a distributed system describes the state of each of its processes and the collection of messages that have been sent but not delivered.

Example: The set C of configurations is S¹² × F(M).

Each process has a transition function that determines what, if anything, it may do in a given state. Transition functions are sometimes described explicitly, as in the next example, and sometimes implicitly, by some text, pseudo-code, or program.

Example: The transition function of P_i, for 0 £ i < 12, is d_i : S S × F(M) defined by:

d_i(init, n, in) = ((active, n, in), {(P_i, n, P_i+1)})
d_i(q, n, in) =	((q, n, Ø), Ø)
	if q=leader or q=follower or (q=active and in=Ø)
d_i(active, n, in) =
if done Î data(in) then ((follower, n, Ø), {(P_i, done, P_i+1)})
else if n Î data(in) then ((leader, n, Ø),{(P_i, done, P_i+1)})
else if n > max(in) then ((active, n, Ø), Ø)
else if n < max(in) then ((active, n, Ø, {(P_i, max(in), P_i+1)})
where	data(in) is the set of data in the message set in,
and	max(in) is the largest integer in that set (and in ¹ Ø)

Very rough sketch of d_i:

                        /~\
                        | | 
                        | v 
              init -> active -> leader <-\
                         |            \__/
                         v
                      follower <-\            [mmmm, ASCII]
                              \__/

There are two kinds of event:

Computation events, where a process makes a transition
Delivery events, where a message is delived to a process

If an event E is possible in a configuration C, it may occur and result in a configuration C': C ¾®^e C'.

Example

Suppose C = (S₀,...,S₁₁,U) and d_i(S_i) = (S_i', out). Then C ¾®^comp(i) C' = (S₀, ..., S_i', ..., S₁₁, U È out).
Suppose C = (S₀, ..., S₁₁, U È {m}) where m = (P_j, d, P_i). Then C ¾®^del(m) C' = (S₀, ..., S_i, ..., S₁₁, U) where if S_i = (q, n, in) then S_i' = (q, n, in È {m}).

Q has a nonempty subset Q_init of initial control states, and a subset Q_final of final control states. A state is initial (final) if its control state is. A configuration is initial (final) if the state of each of its processes is.

Example: Q_init = {init} and Q_final = {leader, follower}.

Also, S_init = Q_init × N × {Ø} and S_final = Q_final × N × F(M).

Also, C_init = (S_init)¹² × F(M) and C_final = (S_final)¹² × F(M).

In general, if d is a transition function, S is a final state, and d(S) = (S', out), then S' is final. Hence, if C is final and C ¾®^e C', then C' is final.

An evolution of a distributed system described by a configuration C₀ is of the form C₀ ¾®^e₀ C₁ ¾®^e₁ C₂ ¾®^e₂ .... An evolution C₀ ¾®^e₀ C₁ ¾®^e₁ ... is terminating is there exists j such that C_j is final.

If C₀ describes an asynchronous system then an evolution C₀ ¾®^e₀ C₁ ¾®^e₁ ... is admissible if:

For each process P_i there are infinitely many j such that e_j = comp(i).
If e_j = comp(i) and m is created in e_j, then e_k = del(m) for some k.

In an admissible evolution of an asynchronous system, each process performs infinitely many transitions, and every message that is created is delivered. There need be no finite upper bound on the number of events between two transitions of a process or between the creation and the delivery of a message.

If C₀ describes a synchronous system then an evolution C₀ ¾®^e₀ C₁ ¾®^e₁ ... is admissible if it consists of a sequence of rounds. In a round,

all outstanding messages are delivered, and then
each process performs a computation step which results in at most one message being created, addressed to any given process.

There is a fixed bound on the number of events between the creation and delivery of a message.

2 Leader Election

The leader election problem is to design a process that stores just two data -- its name n:N and its status s:{null, leader, follower} -- so that for any k>1, if S is a system consisting of k copies of P -- that may differ only in storing different names -- connected in a ring, and in which there are initially no undelivered messages, then every admissible evolution of S contains a final configuration in which the status of one P is leader, and the status of every other P is follower.

A. The synchronous case

Fix k>1. Choose distinct n₀, ..., n_k-1. P has variables s, initially null, and n, initially n_i in P_i. P's transitions are described as follows, where c refers to the clockwise neighbour:

init	=	c!n active
leader	=	?in leader
follower	=	?in follower
active	=	?in IF (
		done Î data(in): s:=follower c?done follower
		n Î data(in) : s:= leader c!done leader
		n < max(in) : c!max(in) active
		n > max(in) : active)

[Notes: (1) The IF construct means ``if...then...elsif...then...elsif...then...'' and so on. (2) An empty set doesn't actually have a maximum, to the IF wouldn't appear to cover the case where in=Ø. It works (I think) if you pretend that max(Ø) = -1, or that the last branch will always be executed if in=Ø.]

Let n* = maxn₀, ..., n_k-1, and suppose n* = n_m.

Lemma 1

Suppose 0 £ r < k.

In round r,
1. P_m+r sends n*
2. If j ¹ m+r then P_j does not send n*
3. No P_j sends done
4. If j ¹ m then no P_i with i Î [m... j-1] sends j
1. P_j is in control state active.
2. s_j = null

Lemma 2

Suppose 0 £ r < k.

In round k+r, P_m+r sends done.
At the end of round k+r,
1. P_m is in control state leader and s_m = leader
2. If j Î (m,m+r] then P_j is in control state follower and s_j = follower
3. If j Î (m+r, m) then P_j is in control state active and s_j = null

Corollary 3

At the end of round 2k-1,

s_m = leader
If j ¹ m then s_j = follower

Message complexity

The number of messages created in the first 2k rounds is O(k²).

B. The asynchronous case

Consider the system S, but now viewed as being asynchronous.

Lemma 4

Suppose C is a configuration in an admissible evolution of S.

P_j is leader iff s_j = leader.
P_j is follower iff s_j = follower.
P_j is init or active iff s_j = null.
If C contains (P_i, n*, P_i+1) then P_m is active and s_j = null for j ¹ m.
C contains at most one message of the form (P_i, n*, P_i+1).
If j ¹ m and i Î [m ... j-1], then C does not contain (P_i, n_j, P_i+1).
If j ¹ m then s_j ¹ leader.
If C contains (P_m+j, done, P_m+j+1), where d £ j < k, then s_m = leader and s_i = follower for i Î (m, m+j] and P_i is active for i Î (m+j, m).
C contains at most one message of the form (P_i, done, P_i+1).

Lemma 5

Suppose C₀ ¾®^e₀ C₁ ¾®^e₁ ... is an admissible evolution of S.

If 0 £ j<k then some C_h contains an undelivered message (P_m+j, n*, P_m+j+1).
If o £ j<k then in some C_h, s_m = leader and s_i = follower for i Î (m, m+j] and there is an undelivered message (P_m+j, done, P_m+j+1)

Corollary 6

Every admissible evolution of S contains a configuration in which s_m = leader and s_j = follower for j ¹ m.

Further reading on Leader Election: Lynch ch. 3 and 15, Attiya & Welch ch. 3

3 Logical clocks and vector clocks

Suppose C₀ ¾®^e₀ C₁ ¾®^e₁ ... is an evolution of an asynchronous system S consisting of n processes P₁, ..., P_n. Assume that e_i ¹ e_j if i ¹ j and that no two messages created in the evolution are identical. Let e₀ⁱ, e₁ⁱ, ... be the events performed by P_i (in order of occurrence in the evolution).

Definition

The happens before relation is the smallest transitive relation such that:

e_jⁱ e_j+1ⁱ for all i, j, and
if m is created in e and consumed in e' then e e'.

References Lynch 18, A&W 6, Coulouris 10, Mullender 4

We derive an evolution of a system S'. Each P_i' has a logical clock, and integer variable c_i, initially 0. If P_i performs e_j then P_i' performs e_j' by carrying out the non-sending actions of E_j, then executing: c_i := 1 + max{c_i, max{t | (P_j', (d,t), P_i') is consumed in e_j'}}, and then carrying out the sending actions of e_j, but sending (d, c_i instead of d.

l(e_j) is defined to be the value of c_i when execution of e_j' finishes.

Lemma 7

If e f is in the evolution of S then l(e) < l(f). The converse does not hold, however.

Example (Send-Consume Diagram)

(7147,2085)(76,-3961) (1201,-2161)150 (1801,-3661)150 (2101,-2161)150 (2401,-3661)150 (2701,-3661)150 (3301,-2161)150 (3901,-2161)150 (4501,-2161)150 (6301,-3661)150 (5326,-3661)150 (5776,-2161)150 (601,-2161)( 1, 0)6600 (601,-3661)( 1, 0)6600 (2401,-3661)( 1, 1)1425 (1276,-2236)( 1,-3)450 (2776,-3586)( 1, 3)450 (3976,-2236)( 1,-1)1275 (4426,-2011)(0,0)[lb]1214.4pt7 (5701,-2011)(0,0)[lb]1214.4pt8 (1726,-3961)(0,0)[lb]1214.4pt2 (2326,-3961)(0,0)[lb]1214.4pt3 (2626,-3961)(0,0)[lb]1214.4pt4 (1126,-2011)(0,0)[lb]1214.4pt1 (2026,-2011)(0,0)[lb]1214.4pt2 (3226,-2011)(0,0)[lb]1214.4pt5 (3826,-2011)(0,0)[lb]1214.4pt6 (5251,-3961)(0,0)[lb]1214.4pt7 (6301,-3961)(0,0)[lb]1214.4pt8 (2101,-2461)(0,0)[lb]1214.4pte (2251,-3511)(0,0)[lb]1214.4ptf (296,-2236)(0,0)[lb]1214.4ptP₁ (296,-3736)(0,0)[lb]1214.4ptP₂

We derive an evolution of a system S''. Each P_i'' has a vector clock, an array variable v_i[1..n], initially v_i[j]=0 for 1 £ j £ n. If P performs e_h then P_i'' performs e_h'' by carrying out the non-sending actions of e_h, then executing

v_i[i]	:=	v_i[i] + 1 and for j ¹ i,
v_i[j]	:=	max{v_i[j], max{t_j \| (P_k'',(d,á t₁..t_nñ),P_i'') is consumed in e_h''}}

and then carrying out the sending actions of e_h, but sending (d,v_i), instead of d. v(e_h) is defined to be the value of v_i when execution of e_h'' finishes.

Note: In any configuration of the evolution of S'', v_j[i] £ v_i[i].

Define v £ w if v[i] £ w[i] " i, and v<w if v £ w and v ¹ w. Also, v and w are incomparable if v ¬ £w and w ¬ £v.

Lemma 8

e f is the evolution of S iff v(e)<v(f).

Proof (sketch)

Þ From the definitions.
Ü Suppose e ¬ f where P_i performs e and P_j performs f. Then v(f)[i]<v(e)[i] so v(e) ¬ £v(f).

Corollary 9

e and f are concurrent, i.e. e ¬ f and f ¬ e, iff v(e) and v(f) are incomparable.

Suppose the links between the processes of S are FIFO queues. For convenience, assume that when an event involves the creation of a message m from P_i to P_j, it also involves the delivery of m to the FIFO queue q_ij, from which P_j can consume messages.

A cut is a (finite) subset of {e₀,e₁,...} that is the union of the events in finite initial segments e₀ⁱ,,...,e_s(i)ⁱ for 1 £ i £ n. A cut determines a configuration: each P_i is in the state after e_s(i)ⁱ has occurred. A cut K is consistent if e Î K and e' e implies e' Î K.

Example

(8422,2562)(601,-3811) (1501,-1561)150 (2251,-3061)150 (2851,-1561)150 (3751,-3061)150 (4201,-3061)150 (6901,-3061)150 (7651,-1561)150 (8251,-3061)150 (4801,-1561)150 (5926,-1561)150 (5326,-1561)150 (901,-1561)( 1, 0)8100 (901,-3061)( 1, 0)8100 (1501,-1636)( 1,-2)675 (3751,-3061)( 3, 2)2025 (4276,-2986)( 1, 3)450 (5476,-1636)( 1,-1)1350 (4051,-3361)( 0, 1)825 (4051,-2536)( 3, 2)900 (4951,-1936)( 1, 4)150 (5551,-3361)( 0, 1)2100 (3976,-3586)(0,0)[lb]1012.0ptcut 1 (3976,-3811)(0,0)[lb]1012.0pt(inconsistent) (5551,-3661)(0,0)[lb]1012.0ptcut 2 (consistent) (601,-1636)(0,0)[lb]1012.0ptP_1 (601,-3136)(0,0)[lb]1012.0ptP_2

A configuration is consistent if it is the configuration determined by a consistent cut.

The Chandy, Lamport distributed snapshot algorithm can be used to construct a consistent configuration in evolution of an asynchronous system.²

For concreteness, suppose P₁,...,P_n are bank processes that send one another amounts of money from accounts they maintain. Assume that if P_i and P_j are adjacent, then P_i sends infinitely many messages to P_j, and assume that each P_i consumes from each q_ki infinitely often.

Introduce a monitor process P₀, which is directly connected to each P_i. P₀ acts as follows (repeatedly):

send take snapshot to each P_i
receive (take snapshot and) state from each P_i, and construct configuration.

The behaviour of P_i (1 £ i £ n) is augmented as follows:

s := state
send take snapshot to each adjacent P_k
Q_hi := áñ (all h¹ 0 adjacent)
status_j := finished
status_k := unifinished (k¹ j adjacent)
(without performing any money-transfer events)
Resume executing money-transfer events.
On consuming v from q_ki with status_k = unifinished, record v in Q_ki.
On consuming take snapshot from q_ki, status_k := finished.
If every status_k = finished, send s together with all Q_ki to P₀, and cease recording.

Global snapshots can be used to detect if a stable property holds, e.g. if there is a deadlock in the system. (Suggested reading: Mullender ch. 4, Lynch ch. 19).

4 Commit Protocols

Consider a synchronous distributed database, whose component processes can fail by stopping. To maintain consistency, no transaction can be committed by one process but aborted by another. More precisely, we consider first the two-phase commit protocol, which ensures that if no process fails then a configuration is reached in which every process fails then a configuration is reached in which every process has committed or every process has aborted. One process is deemed the coordinator.

In phase 1, the coordinator sets its status to the appropriate one of can commit and must abort, and each non-coordinator does likewise and sends its status to the coordinator. If the coordinator's status is can commit and it receives notification from each non-coordinator that its status is can commit, then the decision is commit; otherwise the decision is abort. In phase 2, the coordinator sens the decision to each non-coordinator, which acts on it.

Lemma 10

If in an admissible evolution of the two-phase commit protocol no process fails, then a configuration is reached in which every process has committed or every process has aborted.

The three-phase commit protocol modifies and extends the two-phase commit protocol to take account of the possibility that th coordinator may fail. It achieves the stronger property that a configuration is reached in which every process that has not failed has committed or aborted, and in which it is not the case that some process has committed and some [other] process has aborted.

Suggested reading: Lynch, section 7.

1: I think a multiset (a.k.a. a bag) is like a normal set, but you can have multiple copies of the same thing in it.
2: Hooray.

This document was translated from L^AT_EX by H^EV^EA.