Distributed cache simulator

A cache cluster sits in front of a database that can answer only so many queries at once. Here 2,000 requests a second for 1,000 keys, a few of them very popular, are simulated one by one. Add and remove nodes, restart the cache, and watch the database when popular keys expire, with and without the usual protections.

Hit rate, last 5 s
–
Database queries/s
–
Nodes
4
Key placement
Nodes
4
Protection
Events
Time
The last 60 simulated seconds t = 0 s
Hit rate50%Database queries per secondcapacity 400/sSlowest 1% of requests (p99)80 ms, one query

queries, database keeping up queries, with a queue at the database

Nodes: share of requests in the last second
  • node 0 – 0 keys
  • node 1 – 0 keys
  • node 2 – 0 keys
  • node 3 – 0 keys

Even when keys are spread evenly, requests are not: the hottest key alone gets 16% of all requests, and it lives on one node.

Since the start
Requests
0
Served from cache
–
Served stale
0
Database queries
0
…for the hottest key
0
Most queries in flight
0
In flight now
0 of 32 connections
What happened
  1. –The cache starts empty, so the first seconds are all misses. Then try a button above.

How to use the simulator

The simulation runs in simulated time, one request at a time, at 2,000 requests a second. Each request picks a key, the cache node that owns the key answers if it holds a fresh copy, and otherwise the request queries the database, which has 32 connections and takes 80 ms per query. Entries live for 10 seconds. The charts show one bar per simulated second.

  • Watch the first seconds: the cache starts empty and the database is swamped. Press Restart cache later to see it again, then turn on Coalesce and restart once more.
  • Every 10 seconds the hot keys expire together and the query chart spikes. Turn on TTL jitter to spread expiries out, or Serve stale to answer from the old copy while one request refreshes it.
  • Press + under Nodes with hash mod N selected, and again with Hash ring. The log reports the share of keys that moved and the hit rate in the five seconds after.
  • Update hottest key writes a new value and deletes the cached copy. The log counts how many database queries the next second brought for that one key.

What is a distributed cache?

A distributed cache is a key-value store held in the memory of several machines, placed in front of a database or service so that repeated reads are answered quickly and the slower system sees only a fraction of the traffic. Each key lives on one node (or a primary and its replicas), chosen from a hash of the key, so the cache's total size and throughput grow with the number of nodes.

The arithmetic is what makes caches worth running: if 95% of reads are hits, the database handles one read in twenty. It is also what makes them dangerous. A database sized for the misses of a warm cache cannot handle the load of a cold one, so anything that makes many keys miss at once (a restart, a change in the number of nodes, popular keys expiring together) can overload it.

Reading and writing through a cache

  • Cache-aside (lazy loading): the application looks in the cache; on a miss it reads the database and stores the result with a TTL. On a write it updates the database and deletes the cached entry, so the next read loads the new value. This is the pattern simulated here.
  • Read-through: the same, but the cache itself loads missing entries from the database.
  • Write-through: writes go to the cache, which writes them to the database before acknowledging, so recently written data is always cached.
  • Write-back (write-behind): writes are acknowledged once in the cache and written to the database later, in batches. Writes are fast, but anything not yet written is lost if a node fails.

Deleting on write, rather than writing the new value into the cache, avoids one race: two writers updating the cache in a different order from the database. It leaves another, in which a slow reader puts an old value back after the delete. Facebook's memcache deployment handles that with leases: on a miss, the cache hands the reader a token, and a later delete invalidates the token so the old value cannot be stored.

Placing keys on nodes

The obvious way to pick a node is hash(key) mod N. It spreads keys evenly, but when N changes almost every key gets a new node: a key stays put only if its hash gives the same remainder for both values of N. Going from N to N + 1 nodes moves about N/(N + 1) of all keys, and every moved key is a miss.

Consistent hashing (Karger and others, 1997) places nodes and keys on the same circle of hash values; each key belongs to the first node clockwise from it. A new node takes over only the keys between itself and its predecessor, about 1/(N + 1) of them. Measured on 5,000 keys with this page's hash function:

Nodes, before → afterKeys moved, mod NKeys moved, ringIdeal minimum
3 → 475.0%23.7%25.0%
4 → 580.1%21.6%20.0%
5 → 683.5%18.3%16.7%
8 → 988.5%11.9%11.1%

In the simulation, adding a fifth node to a warm cache with coalescing and jitter on dropped the hit rate over the next five seconds from 95.5% to 90.1% with mod N, and database queries from 380 to 791. With the ring the hit rate fell to 93.7% and queries rose to 538.

A ring with one point per node divides the circle unevenly. The usual fix is virtual nodes: each node is hashed onto the ring many times under different names. With four nodes and 4,000 keys:

Points per nodeBusiest nodeQuietest node
11.58 × fair share0.70 × fair share
101.30 × fair share0.65 × fair share
1001.13 × fair share0.92 × fair share

The simulator uses 100 points per node. Redis Cluster takes a different route to the same goal: it splits keys into 16,384 hash slots (CRC16 of the key modulo 16,384) and assigns slots to nodes explicitly, so adding a node means migrating some slots to it. Memcached servers do not talk to each other at all; the client library decides which server holds each key, commonly with a consistent hashing scheme.

Cache stampedes

A cache stampede (also thundering herd or dogpile) is what happens in the gap between a popular key going missing and its value being stored again. Every request for that key in the gap misses and queries the database. If the hottest key gets a few hundred requests a second and the query takes 80 ms, one expiry turns one query into dozens. If the extra queries push the database past its capacity, queries queue, the gap gets longer, and more requests pile in.

The simplest case is a cold start. In the first five simulated seconds after the cache starts empty:

CoalescingDatabase queriesMost in flightSlowest 1% of requests
off1,6875821438 ms
on860142340 ms

The database here has 32 connections at 80 ms a query, so it can answer 400 queries a second. Without coalescing, 582 queries were running or queued at the worst moment and the slowest 1% of requests took 1438 ms; with it, 142 and 340 ms. A real database under that kind of queue tends to do worse than a fixed queue, as connections time out and clients retry.

Stampedes also happen in steady state. Keys loaded at the same moment, as after a cold start, expire at the same moment, and the hottest keys expire every TTL. Facebook reported that for a set of keys prone to thundering herds, its lease mechanism (which lets one client per key fetch from the database and tells the others to wait briefly) cut the peak database query rate from 17,000 to 1,300 a second.

Coalescing, stale data and jitter

  • Coalescing (request collapsing): when a request misses and a query for the same key is already running, wait for that query's answer instead of starting another. Go's singleflight package does this within one process, nginx's proxy_cache_lock within one proxy, and Facebook's memcache leases across a whole cluster.
  • Serve stale while revalidating: keep an expired entry for a grace period; a request in that period gets the old value at once while one background request refreshes it. HTTP caches support this as the stale-while-revalidate directive (RFC 5861), and nginx as proxy_cache_background_update. The cost is that some readers see data up to the TTL plus the grace period old.
  • TTL jitter: randomize each entry's TTL (here by ±20%) so that keys loaded together expire apart.
  • Probabilistic early expiration: refresh an entry before it expires, with a probability that rises as expiry approaches and with how long the value takes to compute. Vattani, Chierichetti and Lowenstein (2015) showed an optimal form, often called XFetch. It is not simulated here.

Each combination was run for 60 simulated seconds. The hottest key's queries are counted over the whole run, the other columns after the first ten seconds, once the cold start is over:

CoalesceStaleJitterHottest key queriesMost in flightSlowest 1%Hit rateServed stale
offoffoff15661120 ms95.0%0.0%
offoffon1453780 ms95.1%0.0%
offonoff24210 ms99.6%4.5%
offonon25180 ms99.6%4.5%
onoffoff62380 ms95.0%0.0%
onoffon61980 ms95.0%0.0%
ononoff6230 ms99.6%4.6%
ononon6190 ms99.6%4.6%

Coalescing cuts the hottest key's queries from 156 to 6 over the run: one per expiry instead of one per request that arrives during the query. Serving stale data alone cuts them to 24 and lifts the hit rate from 95.0% to 99.6%, because requests for an expired popular key get the old copy while a single request refreshes it; in exchange 4.5% of responses were older than the TTL. With all three on, the most queries in flight drops from 61 to 19. A slowest-1% time of 0 ms means at least 99% of requests in each second were hits, which this model counts as taking no time.

How the simulation works

Requests arrive evenly, 2,000 a second. Each picks one of 1,000 keys with Zipf-distributed popularity (exponent 1.05), so key 0 is the hottest. Each node holds up to 400 entries with least-recently-used eviction. The database runs 32 queries at a time, each taking 80 ms; others wait in line. Keys are placed with FNV-1a followed by MurmurHash3's finalizer, by modulo or on a ring with 100 points per node. Network time between client and cache is not modelled, so a hit takes 0 ms.

The model is checked headless: the share of keys moved when a node is added matches N/(N + 1) for modulo and 1/(N + 1) for the ring to within a few percent, a single hot key with coalescing costs one query per TTL, and a read after an update returns the new value. The numbers in this article are computed from the same code when the page is built.

History

David Karger, Tom Leighton, Daniel Lewin and colleagues at MIT introduced consistent hashing in 1997 as a way to spread web caching across many machines; Leighton and Lewin went on to found Akamai. Brad Fitzpatrick wrote Memcached in 2003 for LiveJournal, and it became the standard cache tier for large websites; Facebook's 2013 paper "Scaling Memcache at Facebook" describes running it across clusters and regions, including leases against stale sets and thundering herds. Redis, first released by Salvatore Sanfilippo in 2009, gained Redis Cluster with hash slots in version 3.0 in April 2015.

Distributed cache questions and answers

What is a cache stampede?

A cache stampede, also called a thundering herd or dogpile, happens when a popular entry is missing from the cache and every request for it goes to the database at once, until one of them puts the value back. With 2,000 requests a second and a database query taking 80 ms, this page's hottest key sends 26 or so queries to the database each time it expires, where one would do.

How do you prevent a cache stampede?

Three common fixes: coalescing (only one request per key goes to the database and the others wait for its answer, as in Go’s singleflight or nginx’s proxy_cache_lock), serving the old value for a short time while one request refreshes it in the background (stale-while-revalidate), and spreading expiry times with random jitter or probabilistic early refresh so that keys loaded together do not expire together.

Why use consistent hashing for a cache cluster?

With hash(key) mod N, changing N changes the node for most keys: going from 4 to 5 nodes moves 80% of keys on this page, and each moved key is a cache miss. With a hash ring only the keys in the arcs taken over by the new node move: about 1/(N + 1) of them when going from N to N + 1 nodes (22% here, against an ideal 20%).

What are virtual nodes in consistent hashing?

Each physical node is placed on the ring many times under different names, so the arcs it owns are many and small rather than one large one. That evens out the load: with one point per node, the busiest of four nodes on this page gets 1.58 times its fair share of keys; with 100 points per node, 1.13 times.

What is the difference between cache-aside and write-through?

In cache-aside (lazy loading) the application reads the cache, and on a miss reads the database and fills the cache itself; writes go to the database and delete or update the cache entry. In write-through the application writes to the cache, which writes to the database synchronously, so the cache always holds recent writes. Write-back (write-behind) acknowledges the write once it is in the cache and writes to the database later, which is faster but can lose data.

Is Redis Cluster consistent hashing?

Not exactly. Redis Cluster splits the key space into 16384 hash slots (CRC16 of the key modulo 16384) and assigns slots to nodes. Adding a node means moving some slots to it, so as with a hash ring only those keys move, but the assignment is an explicit table rather than positions on a ring.

Further reading

Published Updated