Distributed Cache: System Design

A database answers a question by doing work: it locates pages on disk, follows an index, perhaps joins several tables and sorts the result. When the same question is asked again a moment later, it does all of that work again, although nothing has changed and the answer will be the same. A cache is a component that remembers recent answers, keeps them in memory, and returns them without consulting the database, so that a question which took 20 ms the first time takes less than 1 ms on every subsequent occasion and places no load on the database at all.

That description contains the whole appeal of caching and conceals nearly all of its difficulty. The remembered answer is a copy, and from the moment it is made the original is free to change without the copy’s knowledge. A cache also has less room than the data it copies from, so that it must continually decide what to forget. And because a system that has a cache soon comes to rely on it, the moments at which the cache cannot help, when an entry has just expired or a cache node has just been lost, are precisely the moments at which the database receives a load it has not carried for months. This chapter treats each of these in turn: what to keep, how a copy and its original drift apart and how far that can be controlled, and how a cache fails and takes other things with it.

What a cache is worth

The fraction of reads that a cache is able to answer is called its hit ratio, a read answered from the cache being a hit and a read that has to go on to the database a miss. Almost everything about the value of a cache follows from this single number, and the relationship is less linear than intuition suggests.

Suppose that a read from the cache takes 1 ms and a read from the database 20 ms. The average read then costs the hit ratio times 1 ms plus the miss ratio times 20 ms.

A cache that answers half of all reads has barely halved the average, since the slow half still dominates. At 90% the average has fallen to 2.9 ms, and at 99% to 1.19 ms. The more important effect, however, is on the database. Between a hit ratio of 90% and one of 99%, the reads reaching the database fall from 10% of the total to 1%, which is a tenfold reduction, and it is usually this reduction, more than the gain in latency, that a cache is installed to obtain. A database that could not possibly serve the whole read load directly is able to serve the 1% that the cache passes on.

The same arithmetic contains a warning that the rest of the chapter develops. A system whose database is sized for 1% of the read load is a system that cannot survive without its cache. If the hit ratio falls from 99% to 90%, the load on the database multiplies by 10, and if the cache is emptied altogether it multiplies by 100.

Whether a high hit ratio is attainable depends on the traffic. Caching works because requests are unevenly distributed, a small number of keys being asked for again and again while the great majority are asked for rarely. The simulation for this chapter sends a million requests for 100000 keys, with popularity concentrated on a few, to a cache of various sizes.

A cache holding a quarter of the keys answers nearly 3 quarters of the requests. With traffic more sharply concentrated than this, which is common, the same hit ratio would be reached with a much smaller cache, and with requests spread evenly over the keys a cache of 25% would answer only 25%. The distribution of real traffic should therefore be measured before a cache is sized, and no figure quoted from elsewhere, including the figures above, is a substitute.

Arranging a cache and a database

There are 4 established ways of placing a cache beside a database, which differ in who is responsible for filling the cache and in what order the 2 are written.

In the first and commonest arrangement, called cache-aside, the application manages everything. To read, it asks the cache, and on a miss it reads the database and stores the result in the cache for the benefit of later readers. To write, it updates the database and then removes the corresponding entry from the cache, so that the next reader will miss and fetch the new value. The cache is a passive store that knows nothing of the database. The great merit of this arrangement is that the cache is optional: if it is unreachable, the application reads the database directly and continues to work, only more slowly.

Read-through moves the same logic into the cache. The application asks the cache for a key, and the cache, on a miss, fetches the value from the database itself. The behaviour is that of cache-aside, with the advantage that the fetching code exists in 1 place, where every application shares it, and the disadvantage that the cache must be taught how to query the database.

In write-through, the application writes to the cache, and the cache writes to the database before confirming. Every write thus refreshes the cache, which consequently always holds the latest value of whatever has been written through it. The cost is that every write pays for 2 stores, and that entries are cached on writing whether or not anybody will read them.

Write-behind is write-through with the database write postponed. The cache confirms the write at once and passes it to the database later, usually in batches. Writes become very fast, and a burst of them is smoothed into a steady stream. The price is durability: writes that the cache has confirmed and has not yet passed on are lost if the cache node fails, which is the kind of loss that chapter 2 went to such lengths to prevent. Write-behind is appropriate for data whose occasional loss is tolerable, such as counters of page views, and for little else.

The design of this chapter uses cache-aside, because requirement by requirement it is the arrangement in which a failure of the cache does the least harm.

What to remove when memory is full

In steady operation a cache is always full, since it has less memory than there is data and something is inserted on every miss. Each insertion therefore requires an eviction, the removal of an existing entry, and the rule by which the victim is chosen determines the hit ratio that a cache of a given size achieves. The ideal rule would remove the entry that will not be needed for the longest time to come. As the future is unknown, practical rules estimate it from the past, and they differ in which aspect of the past they consult.

Least recently used

The rule known as least recently used, or LRU, removes the entry that has gone longest without being read or written. Its justification is the observation that requests come in clusters, so that something used a moment ago is likely to be used again soon, and something untouched for a long time is likely to remain so. The cache keeps its entries in order of last use, moves an entry to the front whenever it is used, and evicts from the back.

The final step exhibits the characteristic weakness of the rule. A had been requested 3 times in succession and was plainly the most popular key, and yet a single pass over 3 other keys, each requested once, was enough to push it out. In practice this happens when a job scans a large number of keys once, for a report or a backup: the scan passes through the cache, evicting the truly popular entries in favour of entries that will never be requested again, and the hit ratio collapses until ordinary traffic has restored the popular entries.

Least frequently used

Least frequently used, or LFU, consults how often an entry has been used and disregards when. Every entry carries a count of its uses, and the entry with the lowest count is evicted.

On this sequence the rule does better, since the count of 3 protects A from the single requests for B, C and D, and the last request is a hit. A scan can no longer displace popular entries, because the scanned keys never accumulate a count.

LFU has the opposite weakness. A count, once earned, lasts indefinitely, so that an entry which was requested 10000 times last month and has not been requested since continues to occupy its place, while a newly popular entry begins at 1 and is in danger of eviction before it has had time to prove itself. Practical versions therefore let the counts decay, halving them at intervals, so that popularity has to be maintained to be kept.

Other rules

First in, first out evicts the entry that was inserted longest ago, without regard to use. It needs no bookkeeping on reads, which makes it the cheapest rule, and it evicts popular and unpopular entries alike, which generally makes it the least effective.

Evicting an entry chosen at random sounds like no rule at all and performs surprisingly close to LRU on many workloads. A refinement that several in-memory stores have adopted is to pick a handful of entries at random and evict the least recently used among them. This approximates LRU closely while avoiding the ordered list that true LRU must update on every single read, an update which becomes a point of contention when many threads read the cache at once.

The most effective rules in current use combine recency and frequency. One widely adopted approach admits a new entry to the main part of the cache only if an approximate count of its past requests exceeds that of the entry it would displace, so that keys seen once, whether from a scan or from the long tail of unpopular keys, are turned away at the door. The details are beyond this chapter, and the principle is the one the 2 pictures above have already shown, namely that recency and frequency each catch what the other misses.

Our design uses least recently used, as most caches do. It is simple, it suits traffic in which popularity changes from hour to hour, and its weakness can be avoided by keeping scans away from the cache.

How a copy becomes stale

A cached value is stale when the database holds a newer one. Some staleness is inherent in caching and has to be accepted, and the purpose of the design is to bound it and to make sure that it is never permanent.

Under cache-aside, staleness is addressed by the rule that a writer, having updated the database, removes the entry from the cache. It may be asked why the writer does not place the new value in the cache directly, which would spare the next reader a miss. The reason is that 2 writers can then interfere. If writer 1 sets the database to 11 and writer 2 sets it to 12, and their subsequent writes to the cache happen to arrive in the opposite order, the database ends at 12 and the cache at 11. Removing the entry is immune to this, because removing something twice, in either order, leaves it removed.

The order of the 2 steps of a write also matters, the database being written before the entry is removed. If the entry were removed first, a reader arriving between the removal and the database write would miss, read the old value from the database, and put it back into the cache, where it would remain after the write had completed.

Even with both precautions observed, a stale value can become established, through an interleaving of a reader and a writer that is rare and entirely possible.

What makes this race instructive is that neither request did anything wrong. The reader followed the rule for reads and the writer the rule for writes, and the result is a cache that disagrees with the database and will go on disagreeing, since no further event is scheduled to correct it. The race needs a read to be delayed, between fetching from the database and storing in the cache, for longer than an entire write takes, which is unusual, and on a busy system unusual events occur many times a day.

There are 3 defences, of increasing strength. The first is to give every entry a time limit, discussed in the next section, so that even an entry that should have been removed disappears eventually. The second is for the writer to remove the entry a second time after a short delay, which catches a reader that stored its stale value just after the first removal. The third makes the reader’s store conditional. On a miss the cache hands the reader a token, the removal of an entry invalidates any token outstanding for it, and a store that presents an invalidated token is rejected. In the steps above, request 2’s removal would have invalidated the token of request 1, whose attempt to store 10 would then have failed, leaving the cache empty and correct.

Time limits

Every entry in a cache should carry a time to live, or TTL, after which the cache discards it regardless of anything else. The TTL is what converts every failure of invalidation, whether the race just described, a removal that was lost because the cache was briefly unreachable, or a code path that updates the database and forgets the cache, from a permanent error into a temporary one. It is the single mechanism that bounds staleness unconditionally, and for that reason it should be present even where explicit removal is believed to be complete.

Choosing the value is a judgement about the data. A long TTL gives a higher hit ratio and allows stale values to live longer, and a short one the reverse. The question to ask of each kind of data is how long a wrong answer could be shown before somebody is harmed. A product description may tolerate an hour, the number of items in stock a few seconds, and an account balance nothing at all, which is another way of saying that it should not be cached in this manner.

3 ways in which a cache fails under load

The failures considered so far concern correctness. The 3 considered next concern load, and each of them has brought down systems whose caches were functioning exactly as designed.

The stampede

When a popular entry expires or is removed, every request that arrives before it has been restored is a miss. For an entry that is read 500 times per second and takes 100 ms to recompute, some 50 requests will arrive during the recomputation, and under plain cache-aside every one of them queries the database for the same value.

If the query is expensive, the database slows down under 50 simultaneous copies of it, the recomputation takes longer, still more requests arrive during it, and the situation feeds on itself. This is called a stampede.

The principal remedy is to ensure that only 1 request recomputes a given entry. The first request to miss takes a short-lived lock on the key and proceeds to the database, and the others, finding the lock taken, wait briefly and read the cache again, by which time the value is usually there. A complementary technique avoids the miss altogether by refreshing popular entries shortly before they expire: a request that finds an entry close to its expiry may, with a probability that rises as the expiry approaches, recompute it early while all other requests continue to be served the existing value. A third approach is to serve the expired value for a short grace period while a single request fetches a fresh one in the background, which trades a little staleness for the complete absence of a stampede.

Penetration by keys that do not exist

A cache stores answers, and a request for something that does not exist has, on the face of it, no answer to store. Consequently every request for a nonexistent key misses the cache and reaches the database, which searches and finds nothing, and the next request for the same key does the same. An attacker who requests random ids, or merely a defective client that keeps asking for a deleted record, thus obtains unimpeded access to the database, the cache contributing nothing. This is known as cache penetration.

The straightforward remedy is to cache the absence. When the database reports that a key does not exist, the application stores a marker to that effect under a short TTL, and subsequent requests for the key are answered from the cache. Where the set of possible nonexistent keys is very large, as with random ids, the markers themselves could fill the cache, and a more economical structure is placed in front: a Bloom filter, which is a compact summary of the set of keys that do exist. It can state with certainty that a key is absent, and it occasionally reports an absent key as possibly present, in which case the request proceeds to the cache and the database as usual. A filter covering a billion keys occupies little more than a gigabyte.

The avalanche

The third failure arises when a large part of the cache becomes empty at once. This happens when many entries were loaded together and given the same TTL, for example by a job that warms the cache at startup, so that they all expire in the same instant. It happens on a larger scale when a cache node fails or is restarted. In either case the hit ratio falls abruptly, and by the arithmetic of the first section a fall from 99% to 90% multiplies the load on the database by 10.

Against synchronised expiry the remedy is to add a random amount to each TTL, so that entries loaded together expire over a spread of time. Against the loss of a node, the remedies are structural and are the subject of the next section. In both cases the database should be defended independently, by a limit on the number of concurrent queries that the application will send to it, so that an empty cache results in slow responses and some refusals while the database itself stays up. The rate limiter of chapter 7 and load shedding serve exactly this purpose.

A cache on many nodes

With 100 GB of entries and nodes of 16 GB, the cache has to be spread over several machines, and each key must be assigned to one of them in a way that every client computes identically. This is the problem that chapter 1 solved for the key-value store, and the solution is the same: a consistent hash ring on which each node owns many small ranges, so that adding or removing a node moves only the keys of that node.

The argument for consistent hashing is, if anything, stronger for a cache than for a store. A key that moves to a different node is, from the cache’s point of view, a key that has been forgotten, since the new node does not have it. Were keys assigned by taking a hash modulo the number of nodes, the addition of a single node would move nearly all keys, which is to say that it would empty the cache, with the consequences just described. On the ring, adding an eighth node to 7 moves about an eighth of the keys.

In this design the ring lives in a library inside each service, and the cache nodes are unaware of one another. A node holds entries, evicts and expires them, and that is all. There is no leader and no replication between nodes in the basic design, on the reasoning that every entry is a copy of something held durably elsewhere, so that the loss of a node loses nothing that cannot be fetched again.

That reasoning is sound with respect to data and incomplete with respect to load. The loss of 1 node in 8 turns an eighth of all reads into misses until those entries have been fetched again, and the database must be able to bear this. Where it cannot, each entry can be written to 2 nodes, the next one clockwise on the ring serving as a standby, at the cost of half the cache’s capacity. A related precaution concerns a node that returns after a restart with its memory empty: its entries can be loaded gradually by admitting only a fraction of its normal traffic at first, so that its misses arrive at the database as a trickle.

A last difficulty on many nodes is the hot key, a single entry so popular that the 1 node holding it is overloaded while the others are idle. No arrangement of the ring helps, since the ring spreads keys and this is 1 key. The usual answers are to keep a copy of such entries in the memory of each application process for a second or 2, in front of the distributed cache, or to store the entry under several derived keys on different nodes and have readers choose among them at random.

Use cases

Services read through the cache and inform it when the data behind an entry has changed. Operators add and remove nodes.

Functional requirements

Non-functional requirements

A cache is eventually consistent by its nature, and non-functional requirement 2 says so openly. For some period after a write, a reader may still receive the previous value, and the TTL of the entry is the bound on that period. The design does not attempt to remove this window. It attempts to make its length a deliberate choice for each kind of data, and to ensure, by means of the TTL, that no failure can extend it indefinitely.

Core entities

Capacity estimate

Arithmetic ahead. 100 GB at 16 GB of memory per node needs 7 nodes, and we run 8.

Skip to the high-level design

Show the calculation

Quantity	Calculation	Result
Memory needed	given	100 GB
Usable memory per node	16 GB × 0.9	14.4 GB
Nodes	100 ÷ 14.4	6.9, so 7
Reads per node	200000 ÷ 8	25000 per second

The API

Values are opaque bytes, as in the key-value store of chapter 1, and the time limit of an entry travels in a header.

Store a value for 60 seconds

PUT /v1/entries/user-42 HTTP/1.1
Host: cache.example
Cache-Control: max-age=60
Content-Length: 9

ada, 1815

HTTP/1.1 204 No Content

After 60 seconds the entry is gone, whether or not it was read.

Read a value that is not cached

GET /v1/entries/user-43 HTTP/1.1
Host: cache.example

HTTP/1.1 404 Not Found

Misses are ordinary answers: the caller reads the database and stores the result.

A miss is reported with 404, since from the point of view of HTTP the resource does not exist at that address. It is an ordinary outcome and no error, and the caller responds to it by reading the database and storing the result.

High-level design

One request, step by step

The steps follow the arrows of figure 9.8.

1. A service asks the cache library for user-42.
2. The library hashes the key on its ring and asks the 1 cache node that owns it. That node answers with the value, or with 404.
3. On 404, the service reads user-42 from the database, which is slower.
4. The service stores the value in the cache with a time limit, so that the next reader gets a hit.

A write follows the complementary path. The service updates the database first, and when the database has confirmed, it sends DELETE for the key to the cache node that owns it. If that DELETE fails because the node is unreachable, the entry survives with the old value until its TTL expires, which is the case the TTL exists for.

The design, deployed

The deployment has a feature that distinguishes it from most others in this book, which is the absence of a load balancer. Each of the 400 service instances contains the cache library with its copy of the ring, and sends every request directly to the 1 cache node that owns the key. The 8 nodes are spread over 3 zones. With a hit ratio of 99%, about 2000 of the 200000 reads per second go on to the database, whose followers serve them. The figure makes the dependence plain: the database tier is sized for the traffic that passes the cache, and would receive 100 times as much if the cache tier were lost.

How real systems do it

The most thorough public account of a cache at scale is Facebook’s paper on its use of memcached, and nearly every difficulty in this chapter appears in it with the remedy that Facebook adopted.

The stale value left by an overlapping read and write, and the stampede on a missing entry, are both solved there by 1 mechanism, the lease. On a miss, the cache gives the client a token that entitles it to fill the entry. A delete invalidates any outstanding token, so that a late fill with an old value is rejected, which is the conditional store described above. And the cache issues a token for a given key only at intervals, telling other clients that miss in the meantime to wait briefly and try again, which prevents the stampede. The paper reports that leases reduced the peak rate of database queries for a set of keys prone to stampedes from 17000 per second to 1300.

The paper also describes what it does about the avalanche. When a cache server fails, its keys are not moved to the remaining servers, which might then be overloaded in turn. Requests for them go to a small set of idle machines kept for the purpose, called the gutter. Invalidation is driven from the database’s own log of committed changes, which is more dependable than relying on every application to issue its deletes, and a cluster that starts empty is warmed by reading from a neighbouring cluster’s cache before it is allowed to reach the database.

On eviction, the rule of sampling a few entries and removing the least recently used among them is what Redis does by default, with 5 samples. The admission policy that compares the estimated frequency of a newcomer with that of the entry it would replace is TinyLFU, implemented in the Caffeine library. A study of the cache clusters at Twitter found, among other things, that for many workloads at realistic cache sizes plain first in, first out performed about as well as LRU, which has since led to a family of simple queue-based policies. The technique of refreshing an entry early with a probability that rises towards its expiry has been analysed formally and shown to be optimal in a certain sense.

Mistakes to avoid

Failures

Trade-offs

The simulation

References

• R. Nishtala and others, “Scaling Memcache at Facebook”, NSDI 2013.
• G. Einziger, R. Friedman and B. Manes, “TinyLFU: A Highly Efficient Cache Admission Policy”, ACM Transactions on Storage, 2017.
• J. Yang, Y. Yue and K. Rashmi, “A Large Scale Analysis of Hundreds of In-memory Cache Clusters at Twitter”, OSDI 2020.
• A. Vattani, F. Chierichetti and K. Lowenstein, “Optimal Probabilistic Cache Stampede Prevention”, VLDB 2015.
• Redis documentation, “Key eviction”, on approximated LRU.
• B. Bloom, “Space/Time Trade-offs in Hash Coding with Allowable Errors”, Communications of the ACM, 1970.

Terms introduced in this chapter