Post
Load balancing with Consistent Hashing

In a recent conversation, I realised I knew consistent hashing existed, and what you could use it for. However, I couldn’t immediately bring to mind how it actually worked. Not awful — I knew I could easily find out — but I also felt I should have known it.

I’ve found that a good way to get an algorithm to stick in my head is to implement it. Even better to then write about it. I remember the parts of toykv that I wrote about more clearly than those I only coded.

There are a few other algorithms that I like but I can’t quite remember. Rendezvous hashing and reservoir sampling come to mind as ones where embedding the idea solidly would be valuable. I’d also like a deeper understanding of bloom filters. I like data sketches, so perhaps I’ll be brave enough to look harder at hyperloglog and count-min. We’ll see how it goes.

So let’s start a small series here to get some more depth and memories of these useful algorithms. And where better to start than getting consistent hashing fixed securely into my brain.

A high level view

Consistent hashing allows a set of servers to agree on where a given item belongs without needing to communicate with each other. Use-cases include:

• A set of frontend routers receiving requests that wish to send requests from the same source to the same backend servers.
• A sharded key-value store that wants all of its servers to agree on the server that stores each key.

A key strength of consistent hashing is that if the set of servers change, only a minimum number of items change their mapping to servers. This is critical for many use-cases. For example, if you shard your data, you don’t want every key to move when you add a new server! We’ll talk about how this works later.

One caveat: agreement requires that each routing server holds the same set of destinations. If they disagree, then they will select different backends for some requests. But consistent hashing will minimise the difference.

How it works

First, we have a hash function, h(a) that hashes a to a number between 0 and 2⁶⁴-1 — a 64-bit non-negative integer. We use this hash function to decide on destinations for items as follows:

1. For every server, we generate its hash. We put the hashes on a ring that has 2⁶⁴ locations.
2. When we need to decide upon a server for an item, we take the item’s key and apply h(key) to get a position for the item on the ring.
3. In this implementation, we choose the first server in the ring that is greater than or equal to h(key).

I find it a bit easier to understand this using a diagram:

That is, server B owns items with h(A) < h(key) <= h(B). Server A owns all the way from h(C) < h(key) <= h(A), that is, it wraps around.

It’s a little confusing that the item is assigned to the next server in the ring, but it simplifies the maths slightly.

Uneven load

Statistically, using a single hash location for each server will result in uneven load. In the diagram above, server A owns the vast majority of the keys.

We solve this problem by granting each server several locations on the ring. The random nature of the hash function means that the more locations we have for each server, the more evenly the load will be distributed. We can add a location number to the hash function to perturb the locations: h(server, location).

This is shown below. By adding two locations for each server, we even the load somewhat. Server C still owns the largest part of the ring, but it’s not as uneven. It should be a bit clearer after looking at the diagram why adding more locations will even out the load further.

Changing servers

One key piece of consistent hashing “magic” appears when servers change: instead of servers A, B and C, you now have A, B, C and D. Ideally most of your items end up being directed to the same servers after the change. Eg, if you are sharding to several Redis servers as a cache, adding a new Redis server shouldn’t invalidate most of your following calls to the cache.

Consistent hashing gives you this because:

• Adding a new server doesn’t change the positions of any other server’s locations on the ring.
• Items are assigned based on the next server location on the ring.
• Therefore only those items whose next server on the ring is now server D change their server — other keys retain their current server mapping.

Adding server D gives server D some locations on the ring. Only those items which now hash to locations for which D is the next server on the ring move servers. All other items stay the same!

Hopefully this diagram makes it clearer what happens:

Implementing

I chose to implement this in Rust. We will add multiple locations per server to even out the load. I chose to use 5 locations as the default, but this was arbitrary; I didn’t do any testing of the evenness.

First we need a Router:

struct Server {
    hash: u64,
    name: String,
}

struct Router {
    /// List of servers; one entry per server location
    servers: Vec<Server>,
    /// Number of locations on the ring for each server
    locations: u64,
}


impl Router {
    /// Create a new Router with 5 locations on the ring per server
    fn new() -> Self {
        Self {
            servers: vec![],
            locations: 5,
        }
    }
}

Adding servers

First we need a hash function for a server + location combination:

/// Hashes a name to a ring location
fn hash_server(name: &str, location: u64) -> u64 {
    let mut h = Xxh3::new(); // new() uses a consistent seed
    h.write_u64(location);
    h.write(name.as_bytes());
    h.finish()
}

Digression on the hasher

If you are running multiple Router instances, the hash function must return the same hash for the same input on all servers. To achieve this, I’ve used xxhash-rust’s xxh3 with a fixed key, which provides this guarantee. I used xxhash in toykv too, for the on-disk integrity hash.

To make my code work in your project, add xxh3 using:

cargo add xxhash-rust --features xxh3

Then it’s the following use line:

use xxhash_rust::xxh3::{Xxh3, xxh3_64};

Digression in the digression: hash flooding

I read a bit deeper about hash functions while deciding to use xxh3. One place I hadn’t looked at in detail before was the hash flooding attack. It came up because Rust’s default hash algorithm (used in its hashmap) is SipHasher which was chosen to be resistant to hash flooding.

Claude gave me a useful summary of the attack:

find a large set of keys that all hash to the same bucket, then send them to a server that puts attacker-controlled strings into a hash table. If every key lands in the same bucket, the chain in that bucket is just a linked list, and each insert or lookup is O(n). Insert n colliding keys and the total work to build the table is O(n²). It’s similar for probing hashtables.

Classic targets were web frameworks that parsed HTTP POST parameters into a hash map before your code ever ran. An attacker sends one POST body with, say, a million colliding parameter names — a couple of MB of traffic — and the server burns minutes of CPU inserting them.

The point was that most languages and web frameworks used deterministic hash functions, and so once you’d worked out your collisions for a language, you could use them everywhere. Happy days — for the attacker.

SipHasher prevents the attack by using a secret key. The hasher is seeded by a random key at process start, and an attacker who doesn’t know the key cannot predict which inputs will collide. On top of that, SipHasher is carefully designed to avoid leaking information about the key if an attacker knows input X and the result of hashing X. That makes it impossible to for an attacker to derive a key at runtime by sending many hash inputs.

SipHasher is fast enough that it’s used as a safe default in Rust’s hash maps. But we choose xxh3 because it is faster and safe for this use-case. Hash-flooding doesn’t apply to consistent hashing: we are using a ring, so there’s no hash table with degenerate performance properties under collisions. Our get search time is binary search’s O(log n) regardless of collisions, so an attacker cannot eat our CPU via crafted keys.

Back to our scheduled program

We then need to use the function:

impl Router {
    // ...

    /// Generate the set of hashes for a server
    fn hashes(&self, name: &str) -> Vec<u64> {
        (0..self.locations)
            .map(|i| hash_server(name, i))
            .collect() // O(ring_locations)
    }

    /// Add a server to the ring
    pub fn add_server(&mut self, name: String) {
        let hashes = self.hashes(&name);
        self.add_server_with_hashes(name, hashes);
    }
    fn add_server_with_hashes(&mut self, name: String, hashes: Vec<u64>) {
        for hash in hashes {
            self.servers.push(Server {
                hash,
                name: name.clone(),
            });
        }

        // O(N log N), N = num_servers * ring_locations
        self.servers.sort_by_key(|s| s.hash);
    }
}

• First we create Router::hashes which generates the list of hashes for a server name.
• Next we create the “public” API, add_server(&mut self, name: String). This first generates the hashes and delegates to add_server_with_hashes.
• add_server_with_hashes appends an entry to the ring for each server location. It then sorts by the hash so we can use binary search to select the server. add_server is (probably) called rarely, so it can afford a relatively expensive O(n log n) sort operation to save time in the hot path later.

We use add_server_with_hashes to help with testing. It allows us to add servers to explicit locations on the ring during tests, which makes the tests clearer because hashes are not “hidden” by calling the hash function.

Removing servers

I didn’t think too hard on this code, I’m sure there are more efficient ways to do this, but for now a simple linear scan to remove them:

impl Router {
    // ...

    /// Remove a server from the ring
    fn remove_server(&mut self, name: &str) {
        let old = mem::take(&mut self.servers);
        self.servers = old.into_iter().filter(|s| s.name != name).collect();
    }
}

Selecting the server for an item

Now we get to the meat of it. Recall that at this point we have a list (sorted by hash) of servers that looks something like this (if locations were two not five):

  hash  server
[ 1234,   A   ]
[ 2345,   C   ]
[ 3456,   A   ]
[ 4567,   B   ] <-- server for h(item_key) = 3900
[ 5678,   C   ]
[ 6789,   B   ]

Recall the plan: we will generate h(item_key) and then find the server that comes next on the ring from the location of the item. So if h(item_key) = 3900, we would choose server B.

One note is that if the item’s key hash is, say, 9000, then the “next” server on the ring is at location 1234 — because we wrap around.

The simplest way to find the right server is linear search: look at each item, from lowest hash (at index 0) to highest. Remember we’re looking for the server after the item key’s hash. So for 3900, we’d say: 1234 < 3900, next; 2345 < 3900, next; 3456 < 3900, next; 4567 >= 3900, got it!

This works fine for 6 entries in the server Vec, but we want it to work for 1,000 or even 10,000 servers. At even 5 locations on the ring per server, that’s up to 5,000 (or 50,000) comparisons for every server selection. In big-O notation, that’s O(n*m) where n is the number of servers and m is the number of locations.

We can do better. We turn to the first “computer science” algorithm you learn: binary search. Binary search is much faster than linear search as it splits the search space in half for every comparison, so it only needs to do O(log (n*m)) comparisons.

We have a slight wrinkle in that we’re not looking for an exact item, but instead which server’s range our item’s key hash falls into. Traditional binary search looks for an exact match, but it turns out there are simple variants on binary search to find:

• lower bound: the lowest index in the array whose hash is greater than or equal to the searched item.
• upper bound: the lowest index in the array whose hash is greater than the searched item.

These are traditionally used to search for which items fall into an interval (eg, “which items in the array are between 1234 and 56049493”) without checking every item.

But the lower bound is exactly what we need for our server choice, recalling that we want to choose the first server whose hash is greater than or equal to our item’s key hash.

To remind us:

The lower bound binary search algorithm looks like this, returning the size of the array if the item is greater than any item in the array (our wraparound case):

/// Return the lowest index in servers where the server's hash
/// is greater than or equal to the key hash. If no server has
/// a greater or equal hash, returns servers.len().
fn lower_bound(servers: &[Server], key_hash: u64) -> usize {
    let mut low = 0;
    let mut high = servers.len();
    while low < high {
        let mid = low + (high - low) / 2;
        if servers[mid].hash < key_hash {
            low = mid + 1;
        } else {
            high = mid;
        }
    }
    low
}

We also need the function to hash the item key, which is trivial:

/// Hashes a key
fn hash_key(key: &str) -> u64 {
    xxh3_64(key.as_bytes())
}

Now that we have this, the get method is trivial:

impl Router {
    // ...

    /// Return the server to use for key
    pub fn get(&self, key: &str) -> Option<&String> {
        self.get_with_hash(hash_key(key))
    }
    fn get_with_hash(&self, key_hash: u64) -> Option<&String> {
        if self.servers.is_empty() {
            return None;
        }
        let n = lower_bound(&self.servers, key_hash);

        // If key_hash is after last item, n will
        // be set to self.servers.len(); `mod len` puts us
        // back at index zero, wrapping around, in that
        // case.
        let server = &self.servers[n % self.servers.len()];

        Some(&server.name)
    }
}

1. Get the hash of the item’s key.
2. Special case empty.
3. Find the server via our lower_bound method.
4. mod the returned value to wrap around if needed.
5. Done.

Again, for ease of testing, we separate the public method that takes the key from one where we specify the hash directly.

Testing

To show why testing is easier when you can set the hashes directly, here’s an example from the test suite I wrote for this:

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn single_hash_per_server() {
        let mut router = Router::new();
        router.add_server_with_hashes("A".to_string(), vec![100]);
        router.add_server_with_hashes("B".to_string(), vec![200]);
        router.add_server_with_hashes("C".to_string(), vec![300]);

        // Check that we assign things correctly to start with
        assert_eq!(Some(&"A".to_string()), router.get_with_hash(0));
        assert_eq!(Some(&"A".to_string()), router.get_with_hash(10));
        assert_eq!(Some(&"A".to_string()), router.get_with_hash(100));
        assert_eq!(Some(&"B".to_string()), router.get_with_hash(120));
        assert_eq!(Some(&"B".to_string()), router.get_with_hash(200));
        assert_eq!(Some(&"C".to_string()), router.get_with_hash(220));
        assert_eq!(Some(&"C".to_string()), router.get_with_hash(300));
        assert_eq!(Some(&"A".to_string()), router.get_with_hash(320));
    }
}

By using the *_with_hash variants we get to take the hash implementation out of the equation and concentrate on the more complicated bits that we had to implement. We’re unlikely to fail in the generation of the hashes from the server names or the item keys, but we need to be sure we got the selection logic right; that we followed the rules so we can be sure each independent Router would select the same servers for each item.

And so…

And so I hope to have fixed this algorithm in my mind. At least I’ll be able to bring to mind the pictures of the ring, and the servers placed upon it, and piece back together how to write that down in code. I’ve also hopefully committed to memory that there are specialisations of binary search for lower and upper bounds.

To further fix those upper and lower bounds, it occurs to me I have exactly this need in toykv where I have a half-way there version that I should replace now I know something better…