diff options
| -rw-r--r-- | include/grpcpp/impl/codegen/callback_common.h | 37 | ||||
| -rw-r--r-- | include/grpcpp/impl/codegen/server_callback.h | 6 | ||||
| -rw-r--r-- | src/cpp/server/server_context.cc | 3 |
3 files changed, 31 insertions, 15 deletions
diff --git a/include/grpcpp/impl/codegen/callback_common.h b/include/grpcpp/impl/codegen/callback_common.h index 29deef658f..11f54295cd 100644 --- a/include/grpcpp/impl/codegen/callback_common.h +++ b/include/grpcpp/impl/codegen/callback_common.h @@ -110,6 +110,9 @@ class CallbackWithStatusTag } }; +/// CallbackWithSuccessTag can be reused multiple times, and will be used in +/// this fashion for streaming operations. As a result, it shouldn't clear +/// anything up until its destructor class CallbackWithSuccessTag : public grpc_experimental_completion_queue_functor { public: @@ -125,11 +128,31 @@ class CallbackWithSuccessTag // there are no tests catching the compiler warning. static void operator delete(void*, void*) { assert(0); } - CallbackWithSuccessTag() : call_(nullptr), ops_(nullptr) {} + CallbackWithSuccessTag() : call_(nullptr) {} CallbackWithSuccessTag(grpc_call* call, std::function<void(bool)> f, - CompletionQueueTag* ops) - : call_(call), func_(std::move(f)), ops_(ops) { + CompletionQueueTag* ops) { + Set(call, f, ops); + } + + CallbackWithSuccessTag(const CallbackWithSuccessTag&) = delete; + CallbackWithSuccessTag& operator=(const CallbackWithSuccessTag&) = delete; + + ~CallbackWithSuccessTag() { + if (call_ != nullptr) { + func_ = nullptr; + g_core_codegen_interface->grpc_call_unref(call_); + } + } + + // Set can only be called on a default-constructed tag, and it can only + // be called exactly once. It should never be called on a tag that was + // constructed with arguments or on a tag that has been Set before + void Set(grpc_call* call, std::function<void(bool)> f, + CompletionQueueTag* ops) { + call_ = call; + func_ = std::move(f); + ops_ = ops; g_core_codegen_interface->grpc_call_ref(call); functor_run = &CallbackWithSuccessTag::StaticRun; } @@ -162,14 +185,8 @@ class CallbackWithSuccessTag GPR_CODEGEN_ASSERT(ignored == ops_); if (do_callback) { - // Last use of func_, so ok to move it out for rvalue call above - auto func = std::move(func_); - func_ = nullptr; // reset to clear this out for sure - CatchingCallback(std::move(func), ok); - } else { - func_ = nullptr; // reset to clear this out for sure + CatchingCallback(func_, ok); } - g_core_codegen_interface->grpc_call_unref(call_); } }; diff --git a/include/grpcpp/impl/codegen/server_callback.h b/include/grpcpp/impl/codegen/server_callback.h index 5d56cbf1df..b866fc16dc 100644 --- a/include/grpcpp/impl/codegen/server_callback.h +++ b/include/grpcpp/impl/codegen/server_callback.h @@ -22,6 +22,7 @@ #include <functional> #include <grpcpp/impl/codegen/call.h> +#include <grpcpp/impl/codegen/call_op_set.h> #include <grpcpp/impl/codegen/callback_common.h> #include <grpcpp/impl/codegen/config.h> #include <grpcpp/impl/codegen/core_codegen_interface.h> @@ -116,7 +117,7 @@ class CallbackUnaryHandler : public MethodHandler { : public experimental::ServerCallbackRpcController { public: void Finish(Status s) override { - finish_tag_ = CallbackWithSuccessTag( + finish_tag_.Set( call_.call(), [this](bool) { grpc_call* call = call_.call(); @@ -149,8 +150,7 @@ class CallbackUnaryHandler : public MethodHandler { void SendInitialMetadata(std::function<void(bool)> f) override { GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); - meta_tag_ = - CallbackWithSuccessTag(call_.call(), std::move(f), &meta_buf_); + meta_tag_.Set(call_.call(), std::move(f), &meta_buf_); meta_buf_.SendInitialMetadata(&ctx_->initial_metadata_, ctx_->initial_metadata_flags()); if (ctx_->compression_level_set()) { diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc index 355debb3fb..c5707e2b97 100644 --- a/src/cpp/server/server_context.cc +++ b/src/cpp/server/server_context.cc @@ -270,8 +270,7 @@ void ServerContext::BeginCompletionOp(internal::Call* call, bool callback) { new (grpc_call_arena_alloc(call->call(), sizeof(CompletionOp))) CompletionOp(call); if (callback) { - completion_tag_ = - internal::CallbackWithSuccessTag(call->call(), nullptr, completion_op_); + completion_tag_.Set(call->call(), nullptr, completion_op_); completion_op_->set_core_cq_tag(&completion_tag_); } else if (has_notify_when_done_tag_) { completion_op_->set_tag(async_notify_when_done_tag_); |