GRPC async Server: ServerCompletionQueue::AsyncNext does not return NextStatus::SHUTDOWN after calling Shutdown on the queue

Question

I have a C++ GRPC server code that uses the async API. On shutdown of the server, I run this to shutdown the server completion queue and drain it:

queue->Shutdown();

void* tag = nullptr;
bool ok = false;

for (auto status = queue->AsyncNext (&tag, &ok, std::chrono::system_clock::now());
     status != grpc::CompletionQueue::NextStatus::SHUTDOWN;
     status = queue->AsyncNext (&tag, &ok, std::chrono::system_clock::now() + std::chrono::milliseconds (50)))
{
    if (status == grpc::CompletionQueue::NextStatus::GOT_EVENT)
    {
        // do some cleanup work
    }
}

However, my code will never break out of that for loop since the status is always TIMEOUT but never SHUTDOWN. What am I doing wrong here? From my understanding of the documentation after calling Shutdown on a queue, AsyncNext should either return GOT_EVENT if there are still events in the queue that I can cleanup then or it should return SHUTDOWN if there are no more events in the queue.

Answer 1

I figured it out myself in the meantime, so I'll answer my own question to help others coming here in future.

The reason why the server queue never shut down was because the server that is associated with it had not been shut down before. So in order to make the queue shut down, we first need to call Shutdown on the server. After the server has been shut down cleanly, the queue will shut down.

Still this didn't work at the time, because in order to shut down the server, its queue which might still contain events must be handled first. In my scenario there were no more calls from other threads to handle the queue events at that point. So what I came up with was spinning up a temporary thread which drains the queue, shut down the server, then shutdown the queue and finally join the thread:

std::thread queueDrainer ([&] {
    void* tag = nullptr;
    bool ok = false;

    // Next will return false as soon as the queue is in shutdown mode
    while (queue->Next (&tag, &ok))
    {
        if (ok)
        {
            // do some cleanup work
        }
    }
});

server->Shutdown();

queue->Shutdown();

queueDrainer.join();