GRPC C++  1.66.0
completion_queue.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 
32 #ifndef GRPCPP_COMPLETION_QUEUE_H
33 #define GRPCPP_COMPLETION_QUEUE_H
34 
35 #include <list>
36 
37 #include "absl/log/absl_check.h"
38 
39 #include <grpc/grpc.h>
40 #include <grpc/support/atm.h>
41 #include <grpc/support/log.h>
42 #include <grpc/support/time.h>
49 #include <grpcpp/impl/sync.h>
50 
52 
53 namespace grpc {
54 template <class R>
55 class ClientReader;
56 template <class W>
57 class ClientWriter;
58 template <class W, class R>
59 class ClientReaderWriter;
60 template <class R>
62 template <class W>
64 namespace internal {
65 template <class W, class R>
67 
68 template <class ResponseType>
71  grpc::Status&);
72 template <class ServiceType, class RequestType, class ResponseType,
73  class BaseRequestType, class BaseResponseType>
75 template <class ServiceType, class RequestType, class ResponseType>
77 template <class ServiceType, class RequestType, class ResponseType>
79 template <class Streamer, bool WriteNeeded>
81 template <grpc::StatusCode code>
83 } // namespace internal
84 
85 class Channel;
86 class ChannelInterface;
87 class Server;
88 class ServerBuilder;
89 class ServerContextBase;
90 class ServerInterface;
91 
92 namespace internal {
93 class CompletionQueueTag;
94 class RpcMethod;
95 template <class InputMessage, class OutputMessage>
97 template <class Op1, class Op2, class Op3, class Op4, class Op5, class Op6>
98 class CallOpSet;
99 } // namespace internal
100 
106  public:
112  nullptr}) {}
113 
117  explicit CompletionQueue(grpc_completion_queue* take);
118 
121 
123  enum NextStatus {
126  TIMEOUT
128  };
129 
178  bool Next(void** tag, bool* ok) {
179  // Check return type == GOT_EVENT... cases:
180  // SHUTDOWN - queue has been shutdown, return false.
181  // TIMEOUT - we passed infinity time => queue has been shutdown, return
182  // false.
183  // GOT_EVENT - we actually got an event, return true.
184  return (AsyncNextInternal(tag, ok, gpr_inf_future(GPR_CLOCK_REALTIME)) ==
185  GOT_EVENT);
186  }
187 
199  template <typename T>
200  NextStatus AsyncNext(void** tag, bool* ok, const T& deadline) {
201  grpc::TimePoint<T> deadline_tp(deadline);
202  return AsyncNextInternal(tag, ok, deadline_tp.raw_time());
203  }
204 
219  template <typename T, typename F>
220  NextStatus DoThenAsyncNext(F&& f, void** tag, bool* ok, const T& deadline) {
221  CompletionQueueTLSCache cache = CompletionQueueTLSCache(this);
222  f();
223  if (cache.Flush(tag, ok)) {
224  return GOT_EVENT;
225  } else {
226  return AsyncNext(tag, ok, deadline);
227  }
228  }
229 
240  void Shutdown();
241 
247  grpc_completion_queue* cq() { return cq_; }
248 
249  protected:
251  explicit CompletionQueue(const grpc_completion_queue_attributes& attributes) {
253  grpc_completion_queue_factory_lookup(&attributes), &attributes,
254  nullptr);
255  InitialAvalanching(); // reserve this for the future shutdown
256  }
257 
258  private:
259  // Friends for access to server registration lists that enable checking and
260  // logging on shutdown
261  friend class grpc::ServerBuilder;
262  friend class grpc::Server;
263 
264  // Friend synchronous wrappers so that they can access Pluck(), which is
265  // a semi-private API geared towards the synchronous implementation.
266  template <class R>
267  friend class grpc::ClientReader;
268  template <class W>
269  friend class grpc::ClientWriter;
270  template <class W, class R>
272  template <class R>
273  friend class grpc::ServerReader;
274  template <class W>
275  friend class grpc::ServerWriter;
276  template <class W, class R>
278  template <class ResponseType>
281  grpc::Status&);
282  template <class ServiceType, class RequestType, class ResponseType>
284  template <class ServiceType, class RequestType, class ResponseType>
286  template <class Streamer, bool WriteNeeded>
288  template <grpc::StatusCode code>
291  friend class grpc::ServerInterface;
292  template <class InputMessage, class OutputMessage>
294 
295  // Friends that need access to constructor for callback CQ
296  friend class grpc::Channel;
297 
298  // For access to Register/CompleteAvalanching
299  template <class Op1, class Op2, class Op3, class Op4, class Op5, class Op6>
301 
306  class CompletionQueueTLSCache {
307  public:
308  explicit CompletionQueueTLSCache(CompletionQueue* cq);
309  ~CompletionQueueTLSCache();
310  bool Flush(void** tag, bool* ok);
311 
312  private:
313  CompletionQueue* cq_;
314  bool flushed_;
315  };
316 
317  NextStatus AsyncNextInternal(void** tag, bool* ok, gpr_timespec deadline);
318 
321  bool Pluck(grpc::internal::CompletionQueueTag* tag) {
322  auto deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
323  while (true) {
324  auto ev = grpc_completion_queue_pluck(cq_, tag, deadline, nullptr);
325  bool ok = ev.success != 0;
326  void* ignored = tag;
327  if (tag->FinalizeResult(&ignored, &ok)) {
328  ABSL_CHECK(ignored == tag);
329  return ok;
330  }
331  }
332  }
333 
342  void TryPluck(grpc::internal::CompletionQueueTag* tag) {
343  auto deadline = gpr_time_0(GPR_CLOCK_REALTIME);
344  auto ev = grpc_completion_queue_pluck(cq_, tag, deadline, nullptr);
345  if (ev.type == GRPC_QUEUE_TIMEOUT) return;
346  bool ok = ev.success != 0;
347  void* ignored = tag;
348  // the tag must be swallowed if using TryPluck
349  ABSL_CHECK(!tag->FinalizeResult(&ignored, &ok));
350  }
351 
357  void TryPluck(grpc::internal::CompletionQueueTag* tag,
358  gpr_timespec deadline) {
359  auto ev = grpc_completion_queue_pluck(cq_, tag, deadline, nullptr);
360  if (ev.type == GRPC_QUEUE_TIMEOUT || ev.type == GRPC_QUEUE_SHUTDOWN) {
361  return;
362  }
363 
364  bool ok = ev.success != 0;
365  void* ignored = tag;
366  ABSL_CHECK(!tag->FinalizeResult(&ignored, &ok));
367  }
368 
375  void InitialAvalanching() {
376  gpr_atm_rel_store(&avalanches_in_flight_, gpr_atm{1});
377  }
378  void RegisterAvalanching() {
379  gpr_atm_no_barrier_fetch_add(&avalanches_in_flight_, gpr_atm{1});
380  }
381  void CompleteAvalanching() {
382  if (gpr_atm_no_barrier_fetch_add(&avalanches_in_flight_, gpr_atm{-1}) ==
383  1) {
385  }
386  }
387 
388  void RegisterServer(const grpc::Server* server) {
389  (void)server;
390 #ifndef NDEBUG
391  grpc::internal::MutexLock l(&server_list_mutex_);
392  server_list_.push_back(server);
393 #endif
394  }
395  void UnregisterServer(const grpc::Server* server) {
396  (void)server;
397 #ifndef NDEBUG
398  grpc::internal::MutexLock l(&server_list_mutex_);
399  server_list_.remove(server);
400 #endif
401  }
402  bool ServerListEmpty() const {
403 #ifndef NDEBUG
404  grpc::internal::MutexLock l(&server_list_mutex_);
405  return server_list_.empty();
406 #endif
407  return true;
408  }
409 
410  static CompletionQueue* CallbackAlternativeCQ();
411  static void ReleaseCallbackAlternativeCQ(CompletionQueue* cq);
412 
413  grpc_completion_queue* cq_; // owned
414 
415  gpr_atm avalanches_in_flight_;
416 
417  // List of servers associated with this CQ. Even though this is only used with
418  // NDEBUG, instantiate it in all cases since otherwise the size will be
419  // inconsistent.
420  mutable grpc::internal::Mutex server_list_mutex_;
421  std::list<const grpc::Server*>
422  server_list_ /* GUARDED_BY(server_list_mutex_) */;
423 };
424 
428  public:
429  bool IsFrequentlyPolled() { return polling_type_ != GRPC_CQ_NON_LISTENING; }
430 
431  protected:
434 
435  private:
443  grpc_cq_polling_type polling_type,
444  grpc_completion_queue_functor* shutdown_cb)
446  GRPC_CQ_CURRENT_VERSION, completion_type, polling_type,
447  shutdown_cb}),
448  polling_type_(polling_type) {}
449 
450  grpc_cq_polling_type polling_type_;
451  friend class grpc::ServerBuilder;
452  friend class grpc::Server;
453 };
454 
455 } // namespace grpc
456 
457 #endif // GRPCPP_COMPLETION_QUEUE_H
grpc::CompletionQueue::Shutdown
void Shutdown()
Request the shutdown of the queue.
grpc::ClientWriter
Synchronous (blocking) client-side API for doing client-streaming RPCs, where the outgoing message st...
Definition: client_context.h:83
GRPC_CQ_NEXT
@ GRPC_CQ_NEXT
Events are popped out by calling grpc_completion_queue_next() API ONLY.
Definition: grpc_types.h:431
grpc::Server
Represents a gRPC server.
Definition: server.h:57
time.h
rpc_service_method.h
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:98
status.h
grpc_cq_polling_type
grpc_cq_polling_type
Completion queues internally MAY maintain a set of file descriptors in a structure called 'pollset'.
Definition: grpc_types.h:411
grpc::internal::ErrorMethodHandler
General method handler class for errors that prevent real method use e.g., handle unknown method by r...
Definition: completion_queue.h:82
grpc::internal::MethodHandler::HandlerParameter
Definition: rpc_service_method.h:43
grpc::internal::RpcMethodHandler
A wrapper class of an application provided rpc method handler.
Definition: completion_queue.h:74
grpc::ServerWriter
Synchronous (blocking) server-side API for doing for doing a server-streaming RPCs,...
Definition: completion_queue.h:63
grpc_cq_completion_type
grpc_cq_completion_type
Specifies the type of APIs to use to pop events from the completion queue.
Definition: grpc_types.h:429
grpc::CompletionQueue::AsyncNext
NextStatus AsyncNext(void **tag, bool *ok, const T &deadline)
Read from the queue, blocking up to deadline (or the queue's shutdown).
Definition: completion_queue.h:200
grpc::internal::BlockingUnaryCallImpl
Definition: client_context.h:103
GRPC_QUEUE_SHUTDOWN
@ GRPC_QUEUE_SHUTDOWN
Shutting down.
Definition: grpc_types.h:227
grpc::ServerContextBase
Base class of ServerContext.
Definition: server_context.h:124
completion_queue_tag.h
gpr_atm
intptr_t gpr_atm
Definition: atm_gcc_atomic.h:32
grpc::CompletionQueue::DoThenAsyncNext
NextStatus DoThenAsyncNext(F &&f, void **tag, bool *ok, const T &deadline)
EXPERIMENTAL First executes F, then reads from the queue, blocking up to deadline (or the queue's shu...
Definition: completion_queue.h:220
gpr_inf_future
GPRAPI gpr_timespec gpr_inf_future(gpr_clock_type type)
The far future.
grpc_completion_queue_create
GRPCAPI grpc_completion_queue * grpc_completion_queue_create(const grpc_completion_queue_factory *factory, const grpc_completion_queue_attributes *attributes, void *reserved)
Create a completion queue.
grpc::internal::ServerReaderWriterBody
Definition: completion_queue.h:66
grpc::CompletionQueue::CompletionQueue
CompletionQueue(const grpc_completion_queue_attributes &attributes)
Private constructor of CompletionQueue only visible to friend classes.
Definition: completion_queue.h:251
grpc::Status
Did it work? If it didn't, why?
Definition: status.h:34
GRPC_CQ_DEFAULT_POLLING
@ GRPC_CQ_DEFAULT_POLLING
The completion queue will have an associated pollset and there is no restriction on the type of file ...
Definition: grpc_types.h:414
GRPC_CQ_NON_LISTENING
@ GRPC_CQ_NON_LISTENING
Similar to GRPC_CQ_DEFAULT_POLLING except that the completion queues will not contain any 'listening ...
Definition: grpc_types.h:419
grpc::ClientReader
Synchronous (blocking) client-side API for doing server-streaming RPCs, where the stream of messages ...
Definition: client_context.h:81
log.h
grpc::ServerReader
Synchronous (blocking) server-side API for doing client-streaming RPCs, where the incoming message st...
Definition: completion_queue.h:61
grpc_completion_queue_factory_lookup
const GRPCAPI grpc_completion_queue_factory * grpc_completion_queue_factory_lookup(const grpc_completion_queue_attributes *attributes)
Returns the completion queue factory based on the attributes.
gpr_time_0
GPRAPI gpr_timespec gpr_time_0(gpr_clock_type type)
Time constants.
grpc.h
grpc::CompletionQueue::NextStatus
NextStatus
Tri-state return for AsyncNext: SHUTDOWN, GOT_EVENT, TIMEOUT.
Definition: completion_queue.h:123
grpc::TimePoint::raw_time
gpr_timespec raw_time()=delete
grpc::ChannelInterface
Codegen interface for grpc::Channel.
Definition: channel_interface.h:71
grpc::CompletionQueue::TIMEOUT
@ TIMEOUT
deadline was reached.
Definition: completion_queue.h:127
grpc::CompletionQueue::cq
grpc_completion_queue * cq()
Returns a raw pointer to the underlying grpc_completion_queue instance.
Definition: completion_queue.h:247
sync.h
grpc_completion_queue_attributes
Definition: grpc_types.h:462
grpc::internal::GrpcLibrary
Classes that require gRPC to be initialized should inherit from this class.
Definition: grpc_library.h:32
grpc::internal::CompletionQueueTag
An interface allowing implementors to process and filter event tags.
Definition: completion_queue_tag.h:26
grpc_completion_queue_pluck
GRPCAPI grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cq, void *tag, gpr_timespec deadline, void *reserved)
Blocks until an event with tag 'tag' is available, the completion queue is being shutdown or deadline...
grpc::internal::ServerStreamingHandler
A wrapper class of an application provided server streaming handler.
Definition: completion_queue.h:78
grpc::ServerCompletionQueue::IsFrequentlyPolled
bool IsFrequentlyPolled()
Definition: completion_queue.h:429
grpc::internal::MutexLock
Definition: sync.h:81
grpc_library.h
grpc::internal::ClientStreamingHandler
A wrapper class of an application provided client streaming handler.
Definition: completion_queue.h:76
grpc::CompletionQueue::GOT_EVENT
@ GOT_EVENT
Got a new event; tag will be filled in with its associated value; ok indicating its success.
Definition: completion_queue.h:125
grpc_completion_queue_destroy
GRPCAPI void grpc_completion_queue_destroy(grpc_completion_queue *cq)
Destroy a completion queue.
grpc::internal::TemplatedBidiStreamingHandler
A wrapper class of an application provided bidi-streaming handler.
Definition: completion_queue.h:80
GRPC_CQ_CURRENT_VERSION
#define GRPC_CQ_CURRENT_VERSION
Definition: grpc_types.h:460
grpc::CompletionQueue::CompletionQueue
CompletionQueue()
Default constructor.
Definition: completion_queue.h:109
grpc::CompletionQueue::~CompletionQueue
~CompletionQueue() override
Destructor. Destroys the owned wrapped completion queue / instance.
Definition: completion_queue.h:120
grpc::ClientReaderWriter
Synchronous (blocking) client-side API for bi-directional streaming RPCs, where the outgoing message ...
Definition: client_context.h:85
grpc::ServerCompletionQueue
A specific type of completion queue used by the processing of notifications by servers.
Definition: completion_queue.h:427
grpc_completion_queue
struct grpc_completion_queue grpc_completion_queue
Completion Queues enable notification of the completion of asynchronous actions.
Definition: grpc_types.h:58
grpc::internal::CompletionQueueTag::FinalizeResult
virtual bool FinalizeResult(void **tag, bool *status)=0
FinalizeResult must be called before informing user code that the operation bound to the underlying c...
grpc_completion_queue_shutdown
GRPCAPI void grpc_completion_queue_shutdown(grpc_completion_queue *cq)
Begin destruction of a completion queue.
grpc_completion_queue_functor
Specifies an interface class to be used as a tag for callback-based completion queues.
Definition: grpc_types.h:444
grpc::CompletionQueue
A thin wrapper around grpc_completion_queue (see src/core/lib/surface/completion_queue....
Definition: completion_queue.h:105
grpc::Channel
Channels represent a connection to an endpoint. Created by CreateChannel.
Definition: channel.h:54
gpr_atm_rel_store
#define gpr_atm_rel_store(p, value)
Definition: atm_gcc_atomic.h:40
grpc::internal::Mutex
Definition: sync.h:58
grpc::internal::RpcMethod
Descriptor of an RPC method.
Definition: rpc_method.h:29
grpc::ServerCompletionQueue::ServerCompletionQueue
ServerCompletionQueue()
Default constructor.
Definition: completion_queue.h:433
grpc::ServerBuilder
A builder class for the creation and startup of grpc::Server instances.
Definition: server_builder.h:88
grpc::ServerInterface
Definition: server_interface.h:61
atm.h
gpr_timespec
Analogous to struct timespec.
Definition: time.h:48
grpc::internal::UnaryRunHandlerHelper
void UnaryRunHandlerHelper(const grpc::internal::MethodHandler::HandlerParameter &, ResponseType *, grpc::Status &)
A helper function with reduced templating to do the common work needed to actually send the server re...
Definition: method_handler.h:60
grpc::CompletionQueue::SHUTDOWN
@ SHUTDOWN
The completion queue has been shutdown and fully-drained.
Definition: completion_queue.h:124
sync.h
GRPC_QUEUE_TIMEOUT
@ GRPC_QUEUE_TIMEOUT
No event before timeout.
Definition: grpc_types.h:229
GPR_CLOCK_REALTIME
@ GPR_CLOCK_REALTIME
Realtime clock.
Definition: time.h:37
grpc::TimePoint
If you are trying to use CompletionQueue::AsyncNext with a time class that isn't either gpr_timespec ...
Definition: time.h:40
time.h
grpc::CompletionQueue::Next
bool Next(void **tag, bool *ok)
Read from the queue, blocking until an event is available or the queue is shutting down.
Definition: completion_queue.h:178
gpr_atm_no_barrier_fetch_add
#define gpr_atm_no_barrier_fetch_add(p, delta)
Definition: atm_gcc_atomic.h:45