Distributed Consensus with Paxos

The Problem of Distributed Consensus

Paxos is an algorithm: a series of repeatable steps by which a distributed system reaches consensus. Consensus here is a majority. More than half the nodes agree on a particular value. Distributed means the components can fail individually and the messages between them can be dropped.

Apache Zookeeper uses Paxos for distributed locking. So does Google Chubby.

The problem shows up with five servers, S1 to S5, writing to three databases.

1. S2 receives "add 10 to X" and tells all three databases to persist it. The log position is zero.
2. Two other servers receive "subtract 10 from Y" and "add 10 to Y". These requests can arrive in any order.
3. One database logs the subtraction first. Another logs the addition first.
4. Y starts at 5. Where the subtraction comes first it fails for insufficient balance, and Y ends at 15. Where the addition comes first, Y goes to 15 and then to 5.

The two databases hold inconsistent data. Two-phase commit and quorum are two answers. Paxos has advantages over both.

Basic Algorithm

Paxos writes to a log in two phases: first get the log line and lock it, then write to it.

Three servers, S1, S2 and S3, broadcast every write to three databases, D1 to D3. A write is durable when at least two accept it.

1. A plain broadcast can be overwritten by a concurrent request. The position of a write is as important as its data.
2. So a server first asks for the log line. The databases answer log line zero, and the server writes at zero. This is the propose.
3. S2 and S3 can both be told zero before either writes. S3 then overwrites S2: a stale read.
4. So the database records the owner of the log line. When S3 asks after S2, the line is given to S3, because S2 might be dead.
5. A write from S2 is now rejected. S2 asks again, becomes the owner, and its write executes.

A server has to own the log line and execute the write consecutively for there to be consensus.

Improving the algorithm

A lock request that carries a timestamp instead of the name of its server can be compared with any other request.

In the basic algorithm a request to acquire the lock is tied to the identity of the server that sends it. Another server cannot agree on the previous value, and errors across generations of elections are hard to debug.

1. S2 sends a timestamp, T2, instead of its server ID. The latest operation is picked up.
2. The databases store the timestamp beside the log line. A server sends the same timestamp to D1, D2 and D3.
3. Timestamps are unique across the system. S1 takes T1 and T4, S2 takes T2 and T5, S3 takes T3 and T6: each server takes the numbers with one remainder when divided by 3.
4. Two timestamps can now be compared, and the more recent request is known.

Writing is isolated from acquiring the lock: another server can write without having made the initial claim. A server that asks again with a new timestamp runs a fresh election.

Termination Conditions

Paxos is a two-phase protocol: a server locks a log line with a timestamp, then writes to it, and a majority of the databases has to agree.

Servers do not talk to each other, and databases do not talk to each other. A server reaches consensus by observing the values in the databases.

1. S2 sends a log line request with its timestamp. The database answers with the log line and logs the row with that timestamp.
2. S3 sends a request with a higher timestamp. The database overrides the row only because the timestamp is greater.
3. S3 writes with its timestamp. It matches, and the write is accepted.
4. S2 writes with its lower timestamp. The database rejects it. An accept without a prepare fails too.
5. For the next write the database tells the server which log line comes next.

If the databases answer with different log lines, there is no majority value and the server does nothing. The algorithm terminates when a majority of databases agree on the accept: the server alone knowing this is sufficient.

Practical Considerations

Paxos in the real world is optimized further: servers retry with jitter, and a leader is elected so that a write is no longer a log line request.

1. Timing. S1 and S2 retrying a log line request after the same interval hit the server together. With jitter S1 sends every three seconds, S2 every one second and S3 every five seconds, so one of them wins.
2. Speed. Running the algorithm for every read or write takes a lot of bandwidth. Instead of a log line, the servers select a leader. The one that gets a majority wins, and all write operations are directed to it. This is multi-Paxos.
3. Catching up. A database far behind the others has to catch up. The Paxos algorithm does not define how.

Google Chubby and Apache Zookeeper use Paxos for distributed locking. The lock is a small file in a file system: the owner that the majority of the servers accept has the permission to write to a node.

The testing of the algorithm costs more than its implementation.