include/grpcpp/impl/codegen/callback_common.h

   1 /*
   2  *
   3  * Copyright 2018 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #ifndef GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H
  20 #define GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H
  21
  22 #include <functional>
  23
  24 #include <grpc/impl/codegen/grpc_types.h>
  25 #include <grpcpp/impl/codegen/call.h>
  26 #include <grpcpp/impl/codegen/channel_interface.h>
  27 #include <grpcpp/impl/codegen/completion_queue_tag.h>
  28 #include <grpcpp/impl/codegen/config.h>
  29 #include <grpcpp/impl/codegen/core_codegen_interface.h>
  30 #include <grpcpp/impl/codegen/status.h>
  31
  32 namespace grpc {
  33 namespace internal {
  34
  35 /// An exception-safe way of invoking a user-specified callback function
  36 // TODO(vjpai): decide whether it is better for this to take a const lvalue
  37 //              parameter or an rvalue parameter, or if it even matters
  38 template <class Func, class... Args>
  39 void CatchingCallback(Func&& func, Args&&... args) {
  40 #if GRPC_ALLOW_EXCEPTIONS
  41   try {
  42     func(std::forward<Args>(args)...);
  43   } catch (...) {
  44     // nothing to return or change here, just don't crash the library
  45   }
  46 #else   // GRPC_ALLOW_EXCEPTIONS
  47   func(std::forward<Args>(args)...);
  48 #endif  // GRPC_ALLOW_EXCEPTIONS
  49 }
  50
  51 template <class Reactor, class Func, class... Args>
  52 Reactor* CatchingReactorGetter(Func&& func, Args&&... args) {
  53 #if GRPC_ALLOW_EXCEPTIONS
  54   try {
  55     return func(std::forward<Args>(args)...);
  56   } catch (...) {
  57     // fail the RPC, don't crash the library
  58     return nullptr;
  59   }
  60 #else   // GRPC_ALLOW_EXCEPTIONS
  61   return func(std::forward<Args>(args)...);
  62 #endif  // GRPC_ALLOW_EXCEPTIONS
  63 }
  64
  65 // The contract on these tags is that they are single-shot. They must be
  66 // constructed and then fired at exactly one point. There is no expectation
  67 // that they can be reused without reconstruction.
  68
  69 class CallbackWithStatusTag : public grpc_completion_queue_functor {
  70  public:
  71   // always allocated against a call arena, no memory free required
  72   static void operator delete(void* /*ptr*/, std::size_t size) {
  73     GPR_CODEGEN_ASSERT(size == sizeof(CallbackWithStatusTag));
  74   }
  75
  76   // This operator should never be called as the memory should be freed as part
  77   // of the arena destruction. It only exists to provide a matching operator
  78   // delete to the operator new so that some compilers will not complain (see
  79   // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
  80   // there are no tests catching the compiler warning.
  81   static void operator delete(void*, void*) { GPR_CODEGEN_ASSERT(false); }
  82
  83   CallbackWithStatusTag(grpc_call* call, std::function<void(Status)> f,
  84                         CompletionQueueTag* ops)
  85       : call_(call), func_(std::move(f)), ops_(ops) {
  86     g_core_codegen_interface->grpc_call_ref(call);
  87     functor_run = &CallbackWithStatusTag::StaticRun;
  88     // A client-side callback should never be run inline since they will always
  89     // have work to do from the user application. So, set the parent's
  90     // inlineable field to false
  91     inlineable = false;
  92   }
  93   ~CallbackWithStatusTag() {}
  94   Status* status_ptr() { return &status_; }
  95
  96   // force_run can not be performed on a tag if operations using this tag
  97   // have been sent to PerformOpsOnCall. It is intended for error conditions
  98   // that are detected before the operations are internally processed.
  99   void force_run(Status s) {
 100     status_ = std::move(s);
 101     Run(true);
 102   }
 103
 104  private:
 105   grpc_call* call_;
 106   std::function<void(Status)> func_;
 107   CompletionQueueTag* ops_;
 108   Status status_;
 109
 110   static void StaticRun(grpc_completion_queue_functor* cb, int ok) {
 111     static_cast<CallbackWithStatusTag*>(cb)->Run(static_cast<bool>(ok));
 112   }
 113   void Run(bool ok) {
 114     void* ignored = ops_;
 115
 116     if (!ops_->FinalizeResult(&ignored, &ok)) {
 117       // The tag was swallowed
 118       return;
 119     }
 120     GPR_CODEGEN_ASSERT(ignored == ops_);
 121
 122     // Last use of func_ or status_, so ok to move them out
 123     auto func = std::move(func_);
 124     auto status = std::move(status_);
 125     func_ = nullptr;     // reset to clear this out for sure
 126     status_ = Status();  // reset to clear this out for sure
 127     CatchingCallback(std::move(func), std::move(status));
 128     g_core_codegen_interface->grpc_call_unref(call_);
 129   }
 130 };
 131
 132 /// CallbackWithSuccessTag can be reused multiple times, and will be used in
 133 /// this fashion for streaming operations. As a result, it shouldn't clear
 134 /// anything up until its destructor
 135 class CallbackWithSuccessTag : public grpc_completion_queue_functor {
 136  public:
 137   // always allocated against a call arena, no memory free required
 138   static void operator delete(void* /*ptr*/, std::size_t size) {
 139     GPR_CODEGEN_ASSERT(size == sizeof(CallbackWithSuccessTag));
 140   }
 141
 142   // This operator should never be called as the memory should be freed as part
 143   // of the arena destruction. It only exists to provide a matching operator
 144   // delete to the operator new so that some compilers will not complain (see
 145   // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
 146   // there are no tests catching the compiler warning.
 147   static void operator delete(void*, void*) { GPR_CODEGEN_ASSERT(false); }
 148
 149   CallbackWithSuccessTag() : call_(nullptr) {}
 150
 151   CallbackWithSuccessTag(const CallbackWithSuccessTag&) = delete;
 152   CallbackWithSuccessTag& operator=(const CallbackWithSuccessTag&) = delete;
 153
 154   ~CallbackWithSuccessTag() { Clear(); }
 155
 156   // Set can only be called on a default-constructed or Clear'ed tag.
 157   // It should never be called on a tag that was constructed with arguments
 158   // or on a tag that has been Set before unless the tag has been cleared.
 159   // can_inline indicates that this particular callback can be executed inline
 160   // (without needing a thread hop) and is only used for library-provided server
 161   // callbacks.
 162   void Set(grpc_call* call, std::function<void(bool)> f,
 163            CompletionQueueTag* ops, bool can_inline) {
 164     GPR_CODEGEN_ASSERT(call_ == nullptr);
 165     g_core_codegen_interface->grpc_call_ref(call);
 166     call_ = call;
 167     func_ = std::move(f);
 168     ops_ = ops;
 169     functor_run = &CallbackWithSuccessTag::StaticRun;
 170     inlineable = can_inline;
 171   }
 172
 173   void Clear() {
 174     if (call_ != nullptr) {
 175       grpc_call* call = call_;
 176       call_ = nullptr;
 177       func_ = nullptr;
 178       g_core_codegen_interface->grpc_call_unref(call);
 179     }
 180   }
 181
 182   CompletionQueueTag* ops() { return ops_; }
 183
 184   // force_run can not be performed on a tag if operations using this tag
 185   // have been sent to PerformOpsOnCall. It is intended for error conditions
 186   // that are detected before the operations are internally processed.
 187   void force_run(bool ok) { Run(ok); }
 188
 189   /// check if this tag is currently set
 190   /* NOLINTNEXTLINE(google-explicit-constructor) */
 191   operator bool() const { return call_ != nullptr; }
 192
 193  private:
 194   grpc_call* call_;
 195   std::function<void(bool)> func_;
 196   CompletionQueueTag* ops_;
 197
 198   static void StaticRun(grpc_completion_queue_functor* cb, int ok) {
 199     static_cast<CallbackWithSuccessTag*>(cb)->Run(static_cast<bool>(ok));
 200   }
 201   void Run(bool ok) {
 202     void* ignored = ops_;
 203     // Allow a "false" return value from FinalizeResult to silence the
 204     // callback, just as it silences a CQ tag in the async cases
 205 #ifndef NDEBUG
 206     auto* ops = ops_;
 207 #endif
 208     bool do_callback = ops_->FinalizeResult(&ignored, &ok);
 209     GPR_CODEGEN_DEBUG_ASSERT(ignored == ops);
 210
 211     if (do_callback) {
 212       CatchingCallback(func_, ok);
 213     }
 214   }
 215 };
 216
 217 }  // namespace internal
 218 }  // namespace grpc
 219
 220 #endif  // GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H