19 Apr 2016

Fault Tolerance in Action -- RPC

Before we start to talk fault tolerance, one important thing we must care about is communication. There are lots of paradigms to do communication between nodes such as message passing, remote procedure call, message queue, active message, RDMA and so on. These communication patterns differ each other from different levels and usages of different applications. In this series of blog posts, I choose remote procedure call(RPC) as the basic block of communication for its simplicity and general semantic.

RPC is a widely used request-response protocol which has a long history. The protocol dates back to early distributed computing in late 1960s, theoretical proposals of remote procedure calls as the model of network operation dates back to the 1970s, and practical implementation dates back to the early 1980s. Within a RPC service, client sends a request message to server to invoke a function call with given parameters. After receiving the command from client, the remote server sends a response back to the client. In the client’s sight, this procedure has no difference with an ordinary function call. It’s rather a simple way for programming since RPC hides many low level communication details in client’s end. A client object could use the functionality of another object via network.

RPC aims for the level of transparency as below:

Client:
 c = func(a, b)

Server:
 func(a, b) {
   // implementation of func
   return c
 }

Since RPC always plays a foundational role in developing distributed systems, it is necessary to think more of its design.

Firstly, failures. RPC decouples a functional procedure into client process, networking transmitting and server process. There are big chances occurring failure during a RPC: lost package, broken network, crash server, etc. Therefore, it’s indispensable for client to realize the failure and known the exact reason of failure. There are two basic schemes of failure handling:

1. At least once: client will continue sending requests if it couldn’t get reply in a expected time period. After several tries, it will return an error.
2. At most once: client will send request together with a global request ID for server to detect duplicate requests.

I couldn’t say one of them is better or worse than the other because it is just like a commitment between RPC library and high level system. At-least-once scheme works well in read-only scenario and depends on upper module codes to handle write and update case. At-most-once scheme is more widely implemented. Actually, a real system will handle the logic of RPC failure by itself. For example, use TCP to ensure reliable connections.

Another important aspect of RPC design is thread model. Client may create lots of threads to establish connection. Moreover, one server process might response to many many connected clients. For the reason why it is existed, please check out C10K problem and C10M problem. A straightforward method for server process to make it work is to fork new process for each client connection which handles all of the job in this connection. However, process is too heavy in OS and it is hard to share global information between thousands of connections. A better way is using light-weight threads. A main IO thread takes charge of dispatching requests, when new request comes, he creates a new thread to do the real job. Actually, the famous strategy is called event IO. The essential thinking is similar: some event IO threads are responsible for dispatching requests and sending replies. When a request come, they deliver actual requests to a thread pool which owns many threads to execute function calls. When result or data is ready, they send replies back to client. In this stage, event IO threads might also use threads in thread pool to do copy and writing. This strategy also requires starting event IO threads in client end to talk with server. Don’t worry, good RPC implementations always take care of this problem.

Thirdly, request could be executed only if it has real meaning. So we need a protocol sending and receiving messages such as type and value, that’s called serialization and deserialization. There are a lot of basic tools for doing this, the key point you must remember is not only the efficiency of serialization and deserialization itself but also the message size to transmitting. These two parts determine the main performance of communication together. Moreover, to support different programming languages of caller and callee, we need a IDL(Interface Definition Language) which maps a general, readable message format to different languages.

Besides, there are some other problems in designing a RPC library such as QoS control with authentication to ensure safety. But these kinds of design are out of scope in these series.

Till now I hope you have understand the basic concept and logic of RPC. Now let’s write some real code of using a RPC library! Some high level languages implemented advanced utilities inside language code which is easy to implement RPC. In Python, we could use exec(inspect.getsource(fun)), you must take care of closure. In other networking oriented programming languages, RPC implementation is included inside the package of the languages. For instance, Golang‘s net/rpc package is very powerful and easy to use, it is closed to the original logic of doing RPC. But as I mentioned in last blog, I will use C++. So we must choose a third party library to do RPC. Although C++ is a bad language, we have to use it in many scenes for its high performance…

Here I choose Thrift which is invented by Facebook.Inc and now it is a famous apache project. Thrift may be not that fast and flexible comparing to other RPC libraries, the simple reason for me to use it is that thrift not only have serialization and deserialization module, it also implements RPC service. Let’s see a toy example:

First of all, write a thrift format file describing your remote function call utility:

# divide.thrift
# Toy rpc example: implement divide in server end

enum Errors {
  SUCCESS = 0,
  INVALID_REQUEST = 1
}

struct Response {
  1:double result,
  2:Errors error
}

service divide {
  Response div(1:double numerator, 2:double denominator),
  string sayhi()
}

Because it aims to support different languages for remote functional call, we need an additional step to convert the thrift file to C++ code: thrift --gen cpp divide.thrift

After doing that, we can start writing codes in server end. The logic is simple: inherit base class divideIf and implement the function div and sayhi we defined above, define a server object and use serve to continue replying requests. In this toy example, I use TSimpleServer for simplicity. Thrift supplies more thread model such as TThreadedServer, TNonblockingServer and TThreadPoolServer. I recommend to use TThreadedServer and TNonblockingServer in practice.

// toy_server.cpp
class divideHandler : virtual public divideIf {
 public:
  void div(Response& _return, const double numerator, const double denominator) {
    if(denominator == 0.) {
      _return.error = Errors::INVALID_REQUEST;
      printf("Cannot divide by 0!\n");
      return;
    }
    _return.result = numerator / denominator;
    _return.error = Errors::SUCCESS;
    printf("div: %lf div %lf is %lf.\n", numerator, denominator, _return.result);
  }

  void sayhi(std::string& _return) {
    _return = "HELLO WORLD.";
   printf("sayh: .\n");
  }
};

int main(int argc, char **argv) {
  TSimpleServer server(...);
  server.serve();
  return 0;
}

Write client code. In client end, we must specify the address of server to connect. Also, we can specify connection type such as TCP, UDP and so on. The code seems like this:

// toy_client.cpp
int main(int argc, char *argv[])
{
  divideClient client(server_address);

  try {
    transport->open();

    Response res;
    client.div(res, 10., 3.1415926);
    if(res.error == Errors::SUCCESS) {
      std::cout << res.result << std::endl;
    } else if (res.error == Errors::INVALID_REQUEST){
      std::cout << "Cannot divide by 0!" << std::endl;
    } else {
      std::cout << "Unknown error!" << std::endl;
    }
    std::string what;
    client.sayhi(what);
    std::cout << what << std::endl;

    transport->close();
  } catch (TException& tx) {
    std::cout << "ERROR: " << tx.what() << std::endl;
  }
}

You are almost done! Hope you can understand the simple logic of the code. Till now, we have implemented a remote div and sayhi. But that’s not enough because in real case, we want invoke a member function of remote objects. For this, we must define a member variable obj of divideHandler class and invoke the actual implementation of obj.div. Meanwhile, we need create a subthread to serve requests which doesn’t hang main thread. If you want to try it out, refer to compilable code here.

Hong Wu at 00:11