GRPC C++  1.62.0
async_unary_call.h
Go to the documentation of this file.
1 //
2 //
3 // Copyright 2015 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_ASYNC_UNARY_CALL_H
20 #define GRPCPP_SUPPORT_ASYNC_UNARY_CALL_H
21 
22 #include <grpc/grpc.h>
23 #include <grpc/support/log.h>
24 #include <grpcpp/client_context.h>
25 #include <grpcpp/impl/call.h>
30 #include <grpcpp/server_context.h>
31 #include <grpcpp/support/status.h>
32 
33 namespace grpc {
34 
35 // Forward declaration for use in Helper class
36 template <class R>
37 class ClientAsyncResponseReader;
38 
41 template <class R>
43  public:
45 
49  virtual void StartCall() = 0;
50 
57  virtual void ReadInitialMetadata(void* tag) = 0;
58 
73  virtual void Finish(R* msg, grpc::Status* status, void* tag) = 0;
74 };
75 
76 namespace internal {
77 
79  public:
92  template <class R, class W, class BaseR = R, class BaseW = W>
95  const grpc::internal::RpcMethod& method, grpc::ClientContext* context,
96  const W& request) /* __attribute__((noinline)) */ {
97  grpc::internal::Call call = channel->CreateCall(method, context, cq);
99  call.call(), sizeof(ClientAsyncResponseReader<R>)))
100  ClientAsyncResponseReader<R>(call, context);
101  SetupRequest<BaseR, BaseW>(
102  call.call(), &result->single_buf_, &result->read_initial_metadata_,
103  &result->finish_, static_cast<const BaseW&>(request));
104 
105  return result;
106  }
107 
108  // Various helper functions to reduce templating use
109 
110  template <class R, class W>
111  static void SetupRequest(
112  grpc_call* call,
114  std::function<void(ClientContext*, internal::Call*,
116  read_initial_metadata,
117  std::function<
118  void(ClientContext*, internal::Call*, bool initial_metadata_read,
120  internal::CallOpSetInterface**, void*, Status*, void*)>* finish,
121  const W& request) {
122  using SingleBufType =
129  SingleBufType* single_buf =
130  new (grpc_call_arena_alloc(call, sizeof(SingleBufType))) SingleBufType;
131  *single_buf_ptr = single_buf;
132  // TODO(ctiller): don't assert
133  GPR_ASSERT(single_buf->SendMessage(request).ok());
134  single_buf->ClientSendClose();
135 
136  // The purpose of the following functions is to type-erase the actual
137  // templated type of the CallOpSet being used by hiding that type inside the
138  // function definition rather than specifying it as an argument of the
139  // function or a member of the class. The type-erased CallOpSet will get
140  // static_cast'ed back to the real type so that it can be used properly.
141  *read_initial_metadata =
142  [](ClientContext* context, internal::Call* call,
143  internal::CallOpSendInitialMetadata* single_buf_view, void* tag) {
144  auto* single_buf = static_cast<SingleBufType*>(single_buf_view);
145  single_buf->set_output_tag(tag);
146  single_buf->RecvInitialMetadata(context);
147  call->PerformOps(single_buf);
148  };
149 
150  // Note that this function goes one step further than the previous one
151  // because it type-erases the message being written down to a void*. This
152  // will be static-cast'ed back to the class specified here by hiding that
153  // class information inside the function definition. Note that this feature
154  // expects the class being specified here for R to be a base-class of the
155  // "real" R without any multiple-inheritance (as applies in protobuf wrt
156  // MessageLite)
157  *finish = [](ClientContext* context, internal::Call* call,
158  bool initial_metadata_read,
159  internal::CallOpSendInitialMetadata* single_buf_view,
160  internal::CallOpSetInterface** finish_buf_ptr, void* msg,
161  Status* status, void* tag) {
162  if (initial_metadata_read) {
163  using FinishBufType =
166  FinishBufType* finish_buf =
167  new (grpc_call_arena_alloc(call->call(), sizeof(FinishBufType)))
168  FinishBufType;
169  *finish_buf_ptr = finish_buf;
170  finish_buf->set_output_tag(tag);
171  finish_buf->RecvMessage(static_cast<R*>(msg));
172  finish_buf->AllowNoMessage();
173  finish_buf->ClientRecvStatus(context, status);
174  call->PerformOps(finish_buf);
175  } else {
176  auto* single_buf = static_cast<SingleBufType*>(single_buf_view);
177  single_buf->set_output_tag(tag);
178  single_buf->RecvInitialMetadata(context);
179  single_buf->RecvMessage(static_cast<R*>(msg));
180  single_buf->AllowNoMessage();
181  single_buf->ClientRecvStatus(context, status);
182  call->PerformOps(single_buf);
183  }
184  };
185  }
186 
187  static void StartCall(grpc::ClientContext* context,
189  single_buf->SendInitialMetadata(&context->send_initial_metadata_,
190  context->initial_metadata_flags());
191  }
192 };
193 
194 // TODO(vjpai): This templated factory is deprecated and will be replaced by
195 //. the non-templated helper as soon as possible.
196 template <class R>
198  public:
199  template <class W>
202  const grpc::internal::RpcMethod& method, grpc::ClientContext* context,
203  const W& request, bool start) {
204  auto* result = ClientAsyncResponseReaderHelper::Create<R>(
205  channel, cq, method, context, request);
206  if (start) {
207  result->StartCall();
208  }
209  return result;
210  }
211 };
212 
213 } // namespace internal
214 
217 template <class R>
218 class ClientAsyncResponseReader final
219  : public ClientAsyncResponseReaderInterface<R> {
220  public:
221  // always allocated against a call arena, no memory free required
222  static void operator delete(void* /*ptr*/, std::size_t size) {
223  GPR_ASSERT(size == sizeof(ClientAsyncResponseReader));
224  }
225 
226  // This operator should never be called as the memory should be freed as part
227  // of the arena destruction. It only exists to provide a matching operator
228  // delete to the operator new so that some compilers will not complain (see
229  // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
230  // there are no tests catching the compiler warning.
231  static void operator delete(void*, void*) { GPR_ASSERT(false); }
232 
233  void StartCall() override {
234  GPR_DEBUG_ASSERT(!started_);
235  started_ = true;
237  }
238 
245  void ReadInitialMetadata(void* tag) override {
246  GPR_DEBUG_ASSERT(started_);
247  GPR_DEBUG_ASSERT(!context_->initial_metadata_received_);
248  read_initial_metadata_(context_, &call_, single_buf_, tag);
249  initial_metadata_read_ = true;
250  }
251 
257  void Finish(R* msg, grpc::Status* status, void* tag) override {
258  GPR_DEBUG_ASSERT(started_);
259  finish_(context_, &call_, initial_metadata_read_, single_buf_, &finish_buf_,
260  static_cast<void*>(msg), status, tag);
261  }
262 
263  private:
265  grpc::ClientContext* const context_;
266  grpc::internal::Call call_;
267  bool started_ = false;
268  bool initial_metadata_read_ = false;
269 
271  grpc::ClientContext* context)
272  : context_(context), call_(call) {}
273 
274  // disable operator new
275  static void* operator new(std::size_t size);
276  static void* operator new(std::size_t /*size*/, void* p) { return p; }
277 
278  internal::CallOpSendInitialMetadata* single_buf_;
279  internal::CallOpSetInterface* finish_buf_ = nullptr;
280  std::function<void(ClientContext*, internal::Call*,
281  internal::CallOpSendInitialMetadata*, void*)>
282  read_initial_metadata_;
283  std::function<void(ClientContext*, internal::Call*,
284  bool initial_metadata_read,
285  internal::CallOpSendInitialMetadata*,
286  internal::CallOpSetInterface**, void*, Status*, void*)>
287  finish_;
288 };
289 
292 template <class W>
293 class ServerAsyncResponseWriter final
295  public:
297  : call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
298 
306  void SendInitialMetadata(void* tag) override {
307  GPR_ASSERT(!ctx_->sent_initial_metadata_);
308 
309  meta_buf_.set_output_tag(tag);
310  meta_buf_.SendInitialMetadata(&ctx_->initial_metadata_,
311  ctx_->initial_metadata_flags());
312  if (ctx_->compression_level_set()) {
313  meta_buf_.set_compression_level(ctx_->compression_level());
314  }
315  ctx_->sent_initial_metadata_ = true;
316  call_.PerformOps(&meta_buf_);
317  }
318 
338  void Finish(const W& msg, const grpc::Status& status, void* tag) {
339  finish_buf_.set_output_tag(tag);
340  finish_buf_.set_core_cq_tag(&finish_buf_);
341  if (!ctx_->sent_initial_metadata_) {
342  finish_buf_.SendInitialMetadata(&ctx_->initial_metadata_,
343  ctx_->initial_metadata_flags());
344  if (ctx_->compression_level_set()) {
345  finish_buf_.set_compression_level(ctx_->compression_level());
346  }
347  ctx_->sent_initial_metadata_ = true;
348  }
349  // The response is dropped if the status is not OK.
350  if (status.ok()) {
351  finish_buf_.ServerSendStatus(&ctx_->trailing_metadata_,
352  finish_buf_.SendMessage(msg));
353  } else {
354  finish_buf_.ServerSendStatus(&ctx_->trailing_metadata_, status);
355  }
356  call_.PerformOps(&finish_buf_);
357  }
358 
375  void FinishWithError(const grpc::Status& status, void* tag) {
376  GPR_ASSERT(!status.ok());
377  finish_buf_.set_output_tag(tag);
378  if (!ctx_->sent_initial_metadata_) {
379  finish_buf_.SendInitialMetadata(&ctx_->initial_metadata_,
380  ctx_->initial_metadata_flags());
381  if (ctx_->compression_level_set()) {
382  finish_buf_.set_compression_level(ctx_->compression_level());
383  }
384  ctx_->sent_initial_metadata_ = true;
385  }
386  finish_buf_.ServerSendStatus(&ctx_->trailing_metadata_, status);
387  call_.PerformOps(&finish_buf_);
388  }
389 
390  private:
391  void BindCall(grpc::internal::Call* call) override { call_ = *call; }
392 
393  grpc::internal::Call call_;
394  grpc::ServerContext* ctx_;
396  meta_buf_;
400  finish_buf_;
401 };
402 
403 } // namespace grpc
404 
405 namespace std {
406 template <class R>
407 class default_delete<grpc::ClientAsyncResponseReader<R>> {
408  public:
409  void operator()(void* /*p*/) {}
410 };
411 template <class R>
412 class default_delete<grpc::ClientAsyncResponseReaderInterface<R>> {
413  public:
414  void operator()(void* /*p*/) {}
415 };
416 } // namespace std
417 
418 #endif // GRPCPP_SUPPORT_ASYNC_UNARY_CALL_H
grpc::ClientAsyncResponseReader::StartCall
void StartCall() override
Definition: async_unary_call.h:233
grpc::internal::CallOpRecvInitialMetadata
Definition: call_op_set.h:721
grpc::internal::ServerAsyncStreamingInterface
Definition: service_type.h:37
grpc::internal::CallOpClientSendClose
Definition: call_op_set.h:619
grpc_call_arena_alloc
GRPCAPI void * grpc_call_arena_alloc(grpc_call *call, size_t size)
Allocate memory in the grpc_call arena: this memory is automatically discarded at call completion.
grpc::ServerContext
A ServerContext or CallbackServerContext allows the code implementing a service handler to:
Definition: server_context.h:572
grpc::internal::CallOpServerSendStatus
Definition: call_op_set.h:654
grpc
An Alarm posts the user-provided tag to its associated completion queue or invokes the user-provided ...
Definition: alarm.h:33
grpc::internal::CallOpSet
Primary implementation of CallOpSetInterface.
Definition: completion_queue.h:96
GPR_DEBUG_ASSERT
#define GPR_DEBUG_ASSERT(x)
Definition: log.h:103
grpc::internal::CallOpSendMessage
Definition: call_op_set.h:286
grpc::internal::ClientAsyncResponseReaderHelper::Create
static ClientAsyncResponseReader< R > * Create(grpc::ChannelInterface *channel, grpc::CompletionQueue *cq, const grpc::internal::RpcMethod &method, grpc::ClientContext *context, const W &request)
Start a call and write the request out if start is set.
Definition: async_unary_call.h:93
grpc::internal::ClientAsyncResponseReaderFactory
Definition: async_unary_call.h:197
grpc::internal::Call
Straightforward wrapping of the C call object.
Definition: call.h:36
GPR_ASSERT
#define GPR_ASSERT(x)
abort() the process if x is zero, having written a line to the log.
Definition: log.h:95
status.h
grpc::internal::CallOpSendInitialMetadata
Definition: call_op_set.h:216
grpc::Status::ok
bool ok() const
Is the status OK?
Definition: status.h:126
grpc::ServerAsyncResponseWriter::Finish
void Finish(const W &msg, const grpc::Status &status, void *tag)
Indicate that the stream is to be finished and request notification when the server has sent the appr...
Definition: async_unary_call.h:338
std::default_delete< grpc::ClientAsyncResponseReader< R > >::operator()
void operator()(void *)
Definition: async_unary_call.h:409
grpc::ClientAsyncResponseReaderInterface::Finish
virtual void Finish(R *msg, grpc::Status *status, void *tag)=0
Request to receive the server's response msg and final status for the call, and to notify tag on this...
grpc::Status
Did it work? If it didn't, why?
Definition: status.h:35
grpc::ClientAsyncResponseReaderInterface
An interface relevant for async client side unary RPCs (which send one request message to a server an...
Definition: async_unary_call.h:42
grpc::ClientAsyncResponseReaderInterface::StartCall
virtual void StartCall()=0
Start the call that was set up by the constructor, but only if the constructor was invoked through th...
grpc::ClientAsyncResponseReaderInterface::ReadInitialMetadata
virtual void ReadInitialMetadata(void *tag)=0
Request notification of the reading of initial metadata.
grpc::ServerContext::compression_level
grpc_compression_level compression_level() const
Return the compression algorithm to be used by the server call.
Definition: server_context.h:236
grpc::internal::CallOpClientRecvStatus::ClientRecvStatus
void ClientRecvStatus(grpc::ClientContext *context, Status *status)
Definition: call_op_set.h:774
log.h
grpc::ClientContext
A ClientContext allows the person implementing a service client to:
Definition: client_context.h:193
grpc::ClientAsyncResponseReader
Async API for client-side unary RPCs, where the message response received from the server is of type ...
Definition: client_context.h:91
grpc.h
grpc_call
struct grpc_call grpc_call
A Call represents an RPC.
Definition: grpc_types.h:70
std::default_delete< grpc::ClientAsyncResponseReaderInterface< R > >::operator()
void operator()(void *)
Definition: async_unary_call.h:414
grpc::internal::CallOpSetInterface
An abstract collection of call ops, used to generate the grpc_call_op structure to pass down to the l...
Definition: call_op_set_interface.h:36
grpc::ServerAsyncResponseWriter::ServerAsyncResponseWriter
ServerAsyncResponseWriter(grpc::ServerContext *ctx)
Definition: async_unary_call.h:296
channel_interface.h
grpc::ChannelInterface
Codegen interface for grpc::Channel.
Definition: channel_interface.h:71
grpc::ServerAsyncResponseWriter::FinishWithError
void FinishWithError(const grpc::Status &status, void *tag)
Indicate that the stream is to be finished with a non-OK status, and request notification for when th...
Definition: async_unary_call.h:375
grpc::internal::ClientAsyncResponseReaderHelper
Definition: async_unary_call.h:78
grpc::internal::ClientAsyncResponseReaderHelper::SetupRequest
static void SetupRequest(grpc_call *call, grpc::internal::CallOpSendInitialMetadata **single_buf_ptr, std::function< void(ClientContext *, internal::Call *, internal::CallOpSendInitialMetadata *, void *)> *read_initial_metadata, std::function< void(ClientContext *, internal::Call *, bool initial_metadata_read, internal::CallOpSendInitialMetadata *, internal::CallOpSetInterface **, void *, Status *, void *)> *finish, const W &request)
Definition: async_unary_call.h:111
grpc::ClientAsyncResponseReaderInterface::~ClientAsyncResponseReaderInterface
virtual ~ClientAsyncResponseReaderInterface()
Definition: async_unary_call.h:44
call_op_set_interface.h
grpc::ClientAsyncResponseReader::Finish
void Finish(R *msg, grpc::Status *status, void *tag) override
See ClientAsyncResponseReaderInterface::Finish for semantics.
Definition: async_unary_call.h:257
grpc::ClientAsyncResponseReader::ReadInitialMetadata
void ReadInitialMetadata(void *tag) override
See ClientAsyncResponseReaderInterface::ReadInitialMetadata for semantics.
Definition: async_unary_call.h:245
grpc::internal::CallOpSet::set_output_tag
void set_output_tag(void *return_tag)
Definition: call_op_set.h:935
grpc::internal::ClientAsyncResponseReaderFactory::Create
static ClientAsyncResponseReader< R > * Create(grpc::ChannelInterface *channel, grpc::CompletionQueue *cq, const grpc::internal::RpcMethod &method, grpc::ClientContext *context, const W &request, bool start)
Definition: async_unary_call.h:200
client_context.h
grpc::internal::ClientAsyncResponseReaderHelper::StartCall
static void StartCall(grpc::ClientContext *context, grpc::internal::CallOpSendInitialMetadata *single_buf)
Definition: async_unary_call.h:187
grpc::ServerAsyncResponseWriter::SendInitialMetadata
void SendInitialMetadata(void *tag) override
See ServerAsyncStreamingInterface::SendInitialMetadata for semantics.
Definition: async_unary_call.h:306
grpc::internal::Call::PerformOps
void PerformOps(CallOpSetInterface *ops)
Definition: call.h:66
grpc::internal::Call::call
grpc_call * call() const
Definition: call.h:70
server_context.h
call_op_set.h
std
Definition: async_unary_call.h:405
call.h
grpc::ServerContext::compression_level_set
bool compression_level_set() const
Return a bool indicating whether the compression level for this call has been set (either implicitly ...
Definition: server_context.h:251
grpc::internal::CallOpClientRecvStatus
Definition: call_op_set.h:769
service_type.h
grpc::CompletionQueue
A thin wrapper around grpc_completion_queue (see src/core/lib/surface/completion_queue....
Definition: completion_queue.h:103
grpc::internal::CallOpRecvMessage
Definition: call_op_set.h:424
grpc::internal::RpcMethod
Descriptor of an RPC method.
Definition: rpc_method.h:29
grpc::internal::CallOpSendInitialMetadata::SendInitialMetadata
void SendInitialMetadata(std::multimap< std::string, std::string > *metadata, uint32_t flags)
Definition: call_op_set.h:222
grpc::protobuf::util::Status
::absl::Status Status
Definition: config_protobuf.h:97
grpc::internal::CallOpSet::set_core_cq_tag
void set_core_cq_tag(void *core_cq_tag)
set_core_cq_tag is used to provide a different core CQ tag than "this".
Definition: call_op_set.h:943