GRPC C++  1.26.0
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 
30 
31 namespace grpc {
32 namespace internal {
33 
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 void CatchingCallback(Func&& func, Args&&... args) {
39 #if GRPC_ALLOW_EXCEPTIONS
40  try {
41  func(std::forward<Args>(args)...);
42  } 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 }
49 
50 template <class Reactor, class Func, class... Args>
51 Reactor* CatchingReactorGetter(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 
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_experimental_completion_queue_functor* cb,
111  int ok) {
112  static_cast<CallbackWithStatusTag*>(cb)->Run(static_cast<bool>(ok));
113  }
114  void Run(bool ok) {
115  void* ignored = ops_;
116 
117  if (!ops_->FinalizeResult(&ignored, &ok)) {
118  // The tag was swallowed
119  return;
120  }
121  GPR_CODEGEN_ASSERT(ignored == ops_);
122 
123  // Last use of func_ or status_, so ok to move them out
124  auto func = std::move(func_);
125  auto status = std::move(status_);
126  func_ = nullptr; // reset to clear this out for sure
127  status_ = Status(); // reset to clear this out for sure
128  CatchingCallback(std::move(func), std::move(status));
130  }
131 };
132 
138  public:
139  // always allocated against a call arena, no memory free required
140  static void operator delete(void* /*ptr*/, std::size_t size) {
142  }
143 
144  // This operator should never be called as the memory should be freed as part
145  // of the arena destruction. It only exists to provide a matching operator
146  // delete to the operator new so that some compilers will not complain (see
147  // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
148  // there are no tests catching the compiler warning.
149  static void operator delete(void*, void*) { GPR_CODEGEN_ASSERT(false); }
150 
151  CallbackWithSuccessTag() : call_(nullptr) {}
152 
153  CallbackWithSuccessTag(grpc_call* call, std::function<void(bool)> f,
154  CompletionQueueTag* ops, bool can_inline) {
155  Set(call, f, ops, can_inline);
156  }
157 
160 
162 
163  // Set can only be called on a default-constructed or Clear'ed tag.
164  // It should never be called on a tag that was constructed with arguments
165  // or on a tag that has been Set before unless the tag has been cleared.
166  // can_inline indicates that this particular callback can be executed inline
167  // (without needing a thread hop) and is only used for library-provided server
168  // callbacks.
169  void Set(grpc_call* call, std::function<void(bool)> f,
170  CompletionQueueTag* ops, bool can_inline) {
171  GPR_CODEGEN_ASSERT(call_ == nullptr);
173  call_ = call;
174  func_ = std::move(f);
175  ops_ = ops;
176  functor_run = &CallbackWithSuccessTag::StaticRun;
177  inlineable = can_inline;
178  }
179 
180  void Clear() {
181  if (call_ != nullptr) {
182  grpc_call* call = call_;
183  call_ = nullptr;
184  func_ = nullptr;
186  }
187  }
188 
189  CompletionQueueTag* ops() { return ops_; }
190 
191  // force_run can not be performed on a tag if operations using this tag
192  // have been sent to PerformOpsOnCall. It is intended for error conditions
193  // that are detected before the operations are internally processed.
194  void force_run(bool ok) { Run(ok); }
195 
197  operator bool() const { return call_ != nullptr; }
198 
199  private:
200  grpc_call* call_;
201  std::function<void(bool)> func_;
202  CompletionQueueTag* ops_;
203 
204  static void StaticRun(grpc_experimental_completion_queue_functor* cb,
205  int ok) {
206  static_cast<CallbackWithSuccessTag*>(cb)->Run(static_cast<bool>(ok));
207  }
208  void Run(bool ok) {
209  void* ignored = ops_;
210  // Allow a "false" return value from FinalizeResult to silence the
211  // callback, just as it silences a CQ tag in the async cases
212 #ifndef NDEBUG
213  auto* ops = ops_;
214 #endif
215  bool do_callback = ops_->FinalizeResult(&ignored, &ok);
216  GPR_CODEGEN_DEBUG_ASSERT(ignored == ops);
217 
218  if (do_callback) {
219  CatchingCallback(func_, ok);
220  }
221  }
222 };
223 
224 } // namespace internal
225 } // namespace grpc
226 
227 #endif // GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H
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
void force_run(bool ok)
Definition: callback_common.h:194
virtual void grpc_call_ref(grpc_call *call)=0
int inlineable
The inlineable member specifies whether this functor can be run inline.
Definition: grpc_types.h:731
An interface allowing implementors to process and filter event tags.
Definition: completion_queue_tag.h:26
Reactor * CatchingReactorGetter(Func &&func, Args &&... args)
Definition: callback_common.h:51
virtual bool FinalizeResult(void **tag, bool *status)=0
FinalizeResult must be called before informing user code that the operation bound to the underlying c...
virtual void grpc_call_unref(grpc_call *call)=0
CompletionQueueTag * ops()
Definition: callback_common.h:189
Definition: async_unary_call_impl.h:301
::google::protobuf::util::Status Status
Definition: config_protobuf.h:90
void Set(grpc_call *call, std::function< void(bool)> f, CompletionQueueTag *ops, bool can_inline)
Definition: callback_common.h:169
~CallbackWithSuccessTag()
Definition: callback_common.h:161
CallbackWithSuccessTag(grpc_call *call, std::function< void(bool)> f, CompletionQueueTag *ops, bool can_inline)
Definition: callback_common.h:153
CallbackWithStatusTag(grpc_call *call, std::function< void(Status)> f, CompletionQueueTag *ops)
Definition: callback_common.h:83
Status * status_ptr()
Definition: callback_common.h:94
This header provides an object that reads bytes directly from a grpc::ByteBuffer, via the ZeroCopyInp...
Definition: alarm.h:24
void force_run(Status s)
Definition: callback_common.h:99
~CallbackWithStatusTag()
Definition: callback_common.h:93
CoreCodegenInterface * g_core_codegen_interface
Definition: completion_queue_impl.h:90
CallbackWithSuccessTag & operator=(const CallbackWithSuccessTag &)=delete
CallbackWithSuccessTag can be reused multiple times, and will be used in this fashion for streaming o...
Definition: callback_common.h:136
Did it work? If it didn&#39;t, why?
Definition: status.h:31
void Clear()
Definition: callback_common.h:180
void CatchingCallback(Func &&func, Args &&... args)
An exception-safe way of invoking a user-specified callback function.
Definition: callback_common.h:38
Definition: callback_common.h:68
void(* functor_run)(struct grpc_experimental_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:727
EXPERIMENTAL: Specifies an interface class to be used as a tag for callback-based completion queues...
Definition: grpc_types.h:722
#define GPR_CODEGEN_DEBUG_ASSERT(x)
Codegen specific version of GPR_DEBUG_ASSERT.
Definition: core_codegen_interface.h:155
CallbackWithSuccessTag()
Definition: callback_common.h:151