Chapter 2 - Communication

Introduction
 interprocess communication is at the heart of all distributed
systems
 communication in distributed systems is based on message
passing as offered by the underlying network as opposed to
using shared memory
 modern distributed systems consist of thousands of
processes scattered across an unreliable network such as
the Internet
 unless the primitive communication facilities of the network
are replaced by more advanced ones, development of large
scale Distributed Systems becomes extremely difficult

2
 Objectives of the Chapter
 review of how processes communicate in a network (the
rules or the protocols) and their structures
 introduce the four widely used communication models for
distributed systems:
 Remote Procedure Call (RPC)
 Remote Method Invocation (RMI)
 Message-Oriented Middleware (MOM)
 Streams

3
 2.1 Layered Protocols
 two computers, possibly from different manufacturers, must
be able to talk to each other
 for such a communication, there has to be a standard
 The ISO OSI (Open Systems Interconnection) Reference
Model is one of such standards - 7 layers
 TCP/IP protocol suite is the other; has 4 or 5 layers
 OSI
 Open – to connect open systems or systems that are open
for communication with other open systems using standard
rules that govern the format, contents, and meaning of the
messages sent and received
 these rules are called protocols
 two types of protocols: connection-oriented and
connectionless

4
layers, interfaces, and protocols in the OSI model
5
 Physical Layer
 Definition: The physical layer is the lowest layer of the OSI
model and deals with the transmission of raw binary data over a
physical medium.
 It defines the electrical, mechanical, procedural aspects of
transmission, such as voltage levels, cable types, and data rates.
 Functionality: Its primary function is to transmit bits over a
communication channel reliably.
 It handles tasks such as signal encoding, modulation, and
transmission media.
 Example: Ethernet, Wi-Fi, Fiber optic cables.

6
 Data Link Layer:
 Definition: The data link layer is responsible for reliable data
delivery across the link between two directly connected nodes.
 It provides error detection and correction, as well as flow control
to ensure data integrity during transmission.
 Functionality: It manages the link between devices, establishes
and terminates connections, and handles framing, error detection,
and flow control.
 Example: Ethernet (MAC layer), Point-to-Point Protocol (PPP).

7
 Network Layer:

8
 Transport Layer:

9
 Session Layer:

10
 Media (lower) Layers
 Physical: Physical characteristics of the media
 Data Link: Reliable data delivery across the link
 Network: Managing connections across the network
or routing
 Transport: End-to-end connection and reliability
(handles
lost packets); TCP (connection-oriented),
UDP (connectionless), etc.
 Session: Managing sessions between applications
(dialog control and synchronization); rarely
supported
 Presentation: Data presentation to applications; concerned
with the syntax and semantics of the
information transmitted
 Application: Network services to applications; contains
protocols that are commonly needed by
users;
Host (upper) FTP, HTTP, SMTP, ...
Layers
11
a typical message as it appears on the network

12
 a conversation occurs between a sender and a receiver at each
layer
 e.g., at the data link layer

discussion between a receiver and a sender in the data link layer

13
 Transport Protocols: Client-Server TCP

assuming no messages are lost,

 the client initiates a setup
connection using a three-way
handshake (1-3)
 the client sends its request (4)
 it then sends a message to close
the connection (5)
 the server acknowledges receipt
and informs the client that the
connection will be closed down (6)
 then sends the answer (7)
followed by a request to close the
connection (8)
 the client responds with an ack to
finish conversation (9)

normal operation of TCP

w114
 much of the overhead in TCP is for managing the connection
 combine connection setup with request
and closing connection with answer
 such protocol is called TCP for
Transactions (T/TCP)

 the client sends a single message

consisting of a setup request,
service request, and information to
the server that the connection will be
closed down immediately after
receiving the answer (1)
 the server sends acceptance of
connection request, the answer, and
a connection release (2)
 the client acknowledges tear down
of the connection (3)

transactional TCP
15
 2.2 Remote Procedure Call
 the first distributed systems were based on explicit message
exchange between processes through the use of explicit
send and receive procedures; but do not allow access
transparency
 in 1984, Birrel and Nelson introduced a different way of
handling communication: RPC
 it allows a program to call a procedure located on another
machine
 simple and elegant, but there are implementation problems
 the calling and called procedures run in different address
spaces
 parameters and results have to be exchanged; what if the
machines are not identical?
 what happens if both machines crash?

16
 Conventional Procedure Call, i.e., on a single machine
 e.g. count = read (fd, buf, bytes); a C like statement, where
fd is an integer indicating a file
buf is an array of characters into which data are read
bytes is the number of bytes to be read
Stack pointer

Stack pointer

the stack while the called

parameter passing in a local procedure
procedure is active
call: the stack before the call to read
 parameters can be call-by-value (fd and bytes) or call-by reference (buf) or in
some languages call-by-copy/restore 17
 Client and Server Stubs
 RPC would like to make a remote procedure call look the
same as a local one; it should be transparent, i.e., the calling
procedure should not know that the called procedure is
executing on a different machine or vice versa

principle of RPC between a client and server program

 when a program is compiled, it uses different versions of
library functions called client stubs
 a server stub is the server-side equivalent of a client stub
18
 Steps of a Remote Procedure Call
1. Client procedure calls client stub in the normal way
2. Client stub builds a message and calls the local OS (packing
parameters into a message is called parameter marshaling)
3. Client's OS sends the message to the remote OS
4. Remote OS gives the message to the server stub
5. Server stub unpacks the parameters and calls the server
6. Server does the work and returns the result to the stub
7. Server stub packs it in a message and calls the local OS
8. Server's OS sends the message to the client's OS
9. Client's OS gives the message to the client stub
10. Stub unpacks the result and returns to client
 hence, for the client remote services are accessed by making
ordinary (local) procedure calls; not by calling send and receive

 server machine vs server process; client machine vs client process

19
 2.2.2 Parameter Passing
1. Passing Value Parameters
 e.g., consider a remote procedure add(i, j), where i and j are
integer parameters

steps involved in doing remote computation through RPC

20
2. Passing Reference Parameters
 assume the parameter is a pointer to an array
 copy the array into the message and send it to the server
 the server stub can then call the server with a pointer to this
array
 the server then makes any changes to the array and sends it
back to the client stub which copies it to the client
 this is in effect call-by-copy/restore
 optimization of the method
 one of the copy operations can be eliminated if the stub
knows whether the parameter is input or output to the
server
 if it is an input to the server (e.g., in a call to write), it need
not be copied back
 if it is an output, it need not be sent over in the first place;
only send the size
 the above procedure can handle pointers to simple arrays
and structures, but difficult to generalize it to an arbitrary
data structure
21
2.2.3 Extended RPC Models
 to solve some of the shortcomings of the original model
 no need of network communication if server and client are on
the same machine
 no need of blocking for the client in some cases

a. Doors
 the original RPC model assumes that the caller and the callee
can communicate only by means of passing messages over a
network; what if they are colocated on the same machine?

 a door is a generic name for a procedure in the address

space of a server process that can be called by a process
colocated with the server
 support from the local OS is required

22
1. the server process registers a door before it can be called (door_create)
and a name is attached to it
1. a client calls a door by a system call (door_call) including all parameters
2. results are returned by the system call door_return

the principle of using doors as IPC mechanism

23
 benefit: it allows the use of a single mechanism (procedure
calls) for communication
 disadv: application developers have to be aware of where a
procedure is located; is it
 local within the current process
 local to a different process on the same machine
 a remote process

24
b. Asynchronous RPC
 if there is no need to block the client until it gets a reply
 two cases
1. if there is no result to be returned
 e.g., adding entries in a database, ...
 the server immediately sends an ack promising that it will carryout
the request
 the client can now proceed without blocking

a) the interconnection between client and server in a traditional RPC

b) the interaction using asynchronous RPC 25
2. if the result can be collected later
 e.g., prefetching network addresses of a set of hosts, ...
 the server immediately sends an ack promising that it will
carryout the request
 the client can now proceed without blocking
 the server later sends the result

a client and server interacting through two asynchronous RPCs

26
 the above method combines two asynchronous RPCs and is sometimes
called deferred synchronous RPC

 variants of asynchronous RPC

 let the client continue without waiting even for an ack,
called one-way RPC
 problem: if reliability of communication is not guaranteed

27
2.2.4 DCE (Distributed Computing Environment) RPC
 a middleware and an example RPC system developed by OSF
(Open Software Foundation), now The Open Group
 it is designed to execute as a layer of abstraction between
existing OSs and distributed applications
 the Open Group sells the source code and vendors integrate it
into their systems
 it uses the client-server programming model and
communication is by means of RPCs
 services
 distributed file service: a worldwide file system that provides
a transparent way of accessing files
 directory service: to keep track of the location of all
resources in the system (machines, printers, data,
servers, ...); a process can ask for a resource without
knowing its location
 security service: for protecting resources; access is only
through authorization
28
  distributed time service: to maintain clocks on different
machines synchronized (clock synchronization is covered in
Chapter 5)

 Steps in writing a Client and a Server in DCE RPC

 the system consists of languages, libraries, daemons,
utility programs, ... for writing clients and servers
 IDL (Interface Definition Language) is the interface
language - the glue that holds everything together
 it contains type definitions, constant declarations
and what the procedures do (only their syntax)

29
 Uuidgen generates a prototype IDL file with a globally unique interface identifier
 the IDL file is edited (filling the names of procedures and parameters) and the
IDL compiler is called to generate 3 files
 the application writer writes the client and server codes and are then compiled
and linked together with the stubs
30
 Binding a Client to a Server in DCE RPC

 for a client to call a server, the server must be registered (1 &

2)
 the registration allows the client to locate the server and
bind to it
 the DCE daemon maintains a table (server, endpoint) and the
protocols the server uses
 the directory server maintains the locations of all resources
in the system (machines, servers, data,, ...)
 two steps for server location
 locate the server’s machine (3)
 locate the server process on that machine (which has
what is called an endpoint or port) (4)

31
32
 2.3 Remote Object (Method) Invocation (RMI)
 resulted from object-based technology that has proven its
value in developing nondistributed applications
 it is an expansion of the RPC mechanisms
 it enhances distribution transparency as a consequence of
an object that hides its internal from the outside world by
means of a well-defined interface
 Distributed Objects
 an object encapsulates data, called the state, and the
operations on those data, called methods
 methods are made available through interfaces
 the state of an object can be manipulated only by invoking
methods
 this allows an interface to be placed on one machine while
the object itself resides on another machine; such an
organization is referred to as a distributed object
 the state of an object is not distributed, only the interfaces
are; such objects are also referred to as remote objects 33
 the implementation of an object’s interface is called a proxy (analogous to a client stub
in RPC systems)
 it is loaded into the client’s address space when a client binds to a distributed object

 tasks: a proxy marshals method invocation into messages and unmarshals

reply messages to return the result of the method invocation to the client
 a server stub, called a skeleton, unmarshals messages and marshals replies

34
common organization of a remote object with client-side proxy

35
 Binding a Client to an Object
 a process must first bind to an object before invoking its
methods, which results in a proxy being placed in the
process’s address space
 binding can be implicit (directly invoke methods using
only a reference to an object) or explicit (by calling a
special function)
 an object reference could contain
 network address of the machine where the object
resides
 endpoint of the server
 an identification of which object
 the protocol used
 ...

36
 Parameter Passing
 there are two situations when invoking a method with
object reference as parameter; is the object local or
remote to the client?
 remote object: copy and pass the reference of the object
as a value parameter; this means the object is passed by
reference
 local object: a copy of the object is passed; this means the
object is passed by value

37
the situation when passing an object by reference or by value

38
 2.4 Message Oriented Communication

 RPCs and RMIs are not adequate for all distributed system
applications
 the provision of access transparency may be good but
they have semantics that is not adequate for all
applications
 example problems
 they assume that the receiving side is running at the
time of communication
 a client is blocked until its request has been processed

39
 communication can be
 persistent or transient
 asynchronous or synchronous
 persistent: a message that has been submitted for
transmission is stored by the communication system as long
as it takes to deliver it to the receiver
 e.g., email delivery, snail mail delivery

persistent communication of letters back in the days of the Pony Express

40
 transient: a message that has been submitted for transmission is
stored by the communication system only as long as the sending
and receiving applications are executing
 asynchronous: a sender continues immediately after it has
submitted its message for transmission
 synchronous: the sender is blocked until its message is
stored in a local buffer at the receiving host or delivered to the
receiver
 the different types of communication can be combined
 persistent asynchronous: e.g., email
 transient asynchronous: e.g., UDP, asynchronous RPC
 in general there are six possibilities

Persistent Transient

Asynchronous  

Synchronous  message-oriented; three forms

41
persistent asynchronous communication persistent synchronous communication

42
2.4.2 Message-Oriented Transient Communication
 many applications are built on top of the simple message-
oriented model offered by the transport layer
 standardizing the interface of the transport layer by
providing a set of primitives allows programmers to use
messaging protocols
 they also allow porting applications

1. Berkley Sockets
 an example is the socket interface as used in Berkley
UNIX
 a socket is a communication endpoint to which an
application can write data that are to be sent over the
network, and from which incoming data can be read

43 B
Primitive Meaning Executed by

Socket Create a new communication endpoint; also reserve both

resources to send and receive messages

Bind Attach a local address to a socket; e.g., IP address

with a known port number

Listen Announce willingness to accept connections; for

connection-oriented communication
Accept Block caller until a connection request arrives
servers
Connect Actively attempt to establish a connection; the client
is blocked until connection is set up
Send Send some data over the connection
Receive Receive some data over the connection
Close Release the connection

socket primitives for TCP/IP

44
connection-oriented communication pattern using sockets

45
2. The Message-Passing Interface (MPI)
 sockets were designed to communicate across networks
using general-purpose protocol stacks such as TCP/IP
 they were not designed for proprietary protocols
developed for high-speed interconnection networks; of
course portability will suffer
 MPI is designed for parallel applications and tailored for
transient communication
 MPI assumes communication takes place within a known
group of processes, where each group is assigned an
identifier (groupID)
 each process within a group is also assigned an identifier
(processID)
 a (groupID, processID) identifies the source or destination
of a message, and is used instead of a transport-level
address

46
Primitive Meaning

MPI_bsend Append outgoing message to a local send buffer; to support

transient asynchronous communication
Send a message and wait until copied to local or remote
MPI_send buffer (to support receipt-based transient synchronous
communication)

MPI_ssend Send a message and wait until receipt starts (to support
delivery-based transient synchronous communication)

MPI_sendrecv Send a message and wait for reply (to support response-
based transient synchronous communication)

MPI_isend Pass reference to outgoing message, and continue (a

variant of MPI_send)

MPI_issend Pass reference to outgoing message, and wait until receipt

starts (a variant of MPI_ssend)
MPI_recv Receive a message; block if there are none
some of the most intuitive message-passing primitives of MPI

MPI_irecv Check if there is an incoming message, but do not block 47

2.4.3 Message-Oriented Persistent Communication
 there are message-oriented middleware services, called
message-queuing systems or Message-Oriented Middleware
(MOM)
 they support persistent asynchronous communication
 they have intermediate-term storage capacity for messages,
without requiring the sender or the receiver to be active during
message transmission
 unlike Berkley sockets and MPI, message transfer may take
minutes instead of seconds or milliseconds
Message-Queuing Model
 applications communicate by inserting messages in specific
queues
 it permits loosely-coupled communication
 the sender may or may not be running; similarly the receiver
may or may not be running, giving four possible combinations

48
four combinations for loosely-coupled communications using queues
49
Primitive Meaning

Put Append a message to a specified queue; by the sender

and is nonblocking

Get Block until the specified queue is nonempty, and remove

the first message

Poll Check a specified queue for messages, and remove the

first. Never block

Notify Install a handler to be called when a message is put into

the specified queue; usually a daemon
basic interface to a queue in a message-queuing system

50
 General Architecture of a Message-Queuing System
 messages can be put only into queues that are local to the
sender (same machine or on a nearby machine on a LAN)
 such a queue is called the source queue
 messages can also be read only from local queues
 a message put into a local queue must contain the specification
of the destination queue; hence a message-queuing system must
maintain a mapping of queues to network locations; like in DNS

the relationship between queue-level addressing and network-level addressing

51
 messages are managed by queue managers
 they generally interact with the application that sends and
receives messages
 some also serve as routers or relays, i.e., they forward
incoming messages to other queue managers
 however, each queue manager needs a copy of the queue-
to-location mapping, leading to network management
problems for large-scale queuing systems
 the solution is to use a few routers that know about the
network topology

52
 Message Brokers
 how can applications understand the messages they receive
 each receiver can not be made to understand message formats
of new applications
 hence, in a message-queuing system conversations are
handled by message brokers
 a message broker converts incoming messages to a format
that can be understood by the destination application based on
a set of rules

the general organization of a message broker in a message-queuing system

53
 2.4 Stream Oriented Communication
 until now, we focused on exchanging independent and complete
units of information
 time has no effect on correctness; a system can be slow or fast
 however, there are communications where time has a critical role
 Multimedia
 media
 storage, transmission, interchange, presentation,
representation and perception of different data types:
 text, graphics, images, voice, audio, video, animation, ...
 movie: video + audio + …
 multimedia: handling of a variety of representation media
 end user pull
 information overload and starvation
 technology push
 emerging technology to integrate media

54
 Types of Media
 two types
 discrete media: text, executable code, graphics, images;
temporal relationships between data items are not
fundamental to correctly interpret the data
 continuous media: video, audio, animation; temporal
relationships between data items are fundamental to
correctly interpret the data
 a data stream is a sequence of data units and can be applied
to discrete as well as continuous media
 stream-oriented communication provides facilities for the
exchange of time-dependent information (continuous media)
such as audio and video streams

55
 timing in transmission modes
 asynchronous transmission mode: data items are transmitted
one after the other, but no timing constraints; e.g. text
transfer
 synchronous transmission mode: a maximum end-to-end
delay defined for each data unit; it is possible that data can
be transmitted faster than the maximum delay, but not slower
 isochronous transmission mode: maximum and minimum
end-to-end delay are defined; also called bounded delay
jitter; applicable for distributed multimedia systems
 a continuous data stream can be simple or complex
 simple stream: consists of a single sequence of data; e.g.,
mono audio, video only
 complex stream: consists of several related simple streams
that must be synchronized; e.g., stereo audio, video
consisting of audio and video (may also contain subtitles,
translation to other languages, ...)
56
movie as a set of simple streams

57
 a stream can be considered as a virtual connection between a
source and a sink
 the source or the sink could be a process or a device

setting up a stream between two processes across a network

setting up a stream directly between two devices 58

 Quality of Service (QoS)
 QoS requirements describe what is needed from the
underlying distributed system and network to ensure
acceptable delivery; e.g. viewing experience of a user
 for continuous data, the concerns are
 timeliness: data must be delivered in time
 volume: the required throughput must be met
 reliability: a given level of loss of data must not be
exceeded
 quality of perception; highly subjective

59
 QoS Dimensions
 timeliness dimensions
 latency (maximum delay between consecutive frames)
 start-up latency (maximum delay before starting a
presentation)
 jitter (delay variance)
 volume dimensions
 throughput in frames/sec or bits/sec or bytes/sec
 reliability dimensions
 MTBF (Mean Time Between Failure) of disks
 MTTR (Mean Time To Repair)
 error rates on the telecommunication lines

60
 QoS Requirements
 deterministic
 precise values or ranges
 e.g., latency must be between 45 and 55 ms
 probabilistic
 probability of the required QoS
 e.g., latency should be < 50 ms for 95% of the frames
 stochastic distributions
 e.g., frame arrival should follow normal distribution with
mean interval-time of 40 ms and 5 ms variance
 classes
 e.g., guaranteed and best effort

61
 QoS Management
 can be static or dynamic
 Static QoS Management Functions
 specification
 e.g., deterministic range for timeliness, volume and
reliability categories
 negotiation
 the application may accept lower level of QoS for
lower cost
 admission control
 if this test is passed, the system has to guarantee the
promised QoS
 resource reservation
 may be necessary to provide guaranteed QoS

62
 Dynamic QoS Management Functions
 monitoring
 notices deviation from QoS level
 at a certain level of granularity (e.g., every 100 ms)
 policing
 detect participants not keeping themselves to the contract
 e.g., source sends faster than negotiated (e.g., 25 fps)
 maintenance
 sustaining the negotiated QoS
 e.g., the system requires more resources
 renegotiation
 client tries to adapt – may be can accept lower QoS

63
 QoS requirements can be specified using flow specification
containing bandwidth requirements, transmission rates,
delays, ...
 e.g. by Partridge (1992)
 it uses the token bucket algorithm which specifies how the
stream will shape its network traffic (in fact the leaky
bucket, as used in networking)
 the idea is to shape bursty traffic into fixed-rate traffic by
averaging the data rate
 packets may be dropped if the bucket is full
 the input rate may vary, but the output rate remains
constant

64
 Stream Synchronization
 how to maintain temporal relations between streams, e.g., lip
synchronization
 two approaches
1. explicitly by operating on the data units of simple streams; the
responsibility of the application

the principle of explicit synchronization on the level of data units

65
2. through a multimedia middleware that offers a collection of
interfaces for controlling audio and video streams as well as
devices such as monitors, cameras, microphones, ...

the principle of synchronization as supported by high-level interfaces

66