Google Chubby and Apache ZooKeeper: coordination and locks

01 · CONTEXT

Who's in charge now? The coordination problem

Every system that runs several copies of a server has to answer a simple question: who is in charge now? The answer must hold for everyone at the same time, even with machines and networks failing.

The authors of Chubby, Google's lock service, expected it to help with coarse-grained synchronization and, in particular, with electing a leader among equivalent servers. GFS uses a Chubby lock to designate its master, and Bigtable uses it to elect the master, for the master to discover the servers it controls, and for clients to find the master.

Before Chubby, most of Google's distributed systems used ad hoc methods to elect the primary, when the work could be duplicated without harm, or required operator intervention, when correctness was essential. In the second case, the paper reports a significant availability improvement in systems that no longer needed human intervention on failure.

The service offers an interface similar to a distributed file system, with advisory locks, and the design prioritizes availability and reliability over high performance. The paper notes that anyone familiar with distributed computing will recognize electing a primary among peers as an instance of the distributed consensus.

Why a lock service and not a consensus library? One reason is that systems often start as prototypes, with little load and loose availability guarantees, and a lock server makes it easier to keep the program structure and the existing communication patterns. To elect a master that writes to an existing file server, two extra statements and one extra RPC parameter are enough. Another reason is that, with a lock service, even a single client can acquire the lock and make progress safely.

The second central decision was to serve small files, so that the elected primary can announce itself and its parameters without requiring a second service. Another choice, which the paper says may be surprising, is to expect coarse-grained use, not fine-grained locks held for seconds or less. Coarse-grained locks put much less load on the server, and one example is electing a primary that then handles all access to some data for a considerable time, perhaps hours or days.

02 · THE CHUBBY CELL

Replicas, a master, and a master lease

A Chubby cell is a small set of servers, typically five, called replicas and placed so as to reduce the chance of correlated failure, for example in different racks. Because consensus algorithms decide by quorum, Chubby usually has five replicas per cell, of which three must be running for the cell to work.

The replicas use a consenso protocol to elect a master. The master needs votes from a majority of the replicas and promises from them that they will not elect another master for a few seconds, the master lease. Only the master initiates reads and writes on the database; the other replicas just copy the updates.

1. The cell's replicas are listed in DNS, and that is where the client sends the request to locate the master.

2. The client asks a replica where the master is, and a replica that is not the master answers with its identity.

3. Once the master is located, the client sends all requests to it until it stops responding or says it is no longer the master.

4. A write is propagated to the replicas by the consensus protocol and is only acknowledged when it reaches the majority.

5. Reads are served by the master alone, which is safe as long as the master lease has not expired.

6. Every few hours the master writes a snapshot of the database to a GFS server in another building.

Writes are propagated to all replicas by the consensus protocol and acknowledged when they reach the majority. Reads are served by the master alone, which is safe as long as the master lease has not expired, because no other master can exist. If the master fails, the other replicas run the election when their master leases expire, and a new master is usually elected within a few seconds; the paper cites two recent elections that took 6 s and 4 s, and values of up to 30 s.

The client sees a strict tree of files and directories, in an interface similar to UNIX's but simpler, and any node can work as an advisory read/write lock. Ephemeral files serve as temporary files and as indicators to others that a client is alive.

O backup works like this: every few hours the master of each cell writes a snapshot of the database to a GFS server in another building. This ensures the backup survives damage to the building and avoids a circular dependency, since a GFS in the same building could depend on Chubby to elect its master.

03 · LOCKS, SEQUENCERS, AND SESSIONS

Advisory locks and the delayed message

Chubby locks are advisory: they conflict only with other attempts to acquire the same lock. The authors rejected mandatory locks, which make locked objects inaccessible to clients that do not hold the lock, and one reason is that Chubby locks often protect resources implemented by other services, not just the file associated with the lock.

In distributed systems locking is complex because communication is uncertain and processes fail independently. A process holding lock L may issue a request R and fail; another process may acquire L and act before R reaches its destination. If R arrives later, it may be executed without the protection of L and on potentially inconsistent data.

Chubby offers the sequencer: the lock owner can request an opaque byte string that describes the state of the lock right after acquisition, with the lock name, the acquisition mode, and the lock's generation number. The client passes it to the servers whose operations it wants to protect, and the receiving server must test whether the sequencer is still valid and, if not, reject the request. The paper also describes, in the design rationale, passing the lock's acquisition count in the write RPC, with the file server rejecting the write if the count is lower than the current value.

For servers that do not support sequencers, the paper describes an imperfect but easier mechanism to reduce the risk of delayed or reordered requests. If the lock becomes free because the owner failed or became unreachable, the server prevents other clients from claiming it for a period called the lock-delay. The client can choose any value up to a limit, at the time one minute.

The relationship between client and cell is the session, maintained by periodic handshakes called KeepAlives. Each session has a lease: a future interval in which the master guarantees not to terminate the session unilaterally. The default extension is 12 s, and an overloaded master can use larger values to reduce the number of KeepAlives. The master normally blocks the KeepAlive until the client's previous lease is close to expiring, and the client starts a new one right after receiving the reply, so there is almost always a KeepAlive blocked at the master.

O cache on the client also depends on the session. Clients keep a consistent, write-through cache of file data and metadata in memory, including the absence of files. To change a piece of data, the master blocks the modification while it sends invalidations to every client that may have it cached, and the modification only proceeds after each client has acknowledged the invalidation or let its cache lease expire. Invalidations travel in KeepAlive replies, which prevents the client from keeping the session without acknowledging them.

1. In normal operation, the client almost always has a KeepAlive blocked at the master, and the reply states the new lease deadline.

2. The master dies before answering the next KeepAlive, and the other replicas run the election when their master leases expire.

3. The client's local lease expires: it flushes and disables the cache, blocks application calls, and waits out the grace period.

4. The new master picks a new epoch number, which clients must present on every call, and rejects calls with an old epoch.

5. The first KeepAlive to the new master is rejected for carrying the wrong epoch; the reply to the resent request lets the client extend the lease.

6. Because the grace period covered the interval between the end of the old lease and the start of the new one, the application noticed only a delay.

The outcome depends on a comparison. If the grace period covers the interval between the end of the old lease and the start of the new one, the client sees only a delay; if it were shorter than that interval, it would have abandoned the session and reported the failure to the application.

04 · APACHE ZOOKEEPER

A coordination kernel with no locks in the API

The ZooKeeper paper describes Chubby as a lock service with strong synchronization guarantees, on top of which locks implement leader election and group membership. ZooKeeper, in turn, aims to be a simple, high-performance kernel for building more complex coordination primitives on the client.

In designing the API, the authors moved away from blocking primitives, such as locks, because they can let slow or faulty clients hurt the performance of faster ones. Looking only at the signatures, ZooKeeper looks like Chubby without the lock, open, and close methods.

Data lives in znodes organized in a hierarchical namespace. An ephemeral znode is removed automatically when the session that created it ends, deliberately or through failure, and the sequential flag appends a monotonically increasing counter to the znode's name. Znodes were not designed for general data storage.

There are two ordering guarantees. The first is writes that are linearizable: all requests that change state are serializable and respect precedence. The second is per-client FIFO order: a client's requests execute in the order they were sent.

The replicated database is an in-memory database holding the whole data tree. All requests that update state are forwarded to the leader, which computes the system state after the write and turns it into a transaction; unlike client requests, transactions are idempotent. The leader broadcasts the change through Zab, which uses majority quorums by default, so ZooKeeper only works if a majority of the servers are correct.

1. Each client connects to exactly one server in the ensemble and sends its requests to it.

2. A read is answered from the server's local database, without an agreement protocol, and may return a stale value.

3. A write is forwarded by the server to the leader.

4. The leader turns the write into an idempotent transaction and broadcasts it through Zab, which decides by majority.

5. The server that received the request replies to the client when it delivers the change, and that server is the one that triggers that client's watch notifications.

6. Whoever needs the latest value calls sync before the read.

The price of local reads is possible staleness: a fast read does not guarantee precedence order and may return a stale value, even if a more recent update to the same znode has already been acknowledged. When that matters, the client calls sync followed by the read to get the latest value.

Change notification follows a different model. Watches let clients receive notifications without polling; they are one-time triggers associated with a session and indicate that a change happened, but do not deliver it. The paper contrasts this with Chubby, which directly manages client caches and blocks updates to invalidate the cache of everyone holding the data; if any client is slow or faulty, the update is delayed. Leases only limit this impact, whereas ZooKeeper's watches avoid the problem entirely.

Locks exist, but as a client-side recipe. The service does not know about these more powerful primitives, which are implemented entirely on the client with the API. In the simplest lock, the client tries to create an ephemeral znode and, if creation succeeds, holds the lock; this version suffers from the herd effect, because everyone waiting competes for the lock when it is released, although only one can get it.

The recipe without the herd effect uses the sequential flag to order each client's attempt relative to all the others. If the client's znode has the lowest sequence number, it holds the lock; otherwise, the client waits for the removal of the znode that holds the lock or will receive it before it. Since each client watches only the znode that precedes its own, only one process is woken when a lock is released or a request is abandoned.

The paper also shows a case where locks like Chubby's are not enough. When a new leader takes over, it must change many configuration parameters, and there are two requirements: while it makes the changes, the other processes must not start using the configuration being altered; and, if it dies before finishing, they must not use the partial configuration. Locks like Chubby's would help with the first requirement but are insufficient for the second. In ZooKeeper, the new leader deletes the ready znode, updates the configuration znodes, and recreates ready; by the ordering guarantees, whoever sees ready also sees all the changes.

In real use, the Yahoo! Message Broker employs ZooKeeper to manage topic distribution, handle machine failures, and control system operation. For the target workloads, with read-to-write ratios from 2:1 to 100:1, the paper shows that ZooKeeper handles tens to hundreds of thousands of transactions per second. Regarding disponibilidade, the service is available if a majority of the servers are up and communicating, and an acknowledged change persists as long as a quorum can recover.

05 · SIDE BY SIDE

Chubby and ZooKeeper on the same questions

06 · PRODUCTION LESSONS

What operations taught Chubby

Chubby's clients are individual processes, so the master can have many more clients than expected: the paper reports having seen 90 thousand clients talking directly to one master, far more than the number of machines. Since there is one master per cell, on a machine like the clients', they can overload it by an enormous margin.

RPC traffic is dominated by session KeepAlives, with few reads and very few writes or lock acquisitions. The authors conclude that the key to scaling Chubby is not server performance, but reducing communication with it.

The paper also measures the availability side: in a sample of 61 outages over a few weeks, totaling 700 cell-days, most lasted 15 s or less and 52 were under 30 s, and most applications are not significantly affected by outages under 30 s.

Although designed as a lock service, Chubby's most popular use was as a name server. In DNS, if quick replacement of failed services is desired, the TTL may have to be small enough to overload the servers. The paper considers a TTL of 60 s and calculates that keeping the DNS caches of a 3,000-client job would require 150 thousand queries per second. Chubby's cache uses explicit invalidations, so a constant rate of KeepAlives keeps an arbitrary number of entries cached indefinitely, in the absence of changes.

Even so, there were spikes. When deploying the name service, starting a job of 3,000 processes, which generates 9 million requests, could bring the master to its knees. The fix was to group name entries into batches, so that one lookup returns the mappings of a large number, typically 100, of related processes.

Another lesson came from abusive clients. Two other services started using the same upload-tracking module for a larger population of users, and a single 1.5 MB file began to be rewritten in its entirety on every user action, with the space used by the service exceeding that of all other clients combined. Chubby introduced a limit of 256 kB per file, but migrating the data to another system took about a year.

The last lesson is about who uses the service. Developers rarely think about failure probabilities and tend to treat Chubby as if it were always available. Chubby provides an event for clients to detect that a master fail-over happened, intended for them to check for possible changes, but many developers chose to bring the application down on receiving it, greatly reducing the availability of their systems.