sehaj/learn
System design/06 Caching

06.06 · concept · 12 min

Designing a distributed cache

Designing the cache tier itself: sizing it, spreading keys over nodes, routing requests, replicas and failover, what happens when a node dies, and caches in more than one region.

The problem: Snip's cache is now a system of its own

Snip's four Redis nodes from 05.03 Consistent hashing were enough for 100,000 cache reads a second. A year later the targets are:

  • 300,000 reads a second at peak, almost all redirects, plus a few thousand writes (new links, edits, deletes).
  • 40 million links in the working set.
  • p99 under 2 ms for a cache read, so the cache stays far faster than the database.
  • Losing a node mustn't overload Postgres, which is sized for about 1% of reads.

Treat it like any design (01.07 The design framework, step by step): estimate, choose how keys are placed and found, then decide what happens when things fail.

Sizing

Memory. A cached link is a key, a URL and the cache's own bookkeeping per entry: say 500 bytes in all. 40 million × 500 bytes is 20 GB. Leave room for growth and for memory fragmentation, and plan for about 32 GB of primaries.

Throughput. Snip's load tests show a node comfortably serving 50,000 reads a second at the latency target. 300,000 ÷ 50,000 is 6 nodes at full load. To leave headroom, and so that losing one node moves only a small share of keys, Snip runs 12 primaries of about 4 GB each, each with one replica.

Running more, smaller nodes has a real benefit: when one fails, a twelfth of the keys are affected, not a quarter.

Finding the right node

Every request must reach the node that owns its key. There are three ways to arrange that:

ApproachHow a key finds its nodeTrade-off
Client-side hashingEach app server's client library hashes the key onto a ring (05.03 Consistent hashing)No extra hop; every client must agree on the node list
A proxyApps talk to a local proxy (such as mcrouter or twemproxy), which routesOne place for routing and config; one more hop
A cluster-aware storeThe nodes know the layout and redirect clients that ask the wrong oneSelf-managing; clients must support it

Redis Cluster is the third kind. It divides the key space into 16,384 hash slots: a key's slot is CRC16(key) mod 16384, and each primary owns a range of slots. A client that asks the wrong node gets -MOVED 3999 10.0.0.7:6379 and updates its map. While a slot is being moved to another node, the client may get ASK, meaning "try that node for this one request". Slots are fixed logical partitions, exactly like Snip's 256 logical shards (05.04 Rebalancing without downtime): adding a node moves whole slots, not individual keys.

Hash tags let you keep related keys together: only the part inside {...} is hashed, so {x7Kp}:url and {x7Kp}:owner land on the same node and can be read in one command. Use them sparingly, since one popular tag puts all its keys on one node.

Replicas and failover

Each primary copies its writes to a replica asynchronously. If a primary dies, the replica is promoted. Because copying is asynchronous, the last few writes before a crash can be lost, and the Redis Cluster specification says so plainly: there are "small windows where acknowledged writes can be lost".

For a cache, that's acceptable: the database is the source of truth, and a lost entry is just a miss. A lost delete is worse, since the promoted replica may still hold a value that was invalidated. That's another reason every entry has a TTL (06.04 Invalidation and staleness).

Replicas can also serve reads, which adds read capacity, at the price of reading slightly behind the primary.

When a node fails

Without a replica, or during the seconds before a replica takes over, a node's keys are unreachable. What should clients do?

  • Treat it as a miss and go to the database. Simple, but a twelfth of Snip's cached traffic, 25,000 reads a second, suddenly hits a database sized for 3,000. It can cascade.
  • Rehash onto the remaining nodes. With a ring, the failed node's keys spread to its neighbours, which start cold. A single hot key can account for a big share of a node's traffic, so the neighbour that inherits it can be overloaded in turn.
  • Send them to a small spare pool. Facebook's answer, the Gutter pool: about 1% of servers, idle normally, that take over failed servers' requests with short-lived entries. Misses there go to the database once, and then become hits (below).

Snip uses replicas for failover, and limits the misses any one app server can send to Postgres at once (03.11 Backpressure and load shedding), so a slow failover degrades latency instead of taking the database down.

The client matters as much as the servers

  • Timeouts. A cache read that takes 50 ms is worse than none. Give cache calls a tight timeout (a few milliseconds), and treat a timeout like a failed node, not as a reason to wait.
  • Batch, but watch the tail. Reading 20 keys in one multi-get is far cheaper than 20 round trips, but a request that waits for 10 nodes is as slow as the slowest of them (03.12 Tail latency and why p99 matters).
  • Hot keys. Spread them with a small in-process cache and suffixed copies (05.02 Hot partitions and celebrity keys).
  • Coalescing. Collapse concurrent misses for the same key (06.05 Stampedes and request coalescing).

More than one region

If Snip serves from Virginia and Frankfurt, each region gets its own cache, filled from its own database replica, because a cache read across an ocean costs more than a local database read. The hard part is invalidation: an edit in Virginia must delete the entry in Frankfurt's cache too, and not before Frankfurt's database replica has the new row, or a miss there will refill the old value. Driving deletes from each region's own replication stream solves both: a region deletes a key when its database applies the change (06.04 Invalidation and staleness).

Snip's cache tier
getcopiesmisschangesdeleteSnip appin-process L1Cluster client16,384 slotsPrimaries ×12~4 GB eachReplicas ×12failoverPostgreslink shardsInvalidatorreads the WAL

5 steps. Use Next and Back to follow the request.

Text version
  1. A redirect first checks the app's own in-process cache, for the few thousand hottest codes (05.02).
  2. Otherwise the cluster client hashes the code to one of 16,384 slots and asks the primary that owns it: one round trip.
  3. Each primary copies to a replica, which takes over if it fails. Losing one of 12 primaries affects a twelfth of keys.
  4. Misses, coalesced per key, go to the link's Postgres shard. The miss rate is what Postgres is sized for.
  5. An invalidator reads committed changes from the database's log and deletes the affected keys, whoever made the change (06.04).
  • Snip app → Cluster client
  • Cluster client → Primaries ×12: get
  • Primaries ×12 → Replicas ×12: copies
  • Cluster client → Postgres: miss
  • Postgres → Invalidator: changes
  • Invalidator → Primaries ×12: delete

Try it

Design it yourself firstmidseniorcaching

An online shop wants a cache for its product pages: 5 million products, each rendered as about 20 KB of JSON, read 200,000 times a second at peak and changed a few hundred times a second by merchants. During a flash sale, one product can get a fifth of all reads. Design the cache tier: how big, how many nodes, how keys are placed, what happens when a node dies, and how changes reach the cache.

15:00

Seen in production

Facebook

2013

memcache: Gutter and cold clusters

Facebook dedicates about 1% of the memcached servers in a cluster to a pool called Gutter. When a client gets no answer from a server, it retries in Gutter, filling it from the database on a miss, with entries that expire quickly. The paper explains why it doesn't rehash onto the remaining servers: a single key can account for 20% of a server's requests, and the server inheriting it could be overloaded. Gutter cut client-visible failures by 99%, and when a server failed entirely, Gutter's hit rate usually exceeded 35% within four minutes. New or emptied clusters warm up by reading from a warm cluster instead of the database, which brings them to full capacity in a few hours instead of a few days.

What we foundSend a failed server's traffic to idle spares, not to its neighbours, and warm new caches from old ones.

Scaling Memcache at Facebook (NSDI 2013)usenix.org ↗

Redis

Docs

Redis Cluster

The key space is split into 16,384 slots, with HASH_SLOT = CRC16(key) mod 16384; hash tags in braces make related keys share a slot. A node that doesn't own a key's slot replies with a MOVED redirection, and ASK covers slots in the middle of moving. Replication between nodes is asynchronous with "last failover wins", so the specification warns that there are small windows where acknowledged writes can be lost.

What we foundFixed hash slots make resharding a matter of moving slots, and the spec is honest that async replication can lose writes.

Redis cluster specificationredis.io ↗
Predict first

One of Snip's 12 cache primaries fails, and its replica takes 10 seconds to be promoted. Clients treat every error as a miss and go to Postgres. What's the risk, and the fix?

Commit to an answer before you read on.

Recap

Interview questions

SeniorfreqMust-knowDesign

Design a distributed cache.

Report a mistakeEdit this page
☰ Caching06.06 · 52/147next →