« Back to the main CS 0300 website

Project 6: Distributed Store

Due May 12, 2022 at 6:00pm EST
(Note on late hours: owing to the tight grading schedule, you can use at most 24 late hours on this project. Since TA staff have finals, there will be no project hours after the May 12 deadline.)

Introduction

In this assignment, you’ll implement a distributed, sharded key-value storage system that stores user data and post data for a new social media platform called Bacefook. Many major websites like Twitter, AirBnB, and Facebook have so many users that no single server can store all their data, or handle all the requests for data. Instead, they carefully split the data (and the requests accessing it) across many servers. This idea is called sharding.

Outside of a programming context, think about how large events or concerts might speed up their check-in process by having separate tables to check-in attendees with names starting with A-K and L-Z. In systems programming, this idea is called “sharding”, as it splits the overall key space (e.g., A-Z) into many shards that can each be stored and managed on different servers.

Key-Value Stores

A key-value store is a storage system in which data (a value) is addressed by a key. Similar to a Python dictionary or a Java/C++ hashmap, the key maps directly to the value. For example, the key cs300_students might map to a list of student CS logins ([login1, login2, login3]). Note that this is unlike, say, a filesystem, in which files are arranged in a hierarchical directory structure, and you would need to navigate a path like /courses/cs300/students to get to the data. Key-value stores are popular for large-scale web applications because they are particularly amenable to sharding and other techniques for high performance.

The WeensyDB we saw in lectures is a key-value store. Examples of key-value stores used in industry include memcached, Redis, RocksDB, and LevelDB.

Sharding

A shard is a subset of the key/value pairs stored in the overall key-value store. For example, if our key space is the strings[AA, ZZ], the keys in the range[AA, AZ] might be one shard, the keys [BA, ZZ] another shard, etc.

When thinking about sharding, it’s useful to understand the difference between vertical scaling and horizontal scaling. Vertical scaling is the act of scaling up a service or system by adding more computing power (CPU, RAM, etc.) to an existing machine. You might, for example then run more threads to handle additional requests, like you did in the Vunmo project. Horizontal scaling, by contrast, is scaling by adding more machines. While vertical scaling seems simpler at first, it’s easy to run into hardware limits (at some point you can’t have more processors on a machine). In contrast, in a world where cloud providers like AWS, Azure, and Google Compute Engine sell you virtual machines on demand, you can get a practically unlimited amount of compute power, memory, and storage space through vertical scaling.

In contrast to multithreading (a vertical scaling technique), sharding is a horizontal scaling technique. By having multiple key-value servers, each handling operations for just a few of the shards, we increase our system throughput (operations per unit time) by just adding more key-value servers. Naturally this isn’t free – having all these servers work together requires coordination to make sure clients contact the right server. Moreover, we would like to be able to add and remove servers as needed without having to restart the entire system. We’ll build toward such a dynamic system in this assignment!

Conceptual Questions

• Write your answers to the following questions in your README.
• We strongly recommend that you answer these questions as you complete the project, rather than waiting until after finishing the project code. Answering the questions will help you immensely in understanding the concepts behind the code that you’ll be writing.
• The due date for the conceptual questions is the same as the project due date.

1. Describe the roles of the controller, the key-value servers, and the clients in this distributed system. How are they involved in processing a client request for key k?
2. Currently, the clients contact the shardmaster to get the current configuration of shards and then make RPCs to key-value servers. Clients will only re-query the shardmaster if a request to a server fails because the shard containing they key was moved away from that server. Instead of this model, the shardmaster could also be in charge of making requests to the key-value servers on clients’ behalf and pass the responses back to the clients. Why will this design result in lower system throughput?

Assignment

Assignment installation

First, update your development environment. Inside your DEV-ENVIRONMENT folder (where you cloned the cs300-s22-devenv repository), type:

$ git pull

Recreate your Docker container by running the cs300-run-docker script with the --clean flag, like so:

$ ./cs300-run-docker --clean

You only need to do this once – in future runs, you do not need the --clean flag. The reason this is required is that we’ve updated the cs300-run-docker script to include additional configuration for exposing ports that the Bacefook UI containers need to access.

Next, as usual, ensure that your project repository has a handout remote. Inside your project repository, type:

$ git remote show handout

If this reports an error, run:

$ git remote add handout https://github.com/csci0300/cs300-s22-projects.git

Then run:

$ git pull
$ git pull handout master

This will merge our Project 6 stencil code with your repository.

Once you have a local working copy of the repository that is up to date with our stencils, you are good to proceed. You’ll be doing all your work for this project inside the distributed directory in the working copy of your projects repository.

Infrastructure Help

Stencil & Initial State

Switch to the distributed directory (insider your new Docker container), and run ./install.sh. This will install all required dependencies, so it might also take a while.

Stencil Breakdown

How to Build/Test

cd into the build directory and make to build your code.

To run the tests, run either make check or ./test.sh from inside the build directory.

The tests are in the tests directory, with each subdirectory corresponding to a part of the roadmap. After implementing part 1, you should pass the simple_shardkv tests. After part 2, the shardmaster_tests should pass and after part 3, theshardkv_tests and integrated_tests should pass.

If you want to write your own tests just add new files to the subdirectories in tests. Be sure to take a look at the functions in test_utils/test_utils.h – they will make writing your own tests much easier. Note that if you write your own tests you will need to add them to the Makefile (see lines 139-184 in the Makefile for examples) before they will be built by make.

Assignment Roadmap

1. A simple, generic key-value server

Overview

Your first task is to implement a simple, generic key-value server, called simple_shardkv that is not specific to BaceFook. You can think of this component as the code that each shard in the system needs to run. Clients will use Remote Procedure Calls (RPCs) to interact with your server. RPCs are a way of making function or method calls between two processes (even if they’re on different machines!), typically without the programmer having to specify exactly how that communication takes place. Your server should support the following RPCs:

Note: In this project, the keys and the values are strings.

We will use gRPC along with Protobuf to help us set up the RPCs. Protobufs (protocol buffers) are a convenient way to specify message formats, and we’ll use them to define our RPC services. gRPC lets us take a .proto file and generate a lot of the boilerplate code we’d need to otherwise write to implement the RPCs. Take a look at protos/shardkv.proto (the protobuf file) and compare it tobuild/shardkv.grpc.pb.{h, cc} and build/shardkv.pb.{h, cc} (the generated files, can be built by running make from the build directory) if you want to see exactly how much code gets generated. (Check out Lab 9 for more information about Protobufs and gRPC.)

Specification:

Take a look at simple_shardkv/main.cc first. While you shouldn’t modify this file, it’s useful to understand what’s going on. We read in the specified port, construct an instance of the SimpleShardkvServer class, and then register that instance as a listener on the specified port. Finally, we call server->Wait() which actually starts listening for RPCs.

Testing it out with a Client

Each simple_shardkv server is standalone, and only knows about the requests it receives. This is already sufficient to run a simple distributed key-value store: you can run two simple_shardkv servers and arrange for the clients to send all requests for the first half of the keyspace to the first server, and all requests for the second half of the key space to the second server.

We provide you with a client (clients/simple_client) that implements its own, hard-coded sharding scheme to interact with several simple_shardkv servers.

The client will construct an RPC “stub” on its end. A stub is dummy object that provides the interface specified by Shardkv::Service. When the client makes method calls on the stub, your SimpleShardkvServer methods will be invoked, even if the client and your server run on different machines!

The simple_client uses a scheme that shards keys by using id % n where n is the number of servers to chose the server for a key and id is the “id” field extracted from the key. For example, if you connected the client to two simple_shardkv servers, the client will ensure that the Put, Get, and Append RPCs on even keys are sent to the first server and the RPCs on odd keys are sent to the second server. You tell the client on startup how many servers there are and at what address and port they listen.

$ ./simple_shardkv <PORT>

$ ./simple_client <SIMPLE_SHARDKV_SERVER1> <SIMPLE_SHARDKV_SERVER2> ...  

$ ./simple_shardkv 13101

$ ./simple_shardkv 13102

$ ./simple_client `hostname`:13101 `hostname`:13102
put post_10 "golden_snitch" user_7
get post_10
Get returned: "golden_snitch"

While this scheme does distribute load across servers, it has a few problems. One problem is that we have no capacity to add new key-value servers or deal with existing ones leaving, and that all server addresses and ports need to be known when starting the client. Don’t fret, though! You’ll write a better shard scheme in part 3!

2. Shardmaster v1

Overview:

So how can we handle updating clients about key-value servers joining or leaving? To do so, you’ll implement a controller program called “Shardmaster”, which is responsible for assigning shards to servers. The design of this program is inspired by the Slicer system at Google, which works the same way. Clients can now query your shardmaster to find out which server is responsible for the key(s) they are interested in, before directly querying those servers. See the shardmaster directory for the stencil, and the common directory for useful utilities.

Specifications:

Your shardmaster will organize data by partitioning the integer ID range [MIN_KEY, MAX_KEY] (defined in common.h).

Here’s a table of the RPCs your shardmaster should support (take a look at protos/shardmaster.proto to see the message specifications):

Administrators can use the Join/Leave RPCs to add/remove servers. Both of these operations should cause the shardmaster to adjust how it has distributed shards – either giving the new server shards from an existing server, or redistributing the shards from a leaving server. Together with the Query RPC, this makes it possible for clients to adapt to a changing set of key-value servers in the system, as the shardmaster can always tell clients what the most up-to-date configuration is.

The Move RPC transfers a shard between two servers and shouldn’t cause any other configuration changes. We will use it to test your shardmaster.

Finally, this new API also includes a GDPRDelete RPC that should remove all information related to a user from the key-value store. You can implement this for extra credit in part 3.

Note that in this part of the project, you are only expected to adapt the configuration data on the shardmaster. In Part 3, you will actually move keys between key-value servers.

Shard redistribution for Join and Leave

We’ve alluded to this in our explanation of the Join and Leave RPCs, but how exactly should redistributing shards work? You’ll have more freedom to explore this in subsequent parts, but for now Join and Leave should redistribute keys using the following approach.

Maintain a list of key-value servers on the shardmaster and regardless of the current configuration, assign them each an equal proportion of the keyspace, with servers that joined earlier getting lower keys. If the keys don’t evenly distribute, we distribute the remaining across the servers who joined first. Below is an example of how that works.

While this key distributed scheme works, it’s not always ideal. You can implement an improved schema as an extra credit exercise later!

Testing it out with a Client

The new client you’ll use to test this part is clients/client. The client can invoke all the RPCs specified in the above table on your shardmaster (run help for the commands you need), as well as Get/Put/Append.

This client initially queries the shardmaster for its current configuration of how shards are distributed among servers, then directly makes each RPC call on whatever server the shardmaster said has the key in question. Remember, if the configuration of shards has changed you may need to query the shardmaster for the updated configuration to find keys!

$ ./shardmaster <PORT>

$ ./client <SHARDMASTER_HOST> <SHARDMASTER_PORT> ... 

$ ./shardmaster 13100

$ ./simple_shardkv 13101

$ ./client `hostname` 13100
join fake_server:1234
join 29f0c19d87b:13101
query
Shard {0, 500} on server fake_server:1234
Shard {501, 1000} on server 29f0c19d87b:13101
put user_502 Bob
Put server: 29f0c19d87b:13101
put user_500 Alice
Put server: fake_server:1234
method Put failed with status code 14
the error message was: DNS resolution failed

You can provide fake addresses in your Join/Leave commands – the client doesn’t actually verify if key-value servers are running on the addresses until you try to Get/Put/Append. This means you can test your id redistribution without updating your key-value servers (the next part!).

3. A more complex key-value server

Introducing Bacefook

In this part of the assignment, you’ll write a specialized and more complicated key-value server for the new social media platform, Bacefook. On Bacefook, users can choose their names and create posts on their timelines. Surprisingly, these timelines are very similar to “walls” on Facebook! Even though there can be multiple users with the same user names, every user_id is unique. The same goes for posts: even though there can be multiple posts with the same text, every post has a unique post_id.

Specializing your key-value store

In Bacefook, there are four kinds of key-value pairs.

You’ll now write the key-value server called shardkv (see the shardkv directory for the stencil). The logic for handling Get/Put/Append requests should be quite similar to simpleshardkv; however, to specialize it to BaceFook, you should make two major changes:

1. As you may recall, the put RPC includes a user field that stores the user’s ID. This is used in the case the put is for a post. The shardkv server should handle appending the new post ID to the user_id_posts key. The key might not be on this server!
2. We want to be able to query all of the users currently in the system so users can make new friends! Automatically add a new all_users key on each server that keeps track of the user IDs of every user being stored on that server. Think about what other functions you’ll need to modify to keep this list up to date!

Note: You may notice that even though the shardkv server appends the post_id to user_id_posts on the post’s Put RPC, there isn’t a similarly efficient mechanism to delete that post ID from the list on a Delete RPC for the post. In BaceFook and many similar social media sites, deletes are much less common than puts, so it is acceptable to eat the cost of manually retrieving and updating user_id_posts on the client side.

Applying configuration changes to the key-value servers

So far, the shardmaster sounds nice, but it doesn’t actually move keys between key-value servers. In other words, we haven’t touched on how the key-value servers handle configuration changes. If a subset of ids on server1 gets moved to server2, somehow we need to send the key-value pairs stored previously on server1 to their new home!

Your key-value store should now:

• Periodically query the shardmaster for the current configuration of shards. If your server sees that shards it previously stored are now on another server, it should use the Put RPC to transfer the appropriate key-value pairs and cease to serve the shards after the transfer completes.
• Only respond to queries for keys that it handles as per its last contact with the shardmaster, and return an error for all other keys.

When transferring data from one shardkv server to another via Put RPCs, it’s important to retry until success (grpc::Status::OK) – we don’t want key-value pairs to be lost! The diagram in the walkthrough below shows an example of why this is important.

Implementation and Testing

Note: QueryShardmaster is an RPC that the ShardkvServer periodically calls on itself, but other servers could also invoke this RPC remotely to make a ShardkvServer update its configuration!

$ ./shardkv <PORT> <SHARDMASTER_HOSTNAME> <SHARDMASTER_PORT>

The Bacefook UI

To truly see BaceFook come to life and test your implementation, we’ve provided a GUI where you can test out your key-value store! Using the GUI is entirely optional and we will only be running the tests during grading.

There are two components to the GUI: a frontend written using React (in the frontend directory) and a frontend server written using Flask (in the frontend-server). Don’t worry if you aren’t familiar with these technologies – all you need to know is that the frontend server makes RPC requests to the backend, consisting of your shardmaster and shardkv servers, and provides the data to the frontend UI, which is running client-side in a user’s browser.

$ docker compose up

$ ./shardmaster 9095

$ ./shardkv 13101 `hostname` 9095

$ ./shardkv 13102 `hostname` 9095

$ ./client `hostname` 9095
join 29f0c19d87b:13101
join 29f0c19d87b:13102
query
Shard {0, 500} on server 29f0c19d87b:13101
Shard {501, 1000} on server 29f0c19d87b:13102

$ docker compose up

Extra Credit

Here are some extra credit ideas that you can explore to make your Bacefook social network more awesome! For this project, graduate students taking CSCI 1310 do not need to complete extra credit work.

GDPR

The General Data Protection Regulation (GDPR) is an EU law implemented in 2018 that protects the security and privacy of user data. This law affects any corporation that collects data from people in the European Union. These new regulations require all personal data to be protected by companies, from names and addresses to web data like IP addresses and cookie data. In order to protect personal data, companies must design information systems with privacy in mind, such as anonymizing any identifiable information and giving users the right to request the erasure of their data.

Although Bacefook is based in the United States, many of its users live in the EU. As a result, you must ensure that your storage system complies with GDPR regulations.

For extra credit, you can implement shardmaster’s GDPRDelete and remove all information related to a user from the key-value store. Think about what information that includes, and how you can track it so that GDPRDelete can efficiently obtain that information and remove it. This should include changes to both the StaticShardmaster and ShardkvServer classes. After implementing this, you should pass all of the provided tests!

Implementing a Dynamic Shardmaster

You have a fully functional distributed system capable of reconfiguration to move shards between servers now. But it still implements the simple scheme that partitions the key space equally between your shardkv servers.

In the real world, like in Bacefook, keys (and therefore shards) are not equally popular. For example, celebrities on Twitter or Instagram have many more followers than most other users. If a shard stores a celebrity’s timeline, it will receive many more requests than others, which could overload that shard!

Think about how you could adapt your shardmaster to implement a smarter redistribution scheme that takes into account how busy the shards are.

Then, design and implement a scheme for a dynamic shardmaster that reacts to load. Your design will likely involve adding new state and messages to the shardmaster and shardkv servers. You’ll need to provide proof of your improvements (for example, subjecting your new system and your old one to similar loads and collecting statistics on how frequently each server is contacted) to show how you’ve improved system throughput.

There is no single right answer here; we will award credit for any reasonable scheme.

Debugging

All the debugging techniques you’ve learned throughout the course apply to this project too. Finding bugs in distributed systems can be challenging, but we are sure you can do it!

Troubleshooting failure messages:

Here are some common error outputs from the test suite, and hints for how do deal with them.

• STDERR: {test_name}: {path_to_test}:{line_number}: {function}: Assertion {test_function} failed
  • This type of error means that your implementation runs without failure. However, your implementation’s behavior does not match the expected output. GDB will be the most helpful tool for debugging this kind of failure.
  • The executables that correspond to all tests can be found in the build directory and can be run in GDB with the command gdb {test_name}.
  • The best place to start is to set a breakpoint at the test function you’re failing (for example test_query) and then step through this function to see where your implementation’s behavior differs from what is expected.
• ./test.sh: line 69: Segmentation fault (core dumped) "$EXEC" > "$TMP_STDOUT" 2> "$TMP_STDERR"
  • This indicates that a segfault occured during the execution of the test, so you likely access invalid memory somewhere.
  • You can use the Address Sanitizer to get more information about the segfault by running make SAN=1.
  • GDB is again going to help you debug this. You can run the failing test in GDB with the command gdb {test_name} from the build directory to run a specific test in GDB.
  • For a segfault, the best place to start is to just let the program run in GDB without any breakpoints. Once the segfault is reached, do a backtrace and look for the last frame in the trace that corresponds with code that you were responsible for writing. This function will be the best place to start looking in terms of tracking down the source of the fault!

Hanging Tests

If a specific test remains running for more than about a minute, this is a sign that your implementation contains a deadlock. For this, GDB’s multi-threaded debugging tools will be useful. Again, you can run the particular test in GDB with the command gdb {test_name} from the build directory.

After letting the program run in GDB for a bit, you can use Ctrl-C to pause the execution of all threads. After doing this, you can use the command info threads to see a list of all of the running threads and where they are in execution.

If it seems like a lot of threads are waiting on a mutex (their function is listed as _lock_wait), you can select one of these threads (with the command thread <t_id>), and look at its backtrace. If you look at the frame just before the frame for the lock() function, you should be able to see which of your mutexes it is waiting for. You can get information about this mutex with the command p <mutex>. Some of the more useful information that this shows you is the ID of the thread that currently holds the mutex!

Handing In & Grading

Handin instructions

As before, you will hand in your code using Git. In the sharding subdirectory of your project repository, you MUST fill in the text file called README.md.

Grading breakdown

• 93% (84 points) for passing the functional tests in categories simple_shardkv (15 points), shardmaster (25 points), shardkv (10 points), and integration (34 points). If your tests pass on the grading server, you’ve probably got all these points.
• 7% (6 points) answers to conceptual questions.
• Extra Credit: up to 15 points for implementing GDPRDelete and coming up with your own dynamic shard scheme + proving that your scheme is more efficient.

Now head to the grading server. Make sure that you have the “Distributed Store” page configured correctly with your project repository, and check that your tests pass on the grading server as expected.

Congrats! You’re officially done with the last project for the class

Acknowledgements: This project was developed for CS 300.