Basics and models of distributed systems

Table of Content

• Chapter 1: Introduction
  
• Chapter 2: A Model of Distributed Computations
  
• Assignment 1: Basics
  

Chapter 1: Introduction

Definition

·        Autonomous processors communicating over a communication network Some characteristics

·        No common physical clock

§  No shared memory

§  Geographical seperation

§  Autonomy and heterogeneity

Distributed System Model

 Relation between Software Components

Motivation for Distributed System

• Inherently distributed computation
•  Resource sharing
• Access to remote resources Increased performance/cost ratio 
• Reliability

Ø availability, integrity, fault-tolerance

Ø Scalability

Ø Modularity and incremental expandability

Parallel Systems

·    Multiprocessor systems (direct access to shared memory, UMA model)

Ø Interconnection network - bus, multi-stage sweitch

Ø E.g., Omega, Butterfly, Clos, Shuffle-exchange networks

Ø Interconnection generation function, routing function 

·    Multicomputer parallel systems (no direct access to shared memory, NUMA model)

Ø bus, ring, mesh (w w/o wraparound), hypercube topologies

Ø E.g., NYU Ultracomputer, CM* Conneciton Machine, IBM Blue gene 

·    Array processors (colocated, tightly coupled, common system clock)

Ø Niche market, e.g., DSP applications

UMA vs. NUMA Models

Omega, Butterfly Interconnects

Omega Network

·    n processors, n memory banks

·    log n stages: with n/2 switches of size 2x2 in each stage

·    Interconnection function: Output i of a stage connected to input j of next stage:

                                     

·    Routing function: in any stage s at any switch:

to route to dest. j,

          if s + 1th MSB of j = 0 then route on upper wire

          else [s + 1th MSB of j = 1] then route on lower wire

Interconnection Topologies for Multiprocesors

Figure 1.5: (a) 2-D Mesh with wraparound (a.k.a. torus) (b) 3-D hypercube

Flynn’s Taxonomy

Figure 1.6: SIMD, MISD, and MIMD modes.

·    SISD: Single Instruction Stream Single Data Stream (traditional)

·    SIMD: Single Instruction Stream Multiple Data Stream

Ø scientific applicaitons, applications on large arrays

Ø vector processors, systolic arrays, Pentium/SSE, DSP chips 

·    MISD: Multiple Instruciton Stream Single Data Stream

Ø E.g., visualization 

·    MIMD: Multiple Instruction Stream Multiple Data Stream

Ø distributed systems, vast majority of parallel systems

Terminology

·    Coupling

Ø  Interdependency/binding among modules, whether hardware or software (e.g., OS, middleware)

·    Parallelism: T (1)/T (n).

Ø Function of program and system

·    Concurrency of a program

Ø Measures productive CPU time vs. waiting for synchronization operations

·    Granularity of a program

Ø  Amt. of computation vs. amt. of communication

Ø  Fine-grained program suited for tightly-coupled system

Message-passing vs. Shared Memory

·    Emulating MP over SM:

Ø Partition shared address space

Ø Send/Receive emulated by writing/reading from special mailbox per pair of processes 

·    Emulating SM over MP:

Ø  Model each shared object as a process

Ø  Write to shared object emulated by sending message to owner process for the object

Ø  Read from shared object emulated by sending query to owner of shared object

Classification of Primitives (1)

·    Synchronous (send/receive)

Ø Handshake between sender and receiver

Ø Send completes when Receive completes

Ø Receive completes when data copied into buffer

·    Asynchronous (send)

Ø Control returns to process when data copied out of user-specified buffer

Classification of Primitives (2)

·    Blocking (send/receive)

Ø Control returns to invoking process after processing of primitive (whether sync or async) completes

·    Nonblocking (send/receive)

Ø Control returns to process immediately after invocation

Ø Send: even before data copied out of user buffer

Ø Receive: even before data may have arrived from sender

Non-blocking Primitive

Figure 1.7: A nonblocking send primitive. When the Wait call returns, at least one of its parameters is posted.

·    Return parameter returns a system-generated handle

Ø    Use later to check for status of completion of call

Ø    Keep checking (loop or periodically) if handle has been posted

Ø    Issue Wait(handle1, handle2, . . .) call with list of handles

Ø    Wait call blocks until one of the stipulated handles is posted

Blocking/nonblocking; Synchronous/asynchronous; send/receive primities

Figure 1.8:Illustration of 4 send and 2 receive primitives

 Asynchronous Executions; Message-passing System

Synchronous Executions: Message-passing System

Figure 1.10: Synchronous execution in a message-passing system

In any round/step/phase: (send | internal)∗(receive | internal)∗

(1) Sync Execution(int k, n) //k rounds, n processes.

(2) for r = 1 to k do

(3) proc i sends msg to (i + 1) mod n and (i − 1) mod n;

(4) each proc i receives msg from (i + 1) mod n and (i − 1) mod n;

(5) compute app-specific function on received values.

Synchronous vs. Asynchronous Executions (1)

·    Sync vs async processors; Sync vs async primitives

·    Sync vs async executions

·    Async execution

Ø No processor synchrony, no bound on drift rate of clocks

Ø Message delays finite but unbounded

Ø No bound on time for a step at a process

·    Sync execution

Ø Processors are synchronized; clock drift rate bounded

Ø Message delivery occurs in one logical step/round

Ø Known upper bound on time to execute a step at a process

Synchronous vs. Asynchronous Executions (2)

·    Difficult to build a truly synchronous system; can simulate this abstraction

·    Virtual synchrony:

Ø  async execution, processes synchronize as per application requirement;

Ø  execute in rounds/steps

·    Emulations:

Ø  Async program on sync system: trivial (A is special case of S)

Ø  Sync program on async system: tool called synchronizer

System Emulations

Figure 1.11: Sync ↔ async, and shared memory ↔ msg-passing emulations

·    Assumption: failure-free system

·    System A emulated by system B:

Ø  If not solvable in B, not solvable in A

Ø  If solvable in A, solvable in B

Challenges: System Perspective (1)

·    Communication mechanisms: E.g., Remote Procedure Call (RPC), remote object invocation (ROI), message-oriented vs. stream-oriented communication

·    Processes: Code migration, process/thread management at clients and servers, design of software and mobile agents

·    Naming: Easy to use identifiers needed to locate resources and processes transparently and scalably

·    Synchronization

·    Data storage and access

Ø Schemes for data storage, search, and lookup should be fast and scalable across network

Ø Revisit file system design

·    Consistency and replication

Ø Replication for fast access, scalability, avoid bottlenecks

Ø Require consistency management among replicas

Challenges: System Perspective (2)

·    Fault-tolerance: correct and efficient operation despite link, node, process failures

·    Distributed systems security

Ø Secure channels, access control, key management (key generation and key distribution), authorization, secure group management

·    Scalability and modularity of algorithms, data, services

·    Some experimental systems: Globe, Globus, Grid

Challenges: System Perspective (3)

·    API for communications, services: ease of use

·    Transparency: hiding implementation policies from user

Ø  Access: hide differences in data rep across systems, provide uniform operations to access resources

Ø  Location: locations of resources are transparent

Ø  Migration: relocate resources without renaming

Ø  Relocation: relocate resources as they are being accessed

Ø  Replication: hide replication from the users

Ø  Concurrency: mask the use of shared resources

Ø  Failure: reliable and fault-tolerant operation

Challenges: Algorithm/Design (1)

·    Useful execution models and frameworks: to reason with and design correct distributed programs

Ø Interleaving model

Ø Partial order model

Ø Input/Output automata

Ø Temporal Logic of Actions

·    Dynamic distributed graph algorithms and routing algorithms

Ø System topology: distributed graph, with only local neighborhood knowledge

Ø Graph algorithms: building blocks for group communication, data dissemination, object location

Ø Algorithms need to deal with dynamically changing graphs

Ø Algorithm efficiency: also impacts resource consumption, latency, traffic, congestion

Challenges: Algorithm/Design (2)

·    Time and global state

Ø 3D space, 1D time

Ø Physical time (clock) accuracy

Ø Logical time captures inter-process dependencies and tracks relative time progression

Ø Global state observation: inherent distributed nature of system

Ø Concurrency measures: concurrency depends on program logic, execution speeds within logical threads, communication speeds

Challenges: Algorithm/Design (3)

·    Synchronization/coordination mechanisms

Ø Physical clock synchronization: hardware drift needs correction

Ø Leader election: select a distinguished process, due to inherent symmetry

Ø Mutual exclusion: coordinate access to critical resources

Ø Distributed deadlock detection and resolution: need to observe global state; avoid duplicate detection, unnecessary aborts

Ø Termination detection: global state of quiescence; no CPU processing and no in-transit messages

Ø Garbage collection: Reclaim objects no longer pointed to by any process

Challenges: Algorithm/Design (4)

·    Group communication, multicast, and ordered message delivery

Ø Group: processes sharing a context, collaborating

Ø Multiple joins, leaves, fails

Ø Concurrent sends: semantics of delivery order

·    Monitoring distributed events and predicates

Ø Predicate: condition on global system state

Ø Debugging, environmental sensing, industrial process control, analyzing event streams

·    Distributed program design and verification tools

·    Debugging distributed programs

Challenges: Algorithm/Design (5)

·    Data replication, consistency models, and caching

Ø Fast, scalable access;

Ø coordinate replica updates;

Ø optimize replica placement

·    World Wide Web design: caching, searching, scheduling

Ø Global scale distributed system; end-users

Ø Read-intensive; prefetching over caching

Ø Object search and navigation are resource-intensive

Ø User-perceived latency

Challenges: Algorithm/Design (6)

·    Distributed shared memory abstraction

Ø Wait-free algorithm design: process completes execution, irrespective of actions of other processes, i.e., n − 1 fault-resilience 

Ø Mutual exclusion

§ Bakery algorithm, semaphores, based on atomic hardware primitives, fast algorithms when contention-free access

Ø Register constructions

§  Revisit assumptions about memory access

§  What behavior under concurrent unrestricted access to memory? Foundation for future architectures, decoupled with technology (semiconductor, biocomputing, quantum . . .) 

·    Consistency models:

§ coherence versus access cost trade-off

§ Weaker models than strict consistency of uniprocessors

Challenges: Algorithm/Design (7)

·    Reliable and fault-tolerant distributed systems

Ø Consensus algorithms: processes reach agreement in spite of faults (under various fault models)

Ø Replication and replica management

Ø Voting and quorum systems

Ø Distributed databases, commit: ACID properties

Ø Self-stabilizing systems: ”illegal” system state changes to ”legal” state; requires built-in redundancy

Ø Checkpointing and recovery algorithms: roll back and restart from earlier ”saved” state

Ø Failure detectors:

§ Difficult to distinguish a” slow” process/message from a failed process/ never sent message

§ algorithms that” suspect” a process as having failed and converge on a determination of its up/down status

Challenges: Algorithm/Design (8)

·    Load balancing: to reduce latency, increase throughput, dynamically. E.g., server farms

Ø Computation migration: relocate processes to redistribute workload

Ø Data migration: move data, based on access patterns

Ø Distributed scheduling: across processors

·    Real-time scheduling: difficult without global view, network delays make task harder

·    Performance modeling and analysis: Network latency to access resources must be reduced

Ø Metrics: theoretical measures for algorithms, practical measures for systems

Ø Measurement methodologies and tools

Applications and Emerging Challenges (1)

·    Mobile systems

Ø Wireless communication: unit disk model; broadcast medium (MAC), power management etc.

Ø CS perspective: routing, location management, channel allocation, localization and position estimation, mobility management

Ø Base station model (cellular model)

Ø Ad-hoc network model (rich in distributed graph theory problems)

·    Sensor networks: Processor with electro-mechanical interface

·    Ubiquitous or pervasive computing

Ø Processors embedded in and seamlessly pervading environment

Ø Wireless sensor and actuator mechanisms; self-organizing; network-centric, resource-constrained

Ø E.g., intelligent home, smart workplace

Applications and Emerging Challenges (2)

·    Peer-to-peer computing

Ø No hierarchy; symmetric role; self-organizing; efficient object storage and lookup;scalable; dynamic reconfig 

·    Publish/subscribe, content distribution

Ø Filtering information to extract that of interest

·    Distributed agents

Ø  Processes that move and cooperate to perform specific tasks; coordination, controlling mobility, software design and interfaces

·    Distributed data mining

Ø Extract patterns/trends of interest

Ø Data not available in a single repository

Applications and Emerging Challenges (3)

·    Grid computing

Ø Grid of shared computing resources; use idle CPU cycles

Ø Issues: scheduling, QOS guarantees, security of machines and jobs 

·    Security

Ø Confidentiality, authentication, availability in a distributed setting

Ø Manage wireless, peer-to-peer, grid environments

§ Issues: e.g., Lack of trust, broadcast media, resource-constrained, lack of  structure

Chapter 2: A Model of Distributed Computations

A Distributed Program

·    A distributed program is composed of a set of n asynchronous processes, p1, p2, ..., pi , ..., pn.

·    The processes do not share a global memory and communicate solely by passing messages.

·    The processes do not share a global clock that is instantaneously accessible to these processes.

·    Process execution and message transfer are asynchronous.

·    Without loss of generality, we assume that each process is running on a different processor.

·    Let Cij denote the channel from process pi to process pj and let mij denote a message sent by pi to pj.

·    The message transmission delay is finite and unpredictable.

A Model of Distributed Executions

·    The execution of a process consists of a sequential execution of its actions.

·    The actions are atomic and the actions of a process are modeled as three types of events, namely, internal events, message send events, and message receive events.

·   Let e denote the xth event at process pi.

·    For a message m, let send(m) and rec(m) denote its send and receive events, respectively.

·    The occurrence of events changes the states of respective processes and channels.

·    An internal event changes the state of the process at which it occurs.

·    A send event changes the state of the process that sends the message and the state of the channel on which the message is sent.

·    A receive event changes the state of the process that receives the message and the state of the channel on which the message is received.

A Model of Distributed Executions

·    The events at a process are linearly ordered by their order of occurrence

The execution of process pi produces a sequence of events e, e…., e, e_i^x+1,… and  is denoted by H_i where

                             Hi = (hi, →i)

h_i is the set of events produced by pi and

binary relation →_i defines a linear order on these events.

·    Relation →_i expresses causal dependencies among the events of pi.

A Model of Distributed Executions

·    The send and the receive events signify the flow of information between processes and establish causal dependency from the sender process to the receiver process.

·    A relation →msg that captures the causal dependency due to message exchange, is defined as follows. For every message m that is exchanged between two processes, we have                           send(m) →msg rec(m).

·    Relation →_msg defines causal dependencies between the pairs of corresponding send and receive events.

A Model of Distributed Executions

·    The evolution of a distributed execution is depicted by a space-time diagram.

·    A horizontal line represents the progress of the process; a dot indicates an event; a slant arrow indicates a message transfer.

·  Since we assume that an event execution is atomic (hence, indivisible, and instantaneous), it is justified to denote it as a dot on a process line.

·    In the Figure 2.1, for process _p1, the second event is a message send event, the third event is an internal event, and the fourth event is a message receive event.

A Model of Distributed Executions

Figure 2.1: The space-time diagram of a distributed execution

A Model of Distributed Executions

Causal Precedence Relation

·    The execution of a distributed application results in a set of distributed events produced by the processes.

·    Let H=∪_ih_i denote the set of events executed in a distributed computation.

Define a binary relation → on the set H as follows that expresses causal dependencies between events in the distributed execution.

·The causal precedence relation induces an irreflexive partial order on the events of a distributed computation that is denoted as H=(H, →).

A Model of Distributed Executions

. . . Causal Precedence Relation

·    Note that the relation → is nothing but Lamport’s “happens before” relation.

·    For any two events e_i and e_j, if e_i → e_j, then event e_j is directly or transitively dependent on event e_i. (Graphically, it means that there exists a path consisting of message arrows and process-line segments (along increasing time) in the space-time diagram that starts at e_i and ends at e_j.

For example, in Figure 2.1, e → e  and e  →e

·    The relation → denotes flow of information in a distributed computation and ei → ej dictates that all the information available at ei is potentially accessible at ej.

·   For example, in Figure 2.1, event e has the knowledge of all other events shown in the figure.

A Model of Distributed Executions

. . . Causal Precedence Relation

·    For any two events ei and ej, ei /→ ej denotes the fact that event ej does not directly or transitively dependent on event ei. That is, event ei does not causally affect event ej.

·    In this case, event ej is not aware of the execution of ei or any event executed after ei on the same process.

A Model of Distributed Executions

Concurrent events

 

A Model of Distributed Executions

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.

·        Two or more events may be logically concurrent even though they do not occur at the same instant in physical time.

·        However, if processor speed and message delays would have been different, the execution of these events could have very well coincided in physical time.

·        Whether a set of logically concurrent events coincide in the physical time or not, does not change the outcome of the computation.

·        Therefore, even though a set of logically concurrent events may not have occurred at the same instant in physical time, we can assume that these events occured at the same instant in physical time.

Models of Communication Networks

·        There are several models of the service provided by communication networks, namely, FIFO, Non-FIFO, and causal ordering.

·        In the FIFO model, each channel acts as a first-in first-out message queue and thus, message ordering is preserved by a channel.

·        In the 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.

Models of Communication Networks

·    The “causal ordering” model is based on Lamport’s “happens before” relation.

·    A system that supports the causal ordering model satisfies the following property:

·    This property ensures that causally related messages destined to the same destination are delivered in an order that is consistent with their causality relation.

·    Causally ordered delivery of messages implies FIFO message delivery. (Note that CO ⊂ FIFO ⊂ Non-FIFO.)

·    Causal ordering model considerably simplifies the design of distributed algorithms because it provides a built-in synchronization.

Global State of a Distributed System

“A collection of the local states of its components, namely, the processes and the communication channels.”

·    The state of a process is defined by the contents of processor registers, stacks, local memory, etc. and depends on the local context of the distributed application.

·    The state of channel is given by the set of messages in transit in the channel. 

·    The occurrence of events changes the states of respective processes and channels.

·    An internal event changes the state of the process at which it occurs.

·    A send event changes the state of the process that sends the message and the state of the channel on which the message is sent.

·    A receive event changes the state of the process that or receives the message and the state of the channel on which the message is received.

. . . Global State of a Distributed System

Notations

. . . Global State of a Distributed System

A Channel State

    The state of a channel depends upon the states of the processes it connects.

. . . Global State of a Distributed System

Global State

·    The global state of a distributed system is a collection of the local states of the processes and the channels.

·    Notationally, global state GS is defined as,

For a global state 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 if there were a global system clock that can be instantaneously read by the processes. (However, both are impossible.)

. . . Global State of a Distributed System

A Consistent Global State

·    Even if the state of all the components is not recorded at the same instant, such a state will be meaningful provided every message that is recorded as received is also recorded as sent.

·    Basic idea is that a state should not violate causality – an effect should not be present without its cause. A message cannot be received if it was not sent.

·    Such states are called consistent global states and are meaningful global states.

·    Inconsistent global states are not meaningful in the sense that a distributed system can never be in an inconsistent state.

·  

· .....Global State of a Distributed System

An Example

Consider the distributed execution of Figure 2.2.

                     Figure 2.2: The space-time diagram of a distributed execution.

. . . Global State of a Distributed System

In Figure 2.2:

·   because the state of p2 has recorded the receipt of message m₁₂, however, the state of p₁ has not recorded its send.

·    A global state GS₂ consisting of local states

       is consistent; all the channels are empty except C₂₁ that contains message m₂₁.

Cuts of a Distributed Computation

“In the space-time diagram of a distributed computation, a cut is a zigzag line joining one arbitrary point on each process line.”

·    A cut slices the space-time diagram, and thus the set of events in the distributed computation, into a PAST and a FUTURE.

·    The PAST contains all the events to the left of the cut and the FUTURE contains all the events to the right of the cut.

·    For a cut C, let PAST(C) and FUTURE(C) denote the set of events in the PAST and FUTURE of C, respectively.

·    Every cut corresponds to a global state and every global state can be graphically represented as a cut in the computation’s space-time diagram.

·    Cuts in a space-time diagram provide a powerful graphical aid in representing and reasoning about global states of a computation.

. . . Cuts of a Distributed Computation

                       Figure 2.3: Illustration of cuts in a distributed execution.

. . . Cuts of a Distributed Computation

·    In a consistent cut, every message received in the PAST of the cut was sent in the PAST of that cut. (In Figure 2.3, cut C₂ is a consistent cut.)

·    All messages that cross the cut from the PAST to the FUTURE are in transit in the corresponding consistent global state.

·    A cut is inconsistent if a message crosses the cut from the FUTURE to the PAST. (In Figure 2.3, cut C₁ is an inconsistent cut.)

Past and Future Cones of an Event

Past Cone of an Event

·    An event e_j could have been affected only by all events e_i such that e_i → e_j

·    In this situtaion, all the information available at ei could be made accessible at e_j.

·    All such events ei belong to the past of e_j.

Let Past(e_j) denote all events in the past of ej in a computation (H, →). Then,

·    Figure 2.4 (next slide) shows the past of an event e_j

. . . Past and Future Cones of an Event

. . . Past and Future Cones of an Event

Models of Process Communications

·    There are two basic models of process communications – synchronous and asynchronous.

·    The synchronous communication model is a blocking type where on a message send, the sender process blocks until the message has been received by the receiver process.

·    The sender process resumes execution only after it learns that the receiver process has accepted the message.

·    Thus, the sender and the receiver processes must synchronize to exchange a message. On the other hand, asynchronous communication model is a non-blocking type where the sender and the receiver do not synchronize to exchange a message.

·    After having sent a message, the sender process does not wait for the message to be delivered to the receiver process.

·    The message is bufferred by the system and is delivered to the receiver process when it is ready to accept the message.

. . . Models of Process Communications

·    Neither of the communication models is superior to the other.

·    Asynchronous communication provides higher parallelism because the sender process can execute while the message is in transit to the receiver.

·    However, A buffer overflow may occur if a process sends a large number of messages in a burst to another process.

·    Thus, an implementation of asynchronous communication requires more complex buffer management.

·    In addition, due to higher degree of parallelism and non-determinism, it is much more difficult to design, verify, and implement distributed algorithms for asynchronous communications.

·    Synchronous communication is simpler to handle and implement.

·    However, due to frequent blocking, it is likely to have poor performance and is likely to be more prone to deadlocks.

Assignment 1: Basics

1.     What do you understand by the term “models” for distributed systems?

2. Redesign the distributed search algorithm (from slides) for the model – “Completely connected, synchronous, unreliable”

3. Redesign the distributed search algorithm (from slides) for the model – “Connected in ring topology, asynchronous, reliable”

4. Create a scenario where the atomicity issue in distributed systems can be understood.

5.     Why two clocks cannot be perfectly synchronized?

6. To address faults, we can use replication – what kind of problems/challenges this approach can cause?

7.     Identify some distributed systems in real life that you interact with. How do you know, they are distributed systems?

8.     What do you mean by weakening a model? Give example.

9.Discuss the advantages of distributed systems. Explain with clear examples why they are hard to implement.

10.  What is the use of replication in distributed systems? How is it related to fault tolerance.

11.  Explain with example the issue of global clocks in distributed systems.

12.  Explain with example the issue of global memory in distributed systems.

13.What are the difference between synchronous and asynchronous models?

14.If synchronous network is better than asynchronous networks, then why do we even consider the other one?

15.Explain with figures, what do you mean by different topologies?

16.Explain with example various complexity measures for distributed systems.