1. Review and Overview 2. Deadlocks 3. Distributed Systems Architecture 4. Theoretical Foundations 5. Distributed Mutual Exclusions

6. Agreement Protocols 7. Distributed Resource Management 8. Distributed Scheduling 9. Secutiry and Protection 10. Recovery and Fault Tolerance

In a few hundred years, when the history of our time
is written from a long-term perspective, it is likely
that the most important event those historians will see
is not technology, not the Internet, not e-commerce. It
is an unprecedented change in human condition.  For the
first time, they will have to manage themselves.

					Peter Drucker

1. Inherent Limitations of a Distributed System
   Absence of Global clock

   • difficult to make temporal order of events
   • difficult to collect up-to-date information on the state of the entire system

   Absence of Shared Memory

   • no up-to-date state of the entire system to any individual process as there's no shared memory
   • coherent view -- all observations of different processes ( computers ) are made at the same physical time we can obtain a coherent but partial view of the system or
     incoherent view of the system
   • complete view ( global state ) -- local views ( local states ) + messages in transit
     difficult to obtain a coherent global state

2. Lamport's Logical Clock

   happened before →

   a → b , if a and b are events in the same process and a occurred before b
   a → b , if a is the event of sending a message m in a process and b is the event of receipt of the same message m by another process
   if a → b and b → c, then a → c ( transitive )

   event a causally affects b if a → b
   concurrent: a || b if !( a → b ) and !( b → a )
   for any two events in a system, either a → b or b → a or a || b

   e11 → e12   , e12 → e22 e21 → e13   , e14 || e24

   To realize the relation → we need a clock C_i at each process P_i in the system

   C_i(a) -- timestamp of event a at P_i
   if a → b, then C(a) < C(b)

   Condition requirements:
   1. for any two events a and b in a process P_i, if a occurs before b, then C_i(a) < C_i(b)
   2. if a is the event of sending a message m in P_i and b is the event of receiving the same message m at process P_j, then C_i(a) < C_j(b)
   Implementation rules:
   1. two successive events in P_i C_i = C_i + d ( d > 0 ) if a and b are two successive events in P_i and a → b then C_i(b) = C_i(a) + d ( d > 0 )
   2. event a: sending of message m by process P_i,
      timestamp of message m : t_m = C_i(a ) then C_j = max ( C_j, t_m + d )    d > 0

      → is irreflixive, defines partial order among events

      Totally ordering relation ( => ) can be defined by ( on top of the above )

      a is any event in process P_i
      b is any event in process P_j a => b iff
      either C_i(a) < C_j(b)
      or C_i(a) = C_j(b) and P_i P_j ( e.g. P_i P_j if i ≤ j, to break ties )
      
   Limitation of Lamport's Clocks
   if a → b then C(a) < C(b)
   but C(a) < C(b) does not necessarily imply a → b

3. Vector Clocks

   n = number of processes in a distributed system
   Each event in process P_i ~ vector clock C_i ( integer vector of length n )

   Ci =

   Ci[1] Ci[2] .. Ci[n]

   C_i[i] ~ P_i's own logical clock
   C_i[j] ~ P_i's best guess of logical time at P_j. More precisely, the time of occurrence of the last event at P_j which "happenned before" the current point in time at P_j
   C_i(a) is referred to as the timestamp of event a at P_i

   Implementation Rules:

   1. two successive events a, b in process P_i C_i(b)[i] = C_i(a)[i] + d    ( d > 0 )
   2. event a at P_i sending message m to process P_j with receiving event b; vector timestamp t_m = C_i(a) is assigned to m all k, C_j(b)[k] = max(C_j(b)[k],t_m[k])

   Assertion.
   At any instant

   all i, all j : C_i[i] ≥ C_j[i]
   

   Comparing two vector timestamps of events a and b
   

   Equal	ta = tb	iff	all i,	ta[i] = tb[i]
   Not Equal	ta ≠ tb	iff	some i,	ta[i] ≠ tb[i]
   Less Than or Equal	ta ≤ tb	iff	all i,	ta[i] ≤ tb[i]
   Not Less Than or Equal To	ta tb	iff	some i,	ta[i] > tb[i]
   Less Than	ta < tb	iff	all i, some j,	ta[i] ≤ tb[i], ta[j] < tb[j]
   Not Less Than	ta tb	iff		!(ta < tb)
   Concurrent	ta || tb	iff		ta[i] tb[i] and tb[i] ta[i]

   Events are causally related if t^a < t^b or t^b < t^a

   Now, a → b   iff   t^a < t^b
   otherwise concurrent

4. Causal Ordering of Messages
   

   if Send( M₁ ) → Send( M₂ )
   then the receipient should receive M₁ before M₂

   i.e. Send( M₁ ) → Send( M₂ ) requires Receive( M₁ ) → Receive( M₂ )

   
   Figure: Violation of causal ordering of messages
   Applications: database replication management, monitoring distributed computations, simplifying distributed algorithms,...

   Solution idea: upon arrival of a message at a process, buffer (delay delivery) the message until the message immediately preceding it is delivered

   Birman-Schiper-Stephenson Protocol: Enforcing Causal Ordering of Messages

   Assumes broadcast communication channels that do not loose or corrupt messages. ( i.e. everyone talks to everyone ). Use vector clocks to "count" number of messages ( i.e. set d = 1 ). n processes.

   1. Process P_i updates vector time C_i and broadcasts message m with timestamp t_m = C_i.
   2. Process P_j ( j ≠ i ) upon receiving message m with timestamp t_m, P_j buffers the message until
      • all messages sent by P_i preceding m have arrived i.e. C_j[i] = t_m[i] - 1
        and
      • P_j has received all messages that P_i had received before sending m. i.e. C_j[k] ≥ t_m[k]    k = 1, 2, .. n, k ≠ i
   3. When the message is finally delivered at P_j, vector time C_j is updated according to vector clock rule 2.

   Schiper-Eggli-Sandoz were able to solve the problem without broadcasting channels

5. Global State
   no global clock, no global memory
   To determine a global system state, a process p must enlist the cooperation of other processes that must record their states and send the recorded local states to p

   processes cannot record their local states at precisely the same instant unless they have access to a common clock

   the global-state-detection algorithm is to be superimposed on the underlying computation; it must run concurrently with but not alter the underlying computation
   diagrams

   Distributed system finite set of processes
   finite set of channels

   process state, channel state

   Single-token conservation system
   ( system has one token )

   

   each process has two states S₁, S₀

   S₁ -- process possesses token
   S₀ -- process does not possess token

   Global state: token in P1	→	Global state: token in C1
   ↑		↓
   Global state: token in C2	←	Global state: token in P2

   four global states : in-P₁, in-C₁, in-P₂, in-C₂

   Suppose

   P₁ records local state at global state in-P₁,

   P₂, channels record local state at global state in-C₁, in-C₁,

   composite global state recorded in this fashion would show two tokens in system, one in P₁, and one in C₁,

   inconsistency arises because P₁ recorded its state before P sent a message along C and C recorded after P sent the message

   let n = # of messages sent along C₁ before P₁ recorded its state
   and
   let n' = # of messages sent along C₁ before C₁ recorded its state

   e.g. n = 0, n' = 1 in the above example
   we see that recorded global state may be inconsistent if n < n'

   Suppose

   C₁ records local state at G.S. in-P₁, ( n' = 0 )

   C₂, P₁, P₂ record local state at G.S. in-P₁, ( n = 1 )

   Here n > n' and again we have inconsistency

   Thus consistent global state requires n = n'

   Similarly, if

   let m = # of messages received along C₁ before P₂ recorded its state
   and
   let m' = # of messages received along C₁ before C₁ recorded its state consitent global state requires m = m'

   Also a consistent global state must satisfy

   n' ≥ m
   ( i.e. # of messages sent along a channel is always larger than or equal to the number of messages received )

   or from above we have

   n ≥ m

6. Global-State-Detection Algorithm Send a special message called marker

   Chandy-Lamport Global State Recording Protocol

   The goal of this distributed algorithm is to capture a consistent global state. It assumes all communication channels are FIFO. It uses a distinguished message called a marker to start the algorithm.

   P_i sends marker

   1. P_i records its local state
   2. For each channel C_ij on which P_i has not already sent a marker, P_i sends a marker before sending other messages.

   P_j receives marker from P_i

   1. If P_j has not recorded its state:
      • a) Records the state of C_ij as empty
      • b) Sends the marker as described above ( Note: it records local state before sending out marker )

   2. If P_j has recorded its state local state LS_j
      • a) Record the state of C_ij to be the sequence of messages received between the computation of LS_j and the marker from C_ij.

   Example

   In this example, all processes are connected by communications channels C_ij. Messages being sent over the channels are represented by arrows between the processes.

   Snapshot s₁:

   • P₁ records LS₁, sends markers on C₁₂ and C₁₃
   • P₂ receives marker from P₁ on C₁₂; it records its state LS₂, records state of C₁₂ as empty, and sends marker on C₂₁ and C₂₃
   • P₃ receives marker from P₁ on C₁₃; it records its state LS₃, records state of C₁₃ as empty, and sends markers on C₃₁ and C₃₂.
   • P₁ receives marker from P₂ on C₂₁; as LS₁ is recorded, it records the state of C₂₁ as empty.
   • P₁ receives marker from P₃ on C₃₁; as LS₁ is recorded, it records the state of C₃₁ as empty.
   • P₂ receives marker from P3 on C₃₂; as LS₂ is recorded, it records the state of C₃₂ as empty.
   • P₃ receives marker from P₂ on C₂₃; as LS₃ is recorded, it records the state of C₂₃ as empty.

   Snapshot s2: now a message is in transit on C₁₂ and C₂₁.

   • P₁ records LS₁, sends markers on C₁₂ and C₁₃
   • P₂ receives marker from P₁ on C₁₂ after the message from P₁ arrives; it records its state LS₂, records state of C₁₂ as empty, and sends marker on C₂₁ and C₂₃
   • P₃ receives marker from P₁ on C₁₃; it records its state LS₃, records state of C₁₃ as empty, and sends markers on C₃₁ and C₃₂.
   • P₁ receives marker from P₂ on C₂₁; as LS₁ is recorded, and a message has arrived since LS₁ was recorded, it records the state of C₂₁ as containing that message.
   • P₁ receives marker from P₃ on C₃₁; as LS₁ is recorded, it records the state of C₃₁ as empty.
   • P₂ receives marker from P₃ on C₃₂; as LS₂ is recorded, it records the state of C₃₂ as empty.
   • P₃ receives marker from P₂ on C₂₃; as LS₃ is recorded, it records the state of C₂₃ as empty.

   The recorded process states and channel states must be collected and assembled to form the global state. ( e.g. send G.S. to all processes in finite time )

   Termination
   each process must ensure that

   • no marker remains forever in an incident input channel
   • it records its state within finite time of initiation of the algorithm

   A remark ( state without proof here )

   Let
   • GS_i = the global state when the alogrithm is initiated,
   • GS_t = the global state when the algorithm is terminated
   • GS_c = collected global state by the algorithm
   • Seq = sequence of action to take the system from GS_i to GS_t
   Then there exists a sequence Seq' that is a permutation of Seq such that GS_c can be reached from GS_i by executing a prefix of Seq' and GS_t can be reached by executing the rest of the actions in Seq'.
   

7. Some definitions
   LS_i -- local state of S_i ( site )

   events -- send( m_ij ), recv( m_ij )

   

   time ( x ) -- time at which state x was recorded
   e.g. time ( LS_i )

   send ( m_ij ) ∈ LS_i iff time ( send ( m_ij ) ) < time ( LS_i )

   recv ( m_ij ) ∈ LS_j iff time ( recv ( m_ij ) ) < time ( LS_j )

   

   transit ( LS_i, LS_j ) = { m_ij | send( m_ij ) ∈ LS_i Λ recv( m_ij ) !∈ LS_j }
   i.e. message in channel

   

   inconsistent ( LS_i, LS_j ) = { m_ij | send( m_ij) !∈ LS_i Λ recv( m_ij ) ∈ LS_j }

   Global State GS = { LS₁, LS₂, ..., LS_n }
   i.e. collection of local states ( may be consistent or inconsistent )

   Consistent Global State: A global state GS = { LS₁, LS₂, ..., LS_n } is consistent iff all i, all j: 1 ≤ i, j ≤ n :: inconsistent( LS_i, LS_j ) = Φ

   Transitless global state: A global state is transitless iff all i, all j: 1 ≤ i, j ≤ n :: transit( LS_i, LS_j ) = Φ

   Strongly consistent global state: A global state is strongly consistent if it is consistent and transitless.

   

8. Cuts of a distributed Computation
   Graphical representation of GS

   C = { c₁, c₂, ... ,c_n }
   c_i -- cut event, local state of site ( or process ) S_i at that instant

   Consistent Cut:

   all S_i, all S_j, no e_i, no e_j such that

   ( e_i → e_j ) and ( e_i → c_j ) and ( e_i c_i )

   

   
   Theorem

   A cut C = { c₁, c₂, ... ,c_n } is a consistent cut iff no two cut events are causally related. ( i.e. every pair of cut events are concurrent )

   Time of a cut

   C = { c₁, c₂, ... ,c_n }

   C_i -- vector clock of c_i

   T_C = sup ( C₁, C₂, ... , C_n )

   T_C[k] = max ( C₁[k], C₂[k], ... , C_n[k] )

   • Theorem
     if C = { c₁, c₂, ... ,c_n } is a cut with vector time T_C, then the cut is consistent iff

     TC =

     C1[1] C2[2] . . Cn[n]

     -------------- (1)

     Proof:
     If C is a consistent cut, then all its events are concurrent. Thus C_i[i] ≥ C_j[i] for all i, j and hence

     TC = sup ( C1, C2, ... , Cn ) =

     C1[1] C2[2] . . Cn[n]

     On the other hand if (1) is true
     we have C_i[i] ≥ C_j[i] for all i, j. This implies that the the events c_i are concurrent and the cut is consistent.

9. Termination Detection

   Huang's Termination Detection Protocol

   The goal of this protocol is to detect when a distributed computation terminates.

   Notation:

   n processes
   P_i process; without loss of generality, let P₀ be the controlling agent
   W_i. weight of process P_i; initially, W₀ = 1 and W_i = 0 for all other i.
   B(W) computation message with assigned weight W
   C(W) control message sent from process to controlling agent with assigned weight W

   
   Protocol

   an active process P_i sends a computation message to P_j

   1. Set W_i' and W_ij to values such that W_i' + W_ij = W_i,
      W_i' > 0, W_ij > 0. (W_i' is the new weight of P_i.)
   2. Send B(W_ij) to P_j
   P_j receives a computation message B(W_ij) from P_i

   1. W_j = W_j + W_ij
   2. If P_j is idle, P_j becomes active

   P_i becomes idle by:

   1. Send C(W_i) to P₀
   2. W_i = 0
   3. P_i becomes idle

   P_i receives a control message C(W):

   1. W_i = W_i + W
   2. If W_i = 1, the computation has completed.

   Example

   The picture shows a process P₀, designated the controlling agent, with W₀ = 1. It asks P₁ and P₂ to do some computation. It sets

   W₀₁ = 0.2
   W₀₂ = 0.3
   W₀ = 0.5

   P₂ in turn asks P₃ and P₄ to do some computations. It sets

   W₂₃ = 0.1
   W₂₄ = 0.1

   When P₃ terminates, it sends C(W₃) = C(0.1) to P₂, which changes W₂ to 0.1 + 0.1 = 0.2.

   When P₂ terminates, it sends C(W₂) = C(0.2) to P₀, which changes W₀ to 0.5 + 0.2 = 0.7.

   When P₄ terminates, it sends C(W₄) = C(0.1) to P₀, which changes W₀ to 0.7 + 0.1 = 0.8.

   When P₁ terminates, it sends C(W₁) = C(0.2) to P₀, which changes W₀ to 0.8 + 0.2 = 1.

   P₀ thereupon concludes that the computation is finished.

   Total number of messages passed: 8 (one to start each computation, one to return the weight).