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