This tutorial shows you how to write a simple server and client in C++ usinggRPC’s asynchronous/non-blocking APIs. It assumes you are already familiar withwriting simple synchronous gRPC code, as described ingRPC Basics:C++. The example used in this tutorial follows onfrom the basic we used in theoverview. You’ll find it along with installationinstructions in.

gRPC uses theAPI for asynchronous operations. The basic work flowis as follows:

  • bind a CompletionQueue to an RPC call
  • do something like a read or write, present with a unique void* tag
  • call CompletionQueue::Next to wait for operations to complete. If a tagappears, it indicates that the corresponding operation is complete.

To use an asynchronous client to call a remote method, you first create achannel and stub, just as you do in a. Once you have your stub, you dothe following to make an asynchronous call:

  • Initiate the RPC and create a handle for it. Bind the RPC to aCompletionQueue.
  • Ask for the reply and final status, with a unique tag
  1. Status status;
  2. rpc->Finish(&reply, &status, (void*)1);
  • Wait for the completion queue to return the next tag. The reply and status areready once the tag passed into the corresponding Finish() call is returned.
  1. void* got_tag;
  2. bool ok = false;
  3. cq.Next(&got_tag, &ok);
  4. if (ok && got_tag == (void*)1) {
  5.   // check reply and status
  6. }

The server implementation requests an RPC call with a tag and then waits for thecompletion queue to return the tag. The basic flow for handling an RPCasynchronously is:

  • Build a server exporting the async service
  • Request one RPC, providing a unique tag
  1. ServerContext context;
  2. HelloRequest request;
  3. ServerAsyncResponseWriter<HelloReply> responder;
  4. service.RequestSayHello(&context, &request, &responder, &cq, &cq, (void*)1);
  • Wait for the completion queue to return the tag. The context, request andresponder are ready once the tag is retrieved.
  1. HelloReply reply;
  2. Status status;
  3. void* got_tag;
  4. bool ok = false;
  5. cq.Next(&got_tag, &ok);
  6. if (ok && got_tag == (void*)1) {
  7.   responder.Finish(reply, status, (void*)2);
  8. }
  • Wait for the completion queue to return the tag. The RPC is finished when thetag is back.

This basic flow, however, doesn’t take into account the server handling multiplerequests concurrently. To deal with this, our complete async server example usesa CallData object to maintain the state of each RPC, and uses the address ofthis object as the unique tag for the call.

  1. class CallData {
  2. public:
  3.   // Take in the "service" instance (in this case representing an asynchronous
  5.   // with the gRPC runtime.
  6.   CallData(Greeter::AsyncService* service, ServerCompletionQueue* cq)
  7.       : service_(service), cq_(cq), responder_(&ctx_), status_(CREATE) {
  8.     // Invoke the serving logic right away.
  9.     Proceed();
  10.   }
  12.   void Proceed() {
  13.     if (status_ == CREATE) {
  14.       // As part of the initial CREATE state, we *request* that the system
  15.       // start processing SayHello requests. In this request, "this" acts are
  16.       // the tag uniquely identifying the request (so that different CallData
  17.       // instances can serve different requests concurrently), in this case
  18.       // the memory address of this CallData instance.
  19.       service_->RequestSayHello(&ctx_, &request_, &responder_, cq_, cq_,
  20.                                 this);
  21.       // Make this instance progress to the PROCESS state.
  22.       status_ = PROCESS;
  23.     } else if (status_ == PROCESS) {
  24.       // Spawn a new CallData instance to serve new clients while we process
  25.       // the one for this CallData. The instance will deallocate itself as
  26.       // part of its FINISH state.
  27.       new CallData(service_, cq_);
  28.       // The actual processing.
  29.       std::string prefix("Hello ");
  30.       reply_.set_message(prefix + request_.name());
  32.       // And we are done! Let the gRPC runtime know we've finished, using the
  33.       // memory address of this instance as the uniquely identifying tag for
  35.       responder_.Finish(reply_, Status::OK, this);
  36.       status_ = FINISH;
  37.     } else {
  38.       GPR_ASSERT(status_ == FINISH);
  39.       // Once in the FINISH state, deallocate ourselves (CallData).
  40.       delete this;
  41.     }
  42.   }
  43. }

For simplicity the server only uses one completion queue for all events, andruns a main loop in HandleRpcs to query the queue:

  1. void HandleRpcs() {
  2.   // Spawn a new CallData instance to serve new clients.
  3.   new CallData(&service_, cq_.get());
  4.   void* tag;  // uniquely identifies a request.
  5.   bool ok;
  6.   while (true) {
  7.     // Block waiting to read the next event from the completion queue. The
  8.     // event is uniquely identified by its tag, which in this case is the
  9.     // memory address of a CallData instance.
  10.     cq_->Next(&tag, &ok);
  11.     GPR_ASSERT(ok);
  12.     static_cast<CallData*>(tag)->Proceed();
  13.   }
  14. }

Remember we got our completion queue instance cq in ServerImpl::Run() byrunning cq = builder.AddCompletionQueue(). Looking atServerBuilder::AddCompletionQueue's documentation we see that

Refer to ServerBuilder::AddCompletionQueue's full docstring for more details.What this means in our example is that destructor looks like: