How do I use select() and gRPC to create a server?

Question

I need to use gRPC but in a single-threaded application (with additional socket channels). Naively, I'm thinking of using select() and depending on which file descriptor pops, calling gRPC to handle the message. My question is, can someone give me a rough (5-10 lines of code) outline skeleton on what I need to call after the select() pops?

Looking at Google's "hello world" example in the synchronous case implies a thread pool (which I can't use), and in the asynchronous case shows the main loop blocking -- which doesn't work for me because I need to handle other socket operations.

Answer 1

You can't do it, at this point (and probably ever).

One of the big weaknesses of event loops, including direct use of select()/poll() style APIs, is that they aren't composable in any natural way short of direct integration between the two.

We could theoretically add such functionality for Linux -- exporting an epoll_fd with a timerfd which becomes readable if it would be productive to call into a completion queue, but doing so would impose substantial constraints and architectural overhead on the rest of the stack just to support this usecase only on Linux. Everywhere else would require a background thread to manage that fd's readability.

Answer 2

This can be done using a gRPC async service along with grpc::Alarm to send any events that come from select or other polling APIs onto the gRPC completion queue. You can see an example using Epoll and gRPC together in this gist. The important functions are these two:

bool grpc_tick(grpc::ServerCompletionQueue& queue) {
  void* tag = nullptr;
  bool ok = false;
  auto next_status = queue.AsyncNext(&tag, &ok, std::chrono::system_clock::now());
  if (next_status == grpc::CompletionQueue::GOT_EVENT) {
    if (ok && tag) {
      static_cast<RequestProcessor*>(tag)->grpc_queue_tick();
    } else {
      std::cerr << "Not OK or bad tag: " << ok << "; " << tag << std::endl;
      return false;
    }
  }
  return next_status != grpc::CompletionQueue::SHUTDOWN;
}

bool tick_loops(int epoll, grpc::ServerCompletionQueue& queue) {
  // Pump epoll events over to gRPC's completion queue.
  epoll_event event{0};
  while (epoll_wait(epoll, &event, /*maxevents=*/1, /*timeout=*/0)) {
    grpc::Alarm alarm;
    alarm.Set(&queue, std::chrono::system_clock::now(), event.data.ptr);
    if (!grpc_tick(queue)) return false;
  }

  // Make sure gRPC gets at least 1 tick.
  return grpc_tick(queue);
}

Here you can see the tick_loops function repeatedly calls epoll_wait until no more events are returned. For each epoll event, a grpc::Alarm is constructed with the deadline set to right now. After that, the gRPC event loop is immediately pumped with grpc_tick.

Note that the grpc::Alarm instance MUST outlive its time on the completion queue. In a real-world application, the alarm should be somehow attached to the tag (event.data.ptr in this example) so it can be cleaned up in the completion callback.

The gRPC event loop is then pumped again to ensure that any non-epoll events are also processed.

Completion queues are thread safe, so you could also put the epoll pump on one thread and the gRPC pump on another. With this setup you would not need to set the polling timeouts for each to 0 as they are in this example. This would reduce CPU usage by limiting dry cycles of the event loop pumps.