Experimental infrastructure for callback-based CQ

grpc.def

@@ -20,6 +20,7 @@ EXPORTS
     grpc_completion_queue_factory_lookup
     grpc_completion_queue_create_for_next
     grpc_completion_queue_create_for_pluck
+    grpc_completion_queue_create_for_callback
     grpc_completion_queue_create
     grpc_completion_queue_next
     grpc_completion_queue_pluck

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,

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 */

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>,

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_;

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>

src/core/lib/surface/completion_queue.cc

@@ -184,7 +184,7 @@ static const cq_poller_vtable g_poller_vtable_by_poller_type[] = {
 typedef struct cq_vtable {
   grpc_cq_completion_type cq_completion_type;
   size_t data_size;
-  void (*init)(void* data);
+  void (*init)(void* data, grpc_core::CQCallbackInterface* shutdown_callback);
   void (*shutdown)(grpc_completion_queue* cq);
   void (*destroy)(void* data);
   bool (*begin_op)(grpc_completion_queue* cq, void* tag);
@@ -253,6 +253,29 @@ typedef struct cq_pluck_data {
   plucker pluckers[GRPC_MAX_COMPLETION_QUEUE_PLUCKERS];
 } cq_pluck_data;
 
+typedef struct cq_callback_data {
+  /** No actual completed events queue, unlike other types */
+
+  /** Number of pending events (+1 if we're not shutdown) */
+  gpr_atm pending_events;
+
+  /** Counter of how many things have ever been queued on this completion queue
+      useful for avoiding locks to check the queue */
+  gpr_atm things_queued_ever;
+
+  /** 0 initially. 1 once we completed shutting */
+  /* TODO: (sreek) This is not needed since (shutdown == 1) if and only if
+   * (pending_events == 0). So consider removing this in future and use
+   * pending_events */
+  gpr_atm shutdown;
+
+  /** 0 initially. 1 once we initiated shutdown */
+  bool shutdown_called;
+
+  /** A callback that gets invoked when the CQ completes shutdown */
+  grpc_core::CQCallbackInterface* shutdown_callback;
+} cq_callback_data;
+
 /* Completion queue structure */
 struct grpc_completion_queue {
   /** Once owning_refs drops to zero, we will destroy the cq */
@@ -276,11 +299,14 @@ struct grpc_completion_queue {
 /* Forward declarations */
 static void cq_finish_shutdown_next(grpc_completion_queue* cq);
 static void cq_finish_shutdown_pluck(grpc_completion_queue* cq);
+static void cq_finish_shutdown_callback(grpc_completion_queue* cq);
 static void cq_shutdown_next(grpc_completion_queue* cq);
 static void cq_shutdown_pluck(grpc_completion_queue* cq);
+static void cq_shutdown_callback(grpc_completion_queue* cq);
 
 static bool cq_begin_op_for_next(grpc_completion_queue* cq, void* tag);
 static bool cq_begin_op_for_pluck(grpc_completion_queue* cq, void* tag);
+static bool cq_begin_op_for_callback(grpc_completion_queue* cq, void* tag);
 
 static void cq_end_op_for_next(grpc_completion_queue* cq, void* tag,
                                grpc_error* error,
@@ -294,16 +320,25 @@ static void cq_end_op_for_pluck(grpc_completion_queue* cq, void* tag,
                                              grpc_cq_completion* storage),
                                 void* done_arg, grpc_cq_completion* storage);
 
+static void cq_end_op_for_callback(grpc_completion_queue* cq, void* tag,
+                                   grpc_error* error,
+                                   void (*done)(void* done_arg,
+                                                grpc_cq_completion* storage),
+                                   void* done_arg, grpc_cq_completion* storage);
+
 static grpc_event cq_next(grpc_completion_queue* cq, gpr_timespec deadline,
                           void* reserved);
 
 static grpc_event cq_pluck(grpc_completion_queue* cq, void* tag,
                            gpr_timespec deadline, void* reserved);
 
-static void cq_init_next(void* data);
-static void cq_init_pluck(void* data);
+static void cq_init_next(void* data, grpc_core::CQCallbackInterface*);
+static void cq_init_pluck(void* data, grpc_core::CQCallbackInterface*);
+static void cq_init_callback(void* data,
+                             grpc_core::CQCallbackInterface* shutdown_callback);
 static void cq_destroy_next(void* data);
 static void cq_destroy_pluck(void* data);
+static void cq_destroy_callback(void* data);
 
 /* Completion queue vtables based on the completion-type */
 static const cq_vtable g_cq_vtable[] = {
@@ -315,6 +350,10 @@ static const cq_vtable g_cq_vtable[] = {
     {GRPC_CQ_PLUCK, sizeof(cq_pluck_data), cq_init_pluck, cq_shutdown_pluck,
      cq_destroy_pluck, cq_begin_op_for_pluck, cq_end_op_for_pluck, nullptr,
      cq_pluck},
+    /* GRPC_CQ_CALLBACK */
+    {GRPC_CQ_CALLBACK, sizeof(cq_callback_data), cq_init_callback,
+     cq_shutdown_callback, cq_destroy_callback, cq_begin_op_for_callback,
+     cq_end_op_for_callback, nullptr, nullptr},
 };
 
 #define DATA_FROM_CQ(cq) ((void*)(cq + 1))
@@ -419,8 +458,8 @@ static long cq_event_queue_num_items(grpc_cq_event_queue* q) {
 }
 
 grpc_completion_queue* grpc_completion_queue_create_internal(
-    grpc_cq_completion_type completion_type,
-    grpc_cq_polling_type polling_type) {
+    grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type,
+    grpc_core::CQCallbackInterface* shutdown_callback) {
   GPR_TIMER_SCOPE("grpc_completion_queue_create_internal", 0);
 
   grpc_completion_queue* cq;
@@ -448,14 +487,14 @@ grpc_completion_queue* grpc_completion_queue_create_internal(
   gpr_ref_init(&cq->owning_refs, 2);
 
   poller_vtable->init(POLLSET_FROM_CQ(cq), &cq->mu);
-  vtable->init(DATA_FROM_CQ(cq));
+  vtable->init(DATA_FROM_CQ(cq), shutdown_callback);
 
   GRPC_CLOSURE_INIT(&cq->pollset_shutdown_done, on_pollset_shutdown_done, cq,
                     grpc_schedule_on_exec_ctx);
   return cq;
 }
 
-static void cq_init_next(void* ptr) {
+static void cq_init_next(void* ptr, grpc_core::CQCallbackInterface*) {
   cq_next_data* cqd = static_cast<cq_next_data*>(ptr);
   /* Initial count is dropped by grpc_completion_queue_shutdown */
   gpr_atm_no_barrier_store(&cqd->pending_events, 1);
@@ -470,7 +509,7 @@ static void cq_destroy_next(void* ptr) {
   cq_event_queue_destroy(&cqd->queue);
 }
 
-static void cq_init_pluck(void* ptr) {
+static void cq_init_pluck(void* ptr, grpc_core::CQCallbackInterface*) {
   cq_pluck_data* cqd = static_cast<cq_pluck_data*>(ptr);
   /* Initial count is dropped by grpc_completion_queue_shutdown */
   gpr_atm_no_barrier_store(&cqd->pending_events, 1);
@@ -487,6 +526,19 @@ static void cq_destroy_pluck(void* ptr) {
   GPR_ASSERT(cqd->completed_head.next == (uintptr_t)&cqd->completed_head);
 }
 
+static void cq_init_callback(
+    void* ptr, grpc_core::CQCallbackInterface* shutdown_callback) {
+  cq_callback_data* cqd = static_cast<cq_callback_data*>(ptr);
+  /* Initial count is dropped by grpc_completion_queue_shutdown */
+  gpr_atm_no_barrier_store(&cqd->pending_events, 1);
+  gpr_atm_no_barrier_store(&cqd->shutdown, 0);
+  cqd->shutdown_called = false;
+  gpr_atm_no_barrier_store(&cqd->things_queued_ever, 0);
+  cqd->shutdown_callback = shutdown_callback;
+}
+
+static void cq_destroy_callback(void* ptr) {}
+
 grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue* cq) {
   return cq->vtable->cq_completion_type;
 }
@@ -596,6 +648,11 @@ static bool cq_begin_op_for_pluck(grpc_completion_queue* cq, void* tag) {
   return atm_inc_if_nonzero(&cqd->pending_events);
 }
 
+static bool cq_begin_op_for_callback(grpc_completion_queue* cq, void* tag) {
+  cq_callback_data* cqd = static_cast<cq_callback_data*> DATA_FROM_CQ(cq);
+  return atm_inc_if_nonzero(&cqd->pending_events);
+}
+
 bool grpc_cq_begin_op(grpc_completion_queue* cq, void* tag) {
 #ifndef NDEBUG
   gpr_mu_lock(cq->mu);
@@ -759,6 +816,47 @@ static void cq_end_op_for_pluck(grpc_completion_queue* cq, void* tag,
   GRPC_ERROR_UNREF(error);
 }
 
+/* Complete an event on a completion queue of type GRPC_CQ_CALLBACK */
+static void cq_end_op_for_callback(
+    grpc_completion_queue* cq, void* tag, grpc_error* error,
+    void (*done)(void* done_arg, grpc_cq_completion* storage), void* done_arg,
+    grpc_cq_completion* storage) {
+  GPR_TIMER_SCOPE("cq_end_op_for_callback", 0);
+
+  cq_callback_data* cqd = static_cast<cq_callback_data*> DATA_FROM_CQ(cq);
+  bool is_success = (error == GRPC_ERROR_NONE);
+
+  if (grpc_api_trace.enabled() ||
+      (grpc_trace_operation_failures.enabled() && error != GRPC_ERROR_NONE)) {
+    const char* errmsg = grpc_error_string(error);
+    GRPC_API_TRACE(
+        "cq_end_op_for_callback(cq=%p, tag=%p, error=%s, "
+        "done=%p, done_arg=%p, storage=%p)",
+        6, (cq, tag, errmsg, done, done_arg, storage));
+    if (grpc_trace_operation_failures.enabled() && error != GRPC_ERROR_NONE) {
+      gpr_log(GPR_ERROR, "Operation failed: tag=%p, error=%s", tag, errmsg);
+    }
+  }
+
+  /* We don't care for the storage content */
+  done(done_arg, storage);
+
+  gpr_mu_lock(cq->mu);
+  cq_check_tag(cq, tag, false); /* Used in debug builds only */
+
+  gpr_atm_no_barrier_fetch_add(&cqd->things_queued_ever, 1);
+  if (gpr_atm_full_fetch_add(&cqd->pending_events, -1) == 1) {
+    cq_finish_shutdown_callback(cq);
+    gpr_mu_unlock(cq->mu);
+  } else {
+    gpr_mu_unlock(cq->mu);
+  }
+
+  GRPC_ERROR_UNREF(error);
+
+  (static_cast<grpc_core::CQCallbackInterface*>(tag))->Run(is_success);
+}
+
 void grpc_cq_end_op(grpc_completion_queue* cq, void* tag, grpc_error* error,
                     void (*done)(void* done_arg, grpc_cq_completion* storage),
                     void* done_arg, grpc_cq_completion* storage) {
@@ -1233,6 +1331,42 @@ static void cq_shutdown_pluck(grpc_completion_queue* cq) {
   GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down (pluck cq)");
 }
 
+static void cq_finish_shutdown_callback(grpc_completion_queue* cq) {
+  cq_callback_data* cqd = static_cast<cq_callback_data*> DATA_FROM_CQ(cq);
+  auto* callback = cqd->shutdown_callback;
+
+  GPR_ASSERT(cqd->shutdown_called);
+  GPR_ASSERT(!gpr_atm_no_barrier_load(&cqd->shutdown));
+  gpr_atm_no_barrier_store(&cqd->shutdown, 1);
+
+  cq->poller_vtable->shutdown(POLLSET_FROM_CQ(cq), &cq->pollset_shutdown_done);
+  callback->Run(true);
+}
+
+static void cq_shutdown_callback(grpc_completion_queue* cq) {
+  cq_callback_data* cqd = static_cast<cq_callback_data*> DATA_FROM_CQ(cq);
+
+  /* Need an extra ref for cq here because:
+   * We call cq_finish_shutdown_pluck() below, that would call pollset shutdown.
+   * Pollset shutdown decrements the cq ref count which can potentially destroy
+   * the cq (if that happens to be the last ref).
+   * Creating an extra ref here prevents the cq from getting destroyed while
+   * this function is still active */
+  GRPC_CQ_INTERNAL_REF(cq, "shutting_down (callback cq)");
+  gpr_mu_lock(cq->mu);
+  if (cqd->shutdown_called) {
+    gpr_mu_unlock(cq->mu);
+    GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down (callback cq)");
+    return;
+  }
+  cqd->shutdown_called = true;
+  if (gpr_atm_full_fetch_add(&cqd->pending_events, -1) == 1) {
+    cq_finish_shutdown_callback(cq);
+  }
+  gpr_mu_unlock(cq->mu);
+  GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down (callback cq)");
+}
+
 /* Shutdown simply drops a ref that we reserved at creation time; if we drop
    to zero here, then enter shutdown mode and wake up any waiters */
 void grpc_completion_queue_shutdown(grpc_completion_queue* cq) {

src/core/lib/surface/completion_queue.h

@@ -47,6 +47,16 @@ typedef struct grpc_cq_completion {
   uintptr_t next;
 } grpc_cq_completion;
 
+/// For callback CQs, the following is what is actually intended by
+/// the tag.
+namespace grpc_core {
+class CQCallbackInterface {
+ public:
+  virtual ~CQCallbackInterface() {}
+  virtual void Run(bool) = 0;
+};
+}  // namespace grpc_core
+
 #ifndef NDEBUG
 void grpc_cq_internal_ref(grpc_completion_queue* cc, const char* reason,
                           const char* file, int line);
@@ -87,6 +97,7 @@ grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue* cc);
 int grpc_get_cq_poll_num(grpc_completion_queue* cc);
 
 grpc_completion_queue* grpc_completion_queue_create_internal(
-    grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type);
+    grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type,
+    grpc_core::CQCallbackInterface* shutdown_callback);
 
 #endif /* GRPC_CORE_LIB_SURFACE_COMPLETION_QUEUE_H */

src/core/lib/surface/completion_queue_factory.cc

@@ -30,8 +30,9 @@
 static grpc_completion_queue* default_create(
     const grpc_completion_queue_factory* factory,
     const grpc_completion_queue_attributes* attr) {
-  return grpc_completion_queue_create_internal(attr->cq_completion_type,
-                                               attr->cq_polling_type);
+  return grpc_completion_queue_create_internal(
+      attr->cq_completion_type, attr->cq_polling_type,
+      static_cast<grpc_core::CQCallbackInterface*>(attr->cq_shutdown_cb));
 }
 
 static grpc_completion_queue_factory_vtable default_vtable = {default_create};
@@ -60,14 +61,22 @@ const grpc_completion_queue_factory* grpc_completion_queue_factory_lookup(
 grpc_completion_queue* grpc_completion_queue_create_for_next(void* reserved) {
   GPR_ASSERT(!reserved);
   grpc_completion_queue_attributes attr = {1, GRPC_CQ_NEXT,
-                                           GRPC_CQ_DEFAULT_POLLING};
+                                           GRPC_CQ_DEFAULT_POLLING, nullptr};
   return g_default_cq_factory.vtable->create(&g_default_cq_factory, &attr);
 }
 
 grpc_completion_queue* grpc_completion_queue_create_for_pluck(void* reserved) {
   GPR_ASSERT(!reserved);
   grpc_completion_queue_attributes attr = {1, GRPC_CQ_PLUCK,
-                                           GRPC_CQ_DEFAULT_POLLING};
+                                           GRPC_CQ_DEFAULT_POLLING, nullptr};
+  return g_default_cq_factory.vtable->create(&g_default_cq_factory, &attr);
+}
+
+grpc_completion_queue* grpc_completion_queue_create_for_callback(
+    void* shutdown_callback, void* reserved) {
+  GPR_ASSERT(!reserved);
+  grpc_completion_queue_attributes attr = {
+      2, GRPC_CQ_CALLBACK, GRPC_CQ_DEFAULT_POLLING, shutdown_callback};
   return g_default_cq_factory.vtable->create(&g_default_cq_factory, &attr);
 }
 

src/objective-c/GRPCClient/private/GRPCCompletionQueue.m

@@ -21,11 +21,11 @@
 #import <grpc/grpc.h>
 
 #ifdef GRPC_CFSTREAM
-const grpc_completion_queue_attributes kCompletionQueueAttr = {GRPC_CQ_CURRENT_VERSION,
-                                                               GRPC_CQ_NEXT, GRPC_CQ_NON_POLLING};
+const grpc_completion_queue_attributes kCompletionQueueAttr = {
+    GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_NON_POLLING, nullptr};
 #else
 const grpc_completion_queue_attributes kCompletionQueueAttr = {
-    GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING};
+    GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING, nullptr};
 #endif
 
 @implementation GRPCCompletionQueue

src/ruby/ext/grpc/rb_grpc_imports.generated.c

@@ -43,6 +43,7 @@ grpc_g_stands_for_type grpc_g_stands_for_import;
 grpc_completion_queue_factory_lookup_type grpc_completion_queue_factory_lookup_import;
 grpc_completion_queue_create_for_next_type grpc_completion_queue_create_for_next_import;
 grpc_completion_queue_create_for_pluck_type grpc_completion_queue_create_for_pluck_import;
+grpc_completion_queue_create_for_callback_type grpc_completion_queue_create_for_callback_import;
 grpc_completion_queue_create_type grpc_completion_queue_create_import;
 grpc_completion_queue_next_type grpc_completion_queue_next_import;
 grpc_completion_queue_pluck_type grpc_completion_queue_pluck_import;
@@ -294,6 +295,7 @@ void grpc_rb_load_imports(HMODULE library) {
   grpc_completion_queue_factory_lookup_import = (grpc_completion_queue_factory_lookup_type) GetProcAddress(library, "grpc_completion_queue_factory_lookup");
   grpc_completion_queue_create_for_next_import = (grpc_completion_queue_create_for_next_type) GetProcAddress(library, "grpc_completion_queue_create_for_next");
   grpc_completion_queue_create_for_pluck_import = (grpc_completion_queue_create_for_pluck_type) GetProcAddress(library, "grpc_completion_queue_create_for_pluck");
+  grpc_completion_queue_create_for_callback_import = (grpc_completion_queue_create_for_callback_type) GetProcAddress(library, "grpc_completion_queue_create_for_callback");
   grpc_completion_queue_create_import = (grpc_completion_queue_create_type) GetProcAddress(library, "grpc_completion_queue_create");
   grpc_completion_queue_next_import = (grpc_completion_queue_next_type) GetProcAddress(library, "grpc_completion_queue_next");
   grpc_completion_queue_pluck_import = (grpc_completion_queue_pluck_type) GetProcAddress(library, "grpc_completion_queue_pluck");

src/ruby/ext/grpc/rb_grpc_imports.generated.h

@@ -104,6 +104,9 @@ extern grpc_completion_queue_create_for_next_type grpc_completion_queue_create_f
 typedef grpc_completion_queue*(*grpc_completion_queue_create_for_pluck_type)(void* reserved);
 extern grpc_completion_queue_create_for_pluck_type grpc_completion_queue_create_for_pluck_import;
 #define grpc_completion_queue_create_for_pluck grpc_completion_queue_create_for_pluck_import
+typedef grpc_completion_queue*(*grpc_completion_queue_create_for_callback_type)(void* shutdown_callback, void* reserved);
+extern grpc_completion_queue_create_for_callback_type grpc_completion_queue_create_for_callback_import;
+#define grpc_completion_queue_create_for_callback grpc_completion_queue_create_for_callback_import
 typedef grpc_completion_queue*(*grpc_completion_queue_create_type)(const grpc_completion_queue_factory* factory, const grpc_completion_queue_attributes* attributes, void* reserved);
 extern grpc_completion_queue_create_type grpc_completion_queue_create_import;
 #define grpc_completion_queue_create grpc_completion_queue_create_import

test/core/surface/public_headers_must_be_c89.c

@@ -82,6 +82,7 @@ int main(int argc, char **argv) {
   printf("%lx", (unsigned long) grpc_completion_queue_factory_lookup);
   printf("%lx", (unsigned long) grpc_completion_queue_create_for_next);
   printf("%lx", (unsigned long) grpc_completion_queue_create_for_pluck);
+  printf("%lx", (unsigned long) grpc_completion_queue_create_for_callback);
   printf("%lx", (unsigned long) grpc_completion_queue_create);
   printf("%lx", (unsigned long) grpc_completion_queue_next);
   printf("%lx", (unsigned long) grpc_completion_queue_pluck);