C++ GRPC ClientAsyncReaderWriter: how to check if data is available for read?

Sequencing Read() calls

Can I try to read if it no data available?

Yep, and it's going to be case more often than not. Read() will do nothing until data is available, and only then put its passed tag into the completion queue. (see below for details)

Is it blocking call?

Nope. Read() and Write() return immediately. However, you can only have one of each in flight at any given moment. If you try to send a second one before the previous has completed, it (the second one) will fail.

What is the proper way to async reading?

Each time a Read() is done, start a new one. For that, you need to be able to tell when a Read() is done. This is where tags come in!

When you call Read(&msg, tag), or Write(request, tag),you are telling grpc to put tag in the completion queue associated with that responder once that operation has completed. grpc doesn't care what the tag is, it just hands it off.

So the general strategy you will want to go for is:

• As soon as you are ready to start receiving messages:
  • call responder->Read() once with some tag that you will recognize as a "read done".
• Whenever cq_.Next() gives you back that tag, and ok == true:
  • consume the message
  • Queue up a new responder->Read() with that same tag.

Obviously, you'll also want to do something similar for your calls to Write().

But since you still want to be able to lookup the handler instance from a given tag, you'll need a way to pack a reference to the handler as well as information about which operation is being finished in a single tag.

Completion queues

Lookup the handler instance from a given tag? Why?

The true raison d'être of completion queues is unfortunately not evident from the examples. They allow multiple asynchronous rpcs to share the same thread. Unless your application only ever makes a single rpc call, the handling thread should not be associated with a specific responder. Instead, that thread should be a general-purpose worker that dispatches events to the correct handler based on the content of the tag.

The official examples tend to do that by using pointer to the handler object as the tag. That works when there's a specific sequence of events to expect since you can easily predict what a handler is reacting to. You often can't do that with async bidirectional streams, since any given completion event could be a Read() or a Write() finishing.

Example

Here's a general outline of what I personally consider to be a clean way to go about all that:

// Base class for async bidir RPCs handlers. 
// This is so that the handling thread is not associated with a specific rpc method.
class RpcHandler {
  // This will be used as the "tag" argument to the various grpc calls.
  struct TagData {
    enum class Type {
      start_done,
      read_done,
      write_done,
      // add more as needed...
    };

    RpcHandler* handler;
    Type evt;
  };

  struct TagSet {
    TagSet(RpcHandler* self)
        : start_done{self, TagData::Type::start_done},
          read_done{self, TagData::Type::read_done},
          write_done{self, TagData::Type::write_done} {}
    TagData start_done;
    TagData read_done;
    TagData write_done;
  };

 public:
  RpcHandler() : tags(this) {}

  virtual ~RpcHandler() = default;

  // The actual tag objects we'll be passing
  TagSet tags;

  virtual void on_ready() = 0;
  virtual void on_recv() = 0;
  virtual void on_write_done() = 0;

  static void handling_thread_main(grpc::CompletionQueue* cq) {
    void* raw_tag = nullptr;
    bool ok = false;

    while (cq->Next(&raw_tag, &ok)) {
      TagData* tag = reinterpret_cast<TagData*>(raw_tag);
      if(!ok) {
        // Handle error
      }
      else {
        switch (tag->evt) {
          case TagData::Type::start_done:
            tag->handler->on_ready();
            break;
          case TagData::Type::read_done:
            tag->handler->on_recv();
            break;
          case TagData::Type::write_done:
            tag->handler->on_write_done();
            break;
        }
      }
    }
  }
};

void do_something_with_response(Response const&);

class MyHandler final : public RpcHandler {
 public:
  using responder_ptr =
      std::unique_ptr<grpc::ClientAsyncReaderWriter<Request, Response>>;

  MyHandler(responder_ptr responder) : responder_(std::move(responder)) {
    // This lock is needed because StartCall() can
    // cause the handler thread to access the object.
    std::lock_guard lock(mutex_);

    responder_->StartCall(&tags.start_done);
  }

  ~MyHandler() {
    // TODO: finish/abort the streaming rpc as appropriate.
  }

  void send(const Request& msg) {
    std::lock_guard lock(mutex_);
    if (!sending_) {
      sending_ = true;
      responder_->Write(msg, &tags.write_done);
    } else {
      // TODO: add some form of synchronous wait, or outright failure
      // if the queue starts to get too big.
      queued_msgs_.push(msg);
    }
  }

 private:
  // When the rpc is ready, queue the first read
  void on_ready() override {
    std::lock_guard l(mutex_);  // To synchronize with the constructor
    responder_->Read(&incoming_, &tags.read_done);
  };

  // When a message arrives, use it, and start reading the next one
  void on_recv() override {
    // incoming_ never leaves the handling thread, so no need to lock

    // ------ If handling is cheap and stays in the handling thread.
    do_something_with_response(incoming_);
    responder_->Read(&incoming_, &tags.read_done);

    // ------ If responses is expensive or involves another thread.
    // Response msg = std::move(incoming_); 
    // responder_->Read(&incoming_, &tags.read_done); 
    // do_something_with_response(msg);
  };

  // When has been sent, send the next one is there is any
  void on_write_done() override {
    std::lock_guard lock(mutex_);
    if (!queued_msgs_.empty()) {
      responder_->Write(queued_msgs_.front(), &tags.write_done);
      queued_msgs_.pop();
    } else {
      sending_ = false;
    }
  };

  responder_ptr responder_;

  // Only ever touched by the handler thread post-construction.
  Response incoming_;

  bool sending_ = false;
  std::queue<Request> queued_msgs_;

  std::mutex mutex_;  // grpc might be thread-safe, MyHandler isn't...
};

int main() {
  // Start the thread as soon as you have a completion queue.
  auto cq = std::make_unique<grpc::CompletionQueue>();
  std::thread t(RpcHandler::handling_thread_main, cq.get());

  // Multiple concurent RPCs sharing the same handling thread:
  MyHandler handler1(serviceA->MethodA(&context, cq.get()));
  MyHandler handler2(serviceA->MethodA(&context, cq.get()));
  MyHandlerB handler3(serviceA->MethodB(&context, cq.get()));
  MyHandlerC handler4(serviceB->MethodC(&context, cq.get()));
}

If you have a keen eye, you will notice that the code above stores a bunch (1 per event type) of redundant this pointers in the handler. It's generally not a big deal, but it is possible to do without them via multiple inheritance and downcasting, but that's starting to be somewhat beyond the scope of this question.