Merge pull request #16302 from vjpai/callback_cq

EXPERIMENTAL: Infrastructure for callback-based core CQ

5 files changed, 34 insertions, 12 deletions

diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h
index dc22008932..897b89851a 100644
--- a/include/grpc/grpc.h
+++ b/include/grpc/grpc.h
@@ -101,6 +101,12 @@ GRPCAPI grpc_completion_queue* grpc_completion_queue_create_for_next(
 GRPCAPI grpc_completion_queue* grpc_completion_queue_create_for_pluck(
     void* reserved);
 
+/** Helper function to create a completion queue with grpc_cq_completion_type
+    of GRPC_CQ_CALLBACK and grpc_cq_polling_type of GRPC_CQ_DEFAULT_POLLING.
+    This function is experimental. */
+GRPCAPI grpc_completion_queue* grpc_completion_queue_create_for_callback(
+    void* shutdown_callback, void* reserved);
+
 /** Create a completion queue */
 GRPCAPI grpc_completion_queue* grpc_completion_queue_create(
     const grpc_completion_queue_factory* factory,
diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h
index 5fd080c48b..b5353c1dea 100644
--- a/include/grpc/impl/codegen/grpc_types.h
+++ b/include/grpc/impl/codegen/grpc_types.h
@@ -651,10 +651,16 @@ typedef enum {
   GRPC_CQ_NEXT,
 
   /** Events are popped out by calling grpc_completion_queue_pluck() API ONLY*/
-  GRPC_CQ_PLUCK
+  GRPC_CQ_PLUCK,
+
+  /** EXPERIMENTAL: Events trigger a callback specified as the tag */
+  GRPC_CQ_CALLBACK
 } grpc_cq_completion_type;
 
-#define GRPC_CQ_CURRENT_VERSION 1
+/* The upgrade to version 2 is currently experimental. */
+
+#define GRPC_CQ_CURRENT_VERSION 2
+#define GRPC_CQ_VERSION_MINIMUM_FOR_CALLBACKABLE 2
 typedef struct grpc_completion_queue_attributes {
   /** The version number of this structure. More fields might be added to this
      structure in future. */
@@ -663,6 +669,15 @@ typedef struct grpc_completion_queue_attributes {
   grpc_cq_completion_type cq_completion_type;
 
   grpc_cq_polling_type cq_polling_type;
+
+  /* END OF VERSION 1 CQ ATTRIBUTES */
+
+  /* EXPERIMENTAL: START OF VERSION 2 CQ ATTRIBUTES */
+  /** When creating a callbackable CQ, pass in a functor to get invoked when
+   * shutdown is complete */
+  void* cq_shutdown_cb;
+
+  /* END OF VERSION 2 CQ ATTRIBUTES */
 } grpc_completion_queue_attributes;
 
 /** The completion queue factory structure is opaque to the callers of grpc */
diff --git a/include/grpcpp/impl/codegen/client_unary_call.h b/include/grpcpp/impl/codegen/client_unary_call.h
index a37a81b75b..e4e8364e07 100644
--- a/include/grpcpp/impl/codegen/client_unary_call.h
+++ b/include/grpcpp/impl/codegen/client_unary_call.h
@@ -50,8 +50,8 @@ class BlockingUnaryCallImpl {
                         ClientContext* context, const InputMessage& request,
                         OutputMessage* result) {
     CompletionQueue cq(grpc_completion_queue_attributes{
-        GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
-        GRPC_CQ_DEFAULT_POLLING});  // Pluckable completion queue
+        GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_DEFAULT_POLLING,
+        nullptr});  // Pluckable completion queue
     Call call(channel->CreateCall(method, context, &cq));
     CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
               CallOpRecvInitialMetadata, CallOpRecvMessage<OutputMessage>,
diff --git a/include/grpcpp/impl/codegen/completion_queue.h b/include/grpcpp/impl/codegen/completion_queue.h
index 5819e068ba..272575dac2 100644
--- a/include/grpcpp/impl/codegen/completion_queue.h
+++ b/include/grpcpp/impl/codegen/completion_queue.h
@@ -97,7 +97,8 @@ class CompletionQueue : private GrpcLibraryCodegen {
   /// instance.
   CompletionQueue()
       : CompletionQueue(grpc_completion_queue_attributes{
-            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING}) {}
+            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING,
+            nullptr}) {}
 
   /// Wrap \a take, taking ownership of the instance.
   ///
@@ -376,7 +377,7 @@ class ServerCompletionQueue : public CompletionQueue {
   /// frequently polled.
   ServerCompletionQueue(grpc_cq_polling_type polling_type)
       : CompletionQueue(grpc_completion_queue_attributes{
-            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, polling_type}),
+            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, polling_type, nullptr}),
         polling_type_(polling_type) {}
 
   grpc_cq_polling_type polling_type_;
diff --git a/include/grpcpp/impl/codegen/sync_stream.h b/include/grpcpp/impl/codegen/sync_stream.h
index 7152eaf41f..cbfcf25d0a 100644
--- a/include/grpcpp/impl/codegen/sync_stream.h
+++ b/include/grpcpp/impl/codegen/sync_stream.h
@@ -243,8 +243,8 @@ class ClientReader final : public ClientReaderInterface<R> {
                ClientContext* context, const W& request)
       : context_(context),
         cq_(grpc_completion_queue_attributes{
-            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
-            GRPC_CQ_DEFAULT_POLLING}),  // Pluckable cq
+            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_DEFAULT_POLLING,
+            nullptr}),  // Pluckable cq
         call_(channel->CreateCall(method, context, &cq_)) {
     ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
                                 ::grpc::internal::CallOpSendMessage,
@@ -377,8 +377,8 @@ class ClientWriter : public ClientWriterInterface<W> {
                ClientContext* context, R* response)
       : context_(context),
         cq_(grpc_completion_queue_attributes{
-            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
-            GRPC_CQ_DEFAULT_POLLING}),  // Pluckable cq
+            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_DEFAULT_POLLING,
+            nullptr}),  // Pluckable cq
         call_(channel->CreateCall(method, context, &cq_)) {
     finish_ops_.RecvMessage(response);
     finish_ops_.AllowNoMessage();
@@ -551,8 +551,8 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
                      ClientContext* context)
       : context_(context),
         cq_(grpc_completion_queue_attributes{
-            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
-            GRPC_CQ_DEFAULT_POLLING}),  // Pluckable cq
+            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_DEFAULT_POLLING,
+            nullptr}),  // Pluckable cq
         call_(channel->CreateCall(method, context, &cq_)) {
     if (!context_->initial_metadata_corked_) {
       ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>