Distributed systems basics

In short

A distributed system is a collection of independent computers, connected by a network and without shared memory or a common clock, that cooperate to look to users like one coherent system. Everything hard about it comes from three facts: messages take unpredictable time, parts fail independently, and there is no global "now". It is a standard final-year elective (AICTE/CBCS "Distributed Systems", "Cloud Computing", "Advanced Operating Systems"), appears in GATE/UGC-NET through clocks, mutual exclusion, commit protocols and consistency, and dominates backend interviews at Indian product and services firms — UPI/NPCI-scale payment systems, Flipkart, Zomato, PhonePe, Zerodha, Razorpay, Infosys/TCS cloud teams — where "design a system that stays correct when a node dies" is the standard question.

Explained

What a distributed system is, and the models used to reason about it

• Definition (Tanenbaum): independent computers that appear to users as a single system; (Coulouris): components at networked computers communicate and coordinate only by passing messages.
• No shared memory, no shared clock, independent failure — these three absences generate every algorithm in the syllabus.
• Goals: resource sharing, openness (standard interfaces), concurrency, scalability (size, geographic, administrative), and fault tolerance.
• Transparencies (ISO/ANSA/RM-ODP list): *access* (same operations locally and remotely), *location* (name hides where), *migration* (object can move), *relocation* (can move while in use), *replication* (copies hidden), *concurrency* (sharing hidden), *failure* (recovery hidden), *persistence*.
• Synchronous model: known bounds on message delay, clock drift and processing time — consensus is solvable with crash faults. Asynchronous model: no bounds — this is the Internet, and it is where impossibility results bite. Partially synchronous: bounds hold eventually; real systems (Raft, PBFT) are designed here.
• Failure types, in increasing severity: *crash* (stops, stays stopped), *omission* (drops send/receive), *timing* (too slow/early), *response* (wrong value), *arbitrary/Byzantine* (anything, including malicious). A fail-stop failure is a crash that others can detect reliably.
• Byzantine agreement requires n ≥ 3f + 1 nodes to tolerate f Byzantine faults; crash-tolerant consensus needs only n ≥ 2f + 1 (a majority quorum).
• FLP impossibility (1985): in a purely asynchronous system, no deterministic algorithm guarantees consensus if even one process may crash — because a crash and a slow node are indistinguishable. Practical systems escape it with timeouts (partial synchrony) or randomisation, giving safety always and liveness usually.

Architectures, communication and middleware

• Architectural styles: client–server (possibly multi-tier: presentation / application / data), peer-to-peer (structured DHT such as Chord, or unstructured gossip), hybrid (BitTorrent tracker + peers), and modern microservices.
• Middleware sits between OS and application providing naming, RPC, messaging, transactions and security: examples CORBA (ORB + IDL), Java RMI, DCOM, gRPC, Apache Thrift, message brokers (Kafka, RabbitMQ).
• Message passing primitives: *blocking/synchronous* vs *non-blocking/asynchronous* send; *buffered* vs *unbuffered*; *reliable* vs *unreliable*.
• RPC (Remote Procedure Call): client calls a local stub → stub marshals parameters into a message → kernel sends → server stub unmarshals, calls the real procedure → result returns the same way. Steps: the classic ten-step account (Tanenbaum) (call, marshal, send, receive, unmarshal, execute, marshal result, send, unmarshal, return).
• Why RPC is not a local call: pointers cannot be passed (no shared address space — pass by copy/restore or use flattened structures), machines differ in byte order (big- vs little-endian, solved by an external data representation: XDR, CDR, Protocol Buffers), and failures are *partial*.
• RPC semantics: *maybe* (no retry), *at-least-once* (retry; needs idempotent operations), *at-most-once* (duplicate filtering by request ID — the usual choice), *exactly-once* (only as at-least-once retries plus duplicate suppression and persistent state; not achievable at the delivery layer under failures).
• RMI vs RPC: RMI is object-oriented — remote object references, methods, distributed garbage collection; RPC is procedural.
• Message-oriented vs request-reply: brokers decouple sender and receiver in time (persistent queues), enabling at-least-once with idempotent consumers — the dominant pattern in Indian payment and order pipelines.

Time, logical clocks and global state

• Physical clock problems: quartz clocks drift (typical drift rate ρ ≈ 10⁻⁶ s/s, i.e. ~1 s in 11.6 days); two clocks can differ by 2ρ·Δt after Δt seconds, so resynchronisation interval for max skew δ is Δt = δ/(2ρ).
• Cristian's algorithm: client asks a time server, measures round-trip RTT, sets clock to T_server + RTT/2; accuracy ≈ ±(RTT/2 − min_delay).
• Berkeley algorithm: a coordinator polls all machines, averages (discarding outliers), and sends each node an adjustment, not an absolute time — suited to a LAN with no external source.
• NTP uses a stratum hierarchy and offset/delay filtering; India's reference is CSIR-NPL, New Delhi (IST = UTC + 5:30), with NPL NTP servers (time.nplindia.org) and the Legal Metrology framework making NPL's IST the legal time; IST-accurate stamping matters for stock-exchange and banking logs (SEBI/RBI audit requirements mandate synchronised, traceable timestamps — exact clauses vary by circular).
• Lamport logical clock (happened-before, →): rules — (1) before each event, Li = Li + 1; (2) a send carries its timestamp t; (3) on receive, Lj = max(Lj, t) + 1. Gives a → b ⇒ L(a) < L(b); the converse is NOT true.
• Vector clocks: each process keeps a vector V of size n; increment own entry on an event, send V, on receive take element-wise max then increment own. V(a) < V(b) ⇔ a → b; if neither ≤ the other, the events are concurrent. Detects causality exactly, at O(n) message overhead.
• Chandy–Lamport snapshot: a process records its state, then sends a marker on every outgoing channel; on first marker receipt a process records its state and starts recording incoming messages on other channels as channel state, stopping each channel when its marker arrives. Yields a consistent global state (no received-but-not-sent message) without halting the system — used for checkpointing and deadlock/garbage detection.
• Cut terminology: a cut is consistent if, for every event in it, all events that happened-before it are also in it.

Mutual exclusion, election and deadlock

• Requirements: safety (at most one in the critical section), liveness (no deadlock/starvation), and sometimes ordering (entry in happened-before order).
• Central/coordinator algorithm: one coordinator grants tokens — 3 messages per entry (request, grant, release); simple but single point of failure and a bottleneck.
• Ricart–Agrawala (permission-based): broadcast a timestamped request, enter when all n−1 replies arrive; a node defers a reply if it is in or wants the CS with a smaller timestamp. 2(n−1) messages per entry, delay 1 round-trip.
• Lamport's algorithm: same idea but with explicit release messages — 3(n−1) messages.
• Token ring: a token circulates; 1 to ∞ messages per entry (unpredictable), entry delay 0 to n−1. Lost token and dead-node recovery are the problems.
• Maekawa's quorum algorithm: request only a voting set of size ≈ √n → about 3√n messages.
• Election (choosing a coordinator): Bully algorithm — a node that notices the coordinator is dead sends ELECTION to all higher-id nodes; if any answers it stands down, else it declares itself COORDINATOR. Worst case O(n²) messages. Ring algorithm — pass an election message with a list of live ids around the ring; the highest id wins; about 2n messages (or 3n−1 in common accounting).
• Distributed deadlock: handled by *prevention* (wait-die / wound-wait with timestamps), *avoidance*, or *detection* using a wait-for graph (WFG) — centralised (false deadlocks from stale information), hierarchical, or distributed (Chandy–Misra–Haas edge-chasing probes; a deadlock exists if a probe returns to its initiator).

Replication, consistency and the CAP theorem

• Why replicate: performance (read scaling, locality), availability, and fault tolerance. The cost is keeping copies consistent.
• CAP theorem (Brewer; proved by Gilbert–Lynch): of Consistency (linearizability), Availability, Partition tolerance, a system can guarantee only two. Since network partitions cannot be ruled out, the real choice is CP (reject requests during a partition — HBase, ZooKeeper, etcd, classic RDBMS clusters) vs AP (serve possibly stale data — Cassandra, DynamoDB, Riak).
• PACELC extends it: *if Partition then A or C, Else* trade Latency vs Consistency — Cassandra is PA/EL, VoltDB PC/EC.
• Consistency models, strongest first: *linearizability* (every read sees the latest write, as if one copy, in real-time order; *strict consistency* is its idealised absolute-time form) → *sequential* (all processes see the same single interleaving) → *causal* (causally related writes seen in order; concurrent writes may differ) → *FIFO/PRAM* → *eventual* (stops changing if updates stop).
• Client-centric models: monotonic reads, monotonic writes, read-your-writes, writes-follow-reads — what a session needs when the server set changes.
• Quorum (Gifford): with N replicas, a write quorum W and read quorum R give strong consistency if R + W > N (read and write sets must overlap), and write serialisation if W > N/2. Cassandra exposes exactly this as ONE/QUORUM/ALL.
• Protocols: *primary-backup* (remote-write) — all writes to one primary, simple ordering, primary is a bottleneck; *local-write/migration*; *active replication* with a totally ordered multicast; *gossip/anti-entropy* with Merkle trees and read repair for AP stores.
• Conflict handling in AP systems: last-write-wins by timestamp (lossy), vector-clock/version-vector sibling detection, or CRDTs (commutative, convergent data types — counters, sets).

Distributed transactions, commit and consensus

• ACID across nodes needs atomic commitment: either all participants commit or none.
• Two-phase commit (2PC): *Phase 1 (voting)* coordinator sends CAN-COMMIT, each participant writes a prepare record to its log and answers YES/NO; *Phase 2 (completion)* if all YES the coordinator logs COMMIT and sends DO-COMMIT, else ABORT; participants acknowledge. Messages: O(n) — 4 message rounds, 4n messages for n participants (3n if the final acknowledgement is not counted).
• 2PC is blocking: if the coordinator crashes after participants have voted YES, a participant in the *prepared* (uncertain) state must hold its locks until the coordinator recovers. This is 2PC's defining weakness.
• Three-phase commit (3PC) inserts a pre-commit phase so a recovering node can infer the outcome from any peer's state; it is non-blocking only under synchronous assumptions with no network partitions, which is why real systems use consensus instead.
• Consensus properties: *agreement* (all decide the same), *validity/integrity* (the value was proposed), *termination* (all eventually decide).
• Paxos: roles proposer/acceptor/learner; Phase 1a prepare(n) → 1b promise (acceptor promises not to accept lower n, returns any accepted value) → 2a accept(n, v) → 2b accepted; a value chosen once a majority accepts. Multi-Paxos adds a stable leader to skip Phase 1.
• Raft (the one usually asked for in interviews) decomposes it into leader election (randomised election timeouts, terms), log replication (leader appends, commits an entry once replicated on a majority) and safety (only a node with an up-to-date log can win). Understandable equivalent of Multi-Paxos; used by etcd, Consul, CockroachDB, TiKV.
• ZooKeeper (ZAB) gives ordered atomic broadcast and is the coordination service behind Hadoop, Kafka (older versions), HBase.
• Byzantine fault tolerance: PBFT (3f+1 replicas, O(n²) messages) and blockchain-style protocols; India's blockchain pilots (certificate verification, land records) rest on these.
• Saga pattern (practical microservices answer): replace a distributed ACID transaction with a sequence of local transactions plus compensating transactions, relying on idempotency instead of global locks — this is what UPI-style refund/reversal flows look like.

Distributed file systems, naming and big-data processing

• DFS requirements: transparency, concurrency, replication, fault tolerance, consistency, security, efficiency.
• NFS: stateless (classically), mount protocol, VFS interface, client caching with timestamp validation; AFS uses whole-file caching with callbacks and scales better over WANs.
• GFS/HDFS: single NameNode holding metadata, many DataNodes; files split into large blocks (HDFS default 128 MB) with replication factor 3, placed one local, one on another rack, one on the same remote rack (rack awareness); write-once, read-many, no random writes.
• Naming: flat, structured (DNS hierarchy), attribute-based (LDAP/directory); DHT/Chord gives O(log N) lookup hops with a finger table.
• MapReduce: *map* produces (key, value) pairs → *shuffle and sort* groups by key → *reduce* aggregates; failures handled by re-executing tasks, which requires tasks to be deterministic and idempotent.
• Lamport's mutual-exclusion algorithm and logical clocks appear in exams; MapReduce and HDFS appear in interviews — know both ends.

Key points

• Distributed system - independent computers with no shared memory or clock, coordinating only by messages, appearing as one system.
• Transparency - hiding distribution; the eight standard kinds include access, location, replication, failure and concurrency.
• Happened-before (→) - the only ordering a distributed system can know; events not related by it are concurrent.
• Lamport clock - a → b ⇒ L(a) < L(b), but the converse fails; cannot detect concurrency.
• Vector clock - V(a) < V(b) ⇔ a → b; detects concurrency exactly, costs O(n) per message.
• Cristian's vs Berkeley - external server with RTT/2 correction vs internal averaging by a coordinator.
• Chandy–Lamport - marker-based snapshot giving a consistent global state without stopping the system.
• Mutual exclusion costs - central 3, Ricart–Agrawala 2(n−1), Lamport 3(n−1), Maekawa ≈3√n, token ring 1 to ∞.
• Bully vs Ring election - highest live id wins; O(n²) vs about 2n messages.
• Chandy–Misra–Haas - edge-chasing probe; deadlock if the probe returns to its initiator.
• FLP impossibility - no deterministic consensus in an asynchronous system with even one crash fault.
• Byzantine bound - n ≥ 3f + 1 replicas to tolerate f arbitrary faults; n ≥ 2f + 1 for crash faults.
• CAP - under a partition, choose consistency or availability; partition tolerance is not optional.
• Quorum rule - R + W > N gives strongly consistent reads; W > N/2 serialises writes.
• 2PC - atomic commitment in voting + completion phases; blocks if the coordinator dies while participants are prepared.
• Paxos/Raft - majority-based consensus for a replicated log; Raft splits it into election, log replication and safety.
• Idempotency - the practical substitute for exactly-once delivery; at-least-once + idempotent handler is the industry pattern.
• HDFS - NameNode metadata, 128 MB blocks, replication factor 3, rack-aware placement, write-once read-many.

Formulas and examples

Example 1 — Clock resynchronisation interval. Two machines have quartz clocks with drift rate ρ = 2 × 10⁻⁶ s/s. The application needs clocks never more than 50 ms apart.

• Relative drift between the two = 2ρ = 4 × 10⁻⁶ s/s.
• Δt = δ/(2ρ) = 0.050 / (4 × 10⁻⁶) = 12,500 s ≈ 3 h 28 min.
• So resynchronise at least every ~3.5 hours. If the requirement tightens to 1 ms, Δt = 0.001/(4 × 10⁻⁶) = 250 s — about every 4 minutes.

Example 2 — Cristian's algorithm with real numbers. A client sends a time request at T₀ = 10:00:00.000 (its clock) and the reply arrives at T₁ = 10:00:00.240. The server's timestamp in the reply is 10:00:04.000. Minimum one-way network delay is known to be 20 ms.

• RTT = 240 ms → one-way estimate = 120 ms.
• New client time = 10:00:04.000 + 0.120 = 10:00:04.120; since the client's own clock read 10:00:00.240 at that moment, it jumps forward by about 3.88 s (in practice applied gradually by slewing).
• Accuracy = ±(120 − 20) = ±100 ms, so the true time lies in 10:00:04.020 to 10:00:04.220.

Example 3 — Lamport vs vector clocks on the same execution. Three processes; events: P1: a (local), then send m1 to P2. P2: b (local), receive m1 → c, then send m2 to P3. P3: d (local), then receive m2 → e. All clocks start at 0.

• Lamport gives L(a) = L(b) = L(d) = 1 — equal timestamps, with ties broken only by process id; the clocks give no way to tell that a and b (or a and d) are concurrent.
• Vector clocks do: (1,0,0) vs (0,1,0) — neither dominates, so a ∥ b (concurrent). For a and c: (1,0,0) ≤ (2,2,0) and strictly less ⇒ a → c, which is correct.

Example 4 — Quorum design. A Cassandra-style store in an Indian data-centre keeps N = 5 replicas of each row.

• Strong consistency requires R + W > 5. Choices: (W=3, R=3) — tolerates 2 node failures for both reads and writes; (W=5, R=1) — fast reads but no write availability if any node is down; (W=1, R=5) — fast writes, fragile reads.
• Check W > N/2: W = 3 > 2.5 ✓ so writes are serialised. Recommended: W = R = 3 (QUORUM/QUORUM).
• Node availability p = 0.99 each, independent. Probability at least 3 of 5 are up = C(5,3)(0.99)³(0.01)² + C(5,4)(0.99)⁴(0.01)¹ + (0.99)⁵ = 10(0.970299)(0.0001) + 5(0.960596)(0.01) + 0.950990 = 0.00097030 + 0.04802980 + 0.95099005 ≈ 0.99999 (about five nines).

Example 5 — Mutual exclusion message cost. A system has n = 25 nodes, with 40 critical-section entries per second in total across the system.

• The central algorithm is cheapest in messages but fails entirely if the coordinator dies — which is why production systems use a replicated coordinator (ZooKeeper/etcd on Raft) rather than a single one.

Comparison — CP vs AP stores

Where it is used

• Payments and banking: UPI/NPCI switches, RTGS/NEFT, bank core systems — idempotent request IDs, two-phase/saga reversal flows, synchronised IST timestamps for RBI-mandated audit trails.
• E-commerce and consumer internet: Flipkart, Myntra, Zomato, Swiggy, Ola — sharded databases, Kafka event pipelines, eventual consistency for catalogues and carts, strong consistency for inventory and payments.
• Capital markets: NSE/BSE order matching and SEBI-regulated audit logging; Zerodha/Groww backends where clock accuracy and ordering are compliance issues.
• Telecom and cloud: Jio and Airtel OSS/BSS, Aadhaar/UIDAI authentication at very high request rates, NIC/MeghRaj government cloud, AWS/Azure India regions.
• Big-data engineering roles: Hadoop/HDFS, Spark, Kafka, Flink clusters at Indian analytics and IT-services firms (TCS, Infosys, Wipro, Accenture) — "Big Data Engineer" and "SRE" job descriptions list exactly these.
• Coordination infrastructure: ZooKeeper, etcd, Consul for leader election, service discovery and distributed locks in any Kubernetes-based deployment.

Interview questions

• Q: What makes a distributed system fundamentally different from a multiprocessor system? A: No shared memory and no global clock, and components fail independently — so coordination is by messages with unbounded delay, and failure of a part does not stop the whole.
• Q: Why can a Lamport clock not detect concurrent events? A: It guarantees only a → b ⇒ L(a) < L(b); the converse does not hold, so two unrelated events can get ordered timestamps. Vector clocks give the if-and-only-if relation.
• Q: State the CAP theorem and say what choice a real system actually makes. A: Of consistency, availability and partition tolerance you get two; since partitions happen, the real decision is whether to refuse service (CP) or serve stale data (AP) during a partition.
• Q: Why is two-phase commit called blocking, and how is it fixed? A: If the coordinator crashes after participants vote YES, they sit in the prepared state holding locks with no way to decide. 3PC helps only under synchrony; the real fix is a consensus-replicated coordinator (Paxos/Raft) so the decision log survives.
• Q: What does FLP impossibility mean in practice, if Raft clearly works? A: FLP rules out a *deterministic* algorithm that always terminates in a fully asynchronous system. Raft keeps safety unconditionally and uses randomised timeouts for liveness, so it terminates in practice but has no bound in theory.
• Q: Explain exactly-once delivery. A: It is not achievable at the messaging layer with failures; you get at-least-once delivery plus idempotent processing (dedup by request ID) or at-most-once, which together give exactly-once *effects*.
• Q: With 5 replicas, what read and write quorums give strong consistency? A: Any R + W > 5, for example R = W = 3; that also satisfies W > N/2, so writes are serialised and two node failures are tolerated.
• Q: Compare the bully and ring election algorithms. A: Bully: a node contacts all higher ids and the highest live one becomes coordinator, O(n²) messages but fast; Ring: an election message carrying live ids circulates and the maximum wins, about 2n messages but needs a known ring order.
• Q: How many nodes are needed to survive 2 Byzantine failures? A: n ≥ 3f + 1 = 7 nodes; for 2 crash failures a majority quorum of n ≥ 2f + 1 = 5 suffices.
• Q: What is a consistent cut, and how does Chandy–Lamport get one? A: A cut containing all events that happened-before any event in it (no message received without its send). Markers sent on every channel separate pre-snapshot from post-snapshot messages, so in-flight messages are captured as channel state.
• Q: Why is HDFS unsuitable for a transactional application? A: It is write-once/append-only with 128 MB blocks and a single metadata NameNode — optimised for large sequential scans, not small random reads and updates.
• Q: A payment times out with no response. What should the client do? A: Retry the *same* idempotency key, not a new transaction, and reconcile by querying status — a timeout is an unknown outcome, not a failure.

Common mistakes

• Treating a timeout as proof of failure — a slow node and a dead node are indistinguishable asynchronously, which is why retries must be idempotent.
• Claiming CAP lets you "give up partition tolerance": partitions are imposed by the network, not chosen; only C or A can be traded.
• Writing the Lamport receive rule as Lj = max(Lj, t) without the +1, or incrementing only on sends.
• Forgetting the final +1 (own-entry increment) in the vector-clock receive rule, or comparing vectors only element-by-element without checking strictness for ordering.
• Saying 3PC solves blocking in general — it does not under network partitions or asynchrony.
• Confusing consensus (agree on one value, needs majority) with atomic commitment (2PC, needs *unanimity* to commit), and quoting majority rules for 2PC.
• Mixing up eventual consistency with "no consistency", or assuming a QUORUM write makes every subsequent read strong without a QUORUM read (R + W > N must hold on both sides).

Related topics

Operating systems: processes, threads and scheduling · Computer networks: TCP/IP and sockets · Database management systems and transactions · Cloud computing and virtualisation · Parallel computing and GPUs · Microservices and system design.