Consistent Hashing

1. The problem

Imagine you are scaling an in-memory cache cluster (e.g. Memcached or Redis) across $N=4$ servers to store 10 million user sessions.

The traditional approach to partitioning keys across servers is Modulo Hashing: $$\text{Server Index}=\text{hash}(\text{key})(\mathrm{mod}N)$$

Server 0: hash(key) % 4 == 0
Server 1: hash(key) % 4 == 1
Server 2: hash(key) % 4 == 2
Server 3: hash(key) % 4 == 3

This works well—until the cluster needs to scale.

The Modulo Rehash Disaster

Suppose traffic spikes and you add a 5th server ($N=5$):

• A key with $\text{hash}=14$:
  • Previously: $14(\mathrm{mod}4)=\text{Server }2$
  • Now: $14(\mathrm{mod}5)=\text{Server }4$
• A key with $\text{hash}=27$:
  • Previously: $27(\mathrm{mod}4)=\text{Server }3$
  • Now: $27(\mathrm{mod}5)=\text{Server }2$

(By the Chinese Remainder Theorem, since $\gcd (N,N+1)=1$ over a uniform hash space $R\gg N(N+1)$, only $N$ out of $N(N+1)$ residue pairs satisfy $(h\mathrm{mod}N)=(h\mathrm{mod}(N+1))$, leaving $\frac{1}{N+1}$ unchanged and remapping $\frac{N}{N+1}$.)

When scaling from 9 to 10 nodes, $90\%$ of all cached keys remap to different servers. The entire cache is effectively wiped clean simultaneously. Incoming traffic bypasses the cache and floods the primary database (Cache Stampede / Thundering Herd), driving the database into complete outage.

We need a partitioning algorithm where adding or removing a node migrates only the minimum necessary keys, leaving all other mappings undisturbed.

2. Mental model: The Circular Hash Ring

Instead of mapping keys to a discrete array of $N$ slots, Consistent Hashing (introduced by Karger et al. at MIT in 1997) maps both servers and data keys onto a continuous circular coordinate space: the Hash Ring.

                               Token 0 / 2^32
                                     ▲
                                     │
                 [ Node D ]          │          [ Node A ]
                   (315°)            │            (45°)
                                     │
           270° ─────────────────────┼───────────────────── 90°
                                     │
                                     │
                 [ Node C ]          │          [ Node B ]
                   (225°)            │            (135°)
                                     │
                                     ▼
                                 Token 2^31

• Imagine an analog clock face ranging from $0$ to $2^{32}-1$ (for a standard 32-bit hash function like MurmurHash3).
• Place Servers on the Ring: Hash each server’s identifier (e.g. IP address or hostname) to a position on the ring.
• Place Keys on the Ring: Hash each data key to a position on the same ring.
• Routing Rule: To find which server owns a key, start at the key’s position on the ring and walk clockwise until you encounter the first server.

3. Internal workings

SYSTEM ARCHITECTURE BLUEPRINTDATA STRUCTURES

⤢Expand

ARCHITECTURAL FLOW · STEP 1 OF 5

1. Uniform Ring Token Placement

Nodes A, B, and C are hashed onto the 32-bit integer ring [0, 2^32 - 1] using MD5/Murmur3. Each server owns a continuous token arc.

Architectural Invariant: Virtual Nodes: Production rings (Dynamo, Cassandra) assign 128-256 virtual tokens per physical node to prevent hotspot skew.

ARCHITECTURAL FLOW · STEP 2 OF 5

2. Key 1 Clockwise Routing

hash('item:alpha') evaluates to token ~20°. The coordinator walks clockwise along the ring to locate the first node with token >= hash (Node A).

Architectural Invariant: Binary Search: In memory, servers are stored in a sorted array; binary search locates the owning node in O(log N) time.

ARCHITECTURAL FLOW · STEP 3 OF 5

3. Key 2 Clockwise Routing

hash('item:beta') evaluates to token ~90°. Following the ring clockwise leads directly to Node B.

Architectural Invariant: Independent Routing: Every client with the ring topology can route requests directly without querying a central master.

ARCHITECTURAL FLOW · STEP 4 OF 5

4. Cluster Expansion: Node D Joins Ring

A new server (Node D) is introduced at token position ~315°. It takes responsibility for the segment between Node C and Node D.

Architectural Invariant: Scale-Out Property: Unlike modulo hashing where changing N remaps nearly 100% of keys, consistent hashing limits remapping strictly.

ARCHITECTURAL FLOW · STEP 5 OF 5

5. Minimal Key Migration Math

Only Key 3 (located between Node C and Node D) migrates from Node A to Node D. All keys on Node B and Node C remain completely untouched!

Architectural Invariant: Remapping Invariant: Adding 1 node to an N-node cluster requires moving exactly K / (N + 1) keys out of K total keys.

1. Key Assignment

To locate the owning server for a key $K$:

1. Calculate the token: $T_K=\text{hash}(K)$.
2. Search clockwise for the first server whose token $T_S\ge T_K$.
3. If $T_K$ is greater than all server tokens, wrap around to the first server on the ring (at the top).

2. Node Addition (Graceful Scaling)

Suppose we have servers $A,B,C$, and we insert a new server $D$ between $C$ and $A$:

• Only the keys that fall between $C$ and $D$ on the ring are re-assigned from $A$ to $D$.
• Keys owned by server $B$ and server $C$ are completely unaffected.
• On average, exactly $\frac{1}{N+1}$ of keys migrate!

3. Node Removal / Failure

If server $B$ crashes:

• Only keys previously owned by $B$ migrate to its immediate clockwise successor ($C$).
• All keys on servers $A$ and $D$ remain completely untouched.

4. The Two Fundamental Flaws of Naive Consistent Hashing

A simple ring with one token per server exhibits two fatal failure modes in production:

Flaw 1: Non-Uniform Key Distribution (Hotspots)

When servers are assigned random positions on the ring, partition sizes vary wildly. By chance, two servers might be positioned close together, leaving another server responsible for $70\%$ of the ring’s arc.

Flaw 2: Cascading Domino Failover

If Node $B$ crashes, 100% of Node $B$‘s traffic lands on its single immediate clockwise neighbor (Node $C$). Node $C$ now receives its own traffic plus all of $B$‘s load ($2\times$ traffic). Node $C$ promptly crashes under overload, dumping $3\times$ traffic onto the next node, triggering a cascading total cluster blackout.

4. The Solution: Virtual Nodes (Vnodes)

To achieve uniformity and fault isolation, real-world systems (Cassandra, DynamoDB, Discord) do not place physical machines directly on the ring. Instead, they use Virtual Nodes (Vnodes).

Physical Server A ──► Hashes 128 times ──► [ A#1, A#2, A#3, ..., A#128 ]
Physical Server B ──► Hashes 128 times ──► [ B#1, B#2, B#3, ..., B#128 ]
Physical Server C ──► Hashes 128 times ──► [ C#1, C#2, C#3, ..., C#128 ]

Each physical machine is assigned multiple tokens (typically 128 to 256) distributed randomly across the ring:

Ring Order: [ A#12 ] ──► [ B#4 ] ──► [ C#99 ] ──► [ A#45 ] ──► [ B#77 ] ──► [ C#3 ] ...

Why Vnodes Transform Cluster Reliability:

1. Statistical Balance: By the Law of Large Numbers, distributing hundreds of points per server ensures every physical node owns an almost identical total fraction of the ring ($\approx \pm 3-5\%$ variance).
2. Distributed Failover: If physical server $B$ crashes, its 128 virtual tokens are scattered across the ring. Their immediate clockwise successors are split evenly between Server $A$ and Server $C$. Instead of one neighbor absorbing $100\%$ of the failure load, the load is distributed across all surviving machines in the cluster.
3. Heterogeneous Hardware Weighting: A powerful server with 128 GB RAM can be allocated 256 vnodes, while an older 64 GB server is assigned 128 vnodes, proportionally balancing load according to hardware capacity.

5. Mathematical foundations

1. Key Migration Comparison

For a cluster of $N=9$ nodes storing $10,000,000$ keys:

• Modulo hashing forces $9,000,000$ keys to move.
• Consistent hashing moves only $1,000,000$ keys ($9\times$ less data movement!).

2. Load Variance vs. Number of Vnodes

Let $N$ physical nodes each place $V$ independent, uniformly distributed tokens onto the unit ring ($M=N\cdot V$ total tokens). Each node’s total owned ring fraction $L_i$ is the sum of $V$ out of $M$ symmetric Dirichlet spacings, so $L_i\sim \text{Beta}(V,(N-1)V)$ with mean $\mu =\frac{1}{N}$. Its coefficient of variation (standard deviation relative to mean node load) is:

3. Lookup Algorithmic Complexity

In a cluster with $N$ physical nodes and $V$ virtual nodes per server:

• Total tokens on the ring: $M=N\times V$.
• Tokens are stored in a balanced binary search tree (e.g. std::map in C++, java.util.TreeMap in Java, or a sorted array).
• Lookup Time: $O(\log M)=O(\log (N\cdot V))$ using binary search.
• Memory Footprint: Storing 100 nodes $\times 256$ vnodes = 25,600 integer tokens $\times 16\text{ bytes}\approx 400\text{ KB}$ (trivially fits in L2/L3 CPU cache).

6. Implementation: Minimal Consistent Hash Ring in TypeScript

/**
 * Portable Consistent Hash Ring using 32-bit MurmurHash3 (x86_32)
 * Zero Node-specific dependencies; runs identically in browser and server runtimes.
 */
export class ConsistentHashRing {
  private readonly vnodesPerNode: number;
  private readonly ring: Map<number, string> = new Map();
  private sortedTokens: number[] = [];
  private readonly encoder = new TextEncoder();

  constructor(vnodesPerNode = 128) {
    this.vnodesPerNode = vnodesPerNode;
  }

  private murmurHash3_32(value: string, seed = 0x9747b28c): number {
    const data = this.encoder.encode(value);
    const len = data.length;
    const c1 = 0xcc9e2d51;
    const c2 = 0x1b873593;
    let h1 = seed >>> 0;
    const roundedEnd = len & ~0x3;

    for (let i = 0; i < roundedEnd; i += 4) {
      let k1 =
        (data[i] & 0xff) |
        ((data[i + 1] & 0xff) << 8) |
        ((data[i + 2] & 0xff) << 16) |
        ((data[i + 3] & 0xff) << 24);
      k1 = Math.imul(k1, c1);
      k1 = (k1 << 15) | (k1 >>> 17);
      k1 = Math.imul(k1, c2);

      h1 ^= k1;
      h1 = (h1 << 13) | (h1 >>> 19);
      h1 = (Math.imul(h1, 5) + 0xe6546b64) >>> 0;
    }

    let k1 = 0;
    switch (len & 3) {
      case 3:
        k1 ^= (data[roundedEnd + 2] & 0xff) << 16;
      // falls through
      case 2:
        k1 ^= (data[roundedEnd + 1] & 0xff) << 8;
      // falls through
      case 1:
        k1 ^= data[roundedEnd] & 0xff;
        k1 = Math.imul(k1, c1);
        k1 = (k1 << 15) | (k1 >>> 17);
        k1 = Math.imul(k1, c2);
        h1 ^= k1;
    }

    h1 ^= len;
    h1 ^= h1 >>> 16;
    h1 = Math.imul(h1, 0x85ebca6b);
    h1 ^= h1 >>> 13;
    h1 = Math.imul(h1, 0xc2b2ae35);
    h1 ^= h1 >>> 16;
    return h1 >>> 0; // Unsigned 32-bit integer in [0, 2^32 - 1]
  }

  public addNode(node: string): void {
    for (let i = 0; i < this.vnodesPerNode; i++) {
      let attempt = 0;
      let token = this.murmurHash3_32(`${node}#vnode-${i}`);
      while (this.ring.has(token) && this.ring.get(token) !== node) {
        attempt++;
        token = this.murmurHash3_32(`${node}#vnode-${i}#${attempt}`);
      }
      if (!this.ring.has(token)) {
        this.sortedTokens.push(token);
      }
      this.ring.set(token, node);
    }
    this.sortedTokens.sort((a, b) => a - b);
  }

  public removeNode(node: string): void {
    for (const [token, owner] of this.ring.entries()) {
      if (owner === node) {
        this.ring.delete(token);
      }
    }
    this.sortedTokens = this.sortedTokens.filter((t) => this.ring.has(t));
  }

  public getNode(key: string): string | null {
    if (this.sortedTokens.length === 0) return null;
    const token = this.murmurHash3_32(key);

    // Binary search for first token >= key token (clockwise walk)
    let low = 0;
    let high = this.sortedTokens.length - 1;
    let index = 0;

    while (low <= high) {
      const mid = (low + high) >>> 1;
      if (this.sortedTokens[mid] >= token) {
        index = mid;
        high = mid - 1;
      } else {
        low = mid + 1;
      }
    }

    // Wrap around to ring origin if token > all server tokens
    if (low >= this.sortedTokens.length) {
      index = 0;
    }

    return this.ring.get(this.sortedTokens[index]) ?? null;
  }

  public getTokenCount(): number {
    return this.sortedTokens.length;
  }
}

function runDemo(): void {
  console.log("=== Consistent Hashing: Virtual Nodes & Minimal Key Migration ===");
  const ring = new ConsistentHashRing(128);
  const nodes = ["cache-node-A", "cache-node-B", "cache-node-C"];
  for (const n of nodes) ring.addNode(n);

  const TOTAL_KEYS = 10000;
  const beforeMapping = new Map<string, string>();
  const counts: Record<string, number> = { "cache-node-A": 0, "cache-node-B": 0, "cache-node-C": 0 };

  for (let i = 0; i < TOTAL_KEYS; i++) {
    const key = `session:user-${i}`;
    const owner = ring.getNode(key)!;
    beforeMapping.set(key, owner);
    counts[owner]++;
  }

  console.log(`[1] Initialized 3 nodes (${ring.getTokenCount()} total vnode tokens):`);
  for (const n of nodes) {
    console.log(`    ${n}: ${counts[n]} keys (${((counts[n] / TOTAL_KEYS) * 100).toFixed(1)}%)`);
  }

  ring.addNode("cache-node-D");
  let migrated = 0;
  for (let i = 0; i < TOTAL_KEYS; i++) {
    const key = `session:user-${i}`;
    if (ring.getNode(key) !== beforeMapping.get(key)) {
      migrated++;
    }
  }
  console.log(
    `[2] Added cache-node-D (${ring.getTokenCount()} tokens): migrated ${migrated}/${TOTAL_KEYS} keys (${((migrated / TOTAL_KEYS) * 100).toFixed(1)}%, theoretical 1/(N+1) = 25.0%)`
  );
}

runDemo();

7. Real-world implementations

1. Apache Cassandra & Amazon DynamoDB

• Both databases use a distributed token ring for partitioning rows by primary key partition token.
• Apache Cassandra’s Murmur3Partitioner (org.apache.cassandra.dht.Murmur3Partitioner) hashes partition keys into a 64-bit signed integer token space ($-2^{63}+1$ to $2^{63}-1$, with $-2^{63}$ reserved as the minimum sentinel).
• Replication: To replicate a key with replication factor $RF=3$, the coordinator walks clockwise along the ring and places copies on the first 3 distinct physical nodes (and distinct racks when NetworkTopologyStrategy is enabled).

2. Discord Gateway Architecture

• Discord routes millions of active WebSocket guild connections using consistent hashing (via Elixir’s hash-ring routing).
• When gateway instances restart or scale out during peak hours, only $\approx \frac{1}{N+1}$ of stateful guild processes migrate, preventing cluster-wide reconnect storms.

8. Failure modes and limitations

1. The Celebrity / Super-Hot Key: Consistent hashing guarantees uniform distribution across distinct random keys. However, if a single key (e.g. a viral livestream room or breaking news counter) receives 1 million requests/second, that entire load hits the single replica set owning that key. Mitigation: Apply key salting / write sharding ($key\#0,key\#1,\dots ,key\#(k-1)$) for writes and scatter-gather or tiered local edge caching for reads, or use Consistent Hashing with Bounded Loads (Mirrokni et al., 2017) to cap any server at $(1+\epsilon )$ times average load.
2. Replication Overlap & Repair Fan-Out with High Vnode Counts: When selecting $RF=3$ successive nodes clockwise on the ring, virtual tokens from the same physical node might appear consecutively, so the ring walker must skip physical nodes (and racks) already in the replica set. Furthermore, with $V=256$ random vnodes, every physical node shares token ranges with almost every other node in the cluster, increasing simultaneous multi-node failure correlation and slowing anti-entropy Merkle-tree repairs—which motivated Cassandra 4.0+ to reduce default num_tokens to 16 with deterministic token allocation.
3. Ring State Inconsistency: If Node A and Node B hold temporarily divergent ring membership views during gossip convergence or network partitions, they route the same key to different servers, causing cache incoherence or quorum read/write retries (handled via sloppy quorums and hinted handoff in Dynamo-style stores).

9. Operational perspective

What to Monitor in Production

• Token Skew Ratio: Measure maximum-to-minimum data ownership across physical storage nodes: $$\text{Skew Ratio}=\frac{{\max }_i({\text{Owned Bytes}}_i)}{{\min }_i({\text{Owned Bytes}}_i)}$$ A ratio $>1.25$ on equal-sized hardware indicates either insufficient vnode count without deterministic token allocation or low-entropy partition keys.
• Node Join Streaming Bandwidth: When adding or decommissioning a node, cap background streaming throughput (e.g. stream_throughput_outbound_megabits_per_sec in Cassandra) so rebalancing traffic does not starve foreground client p99 latency.

10. Knowledge check

Complete the interactive assessment quiz at the end of this lesson to test your understanding of modulo remapping math, vnode variance bounds, and $O(\log (N\cdot V))$ ring lookups.

11. Exercises

Exercise 1: Multi-Node Migration Math

• Scenario: A distributed cache cluster currently runs $N=15$ equal-capacity servers storing $K=30,000,000$ uniformly hashed objects. During a traffic surge, 2 new servers are added simultaneously ($N^{\prime }=17$).
• Questions:
  1. Under traditional modulo hashing ($\text{hash}(k)\mathrm{mod}N$), assuming a uniform 64-bit hash space, what fraction and exact expected count of objects remap to a different server? (Hint: check $\gcd (15,17)$ via the Chinese Remainder Theorem.)
  2. Under consistent hashing with 128 balanced vnodes per server, how many objects migrate to the two new servers, and how many objects move between the original 15 servers?

Exercise 2: Vnode Load Variance & Domino Failover Sizing

• Scenario: You operate an $N=10$ node caching cluster where each server is provisioned to run at $70\%$ memory and CPU utilization under balanced traffic.
• Questions:
  1. Using the Dirichlet/Beta coefficient of variation formula $CV=\sqrt{\frac{N-1}{N\cdot V+1}}$, calculate the relative load standard deviation ($CV$) for $V=1$ (naive ring) vs. $V=144$ vnodes per node. How many vnodes $V$ are required so that $CV\le 5\%$?
  2. If 1 physical server crashes, compare the peak utilization of the hottest surviving neighbor under $V=1$ versus $V=144$ vnodes, and explain why $V=1$ triggers a cascading cluster outage when nodes crash at $100\%$ utilization.

12. Summary, Design Alternatives & Decision Guide

13. Authoritative References & Version Context

• Karger, D., Lehman, E., Leighton, T., Panigrahy, R., Levine, M., & Lewin, D. (1997). Consistent Hashing and Random Trees: Distributed Caching Protocols for Relieving Hot Spots on the World Wide Web. Proceedings of the 29th Annual ACM Symposium on Theory of Computing (STOC ‘97), pp. 654–663. https://dl.acm.org/doi/10.1145/258533.258660
• DeCandia, G., Hastorun, D., Jampani, M., Kakulapati, G., Lakshman, A., Pilchin, A., Sivasubramanian, S., Vosshall, P., & Vogels, W. (2007). Dynamo: Amazon’s Highly Available Key-value Store. ACM SOSP ‘07, pp. 205–220. https://www.allthingsdistributed.com/files/amazon-dynamo-sosp2007.pdf
• Lamping, J., & Veach, E. (2014). A Fast, Minimal Memory, Consistent Hash Algorithm (Jump Consistent Hash). Google Technical Report. https://arxiv.org/abs/1406.2294
• Mirrokni, V., Thorup, M., & Zadimoghaddam, M. (2018). Consistent Hashing with Bounded Loads. ACM-SIAM SODA ‘18. https://arxiv.org/abs/1608.01350
• Apache Cassandra Official Documentation (v4.1 / v5.0 baseline). Dynamo Ring Architecture, Virtual Nodes (num_tokens: 16 default in 4.0+), and Murmur3Partitioner. https://cassandra.apache.org/doc/latest/cassandra/architecture/dynamo.html