Message Passing

Communication Primitives

 For the purposes of developing communication

algorithms, we divide distributed systems into
two categories:
◦ Synchronous Systems
◦ Asynchronous Systems
 Synchronous Systems
• The key property of such systems is that all
communication and processing takes bounded
time.

• Assuming a reliable communication medium and

a crash failure model for processors, a sender
can know for sure that a receiver has failed if the
sender didn't get an acknowledgement from the
receiver within some finite time
 Asynchronous Systems
• The key property of such systems is that
communication and processing delays can be
arbitrarily long even in the absence of failures.
• Hence, there is no way we can distinguish
between an arbitrarily slow processor, and one
that has failed.
• Implementing group communication semantics in
the presence of failures becomes more
challenging in the asynchronous model.
 Introduction
 A process is a program in execution.
 Each computer of a distributed system may have a
resource manager process to monitor the current status
of usage of its local resources.
 Resource managers of all the computers might
communicate with each other from time to time to
dynamically balance the system load among all the
computers.
◦ A distributed operating system needs to provide interprocess
communication (IPC) mechanisms to facilitate such
communication activities.
 Message Passing

 Two process communicate by IPC. It requires

information sharing either by:

◦ Original Sharing or Shared Data Approach

◦ Copy sharing or Message Passing Approach

 The shared-data approach

Shared
P1 Memory P2
Area

The information to be shared is placed in a common

memory area that is accessible to all the processes
involved in an IPC.
Message-Passing approach

P1 P2

Data to be shared is physically copied from

sender process address space to receiver by
transmitted data.
 Message Passing System

 Provides a set of message based IPC protocol.

 Shields details of complex network protocol.
 Shields details of multiple heterogeneous platform.
 Allows programs to be written using simple
communication primitives.
 It provides infra structure to build other higher level IPC
system such as RPC and DSM.
 Desirable Features of MPS
 Simplicity
 Uniform semantics
 Efficiency
 Reliability
 Correctness
 Flexibility
 Security
 Portability
Issues in IPC by message passing

 Sender & Receiver

 Acceptance of message by receiver.
 Reply of the message
 Failures during communication
 Buffering by the receiver
 Size of the buffer
 Order of the outstanding messages
Synchronization
 Synchronization primitives:
◦ BLOCKING PRIMITIVE (SYNCHRONOUS)
 Its invocation blocks execution of invoker.
◦ NON-BLOCKING (ASYNCHRONOUS)
 It does not block execution.

 Other primitives:
◦ Send primitive
◦ Receive primitive
 Cont…
 Blocking send primitive:
◦ After execution of the send statement, the sending
process is blocked until it receives an
acknowledgement.

 Non-blocking send primitive:

◦ After execution of the send statement, the sending
process is allowed to proceed with its execution as soon
as the message has been copied to a buffer.
 Cont…
 Blocking receive primitive:
◦ After execution of the receive statement, the receiving
process is blocked until it receives a message.

 Non-blocking receive primitive:

◦ The receiving process proceeds with its execution after
execution of the receive statement, which returns
control almost immediately.
 Blocking send and receive primitives
Sender’s Receiver’s
execution execution

Receive (message);
Execution suspended
Send (message); Message
Execution suspended
Execution resumed

Execution resumed Send (ack)

acknowledgment

Blocked state
executing state
 Issues In Non-blocking

 How the receiving process knows that the message

has arrived in the message buffer?

 Done by Polling or Interrupt

◦ Polling: Receiver uses “test” primitive to allow it
to check buffer status.
◦ Interrupt: When message has arrived in the buffer,
a s/w interrupt is used to notify the receiving
process.
 Issues In Blocking Send

 Sending process could get blocked forever if

receiving process crashes or message lost in
network.

 Hence it should use timeout values.

 It could be a parameter of “send” primitive.

 Implementation Ease

• Synchronous communication is easy to

implement.
• If message gets lost or is undelivered, no
backward error recovery is required.
• Synchronous communication limits concurrency.
• Subject to communication deadlock.
Buffering
 Message are transmitted from one process to another by
copying the body of the message from the address space of
sending process to address space of receiver process
(possibly via address space of kernels of sending and
receiving computers).

 In some cases, the receiver process may not be ready to

receive a message but may want O.S. to save messages for
later reception.
Cont…

 Message buffering strategy related to

synchronization strategy

◦ Synchronous Mode: Null /No Buffer

◦ Asynchronous Mode: Buffer with unbounded capacity

 Null Buffering or no buffering

It has two strategies.

1st strategy
– Message remains in SPAS (Sender Process
Address Space) and the execution of send is
delayed until the receiver executes the
corresponding receive.

– After send, when ACK is received, it executes

“send” again to send the message.
 Cont…

2nd strategy
– The message is simply discarded and the time
out mechanism is used to resend message
after a time out period.

– After executing send, sender process wait for

an ACK. After time out, sender retries
executing send
Cont…
Sending Receiving
process process

Message

Message transfer in synchronous send with no buffering

strategy (only one copy operation is needed).

23
 Cont..
 The null buffer strategy is not suitable for
synchronous communication between two
processes in a distributed system.

◦ If the receiver is not ready, a message has to be

transferred two or more times and the receiver of the
message has to wait.
 Single-message buffer
 A buffer having a capacity to store a single
message is used on the receiver’s node.

 Used for synchronous communication.

◦ An application module may have at most one message
outstanding at a time.
 Cont…
 This strategy keeps the message ready for use
at the location of the receiver.
◦ Buffer is used if receiver is not ready to receive the
message.

 The buffer may be located in the kernel’s

address space or in the receiver process address
space.
Cont…

Sending Receiving
process process

Message

Single Message Buffer

Node
boundary

Message transfer in synchronous send with single

buffering strategy (two copy operation is needed).

27
 Unbounded-capacity buffer
 Buffer with unbounded-capacity message may be
used in the asynchronous mode of
communication.
 It can store all unreceived messages with the
assurance that all messages sent to the receiver
will be delivered.
 This strategy is practically impossible.
 Finite Bound Buffer
 Asynchronous mode of communication uses
this strategy of finite bound buffer.

 This strategy may lead to the problem of

possible buffer overflow.

29
 Cont…
 Methods to deal with the problem of buffer
overflow:
◦ Unsuccessful communication
 Message transfers simply fail whenever there is no more
buffer space.

◦ Flow-controlled communication
 The sender is blocked until the receiver accepts some
messages, thus creating space in the buffer for new
messages.

30
 Cont…
 A create-buffer system call is provided to the user.

 This system call when executed by a receiver

process creates a buffer of a size specified by
receiver either in kernel AS or receiver process
AS.

31
Cont…
Sending Receiving
process Message 1 process
Message 2
Message Message 3

Message n

Multiple-message
Buffer/mailbox/port

Message transfer in asynchronous send with multiple-message

buffering strategy (two copy operations are needed).

32
 Multidatagram messages
 A message whose size is greater than the MTU
(maximum transfer unit) is fragmented into
multiples of the MTU.

 Each fragment is sent separately.

 Responsibilities of MPS includes:

◦ The disassembling of a multidatagram message into
multiple packets on the sender side,
◦ The reassembling of the packets on the receiver side.
 Encoding & Decoding of Message Data

 Structure of program/object should be preserved

while they are being transmitted from address space
of sending process to receiving process’s address
space.
Difficult to achieve because:
 An absolute pointer losses its meaning when
transferred from one process AS to another.
◦ Requires flattening and shaping of objects.
 Due to above problems, program object are not
transferred in their original form. They are first
converted to stream form by encoding

34
 Process Addressing

 Processing addressing is naming of parties

involved in interaction.

 Two types of process Addressing:

1. Explicit Addressing
2. Implicit Addressing

35
 Explicit Addressing
 The process with which communication is desired
is explicitly named as a parameter in the
communication primitive used.

 Primitives for explicit process addressing are

◦ Send ( process-id, message )
◦ Receive(process-id, message )

36
 Implicit Addressing

 A process willing to communicate does not explicitly

name a process for communication.
 Primitives for implicit addressing are:
◦ Send_any (service_id, message)
Send a message to any process that provides the
service of type “service_id”
◦ Receive_any (process_id, message)
Receive a message from any process and returns the
“process_id” of the process from which the message
was received.
 Functional addressing
 The address used in the communication
primitive identifies a service rather than a
process.

 The client is not concerned with which particular

server out of a set of servers providing the
service desired by the client actually services its
request.
 Process addressing
 Simple method to identify a process is by a combination of
machine-id and local-id such as machine-id@ local-id.

 Local-id can be process-id or port-id, that uniquely

identifies a process on a machine.

 Machine-id is used by sending machine kernel to send the

message to the receiver process machine.

 Eg: Berkely UNIX

◦ 32 bit Internet address for machine-id
◦ 16 bit for local-id

39
 Cont…
 Benefit:
 No global coordination required for local-id
 Drawback
◦ It does not allow a process to migrate from one machine
to another on requirement like
 One or more processes of a heavily loaded machine may be
migrated to a lightly loaded machine to balance the overall
system load.
 Cont…
 Solution
◦ Processes can be identified by a combination of the
three fields
 machine_id
 Local_id
 machine_id
 Cont…
Machine-id , Local-id , Machine-id ,

Node on which
process created Id of the process Last known Node
Location of the process

Never Change This may

 This type of adding is known as link based process

addressing.
 During Migration, a link Information (p-id + m/c id of
new node) is left on previous node.

42
 Cont…
 Drawbacks in this:
◦ The overload of locating a process may be large if the
process has migrated several times during its lifetime.

◦ It may not be possible to locate a process if an

intermediate node on which the process once resided
during its lifetime is down.
 Cont…
 Both process-addressing methods are
nontransparent due to the need to specify the
machine identifier.

 Solutions are:
– Ensure that the system wide unique identifier of a
process does not contain an embedded machine
identifier.
– Use Two-level naming scheme for processes.
 Two Level Naming Scheme

 Each process has two id:

◦ A high level name that is m/c independent (ASCII
string).
◦ A low level name that is machine dependent
(machine_id@local_id).

 A name server is used to maintain a mapping

table.

45
 Cont…
 When a process wants to send a message to another
process, it specifies high level name of the receiver
process.

 The kernel of the machine first contacts the name server

to get low level name.

 The name server can also be used for functional

addressing.
◦ High level name identifies a service instead of a
process.
◦ The name server maps a service identifier to one or
more processes that provide that service.
 Cont…
 Name server
◦ Drawbacks – poor reliability and poor scalability.
◦ Solution – replicate the name server.

 Extra overhead in replicating name server.

 Failure Handling
Failure can be due to:
1. Loss of Request message
2. Loss of Response Message
3. Unsuccessful execution of request
• Due to the receiver’s node crashing while the
request is being processed.

48
Cont…

sender Receiver

Send request
Request message

lost

Request message is lost.

Cont…

sender Receiver

Send request Request message

Successful request
execution
Send response

lost

Response message is lost.

Cont…

sender Receiver

Send request Request message

unsuccessful request
crash execution

restarted

Receiver’s PC crashed.
 Cont…
To overcome these problems
• A reliable IPC protocol of a message-passing system is
designed.
• It is based on the ideas of internal retransmission of
messages after time out and
• Return of an ACK to sending m/c kernel by receiver
m/c kernel.

52
 Cont…
 Protocols used for client-server communication
between two processes are:
◦ Four-message reliable IPC protocol
◦ Three -message reliable IPC protocol
◦ Two-message reliable IPC protocol
 Four-message reliable IPC protocol

 The time for which the sender waits is slightly

more than the approximate round trip time +
the average time required for executing the
request.

54
Four-message reliable IPC protocol
Sender’s Receiver’s
execution execution

request

acknowledgment

reply

Blocked state
acknowledgment
executing state
Three -message reliable IPC
protocol
The result of the processed request is sufficient
acknowledgment that the request message was
received by the server.
Three-message reliable IPC protocol
Sender’s Receiver’s
execution execution

request

reply

Blocked state
acknowledgment
executing state
 Cont…
Problem is with timeout value.
• If the request message is lost, it will be
retransmitted only after the timeout period,
which might have been set to a larger value.

• If the timeout value is not set properly taking

into consideration the long time needed for
request processing, unneccessary
retransmissions will take place.
Two Message Reliable IPC Protocol
 When request is received at servers machine, it
starts a timer.
 If server finishes processing the req. before time
expires, reply acts as ACK
 Else a separate ACK is sent by kernel.
Two-message reliable IPC
Sender’s Receiver’s
execution execution

request

reply

Blocked state
executing state
 Fault-tolerant communication
client server
Send Request message
request
timeout
lost
Send
request Retransmit
timeout Request message

Unsuccessful Request execution

Send crash
request Retransmit
Request message
timeout Successful Request execution
Send response
These two successful
lost executions of the same
Send request may produce
request Retransmit different results.
Request message Successful Request execution
Send response

Response message
 Idempotency
 It means repeatibility.
 An Idempotent operation produces the same
result without any side effects, no matter how
many times it is performed with the same
argument.
 Example: using procedure GetSqrt(64) for
calculating the square root of a given number.
 ISSUE : Duplicate Requests.
◦ If the execution of the request is nonidempotent, then
its repeated execution will destroy the consistency of
information.
Handling of Duplicate Requests
 If client makes a request.
 Server processes the request.
 Client doesn't receive the response.
 After time out, again issues REQ.

 What Happens?
Cont…
 Solution
◦ Use unique id for every REQ
◦ maintains a reply cache in the kernel’s address space
on the server machine to cache replies.

 The use of a reply cache does not make a

nonidempotent routine idempotent.
 Multidatagram Messages
 A complete message transmission is:
◦ When all the packets of the message have been received
by the process to which it is sent.

◦ For this, reliable delivery of every packet is important.

 Cont…

 Use of Blast Protocol

◦ Requires Single ACK for all packets of multidatagram
message.
◦ Two fields in each packet – total no. of packets and seq
no. of packet.
◦ After timeout, it follows Selective Repeat.

 Use of Stop and Wait Protocol

◦ Requires ACK for each packet.
 Group Communication

 A group is a set of parties that, presumably, want to

exchange information in a reliable, consistent manner.

 Group communication is a paradigm for multi-party

communication that is based on the notion of groups as a
main abstraction.
Cont…
 The set of replicas of a fault-tolerant database
server may constitute a group.
 Consider update messages to the server where the
contents of the database depends on the history of
all update messages received, all updates must be
delivered to all replicas.
 Furthermore, all updates must be delivered in the
same order, otherwise, inconsistencies may arise.
 Cont…
Following three types of group communication are
popular:
◦ One to Many
 Single sender and multiple receiver
 multicast communication
◦ Many to One
 Multiple senders and single receiver
◦ Many to Many
 Multiple senders and multiple receivers
 Group management
 A message-passing system with group
communication facility provides the flexibility:
◦ to create and delete groups dynamically, &
◦ to allow a process to join or leave a group at any time.

 It uses centralized group server to manage the

groups.

 Replication of group servers is required.

 Group addressing
 A two-level naming scheme is used for group
addressing.

 The high-level group name is an ASCII string that

is independent of the location information of the
processes in the group.

 The low-level group name depends to a large

extent on the underlying hardware.
◦ Like a multicast address for a group.
Message Delivery to Receiver Process

 User applications use high level group names in

programs.
 Centralized group server (GS) maintains a mapping of
high-level group names to their low level names.
 Group server also maintains a list of process identifier
of all the processes for each group.
 Kernel contacts the group server to obtain low level name
& process_id of processes belonging to that group.
 This list of process_id is inserted in the message.
Buffered / Unbuffered Multicast

 Multicasting is an Asynchronous
communication mechanism.

 So which one to use

◦ BUFFERED or UNBUFFERED?
Reliability in Multicasting
 Depends on degree of reliability required.
 Sender of a multicast message can specify the
number of receivers from which a response
message is needed.
 This is expressed in the following form:
◦ 0-reliable
◦ 1-reliable
◦ ‘m’ out of ‘n’ reliable
◦ all reliable
 Atomic Multicast

 This has all or nothing property i.e. when a message

is sent to a group, either all or none receive it.
 Only “all-reliable” kind of reliability needs this strict
paradigm.
 A flexible message-passing system should support
both atomic and nonatomic multicast facilities.
 Many to Many Communication
 One to many and many to one are implicit in this
scheme.
 Issue: ordered message delivery.
 Ordered message delivery
◦ Ensures that all messages are delivered to all receivers
in an order acceptable to the application.
◦ Requires message sequencing.
 Cont…
 Semantics of ordered delivery are:
• Absolute ordering
• Consistent Ordering
• Causal Ordering
 Absolute Ordering
 This semantics ensures that all messages are
delivered to all receiver processes in the exact
order in which they were sent.
 It uses global timestamps /synchronized clocks as
message identifiers (not easy to implement).
 Uses a sliding-window mechanism to periodically
deliver the message from the queue to the
receiver.
Cont…

S1 R1 R2 S2

t1
m1 Time
t2
t1 < t2
m2
m1
m2
 Consistent Ordering
 This semantics ensures that all messages are
delivered to all receiver processes in the same
order.

 This order may be different from the order in

which messages were sent.
 Cont…
 Method to implement this semantics
◦ Make the many-to-many scheme appear as a
combination of many-to-one and one-to-many
schemes.

◦ The kernels of the sending machines send messages

to a single receiver (known as sequencer) that
assigns a sequence number to each message and
then multicasts it.
Cont…
S1 R1 R2 S2

t1
Time
t2
m2 m2 t1 < t2

m1 m1
Causal Ordering
• PCR
 CBCAST Protocol
• For Causal Ordering – USED in ISIS, a real commercial
distributed system, based on process groups.
• ISIS project has moved from Cornell University to ISIS
Distributed System a subsidiary of Stratus Computer Inc.
• Each member process of a group maintains a vector of
“n” components, where “n” is the total number of
members in the group.
• Each member is assigned a sequence number from 0 to
n, and the ith component of the vector corresponds to
the member with sequence number i’.
• In particular, the value of the ith component of a
member’s vector is equal to the number of the last
message received in sequence by this member from
member ‘i’.
Cont…
• To send a message, a process increments the value of
its own component in its own vector and sends the
vector as part of the message.

• When the message arrives at a receiver process’s site, it

is buffered by the runtime system.

• The runtime system tests the two conditions to decide

whether the message can be delivered to the user
process or its delivery must be delayed to ensure causal
ordering semantics.
Cont…
• Let
• “S”= vector of sender process that is
attached to the message
• “R” = vector of receiver process
• “i” = sequence number of sender process

• Two conditions to be tested are:

• S[i] = R[i] +1 and
• S[j] <= R[j] for all j != i
 Contd…
• First condition ensures that the receiver has not
missed any message from the sender.
• It ensures that the two messages from the same
sender are always causally related.
• Second condition ensures that the sender has
not received any message that the receiver has
not yet received.
• It make sure that the senders’ message is not
causally related to the message missed by the
receiver.
cbcast
 S[i] = R[i] +1 and S[j] <= R[j] for all j != i