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

Distributed System

A distributed system is a collection of independent (autonomous) computers, connected by a network and equipped with distribution middleware, that appears to its users as a single coherent system. The components coordinate their actions only by passing messages, as they have no shared memory or global clock.

Characteristics

• Resource sharing of hardware, data and services across nodes.
• Concurrency — many users and processes execute simultaneously.
• No global clock — coordination is achieved purely by message passing.
• Independent failures — components can fail independently and partially.

Key Challenges in Design

Client–Server vs Peer-to-Peer Model

Client–Server model (description of diagram): several clients send request messages to one (or a few) dedicated server(s); the server processes the request and returns a reply. Roles are fixed and asymmetric.

  Client1 ---req--->  +--------+
  Client2 ---req--->  | Server | ---> shared resource/DB
  Client3 <--reply--  +--------+

Peer-to-Peer model (description of diagram): all nodes (peers) are functionally equivalent; each acts as both client and server, sharing resources directly with other peers without a central server.

  Peer A <----> Peer B
     ^   \    /   ^
     |    \  /    |
     v     \/     v
  Peer D <----> Peer C

Practical applications: Client–server → World Wide Web / a web mail service. Peer-to-peer → BitTorrent file distribution.

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

Need for clock synchronization: Each computer has its own physical clock that drifts; there is no global clock. Yet many distributed tasks need a common notion of time — ordering events, timestamping transactions, cache/lease expiry, make/file-timestamp consistency, and authentication (e.g. Kerberos ticket validity). Hence clocks must be kept close to each other (internal sync) or to UTC (external sync).

Cristian's Algorithm

A client synchronizes with a time server holding an accurate (UTC) clock:

1. Client records local send time $T_0$ and sends a request to the time server.
2. Server inserts its current time $T_s$ in the reply.
3. Client receives the reply at local time $T_1$.

The round-trip time is $RTT = T_1 - T_0$. Assuming the request and reply paths take roughly equal time, the message took about $RTT/2$ to arrive, so the client sets its clock to:

$$T_{client} = T_s + \frac{T_1 - T_0}{2}$$

Error bound: Let $T_{min}$ be the minimum one-way transit time. The server's timestamp was generated somewhere between $T_0 + T_{min}$ and $T_1 - T_{min}$, so the uncertainty (accuracy) is bounded by:

$$\pm\left(\frac{RTT}{2} - T_{min}\right)$$

A smaller and more symmetric $RTT$ gives a tighter bound; taking the reply with the smallest $RTT$ over several trials improves accuracy.

(b) Lamport's Happened-Before Relation [5]

Definition — the happened-before relation $\rightarrow$ is the smallest relation satisfying:

1. If $a$ and $b$ are events in the same process and $a$ comes 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 timestamps (rules): each process keeps a counter $C$. Before every event $C := C+1$. A message carries its sender's timestamp $t$; on receipt the receiver sets $C := \max(C, t) + 1$.

Example: Process P1 events a, b, c get $1,2,3$. If b sends a message to event d in P2 (where P2's clock was 1), then $C(d) = \max(1,2)+1 = 3$. The clock guarantees: if $a \rightarrow b$ then $C(a) < C(b)$.

Why logical clocks cannot capture causality: The implication is one-way. $a \rightarrow b \Rightarrow C(a) < C(b)$, but the converse is false: $C(a) < C(b)$ does not imply $a \rightarrow b$ — the two events may be concurrent yet receive ordered timestamps. So scalar timestamps cannot tell whether two events are causally related or independent.

Motivation for vector clocks: A vector clock $V$ keeps one counter per process. Then $a \rightarrow b \iff V(a) < V(b)$ (component-wise, with at least one strictly smaller), and $a \parallel b$ when neither vector dominates. Thus vector clocks fully and exactly characterise causality, which scalar Lamport clocks cannot.

(a) Ricart–Agrawala Distributed Mutual Exclusion [7]

This is a timestamp-based, fully distributed algorithm using Lamport logical clocks among $N$ processes.

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

1. Sets its state to Wanted, builds a timestamped REQUEST $(T_i, i)$ and sends it 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 a REQUEST $(T_j, j)$, process $P_i$:

• If $P_i$ is in the CS, it defers the reply (queues it).
• If $P_i$ is also Wanted, it compares timestamps: it replies OK if the incoming request has the lower $(T,id)$ pair, otherwise it defers.
• If $P_i$ is Released (not interested), it replies OK immediately.

On exiting the CS: $P_i$ sends the deferred OK replies to all queued processes.

Message complexity: $2(N-1)$ messages per CS entry — $(N-1)$ REQUESTs + $(N-1)$ REPLYs.

Comparison

Ricart–Agrawala removes the central bottleneck but costs more messages and tolerates failures worse, because every process must respond.

(b) Bully Election Algorithm [5]

Used to elect the process with the highest id as coordinator.

Steps: When a process $P$ notices the coordinator has crashed, it sends an ELECTION message to all processes with higher ids. If none replies, $P$ wins and broadcasts a COORDINATOR message. If a higher process replies with OK, $P$ drops out and that higher process restarts the election among still-higher processes.

Example (processes 1–5, coordinator 5 crashes):

• Suppose process 2 detects the crash. It sends ELECTION to 3, 4, 5.
• 3 and 4 reply OK (5 is dead); 2 steps down.
• 3 sends ELECTION to 4, 5; 4 replies OK; 3 steps down.
• 4 sends ELECTION to 5; no reply (5 dead). 4 wins and sends COORDINATOR to 1, 2, 3.

Process 4 becomes the new coordinator.

Worst-case messages: The worst case occurs when the lowest-id process detects the failure, giving $O(N^2)$ messages — approximately $\sum (\text{election} + \text{OK} + \text{coordinator})$ on the order of $N^2$ messages.

Sun Network File System (NFS)

NFS allows a client to access files stored on a remote server transparently, as if they were local, over a network using RPC.

Design Goals

• Access transparency — same system calls (open, read, write) work on remote files.
• OS and hardware independence — protocol defined via an external IDL (XDR).
• Crash recovery — a stateless server simplifies recovery.
• Reasonable performance with caching, and good portability.

Architecture

Virtual File System (VFS): An abstraction layer in the kernel that presents a uniform file-system interface. It maintains a vnode for every open file. The VFS decides, per request, whether to route the operation to the local file system or to the NFS client, which issues an RPC to the remote server. This makes the location of the file invisible to applications.

Stateless server design: The NFS server keeps no per-client state (no open-file table, no record of past requests). Every request is self-contained, carrying a file handle and an explicit byte offset. Benefits: if the server crashes and restarts, the client simply retries — no recovery protocol is needed; operations are made idempotent. The drawback is that true open/close semantics and server-enforced locking are hard (handled by a separate lock manager).

Mount protocol: A client mounts a remote exported directory into its own namespace. The client sends the pathname to the server's mount service, which returns a file handle for the root of that subtree. Thereafter the client uses this handle. Mounting can be done at boot or on demand (automounter).

NFS vs Andrew File System (AFS)

Summary: NFS favours simplicity and crash-resilience via statelessness but generates more validation traffic; AFS uses whole-file caching with server callbacks to greatly reduce server interaction, giving much better scalability at the cost of a stateful server.

Remote Procedure Call (RPC)

RPC lets a program call a procedure located on another machine as if it were a local call, hiding the message passing from the programmer.

Working (description of diagram)

 Client process                         Server process
  call foo(x)                            foo(x){...}
      |                                       ^
      v                                       |
  [Client Stub] --marshal--> network --> [Server Stub] --unmarshal--> server proc
      ^                                       |
      |  <--unmarshal-- network <--marshal--  v
  return value                           return result

1. The client calls a local client stub (proxy) with parameters.
2. The client stub marshals the parameters into a message (packs them into a standard byte format) and sends it to the server.
3. The server stub receives and unmarshals the message, extracting the parameters.
4. The server stub calls the actual server procedure.
5. The result is marshalled by the server stub, sent back, unmarshalled by the client stub, and returned to the caller.

Marshalling = converting parameters/results into a flat, machine-independent transmission format; unmarshalling = the reverse.

RPC Call Semantics

Maybe semantics (no retransmission) also exists but is unreliable and rarely used.

TCP vs UDP Interprocess Communication

Request–Reply Protocol

The basic protocol underlying RPC/RMI. The client sends a request message (operation + arguments) and blocks; the server executes and returns a reply. Often a third acknowledge-reply is added. It is typically built over UDP for efficiency, with reliability added at the application level.

Problems in IPC and their Handling

• Message loss: A request or reply may be dropped. Handled by timeouts and retransmission of the request.
• Duplication: Retransmission can cause the server to receive the same request twice. Handled by attaching a unique request ID; the server keeps a history/cache to detect and discard duplicates and re-send the saved reply (giving at-most-once semantics).
• Ordering: Messages may arrive out of order over a connectionless transport. Handled by sequence numbers so the receiver can reorder or discard out-of-order messages.

TCP solves all three at the transport layer; with UDP these mechanisms must be implemented in the request-reply layer itself.

Replication

Replication is maintaining multiple copies (replicas) of data or a service on different machines to improve reliability/fault tolerance, availability, and performance (load sharing and lower latency).

Active vs Passive Replication

Active replication (state-machine replication)

• Every replica is a full server and processes every client request in the same order (using a totally-ordered, reliable multicast).
• All replicas independently compute and produce identical results; the client takes the first or majority response.
• Replica failure handling: Because every replica does the work, the failure of one replica is transparent — the remaining replicas still produce the result; no failover delay. It can even mask Byzantine faults using voting (with enough replicas).

Passive (primary-backup) replication

• One replica is the primary; it handles all requests, updates its state, and propagates the new state to one or more backups, then replies to the client.
• Backups stay idle, only receiving state updates.
• Replica failure handling: If a backup fails, service continues unaffected. If the primary fails, a backup is promoted to primary (failover/election); requests in flight may need to be re-sent, causing a short interruption. Tolerates crash (fail-stop) faults but not Byzantine faults.

Token-Ring Algorithm for Mutual Exclusion

The $N$ processes are arranged in a logical ring, each knowing its successor. A single token circulates around the ring in one direction.

Rule: A process may enter the critical section (CS) only if it holds the token. After leaving the CS (or if it does not need the CS), it passes the token to its successor. This guarantees mutual exclusion (only one token) and freedom from starvation (token visits everyone in bounded time).

Message cost: between 1 and infinitely many messages per entry, since the token keeps circulating even when no one needs the CS.

Failure Cases and Recovery

(a) Token is lost (e.g. dropped on a link or by a crash while in transit): No process can ever enter the CS — the system deadlocks. Recovery: Detect the loss using a timeout (no token seen for too long) and regenerate a single token. A monitor/coordinator or a leader election (e.g. Bully) is used to ensure exactly one new token is created, avoiding duplicate tokens.

(b) Process holding the token crashes: The token is lost and the ring is broken — the dead process can neither use nor forward the token, and its predecessor cannot reach its successor. Recovery: Detect the crash (acknowledged token hand-off: the sender keeps the token until the successor acknowledges receipt, so it can retry). Reconfigure the ring to bypass the dead node by connecting its predecessor to the next live successor, then regenerate the token via election. Acknowledged hand-off plus ring repair restores operation.

Berkeley Algorithm (Internal Synchronization)

The Berkeley algorithm keeps a group of machines' clocks mutually consistent without assuming any machine has accurate UTC.

Steps

1. One machine is chosen as the master (coordinator); the rest are slaves.
2. The master polls every slave, asking for its current time.
3. Each slave replies; the master estimates each slave's time accounting for round-trip delay.
4. The master computes the average of all the clock values (its own included), often discarding outliers.
5. Instead of sending absolute times, the master sends each machine the adjustment (offset) it must apply to its clock to match the average.

Example

Master reads 3:00:00. Slaves report 2:59:50 and 3:00:10 (after RTT correction). Average $= (3{:}00{:}00 + 2{:}59{:}50 + 3{:}00{:}10)/3 = 3{:}00{:}00$. Master tells slave 1 to advance by +10 s, slave 2 to set back by −10 s, and itself by 0. All clocks now agree at 3:00:00. (Clocks are slowed/sped gradually, never set backward abruptly.)

Difference from Cristian's Algorithm

Why suitable without UTC: Berkeley only aims to keep clocks close to each other, using their average as the reference. It therefore works perfectly in an isolated network where no machine has access to a UTC/atomic source.

Remote Method Invocation (RMI)

RMI extends the RPC concept to objects: it lets an object in one process (JVM) invoke methods on a remote object living in another process/machine, as if it were local. It is object-oriented and supports passing object references and exceptions.

Key Components

• Proxy (stub): A local stand-in object on the client that implements the same remote interface. The client calls the proxy; it marshals the method id and arguments, forwards them to the remote object, and unmarshals the result. It makes the remote object look local.
• Remote Reference Module: Translates between local and remote object references. It maintains the mapping (remote reference ↔ local object via a remote object table) and is responsible for creating remote references and routing invocations to the correct object.
• Object Registry (e.g. rmiregistry): A naming/lookup service. The server binds a remote object to a name; clients look up that name to obtain a remote reference (proxy). This is how a client first locates a remote object.

A skeleton (server-side stub) unmarshals the request and dispatches it to the actual object.

RMI vs Conventional RPC

In short, RMI is the object-oriented evolution of RPC.

Types of Transparency in a Distributed System

Transparency means hiding the distributed nature of the system from users and programmers, so it appears as a single coherent system. Four important types:

1. Access transparency — Local and remote resources are accessed using identical operations. A user reads a remote file with the same open/read calls used for a local file; differences in data representation and access mechanism are hidden.

2. Location transparency — Resources are accessed without knowing their physical (network/host) location. A URL or logical name identifies the resource regardless of which server holds it.

3. Replication transparency — Multiple copies (replicas) of a resource may exist for reliability and performance, but the user sees a single logical resource and is unaware that replication exists or which replica is used.

4. Failure transparency — Faults in components are hidden (masked); the system continues to function (via retries, replication) so users and applications can complete their work despite partial failures.

(Others: concurrency transparency — concurrent users share resources without interference; migration transparency — resources can move without affecting access.)