Case study: Distributed cache and key-value store

"Design a key-value store" and "Design a distributed cache" are common questions. In both, a program saves a value under a key and reads it back later. The data is too big, or too busy, for one server. The interviewer then asks: "A server dies. What happens?"

A weak answer says "use Redis" or "hash the key to a server". A strong answer works out the number of servers from numbers. It says where each key lives, how many copies exist, and how many copies a write and a read must reach. It also says how one server saves data fast, and what changes when the system is only a cache.

At SDE-2 level you should place the keys and choose the copies. At SDE-3 level you should also explain what two servers failing at once does, and how a delete works.

A wedding hall has 15 cloakroom counters. Each coat has a ticket number. A rule decides which three counters keep the coat for a ticket. When the guest comes back, the staff ask two of those three counters and trust the newer tag. If one counter closes, the guest still gets the coat.

That is the whole design. Below, each part uses an idea from an earlier lesson.

What we build

A key-value store (lesson 17) has three calls: put(key, value), get(key) and delete(key). We assume a value of about 1 KB. We do not build search by value, joins or changes to many keys at once. A request must finish in a few milliseconds, and one broken server must not stop requests.

A store is the place where the data lives. If it loses data, the data is gone. A cache (lesson 31) holds copies of data that lives somewhere else, so it may lose entries. The two share most of the design. We build the store first and then list what a cache changes.

Part 1: find the servers of a key

Use the ring of lesson 44. Servers get many positions (virtual nodes) on the ring. A key goes to the first server at or after its position. With N copies, the key is also kept on the next N - 1 different servers on the ring. These N servers are the owners of the key. We use N = 3.

Every server keeps the whole ring table. So the client can send a request to any server. That server becomes the coordinator of the request (lesson 48). It finds the three owners and talks to them. Another choice is a client library that keeps the ring and goes to an owner directly.

How does a server know that another server is down? Each server sends a short "I am alive" message to a few random servers every second. A server that stays silent for a set time is marked down. A short time marks slow servers as down by mistake. A long time sends requests to dead servers. Each such request waits for the timeout, the longest time a call may wait, before it fails.

Part 2: copies and quorums

Use lesson 48 with N = 3, W = 2 and R = 2. For a write, the coordinator sends the value to all three owners. It tells the client "saved" when two owners confirm. It gives each write a version: its own clock time. For a read, it asks two owners, waits for both answers, and returns the newest version.

Because 2 + 2 is more than 3, a read meets at least one owner that has the newest write.

A server was down. The write still succeeds on two owners. The third owner is behind. When a later read sees two different versions, the coordinator sends the newest one to the old owner. This is read repair. A background job also compares copies and fixes the differences, for keys that nobody reads.

Two clients write one key together. The newest version wins and the other write is lost. Clocks can differ between servers (lesson 49), so the wrong write can win. This design accepts that. If a key cannot accept it, send all writes of that key through one owner (lesson 39). That costs availability when this owner is down.

Part 3: how one server stores data

A server must write fast and must not lose data if it crashes. It does not change files in place. This way of storing data is the log-structured merge tree (LSM tree). Lesson 46 explains it in full. Here is the short version.

1. Write. The server adds the write at the end of the commit log, a file that only grows. After a crash, the server replays the log. The server also puts the write in the memtable, a sorted list in memory.
2. Flush. When the memtable is full, for example 64 MB, the server writes it to disk as one sorted file that is never changed. This file is an SSTable. A new memtable starts.
3. Delete. Files do not change, so a delete writes a tombstone: a marker "this key is deleted", with a version.
4. Read. The server looks in the memtable, then in the files from newest to oldest. The newest version found wins.
5. Too many files. Each flush adds a file, so reads check more files. Two things help. Compaction merges files into one, keeps the newest version of each key, and drops old versions. A Bloom filter for each file is a small table in memory. It says "this key is definitely not in this file" or "maybe". It is never wrong when it says "definitely not". With 10 bits per key and 7 hash functions, a missing key gets a wrong "maybe" about 0.82% of the time.

A tombstone cannot be dropped at once. A server that was down during the delete still holds the old value. If the tombstone is gone, repair copies the old value back, and the deleted key returns. So tombstones stay for a grace period, a set time of some days. In Cassandra the default is 10 days.

Compare lesson 20. A B-tree changes pages in place, so a write goes to a random place on disk. An LSM tree appends, so writes are fast. It pays later: reads check several files, and compaction rewrites data in the background.

What a cache changes

1. No disk. Entries live in memory only. There is no commit log and no file. A lost server means lost entries.
2. Another source of truth. A database holds the real data (lesson 32). A lost entry is a miss, not a loss.
3. Eviction and TTL. The memory is full at some point. The server removes entries by a rule (lesson 33) and by TTL (lesson 31). Size it for the working set.
4. Fewer copies. Usually N = 1, with no quorum. A copy costs memory and only helps when a lost server would overload the database.

When a cache server dies, its keys miss together. If many of them are hot keys, you get a stampede (lesson 35).

SwadKart keeps one kind of data in the store. We find the number of servers, then test failures.

Assumptions (made up to teach with, not real traffic):

• 500,000,000 keys (50 crore). One key with its value and extra bytes takes 1 KB.
• N = 3, W = 2, R = 2. A read asks 2 owners.
• Peak: 100,000 requests per second. 90% are reads (90,000) and 10% are writes (10,000).
• One server does 20,000 requests per second. We plan for 70%: 14,000.
• A server has a 2 TB disk. We fill it to 50% (1,000 GB), because compaction needs spare room. This is our choice.

Step 1: data. 500,000,000 x 1 KB = 500 GB. With 3 copies, 1,500 GB. By disk alone, 1,500 / 1,000 = 1.5, so 2 servers.

Step 2: load. A write reaches 3 servers: 10,000 x 3 = 30,000. A read reaches 2: 90,000 x 2 = 180,000. Total: 210,000 server requests per second. 210,000 / 14,000 = 15 servers. Each holds 1,500 / 15 = 100 GB. Load decides the number, not disk.

Step 3: one server breaks. Each key has 3 owners among 15 servers. A server is one of the 3 owners for 3 / 15 = 20% of the keys. A program placed 100,000 keys on a ring of 15 servers with 200 virtual nodes each (as in lesson 44). Each server was an owner of between 18.3% and 21.4% of the keys. This answers the predict box: about 20%. Those keys still have 2 live owners, so W = 2 and R = 2 work. No request fails. The 14 servers share about 210,000 / 14 = 15,000 requests each. That is 75% of 20,000, above our 70% plan, but they survive.

Step 4: two servers break together. A key fails when two of its owners are the two broken servers. There are 15 x 14 x 13 / 6 = 455 ways to choose 3 owners from 15 servers. In 13 of these 455 sets, both broken servers are included. The third owner can be any of the other 13 servers. So 13 / 455 = 2.86% of the keys have one live owner. For them, W = 2 and R = 2 fail. The program gave 2.86% on average, between 1.5% and 4.2% for different pairs. So a second failure hurts a small share of keys, not all and not none.

Step 5: slow copies. The three owners answer in 1 ms, 1.2 ms and 15 ms. The third is busy with compaction. With W = 2, a write waits for the second answer: 1.2 ms. With W = 3 it would wait 15 ms.

Step 6: the node and its files. A server gets 30,000 / 15 = 2,000 writes per second, which is 2 MB per second. A 64 MB memtable fills in 64 / 2 = 32 seconds. Without compaction, an hour gives 3,600 / 32 = 112 files. A read for a key that is in none of the files would check all 112 files. With Bloom filters, the expected checks are 112 x 0.0082 = 0.92. Assumption: compaction brings the files down to 10. Then 10 x 0.0082 = 0.08. A server holds 100 GB / 1 KB = 100,000,000 keys. The filters need 100,000,000 x 10 bits = 125 MB of memory.

Step 7: add a 16th server. It takes 1 / 16 of the copies: 1,500 / 16 = 94 GB. Assumption: copying is limited to 50 MB per second, so live traffic is not hurt. 94 GB (rounded) is 94,000 MB. Then 94,000 / 50 = 1,880 seconds, about 31 minutes. Hash mod (lesson 43) would move 15 / 16 of 1,500 GB, which is 1,406 GB. That takes 28,100 seconds, about 7.8 hours.

Step 8: the cache. Assumptions: 100,000 reads per second, hit rate 95%, so 5,000 reads per second reach the database, which can do 12,000. The cache has 10 servers. One dies. Its 10% of the keys lose their entries. The hits that were served by it, 0.95 x 100,000 x 10% = 9,500 per second, become misses. The database gets 5,000 + 9,500 = 14,500 per second, which is 121% of its limit. With 20 servers, the extra is 4,750, and the total is 9,750 (81%). With hash mod over the 9 left servers, 9 / 10 of the keys change server. The misses are 5,000 + 0.95 x 100,000 x 0.9 = 90,500 per second. This lasts only until each key has missed once and is cached again.

• Three copies with W = 2 and R = 2 solve one failed server and slow copies. They cost three times the disk and a request to several servers for each call. I would change to R = 1 and W = 3 if reads are far more than writes. I would use a leader for each key if lost writes are not acceptable.
• An LSM tree solves fast writes. It costs reads that check several files, and background rewrites. I would change to a B-tree store if reads dominate and writes are few.
• A cache solves a slow database for popular keys. It costs a miss spike when a server dies. I would use more small servers, or two copies of hot keys, when the database cannot take that spike.

"A distributed cache is a key-value store with a different name." A store must not lose data. A cache may. The cache needs no disk, no log and usually no copies. In Step 8, a lost server only caused misses.

"With three copies, losing two servers is the same as losing one." One failed server hurt no key. Two failed servers made 2.86% of the keys fail for writes and reads.

"A delete removes the value from disk at once." Files do not change. A delete adds a tombstone. The value goes away at a later compaction, after the grace period.

1. List the steps a server takes for a write in an LSM tree. Why does a delete write a tombstone?
2. A store has 200,000,000 keys of 2 KB, with N = 3, W = 2 and R = 2. Peak is 50,000 requests per second, 80% reads. One server does 10,000 requests per second, and you plan for 70%. How many servers do you need for the load? How much data does each hold?
3. A cache has 8 servers, 100,000 reads per second and a hit rate of 90%. The database can do 20,000 reads per second. One cache server dies. Is the database safe? What do you change?