This is HO NGOK CHAO's Readme file.
This project has two parts.
Part 1 is a basic gRPC file sharing service with 4 synchronous Remote Procedure Calls (rpc) fetch, store, list, and Stat.
sequenceDiagram
Client->>Server: Fetch a file request(unary)
Server-->>Client: Transfer file to client(stream)
Client->>Server: Store a file request (stream data)
Server-->>Client: Empty response (unary)
Client->>Server: List all file request (unary)
Server-->>Client: List of files (stream)
Client->>Server: Get attributes of a file (unary)
Server-->>Client: mtimes,ctime,size (unary)
part 2 utilize above 4 fundamental methods plus RequestWriteLock (synchronous) which would be used by Store and Delete (synchronous) so that only one client can change the state of the server at a time, and a asynchronous rpc CallBackList which client would use to asynchronously request/receive about the attributes (including modification time) of files in the server so that multiple clients' files can stay sync with those of server via calling fetch/store/delete respectively according existence/mtime. This constitutes a Distributed File System (DFS) and is achieved by using multiple threads on both clients and server.
stateDiagram
Server
Client1 --> Server
Client2 --> Server
Client3 --> Server
Note left of Client1: Store
Note Left of Client2: Fetch
Note Left of Client3: Fetch
Almost all are hard requirements in English translated to Protocol Buffer (Interface Definition Language) then protocol buffer compiler generates the client/server interfaces in C++ for service DFSService.
Trade-offs explained:
// unary to stream
rpc Fetch(FileRequest) returns (stream FetchResponse);
message FileRequest {
string name = 1;
}
message FetchResponse {
bytes chunk_data = 2;
}- Fetch: This rpc is for client to fetch a file from the server, naturally the input is the filename that it wants which server would resolve to a full server side path, the response is a stream of bytes of the file since the file can be of arbitrary size (the alternative is a Unary response with repeated field inside a single message, but it is unnecessary and make implementation harder).
- On the server side, we read to main memory and send 1000 bytes (reasonably small to make sure we have enough memory) each to client.
- On the client side, it just open a file initially and keep appending to it until the stream is closed.
// stream to unary
rpc Store(stream StoreRequest) returns (google.protobuf.Empty) {};
message StoreRequest {
string name = 1;
bytes chunk_data = 2;
}- Store: This rpc is for client to store a file to the server, again naturally the input of the rpc is the file name and a stream of bytes of the file (with similar reason as fetch).
- On the client side,it read from a file and keep putting sequence of messages into the stream, so every message has both name and chunk_data.
- On the server side, it read one message from the stream every time, and every time open a file with name and append the chunk_data received (for first message of the stream, I would clear all the content inside it first). This design is a bit awkward but since server side doesn't know when the stream would end, it has to open and close a file multiple times. One alternative is to include file size into the message, so that server can create a file on the first message and only close it after receiving enough data (or timeout), but that increase the implementation complexity hence I didn't do it this time.
// unary to unary
rpc Delete(FileRequest) returns (google.protobuf.Empty);- Delete: This rpc is for client to delete a file in the server. Not much design on server side delete if found and return grpc::StatusCode::OK else grpc::StatusCode::NotFound
// unary to stream
rpc List(google.protobuf.Empty) returns (FileList);
message FileList {
map<string,fixed64> mtime =1;
}- List: This rpc is for client to retrieve a map from server for all its files whose key is the filename of a file in server and value is the mtime of the file.
- On the client side, we call the rpc and simply convert grpc::protobuff:map into std::map of C++, the alternative is to use other type like a vector of pair as a response from server and convert that to std::map but that would be a detour.
- On the server side, retrieve the file list using system call opendir() and stat to retrieve the list and attributes including mtimes; the alternative is to use Filesystem library since C++17 but this project restrict the C++ version to be C++14 and the <experimental/filesystem> in C++14 is possibly not stable enough hence I stick with the C style. Othewise, I could have used std::experimental::filesystem::last_write_time
// unary to unary
rpc Stat(FileRequest) returns (Attributes);
message Attributes {
uint64 size = 1;
int64 mtime = 2;
int64 ctime = 3;
}- Stat: This rpc is for client to retrieve Stat of a file in server. The only decision made is to use int64 as the type in protobuf; the alternative is to use google::protobuf::timestamp but I feel like calling timestamp.set_seconds to set the value and timestamp.seconds() to compare two timestamp too troublesome for this project since I am going to always compare in seconds.
First, let's talk about design for DFSService in this part. It is an extension to part1 plus two rpc RequestWriteLock and CallBackList which are necessary for this Distributes File System where multiple clients and one server exists.
Trade-offs explained:
For part1's rpc:
- Store: On the client side, to prevent multiple clients try to store the same file to the server at the same time. It would have to call RequestWriteLock first and only seeing grpc::StatusCode::OK it would continue to make the Store rpc to the server. Need to sync mtimes on both server and the client calling since that would impact all clients' decisions calling fetch. It also need to submit CRC checksum (a hash representation of the file) to the server, so that the server side can decide if the a store is necessary based on both mtime and CRC. The alternative is to do the CRC checking on the client side before calling store, which is possible we just need to make one more RPC call to retrieve both CRC and mtime from server (store if the file doesn't exist on the server of the client's file mtime is newer), which I choose not to do to reduce complexity.
- Delete: Same as Store, need to RequestWriteLock;
- Fetch: Doesn't require a WriteLock since it doesn't change state of the server, but client need to sync the mtime with server when writing the file down, otherwise it would trigger a store (since the file become "newer" than server). Note that the comparison using CRC is done on client side, because mtime comparison is already on client side (via HandleCallbackList) so keeping all comparison to make the decision on the same side makes it easier to debug. The alternative is make the CRC comparison on server side just like Store.
For the additional rpc:
-
RequestWriteLock: On the server side There is one global
std::map<std::string,std::string> FileOwnerwhere first element is filename, and second element is meant for clientid so that only if second element equals to "released" the client may obtain the lock for the file. The map is guarded by one globalstd::lock_guard<std::mutex> guard(g_owners_mutex)so that only one client is changing the state of the map at a time since it is possible multiple clients are calling Store at the same time and is being served by different worker threads of gRPC, while there is only one server thread (Details explained in testing). Alternaive design is to have multiple variables of type pair to represents locks to different file, so that not all file locks is blocked from changing while we are only changing one of the entry, but then we would have multiple mutex as well. Here I choose to save resource and easy implementation over performance (which I am unsure also, since creating/managing multiple mutexes also consumes time). -
ReleaseWriteLock: Similar to RequestWriteLock but the opposite which is used for changing the second element of
std::map<std::string,std::string> FileOwnerto "released" -
CallBackList: On the server side, there are two methods concerning this asynchronous rpc on dfs-lib-servernode-p2 i.e. ProcessCallback ProcessQueuedRequests
-
ProcessCallback: In ProcessCallback I did something similar to List in part1 i.e. list all the file and get attributes via opendir() and stat(). Alternatively, the client need to send the file list with mtime to server, than server can compute the change and send back to client. But since the original implementation in dfslibx-clientnode-p2.h only send a empty string to init the rpc, I cannot use this alternative design, but just do the comparision on client side in HandleCallbackList
-
ProcessQueuedRequests: This method decide when server async threads would call RequestCallbackList so that one request from client in Completion Queue would be ready to be service and proceed to next stage by calling ProcessCallback; I called RequestCallbackList when there is file change or new connection (requiring mutex protecting file changes flag and rpc notifying the server there is new connection) so that only latest detailed file list is send to client otherwise it is easy to have a stale CallBackList sent to HandleCallbackList which would trigger a Fetch (due to file not found in local) and then a Store (due to file creation in Fetch)and then a Delete again. Detailed explained in implementation.
-
HandleCallbackList: On Client side, once the client side completion queue (grpc::CompletionQueue DFSClientNode::completion_queue) provides call_data->reply
dfs_service::FileDetails meta_list = call_data->reply;The client would get a list of file name and its mtimes.
- if both server and client have the file and local file is newer, we call store
- if both side have the file but server is newer, we call fetch and set the mtime (so that it won't trigger another store).
- if file doesn't exist on local, we call fetch and set mtimes. (problematic, if we have delete timestamp, then it's better to compare delete timestamp from OS with server mtime)
- if file only exist on local, we call delete if the file is in server delete history, but since It is also possible that the INOTIFY event triggered store failed or the async thread is slower than the watcher thread, we call store.
stateDiagram
Server
Client1 --> Server
Client2 --> Server
Client3 --> Server
Note left of Client1: Store
Note Left of Client2: Fetch (Store rpc return ALREADY_EXISTS)
Note Left of Client3: Fetch (Store rpc return ALREADY_EXISTS)
stateDiagram
Server
Client1 --> Server
Client2 --> Server
Client3 --> Server
Note left of Client1: Delete
Note Left of Client2: locally Delete (Delete rpc return cancel)
Note Left of Client3: locally Delete (Delete rpc return cancel)
Secondly, let's talk about threading design.
Client side, we have watcher thread running InotifyWatcher which would monitor for creation/modification and delete event and trigger Store and Delete respectively. We also have async thread which would continuously calling the async rpc CallBackList and wait for response from server. We don't want watcher thread to compete with CallBackList since it is meaningless to handle a callbacklist that's stale if watcher thread hasn't complete yet. I used mutex g_local_ctrl here to make this explicit so that if a delete event happened, CallBackList would only be handled after the completion of the Delete rpc call.
Server side, we have Async threads running HandleAsyncRPC to serve the asynchronous rpc CallBackList and worker threads sprawn by gRPC to serve implicitly to serve synchronous rpc; note that there is one Server Thread running HandleSyncRPC but that's really holding the server up. Queue thread will be running queued_requests_callback which is set to ProcessQueuedRequests to make request available for HandleAsyncRPC to serve. Again, we don't want worker threads to compete with Queue thread since just like client it is meaning less to broadcast a stale callbacklist; here not only a mutex is used but also a boolean flag so that the sequence is always Store/Delete finished first and then we process queued requests. We cannot do it on the client side, because for client the change might not be originally from local.
part 1 is a synchrnous gRPC service with client and server. The sequence chart from the repository already fully explained the flow, it is perhaps more important to spend time on part2 from threading perspective. Part1 sequence chart attached at reference.
Basically, Server start up first and client connect to it; then we can perform various synchronous gRPC call waiting for unary or stream response from server; each response is a message with structure as described in .proto file.
On Client side, it would exit after performing the rpc since there is no watcher thread as in part2 to hold the client up by always running DFSClient::InotifyWatcher in a loop, also there is no async thread like part 2 which also can hold the client from exiting because of completion_queue.Next(&tag, &ok) is blocking and upon completion of HandleCallbackList we would trigger another rpc from client by calling InitCallbackList.
On server side, the main thread would be blocked on this->server->Wait(); Hence the server would be running until is destroyed or HandleSignal called exit(0).
Part 2, is a hybrid gRPC (we have synchronous and asynchronous rpc) supported DFS. It already has a sequence graph from the project repository. I attached it at Reference as well.
On client side, it has watcher thread and async thread.
watcher thread running InotifyWatcher which use registered callback method called InotifyWatcherCallback which it would continuously run in a while loop. As described in project design we don't want it to compete with async thread to avoid handling of stale callbacklist hence a mutex g_local_ctrl is used in InotifyWatcherCallback
std::lock_guard<std::mutex> lock(g_local_ctrl);
callback();The implementation of callback() is a while loop processing inotify events running InotifyEventCallback() on each of them. The event possible are described in inotify.h and we only care about three type of events IN_CREATE, IN_MODIFY and IN_DELETE.
In dfs-client-p2.cpp::InotifyWatcher
graph TD
M[Read the next inotify event as it becomes available] --> S[This loop handles each of the inotify events as they come through]
O[inotify event becomes avaialbe]
A[inotify IN_CREATE/IN_MODIFY/IN_DELETE event occurred] --> IN_CREATE[trigger rpc Store/Delete]
S --> O
O --> A
S -->M
IN_CREATE -->S
watcher thread is running HandleCallbackList. The first callbacklist rpc is triggered by the main thread and then subsequently after each completion of HandleCallbackList it would trigger another InitCallbackList() to continue to submit asynchronous rpc CallBackList request in to completion queue for server to handle.
// Initialize the callback list
this->client_node.InitCallbackList()The response from server is a message listing File Attributes of a file including name, size, mtime and ctime.
message FileDetail {
string name = 1;
uint64 size = 2;
int64 mtime = 3;
int64 ctime = 4;
}
message FileDetails{
repeated FileDetail filedetail = 1;
}graph TD
N[Notify server for new connection so that server would respond with one FileDetails even no Store/Delete has happened on server] --> I[completion_queue.Next loop] -->Lock[guard with g_local_ctrl mutex] --> local_file[get local_file_list to compare with FileDetails] --> file_details[get FileDetails by call_data->reply] --> ignore[ignore files without certain suffix]
ignore -->Store[if both have the files and local is newer] --> SS[Store]
ignore -->Fetch[if both have the files and server is newer] --> FF[Fetch and set mtime]
ignore -->Fetch2[if file doesn't exist in local] --> FF
ignore -->Store2[if file only exist in local but not in server delete hsitory] --> SS
ignore -->Delete[if file only exist in local but not in server delete hsitory] --> DD[Delete]
SS --> I
DD --> I
FF --> I
On Server Side, we have Async thread, Server thread and Queue thread.
Async thread is running HandleAsyncRPC in which it retrieve available request from server side completion queue by cq->Next(&tag, &ok) and then call Proceed to use ProcessCallback (which is just preparing the attributes of files on server and inject to CallData instance given).
graph TD
Create[Spawn a new CallData instance to serve new clients. CallData in CREATE state] --> Queue[Put tag into Request Queue]-->loop[While loop]-->
Next[cq->Next Block waiting to read the next event from the completion queue and get tag] --> P[Call ProcessCallback with tag and then set status to FINISH]
new[Spawn a new CallData instance to serve new clients]
P --> new --> loop
Next --> D[Delete CallData instance if status = FINISH]
Server thread is just there to hold the server, can be seen in Testing.
Queue thread is running queued_requests_callback which in turn is actually running logic in ProcessQueuedRequests in which this thread call RequestCallbackList to make server side request (using tag of a CallData instance created in Async thread and stored in queue_request with status CREATE) to unblock cq->Next above to proceed to ProcessCallback.
Most of the code is already discussed in Project Design and Flow of Control Graphs.
Part1 is a pure synchronous gRPC service. Part 2 is a hybrid gRPC service with asynchronous and synchronous rpc with multithreading.
In part1, as shown by Class diagram in the original GaTech repository.
DFSClient is a composition of ./bin/dfs-client-p1, the instance of DFSClient live and die with ./bin/dfs-client-p1. DFSClientNode class is a part of DFSClient, they have separate lifetimes. DFSClient construct and destroy DFSClientNode.
Similarly DFSServerNode is a composition of ./bin/dfs-server-p1.
In part2 Client Side is the same in terms of Class Design. But now the Server side has DFSServiceRunner in which it sprawn the threads and is a composition of DFSServerNode. And DFSData is a part of DFSServiceRunner, this class handles async CallBackList together with DFSCallDataManager (a part of DFSData)
std::mutex g_owners_mutex;
std::mutex g_modified_file;g_owners_mutex is used to protect multiple worker threads try to access the owner map representing WriteLock, g_modified_file is used to prevent both queue thread and worker threads try to change the state change happened in server but not yet serving requests from client for CallBackList.
On client side, even using mutex there is no guarantee of order when a local delete event happen, which of watcher thread (sending Delete rpc) async thread (getting FileDetails list from server) run first so if Server side give a stale version of list (triggerd by something else not the Delete rpc that hasn't happened yet) we could triggered a fetch. Note that from project design, you can see that I directly call fetch without checking mtime of the file on server if the file doesn't exist on local, this is because I haven't find a way to record the accurate delete time of a file so that we can compare with mtime of the file on server and decide either do nothing or call fetch.
On serverside, again even using mutex there is no guarantee of order when a Delete rpc is made from client, which of async thread (injecting the file details on server into CallData) or grpc background worker Thread (serving the Delete rpc call) get run first, so just like above it would broadcast a stale FileDetail list; however upon compeletion of serving Delete rpc call, I designed to send another FileDetails list to client. Now since as shown in project design, when the file on exist on local and in server delete history, I would remove the file locally hence after sometimes, server and client would stay sync eventually.
For part 1, I simply manually tested with using the command List,Stat,Store,Delete,Fetch directly using the binary of client. The other way is to use grpcurl.
For part 2, I manually tested with mounting two clients with the server and use terminal to add/remove file to generates the INOTIFY events.
The other way to do stress is listed in this post Helder Stress Test for project 4
For synchronous gRPC server there are multiple worker threads serving requests. For example, I log out the thread ID serving the store request.
>> DEBUG1: Thread 140000620431104 is serving store
>> DEBUG1: Thread 140000561682176 is serving storeSo even though there is only one so called server thread, we still need to take care of shared resources protections those synchronous rpc.
// Start the synchronous server on a separate thread
std::thread thread_server(HandleSyncRPC<RequestT, ResponseT>, this->server);
dfs_log(LL_SYSINFO) << "Server thread " << " started";
threads.push_back(std::move(thread_server));Hence, I think the server thread running HandleSyncRPC only exists to hold the server from exiting, given from documentation void grpc::Server::Wait is blocking.
/**
* Static callback for handling synchronous requests to the service protocol.
*
* @tparam RequestT
* @tparam ResponseT
* @param server
*/
template <typename RequestT, typename ResponseT>
static void HandleSyncRPC(std::shared_ptr<grpc::Server> server) {
server->Wait();
}opendir https://man7.org/linux/man-pages/man3/opendir.3.html
Filesystem library (since C++17) https://en.cppreference.com/w/cpp/filesystem
void grpc::Server::Wait https://grpc.github.io/grpc/cpp/classgrpc_1_1_server.html#a009682cfb2c453839b9eab318bbb8222
P4L1 Remote Procedure Calls https://gatech.instructure.com/courses/270294/pages/p4l1-remote-procedure-calls?module_item_id=2666700
gRPC C++ Reference https://grpc.github.io/grpc/cpp/index.html Protocol Buffers 3 Language Guide https://developers.google.com/protocol-buffers/docs/proto3 gRPC C++ Examples https://github.com/grpc/grpc/tree/master/examples/cpp gRPC C++ Tutorial https://grpc.io/docs/tutorials/basic/cpp/ Protobuffers Scalar types https://developers.google.com/protocol-buffers/docs/proto3#scalar gRPC Status Codes https://github.com/grpc/grpc/blob/master/doc/statuscodes.md gRPC Deadline https://grpc.io/blog/deadlines/ Helder Stress Test for project 4 https://piazza.com/class/lco3fd6h1yo48k/post/1212
Part 1 sequence chart
Part 2 sequence chart
