BE Computer Engineering (IOE, TU) Distributed System (IOE, CT 702 / ENCT 411) Question Paper 2078 Nepal

Distributed System

A distributed system is a collection of independent (autonomous) computers connected by a network that appears to its users as a single coherent system. The computers coordinate by passing messages and share resources without a global clock or shared memory.

Key Characteristics

• Concurrency of components — multiple machines execute simultaneously and share resources.
• No global clock — coordination relies on message passing, not a common physical time.
• Independent failures — components can fail individually without halting the whole system.
• Resource sharing — hardware, data, and services are shared across nodes.
• Transparency — the distribution is hidden from users.

Architectural Models

Client–Server Model

  Client ---- request ----> [ Server ]
  Client <--- response ----  (resource/service)

A central server holds resources/services; clients send requests and receive responses. Roles are fixed and asymmetric.

Peer-to-Peer (P2P) Model

   Peer <----> Peer
     ^   \    /   ^
     |    \  /    |
   Peer <-- X --> Peer   (every node is both client and server)

All nodes (peers) are functionally equivalent — each acts as both a client and a server, sharing its own resources directly with others.

Comparison

Major Design Challenges

• Heterogeneity — diverse hardware, OS, networks, and languages; addressed by middleware and common standards/protocols (e.g., XML, IDL).
• Openness — the system can be extended and re-implemented via published, standard interfaces.
• Scalability — performance must remain acceptable as the number of users/resources grows; achieved through replication, caching, and avoiding centralized bottlenecks.
• Transparency — hiding distribution: access, location, concurrency, replication, failure, migration, performance, and scaling transparency.
• Fault tolerance, security, and concurrency control are further key concerns.

(a) Clock Synchronization and Cristian's Algorithm [6]

Why synchronization is needed: Each node has its own physical clock that drifts at a different rate, so clocks gradually diverge. Many distributed tasks — ordering events, timestamps, make-style consistency checks, cache expiry, authentication tickets — require nodes to agree on time. Hence clocks must be periodically synchronized.

Cristian's Algorithm

A passive time server $S$ holds the reference time (e.g., from UTC). A client $C$ synchronizes as follows:

1. At time $T_0$ (by its own clock), $C$ sends a request to $S$.
2. $S$ receives it, reads its time $T_{server}$, and replies with $T_{server}$.
3. $C$ receives the reply at $T_1$ (its own clock).

The round-trip time is $RTT = T_1 - T_0$. Assuming the request and reply legs take equal time, the one-way propagation delay is estimated as $\frac{T_1 - T_0}{2}$. The client therefore sets its clock to:

$$T_{new} = T_{server} + \frac{T_1 - T_0}{2}$$

Compensation: because the server's timestamp is already $\frac{RTT}{2}$ old by the time it arrives, adding half the round-trip delay corrects for message propagation. If the minimum one-way transmission time $T_{min}$ is known, accuracy is bounded by $\pm\left(\frac{RTT}{2} - T_{min}\right)$. The clock should be adjusted gradually (slewing) rather than set backward, to avoid time running in reverse.

(b) Lamport's Happened-Before and Logical Clocks [6]

Happened-before relation ($\rightarrow$): the smallest relation satisfying:

1. If $a$ and $b$ are events in the same process and $a$ occurs before $b$, then $a \rightarrow b$.
2. If $a$ is the send of a message and $b$ is its receive, then $a \rightarrow b$.
3. Transitivity: if $a \rightarrow b$ and $b \rightarrow c$ then $a \rightarrow c$.

If neither $a \rightarrow b$ nor $b \rightarrow a$, the events are concurrent ($a \parallel b$).

Lamport clock rules: each process $P_i$ keeps a counter $C_i$.

• Before each event, $C_i := C_i + 1$.
• A message carries timestamp $t = C_i$; on receipt, $C_j := \max(C_j, t) + 1$.

Worked example — three processes; arrows = messages:

For every message, the send timestamp is strictly less than the receive timestamp (e.g., m1: $2 < 3$; m2: $5 < 6$), so the property $a \rightarrow b \Rightarrow C(a) < C(b)$ holds, enforcing causal ordering. (Note: the converse is not guaranteed — equal/ordered clock values do not imply causality; that requires vector clocks.)

(a) Ricart–Agrawala Algorithm [7]

A fully distributed, timestamp-based mutual exclusion algorithm (no coordinator). Each process keeps a Lamport clock.

To enter the critical section (CS), process $P_i$:

1. Sets its state to Wanted, builds timestamp $T_i$, and sends a REQUEST(T_i, i) to all other $N-1$ processes.
2. Waits until it has received an OK (reply) from all $N-1$ processes, then enters the CS.

On receiving REQUEST(T_j, j) at $P_i$:

• If $P_i$ is not interested in the CS → reply OK immediately.
• If $P_i$ is in the CS → queue the request (defer reply).
• If $P_i$ also wants the CS → compare timestamps; the lower $(T,\text{id})$ wins. If the requester wins, reply OK; otherwise defer.

On exiting the CS: send OK to all deferred (queued) requests.

Message Cost

Per CS entry: $N-1$ requests $+$ $N-1$ replies $= \mathbf{2(N-1)}$ messages.

Correctness

• Safety (mutual exclusion): two processes wanting the CS concurrently cannot both proceed, because the lower-timestamp request is granted first and the other's reply is deferred until exit. Lamport timestamps with process IDs break ties, giving a total order — only one can hold all OKs at once.
• Liveness (no deadlock/starvation): the timestamp total order ensures the oldest request is always served, so every request eventually succeeds; there is no circular wait.
• Fairness: requests are honored in increasing timestamp order (FCFS by logical time), so no process can be indefinitely overtaken.

(b) Bully Election Algorithm [5]

Used to elect a coordinator (highest process ID) when the current one fails. When $P_i$ notices the coordinator is unresponsive:

1. $P_i$ sends an ELECTION message to all processes with higher IDs.
2. If no higher process replies (with OK/ALIVE) within a timeout, $P_i$ wins and sends COORDINATOR to all → it becomes the new coordinator.
3. If a higher process replies, $P_i$ drops out; that higher process now runs its own election.

Example — processes $P_1 \dots P_7$; coordinator $P_7$ crashes. Suppose $P_4$ detects this:

• $P_4$ sends ELECTION to $P_5, P_6, P_7$.
• $P_5$ and $P_6$ reply OK (so $P_4$ steps down) and each start their own elections.
• $P_6$ sends ELECTION to $P_7$; $P_7$ is down, no reply.
• $P_6$ gets no higher reply → $P_6$ becomes coordinator and broadcasts COORDINATOR to all lower processes.

The highest live process always "bullies" the others into submission. Worst-case message complexity is $O(N^2)$.

Remote Procedure Call (RPC)

RPC lets a program call a procedure on a remote machine as if it were local, hiding the message passing. It provides access and location transparency.

Components and Flow

  Client                                    Server
  ------                                    ------
  call f(args)                              real f(args)
     |                                         ^
     v                                         |
 [Client Stub] --marshal--> network --> [Server Stub/Skeleton]
     ^                                         |
     | <--unmarshal-- network <--marshal-- (result)
  RPC Runtime  <===== messages =====>  RPC Runtime

1. Client invokes the local client stub (proxy) like an ordinary procedure.
2. The client stub marshals the parameters into a message and asks the RPC runtime to send it.
3. The runtime transmits the request over the network to the server's runtime.
4. The server stub (skeleton) unmarshals the parameters and calls the real server procedure.
5. The result is marshalled back through the skeleton → runtime → network → client stub, which unmarshals it and returns the value to the client.

Parameter Marshalling

Marshalling is packing parameters/results into a flat, machine-independent message; unmarshalling reverses it. It handles differences in byte order (big- vs little-endian), data representation, and alignment, often using a standard format (e.g., XDR). Pointers cannot be passed directly, so call-by-reference is usually emulated by copy/restore of the referenced data.

Call Semantics

• Maybe — no retransmission; the call may or may not execute (used when failures are acceptable; weakest guarantee).
• At-least-once — the client retransmits on timeout, so the call executes one or more times. Safe only for idempotent operations.
• At-most-once — duplicates are filtered (via request IDs), so the call executes exactly zero or one time; the strongest practical guarantee.

RMI vs RPC

In short, RMI is the object-oriented analogue of RPC: it extends the same client-stub/server-skeleton mechanism to method invocation on remote objects with object serialization.

Synchronous vs Asynchronous Communication

• Synchronous: sender and receiver are tightly coupled in time — the sending and receiving operations are coordinated. The sender blocks until the message is received (and often acknowledged); both must be ready together.
• Asynchronous: the send operation is decoupled — the sender continues without waiting for the receiver. Messages may be buffered, so sender and receiver need not be active simultaneously.

Blocking vs Non-Blocking Primitives

• Blocking send: the sender is suspended until the message has been transmitted (or buffered/delivered). Blocking receive: the receiver waits until a message arrives.
• Non-blocking send: control returns immediately after the message is copied to a buffer; the process continues, checking completion later. Non-blocking receive: returns at once whether or not a message is available (the process polls or is signaled later).

Blocking primitives are simpler and safer; non-blocking primitives give higher concurrency but require care (buffer must not be reused until the send completes).

Sockets and the Message-Passing Model

A socket is an endpoint of communication identified by an (IP address, port) pair. Two sockets, one per process, form a channel over which messages are exchanged — directly realizing message passing.

• TCP-based (stream) sockets: connection-oriented and reliable. A connection is established (3-way handshake); data is delivered as an ordered, error-checked byte stream with retransmission and flow control. Used when reliability matters (file transfer, HTTP).
• UDP-based (datagram) sockets: connectionless and unreliable. Independent datagrams are sent with no setup, no ordering, and no delivery guarantee, giving lower latency/overhead. Used for DNS, streaming, and latency-sensitive messaging.

Distributed File System (DFS) Architecture

A DFS lets files stored on remote servers be accessed by clients across a network as if they were local, providing location and access transparency. Typical components:

• Client module — intercepts file operations and forwards remote ones.
• Flat file / storage service — stores file contents on servers.
• Directory (name) service — maps human-readable names to file identifiers.

Sun NFS

NFS is a widely used DFS based on the client–server model using RPC over (originally) UDP/TCP.

Virtual File System (VFS)

The VFS is an abstraction layer in the OS kernel that gives a uniform interface for both local and remote files. It distinguishes local files from NFS-mounted files using vnodes: each open file is represented by a vnode that points either to a local inode or to a remote NFS file (via a file handle). This lets applications use the same system calls regardless of where the file physically lives, achieving access transparency.

Stateless Servers

NFS servers (classically NFSv2/v3) are stateless — the server keeps no record of which clients have which files open. Each request (e.g., read with an explicit offset and file handle) is self-contained and idempotent.

• Advantage: crash recovery is trivial — after a server reboot, clients simply retry; there is no per-client state to rebuild.
• Trade-off: the server cannot easily support file locking on its own (handled by a separate lock manager).

Caching for Performance

• Client-side caching of file data blocks and attributes reduces repeated network round-trips and server load.
• Read-ahead and delayed write improve throughput.
• Cached attributes are revalidated using timestamps/timeouts (validation period), and writes are flushed periodically, balancing performance against consistency (NFS offers approximate, not strict, consistency).

Replication

Replication is maintaining multiple copies (replicas) of data or services on different machines. It is used to:

• Improve reliability / fault tolerance — if one replica fails, others continue.
• Improve performance / availability — requests are served from a nearby or less-loaded replica, increasing throughput and reducing latency.

The central challenge is keeping replicas consistent.

Active Replication

Every replica is a full server and processes every client request independently in the same order (using totally-ordered multicast). The client (or a front end) collects responses and uses the majority/first result.

• Handling failure: because all replicas execute every request, the crash of one replica is masked automatically — surviving replicas already have identical state and continue without any takeover delay. It can even tolerate Byzantine failures via voting (with enough replicas).

Passive (Primary–Backup) Replication

One replica is the primary; it handles all requests and, before responding, propagates its state updates to one or more backups. Backups stay idle, only receiving updates.

• Handling failure: if a backup fails, nothing changes for clients. If the primary fails, the remaining replicas elect a new primary from the backups, which takes over using the last received state. This introduces a short fail-over delay, and updates not yet propagated may be lost (tolerates only crash, not Byzantine, failures).

Comparison

Berkeley Algorithm

The Berkeley algorithm is an internal clock synchronization method that brings a set of clocks into agreement without a reference UTC source.

Procedure

1. A coordinator (master/time daemon) is chosen; it polls all slaves periodically for their clock values.
2. The master estimates each slave's time (accounting for round-trip delay, like Cristian's) and computes the average of all clocks, discarding outliers (clocks differing too much).
3. Instead of sending the absolute time, the master sends each machine an adjustment (offset) — how much to advance or retard its clock — and adjusts its own.

Clocks are slewed gradually so time never jumps backward.

Difference from Cristian's Algorithm

When More Appropriate

Berkeley is better when no machine has an accurate UTC source and the goal is merely that all nodes agree with each other (internal consistency), e.g., an isolated LAN.

Example — master polls three slaves

Suppose master time = 3:00:00. After delay-corrected polling it gets: master 3:00:00, slave A 3:00:25, slave B 2:59:50, slave C 3:00:15.

• Sum of offsets relative to master $= 0 + 25 + (-10) + 15 = 30$ s over 4 clocks → average offset $= +7.5$ s, so target time $\approx 3:00:07.5$.
• Master sends adjustments: master +7.5 s, A −17.5 s, B +17.5 s, C −7.5 s.

All four clocks converge to the agreed average time.

Mutual Exclusion Approaches Compared

Comparison Table

Centralized

A single coordinator grants the token/permission.

• Advantage: simple, fair (FCFS), and minimal messages (only 3 per entry).
• Disadvantage: the coordinator is a single point of failure and a performance bottleneck.

Token-Ring

Processes are logically arranged in a ring; a token circulates and only its holder may enter the CS.

• Advantage: no starvation — every process gets the token in turn; correctness is easy to ensure.
• Disadvantage: if the token is lost or a process crashes, the ring breaks and must be regenerated; wasted messages when no one needs the CS.

Distributed (Ricart–Agrawala)

A process requests permission from all others using timestamps and enters when all reply.

• Advantage: no single coordinator — fully decentralized.
• Disadvantage: high message overhead ($2(N-1)$) and the crash of any process can stall the system (every process is a potential point of failure).

Fault Tolerance

Fault tolerance is the ability of a distributed system to continue providing correct service even when some of its components fail. A system is dependable if it remains available, reliable, safe, and maintainable despite faults.

Failure Types

• Crash (fail-stop) failure: a component halts completely and permanently; before failing it worked correctly, and after failing it does nothing (e.g., a server that stops responding).
• Omission failure: a component fails to perform some action — e.g., a send omission (server fails to send a reply) or receive omission (server never receives a request). Messages are lost.
• Timing failure: a (correct) response is produced outside its required time interval — too late (or too early). Relevant in real-time systems where deadlines matter.
• Byzantine (arbitrary) failure: the most severe — a component behaves arbitrarily/maliciously, may produce wrong or inconsistent results, send conflicting information to different nodes, or actively appear correct. Hardest to detect and tolerate.

Masking Failures with Redundancy

• Information redundancy: add extra bits (checksums, error-correcting codes) to detect/correct corrupted data.
• Time redundancy: repeat an operation (retransmit/retry); masks transient and omission failures.
• Physical (hardware/process) redundancy: replicate components — e.g., Triple Modular Redundancy (TMR) runs three units and votes on the output, masking a single faulty unit. Tolerating $k$ Byzantine faults requires $3k+1$ replicas.

By combining replication with voting and acknowledgements, failures are masked so the system as a whole continues to deliver correct results.

Ring (Chang–Roberts) Election Algorithm

Processes are arranged in a logical ring; each knows its successor. The algorithm elects the process with the highest ID as coordinator.

Procedure

1. A process noticing the coordinator has failed starts an election: it creates an ELECTION message containing its own ID and sends it to its successor (skipping dead nodes).
2. On receiving an ELECTION message, each process adds its ID (Chang–Roberts optimization: it forwards only if its own ID is higher; otherwise just relays the highest) and passes it on.
3. When the message returns to the initiator (it sees its own ID, now the maximum), that process knows the highest ID and sends a COORDINATOR message around the ring announcing the winner.

Example — ring of five processes

Ring order: $P_2 \to P_5 \to P_1 \to P_4 \to P_3 \to (P_2)$. Coordinator $P_5$ fails. Suppose $P_1$ detects it:

1. $P_1$ sends ELECTION(1) to $P_4$.
2. $P_4$: $4 > 1$ → forwards ELECTION(4) to $P_3$.
3. $P_3$: $4 > 3$ → forwards ELECTION(4) to $P_2$ (skipping dead $P_5$).
4. $P_2$: $4 > 2$ → forwards ELECTION(4) to $P_1$.
5. $P_1$ receives ID 4 (> its own 1) → forwards to $P_4$; $P_4$ sees its own ID → $P_4$ is the new coordinator and circulates COORDINATOR(4).

(With $P_5$ dead, $P_4$ — the highest surviving ID — wins.)

Message Complexity vs Bully

The Ring algorithm has lower (linear) message complexity and no message storms, whereas the Bully algorithm is $O(N^2)$ in the worst case but can elect a coordinator faster when the highest process detects the failure itself.