Distributed Systems

Remote Procedure Calls

Interprocess Communication
• It is the heart of all distributed systems because :
• guide the processes on different machines to exchange
information
• Processes scattered across a network with unreliable comm.

• Based on low-level message passing

2
Middleware layers
Middleware Layer
• Middleware is:
• an application that lives in the OSI application layer.
• invented to provide common services and protocols that can be used
by many different applications:
• A rich set of communication protocols, but which allow different applications
to communicate
• (Un)marshaling of data, necessary for integrated systems
• Naming protocols, so that different applications can easily share resources
• Security protocols, to allow different applications to communicate in a secure
way
• Scaling mechanisms, such as support for replication and caching
• Examples: Domain name system (DNS), authentication
protocols, distributed locking protocols, remote procedure call
• Note: what remains are truly application-specific protocols
4
Problems with sockets
Sockets interface is
straightforward
• [connect]
• read/write
• [disconnect]

BUT … it forces read/write

mechanism
• We usually use a procedure call

To make distributed computing

look more like centralized:
• I/O is not the way to go
Java Socket – Server
1. import java.io.*;
2. import java.net.*;
3. public class MyServer {

4. public static void main(String[] args){

5. try{
6. ServerSocket ss=new ServerSocket(6666);
7. Socket s=ss.accept();//establishes connection
8. DataInputStream dis=new DataInputStream(s.getInputStream());
9. String str=(String)dis.readUTF();
10. System.out.println("message= "+str);
11. ss.close();
12. }catch(Exception e){System.out.println(e);}
13. }
14.}
Java Socket - Client
1. import java.io.*;
2. import java.net.*;
3. public class MyClient {
4. public static void main(String[] args) {
5. try{
6. Socket s=new Socket("localhost",6666);
7. DataOutputStream dout=new DataOutputStream(s.getOutputStream());
8. dout.writeUTF("Hello Server");
9. dout.flush();
10. dout.close();
11. s.close();
12. }catch(Exception e){System.out.println(e);}
13. }
14.}
RMI – Interface
Hello.java
1. import java.rmi.Remote;
2. import java.rmi.RemoteException;
3. // Creating Remote interface for our application
4. public interface Hello extends Remote {
5. void printMsg() throws RemoteException;
6. String getMsg(String a) throws RemoteException;
7. }

ImpleExample.java
1. public class ImplExample implements Hello {
2. // Implementing the interface method
3. public void printMsg() {
4. System.out.println("This is an example RMI program");
5. }
6. public String getMsg(String a) {
7. return "Server:$ Hello Client " + a;
8. }
9. }
RMI - Server
1. import java.rmi.registry.*;
2. import java.rmi.*;
3. public class HelloServer {
4. public static void main(String args[]) {
5. try {
6. System.setProperty("java.rmi.server.hostname", "192.168.1.1");
7. // Instantiating the implementation class
8. ImplExample obj = new ImplExample();
9. // Binding the remote object (stub) in the registry
10. Naming.rebind("rmi://192.168.1.1:5000/Hello",obj);
11. System.err.println("Server ready");
12. } catch (Exception e) {
13. System.err.println("Server exception: " + e.toString());
14. e.printStackTrace();}
15. }
16.}
RMI - Client

1. import java.rmi.*;
2. public class HelloClient {
3. private HelloClient() {}
4. public static void main(String[] args) {
5. try { // Looking up the registry for the remote object
6. Hello stub = (Hello) Naming.lookup("rmi://192.168.1.1:5000/Hello");
7. // Calling the remote method using the obtained object
8. stub.printMsg();
9. } catch (Exception e) { System.err.println("Client exception: " + e.toString());
e.printStackTrace(); }
10. }
11. }
RPC
1984: Birrell & Nelson
• Mechanism to call procedures on other machines

Remote Procedure Call

Goal: it should appear to the programmer

that a normal call is taking place
 How do regular
procedure calls work
in programming
languages?
Regular procedure calls
Machine instructions for call & return but the compiler really
makes the procedure call abstraction work:
• Parameter passing
• Local variables
• Return data
Regular procedure calls
You write:
x = f(a, “test”, 5);

The compiler parses this and generates code to:

a.Push the value 5 on the stack
b.Push the address of the string “test” on the stack
c.Push the current value of a on the stack
d.Generate a call to the function f
In compiling f, the compiler generates code to:
a.Push registers that will be clobbered on the stack to save the values
b.Adjust the stack to make room for local and temporary variables
c.Before a return, unadjust the stack, put the return data in a register,
and issue a return instruction
Implementing RPC
No architectural support for remote procedure calls
Simulate it with tools we have
(local procedure calls)

Simulation makes RPC a

language-level construct
instead of an
operating system construct
Implementing RPC
The trick:

Create stub functions to make it appear to the user that the call is
local

Stub function contains the function’s interface

Stub functions
1. Client calls stub (params on stack)

client functions server functions

server stub
client stub
(skeleton)

network routines network routines

client server
Stub functions
2. Stub marshals params to net message

client functions server functions

server stub
client stub
(skeleton)

network routines network routines

client server
Stub functions
3. Network message sent to server

client functions server functions

server stub
client stub
(skeleton)

network routines network routines

client server
Stub functions
4. Receive message: send to stub

client functions server functions

server stub
client stub
(skeleton)

network routines network routines

client server
Stub functions
5. Unmarshal parameters, call server func

client functions server functions

server stub
client stub
(skeleton)

network routines network routines

client server
Stub functions
6. Return from server function

client functions server functions

server stub
client stub
(skeleton)

network routines network routines

client server
Stub functions
7. Marshal return value and send message

client functions server functions

server stub
client stub
(skeleton)

network routines network routines

client server
Stub functions
8. Transfer message over network

client functions server functions

server stub
client stub
(skeleton)

network routines network routines

client server
Stub functions
9. Receive message: direct to stub

client functions server functions

server stub
client stub
(skeleton)

network routines network routines

client server
Stub functions
10. Unmarshal return, return to client code

client functions server functions

server stub
client stub
(skeleton)

network routines network routines

client server
RPC: Characteristics
• Called procedure is in another process which may reside
in another machine.
• The processes do not share address space.
• Passing of parameters by reference and passing pointer values
are not allowed.
• Parameters are passed by values.
• Called remote procedure executes within the
environment of the server process.
• The called procedure does not have access to the calling
procedure's (client’s) environment.

27
RPC: Features
• Simple call syntax
• Familiar semantics
• Well defined interface
• Ease of use
• Being able to communicate between processes on the
same machine or different machines

28
RPC: Limitations
• Parameters: passed by values only and pointer values are not
allowed.
• Speed: remote procedure calling (and return) time (i.e.,
overheads) can be significantly (1 - 3 orders of magnitude)
slower than that for local procedure.
• This may affect real-time design and the programmer should be aware
of its impact.
• Failure issues: RPC is more vulnerable to failure (why?).
• The programmers should be aware of the call semantics, i.e., programs
that make use of RPC must have the capability of handling errors that
cannot occur in local procedure calls.

29
RPC: Mechanism
• Marshalling (at sender): the process of taking a collection
of data items and assembling them into a form suitable for
transmission in a message;
• Unmarshalling (at receiver): the process of disassembling
them on arrival to produce an equivalent collection of
data items at the destination.

30
RPC: Mechanism
• How does the client transfer its call request (the procedure
name) and the arguments to the server via network?
• Marshalling and communication with server:
• For each RPC, a (client) stub procedure is generated and attached to
the (client) program.
• Replace the remote procedure call to a (local) call to the stub
procedure.
• The (codes in the) stub procedure marshals (the input) arguments and
places them into a message together with the procedure identifier (of
the remote procedure).
• Use IPC primitive to send the (call request) message to the server and
wait the reply (call return) message.

31
RPC: Mechanism
• How does the server react the request of the client? From
which port? How to select the procedure? How to interpret the
arguments?
• Dispatching, unmarshalling, communication with client:
• A despatcher is provided. It receives the call request message from the
client and uses the procedure identifier in the message to select one of
the server stub procedures and passes on the arguments.

• For each procedure at the server which is declared (at the

sever interface) as callable remotely, a (server) stub procedure
is generated.
• The task of a server stub procedure is to unmarshal the
arguments, call the corresponding (local) service procedure.
32
RPC: Mechanism
• How does the server transmit the reply back?
• On return, the stub marshals the output arguments into a reply
(call return) message and sends it back to the client.

• How does the client receive the reply?

• The stub procedure of the client unmarshals the result arguments
and returns (local call return). Note that the original remote
procedure call was transformed into a (local) call to the stub
procedure.

33
RPC: Implementation
• The software that supports RPC has three main tasks:
• Interface processing: integrate the RPC mechanism with client
and server programs in conventional programming languages.
• Communication handling: transmitting and receiving request
and reply messages.
• Binding: locating an appropriate server for a particular service.

34
RPC: Implementation
• Interface Processing: An interface definition may be used as
basis on which to construct the extra software components of
the client and server programs enabling RPC.
• Building the client program: An RPC system provides a means to build
a complete client program by having sub procedure to stand in for
each RPC.
• Building the server program: An RPC system provides a dispatcher and
set of server sub procedures.
• An interface compiler in an RPC system processes interface definitions
to produce components that can be combined with client and server
programs:
• Generating a client stub procedure: The sub procedures will be compiled and
linked with client program.

35
RPC: Implementation
• Generating a server stub procedure, the dispatcher and the server stub
will be compiled and linked with the server program;
• Use the signatures of the procedures in the interface to generate
marshalling and unmarshalling operations in each stub procedure;
• Generating procedure headings for each procedure in the service from
the interface definition;

• Communication handling: the module in both client and

server programs is to deal with communication between
them (generally using request-reply). The module is
provided in forms suitable for linking with client and server
programs.
RPC: Implementation (in C, Unix)
• Binding
• It specifies a mapping from a name to a particular object usually
identified by a communication ID (e.g., server port).
• An interface definition specifies a textual service name for use by
clients and servers. Clients that request the service (specified by
service name) must send the request message to the server port
binding to the service.

37
RPC: Implementation
• Binder: a separate service that maintains a table containing
mappings from service names to server ports.
• All other services depend on the binder service.
• Binder interface used by server
• Register (String serviceName, Port serverPort, int version)
• Withdraw (String serviceName, Port serverPort, int version)
• Binder interface used by client
• PortLookUp (String serviceName, int version)

38
Where to bind?
Need to locate host and correct server process
Where to bind? – Solution 1
• Maintain centralized DB that can locate a host that
provides a particular service (Birrell & Nelson’s 1984
proposal)
Where to bind? – Solution 2
• A server on each host maintains a DB of locally provided
services

• Solution 1 is problematic for Sun NFS – identical file servers

serve different file systems
Transport protocol
Which one?
• Some implementations may offer only one
(e.g. TCP)
• Most support several protocols
• Allow programmer (or end user) to choose
When things go wrong
• Local procedure calls do not fail
• If they core dump, entire process dies

• More opportunities for error with RPC:

• Transparency breaks here
• Applications should be prepared to deal with RPC failure
When things go wrong
• Semantics of remote procedure calls
• Local procedure call: exactly once

• A remote procedure call may be called:

• 0 times: server crashed or server process died before executing
server code
• 1 time: everything worked well
• 1 or more: excess latency or lost reply from server and client
retransmission
More issues
Performance
• RPC is slower … a lot slower

Security
• messages visible over network
• Authenticate client
• Authenticate server
Interface Definition Language
• A specification language which allow programmer to
specify remote procedure interfaces
(names, parameters, return values)

• Pre-compiler can use this to generate client and server

stubs:
• Marshaling code
• Unmarshaling code
• Network transport routines
• Conform to defined interface
• Similar to function prototypes
RPC compiler client code (main)

client stub

data conv. compiler client

RPC
IDL compiler
headers

data conv. compiler server

server skeleton

server functions
Code you write

Code RPC compiler generates

Writing the program
Client code has to be modified
• Initialize RPC-related options
• Transport type
• Locate server/service
• Handle failure of remote procedure call

Server functions
• Generally need little or no modification
RPC API
What kind of services does an RPC system need?
• Name service operations
• Export/lookup binding information (ports, machines)
• Support dynamic ports

• Binding operations
• Establish client/server communications using appropriate protocol
(establish endpoints)

• Endpoint operations
• Listen for requests, export endpoint to name server
RPC API
What kind of services does an RPC system need?
• Security operations
• Authenticate client/server
• Internationalization operations
• Marshaling/data conversion operations
• Stub memory management
• Dealing with “reference” data, temporary buffers
• Program ID operations
• Allow applications to access IDs of RPC interfaces
Types of Communication (1/3)
• Distinguish:
• Transient versus persistent communication
• Asynchronous versus synchronous communication

51
Types of Communication (2/3)
• Transient (impermanent) communication:
• A message is discarded by a communication server as soon as it
cannot be delivered at the next server, or at the receiver (store and
forward router – drop the message if it cannot deliver the message to
the next or destination host)
• All transport level communication services offer transient
communication.
• Persistent (nonstop / uninterrupted) communication:
• A message is stored at a communication server as long as it takes to
deliver it at the receiver (message that has been submitted for
transmission is stored by communication middleware as long as it takes
to deliver it to the receiver)
• Example: email

52
Types of Communication (3/3)
• Asynchronous communication
• Sender continues immediately after it has submitted its message for
transmission
• E.g. discussion boards (Dialogue that takes place over a period of time),
messaging (One-to-one or one-to-many communications)
• Synchronous communication
• Sender is blocked until its request is known to be accepted
• Places for synchronization:
• At request submission – middleware notifies that it will take over transmission of the
request
• At request delivery to the intended recipient
• After request is fully processing – up the time that the recipient returns a response
• Ex: Audio conferencing and video conferencing.

53
Client / Server communication
• Client/Server communication is generally based on a model of
transient synchronous communication:
• It’s designed to support the roles and message exchanges in typical
client-server interactions.
• Client and server have to be active at the time of communication
• Client issues request and blocks until it receives reply
• Server essentially waits only for incoming requests, and subsequently
processes them

• Drawbacks synchronous communication:

• Client cannot do any other work while waiting for reply
• Failures have to be dealt with immediately (the client is waiting)
• In many cases the model is simply not appropriate (mail, news)
54
Combinations of persistence, transient, synchronous
and asynchronous communications
• Persistence and Asynchronization
• Common for message queuing system
• Transient and synchronization
• RPC

55
Messaging
• Message-oriented middleware: Aims at high-level
persistent asynchronous communication:
• Processes send each other messages, which are queued
• Sender need not wait for immediate reply, but can do other
things
• Middleware often ensures fault tolerance

56
RPC: Design Issues
• Exception handling
• Necessary because of possibility of network and nodes failures;
• RPC uses return value to indicate errors;

• Transparency
• Syntactic → achievable, exactly the same syntax as a local
procedure call;
• Semantic → impossible because of RPC limitation: failure (similar
but not exactly the same);

57
RPC: Design Issues
• Delivery guarantees
• Retry request message: whether to retransmit the request
message until either a reply or the server is assumed to be failed;
• Duplicate filtering: when retransmission are used, whether to filter
out duplicates at the server;
• Retransmission of replies: whether to keep a history of reply
messages to enable lost replies to be retransmitted without re-
executing the server operations.

58
Asynchronous RPCs (1/2)
• Try to get rid of the strict request-reply behavior, but let the
client continue without waiting for an answer from the
server.

(a) The interaction between client and server in a traditional RPC.

(b) The interaction using asynchronous RPC
59
Asynchronous RPCs (2/2)
• Variation:
• Deferred synchronous RPC: Combining two asynchronous RPCs.
• The client first calls the server, waits for acceptance, and continues
• Results become available, the server sends a response message that leads to a
callback at the client side
• One-way RPCs: client does not wait for an acknowledgment of the server’s
acceptance of the request
• Problem : reliability is not guaranteed

A client and server interacting through asynchronous RPCs

60
Message-Oriented Communication
• Transient Messaging
• Message-Queuing System
• Message Brokers

61
Transient Messaging: Socket
• Many distributed systems and
application are built on top
of the simple message-
oriented model offered by
the transport layer.
• Example: Consider the
Berkeley socket interface,
which has been adopted by
all UNIX systems, as well as
Windows
Connection-oriented communication pattern
using sockets
Socket Programming
• Refer to eLeap for self-study material on Socket
Programming.
• You are encourage to try, to ensure you understand the
later topics.
Message-Oriented Middleware
• It is an asynchronous persistent communication through support of
middleware-level queues.
• How it works?
• Application communicate by inserting message s in specific queues (Queues
correspond to buffers at communication servers).
• Message are forwarded over a series of communication servers and delivered
to the destination (even if it was down when the message was sent)
• Each application has it own private queue, and can be read only by its
associated application, and possible for multiple applications to share a single
queue.
• Guarantee the message will inserted in the recipient’s queue, but
not when if the message will be read (depends on the recipient’s
behavior)
64
Basic interface to a queue in a message-queuing system
Message Broker
• Message queuing systems assume a common messaging
protocol: all applications agree on message format (i.e.,
structure and data presentation), but with a highly diverse
system it is not a simply task
• Application can provide information on the organization of its
message and semantics of the message
• Message broker: Centralized component that takes care of
application heterogeneity in an MQ system:
• Transforms incoming messages to target format
• Very often acts as an application gateway
• May provide subject-based routing capabilities ⇒ Enterprise
Application Integration

66
The general organization of a message broker in a message-queue
system

67
Multicast Communication
• Why multicast communication?
• Sending data to multiple receivers
• Issue
• Setting up the communication paths for information dissemination
• Involved a huge management effort -> require human intervention
• Today with peer-to-peer technology and notably structured
overlay management, it because easier to set up
communication paths
• Deploy at application layer
• Gossip-based information dissemination
68
Application-Level tree-based multicasting
• Organize nodes of a distributed system into an overlay
network and use that network to disseminate data
• Initiator generates a multicast identifier (mid).
• Lookup succ(mid), the node responsible for mid
• Request is routed to succ(mid), which will become the root.
• If P wants to join, it sends a join request to the root.
• When request arrives at Q:
• Q has not seen a join request before ⇒ it becomes forwarder; P becomes
child of Q. Join request continues to be forwarded.
• Q knows about tree ⇒ P becomes child of Q. No need to forward join
request anymore.

69
Gossip based information Dissemination -
Epidemic Algorithms - Outline
• General background
• Update models
• Removing objects

70
Principles
• Basic idea:
• Update operations are initially performed at one or only a few replicas
• A replica passes its updated state to a limited number of neighbors
• Update propagation is lazy, i.e., not immediate
• Eventually, each update should reach every replica
• Epidemic / Gossiping :
• A replica which has just been updated (i.e., has been contaminated), tells a
number of other replicas about its update (contaminating them as well).
• Infected: holds data that it is willing to spread to other
• Susceptible: A node that has not yet seen this data
• Removed: An updated node that is not willing or able to spread its data
• Goal: to rapidly propagate information among a large collection of nodes
using only local information, no central component to coordinate the
dissemination

71
Anti-Entropy
• Each replica regularly chooses another replica at random,
and exchanges state differences, leading to identical
states at both afterwards
• Approaches to exchanging updates:
• Push: P only sends its updates to Q (Chance of Q to be
susceptible?)
• Pull: P only retrieves updates from Q (Chance of P to be
infected?)
• Push-Pull: P and Q exchange mutual updates (after which they
hold the same information).
72
Deleting Values
• Fundamental problem: We cannot remove an old value from a
server and expect the removal to propagate. Instead, removal will
be undone in due time using epidemic algorithms
• Solution: Removal has to be registered as a special update by
inserting a death certificate
• Next problem: When to remove a death certificate (it is not allowed
to stay for ever):
• Run a global algorithm to detect whether the removal is known everywhere,
and then collect the death certificates (looks like garbage collection)
• Assume death certificates propagate in finite time, and associate a maximum
lifetime for a certificate (can be done at risk of not reaching all servers)
• Note: it is necessary that a removal actually reaches all servers.
73
Unsupervised Lab
• Please refer to eLeap for lab material on RPC
• This is unsupervised lab (you are on your 3rd / 4th year
anyway), please do it on your own time
Thank You
Any Question?

75