GRPC C++  1.39.1
callback_common.h
Go to the documentation of this file.
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 
31 
32 namespace grpc {
33 namespace internal {
34 
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 
70  public:
71  // always allocated against a call arena, no memory free required
72  static void operator delete(void* /*ptr*/, std::size_t size) {
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) {
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  }
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));
129  }
130 };
131 
136  public:
137  // always allocated against a call arena, no memory free required
138  static void operator delete(void* /*ptr*/, std::size_t size) {
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 
153 
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);
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;
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 
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
virtual void grpc_call_unref(grpc_call *call)=0
virtual void grpc_call_ref(grpc_call *call)=0
Did it work? If it didn't, why?
Definition: status.h:31
Definition: callback_common.h:69
CallbackWithStatusTag(grpc_call *call, std::function< void(Status)> f, CompletionQueueTag *ops)
Definition: callback_common.h:83
~CallbackWithStatusTag()
Definition: callback_common.h:93
void force_run(Status s)
Definition: callback_common.h:99
Status * status_ptr()
Definition: callback_common.h:94
CallbackWithSuccessTag can be reused multiple times, and will be used in this fashion for streaming o...
Definition: callback_common.h:135
CallbackWithSuccessTag & operator=(const CallbackWithSuccessTag &)=delete
void Set(grpc_call *call, std::function< void(bool)> f, CompletionQueueTag *ops, bool can_inline)
Definition: callback_common.h:162
void force_run(bool ok)
Definition: callback_common.h:187
~CallbackWithSuccessTag()
Definition: callback_common.h:154
CallbackWithSuccessTag(const CallbackWithSuccessTag &)=delete
CallbackWithSuccessTag()
Definition: callback_common.h:149
void Clear()
Definition: callback_common.h:173
CompletionQueueTag * ops()
Definition: callback_common.h:182
An interface allowing implementors to process and filter event tags.
Definition: completion_queue_tag.h:26
virtual bool FinalizeResult(void **tag, bool *status)=0
FinalizeResult must be called before informing user code that the operation bound to the underlying c...
struct grpc_call grpc_call
A Call represents an RPC.
Definition: grpc_types.h:70
#define GPR_CODEGEN_ASSERT(x)
Codegen specific version of GPR_ASSERT.
Definition: core_codegen_interface.h:146
#define GPR_CODEGEN_DEBUG_ASSERT(x)
Codegen specific version of GPR_DEBUG_ASSERT.
Definition: core_codegen_interface.h:155
void CatchingCallback(Func &&func, Args &&... args)
An exception-safe way of invoking a user-specified callback function.
Definition: callback_common.h:39
Reactor * CatchingReactorGetter(Func &&func, Args &&... args)
Definition: callback_common.h:52
::google::protobuf::util::Status Status
Definition: config_protobuf.h:91
An Alarm posts the user-provided tag to its associated completion queue or invokes the user-provided ...
Definition: alarm.h:33
CoreCodegenInterface * g_core_codegen_interface
Definition: completion_queue.h:96
Definition: async_unary_call.h:398
Specifies an interface class to be used as a tag for callback-based completion queues.
Definition: grpc_types.h:757
int inlineable
The inlineable member specifies whether this functor can be run inline.
Definition: grpc_types.h:766
void(* functor_run)(struct grpc_completion_queue_functor *, int)
The run member specifies a function that will be called when this tag is extracted from the completio...
Definition: grpc_types.h:762