The C based gRPC (C++, Python, Ruby, Objective-C, PHP, C#) https://grpc.io/
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

218 lines
7.4 KiB

/*
*
* Copyright 2018 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
#ifndef GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H
#define GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H
#include <functional>
#include <grpc/impl/codegen/grpc_types.h>
#include <grpcpp/impl/codegen/call.h>
#include <grpcpp/impl/codegen/channel_interface.h>
#include <grpcpp/impl/codegen/config.h>
#include <grpcpp/impl/codegen/core_codegen_interface.h>
#include <grpcpp/impl/codegen/status.h>
namespace grpc {
namespace internal {
/// An exception-safe way of invoking a user-specified callback function
// TODO(vjpai): decide whether it is better for this to take a const lvalue
// parameter or an rvalue parameter, or if it even matters
template <class Func, class... Args>
void CatchingCallback(Func&& func, Args&&... args) {
#if GRPC_ALLOW_EXCEPTIONS
try {
func(std::forward<Args>(args)...);
} catch (...) {
// nothing to return or change here, just don't crash the library
}
#else // GRPC_ALLOW_EXCEPTIONS
func(std::forward<Args>(args)...);
#endif // GRPC_ALLOW_EXCEPTIONS
}
template <class ReturnType, class Func, class... Args>
ReturnType* CatchingReactorCreator(Func&& func, Args&&... args) {
#if GRPC_ALLOW_EXCEPTIONS
try {
return func(std::forward<Args>(args)...);
} catch (...) {
// fail the RPC, don't crash the library
return nullptr;
}
#else // GRPC_ALLOW_EXCEPTIONS
return func(std::forward<Args>(args)...);
#endif // GRPC_ALLOW_EXCEPTIONS
}
// The contract on these tags is that they are single-shot. They must be
// constructed and then fired at exactly one point. There is no expectation
// that they can be reused without reconstruction.
class CallbackWithStatusTag
: public grpc_experimental_completion_queue_functor {
public:
// always allocated against a call arena, no memory free required
static void operator delete(void* ptr, std::size_t size) {
assert(size == sizeof(CallbackWithStatusTag));
}
// This operator should never be called as the memory should be freed as part
// of the arena destruction. It only exists to provide a matching operator
// delete to the operator new so that some compilers will not complain (see
// https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
// there are no tests catching the compiler warning.
static void operator delete(void*, void*) { assert(0); }
CallbackWithStatusTag(grpc_call* call, std::function<void(Status)> f,
CompletionQueueTag* ops)
: call_(call), func_(std::move(f)), ops_(ops) {
g_core_codegen_interface->grpc_call_ref(call);
functor_run = &CallbackWithStatusTag::StaticRun;
}
~CallbackWithStatusTag() {}
Status* status_ptr() { return &status_; }
// force_run can not be performed on a tag if operations using this tag
// have been sent to PerformOpsOnCall. It is intended for error conditions
// that are detected before the operations are internally processed.
void force_run(Status s) {
status_ = std::move(s);
Run(true);
}
private:
grpc_call* call_;
std::function<void(Status)> func_;
CompletionQueueTag* ops_;
Status status_;
static void StaticRun(grpc_experimental_completion_queue_functor* cb,
int ok) {
static_cast<CallbackWithStatusTag*>(cb)->Run(static_cast<bool>(ok));
}
void Run(bool ok) {
void* ignored = ops_;
if (!ops_->FinalizeResult(&ignored, &ok)) {
// The tag was swallowed
return;
}
GPR_CODEGEN_ASSERT(ignored == ops_);
// Last use of func_ or status_, so ok to move them out
auto func = std::move(func_);
auto status = std::move(status_);
func_ = nullptr; // reset to clear this out for sure
status_ = Status(); // reset to clear this out for sure
CatchingCallback(std::move(func), std::move(status));
g_core_codegen_interface->grpc_call_unref(call_);
}
};
/// 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:
// always allocated against a call arena, no memory free required
static void operator delete(void* ptr, std::size_t size) {
assert(size == sizeof(CallbackWithSuccessTag));
}
// This operator should never be called as the memory should be freed as part
// of the arena destruction. It only exists to provide a matching operator
// delete to the operator new so that some compilers will not complain (see
// https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
// there are no tests catching the compiler warning.
static void operator delete(void*, void*) { assert(0); }
CallbackWithSuccessTag() : call_(nullptr) {}
CallbackWithSuccessTag(grpc_call* call, std::function<void(bool)> f,
CompletionQueueTag* ops) {
Set(call, f, ops);
}
CallbackWithSuccessTag(const CallbackWithSuccessTag&) = delete;
CallbackWithSuccessTag& operator=(const CallbackWithSuccessTag&) = delete;
~CallbackWithSuccessTag() { Clear(); }
// Set can only be called on a default-constructed or Clear'ed tag.
// It should never be called on a tag that was constructed with arguments
// or on a tag that has been Set before unless the tag has been cleared.
void Set(grpc_call* call, std::function<void(bool)> f,
CompletionQueueTag* ops) {
GPR_CODEGEN_ASSERT(call_ == nullptr);
g_core_codegen_interface->grpc_call_ref(call);
call_ = call;
func_ = std::move(f);
ops_ = ops;
functor_run = &CallbackWithSuccessTag::StaticRun;
}
void Clear() {
if (call_ != nullptr) {
grpc_call* call = call_;
call_ = nullptr;
func_ = nullptr;
g_core_codegen_interface->grpc_call_unref(call);
}
}
CompletionQueueTag* ops() { return ops_; }
// force_run can not be performed on a tag if operations using this tag
// have been sent to PerformOpsOnCall. It is intended for error conditions
// that are detected before the operations are internally processed.
void force_run(bool ok) { Run(ok); }
/// check if this tag is currently set
operator bool() const { return call_ != nullptr; }
private:
grpc_call* call_;
std::function<void(bool)> func_;
CompletionQueueTag* ops_;
static void StaticRun(grpc_experimental_completion_queue_functor* cb,
int ok) {
static_cast<CallbackWithSuccessTag*>(cb)->Run(static_cast<bool>(ok));
}
void Run(bool ok) {
void* ignored = ops_;
// Allow a "false" return value from FinalizeResult to silence the
// callback, just as it silences a CQ tag in the async cases
auto* ops = ops_;
bool do_callback = ops_->FinalizeResult(&ignored, &ok);
GPR_CODEGEN_ASSERT(ignored == ops);
if (do_callback) {
CatchingCallback(func_, ok);
}
}
};
} // namespace internal
} // namespace grpc
#endif // GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H