CS 3551

DISTRIBUTED
COMPUTING
What is distributed
Computing?
HOD wants to assign paper correction (500 papers
in total)

Scenario 1 Scenario 2
Assign paper to single faculty Arun Assign paper to 5 faculties

Arun => 500 papers => 5 days Arun => 100 papers
=> Gopal => 100
papers => Ashif =>
100 papers => Hari
=> 100 papers =>
Murali => 100 papers
=>
Eg - Online banking transaction for 10 million
users
Eg 2 Image Rendering - Resize, Filter, Color,
Effects
(systems that are distributed by their own
nature)
Characteristi
cs
Relation to Computer System
Components
Key
Points
● The distributed software is also termed as middleware.
● A distributed execution is the execution of processes across the
distributed system to collaboratively achieve a common goal. An
execution is also sometimes termed a computation or a run.
● The middleware is the distributed software that drives the
distributed system, while providing transparency of heterogeneity
at the platform level.
● Middleware layer does not contain the traditional application layer
functions of the network protocol stack, such as http, mail, ftp, and
telnet.
● Various primitives and calls to functions defined in various
libraries of the middleware layer are embedded in the user
program code.
● There exist several libraries to choose from to invoke primitives for
the more common functions – such as reliable and ordered
multicasting – of the middleware layer.
● There are several standards such as Object Management Group’s
Motivation/ Benefit of Distributed
Computing
(systems that are distributed
by their own nature)
 Key Points
1. Inherently distributed computations
a. In many applications such as money transfer in banking, or reaching consensus among parties that are
geographically distant, the computation is inherently distributed.
2. Resource sharing
a. Resources such as peripherals, complete data sets in databases, special libraries, as well as data
(variable/files) cannot be fully replicated at all the sites because it is often neither practical nor
cost-effective.
b. Further, they cannot be placed at a single site because access to that site might prove to be a
bottleneck.
c. Therefore, such resources are typically distributed across the system. For example, distributed databases
such as DB2 partition the data sets across several servers, in addition to replicating them at a few sites
for rapid access as well as reliability.
3. Access to geographically remote data and resources
a. In many scenarios, the data cannot be replicated at every site participating in the distributed execution
because it may be too large or too sensitive to be replicated. For example, payroll data within a
multinational corporation is both too large and too sensitive to be replicated at every branch office/site.
It is therefore stored at a central server which can be queried by branch offices. Similarly, special
resources such as supercomputers exist only in certain locations, and to access such supercomputers,
users need to log in remotely.
4. Enhanced reliability
a. A distributed system has the inherent potential to provide increased reliability because of the possibility
of replicating resources and executions, as well as the reality that geographically distributed resources
are not likely to crash/malfunction at the same time under normal circumstances.
b. Reliability entails several aspects:
i. availability, i.e., the resource should be accessible at all times; •
ii. integrity, i.e., the value/state of the resource should be correct, in the face of concurrent access
from multiple processors, as per the semantics expected by the application;
iii. fault-tolerance, i.e., the ability to recover from system failures, where such failures may be
5. Increased performance/cost ratio
a. By resource sharing and accessing geographically remote data and resources, the
performance/cost ratio is increased.
b. Although higher throughput has not necessarily been the main objective behind using a
distributed system, nevertheless, any task can be partitioned across the various computers in
the distributed system.
c. Such a configuration provides a better performance/cost ratio than using special parallel
machines.
6. Scalability
a. As the processors are usually connected by a wide-area network, adding more processors does
not pose a direct bottleneck for the communication network.
7. Modularity and incremental expandability
a. Heterogeneous processors may be easily added into the system without affecting the
performance, as long as those processors are running the same middleware algorithms.
b. Similarly, existing processors may be easily replaced by other processors.
Distributed Vs Parallel
Computing
Message Passing vs Shared
Memory
Key
Points
● Shared memory systems are those in which there is a (common)
shared address space throughout the system.
● Communication among processors takes place via shared data
variables, and
control variables for synchronization among the processors.
● Semaphores and monitors that were originally designed for shared
memory uniprocessors and multiprocessors are examples of how
synchronization can be achieved in shared memory systems.
● All multicomputer (NUMA as well as message-passing) systems that do
not have a
shared address space provided by the underlying architecture and
hardware necessarily communicate by message passing.
● For a distributed system, this abstraction is called distributed
shared memory.
Implementing this abstraction has a certain cost but it simplifies the
task of the application programmer.
Emulating MP in
SM Shared Memory P1 P2 P3

1. The shared address space can be partitioned into disjoint parts,

one part being assigned to each processor.
2. “Send” and “receive” operations can be implemented by writing to
and reading from the destination/sender processor’s address space,
respectively. Specifically, a separate location can be reserved as the
mailbox for each ordered pair of processes.
3. A Pi–Pj message-passing can be emulated by a write by Pi to the
mailbox and then a read by Pj from the mailbox.
4. The write and read operations need to be controlled using
synchronization primitives to inform the receiver/sender after the
data has been sent/received.
Emulating SM in
MP
1. This involves the use of “send” and “receive” operations for “write”
and “read” operations.
2. Each shared location can be modeled as a separate process;
a. “write” to a shared location is emulated by sending an update message to the
corresponding owner process;
b. a “read” to a shared location is emulated by sending a query message to the
owner process.
3. the latencies involved in read and write operations may be high
even when using shared memory emulation
4. An application can of course use a combination of shared
memory and message-passing.
5. In a MIMD message-passing multicomputer system, each
“processor” may be a tightly coupled multiprocessor system with
shared memory. Within the multiprocessor system, the processors
communicate via shared memory. Between two computers, the
communication is by message passing
Primitives for Distributed
Computing
Function Parameters Need

Send() 1. destination, To send data from a

2. buffer in the user space, system to another
containing the data to be sent.

Receive() 1. source from which the data is To receive data

to be received from another
2. and the user buffer into which the system
data is to be received.

Sl Primitive Shortcut

1 Synchronous I care for you, you care for me

2 Asynchronous I don’t care for you

3 Blocking I care for my task completion

4 Non-blocking I don’t even care for my own task

Let’s Understand With Example
Need of
Handles
Key
Points
Primitives Part
Send
2a
● The data gets copied from the user buffer
to the kernel buffer and is then sent over
the network.
● After the data is copied to the receiver’s
system buffer and a Receive call has
been issued, an acknowledgement is sent
to sender.
● The process that invoked the Send
operation and completes the Send.

Receive

● The Receive call blocks until the data

expected arrives and is written in the
specified user buffer.
● Then control is returned to the user
process
Primitives Part
Send
2b
● Control returns back to the invoking process
as soon as the copy of data from the user
buffer to the kernel buffer is initiated
● Handle is returned as parameter which can
be used to check the process completion.

Receive

● When you make a Receive call, the system

gives you a special ID (handle).
● Handle can be used to check if the
non-blocking Receive operation is finished.
● The system sets this handle as "done"
when the expected data arrives and is put
in your specified place (buffer).
Primitives Part
Send
2c
● The user process that invokes the Send is
blocked until the data is copied from the
user’s buffer to the kernel buffer.
Primitives Part
Send
2d
● The user process that invokes the Send is
blocked until the transfer of the data from
the user’s buffer to the kernel buffer is
initiated.
● Control returns to the user process as soon
as this transfer is initiated, and handle is
given back.
● The asynchronous Send completes when
the data has been copied out of the user’s
buffer
Key
Points
1. A synchronous Send is easier to use from a programmer’s perspective
because the handshake between the Send and the Receive makes the
communication appear instantaneous, thereby simplifying the program
logic.
2. The non-blocking asynchronous Send (see Figure 1.8(d)) is useful when a
large
data item is being sent because it allows the process to perform other
instructions in parallel with the completion of the Send.
3. The non-blocking synchronous Send (see Figure 1.8(b)) also avoids the
potentially
large delays for handshaking, particularly when the receiver has not yet
issued the Receive call.
4. The non-blocking Receive (see Figure 1.8(b)) is useful when a large data
item is
being received and/or when the sender has not yet issued the Send call,
because it allows the process to perform other instructions in parallel with
Process Synchrony
Process Synchrony

• Processor synchrony indicates that all the processors execute in lock-step with their
clocks synchronized. As this synchrony is not attainable in a distributed system, what is
more generally indicated is that for a large granularity of code, usually termed as a
step, the processors are synchronized.

• This abstraction is implemented using some form of barrier synchronization to ensure

that no processor begins executing the next step of code until all the processors have
completed executing the previous steps of code assigned to each of the processors.
Libraries and
Standards

● In computer systems, there are many ways for programs to talk to each
other, like sending messages or making remote calls. Different software
products and scientific tools use their own special ways to do this.
● For example, some big companies use their custom methods, like
IBM's CICS software. Scientists often use libraries called MPI or PVM.
Commercial software often uses a method called RPC, which lets you
call functions on a different computer like you would on your own
computer.
● All these methods use something like a hidden network phone line
(called "sockets") to make these remote calls work.
● There are many types of RPC, like Sun RPC and DCE RPC. There are
also other ways to communicate, like "messaging" and "streaming."
● As software evolves, there are new methods like RMI and ROI for
object-based programs, and big standardized systems like CORBA
The text appears to describe characteristics of an asynchronous distributed system. Here’s a breakdown of each
point:

1. **No Processor Synchrony**: There is no guarantee that the processors (or nodes) in the system operate in
lockstep or maintain the same pace. This means that the timing of operations across different processors is not
coordinated, and each processor may run at a different speed.

2. **Unbounded Message Delays**: Messages sent between processors experience delays that are finite (they
eventually arrive) but can be unpredictably long. There's no strict upper limit on how long it might take for a
message to be delivered, making it difficult to rely on precise timing for communication.

3. **No Upper Bound on Process Execution Time**: The time it takes for a process (or task) on a processor to
execute can vary and has no guaranteed maximum. This means a process might take an unpredictable amount of
time to complete, further contributing to the asynchrony of the system.

In summary, these characteristics highlight the challenges of working with asynchronous systems, where timing is
uncertain and cannot be relied upon to ensure coordinated or predictable operation across different parts of the
system.
Synchronous vs Asynchronous Execution

Asynchronous Synchronous

1. there is no processor synchrony 1. processors are

and there is no bound on the synchronized and the clock drift
drift rate of processor rate between any two processors
clocks, is bounded
2. message delays 2. message delivery
(transmission + (transmission + delivery)
propagation times) are times are such that they
finite but unbounded occur in one logical
3. there is no upper bound on the step or round
time taken by a process to 3. there is a known upper bound
Key
Points
● It is easier to design and verify algorithms assuming synchronous
executions because of the coordinated nature of the executions at
all the processes.
● It is practically difficult to build a completely synchronous
system, and have the messages delivered within a bounded time.
(Simulated)
● Thus, synchronous execution is an abstraction that needs to be
provided to the programs. When implementing this abstraction,
observe that the fewer the steps or “synchronizations” of the
processors, the lower the delays and costs.
Emulating
Systems
Design Issue and
Challenges

1. From System Perspective

2. Algorithmic Challenges
3. Applications of distributed
computing
1. From System Perspective ( CSCANS-
PDF)
Challenge Description

Communication designing appropriate mechanisms for communication

among the processes in the network
Eg RPC,ROI, Message oriented vs stream oriented
communication
Synchronization Mechanism for synch among
processes. Eg Mutual exclusion &
leader election algos
Global state recording algo & physical
clocks need synch
Consistency Replicate for performance but it is essential that replicated
and replication compute should be consistent.

API and transparency 1. API Required for ease of use.

2. Access transparency hides differences in data
representation on different systems and provides
uniform operations to access system resources.
3. Location transparency makes the locations of
resources transparent to the users.
4. Migration transparency allows relocating resources
1. From System Perspective ( CSCANS-
PDF)
Challenge Description

Naming Easy to use and robust naming for identifiers, and addresses is essential
for locating resources and processes in a transparent and scalable
manner.
Security & Involves various aspects of cryptography, secure channels, access
Scalability control, key management – generation and distribution,
authorization, and secure group management
The system must be scalable to handle large loads.

Processes management of processes and threads at clients/servers; code

migration; and the design of software and mobile agents.

Data storage Schemes for data storage, and implicitly for accessing the data in a fast
and access and scalable manner across the network are important for efficiency

Fault tolerance ● Fault tolerance requires maintaining correct and efficient operation
in spite of any failures of links, nodes, and processes.
● Process resilience, reliable communication, distributed commit,
checkpointing and recovery, agreement and consensus, failure
detection, and
2. Algorithmic Challenges ( CREPT-MGW)
Challenge Description

group 1. A group is a collection of processes that share a common context and

Communication, collaborate
common ontask
a within an application
multicast, and domain
2. Algorithms need to be designed to enable efficient group communication
ordered message and group management wherein processes can join and leave groups
delivery 3. dynamically
Specify order of delivery when multiple process send message
concurrently
Replication ● Replicate for performance.
and ● Managing such replicas in the face of updates introduces the problems of
consistency ensuring consistency among the replicas and cached copies.

Execution models ● interleaving model and partial order model are two widely adopted models of
and frameworks distributed system executions
● The input/output automata model [25] and the TLA (temporal logic of
actions) are two other examples of models that provide different degrees
of infrastructure
Program design ● Methodically designed and verifiably correct programs can greatly reduce
and verification the
of software
overheaddesign, debugging, and
tools ● Designing
engineering mechanisms to achieve these design and verification goals is a
challenge.
2. Algorithmic Challenges ( CREPT-
MGW)
Challenge Description

Time and global 1. The processes in the system are spread across three-dimensional
state physical space. Another dimension, time, has to be superimposed
uniformly across space.
2. The challenges pertain to providing accurate physical time, and to providing
a variant of time, called logical time.
3. Logical time is relative time, and eliminates the overheads of providing
physical time for applications where physical time is not required.
4. Observing the global state of the system (across space) also involves
the time dimension for consistent observation

Monitoring ● On-line algorithms for monitoring such predicates are hence important.
distributed ● Event streaming is used where streams of relevant events reported
events and from different processes are examined collectively to detect predicates
predicates
Graph ● The distributed system is modeled as a distributed graph, and the graph
algorithms and algorithms form the building blocks for a large number of higher level
distributed communication, data dissemination, object location, and object search
routing functions.
● the design of efficient distributed graph algorithms is of paramount
algorithms
importance
2. Algorithmic Challenges (Synch
Mechanism)
2. Algorithmic Challenges (Distributed shared memory abstraction)
2. Algorithmic Challenges (Fault Tolerant)
2. Algorithmic Challenges (Load
Balancing)
2. Algorithmic Challenges (performance)
3. Applications of distributed
computing
3. Applications of distributed computing
Sensor Networks:

1. Sensors, which can measure physical properties like temperature and humidity, have
become affordable and are deployed in large numbers (over a million).
2. They report external events, not internal computer processes. These networks have various
applications, including mobile or static sensors that communicate wirelessly or through
wires. Self-configuring ad-hoc networks introduce challenges like position and time
estimation.

Ubiquitous Computing:

1. Ubiquitous systems involve processors integrated into the environment, working in the
background, like in sci-fi scenarios. Examples include smart homes and workplaces.
2. These systems are essentially distributed, use wireless tech, sensors, and actuators, and can
self-organize. They often consist of many small processors in a dynamic network, connecting to
more powerful resources for data processing.
3. Applications of distributed computing
Peer-to-Peer (P2P) Computing:

1. In P2P computing, all processors interact as equals without any hierarchy, unlike client-server
systems. P2P networks are often self-organizing and may lack a regular structure.
2. They don't use central directories for name resolution. Challenges include efficient object
storage and lookup, dynamic reconfiguration, replication strategies, and addressing issues
like privacy and security.

Publish-Subscribe, Content Distribution, and Multimedia: (Netflix)

3. As information grows, we need efficient ways to distribute and filter it. Publish-Subscribe
involves distributing information, letting users subscribe to what interests them, and then
filtering it based on user preferences.
4. Content distribution is about sending data with specific characteristics to interested users,
often used in web and P2P settings. When dealing with multimedia, we face challenges like
large data, compression, and synchronization during storage and playback.
3. Applications of distributed
computing
Data Mining Algorithms:

1. They analyze large data sets to find patterns and useful information. For
example, studying customer buying habits for targeted marketing.
2. This involves applying database and AI techniques to data. When data is distributed,
as in private banking or large-scale weather prediction, efficient distributed data
mining algorithms are needed.

Security Challenges in Distributed Systems:

1. Traditional challenges include ensuring confidentiality, authentication, and

availability.
2. The goal is efficient and scalable solutions.
3. In newer distributed architectures like wireless, peer-to-peer, grid, and pervasive
computing, these challenges become more complex due to resource constraints,
broadcast mediums, lack of structure, and network trust issues.
Distributed
Program
● A distributed program is composed of a set of n asynchronous processes p1, p2,
………, pi, , pn that communicate by message passing over the communication
network.
● We assume that each process is running on a different processor.
● The processes do not share a global memory and communicate solely
by passing messages.
● Cij - Channel from pi to process pj
● mij- a message sent by pi to pj.
● Don’t share a global clock.
● Process execution and message transfer are asynchronous
● The global state of a distributed computation is composed of the states of the
processes and the communication channels
● The state of a process is characterized by the state of its local memory and
depends upon the context. The state of a channel is characterized by the
set of messages in transit in the channel
Model for Distributed Execution
 A B
Model for Distributed
Execution
1. Execution of a process consists of a sequential execution of its
actions.
2. The actions are atomic and the actions of a process are modeled
as three types of events, internal events, message send
events(send (m)), and message receive events(rec(m)).
3. The occurrence of events changes the states of respective
processes and channels, thus causing transitions in the global
system state.
4. An internal event changes the state of the process at which it
occurs.
5. A send event (or a receive event) changes the state of the process
that sends (or receives) the message and the state of the
channel on which the message is sent (or received)
Causal Precedence Relation
Logical vs Physical
Concurrency
● In a distributed computation, two events are logically concurrent if
and only if they do not causally affect each other.
● Physical concurrency, on the other hand, has a connotation that
the events occur at the same instant in physical time.
● Note that two or more events may be logically concurrent even
though they do not occur at the same instant in physical time.
Models of Communication Networks (FIFO, Non-FIFO, Causal
Ordering)
1. FIFO - each channel acts as a first-in first-out message queue and thus, message
ordering is preserved by a channel.
2. non-FIFO model, a channel acts like a set in which the sender process adds
messages and the receiver process removes messages from it in a random order.
3. Causal Ordering (built-in synch)- based on Lamport’s “happens before” relation. A
system that supports the causal ordering model satisfies the following property:
Global State of Distributed System

1. The state of a process at any time is defined by the contents of processor registers, stacks, local
memory, etc. and depends on the local context of the distributed application.
2. The state of a channel is given by the set of messages in transit in the channel.
3. The occurrence of events changes the states of respective processes and channels, thus causing
transitions in global system state.
a. For example, an internal event changes the state of the process at which it occurs. A send event (or a receive event)
changes
the state of the process that sends (or receives) the message and the state of the channel on which the message is sent (or
received).
State of Process

State of
Channel
How to find global
state?
❖ For a global snapshot to be meaningful, the states of all the
components of the distributed system must be recorded at the same
instant.
❖ This will be possible if the local clocks at processes were perfectly
synchronized or
there was a global system clock that could be instantaneously
read by the processes. However, both are impossible.
❖ Solution:
❖ Recording at different time will be meaningful provided every
message that is recorded as received is also recorded as sent.
❖ Basic idea is that an effect should not be present without its cause.
❖ States that don’t violate causality are called consistent global states and are
meaningful global states
Inconsistent
state
Consistent
state
Strongly Consistent