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 | 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 | |
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 | 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(ok: 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(ok: static_cast<bool>(ok)); |
113 | } |
114 | void Run(bool ok) { |
115 | void* ignored = ops_; |
116 | |
117 | if (!ops_->FinalizeResult(tag: &ignored, status: &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(func: std::move(func), args: std::move(status)); |
129 | g_core_codegen_interface->grpc_call_unref(call: call_); |
130 | } |
131 | }; |
132 | |
133 | /// CallbackWithSuccessTag can be reused multiple times, and will be used in |
134 | /// this fashion for streaming operations. As a result, it shouldn't clear |
135 | /// anything up until its destructor |
136 | class CallbackWithSuccessTag |
137 | : public grpc_experimental_completion_queue_functor { |
138 | public: |
139 | // always allocated against a call arena, no memory free required |
140 | static void operator delete(void* /*ptr*/, std::size_t size) { |
141 | GPR_CODEGEN_ASSERT(size == sizeof(CallbackWithSuccessTag)); |
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(const CallbackWithSuccessTag&) = delete; |
154 | CallbackWithSuccessTag& operator=(const CallbackWithSuccessTag&) = delete; |
155 | |
156 | ~CallbackWithSuccessTag() { Clear(); } |
157 | |
158 | // Set can only be called on a default-constructed or Clear'ed tag. |
159 | // It should never be called on a tag that was constructed with arguments |
160 | // or on a tag that has been Set before unless the tag has been cleared. |
161 | // can_inline indicates that this particular callback can be executed inline |
162 | // (without needing a thread hop) and is only used for library-provided server |
163 | // callbacks. |
164 | void Set(grpc_call* call, std::function<void(bool)> f, |
165 | CompletionQueueTag* ops, bool can_inline) { |
166 | GPR_CODEGEN_ASSERT(call_ == nullptr); |
167 | g_core_codegen_interface->grpc_call_ref(call); |
168 | call_ = call; |
169 | func_ = std::move(f); |
170 | ops_ = ops; |
171 | functor_run = &CallbackWithSuccessTag::StaticRun; |
172 | inlineable = can_inline; |
173 | } |
174 | |
175 | void Clear() { |
176 | if (call_ != nullptr) { |
177 | grpc_call* call = call_; |
178 | call_ = nullptr; |
179 | func_ = nullptr; |
180 | g_core_codegen_interface->grpc_call_unref(call); |
181 | } |
182 | } |
183 | |
184 | CompletionQueueTag* ops() { return ops_; } |
185 | |
186 | // force_run can not be performed on a tag if operations using this tag |
187 | // have been sent to PerformOpsOnCall. It is intended for error conditions |
188 | // that are detected before the operations are internally processed. |
189 | void force_run(bool ok) { Run(ok); } |
190 | |
191 | /// check if this tag is currently set |
192 | operator bool() const { return call_ != nullptr; } |
193 | |
194 | private: |
195 | grpc_call* call_; |
196 | std::function<void(bool)> func_; |
197 | CompletionQueueTag* ops_; |
198 | |
199 | static void StaticRun(grpc_experimental_completion_queue_functor* cb, |
200 | int ok) { |
201 | static_cast<CallbackWithSuccessTag*>(cb)->Run(ok: static_cast<bool>(ok)); |
202 | } |
203 | void Run(bool ok) { |
204 | void* ignored = ops_; |
205 | // Allow a "false" return value from FinalizeResult to silence the |
206 | // callback, just as it silences a CQ tag in the async cases |
207 | #ifndef NDEBUG |
208 | auto* ops = ops_; |
209 | #endif |
210 | bool do_callback = ops_->FinalizeResult(tag: &ignored, status: &ok); |
211 | GPR_CODEGEN_DEBUG_ASSERT(ignored == ops); |
212 | |
213 | if (do_callback) { |
214 | CatchingCallback(func&: func_, args&: ok); |
215 | } |
216 | } |
217 | }; |
218 | |
219 | } // namespace internal |
220 | } // namespace grpc |
221 | |
222 | #endif // GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H |
223 | |