Shards and consistent hashing

The caches of chapter 13 serve one thread at a time. This chapter asks what happens when many threads, or many machines, share one set of keys.

Both halves of the chapter use the same idea. You split the keys into groups by their hash, and each group is handled separately. A group of keys handled separately is called a shard.

Section 14.1 splits a map inside one program, so threads that use different keys do not wait for each other. Section 14.2 splits keys across machines, so that adding or removing a machine moves as few keys as possible.

14.1 A map split into shards

Section 8.6 showed how to share a value between threads: wrap it in a Mutex, and share the Mutex with an Arc. A thread calls lock() before it touches the value. While one thread holds the lock, every other thread that calls lock() waits.

A single Mutex<HashMap<K, V>> is safe, but it has a cost. If four threads use the map at once, three of them wait, even when each thread wants a different key. Waiting for a lock that another thread holds is called lock contention.

The fix is to give the map several locks. Split it into, say, eight smaller maps, each with its own Mutex (figure 14.1). A key always goes to the same small map. Two threads then wait for each other only when their keys land in the same shard.

Animation 14.1 sends one key and value from each thread into the map. Under the single lock the pairs go in one at a time and stay. Under eight shards they go in together, each into the shard that owns its key.

14.1.1 The type

The cache holds a list of shards, each a HashMap behind its own Mutex. It also holds the hasher that decides which shard a key belongs to.

Listing 14.1 The type and the constructor (lines 16 to 42). src/problems/sharded_cache.rs

use std::{
    collections::{HashMap, hash_map::RandomState},
    hash::{BuildHasher, Hash},
    sync::Mutex,
};

/// A fixed-size map split across independently locked shards.
///
/// The shard count is rounded up to a power of two so the index is a mask instead
/// of a modulo.
pub struct ShardedCache<K, V> {
    shards: Vec<Mutex<HashMap<K, V>>>,
    hasher: RandomState,
    mask: usize,
}

impl<K, V> ShardedCache<K, V>
where
    K: Hash + Eq,
{
    /// Create a cache with at least `shard_count` shards (rounded to a power of two).
    pub fn new(shard_count: usize) -> Self {
        let shard_count = shard_count.max(1).next_power_of_two();

        let shards = (0..shard_count)
            .map(|_| Mutex::new(HashMap::new()))
            .collect();
        Self {
            shards,
            hasher: RandomState::new(),
            mask: shard_count - 1,
        }
    }
    // ...
}

ShardedCache has three fields:

• shards is a Vec of maps, each inside its own Mutex.
• hasher produces the hash that picks a shard for a key.
• mask turns a hash into a shard number, as the next section explains.

new builds the Vec with an iterator. (0..shard_count).map(|_| Mutex::new(HashMap::new())) makes one empty, locked map per shard, and collect() gathers them. The _ in the closure ignores the shard number, which is not needed.

The first line of new adjusts the requested count. shard_count.max(1) turns a request for 0 shards into 1. .next_power_of_two() rounds up to the next power of two, so a request for 6 shards gives 8.

14.1.2 Picking a shard with a mask

A shard count that is a power of two lets the code pick a shard cheaply. For 8 shards, the mask is 8 − 1 = 7, which is 0b111 in binary. The operation hash & mask keeps the lowest three bits of the hash and clears the rest. The result is a number from 0 to 7 (figure 14.2).

hash % 8 would give the same answer. The & is one simple CPU instruction, while % is a division, which takes more time. The difference is small, but this code runs on every call.

Listing 14.2 Picking a shard (lines 45 to 48).

impl<K, V> ShardedCache<K, V>
where
    K: Hash + Eq,
{
    // ...
    fn shard(&self, key: &K) -> &Mutex<HashMap<K, V>> {
        let index = (self.hasher.hash_one(key) as usize) & self.mask;
        &self.shards[index]
    }
    // ...
}

hash_one hashes the key with the cache’s RandomState and returns a u64. The mask keeps its low bits as the shard index. It returns a reference to the chosen shard’s Mutex.

The cache keeps one RandomState for its whole life. RandomState::new() picks random secret keys for the hash function when the cache is created. Every lookup in the same cache uses the same keys, so a key always maps to the same shard. A different cache, or the same program run again, would place the keys differently.

HashMap uses RandomState for the same reason. Random keys stop an attacker from choosing inputs that all hash to the same place and slow the map down.

// 1. The bytes added up. Every order of the same bytes gives one number.
fn hash_sum(bytes: &[u8]) -> u64 {
    bytes.iter().map(|b| u64::from(*b)).sum()
}

// 2. The first byte only. Every key with the same first letter collides.
fn hash_first(bytes: &[u8]) -> u64 {
    u64::from(bytes.first().copied().unwrap_or(0))
}

// 3. The key itself, for a numeric key. The low bits repeat every eight.
fn hash_identity(key: u64) -> u64 {
    key
}

14.1.3 Each method locks one shard

A single-key operation touches one shard only. It hashes the key, locks that shard, and leaves every other shard free for other threads.

Listing 14.3 insert, get, and with (lines 50 to 80).

impl<K, V> ShardedCache<K, V>
where
    K: Hash + Eq,
{
    // ...
    /// Insert a value, returning the previous one if the key was present.
    pub fn insert(&self, key: K, value: V) -> Option<V> {
        self.shard(&key)
            .lock()
            .expect("shard mutex poisoned")
            .insert(key, value)
    }

    /// Read a value by cloning it out of the shard.
    pub fn get(&self, key: &K) -> Option<V>
    where
        V: Clone,
    {
        self.shard(key)
            .lock()
            .expect("shard mutex poisoned")
            .get(key)
            .cloned()
    }

    /// Run `inspect` against the value under the shard lock, without cloning.
    ///
    /// Use this for read-modify-write on a single key; the closure holds the lock
    /// for its duration, so keep it short and do not call back into the cache.
    pub fn with<R>(&self, key: &K, inspect: impl FnOnce(Option<&V>) -> R) -> R {
        let guard = self
            .shard(key)
            .lock()
            .expect("shard mutex poisoned");
        inspect(guard.get(key))
    }
    // ...
}

Each method follows the same three steps: pick the shard, lock it, and call the matching HashMap method. The methods take &self, not &mut self. The Mutex provides the permission to change the map, so a shared reference to the cache is enough. Several threads can then share one cache through an Arc.

lock() returns a Result. It is an error only if another thread panicked while it held the same lock. Such a lock is called poisoned, and chapter 16 covers it. Here .expect(...) turns that error into a panic with a clear message.

The lock stays held until the guard returned by lock() is dropped. In insert, the guard is a temporary value in one expression, so it is dropped at the end of the statement.

get must return a clone of the value, which is why it asks for V: Clone. A reference to the value would point into the map, and the map is only reachable through the guard. When get returns, the guard is dropped and the lock released. A reference could then be read while another thread changes the map, so the compiler does not allow it to escape.

14.1.4 Reading under the lock with a closure

Cloning a large value only to read one field of it wastes work. with avoids the clone. You pass it a closure, and with runs the closure while it holds the lock:

pub fn with<R>(&self, key: &K, inspect: impl FnOnce(Option<&V>) -> R) -> R

• inspect: impl FnOnce(Option<&V>) -> R accepts any closure that takes an Option<&V> and returns some type R. FnOnce means with calls it at most once.
• The closure receives a reference into the map, valid while the lock is held.
• Only the closure’s result, of type R, leaves the function. R cannot borrow from the map.

The test with_exposes_the_value_without_cloning uses it to read the length of a String:

let observed = cache.with(&"node", |value| value.map(|v| v.len()));

The doc comment of with gives two rules. Keep the closure short, because every other thread that needs the same shard waits while it runs. And do not call the cache from inside the closure. A call that lands on the same shard would try to lock a Mutex that this thread already holds. The standard Mutex does not allow that: the thread blocks forever or panics. A thread waiting forever for a lock is called a deadlock.

14.1.5 Operations on every shard

Some operations need every shard: the total size, and clearing the cache. They visit the shards one at a time.

Listing 14.4 len and clear (lines 96 to 119).

impl<K, V> ShardedCache<K, V>
where
    K: Hash + Eq,
{
    // ...
    pub fn len(&self) -> usize {
        self.shards
            .iter()
            .map(|shard| {
                shard
                    .lock()
                    .expect("shard mutex poisoned")
                    .len()
            })
            .sum()
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    pub fn clear(&self) {
        for shard in &self.shards {
            shard
                .lock()
                .expect("shard mutex poisoned")
                .clear();
        }
    }
    // ...
}

len locks each shard in turn, reads its size, releases the lock, and adds up the sizes. clear also visits the shards one at a time.

This is the price of sharding, and the comment at the top of the file states it. No moment exists when all shards are locked together. While len runs, other threads can insert into shards it has already counted. The total is correct if no other thread is writing, and approximate if one is.

14.1.6 The complete file

Listing 14.5 The complete file, with tests. src/problems/sharded_cache.rs

//! Sharded concurrent map to reduce lock contention.
//!
//! A single `Mutex<HashMap>` serializes every access. Splitting the map into N
//! shards, each with its own lock, lets independent keys proceed in parallel. The
//! cost is that `len` and `clear` must visit every shard, and there is no global
//! atomic snapshot.
//!
//! Typical uses: per-account token caches, metadata caches, and request
//! de-duplication tables.

use std::{
    collections::{HashMap, hash_map::RandomState},
    hash::{BuildHasher, Hash},
    sync::Mutex,
};

/// A fixed-size map split across independently locked shards.
///
/// The shard count is rounded up to a power of two so the index is a mask instead
/// of a modulo.
pub struct ShardedCache<K, V> {
    shards: Vec<Mutex<HashMap<K, V>>>,
    hasher: RandomState,
    mask: usize,
}

impl<K, V> ShardedCache<K, V>
where
    K: Hash + Eq,
{
    /// Create a cache with at least `shard_count` shards (rounded to a power of two).
    pub fn new(shard_count: usize) -> Self {
        let shard_count = shard_count.max(1).next_power_of_two();

        let shards = (0..shard_count)
            .map(|_| Mutex::new(HashMap::new()))
            .collect();
        Self {
            shards,
            hasher: RandomState::new(),
            mask: shard_count - 1,
        }
    }

    fn shard(&self, key: &K) -> &Mutex<HashMap<K, V>> {
        let index = (self.hasher.hash_one(key) as usize) & self.mask;
        &self.shards[index]
    }

    /// Insert a value, returning the previous one if the key was present.
    pub fn insert(&self, key: K, value: V) -> Option<V> {
        self.shard(&key)
            .lock()
            .expect("shard mutex poisoned")
            .insert(key, value)
    }

    /// Read a value by cloning it out of the shard.
    pub fn get(&self, key: &K) -> Option<V>
    where
        V: Clone,
    {
        self.shard(key)
            .lock()
            .expect("shard mutex poisoned")
            .get(key)
            .cloned()
    }

    /// Run `inspect` against the value under the shard lock, without cloning.
    ///
    /// Use this for read-modify-write on a single key; the closure holds the lock
    /// for its duration, so keep it short and do not call back into the cache.
    pub fn with<R>(&self, key: &K, inspect: impl FnOnce(Option<&V>) -> R) -> R {
        let guard = self
            .shard(key)
            .lock()
            .expect("shard mutex poisoned");
        inspect(guard.get(key))
    }

    pub fn remove(&self, key: &K) -> Option<V> {
        self.shard(key)
            .lock()
            .expect("shard mutex poisoned")
            .remove(key)
    }

    pub fn contains_key(&self, key: &K) -> bool {
        self.shard(key)
            .lock()
            .expect("shard mutex poisoned")
            .contains_key(key)
    }

    pub fn len(&self) -> usize {
        self.shards
            .iter()
            .map(|shard| {
                shard
                    .lock()
                    .expect("shard mutex poisoned")
                    .len()
            })
            .sum()
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    pub fn clear(&self) {
        for shard in &self.shards {
            shard
                .lock()
                .expect("shard mutex poisoned")
                .clear();
        }
    }

    pub fn shard_count(&self) -> usize {
        self.shards.len()
    }
}

#[cfg(test)]
mod tests {
    use std::{sync::Arc, thread};

    use super::*;

    #[test]
    fn insert_get_remove() {
        let cache: ShardedCache<String, u32> = ShardedCache::new(4);
        assert_eq!(cache.insert("server-1".into(), 8), None);
        assert_eq!(cache.get(&"server-1".into()), Some(8));
        assert_eq!(cache.insert("server-1".into(), 16), Some(8));
        assert_eq!(cache.get(&"server-1".into()), Some(16));
        assert_eq!(cache.remove(&"server-1".into()), Some(16));
        assert!(!cache.contains_key(&"server-1".into()));
        assert!(cache.is_empty());
    }

    #[test]
    fn with_exposes_the_value_without_cloning() {
        let cache: ShardedCache<&str, String> = ShardedCache::new(2);
        cache.insert("node", "ABCD".to_string());
        let observed = cache.with(&"node", |value| value.map(|v| v.len()));
        assert_eq!(observed, Some(4));
    }

    #[test]
    fn concurrent_writers_all_land() {
        let cache = Arc::new(ShardedCache::new(16));
        let mut handles = Vec::new();

        for writer in 0..8 {
            let cache = Arc::clone(&cache);
            handles.push(thread::spawn(move || {
                for key in 0..500 {
                    cache.insert(format!("{writer}-{key}"), key);
                }
            }));
        }
        for handle in handles {
            handle.join().expect("writer panicked");
        }

        assert_eq!(cache.len(), 8 * 500);
    }

    #[test]
    fn keys_spread_across_shards() {
        let cache: ShardedCache<u32, u32> = ShardedCache::new(8);
        for key in 0..4_096 {
            cache.insert(key, key);
        }
        // Every shard should hold roughly 1/8 of the keys; assert none is starved.
        let smallest = cache
            .shards
            .iter()
            .map(|shard| {
                shard
                    .lock()
                    .expect("shard mutex poisoned")
                    .len()
            })
            .min()
            .expect("at least one shard");
        assert!(smallest > 0, "at least one shard was empty");
    }
}

The test concurrent_writers_all_land is the one that uses threads. It wraps the cache in an Arc, and starts eight threads with thread::spawn. Each thread gets its own clone of the Arc and inserts 500 keys. join() waits for each thread to finish. The final len must be 8 × 500 = 4,000, which shows that no insert was lost.

The test keys_spread_across_shards inserts 4,096 keys into 8 shards and checks that no shard is empty. It reaches the private shards field directly. A test module inside the file can see private items of its parent module.

$ cargo test --lib problems::sharded_cache
running 4 tests
test problems::sharded_cache::tests::with_exposes_the_value_without_cloning ... ok
test problems::sharded_cache::tests::insert_get_remove ... ok
test problems::sharded_cache::tests::concurrent_writers_all_land ... ok
test problems::sharded_cache::tests::keys_spread_across_shards ... ok

test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 133 filtered out

14.2 Spreading keys across machines

Now move from threads to machines. Suppose a cache is spread over several servers, and each key must live on exactly one of them. Every client must compute the same server for the same key, with no central directory to ask.

14.2.1 The problem with hash % N

The obvious rule is hash(key) % N, where N is the number of servers. It spreads keys evenly and needs no storage. It breaks down when N changes.

Figure 14.3 shows keys with hashes 0 to 11 when a fourth server joins three others. A key stays in place only if hash % 3 equals hash % 4.

The pattern repeats every 12 hashes, so 75% of all keys change servers. For a cache, a moved key is a miss: the new server does not have it. Adding one server would empty most of the cache at once.

14.2.2 The ring

Consistent hashing places both servers and keys on a circle of hash values, called the ring. The ring covers every u64 value. After the largest value, it wraps around to 0.

Each server is hashed to a point on the ring. A key is hashed too, and it belongs to the first server point clockwise from it. In practice, “clockwise” means “the next larger hash”.

When a server joins, it takes over only the keys between its point and the point before it. Every other key keeps its server. When a server leaves, only its own keys move, to the next point clockwise.

With one point per server, the arcs between points are uneven, so some servers get much more of the ring than others. The fix is to place each server at many points, called virtual nodes. With many small arcs per server, the shares even out (figure 14.4).

Figure 14.5 shows the ring cut open and laid flat, before and after a third node joins.

Animation 14.2 follows the join through the whole system. It shows the ring, the keys each node holds, the key that moves, and the clients that learn about it.

Each node keeps the keys it owns. A join moves the keys in the new arc from the old owner to the new one. Every client also keeps a ConsistentHash of its own and computes the owner locally, so no lookup needs a central directory. When a node joins or leaves, each client has to learn the new ring. Until it does, it sends keys to the old owner. A client that still points k30 at node b asks a node that has given that key up.

With n servers, a new server takes over about 1/(n + 1) of the keys. Going from 3 servers to 4 moves about 25% of the keys, where hash % N moved 75%.

14.2.3 The ring as a sorted Vec

The ring of figure 14.4 can be stored as a vector of points, kept sorted by hash. Each point records its position on the ring and the server it belongs to.

Listing 14.6 The type and add_node (lines 17 to 55). src/problems/consistent_hash.rs

use std::{
    collections::{HashSet, hash_map::DefaultHasher},
    hash::{Hash, Hasher},
    sync::Arc,
};

/// A hash ring mapping keys to node names.
pub struct ConsistentHash {
    replicas: usize,
    /// Sorted `(hash, node)` pairs. The node name is an `Arc<str>` so that the
    /// many virtual nodes for one physical node share a single allocation.
    ring: Vec<(u64, Arc<str>)>,
}

impl ConsistentHash {
    /// Create an empty ring with `replicas` virtual nodes per physical node.
    /// Panics if `replicas` is zero.
    pub fn new(replicas: usize) -> Self {
        assert!(replicas > 0, "replicas must be > 0");
        Self {
            replicas,
            ring: Vec::new(),
        }
    }

    /// Add a node and its virtual nodes. Adding a node that already exists is a
    /// no-op.
    pub fn add_node(&mut self, node: &str) {
        if self
            .ring
            .iter()
            .any(|(_, name)| name.as_ref() == node)
        {
            return;
        }

        // One allocation for the name, then a refcount bump per virtual node.
        let name: Arc<str> = Arc::from(node);
        for replica in 0..self.replicas {
            let hash = virtual_node_hash(node, replica);
            self.ring
                .push((hash, Arc::clone(&name)));
        }
        self.ring.sort_by_key(|(hash, _)| *hash);
    }

    /// Remove a node and all of its virtual nodes.
    pub fn remove_node(&mut self, node: &str) {
        self.ring
            .retain(|(_, name)| name.as_ref() != node);
    }
    // ...
}

The ring is a Vec<(u64, Arc<str>)>, kept sorted by hash. Each entry is one virtual node: its position on the ring, and the name of the server it belongs to.

With 128 virtual nodes per server, each name would be stored 128 times as a String. The code stores it as Arc<str> instead. Arc::from(node) copies the name into one heap block once. Each Arc::clone(&name) then adds a pointer to that same block and increases its reference count. No text is copied.

add_node does three things:

1. It checks whether the server is already on the ring, by scanning the entries. If so, it returns.
2. It pushes one entry per virtual node, with the hash from virtual_node_hash.
3. It sorts the ring by hash again.

remove_node keeps every entry that belongs to another server, with retain. Removing entries does not change the order of the rest, so the ring stays sorted.

Listing 14.7 The hash functions (lines 94 to 109).

fn hash_of(input: &str) -> u64 {
    let mut hasher = DefaultHasher::new();
    input.hash(&mut hasher);
    hasher.finish()
}

/// Hash of the `replica`-th virtual node for `node`.
///
/// Hashing the two fields separately avoids allocating a `"{node}#{replica}"`
/// string for every virtual node on the ring.
fn virtual_node_hash(node: &str, replica: usize) -> u64 {
    let mut hasher = DefaultHasher::new();
    node.hash(&mut hasher);
    replica.hash(&mut hasher);
    hasher.finish()
}

virtual_node_hash feeds both the server name and the replica number into one hasher. Each replica number gives a different point for the same server. Feeding the two values separately avoids building a string such as "node-a#17" for every virtual node.

14.2.4 Finding a key’s server

To find a key’s server, hash the key and find the first point on the ring after that hash. If there is none, wrap around to the first point.

Listing 14.8 get (lines 64 to 77).

impl ConsistentHash {
    // ...
    /// Return the node responsible for `key`, or `None` when the ring is empty.
    pub fn get(&self, key: &str) -> Option<&str> {
        if self.ring.is_empty() {
            return None;
        }

        let hash = hash_of(key);
        // First virtual node strictly clockwise from the key.
        let index = self
            .ring
            .partition_point(|(point, _)| *point <= hash);
        let index = if index == self.ring.len() { 0 } else { index };
        Some(self.ring[index].1.as_ref())
    }
    // ...
}

get must find the first ring entry with a hash larger than the key’s hash. The ring is sorted, so a binary search can find it in O(log n) steps, as in section 3.10.

partition_point performs that binary search. It takes a condition that is true for a first part of the slice and false for the rest. It returns the index where the condition turns false (figure 14.6). Here the condition is point <= hash, so the result is the index of the first point larger than the key’s hash.

If the key’s hash is larger than every point, partition_point returns the length of the ring. The ring wraps around there, so the code uses index 0 instead.

get returns Option<&str>, a reference into the ring. .as_ref() turns the Arc<str> into a &str. An empty ring has no servers, so the function returns None for it.

14.2.5 The complete file

Listing 14.9 The complete file, with tests. src/problems/consistent_hash.rs

//! Consistent hashing ring.
//!
//! When a distributed gateway shards work across N backends, a plain
//! `hash(key) % N` remaps nearly every key the moment N changes. Consistent
//! hashing places each node at several points on a hash ring, so adding or
//! removing one node only moves the keys that node owns. That property is the
//! reason it shows up in caches, sharded queues, and load balancers.
//!
//! Each physical node is placed at `replicas` points (virtual nodes) to smooth the
//! key distribution; more replicas means a less lumpy ring.

use std::{
    collections::{HashSet, hash_map::DefaultHasher},
    hash::{Hash, Hasher},
    sync::Arc,
};

/// A hash ring mapping keys to node names.
pub struct ConsistentHash {
    replicas: usize,
    /// Sorted `(hash, node)` pairs. The node name is an `Arc<str>` so that the
    /// many virtual nodes for one physical node share a single allocation.
    ring: Vec<(u64, Arc<str>)>,
}

impl ConsistentHash {
    /// Create an empty ring with `replicas` virtual nodes per physical node.
    /// Panics if `replicas` is zero.
    pub fn new(replicas: usize) -> Self {
        assert!(replicas > 0, "replicas must be > 0");
        Self {
            replicas,
            ring: Vec::new(),
        }
    }

    /// Add a node and its virtual nodes. Adding a node that already exists is a
    /// no-op.
    pub fn add_node(&mut self, node: &str) {
        if self
            .ring
            .iter()
            .any(|(_, name)| name.as_ref() == node)
        {
            return;
        }

        // One allocation for the name, then a refcount bump per virtual node.
        let name: Arc<str> = Arc::from(node);
        for replica in 0..self.replicas {
            let hash = virtual_node_hash(node, replica);
            self.ring
                .push((hash, Arc::clone(&name)));
        }
        self.ring.sort_by_key(|(hash, _)| *hash);
    }

    /// Remove a node and all of its virtual nodes.
    pub fn remove_node(&mut self, node: &str) {
        self.ring
            .retain(|(_, name)| name.as_ref() != node);
    }

    /// Return the node responsible for `key`, or `None` when the ring is empty.
    pub fn get(&self, key: &str) -> Option<&str> {
        if self.ring.is_empty() {
            return None;
        }

        let hash = hash_of(key);
        // First virtual node strictly clockwise from the key.
        let index = self
            .ring
            .partition_point(|(point, _)| *point <= hash);
        let index = if index == self.ring.len() { 0 } else { index };
        Some(self.ring[index].1.as_ref())
    }

    /// Number of distinct physical nodes.
    pub fn node_count(&self) -> usize {
        self.ring
            .iter()
            .map(|(_, name)| name.as_ref())
            .collect::<HashSet<_>>()
            .len()
    }

    /// Number of virtual nodes on the ring.
    pub fn ring_len(&self) -> usize {
        self.ring.len()
    }
}

fn hash_of(input: &str) -> u64 {
    let mut hasher = DefaultHasher::new();
    input.hash(&mut hasher);
    hasher.finish()
}

/// Hash of the `replica`-th virtual node for `node`.
///
/// Hashing the two fields separately avoids allocating a `"{node}#{replica}"`
/// string for every virtual node on the ring.
fn virtual_node_hash(node: &str, replica: usize) -> u64 {
    let mut hasher = DefaultHasher::new();
    node.hash(&mut hasher);
    replica.hash(&mut hasher);
    hasher.finish()
}

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

    #[test]
    fn empty_ring_has_no_owner() {
        let ring = ConsistentHash::new(64);
        assert_eq!(ring.get("workload-1"), None);
        assert_eq!(ring.node_count(), 0);
    }

    #[test]
    fn lookup_is_stable_for_a_fixed_ring() {
        let mut ring = ConsistentHash::new(64);
        ring.add_node("node-a");
        ring.add_node("node-b");

        let first = ring.get("job-42").map(str::to_string);
        for _ in 0..100 {
            assert_eq!(ring.get("job-42").map(str::to_string), first);
        }
        assert!(first.is_some());
    }

    #[test]
    fn keys_spread_across_nodes() {
        let mut ring = ConsistentHash::new(128);
        for node in ["node-a", "node-b", "node-c"] {
            ring.add_node(node);
        }

        let mut counts = std::collections::HashMap::new();
        for key in 0..3_000 {
            let owner = ring
                .get(&format!("key-{key}"))
                .expect("ring is populated");
            *counts
                .entry(owner.to_string())
                .or_insert(0) += 1;
        }

        assert_eq!(counts.len(), 3, "every node should own some keys");
        let smallest = *counts
            .values()
            .min()
            .expect("non-empty");
        assert!(smallest > 300, "distribution is too lumpy: {counts:?}");
    }

    #[test]
    fn adding_a_node_only_moves_keys_to_the_new_node() {
        let keys: Vec<String> = (0..2_000)
            .map(|i| format!("workload-{i}"))
            .collect();

        let mut ring = ConsistentHash::new(128);
        ring.add_node("node-a");
        ring.add_node("node-b");
        let before: Vec<String> = keys
            .iter()
            .map(|key| {
                ring.get(key)
                    .expect("ring is populated")
                    .to_string()
            })
            .collect();

        ring.add_node("node-c");
        let after: Vec<String> = keys
            .iter()
            .map(|key| {
                ring.get(key)
                    .expect("ring is populated")
                    .to_string()
            })
            .collect();

        for (old, new) in before.iter().zip(after.iter()) {
            assert!(
                old == new || new == "node-c",
                "adding a node moved a key from {old} to {new}"
            );
        }
    }

    #[test]
    fn removing_a_node_drops_it_from_the_ring() {
        let mut ring = ConsistentHash::new(32);
        ring.add_node("node-a");
        ring.add_node("node-b");
        assert_eq!(ring.node_count(), 2);
        assert_eq!(ring.ring_len(), 64);

        ring.remove_node("node-a");
        assert_eq!(ring.node_count(), 1);
        for _ in 0..500 {
            assert!(matches!(ring.get("any-key"), Some("node-b")));
        }
    }
}

node_count counts distinct servers by collecting the names into a HashSet, which keeps one copy of each.

The tests check the properties this section described:

• keys_spread_across_nodes places 3,000 keys on three servers with 128 virtual nodes each. Each server must own more than 300 keys. An even split would give each one 1,000.
• adding_a_node_only_moves_keys_to_the_new_node records the owner of 2,000 keys, adds a third server, and looks again. Every key must either keep its owner or move to the new server.
• removing_a_node_drops_it_from_the_ring checks that 2 servers with 32 virtual nodes give 64 ring entries, and that removing one leaves every key with the other.

$ cargo test --lib problems::consistent_hash
running 5 tests
test problems::consistent_hash::tests::empty_ring_has_no_owner ... ok
test problems::consistent_hash::tests::lookup_is_stable_for_a_fixed_ring ... ok
test problems::consistent_hash::tests::removing_a_node_drops_it_from_the_ring ... ok
test problems::consistent_hash::tests::adding_a_node_only_moves_keys_to_the_new_node ... ok
test problems::consistent_hash::tests::keys_spread_across_nodes ... ok

test result: ok. 5 passed; 0 failed; 0 ignored; 0 measured; 132 filtered out

Chapter 15 looks below the data structures, at how the CPU caches memory and how the operating system hands it out.

Exercises

1. Add update(&self, key: &K, change: impl FnOnce(&mut V)) to ShardedCache, which changes a value in place under the lock.
2. Write a test that counts how many keys change owner when a fourth server joins a ring of three. Compare the result with 25%.
3. Measure how even the ring is. For 10, 50, and 200 virtual nodes per server, place 100,000 keys and report the largest and smallest share.
4. add_node sorts the whole ring after every call. Rewrite it to insert each new entry at its sorted position with partition_point and Vec::insert, and compare the two for 100 servers.