diff --git a/include/grpc++/impl/codegen/completion_queue.h b/include/grpc++/impl/codegen/completion_queue.h index 025eea3ef5b..90595de4826 100644 --- a/include/grpc++/impl/codegen/completion_queue.h +++ b/include/grpc++/impl/codegen/completion_queue.h @@ -247,6 +247,12 @@ class CompletionQueue : private GrpcLibraryCodegen { /// Performs a single polling pluck on \a tag. /// \warning Must not be mixed with calls to \a Next. + /// + /// TODO: sreek - This calls tag->FinalizeResult() even if the cq_ is already + /// shutdown. This is most likely a bug and if it is a bug, then change this + /// implementation to simple call the other TryPluck function with a zero + /// timeout. i.e: + /// TryPluck(tag, gpr_time_0(GPR_CLOCK_REALTIME)) void TryPluck(CompletionQueueTag* tag) { auto deadline = g_core_codegen_interface->gpr_time_0(GPR_CLOCK_REALTIME); auto ev = g_core_codegen_interface->grpc_completion_queue_pluck( @@ -258,6 +264,23 @@ class CompletionQueue : private GrpcLibraryCodegen { GPR_CODEGEN_ASSERT(!tag->FinalizeResult(&ignored, &ok)); } + /// Performs a single polling pluck on \a tag. Calls tag->FinalizeResult if + /// the pluck() was successful and returned the tag. + /// + /// This exects tag->FinalizeResult (if called) to return 'false' i.e expects + /// that the tag is internal not something that is returned to the user. + void TryPluck(CompletionQueueTag* tag, gpr_timespec deadline) { + auto ev = g_core_codegen_interface->grpc_completion_queue_pluck( + cq_, tag, deadline, nullptr); + if (ev.type == GRPC_QUEUE_TIMEOUT || ev.type == GRPC_QUEUE_SHUTDOWN) { + return; + } + + bool ok = ev.success != 0; + void* ignored = tag; + GPR_CODEGEN_ASSERT(!tag->FinalizeResult(&ignored, &ok)); + } + grpc_completion_queue* cq_; // owned gpr_atm avalanches_in_flight_; diff --git a/include/grpc++/impl/codegen/server_context.h b/include/grpc++/impl/codegen/server_context.h index bf9a9b6f1a5..f84ede9eb11 100644 --- a/include/grpc++/impl/codegen/server_context.h +++ b/include/grpc++/impl/codegen/server_context.h @@ -41,6 +41,7 @@ #include #include +#include #include #include #include @@ -212,6 +213,8 @@ class ServerContext { class CompletionOp; void BeginCompletionOp(Call* call); + // Return the tag queued by BeginCompletionOp() + CompletionQueueTag* GetCompletionOpTag(); ServerContext(gpr_timespec deadline, grpc_metadata_array* arr); diff --git a/src/cpp/server/server_cc.cc b/src/cpp/server/server_cc.cc index 6adabef7636..6bb88bcd945 100644 --- a/src/cpp/server/server_cc.cc +++ b/src/cpp/server/server_cc.cc @@ -221,9 +221,14 @@ class Server::SyncRequest final : public CompletionQueueTag { MethodHandler::HandlerParameter(&call_, &ctx_, request_payload_)); global_callbacks->PostSynchronousRequest(&ctx_); request_payload_ = nullptr; - DummyTag ignored_tag; + cq_.Shutdown(); - /* Ensure the cq_ is shutdown (else this will hang indefinitely) */ + + CompletionQueueTag* op_tag = ctx_.GetCompletionOpTag(); + cq_.TryPluck(op_tag, gpr_inf_future(GPR_CLOCK_REALTIME)); + + /* Ensure the cq_ is shutdown */ + DummyTag ignored_tag; GPR_ASSERT(cq_.Pluck(&ignored_tag) == false); } diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc index 05c05c86953..01e9c004346 100644 --- a/src/cpp/server/server_context.cc +++ b/src/cpp/server/server_context.cc @@ -166,6 +166,10 @@ void ServerContext::BeginCompletionOp(Call* call) { call->PerformOps(completion_op_); } +CompletionQueueTag* ServerContext::GetCompletionOpTag() { + return static_cast(completion_op_); +} + void ServerContext::AddInitialMetadata(const grpc::string& key, const grpc::string& value) { initial_metadata_.insert(std::make_pair(key, value));