Global state, snapshots and coordination

Many tasks in a distributed system need a picture of the whole system at once. A backup should capture the state of every partition. A debugger wants to know whether the system is deadlocked. A garbage collector wants to know whether any process still refers to an object. On one machine you could pause everything and look. In a distributed system there is no global pause button and no global clock, and messages are always in flight between processes. This lesson asks what a meaningful global state is, shows the Chandy–Lamport algorithm for recording one without stopping the system, and then turns to two coordination problems that recur constantly: ensuring that only one process at a time does something (mutual exclusion) and choosing one process to lead (leader election).

Cuts and consistent global states

A global state consists of the local state of every process together with the state of every channel, the messages sent but not yet received. Imagine recording each process's state at some moment of its own choosing. The set of recording points is called a cut: it divides each process's history into a past and a future. A cut is consistent if, for every message whose receipt is in the past, its sending is also in the past. In terms of lesson 3, a consistent cut is closed under happened-before: if it contains an event, it contains every event that happened before it.

An inconsistent cut records an effect without its cause, such as money arriving at one bank branch that has not yet left another. Such a state never existed and could never have existed, so conclusions drawn from it are wrong. A consistent cut may not correspond to a moment in real time either, but it is a state the system could have passed through, which is enough for most purposes. In particular, if a property is stable, meaning that once it becomes true it stays true, such as "the computation has terminated", "these processes are deadlocked" or "this object is garbage", then evaluating it on a consistent cut gives a correct answer: if it holds in the recorded state, it holds now.

The Chandy–Lamport snapshot algorithm

K. Mani Chandy and Leslie Lamport published an algorithm in 1985 that records a consistent global state of a running system without stopping it. It assumes reliable, first-in-first-out channels between every pair of processes. Any process can start a snapshot. The algorithm uses a special message, a marker, which separates messages sent before a process recorded its state from those sent after.

The rules are short. To start, a process records its own state and sends a marker on every outgoing channel before sending anything else. When a process receives a marker on some channel for the first time, it records its own state, records that channel as empty, sends markers on all its outgoing channels, and begins recording every other incoming channel. When a process receives a marker on a channel it was recording, it stops recording that channel; the messages it saw on that channel between recording its own state and receiving the marker are the channel's recorded state. The snapshot is complete when every process has recorded its state and received a marker on every incoming channel.

Why is the result consistent? Because channels are FIFO, any message a process sent after recording its state arrives after the marker it sent on the same channel. So when the receiver records its state, which happens no later than receiving that marker, it has not yet received any message sent after the sender's recording point. Every receipt inside the cut therefore has its send inside the cut. Messages sent before the sender's recording but received after the receiver's are exactly those the receiver records as channel state.

# Chandy-Lamport snapshot of three bank branches that keep transferring money.
# Channels are FIFO. Total money is 300; a correct snapshot must also total 300.
import random
from collections import deque
random.seed(450)
P = ["A", "B", "C"]
bal = {"A": 100, "B": 100, "C": 100}
chan = {(i, j): deque() for i in P for j in P if i != j}   # messages in flight i -> j
recorded, chan_state, recording = {}, {}, {}

def start_snapshot(p):
    recorded[p] = bal[p]                                  # record own state
    for q in P:
        if q != p:
            chan[(p, q)].append(("MARKER", 0))             # marker on every outgoing channel
            recording[(q, p)] = True                       # record incoming channels
            chan_state[(q, p)] = []

def deliver(i, j):
    kind, amt = chan[(i, j)].popleft()
    if kind == "MARKER":
        if j not in recorded:                              # first marker: snapshot now
            start_snapshot(j)
            chan_state[(i, j)] = []                        # this channel recorded as empty
        recording[(i, j)] = False                          # stop recording this channel
    else:
        bal[j] += amt
        if recording.get((i, j)):
            chan_state[(i, j)].append(amt)                 # in flight when the cut was taken

steps, started = 0, False
while steps < 400:
    steps += 1
    if steps == 40 and not started:
        start_snapshot("A"); started = True                # A initiates the snapshot
    if random.random() < 0.5:                              # a branch sends a transfer
        i = random.choice(P); j = random.choice([q for q in P if q != i])
        amt = random.randint(1, 20)
        if bal[i] >= amt:
            bal[i] -= amt; chan[(i, j)].append(("MONEY", amt))
    else:                                                  # some channel delivers a message
        busy = [k for k in chan if chan[k]]
        if busy: deliver(*random.choice(busy))
    if started and len(recorded) == 3 and not any(recording.values()):
        break

in_flight = sum(sum(v) for v in chan_state.values())
print("recorded branch balances:", recorded)
print("recorded in-flight money:", {f"{i}->{j}": v for (i, j), v in chan_state.items() if v})
print("snapshot total:", sum(recorded.values()) + in_flight)
naive = dict(bal)                          # what you would get by just reading balances now
print("live balances now:", naive, "sum", sum(naive.values()),
      "+ in channels", sum(a for q in chan.values() for k, a in q if k == "MONEY"))

recorded branch balances: {'A': 126, 'C': 20, 'B': 127}
recorded in-flight money: {'B->A': [14], 'C->B': [13]}
snapshot total: 300
live balances now: {'A': 68, 'B': 131, 'C': 3} sum 202 + in channels 98

The recorded balances alone may not sum to 300, because some money was in flight when the cut was taken; adding the recorded channel contents restores the total exactly. That conservation check is the classic test of a correct snapshot. Snapshot algorithms of this family are used in practice. Apache Flink, a stream-processing system, takes periodic distributed snapshots of operator state by injecting barriers, which play the role of markers, into its data streams; after a failure, every operator restores the last completed snapshot and replays input from the corresponding positions in the source logs. Lesson 12 returns to this as the basis of exactly-once stream processing.

Mutual exclusion

CMPUT-379 (lessons 5 and 6) solved mutual exclusion among threads with locks built on atomic hardware instructions. Between machines there is no shared memory, so mutual exclusion must be built from messages. Three classic designs illustrate the trade-offs.

The simplest is a central coordinator. A process wanting the lock sends a request to the coordinator, which grants it if the lock is free and otherwise queues the request; the holder sends a release when done. It takes three messages per entry and is easy to understand, but the coordinator is a single point of failure and a potential bottleneck. In the Ricart–Agrawala algorithm (1981), a process wanting entry sends a request timestamped with a Lamport clock to every other process and enters only after all have replied; a process defers its reply if it is in the critical section or has an earlier request of its own. It needs \(2(n - 1)\) messages per entry and has no central point of failure, but any single crashed process blocks everyone, so it is less fault tolerant than the central design. In a token ring, a single token circulates around a logical ring and only its holder may enter; it is fair and simple, but the token can be lost when its holder crashes, and detecting that is hard.

None of the classic algorithms tolerates failures well, which is why real systems take a different route: they obtain locks from a small, replicated coordination service such as ZooKeeper or etcd, which uses consensus internally (lesson 9) and therefore keeps working while a minority of its servers are down.

Leases and fencing tokens

A lock held by a process that crashes would block everyone forever, so distributed locks are granted as leases: a lock that expires after a fixed time unless renewed. Leases were introduced by Gray and Cheriton in 1989 for cache consistency and are now everywhere. A lease solves the crashed-holder problem but creates a subtler one. Suppose a client holding a lease is paused, perhaps by a long garbage-collection pause, a virtual machine migration or a page fault on a busy disk. The lease expires, the service grants it to another client, and then the first client wakes up, still believing it holds the lease, and writes to shared storage. Two clients now act as lock holders at once.

The fix, described clearly by Martin Kleppmann, is a fencing token: a number that the lock service increases every time it grants the lock. The client must present its token with every write, and the storage system rejects any write carrying a token smaller than one it has already seen. The paused client's token is stale, so its late write is refused. Note where the check happens: in the resource being protected, not in the client, because the client is precisely the party that cannot be trusted to know its lease has expired. ZooKeeper's znode version numbers and etcd's revision numbers serve as fencing tokens.

# Leases and fencing tokens: a paused client must not overwrite newer work.
class Storage:
    def __init__(self, fencing):
        self.data, self.max_token, self.fencing = {}, 0, fencing
    def write(self, key, value, token):
        if self.fencing and token < self.max_token:
            return f"REJECTED (token {token} < {self.max_token})"
        self.max_token = max(self.max_token, token)
        self.data[key] = value
        return "ok"

for fencing in (False, True):
    s = Storage(fencing)
    # t=0: client 1 acquires the lease with token 33, then pauses (garbage collection)
    # t=12: the lease (10 s) has expired; client 2 acquires it with token 34 and writes
    r2 = s.write("invoice-812", "paid by client 2", 34)
    # t=15: client 1 wakes up, still believing it holds the lease, and writes
    r1 = s.write("invoice-812", "stale write by client 1", 33)
    print(f"fencing {'on ' if fencing else 'off'}: client2 {r2}; client1 {r1}; final = {s.data['invoice-812']!r}")

fencing off: client2 ok; client1 ok; final = 'stale write by client 1'
fencing on : client2 ok; client1 REJECTED (token 33 < 34); final = 'paid by client 2'

Leader election

Many protocols work best with a single coordinator, or leader, that orders requests or makes decisions, while the other processes follow. When the leader crashes, the others must choose a new one. A leader-election algorithm must ensure that, eventually, every live process agrees on the same leader. The two classic algorithms assume processes have unique, comparable identifiers and choose the largest live one.

In Garcia-Molina's bully algorithm (1982), a process that suspects the leader has failed sends an ELECTION message to every process with a higher identifier. If none replies, it declares itself leader and sends COORDINATOR to everyone. If a higher process replies, that process takes over the election, and so on, until the highest live process wins; hence the name, since the biggest process always takes over. In the ring algorithm (Chang and Roberts, 1979), processes are arranged in a logical ring and an election message circulates collecting identifiers, or carrying only the largest seen so far; when it returns to its initiator, the largest identifier is announced as leader.

# The bully algorithm: the highest-numbered live process becomes coordinator.
def bully(initiator, alive, log):
    higher = [p for p in alive if p > initiator]
    log.append(f"{initiator} sends ELECTION to {higher or 'nobody'}")
    if not higher:
        log.append(f"{initiator} wins and sends COORDINATOR to all")
        return initiator
    log.append(f"{', '.join(map(str, higher))} {'replies' if len(higher) == 1 else 'reply'} OK to {initiator}; {initiator} waits")
    # every live higher process starts its own election; the highest finishes it
    return bully(min(higher), alive, log)

alive = {1, 2, 3, 4, 6}          # process 5 has crashed; 7 crashed earlier
log = []
winner = bully(2, alive, log)   # process 2 noticed the old coordinator (7) is gone
print("\n".join(log))
print("coordinator:", winner)
msgs = sum(1 for p in alive for q in alive if q > p)
print("ELECTION messages if every process starts an election (worst case):", msgs)

2 sends ELECTION to [3, 4, 6]
3, 4, 6 reply OK to 2; 2 waits
3 sends ELECTION to [4, 6]
4, 6 reply OK to 3; 3 waits
4 sends ELECTION to [6]
6 replies OK to 4; 4 waits
6 sends ELECTION to nobody
6 wins and sends COORDINATOR to all
coordinator: 6
ELECTION messages if every process starts an election (worst case): 10

Both classic algorithms assume reliable failure detection: a process that does not reply is assumed dead. Lesson 1 explained why that assumption fails in practice. If the old leader is merely slow or partitioned away, it may still believe it is leader while the others elect a new one, and both act as leader at once. Avoiding this split brain requires two ideas developed in lessons 8 and 9. First, a leader must be chosen by a majority (a quorum), and any two majorities overlap, so two leaders cannot both be chosen in the same round. Second, each leadership period carries a term or epoch number, increasing with every election, and every message carries it, so that a deposed leader's messages are recognized as stale, which is exactly the fencing idea applied to leaders. Raft's elections in lesson 9 combine both ideas with randomized timeouts.

Coordination services

Because mutual exclusion, leader election, configuration management and group membership are needed by almost every distributed system, and are hard to implement correctly, they have been factored into dedicated coordination services. Google's Chubby (2006) provided a lock service with small files, used for electing masters in systems such as Bigtable. Apache ZooKeeper (2010), developed at Yahoo, offers a hierarchical namespace of small data nodes with ordered updates, watches that notify clients of changes, and ephemeral nodes that vanish when their creator's session ends, which makes leader election a few lines of client code. etcd, created at CoreOS in 2013 and now part of the Cloud Native Computing Foundation, offers a key-value store with leases and watches and is the configuration store at the heart of every Kubernetes cluster. All three replicate their state with consensus protocols, which lesson 9 examines.

Worked examples