{ tinou bao }

Causal Atomic Broadcast

Spent the other day reading up on ZooKeeper a little.  ZooKeeper is a distributed, open-source coordination service for distributed applications.  Out of the box ZooKeeper can be used for name service, configuration and group membership.  Building on ZooKeeper's primitives, you can create barriers, locks, queues, etc.

Sounds interesting enough, but what got me really interested was ZooKeeper's atomic broadcast, the guts of ZooKeeper.  Atomic broadcast is very closely related to consensus (apparently equivalent in certain asynchronous systems).  Both are fundamental problems in distributed systems.

So what exactly is atomic broadcast?  To answer this, lets work our way from Reliable Broadcast.

Reliable Broadcast

Reliable Broadcast is the weakest type of fault-tolerant broadcast.  Informally, if a correct process broadcasts a message then all correct processes eventually receive that message.  Reliable Broadcast doesn't impose any message delivery ordering.

FIFO Broadcast

FIFO Broadcast is reliable broadcast that satisifies FIFO Order.  If a process broadcasts message m₁ before message m₂, then no correct process delivers m₂ before it delivers m₁.  For example, if message m₁ is a deposit of $100 into your banking account and message m₂ is a subsequent withdrawal of $75, you most definitely want FIFO Broadcast, otherwise your bank will charge you an overdraft fee (on top of their ridiculous $5 ATM fee).

Causal Broadcast

Sometimes FIFO Order is not enough because FIFO ordering too limited in context (a single process).  You need Causal Order to guarantee that if the broadcast of message m₁ "happens before" or "causally precedes" the broadcast of message m₂, then no correct process delivers m₂ before it delivers m₁.

For example, imagine three processes a, b and c.  Process a broadcasts m_1a, "Banks to charge $5 ATM fee."  Process b delivers m_1a then broadcasts message m_1b, "That's outrageous!"  Without Causal Order, process c could deliver message m_1b before m_1a,

m_1b : "That's outrageous!"
m_1a : "Banks to charge $5 ATM fee."

which doesn't make sense (what is outrageous?).  Causal Order requires delivery of m_1a before m_1b.

Atomic Broadcast

Causal Order imposes a partial ordering in the system, so messages without causal relationships are logically concurrent and do not have any delivery order guarantees.  This can be problematic in some cases.  For example, imagine two replicated databases DB₁ and DB₂.  Process a broadcasts message m_1a, "deposit $100."  Process b broadcasts message m_1b, "charge 10% fee."  Since there is no causal relationship between the messages, this situation may arise,

DB₁ : $0 + $100 - ($100 * 10%) = $90
DB₂ : $0 - ($0 * 10%) + $100 = $100

which is obviously not desirably.

Atomic Broadcast imposes a Total Order on the system so that all messages are delivered in the same order, whatever that order maybe.  So in this example, it's either $90 or $100 but never $90 in one database and $100 in the other.

We can further classify Atomic Broadcast by the actual ordering it imposes.  FIFO Atomic Broadcast is then Reliable Broadcast that is FIFO Order and Total Order.  Causal Atomic Broadcast is then Reliable Broadcast that is Causal Order and Total Order.  Causal Atomic Broadcast is the strongest guarantee we've examined.

Summary

This diagram nicely summarizes Reliable Broadcast and the orderings above (taken from "A Modular Approach to Fault-Tolerant Broadcasts and Related Problems," which the above is basically a summary of).

So there you have, atomic broadcast.

Fail Fast, Debugging and Systems that Never Stop

If you're a Java guy you're probably familiar with the phrase "fail fucking fast" in the context of Iterators (I added the fucking part).  But fail fast is much more than just iterators, it's a general design principle--and a very important one at that. 

Jim Gray on fail fast: "it [the process] should either function correctly or it should detect the fault, signal failure and stop operating."  You can replace "the process" with component, class, etc.  Whatever you designate as your logical unit, the fail fast principle remains.  What you absolutely don't want to do is detect the fault, don't signal failure and continue.  This is why swallowing exceptions is one of my all time pet peeves.  Fail fast is often described in terms of fault detection latency, the time between initial error detection and the fault.  You want your latency to be small. 

There are two advantages to failing fast. First, easier debugging and analysis.  It complicates things greatly if your code detects errors but continues.  Eventually something will go wrong but now you can't easily track down the root cause.  I recently worked with a component (one that shall remain nameless) that doesn't follow the fail fast principle.  During initialization the component knew it was missing a configuration file but instead of halting it continued on.  When the component was finally invoked things obviously did not work, but the root cause was not obvious.  Ended up wasting two days tracking down the root cause, when it really should have taken 30 seconds. 

Second, ability to build "systems that never stop".  Failing fast is so important to building highly available systems that Joe Armstrong talks about it at great lengths in his presentation on "systems that never stop (and Erlang)."  In this presentation he presents 6 laws,

1. Isolation
2. Concurrency
3. Failure Detection
4. Fault Identification
5. Live Code Upgrade
6. Stable Storage

Without failure detection you can't fail fast, and if you can't fail fast you can't achieve isolation.  Real world example: Amazon S3's massive outage last year.  A huge embarrassment to Amazon's otherwise great cloud stack.  Their code didn't have adequate failure detection, allowing a single corrupt bit to propagate throughout the system.  The result was hours of downtime while engineers scrambled.  If Amazon had been able to detect the single corrupt bit then they could have isolated the failing component instead of letting the error bring down the entire system.

Few of us will be coding S3 like systems, but it doesn't mean we can't follow the fail fast approach.  It will make debugging easier and just might prevent similar embarrassments.  I think some programmers don't like to fail fast because they worry they'll be blamed for the failure, so just pretend things are good and pray nothing bad happens downstream.  I suppose it depends on how much you believe in God. 

Fault-Tolerant Distributed Systems Made Simple

Say you want to build a fault-tolerant distributed system (who doesn't). Easy enough, you just replicate a bunch of state machines, the so-called state machine approach. Here's a little more detail to get you started.

At the core is the consensus problem. In order to keep the replicas consistent we need an algorithm for choosing a single value when multiple values may be proposed.

The Paxos algorithms can help us. The specific Paxos algorithm discussed below is the "basic" one (there's a bunch of Paxos algorithms).  The model is asynchronous and non-Byzantine.  This means that a process may crash and recover, but doesn't do anything weird like sending random, deceitful messages to other processes.

There are three players in the algorithm: proposers, acceptors and learners.  Proposers propose values, acceptors accept (or reject) proposals, and learners become aware of the accepted proposal values.  These are logical distinctions, a single process could be all three.

A proposer first sends out a prepare message, basically asking the acceptors if they promise to accept the proposer's proposal.

The prepare message is pretty simple, it just has a unique proposal number n.  Initially, when the acceptors receive the prepare message they will reply with a prepare OK and remember the proposal number they promised to accept.

In this example the acceptor promises to accept proposal for n = 100 and will ignore requests for n < 100.  But the prepare OK message may also contain an already accepted proposal value (say when a different proposer tries to send a prepare message).

Here, a proposer with n = 105 sends a prepare message to an acceptor that has already accepted a value for n = 100. The proposer will need to use that value in subsequent requests to maintain consistency. In the "normal" case the value sent back is null and the proposer is unconstrained, so it can propose any value it wants (the value requested by the client).

Once the proposer receives a majority of N response (prepare OK messages) it can issue accept requests.

As long as the acceptor has not promised to accept requests for a higher n then it will accept the proposal and send an accepted acknowledgment.

When a majority of N accepted messages are received by the learner the learner becomes aware of the accepted proposal value. Notice that at key points in the algorithm data is saved to stable storage so that upon crash recovery things remain consistent.

The reason for requiring a majority N in the algorithm is intuitive. If a majority has accepted the proposal then losing any minority is still acceptable.

Returning to the proposal number. Each proposer needs to use proposal numbers that are unique.

In this example we have three proposers and three disjoint sets. Each proposer uses numbers from one of the sets. The proposal numbers in the sets are calculated by a simple mod operation.

Pretty simple, huh?  Of course there are many challenges when you actually want to implement this algorithm, but let us leave that as an exercise to the reader.

And that's my interesting algorithm of the week (corrections appreciated). 

Incidentally, I love this OmniGraffle Professional drawing program.  For the first time I'm thinking, wow, this program blows anything I've used on Windows (i.e., Visio).