sehaj/learn
System design/05 Partitioning and IDs

05.03 · with code · 13 min

Consistent hashing

Mapping keys to servers so that adding or removing one moves only a fair share of keys, not nearly all of them. The ring, virtual nodes, and a small implementation.

The problem: hash mod N moves almost everything

At peak, Snip's redirects make about 100,000 cache reads a second, spread over four Redis nodes. Each code goes to node hash(code) mod 4. The cache is running hot, so you add a fifth node, and the formula becomes hash(code) mod 5.

For most codes, mod 5 gives a different answer from mod 4. About 80% of keys now map to a different node, where they aren't cached. For the next few minutes, 80% of redirects miss and fall through to the database shards: tens of thousands of extra queries a second, the moment you tried to add capacity. Removing a node (or losing one) does the same.

Only a fifth of the keys needed to move, the share the new node should take. Consistent hashing moves roughly that share and no more.

The ring

Picture the output of the hash function, 0 to 2³²−1, bent into a circle.

  1. Place each server on the ring by hashing its name: hash("redis-A") gives its position.
  2. Place each key on the ring by hashing it: hash("x7Kp").
  3. A key belongs to the first server clockwise from it.

Now add server E. It lands at one point on the ring and takes over only the keys between itself and the server before it, keys that used to belong to the next server along. Every other key stays where it was. Remove a server, and only its keys move, to the next server clockwise.

Virtual nodes

With one point per server, the arcs between servers are random lengths, so one server may own three times the keys of another. And when a server leaves, all its keys go to a single neighbour, which suddenly has double the load.

The fix is virtual nodes: put each server on the ring many times (redis-A#0, redis-A#1, … redis-A#199). With hundreds of points each, every server owns many small arcs scattered round the ring, so loads even out, and a departing server's keys spread across all the others. A bigger machine can simply get more virtual nodes.

LabAdd a server, count what moves

Keys moved

–

Make a change to see

Busiest server

1.02×

the average load

A2,432
B2,508
C2,538
D2,522

Keys per server, out of 10,000. Red dots moved in the last change.

Try this

With 10,000 keys, the lab shows the pattern clearly. mod N moves about 80% of keys going from four servers to five. The ring with one point per server moves far fewer, but its busiest server can hold nearly twice the average. With 200 virtual nodes, adding a fifth server moves about 21% of keys (a fair share is 20%), and the busiest server is within a few percent of the average.

A small implementation

A ring is a sorted array of positions. Finding a key's server is a binary search for the first position at or after the key's hash, wrapping to the start.

ts
import { createHash } from "node:crypto";

// 32-bit position on the ring, from the first 4 bytes of an MD5 hash
const position = (s: string) => createHash("md5").update(s).digest().readUInt32BE(0);

export class HashRing {
  private points: { at: number; node: string }[] = [];

  constructor(private vnodes = 200) {}

  add(node: string) {
    for (let v = 0; v < this.vnodes; v++) this.points.push({ at: position(`${node}#${v}`), node });
    this.points.sort((a, b) => a.at - b.at);
  }

  remove(node: string) {
    this.points = this.points.filter((p) => p.node !== node);
  }

  /** The node that owns `key`: the first point clockwise from the key's position. */
  get(key: string): string {
    if (this.points.length === 0) throw new Error("empty ring");
    const h = position(key);
    let lo = 0;
    let hi = this.points.length;
    while (lo < hi) {
      const mid = (lo + hi) >> 1;
      if (this.points[mid].at < h) lo = mid + 1;
      else hi = mid;
    }
    return this.points[lo % this.points.length].node;
  }
}
ts
const ring = new HashRing();
["redis-A", "redis-B", "redis-C", "redis-D"].forEach((n) => ring.add(n));
ring.get("x7Kp"); // → "redis-B", and the same answer on every app server

Every client that builds the ring from the same list of nodes gets the same answers, with no coordination. The hash doesn't need to be secure, only fast and evenly spread; MD5 is used here because it's built into Node.

Replicas: walk further round

To keep N copies of each key, walk clockwise from the key and take the first N distinct physical servers, skipping virtual nodes of a server you already have. If one fails, the next server along already holds a copy. This is how Dynamo-style databases place replicas (08.04 Replication and quorums).

Where it's used, and the alternatives

Consistent hashing shows up wherever many clients must agree, without asking anyone, which server owns a key: cache clusters (06.06 Designing a distributed cache), distributed databases, and load balancers that send the same user or object to the same backend (03.06 Load balancers).

Two related techniques are worth recognising. Rendezvous hashing scores every server for a key (hash of key plus server) and picks the highest: simple, with no ring to store, at the cost of checking every server per lookup. Bounded loads caps each server at a little above the average; a key whose server is full moves on clockwise to the next one with room, which keeps hot spots from overloading one server.

Seen in production

Amazon

2007

Dynamo

Dynamo, which ran Amazon's shopping cart, placed keys on a consistent hashing ring and gave each node many positions ("tokens"), so a failed node's load spread evenly over the rest and bigger machines could take more tokens. Its first scheme, random tokens, made a joining node scan other nodes' data to take over its ranges; in busy shopping seasons that bootstrapping took almost a day. Amazon moved to splitting the ring into a fixed number of equal partitions and assigning whole partitions to nodes, which balanced load better and made moves simpler.

What we foundBasic consistent hashing gave uneven load and ignored machine size; virtual nodes, then fixed equal-sized partitions, fixed both.

Dynamo: Amazon's Highly Available Key-value Store (SOSP 2007)allthingsdistributed.com ↗

Google and Vimeo

2017

Consistent hashing with bounded loads

Google researchers bounded each server's load at a factor above the average: a key whose server is full moves clockwise to the next server with spare capacity, and each change moves only a bounded number of other keys. It was first used in Google Cloud Pub/Sub. An engineer at Vimeo found the paper and implemented it in HAProxy, which reduced Vimeo's cache bandwidth by a factor of almost eight.

What we foundCapping each server's load and overflowing to the next server on the ring cut Vimeo's cache bandwidth by almost 8×.

Consistent Hashing with Bounded Loadsresearch.google ↗
Predict first

A cache cluster uses consistent hashing with 200 virtual nodes per server across 10 servers. One server dies. Roughly what happens?

Commit to an answer before you read on.

Recap

Interview questions

MidfreqMust-knowConcept

What is consistent hashing, and why is it better than hash mod N?

Report a mistakeEdit this page