C++20 coroutines for asynchronous gRPC services

Writing asynchronous gRPC clients and servers in C++ can be a challenging task. The official documentation provides a minimal example with explicit state and memory management that is hardly scalable to real-world use cases. The recently stabilized callback API aims to make things easier, but writing callback-style code in a modern C++ world still does not feel right. Thankfully, with the power of abstractions, adding C++20 coroutine support on top of the gRPC CompletionQueue API can be achieved with relative ease.

For the impatient head over to the asio-grpc repository for a production-ready implementation based on Boost.Asio.
This article describes what a gRPC CompletionQueue is, how to adapt it to Boost.Asio’s ExecutionContext and how to use CompletionToken and CompletionHandler to process RPCs in an asynchronous manner.

The gRPC CompletionQueue API works with user-provided tags (void*). Whenever an RPC step is initiated, like reading from a server-side stream, a tag must be provided as the last argument. The CompletionQueue, essentially an event loop, should then be run by repeatedly invoking its Next function which will block until any asynchronous operation has completed. The tag that was originally passed to the initiating function and the result of the asynchronous operation are returned from the call to Next, so that the user can initiate the next RPC step.

Now that we know how the CompletionQueue API works, let us look at the second ingredient for adding C++20 coroutine support: Boost.Asio.

Boost.Asio provides a generic framework for writing and composing asynchronous operations — with plenty of customization points for the user. For our purpose we will adapt the CompletionQueue to the ExecutionContext. This will later allow us to co_spawn C++20 coroutines that are run on the CompletionQueue. We start by creating a wrapper around the queue:

Since we inherit from execution_context we need to perform the sequence of execution_context::shutdown and execution_context::destroy as described in Boost.Asio’s documentation. We also want to provide a way of draining the CompletionQueue. Let us add a function called run() which does exactly that:

Next we need to implement the Executor for our GrpcContext. While its requirements might seem daunting at first, its main task is to submit a callable object to the CompletionQueue such that it will be invoked within the GrpcContext::run function. To achieve that we equip our GrpcContext with a grpc::Alarm and set it with an immediate deadline. We also allocate memory for the callable object, wrap it into an Operation and add it to a container (queued_operations). The container allows us to keep track of all callable objects that should be invoked within GrpcContext::run.

Once the alarm expires, the MARKER_TAG will be returned from CompletionQueue::Next. This is where we iterate through all queued_operations and complete them.

In the snippets above I have used Operation, TypeErasedOperation and queued_operations. Let us see how they are implemented:

Here we make use of type-erasure since our GrpcContext cannot know about all possible types of callable objects that are submitted by the executor. Additionally, we avoid performing additional memory allocations by making our TypeErasedOperation fit into singly-linked lists provided by Boost.Intrusive.

With all of this in place we can now use GrpcContext and its executor_type as arguments to Boost.Asio functions like co_spawn to run a C++20 coroutine on the CompletionQueue. To summarize, here is the structure that we have created:

Now let us see how to actually process RPCs. As we have learned earlier, the CompletionQueue API is tag-based. Whenever we want to initiate an RPC step we need to provide a tag. We will use a callable object as our tag, that, when invoked by the GrpcContext will inform us about the completion of the asynchronous operation. Such callable objects can be obtained in the form of Boost.Asio’s CompletionToken and CompletionHandler concepts, in particular with the helper function async_initiate that turns a token into a handler. Reading from a server-side streaming RPC and awaiting its result can thereby be implemented as follows:

That is it, now we can process RPCs using C++20 coroutines. We can also use other asynchronous communication primitives like std::future, stackless coroutine, Boost.Coroutine and the good-old callback. All we have to do is to replace boost::asio::use_awaitable with them.

The code snippets above are a good foundation for writing your own Boost.Asio based asynchronous gRPC service. Additional features such as tracking the number of outstanding operations so that GrpcContext::run returns when there are no more left, stopping the GrpcContext prematurely, thread-safety and so on can easily be added on top. For a possible implementation of those and more see asio-grpc.

As a bonus of using Boost.Asio we also get access to other non-blocking I/O operation without having to create a new thread. We can i.e. take code from the Boost.Beast examples, replace all occurrences of net::io_context with our GrpcContext et violà, we have HTTP requests running in the same thread as our RPCs.

Conclusion

We learned how the challenges of writing asynchronous gRPC clients and servers in C++ can be overcome with the help of generic libraries like Boost.Asio. We managed to implement an ExecutionContext and an associated Executor which submit all work to a gRPC CompletionQueue, allowing us to handle RPCs with C++20 coroutines, std::futures, stackless coroutines, Boost.Coroutines or callbacks — whatever we prefer. If you are looking to write gRPC services in such a way then consider using asio-grpc which is successfully used in production at 3YOURMIND.

Thank you Simone Dal Poz, Felix Bauer and Lewis Cowper for helping me write this article.