Distributed Systems

Dr. Rajiv Misra

Department of Computer Science and Engineering
Indian Institute of Technology, Patna

Lecture – 16
Message Ordering and Group Communication

Lecture 16 Message Ordering and Group Communication.

(Refer Slide Time: 00:21)

Preface Recap of Previous Lecture. In previous lecture we have discussed about

‘Termination Detection’ and set of representative termination detection algorithm based
on the concept of snapshot collection, weight throwing and spanning-tree. Content of this
Lecture in this lecture we will discuss about Message ordering, Group Communication
and Multicast that is multi casting.
(Refer Slide Time: 00:49)

Introduction: at the core of the distributed computing is the communication by message

passing among the processes participating in the application. So, in this lecture we will
study several message ordering paradigms for communication such as asynchronous,
synchronous, FIFO, causally ordered, and non-FIFO ordered these orders form the
hierarchy will examine few algorithms to implement these orderings.

Group communication is an important aspect of communication in a distributed system

causally order and total order are popular forms of ordering when doing group multicast
and broadcasts, algorithm to implement these orderings will also be discussed before we
go ahead let me give you few examples where this message ordering and multi casting
will be useful. So, as you know that in a distributed databases the database is normally
replicated to add more than one sites.

Now, whenever there is an update at the database then at all places that is at all the
replicas have to be updated instantaneously. So, if 2 messages are sent for update at
different point of time. So, this particular order in which 2 messages are sent these
updates are to be made in that same order irrespective of how much delay the messages
are ensuring, how this is all done this is all done here in ordering of messages and also
another form of message ordering or communication in a group is basically called multi
casting.
So, multi casting is an example of 2 cases one is called closed group and the other is
called open group. Open group is like online railway reservation system where any
customer at any point of time can come and enter into a system and perform the
operations. So, that client an external person is using a set of nodes for it is application
then it is called an open system open system communication is basically done through
the multicasting.

So, if large number of users are allowed in reservation systems then it is not an easy task
it is a difficult task to basically ensure the correctness and running the application. So,
this will be talked about during the group communication and multi casting how this
paradigm is going to be implemented in the system of message passing system.

(Refer Slide Time: 03:43)

That is called distributed system there are few notations and some of these notations are
explained so we skip this.
(Refer Slide Time: 03:55)

Now, message ordering paradigms message ordering paradigms the order of messages in
a distributed system is an important aspect of system executions because it determines
the messaging behavior that can be expected by the distributed program. So, distributed
program logic greatly depends on this order of delivery to simplify the task of a
programming languages in conjunction with the middleware provides well defined
message behavior.

So, again recreate on the same point the message the order of delivery of the message is
most important as far as distributed applications are concerned and it reflects and it
(Refer Time: 04:41) and it that particular ordering comes from the applications and the
implementations will be basically governed here using the message ordering paradigms.
So, the programmer can send the code for the programming logic with respect to this
particular behavior there are several ordering on the messages which are defined as non-
FIFO; FIFO causal order and synchronous order.
(Refer Slide Time: 05:09)

Let us see a few definitions on these different kind of message orders. So, asynchronous
execution you know that is an execution for which causality relation is a partial order
that we are seen another thing is called FIFO in a first in first out executions is an
asynchronous execution in which for all pairs sender and receiver and another pair of
sender receiver. So, if these sender and these receivers they are happening within a
system and there is an order between the sender then there will be an order in the
receiver also because these senders and these receivers are happening within a particular
process and so within or process they are happening.

So, if this particular send is proceeding over the other send that is s’. So, their receive
also requires this kind of precedence relation and that is called FIFO that is being
preserved. So, here the communication channel has to ensure that the delivery is to be
following the FIFO manner. Similarly the causal order is an execution in which for all
senders and receivers where the sender s and s’ they are basically sending to the same
destination that is that is r.

And if there is a precedence between the send operation; that means, if s precedes s’,
then this particular receive process are also proceeds r’ and r and r ‘ is happening at the
same process that is what this particular symbol indicates. So, the diagram reflects
whatever is the definition of a causal order. So, if there is a precedence between 2 sents
which are happening at different sites then when they will be received and if they have
precedence relations. So, that relation also will be insured at the time of delivery and that
ordering of the message is called causal ordering of messages.

(Refer Slide Time: 08:02)

Now, another kind of ordering is called synchronous; synchronous that sync order and
execution for which the causality relation is a partial order is called the synchronous
order. So, synchronous order is like for every send there will be instantaneous receive.
So, this will be handled as far as synchronous ordering is concerned over an
asynchronous communication channel.

So; that means, the sender and the receiver they basically work in a synchronization and
whatever sender is sending receiver will be receiving within that synchronization they
are called synchronous ordering. So, it is also called the instantaneous communication
and this is the modified definition of causality.
(Refer Slide Time: 09:06)

These kind of ordering that is we have defined like non-FIFO order FIFO order causal
and synchronous order they basically follow in hierarchy in the execution class which is
shown over here in the figure at A. So, in this particular figure you can see this kind of
relation holds where the causal order is a proper set of FIFO and synchronous is a proper
set of the causal order and a means non-FIFO a means a synchronous. So, asynchronous
does not follow any order and that is why it is non-FIFO.

So, here we can see from this particular hierarchy that that sync or a synchronous order is
basically a proper set of causal order and causal order is a proper set of FIFO and FIFO
is a proper set of non-FIFO or an asynchronous the examples here we can see that if it is
a sync; that means, sender and receiver they works in synchronization they can be
represented by this particular arrows.

Causal order in a causal order we can see that if these 2 sends they are having the
precedence relation or this send is happened before the other send then as far as when
they when they receive. So, this particular order is also ensured at the delivery time
hence s 1 sorry hence s precedes s’ at the delivery also why because here they are
following the receive order.

In FIFO we can see that the order in which these messages are sent the same order is
preserved at the time of delivery of these messages asynchronous does not follow any
order. So, the order at the time of sending these messages is not preserved at the time of
delivery of the messages. So, this will include a hierarchy of these kinds now underlying
the asynchronous mode where the message passing follows the delivery of the message
having unpredictable delays these particular ordering, how they are going to be insured
we are going to see in the further slides how they are going to be implemented and used
up.

(Refer Slide Time: 12:10)

Now, the group communication processes across a distributed system cooperate to solve
the joint tasks often they need to communicate with each other as a group and therefore,
there needs to be support for the group communication. So, example of such applications
are for example, if there is a railway reservation system which comprises of several
nodes. So, the person outside can directly communicate to these set of system which
represents the railway reservation system which includes a bank and the railways and so
on and this particular system requires a group communication.

Why because this particular client or a user has to communicate only to a set of
processor and this is called a group of processor and this kind of communication is called
a group communication. Group Communications are to be supported by the applications.
A message broadcast is sending up a message to all the members in a distributed system,
the notion of a system can be confined only to those sites processes participating in the
joint application refining the notion of the broadcasting there is a multi casting wherein
the message is sent to a certain subset identified as a group of the processes in the
system.

So, at the other extreme is the unicasting which is familiar point to point message
communication let me tell you that that broadcasting means sending a message to all is
called broadcasting if sending message not to all, but to a subset or to a group then it is
called basically a multi casting now if sending a message to another node then it is point
to point communication and it is called a unicasting.

(Refer Slide Time: 14:32)

Now, network layer or the hardware assist the multicast cannot easily provide application
specific semantics on message delivery order adopt groups to dynamic membership
multicast to arbitrary process set at each send provide multiple fault tolerance semantics.
So, these are basically the few things or a few properties which has to be implemented
why because in the network layer this hardware assists multicast cannot be cannot be
provided.

So, I told you that the closed group that is the source is a part of the group. So, the source
is also a part of the group to which they have to communicate then it is called a closed
group and an open group means if the source is outside the group then it is called the
open group. So, these this particular example which will tell you about the application is
specific semantics on the message delivery order.
For example there are 2 process P 1 and P 2 they are causally related through this
particular message M which is shown over here and there are 3 sides R 1 R 2 R 3 which
basically will maintain the replicas. So, if message M 1 is multicast this is a multicast to
all R 1 R 2 and R 3 after that P 2 multicast another message M 2 for updates on these
replicas then let us see the scenario what happens.

Now, you know that P 1 is sent before P 2 and this particular message m ensures the
precedence relation between P 1 and P 2. So, as far as R 1 is concerned R 1 will receive P
2 first and then P 1. So, it is not following this particular precedence relation it is
violating R 2 is concerned R 2 is receiving P 1 first and then P 2 R 3 is also receiving.

So, this particular update as far as the b is concerned is not basically causal order and
also total order is also violated, total order in the sense which I will explain later on total
order meaning to say that each site will see same ordering. So, here you see that here P 2
followed by P 1 is the order in which the requests are coming for the updates as far as R
2 is concerned R 2 will have another view P 1 followed by P 2.

So, basically it is not having the total order as well. So, causal order is also violated and
total order is also violated another scenario which says that in all other sites P 2 P 2 and
then P 1. So, P 2 is occurring first then P 1 violating causal order is violated here in this
particular scenario we will discuss all these things in great details in this part of the
lecture.

(Refer Slide Time: 18:34)

Raynal-Schiper-Toueg algorithm in 1991 let us discuss this algorithm for the group
communication intuitively it seems logical that message M should carry a log of all other
messages or their identifiers, send causal before M’s send event, and sent to the same
destination M. This log can then be examined to ensure whether it is safe to deliver the
message all the algorithms aim to reduce this log overhead and space and time overhead
of maintaining log information at the processes.

So, using this algorithm the ordering which is required in causal order or the total order
can be ensured before the delivery like the messages are arrived which are not following
causal order then they have to be buffered and applied this particular Raynal-Schiper
Toueg algorithm before delivery they have to be buffered and ordered the way they want
to be delivered and that is why each process carries the log each message carries the log
of other messages.

So, the message becomes heavy why because it contains all this information of ordering
and when this message reaches to a site it will be buffered and based on the information
which is contained in the log the destination will try to order.

(Refer Slide Time: 20:27)

The messages in the buffer itself and then accordingly it will be delivered as per the
order. So, let us see this particular algorithm which ensures the delivery this algorithm
assumes the FIFO channels. And there are 2 properties safety and liveness safety
property is ensured here in 2 way that is deliver message m which is received at the Pi
where this particular condition ensures that the all previously sent messages are to be
delivered before this message will deliver otherwise it will be stored in the buffer itself.
Now this particular algorithm you can see that it requires the message space in the form
of a log and also the local space because these data structures array of sent array of
delivered they are to be maintained.

So, basically different techniques we will see that how they are going to reduce it. So,
that the overhead gets reduced and the algorithm becomes efficient and affordable to be
implemented.

(Refer Slide Time: 21:51)

Now, we are going to see the total order total order requires that all the messages
received in the same order by the recipients of the messages is formally defined as for
each pair of processors Pi and Pj and for each pair of messages that are delivered to both
the processes Pi and Pi both the processors Pi is delivered Px before Mx before my if and
only if Pj is delivered Mx before My.
(Refer Slide Time: 22:46)

So, this is called total order. So, this particular figure I have already explained that this
does not satisfy the total order, but this is satisfying the total order why because every
process will have the same view and if every process is having same view of the message
delivery then it is called the total order. How the total order total ordering of the
messages is implemented. So, we are going to see the first algorithm which is called a
centralized algorithm or total ordering of messages.

So, if Pi want to multicast a message M to a group G then it will send the Pi will send to
the group coordinator and the group coordinator in turn will send it to the group of
messages. So, this is called coordinator and hence it is called a centralized algorithm
meaning to say that considering these particular channels are FIFO. So, the order in
which all the messages are being received at the coordinator and when the coordinator
will send to the group members the messages will be ordered because they are assumed
as a FIFO.

So, assuming all the process broadcast messages the centralized solution source in this
algorithm and forces the order in the system with a FIFO channels I explained you each
process sends a message it wants to broadcast to a centralized process which simply
release all the message it receives to every other process over FIFO channel in the group.
(Refer Slide Time: 24:37)

So, it is a straightforward to see that their total order is satisfied why because the order in
which the messages are given by the coordinator the same order it will be delivered
hence this particular since it is a total order it also satisfies the casual order why because
we have seen in the hierarchy class of this model complexity of this algorithm is that
each transmission takes 2 message hops and exactly n messages in the system of n
processes the Drawback is the centralized system has a single point of failure and also
there will be condition towards that.

(Refer Slide Time: 24:57)

So, another algorithm is called three-phase algorithm three-phase algorithm also ensures
the total order. So, if it is ensuring total order; that means, causal order is also ensured as
far as the hierarchy of these classes are concerned. So, a distributed algorithm that
enforces total order and causal order for closed groups is given by three-phase algorithm.
So, the three-phases of the algorithms are first described from viewpoint of the sender
and then from the viewpoint of receiver.

So, this algorithm has 2 parts one is the sender part of the algorithm the other is the
receiver part of the algorithm. So, first we will see the all three phases of the sender and
in the receiver also we will see those corresponding three phases to the sender. So, sender
phase 1 in the first phase a process multicasts which is given in line number 1 of the
algorithm that we will see multicast message them with a locally unique tag and a local
timestamp to the group members.

Phase 2 in the second phase the sender process awaits a reply from all the group member
who respond with their tentative proposal for a revised timestamp for that message m the
await call in line 1 d is non-blocking that is any other message messages received in a
meanwhile are processed once all expected replies are received, the process computes the
maximum of the proposed timestamp for M, and uses maximum as the final timestamp.

Third phase is that in the third phase the process multicasts the final timestamp to the
groups in the line 1 f. So, the sender will send 3 times in the first phase 1 it will send the
message 1 with the timestamp, second time it will send the sender waits awaits the
replies from all then it will send a tentative proposal for the revised timestamp and third
time it will send the messages with the final timestamp and on the other side receiver
will receive them and process them and replies them in all 3 phases that we are going to
see here.
(Refer Slide Time: 27:42)

So, in phase 1 the receiver receives the message with a tentative proposal time stamp
which is sent by the sender it updates the variable priority that tracks the highest
proposed time stamp and then realizes the proposed time stamp to the priority and places
the message on it is tag and revises the time stamp at the tail of the Q in the Q the entry
is marked as undeliverable.

So, meaning to say that after receiving the messages from the sender it basically finalized
a tentative proposal time stamp and accordingly the messages are placed in the in the
queues and they are marked as undeliverable they will not be delivered why because they
are stored in the Q and according timestamp they are being they will be ordered and then
delivered.
(Refer Slide Time: 28:46)

So, to do that it will go for a second phase the receiver sends the revised timestamp back
to the sender the receiver then waits in a non-blocking manner for the final in phase 3 the
final timestamp is received from the from the multicaster or from the sender the
corresponding message entry in a Q is identified using the tag and marked as deliverable
after the revised timestamp is over written by the final timestamp the Q is then restored
using timestamp field of the entries.

So, as the Q is already sorted except for the modified entry for the message under
construction that entry has to be placed into a sorted position if the message entry is at
the head of the Q that entry and all the corresponding subsequent entries are also marked
as deliverable are dequeued from the Q and delivered to that particular process.
(Refer Slide Time: 29:32)

So, this particular code I have already explained and let us go through the illustrative
example for the working or the execution of that 3 phase algorithm.

(Refer Slide Time: 29:37)

Now, here you can see that side A and side B they want to send a message to those 2
replicas C and D. So, a will send with a tag 7 the message to D and B will send with a
tag 9 to C. Now D when it receives this particular message with a with a tag 7 it will it
will give it is own timestamp and position it in the Q that is called temporary Q it will
not be delivered and similarly first message D will receive the first message from B, it
will be queued with the timestamp and the next message 7 minute delivered.

So, time this time will be incremented and will be given to this particular message at
time ten. So, on the head of the Q will be the message which comes from B, it will not be
delivered it will be just queued this particular timing which is basically the revised time
or a proposed time will be sent back from a to B in the second phase. Similarly the 9 tag
or a timestamp 9 is sent from D to B. Similarly here 7 will be received first, it will be n
queued on the top of the Q and 7 will be returned back to A and once 9 will be received.
So, 9 will be in the order, 9 stamp will be returned back.

(Refer Slide Time: 31:34)

Now, it will process them, after receiving these values as 7 and 10 it will take the
maximum over here and this will be the final value and a will send the final value 10 to
both of them similarly here after receiving 7 and 9 2 values the maximum is 9 and 9 will
be sent back to both of them. So, if you see here in this scenario mine is at the head of
the queue. So, from temporary it will be put on the delivery of the queue and this
message will be delivered to the process.

Here also the same ordering will be done since 9 is at the head of the queue it will be put
on the delivery and 9 will be delivered. So, first 9 will be delivered at both the ends and
then 10 will be delivered. So, hence it follows the total order according to the algorithm.
(Refer Slide Time: 32:36)

So, all these steps of the algorithms are basically I have explained you through the
example the Complexity of this algorithm since it uses 3 phases and every phases a
message is being exchanged.

So, in every phase n minus one messages are exchanged. So, out of 3 phases 3(n-1)
message will be exchanged delay will be 3 message hops because 3 times the message
will go and come back. So, it will be 3 message hop delay will be there it also
implements casual order that we have already seen. So, 3 phase algorithm is closely
structured along the lines of Lamport’s clock for the mutual exclusion that algorithm we
have studied.

So, Lamport’s mutual exclusion algorithm has the property that when a process is at the
head of it is own queue and as it received a reply from all the process requests of that
process the head of all the queues this can be exploited to deliver the message by all the
process in the same order instead of entering into a critical section. So, exactly on the
same lines this particular algorithm was designed and is structured.
(Refer Slide Time: 33:33)

So, we have seen the algorithm for total order and that will be achieved through the
through the algorithm why because underlying network is an asynchronous network
which we are now assuming. So, the ordering which is required from the application is to
be ensured through that algorithm which we have just covered.

Now the next topic is about the multi casting. So, as I told you that unicasting multi
casting and broadcasting there are 3 different ways of communication in a for the
applications in a distributed system which is being utilized.

So, now we are talking about the multi casting that is the communication to a to a group
of processes. So, 4 classes of source destination relations for the open groups are there.
So, that previous algorithm was for closed group here this example is basically or this
particular nomenclature is for the open groups. So, the first nomenclature for the open
group is SSSG that is single source and single destination group where there is a single
source and the single group is for the destination that is called SSSG.

MSSG is basically multiple sources here you see there are multiple sources and a single
destination called multiple sources single destination M SSG then comes SSSMG means
single source and multiple groups. So, here these multiple groups are overlapping also
and then MSMG that is multiple sources here you see 2 sources are there and multiple
groups which are overlapping also. So, there are 4 different ways in which these open
groups can be organized or can be classified.
Now we can see that before we go ahead let us see that it is quite easy to implement
SSSG why because this can be done using the centralized approach. So, this will be a
coordinator for example, this will send to a coordinator and coordinator will basically
sequence the messages in the order. So, this is quite trivial this also can be handled using
the centralized approach and MSSG can also be handled through that approach why
because these 2 centers are there and they can be converted into a single source single
group communication why because these 2 sources can send to the coordinator and
coordinator intern will send to one group that also can be handled, multiple sources and
multiple groups is very difficult it cannot be straightaway implemented in triple SG way.

(Refer Slide Time: 37:17)

So, now we are going to see the implementation for multiple source and multiple group.
One way to implement multiple source and multiple groups is through the propagation
trees. So, propagation trees for multi casting are going to basically solve the purpose.
(Refer Slide Time: 37:46)

So, let us see how the propagation trees are constructed if this is the set of nodes. So, we
are forming the Meta-group out of these groups and these Meta-groups can be organized
in the form of a tree and this is called a propagation tree.

So, in this example illustrating the proposition tree Meta-groups are shown in the bold
phases these are all Meta-groups. So, you can see that (ABC) is a primary Meta-group.
So, primary Meta-groups will basically have group (AB), (AC), (A) then (B) and then
(C) and then (BC) also and (BCD) also. So, this particular ABC will become the primary
Meta-group and becomes the root of all those Meta-groups.

Now, in turn (BCD) again will become the primary Meta-group for these Meta-groups
means (BCD) is here. So, (BCD) will have (BD) then (CD) and (BC) is also absorbed
already absorbed in (ABC) it will not be reflected over here and (D) and (DE). Now as
far as the (D) will become the primary Meta-group for € so the Meta-group for the
primary Meta-group (DE) will be (E) then (CE) and (EF) now (EF) will become the
primary Meta group for (F).

So, just see that these complete groups of nodes they are organized in a form of the
propagation tree implementation of a propagation tree will now becoming can be easily
handled. So, basically in this once the propagation tree is there and you want to send the
messages to a particular or in a multicast way. So, first messages are sent to the primary
Meta-group node and that in turn will communicate to it is Meta-groups and if it is has to
reach to a Meta group (F). So, it has to traverse through all the primary Meta-groups till
that meta-group which contains that primary meta-group which contains that meta-group
reaches to that point there it will be delivered in that particular tree fashion.

(Refer Slide Time: 40:42)

Now, there are various classification of application level multicast algorithms. So, first
there are 4 different class of multi class algorithm first one is called privilege based
algorithm in privilege based algorithm the token rotates. So, the node which is having the
token will be allowed to send to the destination. So, the process delivers message in the
order of the sequence numbers typically the closed groups and casual order and total
order will be there are 2 algorithms which are implemented based on this particular
classification to totem and on demand multicast algorithms are available.
(Refer Slide Time: 41:33)

The next one is called moving sequencer. So, here there will be a sequencer nodes are
sending the message to the sequencers. So, sequencers token has the sequence numbers
and list of the messages for which the sequence number has been assigned on receiving
the token the sequencer assigns the sequence number to the received, but unsequenced
messages and sends the newly sequenced messages to the destination.

Another next one is called a fixed sequencer fixed sequencer is like centralized approach
that we have already seen centralized approach means there is a one single coordinator
and fifth one is called destination agreement. So, in destination agreement the ordering
will be done among the destination. So, destination received the limited ordering
information and using time based Lamports 3 phase algorithm the agreement will be
evolved.
(Refer Slide Time: 42:44)

Few other algorithms for message ordering and group communication are available in
the literatures that are listed over here Conclusion.

(Refer Slide Time: 42:53)

Inter process communication via message passing is at the core of any distributed system
in this lecture we have discussed non-FIFO, FIFO causal order, asynchronous order,
synchronous communication paradigms for message ordering, then examine several
algorithms to implement these orderings group communication is the important aspect of
communication in a distributed system.
Causal order and total order are the popular forms of ordering when doing the group
multi casting and broadcasting then we have explained the propagation trees for
multicast algorithm and classification of application level multicast algorithms in the
upcoming lectures we will discuss about self stabilization.

Thank you.