Line data    Source code

       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/config.h>
      28             : #include <grpcpp/impl/codegen/core_codegen_interface.h>
      29             : #include <grpcpp/impl/codegen/status.h>
      30             : 
      31             : namespace grpc {
      32             : namespace internal {
      33             : 
      34             : /// An exception-safe way of invoking a user-specified callback function
      35             : // TODO(vjpai): decide whether it is better for this to take a const lvalue
      36             : //              parameter or an rvalue parameter, or if it even matters
      37             : template <class Func, class... Args>
      38           0 : void CatchingCallback(Func&& func, Args&&... args) {
      39             : #if GRPC_ALLOW_EXCEPTIONS
      40             :   try {
      41           0 :     func(std::forward<Args>(args)...);
      42           0 :   } catch (...) {
      43             :     // nothing to return or change here, just don't crash the library
      44             :   }
      45             : #else   // GRPC_ALLOW_EXCEPTIONS
      46             :   func(std::forward<Args>(args)...);
      47             : #endif  // GRPC_ALLOW_EXCEPTIONS
      48           0 : }
      49             : 
      50             : template <class ReturnType, class Func, class... Args>
      51             : ReturnType* CatchingReactorCreator(Func&& func, Args&&... args) {
      52             : #if GRPC_ALLOW_EXCEPTIONS
      53             :   try {
      54             :     return func(std::forward<Args>(args)...);
      55             :   } catch (...) {
      56             :     // fail the RPC, don't crash the library
      57             :     return nullptr;
      58             :   }
      59             : #else   // GRPC_ALLOW_EXCEPTIONS
      60             :   return func(std::forward<Args>(args)...);
      61             : #endif  // GRPC_ALLOW_EXCEPTIONS
      62             : }
      63             : 
      64             : // The contract on these tags is that they are single-shot. They must be
      65             : // constructed and then fired at exactly one point. There is no expectation
      66             : // that they can be reused without reconstruction.
      67             : 
      68             : class CallbackWithStatusTag
      69             :     : public grpc_experimental_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             :     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           0 :   static void operator delete(void*, void*) { assert(0); }
      82             : 
      83           0 :   CallbackWithStatusTag(grpc_call* call, std::function<void(Status)> f,
      84             :                         CompletionQueueTag* ops)
      85           0 :       : call_(call), func_(std::move(f)), ops_(ops) {
      86           0 :     g_core_codegen_interface->grpc_call_ref(call);
      87           0 :     functor_run = &CallbackWithStatusTag::StaticRun;
      88           0 :   }
      89             :   ~CallbackWithStatusTag() {}
      90           0 :   Status* status_ptr() { return &status_; }
      91             : 
      92             :   // force_run can not be performed on a tag if operations using this tag
      93             :   // have been sent to PerformOpsOnCall. It is intended for error conditions
      94             :   // that are detected before the operations are internally processed.
      95           0 :   void force_run(Status s) {
      96           0 :     status_ = std::move(s);
      97           0 :     Run(true);
      98           0 :   }
      99             : 
     100             :  private:
     101             :   grpc_call* call_;
     102             :   std::function<void(Status)> func_;
     103             :   CompletionQueueTag* ops_;
     104             :   Status status_;
     105             : 
     106           0 :   static void StaticRun(grpc_experimental_completion_queue_functor* cb,
     107             :                         int ok) {
     108           0 :     static_cast<CallbackWithStatusTag*>(cb)->Run(static_cast<bool>(ok));
     109           0 :   }
     110           0 :   void Run(bool ok) {
     111           0 :     void* ignored = ops_;
     112             : 
     113           0 :     if (!ops_->FinalizeResult(&ignored, &ok)) {
     114             :       // The tag was swallowed
     115           0 :       return;
     116             :     }
     117           0 :     GPR_CODEGEN_ASSERT(ignored == ops_);
     118             : 
     119             :     // Last use of func_ or status_, so ok to move them out
     120           0 :     auto func = std::move(func_);
     121           0 :     auto status = std::move(status_);
     122           0 :     func_ = nullptr;     // reset to clear this out for sure
     123           0 :     status_ = Status();  // reset to clear this out for sure
     124           0 :     CatchingCallback(std::move(func), std::move(status));
     125           0 :     g_core_codegen_interface->grpc_call_unref(call_);
     126             :   }
     127             : };
     128             : 
     129             : /// CallbackWithSuccessTag can be reused multiple times, and will be used in
     130             : /// this fashion for streaming operations. As a result, it shouldn't clear
     131             : /// anything up until its destructor
     132             : class CallbackWithSuccessTag
     133             :     : public grpc_experimental_completion_queue_functor {
     134             :  public:
     135             :   // always allocated against a call arena, no memory free required
     136             :   static void operator delete(void* ptr, std::size_t size) {
     137             :     assert(size == sizeof(CallbackWithSuccessTag));
     138             :   }
     139             : 
     140             :   // This operator should never be called as the memory should be freed as part
     141             :   // of the arena destruction. It only exists to provide a matching operator
     142             :   // delete to the operator new so that some compilers will not complain (see
     143             :   // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
     144             :   // there are no tests catching the compiler warning.
     145             :   static void operator delete(void*, void*) { assert(0); }
     146             : 
     147             :   CallbackWithSuccessTag() : call_(nullptr) {}
     148             : 
     149             :   CallbackWithSuccessTag(grpc_call* call, std::function<void(bool)> f,
     150             :                          CompletionQueueTag* ops) {
     151             :     Set(call, f, ops);
     152             :   }
     153             : 
     154             :   CallbackWithSuccessTag(const CallbackWithSuccessTag&) = delete;
     155             :   CallbackWithSuccessTag& operator=(const CallbackWithSuccessTag&) = delete;
     156             : 
     157             :   ~CallbackWithSuccessTag() { Clear(); }
     158             : 
     159             :   // Set can only be called on a default-constructed or Clear'ed tag.
     160             :   // It should never be called on a tag that was constructed with arguments
     161             :   // or on a tag that has been Set before unless the tag has been cleared.
     162             :   void Set(grpc_call* call, std::function<void(bool)> f,
     163             :            CompletionQueueTag* ops) {
     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             :   }
     171             : 
     172             :   void Clear() {
     173             :     if (call_ != nullptr) {
     174             :       grpc_call* call = call_;
     175             :       call_ = nullptr;
     176             :       func_ = nullptr;
     177             :       g_core_codegen_interface->grpc_call_unref(call);
     178             :     }
     179             :   }
     180             : 
     181             :   CompletionQueueTag* ops() { return ops_; }
     182             : 
     183             :   // force_run can not be performed on a tag if operations using this tag
     184             :   // have been sent to PerformOpsOnCall. It is intended for error conditions
     185             :   // that are detected before the operations are internally processed.
     186             :   void force_run(bool ok) { Run(ok); }
     187             : 
     188             :   /// check if this tag is currently set
     189             :   operator bool() const { return call_ != nullptr; }
     190             : 
     191             :  private:
     192             :   grpc_call* call_;
     193             :   std::function<void(bool)> func_;
     194             :   CompletionQueueTag* ops_;
     195             : 
     196             :   static void StaticRun(grpc_experimental_completion_queue_functor* cb,
     197             :                         int ok) {
     198             :     static_cast<CallbackWithSuccessTag*>(cb)->Run(static_cast<bool>(ok));
     199             :   }
     200             :   void Run(bool ok) {
     201             :     void* ignored = ops_;
     202             :     // Allow a "false" return value from FinalizeResult to silence the
     203             :     // callback, just as it silences a CQ tag in the async cases
     204             :     auto* ops = ops_;
     205             :     bool do_callback = ops_->FinalizeResult(&ignored, &ok);
     206             :     GPR_CODEGEN_ASSERT(ignored == ops);
     207             : 
     208             :     if (do_callback) {
     209             :       CatchingCallback(func_, ok);
     210             :     }
     211             :   }
     212             : };
     213             : 
     214             : }  // namespace internal
     215             : }  // namespace grpc
     216             : 
     217             : #endif  // GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H