πŸ“š Study Notes / Home / HLD / Session 2
Session 02 Β· Load Balancing & Consistent Hashing

How big systems spread the work β€” fairly and without chaos

In Session 1 we learned that real systems run on many servers, not one. But that raises a question: when a request arrives, which server should handle it? And when you add or remove servers, how do you avoid breaking everything? This session answers both. We assume you've studied none of this before β€” every topic starts with a tiny "explain like I'm 5" story, then we go deeper with real examples and code. Take it slow.

⏱ 45 min readπŸ“– 9 topics

1 Load balancers & their types


Explain like I'm 5

Imagine a busy ice-cream shop with one queue but five servers behind the counter. If everyone lined up for the same person, that one server would be exhausted and the other four would be bored. So a friendly host at the door points each new customer to whichever server is free. That host is a load balancer β€” it stands in front of your servers and sends each request to one of them so the work is shared.

Recall from Session 1 that we scale a system horizontally by running many identical copies of our server. A load balancer (LB) is the piece that sits in front of those copies and decides which one gets each incoming request. It gives us three superpowers at once:

  • Spreading load β€” no single server gets overwhelmed. It distributes the load (requests and data) equally across the app/DB servers behind it.
  • High availability β€” if one server dies, the LB stops sending traffic to it, and users never notice.
  • A single, unified front door β€” clients talk to one address (the LB) instead of needing to know about every server behind it. This gives the end user a unified view of the entire backend: the user doesn't care which exact server handles their request.
The purpose of a load balancer

Two jobs, stated the way the class did: (1) provide a unified view of the entire backend system to the end user β€” because the end user doesn't care which exact server handles their request; and (2) distribute the load (requests / data) equally across the app / DB servers.

πŸ‘€
Clients
Browsers / apps
β†’
βš–οΈ
Load balancer
One front door
β†’
πŸ–₯️
Server pool
Many identical backends

L4 vs L7 β€” two places the LB can do its job

Load balancers are usually described by which OSI layer they operate on. The OSI model is just a 7-layer map of how network data travels; you only need two layers here. Layer 4 (L4) is the transport layer β€” it deals with TCP/UDP connections, IP addresses, and ports. Layer 7 (L7) is the application layer β€” it understands HTTP, so it can read URLs, headers, and cookies.

AspectL4 (transport)L7 (application)
Works withTCP/UDP packets, IP + portFull HTTP requests (URL, headers, cookies)
Can route onSource/destination IP, port, protocolURL path, hostname, headers, cookies, request body
Sees the content?No β€” it just forwards the connectionYes β€” it reads and understands the message
SpeedVery fast (less work per packet)Slightly slower (must parse HTTP)
SmartsLow β€” can't tell a login from an image requestHigh β€” can send /api/* one way, /images/* another
TLSPasses encrypted traffic straight throughOften terminates TLS (decrypts) so it can read the request
When the difference actually matters

Say all your image requests live at /images/* and your API lives at /api/*, and you want images served by a separate, cheaper pool of servers.

  • An L4 LB only sees "a TCP connection to port 443" β€” it has no idea the URL says /images, so it cannot split this traffic.
  • An L7 LB reads the HTTP request, sees the path is /images/cat.jpg, and routes it to the image pool. This is called content-based routing.

Hardware vs software load balancers

Historically, load balancers were expensive dedicated hardware appliances (think specialized boxes from vendors like F5 or Citrix) that you bought and racked in your data center. Today most teams use software load balancers β€” programs running on ordinary servers β€” or a cloud-managed LB you rent by the hour. On AWS the cloud-managed option is the Elastic Load Balancer (ELB).

TypeExamplesNotes
Hardware applianceF5 BIG-IP, Citrix ADCVery fast, very expensive, fixed capacity, you maintain it.
Software (self-run)Nginx, HAProxy, EnvoyRun on normal Linux boxes, cheap, flexible, you configure & scale them.
Cloud-managedAWS ELB/ALB/NLB, GCP Cloud Load Balancing, Azure Load BalancerYou click a few buttons; the cloud runs and scales it for you. ALB = L7, NLB = L4.
Quick name guide

Nginx and HAProxy are the two most common open-source software LBs β€” Nginx is also a web server and reverse proxy, HAProxy specializes in load balancing. On AWS, the ALB (Application Load Balancer) is L7 and the NLB (Network Load Balancer) is L4. The words "appliance," "reverse proxy," and "load balancer" overlap a lot in practice β€” don't let the jargon scare you.

Health checks β€” how the LB knows who's alive

A load balancer must never send a request to a dead server. So it constantly runs health checks: it periodically pings each backend (often a tiny HTTP request to a special URL like /healthz) and waits for an "I'm OK" reply.

  • If a server answers correctly β†’ it stays in the rotation ("healthy").
  • If it fails a few checks in a row β†’ the LB marks it "unhealthy" and stops routing traffic to it. Users never hit the broken box.
  • When it starts answering again β†’ the LB quietly adds it back.
A tiny health-check endpoint
# The backend exposes a cheap "am I alive?" route
GET /healthz  # LB calls this every few seconds

# Healthy response:
200 OK   {"status": "ok"}

# If the server is overloaded or broken:
503 Service Unavailable   # β†’ LB pulls it out of rotation
Where to learn the practical setup

The class linked hands-on configuration guides: a setup doc, the AviNetworks "What is load balancing?" tutorial, and AWS's "Getting started with Elastic Load Balancing." See References for the links.

Recap A load balancer sits in front of a pool of identical servers and spreads requests across them, giving us a unified front door, shared load, and high availability. L4 balancers route on IP/port (fast, blind to content); L7 balancers read HTTP and can route on URL, headers, or cookies (smarter, slightly slower). They come as hardware appliances, software (Nginx, HAProxy), or cloud-managed services (AWS ELB), and they use health checks to route around dead servers automatically.

2 A detour: IP addresses


Explain like I'm 5

Every house needs an address so the mail carrier can find it. On the internet, every machine needs an IP address so packets can find it. An old-style address (IPv4) is like a short street number that's starting to run out β€” there just aren't enough numbers for all the houses being built. So we invented a longer address (IPv6) with so many numbers we'll never run out.

Before we decide which server a request goes to, it helps to remember how machines are even addressed. An IPv4 address looks like 142.250.192.4 β€” four numbers separated by dots, each in the range 0–255.

# IPv4 layout: four octets, each 0–255
IPv4:  0-255 . 0-255 . 0-255 . 0-255

# Each octet is 8 bits, so total possibilities:
2^8 * 2^8 * 2^8 * 2^8  =  2^32  ~  4 billion addresses

The problem: there are roughly 100 billion to 1 trillion devices on the internet, but IPv4 offers only ~4 billion addresses. That's nowhere near enough β€” which is why we also have static vs dynamic IPs (some machines keep a fixed address; most devices borrow one temporarily) and, ultimately, a bigger address space.

VersionSizeTotal possibilities
IPv432 bits2Β³Β² β‰ˆ 4 billion
IPv6128 bits2¹²⁸ β‰ˆ 256 billion billion billion billion

IPv6 uses 128 bits, giving 2¹²⁸ β‰ˆ 256 billion billion billion billion possibilities β€” effectively unlimited. Static IPs stay the same over time (useful for servers you must always reach at the same address); dynamic IPs are assigned on the fly and can change (typical for home/phone devices).

Recap IPv4 is four 0–255 octets (32 bits, 2Β³Β² β‰ˆ 4 billion addresses) β€” too few for the 100 billion–1 trillion devices online, so we have static vs dynamic IPs and IPv6 (128 bits, 2¹²⁸ β‰ˆ practically infinite). This addressing is the backdrop for the real question: which server should a given request go to?

3 Storing data: sharding & partitioning


Explain like I'm 5

You have too many toys for one toy box, so you buy more boxes. But you can't just throw toys in randomly β€” then you'd never find anything! You need a rule: "all the cars go in box 1, all the dolls in box 2." That way, when you want a toy, you know exactly which box to open. Splitting your stuff across boxes with a findable rule is sharding.

The big question

Which server should a particular request be sent to? To answer that, we first have to understand how the data is spread across servers β€” otherwise we'd send requests to servers that don't even have the right data (Sanjana's request lands on a server holding Ashok's data).

Why not just one server?

Q: Can we store all the data on 1 server? No. We added more machines in the first place because we were running out of resources β€” disk space, CPU, memory. One server can't hold it all.

Q: Should we split the data randomly? No β€” then how would we retrieve it? Data storage has to follow some logic that can be repeated. At any moment we must be able to figure out which data lives on which server.

Two kinds of partitioning

"Splitting data" has two flavors, and each can happen within a database or across servers.

KindWhat it splitsWhy you'd do it
Vertical partitioning (within a DB)Splits a table by columnsNormalization β€” break a wide table into related tables.
Horizontal partitioning (within a DB)Splits a table by rowsImprove indexing performance; support multi-tenancy (prevent data cross-communication between clients).
Vertical partitioning across serversDifferent features/tables on different serversMigrate to microservices / separation of concerns.
Horizontal partitioning across serversDifferent rows on different serversAll the data cannot fit on a single server.
Sharding, defined

Sharding is simply "horizontal partitioning across servers." You split your rows across many machines because they won't all fit on one.

The sharding key

Sharding is always based on some value β€” the sharding key. The key is what decides which shard (server) a piece of data lives on. (How to choose a good sharding key is a topic for a later lecture.)

Recap We can't store everything on one server (we ran out of resources), and we can't split data randomly (we'd never find it again) β€” storage must follow repeatable logic. Vertical partitioning splits by columns (normalization / microservices); horizontal partitioning splits by rows (indexing, multi-tenancy, capacity). Sharding = horizontal partitioning across servers, always driven by a sharding key.

4 Routing = sharding, and what makes a good algorithm


Explain like I'm 5

Pretend the host at the door always sends you to the same scooper every time you visit. Then, without anyone planning it, your ice-cream preferences end up known only by that scooper β€” because no one else ever served you! Deciding "which person serves which customer" automatically decides "who remembers which customer." Routing and storing are two sides of the same coin.

Sharding ⟺ Routing

It must not be the case that sharding follows logic A while routing follows logic B. If they differ, requests land on servers that don't hold the needed data! So the algorithm must be the same for both. In fact, sharding happens via routing: routing is the thing that happens, and sharding is just a side effect of it.

Here's the insight. If we route based on user_id, then sharding automatically happens based on user_id too. Why? Because if Sanjana's requests are always routed to server A, then only server A ever sees (and therefore stores) Sanjana's data β€” servers B and C never received her requests in the first place.

The one decision that decides two things

We only decide (which request β†’ which server). The moment we decide that, we've also automatically decided (which data β†’ which server). So all we really need is a good routing algorithm.

Where routing lives

The routing algorithm runs inside the load balancers. It's how the LB decides which request goes to which server.

What makes a routing algorithm "good"?

A good routing algorithm should have these characteristics:

  • Fast β€” computationally easy to decide which request goes to which server.
  • Equal distribution β€” divides the load (data & requests) equally across all servers.
  • Freely add / remove servers β€” servers are unreliable and crash (count goes down), and we scale out by adding more (count goes up). The algorithm must cope with both.
  • Minimal data movement β€” when the number of servers changes, the amount of data that has to move should be minimal.
  • Deterministic without exchanging information β€” we have multiple load balancers, and they must all stay "in sync," forwarding the same type of request (say Sanjana's) to the same server. Otherwise Sanjana's data ends up scattered everywhere. Crucially, they must achieve this without constantly communicating β€” per-request chatter between LBs would be far too much overhead.
Recap Sharding and routing must use the same logic, because sharding is a side effect of routing β€” deciding which request goes to which server automatically decides which data lives where. The routing algorithm runs inside the LBs, and a good one is fast, gives equal distribution, lets you freely add/remove servers with minimal data movement, and is deterministic without LBs syncing with each other. The next topics test candidate algorithms against this checklist.

5 Load balancing algorithms


Explain like I'm 5

Back at the ice-cream shop, how does the host decide which server to send you to? They could just go around the circle: you, then the next person, then the next (taking turns). Or they could look and send you to whoever has the shortest line. Or send more customers to the fastest scooper. These different "rules for choosing" are called load balancing algorithms β€” and picking the right rule keeps everyone happy.

Once the LB has a list of healthy servers, it needs a rule to pick one for each request. Here are the classic algorithms, from simplest to smartest.

Round Robin

Round Robin just cycles through the servers in order: request 1 β†’ server A, request 2 β†’ server B, request 3 β†’ server C, request 4 β†’ back to A, and so on. Simple, fair, and needs no extra information. Its weakness: it assumes every server and every request is equal β€” which isn't always true. (When applied to data routing via a modulo trick, it has a bigger flaw we'll dissect in Topic 6.)

Weighted Round Robin

Weighted Round Robin is Round Robin with a "strength" number per server. A beefier server gets a higher weight and therefore more requests. If A has weight 3 and B has weight 1, A receives three requests for every one B receives. Great when your servers aren't all the same size.

# Python: Round Robin and Weighted Round Robin balancers
from itertools import cycle


class RoundRobinBalancer:
    """Cycle through servers in order: A, B, C, A, B, C, ..."""
    def __init__(self, servers):
        self.servers = list(servers)
        self._i = 0

    def next_server(self):
        server = self.servers[self._i]
        self._i = (self._i + 1) % len(self.servers)
        return server


class WeightedRoundRobinBalancer:
    """A=3, B=1 -> A, A, A, B, A, A, A, B, ... (expanded by weight)."""
    def __init__(self, weighted):   # weighted = [(name, weight), ...]
        sequence = []
        for name, weight in weighted:
            sequence.extend([name] * weight)
        self._cycle = cycle(sequence)

    def next_server(self):
        return next(self._cycle)


if __name__ == "__main__":
    rr = RoundRobinBalancer(["A", "B", "C"])
    print("RR:  ", [rr.next_server() for _ in range(6)])
    # RR:   ['A', 'B', 'C', 'A', 'B', 'C']

    wrr = WeightedRoundRobinBalancer([("A", 3), ("B", 1)])
    print("WRR: ", [wrr.next_server() for _ in range(8)])
    # WRR:  ['A', 'A', 'A', 'B', 'A', 'A', 'A', 'B']
// C++: Round Robin and Weighted Round Robin balancers
#include <iostream>
#include <string>
#include <vector>

class RoundRobinBalancer {
    std::vector<std::string> servers_;
    size_t i_ = 0;
public:
    explicit RoundRobinBalancer(std::vector<std::string> servers)
        : servers_(std::move(servers)) {}

    std::string next_server() {
        std::string s = servers_[i_];
        i_ = (i_ + 1) % servers_.size();
        return s;
    }
};

class WeightedRoundRobinBalancer {
    std::vector<std::string> sequence_;   // expanded by weight
    size_t i_ = 0;
public:
    // weighted = {{name, weight}, ...}
    explicit WeightedRoundRobinBalancer(
            const std::vector<std::pair<std::string, int>>& weighted) {
        for (const auto& [name, weight] : weighted)
            for (int n = 0; n < weight; ++n) sequence_.push_back(name);
    }

    std::string next_server() {
        std::string s = sequence_[i_];
        i_ = (i_ + 1) % sequence_.size();
        return s;
    }
};

int main() {
    RoundRobinBalancer rr({"A", "B", "C"});
    std::cout << "RR:  ";
    for (int n = 0; n < 6; ++n) std::cout << rr.next_server() << " ";
    std::cout << "\n";                       // RR:  A B C A B C

    WeightedRoundRobinBalancer wrr({{"A", 3}, {"B", 1}});
    std::cout << "WRR: ";
    for (int n = 0; n < 8; ++n) std::cout << wrr.next_server() << " ";
    std::cout << "\n";                       // WRR: A A A B A A A B
}
// Java: Round Robin and Weighted Round Robin balancers
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

class RoundRobinBalancer {
    private final List<String> servers;
    private int i = 0;

    RoundRobinBalancer(List<String> servers) {
        this.servers = new ArrayList<>(servers);
    }

    String nextServer() {
        String s = servers.get(i);
        i = (i + 1) % servers.size();
        return s;
    }
}

class WeightedRoundRobinBalancer {
    private final List<String> sequence = new ArrayList<>();  // expanded by weight
    private int i = 0;

    WeightedRoundRobinBalancer(List<Map.Entry<String, Integer>> weighted) {
        for (Map.Entry<String, Integer> e : weighted)
            for (int n = 0; n < e.getValue(); n++) sequence.add(e.getKey());
    }

    String nextServer() {
        String s = sequence.get(i);
        i = (i + 1) % sequence.size();
        return s;
    }
}

public class Balancers {
    public static void main(String[] args) {
        RoundRobinBalancer rr = new RoundRobinBalancer(List.of("A", "B", "C"));
        System.out.print("RR:  ");
        for (int n = 0; n < 6; n++) System.out.print(rr.nextServer() + " ");
        System.out.println();                  // RR:  A B C A B C

        WeightedRoundRobinBalancer wrr = new WeightedRoundRobinBalancer(
            List.of(Map.entry("A", 3), Map.entry("B", 1)));
        System.out.print("WRR: ");
        for (int n = 0; n < 8; n++) System.out.print(wrr.nextServer() + " ");
        System.out.println();                  // WRR: A A A B A A A B
    }
}

Least Connections

Least Connections sends each new request to the server currently handling the fewest active connections. This adapts to reality: if one server is stuck on a few slow, long-lived requests, it stops getting new ones. Ideal when requests vary a lot in how long they take.

Least Response Time

Least Response Time goes further: it picks the server with the lowest combination of active connections and fastest recent response time. It's the most "performance-aware" of the common algorithms β€” it actively favors the server that is currently quickest to answer.

IP hash / hash-based

IP hash (a form of hash-based balancing) computes a hash of the client's IP address (or some other key) and uses it to pick a server. The key property: the same client always lands on the same server. This gives us session stickiness β€” useful when a server keeps some per-user state in memory. (Spoiler: this is the bridge to consistent hashing in the next topics.)

Random

Random simply picks a server at random. With enough requests, the math of large numbers spreads load fairly evenly. A popular refinement is "power of two choices": pick two servers at random and send the request to whichever has fewer connections β€” surprisingly close to optimal, with almost no bookkeeping.

Worked example: 6 requests, 3 servers

Servers A, B, C. Watch how the choice differs:

  • Round Robin: A, B, C, A, B, C β€” perfectly even, ignores how busy each is.
  • Weighted RR (A=3, B=2, C=1): A, A, A, B, B, C β€” A does the most work.
  • Least Connections: if A is stuck on a slow upload, requests skip A and go to B and C until A frees up.
  • IP hash: client 203.0.113.7 always β†’ C, every time, so its session stays put.

Comparison β€” which fits when

AlgorithmHow it choosesNeeds server state?Best when…
Round RobinTake turns in orderNoServers & requests are roughly equal
Weighted Round RobinTake turns, more for stronger serversNo (weights are fixed)Servers have different capacities
Least ConnectionsFewest active connectionsYes (live counts)Request durations vary a lot
Least Response TimeFastest + least busyYes (timings + counts)Latency matters most
IP / hash-basedHash of a key β†’ serverNoYou need the same client on the same server (stickiness)
RandomPick at random (or best of two)No (or tiny)Simplicity at large scale
Watch out: "sticky" sessions are a trap

IP-hash stickiness feels handy, but it ties a user to one specific server. If that server dies, everyone stuck to it loses their session. The cleaner design is to make servers stateless (keep session data in a shared cache/database β€” we cover caching in Session 3) so any server can handle any request. Use stickiness only when you truly must.

Recap Round Robin takes turns; Weighted RR gives stronger servers more turns; Least Connections and Least Response Time react to how busy and fast each server is; IP/hash-based pins a client to one server for stickiness; Random (especially "power of two choices") is simple yet effective at scale. Pick based on whether your servers are equal, how much requests vary, and whether you need stickiness.

6 Naive routing schemes & their problems


Explain like I'm 5

Imagine you have 3 toy boxes and a rule for where each toy lives: "count the letters in the toy's name, divide by 3, and use the leftover to pick a box." It works great… until you buy a 4th box. Now your rule says "divide by 4," and suddenly almost every toy has to move to a different box! That huge, annoying reshuffle is the problem with naive routing.

Let's test simple routing ideas against our checklist from Topic 4 (fast, equal, free add/remove, minimal data movement, no syncing). Each one fails somewhere β€” which is exactly why consistent hashing exists.

Attempt 1 β€” Round Robin via modulo

The simplest routing is a % (modulo) based technique: send each request to the "next" server, computed as key % N.

server_list = ["10.11.6.12", "10.11.5.17", ...]

fn handle_request(request):
    key = request.user_id
    N = len(server_list)
    server_id = key % N
    server = server_list[server_id]
    forward_request(request, server)

With server_list = [A, B, C, D] and user_ids 0…100, the initial distribution (user_id % 4) is tidy:

Serveruser_ids it holds (n = 4)
A048 …
B159 …
C2610 …
D3711 …

Now server B crashes, so N = 3 and everything becomes user_id % 3:

Serveruser_ids it now holds (n = 3)
A0369 …
B(crashed)
C14710 …
D25811 …

Yes, users (1, 5, 9, …) who were on the crashed server B obviously have to move β€” that's expected (their data is migrated; how is a separate topic the class jokingly tags "Son Pari"). But other users' data should not need to move, since their servers are still fine. Is that the case? No! Pretty much all the data gets shuffled around, because changing N changes user_id % N for almost everyone. Adding servers has the same effect.

Round Robin (modulo)
ProsFast; equal distribution; no need to sync information between LBs.
ConsLots of unnecessary data movement when N changes.

Attempt 2 β€” Bucketing

Bucketing assigns user_ids to servers in fixed ranges:

Serveruser_ids (total users = 400)
A0 … 99
B100 … 199
C200 … 299
D300 … 399

If server B crashes, the ranges have to be recomputed across the remaining three servers β€” moving a lot of data:

Serveruser_ids after B crashes
A0 … 132
C133 … 265
D266 … 399

And when more users sign up, there's a second problem: we already decided the buckets, so there's no room for new users in existing servers. It's impossible to add new users without first buying more servers.

Bucketing
ProsEqual data distribution; fast.
ConsToo much data re-shuffling when servers change; can't even add new users without buying servers.

Attempt 3 β€” Mapping table

What if the LB just keeps a hashmap from user_id to server_id?

server_list = ["10.11.6.12", "10.11.5.17", ...]
mapping = {
    sanjana: A,
    prem:    B,
    bathula: A,
    pallavi: C,
    venkat:  B,
    ...
}

fn handle_request(request):
    key = request.user_name
    server = mapping[key]
    forward_request(request, server)

Handling crashes and new users is straightforward β€” only the affected entries change:

fn handle_server_crash(crashed_server):
    for user, server in mapping:
        if server == crashed_server:
            # assign a new server to this user
            new_server = get_random_server()
            mapping[user] = new_server
            son_pari_please_migrate(user, crashed_server, new_server)

fn handle_new_user(user):
    server = get_random_server()
    mapping[user] = server

This is much better on data movement: users who were on the crashed server get moved (desired), and users whose servers are still running do not get moved. But there's a fatal catch.

The killer problem: keeping the table in sync

We have many load balancers, and they must all share the exact same mapping table. If different LBs hold different mappings, requests land on random servers and a user's data scatters. And keeping data in sync, always, with very low latency, across many machines is an extremely hard problem β€” practically impossible.

Mapping table
ProsEqual data distribution; fast; minimizes data movement.
ConsMust keep the mapping table in sync across all LBs β€” keeping data in sync always with low latency is essentially impossible.

The deeper flaw behind modulo hashing

When the goal is to spread data (e.g. which cache server stores which key), the modulo trick hash(key) % N is the same Attempt 1 in disguise:

# N = number of servers
server_index = hash(key) % N

Here hash(key) turns the key into a big number, and % N squashes it into 0 … N-1 β€” a valid server index. Looks perfect, until N changes.

Worked numeric example: adding one server remaps almost everything

Suppose 5 keys hash to these numbers, and we start with N = 4 servers:

Keyhash(key)% 4 β†’ server (N=4)% 5 β†’ server (N=5)Moved?
"apple"2020 % 4 = 020 % 5 = 0no
"banana"2121 % 4 = 121 % 5 = 1no
"cherry"2222 % 4 = 222 % 5 = 2no
"date"2323 % 4 = 323 % 5 = 3no
"elder"2424 % 4 = 024 % 5 = 4YES

That small set looks tame, but it's a coincidence of tidy numbers. With realistic, scattered hash values the effect is brutal. Watch what happens to these:

hash(key)% 4 (old)% 5 (new)Moved?
10000no
10111no
10222changed β†’ moves
10333no
10404moves
10510moves
10621moves
10732moves

Of 8 keys, 6 moved. In general, changing from N to N+1 servers forces roughly (N / (N+1)) of all keys to relocate β€” for N=4 that's about 80%. This mass relocation is called the rehashing problem.

Why this is so painful

If those servers are caches, "moving a key" means the data for that key is now looked for on a server that doesn't have it β€” a cache miss. Adding one cache node could invalidate ~80% of your cache at once, sending a flood of traffic to your database. This is sometimes called a cache stampede, and it can take a system down. If those servers are a database shard, you'd have to physically copy most of your data between machines just to add capacity. Unacceptable.

The core flaw

With hash(key) % N, the server a key maps to depends on N. Change N β€” even by one β€” and you've changed the answer for almost every key. We need a scheme where adding or removing a server moves only a small fraction of keys. That's exactly what consistent hashing delivers.

Recap Round Robin / modulo is fast and even but reshuffles almost all data when N changes. Bucketing uses fixed ranges β€” equal & fast but re-shuffles on change and can't even admit new users without new servers. A mapping table minimizes data movement but must be kept perfectly in sync across all LBs β€” effectively impossible. And modulo hashing's rehashing problem remaps ~N/(N+1) of keys (mass cache misses / data copying). Every naive scheme fails the Topic 4 checklist β€” we need something smarter.

7 Hash functions & the hash ring


Explain like I'm 5

A hash function is a magic blender. You drop in anything β€” a name, a number, a whole book β€” and out comes a number inside a fixed range, like 0 to 99. And it's reliable: blend the same thing twice and you always get the same number. We use this to give every user and every server a spot on a giant circle.

What is a function?

A function is a deterministic mapping from inputs to outputs. No matter how many times you call it with the same input, you get the same output:

fn add(a, b):
    return a + b

add(2, 3)  => 5
add(2, 3)  => 5
add(2, 3)  => 5   # always the same for the same input

Compare that to a procedure (an "impure" function) that has side effects β€” it can return different values for the same input because it reads or writes external state:

state = 0
proc add(a, b):
    state = b
    return a + b + state
# Not always the same output β€” it has side effects.

What is a hash function?

A hash function is a "digest" function: it takes anything as input but returns values within a fixed range. Some toy examples that all output a value in 0 … 99 (or some fixed range):

fn hash_1(data):
    return (sum of ascii values of chars of data) % 100
fn hash_2(data):
    return (sum of squares of ascii values of chars of data) % 799
fn hash_3(data):
    return (multiply even and odd numbers in the data) % 1337

# No matter the input, the output is always within the fixed range.
Worth a read

The class linked Wikipedia on hash functions (definitions & properties), cryptographic hash functions (extra guarantees), and universal hashing (families of hashes). See References.

The hash ring

The hash ring is simply the output space of a given hash function, drawn as a circle. If your hash outputs values in 0 … (2⁢⁴ βˆ’ 1), the ring runs from 0 around to 2βΆβ΄βˆ’1 and wraps back to 0.

Placing users & servers on the ring

  • Use 1 hash function hash_0 for hashing the users.
  • Use k hash functions hash_1 … hash_k for hashing the servers β€” a typical value of k is 32 or 64. (Each server is placed at k spots on the ring; this is the "virtual nodes" idea of Topic 8.)

These hash functions are set up while the LB is being coded/configured. All LBs for a system share the exact same set of hash functions, and all (k+1) functions have the same output space (say 0 … 2⁢⁴ βˆ’ 1).

Interview question: will hashes collide?

Suppose the hash function outputs values in 0 … (2⁢⁴ βˆ’ 1). Given 1 billion users and 1 million servers, what's the probability that two hashes collide? Very close to 0 β€” the output space (2⁢⁴ β‰ˆ 18 quintillion) dwarfs the number of things we're placing on it, so collisions are essentially negligible.

Recap A function is a deterministic inputβ†’output mapping; a procedure with side effects is not. A hash function is a digest that maps any input into a fixed range. The hash ring is that output space drawn as a circle. We hash users with one function (hash_0) and servers with k functions (k β‰ˆ 32–64), all sharing the same output space across every LB β€” and with a 2⁢⁴ space, collisions are effectively impossible.

8 Consistent hashing


Explain like I'm 5

Imagine a giant clock face. We give each toy box a spot on the clock, and we give each toy a spot too. To find a toy's box, start at the toy and walk clockwise until you bump into the first box β€” that's its home. Now if you add a new box, you only steal the toys sitting just before it on the clock; every other toy stays exactly where it was. No giant reshuffle! That clock is the hash ring.

Consistent hashing: the only scheme that ticks every box

Against the Topic 4 checklist, consistent hashing scores all of the above: fast Β· equal load distribution Β· freely add/remove servers Β· minimal data movement Β· no need for any sync between LBs. Cons? None.

Consistent hashing solves the rehashing problem with one clever change: instead of hashing keys directly into "server slots," we hash both servers and keys onto the same circle, called the hash ring.

The algorithm

  1. For each server, hash it using all the hash functions hash_1 … hash_k and place the server at those spots on the ring.
  2. When a request arrives, hash the user_id using hash_0 and place the user on the ring.
  3. Forward the request to the first server clockwise from the user's position.
Diagram: the hash ring
        0 / 360Β°  (top of the ring)
                  β”‚
        S_A ●─────┼─────● S_B
          β•±       β”‚       β•²
         β•±   k1   β”‚   k2    β•²     keys walk CLOCKWISE β†’ first server
        β”‚    ↓    β”‚    ↓     β”‚
        β”‚   (β†’A)  β”‚   (β†’B)   β”‚
         β•²        β”‚        β•±
          β•²   k3  β”‚       β•±
        S_C ●─────┼─────● (back to S_A)
                  β”‚
        k1 β†’ next server clockwise = A
        k2 β†’ next server clockwise = B
        k3 β†’ next server clockwise = C
πŸ”‘
Key
hash(key) β†’ a point on the ring
β†’
🧭
Walk clockwise
until you hit a server
β†’
πŸ–₯️
First server
owns this key

A working implementation

Here is a full implementation in three languages. The ring is a sorted structure of (position, server) points (each server placed at many virtual node spots), and get_node is a binary search for the first server clockwise of the key, wrapping around the end of the ring. The demo adds a 4th server and shows that only a small fraction of keys move β€” the rest stay mapped to their original server.

# Python: consistent-hash ring with virtual nodes + binary search
import bisect
import hashlib

RING_SIZE = 1 << 32


def _hash(label):
    digest = hashlib.md5(label.encode()).hexdigest()
    return int(digest, 16) % RING_SIZE


class ConsistentHashRing:
    def __init__(self, vnodes=150):
        self.vnodes = vnodes
        self._positions = []          # sorted ring positions
        self._owner = {}              # position -> server name

    def add_node(self, name):
        for i in range(self.vnodes):
            pos = _hash(f"{name}#{i}")
            if pos not in self._owner:
                bisect.insort(self._positions, pos)
                self._owner[pos] = name

    def remove_node(self, name):
        for i in range(self.vnodes):
            pos = _hash(f"{name}#{i}")
            if self._owner.get(pos) == name:
                del self._owner[pos]
                idx = bisect.bisect_left(self._positions, pos)
                self._positions.pop(idx)

    def get_node(self, key):
        if not self._positions:
            return None
        pos = _hash(key)
        # first ring position clockwise of the key (wrap to 0)
        idx = bisect.bisect_right(self._positions, pos) % len(self._positions)
        return self._owner[self._positions[idx]]


if __name__ == "__main__":
    ring = ConsistentHashRing()
    for s in ["A", "B", "C"]:
        ring.add_node(s)

    keys = [f"user-{i}" for i in range(10)]
    before = {k: ring.get_node(k) for k in keys}

    ring.add_node("D")                 # add a 4th server
    after = {k: ring.get_node(k) for k in keys}

    moved = [k for k in keys if before[k] != after[k]]
    print("moved:", moved)         # only a small fraction remap to D
    print("stayed:", [k for k in keys if k not in moved])
// C++: consistent-hash ring with virtual nodes + binary search
#include <cstdint>
#include <functional>
#include <iostream>
#include <map>
#include <string>
#include <vector>

class ConsistentHashRing {
    int vnodes_;
    std::map<uint64_t, std::string> ring_;   // sorted: position -> server

    static uint64_t hashLabel(const std::string& label) {
        return std::hash<std::string>{}(label);
    }
public:
    explicit ConsistentHashRing(int vnodes = 150) : vnodes_(vnodes) {}

    void add_node(const std::string& name) {
        for (int i = 0; i < vnodes_; ++i)
            ring_[hashLabel(name + "#" + std::to_string(i))] = name;
    }

    void remove_node(const std::string& name) {
        for (int i = 0; i < vnodes_; ++i) {
            auto it = ring_.find(hashLabel(name + "#" + std::to_string(i)));
            if (it != ring_.end() && it->second == name) ring_.erase(it);
        }
    }

    std::string get_node(const std::string& key) const {
        if (ring_.empty()) return {};
        // first position clockwise of the key (wrap to begin)
        auto it = ring_.upper_bound(hashLabel(key));
        if (it == ring_.end()) it = ring_.begin();
        return it->second;
    }
};

int main() {
    ConsistentHashRing ring;
    for (const std::string& s : {"A", "B", "C"}) ring.add_node(s);

    std::vector<std::string> keys;
    for (int i = 0; i < 10; ++i) keys.push_back("user-" + std::to_string(i));

    std::vector<std::string> before;
    for (const auto& k : keys) before.push_back(ring.get_node(k));

    ring.add_node("D");                  // add a 4th server

    std::cout << "moved:";
    for (size_t i = 0; i < keys.size(); ++i)
        if (ring.get_node(keys[i]) != before[i]) std::cout << " " << keys[i];
    std::cout << "\n";                       // only a small fraction remap to D
}
// Java: consistent-hash ring with virtual nodes + binary search
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;

class ConsistentHashRing {
    private final int vnodes;
    private final SortedMap<Long, String> ring = new TreeMap<>();

    ConsistentHashRing(int vnodes) { this.vnodes = vnodes; }
    ConsistentHashRing() { this(150); }

    private static long hash(String label) {
        try {
            byte[] d = MessageDigest.getInstance("MD5")
                .digest(label.getBytes(StandardCharsets.UTF_8));
            long h = 0;
            for (int i = 0; i < 8; i++) h = (h << 8) | (d[i] & 0xff);
            return h >>> 32;          // 32-bit ring position
        } catch (Exception e) { throw new RuntimeException(e); }
    }

    void addNode(String name) {
        for (int i = 0; i < vnodes; i++) ring.put(hash(name + "#" + i), name);
    }

    void removeNode(String name) {
        for (int i = 0; i < vnodes; i++) {
            long pos = hash(name + "#" + i);
            if (name.equals(ring.get(pos))) ring.remove(pos);
        }
    }

    String getNode(String key) {
        if (ring.isEmpty()) return null;
        // first position clockwise of the key (wrap to first)
        SortedMap<Long, String> tail = ring.tailMap(hash(key) + 1);
        Long pos = tail.isEmpty() ? ring.firstKey() : tail.firstKey();
        return ring.get(pos);
    }
}

public class Ring {
    public static void main(String[] args) {
        ConsistentHashRing ring = new ConsistentHashRing();
        for (String s : List.of("A", "B", "C")) ring.addNode(s);

        List<String> keys = new ArrayList<>();
        for (int i = 0; i < 10; i++) keys.add("user-" + i);

        List<String> before = new ArrayList<>();
        for (String k : keys) before.add(ring.getNode(k));

        ring.addNode("D");                 // add a 4th server

        System.out.print("moved:");
        for (int i = 0; i < keys.size(); i++)
            if (!ring.getNode(keys.get(i)).equals(before.get(i)))
                System.out.print(" " + keys.get(i));
        System.out.println();              // only a small fraction remap to D
    }
}

Checking it against the requirements

Interview-style Q&A on this algorithm
  • Is the algo deterministic? Will Sanjana always go to server C (as long as Sanjana is alive, server C is running, and no servers are added/removed)? Yes.
  • Is the algo fast? Yes. The ring is a sorted array, so finding the nearest server clockwise is a binary search (upper bound) for the user's hash: O(log(Nk)), where N = number of servers and k = number of server hashes.
  • Do the LBs have to share any data to stay in sync? No. Each LB knows which servers are running (via health-check / heartbeat), and the code in every LB is identical β€” so each LB computes the same hashes, builds the same ring, and picks the same server for any user.
  • Is the data distribution even? With enough server hashes (k β‰ˆ 32–64) the points interleave finely around the ring, so each server owns many small arcs that add up to a fair share β€” this is what virtual nodes fix (below).

The payoff: only K/N keys move

Here's the magic. When a new server S_D joins, it lands at some single spot on the ring. The only keys that change homes are the ones sitting in the arc between S_D and the previous server clockwise-before it β€” those keys now bump into S_D first. Every other key on the ring is completely unaffected.

On average, a single node owns about 1/N of the ring, so adding or removing a node only moves about K/N keys (where K = total keys, N = number of servers) instead of nearly all of them. Removing a server is the mirror image: its keys simply flow clockwise to the next server, and nobody else moves.

Worked example: adding a node

Ring positions (0–100 for simplicity). Servers: A=10, B=40, C=70. Keys land at: k1=5, k2=25, k3=55, k4=85. Walking clockwise:

  • k1=5 β†’ first server clockwise is A(10) β†’ A
  • k2=25 β†’ B(40) β†’ B
  • k3=55 β†’ C(70) β†’ C
  • k4=85 β†’ wraps past 100 back to A(10) β†’ A

Now we add D at position 60. Re-walk only what could change:

  • k1 β†’ A (unchanged)
  • k2 β†’ B (unchanged)
  • k3=55 β†’ first server clockwise is now D(60) β†’ moved from C to D
  • k4 β†’ A (unchanged)

Only k3 moved. Compare that to naive modulo, where adding a 4th server would have relocated most keys. That is the win.

A simpler pseudocode view of the ring

# A sorted list of (position, server) points around the ring.
ring = sorted_map()   # keyed by position 0 .. 2^32-1

def add_server(name):
    pos = hash(name) % RING_SIZE
    ring[pos] = name

def remove_server(name):
    pos = hash(name) % RING_SIZE
    del ring[pos]

def get_server(key):
    pos = hash(key) % RING_SIZE
    # walk clockwise: first server position >= pos
    for p in ring.keys_sorted():
        if p >= pos:
            return ring[p]
    # wrapped past the end β†’ take the first server (smallest pos)
    return ring.first_value()

Virtual nodes β€” fixing uneven balance

One catch: if you only place one point per server, the ring can be lumpy. By bad luck, two servers might land close together, leaving a huge arc owned by a third server β€” which then gets far more keys. The fix is virtual nodes (also called vnodes): place each physical server at many points around the ring (e.g. A#1, A#2, … A#150), each hashed separately. This is exactly the role of the k server hash functions (k β‰ˆ 32–64) from Topic 7.

  • With ~100–200 vnodes per server, the points interleave finely, so each physical server owns many small arcs that add up to a fair, even share.
  • Bonus: when a server dies, its load spreads across many other servers (one per vnode arc), rather than dumping entirely onto a single unlucky neighbor.
def add_server(name, vnodes=150):
    for i in range(vnodes):
        pos = hash(f"{name}#{i}") % RING_SIZE
        ring[pos] = name   # many points, all mapping back to 'name'
Key takeaway

Consistent hashing decouples a key's location from the count of servers. Because keys and servers share one ring and a key belongs to "the next server clockwise," adding or removing a node only disturbs the keys in one arc β€” about K/N of them β€” instead of nearly all. Virtual nodes make the arcs even and spread failure load smoothly.

Recap Consistent hashing places servers and keys on a circular hash ring; each key is owned by the first server clockwise from it. It is the only scheme that's fast, evenly distributed, add/remove-friendly, low-movement, and needs no LB sync (all LBs run identical code + know live servers via heartbeat). Routing is a binary search, O(log(Nk)). Adding or removing a node moves only the keys in one arc (~K/N), not everything. Virtual nodes (the k server hashes) give each server many ring points so load stays balanced and a failed node's keys redistribute across many peers.

9 Where consistent hashing is used


Explain like I'm 5

The "clock with toy boxes" trick is so useful that grown-up systems use it everywhere they have to decide "which machine holds this piece of data?" β€” caches, databases, even the networks that deliver videos. Anytime you might want to add or remove machines without shuffling everything, the ring shows up.

Consistent hashing is one of the most reused ideas in distributed systems. It appears wherever data must be sharded (split) across many nodes that can come and go. Here's where you'll meet it in the real world.

AreaReal systemsWhat the ring decides
Distributed cachesMemcached clients, Redis Cluster, Twitter's TwemproxyWhich cache node stores each key, so adding nodes doesn't wipe the cache (Topic 6 problem).
NoSQL databasesAmazon DynamoDB, Apache Cassandra, Riak, ScyllaDBWhich node owns each row/partition; nodes join/leave with minimal data movement.
CDNsAkamai, Cloudflare, Fastly edge cachesWhich edge server caches a given object, so a popular file isn't duplicated everywhere.
Sharding / partitioningApplication-level shard routers, message-queue partitionersWhich shard a user or record belongs to, allowing the fleet to grow smoothly.
Real example: Amazon DynamoDB & Cassandra

The famous 2007 Amazon Dynamo paper popularized consistent hashing in databases, and its ideas live on in DynamoDB and Cassandra. In Cassandra, every node owns a set of token ranges on a ring β€” exactly the virtual-node arcs from Topic 8. When you add a node to a Cassandra cluster, it claims some ranges and only the data in those ranges streams over from neighbors; the rest of the cluster keeps serving traffic untouched. That's why these databases can scale by "just adding nodes."

Real example: distributed caches & CDNs

A memcached cluster with a consistent-hashing client (sometimes called ketama hashing) means restarting or adding a cache node only invalidates a small slice of cached keys β€” avoiding the cache stampede we saw in Topic 6. CDNs use the same trick at the edge: a request for /video/123.mp4 is consistently routed to the one edge server that should cache it, so the file isn't needlessly stored on every edge box and cache hit rates stay high.

Connecting the dots

We'll use consistent hashing directly when we build distributed caches in Session 3 (Caching 1), and again when we discuss database sharding and replication in later sessions. Notice it also overlaps with the IP-hash load balancing from Topic 5 β€” both use a hash to pin "this thing" to "that machine," just for different purposes (requests vs. data).

Recap Consistent hashing powers distributed caches (Memcached, Redis Cluster), NoSQL databases (DynamoDB, Cassandra), CDNs (edge caching), and general sharding. Everywhere the question is "which machine holds this key and can I add/remove machines cheaply?", the hash ring is the answer β€” and we'll lean on it again in upcoming caching and sharding sessions.

β˜… Putting it all together


This session was all about spreading work and data across many machines β€” gracefully. Here's the one-paragraph story tying the topics together:

A load balancer gives users a unified front door and routes each request to a healthy backend, operating at L4 (IP/port, fast and blind) or L7 (HTTP-aware), as hardware, software like Nginx/HAProxy, or a cloud service (AWS ELB) β€” using health checks to skip dead nodes. But which server gets a request is the same question as which data lives where: sharding is a side effect of routing. A good routing algorithm must be fast, evenly distributed, add/remove-friendly, low-movement, and need no LB sync. Naive schemes all fail it: Round Robin/modulo and bucketing reshuffle almost everything when N changes (the rehashing problem, ~N/(N+1) of keys), and a mapping table can't be kept in sync across LBs. Consistent hashing wins on every count by putting servers and keys on a hash ring (the output space of k+1 shared hash functions), where each key belongs to the next server clockwise β€” routing is an O(log Nk) binary search, only ~K/N keys move on change, virtual nodes keep load even, and identical code in every LB needs zero syncing. That's why real systems like DynamoDB, Cassandra, Memcached, and CDNs rely on it to scale by simply adding machines.

Quick self-check

What are the two purposes of a load balancer?

(1) Provide a unified view of the whole backend to the end user (who doesn't care which exact server handles their request), and (2) distribute the load β€” requests and data β€” equally across the app/DB servers.

How many IPv4 addresses exist, and why isn't that enough?

IPv4 is 32 bits β†’ 2Β³Β² β‰ˆ 4 billion addresses. But there are roughly 100 billion–1 trillion devices online, so it's far too few β€” which is why IPv6 uses 128 bits (2¹²⁸ β‰ˆ 256 billion billion billion billion).

Why is sharding "a side effect of routing"?

If we always route Sanjana's requests to server A, then only server A ever sees and stores her data β€” B and C never received it. So deciding (request β†’ server) automatically decides (data β†’ server). The same algorithm must drive both, or requests hit servers without the right data.

What can an L7 load balancer route on that an L4 one cannot?

Application-layer details of the HTTP request β€” URL path, hostname, headers, and cookies (e.g. send /api/* to one pool, /images/* to another). An L4 LB only sees IP addresses, ports, and the transport protocol.

Why does bucketing make it impossible to add new users?

The buckets (id ranges per server) are fixed in advance, so existing servers have no room for new users. You can't admit new users without first buying more servers β€” and rebalancing on a crash also reshuffles huge amounts of data.

What's the fatal flaw of the mapping-table approach?

All LBs must hold the exact same user→server map. Keeping that data in sync, always, with very low latency across many LBs is practically impossible; if maps diverge, requests land on random servers and a user's data scatters.

With hash(key) % N, why does adding one server cause chaos?

The chosen server depends on N. Going from N to N+1 changes the remainder for roughly N/(N+1) of all keys, so almost everything remaps β€” the rehashing problem. For caches that means mass misses; for databases, copying most of the data.

Given a 2⁢⁴ hash space, 1 billion users and 1 million servers, will hashes collide?

The probability is very close to 0. The output space (2⁢⁴ β‰ˆ 18 quintillion) is vastly larger than the number of items being placed, so collisions are negligible.

In consistent hashing, how do you find which server owns a key, and how fast is it?

Hash the key to a point on the ring, then walk clockwise to the first server (wrapping around the top if needed). On a sorted-array ring this is a binary search: O(log(Nk)) for N servers and k hashes each.

Do load balancers need to talk to each other to stay in sync under consistent hashing?

No. Each LB runs identical code (same hash functions) and learns which servers are alive via health-checks/heartbeats. So every LB builds the same ring and routes any given user to the same server β€” no inter-LB communication needed.

Why do we use virtual nodes?

To balance load. Placing each physical server at many points on the ring (the k server hashes) breaks it into many small arcs, so each server owns a fair, even share β€” and when a server fails, its keys spread across many peers instead of dumping onto one neighbor.

Name two real systems that use consistent hashing and what it decides for them.

For example: Cassandra/DynamoDB use it to decide which node owns each partition (so nodes can join/leave cheaply); Memcached/Redis Cluster and CDNs use it to decide which cache/edge node stores each key (so adding nodes doesn't wipe the cache).

πŸ“š References & Further Reading


Class material

Tutorials & cloud docs

Hash functions & consistent hashing