Confusion around CompletionQueue in an async C++ client

user_1307094 0
Viewed 3342

I have some confusion with regards on how to use CompletionQueue for async C++ client. My server is in C#, so my question here is purely around the client sending a request in an async manner to the server.

For reference, here is how I setup my client code to make an async request:

template<typename ResponseType, typename AsyncOpExecutor>
bool DoAsyncOp(const AsyncOpExecutor& op, const unsigned int deadlineMs, ResponseType& response)
{
    grpc::ClientContext ctx;
    grpc::CompletionQueue queue;

    const std::unique_ptr<grpc::ClientAsyncResponseReaderInterface<ResponseType>> asyncOpResponse = op(ctx, queue);

    grpc::Status status;
    int requestTag = 1;
    asyncOpResponse->Finish(&response, &status, (void*)requestTag);

    bool result = false;
    bool gotEvent = false;

    do
    {
        const std::chrono::time_point<std::chrono::system_clock> deadline = std::chrono::system_clock::now() + std::chrono::milliseconds(deadlineMs);

        void* got_tag;
        bool ok = false;
        const grpc::CompletionQueue::NextStatus nextStatus = queue.AsyncNext(&got_tag, &ok, deadline);

        switch (nextStatus)
        {
        case grpc::CompletionQueue::NextStatus::TIMEOUT:
            continue;

        case grpc::CompletionQueue::NextStatus::GOT_EVENT:
            assert(got_tag == (void*)requestTag);
                        // ok is always true even if I close the server while request is in progress.
            assert(ok);
            result = status.ok();
            gotEvent = true;
            break;

        // Given that I am creating a new CompletionQueue per request (not using a shared one), is this flag likely to occur?
        case grpc::CompletionQueue::NextStatus::SHUTDOWN:
            result = false;
            gotEvent = false;
            break;

        default:
            result = false;
            gotEvent = false;
            break;
        }
    } while (!gotEvent);

    return result;
}

My first confusion is around the best way to setup the CompletionQueue. This answer seems to suggest that a single CompletionQueue can be used across requests. How will this behave if multiple threads use the same queue to make requests. Let's say that I change my code above to use a shared queue rather than creating a fresh one per request.

  • How will one thread know that the response it receives is intended for it, rather than for a different thread?

  • Do I need to assign a unique tag per thread, and on each thread check that the received tag from the queue matches the one I originally sent?

  • If thread A receives a tag intended for thread B, does that mean thread B will be able to query for its tag later, or is that tag lost because thread A saw it first?

  • Is there actually a major issue of using a fresh queue per request, rather than sharing?

My second confusion is around grpc::CompletionQueue::NextStatus::SHUTDOWN result. If I use a fresh queue per request, and not explicitly call shutdown on the queue, is this result ever likely to occur? If yes, what will trigger it? One test I carried out is closing the server while a request is in progress, however rather than getting the shutdown result I got the grpc::CompletionQueue::NextStatus::GOT_EVENT result, with status set as UNAVAILABLE.

My final confusion is around the ok flag. I have read this answer, however it's still not very clear. Given the use case and code posted above, if the result I get from the queue is grpc::CompletionQueue::NextStatus::GOT_EVENT, can the ok flag be false, and if yes, what will cause it to be false? Again this is purely around the client and not how CompletionQueue is handles on the server.

grpc
1 Answers
  1. Use the tag returned from the completion queue to know which completion queue operation initiated this action. You have to make sure that there are not two CQ operations in flight at the same time using the same tag.
  2. Each CQ operation in flight simultaneously needs a distinct tag. This applies whether those ops were started from the same thread or different threads.
  3. Each tag comes out of the Next function only once. If you have multiple threads calling Next on the same CompletionQueue, they should all be able to process (or pass off) any of the tags registered to that CQ.
  4. You can use a fresh queue per RPC, like the sync API actually does internally. It's just less efficient and basically devolves to the sync API in that case since you won't be able to have many operations outstanding at the same time.

You need to call Shutdown at some point and then drain the queue. That's the standard for the API. Otherwise, especially for server CQs, there will likely be leaks. By drain, I mean call Next until it returns false (or AsyncNext until it returns SHUTDOWN).

ok can be false for GOT_EVENT for streaming calls, server-side calls, etc. If you're only looking at client unary calls (like your example), it won't be false. A false ok essentially means that this side of the RPC is broken. The documentation for this is in the header file for the CQ.

user_81912230
Related