Design a Distributed Cache: Sharding, Eviction, Hot Keys and Stampedes

One read and one write, end to end

Follow a GET for a user's profile, then a change to that profile. This is a good order to explain it in an interview.

1. 1

   GET: local check

   The app server needs user:42. If user:42 is a known hot key, it first checks its own local cache, which holds such keys for about 2 seconds.

2. 2

   GET: pick the node

   The client library hashes the key user:42 to a number and finds the first ring point clockwise from it. That point belongs to cache node B.

3. 3

   GET: hit

   Node B finds user:42 in its hash table, moves it to the front of its LRU list, and returns the value. Most reads end here, in well under a millisecond.

4. 4

   GET: miss

   If B does not have it, B says so, and with leases it also hands back a token that allows this one client to refill the key. Other clients that miss now are told to wait a moment.

5. 5

   GET: refill

   The client reads user:42 from the database and calls set on B with the value, a TTL of, say, 300 seconds, and the token. B stores it only if the token is still valid.

6. 6

   SET: write the database

   The user changes their name. The app writes the new name to the database. The database is updated first because it is the source of truth.

7. 7

   SET: delete the key

   The cached user:42 is now out of date, so it is deleted from node B (and from any copies). It is deleted, not overwritten, because a delete can safely be repeated.

8. 8

   SET: backstop

   The invalidator, reading the database's commit log, also sends delete user:42. If the app crashed before its own delete, this one still arrives. The TTL is the last safety net.

9. 9

   Next GET

   The next read misses, refills from the database, and caches the new name.

These are the conversations the interviewer is steering you toward. Practice each one until you can talk through it without notes.

Cache-aside, write-through and write-back

There are three common ways to keep a cache and a database together. They differ in who writes what, and in what can go wrong. In cache-aside, also called lazy loading, the app does everything. On a read, it asks the cache. On a miss, it reads the database and puts the value into the cache. On a write, it updates the database and then deletes the cached key. Only data someone asked for gets cached, and a dead cache node only makes reads slower. The AWS ElastiCache guide lists the cost: a miss takes three trips (cache, database, cache again), and cached data can go stale. Facebook uses this pattern and calls it a demand-filled look-aside cache. Its paper explains why it deletes instead of updating the cached value: deletes are idempotent, which means doing one twice has the effect of doing it once. Two updates that arrive in the wrong order would leave the older value cached. In write-through, every write goes to the cache layer and to the database before the app gets an answer. Cached data is current, but each write takes two steps, and much of what is written is never read. A new, empty node also stays empty until its keys are written again, so in practice write-through is combined with lazy loading. In write-back, also called write-behind, the cache replies at once and writes the database later. Writes are very fast, and many writes to one key can be merged into one. But if the cache node dies before the write reaches the database, data the app was told was saved is gone. A general cache design should start with cache-aside plus a TTL on every key. Use write-back only for data you can afford to lose, such as view counters.

Eviction: LRU, LFU and TTL

Memory is limited, so a full node must throw something away. That is eviction. The goal is to keep the items most likely to be read again. LRU, least recently used, throws away the item that was read longest ago. It works well because popular items tend to be read again soon. The classic way to do it in O(1) time, meaning a fixed amount of work no matter how many items are stored, uses two parts. A hash map finds a key's entry in one step. A doubly linked list keeps entries in order of last use, newest at the front. Each list entry points both forward and backward, so it can be cut out without walking the list. A read moves the entry to the front. A write to a full cache removes the entry at the back. Real systems bend this idea. Redis does not keep an exact list, because the pointers cost memory. It samples a few keys at random, 5 by default, and evicts the oldest of them. Its docs say this comes very close to true LRU. Memcached splits its LRU into HOT, WARM and COLD parts. New items enter HOT, and items that are read again move to WARM. This protects busy items from a one-time scan of many keys that are never read again. LFU, least frequently used, evicts the item read the fewest times. It is better when some keys are steadily popular. It needs a counter that fades over time, or an item that was popular last week never leaves. Redis's LFU uses a small probabilistic counter that is reduced every minute by default. TTL is separate from eviction. A TTL says how long an item may live even if there is room, which bounds how stale it can get. Set a TTL on every key. Add a little randomness to it (for example 300 seconds plus or minus 30), so keys written together do not all expire together.

class Node:
    __slots__ = ("key", "val", "prev", "next")
    def __init__(self, key=None, val=None):
        self.key, self.val, self.prev, self.next = key, val, None, None

class LRUCache:
    def __init__(self, capacity):
        self.cap, self.map = capacity, {}          # map: key -> list node
        self.head, self.tail = Node(), Node()      # two empty end markers
        self.head.next, self.tail.prev = self.tail, self.head

    def _unlink(self, n):
        n.prev.next, n.next.prev = n.next, n.prev

    def _push_front(self, n):                      # front = most recently used
        n.prev, n.next = self.head, self.head.next
        self.head.next.prev = n
        self.head.next = n

    def get(self, key):
        n = self.map.get(key)
        if n is None:
            return None                            # miss
        self._unlink(n)
        self._push_front(n)
        return n.val

    def set(self, key, val):
        n = self.map.get(key)
        if n is not None:
            n.val = val
            self._unlink(n)
        else:
            if len(self.map) >= self.cap:          # full: drop the oldest
                old = self.tail.prev
                self._unlink(old)
                del self.map[old.key]
            n = Node(key, val)
            self.map[key] = n
        self._push_front(n)

c = LRUCache(2)
c.set("a", 1); c.set("b", 2)
c.get("a")          # a is now the newest
c.set("c", 3)       # full: b is the oldest, so b goes
print(c.get("b"), c.get("a"), c.get("c"))   # None 1 3

Sharding with consistent hashing and virtual nodes

Sharding means splitting the keys so that each node holds only part of them. The simplest rule is node = hash(key) mod N, where N is the number of nodes and mod is the remainder after division. It spreads keys well, but it breaks badly when N changes. Going from 14 to 15 nodes changes the remainder for about 14 keys in 15. In our test, 93.3% of keys moved. For a cache, a moved key is a miss, so adding one node would empty most of the cache and send that load to the database. Consistent hashing fixes this. It comes from a 1997 paper by Karger and others, written to spread web caching load; the paper describes a hash function that 'changes minimally' when the set of caches changes. Picture a circle of numbers from 0 to 2^32 - 1. Each node is hashed to points on that circle. Each key is hashed to a spot on the circle too, and it belongs to the first node point clockwise. When a node joins, it takes over only the keys just before its points. When a node leaves, only its keys move. With one point per node, though, the gaps between points are uneven, and the circle gives one node a much bigger share. The fix is virtual nodes: each real node gets many points. Amazon's Dynamo paper lists the gains. A failed node's load spreads evenly over the remaining nodes. A new node takes a little load from each existing node. And a bigger machine can be given more points. Twemproxy's ketama code uses 160 points per server. Running the code below on 200,000 keys and 14 nodes: with 1 point per node, the busiest node held 2.7 times its fair share; with 160 points, 1.13 times. Adding a 15th node moved 6.9% of keys, close to the ideal of 1 in 15 (6.7%). Redis Cluster uses a different scheme: a fixed set of 16,384 hash slots, slot = CRC16(key) mod 16384 (CRC16 is a simple checksum function), with each primary owning a range of slots. Moving a slot moves its keys, and clients learn the new owner from a redirect reply.

import bisect, hashlib

def h(s):                                          # a 32-bit spot on the ring
    return int(hashlib.md5(s.encode()).hexdigest()[:8], 16)

class Ring:
    def __init__(self, nodes, vnodes=160):
        self.points = sorted((h(f"{n}#{i}"), n) for n in nodes for i in range(vnodes))
        self.spots = [p for p, _ in self.points]

    def node_for(self, key):
        i = bisect.bisect_right(self.spots, h(key))   # first point clockwise
        return self.points[i % len(self.points)][1]   # past the top: wrap to 0

keys = [f"user:{i}" for i in range(200_000)]
old = Ring([f"cache-{n}" for n in range(14)])
new = Ring([f"cache-{n}" for n in range(15)])
moved = sum(old.node_for(k) != new.node_for(k) for k in keys) / len(keys)
print(f"{moved:.1%} of keys moved")

Routing: client library, proxy, or redirects

Something has to turn a key into a node address. There are three places to put that logic. The first is a client library inside each app. It is the fastest, because there is no extra hop. Memcached works this way: the servers do not know about each other, and the docs note that finding the server is just a hash lookup on the client. The weak point is that every client must hold an identical node list. Karger's paper calls each client's list its 'view', and its analysis allows views to differ, but two different lists still send one key to two nodes for a while. The second is a proxy. Apps send everything to a local or nearby proxy, which routes the request. Facebook's mcrouter is described in its README as a memcached protocol router, and it handles almost 5 billion requests a second at peak across Facebook and Instagram. It adds connection pooling (keeping a few connections open and reusing them), replicated pools, failover and cold cache warm-up. Twemproxy was built mainly to cut the number of connections to the cache servers. It merges many client connections into a few server connections, and supports ketama consistent hashing. Facebook uses both styles at once: gets go straight from the client to memcached over UDP, a lighter network protocol, while sets and deletes go over TCP, the usual reliable protocol, through an mcrouter running on the web server itself. The third is redirects from the servers themselves. In Redis Cluster, a client may ask any node; a node that does not own the key replies MOVED with the right address, and smart clients cache the slot map so later requests go direct. Pick the client library when you control every app, a proxy when many languages or very many connections are involved, and redirects when you use Redis Cluster. The Design Redis page explains the slot map and resharding in more depth.

Replication and failover

A cache can lose data, because the database still has it. So why replicate at all? Because a dead node does not just lose data; it moves load. In our numbers, losing 1 of 14 nodes with no copy sends database reads from 50,000 to about 118,000 a second. There are three common answers. The first is a replica per node. Each primary (the main copy) sends its writes to a replica, and if the primary dies, the replica is promoted and serves its keys. Replication is usually asynchronous, which means the primary answers before the replica has the write. The Redis Cluster spec says plainly that there are small windows where acknowledged writes can be lost. For a cache this is usually fine, with one trap: a lost delete. If a delete reached the old primary but not the replica, the promoted replica still serves the old value. The TTL is what limits how long that lasts. The second answer is a spare pool. Facebook's Gutter is a small set of machines, about 1% of a cluster. When a client gets no answer from a node, it retries against Gutter, and on a Gutter miss it reads the database and fills Gutter, with a short expiry. Facebook reports this cut client-visible failures by 99% and turned 10% to 25% of failures into hits each day. The third answer is the tempting wrong one: rehash the dead node's keys onto the remaining nodes. Facebook's paper warns against it, because one key can account for 20% of a server's requests, and the node that inherits a hot key can be overloaded in turn. Failure detection matters too. Clients mark a node down after a few timeouts in a row; twemproxy has a setting for this, auto_eject_hosts. Mark it down too quickly and a short network blip moves keys for nothing. Too slowly, and requests wait on a dead machine. Finally, an empty cluster is dangerous. Facebook's cold cluster warmup lets a new cluster read from a warm one instead of the database, and says this brings a cold cluster to full capacity in a few hours instead of a few days.

Hot keys

Consistent hashing spreads keys evenly, but it cannot spread a single key. If one celebrity's profile is read 200,000 times a second, the one node that owns it gets all 200,000 on top of its normal load. In our numbers that node goes from 71,400 to 271,400 reads a second, far over a 100,000 budget. You first have to find hot keys: count reads per key on each node or in the client, and alert on keys above a threshold. Then there are three fixes. The first is a local cache in each app server, holding hot keys for a second or two. With 1,000 app servers and a 2-second local TTL, at most 1,000 / 2 = 500 reads a second for that key reach the cache cluster. The cost is that each app server may be up to 2 seconds behind. The second is to store several copies of the key under different names, such as key#0 to key#7, which land on different nodes. Readers pick one at random, so each copy gets 200,000 / 8 = 25,000 reads a second. Every write must delete all eight copies. The third is replication of a whole group of keys inside a pool. Facebook does this when an app reads many keys at once, the whole set fits on one or two servers, and the request rate is more than one server can manage. Its paper shows why splitting the keys would not help in that case: a request for 100 keys split over two servers still costs each server one request, so each server's request rate does not drop. With full copies, each request goes to one replica, and the load per server halves. One more source of load is reads for rows that do not exist. If a missing user id is never cached, every request for it reaches the database, so cache the 'not found' answer too, with a short TTL. Hot writes are a different problem: if one key is written thousands of times a second, its cached copy is deleted constantly and every reader misses. That case needs the stampede protection in the next section, or a cache with a short TTL that is not deleted on every write.

Cache stampedes: leases, request coalescing and early refresh

A cache stampede, also called a thundering herd, happens when a popular key disappears, by expiry, eviction or a delete, and many requests miss at once. Every one of them goes to the database to rebuild one value. The database slows down, the rebuild takes longer, and more requests pile up behind it. Facebook's paper describes the cause plainly: a key with heavy reads and writes, where each write invalidates the value and many reads fall through to the costly path. There are three standard fixes. The first is leases, from that paper. On a miss, the cache gives the client a lease, a 64-bit token tied to that key. Only a set that carries a valid token is stored. The server hands out at most one token per key every 10 seconds; other clients that miss in that time are told to wait a short time and retry. Usually the lease holder has set the value within a few milliseconds, so the retries hit. Over one week, for keys prone to stampedes, peak database reads fell from 17,000 to 1,300 a second. Leases also fix stale sets: if a delete for the key arrives while the lease holder is still reading the database, the token is cancelled, so an old value cannot be written over the newer state. The second fix is request coalescing inside each app server: when many threads miss on one key, one of them reads the database and the rest wait for its result. Go's singleflight package does this, making sure that only one call is in flight for a given key at a time. It limits the herd to one request per app server, not one in total. Outside Facebook's memcache, a short lock key does the job of a lease, as in the code below. The third fix refreshes a key before it expires. The XFetch paper (VLDB 2015) has each reader, near the expiry time, decide at random whether to recompute early, with the chance rising as expiry gets closer and as the recompute gets slower. One reader usually refreshes the value while the others keep getting hits. Random jitter on TTLs helps too, by spreading expiries out.

import time

def get_or_load(cache, key, load, ttl=300, lock_ttl=10):
    val = cache.get(key)
    if val is not None:
        return val                                  # hit
    if cache.add(f"lock:{key}", 1, lock_ttl):       # only one caller wins this
        try:
            val = load(key)                         # the one database read
            cache.set(key, val, ttl)
        finally:
            cache.delete(f"lock:{key}")
        return val
    for _ in range(50):                             # everyone else: wait up to ~1 s
        time.sleep(0.02)
        val = cache.get(key)
        if val is not None:
            return val
    return load(key)                                # still nothing: read the source

Consistency and invalidation

A cache holds copies, and copies can be out of date. The design question is how out of date, and for how long. The most common bug is a race in cache-aside. Reader A misses and reads the old value from the database. Before A writes it to the cache, writer B updates the database and deletes the key. Then A sets the old value. Now the cache holds stale data until its TTL ends. Leases stop this, as described above, because B's delete cancels A's token. A versioned set does too: memcached's cas command stores a value only if no one else has updated the key since the caller last read it. Second, deletes can get lost: the app can crash between writing the database and deleting the key. The fix is to drive deletes from the database itself. Facebook runs a background program called mcsqueal on every database. It reads the statements the database committed, pulls out the cache keys to delete, and sends them to the cache in every front-end cluster (a group of web servers and their cache servers) in that region. It batches them into fewer packets through mcrouter; Facebook measured 18 times more deletes per packet (the median) this way. Most of these deletes find nothing to delete: only 4% of them removed cached data. Third, across regions, a delete can arrive before the database change itself reaches that region's replica database. A read in between would put old data back into the cache. Facebook avoids part of this by sending invalidations from the database side, after the change is committed, and uses markers to send some reads to the main region while a change is still on its way. Last, the TTL is the safety net for every case above. Pick it from how stale the data may be. A product price might allow 60 seconds; a user's own just-saved profile should be deleted on write, never left to the TTL. Facebook's paper sums up its own choice as best-effort eventual consistency, with the emphasis on performance and availability. In an interview, say your own version out loud: which reads may be stale, for how long, and what bounds it.

Memory, multi-gets and the network

Three practical details separate a working cache from a slow one. First, memory is handed out in chunks. Memcached splits its memory into 1 MB pages, gives each page to a slab class (a group of chunks of one size), and cuts it into equal chunks. An item goes into the smallest chunk that fits, so the rest of the chunk is wasted. The docs show a 50-byte item losing 30 bytes in a class of small chunks. Facebook's version uses slab classes from 64 bytes up to 1 MB, each about 1.07 times the one before. That is why the estimate above adds 25% headroom instead of using the raw 550 GB. Second, pages need many keys. On Facebook, loading a popular page fetched 521 distinct items from memcache on average. So the client sends gets in batches, a multi-get, averaging 24 keys per request in Facebook's paper. Third, the network can choke. When one web server asks many cache servers at once, all the replies arrive together and overflow the network switch buffers. This is called incast congestion. Facebook's clients limit how many requests are outstanding with a sliding window, which grows slowly while requests succeed and shrinks when a request goes unanswered. Facebook also sends gets over UDP, which skips the setup cost of a TCP connection; under peak load, 0.25% of those gets were dropped, and the client treats a dropped get as a miss.

How a Distributed Cache actually does it

The best documented distributed cache is Facebook's memcache, described in 'Scaling Memcache at Facebook' (NSDI 2013). It handled billions of requests a second and held trillions of items. It is a demand-filled look-aside cache: web servers read memcache first, fill it on a miss, and delete keys after writing to MySQL. Keys are spread over hundreds of memcached servers per cluster by consistent hashing, and the servers never talk to each other. Leases solve two problems: stale sets and thundering herds. One token per key every 10 seconds cut peak database reads for stampede-prone keys from 17,000 to 1,300 a second. A Gutter pool of about 1% of the servers absorbs the load of failed ones, cutting client-visible failures by 99%. An invalidation daemon, mcsqueal, reads the database's commits and sends deletes in batches. Facebook's mcrouter, now open source, routes memcached requests and handles almost 5 billion requests a second at peak across Facebook and Instagram. Twitter's twemproxy is a lighter proxy for memcached and Redis, built to cut connections to the cache servers, with ketama consistent hashing at 160 points per server. The ring itself comes from Karger and others (STOC 1997), and Amazon's Dynamo paper (SOSP 2007) made virtual nodes standard. Redis Cluster splits keys into 16,384 slots, uses asynchronous replication, and says openly that acknowledged writes can be lost in some failovers. Memcached's own docs describe client-side hashing, the slab allocator, and a segmented HOT, WARM and COLD LRU. Two research results finish the picture: the XFetch paper on refreshing keys early at random to prevent stampedes, and Go's singleflight package for collapsing duplicate misses inside one process.

Frequently asked questions