UNIT II

This technique results in considerable saving in the cost; only one scalar is
piggybacked on every message.

1.18 PHYSICAL CLOCK SYNCHRONIZATION: NEWTWORK TIME PROTOCOL

(NTP)
Centralized systems do not need clock synchronization, as they work under a common
clock. But the distributed systems do not follow common clock: each system functions based
on its own internal clock and its own notion of time.The time in distributed systems is
measured in the following contexts:
• The time of the day at which an event happened on a specific machine in the network.
• The time interval between two events that happened on different machines in the network.
• The relative ordering of events that happened on different machines in the network.
 UNIT II Distributed systems CS3551

Clock synchronization is the process of ensuring that physically distributed processors have a
common notion of time.

Due to different clocks rates, the clocks at various sites may diverge with time, and
periodically a clock synchrinization must be performed to correct this clock skew in
distributed systems. Clocks are synchronized to an accurate real-time standard like UTC
(Universal Coordinated Time). Clocks that must not only be synchronized with each other but
also have to adhere to physical time are termed physical clocks. This degree of
synchronization additionally enables to coordinate and schedule actions between multiple
computers connected to a common network.

1.18.1 Basic terminologies:

If Ca and Cb are two different clocks, then:
• Time: The time of a clock in a machine p is given by the function Cp(t),where Cp(t)= t for a
perfect clock.
• Frequency: Frequency is the rate at which a clock progresses. The frequency at time t of
clock CaisCa’(t).
• Offset:Clock offset is the difference between the time reported by a clockand the real time.
The offset of the clock Ca is given by Ca(t)− t. Theoffset of clock C a relative to Cb at time t ≥
0 is given by Ca(t)- Cb(t)
• Skew: The skew of a clock is the difference in the frequencies of the clockand the perfect
clock. The skew of a clock Ca relative to clock Cb at timet is Ca’(t)- Cb’(t).
• Drift (rate): The drift of clock Ca the second derivative of the clockvalue with respect to
time. The drift is calculated as:

1.18.2 Clocking Inaccuracies

Physical clocks are synchronized to an accurate real-time standard like UTC
(Universal Coordinated Time). Due to the clock inaccuracy discussed above, a timer (clock)
is said to be working within its specification if:

- maximum skew rate.

1. Offset delay estimation

A time service for the Internet - synchronizes clients to UTC Reliability from
redundant paths, scalable, authenticates time sources Architecture. Thedesign of NTP
involves a hierarchical tree of time servers with primary serverat the root synchronizes with
the UTC. The next level contains secondaryservers, which act as a backup to the primary
server. At the lowest level isthe synchronization subnet which has the clients.

2. Clock offset and delay estimation

A source node cannot accurately estimate the local time on thetarget node due to varying
message or network delays between the nodes.This protocol employs a very common practice
of performing several trialsand chooses the trial with the minimum delay.

Dr. Gopikrishnan M 37
 UNIT II Distributed systems CS3551

Fig 1.24: Behavior of clocks

Fig 1.30 a) Offset and delay estimation Fig 1.30 b) Offset and delay estimation
between processes from same server between processes from different servers

Let T1, T2, T3, T4 be the values of the four mostrecent timestamps. The clocks A and
B are stable andrunning at the same speed. Let a = T1 − T3 and b = T2 − T4. If the
networkdelay difference from A to B and from B to A, called differential delay, is
small, the clock offset and roundtrip delay of B relative to A at time T4are approximately
given by the following:

Each NTP message includes the latest three timestamps T1, T2, andT3, while T4 is
determined upon arrival.

Dr. Gopikrishnan M 38
 UNIT II Distributed systems CS3551

UNIT II

1.14 MODELS OF PROCESS COMMUNICATIONS

There are two basic models of process communications

➢ Synchronous: The sender process blocks until the message has been received by the receiver
process. The sender process resumes executiononly after it learns that the receiver process
has accepted the message. The sender and the receiver processes must synchronize to
exchange a message.
➢ Asynchronous: It is non- blocking communication where the sender and the receiver do not
synchronize to exchange a message. The sender process does not wait for the message to be
delivered to the receiver process. The message is buffered by the system and is delivered to
the receiver process when it is ready to accept the message. A buffer overflow may occur if a
process sends a large number of messages in a burst to another process, thus causing a
message burst.

Asynchronous communication achieves high degree of parallelism and non-

determinism at the cost of implementation complexity with buffers. On the other hand,
synchronization is simpler with low performance. The occurrence of deadlocks and frequent
blocking of events prevents it from reaching higher performance levels.

1.15 LOGICAL TIME

Logical clocks are based on capturing chronological and causal relationships of processes and
ordering events based on these relationships.

Dr. Gopikrishnan M 30
 UNIT II Distributed systems CS3551

Precise physical clocking is not possible in distributed systems. The asynchronous

distributed systems spans logical clock for coordinating the events.Three types of logical
clock are maintained in distributed systems:
• Scalar clock
• Vector clock
• Matrix clock

In a system of logical clocks, every process has a logical clock that is advanced using
a set of rules. Every event is assigned a timestamp and the causality relation between events
can be generally inferred from their timestamps.
The timestamps assigned to events obey the fundamental monotonicity property; that
is, if an event a causally affects an event b, then the timestamp of a is smaller than the
timestamp of b.

Differences between physical and logical clock

Physical Clock Logical Clock
A physical clock is a physical procedure A logical clock is a component for catching
combined with a strategy for measuring that sequential and causal connections in a dispersed
procedure to record the progression of time. framework.
The physical clocks are based on cyclic processes A logical clock allows global ordering on
such as a events from different processes.
celestial rotation.

1.15.1 A Framework for a system of logical clocks

A system of logical clocks consists of a time domain T and a logical clock C. Elements of T form a
partially ordered set over a relation <. This relation is usually called the happened before or causal
precedence.
The logical clock C is a function that maps an event e in a distributed system to an element in
the time domain T denoted as C(e).
such that
for any two events ei and ej,. ei→ej C(ei)< C(ej).
This monotonicity property is called the clock consistency condition.When T and C satisfy
the following condition,

Then the system of clocks is strongly consistent.

1.15.2 Implementing logical clocks

The two major issues in implanting logical clocks are:
▪ Data structures: representation of each process
▪ Protocols: rules for updating the data structures to ensure consistent conditions.

Data structures:
Each process pimaintains data structures with the given capabilities:
• A local logical clock (lci), that helps process pi measure itsown progress.
• A logical global clock (gci), that is a representation of process pi’s local view of the logical
global time. It allows this process to assignconsistent timestamps to its local events.

Dr. Gopikrishnan M 31
 UNIT II Distributed systems CS3551

Protocol:
The protocol ensures that a process’s logical clock, and thus its view of theglobal
time, is managed consistently with the following rules:
Rule 1: Decides the updates of the logical clock by a process. It controls send, receive and
other operations.
Rule 2: Decides how a process updates its global logical clock to update its view of the
global time and global progress. It dictates what information about the logical time is
piggybacked in a message and how this information is used by the receiving process to
update its view of the global time.

1.16 Scalar Time

Scalar time is designed by Lamport to synchronize all the events in distributed
systems. A Lamport logical clock is an incrementing counter maintained in each process.
This logical clock has meaning only in relation to messages moving between processes.
When a process receives a message, it resynchronizes its logical clock with that sender
maintainingcausal relationship.
The Lamport’s algorithm is governed using the following rules:
• The algorithm of Lamport Timestamps can be captured in a few rules:
• All the process counters start with value 0.
• A process increments its counter for each event (internal event, message sending, message
receiving) in that process.
• When a process sends a message, it includes its (incremented) counter value with the
message.
• On receiving a message, the counter of the recipient is updated to the greater of its current
counter and the timestamp in the received message, and then incremented by one.

If Ci is the local clock for process Pi then,

• if a and b are two successive events in Pi, then Ci(b) = Ci(a) + d1, where d1 > 0
• if a is the sending of message m by Pi, then m is assigned timestamp tm = Ci(a)
• if b is the receipt of m by Pj, then Cj(b) = max{Cj(b), tm + d2}, where d2 > 0

Rules of Lamport’s clock

Rule 1: Ci(b) = Ci(a) + d1, where d1 > 0
Rule 2: The following actions are implemented when pi receives a message m with timestamp Cm:
a) Ci= max(Ci, Cm)
b) Execute Rule 1
c) deliver the message

Fig 1.20: Evolution of scalar time

Basic properties of scalar time:

Dr. Gopikrishnan M 32
 UNIT II Distributed systems CS3551

1. Consistency property: Scalar clock always satisfies monotonicity. A monotonic clock

only increments its timestamp and never jump.Hence it is consistent.

2. Total Reordering:Scalar clocks order the events in distributed systems.But all the events
do not follow a common identical timestamp. Hence a tie breaking mechanism is essential to
order the events. The tie breaking is done through:
• Linearly order process identifiers.
• Process with low identifier value will be given higher priority.

The term (t, i) indicates timestamp of an event, where t is its time of occurrence and i is the
identity of the process where it occurred.
The total order relation ( ) over two events x and y with timestamp (h, i) and (k, j) is given by:

A total order is generally used to ensure liveness properties in distributed algorithms.

3. Event Counting
If event e has a timestamp h, then h−1 represents the minimum logical duration,
counted in units of events, required before producing the event e. This is called height of the
event e. h-1 events have been produced sequentially before the event e regardless of the
processes that produced these events.

4. No strong consistency
The scalar clocks are not strongly consistent is that the logical local clock and logical
global clock of a process are squashed into one, resulting in the loss causal dependency
information among events at different processes.

1.17 Vector Time

The ordering from Lamport's clocks is not enough to guarantee that if two events
precede one another in the ordering relation they are also causally related. Vector Clocks use
a vector counter instead of an integer counter. The vector clock of a system with N processes
is a vector of N counters, one counter per process. Vector counters have to follow the
following update rules:
• Initially, all counters are zero.
• Each time a process experiences an event, it increments its own counter in the vector by one.
• Each time a process sends a message, it includes a copy of its own (incremented) vector in
the message.
• Each time a process receives a message, it increments its own counter in the vector by one
and updates each element in its vector by taking the maximum of the value in its own vector
counter and the value in the vector in the received message.

The time domain is represented by a set of n-dimensional non-negative integer vectors in vector
time.

Rules of Vector Time

Rule 1: Before executing an event, process pi updates its local logical time
as follows:

Dr. Gopikrishnan M 33
 UNIT II Distributed systems CS3551

Rule 2: Each message m is piggybacked with the vector clock vt of the sender
process at sending time. On the receipt of such a message (m,vt), process
pi executes the following sequence of actions:
1. update its global logical time

2. execute R1
3. deliver the message m

Fig 1.21: Evolution of vector scale

1.17.1 Basic properties of vector time
1. Isomorphism:
• “→” induces a partial order on the set of events that are produced by a distributed execution.
• If events x and y are timestamped as vh and vk then,

• There is an isomorphism between the set of partially ordered events produced by a distributed
computation and their vector timestamps.
• If the process at which an event occurred is known, the test to compare two timestamps can
be simplified as:

2. Strong consistency
The system of vector clocks is strongly consistent; thus, by examining the vector
timestamp of two events, we can determine if the events are causally related.

3. Event counting
If an event e has timestamp vh, vh[j] denotes the number of events executed by
process pj that causally precede e.

Vector clock ordering relation

t[i]- timestamp of process i.

Dr. Gopikrishnan M 34
 UNIT II Distributed systems CS3551

1.17.2 Implementation of vector clock

If the number of processes in a distributed computation is large, then vector clocks
will require piggybacking of huge amount of information in messages for the purpose of
disseminating time progress and updating clocks. There are two implementation techniques:
• Singhal–Kshemkalyani’s differential technique:
− This approach improves the message passing mechanism by only sending updates to the
vector clock that have occurred since the last message sent from Process(i) → Process(j).
− This drastically reduces the message size being sent, but does require O(n²) storage. This is
due to each node now needing to remember, for each other process, the state of the vector at
the last message sent.
− This also requires FIFO message passing between processes, as it relies upon the guarantee of
knowing what the last message sent is, and if messages arrive out of order this would not be
possible.
− If entries i1, i2, …, in of the vector clock at pi have changed to v1, v2,…, vn respectively, since
the last message sentto pj, then process pi piggybacks a compressed timestamp of the form.

− When pj receives this message, it updates its vector clock as follows:

− This cuts down the message size, communication bandwidth and buffer (to store messages)
requirements.
− The storage overhead is resolved by maintaining two vectors by process pi :

Fig 1.22: Vector clocks progress in Singhal–Kshemkalyani technique

• Fowler-Zwaenepoel direct-dependency technique:
− This technique further reduces the message size by only sending the single clock value of the
sending process with a message.

Dr. Gopikrishnan M 35
 UNIT II Distributed systems CS3551

− However, this means processes cannot know their transitive dependencies when looking at
the causality of events.
− In order to gain a full view of all dependencies that lead to a specific event, an offline search
must be made across processes.
− Each process pimaintains a dependency vector Di. Initially,

− i is updated as follows:
1. Whenever an event occurs at pi such that,

2. When a process pi sends a message to process pj, it piggybacks the updatedvalue of Di[i] in
the message.
3. When pi receives a message from pj with piggybacked value d, piupdates its dependency
vector as follows: Di[j]:= max{Di[j], d}.

Fig 1.23: Vector clock progress in Fowler–Zwaenepoel technique

• Process p4 sends a message to process p3, it piggybacks a scalar that indicates the direct
dependency of p3 on p4 because of this message.
• Process p3 sends a message to process p2 piggybacking a scalar to indicate the direct
dependency of p2 on p3 because of this message.
• Process p2 is in fact indirectly dependent on process p4 since process p3 is dependent on
process p4. However, process p2 is never informed about its indirect dependency on p4.

Dr. Gopikrishnan M 36
 CS3551 Distributed Systems – UNIT II

MESSAGE ORDERING & SNAPSHOTS

2.1 MESSAGE ORDERING AND GROUP COMMUNICATION
As the distributed systems are a network of systems at various physical locations, the
coordination between them should always be preserved. The message ordering means the
order of delivering the messages to the intended recipients. The common message order
schemes are First in First out (FIFO), non FIFO, causal order and synchronous order. In case
of group communication with multicasting, the causal and total ordering scheme is followed.
It is also essential to define the behaviour of the system in case of failures. The following
are the notations that are widely used in this chapter:
• Distributed systems are denoted by a graph (N, L).
• The set of events are represented by event set {E, }
• Message is denoted as mi: send and receive events as si and ri respectively.
• Send (M) and receive (M) indicates the message M send and received.
• ab denotes a and b occurs at the same process
• The send receive pairs ={(s, r) Ei x Ejcorresponds to r}

2.1.1 Message Ordering Paradigms

The message orderings are
(i) non-FIFO
(ii) FIFO
(iii) causal order
(iv) synchronous order
There is always a trade-off between concurrency and ease of use and implementation.

Asynchronous Executions
An asynchronous execution (or A-execution) is an execution (E, ≺) for which the causality relation
is a partial order.
• There cannot be any causal relationship between events in asynchronous execution.
• The messages can be delivered in any order even in non FIFO.
• Though there is a physical link that delivers the messages sent on it in FIFO order due
to the physical properties of the medium, a logicallink may be formed as a composite
of physical links and multiple paths mayexist between the two end points of the
logical link.

Fig 2.1: a) FIFO executions b) non FIFO executions

Dr. Gopikrishnan M 1
 CS3551 Distributed Systems – UNIT II

FIFO executions

A FIFO execution is an A-execution in which, for all

• The logical link is non-FIFO.

• FIFO logical channels can be realistically assumed when designing distributed
algorithms since most of the transport layer protocols follow connection oriented
service.
• A FIFO logical channel can be created over a non-FIFO channel by using a
separate numbering scheme to sequence the messages on each logical channel.
• The sender assigns and appends a <sequence_num, connection_id> tuple to each
message.
• The receiver uses a buffer to order the incoming messages as per the sender’s
sequence numbers, and accepts only the “next” message in sequence.

Causally Ordered (CO) executions

CO execution is an A-execution in which, for all,

• Two send events s and s’ are related by causality ordering (not physical time
ordering), then a causally ordered execution requires that their corresponding receive
events r and r’ occur in the same order at all common destinations.
• If s and s’ are not related by causality, then CO is vacuously satisfied.
• Causal order is used in applications that update shared data, distributed shared
memory, or fair resource allocation.
• A message m that arrives in the local OS buffer at Pi may have to be delayed until the
messages that were sent to Pi causally before m was sent have arrived and are
processed by the application.
• The delayed message m is then given to the application for processing. The event of
an application processing an arrived message is referred to as a delivery event.
• No message overtaken by a chain of messages between the same (sender, receiver)
pair.

If send(m1) ≺ send(m2) then for each common destination d of messages m1 and m2,
deliverd(m1) ≺deliverd(m2) must be satisfied.

Other properties of causal ordering

1. Message Order (MO): A MO execution is an A-execution in which, for all

.
2. Empty Interval Execution: An execution (E ≺) is an empty-interval (EI)
execution if for each pair of events (s, r) ∈ T, the open interval set

Dr. Gopikrishnan M 2
 CS3551 Distributed Systems – UNIT II

in the partial order is empty.

3. An execution (E, ≺) is CO if and only if for each pair of events (s, r) ∈ T and each
event e ∈ E,
• weak common past:

• weak common future:

Synchronous Execution
• When all the communication between pairs of processes uses synchronous send and receives
primitives, the resulting order is the synchronous order.
• The synchronous communication always involves a handshake between the receiver and the
sender, the handshake events may appear to be occurring instantaneously and atomically.
• The instantaneous communication property of synchronous executions requires a modified
definition of the causality relation because for each (s, r) ∈ T, the send event is not causally
ordered before the receive event.
• The two events are viewed as being atomic and simultaneous, and neither event precedes the
other.

Fig 2.2 a) Execution in an asynchronous Fig 2.2 b) Equivalent synchronous

system communication

Causality in a synchronous execution: The synchronous causality relation << on E is the

smallest transitive relation that satisfies the following:
S1: If x occurs before y at the same process, then x << y.

S2: If (s, r ∈ T, then) for all x ∈ E, [(x<< s ⇐⇒ x<<r) and (s<< x ⇐⇒ r<< x)].

S3: If x<<y and y<<z, then x<<z.

Synchronous execution: A synchronous execution or S-execution is an execution (E, <<) for

which the causality relation << is a partial order.

Timestamping a synchronous execution: An execution (E, ≺) is synchronous if and only if

there exists a mapping from E to T (scalar timestamps) such that
• for any message M, T(s(M)) = T(r(M))

• for each process Pi , if ei≺ei’, then T(ei) < T(ei’).

Dr. Gopikrishnan M 3
 CS3551 Distributed Systems – UNIT II

2.2 Asynchronous execution with synchronous communication

When all the communication between pairs of processes is by using synchronous send
and receive primitives, the resulting order is synchronous order. The algorithms run on
asynchronous systems will not work in synchronous system and vice versa is also true.

Realizable Synchronous Communication (RSC)

A-execution can be realized under synchronous communication is called a realizable with
synchronous communication (RSC).

• An execution can be modeled to give a total order that extends the partial order
(E, ≺).
• In an A-execution, the messages can be made to appear instantaneous if there exist a
linear extension of the execution, such that each send event is immediately followed
by its corresponding receive event in this linear extension.

Non-separated linear extension is an extension of (E, ≺) is a linear extension of (E, ≺) such that
for each pair (s, r) ∈ T, the interval { x∈ E s ≺ x ≺ r } is empty.

A A-execution (E, ≺) is an RSC execution if and only if there exists a non-separated linear
extension of the partial order (E, ≺).
• In the non-separated linear extension, if the adjacent send event and its corresponding
receive event are viewed atomically, then that pair of events shares a common past
and a common future with each other.

Crown
Let E be an execution. A crown of size k in E is a sequence <(si, ri), i ∈{0,…, k-1}> of pairs of
corresponding send and receive events such that: s0 ≺ r1, s1 ≺ r2, sk−2 ≺ rk−1, sk−1 ≺ r0.

The crown is <(s1, r1) (s2, r2)> as we have s1 ≺ r2 and s2 ≺ r1. Cyclic dependencies
may exist in a crown. The crown criterion states that an A-computation is RSC, i.e., it can be
realized on a system with synchronous communication, if and only if it contains no crown.

Timestamp criterion for RSC execution

An execution (E, ≺) is RSC if and only if there exists a mapping from E to T (scalar
timestamps) such that

2.2.1 Hierarchy of ordering paradigms

The orders of executions are:

• Synchronous order (SYNC)
• Causal order (CO)
• FIFO order (FIFO)
• Non FIFO order (non-FIFO)
The Execution order have the following results
− For an A-execution, A is RSC if and only if A is an S-execution.

Dr. Gopikrishnan M 4
 CS3551 Distributed Systems – UNIT II

− RSC ⊂ CO ⊂ FIFO ⊂ A
− This hierarchy is illustrated in Figure 2.3(a), and example executions of each class are
shown side-by-side in Figure 2.3(b)
− The above hierarchy implies that some executions belonging to a class X will not
belong to any of the classes included in X. The degree of concurrency is most in A
and least in SYNC.
− A program using synchronous communication is easiest to develop and verify.
− A program using non-FIFO communication, resulting in an A execution, is hardest to
design and verify.

Fig (a) Fig (b)

Fig 2.3: Hierarchy of execution classes

2.2.3 Simulations
− The events in the RSC execution are scheduled as per some non-separated linear
extension, and adjacent (s, r) events in this linear extension are executed sequentially
in the synchronous system.
− The partial order of the asynchronous execution remains unchanged.
− If an A-execution is not RSC, then there is no way to schedule the events to make
them RSC, without actually altering the partial order of the given A-execution.
− However, the following indirect strategy that does not alter the partial order can be
used.
− Each channel Ci,j is modeled by a control process Pi,j that simulates the channel buffer.
− An asynchronous communication from i to j becomes a synchronous communication
from i to Pi,j followed by a synchronous communication from Pi,j to j.
− This enables the decoupling of the sender from the receiver, a feature that is essential
in asynchronous systems.

Dr. Gopikrishnan M 5
 CS3551 Distributed Systems – UNIT II

Fig 2.4: Modeling channels as processes to simulate an execution using

asynchronous primitives on synchronous system

Synchronous programs on asynchronous systems

− A (valid) S-execution can be trivially realized on an asynchronous system by
scheduling the messages in the order in which they appear in the S-execution.
− The partial order of the S-execution remains unchanged but the communication
occurs on an asynchronous system that uses asynchronous communication primitives.
− Once a message send event is scheduled, the middleware layer waits for
acknowledgment; after the ack is received, the synchronous send primitive completes.

2.3 SYNCHRONOUS PROGRAM ORDER ON AN ASYNCHRONOUS SYSTEM

Non deterministic programs

The partial ordering of messages in the distributed systems makes the repeated runs of
the same program will produce the same partial order, thus preserving deterministic nature.
But sometimes the distributed systems exhibit non determinism:
• A receive call can receive a message from any sender who has sent a message, if the
expected sender is not specified.
• Multiple send and receive calls which are enabled at a process can be executed in an
interchangeable order.
• If i sends to j, and j sends to i concurrently using blocking synchronous calls, there
results a deadlock.
• There is no semantic dependency between the send and the immediately following
receive at each of the processes. If the receive call at one of the processes can be
scheduled before the send call, then there is no deadlock.

2.3.1 Rendezvous
Rendezvous systems are a form of synchronous communication among an arbitrary
number of asynchronous processes. All the processes involved meet with each other, i.e.,
communicate synchronously with each other at one time. Two types of rendezvous systems
are possible:
• Binary rendezvous: When two processes agree to synchronize.
• Multi-way rendezvous: When more than two processes agree to synchronize.

Features of binary rendezvous:

• For the receive command, the sender must be specified. However, multiple recieve
commands can exist. A type check on the data is implicitly performed.
• Send and received commands may be individually disabled or enabled. A command is
disabled if it is guarded and the guard evaluates to false. The guard would likely
contain an expression on some local variables.
• Synchronous communication is implemented by scheduling messages under the
covers using asynchronous communication.

Dr. Gopikrishnan M 6
 CS3551 Distributed Systems – UNIT II

• Scheduling involves pairing of matching send and receives commands that are both
enabled. The communication events for the control messages under the covers do not
alter the partial order of the execution.

2.3.2 Binary rendezvous algorithm

If multiple interactions are enabled, a process chooses one of them and tries to
synchronize with the partner process. The problem reduces to one of scheduling messages
satisfying the following constraints:
• Schedule on-line, atomically, and in a distributed manner.
• Schedule in a deadlock-free manner (i.e., crown-free).
• Schedule to satisfy the progress property in addition to the safety property.

Steps in Bagrodia algorithm

1. Receive commands are forever enabled from all processes.
2. A send command, once enabled, remains enabled until it completes, i.e., it is not
possible that a send command gets before the send is executed.
3. To prevent deadlock, process identifiers are used to introduce asymmetry to break
potential crowns that arise.
4. Each process attempts to schedule only one send event at any time.

The message (M) types used are: M, ack(M), request(M), and permission(M). Execution
events in the synchronous execution are only the send of the message M and receive of the
message M. The send and receive events for the other message types – ack(M), request(M),
and permission(M) which are control messages. The messages request(M), ack(M), and
permission(M) use M’s unique tag; the message M is not included in these messages.
--------------------------------------------------------------------------------------------------------------------------

(message types)

M, ack(M), request(M), permission(M)

(1) Pi wants to execute SEND(M) to a lower priority process Pj:

Pi executes send(M) and blocks until it receives ack(M) from Pj . The send event SEND(M) now
completes.

Any M’ message (from a higher priority processes) and request(M’) request for synchronization (from
a lower priority processes) received during the blocking period are queued.

(2) Pi wants to execute SEND(M) to a higher priority process Pj:

(2a) Pi seeks permission from Pj by executing send(request(M)).

// to avoid deadlock in which cyclically blocked processes queue // messages.

(2b) While Pi is waiting for permission, it remains unblocked.

(i) If a message M’ arrives from a higher priority process Pk, Pi accepts M’ by scheduling a
RECEIVE(M’) event and then executes send(ack(M’)) to Pk.

Dr. Gopikrishnan M 7
 CS3551 Distributed Systems – UNIT II

(ii) If a request(M’) arrives from a lower priority process Pk, Pi executes send(permission(M’)) to Pk
and blocks waiting for the messageM’. WhenM’ arrives, the RECEIVE(M’) event is executed.

(2c) When the permission(M) arrives, Pi knows partner Pj is synchronized and Pi executes send(M).
The SEND(M) now completes.

(3) request(M) arrival at Pi from a lower priority process Pj:

At the time a request(M) is processed by Pi, process Pi executes send(permission(M)) to Pj and blocks
waiting for the message M. When M arrives, the RECEIVE(M) event is executed and the process
unblocks.

(4) Message M arrival at Pi from a higher priority process Pj:

At the time a message M is processed by Pi, process Pi executes RECEIVE(M) (which is assumed to
be always enabled) and then send(ack(M)) to Pj .

(5) Processing when Pi is unblocked:

When Pi is unblocked, it dequeues the next (if any) message from the queue and processes it as a
message arrival (as per rules 3 or 4).

---------------------------------------------------------------------------------------------------------

Fig 2.5: Bagrodia Algorithm

2.4 GROUP COMMUNICATION

Group communication is done by broadcasting of messages. A message broadcast is
the sending of a message to all members in the distributed system. The communication may
be
• Multicast: A message is sent to a certain subset or a group.
• Unicasting: A point-to-point message communication.

The network layer protocol cannot provide the following functionalities:

▪ Application-specific ordering semantics on the order of delivery of messages.
▪ Adapting groups to dynamically changing membership.
▪ Sending multicasts to an arbitrary set of processes at each send event.
▪ Providing various fault-tolerance semantics.
▪ The multicast algorithms can be open or closed group.

Differences between closed and open group algorithms:

Closed group algorithms Open group algorithms

If sender is also one of the receiver in the If sender is not a part of the communication
multicast algorithm, then it is closed group group, then it is open group algorithm.
algorithm.
They are specific and easy to implement. They are more general, difficult to design and
expensive.
It does not support large systems where client It can support large systems.
processes have short life.

Dr. Gopikrishnan M 8
 CS3551 Distributed Systems – UNIT II

2.5 CAUSAL ORDER (CO)

In the context of group communication, there are two modes of communication:
causal order and total order. Given a system with FIFO channels, causal order needs to be
explicitly enforced by a protocol. The following two criteria must be met by a causal
ordering protocol:
• Safety: In order to prevent causal order from being violated, a message M that
arrives at a process may need to be buffered until all system wide messages sent in the
causal past of the send (M) event to that same destination have already arrived. The
arrival of a message is transparent to the application process. The delivery event
corresponds to the receive event in the execution model.
• Liveness: A message that arrives at a process must eventually be delivered to the
process.

2.5.1 The Raynal–Schiper–Toueg algorithm

• Each message M should carry a log of all other messages sent causally before M’s
send event, and sent to the same destination dest(M).
• The Raynal–Schiper–Toueg algorithm canonical algorithm is a representative of
several algorithms that reduces the size of the local space and message space
overhead by various techniques.
• This log can then be examined to ensure whether it is safe to deliver a message.
• All algorithms aim to reduce this log overhead, and the space and time overhead of
maintaining the log information at the processes.
• To distribute this log information, broadcast and multicast communication is used.
• The hardware-assisted or network layer protocol assisted multicast cannot efficiently
provide features:
➢ Application-specific ordering semantics on the order of delivery of messages.
➢ Adapting groups to dynamically changing membership.
➢ Sending multicasts to an arbitrary set of processes at each send event.
➢ Providing various fault-tolerance semantics

2.6 Causal Order (CO)

An optimal CO algorithm stores in local message logs and propagates on messages,
information of the form d is a destination of M about a messageM sent in the causal past, as
long as and only as long as:
Propagation Constraint I: it is not known that the message M is delivered to d.
Propagation Constraint II: it is not known that a message has been sent to d in the causal
future of Send(M), and hence it is not guaranteed using a reasoning based on transitivity that
the message M will be delivered to d in CO.

Dr. Gopikrishnan M 9
 CS3551 Distributed Systems – UNIT II

Fig 2.6: Conditions for causal ordering

The Propagation Constraints also imply that if either (I) or (II) is false, the information
“d ∈ M.Dests” must not be stored or propagated, even to remember that (I) or (II) has been
falsified:
▪ not in the causal future of Deliverd(M1, a)
▪ not in the causal future of e k, c where d ∈Mk,cDests and there is no other
message sent causally between Mi,a and Mk, c to the same destination d.

Information about messages:

(i) not known to be delivered
(ii) not guaranteed to be delivered in CO, is explicitly tracked by the algorithm using (source,
timestamp, destination) information.
Information about messages already delivered and messages guaranteed to be delivered in
CO is implicitly tracked without storing or propagating it, and is derived from the explicit
information. The algorithm for the send and receive operations is given in Fig. 2.7 a) and b).
Procedure SND is executed atomically. Procedure RCV is executed atomically except for a
possible interruptionin line 2a where a non-blocking wait is required to meet the Delivery
Condition.

Dr. Gopikrishnan M 10
 CS3551 Distributed Systems – UNIT II

Fig 2.7 a) Send algorithm by Kshemkalyani–Singhal to optimally implement causal

ordering

Fig 2.7 b) Receive algorithm by Kshemkalyani–Singhal to optimally implement causal

ordering

The data structures maintained are sorted row–major and then column–major:

Dr. Gopikrishnan M 11
 CS3551 Distributed Systems – UNIT II

1. Explicit tracking:
▪ Tracking of (source, timestamp, destination) information for messages (i) not known to be
delivered and (ii) not guaranteed tobe delivered in CO, is done explicitly using the I.Dests
field of entries inlocal logs at nodes and o.Dests field of entries in messages.
▪ Sets li,aDestsand oi,a. Dests contain explicit information of destinations to which Mi,ais not
guaranteed to be delivered in CO and is not known to be delivered.
▪ The information about d ∈Mi,a .Destsis propagated up to the earliestevents on all causal
paths from (i, a) at which it is known that Mi,a isdelivered to d or is guaranteed to be
delivered to d in CO.

2. Implicit tracking:
▪ Tracking of messages that are either (i) already delivered, or (ii) guaranteed to be
delivered in CO, is performed implicitly.
▪ The information about messages (i) already delivered or (ii) guaranteed tobe delivered
in CO is deleted and not propagated because it is redundantas far as enforcing CO is
concerned.
▪ It is useful in determiningwhat information that is being carried in other messages and
is being storedin logs at other nodes has become redundant and thus can be purged.
▪ Thesemantics are implicitly stored and propagated. This information about messages
that are (i) already delivered or (ii) guaranteed to be delivered in CO is tracked
without explicitly storing it.
▪ The algorithm derives it from the existing explicit information about messages (i) not
known to be delivered and (ii) not guaranteed to be delivered in CO, by examining
only oi,aDests or li,aDests, which is a part of the explicit information.

Fig 2.8: Illustration of propagation constraints

Multicasts M5,1and M4,1

Message M5,1 sent to processes P4 and P6 contains the piggybacked information M5,1.
Dest= {P4, P6}. Additionally, at the send event (5, 1), the information M5,1.Dests = {P4,P6}
is also inserted in the local log Log5. When M5,1 is delivered to P6, the (new) piggybacked
information P4 ∈ M5,1 .Dests is stored in Log6 as M5,1.Dests ={P4} information about P6 ∈

Dr. Gopikrishnan M 12
 CS3551 Distributed Systems – UNIT II

M5,1.Dests which was needed for routing, must not be stored in Log6 because of constraint I.
In the same way when M5,1 is delivered to process P4
at event (4, 1), only the new piggybacked information P6 ∈ M5,1 .Dests is inserted in Log4 as
M5,1.Dests =P6which is later propagated duringmulticast M4,2.

Multicast M4,3
At event (4, 3), the information P6 ∈M5,1.Dests in Log4 is propagated onmulticast M4,3only to
process P6 to ensure causal delivery using the DeliveryCondition. The piggybacked
information on message M4,3sent to process P3must not contain this information because of
constraint II. As long as any future message sent to P6 is delivered in causal order w.r.t.
M4,3sent to P6, it will also be delivered in causal order w.r.t. M5,1. And as M5,1 is already
delivered to P4, the information M5,1Dests = ∅ is piggybacked on M4,3 sent to P 3. Similarly,
the information P6 ∈ M5,1Dests must be deleted from Log4 as it will no longer be needed,
because of constraint II. M5,1Dests = ∅ is stored in Log4 to remember that M5,1 has been
delivered or is guaranteed to be delivered in causal order to all its destinations.

Learning implicit information at P2 and P3

When message M4,2is received by processes P2 and P3, they insert the (new) piggybacked
information in their local logs, as information M5,1.Dests = P6. They both continue to store
this in Log2 and Log3 and propagate this information on multicasts until they learn at events
(2, 4) and (3, 2) on receipt of messages M3,3and M4,3, respectively, that any future message is
expected to be delivered in causal order to process P6, w.r.t. M5,1sent toP6. Hence by
constraint II, this information must be deleted from Log2 andLog3. The flow of events is
given by;
• When M4,3 with piggybacked information M5,1Dests = ∅ is received byP3at (3, 2), this
is inferred to be valid current implicit information aboutmulticast M5,1because the log
Log3 already contains explicit informationP6 ∈M5,1.Dests about that multicast.
Therefore, the explicit informationin Log3 is inferred to be old and must be deleted to
achieve optimality. M5,1Dests is set to ∅ in Log3.
• The logic by which P2 learns this implicit knowledge on the arrival of M3,3is
identical.

Processing at P6
When message M5,1 is delivered to P6, only M5,1.Dests = P4 is added to Log6. Further, P6
propagates only M5,1.Dests = P4 on message M6,2, and this conveys the current implicit
information M5,1 has been delivered to P6 by its very absence in the explicit information.
• When the information P6 ∈ M5,1Dests arrives on M4,3, piggybacked as M5,1 .Dests
= P6 it is used only to ensure causal delivery of M4,3 using the Delivery Condition,
and is not inserted in Log6 (constraint I) – further, the presence of M5,1 .Dests = P4
in Log6 implies the implicit information that M5,1 has already been delivered to
P6. Also, the absence of P4 in M5,1 .Dests in the explicit piggybacked information
implies the implicit information that M5,1 has been delivered or is guaranteed to be
delivered in causal order to P4, and, therefore, M5,1. Dests is set to ∅ in Log6.
• When the information P6 ∈ M5,1 .Dests arrives on M5,2 piggybacked as M5,1. Dests
= {P4, P6} it is used only to ensure causal delivery of M4,3 using the Delivery
Condition, and is not inserted in Log6 because Log6 contains M5,1 .Dests = ∅,

Dr. Gopikrishnan M 13
 CS3551 Distributed Systems – UNIT II

which gives the implicit information that M5,1 has been delivered or is guaranteed
to be delivered in causal order to both P4 and P6.

Processing at P1
• When M2,2arrives carrying piggybacked information M5,1.Dests = P6 this (new)
information is inserted in Log1.
• When M6,2arrives with piggybacked information M5,1.Dests ={P4}, P1learns implicit
information M5,1has been delivered to P6 by the very absence of explicit information
P6 ∈ M5,1.Dests in the piggybacked information, and hence marks information P6 ∈
M5,1Dests for deletion from Log1. Simultaneously, M5,1Dests = P6 in Log1 implies
the implicit information that M5,1has been delivered or is guaranteed to be delivered in
causal order to P4.Thus, P1 also learns that the explicit piggybacked information
M5,1.Dests = P4 is outdated. M5,1.Dests in Log1 is set to ∅.
• The information “P6 ∈M5,1.Dests piggybacked on M2,3,which arrives at P 1, is
inferred to be outdated usingthe implicit knowledge derived from M5,1.Dest= ∅” in
Log1.

2.7 TOTAL ORDER

For each pair of processes Pi and Pj and for each pair of messages Mx and My that are delivered to
both the processes, Pi is delivered Mx before My if and only if Pj is delivered Mxbefore My.

Centralized Algorithm for total ordering

Each process sends the message it wants to broadcast to a centralized process, which
relays all the messages it receives to every other process over FIFO channels.

Complexity: Each message transmission takes two message hops and exactly n messages
in a system of n processes.

Drawbacks: A centralized algorithm has a single point of failure and congestion, and is
not an elegant solution.

Three phase distributed algorithm

Three phases can be seen in both sender and receiver side.
Sender side
Phase 1
• In the first phase, a process multicasts the message M with a locally unique tag and
the local timestamp to the group members.

Phase 2
• The sender process awaits a reply from all the group members who respond with a
tentative proposal for a revised timestamp for that message M.

Dr. Gopikrishnan M 14
 CS3551 Distributed Systems – UNIT II

• The await call is non-blocking.

Phase 3
• The process multicasts the final timestamp to the group.

Fig 2.9: Sender side of three phase distributed algorithm

Receiver Side
Phase 1
• The receiver receives the message with a tentative timestamp. It updates the variable
priority that tracks the highest proposed timestamp, then revises the proposed
timestamp to the priority, and places the message with its tag and the revised
timestamp at the tail of the queue temp_Q. In the queue, the entry is marked as
undeliverable.

Phase 2
• The receiver sends the revised timestamp back to the sender. The receiver then waits
in a non-blocking manner for the final timestamp.

Phase 3
• The final timestamp is received from the multicaster. The corresponding message
entry in temp_Q is identified using the tag, and is marked as deliverable after the
revised timestamp is overwritten by the final timestamp.
• The queue is then resorted using the timestamp field of the entries as the key. As the
queue is already sorted except for the modified entry for the message under
consideration, that message entry has to be placed in its sorted position in the queue.

Dr. Gopikrishnan M 15
 CS3551 Distributed Systems – UNIT II

• If the message entry is at the head of the temp_Q, that entry, and all consecutive
subsequent entries that are also marked as deliverable, are dequeued from temp_Q,
and enqueued in deliver_Q.

Complexity
This algorithm uses three phases, and, to send a message to n − 1 processes, it uses 3(n – 1)
messages and incurs a delay of three message hops

2.8 GLOBAL STATE AND SNAPSHOT RECORDING ALGORITHMS

• A distributed computing system consists of processes that do not share a common
memory and communicate asynchronously with eachother by message passing.
• Each component ofhas a local state. The state of the process is the local memory and a
history of its activity.
• The state of achannel is characterized by the set of messages sent along the channel
lessthe messages received along the channel. The global state of a distributedsystem is
a collection of the local states of its components.
• If shared memory were available, an up-to-date state of the entire systemwould be
available to the processes sharing the memory.
• The absence ofshared memory necessitates ways of getting a coherent and complete
view ofthe system based on the local states of individual processes.
• A meaningfulglobal snapshot can be obtained if the components of the distributed
systemrecord their local states at the same time.
• This would be possible if thelocal clocks at processes were perfectly synchronized or
if there were aglobal system clock that could be instantaneously read by the processes.
• If processes read time froma single common clock, various indeterminatetransmission
delays during the read operation will cause the processes toidentify various physical
instants as the same time.

2.8.1 System Model

• The system consists of a collection of n processes, p1, p2,…,pn that are connected
by channels.
• Let Cij denote the channel from process pi to process pj.
• Processes and channels have states associated with them.
• The state of a process at any time is defined by the contents of processor registers,
stacks, local memory, etc., and may be highly dependent on the local context of
the distributed application.
• The state of channel Cij, denoted by SCij, is given by the set of messages in transit
in the channel.
• The events that may happen are: internal event, send (send (mij)) and receive
(rec(mij)) events.
• The occurrences of events cause changes in the processstate.
• A channel is a distributed entity and its state depends on the local states of the
processes on which it is incident.

Dr. Gopikrishnan M 16
 CS3551 Distributed Systems – UNIT II

• The transit function records the state of the channel Cij.

• In the FIFO model, each channel acts as a first-in first-out message queue and,
thus, message ordering is preserved by a channel.
• In the non-FIFO model, achannel acts like a set in which the sender process adds
messages and thereceiver process removes messages from it in a random order.

2.8.2 A consistent global state

The global state of a distributed system is a collection of the local states ofthe
processes and the channels. The global state is given by:

The two conditions for global state are:

Condition 1 preserves law of conservation of messages.Condition C2 states that in the

collected global state, for everyeffect, its cause must be present.

Law of conservation of messages: Every messagemijthat is recorded as sent in the local state of a
process pi must be capturedin the state of the channel Cij or in the collected local state of the
receiver process pj.
➢ In a consistent global state, every message that is recorded as received isalso recorded
as sent. Such a global state captures the notion of causalitythat a message cannot be
received if it was not sent.
➢ Consistent global statesare meaningful global states and inconsistent global states are
not meaningful in the sense that a distributed system can never be in an
inconsistentstate.

2.8.3 Interpretation of cuts

• Cuts in a space–time diagram provide a powerful graphical aid in representingand
reasoning about the global states of a computation. A cut is a line joiningan arbitrary
point on each process line that slices the space–time diagraminto a PAST and a
FUTURE.
• A consistent global state corresponds to a cut in which every messagereceived in the
PAST of the cut has been sent in the PAST of that cut. Sucha cut is known as a
consistent cut.
• In a consistent snapshot, all the recorded local states of processes are concurrent; that
is, the recorded local state of no process casuallyaffects the recorded local state of any
other process.

2.8.4 Issues in recording global state

The non-availability of global clock in distributed system, raises the following issues:
Issue 1:
How to distinguish between the messages to be recorded in the snapshot from those
not to be recorded?
Answer:
• Any message that is sent by a process before recording its snapshot,must be
recorded in the global snapshot (from C1).

Dr. Gopikrishnan M 17
 CS3551 Distributed Systems – UNIT II

• Any message that is sent by a process after recording its snapshot, mustnot be
recorded in the global snapshot (from C2).

Issue 2:
How to determine the instant when a process takes its snapshot?
The answer
Answer:
A process pj must record its snapshot before processing a message mij that was sent by
process pi after recording its snapshot.

2.9 SNAPSHOT ALGORITHMS FOR FIFO CHANNELS

Each distributed application has number of processes running on different physical
servers. These processes communicate with each other through messaging channels.

A snapshot captures the local states of each process along with the state of each communication channel.

Snapshots are required to:

• Checkpointing
• Collecting garbage
• Detecting deadlocks
• Debugging

2.9.1Chandy–Lamport algorithm
• The algorithm will record a global snapshot for each process channel.
• The Chandy-Lamport algorithm uses a control message, called a marker.
• Aftera site has recorded its snapshot, it sends a marker along all of its
outgoingchannels before sending out any more messages.
• Since channels are FIFO, amarker separates the messages in the channel into those to
be included in the snapshot from those not to be recorded inthe snapshot.
• This addresses issue I1. The role of markers in a FIFO systemis to act as delimiters
for the messages in the channels so that the channelstate recorded by the process at
the receiving end of the channel satisfies thecondition C2.

Dr. Gopikrishnan M 18
 CS3551 Distributed Systems – UNIT II

Fig 2.10: Chandy–Lamport algorithm

Initiating a snapshot
• Process Pi initiates the snapshot
• Pi records its own state and prepares a special marker message.
• Send the marker message to all other processes.
• Start recording all incoming messages from channels Cij for j not equal to i.

Propagating a snapshot
• For all processes Pjconsider a message on channel Ckj.
• If marker message is seen for the first time:
− Pjrecords own sate and marks Ckj as empty
− Send the marker message to all other processes.
− Record all incoming messages from channels Clj for 1 not equal to j or k.
− Else add all messages from inbound channels.

Terminating a snapshot
• All processes have received a marker.
• All process have received a marker on all the N-1 incoming channels.
• A central server can gather the partial state to build a global snapshot.

Correctness of the algorithm

• Since a process records its snapshot when itreceives the first marker on any incoming
channel, no messages that followmarkers on the channels incoming to it are recorded in the
process’s snapshot.
• A process stops recording the state of an incoming channel whena marker is received on that
channel.
• Due to FIFO property of channels, itfollows that no message sent after the marker on that
channel is recorded inthe channel state. Thus, condition C2 is satisfied.
• When a process pj receives message mij that precedes the marker on channel Cij, it acts as
follows: ifprocess pj has not taken its snapshot yet, then it includes mij in its recorded
snapshot. Otherwise, it records mij in the state of the channel Cij. Thus,condition C1 is
satisfied.

Dr. Gopikrishnan M 19
 CS3551 Distributed Systems – UNIT II

Complexity
The recording part of a single instance of the algorithm requires O(e) messages
and O(d) time, where e is the number of edges in the network and d is thediameter of the
network.

2.9.2 Properties of the recorded global state

The recorded global state may not correspond to any of the global states that occurred during
the computation.
This happens because a process can change its state asynchronously before the markers it sent
are received by other sites and the other sites record their states.
But the system could have passed through the recorded global states in some equivalent
executions.
The recorded global state is a valid state in an equivalent execution and if a stable property
(i.e., a property that persists) holds in the system before the snapshot algorithm begins, it holds in the
recorded global snapshot.
Therefore, a recorded global state is useful in detecting stable properties.

Dr. Gopikrishnan M 20